Merge pull request #111 from usherlabs/feat/deposit-poller-liveness-signal

feat: report deposit poller liveness and bound the venue call

Author: outerlook

src/helpers/deposit-archive-poller.ts

@@ -28,6 +28,11 @@ export type DepositArchivePollerConfig = {
 	// Constant defaults (no env vars): every broker env var must be allowlisted in a
 	// Gramine manifest in another repo, so the poller intentionally introduces none.
 	pollIntervalMs: number;
+	// Bounds the venue call. A fetchDeposits promise that never settles would
+	// strand #pollOne forever: no error, no metric, no reschedule — the one
+	// poller death mode that leaves no trace at all. The bound converts it into
+	// an ordinary poll failure, which is already observable.
+	fetchTimeoutMs: number;
 	// How far back the first poll of an account reaches. A restart loses the
 	// in-memory cursor and re-scans this window; duplicate rows are acceptable
 	// because transfer_events is plain MergeTree and consumers deduplicate at read
@@ -39,10 +44,27 @@ export type DepositArchivePollerConfig = {
 
 const DEFAULT_CONFIG: DepositArchivePollerConfig = {
 	pollIntervalMs: 60_000,
+	fetchTimeoutMs: 30_000,
 	lookbackMs: 24 * 60 * 60 * 1000,
 	depositsLimit: 50,
 };
 
+function withTimeout<T>(
+	promise: Promise<T>,
+	timeoutMs: number,
+	label: string,
+): Promise<T> {
+	let timer: ReturnType<typeof setTimeout> | undefined;
+	const expiry = new Promise<never>((_resolve, reject) => {
+		timer = setTimeout(
+			() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)),
+			timeoutMs,
+		);
+		timer.unref?.();
+	});
+	return Promise.race([promise, expiry]).finally(() => clearTimeout(timer));
+}
+
 const ALL_CURRENCIES_CODE = "*";
 
 type DepositPollTarget = {
@@ -51,6 +73,8 @@ type DepositPollTarget = {
 	code: typeof ALL_CURRENCIES_CODE;
 };
 
+type PollOutcome = "ok" | "error" | "unsupported";
+
 type LastArchivedDeposit = {
 	status: string | undefined;
 	timestamp: number | undefined;
@@ -218,7 +242,24 @@ export class DepositArchivePoller {
 		return true;
 	}
 
+	// The heartbeat is the only signal that separates a healthy-idle poller from a
+	// hung one: the archive and error counters are both silent on a quiet venue.
+	// It therefore records on every exit, including an unexpected throw, which is
+	// why the outcome starts pessimistic and is only narrowed by a completed poll.
 	async #pollOne(target: DepositPollTarget): Promise<void> {
+		let outcome: PollOutcome = "error";
+		try {
+			outcome = await this.#pollTarget(target);
+		} finally {
+			void this.params.metrics?.recordCounter(
+				"cex_deposit_poller_polls_total",
+				1,
+				{ exchange: target.exchangeId, outcome },
+			);
+		}
+	}
+
+	async #pollTarget(target: DepositPollTarget): Promise<PollOutcome> {
 		const exchange = target.account.exchange as unknown as ExchangeWithDeposits;
 		const key = this.#targetKey(target);
 		if (
@@ -232,17 +273,17 @@ export class DepositArchivePoller {
 					account: target.account.label,
 				});
 			}
-			return;
+			return "unsupported";
 		}
 
 		const since =
 			this.#cursors.get(key) ?? Date.now() - this.#config.lookbackMs;
 		let deposits: unknown[];
 		try {
-			deposits = await exchange.fetchDeposits(
-				undefined,
-				since,
-				this.#config.depositsLimit,
+			deposits = await withTimeout(
+				exchange.fetchDeposits(undefined, since, this.#config.depositsLimit),
+				this.#config.fetchTimeoutMs,
+				"fetchDeposits",
 			);
 		} catch (error) {
 			void this.params.metrics?.recordCounter(
@@ -255,10 +296,10 @@ export class DepositArchivePoller {
 				account: target.account.label,
 				error,
 			});
-			return;
+			return "error";
 		}
 		if (!Array.isArray(deposits) || deposits.length === 0) {
-			return;
+			return "ok";
 		}
 
 		let archived = 0;
@@ -359,6 +400,7 @@ export class DepositArchivePoller {
 				this.#lastArchivedByTarget.delete(key);
 			}
 		}
+		return "ok";
 	}
 
 	#targetKey(target: DepositPollTarget): string {

test/deposit-archive-poller.test.ts

@@ -27,6 +27,26 @@ function poolWith(exchange: unknown): Record<string, BrokerPoolEntry> {
 	};
 }
 
+type RecordedCounter = {
+	name: string;
+	value: number;
+	labels: Record<string, string | number>;
+};
+
+function fakeMetrics(sink: RecordedCounter[]) {
+	return {
+		recordCounter: async (
+			name: string,
+			value: number,
+			labels: Record<string, string | number>,
+		) => {
+			sink.push({ name, value, labels });
+		},
+	} as unknown as ConstructorParameters<
+		typeof DepositArchivePoller
+	>[0]["metrics"];
+}
+
 function fakeArchiver(sink: BrokerArchiveRow[]): BrokerExecutionArchiver {
 	return {
 		isEnabled: () => true,
@@ -392,3 +412,111 @@ describe("DepositArchivePoller.pollAllOnce", () => {
 		}
 	});
 });
+
+describe("DepositArchivePoller liveness signal", () => {
+	test("records a heartbeat for a successful poll", async () => {
+		const counters: RecordedCounter[] = [];
+		const exchange = {
+			has: { fetchDeposits: true },
+			fetchDeposits: async () => [],
+		};
+		const poller = new DepositArchivePoller({
+			brokers: poolWith(exchange),
+			archiver: fakeArchiver([]),
+			metrics: fakeMetrics(counters),
+		});
+
+		await poller.pollAllOnce();
+
+		expect(counters).toEqual([
+			{
+				name: "cex_deposit_poller_polls_total",
+				value: 1,
+				labels: { exchange: "binance", outcome: "ok" },
+			},
+		]);
+	});
+
+	test("records a heartbeat for an account without fetchDeposits", async () => {
+		const info = spyOn(log, "info").mockImplementation(() => {});
+		try {
+			const counters: RecordedCounter[] = [];
+			const poller = new DepositArchivePoller({
+				brokers: poolWith({ has: { fetchDeposits: false } }),
+				archiver: fakeArchiver([]),
+				metrics: fakeMetrics(counters),
+			});
+
+			await poller.pollAllOnce();
+
+			expect(counters).toEqual([
+				{
+					name: "cex_deposit_poller_polls_total",
+					value: 1,
+					labels: { exchange: "binance", outcome: "unsupported" },
+				},
+			]);
+		} finally {
+			info.mockRestore();
+		}
+	});
+
+	test("counts a hung fetchDeposits as a failed poll and polls again", async () => {
+		const warn = spyOn(log, "warn").mockImplementation(() => {});
+		try {
+			let calls = 0;
+			const exchange = {
+				has: { fetchDeposits: true },
+				fetchDeposits: async () => {
+					calls += 1;
+					if (calls === 1) {
+						// Never settles: the silent-death mode the timeout exists for.
+						return new Promise<unknown[]>(() => {});
+					}
+					return [
+						{
+							txid: "0xafter-hang",
+							currency: "USDC",
+							amount: "5",
+							status: "ok",
+							timestamp: Date.now() + 10_000,
+						},
+					];
+				},
+			};
+			const counters: RecordedCounter[] = [];
+			const sink: BrokerArchiveRow[] = [];
+			const poller = new DepositArchivePoller({
+				brokers: poolWith(exchange),
+				archiver: fakeArchiver(sink),
+				metrics: fakeMetrics(counters),
+				config: { fetchTimeoutMs: 10 },
+			});
+
+			expect(await poller.pollAllOnce()).toBe(true);
+
+			expect(warn.mock.calls[0]?.[0]).toBe("Deposit archive poll failed");
+			expect(
+				String((warn.mock.calls[0]?.[1] as { error: unknown }).error),
+			).toContain("fetchDeposits timed out after 10ms");
+			expect(counters).toEqual([
+				{
+					name: "cex_deposit_poller_errors_total",
+					value: 1,
+					labels: { exchange: "binance" },
+				},
+				{
+					name: "cex_deposit_poller_polls_total",
+					value: 1,
+					labels: { exchange: "binance", outcome: "error" },
+				},
+			]);
+
+			expect(await poller.pollAllOnce()).toBe(true);
+			expect(sink).toHaveLength(1);
+			expect(sink[0]?.row).toMatchObject({ external_id: "0xafter-hang" });
+		} finally {
+			warn.mockRestore();
+		}
+	});
+});