1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Test from "@/Test/Alchemy";
3import { describe, expect } from "alchemy-test";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as HttpClientRequest from "effect/http/HttpClientRequest";
7import { requestWorker } from "../Utils/WorkerRequest.ts";
8import type {
9 MigrationProbe,
10 Snapshot,
11} from "./fixtures/alarm-upgrade/types.ts";
12import AlarmUpgradeWorker from "./fixtures/alarm-upgrade/v2.ts";
13
14class WorkerVersionPending extends Error {}
15
16const requestJson = Effect.fn(
17 function* (
18 url: string,
19 action = "snapshot",
20 workerVersion: Snapshot["version"] = "v2",
21 ) {
22 const fresh = yield* Effect.sync(() => {
23 const fresh = new URL(`${url}/${action}`);
24 fresh.searchParams.set("cb", String(Date.now()));
25 return fresh;
26 });
27 const response = yield* requestWorker(
28 (fresh.pathname === "/snapshot"
29 ? HttpClientRequest.get(fresh.href)
30 : HttpClientRequest.post(fresh.href)
31 ).pipe(
32 // A fresh connection avoids polling an edge still pinned to the old deployment.
33 HttpClientRequest.setHeaders({
34 connection: "close",
35 "cache-control": "no-cache",
36 "x-alarm-worker-version": workerVersion,
37 }),
38 ),
39 ).pipe(Effect.timeout("15 seconds"));
40 if (response.status !== 200) {
41 const body = yield* response.text;
42 const actualVersion = response.headers["x-alarm-worker-version"];
43 if (
44 response.status === 409 &&
45 actualVersion === (workerVersion === "v1" ? "v2" : "v1") &&
46 body === "Alarm worker version mismatch"
47 ) {
48 return yield* Effect.fail(
49 new WorkerVersionPending(
50 `Waiting for Worker ${workerVersion}; got ${actualVersion}`,
51 ),
52 );
53 }
54 return yield* Effect.fail(
55 new Error(
56 `Upgrade fixture ${action}: HTTP ${response.status}\n${body}`,
57 ),
58 );
59 }
60 const body: unknown = yield* response.json;
61 return body;
62 },
63 Effect.retry({
64 while: (error) => error instanceof WorkerVersionPending,
65 schedule: Schedule.spaced("1 second"),
66 times: 10,
67 }),
68);
69
70const request = (
71 url: string,
72 action = "snapshot",
73 workerVersion: Snapshot["version"] = "v2",
74) =>
75 requestJson(url, action, workerVersion).pipe(
76 Effect.map((body) => body as Snapshot),
77 );
78
79const ready = (url: string, version: Snapshot["version"], name?: string) =>
80 request(
81 url,
82 name === undefined
83 ? "snapshot"
84 : `snapshot?name=${encodeURIComponent(name)}`,
85 version,
86 ).pipe(
87 Effect.flatMap((snapshot) =>
88 snapshot.version === version
89 ? Effect.succeed(snapshot)
90 : Effect.fail(
91 new Error(`Waiting for ${version}; got ${snapshot.version}`),
92 ),
93 ),
94 Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 10 }),
95 );
96
97const delivered = (
98 url: string,
99 until: (snapshot: Snapshot) => boolean,
100 workerVersion: Snapshot["version"] = "v2",
101) =>
102 request(url, "snapshot", workerVersion).pipe(
103 Effect.repeat({ schedule: Schedule.spaced("1 second"), times: 10, until }),
104 Effect.tap((snapshot) =>
105 Effect.sync(() => expect(until(snapshot)).toBe(true)),
106 ),
107 );
108
109const count = (
110 snapshot: Snapshot,
111 channel: "legacy" | "callback",
112 id: string,
113) =>
114 snapshot.deliveries.filter(
115 (delivery) => delivery.channel === channel && delivery.id === id,
116 ).length;
117
118for (const dev of [true, false]) {
119 describe(
120 dev ? "local alarm upgrade" : "live alarm upgrade",
121 { tags: ["provider:cloudflare", "provider:cloudflare:worker"] },
122 () => {
123 const { test } = Test.make({ providers: Cloudflare.providers(), dev });
124
125 test.provider(
126 `${dev ? "local" : "live"} V1 to V2 deployment preserves legacy schema, jobs and object identity`,
127 (stack) =>
128 Effect.gen(function* () {
129 yield* stack.destroy();
130 const main = yield* Effect.sync(
131 () =>
132 new URL("./fixtures/alarm-upgrade/v1.ts", import.meta.url)
133 .pathname,
134 );
135 const originalDeployment = yield* stack.deploy(
136 Effect.gen(function* () {
137 const host = yield* Cloudflare.Worker("AlarmUpgradeWorker", {
138 main,
139 env: {
140 UpgradeObject: Cloudflare.DurableObject("UpgradeObject"),
141 },
142 });
143 const reader = dev
144 ? undefined
145 : yield* Cloudflare.Worker("AlarmUpgradeV1Reader", {
146 main,
147 env: {
148 UpgradeObject: Cloudflare.DurableObject(
149 "UpgradeObject",
150 {
151 scriptName: host.workerName,
152 },
153 ),
154 },
155 });
156 return { host, reader };
157 }),
158 );
159 const original = originalDeployment.host;
160 expect(original.url).toBeDefined();
161 expect(
162 original.durableObjectNamespaces.UpgradeObject,
163 ).toBeDefined();
164 if (dev) {
165 expect(original.url).toMatch(/^http:\/\/localhost:\d+$/);
166 } else {
167 expect(original.url).toMatch(/^https:\/\//);
168 }
169 const initial = yield* ready(original.url!, "v1");
170 expect(initial.marker).toBeNull();
171 yield* request(original.url!, "seed", "v1");
172 yield* ready(original.url!, "v1", "future-version");
173 const futureOriginal = yield* request(
174 original.url!,
175 "seed?name=future-version",
176 "v1",
177 );
178 yield* ready(original.url!, "v1", "atomic-migration");
179 const rollbackOriginal = yield* request(
180 original.url!,
181 "seed?name=atomic-migration",
182 "v1",
183 );
184 for (const seeded of [futureOriginal, rollbackOriginal]) {
185 expect(seeded.tables).toEqual(["alchemy_scheduled_events"]);
186 expect(seeded.schemaVersion).toBeNull();
187 expect(seeded.schemaRows).toEqual([]);
188 expect(seeded.legacyRows).toHaveLength(3);
189 }
190 const before = yield* delivered(
191 original.url!,
192 (snapshot) =>
193 count(snapshot, "legacy", "v1-proof") === 1 &&
194 snapshot.legacyRows.length === 3 &&
195 snapshot.alarm ===
196 Math.min(...snapshot.legacyRows.map((row) => row.run_at)),
197 "v1",
198 );
199 expect(before.version).toBe("v1");
200 expect(before.marker).toBe("written-by-v1");
201 expect(before.tables).toEqual(["alchemy_scheduled_events"]);
202 expect(before.schemaVersion).toBeNull();
203 expect(before.schemaRows).toEqual([]);
204 expect(before.legacyRows.map((row) => row.id)).toEqual([
205 "legacy-cancel",
206 "legacy-one",
207 "legacy-repeat",
208 ]);
209 expect(before.legacyColumns).toEqual([
210 { name: "id", type: "TEXT", notnull: 0, pk: 1 },
211 { name: "run_at", type: "INTEGER", notnull: 1, pk: 0 },
212 { name: "repeat_ms", type: "INTEGER", notnull: 0, pk: 0 },
213 { name: "payload", type: "TEXT", notnull: 1, pk: 0 },
214 ]);
215 expect(before.alarm).toBe(
216 Math.min(...before.legacyRows.map((row) => row.run_at)),
217 );
218 expect(before.deliveries).toEqual([
219 {
220 version: "v1",
221 channel: "legacy",
222 id: "v1-proof",
223 payload: { callback: "archive", value: "v1-proof" },
224 },
225 ]);
226
227 // This is an update of the existing resource, not a fresh namespace or a schema-only fixture.
228 const upgradedDeployment = yield* stack.deploy(
229 Effect.gen(function* () {
230 const host = yield* AlarmUpgradeWorker;
231 const reader = dev
232 ? undefined
233 : yield* Cloudflare.Worker("AlarmUpgradeV1Reader", {
234 main,
235 env: {
236 UpgradeObject: Cloudflare.DurableObject(
237 "UpgradeObject",
238 {
239 scriptName: host.workerName,
240 },
241 ),
242 },
243 });
244 return { host, reader };
245 }),
246 );
247 const upgraded = upgradedDeployment.host;
248 expect(upgraded.workerName).toBe(original.workerName);
249 expect(upgraded.workerId).toBe(original.workerId);
250 expect(upgraded.durableObjectNamespaces.UpgradeObject).toBe(
251 original.durableObjectNamespaces.UpgradeObject,
252 );
253 const after = yield* ready(upgraded.url!, "v2");
254 expect(after.id).toBe(before.id);
255 expect(after.marker).toBe(before.marker);
256 expect(after.constructors).toContain("v1");
257 expect(after.constructors).toContain("v2");
258 expect(after.legacyColumns).toEqual(before.legacyColumns);
259 expect(after.legacyRows).toEqual(before.legacyRows);
260 expect(after.deliveries).toEqual(before.deliveries);
261 expect(after.alarm).toBe(before.alarm);
262 expect(after.tables).toContain("alchemy_scheduled_events");
263 expect(after.tables).toContain("alchemy_alarm_callbacks");
264 expect(after.tables).toContain("alchemy_alarm_schema");
265 expect(after.schemaVersion).toBe(1);
266 expect(after.schemaRows).toEqual([{ id: 1, version: 1 }]);
267 expect(after.callbacks).toEqual([]);
268 if (upgradedDeployment.reader) {
269 // Local restarts are atomic; live edges can still run V1 against the V2 object.
270 const viaV1 = yield* request(
271 upgradedDeployment.reader.url!,
272 "snapshot",
273 "v1",
274 );
275 expect(viaV1).toEqual(after);
276 const rejected = yield* requestWorker(
277 HttpClientRequest.post(
278 `${upgradedDeployment.reader.url!}/callback-first`,
279 ).pipe(
280 HttpClientRequest.setHeader("x-alarm-worker-version", "v2"),
281 ),
282 );
283 expect(rejected.status).toBe(409);
284 expect(rejected.headers["x-alarm-worker-version"]).toBe("v1");
285 expect(yield* rejected.text).toBe(
286 "Alarm worker version mismatch",
287 );
288 expect(yield* request(upgraded.url!)).toEqual(after);
289 }
290
291 const callbackFirst = yield* request(
292 upgraded.url!,
293 "callback-first",
294 );
295 expect(callbackFirst.callbacks).toHaveLength(1);
296 expect(callbackFirst.alarm).toBe(callbackFirst.callbacks[0].run_at);
297 expect(callbackFirst.alarm!).toBeLessThan(before.alarm!);
298 const reset = (yield* requestJson(
299 upgraded.url!,
300 "reconstruct",
301 )) as {
302 aborted: boolean;
303 };
304 expect(reset.aborted).toBe(true);
305 const reconstructed = yield* ready(upgraded.url!, "v2");
306 expect(reconstructed.constructors.length).toBeGreaterThan(
307 callbackFirst.constructors.length,
308 );
309 expect(reconstructed.id).toBe(before.id);
310 expect(reconstructed.marker).toBe(before.marker);
311 expect(reconstructed.schemaVersion).toBe(1);
312 expect(reconstructed.schemaRows).toEqual([{ id: 1, version: 1 }]);
313 expect(reconstructed.schema).toEqual(callbackFirst.schema);
314 expect(reconstructed.legacyRows).toEqual(callbackFirst.legacyRows);
315 expect(reconstructed.callbacks).toEqual(callbackFirst.callbacks);
316 expect(reconstructed.alarm).toBe(callbackFirst.alarm);
317 const repeated = yield* request(upgraded.url!);
318 expect(repeated.schemaVersion).toBe(1);
319 expect(repeated.schemaRows).toEqual(reconstructed.schemaRows);
320 expect(repeated.schema).toEqual(reconstructed.schema);
321 expect(repeated.legacyRows).toEqual(reconstructed.legacyRows);
322 expect(repeated.callbacks).toEqual(reconstructed.callbacks);
323 expect(repeated.alarm).toBe(reconstructed.alarm);
324 const legacyCanceled = yield* request(
325 upgraded.url!,
326 "cancel-legacy",
327 );
328 expect(legacyCanceled.legacyRows.map((row) => row.id)).toEqual([
329 "legacy-one",
330 "legacy-repeat",
331 ]);
332 expect(legacyCanceled.callbacks).toEqual(callbackFirst.callbacks);
333 expect(legacyCanceled.alarm).toBe(callbackFirst.alarm);
334 const callbackCanceled = yield* request(
335 upgraded.url!,
336 "cancel-callback",
337 );
338 expect(callbackCanceled.callbacks).toEqual([]);
339 expect(callbackCanceled.legacyRows).toEqual(
340 legacyCanceled.legacyRows,
341 );
342 expect(callbackCanceled.alarm).toBe(before.alarm);
343
344 // Pending jobs start five minutes ahead so deployment latency cannot consume them in V1.
345 // Moving only their due times forward keeps the native-alarm assertion bounded.
346 const released = yield* request(upgraded.url!, "release");
347 expect(released.callbacks).toHaveLength(1);
348 expect(released.alarm).toBe(released.callbacks[0].run_at);
349 for (const row of released.legacyRows) {
350 const old = before.legacyRows.find(
351 (event) => event.id === row.id,
352 )!;
353 expect(row.payload).toBe(old.payload);
354 expect(row.repeat_ms).toBe(old.repeat_ms);
355 expect(row.run_at).toBe(released.callbacks[0].run_at);
356 }
357 const fired = yield* delivered(
358 upgraded.url!,
359 (snapshot) =>
360 count(snapshot, "legacy", "legacy-one") === 1 &&
361 count(snapshot, "legacy", "legacy-repeat") >= 2 &&
362 count(snapshot, "callback", "new-callback") === 1 &&
363 snapshot.callbacks.length === 0 &&
364 snapshot.legacyRows.length === 1 &&
365 snapshot.alarm === snapshot.legacyRows[0].run_at,
366 );
367 expect(fired.legacyRows).toHaveLength(1);
368 expect(fired.legacyRows[0].id).toBe("legacy-repeat");
369 expect(fired.legacyRows[0].repeat_ms).toBe(2_000);
370 expect(fired.legacyRows[0].run_at).toBeGreaterThan(
371 released.legacyRows.find((row) => row.id === "legacy-repeat")!
372 .run_at,
373 );
374 expect(fired.callbacks).toEqual([]);
375 expect(fired.alarm).toBe(fired.legacyRows[0].run_at);
376 for (const delivery of fired.deliveries.filter(
377 (event) => event.id !== "v1-proof",
378 )) {
379 expect(delivery.version).toBe("v2");
380 if (delivery.channel === "legacy") {
381 expect(delivery.payload).toEqual({
382 callback: "archive",
383 value: delivery.id,
384 });
385 }
386 }
387 expect(count(fired, "legacy", "legacy-cancel")).toBe(0);
388 expect(
389 fired.deliveries.filter((event) => event.channel === "callback"),
390 ).toEqual([
391 {
392 version: "v2",
393 channel: "callback",
394 id: "new-callback",
395 payload: { value: "new-callback" },
396 },
397 ]);
398
399 const onlyCallback = yield* request(upgraded.url!, "only-callback");
400 expect(onlyCallback.legacyRows).toEqual([]);
401 expect(onlyCallback.callbacks).toHaveLength(1);
402 expect(onlyCallback.alarm).toBe(onlyCallback.callbacks[0].run_at);
403 const callbackSurvived = yield* delivered(
404 upgraded.url!,
405 (snapshot) =>
406 count(snapshot, "callback", "callback-survivor") === 1 &&
407 snapshot.callbacks.length === 0 &&
408 snapshot.alarm === null,
409 );
410 expect(callbackSurvived.callbacks).toEqual([]);
411 expect(callbackSurvived.alarm).toBeNull();
412
413 const onlyLegacy = yield* request(upgraded.url!, "only-legacy");
414 expect(onlyLegacy.callbacks).toEqual([]);
415 expect(onlyLegacy.legacyRows).toHaveLength(1);
416 expect(onlyLegacy.alarm).toBe(onlyLegacy.legacyRows[0].run_at);
417 const legacySurvived = yield* delivered(
418 upgraded.url!,
419 (snapshot) =>
420 count(snapshot, "legacy", "legacy-survivor") === 1 &&
421 snapshot.legacyRows.length === 0 &&
422 snapshot.alarm === null,
423 );
424 expect(count(legacySurvived, "callback", "must-not-fire")).toBe(0);
425 expect(count(legacySurvived, "callback", "canceled-callback")).toBe(
426 0,
427 );
428 expect(count(legacySurvived, "legacy", "legacy-cancel")).toBe(0);
429 expect(legacySurvived.legacyRows).toEqual([]);
430 expect(legacySurvived.callbacks).toEqual([]);
431 expect(legacySurvived.alarm).toBeNull();
432 expect(legacySurvived.id).toBe(before.id);
433 expect(legacySurvived.marker).toBe("written-by-v1");
434
435 const future = (yield* requestJson(
436 upgraded.url!,
437 "migration-future?name=future-version",
438 )) as MigrationProbe;
439 expect(future.before.id).toBe(futureOriginal.id);
440 expect(future.before.legacyRows).toEqual(futureOriginal.legacyRows);
441 expect(future.before.schemaVersion).toBe(2);
442 expect(future.before.schemaRows).toEqual([{ id: 1, version: 2 }]);
443 expect(future.before.callbacks).toHaveLength(1);
444 expect(future.before.alarm).toBe(futureOriginal.alarm);
445 expect(future.failure?.tag).toBe("UnsupportedAlarmSchemaVersion");
446 expect(future.failure?.version).toBe(2);
447 expect(future.failure?.supportedVersion).toBe(1);
448 expect(future.after).toEqual(future.before);
449 expect(future.recovered).toBeNull();
450
451 const rollback = (yield* requestJson(
452 upgraded.url!,
453 "migration-rollback?name=atomic-migration",
454 )) as MigrationProbe;
455 expect(rollback.before.id).toBe(rollbackOriginal.id);
456 expect(rollback.before.legacyRows).toEqual(
457 rollbackOriginal.legacyRows,
458 );
459 expect(rollback.before.schemaVersion).toBeNull();
460 expect(rollback.before.schemaRows).toEqual([]);
461 expect(rollback.before.tables).not.toContain(
462 "alchemy_alarm_callbacks",
463 );
464 expect(rollback.before.tables).not.toContain(
465 "alchemy_alarm_schema",
466 );
467 expect(
468 rollback.before.schema.some(
469 (row) => row.name === "idx_alchemy_scheduled_events_run_at",
470 ),
471 ).toBe(false);
472 expect(rollback.failure).not.toBeNull();
473 expect(rollback.failure!.message).toContain(
474 "idx_alchemy_alarm_callbacks_run_at",
475 );
476 expect(rollback.after).toEqual(rollback.before);
477 expect(rollback.after.alarm).toBe(rollbackOriginal.alarm);
478 expect(rollback.retryBefore?.schemaVersion).toBe(0);
479 expect(rollback.retryBefore?.schemaRows).toEqual([
480 { id: 1, version: 0 },
481 ]);
482 expect(rollback.retryBefore?.legacyRows).toEqual(
483 rollbackOriginal.legacyRows,
484 );
485 expect(rollback.recovered?.schemaVersion).toBe(1);
486 expect(rollback.recovered?.schemaRows).toEqual([
487 { id: 1, version: 1 },
488 ]);
489 expect(rollback.recovered?.legacyRows).toEqual(
490 rollbackOriginal.legacyRows,
491 );
492 expect(rollback.recovered?.callbacks).toEqual([]);
493 expect(rollback.recovered?.tables).toContain(
494 "alchemy_alarm_callbacks",
495 );
496 expect(
497 rollback.recovered?.schema.some(
498 (row) =>
499 row.name === "idx_alchemy_scheduled_events_run_at" &&
500 row.type === "index",
501 ),
502 ).toBe(true);
503 expect(rollback.recovered?.alarm).toBe(rollbackOriginal.alarm);
504 yield* stack.destroy();
505 }),
506 { tags: [...(dev ? ["local"] : ["live"])], timeout: 120_000 },
507 );
508 },
509 );
510}