Keep the stored metadata on an upstream 304

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)
+	}
+}

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

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.

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
 }
 

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)
+	}
+}