From 902a12fd6fe48c2a6e55b4c4c0fc82a5de00336d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 27 Aug 2026 22:27:45 -0700 Subject: [PATCH] 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 | 4 +- .../multidisk_shardloss_test.go | 2 +- .../multidisk_shell_lifecycle_test.go | 2 +- weed/admin/dash/client_management.go | 2 +- weed/credential/propagating_store.go | 2 +- weed/filer/filer_delete_entry.go | 10 ++- .../integration/seaweedmq_handler_utils.go | 16 ++-- weed/server/filer_grpc_server.go | 6 +- weed/server/filer_grpc_server_admin.go | 4 +- weed/shell/command_cluster_check.go | 2 +- weed/shell/command_cluster_ps.go | 2 +- weed/shell/command_cluster_raft_add.go | 2 +- .../command_cluster_raft_leader_transfer.go | 4 +- weed/shell/command_cluster_raft_ps.go | 2 +- weed/shell/command_cluster_raft_remove.go | 2 +- weed/shell/command_collection_delete.go | 2 +- weed/shell/command_collection_list.go | 2 +- weed/shell/command_ec_common.go | 4 +- weed/shell/command_ec_decode.go | 2 +- weed/shell/command_fs_merge_volumes.go | 2 +- weed/shell/command_s3_bucket_delete.go | 2 +- weed/shell/command_volume_grow.go | 2 +- weed/shell/command_volume_vacuum.go | 2 +- weed/shell/command_volume_vacuum_disable.go | 2 +- weed/shell/command_volume_vacuum_enable.go | 2 +- weed/shell/commands.go | 8 +- weed/shell/shell_liner.go | 2 +- .../exclusive_locks/exclusive_locker.go | 6 +- weed/wdclient/masterclient.go | 20 +++-- weed/wdclient/masterclient_test.go | 83 +++++++++++++++++++ 30 files changed, 148 insertions(+), 55 deletions(-) diff --git a/test/erasure_coding/chaos_lifecycle_test.go b/test/erasure_coding/chaos_lifecycle_test.go index 68678b914..7fb1002b1 100644 --- a/test/erasure_coding/chaos_lifecycle_test.go +++ b/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 diff --git a/test/erasure_coding/multidisk_shardloss_test.go b/test/erasure_coding/multidisk_shardloss_test.go index c58c6ec3d..dc86b5bad 100644 --- a/test/erasure_coding/multidisk_shardloss_test.go +++ b/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 diff --git a/test/erasure_coding/multidisk_shell_lifecycle_test.go b/test/erasure_coding/multidisk_shell_lifecycle_test.go index a2f1ea7de..3a9408ddd 100644 --- a/test/erasure_coding/multidisk_shell_lifecycle_test.go +++ b/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 diff --git a/weed/admin/dash/client_management.go b/weed/admin/dash/client_management.go index 46948a628..6de7d1901 100644 --- a/weed/admin/dash/client_management.go +++ b/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 diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index f561d2007..8d42feb47 100644 --- a/weed/credential/propagating_store.go +++ b/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, diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go index 207fdc7e6..cbffcced0 100644 --- a/weed/filer/filer_delete_entry.go +++ b/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 { diff --git a/weed/mq/kafka/integration/seaweedmq_handler_utils.go b/weed/mq/kafka/integration/seaweedmq_handler_utils.go index 843b72280..33c3590ff 100644 --- a/weed/mq/kafka/integration/seaweedmq_handler_utils.go +++ b/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, diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 9bc9f9a7a..0454396ba 100644 --- a/weed/server/filer_grpc_server.go +++ b/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 } diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go index 0cef9e0c4..c013cff2e 100644 --- a/weed/server/filer_grpc_server_admin.go +++ b/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, diff --git a/weed/shell/command_cluster_check.go b/weed/shell/command_cluster_check.go index ca94ab7be..18e4232a3 100644 --- a/weed/shell/command_cluster_check.go +++ b/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, diff --git a/weed/shell/command_cluster_ps.go b/weed/shell/command_cluster_ps.go index a2fa8dc19..0c1e04fa8 100644 --- a/weed/shell/command_cluster_ps.go +++ b/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, diff --git a/weed/shell/command_cluster_raft_add.go b/weed/shell/command_cluster_raft_add.go index a957d7c44..f54083180 100644 --- a/weed/shell/command_cluster_raft_add.go +++ b/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, diff --git a/weed/shell/command_cluster_raft_leader_transfer.go b/weed/shell/command_cluster_raft_leader_transfer.go index 32dc83f51..0047d8efc 100644 --- a/weed/shell/command_cluster_raft_leader_transfer.go +++ b/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() diff --git a/weed/shell/command_cluster_raft_ps.go b/weed/shell/command_cluster_raft_ps.go index 714e17649..d97272b42 100644 --- a/weed/shell/command_cluster_raft_ps.go +++ b/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) diff --git a/weed/shell/command_cluster_raft_remove.go b/weed/shell/command_cluster_raft_remove.go index 6f5324731..bc18fc05a 100644 --- a/weed/shell/command_cluster_raft_remove.go +++ b/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, diff --git a/weed/shell/command_collection_delete.go b/weed/shell/command_collection_delete.go index 92473e8e3..889da7f99 100644 --- a/weed/shell/command_collection_delete.go +++ b/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, }) diff --git a/weed/shell/command_collection_list.go b/weed/shell/command_collection_list.go index 0287b1e96..4114f73d6 100644 --- a/weed/shell/command_collection_list.go +++ b/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, diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 1fe691522..7922f876b 100644 --- a/weed/shell/command_ec_common.go +++ b/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 }) diff --git a/weed/shell/command_ec_decode.go b/weed/shell/command_ec_decode.go index 9a10fbf95..33e640d18 100644 --- a/weed/shell/command_ec_decode.go +++ b/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 }) diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 6a888cff5..2156c6f3a 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/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 diff --git a/weed/shell/command_s3_bucket_delete.go b/weed/shell/command_s3_bucket_delete.go index ddd3201e9..19ea4f329 100644 --- a/weed/shell/command_s3_bucket_delete.go +++ b/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), }) diff --git a/weed/shell/command_volume_grow.go b/weed/shell/command_volume_grow.go index 1a9a043d7..240506deb 100644 --- a/weed/shell/command_volume_grow.go +++ b/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 } diff --git a/weed/shell/command_volume_vacuum.go b/weed/shell/command_volume_vacuum.go index 34f7870d9..4a761f2b5 100644 --- a/weed/shell/command_volume_vacuum.go +++ b/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, diff --git a/weed/shell/command_volume_vacuum_disable.go b/weed/shell/command_volume_vacuum_disable.go index 15897ddb0..436e963ff 100644 --- a/weed/shell/command_volume_vacuum_disable.go +++ b/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 }) diff --git a/weed/shell/command_volume_vacuum_enable.go b/weed/shell/command_volume_vacuum_enable.go index b8f779b14..0d5104e9d 100644 --- a/weed/shell/command_volume_vacuum_enable.go +++ b/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 }) diff --git a/weed/shell/commands.go b/weed/shell/commands.go index 896a2d81b..315dbcb58 100644 --- a/weed/shell/commands.go +++ b/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, }) diff --git a/weed/shell/shell_liner.go b/weed/shell/shell_liner.go index aaeaf6ae2..1ae4ce5c5 100644 --- a/weed/shell/shell_liner.go +++ b/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, diff --git a/weed/wdclient/exclusive_locks/exclusive_locker.go b/weed/wdclient/exclusive_locks/exclusive_locker.go index e45775165..b1cd94f8a 100644 --- a/weed/wdclient/exclusive_locks/exclusive_locker.go +++ b/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, diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index b06dffad0..4fe8a1e2a 100644 --- a/weed/wdclient/masterclient.go +++ b/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) }) }) diff --git a/weed/wdclient/masterclient_test.go b/weed/wdclient/masterclient_test.go index e19ce9e80..c6450ce64 100644 --- a/weed/wdclient/masterclient_test.go +++ b/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) + } +}