feat: retry and redrive dead effects and broadcasts

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 ## 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. 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.
+- 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

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

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/20260922000000_add_solid_objects_administration_events.rb

@@ -0,0 +1,26 @@
+# 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
+      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
+
+      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_type
+    connection.adapter_name.match?(/postgres/i) ? :jsonb : :json
+  end
+end

db/migrate/20260922000001_add_solid_objects_redrives.rb

@@ -0,0 +1,32 @@
+# 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
+      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
+      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_type
+    connection.adapter_name.match?(/postgres/i) ? :jsonb : :json
+  end
+end

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:
 

docs/operations.md

@@ -282,6 +282,95 @@ 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 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
+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`.
+
+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
+`UPDATE` against a runtime table.
+
 ## Graceful shutdown
 
 The supervisor requests shutdown, stops new claims, lets active loops return,

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

lib/solid_objects.rb

@@ -27,6 +27,11 @@
 require "solid_objects/actor"
 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"
 require "solid_objects/instance_pruner"
@@ -160,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
@@ -192,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

@@ -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]?, ?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:,
+        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

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
@@ -49,6 +51,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,
@@ -80,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,
@@ -98,6 +103,7 @@ class Configuration
       :authorize_subscription,
       :authorize_administration,
       :authorize_transmission,
+      :administration_identity,
       :transmission_actor_type_resolver
 
     # @rbs @additional_components: Array[untyped]
@@ -134,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
@@ -156,6 +164,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
@@ -226,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
@@ -247,6 +259,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
@@ -295,7 +310,8 @@ def positive_values
         shutdown_timeout:,
         message_retention:,
         process_retention:,
-        prune_batch_size:
+        prune_batch_size:,
+        redrive_batch_size:
       }
     end