Report a delete the filer rejected instead of answering success (#11003)

* s3tables: report a delete the filer rejected

deleteDirectory discarded DeleteEntryResponse and checked only the
transport error, so DeleteTable, DeleteNamespace, DeleteView and
DeleteTableBucket answered 200 for a delete the filer refused. Call
filer_pb.DoRemove, which reads resp.Error and still treats a missing
entry as success.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* admin: report a delete the filer rejected

The bucket delete, the file browser handlers and the topic retention
purger all discarded DeleteEntryResponse, so a delete the filer refused
came back as success. Call filer_pb.DoRemove, which reads resp.Error.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* credential: report a delete the filer rejected

DeleteUser, DeletePolicy and the full-sync cleanup loops discarded
DeleteEntryResponse, so a rejected delete answered success and left the
credential file in place. The service account path in the same store
already read resp.Error; the rest now do too, via filer_pb.DoRemove
where not-found is already tolerated.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* shell: report a delete the filer rejected

remote.configure -delete, remote.cache and the remote metadata sync
discarded DeleteEntryResponse, so a rejected delete printed as removed.
Call filer_pb.DoRemove, which reads resp.Error.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* mq: report a delete the filer rejected

The consumer offset group purge and the coordinator assignment delete
discarded DeleteEntryResponse. Call filer_pb.DoRemove, which reads
resp.Error.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* iam: count only the revocation entries the filer actually deleted

The expiry sweep discarded DeleteEntryResponse, so a rejected delete was
counted as purged and the entry stayed. Call filer_pb.DoRemove, which
reads resp.Error, matching the role and provider stores beside it.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* mount: fail rmdir when the unary fallback delete was rejected

The streaming branch turns DeleteEntryResponse.Error into an error, the
unary fallback dropped it, so rmdir of a non-empty directory answered OK
off the stream and ENOTEMPTY on it. Surface it in both.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

* s3tables: fail DeleteTableBucket when the directory delete is refused

The handler only failed when both the leaf entry and the directory
delete failed, so a refused bucket directory delete still answered 200
with the bucket in place. The directory is the bucket, so it decides;
the leaf entry stays best-effort.

Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU

weed/s3api/s3tables/filer_ops.go

@@ -295,17 +295,9 @@ func (h *S3TablesHandler) deleteExtendedAttribute(ctx context.Context, client fi
 }
 
 // deleteDirectory deletes a directory and all its contents
-// Note: DeleteEntry RPC response doesn't have an Error field, so we only check the RPC err
 func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
 	dir, name := splitPath(path)
-	_, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{
-		Directory:            dir,
-		Name:                 name,
-		IsDeleteData:         true,
-		IsRecursive:          true,
-		IgnoreRecursiveError: true,
-	})
-	return err
+	return filer_pb.DoRemove(ctx, client, dir, name, true, true, true, false, nil)
 }
 
 // entryExists checks if an entry exists at the given path

weed/s3api/s3tables/filer_ops_test.go

@@ -0,0 +1,102 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// deleteEntryStub answers DeleteEntry the way the filer does: a rejected delete
+// comes back as a nil transport error with the reason in the response.
+type deleteEntryStub struct {
+	filer_pb.SeaweedFilerClient
+	resp *filer_pb.DeleteEntryResponse
+	err  error
+	req  *filer_pb.DeleteEntryRequest
+}
+
+func (s *deleteEntryStub) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) {
+	s.req = req
+	return s.resp, s.err
+}
+
+func TestDeleteDirectoryReportsRejectedDelete(t *testing.T) {
+	stub := &deleteEntryStub{resp: &filer_pb.DeleteEntryResponse{Error: "fail to delete non-empty folder"}}
+
+	err := (&S3TablesHandler{}).deleteDirectory(context.Background(), stub, "/buckets/b/ns/t")
+
+	require.Error(t, err, "a delete the filer rejected must not be reported as success")
+	assert.Contains(t, err.Error(), "fail to delete non-empty folder")
+}
+
+func TestDeleteDirectoryToleratesMissingEntry(t *testing.T) {
+	t.Run("reported in the response", func(t *testing.T) {
+		stub := &deleteEntryStub{resp: &filer_pb.DeleteEntryResponse{Error: filer_pb.ErrNotFound.Error()}}
+		assert.NoError(t, (&S3TablesHandler{}).deleteDirectory(context.Background(), stub, "/buckets/b/ns/t"))
+	})
+	t.Run("reported as a transport error", func(t *testing.T) {
+		stub := &deleteEntryStub{err: status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())}
+		assert.NoError(t, (&S3TablesHandler{}).deleteDirectory(context.Background(), stub, "/buckets/b/ns/t"))
+	})
+}
+
+func TestDeleteDirectoryPropagatesTransportError(t *testing.T) {
+	stub := &deleteEntryStub{err: errors.New("filer unreachable")}
+
+	err := (&S3TablesHandler{}).deleteDirectory(context.Background(), stub, "/buckets/b/ns/t")
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "filer unreachable")
+}
+
+// deleteDirectory drops a whole subtree, so the delete has to stay recursive and
+// take the data with it.
+func TestDeleteDirectoryDeletesTheSubtree(t *testing.T) {
+	stub := &deleteEntryStub{resp: &filer_pb.DeleteEntryResponse{}}
+
+	require.NoError(t, (&S3TablesHandler{}).deleteDirectory(context.Background(), stub, "/buckets/b/ns/t"))
+
+	require.NotNil(t, stub.req)
+	assert.Equal(t, "/buckets/b/ns", stub.req.Directory)
+	assert.Equal(t, "t", stub.req.Name)
+	assert.True(t, stub.req.IsDeleteData)
+	assert.True(t, stub.req.IsRecursive)
+	assert.True(t, stub.req.IgnoreRecursiveError)
+}
+
+// The bucket is the directory, so DeleteTableBucket must not answer success
+// when the filer refused to remove it.
+func TestDeleteTableBucketReportsRejectedDirectoryDelete(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	bucketMeta, err := json.Marshal(tableBucketMetadata{Name: renameTestBucket, OwnerAccountID: DefaultAccountID})
+	require.NoError(t, err)
+	fs.Put(TablesPath, renameTestBucket, map[string][]byte{
+		ExtendedKeyTableBucket: []byte("{}"),
+		ExtendedKeyMetadata:    bucketMeta,
+	})
+
+	bucketPath := GetTableBucketPath(renameTestBucket)
+	fs.RejectDelete = func(dir, name string) string {
+		if dir+"/"+name == bucketPath {
+			return "fail to delete non-empty folder"
+		}
+		return ""
+	}
+
+	m := NewManager()
+	m.SetTrusted(true)
+	err = m.Execute(context.Background(), NewManagerClient(fs.Client), "DeleteTableBucket",
+		&DeleteTableBucketRequest{TableBucketARN: mustBucketARN(t)}, nil, "")
+
+	require.Error(t, err, "a bucket the filer refused to delete must not be reported as deleted")
+	assert.NotNil(t, fs.Get(TablesPath, renameTestBucket), "the bucket is still there")
+}

weed/s3api/s3tables/handler_bucket_get_list_delete.go

@@ -347,22 +347,13 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
 
 	// Delete the bucket
 	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		// Delete table object entry first, then directory
-		// This ensures we clean up the leaf entry even if directory deletion fails
-		tableObjErr := h.deleteEntryIfExists(r.Context(), client, GetTableObjectBucketPath(bucketName))
-		dirErr := h.deleteDirectory(r.Context(), client, bucketPath)
-
-		// Log any errors but don't fail if one succeeds
-		if tableObjErr != nil && dirErr != nil {
-			return fmt.Errorf("delete table object failed: %w, delete directory failed: %w", tableObjErr, dirErr)
-		}
-		if tableObjErr != nil {
+		// Drop the leaf entry first, so it does not outlive the directory.
+		if tableObjErr := h.deleteEntryIfExists(r.Context(), client, GetTableObjectBucketPath(bucketName)); tableObjErr != nil {
 			glog.V(1).Infof("failed to delete table object for %s: %v", bucketName, tableObjErr)
 		}
-		if dirErr != nil {
-			glog.V(1).Infof("failed to delete table bucket dir for %s: %v", bucketName, dirErr)
-		}
-		return nil
+		// The bucket is the directory, so a refused delete leaves it in place
+		// and the caller must hear about it.
+		return h.deleteDirectory(r.Context(), client, bucketPath)
 	})
 
 	if err != nil {

weed/s3api/s3tables/s3tablestest/memfiler.go

@@ -34,6 +34,9 @@ type MemFiler struct {
 	// BeforeUpdate runs once, at the start of the next UpdateEntry, so a test
 	// can land a competing write in a handler's read-to-write window.
 	BeforeUpdate func()
+	// RejectDelete returns the reason DeleteEntry refuses a path, answered the
+	// way the filer answers one: no transport error, the reason in the response.
+	RejectDelete func(dir, name string) string
 }
 
 func newMemFiler() *MemFiler {
@@ -167,6 +170,11 @@ func (f *MemFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryReque
 }
 
 func (f *MemFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
+	if reject := f.RejectDelete; reject != nil {
+		if reason := reject(req.Directory, req.Name); reason != "" {
+			return &filer_pb.DeleteEntryResponse{Error: reason}, nil
+		}
+	}
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	if d, ok := f.entries[req.Directory]; ok {