diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 87d466e..3c7564f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,16 +7,6 @@ on: branches: [main] workflow_dispatch: -# TEMPORARY (backlog merge): the type check, tests and build are set to -# continue-on-error so a red result cannot block merging the open contributor -# PRs. The steps still run and their output is still in the logs — only the -# gate is lifted. -# -# The workflow is deliberately NOT deleted: if a branch protection rule requires -# the "build" check, removing the workflow means the check never reports at all -# and merges stay blocked permanently. Keeping it green is what actually unblocks. -# -# TO RESTORE: delete the four `continue-on-error: true` lines below. jobs: build: runs-on: ubuntu-latest @@ -33,22 +23,10 @@ jobs: run: npm ci - name: Type check - continue-on-error: true run: npx tsc --noEmit - name: Test - continue-on-error: true run: npm test - name: Build - continue-on-error: true run: npm run build - - # Reports the real status without failing the job, so the gate stays open - # while the actual state of the branch is still visible at a glance. - - name: Report status - continue-on-error: true - run: | - echo "::notice::CI gating is temporarily disabled for the backlog merge." - npx tsc --noEmit 2>&1 | grep -c "error TS" \ - | xargs -I{} echo "::notice::tsc reports {} error(s)" diff --git a/__tests__/database-writer-pool-migration-hooks.test.ts b/__tests__/database-writer-pool-migration-hooks.test.ts index cdeebe9..e77b4af 100644 --- a/__tests__/database-writer-pool-migration-hooks.test.ts +++ b/__tests__/database-writer-pool-migration-hooks.test.ts @@ -1,5 +1,10 @@ import Database from "better-sqlite3"; -import { setDb, runMigrations, closeDb } from "../src/indexer/db.js"; +import { + setDb, + runMigrations, + closeDb, + getShippedMigrationVersions, +} from "../src/indexer/db.js"; import { WriterPoolSchemaError, assertWriterPoolSchemaReady, @@ -63,7 +68,10 @@ describe("database_writer_pool – migration verification hooks (#331)", () => { const report = verifyWriterPoolSchema(); expect(report.valid).toBe(false); - expect(report.missingVersions).toEqual([5, 6, 7]); + // Derived from the shipped list rather than hardcoded, so adding a + // migration does not require editing this expectation. + const expectedMissing = getShippedMigrationVersions().filter((v) => v >= 5); + expect(report.missingVersions).toEqual(expectedMissing); expect(report.issues.join(" ")).toContain("out of sync"); }); diff --git a/__tests__/duplicate-prevention-index.test.ts b/__tests__/duplicate-prevention-index.test.ts index 3aa233e..bd06a87 100644 --- a/__tests__/duplicate-prevention-index.test.ts +++ b/__tests__/duplicate-prevention-index.test.ts @@ -59,11 +59,12 @@ describe("SQLite index optimization — duplicate_prevention lookups", () => { .prepare("SELECT name, \"unique\" FROM pragma_index_list('events')") .all() as Array<{ name: string; unique: number }>; - // Only the auto-index backing the UNIQUE constraint should exist - no - // speculative index was added for this ticket. - expect(indexes).toHaveLength(1); - expect(indexes[0].name).toBe("sqlite_autoindex_events_1"); - expect(indexes[0].unique).toBe(1); + // Later migrations added non-unique lookup indexes to `events`, so this + // asserts what the ticket actually cares about: the UNIQUE constraint's + // auto-index is still the one and only unique index on the table. + const uniqueIndexes = indexes.filter((i) => i.unique === 1); + expect(uniqueIndexes).toHaveLength(1); + expect(uniqueIndexes[0].name).toBe("sqlite_autoindex_events_1"); const columns = testDb .prepare("SELECT name FROM pragma_index_info('sqlite_autoindex_events_1') ORDER BY seqno") @@ -108,7 +109,11 @@ describe("SQLite index optimization — duplicate_prevention lookups", () => { .all("C1", 10, 0) as QueryPlanRow[]; const detail = plan.map((row) => row.detail).join(" | "); - expect(detail).toContain("USING INDEX sqlite_autoindex_events_1"); + // Later migrations gave the planner a better-suited composite index for + // this predicate, so assert the property that matters — the lookup is + // index-backed rather than a full table scan — instead of naming one. + expect(detail).toMatch(/USING (COVERING )?INDEX/); + expect(detail).not.toMatch(/SCAN events\b/); }); }); diff --git a/__tests__/event-type-filter.test.ts b/__tests__/event-type-filter.test.ts index e86064d..947ae02 100644 --- a/__tests__/event-type-filter.test.ts +++ b/__tests__/event-type-filter.test.ts @@ -34,18 +34,31 @@ function makeSuccessResult( }; } +/** + * Stand-in for the slice of `Server` that `fetchEventsWithRetry` needs. + * + * A bare `jest.fn()` infers `Mock`, which makes + * `mockResolvedValue` expect `never` and leaves the object unassignable to the + * `Pick` parameter. Typing the mock explicitly keeps both + * the call sites and the `.mock.calls` assertions well typed. + */ +type MockEventsFn = jest.Mock<(params: GetEventsParams) => Promise>; +type MockServer = Parameters[0] & { + getEvents: MockEventsFn; +}; + /** Build a mock `Server.getEvents` that returns `result`. */ -function makeSuccessServer(result: RpcGetEventsResult = makeSuccessResult()) { - return { - getEvents: jest.fn().mockResolvedValue(result), - }; +function makeSuccessServer(result: RpcGetEventsResult = makeSuccessResult()): MockServer { + const getEvents = jest.fn() as MockEventsFn; + getEvents.mockResolvedValue(result); + return { getEvents } as unknown as MockServer; } /** Build a mock `Server.getEvents` that always throws `err`. */ -function makeFailingServer(err: Error) { - return { - getEvents: jest.fn().mockRejectedValue(err), - }; +function makeFailingServer(err: Error): MockServer { + const getEvents = jest.fn() as MockEventsFn; + getEvents.mockRejectedValue(err); + return { getEvents } as unknown as MockServer; } /** @@ -58,13 +71,13 @@ function makePartialFailServer( successResult: RpcGetEventsResult = makeSuccessResult(1) ) { let calls = 0; - return { - getEvents: jest.fn().mockImplementation(() => { - calls += 1; - if (calls <= failCount) return Promise.reject(err); - return Promise.resolve(successResult); - }), - }; + const getEvents = jest.fn() as MockEventsFn; + getEvents.mockImplementation(() => { + calls += 1; + if (calls <= failCount) return Promise.reject(err); + return Promise.resolve(successResult); + }); + return { getEvents } as unknown as MockServer; } const BASE_PARAMS: GetEventsParams = { @@ -196,7 +209,7 @@ describe("fetchEventsWithRetry – success path", () => { await fetchEventsWithRetry(server, BASE_PARAMS, { sleep: noopSleep }); - const callArgs = server.getEvents.mock.calls[0][0] as { + const callArgs = server.getEvents.mock.calls[0][0] as unknown as { filters: Array<{ topics: string[][] }>; }; expect(callArgs.filters[0].topics[0]).toEqual([...EVENT_TYPES]); @@ -212,7 +225,7 @@ describe("fetchEventsWithRetry – success path", () => { await fetchEventsWithRetry(server, params, { sleep: noopSleep }); - const callArgs = server.getEvents.mock.calls[0][0] as { + const callArgs = server.getEvents.mock.calls[0][0] as unknown as { startLedger: number; limit: number; filters: Array<{ contractIds: string[] }>; @@ -231,7 +244,7 @@ describe("fetchEventsWithRetry – success path", () => { { sleep: noopSleep } ); - const callArgs = server.getEvents.mock.calls[0][0] as { limit: number }; + const callArgs = server.getEvents.mock.calls[0][0] as unknown as { limit: number }; expect(callArgs.limit).toBe(100); }); @@ -581,7 +594,7 @@ describe("fetchEventsWithRetry – retry frequency increases up to max attempts" fakeNow += 1; return Promise.reject(err); }), - }; + } as unknown as MockServer; await fetchEventsWithRetry(server, BASE_PARAMS, { maxAttempts: MAX_ATTEMPTS, @@ -603,7 +616,7 @@ describe("fetchEventsWithRetry – retry frequency increases up to max attempts" callCount++; return Promise.reject(err); }), - }; + } as unknown as MockServer; const maxAttempts = 4; @@ -634,7 +647,7 @@ describe("fetchEventsWithRetry – mixed error scenarios", () => { if (callCount <= 2) return Promise.reject(connectionErr); return Promise.reject(nonConnectionErr); }), - }; + } as unknown as MockServer; await expect( fetchEventsWithRetry(server, BASE_PARAMS, { sleep: noopSleep }) @@ -656,7 +669,7 @@ describe("fetchEventsWithRetry – mixed error scenarios", () => { if (callCount === 1) return Promise.reject(connErr); return Promise.reject(badErr); }), - }; + } as unknown as MockServer; await expect( fetchEventsWithRetry(server, BASE_PARAMS, { sleep: noopSleep }) diff --git a/__tests__/failover-recovery-poll-diagnostics.test.ts b/__tests__/failover-recovery-poll-diagnostics.test.ts index 57273ec..c907b9f 100644 --- a/__tests__/failover-recovery-poll-diagnostics.test.ts +++ b/__tests__/failover-recovery-poll-diagnostics.test.ts @@ -1,6 +1,5 @@ import { jest } from "@jest/globals"; import Database from "better-sqlite3"; -import { jest } from "@jest/globals"; import { setDb, runMigrations } from "../src/indexer/db.js"; import { initializeNodeHealthTables, diff --git a/__tests__/indexer.test.ts b/__tests__/indexer.test.ts index cac7f8d..3553b5a 100644 --- a/__tests__/indexer.test.ts +++ b/__tests__/indexer.test.ts @@ -60,13 +60,9 @@ describe("Indexer Database", () => { }); it("does not re-apply already-applied migrations (idempotent)", () => { - const before = testDb - .prepare("SELECT version FROM schema_migrations ORDER BY version") - .all() as Array<{ version: number }>; - // Running again should not throw and should not duplicate rows const before = testDb - .prepare("SELECT version FROM schema_migrations") + .prepare("SELECT version FROM schema_migrations ORDER BY version") .all() as Array<{ version: number }>; runMigrations(); diff --git a/__tests__/sqlite-schema-manager.test.ts b/__tests__/sqlite-schema-manager.test.ts index 0f7cf6d..33274e8 100644 --- a/__tests__/sqlite-schema-manager.test.ts +++ b/__tests__/sqlite-schema-manager.test.ts @@ -18,7 +18,6 @@ import { } from "../src/indexer/db.js"; import { jest } from "@jest/globals"; import logger from "../src/utils/logger.js"; -import { SCHEMA_MANAGER_INDEXES } from "../src/indexer/db.js"; describe("SQLite Schema Manager – in-memory integration tests", () => { let testDb: Database.Database; diff --git a/__tests__/sqlite_vacuum_cleaner.test.ts b/__tests__/sqlite_vacuum_cleaner.test.ts index 2acd57b..da09411 100644 --- a/__tests__/sqlite_vacuum_cleaner.test.ts +++ b/__tests__/sqlite_vacuum_cleaner.test.ts @@ -1078,6 +1078,14 @@ describe("sqlite_vacuum_cleaner — failure alerting (#347)", () => { describe("runVacuumCleanup integration", () => { let testDb: Database.Database; + // Near-zero backoff so the retry path is exercised without real delays. + const fastConfig = { + maxRetries: 3, + initialBackoffMs: 1, + backoffMultiplier: 2, + maxBackoffMs: 5, + }; + beforeEach(() => { testDb = new Database(":memory:"); setDb(testDb); diff --git a/src/indexer/database-writer-pool.ts b/src/indexer/database-writer-pool.ts index f561cc9..5a1a002 100644 --- a/src/indexer/database-writer-pool.ts +++ b/src/indexer/database-writer-pool.ts @@ -3,6 +3,7 @@ import { getDb, getLastIndexedLedger, getShippedMigrationVersions, + insertEvent, verifySchemaIntegrity, verifySchemaUpToDate, type EventRow, @@ -1531,6 +1532,11 @@ export function resetWriterPoolStartState(): void { lastSchemaReport = null; migrationHooks.clear(); resetWriterPoolHistoricalRangeConfig(); + + // The queue's "already persisted" cache describes one specific database. + // Carrying it across a restart (or a setDb swap) would make the pool skip + // inserts for rows the new database has never seen. + defaultEventQueue.reset(); } // --------------------------------------------------------------------------- diff --git a/src/indexer/db.ts b/src/indexer/db.ts index 6503b4f..f330202 100644 --- a/src/indexer/db.ts +++ b/src/indexer/db.ts @@ -75,12 +75,6 @@ export const INDEXER_RUNNER_INDEXES = { } as const; /** Index names created by the schema-manager migration (#259). */ -export const SCHEMA_MANAGER_INDEXES = { - monitoredContractsActive: "idx_monitored_contracts_active", - eventsCreatedAt: "idx_events_created_at", - eventsContractTypeLedger: "idx_events_contract_type_ledger", -} as const; - // --------------------------------------------------------------------------- // Migration manager (#84) // --------------------------------------------------------------------------- @@ -198,15 +192,23 @@ const MIGRATIONS: Migration[] = [ ON webhook_subscriptions (webhook_url); `, }, + { + version: 8, + description: "add event_type_filter lookup indexes (#276)", + // The topic filter groups and filters by event_type. Migration 3 indexed + // (contract_id, event_type), which cannot serve a bare event_type + // predicate because event_type is not the leading column. + up: ` + CREATE INDEX IF NOT EXISTS idx_events_event_type + ON events (event_type); + + CREATE INDEX IF NOT EXISTS idx_events_contract_event_type + ON events (contract_id, event_type); + `, + }, ]; /** Index names created by the SQLite schema manager lookup-index migration (#259). */ -export const SCHEMA_MANAGER_INDEXES = { - monitoredContractsActive: "idx_monitored_contracts_active", - eventsCreatedAt: "idx_events_created_at", - eventsContractTypeLedger: "idx_events_contract_type_ledger", -} as const; - /** * Migration versions this build ships, ascending. Callers compare these * against `schema_migrations` to detect a database that is behind the code. @@ -216,12 +218,6 @@ export function getShippedMigrationVersions(): number[] { } /** Index names created by the version-5 migration (#259), for test assertions. */ -export const SCHEMA_MANAGER_INDEXES = { - monitoredContractsActive: "idx_monitored_contracts_active", - eventsCreatedAt: "idx_events_created_at", - eventsContractTypeLedger: "idx_events_contract_type_ledger", -} as const; - // --------------------------------------------------------------------------- // Exponential backoff retry for schema manager (#258) // Retries transient SQLite / connection / timeout failures during migrations. @@ -377,13 +373,18 @@ export async function withSchemaRetry( * Ensures the schema_migrations tracking table exists, then applies any * pending migrations in version order, each wrapped in its own transaction. */ -export function runMigrations(): void { +export function runMigrations( + retryConfig: Partial = {}, +): void { const monitor = getSqliteSchemaManagerFailureMonitor(); monitor.checkStall(); const startedAt = performance.now(); - const database = getDb(); let failureRecorded = false; + try { + const database = getDb(); + + const runAll = database.transaction(() => { // Bootstrap: create the migrations tracking table if it doesn't exist yet database.exec(` CREATE TABLE IF NOT EXISTS schema_migrations ( @@ -431,16 +432,43 @@ export function runMigrations(): void { }); try { - applyMigration(); + // Transient SQLite failures (locked/busy database) are retried with + // backoff; the savepoint means a failed attempt leaves nothing behind. + withSchemaRetrySync( + applyMigration, + retryConfig, + `migration_${migration.version}`, + ); logger.info("Migration applied", { version: migration.version }); } catch (err) { logger.error("Migration failed – rolled back", { version: migration.version, error: err instanceof Error ? err.message : String(err), }); + monitor.recordFailure("migration", { + error: err instanceof Error ? err.message : String(err), + version: migration.version, + description: migration.description, + elapsedMs: Math.round(performance.now() - startedAt), + }); + failureRecorded = true; throw err; } } + }); + + runAll(); + monitor.recordSuccess(); + } catch (err) { + // A bootstrap failure never reached the per-migration handler above. + if (!failureRecorded) { + monitor.recordFailure("bootstrap", { + error: err instanceof Error ? err.message : String(err), + elapsedMs: Math.round(performance.now() - startedAt), + }); + } + throw err; + } } /** @@ -997,33 +1025,10 @@ export function insertEventBatch(events: EventRow[], newLedger: number): void { * * Returns the number of rows actually inserted (excludes rows ignored as * duplicates). + * + * Superseded by the range-validating overload defined later in this file, + * which every caller uses. */ -export function insertHistoricalEventBatch(events: EventRow[]): number { - const db = getDb(); - - const insertStmt = db.prepare(` - INSERT OR IGNORE INTO events - (contract_id, event_type, ledger_sequence, timestamp, data_json) - VALUES (?, ?, ?, ?, ?) - `); - - const batchTransaction = db.transaction(() => { - let inserted = 0; - for (const ev of events) { - const result = insertStmt.run( - ev.contractId, - ev.eventType, - ev.ledgerSequence, - ev.timestamp, - ev.dataJson - ); - if (result.changes > 0) inserted++; - } - return inserted; - }); - - return batchTransaction(); -} // --------------------------------------------------------------------------- // In-memory event queue locks for concurrent inserts (#260) @@ -1162,33 +1167,10 @@ export async function insertEventBatchLocked( * * Returns the number of rows actually inserted (excludes rows ignored as * duplicates). + * + * Superseded by the range-validating overload defined later in this file, + * which every caller uses. */ -export function insertHistoricalEventBatch(events: EventRow[]): number { - const db = getDb(); - - const insertStmt = db.prepare(` - INSERT OR IGNORE INTO events - (contract_id, event_type, ledger_sequence, timestamp, data_json) - VALUES (?, ?, ?, ?, ?) - `); - - const batchTransaction = db.transaction(() => { - let inserted = 0; - for (const ev of events) { - const result = insertStmt.run( - ev.contractId, - ev.eventType, - ev.ledgerSequence, - ev.timestamp, - ev.dataJson - ); - if (result.changes > 0) inserted++; - } - return inserted; - }); - - return batchTransaction(); -} // --------------------------------------------------------------------------- // Event queries diff --git a/src/indexer/failover-recovery.ts b/src/indexer/failover-recovery.ts index 8f8d9d4..f51ced2 100644 --- a/src/indexer/failover-recovery.ts +++ b/src/indexer/failover-recovery.ts @@ -1,6 +1,26 @@ import { getDb } from "./db.js"; import logger from "../utils/logger.js"; +/** + * Debug line for one RPC round against a failover node (#249). + * + * The elapsed time and payload size are embedded in the message string as well + * as the metadata, so operators can grep slow or oversized rounds straight out + * of plain-text logs without a structured log backend. + */ +export function logPollDiagnostics( + nodeUrl: string, + startedAt: number, + payloadSizeBytes: number, +): void { + const elapsedMs = Math.max(0, Date.now() - startedAt); + + logger.debug( + `failover_recovery poll elapsedMs=${elapsedMs} payloadSizeBytes=${payloadSizeBytes}`, + { nodeUrl, elapsedMs, payloadSizeBytes }, + ); +} + /** * FailoverRecovery tracks the health of the RPC nodes the indexer reads from and * manages failover between them. Every write runs inside a SQLite transaction so diff --git a/src/indexer/indexer_metrics_collector.ts b/src/indexer/indexer_metrics_collector.ts index c2ce200..37e9799 100644 --- a/src/indexer/indexer_metrics_collector.ts +++ b/src/indexer/indexer_metrics_collector.ts @@ -1,5 +1,10 @@ import type Database from "better-sqlite3"; -import { getDb, insertEvent, type EventRow } from "./db.js"; +import { + getDb, + getShippedMigrationVersions, + insertEvent, + type EventRow, +} from "./db.js"; import logger from "../utils/logger.js"; import { withRetry, @@ -162,6 +167,9 @@ export interface IndexerMetricsDiagnostics { rowCount?: number; totalEvents?: number; lastIndexedLedger?: number; + /** Inclusive ledger window, set by the historical-sync collection path. */ + startLedger?: number; + endLedger?: number; error?: string; } @@ -698,6 +706,176 @@ export function getIndexerMetricsQueue(): IndexerMetricsEventQueue { return defaultQueue; } +// --------------------------------------------------------------------------- +// RPC health check with backoff retry (#334) +// --------------------------------------------------------------------------- + +export interface RpcHealthMetrics { + latestLedgerSequence: number; + collectedAt: string; +} + +/** + * Probe RPC liveness by reading the latest ledger, retrying transient + * failures with the shared backoff policy. + * + * Retries are delegated to `withRetry`, so a non-retryable error (a malformed + * request, say) surfaces on the first attempt instead of burning the budget. + * Either way a failure is recorded on the shared monitor as `rpc_timeout`, so + * repeated RPC trouble trips the same consecutive-failure alert as a failing + * collection. + */ +export async function collectRpcHealthMetrics( + server: RpcServerLike, + config: Partial = {}, +): Promise { + const startedAt = performance.now(); + + try { + const ledger = await withRetry( + () => server.getLatestLedger(), + config, + "rpc_health_check", + ); + + defaultMonitor.recordSuccess(); + logIndexerMetricsDiagnostics({ + collector: COLLECTOR_NAME, + operation: "rpc_health_check", + status: "success", + elapsedMs: roundElapsed(performance.now() - startedAt), + lastIndexedLedger: ledger.sequence, + }); + + return { + latestLedgerSequence: ledger.sequence, + collectedAt: new Date().toISOString(), + }; + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + + defaultMonitor.recordFailure("rpc_timeout", { + error, + operation: "rpc_health_check", + }); + + throw err; + } +} + +// --------------------------------------------------------------------------- +// Migration verification hooks (#340) +// --------------------------------------------------------------------------- +// The collector reads `events`, `indexer_state` and `schema_migrations` +// directly. Running it against a half-migrated database yields silently wrong +// metrics rather than an error, so callers can verify the schema up front. + +/** Tables and columns the collector's queries depend on. */ +const METRICS_REQUIRED_SCHEMA: Record = { + events: ["event_type", "ledger_sequence", "created_at"], + indexer_state: ["key", "value"], + schema_migrations: ["version"], +}; + +export interface IndexerMetricsSchemaReport { + valid: boolean; + missingTables: string[]; + missingColumns: Record; + missingMigrations: number[]; + errors: string[]; +} + +/** + * Check that every table, column and migration the collector relies on is + * present. Reports all problems at once rather than failing on the first. + */ +export function validateIndexerMetricsSchema( + targetDb?: Database.Database, +): IndexerMetricsSchemaReport { + const database = targetDb || getDb(); + const missingTables: string[] = []; + const missingColumns: Record = {}; + const missingMigrations: number[] = []; + const errors: string[] = []; + + for (const [table, requiredColumns] of Object.entries(METRICS_REQUIRED_SCHEMA)) { + const exists = database + .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?") + .get(table); + + if (!exists) { + missingTables.push(table); + errors.push(`Missing table: ${table}`); + continue; + } + + const columns = ( + database.prepare(`PRAGMA table_info(${table})`).all() as Array<{ + name: string; + }> + ).map((c) => c.name); + + const absent = requiredColumns.filter((c) => !columns.includes(c)); + if (absent.length > 0) { + missingColumns[table] = absent; + errors.push(`Missing columns in ${table}: ${absent.join(", ")}`); + } + } + + // Only meaningful when schema_migrations itself survived. + if (!missingTables.includes("schema_migrations")) { + const applied = new Set( + ( + database.prepare("SELECT version FROM schema_migrations").all() as Array<{ + version: number; + }> + ).map((r) => r.version), + ); + + for (const version of getShippedMigrationVersions()) { + if (!applied.has(version)) missingMigrations.push(version); + } + + if (missingMigrations.length > 0) { + errors.push(`Missing applied migrations: ${missingMigrations.join(", ")}`); + } + } + + return { + valid: errors.length === 0, + missingTables, + missingColumns, + missingMigrations, + errors, + }; +} + +/** Throw unless the collector's schema dependencies are all satisfied. */ +export function assertIndexerMetricsSchemaValid( + targetDb?: Database.Database, +): void { + const report = validateIndexerMetricsSchema(targetDb); + if (report.valid) return; + + logger.error("indexer_metrics_collector schema verification failed", { + collector: COLLECTOR_NAME, + missingTables: report.missingTables, + missingColumns: report.missingColumns, + missingMigrations: report.missingMigrations, + }); + + throw new Error( + `indexer_metrics_collector: database schema is out of sync – ${report.errors.join("; ")}`, + ); +} + +/** Drop queue, alert monitor and in-flight collection state. For tests. */ +export function resetIndexerMetricsCollectorState(): void { + defaultMonitor = new IndexerMetricsFailureMonitor(); + defaultQueue = new IndexerMetricsEventQueue(); + inFlightCollection = null; +} + // --------------------------------------------------------------------------- // Dynamic poller throttling parameters (#341) // --------------------------------------------------------------------------- diff --git a/src/indexer/indexer_runner.ts b/src/indexer/indexer_runner.ts index ec7cd1d..dd0a5f1 100644 --- a/src/indexer/indexer_runner.ts +++ b/src/indexer/indexer_runner.ts @@ -8,6 +8,117 @@ import logger from "../utils/logger.js"; * operators can spot slow RPC rounds or unexpectedly large event batches. */ +// --------------------------------------------------------------------------- +// Failure and stall alerting (#253) +// --------------------------------------------------------------------------- + +export interface IndexerRunnerFailureMonitorOptions { + /** Identifies the runner in alert payloads (default: "indexer_runner"). */ + name?: string; + /** Consecutive failures required before alerting (default: 3). */ + failureThreshold?: number; + /** Silence after which a non-failing runner is considered stalled. */ + stallThresholdMs?: number; +} + +const DEFAULT_RUNNER_FAILURE_THRESHOLD = 3; +const DEFAULT_RUNNER_STALL_THRESHOLD_MS = 120_000; + +/** + * Tracks consecutive failures for a runner and raises a single warning when + * the threshold is crossed. + * + * The alert latches: it fires once per episode rather than on every subsequent + * failure, so a persistently broken poll loop produces one alert instead of a + * stream of them. A success clears the latch. + */ +export class IndexerRunnerFailureMonitor { + private readonly name: string; + private readonly failureThreshold: number; + readonly stallThresholdMs: number; + + private consecutiveFailures = 0; + private lastSuccessfulAt: number | null = null; + private alertActive = false; + private stallAlerted = false; + + constructor(options: IndexerRunnerFailureMonitorOptions = {}) { + this.name = options.name ?? "indexer_runner"; + this.failureThreshold = + options.failureThreshold ?? DEFAULT_RUNNER_FAILURE_THRESHOLD; + this.stallThresholdMs = + options.stallThresholdMs ?? DEFAULT_RUNNER_STALL_THRESHOLD_MS; + } + + getConsecutiveFailures(): number { + return this.consecutiveFailures; + } + + getFailureThreshold(): number { + return this.failureThreshold; + } + + getLastSuccessfulAt(): number | null { + return this.lastSuccessfulAt; + } + + isAlertActive(): boolean { + return this.alertActive; + } + + recordFailure( + failureType: string, + details: { error?: string; operation?: string } = {}, + ): number { + this.consecutiveFailures += 1; + + if (this.consecutiveFailures >= this.failureThreshold && !this.alertActive) { + this.alertActive = true; + logger.warn("indexer_runner alert: consecutive failure threshold reached", { + runner: this.name, + failureType, + operation: details.operation, + consecutiveFailures: this.consecutiveFailures, + threshold: this.failureThreshold, + error: details.error, + }); + } + + return this.consecutiveFailures; + } + + recordSuccess(): void { + this.consecutiveFailures = 0; + this.lastSuccessfulAt = Date.now(); + this.alertActive = false; + this.stallAlerted = false; + } + + /** Warn once per episode when nothing has succeeded inside the window. */ + checkStall(): void { + if (this.lastSuccessfulAt === null || this.stallAlerted) return; + + const elapsed = Date.now() - this.lastSuccessfulAt; + if (elapsed <= this.stallThresholdMs) return; + + this.stallAlerted = true; + logger.warn("Poller stall detected – no successful poll for threshold period", { + runner: this.name, + elapsedMs: elapsed, + stallThresholdMs: this.stallThresholdMs, + consecutiveFailures: this.consecutiveFailures, + }); + } + + /** Reset in place – callers hold a reference to this instance. */ + reset(): void { + this.consecutiveFailures = 0; + this.lastSuccessfulAt = null; + this.alertActive = false; + this.stallAlerted = false; + } +} + export interface IndexerRunnerPollDiagnostics { operation: string; status: "started" | "success" | "failure"; @@ -175,11 +286,9 @@ export function adjustIndexerRunnerPollInterval( state.lastLoadAdjustmentAt = Date.now(); - logger.debug("indexer_runner throttle adjustment", { - processedEventCount, - currentIntervalMs: state.currentIntervalMs, - idleCycles: state.idleCycles, - }); - + // Deliberately silent: this runs once per poll, and the poller already emits + // a single consolidated diagnostics line per cycle. Logging here as well + // doubled every poll's debug output. The state is returned for callers that + // want to report it. return { ...state }; } diff --git a/src/indexer/ledger-range-tracker.ts b/src/indexer/ledger-range-tracker.ts index f2b33ac..1917e15 100644 --- a/src/indexer/ledger-range-tracker.ts +++ b/src/indexer/ledger-range-tracker.ts @@ -1,6 +1,114 @@ import { getDb, getLastIndexedLedger, insertEvent, type EventRow } from "./db.js"; import logger from "../utils/logger.js"; +// --------------------------------------------------------------------------- +// Schema verification (#259) +// --------------------------------------------------------------------------- +// The tracker reads and writes `events` and `indexer_state` directly. Starting +// it against a half-migrated database silently mis-tracks ledger progress, so +// callers verify the schema before the tracker is allowed to run. + +/** Tables and columns the tracker's statements depend on. */ +const LEDGER_RANGE_REQUIRED_SCHEMA: Record = { + events: ["contract_id", "event_type", "ledger_sequence", "data_json"], + indexer_state: ["key", "value"], +}; + +export interface LedgerRangeSchemaReport { + valid: boolean; + missingTables: string[]; + missingColumns: Record; + errors: string[]; +} + +/** + * Check every table and column the tracker needs, reporting all problems at + * once rather than failing on the first. + */ +export function verifyLedgerRangeTrackerSchema(): LedgerRangeSchemaReport { + const database = getDb(); + const missingTables: string[] = []; + const missingColumns: Record = {}; + const errors: string[] = []; + + for (const [table, requiredColumns] of Object.entries( + LEDGER_RANGE_REQUIRED_SCHEMA, + )) { + const exists = database + .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?") + .get(table); + + if (!exists) { + missingTables.push(table); + errors.push(`missing table: ${table}`); + continue; + } + + const columns = ( + database.prepare(`PRAGMA table_info(${table})`).all() as Array<{ + name: string; + }> + ).map((c) => c.name); + + const absent = requiredColumns.filter((c) => !columns.includes(c)); + if (absent.length > 0) { + missingColumns[table] = absent; + errors.push(`missing columns in ${table}: ${absent.join(", ")}`); + } + } + + // A hole in the applied migration versions means some migrations ran and + // others did not — the tracker's tables may exist but be the wrong shape. + if (!missingTables.includes("schema_migrations")) { + try { + const applied = ( + database + .prepare("SELECT version FROM schema_migrations ORDER BY version") + .all() as Array<{ version: number }> + ).map((r) => r.version); + + if (applied.length > 0) { + const gaps: number[] = []; + for (let v = applied[0]; v < applied[applied.length - 1]; v++) { + if (!applied.includes(v)) gaps.push(v); + } + if (gaps.length > 0) { + errors.push( + `migration version gap – missing applied migrations: ${gaps.join(", ")}`, + ); + } + } + } catch { + // schema_migrations exists but is unreadable; the table checks above + // already describe the damage. + } + } + + return { + valid: errors.length === 0, + missingTables, + missingColumns, + errors, + }; +} + +/** Throw unless the tracker's schema dependencies are all satisfied. */ +export function assertLedgerRangeTrackerSchemaValid(): void { + const report = verifyLedgerRangeTrackerSchema(); + if (report.valid) return; + + logger.error("LedgerRangeTracker schema verification failed", { + missingTables: report.missingTables, + missingColumns: report.missingColumns, + errors: report.errors, + }); + + throw new Error( + `LedgerRangeTracker schema verification failed – the tracker cannot start: ` + + report.errors.join("; "), + ); +} + /** * LedgerRangeTracker manages ledger range operations with full transaction support. * Ensures that operations on ledger sequence tracking are atomic and consistent diff --git a/src/indexer/poller.ts b/src/indexer/poller.ts index b84f470..5d12af9 100644 --- a/src/indexer/poller.ts +++ b/src/indexer/poller.ts @@ -8,11 +8,17 @@ import { getActiveContractIds, registerContract, adjustPollerInterval, - getCurrentPollIntervalMs, + getCurrentPollIntervalMs as getDbPollIntervalMs, + resetPollerThrottleState, verifySchemaUpToDate, assertSchemaValid, type EventRow, } from "./db.js"; +import { RpcPollerClient } from "./rpc-poller-client.js"; +import { + IndexerRunnerFailureMonitor, + adjustIndexerRunnerPollInterval, +} from "./indexer_runner.js"; import { deliverWebhooks } from "./webhook-delivery.js"; import { fetchEventsWithRetry } from "./event_type_filter.js"; import logger from "../utils/logger.js"; @@ -26,10 +32,15 @@ const RPC_URL = // All RPC calls in the indexer poll loop go through RpcPollerClient, which // retries transient failures (timeouts, connection resets, rate limits, 5xx) // with a doubling backoff up to maxRetries, then resets on success. +// Under test the backoff is collapsed to ~1ms: a rejected RPC mock would +// otherwise spend ~31s walking the production backoff curve before the poll +// reports failure, blowing past Jest's default timeout. +const IS_TEST_ENV = process.env.NODE_ENV === "test"; + const rpcClient = new RpcPollerClient(RPC_URL, { maxRetries: parseInt(process.env.INDEXER_RPC_MAX_RETRIES || "5", 10), initialBackoffMs: parseInt( - process.env.INDEXER_RPC_INITIAL_BACKOFF_MS || "1000", + process.env.INDEXER_RPC_INITIAL_BACKOFF_MS || (IS_TEST_ENV ? "1" : "1000"), 10, ), backoffMultiplier: parseInt( @@ -37,12 +48,33 @@ const rpcClient = new RpcPollerClient(RPC_URL, { 10, ), maxBackoffMs: parseInt( - process.env.INDEXER_RPC_MAX_BACKOFF_MS || "30000", + process.env.INDEXER_RPC_MAX_BACKOFF_MS || (IS_TEST_ENV ? "5" : "30000"), 10, ), }); -const failureMonitor = getIndexerRunnerFailureMonitor(); +// --------------------------------------------------------------------------- +// Failure and stall tracking for the poll loop (#253, #271) +// --------------------------------------------------------------------------- +// Consecutive failures escalate to an alert once the threshold is reached, and +// a poll loop that stops succeeding for longer than the stall threshold is +// reported even while individual polls keep "working". + +const failureMonitor = new IndexerRunnerFailureMonitor({ + name: "indexer_runner", + failureThreshold: parseInt( + process.env.INDEXER_RUNNER_FAILURE_THRESHOLD || + process.env.POLLER_FAILURE_THRESHOLD || + "3", + 10, + ), + stallThresholdMs: parseInt( + process.env.INDEXER_RUNNER_STALL_THRESHOLD_MS || + process.env.POLLER_STALL_THRESHOLD_MS || + "120000", + 10, + ), +}); export function getConsecutiveFailures(): number { return failureMonitor.getConsecutiveFailures(); @@ -53,7 +85,262 @@ export function getLastSuccessfulPollAt(): number | null { } export function resetFailureState(): void { - resetIndexerRunnerFailureState(); + failureMonitor.reset(); + resetPollDiagnosticsThrottle(); +} + +// --------------------------------------------------------------------------- +// Dynamic poll interval (#265) +// --------------------------------------------------------------------------- + +const POLL_INTERVAL_MIN_MS = parseInt(process.env.POLL_INTERVAL_MS || "15000", 10); +const POLL_INTERVAL_MAX_MS = parseInt( + process.env.POLL_INTERVAL_MAX_MS || "120000", + 10, +); +const POLL_INTERVAL_BACKOFF = 2; + +/** + * Next poll delay given the current one and whether the last poll saw activity. + * + * Idle polls back off geometrically up to POLL_INTERVAL_MAX_MS; the first + * active poll drops straight back to the minimum. Pure function so the backoff + * curve can be reasoned about (and tested) without running the loop. + */ +export function nextPollIntervalMs( + currentIntervalMs: number, + sawActivity: boolean, +): number { + if (sawActivity) return POLL_INTERVAL_MIN_MS; + return Math.min(currentIntervalMs * POLL_INTERVAL_BACKOFF, POLL_INTERVAL_MAX_MS); +} + +let currentPollIntervalMs = POLL_INTERVAL_MIN_MS; + +/** The interval the poll loop is currently waiting between cycles. */ +export function getCurrentPollIntervalMs(): number { + return currentPollIntervalMs; +} + +/** Map an RPC event notification onto the row shape `events` stores. */ +function toEventRow(event: any, fallbackContractId: string): EventRow { + return { + contractId: event.contractId?.contractId?.() ?? fallbackContractId, + eventType: scValToNative(event.topic[0]) as string, + ledgerSequence: event.ledger, + timestamp: event.ledgerClosedAt + ? Math.floor(new Date(event.ledgerClosedAt).getTime() / 1000) + : Math.floor(Date.now() / 1000), + dataJson: JSON.stringify(scValToNative(event.value)), + }; +} + +// --------------------------------------------------------------------------- +// Poll diagnostics (#270) +// --------------------------------------------------------------------------- + +const POLL_DIAGNOSTIC_LOG_MIN_INTERVAL_MS = parseInt( + process.env.POLL_DIAGNOSTIC_LOG_MIN_INTERVAL_MS || "60000", + 10, +); +let lastPollDiagnosticAt = 0; +let stallWindowReported = false; + +interface StallWindow { + elapsedMsSinceLastSuccess: number; + stallThresholdMs: number; +} + +/** Clear the diagnostics throttle so the next poll logs unconditionally. */ +export function resetPollDiagnosticsThrottle(): void { + lastPollDiagnosticAt = 0; + stallWindowReported = false; +} + +/** + * One debug line per poll carrying duration and payload size. + * + * Only byte counts are logged, never the payload itself — event data carries + * participant wallet addresses and amounts that have no business in a log. + * + * Throttled to at most one line per POLL_DIAGNOSTIC_LOG_MIN_INTERVAL_MS so a + * fast poll loop cannot flood the log. + */ +function logPollDiagnostics( + elapsedMs: number, + batch: EventRow[], + stall: StallWindow | null = null, +): void { + const now = Date.now(); + + // The first line carrying a stall window is new information, so it is never + // throttled away; repeats of it are. + const stallWindowIsNews = stall !== null && !stallWindowReported; + + if ( + !stallWindowIsNews && + POLL_DIAGNOSTIC_LOG_MIN_INTERVAL_MS > 0 && + lastPollDiagnosticAt > 0 && + now - lastPollDiagnosticAt < POLL_DIAGNOSTIC_LOG_MIN_INTERVAL_MS + ) { + return; + } + lastPollDiagnosticAt = now; + if (stall !== null) stallWindowReported = true; + + const totalPayloadBytes = batch.reduce( + (sum, row) => sum + Buffer.byteLength(row.dataJson, "utf8"), + 0, + ); + const avgPayloadBytes = + batch.length > 0 ? Math.round(totalPayloadBytes / batch.length) : 0; + const rounded = Math.round(elapsedMs); + + const stallSegment = stall + ? ` | Poller stall diagnostics elapsedMsSinceLastSuccess=${stall.elapsedMsSinceLastSuccess}` + + ` stallThresholdMs=${stall.stallThresholdMs}` + : ""; + + logger.debug( + `RPC getEvents diagnostics elapsedMs=${rounded} payloadSizeBytes=${totalPayloadBytes}` + + stallSegment, + { + elapsedMs: rounded, + payloadSizeBytes: totalPayloadBytes, + totalPayloadBytes, + avgPayloadBytes, + eventCount: batch.length, + ...(stall ?? {}), + }, + ); +} + +// --------------------------------------------------------------------------- +// Historical range import (#254) +// --------------------------------------------------------------------------- + +/** Upper bound on how many ledgers a single backfill may cover. */ +export const MAX_LEDGERS_PER_IMPORT = parseInt( + process.env.MAX_LEDGERS_PER_IMPORT || "10000", + 10, +); + +const HISTORICAL_PAGE_SIZE = 100; + +export interface HistoricalRangeValidation { + valid: boolean; + error?: string; +} + +export interface HistoricalImportResult { + eventsFound: number; + eventsImported: number; +} + +/** + * Check a requested backfill window against the chain head and the per-import + * ceiling. Pure, so callers can validate before opening any RPC connection. + */ +export function validateHistoricalRange( + startLedger: number, + endLedger: number, + chainHeadLedger: number, +): HistoricalRangeValidation { + if (!Number.isInteger(startLedger) || startLedger <= 0) { + return { valid: false, error: "startLedger must be a positive integer" }; + } + if (!Number.isInteger(endLedger) || endLedger <= 0) { + return { valid: false, error: "endLedger must be a positive integer" }; + } + if (startLedger > endLedger) { + return { valid: false, error: "startLedger must be <= endLedger" }; + } + if (endLedger > chainHeadLedger) { + return { + valid: false, + error: + `endLedger ${endLedger} does not exist yet – ` + + `the chain head is ${chainHeadLedger}`, + }; + } + + const span = endLedger - startLedger + 1; + if (span > MAX_LEDGERS_PER_IMPORT) { + return { + valid: false, + error: + `range covers ${span} ledgers, exceeding the ` + + `${MAX_LEDGERS_PER_IMPORT}-ledger maximum per import`, + }; + } + + return { valid: true }; +} + +/** + * Import a historical ledger window without disturbing the live poller. + * + * Rows go through `insertHistoricalEventBatch`, which keeps the live ledger + * pointer where it is and relies on the same UNIQUE constraint as the poller — + * so re-running an import is idempotent. Pages are followed by cursor until a + * short page signals the range is fully collected. + */ +export async function fetchHistoricalEvents( + startLedger: number, + endLedger: number, +): Promise { + const chainHead = (await rpcClient.getLatestLedger()).sequence; + const validation = validateHistoricalRange(startLedger, endLedger, chainHead); + if (!validation.valid) { + throw new Error(`Invalid historical range: ${validation.error}`); + } + + let contractIds: string[] = getActiveContractIds(); + if (contractIds.length === 0 && process.env.CONTRACT_ID) { + contractIds = [process.env.CONTRACT_ID]; + } + + let eventsFound = 0; + let eventsImported = 0; + let cursor: string | undefined; + + for (;;) { + // Cursor mode supersedes the ledger window once paging has started; sending + // both is rejected by the RPC. + const params = cursor + ? { cursor, limit: HISTORICAL_PAGE_SIZE } + : { startLedger, endLedger, contractIds, limit: HISTORICAL_PAGE_SIZE }; + + const page = await rpcClient.getEvents(params); + const pageEvents: any[] = page?.events ?? []; + if (pageEvents.length === 0) break; + + eventsFound += pageEvents.length; + + const batch: EventRow[] = pageEvents + .map((event) => toEventRow(event, contractIds[0] ?? "")) + .filter( + (row) => + row.ledgerSequence >= startLedger && row.ledgerSequence <= endLedger, + ); + + if (batch.length > 0) { + const result = insertHistoricalEventBatch(batch, { startLedger, endLedger }); + eventsImported += result.inserted; + } + + cursor = page?.cursor; + if (pageEvents.length < HISTORICAL_PAGE_SIZE || !cursor) break; + } + + logger.info("Historical import complete", { + startLedger, + endLedger, + eventsFound, + eventsImported, + }); + + return { eventsFound, eventsImported }; } // --------------------------------------------------------------------------- @@ -130,19 +417,26 @@ export async function pollEvents(): Promise { } // --- Alerting: stall detection before polling (#253, #271) --- - failureMonitor.checkStall(); - if (failureMonitor.getLastSuccessfulAt()) { - const elapsed = Date.now() - (failureMonitor.getLastSuccessfulAt() as number); + const lastSuccessAt = failureMonitor.getLastSuccessfulAt(); + let stallWindow: StallWindow | null = null; + + if (lastSuccessAt) { + const elapsedMsSinceLastSuccess = Date.now() - lastSuccessAt; const stallThresholdMs = parseInt( process.env.INDEXER_RUNNER_STALL_THRESHOLD_MS || process.env.POLLER_STALL_THRESHOLD_MS || String(failureMonitor.stallThresholdMs), 10, ); - logger.debug("Poller stall diagnostics", { - elapsedMsSinceLastSuccess: elapsed, - stallThresholdMs, - }); + stallWindow = { elapsedMsSinceLastSuccess, stallThresholdMs }; + + if (elapsedMsSinceLastSuccess > stallThresholdMs) { + logger.warn("Poller stall detected – no successful poll for threshold period", { + elapsedMs: elapsedMsSinceLastSuccess, + stallThresholdMs, + consecutiveFailures: failureMonitor.getConsecutiveFailures(), + }); + } } const pollStart = performance.now(); @@ -152,11 +446,33 @@ export async function pollEvents(): Promise { // schema must not silently pass through the topic filter (#282). verifySchemaUpToDate(); + // --- Dynamic historical sync ranges (#254) --- + // With LEDGER_RANGE_START/END set the poller imports that inclusive window + // instead of following the chain head, and leaves the live ledger pointer + // untouched so a backfill cannot disturb live indexing. + const rangeStartRaw = process.env.LEDGER_RANGE_START; + const rangeEndRaw = process.env.LEDGER_RANGE_END; + if (rangeStartRaw || rangeEndRaw) { + const rangeStart = parseInt(rangeStartRaw ?? rangeEndRaw ?? "0", 10); + const rangeEnd = parseInt(rangeEndRaw ?? rangeStartRaw ?? "0", 10); + + const imported = await fetchHistoricalEvents(rangeStart, rangeEnd); + failureMonitor.recordSuccess(); + + logger.info("Historical range import cycle complete", { + startLedger: rangeStart, + endLedger: rangeEnd, + ...imported, + }); + return true; + } + const lastLedger = getLastIndexedLedger(); - const currentLedger = (await server.getLatestLedger()).sequence; + const currentLedger = (await rpcClient.getLatestLedger()).sequence; if (currentLedger <= lastLedger) { - // --- Dynamic throttling: idle cycle (#265) --- + // --- Dynamic throttling: idle cycle (#265, #256) --- adjustPollerInterval(0); + adjustIndexerRunnerPollInterval(0); return false; } @@ -165,23 +481,13 @@ export async function pollEvents(): Promise { logger.info("Polling events", { startLedger, currentLedger }); const eventsStart = performance.now(); - const events = await fetchEventsWithRetry(server, { + const events = await fetchEventsWithRetry(rpcClient.rpcServer, { startLedger, contractIds, limit: 100, }); const eventsElapsed = performance.now() - eventsStart; - // --- Diagnostics: payload size and timing (#270) --- - const payloadSizeBytes = JSON.stringify(events.events).length; - logger.debug("RPC getEvents diagnostics", { - elapsedMs: Math.round(eventsElapsed), - payloadSizeBytes, - eventCount: events.events.length, - startLedger, - currentLedger, - }); - // Build the batch to be written atomically (#84) const batch: EventRow[] = events.events.map((event) => toEventRow(event, contractIds[0]) @@ -193,11 +499,11 @@ export async function pollEvents(): Promise { await enqueueEventInsert(batch, currentLedger); const totalElapsed = performance.now() - pollStart; - consecutiveFailures = 0; - lastSuccessfulPollAt = Date.now(); + failureMonitor.recordSuccess(); // --- Dynamic poller throttling (#265) --- const throttleState = adjustPollerInterval(events.events.length); + adjustIndexerRunnerPollInterval(events.events.length); logger.info("Processed indexer poll", { eventCount: events.events.length, @@ -206,7 +512,7 @@ export async function pollEvents(): Promise { pollIntervalMs: throttleState.currentIntervalMs, }); - logPollDiagnostics(performance.now() - pollStartedAt, batch); + logPollDiagnostics(performance.now() - pollStartedAt, batch, stallWindow); deliverWebhooks(startLedger, currentLedger).catch((err) => logger.error("Error delivering webhooks", { @@ -217,7 +523,10 @@ export async function pollEvents(): Promise { return true; } catch (err) { const totalElapsed = performance.now() - pollStart; - consecutiveFailures += 1; + const consecutiveFailures = failureMonitor.recordFailure("poll", { + error: err instanceof Error ? err.message : String(err), + operation: "poll_events", + }); logger.error("Error polling events", { error: err instanceof Error ? err.message : String(err), @@ -245,11 +554,17 @@ export async function pollEvents(): Promise { let pollerTimeout: NodeJS.Timeout | null = null; let pollerRunning = false; +/** One poll plus the interval adjustment its outcome implies. */ +async function runPollCycle(): Promise { + const sawActivity = await pollEvents(); + currentPollIntervalMs = nextPollIntervalMs(currentPollIntervalMs, sawActivity); +} + async function pollLoop() { if (!pollerRunning) return; - await pollEvents(); - const interval = getCurrentPollIntervalMs(); - pollerTimeout = setTimeout(pollLoop, interval); + await runPollCycle(); + if (!pollerRunning) return; + pollerTimeout = setTimeout(pollLoop, currentPollIntervalMs); } export function startPoller() { @@ -260,11 +575,18 @@ export function startPoller() { assertSchemaValid(); pollerRunning = true; + currentPollIntervalMs = POLL_INTERVAL_MIN_MS; logger.info("Starting event indexer poller", { - intervalMs: getCurrentPollIntervalMs(), + intervalMs: currentPollIntervalMs, + }); + + // The first cycle runs immediately; the loop is scheduled off its outcome so + // the very first idle poll already widens the wait. + void runPollCycle().then(() => { + if (pollerRunning) { + pollerTimeout = setTimeout(pollLoop, currentPollIntervalMs); + } }); - pollEvents(); - pollerTimeout = setTimeout(pollLoop, getCurrentPollIntervalMs()); } export function stopPoller() { @@ -273,5 +595,6 @@ export function stopPoller() { clearTimeout(pollerTimeout); pollerTimeout = null; } + resetPollerThrottleState(); currentPollIntervalMs = POLL_INTERVAL_MIN_MS; } diff --git a/src/indexer/rpc-poller-client.ts b/src/indexer/rpc-poller-client.ts index 0525e35..165100c 100644 --- a/src/indexer/rpc-poller-client.ts +++ b/src/indexer/rpc-poller-client.ts @@ -580,6 +580,15 @@ export class RpcPollerClient { }); } + /** + * The underlying RPC server, for helpers that take a bare server and apply + * their own retry policy (e.g. `fetchEventsWithRetry`). Going through + * `getEvents()` instead would stack this client's retries on top of theirs. + */ + get rpcServer(): RpcServerLike { + return this.server; + } + async getLatestLedger(): Promise<{ sequence: number }> { return withRetry( () => this.server.getLatestLedger(), diff --git a/src/indexer/sqlite_vacuum_cleaner.ts b/src/indexer/sqlite_vacuum_cleaner.ts index c0f03b9..a9fab36 100644 --- a/src/indexer/sqlite_vacuum_cleaner.ts +++ b/src/indexer/sqlite_vacuum_cleaner.ts @@ -1,6 +1,70 @@ import type Database from "better-sqlite3"; +import { getDb } from "./db.js"; import logger from "../utils/logger.js"; +// --------------------------------------------------------------------------- +// SQLite index structures (#344) +// --------------------------------------------------------------------------- +// The cleaner's two hot predicates are the retention sweep (created_at < …) +// and the ledger-range prune (ledger_sequence BETWEEN …). Migration 6 creates +// these indexes; the helpers below name them and prove the planner uses them. + +export const VACUUM_CLEANER_INDEXES = { + /** Retention sweep: DELETE ... WHERE created_at < cutoff. */ + eventsCreatedAt: "idx_events_created_at", + /** Ledger-range prune: DELETE ... WHERE ledger_sequence BETWEEN ? AND ?. */ + eventsLedgerSequence: "idx_events_ledger_sequence", + /** Composite for range-scoped retention sweeps. */ + eventsCreatedAtLedger: "idx_events_created_at_ledger", + /** Composite for retention-scoped range prunes. */ + eventsLedgerCreatedAt: "idx_events_ledger_created_at", +} as const; + +/** Every index name the vacuum cleaner manages, in declaration order. */ +export function getVacuumIndexNames(): string[] { + return Object.values(VACUUM_CLEANER_INDEXES); +} + +/** + * Create any managed index that is missing and return all of their names. + * Idempotent: safe to call on a database migration 6 has already touched. + */ +export function ensureVacuumIndexes(targetDb?: Database.Database): string[] { + const database = targetDb || getDb(); + + database.exec(` + CREATE INDEX IF NOT EXISTS ${VACUUM_CLEANER_INDEXES.eventsCreatedAt} + ON events (created_at); + CREATE INDEX IF NOT EXISTS ${VACUUM_CLEANER_INDEXES.eventsLedgerSequence} + ON events (ledger_sequence); + CREATE INDEX IF NOT EXISTS ${VACUUM_CLEANER_INDEXES.eventsCreatedAtLedger} + ON events (created_at, ledger_sequence); + CREATE INDEX IF NOT EXISTS ${VACUUM_CLEANER_INDEXES.eventsLedgerCreatedAt} + ON events (ledger_sequence, created_at); + `); + + return getVacuumIndexNames(); +} + +/** EXPLAIN QUERY PLAN rows for a statement, with its bind parameters. */ +export function vacuumExplainQueryPlan( + targetDb: Database.Database, + sql: string, + ...params: unknown[] +): Array> { + return targetDb + .prepare(`EXPLAIN QUERY PLAN ${sql}`) + .all(...(params as never[])) as Array>; +} + +/** True when any plan row references the expected index name. */ +export function vacuumQueryPlanUsesIndex( + plan: Array>, + indexName: string, +): boolean { + return plan.some((row) => String(row.detail ?? "").includes(indexName)); +} + // --------------------------------------------------------------------------- // SQLite vacuum cleaner (#193) // --------------------------------------------------------------------------- @@ -299,6 +363,16 @@ export function runVacuumCleanup( logger.info("Starting sqlite vacuum cleanup", { retentionDays }); + // Opening boundary of the cycle; the closing one is emitted on both the + // success and failure paths so every cycle is bracketed in the logs. + logVacuumPollDiagnostics({ + component: VACUUM_COMPONENT_NAME, + operation: "vacuum_cleanup", + status: "started", + elapsedMs: 0, + retentionDays, + }); + // Step 1: transactional prune. If this throws, record the failure (which // alerts once the consecutive-failure threshold is reached) and propagate // immediately — VACUUM is intentionally skipped. diff --git a/src/routes/jobs.ts b/src/routes/jobs.ts index 309272d..1c14cce 100644 --- a/src/routes/jobs.ts +++ b/src/routes/jobs.ts @@ -163,6 +163,21 @@ export function resetPartialReleaseCache(): void { inFlightPartialReleaseRequests.clear(); } +/** + * Cache key for a partial-release build. The amount and source address are + * part of the key because each combination produces a different unsigned + * transaction — keying on the contract and milestone alone would serve one + * caller another caller's XDR. + */ +function partialReleaseCacheKey( + contractId: string, + index: string | number, + amount: unknown, + sourceAddress: string, +): string { + return `${contractId}:${index}:${String(amount)}:${sourceAddress}`; +} + // --------------------------------------------------------------------------- // Simulation error helpers (#83) // --------------------------------------------------------------------------- @@ -1162,6 +1177,27 @@ router.post( return; } + const cacheKey = partialReleaseCacheKey( + contractId as string, + index as string, + amount, + sourceAddress, + ); + + logger.debug("Checking partial-release cache", { traceId, contractId, index, cacheKey }); + const cachedXdr = partialReleaseCache.get(cacheKey); + if (cachedXdr !== undefined) { + logger.info("Partial-release XDR served from cache", { + traceId, + contractId, + index, + source: "cache", + xdrLength: cachedXdr.length, + }); + res.json({ success: true, xdr: cachedXdr }); + return; + } + let requestPromise = inFlightPartialReleaseRequests.get(cacheKey); const servedFromInFlight = Boolean(requestPromise); @@ -1200,7 +1236,12 @@ router.post( } catch (err: any) { const errMsg = String(err?.message || err); const { status, message } = classifySimError(errMsg); - logger.error("Failed to prepare transaction for partial release", { contractId, error: errMsg }); + logger.error("Failed to prepare transaction for partial release", { + traceId, + contractId, + index, + error: errMsg, + }); throw { status, message }; }