From da65b7f569a8780d0da44ec260827a6fb65d011d Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Sat, 5 Sep 2026 14:40:44 -0700 Subject: [PATCH] feat: the invoking backend deploys an events worker the dispatcher cannot find (#3807) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatcher's rehydrate callback reaches whichever backend answers the API's public hostname. A backend with the runtime flag on that is not behind that hostname, or the only one in the fleet with it on, could never get its scripts deployed that way — the callback answered "disabled". The invoking backend already knows the app and script, so on a dispatcher miss it deploys the set itself and retries once, telling the dispatcher to skip its callback and negative cache. The callback stays the path for evicted scripts. --- .../events/EventsWorkerInvokerClient.test.ts | 174 ++++++++++++++++++ .../events/EventsWorkerInvokerClient.ts | 72 +++++++- .../controllers/events/EventsController.ts | 16 +- src/backend/services/events/EventsService.ts | 13 +- .../events/workerInvoker.integration.test.ts | 46 ++++- 5 files changed, 314 insertions(+), 7 deletions(-) diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.test.ts b/src/backend/clients/events/EventsWorkerInvokerClient.test.ts index 60a6d169b..d7efb12fc 100644 --- a/src/backend/clients/events/EventsWorkerInvokerClient.test.ts +++ b/src/backend/clients/events/EventsWorkerInvokerClient.test.ts @@ -24,12 +24,16 @@ import type { fetch as undiciFetch } from 'undici'; import { describe, expect, it } from 'vitest'; +import type { IConfig } from '../../types.js'; import { DispatcherInvokeTransport, + EVENTS_DEPLOYED_HEADER, EVENTS_DISPATCH_ERROR_HEADER, EVENTS_DISPATCH_PATH, EVENTS_ERROR_HEADER, EVENTS_HANDLED_HEADER, + EventsWorkerInvokerClient, + type WorkerInvokeRequest, } from './EventsWorkerInvokerClient.js'; type FetchImpl = typeof undiciFetch; @@ -96,9 +100,27 @@ describe('DispatcherInvokeTransport', () => { expect(await transport.send(CALL)).toEqual({ status: null, error: 'dispatcher: deploy-failed (502)', + dispatchReason: 'deploy-failed', }); }); + it('adds the deployed header when the call says so', async () => { + const headers: Array> = []; + const fetchImpl = (async (_url: string, init?: RequestInit) => { + headers.push(init?.headers as Record); + return new Response(null, { status: 200 }); + }) as unknown as FetchImpl; + const transport = new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl, + }); + + await transport.send(CALL); + await transport.send({ ...CALL, deployed: true }); + + expect(headers[0][EVENTS_DEPLOYED_HEADER]).toBeUndefined(); + expect(headers[1][EVENTS_DEPLOYED_HEADER]).toBe('1'); + }); + it('reports a handled 200 as settled with no error', async () => { const transport = new DispatcherInvokeTransport('http://dispatcher', 's', { fetchImpl: stubFetch(200, { [EVENTS_HANDLED_HEADER]: '1' }), @@ -153,3 +175,155 @@ describe('DispatcherInvokeTransport', () => { expect(calls).toEqual([`http://dispatcher/prefix${EVENTS_DISPATCH_PATH}`]); }); }); + +const REQUEST: WorkerInvokeRequest = { + script: 'evw-test', + appUid: 'app-test', + handler: 'h', + token: 't', + key: 'k1:test', + event: {}, + ctx: {}, +}; + +const makeClient = () => new EventsWorkerInvokerClient({} as unknown as IConfig); + +/** Answers `missing` until it sees the deployed header, then settles. */ +const missUntilDeployedFetch = ( + headersSeen: Array>, +): FetchImpl => + (async (_url: string, init?: RequestInit) => { + const headers = (init?.headers ?? {}) as Record; + headersSeen.push(headers); + if (headers[EVENTS_DEPLOYED_HEADER] === '1') + return new Response(null, { + status: 200, + headers: { [EVENTS_HANDLED_HEADER]: '1' }, + }); + return new Response(null, { + status: 404, + headers: { [EVENTS_DISPATCH_ERROR_HEADER]: 'missing' }, + }); + }) as unknown as FetchImpl; + +/** Answers `missing` no matter what, deployed header included. */ +const alwaysMissingFetch = (calls: unknown[]): FetchImpl => + (async () => { + calls.push(null); + return new Response(null, { + status: 404, + headers: { [EVENTS_DISPATCH_ERROR_HEADER]: 'missing' }, + }); + }) as unknown as FetchImpl; + +describe('EventsWorkerInvokerClient deploy-on-miss', () => { + it('deploys the script and retries once with the deployed header', async () => { + const headersSeen: Array> = []; + const client = makeClient(); + client.setTransport( + new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl: missUntilDeployedFetch(headersSeen), + }), + ); + const missCalls: Array<[string, string]> = []; + client.setMissHandler(async (appUid, script) => { + missCalls.push([appUid, script]); + return 'deployed'; + }); + + const result = await client.invoke(REQUEST); + + expect(missCalls).toEqual([[REQUEST.appUid, REQUEST.script]]); + expect(headersSeen).toHaveLength(2); + expect(headersSeen[0][EVENTS_DEPLOYED_HEADER]).toBeUndefined(); + expect(headersSeen[1][EVENTS_DEPLOYED_HEADER]).toBe('1'); + expect(result).toEqual({ outcome: 'settled', status: 200 }); + }); + + it('does not retry when the miss handler reports the script is stale', async () => { + const calls: unknown[] = []; + const client = makeClient(); + client.setTransport( + new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl: alwaysMissingFetch(calls), + }), + ); + client.setMissHandler(async () => 'stale'); + + const result = await client.invoke(REQUEST); + + expect(calls).toHaveLength(1); + expect(result).toEqual({ + outcome: 'retriable', + status: null, + error: 'deploy: stale', + }); + }); + + it('never asks the miss handler twice for one invocation', async () => { + const calls: unknown[] = []; + const client = makeClient(); + client.setTransport( + new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl: alwaysMissingFetch(calls), + }), + ); + let handlerCalls = 0; + client.setMissHandler(async () => { + handlerCalls++; + return 'deployed'; + }); + + const result = await client.invoke(REQUEST); + + expect(handlerCalls).toBe(1); + // The first send plus the one retry — never a third. + expect(calls).toHaveLength(2); + expect(result.outcome).toBe('retriable'); + expect(result.status).toBeNull(); + }); + + it('leaves behavior unchanged with no miss handler set', async () => { + const calls: unknown[] = []; + const client = makeClient(); + client.setTransport( + new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl: alwaysMissingFetch(calls), + }), + ); + + const result = await client.invoke(REQUEST); + + expect(calls).toHaveLength(1); + expect(result).toEqual({ + outcome: 'retriable', + status: null, + error: 'dispatcher: missing (404)', + }); + }); + + it('does not trigger the miss handler for a forbidden dispatch reason', async () => { + const client = makeClient(); + client.setTransport( + new DispatcherInvokeTransport('http://dispatcher', 's', { + fetchImpl: stubFetch(403, { + [EVENTS_DISPATCH_ERROR_HEADER]: 'forbidden', + }), + }), + ); + let handlerCalled = false; + client.setMissHandler(async () => { + handlerCalled = true; + return 'deployed'; + }); + + const result = await client.invoke(REQUEST); + + expect(handlerCalled).toBe(false); + expect(result).toEqual({ + outcome: 'retriable', + status: null, + error: 'dispatcher: forbidden (403)', + }); + }); +}); diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.ts b/src/backend/clients/events/EventsWorkerInvokerClient.ts index b525e7df3..b70689f1b 100644 --- a/src/backend/clients/events/EventsWorkerInvokerClient.ts +++ b/src/backend/clients/events/EventsWorkerInvokerClient.ts @@ -40,6 +40,9 @@ import { PuterClient } from '../types.js'; * * The delivery token rides in the body, not a header, so nothing between here * and the isolate can treat it as its own authorization. + * + * 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. */ /** The one route an events worker answers. */ @@ -65,6 +68,31 @@ export const EVENTS_HANDLED_HEADER = 'x-puter-events-handled'; /** Machine-readable reason on the runtime's own failure answers. */ export const EVENTS_ERROR_HEADER = 'x-puter-events-error'; +/** + * Set on the retry after this backend deploys a missing script itself, so the + * dispatcher skips its own deploy-on-miss callback and negative cache and goes + * straight to retrying the namespace. + */ +export const EVENTS_DEPLOYED_HEADER = 'x-puter-events-deployed'; + +/** + * Dispatch-error reasons this backend can resolve itself by deploying, rather + * than wait on the rehydrate callback finding a backend behind the dispatcher's + * own hostname. + */ +const SELF_DEPLOYABLE_MISS_REASONS = new Set([ + 'missing', + 'deploy-failed', + 'deploy-timeout', + 'deploy-error', +]); + +/** What a self-deploy attempt did, and to which app/script it applies. */ +export type WorkerMissHandler = ( + appUid: string, + script: string, +) => Promise<'deployed' | string>; + /** How long a handler has to answer before the attempt is abandoned. */ export const EVENTS_INVOKE_TIMEOUT_MS = 30_000; @@ -99,6 +127,8 @@ export interface EventsInvokeCall { /** Serialized `{ handler, event, ctx, token }`. */ body: string; timeoutMs: number; + /** Set on the retry after this backend deployed the script itself. */ + deployed?: boolean; } /** @@ -115,6 +145,8 @@ export interface EventsInvokeTransport { status: number | null; handled?: boolean; error?: string; + /** Set only by the dispatcher transport, naming its own failure. */ + dispatchReason?: string; }>; } @@ -145,12 +177,24 @@ export class EventsWorkerInvokerClient extends PuterClient { /** Set only where invocations do not leave the process — local development. */ #transport: EventsInvokeTransport | null = null; #dispatcher: DispatcherInvokeTransport | null = null; + /** Lets this backend resolve a dispatcher miss instead of waiting on it. */ + #missHandler: WorkerMissHandler | null = null; /** Replaces the dispatcher with an in-process runtime. */ setTransport(transport: EventsInvokeTransport): void { this.#transport = transport; } + /** + * Deploys a script the dispatcher reports missing, in place of the + * rehydrate callback — for the case where that callback lands on a backend + * other than this one. Set once; `invoke()` calls it at most once per + * delivery and never loops. + */ + setMissHandler(handler: WorkerMissHandler): void { + this.#missHandler = handler; + } + override onServerShutdown(): void { this.#dispatcher?.close(); this.#dispatcher = null; @@ -161,7 +205,7 @@ export class EventsWorkerInvokerClient extends PuterClient { if (!transport) return { outcome: 'retriable', status: null, error: NO_TRANSPORT }; - const { status, handled, error } = await transport.send({ + const call: EventsInvokeCall = { script: request.script, appUid: request.appUid, key: request.key, @@ -172,8 +216,29 @@ export class EventsWorkerInvokerClient extends PuterClient { token: request.token, }), timeoutMs: this.#timeoutMs(), - }); + }; + let result = await transport.send(call); + if ( + result.status === null && + this.#missHandler && + result.dispatchReason && + SELF_DEPLOYABLE_MISS_REASONS.has(result.dispatchReason) + ) { + const outcome = await this.#missHandler( + request.appUid, + request.script, + ); + if (outcome !== 'deployed') + return { + outcome: 'retriable', + status: null, + error: `deploy: ${outcome}`, + }; + result = await transport.send({ ...call, deployed: true }); + } + + const { status, handled, error } = result; if (status === null) return { outcome: 'retriable', status: null, error }; @@ -241,6 +306,7 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport { status: number | null; handled?: boolean; error?: string; + dispatchReason?: string; }> { try { // `new URL(path, base)` resolves an absolute path against the @@ -254,6 +320,7 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport { '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.body, signal: AbortSignal.timeout(call.timeoutMs), @@ -272,6 +339,7 @@ export class DispatcherInvokeTransport implements EventsInvokeTransport { return { status: null, error: `dispatcher: ${dispatchError} (${response.status})`, + dispatchReason: dispatchError, }; const handled = response.headers.get(EVENTS_HANDLED_HEADER) === '1'; diff --git a/src/backend/controllers/events/EventsController.ts b/src/backend/controllers/events/EventsController.ts index 3503a4db3..fd9502d0d 100644 --- a/src/backend/controllers/events/EventsController.ts +++ b/src/backend/controllers/events/EventsController.ts @@ -388,12 +388,22 @@ export class EventsController extends PuterController { * Local development has no events dispatcher, so invocations are handed * straight to the local worker runtime — including the deploy-on-miss the * dispatcher would otherwise ask for over the rehydrate route above. + * + * Elsewhere, this backend already knows the app and script a miss names, so + * it deploys the script itself rather than count on the rehydrate callback + * landing on a backend behind the dispatcher's own hostname. */ override onServerStart(): void { const deployer = this.#eventsWorkerDeployer(); - if (!deployer.enabled || !this.config.workers?.localServer) return; - this.services.events.useWorkerTransport( - new LocalEventsInvokeTransport(this.services, deployer), + if (!deployer.enabled) return; + if (this.config.workers?.localServer) { + this.services.events.useWorkerTransport( + new LocalEventsInvokeTransport(this.services, deployer), + ); + return; + } + this.services.events.useWorkerMissHandler((appUid, script) => + deployer.ensure(appUid, script), ); } diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 4ce9fcb4e..b965a2848 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -203,7 +203,10 @@ import { type SubjectOp, } from './subjects.js'; import { backlogPolicyFor, isResumable } from './suspension.js'; -import type { EventsInvokeTransport } from '../../clients/events/EventsWorkerInvokerClient.js'; +import type { + EventsInvokeTransport, + WorkerMissHandler, +} from '../../clients/events/EventsWorkerInvokerClient.js'; import { EVENTS_WORKER_SESSION_NAME, eventsInvokeKey, @@ -4502,6 +4505,14 @@ export class EventsService extends PuterService { this.clients.eventsWorkerInvoker.setTransport(transport); } + /** + * Let this backend deploy a script the dispatcher reports missing, instead + * of waiting on the rehydrate callback to reach it. + */ + useWorkerMissHandler(handler: WorkerMissHandler): void { + this.clients.eventsWorkerInvoker.setMissHandler(handler); + } + /** * Where an app's handlers currently live: the script its published set * hashes to, and the key an invocation of it carries. diff --git a/src/backend/services/events/workerInvoker.integration.test.ts b/src/backend/services/events/workerInvoker.integration.test.ts index e7e0d92b0..837b4c6c5 100644 --- a/src/backend/services/events/workerInvoker.integration.test.ts +++ b/src/backend/services/events/workerInvoker.integration.test.ts @@ -68,7 +68,7 @@ const SOURCE = interface StubCall { method: string; path: string; - headers: Record; + headers: Record & { deployed?: string }; body: { handler?: string; token?: string; @@ -89,6 +89,8 @@ let calls: StubCall[]; let answer: number | 'hang' = 200; /** Whether the stub's answer carries the handled header, as a real one would. */ let answerHandled = true; +/** Whether the stub reports the script missing until it sees the deployed header. */ +let dispatchMissing = false; const events = () => env.server.services.events; const pending = () => env.server.stores.pendingDelivery; @@ -103,6 +105,9 @@ const readBody = async (req: http.IncomingMessage): Promise => { const startStub = async (): Promise => { stub = http.createServer((req, res) => { void readBody(req).then((raw) => { + const deployed = req.headers['x-puter-events-deployed'] as + | string + | undefined; calls.push({ method: req.method ?? '', path: req.url ?? '', @@ -111,10 +116,17 @@ const startStub = async (): Promise => { script: req.headers['x-puter-events-script'] as string, app: req.headers['x-puter-events-app'] as string, key: req.headers['x-puter-events-key'] as string, + deployed, }, body: JSON.parse(raw || '{}') as StubCall['body'], }); if (answer === 'hang') return; + // Stands in for the events dispatcher reporting a namespace miss + // until the deploy-on-miss retry carries the deployed header. + if (dispatchMissing && deployed !== '1') { + res.writeHead(404, { 'x-puter-events-dispatch': 'missing' }).end(); + return; + } // Stands in for the dispatcher forwarding a genuine answer from // the script, so it carries the marker a real one would — unless // a test is specifically simulating something that never reached one. @@ -264,6 +276,10 @@ beforeAll(async () => { invokeTimeoutMs: INVOKE_TIMEOUT_MS, dispatcherUrl, internalSecret: INTERNAL_SECRET, + // Only reached by the deploy-on-miss test below, whose + // `drivers.workers.create` is stubbed — a real deploy target + // never sees this. + workerNamespace: 'ev-test-ns', }, // Seeded accounts carry no email, which the plan machinery reads as a // temporary account — and a temporary account holds no durable rows. @@ -317,6 +333,7 @@ beforeEach(async () => { calls = []; answer = 200; answerHandled = true; + dispatchMissing = false; await env.server.clients.db.write('DELETE FROM `event_subscriptions`', []); events().invalidateUser(userId); await env.server.stores.eventSubscription.markRegionCold(userId); @@ -820,3 +837,30 @@ describe("what withdrawing an app's standing does to its worker session", () => ).resolves.toMatchObject({ actor: expect.anything() }); }); }); + +describe('deploying on a dispatcher miss', () => { + it('deploys the app`s own script itself and settles once the dispatcher sees it', async () => { + dispatchMissing = true; + // Standing in for the real deploy target (Cloudflare or a local + // worker runtime) — this test pins the retry protocol, not the + // deploy mechanics, which `workerDeploy.test.ts` already covers. + const createSpy = vi + .spyOn(env.server.drivers.workers, 'create') + .mockResolvedValue({ success: true, errors: [] }); + + try { + const subId = await subscribe(); + await touch('deploy-on-miss.txt'); + await invoked(2); + + expect(createSpy).toHaveBeenCalledTimes(1); + expect(calls[0].headers.deployed).toBeUndefined(); + expect(calls[1].headers.deployed).toBe('1'); + await vi.waitFor(async () => + expect(await pending().depth(subId)).toBe(0), + ); + } finally { + createSpy.mockRestore(); + } + }); +});