diff --git a/weed/pb/grpc_client_cascade_test.go b/weed/pb/grpc_client_cascade_test.go new file mode 100644 index 0000000000..33acaa5e06 --- /dev/null +++ b/weed/pb/grpc_client_cascade_test.go @@ -0,0 +1,147 @@ +package pb + +import ( + "context" + "net" + "sync" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// cascadeFilerServer answers a "slow" lookup only once its caller is known to be +// in flight, so a second caller can fail while the first still holds the shared +// cached ClientConn. +type cascadeFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + inFlight chan struct{} + release chan struct{} + closeOnce sync.Once +} + +func (s *cascadeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if req.Name == "slow" { + s.closeOnce.Do(func() { close(s.inFlight) }) + select { + case <-s.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return &filer_pb.LookupDirectoryEntryResponse{}, nil +} + +// startCascadeFiler serves a fake filer and resets the process-wide connection +// cache so the test owns the shared ClientConn under exercise. +func startCascadeFiler(t *testing.T) (*cascadeFilerServer, string, grpc.DialOption) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + fake := &cascadeFilerServer{inFlight: make(chan struct{}), release: make(chan struct{})} + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, fake) + go server.Serve(listener) + + grpcClientsLock.Lock() + previous := grpcClients + grpcClients = make(map[string]*versionedGrpcClient) + grpcClientsLock.Unlock() + t.Cleanup(func() { + grpcClientsLock.Lock() + cached := grpcClients + grpcClients = previous + grpcClientsLock.Unlock() + for _, connection := range cached { + connection.Close() + } + server.Stop() + }) + + return fake, listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()) +} + +// TestWithGrpcClient_AbandonedRequestKeepsSharedConnection reproduces +// seaweedfs#10947 end to end: an S3-style caller that hands WithGrpcClient a +// context.Background() while its RPC rides the (now abandoned) HTTP request +// context used to close the shared filer ClientConn, killing every concurrent +// upload on it with "the client connection is closing". +func TestWithGrpcClient_AbandonedRequestKeepsSharedConnection(t *testing.T) { + fake, address, dialOption := startCascadeFiler(t) + + var concurrentErr error + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + concurrentErr = WithGrpcClient(context.Background(), false, 0, func(connection *grpc.ClientConn) error { + _, err := filer_pb.NewSeaweedFilerClient(connection).LookupDirectoryEntry(context.Background(), + &filer_pb.LookupDirectoryEntryRequest{Directory: "/buckets/b", Name: "slow"}) + return err + }, address, false, dialOption) + }() + <-fake.inFlight + + abandoned, cancel := context.WithCancel(context.Background()) + cancel() + if err := WithGrpcClient(context.Background(), false, 0, func(connection *grpc.ClientConn) error { + _, err := filer_pb.NewSeaweedFilerClient(connection).LookupDirectoryEntry(abandoned, + &filer_pb.LookupDirectoryEntryRequest{Directory: "/buckets/b", Name: "fast"}) + return err + }, address, false, dialOption); err == nil { + t.Fatal("the abandoned request should have failed") + } + + close(fake.release) + wg.Wait() + if concurrentErr != nil { + t.Fatalf("concurrent caller must survive an unrelated abandoned request: %v", concurrentErr) + } +} + +func (s *cascadeFilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error { + return nil +} + +// TestWithGrpcClient_EndedStreamKeepsSharedConnection covers the other half of +// seaweedfs#10947: the S3 gateway follows filer metadata on a long-lived +// SubscribeMetadata stream and reconnects forever. Every ordinary end of that +// stream used to drop the shared filer ClientConn the request handlers use, +// firing the same burst of "the client connection is closing" failures. +func TestWithGrpcClient_EndedStreamKeepsSharedConnection(t *testing.T) { + fake, address, dialOption := startCascadeFiler(t) + + var concurrentErr error + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + concurrentErr = WithGrpcClient(context.Background(), false, 0, func(connection *grpc.ClientConn) error { + _, err := filer_pb.NewSeaweedFilerClient(connection).LookupDirectoryEntry(context.Background(), + &filer_pb.LookupDirectoryEntryRequest{Directory: "/buckets/b", Name: "slow"}) + return err + }, address, false, dialOption) + }() + <-fake.inFlight + + if err := WithGrpcClient(context.Background(), true, 0, func(connection *grpc.ClientConn) error { + stream, err := filer_pb.NewSeaweedFilerClient(connection).SubscribeMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{}) + if err != nil { + return err + } + _, err = stream.Recv() + return err + }, address, false, dialOption); err == nil { + t.Fatal("the ended stream should have reported io.EOF") + } + + close(fake.release) + wg.Wait() + if concurrentErr != nil { + t.Fatalf("concurrent caller must survive an unrelated stream ending: %v", concurrentErr) + } +} diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index 878775b9b7..cd1a2a81f9 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -386,9 +386,9 @@ func isClientSideMarshalError(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. +// should be invalidated. ctx must be the context that bounds this RPC attempt; it +// disambiguates a genuinely broken channel from the RPC hitting its own deadline +// or cancellation. A nil or non-cancellable ctx carries no such evidence. func shouldInvalidateConnection(ctx context.Context, err error) bool { if err == nil { return false @@ -401,6 +401,16 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { return false } + // gRPC raises this locally, before the RPC reaches the wire, when this + // process already closed the ClientConn. The caller is a bystander of + // someone else's teardown, and knows nothing about the peer. Its message + // is the only thing separating it from any other Canceled, so the plain + // codes.Canceled its deprecation notice recommends cannot stand in: that + // is the very code this function has to keep telling apart below. + if errors.Is(err, grpc.ErrClientConnClosing) { + return false + } + // Check gRPC status codes first (more reliable) if s, ok := status.FromError(err); ok { code := s.Code() @@ -415,16 +425,15 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { // the caller retry. return false 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 + // Ambiguous: a stale cached channel rejects RPCs this way (a peer + // restart behind a k8s Service VIP), and so does the RPC's own context + // expiring. Tearing the channel down for the latter cancels every other + // in-flight RPC on it with "the client connection is closing", turning + // one abandoned request into a flood of failures. Only a cancellable + // context bounds an RPC attempt and can tell the two apart: + // Background/TODO never expires, so its Err() would answer + // "stale channel" for every caller with no deadline to honor. + return ctx != nil && ctx.Done() != nil && ctx.Err() == nil } } @@ -441,10 +450,11 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { } // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection. -// 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. +// 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. func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error { if !streamingMode { @@ -481,14 +491,15 @@ func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn } defer grpcConnection.Close() executionErr := fn(grpcConnection) - if executionErr != nil { + if executionErr != nil && shouldInvalidateConnection(ctx, executionErr) { // 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. + // same peer. When the stream dies of a peer that went away, 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. A stream that merely + // ended, or that its own caller gave up on, says nothing about the peer. InvalidateGrpcConnection(address) } return executionErr @@ -552,10 +563,10 @@ func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) { return util.JoinHostPort(host, port) } -// WithMasterClient threads the caller's per-request context into the +// WithMasterClient threads the context bounding this RPC attempt 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. +// attempt's own timeout does not invalidate the shared cached master connection. +// Pass context.Background() when fn picks its own context. 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) diff --git a/weed/pb/grpc_client_server_test.go b/weed/pb/grpc_client_server_test.go index 89b6d6ea5d..a544392c86 100644 --- a/weed/pb/grpc_client_server_test.go +++ b/weed/pb/grpc_client_server_test.go @@ -6,6 +6,7 @@ import ( "runtime" "testing" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -49,19 +50,50 @@ func TestShouldInvalidateConnection_CallerContextExpiryIsPerRequest(t *testing.T } // 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 +// carve-out above is gated on the RPC's context: a Canceled/DeadlineExceeded +// while that 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) { + live, cancel := context.WithCancel(context.Background()) + defer cancel() 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) + err := status.Error(code, "stale channel") + if !shouldInvalidateConnection(live, err) { + t.Fatalf("%v with a live attempt 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_NonCancellableContextIsNoEvidence covers +// seaweedfs#10947. Background/TODO never expires, so Err() stays nil forever and +// the guard above answered "stale channel" for every caller that has no deadline +// to honor — which is almost all of them, the S3 gateway included. One client +// abandoning a request then closed the shared filer channel and every concurrent +// multipart part died with "the client connection is closing". +func TestShouldInvalidateConnection_NonCancellableContextIsNoEvidence(t *testing.T) { + for _, ctx := range []context.Context{context.Background(), context.TODO(), nil} { + for _, code := range []codes.Code{codes.Canceled, codes.DeadlineExceeded} { + err := status.Error(code, "context canceled") + if shouldInvalidateConnection(ctx, err) { + t.Fatalf("%v must not invalidate the shared connection on a non-cancellable context", code) + } + } + } +} + +// TestShouldInvalidateConnection_ClientConnClosingIsNotStale ensures a caller +// whose RPC was rejected because this process is already closing the shared +// ClientConn does not close it again. gRPC raises ErrClientConnClosing locally, +// so it is only ever the footprint of another goroutine's teardown - reading it +// as a stale-channel signal lets one teardown re-arm itself across the herd of +// callers it just cancelled. +func TestShouldInvalidateConnection_ClientConnClosingIsNotStale(t *testing.T) { + live, cancel := context.WithCancel(context.Background()) + defer cancel() + for _, err := range []error{grpc.ErrClientConnClosing, fmt.Errorf("assign volume: %w", grpc.ErrClientConnClosing)} { + if shouldInvalidateConnection(live, err) { + t.Fatalf("%v must not invalidate the shared connection", err) } } }