diff --git a/internal/database/queries.go b/internal/database/queries.go index 3afb1348..4b039d8d 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -1290,6 +1290,35 @@ func (db *DB) RefreshMetadataCache(ecosystem, name, digest string, fetchedAt tim return n > 0, err } +// MoveMetadataCache points the metadata cache row for ecosystem and name at +// to, provided it is still at from and records digest, and reports whether +// it did. The caller has copied the document from from to to. +func (db *DB) MoveMetadataCache(ecosystem, name, from, to, digest string) (bool, error) { + res, err := db.Exec(db.Rebind(` + UPDATE metadata_cache SET storage_path = ?, updated_at = ? + WHERE ecosystem = ? AND name = ? AND storage_path = ? AND content_digest = ? + `), to, time.Now(), ecosystem, name, from, digest) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n > 0, err +} + +// ForgetMetadataDocument points the metadata cache row for ecosystem and name +// at to, provided it is still at from, and clears what described the document +// at from: its digest, validators and fetch time. The row stays, so the next +// lookup fetches the document again rather than serving or revalidating it. +func (db *DB) ForgetMetadataDocument(ecosystem, name, from, to string) error { + _, err := db.Exec(db.Rebind(` + UPDATE metadata_cache + SET storage_path = ?, content_digest = NULL, etag = NULL, last_modified = NULL, + fetched_at = NULL, updated_at = ? + WHERE ecosystem = ? AND name = ? AND storage_path = ? + `), to, time.Now(), ecosystem, name, from) + return err +} + // UpsertMetadataCache writes entry and marks it accessed now, since a // document is stored because a request asked for it. func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 8bcac24f..162a3164 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -1296,11 +1296,6 @@ func (p *Proxy) serveArtifactError(w http.ResponseWriter, err error, clientMsg s // errStale304 is returned when upstream sends 304 but the cached file is missing. var errStale304 = fmt.Errorf("upstream returned 304 but cached file is missing") -// metadataStoragePath builds a storage path for cached metadata. -func metadataStoragePath(ecosystem, cacheKey string) string { - return "_metadata/" + ecosystem + "/" + cacheKey + "/metadata" -} - // FetchOrCacheMetadata fetches metadata from upstream with caching. // On success it returns the raw response bytes and content type. // If upstream fails and a cached copy exists, the cached version is returned. @@ -1392,6 +1387,7 @@ func (p *Proxy) cachedMetadataState(ctx context.Context, ecosystem, cacheKey str return nil, nil } entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey) + entry = p.migrateLegacyMetadataEntry(ctx, ecosystem, cacheKey, entry) if entry != nil && p.MetadataTTL > 0 && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) < p.MetadataTTL { data, ct, err := p.readCachedMetadata(ctx, entry, validate) if err == nil { diff --git a/internal/handler/metadata_path.go b/internal/handler/metadata_path.go new file mode 100644 index 00000000..4637909f --- /dev/null +++ b/internal/handler/metadata_path.go @@ -0,0 +1,134 @@ +package handler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + + "github.com/git-pkgs/proxy/internal/database" +) + +// metadataStoragePath builds the storage path for cached metadata. Cache +// keys are case-sensitive, but on a case-insensitive filesystem (APFS, NTFS) +// keys that differ only in case, such as Go's github.com/Sirupsen/logrus and +// github.com/sirupsen/logrus, would share one file. Keys may also hold names +// Windows reserves or end in a dot. So the path carries a hash of the exact +// key. Keys that already are lowercase SHA-256 hex digests are used as they +// are, so their existing entries stay valid. +func metadataStoragePath(ecosystem, cacheKey string) string { + if !isSHA256Hex(cacheKey) { + digest := sha256.Sum256([]byte(cacheKey)) + cacheKey = hex.EncodeToString(digest[:]) + } + return legacyMetadataStoragePath(ecosystem, cacheKey) +} + +// legacyMetadataStoragePath is where metadata was stored before paths were +// hashed: under the cache key itself. +func legacyMetadataStoragePath(ecosystem, cacheKey string) string { + return "_metadata/" + ecosystem + "/" + cacheKey + "/metadata" +} + +func isSHA256Hex(s string) bool { + if len(s) != 2*sha256.Size { + return false + } + for i := range len(s) { + c := s[i] + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return false + } + } + return true +} + +// migrateLegacyMetadataEntry handles a cache row whose blob is still at its +// legacy, unhashed path. On a case-insensitive filesystem keys differing only +// in case shared that blob, so it is served only if it matches the row's digest. +func (p *Proxy) migrateLegacyMetadataEntry(ctx context.Context, ecosystem, cacheKey string, entry *database.MetadataCacheEntry) *database.MetadataCacheEntry { + if entry == nil || p.Storage == nil { + return entry + } + legacy := legacyMetadataStoragePath(ecosystem, cacheKey) + hashed := metadataStoragePath(ecosystem, cacheKey) + if entry.StoragePath != legacy || legacy == hashed { + return entry + } + var matches bool + if entry.ContentDigest.Valid { + digest, err := p.storedDigest(ctx, legacy) + if err != nil { + p.Logger.Warn("failed to read legacy metadata blob", "ecosystem", ecosystem, "key", cacheKey, "error", err) + return nil + } + matches = digest == entry.ContentDigest.String + } + if !matches { + if err := p.DB.ForgetMetadataDocument(ecosystem, cacheKey, legacy, hashed); err != nil { + p.Logger.Warn("failed to drop legacy metadata row", "ecosystem", ecosystem, "key", cacheKey, "error", err) + return nil + } + p.queueLegacyMetadataBlob(ecosystem, cacheKey, legacy) + return nil + } + if !p.moveLegacyMetadataBlob(ctx, entry, legacy, hashed) { + return entry + } + p.queueLegacyMetadataBlob(ecosystem, cacheKey, legacy) + moved := *entry + moved.StoragePath = hashed + return &moved +} + +// storedDigest hashes the blob at path as cacheMetadataBlob records it. +func (p *Proxy) storedDigest(ctx context.Context, path string) (string, error) { + r, err := p.Storage.Open(ctx, path) + if err != nil { + return "", err + } + defer func() { _ = r.Close() }() + h := sha256.New() + if _, err := io.Copy(h, r); err != nil { + return "", err + } + return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil +} + +// moveLegacyMetadataBlob copies entry's verified blob from legacy to hashed +// and points the row there, reporting whether it did. +func (p *Proxy) moveLegacyMetadataBlob(ctx context.Context, entry *database.MetadataCacheEntry, legacy, hashed string) bool { + logFailure := func(err error) { + p.Logger.Warn("failed to move legacy metadata blob", "ecosystem", entry.Ecosystem, "key", entry.Name, "error", err) + } + r, err := p.Storage.Open(ctx, legacy) + if err != nil { + logFailure(err) + return false + } + defer func() { _ = r.Close() }() + _, hash, err := p.Storage.Store(ctx, hashed, r) + if err != nil { + logFailure(err) + return false + } + if "sha256:"+hash != entry.ContentDigest.String { + logFailure(fmt.Errorf("blob changed while being copied")) + return false + } + moved, err := p.DB.MoveMetadataCache(entry.Ecosystem, entry.Name, legacy, hashed, entry.ContentDigest.String) + if err != nil { + logFailure(err) + return false + } + return moved +} + +// queueLegacyMetadataBlob queues a row's legacy blob for deletion once the row +// no longer reads it. Reclaim skips it while another row still points there. +func (p *Proxy) queueLegacyMetadataBlob(ecosystem, cacheKey, legacy string) { + if err := p.DB.QueuePendingDelete(legacy); err != nil { + p.Logger.Warn("failed to queue legacy metadata blob for deletion", "ecosystem", ecosystem, "key", cacheKey, "error", err) + } +} diff --git a/internal/handler/metadata_path_test.go b/internal/handler/metadata_path_test.go new file mode 100644 index 00000000..b1f040c1 --- /dev/null +++ b/internal/handler/metadata_path_test.go @@ -0,0 +1,351 @@ +package handler + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/registries/fetch" +) + +func TestMetadataStoragePathKeepsCaseApart(t *testing.T) { + for _, pair := range [][2]string{ + {"github.com/Sirupsen/logrus", "github.com/sirupsen/logrus"}, + {"JSONStream", "jsonstream"}, + } { + a, b := metadataStoragePath("go", pair[0]), metadataStoragePath("go", pair[1]) + if strings.EqualFold(a, b) { + t.Errorf("%q and %q map to paths %q and %q, which a case-insensitive filesystem treats as one", pair[0], pair[1], a, b) + } + } + for _, key := range []string{"CON", "nul", "trailing.", "a/../b"} { + p := metadataStoragePath("npm", key) + if strings.Contains(p, key) { + t.Errorf("path %q contains the raw key %q", p, key) + } + } + // Keys that already are digests keep their path, so existing entries + // stay valid. + sum := sha256.Sum256([]byte("generic")) + digest := hex.EncodeToString(sum[:]) + if got, want := metadataStoragePath("generic", digest), "_metadata/generic/"+digest+"/metadata"; got != want { + t.Errorf("digest key path = %q, want %q", got, want) + } +} + +// fileStorageProxy returns a proxy whose metadata is cached on the local +// filesystem, as with file:// storage. +func fileStorageProxy(t *testing.T, client *http.Client) (*Proxy, *database.DB) { + t.Helper() + dir := t.TempDir() + db, err := database.Create(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + store, err := storage.OpenBucket(context.Background(), "file://"+filepath.Join(dir, "cache")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + proxy := NewProxy(db, store, &mockFetcher{}, fetch.NewResolver(), slog.New(slog.NewTextHandler(io.Discard, nil))) + proxy.HTTPClient = client + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + return proxy, db +} + +// TestMetadataKeysDifferingInCaseCachedApart caches metadata for two keys +// that differ only in case on file storage. On a case-insensitive filesystem +// they shared one file, and the first key was served the second's document. +func TestMetadataKeysDifferingInCaseCachedApart(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + _, _ = io.WriteString(w, "document for "+r.URL.Path) + })) + defer upstream.Close() + proxy, _ := fileStorageProxy(t, upstream.Client()) + + keys := []string{"github.com/Sirupsen/logrus", "github.com/sirupsen/logrus"} + for range 2 { + for _, key := range keys { + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/"+key) + if err != nil { + t.Fatal(err) + } + if want := "document for /" + key; string(body) != want { + t.Errorf("%s: served %q, want %q", key, body, want) + } + } + } + if got := requests.Load(); got != 2 { + t.Errorf("upstream requests = %d, want 2", got) + } +} + +// seedLegacyMetadata stores blob at key's legacy path and a row pointing at +// it, fetched at fetchedAt. The row's digest is that of document, or unset +// when document is empty. +func seedLegacyMetadata(t *testing.T, proxy *Proxy, db *database.DB, key, blob, document string, fetchedAt time.Time) string { + t.Helper() + legacy := legacyMetadataStoragePath("go", key) + if _, _, err := proxy.Storage.Store(context.Background(), legacy, strings.NewReader(blob)); err != nil { + t.Fatal(err) + } + entry := &database.MetadataCacheEntry{ + Ecosystem: "go", + Name: key, + StoragePath: legacy, + ETag: sql.NullString{String: `"v1"`, Valid: true}, + Size: sql.NullInt64{Int64: int64(len(blob)), Valid: true}, + FetchedAt: sql.NullTime{Time: fetchedAt, Valid: true}, + } + if document != "" { + sum := sha256.Sum256([]byte(document)) + entry.ContentDigest = sql.NullString{String: "sha256:" + hex.EncodeToString(sum[:]), Valid: true} + } + if err := db.UpsertMetadataCache(entry); err != nil { + t.Fatal(err) + } + return legacy +} + +func pendingDeletes(t *testing.T, db *database.DB) []string { + t.Helper() + due, err := db.GetDuePendingDeletes(time.Now().Add(time.Hour), 100) + if err != nil { + t.Fatal(err) + } + return due +} + +func metadataRowPath(t *testing.T, db *database.DB, key string) string { + t.Helper() + entry, err := db.GetMetadataCache("go", key) + if err != nil || entry == nil { + t.Fatalf("cache row for %s: %v %v", key, entry, err) + } + return entry.StoragePath +} + +// TestLegacyMetadataMatchingBlobServedWhenUpstreamDown checks that a row at +// its legacy path whose blob matches its digest is still served as a stale +// fallback, and is moved to the hashed path. +func TestLegacyMetadataMatchingBlobServedWhenUpstreamDown(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + defer upstream.Close() + proxy, db := fileStorageProxy(t, upstream.Client()) + + const key = "github.com/Sirupsen/logrus" + legacy := seedLegacyMetadata(t, proxy, db, key, "old", "old", time.Now().Add(-2*time.Hour)) + + for range 2 { + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/x") + if err != nil { + t.Fatalf("legacy document not served while upstream is down: %v", err) + } + if string(body) != "old" { + t.Fatalf("served %q, want %q", body, "old") + } + } + if got, want := metadataRowPath(t, db, key), metadataStoragePath("go", key); got != want { + t.Errorf("storage path = %q, want %q", got, want) + } + if due := pendingDeletes(t, db); !slices.Contains(due, legacy) { + t.Errorf("pending deletes = %v, want %q", due, legacy) + } +} + +// TestLegacyMetadataMatchingBlobServedFromCache checks that a fresh row at +// its legacy path whose blob matches is served without asking upstream. +func TestLegacyMetadataMatchingBlobServedFromCache(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + _, _ = io.WriteString(w, "fresh") + })) + defer upstream.Close() + proxy, db := fileStorageProxy(t, upstream.Client()) + + const key = "github.com/Sirupsen/logrus" + seedLegacyMetadata(t, proxy, db, key, "old", "old", time.Now()) + + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/x") + if err != nil { + t.Fatal(err) + } + if string(body) != "old" || requests.Load() != 0 { + t.Errorf("served %q after %d upstream requests, want the cached document", body, requests.Load()) + } +} + +// TestLegacyMetadataMismatchedBlobNotServed checks that a legacy blob that +// does not match the row's digest, as when another key's document +// overwrote it, is neither served nor used as a fallback. +func TestLegacyMetadataMismatchedBlobNotServed(t *testing.T) { + var down atomic.Bool + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if down.Load() { + w.WriteHeader(http.StatusBadGateway) + return + } + if r.Header.Get("If-None-Match") != "" { + t.Errorf("revalidated a document that was dropped") + } + _, _ = io.WriteString(w, "fresh") + })) + defer upstream.Close() + proxy, db := fileStorageProxy(t, upstream.Client()) + + const key = "github.com/Sirupsen/logrus" + seedLegacyMetadata(t, proxy, db, key, "other key's document", "old", time.Now()) + down.Store(true) + if body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/x"); err == nil { + t.Fatalf("served %q, want an error while upstream is down", body) + } + + down.Store(false) + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/x") + if err != nil { + t.Fatal(err) + } + if string(body) != "fresh" { + t.Errorf("served %q, want %q", body, "fresh") + } + if got, want := metadataRowPath(t, db, key), metadataStoragePath("go", key); got != want { + t.Errorf("storage path = %q, want %q", got, want) + } +} + +// caseFoldingStorage stores paths differing only in case as one file, as a +// case-insensitive filesystem does. +type caseFoldingStorage struct{ storage.Storage } + +func (s caseFoldingStorage) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) { + return s.Storage.Store(ctx, strings.ToLower(path), r) +} + +func (s caseFoldingStorage) Open(ctx context.Context, path string) (io.ReadCloser, error) { + return s.Storage.Open(ctx, strings.ToLower(path)) +} + +func (s caseFoldingStorage) Exists(ctx context.Context, path string) (bool, error) { + return s.Storage.Exists(ctx, strings.ToLower(path)) +} + +func (s caseFoldingStorage) Delete(ctx context.Context, path string) error { + return s.Storage.Delete(ctx, strings.ToLower(path)) +} + +func (s caseFoldingStorage) Size(ctx context.Context, path string) (int64, error) { + return s.Storage.Size(ctx, strings.ToLower(path)) +} + +// TestLegacyMetadataCaseVariantRefetchedAfterReclaim checks that on +// case-insensitive storage, a key whose legacy blob was shared with a case +// variant and deleted after that variant moved is fetched again rather than +// served. +func TestLegacyMetadataCaseVariantRefetchedAfterReclaim(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + _, _ = io.WriteString(w, "fresh") + })) + defer upstream.Close() + proxy, db := fileStorageProxy(t, upstream.Client()) + proxy.Storage = caseFoldingStorage{proxy.Storage} + + const upper, lower = "github.com/Sirupsen/logrus", "github.com/sirupsen/logrus" + seedLegacyMetadata(t, proxy, db, upper, "doc", "doc", time.Now()) + seedLegacyMetadata(t, proxy, db, lower, "doc", "doc", time.Now()) + + if body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", upper, upstream.URL+"/x"); err != nil || string(body) != "doc" { + t.Fatalf("FetchOrCacheMetadata(%s) = %q, %v; want the migrated document", upper, body, err) + } + // Reclaim deletes the queued legacy blob, which is also lower's. + for _, path := range pendingDeletes(t, db) { + if err := proxy.Storage.Delete(context.Background(), path); err != nil { + t.Fatal(err) + } + } + + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", lower, upstream.URL+"/x") + if err != nil { + t.Fatal(err) + } + if string(body) != "fresh" || requests.Load() != 1 { + t.Errorf("served %q after %d upstream requests, want a fresh fetch", body, requests.Load()) + } + if got, want := metadataRowPath(t, db, lower), metadataStoragePath("go", lower); got != want { + t.Errorf("storage path = %q, want %q", got, want) + } +} + +// TestLegacyMetadataPathRefetched checks that a row at its legacy path with +// no digest to check its blob against is fetched again, moved to the hashed +// path, and its old blob queued for deletion. +func TestLegacyMetadataPathRefetched(t *testing.T) { + var requests atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + _, _ = io.WriteString(w, "fresh") + })) + defer upstream.Close() + proxy, db := fileStorageProxy(t, upstream.Client()) + + const key = "github.com/Sirupsen/logrus" + legacy := legacyMetadataStoragePath("go", key) + if _, _, err := proxy.Storage.Store(context.Background(), legacy, bytes.NewReader([]byte("old"))); err != nil { + t.Fatal(err) + } + if err := db.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: "go", + Name: key, + StoragePath: legacy, + Size: sql.NullInt64{Int64: 3, Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, + }); err != nil { + t.Fatal(err) + } + + body, _, err := proxy.FetchOrCacheMetadata(context.Background(), "go", key, upstream.URL+"/x") + if err != nil { + t.Fatal(err) + } + if string(body) != "fresh" || requests.Load() != 1 { + t.Fatalf("served %q after %d upstream requests, want a fresh fetch", body, requests.Load()) + } + entry, err := db.GetMetadataCache("go", key) + if err != nil || entry == nil { + t.Fatalf("cache row: %v %v", entry, err) + } + if entry.StoragePath != metadataStoragePath("go", key) { + t.Errorf("storage path = %q, want %q", entry.StoragePath, metadataStoragePath("go", key)) + } + due, err := db.GetDuePendingDeletes(time.Now().Add(time.Hour), 10) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(due, legacy) { + t.Errorf("pending deletes = %v, want %q", due, legacy) + } +}