fix(plugin-hono-server): keep a graceful close from cutting off a response still flushing to a slow client

.changeset/31208-hono-drain-keeps-flushing-response.md

@@ -0,0 +1,10 @@
+---
+'@objectstack/plugin-hono-server': patch
+---
+
+fix(plugin-hono-server): a graceful `close()` no longer cuts off a response that has ended but is still being sent to a slow client
+
+Clause-②: no
+Metadata-protocol: none
+
+When `HonoHttpServer.close()` started, Node's own `server.close()` released idle keep-alive connections first thing, and counted a response as done as soon as it had called `end()`, before its bytes were sent. A large response still flushing to a slow client at that moment was destroyed mid-body, so the client got a truncated download. The release of idle connections now waits until no ended response is still flushing; each such response closes its own connection once every byte has been sent, and idle connections are released right after. The drain window still force-closes whatever outlasts it.

packages/plugins/plugin-hono-server/src/adapter-drain.test.ts

@@ -185,6 +185,88 @@ describe('HonoHttpServer — graceful close drains in-flight requests (P1-3)', (
     await closeP;
   });
 
+  it('a response that ended before close() but is still flushing to a slow reader arrives whole', async () => {
+    // Node counts a response as done once it has called `end()`, before its
+    // bytes are flushed. `server.close()` sweeps idle connections first thing
+    // (Node 19 and later), so without the guard this response's connection is
+    // destroyed at drain start: 3,919,702 of 33,554,432 bytes arrived on main.
+    // An idle keep-alive connection alongside it must still be released by the
+    // drain, once the flush is done, rather than holding `close()` open until
+    // the server's keep-alive timer fires.
+    const BIG = new Uint8Array(32 * 1024 * 1024).fill(0x61);
+    let bigResponse: ServerResponse | undefined;
+    const server = new HonoHttpServer(0, undefined, 10_000);
+    server.getRawApp().get('/big', (c) => {
+      bigResponse = (c.env as { outgoing: ServerResponse }).outgoing;
+      return c.body(BIG);
+    });
+    server.getRawApp().get('/probe', (c) => c.text('idle-after-this'));
+    await server.listen(0);
+    const port = server.getPort();
+
+    const idle = await rawConnection(port);
+    idle.socket.write('GET /probe HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n');
+    const slowReader = await rawConnection(port);
+    slowReader.socket.pause(); // reads nothing until resumed below
+    slowReader.socket.write('GET /big HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n');
+    await sleep(100);
+    expect(splitResponse(idle.received()).headers.connection).toBe('keep-alive');
+    // The premise: the big response has called end() and is still flushing.
+    expect(bigResponse?.writableEnded, 'the big response has called end()').toBe(true);
+    expect(bigResponse?.writableFinished, 'the big response is still flushing').toBe(false);
+
+    const closeP = server.close();
+    await sleep(50);
+    slowReader.socket.resume();
+    await slowReader.ended;
+    const flushed = Date.now();
+    const { head, headers, body } = splitResponse(slowReader.received());
+    expect(head).toMatch(/^HTTP\/1\.1 200/);
+    expect(Number(headers['content-length'])).toBe(BIG.length);
+    expect(body.length, 'every byte of the body arrived').toBe(BIG.length);
+
+    await idle.ended;
+    await closeP;
+    // The keep-alive timer would hold it 5 s; the release takes milliseconds.
+    expect(Date.now() - flushed).toBeLessThan(1000);
+  });
+
+  it('an idle keep-alive connection is released at drain start', async () => {
+    const server = new HonoHttpServer(0, undefined, 10_000);
+    server.getRawApp().get('/probe', (c) => c.text('idle-after-this'));
+    await server.listen(0);
+    const idle = await rawConnection(server.getPort());
+    idle.socket.write('GET /probe HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n');
+    await sleep(50);
+    expect(splitResponse(idle.received()).headers.connection).toBe('keep-alive');
+
+    const t0 = Date.now();
+    await server.close();
+    await idle.ended;
+    // Nothing is flushing, so the sweep runs at once: no keep-alive wait.
+    expect(Date.now() - t0).toBeLessThan(1000);
+  });
+
+  it('the drain window still force-closes a reader that never drains', async () => {
+    const BIG = new Uint8Array(32 * 1024 * 1024).fill(0x61);
+    const server = new HonoHttpServer(0, undefined, 300);
+    server.getRawApp().get('/big', (c) => c.body(BIG));
+    await server.listen(0);
+    const stalled = await rawConnection(server.getPort());
+    stalled.socket.pause(); // never resumed
+    stalled.socket.write('GET /big HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n');
+    await sleep(100);
+
+    const t0 = Date.now();
+    await server.close();
+    const elapsed = Date.now() - t0;
+    expect(elapsed).toBeGreaterThanOrEqual(250); // held for the window, not swept at drain start
+    expect(elapsed).toBeLessThan(2000); // the 300 ms window, not the reader
+    stalled.socket.resume(); // read what made it out, to see the connection end
+    await stalled.ended; // the backstop destroyed it
+    expect(stalled.received().length).toBeLessThan(BIG.length);
+  });
+
   it('forgets a pipelined response whose connection died before it was written', async () => {
     // Pipelined requests queue behind the first one's response; the queued
     // responses have no socket of their own, so if the connection dies they

packages/plugins/plugin-hono-server/src/adapter.ts

@@ -595,6 +595,12 @@ export class HonoHttpServer implements IHttpServer {
      * another on a connection that died) is not held past its connection.
      */
     private inFlightResponses = new Map<Socket, OwedResponses>();
+    /**
+     * Node's idle-connection sweep, held back because a response was still
+     * flushing when it was asked for. {@link trackForDrain} runs it once none
+     * is — see {@link closeIdleConnectionsOnceFlushed}.
+     */
+    private deferredIdleSweep: (() => void) | undefined;
     /**
      * Every `(method, pattern)` pair registered through this server, kept so
      * the `notFound` handler can answer "the path exists but the method is
@@ -1849,6 +1855,12 @@ export class HonoHttpServer implements IHttpServer {
                     resolve();
                 });
             this.server = server;
+            // On the instance, so `server.close()` reaches it too: Node's
+            // `httpServerPreClose` calls `server.closeIdleConnections()`.
+            if ('closeIdleConnections' in server) {
+                const sweepIdle: () => void = server.closeIdleConnections.bind(server);
+                server.closeIdleConnections = () => this.closeIdleConnectionsOnceFlushed(sweepIdle);
+            }
             // Prepended so it runs before the Hono listener can write a
             // synchronous response's headers.
             server.prependListener('request', (req: IncomingMessage, res: ServerResponse) => {
@@ -1880,10 +1892,13 @@ export class HonoHttpServer implements IHttpServer {
      *   handed to the socket — and no other response on that connection is
      *   pending. `destroySoon()` flushes what is queued before it closes.
      *
-     * ⛔ Never the server-wide `closeIdleConnections()` from here: Node counts
-     * a response as done once it has called `end()`, before its bytes are
-     * flushed, so a sweep when one response ends destroys any other connection
-     * whose response is ended but still flushing to a slow reader.
+     * ⛔ Never Node's own server-wide sweep from here: Node counts a response
+     * as done once it has called `end()`, before its bytes are flushed, so a
+     * sweep when one response ends destroys any other connection whose
+     * response is ended but still flushing to a slow reader. The only sweep
+     * run here is one `close()` asked for and
+     * {@link closeIdleConnectionsOnceFlushed} held back, and it still runs only
+     * once no response is flushing.
      *
      * A connection's entry is dropped when the connection closes, not only
      * when each response does: a response pipelined behind another never gets
@@ -1898,12 +1913,42 @@ export class HonoHttpServer implements IHttpServer {
         owed.add(res);
         res.once('close', () => {
             owed.delete(res);
-            if (!this.draining || owed.size > 0 || !res.writableFinished) return;
-            if (socket.destroyed || socket.writableEnded) return;
-            socket.destroySoon();
+            if (this.draining && owed.size === 0 && res.writableFinished
+                && !socket.destroyed && !socket.writableEnded) {
+                socket.destroySoon();
+            }
+            const sweep = this.deferredIdleSweep;
+            if (sweep) this.closeIdleConnectionsOnceFlushed(sweep);
         });
     }
 
+    /**
+     * Node's `closeIdleConnections()`, but never while a response is ended and
+     * still flushing.
+     *
+     * Node counts a connection as idle once its request has been read, and
+     * skips it only while its response has not called `end()`. `end()` comes
+     * before the bytes are flushed, so a sweep destroys a connection that is
+     * still sending an ended response to a slow reader, and the client gets a
+     * truncated body. `server.close()` runs that sweep first thing (Node 19 and
+     * later), so the drain's force-close backstop never gets a say. Such a
+     * connection closes on its own once flushed ({@link trackForDrain}); the
+     * sweep is held back until no response is flushing, then run for the idle
+     * connections it is for.
+     */
+    private closeIdleConnectionsOnceFlushed(sweep: () => void) {
+        for (const owed of this.inFlightResponses.values()) {
+            for (const res of owed) {
+                if (res.writableEnded && !res.writableFinished) {
+                    this.deferredIdleSweep = sweep;
+                    return;
+                }
+            }
+        }
+        this.deferredIdleSweep = undefined;
+        sweep();
+    }
+
     /** The {@link inFlightResponses} entry for one connection, made on its first request and dropped when it closes. */
     private responsesOwedOn(socket: Socket): OwedResponses {
         const known: OwedResponses | undefined = this.inFlightResponses.get(socket);
@@ -1931,10 +1976,12 @@ export class HonoHttpServer implements IHttpServer {
         // Graceful drain (P1-3): stop accepting new connections and let in-flight
         // requests finish rather than force-killing them mid-response.
         // `closeIdleConnections()` releases idle keep-alive sockets so the process
-        // can exit promptly; active requests keep running until they complete or
-        // the drain window elapses, and each one's connection closes as soon as
-        // its response ends (`trackForDrain`), so the drain finishes with the
-        // last in-flight request rather than a keep-alive window after it.
+        // can exit promptly, once no ended response is still flushing
+        // (`closeIdleConnectionsOnceFlushed`); active requests keep running
+        // until they complete or the drain window elapses, and each one's
+        // connection closes as soon as its response ends (`trackForDrain`), so
+        // the drain finishes with the last in-flight request rather than a
+        // keep-alive window after it.
         this.draining = true;
         for (const owed of this.inFlightResponses.values()) {
             for (const res of owed) {
@@ -1963,6 +2010,7 @@ export class HonoHttpServer implements IHttpServer {
         });
         this.server = undefined;
         this.draining = false;
+        this.deferredIdleSweep = undefined;
         this.inFlightResponses.clear();
     }
 }