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..ba07755 --- /dev/null +++ b/examples/at-least-once/effect-worker.ts @@ -0,0 +1,62 @@ +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)) + } + if (processed === 0) throw new Error("no effect became claimable") + 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..cb5f70b --- /dev/null +++ b/examples/at-least-once/sink.ts @@ -0,0 +1,34 @@ +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 (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return { deliveries: [] } + throw error + } +} + +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..f4f3c41 --- /dev/null +++ b/test/at-least-once-sink.test.ts @@ -0,0 +1,82 @@ +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" +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: [] }) + }) + + 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() + }) +})