From db02eea3496c98601236354ea8ea6782b2bc04ac Mon Sep 17 00:00:00 2001 From: "user.email" Date: Mon, 26 Oct 2026 18:42:49 -0700 Subject: [PATCH] cluster: bound connection acquisition inside the lock attempt A blocking dial ran under context.Background(), so a caller supplying grpc.WithBlock() could wait past the attempt window before the RPC deadline even started. The attempt context now covers both getOrCreateConnection and the DistributedLock call, and WithGrpcClient threads it into GrpcDial on both the cached and streaming paths. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/cluster/lock_client.go | 13 +++++++------ weed/pb/grpc_client_server.go | 18 +++++++++--------- 2 files changed, 16 insertions(+), 15 deletions(-) diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index 1da611cf25..5b3c76cb6f 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -435,11 +435,12 @@ func (lock *LiveLock) doLock(lockDuration, attemptTimeout time.Duration) (errorM previousHostFiler := lock.hostFiler previousOwner := lock.owner - err = pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - // Bound the attempt so an unreachable owner fails within the retry - // window. - ctx, cancel := context.WithTimeout(context.Background(), attemptTimeout) - defer cancel() + // Bound the whole attempt — connection acquisition included — so an + // unreachable owner fails within the retry window. + ctx, cancel := context.WithTimeout(context.Background(), attemptTimeout) + defer cancel() + err = pb.WithGrpcClient(ctx, false, 0, func(grpcConnection *grpc.ClientConn) error { + client := filer_pb.NewSeaweedFilerClient(grpcConnection) resp, err := client.DistributedLock(ctx, &filer_pb.LockRequest{ Name: lock.key, SecondsToLock: int64(lockDuration.Seconds()), @@ -484,7 +485,7 @@ func (lock *LiveLock) doLock(lockDuration, attemptTimeout time.Duration) (errorM } } return err - }) + }, lock.hostFiler.ToGrpcAddress(), false, lock.grpcDialOption) if err != nil && !lock.hostFiler.Equals(lock.lc.seedFiler) { lock.consecutiveFailures++ diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index 2a119c075e..f8f5837d8a 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -272,7 +272,7 @@ func GrpcDial(ctx context.Context, address string, waitForReady bool, opts ...gr return grpc.DialContext(ctx, address, options...) } -func getOrCreateConnection(address string, waitForReady bool, opts ...grpc.DialOption) (*versionedGrpcClient, error) { +func getOrCreateConnection(ctx context.Context, address string, waitForReady bool, opts ...grpc.DialOption) (*versionedGrpcClient, error) { grpcClientsLock.Lock() defer grpcClientsLock.Unlock() @@ -284,7 +284,6 @@ func getOrCreateConnection(address string, waitForReady bool, opts ...grpc.DialO } glog.V(2).Infof("Creating new gRPC connection to %s", address) - ctx := context.Background() grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...) if err != nil { return nil, fmt.Errorf("fail to dial %s: %v", address, err) @@ -468,15 +467,16 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { } // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection. -// ctx must be the context that bounds the RPC fn issues, so that a -// Canceled/DeadlineExceeded fired by ctx itself does not tear down the shared -// cached ClientConn out from under other concurrent callers. Pass -// context.Background() when fn picks its own context: a Canceled/DeadlineExceeded -// then never invalidates, since nothing here can attribute it to the channel. +// ctx must be the context that bounds this call's connection acquisition and +// the RPC fn issues, so that a Canceled/DeadlineExceeded fired by ctx itself +// does not tear down the shared cached ClientConn out from under other +// concurrent callers. Pass context.Background() when fn picks its own +// context: a Canceled/DeadlineExceeded then never invalidates, since nothing +// here can attribute it to the channel. func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error { if !streamingMode { - vgc, err := getOrCreateConnection(address, waitForReady, opts...) + vgc, err := getOrCreateConnection(ctx, address, waitForReady, opts...) if err != nil { return fmt.Errorf("getOrCreateConnection %s: %v", address, err) } @@ -498,7 +498,7 @@ func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn } // Streaming mode: dedicate a fresh ClientConn to this call. - dialCtx := context.Background() + dialCtx := ctx if signature != 0 { // Optimize: Use AppendToOutgoingContext instead of creating new map dialCtx = metadata.AppendToOutgoingContext(dialCtx, "sw-client-id", fmt.Sprintf("%d", signature))