diff --git a/schema/clickhouse/broker_execution.sql b/schema/clickhouse/broker_execution.sql index ef5c030..e7f6d38 100644 --- a/schema/clickhouse/broker_execution.sql +++ b/schema/clickhouse/broker_execution.sql @@ -15,11 +15,19 @@ CREATE DATABASE IF NOT EXISTS broker_execution; -- Order lifecycle events: execute-action results and user-stream order updates. +-- Plain MergeTree intentionally retains duplicates. They are expected from status +-- polling (every GetOrderDetails observation is archived, while the strategy polls +-- twice per cycle) and at-least-once WebSocket delivery. The canonical read-time +-- dedup key is (exchange, account_selector, symbol, order_id, status, +-- filled_amount), taking argMin(..., broker_observed_timestamp); there is no +-- sequence or updated_at column from which to infer a later authoritative row. CREATE TABLE IF NOT EXISTS broker_execution.order_events ( source LowCardinality(String), deployment_id LowCardinality(String), account_selector LowCardinality(String), + -- Caller-declared order origin; a primary read key, default-empty when absent. + order_author LowCardinality(String) DEFAULT '', exchange LowCardinality(String), symbol LowCardinality(String), @@ -65,6 +73,9 @@ ENGINE = MergeTree PARTITION BY toYYYYMM(parseDateTimeBestEffortOrZero(broker_observed_timestamp)) ORDER BY (exchange, symbol, broker_observed_timestamp); +ALTER TABLE broker_execution.order_events +ADD COLUMN IF NOT EXISTS order_author LowCardinality(String) DEFAULT '' AFTER account_selector; + -- CEX value movements: withdrawals, deposits, and sub<->master internal transfers. -- -- Column names/types/ORDER BY match the fiet-maker consumer contract @@ -131,9 +142,11 @@ ADD COLUMN IF NOT EXISTS client_withdrawal_id String DEFAULT '' AFTER external_i -- Column names/types/ORDER BY match the fiet-maker consumer contract: MergeTree, -- DateTime64 timestamps, string quantities, fill_index UInt32. The contract does -- not constrain retention; fills are execution audit facts and carry no TTL. Plain --- MergeTree (contract): the poller re-scans a lookback window after a restart, so --- the same trade can be re-inserted; dedup is at read time (GROUP BY / argMax over --- exchange, account_selector, symbol, order_id, fill_id). +-- MergeTree (contract): the fill poller re-scans a 24-hour lookback window, so the +-- same trade can be re-inserted. The canonical read-time dedup key is (exchange, +-- account_selector, symbol, order_id, fill_id). fill_index is NOT a stable +-- identifier and must not be used as a dedup key. There is no sequence or +-- updated_at column from which to infer a later authoritative row. CREATE TABLE IF NOT EXISTS broker_execution.fill_events ( broker_observed_timestamp DateTime64(3, 'UTC'), diff --git a/src/handlers/execute-action/orders.ts b/src/handlers/execute-action/orders.ts index f24ccea..ada425a 100644 --- a/src/handlers/execute-action/orders.ts +++ b/src/handlers/execute-action/orders.ts @@ -155,6 +155,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { orderType: orderValue.orderType, requestedQuantity: resolvedOrderTelemetry.requestedQuantity, requestedNotional: orderValue.amount * orderValue.price, + orderAuthor: orderValue.orderAuthor, brokerObservedTimestamp: submissionTimestamp, ...telemetryIds, }; @@ -191,6 +192,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise { requestedQuantity: resolvedOrderTelemetry.requestedQuantity ?? orderValue.amount, requestedNotional: orderValue.amount * orderValue.price, + orderAuthor: orderValue.orderAuthor, ...extractOrderTelemetryIds(createOrderParams), }; emitOrderExecutionTelemetryInBackground( diff --git a/src/handlers/execute-action/treasury-call.ts b/src/handlers/execute-action/treasury-call.ts index 6b5fa4e..832f27e 100644 --- a/src/handlers/execute-action/treasury-call.ts +++ b/src/handlers/execute-action/treasury-call.ts @@ -91,6 +91,7 @@ export async function handleTreasuryCall( side: asNonEmptyString(side), requestedQuantity, requestedNotional, + orderAuthor: callValue.orderAuthor, brokerObservedTimestamp: submissionTimestamp, ...telemetryIds, }; diff --git a/src/helpers/broker-execution-archive/rows.ts b/src/helpers/broker-execution-archive/rows.ts index 0bddb0e..a2cadbf 100644 --- a/src/helpers/broker-execution-archive/rows.ts +++ b/src/helpers/broker-execution-archive/rows.ts @@ -287,6 +287,7 @@ export function buildOrderEventArchiveRow(input: { action, subscription_type: input.subscriptionType, order_id: telemetry.orderId, + order_author: telemetry.orderAuthor ?? "", client_order_id: telemetry.clientOrderId, idempotency_id: telemetry.idempotencyId, maker_action_id: telemetry.makerActionId, @@ -357,11 +358,12 @@ export function buildSubscribeStreamArchiveRow(input: { // Column shapes below follow the fiet-maker CEX_EXECUTION_ARCHIVE_CONTRACT: shared // tags + contract columns, all quantities/prices as strings (venue precision -// varies by asset), fill_index/result_index as numbers (UInt32 columns). Two +// varies by asset), fill_index/result_index as numbers (UInt32 columns). Three // deliberate ADDITIVE divergences the consumer contract doesn't yet list: +// order_events carries order_author as a caller-declared primary read key; // transfer_events carries client_withdrawal_id plus fee_amount/fee_currency (the // ccxt withdrawal object exposes the fee, which is the dominant small-commit -// cost), and fill_events.event_kind is stamped with the true trade-history-poller +// cost); and fill_events.event_kind is stamped with the true trade-history-poller // source rather than "create_order_fill". // Preserve venue precision: prefer the raw string the venue returned (usually in diff --git a/src/helpers/order-telemetry.ts b/src/helpers/order-telemetry.ts index 6201f87..f9c4f6a 100644 --- a/src/helpers/order-telemetry.ts +++ b/src/helpers/order-telemetry.ts @@ -19,6 +19,7 @@ export type OrderTelemetryContext = { orderType?: string; requestedQuantity?: number; requestedNotional?: number; + orderAuthor?: string; clientOrderId?: string; idempotencyId?: string; makerActionId?: string; @@ -34,6 +35,7 @@ export type OrderExecutionTelemetry = { side: string; orderType: string; orderId?: string; + orderAuthor?: string; clientOrderId?: string; idempotencyId?: string; makerActionId?: string; @@ -166,6 +168,7 @@ export function buildOrderExecutionTelemetry( orderType: firstString(record?.type, info?.type, context.orderType) ?? "unknown", orderId: firstString(record?.id, info?.orderId, info?.orderID), + orderAuthor: context.orderAuthor, clientOrderId: firstString( context.clientOrderId, record?.clientOrderId, diff --git a/src/schemas/action-payloads.ts b/src/schemas/action-payloads.ts index b20f6cb..e438470 100644 --- a/src/schemas/action-payloads.ts +++ b/src/schemas/action-payloads.ts @@ -41,6 +41,7 @@ export const DepositPayloadSchema = z.object({ export const CallPayloadSchema = z.object({ functionName: z.string().regex(/^[A-Za-z][A-Za-z0-9]*$/), args: z.preprocess(parseJsonString, z.array(z.unknown())).default([]), + orderAuthor: z.string().min(1).optional(), params: z .preprocess(parseJsonString, z.record(z.string(), z.unknown())) .default({}), @@ -84,6 +85,7 @@ export const CreateOrderPayloadSchema = z.object({ price: z.coerce.number().positive(), marketType: marketTypeSchema, clientOrderId: z.string().min(1).optional(), + orderAuthor: z.string().min(1).optional(), params: z.preprocess(parseJsonString, stringNumberRecordSchema).default({}), }); diff --git a/test/broker-execution-archive.test.ts b/test/broker-execution-archive.test.ts index 3b43bdd..e18bf7f 100644 --- a/test/broker-execution-archive.test.ts +++ b/test/broker-execution-archive.test.ts @@ -14,6 +14,8 @@ import type { LogRecord } from "@opentelemetry/api-logs"; import type { Exchange } from "@usherlabs/ccxt"; import { MAX_ARCHIVE_BODY_BYTES } from "../services/archive-forwarder/limits"; import type { ExecuteActionContext } from "../src/handlers/execute-action/context"; +import { handleOrders } from "../src/handlers/execute-action/orders"; +import { handleTreasuryCall } from "../src/handlers/execute-action/treasury-call"; import { handleWithdraw } from "../src/handlers/execute-action/withdraw"; import { redactSecretLiterals, @@ -44,9 +46,11 @@ import { resolveArchiveForwarderUrlFromEnv, rethrowArchiveDurabilityError, } from "../src/helpers/broker-execution-archive/writer"; +import { Action } from "../src/helpers/constants"; import { log } from "../src/helpers/logger"; import { buildOrderExecutionTelemetry } from "../src/helpers/order-telemetry"; import type { OtelLogs } from "../src/helpers/otel"; +import type { PolicyConfig } from "../src/types"; import { startForwarderServer } from "./archive-forwarder-server"; const archiveTestDirectory = mkdtempSync( @@ -286,6 +290,7 @@ describe("broker execution archive rows", () => { cex: "binance", accountLabel: "primary", symbol: "ARB/USDT", + orderAuthor: "maker-alpha", clientOrderId: "client-1", makerActionId: "maker-1", }, @@ -311,6 +316,7 @@ describe("broker execution archive rows", () => { action: "CancelOrder", event_kind: "execute_action", order_id: "99", + order_author: "maker-alpha", client_order_id: "client-1", maker_action_id: "maker-1", }); @@ -368,6 +374,7 @@ describe("broker execution archive rows", () => { ]) { expect(orderRow.row).not.toHaveProperty(key); } + expect(orderRow.row.order_author).toBe(""); const snapshotRow = buildMarketMetadataSnapshotRow({ tags: buildCommonArchiveTags({ @@ -661,6 +668,117 @@ describe("broker execution archive rows", () => { }); }); +describe("order author archive plumbing", () => { + test("archives authors from typed and Call createOrder without forwarding them to the venue", async () => { + const forwarder = await startForwarderServer(); + const archiver = BrokerExecutionArchiver.create({ + forwarderUrl: forwarder.url, + deadLetterPath: createDeadLetterPath(), + deploymentId: "test-deploy", + batchSize: 100, + flushIntervalMs: 60_000, + }); + const createOrderCalls: unknown[][] = []; + const broker = { + loadMarkets: async () => {}, + markets: { + "USDC/USDT": { + symbol: "USDC/USDT", + base: "USDC", + quote: "USDT", + spot: true, + type: "spot", + }, + }, + createOrder: async (...args: unknown[]) => { + createOrderCalls.push(args); + return { + id: `order-${createOrderCalls.length}`, + symbol: args[0], + type: args[1], + side: args[2], + amount: args[3], + status: "open", + filled: 0, + }; + }, + } as unknown as Exchange; + const policy = { + order: { rule: { markets: ["*"], limits: [] } }, + } as unknown as PolicyConfig; + const context = ( + action: (typeof Action)[keyof typeof Action], + payload: Record, + ) => + ({ + action, + call: { request: { payload } }, + wrappedCallback: () => {}, + policy, + brokers: {}, + normalizedCex: "binance", + cex: "binance", + symbol: "USDC/USDT", + selectedBrokerAccount: { exchange: broker, label: "primary" }, + broker, + verity: { proof: "" }, + brokerArchiver: archiver, + }) as unknown as ExecuteActionContext; + const typedPayload = (orderAuthor?: string) => ({ + orderType: "limit", + amount: "10", + fromToken: "USDC", + toToken: "USDT", + price: "1", + marketType: "spot", + ...(orderAuthor !== undefined && { orderAuthor }), + params: JSON.stringify({ timeInForce: "GTC" }), + }); + + try { + await handleOrders( + context(Action.CreateOrder, typedPayload("maker-alpha")), + ); + await handleTreasuryCall( + context(Action.Call, { + functionName: "createOrder", + args: JSON.stringify(["USDC/USDT", "limit", "buy", 5, 1]), + orderAuthor: "funding-executor", + params: JSON.stringify({ postOnly: true }), + }), + ); + await handleOrders(context(Action.CreateOrder, typedPayload())); + + expect(createOrderCalls[0]?.[5]).toEqual({ timeInForce: "GTC" }); + expect(createOrderCalls[1]?.[5]).toEqual({ postOnly: true }); + expect(createOrderCalls[2]?.[5]).toEqual({ timeInForce: "GTC" }); + for (const call of createOrderCalls) { + expect(call[5]).not.toHaveProperty("orderAuthor"); + } + + await Promise.resolve(); + await archiver.flush(); + const orderRows = forwarder.requests + .flatMap((request) => request.body.rows ?? []) + .filter( + (entry) => entry.table === "broker_execution.order_events", + ) as Array<{ row: Record }>; + const orderAuthors = Object.fromEntries( + orderRows.map(({ row }) => [row.order_id, row.order_author]), + ); + + expect(orderAuthors).toEqual({ + "order-1": "maker-alpha", + "order-2": "funding-executor", + "order-3": "", + }); + } finally { + await archiver.close(); + await forwarder.close(); + } + }); +}); + describe("withdrawal observation tracker", () => { function shouldArchive( tracker: WithdrawalObservationTracker,