diff --git a/weed/filer/filer_delete_collection_test.go b/weed/filer/filer_delete_collection_test.go
new file mode 100644
index 000000000..ac7458ce5
--- /dev/null
+++ b/weed/filer/filer_delete_collection_test.go
@@ -0,0 +1,183 @@
+package filer
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+)
+
+// collectionDeleteMaster is just enough of a master for a MasterClient to
+// consider itself connected, and records every CollectionDelete it is asked for.
+type collectionDeleteMaster struct {
+	master_pb.UnimplementedSeaweedServer
+	calls chan collectionDeleteCall
+}
+
+type collectionDeleteCall struct {
+	name string
+	// budget is the time the RPC arrived with, or 0 when it carried no deadline.
+	budget time.Duration
+}
+
+// KeepConnected is what MasterClient waits on before it reports a master: one
+// response is enough, then the stream idles until the server is torn down.
+func (m *collectionDeleteMaster) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error {
+	if _, err := stream.Recv(); err != nil {
+		return err
+	}
+	if err := stream.Send(&master_pb.KeepConnectedResponse{}); err != nil {
+		return err
+	}
+	<-stream.Context().Done()
+	return nil
+}
+
+func (m *collectionDeleteMaster) CollectionDelete(ctx context.Context, req *master_pb.CollectionDeleteRequest) (*master_pb.CollectionDeleteResponse, error) {
+	call := collectionDeleteCall{name: req.Name}
+	if deadline, ok := ctx.Deadline(); ok {
+		call.budget = time.Until(deadline)
+	}
+	m.calls <- call
+	return &master_pb.CollectionDeleteResponse{}, nil
+}
+
+// hookedStore is the stub store with one extra seam: a callback that runs once
+// an entry has actually been removed, so a test can interleave an event -- a
+// client hanging up, say -- between the store write and whatever follows it.
+type hookedStore struct {
+	*stubFilerStore
+	onDeleteEntry func(util.FullPath)
+}
+
+func (s *hookedStore) DeleteEntry(ctx context.Context, p util.FullPath) error {
+	if err := s.stubFilerStore.DeleteEntry(ctx, p); err != nil {
+		return err
+	}
+	if s.onDeleteEntry != nil {
+		s.onDeleteEntry(p)
+	}
+	return nil
+}
+
+// newFilerWithFakeMaster builds a filer backed by the stub store, with its
+// MasterClient connected to a fake master the test can observe.
+func newFilerWithFakeMaster(t *testing.T) (*Filer, *hookedStore, *collectionDeleteMaster) {
+	t.Helper()
+
+	master := &collectionDeleteMaster{calls: make(chan collectionDeleteCall, 4)}
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	grpcServer := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(grpcServer, master)
+	go func() { _ = grpcServer.Serve(lis) }()
+	t.Cleanup(grpcServer.Stop)
+
+	_, port, err := net.SplitHostPort(lis.Addr().String())
+	if err != nil {
+		t.Fatalf("split listener address: %v", err)
+	}
+	masterAddress := pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
+
+	mc := wdclient.NewMasterClient(
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		"test", cluster.FilerType, pb.ServerAddress("localhost:0"), "", "",
+		*pb.NewServiceDiscoveryFromMap(map[string]pb.ServerAddress{"m": masterAddress}),
+	)
+
+	connecting, stopConnecting := context.WithCancel(context.Background())
+	t.Cleanup(stopConnecting)
+	go mc.KeepConnectedToMaster(connecting)
+
+	waiting, cancelWaiting := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancelWaiting()
+	mc.WaitUntilConnected(waiting)
+	if waiting.Err() != nil {
+		t.Fatal("the master client never connected to the fake master")
+	}
+
+	store := &hookedStore{stubFilerStore: newStubFilerStore()}
+	f := &Filer{
+		DirBucketsPath:      "/buckets",
+		RemoteStorage:       NewFilerRemoteStorage(),
+		Store:               NewFilerStoreWrapper(store),
+		FilerConf:           NewFilerConf(),
+		MaxFilenameLength:   255,
+		MasterClient:        mc,
+		FileIdDeletionQueue: util.NewUnboundedQueue(),
+		deletionQuit:        make(chan struct{}),
+		LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute,
+			func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}),
+	}
+	return f, store, master
+}
+
+// Deleting a bucket entry deletes its collection, and the request hanging up
+// partway must not stop that: the bucket's metadata is already gone, so skipping
+// it strands the bucket's volumes with nothing left to come back for them. The
+// cleanup still has to carry a deadline of its own, or a master that is down
+// parks this handler and every client retry behind it parks another.
+//
+// The cancellation lands between the store removing the entry and the collection
+// delete, which is where a client disconnect actually bites: FilerStoreWrapper
+// refuses a context that is already dead on entry, so an up-front cancellation
+// fails the delete long before this point instead.
+func TestDeleteEntryMetaAndDataDeletesCollectionWhenTheRequestIsCancelledMidDelete(t *testing.T) {
+	f, store, master := newFilerWithFakeMaster(t)
+
+	const bucket = "bucket-a"
+	bucketPath := util.FullPath(f.DirBucketsPath + "/" + bucket)
+	if err := store.InsertEntry(context.Background(), &Entry{
+		FullPath: bucketPath,
+		Attr:     Attr{Mode: os.ModeDir | 0755},
+	}); err != nil {
+		t.Fatalf("seed the bucket entry: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	// The client hangs up just as the bucket entry comes out of the store.
+	store.onDeleteEntry = func(util.FullPath) { cancel() }
+
+	if err := f.DeleteEntryMetaAndData(ctx, bucketPath, true, false, true, false, nil, 0); err != nil {
+		t.Fatalf("DeleteEntryMetaAndData: %v", err)
+	}
+	if ctx.Err() == nil {
+		t.Fatal("test setup: the request was never cancelled, so nothing was exercised")
+	}
+
+	var call collectionDeleteCall
+	select {
+	case call = <-master.calls:
+	case <-time.After(20 * time.Second):
+		t.Fatal("CollectionDelete never reached the master")
+	}
+	if call.name != bucket {
+		t.Errorf("master was asked to delete collection %q, want %q", call.name, bucket)
+	}
+	if call.budget <= 0 {
+		t.Error("CollectionDelete arrived with no deadline; a master that stops answering would hold this handler open")
+	}
+	if call.budget > collectionDeleteTimeout {
+		t.Errorf("CollectionDelete budget = %v, want at most collectionDeleteTimeout %v", call.budget, collectionDeleteTimeout)
+	}
+
+	if store.getEntry(string(bucketPath)) != nil {
+		t.Error("the bucket entry survived the delete")
+	}
+}
diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go
index d7be43fd7..1cc2019df 100644
--- a/weed/filer/filer_delete_entry.go
+++ b/weed/filer/filer_delete_entry.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -103,8 +104,12 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 	if isDeleteCollection {
 		collectionName := entry.Name()
 		// the entry is already gone: a caller that hung up must not leave the
-		// collection behind, so this cleanup outlives the request
-		f.DoDeleteCollection(context.Background(), collectionName)
+		// collection behind, so this cleanup outlives the request -- bounded all
+		// the same, or a master that is down parks this handler indefinitely and
+		// every client retry behind it parks another
+		collectionCtx, cancelCollection := context.WithTimeout(context.WithoutCancel(ctx), collectionDeleteTimeout)
+		f.DoDeleteCollection(collectionCtx, collectionName)
+		cancelCollection()
 		// drop bucket-labeled series held by this process; the S3 gateway
 		// only cleans its own registry
 		stats.DeleteBucketMetrics(collectionName)
@@ -208,6 +213,14 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou
 	return nil
 }
 
+// collectionDeleteTimeout bounds the collection delete a bucket entry's own
+// delete leaves behind, which carries no deadline of its own. It bounds the wait
+// and not the work: the master keeps deleting on its own fan-out once asked, so
+// giving up costs the confirmation. Short enough that the S3 client waiting on
+// the bucket delete, which pays this and then the gateway's own follow-up
+// DeleteCollection, still has retry budget left.
+const collectionDeleteTimeout = 15 * time.Second
+
 func (f *Filer) DoDeleteCollection(ctx context.Context, collectionName string) (err error) {
 
 	return f.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go
index 5b335089e..adc860dde 100644
--- a/weed/s3api/s3api_bucket_handlers.go
+++ b/weed/s3api/s3api_bucket_handlers.go
@@ -35,6 +35,21 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
+// A bucket creation lists collections, and a bucket deletion deletes one.
+// Neither RPC carried a deadline, so a transient failure anywhere down the chain
+// -- gateway to filer, filer to master, master to volume server -- held the S3
+// request open until the client gave up on it. Both budgets are taken outside
+// the filer failover walk, so they cover the whole walk rather than granting
+// each filer a fresh one.
+//
+// The delete is the shorter of the two: the filer has already spent its own
+// budget on this collection, under the bucket entry's delete inside s3a.rm, and
+// this call is the follow-up for when that did not happen.
+const (
+	collectionListTimeout   = 15 * time.Second
+	collectionDeleteTimeout = 10 * time.Second
+)
+
 func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Request) {
 
 	glog.V(3).Infof("ListBucketsHandler")
@@ -259,12 +274,12 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists)
 		return
 	}
-	if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		if resp, err := client.CollectionList(context.Background(), &filer_pb.CollectionListRequest{
+	listCtx, cancelList := context.WithTimeout(r.Context(), collectionListTimeout)
+	if err := s3a.withFilerClient(listCtx, false, func(client filer_pb.SeaweedFilerClient) error {
+		if resp, err := client.CollectionList(listCtx, &filer_pb.CollectionListRequest{
 			IncludeEcVolumes:     true,
 			IncludeNormalVolumes: true,
 		}); err != nil {
-			glog.Errorf("list collection: %v", err)
 			return fmt.Errorf("list collections: %w", err)
 		} else {
 			for _, c := range resp.Collections {
@@ -276,9 +291,13 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		}
 		return nil
 	}); err != nil {
-		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
-		return
+		// Advisory: the answer decides nothing below except whether to log that a
+		// leftover collection is being reused. s3a.exists is what decides whether
+		// the bucket already exists, so a listing that failed is no reason to
+		// refuse the creation.
+		glog.Warningf("PutBucketHandler: list collections for %s: %v", bucket, err)
 	}
+	cancelList()
 
 	// Bucket already exists: report whether the caller already owns it or the
 	// name is taken / the request conflicts.
@@ -478,23 +497,34 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 		}
 	}
 
-	err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+	// Bounded on a background context: the bucket directory is already gone, so
+	// this follow-up must survive a client disconnect, but it must not outlive the
+	// client by an unbounded amount either.
+	deleteCtx, cancelDelete := context.WithTimeout(context.Background(), collectionDeleteTimeout)
+	err = s3a.withFilerClient(deleteCtx, false, func(client filer_pb.SeaweedFilerClient) error {
 		deleteCollectionRequest := &filer_pb.DeleteCollectionRequest{
 			Collection: s3a.getCollectionName(bucket),
 		}
 
 		glog.V(1).Infof("delete collection: %v", deleteCollectionRequest)
-		if _, err := client.DeleteCollection(context.Background(), deleteCollectionRequest); err != nil {
+		if _, err := client.DeleteCollection(deleteCtx, deleteCollectionRequest); err != nil {
 			return fmt.Errorf("delete collection %s: %v", bucket, err)
 		}
 
 		return nil
 	})
+	timedOut := deleteCtx.Err() != nil
+	cancelDelete()
 
 	if err != nil {
 		// Log but don't fail — the bucket directory is already removed, so the bucket
 		// is effectively deleted. The orphaned collection will be cleaned up or reused.
-		glog.Errorf("DeleteBucketHandler: failed to delete collection for bucket %s: %v", bucket, err)
+		if timedOut {
+			// Our own budget, not a refusal: the master carries on deleting once asked.
+			glog.Warningf("DeleteBucketHandler: stopped waiting for the collection delete for bucket %s: %v", bucket, err)
+		} else {
+			glog.Errorf("DeleteBucketHandler: failed to delete collection for bucket %s: %v", bucket, err)
+		}
 	}
 
 	// Clean up bucket-related caches, locks, and metrics after successful deletion
diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go
index 59ac9f996..4c3dd7f57 100644
--- a/weed/s3api/s3api_handlers.go
+++ b/weed/s3api/s3api_handlers.go
@@ -17,10 +17,20 @@ import (
 
 var _ = filer_pb.FilerClient(&S3ApiServer{})
 
+// WithFilerClient satisfies filer_pb.FilerClient, whose signature carries no
+// context. Callers with a budget to spend call withFilerClient directly.
 func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return s3a.withFilerClient(context.Background(), streamingMode, fn)
+}
+
+// withFilerClient runs fn with ctx as the budget bounding the calls fn makes. It
+// cancels nothing itself -- fn owns its own RPC contexts -- but the walk needs to
+// know whose budget ran out: a caller's expiry is not evidence against the filer
+// that was answering it.
+func (s3a *S3ApiServer) withFilerClient(ctx context.Context, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 	// Use filerClient for proper connection management and failover
 	if s3a.filerClient != nil {
-		return s3a.withFilerClientFailover("", streamingMode, fn)
+		return s3a.withFilerClientFailover(ctx, "", streamingMode, fn)
 	}
 
 	// Fallback to direct connection if filerClient not initialized
@@ -43,8 +53,10 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea
 // Failover replays fn from scratch, so it stops once any response has reached
 // fn: a replay after that could silently duplicate state fn accumulated (a
 // listing that failed mid-stream, say), so the error surfaces instead and the
-// caller decides whether a clean-slate retry is safe.
-func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+// caller decides whether a clean-slate retry is safe. ctx is the budget bounding
+// fn's own calls; it is only read, to tell a filer's failure apart from the
+// caller running out of time.
+func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred pb.ServerAddress, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 	currentFiler := s3a.filerClient.GetCurrentFiler()
 
 	candidates := make([]pb.ServerAddress, 0, 2+len(s3a.option.Filers))
@@ -87,6 +99,9 @@ func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, stre
 	var lastErr error
 	for _, filer := range ordered {
 		received := false
+		// Background, not ctx: WithGrpcClient's context only decides whether an
+		// error invalidates the shared connection, and that call is fn's to make,
+		// with the context fn's own RPC ran on.
 		err := pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
 			return fn(filer_pb.NewSeaweedFilerClient(receiveTrackingConn{ClientConnInterface: grpcConnection, received: &received}))
 		}, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption)
@@ -103,6 +118,15 @@ func (s3a *S3ApiServer) withFilerClientFailover(preferred pb.ServerAddress, stre
 			return err
 		}
 
+		// The caller's own budget expiring is not evidence against this filer, and
+		// the next one has no time left to answer either. Recorded as a failure, a
+		// slow master upstream would flag every filer in the walk, and the three
+		// that open the circuit take unrelated object reads down with them.
+		if ctx.Err() != nil {
+			glog.V(2).Infof("WithFilerClient: giving up on %s, the caller's context ended: %v", filer, err)
+			return err
+		}
+
 		s3a.filerClient.RecordFilerFailure(filer)
 		// A preferred owner is often outside the static filer list, where the health
 		// tracking above no-ops; flag it so route-by-key reads skip it briefly.
diff --git a/weed/s3api/s3api_handlers_failover_test.go b/weed/s3api/s3api_handlers_failover_test.go
index 7c10ace09..4606d6b23 100644
--- a/weed/s3api/s3api_handlers_failover_test.go
+++ b/weed/s3api/s3api_handlers_failover_test.go
@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -135,3 +136,54 @@ func TestListAfterMidStreamFailureHasNoDuplicates(t *testing.T) {
 		t.Fatal("want isLast")
 	}
 }
+
+// hangingCollectionFiler answers CollectionList only when the caller gives up,
+// standing in for a filer waiting on a master that has stopped answering.
+type hangingCollectionFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	calls atomic.Int32
+}
+
+func (f *hangingCollectionFiler) CollectionList(ctx context.Context, _ *filer_pb.CollectionListRequest) (*filer_pb.CollectionListResponse, error) {
+	f.calls.Add(1)
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+// The caller's own budget expiring says nothing about the filer that was
+// answering it. Counted as a filer failure it would, after the three that open
+// the circuit, mark every filer in the walk unhealthy -- and LookupVolumeIds
+// skips unhealthy filers, so unrelated object reads would start failing over a
+// slow master they never touched.
+func TestFailoverDoesNotBlameFilersForTheCallersExpiredBudget(t *testing.T) {
+	first := &hangingCollectionFiler{}
+	second := &hangingCollectionFiler{}
+	firstAddr := startFakeFiler(t, first)
+	secondAddr := startFakeFiler(t, second)
+	s3a := newFailoverTestServer(t, firstAddr, secondAddr)
+
+	// More attempts than the three failures it takes to open the circuit.
+	for attempt := 0; attempt < 5; attempt++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		err := s3a.withFilerClient(ctx, false, func(client filer_pb.SeaweedFilerClient) error {
+			_, listErr := client.CollectionList(ctx, &filer_pb.CollectionListRequest{})
+			return listErr
+		})
+		cancel()
+		if err == nil {
+			t.Fatalf("attempt %d: want the expired budget surfaced as an error", attempt)
+		}
+	}
+
+	for _, addr := range []pb.ServerAddress{firstAddr, secondAddr} {
+		if s3a.filerClient.ShouldSkipUnhealthyFiler(addr) {
+			t.Errorf("filer %s was flagged unhealthy for the caller's own timeout; unrelated reads would now skip it", addr)
+		}
+	}
+
+	// And the walk stops rather than spending an already-spent budget on the next
+	// filer, which cannot answer any faster.
+	if n := second.calls.Load(); n != 0 {
+		t.Errorf("the second filer was tried %d times with no budget left to answer in", n)
+	}
+}
diff --git a/weed/s3api/s3api_object_routed_read.go b/weed/s3api/s3api_object_routed_read.go
index 3bfff77b9..ae8015926 100644
--- a/weed/s3api/s3api_object_routed_read.go
+++ b/weed/s3api/s3api_object_routed_read.go
@@ -38,7 +38,7 @@ func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer
 
 	dir, name := fullPath.DirAndName()
 	var entry *filer_pb.Entry
-	err := s3a.withFilerClientFailover(preferred, false, func(client filer_pb.SeaweedFilerClient) error {
+	err := s3a.withFilerClientFailover(context.Background(), preferred, false, func(client filer_pb.SeaweedFilerClient) error {
 		resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
 			Directory: dir,
 			Name:      name,
diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go
index e80880045..b2a602722 100644
--- a/weed/s3api/s3api_object_routed_write.go
+++ b/weed/s3api/s3api_object_routed_write.go
@@ -170,7 +170,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O
 	if s3a.ownerRecentlyUnreachable(owner) {
 		preferred = ""
 	}
-	err := s3a.withFilerClientFailover(preferred, false, txn)
+	err := s3a.withFilerClientFailover(context.Background(), preferred, false, txn)
 	return resp, err
 }
 
diff --git a/weed/server/master_grpc_server_collection.go b/weed/server/master_grpc_server_collection.go
index 3a3f084f2..d85dd61f2 100644
--- a/weed/server/master_grpc_server_collection.go
+++ b/weed/server/master_grpc_server_collection.go
@@ -2,14 +2,25 @@ package weed_server
 
 import (
 	"context"
+	"time"
 
 	"github.com/seaweedfs/raft"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
+// deleteCollectionTimeout bounds one DeleteCollection RPC to a volume server, so
+// a server that accepts the connection and then stops answering cannot hold the
+// fan-out open with nothing to end it. Same bound as allocateVolumeTimeout, the
+// other master-to-volume-server admin RPC. The volume server runs the delete to
+// completion regardless of the request context, so giving up costs the
+// confirmation and not the deletion.
+const deleteCollectionTimeout = 1 * time.Minute
+
 func (ms *MasterServer) CollectionList(ctx context.Context, req *master_pb.CollectionListRequest) (*master_pb.CollectionListResponse, error) {
 
 	if !ms.Topo.IsLeader() {
@@ -33,63 +44,100 @@ func (ms *MasterServer) CollectionDelete(ctx context.Context, req *master_pb.Col
 		return nil, raft.NotLeaderError
 	}
 
-	resp := &master_pb.CollectionDeleteResponse{}
-
-	err := ms.doDeleteNormalCollection(req.Name)
-
-	if err != nil {
+	if err := ms.deleteCollection(ctx, req.Name); err != nil {
 		return nil, err
 	}
 
-	err = ms.doDeleteEcCollection(req.Name)
-
-	if err != nil {
-		return nil, err
-	}
-
-	return resp, nil
+	return &master_pb.CollectionDeleteResponse{}, nil
 }
 
-func (ms *MasterServer) doDeleteNormalCollection(collectionName string) error {
+// deleteCollection removes a collection's normal volumes and its EC shards. Both
+// passes run: one collection can hold both, and returning after a failed normal
+// pass left the shards in place with no request left to come back for them. The
+// normal pass keeps precedence in what is reported, as it did before.
+func (ms *MasterServer) deleteCollection(ctx context.Context, collectionName string) error {
+
+	// Values only, no deadline: a caller that hangs up must not abandon a
+	// destructive fan-out part-done, with volumes left behind and no request still
+	// running to come back for them. Each RPC is bounded on its own.
+	ctx = context.WithoutCancel(ctx)
+
+	normalErr := ms.doDeleteNormalCollection(ctx, collectionName)
+	ecErr := ms.doDeleteEcCollection(ctx, collectionName)
+
+	if normalErr != nil {
+		if ecErr != nil {
+			glog.ErrorfCtx(ctx, "delete collection %s ec shards: %v", collectionName, ecErr)
+		}
+		return normalErr
+	}
+
+	return ecErr
+}
+
+func (ms *MasterServer) doDeleteNormalCollection(ctx context.Context, collectionName string) error {
 
 	collection, ok := ms.Topo.FindCollection(collectionName)
 	if !ok {
 		return nil
 	}
 
-	for _, server := range collection.ListVolumeServers() {
-		err := operation.WithVolumeServerClient(false, server.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
-			_, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{
-				Collection: collectionName,
-			})
-			return deleteErr
-		})
-		if err != nil {
-			return err
+	// One RPC per server, not one per replica: ListVolumeServers reports a node
+	// once for every replica it holds, while DeleteCollection removes the whole
+	// collection from the server it reaches. ListEcServersByCollection already
+	// returns each server once.
+	var servers []pb.ServerAddress
+	seen := make(map[pb.ServerAddress]struct{})
+	for _, node := range collection.ListVolumeServers() {
+		address := node.ServerAddress()
+		if _, done := seen[address]; done {
+			continue
 		}
+		seen[address] = struct{}{}
+		servers = append(servers, address)
+	}
+
+	if err := ms.deleteCollectionFrom(ctx, collectionName, servers); err != nil {
+		return err
 	}
 	ms.Topo.DeleteCollection(collectionName)
 
 	return nil
 }
 
-func (ms *MasterServer) doDeleteEcCollection(collectionName string) error {
+func (ms *MasterServer) doDeleteEcCollection(ctx context.Context, collectionName string) error {
 
-	listOfEcServers := ms.Topo.ListEcServersByCollection(collectionName)
+	if err := ms.deleteCollectionFrom(ctx, collectionName, ms.Topo.ListEcServersByCollection(collectionName)); err != nil {
+		return err
+	}
+	ms.Topo.DeleteEcCollection(collectionName)
 
-	for _, server := range listOfEcServers {
+	return nil
+}
+
+// deleteCollectionFrom asks every server to drop the collection and keeps going
+// past a failure, so one server that is down does not leave the collection on
+// every server after it in the list. The first failure is what is reported, and
+// the collection stays in the topology so a later delete comes back for the rest.
+func (ms *MasterServer) deleteCollectionFrom(ctx context.Context, collectionName string, servers []pb.ServerAddress) error {
+
+	var firstErr error
+	for _, server := range servers {
 		err := operation.WithVolumeServerClient(false, server, ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
-			_, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{
+			rpcCtx, cancel := context.WithTimeout(ctx, deleteCollectionTimeout)
+			defer cancel()
+			_, deleteErr := client.DeleteCollection(rpcCtx, &volume_server_pb.DeleteCollectionRequest{
 				Collection: collectionName,
 			})
 			return deleteErr
 		})
 		if err != nil {
-			return err
+			glog.ErrorfCtx(ctx, "delete collection %s on %s: %v", collectionName, server, err)
+			if firstErr == nil {
+				firstErr = err
+			}
 		}
 	}
 
-	ms.Topo.DeleteEcCollection(collectionName)
-
-	return nil
+	return firstErr
 }
diff --git a/weed/server/master_grpc_server_collection_test.go b/weed/server/master_grpc_server_collection_test.go
new file mode 100644
index 000000000..49a671944
--- /dev/null
+++ b/weed/server/master_grpc_server_collection_test.go
@@ -0,0 +1,287 @@
+package weed_server
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// recordingVolumeServer reports what each DeleteCollection arrived carrying.
+// failWith answers an error at once; hang holds the call until the caller gives
+// up or the test releases it, the way a server that is still reachable but
+// wedged behaves.
+type recordingVolumeServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+
+	hang     bool
+	failWith error
+
+	calls       chan deleteCollectionCall
+	release     chan struct{}
+	releaseOnce sync.Once
+}
+
+type deleteCollectionCall struct {
+	collection string
+	// budget is the time the RPC arrived with, or 0 when it carried no deadline.
+	budget time.Duration
+}
+
+func newRecordingVolumeServer() *recordingVolumeServer {
+	return &recordingVolumeServer{
+		calls:   make(chan deleteCollectionCall, 8),
+		release: make(chan struct{}),
+	}
+}
+
+func (s *recordingVolumeServer) DeleteCollection(ctx context.Context, req *volume_server_pb.DeleteCollectionRequest) (*volume_server_pb.DeleteCollectionResponse, error) {
+	call := deleteCollectionCall{collection: req.Collection}
+	if deadline, ok := ctx.Deadline(); ok {
+		call.budget = time.Until(deadline)
+	}
+	s.calls <- call
+
+	if s.failWith != nil {
+		return nil, s.failWith
+	}
+	if s.hang {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-s.release:
+		}
+	}
+	return &volume_server_pb.DeleteCollectionResponse{}, nil
+}
+
+// Release unblocks every held DeleteCollection. Safe to call more than once.
+func (s *recordingVolumeServer) Release() {
+	s.releaseOnce.Do(func() { close(s.release) })
+}
+
+// addVolumeServer publishes stub on a fresh listener and registers it as a data
+// node of topo. Only the grpc port is dialed; the http port just has to stay
+// distinct per server so the topology does not treat two nodes as one address.
+func addVolumeServer(t *testing.T, topo *topology.Topology, id string, stub *recordingVolumeServer) *topology.DataNode {
+	t.Helper()
+	t.Cleanup(stub.Release)
+	grpcPort := serveGrpc(t, func(s *grpc.Server) {
+		volume_server_pb.RegisterVolumeServerServer(s, stub)
+	})
+	return topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", grpcPort-10000, grpcPort, "", id, map[string]uint32{"": 10})
+}
+
+func newCollectionTestMaster(topo *topology.Topology) *MasterServer {
+	return &MasterServer{
+		Topo:           topo,
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}
+}
+
+// awaitCall waits for one DeleteCollection to reach the stub.
+func awaitCall(t *testing.T, stub *recordingVolumeServer, what string) deleteCollectionCall {
+	t.Helper()
+	select {
+	case call := <-stub.calls:
+		return call
+	case <-time.After(15 * time.Second):
+		t.Fatalf("%s: the volume server never received DeleteCollection", what)
+		return deleteCollectionCall{}
+	}
+}
+
+// A caller with no deadline of its own must still not wait forever: the master
+// has to bound each RPC, or one wedged volume server holds the fan-out open with
+// nothing to end it.
+func TestDoDeleteNormalCollectionBoundsEachVolumeServerRPC(t *testing.T) {
+	const collection = "bucket-a"
+
+	stub := newRecordingVolumeServer()
+	stub.hang = true
+
+	topo := topology.NewTopology("test", nil, 32*1024, 5, false)
+	dn := addVolumeServer(t, topo, "vs1", stub)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+	}, dn)
+	ms := newCollectionTestMaster(topo)
+
+	done := make(chan error, 1)
+	go func() { done <- ms.doDeleteNormalCollection(context.Background(), collection) }()
+
+	call := awaitCall(t, stub, "bounded RPC")
+	if call.budget <= 0 {
+		t.Fatal("DeleteCollection reached the volume server with no deadline: a wedged server holds the fan-out open with nothing to end it")
+	}
+	if call.budget > deleteCollectionTimeout {
+		t.Errorf("DeleteCollection budget = %v, want at most deleteCollectionTimeout %v", call.budget, deleteCollectionTimeout)
+	}
+
+	stub.Release()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("doDeleteNormalCollection returned %v, want nil once the volume server answers", err)
+		}
+	case <-time.After(15 * time.Second):
+		t.Fatal("doDeleteNormalCollection did not return after the volume server answered")
+	}
+}
+
+// The fan-out must outlive the caller. A collection can span many servers and
+// the deletes are sequential, so letting an inbound cancellation end the loop
+// would abandon a large delete part-done, with volumes left behind and no
+// request still running to come back for them.
+func TestDeleteCollectionOutlivesTheCallersCancellation(t *testing.T) {
+	const collection = "bucket-b"
+
+	normal := newRecordingVolumeServer()
+	ec := newRecordingVolumeServer()
+
+	topo := topology.NewTopology("test", nil, 32*1024, 5, false)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+	}, addVolumeServer(t, topo, "vs-normal", normal))
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		{Id: 9, Collection: collection, EcIndexBits: 0x1f},
+	}, addVolumeServer(t, topo, "vs-ec", ec))
+	ms := newCollectionTestMaster(topo)
+
+	// The caller is already gone by the time the delete starts.
+	expired, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	if err := ms.deleteCollection(expired, collection); err != nil {
+		t.Fatalf("deleteCollection returned %v; a cancelled caller must not fail the fan-out", err)
+	}
+
+	// The EC pass runs second, so it is the one most likely to find the caller
+	// already gone.
+	for _, tc := range []struct {
+		stub *recordingVolumeServer
+		what string
+	}{{normal, "normal pass"}, {ec, "ec pass"}} {
+		call := awaitCall(t, tc.stub, "cancelled caller, "+tc.what)
+		if call.collection != collection {
+			t.Errorf("%s: server was told to delete %q, want %q", tc.what, call.collection, collection)
+		}
+		if call.budget <= 0 {
+			t.Errorf("%s: DeleteCollection arrived with no deadline; each RPC should still be bounded", tc.what)
+		}
+	}
+
+	if _, found := topo.FindCollection(collection); found {
+		t.Error("the collection was left in the topology; the delete did not run to completion")
+	}
+}
+
+// ListVolumeServers reports a node once per replica it holds. DeleteCollection
+// removes the whole collection from the server it reaches, so the master must
+// send it once per server, not once per replica.
+func TestDoDeleteNormalCollectionSendsOneRPCPerVolumeServer(t *testing.T) {
+	const collection = "bucket-c"
+
+	stub := newRecordingVolumeServer()
+
+	topo := topology.NewTopology("test", nil, 32*1024, 5, false)
+	dn := addVolumeServer(t, topo, "vs1", stub)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+		{Id: 2, Collection: collection, Version: 3},
+		{Id: 3, Collection: collection, Version: 3},
+	}, dn)
+
+	collectionInTopology, found := topo.FindCollection(collection)
+	if !found {
+		t.Fatalf("test setup: collection %s was not registered", collection)
+	}
+	if listed := len(collectionInTopology.ListVolumeServers()); listed < 2 {
+		t.Fatalf("test setup: the one node is listed %d times, expected once per replica", listed)
+	}
+
+	ms := newCollectionTestMaster(topo)
+	if err := ms.doDeleteNormalCollection(context.Background(), collection); err != nil {
+		t.Fatalf("doDeleteNormalCollection: %v", err)
+	}
+
+	awaitCall(t, stub, "one RPC per server")
+	select {
+	case extra := <-stub.calls:
+		t.Errorf("the same server was told to delete %q more than once", extra.collection)
+	default:
+	}
+}
+
+// One server refusing must not leave the collection on every server after it in
+// the list: returning at the first failure meant a single node that was down
+// stranded the rest, and the failure is reported either way.
+func TestDoDeleteNormalCollectionKeepsGoingPastAFailingServer(t *testing.T) {
+	const collection = "bucket-d"
+
+	failing := newRecordingVolumeServer()
+	failing.failWith = errors.New("volume server is out of disk")
+	healthy := newRecordingVolumeServer()
+
+	topo := topology.NewTopology("test", nil, 32*1024, 5, false)
+	failingNode := addVolumeServer(t, topo, "vs-failing", failing)
+	healthyNode := addVolumeServer(t, topo, "vs-healthy", healthy)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+	}, failingNode)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+	}, healthyNode)
+
+	ms := newCollectionTestMaster(topo)
+	if err := ms.doDeleteNormalCollection(context.Background(), collection); err == nil {
+		t.Fatal("doDeleteNormalCollection reported success while a server refused")
+	}
+
+	awaitCall(t, failing, "failing server")
+	awaitCall(t, healthy, "server behind the failing one")
+
+	if _, found := topo.FindCollection(collection); !found {
+		t.Error("the collection was dropped from the topology despite failing; nothing would come back for the rest")
+	}
+}
+
+// A collection can hold normal volumes and EC shards at once. Returning after a
+// failed normal pass left the EC shards in place with no request left to come
+// back for them, so both passes have to run.
+func TestDeleteCollectionRunsTheEcPassWhenTheNormalPassFails(t *testing.T) {
+	const collection = "bucket-e"
+
+	normal := newRecordingVolumeServer()
+	normal.failWith = errors.New("volume server is out of disk")
+	ec := newRecordingVolumeServer()
+
+	topo := topology.NewTopology("test", nil, 32*1024, 5, false)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		{Id: 1, Collection: collection, Version: 3},
+	}, addVolumeServer(t, topo, "vs-normal", normal))
+	topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{
+		{Id: 9, Collection: collection, EcIndexBits: 0x1f},
+	}, addVolumeServer(t, topo, "vs-ec", ec))
+
+	ms := newCollectionTestMaster(topo)
+	if err := ms.deleteCollection(context.Background(), collection); err == nil {
+		t.Fatal("deleteCollection reported success while the normal pass failed")
+	}
+
+	// The failure is reported, and the EC shards are cleaned up anyway.
+	awaitCall(t, ec, "ec pass after a failed normal pass")
+
+	if _, found := topo.FindCollection(collection); !found {
+		t.Error("the normal collection was dropped from the topology despite failing; nothing would retry it")
+	}
+}
diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go
index 76e4ebc59..03f485a09 100644
--- a/weed/server/master_server_handlers_admin.go
+++ b/weed/server/master_server_handlers_admin.go
@@ -13,8 +13,6 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/operation"
-	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend/memory_map"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
@@ -25,24 +23,18 @@ import (
 
 func (ms *MasterServer) collectionDeleteHandler(w http.ResponseWriter, r *http.Request) {
 	collectionName := r.FormValue("collection")
-	collection, ok := ms.Topo.FindCollection(collectionName)
-	if !ok {
+	if _, ok := ms.Topo.FindCollection(collectionName); !ok {
 		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("collection %s does not exist", collectionName))
 		return
 	}
-	for _, server := range collection.ListVolumeServers() {
-		err := operation.WithVolumeServerClient(false, server.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
-			_, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{
-				Collection: collection.Name,
-			})
-			return deleteErr
-		})
-		if err != nil {
-			writeJsonError(w, r, http.StatusInternalServerError, err)
-			return
-		}
+
+	// The same two passes the gRPC CollectionDelete runs, rather than a second
+	// copy of the volume-server walk: this handler deleted only the normal
+	// volumes and left the collection's EC shards behind.
+	if err := ms.deleteCollection(r.Context(), collectionName); err != nil {
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
 	}
-	ms.Topo.DeleteCollection(collectionName)
 
 	w.WriteHeader(http.StatusNoContent)
 	return