diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 273c7989d4..ecc4e1410d 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -9,6 +9,7 @@ import ( "os" "time" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -118,6 +119,11 @@ func (h *S3TablesHandler) mutateEntryExtended( if err == nil { return nil } + // The filer answers an entry removed since the lookup with + // codes.NotFound; report it as the lookup's own miss. + if status.Code(err) == codes.NotFound { + return fmt.Errorf("%w: %s", filer_pb.ErrNotFound, path) + } if status.Code(err) != codes.FailedPrecondition { return err } @@ -185,12 +191,25 @@ func (h *S3TablesHandler) removeExtendedAttributesIf( // delete's restore or a competing create that landed since the caller's check // is a name conflict, not a record this write may overwrite — an overwrite // would leave the loser's tags, policy, and maintenance settings on the new -// record. +// record. A record still marked by an unfinished create is the exception: +// invisible to every reader, it is adopted rather than refused. A rename +// destination is not adoptable — its writer still owes the record a publish +// or a keep, and adopting it strands the renamed record with neither name. func (h *S3TablesHandler) claimCatalogEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error { return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { - for _, key := range s3tablesExtendedKeys { - if _, ok := extended[key]; ok { - return fmt.Errorf("%w: %s set on %s", ErrNameTaken, key, path) + if string(extended[ExtendedKeyDeleting]) == createClaim { + // A marked record is absent to every reader, so it holds nothing a + // caller could have seen: the create that left it confirmed or lost + // the name already. Take the record over, dropping whatever the + // earlier write left, or an abandoned mark wedges the name forever. + for _, key := range s3tablesExtendedKeys { + delete(extended, key) + } + } else { + for _, key := range s3tablesExtendedKeys { + if _, ok := extended[key]; ok { + return fmt.Errorf("%w: %s set on %s", ErrNameTaken, key, path) + } } } for key, data := range attrs { @@ -407,6 +426,215 @@ func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client f return err } +var ( + errParentDeleting = errors.New("parent is being deleted") + errParentGone = errors.New("parent was deleted") + errNoDeleteClaim = errors.New("claim not held") + errBackOutFailed = errors.New("backing out the create failed") + errSourceGone = errors.New("rename source was deleted") +) + +// claimForDelete writes a fresh token into ExtendedKeyDeleting, replacing any +// earlier claim, and returns it with the attributes the delete must still find +// when it removes the entry. The token makes a stale delete's removal fail. +func (h *S3TablesHandler) claimForDelete(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, isParent func(*filer_pb.Entry) bool) (string, map[string][]byte, error) { + claim := generateVersionToken() + var claimed map[string][]byte + err := h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { + if !isParent(&filer_pb.Entry{Extended: extended}) { + return filer_pb.ErrNotFound + } + extended[ExtendedKeyDeleting] = []byte(claim) + claimed = snapshotExtended(extended) + return nil + }) + if err != nil { + return "", nil, err + } + return claim, claimed, nil +} + +// releaseDeleteClaim removes claim from path while path still holds it, even +// after the request was cancelled. +func (h *S3TablesHandler) releaseDeleteClaim(ctx context.Context, client filer_pb.SeaweedFilerClient, path, claim string) { + err := h.mutateEntryExtended(context.WithoutCancel(ctx), client, path, func(extended map[string][]byte) error { + if string(extended[ExtendedKeyDeleting]) != claim { + return errNoDeleteClaim + } + delete(extended, ExtendedKeyDeleting) + return nil + }) + if err != nil && !errors.Is(err, errNoDeleteClaim) && !errors.Is(err, filer_pb.ErrNotFound) { + glog.Warningf("release delete claim on %s: %v", path, err) + } +} + +// isNamespaceEntry reports whether entry carries a namespace's catalog record. +func isNamespaceEntry(entry *filer_pb.Entry) bool { + _, ok := entry.Extended[ExtendedKeyMetadata] + return ok +} + +// checkParentOpen returns errParentGone if parentPath is missing or not a +// catalog entry, and errParentDeleting if a delete has claimed it. +func (h *S3TablesHandler) checkParentOpen(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool) error { + _, err := h.openParent(ctx, client, parentPath, isParent) + return err +} + +// openParent is checkParentOpen returning the entry it accepted, so a caller +// can later ask whether the parent is still that same record. +func (h *S3TablesHandler) openParent(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool) (*filer_pb.Entry, error) { + entry, err := h.lookupEntry(ctx, client, parentPath) + if errors.Is(err, filer_pb.ErrNotFound) { + return nil, errParentGone + } + if err != nil { + return nil, err + } + if !isParent(entry) { + return nil, errParentGone + } + if _, claimed := entry.Extended[ExtendedKeyDeleting]; claimed { + return nil, errParentDeleting + } + return entry, nil +} + +// parentStillOpen reports whether parentPath still holds the record parent +// captured. A delete claims, removes, or replaces the parent between the two +// reads; createdAt identifies the record since an update preserves it while a +// delete+recreate mints a new one, so an editable property changing in +// between does not read as a removal. +func (h *S3TablesHandler) parentStillOpen(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, parent *filer_pb.Entry) error { + current, err := h.openParent(ctx, client, parentPath, isParent) + if err != nil { + return err + } + var before, now namespaceMetadata + if json.Unmarshal(parent.Extended[ExtendedKeyMetadata], &before) != nil || + json.Unmarshal(current.Extended[ExtendedKeyMetadata], &now) != nil || + !now.CreatedAt.Equal(before.CreatedAt) { + return errParentGone + } + return nil +} + +// createClaim is the ExtendedKeyDeleting value of a catalog record whose +// create has written it but not yet confirmed its parent. renameDestClaim is +// the same not-yet-published mark on a rename's destination, with a value of +// its own so a create refuses it instead of adopting the record. +const ( + createClaim = "create" + renameDestClaim = "rename-dest" +) + +// isUnfinishedCreate reports whether extended belongs to a record still +// marked by the operation that wrote it — invisible to every reader. +func isUnfinishedCreate(extended map[string][]byte) bool { + switch string(extended[ExtendedKeyDeleting]) { + case createClaim, renameDestClaim: + return true + } + return false +} + +// mutateCatalogRecord applies mutate to a catalog record, refusing a record +// an in-flight operation claimed: a create's unfinished record is absent to +// every reader, and a record claimed by a delete or rename must not take a +// write the operation's compare then trips over. +func (h *S3TablesHandler) mutateCatalogRecord( + ctx context.Context, + client filer_pb.SeaweedFilerClient, + path string, + mutate func(extended map[string][]byte) error, +) error { + return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { + if _, claimed := extended[ExtendedKeyDeleting]; !claimed { + return mutate(extended) + } + if isUnfinishedCreate(extended) { + return filer_pb.ErrNotFound + } + return fmt.Errorf("%w: %s claimed on %s", ErrConcurrentUpdate, ExtendedKeyDeleting, path) + }) +} + +// confirmParent re-checks the parent after a create wrote written onto +// childPath. If the parent is open, it clears the record's unfinished mark — +// then checks the parent once more, since a delete claiming it in between +// scans the still-marked record, skips it, and removes it under a create that +// already answered success. If the parent is gone or claimed at either check, +// it removes what the create wrote and returns why, or errBackOutFailed if +// the record stays. +func (h *S3TablesHandler) confirmParent(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, childPath string, written map[string][]byte, emptyDirs ...string) error { + parent, err := h.openParent(ctx, client, parentPath, isParent) + confirmed := false + if err == nil { + // The mark comes off only while the record still holds what this + // create wrote; a failed write backs the record out like a closed parent. + if err = h.removeExtendedAttributesIf(ctx, client, childPath, written, ExtendedKeyDeleting); err == nil { + confirmed = true + err = h.parentStillOpen(ctx, client, parentPath, isParent, parent) + } + } + if err == nil { + return nil + } + ctx = context.WithoutCancel(ctx) + expected := written + keys := make([]string, 0, len(written)) + if confirmed { + // The confirmed record no longer carries the mark written expected. + expected = make(map[string][]byte, len(written)) + for key, data := range written { + if key == ExtendedKeyDeleting { + continue + } + expected[key] = data + keys = append(keys, key) + } + } else { + for key := range written { + keys = append(keys, key) + } + } + if backOutErr := h.removeExtendedAttributesIf(ctx, client, childPath, expected, keys...); backOutErr != nil && !errors.Is(backOutErr, filer_pb.ErrNotFound) { + glog.Warningf("back out %s: %v", childPath, backOutErr) + if errors.Is(backOutErr, ErrConcurrentUpdate) { + // The record moved past this create — adopted or rewritten — so + // there is nothing of its own left to remove. + return fmt.Errorf("%w: %s changed during create", ErrConcurrentUpdate, childPath) + } + return fmt.Errorf("%w: %s: %v (%v)", errBackOutFailed, childPath, backOutErr, err) + } + for _, dir := range emptyDirs { + h.removeLeftoverDirectory(ctx, client, dir) + } + if errors.Is(err, errParentGone) { + h.removeLeftoverDirectory(ctx, client, parentPath) + } + return err +} + +// removeLeftoverDirectory deletes path if it is an empty directory without +// attributes; the filer refuses the non-recursive delete otherwise. +func (h *S3TablesHandler) removeLeftoverDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) { + entry, err := h.lookupEntry(ctx, client, path) + if err != nil || !entry.IsDirectory || len(entry.Extended) > 0 { + return + } + dir, name := splitPath(path) + if _, err := filer_pb.DoRemoveRequest(ctx, client, &filer_pb.DeleteEntryRequest{ + Directory: dir, + Name: name, + IsDeleteData: true, + ExpectedExtended: snapshotExtended(entry.Extended), + }); err != nil { + glog.V(1).Infof("keep %s: %v", path, err) + } +} + // entryExists checks if an entry exists at the given path func (h *S3TablesHandler) entryExists(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) bool { dir, name := splitPath(path) diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 4e33146e32..3350ef1f43 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -33,8 +33,10 @@ const ( ExtendedKeyMaintenanceStatus = s3_constants.ExtS3TablesPrefix + "maintenanceStatus" ExtendedKeyEntryType = s3_constants.ExtS3TablesPrefix + "entryType" // ExtendedKeyDeleting marks a catalog record claimed by an in-flight - // delete or rename, so the claim survives until the operation finishes - // or releases it. + // create, delete or rename, so the claim survives until the operation + // finishes or releases it. A table carries "create", "delete" or "rename"; + // a namespace being deleted carries the random claim token written by + // claimForDelete. ExtendedKeyDeleting = s3_constants.ExtS3TablesPrefix + "deleting" // ExtendedKeyVectorPolicy is written by the S3 Vectors service on table diff --git a/weed/s3api/s3tables/handler_delete_create_race_test.go b/weed/s3api/s3tables/handler_delete_create_race_test.go new file mode 100644 index 0000000000..38f075d3a9 --- /dev/null +++ b/weed/s3api/s3tables/handler_delete_create_race_test.go @@ -0,0 +1,767 @@ +package s3tables + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "path" + "sync" + "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" +) + +// lateEntry is the name every create in these tests tries to add. +const lateEntry = "late" + +// pausedCreateClient holds the first CreateEntry for target until release is +// closed, after closing reached, so a test can land another operation after a +// create has passed its checks and before it writes anything. +type pausedCreateClient struct { + filer_pb.SeaweedFilerClient + target string + reached chan struct{} + release chan struct{} + once sync.Once +} + +func newPausedCreateClient(client filer_pb.SeaweedFilerClient, target string) *pausedCreateClient { + return &pausedCreateClient{ + SeaweedFilerClient: client, + target: target, + reached: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *pausedCreateClient) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest, opts ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + if req.Entry != nil && path.Join(req.Directory, req.Entry.Name) == c.target { + c.once.Do(func() { + close(c.reached) + <-c.release + }) + } + return c.SeaweedFilerClient.CreateEntry(ctx, req, opts...) +} + +// pausedAfterWriteClient holds the create that has just written the first +// update to target until release is closed, after closing reached, so a test +// can land other requests after a create wrote its record and before it checks +// its parent. +type pausedAfterWriteClient struct { + filer_pb.SeaweedFilerClient + target string + reached chan struct{} + release chan struct{} + once sync.Once +} + +func newPausedAfterWriteClient(client filer_pb.SeaweedFilerClient, target string) *pausedAfterWriteClient { + return &pausedAfterWriteClient{ + SeaweedFilerClient: client, + target: target, + reached: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *pausedAfterWriteClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + resp, err := c.SeaweedFilerClient.UpdateEntry(ctx, req, opts...) + if err == nil && req.Entry != nil && path.Join(req.Directory, req.Entry.Name) == c.target { + c.once.Do(func() { + close(c.reached) + <-c.release + }) + } + return resp, err +} + +// pausedConfirmClient holds the update that lifts a create's mark on target — +// the write that publishes the record — until release is closed, so a test +// can land a namespace claim after the create's parent check and before the +// record goes live. +type pausedConfirmClient struct { + filer_pb.SeaweedFilerClient + target string + reached chan struct{} + release chan struct{} + once sync.Once +} + +func newPausedConfirmClient(client filer_pb.SeaweedFilerClient, target string) *pausedConfirmClient { + return &pausedConfirmClient{ + SeaweedFilerClient: client, + target: target, + reached: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *pausedConfirmClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if req.Entry != nil && path.Join(req.Directory, req.Entry.Name) == c.target && req.Entry.Extended[ExtendedKeyDeleting] == nil { + c.once.Do(func() { + close(c.reached) + <-c.release + }) + } + return c.SeaweedFilerClient.UpdateEntry(ctx, req, opts...) +} + +type handlerResult struct { + status int + resp S3TablesError +} + +func startHandler(t *testing.T, m *Manager, client filer_pb.SeaweedFilerClient, operation string, input interface{}) <-chan handlerResult { + t.Helper() + done := make(chan handlerResult, 1) + go func() { + status, resp := runHandler(t, m, client, operation, input) + done <- handlerResult{status, resp} + }() + return done +} + +// waitForPause returns once the create is held. A create that ends before that +// fails the test instead of hanging it. +func waitForPause(t *testing.T, reached <-chan struct{}, created <-chan handlerResult) { + t.Helper() + select { + case <-reached: + case early := <-created: + t.Fatalf("the create ended before it was held: %+v", early) + } +} + +func dropSeedTable(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) { + t.Helper() + require.NoError(t, runDeleteTable(t, m, fs, "ns", "t")) +} + +// plantDeleteClaim puts a delete's claim on an entry, the way a delete that is +// still scanning leaves it. +func plantDeleteClaim(fs *s3tablestest.MemFiler, dir, name string) { + extended := map[string][]byte{} + for key, value := range fs.Get(dir, name).Extended { + extended[key] = value + } + extended[ExtendedKeyDeleting] = []byte("planted-token") + fs.Put(dir, name, extended) +} + +// plantCreateMark puts an unfinished create's mark on an entry, the way a +// create still confirming its parent leaves it. +func plantCreateMark(fs *s3tablestest.MemFiler, dir, name string) { + extended := map[string][]byte{} + for key, value := range fs.Get(dir, name).Extended { + extended[key] = value + } + extended[ExtendedKeyDeleting] = []byte(createClaim) + fs.Put(dir, name, extended) +} + +func createTableRequest(t *testing.T, namespace, name string) *CreateTableRequest { + return &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{namespace}, + Name: name, + Format: FormatIceberg, + } +} + +// createOp is one of the operations that add an entry to a namespace, with the +// read that loads the entry back and the error code that read gives for an +// entry that is not there. +type createOp struct { + operation string + input interface{} + read string + readInput interface{} + readMissing string +} + +func getLateTableRequest(t *testing.T) *GetTableRequest { + return &GetTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: lateEntry, + } +} + +func createTableOp(t *testing.T) createOp { + t.Helper() + return createOp{ + operation: "CreateTable", + input: createTableRequest(t, "ns", lateEntry), + read: "GetTable", + readInput: getLateTableRequest(t), + readMissing: ErrCodeNoSuchTable, + } +} + +// createOps lists every operation that creates a table or view under "ns", +// each adding lateEntry. +func createOps(t *testing.T) []createOp { + t.Helper() + location := "s3://" + renameTestBucket + "/ns/" + lateEntry + "/metadata/v1.metadata.json" + return []createOp{ + createTableOp(t), + { + operation: "RegisterTable", + input: &RegisterTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: lateEntry, + MetadataLocation: location, + }, + read: "GetTable", + readInput: getLateTableRequest(t), + readMissing: ErrCodeNoSuchTable, + }, + { + operation: "CreateView", + input: &CreateViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: lateEntry, + MetadataLocation: location, + }, + read: "GetView", + readInput: &GetViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: lateEntry, + }, + readMissing: ErrCodeNoSuchView, + }, + } +} + +func deleteNamespaceRequest(t *testing.T) *DeleteNamespaceRequest { + return &DeleteNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + } +} + +// A create that finds the namespace already claimed by a delete is refused and +// writes nothing. +func TestCreatesRefusedInClaimedNamespace(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + plantDeleteClaim(fs, GetTableBucketPath(renameTestBucket), "ns") + + status, resp := runHandler(t, m, fs.Client, op.operation, op.input) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), lateEntry), "nothing may be written under a namespace being deleted") + }) + } +} + +// A create that passed its checks writes after the delete scanned the namespace +// empty and before it removed the directory. The create sees the claim, backs +// out, and answers 409 instead of reporting an entry the delete then removes. +func TestCreateBetweenNamespaceScanAndDeleteBacksOut(t *testing.T) { + type gap struct { + create handlerResult + leftover *filer_pb.Entry + } + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + nsPath := GetNamespacePath(renameTestBucket, "ns") + client := newPausedCreateClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + // The create finishes inside the delete's check-to-delete window. + inGap := make(chan gap, 1) + fs.BeforeDelete = func() { + close(client.release) + create := <-created + inGap <- gap{create, fs.Get(nsPath, lateEntry)} + } + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + got := <-inGap + assert.Equal(t, http.StatusConflict, got.create.status, "an entry the delete removes must not be reported created: %+v", got.create.resp) + assert.Equal(t, ErrCodeConflict, got.create.resp.Type) + assert.Nil(t, got.leftover, "the create must remove what it wrote before the delete runs") + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns")) + }) + } +} + +// A create that passed its parent check publishes the record while the delete +// holds the namespace claim: the delete already scanned the marked record and +// skipped it, so the create must re-check the parent after lifting the mark — +// reporting success here leaves the delete to remove a record its caller was +// told exists. +func TestCreateConfirmingUnderANamespaceClaimBacksOut(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + nsPath := GetNamespacePath(renameTestBucket, "ns") + client := newPausedConfirmClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + // The delete claims the namespace and scans the still-marked + // record before the create's publishing write lands. + inGap := make(chan handlerResult, 1) + fs.BeforeDelete = func() { + close(client.release) + inGap <- <-created + } + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + create := <-inGap + assert.Equal(t, http.StatusConflict, create.status, "a create confirmed under the claim must not report success: %+v", create.resp) + assert.Equal(t, ErrCodeConflict, create.resp.Type) + assert.Nil(t, fs.Get(nsPath, lateEntry)) + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns")) + }) + } +} + +// A namespace update inside the confirm window edits the same record — its +// createdAt survives — so the create still confirms: reading the edit as a +// delete+recreate would back a live entry out and report 404 for a namespace +// that never went away. +func TestCreateConfirmingAcrossANamespaceUpdateSucceeds(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + bucketPath := GetTableBucketPath(renameTestBucket) + nsPath := GetNamespacePath(renameTestBucket, "ns") + client := newPausedConfirmClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + // Rewrite the namespace metadata the way UpdateNamespace does — + // same record, new properties — while the create's record waits. + extended := map[string][]byte{} + for key, value := range fs.Get(bucketPath, "ns").Extended { + extended[key] = value + } + var nsMeta namespaceMetadata + require.NoError(t, json.Unmarshal(extended[ExtendedKeyMetadata], &nsMeta)) + nsMeta.Properties = map[string]string{"edited": "true"} + data, err := json.Marshal(&nsMeta) + require.NoError(t, err) + extended[ExtendedKeyMetadata] = data + fs.Put(bucketPath, "ns", extended) + + close(client.release) + create := <-created + assert.Equal(t, http.StatusOK, create.status, "a namespace update must not read as a delete: %+v", create.resp) + assert.NotNil(t, fs.Get(nsPath, lateEntry), "the create's record must survive an unchanged namespace") + }) + } +} + +// A create that passed its checks writes after the delete finished. The filer +// makes the missing namespace directory for it, without the namespace record, +// so the create answers 404 and removes that directory too. +func TestCreateAfterNamespaceDeleteBacksOut(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + bucketPath := GetTableBucketPath(renameTestBucket) + client := newPausedCreateClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + require.Equal(t, http.StatusOK, status, "%+v", resp) + // The filer creates missing parents for the entry directory, so the + // namespace directory comes back, without its record. + fs.Put(bucketPath, "ns", nil) + close(client.release) + + create := <-created + assert.Equal(t, http.StatusNotFound, create.status, "an entry under a deleted namespace must not be reported created: %+v", create.resp) + assert.Equal(t, ErrCodeNoSuchNamespace, create.resp.Type) + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), lateEntry)) + assert.Nil(t, fs.Get(bucketPath, "ns"), "the re-created namespace directory must go with the entry") + }) + } +} + +// A create that has written its record but not yet checked the namespace may +// still back the record out, so nothing may report it as existing. Otherwise a +// caller is told the entry exists, a delete then claims the namespace, and the +// create removes the entry that caller was told about. +func TestUnfinishedCreateIsNotReportedAsExisting(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + bucketPath := GetTableBucketPath(renameTestBucket) + client := newPausedAfterWriteClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + status, resp := runHandler(t, m, fs.Client, op.read, op.readInput) + assert.Equal(t, http.StatusNotFound, status, "an unfinished entry must not load: %+v", resp) + assert.Equal(t, op.readMissing, resp.Type) + // A second create is not told the entry exists; it takes the marked + // record over, since nothing a reader can see lives there. + status, resp = runHandler(t, m, fs.Client, op.operation, op.input) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + + plantDeleteClaim(fs, bucketPath, "ns") + close(client.release) + + create := <-created + assert.Equal(t, http.StatusConflict, create.status, "%+v", create.resp) + assert.Equal(t, ErrCodeConflict, create.resp.Type) + status, resp = runHandler(t, m, fs.Client, op.read, op.readInput) + assert.Equal(t, http.StatusOK, status, "the adopted record survives the loser's back-out: %+v", resp) + }) + } +} + +// A caller who may not create is told so whether or not a delete holds the +// namespace; the claim must not show through a refusal. +func TestDeniedCreatesDoNotSeeADeleteClaim(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + m.SetTrusted(false) + m.SetDefaultAllow(false) + renameOnly := []string{"s3tables:RenameTable"} + + unclaimed := runRequestAs(t, m, fs, op.operation, op.input, renameOnly) + require.Equal(t, http.StatusForbidden, unclaimed.status, "%+v", unclaimed.body) + + plantDeleteClaim(fs, GetTableBucketPath(renameTestBucket), "ns") + claimed := runRequestAs(t, m, fs, op.operation, op.input, renameOnly) + assert.Equal(t, unclaimed, claimed) + assert.NotEqual(t, http.StatusConflict, claimed.status) + }) + } +} + +// A record still marked by its create is absent to readers, so listings skip +// it the way the point reads do. +func TestListingsSkipAnUnfinishedCreate(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + client := newPausedAfterWriteClient(fs.Client, GetTablePath(renameTestBucket, "ns", lateEntry)) + created := startHandler(t, m, client, op.operation, op.input) + waitForPause(t, client.reached, created) + + var names []string + if op.operation == "CreateView" { + list := &ListViewsResponse{} + require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.Client), "ListViews", &ListViewsRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + }, list, "")) + for _, v := range list.Views { + names = append(names, v.Name) + } + } else { + list := &ListTablesResponse{} + require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.Client), "ListTables", &ListTablesRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + }, list, "")) + for _, tb := range list.Tables { + names = append(names, tb.Name) + } + } + assert.NotContains(t, names, lateEntry, "an unfinished create must not be listed") + + close(client.release) + require.Equal(t, http.StatusOK, (<-created).status) + }) + } +} + +// A write that lands on a record a create marked refuses instead of updating +// an entry every reader reports absent and breaking the create's back-out. +func TestUpdateRefusedOnAMarkedRecord(t *testing.T) { + fs, m := startRenameManager(t) + fs.BeforeUpdate = func() { + plantCreateMark(fs, GetNamespacePath(renameTestBucket, "ns"), "t") + } + status, resp := runHandler(t, m, fs.Client, "UpdateTable", &UpdateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + MetadataLocation: "s3://" + renameTestBucket + "/ns/t/metadata/v4.metadata.json", + }) + assert.Equal(t, http.StatusNotFound, status, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchTable, resp.Type) +} + +// A rename into a namespace a delete claimed is refused before it touches the +// source: the destination cannot be confirmed, so the name must stay where it +// was. +func TestRenameIntoClaimedNamespaceIsRefused(t *testing.T) { + fs, m := startRenameManager(t) + ns2Meta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"ns2"}, OwnerAccountID: DefaultAccountID}) + fs.Put(GetTableBucketPath(renameTestBucket), "ns2", map[string][]byte{ExtendedKeyMetadata: ns2Meta}) + plantDeleteClaim(fs, GetTableBucketPath(renameTestBucket), "ns2") + + status, resp := runHandler(t, m, fs.Client, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns2"}, + DestName: "t2", + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + + src := fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, src) + assert.Contains(t, src.Extended, ExtendedKeyMetadata, "the refused rename must leave the source record") + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns2"), "t2")) +} + +// The rename's destination goes up with the same unfinished mark a create +// leaves and ends without it once the namespace is confirmed. +func TestRenameClearsTheDestinationMark(t *testing.T) { + fs, m := startRenameManager(t) + require.NoError(t, runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + })) + dest := fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t2") + require.NotNil(t, dest) + assert.NotContains(t, dest.Extended, ExtendedKeyDeleting, "a finished rename must lift its unfinished mark") +} + +// A rename destination is marked like a create but still owed a publish by the +// rename that wrote it: a create that adopted it the way it adopts an orphaned +// mark would strand the renamed record once its source is cleared. +func TestCreateCannotAdoptARenameDestination(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + nsPath := GetNamespacePath(renameTestBucket, "ns") + marked, _ := json.Marshal(tableMetadataInternal{ + Name: lateEntry, + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataLocation: "s3://" + renameTestBucket + "/ns/" + lateEntry + "/metadata/v1.metadata.json", + }) + fs.Put(nsPath, lateEntry, map[string][]byte{ + ExtendedKeyMetadata: marked, + ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyDeleting: []byte(renameDestClaim), + }) + + status, resp := runHandler(t, m, fs.Client, op.operation, op.input) + assert.Equal(t, http.StatusConflict, status, "a rename destination must refuse a create: %+v", resp) + entry := fs.Get(nsPath, lateEntry) + require.NotNil(t, entry) + assert.Equal(t, []byte(renameDestClaim), entry.Extended[ExtendedKeyDeleting], "the rename's record must be untouched") + }) + } +} + +// A rename that commits while a delete claims the destination namespace has +// the same shape as a confirming create: the delete scanned the still-marked +// destination and skipped it, so the commit must re-check the namespace after +// the record goes live and restore the source instead of losing it. +func TestRenameCommittingUnderAClaimedNamespaceRestoresSource(t *testing.T) { + fs, m := startRenameManager(t) + ns2Meta, _ := json.Marshal(namespaceMetadata{Namespace: []string{"ns2"}, OwnerAccountID: DefaultAccountID}) + fs.Put(GetTableBucketPath(renameTestBucket), "ns2", map[string][]byte{ExtendedKeyMetadata: ns2Meta}) + + client := newPausedConfirmClient(fs.Client, GetTablePath(renameTestBucket, "ns2", "t2")) + renamed := startHandler(t, m, client, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns2"}, + DestName: "t2", + }) + waitForPause(t, client.reached, renamed) + + // The delete claims ns2 and scans the still-marked destination before the + // rename's publishing write lands. + inGap := make(chan handlerResult, 1) + fs.BeforeDelete = func() { + close(client.release) + inGap <- <-renamed + } + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", &DeleteNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + }) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + rename := <-inGap + assert.Equal(t, http.StatusConflict, rename.status, "a rename committed under the claim must not report success: %+v", rename.resp) + src := fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t") + require.NotNil(t, src) + assert.Contains(t, src.Extended, ExtendedKeyMetadata, "the failed rename must restore its source record") + assert.NotContains(t, src.Extended, ExtendedKeyDeleting) + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns2")) +} + +// A delete refused because the namespace still holds a table lifts its claim +// again, so the namespace takes creates as before. +func TestDeleteNamespaceWithTableLiftsItsClaim(t *testing.T) { + for _, op := range []createOp{createTableOp(t)} { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + bucketPath := GetTableBucketPath(renameTestBucket) + before := fs.Get(bucketPath, "ns").Extended + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeNamespaceNotEmpty, resp.Type) + assert.Equal(t, before, fs.Get(bucketPath, "ns").Extended, "the refused delete must not leave its claim behind") + + status, resp = runHandler(t, m, fs.Client, op.operation, op.input) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + created := fs.Get(GetNamespacePath(renameTestBucket, "ns"), lateEntry) + require.NotNil(t, created) + assert.NotContains(t, created.Extended, ExtendedKeyDeleting, "a finished create must lift its unfinished mark") + }) + } +} + +// orphanMarkedRecord plants what a create that crashed after writing its +// record leaves behind: catalog attributes under the unfinished mark. +func orphanMarkedRecord(fs *s3tablestest.MemFiler, dir, name string) { + meta, _ := json.Marshal(tableMetadataInternal{ + Name: name, + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataVersion: 1, + MetadataLocation: "s3://" + renameTestBucket + "/ns/" + name + "/metadata/v1.metadata.json", + }) + fs.Put(dir, name, map[string][]byte{ + ExtendedKeyMetadata: meta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyPolicy: []byte("stale"), + ExtendedKeyDeleting: []byte(createClaim), + }) +} + +// A record whose create never lifted its mark — the request died after the +// write — must not wedge the name: the marked record is absent to every +// reader, so a create takes it over instead of refusing the name forever. +func TestCreateAdoptsAnOrphanedMark(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + orphanMarkedRecord(fs, nsPath, lateEntry) + + status, resp := runHandler(t, m, fs.Client, op.operation, op.input) + require.Equal(t, http.StatusOK, status, "%+v", resp) + + entry := fs.Get(nsPath, lateEntry) + require.NotNil(t, entry) + assert.NotContains(t, entry.Extended, ExtendedKeyDeleting, "the adopted record must end confirmed") + assert.NotContains(t, entry.Extended, ExtendedKeyPolicy, "the stale record's keys must not survive") + + status, resp = runHandler(t, m, fs.Client, op.read, op.readInput) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + }) + } +} + +// A marked record is no destination a rename must refuse: the rename takes +// the record over like a create does. +func TestRenameAdoptsAMarkedDestination(t *testing.T) { + fs, m := startRenameManager(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + orphanMarkedRecord(fs, nsPath, "t2") + + require.NoError(t, runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + })) + + dest := fs.Get(nsPath, "t2") + require.NotNil(t, dest) + assert.NotContains(t, dest.Extended, ExtendedKeyDeleting) + assert.NotContains(t, dest.Extended, ExtendedKeyPolicy, "the stale record's keys must not survive") + var metadata tableMetadataInternal + require.NoError(t, json.Unmarshal(dest.Extended[ExtendedKeyMetadata], &metadata)) + assert.Equal(t, "t2", metadata.Name) +} + +// A marked record is no child either: its create cannot confirm under the +// claim the delete holds, and an abandoned one would refuse the delete +// forever. The delete removes it with the namespace. +func TestDeleteNamespaceSkipsAnUnfinishedCreate(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + orphanMarkedRecord(fs, GetNamespacePath(renameTestBucket, "ns"), "late") + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + require.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns")) +} + +// A namespace whose first page of entries is all skippable — bare directories +// a dropped table left behind — can still hold a table on a later page: the +// scan must keep listing instead of deleting the namespace out from under it. +func TestDeleteNamespaceScansPastASkippedPage(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + nsPath := GetNamespacePath(renameTestBucket, "ns") + for i := 0; i < 12; i++ { + fs.Put(nsPath, fmt.Sprintf("a%02d", i), nil) + } + tableMeta, _ := json.Marshal(tableMetadataInternal{ + Name: "zz", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + }) + fs.Put(nsPath, "zz", map[string][]byte{ExtendedKeyMetadata: tableMeta}) + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeNamespaceNotEmpty, resp.Type) + assert.NotNil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns")) +} + +// A first page of skippables and nothing after it is still an empty +// namespace: the scan must page to the end of the listing, not stop early. +func TestDeleteNamespacePagesToTheEnd(t *testing.T) { + fs, m := startRenameManager(t) + dropSeedTable(t, m, fs) + nsPath := GetNamespacePath(renameTestBucket, "ns") + for i := 0; i < 12; i++ { + fs.Put(nsPath, fmt.Sprintf("a%02d", i), nil) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t)) + require.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns")) +} diff --git a/weed/s3api/s3tables/handler_delete_decouple_test.go b/weed/s3api/s3tables/handler_delete_decouple_test.go index c22c433325..896ff730a3 100644 --- a/weed/s3api/s3tables/handler_delete_decouple_test.go +++ b/weed/s3api/s3tables/handler_delete_decouple_test.go @@ -142,6 +142,108 @@ func TestDeleteTableDecoupledKeepsReusedNamePath(t *testing.T) { } } +// A decoupled record still carrying a claim its operation never lifted — a +// delete or rename that ended between the claim and the marker clear — must go +// with the rest of the marker, or the leftover claim blocks every create of +// the name from then on. +func TestDeleteTableDecoupledClearsLeftoverClaim(t *testing.T) { + fs, m := startRenameManager(t) + + newtMeta, _ := json.Marshal(tableMetadataInternal{ + Name: "newt", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataLocation: "s3://" + renameTestBucket + "/ns/newt-x/metadata/v1.metadata.json", + }) + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.Put(nsPath, "newt", map[string][]byte{ + ExtendedKeyMetadata: newtMeta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyDeleting: []byte("delete"), + }) + fs.Put(nsPath, "newt-x", nil) + fs.Put(GetTablePath(renameTestBucket, "ns", "newt-x"), "metadata", nil) + + require.NoError(t, runDeleteTable(t, m, fs, "ns", "newt")) + + marker := fs.Get(nsPath, "newt") + require.NotNil(t, marker) + assert.NotContains(t, marker.Extended, ExtendedKeyDeleting, "a leftover claim must not survive the delete") + + status, resp := runHandler(t, m, fs.Client, "CreateTable", &CreateTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "newt", + Format: FormatIceberg, + }) + assert.Equal(t, http.StatusOK, status, "the name must take a create once the claim is gone: %+v", resp) +} + +// An update that moves a decoupled table's MetadataLocation between the +// delete's authorized read and the read its claim and purge run on must turn +// the delete into a conflict: the purge otherwise empties the stale directory +// and unmarks a record whose data lives somewhere else entirely. +func TestDeleteTableDecoupledLosesToMovedLocation(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + metaA, _ := json.Marshal(tableMetadataInternal{ + Name: "mv", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataLocation: "s3://" + renameTestBucket + "/ns/mv-a/metadata/v1.metadata.json", + }) + metaB, _ := json.Marshal(tableMetadataInternal{ + Name: "mv", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataLocation: "s3://" + renameTestBucket + "/ns/mv-b/metadata/v1.metadata.json", + }) + fs.Put(nsPath, "mv", map[string][]byte{ + ExtendedKeyMetadata: metaA, + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + fs.Put(GetTablePath(renameTestBucket, "ns", "mv-a"), "data", nil) + fs.Put(GetTablePath(renameTestBucket, "ns", "mv-b"), "data", nil) + + // The first read of the marker is the authorized one; land the update's + // write in front of the read the claim and purge run on. + lookups := 0 + fs.OnLookup = func(dir, name string) { + if dir != nsPath || name != "mv" { + return + } + lookups++ + if lookups > 1 { + fs.OnLookup = nil + fs.Put(nsPath, "mv", map[string][]byte{ + ExtendedKeyMetadata: metaB, + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + } + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "mv", + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "mv-a"), "data"), + "the stale location's data must survive") + assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "mv-b"), "data"), + "the live location's data must survive") + rec := fs.Get(nsPath, "mv") + require.NotNil(t, rec) + assert.Equal(t, metaB, rec.Extended[ExtendedKeyMetadata]) + assert.NotContains(t, rec.Extended, ExtendedKeyDeleting, "a refused delete must not leave a claim") +} + // A table whose MetadataLocation resolves to an ancestor of its own name path // (here the namespace root, e.g. from corrupt metadata) must not be deleted: a // recursive purge of that data path would wipe sibling tables. The delete is @@ -323,3 +425,66 @@ func TestRenameTableLosesToDeleteClaim(t *testing.T) { assert.NotNil(t, src.Extended[ExtendedKeyMetadata], "the claimed source record must survive") assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "t"), "data")) } + +// A no-token delete that meets a record claimed by a rename loses to it: the +// claim is what keeps the rename's destination from pointing at purged data. +func TestDeleteTableLosesToRenameClaim(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + e := fs.Get(nsPath, "t") + claimed := map[string][]byte{} + for k, v := range e.Extended { + claimed[k] = v + } + claimed[ExtendedKeyDeleting] = []byte("rename") + fs.Put(nsPath, "t", claimed) + + status, resp := runHandler(t, m, fs.Client, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + src := fs.Get(nsPath, "t") + require.NotNil(t, src) + assert.NotNil(t, src.Extended[ExtendedKeyMetadata], "the claimed record must survive") + assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "t"), "data")) +} + +// The token delete takes its snapshot after the rename's claim landed; it must +// still refuse the claimed record rather than deleting it. +func TestDeleteTableTokenLosesToRenameClaim(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + e := fs.Get(nsPath, "t") + claimed := map[string][]byte{} + for k, v := range e.Extended { + claimed[k] = v + } + var meta tableMetadataInternal + require.NoError(t, json.Unmarshal(claimed[ExtendedKeyMetadata], &meta)) + meta.VersionToken = "tok" + metaBytes, err := json.Marshal(&meta) + require.NoError(t, err) + claimed[ExtendedKeyMetadata] = metaBytes + claimed[ExtendedKeyDeleting] = []byte("rename") + fs.Put(nsPath, "t", claimed) + + status, resp := runHandler(t, m, fs.Client, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + VersionToken: "tok", + }) + assert.Equal(t, http.StatusConflict, status) + assert.Equal(t, ErrCodeConflict, resp.Type) + + src := fs.Get(nsPath, "t") + require.NotNil(t, src) + assert.NotNil(t, src.Extended[ExtendedKeyMetadata], "the claimed record must survive") + assert.NotNil(t, fs.Get(GetTablePath(renameTestBucket, "ns", "t"), "data")) +} diff --git a/weed/s3api/s3tables/handler_delete_replacement_test.go b/weed/s3api/s3tables/handler_delete_replacement_test.go new file mode 100644 index 0000000000..4c4f73b448 --- /dev/null +++ b/weed/s3api/s3tables/handler_delete_replacement_test.go @@ -0,0 +1,264 @@ +package s3tables + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "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" +) + +// pausedLookupClient holds the first lookup of dir/name until release is +// closed, after closing reached. A delete reads the table bucket between the +// read that authorizes it and the read that claims the record, so holding that +// lookup lands another request exactly there. +type pausedLookupClient struct { + filer_pb.SeaweedFilerClient + dir string + name string + reached chan struct{} + release chan struct{} + once sync.Once +} + +func newPausedLookupClient(client filer_pb.SeaweedFilerClient, dir, name string) *pausedLookupClient { + return &pausedLookupClient{ + SeaweedFilerClient: client, + dir: dir, + name: name, + reached: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *pausedLookupClient) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + if req.Directory == c.dir && req.Name == c.name { + c.once.Do(func() { + close(c.reached) + <-c.release + }) + } + return c.SeaweedFilerClient.LookupDirectoryEntry(ctx, req, opts...) +} + +func createViewRequest(t *testing.T, name string) *CreateViewRequest { + return &CreateViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: name, + MetadataLocation: "s3://" + renameTestBucket + "/ns/" + name + "/metadata/v1.metadata.json", + } +} + +func deleteViewRequest(t *testing.T, name, token string) *DeleteViewRequest { + return &DeleteViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: name, + VersionToken: token, + } +} + +func deleteTableRequest(t *testing.T, name string) *DeleteTableRequest { + return &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: name, + } +} + +func getViewRequest(t *testing.T, name string) *GetViewRequest { + return &GetViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: name, + } +} + +func getTableRequest(t *testing.T, name string) *GetTableRequest { + return &GetTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: name, + } +} + +// versionTokenOf returns the token the record under name carries. +func versionTokenOf(t *testing.T, fs *s3tablestest.MemFiler, name string) string { + t.Helper() + entry := fs.Get(GetNamespacePath(renameTestBucket, "ns"), name) + require.NotNil(t, entry) + var metadata tableMetadataInternal + require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata)) + return metadata.VersionToken +} + +// requireOK runs an operation that has to succeed. +func requireOK(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, operation string, input interface{}) { + t.Helper() + status, resp := runHandler(t, m, fs.Client, operation, input) + require.Equal(t, http.StatusOK, status, "%s: %+v", operation, resp) +} + +// A delete authorized on one record must not remove a record that took its name +// afterwards. Another delete removes the view and a create reuses the name while +// the first delete is held between its two reads. +func TestDeleteViewSparesAReplacement(t *testing.T) { + const name = "v" + cases := []struct { + name string + replace func(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) + withToken bool + wantStatus int + wantType string + wantMessage string + read string + readInput func(t *testing.T) interface{} + }{ + { + name: "table without token", + replace: func(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) { + requireOK(t, m, fs, "CreateTable", createTableRequest(t, "ns", name)) + }, + wantStatus: http.StatusNotFound, + wantType: ErrCodeNoSuchView, + read: "GetTable", + readInput: func(t *testing.T) interface{} { return getTableRequest(t, name) }, + }, + { + name: "view with the old token", + replace: func(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) { + requireOK(t, m, fs, "CreateView", createViewRequest(t, name)) + }, + withToken: true, + wantStatus: http.StatusConflict, + wantType: ErrCodeConflict, + wantMessage: "version token mismatch", + read: "GetView", + readInput: func(t *testing.T) interface{} { return getViewRequest(t, name) }, + }, + { + name: "view without token", + replace: func(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) { + requireOK(t, m, fs, "CreateView", createViewRequest(t, name)) + }, + wantStatus: http.StatusConflict, + wantType: ErrCodeConflict, + wantMessage: "changed during delete", + read: "GetView", + readInput: func(t *testing.T) interface{} { return getViewRequest(t, name) }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fs, m := startRenameManager(t) + requireOK(t, m, fs, "CreateView", createViewRequest(t, name)) + oldToken := versionTokenOf(t, fs, name) + token := "" + if tc.withToken { + token = oldToken + } + + client := newPausedLookupClient(fs.Client, TablesPath, renameTestBucket) + deleted := startHandler(t, m, client, "DeleteView", deleteViewRequest(t, name, token)) + waitForPause(t, client.reached, deleted) + + requireOK(t, m, fs, "DeleteView", deleteViewRequest(t, name, "")) + tc.replace(t, m, fs) + close(client.release) + + got := <-deleted + assert.Equal(t, tc.wantStatus, got.status, "%+v", got.resp) + assert.Equal(t, tc.wantType, got.resp.Type) + assert.Contains(t, got.resp.Message, tc.wantMessage) + + status, resp := runHandler(t, m, fs.Client, tc.read, tc.readInput(t)) + assert.Equal(t, http.StatusOK, status, "the replacement must survive: %+v", resp) + survivor := fs.Get(GetNamespacePath(renameTestBucket, "ns"), name) + require.NotNil(t, survivor) + assert.NotContains(t, survivor.Extended, ExtendedKeyDeleting, "a refused delete must not leave its claim") + }) + } +} + +// The same gap in DeleteTable: a view that takes the table's name while the +// delete is held between its reads is not the record the delete authorized. +func TestDeleteTableSparesAReplacementView(t *testing.T) { + cases := []struct { + name string + // seed makes the table to delete and returns its name and the location + // the replacement view points at, so a delete that wrongly took the view + // would purge from there. + seed func(fs *s3tablestest.MemFiler) (name, location string) + }{ + { + name: "colocated", + seed: func(fs *s3tablestest.MemFiler) (string, string) { + return "t", "s3://" + renameTestBucket + "/ns/t/metadata/v3.metadata.json" + }, + }, + { + name: "decoupled", + seed: func(fs *s3tablestest.MemFiler) (string, string) { + location := "s3://" + renameTestBucket + "/ns/dt-x/metadata/v1.metadata.json" + meta, _ := json.Marshal(tableMetadataInternal{ + Name: "dt", + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataLocation: location, + }) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "dt", map[string][]byte{ + ExtendedKeyMetadata: meta, + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), "dt-x", nil) + fs.Put(GetTablePath(renameTestBucket, "ns", "dt-x"), "metadata", nil) + return "dt", location + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fs, m := startRenameManager(t) + name, location := tc.seed(fs) + + client := newPausedLookupClient(fs.Client, TablesPath, renameTestBucket) + deleted := startHandler(t, m, client, "DeleteTable", deleteTableRequest(t, name)) + waitForPause(t, client.reached, deleted) + + requireOK(t, m, fs, "DeleteTable", deleteTableRequest(t, name)) + // CreateView refuses a name that a decoupled delete left a directory + // on, so the replacement is put the way a create leaves it. + viewMeta, _ := json.Marshal(tableMetadataInternal{ + Name: name, + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + VersionToken: "replacement-token", + MetadataLocation: location, + }) + fs.Put(GetNamespacePath(renameTestBucket, "ns"), name, map[string][]byte{ + ExtendedKeyMetadata: viewMeta, + ExtendedKeyEntryType: []byte(EntryTypeView), + }) + close(client.release) + + got := <-deleted + assert.Equal(t, http.StatusNotFound, got.status, "%+v", got.resp) + assert.Equal(t, ErrCodeNoSuchTable, got.resp.Type) + + status, resp := runHandler(t, m, fs.Client, "GetView", getViewRequest(t, name)) + assert.Equal(t, http.StatusOK, status, "the replacement view must survive: %+v", resp) + survivor := fs.Get(GetNamespacePath(renameTestBucket, "ns"), name) + require.NotNil(t, survivor) + assert.NotContains(t, survivor.Extended, ExtendedKeyDeleting, "a refused delete must not leave its claim") + }) + } +} diff --git a/weed/s3api/s3tables/handler_maintenance.go b/weed/s3api/s3tables/handler_maintenance.go index 34d42cba3d..97890d3742 100644 --- a/weed/s3api/s3tables/handler_maintenance.go +++ b/weed/s3api/s3tables/handler_maintenance.go @@ -261,7 +261,7 @@ func (h *S3TablesHandler) putMaintenanceConfiguration( value *MaintenanceConfigurationValue, ) error { return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.mutateEntryExtended(r.Context(), client, resourcePath, func(extended map[string][]byte) error { + return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { if len(extended[ExtendedKeyMetadata]) == 0 { return fmt.Errorf("%w: %s", filer_pb.ErrNotFound, resourcePath) } diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index 7aceece5ca..c54ffefe49 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -412,9 +412,20 @@ func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.R } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.setExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata, metadataBytes) + return h.mutateCatalogRecord(r.Context(), client, namespacePath, func(extended map[string][]byte) error { + extended[ExtendedKeyMetadata] = metadataBytes + return nil + }) }) if err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "namespace changed during update, retry the request") + return err + } + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", flattenNamespace(req.Namespace))) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update namespace") return err } @@ -694,41 +705,93 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R return ErrAccessDenied } + // Claim before scanning, so a create that writes after the scan sees it. + var claim string + var claimed map[string][]byte + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + var claimErr error + claim, claimed, claimErr = h.claimForDelete(r.Context(), client, namespacePath, isNamespaceEntry) + return claimErr + }) + if err != nil { + switch { + case errors.Is(err, filer_pb.ErrNotFound): + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", flattenNamespace(req.Namespace))) + case errors.Is(err, ErrConcurrentUpdate): + h.writeError(w, http.StatusConflict, ErrCodeConflict, "namespace changed during delete, retry the request") + default: + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to claim namespace for delete: %v", err)) + } + return err + } + releaseClaim := func() { + _ = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + h.releaseDeleteClaim(r.Context(), client, namespacePath, claim) + return nil + }) + } + // Check if namespace is empty hasChildren := false err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{ - Directory: namespacePath, - Limit: 10, - }) - if err != nil { - return err - } - - for { - entry, err := resp.Recv() + lastFileName := "" + for !hasChildren { + resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{ + Directory: namespacePath, + Limit: 10, + StartFromFileName: lastFileName, + InclusiveStartFrom: lastFileName == "", + }) if err != nil { - if err == io.EOF { - break - } return err } - if entry.Entry != nil && !strings.HasPrefix(entry.Entry.Name, ".") { - // Only consider it a child if it's a table (has metadata) or a file. - // Empty directories left by dropped tables or S3 clients should not - // prevent the namespace from being deleted. + + hasMore := false + for { + entry, err := resp.Recv() + if err != nil { + if err == io.EOF { + break + } + return err + } + if entry.Entry == nil { + continue + } + hasMore = true + lastFileName = entry.Entry.Name + if strings.HasPrefix(entry.Entry.Name, ".") { + continue + } + // A record still marked by its create is absent to readers and + // cannot confirm under the claim this delete holds, so it is no + // child — the delete removes it with the namespace. + if isUnfinishedCreate(entry.Entry.Extended) { + continue + } + // Only consider it a child if it's a table (has metadata), a file, + // or a record an in-flight operation claimed — a rename's cleared + // source still holds its claim, and removing its directory takes + // the data the destination keeps pointing at. Bare directories + // left by dropped tables or S3 clients should not prevent the + // namespace from being deleted. _, hasMetadata := entry.Entry.Extended[ExtendedKeyMetadata] - if hasMetadata || !entry.Entry.IsDirectory { + _, claimed := entry.Entry.Extended[ExtendedKeyDeleting] + if hasMetadata || claimed || !entry.Entry.IsDirectory { hasChildren = true break } } + if !hasMore { + break + } } return nil }) if err != nil { + releaseClaim() if errors.Is(err, filer_pb.ErrNotFound) { h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", flattenNamespace(req.Namespace))) } else { @@ -738,17 +801,23 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R } if hasChildren { + releaseClaim() h.writeError(w, http.StatusConflict, ErrCodeNamespaceNotEmpty, "namespace is not empty") return fmt.Errorf("namespace not empty") } // Delete the namespace err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.deleteDirectory(r.Context(), client, namespacePath) + return h.deleteDirectoryIfMatches(r.Context(), client, namespacePath, &filer_pb.Entry{Extended: claimed}) }) if err != nil { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete namespace") + releaseClaim() + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "namespace changed during delete, retry the request") + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete namespace") + } return err } diff --git a/weed/s3api/s3tables/handler_policy.go b/weed/s3api/s3tables/handler_policy.go index c6c6002a7f..947fa1fef1 100644 --- a/weed/s3api/s3tables/handler_policy.go +++ b/weed/s3api/s3tables/handler_policy.go @@ -400,7 +400,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.mutateEntryExtended(r.Context(), client, tablePath, func(extended map[string][]byte) error { + return h.mutateCatalogRecord(r.Context(), client, tablePath, func(extended map[string][]byte) error { if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(tablePolicy)) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) } @@ -412,6 +412,8 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re if err != nil { if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "table policy changed concurrently, retry the request") + } else if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) } else { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table policy") } @@ -766,10 +768,25 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque return fmt.Errorf("failed to marshal tags: %w", err) } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.setExtendedAttribute(r.Context(), client, resourcePath, extendedKey, tagsBytes) + return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { + extended[extendedKey] = tagsBytes + return nil + }) }) if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + errorCode := ErrCodeNoSuchBucket + if rType == ResourceTypeTable { + errorCode = ErrCodeNoSuchTable + } + h.writeError(w, http.StatusNotFound, errorCode, "resource not found") + return err + } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "resource tags changed concurrently, retry the request") + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to tag resource") return err } @@ -1024,10 +1041,25 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req return fmt.Errorf("failed to marshal tags: %w", err) } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.setExtendedAttribute(r.Context(), client, resourcePath, extendedKey, tagsBytes) + return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { + extended[extendedKey] = tagsBytes + return nil + }) }) if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + errorCode := ErrCodeNoSuchBucket + if rType == ResourceTypeTable { + errorCode = ErrCodeNoSuchTable + } + h.writeError(w, http.StatusNotFound, errorCode, "resource not found") + return err + } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "resource tags changed concurrently, retry the request") + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to untag resource") return err } diff --git a/weed/s3api/s3tables/handler_rename_test.go b/weed/s3api/s3tables/handler_rename_test.go index 326a3f3bb7..e21ec2ede1 100644 --- a/weed/s3api/s3tables/handler_rename_test.go +++ b/weed/s3api/s3tables/handler_rename_test.go @@ -168,6 +168,67 @@ func TestRenameTableDestNamespaceMissing(t *testing.T) { assert.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t"), "source must be untouched") } +// A delete that purges the source between the rename's read and its claim loses +// the source, not the destination namespace — the answer must name the table. +func TestRenameTableSourceDeletedMidRename(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.BeforeUpdate = func() { + _, _ = fs.Client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{ + Directory: nsPath, + Name: "t", + IsRecursive: true, + IsDeleteData: true, + }) + } + + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) + assert.Nil(t, fs.Get(nsPath, "t2"), "a failed rename must not publish the destination") +} + +// A source that vanishes after the destination is claimed must not leave the +// destination behind: the failed rename removes the marked record on the way +// out, before it is ever published. +func TestRenameTableBacksOutConfirmedDest(t *testing.T) { + fs, m := startRenameManager(t) + + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.OnUpdate = func(dir, name string) { + if dir == nsPath && name == "t2" { + _, _ = fs.Client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{ + Directory: nsPath, + Name: "t", + IsRecursive: true, + IsDeleteData: true, + }) + } + } + + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns"}, + DestName: "t2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) + assert.Nil(t, fs.Get(nsPath, "t2"), "a failed rename must not leave the claimed destination behind") +} + // A principal allowed to rename the source must still be denied when it cannot // create a table in the destination namespace. The denial reports the same // not-found as a missing destination namespace so the caller cannot probe for diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 46687901b9..97859fefdc 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -63,6 +63,11 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque if err != nil { return err } + if isUnfinishedCreate(entry.Extended) { + // Another create wrote this record and may still back it out, so + // it is not an existing table; the claim below refuses the name. + return filer_pb.ErrNotFound + } if EntryType(entry.Extended) == EntryTypeView { existingIsView = true return nil @@ -153,11 +158,15 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque attrs := map[string][]byte{ ExtendedKeyMetadata: metadataBytes, ExtendedKeyEntryType: []byte(EntryTypeTable), + ExtendedKeyDeleting: []byte(createClaim), } if len(tagsBytes) > 0 { attrs[ExtendedKeyTags] = tagsBytes } - return h.claimCatalogEntry(r.Context(), client, tablePath, attrs) + if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { + return err + } + return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceEntry, tablePath, attrs, dataPath, tablePath) }) if err != nil { @@ -165,6 +174,13 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName)) return err } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during create, retry the request") + return err + } + if h.writeNamespaceClosedError(w, err, namespaceName) { + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create table") return err } @@ -240,13 +256,20 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req tablePath := GetTablePath(bucketName, namespaceName, tableName) - // Table must be absent. + // Table must be absent; a record still marked by its unfinished create is. var existingMetadata tableMetadataInternal err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata) + entry, err := h.lookupEntry(r.Context(), client, tablePath) if err != nil { return err } + if isUnfinishedCreate(entry.Extended) { + return filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return ErrAttributeNotFound + } return json.Unmarshal(data, &existingMetadata) }) if err == nil { @@ -278,19 +301,31 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req return fmt.Errorf("failed to marshal metadata: %w", err) } + attrs := map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + ExtendedKeyDeleting: []byte(createClaim), + } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { if err := h.ensureDirectory(r.Context(), client, tablePath); err != nil { return err } - return h.claimCatalogEntry(r.Context(), client, tablePath, map[string][]byte{ - ExtendedKeyMetadata: metadataBytes, - }) + if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { + return err + } + return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceEntry, tablePath, attrs, tablePath) }) if err != nil { if errors.Is(err, ErrNameTaken) { h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", tableName)) return err } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during register, retry the request") + return err + } + if h.writeNamespaceClosedError(w, err, namespaceName) { + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to register table") return err } @@ -729,8 +764,9 @@ func (h *S3TablesHandler) listTablesWithClient(r *http.Request, client filer_pb. continue } - // Views share the table layout; exclude them from table listings. - if EntryType(entry.Entry.Extended) == EntryTypeView { + // Views share the table layout; exclude them from table listings, + // and a record its create has not confirmed is not a table yet. + if EntryType(entry.Entry.Extended) == EntryTypeView || isUnfinishedCreate(entry.Entry.Extended) { continue } @@ -1067,12 +1103,12 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque claimed := snapshotExtended(tableEntry.Extended) claimed[ExtendedKeyDeleting] = []byte("delete") if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { - if relErr := h.releaseCatalogRecord(r.Context(), client, tablePath, claimed); relErr != nil { + if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, tablePath, claimed); relErr != nil { glog.Warningf("delete table %s: release claim: %v", tablePath, relErr) } return err } - err := h.removeExtendedAttributesIf(r.Context(), client, tablePath, claimed, + err := h.removeExtendedAttributesIf(context.WithoutCancel(r.Context()), client, tablePath, claimed, ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting) if err != nil { @@ -1083,18 +1119,50 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque } return nil } - if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { + // A no-token decoupled delete claims the record like the token path + // above: a rename holding the record turns the delete away instead + // of losing its data to the purge. + entry, err := h.lookupEntry(r.Context(), client, tablePath) + if err != nil { return err } - return h.removeExtendedAttributes(r.Context(), client, tablePath, + // Authorization ran on the first read; only that same record may be + // purged. + if err := ensureAuthorizedTable(entry, metadata); err != nil { + return err + } + // dataPath was resolved on the earlier read, but the claim and the + // purge run on this one: an update that moved the location in + // between must send the delete back instead of purging a stale dir. + if purge, err := tablePurgeDir(entry, tablePath, bucketName); err != nil { + return err + } else if purge != dataPath { + return ErrConcurrentUpdate + } + snapshot := snapshotExtended(entry.Extended) + if err := h.claimCatalogRecord(r.Context(), client, tablePath, snapshot, "delete"); err != nil { + return err + } + claimed := snapshotExtended(entry.Extended) + claimed[ExtendedKeyDeleting] = []byte("delete") + if err := h.deleteDirectory(r.Context(), client, dataPath); err != nil { + if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, tablePath, claimed); relErr != nil { + glog.Warningf("delete table %s: release claim: %v", tablePath, relErr) + } + return err + } + return h.removeExtendedAttributesIf(context.WithoutCancel(r.Context()), client, tablePath, claimed, ExtendedKeyMetadata, ExtendedKeyMetadataVersion, ExtendedKeyPolicy, ExtendedKeyTags, ExtendedKeyEntryType, - ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus) + ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, ExtendedKeyDeleting) } // Colocated table: the name path holds the data. A token caller's // delete carries the captured attributes as a filer precondition, so // compare and removal are one operation under the path lock: an xattr // write that keeps Mtime, or a delete+recreate in between, refuses it. if req.VersionToken != "" && tableEntry != nil { + if claim := string(tableEntry.Extended[ExtendedKeyDeleting]); claim != "" && claim != "delete" { + return ErrVersionTokenMismatch + } if err := h.deleteDirectoryIfMatches(r.Context(), client, tablePath, tableEntry); err != nil { if errors.Is(err, ErrConcurrentUpdate) { return ErrVersionTokenMismatch @@ -1103,7 +1171,36 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque } return nil } - return h.deleteDirectory(r.Context(), client, tablePath) + // The no-token delete claims the record before removing it, so a + // rename or another delete holding the claim loses the record instead + // of the directory under it. + entry, err := h.lookupEntry(r.Context(), client, tablePath) + if err != nil { + return err + } + if err := ensureAuthorizedTable(entry, metadata); err != nil { + return err + } + // Same staleness check as the decoupled path: a location update that + // landed since the authorized read turns this delete away, or it would + // remove the name path the record no longer stores its data under. + if purge, err := tablePurgeDir(entry, tablePath, bucketName); err != nil { + return err + } else if purge != tablePath { + return ErrConcurrentUpdate + } + claimed := snapshotExtended(entry.Extended) + claimed[ExtendedKeyDeleting] = []byte("delete") + if err := h.claimCatalogRecord(r.Context(), client, tablePath, snapshotExtended(entry.Extended), "delete"); err != nil { + return err + } + if err := h.deleteDirectoryIfMatches(r.Context(), client, tablePath, &filer_pb.Entry{Extended: claimed}); err != nil { + if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, tablePath, claimed); relErr != nil { + glog.Warningf("delete table %s: release claim: %v", tablePath, relErr) + } + return err + } + return nil }) if err != nil { @@ -1111,6 +1208,14 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch") return err } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during delete, retry the request") + return err + } + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table") return err } @@ -1192,9 +1297,10 @@ func (h *S3TablesHandler) readTableMetadata(ctx context.Context, client filer_pb // tableMetadata extracts the table-metadata xattr from an entry. A view stored // at the same name is reported absent, the way GetTable and the renamer's -// entry-type check already treat kind mismatches. +// entry-type check already treat kind mismatches, and so is a record its +// create has not finished writing. func tableMetadata(entry *filer_pb.Entry) ([]byte, error) { - if EntryType(entry.Extended) == EntryTypeView { + if EntryType(entry.Extended) == EntryTypeView || isUnfinishedCreate(entry.Extended) { return nil, filer_pb.ErrNotFound } data, ok := entry.Extended[ExtendedKeyMetadata] @@ -1204,6 +1310,48 @@ func tableMetadata(entry *filer_pb.Entry) ([]byte, error) { return data, nil } +// ensureAuthorizedTable reports whether entry is still the table record a +// delete authorized on its first read. A view, an unfinished create or a record +// whose metadata is gone is absent; a record with another version token took +// the name since, and the authorization does not cover it. +func ensureAuthorizedTable(entry *filer_pb.Entry, authorized tableMetadataInternal) error { + data, err := tableMetadata(entry) + if err != nil { + if errors.Is(err, ErrAttributeNotFound) { + return filer_pb.ErrNotFound + } + return err + } + var current tableMetadataInternal + if err := json.Unmarshal(data, ¤t); err != nil { + return fmt.Errorf("failed to unmarshal table metadata: %w", err) + } + if current.VersionToken != authorized.VersionToken { + return ErrConcurrentUpdate + } + return nil +} + +// tablePurgeDir resolves where a delete must purge from the metadata stored on +// entry, the way handleDeleteTable resolved dataPath on its authorized read: a +// location inside the bucket and off the name path is decoupled, anything else +// leaves the data under the name path. +func tablePurgeDir(entry *filer_pb.Entry, tablePath, bucketName string) (string, error) { + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return "", filer_pb.ErrNotFound + } + var metadata tableMetadataInternal + if err := json.Unmarshal(data, &metadata); err != nil { + return "", fmt.Errorf("failed to unmarshal table metadata: %w", err) + } + dataPath := TableDataDirFromMetadataLocation(metadata.MetadataLocation) + if dataPath != "" && dataPath != tablePath && strings.HasPrefix(dataPath+"/", GetTableBucketPath(bucketName)+"/") { + return dataPath, nil + } + return tablePath, nil +} + // canReadEntry reports whether the principal may read the catalog entry a // denied write would touch. A denied write is then answered with 403 — the // refusal leaks nothing a reader could not load anyway — while a caller that @@ -1239,6 +1387,15 @@ func (h *S3TablesHandler) handleRenameView(w http.ResponseWriter, r *http.Reques return h.renameCatalogEntry(w, r, filerClient, viewEntryKind) } +// markSourceGone tags a missing source path so the rename reports it under the +// source name instead of the closed-parent mapping naming the destination. +func markSourceGone(err error) error { + if errors.Is(err, filer_pb.ErrNotFound) { + return fmt.Errorf("%w: %w", errSourceGone, err) + } + return err +} + func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Request, filerClient FilerClient, kind catalogEntryKind) error { var req RenameTableRequest if err := h.readRequestBody(r, &req); err != nil { @@ -1375,8 +1532,9 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ } // Tables and views share the namespace directory, so a rename must not pick - // up the other kind under the same name. - if EntryType(srcExtended) != kind.entryType { + // up the other kind under the same name, nor a record its create may still + // back out. + if EntryType(srcExtended) != kind.entryType || isUnfinishedCreate(srcExtended) { h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName)) return fmt.Errorf("%s %s not found", kind.noun, srcName) } @@ -1432,9 +1590,10 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ } else if !errors.Is(err, ErrAttributeNotFound) { return fmt.Errorf("failed to fetch destination namespace policy: %w", err) } - if _, err := h.getExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata); err == nil { - destExists = true - } else if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrAttributeNotFound) { + if destEntry, err := h.lookupEntry(r.Context(), client, destPath); err == nil { + _, hasMetadata := destEntry.Extended[ExtendedKeyMetadata] + destExists = hasMetadata && !isUnfinishedCreate(destEntry.Extended) + } else if !errors.Is(err, filer_pb.ErrNotFound) { return err } return nil @@ -1473,6 +1632,9 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.writeError(w, http.StatusConflict, kind.existsCode, fmt.Sprintf("%s %s already exists", kind.noun, destName)) return ErrTableAlreadyExists } + if err := h.refuseClaimedNamespace(w, r, filerClient, destNamespacePath, destNamespace); err != nil { + return err + } metadata.Name = destName metadata.Namespace = destNamespace @@ -1490,61 +1652,177 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ // data a delete already removed. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { if err := h.claimCatalogRecord(r.Context(), client, srcPath, snapshotExtended(srcExtended), "rename"); err != nil { - return err + return markSourceGone(err) } claimed := snapshotExtended(srcExtended) claimed[ExtendedKeyDeleting] = []byte("rename") releaseClaim := func(err error) error { - if relErr := h.releaseCatalogRecord(r.Context(), client, srcPath, claimed); relErr != nil { + if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, srcPath, claimed); relErr != nil { glog.Warningf("rename %s: release claim: %v", srcPath, relErr) } return err } - if err := h.createDirectory(r.Context(), client, destPath); err != nil { + if err := h.ensureDirectory(r.Context(), client, destPath); err != nil { return releaseClaim(err) } - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata, metadataBytes); err != nil { - return releaseClaim(err) - } - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyEntryType, []byte(kind.entryType)); err != nil { - return releaseClaim(err) + written := map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + ExtendedKeyEntryType: []byte(kind.entryType), + ExtendedKeyDeleting: []byte(renameDestClaim), } if len(metadataVersionXattr) > 0 { - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadataVersion, metadataVersionXattr); err != nil { - return releaseClaim(err) - } + written[ExtendedKeyMetadataVersion] = metadataVersionXattr } if len(tableTags) > 0 { tagsBytes, err := json.Marshal(tableTags) if err != nil { return releaseClaim(fmt.Errorf("failed to marshal tags: %w", err)) } - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyTags, tagsBytes); err != nil { - return releaseClaim(err) - } + written[ExtendedKeyTags] = tagsBytes } if tablePolicy != "" { - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyPolicy, []byte(tablePolicy)); err != nil { - return releaseClaim(err) - } + written[ExtendedKeyPolicy] = []byte(tablePolicy) } if len(maintenanceXattr) > 0 { - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMaintenance, maintenanceXattr); err != nil { - return releaseClaim(err) - } + written[ExtendedKeyMaintenance] = maintenanceXattr } if len(maintenanceStatusXattr) > 0 { - if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMaintenanceStatus, maintenanceStatusXattr); err != nil { - return releaseClaim(err) + written[ExtendedKeyMaintenanceStatus] = maintenanceStatusXattr + } + // The destination goes up marked like a create, but under the rename's + // own claim so a create cannot adopt a record the rename still owes a + // publish or a keep. It is confirmed only while its namespace stays + // open: a delete that claims the namespace during the rename must find + // this record refused, not take a renamed table whose source the + // soft-delete below already released. + if err := h.claimCatalogEntry(r.Context(), client, destPath, written); err != nil { + return releaseClaim(err) + } + // The destination stays marked — and so invisible to every other + // operation — until the source clears. Confirming it first would publish + // a record that a failed source clear would then remove from under + // callers that already saw it. + backOutDest := func(err error) error { + ctx := context.WithoutCancel(r.Context()) + keys := make([]string, 0, len(written)) + for key := range written { + keys = append(keys, key) } + if backErr := h.removeExtendedAttributesIf(ctx, client, destPath, written, keys...); backErr != nil && !errors.Is(backErr, filer_pb.ErrNotFound) { + glog.Warningf("rename %s: back out destination: %v", destPath, backErr) + } + h.removeLeftoverDirectory(ctx, client, destPath) + return releaseClaim(err) + } + destParent, err := h.openParent(r.Context(), client, destNamespacePath, isNamespaceEntry) + if err != nil { + return backOutDest(err) } // Soft-delete the source catalog identity in place: drop its catalog xattrs // so the name stops resolving while the metadata/ and data/ children stay put - // (manifests embed absolute paths, so the data must not move). - err := h.removeExtendedAttributesIf(r.Context(), client, srcPath, claimed, - append(append([]string{}, renamedTableAttributes...), ExtendedKeyDeleting)...) + // (manifests embed absolute paths, so the data must not move). The rename's + // claim stays on the cleared record until the commit lands: a namespace + // delete scanning it in between must still refuse, or it removes the + // directory the destination keeps pointing at. + err = h.removeExtendedAttributesIf(r.Context(), client, srcPath, claimed, renamedTableAttributes...) if err != nil { - return releaseClaim(err) + return backOutDest(markSourceGone(err)) + } + // Commit on an uncanceled context so a client cancel cannot wedge the + // record half-published. Only this rename can touch its marked record, + // so the compare cannot lose; the namespace check then runs once more, + // since a delete claiming it after the first check scans the still- + // marked destination, skips it, and removes it under a rename that + // already answered success. On a filer error the source's cleared + // attributes go back rather than dropping the table's catalog identity + // entirely. + commitCtx := context.WithoutCancel(r.Context()) + destConfirmed := false + if err = h.removeExtendedAttributesIf(commitCtx, client, destPath, written, ExtendedKeyDeleting); err == nil { + destConfirmed = true + err = h.parentStillOpen(commitCtx, client, destNamespacePath, isNamespaceEntry, destParent) + } + if err != nil { + if destConfirmed { + // The recheck failed after the record went live: strip it, so + // the failed rename leaves no confirmed destination behind. + confirmed := make(map[string][]byte, len(written)) + keys := make([]string, 0, len(written)) + for key, data := range written { + if key == ExtendedKeyDeleting { + continue + } + confirmed[key] = data + keys = append(keys, key) + } + if backErr := h.removeExtendedAttributesIf(commitCtx, client, destPath, confirmed, keys...); backErr != nil && !errors.Is(backErr, filer_pb.ErrNotFound) { + glog.Warningf("rename %s: remove confirmed destination: %v", destPath, backErr) + } else { + h.removeLeftoverDirectory(commitCtx, client, destPath) + } + } + cleared := snapshotExtended(claimed) + for _, key := range renamedTableAttributes { + delete(cleared, key) + } + if rErr := h.mutateEntryExtended(commitCtx, client, srcPath, func(extended map[string][]byte) error { + for key, want := range cleared { + if !bytes.Equal(extended[key], want) { + return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, srcPath) + } + } + // A create that reused the name since the clear owns it now: + // restoring over its attributes would clobber a live table. + for _, key := range renamedTableAttributes { + if _, ok := extended[key]; ok { + return fmt.Errorf("%w: %s recreated on %s", ErrConcurrentUpdate, key, srcPath) + } + } + for key, data := range srcExtended { + if key != ExtendedKeyDeleting { + extended[key] = data + } + } + delete(extended, ExtendedKeyDeleting) + return nil + }); rErr != nil { + glog.Warningf("rename %s: restore source record: %v", srcPath, rErr) + } else if !destConfirmed { + // Source restored: drop the unfinished destination too, since a + // marked record reads as absent but still refuses its name to + // later creates. When the restore fails the marked record is + // the only catalog identity left — keep it for recovery. The + // compare keeps a record that moved on since the failed publish. + keys := make([]string, 0, len(written)) + for key := range written { + keys = append(keys, key) + } + if backErr := h.removeExtendedAttributesIf(commitCtx, client, destPath, written, keys...); backErr != nil && !errors.Is(backErr, filer_pb.ErrNotFound) { + glog.Warningf("rename %s: remove unfinished destination: %v", destPath, backErr) + } else { + h.removeLeftoverDirectory(commitCtx, client, destPath) + } + } + return err + } + // The record is published: lift the source claim so the cleared source + // directory stops refusing its namespace's delete and the old name + // frees for reuse. Nothing else writes the cleared record, so a loss + // here is a transient filer error; retry before letting a leftover + // claim wedge the name and its namespace. + var relErr error + for attempt := 0; attempt < extendedAttributeUpdateAttempts; attempt++ { + if relErr = h.removeExtendedAttributesIf(commitCtx, client, srcPath, + map[string][]byte{ExtendedKeyDeleting: []byte("rename")}, ExtendedKeyDeleting); relErr == nil { + break + } + if errors.Is(relErr, filer_pb.ErrNotFound) || errors.Is(relErr, ErrConcurrentUpdate) { + break + } + time.Sleep(100 * time.Millisecond) + } + if relErr != nil && !errors.Is(relErr, filer_pb.ErrNotFound) { + glog.Warningf("rename %s: release source claim: %v", srcPath, relErr) } return nil }) @@ -1554,6 +1832,17 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.writeError(w, http.StatusConflict, ErrCodeConflict, kind.noun+" changed during rename, retry the request") return err } + if errors.Is(err, ErrNameTaken) { + h.writeError(w, http.StatusConflict, kind.existsCode, fmt.Sprintf("%s %s already exists", kind.noun, destName)) + return ErrTableAlreadyExists + } + if errors.Is(err, errSourceGone) { + h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName)) + return err + } + if h.writeNamespaceClosedError(w, err, destNamespace) { + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to rename "+kind.noun) return err } @@ -1785,7 +2074,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque // 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 { - return h.mutateEntryExtended(r.Context(), client, target.tablePath, func(extended map[string][]byte) error { + return h.mutateCatalogRecord(r.Context(), client, target.tablePath, func(extended map[string][]byte) error { if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata) } @@ -1806,6 +2095,10 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusConflict, ErrCodeConflict, "table was updated concurrently") return ErrVersionTokenMismatch } + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", target.tableName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update metadata") return err } @@ -1958,6 +2251,12 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re return nil, ErrAccessDenied } + if namespaceExists { + if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName); err != nil { + return nil, err + } + } + return &createTableTarget{ bucketName: bucketName, namespaceName: namespaceName, @@ -1966,3 +2265,32 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re ownerAccountID: nsMetadata.OwnerAccountID, }, nil } + +// writeNamespaceClosedError answers a create refused because its namespace is +// being deleted or was deleted, and reports whether err was one of those. +func (h *S3TablesHandler) writeNamespaceClosedError(w http.ResponseWriter, err error, namespaceName string) bool { + switch { + case errors.Is(err, errParentDeleting): + h.writeError(w, http.StatusConflict, ErrCodeConflict, fmt.Sprintf("namespace %s is being deleted", namespaceName)) + case errors.Is(err, errParentGone), errors.Is(err, filer_pb.ErrNotFound): + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName)) + case errors.Is(err, errBackOutFailed): + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("namespace %s was deleted during the request and the new entry may remain", namespaceName)) + default: + return false + } + return true +} + +// refuseClaimedNamespace answers 409 when a delete has claimed the namespace. +// Call it after the authorization check, so denied callers never see the delete. +func (h *S3TablesHandler) refuseClaimedNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient, namespacePath, namespaceName string) error { + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return h.checkParentOpen(r.Context(), client, namespacePath, isNamespaceEntry) + }) + if errors.Is(err, errParentDeleting) { + h.writeNamespaceClosedError(w, err, namespaceName) + return err + } + return nil +} diff --git a/weed/s3api/s3tables/handler_update_conflict_test.go b/weed/s3api/s3tables/handler_update_conflict_test.go new file mode 100644 index 0000000000..0117f736dc --- /dev/null +++ b/weed/s3api/s3tables/handler_update_conflict_test.go @@ -0,0 +1,69 @@ +package s3tables + +import ( + "fmt" + "net/http" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// claimRecord puts a delete claim on an existing catalog entry, the state a +// record is in while another operation owns it. +func claimRecord(t *testing.T, fs *s3tablestest.MemFiler, dir, name string) { + t.Helper() + entry := fs.Get(dir, name) + require.NotNil(t, entry) + claimed := map[string][]byte{} + for k, v := range entry.Extended { + claimed[k] = v + } + claimed[ExtendedKeyDeleting] = []byte("delete") + fs.Put(dir, name, claimed) +} + +// A write landing on a record another operation claimed is an expected +// conflict, not a server fault: the caller must see 409. +func TestUpdateViewClaimedRecordConflicts(t *testing.T) { + fs, m := startRenameManager(t) + seedView(t, fs, "v") + claimRecord(t, fs, GetNamespacePath(renameTestBucket, "ns"), "v") + + status, resp := runHandler(t, m, fs.Client, "UpdateView", &UpdateViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "v", + MetadataLocation: "s3://" + renameTestBucket + "/ns/v/metadata/v2.metadata.json", + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) +} + +func TestTagResourceClaimedRecordConflicts(t *testing.T) { + fs, m := startRenameManager(t) + claimRecord(t, fs, GetNamespacePath(renameTestBucket, "ns"), "t") + + tableARN := fmt.Sprintf("arn:aws:s3tables:%s:%s:bucket/%s/table/ns/t", DefaultRegion, DefaultAccountID, renameTestBucket) + status, resp := runHandler(t, m, fs.Client, "TagResource", &TagResourceRequest{ + ResourceARN: tableARN, + Tags: map[string]string{"k": "v"}, + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) +} + +func TestUntagResourceClaimedRecordConflicts(t *testing.T) { + fs, m := startRenameManager(t) + claimRecord(t, fs, GetNamespacePath(renameTestBucket, "ns"), "t") + + tableARN := fmt.Sprintf("arn:aws:s3tables:%s:%s:bucket/%s/table/ns/t", DefaultRegion, DefaultAccountID, renameTestBucket) + status, resp := runHandler(t, m, fs.Client, "UntagResource", &UntagResourceRequest{ + ResourceARN: tableARN, + TagKeys: []string{"k"}, + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type) +} diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index f1319c7c07..2c2ef5591e 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/weed/s3api/s3tables/handler_view.go @@ -1,6 +1,7 @@ package s3tables import ( + "context" "encoding/json" "errors" "fmt" @@ -9,6 +10,7 @@ import ( "strings" "time" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) @@ -83,6 +85,9 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create view in this namespace") return ErrAccessDenied } + if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName); err != nil { + return err + } viewPath := GetTablePath(bucketName, namespaceName, viewName) @@ -94,6 +99,11 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques if err != nil { return err } + if isUnfinishedCreate(entry.Extended) { + // Another create wrote this record and may still back it out, so + // it is not an existing view; the claim below refuses the name. + return filer_pb.ErrNotFound + } if EntryType(entry.Extended) != EntryTypeView { existingIsTable = true return nil @@ -151,16 +161,28 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques if err := h.ensureDirectory(r.Context(), client, viewPath); err != nil { return err } - return h.claimCatalogEntry(r.Context(), client, viewPath, map[string][]byte{ + attrs := map[string][]byte{ ExtendedKeyMetadata: metadataBytes, ExtendedKeyEntryType: []byte(EntryTypeView), - }) + ExtendedKeyDeleting: []byte(createClaim), + } + if err := h.claimCatalogEntry(r.Context(), client, viewPath, attrs); err != nil { + return err + } + return h.confirmParent(r.Context(), client, namespacePath, isNamespaceEntry, viewPath, attrs, viewPath) }) if err != nil { if errors.Is(err, ErrNameTaken) { h.writeError(w, http.StatusConflict, ErrCodeViewAlreadyExists, fmt.Sprintf("view %s already exists", viewName)) return err } + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed during create, retry the request") + return err + } + if h.writeNamespaceClosedError(w, err, namespaceName) { + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create view") return err } @@ -195,7 +217,9 @@ func (h *S3TablesHandler) handleGetView(w http.ResponseWriter, r *http.Request, if err != nil { return err } - if EntryType(entry.Extended) != EntryTypeView { + // A view whose create has not confirmed its namespace may still be + // backed out, so it is reported absent. + if EntryType(entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Extended) { return filer_pb.ErrNotFound } data, ok := entry.Extended[ExtendedKeyMetadata] @@ -369,8 +393,9 @@ func (h *S3TablesHandler) listViewsInNamespace(r *http.Request, client filer_pb. if prefix != "" && !strings.HasPrefix(entry.Entry.Name, prefix) { continue } - // Only include view entries; skip tables and untagged entries. - if EntryType(entry.Entry.Extended) != EntryTypeView { + // Only include view entries; skip tables, untagged entries, and a + // record its create has not confirmed. + if EntryType(entry.Entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Entry.Extended) { continue } data, ok := entry.Entry.Extended[ExtendedKeyMetadata] @@ -440,7 +465,7 @@ func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Req if err != nil { return err } - if EntryType(entry.Extended) != EntryTypeView { + if EntryType(entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Extended) { return filer_pb.ErrNotFound } data, ok := entry.Extended[ExtendedKeyMetadata] @@ -534,9 +559,20 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.setExtendedAttribute(r.Context(), client, viewPath, ExtendedKeyMetadata, metadataBytes) + return h.mutateCatalogRecord(r.Context(), client, viewPath, func(extended map[string][]byte) error { + extended[ExtendedKeyMetadata] = metadataBytes + return nil + }) }) if err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed concurrently, retry the request") + return err + } + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", target.viewName)) + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to update metadata") return err } @@ -549,6 +585,26 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques return nil } +// completedViewMetadata returns entry's metadata if it is a view whose create +// has finished, checking token against it when token is set. +func completedViewMetadata(entry *filer_pb.Entry, token string) (tableMetadataInternal, error) { + var metadata tableMetadataInternal + if EntryType(entry.Extended) != EntryTypeView || isUnfinishedCreate(entry.Extended) { + return metadata, filer_pb.ErrNotFound + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok { + return metadata, fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) + } + if err := json.Unmarshal(data, &metadata); err != nil { + return metadata, err + } + if token != "" && metadata.VersionToken != token { + return metadata, ErrVersionTokenMismatch + } + return metadata, nil +} + // handleDeleteView deletes a view. func (h *S3TablesHandler) handleDeleteView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { var req DeleteViewRequest @@ -570,20 +626,8 @@ func (h *S3TablesHandler) handleDeleteView(w http.ResponseWriter, r *http.Reques if err != nil { return err } - if EntryType(entry.Extended) != EntryTypeView { - return filer_pb.ErrNotFound - } - data, ok := entry.Extended[ExtendedKeyMetadata] - if !ok { - return fmt.Errorf("%w: %s", ErrAttributeNotFound, ExtendedKeyMetadata) - } - if err := json.Unmarshal(data, &metadata); err != nil { - return err - } - if req.VersionToken != "" && metadata.VersionToken != req.VersionToken { - return ErrVersionTokenMismatch - } - return nil + metadata, err = completedViewMetadata(entry, req.VersionToken) + return err }) if err != nil { if errors.Is(err, filer_pb.ErrNotFound) { @@ -609,9 +653,44 @@ func (h *S3TablesHandler) handleDeleteView(w http.ResponseWriter, r *http.Reques } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.deleteDirectory(r.Context(), client, viewPath) + entry, err := h.lookupEntry(r.Context(), client, viewPath) + if err != nil { + return err + } + current, err := completedViewMetadata(entry, req.VersionToken) + if err != nil { + return err + } + // Authorization ran on the first read; only that same record may be deleted. + if current.VersionToken != metadata.VersionToken { + return ErrConcurrentUpdate + } + claimed := snapshotExtended(entry.Extended) + claimed[ExtendedKeyDeleting] = []byte("delete") + if err := h.claimCatalogRecord(r.Context(), client, viewPath, snapshotExtended(entry.Extended), "delete"); err != nil { + return err + } + if err := h.deleteDirectoryIfMatches(r.Context(), client, viewPath, &filer_pb.Entry{Extended: claimed}); err != nil { + if relErr := h.releaseCatalogRecord(context.WithoutCancel(r.Context()), client, viewPath, claimed); relErr != nil { + glog.Warningf("delete view %s: release claim: %v", viewPath, relErr) + } + return err + } + return nil }) if err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "view changed during delete, retry the request") + return err + } + if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchView, fmt.Sprintf("view %s not found", viewName)) + return err + } + if errors.Is(err, ErrVersionTokenMismatch) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "version token mismatch") + return err + } h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete view") return err } diff --git a/weed/s3api/s3tables/handler_view_rename_test.go b/weed/s3api/s3tables/handler_view_rename_test.go index 0afd1be118..3f7149a0aa 100644 --- a/weed/s3api/s3tables/handler_view_rename_test.go +++ b/weed/s3api/s3tables/handler_view_rename_test.go @@ -6,6 +6,7 @@ import ( "path" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" "github.com/stretchr/testify/assert" @@ -71,6 +72,30 @@ func TestRenameViewMovesCatalogPointer(t *testing.T) { "the view's metadata file must survive") } +// A delete that purges the source view between the rename's read and its claim +// loses the source, so the answer must name the view. +func TestRenameViewSourceDeletedMidRename(t *testing.T) { + fs, m := startRenameManager(t) + seedView(t, fs, "v") + + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.BeforeUpdate = func() { + _, _ = fs.Client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{ + Directory: nsPath, + Name: "v", + IsRecursive: true, + IsDeleteData: true, + }) + } + + err := runRenameView(t, m, fs, "v", "v2") + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchView, s3Err.Type) + assert.Nil(t, fs.Get(nsPath, "v2"), "a failed rename must not publish the destination") +} + // Tables and views share one namespace directory, so each rename must refuse // the other kind rather than moving it. func TestRenameViewRejectsTable(t *testing.T) { diff --git a/weed/s3api/s3tables/s3tablestest/memfiler.go b/weed/s3api/s3tables/s3tablestest/memfiler.go index c1047ba9d0..5fb4e309f3 100644 --- a/weed/s3api/s3tables/s3tablestest/memfiler.go +++ b/weed/s3api/s3tables/s3tablestest/memfiler.go @@ -37,6 +37,13 @@ type MemFiler struct { // BeforeDelete runs once before the next DeleteEntry, so a test can land a // competing write in a handler's check-to-delete window. BeforeDelete func() + // OnLookup runs at the start of every LookupDirectoryEntry with the target + // path, so a test can land a competing write between two reads. + OnLookup func(dir, name string) + // OnUpdate runs at the start of every UpdateEntry with the target path, so + // a test can land a competing write at a chosen step of a multi-update + // operation. + OnUpdate func(dir, name string) // RejectDelete returns the reason DeleteEntry refuses a path, answered the // way the filer answers one: no transport error, the reason in the response. RejectDelete func(dir, name string) string @@ -79,6 +86,9 @@ func (f *MemFiler) PutFile(dir, name string, mtime time.Time) { } func (f *MemFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if hook := f.OnLookup; hook != nil { + hook(req.Directory, req.Name) + } if e := f.Get(req.Directory, req.Name); e != nil { return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil } @@ -144,23 +154,27 @@ func (f *MemFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryReque } func (f *MemFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { - // The hook runs before the lock is taken: its whole purpose is to land a + // The hooks run before the lock is taken: their whole purpose is to land a // competing write in the read-to-write window, and that write needs the lock. + if hook := f.OnUpdate; hook != nil { + hook(req.Directory, req.Entry.Name) + } if hook := f.BeforeUpdate; hook != nil { f.BeforeUpdate = nil hook() } f.mu.Lock() defer f.mu.Unlock() + // The real filer looks the entry up before anything else and answers a + // missing one with codes.NotFound, not a failed precondition. + existing := f.entries[req.Directory][req.Entry.Name] + if existing == nil { + return nil, status.Errorf(codes.NotFound, "not found %s: %v", path.Join(req.Directory, req.Entry.Name), filer_pb.ErrNotFound) + } // 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 d, ok := f.entries[req.Directory]; ok { - if existing := d[req.Entry.Name]; existing != nil { - actual = existing.Extended[key] - } - } + actual := existing.Extended[key] if !bytes.Equal(actual, expected) { return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key) } @@ -184,6 +198,10 @@ func (f *MemFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryReque } f.mu.Lock() defer f.mu.Unlock() + // The real filer refuses a non-recursive delete of a non-empty folder. + if child := path.Join(req.Directory, req.Name); !req.IsRecursive && len(f.entries[child]) > 0 { + return &filer_pb.DeleteEntryResponse{Error: "fail to delete non-empty folder: " + child}, nil + } // The real filer evaluates ExpectedExtended under the path lock and refuses // the delete when the entry no longer matches what the caller captured. for key, expected := range req.ExpectedExtended {