Handle errors and clean up after disconnection in client API

Author: pimterry

src/api/api-model.ts

@@ -199,7 +199,7 @@ export class ApiModel {
     sendRequest(
         requestDefinition: Client.RequestDefinition,
         requestOptions: Client.RequestOptions
-    ): stream.Readable {
+    ) {
         return Client.sendRequest(requestDefinition, requestOptions);
     }
 

src/api/rest-api.ts

@@ -88,18 +88,25 @@ export function exposeRestAPI(
         if (!request) throw new StatusError(400, "No request definition provided");
         if (!options) throw new StatusError(400, "No request options provided");
 
+        const abortController = new AbortController();
+
         const resultStream = apiModel.sendRequest({
             ...request,
             // Body buffers are serialized as base64 (for both requests & responses)
             rawBody: Buffer.from(request.rawBody ?? '', 'base64')
         }, {
-            ...options
+            ...options,
+            abortSignal: abortController.signal
         });
 
+        // If the client closes the connection, abort the outgoing request:
+        res.on('close', () => abortController.abort());
+
         res.writeHead(200, {
             'content-type': 'application/x-ndjson'
         });
 
+        // Write all upstream events as JSON data to the client here:
         resultStream.on('data', (evt: Client.ResponseStreamEvents) => {
             if (evt.type === 'response-body-part') {
                 res.write(JSON.stringify({

src/client/client.ts

@@ -2,7 +2,6 @@ import * as stream from 'stream';
 import * as net from 'net';
 import * as http from 'http';
 import * as https from 'https';
-import { streamToBuffer } from '../util/stream';
 
 export type RawHeaders = Array<[key: string, value: string]>;
 
@@ -23,6 +22,11 @@ export interface RequestDefinition {
 }
 
 export interface RequestOptions {
+    /**
+     * An abort signal, which can be used to cancel the in-process request if
+     * required.
+     */
+    abortSignal?: AbortSignal;
 }
 
 export type ResponseStreamEvents =
@@ -50,6 +54,12 @@ export function sendRequest(
 
     const request = (url.protocol === 'https' ? https : http).request(requestDefn.url, {
         method: requestDefn.method,
+        signal: options.abortSignal
+    });
+
+    options.abortSignal?.addEventListener('abort', () => {
+        // In older Node versions, this seems to be required to _actually_ abort the request:
+        request.abort();
     });
 
     // Node supports sending raw headers via [key, value, key, value] array, but if we do
@@ -93,6 +103,7 @@ export function sendRequest(
         }));
 
         response.on('end', () => resultsStream.push(null));
+        response.on('error', (error) => resultsStream.destroy(error));
     }).catch((error) => {
         resultsStream.destroy(error);
         request.destroy();

test/client/send-request.spec.ts

@@ -3,6 +3,7 @@ import * as mockttp from 'mockttp';
 
 import { ResponseStreamEvents, sendRequest } from '../../src/client/client';
 import { streamToArray } from '../../src/util/stream';
+import { delay } from '../../src/util/promise';
 
 describe("The HTTP client API", () => {
 
@@ -55,5 +56,48 @@ describe("The HTTP client API", () => {
 
         expect(responseEvents[1].type).equal('response-body-part');
         expect(responseEvents[1].rawBody.toString()).to.equal('Mock response body');
+    });
+
+    it("should stop requests if cancelled", async () => {
+        await mockServer.forAnyRequest().thenTimeout()
+
+        const requests: any[] = [];
+        mockServer.on('request', (req) => requests.push(req));
+        const aborts: any[] = [];
+        mockServer.on('abort', (req) => aborts.push(req));
+
+        const abortController = new AbortController();
+
+        const responseStream = sendRequest({
+            url: mockServer.url,
+            method: 'GET',
+            headers: [['host', `localhost:${mockServer.port}`]]
+        }, {
+            abortSignal: abortController.signal
+        });
+
+        const responseEvents: any[] = [];
+        responseStream.on('data', (d) => responseEvents.push(d));
+        responseStream.on('error', (e) => responseEvents.push(e));
+        await delay(10);
+
+        expect(requests.length).to.equal(1);
+        expect(aborts.length).to.equal(0);
+        expect(responseEvents.length).to.equal(0);
+
+        abortController.abort();
+        await delay(10);
+
+        expect(requests.length).to.equal(1);
+        expect(aborts.length).to.equal(1); // <-- Server sees the request cancelled
+
+        // Only emitted event is a thrown error:
+        expect(responseEvents.length).to.equal(1);
+        expect(responseEvents[0]).to.be.instanceOf(Error);
+        expect(responseEvents[0].code).to.be.oneOf([
+            // Depends on the Node version you're testing with:
+            'ECONNRESET',
+            'ABORT_ERR'
+        ]);
     })
 });