nghttp2: a peer's RST_STREAM(NO_ERROR) on a request still arriving releases it, as any other reset does

src/protocols/ioxide.nghttp2/Interop/Nghttp2.cs

@@ -34,6 +34,8 @@ internal struct Callbacks
     /// validated out of <see cref="ih2_read"/>. With <paramref name="manualFlow"/> non-zero nghttp2
     /// sends no WINDOW_UPDATE for DATA by itself - every byte handed to OnData is credited once,
     /// through <see cref="ih2_consume"/> - and OnStreamError reports every close, NO_ERROR included.
+    /// Without it, OnStreamError reports every close but the normal end of a stream: NO_ERROR once
+    /// the response's END_STREAM has gone out.
     /// Returns the handle, or 0 on failure.</summary>
     [DllImport(Lib)] internal static extern nint ih2_server_new(Callbacks callbacks, void* user, int manualFlow);
 

src/protocols/ioxide.nghttp2/native/ioxide_nghttp2_shim.c

@@ -163,13 +163,12 @@ static int on_data_chunk_recv(nghttp2_session *session, uint8_t flags, int32_t s
 static int on_stream_close(nghttp2_session *session, int32_t stream_id, uint32_t error_code,
                            void *user_data)
 {
-    (void)session;
     ih2_conn *c = user_data;
 
-    /* NO_ERROR here is the normal end of a stream whose response already completed, so only a
-     * real error is worth waking the managed side for - unless it reads request bodies itself:
-     * a peer's RST_STREAM(NO_ERROR) ends one of those too. */
-    if ((error_code != NGHTTP2_NO_ERROR || c->manual_flow) && c->cbs.on_stream_error != NULL) {
+    /* NO_ERROR before our END_STREAM went out is a peer's reset, not the normal end; manual flow hears every close. */
+    if ((error_code != NGHTTP2_NO_ERROR || c->manual_flow ||
+         nghttp2_session_get_stream_local_close(session, stream_id) == 0) &&
+        c->cbs.on_stream_error != NULL) {
         c->cbs.on_stream_error(c->user, stream_id, error_code);
     }
 

tests/Ioxide.Tests.Unit/Nghttp2StreamCloseTests.cs

@@ -0,0 +1,98 @@
+using System.Collections;
+using System.IO.Pipelines;
+using System.Reflection;
+using ioxide.nghttp2;
+
+namespace Ioxide.Tests;
+
+/// <summary>A request its peer resets before it ends leaves the connection, whatever code the reset carries.</summary>
+internal static class Nghttp2StreamCloseTests
+{
+    public static void Register(Runner runner)
+    {
+        foreach ((uint code, string name) in new[] { (0u, "NO_ERROR"), (8u, "CANCEL") })
+        {
+            uint error = code;
+            runner.Test($"nghttp2: a request its peer resets with {name} before it ends is released", () =>
+            {
+                using var client = new StrictClient();
+                var connection = new Nghttp2Connection(client);
+
+                int served = 0;
+                Task run = connection.RunAsync(_ =>
+                {
+                    served++;
+                    return Nghttp2Response.Text("ok");
+                });
+
+                client.SendRequests();                   // preface and SETTINGS
+                client.Drain();
+
+                // A POST's headers without END_STREAM, so its body is still owed, and then the reset.
+                client.Feed([.. Headers(1, post: true, endStream: false), .. Http2FlowControlTests.RstStream(1, error)]);
+                client.Drain();
+                Assert.Equal(0, Pending(connection));
+                Assert.Equal(0, served);
+
+                client.Feed(Headers(3, post: false, endStream: true));   // and the connection serves on
+                client.Drain();
+                Assert.Equal(1, served);
+
+                client.Close(run);
+                connection.Dispose();
+            });
+        }
+
+        // Its handler is dispatched and parked, with no headers out, so nothing it could send would ever call back.
+        foreach (int staged in new[] { 128 * 1024, 0 })
+        {
+            int bytes = staged;
+            runner.Test($"nghttp2: a streamed response its peer resets with NO_ERROR hears the peer gone at a flush of {bytes} B", () =>
+            {
+                using var client = new StrictClient();
+                var connection = new Nghttp2Connection(client);
+
+                var go = new TaskCompletionSource();
+                var outcome = new TaskCompletionSource<string>();
+                Task run = connection.RunAsync(async (_, writer) =>
+                {
+                    await go.Task;
+                    if (bytes > 0)                       // past the 64 KiB a flush may run ahead of nghttp2
+                    {
+                        writer.GetSpan(bytes)[..bytes].Fill((byte)'x');
+                        writer.Advance(bytes);
+                    }
+                    FlushResult flushed = await writer.FlushAsync();
+                    outcome.SetResult(flushed.IsCompleted ? "peer gone" : "flushed");
+                });
+
+                client.SendRequests();                   // preface and SETTINGS
+                client.Drain();
+                client.Feed(Headers(1, post: false, endStream: true));
+                client.Drain();
+                client.Feed(Http2FlowControlTests.RstStream(1, 0));
+                client.Drain();
+                go.SetResult();
+                client.Drain();
+
+                Assert.True(outcome.Task.IsCompleted, "the flush of a stream its peer reset never returned");
+                Assert.Equal("peer gone", outcome.Task.Result);
+                client.Close(run);
+                connection.Dispose();
+            });
+        }
+    }
+
+    // :method, :scheme http, :path /, and an :authority, which nghttp2 requires.
+    private static byte[] Headers(int streamId, bool post, bool endStream)
+    {
+        byte[] block = [post ? (byte)0x83 : (byte)0x82, 0x86, 0x84, 0x01, 9, .. "a.example"u8.ToArray()];
+        return Http2FlowControlTests.FrameBytes(0x1, endStream ? (byte)0x5 : (byte)0x4, streamId, block);
+    }
+
+    // The requests still being assembled. Nothing outside the connection can see them, so a leak shows nowhere else.
+    private static int Pending(Nghttp2Connection connection)
+        => ((IDictionary)typeof(Nghttp2Connection)
+            .GetField("_pending", BindingFlags.NonPublic | BindingFlags.Instance)!
+            .GetValue(connection)!).Count;
+}

tests/Ioxide.Tests.Unit/Program.cs

@@ -29,6 +29,7 @@ private static int Main()
         Http2StreamedRequestTests.Register(runner);
         Http2TrailerTests.Register(runner);
         Http2PseudoHeaderTests.Register(runner);
+        Nghttp2StreamCloseTests.Register(runner);
         Http2RequestHeaderTests.Register(runner);
         Http2RequestReuseTests.Register(runner);
         Http2StreamedBodyTests.Register(runner);