Add durable broker stream health archive

README.md

@@ -737,6 +737,12 @@ Every successful `CreateOrder` response, successful `GetOrderDetails` response,
 
 Use metrics for aggregations and alerts. For the durable execution audit trail, the broker archives every order lifecycle event to `broker_execution.order_events` (and pre-order top-of-book to `broker_execution.market_metadata_snapshots`) through the **archive forwarder** — the same HTTP `/archive` → ClickHouse path used for `market_data.*`. Set `CEX_BROKER_ARCHIVE_ENABLED=true`, an explicit HTTP(S) `CEX_BROKER_ARCHIVE_FORWARDER_URL`, and a writable durable JSONL path in `CEX_BROKER_ARCHIVE_DEAD_LETTER_PATH`; startup fails if either required sink configuration is missing or invalid. In production, that path must be on persistent writable storage or a mounted volume rather than the container's ephemeral filesystem. Queue shedding and rows that remain undeliverable during shutdown are written to that loss journal with their original `{table,row}` payload before being discarded. Setting `CEX_BROKER_ARCHIVE_OTEL_LOGS_ENABLED=true` additionally mirrors execution rows to OTel logs for observability, but OTel is never the archive sink of record. Analysts join Maker action rows to `broker_execution.order_events` using `maker_action_id`, `idempotency_id`, `client_order_id`, or the exchange `order_id`, then compare Maker propAMM execution price against `average_execution_price` and fees. Failed CreateOrder rows keep bounded exchange error detail in `error_message`; their telemetry-shaped `payload_json`, metrics, and ordinary telemetry logs remain redacted. The broker does not emit raw exchange payloads, API keys, secrets, or credentials in telemetry fields.
 
+### User-stream health archive contract
+
+`broker_stream_health.snapshots` is a dedicated, source-authoritative archive path for the configured-account user-stream registry. The publisher must emit every active stream for one registry revision in one batch; retired entries may remain as explicit lifecycle evidence. The forwarder verifies total and active counts, unique normalized stream identities, and one recomputed `batch_id` before making one ClickHouse write. It records connection state, authentication and received-event watermarks, counters, and explicit failure kinds; a quiet event-driven stream is represented by `last_received_at = NULL`, not by missing health data. Failure reasons are bounded diagnostics and are redacted before storage; state, failure kind, and traffic mode remain the machine-readable fields.
+
+The archive forwarder must be a single active deployment for this table. Replays are accepted only when the whole batch is already present with the same canonical hashes. A partial batch, a same-id different payload, or more than one existing hash for an ID is persisted to `broker_stream_health.replay_conflicts` and the candidate batch is rejected. Readers must deduplicate exact race duplicates by `(snapshot_id, payload_sha256)` and treat more than one hash for one `snapshot_id` as an integrity incident rather than selecting a winner. Neither health table has a TTL.
+
 ### Telemetry Test Harness
 
 Order telemetry tests use `test/order-telemetry-fixtures.ts` to run the real gRPC server with mocked CCXT exchanges. The fixture can simulate create-order responses, order-detail responses, partial fills, rejected orders, failed create-order calls, and fee/no-fee exchange payloads without live credentials.

schema/clickhouse/broker_stream_health.sql

@@ -0,0 +1,78 @@
+-- Durable, source-authoritative CEX user-stream health snapshots.
+-- NO TTL: quiet, disconnected, and failed states are replay evidence.
+
+CREATE DATABASE IF NOT EXISTS broker_stream_health;
+
+CREATE TABLE IF NOT EXISTS broker_stream_health.snapshots
+(
+    schema_version LowCardinality(String),
+    source LowCardinality(String),
+    deployment_id LowCardinality(String),
+    producer_id LowCardinality(String),
+    producer_epoch UInt64,
+    run_id UUID,
+    batch_id FixedString(64),
+    batch_sequence UInt64,
+    batch_snapshot_count UInt32,
+    batch_active_stream_count UInt32,
+    registry_revision FixedString(64),
+    registry_status Enum8('active' = 1, 'retired' = 2),
+    retired_at Nullable(DateTime64(3, 'UTC')),
+    snapshot_id FixedString(64),
+    stream_key String,
+    exchange LowCardinality(String),
+    account_selector LowCardinality(String),
+    account_role LowCardinality(Nullable(String)),
+    stream_kind LowCardinality(String),
+    account_scope LowCardinality(String),
+    sequence UInt64,
+    state Enum8('connecting' = 1, 'connected' = 2, 'disconnected' = 3, 'error' = 4),
+    state_changed_at DateTime64(3, 'UTC'),
+    last_connected_at Nullable(DateTime64(3, 'UTC')),
+    last_authenticated_at Nullable(DateTime64(3, 'UTC')),
+    last_received_at Nullable(DateTime64(3, 'UTC')),
+    heartbeat_at DateTime64(3, 'UTC'),
+    connect_attempt_count UInt64,
+    reconnect_count UInt64,
+    error_count UInt64,
+    last_failure_kind Enum8(
+        'none' = 0,
+        'auth_failed' = 1,
+        'transport_error' = 2,
+        'remote_closed' = 3,
+        'protocol_error' = 4,
+        'backpressure' = 5,
+        'unsupported_connector' = 6,
+        'shutdown' = 7
+    ),
+    last_failure_reason String,
+    traffic_mode Enum8('event_driven' = 1, 'continuous' = 2, 'unknown' = 3),
+    source_watermark Nullable(String),
+    payload_sha256 FixedString(64),
+    payload_json String,
+    ingested_at DateTime64(3, 'UTC') DEFAULT now64(3)
+)
+ENGINE = MergeTree
+PARTITION BY toYYYYMM(heartbeat_at)
+ORDER BY (producer_id, producer_epoch, run_id, stream_key, sequence, snapshot_id);
+
+-- A collision is never reconciled by a reader. The forwarder records the
+-- evidence separately and fails the whole candidate batch closed.
+CREATE TABLE IF NOT EXISTS broker_stream_health.replay_conflicts
+(
+    detected_at DateTime64(3, 'UTC') DEFAULT now64(3),
+    batch_id FixedString(64),
+    snapshot_id FixedString(64),
+    conflict_kind Enum8(
+        'payload_mismatch' = 1,
+        'partial_batch' = 2,
+        'multiple_existing_hashes' = 3
+    ),
+    existing_payload_sha256 String,
+    incoming_payload_sha256 FixedString(64),
+    existing_payload_json String,
+    incoming_payload_json String
+)
+ENGINE = MergeTree
+PARTITION BY toYYYYMM(detected_at)
+ORDER BY (snapshot_id, detected_at);

services/archive-forwarder/index.ts

@@ -4,6 +4,7 @@ import { createClickHouseInserter } from "./insert";
 import { evaluateForwarderHealth, pingClickHouse } from "./health";
 import { handleArchiveRequest } from "./request";
 import { ensureArchiveSchema } from "./schema";
+import { createClickHouseStreamHealthReplayStore } from "./stream-health-contract";
 import { createArchiveForwarderTelemetry } from "./telemetry";
 import {
 	StrategyArchiveSpool,
@@ -25,6 +26,7 @@ const clickhouse = createClient({
 	clickhouse_settings: { date_time_input_format: "best_effort" },
 });
 const inserter = createClickHouseInserter(clickhouse);
+const streamHealthStore = createClickHouseStreamHealthReplayStore(clickhouse);
 const telemetry = createArchiveForwarderTelemetry();
 let spool: StrategyArchiveSpool | undefined;
 let worker: StrategySpoolWorker | undefined;
@@ -46,7 +48,7 @@ async function ensureSchemaAndStartDrainage(): Promise<void> {
 		schemaReady = true;
 		worker?.start();
 		console.log(
-			"ClickHouse archive schema ensured (market_data, broker_execution, broker_account, strategy_data)",
+			"ClickHouse archive schema ensured (market_data, broker_execution, broker_account, broker_stream_health, strategy_data)",
 		);
 	} catch (error) {
 		schemaReady = false;
@@ -102,6 +104,7 @@ const server = Bun.serve({
 				authToken: config.authToken,
 				inserter,
 				spool,
+				streamHealthStore,
 				telemetry,
 			});
 		}

services/archive-forwarder/request.ts

@@ -11,6 +11,12 @@ import {
 	classifyStrategyArchiveBatch,
 	validateStrategyArchiveBatch,
 } from "./strategy-contract";
+import {
+	classifyStreamHealthArchiveBatch,
+	insertStreamHealthArchiveBatch,
+	type StreamHealthReplayStore,
+	validateStreamHealthArchiveBatch,
+} from "./stream-health-contract";
 import {
 	StrategySpoolQuotaError,
 	StrategySpoolUnavailableError,
@@ -22,6 +28,7 @@ export type ArchiveRequestDependencies = {
 	authToken?: string;
 	inserter: RowInserter;
 	spool?: Pick<StrategyArchiveSpool, "admit">;
+	streamHealthStore?: StreamHealthReplayStore;
 	telemetry: ArchiveForwarderTelemetry;
 };
 
@@ -57,6 +64,24 @@ export async function handleArchiveRequest(
 		return Response.json({ error: "Invalid JSON body" }, { status: 400 });
 	}
 
+	const streamHealthClassification = classifyStreamHealthArchiveBatch(body);
+	if (
+		streamHealthClassification === "invalid_stream_health_mix" ||
+		streamHealthClassification === "invalid_stream_health_source"
+	) {
+		return Response.json(
+			{ error: "Invalid stream health archive source or table mix" },
+			{ status: 400 },
+		);
+	}
+	const streamHealthValidation =
+		streamHealthClassification === "stream_health"
+			? validateStreamHealthArchiveBatch(body)
+			: undefined;
+	if (streamHealthValidation && !streamHealthValidation.ok) {
+		return Response.json({ error: streamHealthValidation.error }, { status: 400 });
+	}
+
 	const strategyClassification = classifyStrategyArchiveBatch(body);
 	if (
 		strategyClassification === "invalid_strategy_mix" ||
@@ -125,6 +150,31 @@ export async function handleArchiveRequest(
 		);
 	}
 
+	if (streamHealthValidation?.ok) {
+		if (!dependencies.streamHealthStore) {
+			return Response.json(
+				{ error: "Stream health replay store unavailable" },
+				{ status: 503 },
+			);
+		}
+		const result = await insertStreamHealthArchiveBatch(
+			dependencies.inserter,
+			dependencies.streamHealthStore,
+			streamHealthValidation,
+		);
+		if (!result.ok) {
+			return Response.json({ error: result.error }, { status: result.status });
+		}
+		if (result.inserted > 0) {
+			dependencies.telemetry.recordRowsInserted(
+				"broker_stream_health.snapshots",
+				result.inserted,
+			);
+			dependencies.telemetry.recordSuccessfulFlush();
+		}
+		return Response.json(result);
+	}
+
 	if (strategyClassification === "strategy") {
 		if (!dependencies.spool) {
 			dependencies.telemetry.recordStrategyAdmissionRejected("spool_unavailable");

services/archive-forwarder/schema.ts

@@ -57,6 +57,7 @@ const ARCHIVE_SCHEMA_FILES = [
 	"market_data.sql",
 	"broker_execution.sql",
 	"broker_account.sql",
+	"broker_stream_health.sql",
 	"strategy_data.sql",
 ] as const;