1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { describe, expect } from "alchemy-test";
4import * as Cause from "effect/Cause";
5import type * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import type {
11 AlarmObservationResult,
12 BatchResult,
13 ExplicitRollbackResult,
14 FailedBatchResult,
15 RegistrationResult,
16 RollbackResult,
17 SiblingTransactionResult,
18 Snapshot,
19} from "./fixtures/alarm-callback/object.ts";
20import makeStack from "./fixtures/alarm-callback/stack.ts";
21import { inMemoryState } from "@/State";
22
23const requestJson = <T>(
24 url: string,
25 method: "GET" | "POST",
26 retryDelay: Duration.Input = "2 seconds",
27) =>
28 Effect.gen(function* () {
29 const client = HttpClient.mapRequest(
30 yield* HttpClient.HttpClient,
31 HttpClientRequest.setHeaders({
32 connection: "close",
33 "cache-control": "no-cache",
34 }),
35 );
36 const fresh = new URL(url);
37 fresh.searchParams.set("cb", yield* Effect.sync(() => String(Date.now())));
38 const response = yield* method === "POST"
39 ? client.post(fresh.href)
40 : client.get(fresh.href);
41 const body = yield* response.text;
42 if (response.status !== 200) {
43 const message = `${method} ${url}: HTTP ${response.status}: ${body}`;
44 if (
45 response.status === 404 ||
46 (response.status >= 500 &&
47 (method === "GET" || body.includes("<title>Script not found |")))
48 ) {
49 return yield* Effect.fail(
50 Object.assign(new Test.WorkerNotReady({ status: response.status }), {
51 message,
52 body,
53 url,
54 }),
55 );
56 }
57 return yield* Effect.fail(new Error(message));
58 }
59 return yield* Effect.try({
60 try: () => JSON.parse(body) as T,
61 catch: () => new Error(`${method} ${url}: Invalid JSON: ${body}`),
62 });
63 }).pipe(
64 Effect.timeout("5 seconds"),
65 Effect.retry({
66 while: (error) => error instanceof Test.WorkerNotReady,
67 schedule: Schedule.spaced(retryDelay),
68 times: 8,
69 }),
70 );
71
72const json = <T>(url: string, method: "GET" | "POST" = "GET") =>
73 Effect.gen(function* () {
74 if (method === "POST") {
75 const snapshot = new URL(url);
76 snapshot.pathname = snapshot.pathname.replace(
77 /\/[^/]+$/,
78 snapshot.pathname.endsWith("/legacy") ? "/legacy" : "/snapshot",
79 );
80 snapshot.search = "";
81 yield* requestJson(snapshot.href, "GET");
82 }
83 return yield* requestJson<T>(url, method);
84 });
85
86const poll = <T>(
87 url: string,
88 until: (value: T) => boolean,
89 interval: "2 seconds" | "4 seconds" = "2 seconds",
90) =>
91 json<T>(url).pipe(
92 Effect.repeat({
93 schedule: Schedule.spaced(interval),
94 until,
95 times: 10,
96 }),
97 Effect.flatMap((snapshot) =>
98 until(snapshot)
99 ? Effect.succeed(snapshot)
100 : Effect.fail(
101 new Error(
102 `Alarm polling exhausted for ${url}: ${JSON.stringify(snapshot)}`,
103 ),
104 ),
105 ),
106 );
107
108const deliveries = (snapshot: Snapshot) =>
109 snapshot.deliveries
110 .map(({ callback, value }) => `${callback}:${value}`)
111 .sort();
112
113const drained = (count: number) => (snapshot: Snapshot) =>
114 snapshot.deliveries.length === count && snapshot.alarm === null;
115
116const assertRollback = (snapshot: Snapshot) => {
117 expect(snapshot.application).toBeNull();
118 expect(snapshot.cleanupWrite).toBeNull();
119 expect(snapshot.rows).toEqual([]);
120};
121
122describe.concurrent.each([
123 { dev: true, stage: "alarm-callback-local" },
124 { dev: false, stage: "alarm-callback-live" },
125])(
126 "makeCallback (dev: $dev)",
127 ({ dev, stage }) => {
128 const state = dev ? inMemoryState() : Cloudflare.state();
129 const Stack = makeStack(state);
130 const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
131 providers: Cloudflare.providers(),
132 state,
133 dev,
134 stage,
135 });
136
137 const stack = beforeAll(
138 Effect.gen(function* () {
139 yield* destroy(Stack);
140 const output = yield* deploy(Stack);
141 yield* Effect.all([
142 requestJson<Snapshot>(
143 `${output.url}/readiness/snapshot`,
144 "GET",
145 "4 seconds",
146 ),
147 requestJson(`${output.url}/readiness/legacy`, "GET", "4 seconds"),
148 ]).pipe(
149 Effect.retry({
150 while: Cause.isTimeoutError,
151 schedule: Schedule.spaced("1 second"),
152 times: 3,
153 }),
154 );
155 expect(output.url.startsWith("http://localhost:")).toBe(dev);
156 return output;
157 }),
158 { timeout: 120_000 },
159 );
160 afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
161 timeout: 30_000,
162 });
163
164 test(
165 "Worker callback registration stays unsupported after a Durable Object call",
166 Effect.gen(function* () {
167 const { url } = yield* stack;
168 const result = yield* json<
169 RegistrationResult & { runtimeType: string }
170 >(`${url}/worker-registration/worker-registration`, "POST");
171 expect(result.snapshot.boots).toBeGreaterThan(0);
172 expect(result.runtimeType).toBe("Cloudflare.Worker");
173 expect(result.failure).toEqual({
174 tag: "CallbackError",
175 callback: "unsupported-worker",
176 message: `Durable callbacks are not supported by ${result.runtimeType}`,
177 });
178 expect(result.snapshot.pendingJobs).toEqual([]);
179 expect(result.snapshot.alarm).toBeNull();
180 }),
181 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
182 );
183
184 test(
185 "late DO registration defects without disabling initialized callbacks",
186 Effect.gen(function* () {
187 const { url } = yield* stack;
188 const base = `${url}/late-registration`;
189 const rejected = yield* json<RegistrationResult>(
190 `${base}/late-registration`,
191 "POST",
192 );
193 expect(rejected.failure?.tag).toBe("CallbackError");
194 expect(rejected.failure?.callback).toBe("late");
195 expect(rejected.failure?.message).toContain("instance initialization");
196 expect(rejected.snapshot.pendingJobs).toEqual([]);
197 expect(rejected.snapshot.deliveries).toEqual([]);
198 expect(rejected.snapshot.alarm).toBeNull();
199 const scheduled = yield* json<Snapshot>(`${base}/timing`, "POST");
200 expect(scheduled.id).toBe(rejected.snapshot.id);
201 expect(scheduled.alarm).not.toBeNull();
202 const delivered = yield* poll<Snapshot>(`${base}/snapshot`, drained(5));
203 expect(delivered.id).toBe(rejected.snapshot.id);
204 expect(deliveries(delivered)).toEqual([
205 "archive:checkpoint",
206 "archive:date",
207 "archive:duration",
208 "archive:latest",
209 "secondary:other-callback",
210 ]);
211 expect(delivered.pendingJobs).toEqual([]);
212 }),
213 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
214 );
215
216 test(
217 "batches native alarm bookkeeping across nested callbacks, legacy mutations, and finalizers",
218 Effect.gen(function* () {
219 const { url } = yield* stack;
220 const base = `${url}/bookkeeping`;
221 const result = yield* json<BatchResult>(`${base}/bookkeeping`, "POST");
222 yield* Effect.logInfo("Native alarm bookkeeping", result.counts);
223 expect(result.counts).toEqual({
224 schemaChecks: 1,
225 reconciliations: 1,
226 setAlarm: 1,
227 deleteAlarm: 0,
228 });
229 expect(result.snapshot.pendingJobs).toHaveLength(3);
230 expect(result.snapshot.alarm).not.toBeNull();
231 const delivered = yield* poll<Snapshot>(`${base}/snapshot`, drained(3));
232 expect(deliveries(delivered)).toEqual([
233 "archive:batched",
234 "archive:finalizer",
235 "secondary:nested",
236 ]);
237 }),
238 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
239 );
240
241 test(
242 "batches native alarm deletion for repeated cancellation",
243 Effect.gen(function* () {
244 const { url } = yield* stack;
245 const result = yield* json<BatchResult>(
246 `${url}/cancel-bookkeeping/cancel-bookkeeping`,
247 "POST",
248 );
249 yield* Effect.logInfo("Native cancellation bookkeeping", result.counts);
250 expect(result.counts).toEqual({
251 schemaChecks: 1,
252 reconciliations: 1,
253 setAlarm: 0,
254 deleteAlarm: 1,
255 });
256 expect(result.snapshot.pendingJobs).toEqual([]);
257 expect(result.snapshot.alarm).toBeNull();
258 }),
259 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
260 );
261
262 for (const explicit of [true, false]) {
263 test(
264 `${explicit ? "explicit rollback" : "native reconciliation failure"} discards deferred bookkeeping and retries schema initialization`,
265 Effect.gen(function* () {
266 const { url } = yield* stack;
267 const operation = explicit
268 ? "bookkeeping-rollback"
269 : "bookkeeping-failure";
270 const base = `${url}/${operation}`;
271 const result = yield* json<FailedBatchResult>(
272 `${base}/${operation}`,
273 "POST",
274 );
275 if (explicit) expect(result.failure).toBeNull();
276 else
277 expect(result.failure).toContain(
278 "no such table: alchemy_alarm_callbacks",
279 );
280 expect(result.rolledBack.counts).toEqual({
281 schemaChecks: 1,
282 reconciliations: explicit ? 0 : 1,
283 setAlarm: 0,
284 deleteAlarm: 0,
285 });
286 assertRollback(result.rolledBack.snapshot);
287 expect(result.rolledBack.snapshot.pendingJobs).toEqual([]);
288 expect(result.rolledBack.snapshot.alarm).toBeNull();
289 expect(result.recovered.counts).toEqual({
290 schemaChecks: 1,
291 reconciliations: 1,
292 setAlarm: 1,
293 deleteAlarm: 0,
294 });
295 expect(result.recovered.snapshot.pendingJobs).toHaveLength(1);
296 expect(result.recovered.snapshot.alarm).not.toBeNull();
297 const delivered = yield* poll<Snapshot>(
298 `${base}/snapshot`,
299 drained(1),
300 );
301 expect(deliveries(delivered)).toEqual(["archive:recovered"]);
302 assertRollback(delivered);
303 }),
304 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
305 );
306 }
307
308 test(
309 "preserves getAlarm observations and explicit native alarm write ordering in transactions",
310 Effect.gen(function* () {
311 const { url } = yield* stack;
312 const { at, observations, committed } =
313 yield* json<AlarmObservationResult>(
314 `${url}/alarm-observations/alarm-observations`,
315 "POST",
316 );
317 expect(observations).toEqual([
318 at,
319 null,
320 at + 1_000,
321 at + 2_000,
322 null,
323 null,
324 ]);
325 expect(committed).toBe(at + 3_000);
326 }),
327 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
328 );
329
330 test(
331 "native alarms deliver typed payloads, overwrite by callback and ID, and cancel",
332 Effect.gen(function* () {
333 const { url } = yield* stack;
334 const initial = yield* json<Snapshot>(`${url}/timing/timing`, "POST");
335 expect(initial.alarm).not.toBeNull();
336 const result = yield* poll<Snapshot>(
337 `${url}/timing/snapshot`,
338 drained(5),
339 );
340 expect(deliveries(result)).toEqual([
341 "archive:checkpoint",
342 "archive:date",
343 "archive:duration",
344 "archive:latest",
345 "secondary:other-callback",
346 ]);
347 expect(result.alarm).toBeNull();
348 }),
349 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
350 );
351
352 test(
353 "commits application SQL, KV, and a scheduled callback in one transaction",
354 Effect.gen(function* () {
355 const { url } = yield* stack;
356 const initial = yield* json<{
357 marker: string;
358 legacyOverload: string;
359 snapshot: Snapshot;
360 }>(`${url}/atomic/atomic`, "POST");
361 expect(initial.marker).toBe("committed");
362 expect(initial.legacyOverload).toBe("callback");
363 expect(initial.snapshot.application).toBe("committed");
364 expect(initial.snapshot.rows).toEqual([{ value: "committed" }]);
365 expect(initial.snapshot.alarm).not.toBeNull();
366 const result = yield* poll<Snapshot>(
367 `${url}/atomic/snapshot`,
368 drained(1),
369 );
370 expect(result.deliveries).toEqual([
371 {
372 callback: "archive",
373 value: "committed",
374 application: "committed",
375 boots: result.boots,
376 },
377 ]);
378 expect(result.alarm).toBeNull();
379 }),
380 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
381 );
382
383 test(
384 "delivers a callback registered during transactional initialization",
385 Effect.gen(function* () {
386 const { url } = yield* stack;
387 const base = `${url}/transactional-registration`;
388 const initial = yield* json<Snapshot>(
389 `${base}/transactional-registration`,
390 "POST",
391 );
392 expect(initial.pendingJobs).toHaveLength(1);
393 expect(initial.pendingJobs[0]?.callback).toBe("transactional-init");
394 expect(initial.alarm).not.toBeNull();
395 const result = yield* poll<Snapshot>(`${base}/snapshot`, drained(1));
396 expect(result.id).toBe(initial.id);
397 expect(result.application).toBe("registered-in-transaction");
398 expect(result.deliveries).toEqual([
399 {
400 callback: "transactional-init",
401 value: "registered-in-transaction",
402 application: "registered-in-transaction",
403 boots: result.boots,
404 },
405 ]);
406 expect(result.pendingJobs).toEqual([]);
407 expect(result.alarm).toBeNull();
408 }),
409 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
410 );
411
412 test(
413 "rejects a pre-existing sibling fiber write during another fiber's transaction",
414 Effect.gen(function* () {
415 const { url } = yield* stack;
416 const result = yield* json<SiblingTransactionResult>(
417 `${url}/sibling-transaction/sibling-transaction`,
418 "POST",
419 );
420 expect(result.failure).toBe("sibling-rollback");
421 expect(result.siblingAcknowledged).toBe(false);
422 expect(result.siblingFailure?.tag).toBe("DurableObjectStorageError");
423 expect(result.siblingValue).toBeNull();
424 assertRollback(result.snapshot);
425 expect(result.snapshot.pendingJobs).toEqual([]);
426 expect(result.snapshot.alarm).toBeNull();
427 }),
428 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
429 );
430
431 test(
432 "explicit rollback rejects later storage writes and callback scheduling while repeated rollback succeeds",
433 Effect.gen(function* () {
434 const { url } = yield* stack;
435 const base = `${url}/rollback-explicit`;
436 const result = yield* json<ExplicitRollbackResult>(
437 `${base}/rollback-explicit`,
438 "POST",
439 );
440 expect(result.repeatedRollbackSucceeded).toBe(true);
441 expect(result.operations.map(({ operation }) => operation)).toEqual([
442 "put",
443 "sql",
444 "schedule",
445 ]);
446 for (const operation of result.operations) {
447 expect(operation.acknowledged).toBe(false);
448 expect(operation.failure?.tag).toBe("DurableObjectStorageError");
449 if (operation.operation !== "schedule") {
450 expect(operation.failure?.wrappedBy).toBeNull();
451 }
452 }
453 assertRollback(result.snapshot);
454 expect(result.snapshot.deliveries).toEqual([]);
455 expect(result.snapshot.pendingJobs).toEqual([]);
456 expect(result.snapshot.alarm).toBeNull();
457 const persisted = yield* json<Snapshot>(`${base}/snapshot`);
458 assertRollback(persisted);
459 expect(persisted.deliveries).toEqual([]);
460 expect(persisted.pendingJobs).toEqual([]);
461 expect(persisted.alarm).toBeNull();
462 }),
463 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
464 );
465
466 for (const [kind, failure] of [
467 ["typed", "typed-value"],
468 ["defect", "defect"],
469 ["interrupt", "interrupted"],
470 ] as const) {
471 test(
472 `${kind} failure rolls back SQL, KV, scheduling, cancellation, and the native wake`,
473 Effect.gen(function* () {
474 const { url } = yield* stack;
475 const base = `${url}/rollback-${kind}`;
476 const result = yield* json<RollbackResult>(
477 `${base}/rollback-${kind}`,
478 "POST",
479 );
480 expect(result.failure).toBe(failure);
481 expect(result.cleanupWaited).toBe(true);
482 expect(result.alarmBefore).not.toBeNull();
483 expect(result.alarmAfter).toBe(result.alarmBefore);
484 assertRollback(result.snapshot);
485 const after = yield* poll<Snapshot>(`${base}/snapshot`, drained(2));
486 assertRollback(after);
487 expect(deliveries(after)).toEqual([
488 "archive:checkpoint",
489 "archive:kept",
490 ]);
491 expect(after.alarm).toBeNull();
492 }),
493 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
494 );
495 }
496
497 test(
498 "persists a recovery wake before a fallible handler and acknowledges only success",
499 Effect.gen(function* () {
500 const { url } = yield* stack;
501 yield* json(`${url}/retry/retry`, "POST");
502 const result = yield* poll<Snapshot>(
503 `${url}/retry/snapshot`,
504 drained(1),
505 );
506 expect(deliveries(result)).toEqual(["retry:retried"]);
507 expect(result.attempts).toHaveLength(2);
508 for (const attempt of result.attempts) {
509 expect(attempt.recovery).not.toBeNull();
510 expect(attempt.recovery!).toBeGreaterThan(attempt.now);
511 }
512 expect(result.alarm).toBeNull();
513 }),
514 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
515 );
516
517 test(
518 "acknowledging a callback does not delete its same-ID replacement",
519 Effect.gen(function* () {
520 const { url } = yield* stack;
521 yield* json(`${url}/replacement/replace`, "POST");
522 const result = yield* poll<Snapshot>(
523 `${url}/replacement/snapshot`,
524 drained(2),
525 );
526 expect(deliveries(result)).toEqual([
527 "archive:replace-first",
528 "archive:replace-second",
529 ]);
530 expect(result.alarm).toBeNull();
531 }),
532 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
533 );
534
535 test(
536 "reconstructs instance-local registrations after abort without mixing object IDs",
537 Effect.gen(function* () {
538 const { url } = yield* stack;
539 const [first, second] = yield* Effect.all(
540 [
541 json<Snapshot>(`${url}/reset-first/reset?value=first`, "POST"),
542 json<Snapshot>(`${url}/reset-second/reset?value=second`, "POST"),
543 ],
544 { concurrency: "unbounded" },
545 );
546 expect(first.id).not.toBe(second.id);
547 // Pending jobs must survive a delayed abort.
548 yield* Effect.sleep("5 seconds");
549 const aborted = yield* json<{ aborted: boolean }>(
550 `${url}/reset-first/abort`,
551 "POST",
552 );
553 expect(aborted.aborted).toBe(true);
554 const reconstructed = yield* json<Snapshot>(
555 `${url}/reset-first/snapshot`,
556 );
557 expect(reconstructed.boots).toBeGreaterThan(first.boots);
558 expect(reconstructed.pendingJobs).toEqual(first.pendingJobs);
559 expect(reconstructed.deliveries).toEqual([]);
560 yield* Effect.all(
561 [
562 json<Snapshot>(`${url}/reset-first/release-pending`, "POST"),
563 json<Snapshot>(`${url}/reset-second/release-pending`, "POST"),
564 ],
565 { concurrency: "unbounded" },
566 );
567 const [afterFirst, afterSecond] = yield* Effect.all(
568 [
569 poll<Snapshot>(`${url}/reset-first/snapshot`, drained(1)),
570 poll<Snapshot>(`${url}/reset-second/snapshot`, drained(1)),
571 ],
572 { concurrency: "unbounded" },
573 );
574 expect(afterFirst.id).toBe(first.id);
575 expect(afterFirst.boots).toBeGreaterThan(first.boots);
576 expect(afterFirst.deliveries[0]?.boots).toBe(afterFirst.boots);
577 expect(deliveries(afterFirst)).toEqual(["archive:first"]);
578 expect(deliveries(afterSecond)).toEqual(["archive:second"]);
579 expect(afterFirst.alarm).toBeNull();
580 expect(afterSecond.alarm).toBeNull();
581 }),
582 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
583 );
584
585 test(
586 "recovers a callback after the handler aborts with native alarm retries enabled",
587 Effect.gen(function* () {
588 const { url } = yield* stack;
589 yield* json(`${url}/recovery/recovery`, "POST");
590 const result = yield* poll<Snapshot>(
591 `${url}/recovery/snapshot`,
592 drained(1),
593 "4 seconds",
594 );
595 yield* Effect.logInfo("Alarm crash recovery snapshot", result);
596 expect(deliveries(result)).toEqual(["crash:recovered"]);
597 expect(result.pendingJobs).toEqual([]);
598 expect(result.attempts).toHaveLength(2);
599 expect(result.attempts[1]!.boots).toBeGreaterThan(
600 result.attempts[0]!.boots,
601 );
602 for (const attempt of result.attempts) {
603 expect(attempt.recovery).not.toBeNull();
604 expect(attempt.recovery!).toBeGreaterThan(attempt.now);
605 }
606 expect(result.alarm).toBeNull();
607 }),
608 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
609 );
610
611 test(
612 "retains an aborted callback with native retries disabled and recovers after an explicit wake",
613 Effect.gen(function* () {
614 const { url } = yield* stack;
615 const base = `${url}/recovery-no-retry`;
616 yield* json(`${base}/recovery-no-retry`, "POST");
617 const retained = yield* poll<Snapshot>(
618 `${base}/snapshot`,
619 (snapshot) => snapshot.attempts.length > 0,
620 );
621 expect(retained.attempts[0]!.recovery).not.toBeNull();
622 if (retained.deliveries.length === 0) {
623 expect(retained.pendingJobs).toHaveLength(1);
624 expect(retained.pendingJobs[0]!.callback).toBe("crash");
625 expect(retained.pendingJobs[0]!.id).toBe("recovery");
626 yield* json(`${base}/wake`, "POST");
627 }
628 const recovered = yield* poll<Snapshot>(`${base}/snapshot`, drained(1));
629 expect(deliveries(recovered)).toEqual(["crash:recovered"]);
630 expect(recovered.pendingJobs).toEqual([]);
631 expect(recovered.attempts).toHaveLength(2);
632 expect(recovered.attempts[1]!.boots).toBeGreaterThan(
633 recovered.attempts[0]!.boots,
634 );
635 }),
636 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
637 );
638
639 test(
640 "retains an unknown callback until its registration is restored",
641 Effect.gen(function* () {
642 const { url } = yield* stack;
643 const initial = yield* json<Snapshot>(
644 `${url}/unknown/optional`,
645 "POST",
646 );
647 expect(initial.alarm).not.toBeNull();
648 yield* Effect.sleep("5 seconds");
649 expect(
650 (yield* json<{ aborted: boolean }>(`${url}/unknown/abort`, "POST"))
651 .aborted,
652 ).toBe(true);
653 const reconstructed = yield* json<Snapshot>(`${url}/unknown/snapshot`);
654 expect(reconstructed.boots).toBeGreaterThan(initial.boots);
655 expect(reconstructed.pendingJobs).toEqual(initial.pendingJobs);
656 expect(reconstructed.deliveries).toEqual([]);
657 const released = yield* json<Snapshot>(
658 `${url}/unknown/release-pending`,
659 "POST",
660 );
661 const pending = yield* poll<Snapshot>(
662 `${url}/unknown/snapshot`,
663 (value) => value.alarm !== null && value.alarm > released.alarm!,
664 );
665 expect(pending.boots).toBeGreaterThan(initial.boots);
666 expect(pending.deliveries).toEqual([]);
667 expect(pending.alarm).not.toBeNull();
668 expect(pending.alarm!).toBeGreaterThan(released.alarm!);
669 expect(pending.pendingJobs).toEqual(
670 released.pendingJobs.map((job) => ({
671 ...job,
672 run_at: pending.alarm,
673 })),
674 );
675 yield* json(`${url}/unknown/enable-optional`, "POST");
676 expect(
677 (yield* json<{ aborted: boolean }>(`${url}/unknown/abort`, "POST"))
678 .aborted,
679 ).toBe(true);
680 const restored = yield* poll<Snapshot>(
681 `${url}/unknown/snapshot`,
682 drained(1),
683 "4 seconds",
684 );
685 expect(deliveries(restored)).toEqual(["optional:retained"]);
686 expect(restored.boots).toBeGreaterThan(pending.boots);
687 expect(restored.alarm).toBeNull();
688 }),
689 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
690 );
691
692 test(
693 "drains a backlog over native alarm wakes without losing or duplicating jobs",
694 Effect.gen(function* () {
695 const { url } = yield* stack;
696 yield* json(`${url}/batch/batch`, "POST");
697 const result = yield* poll<Snapshot>(
698 `${url}/batch/snapshot`,
699 drained(105),
700 );
701 expect(deliveries(result)).toEqual(
702 Array.from({ length: 105 }, (_, i) => `archive:batch-${i}`).sort(),
703 );
704 expect(result.alarm).toBeNull();
705 }),
706 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
707 );
708
709 test(
710 "legacy scheduling and its native alarm handler coexist with registered callbacks",
711 Effect.gen(function* () {
712 const { url } = yield* stack;
713 yield* json(`${url}/legacy/legacy`, "POST");
714 const result = yield* poll<{
715 registered: string | null;
716 legacy: { id: string; payload: { value: string } }[];
717 pending: unknown[];
718 alarm: number | null;
719 }>(
720 `${url}/legacy/legacy`,
721 (value) =>
722 value.registered !== null &&
723 value.legacy.length === 1 &&
724 value.alarm === null,
725 );
726 expect(result.registered).toBe("registered");
727 expect(result.legacy).toHaveLength(1);
728 expect(result.legacy[0]?.id).toBe("shared-id");
729 expect(result.legacy[0]?.payload).toEqual({ value: "legacy" });
730 expect(result.pending).toEqual([]);
731 expect(result.alarm).toBeNull();
732 }),
733 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
734 );
735 },
736 { tags: ["provider:cloudflare", "provider:cloudflare:worker"] },
737);