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

@@ -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)

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)

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 {

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

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

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")

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