fix: find recovery candidates through effects

sqlite_stat1 records only the average row count for each effect
status. When most effects are complete, SQLite estimated that
status = 'processing' matched most of the effects table. It then read
every effect_recoveries row in primary key order to skip the ORDER BY
sort, and looked up each effect. A production app saw 5.75 ms per call
on SQLite against 0.68 ms on MySQL.

This is not a missing index. Only a recovery retirement writes
retired_at, so a recovery row keeps it empty after a normal
completion, and the recovery filter matches almost every row.

CROSS JOIN alone does not fix it: when every effect has one status,
SQLite scans the whole effects table. unlikely() is too weak at
500,000 effects. On SQLite the status test now carries
likelihood(..., 0.000001), which overrides the statistics estimate for
that term, so the plan searches idx_so_effects_poll first. The
PostgreSQL and MySQL SQL does not change.

Author: cardmagic

lib/solid_objects/effect_recovery_coordinator.rb

@@ -63,17 +63,22 @@ def recovery_candidates
       effects = Effect.table_name
       owners = Process.table_name
       bindings = EffectRecovery.table_name
-      heartbeat = case DatabaseAdapter.family(Record.connection)
+      family = DatabaseAdapter.family(Record.connection)
+      heartbeat = case family
       when :postgresql then "EXTRACT(EPOCH FROM #{owners}.last_heartbeat_at)"
       when :mysql then "UNIX_TIMESTAMP(#{owners}.last_heartbeat_at)"
       else "CAST(STRFTIME('%s', #{owners}.last_heartbeat_at) AS REAL)"
       end
+      processing = case family
+      when :postgresql, :mysql then "#{effects}.status = ?"
+      else "likelihood(#{effects}.status = ?, 0.000001)"
+      end
       now = SolidObjects.database_adapter.database_clock_now.to_f
       threshold = SolidObjects.configuration.process_alive_threshold
       EffectRecovery.joins("INNER JOIN #{effects} ON #{effects}.effect_id = #{bindings}.effect_id")
         .joins("LEFT JOIN #{owners} ON #{owners}.id = #{effects}.claimed_by")
         .where(retired_at: nil).where.not(recovery_operation: nil)
-        .where("#{effects}.status = ?", "processing")
+        .where(processing, "processing")
         .where("#{owners}.id IS NULL OR #{heartbeat} <= ? - CASE WHEN #{bindings}.recovery_timeout > ? THEN #{bindings}.recovery_timeout ELSE ? END", now, threshold, threshold)
         .order(:effect_id).limit(SolidObjects.configuration.claim_scan_limit)
     end

test/integration/effect_recovery_candidates_test.rb

@@ -0,0 +1,111 @@
+# frozen_string_literal: true
+
+require "database_test_helper"
+
+class EffectRecoveryCandidatesTest < ActiveSupport::TestCase
+  class CandidateActor < SolidObjects::Actor
+    actor_type "effect-recovery-candidates"
+
+    def run
+    end
+  end
+
+  setup do
+    @message = SolidObjects::Message.find(CandidateActor.ref("one").async.run.id)
+  end
+
+  test "recovery candidates are abandoned processing effects with a recovery operation in effect order" do
+    now = SolidObjects.database_adapter.database_clock_now
+    threshold = SolidObjects.configuration.process_alive_threshold
+    live_owner = create_process(last_heartbeat_at: now)
+    stale_owner = create_process(last_heartbeat_at: now - (threshold * 3))
+    unowned = create_effect("00000000-0000-4000-8000-000000000009", status: "processing")
+    create_effect("00000000-0000-4000-8000-000000000001", status: "processing", owner: live_owner)
+    abandoned = create_effect("00000000-0000-4000-8000-000000000002", status: "processing", owner: stale_owner)
+    create_effect("00000000-0000-4000-8000-000000000003", status: "processing", owner: stale_owner, recovery_timeout: threshold * 10)
+    create_effect("00000000-0000-4000-8000-000000000004", status: "processing", retired_at: now)
+    create_effect("00000000-0000-4000-8000-000000000005", status: "processing", recovery_operation: nil)
+    create_effect("00000000-0000-4000-8000-000000000006", status: "pending")
+    create_effect("00000000-0000-4000-8000-000000000007", status: "completed")
+
+    assert_equal [ abandoned, unowned ], recovery_candidates.map(&:effect_id)
+  end
+
+  test "recovery candidates stop at the claim scan limit" do
+    SolidObjects.configuration.claim_scan_limit = 1
+    create_effect("00000000-0000-4000-8000-000000000002", status: "processing")
+    first = create_effect("00000000-0000-4000-8000-000000000001", status: "processing")
+
+    assert_equal [ first ], recovery_candidates.map(&:effect_id)
+  end
+
+  test "SQLite finds recovery candidates through processing effects when most effects are complete" do
+    skip "requires a SQLite query plan" unless database_family == :sqlite
+
+    now = Time.current
+    effect_ids = Array.new(3_000) { SecureRandom.uuid }
+    SolidObjects::Effect.insert_all!(effect_ids.map { |effect_id|
+      { message_id: @message.id, instance_id: @message.instance_id, effect_id:, name: "work", arguments: {},
+        status: "completed", max_attempts: 3, available_at: now, completed_at: now, created_at: now, updated_at: now }
+    })
+    SolidObjects::EffectRecovery.insert_all!(effect_ids.last(300).map { |effect_id|
+      { effect_id:, instance_id: @message.instance_id, recovery_operation: "recover", status_operation: "status",
+        created_at: now, updated_at: now }
+    })
+    connection = SolidObjects::Record.connection
+    connection.execute("ANALYZE")
+    poll_statistics = connection.select_value("SELECT stat FROM sqlite_stat1 WHERE idx = 'idx_so_effects_poll'")
+
+    assert_equal 3_000, poll_statistics.split[1].to_i
+
+    plan = connection.select_all("EXPLAIN QUERY PLAN #{recovery_candidates.to_sql}").map { |row| row["detail"] }
+
+    assert_match(/\ASEARCH #{SolidObjects::Effect.table_name} USING INDEX idx_so_effects_poll \(status=\?\)/, plan.first, plan.join("\n"))
+    assert plan.none? { |step| step.start_with?("SCAN ") }, plan.join("\n")
+  ensure
+    connection&.execute("DELETE FROM sqlite_stat1")
+    connection&.execute("ANALYZE sqlite_schema")
+  end
+
+  private
+
+  def recovery_candidates
+    SolidObjects::EffectRecoveryCoordinator.new.send(:recovery_candidates)
+  end
+
+  def create_effect(effect_id, status:, owner: nil, recovery_operation: "recover", recovery_timeout: nil, retired_at: nil)
+    SolidObjects::Effect.create!(
+      message: @message,
+      instance: @message.instance,
+      effect_id:,
+      name: "work",
+      arguments: {},
+      status:,
+      max_attempts: 3,
+      available_at: Time.current,
+      claimed_by: owner&.id,
+      claimed_at: owner && Time.current
+    )
+    SolidObjects::EffectRecovery.create!(
+      effect_id:,
+      instance: @message.instance,
+      recovery_operation:,
+      status_operation: "status",
+      recovery_timeout:,
+      retired_at:
+    )
+    effect_id
+  end
+
+  def create_process(last_heartbeat_at:)
+    SolidObjects::Process.create!(
+      id: SecureRandom.uuid,
+      kind: "effect",
+      hostname: "test-host",
+      pid: ::Process.pid,
+      started_at: last_heartbeat_at,
+      last_heartbeat_at:,
+      metadata: {}
+    )
+  end
+end