S3: let a key that is a prefix of other keys be an object (#10912)

* filer: keep the sentinel when CreateEntry reports an update failure

CreateEntry flattened the error UpdateEntry wraps, so errors.Is stopped
matching and ErrExistingIsDirectory and ErrExistingIsFile never reached
the S3 mapper, which answered a retryable 500 instead.

* s3: let a key that is a prefix of other keys be an object

S3 keys are flat, so "a/b" and "a/b/c" are independent objects that
coexist in either write order. The filer stores a key as a path, so one
of them has to live on the directory the other is nested under.

Writing the nested key first refused the prefix key outright. Writing it
second promoted the file to a directory, which kept its data but lost the
key: an empty object left nothing to recognise it by and disappeared, and
one with data listed under a trailing slash it never had.

Mark the directory that carries such a key, and write the object onto it
when the path is already a directory. The mark makes an empty prefix
object visible to listings and readable by GET and HEAD, keeps the empty
folder cleaner off it, and lists it under the key it was written with.
Deleting the key strips the mark back off along with the data.

* filer: keep a TTL off a directory that stands for an object

An expired entry is deleted a row at a time, so expiring a directory
removes it and leaves everything under it unreachable. Promoting a file
to a directory carried its TTL across, and a promoted file is exactly the
one that has keys nested under it.

Drop the TTL on promotion, and leave one an older build wrote alone. The
lifecycle worker still expires the object, through the delete that leaves
the directory behind.

* s3: delete the null version of a key other keys are nested under

The routed delete cannot remove an entry that other keys live under, and
answered a retryable 500 rather than falling back to the lock path the
unversioned delete already falls back to. That path then looked the entry
up under the bucket with the whole key as its name, so the demote wrote it
back one directory too high and failed as not found.

Fall back on any non-precondition error, and split the key before deleting
it. Trailing-slash directory markers with children reach the same delete.

* filer: keep the sentinel when MkFile and Mkdir report a create failure

Same flattening one layer out: every mkFile caller lost the sentinel, so
a CopyObject onto a key that other keys are nested under answered a
retryable 500 where a PutObject of the same key answers 409.

* s3: copy and rename a key that other keys are nested under

Such a key is stored on the directory those keys live in, and copy and
rename both refused it: the source lookup maps every directory entry to
NoSuchKey, so a key a plain GET serves could not be copied or moved, and
the destination side refused it as a directory conflict.

The source is read through a view of the entry as the object it names.
The destination is written the way a PutObject of that key writes it. A
rename at either end copies the object's own data across and strips it off
the source key rather than going through AtomicRenameEntry, which moves a
directory by moving everything under it - the nested keys are not part of
what is being renamed.

.github/workflows/s3-example-integration-tests.yml

@@ -64,6 +64,14 @@ jobs:
           echo "=== Running S3 Empty Directory Marker Tests ==="
           go test -v -timeout=180s -run TestS3ListObjectsEmptyDirectoryMarkers ./...
 
+      - name: Run S3 Prefix Object Tests
+        timeout-minutes: 15
+        working-directory: test/s3/normal
+        run: |
+          set -x
+          echo "=== Running S3 Prefix Object Tests ==="
+          go test -v -timeout=180s -run TestS3PrefixObjectKeys ./...
+
       - name: Run IAM Integration Tests
         timeout-minutes: 15
         working-directory: test/s3/normal

test/s3/copying/s3_rename_test.go

@@ -156,8 +156,9 @@ func TestRenameObjectSourceIfMatch(t *testing.T) {
 	assert.True(t, objectExists(t, client, bucketName, "target.txt"))
 }
 
-// TestRenameObjectOntoDirectory: a key that already holds other objects is a
-// directory, and an object must not be allowed to replace one.
+// TestRenameObjectOntoDirectory: S3 keys are flat, so a key that other keys are
+// nested under is still a key of its own. The rename writes it without disturbing
+// them - it does not replace the directory, it stores the object on it.
 func TestRenameObjectOntoDirectory(t *testing.T) {
 	client := getS3Client(t)
 	bucketName := getNewBucketName()
@@ -172,9 +173,10 @@ func TestRenameObjectOntoDirectory(t *testing.T) {
 		Key:          aws.String("target"),
 		RenameSource: aws.String(createRenameSource("source.txt")),
 	})
-	requireRenameStatus(t, err, 409)
-	assert.True(t, objectExists(t, client, bucketName, "source.txt"))
-	assert.True(t, objectExists(t, client, bucketName, "target/child.txt"))
+	require.NoError(t, err)
+	assert.False(t, objectExists(t, client, bucketName, "source.txt"))
+	assert.Equal(t, "content", getObjectBody(t, getObject(t, client, bucketName, "target")))
+	assert.Equal(t, "child", getObjectBody(t, getObject(t, client, bucketName, "target/child.txt")))
 }
 
 // TestRenameObjectDirectorySource: a directory can be named without a trailing

test/s3/normal/s3_prefix_object_test.go

@@ -0,0 +1,546 @@
+package example
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	v1credentials "github.com/aws/aws-sdk-go/aws/credentials"
+	v1signer "github.com/aws/aws-sdk-go/aws/signer/v4"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestS3PrefixObjectKeys covers keys that are a strict prefix of other keys: S3's
+// namespace is flat, so "collision/foo" and "collision/foo/bar" are independent
+// objects that coexist in either write order.
+func TestS3PrefixObjectKeys(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	cluster, err := startMiniCluster(t)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	put := func(t *testing.T, bucket, key string, body []byte) {
+		t.Helper()
+		_, err := cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+			Body:   bytes.NewReader(body),
+		})
+		require.NoError(t, err, "put %s", key)
+	}
+	// read checks both paths a client reaches an object by, since a directory entry
+	// carrying an object is served by neither the directory nor the plain object path
+	// alone.
+	read := func(t *testing.T, bucket, key string, want []byte) {
+		t.Helper()
+		head, err := cluster.s3Client.HeadObject(&s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.NoError(t, err, "head %s", key)
+		assert.Equal(t, int64(len(want)), aws.Int64Value(head.ContentLength), "head %s", key)
+
+		get, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.NoError(t, err, "get %s", key)
+		defer get.Body.Close()
+		got, err := io.ReadAll(get.Body)
+		require.NoError(t, err, "read %s", key)
+		assert.Equal(t, want, got, "get %s", key)
+	}
+	// gone checks the key answers as absent rather than lingering on a directory
+	// entry that outlived the object.
+	gone := func(t *testing.T, bucket, key string) {
+		t.Helper()
+		_, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		var missing awserr.RequestFailure
+		require.ErrorAs(t, err, &missing, "get %s", key)
+		assert.Equal(t, http.StatusNotFound, missing.StatusCode(), "get %s", key)
+
+		_, err = cluster.s3Client.HeadObject(&s3.HeadObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String(key),
+		})
+		require.ErrorAs(t, err, &missing, "head %s", key)
+		assert.Equal(t, http.StatusNotFound, missing.StatusCode(), "head %s", key)
+	}
+	listKeys := func(t *testing.T, bucket string) []string {
+		t.Helper()
+		resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+		keys := collectKeys(resp.Contents)
+		sort.Strings(keys)
+		return keys
+	}
+
+	body := []byte("prefix object")
+	// Distinct bodies, so a read that resolves to the wrong entry cannot pass.
+	nested := []byte("nested under the prefix object")
+
+	// The reported order: the nested key is written first, so the prefix key has to
+	// land on a path the filer already holds a directory at.
+	t.Run("ChildFirst", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-child-first-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// The opposite order used to keep the prefix key's data but hide the key.
+	t.Run("PrefixFirst", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-first-")
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// An empty object leaves no chunks, content or mime behind, so it is the case a
+	// promoted directory carries no other trace of.
+	t.Run("EmptyObject", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-empty-")
+		put(t, bucket, "a/foo/bar", nil)
+		put(t, bucket, "a/foo", nil)
+		put(t, bucket, "b/foo", nil)
+		put(t, bucket, "b/foo/bar", nil)
+
+		assert.Equal(t, []string{"a/foo", "a/foo/bar", "b/foo", "b/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "a/foo", []byte{})
+		read(t, bucket, "b/foo", []byte{})
+	})
+
+	// The key has no trailing slash, and the keys nested under it still roll up into
+	// their own CommonPrefix.
+	t.Run("Delimiter", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-delimiter-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/other", body)
+
+		resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+			Bucket:    aws.String(bucket),
+			Prefix:    aws.String("collision/"),
+			Delimiter: aws.String("/"),
+		})
+		require.NoError(t, err)
+
+		keys := collectKeys(resp.Contents)
+		sort.Strings(keys)
+		assert.Equal(t, []string{"collision/foo", "collision/other"}, keys)
+		assert.Equal(t, []string{"collision/foo/"}, collectPrefixes(resp.CommonPrefixes))
+
+		// Listing the prefix itself names only what is under it.
+		resp, err = cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+			Bucket:    aws.String(bucket),
+			Prefix:    aws.String("collision/foo/"),
+			Delimiter: aws.String("/"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo/bar"}, collectKeys(resp.Contents))
+		assert.Empty(t, collectPrefixes(resp.CommonPrefixes))
+	})
+
+	// The key and the CommonPrefix its nested keys fold into come off one filer
+	// entry, so a page boundary must not drop either of them.
+	t.Run("Paged", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-paged-")
+		for _, key := range []string{"foo", "foo/bar", "foobar", "other", "zed", "zed/a"} {
+			put(t, bucket, key, body)
+		}
+
+		for _, maxKeys := range []int64{1, 2, 3, 4, 5} {
+			var keys, prefixes []string
+			var token *string
+			for page := 0; page < 12; page++ {
+				resp, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+					Bucket:            aws.String(bucket),
+					Delimiter:         aws.String("/"),
+					MaxKeys:           aws.Int64(maxKeys),
+					ContinuationToken: token,
+				})
+				require.NoError(t, err)
+				// One entry over the budget is the documented same-entry exception;
+				// anything more means the unsigned budget wrapped.
+				assert.LessOrEqual(t, int64(len(resp.Contents)+len(resp.CommonPrefixes)), maxKeys+1,
+					"maxKeys=%d page %d", maxKeys, page)
+				keys = append(keys, collectKeys(resp.Contents)...)
+				prefixes = append(prefixes, collectPrefixes(resp.CommonPrefixes)...)
+				if !aws.BoolValue(resp.IsTruncated) {
+					token = nil
+					break
+				}
+				token = resp.NextContinuationToken
+				require.NotNil(t, token, "a truncated page must name where to resume")
+			}
+			require.Nil(t, token, "maxKeys=%d did not finish", maxKeys)
+			assert.Equal(t, []string{"foo", "foobar", "other", "zed"}, keys, "maxKeys=%d", maxKeys)
+			assert.Equal(t, []string{"foo/", "zed/"}, prefixes, "maxKeys=%d", maxKeys)
+		}
+	})
+
+	// Versioning reaches a prefix object from two directions: a suspended bucket
+	// writes the null version at the key's own path, and a bucket versioned later
+	// finds one already sitting there. Both leave a key that is a directory with
+	// version history beside it.
+	t.Run("Versioned", func(t *testing.T) {
+		setVersioning := func(t *testing.T, bucket, status string) {
+			t.Helper()
+			_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+				Bucket:                  aws.String(bucket),
+				VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String(status)},
+			})
+			require.NoError(t, err)
+		}
+
+		// Both write orders, in a bucket that is versioned and in one where versioning
+		// was suspended - the suspended one is the case that writes at the key's path.
+		for _, state := range []string{"Enabled", "Suspended"} {
+			bucket := createTestBucket(t, cluster, "test-prefix-"+strings.ToLower(state)+"-")
+			setVersioning(t, bucket, "Enabled")
+			if state == "Suspended" {
+				setVersioning(t, bucket, "Suspended")
+			}
+			put(t, bucket, "child/foo/bar", nested)
+			put(t, bucket, "child/foo", body)
+			put(t, bucket, "prefix/foo", body)
+			put(t, bucket, "prefix/foo/bar", nested)
+
+			assert.Equal(t, []string{"child/foo", "child/foo/bar", "prefix/foo", "prefix/foo/bar"},
+				listKeys(t, bucket), state)
+			for _, key := range []string{"child/foo", "prefix/foo"} {
+				read(t, bucket, key, body)
+			}
+			for _, key := range []string{"child/foo/bar", "prefix/foo/bar"} {
+				read(t, bucket, key, nested)
+			}
+		}
+
+		// A prefix object written before versioning is the key's null version. Removing
+		// that version by id must not take the keys nested under it with it.
+		bucket := createTestBucket(t, cluster, "test-prefix-nullversion-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		setVersioning(t, bucket, "Enabled")
+		newer := []byte("written after versioning was enabled")
+		versioned, err := cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo"),
+			Body:   bytes.NewReader(newer),
+		})
+		require.NoError(t, err)
+
+		for _, v := range []struct {
+			id   string
+			want []byte
+		}{{"null", body}, {aws.StringValue(versioned.VersionId), newer}} {
+			got, err := cluster.s3Client.GetObject(&s3.GetObjectInput{
+				Bucket:    aws.String(bucket),
+				Key:       aws.String("collision/foo"),
+				VersionId: aws.String(v.id),
+			})
+			require.NoError(t, err, "get version %s", v.id)
+			body, err := io.ReadAll(got.Body)
+			require.NoError(t, err)
+			got.Body.Close()
+			assert.Equal(t, v.want, body, "get version %s", v.id)
+		}
+
+		_, err = cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket:    aws.String(bucket),
+			Key:       aws.String("collision/foo"),
+			VersionId: aws.String("null"),
+		})
+		require.NoError(t, err, "the null version sits on a directory other keys live in")
+
+		read(t, bucket, "collision/foo", newer)
+		read(t, bucket, "collision/foo/bar", nested)
+		remaining, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{
+			Bucket: aws.String(bucket),
+			Prefix: aws.String("collision/foo"),
+		})
+		require.NoError(t, err)
+		for _, v := range remaining.Versions {
+			if aws.StringValue(v.Key) != "collision/foo" {
+				// collision/foo/bar predates versioning too, and keeps its null version.
+				continue
+			}
+			assert.NotEqual(t, "null", aws.StringValue(v.VersionId), "the null version was deleted")
+		}
+	})
+
+	// The two listings walk the tree differently, and a prefix object is the entry
+	// they disagree about: it is a directory the version listing descends through and
+	// a key at the same time. They have to name the same keys and the same prefixes.
+	t.Run("VersionListingMatchesObjectListing", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-parity-")
+		for _, key := range []string{"foo", "foo/bar", "other", "a/foo", "a/foo/bar", "a/z"} {
+			put(t, bucket, key, body)
+		}
+
+		for _, q := range []struct{ prefix, delimiter string }{
+			{"", ""},
+			{"foo/", ""},
+			{"a/", ""},
+			{"a/foo/", ""},
+			{"", "/"},
+			{"a/", "/"},
+		} {
+			name := "prefix=" + q.prefix + " delimiter=" + q.delimiter
+
+			objects, err := cluster.s3Client.ListObjectsV2(&s3.ListObjectsV2Input{
+				Bucket:    aws.String(bucket),
+				Prefix:    aws.String(q.prefix),
+				Delimiter: aws.String(q.delimiter),
+			})
+			require.NoError(t, err, name)
+
+			versions, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{
+				Bucket:    aws.String(bucket),
+				Prefix:    aws.String(q.prefix),
+				Delimiter: aws.String(q.delimiter),
+			})
+			require.NoError(t, err, name)
+
+			versionKeys := make([]string, 0, len(versions.Versions))
+			for _, v := range versions.Versions {
+				versionKeys = append(versionKeys, aws.StringValue(v.Key))
+			}
+			versionPrefixes := make([]string, 0, len(versions.CommonPrefixes))
+			for _, p := range versions.CommonPrefixes {
+				versionPrefixes = append(versionPrefixes, aws.StringValue(p.Prefix))
+			}
+			sort.Strings(versionKeys)
+			sort.Strings(versionPrefixes)
+
+			objectKeys := collectKeys(objects.Contents)
+			objectPrefixes := collectPrefixes(objects.CommonPrefixes)
+			sort.Strings(objectKeys)
+			sort.Strings(objectPrefixes)
+
+			assert.Equal(t, objectKeys, versionKeys, "keys, %s", name)
+			assert.Equal(t, objectPrefixes, versionPrefixes, "prefixes, %s", name)
+		}
+	})
+
+	// A prefix object written before versioning is the key's null version, so the
+	// version written after it has to take the latest flag off it.
+	t.Run("VersionedAfterPrefixObject", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-versioned-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+		_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Enabled")},
+		})
+		require.NoError(t, err)
+		newer := []byte("written after versioning was enabled")
+		put(t, bucket, "collision/foo", newer)
+
+		resp, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+
+		latest := map[string]int{}
+		var nullSize int64 = -1
+		for _, v := range resp.Versions {
+			if aws.BoolValue(v.IsLatest) {
+				latest[aws.StringValue(v.Key)]++
+			}
+			if aws.StringValue(v.Key) == "collision/foo" && aws.StringValue(v.VersionId) == "null" {
+				nullSize = aws.Int64Value(v.Size)
+				assert.False(t, aws.BoolValue(v.IsLatest), "the newer version is the latest one")
+			}
+		}
+		assert.Equal(t, 1, latest["collision/foo"], "exactly one version of a key is the latest")
+		assert.Equal(t, int64(len(body)), nullSize, "the null version keeps the prefix object's size")
+		read(t, bucket, "collision/foo", newer)
+		read(t, bucket, "collision/foo/bar", nested)
+	})
+
+	// A key that other keys are nested under is a copy source and a copy destination
+	// like any other. The keys nested under either end are not part of the copy.
+	t.Run("Copy", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-copy-")
+		copyObject := func(t *testing.T, src, dst string) {
+			t.Helper()
+			_, err := cluster.s3Client.CopyObject(&s3.CopyObjectInput{
+				Bucket:     aws.String(bucket),
+				Key:        aws.String(dst),
+				CopySource: aws.String(bucket + "/" + src),
+			})
+			require.NoError(t, err, "copy %s to %s", src, dst)
+		}
+
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		// Out of a prefix object, into a key of its own.
+		copyObject(t, "collision/foo", "plain")
+		read(t, bucket, "plain", body)
+		read(t, bucket, "collision/foo", body)
+		read(t, bucket, "collision/foo/bar", nested)
+
+		// Into a key that other keys are nested under.
+		put(t, bucket, "target/child", nested)
+		copyObject(t, "plain", "target")
+		read(t, bucket, "target", body)
+		read(t, bucket, "target/child", nested)
+
+		// And between two of them.
+		put(t, bucket, "other", []byte("copied between prefix keys"))
+		copyObject(t, "other", "collision/foo")
+		read(t, bucket, "collision/foo", []byte("copied between prefix keys"))
+		read(t, bucket, "collision/foo/bar", nested)
+
+		assert.Equal(t, []string{"collision/foo", "collision/foo/bar", "other", "plain", "target", "target/child"},
+			listKeys(t, bucket))
+	})
+
+	// Rename moves the object off the key without moving the keys nested under it,
+	// which is not what the filer's atomic rename of a directory would do.
+	t.Run("Rename", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-rename-")
+		renameObject := func(t *testing.T, src, dst string) {
+			t.Helper()
+			req, _ := http.NewRequest(http.MethodPut, cluster.s3Endpoint+"/"+bucket+"/"+dst+"?renameObject=", nil)
+			req.Header.Set("x-amz-rename-source", "/"+bucket+"/"+src)
+			signer := v1signer.NewSigner(v1credentials.NewStaticCredentials(testAccessKey, testSecretKey, ""))
+			_, err := signer.Sign(req, nil, "s3", testRegion, time.Now())
+			require.NoError(t, err)
+
+			resp, err := (&http.Client{Timeout: 20 * time.Second}).Do(req)
+			require.NoError(t, err, "rename %s to %s", src, dst)
+			defer resp.Body.Close()
+			io.Copy(io.Discard, resp.Body)
+			require.Equal(t, http.StatusOK, resp.StatusCode, "rename %s to %s", src, dst)
+		}
+
+		put(t, bucket, "collision/foo", body)
+		put(t, bucket, "collision/foo/bar", nested)
+
+		// Off a prefix object: the key goes, the keys under it stay.
+		renameObject(t, "collision/foo", "moved")
+		read(t, bucket, "moved", body)
+		read(t, bucket, "collision/foo/bar", nested)
+		gone(t, bucket, "collision/foo")
+
+		// Onto a key other keys are nested under.
+		put(t, bucket, "target/child", nested)
+		renameObject(t, "moved", "target")
+		read(t, bucket, "target", body)
+		read(t, bucket, "target/child", nested)
+		gone(t, bucket, "moved")
+
+		assert.Equal(t, []string{"collision/foo/bar", "target", "target/child"}, listKeys(t, bucket))
+	})
+
+	// A directory SeaweedFS keeps its own state in is not a prefix a key can be
+	// stored on: the object would replace that state with its own.
+	t.Run("ReservedDirectory", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-reserved-")
+		_, err := cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Enabled")},
+		})
+		require.NoError(t, err)
+		put(t, bucket, "foo", body)
+		_, err = cluster.s3Client.PutBucketVersioning(&s3.PutBucketVersioningInput{
+			Bucket:                  aws.String(bucket),
+			VersioningConfiguration: &s3.VersioningConfiguration{Status: aws.String("Suspended")},
+		})
+		require.NoError(t, err)
+
+		_, err = cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("foo.versions"),
+			Body:   bytes.NewReader(body),
+		})
+		// Not just any error: a permanently impossible write must not come back as a
+		// 500 the SDK retries.
+		var refused awserr.RequestFailure
+		require.ErrorAs(t, err, &refused, "the version history of foo is not a prefix of foo.versions")
+		assert.Equal(t, http.StatusConflict, refused.StatusCode())
+		assert.Equal(t, "ExistingObjectIsDirectory", refused.Code())
+
+		versions, err := cluster.s3Client.ListObjectVersions(&s3.ListObjectVersionsInput{Bucket: aws.String(bucket)})
+		require.NoError(t, err)
+		require.Len(t, versions.Versions, 1)
+		assert.Equal(t, "foo", aws.StringValue(versions.Versions[0].Key))
+		read(t, bucket, "foo", body)
+
+		// The multipart staging folder is the other one, and an in-flight upload has
+		// to survive the attempt.
+		staging := createTestBucket(t, cluster, "test-prefix-uploads-")
+		created, err := cluster.s3Client.CreateMultipartUpload(&s3.CreateMultipartUploadInput{
+			Bucket: aws.String(staging),
+			Key:    aws.String("mp.bin"),
+		})
+		require.NoError(t, err)
+
+		_, err = cluster.s3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(staging),
+			Key:    aws.String(".uploads"),
+			Body:   bytes.NewReader(body),
+		})
+		require.ErrorAs(t, err, &refused, "the multipart staging folder is not a prefix of .uploads")
+		assert.Equal(t, http.StatusConflict, refused.StatusCode())
+		assert.Equal(t, "ExistingObjectIsDirectory", refused.Code())
+
+		uploads, err := cluster.s3Client.ListMultipartUploads(&s3.ListMultipartUploadsInput{Bucket: aws.String(staging)})
+		require.NoError(t, err)
+		require.Len(t, uploads.Uploads, 1)
+		assert.Equal(t, "mp.bin", aws.StringValue(uploads.Uploads[0].Key))
+		_, err = cluster.s3Client.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
+			Bucket:   aws.String(staging),
+			Key:      aws.String("mp.bin"),
+			UploadId: created.UploadId,
+		})
+		require.NoError(t, err)
+	})
+
+	// Either key can be deleted without touching the other.
+	t.Run("Delete", func(t *testing.T) {
+		bucket := createTestBucket(t, cluster, "test-prefix-delete-")
+		put(t, bucket, "collision/foo/bar", nested)
+		put(t, bucket, "collision/foo", body)
+
+		_, err := cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo/bar"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo/bar", nested)
+		gone(t, bucket, "collision/foo")
+
+		put(t, bucket, "collision/foo", body)
+		_, err = cluster.s3Client.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(bucket),
+			Key:    aws.String("collision/foo/bar"),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, []string{"collision/foo"}, listKeys(t, bucket))
+		read(t, bucket, "collision/foo", body)
+		gone(t, bucket, "collision/foo/bar")
+	})
+}

weed/filer/filer.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
@@ -299,8 +300,12 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry,
 		}
 		glog.V(4).InfofCtx(ctx, "UpdateEntry %s: old entry: %v", entry.FullPath, oldEntry.Name())
 		if err := f.UpdateEntry(ctx, oldEntry, entry); err != nil {
-			glog.ErrorfCtx(ctx, "update entry %s: %v", entry.FullPath, err)
-			return fmt.Errorf("update entry %s: %v", entry.FullPath, err)
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) || errors.Is(err, filer_pb.ErrExistingIsFile) {
+				glog.V(2).InfofCtx(ctx, "update entry %s: %v", entry.FullPath, err)
+			} else {
+				glog.ErrorfCtx(ctx, "update entry %s: %v", entry.FullPath, err)
+			}
+			return fmt.Errorf("update entry %s: %w", entry.FullPath, err)
 		}
 	}
 
@@ -388,6 +393,16 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di
 		// the original object data remains accessible.
 		glog.V(2).InfofCtx(ctx, "promoting %s from file to directory for %s", dirPath, entry.FullPath)
 		dirEntry.Attr.Mode |= os.ModeDir | 0111
+		// Expiring the entry now deletes the directory row and strands the keys under
+		// it, so the prefix object gives up its lazy TTL. The lifecycle worker still
+		// expires it, through the delete that leaves the directory behind.
+		dirEntry.Attr.TtlSec = 0
+		// An empty object leaves no chunks, content or mime behind, so without the
+		// mark the promotion would hide it.
+		if dirEntry.Extended == nil {
+			dirEntry.Extended = make(map[string][]byte)
+		}
+		dirEntry.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true")
 		if updateErr := f.Store.UpdateEntry(ctx, dirEntry); updateErr != nil {
 			return fmt.Errorf("promote %s to directory: %v", dirPath, updateErr)
 		}
@@ -481,12 +496,15 @@ func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry) (err er
 		} else {
 			f.ensureEntryInode(entry)
 		}
+		// A type conflict is reported through the sentinel, and callers act on it -
+		// an S3 write of a key other keys are nested under retries as a prefix object -
+		// so it is the caller's outcome that decides whether anything went wrong.
 		if oldEntry.IsDirectory() && !entry.IsDirectory() {
-			glog.ErrorfCtx(ctx, "existing %s is a directory", oldEntry.FullPath)
+			glog.V(2).InfofCtx(ctx, "existing %s is a directory", oldEntry.FullPath)
 			return fmt.Errorf("%s: %w", oldEntry.FullPath, filer_pb.ErrExistingIsDirectory)
 		}
 		if !oldEntry.IsDirectory() && entry.IsDirectory() {
-			glog.ErrorfCtx(ctx, "existing %s is a file", oldEntry.FullPath)
+			glog.V(2).InfofCtx(ctx, "existing %s is a file", oldEntry.FullPath)
 			return fmt.Errorf("%s: %w", oldEntry.FullPath, filer_pb.ErrExistingIsFile)
 		}
 	}
@@ -522,7 +540,9 @@ func (f *Filer) FindEntry(ctx context.Context, p util.FullPath) (entry *Entry, e
 		return Root, nil
 	}
 	entry, err = f.Store.FindEntry(ctx, p)
-	if entry != nil && entry.TtlSec > 0 {
+	// A directory is deleted here one row at a time, which would strand whatever is
+	// under it, so a TTL an older build left on one is not acted on.
+	if entry != nil && entry.TtlSec > 0 && !entry.IsDirectory() {
 		if entry.IsExpireS3Enabled() {
 			if entry.GetS3ExpireTime().Before(time.Now()) && !entry.IsS3Versioning() {
 				if delErr := f.doDeleteEntryMetaAndData(ctx, entry, true, false, nil); delErr != nil {
@@ -561,7 +581,7 @@ func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, sta
 			glog.Errorf("Context is done.")
 			return false, fmt.Errorf("context canceled: %w", ctx.Err())
 		default:
-			if entry.TtlSec > 0 {
+			if entry.TtlSec > 0 && !entry.IsDirectory() {
 				if entry.IsExpireS3Enabled() {
 					if entry.GetS3ExpireTime().Before(time.Now()) && !entry.IsS3Versioning() {
 						// Collect for deletion after iteration completes to avoid DB deadlock
@@ -722,5 +742,6 @@ func (f *Filer) IsDirectoryKeyObject(ctx context.Context, p util.FullPath) (bool
 		return false, nil
 	}
 	// Mirror filer_pb.Entry.IsDirectoryKeyObject so the cleaner keeps a promoted file's data.
-	return entry.IsDirectory() && (entry.Mime != "" || len(entry.GetChunks()) > 0 || len(entry.Content) > 0 || entry.IsInRemoteOnly()), nil
+	_, isPrefixObject := entry.Extended[s3_constants.SeaweedFSPrefixObject]
+	return entry.IsDirectory() && (entry.Mime != "" || len(entry.GetChunks()) > 0 || len(entry.Content) > 0 || entry.IsInRemoteOnly() || isPrefixObject), nil
 }

weed/filer/filer_prefix_object_test.go

@@ -0,0 +1,77 @@
+package filer
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPromoteToPrefixObject covers a key written before the keys nested under it:
+// the file becomes the directory they live in, and has to stay an object of its own.
+func TestPromoteToPrefixObject(t *testing.T) {
+	f, store := newTestFilerWithStubStore()
+	ctx := context.Background()
+
+	object := &Entry{
+		FullPath: util.FullPath("/buckets/bkt/a/foo"),
+		Attr: Attr{
+			Mode:   0o644,
+			Mime:   "text/plain",
+			TtlSec: 3600,
+		},
+		Chunks:   []*filer_pb.FileChunk{{FileId: "1,01", Size: 4}},
+		Extended: map[string][]byte{s3_constants.SeaweedFSExpiresS3: []byte("true")},
+	}
+	require.NoError(t, f.CreateEntry(ctx, object, nil, false, false, nil, false, f.MaxFilenameLength))
+
+	nested := &Entry{FullPath: util.FullPath("/buckets/bkt/a/foo/bar"), Attr: Attr{Mode: 0o644}}
+	require.NoError(t, f.CreateEntry(ctx, nested, nil, false, false, nil, false, f.MaxFilenameLength))
+
+	promoted, err := store.FindEntry(ctx, object.FullPath)
+	require.NoError(t, err)
+	require.NotNil(t, promoted)
+
+	assert.True(t, promoted.IsDirectory(), "the nested key needs a directory here")
+	assert.Equal(t, object.Chunks, promoted.Chunks, "the object's data stays on it")
+	assert.Contains(t, promoted.Extended, s3_constants.SeaweedFSPrefixObject)
+	// Expiring the entry deletes the directory row on its own and strands the keys
+	// under it, so the promotion gives up the lazy TTL.
+	assert.Zero(t, promoted.Attr.TtlSec)
+}
+
+// TestExpiredDirectoryIsNotDeletedOnRead pins the other half: a TTL an older build
+// left on a promoted directory must not take the keys under it with it.
+func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) {
+	f, store := newTestFilerWithStubStore()
+	ctx := context.Background()
+
+	dirPath := util.FullPath("/buckets/bkt/a/foo")
+	expired := time.Now().Add(-2 * time.Hour)
+	require.NoError(t, store.InsertEntry(ctx, &Entry{
+		FullPath: dirPath,
+		Attr: Attr{
+			Mode:   os.ModeDir | 0o755,
+			Crtime: expired,
+			Mtime:  expired,
+			TtlSec: 60,
+		},
+		Extended: map[string][]byte{s3_constants.SeaweedFSExpiresS3: []byte("true")},
+	}))
+	nested := &Entry{FullPath: dirPath + "/bar", Attr: Attr{Mode: 0o644}}
+	require.NoError(t, store.InsertEntry(ctx, nested))
+
+	found, err := f.FindEntry(ctx, dirPath)
+	require.NoError(t, err)
+	require.NotNil(t, found, "deleting it here would leave the nested key unreachable")
+
+	stillThere, err := store.FindEntry(ctx, nested.FullPath)
+	require.NoError(t, err)
+	require.NotNil(t, stillThere)
+}

weed/pb/filer_pb/filer_client.go

@@ -261,7 +261,7 @@ func DoMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath
 	glog.V(1).InfofCtx(ctx, "mkdir: %v", request)
 	if err := CreateEntry(ctx, client, request); err != nil {
 		glog.V(0).InfofCtx(ctx, "mkdir %v: %v", request, err)
-		return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
+		return fmt.Errorf("mkdir %s/%s: %w", parentDirectoryPath, dirName, err)
 	}
 
 	return nil
@@ -295,7 +295,7 @@ func MkFile(ctx context.Context, filerClient FilerClient, parentDirectoryPath st
 		glog.V(1).InfofCtx(ctx, "create file: %s/%s", parentDirectoryPath, fileName)
 		if err := CreateEntry(ctx, client, request); err != nil {
 			glog.V(0).InfofCtx(ctx, "create file %v:%v", request, err)
-			return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
+			return fmt.Errorf("create file %s/%s: %w", parentDirectoryPath, fileName, err)
 		}
 
 		return nil

weed/pb/filer_pb/filer_pb_helper.go

@@ -26,7 +26,30 @@ func (entry *Entry) IsDirectoryKeyObject() bool {
 	// Also true for a file promoted to a directory by a child write, which keeps its
 	// chunks/content, or its remote entry when the file was tiered to remote storage.
 	return entry.IsDirectory &&
-		((entry.Attributes != nil && entry.Attributes.Mime != "") || len(entry.GetChunks()) > 0 || len(entry.GetContent()) > 0 || entry.IsInRemoteOnly())
+		((entry.Attributes != nil && entry.Attributes.Mime != "") || len(entry.GetChunks()) > 0 || len(entry.GetContent()) > 0 || entry.IsInRemoteOnly() || entry.IsPrefixObject())
+}
+
+// IsPrefixObject reports whether the directory entry also holds the object named by its
+// path without a trailing slash, which an empty object leaves no other trace of.
+func (entry *Entry) IsPrefixObject() bool {
+	if entry == nil || !entry.IsDirectory {
+		return false
+	}
+	_, marked := entry.Extended[s3_constants.SeaweedFSPrefixObject]
+	return marked
+}
+
+// MarkPrefixObject turns a file entry into the directory entry that stores it, so keys
+// nested under its key can be created beside it.
+func (entry *Entry) MarkPrefixObject() {
+	entry.IsDirectory = true
+	if entry.Attributes != nil {
+		entry.Attributes.FileMode |= uint32(os.ModeDir) | 0111
+	}
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+	entry.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true")
 }
 
 func (entry *Entry) GetExpiryTime() (expiryTime int64) {

weed/pb/filer_pb/filer_pb_helper_test.go

@@ -1,7 +1,10 @@
 package filer_pb
 
 import (
+	"os"
 	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
 func TestIsDirectoryKeyObject(t *testing.T) {
@@ -20,6 +23,7 @@ func TestIsDirectoryKeyObject(t *testing.T) {
 		{"directory with chunks and nil attributes", &Entry{IsDirectory: true, Chunks: chunk}, true},
 		{"regular file with chunks", &Entry{IsDirectory: false, Attributes: &FuseAttributes{}, Chunks: chunk}, false},
 		{"remote mount directory has no remote size", &Entry{IsDirectory: true, Attributes: &FuseAttributes{}, RemoteEntry: &RemoteEntry{StorageName: "s3"}}, false},
+		{"empty prefix object has only the mark", &Entry{IsDirectory: true, Attributes: &FuseAttributes{}, Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}}, true},
 	}
 
 	for _, c := range cases {
@@ -30,3 +34,28 @@ func TestIsDirectoryKeyObject(t *testing.T) {
 		})
 	}
 }
+
+func TestMarkPrefixObject(t *testing.T) {
+	entry := &Entry{Name: "foo", Attributes: &FuseAttributes{FileMode: 0644}}
+
+	if entry.IsPrefixObject() {
+		t.Fatal("a file is not a prefix object")
+	}
+
+	entry.MarkPrefixObject()
+
+	if !entry.IsDirectory || !entry.IsPrefixObject() {
+		t.Errorf("MarkPrefixObject() left %+v", entry)
+	}
+	// The filer derives the entry type from the mode, so the directory bit has to be
+	// on it for the store to keep this as a directory.
+	if mode := os.FileMode(entry.Attributes.FileMode); mode&os.ModeDir == 0 || mode.Perm()&0111 != 0111 {
+		t.Errorf("FileMode = %v, want a traversable directory", mode)
+	}
+
+	// A directory the mark was stripped from is a plain directory again.
+	delete(entry.Extended, s3_constants.SeaweedFSPrefixObject)
+	if entry.IsPrefixObject() || entry.IsDirectoryKeyObject() {
+		t.Error("a demoted directory names no key")
+	}
+}

weed/s3api/filer_util.go

@@ -24,7 +24,18 @@ func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn fun
 
 func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk, fn func(entry *filer_pb.Entry)) error {
 
-	return filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn)
+	err := filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn)
+	if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(fileName) {
+		// Other keys are nested under this one, so the object goes onto the directory
+		// they live in - the same place a PutObject of this key writes it.
+		err = filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, func(entry *filer_pb.Entry) {
+			if fn != nil {
+				fn(entry)
+			}
+			entry.MarkPrefixObject()
+		})
+	}
+	return err
 
 }
 
@@ -187,6 +198,10 @@ func clearDirectoryMarkerMetadata(entry *filer_pb.Entry) {
 	filtered := make(map[string][]byte)
 	for k, v := range entry.Extended {
 		lowerKey := strings.ToLower(k)
+		if lowerKey == s3_constants.SeaweedFSPrefixObject {
+			// The path is a plain directory again, not a key of its own.
+			continue
+		}
 		if strings.HasPrefix(lowerKey, "xattr-") || strings.HasPrefix(lowerKey, s3_constants.SeaweedFSInternalPrefix) {
 			filtered[k] = v
 		}

weed/s3api/filer_util_delete_test.go

@@ -96,6 +96,36 @@ func TestDeleteObjectEntryDemotesNonEmptyDirectoryMarker(t *testing.T) {
 	}, updated.Extended)
 }
 
+// A prefix object is demoted the same way, and the mark has to go with the data:
+// the path is a plain directory again, and no longer a key of its own.
+func TestDeleteObjectEntryDemotesPrefixObject(t *testing.T) {
+	client := &deleteObjectEntryTestClient{
+		deleteResp: &filer_pb.DeleteEntryResponse{
+			Error: filer.MsgFailDelNonEmptyFolder + ": /buckets/test/photos",
+		},
+		lookupResp: &filer_pb.LookupDirectoryEntryResponse{
+			Entry: &filer_pb.Entry{
+				Name:        "photos",
+				IsDirectory: true,
+				Attributes:  &filer_pb.FuseAttributes{},
+				Extended: map[string][]byte{
+					s3_constants.SeaweedFSPrefixObject: []byte("true"),
+					s3_constants.ExtETagKey:            []byte("etag"),
+				},
+			},
+		},
+	}
+
+	require.NoError(t, deleteObjectEntry(client, "/buckets/test", "photos", true, false))
+	require.NotNil(t, client.updateReq)
+
+	updated := client.updateReq.Entry
+	require.NotNil(t, updated)
+	assert.False(t, updated.IsPrefixObject())
+	assert.False(t, updated.IsDirectoryKeyObject())
+	assert.Empty(t, updated.Extended)
+}
+
 func TestDeleteObjectEntryTreatsImplicitDirectoryAsSuccessfulNoop(t *testing.T) {
 	client := &deleteObjectEntryTestClient{
 		deleteResp: &filer_pb.DeleteEntryResponse{

weed/s3api/s3_constants/header.go

@@ -139,6 +139,12 @@ const (
 	// SeaweedFS internal metadata prefix (used to filter internal headers from client responses)
 	SeaweedFSInternalPrefix = "x-seaweedfs-"
 
+	// SeaweedFSPrefixObject marks a directory entry that also holds the object named by
+	// its path without a trailing slash. S3 keys are flat, so "a/b" and "a/b/c" are
+	// independent keys, and the filer stores the first one on the directory the second
+	// one lives under.
+	SeaweedFSPrefixObject = "x-seaweedfs-prefix-object"
+
 	// SeaweedFS internal metadata keys for encryption (prefixed to avoid automatic HTTP header conversion)
 	SeaweedFSSSEKMSKey = "x-seaweedfs-sse-kms-key" // Key for storing serialized SSE-KMS metadata
 	SeaweedFSSSES3Key  = "x-seaweedfs-sse-s3-key"  // Key for storing serialized SSE-S3 metadata

weed/s3api/s3api_object_handlers.go

@@ -364,7 +364,7 @@ func (s3a *S3ApiServer) hasChildren(ctx context.Context, bucket, prefix string)
 // Such a path is not an S3 object: GET/HEAD answer 404 like AWS does for a prefix, and
 // Hadoop-style clients then discover the directory through their LIST fallback.
 func isBareDirectory(entry *filer_pb.Entry) bool {
-	return entry != nil && entry.IsDirectory && filer.FileSize(entry) == 0
+	return entry != nil && entry.IsDirectory && filer.FileSize(entry) == 0 && !entry.IsPrefixObject()
 }
 
 // checkDirectoryObject checks if the object is a directory object (ends with "/") and if it exists

weed/s3api/s3api_object_handlers_copy.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"reflect"
 	"strconv"
 	"strings"
@@ -69,6 +70,23 @@ func hasPrefixFold(s, prefix string) bool {
 	return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix)
 }
 
+// prefixObjectSource presents the object stored on a key that other keys are nested
+// under as the plain object that key names. The entry itself stays a directory - it
+// is where those keys live - so a caller that copies or moves the object works off
+// this view and leaves the directory, and everything under it, alone.
+func prefixObjectSource(entry *filer_pb.Entry) *filer_pb.Entry {
+	if entry == nil || !entry.IsPrefixObject() {
+		return entry
+	}
+	flattened := proto.Clone(entry).(*filer_pb.Entry)
+	flattened.IsDirectory = false
+	if flattened.Attributes != nil {
+		flattened.Attributes.FileMode &^= uint32(os.ModeDir)
+	}
+	delete(flattened.Extended, s3_constants.SeaweedFSPrefixObject)
+	return flattened
+}
+
 // classifyCopySourceError maps a copy-source lookup to an S3 error: a missing
 // or directory source is NoSuchKey like AWS, but a transient store error
 // becomes a retryable 5xx so a resumable copy/commit survives a blip.
@@ -180,6 +198,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	}
 
 	entry, err := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
+	entry = prefixObjectSource(entry)
 	if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
@@ -250,6 +269,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		routeInPlace := owner != "" && !mimeChanged
 		selfCopyBody := func() s3err.ErrorCode {
 			currentEntry, currentErr := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
+			currentEntry = prefixObjectSource(currentEntry)
 			if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone {
 				return errCode
 			}

weed/s3api/s3api_object_handlers_delete.go

@@ -299,11 +299,10 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 					}
 				}
 				if versionId == "null" {
-					deleteCode = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass)
+					deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass)
 				} else {
-					deleteCode = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass)
+					deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass), true
 				}
-				deleteHandled = true
 			}
 		}
 	}

weed/s3api/s3api_object_handlers_dir_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
 func TestServeDirectoryContentContentType(t *testing.T) {
@@ -47,3 +48,31 @@ func TestServeDirectoryContentContentType(t *testing.T) {
 		})
 	}
 }
+
+func TestIsBareDirectory(t *testing.T) {
+	tests := []struct {
+		name  string
+		entry *filer_pb.Entry
+		want  bool
+	}{
+		{"nothing there", nil, false},
+		{"file", &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{}}, false},
+		{"directory a listing stands for", &filer_pb.Entry{IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}}, true},
+		{"promoted file with data", &filer_pb.Entry{IsDirectory: true, Content: []byte("data"), Attributes: &filer_pb.FuseAttributes{FileSize: 4}}, false},
+		{
+			// The key is real even at zero bytes, so it answers rather than 404s.
+			name: "empty prefix object",
+			entry: &filer_pb.Entry{IsDirectory: true, Attributes: &filer_pb.FuseAttributes{},
+				Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isBareDirectory(tt.entry); got != tt.want {
+				t.Errorf("isBareDirectory() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

weed/s3api/s3api_object_handlers_list.go

@@ -400,6 +400,28 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 			return ""
 		}
 
+		// addCommonPrefix records a CommonPrefix the page has not emitted yet. maxKeys is
+		// unsigned, so the decrement is clamped rather than allowed to wrap: a prefix
+		// object spends two slots on one entry, and the budget is only checked between
+		// entries.
+		addCommonPrefix := func(prefix string) {
+			for i := range commonPrefixes {
+				if commonPrefixes[i].Prefix == prefix {
+					if prefix == lastCommonPrefix {
+						lastPrefixConfirmed = true
+					}
+					return
+				}
+			}
+			commonPrefixes = append(commonPrefixes, PrefixEntry{Prefix: prefix})
+			if cursor.maxKeys > 0 {
+				cursor.maxKeys--
+			}
+			lastEntryWasCommonPrefix = true
+			lastCommonPrefix = prefix
+			lastPrefixNullBackers, lastPrefixConfirmed = nil, true
+		}
+
 		retractPrefixBacking := func(prefix, dir, name string) {
 			if prefix != lastCommonPrefix || !lastPrefixNullBackers[dir+"/"+name] {
 				return
@@ -514,6 +536,29 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 							}
 						}
 					}
+					// A prefix object's key carries no trailing slash, so it lists as a
+					// plain key. The traversal never descends into it under a delimiter,
+					// so the CommonPrefix its nested keys fold into is added here too.
+					if entry.IsPrefixObject() {
+						key := fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):]
+						if strings.HasPrefix(key, originalPrefix) {
+							if folded := prefixForKey(dir, entry.Name); folded != "" {
+								// The key and everything under it fold into the same prefix.
+								addCommonPrefix(folded)
+								return
+							}
+							appendOrDedup(newListEntry(s3a, entry, "", dirName, entryName, bucketPrefix, fetchOwner, false, false))
+							lastEntryWasCommonPrefix = false
+						}
+						// The key and the prefix come off one entry, which a marker names as a
+						// whole, so a page ending between them cannot resume at the prefix:
+						// a marker of "<key>/" means the subtree is done, not that it is
+						// next. The page runs one item over maxKeys instead of dropping it.
+						if childPrefix := prefixForKey(dir, entry.Name+"/"); childPrefix != "" && s3a.hasChildren(ctx, bucket, key) {
+							addCommonPrefix(childPrefix)
+						}
+						return
+					}
 					// When delimiter is specified, apply delimiter logic to directory key objects too
 					if delimiter != "" && entry.IsDirectoryKeyObject() {
 						// Apply the same delimiter logic as for regular files

weed/s3api/s3api_object_handlers_put.go

@@ -876,7 +876,14 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 				Entry:     entry,
 			}
 			glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath)
-			if err := filer_pb.CreateEntry(context.Background(), client, req); err != nil {
+			err := filer_pb.CreateEntry(context.Background(), client, req)
+			if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) {
+				// Other keys are nested under this one. S3 keys are flat, so the key is
+				// stored on the directory they live under rather than refused.
+				entry.MarkPrefixObject()
+				err = filer_pb.CreateEntry(context.Background(), client, req)
+			}
+			if err != nil {
 				glog.Errorf("putToFiler: CreateEntry returned error: %v", err)
 				return err
 			}
@@ -1192,6 +1199,14 @@ func (s3a *S3ApiServer) setSSEResponseHeaders(w http.ResponseWriter, r *http.Req
 	}
 }
 
+// isReservedDirectoryName reports whether a directory standing at an object's path is
+// one SeaweedFS keeps its own state in - a multipart staging folder, or a key's version
+// history - rather than a prefix the key can be stored on. Writing the object onto it
+// would replace that state with the object's own.
+func isReservedDirectoryName(name string) bool {
+	return name == s3_constants.MultipartUploadsFolder || strings.HasSuffix(name, s3_constants.VersionsFolder)
+}
+
 func filerErrorToS3Error(err error) s3err.ErrorCode {
 	if err == nil {
 		return s3err.ErrNone

weed/s3api/s3api_object_handlers_rename.go

@@ -94,6 +94,8 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
 
 	errCode = s3a.withRenameWriteLocks(bucket, srcObject, dstObject, func() s3err.ErrorCode {
 		entry, err := s3a.resolveCopySourceEntry(bucket, srcObject, "", "")
+		srcIsPrefixObject := entry.IsPrefixObject()
+		entry = prefixObjectSource(entry)
 		if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
 			return errCode
 		}
@@ -103,7 +105,7 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
 		if errCode := s3a.checkConditionalHeaders(r, bucket, dstObject); errCode != s3err.ErrNone {
 			return errCode
 		}
-		return s3a.renameObjectEntry(r.Context(), bucket, srcObject, dstObject)
+		return s3a.renameObjectEntry(r.Context(), bucket, srcObject, dstObject, entry, srcIsPrefixObject)
 	})
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
@@ -218,19 +220,28 @@ func (s3a *S3ApiServer) withRenameWriteLocks(bucket, srcObject, dstObject string
 	})
 }
 
-func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject, dstObject string) s3err.ErrorCode {
+func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject, dstObject string, srcEntry *filer_pb.Entry, srcIsPrefixObject bool) s3err.ErrorCode {
 	srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName()
 	dstDir, dstName := util.FullPath(s3a.toFilerPath(bucket, dstObject)).DirAndName()
 
-	// The move overwrites an existing destination object, but a directory in
-	// the way is a conflict the filer reports as an opaque error.
-	if existing, err := s3a.getEntry(dstDir, dstName); err == nil && existing.IsDirectory {
-		return s3err.ErrExistingObjectIsDirectory
-	} else if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
+	// The move overwrites an existing destination object. A directory there is not a
+	// conflict: it means other keys are nested under the destination key, and the
+	// object goes onto the directory they live in, the way a PutObject of that key
+	// would put it there.
+	dstHoldsNestedKeys := false
+	if existing, err := s3a.getEntry(dstDir, dstName); err == nil {
+		dstHoldsNestedKeys = existing.IsDirectory
+	} else if !errors.Is(err, filer_pb.ErrNotFound) {
 		glog.Errorf("RenameObject %s: destination %s: %v", bucket, dstObject, err)
 		return s3err.ErrInternalError
 	}
 
+	// AtomicRenameEntry moves a directory by moving everything under it, and the keys
+	// nested under either end of this rename are not part of what is being renamed.
+	if srcIsPrefixObject || dstHoldsNestedKeys {
+		return s3a.renameKeyHoldingNestedKeys(bucket, srcObject, dstObject, srcEntry)
+	}
+
 	err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 		_, err := client.AtomicRenameEntry(ctx, &filer_pb.AtomicRenameEntryRequest{
 			OldDirectory: srcDir,
@@ -249,3 +260,38 @@ func (s3a *S3ApiServer) renameObjectEntry(ctx context.Context, bucket, srcObject
 	}
 	return s3err.ErrNone
 }
+
+// renameKeyHoldingNestedKeys moves an object when either key of the rename is one
+// other keys are nested under. Such a key is stored on the directory those keys live
+// in, which has to stay where it is, so the object's own data is written at the
+// destination and then stripped off the source key - the entry survives as the plain
+// directory it also is. Both keys are held under their write locks for the whole
+// move, so no other S3 write interleaves; a crash between the two steps leaves the
+// destination written and the source still there, which a retry settles.
+func (s3a *S3ApiServer) renameKeyHoldingNestedKeys(bucket, srcObject, dstObject string, srcEntry *filer_pb.Entry) s3err.ErrorCode {
+	dstPath := util.FullPath(s3a.toFilerPath(bucket, dstObject))
+	dstDir, dstName := dstPath.DirAndName()
+
+	chunks, err := s3a.copyChunks(srcEntry, string(dstPath))
+	if err != nil {
+		glog.Errorf("RenameObject %s: copy chunks of %s: %v", bucket, srcObject, err)
+		return s3err.ErrInternalError
+	}
+
+	if err := s3a.mkFile(dstDir, dstName, chunks, func(entry *filer_pb.Entry) {
+		copyEntryToTarget(entry, srcEntry)
+		entry.Chunks = chunks
+	}); err != nil {
+		glog.Errorf("RenameObject %s: write %s: %v", bucket, dstObject, err)
+		s3a.deleteOrphanedChunks(chunks)
+		return filerErrorToS3Error(err)
+	}
+
+	// The destination holds copies now, so the source's own chunks go with it.
+	srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName()
+	if err := s3a.rmObject(srcDir, srcName, true, false); err != nil {
+		glog.Errorf("RenameObject %s: strip %s: %v", bucket, srcObject, err)
+		return s3err.ErrInternalError
+	}
+	return s3err.ErrNone
+}

weed/s3api/s3api_object_versioned_finalize.go

@@ -156,8 +156,10 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck
 // routedDeleteNullVersion deletes the null version (the regular object entry, not
 // a .versions file) off the distributed lock. There is no pointer to recompute;
 // the WORM guards, when present, gate the delete on the object entry itself
-// (condition defaults to lock_key).
-func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass bool) s3err.ErrorCode {
+// (condition defaults to lock_key). The second return reports whether the delete
+// was settled here: the raw delete cannot remove an entry other keys are nested
+// under, which the lock path handles by stripping the object off it instead.
+func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass bool) (s3err.ErrorCode, bool) {
 	fullpath := util.NewFullPath(s3a.bucketDir(bucket), object)
 	dir, name := fullpath.DirAndName()
 	resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
@@ -170,15 +172,15 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket,
 	})
 	switch {
 	case err != nil:
-		glog.Errorf("routedDeleteNullVersion: %s/%s on %s: %v", bucket, object, owner, err)
-		return s3err.ErrInternalError
+		glog.Warningf("routedDeleteNullVersion: %s/%s on %s, falling back to lock: %v", bucket, object, owner, err)
+		return s3err.ErrNone, false
 	case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
-		return s3err.ErrAccessDenied
+		return s3err.ErrAccessDenied, true
 	case resp.Error != "":
-		glog.Errorf("routedDeleteNullVersion: %s/%s: %s", bucket, object, resp.Error)
-		return s3err.ErrInternalError
+		glog.Warningf("routedDeleteNullVersion: %s/%s returned %q, falling back to lock", bucket, object, resp.Error)
+		return s3err.ErrNone, false
 	default:
-		return s3err.ErrNone
+		return s3err.ErrNone, true
 	}
 }
 

weed/s3api/s3api_object_versioning.go

@@ -22,6 +22,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -856,6 +857,13 @@ func (vc *versionCollector) collectVersions(currentPath, relativePath string) er
 							vc.commonPrefixes[commonPrefix] = true
 						}
 
+						// The prefix rolled up here belongs to the keys nested under this
+						// entry. A prefix object is also a key of its own, and that key
+						// carries no trailing slash, so it does not roll up with them.
+						if entry.IsPrefixObject() {
+							vc.processPrefixObject(currentPath, entryPath, entry)
+						}
+
 						// Skip further processing (recursion or addition) for this entry
 						// because it has been rolled up into the CommonPrefix
 						continue
@@ -879,13 +887,31 @@ func (vc *versionCollector) collectVersions(currentPath, relativePath string) er
 	return nil
 }
 
+// processPrefixObject emits the null version of a key that other keys are nested
+// under. The key carries no trailing slash, so it is a null object like any other and
+// needs the same reconciliation against a .versions sibling - but the directory it is
+// stored on is walked as an ancestor of the requested prefix, which its own key does
+// not have to match.
+func (vc *versionCollector) processPrefixObject(currentPath, entryPath string, entry *filer_pb.Entry) {
+	if !strings.HasPrefix(entryPath, vc.prefix) {
+		return
+	}
+	if vc.delimiter != "" && strings.Contains(entryPath[len(vc.prefix):], vc.delimiter) {
+		// The key folds into the same CommonPrefix its nested keys do.
+		return
+	}
+	vc.processRegularFile(currentPath, entryPath, entry)
+}
+
 // processDirectory handles directory entries
 func (vc *versionCollector) processDirectory(currentPath, entryPath string, entry *filer_pb.Entry) error {
 	// Handle explicit S3 directory object. Match ListObjectsV2's
 	// IsDirectoryKeyObject (any non-empty mime), not just FolderMimeType:
 	// 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() {
+	if entry.IsPrefixObject() {
+		vc.processPrefixObject(currentPath, entryPath, entry)
+	} else if entry.IsDirectoryKeyObject() {
 		vc.processExplicitDirectory(entryPath, entry)
 	}
 
@@ -1113,8 +1139,11 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket,
 			return nil
 		}
 
-		// Delete the regular file
-		deleteErr := s3a.rmObject(bucketDir, normalizedObject, !metadataOnly, false)
+		// Delete the regular file. rmObject takes a parent and a name, and the demote
+		// it falls back to for an entry other keys are nested under writes the entry
+		// back under that parent - so a key with a slash in it has to be split first.
+		dir, name := util.NewFullPath(bucketDir, normalizedObject).DirAndName()
+		deleteErr := s3a.rmObject(dir, name, !metadataOnly, false)
 		if deleteErr != nil {
 			// Check if file was already deleted by another process
 			if _, checkErr := s3a.getEntry(bucketDir, normalizedObject); checkErr != nil {