diff --git a/README.md b/README.md index 0377f41..553c4e3 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Streams a live view of the Moonlight network over WebSocket. Cold-start self-syncs from `council-platform`'s public endpoints and Soroban RPC; no persistence. -Six event kinds drive the dashboard (see +Seven event kinds drive the dashboard (see [design sketch](../pm-theahaco/network-dashboard-design-sketch.md)): | Kind | Chain source | @@ -21,27 +21,48 @@ Six event kinds drive the dashboard (see | `asset_registered` | New `assetContractId` appearing in `council-platform/public/channels` | | `channel_deposit` | SAC `transfer` event TO a known channel address | | `channel_settlement` | SAC `transfer` event FROM a known channel address | +| `channel_bundle` | SAC `fee` event whose payer is a registered PP | ## Endpoints | Path | Description | | ------------------------ | -------------------------------------------------------------- | | `GET /api/v1/health` | Liveness — `{status,service,version}` | -| `GET /api/v1/network/ws` | Public WebSocket. Subprotocol `moonlight.network.v1`. No auth. | +| `GET /api/v1/network/ws` | Public WebSocket. Subprotocol `moonlight.network.v2`. No auth. | ### WebSocket frame protocol Server → client, JSON-encoded: ```jsonc -// Sent once on open. +// Sent once on open. (v2 additions over v1: throughputPerMin + latencyMs +// counters, sparklines, assetBreakdown, councilRolling.) { "type": "snapshot", "counters": { "councils": 4, "activePPs": 7, "eventsLast24h": 31, - "assetsRegistered": 2 + "assetsRegistered": 2, + "throughputPerMin": 3, + "latencyMs": 4200 + }, + "sparklines": { + "throughput": [/* 60 per-minute counts, oldest first */], + "latency": [/* 60 per-minute avg ms (null where no samples) */], + "volume": [/* 60 per-minute deposit+settlement totals (whole units) */] + }, + "assetBreakdown": [ + { "assetContractId": "CDMLF…", "assetCode": "XLM", "amountStroops": "12000000", "percent": 100 } + ], + "councilRolling": { + "CBPHGAJ4F7…": { + "bundlesLastHour": 2, + "eventsLastHour": 5, + "ratePerMin": 0.1, + "depositVolumeStroops": "200500000", + "settlementVolumeStroops": "100000000" + } }, "topology": [ { @@ -71,8 +92,8 @@ Server → client, JSON-encoded: } ``` -No client → server frames in v1. Clients reconnect rather than ping; on -reconnect they receive a fresh snapshot. +No client → server frames. Clients reconnect rather than ping; on reconnect they +receive a fresh snapshot. ## Architecture @@ -81,15 +102,28 @@ council-platform /public/* ─┐ ├─► in-memory state ─► WS clients Soroban /getEvents (poll) ──┘ ▲ │ - hourly re-sync + minute window sweep + minute topology re-sync + minute window sweep ``` - **Cold start**: fetch `council-platform/api/v1/public/councils` (one call carries councils + channels + providers + jurisdictions), walk trailing 24h on - every watched contract via `rpc.getEvents`, seed the rolling counter window + - activity-feed ring buffer. -- **Forward poll**: 5s cursor-based poll on the same contractId set. -- **Hourly re-sync**: refresh topology + re-anchor the rolling counter window. + the subscription set via `rpc.getEvents`, seed the rolling counter window and + the activity-feed ring buffer. +- **Subscriptions**: Channel Auth (council) contracts are watched by contractId. + SAC contracts are watched by TOPIC only — `transfer` patterns pinned to known + channel addresses (deposit/settlement) and `fee` patterns pinned to registered + PP keys (bundles). The mainnet XLM SAC emits hundreds of events per ledger; an + unfiltered contractId subscription drowns in them. +- **Forward poll**: 5s poll over the subscription set, paged via the RPC event + cursor so consumption is gap-free even mid-ledger; the shared poll position + only advances across fully-consumed pages (re-read overlap is deduped by event + id). +- **New-council discovery**: network-wide `contract_initialized` poll feeds the + adoption pipeline (topology refresh + historical back-fill per adopted + council). +- **Minute topology re-sync**: refresh topology from council-platform — the + backstop for DB-only registrations (channels, jurisdictions, labels), which + emit no chain event. - **Minute sweep**: drop window entries older than 24h. ## Running locally diff --git a/deno.json b/deno.json index 323bf97..ad2bb5d 100644 --- a/deno.json +++ b/deno.json @@ -1,6 +1,6 @@ { "name": "@moonlight-protocol/network-dashboard-platform", - "version": "0.1.15", + "version": "0.1.16", "license": "MIT", "exports": "./src/main.ts", "tasks": { diff --git a/deno.lock b/deno.lock index d386e55..9cf56fa 100644 --- a/deno.lock +++ b/deno.lock @@ -10,6 +10,7 @@ "jsr:@std/encoding@1": "1.0.11", "jsr:@std/encoding@^1.0.11": "1.0.11", "jsr:@std/http@1": "1.1.2", + "jsr:@std/internal@^1.0.12": "1.0.14", "jsr:@std/internal@^1.0.14": "1.0.14", "jsr:@std/media-types@1": "1.1.0", "jsr:@std/path@1": "1.1.6", @@ -42,7 +43,10 @@ ] }, "@std/assert@1.0.19": { - "integrity": "eaada96ee120cb980bc47e040f82814d786fe8162ecc53c91d8df60b8755991e" + "integrity": "eaada96ee120cb980bc47e040f82814d786fe8162ecc53c91d8df60b8755991e", + "dependencies": [ + "jsr:@std/internal@^1.0.12" + ] }, "@std/bytes@1.0.6": { "integrity": "f6ac6adbd8ccd99314045f5703e23af0a68d7f7e58364b47d2c7f408aeb5820a" @@ -71,7 +75,7 @@ "@std/path@1.1.6": { "integrity": "c68485c2a4dfbb5ae3cc74fae4e8c4e5d874cf8a8ed12927917235c758b46cbe", "dependencies": [ - "jsr:@std/internal" + "jsr:@std/internal@^1.0.14" ] } }, diff --git a/src/core/state/store.ts b/src/core/state/store.ts index a5c0697..23314b8 100644 --- a/src/core/state/store.ts +++ b/src/core/state/store.ts @@ -82,6 +82,15 @@ export class NetworkStateStore { private providerToCouncil = new Map(); /** Ring buffer of most-recent events for the activity feed. */ private recent: NetworkEvent[] = []; + /** + * Ids of every event currently tracked (ring buffer + 24h window). + * The watcher's cursor-based drains re-read a small overlap by design + * (the shared poll position advances to the LOWEST drain point across + * calls), so dedup must outlive the 20-entry ring — otherwise a + * re-read event past the ring would double-count metrics and re-emit + * to WS clients. Pruned by the minute sweep alongside `metrics`. + */ + private seenIds = new Set(); /** * Sliding 24h metric records (one per event observed within the window). * Newest at the end; the cold-start scan seeds in chronological order. @@ -168,6 +177,10 @@ export class NetworkStateStore { return this.providerToCouncil.get(publicKey); } + getProviderPublicKeys(): string[] { + return Array.from(this.providerToCouncil.keys()); + } + /** * Surgical, idempotent insert into the providerToCouncil map. Called by * the watcher when a `provider_added` chain event is observed, so a PP @@ -210,9 +223,10 @@ export class NetworkStateStore { * close time; pass null when the watcher couldn't determine it. */ recordEvent(event: NetworkEvent, latencyMs: number | null = null): boolean { - if (this.recent.some((e) => e.id === event.id)) { + if (this.seenIds.has(event.id)) { return false; } + this.seenIds.add(event.id); this.recent.unshift(event); if (this.recent.length > RING_BUFFER_SIZE) { this.recent.length = RING_BUFFER_SIZE; @@ -234,12 +248,24 @@ export class NetworkStateStore { return true; } - /** Drop metric records older than 24h. Called by the minute-sweep. */ + /** + * Drop metric records older than 24h. Called by the minute-sweep. + * `seenIds` is rebuilt from the survivors (+ the ring buffer) so the + * dedup set stays bounded; a pruned id cannot legitimately reappear — + * the watcher's poll position only moves forward. + */ sweepWindow(now: number = Date.now()): number { const cutoff = now - ROLLING_WINDOW_MS; const before = this.metrics.length; this.metrics = this.metrics.filter((m) => m.occurredAt >= cutoff); - return before - this.metrics.length; + const purged = before - this.metrics.length; + if (purged > 0) { + this.seenIds = new Set([ + ...this.metrics.map((m) => m.id), + ...this.recent.map((e) => e.id), + ]); + } + return purged; } /** @@ -259,6 +285,7 @@ export class NetworkStateStore { if (existing.has(e.id)) continue; const ts = Date.parse(e.occurredAt); if (Number.isNaN(ts)) continue; + this.seenIds.add(e.id); const { assetContractId, amountStroops } = readPayload(e); this.metrics.push({ id: e.id, @@ -275,6 +302,7 @@ export class NetworkStateStore { /** Seed the ring buffer at cold start (newest entries first). */ seedRecent(events: NetworkEvent[]): void { this.recent = events.slice(0, RING_BUFFER_SIZE); + for (const e of this.recent) this.seenIds.add(e.id); } recentEvents(): NetworkEvent[] { @@ -467,6 +495,7 @@ export class NetworkStateStore { this.providerToCouncil.clear(); this.recent = []; this.metrics = []; + this.seenIds.clear(); } } diff --git a/src/core/state/store_test.ts b/src/core/state/store_test.ts index c856319..21c4413 100644 --- a/src/core/state/store_test.ts +++ b/src/core/state/store_test.ts @@ -305,3 +305,36 @@ Deno.test("councilRollingMetrics returns zero rows for known but quiet councils" assertEquals(rolling["CB"].bundlesLastHour, 0); assertEquals(rolling["CB"].eventsLastHour, 0); }); + +Deno.test("recordEvent dedups by id past the ring buffer (cursor re-reads)", () => { + const s = makeStore(); + // Overflow the 20-entry ring so the first event is long gone from it. + for (let i = 0; i < RING_BUFFER_CAPACITY + 5; i++) { + assertEquals(s.recordEvent(event(`e${i}`)), true); + } + // The watcher's shared-cursor advance re-reads overlap by design; a + // re-observed event must stay a no-op even when it left the ring. + assertEquals(s.recordEvent(event("e0")), false); + assertEquals(s.countEventsLast24h(), RING_BUFFER_CAPACITY + 5); +}); + +Deno.test("sweepWindow prunes the dedup set with the metrics", () => { + const s = makeStore(); + s.recordEvent(event("old", "provider_added", ROLLING_WINDOW_DURATION_MS * 2)); + s.recordEvent(event("fresh")); + // "old" is beyond 24h: swept from metrics. It is still in the ring + // buffer, so it must survive in the dedup set via the ring rebuild. + assertEquals(s.sweepWindow(), 1); + assertEquals(s.recordEvent(event("old")), false); + assertEquals(s.recordEvent(event("fresh")), false); + assertEquals(s.countEventsLast24h(), 1); +}); + +Deno.test("getProviderPublicKeys reflects topology + chain-event writes", () => { + const s = makeStore(); + s.replaceTopology([council("CA", [], ["GA", "GB"])]); + s.registerProvider("GC", "CA"); + assertEquals(s.getProviderPublicKeys().sort(), ["GA", "GB", "GC"]); + s.unregisterProvider("GB"); + assertEquals(s.getProviderPublicKeys().sort(), ["GA", "GC"]); +}); diff --git a/src/core/sync/contract-init-listener.ts b/src/core/sync/contract-init-listener.ts index 7d0479b..f897c46 100644 --- a/src/core/sync/contract-init-listener.ts +++ b/src/core/sync/contract-init-listener.ts @@ -29,14 +29,25 @@ import type { NetworkEventBus } from "@/core/events/bus.ts"; /** * `contract_initialized` fires at deploy time, which precedes the - * council-platform `PUT /council/metadata` call by some script-controlled - * window. We keep an unknown contract in `pendingAdoption` and retry - * topology refresh each poll tick for up to PENDING_TTL_MS so a council - * registered AFTER its on-chain deploy still gets adopted. Past the TTL we - * cache as notMoonlight so a chatty unrelated contract can't drag us into - * repeated topology refreshes forever. + * council-platform `PUT /council/metadata` call by some window — script + * flows register within seconds, but a HUMAN-driven flow (console/CLI, + * the mainnet demo path) can take many minutes. We keep an unknown + * contract in `pendingAdoption` for up to PENDING_TTL_MS so a council + * registered well after its on-chain deploy still gets adopted (adoption + * is what triggers the historical back-fill; the periodic topology + * re-sync would surface the council's topology either way, but not its + * pre-adoption events). Past the TTL we cache as notMoonlight so an + * unrelated contract doesn't stay pending forever. The previous 120s TTL + * permanently blacklisted the mainnet demo councils — their platform + * registration landed minutes after deploy. + * + * Refreshes for pending unknowns are throttled to REFRESH_MIN_INTERVAL_MS + * (except when a NEW unknown just arrived, which refreshes immediately) so + * a chatty unrelated contract costs at most a couple of council-platform + * fetches per minute during its pending hour. */ -const PENDING_TTL_MS = 120_000; +const PENDING_TTL_MS = 60 * 60 * 1000; +const REFRESH_MIN_INTERVAL_MS = 30_000; interface PendingEntry { firstSeenMs: number; @@ -46,6 +57,9 @@ interface PendingEntry { const notMoonlight = new Set(); const pendingAdoption = new Map(); +/** Set when an unknown was JUST registered — bypasses the refresh throttle. */ +let hasFreshPending = false; +let lastPendingRefreshMs = 0; /** * Soroban event filter pattern matching `contract_initialized` events with @@ -87,6 +101,7 @@ export function evaluateUnknownContract( firstSeenMs: Date.now(), observedAtLedger, }); + hasFreshPending = true; log.event("registered unknown contractId for topology adoption"); } @@ -117,7 +132,17 @@ export async function drainPendingAdoptions( log.info("drainPendingAdoptions"); log.debug("pendingCount", pendingAdoption.size); - await refreshTopology(`pending=${pendingAdoption.size}`, deps); + // Refresh immediately for a brand-new unknown (fast adoption of real + // councils); otherwise throttle — the periodic topology re-sync also + // updates `networkState`, so waiting pendings still get their + // `hasCouncil` check against fresh state below. + const refreshNow = hasFreshPending || + Date.now() - lastPendingRefreshMs >= REFRESH_MIN_INTERVAL_MS; + if (refreshNow) { + hasFreshPending = false; + lastPendingRefreshMs = Date.now(); + await refreshTopology(`pending=${pendingAdoption.size}`, deps); + } const now = Date.now(); /** Earliest observed ledger across freshly-adopted contracts in this drain pass. */ @@ -162,4 +187,6 @@ export async function drainPendingAdoptions( export function __resetForTests(): void { notMoonlight.clear(); pendingAdoption.clear(); + hasFreshPending = false; + lastPendingRefreshMs = 0; } diff --git a/src/core/sync/scheduler.ts b/src/core/sync/scheduler.ts index 2b653aa..35056ad 100644 --- a/src/core/sync/scheduler.ts +++ b/src/core/sync/scheduler.ts @@ -1,28 +1,35 @@ import type { Logger } from "@/utils/logger/index.ts"; +import type { NetworkEventBus } from "@/core/events/bus.ts"; import { networkState } from "@/core/state/store.ts"; +import { refreshTopology } from "./topology-refresh.ts"; /** - * Background scheduler — rolling-window sweep only. + * Background scheduler — rolling-window sweep + periodic topology re-sync. * - * The previous hourly topology re-sync is gone. Topology updates happen on - * the hot path: boot does the initial fetch (see `main.ts:bootstrap`), and - * new councils are discovered via the Soroban `contract_initialized` - * watcher (`contract-init-listener.ts`), which triggers a fresh - * `refreshTopology` the moment a new Channel Auth deploy is observed. - * Mirrors `provider-platform`'s event-watcher pattern: sync at boot, set - * listeners, react to events. + * The periodic re-sync exists because parts of the topology change with NO + * chain event to react to: channel and jurisdiction registration are + * council-platform DB operations (POST /council/channels), and provider + * labels are metadata. The event-driven paths (boot fetch, the + * `contract_initialized` watcher in `contract-init-listener.ts`, the + * `provider_added` piggyback) cover the chain-visible transitions, but a + * council whose channels were registered after its last refresh stayed + * frozen — on mainnet that left a council name-only (no channels, no + * providers, no jurisdictions) for hours. One council-platform fetch per + * minute is the completeness backstop; `refreshTopology` is single-flight, + * so overlap with the event-driven refreshes coalesces. * - * `sweepWindow` is unrelated to topology — it drops stale entries from the - * 24-hour rolling counter window so memory stays bounded. Keeping it on - * a 60-second cadence. + * `sweepWindow` drops stale entries from the 24-hour rolling counter + * window so memory stays bounded. Keeping it on a 60-second cadence. */ const MINUTE_SWEEP_MS = 60 * 1000; +const TOPOLOGY_RESYNC_MS = 60 * 1000; let minuteTimer: number | null = null; +let resyncTimer: number | null = null; let running = false; export function startScheduler( - deps: { log: Logger }, + deps: { log: Logger; bus: NetworkEventBus }, ): void { if (running) return; running = true; @@ -36,9 +43,20 @@ export function startScheduler( } } + function topologyResync(): void { + refreshTopology("periodic-resync", deps).catch((err) => { + log.error(err, "periodic topology re-sync failed"); + }); + } + minuteTimer = setInterval(minuteSweep, MINUTE_SWEEP_MS) as unknown as number; + resyncTimer = setInterval( + topologyResync, + TOPOLOGY_RESYNC_MS, + ) as unknown as number; log.debug("minuteSweepMs", MINUTE_SWEEP_MS); + log.debug("topologyResyncMs", TOPOLOGY_RESYNC_MS); log.event("scheduler started"); } @@ -48,5 +66,9 @@ export function stopScheduler(deps: { log: Logger }): void { clearInterval(minuteTimer); minuteTimer = null; } + if (resyncTimer !== null) { + clearInterval(resyncTimer); + resyncTimer = null; + } deps.log.scope("scheduler").event("scheduler stopped"); } diff --git a/src/core/sync/soroban-watcher-queries_test.ts b/src/core/sync/soroban-watcher-queries_test.ts new file mode 100644 index 0000000..609df8d --- /dev/null +++ b/src/core/sync/soroban-watcher-queries_test.ts @@ -0,0 +1,293 @@ +import { assertEquals } from "@std/assert"; +import { Address, xdr } from "stellar-sdk"; +import type { Server } from "stellar-sdk/rpc"; +import { + __resetWatcherStateForTests, + __setServerForTests, + buildWatchQueryCalls, + drainEvents, + ledgerOfCursor, + minCursor, +} from "./soroban-watcher.ts"; +import { networkState } from "@/core/state/store.ts"; +import { newNoop } from "@/utils/logger/index.ts"; +import type { CouncilTopologyEntry } from "@/core/events/types.ts"; + +const COUNCIL = "CCVYCJF7ONC4DHYKI34XINUVBBISAMFOD7N4SRRZS2JE2IFBWNUDVMRI"; +const CHANNEL = "CCLTT2ZJMMSKMUFTMDGZRRT76LFXK6INYM35VFKVZF5ZB4S7LQVEDZZ7"; +const SAC = "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA"; +const PP = "GAR2WBIXBOXP3GA7XNVOSEIB3QL2OZJRT2QSX24UJFTDVI26M23MEP25"; + +const sym = (s: string) => xdr.ScVal.scvSymbol(s).toXDR("base64"); +const addr = (a: string) => new Address(a).toScVal().toXDR("base64"); + +/** Real-format RPC cursor for a ledger (`-`). */ +const cursorAt = (ledger: number, eventIndex = 0) => + `${(BigInt(ledger) << 32n).toString()}-${eventIndex}`; + +function topology(): CouncilTopologyEntry[] { + return [{ + id: COUNCIL, + name: "Council", + providers: [{ publicKey: PP, label: null }], + channels: [{ + contractId: CHANNEL, + assetCode: "XLM", + assetContractId: SAC, + }], + jurisdictions: ["AU"], + }]; +} + +Deno.test("buildWatchQueryCalls: councils by contractId, SAC by topic pattern, one filter per call", () => { + networkState.__resetForTests(); + networkState.replaceTopology(topology()); + + const calls = buildWatchQueryCalls(); + assertEquals(calls.length, 2); + + assertEquals(calls[0].filter, { type: "contract", contractIds: [COUNCIL] }); + // The SAC is never watched raw — only via channel/PP topic patterns. + assertEquals(calls[1].filter.contractIds, [SAC]); + assertEquals(calls[1].filter.topics, [ + [sym("transfer"), "*", addr(CHANNEL), "*"], // deposit + [sym("transfer"), addr(CHANNEL), "*", "*"], // settlement + [sym("fee"), addr(PP)], // bundle fee + ]); + + networkState.__resetForTests(); +}); + +Deno.test("buildWatchQueryCalls: no councils → no calls; no channels/PPs → no SAC filter", () => { + networkState.__resetForTests(); + assertEquals(buildWatchQueryCalls(), []); + + networkState.replaceTopology([{ + id: COUNCIL, + name: "Council", + providers: [], + channels: [], + jurisdictions: [], + }]); + const calls = buildWatchQueryCalls(); + assertEquals(calls.length, 1); + assertEquals(calls[0].filter, { type: "contract", contractIds: [COUNCIL] }); + + networkState.__resetForTests(); +}); + +// ── drainEvents ─────────────────────────────────────────────────────── + +type StubPage = + | { + events: number; + cursor: string; + latestLedger: number; + startId?: number; + } + | { throws: string | { message: string } }; + +function stubServer(pages: StubPage[]) { + const requests: Array> = []; + let call = 0; + const server = { + // deno-lint-ignore require-await + getEvents: async (request: Record) => { + requests.push(request); + const page = pages[call++]; + if (!page) throw new Error("stub exhausted"); + if ("throws" in page) throw page.throws; + const base = page.startId ?? 0; + return { + events: Array.from({ length: page.events }, (_, i) => ({ + id: `evt-${base + i}`, + ledger: 100 + base + i, + ledgerClosedAt: "2026-07-30T12:00:00Z", + topic: [], + value: xdr.ScVal.scvVoid(), + txHash: `tx-${base + i}`, + contractId: undefined, + })), + cursor: page.cursor, + latestLedger: page.latestLedger, + }; + }, + }; + __setServerForTests(server as unknown as Server); + return { requests }; +} + +const FILTERS = [{ type: "contract" as const, contractIds: [COUNCIL] }]; + +Deno.test("drainEvents: full page → cursor continuation; head when cursor reaches latestLedger", async () => { + __resetWatcherStateForTests(); + const { requests } = stubServer([ + { events: 100, cursor: cursorAt(150, 7), latestLedger: 500 }, + { events: 100, cursor: cursorAt(300, 2), latestLedger: 500, startId: 100 }, + { events: 3, cursor: cursorAt(500), latestLedger: 500, startId: 200 }, + ]); + try { + const out = await drainEvents({ startLedger: 42 }, FILTERS, 10, newNoop()); + assertEquals(out.raws.length, 203); + assertEquals(out.cursor, cursorAt(500)); + assertEquals(out.latestLedger, 500); + assertEquals(out.complete, true); + // Request 1 windowed by ledger; requests 2-3 by the previous cursor. + assertEquals(requests[0].startLedger, 42); + assertEquals(requests[0].endLedger, 42 + 4000 - 1); + assertEquals(requests[1].cursor, cursorAt(150, 7)); + assertEquals(requests[2].cursor, cursorAt(300, 2)); + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } +}); + +Deno.test("drainEvents: sparse bounded scans step windows from scannedTo + 1", async () => { + __resetWatcherStateForTests(); + // The RPC may cover only a slice of the requested range: an empty page + // whose cursor sits below latestLedger means "scanned this far", NOT + // "caught up" — the drain must keep walking. + const { requests } = stubServer([ + { events: 0, cursor: cursorAt(4999, 4294967295), latestLedger: 20000 }, + { events: 1, cursor: cursorAt(8999, 4294967295), latestLedger: 20000 }, + { events: 0, cursor: cursorAt(20000, 4294967295), latestLedger: 20000 }, + ]); + try { + const out = await drainEvents( + { startLedger: 1000 }, + FILTERS, + 10, + newNoop(), + ); + assertEquals(out.raws.length, 1); + assertEquals(out.complete, true); + assertEquals(requests[0].startLedger, 1000); + assertEquals(requests[1].startLedger, 5000); + assertEquals(requests[2].startLedger, 9000); + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } +}); + +Deno.test("drainEvents: processing-limit rejection halves the window; cursor requests re-anchor to a window", async () => { + __resetWatcherStateForTests(); + const limitErr = { + message: "[-32001] request exceeded processing limit threshold", + }; + const { requests } = stubServer([ + { throws: limitErr }, + { events: 0, cursor: cursorAt(2999, 4294967295), latestLedger: 3000 }, + { events: 0, cursor: cursorAt(3000, 4294967295), latestLedger: 3000 }, + ]); + try { + const out = await drainEvents( + { startLedger: 1000 }, + FILTERS, + 10, + newNoop(), + ); + assertEquals(out.complete, true); + assertEquals(requests[0].endLedger, 1000 + 4000 - 1); + assertEquals(requests[1].endLedger, 1000 + 2000 - 1); // halved + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } + + // Cursor-based request hitting the limit falls back to a bounded window + // at the cursor's ledger (tail re-read is deduped downstream). + const second = stubServer([ + { throws: limitErr }, + { events: 0, cursor: cursorAt(700, 4294967295), latestLedger: 700 }, + ]); + try { + const out = await drainEvents( + { cursor: cursorAt(650, 12) }, + FILTERS, + 10, + newNoop(), + ); + assertEquals(out.complete, true); + assertEquals(second.requests[0].cursor, cursorAt(650, 12)); + assertEquals(second.requests[1].startLedger, 650); + assertEquals(second.requests[1].endLedger, 650 + 4000 - 1); + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } +}); + +Deno.test("drainEvents: first-request failure yields no cursor; later failure keeps progress", async () => { + __resetWatcherStateForTests(); + stubServer([{ throws: "boom" }]); + try { + const failed = await drainEvents( + { cursor: cursorAt(1) }, + FILTERS, + 5, + newNoop(), + ); + assertEquals(failed.raws.length, 0); + assertEquals(failed.cursor, null); + assertEquals(failed.complete, false); + + stubServer([ + { events: 100, cursor: cursorAt(150), latestLedger: 500 }, + { throws: "boom" }, + ]); + const partial = await drainEvents( + { cursor: cursorAt(1) }, + FILTERS, + 5, + newNoop(), + ); + assertEquals(partial.raws.length, 100); + assertEquals(partial.cursor, cursorAt(150)); + assertEquals(partial.complete, false); + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } +}); + +Deno.test("drainEvents retries once at the RPC retention floor (raw error object)", async () => { + __resetWatcherStateForTests(); + const { requests } = stubServer([ + // The SDK throws the raw JSON-RPC error object, not an Error. + { + throws: { + message: "startLedger must be within the ledger range: 700 - 1000", + }, + }, + { events: 1, cursor: cursorAt(1000, 4294967295), latestLedger: 1000 }, + ]); + try { + const out = await drainEvents({ startLedger: 5 }, FILTERS, 5, newNoop()); + assertEquals(out.raws.length, 1); + assertEquals(out.complete, true); + assertEquals(requests[1].startLedger, 700); + } finally { + __setServerForTests(null); + __resetWatcherStateForTests(); + } +}); + +Deno.test("minCursor picks the lowest position numerically", () => { + assertEquals( + minCursor([ + "0273690485927157760-0000000000", + "0273685443634651136-0000000157", + "0273685443634651136-0000000020", + ]), + "0273685443634651136-0000000020", + ); + assertEquals(minCursor([]), null); +}); + +Deno.test("ledgerOfCursor decodes the toid ledger", () => { + assertEquals(ledgerOfCursor(cursorAt(63722358, 42)), 63722358); + // Real mainnet cursor of the Salem channel deposit (ledger 63722358). + assertEquals(ledgerOfCursor("0273685443634651136-0000000000"), 63722358); +}); diff --git a/src/core/sync/soroban-watcher.ts b/src/core/sync/soroban-watcher.ts index 8fbeed2..4bf18aa 100644 --- a/src/core/sync/soroban-watcher.ts +++ b/src/core/sync/soroban-watcher.ts @@ -1,4 +1,6 @@ import { Server } from "stellar-sdk/rpc"; +import type { Api } from "stellar-sdk/rpc"; +import { Address, xdr } from "stellar-sdk"; import type { Logger } from "@/utils/logger/index.ts"; import { getStellarRpcUrl } from "@/config/env.ts"; import { networkState } from "@/core/state/store.ts"; @@ -14,19 +16,42 @@ import { refreshTopology } from "./topology-refresh.ts"; const POLL_INTERVAL_MS = 5_000; const LOOKBACK_LEDGERS_24H = 17_280; // ~5s ledgers × 24h const PAGE_LIMIT = 100; +/** Soroban RPC caps `contractIds` per filter (5 in stellar-soroban-rpc). */ +const MAX_CONTRACT_IDS_PER_FILTER = 5; +/** Soroban RPC caps topic patterns per filter (5). */ +const MAX_TOPIC_PATTERNS_PER_FILTER = 5; /** - * Soroban RPC caps `contractIds` per filter (5 in stellar-soroban-rpc as - * of writing). We split the watched-contracts set into chunks of this size - * and issue one getEvents call per chunk — both for the forward poll and - * the cold-start scan. + * Ledger span per windowed getEvents request. RPC providers bound the + * work a single request may do — Quasar returns `-32001 request exceeded + * processing limit threshold` on wide scans, and the cost multiplies per + * FILTER in the call (a combined councils+SAC call failed at a 2 000 + * window where each filter alone handles 4 000+; measured on mainnet). + * Hence one filter per call, `endLedger`-bounded windows, and halving on + * a processing-limit rejection down to MIN_SCAN_WINDOW_LEDGERS. */ -const CONTRACT_IDS_PER_FILTER = 5; +const INITIAL_SCAN_WINDOW_LEDGERS = 4_000; +const MIN_SCAN_WINDOW_LEDGERS = 250; +/** + * Per-call request budget for one forward-poll tick. A tick normally + * resumes at the head cursor and drains in a single request; the budget + * only matters when catching up after degradation (windowed walk: ~5 + * requests per 24h of gap). On exhaustion the drain stops at a safe + * cursor and the next tick resumes from it — nothing is skipped. + */ +const FORWARD_REQUEST_CAP = 20; +/** + * Per-call request budget for the cold-start / back-fill walks. This is + * a runaway guard, not a coverage bound: a sparse 24h walk costs ~5 + * windowed requests, so 200 covers pathological density and window + * halving. When it fires we log loudly — anything past the cap is left + * for the forward poll to pick up from the returned cursor. + */ +const SCAN_REQUEST_CAP = 200; -function chunkContractIds(ids: string[]): string[][] { - if (ids.length === 0) return []; - const out: string[][] = []; - for (let i = 0; i < ids.length; i += CONTRACT_IDS_PER_FILTER) { - out.push(ids.slice(i, i + CONTRACT_IDS_PER_FILTER)); +function chunk(items: T[], size: number): T[][] { + const out: T[][] = []; + for (let i = 0; i < items.length; i += size) { + out.push(items.slice(i, i + size)); } return out; } @@ -44,29 +69,283 @@ function getServer(): Server { } /** - * Cursor tracking. We forward-poll a union of (Channel Auth contracts) + - * (SAC contracts) — the SAC set may grow as the topology gains new - * channels, so we recompute the contractIds filter from networkState on - * every tick. + * Forward-poll position. `lastCursor` is the RPC event cursor everything + * up to which has been consumed — the resume point for the next tick. + * `lastLedgerSeen` tracks the RPC head for /health (`armed`) and as the + * startLedger fallback when no cursor exists yet (fresh boot with a + * failed cold-start scan). */ +let lastCursor: string | null = null; let lastLedgerSeen: number | null = null; let pollTimer: number | null = null; let running = false; /** * Consecutive `pollTick`s that found the watcher armed (`running`) but with no - * forward cursor (`lastLedgerSeen === null`). This is the "stranded" state: - * `bootstrap()` swallows a failed `coldStartScan` and starts the watcher - * anyway, so the poll then no-ops silently forever — the dashboard freezes - * with a green `/health` and no logs. Tracked so the strand is observable - * (loud log + `/health`) instead of invisible. Reset to 0 on any armed tick. + * forward position (`lastCursor` and `lastLedgerSeen` both null). This is the + * "stranded" state: `bootstrap()` swallows a failed `coldStartScan` and starts + * the watcher anyway, so the poll then no-ops silently forever — the dashboard + * freezes with a green `/health` and no logs. Tracked so the strand is + * observable (loud log + `/health`) instead of invisible. Reset to 0 on any + * armed tick. */ let strandedTickCount = 0; -function watchedContractIds(): string[] { - const ids = new Set(); - for (const c of networkState.getCouncilIds()) ids.add(c); - for (const a of networkState.getAssetContractIds()) ids.add(a); - return Array.from(ids); +// ── event query specs ───────────────────────────────────────────────── + +const TRANSFER_TOPIC = xdr.ScVal.scvSymbol("transfer").toXDR("base64"); +const FEE_TOPIC = xdr.ScVal.scvSymbol("fee").toXDR("base64"); + +/** Address → base64 ScVal XDR, cached (topology entries repeat every tick). */ +const addressTopicCache = new Map(); +function addressTopic(address: string): string { + let encoded = addressTopicCache.get(address); + if (!encoded) { + encoded = new Address(address).toScVal().toXDR("base64"); + addressTopicCache.set(address, encoded); + } + return encoded; +} + +export type WatchQueryCall = { + label: string; + filter: Api.EventFilter; +}; + +/** + * Build the getEvents subscriptions for the current topology — ONE filter + * per call, because the RPC's per-request processing limit multiplies with + * each filter in a call (see INITIAL_SCAN_WINDOW_LEDGERS). + * + * Channel Auth (council) contracts are low-volume, so they get plain + * contractIds filters. SAC contracts are NOT watched raw: the mainnet + * XLM SAC emits hundreds of `transfer`/`fee` events per LEDGER, which + * drowned any unfiltered subscription (a 100-event page didn't even span + * one ledger). Instead we subscribe by topic: + * + * - deposit: ["transfer", *, , *] + * - settlement: ["transfer", , *, *] + * - bundle fee: ["fee", ] + * + * Topic filters are exact-length positional (SAC `transfer` carries 4 + * topics, `fee` carries 2 — verified against mainnet), so each pattern + * matches only its event shape. Within a filter the patterns are OR'd; + * RPC caps (5 contractIds / 5 patterns per filter) drive the chunking. + */ +export function buildWatchQueryCalls(): WatchQueryCall[] { + const councilIds = [...networkState.getCouncilIds()].sort(); + const channelIds = [...networkState.getChannelContractIds()].sort(); + const sacIds = [...networkState.getAssetContractIds()].sort(); + const providerKeys = [...networkState.getProviderPublicKeys()].sort(); + + const filters: Api.EventFilter[] = []; + for (const ids of chunk(councilIds, MAX_CONTRACT_IDS_PER_FILTER)) { + filters.push({ type: "contract", contractIds: ids }); + } + + const patterns: string[][] = []; + for (const channel of channelIds) { + patterns.push([TRANSFER_TOPIC, "*", addressTopic(channel), "*"]); + patterns.push([TRANSFER_TOPIC, addressTopic(channel), "*", "*"]); + } + for (const pp of providerKeys) { + patterns.push([FEE_TOPIC, addressTopic(pp)]); + } + if (sacIds.length > 0 && patterns.length > 0) { + for (const sacs of chunk(sacIds, MAX_CONTRACT_IDS_PER_FILTER)) { + for (const pats of chunk(patterns, MAX_TOPIC_PATTERNS_PER_FILTER)) { + filters.push({ type: "contract", contractIds: sacs, topics: pats }); + } + } + } + + return filters.map((filter, i) => ({ label: `watch:${i}`, filter })); +} + +// ── cursor-paged drain ──────────────────────────────────────────────── + +export type DrainOutcome = { + raws: RawChainEvent[]; + /** + * RPC cursor after the last consumed page. Everything at or before it + * has been returned in `raws`, so advancing the poll position to it + * never skips an event. Null when not even the first page succeeded. + */ + cursor: string | null; + latestLedger: number | null; + /** True when the drain reached the RPC head (a partial page). */ + complete: boolean; +}; + +function toRawChainEvent(ev: Api.EventResponse): RawChainEvent { + return { + id: ev.id, + contractId: ev.contractId?.toString() ?? "", + ledger: ev.ledger, + topics: ev.topic, + value: ev.value, + txHash: ev.txHash ?? "", + ledgerClosedAtMs: parseLedgerClosedAt(ev.ledgerClosedAt), + }; +} + +/** Ledger encoded in an RPC event cursor (`-`). */ +export function ledgerOfCursor(cursor: string): number { + return Number(BigInt(cursor.split("-")[0]) >> 32n); +} + +/** + * Message of a getEvents failure. The SDK's JSON-RPC layer throws the raw + * `{code, message}` error object (not an Error), which stringifies to + * `[object Object]` — extract the message so error classification and + * logs stay useful. + */ +function errMessage(err: unknown): string { + if (err instanceof Error) return err.message; + if (typeof err === "object" && err !== null && "message" in err) { + return String((err as { message: unknown }).message); + } + return String(err); +} + +/** Quasar rejects over-wide scans with `-32001 ... processing limit`. */ +function isProcessingLimit(err: unknown): boolean { + return /processing limit/i.test(errMessage(err)); +} + +/** + * Page through getEvents from `base` until the RPC head or the request + * budget. Two RPC behaviours shape the loop (both measured on mainnet + * Quasar): + * + * - A response may cover only a SLICE of the requested range (bounded + * scan work), returning a partial/empty page whose cursor sits well + * below `latestLedger`. "Partial page" therefore does NOT mean + * "caught up" — head is reached only when the cursor's ledger + * reaches `latestLedger`. + * - An over-wide request fails outright with a processing-limit error + * instead of slicing. Sparse stretches are walked with + * `endLedger`-bounded windows (halved on rejection); a full page + * switches to cursor continuation, which resumes mid-ledger and + * stops cheaply at the page limit. + * + * Consumption is gap-free: every position advance is either the RPC's own + * scan cursor or `scannedTo + 1`. A processing-limit rejection of a + * cursor request falls back to a window starting at the cursor's ledger + * (re-reading that ledger's tail; dedup by event id absorbs it). + */ +export async function drainEvents( + base: { startLedger: number } | { cursor: string }, + filters: Api.EventFilter[], + requestCap: number, + log: Logger, +): Promise { + const server = getServer(); + const out: DrainOutcome = { + raws: [], + cursor: null, + latestLedger: null, + complete: false, + }; + let position: { startLedger: number } | { cursor: string } = base; + let window = INITIAL_SCAN_WINDOW_LEDGERS; + let retriedAtFloor = false; + + for (let requests = 0; requests < requestCap; requests++) { + const request: Api.GetEventsRequest = "cursor" in position + ? { cursor: position.cursor, filters, limit: PAGE_LIMIT } + : { + startLedger: position.startLedger, + endLedger: position.startLedger + window - 1, + filters, + limit: PAGE_LIMIT, + }; + let res: Api.GetEventsResponse; + try { + res = await server.getEvents(request); + } catch (err) { + if (isProcessingLimit(err)) { + if ("cursor" in position) { + // Cursor scans carry no endLedger bound; re-anchor to a window. + position = { startLedger: ledgerOfCursor(position.cursor) }; + continue; + } + if (window > MIN_SCAN_WINDOW_LEDGERS) { + window = Math.max(MIN_SCAN_WINDOW_LEDGERS, Math.floor(window / 2)); + log.debug("window", window); + log.event("processing-limit rejection; halving scan window"); + continue; + } + } + if (!retriedAtFloor && "startLedger" in position) { + const floor = parseValidRangeFloor(err); + if (floor !== null && floor > position.startLedger) { + retriedAtFloor = true; + log.debug("requestedStartLedger", position.startLedger); + log.debug("retentionFloor", floor); + log.event("startLedger below retention; retrying at floor"); + position = { startLedger: floor }; + continue; + } + } + log.debug("requests", requests); + log.error( + new Error(errMessage(err)), + "getEvents failed (drain incomplete)", + ); + return out; + } + + out.cursor = res.cursor; + out.latestLedger = res.latestLedger; + for (const ev of res.events) { + out.raws.push(toRawChainEvent(ev)); + } + + if (res.events.length === PAGE_LIMIT) { + // Dense stretch — continue mid-ledger from the exact cursor. + position = { cursor: res.cursor }; + continue; + } + const scannedTo = ledgerOfCursor(res.cursor); + if (scannedTo >= res.latestLedger) { + out.complete = true; + return out; + } + // Partial/empty page below head: the RPC bounded its scan (or our + // window ended) — step the window forward from where scanning stopped. + position = { startLedger: scannedTo + 1 }; + } + + log.debug("requestCap", requestCap); + log.debug("eventsDrained", out.raws.length); + log.event( + "drain hit request cap before reaching head; resuming from cursor next pass", + ); + return out; +} + +/** + * Numeric compare of RPC event cursors (`-`). Used to + * advance the shared poll position to the LOWEST drain point across the + * tick's calls — never past a call that consumed less. The overlap this + * re-reads on the faster calls is deduped by event id in the store. + */ +export function minCursor(cursors: string[]): string | null { + let min: string | null = null; + let minParts: [bigint, bigint] | null = null; + for (const c of cursors) { + const [a, b] = c.split("-"); + const parts: [bigint, bigint] = [BigInt(a), BigInt(b ?? "0")]; + if ( + minParts === null || + parts[0] < minParts[0] || + (parts[0] === minParts[0] && parts[1] < minParts[1]) + ) { + min = c; + minParts = parts; + } + } + return min; } export function publishMappedEvent( @@ -99,19 +378,13 @@ export function publishMappedEvent( // Channels have NO chain event analogue to `provider_added` — the // privacy_channel contract emits nothing on construction and channel // registration is a council-platform-only DB operation - // (POST /council/channels). The contract-init listener fires a refresh - // on Channel Auth deploy, but channels added AFTER that initial - // refresh (the test flow: step 7 add channel → step 9 add_provider - // on-chain) leave `channelContractToCouncil` stale. The next deposit's - // SAC transfer event lands while `resolveChannelToCouncil(privacyChannel)` - // still returns undefined → `channel_deposit` silently dropped. - // - // `add_provider` on-chain is the deterministic signal that - // council-platform definitely has the council's channels persisted by - // now (the flow always adds channels before any PP can join). Piggyback - // a topology refresh here. `refreshTopology` is single-flight - // (topology-refresh.ts:25) so concurrent fires coalesce. Fire-and-forget; - // any fetch failure is logged inside. + // (POST /council/channels). `add_provider` on-chain is a deterministic + // signal that council-platform has the council's channels persisted by + // now (the flow always adds channels before any PP can join), so + // piggyback a topology refresh here for a fast channel-linkage update; + // the periodic re-sync (scheduler) is the completeness backstop. + // `refreshTopology` is single-flight (topology-refresh.ts) so + // concurrent fires coalesce. Fire-and-forget; failures log inside. refreshTopology(`provider_added:${event.councilId}`, { log, bus }).catch( (err) => log.error(err, "refreshTopology on provider_added failed"), ); @@ -200,15 +473,15 @@ function parseLedgerClosedAt(raw: unknown): number | null { * null if the error doesn't match that pattern. */ function parseValidRangeFloor(err: unknown): number | null { - const msg = err instanceof Error ? err.message : String(err); - const match = msg.match(/ledger range:\s*(\d+)\s*-\s*\d+/); + const match = errMessage(err).match(/ledger range:\s*(\d+)\s*-\s*\d+/); return match ? Number(match[1]) : null; } /** - * Cold-start scan: walk trailing 24h on the current contractId set, + * Cold-start scan: walk trailing 24h on the current subscription set, * map events, and seed the rolling window + ring buffer in chronological - * order. Sets the forward cursor to one past the latest ledger seen. + * order. Arms the forward poller with the drain cursor (or, degraded, + * the head ledger). */ export async function coldStartScan( deps: { log: Logger; bus: NetworkEventBus }, @@ -220,10 +493,12 @@ export async function coldStartScan( const latest = await server.getLatestLedger(); // Soroban's *event* retention is much shorter than its ledger retention // (getHealth.oldestLedger). Querying below the events floor returns - // 0 events without erroring, so probe with progressively smaller - // lookbacks until events appear. Long-retention providers (testnet / - // mainnet) typically return events on the first try; the quickstart - // container needs a closer startLedger. + // 0 events without erroring on some providers, so probe with + // progressively smaller lookbacks until events appear. Long-retention + // providers (testnet / mainnet) typically return events on the first + // try; the quickstart container needs a closer startLedger. (Providers + // that ERROR below the floor instead are handled by the drain's + // retry-at-floor.) const desiredStart = Math.max( 1, latest.sequence - LOOKBACK_LEDGERS_24H, @@ -253,10 +528,12 @@ export async function coldStartScan( log.error(err, "cold-start probe failed at lookback"); } } - const contractIds = watchedContractIds(); - if (contractIds.length === 0) { - lastLedgerSeen = latest.sequence; + // Arm the forward poller up front so a scan failure doesn't strand it. + lastLedgerSeen = latest.sequence; + + const calls = buildWatchQueryCalls(); + if (calls.length === 0) { log.debug("latestLedger", latest.sequence); log.event( "cold-start scan skipped — no contracts to watch yet (no councils registered)", @@ -264,94 +541,31 @@ export async function coldStartScan( return; } - // Set the forward cursor up front so a scan failure doesn't strand - // the forward poller with a null cursor. - lastLedgerSeen = latest.sequence; - log.debug("startLedger", initialStart); log.debug("latestLedger", latest.sequence); - log.debug("contractCount", contractIds.length); + log.debug("callCount", calls.length); log.event("cold-start scan starting"); const rawBatch: RawChainEvent[] = []; - - // Walk forward in ledger ranges for each contractIds chunk independently. - // Each chunk holds at most CONTRACT_IDS_PER_FILTER contracts; getEvents - // pages until a partial page indicates "caught up to head" for that - // chunk. The page cap (50) is per-chunk and only fires on pathological - // event volume. - let totalPages = 0; - for (const chunk of chunkContractIds(contractIds)) { - let nextLedger = initialStart; - let page = 0; - while (true) { - let res; - try { - res = await server.getEvents({ - startLedger: nextLedger, - filters: [{ type: "contract", contractIds: chunk }], - limit: PAGE_LIMIT, - }); - } catch (err) { - // First-page out-of-range failure: retry once at the RPC's valid floor. - const floor = page === 0 ? parseValidRangeFloor(err) : null; - if (floor !== null && floor > nextLedger) { - log.debug("requestedStartLedger", nextLedger); - log.debug("retentionFloor", floor); - log.event( - "cold-start scan startLedger below retention; retrying at floor", - ); - nextLedger = floor; - try { - res = await server.getEvents({ - startLedger: nextLedger, - filters: [{ type: "contract", contractIds: chunk }], - limit: PAGE_LIMIT, - }); - } catch (err2) { - log.debug("startLedger", nextLedger); - log.error(err2, "cold-start scan retry at retention floor failed"); - break; - } - } else { - log.debug("page", page); - log.debug("startLedger", nextLedger); - log.debug("chunkSize", chunk.length); - log.error(err, "cold-start scan page failed (stopping chunk)"); - break; - } - } - page++; - totalPages++; - for (const ev of res.events) { - rawBatch.push({ - id: ev.id, - contractId: ev.contractId?.toString() ?? "", - ledger: ev.ledger, - topics: ev.topic, - value: ev.value, - txHash: ev.txHash ?? "", - ledgerClosedAtMs: parseLedgerClosedAt(ev.ledgerClosedAt), - }); - } - lastLedgerSeen = res.latestLedger; - if (res.events.length < PAGE_LIMIT) break; - const lastLedgerInPage = res.events[res.events.length - 1].ledger; - nextLedger = lastLedgerInPage + 1; - if (page >= 50) { - log.debug("eventsSoFar", rawBatch.length); - log.debug("chunkSize", chunk.length); - log.event("cold-start scan hit page cap (50) for chunk; stopping"); - break; - } + const advances: (string | null)[] = []; + for (const call of calls) { + const drained = await drainEvents( + { startLedger: initialStart }, + [call.filter], + SCAN_REQUEST_CAP, + log.scope(call.label), + ); + rawBatch.push(...drained.raws); + advances.push(drained.cursor); + if (drained.latestLedger !== null) { + lastLedgerSeen = Math.max(lastLedgerSeen, drained.latestLedger); } } - const page = totalPages; // Process the whole accumulated batch with per-tx dedup, then seed. // Back-fill records carry null latency — the store only computes the - // avg-latency counter from live observations. Chunked scans return - // events grouped per chunk, so re-sort by ledger here to keep the + // avg-latency counter from live observations. Per-call drains return + // events grouped per call, so re-sort by ledger here to keep the // ring-buffer chronological. rawBatch.sort((a, b) => a.ledger - b.ledger); const chronological = processRawEventBatch(rawBatch, log).map((p) => p.event); @@ -360,8 +574,19 @@ export async function coldStartScan( const newestFirst = [...chronological].reverse(); networkState.seedRecent(newestFirst); - log.debug("pagesWalked", page); + if (advances.length > 0 && advances.every((c) => c !== null)) { + lastCursor = minCursor(advances as string[]); + } else { + // A call failed before its first page: no gap-free cursor exists, so + // the forward poll starts at head (lastLedgerSeen + 1). Loud — the + // window between the failed call's coverage and head is lost. + log.event( + "cold-start drain incomplete; forward poll starts at head (some history may be missing)", + ); + } + log.debug("eventsSeeded", chronological.length); + log.debug("lastCursor", lastCursor); log.debug("lastLedgerSeen", lastLedgerSeen); log.event("cold-start scan complete"); } @@ -370,17 +595,18 @@ async function pollTick( deps: { log: Logger; bus: NetworkEventBus }, ): Promise { if (!running) return; - if (lastLedgerSeen === null) { - // Armed but no cursor — cold-start threw (its error is caught + swallowed - // in `bootstrap()`) yet the watcher was started. Every tick then no-ops - // SILENTLY, freezing the dashboard while `/health` stays green. Make it - // loud (rate-limited to ~once/5min at a 5s interval) and observable via - // `getWatcherHealth()` so the strand surfaces instead of hiding. + if (lastCursor === null && lastLedgerSeen === null) { + // Armed but no position — cold-start threw before establishing one + // (its error is caught + swallowed in `bootstrap()`) yet the watcher + // was started. Every tick then no-ops SILENTLY, freezing the dashboard + // while `/health` stays green. Make it loud (rate-limited to + // ~once/5min at a 5s interval) and observable via `getWatcherHealth()` + // so the strand surfaces instead of hiding. strandedTickCount += 1; if (strandedTickCount === 1 || strandedTickCount % 60 === 0) { deps.log.scope("pollTick").error( - new Error("forward poller stranded: lastLedgerSeen === null"), - "watcher armed but has no cursor — cold-start did not complete; NO events will be ingested until cold-start succeeds (restart/redeploy)", + new Error("forward poller stranded: no cursor and no ledger position"), + "watcher armed but has no position — cold-start did not complete; NO events will be ingested until cold-start succeeds (restart/redeploy)", ); } return; @@ -388,78 +614,49 @@ async function pollTick( strandedTickCount = 0; const log = deps.log.scope("pollTick"); - const contractIds = watchedContractIds(); + const base: { startLedger: number } | { cursor: string } = lastCursor !== null + ? { cursor: lastCursor } + : { startLedger: (lastLedgerSeen as number) + 1 }; - // Soroban's getEvents intersects multiple filter entries within a single - // call: combining `contractIds: [...]` with a separate topic-only filter - // restricts the response to the contractIds — the topic-only filter is - // effectively ignored. So we issue two independent calls per tick: - // A) known contracts (the existing behaviour) - // B) network-wide `contract_initialized` for new-council discovery - // and merge their result sets. - const server = getServer(); - const startLedger = lastLedgerSeen + 1; - let nextLastLedger = lastLedgerSeen; const rawBatch: RawChainEvent[] = []; - const unknownCandidates = new Set(); - const knownIds = new Set(contractIds); - - // Soroban caps contractIds-per-filter at 5; once the network grows past - // that we have to split the known-contracts subscription into multiple - // getEvents calls. One call per chunk per tick. - for (const chunk of chunkContractIds(contractIds)) { - try { - const res = await server.getEvents({ - startLedger, - filters: [{ type: "contract", contractIds: chunk }], - limit: PAGE_LIMIT, - }); - nextLastLedger = Math.max(nextLastLedger, res.latestLedger); - for (const ev of res.events) { - rawBatch.push({ - id: ev.id, - contractId: ev.contractId?.toString() ?? "", - ledger: ev.ledger, - topics: ev.topic, - value: ev.value, - txHash: ev.txHash ?? "", - ledgerClosedAtMs: parseLedgerClosedAt(ev.ledgerClosedAt), - }); - } - } catch (err) { - log.debug("chunkSize", chunk.length); - log.error(err, "Soroban poll (known contracts) failed"); + const advances: (string | null)[] = []; + + // Known-topology subscriptions (councils by contractId, SAC by topic). + for (const call of buildWatchQueryCalls()) { + const drained = await drainEvents( + base, + [call.filter], + FORWARD_REQUEST_CAP, + log.scope(call.label), + ); + rawBatch.push(...drained.raws); + advances.push(drained.cursor); + if (drained.latestLedger !== null) { + lastLedgerSeen = Math.max(lastLedgerSeen ?? 0, drained.latestLedger); } } - // Always poll for fresh `contract_initialized` events from contracts - // outside the watched set. This is the event-driven new-council - // discovery path — the listener no longer gates on a WASM-hash - // registry, so local-dev environments without GitHub access for the - // soroban-core releases listing still discover new councils as they - // deploy. - try { - const res = await server.getEvents({ - startLedger, - filters: [{ - type: "contract", - topics: [CONTRACT_INITIALIZED_TOPIC_PATTERN], - }], - limit: PAGE_LIMIT, - }); - nextLastLedger = Math.max(nextLastLedger, res.latestLedger); - for (const ev of res.events) { - const cid = ev.contractId?.toString() ?? ""; - if (!cid) continue; - // Already-known contracts are handled by the contractIds-filter - // call above (which carries full event data); skip the duplicate. - if (knownIds.has(cid)) continue; - unknownCandidates.add(cid); - } - } catch (err) { - log.error(err, "Soroban poll (contract_initialized) failed"); + // Network-wide `contract_initialized` poll for new-council discovery — + // contracts outside the subscription set feed the adoption pipeline. + const discovery = await drainEvents( + base, + [{ type: "contract", topics: [CONTRACT_INITIALIZED_TOPIC_PATTERN] }], + FORWARD_REQUEST_CAP, + log.scope("discovery"), + ); + advances.push(discovery.cursor); + if (discovery.latestLedger !== null) { + lastLedgerSeen = Math.max(lastLedgerSeen ?? 0, discovery.latestLedger); + } + for (const raw of discovery.raws) { + // Already-known councils are handled by the subscription calls above + // (which carry full event data); skip the duplicate. + if (!raw.contractId) continue; + if (networkState.hasCouncil(raw.contractId)) continue; + evaluateUnknownContract(raw.contractId, raw.ledger, deps); } + rawBatch.sort((a, b) => a.ledger - b.ledger); for (const processed of processRawEventBatch(rawBatch, log)) { publishMappedEvent( processed.event, @@ -468,21 +665,25 @@ async function pollTick( log, ); } - for (const cid of unknownCandidates) { - evaluateUnknownContract(cid, startLedger, deps); - } - // Refresh topology once if there's anything pending, then adopt - // (or cache as not-ours) each unknown. On adoption, back-fill from the - // earliest observed-at-ledger across the freshly-adopted contracts so - // events emitted between deploy and adoption (e.g. provider_added) are - // still published live. + // Refresh topology if there's anything pending, then adopt (or cache as + // not-ours) each unknown. On adoption, back-fill from the earliest + // observed-at-ledger across the freshly-adopted contracts so events + // emitted between deploy and adoption (e.g. provider_added) are still + // published live. drainPendingAdoptions({ ...deps, backfillFromLedger, }).catch((err) => { log.error(err, "drainPendingAdoptions failed"); }); - lastLedgerSeen = nextLastLedger; + + // Advance the shared position to the lowest safe cursor across all + // calls — never past a call that consumed less, and not at all if any + // call failed outright (its events would be skipped). Re-reads on the + // faster calls are deduped by event id in the store. + if (advances.length > 0 && advances.every((c) => c !== null)) { + lastCursor = minCursor(advances as string[]); + } } function scheduleNext(deps: { log: Logger; bus: NetworkEventBus }): void { @@ -497,9 +698,9 @@ function scheduleNext(deps: { log: Logger; bus: NetworkEventBus }): void { * Liveness snapshot of the forward poller for `/health`. * * `running && !armed` is the stranded state (cold-start failed, watcher - * started with a null cursor) — the poller ingests nothing. Surfacing it lets - * `/health` report degraded so the strand is caught by monitoring / the Fly - * health check instead of silently freezing the dashboard for weeks. + * started with no forward position) — the poller ingests nothing. Surfacing + * it lets `/health` report degraded so the strand is caught by monitoring / + * the Fly health check instead of silently freezing the dashboard for weeks. */ export function getWatcherHealth(): { running: boolean; @@ -508,7 +709,7 @@ export function getWatcherHealth(): { } { return { running, - armed: lastLedgerSeen !== null, + armed: lastCursor !== null || lastLedgerSeen !== null, strandedTickCount, }; } @@ -520,6 +721,7 @@ export function startSorobanWatcher( running = true; const log = deps.log.scope("sorobanWatcher"); log.debug("intervalMs", POLL_INTERVAL_MS); + log.debug("lastCursor", lastCursor); log.debug("lastLedgerSeen", lastLedgerSeen); log.event("soroban watcher started"); scheduleNext(deps); @@ -534,26 +736,18 @@ export function stopSorobanWatcher(deps: { log: Logger }): void { deps.log.scope("sorobanWatcher").event("soroban watcher stopped"); } -/** Re-anchor the rolling 24h counter window after the hourly re-sync. */ -export async function rescanRollingWindow( - deps: { log: Logger; bus: NetworkEventBus }, -): Promise { - await coldStartScan(deps); -} - /** * Back-fill scan invoked after a newly-discovered Channel Auth contract is - * adopted into the topology. Walks the current `watchedContractIds()` set - * from `fromLedger` forward, maps each event, and publishes via the bus. + * adopted into the topology. Walks the current subscription set from + * `fromLedger` forward, maps each event, and publishes via the bus. * Dedup is handled by `networkState.recordEvent` in `publishMappedEvent` — * events the forward poller has already published are skipped, so calling * this concurrently with `pollTick` is safe. * - * The scan covers ALL watched contracts (not just the newly-adopted ones) - * because some events involving a fresh council fire on a SHARED contract - * — e.g. the XLM SAC `transfer` and `fee` events fan out across every - * council and need to be mapped via the contract-id linkage that - * `refreshTopology` just installed. + * The scan covers the WHOLE subscription set (not just the newly-adopted + * council) because some events involving a fresh council fire on a SHARED + * contract — e.g. the XLM SAC `transfer`/`fee` events are matched via the + * channel/PP topic patterns that `refreshTopology` just installed. */ export async function backfillFromLedger( fromLedger: number, @@ -563,51 +757,21 @@ export async function backfillFromLedger( log.info("backfillFromLedger"); log.debug("fromLedger", fromLedger); - const contractIds = watchedContractIds(); - if (contractIds.length === 0) { + const calls = buildWatchQueryCalls(); + if (calls.length === 0) { log.event("back-fill skipped — no contracts watched"); return; } - const server = getServer(); const rawBatch: RawChainEvent[] = []; - for (const chunk of chunkContractIds(contractIds)) { - let nextLedger = fromLedger; - let page = 0; - while (true) { - let res; - try { - res = await server.getEvents({ - startLedger: nextLedger, - filters: [{ type: "contract", contractIds: chunk }], - limit: PAGE_LIMIT, - }); - } catch (err) { - log.debug("chunkSize", chunk.length); - log.debug("startLedger", nextLedger); - log.error(err, "back-fill page failed"); - break; - } - page++; - for (const ev of res.events) { - rawBatch.push({ - id: ev.id, - contractId: ev.contractId?.toString() ?? "", - ledger: ev.ledger, - topics: ev.topic, - value: ev.value, - txHash: ev.txHash ?? "", - ledgerClosedAtMs: parseLedgerClosedAt(ev.ledgerClosedAt), - }); - } - if (res.events.length < PAGE_LIMIT) break; - nextLedger = res.events[res.events.length - 1].ledger + 1; - if (page >= 50) { - log.debug("eventsSoFar", rawBatch.length); - log.event("back-fill page cap (50) reached for chunk; stopping"); - break; - } - } + for (const call of calls) { + const drained = await drainEvents( + { startLedger: fromLedger }, + [call.filter], + SCAN_REQUEST_CAP, + log.scope(call.label), + ); + rawBatch.push(...drained.raws); } rawBatch.sort((a, b) => a.ledger - b.ledger); @@ -622,3 +786,15 @@ export async function backfillFromLedger( } log.event("back-fill scan complete"); } + +/** Test-only seams. */ +export function __setServerForTests(server: Server | null): void { + rpcServer = server; +} + +export function __resetWatcherStateForTests(): void { + lastCursor = null; + lastLedgerSeen = null; + strandedTickCount = 0; + addressTopicCache.clear(); +} diff --git a/src/core/sync/topology-refresh.ts b/src/core/sync/topology-refresh.ts index 9973699..4262a48 100644 --- a/src/core/sync/topology-refresh.ts +++ b/src/core/sync/topology-refresh.ts @@ -10,11 +10,9 @@ import { fetchCouncilTopology } from "./council-fetch.ts"; * * The caller is responsible for any catch-up of historical events for * newly-adopted contracts (via `backfillFromLedger` in `soroban-watcher`). - * We deliberately do NOT call `rescanRollingWindow` here: that helper - * runs `coldStartScan` which `seedRecent`s the ring buffer with - * cold-scanned events, and `publishMappedEvent`'s dedup check uses that - * ring buffer — so a `rescanRollingWindow` call followed by a back-fill - * would silently dedup every back-filled event before it could reach the + * We deliberately do NOT re-run the cold-start scan here: it `seedRecent`s + * the ring buffer with cold-scanned events, and `publishMappedEvent`'s + * dedup would then skip every back-filled event before it could reach the * bus. The contract-init-listener path relies on `backfillFromLedger` to * fan out the historical-but-newly-relevant events live. * diff --git a/src/main.ts b/src/main.ts index 97ab34d..18e2d74 100644 --- a/src/main.ts +++ b/src/main.ts @@ -23,9 +23,10 @@ import { startScheduler, stopScheduler } from "@/core/sync/scheduler.ts"; * 3. Start the forward poller (Soroban watcher). It also polls * Soroban-wide for `contract_initialized` events from unknown * contracts and feeds them to the contract-init-listener, which - * triggers an immediate topology refresh on each unknown — this - * replaces the hourly periodic re-sync. - * 4. Start the scheduler (minute window sweep only). + * triggers a topology refresh + adoption back-fill per unknown. + * 4. Start the scheduler (minute window sweep + periodic topology + * re-sync — the backstop for DB-only registrations like channels + * and jurisdictions, which emit no chain event). * 5. Start the HTTP server. * * Steps 1-2 are best-effort: a failure logs + continues so the service