Merge pull request #89 from usherlabs/feat/archive-sequence-column

Add per-run seq column to strategy_data archive tables

Author: outerlook

schema/clickhouse/strategy_data.sql

@@ -10,6 +10,23 @@
 -- envelope source = "hb_runtime".
 --
 -- NO TTL: replay-critical strategy history, retained indefinitely.
+--
+-- Delivery contract: the producer is AT-MOST-ONCE over a bounded in-memory
+-- queue. It drops the oldest row when the queue is full and discards a batch
+-- after two failed POST attempts, so rows can be lost silently. To make loss
+-- distinguishable from "the event never happened", every row carries `seq`: a
+-- monotonic counter starting at 1, scoped to one (controller_id, run_id) and
+-- shared across ALL tables in this database. Gap detection is therefore a UNION
+-- of these tables filtered by run_id, looking for holes in `seq`; the producer's
+-- heartbeat carries dropped_rows/failed_batches as the aggregate counterpart.
+-- A hole proves an allocated row was lost. The converse does not hold: paths
+-- that skip emission entirely (e.g. blocked Layer12 ticks bypassing the archive
+-- tap) never allocate a `seq`, so their absence leaves no hole.
+--
+-- Adding a column here is a POISON PILL if it ships after the producer. Inserts
+-- use JSONEachRow, so a row carrying a column ClickHouse does not have fails the
+-- whole per-table batch; the forwarder returns non-2xx and the producer drops it.
+-- Deploy the forwarder (which applies this file at startup) BEFORE the producer.
 
 CREATE DATABASE IF NOT EXISTS strategy_data;
 
@@ -28,6 +45,8 @@ CREATE TABLE IF NOT EXISTS strategy_data.policy_evaluation_events
     trading_pair LowCardinality(String),
     market_id String,
     run_id String,
+    -- Per-run gap-detection counter; see the delivery contract at the top.
+    seq UInt64,
 
     policy_epoch String,
     fidelity LowCardinality(String),
@@ -61,6 +80,8 @@ CREATE TABLE IF NOT EXISTS strategy_data.strategy_policy_snapshots
     trading_pair LowCardinality(String),
     market_id String,
     run_id String,
+    -- Per-run gap-detection counter; see the delivery contract at the top.
+    seq UInt64,
 
     snapshot_reason LowCardinality(String),
     policy_epoch String,
@@ -88,6 +109,8 @@ CREATE TABLE IF NOT EXISTS strategy_data.market_identity
     trading_pair LowCardinality(String),
     market_id String,
     run_id String,
+    -- Per-run gap-detection counter; see the delivery contract at the top.
+    seq UInt64,
 
     snapshot_reason LowCardinality(String),
     source_hash String,
@@ -115,6 +138,8 @@ CREATE TABLE IF NOT EXISTS strategy_data.symbol_mapping
     trading_pair LowCardinality(String),
     market_id String,
     run_id String,
+    -- Per-run gap-detection counter; see the delivery contract at the top.
+    seq UInt64,
 
     snapshot_reason LowCardinality(String),
     source_hash String,
@@ -140,6 +165,8 @@ CREATE TABLE IF NOT EXISTS strategy_data.inventory_settlement_events
     trading_pair LowCardinality(String),
     market_id String,
     run_id String,
+    -- Per-run gap-detection counter; see the delivery contract at the top.
+    seq UInt64,
 
     event_kind LowCardinality(String),
     token LowCardinality(String),
@@ -151,3 +178,20 @@ CREATE TABLE IF NOT EXISTS strategy_data.inventory_settlement_events
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(fromUnixTimestamp64Milli(event_time_ms))
 ORDER BY (controller_id, trading_pair, event_time_ms);
+
+-- Backfill the gap-detection counter onto already-created tables. Existing rows
+-- keep seq = 0; only runs that start after the producer ships allocate real values.
+ALTER TABLE strategy_data.policy_evaluation_events
+ADD COLUMN IF NOT EXISTS seq UInt64 AFTER run_id;
+
+ALTER TABLE strategy_data.strategy_policy_snapshots
+ADD COLUMN IF NOT EXISTS seq UInt64 AFTER run_id;
+
+ALTER TABLE strategy_data.market_identity
+ADD COLUMN IF NOT EXISTS seq UInt64 AFTER run_id;
+
+ALTER TABLE strategy_data.symbol_mapping
+ADD COLUMN IF NOT EXISTS seq UInt64 AFTER run_id;
+
+ALTER TABLE strategy_data.inventory_settlement_events
+ADD COLUMN IF NOT EXISTS seq UInt64 AFTER run_id;

test/fixtures/archive_forwarder_envelope.json

@@ -13,13 +13,14 @@
         "exchange": "binance",
         "fidelity": "hb_runtime_policy_clock",
         "lag_ms": 12,
-        "source_cursor": "block:12345680:log:3",
         "market_id": "arb-usdc-015",
         "payload_json": "{\"execution_role\":\"dex_lead\",\"mid_price\":\"1.0234\"}",
         "policy_epoch": "42",
         "run_id": "run-2026-07-02T00-00-00Z",
         "schema_version": "1",
+        "seq": 1,
         "source": "hb_runtime",
+        "source_cursor": "block:12345680:log:3",
         "trading_pair": "ARB-USDC"
       },
       "table": "strategy_data.policy_evaluation_events"
@@ -39,6 +40,7 @@
         "policy_epoch": "42",
         "run_id": "run-2026-07-02T00-00-00Z",
         "schema_version": "1",
+        "seq": 2,
         "snapshot_reason": "content_change",
         "source": "hb_runtime",
         "source_hash": "1ed025b4ae41b0865651beb829e16eb62119ecb867c5148858fbe4bfe35e547e",
@@ -62,6 +64,7 @@
         "reservation_id": "resv-0001",
         "run_id": "run-2026-07-02T00-00-00Z",
         "schema_version": "1",
+        "seq": 3,
         "source": "hb_runtime",
         "token": "ARB",
         "trading_pair": "ARB-USDC",