fix(drivers): stream results with pipe so responses stay under the close-listener limit

stream.pipeline adds seven close listeners to the response; with the ones
metering, compression and tracing add, every streamed /drivers/call passed
Node's ten-listener limit and logged MaxListenersExceededWarning. pipe plus
explicit error and close handling adds two, and keeps the same behavior: a
source failing mid-body aborts the response and logs once, and a client
hanging up stops the source.

src/backend/drivers/util/pipeStreamResult.test.ts

@@ -129,6 +129,27 @@ describe('pipeStreamResult', () => {
         expect(warn).not.toHaveBeenCalled();
     });
 
+    it('adds at most two close listeners to the response', async () => {
+        // The response already carries close listeners from metering,
+        // compression and tracing; Node warns past ten.
+        let added = -1;
+        server = createServer((_req, res) => {
+            const before = res.listenerCount('close');
+            pipeStreamResult(Readable.from(['a', 'b']), res, 'test.run');
+            added = res.listenerCount('close') - before;
+        });
+        await new Promise<void>((resolve) => server!.listen(0, resolve));
+        const { port } = server.address() as AddressInfo;
+        await new Promise<void>((resolve) =>
+            get(`http://127.0.0.1:${port}/`, (res) => {
+                res.resume();
+                res.on('end', () => resolve());
+            }),
+        );
+        expect(added).toBeGreaterThan(0);
+        expect(added).toBeLessThanOrEqual(2);
+    });
+
     it('drops the source, without throwing, when the client left before the result was ready', async () => {
         const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
         const source = Readable.from(['late']);

src/backend/drivers/util/pipeStreamResult.ts

@@ -17,7 +17,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-import { pipeline, type Readable } from 'node:stream';
+import type { Readable } from 'node:stream';
 
 // Only what's used here, so callers on another @types/node (or express's
 // Response) still fit.
@@ -32,24 +32,32 @@ type ResponseSink = NodeJS.WritableStream & {
  * mid-body destroys the response, so the client sees an aborted transfer rather
  * than a clean end, and the failure is logged once instead of surfacing as an
  * unhandled stream error. A client that hangs up stops the source.
+ *
+ * `pipe` rather than `pipeline`: pipeline puts seven close listeners on the
+ * response, and with the ones metering, compression and tracing add, every
+ * streamed response tripped Node's ten-listener warning.
  */
 export function pipeStreamResult(
     stream: Readable,
     res: ResponseSink,
     label: string,
 ): void {
-    // The caller hung up while the driver ran; `pipeline` throws on a closed
-    // destination, so drop the source instead.
+    // The caller hung up while the driver ran: nothing will read the source.
     if (res.closed || res.destroyed) {
         stream.destroy();
         return;
     }
-    pipeline(stream, res, (err) => {
-        if (!err) return;
+    stream.once('error', (err) => {
+        stream.unpipe(res);
         if (!res.destroyed) res.destroy();
-        // The client hanging up is not a failure of ours.
-        if ((err as { code?: string }).code === 'ERR_STREAM_PREMATURE_CLOSE')
-            return;
         console.warn(`[drivers] ${label} stream failed mid-response:`, err);
     });
+    // A response that closes before the source ended is a client hanging up.
+    res.once('close', () => {
+        if (stream.readableEnded) return;
+        stream.unpipe(res);
+        stream.destroy();
+    });
+    res.once('error', () => stream.destroy());
+    stream.pipe(res);
 }