1import type { Db, MutationErrorEvent, TableProxy } from "jazz-tools";2import { createDiagnostics } from "./diagnostics.ts";3import type { EffectContext, EffectUnit } from "../schema/effects.ts";4import { assert, assertCount } from "./test-assert.ts";5import {6 createMemoryJournalStorage,7 emptyJournal,8 hashJournalValue,9 type JournalDocument,10 journalIdFor,11} from "./write-journal.ts";12import { WriteLedger, type WriteLedgerEnvironment } from "./write-ledger.ts";1314type Row = { id: string; title: string };15type Init = Omit<Row, "id">;16const table = { _table: "records", _schema: {} } as TableProxy<Row, Init>;1718class FakeRejection extends Error {19 override readonly name = "PersistedWriteRejectedError";20 constructor(readonly code: string, readonly reason: string) {21 super(`${code}: ${reason}`);22 }23}2425type Deferred = { resolve(): void; reject(error: unknown): void };2627class FakeDb {28 mutationListeners = new Set<(event: MutationErrorEvent) => void>();29 globalWaits: Deferred[] = [];30 rows: Array<Record<string, unknown>> = [];31 refuseNext: Error | null = null;32 nextId = 0;33 nextBatch = 0;34 reads = 0;3536 value(): Db {37 return this as unknown as Db;38 }3940 onMutationError(listener: (event: MutationErrorEvent) => void): () => void {41 this.mutationListeners.add(listener);42 return () => this.mutationListeners.delete(listener);43 }4445 emit(event: MutationErrorEvent): void {46 for (const listener of this.mutationListeners) listener(event);47 }4849 insert(_table: TableProxy<Row, Init>, values: Init) {50 if (this.refuseNext) {51 const error = this.refuseNext;52 this.refuseNext = null;53 throw error;54 }55 const row = { id: `row-${++this.nextId}`, ...values };56 return { value: row, ...this.handle() };57 }5859 update(_table: TableProxy<Row, Init>, _id: string, _patch: Partial<Init>) {60 return this.handle();61 }6263 delete(_table: TableProxy<Row, Init>, _id: string) {64 return this.handle();65 }6667 all(): Promise<unknown[]> {68 this.reads += 1;69 return Promise.resolve(this.rows);70 }7172 settleGlobal(index = 0): void {73 this.globalWaits[index]?.resolve();74 }7576 rejectGlobal(error: unknown, index = 0): void {77 this.globalWaits[index]?.reject(error);78 }7980 private handle() {81 return {82 batchId: `batch-${++this.nextBatch}`,83 wait: ({ tier }: { tier: "local" | "global" }) => {84 if (tier === "local") return Promise.resolve(undefined);85 return new Promise<unknown>((resolve, reject) => {86 this.globalWaits.push({ resolve: () => resolve(undefined), reject });87 });88 },89 };90 }91}9293type Harness = {94 db: FakeDb;95 ledger: WriteLedger;96 diagnostics: ReturnType<typeof createDiagnostics>;97 storage: ReturnType<typeof createMemoryJournalStorage>;98 units: Map<string, EffectUnit<{ id: string }>>;99 clock: { value: number };100};101102function harness(options: {103 syncConfigured?: boolean;104 storage?: ReturnType<typeof createMemoryJournalStorage>;105 units?: Map<string, EffectUnit<{ id: string }>>;106 db?: FakeDb;107 guardWrite?: () => Promise<(() => void) | void>;108 sweepIntervalMs?: number;109} = {}): Harness {110 const db = options.db ?? new FakeDb();111 const diagnostics = createDiagnostics();112 const storage = options.storage ?? createMemoryJournalStorage();113 const units = options.units ?? new Map<string, EffectUnit<{ id: string }>>();114 const clock = { value: 1000 };115 const environment: WriteLedgerEnvironment = {116 getDb: () => Promise.resolve(db.value()),117 syncConfigured: () => options.syncConfigured ?? true,118 storage,119 resolveTable: (name) => (name === "records" ? { where: (input) => input } : null),120 resolveEffectUnit: (name) => units.get(name) ?? null,121 subscribeRuntimeRecreation: () => () => undefined,122 updateDiagnostics: (update) => update(diagnostics),123 now: () => clock.value,124 retryDelayMs: () => 1,125 probeTimeoutMs: 50,126 sweepIntervalMs: options.sweepIntervalMs ?? 60_000,127 ...(options.guardWrite ? { guardWrite: options.guardWrite } : {}),128 };129 return { db, ledger: new WriteLedger(environment), diagnostics, storage, units, clock };130}131132function unit(133 name: string,134 calls: Array<{ handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }>,135 options: {136 failFirst?: { onSynced?: boolean };137 failAlways?: boolean;138 expiresAfterMs?: number;139 maxAttempts?: number;140 } = {},141): EffectUnit<{ id: string }> {142 let failed = false;143 return {144 effectName: name,145 handlers: {146 onSynced: (row, context) => {147 if (options.failAlways) throw new Error("handler always explodes");148 if (options.failFirst?.onSynced && !failed) {149 failed = true;150 throw new Error("handler exploded");151 }152 calls.push({ handler: "onSynced", row, context });153 },154 onRejected: (row, context) => {155 calls.push({ handler: "onRejected", row, context });156 },157 },158 expiresAfterMs: options.expiresAfterMs ?? null,159 ...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),160 };161}162163async function flush(times = 6): Promise<void> {164 for (let i = 0; i < times; i += 1) await Promise.resolve();165}166167async function settleTimers(ms = 10): Promise<void> {168 await new Promise((resolve) => setTimeout(resolve, ms));169}170171Deno.test("a verb write progresses saved to synced and fires onSynced exactly once", async () => {172 const calls: Array<173 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }174 > = [];175 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);176 const fixture = harness({ units });177 await fixture.ledger.arm();178 const handle = fixture.ledger.perform<Row>(179 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "one" } },180 { verb: "placeOrder", units: [units.get("chargeCard")!] },181 );182 const row = await handle;183 assert(row.title === "one", "await must resolve the created row at saved");184 assert(handle.stage === "saved", "the write must rest at saved until the store confirms");185 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "a saved write must count as pending");186 assert(fixture.ledger.rowStatus(row.id) === "waiting", "an unsettled row must read waiting");187 fixture.db.settleGlobal();188 await flush();189 assert(String(handle.stage) === "synced", "store confirmation must reach the handle");190 assertCount(calls.length, 1, "onSynced must fire exactly once");191 assert(calls[0].handler === "onSynced", "confirmation must run the action handler");192 assert(193 calls[0].context.journalId === journalIdFor(handle.writeId, "chargeCard"),194 "the handler must receive the (write id, effect name) idempotency key",195 );196 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a settled write must leave pending");197 assert(fixture.ledger.rowStatus(row.id) === "synced", "a confirmed row must read synced");198 await fixture.ledger.flush();199 assert(fixture.storage.text()?.includes(handle.writeId) === false, "settled entries must prune");200 fixture.ledger.dispose();201});202203Deno.test("a synchronous local refusal throws from the call and never journals", async () => {204 const fixture = harness();205 await fixture.ledger.arm();206 fixture.db.refuseNext = new Error("current policy refuses this write");207 let thrown: Error | null = null;208 try {209 fixture.ledger.perform<Row>(210 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "no" } },211 );212 } catch (error) {213 thrown = error as Error;214 }215 assert(thrown?.message === "current policy refuses this write", "the refusal must throw");216 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a refusal must not journal");217 await fixture.ledger.flush();218 assert(219 (fixture.storage.text() ?? "").includes("records") === false,220 "a refusal must leave no journal entry",221 );222 assertCount(fixture.diagnostics.mutationErrors, 1, "the refusal must still be diagnosable");223 fixture.ledger.dispose();224});225226Deno.test("a permanent verdict settles rejected and fires onRejected exactly once", async () => {227 const calls: Array<228 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }229 > = [];230 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);231 const fixture = harness({ units });232 await fixture.ledger.arm();233 const handle = fixture.ledger.perform<Row>(234 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "denied" } },235 { verb: "placeOrder", units: [units.get("chargeCard")!] },236 );237 const row = await handle;238 fixture.db.rejectGlobal(new FakeRejection("permission_denied", "policy tightened"));239 await flush();240 assert(handle.stage === "rejected", "a permanent verdict must settle the stage");241 assert(handle.reason?.code === "permission_denied", "the verdict code must be readable");242 assertCount(calls.length, 1, "onRejected must fire exactly once");243 assert(calls[0].handler === "onRejected", "compensation must run, not the action");244 assert(calls[0].context.fate === "rejected", "the handler must see the rejected fate");245 assert(fixture.ledger.rowStatus(row.id) === "rejected", "the row must read rejected");246 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a rejected write is not pending");247 fixture.ledger.dispose();248});249250Deno.test("a transient failure keeps the write pending and never compensates", async () => {251 const calls: Array<252 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }253 > = [];254 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);255 const fixture = harness({ units });256 await fixture.ledger.arm();257 const handle = fixture.ledger.perform<Row>(258 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "slow" } },259 { units: [units.get("chargeCard")!] },260 );261 await handle;262 fixture.db.rejectGlobal(new FakeRejection("mystery_code", "unknown"));263 await flush();264 assert(handle.stage === "saved", "an uninterpretable code must keep the write pending");265 assertCount(calls.length, 0, "no compensation on a non-permanent code");266 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "the write must stay pending");267 // The scheduled re-derivation asks the store again; confirm it settles.268 await settleTimers();269 fixture.db.settleGlobal(1);270 await flush();271 assert(String(handle.stage) === "synced", "a later confirmation must settle the pending write");272 assertCount(calls.filter((call) => call.handler === "onSynced").length, 1, "then onSynced runs");273 fixture.ledger.dispose();274});275276Deno.test("reload re-arm: journaled obligations fire after the journal is reloaded", async () => {277 const storage = createMemoryJournalStorage();278 // First run: the write settles as synced, but the unit is not registered279 // yet (its declaring module never loaded), so the obligation stays pending.280 const first = harness({ storage });281 await first.ledger.arm();282 const handle = first.ledger.perform<Row>(283 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "kept" } },284 {285 verb: "placeOrder",286 units: [{ effectName: "notify", handlers: {} } as EffectUnit<{ id: string }>],287 },288 );289 await handle;290 first.db.settleGlobal();291 await flush();292 await first.ledger.flush();293 first.ledger.dispose();294 assert(295 (storage.text() ?? "").includes(handle.writeId),296 "the unsatisfied obligation must survive in the persisted journal",297 );298299 const calls: Array<300 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }301 > = [];302 const units = new Map([["notify", unit("notify", calls)]]);303 const second = harness({ storage, units });304 await second.ledger.arm();305 await flush();306 assertCount(calls.length, 1, "the reloaded runtime must re-run the journaled obligation");307 assert(calls[0].handler === "onSynced", "the synced fate must run the action handler");308 assert(309 calls[0].context.journalId === journalIdFor(handle.writeId, "notify"),310 "the idempotency key must survive the reload",311 );312 await second.ledger.flush();313 assert((second.storage.text() ?? "").includes(handle.writeId) === false, "then the entry prunes");314 second.ledger.dispose();315});316317Deno.test("boot reconciliation confirms journaled writes against store state", async () => {318 const calls: Array<319 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }320 > = [];321 const units = new Map([["celebrate", unit("celebrate", calls)]]);322 const document: JournalDocument = emptyJournal();323 document.writes["w-confirmed"] = {324 writeId: "w-confirmed",325 verb: "addTask",326 table: "records",327 op: "insert",328 rowId: "row-a",329 batchId: "batch-a",330 rowHashes: { title: hashJournalValue(document.hashKey, "kept") },331 stage: "saved",332 cause: null,333 code: null,334 reason: null,335 createdAt: 1,336 expiresAt: null,337 effects: {338 celebrate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },339 },340 };341 document.writes["w-waiting"] = {342 writeId: "w-waiting",343 verb: null,344 table: "records",345 op: "insert",346 rowId: "row-b",347 batchId: "batch-b",348 rowHashes: { title: hashJournalValue(document.hashKey, "not there yet") },349 stage: "saved",350 cause: null,351 code: null,352 reason: null,353 createdAt: 2,354 expiresAt: null,355 effects: {},356 };357 const storage = createMemoryJournalStorage(JSON.stringify(document));358 const fixture = harness({ storage, units });359 fixture.db.rows = [{ id: "row-a", title: "kept" }];360 await fixture.ledger.arm();361 await flush();362 await settleTimers();363 assert(364 fixture.ledger.rowStatus("row-a") === "synced",365 "a journaled write the store confirms must settle as synced at boot",366 );367 assertCount(calls.length, 1, "the re-armed effect must fire after the store confirms");368 assert(369 fixture.ledger.rowStatus("row-b") === "waiting",370 "a journaled write the store does not confirm must stay pending",371 );372 assertCount(fixture.ledger.getPendingSnapshot().count, 1, "only the unconfirmed write pends");373 fixture.ledger.dispose();374});375376Deno.test("boot reconciliation applies replayed verdicts to journaled writes", async () => {377 const calls: Array<378 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }379 > = [];380 const units = new Map([["compensate", unit("compensate", calls)]]);381 const document: JournalDocument = emptyJournal();382 document.writes["w-denied"] = {383 writeId: "w-denied",384 verb: "addTask",385 table: "records",386 op: "insert",387 rowId: "row-x",388 batchId: "batch-x",389 rowHashes: { title: hashJournalValue(document.hashKey, "denied") },390 stage: "saved",391 cause: null,392 code: null,393 reason: null,394 createdAt: 1,395 expiresAt: null,396 effects: {397 compensate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },398 },399 };400 const storage = createMemoryJournalStorage(JSON.stringify(document));401 const fixture = harness({ storage, units });402 await fixture.ledger.arm();403 fixture.db.emit(404 {405 code: "permission_denied",406 reason: "policy tightened while offline",407 batch: { batchId: "batch-x" },408 } as unknown as MutationErrorEvent,409 );410 await flush();411 assertCount(calls.length, 1, "the replayed verdict must run compensation once");412 assert(calls[0].handler === "onRejected", "a permanent verdict runs onRejected");413 assert(fixture.ledger.rowStatus("row-x") === "rejected", "the row must read rejected");414 fixture.db.emit(415 {416 code: "permission_denied",417 reason: "duplicate replay",418 batch: { batchId: "batch-x" },419 } as unknown as MutationErrorEvent,420 );421 await flush();422 assertCount(calls.length, 1, "a duplicate verdict must be recorded idempotently");423 fixture.ledger.dispose();424});425426Deno.test("an unknown replayed code keeps the journaled write pending", async () => {427 const document: JournalDocument = emptyJournal();428 document.writes["w-odd"] = {429 writeId: "w-odd",430 verb: null,431 table: "records",432 op: "update",433 rowId: "row-y",434 batchId: "batch-y",435 rowHashes: { title: hashJournalValue(document.hashKey, "edited") },436 stage: "saved",437 cause: null,438 code: null,439 reason: null,440 createdAt: 1,441 expiresAt: null,442 effects: {},443 };444 const storage = createMemoryJournalStorage(JSON.stringify(document));445 const fixture = harness({ storage });446 await fixture.ledger.arm();447 fixture.db.emit(448 {449 code: "mystery",450 reason: "unclassifiable",451 batch: { batchId: "batch-y" },452 } as unknown as MutationErrorEvent,453 );454 await flush();455 assert(456 fixture.ledger.rowStatus("row-y") === "waiting",457 "an uninterpretable code must never settle or compensate a write",458 );459 fixture.ledger.dispose();460});461462Deno.test("a remove is confirmed at boot by the row's absence", async () => {463 const document: JournalDocument = emptyJournal();464 document.writes["w-gone"] = {465 writeId: "w-gone",466 verb: null,467 table: "records",468 op: "remove",469 rowId: "row-z",470 batchId: "batch-z",471 rowHashes: {},472 stage: "saved",473 cause: null,474 code: null,475 reason: null,476 createdAt: 1,477 expiresAt: null,478 effects: {},479 };480 const storage = createMemoryJournalStorage(JSON.stringify(document));481 const fixture = harness({ storage });482 fixture.db.rows = [];483 await fixture.ledger.arm();484 await flush();485 await settleTimers();486 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "the absent row confirms the remove");487 fixture.ledger.dispose();488});489490Deno.test("effect units are independent: one failing handler never blocks another", async () => {491 const calls: Array<492 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }493 > = [];494 const units = new Map<string, EffectUnit<{ id: string }>>([495 ["fragile", unit("fragile", calls, { failFirst: { onSynced: true } })],496 ["sturdy", unit("sturdy", calls)],497 ]);498 const fixture = harness({ units });499 await fixture.ledger.arm();500 const handle = fixture.ledger.perform<Row>(501 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "both" } },502 { units: [units.get("fragile")!, units.get("sturdy")!] },503 );504 await handle;505 fixture.db.settleGlobal();506 await flush();507 assertCount(508 calls.filter((call) => call.handler === "onSynced").length,509 1,510 "the sturdy unit must run even though the fragile one failed",511 );512 assertCount(fixture.diagnostics.effectHandlerFailures, 1, "the failure must be diagnosable");513 await fixture.ledger.flush();514 assert(515 (fixture.storage.text() ?? "").includes("fragile"),516 "the failed obligation must stay journaled so the next boot re-arms it",517 );518 fixture.ledger.dispose();519});520521Deno.test("without configured sync, local durability is settlement and effects run", async () => {522 const calls: Array<523 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }524 > = [];525 const units = new Map([["local-note", unit("local-note", calls)]]);526 const fixture = harness({ syncConfigured: false, units });527 await fixture.ledger.arm();528 const handle = fixture.ledger.perform<Row>(529 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "solo" } },530 { units: [units.get("local-note")!] },531 );532 await handle;533 await flush();534 assert(handle.stage === "synced", "a device without a sync location settles at saved");535 assertCount(calls.length, 1, "effects must run when the write settles");536 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "nothing waits without a store");537 fixture.ledger.dispose();538});539540Deno.test("a write-guard refusal fails the handle and never journals or compensates", async () => {541 const calls: Array<542 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }543 > = [];544 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);545 const fixture = harness({546 units,547 guardWrite: () => Promise.reject(new Error("schema is ahead of this bundle")),548 });549 await fixture.ledger.arm();550 const handle = fixture.ledger.perform<Row>(551 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "held" } },552 { verb: "placeOrder", units: [units.get("chargeCard")!] },553 );554 const error = await handle.saved.then(() => null, (thrown) => thrown as Error);555 assert(556 error?.message === "schema is ahead of this bundle",557 "the guard refusal must surface through the verb call",558 );559 assert(handle.stage !== "rejected", "a refusal is not an adjudicated verdict");560 assertCount(fixture.ledger.getPendingSnapshot().count, 0, "a refused write must not journal");561 await fixture.ledger.flush();562 assert(563 (fixture.storage.text() ?? "").includes("held") === false,564 "a refused write must leave no journal entry",565 );566 assertCount(calls.length, 0, "a refused write must fire no effect");567 assertCount(fixture.diagnostics.mutationErrors, 1, "the refusal must be diagnosable");568 assertCount(fixture.db.nextId, 0, "the write must never reach the database");569 fixture.ledger.dispose();570});571572Deno.test("the write-guard release covers exactly the local-durability window", async () => {573 let released = 0;574 const fixture = harness({575 guardWrite: () => Promise.resolve(() => void (released += 1)),576 });577 await fixture.ledger.arm();578 const handle = fixture.ledger.perform<Row>(579 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "locked" } },580 );581 await handle;582 await flush();583 assertCount(released, 1, "the lock must release once the write settles locally");584 fixture.db.settleGlobal();585 await flush();586 assertCount(released, 1, "global settlement must not release the lock again");587 fixture.ledger.dispose();588});589590Deno.test("the serialized journal holds hashes, never plaintext column values", async () => {591 const fixture = harness();592 await fixture.ledger.arm();593 const handle = fixture.ledger.perform<Row>(594 {595 kind: "insert",596 table: table as TableProxy<unknown, unknown>,597 values: { title: "SECRET-VALUE-XYZ" },598 },599 );600 await handle;601 await fixture.ledger.flush();602 const persisted = fixture.storage.text() ?? "";603 assert(persisted.includes(handle.writeId), "the pending write must be journaled");604 assert(605 persisted.includes("SECRET-VALUE-XYZ") === false,606 "no plaintext column value may reach the persisted journal",607 );608 assert(persisted.includes("rowHashes"), "the equality-probe hashes must be journaled");609 fixture.ledger.dispose();610});611612Deno.test("boot compensation receives write identity only; live handlers keep the row", async () => {613 const calls: Array<614 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }615 > = [];616 const units = new Map([["compensate", unit("compensate", calls)]]);617 const document: JournalDocument = emptyJournal();618 document.writes["w-back"] = {619 writeId: "w-back",620 verb: "placeOrder",621 table: "records",622 op: "insert",623 rowId: "row-q",624 batchId: "batch-q",625 rowHashes: { title: hashJournalValue(document.hashKey, "gone") },626 stage: "saved",627 cause: null,628 code: null,629 reason: null,630 createdAt: 1,631 expiresAt: null,632 effects: {633 compensate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },634 },635 };636 const storage = createMemoryJournalStorage(JSON.stringify(document));637 const fixture = harness({ storage, units });638 await fixture.ledger.arm();639 fixture.db.emit(640 {641 code: "permission_denied",642 reason: "tightened while away",643 batch: { batchId: "batch-q" },644 } as unknown as MutationErrorEvent,645 );646 await flush();647 assertCount(calls.length, 1, "the replayed verdict must run compensation");648 assert(649 Object.keys(calls[0].row).join(",") === "id" && calls[0].row.id === "row-q",650 "after a reload the journal holds no values: compensation receives identity only",651 );652 assert(calls[0].context.cause === "denied", "the structured cause must reach the handler");653 fixture.ledger.dispose();654});655656Deno.test("a synced obligation re-armed at boot receives the live store row", async () => {657 const calls: Array<658 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }659 > = [];660 const units = new Map([["celebrate", unit("celebrate", calls)]]);661 const document: JournalDocument = emptyJournal();662 document.writes["w-live"] = {663 writeId: "w-live",664 verb: "addTask",665 table: "records",666 op: "insert",667 rowId: "row-l",668 batchId: "batch-l",669 rowHashes: { title: hashJournalValue(document.hashKey, "kept") },670 stage: "synced",671 cause: null,672 code: null,673 reason: null,674 createdAt: 1,675 expiresAt: null,676 effects: {677 celebrate: { status: "pending", attempts: 0, lastError: null, expiresAt: null },678 },679 };680 const storage = createMemoryJournalStorage(JSON.stringify(document));681 const fixture = harness({ storage, units });682 fixture.db.rows = [{ id: "row-l", title: "kept" }];683 await fixture.ledger.arm();684 await flush();685 assertCount(calls.length, 1, "the re-armed obligation must run");686 assert(687 (calls[0].row as { title?: string }).title === "kept",688 "a synced write's handler must receive the live row the store confirmed",689 );690 fixture.ledger.dispose();691});692693Deno.test("an overdue intent surfaces in pending state and diagnostics without compensation", async () => {694 const calls: Array<695 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }696 > = [];697 const units = new Map([["chargeCard", unit("chargeCard", calls)]]);698 const fixture = harness({ units, sweepIntervalMs: 5 });699 await fixture.ledger.arm();700 const handle = fixture.ledger.perform<Row>(701 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "slow" } },702 { verb: "placeOrder", units: [units.get("chargeCard")!], expiresAfterMs: 50 },703 );704 await handle;705 assert(706 fixture.ledger.getPendingSnapshot().writes[0]?.expired === false,707 "an intent inside its lifespan is not overdue",708 );709 fixture.clock.value = 2000;710 await settleTimers(20);711 const snapshot = fixture.ledger.getPendingSnapshot();712 assert(snapshot.writes[0]?.expired === true, "the overdue intent must surface");713 assertCount(fixture.diagnostics.expiredPendingWrites, 1, "diagnostics must count it");714 assert(handle.stage === "saved", "no withdrawal exists, so the stage must not change");715 assertCount(calls.length, 0, "no compensation may fire for a write that can still sync");716 // The store later confirms the overdue write: it settles normally.717 fixture.db.settleGlobal();718 await flush();719 assert(String(handle.stage) === "synced", "a late confirmation still settles the write");720 fixture.ledger.dispose();721});722723Deno.test("a closed delivery window retires the obligation: no handler, diagnosable, prunable", async () => {724 const calls: Array<725 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }726 > = [];727 const units = new Map([["notifyOnce", unit("notifyOnce", calls, { expiresAfterMs: 30 })]]);728 const fixture = harness({ units });729 await fixture.ledger.arm();730 const handle = fixture.ledger.perform<Row>(731 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "late" } },732 { units: [units.get("notifyOnce")!] },733 );734 await handle;735 // The device stays away past the delivery window; the write then syncs.736 fixture.clock.value = 5000;737 fixture.db.settleGlobal();738 await flush();739 assert(String(handle.stage) === "synced", "the write itself settled");740 assertCount(calls.length, 0, "the write happened, so no handler may fire past the window");741 assertCount(fixture.diagnostics.expiredObligations, 1, "the retirement must be diagnosable");742 await fixture.ledger.flush();743 assert(744 (fixture.storage.text() ?? "").includes(handle.writeId) === false,745 "a retired obligation makes the entry prunable",746 );747 fixture.ledger.dispose();748});749750Deno.test("quarantine retires a permanently failing handler after its attempt bound", async () => {751 const storage = createMemoryJournalStorage();752 const calls: Array<753 { handler: "onSynced" | "onRejected"; row: { id: string }; context: EffectContext }754 > = [];755 const failing = unit("fragile", calls, { failAlways: true, maxAttempts: 2 });756 const first = harness({ storage, units: new Map([["fragile", failing]]) });757 await first.ledger.arm();758 const handle = first.ledger.perform<Row>(759 { kind: "insert", table: table as TableProxy<unknown, unknown>, values: { title: "boom" } },760 { units: [failing] },761 );762 await handle;763 first.db.settleGlobal();764 await flush();765 await first.ledger.flush();766 assertCount(first.diagnostics.effectHandlerFailures, 1, "the first failure is counted");767 assert((storage.text() ?? "").includes("fragile"), "the failed obligation re-arms at boot");768 first.ledger.dispose();769770 const second = harness({ storage, units: new Map([["fragile", failing]]) });771 second.db.rows = [];772 await second.ledger.arm();773 await flush();774 assertCount(775 second.diagnostics.quarantinedObligations,776 1,777 "the attempt bound must quarantine the obligation",778 );779 await second.ledger.flush();780 assert(781 (second.storage.text() ?? "").includes(handle.writeId) === false,782 "a quarantined obligation makes the entry prunable",783 );784 assertCount(calls.length, 0, "a quarantined handler never reports success");785 second.ledger.dispose();786});