From 6cdbac8a962d14d2db76cf2a5acd899a62f8725f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Fri, 7 Aug 2026 09:53:38 -0700 Subject: [PATCH 1/2] fix: bound sync retries and load actors Keep SQLite caller bookkeeping and result observation inside the original synchronous deadline without retrying actor behavior. Load and register host app actors before CLI workers start, including when development eager loading is disabled. --- CHANGELOG.md | 7 + Gemfile.lock | 4 +- README.md | 11 +- docs/correctness.md | 11 +- docs/operations.md | 6 + lib/solid_objects/activation.rb | 14 +- lib/solid_objects/actor_registry.rb | 7 +- lib/solid_objects/application_actor_loader.rb | 53 ++++++++ lib/solid_objects/caller_process.rb | 10 +- lib/solid_objects/cli.rb | 2 + lib/solid_objects/client.rb | 46 ++++--- lib/solid_objects/database_adapter.rb | 10 ++ lib/solid_objects/database_adapters/sqlite.rb | 48 ++++++- lib/solid_objects/process_registry.rb | 35 +++-- lib/solid_objects/sync_diagnostics.rb | 60 ++++++++- lib/solid_objects/synchronous_invocation.rb | 58 ++++++-- lib/solid_objects/version.rb | 2 +- .../lib/solid_objects/actor_registry.rbs | 3 + .../application_actor_loader.rbs | 30 +++++ .../lib/solid_objects/database_adapter.rbs | 10 +- .../database_adapters/sqlite.rbs | 11 +- .../lib/solid_objects/sync_diagnostics.rbs | 9 ++ .../solid_objects/synchronous_invocation.rbs | 9 ++ test/dummy/app/actors/cli_worker_actor.rb | 9 ++ test/dummy/config/database.yml | 2 +- .../config/initializers/cli_worker_probe.rb | 10 ++ test/dummy/prepare_cli_worker.rb | 39 ++++++ test/integration/cli_test.rb | 57 ++++++++ .../synchronous_invocation_test.rb | 126 ++++++++++++++++++ test/unit/actor_registry_test.rb | 25 ++++ 30 files changed, 645 insertions(+), 79 deletions(-) create mode 100644 lib/solid_objects/application_actor_loader.rb create mode 100644 sig/generated/lib/solid_objects/application_actor_loader.rbs create mode 100644 test/dummy/app/actors/cli_worker_actor.rb create mode 100644 test/dummy/config/initializers/cli_worker_probe.rb create mode 100644 test/dummy/prepare_cli_worker.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 05b0cc4..732c7fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 0.4.3 - 2026-08-07 + +- Bound SQLite caller-process registration, reuse, heartbeat, and synchronous + result observation retries by the original invocation deadline. +- Load host application actors from `app/actors` before CLI workers start, + including development environments with eager loading disabled. + ## 0.4.2 - 2026-08-07 - Decode Action Cable broadcast payloads before parsing observable invalidations diff --git a/Gemfile.lock b/Gemfile.lock index 6ce51ad..1a68f79 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.4.2) + solid_objects (0.4.3) actioncable (>= 8.0) actionpack (>= 8.0) actionview (>= 8.0) @@ -373,7 +373,7 @@ CHECKSUMS rubocop-rails-omakase (1.1.0) sha256=2af73ac8ee5852de2919abbd2618af9c15c19b512c4cfc1f9a5d3b6ef009109d ruby-progressbar (1.13.0) sha256=80fc9c47a9b640d6834e0dc7b3c94c9df37f08cb072b7761e4a71e22cff29b33 securerandom (0.4.1) sha256=cc5193d414a4341b6e225f0cb4446aceca8e50d5e1888743fac16987638ea0b1 - solid_objects (0.4.2) + solid_objects (0.4.3) sqlite3 (2.9.5-aarch64-linux-gnu) sha256=78075b6337d3d182c6d2b4691049ed45cd220826160c9ea18946bf6a1de200dc sqlite3 (2.9.5-aarch64-linux-musl) sha256=18c801185deb4adc01ddb281e8f672a39e3d1729979ca91e39439cd3eac0402d sqlite3 (2.9.5-arm-linux-gnu) sha256=1bdfca0c7d63998c60b0f4a8e3c8df2d33800ccc4abd2d612eddbbbc92a4c48b diff --git a/README.md b/README.md index 8660990..f737a3f 100644 --- a/README.md +++ b/README.md @@ -588,9 +588,9 @@ polling as the fallback. A timeout never cancels the durable invocation. durable status, mailbox blocker, and activation-owner diagnostics without including message arguments. The configured timeout also bounds adapter database lock waits from the enqueue attempt through result observation. -PostgreSQL uses transaction lock and statement timeouts, SQLite uses its busy -timeout, and MySQL uses its execution timeout plus InnoDB's one-second minimum -lock-wait granularity. +PostgreSQL uses transaction lock and statement timeouts, SQLite retries busy +coordination operations only until the original call deadline, and MySQL uses +its execution timeout plus InnoDB's one-second minimum lock-wait granularity. The durable call can finish after its original caller gives up. Reauthorize and recover its eventual result through the durable message identity: @@ -896,6 +896,11 @@ and marks process rows stopped on graceful shutdown. A hard-killed worker's claimed turn is recovered after its process heartbeat or activation lease becomes stale. +Before any role starts, the CLI loads actors from the host application's +`app/actors` directories through Rails' main autoloader. This works when +development eager loading is disabled and does not require actor references in +an initializer. + See the [operations guide](docs/operations.md) for monitoring, reconciliation, shutdown, retention, and backup guidance. diff --git a/docs/correctness.md b/docs/correctness.md index b408ebf..aa8fbb9 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -155,10 +155,13 @@ Timeout raises `SolidObjects::SyncTimeout` but does not cancel the message. The exception reports actor identity, message ID and sequence, durable status, an earlier mailbox blocker, and activation-owner metadata without exposing arguments. Its `message_reference` can reauthorize and wait for the eventual -result. Adapter lock/query deadlines cover the durable enqueue and coordination -transactions. If enqueue cannot commit, `SyncEnqueueTimeout` is raised and no -message reference exists. MySQL lock waits have one-second InnoDB granularity. -Ruby handlers that already started are not preempted. +result. Adapter lock/query deadlines cover the durable enqueue, caller-process +registration and heartbeat, activation coordination, and result observation. +SQLite retries busy coordination operations only within the original call +deadline and reports `waiting_on=database_contention` when the database cannot +be inspected at timeout. If enqueue cannot commit, `SyncEnqueueTimeout` is +raised and no message reference exists. MySQL lock waits have one-second InnoDB +granularity. Ruby handlers that already started are not preempted. A synchronous call made while the Solid Objects connection already has an open transaction raises `SolidObjects::SyncInsideTransaction` before the message is diff --git a/docs/operations.md b/docs/operations.md index f204868..7c597d1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -23,6 +23,12 @@ Start all configured roles: bundle exec solid_objects start ``` +The command loads the host application's `app/actors` directories before +starting any runtime role, even when Rails eager loading is disabled. Actors in +the conventional directory do not need initializer references. The targeted +loader participates in Rails preparation callbacks so a development reload can +replace a registered actor class without loading unrelated application code. + Inspect process records and clean stale ownership: ```bash diff --git a/lib/solid_objects/activation.rb b/lib/solid_objects/activation.rb index 0ac69a7..a82cd53 100644 --- a/lib/solid_objects/activation.rb +++ b/lib/solid_objects/activation.rb @@ -13,7 +13,9 @@ class Activation # @rbs (lease: Lease) -> void def initialize(lease:) @lease = lease - instance = Instance.find(lease.instance_id) + instance = SolidObjects.database_adapter.with_lock_retry do + Instance.find(lease.instance_id) + end @actor_class = SolidObjects.registry.fetch(instance.actor_type) @actor = build_actor(instance) @last_used_at = monotonic_now @@ -74,10 +76,12 @@ def lease_renewal_due? # @rbs () -> void def yield_ready_messages - now = SolidObjects.database_adapter.database_now - ReadyMessage - .where(instance_id: lease.instance_id, available_at: ..now) - .update_all(available_at: now) + SolidObjects.database_adapter.transaction do + now = SolidObjects.database_adapter.database_now + ReadyMessage + .where(instance_id: lease.instance_id, available_at: ..now) + .update_all(available_at: now) + end end # @rbs (Hash[String, untyped]) -> void diff --git a/lib/solid_objects/actor_registry.rb b/lib/solid_objects/actor_registry.rb index 29bdf96..15c1e90 100644 --- a/lib/solid_objects/actor_registry.rb +++ b/lib/solid_objects/actor_registry.rb @@ -18,7 +18,7 @@ def register(type, actor_class) mutex.synchronize do existing = actors[actor_type] - if existing && existing != actor_class + if existing && existing != actor_class && !reload_of?(existing, actor_class) raise InvalidActor, "#{actor_type.inspect} is already registered by #{existing.name}" end @@ -61,5 +61,10 @@ def validate_actor_class!(actor_class) raise InvalidActor, "registered actor must inherit from SolidObjects::Actor" end + + # @rbs (Class, Class) -> bool + def reload_of?(existing, candidate) + !existing.name.nil? && existing.name == candidate.name + end end end diff --git a/lib/solid_objects/application_actor_loader.rb b/lib/solid_objects/application_actor_loader.rb new file mode 100644 index 0000000..07644e5 --- /dev/null +++ b/lib/solid_objects/application_actor_loader.rb @@ -0,0 +1,53 @@ +# rbs_inline: enabled + +module SolidObjects + class ApplicationActorLoader + # @rbs @application: untyped + # @rbs @autoloader: untyped + + # @rbs (?application: untyped, ?autoloader: untyped) -> void + def initialize(application: Rails.application, autoloader: Rails.autoloaders.main) + @application = application + @autoloader = autoloader + end + + # @rbs () -> void + def call + actor_directories.each { |directory| autoloader.eager_load_dir(directory) } + current_actor_classes.each(&:ensure_registered!) + end + + # @rbs () -> void + def install + application.reloader.to_prepare { call } + call + end + + private + + attr_reader :application, :autoloader + + # @rbs () -> Array[String] + def actor_directories + configured_directories = application.paths["app/actors"]&.existent || [] + conventional_directories = application.paths["app"].existent.select do |application_directory| + File.basename(application_directory) == "actors" + end + managed_directories = autoloader.dirs.map { |directory| File.expand_path(directory) } + + (configured_directories + conventional_directories) + .select { |directory| Dir.exist?(directory) } + .map { |directory| File.expand_path(directory) } + .select { |directory| managed_directories.include?(directory) } + .uniq + end + + # @rbs () -> Array[Class] + def current_actor_classes + Actor.descendants.select do |actor_class| + actor_class.name && + actor_class.name.safe_constantize.equal?(actor_class) + end + end + end +end diff --git a/lib/solid_objects/caller_process.rb b/lib/solid_objects/caller_process.rb index 6170e4d..203bd8b 100644 --- a/lib/solid_objects/caller_process.rb +++ b/lib/solid_objects/caller_process.rb @@ -52,19 +52,21 @@ def reset_after_fork def reusable_registry? return false unless registry&.process_record - registry.process_record.reload.shutdown_state == "running" + SolidObjects.database_adapter.with_lock_retry do + registry.process_record.reload.shutdown_state == "running" + end rescue ActiveRecord::RecordNotFound false end # @rbs () -> ProcessRegistry def register - @registry = ProcessRegistry.new - registry.register( + process_registry = ProcessRegistry.new + process_registry.register( kind: "caller", metadata: { execution: "synchronous" } ) - registry + @registry = process_registry end # @rbs () -> void diff --git a/lib/solid_objects/cli.rb b/lib/solid_objects/cli.rb index a8b9f57..f4b0a6f 100644 --- a/lib/solid_objects/cli.rb +++ b/lib/solid_objects/cli.rb @@ -1,6 +1,7 @@ # rbs_inline: enabled require "thor" +require "solid_objects/application_actor_loader" module SolidObjects class CLI < Thor @@ -133,6 +134,7 @@ def boot_application end require path + ApplicationActorLoader.new.install end # @rbs (Symbol, Integer) -> Integer diff --git a/lib/solid_objects/client.rb b/lib/solid_objects/client.rb index ae0a2b8..8019a98 100644 --- a/lib/solid_objects/client.rb +++ b/lib/solid_objects/client.rb @@ -65,27 +65,33 @@ def sync(reference, message_name, arguments, timeout:, idempotency_key: nil, aut # @rbs (MessageReference, timeout: Numeric, ?authorization_context: untyped) -> untyped def wait(message_reference, timeout:, authorization_context: nil) - message = Message.find(message_reference.id) - validate_message_reference!(message_reference, message) - reference = Reference.new( - actor_type: message.actor_type, - actor_id: message.actor_id - ) - actor_class = SolidObjects.registry.fetch(reference.actor_type) - query = actor_class.definition.queries.key?(message.message_name.to_sym) - actor_message = actor_class.definition.messages.key?(message.message_name.to_sym) - unless query || actor_message - raise UnknownMessage, "unknown message #{message.message_name.inspect}" + SyncDeadline.with(timeout:) do + message = SolidObjects.database_adapter.with_lock_retry do + Message.find(message_reference.id) + end + validate_message_reference!(message_reference, message) + reference = Reference.new( + actor_type: message.actor_type, + actor_id: message.actor_id + ) + actor_class = SolidObjects.registry.fetch(reference.actor_type) + query = actor_class.definition.queries.key?(message.message_name.to_sym) + actor_message = actor_class.definition.messages.key?(message.message_name.to_sym) + unless query || actor_message + raise UnknownMessage, "unknown message #{message.message_name.inspect}" + end + authorize!( + query ? SolidObjects.configuration.authorize_query : SolidObjects.configuration.authorize_message, + reference, + message.message_name, + message.arguments, + authorization_context: + ) + reject_sync_inside_transaction!(reference, message.message_name) + SynchronousInvocation.new.call(message_reference, timeout:) end - authorize!( - query ? SolidObjects.configuration.authorize_query : SolidObjects.configuration.authorize_message, - reference, - message.message_name, - message.arguments, - authorization_context: - ) - reject_sync_inside_transaction!(reference, message.message_name) - SynchronousInvocation.new.call(message_reference, timeout:) + rescue DatabaseDeadlineExceeded + raise SyncDiagnostics.new.database_contention_for(message_reference, timeout:) end # @rbs (Reference, ?authorization_context: untyped) -> StateSnapshot diff --git a/lib/solid_objects/database_adapter.rb b/lib/solid_objects/database_adapter.rb index bbde9da..09e1b79 100644 --- a/lib/solid_objects/database_adapter.rb +++ b/lib/solid_objects/database_adapter.rb @@ -52,6 +52,16 @@ def database_now value.is_a?(Time) ? value.utc : Time.parse("#{value} UTC").utc end + # @rbs () { () -> untyped } -> untyped + def with_lock_retry + yield + end + + # @rbs () { () -> untyped } -> untyped + def with_lock_probe + yield + end + # @rbs () { () -> untyped } -> untyped def transaction(&block) raise DatabaseDeadlineExceeded, "synchronous invocation deadline expired" if SyncDeadline.expired? diff --git a/lib/solid_objects/database_adapters/sqlite.rb b/lib/solid_objects/database_adapters/sqlite.rb index 9089cf3..6f83405 100644 --- a/lib/solid_objects/database_adapters/sqlite.rb +++ b/lib/solid_objects/database_adapters/sqlite.rb @@ -3,8 +3,6 @@ module SolidObjects module DatabaseAdapters class Sqlite < DatabaseAdapter - LOCK_RETRY_INTERVAL = 0.001 - # @rbs () -> String def current_time_expression "STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')" @@ -14,12 +12,49 @@ def current_time_expression def transaction(&block) return super unless SyncDeadline.active? - super + with_lock_retry { super } + end + + # @rbs () { () -> untyped } -> untyped + def with_lock_retry + return yield unless SyncDeadline.active? + + raise DatabaseDeadlineExceeded, "synchronous invocation deadline expired" if SyncDeadline.expired? + + with_connection do |connection| + with_transaction_deadline(connection) { yield } + end rescue DatabaseDeadlineExceeded raise if SyncDeadline.expired? - sleep [ LOCK_RETRY_INTERVAL, SyncDeadline.remaining ].min + yield_before_retry retry + rescue => error + raise unless deadline_error?(error) + + if SyncDeadline.expired? + raise DatabaseDeadlineExceeded, + "database lock wait exceeded the synchronous invocation deadline", + cause: error + end + + yield_before_retry + retry + end + + # @rbs () { () -> untyped } -> untyped + def with_lock_probe + return yield unless SyncDeadline.active? + + with_connection do |connection| + with_transaction_deadline(connection) { yield } + end + rescue => error + raise unless deadline_error?(error) + + raise DatabaseDeadlineExceeded, + "database remained locked at the synchronous invocation deadline", + cause: error end private @@ -47,6 +82,11 @@ def deadline_error?(error) end false end + + # @rbs () -> void + def yield_before_retry + Thread.pass + end end end end diff --git a/lib/solid_objects/process_registry.rb b/lib/solid_objects/process_registry.rb index 2a49b06..27caa6a 100644 --- a/lib/solid_objects/process_registry.rb +++ b/lib/solid_objects/process_registry.rb @@ -80,18 +80,21 @@ def initialize # @rbs (?kind: String, ?metadata: Hash[String | Symbol, untyped]) -> Process def register(kind: "worker", metadata: {}) - now = SolidObjects.database_adapter.database_now - @process_record = Process.create!( - id: SecureRandom.uuid, - kind:, - hostname: Socket.gethostname, - pid: ::Process.pid, - started_at: now, - last_heartbeat_at: now, - metadata: Serialization.dump(default_metadata.merge(metadata)) - ) + process_record = SolidObjects.database_adapter.with_lock_retry do + now = SolidObjects.database_adapter.database_now + Process.create!( + id: SecureRandom.uuid, + kind:, + hostname: Socket.gethostname, + pid: ::Process.pid, + started_at: now, + last_heartbeat_at: now, + metadata: Serialization.dump(default_metadata.merge(metadata)) + ) + end + @process_record = process_record @last_heartbeat_at = monotonic_now - @process_record + process_record end # @rbs () -> bool @@ -99,9 +102,13 @@ def heartbeat return false unless process_record return false if heartbeat_recent? - process_record.update( - last_heartbeat_at: SolidObjects.database_adapter.database_now - ).tap { @last_heartbeat_at = monotonic_now } + updated = SolidObjects.database_adapter.with_lock_retry do + process_record.update( + last_heartbeat_at: SolidObjects.database_adapter.database_now + ) + end + @last_heartbeat_at = monotonic_now if updated + updated end # @rbs () -> bool diff --git a/lib/solid_objects/sync_diagnostics.rb b/lib/solid_objects/sync_diagnostics.rb index b5ff27e..9d1042b 100644 --- a/lib/solid_objects/sync_diagnostics.rb +++ b/lib/solid_objects/sync_diagnostics.rb @@ -10,6 +10,62 @@ def call(message, timeout:) status = message_status(message) waiting_on = waiting_reason(message, instance, blocker) activation = activation_details(instance) + build_error( + message, + timeout:, + status:, + waiting_on:, + activation:, + blocker: blocker_details(blocker) + ) + end + + # @rbs (Message, timeout: Numeric) -> SyncTimeout + def database_contention(message, timeout:) + build_error( + message, + timeout:, + status: "unknown", + waiting_on: "database_contention", + activation: {}, + blocker: nil + ) + end + + # @rbs (MessageReference, timeout: Numeric) -> SyncTimeout + def database_contention_for(message_reference, timeout:) + error = SyncTimeout.new( + timeout:, + actor_type: message_reference.actor_type, + actor_id: message_reference.actor_id, + message_name: "unknown", + message_id: message_reference.id, + request_id: message_reference.request_id, + sequence: message_reference.sequence, + status: "unknown", + waiting_on: "database_contention", + activation: {}, + blocker: nil + ) + SolidObjects.instrument( + :"sync.timeout", + message_id: message_reference.id, + request_id: message_reference.request_id, + actor_type: message_reference.actor_type, + actor_id: message_reference.actor_id, + sequence: message_reference.sequence, + status: "unknown", + waiting_on: "database_contention", + activation_owner_id: nil, + activation_generation: nil + ) + error + end + + private + + # @rbs (Message, timeout: Numeric, status: String, waiting_on: String, activation: Hash[String, untyped], blocker: Hash[String, untyped]?) -> SyncTimeout + def build_error(message, timeout:, status:, waiting_on:, activation:, blocker:) error = SyncTimeout.new( timeout:, actor_type: message.actor_type, @@ -21,7 +77,7 @@ def call(message, timeout:) status:, waiting_on:, activation:, - blocker: blocker_details(blocker) + blocker: ) SolidObjects.instrument( :"sync.timeout", @@ -38,8 +94,6 @@ def call(message, timeout:) error end - private - # @rbs (Message) -> String def message_status(message) return "rejected" if message.rejected? diff --git a/lib/solid_objects/synchronous_invocation.rb b/lib/solid_objects/synchronous_invocation.rb index 8f85400..41b5a45 100644 --- a/lib/solid_objects/synchronous_invocation.rb +++ b/lib/solid_objects/synchronous_invocation.rb @@ -22,33 +22,49 @@ def call(message_reference, timeout:) # @rbs (MessageReference, timeout: Numeric) -> untyped def call_before_deadline(message_reference, timeout:) deadline = monotonic_now + SyncDeadline.remaining + message = nil loop do - message = load_message(message_reference) - return completed_result(message) if message.completed? || message.dead? - - remaining = deadline - monotonic_now - unless remaining.positive? - message = load_message(message_reference) - return completed_result(message) if message.completed? || message.dead? + unless (deadline - monotonic_now).positive? + final_message = load_message_at_deadline(message_reference) + return completed_result(final_message) if final_message&.completed? || final_message&.dead? - raise SyncDiagnostics.new.call(message, timeout:) + raise final_message ? + diagnose_timeout(final_message, timeout:) : + contention_timeout(message, message_reference, timeout:) end + message = load_message(message_reference) + return completed_result(message) if message.completed? || message.dead? + processed = assist(message, deadline:) remaining = deadline - monotonic_now wait(remaining) if processed.zero? && remaining.positive? end rescue DatabaseDeadlineExceeded - message = load_message(message_reference) - return completed_result(message) if message.completed? || message.dead? - - raise SyncDiagnostics.new.call(message, timeout:) + raise message ? + diagnose_timeout(message, timeout:) : + contention_timeout(message, message_reference, timeout:) end # @rbs (MessageReference) -> Message def load_message(message_reference) - Message.uncached { Message.find(message_reference.id) } + SolidObjects.database_adapter.with_lock_retry do + Message.uncached do + Message.includes(:dead_letter).find(message_reference.id) + end + end + end + + # @rbs (MessageReference) -> Message? + def load_message_at_deadline(message_reference) + SolidObjects.database_adapter.with_lock_probe do + Message.uncached do + Message.includes(:dead_letter).find(message_reference.id) + end + end + rescue DatabaseDeadlineExceeded + nil end # @rbs (Message) -> untyped @@ -104,6 +120,22 @@ def wait(remaining) ) end + # @rbs (Message, timeout: Numeric) -> SyncTimeout + def diagnose_timeout(message, timeout:) + SolidObjects.database_adapter.with_lock_probe do + SyncDiagnostics.new.call(message, timeout:) + end + rescue DatabaseDeadlineExceeded + SyncDiagnostics.new.database_contention(message, timeout:) + end + + # @rbs (Message?, MessageReference, timeout: Numeric) -> SyncTimeout + def contention_timeout(message, message_reference, timeout:) + return SyncDiagnostics.new.database_contention(message, timeout:) if message + + SyncDiagnostics.new.database_contention_for(message_reference, timeout:) + end + # @rbs () -> Float def monotonic_now ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index c5e512b..36bd9fe 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.4.2" + VERSION = "0.4.3" end diff --git a/sig/generated/lib/solid_objects/actor_registry.rbs b/sig/generated/lib/solid_objects/actor_registry.rbs index 1b64144..7548ec5 100644 --- a/sig/generated/lib/solid_objects/actor_registry.rbs +++ b/sig/generated/lib/solid_objects/actor_registry.rbs @@ -32,5 +32,8 @@ module SolidObjects # @rbs (untyped) -> void def validate_actor_class!: (untyped) -> void + + # @rbs (Class, Class) -> bool + def reload_of?: (Class, Class) -> bool end end diff --git a/sig/generated/lib/solid_objects/application_actor_loader.rbs b/sig/generated/lib/solid_objects/application_actor_loader.rbs new file mode 100644 index 0000000..397e9cc --- /dev/null +++ b/sig/generated/lib/solid_objects/application_actor_loader.rbs @@ -0,0 +1,30 @@ +# Generated from lib/solid_objects/application_actor_loader.rb with RBS::Inline + +module SolidObjects + class ApplicationActorLoader + @application: untyped + + @autoloader: untyped + + # @rbs (?application: untyped, ?autoloader: untyped) -> void + def initialize: (?application: untyped, ?autoloader: untyped) -> void + + # @rbs () -> void + def call: () -> void + + # @rbs () -> void + def install: () -> void + + private + + attr_reader application: untyped + + attr_reader autoloader: untyped + + # @rbs () -> Array[String] + def actor_directories: () -> Array[String] + + # @rbs () -> Array[Class] + def current_actor_classes: () -> Array[Class] + end +end diff --git a/sig/generated/lib/solid_objects/database_adapter.rbs b/sig/generated/lib/solid_objects/database_adapter.rbs index 5519b13..103df0e 100644 --- a/sig/generated/lib/solid_objects/database_adapter.rbs +++ b/sig/generated/lib/solid_objects/database_adapter.rbs @@ -5,10 +5,10 @@ module SolidObjects # @rbs (untyped) -> DatabaseAdapter def self.for: (untyped) -> DatabaseAdapter - @connection_pool: untyped - @fixed_connection: untyped + @connection_pool: untyped + # @rbs (untyped) -> void def initialize: (untyped) -> void @@ -24,6 +24,12 @@ module SolidObjects # @rbs () -> Time def database_now: () -> Time + # @rbs () { () -> untyped } -> untyped + def with_lock_retry: () { () -> untyped } -> untyped + + # @rbs () { () -> untyped } -> untyped + def with_lock_probe: () { () -> untyped } -> untyped + # @rbs () { () -> untyped } -> untyped def transaction: () { () -> untyped } -> untyped diff --git a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs index 513c6f5..28fcd3f 100644 --- a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs +++ b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs @@ -3,14 +3,18 @@ module SolidObjects module DatabaseAdapters class Sqlite < DatabaseAdapter - LOCK_RETRY_INTERVAL: ::Float - # @rbs () -> String def current_time_expression: () -> String # @rbs () { () -> untyped } -> untyped def transaction: () { () -> untyped } -> untyped + # @rbs () { () -> untyped } -> untyped + def with_lock_retry: () { () -> untyped } -> untyped + + # @rbs () { () -> untyped } -> untyped + def with_lock_probe: () { () -> untyped } -> untyped + private # @rbs (untyped) { () -> untyped } -> untyped @@ -18,6 +22,9 @@ module SolidObjects # @rbs (Exception) -> bool def deadline_error?: (Exception) -> bool + + # @rbs () -> void + def yield_before_retry: () -> void end end end diff --git a/sig/generated/lib/solid_objects/sync_diagnostics.rbs b/sig/generated/lib/solid_objects/sync_diagnostics.rbs index a5d1b22..5fc0568 100644 --- a/sig/generated/lib/solid_objects/sync_diagnostics.rbs +++ b/sig/generated/lib/solid_objects/sync_diagnostics.rbs @@ -5,8 +5,17 @@ module SolidObjects # @rbs (Message, timeout: Numeric) -> SyncTimeout def call: (Message, timeout: Numeric) -> SyncTimeout + # @rbs (Message, timeout: Numeric) -> SyncTimeout + def database_contention: (Message, timeout: Numeric) -> SyncTimeout + + # @rbs (MessageReference, timeout: Numeric) -> SyncTimeout + def database_contention_for: (MessageReference, timeout: Numeric) -> SyncTimeout + private + # @rbs (Message, timeout: Numeric, status: String, waiting_on: String, activation: Hash[String, untyped], blocker: Hash[String, untyped]?) -> SyncTimeout + def build_error: (Message, timeout: Numeric, status: String, waiting_on: String, activation: Hash[String, untyped], blocker: Hash[String, untyped]?) -> SyncTimeout + # @rbs (Message) -> String def message_status: (Message) -> String diff --git a/sig/generated/lib/solid_objects/synchronous_invocation.rbs b/sig/generated/lib/solid_objects/synchronous_invocation.rbs index 8943e9c..15e11fc 100644 --- a/sig/generated/lib/solid_objects/synchronous_invocation.rbs +++ b/sig/generated/lib/solid_objects/synchronous_invocation.rbs @@ -13,6 +13,9 @@ module SolidObjects # @rbs (MessageReference) -> Message def load_message: (MessageReference) -> Message + # @rbs (MessageReference) -> Message? + def load_message_at_deadline: (MessageReference) -> Message? + # @rbs (Message) -> untyped def completed_result: (Message) -> untyped @@ -25,6 +28,12 @@ module SolidObjects # @rbs (Numeric) -> void def wait: (Numeric) -> void + # @rbs (Message, timeout: Numeric) -> SyncTimeout + def diagnose_timeout: (Message, timeout: Numeric) -> SyncTimeout + + # @rbs (Message?, MessageReference, timeout: Numeric) -> SyncTimeout + def contention_timeout: (Message?, MessageReference, timeout: Numeric) -> SyncTimeout + # @rbs () -> Float def monotonic_now: () -> Float end diff --git a/test/dummy/app/actors/cli_worker_actor.rb b/test/dummy/app/actors/cli_worker_actor.rb new file mode 100644 index 0000000..fae6c75 --- /dev/null +++ b/test/dummy/app/actors/cli_worker_actor.rb @@ -0,0 +1,9 @@ +# frozen_string_literal: true + +class CliWorkerActor < SolidObjects::Actor + attribute :completed, default: false + + def complete + self.completed = true + end +end diff --git a/test/dummy/config/database.yml b/test/dummy/config/database.yml index 2cfeb7a..f5000d3 100644 --- a/test/dummy/config/database.yml +++ b/test/dummy/config/database.yml @@ -1,3 +1,3 @@ test: adapter: sqlite3 - database: ":memory:" + database: "<%= ENV.fetch("SOLID_OBJECTS_DUMMY_DATABASE", ":memory:") %>" diff --git a/test/dummy/config/initializers/cli_worker_probe.rb b/test/dummy/config/initializers/cli_worker_probe.rb new file mode 100644 index 0000000..82b1e2f --- /dev/null +++ b/test/dummy/config/initializers/cli_worker_probe.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +if ENV["SOLID_OBJECTS_CLI_WORKER_PROBE"] + ActiveSupport::Notifications.subscribe("solid_objects.message.completed") do |event| + next unless event.payload.fetch(:actor_type) == "CliWorkerActor" + + File.write(ENV.fetch("SOLID_OBJECTS_CLI_WORKER_PROBE"), event.payload.fetch(:message_id)) + Process.kill("TERM", Process.pid) + end +end diff --git a/test/dummy/prepare_cli_worker.rb b/test/dummy/prepare_cli_worker.rb new file mode 100644 index 0000000..3a3d7bf --- /dev/null +++ b/test/dummy/prepare_cli_worker.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +ENV["RAILS_ENV"] = "test" + +require_relative "config/environment" +require_relative "../../db/migrate/20260805000000_create_solid_objects_tables" +require_relative "../../db/migrate/20260806000000_add_state_revision_to_solid_objects_instances" + +CreateSolidObjectsTables.new.migrate(:up) +AddStateRevisionToSolidObjectsInstances.new.migrate(:up) + +now = Time.current +instance = SolidObjects::Instance.create!( + actor_type: "CliWorkerActor", + actor_id: "only-in-app-actors", + state: {}, + state_version: 1 +) +message = SolidObjects::Message.create!( + instance:, + actor_type: instance.actor_type, + actor_id: instance.actor_id, + message_name: "complete", + message_kind: "async", + arguments: {}, + sequence: 1, + max_attempts: 1, + request_id: SecureRandom.uuid, + enqueued_at: now, + available_at: now +) +SolidObjects::ReadyMessage.create!( + message:, + instance:, + sequence: message.sequence, + available_at: now +) + +puts message.id diff --git a/test/integration/cli_test.rb b/test/integration/cli_test.rb index af12c35..ce0184e 100644 --- a/test/integration/cli_test.rb +++ b/test/integration/cli_test.rb @@ -3,6 +3,8 @@ require "test_helper" require "open3" require "solid_objects/cli" +require "sqlite3" +require "tmpdir" class CLITest < ActiveSupport::TestCase test "documents the operational commands" do @@ -41,4 +43,59 @@ class CLITest < ActiveSupport::TestCase assert_raises(SolidObjects::Unauthorized) { command.prune_messages } end + + test "start loads and processes application actors when eager loading is disabled" do + Dir.mktmpdir("solid-objects-cli") do |directory| + database = File.join(directory, "dummy.sqlite3") + completed = File.join(directory, "completed") + environment = { + "BUNDLE_GEMFILE" => File.expand_path("../../Gemfile", __dir__), + "RAILS_ENV" => "test", + "SOLID_OBJECTS_CLI_WORKER_PROBE" => completed, + "SOLID_OBJECTS_DUMMY_DATABASE" => database + } + dummy_root = File.expand_path("../dummy", __dir__) + prepare = File.join(dummy_root, "prepare_cli_worker.rb") + _output, prepare_error, prepare_status = Open3.capture3( + environment, + Gem.ruby, + prepare, + chdir: dummy_root + ) + assert prepare_status.success?, prepare_error + + _input, output, error_output, wait_thread = Open3.popen3( + environment, + "bundle", + "exec", + "solid_objects", + "start", + "--workers", + "1", + "--effect-workers", + "0", + "--broadcast-workers", + "0", + "--reminder-schedulers", + "0", + chdir: dummy_root + ) + status = Timeout.timeout(10) { wait_thread.value } + + assert status.success?, [ output.read, error_output.read ].join("\n") + assert File.exist?(completed) + + connection = SQLite3::Database.new(database) + state, completed_at = connection.get_first_row( + "SELECT state, completed_at FROM solid_objects_messages INNER JOIN " \ + "solid_objects_instances ON solid_objects_instances.id = solid_objects_messages.instance_id" + ) + assert_equal({ "completed" => true }, JSON.parse(state)) + assert completed_at + ensure + connection&.close + Process.kill("TERM", wait_thread.pid) if wait_thread&.alive? + wait_thread&.join + end + end end diff --git a/test/integration/synchronous_invocation_test.rb b/test/integration/synchronous_invocation_test.rb index 2a7d8fc..4e96967 100644 --- a/test/integration/synchronous_invocation_test.rb +++ b/test/integration/synchronous_invocation_test.rb @@ -105,6 +105,21 @@ def run end end + class LockRetryActor < SolidObjects::Actor + actor_type "synchronous-lock-retry" + + class << self + attr_accessor :executions + end + + attribute :value, default: 0 + + def increment + self.class.executions += 1 + self.value += 1 + end + end + class BlockingWakeUp # @rbs @waiting: Thread::Queue # @rbs @release: Thread::Queue @@ -133,6 +148,7 @@ def wait(timeout:) BlockingActor.release = Queue.new DeadlineActor.reached = Queue.new DeadlineActor.continue = Queue.new + LockRetryActor.executions = 0 end test "direct actor method durably executes without a worker" do @@ -456,6 +472,63 @@ def wait(timeout:) blocker&.join end + test "sync bounds SQLite contention while registering its caller process" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + + message_reference = SolidObjects::Mailbox.new.enqueue( + LockRetryActor.ref("registration"), + :increment, + {}, + kind: "sync" + ) + lock = hold_sqlite_write_lock + + error, elapsed = invoke_with_immediate_sqlite_lock_failure(message_reference) + + assert_instance_of SolidObjects::SyncTimeout, error + assert_equal message_reference.id, error.message_id + assert_operator elapsed, :<, 0.5 + assert_equal 0, LockRetryActor.executions + + release_sqlite_write_lock(lock) + lock = nil + + assert_equal 1, message_reference.wait(timeout: 1) + assert_equal 1, LockRetryActor.executions + ensure + release_sqlite_write_lock(lock) if lock + end + + test "sync bounds SQLite contention while reusing and heartbeating its caller process" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + + SolidObjects.configuration.process_heartbeat_interval = 0 + process_record = SolidObjects.caller_process.process_registry.process_record + message_reference = SolidObjects::Mailbox.new.enqueue( + LockRetryActor.ref("heartbeat"), + :increment, + {}, + kind: "sync" + ) + lock = hold_sqlite_write_lock + + error, elapsed = invoke_with_immediate_sqlite_lock_failure(message_reference) + + assert_instance_of SolidObjects::SyncTimeout, error + assert_equal message_reference.id, error.message_id + assert_operator elapsed, :<, 0.5 + assert_equal 0, LockRetryActor.executions + + release_sqlite_write_lock(lock) + lock = nil + + assert_equal process_record.id, SolidObjects.caller_process.process_registry.process_record.id + assert_equal 1, message_reference.wait(timeout: 1) + assert_equal 1, LockRetryActor.executions + ensure + release_sqlite_write_lock(lock) if lock + end + test "concurrent sync calls for one actor execute sequentially" do wake_up = BlockingWakeUp.new SolidObjects.configuration.wake_up_adapter = wake_up @@ -650,6 +723,59 @@ def monotonic_now ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) end + def hold_sqlite_write_lock + locked = Queue.new + release = Queue.new + thread = Thread.new do + SolidObjects::Record.connection_pool.with_connection do + SolidObjects::Record.transaction do + SolidObjects::Process.create!( + id: SecureRandom.uuid, + kind: "lock-holder", + hostname: "test-host", + pid: ::Process.pid, + started_at: Time.current, + last_heartbeat_at: Time.current, + metadata: {} + ) + locked << true + release.pop + end + end + end + Timeout.timeout(2) { locked.pop } + [ thread, release ] + end + + def release_sqlite_write_lock(lock) + thread, release = lock + release << true + thread.join + end + + def invoke_with_immediate_sqlite_lock_failure(message_reference) + result = Queue.new + invocation = Thread.new do + error = nil + elapsed = nil + SolidObjects::Record.connection_pool.with_connection do |connection| + previous_timeout = connection.select_value("PRAGMA busy_timeout").to_i + connection.execute("PRAGMA busy_timeout = 0") + started_at = monotonic_now + error = capture_exception do + SolidObjects::SynchronousInvocation.new.call(message_reference, timeout: 0.1) + end + elapsed = monotonic_now - started_at + ensure + connection.execute("PRAGMA busy_timeout = #{previous_timeout}") + end + result << [ error, elapsed ] + end + captured = Timeout.timeout(2) { result.pop } + invocation.join + captured + end + def actor_instance(actor_id) SolidObjects::Instance.find_by!( actor_type: "synchronous-counter", diff --git a/test/unit/actor_registry_test.rb b/test/unit/actor_registry_test.rb index 5243ba6..9989393 100644 --- a/test/unit/actor_registry_test.rb +++ b/test/unit/actor_registry_test.rb @@ -23,6 +23,31 @@ class RegisteredActor < SolidObjects::Actor end end + test "replaces a reloaded actor class with the same name" do + actor_name = "ReloadedRegistryActor" + Object.const_set(actor_name, Class.new(SolidObjects::Actor)) + original_actor = Object.const_get(actor_name) + SolidObjects.register_actor("reloadable", original_actor) + Object.send(:remove_const, actor_name) + Object.const_set(actor_name, Class.new(SolidObjects::Actor)) + reloaded_actor = Object.const_get(actor_name) + + SolidObjects.register_actor("reloadable", reloaded_actor) + + assert_equal reloaded_actor, SolidObjects.registry.fetch("reloadable") + ensure + Object.send(:remove_const, actor_name) if actor_name && Object.const_defined?(actor_name, false) + end + + test "rejects distinct anonymous actors for the same actor type" do + registry = SolidObjects::ActorRegistry.new + registry.register("anonymous", Class.new(SolidObjects::Actor)) + + assert_raises(SolidObjects::InvalidActor) do + registry.register("anonymous", Class.new(SolidObjects::Actor)) + end + end + test "rejects classes that are not actors" do assert_raises(SolidObjects::InvalidActor) do SolidObjects.register_actor("not-an-actor", String) From a37e8d2f24f3ef4f4fd2a4b67fc77b85fb285652 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Fri, 7 Aug 2026 10:02:52 -0700 Subject: [PATCH 2/2] fix: pace SQLite lock retries Yield between deadline-aware lock attempts so a contended caller does not hammer SQLite while preserving the original sync deadline. Assert bounded retry pressure without re-executing actor behavior. --- lib/solid_objects/database_adapters/sqlite.rb | 17 +++++++++++---- .../database_adapters/sqlite.rbs | 8 ++++++- .../synchronous_invocation_test.rb | 21 ++++++++++++++++--- 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/lib/solid_objects/database_adapters/sqlite.rb b/lib/solid_objects/database_adapters/sqlite.rb index 6f83405..aea77b6 100644 --- a/lib/solid_objects/database_adapters/sqlite.rb +++ b/lib/solid_objects/database_adapters/sqlite.rb @@ -3,6 +3,10 @@ module SolidObjects module DatabaseAdapters class Sqlite < DatabaseAdapter + LOCK_RETRY_INTERVAL = 0.001 + LOCK_RETRY_MUTEX = Thread::Mutex.new + LOCK_RETRY_CONDITION = Thread::ConditionVariable.new + # @rbs () -> String def current_time_expression "STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')" @@ -27,7 +31,7 @@ def with_lock_retry rescue DatabaseDeadlineExceeded raise if SyncDeadline.expired? - yield_before_retry + wait_before_retry retry rescue => error raise unless deadline_error?(error) @@ -38,7 +42,7 @@ def with_lock_retry cause: error end - yield_before_retry + wait_before_retry retry end @@ -84,8 +88,13 @@ def deadline_error?(error) end # @rbs () -> void - def yield_before_retry - Thread.pass + def wait_before_retry + LOCK_RETRY_MUTEX.synchronize do + LOCK_RETRY_CONDITION.wait( + LOCK_RETRY_MUTEX, + [ LOCK_RETRY_INTERVAL, SyncDeadline.remaining ].min + ) + end end end end diff --git a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs index 28fcd3f..e186b12 100644 --- a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs +++ b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs @@ -3,6 +3,12 @@ module SolidObjects module DatabaseAdapters class Sqlite < DatabaseAdapter + LOCK_RETRY_INTERVAL: ::Float + + LOCK_RETRY_MUTEX: untyped + + LOCK_RETRY_CONDITION: untyped + # @rbs () -> String def current_time_expression: () -> String @@ -24,7 +30,7 @@ module SolidObjects def deadline_error?: (Exception) -> bool # @rbs () -> void - def yield_before_retry: () -> void + def wait_before_retry: () -> void end end end diff --git a/test/integration/synchronous_invocation_test.rb b/test/integration/synchronous_invocation_test.rb index 4e96967..408a968 100644 --- a/test/integration/synchronous_invocation_test.rb +++ b/test/integration/synchronous_invocation_test.rb @@ -2,6 +2,7 @@ require "database_test_helper" require "solid_objects/mailbox" +require "solid_objects/synchronous_invocation" require "timeout" class SynchronousInvocationTest < ActiveSupport::TestCase @@ -483,11 +484,13 @@ def wait(timeout:) ) lock = hold_sqlite_write_lock - error, elapsed = invoke_with_immediate_sqlite_lock_failure(message_reference) + error, elapsed, attempts = invoke_with_immediate_sqlite_lock_failure(message_reference) assert_instance_of SolidObjects::SyncTimeout, error assert_equal message_reference.id, error.message_id assert_operator elapsed, :<, 0.5 + assert_operator attempts, :>, 1 + assert_operator attempts, :<, 200 assert_equal 0, LockRetryActor.executions release_sqlite_write_lock(lock) @@ -512,11 +515,13 @@ def wait(timeout:) ) lock = hold_sqlite_write_lock - error, elapsed = invoke_with_immediate_sqlite_lock_failure(message_reference) + error, elapsed, attempts = invoke_with_immediate_sqlite_lock_failure(message_reference) assert_instance_of SolidObjects::SyncTimeout, error assert_equal message_reference.id, error.message_id assert_operator elapsed, :<, 0.5 + assert_operator attempts, :>, 1 + assert_operator attempts, :<, 200 assert_equal 0, LockRetryActor.executions release_sqlite_write_lock(lock) @@ -758,6 +763,10 @@ def invoke_with_immediate_sqlite_lock_failure(message_reference) invocation = Thread.new do error = nil elapsed = nil + attempts = 0 + subscription = ActiveSupport::Notifications.subscribe("sql.active_record") do |event| + attempts += 1 if process_write?(event.payload) + end SolidObjects::Record.connection_pool.with_connection do |connection| previous_timeout = connection.select_value("PRAGMA busy_timeout").to_i connection.execute("PRAGMA busy_timeout = 0") @@ -768,14 +777,20 @@ def invoke_with_immediate_sqlite_lock_failure(message_reference) elapsed = monotonic_now - started_at ensure connection.execute("PRAGMA busy_timeout = #{previous_timeout}") + ActiveSupport::Notifications.unsubscribe(subscription) end - result << [ error, elapsed ] + result << [ error, elapsed, attempts ] end captured = Timeout.timeout(2) { result.pop } invocation.join captured end + def process_write?(payload) + payload.fetch(:sql).match?(/\A(?:INSERT|UPDATE)/) && + payload.fetch(:sql).include?(SolidObjects::Process.table_name) + end + def actor_instance(actor_id) SolidObjects::Instance.find_by!( actor_type: "synchronous-counter",