feat: the invoking backend deploys an events worker the dispatcher cannot find (#3807)

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.

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<Record<string, string>> = [];
+        const fetchImpl = (async (_url: string, init?: RequestInit) => {
+            headers.push(init?.headers as Record<string, string>);
+            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<Record<string, string>>,
+): FetchImpl =>
+    (async (_url: string, init?: RequestInit) => {
+        const headers = (init?.headers ?? {}) as Record<string, string>;
+        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<Record<string, string>> = [];
+        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)',
+        });
+    });
+});

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';

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),
         );
     }
 

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.

src/backend/services/events/workerInvoker.integration.test.ts

@@ -68,7 +68,7 @@ const SOURCE =
 interface StubCall {
     method: string;
     path: string;
-    headers: Record<string, string | undefined>;
+    headers: Record<string, string | undefined> & { 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<string> => {
 const startStub = async (): Promise<string> => {
     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<string> => {
                     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();
+        }
+    });
+});