Merge pull request #87 from usherlabs/feat/order-archive-authorship

Archive the authoring component on order events

Author: outerlook

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'),

src/handlers/execute-action/orders.ts

@@ -155,6 +155,7 @@ async function handleCreateOrder(ctx: ExecuteActionContext): Promise<void> {
 			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<void> {
 			requestedQuantity:
 				resolvedOrderTelemetry.requestedQuantity ?? orderValue.amount,
 			requestedNotional: orderValue.amount * orderValue.price,
+			orderAuthor: orderValue.orderAuthor,
 			...extractOrderTelemetryIds(createOrderParams),
 		};
 		emitOrderExecutionTelemetryInBackground(

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

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

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,

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

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<string, unknown>,
+		) =>
+			({
+				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<string, unknown> }>;
+			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,