From da2d140b3d227ddd827aac0aa922aab1860d5f9e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 3 Jul 2026 01:17:03 +0200 Subject: [PATCH] record live state: ch_writer poison-row/dead_letter quarantine hotfix + violet_decisions DDL entry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- prod/ch_writer.py | 158 ++++++++++++++++----- prod/clickhouse/violet/apply_violet_ddl.py | 2 +- 2 files changed, 125 insertions(+), 35 deletions(-) diff --git a/prod/ch_writer.py b/prod/ch_writer.py index 541c150d..65195560 100644 --- a/prod/ch_writer.py +++ b/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 - try: - existing.add((tid, int(ts_us_s))) - except Exception: - continue + tid, event_key = parts + existing.add((tid, event_key)) 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: - kept_ids: List[int] = [] - kept_rows: List[dict] = [] - duplicate_ids: List[int] = [] - for row_id, payload in items: - tid = str(payload.get("trade_id", "")).strip() - try: - ts_us_val = int(payload.get("ts")) - 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: + seen = set(existing) + kept_ids: List[int] = [] + kept_rows: List[dict] = [] + duplicate_ids: List[int] = [] + for row_id, payload in items: + probe = self._trade_event_key(payload) + if probe is not None and probe in seen: + duplicate_ids.append(row_id) 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 diff --git a/prod/clickhouse/violet/apply_violet_ddl.py b/prod/clickhouse/violet/apply_violet_ddl.py index 0e24c316..84b568de 100644 --- a/prod/clickhouse/violet/apply_violet_ddl.py +++ b/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", }