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

Author: cardmagic

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

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
 

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
 

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)

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

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,15 +273,19 @@ 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
 
   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