From 41e26f5aadb0130d49d1131c8abead6928963688 Mon Sep 17 00:00:00 2001 From: praizeD10 Date: Mon, 19 Oct 2026 13:56:35 +0100 Subject: [PATCH] fix(indexer): bind canonical column and derive event identity from TOID --- docs/indexer-health-probe.md | 29 +- .../0030_indexer_event_identity.sql | 21 + src/db/schema.ts | 5 + src/lib/http.ts | 4 +- src/metrics/registry.ts | 23 + src/services/indexerRecovery.ts | 109 ++++- src/services/indexerService.ts | 233 +++++++-- tests/indexerGapScan.test.ts | 2 +- tests/indexerRecovery.test.ts | 67 ++- tests/indexerService.recovery.test.ts | 452 ++++++++++++++++-- tests/integration/indexerPersistence.test.ts | 298 ++++++++++++ 11 files changed, 1129 insertions(+), 114 deletions(-) create mode 100644 drizzle/migrations/0030_indexer_event_identity.sql create mode 100644 tests/integration/indexerPersistence.test.ts diff --git a/docs/indexer-health-probe.md b/docs/indexer-health-probe.md index 26ea21e9..f0f5fbe8 100644 --- a/docs/indexer-health-probe.md +++ b/docs/indexer-health-probe.md @@ -82,6 +82,31 @@ shutdown → stopIndexerHealthProbe(handle) → stopScheduler() → closeDb() ## Implementation - **Job**: `src/jobs/indexerHealthProbe.ts` -- **Metric**: `src/metrics/registry.ts` — `indexerLagLedgers` Gauge +- **Metric**: `src/metrics/registry.ts` — `indexerLagLedgers` Gauge, `indexer_events_inserted_total`, `indexer_reorgs_detected_total` Counters - **Config**: `src/config/env.ts` — `INDEXER_LAG_ALERT_THRESHOLD` -- **Tests**: `tests/indexerHealthProbe.test.ts` +- **Tests**: `tests/indexerHealthProbe.test.ts`, `tests/indexerService.recovery.test.ts`, `tests/integration/indexerPersistence.test.ts` + +--- + +## Event Identity and Reorganization Recovery + +Soroban RPC emits events whose `id` adheres to the format `-`: +- **TOID prefix (64-bit integer)**: + - Ledger sequence: bits 32..63 (32 bits) + - Transaction application order: bits 12..31 (20 bits) + - Operation index: bits 0..11 (12 bits) +- **Event index suffix**: 10-digit zero-padded index of the event within its operation. + +### Identity vs. Position + +- **Event Identity**: `(ledger, tx_hash, op_index, event_index)` uniquely identifies an immutable event record across re-deliveries. +- **Ledger Position**: `(ledger, tx_application_order, op_index, event_index)` defines the exact execution slot in the ledger. + +### Atomic Reorganization Replacement + +When a new transaction appears at an already indexed ledger position with a different transaction hash (`SELECT ... FOR UPDATE`), a true reorganization has occurred: +1. The stale event is marked non-canonical (`canonical = false`). +2. Predictions linked to the reorganized transaction hash are reset to `status = 'pending'`, with results cleared. +3. A reorganization record is inserted into `indexer_reorgs` with `repaired_predictions` count. +4. The new event is inserted with `canonical = true`. +5. The entire transition executes inside a single database transaction (`BEGIN ... COMMIT`), guaranteeing atomicity and preventing false reorgs between distinct transactions in the same ledger. diff --git a/drizzle/migrations/0030_indexer_event_identity.sql b/drizzle/migrations/0030_indexer_event_identity.sql new file mode 100644 index 00000000..ff66f03b --- /dev/null +++ b/drizzle/migrations/0030_indexer_event_identity.sql @@ -0,0 +1,21 @@ +ALTER TABLE "indexer_events" ADD COLUMN IF NOT EXISTS "tx_application_order" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +ALTER TABLE "indexer_events" ADD COLUMN IF NOT EXISTS "event_index" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +UPDATE "indexer_events" SET "event_index" = 0 WHERE "event_index" IS NULL; +--> statement-breakpoint +UPDATE "indexer_events" SET "tx_application_order" = 0 WHERE "tx_application_order" IS NULL; +--> statement-breakpoint +DROP INDEX IF EXISTS "indexer_events_identity_idx"; +--> statement-breakpoint +CREATE UNIQUE INDEX IF NOT EXISTS "indexer_events_identity_idx" ON "indexer_events" ("ledger", "tx_hash", "op_index", "event_index"); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "indexer_events_position_idx" ON "indexer_events" ("ledger", "tx_application_order", "op_index", "event_index"); +--> statement-breakpoint +ALTER TABLE "indexer_reorgs" ADD COLUMN IF NOT EXISTS "tx_application_order" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +ALTER TABLE "indexer_reorgs" ADD COLUMN IF NOT EXISTS "event_index" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +ALTER TABLE "indexer_reorgs" ADD COLUMN IF NOT EXISTS "repaired_predictions" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "indexer_reorgs_position_idx" ON "indexer_reorgs" ("ledger", "tx_application_order", "op_index", "event_index"); diff --git a/src/db/schema.ts b/src/db/schema.ts index 7da4fb56..5cfc70df 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -347,7 +347,9 @@ export const indexerEvents = pgTable("indexer_events", { id: uuid("id").primaryKey().defaultRandom(), ledger: integer("ledger").notNull(), txHash: text("tx_hash").notNull(), + txApplicationOrder: integer("tx_application_order").notNull().default(0), opIndex: integer("op_index").notNull().default(0), + eventIndex: integer("event_index").notNull().default(0), eventType: text("event_type").notNull(), createdAt: timestamp("created_at", { withTimezone: true }) .notNull() @@ -362,9 +364,12 @@ export type IndexerEvent = typeof indexerEvents.$inferSelect; export const indexerReorgs = pgTable("indexer_reorgs", { id: uuid("id").primaryKey().defaultRandom(), ledger: integer("ledger").notNull(), + txApplicationOrder: integer("tx_application_order").notNull().default(0), opIndex: integer("op_index").notNull(), + eventIndex: integer("event_index").notNull().default(0), oldTxHash: text("old_tx_hash").notNull(), newTxHash: text("new_tx_hash").notNull(), + repairedPredictions: integer("repaired_predictions").notNull().default(0), status: text("status").notNull().default("detected"), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }); diff --git a/src/lib/http.ts b/src/lib/http.ts index d196c6b8..e51dfa71 100644 --- a/src/lib/http.ts +++ b/src/lib/http.ts @@ -75,11 +75,11 @@ export async function fetchWithRetry( } return response; - } catch (err: any) { + } catch (err: unknown) { lastError = err; // Do not retry on explicit aborts from the caller's own signal - if (err.name === 'AbortError' && init?.signal?.aborted) { + if ((err as { name?: string }).name === 'AbortError' && init?.signal?.aborted) { throw err; } diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index f2df6f3d..196799da 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -10,6 +10,10 @@ export const register = new Registry(); collectDefaultMetrics({ register }); +export function resetMetrics(): void { + register.resetMetrics(); +} + export const httpRequestDuration = new Histogram({ name: "http_request_duration_seconds", help: "Duration of HTTP requests in seconds, segmented by route template and status code", @@ -58,6 +62,25 @@ export const indexerLagLedgers = new Gauge({ registers: [register], }); +export const indexerEventsInsertedTotal = new Counter({ + name: "indexer_events_inserted_total", + help: "Total number of indexer events inserted into the database", + registers: [register], +}); + +export const indexerReorgsDetectedTotal = new Counter({ + name: "indexer_reorgs_detected_total", + help: "Total number of chain reorganizations detected by the indexer", + registers: [register], +}); + +export const indexerGapDetectedTotal = new Counter({ + name: "indexer_gap_detected_total", + help: "Total number of indexer gaps detected during scanning", + labelNames: ["from", "to"] as const, + registers: [register], +}); + export const scheduledReportRunsTotal = new Counter({ name: "scheduled_report_runs_total", help: "Scheduled report runs by terminal outcome", diff --git a/src/services/indexerRecovery.ts b/src/services/indexerRecovery.ts index 701bfd50..bba93f5c 100644 --- a/src/services/indexerRecovery.ts +++ b/src/services/indexerRecovery.ts @@ -8,10 +8,88 @@ * treated as a reorg rather than a second canonical event. */ +export interface DecodedEventIdentity { + ledger: number; + txApplicationOrder: number; + opIndex: number; + eventIndex: number; +} + +/** + * Decodes a Soroban RPC event id into its positional coordinates. + * + * Soroban RPC event IDs format: + * - + * + * The TOID prefix is a 64-bit integer encoding: + * - ledger sequence (bits 32..63, 32 bits) + * - transaction application order within ledger (bits 12..31, 20 bits) + * - operation index within transaction (bits 0..11, 12 bits) + * + * The suffix after the hyphen is the zero-padded event index within that operation. + * + * Reference: Stellar Horizon / Soroban RPC TOID specification + * https://github.com/stellar/stellar-protocol + */ +export function decodeEventId(eventId: string): DecodedEventIdentity | null { + if (!eventId || typeof eventId !== "string") { + return null; + } + const parts = eventId.trim().split("-"); + if (parts.length !== 2) { + return null; + } + const [toidStr, eventIndexStr] = parts; + if (!/^\d+$/.test(toidStr) || !/^\d+$/.test(eventIndexStr)) { + return null; + } + + try { + const toid = BigInt(toidStr); + const eventIndex = Number.parseInt(eventIndexStr, 10); + if (!Number.isSafeInteger(eventIndex) || eventIndex < 0) { + return null; + } + + const ledger = Number((toid >> 32n) & 0xffffffffn); + const txApplicationOrder = Number((toid >> 12n) & 0xfffffn); + const opIndex = Number(toid & 0xfffn); + + return { + ledger, + txApplicationOrder, + opIndex, + eventIndex, + }; + } catch { + return null; + } +} + +/** + * Encodes positional coordinates into a standard Soroban RPC event ID. + */ +export function formatEventId( + ledger: number, + txApplicationOrder: number, + opIndex: number, + eventIndex: number, +): string { + const toid = + (BigInt(ledger) << 32n) | + (BigInt(txApplicationOrder) << 12n) | + BigInt(opIndex); + const toidStr = toid.toString().padStart(19, "0"); + const eventIndexStr = eventIndex.toString().padStart(10, "0"); + return `${toidStr}-${eventIndexStr}`; +} + export interface RecoveryEvent { ledger: number; txHash: string; + txApplicationOrder?: number; opIndex: number; + eventIndex?: number; eventType?: string; payload?: unknown; } @@ -23,7 +101,9 @@ export interface LedgerObservation { export interface ReorgReplacement { ledger: number; + txApplicationOrder: number; opIndex: number; + eventIndex: number; oldTxHash: string; newTxHash: string; } @@ -33,12 +113,16 @@ export interface ReplayChunk { to: number; } -export function eventIdentity(event: Pick): string { - return `${event.ledger}:${event.txHash}:${event.opIndex}`; +export function eventIdentity( + event: Pick & { eventIndex?: number }, +): string { + return `${event.ledger}:${event.txHash}:${event.opIndex}:${event.eventIndex ?? 0}`; } -export function positionIdentity(event: Pick): string { - return `${event.ledger}:${event.opIndex}`; +export function positionIdentity( + event: Pick & { txApplicationOrder?: number; eventIndex?: number }, +): string { + return `${event.ledger}:${event.txApplicationOrder ?? 0}:${event.opIndex}:${event.eventIndex ?? 0}`; } /** Removes duplicate delivery while preserving first-seen order. */ @@ -109,7 +193,14 @@ export function detectReorgs( for (const event of dedupeEvents(replay)) { const previous = byPosition.get(positionIdentity(event)); if (previous && previous.txHash !== event.txHash) { - replacements.push({ ledger: event.ledger, opIndex: event.opIndex, oldTxHash: previous.txHash, newTxHash: event.txHash }); + replacements.push({ + ledger: event.ledger, + txApplicationOrder: event.txApplicationOrder ?? 0, + opIndex: event.opIndex, + eventIndex: event.eventIndex ?? 0, + oldTxHash: previous.txHash, + newTxHash: event.txHash, + }); } } return replacements; @@ -143,7 +234,9 @@ export function assertCheckpointDecision(decision: CheckpointDecision): number { export function validateRecoveryEvent(event: RecoveryEvent): void { if (!Number.isSafeInteger(event.ledger) || event.ledger < 0) throw new Error("event ledger must be a non-negative safe integer"); + if (event.txApplicationOrder !== undefined && (!Number.isSafeInteger(event.txApplicationOrder) || event.txApplicationOrder < 0)) throw new Error("event transaction application order must be non-negative"); if (!Number.isSafeInteger(event.opIndex) || event.opIndex < 0) throw new Error("event operation index must be non-negative"); + if (event.eventIndex !== undefined && (!Number.isSafeInteger(event.eventIndex) || event.eventIndex < 0)) throw new Error("event index must be non-negative"); if (event.txHash.trim().length === 0) throw new Error("event transaction hash is required"); } @@ -151,7 +244,11 @@ export function validateRecoveryEvent(event: RecoveryEvent): void { export function normalizeRecoveryBatch(events: RecoveryEvent[]): RecoveryEvent[] { for (const event of events) validateRecoveryEvent(event); return dedupeEvents(events).sort((left, right) => - left.ledger - right.ledger || left.opIndex - right.opIndex || left.txHash.localeCompare(right.txHash), + left.ledger - right.ledger || + (left.txApplicationOrder ?? 0) - (right.txApplicationOrder ?? 0) || + left.opIndex - right.opIndex || + (left.eventIndex ?? 0) - (right.eventIndex ?? 0) || + left.txHash.localeCompare(right.txHash), ); } diff --git a/src/services/indexerService.ts b/src/services/indexerService.ts index d7217e39..be1670f0 100644 --- a/src/services/indexerService.ts +++ b/src/services/indexerService.ts @@ -2,7 +2,19 @@ import { rpc } from "@stellar/stellar-sdk"; import { env } from "../config/env"; import { logger } from "../config/logger"; import { getPool } from "../db/client"; -import { dedupeEvents, type RecoveryEvent } from "./indexerRecovery"; +import { + decodeEventId, + dedupeEvents, + formatEventId, + type DecodedEventIdentity, + type RecoveryEvent, +} from "./indexerRecovery"; +import { + indexerEventsInsertedTotal, + indexerReorgsDetectedTotal, +} from "../metrics/registry"; + +export { decodeEventId, formatEventId, type DecodedEventIdentity }; export const INDEXER_CURSOR_ID = 1; @@ -12,10 +24,14 @@ export interface LedgerGap { } export interface IndexerEventInput { + id?: string; ledger: number; txHash: string; + txApplicationOrder?: number; opIndex: number; + eventIndex?: number; eventType?: string; + marketId?: string | null; payload?: unknown; } @@ -31,10 +47,20 @@ export interface SorobanRpcClient { } export function parseEventOpIndex(eventId: string, fallback: number): number { - const parts = eventId.split("-"); - const last = parts[parts.length - 1]; - const parsed = Number.parseInt(last ?? "", 10); - return Number.isFinite(parsed) ? parsed : fallback; + const decoded = decodeEventId(eventId); + return decoded ? decoded.opIndex : fallback; +} + +export function extractMarketId(event: IndexerEventInput): string | null { + if (event.marketId) { + return event.marketId; + } + if (event.payload && typeof event.payload === "object") { + const p = event.payload as Record; + if (typeof p.marketId === "string") return p.marketId; + if (typeof p.market_id === "string") return p.market_id; + } + return null; } export function createSorobanRpcClient(): SorobanRpcClient { @@ -49,7 +75,6 @@ export function createSorobanRpcClient(): SorobanRpcClient { async getEvents(startLedger: number, endLedger: number): Promise { const collected: IndexerEventInput[] = []; let cursor: string | undefined; - let fallbackIndex = 0; while (true) { const response = await server.getEvents({ @@ -73,15 +98,25 @@ export function createSorobanRpcClient(): SorobanRpcClient { return collected; } if (event.ledger >= startLedger) { + const decoded = decodeEventId(event.id); + if (!decoded) { + logger.warn( + { eventId: event.id, txHash: event.txHash, ledger: event.ledger }, + "indexer skipping event with malformed event id", + ); + continue; + } collected.push({ + id: event.id, ledger: event.ledger, txHash: event.txHash, - opIndex: parseEventOpIndex(event.id, fallbackIndex), + txApplicationOrder: decoded.txApplicationOrder, + opIndex: decoded.opIndex, + eventIndex: decoded.eventIndex, eventType: event.type, payload: event.value, }); } - fallbackIndex += 1; } const lastEvent = response.events[response.events.length - 1]; @@ -234,59 +269,163 @@ export class IndexerService { * a different transaction at the same ledger/op position replaces the old * canonical event and resets prediction state tied to the stale transaction. */ + /** + * Persist an at-least-once RPC batch. Exact duplicates are ignored, while + * a different transaction at the same ledger position (ledger, txApplicationOrder, opIndex, eventIndex) + * replaces the old canonical event and resets prediction state tied to the stale transaction atomically. + */ async persistEventsWithRecovery(events: IndexerEventInput[]): Promise { if (events.length === 0) { return { inserted: 0, replacements: 0, repairedPredictions: 0 }; } + const validEvents: IndexerEventInput[] = []; + for (const raw of events) { + if (raw.id) { + const decoded = decodeEventId(raw.id); + if (!decoded) { + logger.warn( + { eventId: raw.id, txHash: raw.txHash, ledger: raw.ledger }, + "indexer rejecting event with malformed event id", + ); + continue; + } + validEvents.push({ + ...raw, + ledger: raw.ledger ?? decoded.ledger, + txApplicationOrder: raw.txApplicationOrder ?? decoded.txApplicationOrder, + opIndex: raw.opIndex ?? decoded.opIndex, + eventIndex: raw.eventIndex ?? decoded.eventIndex, + }); + } else { + validEvents.push({ + ...raw, + txApplicationOrder: raw.txApplicationOrder ?? 0, + opIndex: raw.opIndex ?? 0, + eventIndex: raw.eventIndex ?? 0, + }); + } + } + + if (validEvents.length === 0) { + return { inserted: 0, replacements: 0, repairedPredictions: 0 }; + } + const pool = getPool(); let inserted = 0; let replacements = 0; let repairedPredictions = 0; - const uniqueEvents = dedupeEvents(events as RecoveryEvent[]); + const uniqueEvents = dedupeEvents(validEvents as RecoveryEvent[]) as IndexerEventInput[]; for (const event of uniqueEvents) { - const existing = await pool.query<{ tx_hash: string }>( - `SELECT tx_hash FROM indexer_events - WHERE ledger = $1 AND op_index = $2 AND canonical = TRUE - AND tx_hash <> $3 LIMIT 1`, - [event.ledger, event.opIndex, event.txHash], - ); - if ((existing.rows ?? []).length > 0) { - const oldTxHash = existing.rows[0].tx_hash; - await pool.query( - `UPDATE indexer_events SET canonical = FALSE - WHERE ledger = $1 AND op_index = $2 AND tx_hash = $3`, - [event.ledger, event.opIndex, oldTxHash], - ); - await pool.query( - `INSERT INTO indexer_reorgs (ledger, op_index, old_tx_hash, new_tx_hash) - VALUES ($1, $2, $3, $4)`, - [event.ledger, event.opIndex, oldTxHash, event.txHash], + const client = typeof pool.connect === "function" ? await pool.connect() : pool; + try { + await client.query("BEGIN"); + + const txAppOrder = event.txApplicationOrder ?? 0; + const opIndex = event.opIndex ?? 0; + const eventIndex = event.eventIndex ?? 0; + + // Detect canonical conflict at the same ledger position using SELECT ... FOR UPDATE + const existing = await client.query<{ id: string; tx_hash: string }>( + `SELECT id, tx_hash FROM indexer_events + WHERE ledger = $1 AND tx_application_order = $2 AND op_index = $3 AND event_index = $4 AND canonical = TRUE + FOR UPDATE`, + [event.ledger, txAppOrder, opIndex, eventIndex], ); - const repaired = await pool.query( - `UPDATE predictions SET status = 'pending', result = NULL, last_error = NULL, - confirm_attempts = 0 - WHERE tx_hash = $1 AND status IN ('confirmed', 'failed')`, - [oldTxHash], + + const existingRows = existing?.rows ?? []; + if (existingRows.length > 0) { + const oldRow = existingRows[0]; + if (oldRow.tx_hash !== event.txHash) { + // Reorg replacement: same position, different tx hash + await client.query( + `UPDATE indexer_events SET canonical = FALSE + WHERE id = $1`, + [oldRow.id], + ); + + const repaired = await client.query( + `UPDATE predictions SET status = 'pending', result = NULL, last_error = NULL, + confirm_attempts = 0 + WHERE tx_hash = $1 AND status IN ('confirmed', 'failed')`, + [oldRow.tx_hash], + ); + const repairedCount = repaired.rowCount ?? 0; + + await client.query( + `INSERT INTO indexer_reorgs ( + ledger, tx_application_order, op_index, event_index, old_tx_hash, new_tx_hash, repaired_predictions, status + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'detected')`, + [ + event.ledger, + txAppOrder, + opIndex, + eventIndex, + oldRow.tx_hash, + event.txHash, + repairedCount, + ], + ); + + logger.warn( + { + ledger: event.ledger, + txApplicationOrder: txAppOrder, + opIndex, + eventIndex, + oldTxHash: oldRow.tx_hash, + newTxHash: event.txHash, + repairedPredictions: repairedCount, + }, + "indexer reorg detected and predictions repaired", + ); + + indexerReorgsDetectedTotal.inc(); + replacements += 1; + repairedPredictions += repairedCount; + } + } + + const marketId = extractMarketId(event); + const payloadJson = event.payload === undefined ? null : JSON.stringify(event.payload); + + const result = await client.query( + `INSERT INTO indexer_events ( + ledger, tx_hash, tx_application_order, op_index, event_index, event_type, market_id, data, canonical + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9) + ON CONFLICT (ledger, tx_hash, op_index, event_index) DO NOTHING + RETURNING id`, + [ + event.ledger, + event.txHash, + txAppOrder, + opIndex, + eventIndex, + event.eventType ?? "unknown", + marketId, + payloadJson, + true, + ], ); - replacements += 1; - repairedPredictions += repaired.rowCount ?? 0; + + await client.query("COMMIT"); + + const rowCount = result.rowCount ?? 0; + inserted += rowCount; + if (rowCount > 0) { + indexerEventsInsertedTotal.inc(rowCount); + } + } catch (err) { + await client.query("ROLLBACK"); + throw err; + } finally { + if ("release" in client && typeof client.release === "function") { + client.release(); + } } - const result = await pool.query( - `INSERT INTO indexer_events (ledger, tx_hash, op_index, event_type, data, canonical) - VALUES ($1, $2, $3, $4, $5::jsonb) - ON CONFLICT (ledger, tx_hash, op_index) DO NOTHING - RETURNING id`, - [ - event.ledger, - event.txHash, - event.opIndex, - event.eventType ?? "unknown", - event.payload === undefined ? null : JSON.stringify(event.payload), - ], - ); - inserted += result.rowCount ?? 0; } return { inserted, replacements, repairedPredictions }; diff --git a/tests/indexerGapScan.test.ts b/tests/indexerGapScan.test.ts index 6b6dbab1..50e1772f 100644 --- a/tests/indexerGapScan.test.ts +++ b/tests/indexerGapScan.test.ts @@ -100,7 +100,7 @@ describe("IndexerService.backfillRange", () => { ]); expect(poolQuery).toHaveBeenCalledWith( - expect.stringContaining("ON CONFLICT (ledger, tx_hash, op_index) DO NOTHING"), + expect.stringContaining("ON CONFLICT (ledger, tx_hash, op_index, event_index) DO NOTHING"), expect.any(Array), ); }); diff --git a/tests/indexerRecovery.test.ts b/tests/indexerRecovery.test.ts index e208a2c6..85821e0c 100644 --- a/tests/indexerRecovery.test.ts +++ b/tests/indexerRecovery.test.ts @@ -2,10 +2,12 @@ import { assertCheckpointDecision, chunkReplayRanges, decideCheckpoint, + decodeEventId, dedupeEvents, detectReorgs, eventIdentity, findUnobservedLedgers, + formatEventId, groupLedgerRanges, IndexerRecoveryPlanner, mergeObservations, @@ -15,13 +17,63 @@ import { validateRecoveryEvent, } from "../src/services/indexerRecovery"; -const event = (ledger: number, txHash = `tx-${ledger}`, opIndex = 0) => ({ ledger, txHash, opIndex }); +const event = ( + ledger: number, + txHash = `tx-${ledger}`, + opIndex = 0, + txApplicationOrder = 0, + eventIndex = 0, +) => ({ + ledger, + txHash, + txApplicationOrder, + opIndex, + eventIndex, +}); + +describe("TOID decoding and encoding", () => { + it("decodes a valid Soroban RPC event id into positional components", () => { + // ledger 1000, txApplicationOrder 4, opIndex 2, eventIndex 1 + const eventId = formatEventId(1000, 4, 2, 1); + const decoded = decodeEventId(eventId); + expect(decoded).toEqual({ + ledger: 1000, + txApplicationOrder: 4, + opIndex: 2, + eventIndex: 1, + }); + }); + + it("handles zero values for all components", () => { + const eventId = formatEventId(0, 0, 0, 0); + const decoded = decodeEventId(eventId); + expect(decoded).toEqual({ + ledger: 0, + txApplicationOrder: 0, + opIndex: 0, + eventIndex: 0, + }); + }); + + it("rejects malformed event ids gracefully", () => { + expect(decodeEventId("")).toBeNull(); + expect(decodeEventId(null as unknown as string)).toBeNull(); + expect(decodeEventId(123 as unknown as string)).toBeNull(); + expect(decodeEventId("not-an-event-id")).toBeNull(); + expect(decodeEventId("12345")).toBeNull(); + expect(decodeEventId("123-456-789")).toBeNull(); + expect(decodeEventId("abc-0000000000")).toBeNull(); + expect(decodeEventId("0000000000-xyz")).toBeNull(); + expect(decodeEventId("-1-0")).toBeNull(); + expect(decodeEventId("100-9999999999999999999999999999999999999")).toBeNull(); + }); +}); describe("indexer recovery identities", () => { it("separates event identity from ledger position", () => { - expect(eventIdentity(event(10, "a", 1))).toBe("10:a:1"); - expect(positionIdentity(event(10, "b", 1))).toBe("10:1"); - expect(eventIdentity(event(10, "a", 1))).not.toBe(eventIdentity(event(10, "b", 1))); + expect(eventIdentity(event(10, "a", 1, 0, 2))).toBe("10:a:1:2"); + expect(positionIdentity(event(10, "b", 1, 3, 2))).toBe("10:3:1:2"); + expect(eventIdentity(event(10, "a", 1, 0, 2))).not.toBe(eventIdentity(event(10, "b", 1, 0, 2))); }); it("deduplicates exact redelivery without dropping distinct operations", () => { @@ -56,6 +108,9 @@ describe("indexer recovery identities", () => { ]); expect(() => validateRecoveryEvent(event(-1))).toThrow("non-negative"); expect(() => validateRecoveryEvent({ ledger: 1, txHash: "", opIndex: 0 })).toThrow("transaction hash"); + expect(() => validateRecoveryEvent({ ledger: 1, txHash: "tx", opIndex: -1 })).toThrow("non-negative"); + expect(() => validateRecoveryEvent({ ledger: 1, txHash: "tx", opIndex: 0, txApplicationOrder: -1 })).toThrow("non-negative"); + expect(() => validateRecoveryEvent({ ledger: 1, txHash: "tx", opIndex: 0, eventIndex: -1 })).toThrow("non-negative"); }); it("merges paginated observations while preserving empty-ledger evidence", () => { @@ -84,13 +139,13 @@ describe("indexer recovery identities", () => { describe("reorg detection", () => { it("detects replacement at the same ledger and operation", () => { expect(detectReorgs([event(20, "old", 2)], [event(20, "new", 2)])).toEqual([ - { ledger: 20, opIndex: 2, oldTxHash: "old", newTxHash: "new" }, + { ledger: 20, txApplicationOrder: 0, opIndex: 2, eventIndex: 0, oldTxHash: "old", newTxHash: "new" }, ]); }); it("ignores same-event replay and reports multiple replacements", () => { expect(detectReorgs([event(20, "old", 2), event(21, "keep")], [event(20, "old", 2), event(21, "new"), event(22)])).toEqual([ - { ledger: 21, opIndex: 0, oldTxHash: "keep", newTxHash: "new" }, + { ledger: 21, txApplicationOrder: 0, opIndex: 0, eventIndex: 0, oldTxHash: "keep", newTxHash: "new" }, ]); }); diff --git a/tests/indexerService.recovery.test.ts b/tests/indexerService.recovery.test.ts index a2a30a90..2ba90e21 100644 --- a/tests/indexerService.recovery.test.ts +++ b/tests/indexerService.recovery.test.ts @@ -1,16 +1,61 @@ -import { IndexerService } from "../src/services/indexerService"; +import { + IndexerService, + extractMarketId, + formatEventId, + createSorobanRpcClient, +} from "../src/services/indexerService"; +import { rpc } from "@stellar/stellar-sdk"; const query = jest.fn(); +const release = jest.fn(); +const client = { query, release }; +const pool = { + query, + connect: jest.fn().mockResolvedValue(client), +}; jest.mock("../src/db/client", () => ({ - getPool: () => ({ query }), + getPool: () => pool, })); +jest.mock("@stellar/stellar-sdk", () => { + return { + rpc: { + Server: jest.fn(), + }, + }; +}); + describe("IndexerService durable event recovery", () => { let service: IndexerService; beforeEach(() => { query.mockReset(); + release.mockReset(); + pool.connect.mockReset().mockResolvedValue(client); + + query.mockImplementation(async (sql: string) => { + if (sql === "BEGIN" || sql === "COMMIT" || sql === "ROLLBACK") { + return { rowCount: 0, rows: [] }; + } + if (sql.includes("SELECT id, tx_hash FROM indexer_events")) { + return { rows: [] }; + } + if (sql.includes("INSERT INTO indexer_events")) { + return { rowCount: 1, rows: [{ id: "ev-1" }] }; + } + if (sql.includes("UPDATE indexer_events")) { + return { rowCount: 1, rows: [] }; + } + if (sql.includes("UPDATE predictions")) { + return { rowCount: 0, rows: [] }; + } + if (sql.includes("INSERT INTO indexer_reorgs")) { + return { rowCount: 1, rows: [] }; + } + return { rowCount: 0, rows: [] }; + }); + service = new IndexerService({ getLatestLedger: jest.fn(), getEvents: jest.fn(), @@ -26,95 +71,402 @@ describe("IndexerService durable event recovery", () => { expect(query).not.toHaveBeenCalled(); }); - it("persists a new canonical event with a safe fallback event type", async () => { - query.mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rowCount: 1 }); - await expect(service.persistEventsWithRecovery([{ ledger: 10, txHash: "tx-10", opIndex: 2, payload: { amount: "5" } }])).resolves.toMatchObject({ inserted: 1, replacements: 0 }); - expect(query).toHaveBeenLastCalledWith( - expect.stringContaining("ON CONFLICT (ledger, tx_hash, op_index) DO NOTHING"), - [10, "tx-10", 2, "unknown", JSON.stringify({ amount: "5" })], + it("Scenario 1: persists a new canonical event binding all columns and populating market_id", async () => { + await expect( + service.persistEventsWithRecovery([ + { + ledger: 10, + txHash: "tx-10", + opIndex: 2, + txApplicationOrder: 1, + eventIndex: 0, + marketId: "market-xyz", + payload: { amount: "5" }, + }, + ]), + ).resolves.toMatchObject({ inserted: 1, replacements: 0 }); + + const insertCall = query.mock.calls.find((call) => + typeof call[0] === "string" && call[0].includes("INSERT INTO indexer_events"), + ); + expect(insertCall).toBeDefined(); + expect(insertCall[0]).toContain( + "ON CONFLICT (ledger, tx_hash, op_index, event_index) DO NOTHING", ); + expect(insertCall[1]).toEqual([ + 10, + "tx-10", + 1, + 2, + 0, + "unknown", + "market-xyz", + JSON.stringify({ amount: "5" }), + true, + ]); }); - it("does not count exact redelivery as a new event", async () => { - query.mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rowCount: 0 }); - const report = await service.persistEventsWithRecovery([{ ledger: 10, txHash: "tx-10", opIndex: 2, eventType: "prediction" }]); + it("Scenario 2: two different txs in same ledger with suffix 0000000000 produce two canonical rows without reorg", async () => { + const eventIdTx1 = formatEventId(100, 1, 0, 0); // suffix 0000000000, txAppOrder 1 + const eventIdTx2 = formatEventId(100, 2, 0, 0); // suffix 0000000000, txAppOrder 2 + + const report = await service.persistEventsWithRecovery([ + { id: eventIdTx1, ledger: 100, txHash: "tx-first", opIndex: 0 }, + { id: eventIdTx2, ledger: 100, txHash: "tx-second", opIndex: 0 }, + ]); + + expect(report).toEqual({ + inserted: 2, + replacements: 0, + repairedPredictions: 0, + }); + + const reorgInserts = query.mock.calls.filter((call) => + typeof call[0] === "string" && call[0].includes("INSERT INTO indexer_reorgs"), + ); + expect(reorgInserts).toHaveLength(0); + }); + + it("Scenario 3: does not count exact redelivery as a new event and creates no reorg", async () => { + query.mockImplementation(async (sql: string) => { + if (sql === "BEGIN" || sql === "COMMIT" || sql === "ROLLBACK") { + return { rowCount: 0, rows: [] }; + } + if (sql.includes("SELECT id, tx_hash FROM indexer_events")) { + return { rows: [{ id: "existing-1", tx_hash: "tx-10" }] }; + } + if (sql.includes("INSERT INTO indexer_events")) { + return { rowCount: 0, rows: [] }; + } + return { rowCount: 0, rows: [] }; + }); + + const report = await service.persistEventsWithRecovery([ + { ledger: 10, txHash: "tx-10", opIndex: 2, eventType: "prediction" }, + ]); expect(report).toEqual({ inserted: 0, replacements: 0, repairedPredictions: 0 }); - expect(query).toHaveBeenCalledTimes(2); }); - it("marks the stale event non-canonical and repairs derived prediction state", async () => { - query - .mockResolvedValueOnce({ rows: [{ tx_hash: "old-tx" }] }) - .mockResolvedValueOnce({ rowCount: 1 }) - .mockResolvedValueOnce({ rowCount: 1 }) - .mockResolvedValueOnce({ rowCount: 3 }) - .mockResolvedValueOnce({ rowCount: 1 }); + it("Scenario 4: marks the stale event non-canonical, inserts reorg row, and repairs predictions on real replacement", async () => { + query.mockImplementation(async (sql: string) => { + if (sql === "BEGIN" || sql === "COMMIT" || sql === "ROLLBACK") { + return { rowCount: 0, rows: [] }; + } + if (sql.includes("SELECT id, tx_hash FROM indexer_events")) { + return { rows: [{ id: "row-old", tx_hash: "old-tx" }] }; + } + if (sql.includes("UPDATE indexer_events SET canonical = FALSE")) { + return { rowCount: 1, rows: [] }; + } + if (sql.includes("UPDATE predictions SET status = 'pending'")) { + return { rowCount: 3, rows: [] }; + } + if (sql.includes("INSERT INTO indexer_reorgs")) { + return { rowCount: 1, rows: [] }; + } + if (sql.includes("INSERT INTO indexer_events")) { + return { rowCount: 1, rows: [{ id: "row-new" }] }; + } + return { rowCount: 0, rows: [] }; + }); - const report = await service.persistEventsWithRecovery([{ ledger: 99, txHash: "new-tx", opIndex: 0, eventType: "settled", payload: { outcome: "yes" } }]); + const report = await service.persistEventsWithRecovery([ + { + ledger: 99, + txHash: "new-tx", + txApplicationOrder: 1, + opIndex: 0, + eventIndex: 0, + eventType: "settled", + payload: { outcome: "yes" }, + }, + ]); expect(report).toEqual({ inserted: 1, replacements: 1, repairedPredictions: 3 }); - expect(query.mock.calls[1][0]).toContain("SET canonical = FALSE"); - expect(query.mock.calls[2][0]).toContain("INSERT INTO indexer_reorgs"); - expect(query.mock.calls[3][0]).toContain("UPDATE predictions SET status = 'pending'"); - expect(query.mock.calls[3][1]).toEqual(["old-tx"]); + + const flipCall = query.mock.calls.find((c) => + typeof c[0] === "string" && c[0].includes("SET canonical = FALSE"), + ); + expect(flipCall).toBeDefined(); + expect(flipCall[1]).toEqual(["row-old"]); + + const reorgCall = query.mock.calls.find((c) => + typeof c[0] === "string" && c[0].includes("INSERT INTO indexer_reorgs"), + ); + expect(reorgCall).toBeDefined(); + expect(reorgCall[1]).toEqual([99, 1, 0, 0, "old-tx", "new-tx", 3]); + + const repairCall = query.mock.calls.find((c) => + typeof c[0] === "string" && c[0].includes("UPDATE predictions SET status = 'pending'"), + ); + expect(repairCall).toBeDefined(); + expect(repairCall[1]).toEqual(["old-tx"]); + }); + + it("Scenario 5: failure injected after canonical flip rolls back the transaction", async () => { + query.mockImplementation(async (sql: string) => { + if (sql === "BEGIN" || sql === "ROLLBACK") { + return { rowCount: 0, rows: [] }; + } + if (sql.includes("SELECT id, tx_hash FROM indexer_events")) { + return { rows: [{ id: "row-old", tx_hash: "old-tx" }] }; + } + if (sql.includes("UPDATE indexer_events SET canonical = FALSE")) { + return { rowCount: 1, rows: [] }; + } + if (sql.includes("UPDATE predictions SET status = 'pending'")) { + throw new Error("DB connection failure during prediction repair"); + } + return { rowCount: 0, rows: [] }; + }); + + await expect( + service.persistEventsWithRecovery([ + { + ledger: 99, + txHash: "new-tx", + txApplicationOrder: 1, + opIndex: 0, + eventIndex: 0, + }, + ]), + ).rejects.toThrow("DB connection failure during prediction repair"); + + expect(query).toHaveBeenCalledWith("ROLLBACK"); + expect(release).toHaveBeenCalled(); + }); + + it("Scenario 6: malformed event id is rejected with logged warning and batch continues", async () => { + const report = await service.persistEventsWithRecovery([ + { id: "invalid-event-id", ledger: 5, txHash: "tx-bad", opIndex: 0 }, + { ledger: 5, txHash: "tx-good", opIndex: 1 }, + ]); + + expect(report.inserted).toBe(1); + const insertCalls = query.mock.calls.filter((c) => + typeof c[0] === "string" && c[0].includes("INSERT INTO indexer_events"), + ); + expect(insertCalls).toHaveLength(1); + expect(insertCalls[0][1][1]).toBe("tx-good"); + }); + + it("Scenario 7: multiple events in one operation are inserted as distinct rows", async () => { + const eventId0 = formatEventId(50, 1, 0, 0); + const eventId1 = formatEventId(50, 1, 0, 1); + const eventId2 = formatEventId(50, 1, 0, 2); + + const report = await service.persistEventsWithRecovery([ + { id: eventId0, ledger: 50, txHash: "tx-shared", opIndex: 0 }, + { id: eventId1, ledger: 50, txHash: "tx-shared", opIndex: 0 }, + { id: eventId2, ledger: 50, txHash: "tx-shared", opIndex: 0 }, + ]); + + expect(report.inserted).toBe(3); + expect(report.replacements).toBe(0); }); it("deduplicates repeated events before performing conflict checks", async () => { - query.mockResolvedValue({ rows: [] }); - query.mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rowCount: 1 }); const report = await service.persistEventsWithRecovery([ { ledger: 7, txHash: "tx", opIndex: 1, eventType: "a" }, { ledger: 7, txHash: "tx", opIndex: 1, eventType: "a" }, ]); expect(report.inserted).toBe(1); - expect(query).toHaveBeenCalledTimes(2); + const selectCalls = query.mock.calls.filter((c) => + typeof c[0] === "string" && c[0].includes("SELECT id, tx_hash FROM indexer_events"), + ); + expect(selectCalls).toHaveLength(1); }); - it("keeps distinct operations at one ledger independent", async () => { - query.mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rowCount: 1 }).mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rowCount: 1 }); - const report = await service.persistEventsWithRecovery([ - { ledger: 12, txHash: "tx", opIndex: 0, eventType: "market" }, - { ledger: 12, txHash: "tx", opIndex: 1, eventType: "prediction" }, - ]); - expect(report).toMatchObject({ inserted: 2, replacements: 0 }); - expect(query).toHaveBeenCalledTimes(4); + it("extractMarketId extracts marketId from payload if omitted at top level", () => { + expect( + extractMarketId({ + ledger: 1, + txHash: "tx", + opIndex: 0, + payload: { marketId: "m-123" }, + }), + ).toBe("m-123"); + expect( + extractMarketId({ + ledger: 1, + txHash: "tx", + opIndex: 0, + payload: { market_id: "m-456" }, + }), + ).toBe("m-456"); + expect( + extractMarketId({ + ledger: 1, + txHash: "tx", + opIndex: 0, + marketId: "direct", + payload: { marketId: "ignored" }, + }), + ).toBe("direct"); + expect( + extractMarketId({ + ledger: 1, + txHash: "tx", + opIndex: 0, + }), + ).toBeNull(); }); it("limits replay to the configured chunk size and advances after all chunks", async () => { - const rpc = { + const rpcMock = { getLatestLedger: jest.fn(), - getEvents: jest.fn() - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]), + getEvents: jest.fn().mockResolvedValueOnce([]).mockResolvedValueOnce([]), }; - service = new IndexerService(rpc); + service = new IndexerService(rpcMock); query.mockResolvedValueOnce({ rowCount: 1 }); await service.backfillRange(1000, 1501); - expect(rpc.getEvents).toHaveBeenCalledWith(900, 1399); - expect(rpc.getEvents).toHaveBeenCalledWith(1400, 1501); - expect(query).toHaveBeenLastCalledWith(expect.stringContaining("INSERT INTO indexer_cursor"), [1, 1501]); + expect(rpcMock.getEvents).toHaveBeenCalledWith(900, 1399); + expect(rpcMock.getEvents).toHaveBeenCalledWith(1400, 1501); + expect(query).toHaveBeenLastCalledWith( + expect.stringContaining("INSERT INTO indexer_cursor"), + [1, 1501], + ); }); it("does not rewind below the configured start ledger", async () => { - const rpc = { getLatestLedger: jest.fn(), getEvents: jest.fn().mockResolvedValue([]) }; - service = new IndexerService(rpc); + const rpcMock = { getLatestLedger: jest.fn(), getEvents: jest.fn().mockResolvedValue([]) }; + service = new IndexerService(rpcMock); query.mockResolvedValueOnce({ rowCount: 1 }); await service.backfillRange(1, 1); - expect(rpc.getEvents).toHaveBeenCalledWith(0, 1); + expect(rpcMock.getEvents).toHaveBeenCalledWith(0, 1); }); it("returns no replay work for an inverted range", async () => { - const rpc = { getLatestLedger: jest.fn(), getEvents: jest.fn() }; - service = new IndexerService(rpc); + const rpcMock = { getLatestLedger: jest.fn(), getEvents: jest.fn() }; + service = new IndexerService(rpcMock); await service.backfillRange(20, 19); - expect(rpc.getEvents).not.toHaveBeenCalled(); + expect(rpcMock.getEvents).not.toHaveBeenCalled(); expect(query).not.toHaveBeenCalled(); }); it("preserves checkpoint monotonicity through the SQL upsert", async () => { query.mockResolvedValueOnce({ rowCount: 1 }); await service.advanceCursor(500); - expect(query).toHaveBeenCalledWith(expect.stringContaining("GREATEST(indexer_cursor.last_ledger"), [1, 500]); + expect(query).toHaveBeenCalledWith( + expect.stringContaining("GREATEST(indexer_cursor.last_ledger"), + [1, 500], + ); + }); + + it("finds missing ledgers and max indexed ledger", async () => { + query.mockResolvedValueOnce({ rows: [{ missing_ledger: 5 }, { missing_ledger: 6 }] }); + const missing = await service.findMissingLedgers(1, 10); + expect(missing).toEqual([5, 6]); + + expect(await service.findMissingLedgers(10, 5)).toEqual([]); + + query.mockResolvedValueOnce({ rows: [{ max_ledger: 42 }] }); + const maxLedger = await service.getMaxIndexedLedger(1, 50); + expect(maxLedger).toBe(42); + + query.mockResolvedValueOnce({ rows: [] }); + expect(await service.getMaxIndexedLedger(1, 50)).toBeNull(); + }); + + it("getCursor returns start ledger if no cursor row exists", async () => { + query.mockResolvedValueOnce({ rows: [] }); + const cursor = await service.getCursor(); + expect(cursor).toBeDefined(); + + query.mockResolvedValueOnce({ rows: [{ last_ledger: 55 }] }); + expect(await service.getCursor()).toBe(55); + }); + + it("pollOnce coordinates getCursor, getChainTip, fetchEvents and advanceCursor", async () => { + const rpcMock = { + getLatestLedger: jest.fn().mockResolvedValue(100), + getEvents: jest.fn().mockResolvedValue([]), + }; + service = new IndexerService(rpcMock); + query.mockResolvedValue({ rows: [{ last_ledger: 95 }] }); + + const tip = await service.pollOnce(); + expect(tip).toBe(100); + expect(rpcMock.getEvents).toHaveBeenCalled(); + + // When cursor is already ahead of tip + rpcMock.getLatestLedger.mockResolvedValue(50); + query.mockResolvedValue({ rows: [{ last_ledger: 1000 }] }); + const sameCursor = await service.pollOnce(); + expect(sameCursor).toBe(1000); + }); + + it("createSorobanRpcClient pagination and event decoding", async () => { + const getEventsMock = jest.fn(); + const getLatestLedgerMock = jest.fn().mockResolvedValue({ sequence: 500 }); + + (rpc.Server as unknown as jest.Mock).mockImplementation(() => ({ + getEvents: getEventsMock, + getLatestLedger: getLatestLedgerMock, + })); + + const clientRpc = createSorobanRpcClient(); + expect(await clientRpc.getLatestLedger()).toBe(500); + + const validId = formatEventId(10, 1, 0, 0); + getEventsMock + .mockResolvedValueOnce({ + events: [ + { id: validId, ledger: 10, txHash: "tx-1", type: "contract", value: { a: 1 }, pagingToken: "p1" }, + { id: "bad-id", ledger: 10, txHash: "tx-bad", type: "contract", value: null, pagingToken: "p2" }, + { id: formatEventId(20, 1, 0, 0), ledger: 20, txHash: "tx-over", type: "contract", value: null }, + ], + }); + + const events = await clientRpc.getEvents(10, 15); + expect(events).toHaveLength(1); + expect(events[0].txHash).toBe("tx-1"); + + // Test loop termination when lastEvent.ledger >= endLedger + getEventsMock.mockResolvedValueOnce({ + events: [ + { id: formatEventId(15, 1, 0, 0), ledger: 15, txHash: "tx-end", type: "contract", value: null }, + ], + }); + const eventsEnd = await clientRpc.getEvents(10, 15); + expect(eventsEnd).toHaveLength(1); + + // Test loop termination when pagingToken is absent + getEventsMock.mockResolvedValueOnce({ + events: [ + { id: formatEventId(12, 1, 0, 0), ledger: 12, txHash: "tx-no-token", type: "contract", value: null }, + ], + }); + const eventsNoToken = await clientRpc.getEvents(10, 15); + expect(eventsNoToken).toHaveLength(1); + + // Test loop termination when response has 0 events + getEventsMock.mockResolvedValueOnce({ events: [] }); + const eventsEmpty = await clientRpc.getEvents(10, 15); + expect(eventsEmpty).toHaveLength(0); + }); + + it("covers parseEventOpIndex, mergeGapRanges empty, and all-malformed batch", async () => { + const { parseEventOpIndex, mergeGapRanges } = await import("../src/services/indexerService"); + const validId = formatEventId(10, 1, 3, 0); + expect(parseEventOpIndex(validId, 99)).toBe(3); + expect(parseEventOpIndex("invalid-event-id", 99)).toBe(99); + + expect(mergeGapRanges([])).toEqual([]); + + const report = await service.persistEventsWithRecovery([ + { id: "malformed-1", ledger: 1, txHash: "bad", opIndex: 0 }, + ]); + expect(report).toEqual({ inserted: 0, replacements: 0, repairedPredictions: 0 }); + + expect( + extractMarketId({ + ledger: 1, + txHash: "tx", + opIndex: 0, + payload: "string-payload-not-object", + }), + ).toBeNull(); }); }); diff --git a/tests/integration/indexerPersistence.test.ts b/tests/integration/indexerPersistence.test.ts new file mode 100644 index 00000000..a338aeff --- /dev/null +++ b/tests/integration/indexerPersistence.test.ts @@ -0,0 +1,298 @@ +import { pool, closeDb } from "../../src/db/client"; +import { + IndexerService, + formatEventId, +} from "../../src/services/indexerService"; + +describe("Indexer Persistence Integration Tests", () => { + let service: IndexerService; + + beforeAll(async () => { + service = new IndexerService({ + getLatestLedger: jest.fn().mockResolvedValue(100), + getEvents: jest.fn().mockResolvedValue([]), + }); + }); + + afterAll(async () => { + await closeDb(); + }); + + beforeEach(async () => { + await pool.query("DELETE FROM indexer_reorgs"); + await pool.query("DELETE FROM indexer_events"); + await pool.query("DELETE FROM predictions"); + await pool.query("DELETE FROM markets"); + await pool.query("DELETE FROM users"); + }); + + async function seedUser(address: string): Promise { + const res = await pool.query( + "INSERT INTO users (stellar_address) VALUES ($1) RETURNING id", + [address], + ); + return res.rows[0].id; + } + + async function seedMarket(marketId: string): Promise { + await pool.query( + `INSERT INTO markets (id, question, status, resolution_time, indexed_ledger, archived, version, featured) + VALUES ($1, 'Test Market?', 'active', NOW() + INTERVAL '1 day', 1, false, 1, false)`, + [marketId], + ); + } + + async function seedPrediction( + txHash: string, + marketId: string, + userId: string, + status = "confirmed", + ): Promise { + const res = await pool.query( + `INSERT INTO predictions (market_id, user_id, outcome, amount, status, tx_hash) + VALUES ($1, $2, 'YES', '10.0', $3, $4) + RETURNING id`, + [marketId, userId, status, txHash], + ); + return res.rows[0].id; + } + + it("Scenario 1: Insert one event against real Postgres - canonical is true and market_id is populated", async () => { + const userId = await seedUser("GA1111111111111111111111111111111111111111111111111111111111"); + const marketId = "market-persistent-1"; + await seedMarket(marketId); + await seedPrediction("tx-1", marketId, userId, "confirmed"); + + const eventId = formatEventId(100, 1, 0, 0); + const report = await service.persistEventsWithRecovery([ + { + id: eventId, + ledger: 100, + txHash: "tx-1", + eventType: "bet_placed", + marketId, + payload: { amount: "10" }, + }, + ]); + + expect(report.inserted).toBe(1); + expect(report.replacements).toBe(0); + + const rows = await pool.query( + `SELECT ledger, tx_hash, tx_application_order, op_index, event_index, event_type, market_id, canonical + FROM indexer_events WHERE tx_hash = $1`, + ["tx-1"], + ); + + expect(rows.rows).toHaveLength(1); + const row = rows.rows[0]; + expect(row.canonical).toBe(true); + expect(row.market_id).toBe(marketId); + expect(row.ledger).toBe(100); + expect(row.tx_application_order).toBe(1); + expect(row.op_index).toBe(0); + expect(row.event_index).toBe(0); + }); + + it("Scenario 2: Two different txs in same ledger, both with event-id suffix 0000000000 - two canonical rows, 0 reorgs, predictions untouched", async () => { + const u1 = await seedUser("GA2222222222222222222222222222222222222222222222222222222222"); + const u2 = await seedUser("GA3333333333333333333333333333333333333333333333333333333333"); + const marketId = "market-shared-ledger"; + await seedMarket(marketId); + const p1 = await seedPrediction("tx-bet-1", marketId, u1, "confirmed"); + const p2 = await seedPrediction("tx-bet-2", marketId, u2, "confirmed"); + + // Both have suffix 0000000000, but distinct TOID txApplicationOrder (1 vs 2) + const eventId1 = formatEventId(200, 1, 0, 0); // suffix 0000000000 + const eventId2 = formatEventId(200, 2, 0, 0); // suffix 0000000000 + + const report = await service.persistEventsWithRecovery([ + { id: eventId1, ledger: 200, txHash: "tx-bet-1", marketId }, + { id: eventId2, ledger: 200, txHash: "tx-bet-2", marketId }, + ]); + + expect(report.inserted).toBe(2); + expect(report.replacements).toBe(0); + expect(report.repairedPredictions).toBe(0); + + const events = await pool.query( + `SELECT tx_hash, canonical FROM indexer_events WHERE ledger = 200 ORDER BY tx_application_order`, + ); + expect(events.rows).toHaveLength(2); + expect(events.rows[0].canonical).toBe(true); + expect(events.rows[1].canonical).toBe(true); + + const reorgs = await pool.query("SELECT * FROM indexer_reorgs WHERE ledger = 200"); + expect(reorgs.rows).toHaveLength(0); + + const predRows = await pool.query( + `SELECT id, status FROM predictions WHERE id IN ($1, $2) ORDER BY tx_hash`, + [p1, p2], + ); + expect(predRows.rows[0].status).toBe("confirmed"); + expect(predRows.rows[1].status).toBe("confirmed"); + }); + + it("Scenario 3: The same page delivered twice - no extra rows, no reorg rows", async () => { + const eventId = formatEventId(300, 1, 0, 0); + const batch = [{ id: eventId, ledger: 300, txHash: "tx-idempotent" }]; + + const first = await service.persistEventsWithRecovery(batch); + expect(first.inserted).toBe(1); + expect(first.replacements).toBe(0); + + const second = await service.persistEventsWithRecovery(batch); + expect(second.inserted).toBe(0); + expect(second.replacements).toBe(0); + expect(second.repairedPredictions).toBe(0); + + const events = await pool.query( + "SELECT * FROM indexer_events WHERE ledger = 300", + ); + expect(events.rows).toHaveLength(1); + + const reorgs = await pool.query( + "SELECT * FROM indexer_reorgs WHERE ledger = 300", + ); + expect(reorgs.rows).toHaveLength(0); + }); + + it("Scenario 4: A real replacement: same TOID position, different tx hash - marks old row non-canonical, 1 reorg row, resets prediction", async () => { + const userId = await seedUser("GA4444444444444444444444444444444444444444444444444444444444"); + const marketId = "market-reorg"; + await seedMarket(marketId); + await seedPrediction("tx-fork-a", marketId, userId, "confirmed"); + + const eventId = formatEventId(400, 1, 0, 0); + + // Initial event canonicalized + await service.persistEventsWithRecovery([ + { id: eventId, ledger: 400, txHash: "tx-fork-a", marketId }, + ]); + + // Chain reorg at same TOID position with new transaction + const reorgReport = await service.persistEventsWithRecovery([ + { id: eventId, ledger: 400, txHash: "tx-fork-b", marketId }, + ]); + + expect(reorgReport.inserted).toBe(1); + expect(reorgReport.replacements).toBe(1); + expect(reorgReport.repairedPredictions).toBe(1); + + const oldRow = await pool.query( + "SELECT canonical FROM indexer_events WHERE tx_hash = 'tx-fork-a'", + ); + expect(oldRow.rows[0].canonical).toBe(false); + + const newRow = await pool.query( + "SELECT canonical FROM indexer_events WHERE tx_hash = 'tx-fork-b'", + ); + expect(newRow.rows[0].canonical).toBe(true); + + const reorgRows = await pool.query( + "SELECT old_tx_hash, new_tx_hash, repaired_predictions FROM indexer_reorgs WHERE ledger = 400", + ); + expect(reorgRows.rows).toHaveLength(1); + expect(reorgRows.rows[0].old_tx_hash).toBe("tx-fork-a"); + expect(reorgRows.rows[0].new_tx_hash).toBe("tx-fork-b"); + expect(reorgRows.rows[0].repaired_predictions).toBe(1); + + const pred = await pool.query( + "SELECT status, result FROM predictions WHERE tx_hash = 'tx-fork-a'", + ); + expect(pred.rows[0].status).toBe("pending"); + expect(pred.rows[0].result).toBeNull(); + }); + + it("Scenario 5: Failure injected after canonical flip - transaction rolled back, old row remains canonical, prediction unchanged", async () => { + const userId = await seedUser("GA5555555555555555555555555555555555555555555555555555555555"); + const marketId = "market-rollback"; + await seedMarket(marketId); + await seedPrediction("tx-orig", marketId, userId, "confirmed"); + + const eventId = formatEventId(500, 1, 0, 0); + await service.persistEventsWithRecovery([ + { id: eventId, ledger: 500, txHash: "tx-orig", marketId }, + ]); + + // Create a temporary trigger or failing constraint to inject failure during prediction update or insert + await pool.query(` + CREATE OR REPLACE FUNCTION fail_indexer_reorgs() RETURNS TRIGGER AS $$ + BEGIN + RAISE EXCEPTION 'injected failure during reorg insert'; + END; + $$ LANGUAGE plpgsql; + + CREATE TRIGGER trigger_fail_reorg + BEFORE INSERT ON indexer_reorgs + FOR EACH ROW EXECUTE FUNCTION fail_indexer_reorgs(); + `); + + try { + await expect( + service.persistEventsWithRecovery([ + { id: eventId, ledger: 500, txHash: "tx-competing", marketId }, + ]), + ).rejects.toThrow("injected failure during reorg insert"); + } finally { + await pool.query("DROP TRIGGER IF EXISTS trigger_fail_reorg ON indexer_reorgs"); + await pool.query("DROP FUNCTION IF EXISTS fail_indexer_reorgs"); + } + + // Verify rollback: old row is STILL canonical, prediction is STILL confirmed + const origEvent = await pool.query( + "SELECT canonical FROM indexer_events WHERE tx_hash = 'tx-orig'", + ); + expect(origEvent.rows[0].canonical).toBe(true); + + const competingEvent = await pool.query( + "SELECT * FROM indexer_events WHERE tx_hash = 'tx-competing'", + ); + expect(competingEvent.rows).toHaveLength(0); + + const reorgRows = await pool.query( + "SELECT * FROM indexer_reorgs WHERE ledger = 500", + ); + expect(reorgRows.rows).toHaveLength(0); + + const pred = await pool.query( + "SELECT status FROM predictions WHERE tx_hash = 'tx-orig'", + ); + expect(pred.rows[0].status).toBe("confirmed"); + }); + + it("Scenario 6: Malformed event id (no TOID prefix) - event rejected with warning, batch continues", async () => { + const validId = formatEventId(600, 1, 0, 0); + const report = await service.persistEventsWithRecovery([ + { id: "invalid-toid", ledger: 600, txHash: "tx-malformed" }, + { id: validId, ledger: 600, txHash: "tx-valid" }, + ]); + + expect(report.inserted).toBe(1); + + const events = await pool.query( + "SELECT tx_hash FROM indexer_events WHERE ledger = 600", + ); + expect(events.rows).toHaveLength(1); + expect(events.rows[0].tx_hash).toBe("tx-valid"); + }); + + it("Scenario 7: Multiple events in one operation (event index 0, 1, 2) - three distinct rows", async () => { + const id0 = formatEventId(700, 1, 0, 0); + const id1 = formatEventId(700, 1, 0, 1); + const id2 = formatEventId(700, 1, 0, 2); + + const report = await service.persistEventsWithRecovery([ + { id: id0, ledger: 700, txHash: "tx-multi-event", opIndex: 0 }, + { id: id1, ledger: 700, txHash: "tx-multi-event", opIndex: 0 }, + { id: id2, ledger: 700, txHash: "tx-multi-event", opIndex: 0 }, + ]); + + expect(report.inserted).toBe(3); + + const rows = await pool.query( + "SELECT event_index FROM indexer_events WHERE ledger = 700 AND tx_hash = 'tx-multi-event' ORDER BY event_index", + ); + expect(rows.rows.map((r) => r.event_index)).toEqual([0, 1, 2]); + }); +});