grpc: don't tear down the shared master connection on a caller's own timeout (#9775)

A Canceled/DeadlineExceeded from the caller's per-request context was
treated like a dead channel: it closed the shared cached ClientConn and
cancelled every other in-flight RPC on it with "the client connection is
closing". Under a burst of concurrent chunk assigns (e.g. a large S3
multipart upload) one slow assign hitting its 10s attempt timeout could
poison the connection for all the rest, cascading into a flood of 500s.

Thread the caller's context into shouldInvalidateConnection and only
invalidate on Canceled/DeadlineExceeded while that context is still live,
which isolates the genuine stale-channel signal (a peer restart behind a
k8s Service VIP). To carry the context, add a ctx parameter to the
existing WithGrpcClient, WithMasterClient, and WithMasterServerClient; the
master assign and volume-lookup paths pass their per-attempt context and
every other caller passes context.Background().

unmaintained/stress/ec_stress_runner.go

@@ -301,7 +301,7 @@ func (r *runner) fetchVolumeList(ctx context.Context) (*master_pb.VolumeListResp
 		}
 
 		var response *master_pb.VolumeListResponse
-		err := pb.WithMasterClient(false, pb.ServerAddress(masterAddress), r.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		err := pb.WithMasterClient(context.Background(), false, pb.ServerAddress(masterAddress), r.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			callCtx, cancel := context.WithTimeout(ctx, r.cfg.RequestTimeout)
 			defer cancel()
 

weed/cluster/master_client.go

@@ -12,7 +12,7 @@ import (
 
 func ListExistingPeerUpdates(master pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string, clientType string) (existingNodes []*master_pb.ClusterNodeUpdate) {
 
-	if grpcErr := pb.WithMasterClient(false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+	if grpcErr := pb.WithMasterClient(context.Background(), false, master, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 		resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 			ClientType: clientType,
 			FilerGroup: filerGroup,

weed/command/filer_copy.go

@@ -596,7 +596,7 @@ var _ = filer_pb.FilerClient(&FileCopyWorker{})
 func (worker *FileCopyWorker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) (err error) {
 
 	filerGrpcAddress := worker.filerAddress.ToGrpcAddress()
-	err = pb.WithGrpcClient(streamingMode, worker.signature, func(grpcConnection *grpc.ClientConn) error {
+	err = pb.WithGrpcClient(context.Background(), streamingMode, worker.signature, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, filerGrpcAddress, false, worker.options.grpcDialOption)

weed/command/filer_sync_verify.go

@@ -177,7 +177,7 @@ type simpleFilerClient struct {
 }
 
 func (c *simpleFilerClient) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
-	return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, c.grpcAddress.ToGrpcAddress(), false, c.grpcDialOption)

weed/command/upload.go

@@ -145,7 +145,7 @@ func runUpload(cmd *Command, args []string) bool {
 }
 
 func readMasterConfiguration(grpcDialOption grpc.DialOption, masterAddress pb.ServerAddress) (replication string, err error) {
-	err = pb.WithMasterClient(false, masterAddress, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+	err = pb.WithMasterClient(context.Background(), false, masterAddress, grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 		resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 		if err != nil {
 			return fmt.Errorf("get master %s configuration: %v", masterAddress, err)

weed/credential/grpc/grpc_store.go

@@ -98,7 +98,7 @@ func (store *IamGrpcStore) withIamClient(ctx context.Context, fn func(ctx contex
 		}
 	}
 
-	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error {
 		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
 		return fn(ctx, client)
 	}, filerAddress.ToGrpcAddress(), false, dialOption)

weed/credential/propagating_store.go

@@ -86,7 +86,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun
 		wg.Add(1)
 		go func(server string) {
 			defer wg.Done()
-			err := pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+			err := pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error {
 				glog.V(4).Infof("IAM: successfully connected to S3 server %s for propagation", server)
 				client := s3_pb.NewSeaweedS3IamCacheClient(conn)
 				return fn(propagateCtx, client)

weed/mount/wfs_filer_client.go

@@ -1,6 +1,7 @@
 package mount
 
 import (
+	"context"
 	"sync/atomic"
 
 	"google.golang.org/grpc"
@@ -21,7 +22,7 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile
 		for x := 0; x < n; x++ {
 
 			filerGrpcAddress := wfs.option.FilerAddresses[i].ToGrpcAddress()
-			err = pb.WithGrpcClient(streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error {
+			err = pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error {
 				client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 				return fn(client)
 			}, filerGrpcAddress, false, wfs.option.GrpcDialOption)

weed/operation/assign_file_id.go

@@ -91,7 +91,10 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO
 				// Per-attempt timeout to prevent a single slow RPC from consuming the entire retry budget
 				attemptCtx, attemptCancel := context.WithTimeout(deadlineCtx, 10*time.Second)
 				defer attemptCancel()
-				return WithMasterServerClient(false, masterFn(attemptCtx), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+				// Pass attemptCtx so its expiry is not mistaken for a dead shared
+				// connection: invalidating it would cancel every other in-flight
+				// assign with "the client connection is closing".
+				return WithMasterServerClient(attemptCtx, false, masterFn(attemptCtx), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 					req := &master_pb.AssignRequest{
 						Count:               request.Count,
 						Replication:         request.Replication,
@@ -151,7 +154,7 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO
 
 func LookupJwt(master pb.ServerAddress, grpcDialOption grpc.DialOption, fileId string) (token security.EncodedJwt) {
 
-	WithMasterServerClient(false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	WithMasterServerClient(context.Background(), false, master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 		resp, grpcErr := masterClient.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{
 			VolumeOrFileIds: []string{fileId},

weed/operation/grpc_client.go

@@ -1,6 +1,8 @@
 package operation
 
 import (
+	"context"
+
 	"google.golang.org/grpc"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -10,18 +12,17 @@ import (
 
 func WithVolumeServerClient(streamingMode bool, volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(volume_server_pb.VolumeServerClient) error) error {
 
-	return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 		client := volume_server_pb.NewVolumeServerClient(grpcConnection)
 		return fn(client)
 	}, volumeServer.ToGrpcAddress(), false, grpcDialOption)
 
 }
 
-func WithMasterServerClient(streamingMode bool, masterServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error {
-
-	return pb.WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
-		client := master_pb.NewSeaweedClient(grpcConnection)
-		return fn(client)
-	}, masterServer.ToGrpcAddress(), false, grpcDialOption)
-
+// WithMasterServerClient threads the caller's per-request context into the
+// connection-invalidation decision, so a Canceled/DeadlineExceeded from the
+// caller's own timeout does not invalidate the shared cached master connection.
+// Pass context.Background() when there is no per-request deadline to honor.
+func WithMasterServerClient(ctx context.Context, streamingMode bool, masterServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(masterClient master_pb.SeaweedClient) error) error {
+	return pb.WithMasterClient(ctx, streamingMode, masterServer, grpcDialOption, false, fn)
 }

weed/operation/lookup.go

@@ -82,7 +82,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 
 	//only query unknown_vids
 
-	err := WithMasterServerClient(false, masterFn(context.Background()), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err := WithMasterServerClient(context.Background(), false, masterFn(context.Background()), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 		req := &master_pb.LookupVolumeRequest{
 			VolumeOrFileIds: unknown_vids,

weed/pb/grpc_client_server.go

@@ -311,7 +311,9 @@ func InvalidateGrpcConnection(address string) {
 
 // grpcMarshalErrorPrefix is the library-owned prefix gRPC prepends to every
 // client-side proto marshal failure; see grpc-go rpc_util.go encode():
-//   status.Errorf(codes.Internal, "grpc: error while marshaling: %v", ...)
+//
+//	status.Errorf(codes.Internal, "grpc: error while marshaling: %v", ...)
+//
 // The "grpc:" token is reserved for gRPC internal diagnostics and will not
 // collide with user-produced Internal statuses.
 const grpcMarshalErrorPrefix = "grpc: error while marshaling"
@@ -352,8 +354,11 @@ func isClientSideMarshalError(err error) bool {
 	return s.Code() == codes.Internal && strings.HasPrefix(s.Message(), grpcMarshalErrorPrefix)
 }
 
-// shouldInvalidateConnection checks if an error indicates the cached connection should be invalidated
-func shouldInvalidateConnection(err error) bool {
+// shouldInvalidateConnection checks if an error indicates the cached connection
+// should be invalidated. ctx is the caller's per-request context (nil is treated
+// as a live context); it disambiguates a genuinely broken channel from the
+// caller cancelling or timing out its own request.
+func shouldInvalidateConnection(ctx context.Context, err error) bool {
 	if err == nil {
 		return false
 	}
@@ -369,8 +374,19 @@ func shouldInvalidateConnection(err error) bool {
 	if s, ok := status.FromError(err); ok {
 		code := s.Code()
 		switch code {
-		case codes.Unavailable, codes.Canceled, codes.DeadlineExceeded, codes.Aborted, codes.Internal:
+		case codes.Unavailable, codes.Aborted, codes.Internal:
 			return true
+		case codes.Canceled, codes.DeadlineExceeded:
+			// Ambiguous: this fires both when a stale cached channel rejects
+			// RPCs (e.g. a peer restart behind a k8s Service VIP), where we must
+			// invalidate, and when the caller's own context expired, where the
+			// shared channel is fine. Tearing the channel down for the latter
+			// cancels every other in-flight RPC on it with "the client
+			// connection is closing", a cascade that turns one slow request into
+			// a flood of failures across all concurrent callers. Only invalidate
+			// while the caller's context is still live, isolating the genuine
+			// stale-channel signal.
+			return ctx == nil || ctx.Err() == nil
 		}
 	}
 
@@ -387,7 +403,11 @@ func shouldInvalidateConnection(err error) bool {
 }
 
 // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection.
-func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error {
+// ctx is the caller's per-request context: a Canceled/DeadlineExceeded that fires
+// because ctx itself expired will not tear down the shared cached ClientConn out
+// from under other concurrent callers. Pass context.Background() when there is no
+// per-request deadline to honor.
+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...)
@@ -395,7 +415,7 @@ func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientCon
 			return fmt.Errorf("getOrCreateConnection %s: %v", address, err)
 		}
 		executionErr := fn(vgc.ClientConn)
-		if executionErr != nil && shouldInvalidateConnection(executionErr) {
+		if executionErr != nil && shouldInvalidateConnection(ctx, executionErr) {
 			grpcClientsLock.Lock()
 			t, ok := grpcClients[address]
 			shouldClose := ok && t.version == vgc.version
@@ -412,12 +432,12 @@ func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientCon
 	}
 
 	// Streaming mode: dedicate a fresh ClientConn to this call.
-	ctx := context.Background()
+	dialCtx := context.Background()
 	if signature != 0 {
 		// Optimize: Use AppendToOutgoingContext instead of creating new map
-		ctx = metadata.AppendToOutgoingContext(ctx, "sw-client-id", fmt.Sprintf("%d", signature))
+		dialCtx = metadata.AppendToOutgoingContext(dialCtx, "sw-client-id", fmt.Sprintf("%d", signature))
 	}
-	grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...)
+	grpcConnection, err := GrpcDial(dialCtx, address, waitForReady, opts...)
 	if err != nil {
 		return fmt.Errorf("fail to dial %s: %v", address, err)
 	}
@@ -494,8 +514,12 @@ func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) {
 	return util.JoinHostPort(host, port)
 }
 
-func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption grpc.DialOption, waitForReady bool, fn func(client master_pb.SeaweedClient) error) error {
-	return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+// WithMasterClient threads the caller's per-request context into the
+// connection-invalidation decision, so a Canceled/DeadlineExceeded from the
+// caller's own timeout does not invalidate the shared cached master connection.
+// Pass context.Background() when there is no per-request deadline to honor.
+func WithMasterClient(ctx context.Context, streamingMode bool, master ServerAddress, grpcDialOption grpc.DialOption, waitForReady bool, fn func(client master_pb.SeaweedClient) error) error {
+	return WithGrpcClient(ctx, streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 		client := master_pb.NewSeaweedClient(grpcConnection)
 		return fn(client)
 	}, master.ToGrpcAddress(), waitForReady, grpcDialOption)
@@ -503,7 +527,7 @@ func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption g
 }
 
 func WithVolumeServerClient(streamingMode bool, volumeServer ServerAddress, grpcDialOption grpc.DialOption, fn func(client volume_server_pb.VolumeServerClient) error) error {
-	return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+	return WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 		client := volume_server_pb.NewVolumeServerClient(grpcConnection)
 		return fn(client)
 	}, volumeServer.ToGrpcAddress(), false, grpcDialOption)
@@ -513,7 +537,7 @@ func WithVolumeServerClient(streamingMode bool, volumeServer ServerAddress, grpc
 func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[string]ServerAddress, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) (err error) {
 
 	for _, masterGrpcAddress := range masterGrpcAddresses {
-		err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+		err = WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 			client := master_pb.NewSeaweedClient(grpcConnection)
 			return fn(client)
 		}, masterGrpcAddress.ToGrpcAddress(), false, grpcDialOption)
@@ -527,7 +551,7 @@ func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[stri
 
 func WithBrokerGrpcClient(streamingMode bool, brokerGrpcAddress string, grpcDialOption grpc.DialOption, fn func(client mq_pb.SeaweedMessagingClient) error) error {
 
-	return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+	return WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 		client := mq_pb.NewSeaweedMessagingClient(grpcConnection)
 		return fn(client)
 	}, brokerGrpcAddress, false, grpcDialOption)
@@ -542,7 +566,7 @@ func WithFilerClient(streamingMode bool, signature int32, filer ServerAddress, g
 
 func WithGrpcFilerClient(streamingMode bool, signature int32, filerAddress ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) error {
 
-	return WithGrpcClient(streamingMode, signature, func(grpcConnection *grpc.ClientConn) error {
+	return WithGrpcClient(context.Background(), streamingMode, signature, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, filerAddress.ToGrpcAddress(), false, grpcDialOption)
@@ -552,7 +576,7 @@ func WithGrpcFilerClient(streamingMode bool, signature int32, filerAddress Serve
 func WithOneOfGrpcFilerClients(streamingMode bool, filerAddresses []ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) (err error) {
 
 	for _, filerAddress := range filerAddresses {
-		err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
+		err = WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
 			client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 			return fn(client)
 		}, filerAddress.ToGrpcAddress(), false, grpcDialOption)

weed/pb/grpc_client_server_test.go

@@ -1,6 +1,7 @@
 package pb
 
 import (
+	"context"
 	"fmt"
 	"runtime"
 	"testing"
@@ -19,24 +20,59 @@ func TestShouldInvalidateConnection_MarshalErrorIsPerRequest(t *testing.T) {
 	// outgoing request contains invalid UTF-8 bytes.
 	marshalErr := status.Error(codes.Internal,
 		"grpc: error while marshaling: string field contains invalid UTF-8")
-	if shouldInvalidateConnection(marshalErr) {
+	if shouldInvalidateConnection(context.Background(), marshalErr) {
 		t.Fatalf("client-side marshal error must not invalidate the shared connection")
 	}
 
 	// Same error wrapped with fmt.Errorf (common when callers add context).
 	wrapped := fmt.Errorf("upload data: %w", marshalErr)
-	if shouldInvalidateConnection(wrapped) {
+	if shouldInvalidateConnection(context.Background(), wrapped) {
 		t.Fatalf("wrapped marshal error must not invalidate the shared connection")
 	}
 }
 
+// TestShouldInvalidateConnection_CallerContextExpiryIsPerRequest ensures that a
+// Canceled/DeadlineExceeded caused by the caller's own context expiring does NOT
+// tear down the shared cached ClientConn. Doing so would cancel every other
+// in-flight RPC on it with "the client connection is closing" — the cascade
+// that turned one slow chunk assign into a flood of failures during a
+// high-concurrency upload (seaweedfs#9765).
+func TestShouldInvalidateConnection_CallerContextExpiryIsPerRequest(t *testing.T) {
+	expired, cancel := context.WithCancel(context.Background())
+	cancel()
+	for _, code := range []codes.Code{codes.Canceled, codes.DeadlineExceeded} {
+		err := status.Error(code, "context expired")
+		if shouldInvalidateConnection(expired, err) {
+			t.Fatalf("%v with an expired caller context must not invalidate the shared connection", code)
+		}
+	}
+}
+
+// TestShouldInvalidateConnection_StaleChannelStillInvalidates ensures the
+// carve-out above is gated on the caller's context: a Canceled/DeadlineExceeded
+// while the context is still live is the genuine stale-channel signal (e.g. a
+// peer restart behind a k8s Service VIP) and must still invalidate so the next
+// attempt dials fresh.
+func TestShouldInvalidateConnection_StaleChannelStillInvalidates(t *testing.T) {
+	for _, code := range []codes.Code{codes.Canceled, codes.DeadlineExceeded} {
+		err := status.Error(code, "the client connection is closing")
+		if !shouldInvalidateConnection(context.Background(), err) {
+			t.Fatalf("%v with a live caller context must still invalidate the connection", code)
+		}
+		// nil context is treated as live for the context-free WithGrpcClient path.
+		if !shouldInvalidateConnection(nil, err) {
+			t.Fatalf("%v with a nil caller context must still invalidate the connection", code)
+		}
+	}
+}
+
 // TestShouldInvalidateConnection_GenuineInternalStillInvalidates ensures the
 // marshal-error carve-out does not swallow real server-side Internal errors,
 // which previously caused — and should continue to cause — connection
 // invalidation.
 func TestShouldInvalidateConnection_GenuineInternalStillInvalidates(t *testing.T) {
 	serverInternal := status.Error(codes.Internal, "stream terminated by RST_STREAM with code 2")
-	if !shouldInvalidateConnection(serverInternal) {
+	if !shouldInvalidateConnection(context.Background(), serverInternal) {
 		t.Fatalf("genuine server-side Internal must still invalidate the connection")
 	}
 }
@@ -50,7 +86,7 @@ func TestShouldInvalidateConnection_TransportErrorsStillInvalidate(t *testing.T)
 		"dial tcp: connection refused",
 		"read: connection reset by peer",
 	} {
-		if !shouldInvalidateConnection(fmt.Errorf("%s", msg)) {
+		if !shouldInvalidateConnection(context.Background(), fmt.Errorf("%s", msg)) {
 			t.Fatalf("transport error %q must still invalidate", msg)
 		}
 	}

weed/replication/sink/filersink/fetch_write.go

@@ -509,7 +509,7 @@ var _ = filer_pb.FilerClient(&FilerSink{})
 
 func (fs *FilerSink) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, fs.grpcAddress, false, fs.grpcDialOption)

weed/replication/source/filer_source.go

@@ -149,7 +149,7 @@ var _ = filer_pb.FilerClient(&FilerSource{})
 
 func (fs *FilerSource) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, fs.grpcAddress, false, fs.grpcDialOption)

weed/s3api/s3api_handlers.go

@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"encoding/base64"
 	"errors"
 	"fmt"
@@ -24,7 +25,7 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea
 
 	// Fallback to direct connection if filerClient not initialized
 	// This should only happen during initialization or testing
-	return pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, s3a.getFilerAddress().ToGrpcAddress(), false, s3a.option.GrpcDialOption)
@@ -37,7 +38,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(streamingMode bool, fn func(file
 	currentFiler := s3a.filerClient.GetCurrentFiler()
 
 	// Try current filer first (fast path)
-	err := pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
+	err := pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, currentFiler.ToGrpcAddress(), false, s3a.option.GrpcDialOption)
@@ -70,7 +71,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(streamingMode bool, fn func(file
 			continue
 		}
 
-		err = pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
+		err = pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error {
 			client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 			return fn(client)
 		}, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption)

weed/server/filer_grpc_server_admin.go

@@ -117,7 +117,7 @@ func (fs *FilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (res
 		})
 	}
 	if req.TargetType == cluster.MasterType {
-		pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), fs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), fs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			pingResp, err := client.Ping(ctx, &master_pb.PingRequest{})
 			if pingResp != nil {
 				resp.RemoteTimeNs = pingResp.StartTimeNs

weed/server/filer_server.go

@@ -312,7 +312,7 @@ func (fs *FilerServer) checkWithMaster() {
 	for !isConnected {
 		fs.option.Masters.RefreshBySrvIfAvailable()
 		for _, master := range fs.option.Masters.GetInstances() {
-			readErr := operation.WithMasterServerClient(false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+			readErr := operation.WithMasterServerClient(context.Background(), false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 				resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 				if err != nil {
 					return fmt.Errorf("get master %s configuration: %v", master, err)

weed/server/master_grpc_server_admin.go

@@ -222,7 +222,7 @@ func (ms *MasterServer) Ping(ctx context.Context, req *master_pb.PingRequest) (r
 		})
 	}
 	if req.TargetType == cluster.MasterType {
-		pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), ms.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), ms.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			pingResp, err := client.Ping(ctx, &master_pb.PingRequest{})
 			if pingResp != nil {
 				resp.RemoteTimeNs = pingResp.StartTimeNs

weed/server/master_server.go

@@ -526,7 +526,7 @@ func (ms *MasterServer) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, startF
 				hashicorpRaft.ServerAddress(peerAddress.ToGrpcAddress()), 0, 0)
 		}
 	} else {
-		pb.WithMasterClient(false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error {
+		pb.WithMasterClient(context.Background(), false, peerAddress, ms.grpcDialOption, true, func(client master_pb.SeaweedClient) error {
 			ctx, cancel := context.WithTimeout(context.TODO(), 15*time.Second)
 			defer cancel()
 			if _, err := client.Ping(ctx, &master_pb.PingRequest{Target: string(peerAddress), TargetType: cluster.MasterType}); err != nil {

weed/server/volume_grpc_admin.go

@@ -280,7 +280,7 @@ func (vs *VolumeServer) makeVolumeWritable(ctx context.Context, v *storage.Volum
 }
 
 func (vs *VolumeServer) notifyMasterVolumeReadonly(ctx context.Context, v *storage.Volume, isReadOnly bool) error {
-	if grpcErr := pb.WithMasterClient(false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+	if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(ctx), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 		_, err := client.VolumeMarkReadonly(context.Background(), &master_pb.VolumeMarkReadonlyRequest{
 			Ip:               vs.store.Ip,
 			Port:             uint32(vs.store.Port),
@@ -490,7 +490,7 @@ func (vs *VolumeServer) Ping(ctx context.Context, req *volume_server_pb.PingRequ
 		})
 	}
 	if req.TargetType == cluster.MasterType {
-		pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			pingResp, err := client.Ping(ctx, &master_pb.PingRequest{})
 			if pingResp != nil {
 				resp.RemoteTimeNs = pingResp.StartTimeNs

weed/server/volume_grpc_client_to_master.go

@@ -47,7 +47,7 @@ func (vs *VolumeServer) setCurrentMaster(master pb.ServerAddress) {
 func (vs *VolumeServer) checkWithMaster() (err error) {
 	for {
 		for _, master := range vs.SeedMasterNodes {
-			err = operation.WithMasterServerClient(false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+			err = operation.WithMasterServerClient(context.Background(), false, master, vs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 				resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 				if err != nil {
 					return fmt.Errorf("get master %s configuration: %v", master, err)

weed/server/volume_grpc_copy.go

@@ -93,7 +93,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre
 		}()
 
 		var preallocateSize int64
-		if grpcErr := pb.WithMasterClient(false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 			if err != nil {
 				return fmt.Errorf("get master %s configuration: %v", vs.GetMaster(context.Background()), err)

weed/server/webdav_server.go

@@ -145,7 +145,7 @@ var _ = filer_pb.FilerClient(&WebDavFileSystem{})
 
 func (fs *WebDavFileSystem) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return pb.WithGrpcClient(streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, fs.signature, func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, fs.option.Filer.ToGrpcAddress(), false, fs.option.GrpcDialOption)

weed/sftpd/sftp_filer.go

@@ -89,7 +89,7 @@ func (fs *SftpServer) AdjustedUrl(location *filer_pb.Location) string { return l
 func (fs *SftpServer) GetDataCenter() string                          { return fs.dataCenter }
 func (fs *SftpServer) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 	addr := fs.filerAddr.ToGrpcAddress()
-	return pb.WithGrpcClient(streamingMode, util.RandomInt32(), func(conn *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), streamingMode, util.RandomInt32(), func(conn *grpc.ClientConn) error {
 		return fn(filer_pb.NewSeaweedFilerClient(conn))
 	}, addr, false, fs.grpcDialOption)
 }

weed/shell/command_cluster_check.go

@@ -114,7 +114,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i
 	for _, master := range masters {
 		for _, volumeServer := range volumeServers {
 			fmt.Fprintf(writer, "checking master %s to volume server %s ... ", string(master), string(volumeServer))
-			err := pb.WithMasterClient(false, master, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error {
+			err := pb.WithMasterClient(context.Background(), false, master, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error {
 				pong, err := client.Ping(context.Background(), &master_pb.PingRequest{
 					Target:     string(volumeServer),
 					TargetType: cluster.VolumeServerType,
@@ -137,7 +137,7 @@ func (c *commandClusterCheck) Do(args []string, commandEnv *CommandEnv, writer i
 				continue
 			}
 			fmt.Fprintf(writer, "checking master %s to %s ... ", string(sourceMaster), string(targetMaster))
-			err := pb.WithMasterClient(false, sourceMaster, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error {
+			err := pb.WithMasterClient(context.Background(), false, sourceMaster, commandEnv.option.GrpcDialOption, false, func(client master_pb.SeaweedClient) error {
 				pong, err := client.Ping(context.Background(), &master_pb.PingRequest{
 					Target:     string(targetMaster),
 					TargetType: cluster.MasterType,

weed/shell/command_s3_iam_client.go

@@ -24,7 +24,7 @@ const iamRequestTimeout = 30 * time.Second
 // applied — callers can derive child contexts but should not need their own
 // timeout boilerplate.
 func (ce *CommandEnv) withIamClient(fn func(ctx context.Context, client iam_pb.SeaweedIdentityAccessManagementClient) error) error {
-	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error {
 		ctx, cancel := context.WithTimeout(iamAdminAuthContext(context.Background()), iamRequestTimeout)
 		defer cancel()
 		return fn(ctx, iam_pb.NewSeaweedIdentityAccessManagementClient(conn))

weed/shell/s3tables_helpers.go

@@ -17,7 +17,7 @@ const s3TablesDefaultRegion = ""
 const timeFormat = "2006-01-02T15:04:05Z07:00"
 
 func withFilerClient(commandEnv *CommandEnv, fn func(client filer_pb.SeaweedFilerClient) error) error {
-	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+	return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(conn)
 		return fn(client)
 	}, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption)

weed/storage/store_ec.go

@@ -461,7 +461,7 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume)
 
 	glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId)
 
-	err = operation.WithMasterServerClient(false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+	err = operation.WithMasterServerClient(context.Background(), false, s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 		req := &master_pb.LookupEcVolumeRequest{
 			VolumeId: uint32(ecVolume.VolumeId),
 		}

weed/wdclient/masterclient.go

@@ -75,7 +75,7 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []
 				return status.Errorf(codes.Unavailable, "no master available")
 			}
 
-			return pb.WithMasterClient(false, master, p.masterClient.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+			return pb.WithMasterClient(timeoutCtx, false, master, p.masterClient.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 				resp, err := client.LookupVolume(timeoutCtx, &master_pb.LookupVolumeRequest{
 					VolumeOrFileIds: volumeIds,
 				})
@@ -224,7 +224,7 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server
 	glog.V(1).Infof("%s.%s masterClient Connecting to master %v", mc.FilerGroup, mc.clientType, master)
 	stats.MasterClientConnectCounter.WithLabelValues("total").Inc()
 	connectStartTime := time.Now()
-	gprcErr := pb.WithMasterClient(true, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+	gprcErr := pb.WithMasterClient(context.Background(), true, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
 
@@ -394,7 +394,7 @@ func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.
 
 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(streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		return pb.WithMasterClient(context.Background(), streamingMode, getMasterF(), mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			return fn(client)
 		})
 	})
@@ -511,7 +511,7 @@ func (mc *MasterClient) FindLeaderFromOtherPeers(myMasterAddress pb.ServerAddres
 		if master == myMasterAddress {
 			continue
 		}
-		if grpcErr := pb.WithMasterClient(false, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		if grpcErr := pb.WithMasterClient(context.Background(), false, master, mc.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 			ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
 			defer cancel()
 			resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})