fix(build): use node 24 for drift sdk

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),
+	});
+}