Files
MRLokopandClaude Opus 5 9d09d9872f Break metrics down per vrepo and per upstream, and time them
The counters were process-wide globals: one upstream_requests_total, one
cache hit rate, one pair of byte counters. With several vrepos and
fallback upstream lists that answers none of the questions that actually
come up — which mirror is slow, which one is 404-ing everything so every
miss costs an extra round trip, what each one costs in bandwidth.

Everything attributable now carries a label: vrepo for what the client
asked for, vrepo plus the upstream URL for what the proxy had to fetch.
Series are registered from the config at startup, so a label value can
only ever be a configured name — a client asking for random paths lands
in a vrepo="" bucket instead of minting series. Handlers mark the repo
once they have parsed the path and passed auth; proxycache reads it back
off the context, so no call site had to change shape.

Bytes are now counted at the socket, where they are real: each upstream
gets a counting dialer and the server runs on a counting listener, so the
new wire counters include request lines, headers, chunked framing and the
TLS records and handshake. The body counters stay beside them — one is
what the artifact weighs, the other is what the link is billed for. A
cold 68KB jar reads as 68605 body bytes against 75562 on the wire. Client
sockets keep requests for several vrepos on one keep-alive connection, so
those totals stay process-wide rather than split on a guess.

Timings come from httptrace, per wire attempt: connect (DNS included),
TLS handshake, time-to-first-byte and body streaming, plus the inbound
request duration per vrepo. They are histograms, since the average of a
download time hides exactly the tail worth knowing about. Connection
hooks only fire on a real dial, so a quiet connect histogram beside a
busy request counter means the pool is working.

Two knock-on fixes: a transport with a custom DialContext gets no
automatic HTTP/2, so ForceAttemptHTTP2 is now explicit rather than every
upstream silently dropping to HTTP/1.1; and the counting conn implements
ReadFrom so serving a cached file still uses sendfile.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-10 05:45:11 +03:00

514 lines
17 KiB
Go

// Package github implements a pull-through cache for GitHub and Gitea
// release metadata and assets (the gitea vrepo type shares this handler
// with its own URL prefix and cache namespace).
//
// Paths under /<prefix>/<vrepo>/<owner>/<repo>/ (<prefix> = github|gitea):
//
// latest, snapshot resolve alias; 302 to the tag path (default)
// or serve in place (vrepo redirect: false)
// <tag> 302 to <tag>.json
// <tag>.json, latest.json, snapshot.json
// release metadata (GitHub API JSON with asset
// URLs rewritten to this proxy), TTL-cached
// <tag>/<asset>, latest/<asset>, snapshot/<asset>
// asset download, cached forever
//
// "latest" is the latest stable release (GitHub releases/latest), "snapshot"
// the most recent prerelease. Both names are reserved. Tags containing "/"
// work in the .json and asset forms (asset names never contain "/") but not
// bare. A request ending in ".json" with a "/" before it is ambiguous
// (slash-tag metadata vs .json-named asset): metadata is tried first, and
// an upstream 404 falls back to the asset form.
package github
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"net/url"
"path"
"strconv"
"strings"
"tiny-proxy/internal/auth"
"tiny-proxy/internal/config"
"tiny-proxy/internal/metrics"
"tiny-proxy/internal/proxycache"
"tiny-proxy/internal/upstream"
)
type repo struct {
cfg config.VRepo
client *http.Client
prefix string // "github" | "gitea" — URL and cache-key namespace
webBase string // e.g. https://github.com
apiBase string // e.g. https://api.github.com
apiAccept string // Accept header for API requests
// noStore mirrors upstream cache: false — release metadata, resolved
// aliases and assets are proxied through instead of being stored.
noStore bool
}
// result tags an upstream response with this repo's cache policy, for the
// proxycache callbacks.
func (rp *repo) result(resp *http.Response, err error) (proxycache.Result, error) {
return proxycache.Result{Resp: resp, NoStore: rp.noStore}, err
}
type Handler struct {
repos map[string]*repo
cache *proxycache.Cache
}
// NewHandler collects github and gitea vrepos. The same handler is mounted
// at both /github/ and /gitea/; rp.prefix disambiguates the two namespaces.
func NewHandler(repos []config.VRepo, clients map[string]*http.Client, cache *proxycache.Cache) *Handler {
h := &Handler{repos: map[string]*repo{}, cache: cache}
for _, rc := range repos {
if rc.Type != config.VRepoGithub && rc.Type != config.VRepoGitea {
continue
}
h.repos[rc.Name] = newRepo(rc, clients[rc.Name])
}
return h
}
// newRepo derives the API base and Accept header from the vrepo type: gitea
// serves its API at <web>/api/v1, github.com at api.github.com, and GitHub
// Enterprise at <host>/api/v3.
func newRepo(rc config.VRepo, client *http.Client) *repo {
web := strings.TrimSuffix(rc.Upstream.URL, "/")
rp := &repo{
cfg: rc,
client: client,
prefix: string(rc.Type),
webBase: web,
noStore: !rc.Upstream.CacheEnabled(),
}
if rc.Type == config.VRepoGitea {
rp.apiBase = web + "/api/v1"
rp.apiAccept = "application/json"
} else {
rp.apiBase = apiBase(web)
rp.apiAccept = "application/vnd.github+json"
}
return rp
}
// apiBase derives the REST API base from the web base: github.com uses
// api.github.com, GitHub Enterprise serves the API at <host>/api/v3.
func apiBase(webBase string) string {
if u, err := url.Parse(webBase); err == nil && (u.Host == "github.com" || u.Host == "www.github.com") {
return "https://api.github.com"
}
return webBase + "/api/v3"
}
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet && r.Method != http.MethodHead {
http.Error(w, "read-only repository", http.StatusMethodNotAllowed)
return
}
// The handler is mounted at both /github/ and /gitea/; the first path
// segment is the mount prefix and must match the vrepo's type so the
// two namespaces never mix.
rest := strings.TrimPrefix(r.URL.Path, "/")
mount, rest, ok0 := strings.Cut(rest, "/")
vrepo, rest, ok := strings.Cut(rest, "/")
owner, rest, ok2 := strings.Cut(rest, "/")
ghRepo, rest, _ := strings.Cut(rest, "/")
if !ok0 || !ok || !ok2 || ghRepo == "" {
http.Error(w, "bad path", http.StatusBadRequest)
return
}
rp, exists := h.repos[vrepo]
if !exists || mount != rp.prefix {
http.Error(w, "unknown repository", http.StatusNotFound)
return
}
if !auth.Check(rp.cfg.Auth, r) {
auth.Unauthorized(w)
return
}
// Configured and authorized, so it is safe as a metrics label; the cache
// reads it back from the context to attribute its hits and misses.
metrics.MarkVRepo(r.Context(), vrepo)
if !validName(owner) || !validName(ghRepo) || !allowed(rp.cfg.Allow, owner, ghRepo) {
http.Error(w, "not found", http.StatusNotFound)
return
}
if rest == "" {
h.redirect(w, r, rp, owner, ghRepo, "latest.json")
return
}
// A ".json" suffix marks metadata, but "<tag>/<asset>.json" is
// ambiguous with a slash-containing tag: try metadata, fall back to
// the asset form on an upstream 404.
if s, found := strings.CutSuffix(rest, ".json"); found {
h.serveMetadata(w, r, rp, owner, ghRepo, s, strings.Contains(s, "/"))
return
}
// Otherwise a "/" splits at the LAST slash: tag before, asset after.
selector, asset := rest, ""
if i := strings.LastIndex(rest, "/"); i >= 0 {
selector, asset = rest[:i], rest[i+1:]
if asset == "" {
http.Error(w, "bad path", http.StatusBadRequest)
return
}
}
if isAlias(selector) {
tag, err := h.resolve(r.Context(), rp, owner, ghRepo, selector)
if err != nil {
h.resolveError(w, selector, err)
return
}
// A tag literally named "latest"/"snapshot" would redirect onto
// the same URL forever — serve it in place instead.
if rp.cfg.AliasRedirect() && tag != selector {
target := tag
if asset != "" {
target += "/" + asset
}
h.redirect(w, r, rp, owner, ghRepo, target)
return
}
if asset == "" {
h.serveTagMetadata(w, r, rp, owner, ghRepo, tag, false)
return
}
h.serveAsset(w, r, rp, owner, ghRepo, tag, asset)
return
}
if asset == "" {
// Bare tag: canonical metadata URL is <tag>.json.
h.redirect(w, r, rp, owner, ghRepo, selector+".json")
return
}
h.serveAsset(w, r, rp, owner, ghRepo, selector, asset)
}
// serveMetadata answers <selector>.json, resolving aliases first.
// assetFallback reports that the selector contained a "/", so a 404 from
// the tag endpoint should retry as <tag>/<asset>.json.
func (h *Handler) serveMetadata(w http.ResponseWriter, r *http.Request, rp *repo, owner, ghRepo, selector string, assetFallback bool) {
if isAlias(selector) {
tag, err := h.resolve(r.Context(), rp, owner, ghRepo, selector)
if err != nil {
h.resolveError(w, selector, err)
return
}
selector = tag
}
h.serveTagMetadata(w, r, rp, owner, ghRepo, selector, assetFallback)
}
// serveTagMetadata answers with the cached GitHub release object for tag,
// asset download URLs rewritten to this proxy.
func (h *Handler) serveTagMetadata(w http.ResponseWriter, r *http.Request, rp *repo, owner, ghRepo, tag string, assetFallback bool) {
// Ambiguous ".json" request: if the asset form is already cached, serve
// it directly and skip the API probe entirely.
if assetFallback {
if i := strings.LastIndex(tag, "/"); i > 0 {
key := assetKey(rp, owner, ghRepo, tag[:i], tag[i+1:]+".json")
if _, err := h.cache.Store.Stat(r.Context(), key); err == nil {
h.serveAsset(w, r, rp, owner, ghRepo, tag[:i], tag[i+1:]+".json")
return
}
}
}
key := rp.prefix + "/" + rp.cfg.Name + "/" + owner + "/" + ghRepo + "/" + tag + ".json"
body, err := h.cache.Fetch(r.Context(), key, rp.cfg.MutableTTL(), func(ctx context.Context) (proxycache.Result, error) {
return rp.result(h.fetchRelease(ctx, rp, owner, ghRepo, tag))
})
if err != nil {
var ue *proxycache.UpstreamError
if errors.As(err, &ue) && ue.StatusCode == http.StatusNotFound && assetFallback {
if i := strings.LastIndex(tag, "/"); i > 0 {
h.serveAsset(w, r, rp, owner, ghRepo, tag[:i], tag[i+1:]+".json")
return
}
}
if errors.As(err, &ue) && ue.StatusCode == http.StatusNotFound {
http.Error(w, "not found", http.StatusNotFound)
return
}
slog.Error("metadata fetch failed", "key", key, "err", err)
http.Error(w, "upstream error", http.StatusBadGateway)
return
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Length", strconv.Itoa(len(body)))
w.WriteHeader(http.StatusOK)
if r.Method == http.MethodGet {
w.Write(body)
}
}
// fetchRelease loads the release object for tag from the GitHub API and
// rewrites each asset's browser_download_url to this proxy's path.
func (h *Handler) fetchRelease(ctx context.Context, rp *repo, owner, ghRepo, tag string) (*http.Response, error) {
apiURL := rp.apiBase + "/repos/" + owner + "/" + ghRepo + "/releases/tags/" + url.PathEscape(tag)
resp, err := h.get(ctx, rp, apiURL)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return resp, nil // caller treats as uncacheable
}
defer resp.Body.Close()
var rel map[string]any
if err := json.NewDecoder(resp.Body).Decode(&rel); err != nil {
return nil, err
}
assetBase := escapePath("/" + rp.prefix + "/" + rp.cfg.Name + "/" + owner + "/" + ghRepo + "/" + tag + "/")
if assets, ok := rel["assets"].([]any); ok {
for _, a := range assets {
asset, ok := a.(map[string]any)
if !ok {
continue
}
if name, ok := asset["name"].(string); ok && name != "" {
asset["browser_download_url"] = assetBase + url.PathEscape(name)
}
}
}
body, err := json.Marshal(rel)
if err != nil {
return nil, err
}
return jsonResponse(body), nil
}
// serveAsset answers <tag>/<asset> by proxying the GitHub download URL
// (which itself 302s to a signed CDN URL; the client follows it and the
// upstream transport drops credentials on redirects). Cached as immutable.
func (h *Handler) serveAsset(w http.ResponseWriter, r *http.Request, rp *repo, owner, ghRepo, tag, asset string) {
key := assetKey(rp, owner, ghRepo, tag, asset)
upstreamURL := rp.webBase + escapePath("/"+owner+"/"+ghRepo+"/releases/download/"+tag+"/"+asset)
h.cache.Serve(w, r, key, 0, func(ctx context.Context) (proxycache.Result, error) {
req, err := http.NewRequestWithContext(ctx, r.Method, upstreamURL, nil)
if err != nil {
return proxycache.Result{}, err
}
if rng := r.Header.Get("Range"); rng != "" {
req.Header.Set("Range", rng)
}
res, err := rp.result(rp.client.Do(req))
if err == nil && res.Resp != nil {
res.Resume = upstream.Resumer(rp.client, r.Method, upstreamURL)
res.ResumeAttempts = rp.cfg.Upstream.RetryAttempts()
res.ResumeMaxWait = rp.cfg.Upstream.RetryMaxWait()
}
return res, err
})
}
func assetKey(rp *repo, owner, ghRepo, tag, asset string) string {
return rp.prefix + "/" + rp.cfg.Name + "/" + owner + "/" + ghRepo + "/" + tag + "/" + asset
}
// escapePath escapes a URL path while keeping "/" separators, so special
// characters in tags or asset names (#, %, …) survive into links and
// redirects.
func escapePath(p string) string {
return (&url.URL{Path: p}).String()
}
// resolve maps an alias ("latest" | "snapshot") to a concrete tag, caching
// the answer under <prefix>-refs/<v>/<o>/<r>/<alias> with the metadata TTL —
// a separate namespace from asset keys so no tag/asset pair can collide.
func (h *Handler) resolve(ctx context.Context, rp *repo, owner, ghRepo, alias string) (string, error) {
key := rp.prefix + "-refs/" + rp.cfg.Name + "/" + owner + "/" + ghRepo + "/" + alias
body, err := h.cache.Fetch(ctx, key, rp.cfg.MutableTTL(), func(ctx context.Context) (proxycache.Result, error) {
if alias == "latest" {
return rp.result(h.fetchLatest(ctx, rp, owner, ghRepo))
}
return rp.result(h.fetchSnapshot(ctx, rp, owner, ghRepo))
})
if err != nil {
return "", err
}
tag := string(body)
if !validTag(tag) {
return "", fmt.Errorf("upstream returned invalid tag %q", tag)
}
return tag, nil
}
// fetchLatest asks GitHub for the latest stable release (releases/latest
// excludes prereleases and drafts) and returns the tag as a 200 body.
func (h *Handler) fetchLatest(ctx context.Context, rp *repo, owner, ghRepo string) (*http.Response, error) {
apiURL := rp.apiBase + "/repos/" + owner + "/" + ghRepo + "/releases/latest"
resp, err := h.get(ctx, rp, apiURL)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return resp, nil
}
defer resp.Body.Close()
tag, err := decodeTag(resp.Body)
if err != nil {
return nil, err
}
return textResponse([]byte(tag)), nil
}
// fetchSnapshot lists releases (paging until a prerelease turns up, capped
// at 5 pages) and returns the tag of the most recent non-draft prerelease;
// a 404 body when the repo has none.
func (h *Handler) fetchSnapshot(ctx context.Context, rp *repo, owner, ghRepo string) (*http.Response, error) {
for page := 1; page <= 5; page++ {
apiURL := fmt.Sprintf("%s/repos/%s/%s/releases?per_page=100&page=%d", rp.apiBase, owner, ghRepo, page)
resp, err := h.get(ctx, rp, apiURL)
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return resp, nil
}
var rels []struct {
TagName string `json:"tag_name"`
Prerelease bool `json:"prerelease"`
Draft bool `json:"draft"`
}
err = json.NewDecoder(resp.Body).Decode(&rels)
resp.Body.Close()
if err != nil {
return nil, err
}
for _, rel := range rels {
if rel.Prerelease && !rel.Draft && rel.TagName != "" {
return textResponse([]byte(rel.TagName)), nil
}
}
if len(rels) < 100 {
break
}
}
return &http.Response{
StatusCode: http.StatusNotFound,
Header: http.Header{"Content-Type": {"text/plain"}},
Body: io.NopCloser(strings.NewReader("no prerelease")),
}, nil
}
func (h *Handler) get(ctx context.Context, rp *repo, apiURL string) (*http.Response, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept", rp.apiAccept)
return rp.client.Do(req)
}
// redirect sends a 302 to an absolute-path URL under this vrepo.
func (h *Handler) redirect(w http.ResponseWriter, r *http.Request, rp *repo, owner, ghRepo, target string) {
p := "/" + rp.prefix + "/" + rp.cfg.Name + "/" + owner + "/" + ghRepo + "/" + target
http.Redirect(w, r, escapePath(p), http.StatusFound)
}
// resolveError maps an alias resolution failure to the client: upstream 404
// (no such release) stays a 404, anything else is a bad gateway.
func (h *Handler) resolveError(w http.ResponseWriter, alias string, err error) {
var ue *proxycache.UpstreamError
if errors.As(err, &ue) && ue.StatusCode == http.StatusNotFound {
http.Error(w, "not found", http.StatusNotFound)
return
}
slog.Error("alias resolution failed", "alias", alias, "err", err)
http.Error(w, "upstream error", http.StatusBadGateway)
}
func isAlias(selector string) bool {
return selector == "latest" || selector == "snapshot"
}
// allowed reports whether the allow list permits owner/repo. An empty list
// disables filtering entirely.
func allowed(patterns []string, owner, repo string) bool {
if len(patterns) == 0 {
return true
}
full := owner + "/" + repo
for _, p := range patterns {
if ok, err := path.Match(p, full); err == nil && ok {
return true
}
}
return false
}
// validName restricts owner/repo to GitHub's character set, rejecting
// path traversal and empty segments.
func validName(s string) bool {
if s == "" || s == "." || s == ".." {
return false
}
for _, c := range s {
if c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' ||
c == '-' || c == '_' || c == '.' {
continue
}
return false
}
return true
}
// validTag rejects tags that could escape the vrepo's cache-key namespace
// or produce broken URLs (defense-in-depth; git refnames already forbid
// most of this).
func validTag(tag string) bool {
if tag == "" || strings.Contains(tag, "..") ||
strings.HasPrefix(tag, "/") || strings.HasSuffix(tag, "/") {
return false
}
for _, c := range tag {
if c < 0x20 || c == 0x7f {
return false
}
}
return true
}
func decodeTag(r io.Reader) (string, error) {
var rel struct {
TagName string `json:"tag_name"`
}
if err := json.NewDecoder(r).Decode(&rel); err != nil {
return "", err
}
if rel.TagName == "" {
return "", errors.New("release has no tag_name")
}
return rel.TagName, nil
}
func jsonResponse(body []byte) *http.Response {
return &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": {"application/json"}},
Body: io.NopCloser(bytes.NewReader(body)),
ContentLength: int64(len(body)),
}
}
func textResponse(body []byte) *http.Response {
return &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": {"text/plain"}},
Body: io.NopCloser(bytes.NewReader(body)),
ContentLength: int64(len(body)),
}
}