1import * as Clock from "effect/Clock";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Exit from "effect/Exit";
5import * as Layer from "effect/Layer";
6import * as Queue from "effect/Queue";
7import * as Result from "effect/Result";
8import * as Scope from "effect/Scope";
9import * as Docker from "./Docker.ts";
10import type * as Globals from "./globals/Globals.ts";
11import * as Storage from "./globals/Storage.ts";
12import {
13 defaultDurableObjectUniqueKey,
14 SERVICE_USER_WORKER,
15 SOCKET_USER_ENTRY,
16 withDefaultFlags,
17} from "./internal/constants.ts";
18import { moduleToWorkerd } from "./internal/internal-modules.ts";
19import type { BindingHook } from "./PluginContext.ts";
20import * as PluginContext from "./PluginContext.ts";
21import * as RegistryProxy from "./registry/RegistryProxy.ts";
22import { type RuntimeError, SystemError } from "./RuntimeError.shared.ts";
23import type { BindingHooks, RuntimeWorker } from "./RuntimeWorker.ts";
24import type * as WorkerdConfig from "./workerd/Config.ts";
25import * as Workerd from "./workerd/Workerd.ts";
26
27export class Runtime extends Context.Service<
28 Runtime,
29 {
30 readonly start: <B extends BindingHooks>(
31 worker: RuntimeWorker<B>,
32 ) => Effect.Effect<URL, RuntimeError, BindingRequirements<B> | Scope.Scope>;
33 }
34>()("cloudflare-runtime/Runtime") {}
35
36type BindingRequirements<B extends BindingHooks> =
37 B extends Array<never>
38 ? never
39 : B extends Array<BindingHook<infer R>>
40 ? R
41 : never;
42
43export const RuntimeLive = Layer.effect(
44 Runtime,
45 Effect.gen(function* () {
46 const workerd = yield* Workerd.Workerd;
47 const storage = yield* Storage.Storage;
48 const docker = yield* Docker.Docker;
49 const plugins =
50 yield* PluginContext.pickPluginsFromContext<Globals.Globals>();
51
52 const preparePlugins = Effect.fnUntraced(function* (worker: RuntimeWorker) {
53 const context = yield* PluginContext.make(
54 worker as RuntimeWorker,
55 plugins,
56 );
57 const [bindings, { tails, streamingTails }] = yield* Effect.all(
58 [
59 Effect.all(worker.bindings as ReadonlyArray<BindingHook<never>>, {
60 concurrency: "unbounded",
61 }).pipe(Effect.provideService(PluginContext.PluginContext, context)),
62 prepareTails(worker, context),
63 ],
64 { concurrency: "unbounded" },
65 );
66 return {
67 config: yield* context.config,
68 context,
69 bindings,
70 tails,
71 streamingTails,
72 };
73 });
74
75 /**
76 * Resolve each (streaming) tail consumer name to a `ServiceDesignator`
77 * through the dev registry proxy — the same path service bindings to other
78 * local workers take (`Service.local`). The registry proxy's
79 * `ExternalService` entrypoint forwards `tail()` batches and
80 * `tailStream()` sessions to the consumer's process via the workerd debug
81 * port, so consumers may live in other dev processes and may (re)start at
82 * any time. Subscriptions must be registered before `context.config` is
83 * computed, which is why this runs alongside the binding hooks in
84 * `preparePlugins`.
85 */
86 const prepareTails = Effect.fnUntraced(function* (
87 worker: RuntimeWorker,
88 context: PluginContext.PluginContext["Service"],
89 ) {
90 if (!worker.tails?.length && !worker.streamingTails?.length) {
91 return { tails: undefined, streamingTails: undefined };
92 }
93 const proxy = yield* context.get(RegistryProxy.RegistryProxy);
94 const subscribeAll = (names: ReadonlyArray<string> | undefined) =>
95 names?.length
96 ? Effect.forEach([...new Set(names)], (scriptName) =>
97 proxy.api.subscribe({ kind: "worker", scriptName }),
98 )
99 : Effect.succeed(undefined);
100 const [tails, streamingTails] = yield* Effect.all(
101 [subscribeAll(worker.tails), subscribeAll(worker.streamingTails)],
102 { concurrency: "unbounded" },
103 );
104 return { tails, streamingTails };
105 });
106
107 const prepareContainers = Effect.fnUntraced(function* (
108 worker: RuntimeWorker,
109 ) {
110 const containers = (worker.durableObjectNamespaces ?? []).flatMap(
111 (namespace) => {
112 if (!namespace.container) return [];
113 return {
114 className: namespace.className,
115 container: namespace.container,
116 };
117 },
118 );
119 if (!containers.length) {
120 return { imageNames: new Map() };
121 }
122 // Local development with containers relies on pulling/building `linux/amd64`
123 // images, which the Docker daemon on Windows cannot do (it runs Windows
124 // containers). Upstream workers-sdk bails out on Windows for the same
125 // reason, directing users to WSL.
126 if (process.platform === "win32") {
127 return yield* new SystemError({
128 subtag: "ContainersUnsupportedOnWindows",
129 message:
130 "Local development with containers is not supported on Windows.",
131 hint: "Use WSL to develop the container part of your application, or remove the container configuration if you do not need it.",
132 });
133 }
134 const imageNames = new Map<string, string>();
135
136 const registerImage = (
137 className: string,
138 tag: string,
139 env?: Record<string, string>,
140 ) => {
141 if (env) {
142 // To prevent collisions between images with the same tag but different env,
143 // `registerImageEnv` returns a unique alias for the image, which our Docker
144 // proxy server then maps to the actual tag and injects the env variables.
145 return docker
146 .registerImageEnv(className, tag, env)
147 .pipe(
148 Effect.andThen((alias) =>
149 Effect.sync(() => imageNames.set(className, alias)),
150 ),
151 );
152 }
153 return Effect.sync(() => imageNames.set(className, tag));
154 };
155
156 const [, containerEngine] = yield* Effect.forEach(
157 containers,
158 ({ className, container }) => {
159 if ("tag" in container) {
160 return docker
161 .validate(container.tag)
162 .pipe(
163 Effect.andThen(
164 registerImage(className, container.tag, container.env),
165 ),
166 );
167 }
168 const tag = docker.generateImageTag(className);
169 const prepare =
170 "imageUri" in container
171 ? docker.pull(tag, container)
172 : docker.build(tag, container);
173 return prepare.pipe(
174 Effect.andThen(docker.validate(tag)),
175 Effect.tap(() => {
176 // Each start cleans up ONLY its own image tag when its scope
177 // closes. Do NOT prune other same-name tags as "stale" here: a
178 // dev session starts the worker more than once (precreate stub
179 // → reconcile), and a cleanup that guesses which sibling tags
180 // are dead can untag the tag a live workerd is about to
181 // `docker create` from — every container start then fails and
182 // the session serves 500s until redeploy.
183 return Effect.addFinalizer(() =>
184 docker
185 .removeContainer(tag)
186 .pipe(
187 Effect.andThen(docker.removeImageTag(tag)),
188 Effect.ignore,
189 ),
190 );
191 }),
192 Effect.tap(() => registerImage(className, tag, container.env)),
193 );
194 },
195 { concurrency: "unbounded", discard: true },
196 ).pipe(
197 Effect.zip(docker.getWorkerdDockerConfiguration, { concurrent: true }),
198 );
199 return { imageNames, containerEngine };
200 });
201
202 return Runtime.of({
203 start: Effect.fn(function* (worker) {
204 worker = {
205 ...worker,
206 compatibilityFlags: withDefaultFlags(worker.compatibilityFlags, {
207 date: worker.compatibilityDate,
208 }),
209 };
210 const [
211 { config, context, bindings, tails, streamingTails },
212 { containerEngine, imageNames },
213 ] = yield* Effect.all(
214 [preparePlugins(worker), prepareContainers(worker)],
215 {
216 concurrency: "unbounded",
217 },
218 );
219 const sockets: Array<WorkerdConfig.Socket> = [
220 {
221 name: SOCKET_USER_ENTRY,
222 address: "127.0.0.1:0",
223 service: { name: config.entry ?? SERVICE_USER_WORKER },
224 },
225 ...config.sockets,
226 ];
227 const exits = yield* Queue.unbounded<Workerd.WorkerdExit>();
228 /**
229 * Serves the Worker. `pinned` reuses the ports of a previous process,
230 * so a replacement is reachable through the same URL, proxy target
231 * and registry entry as the process it replaces.
232 */
233 const serveWorker = (pinned?: Workerd.WorkerdPorts) =>
234 workerd.serve(
235 {
236 sockets: pinned ? sockets.map(pinSocket(pinned)) : sockets,
237 services: [
238 {
239 name: SERVICE_USER_WORKER,
240 worker: {
241 compatibilityDate: worker.compatibilityDate,
242 compatibilityFlags: worker.compatibilityFlags,
243 bindings,
244 modules: worker.modules.map(moduleToWorkerd),
245 durableObjectNamespaces:
246 worker.durableObjectNamespaces?.map((namespace) => {
247 const imageName = imageNames.get(namespace.className);
248 return {
249 className: namespace.className,
250 enableSql: namespace.sql,
251 uniqueKey:
252 namespace.uniqueKey ??
253 defaultDurableObjectUniqueKey(
254 worker.name,
255 namespace.className,
256 ),
257 ephemeralLocal: namespace.ephemeralLocal,
258 container: imageName ? { imageName } : undefined,
259 };
260 }),
261 durableObjectStorage: {
262 localDisk: storage.name,
263 },
264 containerEngine,
265 tails,
266 streamingTails,
267 ...config.userWorker,
268 ...worker.unsafe,
269 },
270 },
271 ...config.services,
272 ],
273 extensions: config.extensions,
274 },
275 {
276 "debug-port": `127.0.0.1:${pinned?.[SOCKET_DEBUG_PORT] ?? 0}`,
277 ...(worker.logging?.verbose ? { verbose: true } : undefined),
278 },
279 {
280 onOutput: worker.logging?.onOutput,
281 onExit: (exit) => {
282 Queue.offerUnsafe(exits, exit);
283 },
284 },
285 );
286 const parentScope = yield* Effect.scope;
287 let processScope = yield* Scope.fork(parentScope);
288 const startProcess = (pinned?: Workerd.WorkerdPorts) =>
289 serveWorker(pinned).pipe(
290 Effect.tap((ports) => context.start(ports)),
291 Scope.provide(processScope),
292 Effect.onExit((exit) =>
293 Exit.isFailure(exit)
294 ? Scope.close(processScope, exit)
295 : Effect.void,
296 ),
297 );
298 const ports = yield* startProcess();
299 // A process that dies after startup (V8 aborting on heap exhaustion
300 // is the common case) is replaced on the same ports. Without this
301 // every request fails until the whole dev session is restarted.
302 yield* Effect.gen(function* () {
303 let restartTimes: Array<number> = [];
304 while (true) {
305 const exit = yield* Queue.take(exits);
306 // Release the old process, output listeners and plugin tasks before
307 // starting another generation. Closing also detaches this child
308 // scope from the parent, so repeated crashes do not retain it.
309 yield* Scope.close(processScope, Exit.void);
310 const now = yield* Clock.currentTimeMillis;
311 restartTimes = restartTimes.filter((time) => now - time < 60_000);
312 if (restartTimes.length >= 3) {
313 yield* Effect.logError(
314 `The Workers runtime for "${worker.name}" stopped after 3 restarts within 60 seconds. Fix the crash and restart the dev session.`,
315 );
316 return;
317 }
318 restartTimes.push(now);
319 yield* Effect.sleep(250 * 2 ** (restartTimes.length - 1));
320 yield* Effect.logWarning(
321 `The Workers runtime for "${worker.name}" exited unexpectedly (exit code ${exit.exitCode}, signal ${exit.signal}); starting a replacement on the same ports.${exit.stderr ? `\n${exit.stderr}` : ""}`,
322 );
323 processScope = yield* Scope.fork(parentScope);
324 const restarted = yield* startProcess(ports).pipe(Effect.result);
325 if (Result.isFailure(restarted)) {
326 yield* Effect.logError(
327 `The Workers runtime for "${worker.name}" could not be restarted: ${restarted.failure.message}`,
328 );
329 return;
330 }
331 worker.onRestart?.(exit);
332 }
333 }).pipe(Effect.forkScoped);
334 return new URL(`http://127.0.0.1:${ports[SOCKET_USER_ENTRY]}`);
335 }),
336 });
337 }),
338);
339
340const SOCKET_DEBUG_PORT = "debug-port";
341
342const pinSocket =
343 (ports: Workerd.WorkerdPorts) =>
344 (socket: WorkerdConfig.Socket): WorkerdConfig.Socket => {
345 const port = socket.name === undefined ? undefined : ports[socket.name];
346 return port === undefined
347 ? socket
348 : { ...socket, address: `127.0.0.1:${port}` };
349 };