s3: stop one abandoned request from cancelling every concurrent upload (#10948)

* grpc: a non-cancellable context is no evidence of a stale channel

shouldInvalidateConnection only invalidates on Canceled/DeadlineExceeded
while the context handed to WithGrpcClient is still live, so that an RPC
timing out on its own does not close the shared cached ClientConn and
cancel every other in-flight RPC on it. context.Background()/TODO never
expire, so Err() stays nil forever and that guard always answered
"invalidate" - and Background is what almost every caller passes, the S3
gateway included.

One S3 request whose RPC rode an abandoned HTTP request context therefore
closed the shared filer connection, and every multipart part in flight
died with "the client connection is closing", surfacing to the client as
400 InvalidRequest.

Only a cancellable context bounds an RPC attempt, so require one before
reading it. A genuinely stale channel (a peer restart behind a stable L4
endpoint) surfaces as Unavailable, which invalidates on its own branch.

* grpc: a bystander of a connection teardown is not a stale-channel witness

gRPC raises ErrClientConnClosing locally, before an RPC reaches the wire,
when this process has already closed the ClientConn. Every caller that
touches a channel during another goroutine's teardown gets it, so reading
it as a stale-channel signal lets one teardown re-arm itself across the
whole herd of callers it just cancelled.

The cached-connection version check keeps those callers from closing a
replacement channel, but the streaming path invalidates by address alone
and has no such guard.

* grpc: end a stream without dropping the peer connection under it

A streaming caller gets its own ClientConn, but on any error it also drops
the cached non-streaming ClientConn every request handler shares with that
peer, to recover a peer restart hidden behind a stable L4 endpoint. Any
error includes the ordinary ones: a metadata subscription that reached its
stop point, a follow callback that refused an event, a caller that gave up.

The S3 gateway follows filer metadata on such a stream and reconnects
forever, so each ordinary end of it cancelled every S3 request in flight
against the filer. Drop the shared channel only for errors that say the
peer went away, which is what invalidation is for.

* test: close the connections the cascade tests leave cached

Each test swaps in a fresh connection cache and restores the previous one,
dropping its own entries without closing them, so the ClientConn's
transport and reconnect goroutines outlive the fake filer they dialed.

* grpc: say why ErrClientConnClosing's deprecation notice does not apply

It points at codes.Canceled, which is the code this function exists to
disambiguate. Only the message distinguishes a teardown a caller merely
walked into, so the sentinel stays.

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

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)

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