trade/trade-deploy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dlob): support two sources + per-user switch

Browse Source 
- Add "source" column + composite PKs for DLOB tables\n- Filter public Hasura selects by X-Hasura-Dlob-Source\n- Run parallel workers for mevnode + dlob.drift.trade\n- Frontend proxy sets x-hasura-dlob-source from cookie and injects UI switch
This commit is contained in:
u1
2026-02-13 10:48:20 +01:00
parent 9e7d7b88ac
commit 57433c7e75
17 changed files with 501 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
kustomize/base/dlob-depth-worker/deployment-drift.yaml Normal file
View File

@@ -0,0 +1,50 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: dlob-depth-worker-drift
annotations:
argocd.argoproj.io/sync-wave: "6"
spec:
replicas: 1
selector:
matchLabels:
app.kubernetes.io/name: dlob-depth-worker-drift
template:
metadata:
labels:
app.kubernetes.io/name: dlob-depth-worker-drift
spec:
containers:
- name: worker
image: node:20-slim
imagePullPolicy: IfNotPresent
env:
- name: HASURA_GRAPHQL_URL
value: http://hasura:8080/v1/graphql
- name: HASURA_ADMIN_SECRET
valueFrom:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: drift
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_POLL_MS
value: "1000"
- name: DLOB_DEPTH_BPS_BANDS
value: "5,10,20,50,100,200"
- name: PRICE_PRECISION
value: "1000000"
- name: BASE_PRECISION
value: "1000000000"
command: ["node", "/app/worker.mjs"]
volumeMounts:
- name: script
mountPath: /app/worker.mjs
subPath: worker.mjs
readOnly: true
volumes:
- name: script
configMap:
name: dlob-depth-worker-script

2
kustomize/base/dlob-depth-worker/deployment.yaml
View File

@@ -26,6 +26,8 @@ spec:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: mevnode
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_POLL_MS

11
kustomize/base/dlob-depth-worker/worker.mjs
View File

@@ -64,6 +64,7 @@ function resolveConfig() {
const hasuraAdminSecret = process.env.HASURA_ADMIN_SECRET || process.env.HASURA_GRAPHQL_ADMIN_SECRET || undefined;
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
const markets = envList('DLOB_MARKETS', 'PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP');
const pollMs = clampInt(process.env.DLOB_POLL_MS, 250, 60_000, 1000);
const bandsBps = envIntList('DLOB_DEPTH_BPS_BANDS', '5,10,20,50,100,200');
@@ -79,6 +80,7 @@ function resolveConfig() {
hasuraUrl,
hasuraAdminSecret,
hasuraAuthToken,
dlobSource,
markets,
pollMs,
bandsBps,
@@ -169,8 +171,9 @@ function computeBandDepth({ bids, asks, mid, bandBps }) {
async function fetchL2Latest(cfg) {
const query = `
query DlobL2Latest($markets: [String!]!) {
dlob_l2_latest(where: {market_name: {_in: $markets}}) {
query DlobL2Latest($source: String!, $markets: [String!]!) {
dlob_l2_latest(where: {source: {_eq: $source}, market_name: {_in: $markets}}) {
source
market_name
market_type
market_index
@@ -186,7 +189,7 @@ async function fetchL2Latest(cfg) {
}
}
`;
const data = await graphqlRequest(cfg, query, { markets: cfg.markets });
const data = await graphqlRequest(cfg, query, { source: cfg.dlobSource, markets: cfg.markets });
return Array.isArray(data?.dlob_l2_latest) ? data.dlob_l2_latest : [];
}
@@ -232,6 +235,7 @@ async function main() {
startedAt: getIsoNow(),
hasuraUrl: cfg.hasuraUrl,
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
dlobSource: cfg.dlobSource,
markets: cfg.markets,
pollMs: cfg.pollMs,
bandsBps: cfg.bandsBps,
@@ -268,6 +272,7 @@ async function main() {
for (const bandBps of cfg.bandsBps) {
const d = computeBandDepth({ bids, asks, mid, bandBps });
rows.push({
source: cfg.dlobSource,
market_name: market,
band_bps: bandBps,
market_type: l2.market_type ? String(l2.market_type) : 'perp',

50
kustomize/base/dlob-slippage-worker/deployment-drift.yaml Normal file
View File

@@ -0,0 +1,50 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: dlob-slippage-worker-drift
annotations:
argocd.argoproj.io/sync-wave: "6"
spec:
replicas: 1
selector:
matchLabels:
app.kubernetes.io/name: dlob-slippage-worker-drift
template:
metadata:
labels:
app.kubernetes.io/name: dlob-slippage-worker-drift
spec:
containers:
- name: worker
image: node:20-slim
imagePullPolicy: IfNotPresent
env:
- name: HASURA_GRAPHQL_URL
value: http://hasura:8080/v1/graphql
- name: HASURA_ADMIN_SECRET
valueFrom:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: drift
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_POLL_MS
value: "1000"
- name: DLOB_SLIPPAGE_SIZES_USD
value: "0.1,0.2,0.5,1,2,5,10,25,50,100,250,500,1000,5000,10000,50000"
- name: PRICE_PRECISION
value: "1000000"
- name: BASE_PRECISION
value: "1000000000"
command: ["node", "/app/worker.mjs"]
volumeMounts:
- name: script
mountPath: /app/worker.mjs
subPath: worker.mjs
readOnly: true
volumes:
- name: script
configMap:
name: dlob-slippage-worker-script

2
kustomize/base/dlob-slippage-worker/deployment.yaml
View File

@@ -26,6 +26,8 @@ spec:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: mevnode
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_POLL_MS

12
kustomize/base/dlob-slippage-worker/worker.mjs
View File

@@ -55,6 +55,7 @@ function resolveConfig() {
tokens.hasuraAdminSecret;
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
const markets = envList('DLOB_MARKETS', 'PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP');
const pollMs = clampInt(process.env.DLOB_POLL_MS, 250, 60_000, 1000);
@@ -77,6 +78,7 @@ function resolveConfig() {
hasuraUrl,
hasuraAdminSecret,
hasuraAuthToken,
dlobSource,
markets,
pollMs,
sizesUsd,
@@ -209,6 +211,7 @@ async function main() {
startedAt: getIsoNow(),
hasuraUrl: cfg.hasuraUrl,
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
dlobSource: cfg.dlobSource,
markets: cfg.markets,
pollMs: cfg.pollMs,
sizesUsd: cfg.sizesUsd,
@@ -226,8 +229,9 @@ async function main() {
try {
const query = `
query DlobL2Latest($markets: [String!]!) {
dlob_l2_latest(where: { market_name: { _in: $markets } }) {
query DlobL2Latest($source: String!, $markets: [String!]!) {
dlob_l2_latest(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
source
market_name
market_type
market_index
@@ -242,7 +246,7 @@ async function main() {
}
`;
const data = await graphqlRequest(cfg, query, { markets: cfg.markets });
const data = await graphqlRequest(cfg, query, { source: cfg.dlobSource, markets: cfg.markets });
const rows = Array.isArray(data?.dlob_l2_latest) ? data.dlob_l2_latest : [];
const objectsV1 = [];
@@ -277,6 +281,7 @@ async function main() {
{
const sim = simulateFill(asks, sizeUsd);
const baseObj = {
source: cfg.dlobSource,
market_name: market,
side: 'buy',
market_type: row?.market_type ?? 'perp',
@@ -302,6 +307,7 @@ async function main() {
{
const sim = simulateFill(bids, sizeUsd);
const baseObj = {
source: cfg.dlobSource,
market_name: market,
side: 'sell',
market_type: row?.market_type ?? 'perp',

44
kustomize/base/dlob-ts-archiver/deployment-drift.yaml Normal file
View File

@@ -0,0 +1,44 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: dlob-ts-archiver-drift
annotations:
argocd.argoproj.io/sync-wave: "6"
spec:
replicas: 1
selector:
matchLabels:
app.kubernetes.io/name: dlob-ts-archiver-drift
template:
metadata:
labels:
app.kubernetes.io/name: dlob-ts-archiver-drift
spec:
containers:
- name: worker
image: node:20-slim
imagePullPolicy: IfNotPresent
env:
- name: HASURA_GRAPHQL_URL
value: http://hasura:8080/v1/graphql
- name: HASURA_ADMIN_SECRET
valueFrom:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: drift
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_TS_POLL_MS
value: "1000"
command: ["node", "/app/worker.mjs"]
volumeMounts:
- name: script
mountPath: /app/worker.mjs
subPath: worker.mjs
readOnly: true
volumes:
- name: script
configMap:
name: dlob-ts-archiver-script

2
kustomize/base/dlob-ts-archiver/deployment.yaml
View File

@@ -26,6 +26,8 @@ spec:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: mevnode
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_TS_POLL_MS

20
kustomize/base/dlob-ts-archiver/worker.mjs
View File

@@ -49,10 +49,11 @@ function resolveConfig() {
tokens.hasuraAdminSecret;
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
const markets = envList('DLOB_MARKETS', 'PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP');
const pollMs = clampInt(process.env.DLOB_TS_POLL_MS, 500, 60_000, 1000);
return { hasuraUrl, hasuraAdminSecret, hasuraAuthToken, markets, pollMs };
return { hasuraUrl, hasuraAdminSecret, hasuraAuthToken, dlobSource, markets, pollMs };
}
async function graphqlRequest(cfg, query, variables) {
@@ -97,6 +98,7 @@ async function main() {
startedAt: getIsoNow(),
hasuraUrl: cfg.hasuraUrl,
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
dlobSource: cfg.dlobSource,
markets: cfg.markets,
pollMs: cfg.pollMs,
},
@@ -110,24 +112,24 @@ async function main() {
try {
const query = `
query Latest($markets: [String!]!) {
dlob_stats_latest(where: { market_name: { _in: $markets } }) {
query Latest($source: String!, $markets: [String!]!) {
dlob_stats_latest(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
market_name market_type market_index ts slot
mark_price oracle_price best_bid_price best_ask_price mid_price
spread_abs spread_bps depth_levels depth_bid_base depth_ask_base depth_bid_usd depth_ask_usd imbalance
raw
}
dlob_depth_bps_latest(where: { market_name: { _in: $markets } }) {
dlob_depth_bps_latest(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
market_name band_bps market_type market_index ts slot
mid_price best_bid_price best_ask_price bid_base ask_base bid_usd ask_usd imbalance
raw
}
dlob_slippage_latest(where: { market_name: { _in: $markets } }) {
dlob_slippage_latest(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
market_name side size_usd market_type market_index ts slot
mid_price vwap_price worst_price filled_usd filled_base impact_bps levels_consumed fill_pct
raw
}
dlob_slippage_latest_v2(where: { market_name: { _in: $markets } }) {
dlob_slippage_latest_v2(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
market_name side size_usd market_type market_index ts slot
mid_price vwap_price worst_price filled_usd filled_base impact_bps levels_consumed fill_pct
raw
@@ -135,10 +137,11 @@ async function main() {
}
`;
const data = await graphqlRequest(cfg, query, { markets: cfg.markets });
const data = await graphqlRequest(cfg, query, { source: cfg.dlobSource, markets: cfg.markets });
const statsRows = (data?.dlob_stats_latest || []).map((r) => ({
ts: now,
source: cfg.dlobSource,
market_name: r.market_name,
market_type: r.market_type,
market_index: r.market_index ?? null,
@@ -162,6 +165,7 @@ async function main() {
const depthRows = (data?.dlob_depth_bps_latest || []).map((r) => ({
ts: now,
source: cfg.dlobSource,
market_name: r.market_name,
band_bps: r.band_bps,
market_type: r.market_type,
@@ -181,6 +185,7 @@ async function main() {
const slippageRows = (data?.dlob_slippage_latest || []).map((r) => ({
ts: now,
source: cfg.dlobSource,
market_name: r.market_name,
side: r.side,
size_usd: r.size_usd,
@@ -201,6 +206,7 @@ async function main() {
const slippageRowsV2 = (data?.dlob_slippage_latest_v2 || []).map((r) => ({
ts: now,
source: cfg.dlobSource,
market_name: r.market_name,
side: r.side,
size_usd: r.size_usd,

52
kustomize/base/dlob-worker/deployment-drift.yaml Normal file
View File

@@ -0,0 +1,52 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: dlob-worker-drift
annotations:
argocd.argoproj.io/sync-wave: "5"
spec:
replicas: 1
selector:
matchLabels:
app.kubernetes.io/name: dlob-worker-drift
template:
metadata:
labels:
app.kubernetes.io/name: dlob-worker-drift
spec:
hostNetwork: true
dnsPolicy: ClusterFirstWithHostNet
containers:
- name: worker
image: node:20-slim
imagePullPolicy: IfNotPresent
env:
- name: HASURA_GRAPHQL_URL
value: http://hasura:8080/v1/graphql
- name: HASURA_ADMIN_SECRET
valueFrom:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: drift
- name: DLOB_HTTP_URL
value: https://dlob.drift.trade
- name: DLOB_FORCE_IPV6
value: "true"
- name: DLOB_MARKETS
value: PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP
- name: DLOB_POLL_MS
value: "500"
- name: DLOB_DEPTH
value: "10"
command: ["node", "/app/worker.mjs"]
volumeMounts:
- name: script
mountPath: /app/worker.mjs
subPath: worker.mjs
readOnly: true
volumes:
- name: script
configMap:
name: dlob-worker-script

2
kustomize/base/dlob-worker/deployment.yaml
View File

@@ -26,6 +26,8 @@ spec:
secretKeyRef:
name: trade-hasura
key: HASURA_GRAPHQL_ADMIN_SECRET
- name: DLOB_SOURCE
value: mevnode
- name: DLOB_HTTP_URL
value: http://dlob-server:6969
- name: DLOB_MARKETS

13
kustomize/base/dlob-worker/worker.mjs
View File

@@ -64,6 +64,7 @@ function resolveConfig() {
.trim()
.replace(/\/$/, '');
const dlobForceIpv6 = envBool('DLOB_FORCE_IPV6', false);
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
const markets = envList('DLOB_MARKETS', 'PUMP-PERP,SOL-PERP,1MBONK-PERP,BTC-PERP,ETH-PERP');
const depth = clampInt(process.env.DLOB_DEPTH, 1, 50, 10);
@@ -80,6 +81,7 @@ function resolveConfig() {
hasuraUrl,
hasuraAdminSecret,
hasuraAuthToken,
dlobSource,
dlobHttpBase,
dlobForceIpv6,
markets,
@@ -238,8 +240,9 @@ function computeStats({ l2, depth, pricePrecision, basePrecision }) {
};
}
function l2ToInsertObject({ l2, updatedAt, pricePrecision }) {
function l2ToInsertObject({ dlobSource, l2, updatedAt, pricePrecision }) {
return {
source: dlobSource,
market_name: String(l2.marketName),
market_type: String(l2.marketType || 'perp'),
market_index: typeof l2.marketIndex === 'number' ? l2.marketIndex : null,
@@ -256,8 +259,9 @@ function l2ToInsertObject({ l2, updatedAt, pricePrecision }) {
};
}
function statsToInsertObject({ l2, stats, updatedAt }) {
function statsToInsertObject({ dlobSource, l2, stats, updatedAt }) {
return {
source: dlobSource,
market_name: String(l2.marketName),
market_type: String(l2.marketType || 'perp'),
market_index: typeof l2.marketIndex === 'number' ? l2.marketIndex : null,
@@ -371,6 +375,7 @@ async function main() {
startedAt: getIsoNow(),
hasuraUrl: cfg.hasuraUrl,
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
dlobSource: cfg.dlobSource,
dlobHttpBase: cfg.dlobHttpBase,
dlobForceIpv6: cfg.dlobForceIpv6,
markets: cfg.markets,
@@ -410,8 +415,8 @@ async function main() {
basePrecision: cfg.basePrecision,
});
l2Objects.push(l2ToInsertObject({ l2, updatedAt, pricePrecision: cfg.pricePrecision }));
statsObjects.push(statsToInsertObject({ l2, stats, updatedAt }));
l2Objects.push(l2ToInsertObject({ dlobSource: cfg.dlobSource, l2, updatedAt, pricePrecision: cfg.pricePrecision }));
statsObjects.push(statsToInsertObject({ dlobSource: cfg.dlobSource, l2, stats, updatedAt }));
}
try {

39
kustomize/base/hasura/hasura-bootstrap.mjs
View File

@@ -89,6 +89,8 @@ async function main() {
console.log(`[hasura-bootstrap] HASURA_URL=${HASURA_URL}`);
await waitForHasura();
const PUBLIC_DLOB_SOURCE_HEADER = 'X-Hasura-Dlob-Source';
const apiTokensTable = { schema: 'public', name: 'api_tokens' };
const source = 'default';
@@ -187,7 +189,7 @@ async function main() {
'updated_at',
]);
const ensureDlobTable = async (table, columns) => {
const ensureDlobTable = async (table, columns, { publicFilter } = {}) => {
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table } });
await metadata({ type: 'pg_track_table', args: { source, table } });
@@ -200,7 +202,7 @@ async function main() {
role: 'public',
permission: {
columns,
filter: {},
filter: publicFilter || {},
},
},
});
@@ -236,7 +238,7 @@ async function main() {
});
};
async function ensurePublicSelectTable(table, columns) {
async function ensurePublicSelectTable(table, columns, { publicFilter } = {}) {
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table } });
await metadata({ type: 'pg_track_table', args: { source, table } });
@@ -249,7 +251,7 @@ async function main() {
role: 'public',
permission: {
columns,
filter: {},
filter: publicFilter || {},
},
},
});
@@ -259,7 +261,10 @@ async function main() {
await metadataIgnore({ type: 'pg_drop_update_permission', args: { source, table, role: 'ingestor' } });
}
const dlobPublicFilter = { source: { _eq: PUBLIC_DLOB_SOURCE_HEADER } };
await ensureDlobTable(dlobL2LatestTable, [
'source',
'market_name',
'market_type',
'market_index',
@@ -273,9 +278,10 @@ async function main() {
'asks',
'raw',
'updated_at',
]);
], { publicFilter: dlobPublicFilter });
await ensureDlobTable(dlobStatsLatestTable, [
'source',
'market_name',
'market_type',
'market_index',
@@ -296,9 +302,10 @@ async function main() {
'imbalance',
'raw',
'updated_at',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobDepthBpsLatestTable, [
'source',
'market_name',
'band_bps',
'market_type',
@@ -315,9 +322,10 @@ async function main() {
'imbalance',
'raw',
'updated_at',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobSlippageLatestTable, [
'source',
'market_name',
'side',
'size_usd',
@@ -337,9 +345,10 @@ async function main() {
'fill_pct',
'raw',
'updated_at',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobSlippageLatestV2Table, [
'source',
'market_name',
'side',
'size_usd',
@@ -359,11 +368,12 @@ async function main() {
'fill_pct',
'raw',
'updated_at',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobStatsTsTable, [
'ts',
'id',
'source',
'market_name',
'market_type',
'market_index',
@@ -383,11 +393,12 @@ async function main() {
'depth_ask_usd',
'imbalance',
'raw',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobDepthBpsTsTable, [
'ts',
'id',
'source',
'market_name',
'band_bps',
'market_type',
@@ -403,11 +414,12 @@ async function main() {
'ask_usd',
'imbalance',
'raw',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobSlippageTsTable, [
'ts',
'id',
'source',
'market_name',
'side',
'size_usd',
@@ -424,11 +436,12 @@ async function main() {
'levels_consumed',
'fill_pct',
'raw',
]);
], { publicFilter: dlobPublicFilter });
await ensurePublicSelectTable(dlobSlippageTsV2Table, [
'ts',
'id',
'source',
'market_name',
'side',
'size_usd',
@@ -445,7 +458,7 @@ async function main() {
'levels_consumed',
'fill_pct',
'raw',
]);
], { publicFilter: dlobPublicFilter });
// Return table type for candle functions (needed for Hasura to track the function).
await metadataIgnore({ type: 'pg_track_table', args: { source, table: candlesReturnTable } });

235
kustomize/base/initdb/001_init.sql
View File

@@ -321,7 +321,8 @@ $$;
-- Latest DLOB orderbook snapshots (top-N levels), per market.
-- Filled by a VPS worker (collector) and consumed by the UI via Hasura subscriptions.
CREATE TABLE IF NOT EXISTS public.dlob_l2_latest (
market_name TEXT PRIMARY KEY,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
market_type TEXT NOT NULL DEFAULT 'perp',
market_index INTEGER,
ts BIGINT,
@@ -333,15 +334,52 @@ CREATE TABLE IF NOT EXISTS public.dlob_l2_latest (
bids JSONB,
asks JSONB,
raw JSONB,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (source, market_name)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_l2_latest ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_l2_latest ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_l2_latest SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_l2_latest ALTER COLUMN source SET NOT NULL;
-- Ensure PRIMARY KEY is (source, market_name) (required to keep 2 sources in parallel).
DO $$
DECLARE
pk_name text;
pk_cols text[];
BEGIN
SELECT
con.conname,
array_agg(att.attname ORDER BY ord.ordinality)
INTO pk_name, pk_cols
FROM pg_constraint con
JOIN pg_class rel ON rel.oid = con.conrelid
JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS ord(attnum, ordinality) ON true
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = ord.attnum
WHERE con.contype = 'p' AND nsp.nspname = 'public' AND rel.relname = 'dlob_l2_latest'
GROUP BY con.conname;
IF pk_name IS NULL THEN
EXECUTE 'ALTER TABLE public.dlob_l2_latest ADD CONSTRAINT dlob_l2_latest_pkey PRIMARY KEY (source, market_name)';
ELSIF pk_cols <> ARRAY['source','market_name'] THEN
EXECUTE format('ALTER TABLE public.dlob_l2_latest DROP CONSTRAINT %I', pk_name);
EXECUTE 'ALTER TABLE public.dlob_l2_latest ADD CONSTRAINT dlob_l2_latest_pkey PRIMARY KEY (source, market_name)';
END IF;
END $$;
CREATE INDEX IF NOT EXISTS dlob_l2_latest_updated_at_idx
ON public.dlob_l2_latest (updated_at DESC);
CREATE INDEX IF NOT EXISTS dlob_l2_latest_source_updated_at_idx
ON public.dlob_l2_latest (source, updated_at DESC);
-- Derived stats for fast UI display.
CREATE TABLE IF NOT EXISTS public.dlob_stats_latest (
market_name TEXT PRIMARY KEY,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
market_type TEXT NOT NULL DEFAULT 'perp',
market_index INTEGER,
ts BIGINT,
@@ -360,15 +398,52 @@ CREATE TABLE IF NOT EXISTS public.dlob_stats_latest (
depth_ask_usd NUMERIC,
imbalance NUMERIC,
raw JSONB,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (source, market_name)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_stats_latest ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_stats_latest ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_stats_latest SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_stats_latest ALTER COLUMN source SET NOT NULL;
-- Ensure PRIMARY KEY is (source, market_name) (required to keep 2 sources in parallel).
DO $$
DECLARE
pk_name text;
pk_cols text[];
BEGIN
SELECT
con.conname,
array_agg(att.attname ORDER BY ord.ordinality)
INTO pk_name, pk_cols
FROM pg_constraint con
JOIN pg_class rel ON rel.oid = con.conrelid
JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS ord(attnum, ordinality) ON true
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = ord.attnum
WHERE con.contype = 'p' AND nsp.nspname = 'public' AND rel.relname = 'dlob_stats_latest'
GROUP BY con.conname;
IF pk_name IS NULL THEN
EXECUTE 'ALTER TABLE public.dlob_stats_latest ADD CONSTRAINT dlob_stats_latest_pkey PRIMARY KEY (source, market_name)';
ELSIF pk_cols <> ARRAY['source','market_name'] THEN
EXECUTE format('ALTER TABLE public.dlob_stats_latest DROP CONSTRAINT %I', pk_name);
EXECUTE 'ALTER TABLE public.dlob_stats_latest ADD CONSTRAINT dlob_stats_latest_pkey PRIMARY KEY (source, market_name)';
END IF;
END $$;
CREATE INDEX IF NOT EXISTS dlob_stats_latest_updated_at_idx
ON public.dlob_stats_latest (updated_at DESC);
CREATE INDEX IF NOT EXISTS dlob_stats_latest_source_updated_at_idx
ON public.dlob_stats_latest (source, updated_at DESC);
-- Depth snapshots within bps bands around mid-price (per market, per band).
-- Filled by a derived worker that reads `dlob_l2_latest`.
CREATE TABLE IF NOT EXISTS public.dlob_depth_bps_latest (
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
band_bps INTEGER NOT NULL,
market_type TEXT NOT NULL DEFAULT 'perp',
@@ -385,18 +460,54 @@ CREATE TABLE IF NOT EXISTS public.dlob_depth_bps_latest (
imbalance NUMERIC,
raw JSONB,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (market_name, band_bps)
PRIMARY KEY (source, market_name, band_bps)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_depth_bps_latest ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_depth_bps_latest ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_depth_bps_latest SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_depth_bps_latest ALTER COLUMN source SET NOT NULL;
-- Ensure PRIMARY KEY is (source, market_name, band_bps) (required to keep 2 sources in parallel).
DO $$
DECLARE
pk_name text;
pk_cols text[];
BEGIN
SELECT
con.conname,
array_agg(att.attname ORDER BY ord.ordinality)
INTO pk_name, pk_cols
FROM pg_constraint con
JOIN pg_class rel ON rel.oid = con.conrelid
JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS ord(attnum, ordinality) ON true
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = ord.attnum
WHERE con.contype = 'p' AND nsp.nspname = 'public' AND rel.relname = 'dlob_depth_bps_latest'
GROUP BY con.conname;
IF pk_name IS NULL THEN
EXECUTE 'ALTER TABLE public.dlob_depth_bps_latest ADD CONSTRAINT dlob_depth_bps_latest_pkey PRIMARY KEY (source, market_name, band_bps)';
ELSIF pk_cols <> ARRAY['source','market_name','band_bps'] THEN
EXECUTE format('ALTER TABLE public.dlob_depth_bps_latest DROP CONSTRAINT %I', pk_name);
EXECUTE 'ALTER TABLE public.dlob_depth_bps_latest ADD CONSTRAINT dlob_depth_bps_latest_pkey PRIMARY KEY (source, market_name, band_bps)';
END IF;
END $$;
CREATE INDEX IF NOT EXISTS dlob_depth_bps_latest_updated_at_idx
ON public.dlob_depth_bps_latest (updated_at DESC);
CREATE INDEX IF NOT EXISTS dlob_depth_bps_latest_market_name_idx
ON public.dlob_depth_bps_latest (market_name);
CREATE INDEX IF NOT EXISTS dlob_depth_bps_latest_source_market_name_idx
ON public.dlob_depth_bps_latest (source, market_name);
-- Slippage/impact estimates for "market" orders at common USD sizes.
-- Filled by a derived worker that reads `dlob_l2_latest`.
CREATE TABLE IF NOT EXISTS public.dlob_slippage_latest (
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
side TEXT NOT NULL,
size_usd INTEGER NOT NULL,
@@ -416,19 +527,55 @@ CREATE TABLE IF NOT EXISTS public.dlob_slippage_latest (
fill_pct NUMERIC,
raw JSONB,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (market_name, side, size_usd),
PRIMARY KEY (source, market_name, side, size_usd),
CONSTRAINT dlob_slippage_latest_side_chk CHECK (side IN ('buy', 'sell'))
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_slippage_latest ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_slippage_latest ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_slippage_latest SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_slippage_latest ALTER COLUMN source SET NOT NULL;
-- Ensure PRIMARY KEY is (source, market_name, side, size_usd) (required to keep 2 sources in parallel).
DO $$
DECLARE
pk_name text;
pk_cols text[];
BEGIN
SELECT
con.conname,
array_agg(att.attname ORDER BY ord.ordinality)
INTO pk_name, pk_cols
FROM pg_constraint con
JOIN pg_class rel ON rel.oid = con.conrelid
JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS ord(attnum, ordinality) ON true
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = ord.attnum
WHERE con.contype = 'p' AND nsp.nspname = 'public' AND rel.relname = 'dlob_slippage_latest'
GROUP BY con.conname;
IF pk_name IS NULL THEN
EXECUTE 'ALTER TABLE public.dlob_slippage_latest ADD CONSTRAINT dlob_slippage_latest_pkey PRIMARY KEY (source, market_name, side, size_usd)';
ELSIF pk_cols <> ARRAY['source','market_name','side','size_usd'] THEN
EXECUTE format('ALTER TABLE public.dlob_slippage_latest DROP CONSTRAINT %I', pk_name);
EXECUTE 'ALTER TABLE public.dlob_slippage_latest ADD CONSTRAINT dlob_slippage_latest_pkey PRIMARY KEY (source, market_name, side, size_usd)';
END IF;
END $$;
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_updated_at_idx
ON public.dlob_slippage_latest (updated_at DESC);
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_market_name_idx
ON public.dlob_slippage_latest (market_name);
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_source_market_name_idx
ON public.dlob_slippage_latest (source, market_name);
-- Slippage v2: supports fractional order sizes (e.g. 0.1/0.2/0.5 USD), per market and side.
-- Keep v1 intact for backward compatibility and to avoid data loss.
CREATE TABLE IF NOT EXISTS public.dlob_slippage_latest_v2 (
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
side TEXT NOT NULL, -- buy|sell
size_usd NUMERIC NOT NULL,
@@ -448,22 +595,58 @@ CREATE TABLE IF NOT EXISTS public.dlob_slippage_latest_v2 (
fill_pct NUMERIC,
raw JSONB,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (market_name, side, size_usd),
PRIMARY KEY (source, market_name, side, size_usd),
CONSTRAINT dlob_slippage_latest_v2_side_chk CHECK (side IN ('buy', 'sell'))
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_slippage_latest_v2 ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_slippage_latest_v2 ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_slippage_latest_v2 SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_slippage_latest_v2 ALTER COLUMN source SET NOT NULL;
-- Ensure PRIMARY KEY is (source, market_name, side, size_usd) (required to keep 2 sources in parallel).
DO $$
DECLARE
pk_name text;
pk_cols text[];
BEGIN
SELECT
con.conname,
array_agg(att.attname ORDER BY ord.ordinality)
INTO pk_name, pk_cols
FROM pg_constraint con
JOIN pg_class rel ON rel.oid = con.conrelid
JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS ord(attnum, ordinality) ON true
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = ord.attnum
WHERE con.contype = 'p' AND nsp.nspname = 'public' AND rel.relname = 'dlob_slippage_latest_v2'
GROUP BY con.conname;
IF pk_name IS NULL THEN
EXECUTE 'ALTER TABLE public.dlob_slippage_latest_v2 ADD CONSTRAINT dlob_slippage_latest_v2_pkey PRIMARY KEY (source, market_name, side, size_usd)';
ELSIF pk_cols <> ARRAY['source','market_name','side','size_usd'] THEN
EXECUTE format('ALTER TABLE public.dlob_slippage_latest_v2 DROP CONSTRAINT %I', pk_name);
EXECUTE 'ALTER TABLE public.dlob_slippage_latest_v2 ADD CONSTRAINT dlob_slippage_latest_v2_pkey PRIMARY KEY (source, market_name, side, size_usd)';
END IF;
END $$;
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_v2_updated_at_idx
ON public.dlob_slippage_latest_v2 (updated_at DESC);
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_v2_market_name_idx
ON public.dlob_slippage_latest_v2 (market_name);
CREATE INDEX IF NOT EXISTS dlob_slippage_latest_v2_source_market_name_idx
ON public.dlob_slippage_latest_v2 (source, market_name);
-- Time-series tables for UI history (start: 7 days).
-- Keep these append-only; use Timescale hypertables.
CREATE TABLE IF NOT EXISTS public.dlob_stats_ts (
ts TIMESTAMPTZ NOT NULL,
id BIGSERIAL NOT NULL,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
market_type TEXT NOT NULL DEFAULT 'perp',
market_index INTEGER,
@@ -486,14 +669,24 @@ CREATE TABLE IF NOT EXISTS public.dlob_stats_ts (
PRIMARY KEY (ts, id)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_stats_ts ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_stats_ts ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_stats_ts SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_stats_ts ALTER COLUMN source SET NOT NULL;
SELECT create_hypertable('dlob_stats_ts', 'ts', if_not_exists => TRUE, migrate_data => TRUE);
CREATE INDEX IF NOT EXISTS dlob_stats_ts_market_ts_desc_idx
ON public.dlob_stats_ts (market_name, ts DESC);
CREATE INDEX IF NOT EXISTS dlob_stats_ts_source_market_ts_desc_idx
ON public.dlob_stats_ts (source, market_name, ts DESC);
CREATE TABLE IF NOT EXISTS public.dlob_depth_bps_ts (
ts TIMESTAMPTZ NOT NULL,
id BIGSERIAL NOT NULL,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
band_bps INTEGER NOT NULL,
market_type TEXT NOT NULL DEFAULT 'perp',
@@ -512,14 +705,24 @@ CREATE TABLE IF NOT EXISTS public.dlob_depth_bps_ts (
PRIMARY KEY (ts, id)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_depth_bps_ts ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_depth_bps_ts ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_depth_bps_ts SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_depth_bps_ts ALTER COLUMN source SET NOT NULL;
SELECT create_hypertable('dlob_depth_bps_ts', 'ts', if_not_exists => TRUE, migrate_data => TRUE);
CREATE INDEX IF NOT EXISTS dlob_depth_bps_ts_market_ts_desc_idx
ON public.dlob_depth_bps_ts (market_name, ts DESC);
CREATE INDEX IF NOT EXISTS dlob_depth_bps_ts_source_market_ts_desc_idx
ON public.dlob_depth_bps_ts (source, market_name, ts DESC);
CREATE TABLE IF NOT EXISTS public.dlob_slippage_ts (
ts TIMESTAMPTZ NOT NULL,
id BIGSERIAL NOT NULL,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
side TEXT NOT NULL,
size_usd INTEGER NOT NULL,
@@ -539,14 +742,24 @@ CREATE TABLE IF NOT EXISTS public.dlob_slippage_ts (
PRIMARY KEY (ts, id)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_slippage_ts ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_slippage_ts ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_slippage_ts SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_slippage_ts ALTER COLUMN source SET NOT NULL;
SELECT create_hypertable('dlob_slippage_ts', 'ts', if_not_exists => TRUE, migrate_data => TRUE);
CREATE INDEX IF NOT EXISTS dlob_slippage_ts_market_ts_desc_idx
ON public.dlob_slippage_ts (market_name, ts DESC);
CREATE INDEX IF NOT EXISTS dlob_slippage_ts_source_market_ts_desc_idx
ON public.dlob_slippage_ts (source, market_name, ts DESC);
CREATE TABLE IF NOT EXISTS public.dlob_slippage_ts_v2 (
ts TIMESTAMPTZ NOT NULL,
id BIGSERIAL NOT NULL,
source TEXT NOT NULL DEFAULT 'mevnode',
market_name TEXT NOT NULL,
side TEXT NOT NULL,
size_usd NUMERIC NOT NULL,
@@ -566,11 +779,19 @@ CREATE TABLE IF NOT EXISTS public.dlob_slippage_ts_v2 (
PRIMARY KEY (ts, id)
);
-- Schema upgrades (idempotent for existing volumes)
ALTER TABLE public.dlob_slippage_ts_v2 ADD COLUMN IF NOT EXISTS source TEXT;
ALTER TABLE public.dlob_slippage_ts_v2 ALTER COLUMN source SET DEFAULT 'mevnode';
UPDATE public.dlob_slippage_ts_v2 SET source = 'mevnode' WHERE source IS NULL;
ALTER TABLE public.dlob_slippage_ts_v2 ALTER COLUMN source SET NOT NULL;
SELECT create_hypertable('dlob_slippage_ts_v2', 'ts', if_not_exists => TRUE, migrate_data => TRUE);
CREATE INDEX IF NOT EXISTS dlob_slippage_ts_v2_market_ts_desc_idx
ON public.dlob_slippage_ts_v2 (market_name, ts DESC);
CREATE INDEX IF NOT EXISTS dlob_slippage_ts_v2_source_market_ts_desc_idx
ON public.dlob_slippage_ts_v2 (source, market_name, ts DESC);
-- Retention policies (best-effort; safe if Timescale is present).
DO $$
BEGIN

4
kustomize/base/kustomization.yaml
View File

@@ -19,9 +19,13 @@ resources:
- dlob/server-service.yaml
- dlob/server-deployment.yaml
- dlob-worker/deployment.yaml
- dlob-worker/deployment-drift.yaml
- dlob-depth-worker/deployment.yaml
- dlob-depth-worker/deployment-drift.yaml
- dlob-slippage-worker/deployment.yaml
- dlob-slippage-worker/deployment-drift.yaml
- dlob-ts-archiver/deployment.yaml
- dlob-ts-archiver/deployment-drift.yaml
- candles-cache-worker/deployment.yaml
configMapGenerator:

16
kustomize/overlays/staging/dlob-worker-patch.yaml
View File

@@ -1,16 +0,0 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: dlob-worker
spec:
template:
spec:
hostNetwork: true
dnsPolicy: ClusterFirstWithHostNet
containers:
- name: worker
env:
- name: DLOB_HTTP_URL
value: https://dlob.drift.trade
- name: DLOB_FORCE_IPV6
value: "true"

1
kustomize/overlays/staging/kustomization.yaml
View File

@@ -14,7 +14,6 @@ patchesStrategicMerge:
- frontend-auth-patch.yaml
- frontend-graphql-proxy-patch.yaml
- ingestor-dlob-patch.yaml
- dlob-worker-patch.yaml
configMapGenerator:
- name: trade-dlob-ingestor-script