serve s3: remove a directory with its last key after its marker is deleted

With --directory-markers, deleting the directory marker of a directory
holding other keys had no effect, so the directory was still listed
and wasn't removed when the last key below it was deleted. This left
directories behind when the keys of a directory were deleted in an
order other than deepest first, as rclone's s3 backend does when
purging.

Remember the deletion in memory, so the marker isn't listed and the
directory is removed with the last key below it.

cmd/serve/s3/backend.go

@@ -94,12 +94,13 @@ type tenant struct {
 // objectMeta is the metadata stored for an object, with the size and
 // modification time the object had when it was stored.
 type objectMeta struct {
-	meta      map[string]string
-	hash      []byte  // the ETag of the object, if not its hash
-	partSizes []int64 // the sizes of the parts of an object made by a multipart upload
-	size      int64
-	modTime   time.Time
-	dirMarker bool // a directory marker was put for the directory at this path
+	meta        map[string]string
+	hash        []byte  // the ETag of the object, if not its hash
+	partSizes   []int64 // the sizes of the parts of an object made by a multipart upload
+	size        int64
+	modTime     time.Time
+	dirMarker   bool // a directory marker was put for the directory at this path
+	noDirMarker bool // the directory marker of the directory at this path was deleted, with --directory-markers
 }
 
 // newMetadataStore makes a store for the metadata of up to maxObjects
@@ -161,7 +162,7 @@ func (t *tenant) context() context.Context {
 // stored, as it has been replaced by something other than this user.
 func (t *tenant) getMeta(ctx context.Context, fp string, node vfs.Node) (om objectMeta, ok bool) {
 	om, ok = t.meta.Get(fp)
-	if !ok || om.dirMarker || om.size != node.Size() {
+	if !ok || om.dirMarker || om.noDirMarker || om.size != node.Size() {
 		return objectMeta{}, false
 	}
 	window := fs.GetModifyWindow(ctx, node.VFS().Fs())
@@ -258,12 +259,13 @@ func (t *tenant) storeDirMarker(fp string) {
 }
 
 // hasDirMarker reports whether the directory at fp has a directory
-// marker: one was put for it, or --directory-markers is set.
+// marker: one was put for it, or --directory-markers is set and it
+// hasn't been deleted.
 func (t *tenant) hasDirMarker(fp string) bool {
-	if t.b.s.opt.DirectoryMarkers {
-		return true
-	}
 	om, ok := t.meta.Get(fp)
+	if t.b.s.opt.DirectoryMarkers {
+		return !ok || !om.noDirMarker
+	}
 	return ok && om.dirMarker
 }
 
@@ -800,7 +802,10 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str
 			if err := _vfs.Remove(fp); err != nil {
 				return err
 			}
-			rmdirRecursive(fp, _vfs, t.hasDirMarker)
+			rmdirRecursive(fp, _vfs, t)
+		} else if b.s.opt.DirectoryMarkers {
+			// Otherwise it is removed with the last key below it
+			t.meta.Add(fp, objectMeta{noDirMarker: true})
 		}
 		return nil
 	}
@@ -816,7 +821,7 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str
 	t.removeMeta(fp)
 
 	// FIXME: unsafe operation
-	rmdirRecursive(fp, _vfs, t.hasDirMarker)
+	rmdirRecursive(fp, _vfs, t)
 	return nil
 }
 

cmd/serve/s3/list_test.go

@@ -451,4 +451,22 @@ func TestListDirectoryMarkers(t *testing.T) {
 	assert.Equal(t, []string{"a/", "a/b/", "implicit/", "implicit/file"}, all())
 	require.NoError(t, b.deleteObject(ctx, "bucket", "a/b/"))
 	assert.Equal(t, []string{"a/", "implicit/", "implicit/file"}, all())
+
+	// A marker deleted while its directory holds other keys isn't
+	// listed, and the directory goes with the last key below it, in
+	// whatever order the keys are deleted
+	_, err := b.PutObject(ctx, "bucket", "x/y/z", nil, strings.NewReader("z"), 1)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"a/", "implicit/", "implicit/file", "x/", "x/y/", "x/y/z"}, all())
+	require.NoError(t, b.deleteObject(ctx, "bucket", "x/"))
+	assert.Equal(t, []string{"a/", "implicit/", "implicit/file", "x/y/", "x/y/z"}, all())
+	require.NoError(t, b.deleteObject(ctx, "bucket", "x/y/z"))
+	assert.Equal(t, []string{"a/", "implicit/", "implicit/file", "x/y/"}, all())
+	require.NoError(t, b.deleteObject(ctx, "bucket", "x/y/"))
+	assert.Equal(t, []string{"a/", "implicit/", "implicit/file"}, all())
+
+	// A directory made again has its marker
+	_, err = b.PutObject(ctx, "bucket", "x/w", nil, strings.NewReader("w"), 1)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"a/", "implicit/", "implicit/file", "x/", "x/w"}, all())
 }

cmd/serve/s3/serve_s3.md

@@ -506,8 +506,11 @@ With `--directory-markers` every directory is listed as a directory
 marker, so the markers are listed after a restart too, as needed by
 clients which rely on them, such as rclone's s3 backend with
 `--s3-directory-markers`. As every directory has a marker, a directory
-isn't removed when the last key below it is deleted, and deleting the
-marker of a directory which holds other keys has no effect.
+isn't removed when the last key below it is deleted. Deleting the
+marker of a directory which holds other keys stops it being listed,
+and the directory is then removed with the last key below it, but this
+is only remembered in memory, like the markers put without
+`--directory-markers`.
 
 As keys are file paths, a key can't be both an object and a directory
 holding other objects, for example `a/b` and `a/b/c`. Writing an

cmd/serve/s3/utils.go

@@ -228,8 +228,9 @@ func removeMadeDirs(dir, made string, VFS *vfs.VFS) {
 }
 
 // rmdirRecursive removes the parent directories of p which are empty,
-// stopping at the bucket or at a directory for which keep is true.
-func rmdirRecursive(p string, VFS *vfs.VFS, keep func(dir string) bool) {
+// stopping at the bucket or at a directory with a directory marker of
+// the user t, and forgets what t remembers of the directories removed.
+func rmdirRecursive(p string, VFS *vfs.VFS, t *tenant) {
 	dir := path.Dir(p)
 	if !strings.ContainsAny(dir, "/\\") {
 		// might be bucket(root)
@@ -237,13 +238,14 @@ func rmdirRecursive(p string, VFS *vfs.VFS, keep func(dir string) bool) {
 	}
 	// Check the directory is empty first as the VFS logs an error
 	// removing one which isn't
-	if keep(dir) || !isEmptyDir(dir, VFS) {
+	if t.hasDirMarker(dir) || !isEmptyDir(dir, VFS) {
 		return
 	}
 	if err := VFS.Remove(dir); err != nil {
 		return
 	}
-	rmdirRecursive(dir, VFS, keep)
+	t.removeMeta(dir)
+	rmdirRecursive(dir, VFS, t)
 }
 
 // pathLocks are locks on paths, made as needed.