iceberg: make a table commit a compare-and-swap (#10775)

* iceberg: make a table commit a compare-and-swap

The catalog validated the caller's version token, ran its authorization
checks, and only then wrote the new metadata xattr. Two engines
committing against the same base both passed that check and both wrote,
so the second silently dropped the first one's snapshot. Both also derive
the same v{N}.metadata.json name and the file write overwrote, leaving
the surviving pointer aimed at the loser's metadata - and the loser's
conflict cleanup then deleted the winner's file.

Write the metadata file with an exclusive create and update the xattr
conditionally on the bytes the handler read, the way the maintenance
worker already commits. A writer that lost the race re-reads and retries,
and reports 409 CommitFailedException once out of attempts.

* iceberg: stage a commit under a unique name when the versioned one is taken

Two follow-ups from review of the commit compare-and-swap:

Refusing to overwrite v{N}.metadata.json also refused to get past a file
left behind by a commit that died between staging and updating the
pointer. Every later commit derived the same name, saw the collision, and
reported a conflict, so the table stayed uncommittable until an orphan
sweep removed the file. Stage under v{N}-{uuid} instead: neither writer's
file is overwritten and the catalog pointer still decides who won, which
is how the maintenance worker has always staged its own metadata.
metadataVersionFromLocation learned to read the version back out of that
name.

The conditional update guarded only the metadata attribute while the
write replaced the whole entry, so a policy or tag written in the same
window was silently reverted. Guard every catalog attribute, which turns
that into a conflict the caller retries on fresh state.

* iceberg: give saveMetadataFile the exclusive flag instead of a second name

saveNewMetadataFile, saveMetadataBlobExclusive and uniqueMetadataFileName
were three new names around one existing helper. The flag now rides on
saveMetadataFile and saveMetadataBlob, and the unique-name construction
sits where it is used.

* iceberg: reuse the filer CAS helpers #10773 added, and stage transactions exclusively

#10773 landed mutateEntryExtended, which already writes an entry back under a
whole-entry precondition and retries. Drop the helper this branch added and
route the table commit through it: the check that the metadata is still the
one this request read now lives in the mutation, where it sees current state.

The policy the request was authorized against is asserted too, so an
administrator restricting it mid-commit sends the caller back through
authorization instead of having a stale decision applied. Bucket and
namespace policies live on other entries and a single-entry precondition
cannot cover them.

Multi-table transactions stage their metadata exclusively for the same
reason single-table commits do, and carry the name they landed on into the
pointer flip.

weed/s3api/iceberg/commit_helpers.go

@@ -5,8 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math/rand/v2"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
@@ -17,6 +19,36 @@ import (
 
 const requirementAssertCreate = "assert-create"
 
+// sleepBeforeCommitRetry backs off between commit attempts, jittered so that
+// writers that collided do not line up again on the next try.
+func sleepBeforeCommitRetry(attempt int) {
+	jitter := time.Duration(rand.Int64N(int64(25 * time.Millisecond)))
+	time.Sleep(time.Duration(50*attempt)*time.Millisecond + jitter)
+}
+
+// stageCommitMetadata writes the metadata file a commit will point the catalog
+// at, returning the name and location it landed on. The exclusive create keeps
+// one writer from overwriting another's file; when the name is already taken
+// the metadata goes to a unique one instead of failing, because the file there
+// may be an orphan left by an interrupted commit, and refusing would wedge the
+// table until an orphan sweep ran. The catalog pointer, not the file name,
+// decides which commit won. This mirrors how the maintenance worker stages its
+// own metadata.
+func (s *Server) stageCommitMetadata(ctx context.Context, metadataBucket, metadataPath, location, metadataFileName string, metadataBytes []byte) (string, string, error) {
+	err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, true)
+	if errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+		// v{N}-{uuid}, a form both the catalog and the maintenance worker
+		// still read the version from.
+		metadataFileName = fmt.Sprintf("%s-%s.metadata.json", strings.TrimSuffix(metadataFileName, ".metadata.json"), uuid.NewString())
+		glog.V(1).Infof("Iceberg: metadata file already staged for %s, using %s", location, metadataFileName)
+		err = s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, true)
+	}
+	if err != nil {
+		return "", "", err
+	}
+	return metadataFileName, fmt.Sprintf("%s/metadata/%s", strings.TrimSuffix(location, "/"), metadataFileName), nil
+}
+
 type icebergRequestError struct {
 	status  int
 	errType string
@@ -155,7 +187,7 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi
 			message: "Invalid table location: " + err.Error(),
 		}
 	}
-	if err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
+	if err := s.saveMetadataFile(ctx, metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil {
 		return nil, &icebergRequestError{
 			status:  http.StatusInternalServerError,
 			errType: "InternalServerError",

weed/s3api/iceberg/handlers_commit.go

@@ -4,11 +4,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math/rand/v2"
 	"net/http"
 	"path"
 	"strings"
-	"time"
 
 	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
@@ -311,7 +309,8 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid table location: "+err.Error())
 			return
 		}
-		if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
+		metadataFileName, newMetadataLocation, err = s.stageCommitMetadata(r.Context(), metadataBucket, metadataPath, location, metadataFileName, metadataBytes)
+		if err != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error())
 			return
 		}
@@ -350,8 +349,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			}
 			if attempt < maxCommitAttempts {
 				glog.V(1).Infof("Iceberg: CommitTable conflict for %s (attempt %d/%d), retrying", tableName, attempt, maxCommitAttempts)
-				jitter := time.Duration(rand.Int64N(int64(25 * time.Millisecond)))
-				time.Sleep(time.Duration(50*attempt)*time.Millisecond + jitter)
+				sleepBeforeCommitRetry(attempt)
 				continue
 			}
 			writeError(w, http.StatusConflict, "CommitFailedException", "Version token mismatch")

weed/s3api/iceberg/handlers_table.go

@@ -265,7 +265,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 	// Stage-create persists metadata in the internal staged area and skips S3Tables registration.
 	if req.StageCreate {
 		stagedTablePath := stageCreateStagedTablePath(namespace, tableName, tableUUID)
-		if err := s.saveMetadataFile(r.Context(), metadataBucket, stagedTablePath, metadataFileName, metadataBytes); err != nil {
+		if err := s.saveMetadataFile(r.Context(), metadataBucket, stagedTablePath, metadataFileName, metadataBytes, false); err != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save staged metadata file: "+err.Error())
 			return
 		}
@@ -281,7 +281,7 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		writeLoadResult(w, http.StatusOK, result)
 		return
 	}
-	if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
+	if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil {
 		writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error())
 		return
 	}

weed/s3api/iceberg/handlers_transaction.go

@@ -36,6 +36,7 @@ type preparedTableCommit struct {
 	versionToken     string
 	metadataBucket   string
 	metadataPath     string
+	location         string
 	metadataFileName string
 	metadataBytes    []byte
 	metadataVersion  int
@@ -81,15 +82,21 @@ func (s *Server) handleCommitTransaction(w http.ResponseWriter, r *http.Request)
 		prepared = append(prepared, *pc)
 	}
 
-	// Phase 2: write each new metadata.json object.
+	// Phase 2: write each new metadata.json object. Staging is exclusive for the
+	// same reason a single-table commit stages exclusively: a transaction racing
+	// another writer on one of its tables would otherwise overwrite that
+	// writer's metadata, and the pointer flip below decides who won.
 	for i := range prepared {
 		pc := &prepared[i]
-		if err := s.saveMetadataFile(r.Context(), pc.metadataBucket, pc.metadataPath, pc.metadataFileName, pc.metadataBytes); err != nil {
+		fileName, location, err := s.stageCommitMetadata(r.Context(), pc.metadataBucket, pc.metadataPath, pc.location, pc.metadataFileName, pc.metadataBytes)
+		if err != nil {
 			// No pointer flipped yet, so every written file is safe to delete.
 			s.cleanupPreparedMetadata(r.Context(), prepared[:i+1], nil)
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save metadata file: "+err.Error())
 			return
 		}
+		pc.metadataFileName = fileName
+		pc.newMetadataLoc = location
 	}
 
 	// Phase 3: flip each table's pointer xattr; on failure roll back prior flips.
@@ -208,6 +215,7 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN,
 		versionToken:        getResp.VersionToken,
 		metadataBucket:      metadataBucket,
 		metadataPath:        metadataPath,
+		location:            location,
 		metadataFileName:    metadataFileName,
 		metadataBytes:       metadataBytes,
 		metadataVersion:     metadataVersion,

weed/s3api/iceberg/handlers_view.go

@@ -204,7 +204,7 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
 
 	// Persist the metadata file only after the catalog registers the view, so a
 	// missing namespace or name collision fails before any bytes hit storage.
-	if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
+	if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes, false); err != nil {
 		// Roll back the registered view so it doesn't linger pointing at metadata
 		// that was never written.
 		if dropErr := s.dropView(r, namespace, req.Name); dropErr != nil {

weed/s3api/iceberg/handlers_view_update.go

@@ -3,10 +3,8 @@ package iceberg
 import (
 	"encoding/json"
 	"fmt"
-	"math/rand/v2"
 	"net/http"
 	"strings"
-	"time"
 
 	"github.com/apache/iceberg-go/view"
 	"github.com/google/uuid"
@@ -104,7 +102,8 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Invalid view location: "+err.Error())
 			return
 		}
-		if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
+		metadataFileName, newMetadataLocation, err = s.stageCommitMetadata(r.Context(), metadataBucket, metadataPath, location, metadataFileName, metadataBytes)
+		if err != nil {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to save view metadata file: "+err.Error())
 			return
 		}
@@ -144,8 +143,7 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
 			}
 			if attempt < maxCommitAttempts {
 				glog.V(1).Infof("Iceberg: UpdateView conflict for %s (attempt %d/%d), retrying", viewName, attempt, maxCommitAttempts)
-				jitter := time.Duration(rand.Int64N(int64(25 * time.Millisecond)))
-				time.Sleep(time.Duration(50*attempt)*time.Millisecond + jitter)
+				sleepBeforeCommitRetry(attempt)
 				continue
 			}
 			writeError(w, http.StatusConflict, "CommitFailedException", "Version token mismatch")

weed/s3api/iceberg/manifest_repair.go

@@ -556,7 +556,7 @@ func (st *serverManifestStore) saveFile(ctx context.Context, location string, da
 	if err != nil {
 		return err
 	}
-	return st.server.saveMetadataBlob(ctx, bucket, tablePath, fileName, data, "application/avro")
+	return st.server.saveMetadataBlob(ctx, bucket, tablePath, fileName, data, "application/avro", false)
 }
 
 // repairAddSnapshotManifests is the server entry point used by the commit

weed/s3api/iceberg/metadata_files.go

@@ -22,13 +22,17 @@ func metadataDirPath(bucketName, tablePath string) string {
 }
 
 // saveMetadataFile saves the Iceberg metadata JSON file to the filer.
-// It constructs the filer path from the S3 location components.
-func (s *Server) saveMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte) error {
-	return s.saveMetadataBlob(ctx, bucketName, tablePath, metadataFileName, content, "application/json")
+// It constructs the filer path from the S3 location components. With exclusive
+// set the write fails with filer_pb.ErrEntryAlreadyExists rather than replacing
+// a file that is already there: two commits racing off the same base pick the
+// same v{N} name, and letting the second overwrite the first would leave the
+// winner's catalog pointer aimed at the loser's metadata.
+func (s *Server) saveMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, exclusive bool) error {
+	return s.saveMetadataBlob(ctx, bucketName, tablePath, metadataFileName, content, "application/json", exclusive)
 }
 
 // saveMetadataBlob saves a file into the table's metadata directory.
-func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, mimeType string) error {
+func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, mimeType string, exclusive bool) error {
 
 	// Create context with timeout for file operations
 	opCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
@@ -101,6 +105,7 @@ func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, me
 		// 4. Write the file
 		resp, err := client.CreateEntry(opCtx, &filer_pb.CreateEntryRequest{
 			Directory: metadataDir,
+			OExcl:     exclusive,
 			Entry: &filer_pb.Entry{
 				Name: metadataFileName,
 				Attributes: &filer_pb.FuseAttributes{

weed/s3api/iceberg/metadata_files_test.go

@@ -0,0 +1,128 @@
+package iceberg
+
+import (
+	"context"
+	"errors"
+	"path"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
+)
+
+// stubFilerClient answers only the calls the metadata writers make, tracking
+// which paths exist so exclusive creates can be exercised.
+type stubFilerClient struct {
+	filer_pb.SeaweedFilerClient
+	entries map[string][]byte
+}
+
+func newStubFilerClient() *stubFilerClient {
+	return &stubFilerClient{entries: make(map[string][]byte)}
+}
+
+func (c *stubFilerClient) WithFilerClient(_ bool, fn func(client filer_pb.SeaweedFilerClient) error) error {
+	return fn(c)
+}
+
+func (c *stubFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	key := path.Join(req.Directory, req.Name)
+	if _, ok := c.entries[key]; !ok {
+		return nil, filer_pb.ErrNotFound
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{Name: req.Name}}, nil
+}
+
+func (c *stubFilerClient) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest, _ ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) {
+	key := path.Join(req.Directory, req.Entry.Name)
+	if _, exists := c.entries[key]; exists && req.OExcl {
+		return &filer_pb.CreateEntryResponse{
+			Error:     "entry already exists",
+			ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS,
+		}, nil
+	}
+	c.entries[key] = req.Entry.Content
+	return &filer_pb.CreateEntryResponse{}, nil
+}
+
+// Two commits racing off the same base metadata pick the same v{N} file name.
+// The loser must be told, not allowed to replace the winner's metadata.
+func TestSaveMetadataFileExclusiveRefusesToOverwrite(t *testing.T) {
+	client := newStubFilerClient()
+	s := &Server{filerClient: client}
+	ctx := context.Background()
+
+	if err := s.saveMetadataFile(ctx, "bkt", "ns/tbl", "v2.metadata.json", []byte(`{"winner":true}`), true); err != nil {
+		t.Fatalf("first write failed: %v", err)
+	}
+
+	err := s.saveMetadataFile(ctx, "bkt", "ns/tbl", "v2.metadata.json", []byte(`{"loser":true}`), true)
+	if !errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+		t.Fatalf("second write err = %v, want ErrEntryAlreadyExists", err)
+	}
+
+	stored := string(client.entries[path.Join(metadataDirPath("bkt", "ns/tbl"), "v2.metadata.json")])
+	if stored != `{"winner":true}` {
+		t.Errorf("stored metadata = %s, want the first writer's content", stored)
+	}
+}
+
+// A metadata file left behind by an interrupted commit must not wedge the
+// table: the next commit stages under a unique name rather than failing or
+// overwriting, and the catalog pointer still decides the winner.
+func TestStageCommitMetadataFallsBackToAUniqueName(t *testing.T) {
+	client := newStubFilerClient()
+	s := &Server{filerClient: client}
+	ctx := context.Background()
+
+	name, location, err := s.stageCommitMetadata(ctx, "bkt", "ns/tbl", "s3://bkt/ns/tbl", "v2.metadata.json", []byte(`{"orphan":true}`))
+	if err != nil {
+		t.Fatalf("first stage failed: %v", err)
+	}
+	if name != "v2.metadata.json" {
+		t.Errorf("first stage used %q, want the plain versioned name", name)
+	}
+	if location != "s3://bkt/ns/tbl/metadata/v2.metadata.json" {
+		t.Errorf("location = %q", location)
+	}
+
+	name, location, err = s.stageCommitMetadata(ctx, "bkt", "ns/tbl", "s3://bkt/ns/tbl", "v2.metadata.json", []byte(`{"second":true}`))
+	if err != nil {
+		t.Fatalf("second stage failed: %v", err)
+	}
+	if !strings.HasPrefix(name, "v2-") || !strings.HasSuffix(name, ".metadata.json") {
+		t.Errorf("second stage used %q, want a unique v2-* name", name)
+	}
+	if location != "s3://bkt/ns/tbl/metadata/"+name {
+		t.Errorf("location = %q, want it to match the staged name", location)
+	}
+
+	stored := string(client.entries[path.Join(metadataDirPath("bkt", "ns/tbl"), "v2.metadata.json")])
+	if stored != `{"orphan":true}` {
+		t.Errorf("the first file was overwritten: %s", stored)
+	}
+	if fallback := string(client.entries[path.Join(metadataDirPath("bkt", "ns/tbl"), name)]); fallback != `{"second":true}` {
+		t.Errorf("fallback file holds %s, want the second writer's metadata", fallback)
+	}
+}
+
+// Paths that legitimately rewrite a file, such as manifest repair, keep the
+// overwriting behaviour.
+func TestSaveMetadataFileOverwrites(t *testing.T) {
+	client := newStubFilerClient()
+	s := &Server{filerClient: client}
+	ctx := context.Background()
+
+	if err := s.saveMetadataFile(ctx, "bkt", "ns/tbl", "v2.metadata.json", []byte(`{"first":true}`), false); err != nil {
+		t.Fatalf("first write failed: %v", err)
+	}
+	if err := s.saveMetadataFile(ctx, "bkt", "ns/tbl", "v2.metadata.json", []byte(`{"second":true}`), false); err != nil {
+		t.Fatalf("second write failed: %v", err)
+	}
+
+	stored := string(client.entries[path.Join(metadataDirPath("bkt", "ns/tbl"), "v2.metadata.json")])
+	if stored != `{"second":true}` {
+		t.Errorf("stored metadata = %s, want the second write", stored)
+	}
+}

weed/s3api/s3tables/handler_rename_test.go

@@ -1,6 +1,7 @@
 package s3tables
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"net"
@@ -26,6 +27,9 @@ type memFilerServer struct {
 	filer_pb.UnimplementedSeaweedFilerServer
 	entries map[string]map[string]*filer_pb.Entry // dir -> name -> entry
 	client  filer_pb.SeaweedFilerClient
+	// beforeUpdate runs once, at the start of the next UpdateEntry, so a test
+	// can land a competing write in a handler's read-to-write window.
+	beforeUpdate func()
 }
 
 func newMemFilerServer() *memFilerServer {
@@ -81,6 +85,21 @@ func (f *memFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntr
 }
 
 func (f *memFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
+	if hook := f.beforeUpdate; hook != nil {
+		f.beforeUpdate = nil
+		hook()
+	}
+	// The real filer validates ExpectedExtended under the per-path lock; without
+	// it here a lost update would look like a success.
+	for key, expected := range req.ExpectedExtended {
+		var actual []byte
+		if existing := f.getEntry(req.Directory, req.Entry.Name); existing != nil {
+			actual = existing.Extended[key]
+		}
+		if !bytes.Equal(actual, expected) {
+			return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key)
+		}
+	}
 	if _, ok := f.entries[req.Directory]; !ok {
 		f.entries[req.Directory] = make(map[string]*filer_pb.Entry)
 	}

weed/s3api/s3tables/handler_table.go

@@ -1,6 +1,7 @@
 package s3tables
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -293,6 +294,14 @@ func metadataVersionFromLocation(metadataLocation string) int {
 	if v, err := strconv.Atoi(strings.TrimPrefix(name, "v")); err == nil && v > 0 {
 		return v
 	}
+	// v{N}-{unique} form, written when a commit finds v{N} already staged
+	if trimmed := strings.TrimPrefix(name, "v"); trimmed != name {
+		if idx := strings.IndexByte(trimmed, '-'); idx != -1 {
+			if v, err := strconv.Atoi(trimmed[:idx]); err == nil && v > 0 {
+				return v
+			}
+		}
+	}
 	// {NNNNN}-{uuid} form: the leading integer before the first '-'
 	if idx := strings.IndexByte(name, '-'); idx != -1 {
 		if v, err := strconv.Atoi(name[:idx]); err == nil && v > 0 {
@@ -1566,6 +1575,8 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 
 	// Load existing metadata and policies for authorization
 	var metadata tableMetadataInternal
+	var storedMetadata []byte
+	var storedPolicy []byte
 	var tablePolicy string
 	var bucketPolicy string
 	var bucketTags map[string]string
@@ -1581,11 +1592,13 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		if err := json.Unmarshal(data, &metadata); err != nil {
 			return fmt.Errorf("failed to unmarshal table metadata: %w", err)
 		}
+		storedMetadata = data
 
 		// 2. Get Table Policy & Tags
 		policyData, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy)
 		if err == nil {
 			tablePolicy = string(policyData)
+			storedPolicy = policyData
 		} else if !errors.Is(err, ErrAttributeNotFound) {
 			return fmt.Errorf("failed to fetch table policy: %w", err)
 		}
@@ -1697,14 +1710,33 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		return err
 	}
 
+	// Conditional on the metadata this request read and authorized against:
+	// two commits that both passed the version-token check would otherwise each
+	// write their own metadata and the later one would drop the earlier
+	// snapshot. mutateEntryExtended retries on a changed entry, so the check
+	// lives in the mutation, where it sees the value that is current now.
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
-			return err
-		}
-		return nil
+		return h.mutateEntryExtended(r.Context(), client, tablePath, func(extended map[string][]byte) error {
+			if !bytes.Equal(extended[ExtendedKeyMetadata], storedMetadata) {
+				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata)
+			}
+			// The policy this request was authorized against must still be the
+			// one in force: an administrator restricting it mid-commit should
+			// send the caller back through authorization, not have its decision
+			// applied afterwards.
+			if !bytes.Equal(extended[ExtendedKeyPolicy], storedPolicy) {
+				return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy)
+			}
+			extended[ExtendedKeyMetadata] = metadataBytes
+			return nil
+		})
 	})
 
 	if err != nil {
+		if errors.Is(err, ErrConcurrentUpdate) {
+			h.writeError(w, http.StatusConflict, ErrCodeConflict, "table was updated concurrently")
+			return ErrVersionTokenMismatch
+		}
 		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update metadata")
 		return err
 	}

weed/s3api/s3tables/handler_update_cas_test.go

@@ -0,0 +1,98 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func updateTableRequest(t *testing.T, metadataLocation string, version int) *UpdateTableRequest {
+	t.Helper()
+	return &UpdateTableRequest{
+		TableBucketARN:   mustBucketARN(t),
+		Namespace:        []string{"ns"},
+		Name:             "t",
+		MetadataVersion:  version,
+		MetadataLocation: metadataLocation,
+	}
+}
+
+// A second commit that reads the same base must not overwrite the first one's
+// metadata pointer: the snapshot the first commit published would be lost.
+func TestUpdateTableRejectsLostUpdate(t *testing.T) {
+	fs, m := startRenameManager(t)
+
+	winnerLocation := "s3://" + renameTestBucket + "/ns/t/metadata/v4-winner.metadata.json"
+	fs.beforeUpdate = func() {
+		winner, err := json.Marshal(tableMetadataInternal{
+			Name:             "t",
+			Namespace:        "ns",
+			Format:           "ICEBERG",
+			OwnerAccountID:   DefaultAccountID,
+			MetadataVersion:  4,
+			MetadataLocation: winnerLocation,
+			VersionToken:     generateVersionToken(),
+		})
+		require.NoError(t, err)
+		entry := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
+		require.NotNil(t, entry)
+		entry.Extended[ExtendedKeyMetadata] = winner
+	}
+
+	loserLocation := "s3://" + renameTestBucket + "/ns/t/metadata/v4-loser.metadata.json"
+	err := m.Execute(context.Background(), NewManagerClient(fs.client), "UpdateTable",
+		updateTableRequest(t, loserLocation, 4), nil, "")
+
+	require.Error(t, err)
+	var s3Err *S3TablesError
+	require.ErrorAs(t, err, &s3Err)
+	assert.Equal(t, ErrCodeConflict, s3Err.Type)
+
+	got, err := runGetTable(t, m, fs, "ns", "t")
+	require.NoError(t, err)
+	assert.Equal(t, winnerLocation, got.MetadataLocation, "the first commit must survive")
+}
+
+// A policy written between the authorization check and the write decides
+// whether this caller may still commit, so the commit is rejected rather than
+// applied on a decision that has been overtaken - and the policy itself must
+// survive, not be reverted by the commit's stale copy of the entry.
+func TestUpdateTableRejectsWhenTheAuthorizingPolicyChanges(t *testing.T) {
+	fs, m := startRenameManager(t)
+
+	policy := []byte(`{"Version":"2012-10-17","Statement":[]}`)
+	fs.beforeUpdate = func() {
+		entry := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
+		require.NotNil(t, entry)
+		entry.Extended[ExtendedKeyPolicy] = policy
+	}
+
+	location := "s3://" + renameTestBucket + "/ns/t/metadata/v4.metadata.json"
+	err := m.Execute(context.Background(), NewManagerClient(fs.client), "UpdateTable",
+		updateTableRequest(t, location, 4), nil, "")
+
+	require.Error(t, err)
+	var s3Err *S3TablesError
+	require.ErrorAs(t, err, &s3Err)
+	assert.Equal(t, ErrCodeConflict, s3Err.Type)
+
+	entry := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "t")
+	require.NotNil(t, entry)
+	assert.Equal(t, policy, entry.Extended[ExtendedKeyPolicy], "the concurrent policy write must survive")
+}
+
+func TestUpdateTableAppliesWithoutContention(t *testing.T) {
+	fs, m := startRenameManager(t)
+
+	location := "s3://" + renameTestBucket + "/ns/t/metadata/v4.metadata.json"
+	require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.client), "UpdateTable",
+		updateTableRequest(t, location, 4), nil, ""))
+
+	got, err := runGetTable(t, m, fs, "ns", "t")
+	require.NoError(t, err)
+	assert.Equal(t, location, got.MetadataLocation)
+	assert.Equal(t, 4, got.MetadataVersion)
+}

weed/s3api/s3tables/table_data_dir_test.go

@@ -21,3 +21,17 @@ func TestTableDataDirFromMetadataLocation(t *testing.T) {
 		}
 	}
 }
+
+func TestMetadataVersionFromLocationUniqueSuffix(t *testing.T) {
+	cases := map[string]int{
+		"s3://bkt/ns/t/metadata/v4.metadata.json":                                      4,
+		"s3://bkt/ns/t/metadata/v4-0a1b2c3d-4e5f-6789-abcd-ef0123456789.metadata.json": 4,
+		"s3://bkt/ns/t/metadata/00007-0a1b2c3d.metadata.json":                          7,
+		"s3://bkt/ns/t/metadata/whatever.metadata.json":                                1,
+	}
+	for location, want := range cases {
+		if got := metadataVersionFromLocation(location); got != want {
+			t.Errorf("metadataVersionFromLocation(%q) = %d, want %d", location, got, want)
+		}
+	}
+}