wdclient: bound the wait for a master leader by the caller's context (#11002)

* wdclient: bound the wait for a master leader by the caller's context

WithClient waited on GetMaster with context.Background(), so a caller that
arrived while no master leader was known parked in a 200ms poll loop until one
appeared, whatever deadline it had already set on the RPC. Each retry above it
then left another goroutine in the same wait.

Take the context in WithClient and WithClientCustomGetMaster and hand it to
GetMaster, and stop the retry loop once it is done. The dial keeps
context.Background(): fn brings its own RPC context, so a cancellation seen
here cannot be attributed to the shared connection.

Call sites pass whatever they hold: the request context in the filer's
CollectionList, DeleteCollection and Statistics handlers and in the credential
store's propagation, the operation context in the shell's s3.bucket.delete and
the kafka gateway's broker and filer discovery, and context.Background() where
there is none - the shell commands, the admin dashboard wrapper, and the
exclusive locker's initial lease. The locker's release keeps its own
uncancelled context so a slow unlock cannot turn into a ghost lock.

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

* wdclient: test that WithClient gives up with the caller's context

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

* wdclient: cut the master retry backoff short when the caller gives up

util.Retry sleeps unconditionally between attempts, so a transient error
arriving just before the caller's deadline still cost it a full backoff step.
Use the context-aware util.RetryWithBackoff, the same helper the volume lookup
in this file already uses.

Two call sites went with it: the shell's lock-holder lookup builds its three
second bound before WithClient so it also covers finding the leader, as its
comment already promised, and the filer's post-delete collection cleanup goes
back to an uncancelled context - the entry is already gone, so a caller that
hung up must not leave the collection behind.

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

* wdclient: test that a cancel during backoff ends the retry

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

test/erasure_coding/chaos_lifecycle_test.go

@@ -933,7 +933,7 @@ func copyFileContents(src, dst string) error {
 // chaosDataNodes lists the data nodes from a fresh master topology snapshot.
 func chaosDataNodes(commandEnv *shell.CommandEnv) []*master_pb.DataNodeInfo {
 	var resp *master_pb.VolumeListResponse
-	err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		var e error
 		resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
 		return e
@@ -955,7 +955,7 @@ func chaosDataNodes(commandEnv *shell.CommandEnv) []*master_pb.DataNodeInfo {
 func masterEcGenerations(commandEnv *shell.CommandEnv, volumeId uint32) map[int64]bool {
 	generations := map[int64]bool{}
 	var resp *master_pb.VolumeListResponse
-	err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		var e error
 		resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
 		return e

test/erasure_coding/multidisk_shardloss_test.go

@@ -200,7 +200,7 @@ func disksWithShards(testDir string, volumeId uint32) int {
 func nodeVolumeDiskCounts(t *testing.T, commandEnv *shell.CommandEnv) map[string]int {
 	t.Helper()
 	var resp *master_pb.VolumeListResponse
-	err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		var e error
 		resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
 		return e

test/erasure_coding/multidisk_shell_lifecycle_test.go

@@ -370,7 +370,7 @@ func removeTwoShardFiles(t *testing.T, testDir string, volumeId uint32) []int {
 func masterEcShardIds(commandEnv *shell.CommandEnv, volumeId uint32) map[int]bool {
 	ids := map[int]bool{}
 	var resp *master_pb.VolumeListResponse
-	err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		var e error
 		resp, e = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
 		return e

weed/admin/dash/client_management.go

@@ -20,7 +20,7 @@ import (
 
 // WithMasterClient executes a function with a master client connection
 func (s *AdminServer) WithMasterClient(f func(client master_pb.SeaweedClient) error) error {
-	return s.masterClient.WithClient(false, f)
+	return s.masterClient.WithClient(context.Background(), false, f)
 }
 
 // WithFilerClient executes a function with a filer client connection

weed/credential/propagating_store.go

@@ -55,7 +55,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun
 
 	// List S3 servers
 	var s3Servers []string
-	err := s.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := s.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 		glog.V(4).Infof("IAM: listing S3 servers (FilerGroup: '%s')", s.masterClient.FilerGroup)
 		resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{
 			ClientType: cluster.S3Type,

weed/filer/filer_delete_entry.go

@@ -63,7 +63,9 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 
 	if isDeleteCollection {
 		collectionName := entry.Name()
-		f.DoDeleteCollection(collectionName)
+		// 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)
 		// drop bucket-labeled series held by this process; the S3 gateway
 		// only cleans its own registry
 		stats.DeleteBucketMetrics(collectionName)
@@ -167,10 +169,10 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou
 	return nil
 }
 
-func (f *Filer) DoDeleteCollection(collectionName string) (err error) {
+func (f *Filer) DoDeleteCollection(ctx context.Context, collectionName string) (err error) {
 
-	return f.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
-		_, err := client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{
+	return f.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+		_, err := client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{
 			Name: collectionName,
 		})
 		if err != nil {

weed/mq/kafka/integration/seaweedmq_handler_utils.go

@@ -54,7 +54,7 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str
 
 	// Discover brokers from masters using master client
 	glog.V(1).Infof("About to call discoverBrokersWithMasterClient...")
-	brokerAddresses, err := discoverBrokersWithMasterClient(masterClient, filerGroup)
+	brokerAddresses, err := discoverBrokersWithMasterClient(ctx, masterClient, filerGroup)
 	if err != nil {
 		glog.Errorf("Broker discovery failed: %v", err)
 		return nil, fmt.Errorf("failed to discover brokers: %v", err)
@@ -66,7 +66,7 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str
 	}
 
 	// Discover filers from masters using master client
-	filerAddresses, err := discoverFilersWithMasterClient(masterClient, filerGroup)
+	filerAddresses, err := discoverFilersWithMasterClient(ctx, masterClient, filerGroup)
 	if err != nil {
 		return nil, fmt.Errorf("failed to discover filers: %v", err)
 	}
@@ -107,12 +107,12 @@ func NewSeaweedMQBrokerHandler(masters string, filerGroup string, clientHost str
 }
 
 // discoverBrokersWithMasterClient queries masters for available brokers using reusable master client
-func discoverBrokersWithMasterClient(masterClient *wdclient.MasterClient, filerGroup string) ([]string, error) {
+func discoverBrokersWithMasterClient(ctx context.Context, masterClient *wdclient.MasterClient, filerGroup string) ([]string, error) {
 	var brokers []string
 
-	err := masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 		glog.V(1).Infof("Inside MasterClient.WithClient callback - client obtained successfully")
-		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+		resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{
 			ClientType: cluster.BrokerType,
 			FilerGroup: filerGroup,
 			Limit:      1000,
@@ -144,11 +144,11 @@ func discoverBrokersWithMasterClient(masterClient *wdclient.MasterClient, filerG
 }
 
 // discoverFilersWithMasterClient queries masters for available filers using reusable master client
-func discoverFilersWithMasterClient(masterClient *wdclient.MasterClient, filerGroup string) ([]pb.ServerAddress, error) {
+func discoverFilersWithMasterClient(ctx context.Context, masterClient *wdclient.MasterClient, filerGroup string) ([]pb.ServerAddress, error) {
 	var filers []pb.ServerAddress
 
-	err := masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
-		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
+	err := masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+		resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{
 			ClientType: cluster.FilerType,
 			FilerGroup: filerGroup,
 			Limit:      1000,

weed/server/filer_grpc_server.go

@@ -890,8 +890,8 @@ func (fs *FilerServer) CollectionList(ctx context.Context, req *filer_pb.Collect
 	glog.V(4).InfofCtx(ctx, "CollectionList %v", req)
 	resp = &filer_pb.CollectionListResponse{}
 
-	err = fs.filer.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
-		masterResp, err := client.CollectionList(context.Background(), &master_pb.CollectionListRequest{
+	err = fs.filer.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+		masterResp, err := client.CollectionList(ctx, &master_pb.CollectionListRequest{
 			IncludeNormalVolumes: req.IncludeNormalVolumes,
 			IncludeEcVolumes:     req.IncludeEcVolumes,
 		})
@@ -911,7 +911,7 @@ func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.Delet
 
 	glog.V(4).InfofCtx(ctx, "DeleteCollection %v", req)
 
-	err = fs.filer.DoDeleteCollection(req.GetCollection())
+	err = fs.filer.DoDeleteCollection(ctx, req.GetCollection())
 
 	return &filer_pb.DeleteCollectionResponse{}, err
 }

weed/server/filer_grpc_server_admin.go

@@ -21,8 +21,8 @@ func (fs *FilerServer) Statistics(ctx context.Context, req *filer_pb.StatisticsR
 
 	var output *master_pb.StatisticsResponse
 
-	err = fs.filer.MasterClient.WithClient(false, func(masterClient master_pb.SeaweedClient) error {
-		grpcResponse, grpcErr := masterClient.Statistics(context.Background(), &master_pb.StatisticsRequest{
+	err = fs.filer.MasterClient.WithClient(ctx, false, func(masterClient master_pb.SeaweedClient) error {
+		grpcResponse, grpcErr := masterClient.Statistics(ctx, &master_pb.StatisticsRequest{
 			Replication: fs.statisticsReplication(req.Replication),
 			Collection:  req.Collection,
 			Ttl:         req.Ttl,

weed/shell/command_cluster_check.go

@@ -62,7 +62,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i
 
 	// collect filers
 	var filers []pb.ServerAddress
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 			ClientType: cluster.FilerType,
 			FilerGroup: *commandEnv.option.FilerGroup,

weed/shell/command_cluster_ps.go

@@ -49,7 +49,7 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
 
 	listClusterNodes := func(clientType string) ([]*master_pb.ListClusterNodesResponse_ClusterNode, error) {
 		var nodes []*master_pb.ListClusterNodesResponse_ClusterNode
-		err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+		err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 			resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 				ClientType: clientType,
 				FilerGroup: *commandEnv.option.FilerGroup,

weed/shell/command_cluster_raft_add.go

@@ -46,7 +46,7 @@ func (c *commandRaftServerAdd) Do(args []string, commandEnv *CommandEnv, writer
 		return fmt.Errorf("empty server id or address")
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		_, err := client.RaftAddServer(context.Background(), &master_pb.RaftAddServerRequest{
 			Id:      *serverId,
 			Address: *serverAddress,

weed/shell/command_cluster_raft_leader_transfer.go

@@ -67,7 +67,7 @@ func (c *commandRaftLeaderTransfer) Do(args []string, commandEnv *CommandEnv, wr
 	fmt.Fprintf(writer, "Checking current raft cluster status...\n")
 
 	var currentLeader string
-	err := commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err := commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
 
@@ -107,7 +107,7 @@ func (c *commandRaftLeaderTransfer) Do(args []string, commandEnv *CommandEnv, wr
 	}
 	fmt.Fprintf(writer, "\nTransferring leadership from %s to %s...\n", currentLeader, targetDesc)
 
-	err = commandEnv.MasterClient.WithClient(true, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), true, func(client master_pb.SeaweedClient) error {
 		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
 

weed/shell/command_cluster_raft_ps.go

@@ -38,7 +38,7 @@ func (c *commandRaftClusterPs) Do(args []string, commandEnv *CommandEnv, writer
 		return nil
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err := client.RaftListClusterServers(context.Background(), &master_pb.RaftListClusterServersRequest{})
 		if err != nil {
 			return fmt.Errorf("raft list cluster: %w", err)

weed/shell/command_cluster_raft_remove.go

@@ -44,7 +44,7 @@ func (c *commandRaftServerRemove) Do(args []string, commandEnv *CommandEnv, writ
 		return fmt.Errorf("empty server id")
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		_, err := client.RaftRemoveServer(context.Background(), &master_pb.RaftRemoveServerRequest{
 			Id:    *serverId,
 			Force: true,

weed/shell/command_collection_delete.go

@@ -64,7 +64,7 @@ func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writ
 		return nil
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		_, err = client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{
 			Name: *collectionName,
 		})

weed/shell/command_collection_list.go

@@ -79,7 +79,7 @@ func (c *commandCollectionList) Do(args []string, commandEnv *CommandEnv, writer
 
 func ListCollectionNames(commandEnv *CommandEnv, includeNormalVolumes, includeEcVolumes bool) (collections []string, err error) {
 	var resp *master_pb.CollectionListResponse
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err = client.CollectionList(context.Background(), &master_pb.CollectionListRequest{
 			IncludeNormalVolumes: includeNormalVolumes,
 			IncludeEcVolumes:     includeEcVolumes,

weed/shell/command_ec_common.go

@@ -55,7 +55,7 @@ func _getDefaultReplicaPlacement(commandEnv *CommandEnv) (*super_block.ReplicaPl
 	var resp *master_pb.GetMasterConfigurationResponse
 	var err error
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err = client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 		return err
 	})
@@ -94,7 +94,7 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura
 	}
 
 	var resp *master_pb.VolumeListResponse
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err = pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{})
 		return err
 	})

weed/shell/command_ec_decode.go

@@ -124,7 +124,7 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 
 func lookupVolumeIds(commandEnv *CommandEnv, volumeIds []string) (volumeIdLocations []*master_pb.LookupVolumeResponse_VolumeIdLocation, err error) {
 	var resp *master_pb.LookupVolumeResponse
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		resp, err = client.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: volumeIds})
 		return err
 	})

weed/shell/command_fs_merge_volumes.go

@@ -318,7 +318,7 @@ func (c *commandFsMergeVolumes) volumesAreCompatible(src needle.VolumeId, dest n
 func (c *commandFsMergeVolumes) reloadVolumesInfo(masterClient *wdclient.MasterClient) error {
 	c.volumes = make(map[needle.VolumeId]*master_pb.VolumeInformationMessage)
 
-	return masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	return masterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		volumes, err := pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{})
 		if err != nil {
 			return err

weed/shell/command_s3_bucket_delete.go

@@ -66,7 +66,7 @@ func (c *commandS3BucketDelete) Do(args []string, commandEnv *CommandEnv, writer
 	}
 
 	// delete the collection directly first
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 		_, err = client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{
 			Name: getCollectionName(commandEnv, *bucketName),
 		})

weed/shell/command_volume_grow.go

@@ -115,7 +115,7 @@ func (c *commandGrow) Do(args []string, commandEnv *CommandEnv, writer io.Writer
 	if !collectionFound {
 		return fmt.Errorf("collection not found")
 	}
-	if err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	if err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		if _, err := client.VolumeGrow(context.Background(), volumeGrowRequest); err != nil {
 			return err
 		}

weed/shell/command_volume_vacuum.go

@@ -99,7 +99,7 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ
 	}
 
 	for _, volumeId := range volumeIdInts {
-		err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+		err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 			_, err = client.VacuumVolume(context.Background(), &master_pb.VacuumVolumeRequest{
 				GarbageThreshold: float32(*garbageThreshold),
 				VolumeId:         volumeId,

weed/shell/command_volume_vacuum_disable.go

@@ -36,7 +36,7 @@ func (c *commandDisableVacuum) Do(args []string, commandEnv *CommandEnv, writer
 		return
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		_, err = client.DisableVacuum(context.Background(), &master_pb.DisableVacuumRequest{})
 		return err
 	})

weed/shell/command_volume_vacuum_enable.go

@@ -36,7 +36,7 @@ func (c *commandEnableVacuum) Do(args []string, commandEnv *CommandEnv, writer i
 		return
 	}
 
-	err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 		_, err = client.EnableVacuum(context.Background(), &master_pb.EnableVacuumRequest{})
 		return err
 	})

weed/shell/commands.go

@@ -112,11 +112,11 @@ func (ce *CommandEnv) isLocked() bool {
 
 // shellLockHolder asks the master who currently holds the cluster-wide shell
 // lock. Best effort: masters without GetAdminLockStatus report no holder, and
-// each attempt is bounded so an unresponsive master cannot hang the shell.
+// the lookup is bounded so an unresponsive master cannot hang the shell.
 func (ce *CommandEnv) shellLockHolder() (clientName string, message string, held bool) {
-	ce.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
-		attemptCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-		defer cancel()
+	attemptCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	ce.MasterClient.WithClient(attemptCtx, false, func(client master_pb.SeaweedClient) error {
 		resp, err := client.GetAdminLockStatus(attemptCtx, &master_pb.GetAdminLockStatusRequest{
 			LockName: cluster.AdminShellLockName,
 		})

weed/shell/shell_liner.go

@@ -61,7 +61,7 @@ func RunShell(options ShellOptions) {
 
 	if commandEnv.option.FilerAddress == "" {
 		var filers []pb.ServerAddress
-		commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+		commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 			resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 				ClientType: cluster.FilerType,
 				FilerGroup: *options.FilerGroup,

weed/wdclient/exclusive_locks/exclusive_locker.go

@@ -61,7 +61,7 @@ func (l *ExclusiveLocker) RequestLock(clientName string) {
 
 	// retry to get the lease
 	for {
-		if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+		if err := l.masterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error {
 			attemptCtx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
 			defer cancel()
 			resp, err := client.LeaseAdminToken(attemptCtx, &master_pb.LeaseAdminTokenRequest{
@@ -118,7 +118,7 @@ func (l *ExclusiveLocker) renewLease(ctx context.Context) error {
 	if !l.isLocked.Load() {
 		return nil
 	}
-	return l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	return l.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 		attemptCtx, cancel := context.WithTimeout(ctx, rpcTimeout)
 		defer cancel()
 		resp, err := client.LeaseAdminToken(attemptCtx, &master_pb.LeaseAdminTokenRequest{
@@ -152,7 +152,7 @@ func (l *ExclusiveLocker) ReleaseLock() {
 
 	// single unbounded attempt: a release cut short by a deadline leaves the
 	// lock held until it expires, turning a slow unlock into a ghost lock
-	l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
+	l.masterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
 		client.ReleaseAdminToken(ctx, &master_pb.ReleaseAdminTokenRequest{
 			PreviousToken:    prevToken,
 			PreviousLockTime: prevLockTsNs,

weed/wdclient/masterclient.go

@@ -409,16 +409,24 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
 		len(resp.VolumeLocation.NewEcVids), len(resp.VolumeLocation.DeletedEcVids))
 }
 
-func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
+func (mc *MasterClient) WithClient(ctx context.Context, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
 	getMasterF := func() pb.ServerAddress {
-		return mc.GetMaster(context.Background())
+		return mc.GetMaster(ctx)
 	}
-	return mc.WithClientCustomGetMaster(getMasterF, streamingMode, fn)
+	return mc.WithClientCustomGetMaster(ctx, getMasterF, streamingMode, fn)
 }
 
-func (mc *MasterClient) WithClientCustomGetMaster(getMasterF func() pb.ServerAddress, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
-	return util.Retry("master grpc", func() error {
-		return pb.WithMasterClient(context.Background(), streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+// WithClientCustomGetMaster bounds the wait for a master leader by ctx, so a
+// caller with a deadline is not parked for the length of an election. The dial
+// still gets context.Background(): fn brings its own RPC context, so nothing
+// here can attribute a cancellation to the shared connection.
+func (mc *MasterClient) WithClientCustomGetMaster(ctx context.Context, getMasterF func() pb.ServerAddress, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
+	return util.RetryWithBackoff(ctx, "master grpc", util.RetryWaitTime, util.IsTransientError, func() error {
+		master := getMasterF()
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		return pb.WithMasterClient(context.Background(), streamingMode, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			return fn(client)
 		})
 	})

weed/wdclient/masterclient_test.go

@@ -2,11 +2,14 @@ package wdclient
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 // TestWaitUntilConnectedWithoutKeepConnected verifies that WaitUntilConnected
@@ -101,3 +104,83 @@ func TestMasterClientFilerGroupLogging(t *testing.T) {
 		t.Errorf("Expected clientType %s, got %s", clientType, mc.clientType)
 	}
 }
+
+// TestWithClientStopsWaitingOnCanceledContext verifies that WithClient hands the
+// caller's context to the wait for a master leader, so a caller that has already
+// given up is not parked until an election finishes.
+func TestWithClientStopsWaitingOnCanceledContext(t *testing.T) {
+	mc := NewMasterClient(grpc.EmptyDialOption{}, "test-group", "test-client", "", "", "", pb.ServerDiscovery{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	called := false
+	start := time.Now()
+	err := mc.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+		called = true
+		return nil
+	})
+	elapsed := time.Since(start)
+
+	if !errors.Is(err, context.Canceled) {
+		t.Errorf("expected context.Canceled, got %v", err)
+	}
+	if called {
+		t.Error("callback ran without a master leader")
+	}
+	if elapsed > time.Second {
+		t.Errorf("WithClient blocked for %v with an already canceled context", elapsed)
+	}
+}
+
+// TestWithClientStopsWaitingOnDeadline verifies the same bound applies to a
+// deadline the caller set rather than an outright cancellation.
+func TestWithClientStopsWaitingOnDeadline(t *testing.T) {
+	mc := NewMasterClient(grpc.EmptyDialOption{}, "test-group", "test-client", "", "", "", pb.ServerDiscovery{})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	err := mc.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+		return nil
+	})
+	elapsed := time.Since(start)
+
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("expected context.DeadlineExceeded, got %v", err)
+	}
+	if elapsed > time.Second {
+		t.Errorf("WithClient blocked for %v past a 100ms deadline", elapsed)
+	}
+}
+
+// TestWithClientStopsBackoffOnCancel verifies that a cancellation arriving while
+// the retry is backing off cuts the backoff short rather than sleeping it out.
+func TestWithClientStopsBackoffOnCancel(t *testing.T) {
+	mc := NewMasterClient(grpc.WithTransportCredentials(insecure.NewCredentials()), "test-group", "test-client", "", "", "", pb.ServerDiscovery{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	getMasterF := func() pb.ServerAddress { return "localhost:19333" }
+
+	attempts := 0
+	start := time.Now()
+	err := mc.WithClientCustomGetMaster(ctx, getMasterF, false, func(client master_pb.SeaweedClient) error {
+		attempts++
+		cancel()
+		return errors.New("connection reset by peer")
+	})
+	elapsed := time.Since(start)
+
+	if attempts != 1 {
+		t.Errorf("expected 1 attempt, got %d", attempts)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Errorf("expected context.Canceled, got %v", err)
+	}
+	if elapsed > time.Second {
+		t.Errorf("WithClientCustomGetMaster slept %v after the context was canceled", elapsed)
+	}
+}