From 2385e27cfd45e88f8d889ab54c6e708cfe1e4bca Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Wed, 23 Sep 2026 16:14:29 +0100 Subject: [PATCH] serve s3: keep the state remembered for each user in one place Refactor the state serve s3 keeps between requests - in-flight multipart uploads and object metadata - so that it is reached through a tenant struct for the user making the request. This makes it possible to keep users' state apart when they are serving different backends. There is only one tenant for now, so there is no change in behaviour. --- cmd/serve/s3/backend.go | 100 +++++++++++++++++++++++++-------- cmd/serve/s3/multipart.go | 53 ++++++++++------- cmd/serve/s3/multipart_test.go | 18 +++--- 3 files changed, 119 insertions(+), 52 deletions(-) diff --git a/cmd/serve/s3/backend.go b/cmd/serve/s3/backend.go index a7310c5b5..c4e959a96 100644 --- a/cmd/serve/s3/backend.go +++ b/cmd/serve/s3/backend.go @@ -9,6 +9,7 @@ import ( "math" "os" "path" + "slices" "strings" "sync" "sync/atomic" @@ -42,13 +43,12 @@ const putObjectPrefix = tempObjectPrefix + "put_" // PutStream, instead of being buffered in memory by gofakes3. type s3Backend struct { s *Server - meta *lru.Cache[string, map[string]string] // object metadata by path + meta *lru.Cache[metaKey, map[string]string] // object metadata of all tenants - // multipartUploads tracks in-flight streaming multipart uploads, - // keyed by gofakes3.UploadID. - multipartUploads sync.Map + tenantsMu sync.Mutex + tenants map[string]*tenant // per-user state by tenant ID - // uploads counts the entries in multipartUploads. + // uploads counts the in-flight uploads of all tenants. uploads atomic.Int64 // warnMaxUploadsOnce logs a single NOTICE the first time a multipart @@ -66,14 +66,34 @@ type s3Backend struct { reaperStop sync.Once } +// tenant is the state kept between requests for one user of the server. +// +// Anything remembered about a user's objects or uploads must be reached +// through their tenant, as different users may be serving different +// backends. +type tenant struct { + b *s3Backend + id string + + // uploads tracks in-flight streaming multipart uploads, + // keyed by gofakes3.UploadID. + uploads sync.Map +} + +// metaKey is the key of an object's metadata in s3Backend.meta. +type metaKey struct { + tenant string // tenant ID + fp string // object path +} + // newMetadataStore makes a store for the metadata of up to maxObjects // objects, forgetting that of the least recently used after that. If // maxObjects <= 0 there is no limit. -func newMetadataStore(maxObjects int) *lru.Cache[string, map[string]string] { +func newMetadataStore(maxObjects int) *lru.Cache[metaKey, map[string]string] { if maxObjects <= 0 { maxObjects = math.MaxInt } - meta, err := lru.New[string, map[string]string](maxObjects) + meta, err := lru.New[metaKey, map[string]string](maxObjects) if err != nil { panic(err) // only fails if maxObjects <= 0 } @@ -85,11 +105,47 @@ func newBackend(s *Server) *s3Backend { return &s3Backend{ s: s, meta: newMetadataStore(s.opt.MetadataMaxObjects), + tenants: map[string]*tenant{}, budget: newBufferBudget(int64(s.opt.MultipartStreamingBufferTotal)), reaperQuit: make(chan struct{}), } } +// tenant returns the state of the user making the request in ctx. +func (b *s3Backend) tenant(ctx context.Context) *tenant { + id := "" + b.tenantsMu.Lock() + defer b.tenantsMu.Unlock() + t, ok := b.tenants[id] + if !ok { + t = &tenant{b: b, id: id} + b.tenants[id] = t + } + return t +} + +// allTenants returns the state of every user seen so far. +func (b *s3Backend) allTenants() []*tenant { + b.tenantsMu.Lock() + defer b.tenantsMu.Unlock() + return slices.Collect(maps.Values(b.tenants)) +} + +// getMeta returns the stored metadata of the object at fp. +func (t *tenant) getMeta(fp string) (meta map[string]string, ok bool) { + return t.b.meta.Get(metaKey{tenant: t.id, fp: fp}) +} + +// setMeta stores meta as the metadata of the object at fp. +func (t *tenant) setMeta(fp string, meta map[string]string) { + t.b.meta.Add(metaKey{tenant: t.id, fp: fp}, meta) +} + +// removeMeta forgets the metadata of the object at fp. +func (t *tenant) removeMeta(fp string) { + t.b.meta.Remove(metaKey{tenant: t.id, fp: fp}) +} + // ListBuckets always returns the default bucket. func (b *s3Backend) ListBuckets(ctx context.Context) ([]gofakes3.BucketInfo, error) { _vfs, err := b.s.getVFS(ctx) @@ -203,7 +259,7 @@ func (b *s3Backend) HeadObject(ctx context.Context, bucketName, objectName strin "Content-Type": mimeType, } - if metaMap, ok := b.meta.Get(fp); ok { + if metaMap, ok := b.tenant(ctx).getMeta(fp); ok { maps.Copy(meta, metaMap) } @@ -283,7 +339,7 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string "Content-Type": mimeType, } - if metaMap, ok := b.meta.Get(fp); ok { + if metaMap, ok := b.tenant(ctx).getMeta(fp); ok { maps.Copy(meta, metaMap) } @@ -297,12 +353,12 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string }, nil } -// storeModtime sets both "mtime" and "X-Amz-Meta-Mtime" to val in b.meta. -// Call this whenever modtime is updated. -func (b *s3Backend) storeModtime(fp string, meta map[string]string, val string) { +// storeModtime sets both "mtime" and "X-Amz-Meta-Mtime" to val in the +// stored metadata. Call this whenever modtime is updated. +func (t *tenant) storeModtime(fp string, meta map[string]string, val string) { meta["X-Amz-Meta-Mtime"] = val meta["mtime"] = val - b.meta.Add(fp, meta) + t.setMeta(fp, meta) } // TouchObject creates or updates meta on specified object. @@ -328,12 +384,12 @@ func (b *s3Backend) TouchObject(ctx context.Context, fp string, meta map[string] return result, err } - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) if val, ok := meta["X-Amz-Meta-Mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -342,7 +398,7 @@ func (b *s3Backend) TouchObject(ctx context.Context, fp string, meta map[string] if val, ok := meta["mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -449,12 +505,12 @@ func (b *s3Backend) PutObject( return result, err } - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) if val, ok := meta["X-Amz-Meta-Mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -463,7 +519,7 @@ func (b *s3Backend) PutObject( if val, ok := meta["mtime"]; ok { ti, err := swift.FloatStringToTime(val) if err == nil { - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } // ignore error since the file is successfully created @@ -517,7 +573,7 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str if err := _vfs.Remove(fp); err != nil && !os.IsNotExist(err) { return err } - b.meta.Remove(fp) + b.tenant(ctx).removeMeta(fp) // FIXME: unsafe operation rmdirRecursive(fp, _vfs) @@ -592,7 +648,7 @@ func (b *s3Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket return result, gofakes3.KeyNotFound(srcKey) } if srcBucket == dstBucket && srcKey == dstKey { - b.meta.Add(fp, meta) + b.tenant(ctx).setMeta(fp, meta) val, ok := meta["X-Amz-Meta-Mtime"] if !ok { @@ -605,7 +661,7 @@ func (b *s3Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket if err != nil { return result, nil } - b.storeModtime(fp, meta, val) + b.tenant(ctx).storeModtime(fp, meta, val) return result, _vfs.Chtimes(fp, ti, ti) } diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index 47f863e67..1a833a677 100644 --- a/cmd/serve/s3/multipart.go +++ b/cmd/serve/s3/multipart.go @@ -61,8 +61,9 @@ type multipartUpload struct { streamFp string // path the parts are written to (fp when the remote has no server-side move or copy) meta map[string]string - fh io.WriteCloser // sink the in-order parts are written to, opened by sink (guarded by mu) - vfs *vfs.VFS // the VFS fh is created on, used for all later operations + fh io.WriteCloser // sink the in-order parts are written to, opened by sink (guarded by mu) + vfs *vfs.VFS // the VFS fh is created on, used for all later operations + tenant *tenant // the user the upload belongs to mu sync.Mutex changed chan struct{} // closed when buffered shrinks, nextPart advances or the upload closes @@ -177,9 +178,10 @@ func (up *multipartUpload) endActivity() { up.mu.Unlock() } -// loadUpload looks up an in-flight upload by ID. -func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, error) { - v, ok := b.multipartUploads.Load(uploadID) +// loadUpload looks up an in-flight upload of the user making the request +// in ctx by ID. +func (b *s3Backend) loadUpload(ctx context.Context, uploadID gofakes3.UploadID) (*multipartUpload, error) { + v, ok := b.tenant(ctx).uploads.Load(uploadID) if !ok { return nil, gofakes3.ErrNoSuchUpload } @@ -207,15 +209,16 @@ func (b *s3Backend) releaseUpload() { b.uploads.Add(-1) } -// addUpload records a new in-flight upload, which must hold a place +// addUpload records a new in-flight upload of t, which must hold a place // reserved by reserveUpload. -func (b *s3Backend) addUpload(uploadID gofakes3.UploadID, up *multipartUpload) { - b.multipartUploads.Store(uploadID, up) +func (b *s3Backend) addUpload(t *tenant, uploadID gofakes3.UploadID, up *multipartUpload) { + up.tenant = t + t.uploads.Store(uploadID, up) } // deleteUpload removes the record of an in-flight upload, if present. -func (b *s3Backend) deleteUpload(uploadID gofakes3.UploadID) { - if _, ok := b.multipartUploads.LoadAndDelete(uploadID); ok { +func (b *s3Backend) deleteUpload(uploadID gofakes3.UploadID, up *multipartUpload) { + if _, ok := up.tenant.uploads.LoadAndDelete(uploadID); ok { b.uploads.Add(-1) } } @@ -296,7 +299,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec up.budget = b.budget up.vfs = _vfs - b.addUpload(uploadID, up) + b.addUpload(b.tenant(ctx), uploadID, up) return uploadID, nil } @@ -307,7 +310,7 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec // the stream reaches it, and a re-upload of a part already streamed is just // hashed to check it matches. func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, partNumber int, contentLength int64, body io.Reader) (string, error) { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return "", err } @@ -832,7 +835,7 @@ func pipePart(w io.Writer, rw *pool.RW) error { // S3-style multipart ETag, and stores the user metadata so HeadObject and // GetObject see the same fields the in-memory PutObject path produces. func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, input *gofakes3.CompleteMultipartUploadRequest) (gofakes3.VersionID, string, error) { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return "", "", err } @@ -866,17 +869,17 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj return "", "", err } } - b.deleteUpload(uploadID) + b.deleteUpload(uploadID, up) - b.meta.Add(up.fp, up.meta) + up.tenant.setMeta(up.fp, up.meta) if val, ok := up.meta["X-Amz-Meta-Mtime"]; ok { if ti, err := swift.FloatStringToTime(val); err == nil { - b.storeModtime(up.fp, up.meta, val) + up.tenant.storeModtime(up.fp, up.meta, val) _ = up.vfs.Chtimes(up.fp, ti, ti) } } else if val, ok := up.meta["mtime"]; ok { if ti, err := swift.FloatStringToTime(val); err == nil { - b.storeModtime(up.fp, up.meta, val) + up.tenant.storeModtime(up.fp, up.meta, val) _ = up.vfs.Chtimes(up.fp, ti, ti) } } @@ -887,11 +890,11 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj // AbortMultipartUpload tears down an in-progress upload, discarding any data // already received. func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID) error { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(ctx, uploadID) if err != nil { return err } - defer b.deleteUpload(uploadID) + defer b.deleteUpload(uploadID, up) if err := up.abort(); err != nil { fs.Errorf(up.fp, "aborting multipart upload: %v", err) } @@ -910,7 +913,7 @@ func (b *s3Backend) failUpload(uploadID gofakes3.UploadID, up *multipartUpload) // forgetUpload aborts up, discards what it wrote and removes every record // of it, ours and gofakes3's. func (b *s3Backend) forgetUpload(uploadID gofakes3.UploadID, up *multipartUpload) { - b.deleteUpload(uploadID) + b.deleteUpload(uploadID, up) if err := up.abort(); err != nil { fs.Errorf(up.fp, "aborting multipart upload: %v", err) } @@ -947,7 +950,15 @@ func (b *s3Backend) stopReaper() { // reapExpiredUploads aborts and cleans up multipart uploads which have had // no request activity for longer than expiry. func (b *s3Backend) reapExpiredUploads(now time.Time, expiry time.Duration) { - b.multipartUploads.Range(func(key, value any) bool { + for _, t := range b.allTenants() { + b.reapExpiredTenantUploads(t, now, expiry) + } +} + +// reapExpiredTenantUploads aborts and cleans up the multipart uploads of +// t which have had no request activity for longer than expiry. +func (b *s3Backend) reapExpiredTenantUploads(t *tenant, now time.Time, expiry time.Duration) { + t.uploads.Range(func(key, value any) bool { uploadID := key.(gofakes3.UploadID) up := value.(*multipartUpload) up.mu.Lock() diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 89258f627..3085aac1f 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/cmd/serve/s3/multipart_test.go @@ -476,7 +476,7 @@ func TestMultipartBufferLimit(t *testing.T) { // CreateMultipartUpload does. func storeTestUpload(t *testing.T, b *s3Backend, uploadID gofakes3.UploadID, up *multipartUpload) { require.NoError(t, b.reserveUpload()) - b.addUpload(uploadID, up) + b.addUpload(b.tenant(context.Background()), uploadID, up) } // stubSink is a multipartUpload sink which records how it was closed. @@ -627,7 +627,7 @@ func TestMultipartCompleteRenameFailureKeepsUpload(t *testing.T) { _, _, err = b.CompleteMultipartUpload(ctx, bucket, "key", uploadID, &gofakes3.CompleteMultipartUploadRequest{}) require.Error(t, err) - _, err = b.loadUpload(uploadID) + _, err = b.loadUpload(context.Background(), uploadID) require.NoError(t, err, "the upload record must survive a retryable Complete failure") } @@ -728,11 +728,11 @@ func TestMultipartReapExpiredUploads(t *testing.T) { b.reapExpiredUploads(now, expiry) - _, err = b.loadUpload("fresh") + _, err = b.loadUpload(context.Background(), "fresh") assert.NoError(t, err, "a fresh upload must not be reaped") - _, err = b.loadUpload("busy") + _, err = b.loadUpload(context.Background(), "busy") assert.NoError(t, err, "an upload with a request in flight must not be reaped") - _, err = b.loadUpload("idle") + _, err = b.loadUpload(context.Background(), "idle") assert.ErrorIs(t, err, gofakes3.ErrNoSuchUpload, "an idle upload past the expiry must be reaped") // The reaped upload was aborted, not committed. @@ -1274,7 +1274,7 @@ func (s *bufferSink) Bytes() []byte { func newBufferSinkUpload(t *testing.T, b *s3Backend, bucket, object string) (gofakes3.UploadID, *bufferSink) { uploadID, err := b.CreateMultipartUpload(context.Background(), bucket, object, nil) require.NoError(t, err) - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(context.Background(), uploadID) require.NoError(t, err) if up.fh != nil { // This fails the VFS upload, returning errBoom @@ -1467,7 +1467,7 @@ func TestMultipartPartFailsAfterWholeBody(t *testing.T) { // a part which could never be accepted. _, err = b.UploadPart(ctx, bucket, object, uploadID, 1, int64(len(part1)), bytes.NewReader(part1)) require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) - _, err = b.loadUpload(uploadID) + _, err = b.loadUpload(context.Background(), uploadID) require.ErrorIs(t, err, gofakes3.ErrNoSuchUpload) } @@ -1582,7 +1582,7 @@ func TestMultipartSinkOpenedOnFirstPart(t *testing.T) { ctx := context.Background() sinkOpen := func(uploadID gofakes3.UploadID) bool { - up, err := b.loadUpload(uploadID) + up, err := b.loadUpload(context.Background(), uploadID) require.NoError(t, err) up.mu.Lock() defer up.mu.Unlock() @@ -1661,7 +1661,7 @@ func TestMultipartMaxUploads(t *testing.T) { assert.ErrorIs(t, err, vfs.ENOENT, "a refused upload created directories") // And so does expiring one. - up, err := b.loadUpload(id3) + up, err := b.loadUpload(context.Background(), id3) require.NoError(t, err) up.mu.Lock() up.lastUsed = time.Now().Add(-2 * time.Hour)