1import * as Context from "effect/Context";
2import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Layer from "effect/Layer";
6import * as MutableHashMap from "effect/MutableHashMap";
7import * as Path from "effect/Path";
8import * as Schedule from "effect/Schedule";
9import * as Semaphore from "effect/Semaphore";
10import type * as Scope from "effect/Scope";
11import * as Stream from "effect/Stream";
12import * as SubscriptionRef from "effect/SubscriptionRef";
13import * as NFS from "node:fs";
14import * as Paths from "../internal/Paths.ts";
15import * as System from "../internal/System.ts";
16import { SystemError } from "../RuntimeError.shared.ts";
17import {
18 resolvedTargetKey,
19 type RegistryEntry,
20 type ResolvedTargetMap,
21 type Subscriber,
22} from "./RegistryTypes.shared.ts";
23
24export class Registry extends Context.Service<
25 Registry,
26 {
27 /**
28 * Reads the registry and returns the resolved targets for the given subscribers.
29 */
30 readonly read: (
31 subscribers: ReadonlyArray<Subscriber>,
32 ) => Effect.Effect<ResolvedTargetMap>;
33 /**
34 * Subscribes to changes in the registry for the given subscribers.
35 * Returns a stream containing an updated `ResolvedTargetMap` whenever the registry changes.
36 */
37 readonly subscribe: (
38 subscribers: ReadonlyArray<Subscriber>,
39 ) => Stream.Stream<ResolvedTargetMap>;
40 /**
41 * Writes an entry to the registry.
42 * The entry is removed when the scope closes.
43 */
44 readonly write: (
45 entry: RegistryEntry,
46 ) => Effect.Effect<void, SystemError, Scope.Scope>;
47 }
48>()("cloudflare-runtime/registry/Registry") {}
49
50const STALE_AFTER_MS = 300_000;
51
52/**
53 * Effect's Node `FileSystem.stat` uses `{ bigint: true }` and then converts
54 * `ino`/`dev` with `Number()`. NTFS file indexes routinely exceed
55 * `Number.MAX_SAFE_INTEGER`, so `stat` fails with `BadArgument` on Windows.
56 * The registry then treated live files as missing, wiped the in-memory
57 * snapshot, and skipped reap. `node:fs.stat` without bigint still returns a
58 * `Date` for `mtime`, which is all we need for the staleness check.
59 */
60const readMtime = (entryPath: string) =>
61 Effect.callback<Date | undefined>((resume) => {
62 NFS.stat(entryPath, (error, stats) => {
63 resume(Effect.succeed(error ? undefined : stats.mtime));
64 });
65 });
66
67export const RegistryLive = Layer.effect(
68 Registry,
69 Effect.gen(function* () {
70 const fs = yield* FileSystem.FileSystem;
71 const path = yield* Path.Path;
72 const directory = yield* Paths.state("alchemy", "registry");
73
74 const isNonStale = (entryPath: string) =>
75 Effect.zip(readMtime(entryPath), DateTime.nowAsDate, {
76 concurrent: true,
77 }).pipe(
78 Effect.map(
79 ([mtime, now]) =>
80 !!mtime && mtime.getTime() > now.getTime() - STALE_AFTER_MS,
81 ),
82 );
83
84 const readEntry = (entry: string) => {
85 const entryPath = path.join(directory, entry);
86 return isNonStale(entryPath).pipe(
87 Effect.flatMap((valid) =>
88 valid
89 ? fs.readFileString(entryPath).pipe(
90 Effect.flatMap((content) =>
91 // Other processes (including older runtimes) may expose
92 // a partial write. A parse failure must leave refresh alive.
93 Effect.try(
94 () =>
95 [
96 decodeURIComponent(path.basename(entry, ".json")),
97 JSON.parse(content),
98 ] as const,
99 ),
100 ),
101 )
102 : fs.remove(entryPath).pipe(Effect.as(undefined)),
103 ),
104 Effect.orElseSucceed(() => undefined),
105 );
106 };
107
108 const readAll = fs.readDirectory(directory).pipe(
109 Effect.flatMap((entries) =>
110 Effect.forEach(
111 entries.filter((entry) => entry.endsWith(".json")),
112 readEntry,
113 {
114 concurrency: "unbounded",
115 },
116 ),
117 ),
118 Effect.map((entries) =>
119 MutableHashMap.make(...entries.filter((entry) => entry !== undefined)),
120 ),
121 );
122
123 // Seed the registry with the entries that already exist on disk, so that
124 // workers registered before this process started are resolvable
125 // immediately (the watcher only reports future changes).
126 const ref = yield* SubscriptionRef.make(yield* readAll);
127
128 // Serialize disk snapshots and writes so a poll/watch refresh that started
129 // before `write` cannot overwrite the eager in-memory update with a stale
130 // empty snapshot.
131 const updateLock = yield* Semaphore.make(1);
132 const refresh = readAll.pipe(
133 Effect.flatMap((newValue) => SubscriptionRef.set(ref, newValue)),
134 updateLock.withPermits(1),
135 );
136
137 // The `fileSystemSupportsWatcher` flag is set to false on Windows and true everywhere else.
138 // The flag can be overridden using a ConfigProvider, e.g. for testing.
139 // If the watcher is not supported, we fall back to polling every 100ms.
140 yield* (
141 (yield* System.fileSystemSupportsWatcher)
142 ? fs.watch(directory).pipe(
143 Stream.map(() => undefined),
144 // Trigger one more read once the watcher is running, to cover
145 // changes made between the initial snapshot above and the watcher
146 // subscription. `mapEffect` runs the reads sequentially, so a
147 // watcher-triggered read cannot be overwritten by an older one.
148 Stream.merge(Stream.succeed(undefined)),
149 Stream.mapEffect(() => refresh),
150 )
151 : Stream.fromEffect(refresh).pipe(Stream.repeat(Schedule.spaced(100)))
152 ).pipe(Stream.runDrain, Effect.forkScoped);
153
154 return Registry.of({
155 read: (subscribers) =>
156 SubscriptionRef.get(ref).pipe(
157 Effect.map(pickSubscriberServices(subscribers)),
158 ),
159 subscribe: (subscribers) =>
160 SubscriptionRef.changes(ref).pipe(
161 Stream.map(pickSubscriberServices(subscribers)),
162 Stream.changes,
163 ),
164 write: (entry) => {
165 const entryPath = path.join(
166 directory,
167 `${encodeURIComponent(entry.scriptName)}.json`,
168 );
169 const serialized = JSON.stringify(entry, null, 2);
170 // Publish complete files atomically so other registry processes never
171 // observe a truncated entry during creation or replacement.
172 return Effect.scoped(
173 Effect.gen(function* () {
174 const temporaryPath = yield* fs.makeTempFileScoped({
175 directory,
176 prefix: ".registry-",
177 suffix: ".tmp",
178 });
179 yield* fs.writeFileString(temporaryPath, serialized);
180 yield* fs.rename(temporaryPath, entryPath);
181 }),
182 ).pipe(
183 Effect.andThen(
184 // Immediately update the in-memory registry so it's available without waiting on IO.
185 SubscriptionRef.update(ref, (map) =>
186 MutableHashMap.set(map, entry.scriptName, entry),
187 ),
188 ),
189 updateLock.withPermits(1),
190 Effect.tap(() => {
191 // Remove the entry from the filesystem when the scope closes — but
192 // only while the file still holds THIS write's content. A
193 // replacement instance of the same script re-registers under the
194 // same key; a graceful handoff closes the old scope after the new
195 // instance has already overwritten the file, and removing it here
196 // would unregister the live replacement.
197 return Effect.addFinalizer(() =>
198 fs.readFileString(entryPath).pipe(
199 Effect.flatMap((current) =>
200 current === serialized ? fs.remove(entryPath) : Effect.void,
201 ),
202 Effect.ignore,
203 ),
204 );
205 }),
206 Effect.tap(() =>
207 // Update the `mtime` every 30 seconds so the entry is not considered stale.
208 DateTime.nowAsDate.pipe(
209 Effect.flatMap((now) => fs.utimes(entryPath, now, now)),
210 Effect.schedule(Schedule.spaced("30 seconds")),
211 Effect.forkScoped,
212 ),
213 ),
214 Effect.mapError(
215 (error) =>
216 new SystemError({
217 subtag: "RegistryWriteError",
218 message: "Failed to write registry entry",
219 detail: {
220 entry,
221 },
222 cause: error,
223 }),
224 ),
225 );
226 },
227 });
228 }),
229);
230
231const pickSubscriberServices =
232 (subscribers: ReadonlyArray<Subscriber>) =>
233 (registry: MutableHashMap.MutableHashMap<string, RegistryEntry>) => {
234 const resolved: ResolvedTargetMap = {};
235 for (const subscriber of subscribers) {
236 for (const entry of MutableHashMap.values(registry)) {
237 const service = extractSubscriberService(subscriber, entry);
238 if (service) {
239 resolved[resolvedTargetKey(subscriber)] = {
240 ...service,
241 scriptName: entry.scriptName,
242 debugPortAddress: entry.debugPortAddress,
243 };
244 break;
245 }
246 }
247 }
248 return resolved;
249 };
250
251const extractSubscriberService = (
252 subscriber: Subscriber,
253 entry: RegistryEntry,
254) => {
255 switch (subscriber.kind) {
256 case "worker":
257 return entry.scriptName === subscriber.scriptName
258 ? entry.services[0]
259 : undefined;
260 case "durable-object":
261 return entry.scriptName === subscriber.scriptName
262 ? entry.services.find(
263 (service) =>
264 service.kind === "durable-object" &&
265 service.className === subscriber.className,
266 )
267 : undefined;
268 case "queue-consumer":
269 return entry.services.find(
270 (service) =>
271 service.kind === "queue-consumer" &&
272 service.queueName === subscriber.queueName,
273 );
274 case "workflow":
275 return entry.scriptName === subscriber.scriptName
276 ? entry.services.find(
277 (service) =>
278 service.kind === "workflow" &&
279 service.workflowName === subscriber.workflowName,
280 )
281 : undefined;
282 }
283};