diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 96fae0a259..774743dd69 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -11,8 +11,10 @@ import ( "path" "time" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -107,6 +109,20 @@ func (h *S3TablesHandler) mutateEntry( client filer_pb.SeaweedFilerClient, path string, mutate func(entry *filer_pb.Entry) error, +) error { + return h.mutateEntryAsserting(ctx, client, path, nil, mutate) +} + +// mutateEntryAsserting is mutateEntry with the keys in absent asserted at the +// write too, as this package's own attributes are: one absent when the entry was +// read and written by someone else since fails the compare, instead of being +// written over with the entry as it was read. +func (h *S3TablesHandler) mutateEntryAsserting( + ctx context.Context, + client filer_pb.SeaweedFilerClient, + path string, + absent []string, + mutate func(entry *filer_pb.Entry) error, ) error { dir, name := splitPath(path) @@ -120,7 +136,7 @@ func (h *S3TablesHandler) mutateEntry( } entry := resp.Entry - expected := snapshotExtended(entry.Extended) + expected := snapshotExtended(entry.Extended, absent...) if entry.Extended == nil { entry.Extended = make(map[string][]byte) } @@ -401,13 +417,25 @@ func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer // 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. +// the data under it leaves one — so its absence reads as a missing entry. So +// does a namespace still marked by its create: like a table's, it is absent to +// every reader until the create confirms it. 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) { + entry, err := h.lookupEntry(ctx, client, namespacePath) + if err != nil { + return nil, err + } + data, ok := entry.Extended[ExtendedKeyMetadata] + if !ok || isUnfinishedCreate(entry.Extended) { return nil, fmt.Errorf("%w: %s", filer_pb.ErrNotFound, namespacePath) } - return data, err + return data, nil +} + +// isFinishedNamespaceEntry is isNamespaceEntry for a namespace its create has +// confirmed, the only kind a reader or a delete may act on. +func isFinishedNamespaceEntry(entry *filer_pb.Entry) bool { + return isNamespaceEntry(entry) && !isUnfinishedCreate(entry.Extended) } // loadNamespaceMetadata resolves a namespace to its metadata. A directory that @@ -499,9 +527,46 @@ func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client f return nil } +// errDirNotEmpty is a directory that holds something the caller meant to leave +// alone: a file in a leftover tree, or an entry the filer found in a directory +// it was asked to delete without its contents. +var errDirNotEmpty = errors.New("directory is not empty") + +// deleteEmptyDirectoryIfMatches deletes a directory without its contents, only +// while it still carries the extended attributes the caller captured. The +// filer checks both under the path lock, in the one delete: a changed +// attribute is ErrConcurrentUpdate, and an entry that appeared in the +// directory is errDirNotEmpty, so neither a directory taken back since the +// caller looked at it nor a file written into it since is lost. The mark of an +// empty S3 object stored on the directory is asserted absent too, so one +// written since is kept the same way. +func (h *S3TablesHandler) deleteEmptyDirectoryIfMatches(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, expected *filer_pb.Entry) error { + dir, name := splitPath(path) + resp, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{ + Directory: dir, + Name: name, + IsDeleteData: true, + ExpectedExtended: snapshotExtended(expected.Extended, s3_constants.SeaweedFSPrefixObject), + }) + if status.Code(err) == codes.FailedPrecondition { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path) + } + if err != nil { + return err + } + if resp != nil && resp.Error != "" { + if err := filer.DeleteEntryError(resp.Error); !errors.Is(err, filer.ErrNonEmptyFolder) { + return err + } + return fmt.Errorf("%w: %s", errDirNotEmpty, path) + } + return nil +} + var ( errParentDeleting = errors.New("parent is being deleted") errParentGone = errors.New("parent was deleted") + errParentCreating = errors.New("parent is still being created") errNoDeleteClaim = errors.New("claim not held") errBackOutFailed = errors.New("backing out the create failed") errSourceGone = errors.New("rename source was deleted") @@ -527,18 +592,54 @@ func (h *S3TablesHandler) claimForDelete(ctx context.Context, client filer_pb.Se 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 { +// tryReleaseDeleteClaim removes claim from path once, while path still holds +// it. A claim another delete replaced or lifted, or an entry that is gone, +// leaves nothing to release and is not an error. +func (h *S3TablesHandler) tryReleaseDeleteClaim(ctx context.Context, client filer_pb.SeaweedFilerClient, path, claim string) error { + err := h.mutateEntryExtended(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) + if errors.Is(err, errNoDeleteClaim) || errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + return err +} + +// releaseDeleteClaim removes claim from path while path still holds it, even +// after the request was cancelled. A claim left behind refuses creates under a +// bucket or namespace and makes a leftover directory read as taken, so a failed +// release is retried, including one that lost to another write of the entry. +func (h *S3TablesHandler) releaseDeleteClaim(ctx context.Context, client filer_pb.SeaweedFilerClient, path, claim string) { + ctx = context.WithoutCancel(ctx) + var err error + for attempt := 0; attempt < extendedAttributeUpdateAttempts; attempt++ { + if attempt > 0 { + time.Sleep(100 * time.Millisecond) + } + if err = h.tryReleaseDeleteClaim(ctx, client, path, claim); err == nil { + return + } + } + glog.Warningf("release delete claim on %s: %v", path, err) +} + +// isTableBucketCreatedAt accepts only the table bucket created at createdAt. A +// bucket deleted and created again under the same name has a CreatedAt of its +// own, so a create authorized against the first reads the second as gone. +func isTableBucketCreatedAt(createdAt time.Time) func(*filer_pb.Entry) bool { + return func(entry *filer_pb.Entry) bool { + if !IsTableBucketEntry(entry) { + return false + } + var metadata tableBucketMetadata + if json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata) != nil { + return false + } + return metadata.CreatedAt.Equal(createdAt) } } @@ -549,7 +650,8 @@ func isNamespaceEntry(entry *filer_pb.Entry) bool { } // checkParentOpen returns errParentGone if parentPath is missing or not a -// catalog entry, and errParentDeleting if a delete has claimed it. +// catalog entry, errParentCreating if the create that wrote it has not +// confirmed it yet, 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 @@ -568,6 +670,9 @@ func (h *S3TablesHandler) openParent(ctx context.Context, client filer_pb.Seawee if !isParent(entry) { return nil, errParentGone } + if isUnfinishedCreate(entry.Extended) { + return nil, errParentCreating + } if _, claimed := entry.Extended[ExtendedKeyDeleting]; claimed { return nil, errParentDeleting } @@ -644,6 +749,20 @@ func (h *S3TablesHandler) mutateCatalogRecord( // 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 { + return h.confirmChild(ctx, client, parentPath, isParent, true, childPath, written, emptyDirs) +} + +// confirmParentOnce is confirmParent for a record that other creates build on, +// a namespace. Checking the parent again after the mark comes off would leave a +// window in which a child is created in the record and the check then removes +// the record under it, so the parent is checked once, before the mark comes +// off. That is enough while a delete of the parent counts the marked record: one +// claiming the parent after the check scans the record and is refused. +func (h *S3TablesHandler) confirmParentOnce(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, childPath string, written map[string][]byte, emptyDirs ...string) error { + return h.confirmChild(ctx, client, parentPath, isParent, false, childPath, written, emptyDirs) +} + +func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, recheck bool, childPath string, written map[string][]byte, emptyDirs []string) error { parent, err := h.openParent(ctx, client, parentPath, isParent) confirmed := false if err == nil { @@ -651,7 +770,9 @@ func (h *S3TablesHandler) confirmParent(ctx context.Context, client filer_pb.Sea // 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 recheck { + err = h.parentStillOpen(ctx, client, parentPath, isParent, parent) + } } } if err == nil { @@ -753,6 +874,159 @@ func (h *S3TablesHandler) dirHoldsFiles(ctx context.Context, client filer_pb.Sea return false, nil } +// claimTreeDirectory writes a fresh delete claim onto the directory at dirPath +// while it still carries expected, and returns the claim with the attributes it +// left. A claim seen in expected, one a delete that died left, is replaced like +// any other attribute; one written since fails the compare. +// +// The claim is what ties the later delete to this directory: the delete +// requires it, and only this directory carries it. A file or an S3 object at +// dirPath is refused with errDirNotEmpty, whether it was there when the entry +// was read or took the directory's place before the write, which the filer +// refuses since it never writes a directory entry over a file. One that takes +// the place after the write lacks the claim, so the delete is refused instead. +// An S3 object landing on the directory between the read and the write carries +// the prefix-object mark, which the write asserts absent, so the write fails +// instead of putting the bare directory back over the object. +func (h *S3TablesHandler) claimTreeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, expected map[string][]byte) (string, map[string][]byte, error) { + claim := generateVersionToken() + var claimed map[string][]byte + err := h.mutateEntryAsserting(ctx, client, dirPath, []string{s3_constants.SeaweedFSPrefixObject}, func(entry *filer_pb.Entry) error { + if !entry.IsDirectory || entry.IsDirectoryKeyObject() { + return fmt.Errorf("%w: %s is an object", errDirNotEmpty, dirPath) + } + for key, want := range expected { + if !bytes.Equal(entry.Extended[key], want) { + return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, dirPath) + } + } + entry.Extended[ExtendedKeyDeleting] = []byte(claim) + claimed = snapshotExtended(entry.Extended) + return nil + }) + if errors.Is(err, filer_pb.ErrNotFound) { + return "", nil, fmt.Errorf("%w: %s", ErrConcurrentUpdate, dirPath) + } + if err != nil && !errors.Is(err, errDirNotEmpty) && !errors.Is(err, ErrConcurrentUpdate) { + // The filer reports the refused write over a file only as an opaque + // failure, so look at what holds the path now. + if current, lookupErr := h.lookupEntry(ctx, client, dirPath); lookupErr == nil && !current.IsDirectory { + return "", nil, fmt.Errorf("%w: %s is an object", errDirNotEmpty, dirPath) + } + } + if err != nil { + return "", nil, err + } + return claim, claimed, nil +} + +// removeTreeDirectory removes a directory the walk listed as listed, under a +// claim of its own, and lifts the claim if the delete is refused. +func (h *S3TablesHandler) removeTreeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, listed *filer_pb.Entry) error { + claim, claimed, err := h.claimTreeDirectory(ctx, client, dirPath, snapshotExtended(listed.Extended)) + if err != nil { + return err + } + if err := h.deleteEmptyDirectoryIfMatches(ctx, client, dirPath, &filer_pb.Entry{Extended: claimed}); err != nil { + h.releaseDeleteClaim(ctx, client, dirPath, claim) + return err + } + return nil +} + +// removeEmptyTree removes dirPath and every directory under it, provided no +// file, no S3 object stored on a directory entry, and no catalog record sits +// anywhere in the tree, dirPath included. dirEntry is dirPath as the caller saw +// it. +// +// dirPath is claimed first, in a write that requires the attributes the caller +// saw: a directory taken back as a namespace since then fails it, before +// anything under it is touched, and the claim keeps it from being taken back +// while the tree goes. With dirPath no namespace, nothing under it can become a +// catalog record either, so a record the walk meets was there before and is +// refused, not removed. +// +// The walk lists the tree a page at a time and stops at the first file, object +// or record with errDirNotEmpty, before anything is removed. Directories then go +// deepest first, each claimed with the attributes it had when listed and +// deleted under that claim, which refuses a directory that holds an entry by +// now or was replaced by a file. dirPath goes last, still under its claim; any +// other way out lifts the claim. +func (h *S3TablesHandler) removeEmptyTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, dirEntry *filer_pb.Entry) (err error) { + claim, claimed, err := h.claimTreeDirectory(ctx, client, dirPath, snapshotExtended(dirEntry.Extended)) + if err != nil { + return err + } + defer func() { + if err != nil { + h.releaseDeleteClaim(ctx, client, dirPath, claim) + } + }() + + type listedDir struct { + path string + entry *filer_pb.Entry + } + listed := []listedDir{{dirPath, &filer_pb.Entry{Extended: claimed}}} + + walkCtx, cancel := context.WithCancel(ctx) + defer cancel() + // A directory is listed after its parent, so removing in reverse order + // removes every directory after the ones under it. + for next := 0; next < len(listed); next++ { + dir := listed[next].path + lastFileName := "" + for { + resp, err := client.ListEntries(walkCtx, &filer_pb.ListEntriesRequest{ + Directory: dir, + Limit: dirListPageSize, + StartFromFileName: lastFileName, + InclusiveStartFrom: lastFileName == "", + }) + if err != nil { + if status.Code(err) == codes.NotFound { + break // gone, so its delete below reports the change + } + return err + } + + received := false + for { + entry, err := resp.Recv() + if err != nil { + if err == io.EOF || status.Code(err) == codes.NotFound { + break + } + return err + } + if entry.Entry == nil { + continue + } + received = true + lastFileName = entry.Entry.Name + child := path.Join(dir, entry.Entry.Name) + if !entry.Entry.IsDirectory || entry.Entry.IsDirectoryKeyObject() { + return fmt.Errorf("%w: %s is an object", errDirNotEmpty, child) + } + if _, isRecord := entry.Entry.Extended[ExtendedKeyMetadata]; isRecord { + return fmt.Errorf("%w: %s is a catalog record", errDirNotEmpty, child) + } + listed = append(listed, listedDir{child, entry.Entry}) + } + if !received { + break + } + } + } + + for i := len(listed) - 1; i > 0; i-- { + if err := h.removeTreeDirectory(ctx, client, listed[i].path, listed[i].entry); err != nil { + return err + } + } + return h.deleteEmptyDirectoryIfMatches(ctx, client, dirPath, &filer_pb.Entry{Extended: claimed}) +} + // 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 diff --git a/weed/s3api/s3tables/handler_bucket_get_list_delete.go b/weed/s3api/s3tables/handler_bucket_get_list_delete.go index 1b801eb84a..2e7c604988 100644 --- a/weed/s3api/s3tables/handler_bucket_get_list_delete.go +++ b/weed/s3api/s3tables/handler_bucket_get_list_delete.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http" + "path" "strings" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -275,6 +276,32 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http var metadata tableBucketMetadata var bucketPolicy string hasChildren := false + heldDir := "" + unfinishedNamespace := "" + // What the scan found besides namespaces: the bare directories — a deleted + // namespace's leftovers or hidden ones — as they were seen, and the names + // of the hidden files. + var leftovers []*filer_pb.Entry + var hidden []string + // The claim this delete holds on the bucket. It goes with the bucket when it + // is deleted; any other way out lifts it, or the bucket would refuse every + // create. bucketClaimed is the bucket's record as the claim left it, which the + // removal has to find. + bucketClaim := "" + var bucketClaimed map[string][]byte + defer func() { + if bucketClaim == "" { + return + } + // A claim left behind refuses every create in the bucket until the next + // delete replaces it, so a failed release is retried. + if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + h.releaseDeleteClaim(r.Context(), client, bucketPath, bucketClaim) + return nil + }); err != nil { + glog.Warningf("release delete claim on %s: %v", bucketPath, err) + } + }() err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // 1. Get metadata for ownership check data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata) @@ -310,25 +337,89 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http return NewAuthError("DeleteTableBucket", principal, fmt.Sprintf("not authorized to delete bucket %s", bucketName)) } - // 3. Check if bucket is empty - resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{ - Directory: bucketPath, - Limit: 10, - }) + // 2. Claim the bucket before scanning it. CreateNamespace refuses a + // claimed bucket and checks it again after its write, so a namespace is + // either seen by the scan below or backed out. The bucket is removed only + // while it still carries this claim: once another delete has replaced or + // lifted it, a create may have passed that check, and this delete answers + // 409 instead. A claim an earlier delete left behind is replaced. Only the + // bucket authorized above is claimed: one deleted and created again under + // the same name since then is another bucket and reads as gone. + bucketClaim, bucketClaimed, err = h.claimForDelete(r.Context(), client, bucketPath, isTableBucketCreatedAt(metadata.CreatedAt)) if err != nil { return err } - for { - entry, err := resp.Recv() + // 3. Check if bucket is empty. A file, an S3 object stored on a directory + // entry, or a namespace is a child. A bare directory is what a deleted + // namespace leaves behind to keep the data under it, so it is only a + // child while it still holds a file. The scan + // pages through the whole bucket: the first page may be nothing but bare + // directories sorted ahead of the namespaces. + lastFileName := "" + for !hasChildren { + resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{ + Directory: bucketPath, + 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, ".") { - hasChildren = true + + 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 + // Hidden entries are never children. A hidden directory can + // hold files the same way a leftover does, so it waits with + // them for the walk; a hidden file is removed by name. + if strings.HasPrefix(entry.Entry.Name, ".") { + if entry.Entry.IsDirectory { + leftovers = append(leftovers, entry.Entry) + } else { + hidden = append(hidden, entry.Entry.Name) + } + continue + } + if _, hasMetadata := entry.Entry.Extended[ExtendedKeyMetadata]; hasMetadata || !entry.Entry.IsDirectory || entry.Entry.IsDirectoryKeyObject() { + hasChildren = true + if hasMetadata && isUnfinishedCreate(entry.Entry.Extended) { + unfinishedNamespace = entry.Entry.Name + } + break + } + leftovers = append(leftovers, entry.Entry) + } + if !hasMore { + break + } + } + + // Looked at after the scan, not during it, so a bucket that still has a + // namespace answers at once instead of walking the leftovers first. The + // first directory holding a file settles it. + if hasChildren { + return nil + } + for _, leftover := range leftovers { + holdsFiles, err := h.dirHoldsFiles(r.Context(), client, path.Join(bucketPath, leftover.Name)) + if err != nil { + return err + } + if holdsFiles { + heldDir = leftover.Name break } } @@ -341,6 +432,8 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) } else if isAuthError(err) { h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + } else if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table bucket changed during delete, retry the request") } else { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to delete table bucket: %v", err)) } @@ -348,25 +441,59 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http } if hasChildren { - h.writeError(w, http.StatusConflict, ErrCodeBucketNotEmpty, "table bucket is not empty") + message := "table bucket is not empty" + if unfinishedNamespace != "" { + // Listings hide it, so name it: creating it again finishes it. + message = fmt.Sprintf("table bucket is not empty: namespace %s has an unfinished create", unfinishedNamespace) + } + h.writeError(w, http.StatusConflict, ErrCodeBucketNotEmpty, message) return fmt.Errorf("bucket not empty") } + // Nothing in the bucket is a namespace, but data sits under a directory a + // deleted namespace left. Deleting the bucket would take it, so it stays. + if heldDir != "" { + h.writeError(w, http.StatusConflict, ErrCodeBucketNotEmpty, fmt.Sprintf("table bucket is not empty: directory %s still holds files", heldDir)) + return fmt.Errorf("bucket not empty: directory %s still holds files", heldDir) + } + // Delete the bucket err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Drop the leaf entry first, so it does not outlive the directory. if tableObjErr := h.deleteEntryIfExists(r.Context(), client, GetTableObjectBucketPath(bucketName)); tableObjErr != nil { glog.V(1).Infof("failed to delete table object for %s: %v", bucketName, tableObjErr) } + // Only the hidden files go by a recursive delete. A file written since + // the scan anywhere else makes the filer refuse the delete that would have + // taken it: each leftover goes only if it is still the bare directory the + // scan saw and holds no file or catalog record, and the bucket only if + // nothing is left in it. + for _, leftover := range leftovers { + if err := h.removeEmptyTree(r.Context(), client, path.Join(bucketPath, leftover.Name), leftover); err != nil { + return err + } + } + for _, name := range hidden { + if err := h.deleteDirectory(r.Context(), client, path.Join(bucketPath, name)); err != nil { + return err + } + } // The bucket is the directory, so a refused delete leaves it in place // and the caller must hear about it. - return h.deleteDirectory(r.Context(), client, bucketPath) + return h.deleteEmptyDirectoryIfMatches(r.Context(), client, bucketPath, &filer_pb.Entry{Extended: bucketClaimed}) }) if err != nil { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table bucket") + if errors.Is(err, errDirNotEmpty) { + h.writeError(w, http.StatusConflict, ErrCodeBucketNotEmpty, "table bucket is not empty") + } else if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table bucket changed during delete, retry the request") + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table bucket") + } return err } + bucketClaim = "" h.writeJSON(w, http.StatusOK, nil) return nil diff --git a/weed/s3api/s3tables/handler_bucket_keeps_data_test.go b/weed/s3api/s3tables/handler_bucket_keeps_data_test.go new file mode 100644 index 0000000000..dfeef51285 --- /dev/null +++ b/weed/s3api/s3tables/handler_bucket_keeps_data_test.go @@ -0,0 +1,997 @@ +package s3tables + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "path" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// startEmptyBucket seeds a table bucket with nothing in it, not even a namespace. +func startEmptyBucket(t *testing.T) (*s3tablestest.MemFiler, *Manager) { + t.Helper() + fs := s3tablestest.Start(t) + bucketMeta, err := json.Marshal(tableBucketMetadata{Name: renameTestBucket, OwnerAccountID: DefaultAccountID}) + require.NoError(t, err) + fs.Put(TablesPath, renameTestBucket, map[string][]byte{ + ExtendedKeyTableBucket: []byte("{}"), + ExtendedKeyMetadata: bucketMeta, + }) + m := NewManager() + m.SetTrusted(true) + return fs, m +} + +func deleteTableBucketRequest(t *testing.T) *DeleteTableBucketRequest { + return &DeleteTableBucketRequest{TableBucketARN: mustBucketARN(t)} +} + +// Deleting a namespace that holds renamed-away data leaves its directory behind +// with no record. Once the renamed table is dropped the data is gone, and the +// directory is an empty leftover that must not keep the bucket from being deleted. +func TestDeleteTableBucketRemovesEmptyLeftovers(t *testing.T) { + fs, m := startDeletedNamespaceWithRenamedData(t) + bucketPath := GetTableBucketPath(renameTestBucket) + require.NotNil(t, fs.Get(bucketPath, "ns"), "the delete of ns keeps its directory, which is the leftover under test") + + requireOK(t, m, fs, "DeleteTable", &DeleteTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + Name: "t", + }) + requireOK(t, m, fs, "DeleteNamespace", &DeleteNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + }) + require.Nil(t, fs.Get(keptDataDir(), keptDataFile), "dropping the renamed table purges the data the leftover kept") + require.NotNil(t, fs.Get(bucketPath, "ns"), "the leftover directory is still in the bucket") + + requireOK(t, m, fs, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") + assert.Nil(t, fs.Get(bucketPath, "ns"), "the empty leftover goes with it") +} + +// No table refers to a file under a directory with no record, but deleting the +// bucket would destroy it without anyone deciding to. The bucket stays, and so +// does the file. +func TestDeleteTableBucketRefusesOrphanData(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + orphan := path.Join(bucketPath, "old", "t", "data") + fs.Put(bucketPath, "old", nil) + fs.Put(path.Join(bucketPath, "old"), "t", nil) + fs.Put(path.Join(bucketPath, "old", "t"), "data", nil) + fs.PutFile(orphan, keptDataFile, time.Now()) + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + assert.Contains(t, resp.Message, "directory old still holds files", "the answer names the directory that holds the file") + assert.NotNil(t, fs.Get(orphan, keptDataFile), "the file must survive") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") +} + +// A hidden directory can hold data the same way a visible leftover does. +// Files under it refuse the delete; only a tree holding none goes with the +// bucket, and so does a hidden file. +func TestDeleteTableBucketKeepsDataUnderAHiddenDirectory(t *testing.T) { + cases := []struct { + name string + seed func(t *testing.T, fs *s3tablestest.MemFiler, bucketPath, hiddenPath string) + kept bool + }{ + {"files under a hidden directory", func(t *testing.T, fs *s3tablestest.MemFiler, bucketPath, hiddenPath string) { + fs.Put(bucketPath, ".hidden", nil) + fs.Put(hiddenPath, "data", nil) + fs.PutFile(path.Join(hiddenPath, "data"), keptDataFile, time.Now()) + }, true}, + {"only empty hidden directories", func(t *testing.T, fs *s3tablestest.MemFiler, bucketPath, hiddenPath string) { + fs.Put(bucketPath, ".hidden", nil) + fs.Put(hiddenPath, "data", nil) + }, false}, + {"a hidden file", func(t *testing.T, fs *s3tablestest.MemFiler, bucketPath, hiddenPath string) { + fs.PutFile(bucketPath, ".DS_Store", time.Now()) + }, false}, + {"a hidden directory stored as an object", func(t *testing.T, fs *s3tablestest.MemFiler, bucketPath, hiddenPath string) { + putDirectoryObject(t, fs, bucketPath, ".hidden") + }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + tc.seed(t, fs, bucketPath, path.Join(bucketPath, ".hidden")) + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + if tc.kept { + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + assert.NotNil(t, fs.Get(bucketPath, ".hidden"), "the hidden entry must survive") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") + } else { + assert.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") + } + }) + } +} + +// pauseListClient runs pause just before the nth listing of dir, so a test can +// land a competing write at a chosen step of a delete's directory walk. +type pauseListClient struct { + filer_pb.SeaweedFilerClient + dir string + nth int + listings int + pause func() +} + +func (c *pauseListClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + if in.Directory == c.dir { + c.listings++ + if c.listings == c.nth { + c.pause() + } + } + return c.SeaweedFilerClient.ListEntries(ctx, in, opts...) +} + +func createNamespaceNsRequest(t *testing.T) *CreateNamespaceRequest { + return &CreateNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}} +} + +// A create could take a leftover directory back after the scan listed it, and a +// namespace with nothing in it holds no file for the walk to find, so the +// recursive delete would remove a namespace the create had already answered +// for. The delete's claim on the bucket turns the create away instead. +func TestDeleteTableBucketTurnsAwayATakeBackAfterTheScan(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + fs.Put(bucketPath, "ns", nil) + var createStatus int + var createResp S3TablesError + // The scan lists the bucket again for the page after the leftover, so the + // create lands once the scan has recorded it and before the delete walks it. + client := &pauseListClient{SeaweedFilerClient: fs.Client, dir: bucketPath, nth: 2} + client.pause = func() { + createStatus, createResp = runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + } + + status, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.GreaterOrEqual(t, client.listings, 2, "the create has to land after the scan listed the leftover") + assert.Equal(t, http.StatusConflict, createStatus, "%+v", createResp) + assert.Equal(t, ErrCodeConflict, createResp.Type, "%+v", createResp) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") +} + +// A create that finds the bucket claimed by a delete is refused and writes +// nothing, as a create under a claimed namespace is. +func TestCreateNamespaceRefusedInClaimedBucket(t *testing.T) { + fs, m := startEmptyBucket(t) + plantDeleteClaim(fs, TablesPath, renameTestBucket) + + status, resp := runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(GetTableBucketPath(renameTestBucket), "ns"), "a refused create writes nothing") +} + +// A create is authorized against the bucket it read. If that bucket is deleted +// and another is created under the same name before the create writes, the +// namespace would land in a bucket nobody authorized it for, carrying the first +// bucket's owner. The create checks the bucket by its creation time, reads the +// new one as the old one gone, and writes nothing into it. +func TestCreateNamespaceTreatsARecreatedBucketAsGone(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + recreatedMeta, err := json.Marshal(tableBucketMetadata{ + Name: renameTestBucket, + OwnerAccountID: "111122223333", + CreatedAt: time.Now(), + }) + require.NoError(t, err) + recreated := map[string][]byte{ + ExtendedKeyTableBucket: []byte("{}"), + ExtendedKeyMetadata: recreatedMeta, + } + // The create looks its namespace up right before it writes; the bucket is + // replaced there. + fs.OnLookup = func(dir, name string) { + if dir != bucketPath || name != "ns" { + return + } + fs.OnLookup = nil + fs.Put(TablesPath, renameTestBucket, recreated) + } + + status, resp := runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + + assert.Equal(t, http.StatusNotFound, status, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchBucket, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(bucketPath, "ns"), "nothing is written into the new bucket") + require.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the new bucket stays") + assert.Equal(t, recreated, fs.Get(TablesPath, renameTestBucket).Extended, "and is left as it was created") +} + +// A delete is authorized against the bucket it read. If that bucket is deleted +// and another is created under the same name before the delete claims it, the +// claim and the removal would act on a bucket nobody authorized the delete for. +// The claim takes only the bucket with the creation time the delete read, so +// the new one reads as the old one gone and is left as it was. +func TestDeleteTableBucketTreatsARecreatedBucketAsGone(t *testing.T) { + fs, m := startEmptyBucket(t) + recreatedMeta, err := json.Marshal(tableBucketMetadata{ + Name: renameTestBucket, + OwnerAccountID: "111122223333", + CreatedAt: time.Now(), + }) + require.NoError(t, err) + recreated := map[string][]byte{ + ExtendedKeyTableBucket: []byte("{}"), + ExtendedKeyMetadata: recreatedMeta, + } + // The claim is the delete's first write. The bucket is replaced as it lands, + // so the write loses and its retry reads the new bucket. + fs.BeforeUpdate = func() { fs.Put(TablesPath, renameTestBucket, recreated) } + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusNotFound, code, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchBucket, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the new bucket stays") + assert.Equal(t, recreated, fs.Get(TablesPath, renameTestBucket).Extended, "and is left as it was created") +} + +// failNthLookupClient answers the nth lookup of target with Unavailable, the +// way a filer that is briefly unreachable would, and counts the lookups of the +// entries under target that follow. +type failNthLookupClient struct { + filer_pb.SeaweedFilerClient + target string + nth int + lookups int + childLookups int +} + +func (c *failNthLookupClient) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + switch path.Join(req.Directory, req.Name) { + case c.target: + c.lookups++ + if c.lookups == c.nth { + return nil, status.Error(codes.Unavailable, "filer unavailable") + } + default: + if req.Directory == c.target { + c.childLookups++ + } + } + return c.SeaweedFilerClient.LookupDirectoryEntry(ctx, req, opts...) +} + +// The check of the bucket before the namespace is written either passes or ends +// the create. A bucket it could not read, or one it finds still being created, +// is not one to write into: nothing is written, where a create that went on +// would write and leave its confirmation to back out. +func TestCreateNamespaceStopsAtAFailedBucketCheck(t *testing.T) { + bucketPath := GetTableBucketPath(renameTestBucket) + + t.Run("the bucket could not be read", func(t *testing.T) { + fs, m := startEmptyBucket(t) + // The create reads the bucket three times for its authorization; the + // fourth read is the check before the write. + client := &failNthLookupClient{SeaweedFilerClient: fs.Client, target: bucketPath, nth: 4} + + code, resp := runHandler(t, m, client, "CreateNamespace", createNamespaceNsRequest(t)) + + assert.Equal(t, http.StatusInternalServerError, code, "%+v", resp) + assert.Equal(t, ErrCodeInternalError, resp.Type, "%+v", resp) + assert.Zero(t, client.childLookups, "the create stops before it looks at the namespace") + assert.Nil(t, fs.Get(bucketPath, "ns"), "nothing is written") + }) + t.Run("the bucket is still being created", func(t *testing.T) { + fs, m := startEmptyBucket(t) + plantCreateMark(fs, TablesPath, renameTestBucket) + + code, resp := runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(bucketPath, "ns"), "nothing is written") + }) +} + +// A delete that has scanned an empty bucket must not lose to a create that +// lands before the bucket is removed: the claim the delete holds since before +// the scan turns the create away, so no namespace is answered for and deleted. +func TestDeleteTableBucketTurnsAwayACreateAfterTheScan(t *testing.T) { + fs, m := startEmptyBucket(t) + var createStatus int + var createResp S3TablesError + fs.BeforeDelete = func() { + createStatus, createResp = runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.NotZero(t, createStatus, "the create has to land between the scan and the delete") + assert.Equal(t, http.StatusConflict, createStatus, "%+v", createResp) + assert.Equal(t, ErrCodeConflict, createResp.Type, "%+v", createResp) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") +} + +// A create can pass the claim check before the delete claims the bucket, and +// then write a namespace that was not there when the delete scanned. The scan +// never saw it, and the recursive delete would remove it after the create had +// answered. The create's confirmation after its write finds the claim, so it +// backs out and reports the conflict instead of success. +func TestCreateNamespaceWrittenAfterTheScanBacksOut(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + reached := make(chan struct{}) + release := make(chan struct{}) + // The create looks its namespace up right after the claim check; hold it there. + fs.OnLookup = func(dir, name string) { + if dir != bucketPath || name != "ns" { + return + } + fs.OnLookup = nil + close(reached) + <-release + } + created := startHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + waitForPause(t, reached, created) + // The delete claims the bucket and scans it empty. The create writes only + // now, before the delete removes the bucket. + var createResult handlerResult + fs.BeforeDelete = func() { + close(release) + createResult = <-created + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.NotZero(t, createResult.status, "the create has to write between the scan and the delete") + assert.Equal(t, http.StatusConflict, createResult.status, "%+v", createResult.resp) + assert.Equal(t, ErrCodeConflict, createResult.resp.Type, "%+v", createResult.resp) + assert.Equal(t, http.StatusOK, status, "%+v", resp) + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") + assert.Nil(t, fs.Get(bucketPath, "ns"), "no namespace the create answered for is left behind") +} + +// The claim a delete takes on the bucket can be replaced by a second delete, +// which lifts it when it refuses. A create then finds the bucket open, and the +// first delete, which scanned before any of that, must not remove the bucket +// on a claim it no longer holds: it would take the namespace the create +// answered for, and the file under it, with it. It answers 409 and leaves both. +func TestDeleteTableBucketRefusesAfterItsClaimIsLifted(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + nsPath := path.Join(bucketPath, "ns") + var refusedStatus, createStatus int + var refusedResp, createResp S3TablesError + // The first delete has scanned an empty bucket when it reaches its removal. + fs.BeforeDelete = func() { + // A directory with a file, kept by a namespace that was deleted, turns up + // and the second delete refuses on it, lifting its own claim. + fs.Put(bucketPath, "ns", nil) + fs.PutFile(nsPath, keptDataFile, time.Now()) + refusedStatus, refusedResp = runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + // The bucket is open again, so the create takes the directory back. + createStatus, createResp = runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.NotZero(t, createStatus, "the create has to land between the first delete's scan and its removal") + assert.Equal(t, http.StatusConflict, refusedStatus, "%+v", refusedResp) + assert.Equal(t, ErrCodeBucketNotEmpty, refusedResp.Type, "%+v", refusedResp) + assert.Equal(t, http.StatusOK, createStatus, "%+v", createResp) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") + assert.NotContains(t, fs.Get(TablesPath, renameTestBucket).Extended, ExtendedKeyDeleting) + assert.NotNil(t, fs.Get(bucketPath, "ns"), "the namespace the create answered for must survive") + assert.NotNil(t, fs.Get(nsPath, keptDataFile), "the file under it must survive") +} + +// The walk that finds a leftover directory empty is over before the bucket is +// removed, and a file can be written in between: an upload through the S3 API +// takes no claim. Nothing the removal does may take a file it was not shown, so +// it is refused at the directory that holds one, and a file written into the +// bucket itself makes the removal of the bucket refuse the same way. +func TestDeleteTableBucketKeepsAFileUploadedAfterTheWalk(t *testing.T) { + for _, tc := range []struct { + name string + dir func(bucketPath string) string + }{ + {"into a leftover directory", func(bucketPath string) string { return path.Join(bucketPath, "old") }}, + {"into the bucket", func(bucketPath string) string { return bucketPath }}, + } { + t.Run(tc.name, func(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + fs.Put(bucketPath, "old", nil) + uploadDir := tc.dir(bucketPath) + fs.BeforeDelete = func() { + fs.PutFile(uploadDir, keptDataFile, time.Now()) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + assert.NotNil(t, fs.Get(uploadDir, keptDataFile), "the uploaded file must survive") + require.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") + assert.NotContains(t, fs.Get(TablesPath, renameTestBucket).Extended, ExtendedKeyDeleting, "a refused delete lifts its claim") + if uploadDir != bucketPath { + require.NotNil(t, fs.Get(bucketPath, "old")) + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "and its claim on the leftover") + } + }) + } +} + +// A second delete can replace the first one's claim and lift it, and a create +// then takes a leftover directory back as a namespace after the first delete +// scanned it bare, and an empty table is created in it. Neither holds a file, +// so only the attributes the scan saw on the directory tell the removal it is +// no longer the same directory, and that has to be settled before anything +// under it goes: the delete refuses with a conflict and leaves both. +func TestDeleteTableBucketKeepsALeftoverTakenBackAfterItsClaimIsLifted(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + nsPath := path.Join(bucketPath, "ns") + fs.Put(bucketPath, "ns", nil) + var refusedStatus, createStatus, tableStatus int + var refusedResp, createResp, tableResp S3TablesError + fs.BeforeDelete = func() { + // A file under another leftover makes the second delete refuse, and + // lift its claim; the bucket is open again for the create. + fs.Put(bucketPath, "old", nil) + fs.PutFile(path.Join(bucketPath, "old"), keptDataFile, time.Now()) + refusedStatus, refusedResp = runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + createStatus, createResp = runHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + tableStatus, tableResp = runHandler(t, m, fs.Client, "CreateTable", createTableRequest(t, "ns", "t")) + } + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.NotZero(t, tableStatus, "the creates have to land between the first delete's scan and its removal") + assert.Equal(t, http.StatusConflict, refusedStatus, "%+v", refusedResp) + assert.Equal(t, http.StatusOK, createStatus, "%+v", createResp) + assert.Equal(t, http.StatusOK, tableStatus, "%+v", tableResp) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(bucketPath, "ns"), "the namespace the create answered for must survive") + assert.Contains(t, fs.Get(bucketPath, "ns").Extended, ExtendedKeyMetadata) + require.NotNil(t, fs.Get(nsPath, "t"), "the table the create answered for must survive") + assert.Contains(t, fs.Get(nsPath, "t").Extended, ExtendedKeyMetadata) + assert.NotNil(t, fs.Get(path.Join(nsPath, "t"), "data"), "and so must its data directory") + assert.NotNil(t, fs.Get(path.Join(bucketPath, "old"), keptDataFile), "the file under the other leftover must survive") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") +} + +// A catalog record under a leftover directory is not the leftover's to remove, +// whatever it holds: an empty table is still a table. The delete refuses at +// it, and lifts the claim it took on the leftover. +func TestDeleteTableBucketRefusesACatalogRecordInALeftover(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + leftover := path.Join(bucketPath, "old") + fs.Put(bucketPath, "old", nil) + fs.Put(leftover, "t", map[string][]byte{ + ExtendedKeyMetadata: []byte("{}"), + ExtendedKeyEntryType: []byte(EntryTypeTable), + }) + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(leftover, "t"), "the record must survive") + assert.Contains(t, fs.Get(leftover, "t").Extended, ExtendedKeyMetadata) + require.NotNil(t, fs.Get(bucketPath, "old")) + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a refused delete lifts its claim on the leftover") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") +} + +// A namespace a create left marked unfinished must not wedge the name: the mark +// says its writer never confirmed it, so the next create takes it over, and +// tables can be created in it once it is confirmed. +func TestCreateNamespaceAdoptsAnUnfinishedNamespace(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + plantCreateMark(fs, bucketPath, "ns") + + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + + assert.NotContains(t, fs.Get(bucketPath, "ns").Extended, ExtendedKeyDeleting, "the adopted namespace ends confirmed") + requireOK(t, m, fs, "CreateTable", createTableRequest(t, "ns", "t")) +} + +// A table answers success once its namespace is open, and says nothing of the +// bucket, so a table created while the namespace's own create still has to +// confirm the bucket would survive as a table under a namespace that +// confirmation backs out. The namespace carries the unfinished mark until the +// confirmation has passed, and is absent to a create under it meanwhile. +func TestCreateTableRefusedInNamespaceStillBeingCreated(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + reached := make(chan struct{}) + release := make(chan struct{}) + // The create confirms the bucket right after writing the namespace; hold it there. + fs.OnLookup = func(dir, name string) { + if dir != TablesPath || name != renameTestBucket || fs.Get(bucketPath, "ns") == nil { + return + } + fs.OnLookup = nil + close(reached) + <-release + } + created := startHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + waitForPause(t, reached, created) + + status, resp := runHandler(t, m, fs.Client, "CreateTable", createTableRequest(t, "ns", "t")) + close(release) + createResult := <-created + + assert.Equal(t, http.StatusNotFound, status, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(path.Join(bucketPath, "ns"), "t"), "a refused create writes nothing") + assert.Equal(t, http.StatusOK, createResult.status, "%+v", createResult.resp) + assert.NotContains(t, fs.Get(bucketPath, "ns").Extended, ExtendedKeyDeleting, "a confirmed namespace carries no mark") + requireOK(t, m, fs, "CreateTable", createTableRequest(t, "ns", "t")) +} + +// A namespace whose create has not confirmed it is absent to every reader, as +// a table's is. A delete that claimed it would replace the create's mark, and +// lifting its own claim on a refusal would leave the namespace looking +// confirmed: open to tables before its bucket check passed, and out of reach of +// the create's back-out. Here the create takes back a leftover holding a file, +// which a delete would refuse on. +func TestDeleteNamespaceLeavesAnUnfinishedNamespaceAlone(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + nsPath := path.Join(bucketPath, "ns") + fs.Put(bucketPath, "ns", nil) + fs.PutFile(nsPath, keptDataFile, time.Now()) + reached := make(chan struct{}) + release := make(chan struct{}) + // The create confirms the bucket right after writing the namespace; hold it there. + fs.OnLookup = func(dir, name string) { + if dir != TablesPath || name != renameTestBucket { + return + } + if entry := fs.Get(bucketPath, "ns"); entry == nil || entry.Extended[ExtendedKeyMetadata] == nil { + return + } + fs.OnLookup = nil + close(reached) + <-release + } + created := startHandler(t, m, fs.Client, "CreateNamespace", createNamespaceNsRequest(t)) + waitForPause(t, reached, created) + + deleteStatus, deleteResp := runHandler(t, m, fs.Client, "DeleteNamespace", + &DeleteNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}) + listed := listedNamespaces(t, m, fs) + tableStatus, tableResp := runHandler(t, m, fs.Client, "CreateTable", createTableRequest(t, "ns", "t")) + close(release) + createResult := <-created + + assert.Equal(t, http.StatusNotFound, deleteStatus, "%+v", deleteResp) + assert.Equal(t, ErrCodeNoSuchNamespace, deleteResp.Type, "%+v", deleteResp) + assert.NotContains(t, listed, "ns", "an unfinished namespace is not listed") + assert.Equal(t, http.StatusNotFound, tableStatus, "%+v", tableResp) + assert.Equal(t, ErrCodeNoSuchNamespace, tableResp.Type, "%+v", tableResp) + assert.Nil(t, fs.Get(nsPath, "t"), "no table is created in it") + assert.Equal(t, http.StatusOK, createResult.status, "%+v", createResult.resp) + assert.NotContains(t, fs.Get(bucketPath, "ns").Extended, ExtendedKeyDeleting, "the create confirms it") + assert.NotNil(t, fs.Get(nsPath, keptDataFile), "the file under it stays") +} + +// A namespace can become unfinished between a delete's read and its claim: a +// create took its name back meanwhile. The claim refuses it as absent, rather +// than replacing the create's mark. +func TestDeleteNamespaceDoesNotClaimAnUnfinishedNamespace(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + // The claim is the delete's first write. + fs.BeforeUpdate = func() { plantCreateMark(fs, bucketPath, "ns") } + + status, resp := runHandler(t, m, fs.Client, "DeleteNamespace", + &DeleteNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}}) + + assert.Equal(t, http.StatusNotFound, status, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(bucketPath, "ns"), "the namespace stays") + assert.Equal(t, []byte(createClaim), fs.Get(bucketPath, "ns").Extended[ExtendedKeyDeleting], "with its create's mark") +} + +// A delete that claims the bucket between the namespace's confirmation and the +// end of its create must still see the namespace, or it would remove a +// namespace the create answered for. The mark does not hide it from the scan. +func TestDeleteTableBucketCountsANamespaceStillBeingCreated(t *testing.T) { + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + plantCreateMark(fs, GetTableBucketPath(renameTestBucket), "ns") + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + assert.Contains(t, resp.Message, "namespace ns", "the hidden namespace is named") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is kept") +} + +// The claim on the bucket is lifted whenever the delete refuses, or the bucket +// would refuse every create; a claim left by a delete that died does not keep +// the next delete from running. +func TestDeleteTableBucketClaimOnTheBucket(t *testing.T) { + lifted := func(t *testing.T, fs *s3tablestest.MemFiler, m *Manager) { + t.Helper() + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.NotContains(t, fs.Get(TablesPath, renameTestBucket).Extended, ExtendedKeyDeleting) + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"other"}, + }) + } + t.Run("lifted when a namespace is found", func(t *testing.T) { + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + lifted(t, fs, m) + }) + t.Run("lifted when a directory holds files", func(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + fs.Put(bucketPath, "old", nil) + fs.PutFile(path.Join(bucketPath, "old"), keptDataFile, time.Now()) + lifted(t, fs, m) + }) + t.Run("replaces the claim of a delete that died", func(t *testing.T) { + fs, m := startEmptyBucket(t) + plantDeleteClaim(fs, TablesPath, renameTestBucket) + + requireOK(t, m, fs, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Nil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is deleted") + }) +} + +// putDirectoryObject stores an S3 object with content on the directory entry +// dir/name, the way the S3 API stores a key that other keys sit under: with the +// prefix-object mark the filer expects on an object landing on a directory. +func putDirectoryObject(t *testing.T, fs *s3tablestest.MemFiler, dir, name string) { + t.Helper() + _, err := fs.Client.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{ + Directory: dir, + Entry: &filer_pb.Entry{ + Name: name, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mime: "application/octet-stream"}, + Content: []byte("object"), + Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}, + }, + }) + require.NoError(t, err) +} + +// markPrefixObject marks the directory dir/name as also holding an empty S3 +// object, which leaves no trace but the mark, keeping its other attributes. +func markPrefixObject(fs *s3tablestest.MemFiler, dir, name string) { + extended := map[string][]byte{} + for key, value := range fs.Get(dir, name).Extended { + extended[key] = value + } + extended[s3_constants.SeaweedFSPrefixObject] = []byte("true") + fs.Put(dir, name, extended) +} + +// beforeDeleteClient runs before just ahead of the first delete of target. +type beforeDeleteClient struct { + filer_pb.SeaweedFilerClient + target string + before func() +} + +func (c *beforeDeleteClient) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest, opts ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + if before := c.before; before != nil && path.Join(req.Directory, req.Name) == c.target { + c.before = nil + before() + } + return c.SeaweedFilerClient.DeleteEntry(ctx, req, opts...) +} + +// An S3 object can be stored on a directory entry: a key other keys sit under, +// or an empty object that leaves nothing but a mark on the directory. Such a +// directory holds no file, yet it is user data, not a leftover: the delete +// refuses wherever it finds one and the object stays. +func TestDeleteTableBucketKeepsADirectoryBackedObject(t *testing.T) { + bucketPath := GetTableBucketPath(renameTestBucket) + leftover := path.Join(bucketPath, "old") + refused := func(t *testing.T, code int, resp S3TablesError, wantType string) { + t.Helper() + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, wantType, resp.Type, "%+v", resp) + } + + t.Run("an empty object in the bucket", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}) + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + refused(t, code, resp, ErrCodeBucketNotEmpty) + require.NotNil(t, fs.Get(bucketPath, "old"), "the object must survive") + assert.True(t, fs.Get(bucketPath, "old").IsPrefixObject()) + }) + t.Run("an object under a leftover directory", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + putDirectoryObject(t, fs, leftover, "a") + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + refused(t, code, resp, ErrCodeBucketNotEmpty) + require.NotNil(t, fs.Get(leftover, "a"), "the object must survive") + assert.Equal(t, []byte("object"), fs.Get(leftover, "a").Content) + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a refused delete lifts its claim on the leftover") + }) + t.Run("a leftover stored as an object after the scan", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + fs.BeforeDelete = func() { putDirectoryObject(t, fs, bucketPath, "old") } + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + refused(t, code, resp, ErrCodeBucketNotEmpty) + require.NotNil(t, fs.Get(bucketPath, "old"), "the object must survive") + assert.Equal(t, []byte("object"), fs.Get(bucketPath, "old").Content) + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "no claim is taken on an object") + }) + t.Run("a leftover stored as an object between the read and the write of its claim", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + // The claim read the bare directory; the object lands before its write, + // which would otherwise put the bare directory back over it. + fs.OnUpdate = func(dir, name string) { + if dir != bucketPath || name != "old" { + return + } + fs.OnUpdate = nil + putDirectoryObject(t, fs, bucketPath, "old") + } + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + refused(t, code, resp, ErrCodeBucketNotEmpty) + require.NotNil(t, fs.Get(bucketPath, "old"), "the object must survive") + assert.Equal(t, []byte("object"), fs.Get(bucketPath, "old").Content) + assert.True(t, fs.Get(bucketPath, "old").IsPrefixObject()) + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "no claim is taken on an object") + }) + t.Run("a leftover marked as an empty object after the walk", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + client := &beforeDeleteClient{SeaweedFilerClient: fs.Client, target: leftover} + client.before = func() { markPrefixObject(fs, bucketPath, "old") } + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + refused(t, code, resp, ErrCodeConflict) + require.NotNil(t, fs.Get(bucketPath, "old"), "the object must survive") + assert.True(t, fs.Get(bucketPath, "old").IsPrefixObject()) + }) +} + +// A directory the delete listed can be removed by another client and a file +// uploaded at its path before the delete reaches it. The file carries no +// attribute a directory would not, so only the directory's type and the claim +// written onto it tell the two apart: the delete refuses and the file stays, +// whether it replaced the leftover itself or a directory under it, and whether +// it landed before that directory was claimed or after. +func TestDeleteTableBucketKeepsAFileThatReplacedADirectory(t *testing.T) { + bucketPath := GetTableBucketPath(renameTestBucket) + leftover := path.Join(bucketPath, "old") + kept := func(t *testing.T, fs *s3tablestest.MemFiler, dir, name string) { + t.Helper() + entry := fs.Get(dir, name) + require.NotNil(t, entry, "the uploaded file must survive") + assert.False(t, entry.IsDirectory) + assert.NotContains(t, entry.Extended, ExtendedKeyDeleting, "no claim is left on the file") + require.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") + } + + t.Run("the leftover, after the scan", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + fs.BeforeDelete = func() { fs.PutFile(bucketPath, "old", time.Now()) } + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + kept(t, fs, bucketPath, "old") + }) + t.Run("a directory under it, after the walk listed it", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + fs.Put(leftover, "child", nil) + // The scan's look for files and the removal's walk each list old, then + // old/child. The file takes child's place as the walk reaches it, after + // the walk listed child as a directory and before child is claimed. + client := &pauseListClient{SeaweedFilerClient: fs.Client, dir: path.Join(leftover, "child"), nth: 2} + client.pause = func() { fs.PutFile(leftover, "child", time.Now()) } + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.Equal(t, 2, client.listings, "the file has to land during the removal's walk") + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + kept(t, fs, leftover, "child") + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a refused delete lifts its claim on the leftover") + }) + t.Run("a directory under it, between the read and the write of its claim", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + fs.Put(leftover, "child", nil) + // The filer refuses to write the claimed directory over the file and + // says so only as an opaque failure. + fs.OnUpdate = func(dir, name string) { + if dir != leftover || name != "child" { + return + } + fs.OnUpdate = nil + fs.PutFile(leftover, "child", time.Now()) + } + + code, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + kept(t, fs, leftover, "child") + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a refused delete lifts its claim on the leftover") + }) + t.Run("a directory under it, after its claim", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + fs.Put(leftover, "child", nil) + client := &beforeDeleteClient{SeaweedFilerClient: fs.Client, target: path.Join(leftover, "child")} + client.before = func() { fs.PutFile(leftover, "child", time.Now()) } + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + require.Nil(t, client.before, "the file has to land right before the delete") + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + kept(t, fs, leftover, "child") + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a refused delete lifts its claim on the leftover") + }) +} + +// failClaimReleaseClient answers the first write that lifts a delete claim from +// target with Unavailable, the way a filer that is briefly unreachable would. +type failClaimReleaseClient struct { + filer_pb.SeaweedFilerClient + target string + failed int +} + +func (c *failClaimReleaseClient) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + if c.failed == 0 && req.Entry != nil && path.Join(req.Directory, req.Entry.Name) == c.target && + req.Entry.Extended[ExtendedKeyDeleting] == nil && len(req.ExpectedExtended[ExtendedKeyDeleting]) > 0 { + c.failed++ + return nil, status.Error(codes.Unavailable, "filer unavailable") + } + return c.SeaweedFilerClient.UpdateEntry(ctx, req, opts...) +} + +// A claim the delete fails to lift refuses every create in the bucket until the +// next delete replaces it, so a release the filer turns away once is retried. +func TestDeleteTableBucketRetriesTheClaimRelease(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + fs.Put(bucketPath, "old", nil) + fs.PutFile(path.Join(bucketPath, "old"), keptDataFile, time.Now()) + client := &failClaimReleaseClient{SeaweedFilerClient: fs.Client, target: bucketPath} + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + require.Equal(t, 1, client.failed, "the first release has to fail") + assert.NotContains(t, fs.Get(TablesPath, renameTestBucket).Extended, ExtendedKeyDeleting, "a later attempt lifts the claim") + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) +} + +// A claim the removal fails to lift from a leftover directory makes a create +// read the leftover's name as taken, and one left on a directory under it counts +// as a child of the namespace the leftover may become. A release the filer turns +// away once is retried for both. +func TestDeleteTableBucketRetriesLeftoverClaimReleases(t *testing.T) { + bucketPath := GetTableBucketPath(renameTestBucket) + leftover := path.Join(bucketPath, "old") + + t.Run("on the leftover", func(t *testing.T) { + fs, m := startEmptyBucket(t) + fs.Put(bucketPath, "old", nil) + failing := &failClaimReleaseClient{SeaweedFilerClient: fs.Client, target: leftover} + // A file lands in the leftover after the scan, so the walk refuses it and + // the leftover's claim has to come off. + client := &beforeDeleteClient{SeaweedFilerClient: failing, target: GetTableObjectBucketPath(renameTestBucket)} + client.before = func() { fs.PutFile(leftover, keptDataFile, time.Now()) } + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + require.Equal(t, 1, failing.failed, "the first release has to fail") + assert.NotContains(t, fs.Get(bucketPath, "old").Extended, ExtendedKeyDeleting, "a later attempt lifts the claim") + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"old"}}) + }) + t.Run("on a directory under it", func(t *testing.T) { + fs, m := startEmptyBucket(t) + child := path.Join(leftover, "child") + fs.Put(bucketPath, "old", nil) + fs.Put(leftover, "child", nil) + failing := &failClaimReleaseClient{SeaweedFilerClient: fs.Client, target: child} + // A file lands in child after its claim, so its delete is refused and + // the claim has to come off. + client := &beforeDeleteClient{SeaweedFilerClient: failing, target: child} + client.before = func() { fs.PutFile(child, keptDataFile, time.Now()) } + + code, resp := runHandler(t, m, client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + require.Equal(t, 1, failing.failed, "the first release has to fail") + assert.NotContains(t, fs.Get(leftover, "child").Extended, ExtendedKeyDeleting, "a later attempt lifts the claim") + assert.NotNil(t, fs.Get(child, keptDataFile), "the file stays") + }) +} + +// The scan reads the bucket a page at a time, and the leftovers it collects +// along the way do not decide anything. With more bare directories than one page +// holds, the namespace sorted after them is on a later page: stopping at the +// first page would find nothing but leftovers and delete the bucket with it. +func TestDeleteTableBucketFindsANamespacePastLeftovers(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + for i := 0; i <= 10; i++ { + fs.Put(bucketPath, fmt.Sprintf("a%02d", i), nil) + } + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"zz"}, + }) + + status, resp := runHandler(t, m, fs.Client, "DeleteTableBucket", deleteTableBucketRequest(t)) + + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeBucketNotEmpty, resp.Type, "%+v", resp) + assert.NotNil(t, fs.Get(bucketPath, "zz"), "the namespace must survive") + assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket must survive") +} diff --git a/weed/s3api/s3tables/handler_delete_create_race_test.go b/weed/s3api/s3tables/handler_delete_create_race_test.go index 38f075d3a9..b46b9fb8d0 100644 --- a/weed/s3api/s3tables/handler_delete_create_race_test.go +++ b/weed/s3api/s3tables/handler_delete_create_race_test.go @@ -647,6 +647,22 @@ func TestDeleteNamespaceWithTableLiftsItsClaim(t *testing.T) { } } +// A claim the delete fails to lift refuses every create in the namespace, so a +// release the filer turns away once is retried. +func TestDeleteNamespaceRetriesTheClaimRelease(t *testing.T) { + fs, m := startRenameManager(t) + bucketPath := GetTableBucketPath(renameTestBucket) + before := fs.Get(bucketPath, "ns").Extended + client := &failClaimReleaseClient{SeaweedFilerClient: fs.Client, target: GetNamespacePath(renameTestBucket, "ns")} + + code, resp := runHandler(t, m, client, "DeleteNamespace", deleteNamespaceRequest(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeNamespaceNotEmpty, resp.Type, "%+v", resp) + require.Equal(t, 1, client.failed, "the first release has to fail") + assert.Equal(t, before, fs.Get(bucketPath, "ns").Extended, "a later attempt lifts the claim") +} + // 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) { diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index a4690f491a..2683e06fd0 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -170,20 +170,50 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R } namespacePath := GetNamespacePath(bucketName, namespaceName) + // The authorization above holds for this bucket only. One deleted and + // created again under the same name since then is another bucket, and is + // treated as gone by both checks below. + isAuthorizedBucket := isTableBucketCreatedAt(bucketMetadata.CreatedAt) - // Check if namespace already exists + // A delete that has claimed the bucket is scanning it for namespaces, so + // refuse before writing anything. Checked after the authorization above, so + // denied callers never see the delete. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - _, err := h.readNamespaceMetadataAttr(r.Context(), client, namespacePath) + return h.checkParentOpen(r.Context(), client, bucketPath, isAuthorizedBucket) + }) + if h.writeBucketClosedError(w, err, bucketName) { return err + } + if err != nil { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table bucket: %v", err)) + return err + } + + // Check if namespace already exists. One still marked unfinished does not: + // its create either confirmed it or lost it, and the mark left behind would + // wedge the name, so this create takes it over the way a table's does. + exists := false + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(r.Context(), client, namespacePath) + if errors.Is(err, filer_pb.ErrNotFound) { + return nil + } + if err != nil { + return err + } + _, hasMetadata := entry.Extended[ExtendedKeyMetadata] + exists = hasMetadata && !isUnfinishedCreate(entry.Extended) + return nil }) - if err == nil { - h.writeError(w, http.StatusConflict, ErrCodeNamespaceAlreadyExists, fmt.Sprintf("namespace %s already exists", namespaceName)) - return fmt.Errorf("namespace already exists") - } else if !errors.Is(err, filer_pb.ErrNotFound) { + if err != nil { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err)) return err } + if exists { + h.writeError(w, http.StatusConflict, ErrCodeNamespaceAlreadyExists, fmt.Sprintf("namespace %s already exists", namespaceName)) + return fmt.Errorf("namespace already exists") + } // Create the namespace with bucket owner to maintain consistency // (authorization above ensures the caller has permission to create in this bucket) @@ -201,7 +231,13 @@ 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} + // The unfinished mark keeps a table or view from being created in the + // namespace until the bucket check below has passed: one created before it + // would be left under a namespace that check then backs out. + attrs := map[string][]byte{ + ExtendedKeyMetadata: metadataBytes, + ExtendedKeyDeleting: []byte(createClaim), + } var taken error err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err := h.createDirectoryExclusive(r.Context(), client, namespacePath, attrs) @@ -215,7 +251,19 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R return nil } } - return err + if err != nil { + return err + } + // A delete may have claimed the bucket since the check above, after it + // scanned for namespaces and before this write. Confirm the bucket once + // more now that the namespace is there: a delete claiming it later scans + // the namespace, marked or not, and one that claimed it already makes + // this back out what it wrote. The mark comes off only after that check + // and the bucket is not checked again: a table created in the window + // between the two would be left under a namespace the second check backs out. + // A bucket recreated since the authorization fails the check as gone, so + // the namespace is backed out of it rather than left with this owner. + return h.confirmParentOnce(r.Context(), client, bucketPath, isAuthorizedBucket, namespacePath, attrs, namespacePath) }) if err == nil && taken != nil { err = taken @@ -228,6 +276,9 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R h.writeError(w, http.StatusConflict, ErrCodeConflict, "namespace changed during create, retry the request") return err } + if h.writeBucketClosedError(w, err, bucketName) { + return err + } if err != nil { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to create namespace") return err @@ -243,6 +294,25 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R return nil } +// writeBucketClosedError answers a create refused because its bucket is being +// created, being deleted, or was deleted, and reports whether err was one of +// those. +func (h *S3TablesHandler) writeBucketClosedError(w http.ResponseWriter, err error, bucketName string) bool { + switch { + case errors.Is(err, errParentCreating): + h.writeError(w, http.StatusConflict, ErrCodeConflict, fmt.Sprintf("table bucket %s is still being created, retry the request", bucketName)) + case errors.Is(err, errParentDeleting): + h.writeError(w, http.StatusConflict, ErrCodeConflict, fmt.Sprintf("table bucket %s is being deleted", bucketName)) + case errors.Is(err, errParentGone): + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + case errors.Is(err, errBackOutFailed): + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("table bucket %s was deleted during the request and the new namespace may remain", bucketName)) + default: + return false + } + return true +} + // handleGetNamespace gets details of a namespace func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { var req GetNamespaceRequest @@ -580,9 +650,10 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re continue } - // Read metadata from extended attribute + // Read metadata from extended attribute. A namespace its create has + // not confirmed is not one yet. data, ok := entry.Entry.Extended[ExtendedKeyMetadata] - if !ok { + if !ok || isUnfinishedCreate(entry.Entry.Extended) { continue } @@ -718,11 +789,14 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R } // Claim before scanning, so a create that writes after the scan sees it. + // A namespace whose create has not confirmed it is absent, and claiming it + // would replace the create's mark: a refused delete then lifts the claim and + // leaves the namespace looking confirmed before its create checked the bucket. 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) + claim, claimed, claimErr = h.claimForDelete(r.Context(), client, namespacePath, isFinishedNamespaceEntry) return claimErr }) if err != nil { diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 8e6ce93c96..a9df3373d8 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -2270,9 +2270,12 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re } // writeNamespaceClosedError answers a create refused because its namespace is -// being deleted or was deleted, and reports whether err was one of those. +// being created, 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, errParentCreating): + h.writeError(w, http.StatusConflict, ErrCodeConflict, fmt.Sprintf("namespace %s is still being created, retry the request", namespaceName)) 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): @@ -2285,13 +2288,14 @@ func (h *S3TablesHandler) writeNamespaceClosedError(w http.ResponseWriter, err e 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. +// refuseClaimedNamespace answers 409 when a delete has claimed the namespace or +// its create is not confirmed yet. 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) { + if errors.Is(err, errParentDeleting) || errors.Is(err, errParentCreating) { h.writeNamespaceClosedError(w, err, namespaceName) return err } diff --git a/weed/s3api/s3tables/s3tablestest/memfiler.go b/weed/s3api/s3tables/s3tablestest/memfiler.go index 391a2f0331..292a7c94cd 100644 --- a/weed/s3api/s3tables/s3tablestest/memfiler.go +++ b/weed/s3api/s3tables/s3tablestest/memfiler.go @@ -204,12 +204,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. + // the delete when the entry no longer matches what the caller captured. It + // does so before anything else, so a directory that changed hands is a + // changed entry whether or not it also holds something. for key, expected := range req.ExpectedExtended { var actual []byte if d, ok := f.entries[req.Directory]; ok { @@ -221,6 +219,10 @@ func (f *MemFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryReque return nil, status.Errorf(codes.FailedPrecondition, "extended attribute %q changed", key) } } + // The real filer then 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 + } if d, ok := f.entries[req.Directory]; ok { delete(d, req.Name) } diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 9b3704aa60..d647668b18 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -43,6 +43,10 @@ var handlers = map[string]func(*Server, http.ResponseWriter, *http.Request){ // lance table ids put indexes under a "vector" namespace inside their bucket. const indexNamespace = s3tables.VectorIndexNamespace +// ensureIndexNamespaceAttempts bounds how often ensureIndexNamespace declares +// the namespace again after losing it to a concurrent declare. +const ensureIndexNamespaceAttempts = 3 + // Vector bucket records are s3tables table buckets; the S3 Vectors ARN form is // arn:aws:s3vectors:{region}:{account}:bucket/{name}[/index/{name}]. func (s *Server) vectorBucketArn(bucket string) string { @@ -220,26 +224,36 @@ func vectorErr(w http.ResponseWriter, err error) { // calls are internal: a caller's CreateTable grant is what authorizes the // index, so it must not also need namespace reads or writes. func (s *Server) ensureIndexNamespace(ctx context.Context, bucket string) error { - var resp map[string]interface{} - err := s.internalCatalog(ctx, "GetNamespace", &s3tables.GetNamespaceRequest{ - TableBucketARN: s.s3tablesArn(bucket), - Namespace: []string{indexNamespace}, - }, &resp) - if err == nil { - return nil + var err error + for attempt := 0; attempt < ensureIndexNamespaceAttempts; attempt++ { + if attempt > 0 { + time.Sleep(time.Duration(attempt) * 50 * time.Millisecond) + } + var resp map[string]interface{} + err = s.internalCatalog(ctx, "GetNamespace", &s3tables.GetNamespaceRequest{ + TableBucketARN: s.s3tablesArn(bucket), + Namespace: []string{indexNamespace}, + }, &resp) + if err == nil { + return nil + } + if code := catalogErrCode(err); code != s3tables.ErrCodeNoSuchNamespace && code != s3tables.ErrCodeNoSuchBucket { + return err + } + err = s.internalCatalog(ctx, "CreateNamespace", &s3tables.CreateNamespaceRequest{ + TableBucketARN: s.s3tablesArn(bucket), + Namespace: []string{indexNamespace}, + }, &resp) + // A concurrent CreateIndex may have won the create, or taken over this + // one's namespace before it was confirmed. + if err == nil || catalogErrCode(err) == s3tables.ErrCodeNamespaceAlreadyExists { + return nil + } + if catalogErrCode(err) != s3tables.ErrCodeConflict { + return err + } } - if code := catalogErrCode(err); code != s3tables.ErrCodeNoSuchNamespace && code != s3tables.ErrCodeNoSuchBucket { - return err - } - err = s.internalCatalog(ctx, "CreateNamespace", &s3tables.CreateNamespaceRequest{ - TableBucketARN: s.s3tablesArn(bucket), - Namespace: []string{indexNamespace}, - }, &resp) - // A concurrent CreateIndex may have won the create. - if err != nil && catalogErrCode(err) != s3tables.ErrCodeNamespaceAlreadyExists { - return err - } - return nil + return err } func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request) { diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go index 7aad7447bb..32dbef2d70 100644 --- a/weed/s3api/s3vectors/handlers_test.go +++ b/weed/s3api/s3vectors/handlers_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "strconv" "strings" + "sync" "testing" "time" @@ -983,3 +984,55 @@ func TestVectorKeysBoundsRejectsBeforeEngine(t *testing.T) { t.Fatalf("engine saw %d get/%d delete calls, want none", len(f.gets), len(f.deletes)) } } + +// Two indexes created at once in a new bucket both declare the implicit +// namespace. One create takes over the other's namespace while it is still +// marked unfinished, so the other's confirmation fails; its index is created +// all the same. +func TestCreateIndexSurvivesAConcurrentNamespaceCreate(t *testing.T) { + fs := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0)) + if w := postJSON(t, s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + bucketPath := s3tables.GetTableBucketPath("vec-bucket") + reached := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + releaseFirst := func() { releaseOnce.Do(func() { close(release) }) } + t.Cleanup(releaseFirst) + held := false + // Hold the first create where it confirms the bucket after writing the namespace. + fs.OnLookup = func(dir, name string) { + if held || dir != s3tables.TablesPath || name != "vec-bucket" { + return + } + ns := fs.Get(bucketPath, indexNamespace) + if ns == nil || string(ns.Extended[s3tables.ExtendedKeyDeleting]) != "create" { + return + } + held = true + close(reached) + <-release + } + createIndex := func(name string) *httptest.ResponseRecorder { + return postJSON(t, s.handleCreateIndex, &CreateIndexRequest{ + VectorBucketName: "vec-bucket", IndexName: name, DataType: "float32", Dimension: 2, DistanceMetric: "cosine", + }) + } + first := make(chan *httptest.ResponseRecorder, 1) + go func() { first <- createIndex("first") }() + select { + case <-reached: + case <-time.After(10 * time.Second): + t.Fatal("the first create never wrote an unfinished namespace") + } + + if w := createIndex("second"); w.Code != http.StatusOK { + t.Fatalf("second create index: %d %s", w.Code, w.Body.String()) + } + releaseFirst() + if w := <-first; w.Code != http.StatusOK { + t.Fatalf("first create index: %d %s", w.Code, w.Body.String()) + } +}