Merge pull request #73 from usherlabs/fix/ohlcv-subscribe-watch-stream

Fix OHLCV subscriptions: stream via watchOHLCV and guard bar re-emits

Author: outerlook

src/handlers/subscribe/handler.ts

@@ -679,7 +679,7 @@ export function createSubscribeHandler(deps: SubscribeDeps) {
 							}
 						}
 						while (ohlcvStreamActive && !isStreamClosed()) {
-							const data = await broker.fetchOHLCVWs(resolvedSymbol, timeframe);
+							const data = await broker.watchOHLCV(resolvedSymbol, timeframe);
 							const receivedTimestamp = Date.now();
 							if (
 								!(await writeSubscribeFrame(call, isStreamClosed, {

src/helpers/broker-execution-archive/writer.ts

@@ -41,6 +41,7 @@ const DEFAULT_MAX_QUEUE_SIZE = 10_000;
 const DEFAULT_BATCH_SIZE = 10;
 const DEFAULT_FLUSH_INTERVAL_MS = 1_000;
 const DEFAULT_FORWARDER_TIMEOUT_MS = 3_000;
+const SHED_WARN_INTERVAL_MS = 60_000;
 const DEFAULT_ARCHIVE_FORWARDER_PATH = "/archive";
 const DEFAULT_ARCHIVE_FORWARDER_PORT = 8090;
 
@@ -95,6 +96,7 @@ export class BrokerExecutionArchiver {
 	};
 	private flushTimer: ReturnType<typeof setInterval> | null = null;
 	private flushInFlight: Promise<void> | null = null;
+	private lastShedWarnAtMs = 0;
 	private closed = false;
 	private loggedMissingMarketForwarder = false;
 	private readonly enabled: boolean;
@@ -197,6 +199,17 @@ export class BrokerExecutionArchiver {
 			void this.recordArchiveMetric("cex_archive_rows_shed_total", {
 				table: row.table,
 			});
+			// Shedding means silent archive data loss; the metric alone is invisible
+			// when OTel is disabled, so surface it in logs (rate-limited per row burst).
+			const now = Date.now();
+			if (now - this.lastShedWarnAtMs >= SHED_WARN_INTERVAL_MS) {
+				log.warn("Archive queue full: shedding oldest rows", {
+					shed_total: this.stats.shed,
+					queue_max: this.maxQueueSize,
+					table: row.table,
+				});
+				this.lastShedWarnAtMs = now;
+			}
 		}
 		this.queue.push(row);
 		this.stats.enqueued += 1;

src/helpers/market-data-archive/ohlcv-bar-tracker.ts

@@ -116,21 +116,27 @@ export class OhlcvBarTracker {
 		if (!firstBar || !lastBar) {
 			return [];
 		}
+		const lastOpenTimeMs = this.lastOpenTimeMs;
 
-		if (
-			this.lastOpenTimeMs !== null &&
-			lastBar.openTimeMs < this.lastOpenTimeMs
-		) {
+		if (lastOpenTimeMs !== null && lastBar.openTimeMs < lastOpenTimeMs) {
+			return [];
+		}
+		const barsToProcess =
+			lastOpenTimeMs === null
+				? bars
+				: bars.filter((bar) => bar.openTimeMs >= lastOpenTimeMs);
+		const firstBarToProcess = barsToProcess[0];
+		const lastBarToProcess = barsToProcess[barsToProcess.length - 1];
+		if (!firstBarToProcess || !lastBarToProcess) {
 			return [];
 		}
 
 		const candidates: OhlcvArchiveCandidate[] = [];
 
 		if (
 			this.lastBar !== null &&
-			this.lastOpenTimeMs !== null &&
-			!bars.some((bar) => bar.openTimeMs === this.lastOpenTimeMs) &&
-			this.lastOpenTimeMs < firstBar.openTimeMs
+			lastOpenTimeMs !== null &&
+			lastOpenTimeMs < firstBarToProcess.openTimeMs
 		) {
 			candidates.push({
 				bar: this.lastBar,
@@ -139,8 +145,8 @@ export class OhlcvBarTracker {
 			});
 		}
 
-		for (let index = 0; index < bars.length - 1; index += 1) {
-			const bar = bars[index];
+		for (let index = 0; index < barsToProcess.length - 1; index += 1) {
+			const bar = barsToProcess[index];
 			if (bar) {
 				candidates.push({
 					bar,
@@ -151,13 +157,13 @@ export class OhlcvBarTracker {
 		}
 
 		candidates.push({
-			bar: lastBar,
+			bar: lastBarToProcess,
 			isClosed: false,
 			brokerVersion,
 		});
 
-		this.lastOpenTimeMs = lastBar.openTimeMs;
-		this.lastBar = lastBar;
+		this.lastOpenTimeMs = lastBarToProcess.openTimeMs;
+		this.lastBar = lastBarToProcess;
 		return candidates;
 	}
 }

test/fixtures/ohlcv-collector-fake-exchange.ts

@@ -10,7 +10,7 @@ class ShutdownTestExchange {
 
 	extendExchangeOptions(): void {}
 
-	async fetchOHLCVWs(): Promise<number[][]> {
+	async watchOHLCV(): Promise<number[][]> {
 		this.#fetchCount += 1;
 		const countPath = process.env.OHLCV_TEST_EXCHANGE_COUNT_PATH;
 		if (countPath) {

test/market-data-archive.test.ts

@@ -205,6 +205,15 @@ describe("orderbook sampler", () => {
 });
 
 describe("ohlcv bar tracker", () => {
+	const snapshot = Array.from({ length: 500 }, (_, index) => [
+		1_700_000_000_000 + index * 60_000,
+		1,
+		2,
+		0.5,
+		1.5,
+		10,
+	]);
+
 	test("parseOhlcvBar accepts CCXT tuple shape", () => {
 		expect(parseOhlcvBar([1_000, 1, 2, 0.5, 1.5, 10, 15])).toEqual({
 			openTimeMs: 1_000,
@@ -280,6 +289,63 @@ describe("ohlcv bar tracker", () => {
 			},
 		]);
 	});
+
+	test("preserves all bars in the first batch and leaves the newest open", () => {
+		const tracker = new OhlcvBarTracker();
+
+		const candidates = tracker.process(snapshot, 100);
+
+		expect(candidates).toHaveLength(500);
+		expect(
+			candidates.map(({ bar, isClosed }) => ({
+				openTimeMs: bar.openTimeMs,
+				isClosed,
+			})),
+		).toEqual(
+			snapshot.map(([openTimeMs], index) => ({
+				openTimeMs,
+				isClosed: index < snapshot.length - 1,
+			})),
+		);
+	});
+
+	test("repeated snapshot only re-emits the open bar update", () => {
+		const tracker = new OhlcvBarTracker();
+		tracker.process(snapshot, 100);
+
+		const candidates = tracker.process(snapshot, 200);
+
+		expect(candidates).toHaveLength(1);
+		expect(candidates[0]).toMatchObject({
+			bar: { openTimeMs: snapshot.at(-1)?.[0] },
+			isClosed: false,
+			brokerVersion: 200,
+		});
+	});
+
+	test("overlapping snapshot closes the previous open bar and emits the new open bar", () => {
+		const tracker = new OhlcvBarTracker();
+		tracker.process(snapshot, 100);
+		const previousOpenTimeMs = snapshot.at(-1)?.[0] ?? 0;
+		const nextOpenTimeMs = previousOpenTimeMs + 60_000;
+		const overlappingSnapshot = [
+			...snapshot.slice(1),
+			[nextOpenTimeMs, 1.5, 2.5, 1, 2, 12],
+		];
+
+		const candidates = tracker.process(overlappingSnapshot, 200);
+
+		expect(
+			candidates.map(({ bar, isClosed, brokerVersion }) => ({
+				openTimeMs: bar.openTimeMs,
+				isClosed,
+				brokerVersion,
+			})),
+		).toEqual([
+			{ openTimeMs: previousOpenTimeMs, isClosed: true, brokerVersion: 200 },
+			{ openTimeMs: nextOpenTimeMs, isClosed: false, brokerVersion: 200 },
+		]);
+	});
 });
 
 describe("ohlcv bootstrap limit", () => {

test/subscribe-handler.test.ts

@@ -172,7 +172,7 @@ type ThrowingSubscriptionMethod =
 	| "watchOrderBook"
 	| "watchTrades"
 	| "watchTicker"
-	| "fetchOHLCVWs"
+	| "watchOHLCV"
 	| "watchBalance"
 	| "watchOrders";
 
@@ -267,7 +267,7 @@ describe("subscribe handler", () => {
 		},
 		{
 			type: SubscriptionType.OHLCV,
-			method: "fetchOHLCVWs",
+			method: "watchOHLCV",
 			errorMessage: "ohlcv boom",
 			expectedError: "Failed to fetch OHLCV: ohlcv boom",
 		},
@@ -456,7 +456,7 @@ describe("subscribe handler", () => {
 		try {
 			const controlledWatch = createControlledWatch();
 			const exchange = {
-				fetchOHLCVWs: controlledWatch.watch,
+				watchOHLCV: controlledWatch.watch,
 			} as unknown as Exchange;
 			const archiver = BrokerExecutionArchiver.create({
 				forwarderUrl: server.url,