Prove serialization by message, not by execution

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Unreleased
+
+- Make the failure-recovery demo's serialization proof count messages rather
+  than executions. A worker that loses its lease mid-operation leaves the
+  replacement to execute the same message again, which is the at-least-once
+  contract, so the proof failed on slower machines for behaviour it documents
+  elsewhere. Each serialization event now carries its attempt and process, so a
+  start pairs with its own finish instead of with whichever finish came next.
+  The proof asserts that every start has its own finish, that exactly the two
+  sent messages ran, and that the surviving attempt of each message never
+  overlaps another message's surviving attempt; a superseded attempt may
+  overlap anything, because it keeps running until it notices the lost lease and
+  its write is fenced out. The committed state check is unchanged, and the demo
+  reports the executions it saw. `assertSerializedExecution` moved into its own
+  module with unit coverage for the clean, retried, superseded-overlap,
+  still-running-replacement, unexplained-overlap, boundary, unfinished,
+  unmatched-finish, double-start, restart-after-finish, and lost-message cases.
+
 ## 0.13.2 - 2026-08-17
 
 - Accept a `key` on `schedule`, naming a reminder for the item it is waiting

examples/failure-recovery/actor.ts

@@ -25,16 +25,14 @@ export class RecoveryCounter extends Actor {
   async serialize({ controlDirectory }: { controlDirectory: string }): Promise<number> {
     const message = this.currentMessage
     if (!message) throw new Error("serialize requires a durable message")
-    await appendFile(
-      join(controlDirectory, "serialization.jsonl"),
-      `${JSON.stringify({ event: "start", messageId: message.id, at: Date.now() })}\n`,
-    )
+    // The attempt and the process identify the execution, so a start pairs with
+    // its own finish even when a superseded attempt outlives its replacement.
+    const execution = { messageId: message.id, attempt: message.attempt, processId: process.pid }
+    const path = join(controlDirectory, "serialization.jsonl")
+    await appendFile(path, `${JSON.stringify({ event: "start", ...execution, at: Date.now() })}\n`)
     await new Promise((resolve) => setTimeout(resolve, 100))
     this.count += 1
-    await appendFile(
-      join(controlDirectory, "serialization.jsonl"),
-      `${JSON.stringify({ event: "finish", messageId: message.id, at: Date.now() })}\n`,
-    )
+    await appendFile(path, `${JSON.stringify({ event: "finish", ...execution, at: Date.now() })}\n`)
     return this.count
   }
 }

examples/failure-recovery/demo.ts

@@ -8,19 +8,18 @@ import { fork, type ChildProcess } from "node:child_process"
 import { createRuntime, type ActorReference, type MessageReference } from "solid-objects"
 import { sqlite } from "solid-objects/database/sqlite"
 import { RecoveryCounter } from "./actor.ts"
+import {
+  assertSerializedExecution,
+  parseSerializationEvent,
+  type SerializationProof,
+} from "./serialization.ts"
 
 interface WorkerMessage {
   event: string
   attempt?: number
   processed?: number
 }
 
-interface SerializationEvent {
-  event: "start" | "finish"
-  messageId: string
-  at: number
-}
-
 interface ExternalEffectEvent {
   messageId: string
   attempt: number
@@ -59,7 +58,7 @@ try {
 
 assert.equal(existsSync(directory), false)
 
-async function proveSerialization(): Promise<{ finalState: number; overlap: false }> {
+async function proveSerialization(): Promise<SerializationProof & { finalState: number }> {
   const controlDirectory = join(directory, "serialization")
   await mkdir(controlDirectory)
   const reference = runtime.ref(RecoveryCounter, "serialized")
@@ -74,15 +73,10 @@ async function proveSerialization(): Promise<{ finalState: number; overlap: fals
     join(controlDirectory, "serialization.jsonl"),
     parseSerializationEvent,
   )
-  assert.equal(events.length, 4)
-  const starts = events.filter((event) => event.event === "start")
-  const finishes = events.filter((event) => event.event === "finish")
-  assert.equal(starts.length, 2)
-  assert.equal(finishes.length, 2)
-  assert(Number(starts[1]?.at) >= Number(finishes[0]?.at))
+  const proof = assertSerializedExecution(events, { messageCount: 2 })
   const snapshot = await reference.snapshot()
   assert.equal(snapshot.count, 2)
-  return { finalState: snapshot.count, overlap: false }
+  return { ...proof, finalState: snapshot.count }
 }
 
 async function proveCrashRecovery(): Promise<{
@@ -200,18 +194,6 @@ async function readJsonLines<Value>(
   return (await readFile(path, "utf8")).trim().split("\n").filter(Boolean).map(parse)
 }
 
-function parseSerializationEvent(line: string): SerializationEvent {
-  const event = JSON.parse(line) as Partial<SerializationEvent>
-  if (
-    (event.event !== "start" && event.event !== "finish") ||
-    typeof event.messageId !== "string" ||
-    typeof event.at !== "number"
-  ) {
-    throw new TypeError("invalid serialization event")
-  }
-  return { event: event.event, messageId: event.messageId, at: event.at }
-}
-
 function parseExternalEffectEvent(line: string): ExternalEffectEvent {
   const event = JSON.parse(line) as Partial<ExternalEffectEvent>
   if (

examples/failure-recovery/serialization.ts

@@ -0,0 +1,133 @@
+import assert from "node:assert/strict"
+
+export interface SerializationEvent {
+  event: "start" | "finish"
+  messageId: string
+  attempt: number
+  processId: number
+  at: number
+}
+
+export interface SerializationProof {
+  executions: number
+  retried: boolean
+  supersededOverlap: boolean
+}
+
+interface Execution {
+  messageId: string
+  attempt: number
+  processId: number
+  startedAt: number
+  finishedAt: number
+}
+
+export function parseSerializationEvent(line: string): SerializationEvent {
+  const event = JSON.parse(line) as Partial<SerializationEvent>
+  if (
+    (event.event !== "start" && event.event !== "finish") ||
+    typeof event.messageId !== "string" ||
+    typeof event.attempt !== "number" ||
+    typeof event.processId !== "number" ||
+    typeof event.at !== "number"
+  ) {
+    throw new TypeError("invalid serialization event")
+  }
+  return {
+    event: event.event,
+    messageId: event.messageId,
+    attempt: event.attempt,
+    processId: event.processId,
+    at: event.at,
+  }
+}
+
+// One identity commits one state transition at a time. The control file is
+// written outside the transaction, so it records execution attempts rather than
+// commits: a worker that loses its lease keeps running until it notices, and its
+// replacement executes the same message under a higher attempt. The superseded
+// attempt may therefore overlap anything, because its write is fenced out and
+// the committed state is what proves it.
+//
+// Each event carries its attempt and process, so a start pairs with its own
+// finish rather than with whichever finish arrived next. Without that, a
+// superseded attempt finishing late reads as its replacement finishing, and a
+// second message could then overlap a replacement that is still running.
+export function assertSerializedExecution(
+  events: readonly SerializationEvent[],
+  options: { messageCount: number },
+): SerializationProof {
+  const executions = pairExecutions(events)
+
+  const messageIds = new Set(executions.map((execution) => execution.messageId))
+  assert.equal(
+    messageIds.size,
+    options.messageCount,
+    `expected ${options.messageCount} messages to run, saw ${messageIds.size}`,
+  )
+
+  const survivingAttempt = new Map<string, number>()
+  for (const execution of executions) {
+    const highest = survivingAttempt.get(execution.messageId) ?? 0
+    if (execution.attempt > highest) survivingAttempt.set(execution.messageId, execution.attempt)
+  }
+  const surviving = executions.filter(
+    (execution) => survivingAttempt.get(execution.messageId) === execution.attempt,
+  )
+
+  for (const [index, execution] of surviving.entries()) {
+    for (const other of surviving.slice(index + 1)) {
+      assert(
+        !overlaps(execution, other),
+        `${describe(execution)} and ${describe(other)} overlap, and neither was superseded`,
+      )
+    }
+  }
+
+  const supersededOverlap = executions.some((execution) =>
+    executions.some((other) => other !== execution && overlaps(execution, other)),
+  )
+
+  return {
+    executions: executions.length,
+    retried: executions.length > options.messageCount,
+    supersededOverlap,
+  }
+}
+
+function pairExecutions(events: readonly SerializationEvent[]): Execution[] {
+  const started = new Map<string, SerializationEvent>()
+  const executions: Execution[] = []
+
+  for (const event of [...events].sort((left, right) => left.at - right.at)) {
+    const key = `${event.messageId}#${event.attempt}#${event.processId}`
+    if (event.event === "start") {
+      assert(!started.has(key), `${describe(event)} started twice`)
+      started.set(key, event)
+      continue
+    }
+    const start = started.get(key)
+    assert(start !== undefined, `${describe(event)} finished with no matching start`)
+    started.delete(key)
+    executions.push({
+      messageId: event.messageId,
+      attempt: event.attempt,
+      processId: event.processId,
+      startedAt: start.at,
+      finishedAt: event.at,
+    })
+  }
+
+  const unfinished = [...started.values()].map(describe)
+  assert.equal(unfinished.length, 0, `${unfinished.join(", ")} never wrote a finish`)
+
+  return executions
+}
+
+function overlaps(left: Execution, right: Execution): boolean {
+  return left.startedAt < right.finishedAt && right.startedAt < left.finishedAt
+}
+
+function describe(execution: { messageId: string; attempt: number }): string {
+  return `${execution.messageId} attempt ${execution.attempt}`
+}