From ff4f96c71fc51c0d60632fffef94e22f571ec8e7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 16 Apr 2026 12:10:25 -0700 Subject: [PATCH] fix(filer): drop stale master gRPC cache on stream death (#9102) (#9107) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(filer): drop stale master gRPC cache on stream death (#9102) When the master server restarts behind a stable L4 endpoint (e.g. a Kubernetes ClusterIP Service), the filer's streaming KeepConnected channel detects the disconnect and reconnects, but the shared request-path ClientConn cached in pb.grpcClients can remain in READY state while actually being dead. New AssignVolume/LookupVolume calls reuse that cached channel and return `rpc error: code = Canceled desc = context canceled` for every request, until the filer pod is restarted. - Expose pb.InvalidateGrpcConnection(address) to drop a cached ClientConn when a higher-level signal says it is stale. - In MasterClient.tryConnectToMaster, invalidate the cached request-path channel whenever the KeepConnected stream returns, so unrelated callers dial fresh on their next RPC. - Extend operation.Assign's retry predicate to cover Canceled and DeadlineExceeded while the caller context is still live: the first failure invalidates the stale ClientConn via shouldInvalidateConnection, and the retry dials a new channel. * fix(grpc): invalidate cached peer conn on streaming death in other paths Extends the master-client fix to the other streaming-caller + cached non-streaming-peer pairs that share the same stale-channel failure mode when the peer restarts behind a stable L4 endpoint (k8s Service VIP, external load balancer): - pb.FollowMetadata (s3, mount, webdav, mq broker, filer remote gateway, etc. → filer): invalidate the filer's cached ClientConn when the SubscribeMetadata stream returns an error. - filer.MetaAggregator.loopSubscribeToOneFiler (filer → peer filer): invalidate the peer's cached ClientConn after doSubscribeToOneFiler fails, so the next iteration's readFilerStoreSignature / updateOffset calls dial fresh. - mq sub_client.onEachPartition and doKeepConnectedToSubCoordinator (subscriber → broker): invalidate the broker's cached ClientConn when the SubscribeMessage / SubscriberToSubCoordinator stream errors. - mq broker.BrokerConnectToBalancer (broker → broker-balancer): invalidate the balancer's cached ClientConn after the PublisherToPubBalancer stream errors. * address review feedback on InvalidateGrpcConnection - pb.InvalidateGrpcConnection: drop the cache entry under grpcClientsLock but call ClientConn.Close() after releasing the lock, so Close's internal synchronisation/IO doesn't serialise unrelated callers on the global map lock. - wdclient.tryConnectToMaster: only invalidate the cached request-path channel when the streaming call returned an error. On a healthy leader redirect (gprcErr == nil) the cached channel is still usable and invalidating it just causes a needless re-dial from concurrent callers. * refactor(grpc): centralize peer-conn invalidation in streaming path Previously every streaming caller duplicated the same invalidate-cached- non-streaming-peer-conn wrapper around their WithGrpcClient(true, ...) call. Move that logic into WithGrpcClient itself: when the streaming fn returns an error, invalidate any cached ClientConn for the same address. This removes six near-identical call-site wrappers and gives every current and future streaming caller the fix by default. Also aligns the non-streaming branch with the new Invalidate helper's lock discipline: delete the cache entry under grpcClientsLock, then Close the ClientConn after releasing the lock. --- weed/operation/assign_file_id.go | 17 ++++++- weed/pb/grpc_client_server.go | 79 +++++++++++++++++++++----------- 2 files changed, 68 insertions(+), 28 deletions(-) diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index 525ce515d..783d1a204 100644 --- a/weed/operation/assign_file_id.go +++ b/weed/operation/assign_file_id.go @@ -70,7 +70,22 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO lastError = util.RetryWithBackoff(deadlineCtx, "assign", remaining, func(err error) bool { st, ok := status.FromError(err) - return ok && st.Code() == codes.Unavailable + if !ok { + return false + } + switch st.Code() { + case codes.Unavailable: + return true + case codes.Canceled, codes.DeadlineExceeded: + // A stale cached gRPC channel (e.g., master restart behind + // a k8s Service VIP) can return Canceled/DeadlineExceeded + // immediately even though the caller's context is still + // live. The first failure invalidates the cached ClientConn + // via shouldInvalidateConnection; retry so the next attempt + // dials a fresh channel. + return deadlineCtx.Err() == nil + } + return false }, func() error { // Per-attempt timeout to prevent a single slow RPC from consuming the entire retry budget diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index 82b9a23f5..ddae14219 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -247,6 +247,24 @@ func requestIDUnaryInterceptor() grpc.UnaryServerInterceptor { } } +// InvalidateGrpcConnection drops any cached gRPC ClientConn for the given +// address. Use when a higher-level signal (for example a streaming master +// connection detecting its peer has died) indicates the cached channel is +// stale, even though gRPC itself may still believe the channel is healthy. +// Silently returns if no cached connection exists. +func InvalidateGrpcConnection(address string) { + grpcClientsLock.Lock() + vgc, ok := grpcClients[address] + if ok { + delete(grpcClients, address) + } + grpcClientsLock.Unlock() + if ok { + glog.V(1).Infof("Invalidating cached gRPC connection to %s", address) + vgc.Close() + } +} + // shouldInvalidateConnection checks if an error indicates the cached connection should be invalidated func shouldInvalidateConnection(err error) bool { if err == nil { @@ -283,38 +301,45 @@ 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 { - if shouldInvalidateConnection(executionErr) { - grpcClientsLock.Lock() - if t, ok := grpcClients[address]; ok { - if t.version == vgc.version { - glog.V(1).Infof("Removing cached gRPC connection to %s due to error: %v", address, executionErr) - vgc.Close() - delete(grpcClients, address) - } - } - grpcClientsLock.Unlock() + if executionErr != nil && shouldInvalidateConnection(executionErr) { + grpcClientsLock.Lock() + t, ok := grpcClients[address] + shouldClose := ok && t.version == vgc.version + if shouldClose { + delete(grpcClients, address) + } + grpcClientsLock.Unlock() + if shouldClose { + glog.V(1).Infof("Removing cached gRPC connection to %s due to error: %v", address, executionErr) + vgc.Close() } } return executionErr - } else { - ctx := context.Background() - if signature != 0 { - // Optimize: Use AppendToOutgoingContext instead of creating new map - ctx = metadata.AppendToOutgoingContext(ctx, "sw-client-id", fmt.Sprintf("%d", signature)) - } - grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...) - if err != nil { - return fmt.Errorf("fail to dial %s: %v", address, err) - } - defer grpcConnection.Close() - executionErr := fn(grpcConnection) - if executionErr != nil { - return executionErr - } - return nil } + // Streaming mode: dedicate a fresh ClientConn to this call. + ctx := context.Background() + if signature != 0 { + // Optimize: Use AppendToOutgoingContext instead of creating new map + ctx = metadata.AppendToOutgoingContext(ctx, "sw-client-id", fmt.Sprintf("%d", signature)) + } + grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...) + if err != nil { + return fmt.Errorf("fail to dial %s: %v", address, err) + } + defer grpcConnection.Close() + executionErr := fn(grpcConnection) + if executionErr != nil { + // The streaming channel is dedicated to this caller, but unrelated + // request-path callers share a cached non-streaming ClientConn to the + // same peer. When the stream fails, drop that cached channel so the + // next caller dials fresh: this recovers cases where a stable L4 + // endpoint (k8s Service VIP, external LB) hides a peer restart from + // the transport layer, leaving the cached ClientConn healthy-looking + // but silently cancelling RPCs. + InvalidateGrpcConnection(address) + } + return executionErr } func hostAndPort(address string) (host string, port uint64, err error) {