remote storage: actually delete objects when a directory is removed (#10531)

* remote storage: actually delete objects when a directory is removed

On object-store backends RemoveDirectory returned nil without doing
anything, so a directory delete synced to the remote as a successful
no-op and the objects under that prefix stayed there forever. Nothing
surfaced the divergence: the sync logged rmdir, advanced its offset,
and the local namespace looked clean.

Deleting a bucket-level directory on a filer store that can drop a
whole bucket emits no per-child delete events at all, so the single
rmdir event was the only chance to clean up the remote.

Each backend now lists the prefix and deletes what it finds: S3 in
DeleteObjects batches of one listing page, GCS and Azure per object.
The prefix always ends with a slash so a sibling like dir2 survives
deleting dir, and errors propagate so a failed delete is retried
instead of silently skipped. A directory that maps to the bucket root
is left alone: wiping every object in the bucket from one namespace
event is too destructive, and bucket removal already has its own path.

* gcs remote: wrap the per-object delete error

The listing error in the same function already wraps, so the delete
error should stay inspectable with errors.Is as well.

* s3 remote: name the empty-listing test for what it checks

The prefix in that test is a normal directory; what is empty is the
listing. The bucket-root guard has its own test.

* s3 remote: report the scope of a failed delete batch

A DeleteObjects response can carry per-key errors for up to a
thousand keys. Surfacing only the first hid how much of the batch
failed, and surfacing all of them would build an unbounded error
string, so report the count with the first failure as the sample.

weed/remote_storage/azure/azure_storage_client.go

@@ -19,6 +19,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
@@ -382,6 +383,39 @@ func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageL
 }
 
 func (az *azureRemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) {
+	// the trailing slash keeps sibling prefixes that share the name intact
+	prefix := loc.Path[1:]
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
+		prefix += "/"
+	}
+	if prefix == "" {
+		// the mount root maps to the whole container; wiping every blob from a
+		// single namespace event is too destructive, so keep them
+		glog.Warningf("azure %s: skip removing directory mapped to the container root", loc.Bucket)
+		return nil
+	}
+
+	containerClient := az.client.ServiceClient().NewContainerClient(loc.Bucket)
+	pager := containerClient.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
+		Prefix: &prefix,
+	})
+	for pager.More() {
+		resp, pageErr := pager.NextPage(context.Background())
+		if pageErr != nil {
+			return fmt.Errorf("azure list %s/%s: %w", loc.Bucket, prefix, pageErr)
+		}
+		for _, blobItem := range resp.Segment.BlobItems {
+			if blobItem.Name == nil {
+				continue
+			}
+			_, delErr := containerClient.NewBlobClient(*blobItem.Name).Delete(context.Background(), &blob.DeleteOptions{
+				DeleteSnapshots: to.Ptr(blob.DeleteSnapshotsOptionTypeInclude),
+			})
+			if delErr != nil && !bloberror.HasCode(delErr, bloberror.BlobNotFound) {
+				return fmt.Errorf("azure delete %s/%s: %w", loc.Bucket, *blobItem.Name, delErr)
+			}
+		}
+	}
 	return nil
 }
 

weed/remote_storage/gcs/gcs_storage_client.go

@@ -220,7 +220,34 @@ func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLo
 }
 
 func (gcs *gcsRemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) {
-	return nil
+	// the trailing slash keeps sibling prefixes that share the name intact
+	prefix := loc.Path[1:]
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
+		prefix += "/"
+	}
+	if prefix == "" {
+		// the mount root maps to the whole bucket; wiping every object from a
+		// single namespace event is too destructive, so keep them
+		glog.Warningf("gcs %s: skip removing directory mapped to the bucket root", loc.Bucket)
+		return nil
+	}
+
+	bucket := gcs.client.Bucket(loc.Bucket)
+	objectIterator := bucket.Objects(context.Background(), &storage.Query{
+		Prefix: prefix,
+	})
+	for {
+		objectAttr, iterErr := objectIterator.Next()
+		if iterErr == iterator.Done {
+			return nil
+		}
+		if iterErr != nil {
+			return fmt.Errorf("gcs list %s/%s: %w", loc.Bucket, prefix, iterErr)
+		}
+		if delErr := bucket.Object(objectAttr.Name).Delete(context.Background()); delErr != nil && !errors.Is(delErr, storage.ErrObjectNotExist) {
+			return fmt.Errorf("gcs delete %s/%s: %w", loc.Bucket, objectAttr.Name, delErr)
+		}
+	}
 }
 
 func (gcs *gcsRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (remoteEntry *filer_pb.RemoteEntry, err error) {

weed/remote_storage/s3/s3_storage_client.go

@@ -285,6 +285,57 @@ func (s *s3RemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocat
 }
 
 func (s *s3RemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) {
+	// the trailing slash keeps sibling prefixes that share the name intact
+	prefix := loc.Path[1:]
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
+		prefix += "/"
+	}
+	if prefix == "" {
+		// the mount root maps to the whole bucket; wiping every object from a
+		// single namespace event is too destructive, so keep them
+		glog.Warningf("s3 %s: skip removing directory mapped to the bucket root", loc.Bucket)
+		return nil
+	}
+
+	listInput := &s3.ListObjectsV2Input{
+		Bucket: aws.String(loc.Bucket),
+		Prefix: aws.String(prefix),
+	}
+	var deleteErr error
+	listErr := s.conn.ListObjectsV2Pages(listInput, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
+		var objects []*s3.ObjectIdentifier
+		for _, content := range page.Contents {
+			objects = append(objects, &s3.ObjectIdentifier{Key: content.Key})
+		}
+		if len(objects) == 0 {
+			return true
+		}
+		// a listing page holds at most 1000 keys, the DeleteObjects limit
+		resp, batchErr := s.conn.DeleteObjects(&s3.DeleteObjectsInput{
+			Bucket: aws.String(loc.Bucket),
+			Delete: &s3.Delete{
+				Objects: objects,
+				Quiet:   aws.Bool(true),
+			},
+		})
+		if batchErr != nil {
+			deleteErr = batchErr
+			return false
+		}
+		if len(resp.Errors) > 0 {
+			// a batch can fail 1000 keys; report the scope, not every key
+			failed := resp.Errors[0]
+			deleteErr = fmt.Errorf("%d keys failed, first is %s: %s %s", len(resp.Errors), aws.StringValue(failed.Key), aws.StringValue(failed.Code), aws.StringValue(failed.Message))
+			return false
+		}
+		return true
+	})
+	if listErr != nil {
+		return fmt.Errorf("list %s/%s: %w", loc.Bucket, prefix, listErr)
+	}
+	if deleteErr != nil {
+		return fmt.Errorf("remove directory %s/%s: %w", loc.Bucket, prefix, deleteErr)
+	}
 	return nil
 }
 

weed/remote_storage/s3/s3_storage_client_test.go

@@ -2,13 +2,16 @@ package s3
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	awss3 "github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
@@ -71,6 +74,125 @@ func TestS3ErrRemoteObjectNotFoundIsAccessible(t *testing.T) {
 	require.Equal(t, "remote object not found", remote_storage.ErrRemoteObjectNotFound.Error())
 }
 
+// removeDirectoryMock serves canned listing pages and records the delete batches.
+type removeDirectoryMock struct {
+	s3iface.S3API
+	pages        []*awss3.ListObjectsV2Output
+	listInputs   []*awss3.ListObjectsV2Input
+	deleteInputs []*awss3.DeleteObjectsInput
+	deleteResp   *awss3.DeleteObjectsOutput
+	deleteErr    error
+}
+
+func (m *removeDirectoryMock) ListObjectsV2Pages(input *awss3.ListObjectsV2Input, fn func(*awss3.ListObjectsV2Output, bool) bool) error {
+	m.listInputs = append(m.listInputs, input)
+	for i, page := range m.pages {
+		if !fn(page, i == len(m.pages)-1) {
+			return nil
+		}
+	}
+	return nil
+}
+
+func (m *removeDirectoryMock) DeleteObjects(input *awss3.DeleteObjectsInput) (*awss3.DeleteObjectsOutput, error) {
+	m.deleteInputs = append(m.deleteInputs, input)
+	if m.deleteErr != nil {
+		return nil, m.deleteErr
+	}
+	if m.deleteResp != nil {
+		return m.deleteResp, nil
+	}
+	return &awss3.DeleteObjectsOutput{}, nil
+}
+
+func listPage(keys ...string) *awss3.ListObjectsV2Output {
+	page := &awss3.ListObjectsV2Output{}
+	for _, key := range keys {
+		page.Contents = append(page.Contents, &awss3.Object{Key: aws.String(key)})
+	}
+	return page
+}
+
+func deletedKeys(input *awss3.DeleteObjectsInput) (keys []string) {
+	for _, object := range input.Delete.Objects {
+		keys = append(keys, aws.StringValue(object.Key))
+	}
+	return
+}
+
+func TestS3RemoveDirectoryDeletesEveryListedObject(t *testing.T) {
+	mock := &removeDirectoryMock{
+		pages: []*awss3.ListObjectsV2Output{
+			listPage("testdir/a.bin", "testdir/sub/b.bin"),
+			listPage("testdir/z.bin"),
+		},
+	}
+	client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"}
+
+	require.NoError(t, client.RemoveDirectory(loc))
+
+	require.Len(t, mock.listInputs, 1)
+	// without the trailing slash the listing would also match /testdir2
+	require.Equal(t, "testdir/", aws.StringValue(mock.listInputs[0].Prefix))
+	require.Len(t, mock.deleteInputs, 2)
+	require.Equal(t, []string{"testdir/a.bin", "testdir/sub/b.bin"}, deletedKeys(mock.deleteInputs[0]))
+	require.Equal(t, []string{"testdir/z.bin"}, deletedKeys(mock.deleteInputs[1]))
+}
+
+func TestS3RemoveDirectoryEmptyListingSendsNoDeletes(t *testing.T) {
+	mock := &removeDirectoryMock{pages: []*awss3.ListObjectsV2Output{listPage()}}
+	client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"}
+
+	require.NoError(t, client.RemoveDirectory(loc))
+	require.Empty(t, mock.deleteInputs)
+}
+
+func TestS3RemoveDirectoryRefusesBucketRoot(t *testing.T) {
+	mock := &removeDirectoryMock{pages: []*awss3.ListObjectsV2Output{listPage("a.bin")}}
+	client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/"}
+
+	require.NoError(t, client.RemoveDirectory(loc))
+	require.Empty(t, mock.listInputs)
+	require.Empty(t, mock.deleteInputs)
+}
+
+func TestS3RemoveDirectoryReturnsBatchError(t *testing.T) {
+	mock := &removeDirectoryMock{
+		pages:     []*awss3.ListObjectsV2Output{listPage("testdir/a.bin")},
+		deleteErr: fmt.Errorf("access denied"),
+	}
+	client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"}
+
+	require.ErrorContains(t, client.RemoveDirectory(loc), "access denied")
+}
+
+func TestS3RemoveDirectoryReturnsPerKeyError(t *testing.T) {
+	mock := &removeDirectoryMock{
+		pages: []*awss3.ListObjectsV2Output{listPage("testdir/a.bin", "testdir/b.bin")},
+		deleteResp: &awss3.DeleteObjectsOutput{
+			Errors: []*awss3.Error{{
+				Key:     aws.String("testdir/a.bin"),
+				Code:    aws.String("InternalError"),
+				Message: aws.String("try again"),
+			}, {
+				Key:     aws.String("testdir/b.bin"),
+				Code:    aws.String("InternalError"),
+				Message: aws.String("try again"),
+			}},
+		},
+	}
+	client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock}
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"}
+
+	err := client.RemoveDirectory(loc)
+	require.ErrorContains(t, err, "2 keys failed")
+	require.ErrorContains(t, err, "testdir/a.bin")
+}
+
 // captureRoundTripper records the PUT request that the s3manager uploader
 // sends, and short-circuits all calls with a 200 so the SDK is satisfied.
 type captureRoundTripper struct {