1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import type { RuntimeContext } from "@/RuntimeContext.ts";
4import * as Cause from "effect/Cause";
5import * as Context from "effect/Context";
6import * as Data from "effect/Data";
7import * as Deferred from "effect/Deferred";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Exit from "effect/Exit";
11import * as Fiber from "effect/Fiber";
12import * as Ref from "effect/Ref";
13import * as Result from "effect/Result";
14import * as Scheduler from "effect/Scheduler";
15
16class TransactionMarker extends Context.Service<
17 TransactionMarker,
18 { value: string }
19>()("alarm-callback/TransactionMarker") {}
20
21class Rollback extends Data.TaggedError("Rollback")<{ value: string }> {}
22class RetryArchive extends Data.TaggedError("RetryArchive")<{}> {}
23
24export interface Delivery {
25 callback: string;
26 value: string;
27 application: string | null;
28 boots: number;
29}
30
31export interface Attempt {
32 now: number;
33 recovery: number | null;
34 boots: number;
35}
36
37export interface Snapshot {
38 id: string;
39 boots: number;
40 deliveries: Delivery[];
41 attempts: Attempt[];
42 application: string | null;
43 cleanupWrite: string | null;
44 rows: { value: string }[];
45 alarm: number | null;
46 pendingJobs: {
47 callback: string;
48 id: string;
49 run_at: number;
50 version: string;
51 }[];
52}
53
54export interface RegistrationResult {
55 failure: {
56 tag: "CallbackError";
57 callback: string;
58 message: string;
59 } | null;
60 snapshot: Snapshot;
61}
62
63export interface RollbackResult {
64 failure: string;
65 cleanupWaited: boolean;
66 alarmBefore: number | null;
67 alarmAfter: number | null;
68 snapshot: Snapshot;
69}
70
71export interface SiblingTransactionResult {
72 failure: string;
73 siblingAcknowledged: boolean;
74 siblingFailure: {
75 tag: "DurableObjectStorageError";
76 operation: string;
77 message: string;
78 } | null;
79 siblingValue: string | null;
80 snapshot: Snapshot;
81}
82
83export interface BookkeepingCounts {
84 schemaChecks: number;
85 reconciliations: number;
86 setAlarm: number;
87 deleteAlarm: number;
88}
89
90export interface BatchResult {
91 counts: BookkeepingCounts;
92 snapshot: Snapshot;
93}
94
95export interface FailedBatchResult {
96 failure: string | null;
97 rolledBack: BatchResult;
98 recovered: BatchResult;
99}
100
101export interface AlarmObservationResult {
102 at: number;
103 observations: (number | null)[];
104 committed: number | null;
105}
106
107export interface ExplicitRollbackResult {
108 repeatedRollbackSucceeded: boolean;
109 operations: {
110 operation: "put" | "sql" | "schedule";
111 acknowledged: boolean;
112 failure: {
113 tag: "DurableObjectStorageError";
114 wrappedBy: "CallbackError" | null;
115 message: string;
116 } | null;
117 }[];
118 snapshot: Snapshot;
119}
120
121export class AlarmObject extends Cloudflare.DurableObject<AlarmObject>()(
122 "AlarmObject",
123 Effect.gen(function* () {
124 const state = yield* Cloudflare.DurableObjectState;
125 const storage = state.storage;
126
127 return Effect.gen(function* () {
128 const boots = ((yield* storage.get<number>("boots")) ?? 0) + 1;
129 yield* storage.put("boots", boots);
130 yield* storage.sql.exec(
131 "CREATE TABLE IF NOT EXISTS application_values (value TEXT NOT NULL)",
132 );
133
134 const record = Effect.fn(function* (callback: string, value: string) {
135 const deliveries = (yield* storage.get<Delivery[]>("deliveries")) ?? [];
136 yield* storage.put("deliveries", [
137 ...deliveries,
138 {
139 callback,
140 value,
141 application: (yield* storage.get<string>("application")) ?? null,
142 boots,
143 },
144 ]);
145 });
146
147 const archive: (payload: {
148 value: string;
149 }) => Effect.Effect<
150 void,
151 Alchemy.CallbackError | Cloudflare.DurableObjectStorageError,
152 RuntimeContext
153 > = Effect.fn(function* (payload: { value: string }) {
154 if (payload.value === "replace-first") {
155 yield* onArchive.schedule("replace", {
156 after: "1 second",
157 payload: { value: "replace-second" },
158 });
159 }
160 yield* record("archive", payload.value);
161 });
162
163 const onArchive = yield* Alchemy.makeCallback("archive", archive);
164 const onTransactionalRegistration = yield* storage
165 .transaction(
166 Alchemy.makeCallback(
167 "transactional-init",
168 Effect.fn(function* (payload: { value: string }) {
169 yield* storage.put("application", payload.value);
170 yield* record("transactional-init", payload.value);
171 }),
172 ),
173 )
174 .pipe(Effect.orDie);
175 const onSecondary = yield* Alchemy.makeCallback(
176 "secondary",
177 Effect.fn(function* (payload: { value: string }) {
178 yield* record("secondary", payload.value);
179 }),
180 );
181 const onRetry = yield* Alchemy.makeCallback(
182 "retry",
183 Effect.fn(function* (payload: { value: string }) {
184 const attempts = (yield* storage.get<Attempt[]>("attempts")) ?? [];
185 yield* storage.put("attempts", [
186 ...attempts,
187 {
188 now: yield* Effect.sync(() => Date.now()),
189 recovery: yield* storage.getAlarm(),
190 boots,
191 },
192 ]);
193 if (attempts.length === 0) {
194 yield* Effect.fail(new RetryArchive());
195 }
196 yield* record("retry", payload.value);
197 }),
198 { retry: { delay: "1 second" } },
199 );
200 const onCrash = yield* Alchemy.makeCallback(
201 "crash",
202 Effect.fn(function* (payload: { value: string; retryAlarm?: boolean }) {
203 const attempts = (yield* storage.get<Attempt[]>("attempts")) ?? [];
204 yield* storage.put("attempts", [
205 ...attempts,
206 {
207 now: yield* Effect.sync(() => Date.now()),
208 recovery: yield* storage.getAlarm(),
209 boots,
210 },
211 ]);
212 if (attempts.length === 0) {
213 // abort cancels buffered writes; persist the one-shot crash marker first.
214 yield* storage.sync();
215 yield* state.abort("alarm callback crash", {
216 retryAlarm: payload.retryAlarm ?? true,
217 });
218 }
219 yield* record("crash", payload.value);
220 }),
221 { retry: { delay: "1 second" } },
222 );
223 const onOptional =
224 (yield* storage.get<boolean>("optionalEnabled")) === false
225 ? undefined
226 : yield* Alchemy.makeCallback(
227 "optional",
228 Effect.fn(function* (payload: { value: string }) {
229 yield* record("optional", payload.value);
230 }),
231 { retry: { delay: "1 second" } },
232 );
233
234 const snapshot = Effect.fn(function* () {
235 const tables = yield* (yield* storage.sql.exec<{ name: string }>(
236 "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'alchemy_alarm_callbacks'",
237 )).toArray();
238 const pendingJobs =
239 tables.length === 0
240 ? []
241 : yield* (yield* storage.sql.exec<{
242 callback: string;
243 id: string;
244 run_at: number;
245 version: string;
246 }>(
247 "SELECT callback, id, run_at, version FROM alchemy_alarm_callbacks ORDER BY callback, id",
248 )).toArray();
249 const result: Snapshot = {
250 id: yield* Effect.sync(() => state.id.toString()),
251 boots,
252 deliveries: (yield* storage.get<Delivery[]>("deliveries")) ?? [],
253 attempts: (yield* storage.get<Attempt[]>("attempts")) ?? [],
254 application: (yield* storage.get<string>("application")) ?? null,
255 cleanupWrite: (yield* storage.get<string>("cleanupWrite")) ?? null,
256 rows: yield* (yield* storage.sql.exec<{ value: string }>(
257 "SELECT value FROM application_values ORDER BY value",
258 )).toArray(),
259 alarm: yield* storage.getAlarm(),
260 pendingJobs,
261 };
262 return result;
263 });
264
265 const pending = Effect.fn(function* () {
266 yield* onArchive.schedule("keep", {
267 after: "2 seconds",
268 payload: { value: "kept" },
269 });
270 return yield* storage.getAlarm();
271 });
272 const transactionalWrites = Effect.gen(function* () {
273 yield* storage.put("application", "uncommitted");
274 yield* storage.sql.exec(
275 "INSERT INTO application_values (value) VALUES (?)",
276 "uncommitted",
277 );
278 yield* onArchive.cancel("keep");
279 yield* onArchive.schedule("rolled-back", {
280 after: "1 second",
281 payload: { value: "rolled-back" },
282 });
283 });
284 const rollbackResult = Effect.fn(function* (
285 failure: string,
286 alarmBefore: number | null,
287 cleanupWaited = true,
288 ) {
289 const alarmAfter = yield* storage.getAlarm();
290 const result: RollbackResult = {
291 failure,
292 cleanupWaited,
293 alarmBefore,
294 alarmAfter,
295 snapshot: yield* snapshot(),
296 };
297 yield* onArchive.schedule("checkpoint", {
298 after: "3 seconds",
299 payload: { value: "checkpoint" },
300 });
301 return result;
302 });
303
304 const measureBookkeeping = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
305 Effect.acquireUseRelease(
306 Effect.sync(() => {
307 const raw = state.raw.storage;
308 const counts: BookkeepingCounts = {
309 schemaChecks: 0,
310 reconciliations: 0,
311 setAlarm: 0,
312 deleteAlarm: 0,
313 };
314 const exec = raw.sql.exec;
315 const setAlarm = raw.setAlarm;
316 const deleteAlarm = raw.deleteAlarm;
317 raw.sql.exec = function <
318 T extends Record<string, Cloudflare.SqlStorageValue>,
319 >(query: string, ...bindings: Cloudflare.SqlStorageValue[]) {
320 if (
321 query.includes("SELECT name FROM sqlite_master") &&
322 query.includes("alchemy_alarm_schema")
323 )
324 counts.schemaChecks++;
325 if (query.includes("SELECT MIN(run_at) AS run_at FROM (")) {
326 counts.reconciliations++;
327 }
328 return exec.call(raw.sql, query, ...bindings) as ReturnType<
329 typeof raw.sql.exec<T>
330 >;
331 };
332 raw.setAlarm = (at, options) => {
333 counts.setAlarm++;
334 return setAlarm.call(raw, at, options);
335 };
336 raw.deleteAlarm = (options) => {
337 counts.deleteAlarm++;
338 return deleteAlarm.call(raw, options);
339 };
340 return { raw, counts, exec, setAlarm, deleteAlarm };
341 }),
342 ({ counts }) => effect.pipe(Effect.as(counts)),
343 ({ raw, exec, setAlarm, deleteAlarm }) =>
344 Effect.sync(() => {
345 raw.sql.exec = exec;
346 raw.setAlarm = setAlarm;
347 raw.deleteAlarm = deleteAlarm;
348 }),
349 );
350
351 return {
352 snapshot,
353 bookkeeping: Effect.fn(function* () {
354 const at = yield* Effect.sync(() => Date.now() + 1_500);
355 const counts = yield* measureBookkeeping(
356 storage
357 .transaction(
358 Effect.gen(function* () {
359 for (let i = 0; i < 20; i++) {
360 yield* onArchive.schedule(`cancelled-${i}`, {
361 at,
362 payload: { value: "cancelled" },
363 });
364 yield* onArchive.cancel(`cancelled-${i}`);
365 }
366 yield* onArchive.schedule("kept", {
367 at,
368 payload: { value: "batched" },
369 });
370 yield* storage.transaction(
371 Effect.gen(function* () {
372 yield* onSecondary.schedule("kept", {
373 at,
374 payload: { value: "nested" },
375 });
376 yield* Cloudflare.scheduleEvent(
377 "cancelled-legacy",
378 new Date(at),
379 {},
380 );
381 yield* Cloudflare.cancelEvent("cancelled-legacy");
382 }),
383 );
384 yield* Effect.addFinalizer(() =>
385 onArchive
386 .schedule("finalizer", {
387 at,
388 payload: { value: "finalizer" },
389 })
390 .pipe(Effect.orDie),
391 );
392 }),
393 )
394 .pipe(
395 Effect.provideService(Cloudflare.DurableObjectState, state),
396 ),
397 );
398 const result: BatchResult = { counts, snapshot: yield* snapshot() };
399 return result;
400 }),
401 cancelBookkeeping: Effect.fn(function* () {
402 yield* onArchive.schedule("cancelled", {
403 after: "1 minute",
404 payload: { value: "cancelled" },
405 });
406 const counts = yield* measureBookkeeping(
407 storage.transaction(
408 Effect.gen(function* () {
409 yield* onArchive.cancel("cancelled");
410 for (let i = 0; i < 20; i++)
411 yield* onArchive.cancel(`missing-${i}`);
412 }),
413 ),
414 );
415 const result: BatchResult = { counts, snapshot: yield* snapshot() };
416 return result;
417 }),
418 failedBookkeeping: Effect.fn(function* (explicit: boolean) {
419 let failure: string | null = null;
420 const counts = yield* measureBookkeeping(
421 storage
422 .transaction(
423 Effect.fn(function* (txn: Cloudflare.DurableObjectTransaction) {
424 yield* storage.put("application", "uncommitted");
425 yield* storage.sql.exec(
426 "INSERT INTO application_values (value) VALUES ('uncommitted')",
427 );
428 yield* onArchive.schedule("rolled-back", {
429 after: "1 minute",
430 payload: { value: "rolled-back" },
431 });
432 if (explicit) {
433 yield* txn.rollback();
434 } else {
435 // A real SQLite failure during deferred reconciliation must roll back the batch.
436 yield* storage.sql.exec(
437 "DROP TABLE alchemy_alarm_callbacks",
438 );
439 }
440 }),
441 )
442 .pipe(
443 Effect.exit,
444 Effect.tap((exit) =>
445 Effect.sync(() => {
446 failure = Exit.isFailure(exit)
447 ? String(Cause.squash(exit.cause))
448 : null;
449 }),
450 ),
451 ),
452 );
453 const rolledBack: BatchResult = {
454 counts,
455 snapshot: yield* snapshot(),
456 };
457 const recovered = yield* measureBookkeeping(
458 storage.transaction(
459 Effect.gen(function* () {
460 yield* onArchive.schedule("recovered", {
461 after: "1 second",
462 payload: { value: "recovered" },
463 });
464 yield* onArchive.cancel("missing");
465 }),
466 ),
467 );
468 const result: FailedBatchResult = {
469 failure,
470 rolledBack,
471 recovered: { counts: recovered, snapshot: yield* snapshot() },
472 };
473 return result;
474 }),
475 alarmObservations: Effect.fn(function* () {
476 const at = yield* Effect.sync(() => Date.now() + 60_000);
477 const observations = yield* storage.transaction(
478 Effect.fn(function* (txn: Cloudflare.DurableObjectTransaction) {
479 const values: (number | null)[] = [];
480 yield* onArchive.schedule("observed", {
481 at,
482 payload: { value: "observed" },
483 });
484 values.push(yield* storage.getAlarm());
485 yield* onArchive.cancel("observed");
486 values.push(yield* txn.getAlarm());
487 yield* onArchive.schedule("observed", {
488 at,
489 payload: { value: "observed" },
490 });
491 yield* txn.setAlarm(at + 1_000);
492 values.push(yield* storage.getAlarm());
493 yield* onArchive.cancel("observed");
494 yield* storage.setAlarm(at + 2_000);
495 values.push(yield* txn.getAlarm());
496 yield* onArchive.schedule("observed", {
497 at,
498 payload: { value: "observed" },
499 });
500 yield* txn.deleteAlarm();
501 values.push(yield* storage.getAlarm());
502 yield* onArchive.schedule("observed", {
503 at,
504 payload: { value: "observed" },
505 });
506 yield* storage.deleteAlarm();
507 values.push(yield* txn.getAlarm());
508 yield* onArchive.cancel("observed");
509 yield* txn.setAlarm(at + 3_000);
510 return values;
511 }),
512 );
513 const committed = yield* storage.getAlarm();
514 yield* storage.deleteAlarm();
515 const result: AlarmObservationResult = {
516 at,
517 observations,
518 committed,
519 };
520 return result;
521 }),
522 registerLate: Effect.fn(function* () {
523 const exit = yield* Effect.exit(
524 Alchemy.makeCallback(
525 "late",
526 (_payload: { value: string }) => Effect.void,
527 ),
528 );
529 const defect = Exit.isFailure(exit)
530 ? Result.getOrUndefined(Cause.findDefect(exit.cause))
531 : undefined;
532 const result: RegistrationResult = {
533 failure:
534 defect instanceof Alchemy.CallbackError
535 ? {
536 tag: defect._tag,
537 callback: defect.callback,
538 message: defect.message,
539 }
540 : null,
541 snapshot: yield* snapshot(),
542 };
543 return result;
544 }),
545 timing: Effect.fn(function* () {
546 const at = yield* Effect.sync(() => Date.now() + 1_500);
547 yield* storage.transaction(
548 Effect.gen(function* () {
549 yield* onArchive.schedule("same", {
550 after: "1 second",
551 payload: { value: "superseded" },
552 });
553 yield* onArchive.schedule("same", {
554 at,
555 payload: { value: "latest" },
556 });
557 yield* onSecondary.schedule("same", {
558 at,
559 payload: { value: "other-callback" },
560 });
561 yield* onArchive.schedule("cancelled", {
562 at,
563 payload: { value: "cancelled" },
564 });
565 yield* onArchive.cancel("cancelled");
566 yield* onArchive.cancel("missing");
567 yield* onArchive.schedule("date", {
568 at: new Date(at),
569 payload: { value: "date" },
570 });
571 yield* onArchive.schedule("duration", {
572 after: Duration.millis(1_500),
573 payload: { value: "duration" },
574 });
575 yield* onArchive.schedule("checkpoint", {
576 after: "3 seconds",
577 payload: { value: "checkpoint" },
578 });
579 }),
580 );
581 return yield* snapshot();
582 }),
583 atomic: Effect.fn(function* () {
584 const marker = yield* storage
585 .transaction(
586 Effect.gen(function* () {
587 const { value } = yield* TransactionMarker;
588 yield* storage.put("application", value);
589 yield* storage.sql.exec(
590 "INSERT INTO application_values (value) VALUES (?)",
591 value,
592 );
593 yield* onArchive.schedule("atomic", {
594 after: "1 second",
595 payload: { value },
596 });
597 return value;
598 }),
599 )
600 .pipe(
601 Effect.provideService(TransactionMarker, { value: "committed" }),
602 );
603 const legacyOverload = yield* storage
604 .transaction(
605 Effect.fn(function* (
606 transaction: Cloudflare.DurableObjectTransaction,
607 ) {
608 const { value } = yield* TransactionMarker;
609 yield* transaction.put("legacyOverload", value);
610 return yield* transaction.get<string>("legacyOverload");
611 }),
612 )
613 .pipe(
614 Effect.provideService(TransactionMarker, { value: "callback" }),
615 );
616 return { marker, legacyOverload, snapshot: yield* snapshot() };
617 }),
618 transactionalRegistration: Effect.fn(function* () {
619 yield* onTransactionalRegistration.schedule("transactional-init", {
620 after: "1 second",
621 payload: { value: "registered-in-transaction" },
622 });
623 return yield* snapshot();
624 }),
625 siblingTransaction: Effect.fn(function* () {
626 // Native input gates can block timers used for fiber scheduling.
627 const scheduler = yield* Effect.sync(
628 () => new Scheduler.MixedScheduler("sync"),
629 );
630 return yield* Effect.scoped(
631 Effect.gen(function* () {
632 const ready = yield* Deferred.make<void>();
633 const release = yield* Deferred.make<void>();
634 const completed = yield* Deferred.make<void>();
635 const sibling = yield* Effect.gen(function* () {
636 yield* Deferred.succeed(ready, undefined);
637 yield* Deferred.await(release);
638 yield* storage.put("siblingWrite", "independent-write");
639 }).pipe(
640 Effect.timeout("2 seconds"),
641 Effect.exit,
642 Effect.tap(() => Deferred.succeed(completed, undefined)),
643 Effect.forkScoped,
644 );
645 yield* Deferred.await(ready).pipe(
646 Effect.timeout("2 seconds"),
647 Effect.orDie,
648 );
649 const failure = yield* storage
650 .transaction(
651 Effect.gen(function* () {
652 yield* storage.put("application", "uncommitted");
653 yield* storage.sql.exec(
654 "INSERT INTO application_values (value) VALUES (?)",
655 "uncommitted",
656 );
657 yield* Deferred.succeed(release, undefined);
658 yield* Deferred.await(completed).pipe(
659 Effect.timeout("2 seconds"),
660 Effect.orDie,
661 );
662 return yield* Effect.fail(
663 new Rollback({ value: "sibling-rollback" }),
664 );
665 }),
666 )
667 .pipe(
668 Effect.catchTag("Rollback", (error) =>
669 Effect.succeed(error.value),
670 ),
671 );
672 const exit = yield* Fiber.join(sibling).pipe(
673 Effect.timeout("2 seconds"),
674 Effect.orDie,
675 );
676 const error = Exit.isFailure(exit)
677 ? Cause.squash(exit.cause)
678 : undefined;
679 const result: SiblingTransactionResult = {
680 failure,
681 siblingAcknowledged: Exit.isSuccess(exit),
682 siblingFailure:
683 error instanceof Cloudflare.DurableObjectStorageError
684 ? {
685 tag: error._tag,
686 operation: error.operation,
687 message: error.message,
688 }
689 : null,
690 siblingValue:
691 (yield* storage.get<string>("siblingWrite")) ?? null,
692 snapshot: yield* snapshot(),
693 };
694 return result;
695 }),
696 ).pipe(Effect.provideService(Scheduler.Scheduler, scheduler));
697 }),
698 rollbackExplicit: Effect.fn(function* () {
699 const describeOperation = <A, E>(
700 operation: ExplicitRollbackResult["operations"][number]["operation"],
701 exit: Exit.Exit<A, E>,
702 ): ExplicitRollbackResult["operations"][number] => {
703 const error = Exit.isFailure(exit)
704 ? Cause.squash(exit.cause)
705 : undefined;
706 const cause =
707 error instanceof Alchemy.CallbackError ? error.cause : error;
708 return {
709 operation,
710 acknowledged: Exit.isSuccess(exit),
711 failure:
712 cause instanceof Cloudflare.DurableObjectStorageError
713 ? {
714 tag: cause._tag,
715 wrappedBy:
716 error instanceof Alchemy.CallbackError
717 ? error._tag
718 : null,
719 message: cause.message,
720 }
721 : null,
722 };
723 };
724 const outcome = yield* storage.transaction(
725 Effect.fn(function* (txn: Cloudflare.DurableObjectTransaction) {
726 yield* txn.put("application", "before-rollback");
727 yield* storage.sql.exec(
728 "INSERT INTO application_values (value) VALUES (?)",
729 "before-rollback",
730 );
731 yield* onArchive.schedule("before-explicit-rollback", {
732 after: "1 minute",
733 payload: { value: "before-rollback" },
734 });
735 yield* txn.rollback();
736 const repeatedRollback = yield* Effect.exit(txn.rollback());
737 const put = yield* Effect.exit(
738 storage.put("application", "after-rollback"),
739 );
740 const sql = yield* Effect.exit(
741 storage.sql.exec(
742 "INSERT INTO application_values (value) VALUES (?)",
743 "after-rollback",
744 ),
745 );
746 const schedule = yield* Effect.exit(
747 onArchive.schedule("after-explicit-rollback", {
748 after: "1 minute",
749 payload: { value: "after-rollback" },
750 }),
751 );
752 return {
753 repeatedRollbackSucceeded: Exit.isSuccess(repeatedRollback),
754 operations: [
755 describeOperation("put", put),
756 describeOperation("sql", sql),
757 describeOperation("schedule", schedule),
758 ],
759 };
760 }),
761 );
762 const result: ExplicitRollbackResult = {
763 ...outcome,
764 snapshot: yield* snapshot(),
765 };
766 return result;
767 }),
768 rollbackTyped: Effect.fn(function* () {
769 const alarmBefore = yield* pending();
770 const failure = yield* storage
771 .transaction(
772 Effect.gen(function* () {
773 yield* transactionalWrites;
774 return yield* Effect.fail(
775 new Rollback({ value: "typed-value" }),
776 );
777 }),
778 )
779 .pipe(
780 Effect.catchTag("Rollback", (error) =>
781 Effect.succeed(error.value),
782 ),
783 );
784 return yield* rollbackResult(failure, alarmBefore);
785 }),
786 rollbackDefect: Effect.fn(function* () {
787 const alarmBefore = yield* pending();
788 const exit = yield* storage
789 .transaction(
790 Effect.gen(function* () {
791 yield* transactionalWrites;
792 yield* storage.getAlarm();
793 return yield* Effect.die("rollback-defect");
794 }),
795 )
796 .pipe(Effect.exit);
797 const failure =
798 Exit.isFailure(exit) &&
799 Cause.squash(exit.cause) === "rollback-defect"
800 ? "defect"
801 : "unexpected";
802 return yield* rollbackResult(failure, alarmBefore);
803 }),
804 rollbackInterrupt: Effect.fn(function* () {
805 return yield* Effect.scoped(
806 Effect.gen(function* () {
807 const alarmBefore = yield* pending();
808 const entered = yield* Deferred.make<void>();
809 const cleanupStarted = yield* Deferred.make<void>();
810 const releaseCleanup = yield* Deferred.make<void>();
811 const interruptFinished = yield* Deferred.make<void>();
812 const cleaned = yield* Ref.make(false);
813 const fiber = yield* storage
814 .transaction(
815 Effect.gen(function* () {
816 yield* transactionalWrites;
817 yield* storage.getAlarm();
818 yield* Deferred.succeed(entered, undefined);
819 yield* Effect.never.pipe(
820 Effect.timeout("5 seconds"),
821 Effect.orDie,
822 );
823 }).pipe(
824 Effect.ensuring(
825 Effect.gen(function* () {
826 yield* Deferred.succeed(cleanupStarted, undefined);
827 yield* Deferred.await(releaseCleanup).pipe(
828 Effect.timeout("2 seconds"),
829 Effect.orDie,
830 );
831 yield* storage.put("cleanupWrite", "rolled-back");
832 yield* Ref.set(cleaned, true);
833 }),
834 ),
835 ),
836 )
837 .pipe(Effect.forkScoped);
838 yield* Deferred.await(entered).pipe(
839 Effect.timeout("5 seconds"),
840 Effect.orDie,
841 );
842 const interrupter = yield* Fiber.interrupt(fiber).pipe(
843 Effect.andThen(Deferred.succeed(interruptFinished, undefined)),
844 Effect.forkScoped,
845 );
846 yield* Deferred.await(cleanupStarted).pipe(
847 Effect.timeout("8 seconds"),
848 Effect.orDie,
849 );
850 const finishedEarly = yield* Deferred.isDone(interruptFinished);
851 yield* Deferred.succeed(releaseCleanup, undefined);
852 yield* Fiber.join(interrupter);
853 const exit = yield* Fiber.await(fiber);
854 const failure =
855 Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)
856 ? "interrupted"
857 : "unexpected";
858 return yield* rollbackResult(
859 failure,
860 alarmBefore,
861 !finishedEarly && (yield* Ref.get(cleaned)),
862 );
863 }),
864 );
865 }),
866 retry: Effect.fn(function* () {
867 yield* onRetry.schedule("retry", {
868 after: "1 second",
869 payload: { value: "retried" },
870 });
871 }),
872 replace: Effect.fn(function* () {
873 yield* onArchive.schedule("replace", {
874 after: "1 second",
875 payload: { value: "replace-first" },
876 });
877 }),
878 recovery: Effect.fn(function* (retryAlarm = true) {
879 yield* onCrash.schedule("recovery", {
880 after: "1 second",
881 payload: { value: "recovered", retryAlarm },
882 });
883 }),
884 wake: Effect.fn(function* () {
885 const now = yield* Effect.sync(() => Date.now());
886 yield* storage.setAlarm(now);
887 }),
888 prepareReset: Effect.fn(function* (value: string) {
889 yield* onArchive.schedule("reset", {
890 after: "5 minutes",
891 payload: { value },
892 });
893 return yield* snapshot();
894 }),
895 optional: Effect.fn(function* () {
896 if (!onOptional) {
897 return yield* Effect.die("optional callback not registered");
898 }
899 yield* onOptional.schedule("optional", {
900 after: "5 minutes",
901 payload: { value: "retained" },
902 });
903 yield* storage.put("optionalEnabled", false);
904 return yield* snapshot();
905 }),
906 releasePending: Effect.fn(function* () {
907 // Preserve persisted jobs until reconstruction, then advance only their due times.
908 const at = yield* Effect.sync(() => Date.now() + 1_000);
909 yield* storage.transaction(
910 Effect.gen(function* () {
911 yield* storage.sql.exec(
912 "UPDATE alchemy_alarm_callbacks SET run_at = ?",
913 at,
914 );
915 yield* storage.setAlarm(at);
916 }),
917 );
918 return yield* snapshot();
919 }),
920 enableOptional: Effect.fn(function* () {
921 yield* storage.put("optionalEnabled", true);
922 }),
923 batch: Effect.fn(function* () {
924 const at = yield* Effect.sync(() => Date.now() + 1_500);
925 yield* storage.transaction(
926 Effect.gen(function* () {
927 for (let i = 0; i < 105; i++) {
928 yield* onArchive.schedule(`batch-${i}`, {
929 at,
930 payload: { value: `batch-${i}` },
931 });
932 }
933 }),
934 );
935 }),
936 crash: Effect.fn(function* () {
937 yield* state.abort("alarm callback reconstruction", {
938 retryAlarm: false,
939 });
940 }),
941 };
942 });
943 }),
944) {}