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) <noreply@anthropic.com>

Author: cardmagic

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

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

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
 

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

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
 

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
 

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,19 +41,48 @@ 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
 
       revive(row)
       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

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

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