fix(indexer): bind canonical column and derive event identity from TOID

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>-<event_index>`:
+- **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.

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");

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(),
 });

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;
       }
 

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",

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:
+ *   <toid>-<event_index>
+ *
+ * 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<RecoveryEvent, "ledger" | "txHash" | "opIndex">): string {
-  return `${event.ledger}:${event.txHash}:${event.opIndex}`;
+export function eventIdentity(
+  event: Pick<RecoveryEvent, "ledger" | "txHash" | "opIndex"> & { eventIndex?: number },
+): string {
+  return `${event.ledger}:${event.txHash}:${event.opIndex}:${event.eventIndex ?? 0}`;
 }
 
-export function positionIdentity(event: Pick<RecoveryEvent, "ledger" | "opIndex">): string {
-  return `${event.ledger}:${event.opIndex}`;
+export function positionIdentity(
+  event: Pick<RecoveryEvent, "ledger" | "opIndex"> & { 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,15 +234,21 @@ 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");
 }
 
 /** Validates and normalizes an RPC batch before it can influence a checkpoint. */
 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),
   );
 }