diff --git a/internal/cache/disk.go b/internal/cache/disk.go index fbd5c680..bc14e5d4 100644 --- a/internal/cache/disk.go +++ b/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 diff --git a/internal/cache/s3.go b/internal/cache/s3.go index c384dc35..a8aa6be5 100644 --- a/internal/cache/s3.go +++ b/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 } diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index 786477cc..7a497b54 100644 --- a/internal/cache/tiered.go +++ b/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) { diff --git a/internal/cache/tiered_pinned_test.go b/internal/cache/tiered_pinned_test.go index 2ef02550..974557e7 100644 --- a/internal/cache/tiered_pinned_test.go +++ b/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") +}