feat(events): batch handler invocations to the events dispatcher

Invocations that come due within a few milliseconds go out as one
/invoke-batch request (up to events.invokeBatchSize, default 60). The
dispatcher runs a few at a time and streams a started line and an answer line
per call, so a slow handler delays no other call's answer. A call's timeout
runs from its started line, so time queued behind other calls is never the
handler's; queue time comes out of the connection-wait share instead, and a
call still queued past it is never started and fails as Puter's.

A lone call still uses /invoke, and a dispatcher without the batch route gets
single calls for five minutes before the batch route is tried again.

src/backend/clients/events/EventsWorkerInvokerClient.test.ts

@@ -28,6 +28,7 @@ import type { IConfig } from '../../types.js';
 import {
     DispatcherInvokeTransport,
     EVENTS_DEPLOYED_HEADER,
+    EVENTS_DISPATCH_BATCH_PATH,
     EVENTS_DISPATCH_ERROR_HEADER,
     EVENTS_DISPATCH_PATH,
     EVENTS_ERROR_HEADER,
@@ -36,6 +37,7 @@ import {
     INVOKE_CONNECTIONS,
     SCHEDULE_INVOKE_CONNECTIONS,
     outcomeForStatus,
+    parseAnswerLine,
     type WorkerInvokeRequest,
 } from './EventsWorkerInvokerClient.js';
 
@@ -275,10 +277,12 @@ describe('DispatcherInvokeTransport with every connection busy', () => {
     /** Every connection taken by a call nobody has answered yet. */
     const saturated = async () => {
         const { fetchImpl, held } = heldFetch();
+        // Slots are per call, batched or not; one request per call keeps
+        // the count readable.
         const transport = new DispatcherInvokeTransport(
             'http://dispatcher',
             's',
-            { fetchImpl },
+            { fetchImpl, batchSize: 1 },
         );
         const busy = Array.from({ length: INVOKE_CONNECTIONS }, () =>
             transport.send(CALL),
@@ -365,7 +369,7 @@ describe('DispatcherInvokeTransport schedule lane', () => {
         const transport = new DispatcherInvokeTransport(
             'http://dispatcher',
             's',
-            { fetchImpl },
+            { fetchImpl, batchSize: 1 },
         );
         const scheduled = Array.from(
             { length: SCHEDULE_INVOKE_CONNECTIONS + 5 },
@@ -400,7 +404,7 @@ describe('DispatcherInvokeTransport schedule lane', () => {
         const transport = new DispatcherInvokeTransport(
             'http://dispatcher',
             's',
-            { fetchImpl },
+            { fetchImpl, batchSize: 1 },
         );
         const busy = Array.from({ length: SCHEDULE_INVOKE_CONNECTIONS }, () =>
             transport.send({ ...CALL, lane: 'schedule' as const }),
@@ -424,6 +428,400 @@ describe('DispatcherInvokeTransport schedule lane', () => {
     });
 });
 
+describe('DispatcherInvokeTransport batches', () => {
+    interface BatchRequest {
+        url: string;
+        body: { calls: Array<Record<string, unknown>> };
+        signal: AbortSignal;
+        /** Streams one answer line back. */
+        answer: (line: Record<string, unknown>) => void;
+        end: () => void;
+    }
+
+    /**
+     * A fetch whose batch answers stream only when told to; single calls get a
+     * handled 200 straight away.
+     */
+    const streamingFetch = () => {
+        const batches: BatchRequest[] = [];
+        const singles: Array<{ url: string; headers: Record<string, string> }> =
+            [];
+        const fetchImpl = (async (url: string, init: RequestInit) => {
+            if (!url.endsWith(EVENTS_DISPATCH_BATCH_PATH)) {
+                singles.push({
+                    url,
+                    headers: init.headers as Record<string, string>,
+                });
+                return new Response(null, {
+                    status: 200,
+                    headers: { [EVENTS_HANDLED_HEADER]: '1' },
+                });
+            }
+            let controller!: ReadableStreamDefaultController<Uint8Array>;
+            const stream = new ReadableStream<Uint8Array>({
+                start: (c) => {
+                    controller = c;
+                },
+            });
+            const encoder = new TextEncoder();
+            init.signal?.addEventListener('abort', () => {
+                try {
+                    controller.error(init.signal!.reason);
+                } catch {
+                    // already closed
+                }
+            });
+            batches.push({
+                url,
+                body: JSON.parse(String(init.body)),
+                signal: init.signal!,
+                answer: (line) =>
+                    controller.enqueue(
+                        encoder.encode(`${JSON.stringify(line)}\n`),
+                    ),
+                end: () => controller.close(),
+            });
+            return new Response(stream, {
+                status: 200,
+                headers: { 'content-type': 'application/x-ndjson' },
+            });
+        }) as unknown as FetchImpl;
+        return { fetchImpl, batches, singles };
+    };
+
+    const transportFor = (
+        fetchImpl: FetchImpl,
+        options: { batchSize?: number; batchWindowMs?: number } = {},
+    ) =>
+        new DispatcherInvokeTransport('http://dispatcher', 's', {
+            fetchImpl,
+            ...options,
+        });
+
+    it('sends calls made together as one request, and answers each from its own line', async () => {
+        const { fetchImpl, batches, singles } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        const first = transport.send({ ...CALL, script: 'evw-a', key: 'k1:a' });
+        const second = transport.send({
+            ...CALL,
+            script: 'evw-b',
+            deployed: true,
+            body: '{"b":1}',
+        });
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+
+        expect(batches[0].url).toBe(
+            `http://dispatcher${EVENTS_DISPATCH_BATCH_PATH}`,
+        );
+        expect(batches[0].body.calls).toEqual([
+            {
+                script: 'evw-a',
+                appUid: 'app-test',
+                key: 'k1:a',
+                body: '{}',
+                waitMs: expect.any(Number),
+            },
+            {
+                script: 'evw-b',
+                appUid: 'app-test',
+                key: 'k1:test',
+                deployed: true,
+                body: '{"b":1}',
+                waitMs: expect.any(Number),
+            },
+        ]);
+        // What is left of half the timeout, the share a call may wait.
+        expect(batches[0].body.calls[0].waitMs).toBeLessThanOrEqual(2_500);
+        expect(batches[0].body.calls[0].waitMs).toBeGreaterThan(2_400);
+
+        // Out of order: each line names the call it answers.
+        batches[0].answer({ i: 1, status: 400, handled: true });
+        await expect(second).resolves.toEqual({ status: 400, handled: true });
+        batches[0].answer({
+            i: 0,
+            status: 503,
+            handled: true,
+            error: 'platform-error',
+            retryAfter: '3',
+        });
+        await expect(first).resolves.toEqual({
+            status: 503,
+            handled: true,
+            error: 'platform-error',
+            retryAfterMs: 3000,
+        });
+        // Every call answered, so the request is let go.
+        expect(batches[0].signal.aborted).toBe(true);
+        expect(singles).toEqual([]);
+    });
+
+    it('sends a call with nobody to batch with on the single route', async () => {
+        const { fetchImpl, batches, singles } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        await expect(transport.send(CALL)).resolves.toMatchObject({
+            status: 200,
+        });
+        expect(batches).toEqual([]);
+        expect(singles[0].url).toBe(`http://dispatcher${EVENTS_DISPATCH_PATH}`);
+    });
+
+    it('sends a full batch at once, without waiting out the window', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl, {
+            batchSize: 2,
+            batchWindowMs: 60_000,
+        });
+        const sent = [transport.send(CALL), transport.send(CALL)];
+        const third = transport.send(CALL);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        expect(batches[0].body.calls).toHaveLength(2);
+        batches[0].answer({ i: 0, status: 200, handled: true });
+        batches[0].answer({ i: 1, status: 200, handled: true });
+        await Promise.all(sent);
+        transport.close();
+        await expect(third).resolves.toMatchObject({ status: 200 });
+    });
+
+    it('gives each call its own timeout, without failing the rest of the batch', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        const quick = transport.send({ ...CALL, timeoutMs: 50 });
+        const slow = transport.send({ ...CALL, timeoutMs: 5_000 });
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        batches[0].answer({ i: 0, started: true });
+        batches[0].answer({ i: 1, started: true });
+
+        await expect(quick).resolves.toMatchObject({
+            status: null,
+            timedOut: true,
+        });
+        expect(batches[0].signal.aborted).toBe(false);
+        batches[0].answer({ i: 1, status: 200, handled: true });
+        await expect(slow).resolves.toEqual({ status: 200, handled: true });
+    });
+
+    it('runs a call`s timeout from its started line, not from the send', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        let settled = false;
+        const queued = transport
+            .send({ ...CALL, timeoutMs: 200 })
+            .then((answer) => {
+                settled = true;
+                return answer;
+            });
+        const other = transport.send(CALL);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+
+        // Queued behind other calls for longer than its whole timeout.
+        await new Promise((resolve) => setTimeout(resolve, 250));
+        expect(settled).toBe(false);
+        batches[0].answer({ i: 0, started: true });
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        expect(settled).toBe(false);
+        await expect(queued).resolves.toMatchObject({
+            status: null,
+            timedOut: true,
+        });
+
+        batches[0].answer({ i: 1, status: 200, handled: true });
+        await other;
+    });
+
+    it('puts a call that never started on Puter, not on its handler', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        const never = transport.send({ ...CALL, timeoutMs: 20 });
+        const queueTimedOut = transport.send(CALL);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        batches[0].answer({ i: 1, status: 503, dispatch: 'queue-timeout' });
+
+        await expect(queueTimedOut).resolves.toEqual({
+            status: null,
+            error: 'dispatcher: queue-timeout (503)',
+            dispatchReason: 'queue-timeout',
+        });
+        // Its wait plus the grace for the started line to arrive.
+        await expect(never).resolves.toEqual({
+            status: null,
+            error: 'dispatcher: call never started',
+        });
+        const client = new EventsWorkerInvokerClient({} as IConfig);
+        client.setTransport({ send: async () => queueTimedOut });
+        await expect(client.invoke(REQUEST)).resolves.toMatchObject({
+            outcome: 'unavailable',
+        });
+    });
+
+    it('puts a call the stream ended without answering on Puter', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        const answered = transport.send(CALL);
+        const dropped = transport.send(CALL);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        batches[0].answer({ i: 0, status: 200, handled: true });
+        batches[0].end();
+
+        await expect(answered).resolves.toEqual({ status: 200, handled: true });
+        await expect(dropped).resolves.toEqual({
+            status: null,
+            error: 'dispatcher: batch ended without an answer',
+        });
+    });
+
+    it('carries a dispatch failure in a line as that call`s reason', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        const missing = transport.send(CALL);
+        const other = transport.send(CALL);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        batches[0].answer({ i: 0, status: 404, dispatch: 'missing' });
+        batches[0].answer({ i: 1, status: 200, handled: true });
+
+        await expect(missing).resolves.toEqual({
+            status: null,
+            error: 'dispatcher: missing (404)',
+            dispatchReason: 'missing',
+        });
+        await expect(other).resolves.toMatchObject({ status: 200 });
+    });
+
+    it('fails every call of a batch the dispatcher refused, or something else answered', async () => {
+        for (const [response, expected] of [
+            [
+                new Response('forbidden', {
+                    status: 403,
+                    headers: { [EVENTS_DISPATCH_ERROR_HEADER]: 'forbidden' },
+                }),
+                {
+                    status: null,
+                    error: 'dispatcher: forbidden (403)',
+                    dispatchReason: 'forbidden',
+                },
+            ],
+            [
+                new Response('not found', { status: 404 }),
+                {
+                    status: null,
+                    error: 'dispatcher: unmarked 404 for a batch',
+                },
+            ],
+        ] as const) {
+            const transport = transportFor(
+                (async () => response) as unknown as FetchImpl,
+            );
+            const answers = await Promise.all([
+                transport.send(CALL),
+                transport.send(CALL),
+            ]);
+            expect(answers).toEqual([expected, expected]);
+        }
+    });
+
+    it('fails every call when the batch request itself fails', async () => {
+        const transport = transportFor(refusedFetch);
+        const answers = await Promise.all([
+            transport.send(CALL),
+            transport.send(CALL),
+        ]);
+        expect(answers).toEqual([
+            { status: null, error: 'fetch failed' },
+            { status: null, error: 'fetch failed' },
+        ]);
+    });
+
+    it('falls back to single calls on a dispatcher without the batch route', async () => {
+        const urls: string[] = [];
+        const fetchImpl = (async (url: string) => {
+            urls.push(url);
+            if (url.endsWith(EVENTS_DISPATCH_BATCH_PATH))
+                return new Response('no-such-route', {
+                    status: 404,
+                    headers: {
+                        [EVENTS_DISPATCH_ERROR_HEADER]: 'no-such-route',
+                    },
+                });
+            return new Response(null, {
+                status: 200,
+                headers: { [EVENTS_HANDLED_HEADER]: '1' },
+            });
+        }) as unknown as FetchImpl;
+        const transport = transportFor(fetchImpl);
+
+        const first = await Promise.all([
+            transport.send(CALL),
+            transport.send(CALL),
+        ]);
+        expect(first).toEqual([
+            { status: 200, handled: true },
+            { status: 200, handled: true },
+        ]);
+        // The next pair goes straight to the single route.
+        await Promise.all([transport.send(CALL), transport.send(CALL)]);
+        expect(urls.map((url) => new URL(url).pathname)).toEqual([
+            EVENTS_DISPATCH_BATCH_PATH,
+            EVENTS_DISPATCH_PATH,
+            EVENTS_DISPATCH_PATH,
+            EVENTS_DISPATCH_PATH,
+            EVENTS_DISPATCH_PATH,
+        ]);
+    });
+
+    it('builds each body when the batch goes out', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const transport = transportFor(fetchImpl);
+        let renders = 0;
+        const sent = [
+            transport.send({ ...CALL, render: () => `{"n":${++renders}}` }),
+            transport.send(CALL),
+        ];
+        expect(renders).toBe(0);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        expect(batches[0].body.calls[0].body).toBe('{"n":1}');
+        batches[0].answer({ i: 0, status: 200, handled: true });
+        batches[0].answer({ i: 1, status: 200, handled: true });
+        await Promise.all(sent);
+    });
+
+    it('deploys a call`s missing script and retries it, batched with the rest', async () => {
+        const { fetchImpl, batches } = streamingFetch();
+        const client = new EventsWorkerInvokerClient({
+            events: { invokeTimeoutMs: 5_000 },
+        } as unknown as IConfig);
+        client.setTransport(transportFor(fetchImpl));
+        const missHandler = vi.fn(async () => 'deployed' as const);
+        client.setMissHandler(missHandler);
+
+        const missing = client.invoke({ ...REQUEST, script: 'evw-gone' });
+        const resident = client.invoke(REQUEST);
+        await vi.waitFor(() => expect(batches).toHaveLength(1));
+        batches[0].answer({ i: 0, status: 404, dispatch: 'missing' });
+        batches[0].answer({ i: 1, status: 200, handled: true });
+        await expect(resident).resolves.toMatchObject({ outcome: 'settled' });
+
+        // The retry is alone, so it goes on the single route.
+        await expect(missing).resolves.toMatchObject({ outcome: 'settled' });
+        expect(missHandler).toHaveBeenCalledWith('app-test', 'evw-gone', {
+            lane: undefined,
+        });
+    });
+
+    it('reads only well-formed answer lines', () => {
+        expect(parseAnswerLine('not json')).toBeNull();
+        expect(parseAnswerLine('{"status":200}')).toBeNull();
+        expect(parseAnswerLine('{"i":-1,"status":200}')).toBeNull();
+        expect(parseAnswerLine('{"i":0}')).toBeNull();
+        expect(parseAnswerLine('{"i":1,"started":true}')).toEqual({
+            i: 1,
+            started: true,
+        });
+        expect(parseAnswerLine('{"i":2,"status":200}')).toEqual({
+            i: 2,
+            answer: { status: 200, handled: false },
+        });
+    });
+});
+
 describe('EventsWorkerInvokerClient invoke options', () => {
     it('stamps the event as each attempt goes out', async () => {
         const bodies: string[] = [];

src/backend/clients/events/EventsWorkerInvokerClient.ts

@@ -43,6 +43,20 @@ import { PuterClient } from '../types.js';
  *
  * On a miss, a handler set here may deploy the script itself and retry once
  * with `x-puter-events-deployed: 1`, skipping the dispatcher's own callback.
+ *
+ * Calls that come due within a few milliseconds of each other go out together:
+ *
+ *     POST <events dispatcher>/invoke-batch
+ *     x-puter-internal-auth:  <internal secret>
+ *     { calls: [{ script, appUid, key, deployed?, body, waitMs }] }
+ *
+ * The dispatcher runs a few at a time, in order, and streams NDJSON: `{ i,
+ * started: true }` when a call goes to its worker, then `{ i, status, handled?,
+ * error?, retryAfter?, dispatch? }` when it finishes — the same facts the
+ * single route carries in its status and headers. A call's timeout runs from
+ * its `started` line, so time spent queued behind other calls is never the
+ * handler's. A call still queued `waitMs` after the batch arrived is never
+ * started and answers `dispatch: "queue-timeout"`.
  */
 
 /** The one route an events worker answers. */
@@ -51,6 +65,27 @@ export const EVENTS_INVOKE_PATH = '/__events/invoke';
 /** The route the events dispatcher takes an invocation on. */
 export const EVENTS_DISPATCH_PATH = '/invoke';
 
+/** The route it takes several on, answering each as it finishes. */
+export const EVENTS_DISPATCH_BATCH_PATH = '/invoke-batch';
+
+/** Calls one batch carries at most. */
+export const INVOKE_BATCH_SIZE = 60;
+
+/** How long the first call of a batch waits for others to join it. */
+export const INVOKE_BATCH_WINDOW_MS = 5;
+
+/**
+ * How long a dispatcher without the batch route is sent single calls before the
+ * batch route is tried again.
+ */
+const BATCH_UNSUPPORTED_RETRY_MS = 5 * 60_000;
+
+/**
+ * How long past its wait a batched call's `started` line may take to arrive
+ * before the call is given up as never started.
+ */
+const BATCH_START_GRACE_MS = 2_000;
+
 /**
  * Header the dispatcher marks its own failures with. Anything carrying it never
  * reached a handler, so its status says nothing about the delivery.
@@ -220,7 +255,10 @@ export interface EventsInvokeTransport {
  * handler); retried like `retriable` but never the handler's fault.
  */
 export type WorkerInvokeOutcome =
-    'settled' | 'terminal' | 'retriable' | 'unavailable';
+    | 'settled'
+    | 'terminal'
+    | 'retriable'
+    | 'unavailable';
 
 export interface WorkerInvokeResult {
     outcome: WorkerInvokeOutcome;
@@ -367,9 +405,19 @@ export class EventsWorkerInvokerClient extends PuterClient {
         // a misconfiguration worth reporting once per attempt rather than a
         // round trip.
         if (!dispatcherUrl || !internalSecret) return null;
+        const { invokeBatchSize, invokeBatchWindowMs } =
+            this.config.events ?? {};
         this.#dispatcher ??= new DispatcherInvokeTransport(
             dispatcherUrl,
             internalSecret,
+            {
+                ...(invokeBatchSize !== undefined
+                    ? { batchSize: invokeBatchSize }
+                    : {}),
+                ...(invokeBatchWindowMs !== undefined
+                    ? { batchWindowMs: invokeBatchWindowMs }
+                    : {}),
+            },
         );
         return this.#dispatcher;
     }
@@ -382,11 +430,28 @@ export class EventsWorkerInvokerClient extends PuterClient {
     }
 }
 
+/** A call waiting in, or sent with, the current batch. */
+interface BatchedCall {
+    call: EventsInvokeCall;
+    resolve: (answer: EventsInvokeAnswer) => void;
+    /** When its wait for a connection began, for what is left of it. */
+    waitingSince: number;
+}
+
+export interface DispatcherInvokeTransportOptions {
+    fetchImpl?: typeof undiciFetch;
+    /** Calls per batch; 1 sends every call on its own. */
+    batchSize?: number;
+    batchWindowMs?: number;
+}
+
 /** Reaches an events worker the only way production can: through the dispatcher. */
 export class DispatcherInvokeTransport implements EventsInvokeTransport {
     readonly #url: string;
     readonly #secret: string;
     readonly #fetchImpl: typeof undiciFetch;
+    readonly #batchSize: number;
+    readonly #batchWindowMs: number;
     #agent: Agent | null = null;
     /** Calls holding a connection slot. */
     #active = 0;
@@ -394,18 +459,34 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport {
     #scheduleActive = 0;
     /** Calls waiting for one, longest-waiting first, with their lane. */
     readonly #waiting = new Map<() => void, InvokeLane>();
+    /** Calls gathered for the next batch. */
+    #batch: BatchedCall[] = [];
+    #batchTimer: ReturnType<typeof setTimeout> | null = null;
+    /** Until when the dispatcher is taken not to have the batch route. */
+    #batchUnsupportedUntil = 0;
 
     constructor(
         dispatcherUrl: string,
         secret: string,
-        options: { fetchImpl?: typeof undiciFetch } = {},
+        options: DispatcherInvokeTransportOptions = {},
     ) {
-        this.#url = dispatcherUrl;
+        // Joined as a string: `new URL(path, base)` resolves an absolute path
+        // against the base's origin, dropping any path prefix it carries.
+        this.#url = dispatcherUrl.replace(/\/$/, '');
         this.#secret = secret;
         this.#fetchImpl = options.fetchImpl ?? undiciFetch;
+        this.#batchSize = Math.max(
+            1,
+            Math.floor(options.batchSize ?? INVOKE_BATCH_SIZE),
+        );
+        this.#batchWindowMs = Math.max(
+            0,
+            options.batchWindowMs ?? INVOKE_BATCH_WINDOW_MS,
+        );
     }
 
     close(): void {
+        this.#flush();
         void this.#agent?.close().catch(() => {});
         this.#agent = null;
     }
@@ -414,6 +495,7 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport {
         // No `timedOut`: the handler never got this call, so the wait is not
         // its failure.
         const lane = call.lane ?? 'delivery';
+        const waitingSince = Date.now();
         if (
             !(await this.#acquire(call.timeoutMs * CONNECTION_WAIT_SHARE, lane))
         )
@@ -422,23 +504,58 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport {
                 error: 'no free connection to the events dispatcher',
             };
         try {
-            // `new URL(path, base)` resolves an absolute path against the
-            // base's origin, dropping any path prefix `base` already carries.
-            const url = `${this.#url.replace(/\/$/, '')}${EVENTS_DISPATCH_PATH}`;
-            const response = await this.#fetchImpl(url, {
-                method: 'POST',
-                headers: {
-                    'content-type': 'application/json',
-                    'x-puter-internal-auth': this.#secret,
-                    'x-puter-events-script': call.script,
-                    'x-puter-events-app': call.appUid,
-                    'x-puter-events-key': call.key,
-                    ...(call.deployed ? { [EVENTS_DEPLOYED_HEADER]: '1' } : {}),
-                },
-                body: call.render ? call.render() : call.body,
-                signal: AbortSignal.timeout(call.timeoutMs),
-                dispatcher: this.#dispatcherFor(),
+            if (this.#batchSize === 1) return await this.#sendOne(call);
+            return await new Promise<EventsInvokeAnswer>((resolve) => {
+                this.#batch.push({ call, resolve, waitingSince });
+                if (this.#batch.length >= this.#batchSize) this.#flush();
+                else
+                    this.#batchTimer ??= setTimeout(
+                        () => this.#flush(),
+                        this.#batchWindowMs,
+                    );
             });
+        } finally {
+            this.#release(lane);
+        }
+    }
+
+    /** Send what has gathered: alone when it is one call, else as a batch. */
+    #flush(): void {
+        if (this.#batchTimer) clearTimeout(this.#batchTimer);
+        this.#batchTimer = null;
+        const batch = this.#batch;
+        this.#batch = [];
+        if (batch.length === 0) return;
+        if (batch.length === 1 || Date.now() < this.#batchUnsupportedUntil) {
+            for (const { call, resolve } of batch)
+                void this.#sendOne(call).then(resolve);
+            return;
+        }
+        void this.#sendBatch(batch);
+    }
+
+    /** One call on the single route. */
+    async #sendOne(call: EventsInvokeCall): Promise<EventsInvokeAnswer> {
+        try {
+            const response = await this.#fetchImpl(
+                `${this.#url}${EVENTS_DISPATCH_PATH}`,
+                {
+                    method: 'POST',
+                    headers: {
+                        'content-type': 'application/json',
+                        'x-puter-internal-auth': this.#secret,
+                        'x-puter-events-script': call.script,
+                        'x-puter-events-app': call.appUid,
+                        'x-puter-events-key': call.key,
+                        ...(call.deployed
+                            ? { [EVENTS_DEPLOYED_HEADER]: '1' }
+                            : {}),
+                    },
+                    body: call.render ? call.render() : call.body,
+                    signal: AbortSignal.timeout(call.timeoutMs),
+                    dispatcher: this.#dispatcherFor(),
+                },
+            );
             // The body is not read: a handler answers with a status, and
             // whatever else it writes is its own business.
             void response.body?.cancel().catch(() => {});
@@ -449,17 +566,160 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport {
                 EVENTS_DISPATCH_ERROR_HEADER,
             );
             if (dispatchError !== null)
-                return {
-                    status: null,
-                    error: `dispatcher: ${dispatchError} (${response.status})`,
-                    dispatchReason: dispatchError,
-                };
+                return dispatchAnswer(dispatchError, response.status);
 
             return answerFrom(response);
         } catch (err) {
             return failedAnswer(err);
-        } finally {
-            this.#release(lane);
+        }
+    }
+
+    /**
+     * Several calls in one request. A call may wait in the dispatcher's queue
+     * for whatever is left of its connection wait, and its timeout runs from
+     * its `started` line; the request itself is dropped once every call has an
+     * answer.
+     */
+    async #sendBatch(batch: BatchedCall[]): Promise<void> {
+        const hangUp = new AbortController();
+        let open = batch.length;
+        const answered = batch.map(() => false);
+        const timers: ReturnType<typeof setTimeout>[] = [];
+        const settle = (i: number, answer: EventsInvokeAnswer): void => {
+            if (answered[i]) return;
+            answered[i] = true;
+            clearTimeout(timers[i]);
+            batch[i].resolve(answer);
+            if (--open === 0) hangUp.abort();
+        };
+        const settleRest = (answer: EventsInvokeAnswer): void => {
+            for (let i = 0; i < batch.length; i++) settle(i, answer);
+        };
+
+        const now = Date.now();
+        const waits = batch.map(({ call, waitingSince }) =>
+            Math.max(
+                0,
+                Math.floor(
+                    call.timeoutMs * CONNECTION_WAIT_SHARE -
+                        (now - waitingSince),
+                ),
+            ),
+        );
+        const body = JSON.stringify({
+            calls: batch.map(({ call }, i) => ({
+                script: call.script,
+                appUid: call.appUid,
+                key: call.key,
+                ...(call.deployed ? { deployed: true } : {}),
+                body: call.render ? call.render() : call.body,
+                waitMs: waits[i],
+            })),
+        });
+        const arm = (
+            i: number,
+            ms: number,
+            answer: () => EventsInvokeAnswer,
+        ) => {
+            clearTimeout(timers[i]);
+            timers[i] = setTimeout(() => settle(i, answer()), ms);
+            timers[i].unref?.();
+        };
+        // No `timedOut`: a call that never started never reached a handler.
+        batch.forEach((_, i) =>
+            arm(i, waits[i] + BATCH_START_GRACE_MS, () => ({
+                status: null,
+                error: 'dispatcher: call never started',
+            })),
+        );
+        const started = (i: number): void => {
+            if (!answered[i])
+                arm(i, batch[i].call.timeoutMs, () =>
+                    failedAnswer(timeoutError()),
+                );
+        };
+        const longest = Math.max(
+            ...batch.map(
+                ({ call }, i) =>
+                    waits[i] + BATCH_START_GRACE_MS + call.timeoutMs,
+            ),
+        );
+
+        try {
+            const response = await this.#fetchImpl(
+                `${this.#url}${EVENTS_DISPATCH_BATCH_PATH}`,
+                {
+                    method: 'POST',
+                    headers: {
+                        'content-type': 'application/json',
+                        'x-puter-internal-auth': this.#secret,
+                    },
+                    body,
+                    signal: AbortSignal.any([
+                        hangUp.signal,
+                        AbortSignal.timeout(longest),
+                    ]),
+                    dispatcher: this.#dispatcherFor(),
+                },
+            );
+
+            const dispatchError = response.headers.get(
+                EVENTS_DISPATCH_ERROR_HEADER,
+            );
+            if (dispatchError === 'no-such-route') {
+                // A dispatcher from before the batch route: nothing ran, so
+                // every call goes again on its own.
+                void response.body?.cancel().catch(() => {});
+                this.#batchUnsupportedUntil =
+                    Date.now() + BATCH_UNSUPPORTED_RETRY_MS;
+                batch.forEach(({ call }, i) => {
+                    if (!answered[i])
+                        void this.#sendOne(call).then((a) => settle(i, a));
+                });
+                return;
+            }
+            if (dispatchError !== null || !response.ok) {
+                void response.body?.cancel().catch(() => {});
+                // An unmarked failure (an edge 404, a WAF 403) reads as
+                // Puter's, never as any handler's answer.
+                settleRest(
+                    dispatchError !== null
+                        ? dispatchAnswer(dispatchError, response.status)
+                        : {
+                              status: null,
+                              error: `dispatcher: unmarked ${response.status} for a batch`,
+                          },
+                );
+                return;
+            }
+
+            let buffered = '';
+            const read = (line: string): void => {
+                if (line.trim() === '') return;
+                const parsed = parseAnswerLine(line);
+                if (!parsed || parsed.i >= batch.length) return;
+                if ('started' in parsed) started(parsed.i);
+                else settle(parsed.i, parsed.answer);
+            };
+            const decoder = new TextDecoder();
+            for await (const chunk of response.body ?? []) {
+                buffered += decoder.decode(chunk, { stream: true });
+                let newline = buffered.indexOf('\n');
+                while (newline !== -1) {
+                    read(buffered.slice(0, newline));
+                    buffered = buffered.slice(newline + 1);
+                    newline = buffered.indexOf('\n');
+                }
+            }
+            read(buffered);
+            settleRest({
+                status: null,
+                error: 'dispatcher: batch ended without an answer',
+            });
+        } catch (err) {
+            // Every call has its answer and the request was dropped on purpose.
+            if (open === 0) return;
+            settleRest(failedAnswer(err));
         }
     }
 
@@ -524,6 +784,65 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport {
     }
 }
 
+/** What a failure the dispatcher marked as its own reads as. */
+const dispatchAnswer = (
+    reason: string,
+    status: number,
+): EventsInvokeAnswer => ({
+    status: null,
+    error: `dispatcher: ${reason} (${status})`,
+    dispatchReason: reason,
+});
+
+/** What `AbortSignal.timeout` rejects with, for a batched call's own timeout. */
+const timeoutError = (): Error =>
+    new DOMException(
+        'The operation was aborted due to timeout',
+        'TimeoutError',
+    );
+
+/** One line of a batch answer, or null when it is not one. */
+export const parseAnswerLine = (
+    line: string,
+):
+    | { i: number; started: true }
+    | { i: number; answer: EventsInvokeAnswer }
+    | null => {
+    let raw: unknown;
+    try {
+        raw = JSON.parse(line);
+    } catch {
+        return null;
+    }
+    if (!raw || typeof raw !== 'object') return null;
+    const { i, started, status, handled, error, retryAfter, dispatch } =
+        raw as Record<string, unknown>;
+    if (!Number.isInteger(i) || (i as number) < 0) return null;
+    if (started === true) return { i: i as number, started: true };
+    if (typeof status !== 'number') return null;
+    if (typeof dispatch === 'string')
+        return { i: i as number, answer: dispatchAnswer(dispatch, status) };
+    return {
+        i: i as number,
+        answer: answerFrom({
+            status,
+            headers: {
+                get: (name) => {
+                    if (name === EVENTS_HANDLED_HEADER)
+                        return handled === true ? '1' : null;
+                    if (name === EVENTS_ERROR_HEADER)
+                        return typeof error === 'string' ? error : null;
+                    if (name === 'retry-after')
+                        return typeof retryAfter === 'string'
+                            ? retryAfter
+                            : null;
+                    return null;
+                },
+            },
+        }),
+    };
+};
+
 /** Whether deploying the script is what fixes this answer. */
 const selfDeployable = (result: EventsInvokeAnswer): boolean =>
     result.status === null

src/backend/types.ts

@@ -256,7 +256,12 @@ export interface IPreludeConfig {
      * an RCS agent provisioned in the Prelude account to actually use RCS.
      */
     preferredChannel?:
-        'sms' | 'rcs' | 'whatsapp' | 'viber' | 'zalo' | 'telegram';
+        | 'sms'
+        | 'rcs'
+        | 'whatsapp'
+        | 'viber'
+        | 'zalo'
+        | 'telegram';
 }
 
 /**
@@ -1285,6 +1290,16 @@ interface IConfigOptional {
         forwardSession?: boolean;
         /** How long a handler has to answer an invocation. Default 30 s. */
         invokeTimeoutMs?: number;
+        /**
+         * Invocations sent to the dispatcher in one request. Default 60; 1
+         * sends each on its own.
+         */
+        invokeBatchSize?: number;
+        /**
+         * How long the first invocation of a batch waits for others to join it.
+         * Default 5 ms.
+         */
+        invokeBatchWindowMs?: number;
         /**
          * Dispatch namespace events workers are deployed into. Must not be the
          * one the public worker dispatcher serves — that namespace is reachable
@@ -1364,8 +1379,7 @@ export interface ServerDrainInfo {
 }
 
 export interface WithCostsReporting extends WithLifecycle {
-    getReportedCosts?: () =>
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    getReportedCosts?: () => // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Promise<Record<string, any>[]>
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Record<string, any>[];