From abe93f02e9e7a71b482dbc8c700c70794bdd386c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 24 Aug 2026 12:34:01 -0700 Subject: [PATCH 1/2] 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. --- .github/workflows/ci.yml | 1 + CHANGELOG.md | 14 ++++ Rakefile | 5 ++ docs/correctness.md | 9 +++ examples/at_least_once/actor.rb | 14 ++++ examples/at_least_once/boot.rb | 47 +++++++++++++ examples/at_least_once/demo.rb | 92 +++++++++++++++++++++++++ examples/at_least_once/effect_worker.rb | 40 +++++++++++ examples/at_least_once/sink.rb | 27 ++++++++ test/unit/at_least_once_sink_test.rb | 45 ++++++++++++ 10 files changed, 294 insertions(+) create mode 100644 examples/at_least_once/actor.rb create mode 100644 examples/at_least_once/boot.rb create mode 100644 examples/at_least_once/demo.rb create mode 100644 examples/at_least_once/effect_worker.rb create mode 100644 examples/at_least_once/sink.rb create mode 100644 test/unit/at_least_once_sink_test.rb diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a832474..1a2cba6 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cc4a83..94c21e7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Rakefile b/Rakefile index 72520b2..817ab12 100644 --- a/Rakefile +++ b/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 ." diff --git a/docs/correctness.md b/docs/correctness.md index fe3ad9b..c8bb2e3 100644 --- a/docs/correctness.md +++ b/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: diff --git a/examples/at_least_once/actor.rb b/examples/at_least_once/actor.rb new file mode 100644 index 0000000..93370e3 --- /dev/null +++ b/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 diff --git a/examples/at_least_once/boot.rb b/examples/at_least_once/boot.rb new file mode 100644 index 0000000..c66c443 --- /dev/null +++ b/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 diff --git a/examples/at_least_once/demo.rb b/examples/at_least_once/demo.rb new file mode 100644 index 0000000..ec8d5b1 --- /dev/null +++ b/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 diff --git a/examples/at_least_once/effect_worker.rb b/examples/at_least_once/effect_worker.rb new file mode 100644 index 0000000..d23ffb1 --- /dev/null +++ b/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 diff --git a/examples/at_least_once/sink.rb b/examples/at_least_once/sink.rb new file mode 100644 index 0000000..f5f528a --- /dev/null +++ b/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 diff --git a/test/unit/at_least_once_sink_test.rb b/test/unit/at_least_once_sink_test.rb new file mode 100644 index 0000000..4f0ed03 --- /dev/null +++ b/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 From b8ea47a287866054d85f062cb34b5958b75c8d9a Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 24 Aug 2026 14:19:30 -0700 Subject: [PATCH 2/2] test: pin the sink against silent read failures Greptile flagged the JS counterpart for reading every failure as an empty sink, which lets the deduplication phase forget the effect id and still pass. The Ruby sink already rescues only Errno::ENOENT, and the Ruby effect worker already fails when no effect became claimable, so neither defect exists here. Both properties were incidental until now. These two tests make them deliberate: a truncated JSON file and a path that is a directory must raise rather than read as a first run. Widening the rescue to StandardError fails both. --- test/unit/at_least_once_sink_test.rb | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/test/unit/at_least_once_sink_test.rb b/test/unit/at_least_once_sink_test.rb index 4f0ed03..f1b08db 100644 --- a/test/unit/at_least_once_sink_test.rb +++ b/test/unit/at_least_once_sink_test.rb @@ -42,4 +42,19 @@ class AtLeastOnceSinkTest < ActiveSupport::TestCase assert_equal [], AtLeastOnceSink.read(File.join(directory, "missing.json")) end end + + test "refuses to read a damaged sink as an empty one" do + Dir.mktmpdir do |directory| + path = File.join(directory, "sink.json") + File.write(path, "{ deliveries: ") + + assert_raises(JSON::ParserError) { AtLeastOnceSink.read(path) } + end + end + + test "refuses to read an unreadable sink as an empty one" do + Dir.mktmpdir do |directory| + assert_raises(Errno::EISDIR) { AtLeastOnceSink.read(directory) } + end + end end