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

@@ -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++

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))