From 45d34182fac3938bb6e8e1d6b3b85dddc3d2313d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 24 Aug 2026 10:18:31 -0700 Subject: [PATCH 1/2] feat: make the at-least-once duplicate observable at a sink Implements #23. The delivery contract promises at-least-once and the docs tell external systems to hold stable idempotency keys, but no artifact showed the duplicate arriving anywhere with deduplication deliberately off. A contract clause nobody can observe firing is decoration; this demo lets anyone watch the clause fire and watch the documented remedy absorb it. examples/at-least-once stages one actor turn whose effect writes to an external sink file. The first effect worker crashes between the sink write and the acknowledgement; a second worker reclaims the stale effect after the liveness threshold and delivers again. With deduplication off the sink reads 2, both deliveries carrying the same stable effect id at attempts 1 and 2. With a guard on that id the sink reads 1. The actor state commits exactly once in both runs, which is the sharpest line of the proof: the state machine kept its exactly-once story while the outside world saw two. pnpm run test:at-least-once runs it; CI runs it beside the recovery demo in both the quality and floor jobs; docs/correctness.md links it from the at-least-once limitation it makes observable. The sink module carries unit coverage for both guard modes. --- .github/workflows/ci.yml | 2 + CHANGELOG.md | 12 +++ docs/correctness.md | 7 +- examples/at-least-once/actor.ts | 13 +++ examples/at-least-once/demo.ts | 129 ++++++++++++++++++++++++ examples/at-least-once/effect-worker.ts | 61 +++++++++++ examples/at-least-once/sink.ts | 33 ++++++ package.json | 1 + test/at-least-once-sink.test.ts | 69 +++++++++++++ 9 files changed, 326 insertions(+), 1 deletion(-) create mode 100644 examples/at-least-once/actor.ts create mode 100644 examples/at-least-once/demo.ts create mode 100644 examples/at-least-once/effect-worker.ts create mode 100644 examples/at-least-once/sink.ts create mode 100644 test/at-least-once-sink.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ce895c1..1e9bb07 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,7 @@ jobs: - run: pnpm run pack:check - run: pnpm run test:package - run: pnpm run test:recovery + - run: pnpm run test:at-least-once - run: pnpm audit --audit-level=high floor: @@ -42,6 +43,7 @@ jobs: - run: pnpm run build - run: pnpm run test:package - run: pnpm run test:recovery + - run: pnpm run test:at-least-once postgresql: runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 72bcfc8..6b7bc66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +- Add `examples/at-least-once` and `pnpm run test:at-least-once` + ([#23](https://github.com/cardmagic/solid-objects-js/issues/23)): an + executable proof that the at-least-once clause fires and that the + documented remedy absorbs it. An effect worker crashes between the + external sink write and the acknowledgement; after restart the sink + reads 2 with deduplication off, and 1 when a guard on the stable + effect id is in place. The state commit happens exactly once in both + runs, and both deliveries carry the same effect id. CI runs the demo + alongside the recovery demo. + ## 0.14.1 - 2026-08-23 - Add `solid-objects/signals`, live signals on actor references diff --git a/docs/correctness.md b/docs/correctness.md index ab9a748..55250d3 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -58,7 +58,12 @@ - At-least-once execution means actor code may begin more than once. State and staged intents from a failed turn roll back, but arbitrary external work does - not. External systems need stable idempotency keys. + not. External systems need stable idempotency keys. This clause is + observable, not decorative: `pnpm run test:at-least-once` crashes an + effect worker between the sink write and the acknowledgement, restarts + it, and shows the sink reading 2 with deduplication off — then shows a + guard on the stable effect id absorbing the same duplicate, with the + sink reading 1. The state commit happens exactly once in both runs. - The activation fence protects the Solid Objects commit. It cannot revoke or undo network calls, files, emails, payments, or other external effects. - One identity processes one write operation at a time. This is the ordering diff --git a/examples/at-least-once/actor.ts b/examples/at-least-once/actor.ts new file mode 100644 index 0000000..167a2ef --- /dev/null +++ b/examples/at-least-once/actor.ts @@ -0,0 +1,13 @@ +import { Actor } from "solid-objects" + +export class DeliveryCounter extends Actor { + static override readonly actorType = "DeliveryCounter" + + count = 0 + + deliver(): number { + this.count += 1 + this.emit("record", { arguments: {} }) + return this.count + } +} diff --git a/examples/at-least-once/demo.ts b/examples/at-least-once/demo.ts new file mode 100644 index 0000000..e3c2c3d --- /dev/null +++ b/examples/at-least-once/demo.ts @@ -0,0 +1,129 @@ +import assert from "node:assert/strict" +import { fork, type ChildProcess } from "node:child_process" +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { fileURLToPath } from "node:url" +import { createRuntime } from "solid-objects" +import { sqlite } from "solid-objects/database/sqlite" +import { DeliveryCounter } from "./actor.ts" +import { readSink } from "./sink.ts" + +const directory = await mkdtemp(join(tmpdir(), "solid-objects-at-least-once-")) +const databasePath = join(directory, "state.sqlite3") +const runtime = createRuntime({ + database: sqlite({ path: databasePath, timeoutMilliseconds: 2_000, lockRetryAttempts: 20 }), + leaseDurationMilliseconds: 250, + leaseRenewalIntervalMilliseconds: 50, + processHeartbeatIntervalMilliseconds: 75, + processAliveThresholdMilliseconds: 300, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeAdministration: () => true, +}) + +try { + runtime.register(DeliveryCounter) + await runtime.install() + const duplicate = await proveDuplicateAtSink() + const remedy = await proveDeduplicationAbsorbsIt() + process.stdout.write(`${JSON.stringify({ duplicate, remedy }, null, 2)}\n`) +} finally { + await runtime.close() + await rm(directory, { recursive: true }) +} + +async function proveDuplicateAtSink(): Promise<{ + stateCommits: number + sinkDeliveries: number + sameEffectId: boolean + attempts: number[] +}> { + const sinkPath = join(directory, "sink-dedup-off.json") + await stageOneDelivery("dedup-off") + await crashThenRecover({ sinkPath, deduplicate: "off" }) + + const sink = await readSink(sinkPath) + const snapshot = await runtime.ref(DeliveryCounter, "dedup-off").snapshot() + assert.equal(snapshot.count, 1, "the state commit happened exactly once") + assert.equal(sink.deliveries.length, 2, "the sink observed the duplicate") + assert.equal( + sink.deliveries[0]?.effectId, + sink.deliveries[1]?.effectId, + "both deliveries carried the same stable effect id", + ) + return { + stateCommits: snapshot.count, + sinkDeliveries: sink.deliveries.length, + sameEffectId: sink.deliveries[0]?.effectId === sink.deliveries[1]?.effectId, + attempts: sink.deliveries.map((delivery) => delivery.attempt), + } +} + +async function proveDeduplicationAbsorbsIt(): Promise<{ + stateCommits: number + sinkDeliveries: number +}> { + const sinkPath = join(directory, "sink-dedup-on.json") + await stageOneDelivery("dedup-on") + await crashThenRecover({ sinkPath, deduplicate: "on" }) + + const sink = await readSink(sinkPath) + const snapshot = await runtime.ref(DeliveryCounter, "dedup-on").snapshot() + assert.equal(snapshot.count, 1, "the state commit happened exactly once") + assert.equal(sink.deliveries.length, 1, "the stable effect id absorbed the duplicate") + return { stateCommits: snapshot.count, sinkDeliveries: sink.deliveries.length } +} + +async function stageOneDelivery(actorId: string): Promise { + const message = await runtime.ref(DeliveryCounter, actorId).send.deliver() + const worker = runtime.worker() + try { + let processed = 0 + for (let attempt = 0; attempt < 200 && processed === 0; attempt += 1) { + processed = await worker.runOnce({ activationRetention: "release" }) + if (processed === 0) await new Promise((resolve) => setTimeout(resolve, 10)) + } + assert.equal(processed, 1, "the actor turn committed and staged the effect") + } finally { + await worker.stop() + } + assert.equal(await message.status(), "completed") +} + +async function crashThenRecover(options: { + sinkPath: string + deduplicate: "on" | "off" +}): Promise { + const crashing = spawnEffectWorker({ ...options, mode: "crash" }) + const crashExit = await crashing.finished + assert.equal(crashExit, 1, "the first delivery crashed before acknowledgement") + + await new Promise((resolve) => setTimeout(resolve, 400)) + + const recovering = spawnEffectWorker({ ...options, mode: "complete" }) + const recoveryExit = await recovering.finished + assert.equal(recoveryExit, 0, "the second delivery completed and acknowledged") +} + +function spawnEffectWorker(options: { + sinkPath: string + deduplicate: "on" | "off" + mode: "crash" | "complete" +}): { child: ChildProcess; finished: Promise } { + const child = fork( + fileURLToPath(new URL("./effect-worker.ts", import.meta.url)), + [databasePath, options.sinkPath, options.mode, options.deduplicate], + { stdio: ["ignore", "inherit", "inherit", "ipc"] }, + ) + const finished = new Promise((resolve, reject) => { + child.once("exit", (code) => resolve(code)) + child.once("error", reject) + }) + return { child, finished } +} diff --git a/examples/at-least-once/effect-worker.ts b/examples/at-least-once/effect-worker.ts new file mode 100644 index 0000000..9314d40 --- /dev/null +++ b/examples/at-least-once/effect-worker.ts @@ -0,0 +1,61 @@ +import { createRuntime } from "solid-objects" +import { sqlite } from "solid-objects/database/sqlite" +import { DeliveryCounter } from "./actor.ts" +import { recordDelivery } from "./sink.ts" + +const databasePath = requiredArgument(2) +const sinkPath = requiredArgument(3) +const mode = requiredArgument(4) +const deduplicate = requiredArgument(5) === "on" + +const runtime = createRuntime({ + database: sqlite({ path: databasePath, timeoutMilliseconds: 2_000, lockRetryAttempts: 20 }), + pollingIntervalMilliseconds: 10, + leaseDurationMilliseconds: 250, + leaseRenewalIntervalMilliseconds: 50, + processHeartbeatIntervalMilliseconds: 75, + processAliveThresholdMilliseconds: 300, + workerCount: 0, + effectWorkerCount: 1, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeAdministration: () => true, +}) + +runtime.register(DeliveryCounter) +runtime.registerEffect("record", async (_argumentsValue, context) => { + const { applied } = await recordDelivery({ + path: sinkPath, + effectId: context.id, + attempt: context.attempt, + deduplicate, + }) + process.send?.({ event: "sink.recorded", effectId: context.id, applied }) + if (mode === "crash") { + process.exit(1) + } + return null +}) +await runtime.install() +const effectWorker = runtime.effectWorker() + +try { + let processed = 0 + for (let attempt = 0; attempt < 200 && processed === 0; attempt += 1) { + processed = await effectWorker.runOnce() + if (processed === 0) await new Promise((resolve) => setTimeout(resolve, 10)) + } + process.send?.({ event: "effects.finished", processed }) +} finally { + await effectWorker.stop() + await runtime.close() +} + +function requiredArgument(index: number): string { + const value = process.argv[index] + if (!value) throw new TypeError(`argument ${index - 1} is required`) + return value +} diff --git a/examples/at-least-once/sink.ts b/examples/at-least-once/sink.ts new file mode 100644 index 0000000..534c83b --- /dev/null +++ b/examples/at-least-once/sink.ts @@ -0,0 +1,33 @@ +import { readFile, writeFile } from "node:fs/promises" + +export interface SinkDelivery { + effectId: string + attempt: number +} + +export interface SinkState { + deliveries: SinkDelivery[] +} + +export async function readSink(path: string): Promise { + try { + const parsed: SinkState = JSON.parse(await readFile(path, "utf-8")) + return parsed + } catch { + return { deliveries: [] } + } +} + +export async function recordDelivery(options: { + path: string + effectId: string + attempt: number + deduplicate: boolean +}): Promise<{ applied: boolean }> { + const sink = await readSink(options.path) + const seen = sink.deliveries.some((delivery) => delivery.effectId === options.effectId) + if (options.deduplicate && seen) return { applied: false } + sink.deliveries.push({ effectId: options.effectId, attempt: options.attempt }) + await writeFile(options.path, JSON.stringify(sink, null, 2)) + return { applied: true } +} diff --git a/package.json b/package.json index a48df50..30a2e70 100644 --- a/package.json +++ b/package.json @@ -107,6 +107,7 @@ "test:mysql": "vitest run test/mysql.test.ts", "test:package": "node scripts/release-artifact-smoke.mjs", "test:recovery": "pnpm run build && node examples/failure-recovery/demo.ts", + "test:at-least-once": "pnpm run build && node examples/at-least-once/demo.ts", "test:redis": "vitest run test/redis-wake-up.test.ts", "test:watch": "vitest", "benchmark": "pnpm run build && node benchmarks/run.ts", diff --git a/test/at-least-once-sink.test.ts b/test/at-least-once-sink.test.ts new file mode 100644 index 0000000..65a18c7 --- /dev/null +++ b/test/at-least-once-sink.test.ts @@ -0,0 +1,69 @@ +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { afterEach, describe, expect, it } from "vitest" +import { readSink, recordDelivery } from "../examples/at-least-once/sink.js" + +let directory: string | undefined + +afterEach(async () => { + if (directory) await rm(directory, { recursive: true, force: true }) + directory = undefined +}) + +async function sinkPath(): Promise { + directory = await mkdtemp(join(tmpdir(), "solid-objects-sink-")) + return join(directory, "sink.json") +} + +describe("at-least-once sink", () => { + it("records every delivery when deduplication is off", async () => { + const path = await sinkPath() + + const first = await recordDelivery({ + path, + effectId: "effect-1", + attempt: 1, + deduplicate: false, + }) + const second = await recordDelivery({ + path, + effectId: "effect-1", + attempt: 2, + deduplicate: false, + }) + + expect(first).toEqual({ applied: true }) + expect(second).toEqual({ applied: true }) + expect((await readSink(path)).deliveries).toEqual([ + { effectId: "effect-1", attempt: 1 }, + { effectId: "effect-1", attempt: 2 }, + ]) + }) + + it("applies a replayed effect id once when deduplication is on", async () => { + const path = await sinkPath() + + const first = await recordDelivery({ + path, + effectId: "effect-1", + attempt: 1, + deduplicate: true, + }) + const replay = await recordDelivery({ + path, + effectId: "effect-1", + attempt: 2, + deduplicate: true, + }) + + expect(first).toEqual({ applied: true }) + expect(replay).toEqual({ applied: false }) + expect((await readSink(path)).deliveries).toEqual([{ effectId: "effect-1", attempt: 1 }]) + }) + + it("reads an empty sink where no file exists", async () => { + const path = await sinkPath() + expect(await readSink(path)).toEqual({ deliveries: [] }) + }) +}) From af82d86eee5b5a71ac41c3b7b0d29688f3973295 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 24 Aug 2026 14:19:16 -0700 Subject: [PATCH 2/2] fix: stop the demo sink from hiding its own failures Greptile flagged two ways the at-least-once proof could report success without proving anything. readSink turned every read failure into an empty sink, so a damaged or unreadable file looked identical to a first run. The deduplication phase would then forget the effect id it needs, append the replay as if it were the first delivery, and still satisfy the assertion that the sink holds one delivery. Only ENOENT means an empty sink now; every other error propagates. Two tests cover it: a truncated JSON file and a path that is a directory. The effect worker exited 0 when its 200 claim attempts all came back empty, and the parent reads that exit status as proof that the delivery completed and was acknowledged. It now fails when nothing was processed, so the recovery assertion means what it claims. The Ruby counterpart already had both properties, so this brings the two runtimes back to the same shape rather than moving them apart. --- examples/at-least-once/effect-worker.ts | 1 + examples/at-least-once/sink.ts | 5 +++-- test/at-least-once-sink.test.ts | 15 ++++++++++++++- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/examples/at-least-once/effect-worker.ts b/examples/at-least-once/effect-worker.ts index 9314d40..ba07755 100644 --- a/examples/at-least-once/effect-worker.ts +++ b/examples/at-least-once/effect-worker.ts @@ -48,6 +48,7 @@ try { processed = await effectWorker.runOnce() if (processed === 0) await new Promise((resolve) => setTimeout(resolve, 10)) } + if (processed === 0) throw new Error("no effect became claimable") process.send?.({ event: "effects.finished", processed }) } finally { await effectWorker.stop() diff --git a/examples/at-least-once/sink.ts b/examples/at-least-once/sink.ts index 534c83b..cb5f70b 100644 --- a/examples/at-least-once/sink.ts +++ b/examples/at-least-once/sink.ts @@ -13,8 +13,9 @@ export async function readSink(path: string): Promise { try { const parsed: SinkState = JSON.parse(await readFile(path, "utf-8")) return parsed - } catch { - return { deliveries: [] } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return { deliveries: [] } + throw error } } diff --git a/test/at-least-once-sink.test.ts b/test/at-least-once-sink.test.ts index 65a18c7..f4f3c41 100644 --- a/test/at-least-once-sink.test.ts +++ b/test/at-least-once-sink.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, rm, writeFile } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { afterEach, describe, expect, it } from "vitest" @@ -66,4 +66,17 @@ describe("at-least-once sink", () => { const path = await sinkPath() expect(await readSink(path)).toEqual({ deliveries: [] }) }) + + it("refuses to read a damaged sink as an empty one", async () => { + const path = await sinkPath() + await writeFile(path, "{ deliveries: ") + + await expect(readSink(path)).rejects.toThrow() + }) + + it("refuses to read an unreadable sink as an empty one", async () => { + await sinkPath() + + await expect(readSink(directory as string)).rejects.toThrow() + }) })