diff --git a/test/s3/versioning/s3_directory_marker_delete_test.go b/test/s3/versioning/s3_directory_marker_delete_test.go new file mode 100644 index 0000000000..0b93976ca8 --- /dev/null +++ b/test/s3/versioning/s3_directory_marker_delete_test.go @@ -0,0 +1,180 @@ +package s3api + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/smithy-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestDeletedDirectoryMarkerDisappears covers the rclone directory_markers flow: a key +// created with PutObject on "m2/" is deleted, and every current-version surface has to +// agree it is gone even though the filer directory that carries it survives. +func TestDeletedDirectoryMarkerDisappears(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, "m2/", "") + putObject(t, client, bucketName, "m2/f.txt", "hi") + + assert.Equal(t, []string{"m2/", "m2/f.txt"}, listKeys(t, client, bucketName, "")) + + deleteKey(t, client, bucketName, "m2/f.txt") + deleteKey(t, client, bucketName, "m2/") + + assert.Empty(t, listKeys(t, client, bucketName, ""), "the deleted marker is not a key") + assert.Empty(t, listPrefixes(t, client, bucketName, ""), "and it names no prefix") + assert.Empty(t, listKeys(t, client, bucketName, "m2/"), "prefix=m2/ answers empty") + + _, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String("m2/"), + }) + require.Error(t, err, "HEAD on a deleted directory marker must not answer 200") + _, err = client.GetObject(context.TODO(), &s3.GetObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String("m2/"), + }) + requireAPIError(t, err, "NoSuchKey") + + // The key appears once in the version listing, as its delete marker. + versions, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{ + Bucket: aws.String(bucketName), + Prefix: aws.String("m2/"), + }) + require.NoError(t, err) + latest := 0 + for _, v := range versions.Versions { + if *v.Key == "m2/" && *v.IsLatest { + latest++ + } + } + for _, m := range versions.DeleteMarkers { + if *m.Key == "m2/" && *m.IsLatest { + latest++ + } + } + assert.Equal(t, 1, latest, "exactly one version of m2/ is the latest") + + // Re-creating the marker retires the delete marker and keeps the history. + putObject(t, client, bucketName, "m2/", "") + assert.Equal(t, []string{"m2/"}, listKeys(t, client, bucketName, "")) + _, err = client.HeadObject(context.TODO(), &s3.HeadObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String("m2/"), + }) + require.NoError(t, err) + + versions, err = client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{ + Bucket: aws.String(bucketName), + Prefix: aws.String("m2/"), + }) + require.NoError(t, err) + assert.NotEmpty(t, versions.DeleteMarkers, "the delete marker stays in the history") + for _, m := range versions.DeleteMarkers { + if *m.Key == "m2/" { + assert.False(t, *m.IsLatest, "the delete marker is no longer current") + } + } +} + +// TestDeletedDirectoryMarkerKeepsItsSubtree pins the boundary: deleting the key "m2/" +// says nothing about the objects under it, which keep listing and keep the prefix. +func TestDeletedDirectoryMarkerKeepsItsSubtree(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, "m2/", "") + putObject(t, client, bucketName, "m2/keep.txt", "still here") + + deleteKey(t, client, bucketName, "m2/") + + assert.Equal(t, []string{"m2/keep.txt"}, listKeys(t, client, bucketName, "")) + assert.Equal(t, []string{"m2/"}, listPrefixes(t, client, bucketName, ""), + "a live child keeps the prefix even though the marker key is gone") + assert.Equal(t, []string{"m2/keep.txt"}, listKeys(t, client, bucketName, "m2/")) +} + +// TestDirectoryMarkerSurvivesWithoutVersioning guards the unversioned path, which has no +// history to consult and must keep answering as before. +func TestDirectoryMarkerSurvivesWithoutVersioning(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + putObject(t, client, bucketName, "m2/", "") + assert.Equal(t, []string{"m2/"}, listKeys(t, client, bucketName, "")) + + _, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String("m2/"), + }) + require.NoError(t, err) +} + +func listObjects(t *testing.T, client *s3.Client, bucketName, prefix, delimiter string) *s3.ListObjectsV2Output { + t.Helper() + input := &s3.ListObjectsV2Input{Bucket: aws.String(bucketName)} + if prefix != "" { + input.Prefix = aws.String(prefix) + } + if delimiter != "" { + input.Delimiter = aws.String(delimiter) + } + resp, err := client.ListObjectsV2(context.TODO(), input) + require.NoError(t, err) + return resp +} + +// listKeys lists without a delimiter, so a directory marker shows as the key it is +// instead of rolling up into its own CommonPrefix. +func listKeys(t *testing.T, client *s3.Client, bucketName, prefix string) []string { + t.Helper() + resp := listObjects(t, client, bucketName, prefix, "") + keys := make([]string, 0, len(resp.Contents)) + for _, c := range resp.Contents { + keys = append(keys, *c.Key) + } + return keys +} + +func listPrefixes(t *testing.T, client *s3.Client, bucketName, prefix string) []string { + t.Helper() + resp := listObjects(t, client, bucketName, prefix, "/") + prefixes := make([]string, 0, len(resp.CommonPrefixes)) + for _, p := range resp.CommonPrefixes { + prefixes = append(prefixes, *p.Prefix) + } + return prefixes +} + +func deleteKey(t *testing.T, client *s3.Client, bucketName, key string) { + t.Helper() + _, err := client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(key), + }) + require.NoError(t, err) +} + +func requireAPIError(t *testing.T, err error, code string) { + t.Helper() + require.Error(t, err) + var apiErr smithy.APIError + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, code, apiErr.ErrorCode()) +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index dee0fb0cb6..1795ba205e 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -463,9 +463,85 @@ func (s3a *S3ApiServer) serveDirectoryContent(w http.ResponseWriter, r *http.Req } } +// handleVersionedDirectoryObjectRequest answers GET/HEAD for the key "/" out of +// its version history, which for a directory marker lives inside the directory the key +// names. Without a history the key is whatever the directory entry says, which is the +// unversioned path the caller falls through to. +func (s3a *S3ApiServer) handleVersionedDirectoryObjectRequest(w http.ResponseWriter, r *http.Request, bucket, object, handlerName string) bool { + if !strings.HasSuffix(object, "/") { + return false + } + // A lookup that fails for any reason other than "not there" leaves the key's state + // unknown, and falling through would serve a directory whose current version may be + // a delete marker. Only a confirmed absence takes the unversioned path. + versioningConfigured, err := s3a.isVersioningConfigured(bucket) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + glog.Errorf("%s: versioning state of %s unknown: %v", handlerName, bucket, err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return true + } + if !versioningConfigured { + return false + } + + if versionId := r.URL.Query().Get("versionId"); versionId != "" { + entry, err := s3a.getSpecificObjectVersion(bucket, object, versionId) + if err != nil { + glog.V(2).Infof("%s: no version %s of directory object %s/%s: %v", handlerName, versionId, bucket, object, err) + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchVersion) + return true + } + w.Header().Set("x-amz-version-id", versionId) + if isDeleteMarkerEntry(entry) { + w.Header().Set(s3_constants.AmzDeleteMarker, "true") + s3err.WriteErrorResponse(w, r, s3err.ErrMethodNotAllowed) + return true + } + s3a.serveDirectoryContent(w, r, entry) + return true + } + + normalizedObject := s3_constants.NormalizeObjectKey(object) + if _, historyErr := s3a.getEntry(s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder); historyErr != nil { + if !errors.Is(historyErr, filer_pb.ErrNotFound) { + glog.Errorf("%s: version history of %s/%s unknown: %v", handlerName, bucket, object, historyErr) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return true + } + return false + } + // A failure here is reported the way the regular-object GET path reports it. + entry, err := s3a.getLatestObjectVersion(bucket, object) + if err != nil { + glog.V(2).Infof("%s: no current version of directory object %s/%s: %v", handlerName, bucket, object, err) + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return true + } + versionId := string(entry.Extended[s3_constants.ExtVersionIdKey]) + if versionId == "" { + versionId = "null" + } + w.Header().Set("x-amz-version-id", versionId) + if isDeleteMarkerEntry(entry) { + w.Header().Set(s3_constants.AmzDeleteMarker, "true") + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return true + } + s3a.serveDirectoryContent(w, r, entry) + return true +} + +func isDeleteMarkerEntry(entry *filer_pb.Entry) bool { + return entry != nil && string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" +} + // handleDirectoryObjectRequest is a helper function that handles directory object requests // for both GET and HEAD operations, eliminating code duplication func (s3a *S3ApiServer) handleDirectoryObjectRequest(w http.ResponseWriter, r *http.Request, bucket, object, handlerName string) bool { + if s3a.handleVersionedDirectoryObjectRequest(w, r, bucket, object, handlerName) { + return true + } + // Check if this is a directory object and handle it directly if dirEntry, isDirectoryObject, err := s3a.checkDirectoryObject(bucket, object); err != nil { glog.Errorf("%s: error checking directory object %s/%s: %v", handlerName, bucket, object, err) diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go index 47975c7a29..835e92ca73 100644 --- a/weed/s3api/s3api_object_handlers_list.go +++ b/weed/s3api/s3api_object_handlers_list.go @@ -717,6 +717,11 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. // Process .versions directories immediately to create logical versioned object entries // These directories are never traversed (we continue here), so each is only encountered once if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) { + if entry.Name == s3_constants.VersionsFolder { + // The history of the key "/", not of a child. The parent + // listing decides that key when it reaches the directory entry. + continue + } // Extract object name from .versions directory name baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder) // Construct full object path relative to bucket @@ -747,7 +752,7 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. if cursor.prefixEndsOnDelimiter { cursor.prefixEndsOnDelimiter = false } - isKeyObject := entry.IsDirectoryKeyObject() + isKeyObject := s3a.isLiveDirectoryKeyObject(ctx, client, entry, dir+"/"+entry.Name, cursor) if isKeyObject { // Directory key objects (created via PutObject with trailing "/") // must appear as regular keys in recursive listing mode. @@ -787,7 +792,7 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. return } // println("doListFilerEntries2 nextMarker", nextMarker) - } else if entry.IsDirectoryKeyObject() || !s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+entry.Name, cursor) { + } else if s3a.isLiveDirectoryKeyObject(ctx, client, entry, dir+"/"+entry.Name, cursor) || !s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+entry.Name, cursor) { eachEntryFn(dir, entry) } } else { @@ -807,6 +812,55 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb. } } +// isLiveDirectoryKeyObject reports whether entry still stands for the key "/", +// and strips entry in place when it does not. A directory marker deleted in a versioned +// bucket keeps its payload on disk — that payload is the key's null version — so the +// copy this listing holds is demoted instead, which is what stops the callback further +// down from reporting the entry as a key while the directory still serves as a prefix. +func (s3a *S3ApiServer) isLiveDirectoryKeyObject(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry, dirPath string, cursor *ListingCursor) bool { + if !entry.IsDirectoryKeyObject() { + return false + } + if !cursor.hideDeletedPrefixes || !directoryMarkerIsDeleted(ctx, client, dirPath) { + return true + } + clearDirectoryMarkerMetadata(entry) + return false +} + +// directoryMarkerIsDeleted reports whether the key "/" currently resolves to a +// delete marker. An explicit directory marker is the filer directory itself, so its +// version history is the container's own .versions entry, whose current-version stamp +// every pointer flip maintains — one lookup, no version scan. An unstamped history +// leaves the key visible, as it was before this check existed. +func directoryMarkerIsDeleted(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string) bool { + // The answer comes off the first entry, so cancel on return rather than draining. + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + stream, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{ + Directory: dirPath, + Prefix: s3_constants.VersionsFolder, + Limit: 1, + }) + if err != nil { + if !errors.Is(err, filer_pb.ErrNotFound) { + glog.V(1).Infof("directoryMarkerIsDeleted %s: %v", dirPath, err) + } + return false + } + for { + resp, recvErr := stream.Recv() + if recvErr != nil { + return false + } + if resp.Entry == nil || resp.Entry.Name != s3_constants.VersionsFolder { + continue + } + return string(resp.Entry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker]) == "true" + } +} + // hiddenProbePageSize is the window one probe request asks the filer for, and // hiddenProbeBudget caps how many entries a single list request may look at while // deciding which directories still stand for a prefix. @@ -893,7 +947,7 @@ func (s3a *S3ApiServer) dirHoldsOnlyHiddenEntries(ctx context.Context, client fi } return false } - if entry.IsDirectoryKeyObject() { + if s3a.isLiveDirectoryKeyObject(ctx, client, entry, dir+"/"+entry.Name, cursor) { return false } if !s3a.dirHoldsOnlyHiddenEntries(ctx, client, bucket, dir+"/"+entry.Name, cursor) { diff --git a/weed/s3api/s3api_object_handlers_list_directory_marker_test.go b/weed/s3api/s3api_object_handlers_list_directory_marker_test.go new file mode 100644 index 0000000000..ac20388c79 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_list_directory_marker_test.go @@ -0,0 +1,122 @@ +package s3api + +import ( + "strconv" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/stretchr/testify/assert" +) + +// directoryMarker builds the filer directory a PutObject on a trailing-slash key +// leaves behind. +func directoryMarker(name string) *filer_pb.Entry { + entry := newDir(name) + entry.Attributes.Mime = s3_constants.FolderMimeType + entry.Extended = map[string][]byte{s3_constants.ExtETagKey: []byte("d41d8cd98f00b204e9800998ecf8427e")} + return entry +} + +// ownVersionsDir builds the history of the key "/", which lives inside the +// directory it names rather than beside it. +func ownVersionsDir(deleted bool) *filer_pb.Entry { + now := time.Now().Unix() + extended := map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("v-dir"), + s3_constants.ExtLatestVersionMtimeKey: []byte(strconv.FormatInt(now, 10)), + s3_constants.ExtLatestVersionSizeKey: []byte("0"), + s3_constants.ExtLatestVersionETagKey: []byte(`"d41d8cd98f00b204e9800998ecf8427e"`), + s3_constants.ExtLatestVersionIsDeleteMarker: []byte(strconv.FormatBool(deleted)), + } + return &filer_pb.Entry{ + Name: s3_constants.VersionsFolder, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mtime: now}, + Extended: extended, + } +} + +// TestDeletedDirectoryMarkerIsNotListed covers the reported flow: PutObject on "m2/", +// DELETE on "m2/" writes a delete marker into m2/.versions, and the key must stop +// being reported even though the filer directory that carries it survives. +func TestDeletedDirectoryMarkerIsNotListed(t *testing.T) { + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {directoryMarker("m2")}, + "/buckets/test/m2": {ownVersionsDir(true)}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Empty(t, seen, "a delete-marked directory marker is not a key and names no prefix") + + seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Empty(t, seen, "the flat listing must not report it either") +} + +// TestLiveDirectoryMarkerIsListed pins the other half: a marker whose current version +// is real is a key, and the listing reports it from the directory that carries it. +func TestLiveDirectoryMarkerIsListed(t *testing.T) { + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {directoryMarker("m2")}, + "/buckets/test/m2": {ownVersionsDir(false)}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Equal(t, []string{"m2"}, seen) + + // A live history must not also surface as a phantom key named after its container. + seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Equal(t, []string{"m2"}, seen) +} + +// TestDeletedDirectoryMarkerKeepsLiveChildren guards the boundary between the two +// questions a directory answers: deleting the key "m2/" says nothing about the objects +// below it, which keep the prefix alive. +func TestDeletedDirectoryMarkerKeepsLiveChildren(t *testing.T) { + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {directoryMarker("m2")}, + "/buckets/test/m2": {ownVersionsDir(true), liveVersionsDir("keep.txt")}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Equal(t, []string{"m2"}, seen, "the directory still holds a live key, so it stays a prefix") + + seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, &ListingCursor{maxKeys: 1000, hideDeletedPrefixes: true}) + assert.Equal(t, []string{"keep.txt"}, seen, "only the surviving child is a key") +} + +// TestDeletedDirectoryMarkerIsNotADirectoryProbe checks the trailing-slash probe: the +// key was deleted, so prefix=m2/ answers empty instead of resurfacing the marker. +func TestDeletedDirectoryMarkerIsNotADirectoryProbe(t *testing.T) { + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {directoryMarker("m2")}, + "/buckets/test/m2": {ownVersionsDir(true)}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", prefix: "m2", delimiter: "/", bucket: "test"}, + &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true, hideDeletedPrefixes: true}) + assert.Empty(t, seen) +} + +// TestUnversionedBucketKeepsItsDirectoryMarkers pins the gate: without versioning there +// is no history to consult and no lookup to pay for. +func TestUnversionedBucketKeepsItsDirectoryMarkers(t *testing.T) { + client := &testFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{ + "/buckets/test": {directoryMarker("m2")}, + "/buckets/test/m2": {ownVersionsDir(true)}, + }, + } + + seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", delimiter: "/", bucket: "test"}, &ListingCursor{maxKeys: 1000}) + assert.Equal(t, []string{"m2"}, seen) +} diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 68363f8992..ddea6b8265 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -189,6 +189,24 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) s3err.WriteErrorResponse(w, r, filerErrorToS3Error(err)) return } + // A directory marker is stored as the directory itself, so it is the key's null + // version. Re-creating one after a delete has to retire that delete marker, + // otherwise the key stays invisible to every versioned read. Reporting the PUT + // as successful before that is known to have happened would hand the client a + // marker it cannot see, so an unreadable versioning state fails the request. + state, stateErr := s3a.getVersioningState(bucket) + if stateErr != nil && !errors.Is(stateErr, filer_pb.ErrNotFound) { + glog.Errorf("PutObjectHandler: versioning state of %s unknown: %v", bucket, stateErr) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + if state != "" { + if err := s3a.restoreNullVersion(bucket, strings.TrimPrefix(object, "/")); err != nil { + glog.Errorf("PutObjectHandler: failed to restore directory marker %s/%s: %v", bucket, object, err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + } setEtag(w, dirEtag) } else { // Get detailed versioning state for the bucket @@ -1293,39 +1311,9 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob glog.V(3).Infof("putSuspendedVersioningObject: START bucket=%s, object=%s, normalized=%s", bucket, object, normalizedObject) - bucketDir := s3a.bucketDir(bucket) - - // Check if there's an existing null version in .versions directory and delete it - // This ensures suspended versioning properly overwrites the null version as per S3 spec - // Note: We only delete null versions, NOT regular versions (those should be preserved) - versionsObjectPath := normalizedObject + s3_constants.VersionsFolder - versionsDir := bucketDir + "/" + versionsObjectPath - entries, _, err := s3a.list(versionsDir, "", "", false, 1000) - if err == nil { - // .versions directory exists - glog.V(3).Infof("putSuspendedVersioningObject: found %d entries in .versions for %s/%s", len(entries), bucket, object) - for _, entry := range entries { - if entry.Extended != nil { - if versionIdBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok { - versionId := string(versionIdBytes) - glog.V(3).Infof("putSuspendedVersioningObject: found version '%s' in .versions", versionId) - if versionId == "null" { - // Only delete null version - preserve real versioned entries - glog.V(3).Infof("putSuspendedVersioningObject: deleting null version from .versions") - err := s3a.rm(versionsDir, entry.Name, true, false) - if err != nil { - glog.Warningf("putSuspendedVersioningObject: failed to delete null version: %v", err) - } else { - glog.V(3).Infof("putSuspendedVersioningObject: successfully deleted null version") - } - break - } - } - } - } - } else { - glog.V(3).Infof("putSuspendedVersioningObject: no .versions directory for %s/%s", bucket, object) - } + // The null version now lives at the regular path, so any null version recorded in + // .versions is stale (S3 has the suspended write overwrite it, not accumulate). + s3a.removeNullVersionFile(bucket, normalizedObject) filePath := s3a.toFilerPath(bucket, normalizedObject) @@ -1413,6 +1401,42 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob return etag, s3err.ErrNone, sseMetadata } +// restoreNullVersion makes the object at the regular path the current version again: +// it drops a stale null version from .versions and clears the latest-version pointer, +// which is the recorded way of saying "the null object is current". The clearing helper +// is named for suspended versioning but does exactly this, and an enabled bucket needs +// the same thing when a directory marker is re-created over its delete marker. +func (s3a *S3ApiServer) restoreNullVersion(bucket, object string) error { + if _, err := s3a.getEntry(s3a.bucketDir(bucket), object+s3_constants.VersionsFolder); err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return nil // no history, so the object at the regular path is already current + } + return fmt.Errorf("read version history of %s/%s: %w", bucket, object, err) + } + s3a.removeNullVersionFile(bucket, object) + return s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, object) +} + +// removeNullVersionFile deletes the "null" version file from an object's .versions +// directory, leaving real versions alone. Best-effort: a leftover null version is +// superseded by the object at the regular path on the next read. +func (s3a *S3ApiServer) removeNullVersionFile(bucket, object string) { + versionsDir := s3a.bucketDir(bucket) + "/" + object + s3_constants.VersionsFolder + entries, _, err := s3a.list(versionsDir, "", "", false, 1000) + if err != nil { + return + } + for _, entry := range entries { + if string(entry.Extended[s3_constants.ExtVersionIdKey]) != "null" { + continue + } + if rmErr := s3a.rm(versionsDir, entry.Name, true, false); rmErr != nil { + glog.Warningf("removeNullVersionFile: %s/%s: %v", bucket, object, rmErr) + } + return + } +} + // updateIsLatestFlagsForSuspendedVersioning sets IsLatest=false on all existing versions/delete markers // when a new "null" version becomes the latest during suspended versioning func (s3a *S3ApiServer) updateIsLatestFlagsForSuspendedVersioning(bucket, object string) error { diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 21d7194ca5..2843e614dd 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -649,8 +649,9 @@ func (vc *versionCollector) processVersionsDirectory(entryPath string) error { return nil } -// processExplicitDirectory handles an explicit S3 directory object -func (vc *versionCollector) processExplicitDirectory(entryPath string, entry *filer_pb.Entry) { +// processExplicitDirectory handles an explicit S3 directory object. containerPath is +// the directory's own filer path, which is where the key's version history lives. +func (vc *versionCollector) processExplicitDirectory(entryPath, containerPath string, entry *filer_pb.Entry) error { directoryKey := entryPath if !strings.HasSuffix(directoryKey, "/") { directoryKey += "/" @@ -662,18 +663,33 @@ func (vc *versionCollector) processExplicitDirectory(entryPath string, entry *fi // this mirrors ListObjectsV2 and AWS, and stops clients like Veeam that // reject unexpected keys in a listing from aborting. if !strings.HasPrefix(directoryKey, vc.prefix) { - return + return nil } // Skip directories at or before keyMarker if vc.keyMarker != "" && directoryKey <= vc.keyMarker { - return + return nil + } + + // The directory entry is this key's null version. A history under it names the + // current version, so the null is only latest when there is no pointer to follow. + // A history we cannot read leaves that unknown, and claiming latest would hide a + // current delete marker, so the listing fails instead of guessing. + isLatest := true + versionsEntry, err := vc.s3a.getEntry(containerPath, s3_constants.VersionsFolder) + switch { + case err == nil: + if _, hasPointer := versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey]; hasPointer { + isLatest = false + } + case !errors.Is(err, filer_pb.ErrNotFound): + return fmt.Errorf("read version history of %s: %w", containerPath, err) } versionEntry := &VersionEntry{ Key: directoryKey, VersionId: "null", - IsLatest: true, + IsLatest: isLatest, LastModified: time.Unix(entry.Attributes.Mtime, 0), ETag: "\"d41d8cd98f00b204e9800998ecf8427e\"", // Empty content ETag Size: 0, @@ -681,6 +697,7 @@ func (vc *versionCollector) processExplicitDirectory(entryPath string, entry *fi StorageClass: StorageClass(vc.s3a.getStorageClassFromExtended(entry.Extended)), } *vc.allVersions = append(*vc.allVersions, versionEntry) + return nil } // processRegularFile handles a regular file entry (pre-versioning or suspended-versioning object) @@ -870,7 +887,9 @@ func (vc *versionCollector) processDirectory(currentPath, entryPath string, entr // an SDK PutObject of "dir/" carries a default Content-Type, so the two // listings must agree on what counts as a directory key. if entry.IsDirectoryKeyObject() { - vc.processExplicitDirectory(entryPath, entry) + if err := vc.processExplicitDirectory(entryPath, path.Join(currentPath, entry.Name), entry); err != nil { + return err + } } // Skip entire subdirectory if all keys within it are before the keyMarker.