From 4e3eb4e65f2814691c4d468b379f9c733fff28fa Mon Sep 17 00:00:00 2001 From: Sam Goodwin Date: Fri, 17 Jul 2026 18:43:14 -0700 Subject: [PATCH] fix(cloudflare/workers): keep the previous dev worker instance serving until its replacement starts (#859) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under alchemy dev, a reconcile that replaced a Worker instance ("Changes detected, interrupting existing instance") tore down the running workerd — and deleted its dev-registry entry — before the replacement completed runtime.start. For DO/container hosts, whose start includes docker image prep, that dark window spans seconds to minutes and re-opens on every change; cross-script DO bindings resolve through the registry and 503 with 'Durable Object "X" defined in worker "y" not found' the whole time. Make the serve path make-before-break: - workerd scopes are forked from the provider root scope, not the instance scope, so tearing down an instance's bundle watcher no longer kills the running workerd; the last good instance keeps serving until the replacement's first serve completes. - serveWith starts the replacement first, cuts the proxy over, then closes the previous scope. Registry safety for the shared key comes from owner-aware entry removal in cloudflare-runtime (cloudflare-tools#72), which unblocks reverting the close-old-first ordering #812 introduced. - delete/external-mode teardown now closes the workerd scope explicitly. Co-Authored-By: Claude Fable 5 --- cloudflare-tools | 2 +- .../Cloudflare/Workers/LocalWorkerProvider.ts | 153 +++++++++------ .../Workers/LocalWorkerHandoff.test.ts | 178 ++++++++++++++++++ .../Workers/fixtures/handoff-worker.ts | 70 +++++++ 4 files changed, 343 insertions(+), 60 deletions(-) create mode 100644 packages/alchemy/test/Cloudflare/Workers/LocalWorkerHandoff.test.ts create mode 100644 packages/alchemy/test/Cloudflare/Workers/fixtures/handoff-worker.ts diff --git a/cloudflare-tools b/cloudflare-tools index c09ef1de2..6c2eebb7b 160000 --- a/cloudflare-tools +++ b/cloudflare-tools @@ -1 +1 @@ -Subproject commit c09ef1de2b1ee23a6ceb4ea8faf0e0ee091a274d +Subproject commit 6c2eebb7b088abbc074e99102db16493c90cfb6c diff --git a/packages/alchemy/src/Cloudflare/Workers/LocalWorkerProvider.ts b/packages/alchemy/src/Cloudflare/Workers/LocalWorkerProvider.ts index 9ba51d900..05b32a263 100644 --- a/packages/alchemy/src/Cloudflare/Workers/LocalWorkerProvider.ts +++ b/packages/alchemy/src/Cloudflare/Workers/LocalWorkerProvider.ts @@ -233,17 +233,13 @@ export const LocalWorkerProvider = () => // Latest successful serve per worker id, so runtime wiring changes // that arrive AFTER workerd started (e.g. a sibling `Consumer` // resource registering this script as a queue consumer) can restart - // the instance with the same bundle. `scope` is the parent scope the - // workerd child scope is forked from (the instance scope), NOT the - // caller's ambient scope — restarts are triggered from other - // providers' reconcile fibers whose scopes must not own workerd. + // the instance with the same bundle. const latestServes = new Map< string, { worker: WorkerConfig; bundle: Bundle.BundleOutput; proxy: WorkerProxy.WorkerProxyInstance; - scope: Scope.Scope; } >(); // Serializes serves per worker id: a restart triggered by a sibling @@ -259,64 +255,90 @@ export const LocalWorkerProvider = () => return lock; }; + // Serve with make-before-break semantics: start the replacement + // workerd while the previous instance (if any) keeps serving — and + // stays registered in the dev registry — then cut the proxy over and + // tear the previous instance down. Cross-script consumers (e.g. a DO + // bound via `scriptName` from another Worker) therefore never observe + // a window where the script has no running instance and no registry + // entry, even when `runtime.start` is slow (container image builds). + // Both instances use the same registry key; the registry's entry + // removal is owner-aware, so closing the old scope after the + // replacement has re-registered cannot delete the replacement's + // registration. + // + // The workerd scope is forked from the provider's `rootScope`, NOT + // the instance scope: a reconcile that replaces the instance (or a + // restart triggered from a sibling provider's fiber) tears down the + // bundle watcher without killing the currently serving workerd — the + // last good instance keeps serving until the replacement's first + // serve completes. Ownership is tracked in `workerdScopes`, closed by + // the next successful serve, by `delete`, or by provider shutdown. const serveWith = ( worker: WorkerConfig, bundle: Bundle.BundleOutput, proxy: WorkerProxy.WorkerProxyInstance, - parentScope: Scope.Scope, ) => Semaphore.withPermits( serveLock(worker.id), 1, )( - Effect.gen(function* () { - const previous = workerdScopes.get(worker.id); - if (previous) { - // Both runtimes use the same registry key. Close the old scope first so - // its unregister finalizer cannot delete the replacement registration. - yield* Scope.close(previous, Exit.void); - workerdScopes.delete(worker.id); - } - const scope = yield* Scope.fork(parentScope); - const url = yield* runtime - .start({ - name: worker.name, - compatibilityDate: worker.compatibility.date, - compatibilityFlags: worker.compatibility.flags, - bindings: worker.workerBindings as never, - hyperdrives: worker.hyperdrives, - durableObjectNamespaces: worker.durableObjectNamespaces, - queueConsumers: yield* getQueueConsumers(worker.name), - modules: yield* toRuntimeModules(bundle), - assets: toRuntimeAssets(worker.assets), - }) - .pipe(Scope.provide(scope)); - workerdScopes.set(worker.id, scope); - latestServes.set(worker.id, { - worker, - bundle, - proxy, - scope: parentScope, - }); - MutableHashMap.set( - localRuntimeState.workerRestarts, - worker.name, - restartWorker(worker.id), - ); - yield* proxy.set(url); - return url; - }), + // The bookkeeping after `runtime.start` must not be torn in half + // by an interrupt: once the replacement workerd is up, it must be + // recorded in `workerdScopes` and the previous instance must be + // closed, or one of the two workerds would leak until provider + // shutdown while holding the shared registry key. + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const scope = yield* Scope.fork(rootScope); + const url = yield* restore( + runtime + .start({ + name: worker.name, + compatibilityDate: worker.compatibility.date, + compatibilityFlags: worker.compatibility.flags, + bindings: worker.workerBindings as never, + hyperdrives: worker.hyperdrives, + durableObjectNamespaces: worker.durableObjectNamespaces, + queueConsumers: yield* getQueueConsumers(worker.name), + modules: yield* toRuntimeModules(bundle), + assets: toRuntimeAssets(worker.assets), + }) + .pipe(Scope.provide(scope)), + ).pipe( + // The scope hangs off `rootScope`, so a failed or + // interrupted start must close it here — nothing else owns + // it yet. + Effect.onExit((exit) => + exit._tag === "Failure" + ? Scope.close(scope, exit) + : Effect.void, + ), + ); + const previous = workerdScopes.get(worker.id); + workerdScopes.set(worker.id, scope); + latestServes.set(worker.id, { worker, bundle, proxy }); + MutableHashMap.set( + localRuntimeState.workerRestarts, + worker.name, + restartWorker(worker.id), + ); + yield* proxy.set(url); + if (previous) { + yield* Scope.close(previous, Exit.void).pipe( + Effect.catchCause((cause) => + Effect.logWarning( + `[${worker.id}] Failed to stop previous local worker instance`, + Cause.squash(cause), + ), + ), + ); + } + return url; + }), + ), ); - const serveScoped = Effect.fn(function* ( - worker: WorkerConfig, - bundle: Bundle.BundleOutput, - proxy: WorkerProxy.WorkerProxyInstance, - ) { - const parentScope = yield* Effect.scope; - return yield* serveWith(worker, bundle, proxy, parentScope); - }); - /** * Restart a running worker with its latest bundle so start-time * runtime wiring (queue consumers) is re-read from @@ -327,12 +349,7 @@ export const LocalWorkerProvider = () => Effect.suspend(() => { const latest = latestServes.get(id); if (!latest) return Effect.void; - return serveWith( - latest.worker, - latest.bundle, - latest.proxy, - latest.scope, - ).pipe( + return serveWith(latest.worker, latest.bundle, latest.proxy).pipe( Effect.asVoid, Effect.catchCause((cause) => Effect.logWarning( @@ -343,6 +360,18 @@ export const LocalWorkerProvider = () => ); }); + // Tear down the running workerd for a worker id, if any. Used when the + // Worker is deleted or handed off to an external dev process — + // instance replacement does NOT go through this: the previous workerd + // keeps serving until the replacement's first serve closes it. + const closeWorkerd = Effect.fn(function* (id: string) { + const scope = workerdScopes.get(id); + if (scope) { + workerdScopes.delete(id); + yield* Scope.close(scope, Exit.void); + } + }); + // Note: `serveLocks` entries are intentionally retained — an // in-flight restart may still hold the semaphore when the instance // is torn down, and a same-id re-create must serialize against it. @@ -521,7 +550,7 @@ export const LocalWorkerProvider = () => : Result.failVoid, ), Stream.mapEffect((bundle) => - serveScoped(worker, bundle, proxy).pipe( + serveWith(worker, bundle, proxy).pipe( Effect.exit, Effect.tap((exit) => { if (exit._tag === "Success") { @@ -709,6 +738,7 @@ export const LocalWorkerProvider = () => instances.delete(id); dropServeState(id); } + yield* closeWorkerd(id); const name = yield* createWorkerName(id, news.name); return { workerId: name, @@ -737,6 +767,10 @@ export const LocalWorkerProvider = () => yield* Effect.log( `[${options.id}] Changes detected, interrupting existing instance`, ); + // Tears down the instance's bundle watcher and any in-flight + // serve — but NOT its running workerd, which keeps serving (and + // stays registered in the dev registry) until the replacement + // instance's first serve completes and cuts over. yield* Fiber.interrupt(existing.fiber); yield* Scope.close(existing.scope, Exit.void); instances.delete(options.id); @@ -766,6 +800,7 @@ export const LocalWorkerProvider = () => instances.delete(id); dropServeState(id); } + yield* closeWorkerd(id); }), }; }), diff --git a/packages/alchemy/test/Cloudflare/Workers/LocalWorkerHandoff.test.ts b/packages/alchemy/test/Cloudflare/Workers/LocalWorkerHandoff.test.ts new file mode 100644 index 000000000..d15c545db --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/LocalWorkerHandoff.test.ts @@ -0,0 +1,178 @@ +import * as Cloudflare from "@/Cloudflare"; +import * as Alchemy from "@/index.ts"; +import * as RpcProviderProxy from "@/Local/RpcProviderProxy"; +import * as RpcSpawner from "@/Local/RpcSpawner"; +import * as Test from "@/Test/Alchemy"; +import { expect } from "alchemy-test"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import { MinimumLogLevel } from "effect/References"; +import * as Schedule from "effect/Schedule"; +import * as HttpClient from "effect/unstable/http/HttpClient"; +import * as Option from "effect/Option"; +import * as Os from "node:os"; +import HandoffWorkerLive, { + HandoffWorker, + setDeployN, +} from "./fixtures/handoff-worker.ts"; + +const { test, deploy, destroy } = Test.make({ + providers: Cloudflare.providers(), + state: Alchemy.localState(), + dev: true, +}); + +const logLevel = Effect.provideService( + MinimumLogLevel, + process.env.DEBUG ? "Debug" : "Info", +); + +/** + * Run the local providers in a persistent RPC sidecar, exactly like + * `alchemy dev`. Without this, each `deploy(Stack)` builds a fresh provider + * (fresh in-memory instance maps), so the second deploy never takes the + * "Changes detected, interrupting existing instance" replacement path this + * test exists to cover. The proxy caches its sidecar session per entry URL, + * so both deploys talk to the same provider process. + */ +const Sidecar = Layer.unwrap( + Effect.map(RpcSpawner.RpcSpawner, (spawner) => + RpcProviderProxy.layer(spawner.url), + ), +).pipe( + Layer.provideMerge( + RpcSpawner.layerServer({ + profile: process.env.ALCHEMY_PROFILE, + envFile: undefined, + }), + ), +); + +const Stack = Alchemy.Stack( + "LocalWorkerHandoffStack", + { providers: Cloudflare.providers(), state: Alchemy.localState() }, + Effect.gen(function* () { + const worker = yield* HandoffWorker; + return { + url: worker.url.as(), + workerName: worker.workerName.as(), + }; + }).pipe(Effect.provide(HandoffWorkerLive)), +); + +const fetchDeploy = (url: string) => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + const res = yield* client.get(url); + if (res.status !== 200) { + const text = yield* res.text; + return yield* Effect.fail( + new Error(`status ${res.status}: ${text.slice(0, 500)}`), + ); + } + const body = (yield* res.json) as { deploy: string; pong: string }; + expect(body.pong).toBe("pong"); + return body.deploy; + }).pipe(Effect.timeout("10 seconds")); + +const registryEntryPath = (workerName: string) => + Effect.gen(function* () { + const path = yield* Path.Path; + const stateHome = + process.env.XDG_STATE_HOME ?? path.join(Os.homedir(), ".local", "state"); + return path.join( + stateHome, + "alchemy", + "registry", + `${encodeURIComponent(workerName)}.json`, + ); + }); + +const registryEntryExists = (workerName: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + return yield* fs.exists(yield* registryEntryPath(workerName)); + }); + +/** + * The registry entry's inode. A make-before-break handoff only ever + * overwrites `{scriptName}.json` in place (the replacement re-registers, + * then the owner-aware unregister of the old instance no-ops), so the inode + * is stable across a redeploy. The broken path unlinks the entry when the + * old instance is torn down and recreates it when the replacement finally + * serves — observable as an inode change even when the dark window itself + * is too short to catch with a request. + */ +const registryEntryIno = (workerName: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const stat = yield* fs.stat(yield* registryEntryPath(workerName)); + return Option.getOrUndefined(stat.ino); + }); + +/** + * Regression test for alchemy-run/alchemy#859: replacing a local dev Worker + * instance (a reconcile with a changed structural signature logs "Changes + * detected, interrupting existing instance") must not tear down the running + * workerd before the replacement has started. The last good instance keeps + * serving — and keeps its dev-registry entry, which cross-script DO bindings + * resolve through — until the replacement's first serve completes. + */ +test( + "previous instance keeps serving and stays registered during redeploy", + Effect.gen(function* () { + setDeployN("1"); + const first = yield* deploy(Stack); + + // Wait for the first instance to actually serve (fresh workerd + DO). + const initial = yield* fetchDeploy(first.url).pipe( + Effect.retry({ + schedule: Schedule.min([ + Schedule.exponential("500 millis"), + Schedule.spaced("2 seconds"), + ]), + times: 20, + }), + ); + expect(initial).toBe("1"); + expect(yield* registryEntryExists(first.workerName)).toBe(true); + const inoBefore = yield* registryEntryIno(first.workerName); + + // Redeploy with a changed env var: the Worker's structural signature + // changes, so the provider replaces the instance. The deploy returns as + // soon as reconcile completes — before the replacement workerd has + // started serving. + setDeployN("2"); + const second = yield* deploy(Stack); + + // The previous instance must still be serving (make-before-break): a + // single un-retried request must succeed, whether it is answered by the + // old instance ("1") or an already-cut-over replacement ("2"). + const during = yield* fetchDeploy(second.url); + expect(["1", "2"]).toContain(during); + + // ...and its dev-registry entry must still exist, so cross-script DO + // bindings from other workers keep resolving throughout the handoff. + expect(yield* registryEntryExists(second.workerName)).toBe(true); + + // The replacement eventually takes over. + const final = yield* fetchDeploy(second.url).pipe( + Effect.flatMap((deploy) => + deploy === "2" + ? Effect.succeed(deploy) + : Effect.fail(new Error(`still serving deploy ${deploy}`)), + ), + Effect.retry({ schedule: Schedule.spaced("1 second"), times: 90 }), + ); + expect(final).toBe("2"); + expect(yield* registryEntryExists(second.workerName)).toBe(true); + // The entry was overwritten in place by the replacement, never deleted — + // cross-script DO bindings resolving through it never saw it vanish. + expect(yield* registryEntryIno(second.workerName)).toBe(inoBefore); + + yield* destroy(Stack); + }).pipe(Effect.provide(Sidecar), logLevel), + { timeout: 240_000 }, +); diff --git a/packages/alchemy/test/Cloudflare/Workers/fixtures/handoff-worker.ts b/packages/alchemy/test/Cloudflare/Workers/fixtures/handoff-worker.ts new file mode 100644 index 000000000..2dd6f9533 --- /dev/null +++ b/packages/alchemy/test/Cloudflare/Workers/fixtures/handoff-worker.ts @@ -0,0 +1,70 @@ +import * as Cloudflare from "@/Cloudflare"; +import type { RuntimeContext } from "@/index"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"; + +/** + * A DO-hosting Effect Worker whose `DEPLOY_N` env var is read from a mutable + * module variable at plan time, so consecutive deploys can force a + * structural-signature change — and therefore an instance replacement — + * without editing any source file. The props Effect resolves in the engine + * (test) process; the resulting plain string is what crosses the RPC + * boundary to the dev sidecar. + */ +let deployN = "0"; +export const setDeployN = (n: string) => { + deployN = n; +}; + +export class HandoffObject extends Cloudflare.DurableObject< + HandoffObject, + { + ping: () => Effect.Effect; + } +>()("HandoffObject") {} + +export const HandoffObjectLive = HandoffObject.make( + Effect.gen(function* () { + return Effect.gen(function* () { + return { + ping: () => Effect.succeed("pong"), + }; + }); + }), +); + +export class HandoffWorker extends Cloudflare.Worker< + HandoffWorker, + {}, + HandoffObject +>()("HandoffWorker") {} + +export default HandoffWorker.make( + Effect.sync(() => ({ + main: import.meta.url, + env: { + DEPLOY_N: deployN, + }, + })), + Effect.gen(function* () { + const objects = yield* HandoffObject; + return { + fetch: Effect.gen(function* () { + const env = yield* Cloudflare.Workers.WorkerEnvironment; + const pong = yield* objects.getByName("test").ping(); + return yield* HttpServerResponse.json({ + deploy: String(env.DEPLOY_N), + pong, + }); + }).pipe( + Effect.catchCause((cause) => + HttpServerResponse.json( + { error: Cause.pretty(cause) }, + { status: 500 }, + ), + ), + ), + }; + }).pipe(Effect.provide(HandoffObjectLive)), +);