package handler import ( "io" "net/http" "net/http/httptest" "strings" "sync" "sync/atomic" "testing" "time" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/registries/fetch" ) const apkSnapshotIndexPath = "/v3.22/main/x86_64/APKINDEX.tar.gz" // apkSnapshotUpstream serves the index held in index, initially "index v1", // and counts package downloads. type apkSnapshotUpstream struct { index atomic.Value downloads atomic.Int32 server *httptest.Server } func newAPKSnapshotUpstream(t *testing.T) *apkSnapshotUpstream { t.Helper() u := &apkSnapshotUpstream{} u.index.Store("index v1") u.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case apkSnapshotIndexPath: _, _ = w.Write([]byte(u.index.Load().(string))) case "/v3.22/main/x86_64/curl-8.5.0-r0.apk": u.downloads.Add(1) _, _ = w.Write([]byte("curl 8.5.0")) default: http.NotFound(w, r) } })) t.Cleanup(u.server.Close) return u } // newSnapshotAPKHandler returns an APK handler with snapshots enabled whose // metadata TTL is 0, so every index request asks upstream. func newSnapshotAPKHandler(t *testing.T, upstreamURL string) (*APKHandler, *database.DB) { t.Helper() proxy, db, _, _ := setupTestProxy(t) proxy.CacheMetadata = true proxy.MetadataTTL = 0 proxy.HTTPClient = http.DefaultClient fetcher := fetch.NewFetcher(fetch.WithHTTPClient(http.DefaultClient), fetch.WithMaxRetries(0)) proxy.Fetcher = fetcher t.Cleanup(func() { _ = fetcher.Close() }) h := NewAPKHandler(proxy, "http://proxy.example", map[string]string{"alpine": upstreamURL}). WithSnapshots(30 * 24 * time.Hour) return h, db } func snapshotCount(t *testing.T, db *database.DB) int { t.Helper() var n int if err := db.Get(&n, `SELECT COUNT(*) FROM metadata_snapshots`); err != nil { t.Fatal(err) } return n } func expectAPKResponse(t *testing.T, h *APKHandler, target string, status int, body string) { t.Helper() w := serveAPKRequest(h, target) if w.Code != status { t.Fatalf("%s: status = %d, want %d: %s", target, w.Code, status, w.Body.String()) } if status == http.StatusOK && w.Body.String() != body { t.Errorf("%s: body = %q, want %q", target, w.Body.String(), body) } } // TestAPKSnapshots_ServeIndexCurrentAtDate records two versions of an index // through the plain path and checks that each dated path serves the version // that was current at the end of that day. func TestAPKSnapshots_ServeIndexCurrentAtDate(t *testing.T) { upstream := newAPKSnapshotUpstream(t) h, db := newSnapshotAPKHandler(t, upstream.server.URL) expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") // Pretend v1 was first stored on 2026-01-10. recorded := time.Date(2026, 1, 10, 12, 0, 0, 0, time.UTC) if _, err := db.Exec(`UPDATE metadata_snapshots SET valid_from_ms = ?`, recorded.UnixMilli()); err != nil { t.Fatal(err) } upstream.index.Store("index v2") expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v2") // An unchanged index adds no snapshot. expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v2") if got := snapshotCount(t, db); got != 2 { t.Fatalf("snapshots = %d, want 2", got) } today := time.Now().UTC().Format(apkSnapshotDateLayout) future := time.Now().UTC().AddDate(0, 0, 2).Format(apkSnapshotDateLayout) expectAPKResponse(t, h, "/alpine@2026-01-09"+apkSnapshotIndexPath, http.StatusNotFound, "") expectAPKResponse(t, h, "/alpine@2026-01-10"+apkSnapshotIndexPath, http.StatusOK, "index v1") expectAPKResponse(t, h, "/alpine@"+today+apkSnapshotIndexPath, http.StatusOK, "index v2") expectAPKResponse(t, h, "/alpine@"+future+apkSnapshotIndexPath, http.StatusNotFound, "") expectAPKResponse(t, h, "/alpine@2026-13-01"+apkSnapshotIndexPath, http.StatusNotFound, "") expectAPKResponse(t, h, "/other@2026-01-10"+apkSnapshotIndexPath, http.StatusNotFound, "") expectAPKResponse(t, h, "/alpine@2026-01-10/v3.22/community/x86_64/APKINDEX.tar.gz", http.StatusNotFound, "") } func TestAPKSnapshots_ConditionalRequest(t *testing.T) { upstream := newAPKSnapshotUpstream(t) h, _ := newSnapshotAPKHandler(t, upstream.server.URL) expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") target := "/alpine@" + time.Now().UTC().Format(apkSnapshotDateLayout) + apkSnapshotIndexPath first := serveAPKRequest(h, target) etag := first.Header().Get("ETag") if first.Code != http.StatusOK || etag == "" { t.Fatalf("status = %d, ETag = %q, want 200 with an ETag", first.Code, etag) } req := httptest.NewRequest(http.MethodGet, target, nil) req.Header.Set("If-None-Match", etag) w := httptest.NewRecorder() h.Routes().ServeHTTP(w, req) if w.Code != http.StatusNotModified { t.Errorf("conditional request: status = %d, want 304", w.Code) } } // TestAPKSnapshots_PackagesShareTheCache checks that a package requested // under a snapshot path is the cached artifact of the plain path. func TestAPKSnapshots_PackagesShareTheCache(t *testing.T) { upstream := newAPKSnapshotUpstream(t) h, _ := newSnapshotAPKHandler(t, upstream.server.URL) pkg := "/v3.22/main/x86_64/curl-8.5.0-r0.apk" expectAPKResponse(t, h, "/alpine"+pkg, http.StatusOK, "curl 8.5.0") expectAPKResponse(t, h, "/alpine@2026-01-10"+pkg, http.StatusOK, "curl 8.5.0") if got := upstream.downloads.Load(); got != 1 { t.Errorf("upstream downloads = %d, want 1", got) } expectAPKResponse(t, h, "/alpine@2026-01-10/v3.22/main/x86_64/gone-1.0-r0.apk", http.StatusNotFound, "") } func TestAPKSnapshots_DisabledByDefault(t *testing.T) { upstream := newAPKSnapshotUpstream(t) h, db := newSnapshotAPKHandler(t, upstream.server.URL) h.WithSnapshots(0) expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") if got := snapshotCount(t, db); got != 0 { t.Errorf("snapshots = %d, want 0", got) } today := time.Now().UTC().Format(apkSnapshotDateLayout) expectAPKResponse(t, h, "/alpine@"+today+apkSnapshotIndexPath, http.StatusNotFound, "") } // TestAPKSnapshots_ConfiguredNameWins checks that a repository whose // configured name contains '@' is served as configured, not as a snapshot. func TestAPKSnapshots_ConfiguredNameWins(t *testing.T) { upstream := newAPKSnapshotUpstream(t) h, _ := newSnapshotAPKHandler(t, upstream.server.URL) h.repositories["alpine@2026-01-10"] = upstream.server.URL expectAPKResponse(t, h, "/alpine@2026-01-10"+apkSnapshotIndexPath, http.StatusOK, "index v1") } // TestAPKSnapshots_SameVersionRecordedTwice checks that two requests // recording the same version in the same millisecond, as coalesced fetches // after an upstream update can, leave the recorded snapshot readable: the // request whose row is not added must not delete the other's blob. func TestAPKSnapshots_SameVersionRecordedTwice(t *testing.T) { u := newAPKSnapshotUpstream(t) h, db := newSnapshotAPKHandler(t, u.server.URL) expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") key := h.metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) entry, err := db.GetMetadataCache(apkEcosystem, key) if err != nil || entry == nil { t.Fatalf("cached index row = %v, %v", entry, err) } if _, err := db.Exec(`DELETE FROM metadata_snapshots`); err != nil { t.Fatal(err) } validFrom := time.Now() h.storeSnapshot(t.Context(), key, entry, validFrom) h.storeSnapshot(t.Context(), key, entry, validFrom) if n := snapshotCount(t, db); n != 1 { t.Fatalf("%d snapshots recorded, want 1", n) } snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) if err != nil || snap == nil { t.Fatalf("latest snapshot = %v, %v", snap, err) } body, err := h.proxy.Storage.Open(t.Context(), snap.StoragePath) if err != nil { t.Fatalf("recorded snapshot blob unreadable: %v", err) } _ = body.Close() day := validFrom.UTC().Format(apkSnapshotDateLayout) expectAPKResponse(t, h, "/alpine@"+day+apkSnapshotIndexPath, http.StatusOK, "index v1") } // pausingWriter blocks the first Write until release is closed, after // signalling on writing. type pausingWriter struct { *httptest.ResponseRecorder once sync.Once writing chan struct{} release chan struct{} } func (w *pausingWriter) Write(b []byte) (int, error) { w.once.Do(func() { close(w.writing) <-w.release }) return w.ResponseRecorder.Write(b) } // TestAPKSnapshots_OverlappingRequests pauses a request while it writes // index v1, lets a second request fetch and serve v2, and then lets the first // one finish. Both versions must be kept, in the order they were fetched, // with readable blobs. func TestAPKSnapshots_OverlappingRequests(t *testing.T) { u := newAPKSnapshotUpstream(t) h, db := newSnapshotAPKHandler(t, u.server.URL) target := "/alpine" + apkSnapshotIndexPath first := &pausingWriter{ ResponseRecorder: httptest.NewRecorder(), writing: make(chan struct{}), release: make(chan struct{}), } done := make(chan struct{}) go func() { defer close(done) h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, target, nil)) }() select { case <-first.writing: case <-done: t.Fatal("first request finished without writing a body") } u.index.Store("index v2") expectAPKResponse(t, h, target, http.StatusOK, "index v2") close(first.release) <-done if first.Code != http.StatusOK || first.Body.String() != "index v1" { t.Fatalf("first request = %d %q, want 200 with index v1", first.Code, first.Body.String()) } var snaps []database.MetadataSnapshot if err := db.Select(&snaps, `SELECT `+ `id, ecosystem, name, storage_path, content_type, content_encoding, content_digest, size, valid_from_ms `+ `FROM metadata_snapshots ORDER BY valid_from_ms, id`); err != nil { t.Fatal(err) } if len(snaps) != 2 { t.Fatalf("%d snapshots recorded, want 2", len(snaps)) } for i, want := range []string{"index v1", "index v2"} { body, err := h.proxy.Storage.Open(t.Context(), snaps[i].StoragePath) if err != nil { t.Fatalf("snapshot %d blob unreadable: %v", i, err) } got, _ := io.ReadAll(body) _ = body.Close() if string(got) != want { t.Errorf("snapshot %d = %q, want %q", i, got, want) } if snaps[i].Size != int64(len(want)) { t.Errorf("snapshot %d size = %d, want %d", i, snaps[i].Size, len(want)) } } } // TestAPKSnapshots_ConcurrentReplicas records the same index version from // two replicas sharing a Postgres database, many times at once. One snapshot // row must remain, with a readable blob, and the other copies are deleted. func TestAPKSnapshots_ConcurrentReplicas(t *testing.T) { a, b, db := sharedInstances(t) u := newAPKSnapshotUpstream(t) handlers := make([]*APKHandler, 0, 2) for _, p := range []*Proxy{a, b} { p.CacheMetadata = true p.HTTPClient = http.DefaultClient handlers = append(handlers, NewAPKHandler(p, "http://proxy.example", map[string]string{"alpine": u.server.URL}). WithSnapshots(30*24*time.Hour)) } expectAPKResponse(t, handlers[0], "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") key := handlers[0].metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) entry, err := db.GetMetadataCache(apkEcosystem, key) if err != nil || entry == nil { t.Fatalf("cached index row = %v, %v", entry, err) } if _, err := db.Exec(`DELETE FROM metadata_snapshots`); err != nil { t.Fatal(err) } // The request above recorded a copy of its own, which the test leaves be. store := a.Storage.(*mockStorage) store.mu.Lock() before := map[string]bool{} for path := range store.files { before[path] = true } store.mu.Unlock() validFrom := time.Now() var wg sync.WaitGroup for i := range 16 { wg.Go(func() { handlers[i%2].storeSnapshot(t.Context(), key, entry, validFrom) }) } wg.Wait() if n := snapshotCount(t, db); n != 1 { t.Fatalf("%d snapshots recorded, want 1", n) } snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) if err != nil || snap == nil { t.Fatalf("latest snapshot = %v, %v", snap, err) } store.mu.Lock() defer store.mu.Unlock() if got := string(store.files[snap.StoragePath]); got != "index v1" { t.Errorf("snapshot blob = %q, want the index", got) } for path := range store.files { if strings.HasPrefix(path, "_snapshots/") && path != snap.StoragePath && !before[path] { t.Errorf("copy %s of a snapshot that wasn't recorded is left in storage", path) } } } // TestAPKSnapshots_CacheHit records an index cached before snapshots were // enabled when a request is served from the cache within its TTL, and adds // no row for a later hit on the same version. func TestAPKSnapshots_CacheHit(t *testing.T) { u := newAPKSnapshotUpstream(t) h, db := newSnapshotAPKHandler(t, u.server.URL) h.proxy.MetadataTTL = 5 * time.Minute h.WithSnapshots(0) expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") if n := snapshotCount(t, db); n != 0 { t.Fatalf("%d snapshots recorded with snapshots off, want 0", n) } h.WithSnapshots(30 * 24 * time.Hour) // Upstream moved on; a hit still serves the cached v1. u.index.Store("index v2") expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") if n := snapshotCount(t, db); n != 1 { t.Fatalf("%d snapshots recorded on a cache hit, want 1", n) } key := h.metadataCacheKey("alpine", u.server.URL, apkSnapshotIndexPath[1:]) snap, err := db.LatestMetadataSnapshot(apkEcosystem, key) if err != nil || snap == nil { t.Fatalf("latest snapshot = %v, %v", snap, err) } body, err := h.proxy.Storage.Open(t.Context(), snap.StoragePath) if err != nil { t.Fatalf("recorded snapshot blob unreadable: %v", err) } got, _ := io.ReadAll(body) _ = body.Close() if string(got) != "index v1" { t.Fatalf("snapshot blob = %q, want %q", got, "index v1") } expectAPKResponse(t, h, "/alpine"+apkSnapshotIndexPath, http.StatusOK, "index v1") if n := snapshotCount(t, db); n != 1 { t.Fatalf("%d snapshots after a second hit on the same version, want 1", n) } }