From 8e483892d420f63402f70bb0ca2665cf33b4c734 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 12:23:19 -0700 Subject: [PATCH 01/15] feat: retry a dead effect or broadcast A dead effect or broadcast had no retry API. The only way back was an operator writing an UPDATE against a runtime table, which is the one thing the deny-by-default posture exists to prevent. A dead transmit effect was a lost replay. `SolidObjects.dead_letters` keeps its message meaning and answers two scopes, so the kind rides on the receiver rather than on an argument: SolidObjects.dead_letters.effects.all(authorization_context:) SolidObjects.dead_letters.effects.retry(effect_id, ...) SolidObjects.dead_letters.broadcasts.retry(broadcast_id, ...) `retry` returns a dead row to pending with a zero attempt count, no claim, and an immediate availability. It reuses the stable id, so a deduplicating handler still sees the same key, and it acts only on a dead row, so calling it twice cannot double-enqueue and cannot yank a row a worker holds. Each scope authorizes under its own resource name, and reads only its own kind. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects.rb | 1 + lib/solid_objects/dead_letter_manager.rb | 18 ++ lib/solid_objects/dead_letter_scope.rb | 79 ++++++ .../lib/solid_objects/dead_letter_manager.rbs | 6 + .../lib/solid_objects/dead_letter_scope.rbs | 44 ++++ test/integration/dead_letter_scopes_test.rb | 235 ++++++++++++++++++ 6 files changed, 383 insertions(+) create mode 100644 lib/solid_objects/dead_letter_scope.rb create mode 100644 sig/generated/lib/solid_objects/dead_letter_scope.rbs create mode 100644 test/integration/dead_letter_scopes_test.rb diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index 4cbbf03..59529fc 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -27,6 +27,7 @@ require "solid_objects/actor" require "solid_objects/reference" require "solid_objects/message_reference" +require "solid_objects/dead_letter_scope" require "solid_objects/dead_letter_manager" require "solid_objects/message_pruner" require "solid_objects/instance_pruner" diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb index 65cb457..0e109bb 100644 --- a/lib/solid_objects/dead_letter_manager.rb +++ b/lib/solid_objects/dead_letter_manager.rb @@ -29,6 +29,24 @@ def retry(dead_letter_id, authorization_context: nil) message_reference end + # @rbs () -> DeadLetterScope + def effects + @effects ||= DeadLetterScope.new( + model: Effect, + resource: "effect_dead_letters", + identifier: :effect_id + ) + end + + # @rbs () -> DeadLetterScope + def broadcasts + @broadcasts ||= DeadLetterScope.new( + model: Broadcast, + resource: "broadcast_dead_letters", + identifier: :broadcast_id + ) + end + private # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb new file mode 100644 index 0000000..9d6fe03 --- /dev/null +++ b/lib/solid_objects/dead_letter_scope.rb @@ -0,0 +1,79 @@ +# rbs_inline: enabled + +module SolidObjects + class DeadLetterScope + DEAD = "dead" + PENDING = "pending" + + # @rbs @model: untyped + # @rbs @resource: String + # @rbs @identifier: Symbol + + attr_reader :resource + + # @rbs (model: untyped, resource: String, identifier: Symbol) -> void + def initialize(model:, resource:, identifier:) + @model = model + @resource = resource + @identifier = identifier + end + + # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] + def all(authorization_context: nil) + authorize!(:inspect, authorization_context:) + dead.order(updated_at: :desc, id: :desc) + end + + # @rbs (String, ?authorization_context: untyped) -> untyped + def retry(identifier_value, authorization_context: nil) + authorize!(:retry, authorization_context:, resource_id: identifier_value) + row = model.find_by!(identifier => identifier_value) + return row unless row.status == DEAD + + revive(row) + row + end + + # @rbs () -> ActiveRecord::Relation[untyped] + def dead + model.where(status: DEAD) + end + + # @rbs (untyped) -> Integer + def revive(row) + model.where(id: row.id, status: DEAD).update_all(revival_attributes).tap do + row.reload + end + end + + # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + def authorize!(action, authorization_context:, resource_id: nil) + authorized = SolidObjects.configuration.authorize_administration.call( + action: action.to_s, + resource: resource, + resource_id: resource_id, + authorization_context: + ) + return if authorized + + raise Unauthorized, "actor administration is not authorized" + end + + private + + attr_reader :model, :identifier + + # @rbs () -> Hash[Symbol, untyped] + def revival_attributes + now = SolidObjects.database_adapter.database_now + { + status: PENDING, + attempt_count: 0, + available_at: now, + claimed_by: nil, + claimed_at: nil, + updated_at: now + } + end + end +end diff --git a/sig/generated/lib/solid_objects/dead_letter_manager.rbs b/sig/generated/lib/solid_objects/dead_letter_manager.rbs index 373e5f5..3b75450 100644 --- a/sig/generated/lib/solid_objects/dead_letter_manager.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_manager.rbs @@ -8,6 +8,12 @@ module SolidObjects # @rbs (Integer, ?authorization_context: untyped) -> MessageReference def retry: (Integer, ?authorization_context: untyped) -> MessageReference + # @rbs () -> DeadLetterScope + def effects: () -> DeadLetterScope + + # @rbs () -> DeadLetterScope + def broadcasts: () -> DeadLetterScope + private # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs new file mode 100644 index 0000000..d659942 --- /dev/null +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -0,0 +1,44 @@ +# Generated from lib/solid_objects/dead_letter_scope.rb with RBS::Inline + +module SolidObjects + class DeadLetterScope + DEAD: ::String + + PENDING: ::String + + @model: untyped + + @resource: String + + @identifier: Symbol + + attr_reader resource: untyped + + # @rbs (model: untyped, resource: String, identifier: Symbol) -> void + def initialize: (model: untyped, resource: String, identifier: Symbol) -> void + + # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] + def all: (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] + + # @rbs (String, ?authorization_context: untyped) -> untyped + def retry: (String, ?authorization_context: untyped) -> untyped + + # @rbs () -> ActiveRecord::Relation[untyped] + def dead: () -> ActiveRecord::Relation[untyped] + + # @rbs (untyped) -> Integer + def revive: (untyped) -> Integer + + # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + def authorize!: (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + + private + + attr_reader model: untyped + + attr_reader identifier: untyped + + # @rbs () -> Hash[Symbol, untyped] + def revival_attributes: () -> Hash[Symbol, untyped] + end +end diff --git a/test/integration/dead_letter_scopes_test.rb b/test/integration/dead_letter_scopes_test.rb new file mode 100644 index 0000000..2169879 --- /dev/null +++ b/test/integration/dead_letter_scopes_test.rb @@ -0,0 +1,235 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class DeadLetterScopesTest < ActiveSupport::TestCase + class OrderActor < SolidObjects::Actor + actor_type "scoped-dead-letter-orders" + + attribute :count, default: 0 + attribute :orders, default: 0 + + observable :count + + def place + self.orders += 1 + emit :charge_order, order_id: "order-1" + end + + def touch + self.count += 1 + end + + def send_elsewhere + transmit.touch + end + end + + class PoisonActor < SolidObjects::Actor + actor_type "scoped-dead-letter-poison" + + class << self + attr_accessor :fail + end + + def run + raise "poison message" if self.class.fail + end + end + + setup do + SolidObjects.configuration.retry_delay = ->(_attempt) { 0 } + SolidObjects.configuration.max_attempts = 1 + SolidObjects.configuration.authorize_administration = ->(**) { true } + PoisonActor.fail = true + @charges = [] + end + + test "returns a dead effect to pending and runs it again" do + effect = dead_effect + + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + + assert_equal "pending", effect.reload.status + assert_equal 0, effect.attempt_count + assert_nil effect.claimed_by + SolidObjects.register_effect(:charge_order) { |arguments, _context| @charges << arguments } + run_effects + + assert_equal "completed", effect.reload.status + assert_equal [ { "order_id" => "order-1" } ], @charges + end + + test "reuses the stable effect id when it retries" do + effect = dead_effect + original_id = effect.effect_id + + SolidObjects.dead_letters.effects.retry(original_id, authorization_context: "operator") + + assert_equal original_id, effect.reload.effect_id + assert_equal 1, SolidObjects::Effect.count + end + + test "leaves an effect that is already pending alone" do + effect = dead_effect + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + available_at = effect.reload.available_at + + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + + assert_equal "pending", effect.reload.status + assert_equal available_at, effect.available_at + assert_equal 1, SolidObjects::Effect.count + end + + test "returns a dead broadcast to pending and delivers it" do + broadcast = dead_broadcast + + SolidObjects.dead_letters.broadcasts.retry( + broadcast.broadcast_id, + authorization_context: "operator" + ) + + assert_equal "pending", broadcast.reload.status + assert_equal 0, broadcast.attempt_count + delivered = [] + SolidObjects.configuration.broadcast_adapter = ->(payload) { delivered << payload } + run_broadcasts + + assert_equal "delivered", broadcast.reload.status + assert_equal 1, delivered.size + end + + test "replays a dead transmit effect" do + OrderActor.ref("one").async.send_elsewhere + run_actors + transmitted = [] + SolidObjects.register_effect(SolidObjects::Transmission::EFFECT_NAME) { raise "carrier down" } + run_effects + effect = SolidObjects::Effect.find_by!(name: SolidObjects::Transmission::EFFECT_NAME) + assert_equal "dead", effect.status + + SolidObjects.register_effect(SolidObjects::Transmission::EFFECT_NAME) do |arguments, _context| + transmitted << arguments + end + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + run_effects + + assert_equal "completed", effect.reload.status + assert_equal [ "touch" ], transmitted.map { |arguments| arguments.fetch("operation") } + end + + test "reads and retries message dead letters as it always has" do + PoisonActor.ref("one").async.run + run_actors + dead_letter = SolidObjects::DeadLetter.sole + PoisonActor.fail = false + + listed = SolidObjects.dead_letters.all(authorization_context: "operator") + reference = SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator") + run_actors + + assert_equal [ dead_letter.id ], listed.map(&:id) + assert_equal reference.id, dead_letter.reload.retried_message_id + assert_equal "completed", reference.status + end + + test "each scope reads only its own kind" do + effect = dead_effect + broadcast = dead_broadcast + PoisonActor.ref("one").async.run + run_actors + + effects = SolidObjects.dead_letters.effects.all(authorization_context: "operator") + broadcasts = SolidObjects.dead_letters.broadcasts.all(authorization_context: "operator") + messages = SolidObjects.dead_letters.all(authorization_context: "operator") + + assert_equal [ effect.effect_id ], effects.map(&:effect_id) + assert_equal [ broadcast.broadcast_id ], broadcasts.map(&:broadcast_id) + assert_equal 1, messages.count + end + + test "reads only dead rows, not pending ones" do + dead = dead_effect + OrderActor.ref("two").async.place + run_actors + + effects = SolidObjects.dead_letters.effects.all(authorization_context: "operator") + + assert_equal 2, SolidObjects::Effect.count + assert_equal [ dead.effect_id ], effects.map(&:effect_id) + end + + test "refuses an unauthorized caller" do + effect = dead_effect + broadcast = dead_broadcast + SolidObjects.configuration.authorize_administration = ->(**) { false } + + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.dead_letters.effects.all(authorization_context: "operator") + end + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + end + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.dead_letters.broadcasts.retry( + broadcast.broadcast_id, + authorization_context: "operator" + ) + end + end + + test "names the scope it authorizes" do + effect = dead_effect + seen = [] + SolidObjects.configuration.authorize_administration = lambda do |action:, resource:, **| + seen << [ action, resource ] + true + end + + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + + assert_equal [ [ "retry", "effect_dead_letters" ] ], seen + end + + private + + def dead_effect + OrderActor.ref("one").async.place + run_actors + SolidObjects.register_effect(:charge_order) { raise "charge declined" } + run_effects + SolidObjects::Effect.find_by!(name: "charge_order").tap do |effect| + assert_equal "dead", effect.status + end + end + + def dead_broadcast + SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" } + OrderActor.ref("broadcast").async.touch + run_actors + run_broadcasts + SolidObjects::Broadcast.where(status: "dead").sole + end + + def run_actors + worker = SolidObjects::Worker.new + worker.run_until_idle + ensure + worker&.stop + end + + def run_effects + executor = SolidObjects::EffectExecutor.new + executor.run_once while SolidObjects::Effect.exists?(status: "pending") + ensure + executor&.stop + end + + def run_broadcasts + executor = SolidObjects::BroadcastExecutor.new + executor.run_once while SolidObjects::Broadcast.exists?(status: "pending") + ensure + executor&.stop + end +end From c36cfd03490a3b058e0f2d1a54ee206bac87c4b9 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 12:36:03 -0700 Subject: [PATCH 02/15] feat: record who pressed what The dashboard kept no record of an administration action, and retry is the action that can re-run a side effect. Every retry now writes one row to `solid_objects_administration_events`, holding the action, the kind, the subject, the identity, and when it happened. The identity comes from the existing authorization context through a new `administration_identity` hook, so the library records what the application already knows rather than invent an authentication concept. It defaults to the context's own `to_s` and is bounded to 255 bytes. A refused caller writes nothing, because the audit records what happened rather than what was attempted and denied. A read writes nothing. Co-Authored-By: Claude Opus 5 (1M context) --- .../solid_objects/administration_event.rb | 7 + ...add_solid_objects_administration_events.rb | 31 ++++ lib/solid_objects.rb | 1 + lib/solid_objects/administration_audit.rb | 30 ++++ lib/solid_objects/configuration.rb | 6 + lib/solid_objects/dead_letter_manager.rb | 12 +- lib/solid_objects/dead_letter_scope.rb | 13 +- lib/solid_objects/doctor.rb | 3 +- lib/solid_objects/test_helper.rb | 1 + .../solid_objects/administration_audit.rbs | 11 ++ .../lib/solid_objects/configuration.rbs | 10 +- .../lib/solid_objects/dead_letter_scope.rbs | 6 +- .../solid_objects/administration_event.rbs | 6 + test/database_test_helper.rb | 4 + test/integration/administration_audit_test.rb | 160 ++++++++++++++++++ 15 files changed, 290 insertions(+), 11 deletions(-) create mode 100644 app/models/solid_objects/administration_event.rb create mode 100644 db/migrate/20260922000000_add_solid_objects_administration_events.rb create mode 100644 lib/solid_objects/administration_audit.rb create mode 100644 sig/generated/lib/solid_objects/administration_audit.rbs create mode 100644 sig/generated/models/solid_objects/administration_event.rbs create mode 100644 test/integration/administration_audit_test.rb diff --git a/app/models/solid_objects/administration_event.rb b/app/models/solid_objects/administration_event.rb new file mode 100644 index 0000000..fbfdd20 --- /dev/null +++ b/app/models/solid_objects/administration_event.rb @@ -0,0 +1,7 @@ +# rbs_inline: enabled + +module SolidObjects + class AdministrationEvent < Record + self.table_name = SolidObjects.table_name(:administration_events) + end +end diff --git a/db/migrate/20260922000000_add_solid_objects_administration_events.rb b/db/migrate/20260922000000_add_solid_objects_administration_events.rb new file mode 100644 index 0000000..94aa632 --- /dev/null +++ b/db/migrate/20260922000000_add_solid_objects_administration_events.rb @@ -0,0 +1,31 @@ +# rbs_inline: enabled + +class AddSolidObjectsAdministrationEvents < ActiveRecord::Migration[7.1] + # @rbs () -> void + def change + create_table SolidObjects.table_name(:administration_events) do |definition| + definition.string :action, null: false, limit: 64 + definition.string :kind, null: false, limit: 32 + definition.string :subject_id, limit: 191 + json_column definition, :filters + definition.string :actor, limit: 255 + definition.datetime :occurred_at, null: false, precision: 6 + definition.timestamps precision: 6, null: false + + definition.index [ :occurred_at, :id ], name: "idx_so_admin_events_occurred" + definition.index [ :kind, :subject_id ], name: "idx_so_admin_events_subject" + end + end + + private + + # @rbs () -> Symbol + def json_column_type + connection.adapter_name.match?(/postgres/i) ? :jsonb : :json + end + + # @rbs (untyped, Symbol, ?null: bool) -> void + def json_column(definition, name, null: true) + definition.public_send(json_column_type, name, null:) + end +end diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index 59529fc..9c046fb 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -27,6 +27,7 @@ require "solid_objects/actor" require "solid_objects/reference" require "solid_objects/message_reference" +require "solid_objects/administration_audit" require "solid_objects/dead_letter_scope" require "solid_objects/dead_letter_manager" require "solid_objects/message_pruner" diff --git a/lib/solid_objects/administration_audit.rb b/lib/solid_objects/administration_audit.rb new file mode 100644 index 0000000..96d4a15 --- /dev/null +++ b/lib/solid_objects/administration_audit.rb @@ -0,0 +1,30 @@ +# rbs_inline: enabled + +module SolidObjects + module AdministrationAudit + module_function + + # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, authorization_context: untyped) -> void + def record(action:, kind:, authorization_context:, subject_id: nil, filters: nil) + AdministrationEvent.create!( + action:, + kind:, + subject_id: subject_id&.to_s, + filters:, + actor: identity(authorization_context), + occurred_at: SolidObjects.database_adapter.database_now + ) + nil + end + + # @rbs (untyped) -> String? + def identity(authorization_context) + return nil if authorization_context.nil? + + SolidObjects.configuration.administration_identity + .call(authorization_context) + &.to_s + &.byteslice(0, 255) + end + end +end diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index 87b2523..a29856e 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -49,6 +49,7 @@ class Configuration # @rbs @authorize_subscription: Proc # @rbs @authorize_administration: Proc # @rbs @authorize_transmission: Proc + # @rbs @administration_identity: Proc # @rbs @transmission_actor_type_resolver: Proc attr_accessor :table_name_prefix, @@ -98,6 +99,7 @@ class Configuration :authorize_subscription, :authorize_administration, :authorize_transmission, + :administration_identity, :transmission_actor_type_resolver # @rbs @additional_components: Array[untyped] @@ -156,6 +158,7 @@ def initialize @authorize_subscription = ->(**) { false } @authorize_administration = ->(**) { false } @authorize_transmission = ->(**) { false } + @administration_identity = ->(authorization_context) { authorization_context.to_s } @transmission_actor_type_resolver = ->(actor_type) { actor_type } @additional_components = [] end @@ -247,6 +250,9 @@ def validate! unless payload_authorization_context.respond_to?(:call) raise ArgumentError, "payload_authorization_context must respond to call" end + unless administration_identity.respond_to?(:call) + raise ArgumentError, "administration_identity must respond to call" + end self end diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb index 0e109bb..b808a38 100644 --- a/lib/solid_objects/dead_letter_manager.rb +++ b/lib/solid_objects/dead_letter_manager.rb @@ -12,6 +12,12 @@ def all(authorization_context: nil) def retry(dead_letter_id, authorization_context: nil) authorize!(:retry, authorization_context:, dead_letter_id:) dead_letter = DeadLetter.find(dead_letter_id) + AdministrationAudit.record( + action: "dead_letter.retry", + kind: "message", + subject_id: dead_letter.id, + authorization_context: + ) return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) if dead_letter.retried_message_id original_message = dead_letter.message @@ -34,7 +40,8 @@ def effects @effects ||= DeadLetterScope.new( model: Effect, resource: "effect_dead_letters", - identifier: :effect_id + identifier: :effect_id, + kind: "effect" ) end @@ -43,7 +50,8 @@ def broadcasts @broadcasts ||= DeadLetterScope.new( model: Broadcast, resource: "broadcast_dead_letters", - identifier: :broadcast_id + identifier: :broadcast_id, + kind: "broadcast" ) end diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 9d6fe03..9730c76 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -11,11 +11,12 @@ class DeadLetterScope attr_reader :resource - # @rbs (model: untyped, resource: String, identifier: Symbol) -> void - def initialize(model:, resource:, identifier:) + # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void + def initialize(model:, resource:, identifier:, kind:) @model = model @resource = resource @identifier = identifier + @kind = kind end # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] @@ -28,6 +29,12 @@ def all(authorization_context: nil) def retry(identifier_value, authorization_context: nil) authorize!(:retry, authorization_context:, resource_id: identifier_value) row = model.find_by!(identifier => identifier_value) + AdministrationAudit.record( + action: "dead_letter.retry", + kind: kind, + subject_id: identifier_value, + authorization_context: + ) return row unless row.status == DEAD revive(row) @@ -61,7 +68,7 @@ def authorize!(action, authorization_context:, resource_id: nil) private - attr_reader :model, :identifier + attr_reader :model, :identifier, :kind # @rbs () -> Hash[Symbol, untyped] def revival_attributes diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb index 4ed4a92..5700201 100644 --- a/lib/solid_objects/doctor.rb +++ b/lib/solid_objects/doctor.rb @@ -80,7 +80,8 @@ def to_s effects: %w[id message_id instance_id effect_id status available_at], effect_recoveries: %w[effect_id instance_id recovery_operation status_operation recovery_timeout retired_at], broadcasts: %w[id message_id instance_id broadcast_id status available_at], - dead_letters: %w[id message_id instance_id actor_type actor_id attempts] + dead_letters: %w[id message_id instance_id actor_type actor_id attempts], + administration_events: %w[id action kind subject_id actor occurred_at] }.freeze class ProbeActor < Actor diff --git a/lib/solid_objects/test_helper.rb b/lib/solid_objects/test_helper.rb index 804469c..e4bd1c0 100644 --- a/lib/solid_objects/test_helper.rb +++ b/lib/solid_objects/test_helper.rb @@ -31,6 +31,7 @@ def reset_actors! # @rbs () -> Array[Class] def actor_owned_models [ + AdministrationEvent, DeadLetter, ClaimedMessage, ReadyMessage, diff --git a/sig/generated/lib/solid_objects/administration_audit.rbs b/sig/generated/lib/solid_objects/administration_audit.rbs new file mode 100644 index 0000000..14a839c --- /dev/null +++ b/sig/generated/lib/solid_objects/administration_audit.rbs @@ -0,0 +1,11 @@ +# Generated from lib/solid_objects/administration_audit.rb with RBS::Inline + +module SolidObjects + module AdministrationAudit + # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, authorization_context: untyped) -> void + def self?.record: (action: String, kind: String, authorization_context: untyped, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?) -> void + + # @rbs (untyped) -> String? + def self?.identity: (untyped) -> String? + end +end diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index 0cecf1f..ddc4935 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -2,7 +2,7 @@ module SolidObjects class Configuration - @process_heartbeat_interval: Float + @table_name_prefix: String @process_alive_threshold: Float @@ -60,9 +60,9 @@ module SolidObjects @authorize_transmission: Proc - @transmission_actor_type_resolver: Proc + @administration_identity: Proc - @table_name_prefix: String + @transmission_actor_type_resolver: Proc @polling_interval: Float @@ -98,6 +98,8 @@ module SolidObjects @lock_retry_attempts: Integer + @process_heartbeat_interval: Float + attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -192,6 +194,8 @@ module SolidObjects attr_accessor authorize_transmission: untyped + attr_accessor administration_identity: untyped + attr_accessor transmission_actor_type_resolver: untyped # @rbs @additional_components: Array[untyped] diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index d659942..b058b30 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -14,8 +14,8 @@ module SolidObjects attr_reader resource: untyped - # @rbs (model: untyped, resource: String, identifier: Symbol) -> void - def initialize: (model: untyped, resource: String, identifier: Symbol) -> void + # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void + def initialize: (model: untyped, resource: String, identifier: Symbol, kind: String) -> void # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] def all: (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] @@ -38,6 +38,8 @@ module SolidObjects attr_reader identifier: untyped + attr_reader kind: untyped + # @rbs () -> Hash[Symbol, untyped] def revival_attributes: () -> Hash[Symbol, untyped] end diff --git a/sig/generated/models/solid_objects/administration_event.rbs b/sig/generated/models/solid_objects/administration_event.rbs new file mode 100644 index 0000000..7787bff --- /dev/null +++ b/sig/generated/models/solid_objects/administration_event.rbs @@ -0,0 +1,6 @@ +# Generated from app/models/solid_objects/administration_event.rb with RBS::Inline + +module SolidObjects + class AdministrationEvent < Record + end +end diff --git a/test/database_test_helper.rb b/test/database_test_helper.rb index 56b42b5..c49163a 100644 --- a/test/database_test_helper.rb +++ b/test/database_test_helper.rb @@ -25,11 +25,13 @@ require_relative "../db/migrate/20260806000000_add_state_revision_to_solid_objects_instances" require_relative "../db/migrate/20260813000000_rename_message_dispatch_columns" require_relative "../db/migrate/20260915000000_add_solid_objects_effect_recoveries" +require_relative "../db/migrate/20260922000000_add_solid_objects_administration_events" CreateSolidObjectsTables.new.migrate(:up) AddStateRevisionToSolidObjectsInstances.new.migrate(:up) RenameMessageDispatchColumns.new.migrate(:up) AddSolidObjectsEffectRecoveries.new.migrate(:up) +AddSolidObjectsAdministrationEvents.new.migrate(:up) ActiveRecord::Base.connection.create_table(:solid_objects_test_domain_records) do |table| table.string :name, null: false @@ -37,6 +39,7 @@ require "solid_objects/database_adapter" require_relative "../app/models/solid_objects/record" +require_relative "../app/models/solid_objects/administration_event" require_relative "../app/models/solid_objects/process" require_relative "../app/models/solid_objects/instance" require_relative "../app/models/solid_objects/message" @@ -59,6 +62,7 @@ class ActiveSupport::TestCase teardown do SolidObjects::Instance.delete_all SolidObjects::Process.delete_all + SolidObjects::AdministrationEvent.delete_all SolidObjectsTestDomainRecord.delete_all end diff --git a/test/integration/administration_audit_test.rb b/test/integration/administration_audit_test.rb new file mode 100644 index 0000000..612a3d4 --- /dev/null +++ b/test/integration/administration_audit_test.rb @@ -0,0 +1,160 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class AdministrationAuditTest < ActiveSupport::TestCase + class LedgerActor < SolidObjects::Actor + actor_type "audited-ledger" + + attribute :count, default: 0 + attribute :entries, default: 0 + + observable :count + + def post + self.entries += 1 + emit :settle, entry: "one" + end + + def touch + self.count += 1 + end + end + + class PoisonActor < SolidObjects::Actor + actor_type "audited-poison" + + class << self + attr_accessor :fail + end + + def run + raise "poison message" if self.class.fail + end + end + + setup do + SolidObjects.configuration.retry_delay = ->(_attempt) { 0 } + SolidObjects.configuration.max_attempts = 1 + SolidObjects.configuration.authorize_administration = ->(**) { true } + PoisonActor.fail = true + end + + test "writes one audit row for an effect retry" do + effect = dead_effect + + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + + event = SolidObjects::AdministrationEvent.sole + assert_equal "dead_letter.retry", event.action + assert_equal "effect", event.kind + assert_equal effect.effect_id, event.subject_id + assert_equal "operator", event.actor + assert event.occurred_at + end + + test "writes one audit row for a broadcast retry" do + broadcast = dead_broadcast + + SolidObjects.dead_letters.broadcasts.retry( + broadcast.broadcast_id, + authorization_context: "operator" + ) + + event = SolidObjects::AdministrationEvent.sole + assert_equal "dead_letter.retry", event.action + assert_equal "broadcast", event.kind + assert_equal broadcast.broadcast_id, event.subject_id + end + + test "writes one audit row for a message retry" do + PoisonActor.ref("one").async.run + run_actors + dead_letter = SolidObjects::DeadLetter.sole + PoisonActor.fail = false + + SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator") + + event = SolidObjects::AdministrationEvent.sole + assert_equal "dead_letter.retry", event.action + assert_equal "message", event.kind + assert_equal dead_letter.id.to_s, event.subject_id + end + + test "writes one audit row for each press, including a repeat" do + effect = dead_effect + + 2.times do + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + end + + assert_equal 2, SolidObjects::AdministrationEvent.count + end + + test "records the identity the application names" do + effect = dead_effect + SolidObjects.configuration.administration_identity = ->(context) { "user:#{context.fetch(:id)}" } + + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: { id: 42 }) + + assert_equal "user:42", SolidObjects::AdministrationEvent.sole.actor + end + + test "writes no audit row when the caller is refused" do + effect = dead_effect + SolidObjects.configuration.authorize_administration = ->(**) { false } + + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + end + + assert_equal 0, SolidObjects::AdministrationEvent.count + end + + test "reading dead letters writes no audit row" do + dead_effect + + SolidObjects.dead_letters.effects.all(authorization_context: "operator").to_a + + assert_equal 0, SolidObjects::AdministrationEvent.count + end + + private + + def dead_effect + LedgerActor.ref("one").async.post + run_actors + SolidObjects.register_effect(:settle) { raise "settlement declined" } + run_effects + SolidObjects::Effect.find_by!(name: "settle") + end + + def dead_broadcast + SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" } + LedgerActor.ref("broadcast").async.touch + run_actors + run_broadcasts + SolidObjects::Broadcast.where(status: "dead").sole + end + + def run_actors + worker = SolidObjects::Worker.new + worker.run_until_idle + ensure + worker&.stop + end + + def run_effects + executor = SolidObjects::EffectExecutor.new + executor.run_once while SolidObjects::Effect.exists?(status: "pending") + ensure + executor&.stop + end + + def run_broadcasts + executor = SolidObjects::BroadcastExecutor.new + executor.run_once while SolidObjects::Broadcast.exists?(status: "pending") + ensure + executor&.stop + end +end From 1e9d9cc5fc4c2395e6768eca469b6bbf1eee6de0 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 12:54:13 -0700 Subject: [PATCH 03/15] feat: redrive dead effects and broadcasts in bulk Retry was one row at a time, and an incident produces dead rows in the hundreds. A scope answers `redrive` now, which opens a durable task and returns at once: task = SolidObjects.dead_letters.effects.redrive( actor_type: "payments", failed_after: 6.hours.ago, limit: 5_000, authorization_context: current_admin ) task.cancel(authorization_context: current_admin) The task is idempotent over its scope and filters, which a dashboard button needs. A unique index on the active scope enforces that in the database rather than in a read followed by a write, so two processes that start the same redrive at the same time share one task. The index is total rather than partial: the column holds the scope while the task runs and NULL once it finishes, so a later redrive of the same scope starts a new task. The supervisor advances one bounded batch per pass and pauses between batches, so a redrive of thousands of rows never holds a transaction longer than one batch and shares the database with delivery. `SolidObjects.redrives` reads tasks back by id and by status. A running task reports what is left to move rather than a stored estimate, because rows die and are retried while it runs. Every transition writes one audit row under the identity that asked for it. Co-Authored-By: Claude Opus 5 (1M context) --- app/models/solid_objects/redrive.rb | 7 + ...260922000001_add_solid_objects_redrives.rb | 37 +++ lib/solid_objects.rb | 9 + lib/solid_objects/administration_audit.rb | 6 +- lib/solid_objects/configuration.rb | 12 +- lib/solid_objects/dead_letter_manager.rb | 2 +- lib/solid_objects/dead_letter_scope.rb | 43 ++- lib/solid_objects/doctor.rb | 3 +- lib/solid_objects/redrive_manager.rb | 131 ++++++++ lib/solid_objects/redrive_runner.rb | 69 +++++ lib/solid_objects/redrive_task.rb | 12 + lib/solid_objects/supervisor.rb | 47 +++ lib/solid_objects/test_helper.rb | 1 + sig/generated/lib/solid_objects.rbs | 3 + .../solid_objects/administration_audit.rbs | 4 +- .../lib/solid_objects/configuration.rbs | 12 +- .../lib/solid_objects/dead_letter_scope.rbs | 20 +- .../lib/solid_objects/redrive_manager.rbs | 48 +++ .../lib/solid_objects/redrive_runner.rbs | 31 ++ .../lib/solid_objects/redrive_task.rbs | 28 ++ .../lib/solid_objects/supervisor.rbs | 19 ++ .../models/solid_objects/redrive.rbs | 6 + test/database_test_helper.rb | 4 + test/integration/redrive_test.rb | 280 ++++++++++++++++++ 24 files changed, 817 insertions(+), 17 deletions(-) create mode 100644 app/models/solid_objects/redrive.rb create mode 100644 db/migrate/20260922000001_add_solid_objects_redrives.rb create mode 100644 lib/solid_objects/redrive_manager.rb create mode 100644 lib/solid_objects/redrive_runner.rb create mode 100644 lib/solid_objects/redrive_task.rb create mode 100644 sig/generated/lib/solid_objects/redrive_manager.rbs create mode 100644 sig/generated/lib/solid_objects/redrive_runner.rbs create mode 100644 sig/generated/lib/solid_objects/redrive_task.rbs create mode 100644 sig/generated/models/solid_objects/redrive.rbs create mode 100644 test/integration/redrive_test.rb diff --git a/app/models/solid_objects/redrive.rb b/app/models/solid_objects/redrive.rb new file mode 100644 index 0000000..3687698 --- /dev/null +++ b/app/models/solid_objects/redrive.rb @@ -0,0 +1,7 @@ +# rbs_inline: enabled + +module SolidObjects + class Redrive < Record + self.table_name = SolidObjects.table_name(:redrives) + end +end diff --git a/db/migrate/20260922000001_add_solid_objects_redrives.rb b/db/migrate/20260922000001_add_solid_objects_redrives.rb new file mode 100644 index 0000000..4e36d56 --- /dev/null +++ b/db/migrate/20260922000001_add_solid_objects_redrives.rb @@ -0,0 +1,37 @@ +# rbs_inline: enabled + +class AddSolidObjectsRedrives < ActiveRecord::Migration[7.1] + # @rbs () -> void + def change + create_table SolidObjects.table_name(:redrives), id: :string, limit: 64 do |definition| + definition.string :kind, null: false, limit: 32 + json_column definition, :filters, null: false + definition.string :status, null: false, default: "running", limit: 32 + definition.string :active_scope, limit: 191 + definition.integer :moved, null: false, default: 0 + definition.integer :move_limit + definition.string :actor, limit: 255 + definition.datetime :started_at, null: false, precision: 6 + definition.datetime :finished_at, precision: 6 + definition.timestamps precision: 6, null: false + + definition.index :active_scope, unique: true, name: "idx_so_redrives_active_scope" + definition.index [ :status, :started_at, :id ], name: "idx_so_redrives_poll" + definition.check_constraint "moved >= 0", name: "chk_so_redrives_moved" + definition.check_constraint "move_limit IS NULL OR move_limit > 0", name: "chk_so_redrives_limit" + definition.check_constraint "status IN ('running', 'completed', 'cancelled')", name: "chk_so_redrives_status" + end + end + + private + + # @rbs () -> Symbol + def json_column_type + connection.adapter_name.match?(/postgres/i) ? :jsonb : :json + end + + # @rbs (untyped, Symbol, ?null: bool) -> void + def json_column(definition, name, null: true) + definition.public_send(json_column_type, name, null:) + end +end diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb index 9c046fb..faf9e9c 100644 --- a/lib/solid_objects.rb +++ b/lib/solid_objects.rb @@ -28,6 +28,9 @@ require "solid_objects/reference" require "solid_objects/message_reference" require "solid_objects/administration_audit" +require "solid_objects/redrive_task" +require "solid_objects/redrive_manager" +require "solid_objects/redrive_runner" require "solid_objects/dead_letter_scope" require "solid_objects/dead_letter_manager" require "solid_objects/message_pruner" @@ -162,6 +165,11 @@ def dead_letters @dead_letters ||= DeadLetterManager.new end + # @rbs () -> RedriveManager + def redrives + @redrives ||= RedriveManager.new + end + # @rbs () -> Administration def administration @administration ||= Administration.new @@ -194,6 +202,7 @@ def reset! @effect_registry = EffectRegistry.new @commit_action_registry = CommitActionRegistry.new @dead_letters = nil + @redrives = nil @administration = nil end diff --git a/lib/solid_objects/administration_audit.rb b/lib/solid_objects/administration_audit.rb index 96d4a15..840902b 100644 --- a/lib/solid_objects/administration_audit.rb +++ b/lib/solid_objects/administration_audit.rb @@ -4,14 +4,14 @@ module SolidObjects module AdministrationAudit module_function - # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, authorization_context: untyped) -> void - def record(action:, kind:, authorization_context:, subject_id: nil, filters: nil) + # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void + def record(action:, kind:, subject_id: nil, filters: nil, actor: nil) AdministrationEvent.create!( action:, kind:, subject_id: subject_id&.to_s, filters:, - actor: identity(authorization_context), + actor:, occurred_at: SolidObjects.database_adapter.database_now ) nil diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb index a29856e..7a33ec9 100644 --- a/lib/solid_objects/configuration.rb +++ b/lib/solid_objects/configuration.rb @@ -31,6 +31,8 @@ class Configuration # @rbs @instance_retention_by_actor_type: Hash[String, Numeric] # @rbs @process_retention: Numeric # @rbs @prune_batch_size: Integer + # @rbs @redrive_batch_size: Integer + # @rbs @redrive_batch_pause: Float # @rbs @worker_count: Integer # @rbs @effect_worker_count: Integer # @rbs @broadcast_worker_count: Integer @@ -81,6 +83,8 @@ class Configuration :instance_retention_by_actor_type, :process_retention, :prune_batch_size, + :redrive_batch_size, + :redrive_batch_pause, :worker_count, :effect_worker_count, :broadcast_worker_count, @@ -136,6 +140,8 @@ def initialize @instance_retention_by_actor_type = {} @process_retention = 7.days @prune_batch_size = 1_000 + @redrive_batch_size = 100 + @redrive_batch_pause = 0.05 @worker_count = 1 @effect_worker_count = 1 @broadcast_worker_count = 1 @@ -229,6 +235,9 @@ def validate! positive_values.each do |name, value| raise ArgumentError, "#{name} must be positive" unless value.positive? end + if redrive_batch_pause.negative? + raise ArgumentError, "redrive_batch_pause must not be negative" + end if warn_state_bytes > max_state_bytes raise ArgumentError, "warn_state_bytes must not exceed max_state_bytes" end @@ -301,7 +310,8 @@ def positive_values shutdown_timeout:, message_retention:, process_retention:, - prune_batch_size: + prune_batch_size:, + redrive_batch_size: } end diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb index b808a38..8455309 100644 --- a/lib/solid_objects/dead_letter_manager.rb +++ b/lib/solid_objects/dead_letter_manager.rb @@ -16,7 +16,7 @@ def retry(dead_letter_id, authorization_context: nil) action: "dead_letter.retry", kind: "message", subject_id: dead_letter.id, - authorization_context: + actor: AdministrationAudit.identity(authorization_context) ) return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) if dead_letter.retried_message_id diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 9730c76..691c5af 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -9,7 +9,7 @@ class DeadLetterScope # @rbs @resource: String # @rbs @identifier: Symbol - attr_reader :resource + attr_reader :resource, :kind # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void def initialize(model:, resource:, identifier:, kind:) @@ -19,6 +19,14 @@ def initialize(model:, resource:, identifier:, kind:) @kind = kind end + # @rbs (String) -> DeadLetterScope + def self.for_kind(kind) + return SolidObjects.dead_letters.effects if kind == "effect" + return SolidObjects.dead_letters.broadcasts if kind == "broadcast" + + raise ArgumentError, "unknown dead letter kind #{kind.inspect}" + end + # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] def all(authorization_context: nil) authorize!(:inspect, authorization_context:) @@ -33,7 +41,7 @@ def retry(identifier_value, authorization_context: nil) action: "dead_letter.retry", kind: kind, subject_id: identifier_value, - authorization_context: + actor: AdministrationAudit.identity(authorization_context) ) return row unless row.status == DEAD @@ -41,11 +49,40 @@ def retry(identifier_value, authorization_context: nil) row end + # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask + def redrive(actor_type: nil, failed_after: nil, limit: nil, authorization_context: nil) + authorize!(:redrive, authorization_context:) + SolidObjects.redrives.start( + scope: self, + filters: { + "actor_type" => actor_type, + "failed_after" => failed_after&.utc&.iso8601(6), + "limit" => limit + }, + authorization_context: + ) + end + # @rbs () -> ActiveRecord::Relation[untyped] def dead model.where(status: DEAD) end + # @rbs (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] + def matching(filters) + relation = dead + actor_type = filters["actor_type"] + failed_after = filters["failed_after"] + relation = relation.joins(:instance).where(Instance.table_name => { actor_type: }) if actor_type + relation = relation.where(updated_at: Time.parse(failed_after)..) if failed_after + relation + end + + # @rbs (Array[untyped]) -> Integer + def revive_all(identifiers) + model.where(id: identifiers, status: DEAD).update_all(revival_attributes) + end + # @rbs (untyped) -> Integer def revive(row) model.where(id: row.id, status: DEAD).update_all(revival_attributes).tap do @@ -68,7 +105,7 @@ def authorize!(action, authorization_context:, resource_id: nil) private - attr_reader :model, :identifier, :kind + attr_reader :model, :identifier # @rbs () -> Hash[Symbol, untyped] def revival_attributes diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb index 5700201..0a26561 100644 --- a/lib/solid_objects/doctor.rb +++ b/lib/solid_objects/doctor.rb @@ -81,7 +81,8 @@ def to_s effect_recoveries: %w[effect_id instance_id recovery_operation status_operation recovery_timeout retired_at], broadcasts: %w[id message_id instance_id broadcast_id status available_at], dead_letters: %w[id message_id instance_id actor_type actor_id attempts], - administration_events: %w[id action kind subject_id actor occurred_at] + administration_events: %w[id action kind subject_id actor occurred_at], + redrives: %w[id kind filters status active_scope moved move_limit started_at finished_at] }.freeze class ProbeActor < Actor diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb new file mode 100644 index 0000000..aaec6ff --- /dev/null +++ b/lib/solid_objects/redrive_manager.rb @@ -0,0 +1,131 @@ +# rbs_inline: enabled + +require "digest" + +module SolidObjects + class RedriveManager + RUNNING = "running" + COMPLETED = "completed" + CANCELLED = "cancelled" + + # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask + def start(scope:, filters:, authorization_context:) + active_scope = active_scope_for(kind: scope.kind, filters:) + running = Redrive.find_by(active_scope:) + return task_for(running) if running + + task_for(open_task(scope:, filters:, active_scope:, authorization_context:)) + rescue ActiveRecord::RecordNotUnique + task_for(Redrive.find_by!(active_scope:)) + end + + # @rbs (String, ?authorization_context: untyped) -> RedriveTask + def find(id, authorization_context: nil) + authorize!(:inspect, authorization_context:, resource_id: id) + task_for(Redrive.find(id)) + end + + # @rbs (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask] + def all(status: nil, authorization_context: nil) + authorize!(:inspect, authorization_context:) + relation = Redrive.order(started_at: :desc, id: :desc) + relation = relation.where(status: status.to_s) if status + relation.map { |record| task_for(record) } + end + + # @rbs (String, ?authorization_context: untyped) -> RedriveTask + def cancel(id, authorization_context: nil) + authorize!(:cancel, authorization_context:, resource_id: id) + record = Redrive.find(id) + return task_for(record) unless record.status == RUNNING + + close(record, status: CANCELLED) + audit(record, action: "redrive.cancel") + task_for(record) + end + + # @rbs (Redrive, status: String) -> void + def close(record, status:) + record.update!( + status:, + active_scope: nil, + finished_at: SolidObjects.database_adapter.database_now + ) + end + + # @rbs (Redrive, action: String) -> void + def audit(record, action:) + AdministrationAudit.record( + action:, + kind: record.kind, + subject_id: record.id, + filters: record.filters, + actor: record.actor + ) + end + + # @rbs (Redrive) -> RedriveTask + def task_for(record) + RedriveTask.new( + id: record.id, + kind: record.kind, + filters: Serialization.readonly_copy(record.filters), + status: record.status, + moved: record.moved, + remaining: remaining_for(record), + started_at: record.started_at, + finished_at: record.finished_at + ) + end + + private + + # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive + def open_task(scope:, filters:, active_scope:, authorization_context:) + record = Redrive.create!( + id: "redrive_#{SecureRandom.uuid}", + kind: scope.kind, + filters:, + status: RUNNING, + active_scope:, + moved: 0, + move_limit: filters["limit"], + actor: AdministrationAudit.identity(authorization_context), + started_at: SolidObjects.database_adapter.database_now + ) + audit(record, action: "redrive.start") + record + end + + # A running task reports what is left to move rather than a stored + # estimate, because rows die and are retried while it runs. + # @rbs (Redrive) -> Integer + def remaining_for(record) + return 0 unless record.status == RUNNING + + matching = DeadLetterScope.for_kind(record.kind).matching(record.filters).count + limit = record.move_limit + return matching unless limit + + [ matching, limit - record.moved ].min + end + + # @rbs (kind: String, filters: Hash[String, untyped]) -> String + def active_scope_for(kind:, filters:) + "#{kind}:#{Digest::SHA256.hexdigest(filters.to_json)}" + end + + # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + def authorize!(action, authorization_context:, resource_id: nil) + authorized = SolidObjects.configuration.authorize_administration.call( + action: action.to_s, + resource: "redrives", + resource_id:, + authorization_context: + ) + return if authorized + + raise Unauthorized, "actor administration is not authorized" + end + end +end diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb new file mode 100644 index 0000000..663fb64 --- /dev/null +++ b/lib/solid_objects/redrive_runner.rb @@ -0,0 +1,69 @@ +# rbs_inline: enabled + +module SolidObjects + # Moves dead rows back to pending for one redrive task at a time, in bounded + # batches. Each batch is its own short transaction, so a redrive of thousands + # of rows never holds a lock long enough to starve delivery. + class RedriveRunner + # @rbs () -> bool + def run_once + database_adapter.transaction do + record = claim + next false unless record + + advance(record) + end + end + + private + + # @rbs () -> Redrive? + def claim + Redrive.lock.where(status: RedriveManager::RUNNING).order(:started_at, :id).first + end + + # @rbs (Redrive) -> bool + def advance(record) + moved = move_batch(record) + return true if moved.positive? + + finish(record) + false + end + + # @rbs (Redrive) -> Integer + def move_batch(record) + scope = DeadLetterScope.for_kind(record.kind) + size = batch_size(record) + return 0 unless size.positive? + + identifiers = scope.matching(record.filters).order(:id).limit(size).pluck(:id) + return 0 if identifiers.empty? + + revived = scope.revive_all(identifiers) + record.update!(moved: record.moved + revived) + revived + end + + # @rbs (Redrive) -> Integer + def batch_size(record) + configured = SolidObjects.configuration.redrive_batch_size + limit = record.move_limit + return configured unless limit + + [ configured, limit - record.moved ].min + end + + # @rbs (Redrive) -> void + def finish(record) + manager = SolidObjects.redrives + manager.close(record, status: RedriveManager::COMPLETED) + manager.audit(record, action: "redrive.finish") + end + + # @rbs () -> DatabaseAdapter + def database_adapter + SolidObjects.database_adapter + end + end +end diff --git a/lib/solid_objects/redrive_task.rb b/lib/solid_objects/redrive_task.rb new file mode 100644 index 0000000..2f968f8 --- /dev/null +++ b/lib/solid_objects/redrive_task.rb @@ -0,0 +1,12 @@ +# rbs_inline: enabled + +module SolidObjects + RedriveTask = Data.define( + :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at + ) do + # @rbs (?authorization_context: untyped) -> RedriveTask + def cancel(authorization_context: nil) + SolidObjects.redrives.cancel(id, authorization_context:) + end + end +end diff --git a/lib/solid_objects/supervisor.rb b/lib/solid_objects/supervisor.rb index 3d7ff91..009983b 100644 --- a/lib/solid_objects/supervisor.rb +++ b/lib/solid_objects/supervisor.rb @@ -31,6 +31,7 @@ def initialize( @started = false @cleaned_up_at = nil @retention = nil + @redrive = nil @lifecycle = Thread::Mutex.new end @@ -50,6 +51,7 @@ def start @threads = components.map { |component| supervise(component) } @monitor = Thread.new { monitor_loop } @retention = Thread.new { retention_loop } + @redrive = Thread.new { redrive_loop } SolidObjects.instrument(:"supervisor.started", component_count: components.length) end @@ -64,6 +66,7 @@ def stop @lifecycle.synchronize { @started = false } stop_monitor stop_retention + stop_redrive components.each(&:request_shutdown) join_until_timeout components.reject(&:stopped?).each(&:stop) @@ -204,6 +207,50 @@ def retention_pause(failures) [ backoff, interval ].min end + # An operator starts a redrive and expects it to move, so the supervisor + # advances it rather than ask the application to schedule a job. Each pass + # takes one bounded batch, and the loop pauses between batches so a large + # redrive shares the database with delivery. + # @rbs () -> void + def redrive_loop + while @started + advance_redrive ? pause_between_batches : wait_for_next_redrive + end + end + + # @rbs () -> bool + def advance_redrive + RedriveRunner.new.run_once + rescue => error + SolidObjects.instrument( + :"supervisor.redrive_failed", + error_class: error.class.name, + error_message: error.message + ) + false + end + + # @rbs () -> void + def pause_between_batches + pause = SolidObjects.configuration.redrive_batch_pause + sleep pause if pause.positive? + end + + # @rbs () -> void + def wait_for_next_redrive + sleep SolidObjects.configuration.supervisor_monitor_interval + end + + # @rbs () -> void + def stop_redrive + redrive = @redrive + @redrive = nil + return unless redrive + + redrive.join(SolidObjects.configuration.shutdown_timeout) + redrive.kill if redrive.alive? + end + # @rbs () -> void def stop_retention retention = @retention diff --git a/lib/solid_objects/test_helper.rb b/lib/solid_objects/test_helper.rb index e4bd1c0..5cd466e 100644 --- a/lib/solid_objects/test_helper.rb +++ b/lib/solid_objects/test_helper.rb @@ -32,6 +32,7 @@ def reset_actors! def actor_owned_models [ AdministrationEvent, + Redrive, DeadLetter, ClaimedMessage, ReadyMessage, diff --git a/sig/generated/lib/solid_objects.rbs b/sig/generated/lib/solid_objects.rbs index 7c56789..0d7563d 100644 --- a/sig/generated/lib/solid_objects.rbs +++ b/sig/generated/lib/solid_objects.rbs @@ -39,6 +39,9 @@ module SolidObjects # @rbs () -> DeadLetterManager def self.dead_letters: () -> DeadLetterManager + # @rbs () -> RedriveManager + def self.redrives: () -> RedriveManager + # @rbs () -> Administration def self.administration: () -> Administration diff --git a/sig/generated/lib/solid_objects/administration_audit.rbs b/sig/generated/lib/solid_objects/administration_audit.rbs index 14a839c..9a3c096 100644 --- a/sig/generated/lib/solid_objects/administration_audit.rbs +++ b/sig/generated/lib/solid_objects/administration_audit.rbs @@ -2,8 +2,8 @@ module SolidObjects module AdministrationAudit - # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, authorization_context: untyped) -> void - def self?.record: (action: String, kind: String, authorization_context: untyped, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?) -> void + # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void + def self?.record: (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void # @rbs (untyped) -> String? def self?.identity: (untyped) -> String? diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs index ddc4935..af49e26 100644 --- a/sig/generated/lib/solid_objects/configuration.rbs +++ b/sig/generated/lib/solid_objects/configuration.rbs @@ -4,8 +4,6 @@ module SolidObjects class Configuration @table_name_prefix: String - @process_alive_threshold: Float - @shutdown_timeout: Float @supervisor_monitor_interval: Float @@ -24,6 +22,10 @@ module SolidObjects @prune_batch_size: Integer + @redrive_batch_size: Integer + + @redrive_batch_pause: Float + @worker_count: Integer @effect_worker_count: Integer @@ -100,6 +102,8 @@ module SolidObjects @process_heartbeat_interval: Float + @process_alive_threshold: Float + attr_accessor table_name_prefix: untyped attr_accessor polling_interval: untyped @@ -158,6 +162,10 @@ module SolidObjects attr_accessor prune_batch_size: untyped + attr_accessor redrive_batch_size: untyped + + attr_accessor redrive_batch_pause: untyped + attr_accessor worker_count: untyped attr_accessor effect_worker_count: untyped diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index b058b30..2d46545 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -6,26 +6,40 @@ module SolidObjects PENDING: ::String - @model: untyped + @identifier: Symbol @resource: String - @identifier: Symbol + @model: untyped attr_reader resource: untyped + attr_reader kind: untyped + # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void def initialize: (model: untyped, resource: String, identifier: Symbol, kind: String) -> void + # @rbs (String) -> DeadLetterScope + def self.for_kind: (String) -> DeadLetterScope + # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] def all: (?authorization_context: untyped) -> ActiveRecord::Relation[untyped] # @rbs (String, ?authorization_context: untyped) -> untyped def retry: (String, ?authorization_context: untyped) -> untyped + # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask + def redrive: (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask + # @rbs () -> ActiveRecord::Relation[untyped] def dead: () -> ActiveRecord::Relation[untyped] + # @rbs (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] + def matching: (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] + + # @rbs (Array[untyped]) -> Integer + def revive_all: (Array[untyped]) -> Integer + # @rbs (untyped) -> Integer def revive: (untyped) -> Integer @@ -38,8 +52,6 @@ module SolidObjects attr_reader identifier: untyped - attr_reader kind: untyped - # @rbs () -> Hash[Symbol, untyped] def revival_attributes: () -> Hash[Symbol, untyped] end diff --git a/sig/generated/lib/solid_objects/redrive_manager.rbs b/sig/generated/lib/solid_objects/redrive_manager.rbs new file mode 100644 index 0000000..9b7a7d3 --- /dev/null +++ b/sig/generated/lib/solid_objects/redrive_manager.rbs @@ -0,0 +1,48 @@ +# Generated from lib/solid_objects/redrive_manager.rb with RBS::Inline + +module SolidObjects + class RedriveManager + RUNNING: ::String + + COMPLETED: ::String + + CANCELLED: ::String + + # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask + def start: (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask + + # @rbs (String, ?authorization_context: untyped) -> RedriveTask + def find: (String, ?authorization_context: untyped) -> RedriveTask + + # @rbs (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask] + def all: (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask] + + # @rbs (String, ?authorization_context: untyped) -> RedriveTask + def cancel: (String, ?authorization_context: untyped) -> RedriveTask + + # @rbs (Redrive, status: String) -> void + def close: (Redrive, status: String) -> void + + # @rbs (Redrive, action: String) -> void + def audit: (Redrive, action: String) -> void + + # @rbs (Redrive) -> RedriveTask + def task_for: (Redrive) -> RedriveTask + + private + + # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive + def open_task: (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive + + # A running task reports what is left to move rather than a stored + # estimate, because rows die and are retried while it runs. + # @rbs (Redrive) -> Integer + def remaining_for: (Redrive) -> Integer + + # @rbs (kind: String, filters: Hash[String, untyped]) -> String + def active_scope_for: (kind: String, filters: Hash[String, untyped]) -> String + + # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + def authorize!: (Symbol, authorization_context: untyped, ?resource_id: String?) -> void + end +end diff --git a/sig/generated/lib/solid_objects/redrive_runner.rbs b/sig/generated/lib/solid_objects/redrive_runner.rbs new file mode 100644 index 0000000..651aa60 --- /dev/null +++ b/sig/generated/lib/solid_objects/redrive_runner.rbs @@ -0,0 +1,31 @@ +# Generated from lib/solid_objects/redrive_runner.rb with RBS::Inline + +module SolidObjects + # Moves dead rows back to pending for one redrive task at a time, in bounded + # batches. Each batch is its own short transaction, so a redrive of thousands + # of rows never holds a lock long enough to starve delivery. + class RedriveRunner + # @rbs () -> bool + def run_once: () -> bool + + private + + # @rbs () -> Redrive? + def claim: () -> Redrive? + + # @rbs (Redrive) -> bool + def advance: (Redrive) -> bool + + # @rbs (Redrive) -> Integer + def move_batch: (Redrive) -> Integer + + # @rbs (Redrive) -> Integer + def batch_size: (Redrive) -> Integer + + # @rbs (Redrive) -> void + def finish: (Redrive) -> void + + # @rbs () -> DatabaseAdapter + def database_adapter: () -> DatabaseAdapter + end +end diff --git a/sig/generated/lib/solid_objects/redrive_task.rbs b/sig/generated/lib/solid_objects/redrive_task.rbs new file mode 100644 index 0000000..ee52d83 --- /dev/null +++ b/sig/generated/lib/solid_objects/redrive_task.rbs @@ -0,0 +1,28 @@ +# Generated from lib/solid_objects/redrive_task.rb with RBS::Inline + +module SolidObjects + class RedriveTask < Data + attr_reader id(): untyped + + attr_reader kind(): untyped + + attr_reader filters(): untyped + + attr_reader status(): untyped + + attr_reader moved(): untyped + + attr_reader remaining(): untyped + + attr_reader started_at(): untyped + + attr_reader finished_at(): untyped + + def self.new: (untyped id, untyped kind, untyped filters, untyped status, untyped moved, untyped remaining, untyped started_at, untyped finished_at) -> instance + | (id: untyped, kind: untyped, filters: untyped, status: untyped, moved: untyped, remaining: untyped, started_at: untyped, finished_at: untyped) -> instance + + def self.members: () -> [ :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at ] + + def members: () -> [ :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at ] + end +end diff --git a/sig/generated/lib/solid_objects/supervisor.rbs b/sig/generated/lib/solid_objects/supervisor.rbs index 0572fda..a88fd95 100644 --- a/sig/generated/lib/solid_objects/supervisor.rbs +++ b/sig/generated/lib/solid_objects/supervisor.rbs @@ -83,6 +83,25 @@ module SolidObjects # @rbs (Integer) -> Float def retention_pause: (Integer) -> Float + # An operator starts a redrive and expects it to move, so the supervisor + # advances it rather than ask the application to schedule a job. Each pass + # takes one bounded batch, and the loop pauses between batches so a large + # redrive shares the database with delivery. + # @rbs () -> void + def redrive_loop: () -> void + + # @rbs () -> bool + def advance_redrive: () -> bool + + # @rbs () -> void + def pause_between_batches: () -> void + + # @rbs () -> void + def wait_for_next_redrive: () -> void + + # @rbs () -> void + def stop_redrive: () -> void + # @rbs () -> void def stop_retention: () -> void diff --git a/sig/generated/models/solid_objects/redrive.rbs b/sig/generated/models/solid_objects/redrive.rbs new file mode 100644 index 0000000..0b51664 --- /dev/null +++ b/sig/generated/models/solid_objects/redrive.rbs @@ -0,0 +1,6 @@ +# Generated from app/models/solid_objects/redrive.rb with RBS::Inline + +module SolidObjects + class Redrive < Record + end +end diff --git a/test/database_test_helper.rb b/test/database_test_helper.rb index c49163a..0ff0f22 100644 --- a/test/database_test_helper.rb +++ b/test/database_test_helper.rb @@ -26,12 +26,14 @@ require_relative "../db/migrate/20260813000000_rename_message_dispatch_columns" require_relative "../db/migrate/20260915000000_add_solid_objects_effect_recoveries" require_relative "../db/migrate/20260922000000_add_solid_objects_administration_events" +require_relative "../db/migrate/20260922000001_add_solid_objects_redrives" CreateSolidObjectsTables.new.migrate(:up) AddStateRevisionToSolidObjectsInstances.new.migrate(:up) RenameMessageDispatchColumns.new.migrate(:up) AddSolidObjectsEffectRecoveries.new.migrate(:up) AddSolidObjectsAdministrationEvents.new.migrate(:up) +AddSolidObjectsRedrives.new.migrate(:up) ActiveRecord::Base.connection.create_table(:solid_objects_test_domain_records) do |table| table.string :name, null: false @@ -40,6 +42,7 @@ require "solid_objects/database_adapter" require_relative "../app/models/solid_objects/record" require_relative "../app/models/solid_objects/administration_event" +require_relative "../app/models/solid_objects/redrive" require_relative "../app/models/solid_objects/process" require_relative "../app/models/solid_objects/instance" require_relative "../app/models/solid_objects/message" @@ -63,6 +66,7 @@ class ActiveSupport::TestCase SolidObjects::Instance.delete_all SolidObjects::Process.delete_all SolidObjects::AdministrationEvent.delete_all + SolidObjects::Redrive.delete_all SolidObjectsTestDomainRecord.delete_all end diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb new file mode 100644 index 0000000..9f9dbe8 --- /dev/null +++ b/test/integration/redrive_test.rb @@ -0,0 +1,280 @@ +# frozen_string_literal: true + +require "database_test_helper" + +class RedriveTest < ActiveSupport::TestCase + class PaymentActor < SolidObjects::Actor + actor_type "redrive-payments" + + attribute :placed, default: 0 + + def place(order:) + self.placed += 1 + emit :settle, order: order + end + end + + class ShipmentActor < SolidObjects::Actor + actor_type "redrive-shipments" + + attribute :count, default: 0 + + observable :count + + def touch + self.count += 1 + end + end + + setup do + SolidObjects.configuration.retry_delay = ->(_attempt) { 0 } + SolidObjects.configuration.authorize_administration = ->(**) { true } + SolidObjects.configuration.redrive_batch_size = 10 + SolidObjects.configuration.redrive_batch_pause = 0 + end + + test "moves every matching row in bounded batches" do + dead_effects(25) + + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + runner = SolidObjects::RedriveRunner.new + + assert_equal "running", task.status + assert runner.run_once + assert_equal 10, SolidObjects::Effect.where(status: "pending").count + assert runner.run_once + assert_equal 20, SolidObjects::Effect.where(status: "pending").count + assert runner.run_once + assert_equal 25, SolidObjects::Effect.where(status: "pending").count + + refute runner.run_once + finished = SolidObjects.redrives.find(task.id, authorization_context: "operator") + assert_equal "completed", finished.status + assert_equal 25, finished.moved + assert_equal 0, finished.remaining + end + + test "stops at the limit and leaves the rest dead" do + dead_effects(25) + + task = SolidObjects.dead_letters.effects.redrive(limit: 15, authorization_context: "operator") + drain + + finished = SolidObjects.redrives.find(task.id, authorization_context: "operator") + assert_equal 15, finished.moved + assert_equal "completed", finished.status + assert_equal 10, SolidObjects::Effect.where(status: "dead").count + end + + test "returns the running task when the same scope is redriven again" do + dead_effects(25) + + first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + second = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + assert_equal first.id, second.id + assert_equal 1, SolidObjects::Redrive.count + end + + test "starts a separate task for another scope while one runs" do + dead_effects(5) + dead_broadcast + + effects_task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + broadcasts_task = SolidObjects.dead_letters.broadcasts.redrive(authorization_context: "operator") + + refute_equal effects_task.id, broadcasts_task.id + assert_equal [ "broadcast", "effect" ], SolidObjects::Redrive.pluck(:kind).sort + end + + test "starts a separate task for different filters" do + dead_effects(5) + + first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + second = SolidObjects.dead_letters.effects.redrive( + actor_type: "redrive-payments", + authorization_context: "operator" + ) + + refute_equal first.id, second.id + end + + test "starts a new task once the first finishes" do + dead_effects(5) + first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + drain + SolidObjects::Effect.update_all(status: "dead") + + second = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + refute_equal first.id, second.id + assert_equal "running", second.status + end + + test "cancels a running task and keeps the rows it already moved" do + dead_effects(25) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + runner = SolidObjects::RedriveRunner.new + runner.run_once + + task.cancel(authorization_context: "operator") + + refute runner.run_once + cancelled = SolidObjects.redrives.find(task.id, authorization_context: "operator") + assert_equal "cancelled", cancelled.status + assert_equal 10, cancelled.moved + assert_equal 10, SolidObjects::Effect.where(status: "pending").count + assert_equal 15, SolidObjects::Effect.where(status: "dead").count + end + + test "filters by actor type" do + dead_effects(3) + ShipmentActor.ref("one").async.touch + run_actors + + task = SolidObjects.dead_letters.effects.redrive( + actor_type: "redrive-shipments", + authorization_context: "operator" + ) + drain + + finished = SolidObjects.redrives.find(task.id, authorization_context: "operator") + assert_equal 0, finished.moved + assert_equal 3, SolidObjects::Effect.where(status: "dead").count + end + + test "filters by failure time" do + dead_effects(3) + future = SolidObjects.database_adapter.database_now + 60 + + task = SolidObjects.dead_letters.effects.redrive( + failed_after: future, + authorization_context: "operator" + ) + drain + + assert_equal 0, SolidObjects.redrives.find(task.id, authorization_context: "operator").moved + end + + test "reads tasks back by id and by status" do + dead_effects(5) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + running = SolidObjects.redrives.all(status: :running, authorization_context: "operator") + assert_equal [ task.id ], running.map(&:id) + + drain + + assert_empty SolidObjects.redrives.all(status: :running, authorization_context: "operator") + completed = SolidObjects.redrives.all(status: :completed, authorization_context: "operator") + assert_equal [ task.id ], completed.map(&:id) + assert_equal [ task.id ], SolidObjects.redrives.all(authorization_context: "operator").map(&:id) + end + + test "writes one audit row for each task transition" do + dead_effects(5) + + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + drain + + events = SolidObjects::AdministrationEvent.order(:id).map(&:action) + assert_equal [ "redrive.start", "redrive.finish" ], events + assert_equal [ task.id, task.id ], SolidObjects::AdministrationEvent.order(:id).map(&:subject_id) + assert_equal({ "actor_type" => nil, "failed_after" => nil, "limit" => nil }, + SolidObjects::AdministrationEvent.order(:id).first.filters) + end + + test "writes one audit row when a task is cancelled" do + dead_effects(5) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + task.cancel(authorization_context: "operator") + + assert_equal [ "redrive.start", "redrive.cancel" ], + SolidObjects::AdministrationEvent.order(:id).map(&:action) + end + + test "refuses an unauthorized caller" do + dead_effects(5) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + SolidObjects.configuration.authorize_administration = ->(**) { false } + + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + end + assert_raises(SolidObjects::Unauthorized) do + task.cancel(authorization_context: "operator") + end + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.redrives.all(authorization_context: "operator") + end + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.redrives.find(task.id, authorization_context: "operator") + end + end + + test "a supervised runtime advances a redrive without a caller driving it" do + dead_effects(12) + SolidObjects.configuration.redrive_batch_size = 4 + supervisor = SolidObjects::Supervisor.new( + worker_count: 0, + effect_worker_count: 0, + broadcast_worker_count: 0, + reminder_scheduler_count: 1 + ) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + supervisor.start + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 10 + until SolidObjects.redrives.find(task.id, authorization_context: "operator").status == "completed" + flunk "the supervisor did not finish the redrive" if + Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + sleep 0.05 + end + + assert_equal 12, SolidObjects::Effect.where(status: "pending").count + ensure + supervisor&.stop + end + + test "reports a task as a frozen value" do + dead_effects(1) + + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + + assert_kind_of SolidObjects::RedriveTask, task + assert task.frozen? + assert_equal "effect", task.kind + assert task.started_at + assert_nil task.finished_at + end + + private + + def dead_effects(count) + SolidObjects.register_effect(:settle) { raise "declined" } + count.times { |index| PaymentActor.ref("order-#{index}").async.place(order: index) } + run_actors + SolidObjects::Effect.update_all(status: "dead") + assert_equal count, SolidObjects::Effect.where(status: "dead").count + end + + def dead_broadcast + SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" } + ShipmentActor.ref("one").async.touch + run_actors + SolidObjects::Broadcast.update_all(status: "dead") + end + + def drain + runner = SolidObjects::RedriveRunner.new + nil while runner.run_once + end + + def run_actors + worker = SolidObjects::Worker.new + worker.run_until_idle + ensure + worker&.stop + end +end From d7e3e0e2725d9c8544c6e28c4366e552ffef387e Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 12:57:26 -0700 Subject: [PATCH 04/15] docs: describe retry and redrive The roadmap still named the gaps this branch closes: no retry for a dead effect or broadcast, one row at a time, and no record of who pressed what. It now states what exists and what still does not, which is the dashboard surface for the new scopes. Operations gains the API, the idempotency rule, the batching, the authorization resource names, and the audit row. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 22 +++++++++++++ docs/dashboard.md | 10 ++++++ docs/operations.md | 77 ++++++++++++++++++++++++++++++++++++++++++++++ docs/roadmap.md | 21 +++++++------ 4 files changed, 121 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 46cf05e..08c7fd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,28 @@ ## Unreleased +- Retry a dead effect or broadcast. `SolidObjects.dead_letters` keeps its + message meaning and answers `effects` and `broadcasts`, so the kind rides on + the receiver. `retry` returns a dead row to pending with a zero attempt count + and no claim, reuses the stable id so a deduplicating handler sees the same + key, and acts only on a dead row, so a second press cannot double-enqueue. A + dead transmit effect replays rather than stay lost. +- Redrive a whole scope. `redrive` opens a durable task, returns at once, and is + idempotent over its scope and filters, which a dashboard button needs. A + unique index on the active scope enforces that in the database, so two + processes that start the same redrive share one task. The supervisor advances + one bounded batch per pass, so a redrive never holds a transaction longer than + one batch. `SolidObjects.redrives` reads tasks back, and `task.cancel` stops + one and leaves the rows it already moved. +- Record who pressed what. Every retry and every redrive transition writes one + row to `solid_objects_administration_events`. The identity comes from the + authorization context through a new `administration_identity` hook. +- Add `redrive_batch_size`, which defaults to 100, and `redrive_batch_pause`, + which defaults to 0.05 seconds. +- Add two tables, `solid_objects_administration_events` and + `solid_objects_redrives`. Run `bin/rails solid_objects:install:migrations` and + migrate. + - Select a wake-up adapter automatically. `config.wake_up_adapter` now takes a name or an adapter, as `config.cache_store` and `config.active_job.queue_adapter` do, and defaults to `:automatic`. Selection diff --git a/docs/dashboard.md b/docs/dashboard.md index 395c9a5..6a408a2 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -149,6 +149,16 @@ actor class that no longer exists, a full mailbox, a payload over the cap. The dashboard renders the dead letter again with the reason and a 422 status, rather than failing the request. +Dead effects and broadcasts have the same API, which the dashboard does not yet +surface. `SolidObjects.dead_letters.effects` and +`SolidObjects.dead_letters.broadcasts` read and retry their own kind, and +`redrive` moves a whole scope as a durable task. See +[Operations](operations.md) for both. + +Every retry and every redrive transition writes one row to +`solid_objects_administration_events`, holding the action, the kind, the +subject, and the identity that asked for it. + **Pause an instance** sets `paused_at`, and the activation manager stops claiming that identity. Two consequences matter: diff --git a/docs/operations.md b/docs/operations.md index 38b219f..783a07d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -282,6 +282,83 @@ interval, and current interval. The polling-only warning is also emitted as adapters should return `true` for a notification and `false` for a timeout; an older adapter that returns `nil` remains compatible and keeps the fast cadence. +## Dead letters, retry, and redrive + +A message that exhausts its attempts becomes a dead letter. An effect or a +broadcast that exhausts its attempts stays in its own table with +`status = 'dead'`. All three are read and retried through one receiver, which +carries the kind: + +```ruby +SolidObjects.dead_letters.all(authorization_context: current_admin) +SolidObjects.dead_letters.retry(dead_letter_id, authorization_context: current_admin) + +SolidObjects.dead_letters.effects.all(authorization_context: current_admin) +SolidObjects.dead_letters.effects.retry(effect_id, authorization_context: current_admin) +SolidObjects.dead_letters.broadcasts.retry(broadcast_id, authorization_context: current_admin) +``` + +An effect or broadcast retry returns the row to pending with a zero attempt +count, no claim, and immediate availability. It keeps the stable id, so a +handler that deduplicates on `effect_id` still sees the same key. An effect is +at-least-once by contract, so a retried effect can run twice. + +Retry acts only on a dead row. A row that is pending, processing, or completed +comes back unchanged, so pressing a button twice cannot double-enqueue and +cannot take a row away from a worker that holds it. + +An incident produces dead rows in the hundreds, so a scope also answers +`redrive`: + +```ruby +task = SolidObjects.dead_letters.effects.redrive( + actor_type: "payments", + failed_after: 6.hours.ago, + limit: 5_000, + authorization_context: current_admin +) + +task.id # => "redrive_..." +task.status # => "running" +task.moved # => 412 +task.remaining # => 4_588 + +task.cancel(authorization_context: current_admin) +``` + +`redrive` returns at once. The task is durable, and the supervisor advances one +bounded batch per pass, so a redrive of thousands of rows never holds a +transaction longer than one batch. `redrive_batch_size` defaults to 100 and +`redrive_batch_pause` to 0.05 seconds. + +A redrive is idempotent over its scope and its filters. Starting the same one +while it runs returns the running task rather than a second one, which a +dashboard button an operator can press twice needs. A different scope or a +different filter starts its own task, and the same scope can be redriven again +once the first task finishes. + +Read tasks back with `SolidObjects.redrives`: + +```ruby +SolidObjects.redrives.find(task.id, authorization_context: current_admin) +SolidObjects.redrives.all(status: :running, authorization_context: current_admin) +``` + +A running task reports what is left to move rather than a stored estimate, +because rows die and are retried while it runs. + +Retry, redrive, and cancel each go through `authorize_administration` under +their own resource name: `dead_letters`, `effect_dead_letters`, +`broadcast_dead_letters`, and `redrives`. Every retry and every task transition +writes one row to `solid_objects_administration_events`, holding the action, the +kind, the subject, the identity, and when it happened. The identity comes from +`administration_identity`, which receives the authorization context the caller +passed and defaults to its `to_s`. A refused caller writes nothing. + +Automatic redrive on a schedule is deliberately absent. A dead row means a +person decided something, and these APIs give that person an alternative to an +`UPDATE` against a runtime table. + ## Graceful shutdown The supervisor requests shutdown, stops new claims, lets active loops return, diff --git a/docs/roadmap.md b/docs/roadmap.md index b53f245..8fe2c27 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -178,15 +178,18 @@ or turn off. Every route declares its own administration policy and a route declared without one raises at load time, so the deny-by-default posture is enforced by construction rather than - by remembering to add a check. It changes only two things: an idempotent dead - letter retry and instance pause/resume. What does not exist is audit records - of who pressed what, and bulk-safe tools: retry is one dead letter at a time, - because `DeadLetterManager` exposes no bulk operation. Pause is an operator - brake and not a stop, since a pass already in flight finishes its turn and a - synchronous caller waiting on a paused instance times out. Retry also only - exists for message dead letters: a dead effect or broadcast has no retry - API, which matters for transmit effects because a dead one is a lost - replay until an operator returns its row to pending. The page cost was + by remembering to add a check. It changes only three things: an idempotent dead + letter retry, a redrive, and instance pause/resume. Retry covers all three + kinds. `SolidObjects.dead_letters` keeps its message meaning and answers + `effects` and `broadcasts`, so a dead effect or broadcast returns to pending + through an API rather than an operator's `UPDATE`, and a dead transmit effect + is no longer a lost replay. `redrive` moves a whole scope as a durable task + that is idempotent over its filters, cancellable, and advanced in bounded + batches by the supervisor. Every retry and task transition writes one row to + `solid_objects_administration_events`, so who pressed what is recorded. Pause + is an operator brake and not a stop, since a pass already in flight finishes + its turn and a synchronous caller waiting on a paused instance times out. The + dashboard does not yet surface the scopes or redrive; the API does. The page cost was reasoned about rather than measured: the summary bar issues a fixed set of indexed aggregate queries per page, which is why `HEAD /` exists for uptime monitors, but no dashboard latency has been benchmarked against a large From ec42211d16d1a81ac366bdc0d018f412f1b47667 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:28:13 -0700 Subject: [PATCH 05/15] fix: stop a redrive from moving a row twice A redrive read the scope on every pass, so a row it moved that failed again landed straight back in it. With a handler that was still broken, a task without a limit would move the same rows forever and never finish. Porting this to the TypeScript runtime is where it showed: there the workers run beside the redrive, so the churn was immediate. A pass now takes only rows that were already dead when the task started. Co-Authored-By: Claude Opus 5 (1M context) --- docs/operations.md | 4 +++ lib/solid_objects/dead_letter_scope.rb | 8 ++++-- lib/solid_objects/redrive_manager.rb | 5 +++- lib/solid_objects/redrive_runner.rb | 6 +++- .../lib/solid_objects/dead_letter_scope.rbs | 7 +++-- test/integration/redrive_test.rb | 28 +++++++++++++++++-- 6 files changed, 49 insertions(+), 9 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index 783a07d..0f40472 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -331,6 +331,10 @@ bounded batch per pass, so a redrive of thousands of rows never holds a transaction longer than one batch. `redrive_batch_size` defaults to 100 and `redrive_batch_pause` to 0.05 seconds. +A redrive moves the rows that were already dead when it started. A row that +fails again lands back in the same scope, and without that bound a task whose +handler is still broken would move it forever. + A redrive is idempotent over its scope and its filters. Starting the same one while it runs returns the running task rather than a second one, which a dashboard button an operator can press twice needs. A different scope or a diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 691c5af..be42775 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -68,13 +68,17 @@ def dead model.where(status: DEAD) end - # @rbs (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] - def matching(filters) + # A redrive moves what was already dead when it started. Without that bound + # a row that fails again lands back in the same scope, and a task whose + # handler is still broken would move it forever. + # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] + def matching(filters, dead_before: nil) relation = dead actor_type = filters["actor_type"] failed_after = filters["failed_after"] relation = relation.joins(:instance).where(Instance.table_name => { actor_type: }) if actor_type relation = relation.where(updated_at: Time.parse(failed_after)..) if failed_after + relation = relation.where(updated_at: ..dead_before) if dead_before relation end diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb index aaec6ff..7377eb6 100644 --- a/lib/solid_objects/redrive_manager.rb +++ b/lib/solid_objects/redrive_manager.rb @@ -103,7 +103,10 @@ def open_task(scope:, filters:, active_scope:, authorization_context:) def remaining_for(record) return 0 unless record.status == RUNNING - matching = DeadLetterScope.for_kind(record.kind).matching(record.filters).count + matching = DeadLetterScope + .for_kind(record.kind) + .matching(record.filters, dead_before: record.started_at) + .count limit = record.move_limit return matching unless limit diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb index 663fb64..999649a 100644 --- a/lib/solid_objects/redrive_runner.rb +++ b/lib/solid_objects/redrive_runner.rb @@ -37,7 +37,11 @@ def move_batch(record) size = batch_size(record) return 0 unless size.positive? - identifiers = scope.matching(record.filters).order(:id).limit(size).pluck(:id) + identifiers = scope + .matching(record.filters, dead_before: record.started_at) + .order(:id) + .limit(size) + .pluck(:id) return 0 if identifiers.empty? revived = scope.revive_all(identifiers) diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index 2d46545..f8a08fd 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -34,8 +34,11 @@ module SolidObjects # @rbs () -> ActiveRecord::Relation[untyped] def dead: () -> ActiveRecord::Relation[untyped] - # @rbs (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] - def matching: (Hash[String, untyped]) -> ActiveRecord::Relation[untyped] + # A redrive moves what was already dead when it started. Without that bound + # a row that fails again lands back in the same scope, and a task whose + # handler is still broken would move it forever. + # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] + def matching: (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] # @rbs (Array[untyped]) -> Integer def revive_all: (Array[untyped]) -> Integer diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb index 9f9dbe8..4cd64e9 100644 --- a/test/integration/redrive_test.rb +++ b/test/integration/redrive_test.rb @@ -103,7 +103,9 @@ def touch dead_effects(5) first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") drain - SolidObjects::Effect.update_all(status: "dead") + SolidObjects::Effect.where.not(status: "dead").update_all( + status: "dead", updated_at: SolidObjects.database_adapter.database_now + ) second = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") @@ -127,6 +129,22 @@ def touch assert_equal 15, SolidObjects::Effect.where(status: "dead").count end + test "does not move a row that died after the task started" do + dead_effects(2) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + sleep 0.01 + PaymentActor.ref("late").async.place(order: "late") + run_actors + SolidObjects::Effect.where.not(status: "dead").update_all( + status: "dead", updated_at: SolidObjects.database_adapter.database_now + ) + + drain + + assert_equal 2, SolidObjects.redrives.find(task.id, authorization_context: "operator").moved + assert_equal 1, SolidObjects::Effect.where(status: "dead").count + end + test "filters by actor type" do dead_effects(3) ShipmentActor.ref("one").async.touch @@ -255,7 +273,9 @@ def dead_effects(count) SolidObjects.register_effect(:settle) { raise "declined" } count.times { |index| PaymentActor.ref("order-#{index}").async.place(order: index) } run_actors - SolidObjects::Effect.update_all(status: "dead") + SolidObjects::Effect.where.not(status: "dead").update_all( + status: "dead", updated_at: SolidObjects.database_adapter.database_now + ) assert_equal count, SolidObjects::Effect.where(status: "dead").count end @@ -263,7 +283,9 @@ def dead_broadcast SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" } ShipmentActor.ref("one").async.touch run_actors - SolidObjects::Broadcast.update_all(status: "dead") + SolidObjects::Broadcast.where.not(status: "dead").update_all( + status: "dead", updated_at: SolidObjects.database_adapter.database_now + ) end def drain From 29e2152edce789836385b86d6f762a916614e99d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:30:33 -0700 Subject: [PATCH 06/15] docs: record the redrive bound in the changelog Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 08c7fd1..4c34a76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,8 @@ processes that start the same redrive share one task. The supervisor advances one bounded batch per pass, so a redrive never holds a transaction longer than one batch. `SolidObjects.redrives` reads tasks back, and `task.cancel` stops - one and leaves the rows it already moved. + one and leaves the rows it already moved. A redrive moves what was dead when + it started, so a still-broken handler cannot make it run forever. - Record who pressed what. Every retry and every redrive transition writes one row to `solid_objects_administration_events`. The identity comes from the authorization context through a new `administration_identity` hook. From 94786d46047830a233273490d10777c250e3902c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:35:06 -0700 Subject: [PATCH 07/15] fix: authorize a redrive where it cannot be bypassed `RedriveManager#start` is public and ran no check of its own, so the only guard was the scope that normally calls it. A caller that reached the manager directly started a task unauthorized. The check moves into `start`, under the scope's own resource name, so there is one check and no way around it. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/dead_letter_scope.rb | 1 - lib/solid_objects/redrive_manager.rb | 1 + test/integration/redrive_test.rb | 12 ++++++++++++ 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index be42775..fb9d494 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -51,7 +51,6 @@ def retry(identifier_value, authorization_context: nil) # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask def redrive(actor_type: nil, failed_after: nil, limit: nil, authorization_context: nil) - authorize!(:redrive, authorization_context:) SolidObjects.redrives.start( scope: self, filters: { diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb index 7377eb6..a962585 100644 --- a/lib/solid_objects/redrive_manager.rb +++ b/lib/solid_objects/redrive_manager.rb @@ -10,6 +10,7 @@ class RedriveManager # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask def start(scope:, filters:, authorization_context:) + scope.authorize!(:redrive, authorization_context:) active_scope = active_scope_for(kind: scope.kind, filters:) running = Redrive.find_by(active_scope:) return task_for(running) if running diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb index 4cd64e9..8e30da1 100644 --- a/test/integration/redrive_test.rb +++ b/test/integration/redrive_test.rb @@ -212,6 +212,18 @@ def touch SolidObjects::AdministrationEvent.order(:id).map(&:action) end + test "refuses an unauthorized caller that reaches the manager directly" do + SolidObjects.configuration.authorize_administration = ->(**) { false } + + assert_raises(SolidObjects::Unauthorized) do + SolidObjects.redrives.start( + scope: SolidObjects.dead_letters.effects, + filters: { "actor_type" => nil, "failed_after" => nil, "limit" => nil }, + authorization_context: "operator" + ) + end + end + test "refuses an unauthorized caller" do dead_effects(5) task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") From 5c7d4f0634ddfc0ea82d3b9bd740aeae8c41734e Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 13:47:32 -0700 Subject: [PATCH 08/15] fix: claim a redrive the way every other role claims work `Redrive.lock` takes a plain `FOR UPDATE`, so with a redrive thread in every supervisor the second one waits on the first rather than taking the next task. It uses `lock_candidates` now, which is `FOR UPDATE SKIP LOCKED` where the database has it, as the actor, effect, broadcast, and reminder claims already do. `RedriveTask#cancel` moves out of the `Data.define` block, because rbs-inline does not read that block and the shipped signature therefore omitted a documented method. The migrations inline their one-use JSON helper, and four comments that restated the code they sat above are gone. The reasoning they carried is in the commit that introduced each one. Co-Authored-By: Claude Opus 5 (1M context) --- ...00000_add_solid_objects_administration_events.rb | 9 ++------- .../20260922000001_add_solid_objects_redrives.rb | 9 ++------- lib/solid_objects/dead_letter_scope.rb | 3 --- lib/solid_objects/redrive_manager.rb | 2 -- lib/solid_objects/redrive_runner.rb | 13 +++---------- lib/solid_objects/redrive_task.rb | 4 +++- lib/solid_objects/supervisor.rb | 4 ---- .../lib/solid_objects/dead_letter_scope.rbs | 3 --- sig/generated/lib/solid_objects/redrive_manager.rbs | 2 -- sig/generated/lib/solid_objects/redrive_runner.rbs | 6 ------ sig/generated/lib/solid_objects/redrive_task.rbs | 5 +++++ sig/generated/lib/solid_objects/supervisor.rbs | 4 ---- 12 files changed, 15 insertions(+), 49 deletions(-) diff --git a/db/migrate/20260922000000_add_solid_objects_administration_events.rb b/db/migrate/20260922000000_add_solid_objects_administration_events.rb index 94aa632..95e69c4 100644 --- a/db/migrate/20260922000000_add_solid_objects_administration_events.rb +++ b/db/migrate/20260922000000_add_solid_objects_administration_events.rb @@ -7,7 +7,7 @@ def change definition.string :action, null: false, limit: 64 definition.string :kind, null: false, limit: 32 definition.string :subject_id, limit: 191 - json_column definition, :filters + definition.public_send(json_type, :filters) definition.string :actor, limit: 255 definition.datetime :occurred_at, null: false, precision: 6 definition.timestamps precision: 6, null: false @@ -20,12 +20,7 @@ def change private # @rbs () -> Symbol - def json_column_type + def json_type connection.adapter_name.match?(/postgres/i) ? :jsonb : :json end - - # @rbs (untyped, Symbol, ?null: bool) -> void - def json_column(definition, name, null: true) - definition.public_send(json_column_type, name, null:) - end end diff --git a/db/migrate/20260922000001_add_solid_objects_redrives.rb b/db/migrate/20260922000001_add_solid_objects_redrives.rb index 4e36d56..4da7471 100644 --- a/db/migrate/20260922000001_add_solid_objects_redrives.rb +++ b/db/migrate/20260922000001_add_solid_objects_redrives.rb @@ -5,7 +5,7 @@ class AddSolidObjectsRedrives < ActiveRecord::Migration[7.1] def change create_table SolidObjects.table_name(:redrives), id: :string, limit: 64 do |definition| definition.string :kind, null: false, limit: 32 - json_column definition, :filters, null: false + definition.public_send(json_type, :filters, null: false) definition.string :status, null: false, default: "running", limit: 32 definition.string :active_scope, limit: 191 definition.integer :moved, null: false, default: 0 @@ -26,12 +26,7 @@ def change private # @rbs () -> Symbol - def json_column_type + def json_type connection.adapter_name.match?(/postgres/i) ? :jsonb : :json end - - # @rbs (untyped, Symbol, ?null: bool) -> void - def json_column(definition, name, null: true) - definition.public_send(json_column_type, name, null:) - end end diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index fb9d494..9257db2 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -67,9 +67,6 @@ def dead model.where(status: DEAD) end - # A redrive moves what was already dead when it started. Without that bound - # a row that fails again lands back in the same scope, and a task whose - # handler is still broken would move it forever. # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] def matching(filters, dead_before: nil) relation = dead diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb index a962585..4987834 100644 --- a/lib/solid_objects/redrive_manager.rb +++ b/lib/solid_objects/redrive_manager.rb @@ -98,8 +98,6 @@ def open_task(scope:, filters:, active_scope:, authorization_context:) record end - # A running task reports what is left to move rather than a stored - # estimate, because rows die and are retried while it runs. # @rbs (Redrive) -> Integer def remaining_for(record) return 0 unless record.status == RUNNING diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb index 999649a..3839ab6 100644 --- a/lib/solid_objects/redrive_runner.rb +++ b/lib/solid_objects/redrive_runner.rb @@ -1,13 +1,10 @@ # rbs_inline: enabled module SolidObjects - # Moves dead rows back to pending for one redrive task at a time, in bounded - # batches. Each batch is its own short transaction, so a redrive of thousands - # of rows never holds a lock long enough to starve delivery. class RedriveRunner # @rbs () -> bool def run_once - database_adapter.transaction do + SolidObjects.database_adapter.transaction do record = claim next false unless record @@ -19,7 +16,8 @@ def run_once # @rbs () -> Redrive? def claim - Redrive.lock.where(status: RedriveManager::RUNNING).order(:started_at, :id).first + relation = Redrive.where(status: RedriveManager::RUNNING).order(:started_at, :id) + SolidObjects.database_adapter.lock_candidates(relation).first end # @rbs (Redrive) -> bool @@ -64,10 +62,5 @@ def finish(record) manager.close(record, status: RedriveManager::COMPLETED) manager.audit(record, action: "redrive.finish") end - - # @rbs () -> DatabaseAdapter - def database_adapter - SolidObjects.database_adapter - end end end diff --git a/lib/solid_objects/redrive_task.rb b/lib/solid_objects/redrive_task.rb index 2f968f8..1172352 100644 --- a/lib/solid_objects/redrive_task.rb +++ b/lib/solid_objects/redrive_task.rb @@ -3,7 +3,9 @@ module SolidObjects RedriveTask = Data.define( :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at - ) do + ) + + class RedriveTask # @rbs (?authorization_context: untyped) -> RedriveTask def cancel(authorization_context: nil) SolidObjects.redrives.cancel(id, authorization_context:) diff --git a/lib/solid_objects/supervisor.rb b/lib/solid_objects/supervisor.rb index 009983b..f3f27e3 100644 --- a/lib/solid_objects/supervisor.rb +++ b/lib/solid_objects/supervisor.rb @@ -207,10 +207,6 @@ def retention_pause(failures) [ backoff, interval ].min end - # An operator starts a redrive and expects it to move, so the supervisor - # advances it rather than ask the application to schedule a job. Each pass - # takes one bounded batch, and the loop pauses between batches so a large - # redrive shares the database with delivery. # @rbs () -> void def redrive_loop while @started diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index f8a08fd..3f65fd2 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -34,9 +34,6 @@ module SolidObjects # @rbs () -> ActiveRecord::Relation[untyped] def dead: () -> ActiveRecord::Relation[untyped] - # A redrive moves what was already dead when it started. Without that bound - # a row that fails again lands back in the same scope, and a task whose - # handler is still broken would move it forever. # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] def matching: (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped] diff --git a/sig/generated/lib/solid_objects/redrive_manager.rbs b/sig/generated/lib/solid_objects/redrive_manager.rbs index 9b7a7d3..85c5440 100644 --- a/sig/generated/lib/solid_objects/redrive_manager.rbs +++ b/sig/generated/lib/solid_objects/redrive_manager.rbs @@ -34,8 +34,6 @@ module SolidObjects # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive def open_task: (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive - # A running task reports what is left to move rather than a stored - # estimate, because rows die and are retried while it runs. # @rbs (Redrive) -> Integer def remaining_for: (Redrive) -> Integer diff --git a/sig/generated/lib/solid_objects/redrive_runner.rbs b/sig/generated/lib/solid_objects/redrive_runner.rbs index 651aa60..ae6e12b 100644 --- a/sig/generated/lib/solid_objects/redrive_runner.rbs +++ b/sig/generated/lib/solid_objects/redrive_runner.rbs @@ -1,9 +1,6 @@ # Generated from lib/solid_objects/redrive_runner.rb with RBS::Inline module SolidObjects - # Moves dead rows back to pending for one redrive task at a time, in bounded - # batches. Each batch is its own short transaction, so a redrive of thousands - # of rows never holds a lock long enough to starve delivery. class RedriveRunner # @rbs () -> bool def run_once: () -> bool @@ -24,8 +21,5 @@ module SolidObjects # @rbs (Redrive) -> void def finish: (Redrive) -> void - - # @rbs () -> DatabaseAdapter - def database_adapter: () -> DatabaseAdapter end end diff --git a/sig/generated/lib/solid_objects/redrive_task.rbs b/sig/generated/lib/solid_objects/redrive_task.rbs index ee52d83..301e732 100644 --- a/sig/generated/lib/solid_objects/redrive_task.rbs +++ b/sig/generated/lib/solid_objects/redrive_task.rbs @@ -25,4 +25,9 @@ module SolidObjects def members: () -> [ :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at ] end + + class RedriveTask + # @rbs (?authorization_context: untyped) -> RedriveTask + def cancel: (?authorization_context: untyped) -> RedriveTask + end end diff --git a/sig/generated/lib/solid_objects/supervisor.rbs b/sig/generated/lib/solid_objects/supervisor.rbs index a88fd95..bf179f1 100644 --- a/sig/generated/lib/solid_objects/supervisor.rbs +++ b/sig/generated/lib/solid_objects/supervisor.rbs @@ -83,10 +83,6 @@ module SolidObjects # @rbs (Integer) -> Float def retention_pause: (Integer) -> Float - # An operator starts a redrive and expects it to move, so the supervisor - # advances it rather than ask the application to schedule a job. Each pass - # takes one bounded batch, and the loop pauses between batches so a large - # redrive shares the database with delivery. # @rbs () -> void def redrive_loop: () -> void From a954bb156d9ea07a7f5aba51d6e885ff77bee178 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:03:46 -0700 Subject: [PATCH 09/15] fix: validate redrive filters and guard every close Reviewing the TypeScript port surfaced two defects this branch shares. `limit` and `failed_after` were passed through unchecked. A limit of zero or a fraction reached the database as a filter nobody had agreed to, and a value that does not answer `utc` raised a NoMethodError from inside the manager rather than refusing the argument. `close` updated a task by id alone, so a cancel could overwrite a task the runner had already completed and write a second transition event for it. Both closes guard on the running status now and write their event only when the update changed a row. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/dead_letter_scope.rb | 20 +++++++++-- lib/solid_objects/redrive_manager.rb | 12 ++++--- lib/solid_objects/redrive_runner.rb | 3 +- .../lib/solid_objects/dead_letter_scope.rbs | 6 ++++ .../lib/solid_objects/redrive_manager.rbs | 4 +-- test/integration/redrive_test.rb | 35 +++++++++++++++++++ 6 files changed, 70 insertions(+), 10 deletions(-) diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 9257db2..56ddd5e 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -55,13 +55,29 @@ def redrive(actor_type: nil, failed_after: nil, limit: nil, authorization_contex scope: self, filters: { "actor_type" => actor_type, - "failed_after" => failed_after&.utc&.iso8601(6), - "limit" => limit + "failed_after" => failed_after_filter(failed_after), + "limit" => limit_filter(limit) }, authorization_context: ) end + # @rbs (untyped) -> String? + def failed_after_filter(failed_after) + return nil if failed_after.nil? + raise ArgumentError, "failed_after must be a time" unless failed_after.respond_to?(:utc) + + failed_after.utc.iso8601(6) + end + + # @rbs (untyped) -> Integer? + def limit_filter(limit) + return nil if limit.nil? + return limit if limit.is_a?(Integer) && limit.positive? + + raise ArgumentError, "limit must be a positive integer" + end + # @rbs () -> ActiveRecord::Relation[untyped] def dead model.where(status: DEAD) diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb index 4987834..df94183 100644 --- a/lib/solid_objects/redrive_manager.rb +++ b/lib/solid_objects/redrive_manager.rb @@ -40,18 +40,20 @@ def cancel(id, authorization_context: nil) record = Redrive.find(id) return task_for(record) unless record.status == RUNNING - close(record, status: CANCELLED) - audit(record, action: "redrive.cancel") + audit(record, action: "redrive.cancel") if close(record, status: CANCELLED) task_for(record) end - # @rbs (Redrive, status: String) -> void + # @rbs (Redrive, status: String) -> bool def close(record, status:) - record.update!( + changed = Redrive.where(id: record.id, status: RUNNING).update_all( status:, active_scope: nil, - finished_at: SolidObjects.database_adapter.database_now + finished_at: SolidObjects.database_adapter.database_now, + updated_at: SolidObjects.database_adapter.database_now ) + record.reload if changed.positive? + changed.positive? end # @rbs (Redrive, action: String) -> void diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb index 3839ab6..28cd1cb 100644 --- a/lib/solid_objects/redrive_runner.rb +++ b/lib/solid_objects/redrive_runner.rb @@ -59,7 +59,8 @@ def batch_size(record) # @rbs (Redrive) -> void def finish(record) manager = SolidObjects.redrives - manager.close(record, status: RedriveManager::COMPLETED) + return unless manager.close(record, status: RedriveManager::COMPLETED) + manager.audit(record, action: "redrive.finish") end end diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index 3f65fd2..e0d5692 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -31,6 +31,12 @@ module SolidObjects # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask def redrive: (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask + # @rbs (untyped) -> String? + def failed_after_filter: (untyped) -> String? + + # @rbs (untyped) -> Integer? + def limit_filter: (untyped) -> Integer? + # @rbs () -> ActiveRecord::Relation[untyped] def dead: () -> ActiveRecord::Relation[untyped] diff --git a/sig/generated/lib/solid_objects/redrive_manager.rbs b/sig/generated/lib/solid_objects/redrive_manager.rbs index 85c5440..b20c0e8 100644 --- a/sig/generated/lib/solid_objects/redrive_manager.rbs +++ b/sig/generated/lib/solid_objects/redrive_manager.rbs @@ -20,8 +20,8 @@ module SolidObjects # @rbs (String, ?authorization_context: untyped) -> RedriveTask def cancel: (String, ?authorization_context: untyped) -> RedriveTask - # @rbs (Redrive, status: String) -> void - def close: (Redrive, status: String) -> void + # @rbs (Redrive, status: String) -> bool + def close: (Redrive, status: String) -> bool # @rbs (Redrive, action: String) -> void def audit: (Redrive, action: String) -> void diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb index 8e30da1..f8eed0a 100644 --- a/test/integration/redrive_test.rb +++ b/test/integration/redrive_test.rb @@ -212,6 +212,41 @@ def touch SolidObjects::AdministrationEvent.order(:id).map(&:action) end + test "refuses an invalid filter rather than redrive everything" do + dead_effects(2) + + assert_raises(ArgumentError) do + SolidObjects.dead_letters.effects.redrive(limit: 0, authorization_context: "operator") + end + assert_raises(ArgumentError) do + SolidObjects.dead_letters.effects.redrive(limit: 1.5, authorization_context: "operator") + end + + assert_equal 0, SolidObjects::Redrive.count + assert_equal 0, SolidObjects::AdministrationEvent.count + end + + test "writes no audit row when a retry names a row that does not exist" do + assert_raises(ActiveRecord::RecordNotFound) do + SolidObjects.dead_letters.effects.retry("missing", authorization_context: "operator") + end + + assert_equal 0, SolidObjects::AdministrationEvent.count + end + + test "a cancel cannot overwrite a task the runner already finished" do + dead_effects(1) + task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") + drain + + task.cancel(authorization_context: "operator") + + assert_equal "completed", + SolidObjects.redrives.find(task.id, authorization_context: "operator").status + assert_equal [ "redrive.start", "redrive.finish" ], + SolidObjects::AdministrationEvent.order(:id).map(&:action) + end + test "refuses an unauthorized caller that reaches the manager directly" do SolidObjects.configuration.authorize_administration = ->(**) { false } From 453862882e1abf7d85371d75b5fb2a32796f2a3c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:06:30 -0700 Subject: [PATCH 10/15] refactor: inline the one-use redrive helpers `active_scope_for` wrapped one interpolation and `batch_size` wrapped one calculation, each with a single caller. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/redrive_manager.rb | 7 +------ lib/solid_objects/redrive_runner.rb | 13 +++---------- sig/generated/lib/solid_objects/redrive_manager.rbs | 3 --- sig/generated/lib/solid_objects/redrive_runner.rbs | 3 --- 4 files changed, 4 insertions(+), 22 deletions(-) diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb index df94183..dd4cbe9 100644 --- a/lib/solid_objects/redrive_manager.rb +++ b/lib/solid_objects/redrive_manager.rb @@ -11,7 +11,7 @@ class RedriveManager # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask def start(scope:, filters:, authorization_context:) scope.authorize!(:redrive, authorization_context:) - active_scope = active_scope_for(kind: scope.kind, filters:) + active_scope = "#{scope.kind}:#{Digest::SHA256.hexdigest(filters.to_json)}" running = Redrive.find_by(active_scope:) return task_for(running) if running @@ -114,11 +114,6 @@ def remaining_for(record) [ matching, limit - record.moved ].min end - # @rbs (kind: String, filters: Hash[String, untyped]) -> String - def active_scope_for(kind:, filters:) - "#{kind}:#{Digest::SHA256.hexdigest(filters.to_json)}" - end - # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void def authorize!(action, authorization_context:, resource_id: nil) authorized = SolidObjects.configuration.authorize_administration.call( diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb index 28cd1cb..22f6977 100644 --- a/lib/solid_objects/redrive_runner.rb +++ b/lib/solid_objects/redrive_runner.rb @@ -32,7 +32,9 @@ def advance(record) # @rbs (Redrive) -> Integer def move_batch(record) scope = DeadLetterScope.for_kind(record.kind) - size = batch_size(record) + configured = SolidObjects.configuration.redrive_batch_size + limit = record.move_limit + size = limit ? [ configured, limit - record.moved ].min : configured return 0 unless size.positive? identifiers = scope @@ -47,15 +49,6 @@ def move_batch(record) revived end - # @rbs (Redrive) -> Integer - def batch_size(record) - configured = SolidObjects.configuration.redrive_batch_size - limit = record.move_limit - return configured unless limit - - [ configured, limit - record.moved ].min - end - # @rbs (Redrive) -> void def finish(record) manager = SolidObjects.redrives diff --git a/sig/generated/lib/solid_objects/redrive_manager.rbs b/sig/generated/lib/solid_objects/redrive_manager.rbs index b20c0e8..dc68c8a 100644 --- a/sig/generated/lib/solid_objects/redrive_manager.rbs +++ b/sig/generated/lib/solid_objects/redrive_manager.rbs @@ -37,9 +37,6 @@ module SolidObjects # @rbs (Redrive) -> Integer def remaining_for: (Redrive) -> Integer - # @rbs (kind: String, filters: Hash[String, untyped]) -> String - def active_scope_for: (kind: String, filters: Hash[String, untyped]) -> String - # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void def authorize!: (Symbol, authorization_context: untyped, ?resource_id: String?) -> void end diff --git a/sig/generated/lib/solid_objects/redrive_runner.rbs b/sig/generated/lib/solid_objects/redrive_runner.rbs index ae6e12b..d48e357 100644 --- a/sig/generated/lib/solid_objects/redrive_runner.rbs +++ b/sig/generated/lib/solid_objects/redrive_runner.rbs @@ -16,9 +16,6 @@ module SolidObjects # @rbs (Redrive) -> Integer def move_batch: (Redrive) -> Integer - # @rbs (Redrive) -> Integer - def batch_size: (Redrive) -> Integer - # @rbs (Redrive) -> void def finish: (Redrive) -> void end From 5f48a26bfdf2c24074bde0f104997dfdfa208df0 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:13:54 -0700 Subject: [PATCH 11/15] refactor: inline the redrive filter validation Two public helpers validated one argument each, with one caller and a generated signature apiece. They are guard clauses in `redrive` now. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/dead_letter_scope.rb | 27 +++++++------------ .../lib/solid_objects/dead_letter_scope.rbs | 6 ----- 2 files changed, 9 insertions(+), 24 deletions(-) diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 56ddd5e..7d56846 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -51,33 +51,24 @@ def retry(identifier_value, authorization_context: nil) # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask def redrive(actor_type: nil, failed_after: nil, limit: nil, authorization_context: nil) + if failed_after && !failed_after.respond_to?(:utc) + raise ArgumentError, "failed_after must be a time" + end + if limit && !(limit.is_a?(Integer) && limit.positive?) + raise ArgumentError, "limit must be a positive integer" + end + SolidObjects.redrives.start( scope: self, filters: { "actor_type" => actor_type, - "failed_after" => failed_after_filter(failed_after), - "limit" => limit_filter(limit) + "failed_after" => failed_after&.utc&.iso8601(6), + "limit" => limit }, authorization_context: ) end - # @rbs (untyped) -> String? - def failed_after_filter(failed_after) - return nil if failed_after.nil? - raise ArgumentError, "failed_after must be a time" unless failed_after.respond_to?(:utc) - - failed_after.utc.iso8601(6) - end - - # @rbs (untyped) -> Integer? - def limit_filter(limit) - return nil if limit.nil? - return limit if limit.is_a?(Integer) && limit.positive? - - raise ArgumentError, "limit must be a positive integer" - end - # @rbs () -> ActiveRecord::Relation[untyped] def dead model.where(status: DEAD) diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs index e0d5692..3f65fd2 100644 --- a/sig/generated/lib/solid_objects/dead_letter_scope.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs @@ -31,12 +31,6 @@ module SolidObjects # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask def redrive: (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask - # @rbs (untyped) -> String? - def failed_after_filter: (untyped) -> String? - - # @rbs (untyped) -> Integer? - def limit_filter: (untyped) -> Integer? - # @rbs () -> ActiveRecord::Relation[untyped] def dead: () -> ActiveRecord::Relation[untyped] From 783cacb3d031d5dc91b75fb44323c9e00b11339b Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:24:40 -0700 Subject: [PATCH 12/15] test: report what a running redrive has left to move The TypeScript suite covers this and the Ruby suite did not. Co-Authored-By: Claude Opus 5 (1M context) --- test/integration/redrive_test.rb | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb index f8eed0a..64a76cb 100644 --- a/test/integration/redrive_test.rb +++ b/test/integration/redrive_test.rb @@ -174,6 +174,17 @@ def touch assert_equal 0, SolidObjects.redrives.find(task.id, authorization_context: "operator").moved end + test "reports what a running task has left to move" do + dead_effects(25) + + task = SolidObjects.dead_letters.effects.redrive(limit: 15, authorization_context: "operator") + assert_equal 15, task.remaining + + SolidObjects::RedriveRunner.new.run_once + + assert_equal 5, SolidObjects.redrives.find(task.id, authorization_context: "operator").remaining + end + test "reads tasks back by id and by status" do dead_effects(5) task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator") From c358b0a18befab512f1cf052f96b4b7738842226 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:35:44 -0700 Subject: [PATCH 13/15] fix: write an audit row only when the retry happens The audit was written before the work, in its own transaction, so a retry that raised left a record of something that never happened. A message retry that could not enqueue, and a scope retry whose revive failed, both wrote one. Each event goes in the transaction that causes it now. The same defect was in the TypeScript port and is fixed there too, which is where it surfaced. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/dead_letter_manager.rb | 54 +++++++++++-------- lib/solid_objects/dead_letter_scope.rb | 23 ++++---- .../lib/solid_objects/dead_letter_manager.rbs | 3 ++ test/integration/administration_audit_test.rb | 34 ++++++++++++ 4 files changed, 81 insertions(+), 33 deletions(-) diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb index 8455309..f761d45 100644 --- a/lib/solid_objects/dead_letter_manager.rb +++ b/lib/solid_objects/dead_letter_manager.rb @@ -11,28 +11,18 @@ def all(authorization_context: nil) # @rbs (Integer, ?authorization_context: untyped) -> MessageReference def retry(dead_letter_id, authorization_context: nil) authorize!(:retry, authorization_context:, dead_letter_id:) - dead_letter = DeadLetter.find(dead_letter_id) - AdministrationAudit.record( - action: "dead_letter.retry", - kind: "message", - subject_id: dead_letter.id, - actor: AdministrationAudit.identity(authorization_context) - ) - return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) if dead_letter.retried_message_id - - original_message = dead_letter.message - message_reference = Mailbox.new.enqueue( - reference: Reference.new( - actor_type: dead_letter.actor_type, - actor_id: dead_letter.actor_id - ), - operation: dead_letter.operation, - arguments: dead_letter.arguments, - delivery_mode: original_message.delivery_mode, - idempotency_key: "dead-letter:#{dead_letter.id}" - ) - dead_letter.update!(retried_message_id: message_reference.id) - message_reference + actor = AdministrationAudit.identity(authorization_context) + SolidObjects.database_adapter.transaction do + dead_letter = DeadLetter.find(dead_letter_id) + reference = retried_reference(dead_letter) + AdministrationAudit.record( + action: "dead_letter.retry", + kind: "message", + subject_id: dead_letter.id, + actor: + ) + reference + end end # @rbs () -> DeadLetterScope @@ -57,6 +47,26 @@ def broadcasts private + # @rbs (DeadLetter) -> MessageReference + def retried_reference(dead_letter) + if dead_letter.retried_message_id + return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) + end + + message_reference = Mailbox.new.enqueue( + reference: Reference.new( + actor_type: dead_letter.actor_type, + actor_id: dead_letter.actor_id + ), + operation: dead_letter.operation, + arguments: dead_letter.arguments, + delivery_mode: dead_letter.message.delivery_mode, + idempotency_key: "dead-letter:#{dead_letter.id}" + ) + dead_letter.update!(retried_message_id: message_reference.id) + message_reference + end + # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void def authorize!(action, authorization_context:, dead_letter_id: nil) authorized = SolidObjects.configuration.authorize_administration.call( diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb index 7d56846..1cb9b5f 100644 --- a/lib/solid_objects/dead_letter_scope.rb +++ b/lib/solid_objects/dead_letter_scope.rb @@ -36,17 +36,18 @@ def all(authorization_context: nil) # @rbs (String, ?authorization_context: untyped) -> untyped def retry(identifier_value, authorization_context: nil) authorize!(:retry, authorization_context:, resource_id: identifier_value) - row = model.find_by!(identifier => identifier_value) - AdministrationAudit.record( - action: "dead_letter.retry", - kind: kind, - subject_id: identifier_value, - actor: AdministrationAudit.identity(authorization_context) - ) - return row unless row.status == DEAD - - revive(row) - row + actor = AdministrationAudit.identity(authorization_context) + SolidObjects.database_adapter.transaction do + row = model.find_by!(identifier => identifier_value) + revive(row) if row.status == DEAD + AdministrationAudit.record( + action: "dead_letter.retry", + kind: kind, + subject_id: identifier_value, + actor: + ) + row + end end # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask diff --git a/sig/generated/lib/solid_objects/dead_letter_manager.rbs b/sig/generated/lib/solid_objects/dead_letter_manager.rbs index 3b75450..563791f 100644 --- a/sig/generated/lib/solid_objects/dead_letter_manager.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_manager.rbs @@ -16,6 +16,9 @@ module SolidObjects private + # @rbs (DeadLetter) -> MessageReference + def retried_reference: (DeadLetter) -> MessageReference + # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void def authorize!: (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void end diff --git a/test/integration/administration_audit_test.rb b/test/integration/administration_audit_test.rb index 612a3d4..58bdebf 100644 --- a/test/integration/administration_audit_test.rb +++ b/test/integration/administration_audit_test.rb @@ -100,6 +100,32 @@ def run assert_equal "user:42", SolidObjects::AdministrationEvent.sole.actor end + test "writes no audit row when the retry itself fails" do + effect = dead_effect + + with_failing_method(SolidObjects::DeadLetterScope, :revive) do + assert_raises(RuntimeError) do + SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator") + end + end + + assert_equal 0, SolidObjects::AdministrationEvent.count + end + + test "writes no audit row when a message retry fails to enqueue" do + PoisonActor.ref("one").async.run + run_actors + dead_letter = SolidObjects::DeadLetter.sole + + with_failing_method(SolidObjects::Mailbox, :enqueue) do + assert_raises(RuntimeError) do + SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator") + end + end + + assert_equal 0, SolidObjects::AdministrationEvent.count + end + test "writes no audit row when the caller is refused" do effect = dead_effect SolidObjects.configuration.authorize_administration = ->(**) { false } @@ -121,6 +147,14 @@ def run private + def with_failing_method(target, name) + original = target.instance_method(name) + target.define_method(name) { |*, **| raise "injected failure" } + yield + ensure + target.define_method(name, original) + end + def dead_effect LedgerActor.ref("one").async.post run_actors From f6793263f16f7a7416034a74efbcdd99746b4ce5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:44:45 -0700 Subject: [PATCH 14/15] refactor: inline the retried-reference branch `retried_reference` wrapped one branch with one caller. `retry` shows the branch and `enqueue_retry` keeps only the enqueue, which the branch needs a name for. Co-Authored-By: Claude Opus 5 (1M context) --- lib/solid_objects/dead_letter_manager.rb | 12 ++++++------ .../lib/solid_objects/dead_letter_manager.rbs | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb index f761d45..832645b 100644 --- a/lib/solid_objects/dead_letter_manager.rb +++ b/lib/solid_objects/dead_letter_manager.rb @@ -14,7 +14,11 @@ def retry(dead_letter_id, authorization_context: nil) actor = AdministrationAudit.identity(authorization_context) SolidObjects.database_adapter.transaction do dead_letter = DeadLetter.find(dead_letter_id) - reference = retried_reference(dead_letter) + reference = if dead_letter.retried_message_id + MessageReference.from_message(Message.find(dead_letter.retried_message_id)) + else + enqueue_retry(dead_letter) + end AdministrationAudit.record( action: "dead_letter.retry", kind: "message", @@ -48,11 +52,7 @@ def broadcasts private # @rbs (DeadLetter) -> MessageReference - def retried_reference(dead_letter) - if dead_letter.retried_message_id - return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) - end - + def enqueue_retry(dead_letter) message_reference = Mailbox.new.enqueue( reference: Reference.new( actor_type: dead_letter.actor_type, diff --git a/sig/generated/lib/solid_objects/dead_letter_manager.rbs b/sig/generated/lib/solid_objects/dead_letter_manager.rbs index 563791f..856fe79 100644 --- a/sig/generated/lib/solid_objects/dead_letter_manager.rbs +++ b/sig/generated/lib/solid_objects/dead_letter_manager.rbs @@ -17,7 +17,7 @@ module SolidObjects private # @rbs (DeadLetter) -> MessageReference - def retried_reference: (DeadLetter) -> MessageReference + def enqueue_retry: (DeadLetter) -> MessageReference # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void def authorize!: (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void From 92bc474e9415c2db4307ebe7ea622ca35c027ec7 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 22 Sep 2026 14:48:11 -0700 Subject: [PATCH 15/15] docs: state what an administration event records An event records an authorized press, not a state transition, and the redrive transitions are the opposite. The rule was implicit in the tests. Co-Authored-By: Claude Opus 5 (1M context) --- docs/operations.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/docs/operations.md b/docs/operations.md index 0f40472..f11563f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -357,7 +357,15 @@ their own resource name: `dead_letters`, `effect_dead_letters`, writes one row to `solid_objects_administration_events`, holding the action, the kind, the subject, the identity, and when it happened. The identity comes from `administration_identity`, which receives the authorization context the caller -passed and defaults to its `to_s`. A refused caller writes nothing. +passed and defaults to its `to_s`. + +An event records an authorized press, not a state transition. Pressing retry +twice writes two rows, because an operator did two things and a log that shows +one cannot answer who pressed what. The row the event names carries the outcome. +A refused caller writes nothing, and a retry that raises after the lookup writes +nothing, because the event shares the transaction with the work. The redrive +transitions are different: `redrive.start`, `redrive.finish`, and +`redrive.cancel` are written only when the task actually changes. Automatic redrive on a schedule is deliberately absent. A dead row means a person decided something, and these APIs give that person an alternative to an