fix(s3tables): refuse a create that a namespace delete can lose (#12239)

* s3tables: refuse a create that a namespace delete can lose DeleteNamespace checks the namespace for children and then deletes its directory recursively in a second filer call. The filer locks each path on its own, so a table created under the namespace between the two is not seen by the check and is removed by the delete, while both requests answer 200. The mirror case loses the table too: a create that passed its namespace check before the delete finished re-created the namespace directory through the filer's parent creation and recorded a table in a namespace that no longer exists. Each side now writes before it reads. DeleteNamespace claims the namespace record with a fresh token before it scans, deletes only while the record still holds that claim, and lifts its own claim when it stops. CreateTable, RegisterTable and CreateView re-read the namespace after writing their record and back out when it is claimed or gone: they remove what they wrote and answer 409 or 404 instead of reporting an entry that does not survive, or 500 when the record could not be removed and may remain. An entry the scan missed was written after the claim, so its create sees the claim; a create and a delete can both fail, but no longer both succeed. A create is also refused up front while a claim is held, but only after its authorization check, so a caller that may not create there does not learn of the delete. A create writes its record marked "create" and lifts the mark once the namespace is confirmed open. Until then GetTable, GetView, UpdateTable and a second create treat the record as absent, so no request is told about a table or view that the first create may still back out. * s3tables: keep a create's unfinished record out of reads and listings * s3tables: refuse catalog writes on a record an operation claimed * s3tables: answer 404 when a create's write path loses the namespace * s3tables: confirm the rename destination's namespace like a create * s3tables: cover unfinished-record reads, claimed-record writes, rename claim * s3tables: lift a leftover claim when a no-token delete clears the record * s3tables: answer 409 when a claimed record turns away a write UpdateView, TagResource, and UntagResource surfaced ErrConcurrentUpdate as a 500, so an expected conflict looked like a server fault to the caller. Map it to Conflict the way UpdateTable and the policy handlers already do. * s3tables: name the lost source when a rename finds it deleted * s3tables: report an entry deleted before its update as not found The filer answers an UpdateEntry on an entry removed since the lookup with codes.NotFound. mutateEntryExtended returned that error unchanged instead of filer_pb.ErrNotFound, so a rename whose source was deleted mid-claim answered 500 instead of NoSuchTable or NoSuchView. MemFiler now answers a missing entry the same way, so the rename tests for tables and views take the real filer's path. * s3tables: unpublish the destination when a rename's source clear fails * s3tables: refuse to delete a record another operation claimed * s3tables: release catalog claims on an uncanceled cleanup context * s3tables: answer 404 when a racing delete already removed the table * s3tables: keep the rename destination marked until the source clears * s3tables: guard the rename rollback against a recreated source and an unfinished destination * s3tables: drop the unfinished destination only once the source is restored * s3tables: re-resolve the purge target on the record a delete claims * s3tables: cover a delete that meets a moved metadata location * s3tables: answer 409 when a create's record changed under it * s3tables: adopt a record still marked by its unfinished create * s3tables: skip unfinished creates in the namespace empty check * s3tables: delete only the record the caller was authorized on DeleteView and DeleteTable check permissions on the record they read first, then read the name again to claim it. A record that took the name in between, such as a table or another owner's view created after a concurrent delete, was claimed and removed on the first record's authorization, and DeleteView never compared the caller's version token with it. The second read now repeats the first one's checks: a record of the other kind or an unfinished create answers 404, and a token that no longer matches answers 409. A delete without a token also compares the record's version token with the authorized one and answers 409 when they differ. * s3tables: re-check the parent after a create lifts its mark A create that passed checkParentOpen and paused before lifting its mark could still confirm while a namespace delete held the claim: the delete scans the still-marked record, skips it, and removes it under a create that already answered success. confirmParent now re-reads the same parent record after the mark clears and backs the confirmed record out when the namespace moved on. * s3tables: mark the rename destination with a claim of its own claimCatalogEntry adopted every create-marked record, including a rename's destination: a create landing after the source cleared could replace the destination before the rename published it, and with the source name reused the restore could leave the table on neither name. The destination now marks with the rename's own claim, which readers still treat as unpublished but a create must refuse, and the commit re-checks the destination namespace after the record goes live. The source keeps its rename claim until the commit lands so a namespace delete scanning the cleared record still refuses. * s3tables: compare the parent's createdAt, not its editable metadata A namespace update rewrites ExtendedKeyMetadata in place, so a create or rename whose second parent read landed across an UpdateNamespace saw a changed blob and reported the still-live namespace as gone: creates backed their record out and answered 404, renames undid their destination. createdAt survives the edit but changes on a delete+recreate, so it still identifies the record. * s3tables: retry lifting the rename's claim off the cleared source The release's only failure mode is a transient filer error — nothing else can write the cleared record — yet a single loss left the claim behind: the old name stayed refused to creates and the namespace counted the record as a child. Retry the release before warning. * s3tables: page the namespace scan before deleting it A page of skippable entries — bare directories or unfinished creates — ended the scan with hasChildren false, so a table on a later page was removed with the namespace. Keep listing with StartFromFileName until a child shows or the listing ends. --------- Co-authored-by: chrislusf <chrislusf@users.noreply.github.com>
2026-10-30 17:00:48 +01:00 · 2026-10-30 03:18:52 -07:00
1 parent e80430db8e
commit eba2cb926d
@@ -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)
@@ -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
@@ -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"))
+}
@@ -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"))
+}
@@ -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")
+		})
+	}
+}
@@ -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)
 			}
@@ -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
 	}

@@ -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
 	}
@@ -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
@@ -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, &current); 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
+}
@@ -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)
+}
@@ -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
 	}
@@ -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) {
@@ -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 {