diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go
index 4752c904e..b1d847fe6 100644
--- a/weed/s3api/filer_multipart.go
+++ b/weed/s3api/filer_multipart.go
@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"cmp"
+	"context"
 	"crypto/md5"
 	"crypto/rand"
 	"encoding/base64"
@@ -936,12 +937,16 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 	}
 
 	if completionState != nil {
+		// The object is already committed and the client is still waiting, so the
+		// cleanup below runs on its own context but spends one allowance between
+		// all of it rather than a retry backoff per unused entry.
+		cleanupCtx := withFilerRetryBudget(context.Background(), filerRetryRequestBudget)
 		for _, deleteEntry := range completionState.deleteEntries {
-			if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
+			if err := s3a.rm(cleanupCtx, uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
 				glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err)
 			}
 		}
-		if err := s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
+		if err := s3a.rm(cleanupCtx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
 			glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
 		}
 		if len(completionState.supersededPartManifests) > 0 {
@@ -955,7 +960,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 // Metadata-only: the version file's chunks are the still-registered parts'
 // chunks, which a retried completion needs.
 func (s3a *S3ApiServer) rollbackMultipartVersion(versionDir, versionFileName string) error {
-	return s3a.rmObject(versionDir, versionFileName, false, false)
+	return s3a.rmObject(context.Background(), versionDir, versionFileName, false, false)
 }
 
 func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) {
@@ -995,7 +1000,7 @@ func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput
 		return nil, s3err.ErrInternalError
 	}
 	if exists {
-		err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
+		err = s3a.rm(context.Background(), s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
 	}
 	if err != nil {
 		glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)
diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go
index 62a1000c3..240e3b18f 100644
--- a/weed/s3api/filer_util.go
+++ b/weed/s3api/filer_util.go
@@ -94,26 +94,36 @@ func listWithRetry(parentDirectoryPath string, doList func() (entries []*filer_p
 
 }
 
-func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
+// A delete is idempotent at the filer, which answers an entry that is already
+// gone with an empty resp.Error, so a reply the transport dropped can be
+// reissued instead of surfaced: as a 500 on the bucket delete, or as a per-key
+// InternalError inside the 200 of a multi-object delete, which no SDK retries.
+// Each attempt re-enters WithFilerClient, so it walks the failover list again
+// on a connection the failed one had invalidated.
+func (s3a *S3ApiServer) rm(ctx context.Context, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
 
-	return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+	return retryFilerOp(ctx, "rm "+parentDirectoryPath+"/"+entryName, func() error {
+		return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 
-		return doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+			return doDeleteEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+		})
 	})
 
 }
 
-func (s3a *S3ApiServer) rmObject(parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
+func (s3a *S3ApiServer) rmObject(ctx context.Context, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
 
-	return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+	return retryFilerOp(ctx, "rmObject "+parentDirectoryPath+"/"+entryName, func() error {
+		return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 
-		return deleteObjectEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+			return deleteObjectEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+		})
 	})
 
 }
 
-func deleteObjectEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
-	err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+func deleteObjectEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
+	err := doDeleteEntry(ctx, client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
 	if err == nil {
 		return nil
 	}
@@ -121,10 +131,10 @@ func deleteObjectEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath,
 		return err
 	}
 
-	return demoteDirectoryMarkerToImplicitDirectory(client, parentDirectoryPath, entryName)
+	return demoteDirectoryMarkerToImplicitDirectory(ctx, client, parentDirectoryPath, entryName)
 }
 
-func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error {
+func doDeleteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error {
 	request := &filer_pb.DeleteEntryRequest{
 		Directory:            parentDirectoryPath,
 		Name:                 entryName,
@@ -134,7 +144,7 @@ func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath strin
 	}
 
 	glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)
-	if resp, err := client.DeleteEntry(context.Background(), request); err != nil {
+	if resp, err := client.DeleteEntry(ctx, request); err != nil {
 		glog.V(1).Infof("delete entry %v: %v", request, err)
 		return fmt.Errorf("delete entry %s/%s: %w", parentDirectoryPath, entryName, err)
 	} else {
@@ -147,8 +157,8 @@ func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath strin
 	return nil
 }
 
-func demoteDirectoryMarkerToImplicitDirectory(client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string) error {
-	resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+func demoteDirectoryMarkerToImplicitDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, parentDirectoryPath, entryName string) error {
+	resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
 		Directory: parentDirectoryPath,
 		Name:      entryName,
 	})
@@ -167,7 +177,7 @@ func demoteDirectoryMarkerToImplicitDirectory(client filer_pb.SeaweedFilerClient
 
 	clearDirectoryMarkerMetadata(resp.Entry)
 
-	if err := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
+	if err := filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
 		Directory: parentDirectoryPath,
 		Entry:     resp.Entry,
 	}); err != nil {
diff --git a/weed/s3api/filer_util_delete_test.go b/weed/s3api/filer_util_delete_test.go
index 47c6b9428..8ee94a4d3 100644
--- a/weed/s3api/filer_util_delete_test.go
+++ b/weed/s3api/filer_util_delete_test.go
@@ -3,7 +3,9 @@ package s3api
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -24,13 +26,34 @@ type deleteObjectEntryTestClient struct {
 	lookupErr  error
 	updateErr  error
 
-	deleteReq *filer_pb.DeleteEntryRequest
-	lookupReq *filer_pb.LookupDirectoryEntryRequest
-	updateReq *filer_pb.UpdateEntryRequest
+	// deleteCallErrs scripts the RPC itself per attempt, deleteRespErrs the
+	// filer's own message: the two ways a delete fails, told apart because
+	// only the first is a transport that dropped a reply.
+	deleteCallErrs []error
+	deleteRespErrs []string
+
+	deleteAttempts int
+	deleteReqs     []*filer_pb.DeleteEntryRequest
+	deleteReq      *filer_pb.DeleteEntryRequest
+	lookupReq      *filer_pb.LookupDirectoryEntryRequest
+	updateReq      *filer_pb.UpdateEntryRequest
 }
 
-func (c *deleteObjectEntryTestClient) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) {
+func (c *deleteObjectEntryTestClient) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) {
+	attempt := c.deleteAttempts
+	c.deleteAttempts++
 	c.deleteReq = req
+	c.deleteReqs = append(c.deleteReqs, req)
+
+	if err := ctx.Err(); err != nil {
+		return nil, status.FromContextError(err).Err()
+	}
+	if attempt < len(c.deleteCallErrs) && c.deleteCallErrs[attempt] != nil {
+		return nil, c.deleteCallErrs[attempt]
+	}
+	if attempt < len(c.deleteRespErrs) && c.deleteRespErrs[attempt] != "" {
+		return &filer_pb.DeleteEntryResponse{Error: c.deleteRespErrs[attempt]}, nil
+	}
 	if c.deleteResp == nil {
 		return &filer_pb.DeleteEntryResponse{}, c.deleteErr
 	}
@@ -77,7 +100,7 @@ func TestDeleteObjectEntryDemotesNonEmptyDirectoryMarker(t *testing.T) {
 		},
 	}
 
-	err := deleteObjectEntry(client, "/buckets/test", "photos", true, false)
+	err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false)
 	require.NoError(t, err)
 	require.NotNil(t, client.lookupReq)
 	require.NotNil(t, client.updateReq)
@@ -116,7 +139,7 @@ func TestDeleteObjectEntryDemotesPrefixObject(t *testing.T) {
 		},
 	}
 
-	require.NoError(t, deleteObjectEntry(client, "/buckets/test", "photos", true, false))
+	require.NoError(t, deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false))
 	require.NotNil(t, client.updateReq)
 
 	updated := client.updateReq.Entry
@@ -140,7 +163,7 @@ func TestDeleteObjectEntryTreatsImplicitDirectoryAsSuccessfulNoop(t *testing.T)
 		},
 	}
 
-	err := deleteObjectEntry(client, "/buckets/test", "photos", true, false)
+	err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false)
 	require.NoError(t, err)
 	require.NotNil(t, client.lookupReq)
 	assert.Nil(t, client.updateReq)
@@ -163,7 +186,7 @@ func TestDeleteObjectEntryIgnoresConcurrentUpdateNotFound(t *testing.T) {
 		updateErr: status.Error(codes.NotFound, "already removed"),
 	}
 
-	err := deleteObjectEntry(client, "/buckets/test", "photos", true, false)
+	err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false)
 	require.NoError(t, err)
 	require.NotNil(t, client.lookupReq)
 	require.NotNil(t, client.updateReq)
@@ -180,7 +203,7 @@ func TestDeleteObjectEntryIgnoresMarkerSpoofedByKey(t *testing.T) {
 		},
 	}
 
-	err := deleteObjectEntry(client, "/buckets/test", name, true, false)
+	err := deleteObjectEntry(context.Background(), client, "/buckets/test", name, true, false)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "disk full")
 	assert.Nil(t, client.lookupReq)
@@ -192,9 +215,124 @@ func TestDeleteObjectEntryPropagatesNonDirectoryDeleteErrors(t *testing.T) {
 		deleteErr: errors.New("boom"),
 	}
 
-	err := deleteObjectEntry(client, "/buckets/test", "photos", true, false)
+	err := deleteObjectEntry(context.Background(), client, "/buckets/test", "photos", true, false)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "boom")
 	assert.Nil(t, client.lookupReq)
 	assert.Nil(t, client.updateReq)
 }
+
+// A delete is idempotent at the filer, so a reply the transport dropped is
+// reissued rather than surfaced. Issue #7204 saw the surfaced version as a 500
+// on the bucket delete, which boto3 resent and then answered NoSuchBucket;
+// issue #7224 saw it as a per-key InternalError inside a 200, which no SDK
+// retries, so the object silently stayed.
+func TestDeleteUnversionedObjectReplaysADroppedReply(t *testing.T) {
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	client := &deleteObjectEntryTestClient{
+		deleteCallErrs: []error{status.Error(codes.Unavailable, "transport is closing"), nil},
+	}
+
+	err := s3a.deleteUnversionedObjectWithClient(testRetryCtx(10*time.Millisecond), client, "mybucket", "a/b/c.txt", false)
+
+	require.NoError(t, err)
+	assert.Equal(t, 2, client.deleteAttempts)
+	// the replay must ask for the same delete, not a differently shaped one
+	require.Len(t, client.deleteReqs, 2)
+	assert.Equal(t, client.deleteReqs[0].String(), client.deleteReqs[1].String())
+}
+
+// The filer formats the deleted path, and for a recursive delete the child it
+// stopped on, into the message it sends back, so both are the client's text.
+// Nothing here reads that text, and this pins it: the same failure under a key
+// named after a transport condition is replayed exactly as often as under a
+// plain one.
+func TestDeleteUnversionedObjectReplayIgnoresTheObjectName(t *testing.T) {
+	attemptsFor := func(key string) int {
+		s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+		client := &deleteObjectEntryTestClient{
+			deleteRespErrs: []string{"delete file /buckets/b/" + key + ": permission denied", ""},
+		}
+		require.NoError(t, s3a.deleteUnversionedObjectWithClient(testRetryCtx(10*time.Millisecond), client, "b", key, false))
+		return client.deleteAttempts
+	}
+
+	plain := attemptsFor("plain.log")
+	for _, key := range []string{"transport.log", "unavailable.txt", "slowdown.csv", filer_pb.ErrNotFound.Error()} {
+		assert.Equal(t, plain, attemptsFor(key), "key %q must not steer the replay", key)
+	}
+}
+
+// A folder the filer refused because it still has children is an answer about
+// the tree, not a hiccup: it is handed straight to the demote path instead of
+// being replayed until the allowance runs out.
+func TestDeleteUnversionedObjectDoesNotReplayANonEmptyFolder(t *testing.T) {
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	client := &deleteObjectEntryTestClient{
+		deleteResp: &filer_pb.DeleteEntryResponse{
+			Error: filer.MsgFailDelNonEmptyFolder + ": /buckets/b/photos",
+		},
+	}
+
+	require.NoError(t, s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "photos", false))
+	assert.Equal(t, 1, client.deleteAttempts)
+	assert.NotNil(t, client.lookupReq, "the demote must still run")
+}
+
+// A caller that has gone away is not worth replaying for.
+func TestDeleteUnversionedObjectDoesNotReplayForACancelledCaller(t *testing.T) {
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	client := &deleteObjectEntryTestClient{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := s3a.deleteUnversionedObjectWithClient(ctx, client, "b", "k", false)
+
+	require.Error(t, err)
+	assert.Equal(t, 1, client.deleteAttempts)
+}
+
+// The client picks how many keys a multi-object delete carries, so the backoff
+// is drawn from one allowance held by the request rather than one per key.
+func TestDeleteUnversionedObjectDrawsFromTheRequestAllowance(t *testing.T) {
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	ctx := testRetryCtx(20 * time.Millisecond)
+
+	start := time.Now()
+	var attempts int
+	for i := 0; i < 50; i++ {
+		client := &deleteObjectEntryTestClient{deleteErr: status.Error(codes.Unavailable, "transport is closing")}
+		require.Error(t, s3a.deleteUnversionedObjectWithClient(ctx, client, "b", fmt.Sprintf("k%d", i), false))
+		attempts += client.deleteAttempts
+	}
+
+	assert.Less(t, time.Since(start), time.Second, "one allowance, not one per key")
+	assert.Greater(t, attempts, 50, "the allowance must still buy a replay")
+}
+
+// testRetryCtx caps the backoff so a test exercises the replay without waiting
+// out the real per-request allowance.
+func testRetryCtx(allowance time.Duration) context.Context {
+	return withFilerRetryBudget(context.Background(), allowance)
+}
+
+func TestIsRetryableFilerErr(t *testing.T) {
+	assert.False(t, isRetryableFilerErr(nil))
+	assert.True(t, isRetryableFilerErr(status.Error(codes.Unavailable, "transport is closing")))
+	assert.True(t, isRetryableFilerErr(fmt.Errorf("delete entry /b/k: %w", status.Error(codes.Unavailable, "x"))))
+	// the filer's own message carries no status, and a store error may well clear
+	assert.True(t, isRetryableFilerErr(filer.DeleteEntryError("delete file /b/k: filer store delete: disk full")))
+
+	assert.False(t, isRetryableFilerErr(filer.DeleteEntryError(filer.MsgFailDelNonEmptyFolder+": /b/photos")))
+	assert.False(t, isRetryableFilerErr(filer_pb.ErrNotFound))
+	assert.False(t, isRetryableFilerErr(status.Error(codes.NotFound, "gone")))
+	assert.False(t, isRetryableFilerErr(context.Canceled))
+	assert.False(t, isRetryableFilerErr(status.Error(codes.Canceled, "context canceled")))
+	assert.False(t, isRetryableFilerErr(status.Error(codes.DeadlineExceeded, "context deadline exceeded")))
+
+	// the message is never read, so the same code classifies the same way
+	// whatever the client named the key
+	assert.Equal(t,
+		isRetryableFilerErr(status.Error(codes.PermissionDenied, "denied for /b/plain.log")),
+		isRetryableFilerErr(status.Error(codes.PermissionDenied, "denied for /b/transport.log")))
+}
diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go
index 5cc7afd49..5b335089e 100644
--- a/weed/s3api/s3api_bucket_handlers.go
+++ b/weed/s3api/s3api_bucket_handlers.go
@@ -370,7 +370,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 	// This ensures we don't leave a bucket without the requested Object Lock configuration
 	if objectLockSetupError != nil {
 		glog.Errorf("PutBucketHandler: rolling back bucket %s creation due to Object Lock setup failure: %v", bucket, objectLockSetupError)
-		if deleteErr := s3a.rm(s3a.option.BucketsPath, bucket, true, true); deleteErr != nil {
+		if deleteErr := s3a.rm(context.Background(), s3a.option.BucketsPath, bucket, true, true); deleteErr != nil {
 			glog.Errorf("PutBucketHandler: failed to rollback bucket %s after Object Lock setup failure: %v", bucket, deleteErr)
 		}
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -414,6 +414,9 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 
 	bucket, _ := s3_constants.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteBucketHandler %s", bucket)
+	// The teardown below retries, and a failover walk repeats it once per
+	// filer, so the backoff comes out of one allowance held here.
+	r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget))
 
 	if s3a.isTableBucket(bucket) {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
@@ -463,7 +466,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 	// the "collection exists but bucket directory missing" inconsistency that blocks
 	// bucket recreation. An orphaned collection is harmless and will be cleaned up
 	// or reused when the bucket is recreated.
-	err := s3a.rm(s3a.option.BucketsPath, bucket, false, true)
+	err := s3a.rm(r.Context(), s3a.option.BucketsPath, bucket, false, true)
 	if err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
diff --git a/weed/s3api/s3api_bucket_owner_index.go b/weed/s3api/s3api_bucket_owner_index.go
index 977aa39f4..b69cbc66a 100644
--- a/weed/s3api/s3api_bucket_owner_index.go
+++ b/weed/s3api/s3api_bucket_owner_index.go
@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"net/http"
 	"net/url"
 	"slices"
@@ -67,7 +68,7 @@ func (s3a *S3ApiServer) addBucketToOwnerIndex(owner, bucket string, crtime int64
 }
 
 func (s3a *S3ApiServer) removeBucketFromOwnerIndex(owner, bucket string) error {
-	return s3a.rm(s3a.bucketOwnerDir(owner), bucket, false, false)
+	return s3a.rm(context.Background(), s3a.bucketOwnerDir(owner), bucket, false, false)
 }
 
 // maintainBucketOwnerIndex applies a /buckets metadata event to the owner
diff --git a/weed/s3api/s3api_directory_marker.go b/weed/s3api/s3api_directory_marker.go
index 41577bae7..0c97382a5 100644
--- a/weed/s3api/s3api_directory_marker.go
+++ b/weed/s3api/s3api_directory_marker.go
@@ -93,7 +93,7 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object st
 				break
 			}
 		}
-		if rmErr := s3a.rm(markerDir, s3_constants.VersionsFolder, true, true); rmErr != nil {
+		if rmErr := s3a.rm(r.Context(), markerDir, s3_constants.VersionsFolder, true, true); rmErr != nil {
 			glog.Errorf("deleteDirectoryMarker: failed to remove stale history of %s/%s: %v", bucket, object, rmErr)
 			return s3err.ErrInternalError
 		}
@@ -103,7 +103,7 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object st
 	}
 
 	if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
+		return s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object, false)
 	}); err != nil {
 		glog.Errorf("deleteDirectoryMarker: failed to delete %s/%s: %v", bucket, object, err)
 		return s3err.ErrInternalError
diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go
index 164a999ba..f32c478c7 100644
--- a/weed/s3api/s3api_internal_lifecycle.go
+++ b/weed/s3api/s3api_internal_lifecycle.go
@@ -88,7 +88,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 			return done(), nil
 		default:
 			err := s3a.WithFilerClient(false, func(c filer_pb.SeaweedFilerClient) error {
-				return s3a.deleteUnversionedObjectWithClient(c, req.Bucket, req.ObjectPath, metadataOnly)
+				return s3a.deleteUnversionedObjectWithClient(ctx, c, req.Bucket, req.ObjectPath, metadataOnly)
 			})
 			if err != nil {
 				if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrObjectNotFound) {
@@ -179,7 +179,7 @@ func (s3a *S3ApiServer) lifecycleAbortMPU(ctx context.Context, req *s3_lifecycle
 	if !exists {
 		return noopResolved("NOT_FOUND"), nil
 	}
-	if err := s3a.rm(uploadsFolder, uploadID, true, true); err != nil {
+	if err := s3a.rm(ctx, uploadsFolder, uploadID, true, true); err != nil {
 		if errors.Is(err, filer_pb.ErrNotFound) {
 			return noopResolved("NOT_FOUND_AT_DELETE"), nil
 		}
diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go
index 775852c35..f1e69b6e9 100644
--- a/weed/s3api/s3api_object_handlers_copy.go
+++ b/weed/s3api/s3api_object_handlers_copy.go
@@ -643,7 +643,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
 func (s3a *S3ApiServer) rollbackCopyVersion(bucketDir, versionObjectPath string) error {
 	versionPath := util.FullPath(fmt.Sprintf("%s/%s", bucketDir, versionObjectPath))
 	versionDir, versionName := versionPath.DirAndName()
-	return s3a.rmObject(versionDir, versionName, true, false)
+	return s3a.rmObject(context.Background(), versionDir, versionName, true, false)
 }
 
 func (s3a *S3ApiServer) resolveCopySourceEntry(bucket, object, versionId, versioningState string) (*filer_pb.Entry, error) {
@@ -1090,7 +1090,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 	// Save the part entry to the multipart uploads folder
 	// Check if part exists and remove it first (allow re-copying same part)
 	if exists, _ := s3a.exists(uploadDir, partName, false); exists {
-		if err := s3a.rm(uploadDir, partName, false, false); err != nil {
+		if err := s3a.rm(r.Context(), uploadDir, partName, false, false); err != nil {
 			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 			return
 		}
diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go
index 1bcc05a25..7a4968fa9 100644
--- a/weed/s3api/s3api_object_handlers_copy_part_sse.go
+++ b/weed/s3api/s3api_object_handlers_copy_part_sse.go
@@ -415,7 +415,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
 func (s3a *S3ApiServer) writeEmptyCopyPart(dstBucket, uploadID string, partID int) (string, s3err.ErrorCode) {
 	uploadDir, partName := s3a.copyPartLocation(dstBucket, uploadID, partID)
 	if exists, _ := s3a.exists(uploadDir, partName, false); exists {
-		if err := s3a.rm(uploadDir, partName, false, false); err != nil {
+		if err := s3a.rm(context.Background(), uploadDir, partName, false, false); err != nil {
 			return "", s3err.ErrInternalError
 		}
 	}
diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go
index 5f42bc9f5..a21a73fe2 100644
--- a/weed/s3api/s3api_object_handlers_delete.go
+++ b/weed/s3api/s3api_object_handlers_delete.go
@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"encoding/xml"
 	"errors"
 	"io"
@@ -189,19 +190,26 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v
 // relies on the volume's natural TTL to reclaim chunks; pass true only
 // when the entry's Attributes.TtlSec > 0 so the volume is guaranteed to
 // drop the chunks on its own.
-func (s3a *S3ApiServer) deleteUnversionedObjectWithClient(client filer_pb.SeaweedFilerClient, bucket, object string, metadataOnly bool) error {
+func (s3a *S3ApiServer) deleteUnversionedObjectWithClient(ctx context.Context, client filer_pb.SeaweedFilerClient, bucket, object string, metadataOnly bool) error {
 	if !s3_constants.IsValidBucketName(bucket) || !s3_constants.IsValidObjectKey(object) {
 		return errors.New("invalid bucket or object path")
 	}
 	target := util.NewFullPath(s3a.bucketDir(bucket), object)
 	dir, name := target.DirAndName()
-	return deleteObjectEntry(client, dir, name, !metadataOnly, false)
+	// The caller holds one client for a whole batch, so a dropped reply is
+	// replayed on that client rather than by re-entering WithFilerClient.
+	return retryFilerOp(ctx, "delete "+string(target), func() error {
+		return deleteObjectEntry(ctx, client, dir, name, !metadataOnly, false)
+	})
 }
 
 func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := s3_constants.GetBucketAndObject(r)
 	glog.Infof("DeleteObjectHandler %s %s", bucket, object)
+	// The filer ops below each retry, and a failover walk runs the whole set
+	// once per filer, so the backoff comes out of one allowance held here.
+	r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget))
 	if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 		return
@@ -326,7 +334,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 			}
 
 			if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-				return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
+				return s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object, false)
 			}); err != nil {
 				glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err)
 				return s3err.ErrInternalError
@@ -493,7 +501,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 					return s3a.deleteDirectoryMarker(r, bucket, object.Key)
 				}
 
-				if err := s3a.deleteUnversionedObjectWithClient(client, bucket, object.Key, false); err != nil {
+				if err := s3a.deleteUnversionedObjectWithClient(r.Context(), client, bucket, object.Key, false); err != nil {
 					glog.Errorf("DeleteMultipleObjectsHandler: failed to delete %s/%s: %v", bucket, object.Key, err)
 					return s3err.ErrInternalError
 				}
diff --git a/weed/s3api/s3api_object_handlers_delete_test.go b/weed/s3api/s3api_object_handlers_delete_test.go
index b2104e257..dd680d6c0 100644
--- a/weed/s3api/s3api_object_handlers_delete_test.go
+++ b/weed/s3api/s3api_object_handlers_delete_test.go
@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -47,7 +48,7 @@ func TestDeleteUnversionedObjectWithClient_MetadataOnlySkipsChunkDelete(t *testi
 	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
 	client := &deleteObjectEntryTestClient{}
 
-	err := s3a.deleteUnversionedObjectWithClient(client, "b", "k", true)
+	err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", true)
 	require.NoError(t, err)
 	require.NotNil(t, client.deleteReq)
 	assert.Equal(t, "/buckets/b", client.deleteReq.Directory)
@@ -61,7 +62,7 @@ func TestDeleteUnversionedObjectWithClient_FullDeletePreservesIsDeleteData(t *te
 	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
 	client := &deleteObjectEntryTestClient{}
 
-	err := s3a.deleteUnversionedObjectWithClient(client, "b", "k", false)
+	err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", false)
 	require.NoError(t, err)
 	require.NotNil(t, client.deleteReq)
 	assert.True(t, client.deleteReq.IsDeleteData, "default delete must keep IsDeleteData true")
@@ -75,7 +76,7 @@ func TestDeleteUnversionedObjectWithClient_FullPathFromBucketsRoot(t *testing.T)
 	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
 	client := &deleteObjectEntryTestClient{}
 
-	err := s3a.deleteUnversionedObjectWithClient(client, "mybucket", "a/b/c.txt", false)
+	err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "mybucket", "a/b/c.txt", false)
 	require.NoError(t, err)
 	require.NotNil(t, client.deleteReq)
 	assert.Equal(t, "/buckets/mybucket/a/b", client.deleteReq.Directory)
@@ -86,7 +87,7 @@ func TestDeleteUnversionedObjectWithClientRejectsTraversal(t *testing.T) {
 	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
 	client := &deleteObjectEntryTestClient{}
 
-	err := s3a.deleteUnversionedObjectWithClient(client, "source-bucket", "../victim-bucket/secret", false)
+	err := s3a.deleteUnversionedObjectWithClient(context.Background(), client, "source-bucket", "../victim-bucket/secret", false)
 
 	require.Error(t, err)
 	assert.Nil(t, client.deleteReq, "invalid path must be rejected before a filer delete RPC")
@@ -102,7 +103,7 @@ func TestDeleteUnversionedObjectWithClient_PropagatesEntryAttributesIrrelevant(t
 		deleteResp: &filer_pb.DeleteEntryResponse{},
 	}
 
-	require.NoError(t, s3a.deleteUnversionedObjectWithClient(client, "b", "k", true))
+	require.NoError(t, s3a.deleteUnversionedObjectWithClient(context.Background(), client, "b", "k", true))
 	require.NotNil(t, client.deleteReq)
 	assert.False(t, client.deleteReq.IsDeleteData)
 }
diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index a60231551..53a857272 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/weed/s3api/s3api_object_handlers_put.go
@@ -933,7 +933,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 		entryCreated = true
 		if finalize != nil && finalize.afterCreate != nil {
 			if afterCreateCode := finalize.afterCreate(entry); afterCreateCode != s3err.ErrNone {
-				rollbackErr = s3a.rmObject(path.Dir(filePath), path.Base(filePath), true, false)
+				rollbackErr = s3a.rmObject(context.Background(), path.Dir(filePath), path.Base(filePath), true, false)
 				if rollbackErr != nil {
 					glog.Errorf("putToFiler: failed to rollback created entry for %s after post-create error: %v", filePath, rollbackErr)
 				} else {
@@ -1470,7 +1470,7 @@ func (s3a *S3ApiServer) removeNullVersionFile(bucket, object string) {
 		if string(entry.Extended[s3_constants.ExtVersionIdKey]) != "null" {
 			continue
 		}
-		if rmErr := s3a.rm(versionsDir, entry.Name, true, false); rmErr != nil {
+		if rmErr := s3a.rm(context.Background(), versionsDir, entry.Name, true, false); rmErr != nil {
 			glog.Warningf("removeNullVersionFile: %s/%s: %v", bucket, object, rmErr)
 		}
 		return
diff --git a/weed/s3api/s3api_object_handlers_rename.go b/weed/s3api/s3api_object_handlers_rename.go
index a2275c89a..d4d45112c 100644
--- a/weed/s3api/s3api_object_handlers_rename.go
+++ b/weed/s3api/s3api_object_handlers_rename.go
@@ -289,7 +289,7 @@ func (s3a *S3ApiServer) renameKeyHoldingNestedKeys(bucket, srcObject, dstObject
 
 	// 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 {
+	if err := s3a.rmObject(context.Background(), srcDir, srcName, true, false); err != nil {
 		glog.Errorf("RenameObject %s: strip %s: %v", bucket, srcObject, err)
 		return s3err.ErrInternalError
 	}
diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go
index 44ce312c5..29986cbbd 100644
--- a/weed/s3api/s3api_object_versioning.go
+++ b/weed/s3api/s3api_object_versioning.go
@@ -1144,7 +1144,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket,
 		// 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)
+		deleteErr := s3a.rmObject(ctx, dir, name, !metadataOnly, false)
 		if deleteErr != nil {
 			// Check if file was already deleted by another process
 			if _, checkErr := s3a.getEntry(bucketDir, normalizedObject); checkErr != nil {
@@ -1193,7 +1193,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket,
 	// Attempt to delete the version file
 	// Note: We don't check if the file exists first to avoid race conditions
 	// The deletion operation should be idempotent
-	deleteErr := s3a.rm(versionsDir, versionFile, !metadataOnly, false)
+	deleteErr := s3a.rm(ctx, versionsDir, versionFile, !metadataOnly, false)
 	if deleteErr != nil {
 		// Check if file was already deleted by another process (race condition handling)
 		if _, checkErr := s3a.getEntry(versionsDir, versionFile); checkErr != nil {
@@ -1212,7 +1212,7 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(ctx context.Context, bucket,
 		// down. Non-recursive: any orphan from older code paths leaves
 		// the directory in place for the empty-folder cleaner or our
 		// reconciler to handle.
-		if rmErr := s3a.rm(s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder, true, false); rmErr != nil {
+		if rmErr := s3a.rm(ctx, s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder, true, false); rmErr != nil {
 			glog.V(2).Infof("deleteSpecificObjectVersion: deferring .versions/ teardown for %s/%s: %v", bucket, normalizedObject, rmErr)
 		}
 	case isLatestVersion && !prePointerRolled:
@@ -1442,6 +1442,8 @@ func (b *filerRetryBudget) take(d time.Duration) (time.Duration, bool) {
 //   - NotFound: the entry genuinely doesn't exist. Retrying won't make
 //     it appear, and callers (e.g. repointLatestBeforeDeletion) want
 //     to act on this directly.
+//   - non-empty folder: the filer looked and the children are there, so
+//     the answer will not change; callers act on it directly too.
 //   - context.Canceled / DeadlineExceeded: the request was aborted by
 //     the client or hit a deadline. Continuing to retry just delays
 //     the failure return.
@@ -1452,12 +1454,18 @@ func isRetryableFilerErr(err error) bool {
 	if err == nil {
 		return false
 	}
-	if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound {
+	if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, filer.ErrNonEmptyFolder) {
 		return false
 	}
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return false
 	}
+	// the same conditions once they have crossed gRPC, where a cancelled
+	// context arrives as a status and no longer matches the sentinel above
+	switch status.Code(err) {
+	case codes.NotFound, codes.Canceled, codes.DeadlineExceeded:
+		return false
+	}
 	return true
 }
 
@@ -1650,41 +1658,28 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(ctx context.Context, bu
 		// object is correctly absent.
 		glog.V(2).Infof("updateLatestVersionAfterDeletion: no versions left for %s/%s, deleting .versions directory", bucket, object)
 
-		rmErr := s3a.rm(bucketDir, versionsObjectPath, true, false)
+		rmErr := s3a.rm(ctx, bucketDir, versionsObjectPath, true, false)
 		if rmErr == nil {
 			return nil
 		}
-		// Two ways rm can fail here: "non-empty folder" (orphan entries
-		// blocking the teardown — fall through to pointer clear) and a
-		// transient filer error (worth retrying). Distinguish by the
-		// sentinel; if we can't tell, treat as transient.
+		// "non-empty folder" means orphan entries are blocking the
+		// teardown — fall through to the pointer clear. Anything else
+		// has already exhausted rm's own retries. Either way we still
+		// clear the stale pointer so readers get a clean miss; the
+		// directory can be tidied by the reconciler later.
 		if errors.Is(rmErr, filer.ErrNonEmptyFolder) {
 			glog.V(2).Infof("updateLatestVersionAfterDeletion: .versions/ for %s/%s still has orphan entries: %v", bucket, object, rmErr)
 			s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
 			return nil
 		}
-		// Transient — retry the rm a few times before giving up. Even
-		// if it ultimately fails, we still clear the stale pointer so
-		// readers get a clean miss; the directory can be tidied by the
-		// reconciler later.
-		retryErr := retryFilerOp(ctx, "updateLatestVersionAfterDeletion.rm", func() error {
-			return s3a.rm(bucketDir, versionsObjectPath, true, false)
-		})
-		if retryErr == nil {
-			return nil
-		}
-		if errors.Is(retryErr, filer.ErrNonEmptyFolder) {
-			s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
-			return nil
-		}
-		versioningHealWarningf("teardown_failed", "bucket=%s key=%s err=%v (fell through to clearStale)", bucket, object, retryErr)
+		versioningHealWarningf("teardown_failed", "bucket=%s key=%s err=%v (fell through to clearStale)", bucket, object, rmErr)
 		if s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion") {
 			// Pointer is consistent again; reader will get NoSuchKey via
 			// the clean-miss path. Don't emit `produced` or enqueue the
 			// reconciler — there's no stranded state left to heal.
 			return nil
 		}
-		return fmt.Errorf("delete .versions directory: %w", retryErr)
+		return fmt.Errorf("delete .versions directory: %w", rmErr)
 	}
 
 	return nil
@@ -2067,7 +2062,7 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s
 		// non-recursive rm makes this safe against a concurrent PUT — a new
 		// child fails the rm, and a directory removed just before that PUT's
 		// version file lands is recreated by the create's parent handling.
-		if rmErr := s3a.rm(bucketDir, versionsObjectPath, true, false); rmErr == nil {
+		if rmErr := s3a.rm(context.Background(), bucketDir, versionsObjectPath, true, false); rmErr == nil {
 			versioningHealInfof("healed", "bucket=%s key=%s mode=empty_dir_removed", bucket, normalizedObject)
 		} else {
 			// Orphan entries (files in .versions/ that lack the version-id