diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 4434df6704..96fae0a259 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/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) { diff --git a/weed/s3api/s3tables/filer_ops_test.go b/weed/s3api/s3tables/filer_ops_test.go index 22f3e0c8f2..e29b81e196 100644 --- a/weed/s3api/s3tables/filer_ops_test.go +++ b/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) { diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index c54ffefe49..a4690f491a 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/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}) }) diff --git a/weed/s3api/s3tables/handler_namespace_keeps_data_test.go b/weed/s3api/s3tables/handler_namespace_keeps_data_test.go new file mode 100644 index 0000000000..e155cc87ce --- /dev/null +++ b/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") +} diff --git a/weed/s3api/s3tables/s3tablestest/memfiler.go b/weed/s3api/s3tables/s3tablestest/memfiler.go index 5fb4e309f3..391a2f0331 100644 --- a/weed/s3api/s3tables/s3tablestest/memfiler.go +++ b/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 {