diff --git a/CHANGELOG.md b/CHANGELOG.md index e81029a..6b22b15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,29 @@ # Changelog +## 0.5.1 - 2026-08-07 + +- Restore the SQLite busy wait that a synchronous invocation suspends for its + deadline. Rails installs the busy wait as a Ruby busy handler through the + sqlite3 `timeout` configuration, which `PRAGMA busy_timeout` reports as zero + and silently replaces, so the previous save and restore left pooled + connections with no busy handler at all. Every later writer on that + connection, inside or outside Solid Objects, then failed immediately with + `SQLite3::BusyException` instead of waiting for the lock. Suspend the busy + wait only when the adapter can identify how to restore it, so an Active + Record release that stops exposing the configured timeout loosens + synchronous deadline bounds instead of stripping lock waiting from a shared + pooled connection. + +- Run the doctor round-trip probe on a dedicated caller process, and accept an + explicit process registry in `SynchronousInvocation`, so the probe can no + longer stop and delete a shared application caller process, release its + activations, and unclaim its messages. +- Report doctor probe cleanup failures as a failed or warned check instead of + raising a database lock error out of the command and leaking the probe + caller process. +- Instrument component refreshes with actor identity, component name, key, + dependencies, refresh method, revision, and outcome, excluding locals. + ## 0.5.0 - 2026-08-07 - Add repeatable reactive components with signed string or integer keys and diff --git a/Gemfile.lock b/Gemfile.lock index 00cf7a9..77cc4ad 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_objects (0.5.0) + solid_objects (0.5.1) 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.5.0) + solid_objects (0.5.1) 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/app/controllers/solid_objects/components_controller.rb b/app/controllers/solid_objects/components_controller.rb index d593051..688953c 100644 --- a/app/controllers/solid_objects/components_controller.rb +++ b/app/controllers/solid_objects/components_controller.rb @@ -8,12 +8,25 @@ class ComponentsController < ActionController::Base # @rbs () -> void def show + SolidObjects.instrument(:"component.refreshed") { |payload| refresh(payload) } + end + + private + + # @rbs (Hash[Symbol, untyped]) -> void + def refresh(payload) registration = ComponentRegistration.from_token( params.require(:token) ) + payload.merge!(registration_payload(registration)) requested_revision = requested_revision_key snapshot = ActorSnapshot.new(registration.reference) - return head :conflict if newer_than_snapshot?(requested_revision, snapshot) + payload[:instance_id] = snapshot.instance_id + payload[:revision] = snapshot.revision + if newer_than_snapshot?(requested_revision, snapshot) + payload[:outcome] = "conflict" + return head :conflict + end authorization_context = SolidObjects .configuration @@ -26,18 +39,32 @@ def show authorization_context: ).call response.headers["Cache-Control"] = "private, no-store" + payload[:outcome] = "rendered" render html: component_frame(registration, snapshot, rendered) rescue Unauthorized + payload[:outcome] = "unauthorized" head :forbidden rescue UnknownComponent + payload[:outcome] = "unknown_component" head :not_found rescue ActionController::ParameterMissing, ArgumentError, InvalidComponentToken + payload[:outcome] = "invalid_token" head :bad_request end - private + # @rbs (ComponentRegistration) -> Hash[Symbol, untyped] + def registration_payload(registration) + { + actor_type: registration.reference.actor_type, + actor_id: registration.reference.actor_id, + component_name: registration.component_name, + component_key: registration.component_key, + dependencies: registration.dependencies, + refresh_method: registration.refresh_method + } + end # @rbs () -> Array[Integer] def requested_revision_key diff --git a/docs/correctness.md b/docs/correctness.md index 39f4571..ad26e28 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -167,7 +167,21 @@ 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 +be inspected at timeout. To keep those retries in Ruby, the SQLite adapter +suspends the connection's busy wait for the duration of each deadline-bound +transaction and restores it afterwards. Restoration reinstalls the Ruby busy +handler Rails configures from the sqlite3 `timeout` setting, which +`PRAGMA busy_timeout` neither reports nor preserves, so a synchronous call +leaves the connection's lock waiting behaviour exactly as it found it for +later writers inside and outside Solid Objects. + +The adapter suspends the busy wait only when it can identify how to restore +it. When a future Active Record release stops exposing the configured +timeout, the adapter leaves the connection untouched: synchronous deadlines +lose their tight bound and wait as long as the configured busy wait allows, +rather than stripping lock waiting from a pooled connection the rest of the +application shares. A test asserts the timeout stays discoverable so the +looser bound cannot be adopted silently. 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. diff --git a/docs/operations.md b/docs/operations.md index 7c597d1..afd8931 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -15,6 +15,14 @@ so the schema check compares the required shape instead of a fixed timestamp. Warnings such as an all-deny neutral policy do not fail the command because a context-aware production policy may correctly deny the probe. +The round-trip probe runs on its own dedicated caller process rather than the +shared application caller process, and removes that record together with its +temporary actor when it finishes. Running the doctor inside a process that +already serves synchronous calls therefore leaves the application caller +process, its activations, and its claimed messages untouched, including when an +application call overlaps the probe. A database busy enough to block cleanup +reports a failed or warned check rather than raising out of the command. + ## Runtime Start all configured roles: @@ -147,10 +155,20 @@ transaction rejection, commit-action start/completion/failure, effect and broadcast enqueue/completion, reminder enqueue, actor destruction/expiration, retention pruning, process cleanup, and supervisor lifecycle. +`solid_objects.component.refreshed` covers every authorized component refresh +request. Its payload carries the actor identity, `component_name`, +`component_key`, declared `dependencies`, `refresh_method`, the rendered +`instance_id` and `revision`, and an `outcome` of `rendered`, `conflict`, +`unauthorized`, `unknown_component`, or `invalid_token`. Use it to watch +refresh rate per key, authorization denials, superseded requests, and render +duration. A rejected token reports only the outcome, since no signed identity +was recovered. + Payloads contain stable runtime identifiers, actor identity, sequence, attempts, ownership generations, and safe exception summaries where relevant. -Arguments, actor state, results, and outbox payloads are excluded. The bundled -log subscriber turns the same notifications into structured logger hashes. +Arguments, component locals, actor state, results, and outbox payloads are +excluded. The bundled log subscriber turns the same notifications into +structured logger hashes. ## Retention and backups diff --git a/lib/solid_objects/database_adapters/sqlite.rb b/lib/solid_objects/database_adapters/sqlite.rb index aea77b6..01e9bfe 100644 --- a/lib/solid_objects/database_adapters/sqlite.rb +++ b/lib/solid_objects/database_adapters/sqlite.rb @@ -67,11 +67,49 @@ def with_lock_probe def with_transaction_deadline(connection) return yield unless SyncDeadline.active? - previous_timeout = connection.select_value("PRAGMA busy_timeout").to_i - connection.execute("PRAGMA busy_timeout = 0") - yield - ensure - connection.execute("PRAGMA busy_timeout = #{previous_timeout}") if previous_timeout + busy_wait = restorable_busy_wait(connection) + return yield unless busy_wait + + begin + connection.execute("PRAGMA busy_timeout = 0") + yield + ensure + restore_busy_wait(connection, busy_wait) + end + end + + # @rbs (untyped) -> Hash[Symbol, untyped]? + def restorable_busy_wait(connection) + pragma_timeout = connection.select_value("PRAGMA busy_timeout").to_i + return { pragma_timeout: } if pragma_timeout.positive? + + handler_timeout = configured_busy_handler_timeout(connection) + return nil unless handler_timeout + + { pragma_timeout:, handler_timeout: } + end + + # @rbs (untyped, Hash[Symbol, untyped]) -> void + def restore_busy_wait(connection, busy_wait) + handler_timeout = busy_wait[:handler_timeout] + if handler_timeout + connection.raw_connection.busy_handler_timeout = handler_timeout + return + end + + connection.execute("PRAGMA busy_timeout = #{busy_wait.fetch(:pragma_timeout)}") + end + + # @rbs (untyped) -> Integer? + def configured_busy_handler_timeout(connection) + return nil unless connection.respond_to?(:raw_connection) + return nil unless connection.raw_connection.respond_to?(:busy_handler_timeout=) + + pool = connection.respond_to?(:pool) ? connection.pool : nil + return nil unless pool.respond_to?(:db_config) + + timeout = pool.db_config.configuration_hash[:timeout] + timeout&.to_i end # @rbs (Exception) -> bool diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb index 5966aec..c13dc4d 100644 --- a/lib/solid_objects/doctor.rb +++ b/lib/solid_objects/doctor.rb @@ -215,25 +215,63 @@ def check_runtime # @rbs () -> Check def check_sync_round_trip actor_id = SecureRandom.uuid + probe_registry = ProcessRegistry.new + check = run_sync_probe(actor_id, probe_registry) + leftovers = remove_probe_records(actor_id:, probe_registry:) + return check if leftovers.empty? || check.failed? + + warn_check( + :sync_round_trip, + "#{check.message}; could not remove the #{leftovers.join(" and ")}" + ) + end + + # @rbs (String, ProcessRegistry) -> Check + def run_sync_probe(actor_id, probe_registry) + probe_registry.register(kind: "caller", metadata: { execution: "doctor" }) value = SecureRandom.hex(8) - process_registry = SolidObjects.caller_process.process_registry - reference = ProbeActor.ref(actor_id) message_reference = Mailbox.new.enqueue( - reference, + ProbeActor.ref(actor_id), :ping, { value: }, kind: "sync" ) - result = SynchronousInvocation.new.call(message_reference, timeout: 5.seconds) + result = SynchronousInvocation + .new(process_registry: probe_registry) + .call(message_reference, timeout: 5.seconds) raise Error, "unexpected round-trip result" unless result == value pass(:sync_round_trip, "durable synchronous actor call completed without a worker") rescue => error fail_check(:sync_round_trip, "#{error.class}: #{error.message}") - ensure - Instance.where(actor_type: ProbeActor.actor_type, actor_id:).delete_all if actor_id - process_registry&.stop - process_registry&.process_record&.delete + end + + # @rbs (actor_id: String, probe_registry: ProcessRegistry) -> Array[String] + def remove_probe_records(actor_id:, probe_registry:) + leftovers = [] + leftovers << "probe actor" unless delete_probe_actor(actor_id) + leftovers << "probe caller process" unless delete_probe_caller_process(probe_registry) + leftovers + end + + # @rbs (String) -> bool + def delete_probe_actor(actor_id) + Instance.where(actor_type: ProbeActor.actor_type, actor_id:).delete_all + true + rescue + false + end + + # @rbs (ProcessRegistry) -> bool + def delete_probe_caller_process(probe_registry) + process_record = probe_registry.process_record + return true unless process_record + + probe_registry.stop + process_record.delete + true + rescue + false end # @rbs (Check, Check) -> bool diff --git a/lib/solid_objects/synchronous_invocation.rb b/lib/solid_objects/synchronous_invocation.rb index 41b5a45..df81bc8 100644 --- a/lib/solid_objects/synchronous_invocation.rb +++ b/lib/solid_objects/synchronous_invocation.rb @@ -6,6 +6,13 @@ module SolidObjects class SynchronousInvocation + # @rbs @dedicated_process_registry: ProcessRegistry? + + # @rbs (?process_registry: ProcessRegistry?) -> void + def initialize(process_registry: nil) + @dedicated_process_registry = process_registry + end + # @rbs (MessageReference, timeout: Numeric) -> untyped def call(message_reference, timeout:) return call_before_deadline(message_reference, timeout:) if SyncDeadline.active? @@ -92,9 +99,17 @@ def raise_rejection(message) ) end + # @rbs () -> ProcessRegistry + def process_registry + dedicated_registry = @dedicated_process_registry + return SolidObjects.caller_process.process_registry unless dedicated_registry + + dedicated_registry.tap(&:heartbeat) + end + # @rbs (Message, deadline: Float) -> Integer def assist(message, deadline:) - process_registry = SolidObjects.caller_process.process_registry + process_registry = self.process_registry activation = ActivationManager .new(owner_id: process_registry.process_record.id) .claim(instance_id: message.instance_id) diff --git a/lib/solid_objects/version.rb b/lib/solid_objects/version.rb index 2942531..4e07a1e 100644 --- a/lib/solid_objects/version.rb +++ b/lib/solid_objects/version.rb @@ -1,5 +1,5 @@ # rbs_inline: enabled module SolidObjects - VERSION = "0.5.0" + VERSION = "0.5.1" end diff --git a/sig/generated/controllers/solid_objects/components_controller.rbs b/sig/generated/controllers/solid_objects/components_controller.rbs index 87b582e..09d4239 100644 --- a/sig/generated/controllers/solid_objects/components_controller.rbs +++ b/sig/generated/controllers/solid_objects/components_controller.rbs @@ -7,6 +7,12 @@ module SolidObjects private + # @rbs (Hash[Symbol, untyped]) -> void + def refresh: (Hash[Symbol, untyped]) -> void + + # @rbs (ComponentRegistration) -> Hash[Symbol, untyped] + def registration_payload: (ComponentRegistration) -> Hash[Symbol, untyped] + # @rbs () -> Array[Integer] def requested_revision_key: () -> Array[Integer] diff --git a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs index e186b12..225bef9 100644 --- a/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs +++ b/sig/generated/lib/solid_objects/database_adapters/sqlite.rbs @@ -26,6 +26,15 @@ module SolidObjects # @rbs (untyped) { () -> untyped } -> untyped def with_transaction_deadline: (untyped) { () -> untyped } -> untyped + # @rbs (untyped) -> Hash[Symbol, untyped]? + def restorable_busy_wait: (untyped) -> Hash[Symbol, untyped]? + + # @rbs (untyped, Hash[Symbol, untyped]) -> void + def restore_busy_wait: (untyped, Hash[Symbol, untyped]) -> void + + # @rbs (untyped) -> Integer? + def configured_busy_handler_timeout: (untyped) -> Integer? + # @rbs (Exception) -> bool def deadline_error?: (Exception) -> bool diff --git a/sig/generated/lib/solid_objects/doctor.rbs b/sig/generated/lib/solid_objects/doctor.rbs index b5e52ba..e8d4342 100644 --- a/sig/generated/lib/solid_objects/doctor.rbs +++ b/sig/generated/lib/solid_objects/doctor.rbs @@ -78,6 +78,18 @@ module SolidObjects # @rbs () -> Check def check_sync_round_trip: () -> Check + # @rbs (String, ProcessRegistry) -> Check + def run_sync_probe: (String, ProcessRegistry) -> Check + + # @rbs (actor_id: String, probe_registry: ProcessRegistry) -> Array[String] + def remove_probe_records: (actor_id: String, probe_registry: ProcessRegistry) -> Array[String] + + # @rbs (String) -> bool + def delete_probe_actor: (String) -> bool + + # @rbs (ProcessRegistry) -> bool + def delete_probe_caller_process: (ProcessRegistry) -> bool + # @rbs (Check, Check) -> bool def ready_for_round_trip?: (Check, Check) -> bool diff --git a/sig/generated/lib/solid_objects/synchronous_invocation.rbs b/sig/generated/lib/solid_objects/synchronous_invocation.rbs index 15e11fc..ab11534 100644 --- a/sig/generated/lib/solid_objects/synchronous_invocation.rbs +++ b/sig/generated/lib/solid_objects/synchronous_invocation.rbs @@ -2,6 +2,11 @@ module SolidObjects class SynchronousInvocation + @dedicated_process_registry: ProcessRegistry? + + # @rbs (?process_registry: ProcessRegistry?) -> void + def initialize: (?process_registry: ProcessRegistry?) -> void + # @rbs (MessageReference, timeout: Numeric) -> untyped def call: (MessageReference, timeout: Numeric) -> untyped @@ -22,6 +27,9 @@ module SolidObjects # @rbs (Message) -> bot def raise_rejection: (Message) -> bot + # @rbs () -> ProcessRegistry + def process_registry: () -> ProcessRegistry + # @rbs (Message, deadline: Float) -> Integer def assist: (Message, deadline: Float) -> Integer diff --git a/test/database_test_helper.rb b/test/database_test_helper.rb index fddca8e..4de539d 100644 --- a/test/database_test_helper.rb +++ b/test/database_test_helper.rb @@ -56,6 +56,31 @@ class ActiveSupport::TestCase SolidObjects::Process.delete_all SolidObjectsTestDomainRecord.delete_all end + + def with_immediate_sqlite_lock_failure(&block) + SolidObjects::Record.connection_pool.with_connection do |connection| + suspend_sqlite_busy_wait(connection, &block) + end + end + + def suspend_sqlite_busy_wait(connection) + database_adapter = SolidObjects.database_adapter + database_adapter.define_singleton_method(:configured_busy_handler_timeout) { |_connection| 0 } + connection.raw_connection.busy_handler_timeout = 0 + yield + ensure + database_adapter.singleton_class.send(:remove_method, :configured_busy_handler_timeout) + connection.raw_connection.busy_handler_timeout = configured_sqlite_busy_handler_timeout + end + + def configured_sqlite_busy_handler_timeout + SolidObjects::Record + .connection_pool + .db_config + .configuration_hash + .fetch(:timeout, 5_000) + .to_i + end end Minitest.after_run do diff --git a/test/integration/components_controller_test.rb b/test/integration/components_controller_test.rb index a47ba97..1b85955 100644 --- a/test/integration/components_controller_test.rb +++ b/test/integration/components_controller_test.rb @@ -292,8 +292,96 @@ def update_room(messages:, status:) assert_includes @response.body, %(data-solid-objects-refresh="morph") end + test "instruments an authorized component refresh without its locals" do + reference = RoomActor.ref("general") + reference.replace_messages(messages: [ { id: "1", body: "First" } ]) + token = component_token( + reference, + component_name: "player", + component_key: "alice", + dependencies: %w[status], + locals: { player_id: "alice", label: "You" }, + refresh_method: "morph" + ) + event = capture_component_event { render_component(token, viewer: "alice") } + + assert_response :success + assert_equal "component-room", event.payload.fetch(:actor_type) + assert_equal "general", event.payload.fetch(:actor_id) + assert_equal "player", event.payload.fetch(:component_name) + assert_equal "alice", event.payload.fetch(:component_key) + assert_equal %w[status], event.payload.fetch(:dependencies) + assert_equal "morph", event.payload.fetch(:refresh_method) + assert_equal "rendered", event.payload.fetch(:outcome) + assert_equal( + SolidObjects::Instance.find_by(actor_type: "component-room", actor_id: "general").state_revision, + event.payload.fetch(:revision) + ) + assert event.payload.fetch(:instance_id) + assert event.duration + refute event.payload.key?(:locals) + refute event.payload.key?(:token) + end + + test "instruments a denied component refresh" do + reference = RoomActor.ref("general") + SolidObjects.configuration.authorize_query = ->(**) { false } + token = component_token( + reference, + component_name: "messages", + dependencies: %w[recent_messages] + ) + + event = capture_component_event { render_component(token, viewer: "mallory") } + + assert_response :forbidden + assert_equal "unauthorized", event.payload.fetch(:outcome) + assert_equal "messages", event.payload.fetch(:component_name) + end + + test "instruments a superseded component refresh" do + reference = RoomActor.ref("general") + token = component_token( + reference, + component_name: "messages", + dependencies: %w[recent_messages] + ) + registration = SolidObjects::ComponentRegistration.from_token(token) + @request.headers["HTTP_X_VIEWER"] = "alice" + + event = capture_component_event do + get :show, params: { + token:, + instance_id: registration.instance_id + 1, + revision: registration.revision + } + end + + assert_response :conflict + assert_equal "conflict", event.payload.fetch(:outcome) + end + + test "instruments a rejected component token" do + event = capture_component_event { render_component("malformed", viewer: "alice") } + + assert_response :bad_request + assert_equal "invalid_token", event.payload.fetch(:outcome) + refute event.payload.key?(:component_name) + end + private + def capture_component_event + event = nil + subscription = ActiveSupport::Notifications.subscribe( + "solid_objects.component.refreshed" + ) { |notification| event = notification } + yield + event + ensure + ActiveSupport::Notifications.unsubscribe(subscription) if subscription + end + def component_token( reference, component_name:, diff --git a/test/integration/doctor_test.rb b/test/integration/doctor_test.rb index eb99af3..12e4f45 100644 --- a/test/integration/doctor_test.rb +++ b/test/integration/doctor_test.rb @@ -21,6 +21,57 @@ class DoctorTest < ActiveSupport::TestCase assert_empty SolidObjects::Process.where(kind: "caller") end + test "runs its probe on a caller process the application cannot adopt" do + SolidObjects.caller_process.define_singleton_method(:process_registry) do + raise "the doctor probe must not share the application caller process" + end + + report = SolidObjects::Doctor.new.call + + assert_equal :pass, report.check(:sync_round_trip).status + assert_empty SolidObjects::Process.where(kind: "caller") + ensure + SolidObjects.reset_caller_process! + end + + test "preserves a caller process the application registered before the probe" do + existing_record = SolidObjects.caller_process.process_registry.process_record + + report = SolidObjects::Doctor.new.call + + assert_equal :pass, report.check(:sync_round_trip).status + assert SolidObjects::Process.exists?(id: existing_record.id), + "doctor deleted the caller process the application already registered" + assert_equal "running", existing_record.reload.shutdown_state + assert_empty SolidObjects::Instance.where(actor_type: "solid_objects_doctor") + ensure + SolidObjects.reset_caller_process! + end + + test "reports a failed round trip instead of raising while the database stays locked" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + lock = hold_sqlite_write_lock + + report = with_immediate_sqlite_lock_failure { SolidObjects::Doctor.new.call } + + refute report.healthy? + assert_equal :fail, report.check(:sync_round_trip).status + assert_match(/database is locked/, report.check(:sync_round_trip).message) + ensure + release_sqlite_write_lock(lock) if lock + end + + test "warns when probe records outlive a passing round trip" do + doctor = SolidObjects::Doctor.new + doctor.define_singleton_method(:delete_probe_actor) { |_actor_id| false } + + report = doctor.call + + assert report.healthy? + assert_equal :warn, report.check(:sync_round_trip).status + assert_match(/probe actor/, report.check(:sync_round_trip).message) + end + test "warns when every policy denies a neutral context without changing policies" do deny = ->(**) { false } SolidObjects.configuration.authorize_message = deny @@ -93,4 +144,36 @@ class DoctorTest < ActiveSupport::TestCase ensure Rake.application = original_application end + + private + + 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 end diff --git a/test/integration/synchronous_invocation_test.rb b/test/integration/synchronous_invocation_test.rb index 408a968..4031162 100644 --- a/test/integration/synchronous_invocation_test.rb +++ b/test/integration/synchronous_invocation_test.rb @@ -502,6 +502,48 @@ def wait(timeout:) release_sqlite_write_lock(lock) if lock end + test "sync discovers the configured SQLite busy wait it has to restore" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + + SolidObjects::Record.connection_pool.with_connection do |connection| + discovered = SolidObjects + .database_adapter + .send(:configured_busy_handler_timeout, connection) + + assert_equal configured_sqlite_busy_handler_timeout, discovered, + "the adapter can no longer read the configured busy wait, so it stops " \ + "suspending lock waits and synchronous deadlines lose their bound" + end + end + + test "sync leaves an unrestorable busy wait alone" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + database_adapter = SolidObjects.database_adapter + database_adapter.define_singleton_method(:configured_busy_handler_timeout) { |_connection| nil } + + SolidObjects::Record.connection_pool.with_connection do + CounterActor.ref("unrestorable").increment + + assert_nothing_raised do + write_while_write_lock_is_briefly_held + end + end + ensure + database_adapter&.singleton_class&.send(:remove_method, :configured_busy_handler_timeout) + end + + test "sync restores the SQLite busy handler it suspended for the deadline" do + skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) + + SolidObjects::Record.connection_pool.with_connection do + CounterActor.ref("busy-handler").increment + + assert_nothing_raised do + write_while_write_lock_is_briefly_held + end + end + end + test "sync bounds SQLite contention while reusing and heartbeating its caller process" do skip unless SolidObjects::Record.connection.adapter_name.match?(/sqlite/i) @@ -758,6 +800,32 @@ def release_sqlite_write_lock(lock) thread.join end + BRIEF_LOCK_HOLD = 0.2 + + def write_while_write_lock_is_briefly_held + lock = hold_sqlite_write_lock + releaser = Thread.new do + mutex = Thread::Mutex.new + mutex.synchronize { Thread::ConditionVariable.new.wait(mutex, BRIEF_LOCK_HOLD) } + release_sqlite_write_lock(lock) + lock = nil + end + + SolidObjects::Process.create!( + id: SecureRandom.uuid, + kind: "busy-handler-probe", + hostname: "test-host", + pid: ::Process.pid, + started_at: Time.current, + last_heartbeat_at: Time.current, + metadata: {} + ) + releaser.join + ensure + releaser&.join + release_sqlite_write_lock(lock) if lock + end + def invoke_with_immediate_sqlite_lock_failure(message_reference) result = Queue.new invocation = Thread.new do @@ -768,15 +836,14 @@ def invoke_with_immediate_sqlite_lock_failure(message_reference) 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") - started_at = monotonic_now - error = capture_exception do - SolidObjects::SynchronousInvocation.new.call(message_reference, timeout: 0.1) + suspend_sqlite_busy_wait(connection) do + started_at = monotonic_now + error = capture_exception do + SolidObjects::SynchronousInvocation.new.call(message_reference, timeout: 0.1) + end + elapsed = monotonic_now - started_at end - elapsed = monotonic_now - started_at ensure - connection.execute("PRAGMA busy_timeout = #{previous_timeout}") ActiveSupport::Notifications.unsubscribe(subscription) end result << [ error, elapsed, attempts ]