From 883dabb2b554a43c43f8f3666f4af9c1f8dcc10c Mon Sep 17 00:00:00 2001 From: Gorka Date: Wed, 24 Jun 2026 10:25:52 -0300 Subject: [PATCH 1/2] fix(event-watcher): resolve boot start ledger inside the retried poll loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A transient RPC failure while resolving the boot start ledger could permanently disable the watcher. start() resolved the boot ledger via an un-retried getHealth() RPC before scheduling the poll loop; if that call rejected, start() rejected, the loop never ran, the error was swallowed, and the watcher stayed dead for the life of the process — blinding the provider to all live on-chain membership/channel events until a manual restart. Move boot-ledger resolution into the first poll(), inside the existing try/catch + retry machinery. A transient failure is now caught and retried on the next tick, exactly like any poll error, instead of being terminal. Boot-sync semantics are unchanged: oldest available, or the pinned BOOT_SYNC_START_LEDGER_BLOCK override — never "latest"; converge-by-query stays the recovery path; no durable-cursor behaviour. Add a test proving a transient failure on the first boot resolution does not kill the watcher: it retries and, once the RPC recovers, polls and processes events normally. --- .../event-watcher.process.test.ts | 117 +++++++++++++++++- .../event-watcher/event-watcher.process.ts | 37 ++++-- 2 files changed, 140 insertions(+), 14 deletions(-) diff --git a/src/core/service/event-watcher/event-watcher.process.test.ts b/src/core/service/event-watcher/event-watcher.process.test.ts index 226981d..535ea83 100644 --- a/src/core/service/event-watcher/event-watcher.process.test.ts +++ b/src/core/service/event-watcher/event-watcher.process.test.ts @@ -1,11 +1,19 @@ import { assertEquals } from "@std/assert"; +import { Address, Keypair, xdr } from "stellar-sdk"; import type { Server } from "stellar-sdk/rpc"; import { EventWatcher } from "./event-watcher.process.ts"; +import type { ChannelAuthEvent } from "./event-watcher.types.ts"; import { newNoop } from "@/utils/logger/index.ts"; const CONTRACT = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFCT4"; const CONTRACT_B = "CBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBV6"; +// The boot start ledger now resolves inside the first poll (so a transient +// failure retries instead of killing the watcher in start()), so every +// assertion about where the watcher began polling must let that fire-and-forget +// first poll run to completion. +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + /** * Records the `startLedger` of each getEvents call so a test can assert exactly * where a fresh watcher began polling. `latestLedger` controls how far the @@ -60,6 +68,7 @@ Deno.test("EventWatcher - override set → first getEvents at exactly that ledge ); await watcher.start(); + await tick(); // let the first poll resolve the boot ledger and fetch watcher.stop(); assertEquals(startLedgers[0], 12345); @@ -76,14 +85,12 @@ Deno.test("EventWatcher - override unset → first getEvents at oldest available ); await watcher.start(); + await tick(); // let the first poll resolve the boot ledger and fetch watcher.stop(); assertEquals(startLedgers[0], 5000); }); -// Let the fire-and-forget first poll (scheduled by start()) run to completion. -const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); - Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldest", async () => { // First watcher polls once and advances its in-memory cursor past 5100. const first = mockRpc({ oldestLedger: 5000, latestLedger: 5100 }); @@ -104,6 +111,7 @@ Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldes { log: newNoop(), rpc: second.rpc, startLedgerBlock: null }, ); await w2.start(); + await tick(); // let the fresh watcher's first poll resolve + fetch w2.stop(); assertEquals(second.startLedgers[0], 5000); @@ -185,3 +193,106 @@ Deno.test("EventWatcher - empty contract set holds the cursor and skips the RPC" assertEquals(polledContractIds.length, 0); assertEquals(watcher.getLastLedger(), 7000); }); + +// --- Boot resilience --- + +/** A raw provider_added RPC event the real parser accepts, for CONTRACT. */ +function rawProviderAddedEvent(address: string, ledger: number) { + return { + type: "contract" as const, + ledger, + topic: [ + xdr.ScVal.scvSymbol("provider_added"), + new Address(address).toScVal(), + ], + value: xdr.ScVal.scvVoid(), + id: `${ledger}-0`, + pagingToken: `${ledger}-0`, + inSuccessfulContractCall: true, + contractId: CONTRACT, + }; +} + +/** + * RPC mock whose boot-ledger resolution (getHealth) THROWS on its first call + * and recovers afterwards. Once recovered, getEvents serves a single + * provider_added event so a test can prove the watcher actually resumes polling + * and processing — not merely that it survived. + */ +function flakyBootRpc( + opts: { oldestLedger: number; latestLedger: number; address: string }, +) { + let healthCalls = 0; + let eventsServed = false; + const rpc = { + // deno-lint-ignore require-await -- mock satisfies async getHealth contract + getHealth: async () => { + healthCalls++; + if (healthCalls === 1) { + throw new Error("transient RPC failure resolving boot ledger"); + } + return { oldestLedger: opts.oldestLedger }; + }, + // deno-lint-ignore require-await -- mock satisfies async getEvents contract + getEvents: async () => { + // Serve the event exactly once so the watcher processes it after recovery. + const events = eventsServed + ? [] + : [rawProviderAddedEvent(opts.address, opts.latestLedger)]; + eventsServed = true; + return { events, latestLedger: opts.latestLedger }; + }, + // deno-lint-ignore require-await -- mock satisfies async getLatestLedger contract + getLatestLedger: async () => ({ sequence: opts.latestLedger }), + } as unknown as Server; + return { rpc, healthCalls: () => healthCalls }; +} + +/** Poll `predicate` on the macrotask queue until true or `timeoutMs` elapses. */ +async function waitUntil( + predicate: () => boolean, + timeoutMs = 2000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return true; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + return predicate(); +} + +Deno.test("EventWatcher - transient boot-ledger failure retries instead of killing the watcher", async () => { + const ADDRESS = Keypair.random().publicKey(); + const { rpc, healthCalls } = flakyBootRpc({ + oldestLedger: 5000, + latestLedger: 5100, + address: ADDRESS, + }); + + const received: ChannelAuthEvent[] = []; + const watcher = new EventWatcher( + { contractIds: [CONTRACT], intervalMs: 20 }, + { log: newNoop(), rpc, startLedgerBlock: null }, + ); + watcher.onEvent((event) => { + received.push(event); + }); + + // The first poll's getHealth throws. Pre-fix this rejected start() and the + // poll loop never ran; the watcher stayed dead forever. It must instead retry. + await watcher.start(); + + const recovered = await waitUntil(() => received.length > 0); + watcher.stop(); + + // The first boot resolution failed (proving the failure path was exercised)… + // …yet the watcher recovered: it resolved the boot ledger on a later tick, + // polled, and processed the on-chain event. + assertEquals(recovered, true); + assertEquals(healthCalls() >= 2, true); + assertEquals(received.length, 1); + assertEquals(received[0].type, "provider_added"); + assertEquals(received[0].address, ADDRESS); + // Cursor advanced past the latest ledger — the loop is live, not stuck at boot. + assertEquals(watcher.getLastLedger(), 5101); +}); diff --git a/src/core/service/event-watcher/event-watcher.process.ts b/src/core/service/event-watcher/event-watcher.process.ts index b23c812..38c758e 100644 --- a/src/core/service/event-watcher/event-watcher.process.ts +++ b/src/core/service/event-watcher/event-watcher.process.ts @@ -101,26 +101,28 @@ export class EventWatcher { /** * Starts the event watcher polling loop. */ - async start(): Promise { + start(): Promise { if (this.isRunning) { this.log.event("EventWatcher is already running"); - return; + return Promise.resolve(); } this.isRunning = true; - // No persisted cursor: resolve the boot start ledger (oldest available, or - // the configured override) and sync forward. - this.lastLedger = await resolveBootStartLedger( - this.rpc, - this.startLedgerBlock, - ); + // No persisted cursor: the boot start ledger (oldest available, or the + // configured override) is resolved by the FIRST poll, not here. Resolving + // it inside the poll loop means a transient RPC failure while resolving it + // is caught and retried on the next tick — exactly like any poll error — + // instead of rejecting start() and leaving the watcher permanently dead + // for the life of the process. The cursor stays null until resolved. this.log.debug("contractCount", this.contractIds.size); - this.log.debug("startLedger", this.lastLedger); - this.log.event("EventWatcher initialized boot start ledger"); + this.log.event( + "EventWatcher started; boot start ledger resolves on first poll", + ); // Start the self-scheduling loop this.scheduleNext(); + return Promise.resolve(); } /** @@ -166,7 +168,20 @@ export class EventWatcher { private poll(): Promise { return withSpan("EventWatcher.poll", async (span) => { try { - if (this.lastLedger === null) return; + // First poll with no resolved boot position: resolve it now, INSIDE + // this try/catch + retry machinery. A transient failure here (e.g. the + // getHealth RPC blipping) is caught below and retried on the next tick + // rather than terminally killing the watcher. Boot-sync semantics are + // unchanged: oldest available, or the pinned override — never "latest". + if (this.lastLedger === null) { + this.lastLedger = await resolveBootStartLedger( + this.rpc, + this.startLedgerBlock, + ); + this.log.debug("contractCount", this.contractIds.size); + this.log.debug("startLedger", this.lastLedger); + this.log.event("EventWatcher resolved boot start ledger"); + } const { events, latestLedger } = await fetchChannelAuthEvents( this.rpc, From a638299adecea49a99f3c89418e8800965402599 Mon Sep 17 00:00:00 2001 From: Gorka Date: Wed, 24 Jun 2026 10:27:04 -0300 Subject: [PATCH 2/2] chore: bump version to 0.9.7 --- deno.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deno.json b/deno.json index 891120a..eb9a530 100644 --- a/deno.json +++ b/deno.json @@ -1,6 +1,6 @@ { "name": "@moonlight-protocol/provider-platform", - "version": "0.9.6", + "version": "0.9.7", "license": "MIT", "exports": "./src/main.ts", "tasks": {