test: prove the at-least-once clause at a sink

The correctness doc promises at-least-once execution and tells external
systems to hold stable idempotency keys, but nothing in the repository
showed that duplicate arriving anywhere. A contract clause nobody can
watch fire is decoration. This adds the artifact that fires it on demand
and shows the documented remedy absorbing 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 runs
ProcessRegistry.cleanup_dead after the liveness threshold, reclaims the
released effect, and delivers again. With deduplication off the sink
reads 2, both deliveries carrying the same context.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.

bundle exec rake at_least_once runs it, CI runs it in the SQLite job,
and docs/correctness.md links it from the handler idempotency section it
makes observable. The sink module carries unit coverage for both guard
modes. This is the Ruby counterpart of solid-objects-js#26.

Author: cardmagic

.github/workflows/ci.yml

@@ -15,6 +15,7 @@ jobs:
           ruby-version: "3.3"
           bundler-cache: true
       - run: bundle exec rake
+      - run: bundle exec rake at_least_once
 
   javascript:
     runs-on: ubuntu-latest

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## Unreleased
+
+- Add `examples/at_least_once` and `bundle exec rake at_least_once`, an
+  executable proof that the at-least-once clause fires and that the
+  documented remedy absorbs it. One actor turn stages an effect that writes
+  to an external sink file. The first effect worker crashes between the sink
+  write and the acknowledgement, and 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 `context.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. CI runs the demo in the SQLite job, and
+  `docs/correctness.md` links it from the handler idempotency section. This
+  mirrors `pnpm run test:at-least-once` in solid-objects-js.
+
 ## 0.14.0 - 2026-08-22
 
 - Add `SolidObjects::Transmission.receive(envelope)`, the server ingest for

Rakefile

@@ -32,6 +32,11 @@ task :steep do
   sh "bundle exec steep check"
 end
 
+desc "Prove the at-least-once clause by crashing an effect worker at a sink"
+task :at_least_once do
+  sh "bundle exec ruby examples/at_least_once/demo.rb"
+end
+
 desc "Scan the Rails engine for security warnings"
 task :security do
   sh "bundle exec brakeman --force --no-pager -q ."

docs/correctness.md

@@ -93,6 +93,15 @@ end
 The guard prevents a repeated state transition. The effect consumer still
 deduplicates with `context.id`.
 
+This clause is observable, not decorative. `bundle exec rake at_least_once`
+crashes an effect worker between the external sink write and the
+acknowledgement, restarts one after the liveness threshold, and shows the sink
+reading 2 with deduplication off. Both deliveries carry the same `context.id`
+at attempts 1 and 2. A guard on that id absorbs the same duplicate and the
+sink reads 1. The actor state commits exactly once in both runs. The source is
+`examples/at_least_once/`; solid-objects-js runs the same proof with
+`pnpm run test:at-least-once`.
+
 ## Atomic boundaries
 
 The following are atomic:

examples/at_least_once/actor.rb

@@ -0,0 +1,14 @@
+# rbs_inline: enabled
+
+class DeliveryCounter < SolidObjects::Actor
+  actor_type "delivery-counter"
+
+  attribute :count, default: 0
+
+  # @rbs () -> Integer
+  def deliver
+    self.count += 1
+    emit :record
+    count
+  end
+end

examples/at_least_once/boot.rb

@@ -0,0 +1,47 @@
+# rbs_inline: enabled
+
+require "bundler/setup"
+require "active_record"
+require "solid_objects"
+
+# Boots a standalone runtime against a shared SQLite file, the way the
+# demo's parent and its crashing children all attach to one database.
+module AtLeastOnceBoot
+  ROOT = File.expand_path("../..", __dir__)
+
+  # @rbs (String database_path) -> void
+  def self.call(database_path)
+    ActiveRecord::Base.establish_connection(
+      adapter: "sqlite3",
+      database: database_path,
+      pool: 5,
+      timeout: 5_000
+    )
+    ActiveRecord::Migration.verbose = false
+    migrate unless ActiveRecord::Base.connection.table_exists?("solid_objects_instances")
+
+    require "solid_objects/database_adapter"
+    %w[
+      record process instance message ready_message claimed_message
+      reminder effect broadcast dead_letter
+    ].each { |model| require File.join(ROOT, "app/models/solid_objects", model) }
+
+    SolidObjects.configuration.authorize_message = ->(**) { true }
+    SolidObjects.configuration.authorize_query = ->(**) { true }
+    SolidObjects.configuration.polling_interval = 0.01
+    SolidObjects.configuration.process_heartbeat_interval = 0.075
+    SolidObjects.configuration.process_alive_threshold = 0.3
+    SolidObjects.configuration.lease_duration = 0.25
+    SolidObjects.configuration.lease_renewal_interval = 0.05
+  end
+
+  # @rbs () -> void
+  def self.migrate
+    require File.join(ROOT, "db/migrate/20260805000000_create_solid_objects_tables")
+    require File.join(ROOT, "db/migrate/20260806000000_add_state_revision_to_solid_objects_instances")
+    require File.join(ROOT, "db/migrate/20260813000000_rename_message_dispatch_columns")
+    CreateSolidObjectsTables.new.migrate(:up)
+    AddStateRevisionToSolidObjectsInstances.new.migrate(:up)
+    RenameMessageDispatchColumns.new.migrate(:up)
+  end
+end

examples/at_least_once/demo.rb

@@ -0,0 +1,92 @@
+# rbs_inline: enabled
+
+# An executable proof for the at-least-once clause: a contract clause
+# nobody can observe firing is decoration. Run with:
+#
+#   bundle exec rake at_least_once
+#
+# Phase one crashes an effect worker between the external sink write and
+# the acknowledgement, restarts one, and shows the sink reading 2 with
+# deduplication off. Both deliveries carry the same stable effect id.
+# Phase two repeats the crash with a guard on that id; the sink reads 1.
+# The actor state commits exactly once in both phases.
+
+require_relative "boot"
+require_relative "actor"
+require_relative "sink"
+require "fileutils"
+require "json"
+require "rbconfig"
+require "tmpdir"
+
+directory = Dir.mktmpdir("solid_objects_at_least_once_")
+database_path = File.join(directory, "state.sqlite3")
+AtLeastOnceBoot.call(database_path)
+
+# @rbs (String message) -> void
+def prove(message)
+  raise "proof failed: #{message}" unless yield
+end
+
+# @rbs (String actor_id) -> void
+def stage_one_delivery(actor_id)
+  DeliveryCounter.ref(actor_id).async.deliver
+  worker = SolidObjects::Worker.new
+  begin
+    worker.run_until_idle
+  ensure
+    worker.stop
+  end
+end
+
+# @rbs (database_path: String, sink_path: String, mode: String, deduplication: String) -> Integer?
+def run_effect_worker(database_path:, sink_path:, mode:, deduplication:)
+  script = File.expand_path("effect_worker.rb", __dir__)
+  pid = Process.spawn(
+    RbConfig.ruby, script, database_path, sink_path, mode, deduplication,
+    chdir: AtLeastOnceBoot::ROOT
+  )
+  _pid, status = Process.wait2(pid)
+  status.exitstatus
+end
+
+# @rbs (database_path: String, sink_path: String, deduplication: String) -> void
+def crash_then_recover(database_path:, sink_path:, deduplication:)
+  crash = run_effect_worker(database_path:, sink_path:, mode: "crash", deduplication:)
+  prove("the first delivery crashed before acknowledgement") { crash == 1 }
+  sleep 0.4
+  recovery = run_effect_worker(database_path:, sink_path:, mode: "complete", deduplication:)
+  prove("the second delivery completed and acknowledged") { recovery == 0 }
+end
+
+begin
+  sink_off = File.join(directory, "sink-dedup-off.json")
+  stage_one_delivery("dedup-off")
+  crash_then_recover(database_path:, sink_path: sink_off, deduplication: "off")
+  deliveries = AtLeastOnceSink.read(sink_off)
+  effect_ids = deliveries.map { |delivery| delivery.fetch("effect_id") }
+  state_off = SolidObjects::Instance.find_by!(actor_id: "dedup-off").state.fetch("count")
+  prove("the state commit happened exactly once") { state_off == 1 }
+  prove("the sink observed the duplicate") { deliveries.length == 2 }
+  prove("both deliveries carried the same stable effect id") { effect_ids.uniq.length == 1 }
+
+  sink_on = File.join(directory, "sink-dedup-on.json")
+  stage_one_delivery("dedup-on")
+  crash_then_recover(database_path:, sink_path: sink_on, deduplication: "on")
+  guarded = AtLeastOnceSink.read(sink_on)
+  state_on = SolidObjects::Instance.find_by!(actor_id: "dedup-on").state.fetch("count")
+  prove("the state commit happened exactly once") { state_on == 1 }
+  prove("the stable effect id absorbed the duplicate") { guarded.length == 1 }
+
+  puts JSON.pretty_generate(
+    duplicate: {
+      state_commits: state_off,
+      sink_deliveries: deliveries.length,
+      same_effect_id: effect_ids.uniq.length == 1,
+      attempts: deliveries.map { |delivery| delivery.fetch("attempt") }
+    },
+    remedy: { state_commits: state_on, sink_deliveries: guarded.length }
+  )
+ensure
+  FileUtils.remove_entry(directory) if directory
+end

examples/at_least_once/effect_worker.rb

@@ -0,0 +1,40 @@
+# rbs_inline: enabled
+
+require_relative "boot"
+require_relative "actor"
+require_relative "sink"
+
+database_path, sink_path, mode, deduplication = ARGV
+raise ArgumentError, "usage: effect_worker.rb DATABASE SINK crash|complete on|off" unless deduplication
+
+AtLeastOnceBoot.call(database_path.to_s)
+
+SolidObjects.register_effect(:record) do |_arguments, context|
+  AtLeastOnceSink.record(
+    path: sink_path.to_s,
+    effect_id: context.id,
+    attempt: context.attempt,
+    deduplication: deduplication.to_sym
+  )
+  # A crash between the external write and the acknowledgement: the sink
+  # has the delivery, the effect row never completes.
+  Process.exit!(1) if mode == "crash"
+  nil
+end
+
+# Production runs this on the dead-process-cleanup interval; the demo runs
+# it once, after the liveness threshold, to release the crashed claim.
+SolidObjects::ProcessRegistry.cleanup_dead
+
+effect_executor = SolidObjects::EffectExecutor.new
+begin
+  worked = false
+  200.times do
+    worked = effect_executor.run_once
+    break if worked
+    sleep 0.01
+  end
+  raise "no effect became claimable" unless worked
+ensure
+  effect_executor.stop
+end

examples/at_least_once/sink.rb

@@ -0,0 +1,27 @@
+# rbs_inline: enabled
+
+require "json"
+
+# The external system in the at-least-once demo: a JSON file that records
+# every delivery it accepts. With deduplication :off it accepts everything,
+# which makes an at-least-once duplicate visible. With deduplication :on it
+# accepts each stable effect id once, which is the documented remedy.
+module AtLeastOnceSink
+  # @rbs (String path) -> Array[Hash[String, untyped]]
+  def self.read(path)
+    JSON.parse(File.read(path))
+  rescue Errno::ENOENT
+    []
+  end
+
+  # @rbs (path: String, effect_id: String, attempt: Integer, deduplication: Symbol) -> bool
+  def self.record(path:, effect_id:, attempt:, deduplication:)
+    deliveries = read(path)
+    seen = deliveries.any? { |delivery| delivery.fetch("effect_id") == effect_id }
+    return false if deduplication == :on && seen
+
+    deliveries << { "effect_id" => effect_id, "attempt" => attempt }
+    File.write(path, JSON.pretty_generate(deliveries))
+    true
+  end
+end

test/unit/at_least_once_sink_test.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require "tmpdir"
+require_relative "../../examples/at_least_once/sink"
+
+class AtLeastOnceSinkTest < ActiveSupport::TestCase
+  test "records every delivery when deduplication is off" do
+    Dir.mktmpdir do |directory|
+      path = File.join(directory, "sink.json")
+
+      first = AtLeastOnceSink.record(path:, effect_id: "effect-1", attempt: 1, deduplication: :off)
+      second = AtLeastOnceSink.record(path:, effect_id: "effect-1", attempt: 2, deduplication: :off)
+
+      assert first
+      assert second
+      assert_equal(
+        [
+          { "effect_id" => "effect-1", "attempt" => 1 },
+          { "effect_id" => "effect-1", "attempt" => 2 }
+        ],
+        AtLeastOnceSink.read(path)
+      )
+    end
+  end
+
+  test "applies a replayed effect id once when deduplication is on" do
+    Dir.mktmpdir do |directory|
+      path = File.join(directory, "sink.json")
+
+      first = AtLeastOnceSink.record(path:, effect_id: "effect-1", attempt: 1, deduplication: :on)
+      replay = AtLeastOnceSink.record(path:, effect_id: "effect-1", attempt: 2, deduplication: :on)
+
+      assert first
+      refute replay
+      assert_equal([ { "effect_id" => "effect-1", "attempt" => 1 } ], AtLeastOnceSink.read(path))
+    end
+  end
+
+  test "reads an empty sink where no file exists" do
+    Dir.mktmpdir do |directory|
+      assert_equal [], AtLeastOnceSink.read(File.join(directory, "missing.json"))
+    end
+  end
+end