record live state: ch_writer poison-row/dead_letter quarantine hotfix + violet_decisions DDL entry

NO content authored here — this commits the long-uncommitted LIVE hotfix from
incident 2026-06-12 (bars_held UInt16 poison jammed 18M rows 1.5d): poison rows
retried individually after CH_POISON_ATTEMPTS, then quarantined to dead_letter;
includes the ids loop-variable bugfix. Uncommitted live code was one checkout
away from loss.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

prod/ch_writer.py

@@ -60,6 +60,13 @@ CH_WAL_TRUNCATE_BYTES = int(os.environ.get("CH_WAL_TRUNCATE_BYTES", str(64 * 102
 CH_VACUUM_MIN_BYTES = int(os.environ.get("CH_VACUUM_MIN_BYTES", str(512 * 1024 * 1024)))
 CH_VACUUM_MIN_FREE_RATIO = float(os.environ.get("CH_VACUUM_MIN_FREE_RATIO", "1.25"))
 CH_VACUUM_MIN_FREE_BYTES = int(os.environ.get("CH_VACUUM_MIN_FREE_BYTES", str(128 * 1024 * 1024)))
+# Poison-row quarantine: a row CH permanently rejects (schema mismatch, bad
+# value) must not head-of-line-block the spool forever. After this many
+# failed attempts the row is retried INDIVIDUALLY (CH proven up first); if it
+# still fails it moves to the dead_letter table for offline repair/replay.
+# Incident 2026-06-12: one trade_events row with bars_held=-106 (UInt16
+# column) was retried 3.2M times and jammed 18M rows behind it for 1.5 days.
+CH_POISON_ATTEMPTS = int(os.environ.get("CH_POISON_ATTEMPTS", "200"))
 
 
 # ─── Timestamp helpers ────────────────────────────────────────────────────────
@@ -202,6 +209,19 @@ class _CHWriter:
         conn.execute(
             "CREATE INDEX IF NOT EXISTS idx_queue_table ON queue(table_name, id)"
         )
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS dead_letter (
+                id INTEGER PRIMARY KEY,
+                table_name TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                created_ts_us INTEGER NOT NULL,
+                attempts INTEGER NOT NULL,
+                dead_ts_us INTEGER NOT NULL,
+                last_error TEXT
+            )
+            """
+        )
         return conn
 
     _PUT_LOCK_TIMEOUT_S: float = 0.1  # max wait before dropping the row
@@ -276,7 +296,7 @@ class _CHWriter:
         with self._lock:
             cur = self._conn.execute(
                 "SELECT id, attempts FROM queue WHERE id IN (%s)" % ",".join("?" for _ in ids),
-                [int(_id) for _ in ids],
+                [int(_id) for _id in ids],
             )
             high_attempts = [(row[0], int(row[1]) + 1) for row in cur.fetchall() if int(row[1]) >= 1000]
             self._conn.executemany(
@@ -291,6 +311,65 @@ class _CHWriter:
                     row_id, attempt,
                 )
 
+    def _ch_alive(self) -> bool:
+        """True iff ClickHouse answers a trivial query — used to distinguish
+        'CH is down' (retry forever, quarantine nothing) from 'CH rejects this
+        specific row' (quarantine after CH_POISON_ATTEMPTS)."""
+        try:
+            req = urllib.request.Request(f"{CH_URL}/?query=SELECT+1", method="GET")
+            req.add_header("X-ClickHouse-User", CH_USER)
+            req.add_header("X-ClickHouse-Key", CH_PASS)
+            with urllib.request.urlopen(req, timeout=3) as resp:
+                return resp.status == 200
+        except Exception:
+            return False
+
+    def _quarantine_poison(self, table: str, items: List[Tuple[int, dict]]) -> None:
+        """After a batch failure, isolate rows CH permanently rejects.
+
+        Only rows whose attempt count exceeds CH_POISON_ATTEMPTS are touched,
+        and only while CH itself is provably up. Each candidate is retried
+        alone: success → delivered+deleted; failure → moved to dead_letter
+        (payload preserved for offline repair/replay) so the spool can drain.
+        """
+        ids = [row_id for row_id, _ in items]
+        if not ids:
+            return
+        with self._lock:
+            cur = self._conn.execute(
+                "SELECT id, attempts FROM queue WHERE id IN (%s)"
+                % ",".join("?" for _ in ids),
+                [int(i) for i in ids],
+            )
+            attempts_by_id = {int(r[0]): int(r[1]) for r in cur.fetchall()}
+        candidates = [
+            (row_id, payload) for row_id, payload in items
+            if attempts_by_id.get(int(row_id), 0) >= CH_POISON_ATTEMPTS
+        ]
+        if not candidates:
+            return
+        if not self._ch_alive():
+            return  # CH outage — nothing is poison, keep retrying the batch
+        for row_id, payload in candidates:
+            if self._post_rows(table, [payload]):
+                self._delete_ids([row_id])
+                continue
+            now = ts_us()
+            with self._lock:
+                self._conn.execute(
+                    "INSERT OR REPLACE INTO dead_letter "
+                    "(id, table_name, payload, created_ts_us, attempts, dead_ts_us, last_error) "
+                    "SELECT id, table_name, payload, created_ts_us, attempts, ?, ? "
+                    "FROM queue WHERE id=?",
+                    (now, "rejected by CH while CH alive (see ch flush WARNINGs)", int(row_id)),
+                )
+                self._conn.execute("DELETE FROM queue WHERE id=?", (int(row_id),))
+            log.error(
+                "ch_writer[%s]: POISON ROW quarantined to dead_letter: id=%s table=%s "
+                "attempts=%d — spool unblocked; repair/replay offline",
+                self._db, row_id, table, attempts_by_id.get(int(row_id), -1),
+            )
+
     def _queue_count(self) -> int:
         with self._lock:
             row = self._conn.execute("SELECT count(*) FROM queue").fetchone()
@@ -433,7 +512,7 @@ class _CHWriter:
             raw = resp.read().decode("utf-8", errors="replace")
             return [line for line in raw.splitlines() if line]
 
-    def _existing_trade_keys(self, rows: List[dict]) -> set[Tuple[str, int]]:
+    def _existing_trade_keys(self, rows: List[dict]) -> set[Tuple[str, str]]:
         trade_ids: List[str] = []
         for row in rows:
             trade_id = row.get("trade_id")
@@ -447,13 +526,14 @@ class _CHWriter:
             return set()
 
         unique = sorted(set(trade_ids))
-        existing: set[Tuple[str, int]] = set()
+        existing: set[Tuple[str, str]] = set()
         chunk_size = 200
         for i in range(0, len(unique), chunk_size):
             chunk = unique[i : i + chunk_size]
             quoted = ",".join("'" + tid.replace("'", "''") + "'" for tid in chunk)
             sql = (
-                "SELECT trade_id, toInt64(toUnixTimestamp64Micro(ts)) "
+                "SELECT trade_id, "
+                "ifNull(nullIf(event_id, ''), concat(toString(toInt64(toUnixTimestamp64Micro(ts))), ':', ifNull(exit_reason, ''))) "
                 f"FROM trade_events WHERE trade_id IN ({quoted}) FORMAT TSV"
             )
             try:
@@ -466,13 +546,24 @@ class _CHWriter:
                 parts = line.split("\t", 1)
                 if len(parts) != 2:
                     continue
-                tid, ts_us_s = parts
+                tid, event_key = parts
-                try:
+                existing.add((tid, event_key))
-                    existing.add((tid, int(ts_us_s)))
-                except Exception:
-                    continue
         return existing
 
+    @staticmethod
+    def _trade_event_key(payload: dict) -> Tuple[str, str] | None:
+        tid = str(payload.get("trade_id", "") or "").strip()
+        if not tid:
+            return None
+        event_id = str(payload.get("event_id", "") or "").strip()
+        if event_id:
+            return (tid, event_id)
+        try:
+            ts_us_val = int(payload.get("ts"))
+        except Exception:
+            ts_us_val = -1
+        return (tid, f"{ts_us_val}:{payload.get('exit_reason', '')}")
+
     def flush_once(self) -> int:
         """
         Drain a single batch from the local spool.
@@ -494,38 +585,37 @@ class _CHWriter:
             rows = [payload for _, payload in items]
             if table == "trade_events":
                 existing = self._existing_trade_keys(rows)
-                if existing:
+                seen = set(existing)
-                    kept_ids: List[int] = []
+                kept_ids: List[int] = []
-                    kept_rows: List[dict] = []
+                kept_rows: List[dict] = []
-                    duplicate_ids: List[int] = []
+                duplicate_ids: List[int] = []
-                    for row_id, payload in items:
+                for row_id, payload in items:
-                        tid = str(payload.get("trade_id", "")).strip()
+                    probe = self._trade_event_key(payload)
-                        try:
+                    if probe is not None and probe in seen:
-                            ts_us_val = int(payload.get("ts"))
+                        duplicate_ids.append(row_id)
-                        except Exception:
-                            ts_us_val = -1
-                        if tid and ts_us_val >= 0 and (tid, ts_us_val) in existing:
-                            duplicate_ids.append(row_id)
-                        else:
-                            kept_ids.append(row_id)
-                            kept_rows.append(payload)
-                    if duplicate_ids:
-                        self._delete_ids(duplicate_ids)
-                        log.warning(
-                            "ch_writer[%s]: dropped %d duplicate trade_events rows by trade_id",
-                            self._db,
-                            len(duplicate_ids),
-                        )
-                    ids = kept_ids
-                    rows = kept_rows
-                    if not rows:
                         continue
+                    kept_ids.append(row_id)
+                    kept_rows.append(payload)
+                    if probe is not None:
+                        seen.add(probe)
+                if duplicate_ids:
+                    self._delete_ids(duplicate_ids)
+                    log.warning(
+                        "ch_writer[%s]: dropped %d duplicate trade_events rows by stable event key",
+                        self._db,
+                        len(duplicate_ids),
+                    )
+                ids = kept_ids
+                rows = kept_rows
+                if not rows:
+                    continue
             ok = self._post_rows(table, rows)
             if ok:
                 delivered += len(rows)
                 self._delete_ids(ids)
             else:
                 self._bump_attempts(ids)
+                self._quarantine_poison(table, list(zip(ids, rows)))
         self._maybe_maintain_spool()
         return delivered
 

prod/clickhouse/violet/apply_violet_ddl.py

@@ -34,7 +34,7 @@ EXPECTED_TABLES = {
     "status_snapshots", "trade_events", "v7_decision_events",
     "adaptive_exit_shadow", "fee_settled_events",
     "sc_bucket_gauge_shadow", "sc_threshold_advisor_shadow",
-    "violet_feed_divergence",
+    "violet_feed_divergence", "violet_decisions",
 }