s3tables: create a missing directory exclusively in ensureDirectory

ensureDirectory looked the path up and, when it was missing, created it with a
plain CreateEntry. A table create that lost the lookup race then replaced the
directory another create had made, claimed and confirmed since, erasing that
table's record, so a third create could claim the name and win as well.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

weed/s3api/s3tables/filer_ops.go

@@ -72,7 +72,9 @@ func (h *S3TablesHandler) createDirectoryEntry(ctx context.Context, client filer
 	})
 }
 
-// ensureDirectory ensures a directory exists at the specified path
+// ensureDirectory ensures a directory exists at the specified path. The create
+// is exclusive: a directory another writer made since the lookup may already
+// carry a record, which a plain create would replace with a bare directory.
 func (h *S3TablesHandler) ensureDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
 	dir, name := splitPath(path)
 	_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
@@ -83,7 +85,9 @@ func (h *S3TablesHandler) ensureDirectory(ctx context.Context, client filer_pb.S
 		return nil
 	}
 	if errors.Is(err, filer_pb.ErrNotFound) {
-		return h.createDirectory(ctx, client, path)
+		if err = h.createDirectoryExclusive(ctx, client, path, nil); errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+			return nil
+		}
 	}
 	return err
 }

weed/s3api/s3tables/filer_ops_race_test.go

@@ -87,3 +87,16 @@ func TestCreateNamespaceLostRaceKeepsTheFirstCreate(t *testing.T) {
 	require.NotNil(t, survivor)
 	assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata]))
 }
+
+func TestEnsureDirectoryKeepsADirectoryCreatedSinceTheLookup(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	tablePath := GetTablePath("bucket", "ns", "orders")
+	dir, name := splitPath(tablePath)
+	client := &competingCreateClient{SeaweedFilerClient: fs.Client, f: fs, dir: dir, name: name, plant: true}
+
+	require.NoError(t, NewS3TablesHandler().ensureDirectory(context.Background(), client, tablePath))
+
+	survivor := fs.Get(dir, name)
+	require.NotNil(t, survivor)
+	assert.Equal(t, `{"OwnerAccountID":"competitor"}`, string(survivor.Extended[ExtendedKeyMetadata]))
+}