nghttp2: a response given after the peer reset its stream attaches nothing, so its body no longer leaks and a streamed writer hears the peer gone

src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.Streamed.cs

@@ -110,11 +110,12 @@ private Nghttp2ResponseWriter RentWriter(int streamId)
     }
 
     /// <summary>Headers of a streamed response - no END_STREAM, the body follows.</summary>
-    internal unsafe void SendStreamedHeaders(int streamId, Nghttp2Response response)
+    /// <returns>False when nghttp2 no longer knows the stream: the peer reset it.</returns>
+    internal unsafe bool SendStreamedHeaders(int streamId, Nghttp2Response response)
     {
         if (_handle == 0)
         {
-            return;
+            return true;
         }
 
         byte[] headers = PackHeaders(response, out int headersLength);
@@ -126,10 +127,15 @@ internal unsafe void SendStreamedHeaders(int streamId, Nghttp2Response response)
                 result = Nghttp2.ih2_submit_response_stream(_handle, streamId, headerBytes, (nuint)headersLength);
             }
 
+            if (result == ErrStreamClosed)
+            {
+                return false;
+            }
             if (result != 0)
             {
                 _failed = true;
             }
+            return true;
         }
         finally
         {

src/protocols/ioxide.nghttp2/Connection/Nghttp2Connection.cs

@@ -45,6 +45,7 @@ public sealed partial class Nghttp2Connection : IDisposable
     // RFC 9113 ENHANCE_YOUR_CALM: a request body past MaxRequestBytes, as the managed stack resets it.
     private const uint EnhanceYourCalm = 0xb;
     private const int  ErrInvalidArgument = -501;   // NGHTTP2_ERR_INVALID_ARGUMENT
+    private const int  ErrStreamClosed = -510;      // NGHTTP2_ERR_STREAM_CLOSED
 
     private readonly IDuplexPipe _pipe;
     private readonly Nghttp2Options _options;

src/protocols/ioxide.nghttp2/Http/Nghttp2ResponseWriter.cs

@@ -80,9 +80,9 @@ public void WriteHeaders(Nghttp2Response response)
 
     private void SendHeaders(Nghttp2Response response)
     {
-        if (!_gone)
+        if (!_gone && !_connection.SendStreamedHeaders(StreamId, response))
         {
-            _connection.SendStreamedHeaders(StreamId, response);
+            _gone = true;
         }
     }
 

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

@@ -43,17 +43,19 @@ internal struct Callbacks
 
     /// <summary>Answer the request on <paramref name="streamId"/>. Headers are packed
     /// [u16 namelen][name][u16 valuelen][value]..., <c>:status</c> first. The body is copied
-    /// natively and freed on stream close. Returns 0, or a negative nghttp2 error.</summary>
+    /// natively and freed on stream close; for a stream the peer already reset, nothing is kept and
+    /// the result is still 0. Returns 0, or a negative nghttp2 error.</summary>
     [DllImport(Lib)] internal static extern int ih2_submit_response(nint connection, int streamId,
         byte* headers, nuint headersLen, byte* body, nuint bodyLen);
 
     /// <summary>
-    /// Open a response whose body arrives over time. Headers go out at once; the body follows
-    /// through <see cref="ih2_stream_write"/> until <see cref="ih2_stream_close"/> ends it.
+    /// Open a response whose body arrives over time. Headers are queued for the next drain; the body
+    /// follows through <see cref="ih2_stream_write"/> until <see cref="ih2_stream_close"/> ends it.
     ///
     /// nghttp2 PULLS: it asks for body bytes when it is ready to frame them, and its read callback
     /// defers whenever nothing is buffered - which is what holds the stream open through a slow
-    /// producer rather than ending it early. Returns 0, or a negative nghttp2 error.
+    /// producer rather than ending it early. Returns 0, NGHTTP2_ERR_STREAM_CLOSED for a stream the peer
+    /// already reset, or another negative nghttp2 error.
     /// </summary>
     [DllImport(Lib)] internal static extern int ih2_submit_response_stream(nint connection,
         int streamId, byte* headers, nuint headersLen);

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

@@ -395,6 +395,9 @@ int ih2_submit_response_stream(ih2_conn *c, int32_t stream_id,
     if (c == NULL || c->session == NULL) {
         return NGHTTP2_ERR_INVALID_STATE;
     }
+    if (nghttp2_session_find_stream(c->session, stream_id) == NULL) {
+        return NGHTTP2_ERR_STREAM_CLOSED;   /* the peer reset it: no callback would ever free a stream attached now */
+    }
 
     nghttp2_nv nv[IH2_MAX_HEADERS];
     int count = ih2_unpack_headers(headers, headers_len, nv);
@@ -516,6 +519,9 @@ int ih2_submit_response(ih2_conn *c, int32_t stream_id,
     if (c == NULL || c->session == NULL) {
         return NGHTTP2_ERR_INVALID_STATE;
     }
+    if (nghttp2_session_find_stream(c->session, stream_id) == NULL) {
+        return 0;   /* the peer reset it: no callback would ever free a body attached now, and an error fails the connection */
+    }
 
     nghttp2_nv nv[IH2_MAX_HEADERS];
     int count = ih2_unpack_headers(headers, headers_len, nv);

tests/Ioxide.Tests.Unit/Nghttp2ResetResponseTests.cs

@@ -0,0 +1,123 @@
+using System.IO.Pipelines;
+using System.Runtime.InteropServices;
+using ioxide.nghttp2;
+
+namespace Ioxide.Tests;
+
+/// <summary>A response given after the peer reset its stream leaves nothing behind in the shim.</summary>
+internal static class Nghttp2ResetResponseTests
+{
+    public static void Register(Runner runner)
+    {
+        runner.Test("nghttp2: a buffered response to a stream its peer reset while the handler ran is not kept", () =>
+        {
+            using var client = new StrictClient();
+            var connection = new Nghttp2Connection(client);
+
+            TaskCompletionSource<Nghttp2Response>? parked = null;
+            Task run = connection.RunAsync(_ =>
+            {
+                parked = new TaskCompletionSource<Nghttp2Response>();
+                return new ValueTask<Nghttp2Response>(parked.Task);
+            });
+            var response = new Nghttp2Response { Body = new byte[64 * 1024] };
+
+            client.SendRequests();                       // preface and SETTINGS
+            client.Drain();
+
+            const int warm = 20, total = 220;
+            long before = 0;
+            for (int i = 0; i < total; i++)
+            {
+                int id = 1 + (2 * i);
+                client.Feed(Get(id));
+                client.Drain();
+                client.Feed(Http2FlowControlTests.RstStream(id, 8));   // CANCEL, while the handler is parked
+                client.Drain();
+                parked!.SetResult(response);
+                client.Drain();
+                if (i == warm - 1)
+                {
+                    before = NativeHeapInUse();
+                }
+            }
+            long grown = NativeHeapInUse() - before;
+
+            client.Close(run);
+            connection.Dispose();
+            Assert.True(grown < 6 * 1024 * 1024,         // malloc is the whole process's; the leak is 12.8 MiB
+                $"{total - warm} responses of 64 KiB to reset streams left {grown / 1024} KiB on the native heap");
+        });
+
+        foreach (int staged in new[] { 128 * 1024, 0 })
+        {
+            int bytes = staged;
+            runner.Test($"nghttp2: a streamed response to a stream reset in the same read as its end 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>();
+                bool servedNext = false;
+                Task run = connection.RunAsync(async (request, writer) =>
+                {
+                    if (request.StreamId != 1)
+                    {
+                        servedNext = true;
+                        return;
+                    }
+
+                    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();
+                client.Drain();
+                client.Feed([.. Get(1), .. Http2FlowControlTests.RstStream(1, 8)]);   // its writer is made after the reset
+                client.Drain();
+                go.SetResult();
+                client.Drain();
+
+                Assert.True(outcome.Task.IsCompleted, "the flush of a stream nghttp2 had already closed never returned");
+                Assert.Equal("peer gone", outcome.Task.Result);
+
+                Assert.True(!connection.IsBroken, "one stream's reset failed the connection");   // the loop only sees that after its next read
+                client.Feed(Get(3));
+                client.Drain();
+                Assert.True(servedNext, "the connection served nothing after the reset");
+                client.Close(run);
+                connection.Dispose();
+            });
+        }
+    }
+
+    // A GET with END_STREAM: :scheme http, :path /, and an :authority, which nghttp2 requires.
+    private static byte[] Get(int streamId)
+    {
+        byte[] block = [0x82, 0x86, 0x84, 0x01, 9, .. "a.example"u8.ToArray()];
+        return Http2FlowControlTests.FrameBytes(0x1, 0x5, streamId, block);
+    }
+
+    [StructLayout(LayoutKind.Sequential)]
+    private struct MallInfo2
+    {
+        public nuint Arena, Ordblks, Smblks, Hblks, Hblkhd, Usmblks, Fsmblks, Uordblks, Fordblks, Keepcost;
+    }
+
+    [DllImport("libc", EntryPoint = "mallinfo2")]
+    private static extern MallInfo2 GetMallInfo2();
+
+    // What malloc has handed out and not had back, which is where the shim keeps a response's body.
+    private static long NativeHeapInUse()
+    {
+        MallInfo2 info = GetMallInfo2();
+        return (long)info.Uordblks + (long)info.Hblkhd;
+    }
+}

tests/Ioxide.Tests.Unit/Program.cs

@@ -32,6 +32,7 @@ private static int Main()
         Http2StreamedBodyTests.Register(runner);
         Http2FlowControlTests.Register(runner);
         Http2BufferedCreditTests.Register(runner);
+        Nghttp2ResetResponseTests.Register(runner);
         HpackEncodeTests.Register(runner);
         QpackStaticEncodeTests.Register(runner);
         AlpnSelectTests.Register(runner);