diff --git a/internal/database/metadata_cache_test.go b/internal/database/metadata_cache_test.go index 26e0a639..0e19f6c7 100644 --- a/internal/database/metadata_cache_test.go +++ b/internal/database/metadata_cache_test.go @@ -401,3 +401,50 @@ func TestMetadataCacheTimesOffUTC(t *testing.T) { } }) } + +// TestRefreshMetadataCache checks that a refresh moves only fetched_at on, and +// only while the row still records the digest it was asked for. +func TestRefreshMetadataCache(t *testing.T) { + db := setupMetadataCacheDB(t) + fetched := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + entry := &MetadataCacheEntry{ + Ecosystem: testEcosystemNPM, + Name: "lodash", + StoragePath: "_metadata/npm/lodash/metadata", + ETag: sql.NullString{String: `"abc123"`, Valid: true}, + ContentDigest: sql.NullString{String: "sha256:aaaa", Valid: true}, + Size: sql.NullInt64{Int64: 1024, Valid: true}, + FetchedAt: sql.NullTime{Time: fetched, Valid: true}, + } + if err := db.UpsertMetadataCache(entry); err != nil { + t.Fatal(err) + } + + refreshed, err := db.RefreshMetadataCache(testEcosystemNPM, "lodash", "sha256:bbbb", time.Now()) + if err != nil || refreshed { + t.Fatalf("refresh with another digest = %v, %v, want false", refreshed, err) + } + if got, _ := db.GetMetadataCache(testEcosystemNPM, "lodash"); !got.FetchedAt.Time.Equal(fetched) { + t.Errorf("fetched_at = %v after a refresh with another digest, want %v", got.FetchedAt.Time, fetched) + } + + now := time.Now().Truncate(time.Second) + refreshed, err = db.RefreshMetadataCache(testEcosystemNPM, "lodash", "sha256:aaaa", now) + if err != nil || !refreshed { + t.Fatalf("refresh = %v, %v, want true", refreshed, err) + } + got, err := db.GetMetadataCache(testEcosystemNPM, "lodash") + if err != nil { + t.Fatal(err) + } + if !got.FetchedAt.Time.Equal(now) { + t.Errorf("fetched_at = %v, want %v", got.FetchedAt.Time, now) + } + if got.ETag.String != `"abc123"` || got.ContentDigest.String != "sha256:aaaa" || got.Size.Int64 != 1024 || got.StoragePath != entry.StoragePath { + t.Errorf("refresh changed the row: %+v", got) + } + + if refreshed, err := db.RefreshMetadataCache(testEcosystemNPM, "missing", "sha256:aaaa", now); err != nil || refreshed { + t.Errorf("refresh of a missing row = %v, %v, want false", refreshed, err) + } +} diff --git a/internal/database/queries.go b/internal/database/queries.go index cd7c51ce..f1cdbe9f 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -1140,6 +1140,22 @@ func (db *DB) GetMetadataCache(ecosystem, name string) (*MetadataCacheEntry, err return &entry, nil } +// RefreshMetadataCache sets fetched_at on the metadata cache row for ecosystem +// and name, provided the row still records digest, and reports whether it did. +// It is for an upstream 304, which leaves the stored document as it is. +func (db *DB) RefreshMetadataCache(ecosystem, name, digest string, fetchedAt time.Time) (bool, error) { + // fetched_at is stored in UTC, as in UpsertMetadataCache. + res, err := db.Exec(db.Rebind(` + UPDATE metadata_cache SET fetched_at = ?, updated_at = ? + WHERE ecosystem = ? AND name = ? AND content_digest = ? + `), fetchedAt.UTC(), time.Now(), ecosystem, name, digest) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n > 0, err +} + func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { now := time.Now() // fetched_at is a TIMESTAMP without zone on Postgres, and lib/pq reads it diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 62d63aec..6862bc2a 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -1355,7 +1355,10 @@ func (p *Proxy) fetchMetadataFromUpstream(ctx context.Context, ecosystem, cacheK if err == nil { var digest string if p.CacheMetadata { - digest = p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) + digest = p.refreshMetadataBlob(ecosystem, cacheKey, meta) + if digest == "" { + digest = p.cacheMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), meta) + } } return metadataResult{body: meta.body, contentType: meta.contentType, contentEncoding: meta.contentEncoding, digest: digest} } @@ -1505,6 +1508,9 @@ type upstreamMetadata struct { contentEncoding string etag string lastModified time.Time + // notModified is the cache row upstream answered 304 for, whose stored + // bytes body holds. It is nil for a document upstream sent. + notModified *database.MetadataCacheEntry } // fetchUpstreamMetadata fetches metadata from upstream, using ETag for conditional revalidation. @@ -1546,7 +1552,7 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e if readErr != nil { return nil, errStale304 } - meta := &upstreamMetadata{body: data, contentType: contentTypeJSON, etag: entry.ETag.String} + meta := &upstreamMetadata{body: data, contentType: contentTypeJSON, etag: entry.ETag.String, notModified: entry} if entry.ContentType.Valid { meta.contentType = entry.ContentType.String } @@ -1587,6 +1593,28 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e return meta, nil } +// refreshMetadataBlob records that upstream answered 304 for the cached +// document meta was read from, by moving the row's fetched_at on. The bytes in +// storage are unchanged, so they are neither written nor hashed again. It +// returns the row's digest, or "" when the caller has to store meta as for a +// new document: when upstream sent one, the row records no digest, or a +// concurrent fetch has replaced the document since it was read. +func (p *Proxy) refreshMetadataBlob(ecosystem, cacheKey string, meta *upstreamMetadata) string { + entry := meta.notModified + if entry == nil || p.DB == nil || !entry.ContentDigest.Valid { + return "" + } + refreshed, err := p.DB.RefreshMetadataCache(ecosystem, cacheKey, entry.ContentDigest.String, time.Now()) + if err != nil { + p.Logger.Warn("failed to refresh cached metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err) + return "" + } + if !refreshed { + return "" + } + return entry.ContentDigest.String +} + // cacheMetadataBlob stores metadata bytes in storage and updates the // database. It returns the digest it recorded for the bytes, or "" when they // were not cached. diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index ffc0d5f2..1096cdf1 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -38,6 +38,7 @@ type mockStorage struct { signedURL string signErr error seekable bool + stores int } func newMockStorage() *mockStorage { @@ -55,6 +56,7 @@ func (s *mockStorage) Store(_ context.Context, path string, r io.Reader) (int64, return 0, "", err } s.files[path] = data + s.stores++ return int64(len(data)), sha256Hex(string(data)), nil } diff --git a/internal/handler/metadata_not_modified_test.go b/internal/handler/metadata_not_modified_test.go new file mode 100644 index 00000000..a16c82ce --- /dev/null +++ b/internal/handler/metadata_not_modified_test.go @@ -0,0 +1,112 @@ +package handler + +import ( + "context" + "database/sql" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// TestMetadataNotModifiedKeepsStoredDocument checks that a 304 from upstream +// refreshes the cache row without writing the unchanged document to storage +// again, and still serves it. +func TestMetadataNotModifiedKeepsStoredDocument(t *testing.T) { + const body = `{"v":1}` + var notModified atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set(headerETag, `"v1"`) + if r.Header.Get("If-None-Match") == `"v1"` { + notModified.Add(1) + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set(headerContentType, contentTypeJSON) + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(upstream.Close) + + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Nanosecond // revalidate on every request + proxy.HTTPClient = upstream.Client() + + ctx := context.Background() + first := proxy.fetchMetadataDocument(ctx, "test", "pkg", upstream.URL+"/pkg", contentTypeJSON) + if first.err != nil { + t.Fatal(first.err) + } + stored, err := db.GetMetadataCache("test", "pkg") + if err != nil || stored == nil { + t.Fatalf("cache row after first fetch: %v, %v", stored, err) + } + + for range 2 { + res := proxy.fetchMetadataDocument(ctx, "test", "pkg", upstream.URL+"/pkg", contentTypeJSON) + if res.err != nil || string(res.body) != body || res.digest != first.digest { + t.Fatalf("revalidated fetch = %q, digest %q, err %v; want %q, digest %q", res.body, res.digest, res.err, body, first.digest) + } + } + if n := notModified.Load(); n != 2 { + t.Fatalf("upstream answered 304 %d times, want 2", n) + } + if store.stores != 1 { + t.Errorf("document stored %d times, want once", store.stores) + } + refreshed, err := db.GetMetadataCache("test", "pkg") + if err != nil { + t.Fatal(err) + } + if !refreshed.FetchedAt.Time.After(stored.FetchedAt.Time) { + t.Errorf("fetched_at = %v, want later than %v", refreshed.FetchedAt.Time, stored.FetchedAt.Time) + } + if refreshed.ContentDigest != stored.ContentDigest || refreshed.ETag != stored.ETag { + t.Errorf("row validators changed: %+v, want %+v", refreshed, stored) + } +} + +// TestMetadataNotModifiedStoresRowWithoutDigest checks that a 304 for a row +// written before digests were recorded stores the document as before, so the +// row gains its digest. +func TestMetadataNotModifiedStoresRowWithoutDigest(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("If-None-Match") == `"v1"` { + w.WriteHeader(http.StatusNotModified) + return + } + t.Errorf("unconditional upstream request") + })) + t.Cleanup(upstream.Close) + + proxy, db, store, _ := setupTestProxy(t) + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Nanosecond + proxy.HTTPClient = upstream.Client() + + path := metadataStoragePath("test", "pkg") + store.files[path] = []byte(`{"v":1}`) + if err := db.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: "test", + Name: "pkg", + StoragePath: path, + ETag: sql.NullString{String: `"v1"`, Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now().Add(-time.Hour), Valid: true}, + }); err != nil { + t.Fatal(err) + } + + res := proxy.fetchMetadataDocument(context.Background(), "test", "pkg", upstream.URL+"/pkg", contentTypeJSON) + if res.err != nil || string(res.body) != `{"v":1}` { + t.Fatalf("fetch = %q, %v", res.body, res.err) + } + if want := "sha256:" + sha256Hex(`{"v":1}`); res.digest != want { + t.Errorf("digest = %q, want %q", res.digest, want) + } + if store.stores != 1 { + t.Errorf("document stored %d times, want once", store.stores) + } +}