feat(server): add cooperative request cancellation and deadlines

.changeset/request-cancellation.md

@@ -0,0 +1,5 @@
+---
+"@cleverbrush/server": minor
+---
+
+Add withRequestSignal for cooperative request-disconnect, caller and deadline cancellation with deterministic lifecycle cleanup.

libs/server/README.md

@@ -1014,3 +1014,27 @@ return `undefined` when matching. One trailing slash is accepted.
 ## License
 
 BSD-3-Clause — see [LICENSE](../../LICENSE).
+
+## Request cancellation and deadlines
+
+Use `withRequestSignal(context, work, { timeoutMs, signal? })` for bounded,
+request-owned work. The timeout is an application policy, so it is required and
+must be a positive integer no greater than 2,147,483,647 milliseconds.
+
+```ts
+import { withRequestSignal } from '@cleverbrush/server';
+
+const result = await withRequestSignal(
+    context,
+    signal => reports.generate({ signal }),
+    { timeoutMs: 30_000, signal: shutdownSignal }
+);
+```
+
+The signal aborts on an unfinished response closing, request abort, deadline, or
+optional caller cancellation. A completed response closing normally does not
+abort it. Caller cancellation preserves its reason; disconnect and deadline use
+`AbortError` and `TimeoutError` DOMExceptions. Already-cancelled work never starts.
+Timers and listeners are released on success or failure. Cancellation is
+cooperative: the work must observe or forward its signal. This helper neither
+races arbitrary work nor changes HTTP error/status policies.

libs/server/src/index.ts

@@ -142,3 +142,9 @@ export {
     type UploadSchema
 } from './upload.js';
 export { defineWebhook, type WebhookDefinition } from './Webhook.js';
+
+export {
+    type RequestSignalContext,
+    type RequestSignalOptions,
+    withRequestSignal
+} from './withRequestSignal.js';

libs/server/src/withRequestSignal.test.ts

@@ -0,0 +1,133 @@
+import { IncomingMessage, ServerResponse } from 'node:http';
+import { Socket } from 'node:net';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { withRequestSignal } from './withRequestSignal.js';
+
+const fixture = () => {
+    const request = new IncomingMessage(new Socket());
+    return { request, response: new ServerResponse(request) };
+};
+const untilAbort = (signal: AbortSignal) =>
+    new Promise<never>((_, reject) => {
+        signal.addEventListener('abort', () => reject(signal.reason), {
+            once: true
+        });
+    });
+const options = { timeoutMs: 100 };
+afterEach(() => vi.useRealTimers());
+
+describe('withRequestSignal', () => {
+    it('returns the result and removes lifecycle listeners and timer', async () => {
+        vi.useFakeTimers();
+        const context = fixture();
+        expect(await withRequestSignal(context, () => 42, options)).toBe(42);
+        expect(context.request.listenerCount('aborted')).toBe(0);
+        expect(context.response.listenerCount('close')).toBe(0);
+        expect(vi.getTimerCount()).toBe(0);
+    });
+    it('cleans up synchronous and asynchronous failures', async () => {
+        const context = fixture();
+        const failure = new Error('work failed');
+        for (const work of [
+            () => {
+                throw failure;
+            },
+            async () => {
+                throw failure;
+            }
+        ]) {
+            await expect(
+                withRequestSignal(context, work, options)
+            ).rejects.toBe(failure);
+            expect(context.request.listenerCount('aborted')).toBe(0);
+            expect(context.response.listenerCount('close')).toBe(0);
+        }
+    });
+    it('cancels on an aborted request', async () => {
+        const context = fixture();
+        const work = withRequestSignal(context, untilAbort, options);
+        context.request.emit('aborted');
+        await expect(work).rejects.toMatchObject({ name: 'AbortError' });
+    });
+    it('cancels on an unfinished response close', async () => {
+        const context = fixture();
+        const work = withRequestSignal(context, untilAbort, options);
+        context.response.emit('close');
+        await expect(work).rejects.toMatchObject({ name: 'AbortError' });
+    });
+    it('does not cancel on a normally finished response', async () => {
+        const context = fixture();
+        Object.defineProperty(context.response, 'writableFinished', {
+            value: true
+        });
+        await withRequestSignal(
+            context,
+            signal => {
+                context.response.emit('close');
+                expect(signal.aborted).toBe(false);
+            },
+            options
+        );
+    });
+    it('cancels at the deadline and removes the timer', async () => {
+        vi.useFakeTimers();
+        const work = withRequestSignal(fixture(), untilAbort, options);
+        const assertion = expect(work).rejects.toMatchObject({
+            name: 'TimeoutError'
+        });
+        await vi.advanceTimersByTimeAsync(100);
+        await assertion;
+        expect(vi.getTimerCount()).toBe(0);
+    });
+    it('propagates caller cancellation and removes its listener', async () => {
+        const context = fixture();
+        const caller = new AbortController();
+        const remove = vi.spyOn(caller.signal, 'removeEventListener');
+        const reason = new Error('shutdown');
+        const work = withRequestSignal(context, untilAbort, {
+            ...options,
+            signal: caller.signal
+        });
+        caller.abort(reason);
+        await expect(work).rejects.toBe(reason);
+        expect(remove).toHaveBeenCalledWith('abort', expect.any(Function));
+    });
+    it.each(['request', 'response', 'caller'] as const)(
+        'does not start already-cancelled %s work',
+        async kind => {
+            const context = fixture();
+            const caller = new AbortController();
+            if (kind === 'request') context.request.aborted = true;
+            if (kind === 'response') context.response.destroyed = true;
+            if (kind === 'caller') caller.abort();
+            const work = vi.fn();
+            await expect(
+                withRequestSignal(context, work, {
+                    ...options,
+                    signal: caller.signal
+                })
+            ).rejects.toMatchObject({ name: 'AbortError' });
+            expect(work).not.toHaveBeenCalled();
+        }
+    );
+    it('accepts a finished and destroyed response without a false disconnect', async () => {
+        const context = fixture();
+        context.response.destroyed = true;
+        Object.defineProperty(context.response, 'writableFinished', {
+            value: true
+        });
+        expect(await withRequestSignal(context, () => 'done', options)).toBe(
+            'done'
+        );
+    });
+    it.each([0, -1, NaN, Infinity, 1.5, 2_147_483_648])(
+        'rejects invalid deadline %s',
+        async timeoutMs => {
+            const context = fixture();
+            await expect(
+                withRequestSignal(context, () => {}, { timeoutMs })
+            ).rejects.toBeInstanceOf(RangeError);
+            expect(context.response.listenerCount('close')).toBe(0);
+        }
+    );
+});

libs/server/src/withRequestSignal.ts

@@ -0,0 +1,80 @@
+import type { IncomingMessage, ServerResponse } from 'node:http';
+
+/** HTTP lifecycle fields required to cancel request-owned work. */
+export interface RequestSignalContext {
+    readonly request: Pick<IncomingMessage, 'aborted' | 'once' | 'off'>;
+    readonly response: Pick<
+        ServerResponse,
+        'destroyed' | 'writableFinished' | 'once' | 'off'
+    >;
+}
+
+/** Application-owned deadline and optional cancellation for request work. */
+export interface RequestSignalOptions {
+    /** Positive integer deadline in milliseconds, at most 2,147,483,647. */
+    timeoutMs: number;
+    /** Additional cancellation, such as application shutdown. */
+    signal?: AbortSignal;
+}
+
+/**
+ * Run cooperative work with cancellation on disconnect, deadline or caller abort.
+ *
+ * The visitor must observe its signal; this helper does not forcibly interrupt
+ * arbitrary promises. A normal response finish does not cancel the work. The
+ * deadline timer and all listeners are removed when work settles, including on
+ * failure. Already-aborted inputs reject without invoking the visitor.
+ *
+ * @example
+ * ```ts
+ * await withRequestSignal(context, signal => loadReport(signal), {
+ *     timeoutMs: 30_000
+ * });
+ * ```
+ */
+export async function withRequestSignal<T>(
+    context: RequestSignalContext,
+    work: (signal: AbortSignal) => T | PromiseLike<T>,
+    options: RequestSignalOptions
+): Promise<T> {
+    if (
+        !Number.isSafeInteger(options.timeoutMs) ||
+        options.timeoutMs <= 0 ||
+        options.timeoutMs > 2_147_483_647
+    ) {
+        throw new RangeError(
+            'timeoutMs must be a positive timer-safe integer.'
+        );
+    }
+    const controller = new AbortController();
+    const disconnected = () =>
+        controller.abort(
+            new DOMException('Request disconnected.', 'AbortError')
+        );
+    const closed = () => {
+        if (!context.response.writableFinished) disconnected();
+    };
+    const cancelled = () => controller.abort(options.signal!.reason);
+    const timer = setTimeout(
+        () =>
+            controller.abort(
+                new DOMException('Request deadline exceeded.', 'TimeoutError')
+            ),
+        options.timeoutMs
+    );
+    context.request.once('aborted', disconnected);
+    context.response.once('close', closed);
+    options.signal?.addEventListener('abort', cancelled, { once: true });
+    try {
+        if (context.request.aborted || context.response.destroyed) closed();
+        if (context.request.aborted) disconnected();
+        if (options.signal?.aborted) cancelled();
+        controller.signal.throwIfAborted();
+        return await work(controller.signal);
+    } finally {
+        clearTimeout(timer);
+        context.request.off('aborted', disconnected);
+        context.response.off('close', closed);
+        options.signal?.removeEventListener('abort', cancelled);
+    }
+}