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

Author: cardmagic

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)

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

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

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]
 

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

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 }