FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

examples: prove the at-least-once clause at a sink by cardmagic · Pull Request #51 · cardmagic/solid-objects-ruby · GitHub

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .md  (2) .rb  (6) .yml  (1) No extension  (1) All 4 file types selected
Only manifest files
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
1 change: 1 addition & 0 deletions .github/workflows/ci.yml
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 14 additions & 0 deletions CHANGELOG.md
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
Expand Down
5 changes: 5 additions & 0 deletions Rakefile
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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 ."
Expand Down
9 changes: 9 additions & 0 deletions docs/correctness.md
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
14 changes: 14 additions & 0 deletions examples/at_least_once/actor.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
47 changes: 47 additions & 0 deletions examples/at_least_once/boot.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
92 changes: 92 additions & 0 deletions examples/at_least_once/demo.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
40 changes: 40 additions & 0 deletions examples/at_least_once/effect_worker.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
27 changes: 27 additions & 0 deletions examples/at_least_once/sink.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
60 changes: 60 additions & 0 deletions test/unit/at_least_once_sink_test.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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
Loading

Back | FazBrowse Home | New Git URL