Merge pull request #117 from usherlabs/fix/market-capture-fail-closed

fix(market-data): refuse to start when production capture identity is incomplete

Author: outerlook

src/helpers/market-data-archive/capture-context.ts

@@ -98,6 +98,33 @@ export function resolveMarketCaptureArchiveState(input: {
 	return { enabled: true };
 }
 
+/**
+ * Throws when archival was requested but its capture identity is incomplete.
+ *
+ * `archive_disabled` and `market_archive_disabled` are deliberate opt-outs and
+ * must keep the broker available — that is the whole point of resolving to a
+ * typed state instead of throwing. Every other disabled reason means archival
+ * was asked for and cannot be honoured, which is a misconfiguration: a broker
+ * started in that state serves RPC and passes health while writing no
+ * market-data row at all, and the only trace is one line at boot.
+ */
+export function assertMarketCaptureArchiveStartable(
+	state: MarketCaptureArchiveState,
+): void {
+	if (
+		state.enabled ||
+		state.reason === "archive_disabled" ||
+		state.reason === "market_archive_disabled"
+	) {
+		return;
+	}
+	throw new Error(
+		`Refusing to start: canonical market-data archival was requested but its capture identity is incomplete (${state.reason}). ` +
+			"Set CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT, CEX_BROKER_DEPLOYMENT_ID and CEX_BROKER_CAPTURE_BUNDLE_ID, " +
+			"or disable archival explicitly via CEX_BROKER_MARKET_ARCHIVE_ENABLED.",
+	);
+}
+
 export function validateExternalFallbackContext(input: {
 	configuredExchange: string;
 	configuredSymbol: string;

src/helpers/market-data-archive/index.ts

@@ -15,6 +15,7 @@ export {
 	createOrderbookTobSampler,
 } from "./capture";
 export {
+	assertMarketCaptureArchiveStartable,
 	captureEnvironmentFromEnv,
 	createMarketCaptureContext,
 	type MarketCaptureArchiveDisabledReason,

src/index.ts

@@ -19,7 +19,10 @@ import {
 import { DepositArchivePoller } from "./helpers/deposit-archive-poller";
 import { FillArchivePoller } from "./helpers/fill-archive-poller";
 import { log } from "./helpers/logger";
-import { resolveMarketCaptureArchiveState } from "./helpers/market-data-archive/capture-context";
+import {
+	assertMarketCaptureArchiveStartable,
+	resolveMarketCaptureArchiveState,
+} from "./helpers/market-data-archive/capture-context";
 import { isMarketArchiveEnabled } from "./helpers/market-data-archive/orderbook-sampler";
 import { OrderActivityTracker } from "./helpers/order-activity-tracker";
 import {
@@ -336,16 +339,7 @@ export default class CEXBroker {
 			deploymentId: this.brokerArchiver?.getDeploymentId(),
 			captureBundleId: process.env.CEX_BROKER_CAPTURE_BUNDLE_ID,
 		});
-		if (
-			!marketArchiveState.enabled &&
-			marketArchiveState.reason !== "archive_disabled" &&
-			marketArchiveState.reason !== "market_archive_disabled"
-		) {
-			log.warn(
-				"Canonical market-data archival is disabled; broker RPC service remains available",
-				{ reason: marketArchiveState.reason },
-			);
-		}
+		assertMarketCaptureArchiveStartable(marketArchiveState);
 		if (this.server) {
 			await this.server.forceShutdown();
 		}

test/canonical-market-data-contract.test.ts

@@ -4,6 +4,7 @@ import {
 	OrderBookValidationError,
 } from "../src/helpers/market-data-archive/canonical-orderbook";
 import {
+	assertMarketCaptureArchiveStartable,
 	createMarketCaptureContext,
 	resolveMarketCaptureArchiveState,
 	validateExternalFallbackContext,
@@ -221,6 +222,36 @@ describe("canonical market capture contract", () => {
 		).toEqual({ enabled: true });
 	});
 
+	test("startup fails closed when archival was requested but capture identity is incomplete", () => {
+		// The resolver stays non-throwing (above); the START decision is what must
+		// fail closed. Without this, a production broker missing its deployment or
+		// bundle id runs healthy, serves RPC, and writes no market-data row at all —
+		// one warn line at boot and no counter, metric or alarm anywhere.
+		for (const reason of [
+			"invalid_capture_environment",
+			"missing_deployment_id",
+			"missing_capture_bundle_id",
+		] as const) {
+			expect(() =>
+				assertMarketCaptureArchiveStartable({ enabled: false, reason }),
+			).toThrow(reason);
+		}
+
+		// Deliberate opt-outs must keep the broker available — that is the whole
+		// point of resolving to a typed state rather than throwing outright.
+		for (const reason of [
+			"archive_disabled",
+			"market_archive_disabled",
+		] as const) {
+			expect(() =>
+				assertMarketCaptureArchiveStartable({ enabled: false, reason }),
+			).not.toThrow();
+		}
+		expect(() =>
+			assertMarketCaptureArchiveStartable({ enabled: true }),
+		).not.toThrow();
+	});
+
 	test("external fallback rows cannot cross venue or omit their reason", () => {
 		expect(() =>
 			validateExternalFallbackContext({

test/production-market-capture-startup.test.ts

@@ -128,11 +128,16 @@ test("production broker starts its full RPC service without archive configuratio
 	}
 });
 
-test("incomplete production market provenance does not gate the full RPC service", async () => {
+test("incomplete production market provenance refuses to start", async () => {
+	// A production broker that asked for archival but cannot identify its capture
+	// must NOT come up. Starting is the dangerous outcome, not a degraded one: it
+	// serves RPC and passes health while writing no market-data row, leaving one
+	// warn line at boot as the only trace. Availability is still preserved for
+	// DELIBERATE opt-outs (archive disabled / market archive disabled), which the
+	// surrounding tests cover.
 	const original = captureEnvironment();
 	const deadLetterPath = `/tmp/cex-broker-production-ineligible-${crypto.randomUUID()}.jsonl`;
 	let broker: CEXBroker | undefined;
-	let client: FullBrokerClient | undefined;
 	try {
 		const port = await reservePort();
 		process.env.CEX_BROKER_MARKET_CAPTURE_ENVIRONMENT = "production";
@@ -145,15 +150,8 @@ test("incomplete production market provenance does not gate the full RPC service
 
 		broker = new CEXBroker({}, policy);
 		broker.port = port;
-		await broker.run();
-		client = new grpcObject.cex_broker.cex_service(
-			`127.0.0.1:${port}`,
-			grpc.credentials.createInsecure(),
-		);
-		await waitForReady(client);
-		await expectFullRpcService(client);
+		await expect(broker.run()).rejects.toThrow("missing_deployment_id");
 	} finally {
-		client?.close();
 		await broker?.stop();
 		restoreEnvironment(original);
 		if (await Bun.file(deadLetterPath).exists()) {