From a724f4296c835667468db9312044cf2dd56b29b0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 19:01:08 -0700 Subject: [PATCH] s3, s3tables: create buckets and namespaces exclusively (#11872) * filer_pb: add MkdirExclusive A taken name leaves the callback as a result, not an error, so filer failover does not count a lost race as a filer failure. * s3tables: create the entry with its attributes in one exclusive create CreateTableBucket and CreateNamespace checked that the name was free and then wrote the entry in several calls, so a second create in the window replaced the first caller's entry including its owner. The create now carries the marker, metadata and tags in one O_EXCL CreateEntry that answers ErrEntryAlreadyExists with the same 409. * s3: PutBucket uses an exclusive mkdir The same window let a plain PutBucket replace a table bucket created concurrently. The already-exists branch now answers from the entry itself, since the registry just cached the name as absent, and the admin auto-create path is exclusive too. * s3: read the winning entry when a create loses the race The concurrent-create branch called existingBucketError on a name the registry had just cached as absent, answering NoSuchBucket, and the auto-create fallback could assign an S3 owner to a table bucket it met there. The negative cache is cleared before the existing-bucket error is built, and a table bucket keeps its owner-less shape. * s3: fail an auto-create that met a table bucket A table bucket carries no S3 owner and must not gain one, and the upload that auto-create ran for must not continue into it either. * s3: fail an auto-create when the winning entry cannot be read An unreadable existing entry fell through to success, so the write continued into an unvalidated entry, possibly a table bucket. --- weed/pb/filer_pb/filer_client.go | 25 ++++++ weed/s3api/filer_util.go | 18 ++++ weed/s3api/s3api_bucket_handlers.go | 51 +++++++---- weed/s3api/s3tables/filer_ops.go | 14 +++ weed/s3api/s3tables/filer_ops_race_test.go | 89 ++++++++++++++++++++ weed/s3api/s3tables/handler_bucket_create.go | 35 ++++---- weed/s3api/s3tables/handler_namespace.go | 24 +++--- 7 files changed, 212 insertions(+), 44 deletions(-) create mode 100644 weed/s3api/s3tables/filer_ops_race_test.go diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go index ffd11e93e7..a3903cff48 100644 --- a/weed/pb/filer_pb/filer_client.go +++ b/weed/pb/filer_pb/filer_client.go @@ -244,7 +244,31 @@ func Mkdir(ctx context.Context, filerClient FilerClient, parentDirectoryPath str }) } +// MkdirExclusive is Mkdir that never replaces an entry already at the path: +// the create is O_EXCL and a taken name answers ErrEntryAlreadyExists. The +// taken name leaves the callback as a result, not an error: failover counts a +// callback error against the filer and retries on the next one. +func MkdirExclusive(ctx context.Context, filerClient FilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error { + var taken error + err := filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error { + err := doMkdir(ctx, client, parentDirectoryPath, dirName, true, fn) + if errors.Is(err, ErrEntryAlreadyExists) { + taken = err + return nil + } + return err + }) + if err == nil { + err = taken + } + return err +} + func DoMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error { + return doMkdir(ctx, client, parentDirectoryPath, dirName, false, fn) +} + +func doMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath string, dirName string, exclusive bool, fn func(entry *Entry)) error { entry := &Entry{ Name: dirName, IsDirectory: true, @@ -264,6 +288,7 @@ func DoMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath request := &CreateEntryRequest{ Directory: parentDirectoryPath, Entry: entry, + OExcl: exclusive, } glog.V(1).InfofCtx(ctx, "mkdir: %v", request) diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go index 240e3b18ff..5ebffe8f83 100644 --- a/weed/s3api/filer_util.go +++ b/weed/s3api/filer_util.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -22,6 +23,23 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun } +// mkdirExclusive never replaces an entry already at the path: PutBucket uses +// it so a plain bucket cannot replace a table bucket created in its +// check-to-create window. +func (s3a *S3ApiServer) mkdirExclusive(parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error { + + return filer_pb.MkdirExclusive(context.Background(), s3a, parentDirectoryPath, dirName, fn) + +} + +// concurrentEntryIsForeign: after PutBucket's exclusive mkdir met an entry, +// whether that entry must be answered BucketAlreadyExists without looking at +// its owner: a table bucket (it carries no S3 owner and must never read as +// "owned by you"), or an entry that could not be read. +func concurrentEntryIsForeign(entry *filer_pb.Entry, err error) bool { + return err != nil || entry == nil || s3tables.IsTableBucketEntry(entry) +} + func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk, fn func(entry *filer_pb.Entry)) error { err := filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn) diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go index 1a3c756412..0f74211e65 100644 --- a/weed/s3api/s3api_bucket_handlers.go +++ b/weed/s3api/s3api_bucket_handlers.go @@ -330,7 +330,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) // This ensures Object Lock configuration is set in the same CreateEntry call, // preventing race conditions where the bucket exists without Object Lock enabled var bucketCrtime int64 - if err := s3a.mkdir(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) { + if err := s3a.mkdirExclusive(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) { bucketCrtime = entry.Attributes.Crtime // Set bucket owner @@ -373,6 +373,16 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) // return the appropriate already-exists error instead of InternalError. if exist, checkErr := s3a.exists(s3a.option.BucketsPath, bucket, true); checkErr == nil && exist { glog.V(3).Infof("PutBucketHandler: bucket %s was created concurrently", bucket) + // From the entry itself, not the bucket registry: the pre-check + // above has just cached this name as not found, so in the race this + // branch exists for the registry would still say "no table bucket". + if concurrentEntryIsForeign(s3a.getEntry(s3a.option.BucketsPath, bucket)) { + s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists) + return + } + if s3a.bucketConfigCache != nil { + s3a.bucketConfigCache.RemoveNegativeCache(bucket) + } errCode := s3a.existingBucketError(r, bucket, currentIdentityId, requestHasACL) if errCode == s3err.ErrBucketAlreadyOwnedByYou { s3a.healBucketOwnerIndex(bucket, currentIdentityId) @@ -744,7 +754,7 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error { identityId := s3_constants.GetIdentityNameFromContext(r) var bucketCrtime int64 - if err := s3a.mkdir(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) { + if err := s3a.mkdirExclusive(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) { bucketCrtime = entry.Attributes.Crtime setBucketOwner(r)(entry) }); err != nil { @@ -754,26 +764,35 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error { glog.Warningf("autoCreateBucket: failed to check existence for bucket %s: %v", bucket, err2) return fmt.Errorf("failed to auto-create bucket %s: %w", bucket, errors.Join(err, err2)) } else if exist { + entry, getErr := s3a.getEntry(s3a.option.BucketsPath, bucket) + if getErr != nil { + glog.Warningf("autoCreateBucket: failed to get entry for existing bucket %s: %v", bucket, getErr) + return fmt.Errorf("failed to auto-create bucket %s: %w", bucket, errors.Join(err, getErr)) + } + if concurrentEntryIsForeign(entry, nil) { + // A table bucket carries no S3 owner and must not gain one — + // or a write intended for a missing plain bucket. + if s3a.bucketConfigCache != nil { + s3a.bucketConfigCache.RemoveNegativeCache(bucket) + } + return fmt.Errorf("auto-create bucket %s: name taken by a table bucket: %w", bucket, err) + } // The bucket exists, which is fine. However, we should ensure it has an owner. // If it was created by a concurrent request that didn't set an owner, // we'll set it here to ensure consistency. - if entry, getErr := s3a.getEntry(s3a.option.BucketsPath, bucket); getErr == nil { - if entry.Extended == nil || len(entry.Extended[s3_constants.AmzIdentityId]) == 0 { - // No owner set, assign current admin as owner - setBucketOwner(r)(entry) - if updateErr := s3a.updateEntry(s3a.option.BucketsPath, entry); updateErr != nil { - glog.Warningf("autoCreateBucket: failed to set owner for existing bucket %s: %v", bucket, updateErr) - } else { - glog.V(1).Infof("Set owner for existing bucket %s (created by concurrent request)", bucket) - if owner := bucketEntryOwner(entry); owner != "" { - if indexErr := s3a.addBucketToOwnerIndex(owner, bucket, entry.Attributes.Crtime); indexErr != nil { - glog.Warningf("autoCreateBucket: owner index add %s/%s: %v", owner, bucket, indexErr) - } + if entry.Extended == nil || len(entry.Extended[s3_constants.AmzIdentityId]) == 0 { + // No owner set, assign current admin as owner + setBucketOwner(r)(entry) + if updateErr := s3a.updateEntry(s3a.option.BucketsPath, entry); updateErr != nil { + glog.Warningf("autoCreateBucket: failed to set owner for existing bucket %s: %v", bucket, updateErr) + } else { + glog.V(1).Infof("Set owner for existing bucket %s (created by concurrent request)", bucket) + if owner := bucketEntryOwner(entry); owner != "" { + if indexErr := s3a.addBucketToOwnerIndex(owner, bucket, entry.Attributes.Crtime); indexErr != nil { + glog.Warningf("autoCreateBucket: owner index add %s/%s: %v", owner, bucket, indexErr) } } } - } else { - glog.Warningf("autoCreateBucket: failed to get entry for existing bucket %s: %v", bucket, getErr) } // Remove bucket from negative cache — it exists now if s3a.bucketConfigCache != nil { diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 87ead2af4a..4ed73cbc39 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -27,10 +27,23 @@ const extendedAttributeUpdateAttempts = 5 // createDirectory creates a new directory at the specified path func (h *S3TablesHandler) createDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error { + return h.createDirectoryEntry(ctx, client, path, false, nil) +} + +// createDirectoryExclusive creates a directory entry together with its +// extended attributes in one CreateEntry that fails with +// filer_pb.ErrEntryAlreadyExists when an entry is already there: nothing is +// replaced and no reader sees the directory without its attributes. +func (h *S3TablesHandler) createDirectoryExclusive(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, extended map[string][]byte) error { + return h.createDirectoryEntry(ctx, client, path, true, extended) +} + +func (h *S3TablesHandler) createDirectoryEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, exclusive bool, extended map[string][]byte) error { dir, name := splitPath(path) now := time.Now().Unix() return filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{ Directory: dir, + OExcl: exclusive, Entry: &filer_pb.Entry{ Name: name, IsDirectory: true, @@ -39,6 +52,7 @@ func (h *S3TablesHandler) createDirectory(ctx context.Context, client filer_pb.S Crtime: now, FileMode: uint32(0755 | os.ModeDir), }, + Extended: extended, }, }) } diff --git a/weed/s3api/s3tables/filer_ops_race_test.go b/weed/s3api/s3tables/filer_ops_race_test.go new file mode 100644 index 0000000000..a1f6533862 --- /dev/null +++ b/weed/s3api/s3tables/filer_ops_race_test.go @@ -0,0 +1,89 @@ +package s3tables + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +// competingCreateClient lands a foreign entry where the handler's existence +// check just answered "not there": the create that follows must lose to it +// rather than replace it, the way a second caller's create wins the window in +// a check-then-create race. +type competingCreateClient struct { + filer_pb.SeaweedFilerClient + f *s3tablestest.MemFiler + dir string + name string + plant bool +} + +func (c *competingCreateClient) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + resp, err := c.SeaweedFilerClient.LookupDirectoryEntry(ctx, req, opts...) + if c.plant && err != nil && req.Directory == c.dir && req.Name == c.name { + c.plant = false + c.f.Put(c.dir, c.name, map[string][]byte{ + ExtendedKeyTableBucket: []byte("true"), + ExtendedKeyMetadata: []byte(`{"OwnerAccountID":"competitor"}`), + }) + } + return resp, err +} + +func runHandler(t *testing.T, m *Manager, client filer_pb.SeaweedFilerClient, operation string, input interface{}) (int, S3TablesError) { + t.Helper() + body, err := json.Marshal(input) + require.NoError(t, err) + req, err := newManagerRequest(context.Background(), operation, body, "admin") + require.NoError(t, err) + recorder := httptest.NewRecorder() + m.handler.HandleRequest(recorder, req, NewManagerClient(client)) + result := recorder.Result() + defer result.Body.Close() + var response S3TablesError + json.NewDecoder(result.Body).Decode(&response) + return result.StatusCode, response +} + +func TestCreateTableBucketLostRaceKeepsTheFirstCreate(t *testing.T) { + fs := s3tablestest.Start(t) + m := NewManager() + client := &competingCreateClient{SeaweedFilerClient: fs.Client, f: fs, dir: TablesPath, name: "bucket", plant: true} + + status, resp := runHandler(t, m, client, "CreateTableBucket", &CreateTableBucketRequest{Name: "bucket"}) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeBucketAlreadyExists, resp.Type) + + survivor := fs.Get(TablesPath, "bucket") + require.NotNil(t, survivor) + assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata])) +} + +func TestCreateNamespaceLostRaceKeepsTheFirstCreate(t *testing.T) { + fs := s3tablestest.Start(t) + m := NewManager() + bucketMetadata, _ := json.Marshal(&tableBucketMetadata{Name: "bucket", OwnerAccountID: DefaultAccountID}) + fs.Put(TablesPath, "bucket", map[string][]byte{ExtendedKeyTableBucket: []byte("true"), ExtendedKeyMetadata: bucketMetadata}) + + arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, "bucket") + require.NoError(t, err) + client := &competingCreateClient{SeaweedFilerClient: fs.Client, f: fs, dir: GetTableBucketPath("bucket"), name: "ns", plant: true} + status, resp := runHandler(t, m, client, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: arn, + Namespace: []string{"ns"}, + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeNamespaceAlreadyExists, resp.Type) + + survivor := fs.Get(GetTableBucketPath("bucket"), "ns") + require.NotNil(t, survivor) + assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata])) +} diff --git a/weed/s3api/s3tables/handler_bucket_create.go b/weed/s3api/s3tables/handler_bucket_create.go index acf34182b5..3ecd0f83d1 100644 --- a/weed/s3api/s3tables/handler_bucket_create.go +++ b/weed/s3api/s3tables/handler_bucket_create.go @@ -140,6 +140,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http return err } + var taken error err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Ensure root tables directory exists if !h.entryExists(r.Context(), client, TablesPath) { @@ -148,35 +149,35 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http } } - // Create bucket directory - if err := h.createDirectory(r.Context(), client, bucketPath); err != nil { - return err + extended := map[string][]byte{ + ExtendedKeyTableBucket: []byte("true"), + ExtendedKeyMetadata: metadataBytes, } - - // Mark as a table bucket - if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyTableBucket, []byte("true")); err != nil { - return err - } - - // Set metadata as extended attribute - if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata, metadataBytes); err != nil { - return err - } - - // Set tags if provided if len(req.Tags) > 0 { tagsBytes, err := json.Marshal(req.Tags) if err != nil { return fmt.Errorf("failed to marshal tags: %w", err) } - if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyTags, tagsBytes); err != nil { - return err + extended[ExtendedKeyTags] = tagsBytes + } + if err := h.createDirectoryExclusive(r.Context(), client, bucketPath, extended); err != nil { + if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { + taken = err // an answer, not a filer failure + return nil } + return err } return nil }) + if err == nil && taken != nil { + err = taken + } + if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { + h.writeError(w, http.StatusConflict, ErrCodeBucketAlreadyExists, fmt.Sprintf("table bucket %s already exists", req.Name)) + return err + } if err != nil { glog.Errorf("S3Tables: failed to create table bucket %s: %v", req.Name, err) h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create table bucket") diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index 60a2e942bb..8e6a7ee6d5 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -197,20 +197,22 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R return fmt.Errorf("failed to marshal metadata: %w", err) } + var taken error err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - // Create namespace directory - if err := h.createDirectory(r.Context(), client, namespacePath); err != nil { - return err + err := h.createDirectoryExclusive(r.Context(), client, namespacePath, map[string][]byte{ExtendedKeyMetadata: metadataBytes}) + if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { + taken = err // an answer, not a filer failure + return nil } - - // Set metadata as extended attribute - if err := h.setExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata, metadataBytes); err != nil { - return err - } - - return nil + return err }) - + if err == nil && taken != nil { + err = taken + } + if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { + h.writeError(w, http.StatusConflict, ErrCodeNamespaceAlreadyExists, fmt.Sprintf("namespace %s already exists", namespaceName)) + return err + } if err != nil { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create namespace") return err