fix: find recovery candidates through effects

CHANGELOG.md

@@ -2,6 +2,19 @@
 
 ## Unreleased
 
+- Find SQLite effect recovery candidates through the processing 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 recovery row in index
+  order, on each effect poll. On SQLite the status test now carries
+  `likelihood(..., 0.000001)`, so the plan starts from the `effects_poll`
+  index. The PostgreSQL and MySQL queries do not change. Ruby 0.16.1 has the
+  same fix.
+- Ruby 0.16.1 also fixes the SQLite join order of its claimed-message scan. The
+  JavaScript claim path has no such scan, and SQLite already starts
+  `recoverExpiredClaims` from the claimed messages, so that fix needs no
+  JavaScript change.
+
 - Build before `npm publish` reads the manifest. npm validates `bin` against the
   working tree before `prepack` produces `dist`, so every release logged
   `No bin file found at dist/executable.js` twice. The published package was

docs/parity.md

@@ -310,3 +310,8 @@ checks. Per-effect recovery timeouts extend the runtime heartbeat threshold
 (milliseconds in JS, seconds in Ruby). Cloudflare returns emit handles but rejects
 process-heartbeat recovery options and intents before commit. See
 [the transaction protocol](effect-recovery.md).
+
+On SQLite, both runtimes mark the `status = 'processing'` test in the recovery
+candidate query with `likelihood()`. The candidate scan then starts from the
+effects poll index, also when `sqlite_stat1` shows that most effects have one
+status.

src/effect-recovery-coordinator.ts

@@ -40,12 +40,16 @@ export class EffectRecoveryCoordinator {
     const candidates = await this.options.settings.database.connection(async (connection) => {
       const now = await connection.nowMilliseconds()
       const threshold = this.options.settings.processAliveThresholdMilliseconds
+      const processing =
+        this.options.settings.database.family === "sqlite"
+          ? "likelihood(effects.status = 'processing', 0.000001)"
+          : "effects.status = 'processing'"
       return connection.all<RecoveryBinding>(
         `SELECT recoveries.* FROM ${this.table("effect_recoveries")} recoveries
        JOIN ${this.table("effects")} effects ON effects.id = recoveries.effect_id
        LEFT JOIN ${this.table("processes")} owners ON owners.id = effects.claimed_by
        WHERE recoveries.retired_at_ms IS NULL AND recoveries.recovery_operation IS NOT NULL
-         AND effects.status = 'processing'
+         AND ${processing}
          AND (owners.id IS NULL OR owners.heartbeat_at_ms <= ? -
            CASE WHEN recoveries.recovery_timeout_ms > ? THEN recoveries.recovery_timeout_ms ELSE ? END)
          ORDER BY recoveries.instance_id, recoveries.effect_id LIMIT ?`,

test/polling-queries.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, it } from "vitest"
-import type { Database, DatabaseConnection } from "../src/database/types.js"
+import type { Database, DatabaseConnection, DatabaseFamily } from "../src/database/types.js"
 import { sqlite } from "../src/database/sqlite.js"
 import { configure, type SolidObjectsRuntime } from "../src/runtime.js"
 
@@ -59,8 +59,119 @@ describe("polling queries", () => {
     )
     expect(availableReminder).not.toMatch(/reminders\.claimed_by IS NOT NULL/)
   })
+
+  it("finds SQLite recovery candidates through processing effects when most effects are complete", async () => {
+    const database = sqlite({ path: ":memory:" })
+    const queries: RecordedQuery[] = []
+    runtime = configuredRuntime(new RecordingQueryDatabase(database, queries))
+    await runtime.install()
+    await database.transaction(async (connection) => {
+      const now = await connection.nowMilliseconds()
+      await connection.run(
+        `INSERT INTO solid_objects_instances(id, actor_type, actor_id, state, state_version, created_at_ms, updated_at_ms)
+         VALUES ('instance', 'Probe', 'one', '{}', 1, ?, ?)`,
+        [now, now],
+      )
+      await connection.run(
+        `INSERT INTO solid_objects_messages(id, request_id, instance_id, actor_type, actor_id, sequence, operation,
+           delivery_mode, arguments, max_attempts, created_at_ms, updated_at_ms)
+         VALUES ('message', 'request', 'instance', 'Probe', 'one', 1, 'run', 'async', '{}', 3, ?, ?)`,
+        [now, now],
+      )
+      for (let index = 0; index < 3_000; index++) {
+        const effectId = `effect-${String(index).padStart(4, "0")}`
+        await connection.run(
+          `INSERT INTO solid_objects_effects(id, message_id, instance_id, name, arguments, status, max_attempts, available_at_ms)
+           VALUES (?, 'message', 'instance', 'work', '{}', 'completed', 3, ?)`,
+          [effectId, now],
+        )
+        if (index < 2_700) continue
+        await connection.run(
+          `INSERT INTO solid_objects_effect_recoveries(effect_id, instance_id, recovery_operation, status_operation)
+           VALUES (?, 'instance', 'recover', 'inspect')`,
+          [effectId],
+        )
+      }
+    })
+    await database.connection((connection) => connection.run("ANALYZE"))
+    const pollStatistics = await database.connection((connection) =>
+      connection.get<{ stat: string }>(
+        "SELECT stat FROM sqlite_stat1 WHERE idx = 'solid_objects_effects_poll'",
+      ),
+    )
+    expect(pollStatistics?.stat.split(" ")[1]).toBe("3000")
+
+    await runtime.repository.claimEffect("effect-worker")
+
+    const candidates = queries.find((query) =>
+      query.sql.includes("FROM solid_objects_effect_recoveries recoveries"),
+    )
+    expect(candidates).toBeDefined()
+    const plan = await database.connection((connection) =>
+      connection.all<{ detail: string }>(
+        `EXPLAIN QUERY PLAN ${candidates!.sql}`,
+        candidates!.parameters,
+      ),
+    )
+    const steps = plan.map((row) => row.detail)
+    expect(steps[0]).toMatch(/^SEARCH effects USING INDEX solid_objects_effects_poll \(status=\?\)/)
+    expect(steps.filter((step) => step.startsWith("SCAN "))).toEqual([])
+  })
 })
 
+type QueryParameters = Parameters<DatabaseConnection["all"]>[1]
+
+interface RecordedQuery {
+  sql: string
+  parameters: QueryParameters
+}
+
+class RecordingQueryDatabase implements Database {
+  readonly family: DatabaseFamily
+  readonly schemaIdentity: string
+
+  constructor(
+    private readonly database: Database,
+    private readonly queries: RecordedQuery[],
+  ) {
+    this.family = database.family
+    this.schemaIdentity = database.schemaIdentity
+  }
+
+  connection<Result>(
+    callback: (connection: DatabaseConnection) => Promise<Result>,
+  ): Promise<Result> {
+    return this.database.connection((connection) => callback(this.recordingConnection(connection)))
+  }
+
+  transaction<Result>(
+    callback: (connection: DatabaseConnection) => Promise<Result>,
+  ): Promise<Result> {
+    return this.database.transaction((connection) => callback(this.recordingConnection(connection)))
+  }
+
+  transactionActive(): boolean {
+    return this.database.transactionActive?.() ?? false
+  }
+
+  close(): Promise<void> {
+    return this.database.close()
+  }
+
+  private recordingConnection(connection: DatabaseConnection): DatabaseConnection {
+    return {
+      run: (sql, parameters) => connection.run(sql, parameters),
+      get: <Row extends object>(sql: string, parameters?: QueryParameters) =>
+        connection.get<Row>(sql, parameters),
+      all: <Row extends object>(sql: string, parameters?: QueryParameters) => {
+        this.queries.push({ sql, parameters })
+        return connection.all<Row>(sql, parameters)
+      },
+      nowMilliseconds: () => connection.nowMilliseconds(),
+    }
+  }
+}
+
 class RecordingPostgreSQLDatabase implements Database {
   readonly family = "postgresql" as const
   readonly schemaIdentity: string
@@ -95,11 +206,11 @@ class RecordingPostgreSQLDatabase implements Database {
   private recordingConnection(connection: DatabaseConnection): DatabaseConnection {
     return {
       run: (sql, parameters) => connection.run(sql, parameters),
-      get: <Row extends object>(sql: string, parameters?: readonly unknown[]) => {
+      get: <Row extends object>(sql: string, parameters?: QueryParameters) => {
         this.statements.push(sql.replace(/\s+/g, " ").trim())
         return connection.get<Row>(sql.replace(/\s+FOR UPDATE SKIP LOCKED\s*$/i, ""), parameters)
       },
-      all: <Row extends object>(sql: string, parameters?: readonly unknown[]) =>
+      all: <Row extends object>(sql: string, parameters?: QueryParameters) =>
         connection.all<Row>(sql, parameters),
       nowMilliseconds: () => connection.nowMilliseconds(),
     }