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