From e6bfd8b346715f635feba33bffc0c4c3d2637b00 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Fri, 2 Oct 2026 10:05:12 -0700 Subject: [PATCH 1/2] 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 through the instance index on each effect poll. On generated data with 100,000 completed effects and 10,000 recovery rows, the query took 6.3 ms. On SQLite the status test now carries likelihood(..., 0.000001), which overrides the statistics estimate for that term. The plan searches the effects poll index first and takes 0.015 ms on the same data. The PostgreSQL and MySQL SQL does not change. Ruby 0.16.1 has the same fix. Ruby 0.16.1 also fixes the join order of its claimed-message scan. JavaScript has no such scan, and SQLite already starts recoverExpiredClaims from the claimed messages, so that fix needs no change here. --- CHANGELOG.md | 13 ++++ docs/parity.md | 5 ++ src/effect-recovery-coordinator.ts | 7 +- test/polling-queries.test.ts | 111 ++++++++++++++++++++++++++++- 4 files changed, 134 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d79a993..55393f5 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/parity.md b/docs/parity.md index 176c1fa..08828d6 100644 --- a/docs/parity.md +++ b/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. diff --git a/src/effect-recovery-coordinator.ts b/src/effect-recovery-coordinator.ts index 8146f23..aa93b78 100644 --- a/src/effect-recovery-coordinator.ts +++ b/src/effect-recovery-coordinator.ts @@ -45,7 +45,7 @@ export class EffectRecoveryCoordinator { 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 ${this.processingCondition()} 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 ?`, @@ -250,4 +250,9 @@ export class EffectRecoveryCoordinator { private lockClause(): string { return this.options.settings.database.family === "sqlite" ? "" : " FOR UPDATE" } + private processingCondition(): string { + return this.options.settings.database.family === "sqlite" + ? "likelihood(effects.status = 'processing', 0.000001)" + : "effects.status = 'processing'" + } } diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts index 0a860cc..14cd998 100644 --- a/test/polling-queries.test.ts +++ b/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,117 @@ 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([]) + }) }) +interface RecordedQuery { + sql: string + parameters: readonly unknown[] | undefined +} + +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( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.connection((connection) => callback(this.recordingConnection(connection))) + } + + transaction( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.transaction((connection) => callback(this.recordingConnection(connection))) + } + + transactionActive(): boolean { + return this.database.transactionActive?.() ?? false + } + + close(): Promise { + return this.database.close() + } + + private recordingConnection(connection: DatabaseConnection): DatabaseConnection { + return { + run: (sql, parameters) => connection.run(sql, parameters), + get: (sql: string, parameters?: readonly unknown[]) => + connection.get(sql, parameters), + all: (sql: string, parameters?: readonly unknown[]) => { + this.queries.push({ sql, parameters }) + return connection.all(sql, parameters) + }, + nowMilliseconds: () => connection.nowMilliseconds(), + } + } +} + class RecordingPostgreSQLDatabase implements Database { readonly family = "postgresql" as const readonly schemaIdentity: string From d97b3606f506ecdf4f8927615b2d43561606844d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Fri, 2 Oct 2026 10:18:27 -0700 Subject: [PATCH 2/2] refactor: keep the SQLite hint beside the query recoverAvailable calls processingCondition() once, so the method only moved the SQLite-specific SQL away from the query that uses it. The condition is now a local value next to the query. The recording databases in polling-queries.test.ts typed SQL parameters as unknown. The repository rule disallows unknown in type annotations, so they now use the parameter type that DatabaseConnection declares. --- src/effect-recovery-coordinator.ts | 11 +++++------ test/polling-queries.test.ts | 12 +++++++----- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/effect-recovery-coordinator.ts b/src/effect-recovery-coordinator.ts index aa93b78..b4dd500 100644 --- a/src/effect-recovery-coordinator.ts +++ b/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( `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 ${this.processingCondition()} + 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 ?`, @@ -250,9 +254,4 @@ export class EffectRecoveryCoordinator { private lockClause(): string { return this.options.settings.database.family === "sqlite" ? "" : " FOR UPDATE" } - private processingCondition(): string { - return this.options.settings.database.family === "sqlite" - ? "likelihood(effects.status = 'processing', 0.000001)" - : "effects.status = 'processing'" - } } diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts index 14cd998..9d9c6db 100644 --- a/test/polling-queries.test.ts +++ b/test/polling-queries.test.ts @@ -119,9 +119,11 @@ describe("polling queries", () => { }) }) +type QueryParameters = Parameters[1] + interface RecordedQuery { sql: string - parameters: readonly unknown[] | undefined + parameters: QueryParameters } class RecordingQueryDatabase implements Database { @@ -159,9 +161,9 @@ class RecordingQueryDatabase implements Database { private recordingConnection(connection: DatabaseConnection): DatabaseConnection { return { run: (sql, parameters) => connection.run(sql, parameters), - get: (sql: string, parameters?: readonly unknown[]) => + get: (sql: string, parameters?: QueryParameters) => connection.get(sql, parameters), - all: (sql: string, parameters?: readonly unknown[]) => { + all: (sql: string, parameters?: QueryParameters) => { this.queries.push({ sql, parameters }) return connection.all(sql, parameters) }, @@ -204,11 +206,11 @@ class RecordingPostgreSQLDatabase implements Database { private recordingConnection(connection: DatabaseConnection): DatabaseConnection { return { run: (sql, parameters) => connection.run(sql, parameters), - get: (sql: string, parameters?: readonly unknown[]) => { + get: (sql: string, parameters?: QueryParameters) => { this.statements.push(sql.replace(/\s+/g, " ").trim()) return connection.get(sql.replace(/\s+FOR UPDATE SKIP LOCKED\s*$/i, ""), parameters) }, - all: (sql: string, parameters?: readonly unknown[]) => + all: (sql: string, parameters?: QueryParameters) => connection.all(sql, parameters), nowMilliseconds: () => connection.nowMilliseconds(), }