s3: honor the version history of a directory marker (#10571)

* s3: stop listing a directory marker whose latest version is a delete marker

A directory marker is stored as the filer directory itself, so deleting the
key "dir/" writes its delete marker into dir/.versions while the directory
keeps its mime and stays a key object. Every listing kept reporting the key.
Consult that history before treating the entry as a key, and demote it in
memory when it is delete-marked so live children still hold the prefix.

Also skip the container's own .versions entry while listing inside it: the
suffix match read it as the history of a nested object named "", which
surfaces as a phantom dir/dir key as soon as a live directory version exists.

* s3: a directory marker with version history is not also the latest null version

The directory entry behind the key "dir/" is that key's null version, but
list-object-versions reported it with IsLatest hardcoded true. After a delete
the key came back twice, once as the delete marker and once as a null version,
both claiming to be latest. Read the pointer under the directory instead.

* s3: resolve directory markers through their version history on GET and HEAD

GET and HEAD short-circuit any trailing-slash key straight to the filer
directory, so a directory marker kept answering 200 after its delete marker
was written. Resolve the key from dir/.versions first when the bucket is
versioned: a delete-marked current version answers 404 with
x-amz-delete-marker, a named delete-marker version answers 405, and a key
with no history keeps today's directory-probe behavior untouched.

* s3: re-creating a directory marker retires its delete marker

PutObject on a trailing-slash key never looked at the bucket's versioning
state, so re-creating a marker after a delete left the latest-version pointer
on the delete marker and the key stayed invisible to every versioned read.
Point the key back at the directory entry, which is its null version, and drop
the null version .versions may still hold — the same two steps a suspended
write already takes, now shared.

* s3: fail a directory-marker request whose version history cannot be read

Every lookup of dir/.versions treated any error as "no history", so a filer
hiccup served the directory entry for a key whose current version may be a
delete marker, reported a null version as latest over one, and let a PUT
report success without retiring the delete marker it was meant to retire.
Only a confirmed absence takes the no-history path now.

* s3: cancel the directory-marker probe stream instead of abandoning it

The probe answers off the first entry and returns, leaving the ListEntries
stream open for the life of the parent context. Give it a context of its own.

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())
+}

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 "<dir>/" 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)

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 "<dir>/", 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 "<dir>/",
+// 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 "<dir>/" 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) {

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 "<dir>/", 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)
+}

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 {

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.