From d044839ab2c65825278ed7880f560d6080ab6f0d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 16 Aug 2026 12:55:42 -0700 Subject: [PATCH] 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 | 34 ++++- weed/s3api/iceberg/handlers_commit.go | 8 +- weed/s3api/iceberg/handlers_table.go | 4 +- weed/s3api/iceberg/handlers_transaction.go | 12 +- weed/s3api/iceberg/handlers_view.go | 2 +- weed/s3api/iceberg/handlers_view_update.go | 8 +- weed/s3api/iceberg/manifest_repair.go | 2 +- weed/s3api/iceberg/metadata_files.go | 13 +- weed/s3api/iceberg/metadata_files_test.go | 128 ++++++++++++++++++ weed/s3api/s3tables/handler_rename_test.go | 19 +++ weed/s3api/s3tables/handler_table.go | 40 +++++- .../s3api/s3tables/handler_update_cas_test.go | 98 ++++++++++++++ weed/s3api/s3tables/table_data_dir_test.go | 14 ++ 13 files changed, 357 insertions(+), 25 deletions(-) create mode 100644 weed/s3api/iceberg/metadata_files_test.go create mode 100644 weed/s3api/s3tables/handler_update_cas_test.go diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index 2acf34044..764385766 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/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", diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index c62c774cb..e6922465e 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/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") diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 6d9a4e7dc..4e130bfa1 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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 } diff --git a/weed/s3api/iceberg/handlers_transaction.go b/weed/s3api/iceberg/handlers_transaction.go index f46bdfa37..6b7065cca 100644 --- a/weed/s3api/iceberg/handlers_transaction.go +++ b/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, diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index ba3d29a45..2a08b21d8 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/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 { diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go index 73ebcadd2..9ee23065c 100644 --- a/weed/s3api/iceberg/handlers_view_update.go +++ b/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") diff --git a/weed/s3api/iceberg/manifest_repair.go b/weed/s3api/iceberg/manifest_repair.go index ea629ffdd..e1326d9d0 100644 --- a/weed/s3api/iceberg/manifest_repair.go +++ b/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 diff --git a/weed/s3api/iceberg/metadata_files.go b/weed/s3api/iceberg/metadata_files.go index 1e6223f0a..7d9021dc4 100644 --- a/weed/s3api/iceberg/metadata_files.go +++ b/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{ diff --git a/weed/s3api/iceberg/metadata_files_test.go b/weed/s3api/iceberg/metadata_files_test.go new file mode 100644 index 000000000..195247582 --- /dev/null +++ b/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) + } +} diff --git a/weed/s3api/s3tables/handler_rename_test.go b/weed/s3api/s3tables/handler_rename_test.go index 7f522704f..a7f265937 100644 --- a/weed/s3api/s3tables/handler_rename_test.go +++ b/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) } diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 0c0bda04f..04014ac2a 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/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 } diff --git a/weed/s3api/s3tables/handler_update_cas_test.go b/weed/s3api/s3tables/handler_update_cas_test.go new file mode 100644 index 000000000..ab5b3eff7 --- /dev/null +++ b/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) +} diff --git a/weed/s3api/s3tables/table_data_dir_test.go b/weed/s3api/s3tables/table_data_dir_test.go index ffbcc3c79..c702f14e3 100644 --- a/weed/s3api/s3tables/table_data_dir_test.go +++ b/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) + } + } +}