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..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 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 ?`, diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts index 0a860cc..9d9c6db 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,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[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( + 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?: QueryParameters) => + connection.get(sql, parameters), + all: (sql: string, parameters?: QueryParameters) => { + 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 @@ -95,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(), }