fix(build): use node 24 for drift sdk

add ws redis switchover with hysteresis

Dockerfile

@@ -1,4 +1,4 @@
-FROM public.ecr.aws/docker/library/node:22 AS builder
+FROM public.ecr.aws/docker/library/node:24 AS builder
 RUN npm install -g typescript@5.4.5
 
 WORKDIR /app
@@ -14,7 +14,7 @@ RUN yarn && yarn build
 WORKDIR /app
 RUN yarn && yarn build
 
-FROM public.ecr.aws/docker/library/node:22-alpine
+FROM public.ecr.aws/docker/library/node:24-alpine
 RUN apk add python3 make g++ --virtual .build &&\
     npm install -C /lib bigint-buffer @triton-one/yellowstone-grpc@1.3.0 helius-laserstream@0.1.8 rpc-websockets@7.5.1 &&\
     apk del .build

README.md

@@ -82,6 +82,12 @@ To properly configure the DLOB server, set the following environment variables i
 | `ELASTICACHE_HOST`            | (for websocket server) Redis host endpoint.                     | `localhost`                         |
 | `ELASTICACHE_PORT`            | (for websocket server) Redis port.                              | `6379`                              |
 | `REDIS_CLIENT`                | (for websocket server) Redis client type (DLOB/DLOB_HELIUS).    | `DLOB`                              |
+| `ENABLE_PERSISTENT_STORE`     | Persist canonical latest DLOB snapshots to Postgres.            | `true`                              |
+| `POSTGRES_URL`                | Connection string for canonical DLOB store.                     | `postgres://user:pass@host:5432/db` |
+| `DLOB_SOURCE`                 | Source key stored with canonical DLOB rows.                     | `mevnode`                           |
+| `PRICE_PRECISION`             | Price scale used by projection views.                           | `1000000`                           |
+| `BASE_PRECISION`              | Base size scale used by projection views.                       | `1000000000`                        |
+| `PERSISTENT_STATS_DEPTH`      | Depth used by SQL projection views for stats.                   | `10`                                |
 | `WS_PORT`                     | (for websocket server) The port to run the websocket server on. | `3000`                              |
 
 Note: multiple Redis hosts can be provided by providing a comma separated string.
@@ -131,6 +137,30 @@ When you're done, stop the redis cluster:
 bash redisCluster.sh stop
 ```
 
+## Persistent canonical store
+
+The publisher can now persist canonical latest DLOB snapshots directly to Postgres while still using Redis as the hot path.
+
+- Redis remains the runtime cache and pub/sub layer.
+- Postgres stores canonical latest snapshots in:
+  - `dlob_orderbook_l2_latest`
+  - `dlob_orderbook_l3_latest`
+  - `dlob_best_makers_latest`
+- Postgres exposes projection views for Hasura and frontend adaptation:
+  - `dlob_l2_latest_projection`
+  - `dlob_l3_latest_projection`
+  - `dlob_best_makers_latest_projection`
+  - `dlob_stats_latest_projection`
+- This aligns the DLOB stack with a `Redis hot state + Postgres durable canonical store + projection views` model.
+
+To initialize the persistent tables without connecting to chain inputs yet:
+
+```bash
+ENABLE_PERSISTENT_STORE=true \
+POSTGRES_URL=postgres://user:pass@host:5432/db \
+yarn dlob-storage:init
+```
+
 # Run the example client
 
 Documentation for connecting to the dlob server are available [here](https://drift-labs.github.io/v2-teacher/?python#orderbook-trades-dlob-server)

package.json

@@ -32,6 +32,7 @@
 		"express": "^4.18.2",
 		"ioredis": "^5.4.1",
 		"morgan": "^1.10.0",
+		"pg": "^8.13.1",
 		"redis": "^4.6.10",
 		"response-time": "^2.3.2",
 		"rxjs": "^7.8.1",
@@ -70,6 +71,7 @@
 		"dev": "ts-node src/index.ts",
 		"server-lite": "ts-node src/serverLite.ts",
 		"dlob-publish": "ts-node src/publishers/dlobPublisher.ts",
+		"dlob-storage:init": "ts-node src/scripts/initPersistentStore.ts",
 		"trades-publish": "ts-node src/publishers/tradesPublisher.ts",
 		"fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts",
 		"pnl-publish": "ts-node src/scripts/publishUnsettledPnlUsers.ts",

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -32,6 +32,7 @@ import { OffloadQueue } from '../utils/offload';
 import { setHealthStatus, HEALTH_STATUS } from '../core/healthCheck';
 import { CounterValue } from '../core/metricsV2';
 import { COMMON_MATH } from '@drift/common';
+import { DlobSnapshotStore } from '../storage/dlobSnapshotStore';
 
 export type wsMarketArgs = {
 	marketIndex: number;
@@ -70,6 +71,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
 	redisClient: RedisClient;
 	indicativeQuotesRedisClient?: RedisClient;
+	persistentStore?: DlobSnapshotStore;
 	public killSwitchSlotDiffThreshold: number;
 	public offloadQueue?: ReturnType<typeof OffloadQueue>;
 	public enableOffload: boolean;
@@ -79,6 +81,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			env: DriftEnv;
 			redisClient: RedisClient;
 			indicativeQuotesRedisClient?: RedisClient;
+			persistentStore?: DlobSnapshotStore;
 			enableOffloadQueue?: boolean;
 			offloadQueueCounter?: CounterValue;
 			perpMarketInfos: wsMarketInfo[];
@@ -90,6 +93,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		super(config);
 		this.redisClient = config.redisClient;
 		this.indicativeQuotesRedisClient = config.indicativeQuotesRedisClient;
+		this.persistentStore = config.persistentStore;
 
 		this.killSwitchSlotDiffThreshold =
 			config.killSwitchSlotDiffThreshold || 200;
@@ -470,6 +474,19 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Formatted_depth100
 		);
 
+		if (this.persistentStore) {
+			void this.persistentStore
+				.writeL2Snapshot(l2Formatted, {
+					indicative: Boolean(this.indicativeQuotesRedisClient),
+				})
+				.catch((error) => {
+				logger.error(
+					`Failed to persist canonical L2 snapshot for ${marketName}`,
+					error
+				);
+				});
+		}
+
 		publishGroupings(
 			l2Formatted,
 			marketArgs,
@@ -508,6 +525,23 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				`last_update_orderbook_best_makers_${marketType}_${marketArgs.marketIndex}`,
 				{ bids, asks, slot }
 			);
+			if (this.persistentStore) {
+				void this.persistentStore
+					.writeBestMakersSnapshot({
+						marketName: marketName?.toUpperCase() ?? String(marketArgs.marketIndex),
+						marketType: marketType?.toLowerCase(),
+						marketIndex: marketArgs.marketIndex,
+						slot: slot ?? null,
+						bids,
+						asks,
+					})
+					.catch((error) => {
+						logger.error(
+							`Failed to persist canonical best makers for ${marketName}`,
+							error
+						);
+					});
+			}
 		}
 	}
 
@@ -571,5 +605,18 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			}`,
 			l3
 		);
+
+		if (this.persistentStore) {
+			void this.persistentStore
+				.writeL3Snapshot(l3, {
+					indicative: Boolean(this.indicativeQuotesRedisClient),
+				})
+				.catch((error) => {
+					logger.error(
+						`Failed to persist canonical L3 snapshot for ${marketName}`,
+						error
+					);
+				});
+		}
 	}
 }

src/publishers/dlobPublisher.ts

@@ -47,6 +47,7 @@ import {
 	FillQualityAnalyticsRepository,
 	TakerFillVsOracleBpsRedisResult,
 } from '../athena/repositories/fillQualityAnalytics';
+import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore';
 
 setGlobalDispatcher(
 	new Agent({
@@ -383,10 +384,20 @@ const main = async () => {
 		prefix: RedisClientPrefix[REDIS_CLIENT],
 	});
 	await redisClient.connect();
+	const persistentStore = createPostgresCanonicalStoreFromEnv();
+	await persistentStore.init();
 
 	const indicativeRedisClient = new RedisClient({});
 	await indicativeRedisClient.connect();
 
+	logger.info(
+		`Persistent store: ${
+			process.env.ENABLE_PERSISTENT_STORE?.toLowerCase() === 'true'
+				? 'postgres canonical + projections'
+				: 'disabled'
+		}`
+	);
+
 	const connection = new Connection(endpoint, {
 		wsEndpoint: wsEndpoint,
 		commitment: stateCommitment,
@@ -582,6 +593,7 @@ const main = async () => {
 		spotMarketSubscribers: MARKET_SUBSCRIBERS,
 		perpMarketInfos,
 		spotMarketInfos,
+		persistentStore,
 		killSwitchSlotDiffThreshold: KILLSWITCH_SLOT_DIFF_THRESHOLD,
 		protectedMakerView: false,
 	});

src/scripts/initPersistentStore.ts

@@ -0,0 +1,16 @@
+import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore';
+import { logger } from '../utils/logger';
+
+require('dotenv').config();
+
+async function main() {
+	const store = createPostgresCanonicalStoreFromEnv();
+	await store.init();
+	logger.info('Persistent DLOB store initialized');
+	await store.close();
+}
+
+main().catch((error) => {
+	logger.error('Failed to initialize persistent DLOB store', error);
+	process.exit(1);
+});

src/storage/dlobSnapshotStore.ts

@@ -0,0 +1,45 @@
+export interface DlobSnapshotStore {
+	init(): Promise<void>;
+	writeL2Snapshot(
+		l2Snapshot: Record<string, any>,
+		options: { indicative: boolean }
+	): Promise<void>;
+	writeL3Snapshot(
+		l3Snapshot: Record<string, any>,
+		options: { indicative: boolean }
+	): Promise<void>;
+	writeBestMakersSnapshot(snapshot: {
+		marketName: string;
+		marketType: string;
+		marketIndex: number;
+		slot: number | null;
+		bids: string[];
+		asks: string[];
+	}): Promise<void>;
+	close(): Promise<void>;
+}
+
+export class NoopDlobSnapshotStore implements DlobSnapshotStore {
+	async init(): Promise<void> {}
+
+	async writeL2Snapshot(
+		_l2Snapshot: Record<string, any>,
+		_options: { indicative: boolean }
+	): Promise<void> {}
+
+	async writeL3Snapshot(
+		_l3Snapshot: Record<string, any>,
+		_options: { indicative: boolean }
+	): Promise<void> {}
+
+	async writeBestMakersSnapshot(_snapshot: {
+		marketName: string;
+		marketType: string;
+		marketIndex: number;
+		slot: number | null;
+		bids: string[];
+		asks: string[];
+	}): Promise<void> {}
+
+	async close(): Promise<void> {}
+}

src/storage/postgresCanonicalStore.ts

@@ -0,0 +1,560 @@
+import { Pool } from 'pg';
+
+import { logger } from '../utils/logger';
+import {
+	DlobSnapshotStore,
+	NoopDlobSnapshotStore,
+} from './dlobSnapshotStore';
+
+type PostgresCanonicalStoreConfig = {
+	connectionString?: string;
+	host?: string;
+	port?: number;
+	user?: string;
+	password?: string;
+	database?: string;
+	source: string;
+	pricePrecision: number;
+	basePrecision: number;
+	statsDepth: number;
+};
+
+function envBool(name: string, fallback = false): boolean {
+	const raw = process.env[name];
+	if (raw == null) return fallback;
+	const value = String(raw).trim().toLowerCase();
+	if (['1', 'true', 'yes', 'y', 'on'].includes(value)) return true;
+	if (['0', 'false', 'no', 'n', 'off'].includes(value)) return false;
+	return fallback;
+}
+
+function envInt(name: string, fallback: number): number {
+	const raw = process.env[name];
+	if (raw == null) return fallback;
+	const parsed = Number.parseInt(String(raw), 10);
+	return Number.isFinite(parsed) ? parsed : fallback;
+}
+
+function toIntOrNull(value: unknown): number | null {
+	if (value == null) return null;
+	const parsed =
+		typeof value === 'number' ? value : Number.parseInt(String(value), 10);
+	return Number.isFinite(parsed) ? Math.trunc(parsed) : null;
+}
+
+function toJson(value: unknown): string {
+	return JSON.stringify(value ?? null);
+}
+
+export class PostgresCanonicalStore implements DlobSnapshotStore {
+	private readonly pool: Pool;
+	private writeChain: Promise<void> = Promise.resolve();
+
+	constructor(private readonly config: PostgresCanonicalStoreConfig) {
+		this.pool = new Pool(
+			config.connectionString
+				? { connectionString: config.connectionString }
+				: {
+						host: config.host,
+						port: config.port,
+						user: config.user,
+						password: config.password,
+						database: config.database,
+				  }
+		);
+	}
+
+	async init(): Promise<void> {
+		await this.pool.query(`
+			CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l2_latest (
+				source TEXT NOT NULL,
+				market_name TEXT NOT NULL,
+				market_type TEXT NOT NULL,
+				market_index INTEGER,
+				is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
+				ts BIGINT,
+				slot BIGINT,
+				mark_price_raw TEXT,
+				oracle_price_raw TEXT,
+				best_bid_price_raw TEXT,
+				best_ask_price_raw TEXT,
+				spread_pct_raw TEXT,
+				spread_quote_raw TEXT,
+				oracle_data JSONB,
+				bids JSONB,
+				asks JSONB,
+				payload JSONB NOT NULL,
+				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+				PRIMARY KEY (source, market_name, is_indicative)
+			);
+		`);
+
+		await this.pool.query(`
+			CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l3_latest (
+				source TEXT NOT NULL,
+				market_name TEXT NOT NULL,
+				market_type TEXT NOT NULL,
+				market_index INTEGER,
+				is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
+				ts BIGINT,
+				slot BIGINT,
+				oracle_data JSONB,
+				bids JSONB,
+				asks JSONB,
+				payload JSONB NOT NULL,
+				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+				PRIMARY KEY (source, market_name, is_indicative)
+			);
+		`);
+
+		await this.pool.query(`
+			CREATE TABLE IF NOT EXISTS public.dlob_best_makers_latest (
+				source TEXT NOT NULL,
+				market_name TEXT NOT NULL,
+				market_type TEXT NOT NULL,
+				market_index INTEGER,
+				slot BIGINT,
+				bids JSONB,
+				asks JSONB,
+				payload JSONB NOT NULL,
+				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+				PRIMARY KEY (source, market_name)
+			);
+		`);
+
+		await this.pool.query(`
+			CREATE INDEX IF NOT EXISTS dlob_orderbook_l2_latest_updated_at_idx
+			ON public.dlob_orderbook_l2_latest (updated_at DESC);
+		`);
+		await this.pool.query(`
+			CREATE INDEX IF NOT EXISTS dlob_orderbook_l3_latest_updated_at_idx
+			ON public.dlob_orderbook_l3_latest (updated_at DESC);
+		`);
+		await this.pool.query(`
+			CREATE INDEX IF NOT EXISTS dlob_best_makers_latest_updated_at_idx
+			ON public.dlob_best_makers_latest (updated_at DESC);
+		`);
+
+		await this.pool.query(`
+			CREATE OR REPLACE VIEW public.dlob_l2_latest_projection AS
+			SELECT
+				source,
+				market_name,
+				market_type,
+				market_index,
+				ts,
+				slot,
+				CASE
+					WHEN NULLIF(mark_price_raw, '') IS NULL THEN NULL
+					ELSE NULLIF(mark_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+				END AS mark_price,
+				CASE
+					WHEN NULLIF(oracle_price_raw, '') IS NULL THEN NULL
+					ELSE NULLIF(oracle_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+				END AS oracle_price,
+				CASE
+					WHEN NULLIF(best_bid_price_raw, '') IS NULL THEN NULL
+					ELSE NULLIF(best_bid_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+				END AS best_bid_price,
+				CASE
+					WHEN NULLIF(best_ask_price_raw, '') IS NULL THEN NULL
+					ELSE NULLIF(best_ask_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+				END AS best_ask_price,
+				bids,
+				asks,
+				payload AS raw,
+				updated_at
+			FROM public.dlob_orderbook_l2_latest
+			WHERE is_indicative = FALSE;
+		`);
+
+		await this.pool.query(`
+			CREATE OR REPLACE VIEW public.dlob_l3_latest_projection AS
+			SELECT
+				source,
+				market_name,
+				market_type,
+				market_index,
+				ts,
+				slot,
+				bids,
+				asks,
+				payload AS raw,
+				updated_at
+			FROM public.dlob_orderbook_l3_latest
+			WHERE is_indicative = FALSE;
+		`);
+
+		await this.pool.query(`
+			CREATE OR REPLACE VIEW public.dlob_best_makers_latest_projection AS
+			SELECT
+				source,
+				market_name,
+				market_type,
+				market_index,
+				slot,
+				bids,
+				asks,
+				payload AS raw,
+				updated_at
+			FROM public.dlob_best_makers_latest;
+		`);
+
+		await this.pool.query(`
+			CREATE OR REPLACE VIEW public.dlob_stats_latest_projection AS
+			WITH base AS (
+				SELECT
+					source,
+					market_name,
+					market_type,
+					market_index,
+					ts,
+					slot,
+					CASE
+						WHEN NULLIF(mark_price_raw, '') IS NULL THEN NULL
+						ELSE NULLIF(mark_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+					END AS mark_price,
+					CASE
+						WHEN NULLIF(oracle_price_raw, '') IS NULL THEN NULL
+						ELSE NULLIF(oracle_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+					END AS oracle_price,
+					CASE
+						WHEN NULLIF(best_bid_price_raw, '') IS NULL THEN NULL
+						ELSE NULLIF(best_bid_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+					END AS best_bid_price,
+					CASE
+						WHEN NULLIF(best_ask_price_raw, '') IS NULL THEN NULL
+						ELSE NULLIF(best_ask_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
+					END AS best_ask_price,
+					bids,
+					asks,
+					payload,
+					updated_at
+				FROM public.dlob_orderbook_l2_latest
+				WHERE is_indicative = FALSE
+			),
+			bid_levels AS (
+				SELECT
+					b.source,
+					b.market_name,
+					COALESCE(
+						SUM((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric),
+						0::numeric
+					) AS depth_bid_base,
+					COALESCE(
+						SUM(
+							((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric) *
+							((NULLIF(level->>'price', '')::numeric) / ${this.config.pricePrecision}::numeric)
+						),
+						0::numeric
+					) AS depth_bid_usd
+				FROM base b
+				LEFT JOIN LATERAL (
+					SELECT level
+					FROM jsonb_array_elements(COALESCE(b.bids, '[]'::jsonb)) WITH ORDINALITY AS e(level, ord)
+					WHERE ord <= ${this.config.statsDepth}
+				) levels ON TRUE
+				GROUP BY b.source, b.market_name
+			),
+			ask_levels AS (
+				SELECT
+					b.source,
+					b.market_name,
+					COALESCE(
+						SUM((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric),
+						0::numeric
+					) AS depth_ask_base,
+					COALESCE(
+						SUM(
+							((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric) *
+							((NULLIF(level->>'price', '')::numeric) / ${this.config.pricePrecision}::numeric)
+						),
+						0::numeric
+					) AS depth_ask_usd
+				FROM base b
+				LEFT JOIN LATERAL (
+					SELECT level
+					FROM jsonb_array_elements(COALESCE(b.asks, '[]'::jsonb)) WITH ORDINALITY AS e(level, ord)
+					WHERE ord <= ${this.config.statsDepth}
+				) levels ON TRUE
+				GROUP BY b.source, b.market_name
+			)
+			SELECT
+				b.source,
+				b.market_name,
+				b.market_type,
+				b.market_index,
+				b.ts,
+				b.slot,
+				b.mark_price,
+				b.oracle_price,
+				b.best_bid_price,
+				b.best_ask_price,
+				CASE
+					WHEN b.best_bid_price IS NOT NULL AND b.best_ask_price IS NOT NULL
+					THEN (b.best_bid_price + b.best_ask_price) / 2
+					ELSE NULL
+				END AS mid_price,
+				CASE
+					WHEN b.best_bid_price IS NOT NULL AND b.best_ask_price IS NOT NULL
+					THEN b.best_ask_price - b.best_bid_price
+					ELSE NULL
+				END AS spread_abs,
+				CASE
+					WHEN b.best_bid_price IS NOT NULL
+						AND b.best_ask_price IS NOT NULL
+						AND (b.best_bid_price + b.best_ask_price) > 0
+					THEN
+						((b.best_ask_price - b.best_bid_price) / ((b.best_bid_price + b.best_ask_price) / 2)) * 10000
+					ELSE NULL
+				END AS spread_bps,
+				${this.config.statsDepth}::integer AS depth_levels,
+				bids.depth_bid_base,
+				asks.depth_ask_base,
+				bids.depth_bid_usd,
+				asks.depth_ask_usd,
+				CASE
+					WHEN (bids.depth_bid_usd + asks.depth_ask_usd) > 0
+					THEN (bids.depth_bid_usd - asks.depth_ask_usd) / (bids.depth_bid_usd + asks.depth_ask_usd)
+					ELSE NULL
+				END AS imbalance,
+				jsonb_build_object(
+					'canonical_table', 'dlob_orderbook_l2_latest',
+					'projection', 'dlob_stats_latest_projection'
+				) AS raw,
+				b.updated_at
+			FROM base b
+			LEFT JOIN bid_levels bids USING (source, market_name)
+			LEFT JOIN ask_levels asks USING (source, market_name);
+		`);
+	}
+
+	async writeL2Snapshot(
+		l2Snapshot: Record<string, any>,
+		options: { indicative: boolean }
+	): Promise<void> {
+		this.writeChain = this.writeChain
+			.then(() =>
+				this.pool.query(
+					`
+						INSERT INTO public.dlob_orderbook_l2_latest (
+							source,
+							market_name,
+							market_type,
+							market_index,
+							is_indicative,
+							ts,
+							slot,
+							mark_price_raw,
+							oracle_price_raw,
+							best_bid_price_raw,
+							best_ask_price_raw,
+							spread_pct_raw,
+							spread_quote_raw,
+							oracle_data,
+							bids,
+							asks,
+							payload,
+							updated_at
+						)
+						VALUES (
+							$1, $2, $3, $4, $5, $6, $7, $8, $9,
+							$10, $11, $12, $13,
+							$14::jsonb, $15::jsonb, $16::jsonb, $17::jsonb, now()
+						)
+						ON CONFLICT (source, market_name, is_indicative) DO UPDATE SET
+							market_type = EXCLUDED.market_type,
+							market_index = EXCLUDED.market_index,
+							ts = EXCLUDED.ts,
+							slot = EXCLUDED.slot,
+							mark_price_raw = EXCLUDED.mark_price_raw,
+							oracle_price_raw = EXCLUDED.oracle_price_raw,
+							best_bid_price_raw = EXCLUDED.best_bid_price_raw,
+							best_ask_price_raw = EXCLUDED.best_ask_price_raw,
+							spread_pct_raw = EXCLUDED.spread_pct_raw,
+							spread_quote_raw = EXCLUDED.spread_quote_raw,
+							oracle_data = EXCLUDED.oracle_data,
+							bids = EXCLUDED.bids,
+							asks = EXCLUDED.asks,
+							payload = EXCLUDED.payload,
+							updated_at = now()
+					`,
+					[
+						this.config.source,
+						String(l2Snapshot.marketName),
+						String(l2Snapshot.marketType || 'perp'),
+						toIntOrNull(l2Snapshot.marketIndex),
+						options.indicative,
+						toIntOrNull(l2Snapshot.ts),
+						toIntOrNull(l2Snapshot.slot),
+						l2Snapshot.markPrice ?? null,
+						l2Snapshot.oracleData?.price ?? l2Snapshot.oracle ?? null,
+						l2Snapshot.bestBidPrice ?? null,
+						l2Snapshot.bestAskPrice ?? null,
+						l2Snapshot.spreadPct ?? null,
+						l2Snapshot.spreadQuote ?? null,
+						toJson(l2Snapshot.oracleData ?? null),
+						toJson(l2Snapshot.bids ?? null),
+						toJson(l2Snapshot.asks ?? null),
+						toJson(l2Snapshot),
+					]
+				)
+			)
+			.catch((error) => {
+				logger.error('Failed to persist canonical DLOB L2 snapshot', error);
+			});
+
+		return this.writeChain;
+	}
+
+	async writeL3Snapshot(
+		l3Snapshot: Record<string, any>,
+		options: { indicative: boolean }
+	): Promise<void> {
+		this.writeChain = this.writeChain
+			.then(() =>
+				this.pool.query(
+					`
+						INSERT INTO public.dlob_orderbook_l3_latest (
+							source,
+							market_name,
+							market_type,
+							market_index,
+							is_indicative,
+							ts,
+							slot,
+							oracle_data,
+							bids,
+							asks,
+							payload,
+							updated_at
+						)
+						VALUES (
+							$1, $2, $3, $4, $5, $6, $7,
+							$8::jsonb, $9::jsonb, $10::jsonb, $11::jsonb, now()
+						)
+						ON CONFLICT (source, market_name, is_indicative) DO UPDATE SET
+							market_type = EXCLUDED.market_type,
+							market_index = EXCLUDED.market_index,
+							ts = EXCLUDED.ts,
+							slot = EXCLUDED.slot,
+							oracle_data = EXCLUDED.oracle_data,
+							bids = EXCLUDED.bids,
+							asks = EXCLUDED.asks,
+							payload = EXCLUDED.payload,
+							updated_at = now()
+					`,
+					[
+						this.config.source,
+						String(l3Snapshot.marketName),
+						String(l3Snapshot.marketType || 'perp'),
+						toIntOrNull(l3Snapshot.marketIndex),
+						options.indicative,
+						toIntOrNull(l3Snapshot.ts),
+						toIntOrNull(l3Snapshot.slot),
+						toJson(l3Snapshot.oracleData ?? null),
+						toJson(l3Snapshot.bids ?? null),
+						toJson(l3Snapshot.asks ?? null),
+						toJson(l3Snapshot),
+					]
+				)
+			)
+			.catch((error) => {
+				logger.error('Failed to persist canonical DLOB L3 snapshot', error);
+			});
+
+		return this.writeChain;
+	}
+
+	async writeBestMakersSnapshot(snapshot: {
+		marketName: string;
+		marketType: string;
+		marketIndex: number;
+		slot: number | null;
+		bids: string[];
+		asks: string[];
+	}): Promise<void> {
+		this.writeChain = this.writeChain
+			.then(() =>
+				this.pool.query(
+					`
+						INSERT INTO public.dlob_best_makers_latest (
+							source,
+							market_name,
+							market_type,
+							market_index,
+							slot,
+							bids,
+							asks,
+							payload,
+							updated_at
+						)
+						VALUES (
+							$1, $2, $3, $4, $5, $6::jsonb, $7::jsonb, $8::jsonb, now()
+						)
+						ON CONFLICT (source, market_name) DO UPDATE SET
+							market_type = EXCLUDED.market_type,
+							market_index = EXCLUDED.market_index,
+							slot = EXCLUDED.slot,
+							bids = EXCLUDED.bids,
+							asks = EXCLUDED.asks,
+							payload = EXCLUDED.payload,
+							updated_at = now()
+					`,
+					[
+						this.config.source,
+						snapshot.marketName,
+						snapshot.marketType,
+						snapshot.marketIndex,
+						snapshot.slot,
+						toJson(snapshot.bids),
+						toJson(snapshot.asks),
+						toJson(snapshot),
+					]
+				)
+			)
+			.catch((error) => {
+				logger.error('Failed to persist canonical DLOB best makers', error);
+			});
+
+		return this.writeChain;
+	}
+
+	async close(): Promise<void> {
+		await this.writeChain.catch(() => undefined);
+		await this.pool.end();
+	}
+}
+
+export function createPostgresCanonicalStoreFromEnv(): DlobSnapshotStore {
+	if (!envBool('ENABLE_PERSISTENT_STORE', false)) {
+		return new NoopDlobSnapshotStore();
+	}
+
+	const connectionString =
+		process.env.POSTGRES_URL || process.env.DATABASE_URL || undefined;
+
+	if (
+		!connectionString &&
+		(!process.env.PGHOST ||
+			!process.env.PGUSER ||
+			!process.env.PGDATABASE)
+	) {
+		throw new Error(
+			'ENABLE_PERSISTENT_STORE=true requires POSTGRES_URL or PGHOST/PGUSER/PGDATABASE'
+		);
+	}
+
+	return new PostgresCanonicalStore({
+		connectionString,
+		host: process.env.PGHOST,
+		port: envInt('PGPORT', 5432),
+		user: process.env.PGUSER,
+		password: process.env.PGPASSWORD,
+		database: process.env.PGDATABASE,
+		source: process.env.DLOB_SOURCE || 'mevnode',
+		pricePrecision: envInt('PRICE_PRECISION', 1_000_000),
+		basePrecision: envInt('BASE_PRECISION', 1_000_000_000),
+		statsDepth: envInt('PERSISTENT_STATS_DEPTH', 10),
+	});
+}

src/wsConnectionManager.ts

@@ -6,7 +6,161 @@ import { WebSocket, WebSocketServer } from 'ws';
 import { sleep, selectMostRecentBySlot, GROUPING_OPTIONS } from './utils/utils';
 import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
 import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-import { Metrics } from './core/metricsV2';
+import { Metrics, GaugeValue } from './core/metricsV2';
+
+// Stream selector with hysteresis to prevent flip-flopping between sources
+const STREAM_SWITCH_THRESHOLD_MS = 10_000; // 10 seconds
+const STREAM_STALE_THRESHOLD_MS = 3_000; // Consider stream stale if no message for 3 seconds
+
+class StreamSelector {
+	private activeStream: string | null = null;
+	private streamLastMessageTime: Map<string, number> = new Map();
+	private streamLastSlot: Map<string, number> = new Map();
+	private streamLeadingSince: Map<string, number> = new Map();
+	private streamHealthyGauge: GaugeValue;
+	private activeStreamGauge: GaugeValue;
+	private streams: string[];
+
+	constructor(
+		streams: string[],
+		streamHealthyGauge: GaugeValue,
+		activeStreamGauge: GaugeValue
+	) {
+		this.streams = streams;
+		this.streamHealthyGauge = streamHealthyGauge;
+		this.activeStreamGauge = activeStreamGauge;
+
+		// Initialize all streams as unhealthy
+		for (const stream of streams) {
+			this.streamHealthyGauge.setLatestValue(0, { source: stream });
+			this.activeStreamGauge.setLatestValue(0, { source: stream });
+		}
+	}
+
+	// Record a message from a stream and return whether it should be forwarded
+	recordMessage(stream: string, slot: number): boolean {
+		const now = Date.now();
+		this.streamLastMessageTime.set(stream, now);
+		this.streamLastSlot.set(stream, slot);
+
+		this.streamHealthyGauge.setLatestValue(1, { source: stream });
+
+		if (this.activeStream === null) {
+			this.setActiveStream(stream);
+			return true;
+		}
+
+		// If this is from the active stream, forward it
+		if (stream === this.activeStream) {
+			// Reset leading times for other streams since active stream is still producing
+			for (const s of this.streams) {
+				if (s !== stream) {
+					this.streamLeadingSince.delete(s);
+				}
+			}
+			return true;
+		}
+
+		// This message is from a non-active stream
+		const activeSlot = this.streamLastSlot.get(this.activeStream) || 0;
+		const activeLastMessage =
+			this.streamLastMessageTime.get(this.activeStream) || 0;
+
+		// Check if active stream is stale
+		if (now - activeLastMessage > STREAM_STALE_THRESHOLD_MS) {
+			console.log(
+				`Active stream ${this.activeStream} is stale (no message for ${
+					now - activeLastMessage
+				}ms), switching to ${stream}`
+			);
+			this.setActiveStream(stream);
+			return true;
+		}
+
+		// Check if this stream has a more recent slot
+		if (slot > activeSlot) {
+			// Track how long this stream has been leading
+			if (!this.streamLeadingSince.has(stream)) {
+				this.streamLeadingSince.set(stream, now);
+			}
+
+			const leadingDuration = now - this.streamLeadingSince.get(stream)!;
+			if (leadingDuration >= STREAM_SWITCH_THRESHOLD_MS) {
+				console.log(
+					`Stream ${stream} has been leading for ${leadingDuration}ms, switching from ${this.activeStream}`
+				);
+				this.setActiveStream(stream);
+				return true;
+			}
+		} else {
+			// This stream is not leading, reset its leading time
+			this.streamLeadingSince.delete(stream);
+		}
+
+		// Don't forward messages from non-active streams that haven't proven themselves
+		return false;
+	}
+
+	private setActiveStream(stream: string) {
+		if (this.activeStream) {
+			this.activeStreamGauge.setLatestValue(0, { source: this.activeStream });
+		}
+
+		this.activeStream = stream;
+		this.activeStreamGauge.setLatestValue(1, { source: stream });
+		this.streamLeadingSince.clear();
+
+		console.log(`Active stream set to: ${stream}`);
+	}
+
+	getActiveStream(): string | null {
+		return this.activeStream;
+	}
+
+	checkHealth() {
+		const now = Date.now();
+		let anyHealthy = false;
+
+		for (const stream of this.streams) {
+			const lastMessage = this.streamLastMessageTime.get(stream);
+			const isHealthy =
+				lastMessage && now - lastMessage < STREAM_STALE_THRESHOLD_MS;
+
+			this.streamHealthyGauge.setLatestValue(isHealthy ? 1 : 0, {
+				source: stream,
+			});
+
+			if (isHealthy) {
+				anyHealthy = true;
+			}
+		}
+
+		// If active stream is no longer healthy, try to switch
+		if (this.activeStream) {
+			const activeLastMessage = this.streamLastMessageTime.get(
+				this.activeStream
+			);
+			if (
+				!activeLastMessage ||
+				now - activeLastMessage > STREAM_STALE_THRESHOLD_MS
+			) {
+				// Find a healthy stream to switch to
+				for (const stream of this.streams) {
+					const lastMessage = this.streamLastMessageTime.get(stream);
+					if (lastMessage && now - lastMessage < STREAM_STALE_THRESHOLD_MS) {
+						console.log(
+							`Active stream ${this.activeStream} is unhealthy, switching to ${stream}`
+						);
+						this.setActiveStream(stream);
+						break;
+					}
+				}
+			}
+		}
+
+		return anyHealthy;
+	}
+}
 
 // Set up env constants
 require('dotenv').config();
@@ -34,6 +188,14 @@ const wsOrderbookSourceLastSlotGauge = metricsV2.addGauge(
 	'websocket_orderbook_source_last_slot',
 	'Last slot of orderbook messages from a source'
 );
+const wsStreamHealthyGauge = metricsV2.addGauge(
+	'websocket_stream_healthy',
+	'Whether a Redis stream source is healthy (1) or not (0). Alert if all sources are 0.'
+);
+const wsActiveStreamGauge = metricsV2.addGauge(
+	'websocket_active_stream',
+	'Whether a Redis stream source is currently active (1) or not (0). Only one source should be active at a time.'
+);
 metricsV2.finalizeObservables();
 
 const server = http.createServer(app);
@@ -125,6 +287,32 @@ const getRedisChannelFromMessage = (message: any): string => {
 async function main() {
 	const subscribedChannelToSlot: Map<string, number> = new Map();
 
+	// Create stream selector for managing failover between Redis sources
+	const streamSelector =
+		REDIS_CLIENTS.length > 1
+			? new StreamSelector(
+					REDIS_CLIENTS,
+					wsStreamHealthyGauge,
+					wsActiveStreamGauge
+			  )
+			: null;
+
+	// If only one client, mark it as healthy and active
+	if (!streamSelector && REDIS_CLIENTS.length === 1) {
+		wsStreamHealthyGauge.setLatestValue(1, { source: REDIS_CLIENTS[0] });
+		wsActiveStreamGauge.setLatestValue(1, { source: REDIS_CLIENTS[0] });
+	}
+
+	// Periodic health check for stream selector
+	if (streamSelector) {
+		setInterval(() => {
+			const anyHealthy = streamSelector.checkHealth();
+			if (!anyHealthy) {
+				console.error('WARNING: No healthy Redis streams available!');
+			}
+		}, 1000);
+	}
+
 	const redisClients: Array<RedisClient> = [];
 	const lastMessageClients: Array<RedisClient> = [];
 	for (let i = 0; i < REDIS_CLIENTS.length; i++) {
@@ -157,12 +345,26 @@ async function main() {
 						source: channelPrefix,
 					});
 
-					const lastMessageSlot = subscribedChannelToSlot.get(sanitizedChannel);
-					if (!lastMessageSlot || lastMessageSlot <= messageSlot) {
-						subscribedChannelToSlot.set(sanitizedChannel, messageSlot);
-					} else if (lastMessageSlot > messageSlot) {
-						return;
+					if (streamSelector) {
+						const shouldForward = streamSelector.recordMessage(
+							channelPrefix,
+							messageSlot
+						);
+						if (!shouldForward) {
+							return;
+						}
+					} else {
+						const lastMessageSlot =
+							subscribedChannelToSlot.get(sanitizedChannel);
+						if (!lastMessageSlot || lastMessageSlot <= messageSlot) {
+							subscribedChannelToSlot.set(sanitizedChannel, messageSlot);
+						} else if (lastMessageSlot > messageSlot) {
+							return;
+						}
 					}
+
+					// Update slot tracking for the forwarded message
+					subscribedChannelToSlot.set(sanitizedChannel, messageSlot);
 				}
 				subscribers.forEach((ws) => {
 					if (