diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go
index cbffcced0..d7be43fd7 100644
--- a/weed/filer/filer_delete_entry.go
+++ b/weed/filer/filer_delete_entry.go
@@ -2,7 +2,9 @@ package filer
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -15,6 +17,40 @@ const (
 	MsgFailDelNonEmptyFolder = "fail to delete non-empty folder"
 )
 
+// ErrNonEmptyFolder is a non-recursive delete refused because the folder still
+// has children. The marker leads the message the filer builds for it and is
+// never wrapped on the way out, so the path that follows it, which the client
+// chose, cannot forge one.
+var ErrNonEmptyFolder = errors.New(MsgFailDelNonEmptyFolder)
+
+// DeleteEntryError turns the text of DeleteEntryResponse.Error back into an
+// error carrying the condition the filer reported. Call it on the response
+// field, before formatting a path around it.
+func DeleteEntryError(msg string) error {
+	if strings.HasPrefix(msg, MsgFailDelNonEmptyFolder) {
+		return &deleteEntryError{msg: msg, cause: ErrNonEmptyFolder}
+	}
+	return errors.New(msg)
+}
+
+// IsNonEmptyFolderError is for callers holding a delete failure that has not
+// been wrapped yet: the sentinel when it survived, the leading marker when the
+// error only crossed the wire as text.
+func IsNonEmptyFolderError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return errors.Is(err, ErrNonEmptyFolder) || strings.HasPrefix(err.Error(), MsgFailDelNonEmptyFolder)
+}
+
+type deleteEntryError struct {
+	msg   string
+	cause error
+}
+
+func (e *deleteEntryError) Error() string { return e.msg }
+func (e *deleteEntryError) Unwrap() error { return e.cause }
+
 type OnChunksFunc func([]*filer_pb.FileChunk) error
 type OnHardLinkIdsFunc func([]HardLinkId) error
 
@@ -43,6 +79,9 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 		})
 		if err != nil {
 			glog.V(2).InfofCtx(ctx, "delete directory %s: %v", p, err)
+			if errors.Is(err, ErrNonEmptyFolder) {
+				return err
+			}
 			return fmt.Errorf("delete directory %s: %v", p, err)
 		}
 	}
@@ -91,7 +130,7 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 			if lastFileName == "" && !isRecursive && len(entries) > 0 {
 				// only for first iteration in the loop
 				glog.V(2).InfofCtx(ctx, "deleting a folder %s has children: %+v ...", entry.FullPath, entries[0].Name())
-				return fmt.Errorf("%s: %s", MsgFailDelNonEmptyFolder, entry.FullPath)
+				return fmt.Errorf("%w: %s", ErrNonEmptyFolder, entry.FullPath)
 			}
 
 			for _, sub := range entries {
diff --git a/weed/filer/filer_delete_entry_test.go b/weed/filer/filer_delete_entry_test.go
new file mode 100644
index 000000000..e62ac7882
--- /dev/null
+++ b/weed/filer/filer_delete_entry_test.go
@@ -0,0 +1,29 @@
+package filer
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+// The path in a non-empty-folder failure is the client's, so the marker has to
+// lead the message for a caller reading it off the wire to trust it.
+func TestNonEmptyFolderClassification(t *testing.T) {
+	err := fmt.Errorf("%w: %s", ErrNonEmptyFolder, "/buckets/b/photos")
+	if !IsNonEmptyFolderError(err) {
+		t.Errorf("expected the sentinel to be recognized: %v", err)
+	}
+	if !errors.Is(DeleteEntryError(err.Error()), ErrNonEmptyFolder) {
+		t.Errorf("expected the wire text to classify: %v", err)
+	}
+
+	// an entry named after the marker cannot forge one: every other delete
+	// failure the filer builds leads with its own wrapper
+	spoofed := "delete file /buckets/b/" + MsgFailDelNonEmptyFolder + ": filer store delete: disk full"
+	if errors.Is(DeleteEntryError(spoofed), ErrNonEmptyFolder) {
+		t.Errorf("expected no forgery from the entry name: %v", spoofed)
+	}
+	if IsNonEmptyFolderError(errors.New(spoofed)) {
+		t.Errorf("expected no forgery from the entry name: %v", spoofed)
+	}
+}
diff --git a/weed/mount/weedfs_dir_mkrm.go b/weed/mount/weedfs_dir_mkrm.go
index 1989bb309..26ba969d2 100644
--- a/weed/mount/weedfs_dir_mkrm.go
+++ b/weed/mount/weedfs_dir_mkrm.go
@@ -3,7 +3,6 @@ package mount
 import (
 	"context"
 	"os"
-	"strings"
 	"syscall"
 	"time"
 
@@ -161,7 +160,7 @@ func (wfs *WFS) Rmdir(cancel <-chan struct{}, header *fuse.InHeader, name string
 	resp, err := wfs.streamDeleteEntry(context.Background(), deleteReq)
 	if err != nil {
 		glog.V(1).Infof("remove %s: %v", entryFullPath, err)
-		if strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
+		if filer.IsNonEmptyFolderError(err) {
 			return fuse.Status(syscall.ENOTEMPTY)
 		}
 		return fuse.ENOENT
diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go
index 7174d807e..aa2425885 100644
--- a/weed/pb/filer_pb/filer_client.go
+++ b/weed/pb/filer_pb/filer_client.go
@@ -155,7 +155,9 @@ func DoSeaweedListWithSnapshot(ctx context.Context, client SeaweedFilerClient, f
 	defer cancel()
 	stream, err := client.ListEntries(ctx, request)
 	if err != nil {
-		return actualSnapshotTsNs, fmt.Errorf("list %s: %v", fullDirPath, err)
+		// fullDirPath is the caller's bucket and prefix; keep the status the filer
+		// sent so a retry is decided on that rather than on this message
+		return actualSnapshotTsNs, fmt.Errorf("list %s: %w", fullDirPath, err)
 	}
 
 	var prevEntry *Entry
diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go
index 74a9680a0..62a1000c3 100644
--- a/weed/s3api/filer_util.go
+++ b/weed/s3api/filer_util.go
@@ -67,10 +67,10 @@ const (
 	listRetryInitialBackoff = 100 * time.Millisecond
 )
 
-// isRetryableListError classifies by message via util.IsTransientError because
-// DoSeaweedListWithSnapshot wraps a failed ListEntries call with %v, dropping
-// the gRPC status from the chain. Not-found is authoritative and must reach the
-// caller unchanged.
+// isRetryableListError defers to util.IsTransientError, which reads the status
+// the filer sent rather than the message DoSeaweedListWithSnapshot builds around
+// it out of the bucket and prefix the client chose. Not-found is authoritative
+// and must reach the caller unchanged.
 func isRetryableListError(err error) bool {
 	return err != nil && !isFilerNotFound(err) && util.IsTransientError(err)
 }
@@ -117,7 +117,7 @@ func deleteObjectEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath,
 	if err == nil {
 		return nil
 	}
-	if !strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
+	if !errors.Is(err, filer.ErrNonEmptyFolder) {
 		return err
 	}
 
@@ -136,10 +136,12 @@ 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 {
 		glog.V(1).Infof("delete entry %v: %v", request, err)
-		return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, err)
+		return fmt.Errorf("delete entry %s/%s: %w", parentDirectoryPath, entryName, err)
 	} else {
 		if resp.Error != "" {
-			return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, resp.Error)
+			// the path wrapped in here is the client's, so classify the filer's
+			// text now, while it still stands alone
+			return fmt.Errorf("delete entry %s/%s: %w", parentDirectoryPath, entryName, filer.DeleteEntryError(resp.Error))
 		}
 	}
 	return nil
diff --git a/weed/s3api/filer_util_delete_test.go b/weed/s3api/filer_util_delete_test.go
index 68389daa0..47c6b9428 100644
--- a/weed/s3api/filer_util_delete_test.go
+++ b/weed/s3api/filer_util_delete_test.go
@@ -169,6 +169,24 @@ func TestDeleteObjectEntryIgnoresConcurrentUpdateNotFound(t *testing.T) {
 	require.NotNil(t, client.updateReq)
 }
 
+// The key is the client's and the filer echoes it in the message it sends back,
+// so a key named after the marker must not turn a real failure into the demote
+// no-op, which would answer a failed delete with a 204.
+func TestDeleteObjectEntryIgnoresMarkerSpoofedByKey(t *testing.T) {
+	name := filer.MsgFailDelNonEmptyFolder
+	client := &deleteObjectEntryTestClient{
+		deleteResp: &filer_pb.DeleteEntryResponse{
+			Error: "delete file /buckets/test/" + name + ": filer store delete: disk full",
+		},
+	}
+
+	err := deleteObjectEntry(client, "/buckets/test", name, true, false)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "disk full")
+	assert.Nil(t, client.lookupReq)
+	assert.Nil(t, client.updateReq)
+}
+
 func TestDeleteObjectEntryPropagatesNonDirectoryDeleteErrors(t *testing.T) {
 	client := &deleteObjectEntryTestClient{
 		deleteErr: errors.New("boom"),
diff --git a/weed/s3api/filer_util_list_retry_test.go b/weed/s3api/filer_util_list_retry_test.go
index 8a89f89e9..dc3280805 100644
--- a/weed/s3api/filer_util_list_retry_test.go
+++ b/weed/s3api/filer_util_list_retry_test.go
@@ -248,3 +248,50 @@ func TestIsRetryableListError(t *testing.T) {
 	assert.False(t, isRetryableListError(status.Error(codes.PermissionDenied, "not allowed")))
 	assert.False(t, isRetryableListError(context.Canceled))
 }
+
+// The bucket and the prefix are the client's to choose and they land inside the
+// message DoSeaweedListWithSnapshot builds, so they must not decide whether the
+// failure is retried.
+func TestListWithRetryIgnoresUserControlledPath(t *testing.T) {
+	for name, listErr := range map[string]error{
+		"permission denied": status.Error(codes.PermissionDenied, "not allowed"),
+		"invalid argument":  status.Error(codes.InvalidArgument, "bad prefix"),
+	} {
+		t.Run(name, func(t *testing.T) {
+			client := &listRetryClient{
+				entries:  testUploadEntries("upload-1"),
+				callErrs: []error{listErr, nil},
+			}
+			accessor := &listRetryAccessor{client: client}
+
+			_, _, err := listWithRetry("/buckets/transport/unavailable", func() ([]*filer_pb.Entry, bool, error) {
+				return listOnce(accessor, "/buckets/transport/unavailable")
+			})
+
+			require.Error(t, err)
+			assert.Equal(t, 1, client.attempts, "a bucket name must not make a non-transient error look transient")
+		})
+	}
+}
+
+// The inverse: a prefix carrying the not-found sentence must not suppress a
+// retry the filer actually asked for.
+func TestListWithRetryIgnoresSpoofedNotFoundPath(t *testing.T) {
+	client := &listRetryClient{
+		entries: testUploadEntries("upload-1"),
+		callErrs: []error{
+			status.Error(codes.Unavailable, "filer is restarting"),
+			nil,
+		},
+	}
+	accessor := &listRetryAccessor{client: client}
+
+	dir := "/buckets/b/filer: no entry is found in filer store"
+	entries, _, err := listWithRetry(dir, func() ([]*filer_pb.Entry, bool, error) {
+		return listOnce(accessor, dir)
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, []string{"upload-1"}, entryNames(entries))
+	assert.Equal(t, 2, client.attempts, "an object name must not suppress a retry")
+}
diff --git a/weed/s3api/s3_error_utils.go b/weed/s3api/s3_error_utils.go
index 13e499cee..e6eed6b56 100644
--- a/weed/s3api/s3_error_utils.go
+++ b/weed/s3api/s3_error_utils.go
@@ -7,20 +7,26 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 // isFilerNotFound reports whether a filer error is a not-found.
 // Unlike lookups (normalized in filer_pb.LookupEntry), list and cache errors
 // cross gRPC as raw status errors, so the sentinel survives as codes.NotFound
-// or only as text. Callers must pass errors from paths without user-controlled
-// segments that could spoof it.
+// or only as text. The text is matched last and only on what the filer itself
+// said, since callers wrap these errors with a path the client chose.
 func isFilerNotFound(err error) bool {
 	if err == nil {
 		return false
 	}
-	return errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound || strings.Contains(err.Error(), filer_pb.ErrNotFound.Error())
+	if errors.Is(err, filer_pb.ErrNotFound) {
+		return true
+	}
+	if st, ok := util.ServerStatus(err); ok {
+		return st.Code() == codes.NotFound || strings.Contains(st.Message(), filer_pb.ErrNotFound.Error())
+	}
+	return strings.Contains(err.Error(), filer_pb.ErrNotFound.Error())
 }
 
 // ErrorHandlers provide common error handling patterns for S3 API operations
diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go
index 77840f387..44ce312c5 100644
--- a/weed/s3api/s3api_object_versioning.go
+++ b/weed/s3api/s3api_object_versioning.go
@@ -1657,8 +1657,8 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(ctx context.Context, bu
 		// 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
-		// canonical error substring; if we can't tell, treat as transient.
-		if strings.Contains(rmErr.Error(), filer.MsgFailDelNonEmptyFolder) {
+		// sentinel; if we can't tell, treat as transient.
+		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
@@ -1673,7 +1673,7 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(ctx context.Context, bu
 		if retryErr == nil {
 			return nil
 		}
-		if strings.Contains(retryErr.Error(), filer.MsgFailDelNonEmptyFolder) {
+		if errors.Is(retryErr, filer.ErrNonEmptyFolder) {
 			s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
 			return nil
 		}
diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go
index 0454396ba..1679b1b4e 100644
--- a/weed/server/filer_grpc_server.go
+++ b/weed/server/filer_grpc_server.go
@@ -9,7 +9,6 @@ import (
 	"os"
 	"path/filepath"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
@@ -433,7 +432,7 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje
 			// the expected no-op, and a failed teardown must not fail the
 			// already-applied delete.
 			parentErr := fs.filer.DeleteEntryMetaAndData(ctx, util.FullPath(m.Directory), false, false, false, fromOtherCluster, signatures, 0)
-			if parentErr != nil && parentErr != filer_pb.ErrNotFound && !strings.Contains(parentErr.Error(), filer.MsgFailDelNonEmptyFolder) {
+			if parentErr != nil && parentErr != filer_pb.ErrNotFound && !errors.Is(parentErr, filer.ErrNonEmptyFolder) {
 				glog.V(1).InfofCtx(ctx, "remove empty parent %s: %v", m.Directory, parentErr)
 			}
 		}
diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go
index e90aad47b..76d344f82 100644
--- a/weed/shell/command_volume_fsck.go
+++ b/weed/shell/command_volume_fsck.go
@@ -747,7 +747,7 @@ func (c *commandVolumeFsck) purgeEmptyDirectories() {
 			continue
 		}
 		if err := c.deleteEmptyDirectory(dir, mtime); err != nil {
-			if !strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
+			if !errors.Is(err, filer.ErrNonEmptyFolder) {
 				fmt.Fprintf(c.writer, "delete empty directory %s: %v\n", dir, err)
 			}
 			continue
@@ -771,7 +771,7 @@ func (c *commandVolumeFsck) deleteEmptyDirectory(dir util.FullPath, mtime int64)
 			return err
 		}
 		if resp.Error != "" {
-			return errors.New(resp.Error)
+			return filer.DeleteEntryError(resp.Error)
 		}
 		return nil
 	})
diff --git a/weed/util/retry.go b/weed/util/retry.go
index 3cee2ce1e..b5017b1d2 100644
--- a/weed/util/retry.go
+++ b/weed/util/retry.go
@@ -10,6 +10,8 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 var RetryWaitTime = 6 * time.Second
@@ -56,6 +58,20 @@ func IsTransientErrorMessage(msg string) bool {
 	return false
 }
 
+// ServerStatus returns the status a gRPC server sent, when the error chain
+// still carries one. Unlike status.FromError it keeps the server's own message
+// instead of the whole wrapped string: callers routinely format a path the
+// client chose into their wrapper, so a classifier that matches substrings has
+// to read what the server said and not what the caller added around it.
+func ServerStatus(err error) (*status.Status, bool) {
+	var carrier interface{ GRPCStatus() *status.Status }
+	if !errors.As(err, &carrier) {
+		return nil, false
+	}
+	st := carrier.GRPCStatus()
+	return st, st != nil
+}
+
 // IsTransientError reports whether err is a network or service condition worth
 // retrying. A cancelled or expired context never is: the caller is already gone.
 func IsTransientError(err error) bool {
@@ -75,6 +91,10 @@ func IsTransientError(err error) bool {
 	if errors.As(err, &netErr) && netErr.Timeout() {
 		return true
 	}
+	if st, ok := ServerStatus(err); ok {
+		return st.Code() == codes.Unavailable || st.Code() == codes.ResourceExhausted ||
+			IsTransientErrorMessage(st.Message())
+	}
 	return IsTransientErrorMessage(err.Error())
 }
 
diff --git a/weed/util/retry_test.go b/weed/util/retry_test.go
index f96cd8236..9bc1d5772 100644
--- a/weed/util/retry_test.go
+++ b/weed/util/retry_test.go
@@ -9,6 +9,9 @@ import (
 	"syscall"
 	"testing"
 	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func TestIsTransientError(t *testing.T) {
@@ -43,6 +46,35 @@ func TestIsTransientError(t *testing.T) {
 	}
 }
 
+// A caller that formats a path into its wrapper must not be able to hand the
+// substring matcher a word the client chose: once the chain carries the status
+// the server sent, that status is what gets classified.
+func TestIsTransientErrorPrefersServerStatus(t *testing.T) {
+	transient := []error{
+		fmt.Errorf("list %s: %w", "/buckets/b/logs", status.Error(codes.Unavailable, "filer is restarting")),
+		fmt.Errorf("list %s: %w", "/buckets/b/filer: no entry is found in filer store", status.Error(codes.Unavailable, "filer is restarting")),
+		status.Error(codes.ResourceExhausted, "too many requests"),
+		// the server's own text still counts when the code does not name the condition
+		status.Error(codes.Internal, "connection reset by peer"),
+	}
+	for _, err := range transient {
+		if !IsTransientError(err) {
+			t.Errorf("expected transient: %v", err)
+		}
+	}
+
+	permanent := []error{
+		fmt.Errorf("list %s: %w", "/buckets/transport/unavailable", status.Error(codes.PermissionDenied, "not allowed")),
+		fmt.Errorf("all filers failed, last error: %w",
+			fmt.Errorf("list %s: %w", "/buckets/slowdown/throttling", status.Error(codes.InvalidArgument, "bad prefix"))),
+	}
+	for _, err := range permanent {
+		if IsTransientError(err) {
+			t.Errorf("expected permanent: %v", err)
+		}
+	}
+}
+
 func TestIsTransientErrorMessage(t *testing.T) {
 	transient := []string{
 		"read tcp 10.0.0.1:8082->10.0.0.1:54848: i/o timeout",