Enforce that rules must end or destroy before they resolve successfully

The one step breaking this invariant was the streaming step, which
resolved immediately. This meant it ignored any possible errors. With
this fix, stream errors will now be exposed in the response stream as
expected.

This also changes logic so that in general any handler throwing an
AbortError will have the underlying response automatically killed
if it hasn't destroyed it itself already. This shouldn't affect any
externally visible behaviour.

Author: pimterry

src/rules/requests/request-step-impls.ts

@@ -285,17 +285,33 @@ export class CallbackStepImpl extends CallbackStep {
 
 export class StreamStepImpl extends StreamStep {
 
-    async handle(_request: OngoingRequest, response: OngoingResponse) {
+    async handle(request: OngoingRequest, response: OngoingResponse) {
         if (!this.stream.done) {
             if (this.headers) dropDefaultHeaders(response);
 
             writeHead(response, this.status, undefined, this.headers);
             response.flushHeaders();
 
-            this.stream.pipe(response);
+            if (this.stream.readableEnded || this.stream.destroyed) {
+                response.end();
+            } else {
+                this.stream.pipe(response);
+            }
             this.stream.done = true;
 
-            this.stream.on('error', (e) => response.destroy(e));
+            return new Promise<void>((resolve, reject) => {
+                if (this.stream.readableEnded || this.stream.destroyed) {
+                    resolve();
+                } else {
+                    this.stream.on('end', () => resolve());
+                    this.stream.on('error', (e: ErrorLike) => {
+                        reject(new AbortError(
+                            `Stream rule error: ${e.message}`,
+                            e.code ?? 'STREAM_RULE_ERROR'
+                        ));
+                    });
+                }
+            });
         } else {
             throw new Error(stripIndent`
                 Stream request step called more than once - this is not supported.

src/server/mockttp-server.ts

@@ -745,10 +745,16 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
             } else {
                 await this.sendUnmatchedRequestError(request, response);
             }
-            result = result || 'responded';
+
+            if (!response.writableEnded && !response.destroyed) {
+                throw new Error("Request handler finished successfully without ending the response");
+            }
+
+            result ||= 'responded';
         } catch (e) {
             if (e instanceof AbortError) {
                 abort(e);
+                response.destroy(e);
 
                 if (this.debug) {
                     console.error("Failed to handle request due to abort:", e);
@@ -770,7 +776,7 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
 
                 try {
                     response.end((isErrorLike(e) && e.toString()) || e);
-                    result = result || 'responded';
+                    result ||= 'responded';
                 } catch (e) {
                     abort(e as Error);
                 }

test/integration/subscriptions/response-events.spec.ts

@@ -59,10 +59,8 @@ describe("Response initiated subscriptions", () => {
 
             const timingEvents = seenResponse.timingEvents;
             expect(timingEvents.startTimestamp).to.be.a('number');
-            expect(timingEvents.bodyReceivedTimestamp).to.be.a('number');
             expect(timingEvents.headersSentTimestamp).to.be.a('number');
 
-            expect(timingEvents.bodyReceivedTimestamp).to.be.greaterThan(timingEvents.startTimestamp);
             expect(timingEvents.headersSentTimestamp).to.be.greaterThan(timingEvents.startTimestamp);
 
             expect(timingEvents.responseSentTimestamp).to.equal(undefined);