examples: make the at-least-once duplicate observable at a sink

.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

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

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

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
+  }
+}

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<void> {
+  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<void> {
+  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<number | null> } {
+  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<number | null>((resolve, reject) => {
+    child.once("exit", (code) => resolve(code))
+    child.once("error", reject)
+  })
+  return { child, finished }
+}

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
+}

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<SinkState> {
+  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 }
+}

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",

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<string> {
+  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()
+  })
+})