feat(cache): stamp S3 content ETag onto disk copy at write time

internal/cache/disk.go

@@ -427,18 +427,19 @@ func (d *Disk) evictBySize(remainingFiles []evictFileInfo) error {
 }
 
 type diskWriter struct {
-	disk      *Disk
-	file      *os.File
-	key       Key
-	namespace Namespace
-	path      string
-	tempPath  string
-	expiresAt time.Time
-	headers   http.Header
-	size      int64
-	ctx       context.Context
-	cancel    context.CancelCauseFunc
-	closed    bool
+	disk        *Disk
+	file        *os.File
+	key         Key
+	namespace   Namespace
+	path        string
+	tempPath    string
+	expiresAt   time.Time
+	headers     http.Header
+	contentETag string
+	size        int64
+	ctx         context.Context
+	cancel      context.CancelCauseFunc
+	closed      bool
 }
 
 func (w *diskWriter) Write(p []byte) (int, error) {
@@ -488,7 +489,19 @@ func (w *diskWriter) Close() error {
 		return errors.Errorf("failed to rename temp file: %w", err)
 	}
 
-	if err := w.disk.db.set(w.key, w.namespace, w.expiresAt, w.headers); err != nil {
+	headers := w.headers
+	if w.contentETag != "" {
+		// Fold the content ETag into this writer's own atomic commit so the
+		// disk copy is immediately pinnable. Clone so the shared headers map
+		// (also referenced by sibling tier writers) is not mutated.
+		headers = headers.Clone()
+		if headers == nil {
+			headers = http.Header{}
+		}
+		headers.Set(ContentETagHeader, w.contentETag)
+	}
+
+	if err := w.disk.db.set(w.key, w.namespace, w.expiresAt, headers); err != nil {
 		return errors.Join(errors.Errorf("failed to set metadata: %w", err), os.Remove(w.path))
 	}
 
@@ -502,6 +515,12 @@ func (w *diskWriter) Close() error {
 	return nil
 }
 
+// setContentETag records the content ETag to persist with this entry so the
+// disk copy is immediately pinnable. It must be called before Close so the ETag
+// is folded into the writer's atomic metadata commit (rather than a racy
+// post-commit update).
+func (w *diskWriter) setContentETag(etag string) { w.contentETag = etag }
+
 // Namespace creates a namespaced view of the disk cache.
 func (d *Disk) Namespace(namespace Namespace) Cache {
 	// Create a shallow copy with the namespace set

internal/cache/s3.go

@@ -334,8 +334,14 @@ type s3Writer struct {
 	errCh     chan error
 	uploadErr error
 	closed    bool
+	etag      string // content ETag from PutObject, valid after a successful Close
 }
 
+// contentETag returns the object's content ETag once the upload has completed.
+// The channel handoff in Close establishes happens-before with the assignment in
+// upload, so the value is safe to read after Close returns nil.
+func (w *s3Writer) contentETag() string { return w.etag }
+
 func (w *s3Writer) Write(p []byte) (int, error) {
 	n, err := w.pipe.Write(p)
 	if err != nil {
@@ -419,7 +425,7 @@ func (w *s3Writer) upload(pr *io.PipeReader, r io.Reader) {
 	}
 
 	// Upload object with streaming (size -1 means unknown size, will use chunked encoding)
-	_, err := w.s3.client.PutObject(
+	info, err := w.s3.client.PutObject(
 		w.ctx,
 		w.s3.config.Bucket,
 		objectName,
@@ -433,6 +439,7 @@ func (w *s3Writer) upload(pr *io.PipeReader, r io.Reader) {
 		return
 	}
 
+	w.etag = info.ETag
 	w.errCh <- nil
 }
 

internal/cache/tiered.go

@@ -274,6 +274,16 @@ type tieredWriter struct {
 	closed  bool
 }
 
+// contentETagProvider is implemented by a tier writer that learns the object's
+// content ETag while closing (S3).
+type contentETagProvider interface{ contentETag() string }
+
+// contentETagReceiver is implemented by a tier writer that can persist the
+// content ETag with its entry so it becomes pinnable without an S3 round-trip
+// (disk). setContentETag must be called before Close so the ETag is folded into
+// the writer's atomic commit.
+type contentETagReceiver interface{ setContentETag(etag string) }
+
 var _ Writer = (*tieredWriter)(nil)
 
 func (t *tieredWriter) Abort(err error) error {
@@ -288,13 +298,58 @@ func (t *tieredWriter) Close() error {
 	}
 	t.closed = true
 
+	// Close ETag providers (S3) first so the content ETag is known before the
+	// receiving tiers (disk) commit, letting each fold the ETag into its own
+	// atomic metadata write. This makes a freshly written disk copy immediately
+	// pinnable (disk-first range serving) instead of proxying pinned reads
+	// through S3 until a backfill persists the ETag. Both tiers were fed the
+	// same byte stream, so the ETag is valid for the disk copy.
+	var providers, receivers []Writer
+	for _, w := range t.writers {
+		switch {
+		case isContentETagProvider(w):
+			providers = append(providers, w)
+		default:
+			receivers = append(receivers, w)
+		}
+	}
+
+	errs := closeAll(providers)
+	if errors.Join(errs...) == nil {
+		if etag := providerETag(providers); etag != "" {
+			for _, w := range receivers {
+				if r, ok := w.(contentETagReceiver); ok {
+					r.setContentETag(etag)
+				}
+			}
+		}
+	}
+	return errors.Join(append(errs, closeAll(receivers)...)...)
+}
+
+func isContentETagProvider(w Writer) bool {
+	_, ok := w.(contentETagProvider)
+	return ok
+}
+
+// providerETag returns the first non-empty content ETag among providers.
+func providerETag(providers []Writer) string {
+	for _, w := range providers {
+		if p, ok := w.(contentETagProvider); ok && p.contentETag() != "" {
+			return p.contentETag()
+		}
+	}
+	return ""
+}
+
+func closeAll(writers []Writer) []error {
 	wg := sync.WaitGroup{}
-	errs := make([]error, len(t.writers))
-	for i, cache := range t.writers {
-		wg.Go(func() { errs[i] = errors.WithStack(cache.Close()) })
+	errs := make([]error, len(writers))
+	for i, w := range writers {
+		wg.Go(func() { errs[i] = errors.WithStack(w.Close()) })
 	}
 	wg.Wait()
-	return errors.Join(errs...)
+	return errs
 }
 
 func (t *tieredWriter) Write(p []byte) (n int, err error) {

internal/cache/tiered_pinned_test.go

@@ -81,3 +81,51 @@ func TestTieredPinnedRangeFallthrough(t *testing.T) {
 	// Disk warm: same pin still serves identical bytes (now from disk).
 	assert.True(t, bytes.Equal(data, stitch()), "disk-served ranges must match")
 }
+
+// TestTieredCreateStampsDiskETag verifies that a tiered Create writing both disk
+// and S3 stamps S3's content ETag onto the disk copy at write time, so the disk
+// tier is immediately pinnable without a prior Open/backfill. The disk and S3
+// pins must report the same token, and a disk range must match.
+func TestTieredCreateStampsDiskETag(t *testing.T) {
+	bucket := s3clienttest.Start(t)
+	_, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug})
+
+	disk, err := cache.NewDisk(ctx, cache.DiskConfig{Root: t.TempDir(), LimitMB: 1024, MaxTTL: time.Hour})
+	assert.NoError(t, err)
+	defer disk.Close()
+
+	clientProvider := s3client.NewClientProvider(ctx, s3client.Config{Endpoint: s3clienttest.Addr, UseSSL: false})
+	s3, err := cache.NewS3(ctx, cache.S3Config{Bucket: bucket, MaxTTL: time.Hour, UploadPartSizeMB: 16}, clientProvider)
+	assert.NoError(t, err)
+	defer s3.Close()
+
+	key := cache.NewKey("tiered-stamp")
+	data := make([]byte, 4<<20)
+	_, err = rand.Read(data)
+	assert.NoError(t, err)
+
+	tiered := cache.MaybeNewTiered(ctx, []cache.Cache{disk, s3})
+	w, err := tiered.Create(ctx, key, nil, 0)
+	assert.NoError(t, err)
+	_, err = w.Write(data)
+	assert.NoError(t, err)
+	assert.NoError(t, w.Close())
+
+	// Disk is pinnable immediately, without any Open/backfill, and reports the
+	// same pin token as the authoritative S3 tier.
+	diskPin, err := disk.Pin(ctx, key)
+	assert.NoError(t, err)
+	s3Pin, err := s3.Pin(ctx, key)
+	assert.NoError(t, err)
+	assert.Equal(t, s3Pin.Pin, diskPin.Pin)
+	assert.Equal(t, int64(len(data)), diskPin.Size)
+
+	// A range served from the disk tier matches.
+	r, total, err := disk.OpenPinnedRange(ctx, key, diskPin.Pin, 0, int64(len(data))-1)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(len(data)), total)
+	got, err := io.ReadAll(r)
+	assert.NoError(t, err)
+	assert.NoError(t, r.Close())
+	assert.True(t, bytes.Equal(data, got), "disk range must match written bytes")
+}