From 3389a3c67e0f8804264329070df723e13ff9a482 Mon Sep 17 00:00:00 2001 From: Codebuff Date: Tue, 20 Oct 2026 21:29:15 +0000 Subject: [PATCH 1/2] feat(indexer): reorg-aware event ingestion with ledger rollback (#724) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detect Stellar ledger rollbacks by anchoring every ingested ledger and event to its hash (LedgerCheckpoint + new IngestedEvent table), walk back to the fork point on a previousLedgerHash mismatch, reverse all affected database side effects in one atomic transaction, emit a warn log with the fork point and affected counts, re-ingest the canonical chain from the fork point, and gate buy/sell confirmations on confirmed ledger depth >= 2. 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- prisma/schema/indexer.prisma | 55 ++ prisma/schema/ledger.prisma | 7 + .../migration.sql | 47 ++ .../contracts/contract.service.test.ts | 8 +- src/modules/contracts/contract.service.ts | 15 + .../contracts/ledger-depth-gate.utils.test.ts | 179 +++++ .../contracts/ledger-depth-gate.utils.ts | 106 +++ .../indexer/reorg-guard.integration.test.ts | 741 ++++++++++++++++++ src/modules/indexer/reorg-guard.service.ts | 669 ++++++++++++++++ 9 files changed, 1826 insertions(+), 1 deletion(-) create mode 100644 prisma/schema/migrations/20261020000000_add_reorg_aware_ingestion/migration.sql create mode 100644 src/modules/contracts/ledger-depth-gate.utils.test.ts create mode 100644 src/modules/contracts/ledger-depth-gate.utils.ts create mode 100644 src/modules/indexer/reorg-guard.integration.test.ts create mode 100644 src/modules/indexer/reorg-guard.service.ts diff --git a/prisma/schema/indexer.prisma b/prisma/schema/indexer.prisma index a0654b31..2725126c 100644 --- a/prisma/schema/indexer.prisma +++ b/prisma/schema/indexer.prisma @@ -1,5 +1,60 @@ // prisma/schema/indexer.prisma +/// Finality status of an ingested chain event with respect to the canonical +/// ledger chain (#724). +enum IngestedEventStatus { + /// Event has been ingested and its ledger is still considered canonical. + CONFIRMED + /// The event's ledger was rolled back by a reorg; its database side + /// effects have been reversed by the reorg-guard rollback. + REORGED +} + +/// One row per ingested chain event, stamped with the ledger sequence and +/// ledger hash it was observed in (#724). +/// +/// Written by the reorg-aware ingestion entry point (`ingestLedger`) so a +/// ledger rollback can identify exactly which events to mark REORGED and +/// reverse, and so re-ingestion after a rollback is auditable. +model IngestedEvent { + id String @id @default(cuid()) + + /// Ledger sequence the event was ingested in. + ledgerSequence Int + /// Hash of the ledger the event was ingested in. + ledgerHash String + + /// On-chain transaction hash (dedup key with eventIndex). + txHash String + /// Index of the event within its transaction. + eventIndex Int + /// Domain event type (e.g. KEY_BOUGHT, KEY_SOLD). + eventType String + + /// Wallet that performed the action, when applicable. + actor String? + /// Creator/key the event applies to, when applicable. + creatorId String? + /// Key quantity the event moved, as a decimal string, when applicable. + amount String? + /// Per-key trade price in stroops, as a decimal string, when applicable. + priceStroops String? + /// Trade timestamp carried by the event, when applicable. + tradeAt DateTime? + + status IngestedEventStatus @default(CONFIRMED) + /// Set when a reorg rolled this event back. + reorgedAt DateTime? + + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([ledgerSequence]) + @@index([status]) + @@index([creatorId]) + @@map("ingested_events") +} + /** * Dead-letter queue (DLQ) for failed indexing jobs. * diff --git a/prisma/schema/ledger.prisma b/prisma/schema/ledger.prisma index fa832bf9..893e47d8 100644 --- a/prisma/schema/ledger.prisma +++ b/prisma/schema/ledger.prisma @@ -10,6 +10,13 @@ model IndexedLedger { model LedgerCheckpoint { id Int @id @default(autoincrement()) ledger Int + /// Hash of the ledger at `ledger`, as reported by the network (#724). + /// Stored for every ledger ingested by the reorg-aware pipeline so the + /// next ledger's `previousLedgerHash` can be verified against it. + /// Nullable only for checkpoints written before hash tracking existed. + ledgerHash String? + /// The `previousLedgerHash` the ingested ledger carried, when available. + previousLedgerHash String? lastProcessedEventIndex Int batchHash String completedAt DateTime @default(now()) diff --git a/prisma/schema/migrations/20261020000000_add_reorg_aware_ingestion/migration.sql b/prisma/schema/migrations/20261020000000_add_reorg_aware_ingestion/migration.sql new file mode 100644 index 00000000..04be19b7 --- /dev/null +++ b/prisma/schema/migrations/20261020000000_add_reorg_aware_ingestion/migration.sql @@ -0,0 +1,47 @@ +-- Reorg-aware Stellar event ingestion (#724). +-- +-- 1. ledger_checkpoints gains ledgerHash/previousLedgerHash so the pipeline +-- can verify that each incoming ledger chains onto the last ingested one +-- and detect ledger rollbacks. +-- 2. New ingested_events table records every ingested event with the ledger +-- sequence and hash it was observed in, so a reorg can mark exactly the +-- affected events as REORGED and reverse their side effects. + +-- AlterTable +ALTER TABLE "ledger_checkpoints" ADD COLUMN "ledgerHash" TEXT, +ADD COLUMN "previousLedgerHash" TEXT; + +-- CreateEnum +CREATE TYPE "IngestedEventStatus" AS ENUM ('CONFIRMED', 'REORGED'); + +-- CreateTable +CREATE TABLE "ingested_events" ( + "id" TEXT NOT NULL, + "ledgerSequence" INTEGER NOT NULL, + "ledgerHash" TEXT NOT NULL, + "txHash" TEXT NOT NULL, + "eventIndex" INTEGER NOT NULL, + "eventType" TEXT NOT NULL, + "actor" TEXT, + "creatorId" TEXT, + "amount" TEXT, + "priceStroops" TEXT, + "tradeAt" TIMESTAMP(3), + "status" "IngestedEventStatus" NOT NULL DEFAULT 'CONFIRMED', + "reorgedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "ingested_events_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "ingested_events_txHash_eventIndex_key" ON "ingested_events"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "ingested_events_ledgerSequence_idx" ON "ingested_events"("ledgerSequence"); + +-- CreateIndex +CREATE INDEX "ingested_events_status_idx" ON "ingested_events"("status"); + +-- CreateIndex +CREATE INDEX "ingested_events_creatorId_idx" ON "ingested_events"("creatorId"); diff --git a/src/modules/contracts/contract.service.test.ts b/src/modules/contracts/contract.service.test.ts index 00af8f65..0c6debe5 100644 --- a/src/modules/contracts/contract.service.test.ts +++ b/src/modules/contracts/contract.service.test.ts @@ -40,9 +40,15 @@ function makeDeps( ) { let currentTime = 0; const pollCalls: number[] = []; + // latestLedger: 102 satisfies the #724 ledger-depth gate (tx ledger 100 + // + depth 2) on the first poll so gate polling adds no extra delays. const getTransaction = overrides.getTransaction ?? - jest.fn().mockResolvedValue({ status: 'SUCCESS', ledger: 100 }); + jest.fn().mockResolvedValue({ + status: 'SUCCESS', + ledger: 100, + latestLedger: 102, + }); return { deps: { diff --git a/src/modules/contracts/contract.service.ts b/src/modules/contracts/contract.service.ts index 096de9fe..3d897e5f 100644 --- a/src/modules/contracts/contract.service.ts +++ b/src/modules/contracts/contract.service.ts @@ -31,6 +31,7 @@ import { pollForConfirmation, type TransactionStatusClient, } from './confirmation-poller.utils'; +import { waitForLedgerDepth } from './ledger-depth-gate.utils'; import { classifyContractError, ContractSubmitError, @@ -283,6 +284,20 @@ export async function submitContractCall( // ── 4. Persist resolution + emit events ───────────────────────────── if (outcome?.status === 'SUCCESS') { + // Ledger-depth gate (#724): buy/sell responses are withheld until the + // ledger that included the transaction is 2 ledgers deep, to reduce + // reorg exposure. Budget exhaustion releases with a warning (logged + // inside the gate) so a stalled RPC head cannot block confirmations + // forever. + await waitForLedgerDepth(outcome.ledger, { + statusClient, + txHash, + intervalMs: pollIntervalMs, + timeoutMs: pollTimeoutMs, + delay, + now, + }); + try { await prisma.contractTransaction.update({ where: { txHash }, diff --git a/src/modules/contracts/ledger-depth-gate.utils.test.ts b/src/modules/contracts/ledger-depth-gate.utils.test.ts new file mode 100644 index 00000000..aacd1086 --- /dev/null +++ b/src/modules/contracts/ledger-depth-gate.utils.test.ts @@ -0,0 +1,179 @@ +// src/modules/contracts/ledger-depth-gate.utils.test.ts +// Unit tests for #724 — buy/sell confirmations are gated on confirmed +// ledger depth ≥ 2. + +import { + CONFIRMED_LEDGER_DEPTH, + waitForLedgerDepth, +} from './ledger-depth-gate.utils'; +import { logger } from '../../utils/logger.utils'; +import { submitContractCall } from './contract.service'; + +jest.mock('../../utils/prisma.utils', () => ({ + prisma: { + contractTransaction: { + create: jest.fn().mockResolvedValue({}), + update: jest.fn().mockResolvedValue({}), + }, + }, +})); + +function makeGateDeps( + getTransaction: jest.Mock, + overrides: { intervalMs?: number; timeoutMs?: number } = {} +) { + let currentTime = 0; + const delays: number[] = []; + return { + deps: { + statusClient: { getTransaction }, + txHash: 'tx-hash-1', + intervalMs: overrides.intervalMs ?? 5, + timeoutMs: overrides.timeoutMs ?? 500, + delay: async (ms: number) => { + delays.push(ms); + currentTime += ms; + }, + now: () => currentTime, + }, + delays, + }; +} + +describe('waitForLedgerDepth (#724)', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('requires a depth of 2 ledgers', () => { + expect(CONFIRMED_LEDGER_DEPTH).toBe(2); + }); + + it('resolves immediately when the ledger is already 2 deep', async () => { + const getTransaction = jest + .fn() + .mockResolvedValue({ status: 'SUCCESS', ledger: 100, latestLedger: 102 }); + const { deps } = makeGateDeps(getTransaction); + + const result = await waitForLedgerDepth(100, deps); + + expect(result).toEqual({ depthConfirmed: true, latestLedger: 102 }); + expect(getTransaction).toHaveBeenCalledTimes(1); + }); + + it('keeps polling until the head advances to the required depth', async () => { + const heads = [100, 101, 102]; + let call = 0; + const getTransaction = jest.fn().mockImplementation(async () => ({ + status: 'SUCCESS', + ledger: 100, + latestLedger: heads[Math.min(call++, heads.length - 1)], + })); + const { deps, delays } = makeGateDeps(getTransaction); + + const result = await waitForLedgerDepth(100, deps); + + expect(result.depthConfirmed).toBe(true); + expect(result.latestLedger).toBe(102); + // Head was polled at least twice and waited between polls. + expect(getTransaction.mock.calls.length).toBeGreaterThanOrEqual(3); + expect(delays.length).toBeGreaterThanOrEqual(2); + }); + + it('reports depthConfirmed=false when the budget is exhausted', async () => { + const warnSpy = jest.spyOn(logger, 'warn'); + const getTransaction = jest + .fn() + .mockResolvedValue({ status: 'SUCCESS', ledger: 100, latestLedger: 100 }); + const { deps } = makeGateDeps(getTransaction, { timeoutMs: 30 }); + + const result = await waitForLedgerDepth(100, deps); + + expect(result.depthConfirmed).toBe(false); + expect(result.latestLedger).toBe(100); + expect(warnSpy).toHaveBeenCalledWith( + expect.objectContaining({ + tx_ledger: 100, + latest_ledger: 100, + required_depth: 2, + }), + 'Ledger depth gate timed out before the required confirmation depth' + ); + warnSpy.mockRestore(); + }); + + it('retries transient RPC failures instead of throwing', async () => { + let call = 0; + const getTransaction = jest.fn().mockImplementation(async () => { + call += 1; + if (call === 1) { + throw new Error('rpc unavailable'); + } + return { status: 'SUCCESS', ledger: 100, latestLedger: 103 }; + }); + const { deps } = makeGateDeps(getTransaction); + + const result = await waitForLedgerDepth(100, deps); + + expect(result.depthConfirmed).toBe(true); + expect(result.latestLedger).toBe(103); + }); + + it('treats a missing latestLedger as "not deep enough yet"', async () => { + const getTransaction = jest + .fn() + .mockResolvedValue({ status: 'SUCCESS', ledger: 100 }); + const { deps } = makeGateDeps(getTransaction, { timeoutMs: 20 }); + + const result = await waitForLedgerDepth(100, deps); + + expect(result.depthConfirmed).toBe(false); + }); +}); + +describe('submitContractCall ledger-depth gate (#724)', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('withholds the CONFIRMED result until the including ledger is 2 deep', async () => { + const sendTransaction = jest + .fn() + .mockResolvedValue({ hash: 'hash-gated', status: 'PENDING' }); + + // Confirmation lands at ledger 100; the head lags at 100/101 and only + // reaches 102 after several polls. + const heads = [100, 100, 101, 102]; + let call = 0; + const getTransaction = jest.fn().mockImplementation(async () => ({ + status: 'SUCCESS', + ledger: 100, + latestLedger: heads[Math.min(call++, heads.length - 1)], + })); + + let currentTime = 0; + const deps = { + submitClient: { sendTransaction }, + statusClient: { getTransaction }, + delay: async (ms: number) => { + currentTime += ms; + }, + now: () => currentTime, + pollIntervalMs: 5, + pollTimeoutMs: 500, + }; + + const result = await submitContractCall( + { + operation: 'buy' as const, + submitterWallet: 'GWALLET', + transactionXdr: 'AAAAbase64xdr==', + }, + deps + ); + + expect(result.status).toBe('CONFIRMED'); + // At least one confirmation poll plus the depth-gate polls. + expect(getTransaction.mock.calls.length).toBeGreaterThan(1); + }); +}); diff --git a/src/modules/contracts/ledger-depth-gate.utils.ts b/src/modules/contracts/ledger-depth-gate.utils.ts new file mode 100644 index 00000000..016fb5ec --- /dev/null +++ b/src/modules/contracts/ledger-depth-gate.utils.ts @@ -0,0 +1,106 @@ +// src/modules/contracts/ledger-depth-gate.utils.ts +// Ledger-depth gate for buy/sell confirmations (#724). +// +// A Stellar ledger is not final the moment it closes — a short reorg can +// still replace it. To reduce reorg exposure, every buy/sell confirmation +// surfaced by the contract interaction service must wait until the ledger +// that included the transaction is at least CONFIRMED_LEDGER_DEPTH ledgers +// deep before the response (and the `contract_tx_confirmed` event) is +// released. +// +// The gate keeps polling `getTransaction` (which reports the network head +// as `latestLedger`) until `latestLedger - txLedger >= CONFIRMED_LEDGER_DEPTH` +// or the polling budget is exhausted. The clock/delay/status client are +// injectable so tests run with fake timers. + +import { logger } from '../../utils/logger.utils'; +import type { TransactionStatusClient } from './confirmation-poller.utils'; + +/** Required confirmed ledger depth before a buy/sell response is released. */ +export const CONFIRMED_LEDGER_DEPTH = 2; + +export interface DepthGateDeps { + statusClient: TransactionStatusClient; + /** Hash of the confirmed transaction; used to read the network head. */ + txHash: string; + /** Delay between head polls, ms. */ + intervalMs: number; + /** Total polling budget, ms. */ + timeoutMs: number; + /** Delay implementation; defaults to a real setTimeout. */ + delay?: (ms: number) => Promise; + /** Monotonic-ish clock; injectable for tests. */ + now?: () => number; +} + +export interface DepthGateResult { + /** True when the including ledger reached the required depth in budget. */ + depthConfirmed: boolean; + /** Last observed network head; 0 when none could be observed. */ + latestLedger: number; +} + +const defaultDelay = (ms: number): Promise => + new Promise(resolve => setTimeout(resolve, ms)); + +/** + * Wait until `txLedger` is CONFIRMED_LEDGER_DEPTH ledgers deep. + * + * Never throws for poll-level errors — transient RPC failures are logged and + * retried until the deadline, mirroring {@link pollForConfirmation}. When + * the budget is exhausted without reaching the required depth the gate + * reports `depthConfirmed: false`; callers decide whether to release or + * withhold the response (the contract service releases with a warning so a + * stalled RPC head cannot block confirmations forever). + */ +export async function waitForLedgerDepth( + txLedger: number, + deps: DepthGateDeps +): Promise { + const { + statusClient, + txHash, + intervalMs, + timeoutMs, + delay = defaultDelay, + now = Date.now, + } = deps; + + const deadline = now() + timeoutMs; + let latestLedger = 0; + let lastError: string | null = null; + + while (now() < deadline) { + try { + const result = await statusClient.getTransaction(txHash); + if (typeof result?.latestLedger === 'number') { + latestLedger = result.latestLedger; + } + if (latestLedger - txLedger >= CONFIRMED_LEDGER_DEPTH) { + return { depthConfirmed: true, latestLedger }; + } + } catch (error) { + lastError = error instanceof Error ? error.message : String(error); + logger.warn( + { tx_hash: txHash, error: lastError }, + 'Ledger depth gate poll failed, will retry' + ); + } + + await delay(intervalMs); + } + + logger.warn( + { + tx_hash: txHash, + tx_ledger: txLedger, + latest_ledger: latestLedger, + required_depth: CONFIRMED_LEDGER_DEPTH, + timeout_ms: timeoutMs, + error: lastError, + }, + 'Ledger depth gate timed out before the required confirmation depth' + ); + + return { depthConfirmed: false, latestLedger }; +} diff --git a/src/modules/indexer/reorg-guard.integration.test.ts b/src/modules/indexer/reorg-guard.integration.test.ts new file mode 100644 index 00000000..5869b865 --- /dev/null +++ b/src/modules/indexer/reorg-guard.integration.test.ts @@ -0,0 +1,741 @@ +// src/modules/indexer/reorg-guard.integration.test.ts +// Integration tests for #724 — reorg-aware Stellar event ingestion. +// +// These run against a real Postgres instance (see docker-compose.yml — +// `pnpm db:up`), not against the mocked Prisma client from jest.setup.ts, +// because the whole point of the reorg guard is the database state it +// rolls back and re-ingests. +// +// Scenario modelled below: +// shared L100 (hash LH100) buyer1 buys 5 @ 1.0 XLM +// forked L101F (hash LF101) buyer2 buys 3 @ 1.1 XLM (abandoned) +// forked L102F (hash LF102) buyer1 buys 1 @ 1.2 XLM (abandoned) +// canonical L101C (hash LC101) buyer2 buys 3 @ 1.1 XLM +// canonical L102C (hash LC102) buyer1 sells 2 @ 1.3 XLM +// +// The forked branch is ingested first; then canonical L102C arrives whose +// previousLedgerHash (LC101) no longer matches the stored hash of L101 +// (LF101), which triggers detection, rollback to the fork point L100, +// re-ingestion of canonical L101C, and ingestion of L102C. +// +// The "identical final state" acceptance criterion is verified by running +// the canonical chain cleanly through a control creator and comparing the +// full normalized read-model state against the reorged creator's state. + +jest.unmock('@prisma/client'); + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + ingestLedger, + recordLedgerIngestion, + rollbackToLedger, + type LedgerHeader, +} from './reorg-guard.service'; +import type { IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; + +const PREFIX = 'reorg724'; +const RUN_ID = `${PREFIX}-${Date.now().toString(36)}`; + +const H_L99 = 'hash-L99'; +const H_L100 = 'hash-L100'; +const H_L101F = 'hash-L101F'; +const H_L102F = 'hash-L102F'; +const H_L101C = 'hash-L101C'; +const H_L102C = 'hash-L102C'; + +const T100 = '2026-01-01T00:00:00.000Z'; +const T101 = '2026-01-01T00:00:05.000Z'; +const T102 = '2026-01-01T00:00:10.000Z'; + +const HEADER_L100: LedgerHeader = { + sequence: 100, + hash: H_L100, + previousLedgerHash: H_L99, +}; +const HEADER_L101F: LedgerHeader = { + sequence: 101, + hash: H_L101F, + previousLedgerHash: H_L100, +}; +const HEADER_L102F: LedgerHeader = { + sequence: 102, + hash: H_L102F, + previousLedgerHash: H_L101F, +}; +const HEADER_L101C: LedgerHeader = { + sequence: 101, + hash: H_L101C, + previousLedgerHash: H_L100, +}; +const HEADER_L102C: LedgerHeader = { + sequence: 102, + hash: H_L102C, + previousLedgerHash: H_L101C, +}; + +function buyEvent( + tag: string, + ledger: number, + actor: string, + creatorId: string, + amount: number, + priceStroops: bigint, + tradeAt: string, + eventIndex = 0 +): IndexerChainEvent { + return { + txHash: `tx724-${tag}-L${ledger}`, + eventIndex, + eventType: 'KEY_BOUGHT', + ledger, + creatorId, + actor, + amount, + price: priceStroops, + feePaid: BigInt(5_000), + tradeAt, + }; +} + +function sellEvent( + tag: string, + ledger: number, + actor: string, + creatorId: string, + amount: number, + priceStroops: bigint, + tradeAt: string, + eventIndex = 0 +): IndexerChainEvent { + return { + txHash: `tx724-${tag}-L${ledger}`, + eventIndex, + eventType: 'KEY_SOLD', + ledger, + creatorId, + actor, + amount, + price: priceStroops, + feePaid: BigInt(5_000), + tradeAt, + }; +} + +async function seedCreator(tag: string): Promise { + const userId = `${PREFIX}-user-${tag}-${RUN_ID}`; + await prisma.user.upsert({ + where: { id: userId }, + create: { + id: userId, + email: `${PREFIX}-${tag}-${RUN_ID}@example.test`, + passwordHash: 'dummy-hash', + firstName: 'Reorg', + lastName: 'Test', + }, + update: {}, + }); + const creator = await prisma.creatorProfile.upsert({ + where: { handle: `${PREFIX}-${tag}-${RUN_ID}` }, + create: { + userId, + handle: `${PREFIX}-${tag}-${RUN_ID}`, + displayName: `Reorg ${tag}`, + }, + update: {}, + }); + return creator.id; +} + +/** + * Collect the complete normalized read-model state produced by ingestion + * for one creator + wallet set, so two runs that consumed the same + * canonical events can be compared for equality. Run-specific identifiers + * (creator id, wallet addresses, tx hashes) are replaced by role tokens. + */ +async function captureState( + creatorId: string, + roleByWallet: Map +): Promise { + const wallets = [...roleByWallet.keys()]; + + const ownership = await prisma.keyOwnership.findMany({ + where: { creatorId, ownerAddress: { in: wallets } }, + orderBy: { ownerAddress: 'asc' }, + }); + + const activities = await prisma.activity.findMany({ + where: { creatorId, type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] } }, + orderBy: [{ createdAt: 'asc' }, { id: 'asc' }], + }); + + const profile = await prisma.creatorProfile.findUnique({ + where: { id: creatorId }, + select: { circulatingSupply: true, currentMilestone: true }, + }); + + const snapshot = await prisma.creatorPriceSnapshot.findUnique({ + where: { creatorId }, + }); + + const priceHistory = await prisma.creatorPriceHistory.findMany({ + where: { creatorId }, + orderBy: [{ recordedAt: 'asc' }, { id: 'asc' }], + }); + + const holdingSnapshots = await prisma.walletHoldingSnapshot.findMany({ + where: { creatorId, ownerAddress: { in: wallets } }, + orderBy: [{ recordedAt: 'asc' }, { id: 'asc' }], + }); + + // Only CONFIRMED events are compared: a reorged run intentionally keeps + // extra REORGED audit rows, which is asserted separately in the + // detection test. + const ingestedEvents = await prisma.ingestedEvent.findMany({ + where: { creatorId, status: 'CONFIRMED' }, + orderBy: [{ ledgerSequence: 'asc' }, { eventIndex: 'asc' }], + }); + + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { in: [100, 101, 102] } }, + orderBy: { ledger: 'asc' }, + }); + + return { + ownership: ownership.map(row => ({ + wallet: roleByWallet.get(row.ownerAddress), + balance: Number(row.balance), + costBasis: Number(row.costBasis ?? 0), + realisedPnl: Number(row.realisedPnl ?? 0), + lastBuyAt: row.lastBuyAt?.toISOString() ?? null, + })), + activities: activities.map(row => ({ + type: row.type, + wallet: roleByWallet.get(row.actor), + amount: (row.payload as any)?.amount, + priceAtTrade: (row.payload as any)?.price_at_trade, + ledgerSequence: (row.payload as any)?.ledger_sequence, + createdAt: row.createdAt.toISOString(), + })), + profile: { + circulatingSupply: + profile?.circulatingSupply !== null && profile?.circulatingSupply !== undefined + ? Number(profile.circulatingSupply) + : null, + currentMilestone: profile?.currentMilestone ?? null, + }, + priceSnapshot: snapshot + ? { + currentPrice: snapshot.currentPrice.toString(), + price24hAgo: snapshot.price24hAgo.toString(), + lastTradeAt: snapshot.lastTradeAt?.toISOString() ?? null, + } + : null, + priceHistory: priceHistory.map(row => ({ + price: row.price.toString(), + supply: row.supply.toString(), + direction: row.direction, + recordedAt: row.recordedAt.toISOString(), + })), + holdingSnapshots: holdingSnapshots.map(row => ({ + wallet: roleByWallet.get(row.ownerAddress), + balance: Number(row.balance), + recordedAt: row.recordedAt.toISOString(), + })), + ingestedEvents: ingestedEvents.map(row => ({ + ledgerSequence: row.ledgerSequence, + ledgerHash: row.ledgerHash, + role: row.actor ? roleByWallet.get(row.actor) ?? row.actor : null, + eventType: row.eventType, + amount: row.amount, + priceStroops: row.priceStroops, + status: row.status, + })), + checkpoints: checkpoints.map(row => ({ + ledger: row.ledger, + ledgerHash: row.ledgerHash, + previousLedgerHash: row.previousLedgerHash, + lastProcessedEventIndex: row.lastProcessedEventIndex, + })), + }; +} + +async function cleanup(): Promise { + const creators = await prisma.creatorProfile.findMany({ + where: { handle: { startsWith: `${PREFIX}-` } }, + select: { id: true }, + }); + const creatorIds = creators.map(c => c.id); + + await prisma.ingestedEvent.deleteMany({ + where: { txHash: { startsWith: `tx724-` } }, + }); + await prisma.ledgerCheckpoint.deleteMany({ + where: { ledger: { in: [99, 100, 101, 102] } }, + }); + if (creatorIds.length > 0) { + await prisma.activity.deleteMany({ + where: { creatorId: { in: creatorIds } }, + }); + await prisma.keyOwnership.deleteMany({ + where: { creatorId: { in: creatorIds } }, + }); + await prisma.creatorPriceHistory.deleteMany({ + where: { creatorId: { in: creatorIds } }, + }); + await prisma.creatorPriceSnapshot.deleteMany({ + where: { creatorId: { in: creatorIds } }, + }); + await prisma.walletHoldingSnapshot.deleteMany({ + where: { creatorId: { in: creatorIds } }, + }); + await prisma.creatorProfile.deleteMany({ + where: { id: { in: creatorIds } }, + }); + } + await prisma.user.deleteMany({ + where: { email: { startsWith: `${PREFIX}-` } }, + }); +} + +describe('#724 reorg-aware ingestion — integration', () => { + // Each test starts from a clean slate: the checkpoint and ingested-event + // tables are global, so leftover rows from a previous test would skew + // reorg detection and rollback counts. + beforeEach(async () => { + await cleanup(); + }); + + afterAll(async () => { + await cleanup(); + await prisma.$disconnect(); + }); + + describe('clean ingestion', () => { + it('advances the checkpoint with hash and sequence for every ingested ledger', async () => { + const creatorId = await seedCreator('clean'); + const buyer1 = `${PREFIX}-w1-${RUN_ID}`; + + await ingestLedger(HEADER_L100, [ + buyEvent('clean', 100, buyer1, creatorId, 5, 10_000_000n, T100), + ]); + await ingestLedger(HEADER_L101C, [ + buyEvent('clean', 101, buyer1, creatorId, 2, 11_000_000n, T101), + ]); + + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { in: [100, 101] } }, + orderBy: { ledger: 'asc' }, + }); + expect(checkpoints).toHaveLength(2); + expect(checkpoints[0]).toMatchObject({ + ledger: 100, + ledgerHash: H_L100, + previousLedgerHash: H_L99, + }); + expect(checkpoints[1]).toMatchObject({ + ledger: 101, + ledgerHash: H_L101C, + previousLedgerHash: H_L100, + }); + + // Every ingested event is recorded with its sequence + hash. + const events = await prisma.ingestedEvent.findMany({ + where: { creatorId }, + orderBy: [{ ledgerSequence: 'asc' }, { eventIndex: 'asc' }], + }); + expect(events).toHaveLength(2); + expect(events[0]).toMatchObject({ + ledgerSequence: 100, + ledgerHash: H_L100, + eventType: 'KEY_BOUGHT', + amount: '5', + priceStroops: '10000000', + status: 'CONFIRMED', + }); + expect(events[1]).toMatchObject({ + ledgerSequence: 101, + ledgerHash: H_L101C, + status: 'CONFIRMED', + }); + + // A ledger that chains correctly reports no reorg. + const result = await ingestLedger( + { sequence: 102, hash: H_L102C, previousLedgerHash: H_L101C }, + [sellEvent('clean', 102, buyer1, creatorId, 1, 13_000_000n, T102)] + ); + expect(result.reorgDetected).toBe(false); + expect(result.eventsIngested).toBe(1); + }); + }); + + describe('reorg detection and rollback', () => { + it('detects the hash mismatch, rolls back to the fork point, re-ingests, and emits the warn log', async () => { + const warnSpy = jest.spyOn(logger, 'warn'); + const creatorId = await seedCreator('reorg'); + const buyer1 = `${PREFIX}-rw1-${RUN_ID}`; + const buyer2 = `${PREFIX}-rw2-${RUN_ID}`; + + const events100 = [ + buyEvent('reorg', 100, buyer1, creatorId, 5, 10_000_000n, T100), + ]; + const events101F = [ + buyEvent('reorgF', 101, buyer2, creatorId, 3, 11_500_000n, T101), + ]; + const events102F = [ + buyEvent('reorgF', 102, buyer1, creatorId, 1, 12_500_000n, T102), + ]; + const events101C = [ + buyEvent('reorgC', 101, buyer2, creatorId, 3, 11_000_000n, T101), + ]; + const events102C = [ + sellEvent('reorgC', 102, buyer1, creatorId, 2, 13_000_000n, T102), + ]; + + // Ingest the branch that will be abandoned. + await ingestLedger(HEADER_L100, events100); + await ingestLedger(HEADER_L101F, events101F); + await ingestLedger(HEADER_L102F, events102F); + + // Sanity: forked-branch state is in place. + const forkedBalance = await prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer1, + creatorId, + }, + }, + }); + expect(Number(forkedBalance?.balance)).toBe(6); // 5 + 1 + + warnSpy.mockClear(); + + // Canonical L102C arrives: its previousLedgerHash (LC101) does not + // match the stored hash of L101 (LF101) → reorg. + const result = await ingestLedger(HEADER_L102C, events102C, { + fetchHeader: async seq => + seq === 101 ? HEADER_L101C : seq === 102 ? HEADER_L102C : null, + fetchEvents: async seq => (seq === 101 ? events101C : []), + }); + + expect(result.reorgDetected).toBe(true); + expect(result.forkPoint).toBe(100); + expect(result.reorgedLedgerCount).toBe(2); // L101F + L102F + expect(result.affectedEventCount).toBe(2); // one event per forked ledger + expect(result.reingestedLedgers).toEqual([101]); + + // Warn-level structured log with fork point and affected counts. + const reorgWarn = warnSpy.mock.calls.find( + call => call[1] === 'Ledger reorg detected — rolled back database state from the fork point' + ); + expect(reorgWarn).toBeDefined(); + const fields = reorgWarn![0] as Record; + expect(fields.forkPoint).toBe(100); + expect(fields.reorgedLedgerCount).toBe(2); + expect(fields.affectedEventCount).toBe(2); + + // The forked events are marked REORGED, the canonical ones + // CONFIRMED, and no forked checkpoint survives. + const events = await prisma.ingestedEvent.findMany({ + where: { creatorId }, + orderBy: [{ ledgerSequence: 'asc' }, { eventIndex: 'asc' }], + }); + expect( + events.map(e => [e.ledgerSequence, e.status, e.ledgerHash]) + ).toEqual([ + [100, 'CONFIRMED', H_L100], + [101, 'REORGED', H_L101F], + [101, 'CONFIRMED', H_L101C], + [102, 'REORGED', H_L102F], + [102, 'CONFIRMED', H_L102C], + ]); + expect(events[1].reorgedAt).not.toBeNull(); + + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { in: [100, 101, 102] } }, + orderBy: { ledger: 'asc' }, + }); + expect(checkpoints.map(c => [c.ledger, c.ledgerHash])).toEqual([ + [100, H_L100], + [101, H_L101C], + [102, H_L102C], + ]); + + // Final balances reflect only the canonical chain: + // buyer1: 5 (L100) - 2 (sell at L102C) = 3; buyer2: 3 (L101C). + const [b1, b2] = await Promise.all([ + prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer1, + creatorId, + }, + }, + }), + prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer2, + creatorId, + }, + }, + }), + ]); + expect(Number(b1?.balance)).toBe(3); + expect(Number(b2?.balance)).toBe(3); + + // Only canonical activities remain (3), and no Trade rows leaked. + const activities = await prisma.activity.findMany({ + where: { creatorId, type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] } }, + }); + expect(activities).toHaveLength(3); + const supply = await prisma.creatorProfile.findUnique({ + where: { id: creatorId }, + select: { circulatingSupply: true }, + }); + expect(Number(supply?.circulatingSupply)).toBe(6); // 5 + 3 - 2 + + warnSpy.mockRestore(); + }); + + it('fully reverses rolled-back balance, supply, price and history side effects', async () => { + const creatorId = await seedCreator('reverse'); + const buyer1 = `${PREFIX}-rv1-${RUN_ID}`; + const buyer2 = `${PREFIX}-rv2-${RUN_ID}`; + + const events100 = [ + buyEvent('reverse', 100, buyer1, creatorId, 5, 10_000_000n, T100), + ]; + const events101F = [ + buyEvent('reverseF', 101, buyer2, creatorId, 3, 11_500_000n, T101), + ]; + const events102F = [ + buyEvent('reverseF', 102, buyer1, creatorId, 1, 12_500_000n, T102), + ]; + + await ingestLedger(HEADER_L100, events100); + await ingestLedger(HEADER_L101F, events101F); + await ingestLedger(HEADER_L102F, events102F); + + // Directly exercise the rollback for the forked ledgers 101 + 102. + const rollback = await rollbackToLedger(100); + expect(rollback.reorgedLedgerCount).toBe(2); + expect(rollback.affectedEventCount).toBe(2); + + // Balance changes reversed: buyer1 back to 5, buyer2 back to 0. + const [b1, b2] = await Promise.all([ + prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer1, + creatorId, + }, + }, + }), + prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer2, + creatorId, + }, + }, + }), + ]); + expect(Number(b1?.balance)).toBe(5); + expect(Number(b1?.costBasis)).toBeCloseTo(1.0, 7); + expect(Number(b2?.balance)).toBe(0); + + // Transaction history entries from the forked ledgers are gone. + const activities = await prisma.activity.findMany({ + where: { creatorId, type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] } }, + }); + expect(activities).toHaveLength(1); + expect( + (activities[0].payload as any)?.ledger_sequence + ).toBe(100); + + // Circulating supply recomputed to the surviving history. + const profile = await prisma.creatorProfile.findUnique({ + where: { id: creatorId }, + select: { circulatingSupply: true }, + }); + expect(Number(profile?.circulatingSupply)).toBe(5); + + // Price snapshot restored to the last surviving trade. + const snapshot = await prisma.creatorPriceSnapshot.findUnique({ + where: { creatorId }, + }); + expect(snapshot?.currentPrice.toString()).toBe('10000000'); + expect(snapshot?.lastTradeAt?.toISOString()).toBe(T100); + const history = await prisma.creatorPriceHistory.findMany({ + where: { creatorId }, + }); + expect(history).toHaveLength(1); + + // Post-trade holding snapshots from forked trades removed. + const snapshots = await prisma.walletHoldingSnapshot.findMany({ + where: { creatorId }, + }); + expect(snapshots).toHaveLength(1); + expect(Number(snapshots[0].balance)).toBe(5); + + // Checkpoints after the fork point removed; fork point survives. + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { in: [100, 101, 102] } }, + }); + expect(checkpoints).toHaveLength(1); + expect(checkpoints[0].ledger).toBe(100); + + // Events marked REORGED. + const reorged = await prisma.ingestedEvent.count({ + where: { creatorId, status: 'REORGED' }, + }); + expect(reorged).toBe(2); + }); + + it('rolls back atomically — a failure mid-rollback leaves all state untouched', async () => { + const creatorId = await seedCreator('atomic'); + const buyer1 = `${PREFIX}-at1-${RUN_ID}`; + const buyer2 = `${PREFIX}-at2-${RUN_ID}`; + + await ingestLedger(HEADER_L100, [ + buyEvent('atomic', 100, buyer1, creatorId, 5, 10_000_000n, T100), + ]); + await ingestLedger(HEADER_L101F, [ + buyEvent('atomicF', 101, buyer2, creatorId, 4, 11_500_000n, T101), + ]); + + // Inject an event whose creator does not exist, so the supply + // recompute step (creatorProfile.update) fails inside the + // transaction after earlier steps already ran. + await prisma.ingestedEvent.create({ + data: { + ledgerSequence: 101, + ledgerHash: H_L101F, + txHash: 'tx724-atomic-bogus', + eventIndex: 9, + eventType: 'KEY_BOUGHT', + actor: buyer1, + creatorId: `${PREFIX}-missing-creator-${RUN_ID}`, + amount: '1', + }, + }); + + await expect(rollbackToLedger(100)).rejects.toThrow(); + + // Nothing changed: history, balances, checkpoints, event statuses. + const activities = await prisma.activity.findMany({ + where: { creatorId, type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] } }, + }); + expect(activities).toHaveLength(2); + + const balance = await prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: buyer2, + creatorId, + }, + }, + }); + expect(Number(balance?.balance)).toBe(4); + + const checkpoints = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { in: [100, 101] } }, + }); + expect(checkpoints).toHaveLength(2); + + const statuses = await prisma.ingestedEvent.findMany({ + where: { txHash: { startsWith: 'tx724-atomic' } }, + select: { status: true }, + }); + expect(statuses.every(s => s.status === 'CONFIRMED')).toBe(true); + }); + }); + + describe('identical final state after re-ingestion', () => { + it('produces the same read-model state as if the reorg never happened', async () => { + // ── Control run: canonical chain ingested cleanly. ──────────── + const controlCreator = await seedCreator('control'); + const cBuyer1 = `${PREFIX}-c1-${RUN_ID}`; + const cBuyer2 = `${PREFIX}-c2-${RUN_ID}`; + + await ingestLedger(HEADER_L100, [ + buyEvent('ctrl', 100, cBuyer1, controlCreator, 5, 10_000_000n, T100), + ]); + await ingestLedger(HEADER_L101C, [ + buyEvent('ctrlC', 101, cBuyer2, controlCreator, 3, 11_000_000n, T101), + ]); + await ingestLedger(HEADER_L102C, [ + sellEvent('ctrlC', 102, cBuyer1, controlCreator, 2, 13_000_000n, T102), + ]); + + // Capture the control state now: the reorg rollback below is + // global (one canonical chain), so it also touches the control + // creator's rows in the shared tables. + const controlState = await captureState(controlCreator, new Map([ + [cBuyer1, 'buyer1'], + [cBuyer2, 'buyer2'], + ])); + + // ── Reorg run: forked branch first, then canonical chain. ───── + const reorgCreator = await seedCreator('finalstate'); + const rBuyer1 = `${PREFIX}-f1-${RUN_ID}`; + const rBuyer2 = `${PREFIX}-f2-${RUN_ID}`; + + await ingestLedger(HEADER_L100, [ + buyEvent('fs', 100, rBuyer1, reorgCreator, 5, 10_000_000n, T100), + ]); + await ingestLedger(HEADER_L101F, [ + buyEvent('fsF', 101, rBuyer2, reorgCreator, 3, 11_500_000n, T101), + ]); + await ingestLedger(HEADER_L102F, [ + buyEvent('fsF', 102, rBuyer1, reorgCreator, 1, 12_500_000n, T102), + ]); + + const result = await ingestLedger(HEADER_L102C, [ + sellEvent('fsC', 102, rBuyer1, reorgCreator, 2, 13_000_000n, T102), + ], { + fetchHeader: async seq => + seq === 101 ? HEADER_L101C : seq === 102 ? HEADER_L102C : null, + fetchEvents: async seq => + seq === 101 + ? [ + buyEvent('fsC', 101, rBuyer2, reorgCreator, 3, 11_000_000n, T101), + ] + : [], + }); + expect(result.reorgDetected).toBe(true); + expect(result.forkPoint).toBe(100); + + const reorgState = await captureState(reorgCreator, new Map([ + [rBuyer1, 'buyer1'], + [rBuyer2, 'buyer2'], + ])); + + expect(reorgState).toEqual(controlState); + }); + }); + + describe('recordLedgerIngestion idempotency', () => { + it('upserts the same checkpoint and event rows on replay', async () => { + const creatorId = await seedCreator('idem'); + const buyer1 = `${PREFIX}-id1-${RUN_ID}`; + const events = [ + buyEvent('idem', 100, buyer1, creatorId, 5, 10_000_000n, T100), + ]; + + await recordLedgerIngestion(HEADER_L100, events); + await recordLedgerIngestion(HEADER_L100, events); + + const checkpoints = await prisma.ledgerCheckpoint.count({ + where: { ledger: 100 }, + }); + expect(checkpoints).toBe(1); + + const ingested = await prisma.ingestedEvent.count({ + where: { txHash: events[0].txHash }, + }); + expect(ingested).toBe(1); + }); + }); +}); diff --git a/src/modules/indexer/reorg-guard.service.ts b/src/modules/indexer/reorg-guard.service.ts new file mode 100644 index 00000000..a09a046b --- /dev/null +++ b/src/modules/indexer/reorg-guard.service.ts @@ -0,0 +1,669 @@ +// src/modules/indexer/reorg-guard.service.ts +// Reorg-aware Stellar event ingestion (#724). +// +// Stellar ledgers are usually final on first observation, but a network +// partition or validator disagreement can cause a short ledger rollback +// (reorg) that invalidates previously ingested events. This module makes the +// ingestion pipeline reorg-aware: +// +// - `recordLedgerIngestion` stores the ledger sequence + ledger hash for +// every ingested ledger (LedgerCheckpoint) and every ingested event +// (IngestedEvent), so the chain linkage is auditable. +// - `ingestLedger` is the reorg-aware ingestion entry point. Before +// ingesting a new ledger it verifies +// `incoming.previousLedgerHash === checkpoint[incoming.sequence - 1].hash`. +// - On mismatch it finds the fork point by walking backwards through the +// stored LedgerCheckpoints until the incoming chain's hashes agree, +// rolls back every database side effect from the fork point forward in a +// single atomic transaction (`rollbackToLedger`), emits a warn-level +// structured log with the fork point and affected counts, and re-ingests +// the canonical ledgers from the fork point forward. +// +// Reversal model for a trade event's side effects: +// - transaction history entries (Activity KEY_BOUGHT/KEY_SOLD rows and +// Trade rows) are deleted, +// - affected KeyOwnership positions are rebuilt by replaying the remaining +// (non-reorged) trade history through the same weighted-average cost +// basis / realised P&L math the pipeline uses, which also corrects +// derived holder counts (they are counted from KeyOwnership balances), +// - affected CreatorPriceSnapshot / CreatorPriceHistory rows are restored +// from the latest surviving price history row, +// - CreatorProfile.circulatingSupply / currentMilestone are recomputed, +// - the reorged LedgerCheckpoints are deleted so the ledgers can be +// re-ingested from the fork point. +// +// Everything above happens inside one `prisma.$transaction` so the rollback +// is atomic: either the whole rollback lands or none of it does. + +import type { Prisma } from '@prisma/client'; +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { dedupeChainEvents } from '../../utils/indexer-dedupe.utils'; +import { computeBatchHash } from './ledger-checkpoint.service'; +import { processTradeEvents } from './indexer-pipeline.service'; +import { getTierForSupply } from '../keys/key-milestones.service'; +import type { IndexerChainEvent } from '../../utils/indexer-event-processor.utils'; + +/** One Stellar ledger header as seen on the incoming (canonical) chain. */ +export interface LedgerHeader { + /** Ledger sequence number. */ + sequence: number; + /** Hash of this ledger itself. */ + hash: string; + /** Hash this ledger's parent, as carried by this ledger. */ + previousLedgerHash: string | null; +} + +/** + * Upper bound on how far backwards the fork-point walk (and the subsequent + * re-ingestion) may go. Stellar reorgs are short — a bound keeps a badly + * wrong hash source from scanning the whole checkpoint history. + */ +export const MAX_FORK_WALKBACK_LEDGERS = 1_000; + +/** Injectable dependencies; production callers normally use the defaults. */ +export interface ReorgGuardDeps { + /** + * Fetch the canonical header for a ledger sequence from the *incoming* + * chain. Used to walk backwards to the fork point and to re-ingest the + * canonical ledgers after a rollback. When absent, intermediate ledgers + * cannot be re-fetched and are left for the caller to re-deliver. + */ + fetchHeader?: (sequence: number) => Promise; + /** + * Fetch the ingested events for a ledger on the canonical chain so it can + * be re-ingested. Required together with `fetchHeader` for re-ingestion. + */ + fetchEvents?: (sequence: number) => Promise; + /** + * Event processor used for both fresh ingestion and re-ingestion. + * Defaults to the trade-event pipeline (`processTradeEvents`). + */ + processEvents?: (events: IndexerChainEvent[]) => Promise; +} + +/** Result of detecting and (if needed) rolling back a reorg. */ +export interface ReorgOutcome { + reorgDetected: boolean; + /** Last canonical ledger that survives the rollback; null when clean. */ + forkPoint: number | null; + /** Number of ledger checkpoints removed by the rollback. */ + reorgedLedgerCount: number; + /** Number of ingested events marked REORGED by the rollback. */ + affectedEventCount: number; +} + +/** Result of ingesting one ledger through the reorg-aware entry point. */ +export interface IngestLedgerResult extends ReorgOutcome { + sequence: number; + /** Unique events ingested for the incoming ledger. */ + eventsIngested: number; + /** Canonical ledgers re-ingested from the fork point, in order. */ + reingestedLedgers: number[]; +} + +/** Interactive-transaction budget for the rollback (rebuild can be slow). */ +const ROLLBACK_TX_TIMEOUT_MS = 60_000; +const ROLLBACK_TX_MAX_WAIT_MS = 10_000; + +type Tx = Prisma.TransactionClient; + +const TRADE_ACTIVITY_TYPES = ['KEY_BOUGHT', 'KEY_SOLD'] as const; + +/** Convert an incoming chain event into an IngestedEvent row payload. */ +function toIngestedEventFields( + event: IndexerChainEvent, + header: LedgerHeader +): Prisma.IngestedEventUncheckedCreateInput { + const tradeAt = + event.tradeAt !== undefined && event.tradeAt !== null + ? new Date(String(event.tradeAt)) + : null; + + return { + ledgerSequence: + typeof event.ledger === 'number' && Number.isFinite(event.ledger) + ? event.ledger + : header.sequence, + ledgerHash: header.hash, + txHash: event.txHash, + eventIndex: event.eventIndex, + eventType: event.eventType, + actor: typeof event.actor === 'string' ? event.actor : null, + creatorId: + typeof event.creatorId === 'string' ? event.creatorId : null, + amount: + event.amount !== undefined && event.amount !== null + ? String(event.amount) + : null, + priceStroops: + event.price !== undefined && event.price !== null + ? String(event.price) + : null, + tradeAt: tradeAt && !Number.isNaN(tradeAt.getTime()) ? tradeAt : null, + }; +} + +/** + * Persist the chain linkage for one ingested ledger: a LedgerCheckpoint + * carrying the ledger sequence + hash (and the parent hash it chained onto), + * plus one IngestedEvent row per unique event, in a single transaction. + * + * Idempotent: replaying the same ledger upserts the same rows. + */ +export async function recordLedgerIngestion( + header: LedgerHeader, + events: IndexerChainEvent[] +): Promise { + const uniqueEvents = dedupeChainEvents(events); + const eventIds = uniqueEvents.map(e => `${e.txHash}:${e.eventIndex}`); + const lastProcessedEventIndex = + uniqueEvents.length > 0 + ? Math.max(...uniqueEvents.map(e => e.eventIndex)) + : 0; + + await prisma.$transaction( + async tx => { + for (const event of uniqueEvents) { + const fields = toIngestedEventFields(event, header); + await tx.ingestedEvent.upsert({ + where: { + txHash_eventIndex: { + txHash: event.txHash, + eventIndex: event.eventIndex, + }, + }, + create: { ...fields, status: 'CONFIRMED' }, + update: { + ...fields, + status: 'CONFIRMED', + reorgedAt: null, + }, + }); + } + + await tx.ledgerCheckpoint.upsert({ + where: { ledger: header.sequence }, + create: { + ledger: header.sequence, + ledgerHash: header.hash, + previousLedgerHash: header.previousLedgerHash, + lastProcessedEventIndex, + batchHash: computeBatchHash(eventIds), + }, + update: { + ledgerHash: header.hash, + previousLedgerHash: header.previousLedgerHash, + lastProcessedEventIndex, + batchHash: computeBatchHash(eventIds), + completedAt: new Date(), + }, + }); + }, + { + timeout: ROLLBACK_TX_TIMEOUT_MS, + maxWait: ROLLBACK_TX_MAX_WAIT_MS, + } + ); + + logger.debug( + { + ledger: header.sequence, + ledger_hash: header.hash, + events: uniqueEvents.length, + }, + 'Ledger ingestion recorded (checkpoint + ingested events)' + ); +} + +/** + * Find the fork point for a diverging incoming ledger. + * + * Walks backwards from `header.sequence - 1` through the stored + * LedgerCheckpoints, comparing each stored hash against the hash the + * *incoming* chain reports for the same sequence. The fork point is the + * highest sequence whose hashes still agree — every ledger after it belongs + * to the abandoned branch and must be rolled back. + * + * Returns `header.sequence - 1` when there is nothing stored to compare + * against (nothing to roll back). + */ +export async function findForkPoint( + header: LedgerHeader, + fetchHeader?: (sequence: number) => Promise +): Promise { + const stored = await prisma.ledgerCheckpoint.findMany({ + where: { ledger: { lt: header.sequence } }, + orderBy: { ledger: 'desc' }, + take: MAX_FORK_WALKBACK_LEDGERS, + select: { ledger: true, ledgerHash: true }, + }); + + if (stored.length === 0) { + return header.sequence - 1; + } + + const oldestStored = stored[stored.length - 1].ledger; + + for (const checkpoint of stored) { + const seq = checkpoint.ledger; + + // The incoming ledger directly names its parent's hash; older + // sequences need the caller's hash source for the incoming chain. + const incomingHash = + seq === header.sequence - 1 + ? header.previousLedgerHash + : fetchHeader + ? (await fetchHeader(seq))?.hash ?? null + : null; + + // Cannot walk further without a hash for this sequence on the new + // chain — roll back everything we can still attribute to the fork. + if (incomingHash === null) { + return seq; + } + + if (incomingHash === checkpoint.ledgerHash) { + return seq; + } + } + + // Every walked ledger disagrees — the whole window is on the abandoned + // branch; roll back from just before its oldest checkpoint. + return oldestStored - 1; +} + +/** + * Atomically roll back every database side effect produced at ledgers after + * `forkPoint`: + * + * - marks the affected IngestedEvent rows REORGED, + * - deletes their transaction history entries (Activity + Trade), + * - rebuilds the affected KeyOwnership positions from the surviving trade + * history (same cost-basis / P&L math as the pipeline), which also + * corrects derived holder counts, + * - restores affected price snapshots from the surviving price history, + * - recomputes circulating supply / milestone for affected creators, + * - deletes the reorged LedgerCheckpoints so the ledgers can be re-ingested. + * + * Runs inside a single interactive transaction — atomic by construction. + */ +export async function rollbackToLedger(forkPoint: number): Promise<{ + reorgedLedgerCount: number; + affectedEventCount: number; +}> { + return prisma.$transaction( + async tx => { + const reorgedEvents = await tx.ingestedEvent.findMany({ + where: { + ledgerSequence: { gt: forkPoint }, + status: 'CONFIRMED', + }, + }); + const affectedEventCount = reorgedEvents.length; + + const reorgedLedgers = await tx.ledgerCheckpoint.count({ + where: { ledger: { gt: forkPoint } }, + }); + + // 1. Mark the affected events REORGED (audit trail). + await tx.ingestedEvent.updateMany({ + where: { id: { in: reorgedEvents.map(e => e.id) } }, + data: { status: 'REORGED', reorgedAt: new Date() }, + }); + + // 2. Remove transaction history entries written by the reorged + // ledgers. Trade-side activities carry the ledger sequence in + // their payload; Trade rows store it directly. + await tx.activity.deleteMany({ + where: { + type: { in: [...TRADE_ACTIVITY_TYPES] }, + payload: { path: ['ledger_sequence'], gt: forkPoint }, + }, + }); + await tx.trade.deleteMany({ + where: { ledger: { gt: forkPoint } }, + }); + + const affectedCreators = [ + ...new Set( + reorgedEvents + .map(e => e.creatorId) + .filter((id): id is string => typeof id === 'string' && id.length > 0) + ), + ]; + + // 3. Restore price history + snapshots for affected creators. + for (const creatorId of affectedCreators) { + const creatorTradeTimes = reorgedEvents + .filter(e => e.creatorId === creatorId && e.tradeAt !== null) + .map(e => e.tradeAt as Date); + if (creatorTradeTimes.length > 0) { + const earliestReorgedTradeAt = new Date( + Math.min(...creatorTradeTimes.map(d => d.getTime())) + ); + await tx.creatorPriceHistory.deleteMany({ + where: { + creatorId, + recordedAt: { gte: earliestReorgedTradeAt }, + }, + }); + } + + const latest = await tx.creatorPriceHistory.findFirst({ + where: { creatorId }, + orderBy: [{ recordedAt: 'desc' }, { id: 'desc' }], + }); + + if (latest) { + await tx.creatorPriceSnapshot.upsert({ + where: { creatorId }, + create: { + creatorId, + currentPrice: latest.price, + price24hAgo: latest.price, + lastTradeAt: latest.recordedAt, + }, + update: { + currentPrice: latest.price, + lastTradeAt: latest.recordedAt, + }, + }); + } else { + await tx.creatorPriceSnapshot.deleteMany({ + where: { creatorId }, + }); + } + } + + // 4. Rebuild the affected ownership positions from the surviving + // trade history. Replaying the same events through the same + // math restores balance, weighted-average cost basis, realised + // P&L and lastBuyAt exactly — so holder counts (derived from + // balances) follow automatically. + const affectedPairs = new Map(); + for (const event of reorgedEvents) { + if ( + TRADE_ACTIVITY_TYPES.includes(event.eventType as any) && + typeof event.actor === 'string' && + event.actor.length > 0 && + typeof event.creatorId === 'string' && + event.creatorId.length > 0 + ) { + affectedPairs.set(`${event.actor}:${event.creatorId}`, { + ownerAddress: event.actor, + creatorId: event.creatorId, + }); + } + } + for (const pair of affectedPairs.values()) { + // Drop the post-trade holding snapshots the reorged trades + // wrote (#1008); re-ingestion re-creates the canonical ones + // through the normal pipeline. + const pairTradeTimes = reorgedEvents + .filter( + e => + e.actor === pair.ownerAddress && + e.creatorId === pair.creatorId && + e.tradeAt !== null + ) + .map(e => e.tradeAt as Date); + if (pairTradeTimes.length > 0) { + const earliestPairTradeAt = new Date( + Math.min(...pairTradeTimes.map(d => d.getTime())) + ); + await tx.walletHoldingSnapshot.deleteMany({ + where: { + ownerAddress: pair.ownerAddress, + creatorId: pair.creatorId, + recordedAt: { gte: earliestPairTradeAt }, + }, + }); + } + + await rebuildOwnershipPosition( + // The $extends-ed client's tx is a structural superset of + // Prisma.TransactionClient. + tx as unknown as Tx, + pair.ownerAddress, + pair.creatorId + ); + } + + // 5. Recompute derived supply state for affected creators. + for (const creatorId of affectedCreators) { + const activities = await tx.activity.findMany({ + where: { + creatorId, + type: { in: [...TRADE_ACTIVITY_TYPES] }, + }, + select: { type: true, payload: true }, + }); + const supply = activities.reduce( + (total: number, activity: { type: string; payload: any }) => { + const amount = Number(activity.payload?.amount ?? 0); + return activity.type === 'KEY_BOUGHT' + ? total + amount + : total - amount; + }, + 0 + ); + await tx.creatorProfile.update({ + where: { id: creatorId }, + data: { + circulatingSupply: supply, + currentMilestone: getTierForSupply(supply), + }, + }); + } + + // 6. Drop the reorged checkpoints so the ledgers can be re-ingested. + await tx.ledgerCheckpoint.deleteMany({ + where: { ledger: { gt: forkPoint } }, + }); + + return { reorgedLedgerCount: reorgedLedgers, affectedEventCount }; + }, + { timeout: ROLLBACK_TX_TIMEOUT_MS, maxWait: ROLLBACK_TX_MAX_WAIT_MS } + ); +} + +/** + * Rebuild one (owner, creator) position by replaying the surviving trade + * activities in ingestion order, using the same math as + * `recordKeyPurchase` / `recordKeySale` (ownership.service). Non-derived + * fields (frozen, lockup) on an existing row are preserved. + */ +async function rebuildOwnershipPosition( + tx: Tx, + ownerAddress: string, + creatorId: string +): Promise { + const activities = await tx.activity.findMany({ + where: { + creatorId, + actor: ownerAddress, + type: { in: [...TRADE_ACTIVITY_TYPES] }, + }, + orderBy: [{ createdAt: 'asc' }, { id: 'asc' }], + select: { type: true, payload: true, createdAt: true }, + }); + + let balance = 0; + let costBasis = 0; + let realisedPnl = 0; + let lastBuyAt: Date | null = null; + + for (const activity of activities) { + const payload = activity.payload as { + amount?: number; + price_at_trade?: string; + }; + const amount = Number(payload?.amount ?? 0); + // Activity stores the per-key price in stroops; the ownership math + // works in XLM (same conversion as indexer-pipeline.service). + let pricePerKeyXlm = Number(payload?.price_at_trade ?? 0) / 10_000_000; + if (!Number.isFinite(pricePerKeyXlm) || pricePerKeyXlm < 0) { + pricePerKeyXlm = 0; + } + if (!Number.isFinite(amount) || amount === 0) { + continue; + } + + if (activity.type === 'KEY_BOUGHT') { + const previousBalance = balance; + const newBalance = Math.max(0, balance + amount); + if (newBalance === 0) { + costBasis = 0; + } else if (previousBalance <= 0) { + costBasis = pricePerKeyXlm; + } else { + costBasis = + (costBasis * previousBalance + pricePerKeyXlm * amount) / + newBalance; + } + balance += amount; + lastBuyAt = activity.createdAt; + } else { + if (balance < amount) { + // Mirror the pipeline's fallback for out-of-sync history: + // balance-only decrement without P&L accounting. + balance -= amount; + continue; + } + realisedPnl += (pricePerKeyXlm - costBasis) * amount; + balance -= amount; + if (balance <= 0) { + costBasis = 0; + } + } + } + + const existing = await tx.keyOwnership.findUnique({ + where: { ownerAddress_creatorId: { ownerAddress, creatorId } }, + select: { id: true }, + }); + + if (!existing && activities.length === 0) { + // Position never existed from the trade history's point of view. + return; + } + + if (existing) { + await tx.keyOwnership.update({ + where: { ownerAddress_creatorId: { ownerAddress, creatorId } }, + data: { balance, costBasis, realisedPnl, lastBuyAt }, + }); + } else { + await tx.keyOwnership.create({ + data: { ownerAddress, creatorId, balance, costBasis, realisedPnl, lastBuyAt }, + }); + } +} + +/** + * Reorg-aware ingestion entry point: ingest one ledger, verifying that it + * chains onto the last ingested ledger. + * + * Flow: + * 1. Compare `header.previousLedgerHash` with the stored hash of ledger + * `header.sequence - 1`. (No checkpoint → nothing to verify yet, e.g. + * first run or a backfill gap — gap handling is #420's domain.) + * 2. On mismatch: walk back to the fork point, roll everything after it + * back in one atomic transaction, emit the reorg warn log, and re-ingest + * the canonical ledgers from the fork point forward. + * 3. Ingest the incoming ledger's events through the trade pipeline and + * record the checkpoint + ingested-event rows. + */ +export async function ingestLedger( + header: LedgerHeader, + events: IndexerChainEvent[], + deps: ReorgGuardDeps = {} +): Promise { + const processEvents = deps.processEvents ?? processTradeEvents; + const uniqueEvents = dedupeChainEvents(events); + + const reorgOutcome: ReorgOutcome = { + reorgDetected: false, + forkPoint: null, + reorgedLedgerCount: 0, + affectedEventCount: 0, + }; + const reingestedLedgers: number[] = []; + + const storedParent = await prisma.ledgerCheckpoint.findUnique({ + where: { ledger: header.sequence - 1 }, + select: { ledgerHash: true }, + }); + + if ( + storedParent && + storedParent.ledgerHash && + header.previousLedgerHash && + header.previousLedgerHash !== storedParent.ledgerHash + ) { + const forkPoint = await findForkPoint(header, deps.fetchHeader); + const { reorgedLedgerCount, affectedEventCount } = + await rollbackToLedger(forkPoint); + + // Warn-level structured log emitted on every reorg (#724). + logger.warn( + { + type: 'ledger_reorg_detected', + forkPoint, + reorgedLedgerCount, + affectedEventCount, + incomingSequence: header.sequence, + storedParentHash: storedParent.ledgerHash, + incomingParentHash: header.previousLedgerHash, + detectedAt: new Date().toISOString(), + }, + 'Ledger reorg detected — rolled back database state from the fork point' + ); + + reorgOutcome.reorgDetected = true; + reorgOutcome.forkPoint = forkPoint; + reorgOutcome.reorgedLedgerCount = reorgedLedgerCount; + reorgOutcome.affectedEventCount = affectedEventCount; + + // Re-ingest the canonical ledgers between the fork point and the + // incoming ledger when the caller can supply them. + if (deps.fetchHeader && deps.fetchEvents) { + for (let seq = forkPoint + 1; seq < header.sequence; seq++) { + const canonical = await deps.fetchHeader(seq); + if (!canonical) { + logger.warn( + { type: 'ledger_reorg_reingest_skipped', ledger: seq }, + 'Cannot re-fetch canonical header during reorg recovery; skipping re-ingest' + ); + break; + } + const canonicalEvents = await deps.fetchEvents(seq); + await processEvents(canonicalEvents); + await recordLedgerIngestion(canonical, canonicalEvents); + reingestedLedgers.push(seq); + } + } else { + logger.warn( + { + type: 'ledger_reorg_reingest_deferred', + forkPoint, + fromLedger: forkPoint + 1, + toLedger: header.sequence - 1, + }, + 'No re-ingest source configured; caller must re-deliver ledgers from the fork point' + ); + } + } + + // Ingest the incoming ledger (fresh, or post-reorg canonical delivery). + await processEvents(uniqueEvents); + await recordLedgerIngestion(header, uniqueEvents); + + return { + ...reorgOutcome, + sequence: header.sequence, + eventsIngested: uniqueEvents.length, + reingestedLedgers, + }; +} From 7c1cd84c14047dca371e12d5376cd8790a5e9654 Mon Sep 17 00:00:00 2001 From: Codebuff Date: Wed, 21 Oct 2026 15:39:22 +0000 Subject: [PATCH 2/2] fix(invoice): return latestLedger from the invoice-protocol RPC stub MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #724 ledger-depth gate reads latestLedger from getTransaction, which a real Soroban node always returns. The invoice-protocol integration stub omitted it, so every buy/sell waited out the full 120s poll budget and CI timed out; report a head two ledgers past inclusion instead. 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- .../integration/invoice-protocol.integration.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/__tests__/integration/invoice-protocol.integration.test.ts b/src/__tests__/integration/invoice-protocol.integration.test.ts index 2b0ab8b1..87d9884d 100644 --- a/src/__tests__/integration/invoice-protocol.integration.test.ts +++ b/src/__tests__/integration/invoice-protocol.integration.test.ts @@ -266,7 +266,10 @@ describeDatabase('invoice protocol against PostgreSQL', () => { return { result: { status: 'PENDING', hash } }; } if (body.method === 'getTransaction') - return { result: { status: 'SUCCESS', ledger: 901 } }; + // `latestLedger` is part of a real getTransaction response and is + // what the #724 ledger-depth gate reads: report the network head + // two ledgers past inclusion so buys/sells confirm immediately. + return { result: { status: 'SUCCESS', ledger: 901, latestLedger: 903 } }; throw new Error(`Unexpected RPC method ${body.method}`); }); });