fix(s3tables): keep the data under a deleted namespace (#12276)

* fix(s3tables): keep the data under a deleted namespace

A rename leaves the source table directory holding data the renamed table
still points at, and DeleteNamespace removed it recursively. The delete now
keeps the directory, minus its catalog record, when files remain under it;
CreateNamespace can take it back.

* fix(s3tables): keep hidden-directory data, report name takeover as 409

- DeleteNamespace also looks for files under hidden directories, since a
  table location may sit under one.
- CreateNamespace answers 409 when a file takes the name during the
  take-back of a kept directory.

* s3tables: a vanished leftover holds nothing to keep

A bare directory removed between the namespace listing and the walk
made dirHoldsFiles answer its ListEntries NotFound, and the delete
reported that miss as a 404 or 500 on a live namespace. Skip the gone
directory instead: whatever it held is already gone, so it cannot keep
the namespace.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3tables: a delete's gone entry is a conflict, not a success

The filer answers a precondition delete on an entry that is already gone with FailedPrecondition carrying the not-found text, which DoRemoveRequest reads as a completed delete -- DeleteNamespace answered 200 for a delete that never ran. deleteDirectoryIfMatches now calls DeleteEntry directly, keeping the FailedPrecondition-to-conflict mapping and refusing a response-level error. keepAsLeftover likewise reports a directory removed since the scan as ErrConcurrentUpdate, so the handler answers 409 rather than 500.

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/s3api/s3tables/filer_ops.go

@@ -6,7 +6,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"os"
+	"path"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -90,6 +92,21 @@ func (h *S3TablesHandler) mutateEntryExtended(
 	client filer_pb.SeaweedFilerClient,
 	path string,
 	mutate func(extended map[string][]byte) error,
+) error {
+	return h.mutateEntry(ctx, client, path, func(entry *filer_pb.Entry) error {
+		return mutate(entry.Extended)
+	})
+}
+
+// mutateEntry is mutateEntryExtended for a mutation that also needs the entry
+// itself, such as its type. The entry mutate inspects is the one written back,
+// and the filer refuses to write a directory entry over a file, so a check on
+// the entry's type holds at the write too.
+func (h *S3TablesHandler) mutateEntry(
+	ctx context.Context,
+	client filer_pb.SeaweedFilerClient,
+	path string,
+	mutate func(entry *filer_pb.Entry) error,
 ) error {
 	dir, name := splitPath(path)
 
@@ -107,7 +124,7 @@ func (h *S3TablesHandler) mutateEntryExtended(
 		if entry.Extended == nil {
 			entry.Extended = make(map[string][]byte)
 		}
-		if err := mutate(entry.Extended); err != nil {
+		if err := mutate(entry); err != nil {
 			return err
 		}
 
@@ -197,28 +214,57 @@ func (h *S3TablesHandler) removeExtendedAttributesIf(
 // 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 {
-		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 {
-			extended[key] = data
-		}
-		return nil
+		return claimExtended(extended, path, attrs)
 	})
 }
 
+// claimExtended is claimCatalogEntry's change to the attributes it read.
+func claimExtended(extended map[string][]byte, path string, attrs map[string][]byte) error {
+	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 {
+		extended[key] = data
+	}
+	return nil
+}
+
+// retakeDirectory writes a create's attributes onto a directory that is already
+// at path, the way an exclusive create would have: a leftover that holds no
+// catalog attribute is taken over, and anything else — a file, or a directory
+// that carries a record — is a taken name. The type is checked on the entry the
+// write carries, so a file that replaces the directory cannot be tagged: the
+// filer refuses to write the directory entry over it. It reports that refusal
+// only as an opaque failure, so a failed write is followed by a look at what
+// holds the name now: a file there is a taken name, not a server error.
+func (h *S3TablesHandler) retakeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error {
+	err := h.mutateEntry(ctx, client, path, func(entry *filer_pb.Entry) error {
+		if !entry.IsDirectory {
+			return fmt.Errorf("%w: %s is not a directory", ErrNameTaken, path)
+		}
+		return claimExtended(entry.Extended, path, attrs)
+	})
+	if err == nil || errors.Is(err, ErrNameTaken) {
+		return err
+	}
+	if current, lookupErr := h.lookupEntry(ctx, client, path); lookupErr == nil && !current.IsDirectory {
+		return fmt.Errorf("%w: %s is not a directory", ErrNameTaken, path)
+	}
+	return err
+}
+
 // claimCatalogRecord reserves a catalog record for an in-flight delete or
 // rename under one compare-checked write: the entry must still hold exactly
 // expected. A record claimed by another kind of operation refuses the claim,
@@ -353,6 +399,17 @@ func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer
 	return data, nil
 }
 
+// readNamespaceMetadataAttr returns the namespace-metadata xattr at
+// namespacePath. A directory without it is not a namespace — a delete that kept
+// the data under it leaves one — so its absence reads as a missing entry.
+func (h *S3TablesHandler) readNamespaceMetadataAttr(ctx context.Context, client filer_pb.SeaweedFilerClient, namespacePath string) ([]byte, error) {
+	data, err := h.getExtendedAttribute(ctx, client, namespacePath, ExtendedKeyMetadata)
+	if errors.Is(err, ErrAttributeNotFound) {
+		return nil, fmt.Errorf("%w: %s", filer_pb.ErrNotFound, namespacePath)
+	}
+	return data, err
+}
+
 // loadNamespaceMetadata resolves a namespace to its metadata. A directory that
 // carries no namespace metadata is not a namespace, so a missing attribute
 // reports the same absence as a missing entry and every caller tests one
@@ -360,11 +417,8 @@ func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer
 func (h *S3TablesHandler) loadNamespaceMetadata(ctx context.Context, filerClient FilerClient, bucketName, namespaceName string) (*namespaceMetadata, error) {
 	var metadata namespaceMetadata
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(ctx, client, GetNamespacePath(bucketName, namespaceName), ExtendedKeyMetadata)
+		data, err := h.readNamespaceMetadataAttr(ctx, client, GetNamespacePath(bucketName, namespaceName))
 		if err != nil {
-			if errors.Is(err, ErrAttributeNotFound) {
-				return filer_pb.ErrNotFound
-			}
 			return err
 		}
 		if err := json.Unmarshal(data, &metadata); err != nil {
@@ -419,10 +473,13 @@ func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.S
 // while the entry still carries the extended attributes the caller captured:
 // the filer compares and removes in one operation under the path lock, so an
 // attribute write that keeps Mtime or a delete+recreate in between refuses the
-// delete with ErrConcurrentUpdate instead of losing the replacement.
+// delete with ErrConcurrentUpdate instead of losing the replacement. The raw
+// RPC, not DoRemoveRequest: the filer answers a precondition delete on an
+// entry that is already gone with FailedPrecondition carrying the not-found
+// text, which that helper reads as a completed delete.
 func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, expected *filer_pb.Entry) error {
 	dir, name := splitPath(path)
-	_, err := filer_pb.DoRemoveRequest(ctx, client, &filer_pb.DeleteEntryRequest{
+	resp, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{
 		Directory:            dir,
 		Name:                 name,
 		IsDeleteData:         true,
@@ -433,7 +490,13 @@ func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client f
 	if status.Code(err) == codes.FailedPrecondition {
 		return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
 	}
-	return err
+	if err != nil {
+		return err
+	}
+	if resp != nil && resp.Error != "" {
+		return errors.New(resp.Error)
+	}
+	return nil
 }
 
 var (
@@ -630,6 +693,88 @@ func (h *S3TablesHandler) confirmParent(ctx context.Context, client filer_pb.Sea
 	return err
 }
 
+// dirListPageSize is how many entries dirHoldsFiles reads per ListEntries call.
+const dirListPageSize = 1024
+
+// dirHoldsFiles reports whether any file sits under dirPath, at any depth. It
+// walks breadth-first, pages through each directory, and stops at the first
+// file it sees. Returning at that file abandons the page being read, so the
+// listings run on a context of their own that ends with the call: otherwise
+// the filer keeps producing that page until the whole request is over.
+func (h *S3TablesHandler) dirHoldsFiles(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string) (bool, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	pending := []string{dirPath}
+	for len(pending) > 0 {
+		dir := pending[0]
+		pending = pending[1:]
+
+		lastFileName := ""
+		for {
+			resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
+				Directory:          dir,
+				Limit:              dirListPageSize,
+				StartFromFileName:  lastFileName,
+				InclusiveStartFrom: lastFileName == "",
+			})
+			if err != nil {
+				if status.Code(err) == codes.NotFound {
+					// the directory is gone, so there is nothing under it to keep
+					break
+				}
+				return false, err
+			}
+
+			received := false
+			for {
+				entry, err := resp.Recv()
+				if err != nil {
+					if err == io.EOF || status.Code(err) == codes.NotFound {
+						break
+					}
+					return false, err
+				}
+				if entry.Entry == nil {
+					continue
+				}
+				received = true
+				lastFileName = entry.Entry.Name
+				if !entry.Entry.IsDirectory {
+					return true, nil
+				}
+				pending = append(pending, path.Join(dir, entry.Entry.Name))
+			}
+			if !received {
+				break
+			}
+		}
+	}
+	return false, nil
+}
+
+// keepAsLeftover clears a namespace's catalog attributes and the delete's
+// claim in one write, while the entry still holds claimed, keeping the
+// directory and the data under it. The write compares against claimed so the
+// delete's token still fails an ABA, and it drops the claim itself: a kept
+// directory that carried it could never be claimed again. A directory removed
+// since the scan is a concurrent change like any other the compare would
+// catch, so its not-found answers ErrConcurrentUpdate rather than a filer
+// failure.
+func (h *S3TablesHandler) keepAsLeftover(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, claimed map[string][]byte) error {
+	keys := make([]string, 0, len(s3tablesExtendedKeys))
+	for _, key := range s3tablesExtendedKeys {
+		if _, ok := claimed[key]; ok {
+			keys = append(keys, key)
+		}
+	}
+	err := h.removeExtendedAttributesIf(ctx, client, path, claimed, keys...)
+	if errors.Is(err, filer_pb.ErrNotFound) {
+		return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
+	}
+	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) {

weed/s3api/s3tables/filer_ops_test.go

@@ -73,6 +73,32 @@ func TestDeleteDirectoryDeletesTheSubtree(t *testing.T) {
 	assert.True(t, stub.req.IgnoreRecursiveError)
 }
 
+// The filer answers a precondition delete on an entry that is already gone
+// with FailedPrecondition carrying the not-found text. DoRemoveRequest reads
+// any error containing that text as a completed delete, so the conditional
+// delete must keep the conflict itself: a name that changed under the delete
+// is a concurrent update, not a success.
+func TestDeleteDirectoryIfMatchesReportsAVanishedEntry(t *testing.T) {
+	stub := &deleteEntryStub{err: status.Error(codes.FailedPrecondition,
+		"extended attribute precondition /buckets/b/ns: "+filer_pb.ErrNotFound.Error())}
+
+	err := (&S3TablesHandler{}).deleteDirectoryIfMatches(context.Background(), stub, "/buckets/b/ns",
+		&filer_pb.Entry{Extended: map[string][]byte{ExtendedKeyMetadata: []byte("{}")}})
+
+	assert.ErrorIs(t, err, ErrConcurrentUpdate)
+}
+
+// A precondition delete the filer refuses inside the response is not a
+// success either.
+func TestDeleteDirectoryIfMatchesReportsResponseError(t *testing.T) {
+	stub := &deleteEntryStub{resp: &filer_pb.DeleteEntryResponse{Error: "fail to delete non-empty folder"}}
+
+	err := (&S3TablesHandler{}).deleteDirectoryIfMatches(context.Background(), stub, "/buckets/b/ns", &filer_pb.Entry{})
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "fail to delete non-empty folder")
+}
+
 // The bucket is the directory, so DeleteTableBucket must not answer success
 // when the filer refused to remove it.
 func TestDeleteTableBucketReportsRejectedDirectoryDelete(t *testing.T) {

weed/s3api/s3tables/handler_namespace.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"path"
 	"strings"
 	"time"
 
@@ -172,7 +173,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 
 	// Check if namespace already exists
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		_, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
+		_, err := h.readNamespaceMetadataAttr(r.Context(), client, namespacePath)
 		return err
 	})
 
@@ -200,12 +201,19 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 		return fmt.Errorf("failed to marshal metadata: %w", err)
 	}
 
+	attrs := map[string][]byte{ExtendedKeyMetadata: metadataBytes}
 	var taken error
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		err := h.createDirectoryExclusive(r.Context(), client, namespacePath, map[string][]byte{ExtendedKeyMetadata: metadataBytes})
+		err := h.createDirectoryExclusive(r.Context(), client, namespacePath, attrs)
 		if errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
-			taken = err // an answer, not a filer failure
-			return nil
+			// A directory is already there. If it is a leftover — a delete kept
+			// it for the data under it and cleared its catalog attributes —
+			// take it back; claimCatalogEntry refuses a real namespace.
+			err = h.retakeDirectory(r.Context(), client, namespacePath, attrs)
+			if errors.Is(err, ErrNameTaken) {
+				taken = filer_pb.ErrEntryAlreadyExists // an answer, not a filer failure
+				return nil
+			}
 		}
 		return err
 	})
@@ -216,6 +224,10 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 		h.writeError(w, http.StatusConflict, ErrCodeNamespaceAlreadyExists, fmt.Sprintf("namespace %s already exists", namespaceName))
 		return err
 	}
+	if errors.Is(err, ErrConcurrentUpdate) || errors.Is(err, filer_pb.ErrNotFound) {
+		h.writeError(w, http.StatusConflict, ErrCodeConflict, "namespace changed during create, retry the request")
+		return err
+	}
 	if err != nil {
 		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create namespace")
 		return err
@@ -264,7 +276,7 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ
 	var bucketPolicy string
 	var bucketTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
+		data, err := h.readNamespaceMetadataAttr(r.Context(), client, namespacePath)
 		if err != nil {
 			return err
 		}
@@ -357,7 +369,7 @@ func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.R
 	var bucketPolicy string
 	var bucketTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
+		data, err := h.readNamespaceMetadataAttr(r.Context(), client, namespacePath)
 		if err != nil {
 			return err
 		}
@@ -657,7 +669,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 	var bucketPolicy string
 	var bucketTags map[string]string
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
+		data, err := h.readNamespaceMetadataAttr(r.Context(), client, namespacePath)
 		if err != nil {
 			return err
 		}
@@ -733,6 +745,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 
 	// Check if namespace is empty
 	hasChildren := false
+	keepDir := false
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		lastFileName := ""
 		for !hasChildren {
@@ -760,26 +773,38 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 				}
 				hasMore = true
 				lastFileName = entry.Entry.Name
-				if strings.HasPrefix(entry.Entry.Name, ".") {
-					continue
+				// Hidden entries are never children.
+				if !strings.HasPrefix(entry.Entry.Name, ".") {
+					// 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]
+					_, claimed := entry.Entry.Extended[ExtendedKeyDeleting]
+					if hasMetadata || claimed || !entry.Entry.IsDirectory {
+						hasChildren = true
+						break
+					}
 				}
-				// 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]
-				_, claimed := entry.Entry.Extended[ExtendedKeyDeleting]
-				if hasMetadata || claimed || !entry.Entry.IsDirectory {
-					hasChildren = true
-					break
+				// A bare directory with files under it, hidden or not, is a
+				// renamed-away table's data, which the destination still points
+				// at — a table's location may sit under any directory of the
+				// namespace: the namespace directory stays, minus its catalog
+				// record. One such directory settles it, so the rest are not walked.
+				if entry.Entry.IsDirectory && !keepDir {
+					holdsFiles, err := h.dirHoldsFiles(r.Context(), client, path.Join(namespacePath, entry.Entry.Name))
+					if err != nil {
+						return err
+					}
+					keepDir = holdsFiles
 				}
 			}
 			if !hasMore {
@@ -806,8 +831,11 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 		return fmt.Errorf("namespace not empty")
 	}
 
-	// Delete the namespace
+	// Delete the namespace, or stop it being one while the data under it stays.
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		if keepDir {
+			return h.keepAsLeftover(r.Context(), client, namespacePath, claimed)
+		}
 		return h.deleteDirectoryIfMatches(r.Context(), client, namespacePath, &filer_pb.Entry{Extended: claimed})
 	})
 

weed/s3api/s3tables/handler_namespace_keeps_data_test.go

@@ -0,0 +1,427 @@
+package s3tables
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"path"
+	"testing"
+	"time"
+
+	"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"
+)
+
+// replaceAfterLookupClient swaps the entry at dir/name for a file as soon as
+// its nth lookup has returned, so the next read finds a file where the one
+// before it found a directory.
+type replaceAfterLookupClient struct {
+	filer_pb.SeaweedFilerClient
+	f       *s3tablestest.MemFiler
+	dir     string
+	name    string
+	nth     int
+	lookups int
+}
+
+func (c *replaceAfterLookupClient) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	resp, err := c.SeaweedFilerClient.LookupDirectoryEntry(ctx, req, opts...)
+	if req.Directory == c.dir && req.Name == c.name {
+		c.lookups++
+		if c.lookups == c.nth {
+			c.f.PutFile(c.dir, c.name, time.Now())
+		}
+	}
+	return resp, err
+}
+
+const keptDataFile = "part-0.parquet"
+
+// keptDataDir is where the renamed table's data stays: a rename is catalog-only,
+// so the destination keeps pointing at the original table directory.
+func keptDataDir() string {
+	return path.Join(GetTablePath(renameTestBucket, "ns", "t"), "data")
+}
+
+func listedNamespaces(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) []string {
+	t.Helper()
+	resp := &ListNamespacesResponse{}
+	require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.Client), "ListNamespaces",
+		&ListNamespacesRequest{TableBucketARN: mustBucketARN(t)}, resp, ""))
+	names := make([]string, 0, len(resp.Namespaces))
+	for _, namespace := range resp.Namespaces {
+		names = append(names, flattenNamespace(namespace.Namespace))
+	}
+	return names
+}
+
+// renameTableToNs2 moves ns/t to ns2/t, which leaves ns/t behind as a directory
+// without a record.
+func renameTableToNs2(t *testing.T, m *Manager, fs *s3tablestest.MemFiler) {
+	t.Helper()
+	requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns2"},
+	})
+	requireOK(t, m, fs, "RenameTable", &RenameTableRequest{
+		TableBucketARN:  mustBucketARN(t),
+		SourceNamespace: []string{"ns"},
+		SourceName:      "t",
+		DestNamespace:   []string{"ns2"},
+		DestName:        "t",
+	})
+	require.NotNil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t"), "the renamed-away directory is the leftover under test")
+}
+
+// startDeletedNamespaceWithRenamedData renames ns/t to ns2/t, then deletes ns.
+// The table's data file sits under ns/t, where the renamed table still reads it.
+func startDeletedNamespaceWithRenamedData(t *testing.T) (*s3tablestest.MemFiler, *Manager) {
+	t.Helper()
+	fs, m := startRenameManager(t)
+	fs.PutFile(keptDataDir(), keptDataFile, time.Now())
+	renameTableToNs2(t, m, fs)
+	requireOK(t, m, fs, "DeleteNamespace", deleteNamespaceRequest(t))
+	return fs, m
+}
+
+// A rename leaves the source table directory behind, holding the data the
+// renamed table still points at. Deleting the namespace must not take it: the
+// namespace stops existing, and the directory stays with the data under it.
+func TestDeleteNamespaceKeepsRenamedTableData(t *testing.T) {
+	fs, m := startDeletedNamespaceWithRenamedData(t)
+
+	status, resp := runHandler(t, m, fs.Client, "GetNamespace", &GetNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns"},
+	})
+	assert.Equal(t, http.StatusNotFound, status, "%+v", resp)
+	assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp)
+	assert.Equal(t, []string{"ns2"}, listedNamespaces(t, m, fs))
+
+	assert.NotNil(t, fs.Get(keptDataDir(), keptDataFile), "the renamed table's data must survive")
+	kept := fs.Get(GetTableBucketPath(renameTestBucket), "ns")
+	require.NotNil(t, kept, "the directory holding the data must stay")
+	for _, key := range s3tablesExtendedKeys {
+		assert.NotContains(t, kept.Extended, key, "the kept directory must carry no catalog attribute, claim included")
+	}
+
+	status, resp = runHandler(t, m, fs.Client, "GetTable", &GetTableRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns2"},
+		Name:           "t",
+	})
+	assert.Equal(t, http.StatusOK, status, "the renamed table must still resolve: %+v", resp)
+}
+
+// The kept directory is not a namespace, so every call that names it answers
+// NoSuchNamespace and writes nothing to it.
+func TestKeptDirectoryIsNoSuchNamespace(t *testing.T) {
+	fs, m := startDeletedNamespaceWithRenamedData(t)
+	before := extendedOf(t, fs, GetTableBucketPath(renameTestBucket), "ns")
+
+	ops := []struct {
+		operation string
+		input     interface{}
+	}{
+		{"GetNamespace", &GetNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}},
+		{"UpdateNamespace", &UpdateNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Properties: map[string]string{"k": "v"}}},
+		{"DeleteNamespace", deleteNamespaceRequest(t)},
+	}
+	for _, op := range ops {
+		t.Run(op.operation, func(t *testing.T) {
+			status, resp := runHandler(t, m, fs.Client, op.operation, op.input)
+			assert.Equal(t, http.StatusNotFound, status, "%+v", resp)
+			assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp)
+		})
+	}
+
+	assert.Equal(t, before, extendedOf(t, fs, GetTableBucketPath(renameTestBucket), "ns"), "the kept directory must not be written to")
+	assert.NotNil(t, fs.Get(keptDataDir(), keptDataFile))
+}
+
+// Directories with no file under them at any depth are not worth keeping, so a
+// namespace left with only those is removed whole, as before.
+func TestDeleteNamespaceRemovesEmptyLeftovers(t *testing.T) {
+	fs, m := startRenameManager(t)
+	renameTableToNs2(t, m, fs)
+
+	requireOK(t, m, fs, "DeleteNamespace", deleteNamespaceRequest(t))
+
+	assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns"), "a namespace holding no file is removed")
+	assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t"))
+	assert.Equal(t, []string{"ns2"}, listedNamespaces(t, m, fs))
+}
+
+// A file that sits deep under a leftover directory, past other empty branches,
+// is still found.
+func TestDeleteNamespaceKeepsDeeplyNestedData(t *testing.T) {
+	fs, m := startRenameManager(t)
+	renameTableToNs2(t, m, fs)
+	leftover := GetTablePath(renameTestBucket, "ns", "leftover")
+	fs.Put(GetNamespacePath(renameTestBucket, "ns"), "leftover", nil)
+	fs.Put(leftover, "empty", nil)
+	fs.Put(leftover, "year=2026", nil)
+	fs.Put(path.Join(leftover, "year=2026"), "month=10", nil)
+	fs.PutFile(path.Join(leftover, "year=2026", "month=10"), keptDataFile, time.Now())
+
+	requireOK(t, m, fs, "DeleteNamespace", deleteNamespaceRequest(t))
+
+	assert.NotNil(t, fs.Get(path.Join(leftover, "year=2026", "month=10"), keptDataFile))
+	assert.Equal(t, []string{"ns2"}, listedNamespaces(t, m, fs))
+}
+
+// A table's location may sit under a hidden directory of its namespace. Hidden
+// entries are never children, but the files under a hidden directory are data
+// the renamed table still reads, so they keep the namespace directory too.
+func TestDeleteNamespaceKeepsDataUnderAHiddenDirectory(t *testing.T) {
+	nsPath := GetNamespacePath(renameTestBucket, "ns")
+	hidden := path.Join(nsPath, ".hidden")
+	cases := []struct {
+		name string
+		seed func(fs *s3tablestest.MemFiler)
+		kept bool
+	}{
+		{"files under a hidden directory", func(fs *s3tablestest.MemFiler) {
+			fs.Put(nsPath, ".hidden", nil)
+			fs.Put(hidden, "data", nil)
+			fs.PutFile(path.Join(hidden, "data"), keptDataFile, time.Now())
+		}, true},
+		{"only empty hidden directories", func(fs *s3tablestest.MemFiler) {
+			fs.Put(nsPath, ".hidden", nil)
+			fs.Put(hidden, "data", nil)
+		}, false},
+		{"a hidden file", func(fs *s3tablestest.MemFiler) {
+			fs.PutFile(nsPath, ".DS_Store", time.Now())
+		}, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			fs, m := startRenameManager(t)
+			renameTableToNs2(t, m, fs)
+			tc.seed(fs)
+
+			requireOK(t, m, fs, "DeleteNamespace", deleteNamespaceRequest(t))
+
+			kept := fs.Get(GetTableBucketPath(renameTestBucket), "ns") != nil
+			assert.Equal(t, tc.kept, kept, "namespace directory kept")
+			if tc.kept {
+				assert.NotNil(t, fs.Get(path.Join(hidden, "data"), keptDataFile), "the hidden directory's data must survive")
+			}
+			assert.Equal(t, []string{"ns2"}, listedNamespaces(t, m, fs))
+		})
+	}
+}
+
+// The walk follows a directory's later pages. With more children than one page
+// holds, the only file sits past the first page — directly in the leftover, or
+// under a directory listed on the second page — and missing it would let the
+// delete take the data.
+func TestDeleteNamespaceKeepsDataPastTheFirstListingPage(t *testing.T) {
+	const children = dirListPageSize + 10
+	cases := []struct {
+		name string
+		put  func(fs *s3tablestest.MemFiler, leftover string)
+	}{
+		{"file on the second page", func(fs *s3tablestest.MemFiler, leftover string) {
+			fs.PutFile(leftover, "z-"+keptDataFile, time.Now())
+		}},
+		{"file under a directory on the second page", func(fs *s3tablestest.MemFiler, leftover string) {
+			fs.PutFile(path.Join(leftover, fmt.Sprintf("d%04d", children-1)), keptDataFile, time.Now())
+		}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			fs, m := startRenameManager(t)
+			renameTableToNs2(t, m, fs)
+			leftover := GetTablePath(renameTestBucket, "ns", "wide")
+			fs.Put(GetNamespacePath(renameTestBucket, "ns"), "wide", nil)
+			for i := 0; i < children; i++ {
+				fs.Put(leftover, fmt.Sprintf("d%04d", i), nil)
+			}
+			tc.put(fs, leftover)
+
+			requireOK(t, m, fs, "DeleteNamespace", deleteNamespaceRequest(t))
+
+			require.NotNil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns"), "the directory holding the data must stay")
+			assert.Equal(t, []string{"ns2"}, listedNamespaces(t, m, fs))
+		})
+	}
+}
+
+// listContextClient remembers the context of every listing it serves.
+type listContextClient struct {
+	filer_pb.SeaweedFilerClient
+	contexts []context.Context
+}
+
+func (c *listContextClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	c.contexts = append(c.contexts, ctx)
+	return c.SeaweedFilerClient.ListEntries(ctx, in, opts...)
+}
+
+// The walk stops at the first file with a page still unread. Its listings must
+// end with the call, not with the request, so the filer stops producing that page.
+func TestDirHoldsFilesEndsItsListings(t *testing.T) {
+	fs, m := startRenameManager(t)
+	leftover := GetTablePath(renameTestBucket, "ns", "leftover")
+	fs.Put(leftover, "a-dir", nil)
+	fs.PutFile(leftover, "b-file", time.Now())
+	fs.PutFile(leftover, "c-file", time.Now())
+	client := &listContextClient{SeaweedFilerClient: fs.Client}
+
+	holds, err := m.handler.dirHoldsFiles(context.Background(), client, leftover)
+
+	require.NoError(t, err)
+	assert.True(t, holds)
+	require.NotEmpty(t, client.contexts)
+	for _, ctx := range client.contexts {
+		assert.Error(t, ctx.Err(), "a listing must not outlive the walk")
+	}
+}
+
+// A table still in the namespace keeps it from being deleted, even with a
+// leftover directory holding data beside it.
+func TestDeleteNamespaceWithTableIsStillNotEmpty(t *testing.T) {
+	fs, m := startRenameManager(t)
+	fs.PutFile(keptDataDir(), keptDataFile, time.Now())
+	fs.Put(GetNamespacePath(renameTestBucket, "ns"), "leftover", nil)
+	fs.PutFile(GetTablePath(renameTestBucket, "ns", "leftover"), keptDataFile, time.Now())
+
+	status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t))
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeNamespaceNotEmpty, resp.Type, "%+v", resp)
+
+	namespace := fs.Get(GetTableBucketPath(renameTestBucket), "ns")
+	require.NotNil(t, namespace)
+	assert.Contains(t, namespace.Extended, ExtendedKeyMetadata, "a refused delete leaves the namespace a namespace")
+	assert.NotContains(t, namespace.Extended, ExtendedKeyDeleting, "a refused delete releases its claim")
+}
+
+// The name of a deleted namespace is free again even though its directory is
+// still there: the create takes the directory back, with no claim stuck on it,
+// and the data under it is untouched.
+func TestCreateNamespaceTakesBackAKeptDirectory(t *testing.T) {
+	fs, m := startDeletedNamespaceWithRenamedData(t)
+
+	requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns"},
+	})
+	requireOK(t, m, fs, "CreateTable", createTableRequest(t, "ns", "x"))
+
+	assert.ElementsMatch(t, []string{"ns", "ns2"}, listedNamespaces(t, m, fs))
+	assert.NotNil(t, fs.Get(keptDataDir(), keptDataFile), "the renamed table's data must survive")
+	status, resp := runHandler(t, m, fs.Client, "GetTable", &GetTableRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns2"},
+		Name:           "t",
+	})
+	assert.Equal(t, http.StatusOK, status, "%+v", resp)
+
+	// The directory is a real namespace again, so a second create is refused.
+	status, resp = runHandler(t, m, fs.Client, "CreateNamespace", &CreateNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns"},
+	})
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeNamespaceAlreadyExists, resp.Type, "%+v", resp)
+}
+
+// A file that replaces the leftover directory while the create is taking it
+// over must not end up tagged as a namespace: the create checks the type on the
+// entry it writes, and the filer refuses to write a directory over the file.
+// The name is the file's now, so the create answers a name conflict.
+func TestCreateNamespaceDoesNotTagAFileThatReplacesTheDirectory(t *testing.T) {
+	fs, m := startDeletedNamespaceWithRenamedData(t)
+	bucketPath := GetTableBucketPath(renameTestBucket)
+	// The create looks the name up twice: once to see whether it exists, and
+	// once to read the leftover it takes over. The file lands after the second.
+	client := &replaceAfterLookupClient{SeaweedFilerClient: fs.Client, f: fs, dir: bucketPath, name: "ns", nth: 2}
+
+	status, resp := runHandler(t, m, client, "CreateNamespace", &CreateNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"ns"},
+	})
+
+	require.GreaterOrEqual(t, client.lookups, 2, "the replacement has to land after the create's second read")
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeNamespaceAlreadyExists, resp.Type, "%+v", resp)
+	file := fs.Get(bucketPath, "ns")
+	require.NotNil(t, file)
+	assert.False(t, file.IsDirectory)
+	assert.NotContains(t, file.Extended, ExtendedKeyMetadata, "the file must not be tagged as a namespace")
+}
+
+// Removing the claimed namespace between the scan and the keep is a
+// concurrent change: the delete answers a conflict the caller can retry, not
+// an internal error.
+func TestDeleteNamespaceConflictsWhenTheKeptDirectoryVanishes(t *testing.T) {
+	fs, m := startRenameManager(t)
+	fs.PutFile(keptDataDir(), keptDataFile, time.Now())
+	renameTableToNs2(t, m, fs)
+
+	bucketPath := GetTableBucketPath(renameTestBucket)
+	// The delete claims the namespace before it scans, so the next lookup of
+	// the claimed directory is the keep's own read — remove it there.
+	fs.OnLookup = func(dir, name string) {
+		if dir != bucketPath || name != "ns" {
+			return
+		}
+		if entry := fs.Get(dir, name); entry == nil || entry.Extended[ExtendedKeyDeleting] == nil {
+			return
+		}
+		fs.OnLookup = nil
+		_, _ = fs.Client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{
+			Directory:    bucketPath,
+			Name:         "ns",
+			IsRecursive:  true,
+			IsDeleteData: true,
+		})
+	}
+
+	status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t))
+
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp)
+	assert.Nil(t, fs.Get(bucketPath, "ns"), "the competing removal keeps the directory it took")
+}
+
+// A namespace removed between the claim and the conditional delete is a
+// concurrent change too, not a delete this call completed.
+func TestDeleteNamespaceConflictsWhenItVanishesBeforeTheDelete(t *testing.T) {
+	fs, m := startRenameManager(t)
+	renameTableToNs2(t, m, fs)
+
+	bucketPath := GetTableBucketPath(renameTestBucket)
+	fs.BeforeDelete = func() {
+		_, _ = fs.Client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{
+			Directory:    bucketPath,
+			Name:         "ns",
+			IsRecursive:  true,
+			IsDeleteData: true,
+		})
+	}
+
+	status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", deleteNamespaceRequest(t))
+
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp)
+}
+
+// A file is not a leftover directory to take over: the create keeps refusing a
+// name a file already holds.
+func TestCreateNamespaceRefusesAFileInTheWay(t *testing.T) {
+	fs, m := startRenameManager(t)
+	fs.PutFile(GetTableBucketPath(renameTestBucket), "occupied", time.Now())
+
+	status, resp := runHandler(t, m, fs.Client, "CreateNamespace", &CreateNamespaceRequest{
+		TableBucketARN: mustBucketARN(t),
+		Namespace:      []string{"occupied"},
+	})
+	assert.Equal(t, http.StatusConflict, status, "%+v", resp)
+	assert.Equal(t, ErrCodeNamespaceAlreadyExists, resp.Type, "%+v", resp)
+	assert.Empty(t, fs.Get(GetTableBucketPath(renameTestBucket), "occupied").Extended, "the file must not be tagged as a namespace")
+}

weed/s3api/s3tables/s3tablestest/memfiler.go

@@ -171,6 +171,12 @@ func (f *MemFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryReque
 	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 will not write a directory entry over a file or the
+	// reverse; without it here a write that raced a replacement would turn the
+	// replacement into the entry it never was.
+	if existing.IsDirectory != req.Entry.IsDirectory {
+		return nil, status.Errorf(codes.Unknown, "%s: existing entry type differs", path.Join(req.Directory, req.Entry.Name))
+	}
 	// 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 {