fix(filer): drop stale master gRPC cache on stream death (#9102) (#9107)

* 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

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

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