From f37141589ad3fd9f98f5120317e4b84930687efa Mon Sep 17 00:00:00 2001 From: Salmatcre8 Date: Sun, 30 Aug 2026 10:12:44 +0100 Subject: [PATCH 1/2] feat(indexer): one loop per configured network (#161, #160) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit startIndexer() ran a single loop over module-global state: totalIndexed, pollCycleCount and startedAt were module-level `let`s, the SAC/NFT watch lists were resolved once at import, and one source switcher was shared by everything. Indexing two networks in one process was impossible. Closes #161. Also closes #160, because #161 cannot be built without it: the loop needs getRpc(network), and shipping the loop against the old global singleton would have produced code that could not be tested. ## Per-network RPC (#160) getRpc(network) caches one client per network. Endpoints resolve from SOROBAN_RPC_URL_TESTNET / _MAINNET, then the unsuffixed SOROBAN_RPC_URL, then the public testnet default; mainnet has no free public RPC so it throws rather than guessing. The unsuffixed variable deliberately applies **only** to the network named by STELLAR_NETWORK. Honouring it for both would let a mainnet loop connect to a testnet endpoint, index happily, and write testnet ledgers tagged network='mainnet' — a corruption with no error anywhere. Scoping it keeps every existing single-network deployment byte-identical while making that mix-up impossible. Covered by a test; removing the scope fails it. ## Per-network loops (#161) Each loop owns a LoopState: its own counters, watch lists, source switcher and cursor. Nothing is shared, which matters beyond tidiness — two loops incrementing one counter make /status unable to say how far either chain got, and a shared source switcher has a mutable `preferred` field, so a testnet RPC outage would silently repoint the mainnet loop at testnet Horizon. Everything that names a chain is now per-network with a shared fallback: SAC_CONTRACT_IDS_*, NFT_CONTRACT_IDS_*, HORIZON_URL_*, START_LEDGER_*. START_LEDGER especially — the same sequence number is a completely different point in history on each chain. NETWORKS=testnet,mainnet opts in; unset means one loop on STELLAR_NETWORK, exactly as before. Unrecognised entries are dropped rather than throwing, and duplicates collapse so two loops never fight over one cursor. Loops are fault-isolated: a crash restarts that loop after 10s instead of exiting the process, because an expiring mainnet RPC key must not stop testnet indexing or take the API down with it. ## Writes that were silently untagged Two paths wrote rows without a network. The column defaults to 'testnet', so both compiled, typechecked and ran while filing mainnet rows as testnet: - upsertHostFnLogs() never set it. Now stamped. - pollParallel() / runPartitionWorker() called upsertTransfers and setLastIndexedLedger unscoped. Now threaded. Also fixed: fetchNftMetadata picked its passphrase from process.env directly, so a mainnet metadata simulation would have been built with the testnet passphrase; and sac-detect's cache was keyed by contract id alone, which is shared mutable state between loops. ## /status Adds a `networks` object with per-network lastIndexedLedger, latestLedger, lag and loop stats. Existing top-level fields are untouched, so current consumers — including tests/chaos/db-restart.test.ts's StatusResponse — keep working. A network with no running loop reports running:false rather than zeroes that look like progress. Verified: tsc --noEmit clean, 290 tests pass (was 270), 20 of them new. Integration tests checked for the shape they assert; the change is additive. Claude-Session: https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB --- docs/DUAL_NETWORK.md | 22 +- src/__tests__/accountSummary.test.ts | 4 + src/__tests__/fetchEventsSafe.test.ts | 4 +- src/__tests__/graphql.test.ts | 4 + src/__tests__/multiNetworkIndexer.test.ts | 198 ++++++++++++++++ src/__tests__/routes/transfers.test.ts | 4 + src/__tests__/staleReads.test.ts | 4 + src/__tests__/webhooks.test.ts | 4 + src/api.ts | 36 ++- src/index.ts | 10 +- src/indexer.ts | 268 ++++++++++++++++------ src/indexer/host-fn-log.ts | 11 +- src/indexer/parallel.ts | 14 +- src/indexer/sac-detect.ts | 33 ++- src/indexer/sources/index.ts | 5 +- src/indexer/sources/rpc.ts | 13 +- src/ingester/nft.ts | 15 +- src/network.ts | 25 ++ src/rpc.ts | 114 +++++---- 19 files changed, 635 insertions(+), 153 deletions(-) create mode 100644 src/__tests__/multiNetworkIndexer.test.ts diff --git a/docs/DUAL_NETWORK.md b/docs/DUAL_NETWORK.md index 70f9e305..e13abe7e 100644 --- a/docs/DUAL_NETWORK.md +++ b/docs/DUAL_NETWORK.md @@ -24,6 +24,7 @@ behave exactly as before; #161 and #163 pass it explicitly. | Var | testnet | mainnet | |-----|---------|---------| +| `NETWORKS` | `testnet` | `testnet,mainnet` to index both in one process | | `STELLAR_NETWORK` | `testnet` | `mainnet` | | `SOROBAN_RPC_URL` | `https://soroban-testnet.stellar.org` | external provider endpoint (**secret — host env only**) | | `SAC_CONTRACT_IDS` | testnet SAC `CDMLFMKM…` | mainnet XLM SAC `CDLZFC3SY…` | @@ -33,6 +34,23 @@ behave exactly as before; #161 and #163 pass it explicitly. > Mainnet has **no free public Soroban RPC** — an external provider endpoint is > required. Never commit the endpoint/key; it lives only in host secrets. +### In-process dual-network (#160, #161) + +Set `NETWORKS=testnet,mainnet` to run one indexer loop per network in a single +process. Each loop owns its cursor, counters, watch list, RPC client and source +switcher, so neither can stall or repoint the other, and `/status` reports both +under `networks`. + +Every setting that names a chain takes a per-network suffix, falling back to the +shared name: `SOROBAN_RPC_URL_MAINNET`, `HORIZON_URL_MAINNET`, +`SAC_CONTRACT_IDS_MAINNET`, `NFT_CONTRACT_IDS_MAINNET`, `START_LEDGER_MAINNET` +(and the `_TESTNET` equivalents). + +> The **unsuffixed** `SOROBAN_RPC_URL` applies only to the network named by +> `STELLAR_NETWORK`. That is deliberate: honouring it for both would let a +> mainnet loop connect to a testnet endpoint and write testnet ledgers tagged +> `network='mainnet'`. Indexing mainnet requires `SOROBAN_RPC_URL_MAINNET`. + ## Ordered work (next Wave) Dependencies: **#159 → #161** and #160 before #161. @@ -40,8 +58,8 @@ Dependencies: **#159 → #161** and #160 before #161. | # | Issue | Dep | |---|-------|-----| | ~~[#159](../../issues/159)~~ | ~~`network` column across all Prisma models~~ (done) | — | -| [#160](../../issues/160) | Per-network `getRpc(network)` factory | — | -| [#161](../../issues/161) | One indexer loop per network | #159, #160 | +| ~~[#160](../../issues/160)~~ | ~~Per-network `getRpc(network)` factory~~ (done) | — | +| ~~[#161](../../issues/161)~~ | ~~One indexer loop per network~~ (done) | #159, #160 | | [#162](../../issues/162) | Per-network SAC/NFT watch-lists | — | | [#163](../../issues/163) | `network` selector on REST/GraphQL/WS | #159–#161 | | [#164](../../issues/164) | Serve stale cached data instead of 503 | — | diff --git a/src/__tests__/accountSummary.test.ts b/src/__tests__/accountSummary.test.ts index 7e7c1a1c..fc4ab83d 100644 --- a/src/__tests__/accountSummary.test.ts +++ b/src/__tests__/accountSummary.test.ts @@ -26,6 +26,10 @@ jest.mock("../rpc", () => ({ })); jest.mock("../indexer", () => ({ + // #161: /status also reads per-network loop state. Listed explicitly + // because a partial mock silently 500s the route rather than failing loudly. + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), getIndexerStats: jest.fn().mockReturnValue({ startedAt: "2024-01-01T00:00:00Z", uptimeSeconds: 0, totalIndexed: 0 }), })); diff --git a/src/__tests__/fetchEventsSafe.test.ts b/src/__tests__/fetchEventsSafe.test.ts index 01e925ea..587dd74c 100644 --- a/src/__tests__/fetchEventsSafe.test.ts +++ b/src/__tests__/fetchEventsSafe.test.ts @@ -115,6 +115,8 @@ describe('fetchEventsSafe — bisection algorithm', () => { await fetchEventsSafe(100, 100, contracts, 5_000, fetch as any) - expect(fetch).toHaveBeenCalledWith(100, contracts, 5_000) + // fetchEventsSafe now forwards the network as a 4th argument (#161); + // undefined here means "the configured network", the single-network default. + expect(fetch).toHaveBeenCalledWith(100, contracts, 5_000, undefined) }) }) diff --git a/src/__tests__/graphql.test.ts b/src/__tests__/graphql.test.ts index 8b4c9386..b4160712 100644 --- a/src/__tests__/graphql.test.ts +++ b/src/__tests__/graphql.test.ts @@ -34,6 +34,10 @@ jest.mock("../rpc", () => ({ })); jest.mock("../indexer", () => ({ + // #161: /status also reads per-network loop state. Listed explicitly + // because a partial mock silently 500s the route rather than failing loudly. + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), getIndexerStats: jest.fn().mockReturnValue({ uptimeSeconds: 0, totalIndexed: 0 }), })); diff --git a/src/__tests__/multiNetworkIndexer.test.ts b/src/__tests__/multiNetworkIndexer.test.ts new file mode 100644 index 00000000..da0f767b --- /dev/null +++ b/src/__tests__/multiNetworkIndexer.test.ts @@ -0,0 +1,198 @@ +/** + * Multi-network indexer tests (#161, #160). + * + * The acceptance criteria are about *isolation*, and isolation bugs are quiet: + * two loops sharing a counter still index correctly, they just report nonsense; + * two loops sharing an RPC client still fetch events, just from one chain. So + * these assert that per-network things are actually distinct, rather than that + * the code runs. + */ + +import { + DEFAULT_XLM_SAC_MAINNET, + DEFAULT_XLM_SAC_TESTNET, + getIndexerStats, + resolveNftContractIds, + resolveSacContractIds, + runningNetworks, + _resetIndexerLoops, +} from "../indexer"; +import { getRpc, validateNetworkConfig, _resetRpcClients } from "../rpc"; +import { currentNetwork, enabledNetworks, parseNetwork } from "../network"; + +const ENV_KEYS = [ + "NETWORKS", + "STELLAR_NETWORK", + "SAC_CONTRACT_IDS", + "SAC_CONTRACT_IDS_TESTNET", + "SAC_CONTRACT_IDS_MAINNET", + "CONTRACT_IDS", + "NFT_CONTRACT_IDS", + "NFT_CONTRACT_IDS_TESTNET", + "NFT_CONTRACT_IDS_MAINNET", + "SOROBAN_RPC_URL", + "STELLAR_RPC_URL", + "SOROBAN_RPC_URL_TESTNET", + "SOROBAN_RPC_URL_MAINNET", +]; + +beforeEach(() => { + for (const key of ENV_KEYS) delete process.env[key]; + _resetRpcClients(); + _resetIndexerLoops(); +}); + +describe("enabledNetworks", () => { + it("defaults to the single configured network, so existing deployments are unchanged", () => { + expect(enabledNetworks()).toEqual(["testnet"]); + + process.env.STELLAR_NETWORK = "mainnet"; + expect(enabledNetworks()).toEqual(["mainnet"]); + }); + + it("parses NETWORKS into an ordered list", () => { + process.env.NETWORKS = "testnet,mainnet"; + expect(enabledNetworks()).toEqual(["testnet", "mainnet"]); + }); + + it("tolerates whitespace and case", () => { + process.env.NETWORKS = " MAINNET , testnet "; + expect(enabledNetworks()).toEqual(["mainnet", "testnet"]); + }); + + it("de-duplicates — two loops on one network would fight over the same cursor", () => { + process.env.NETWORKS = "testnet,testnet"; + expect(enabledNetworks()).toEqual(["testnet"]); + }); + + it("drops unrecognised entries rather than starting a loop for them", () => { + process.env.NETWORKS = "testnet,futurenet"; + expect(enabledNetworks()).toEqual(["testnet"]); + expect(parseNetwork("futurenet")).toBeNull(); + }); + + it("falls back to the configured network when NETWORKS is empty or all junk", () => { + process.env.STELLAR_NETWORK = "mainnet"; + process.env.NETWORKS = " , "; + expect(enabledNetworks()).toEqual(["mainnet"]); + + process.env.NETWORKS = "nope,alsonope"; + expect(enabledNetworks()).toEqual(["mainnet"]); + }); +}); + +describe("per-network watch lists", () => { + it("defaults each network to its own native XLM SAC", () => { + // The bug this prevents: reading STELLAR_NETWORK inside the resolver, so + // both loops watch the same chain's SAC and one indexes nothing. + expect(resolveSacContractIds("testnet")).toEqual([DEFAULT_XLM_SAC_TESTNET]); + expect(resolveSacContractIds("mainnet")).toEqual([DEFAULT_XLM_SAC_MAINNET]); + expect(DEFAULT_XLM_SAC_TESTNET).not.toEqual(DEFAULT_XLM_SAC_MAINNET); + }); + + it("keeps the process-wide default when no network is passed", () => { + process.env.STELLAR_NETWORK = "mainnet"; + expect(resolveSacContractIds()).toEqual([DEFAULT_XLM_SAC_MAINNET]); + }); + + it("prefers the per-network env var over the shared one", () => { + process.env.SAC_CONTRACT_IDS = "CSHARED"; + process.env.SAC_CONTRACT_IDS_MAINNET = "CMAIN1,CMAIN2"; + + expect(resolveSacContractIds("mainnet")).toEqual(["CMAIN1", "CMAIN2"]); + // testnet has no override, so it still sees the shared value + expect(resolveSacContractIds("testnet")).toEqual(["CSHARED"]); + }); + + it("still honours the legacy CONTRACT_IDS alias", () => { + process.env.CONTRACT_IDS = "CLEGACY"; + expect(resolveSacContractIds("testnet")).toEqual(["CLEGACY"]); + }); + + it("resolves NFT watch lists per network", () => { + process.env.NFT_CONTRACT_IDS = "CNFT_SHARED"; + process.env.NFT_CONTRACT_IDS_TESTNET = "CNFT_T"; + + expect(resolveNftContractIds("testnet")).toEqual(["CNFT_T"]); + expect(resolveNftContractIds("mainnet")).toEqual(["CNFT_SHARED"]); + }); +}); + +describe("per-network RPC clients (#160)", () => { + it("returns one cached client per network, and never shares between them", () => { + process.env.SOROBAN_RPC_URL_TESTNET = "https://testnet.example/rpc"; + process.env.SOROBAN_RPC_URL_MAINNET = "https://mainnet.example/rpc"; + + const testnet = getRpc("testnet"); + const mainnet = getRpc("mainnet"); + + // Same network → same instance (cached, so we don't open a pool per call) + expect(getRpc("testnet")).toBe(testnet); + // Different network → different instance. Sharing one was the whole bug. + expect(mainnet).not.toBe(testnet); + }); + + it("scopes the legacy unsuffixed SOROBAN_RPC_URL to the configured network only", () => { + // The dangerous case: a single-network deployment sets SOROBAN_RPC_URL for + // testnet, then enables mainnet. If the legacy var applied to both, the + // mainnet loop would connect to testnet RPC, index happily, and write + // testnet ledgers tagged network='mainnet'. It must fail loudly instead. + process.env.STELLAR_NETWORK = "testnet"; + process.env.SOROBAN_RPC_URL = "https://testnet.example/rpc"; + + expect(() => getRpc("testnet")).not.toThrow(); + expect(() => getRpc("mainnet")).toThrow(/SOROBAN_RPC_URL_MAINNET is required/); + }); + + it("still lets a single-network mainnet deployment use the unsuffixed var", () => { + process.env.STELLAR_NETWORK = "mainnet"; + process.env.SOROBAN_RPC_URL = "https://mainnet.example/rpc"; + + expect(() => getRpc("mainnet")).not.toThrow(); + }); + + it("defaults testnet to the public endpoint but refuses to guess for mainnet", () => { + // There is no free public mainnet Soroban RPC, so guessing would produce a + // client that fails on every call instead of a clear config error. + expect(() => getRpc("testnet")).not.toThrow(); + expect(() => getRpc("mainnet")).toThrow(/no free public Soroban RPC|SOROBAN_RPC_URL_MAINNET/); + }); + + it("validateNetworkConfig checks every network it is given", () => { + process.env.SOROBAN_RPC_URL_TESTNET = "https://testnet.example/rpc"; + + expect(() => validateNetworkConfig(["testnet"])).not.toThrow(); + // Fails at startup rather than after testnet has begun writing. + expect(() => validateNetworkConfig(["testnet", "mainnet"])).toThrow(); + }); +}); + +describe("loop state isolation", () => { + it("reports no running loops before any are started", () => { + expect(runningNetworks()).toEqual([]); + }); + + it("reports zero indexed for a network with no loop, rather than another network's total", () => { + // Guards the shape of the failure: an un-started loop must not inherit + // whatever the other loop has counted. + expect(getIndexerStats("mainnet").totalIndexed).toBe(0); + expect(getIndexerStats("testnet").totalIndexed).toBe(0); + }); + + it("getIndexerStats keeps its original shape for existing /status consumers", () => { + const stats = getIndexerStats(); + expect(stats).toEqual({ + startedAt: expect.any(String), + uptimeSeconds: expect.any(Number), + totalIndexed: expect.any(Number), + }); + expect(new Date(stats.startedAt).toString()).not.toBe("Invalid Date"); + }); + + it("resolves the default network from the environment on every call", () => { + process.env.STELLAR_NETWORK = "mainnet"; + expect(currentNetwork()).toBe("mainnet"); + process.env.STELLAR_NETWORK = "testnet"; + expect(currentNetwork()).toBe("testnet"); + }); +}); diff --git a/src/__tests__/routes/transfers.test.ts b/src/__tests__/routes/transfers.test.ts index 425135c9..b9fecf05 100644 --- a/src/__tests__/routes/transfers.test.ts +++ b/src/__tests__/routes/transfers.test.ts @@ -16,6 +16,10 @@ jest.mock("../../rpc", () => ({ })); jest.mock("../../indexer", () => ({ + // #161: /status also reads per-network loop state. Listed explicitly + // because a partial mock silently 500s the route rather than failing loudly. + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), getIndexerStats: jest .fn() .mockReturnValue({ startedAt: "2024-01-01T00:00:00.000Z", uptimeSeconds: 0, totalIndexed: 0 }), diff --git a/src/__tests__/staleReads.test.ts b/src/__tests__/staleReads.test.ts index faff2368..ef2d3224 100644 --- a/src/__tests__/staleReads.test.ts +++ b/src/__tests__/staleReads.test.ts @@ -16,6 +16,10 @@ jest.mock("../rpc", () => ({ })); jest.mock("../indexer", () => ({ + // #161: /status also reads per-network loop state. Listed explicitly + // because a partial mock silently 500s the route rather than failing loudly. + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), getIndexerStats: jest .fn() .mockReturnValue({ startedAt: "2024-01-01T00:00:00.000Z", uptimeSeconds: 100, totalIndexed: 50 }), diff --git a/src/__tests__/webhooks.test.ts b/src/__tests__/webhooks.test.ts index f8e3a6af..e8cc7db7 100644 --- a/src/__tests__/webhooks.test.ts +++ b/src/__tests__/webhooks.test.ts @@ -48,6 +48,10 @@ jest.mock("../rpc", () => ({ })); jest.mock("../indexer", () => ({ + // #161: /status also reads per-network loop state. Listed explicitly + // because a partial mock silently 500s the route rather than failing loudly. + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), getIndexerStats: jest.fn().mockReturnValue({ startedAt: "2024-01-01T00:00:00Z", uptimeSeconds: 0, diff --git a/src/api.ts b/src/api.ts index 714fc519..042e3695 100644 --- a/src/api.ts +++ b/src/api.ts @@ -5,7 +5,8 @@ import { jsonApiMiddleware } from "./middleware/jsonapi"; import { queryHostFnLogs } from "./indexer/host-fn-log"; import { queryTransfers, queryAllTransfers, queryByTxHash, querySummary, queryNftTransfers, getNftOwner, getNftMetadata, getLastIndexedLedger, prisma } from "./db"; import { getLatestLedger } from "./rpc"; -import { getIndexerStats } from "./indexer"; +import { getIndexerStats, getAllIndexerStats, runningNetworks } from "./indexer"; +import { enabledNetworks, type Network } from "./network"; import { createAccountsRouter } from "./api/accounts"; import { createWebhooksRouter } from "./api/webhooks"; import { createGraphQLMiddleware } from "./graphql/server"; @@ -304,6 +305,38 @@ export function createApp(): express.Application { const latestLedger = rpcSuccess ? latestLedgerResult.value : null; const stats = getIndexerStats(); + // Per-network progress (#161). The top-level fields above stay as they + // were so existing consumers are unaffected; this reports each loop + // separately, which is the only way to see one chain falling behind + // while the other is healthy. + const networks: Network[] = runningNetworks().length > 0 ? runningNetworks() : enabledNetworks(); + const loopStats = getAllIndexerStats(); + const perNetwork = await Promise.all( + networks.map(async (net) => { + const [indexed, tip] = await Promise.allSettled([ + getLastIndexedLedger(net), + getLatestLedger(net), + ]); + const indexedLedger = indexed.status === "fulfilled" ? indexed.value : null; + const tipLedger = + tip.status === "fulfilled" && typeof tip.value === "number" ? tip.value : null; + return [ + net, + { + lastIndexedLedger: indexedLedger, + latestLedger: tipLedger, + lagLedgers: + tipLedger !== null ? tipLedger - (indexedLedger ?? tipLedger) : null, + // A loop that has not started reports no stats rather than zeroes + // pretending to be progress. + running: loopStats[net] !== undefined, + ...(loopStats[net] ?? {}), + }, + ] as const; + }) + ); + const byNetwork = Object.fromEntries(perNetwork); + if (rpcSuccess && latestLedger !== null) { res.json({ ok: true, @@ -312,6 +345,7 @@ export function createApp(): express.Application { latestLedger, lagLedgers: latestLedger - (lastIndexedLedger ?? latestLedger), ...stats, + networks: byNetwork, }); } else { res.setHeader("X-Data-Stale", "true"); diff --git a/src/index.ts b/src/index.ts index 06e63c97..912b1287 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,7 +2,7 @@ import "dotenv/config"; import http from "http"; import { execSync } from "child_process"; import { createApp } from "./api"; -import { startIndexer } from "./indexer"; +import { startAllIndexers } from "./indexer"; import { prisma } from "./db"; import { attachWebSocketServer } from "./ws"; import { attachGraphQLSubscriptions, SUBSCRIPTIONS_PATH } from "./graphql/subscriptions"; @@ -60,10 +60,10 @@ async function main() { if (process.env.SKIP_INDEXER === "true") { console.log("[wraith] SKIP_INDEXER=true — indexer not started (API-only mode)"); } else { - startIndexer().catch((err) => { - console.error("[wraith] Indexer crashed — exiting:", err); - process.exit(1); - }); + // One loop per enabled network (NETWORKS env; defaults to STELLAR_NETWORK). + // Each loop restarts itself on crash rather than taking the process down — + // a mainnet RPC key expiring must not stop testnet indexing, nor the API. + startAllIndexers(); } } diff --git a/src/indexer.ts b/src/indexer.ts index d21b9da9..10535ca9 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -16,15 +16,17 @@ import { parseHostFnEvent, upsertHostFnLogs, type HostFnRecord } from "./indexer import { tagSacTransfers } from "./indexer/sac-detect"; import { pollParallel } from "./indexer/parallel"; import { isNftTransferEvent, parseNftEvents, fetchNftMetadata } from "./ingester/nft"; -import { createSourceSwitcherWithConfig } from "./indexer/sources"; +import { createSourceSwitcherWithConfig, type SourceSwitcher } from "./indexer/sources"; +import { currentNetwork, enabledNetworks, resolveNetwork, type Network } from "./network"; // ─── NFT Contract IDs ───────────────────────────────────────────────────────── /** * Resolve the list of NFT contract IDs to watch. * Falls back to empty — NFT events can still be auto-detected by topic structure. */ -export function resolveNftContractIds(): string[] { - const raw = process.env.NFT_CONTRACT_IDS ?? ""; +export function resolveNftContractIds(network?: Network): string[] { + const suffix = resolveNetwork(network).toUpperCase(); + const raw = process.env[`NFT_CONTRACT_IDS_${suffix}`] ?? process.env.NFT_CONTRACT_IDS ?? ""; return raw.split(",").map((s) => s.trim()).filter(Boolean); } @@ -48,8 +50,12 @@ export const DEFAULT_XLM_SAC_TESTNET = * The native XLM SAC default depends on STELLAR_NETWORK ("mainnet" | "testnet"). * Any unset / empty value falls through to the next tier. */ -export function resolveSacContractIds(): string[] { +export function resolveSacContractIds(network?: Network): string[] { + const net = resolveNetwork(network); + const suffix = net.toUpperCase(); + const raw = + process.env[`SAC_CONTRACT_IDS_${suffix}`] || process.env.SAC_CONTRACT_IDS || process.env.CONTRACT_IDS || ""; @@ -63,29 +69,17 @@ export function resolveSacContractIds(): string[] { return ids; } - // Fall back to the native XLM SAC for the configured network. - const network = (process.env.STELLAR_NETWORK ?? "testnet").toLowerCase(); - return [ - network === "mainnet" ? DEFAULT_XLM_SAC_MAINNET : DEFAULT_XLM_SAC_TESTNET, - ]; + // Fall back to the native XLM SAC for *this* network, not the process-wide + // one — with a loop per network, reading STELLAR_NETWORK here would point + // both loops at the same chain's SAC. + return [net === "mainnet" ? DEFAULT_XLM_SAC_MAINNET : DEFAULT_XLM_SAC_TESTNET]; } // ─── Config ─────────────────────────────────────────────────────────────────── +// These stay process-wide: they describe how hard to poll, not which chain. const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS ?? "6000", 10); const BATCH_SIZE = parseInt(process.env.EVENTS_BATCH_SIZE ?? "10000", 10); const INGEST_WORKERS = parseInt(process.env.INGEST_WORKERS ?? "1", 10); -const SAC_CONTRACT_IDS = resolveSacContractIds(); -const NFT_CONTRACT_IDS = resolveNftContractIds(); -// Combined watch list — deduplicated so we don't request the same contract twice -const ALL_CONTRACT_IDS = [...new Set([...SAC_CONTRACT_IDS, ...NFT_CONTRACT_IDS])]; -const sourceSwitcher = createSourceSwitcherWithConfig({ - horizonUrl: process.env.HORIZON_URL, - horizonEventsPath: process.env.HORIZON_EVENTS_PATH, - fetchImpl: (globalThis as { fetch?: (input: string, init?: unknown) => Promise }).fetch as unknown as ( - input: string, - init?: { headers?: Record } - ) => Promise<{ ok: boolean; status: number; json(): Promise }>, -}); // Stellar testnet RPC retains ~7 days ≈ 120 000 ledgers (at ~5s per ledger). // We cap the back-fill look-back so we never request a ledger that's already pruned. @@ -95,41 +89,126 @@ const RPC_MAX_LOOKBACK_LEDGERS = 100_000; // reading ledgers that haven't fully propagated yet. const TIP_LAG = 2; -// ─── State ──────────────────────────────────────────────────────────────────── -let startedAt = Date.now(); -let totalIndexed = 0; - // Prune old data every ~1 hour (600 poll cycles × 6s = 3600s) const PRUNE_EVERY_CYCLES = 600; -let pollCycleCount = 0; -export function getIndexerStats() { +// ─── Per-network loop state ─────────────────────────────────────────────────── +/** + * Everything one indexer loop owns. + * + * This used to be module-level `let`s (`startedAt`, `totalIndexed`, + * `pollCycleCount`) plus module-level watch lists and one source switcher. With + * a loop per network that becomes a correctness problem, not a tidiness one: + * two loops would increment the same counters, so `/status` could not say how + * far either chain had actually got, and they would share one source switcher + * whose failover `preferred` field is mutable — a testnet RPC outage would + * silently repoint the mainnet loop at testnet Horizon. + */ +type LoopState = { + network: Network; + sacContractIds: string[]; + nftContractIds: string[]; + /** Deduplicated union — we never request the same contract twice. */ + allContractIds: string[]; + sourceSwitcher: SourceSwitcher; + startedAt: number; + totalIndexed: number; + pollCycleCount: number; +}; + +const loops = new Map(); + +/** Build the isolated state for one network's loop. */ +function createLoopState(network: Network): LoopState { + const sacContractIds = resolveSacContractIds(network); + const nftContractIds = resolveNftContractIds(network); + const suffix = network.toUpperCase(); + + return { + network, + sacContractIds, + nftContractIds, + allContractIds: [...new Set([...sacContractIds, ...nftContractIds])], + sourceSwitcher: createSourceSwitcherWithConfig({ + network, + // Horizon is per-network for the same reason as RPC: one shared + // HORIZON_URL would let a failover cross chains. + horizonUrl: process.env[`HORIZON_URL_${suffix}`] ?? process.env.HORIZON_URL, + horizonEventsPath: process.env.HORIZON_EVENTS_PATH, + fetchImpl: (globalThis as { fetch?: (input: string, init?: unknown) => Promise }).fetch as unknown as ( + input: string, + init?: { headers?: Record } + ) => Promise<{ ok: boolean; status: number; json(): Promise }>, + }), + startedAt: Date.now(), + totalIndexed: 0, + pollCycleCount: 0, + }; +} + +const PROCESS_STARTED_AT = Date.now(); + +export type IndexerStats = { + startedAt: string; + uptimeSeconds: number; + totalIndexed: number; +}; + +/** + * Stats for one network. The shape is unchanged from the single-loop version so + * existing `/status` consumers keep working; {@link getAllIndexerStats} is the + * per-network view. + */ +export function getIndexerStats(network?: Network): IndexerStats { + const loop = loops.get(resolveNetwork(network)); + const startedAt = loop?.startedAt ?? PROCESS_STARTED_AT; return { startedAt: new Date(startedAt).toISOString(), uptimeSeconds: Math.floor((Date.now() - startedAt) / 1000), - totalIndexed, + totalIndexed: loop?.totalIndexed ?? 0, }; } +/** Stats for every running loop, keyed by network. */ +export function getAllIndexerStats(): Record { + const out: Record = {}; + for (const [network, loop] of loops) { + out[network] = { ...getIndexerStats(network), watching: loop.allContractIds.length }; + } + return out; +} + +/** The networks with a loop currently running. */ +export function runningNetworks(): Network[] { + return [...loops.keys()]; +} + +/** Test-only: drops loop state between cases. */ +export function _resetIndexerLoops(): void { + loops.clear(); +} + // ─── Core poll step ─────────────────────────────────────────────────────────── /** * Fetch one batch of events starting from `fromLedger`, parse and persist them. * Returns the highest ledger sequence seen in the batch (or fromLedger if empty). */ async function pollOnce( + loop: LoopState, fromLedger: number, latestLedger: number ): Promise { + const net = loop.network; console.log( - `[indexer] Polling ledgers ${fromLedger} → ${latestLedger} (lag: ${latestLedger - fromLedger})` + `[indexer/${net}] Polling ledgers ${fromLedger} → ${latestLedger} (lag: ${latestLedger - fromLedger})` ); - const { events, highestLedger } = await sourceSwitcher.fetchEvents( - fromLedger, latestLedger, ALL_CONTRACT_IDS, BATCH_SIZE + const { events, highestLedger } = await loop.sourceSwitcher.fetchEvents( + fromLedger, latestLedger, loop.allContractIds, BATCH_SIZE ); if (events.length === 0) { - await setLastIndexedLedger(highestLedger); + await setLastIndexedLedger(highestLedger, net); return highestLedger; } @@ -142,16 +221,16 @@ async function pollOnce( const records = parseEvents(fungibleEvents); // Tag each transfer with whether its contract is a SAC (#136). Best-effort: // a detection failure must never block ingest, so default to false on error. - await tagSacTransfers(records).catch((e) => - console.error("[indexer] SAC detection failed:", e) + await tagSacTransfers(records, undefined, net).catch((e: unknown) => + console.error(`[indexer/${net}] SAC detection failed:`, e) ); - const inserted = await upsertTransfers(records); - totalIndexed += inserted; + const inserted = await upsertTransfers(records, net); + loop.totalIndexed += inserted; // Update materialized account summaries alongside transfer inserts if (inserted > 0) { - await upsertAccountSummaries(records).catch((e) => - console.error("[indexer] Account summary upsert failed:", e) + await upsertAccountSummaries(records, net).catch((e: unknown) => + console.error(`[indexer/${net}] Account summary upsert failed:`, e) ); } @@ -165,8 +244,8 @@ async function pollOnce( .map(raw => { try { return parseHostFnEvent(raw); } catch { return null; } }) .filter((r): r is HostFnRecord => r !== null); if (hostFnRecords.length > 0) { - await upsertHostFnLogs(hostFnRecords).catch(err => - console.error("[indexer] host-fn log error:", err), + await upsertHostFnLogs(hostFnRecords, net).catch((err: unknown) => + console.error(`[indexer/${net}] host-fn log error:`, err), ); hostFnRecords.forEach(emitHostFnLog); } @@ -174,8 +253,8 @@ async function pollOnce( // ── NFT path ───────────────────────────────────────────────────────────────── const nftParsed = parseNftEvents(nftRawEvents); const nftRecords = nftParsed.map((p) => p.record); - const nftInserted = await upsertNftTransfers(nftRecords); - totalIndexed += nftInserted; + const nftInserted = await upsertNftTransfers(nftRecords, net); + loop.totalIndexed += nftInserted; // Lazy-load metadata for unique (contractId, tokenId) pairs not yet cached if (nftParsed.length > 0) { @@ -184,69 +263,83 @@ async function pollOnce( const key = `${record.contractId}:${record.tokenId}`; if (seen.has(key)) continue; seen.add(key); - const cached = await getNftMetadata(record.contractId, record.tokenId); + const cached = await getNftMetadata(record.contractId, record.tokenId, net); if (!cached) { - const meta = await fetchNftMetadata(record.contractId, tokenIdScVal).catch(() => ({})); - await upsertNftMetadata(record.contractId, record.tokenId, meta).catch((e) => - console.error("[indexer] NFT metadata upsert failed:", e) + const meta = await fetchNftMetadata(record.contractId, tokenIdScVal, net).catch(() => ({})); + await upsertNftMetadata(record.contractId, record.tokenId, meta, net).catch((e: unknown) => + console.error(`[indexer/${net}] NFT metadata upsert failed:`, e) ); } } } - await setLastIndexedLedger(highestLedger); + await setLastIndexedLedger(highestLedger, net); console.log( - `[indexer] Processed ${events.length} events → ${inserted} fungible + ${nftInserted} NFT records saved (ledger ${highestLedger})` + `[indexer/${net}] Processed ${events.length} events → ${inserted} fungible + ${nftInserted} NFT records saved (ledger ${highestLedger})` ); return highestLedger; } // ─── Main loop ──────────────────────────────────────────────────────────────── -export async function startIndexer(): Promise { +/** + * Run one indexer loop for one network. Never returns. + * + * Each call owns its own {@link LoopState}, so two concurrent loops share no + * mutable state: separate counters, separate watch lists, separate source + * switchers, and separate cursors (IndexerState is keyed by network). + */ +export async function startIndexer(network?: Network): Promise { + const net = resolveNetwork(network); + // Fail fast if RPC is not configured — surfaces env errors before any DB work - validateNetworkConfig(); + validateNetworkConfig([net]); + + const loop = createLoopState(net); + loops.set(net, loop); - console.log("[indexer] Starting Wraith indexer…"); + console.log(`[indexer/${net}] Starting Wraith indexer…`); console.log( - `[indexer] Watching SAC contracts (${SAC_CONTRACT_IDS.length}): ${SAC_CONTRACT_IDS.join(", ")}` + `[indexer/${net}] Watching SAC contracts (${loop.sacContractIds.length}): ${loop.sacContractIds.join(", ")}` ); - if (NFT_CONTRACT_IDS.length > 0) { + if (loop.nftContractIds.length > 0) { console.log( - `[indexer] Watching NFT contracts (${NFT_CONTRACT_IDS.length}): ${NFT_CONTRACT_IDS.join(", ")}` + `[indexer/${net}] Watching NFT contracts (${loop.nftContractIds.length}): ${loop.nftContractIds.join(", ")}` ); } else { - console.log("[indexer] NFT auto-detection enabled (set NFT_CONTRACT_IDS for explicit watch)"); + console.log(`[indexer/${net}] NFT auto-detection enabled (set NFT_CONTRACT_IDS for explicit watch)`); } - startedAt = Date.now(); - // ── Determine start ledger ────────────────────────────────────────────────── - const latestLedger = await withRetry(() => sourceSwitcher.getLatestLedger()); + const latestLedger = await withRetry(() => loop.sourceSwitcher.getLatestLedger()); const minSafeLedger = latestLedger - RPC_MAX_LOOKBACK_LEDGERS; let currentLedger: number; - const envStart = process.env.START_LEDGER ? parseInt(process.env.START_LEDGER, 10) : null; - const dbLedger = await getLastIndexedLedger(); + // START_LEDGER is per-network too: the same sequence number means a + // completely different point in history on each chain. + const rawStart = + process.env[`START_LEDGER_${net.toUpperCase()}`] ?? process.env.START_LEDGER; + const envStart = rawStart ? parseInt(rawStart, 10) : null; + const dbLedger = await getLastIndexedLedger(net); if (envStart !== null && envStart > 0) { currentLedger = Math.max(envStart, minSafeLedger); - console.log(`[indexer] Starting from env START_LEDGER=${envStart} (clamped to ${currentLedger})`); + console.log(`[indexer/${net}] Starting from env START_LEDGER=${envStart} (clamped to ${currentLedger})`); } else if (dbLedger !== null) { currentLedger = Math.max(dbLedger, minSafeLedger); - console.log(`[indexer] Resuming from DB state: ledger ${dbLedger} (clamped to ${currentLedger})`); + console.log(`[indexer/${net}] Resuming from DB state: ledger ${dbLedger} (clamped to ${currentLedger})`); } else { // Fresh start — begin near the tip rather than trying to fetch all history. currentLedger = latestLedger - TIP_LAG; - console.log(`[indexer] No prior state — starting from tip: ledger ${currentLedger}`); + console.log(`[indexer/${net}] No prior state — starting from tip: ledger ${currentLedger}`); } // ── Polling loop ──────────────────────────────────────────────────────────── while (true) { try { - const tip = await withRetry(() => sourceSwitcher.getLatestLedger()); + const tip = await withRetry(() => loop.sourceSwitcher.getLatestLedger()); const target = tip - TIP_LAG; if (currentLedger >= target) { @@ -255,37 +348,64 @@ export async function startIndexer(): Promise { continue; } - if (INGEST_WORKERS > 1 && SAC_CONTRACT_IDS.length > 1) { + if (INGEST_WORKERS > 1 && loop.sacContractIds.length > 1) { // Parallel path: shard contracts across N workers for higher throughput (#83) const { totalInserted, highestLedger } = await pollParallel( - SAC_CONTRACT_IDS, + loop.sacContractIds, currentLedger, target, BATCH_SIZE, INGEST_WORKERS, + net, ); - totalIndexed += totalInserted; + loop.totalIndexed += totalInserted; currentLedger = highestLedger; } else { - currentLedger = await pollOnce(currentLedger, target); + currentLedger = await pollOnce(loop, currentLedger, target); } // Periodic data retention cleanup - pollCycleCount++; - if (pollCycleCount >= PRUNE_EVERY_CYCLES) { - pollCycleCount = 0; - await pruneOldTransfers().catch((e) => - console.error("[indexer] Prune failed:", e) + loop.pollCycleCount++; + if (loop.pollCycleCount >= PRUNE_EVERY_CYCLES) { + loop.pollCycleCount = 0; + await pruneOldTransfers(net).catch((e: unknown) => + console.error(`[indexer/${net}] Prune failed:`, e) ); } } catch (err) { - console.error("[indexer] Unhandled error in poll loop:", err); + console.error(`[indexer/${net}] Unhandled error in poll loop:`, err); // Back off before retrying to avoid hammering the RPC on persistent errors await sleep(POLL_INTERVAL_MS * 2); } } } +/** + * Start one loop per enabled network (see `NETWORKS`). + * + * Loops are fault-isolated from each other: a loop that throws out of its own + * retry handling is restarted on its own, because a mainnet RPC key expiring + * must not stop testnet indexing. `startIndexer` never resolves, so this + * returns immediately with the loops running in the background. + */ +export function startAllIndexers(networks: Network[] = enabledNetworks()): Network[] { + // Validate every network up front so a bad mainnet endpoint is reported at + // startup rather than after testnet has already begun writing. + validateNetworkConfig(networks); + + console.log(`[indexer] Starting loops for: ${networks.join(", ")}`); + + const run = (net: Network) => { + startIndexer(net).catch((err) => { + console.error(`[indexer/${net}] loop crashed, restarting in 10s:`, err); + setTimeout(() => run(net), 10_000); + }); + }; + + for (const net of networks) run(net); + return networks; +} + function sleep(ms: number): Promise { return new Promise((r) => setTimeout(r, ms)); } diff --git a/src/indexer/host-fn-log.ts b/src/indexer/host-fn-log.ts index a2e8f957..c4f18fa8 100644 --- a/src/indexer/host-fn-log.ts +++ b/src/indexer/host-fn-log.ts @@ -15,6 +15,7 @@ import * as StellarSdk from "@stellar/stellar-sdk"; import { Prisma } from "@prisma/client"; +import { resolveNetwork, type Network } from "../network"; import { prisma } from "../db"; import type { RawEvent } from "../rpc"; @@ -100,11 +101,19 @@ export function parseHostFnEvent(raw: RawEvent): HostFnRecord | null { * Idempotently persist a batch of host-fn log records. * Conflicts on `eventId` are silently ignored — safe to replay ledger ranges. */ -export async function upsertHostFnLogs(records: HostFnRecord[]): Promise { +export async function upsertHostFnLogs( + records: HostFnRecord[], + network?: Network, +): Promise { if (records.length === 0) return 0; + // Stamped explicitly: the column defaults to 'testnet', so omitting it + // compiles and files mainnet events under testnet. + const net = resolveNetwork(network); + const result = await prisma.hostFnLog.createMany({ data: records.map(r => ({ + network: net, contractId: r.contractId, functionName: r.functionName, args: r.args as Prisma.InputJsonValue, diff --git a/src/indexer/parallel.ts b/src/indexer/parallel.ts index e3a52a71..c134987b 100644 --- a/src/indexer/parallel.ts +++ b/src/indexer/parallel.ts @@ -13,6 +13,7 @@ */ import { fetchEventsSafe } from "../rpc"; +import { resolveNetwork, type Network } from "../network"; import { parseEvents } from "../decoder"; import { upsertTransfers, setLastIndexedLedger } from "../db"; import { emitTransfer } from "../events"; @@ -46,12 +47,15 @@ async function runPartitionWorker( fromLedger: number, toLedger: number, batchSize: number, + network: Network, ): Promise { const { events, highestLedger } = await fetchEventsSafe( fromLedger, toLedger, partition, batchSize, + undefined, + network, ); if (events.length === 0) { @@ -59,7 +63,7 @@ async function runPartitionWorker( } const records = parseEvents(events); - const inserted = await upsertTransfers(records); + const inserted = await upsertTransfers(records, network); if (inserted > 0) { records.forEach(emitTransfer); @@ -79,12 +83,14 @@ export async function pollParallel( toLedger: number, batchSize: number, workerCount: number = DEFAULT_WORKERS, + network?: Network, ): Promise<{ totalInserted: number; highestLedger: number }> { + const net = resolveNetwork(network); const partitions = partitionByContract(contractIds, Math.min(workerCount, contractIds.length || 1)); const results = await Promise.all( partitions.map(partition => - runPartitionWorker(partition, fromLedger, toLedger, batchSize), + runPartitionWorker(partition, fromLedger, toLedger, batchSize, net), ), ); @@ -94,11 +100,11 @@ export async function pollParallel( fromLedger, ); - await setLastIndexedLedger(highestLedger); + await setLastIndexedLedger(highestLedger, net); if (totalInserted > 0) { console.log( - `[parallel] ${partitions.length} workers processed ${totalInserted} new records (ledger ${highestLedger})`, + `[parallel/${net}] ${partitions.length} workers processed ${totalInserted} new records (ledger ${highestLedger})`, ); } diff --git a/src/indexer/sac-detect.ts b/src/indexer/sac-detect.ts index b422ac42..6687e6ae 100644 --- a/src/indexer/sac-detect.ts +++ b/src/indexer/sac-detect.ts @@ -20,6 +20,7 @@ import { xdr } from "@stellar/stellar-sdk"; import { getRpc } from "../rpc"; +import { resolveNetwork, type Network } from "../network"; // ─── Known SACs ─────────────────────────────────────────────────────────────── // The native XLM SAC on mainnet and testnet. These are fixed by the network and @@ -61,9 +62,12 @@ export function instanceValIsSac(val: xdr.ScVal): boolean { */ export type InstanceFetcher = (contractId: string) => Promise; -async function fetchInstanceVal(contractId: string): Promise { +async function fetchInstanceVal( + contractId: string, + network?: Network, +): Promise { try { - const entry = await getRpc().getContractData( + const entry = await getRpc(network).getContractData( contractId, xdr.ScVal.scvLedgerKeyContractInstance(), ); @@ -78,6 +82,9 @@ async function fetchInstanceVal(contractId: string): Promise { const cache = new Map(); +/** Cache key — namespaced by network so concurrent loops never share entries. */ +const cacheKey = (network: Network, contractId: string) => `${network}:${contractId}`; + /** * Detect whether `contractId` is a Stellar Asset Contract. Cached per contract. * @@ -86,16 +93,21 @@ const cache = new Map(); */ export async function detectSac( contractId: string, - fetchInstance: InstanceFetcher = fetchInstanceVal, + fetchInstance?: InstanceFetcher, + network?: Network, ): Promise { if (KNOWN_SAC_IDS.has(contractId)) return true; - const cached = cache.get(contractId); + const net = resolveNetwork(network); + const key = cacheKey(net, contractId); + + const cached = cache.get(key); if (cached !== undefined) return cached; - const val = await fetchInstance(contractId); + const fetcher = fetchInstance ?? ((id: string) => fetchInstanceVal(id, net)); + const val = await fetcher(contractId); const isSac = val !== null && instanceValIsSac(val); - cache.set(contractId, isSac); + cache.set(key, isSac); return isSac; } @@ -105,11 +117,12 @@ export async function detectSac( */ export async function detectSacBatch( contractIds: Iterable, - fetchInstance: InstanceFetcher = fetchInstanceVal, + fetchInstance?: InstanceFetcher, + network?: Network, ): Promise> { const unique = [...new Set(contractIds)]; const results = await Promise.all( - unique.map(async (id) => [id, await detectSac(id, fetchInstance)] as const), + unique.map(async (id) => [id, await detectSac(id, fetchInstance, network)] as const), ); return new Map(results); } @@ -121,12 +134,14 @@ export async function detectSacBatch( */ export async function tagSacTransfers( records: T[], - fetchInstance: InstanceFetcher = fetchInstanceVal, + fetchInstance?: InstanceFetcher, + network?: Network, ): Promise { if (records.length === 0) return records; const byContract = await detectSacBatch( records.map((r) => r.contractId), fetchInstance, + network, ); for (const record of records) { record.isSac = byContract.get(record.contractId) ?? false; diff --git a/src/indexer/sources/index.ts b/src/indexer/sources/index.ts index 75c00357..c48b96bb 100644 --- a/src/indexer/sources/index.ts +++ b/src/indexer/sources/index.ts @@ -1,11 +1,14 @@ import type { RawEvent } from "../../rpc"; import { createHorizonSource, type FetchLike, type HorizonSourceConfig } from "./horizon"; import { createRpcSource, type EventSource } from "./rpc"; +import type { Network } from "../../network"; export type SourceSwitcherConfig = { horizonUrl?: string; horizonEventsPath?: string; fetchImpl: FetchLike; + /** Which chain this switcher reads. Defaults to the configured network. */ + network?: Network; }; export interface SourceSwitcher { @@ -25,7 +28,7 @@ function isTruthySource(source: EventSource | null): source is EventSource { export function createSourceSwitcherWithConfig(config: SourceSwitcherConfig): SourceSwitcher { const sources = [ - createRpcSource(), + createRpcSource(config.network), config.horizonUrl ? createHorizonSource({ baseUrl: config.horizonUrl.replace(/\/$/, ""), diff --git a/src/indexer/sources/rpc.ts b/src/indexer/sources/rpc.ts index 3900f949..057ee527 100644 --- a/src/indexer/sources/rpc.ts +++ b/src/indexer/sources/rpc.ts @@ -1,4 +1,5 @@ import { fetchEventsSafe, getLatestLedger, type RawEvent } from "../../rpc"; +import type { Network } from "../../network"; export interface EventSource { name: string; @@ -12,20 +13,24 @@ export interface EventSource { ): Promise<{ events: RawEvent[]; highestLedger: number }>; } -export function createRpcSource(): EventSource { +/** + * @param network Which chain this source reads. Omitted means the configured + * network, which is what every single-network deployment gets. + */ +export function createRpcSource(network?: Network): EventSource { return { name: "rpc", async isHealthy() { try { - await getLatestLedger(); + await getLatestLedger(network); return true; } catch { return false; } }, - getLatestLedger, + getLatestLedger: () => getLatestLedger(network), fetchEvents(startLedger, endLedger, contractIds, limit) { - return fetchEventsSafe(startLedger, endLedger, contractIds, limit); + return fetchEventsSafe(startLedger, endLedger, contractIds, limit, undefined, network); }, }; } \ No newline at end of file diff --git a/src/ingester/nft.ts b/src/ingester/nft.ts index 9169f6c2..ac6793f9 100644 --- a/src/ingester/nft.ts +++ b/src/ingester/nft.ts @@ -9,6 +9,7 @@ import { } from "@stellar/stellar-sdk"; import type { RawEvent } from "../rpc"; import { getRpc } from "../rpc"; +import { resolveNetwork, type Network } from "../network"; // ─── Types ──────────────────────────────────────────────────────────────────── @@ -117,11 +118,15 @@ export function parseNftEvents( */ export async function fetchNftMetadata( contractId: string, - tokenIdScVal: xdr.ScVal + tokenIdScVal: xdr.ScVal, + network?: Network ): Promise { - const network = (process.env.STELLAR_NETWORK ?? "testnet").toLowerCase(); - const networkPassphrase = - network === "mainnet" ? Networks.PUBLIC : Networks.TESTNET; + // Read from the caller's network rather than the process-wide env var: with + // one loop per network, a module-level lookup would sign every simulation + // with whichever passphrase happened to be configured, so a mainnet metadata + // read would be built against the testnet passphrase and simulate wrongly. + const net = resolveNetwork(network); + const networkPassphrase = net === "mainnet" ? Networks.PUBLIC : Networks.TESTNET; // Any valid address works as a simulation source — it doesn't need funds. const dummy = new Account( @@ -129,7 +134,7 @@ export async function fetchNftMetadata( "0" ); const contract = new Contract(contractId); - const rpc = getRpc(); + const rpc = getRpc(net); const result: NftMetadataPayload = {}; // Try token_uri(token_id) diff --git a/src/network.ts b/src/network.ts index 1a15d637..048994b3 100644 --- a/src/network.ts +++ b/src/network.ts @@ -53,3 +53,28 @@ export function currentNetwork(): Network { export function resolveNetwork(network?: Network): Network { return network ?? currentNetwork(); } + +/** + * The networks this process should index, from `NETWORKS` (comma-separated). + * + * Defaults to just {@link currentNetwork}, so a deployment that sets only + * `STELLAR_NETWORK` keeps running exactly one loop — the pre-#161 behaviour. + * `NETWORKS=testnet,mainnet` opts into indexing both in one process. + * + * Unrecognised entries are dropped rather than throwing: a typo should not + * take the whole indexer down, and the caller logs what it actually started. + * An empty or entirely invalid list falls back to the configured network for + * the same reason. + */ +export function enabledNetworks(): Network[] { + const raw = process.env.NETWORKS ?? ""; + const parsed = raw + .split(",") + .map((entry) => parseNetwork(entry)) + .filter((entry): entry is Network => entry !== null); + + // De-duplicate: NETWORKS=testnet,testnet must not start two loops writing + // the same rows and fighting over the same cursor. + const unique = [...new Set(parsed)]; + return unique.length > 0 ? unique : [currentNetwork()]; +} diff --git a/src/rpc.ts b/src/rpc.ts index 6f00b738..16ee4a39 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -1,60 +1,79 @@ import { rpc as RPC, xdr } from "@stellar/stellar-sdk"; +import { resolveNetwork, currentNetwork, type Network } from "./network"; // ─── Network config ─────────────────────────────────────────────────────────── const TESTNET_RPC_URL = "https://soroban-testnet.stellar.org"; /** - * Resolve the Soroban RPC endpoint from environment variables. + * Resolve the Soroban RPC endpoint for one network. * - * Resolution order: - * 1. SOROBAN_RPC_URL (explicit — takes precedence) - * 2. STELLAR_RPC_URL (backward-compat alias) - * 3. STELLAR_NETWORK=testnet → default testnet URL - * 4. STELLAR_NETWORK=mainnet → requires explicit SOROBAN_RPC_URL; no free - * public mainnet RPC exists, so we fail fast - * 5. Nothing set → throws with a clear configuration guide + * Resolution order, per network: + * 1. SOROBAN_RPC_URL_TESTNET / SOROBAN_RPC_URL_MAINNET (explicit, per network) + * 2. SOROBAN_RPC_URL / STELLAR_RPC_URL — but **only for the network this + * process is configured as** (STELLAR_NETWORK). See below. + * 3. testnet → default public testnet endpoint + * 4. mainnet → throws; there is no free public mainnet Soroban RPC + * + * Step 2 is deliberately narrow. The unsuffixed variables predate multi-network + * support, so a deployment that sets `SOROBAN_RPC_URL` means "the endpoint for + * the network I run". Honouring it for *both* networks would silently point a + * mainnet indexer at a testnet endpoint — it would connect, index happily, and + * write testnet ledger data tagged `network='mainnet'`. Scoping the legacy + * variable to the configured network keeps every single-network deployment + * behaving exactly as before while making that mix-up impossible. */ -function resolveRpcUrl(): string { - const explicit = process.env.SOROBAN_RPC_URL ?? process.env.STELLAR_RPC_URL; - if (explicit) return explicit; - - const network = (process.env.STELLAR_NETWORK ?? "").toLowerCase(); +function resolveRpcUrl(network: Network): string { + const suffix = network.toUpperCase(); + const perNetwork = + process.env[`SOROBAN_RPC_URL_${suffix}`] || process.env[`STELLAR_RPC_URL_${suffix}`]; + if (perNetwork) return perNetwork; + + if (network === currentNetwork()) { + const legacy = process.env.SOROBAN_RPC_URL || process.env.STELLAR_RPC_URL; + if (legacy) return legacy; + } if (network === "testnet") return TESTNET_RPC_URL; - if (network === "mainnet") { - throw new Error( - "[wraith] SOROBAN_RPC_URL is required when STELLAR_NETWORK=mainnet. " + - "There is no free public Soroban RPC for mainnet — set SOROBAN_RPC_URL " + - "to your provider's endpoint (e.g. Validation Cloud, Ankr, self-hosted)." - ); - } - throw new Error( - "[wraith] RPC endpoint not configured. " + - "Set SOROBAN_RPC_URL to your Soroban RPC endpoint, or set STELLAR_NETWORK=testnet " + - "to use the default public testnet endpoint automatically." + `[wraith] SOROBAN_RPC_URL_MAINNET is required to index mainnet. ` + + "There is no free public Soroban RPC for mainnet — set it to your " + + "provider's endpoint (e.g. Validation Cloud, Ankr, self-hosted). " + + "Single-network deployments may still use SOROBAN_RPC_URL with " + + "STELLAR_NETWORK=mainnet." ); } /** - * Validate the network configuration at startup. - * Call this before opening DB connections so configuration errors surface - * immediately instead of on the first RPC call. + * Validate RPC configuration at startup for every network given (defaults to + * the configured one). Call before opening DB connections so a misconfigured + * endpoint surfaces immediately rather than on the first poll. */ -export function validateNetworkConfig(): void { - resolveRpcUrl(); // throws with a human-readable message if misconfigured +export function validateNetworkConfig(networks: Network[] = [currentNetwork()]): void { + for (const network of networks) { + resolveRpcUrl(network); // throws with a human-readable message + } } -// ─── RPC client singleton ───────────────────────────────────────────────────── -let _rpc: RPC.Server | null = null; - -export function getRpc(): RPC.Server { - if (!_rpc) { - const url = resolveRpcUrl(); - _rpc = new RPC.Server(url, { allowHttp: url.startsWith("http://") }); +// ─── RPC clients, one per network ───────────────────────────────────────────── +// Cached per network: repeated calls reuse a connection, and two networks can +// never share one — which was impossible with the previous single singleton. +const clients = new Map(); + +export function getRpc(network?: Network): RPC.Server { + const net = resolveNetwork(network); + let client = clients.get(net); + if (!client) { + const url = resolveRpcUrl(net); + client = new RPC.Server(url, { allowHttp: url.startsWith("http://") }); + clients.set(net, client); } - return _rpc; + return client; +} + +/** Test-only: drops cached clients so a test can rebind env or mocks. */ +export function _resetRpcClients(): void { + clients.clear(); } // ─── Types ──────────────────────────────────────────────────────────────────── @@ -80,13 +99,15 @@ export interface RawEvent { * @param startLedger First ledger to include (inclusive). * @param contractIds Filter to specific contract IDs. Pass [] to skip filter. * @param limit Max events per call (RPC hard-caps at 10 000). + * @param network Which chain to read. Defaults to the configured network. */ export async function fetchEvents( startLedger: number, contractIds: string[], - limit: number = 10_000 + limit: number = 10_000, + network?: Network ): Promise<{ events: RawEvent[]; latestLedger: number }> { - const rpc = getRpc(); + const rpc = getRpc(network); // Build the request using the correct Server.GetEventsRequest type. // Api.EventFilter allows: type, contractIds (string[]), topics (string[][]). @@ -123,8 +144,8 @@ export async function fetchEvents( } // ─── Network tip helper ─────────────────────────────────────────────────────── -export async function getLatestLedger(): Promise { - const rpc = getRpc(); +export async function getLatestLedger(network?: Network): Promise { + const rpc = getRpc(network); const resp = await rpc.getLatestLedger(); return resp.sequence; } @@ -170,12 +191,13 @@ export async function fetchEventsSafe( endLedger: number, contractIds: string[], limit: number = 10_000, - _fetchFn: FetchFn = fetchEvents + _fetchFn: FetchFn = fetchEvents, + network?: Network ): Promise<{ events: RawEvent[]; highestLedger: number }> { // If the range is a single ledger and it fails, skip it. if (startLedger >= endLedger) { try { - const { events, latestLedger } = await _fetchFn(startLedger, contractIds, limit); + const { events, latestLedger } = await _fetchFn(startLedger, contractIds, limit, network); return { events, highestLedger: Math.max(startLedger, latestLedger) }; } catch (err) { const msg = (err as Error).message ?? ""; @@ -188,7 +210,7 @@ export async function fetchEventsSafe( } try { - const { events, latestLedger } = await _fetchFn(startLedger, contractIds, limit); + const { events, latestLedger } = await _fetchFn(startLedger, contractIds, limit, network); return { events, highestLedger: latestLedger }; } catch (err) { const msg = (err as Error).message ?? ""; @@ -198,8 +220,8 @@ export async function fetchEventsSafe( console.warn(`[rpc] XDR error in ledgers ${startLedger}–${endLedger}, bisecting…`); const mid = Math.floor((startLedger + endLedger) / 2); - const lower = await fetchEventsSafe(startLedger, mid, contractIds, limit, _fetchFn); - const upper = await fetchEventsSafe(mid + 1, endLedger, contractIds, limit, _fetchFn); + const lower = await fetchEventsSafe(startLedger, mid, contractIds, limit, _fetchFn, network); + const upper = await fetchEventsSafe(mid + 1, endLedger, contractIds, limit, _fetchFn, network); return { events: [...lower.events, ...upper.events], From 812786bb52e579fc9212d5286061b8d344bf22a5 Mon Sep 17 00:00:00 2001 From: Salmatcre8 Date: Sun, 30 Aug 2026 10:23:17 +0100 Subject: [PATCH 2/2] fix(status): report per-network progress when degraded too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review catch: `networks` was attached only to the healthy branch of /status. The degraded branch computed it and threw it away. That is backwards. With two loops, "RPC is down" is normally true of one chain and not the other — mainnet runs on a paid provider endpoint whose key can expire while the public testnet endpoint is fine. Degraded is precisely when you need to know *which* chain is stuck, and the top-level fields go null in that branch, so it was the one case the whole field exists for. Also calls runningNetworks() once instead of twice. Both branches now assert `networks` in staleReads.test.ts. Mutation- checked: removing it from the degraded branch fails that test alone. Claude-Session: https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB --- src/__tests__/staleReads.test.ts | 16 ++++++++++++++++ src/api.ts | 7 ++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/src/__tests__/staleReads.test.ts b/src/__tests__/staleReads.test.ts index ef2d3224..fbbec829 100644 --- a/src/__tests__/staleReads.test.ts +++ b/src/__tests__/staleReads.test.ts @@ -106,6 +106,13 @@ describe("Graceful stale reads during RPC outage (#164)", () => { expect(res.body.lastIndexedLedger).toBe(1000); expect(res.body.latestLedger).toBe(1050); expect(res.body.lagLedgers).toBe(50); + // #161: per-network progress alongside the aggregate view. + expect(res.body.networks).toBeDefined(); + expect(res.body.networks.testnet).toMatchObject({ + lastIndexedLedger: 1000, + latestLedger: 1050, + lagLedgers: 50, + }); }); it("returns status 'degraded' when RPC is down but DB is healthy", async () => { @@ -118,6 +125,15 @@ describe("Graceful stale reads during RPC outage (#164)", () => { expect(res.body.status).toBe("degraded"); expect(res.body.stale).toBe(true); expect(res.body.as_of_ledger).toBe(1000); + // Reported when degraded too. With two loops "RPC is down" is usually + // true of one chain only, and the top-level nulls cannot say which — + // so omitting this here would blind exactly the case it exists for. + expect(res.body.networks).toBeDefined(); + expect(res.body.networks.testnet).toMatchObject({ + lastIndexedLedger: 1000, + latestLedger: null, + lagLedgers: null, + }); expect(res.body.latestLedger).toBeNull(); expect(res.body.lagLedgers).toBeNull(); expect(res.headers["x-data-stale"]).toBe("true"); diff --git a/src/api.ts b/src/api.ts index 042e3695..601ce228 100644 --- a/src/api.ts +++ b/src/api.ts @@ -309,7 +309,8 @@ export function createApp(): express.Application { // were so existing consumers are unaffected; this reports each loop // separately, which is the only way to see one chain falling behind // while the other is healthy. - const networks: Network[] = runningNetworks().length > 0 ? runningNetworks() : enabledNetworks(); + const active = runningNetworks(); + const networks: Network[] = active.length > 0 ? active : enabledNetworks(); const loopStats = getAllIndexerStats(); const perNetwork = await Promise.all( networks.map(async (net) => { @@ -361,6 +362,10 @@ export function createApp(): express.Application { latestLedger: null, lagLedgers: null, ...stats, + // Also reported when degraded — with two loops, "RPC is down" is + // usually true of one chain only, and the top-level nulls above + // cannot say which. + networks: byNetwork, }); } } catch (err) {