From c94902a023ffcb303b88a0c42096da303228cd67 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Mon, 2 Nov 2026 18:23:25 -0700 Subject: [PATCH] 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. --- .../events/EventsWorkerInvokerClient.test.ts | 404 +++++++++++++++++- .../events/EventsWorkerInvokerClient.ts | 371 ++++++++++++++-- src/backend/types.ts | 20 +- 3 files changed, 763 insertions(+), 32 deletions(-) diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.test.ts b/src/backend/clients/events/EventsWorkerInvokerClient.test.ts index 93bc83a70..bfeec2f88 100644 --- a/src/backend/clients/events/EventsWorkerInvokerClient.test.ts +++ b/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> }; + signal: AbortSignal; + /** Streams one answer line back. */ + answer: (line: Record) => 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 }> = + []; + const fetchImpl = (async (url: string, init: RequestInit) => { + if (!url.endsWith(EVENTS_DISPATCH_BATCH_PATH)) { + singles.push({ + url, + headers: init.headers as Record, + }); + return new Response(null, { + status: 200, + headers: { [EVENTS_HANDLED_HEADER]: '1' }, + }); + } + let controller!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + 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[] = []; diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.ts b/src/backend/clients/events/EventsWorkerInvokerClient.ts index 6427a58a7..40bcaf4cf 100644 --- a/src/backend/clients/events/EventsWorkerInvokerClient.ts +++ b/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 /invoke-batch + * x-puter-internal-auth: + * { 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 | 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((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 { + 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 { + const hangUp = new AbortController(); + let open = batch.length; + const answered = batch.map(() => false); + const timers: ReturnType[] = []; + 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; + 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 diff --git a/src/backend/types.ts b/src/backend/types.ts index 6d9a1230a..1089d351d 100644 --- a/src/backend/types.ts +++ b/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[]> // eslint-disable-next-line @typescript-eslint/no-explicit-any | Record[];