1import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";2import { createDiagnostics } from "./diagnostics.ts";3import {4 type EffectContext,5 type EffectUnit,6 type MutationDescriptor,7 PermanentEffectError,8} from "../schema/effects.ts";9import { assert, assertCount } from "./test-assert.ts";10import {11 createMemoryJournalStorage,12 emptyJournal,13 hashJournalValue,14 type JournalDocument,15 journalIdFor,16} from "./write-journal.ts";17import { WriteLedger, type WriteLedgerEnvironment } from "./write-ledger.ts";1819type Row = { id: string; title: string };20type Init = Omit<Row, "id">;21const table = { _table: "records", _schema: {} } as TableProxy<Row, Init>;2223class FakeRejection extends Error {24 override readonly name = "PersistedWriteRejectedError";25 constructor(readonly code: string, readonly reason: string) {26 super(`${code}: ${reason}`);27 }28}2930type Deferred = { resolve(): void; reject(error: unknown): void };3132class FakeDb {33 mutationListeners = new Set<(event: MutationErrorEvent) => void>();34 globalWaits: Deferred[] = [];35 rows: Array<Record<string, unknown>> = [];36 refuseNext: Error | null = null;37 nextId = 0;38 nextBatch = 0;39 reads = 0;4041 value(): Db {42 return this as unknown as Db;43 }4445 onMutationError(listener: (event: MutationErrorEvent) => void): () => void {46 this.mutationListeners.add(listener);47 return () => this.mutationListeners.delete(listener);48 }4950 emit(event: MutationErrorEvent): void {51 for (const listener of this.mutationListeners) listener(event);52 }5354 insert(_table: TableProxy<Row, Init>, values: Init) {55 if (this.refuseNext) {56 const error = this.refuseNext;57 this.refuseNext = null;58 throw error;59 }60 const row = { id: `row-${++this.nextId}`, ...values };61 return { value: row, ...this.handle() };62 }6364 update(_table: TableProxy<Row, Init>, _id: string, _patch: Partial<Init>) {65 return this.handle();66 }6768 delete(_table: TableProxy<Row, Init>, _id: string) {69 return this.handle();70 }7172 all(): Promise<unknown[]> {73 this.reads += 1;74 return Promise.resolve(this.rows);75 }7677 settleGlobal(index = 0): void {78 this.globalWaits[index]?.resolve();79 }8081 rejectGlobal(error: unknown, index = 0): void {82 this.globalWaits[index]?.reject(error);83 }8485 private handle() {86 return {87 batchId: `batch-${++this.nextBatch}`,88 wait: ({ tier }: { tier: "local" | "global" }) => {89 if (tier === "local") return Promise.resolve(undefined);90 return new Promise<unknown>((resolve, reject) => {91 this.globalWaits.push({ resolve: () => resolve(undefined), reject });92 });93 },94 };95 }96}9798type Harness = {99 db: FakeDb;100 ledger: WriteLedger;101 diagnostics: ReturnType<typeof createDiagnostics>;102 storage: ReturnType<typeof createMemoryJournalStorage>;103 units: Map<string, EffectUnit<{ id: string }>>;104 clock: { value: number };105};106107function harness(options: {108 syncConfigured?: boolean;109 storage?: ReturnType<typeof createMemoryJournalStorage>;110 units?: Map<string, EffectUnit<{ id: string }>>;111 db?: FakeDb;112 guardWrite?: () => Promise<(() => void) | void>;113 sweepIntervalMs?: number;114} = {}): Harness {115 const db = options.db ?? new FakeDb();116 const diagnostics = createDiagnostics();117 const storage = options.storage ?? createMemoryJournalStorage();118 const units = options.units ?? new Map<string, EffectUnit<{ id: string }>>();119 const clock = { value: 1000 };120 const environment: WriteLedgerEnvironment = {121 getDb: () => Promise.resolve(db.value()),122 syncConfigured: () => options.syncConfigured ?? true,123 storage,124 resolveTable: (name) => (name === "records" ? { where: (input) => input } : null),125 resolveEffectUnit: (name) => units.get(name) ?? null,126 subscribeRuntimeRecreation: () => () => undefined,127 updateDiagnostics: (update) => update(diagnostics),128 now: () => clock.value,129 retryDelayMs: () => 1,130 probeTimeoutMs: 50,131 sweepIntervalMs: options.sweepIntervalMs ?? 60_000,132 ...(options.guardWrite ? { guardWrite: options.guardWrite } : {}),133 };134 return { db, ledger: new WriteLedger(environment), diagnostics, storage, units, clock };135}136137function unit(138 name: string,139 calls: Array<{ handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }>,140 options: {141 failFirst?: { onSynced?: boolean };142 failAlways?: boolean;143 failPermanent?: boolean;144 expiresAfterMs?: number;145 maxAttempts?: number;146 } = {},147): EffectUnit<{ id: string }> {148 let failed = false;149 return {150 effectName: name,151 handlers: {152 onSynced: (row, context) => {153 if (options.failPermanent) throw new PermanentEffectError("receiver refused for good");154 if (options.failAlways) throw new Error("handler always explodes");155 if (options.failFirst?.onSynced && !failed) {156 failed = true;157 throw new Error("handler exploded");158 }159 calls.push({ handler: "onSynced", row, context });160 },161 onRejected: (row, context) => {162 calls.push({ handler: "onRejected", row, context });163 },164 },165 expiresAfterMs: options.expiresAfterMs ?? null,166 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),167 };168}169170async function flush(times = 6): Promise<void> {171 for (let i = 0; i < times; i += 1) await Promise.resolve();172}173174async function settleTimers(ms = 10): Promise<void> {175 await new Promise((resolve) => setTimeout(resolve, ms));176}177178Deno.test("a verb write progresses saved to synced and fires onSynced exactly once", async () => {179 const calls: Array<180 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }181 > = [];182 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);183 const fixture = harness({ units });184 await fixture.ledger.arm();185 const handle = fixture.ledger.perform<Row>(186 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "one" } },187 { verb: "placeOrder", units: [units.get("chargeCard")!] },188 );189 const row = await handle;190 assert(row.title === "one", "await must resolve the created row at saved");191 assert(handle.stage === "saved", "the write must rest at saved until the store confirms");192 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "a saved write must count as pending");193 assert(fixture.ledger.rowStatus(row.id) === "waiting", "an unsettled row must read waiting");194 fixture.db.settleGlobal();195 await flush();196 assert(String(handle.stage) === "synced", "store confirmation must reach the handle");197 assertCount(calls.length, 1, "onSynced must fire exactly once");198 assert(calls[0].handler === "onSynced", "confirmation must run the action handler");199 assert(200 calls[0].context.journalId === journalIdFor(handle.writeId, "chargeCard"),201 "the handler must receive the (write id, effect name) idempotency key",202 );203 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a settled write must leave pending");204 assert(fixture.ledger.rowStatus(row.id) === "synced", "a confirmed row must read synced");205 await fixture.ledger.flush();206 assert(fixture.storage.text()?.includes(handle.writeId) === false, "settled entries must prune");207 fixture.ledger.dispose();208});209210Deno.test("a synchronous local refusal throws from the call and never journals", async () => {211 const fixture = harness();212 await fixture.ledger.arm();213 fixture.db.refuseNext = new Error("current policy refuses this write");214 let thrown: Error | null = null;215 try {216 fixture.ledger.perform<Row>(217 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "no" } },218 );219 } catch (error) {220 thrown = error as Error;221 }222 assert(thrown?.message === "current policy refuses this write", "the refusal must throw");223 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a refusal must not journal");224 await fixture.ledger.flush();225 assert(226 (fixture.storage.text() ?? "").includes("records") === false,227 "a refusal must leave no journal entry",228 );229 assertCount(fixture.diagnostics.mutationErrors, 1, "the refusal must still be diagnosable");230 fixture.ledger.dispose();231});232233Deno.test("a permanent verdict settles rejected and fires onRejected exactly once", async () => {234 const calls: Array<235 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }236 > = [];237 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);238 const fixture = harness({ units });239 await fixture.ledger.arm();240 const handle = fixture.ledger.perform<Row>(241 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "denied" } },242 { verb: "placeOrder", units: [units.get("chargeCard")!] },243 );244 const row = await handle;245 fixture.db.rejectGlobal(new FakeRejection("permission_denied", "policy tightened"));246 await flush();247 assert(handle.stage === "rejected", "a permanent verdict must settle the stage");248 assert(handle.reason?.code === "permission_denied", "the verdict code must be readable");249 assertCount(calls.length, 1, "onRejected must fire exactly once");250 assert(calls[0].handler === "onRejected", "compensation must run, not the action");251 assert(calls[0].context.fate === "rejected", "the handler must see the rejected fate");252 assert(fixture.ledger.rowStatus(row.id) === "rejected", "the row must read rejected");253 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a rejected write is not pending");254 fixture.ledger.dispose();255});256257Deno.test("a transient failure keeps the write pending and never compensates", async () => {258 const calls: Array<259 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }260 > = [];261 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);262 const fixture = harness({ units });263 await fixture.ledger.arm();264 const handle = fixture.ledger.perform<Row>(265 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "slow" } },266 { units: [units.get("chargeCard")!] },267 );268 await handle;269 fixture.db.rejectGlobal(new FakeRejection("mystery_code", "unknown"));270 await flush();271 assert(handle.stage === "saved", "an uninterpretable code must keep the write pending");272 assertCount(calls.length, 0, "no compensation on a non-permanent code");273 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "the write must stay pending");274 // The scheduled re-derivation asks the store again; confirm it settles.275 await settleTimers();276 fixture.db.settleGlobal(1);277 await flush();278 assert(String(handle.stage) === "synced", "a later confirmation must settle the pending write");279 assertCount(calls.filter((call) => call.handler === "onSynced").length, 1, "then onSynced runs");280 fixture.ledger.dispose();281});282283Deno.test("reload re-arm: journaled obligations fire after the journal is reloaded", async () => {284 const storage = createMemoryJournalStorage();285 // First run: the write settles as synced, but the unit is not registered286 // yet (its declaring module never loaded), so the obligation stays pending.287 const first = harness({ storage });288 await first.ledger.arm();289 const handle = first.ledger.perform<Row>(290 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "kept" } },291 {292 verb: "placeOrder",293 units: [{ effectName: "notify", handlers: {} } as EffectUnit<{ id: string }>],294 },295 );296 await handle;297 first.db.settleGlobal();298 await flush();299 await first.ledger.flush();300 first.ledger.dispose();301 assert(302 (storage.text() ?? "").includes(handle.writeId),303 "the unsatisfied obligation must survive in the persisted journal",304 );305306 const calls: Array<307 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }308 > = [];309 const units = new Map([["notify", unit("notify", calls)]]);310 const second = harness({ storage, units });311 await second.ledger.arm();312 await flush();313 assertCount(calls.length, 1, "the reloaded runtime must re-run the journaled obligation");314 assert(calls[0].handler === "onSynced", "the synced fate must run the action handler");315 assert(316 calls[0].context.journalId === journalIdFor(handle.writeId, "notify"),317 "the idempotency key must survive the reload",318 );319 await second.ledger.flush();320 assert((second.storage.text() ?? "").includes(handle.writeId) === false, "then the entry prunes");321 second.ledger.dispose();322});323324Deno.test("boot reconciliation confirms journaled writes against store state", async () => {325 const calls: Array<326 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }327 > = [];328 const units = new Map([["celebrate", unit("celebrate", calls)]]);329 const document: JournalDocument = emptyJournal();330 document.writes["w-confirmed"] = {331 writeId: "w-confirmed",332 verb: "addTask",333 table: "records",334 op: "insert",335 rowId: "row-a",336 batchId: "batch-a",337 rowHashes: { title: hashJournalValue(document.hashKey, "kept") },338 stage: "saved",339 cause: null,340 code: null,341 reason: null,342 createdAt: 1,343 expiresAt: null,344 effects: {345 celebrate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },346 },347 };348 document.writes["w-waiting"] = {349 writeId: "w-waiting",350 verb: null,351 table: "records",352 op: "insert",353 rowId: "row-b",354 batchId: "batch-b",355 rowHashes: { title: hashJournalValue(document.hashKey, "not there yet") },356 stage: "saved",357 cause: null,358 code: null,359 reason: null,360 createdAt: 2,361 expiresAt: null,362 effects: {},363 };364 const storage = createMemoryJournalStorage(JSON.stringify(document));365 const fixture = harness({ storage, units });366 fixture.db.rows = [{ id: "row-a", title: "kept" }];367 await fixture.ledger.arm();368 await flush();369 await settleTimers();370 assert(371 fixture.ledger.rowStatus("row-a") === "synced",372 "a journaled write the store confirms must settle as synced at boot",373 );374 assertCount(calls.length, 1, "the re-armed effect must fire after the store confirms");375 assert(376 fixture.ledger.rowStatus("row-b") === "waiting",377 "a journaled write the store does not confirm must stay pending",378 );379 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "only the unconfirmed write pends");380 fixture.ledger.dispose();381});382383Deno.test("boot reconciliation applies replayed verdicts to journaled writes", async () => {384 const calls: Array<385 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }386 > = [];387 const units = new Map([["compensate", unit("compensate", calls)]]);388 const document: JournalDocument = emptyJournal();389 document.writes["w-denied"] = {390 writeId: "w-denied",391 verb: "addTask",392 table: "records",393 op: "insert",394 rowId: "row-x",395 batchId: "batch-x",396 rowHashes: { title: hashJournalValue(document.hashKey, "denied") },397 stage: "saved",398 cause: null,399 code: null,400 reason: null,401 createdAt: 1,402 expiresAt: null,403 effects: {404 compensate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },405 },406 };407 const storage = createMemoryJournalStorage(JSON.stringify(document));408 const fixture = harness({ storage, units });409 await fixture.ledger.arm();410 fixture.db.emit(411 {412 code: "permission_denied",413 reason: "policy tightened while offline",414 batch: { batchId: "batch-x" },415 } as unknown as MutationErrorEvent,416 );417 await flush();418 assertCount(calls.length, 1, "the replayed verdict must run compensation once");419 assert(calls[0].handler === "onRejected", "a permanent verdict runs onRejected");420 assert(fixture.ledger.rowStatus("row-x") === "rejected", "the row must read rejected");421 fixture.db.emit(422 {423 code: "permission_denied",424 reason: "duplicate replay",425 batch: { batchId: "batch-x" },426 } as unknown as MutationErrorEvent,427 );428 await flush();429 assertCount(calls.length, 1, "a duplicate verdict must be recorded idempotently");430 fixture.ledger.dispose();431});432433Deno.test("an unknown replayed code keeps the journaled write pending", async () => {434 const document: JournalDocument = emptyJournal();435 document.writes["w-odd"] = {436 writeId: "w-odd",437 verb: null,438 table: "records",439 op: "update",440 rowId: "row-y",441 batchId: "batch-y",442 rowHashes: { title: hashJournalValue(document.hashKey, "edited") },443 stage: "saved",444 cause: null,445 code: null,446 reason: null,447 createdAt: 1,448 expiresAt: null,449 effects: {},450 };451 const storage = createMemoryJournalStorage(JSON.stringify(document));452 const fixture = harness({ storage });453 await fixture.ledger.arm();454 fixture.db.emit(455 {456 code: "mystery",457 reason: "unclassifiable",458 batch: { batchId: "batch-y" },459 } as unknown as MutationErrorEvent,460 );461 await flush();462 assert(463 fixture.ledger.rowStatus("row-y") === "waiting",464 "an uninterpretable code must never settle or compensate a write",465 );466 fixture.ledger.dispose();467});468469Deno.test("a remove is confirmed at boot by the row's absence", async () => {470 const document: JournalDocument = emptyJournal();471 document.writes["w-gone"] = {472 writeId: "w-gone",473 verb: null,474 table: "records",475 op: "remove",476 rowId: "row-z",477 batchId: "batch-z",478 rowHashes: {},479 stage: "saved",480 cause: null,481 code: null,482 reason: null,483 createdAt: 1,484 expiresAt: null,485 effects: {},486 };487 const storage = createMemoryJournalStorage(JSON.stringify(document));488 const fixture = harness({ storage });489 fixture.db.rows = [];490 await fixture.ledger.arm();491 await flush();492 await settleTimers();493 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "the absent row confirms the remove");494 fixture.ledger.dispose();495});496497Deno.test("effect units are independent: one failing handler never blocks another", async () => {498 const calls: Array<499 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }500 > = [];501 const units = new Map<string, EffectUnit<{ id: string }>>([502 ["fragile", unit("fragile", calls, { failFirst: { onSynced: true } })],503 ["sturdy", unit("sturdy", calls)],504 ]);505 const fixture = harness({ units });506 await fixture.ledger.arm();507 const handle = fixture.ledger.perform<Row>(508 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "both" } },509 { units: [units.get("fragile")!, units.get("sturdy")!] },510 );511 await handle;512 fixture.db.settleGlobal();513 await flush();514 assertCount(515 calls.filter((call) => call.handler === "onSynced").length,516 1,517 "the sturdy unit must run even though the fragile one failed",518 );519 assertCount(fixture.diagnostics.effectHandlerFailures, 1, "the failure must be diagnosable");520 await fixture.ledger.flush();521 assert(522 (fixture.storage.text() ?? "").includes("fragile"),523 "the failed obligation must stay journaled so the next boot re-arms it",524 );525 fixture.ledger.dispose();526});527528Deno.test("without configured sync, local durability is settlement and effects run", async () => {529 const calls: Array<530 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }531 > = [];532 const units = new Map([["local-note", unit("local-note", calls)]]);533 const fixture = harness({ syncConfigured: false, units });534 await fixture.ledger.arm();535 const handle = fixture.ledger.perform<Row>(536 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "solo" } },537 { units: [units.get("local-note")!] },538 );539 await handle;540 await flush();541 assert(handle.stage === "synced", "a device without a sync location settles at saved");542 assertCount(calls.length, 1, "effects must run when the write settles");543 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "nothing waits without a store");544 fixture.ledger.dispose();545});546547Deno.test("a write-guard refusal fails the handle and never journals or compensates", async () => {548 const calls: Array<549 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }550 > = [];551 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);552 const fixture = harness({553 units,554 guardWrite: () => Promise.reject(new Error("schema is ahead of this bundle")),555 });556 await fixture.ledger.arm();557 const handle = fixture.ledger.perform<Row>(558 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "held" } },559 { verb: "placeOrder", units: [units.get("chargeCard")!] },560 );561 const error = await handle.saved.then(() => null, (thrown) => thrown as Error);562 assert(563 error?.message === "schema is ahead of this bundle",564 "the guard refusal must surface through the verb call",565 );566 assert(handle.stage !== "rejected", "a refusal is not an adjudicated verdict");567 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a refused write must not journal");568 await fixture.ledger.flush();569 assert(570 (fixture.storage.text() ?? "").includes("held") === false,571 "a refused write must leave no journal entry",572 );573 assertCount(calls.length, 0, "a refused write must fire no effect");574 assertCount(fixture.diagnostics.mutationErrors, 1, "the refusal must be diagnosable");575 assertCount(fixture.db.nextId, 0, "the write must never reach the database");576 fixture.ledger.dispose();577});578579Deno.test("the write-guard release covers exactly the local-durability window", async () => {580 let released = 0;581 const fixture = harness({582 guardWrite: () => Promise.resolve(() => void (released += 1)),583 });584 await fixture.ledger.arm();585 const handle = fixture.ledger.perform<Row>(586 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "locked" } },587 );588 await handle;589 await flush();590 assertCount(released, 1, "the lock must release once the write settles locally");591 fixture.db.settleGlobal();592 await flush();593 assertCount(released, 1, "global settlement must not release the lock again");594 fixture.ledger.dispose();595});596597Deno.test("the serialized journal holds hashes, never plaintext column values", async () => {598 const fixture = harness();599 await fixture.ledger.arm();600 const handle = fixture.ledger.perform<Row>(601 {602 kind: "insert",603 table: table as TableProxy<unknown, unknown>,604 values: { title: "SECRET-VALUE-XYZ" },605 },606 );607 await handle;608 await fixture.ledger.flush();609 const persisted = fixture.storage.text() ?? "";610 assert(persisted.includes(handle.writeId), "the pending write must be journaled");611 assert(612 persisted.includes("SECRET-VALUE-XYZ") === false,613 "no plaintext column value may reach the persisted journal",614 );615 assert(persisted.includes("rowHashes"), "the equality-probe hashes must be journaled");616 fixture.ledger.dispose();617});618619Deno.test("boot compensation receives write identity only; live handlers keep the row", async () => {620 const calls: Array<621 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }622 > = [];623 const units = new Map([["compensate", unit("compensate", calls)]]);624 const document: JournalDocument = emptyJournal();625 document.writes["w-back"] = {626 writeId: "w-back",627 verb: "placeOrder",628 table: "records",629 op: "insert",630 rowId: "row-q",631 batchId: "batch-q",632 rowHashes: { title: hashJournalValue(document.hashKey, "gone") },633 stage: "saved",634 cause: null,635 code: null,636 reason: null,637 createdAt: 1,638 expiresAt: null,639 effects: {640 compensate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },641 },642 };643 const storage = createMemoryJournalStorage(JSON.stringify(document));644 const fixture = harness({ storage, units });645 await fixture.ledger.arm();646 fixture.db.emit(647 {648 code: "permission_denied",649 reason: "tightened while away",650 batch: { batchId: "batch-q" },651 } as unknown as MutationErrorEvent,652 );653 await flush();654 assertCount(calls.length, 1, "the replayed verdict must run compensation");655 assert(656 Object.keys(calls[0].row).join(",") === "id" && calls[0].row.id === "row-q",657 "after a reload the journal holds no values: compensation receives identity only",658 );659 assert(calls[0].context.cause === "denied", "the structured cause must reach the handler");660 fixture.ledger.dispose();661});662663Deno.test("a synced obligation re-armed at boot receives the live store row", async () => {664 const calls: Array<665 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }666 > = [];667 const units = new Map([["celebrate", unit("celebrate", calls)]]);668 const document: JournalDocument = emptyJournal();669 document.writes["w-live"] = {670 writeId: "w-live",671 verb: "addTask",672 table: "records",673 op: "insert",674 rowId: "row-l",675 batchId: "batch-l",676 rowHashes: { title: hashJournalValue(document.hashKey, "kept") },677 stage: "synced",678 cause: null,679 code: null,680 reason: null,681 createdAt: 1,682 expiresAt: null,683 effects: {684 celebrate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },685 },686 };687 const storage = createMemoryJournalStorage(JSON.stringify(document));688 const fixture = harness({ storage, units });689 fixture.db.rows = [{ id: "row-l", title: "kept" }];690 await fixture.ledger.arm();691 await flush();692 assertCount(calls.length, 1, "the re-armed obligation must run");693 assert(694 (calls[0].row as { title?: string }).title === "kept",695 "a synced write's handler must receive the live row the store confirmed",696 );697 fixture.ledger.dispose();698});699700Deno.test("an overdue intent surfaces in pending state and diagnostics without compensation", async () => {701 const calls: Array<702 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }703 > = [];704 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);705 const fixture = harness({ units, sweepIntervalMs: 5 });706 await fixture.ledger.arm();707 const handle = fixture.ledger.perform<Row>(708 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "slow" } },709 { verb: "placeOrder", units: [units.get("chargeCard")!], expiresAfterMs: 50 },710 );711 await handle;712 assert(713 fixture.ledger.getPendingSnapshot().writes[0]?.expired === false,714 "an intent inside its lifespan is not overdue",715 );716 fixture.clock.value = 2000;717 await settleTimers(20);718 const snapshot = fixture.ledger.getPendingSnapshot();719 assert(snapshot.writes[0]?.expired === true, "the overdue intent must surface");720 assertCount(fixture.diagnostics.expiredPendingWrites, 1, "diagnostics must count it");721 assert(handle.stage === "saved", "no withdrawal exists, so the stage must not change");722 assertCount(calls.length, 0, "no compensation may fire for a write that can still sync");723 // The store later confirms the overdue write: it settles normally.724 fixture.db.settleGlobal();725 await flush();726 assert(String(handle.stage) === "synced", "a late confirmation still settles the write");727 fixture.ledger.dispose();728});729730Deno.test("a closed delivery window retires the obligation: no handler, diagnosable, prunable", async () => {731 const calls: Array<732 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }733 > = [];734 const units = new Map([["notifyOnce", unit("notifyOnce", calls, { expiresAfterMs: 30 })]]);735 const fixture = harness({ units });736 await fixture.ledger.arm();737 const handle = fixture.ledger.perform<Row>(738 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "late" } },739 { units: [units.get("notifyOnce")!] },740 );741 await handle;742 // The device stays away past the delivery window; the write then syncs.743 fixture.clock.value = 5000;744 fixture.db.settleGlobal();745 await flush();746 assert(String(handle.stage) === "synced", "the write itself settled");747 assertCount(calls.length, 0, "the write happened, so no handler may fire past the window");748 assertCount(fixture.diagnostics.expiredObligations, 1, "the retirement must be diagnosable");749 await fixture.ledger.flush();750 assert(751 (fixture.storage.text() ?? "").includes(handle.writeId) === false,752 "a retired obligation makes the entry prunable",753 );754 fixture.ledger.dispose();755});756757Deno.test("a PermanentEffectError retires the obligation on the first failure", async () => {758 const storage = createMemoryJournalStorage();759 const calls: Array<760 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }761 > = [];762 // A high attempt bound proves the retirement is the permanent signal, not763 // the attempt count: an ordinary throw would re-arm four more times.764 const refusing = unit("permanent", calls, { failPermanent: true, maxAttempts: 5 });765 const fixture = harness({ storage, units: new Map([["permanent", refusing]]) });766 await fixture.ledger.arm();767 const handle = fixture.ledger.perform<Row>(768 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "gone" } },769 { units: [refusing] },770 );771 await handle;772 fixture.db.settleGlobal();773 await flush();774 await fixture.ledger.flush();775 assertCount(fixture.diagnostics.effectHandlerFailures, 1, "the failure is counted once");776 assertCount(777 fixture.diagnostics.quarantinedObligations,778 1,779 "a permanent failure quarantines without exhausting attempts",780 );781 assert(782 (storage.text() ?? "").includes(handle.writeId) === false,783 "the retired obligation makes the entry prunable, so it never re-arms",784 );785 assertCount(calls.length, 0, "a permanently refused handler never reports success");786 fixture.ledger.dispose();787});788789Deno.test("quarantine retires a permanently failing handler after its attempt bound", async () => {790 const storage = createMemoryJournalStorage();791 const calls: Array<792 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }793 > = [];794 const failing = unit("fragile", calls, { failAlways: true, maxAttempts: 2 });795 const first = harness({ storage, units: new Map([["fragile", failing]]) });796 await first.ledger.arm();797 const handle = first.ledger.perform<Row>(798 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "boom" } },799 { units: [failing] },800 );801 await handle;802 first.db.settleGlobal();803 await flush();804 await first.ledger.flush();805 assertCount(first.diagnostics.effectHandlerFailures, 1, "the first failure is counted");806 assert((storage.text() ?? "").includes("fragile"), "the failed obligation re-arms at boot");807 first.ledger.dispose();808809 const second = harness({ storage, units: new Map([["fragile", failing]]) });810 second.db.rows = [];811 await second.ledger.arm();812 await flush();813 assertCount(814 second.diagnostics.quarantinedObligations,815 1,816 "the attempt bound must quarantine the obligation",817 );818 await second.ledger.flush();819 assert(820 (second.storage.text() ?? "").includes(handle.writeId) === false,821 "a quarantined obligation makes the entry prunable",822 );823 assertCount(calls.length, 0, "a quarantined handler never reports success");824 second.ledger.dispose();825});826827Deno.test("a chained child uses one retained deterministic write across replay", async () => {828 const fixture = harness();829 await fixture.ledger.arm();830 const descriptor: MutationDescriptor = {831 verbName: "createChild",832 op: { kind: "insert", table: table as TableProxy<unknown, unknown> },833 units: [],834 expiresAfterMs: null,835 };836 const parentJournalId = "parent-write:chain#0";837 await fixture.ledger.performChainedVerb(descriptor, [{ title: "child" }], parentJournalId);838 assertCount(fixture.db.nextBatch, 1, "the first delivery must issue one child mutation");839 const writeId = `chain:${parentJournalId}`;840 const first = JSON.parse(fixture.storage.text() ?? "{}") as JournalDocument;841 assert(842 first.writes[writeId]?.retainedBy === parentJournalId,843 "the child record must remain retained until its parent obligation commits",844 );845846 await fixture.ledger.performChainedVerb(descriptor, [{ title: "child" }], parentJournalId);847 assertCount(fixture.db.nextBatch, 1, "replay must observe the retained child, not issue another");848 fixture.ledger.dispose();849});