Repository: grpc/grpc-go

Issue:
Failing to delete the stream from the activeStreams map leading to REFUSED_STREAM errors
Original PR with repro: https://github.com/grpc/grpc-go/pull/8504

After upgrading `go-grpc`, we started see the following error:

```
stream terminated by RST_STREAM with error code: REFUSED_STREAM
```

Upon investigation, it appears that in some scenarios, we fail to properly delete streams from the [`activeStreams`](https://github.com/grpc/grpc-go/blob/57b69b47a2da17098c387c8c64c4aab5295cbd3e/internal/transport/http2_server.go#L1309) map, resulting in this error being wrongly returned.

I added a test case that seems to trigger the problem:

PS: To test the behavior i added a "ActiveStreamTracker" hook -> im not sure this is the best way but i wanted just to show the problem

```
end2end_test.go:4015: leak streams:  5
```

The issue seems to have been introduced in the following PR: https://github.com/grpc/grpc-go/pull/8071

Specifically, the check added here:
[https://github.com/grpc/grpc-go/blob/57b69b47a2da17098c387c8c64c4aab5295cbd3e/internal/transport/http2\_server.go#L1357-L1359](https://github.com/grpc/grpc-go/blob/57b69b47a2da17098c387c8c64c4aab5295cbd3e/internal/transport/http2_server.go#L1357-L1359)

If this conditional is removed, all streams are correctly cleaned up from the `activeStreams` map, and the test passes without leaks.

## Root cause
1. The server stream exhausts the stream quota, and starts buffering data frames.
2. The server attempts to finish the RPC gracefully by sending trailers by calling [finishStream](https://github.com/grpc/grpc-go/blob/19c720f666f3d405dec68363733f19e9f80ea7bf/internal/transport/http2_server.go#L1326-L1347).
3. The write of the header frame [gets blocked](https://github.com/grpc/grpc-go/blob/19c720f666f3d405dec68363733f19e9f80ea7bf/internal/transport/controlbuf.go#L692-L696) due to the pending data frames that are awaiting flow control quota.
4. Once the deadline expires, the [server's deadline monitoring timer](https://github.com/grpc/grpc-go/blob/19c720f666f3d405dec68363733f19e9f80ea7bf/internal/transport/http2_server.go#L627-L630) tries to call closeStream, which is a no-op since the stream state is already marked `streamDone` by `finishStream` in point 1.
5. The client sends and RST stream, [the server calls closeStream](https://github.com/grpc/grpc-go/blob/19c720f666f3d405dec68363733f19e9f80ea7bf/internal/transport/http2_server.go#L865-L870) which is again a no-op.

The simplest fix is to remove the guard block added in `closeStream` in https://github.com/grpc/grpc-go/pull/8071. We may end up pushing redundant `cleanupStream` events on to the controlbuf, but it doesn't cause correctness issues since [cleanupStreamHandler returns early](https://github.com/grpc/grpc-go/blob/19c720f666f3d405dec68363733f19e9f80ea7bf/internal/transport/controlbuf.go#L803-L805) in such cases.



Hints from issue comments:
@arjan-bal: @alanprot will be [raising a PR ](https://github.com/grpc/grpc-go/pull/8504#issuecomment-3191445324 )with the fix@arjan-bal: We are delaying release v1.75 to get this fix in.@alanprot: Question: Do we think that is ok to create [this](https://github.com/grpc/grpc-go/pull/8504/files#diff-67213f39b84a013e7db04664cd3ad943b7a237d97b9c6e862d10806ac27def28R248) hook (or similar) in order to be able to run the unit test?@arjan-bal: Avoiding test only hooks in non-test code would be preferable. I can think of the following way to write a test:
1. create a server which allows 1 concurrent stream using [the server option](https://pkg.go.dev/google.golang.org/grpc#MaxConcurrentStreams).
2. create a client and start a bi-di stream.
3. Have the server send a response message of size > [65535 bytes](https://github.com/grpc/grpc-go/blob/33ec81b40e51ce07a92bee417656da502cdb05b9/internal/transport/defaults.go#L28) to exhaust the client's flow control window. Have the server send another message, which should be queued on the server.
4.  Ensure the client DOES NOT read any message.
5. Have the server handler return, ending the stream. This should cause the bug to be triggered.
6. Have the client cancel the RPC by cancelling the stream context. 
7. Try creating a single new stream for short duration, say 5 millis. Read one message from the stream. Every time we should get an error CONNECTION_REFUSED without the fix.

The test may be flaky due to a race b/w step 5 & 6. Even if it fails 50% of the time, it should be good enough. With the fix, the test should always pass.@arjan-bal: @alanprot I suspect the test might be more complicated than I initially estimated. If you'd like, you can open a PR with just the fix, and I can handle the test in a separate PR.@alanprot: Idk if I will have time today tbh.. feel free to open the pr with the test and fix.. :) 

Thanks again for looking into this :)@arjan-bal: Sounds good, I'll raise a PR.

Solution patch:
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 9f725e15..83cee314 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -1353,10 +1353,10 @@ func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCo
 	// called to interrupt the potential blocking on other goroutines.
 	s.cancel()
 
-	oldState := s.swapState(streamDone)
-	if oldState == streamDone {
-		return
-	}
+	// We can't return early even if the stream's state is "done" as the state
+	// might have been set by the `finishStream` method. Deleting the stream via
+	// `finishStream` can get blocked on flow control.
+	s.swapState(streamDone)
 	t.deleteStream(s, eosReceived)
 
 	t.controlBuf.put(&cleanupStream{



Test patch:
diff --git a/test/transport_test.go b/test/transport_test.go
index 10e9ab57..7a8b7581 100644
--- a/test/transport_test.go
+++ b/test/transport_test.go
@@ -229,3 +229,76 @@ func (s) TestRSTDuringMessageRead(t *testing.T) {
 		t.Fatalf("client.EmptyCall() returned %v; want status with code %v", err, codes.Canceled)
 	}
 }
+
+// Test verifies that a client-side cancellation correctly frees up resources on
+// the server. The test setup is designed to simulate a scenario where a server
+// is blocked from sending a large message due to a full client-side flow
+// control window. The client-side cancellation of this blocked RPC then frees
+// up the max concurrent streams quota on the server, allowing a new RPC to be
+// created successfully.
+func (s) TestCancelWhileServerWaitingForFlowControl(t *testing.T) {
+	serverDoneCh := make(chan struct{}, 2)
+	const flowControlWindowSize = 65535
+	ss := &stubserver.StubServer{
+		StreamingOutputCallF: func(_ *testpb.StreamingOutputCallRequest, stream testpb.TestService_StreamingOutputCallServer) error {
+			// Send a large message to exhaust the client's flow control window.
+			stream.Send(&testpb.StreamingOutputCallResponse{
+				Payload: &testpb.Payload{
+					Body: make([]byte, flowControlWindowSize+1),
+				},
+			})
+			serverDoneCh <- struct{}{}
+			return nil
+		},
+	}
+
+	// Create a server that allows only 1 stream at a time.
+	ss = stubserver.StartTestService(t, ss, grpc.MaxConcurrentStreams(1))
+	defer ss.Stop()
+	// Use a static flow control window.
+	if err := ss.StartClient(grpc.WithStaticStreamWindowSize(flowControlWindowSize)); err != nil {
+		t.Fatalf("Error while start test service client: %v", err)
+	}
+	client := ss.Client
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	streamCtx, streamCancel := context.WithCancel(ctx)
+	defer streamCancel()
+
+	if _, err := client.StreamingOutputCall(streamCtx, &testpb.StreamingOutputCallRequest{}); err != nil {
+		t.Fatalf("Failed to create server streaming RPC: %v", err)
+	}
+
+	// Wait for the server handler to return. This should cause the trailers to
+	// be buffered on the server, waiting for flow control quota to first send
+	// the data frame.
+	select {
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for server handler to return.")
+	case <-serverDoneCh:
+	}
+
+	// Attempt to create a stream. It should fail since the previous stream is
+	// still blocked.
+	shortCtx, shortCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer shortCancel()
+	_, err := client.StreamingOutputCall(shortCtx, &testpb.StreamingOutputCallRequest{})
+	if status.Code(err) != codes.DeadlineExceeded {
+		t.Fatalf("Server stream creation returned error with unexpected status code: %v, want code: %v", err, codes.DeadlineExceeded)
+	}
+
+	// Cancel the RPC, this should free up concurrent stream quota on the
+	// server.
+	streamCancel()
+
+	// Attempt to create another stream.
+	stream, err := client.StreamingOutputCall(ctx, &testpb.StreamingOutputCallRequest{})
+	if err != nil {
+		t.Fatalf("Failed to create server streaming RPC: %v", err)
+	}
+	_, err = stream.Recv()
+	if err != nil {
+		t.Fatalf("Failed to read from the stream: %v", err)
+	}
+}