diff --git a/test/s3/versioning/s3_preversioning_delete_test.go b/test/s3/versioning/s3_preversioning_delete_test.go new file mode 100644 index 000000000..bf6653c42 --- /dev/null +++ b/test/s3/versioning/s3_preversioning_delete_test.go @@ -0,0 +1,293 @@ +package s3api + +import ( + "context" + "sort" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Deleting a key whose null version predates versioning records the delete +// marker under .versions but leaves the null object at the base path. +// The listing must retract that stale entry, on whichever page it lands. + +func deleteObject(t *testing.T, client *s3.Client, bucket, key string) { + t.Helper() + resp, err := client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{ + Bucket: aws.String(bucket), Key: aws.String(key), + }) + require.NoError(t, err) + require.True(t, resp.DeleteMarker != nil && *resp.DeleteMarker, "the delete must record a delete marker") +} + +func TestPreVersioningNullObjectDeleteHidesKey(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "k.txt", "pre-versioning") + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "k.txt") + + assert.Empty(t, listAllKeys(t, client, bucketName, 0), "the deleted key must leave the listing") + + _, err := client.GetObject(context.TODO(), &s3.GetObjectInput{ + Bucket: aws.String(bucketName), Key: aws.String("k.txt"), + }) + assert.Error(t, err, "the deleted key must not be readable") + + versions, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{ + Bucket: aws.String(bucketName), + }) + require.NoError(t, err) + require.Len(t, versions.DeleteMarkers, 1) + assert.True(t, *versions.DeleteMarkers[0].IsLatest, "the delete marker is the current version") + require.Len(t, versions.Versions, 1) + assert.Equal(t, "null", *versions.Versions[0].VersionId) + assert.False(t, *versions.Versions[0].IsLatest, "the null version is shadowed by the marker") +} + +func TestPreVersioningNullObjectDeleteAfterOverwriteHidesKey(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "k.txt", "pre-versioning") + enableVersioning(t, client, bucketName) + putObject(t, client, bucketName, "k.txt", "versioned overwrite") + deleteObject(t, client, bucketName, "k.txt") + + assert.Empty(t, listAllKeys(t, client, bucketName, 0), "the deleted key must leave the listing") +} + +// The retraction and the replacement must both survive a page boundary landing +// between a key and its .versions sibling. +func TestPreVersioningNullObjectAcrossPageBoundary(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "a.txt", "old") + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "a.txt") + putObjectVersioned(t, client, bucketName, "b.txt") + + assert.Equal(t, []string{"b.txt"}, listAllKeys(t, client, bucketName, 1), + "a page ending on the stale null object must still retract it") +} + +func TestPreVersioningNullObjectMetadataAcrossPageBoundary(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + overwrite := "versioned overwrite" + putObject(t, client, bucketName, "a.txt", "old") + enableVersioning(t, client, bucketName) + putObject(t, client, bucketName, "a.txt", overwrite) + putObjectVersioned(t, client, bucketName, "b.txt") + + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), MaxKeys: aws.Int32(1), + }) + require.NoError(t, err) + require.Len(t, page.Contents, 1) + assert.Equal(t, "a.txt", *page.Contents[0].Key) + assert.Equal(t, int64(len(overwrite)), *page.Contents[0].Size, + "a page ending on the null object must still pick up the current version's metadata") +} + +// A key such as "a.txt.bak" sorts between "a.txt" and "a.txt.versions", so the +// sibling's outcome arrives entries later, possibly on a later page. +func TestPreVersioningInterveningKeyRetraction(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "a.txt", "pre-versioning") + putObject(t, client, bucketName, "a.txt!between", "pre-versioning") + putObject(t, client, bucketName, "a.txt.bak", "pre-versioning") + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "a.txt") + + for _, maxKeys := range []int32{0, 1, 2} { + assert.Equal(t, []string{"a.txt!between", "a.txt.bak"}, listAllKeys(t, client, bucketName, maxKeys), + "maxKeys=%d must not list the deleted key", maxKeys) + } +} + +func TestPreVersioningInterveningKeyMetadata(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + overwrite := "versioned overwrite" + putObject(t, client, bucketName, "a.txt", "old") + putObject(t, client, bucketName, "a.txt!between", "pre-versioning") + putObject(t, client, bucketName, "a.txt.bak", "pre-versioning") + enableVersioning(t, client, bucketName) + putObject(t, client, bucketName, "a.txt", overwrite) + + for _, maxKeys := range []int32{1, 2} { + var keys []string + var token *string + for { + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), MaxKeys: aws.Int32(maxKeys), ContinuationToken: token, + }) + require.NoError(t, err) + for _, o := range page.Contents { + keys = append(keys, *o.Key) + if *o.Key == "a.txt" { + assert.Equal(t, int64(len(overwrite)), *o.Size, + "maxKeys=%d must list the current version's metadata", maxKeys) + } + } + if page.IsTruncated == nil || !*page.IsTruncated { + break + } + token = page.NextContinuationToken + } + assert.Equal(t, []string{"a.txt", "a.txt!between", "a.txt.bak"}, keys, "maxKeys=%d", maxKeys) + } +} + +// CommonPrefixes derive from listable keys, so deleting the only pre-versioning +// object under a prefix takes the prefix with it. +func TestPreVersioningDeletedPrefixHidesCommonPrefix(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "p/k.txt", "pre-versioning") + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "p/k.txt") + + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), Delimiter: aws.String("/"), + }) + require.NoError(t, err) + assert.Empty(t, page.Contents) + assert.Empty(t, page.CommonPrefixes, "no listable key remains under p/") +} + +// A non-slash delimiter derives prefixes straight from base-path keys, so a +// delete-marked null object must take its prefix along; a live key under the +// same prefix brings it back. +func TestPreVersioningDeletedKeyHidesCustomDelimiterPrefix(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "group-item", "pre-versioning") + putObject(t, client, bucketName, "solo", "pre-versioning") + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "group-item") + + listDashed := func() (keys, prefixes []string) { + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), Delimiter: aws.String("-"), + }) + require.NoError(t, err) + for _, o := range page.Contents { + keys = append(keys, *o.Key) + } + for _, p := range page.CommonPrefixes { + prefixes = append(prefixes, *p.Prefix) + } + return + } + + keys, prefixes := listDashed() + assert.Equal(t, []string{"solo"}, keys) + assert.Empty(t, prefixes, "the deleted key was the prefix's only backer") + + putObject(t, client, bucketName, "group-live", "versioned") + keys, prefixes = listDashed() + assert.Equal(t, []string{"solo"}, keys) + assert.Equal(t, []string{"group-"}, prefixes, "a live key restores the prefix") +} + +// A delete marker for a version-only key (never a base object) must not debit +// the backers of a prefix a live null object still stands behind. +func TestVersionOnlyMarkerLeavesForeignPrefixAlone(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "group-a", "pre-versioning") + enableVersioning(t, client, bucketName) + putObjectVersioned(t, client, bucketName, "group-b") + deleteObject(t, client, bucketName, "group-b") + + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), Delimiter: aws.String("-"), + }) + require.NoError(t, err) + assert.Empty(t, page.Contents) + require.Len(t, page.CommonPrefixes, 1, "group-a still backs the prefix") + assert.Equal(t, "group-", *page.CommonPrefixes[0].Prefix) +} + +// Nested names (k, k!, k!!, ...) can hold more unresolved null objects than the +// pending cap; the evicted key must still be settled, on any page size. +func TestPreVersioningNestedNullObjectsBeyondCap(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + var live []string + for i := 0; i < 10; i++ { + key := "k" + strings.Repeat("!", i) + putObject(t, client, bucketName, key, "pre-versioning") + if i > 0 { + live = append(live, key) + } + } + enableVersioning(t, client, bucketName) + deleteObject(t, client, bucketName, "k") + + sort.Strings(live) + for _, maxKeys := range []int32{0, 3, 9} { + assert.Equal(t, live, listAllKeys(t, client, bucketName, maxKeys), + "maxKeys=%d must not list the deleted key", maxKeys) + } +} + +// A suspended-versioning write is the current null version; its .versions +// sibling (whose latest pointer the write cleared) must not list it a second +// time or resurrect an older version's metadata. +func TestSuspendedNullObjectListsOnce(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + enableVersioning(t, client, bucketName) + putObject(t, client, bucketName, "k.txt", "versioned") + suspendVersioning(t, client, bucketName) + current := "suspended current" + putObject(t, client, bucketName, "k.txt", current) + + page, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{ + Bucket: aws.String(bucketName), + }) + require.NoError(t, err) + require.Len(t, page.Contents, 1, "one key must list exactly once") + assert.Equal(t, int64(len(current)), *page.Contents[0].Size, "the null version is current") +} diff --git a/weed/s3api/s3_constants/extend_key.go b/weed/s3api/s3_constants/extend_key.go index fe7c9ec81..978d021e0 100644 --- a/weed/s3api/s3_constants/extend_key.go +++ b/weed/s3api/s3_constants/extend_key.go @@ -28,6 +28,12 @@ const ( // the entry's own mtime so legacy data still expires. ExtNoncurrentSinceNsKey = "Seaweed-X-Amz-Noncurrent-Since-Ns" + // Set on a .versions directory when a suspended-versioning write made the + // base-path null object the current version; cleared whenever a version in + // the directory becomes current. Unlike an absent latest pointer, which a + // replica may simply not have received yet, this is an explicit signal. + ExtNullVersionIsLatestKey = "Seaweed-X-Amz-Null-Version-Is-Latest" + // Per-bucket opt-in for the PutObject lifecycle TTL fast path ("true" // to enable). When on, an Expiration.Days rule is stamped as a volume // TTL at write time instead of being expired by the worker. Off by diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go index 23785cb31..0b8de645f 100644 --- a/weed/s3api/s3api_object_handlers_list.go +++ b/weed/s3api/s3api_object_handlers_list.go @@ -17,6 +17,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) type OptionalString struct { @@ -298,40 +300,208 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq return } + alignedMarker := marker + // check filer err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + // The failover wrapper retries this callback on another filer after a + // transport error, so every attempt rebuilds the page from scratch: a + // partially built page must not leak into the retry. + contents = nil + commonPrefixes = nil + doErr = nil + nextMarker = "" + marker = alignedMarker + *cursor = ListingCursor{ + maxKeys: maxKeys, + prefixEndsOnDelimiter: strings.HasSuffix(originalPrefix, "/") && len(originalMarker) == 0, + } + var lastEntryWasCommonPrefix bool var lastCommonPrefix string + // Backing for the newest CommonPrefix: which unsettled null objects stand + // behind it, and whether anything definitely listable does. A prefix whose + // null backers all settle as delete-marked names nothing and is retracted; + // tracking backers by key keeps a marker that never joined the prefix (a + // version-only key with no base object) from debiting it. Contributors to + // a prefix stream contiguously, so only the newest prefix needs this. + var lastPrefixNullBackers map[string]bool + var lastPrefixConfirmed bool // Hoist versioning check out of per-entry callback versioningState, _ := s3a.getVersioningState(bucket) - versioningEnabled := versioningState == "Enabled" - // Suspending versioning keeps the delete markers already written, so both states - // can hold a directory whose objects are all gone from the current-version view. - cursor.hideDeletedPrefixes = versioningState != "" + // Suspending versioning keeps the .versions directories already written, so both + // states can emit a key twice (base-path null object plus its .versions sibling) + // and can hold a directory whose objects are all gone from the current-version view. + versioningConfigured := versioningState != "" + cursor.hideDeletedPrefixes = versioningConfigured // Helper function to handle dedup/append logic appendOrDedup := func(newEntry ListEntry) { - if versioningEnabled { - // For versioned buckets, we need to handle duplicates between the main file and the .versions directory - if len(contents) > 0 && contents[len(contents)-1].Key == newEntry.Key { - glog.V(3).Infof("listFilerEntries deduplicating versioned entry: %s", newEntry.Key) - contents[len(contents)-1] = newEntry - } else { - contents = append(contents, newEntry) - cursor.maxKeys-- + if versioningConfigured { + // A key's .versions sibling resolves after every key that sorts between + // them ("k.bak" lists between "k" and "k.versions"), so the base entry is + // found by scanning back through the page and a late resolution is + // inserted where it keeps the page sorted, not at the end. + insertAt := len(contents) + for insertAt > 0 && contents[insertAt-1].Key > newEntry.Key { + insertAt-- } + if insertAt > 0 && contents[insertAt-1].Key == newEntry.Key { + glog.V(3).Infof("listFilerEntries deduplicating versioned entry: %s", newEntry.Key) + contents[insertAt-1] = newEntry + return + } + contents = append(contents, ListEntry{}) + copy(contents[insertAt+1:], contents[insertAt:]) + contents[insertAt] = newEntry + cursor.maxKeys-- } else { contents = append(contents, newEntry) cursor.maxKeys-- } } + // Null objects whose .versions sibling has not streamed yet, and so may still + // turn out to be shadowed by a delete marker. Entries stream in order, so a + // pending null is dropped once the stream passes its sibling's name; the cap + // only binds on pathological prefix nests and falls back to sibling-adjacent + // behavior for the evicted oldest. + type pendingNull struct{ dir, name string } + var pendingNulls []pendingNull + prunePendingNulls := func(dir, passedName string) { + kept := pendingNulls[:0] + for _, p := range pendingNulls { + if p.dir != dir || p.name+s3_constants.VersionsFolder >= passedName { + kept = append(kept, p) + } + } + pendingNulls = kept + } + dropPendingNull := func(dir, name string) { + kept := pendingNulls[:0] + for _, p := range pendingNulls { + if p.dir != dir || p.name != name { + kept = append(kept, p) + } + } + pendingNulls = kept + } + // prefixForKey returns the CommonPrefix a key folds into under the request's + // delimiter, derived exactly as the emission sites derive it, or "". + prefixForKey := func(dir, name string) string { + if delimiter == "" { + return "" + } + undelimited := strings.TrimPrefix((dir + "/" + name)[len(bucketPrefix):], originalPrefix) + if parts := strings.SplitN(undelimited, delimiter, 2); len(parts) == 2 { + return originalPrefix + parts[0] + delimiter + } + return "" + } + + retractPrefixBacking := func(prefix, dir, name string) { + if prefix != lastCommonPrefix || !lastPrefixNullBackers[dir+"/"+name] { + return + } + delete(lastPrefixNullBackers, dir+"/"+name) + if len(lastPrefixNullBackers) == 0 && !lastPrefixConfirmed && + len(commonPrefixes) > 0 && commonPrefixes[len(commonPrefixes)-1].Prefix == prefix { + commonPrefixes = commonPrefixes[:len(commonPrefixes)-1] + cursor.maxKeys++ + lastEntryWasCommonPrefix = false + lastCommonPrefix = "" + } + } + + // The null object for a key lists before its .versions sibling can reveal + // that the current version is a delete marker, so the reveal retracts it - + // from the page's keys, or from the CommonPrefix it was folded into. + cursor.retractEntry = func(dir, name string) { + dropPendingNull(dir, name) + dirName, entryName, _ := entryUrlEncode(dir, name, encodingTypeUrl) + key := fmt.Sprintf("%s/%s", dirName, entryName)[len(bucketPrefix):] + for i := len(contents) - 1; i >= 0 && contents[i].Key >= key; i-- { + if contents[i].Key == key { + contents = append(contents[:i], contents[i+1:]...) + cursor.maxKeys++ + return + } + } + if prefix := prefixForKey(dir, name); prefix != "" { + retractPrefixBacking(prefix, dir, name) + } + } + + // Only a definitive not-found means the null object is live; a transient + // failure leaves the entry unsettled and must not commit it to the page, + // since the next page would then skip the sibling for good. + settlePendingNull := func(p pendingNull) error { + versionsEntry, err := s3a.getEntry(p.dir, p.name+s3_constants.VersionsFolder) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound { + return nil + } + return fmt.Errorf("settle null object %s/%s: %w", p.dir, p.name, err) + } + fullObjectPath := strings.TrimPrefix(p.dir+"/"+p.name, bucketPrefix) + latest, lerr := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, versionsEntry) + switch { + case lerr == nil: + if prefix := prefixForKey(p.dir, p.name); prefix != "" { + // The key folds into a CommonPrefix; a live current version + // confirms the prefix rather than surfacing the key. + if prefix == lastCommonPrefix { + lastPrefixConfirmed = true + } + } else { + dirName, entryName, _ := entryUrlEncode(p.dir, latest.Name, encodingTypeUrl) + appendOrDedup(newListEntry(s3a, latest, "", dirName, entryName, bucketPrefix, fetchOwner, false, false)) + } + case errors.Is(lerr, ErrDeleteMarker), errors.Is(lerr, filer_pb.ErrNotFound): + cursor.retractEntry(p.dir, p.name) + default: + return fmt.Errorf("settle null object %s/%s: %w", p.dir, p.name, lerr) + } + return nil + } + + addPendingNull := func(dir, name string) { + if len(pendingNulls) >= 8 { + // Settle rather than silently evict: an unsettled null would leak past + // the page, and the resume skip would then keep it stale for good. A + // failed settlement is retained for page-close resolution to retry. + settled := pendingNulls[0] + pendingNulls = pendingNulls[1:] + if settleErr := settlePendingNull(settled); settleErr != nil { + pendingNulls = append([]pendingNull{settled}, pendingNulls...) + } + } + pendingNulls = append(pendingNulls, pendingNull{dir, name}) + } + + // A page may fill while a trailing null object's .versions sibling is still + // unstreamed; settle each one by direct lookup before the page is declared + // final. A retraction here reopens the page's quota, and a failure fails the + // listing rather than committing an unsettled entry. + cursor.resolvePendingNulls = func() error { + for len(pendingNulls) > 0 { + p := pendingNulls[0] + pendingNulls = pendingNulls[1:] + if settleErr := settlePendingNull(p); settleErr != nil { + pendingNulls = append([]pendingNull{p}, pendingNulls...) + return settleErr + } + } + return nil + } + for { empty := true nextMarker, doErr = s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: reqDir, prefix: prefix, marker: marker, delimiter: delimiter, bucket: bucket}, cursor, func(dir string, entry *filer_pb.Entry) { empty = false + prunePendingNulls(dir, entry.Name) dirName, entryName, _ := entryUrlEncode(dir, entry.Name, encodingTypeUrl) if entry.IsDirectory { if originalPrefix != "" { @@ -370,9 +540,11 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq delimiterFound = true lastEntryWasCommonPrefix = true lastCommonPrefix = delimitedPrefix + lastPrefixNullBackers, lastPrefixConfirmed = nil, true } else { // This directory object belongs to an existing CommonPrefix, skip it delimiterFound = true + lastPrefixConfirmed = true } } @@ -399,6 +571,7 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq cursor.maxKeys-- lastEntryWasCommonPrefix = true lastCommonPrefix = dirPrefix + lastPrefixNullBackers, lastPrefixConfirmed = nil, true } } else { var delimiterFound bool @@ -416,6 +589,15 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq // S3 clients expect the delimited prefix to contain the delimiter and prefix. delimitedPrefix := originalPrefix + delimitedPath[0] + delimiter + // A null object rolled into a prefix still awaits its .versions + // sibling, and the prefix must not outlive its only backers. + isNullBacker := false + if versioningConfigured { + if vid := string(entry.Extended[s3_constants.ExtVersionIdKey]); vid == "" || vid == "null" { + isNullBacker = true + } + } + for i := range commonPrefixes { if commonPrefixes[i].Prefix == delimitedPrefix { delimiterFound = true @@ -431,10 +613,27 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq delimiterFound = true lastEntryWasCommonPrefix = true lastCommonPrefix = delimitedPrefix + lastPrefixNullBackers, lastPrefixConfirmed = nil, !isNullBacker + if isNullBacker { + lastPrefixNullBackers = map[string]bool{dir + "/" + entry.Name: true} + addPendingNull(dir, entry.Name) + } } else { // This object belongs to an existing CommonPrefix, skip it // but continue processing to maintain correct flow delimiterFound = true + if delimitedPrefix == lastCommonPrefix { + if isNullBacker { + if lastPrefixNullBackers == nil { + lastPrefixNullBackers = map[string]bool{} + } + lastPrefixNullBackers[dir+"/"+entry.Name] = true + addPendingNull(dir, entry.Name) + } else { + lastPrefixConfirmed = true + dropPendingNull(dir, entry.Name) + } + } } } } @@ -443,6 +642,15 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq newEntry := newListEntry(s3a, entry, "", dirName, entryName, bucketPrefix, fetchOwner, false, false) appendOrDedup(newEntry) lastEntryWasCommonPrefix = false + if versioningConfigured { + // A real version id means the .versions sibling resolved this + // key; anything else is a null object the sibling may shadow. + if vid := string(entry.Extended[s3_constants.ExtVersionIdKey]); vid == "" || vid == "null" { + addPendingNull(dir, entry.Name) + } else { + dropPendingNull(dir, entry.Name) + } + } } } }) @@ -511,6 +719,12 @@ type ListingCursor struct { // something to find once a bucket has version history to leave behind. hideDeletedPrefixes bool probedEntries int + // retractEntry undoes the listing of a base-path null object once its .versions + // sibling reveals that the current version is a delete marker. + retractEntry func(dir, name string) + // resolvePendingNulls settles trailing null objects whose .versions sibling + // has not streamed yet before a page is declared full. + resolvePendingNulls func() error } // the prefix and marker may be in different directories @@ -624,6 +838,10 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. ctx, cancel := context.WithCancel(ctx) defer cancel() + // The marker this page started from, unlike marker below, which advances with + // each request window inside the page. + pageMarker := marker + // Entries that emit nothing (empty directories, the .uploads folder, the marker // echo) consume the request window without consuming maxKeys, so one window may // end before maxKeys is satisfied. Keep requesting from the last received entry @@ -677,9 +895,21 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. continue } - if cursor.maxKeys <= 0 { - cursor.isTruncated = true - break + // The .versions sibling of the key just emitted still decides that key's + // fate (metadata replacement or retraction) and consumes no quota of its + // own, so it must not be pushed past the page boundary. + versionsSiblingOfLast := cursor.hideDeletedPrefixes && entry.IsDirectory && + entry.Name == nextMarker+s3_constants.VersionsFolder + if cursor.maxKeys <= 0 && !versionsSiblingOfLast { + if cursor.resolvePendingNulls != nil { + if err = cursor.resolvePendingNulls(); err != nil { + return + } + } + if cursor.maxKeys <= 0 { + cursor.isTruncated = true + break + } } // Set nextMarker only when we have quota to process this entry @@ -713,6 +943,17 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. } // Extract object name from .versions directory name baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder) + // A page resuming from a marker inside the base key's extension + // region ("k.bak" sorts between "k" and "k.versions") means an + // earlier page already listed and settled the base null object; + // resolving this directory again would duplicate the key. Without + // a base object the key has not been listed yet, so it still + // resolves here. + if pageMarker != "" && baseObjectName < pageMarker { + if _, baseErr := s3a.getEntry(dir, baseObjectName); baseErr == nil { + continue + } + } // Construct full object path relative to bucket bucketFullPath := s3a.bucketDir(bucket) bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath) @@ -726,8 +967,13 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. // Use metadata from the already-fetched .versions directory entry if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil { eachEntryFn(dir, latestVersionEntry) - } else if !errors.Is(err, ErrDeleteMarker) { - // Log unexpected errors (delete markers are expected) + } else if errors.Is(err, ErrDeleteMarker) { + // The current version is a delete marker, so a null object listed + // for the base path just before this directory is stale. + if cursor.retractEntry != nil { + cursor.retractEntry(dir, baseObjectName) + } + } else { glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err) } continue @@ -831,6 +1077,10 @@ func (s3a *S3ApiServer) dirHoldsOnlyHiddenEntries(ctx context.Context, client fi sawEntry := false startFrom := "" + // A plain file is a null object that its .versions sibling, streaming later, + // may prove delete-marked; it stays pending until then. A pending file whose + // sibling window closes without one is a live key. + var pendingFiles []string for { request := &filer_pb.ListEntriesRequest{ Directory: dir, @@ -868,8 +1118,14 @@ func (s3a *S3ApiServer) dirHoldsOnlyHiddenEntries(ctx context.Context, client fi return false } + for _, pendingFile := range pendingFiles { + if entry.Name > pendingFile+s3_constants.VersionsFolder { + return false + } + } if !entry.IsDirectory { - return false + pendingFiles = append(pendingFiles, entry.Name) + continue } if entry.Name == s3_constants.MultipartUploadsFolder { continue @@ -883,6 +1139,15 @@ func (s3a *S3ApiServer) dirHoldsOnlyHiddenEntries(ctx context.Context, client fi // serving this list - and an unknown object keeps its prefix rather than // turning one listing into a version rescan per object. if isDeleteMarker, stamped := entry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker]; stamped && string(isDeleteMarker) == "true" { + // The marker also hides the null object the key left at the base path. + base := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder) + kept := pendingFiles[:0] + for _, pendingFile := range pendingFiles { + if pendingFile != base { + kept = append(kept, pendingFile) + } + } + pendingFiles = kept continue } return false @@ -896,7 +1161,7 @@ func (s3a *S3ApiServer) dirHoldsOnlyHiddenEntries(ctx context.Context, client fi } if entriesReceived < request.Limit { - return sawEntry + return sawEntry && len(pendingFiles) == 0 } } } diff --git a/weed/s3api/s3api_object_handlers_list_deleted_prefix_test.go b/weed/s3api/s3api_object_handlers_list_deleted_prefix_test.go index ce3b02c3a..e7ca1345f 100644 --- a/weed/s3api/s3api_object_handlers_list_deleted_prefix_test.go +++ b/weed/s3api/s3api_object_handlers_list_deleted_prefix_test.go @@ -101,6 +101,34 @@ func TestLivePrefixIsStillACommonPrefix(t *testing.T) { assert.Equal(t, []string{"20260102"}, seen, "only the date prefix with a current version is listed") } +// TestDeletedPrefixWithNullObjectIsNotACommonPrefix covers a key written before +// versioning was enabled: the delete marker leaves its null object at the base +// path, and that file must not keep the prefix alive. +func TestDeletedPrefixWithNullObjectIsNotACommonPrefix(t *testing.T) { + nullObject := &filer_pb.Entry{Name: "manifest", Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: 6}} + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {newDir("backup")}, + "/buckets/test/backup": {nullObject, deleteMarkedVersionsDir("manifest")}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Empty(t, seen, "the null object is shadowed by the delete marker, so nothing under the prefix is a key") + + // A second, unshadowed null object keeps the prefix live. + liveNull := &filer_pb.Entry{Name: "kept", Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: 6}} + client = &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {newDir("backup")}, + "/buckets/test/backup": {liveNull, nullObject, deleteMarkedVersionsDir("manifest")}, + }, + } + + seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Equal(t, []string{"backup"}, seen) +} + // TestDeletedPrefixGetsNoDirectoryMarker checks the trailing-slash probe. The empty // directory marker exists for directories created out of band; a directory that only // holds version history names nothing, so the probe answers empty like AWS does. diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 3e30b6774..95d6d9ed0 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -1474,6 +1474,9 @@ func (s3a *S3ApiServer) updateIsLatestFlagsForSuspendedVersioning(bucket, object delete(versionsEntry.Extended, s3_constants.ExtLatestVersionIdKey) delete(versionsEntry.Extended, s3_constants.ExtLatestVersionFileNameKey) clearCachedVersionMetadata(versionsEntry.Extended) + // Record that the null object is current explicitly: an absent pointer + // alone also looks like replication lag to a reader. + versionsEntry.Extended[s3_constants.ExtNullVersionIsLatestKey] = []byte("true") // Update the .versions directory entry err = s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) { @@ -1624,6 +1627,7 @@ func (s3a *S3ApiServer) updateLatestVersionInDirectory(bucket, object, versionId versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(versionId) versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(versionFileName) + delete(versionsEntry.Extended, s3_constants.ExtNullVersionIsLatestKey) // Cache list metadata for single-scan efficiency (avoids extra getEntry per object during list) setCachedListMetadata(versionsEntry, versionEntry) diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index ea0393c8b..ccd18d68f 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -46,6 +46,9 @@ func (s3a *S3ApiServer) latestPointerRecompute(bucket, object string, useInverte s3_constants.ExtLatestVersionOwnerKey: s3_constants.ExtAmzOwnerKey, s3_constants.ExtLatestVersionIsDeleteMarker: s3_constants.ExtDeleteMarkerKey, s3_constants.ExtLatestVersionStorageClassKey: s3_constants.AmzStorageClass, + // Version files never carry the null-current signal, so this mapping + // deletes a stale one from the pointer whenever it recomputes. + s3_constants.ExtNullVersionIsLatestKey: s3_constants.ExtNullVersionIsLatestKey, }, ExcludeName: excludeName, } diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 21d7194ca..62f447bd9 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -708,7 +708,7 @@ func (vc *versionCollector) processRegularFile(currentPath, entryPath string, en // Check if a .versions directory exists for this object versionsEntryName := entry.Name + s3_constants.VersionsFolder - _, versionsErr := vc.s3a.getEntry(currentPath, versionsEntryName) + versionsDirEntry, versionsErr := vc.s3a.getEntry(currentPath, versionsEntryName) if versionsErr == nil && !hasVersionMeta { // .versions exists but file has no version metadata - check for null version in .versions versions, err := vc.s3a.getObjectVersionList(vc.bucket, normalizedObjectKey) @@ -731,10 +731,26 @@ func (vc *versionCollector) processRegularFile(currentPath, entryPath string, en } vc.seenVersionIds[versionKey] = true + // A latest-version pointer on the .versions sibling names the current version + // and outranks a stale null-current signal a recompute may not have cleared + // yet. With no pointer, the explicit signal decides; with neither, the + // sibling may still hold replicated versions the lagging pointer has not + // caught up with, and the nullObjectWins rule decides. + isLatest := true + if versionsErr == nil { + if len(versionsDirEntry.Extended[s3_constants.ExtLatestVersionIdKey]) > 0 { + isLatest = false + } else if !nullVersionIsLatest(versionsDirEntry) { + if latestVersion, _, _, _, scanErr := vc.s3a.scanLatestVersionEntry(currentPath + "/" + versionsEntryName); scanErr == nil && latestVersion != nil && !nullObjectWins(entry, latestVersion) { + isLatest = false + } + } + } + versionEntry := &VersionEntry{ Key: normalizedObjectKey, VersionId: "null", - IsLatest: true, + IsLatest: isLatest, LastModified: time.Unix(entry.Attributes.Mtime, 0), ETag: vc.s3a.calculateETagFromChunks(entry.Chunks), Size: int64(entry.Attributes.FileSize), @@ -2029,7 +2045,7 @@ func (s3a *S3ApiServer) getLatestVersionEntryFromDirectoryEntry(bucket, object s // here). Indexing a nil Extended map is safe and yields !ok. latestVersionIdBytes, hasLatestVersionId := versionsDirEntry.Extended[s3_constants.ExtLatestVersionIdKey] if !hasLatestVersionId { - return s3a.recoverLatestListEntryByScan(bucket, normalizedObject) + return s3a.recoverLatestListEntryByScan(bucket, normalizedObject, nullVersionIsLatest(versionsDirEntry)) } // Check if this is a delete marker (should not be shown in regular list) @@ -2095,7 +2111,7 @@ func (s3a *S3ApiServer) getLatestVersionEntryFromDirectoryEntry(bucket, object s // Fallback: fetch version file if cached metadata not available (for older versions) latestVersionFileBytes, hasLatestVersionFile := versionsDirEntry.Extended[s3_constants.ExtLatestVersionFileNameKey] if !hasLatestVersionFile { - return s3a.recoverLatestListEntryByScan(bucket, normalizedObject) + return s3a.recoverLatestListEntryByScan(bucket, normalizedObject, nullVersionIsLatest(versionsDirEntry)) } latestVersionFile := string(latestVersionFileBytes) @@ -2128,6 +2144,26 @@ func (s3a *S3ApiServer) getLatestVersionEntryFromDirectoryEntry(bucket, object s return logicalEntry, nil } +// nullObjectWins decides, with no latest-version pointer and no null-is-latest +// signal to consult, whether the base-path null object or the newest scanned +// version is the current version: the newer mtime wins, and a tie goes to the +// version, since second-resolution mtimes cannot order same-second writes and +// an intentional null carries the ExtNullVersionIsLatestKey signal. The +// NoncurrentSinceNs demotion stamp is deliberately not consulted: promotions +// do not clear it, so it does not prove the null displaced the version. +func nullObjectWins(regular, latest *filer_pb.Entry) bool { + if latest == nil { + return true + } + return regular.GetAttributes().GetMtime() > latest.GetAttributes().GetMtime() +} + +// nullVersionIsLatest reports the explicit signal a suspended-versioning write +// leaves on the .versions directory when the null object is current. +func nullVersionIsLatest(versionsDirEntry *filer_pb.Entry) bool { + return versionsDirEntry != nil && string(versionsDirEntry.Extended[s3_constants.ExtNullVersionIsLatestKey]) == "true" +} + // recoverLatestListEntryByScan rebuilds an object's current-version list entry by // rescanning .versions/ when the cached latest-version pointer is missing on the // filer serving the list. This is the listing-path counterpart to the read path's @@ -2139,20 +2175,27 @@ func (s3a *S3ApiServer) getLatestVersionEntryFromDirectoryEntry(bucket, object s // a write per diverged object); convergence is handled on the write/replication // side. Returns ErrDeleteMarker when the current version is a delete marker // (excluded from a regular listing) and filer_pb.ErrNotFound when nothing remains. -func (s3a *S3ApiServer) recoverLatestListEntryByScan(bucket, normalizedObject string) (*filer_pb.Entry, error) { +func (s3a *S3ApiServer) recoverLatestListEntryByScan(bucket, normalizedObject string, nullIsLatest bool) (*filer_pb.Entry, error) { bucketDir := s3a.bucketDir(bucket) versionsDir := bucketDir + "/" + normalizedObject + s3_constants.VersionsFolder + // An absent pointer can mean a suspended-versioning write made the null + // object current (the write clears the pointer and leaves the explicit + // nullIsLatest signal), or that the pointer has not replicated to this + // filer while the version files have. + regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + if regularErr == nil && nullIsLatest { + return regularEntry, nil + } + latestEntry, latestVersionId, _, isDeleteMarker, err := s3a.scanLatestVersionEntry(versionsDir) if err != nil { return nil, err } + if regularErr == nil && nullObjectWins(regularEntry, latestEntry) { + return regularEntry, nil + } if latestEntry == nil { - // No version files remain. A pre-versioning / suspended "null" object at the - // base path is the current version if one exists. - if regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject); regularErr == nil { - return regularEntry, nil - } return nil, fmt.Errorf("%w: no current version for %s/%s", filer_pb.ErrNotFound, bucket, normalizedObject) } if isDeleteMarker {