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..f1b08db --- /dev/null +++ b/test/unit/at_least_once_sink_test.rb @@ -0,0 +1,60 @@ +# 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 + + 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