From 82e21b8714bead16284955dc21d6426aeb1d9733 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 17:15:23 +0000 Subject: [PATCH 1/9] fix(runtime): serialize shared spool initialization --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 60 +++++++++++++++++++++++ qdl/transport/sqlite_spool.py | 20 ++++++-- tests/test_fund_phase2_transport.py | 19 +++++++ 3 files changed, 95 insertions(+), 4 deletions(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 32df41b..2a5434b 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6498,3 +6498,63 @@ After an approved cutover, fence the selected Rust slice, restore the matching Python rollback manifest under a newer authority revision/lease, replay from the last common durable watermark and leave all unrelated venue/feed slices untouched. + +#### C.5 Merged Runtime Ingress Closure For Trading System V2_PRIMARY + +**2026-08-20 status: `APPROVED / PREFLIGHT`:** + +- Approved source is merged `origin/dev` commit `f4a7e1c`; the only eligible + release pair remains Python + `sha256:89e359ecc731d68db7a1814885023e1ff9f0aea793e668b6298109eb463ff91c` + and Rust + `sha256:ab57e015da2fb96ef6e4b2180676e0a41b2cc45b64080e820d6a8f29cdab180a` + from topology revision `be35aa7389a37b31c21cc2689c25873dcfc7e73d`. +- The currently running isolated C.2 stack uses superseded images and private + ingress only. Recreate the isolated V2 project from the final bundle while + preserving its Kafka/state volumes and the live V1 project. Rotate all + candidate identities atomically; only query replicas and stream active/ + passive join external `executor_network` with the frozen aliases. +- Kafka, stable Redis, projector, ingestors and Rust cores remain private. V1 + port `8100`, its Redis, storage and every current consumer remain live during + the V2 recreation. DNSE remains V1-only. +- Acceptance requires final image IDs and non-root users, complete process + health, authenticated mTLS query/stream from `executor_network`, real-provider + Binance/OKX event continuity, bounded queue/lag/resources and no V1 restart or + persistent-volume deletion. Synthetic data may not satisfy this gate. +- Rollback recreates the isolated C.2 ingress from its prior immutable image + pair or leaves V2 stopped while V1 remains authoritative. This operation does + not authorize `RUST_PRIMARY`; authority remains `RUST_SHADOW` until a separate + CAS packet proves terminal-watermark handoff for all twelve slices. +- After ingress acceptance, Trading System may make its already-approved exact + routes `V2_PRIMARY` with V1 fallback. No alpha is started by this packet. + +**2026-08-20 final-stack recreation finding: `BLOCKED BEFORE CONSUMER CUTOVER`:** + +- Final immutable images and external aliases were applied while all Kafka, + stable-state and V1 volumes were preserved. V1 remained healthy. +- Concurrent query/stream startup exposed a shared SQLite schema-initialization + race (`sqlite3.OperationalError: database is locked`). The shared spool is + intentional and cannot be split per replica; initialization needs bounded + busy retry while preserving one cache identity. +- Recreating ephemeral stable Redis while retaining a non-empty durable spool + correctly triggered `ProjectionCacheMismatch`. The existing governed cache + rebuild must replay canonical Kafka into a fresh SQLite/Redis cache before + projector readiness. This is recovery behavior, not permission to discard + Kafka or V1 data. +- Fix and gate the concurrent initialization path, rebuild only the isolated + projection cache through the existing confirmation-token runbook, then repeat + mTLS query/stream and real-provider continuity checks. Trading System remains + V1 until all gates pass. + +**2026-08-20 SQLite startup closure result: `PASS / REBUILD PENDING`:** + +- Shared-spool initialization now uses a 30-second SQLite busy timeout and four + bounded lock-only retries. Non-lock operational errors still fail immediately; + all replicas retain one durable cache identity and integrity check. +- Added an eight-replica simultaneous-open regression. Targeted transport and + cache-rebuild tests passed 25/25. The full network-off Python suite passed + 550 tests with six explicit skips using the final runtime dependencies and a + temporary writable log mount. No provider, V1, Kafka, Redis, order or DB state + was mutated by tests. +- Next gate is a new immutable Python image from this exact commit followed by + the confirmation-token projection rebuild; the Rust binary is unchanged. diff --git a/qdl/transport/sqlite_spool.py b/qdl/transport/sqlite_spool.py index 3febf82..deb32e6 100644 --- a/qdl/transport/sqlite_spool.py +++ b/qdl/transport/sqlite_spool.py @@ -98,18 +98,30 @@ def __init__(self, config: SpoolConfig, *, clock_ns=time.time_ns): self._lock = threading.RLock() config.path.parent.mkdir(parents=True, exist_ok=True) self._connection = sqlite3.connect( - str(config.path), timeout=10.0, isolation_level=None, check_same_thread=False + str(config.path), timeout=30.0, isolation_level=None, check_same_thread=False ) self._connection.row_factory = sqlite3.Row - self._configure() - self._migrate() + self._initialize_schema() self._validate_integrity() + def _initialize_schema(self) -> None: + for attempt in range(4): + try: + self._configure() + self._migrate() + return + except sqlite3.OperationalError as error: + if "locked" not in str(error).lower() or attempt == 3: + raise + if self._connection.in_transaction: + self._connection.rollback() + time.sleep(0.25 * (attempt + 1)) + def _configure(self) -> None: self._connection.execute("PRAGMA journal_mode=WAL") self._connection.execute("PRAGMA synchronous=FULL") self._connection.execute("PRAGMA foreign_keys=ON") - self._connection.execute("PRAGMA busy_timeout=10000") + self._connection.execute("PRAGMA busy_timeout=30000") self._connection.execute("PRAGMA wal_autocheckpoint=1000") self._connection.execute("PRAGMA journal_size_limit=67108864") diff --git a/tests/test_fund_phase2_transport.py b/tests/test_fund_phase2_transport.py index 1cde242..ce0fb35 100644 --- a/tests/test_fund_phase2_transport.py +++ b/tests/test_fund_phase2_transport.py @@ -2,7 +2,9 @@ import tempfile import unittest +from concurrent.futures import ThreadPoolExecutor from pathlib import Path +from threading import Barrier from qdl.transport import ( BackpressureRequired, @@ -80,6 +82,23 @@ def spool(self, **overrides) -> SQLiteDurableSpool: ) return SQLiteDurableSpool(config, clock_ns=self.clock) + def test_concurrent_replicas_share_one_initialized_spool(self): + config = SpoolConfig(path=self.path, min_free_disk_bytes=0) + barrier = Barrier(8) + + def open_replica() -> SQLiteDurableSpool: + barrier.wait() + return SQLiteDurableSpool(config) + + with ThreadPoolExecutor(max_workers=8) as executor: + replicas = list(executor.map(lambda _: open_replica(), range(8))) + try: + self.assertEqual(len({replica.cache_id for replica in replicas}), 1) + self.assertTrue(all(replica.integrity_check() for replica in replicas)) + finally: + for replica in replicas: + replica.close() + def test_commit_restart_replay_and_idempotent_retry(self): with self.spool() as spool: first = spool.append(event(1)) From 319d48b8c9d83d79583519463d42e6a05508d8d0 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 17:26:38 +0000 Subject: [PATCH 2/9] perf(recovery): bound stable projection replay tail --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 25 +++++++++++++++ scripts/rebuild_v2_stable_projection_cache.py | 31 +++++++++++++------ tests/test_phaseb_stable_rebuild.py | 27 ++++++++++++++++ 3 files changed, 73 insertions(+), 10 deletions(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 2a5434b..d973fa7 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6558,3 +6558,28 @@ untouched. was mutated by tests. - Next gate is a new immutable Python image from this exact commit followed by the confirmation-token projection rebuild; the Rust binary is unchanged. + +**2026-08-20 replay-efficiency finding: `FIX IN PROGRESS`:** + +- The governed rebuild reset the projector to the beginning of a 7.26-million + event canonical topic although the spool retains only 10,000 records per + partition. Replay progressed correctly but would spend tens of minutes reading + records guaranteed to be trimmed. The orchestrator was interrupted without + deleting Kafka; five cache users were stopped while ingestors/Rust core kept + capturing approved real-provider bytes. +- Rebuild will atomically reset the inactive projector group to each partition + end and then shift back exactly 10,000 records, matching the spool retention + bound. It must still replay all six partitions, reach lag <=250 for three + samples, rebuild non-empty Redis, and prove fresh events for every approved + Binance/OKX route before query readiness. No synthetic event may satisfy the + runtime gate. + +**2026-08-20 bounded-tail recovery closure: `PASS / RUNTIME REBUILD PENDING`:** + +- Recovery now resets the inactive projector group to latest and shifts back + exactly 10,000 records on each of six canonical partitions, matching spool + retention without deleting any Kafka event. The existing <=250 total-lag, + three-consecutive-sample gate remains unchanged. +- Added exact command/plan regression. Targeted recovery/transport tests passed + 26/26; full network-off Python passed 551 with six explicit skips. No runtime + state changed during these tests. diff --git a/scripts/rebuild_v2_stable_projection_cache.py b/scripts/rebuild_v2_stable_projection_cache.py index 988d327..3de62f4 100755 --- a/scripts/rebuild_v2_stable_projection_cache.py +++ b/scripts/rebuild_v2_stable_projection_cache.py @@ -21,6 +21,7 @@ KAFKA_ADMIN_CONFIG = "/etc/kafka/secrets/admin.properties" EXPECTED_CANONICAL_PARTITIONS = 6 MAX_ACCEPTED_LAG = 250 +REPLAY_TAIL_RECORDS_PER_PARTITION = 10_000 REQUIRED_BOUNDED_LAG_SAMPLES = 3 STOP_SERVICES = ( "projector_v2", @@ -64,6 +65,7 @@ def rebuild_plan(env_file: Path) -> dict[str, object]: "flush_service": "stable_redis", "reset_group": PROJECTOR_GROUP, "reset_topic": CANONICAL_TOPIC, + "replay_tail_records_per_partition": REPLAY_TAIL_RECORDS_PER_PARTITION, "lag_gate": { "expected_partitions": EXPECTED_CANONICAL_PARTITIONS, "max_total_records": MAX_ACCEPTED_LAG, @@ -158,6 +160,24 @@ def _kafka_group(env_file: Path, *arguments: str) -> str: return result.stdout +def _reset_projector_to_bounded_tail(env_file: Path) -> None: + common = ( + "--group", + PROJECTOR_GROUP, + "--topic", + CANONICAL_TOPIC, + "--reset-offsets", + ) + _kafka_group(env_file, *common, "--to-latest", "--execute") + _kafka_group( + env_file, + *common, + "--shift-by", + f"-{REPLAY_TAIL_RECORDS_PER_PARTITION}", + "--execute", + ) + + def _validate_project(env_file: Path) -> None: if not env_file.is_file(): raise FileNotFoundError(f"stable env file does not exist: {env_file}") @@ -320,16 +340,7 @@ def execute_rebuild(env_file: Path, *, timeout_seconds: float) -> dict[str, obje if dbsize != "0": raise RuntimeError("isolated stable Redis did not reset to zero keys") - _kafka_group( - env_file, - "--group", - PROJECTOR_GROUP, - "--topic", - CANONICAL_TOPIC, - "--reset-offsets", - "--to-earliest", - "--execute", - ) + _reset_projector_to_bounded_tail(env_file) ssl_context = _stable_client_ssl_context(env_file) _start_services(env_file, *STREAM_SERVICES) _wait_http( diff --git a/tests/test_phaseb_stable_rebuild.py b/tests/test_phaseb_stable_rebuild.py index 6e56e64..66c988d 100644 --- a/tests/test_phaseb_stable_rebuild.py +++ b/tests/test_phaseb_stable_rebuild.py @@ -15,9 +15,11 @@ PROJECT_NAME, PROJECTOR_GROUP, QUERY_SERVICES, + REPLAY_TAIL_RECORDS_PER_PARTITION, STOP_SERVICES, STREAM_SERVICES, _env_value, + _reset_projector_to_bounded_tail, _stable_client_ssl_context, _start_services, _validate_project, @@ -39,6 +41,10 @@ def test_plan_is_exact_isolated_and_v1_safe(self): self.assertEqual(plan["delete_files"], list(CACHE_FILES)) self.assertEqual(plan["reset_group"], PROJECTOR_GROUP) self.assertEqual(plan["reset_topic"], CANONICAL_TOPIC) + self.assertEqual( + plan["replay_tail_records_per_partition"], + REPLAY_TAIL_RECORDS_PER_PARTITION, + ) self.assertEqual( plan["lag_gate"]["expected_partitions"], EXPECTED_CANONICAL_PARTITIONS, @@ -108,6 +114,27 @@ def test_recovery_starts_roles_without_dependency_traversal(self): with self.assertRaisesRegex(ValueError, "at least one"): _start_services(env) + def test_bounded_tail_reset_matches_spool_partition_retention(self): + env = Path("/tmp/stable.env") + with patch( + "scripts.rebuild_v2_stable_projection_cache._kafka_group" + ) as kafka_group: + _reset_projector_to_bounded_tail(env) + kafka_group.assert_any_call( + env, + "--group", PROJECTOR_GROUP, + "--topic", CANONICAL_TOPIC, + "--reset-offsets", "--to-latest", "--execute", + ) + kafka_group.assert_any_call( + env, + "--group", PROJECTOR_GROUP, + "--topic", CANONICAL_TOPIC, + "--reset-offsets", "--shift-by", + f"-{REPLAY_TAIL_RECORDS_PER_PARTITION}", "--execute", + ) + self.assertEqual(kafka_group.call_count, 2) + def test_lag_parser_requires_real_canonical_partitions(self): output = """GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID stable-projector-v1 md.canonical.v2 0 10 12 2 - - - From 571d697c1aa2d5bb2da5a6dab069346145dc2cb9 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 17:32:58 +0000 Subject: [PATCH 3/9] fix(recovery): preserve sparse feeds in cache rebuild --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 22 +++++++ scripts/rebuild_v2_stable_projection_cache.py | 52 +++++++++++----- tests/test_phaseb_stable_rebuild.py | 61 +++++++++++++++---- 3 files changed, 107 insertions(+), 28 deletions(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index d973fa7..0b6aa07 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6574,6 +6574,28 @@ untouched. Binance/OKX route before query readiness. No synthetic event may satisfy the runtime gate. +**2026-08-20 sparse-feed coverage correction: `FIX IN PROGRESS`:** + +- Real acceptance rejected five-bar warmup after the 10,000-record Kafka-tail + rebuild. A physical Kafka partition mixes dense TRADE with sparse BAR events, + so a record-count tail cannot guarantee BAR coverage even though it is bounded. +- Recovery must instead reset to a 15-minute broker timestamp window, require + all six canonical partitions and reject a bootstrap over one million events + before projector startup. This retains at least five expected 1m BAR closes + independently of trade density while keeping recovery bounded. Exact warmup, + source authority and cursor continuity gates remain unchanged. + +**2026-08-20 sparse-feed recovery closure: `PASS / RUNTIME REBUILD PENDING`:** + +- Recovery now derives a UTC broker timestamp exactly 15 minutes behind apply + time, resets the inactive projector group with `--to-datetime`, then verifies + six partitions and at most one million pending events before starting any + cache writer. Missing partition or oversized replay fails closed. +- Targeted recovery tests passed 11/11, including deterministic timestamp, + missing-partition and oversized-window cases. Full network-off Python passed + 552 tests with six explicit skips. + + **2026-08-20 bounded-tail recovery closure: `PASS / RUNTIME REBUILD PENDING`:** - Recovery now resets the inactive projector group to latest and shifts back diff --git a/scripts/rebuild_v2_stable_projection_cache.py b/scripts/rebuild_v2_stable_projection_cache.py index 3de62f4..3e855c7 100755 --- a/scripts/rebuild_v2_stable_projection_cache.py +++ b/scripts/rebuild_v2_stable_projection_cache.py @@ -7,6 +7,7 @@ import subprocess import time import urllib.request +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Callable, Sequence @@ -21,7 +22,8 @@ KAFKA_ADMIN_CONFIG = "/etc/kafka/secrets/admin.properties" EXPECTED_CANONICAL_PARTITIONS = 6 MAX_ACCEPTED_LAG = 250 -REPLAY_TAIL_RECORDS_PER_PARTITION = 10_000 +REPLAY_LOOKBACK_SECONDS = 15 * 60 +MAX_REPLAY_BOOTSTRAP_RECORDS = 1_000_000 REQUIRED_BOUNDED_LAG_SAMPLES = 3 STOP_SERVICES = ( "projector_v2", @@ -65,7 +67,8 @@ def rebuild_plan(env_file: Path) -> dict[str, object]: "flush_service": "stable_redis", "reset_group": PROJECTOR_GROUP, "reset_topic": CANONICAL_TOPIC, - "replay_tail_records_per_partition": REPLAY_TAIL_RECORDS_PER_PARTITION, + "replay_lookback_seconds": REPLAY_LOOKBACK_SECONDS, + "max_replay_bootstrap_records": MAX_REPLAY_BOOTSTRAP_RECORDS, "lag_gate": { "expected_partitions": EXPECTED_CANONICAL_PARTITIONS, "max_total_records": MAX_ACCEPTED_LAG, @@ -160,22 +163,40 @@ def _kafka_group(env_file: Path, *arguments: str) -> str: return result.stdout -def _reset_projector_to_bounded_tail(env_file: Path) -> None: - common = ( - "--group", - PROJECTOR_GROUP, - "--topic", - CANONICAL_TOPIC, - "--reset-offsets", +def _reset_projector_to_bounded_window( + env_file: Path, *, now: datetime | None = None +) -> dict[str, int | str]: + current = now or datetime.now(timezone.utc) + if current.tzinfo is None: + raise ValueError("replay bootstrap time must be timezone-aware") + start = current.astimezone(timezone.utc) - timedelta( + seconds=REPLAY_LOOKBACK_SECONDS + ) + start_text = start.strftime("%Y-%m-%dT%H:%M:%S.") + ( + f"{start.microsecond // 1000:03d}" ) - _kafka_group(env_file, *common, "--to-latest", "--execute") _kafka_group( env_file, - *common, - "--shift-by", - f"-{REPLAY_TAIL_RECORDS_PER_PARTITION}", - "--execute", + "--group", PROJECTOR_GROUP, + "--topic", CANONICAL_TOPIC, + "--reset-offsets", "--to-datetime", start_text, "--execute", ) + total_records, partitions = parse_canonical_lag(_kafka_group( + env_file, "--group", PROJECTOR_GROUP, "--describe" + )) + if partitions != EXPECTED_CANONICAL_PARTITIONS: + raise RuntimeError("replay bootstrap does not cover every canonical partition") + if total_records > MAX_REPLAY_BOOTSTRAP_RECORDS: + raise RuntimeError( + "replay bootstrap exceeds its bounded event budget: " + f"{total_records}>{MAX_REPLAY_BOOTSTRAP_RECORDS}" + ) + return { + "lookback_seconds": REPLAY_LOOKBACK_SECONDS, + "start_datetime_utc": start_text, + "records": total_records, + "partitions": partitions, + } def _validate_project(env_file: Path) -> None: @@ -340,7 +361,7 @@ def execute_rebuild(env_file: Path, *, timeout_seconds: float) -> dict[str, obje if dbsize != "0": raise RuntimeError("isolated stable Redis did not reset to zero keys") - _reset_projector_to_bounded_tail(env_file) + replay_bootstrap = _reset_projector_to_bounded_window(env_file) ssl_context = _stable_client_ssl_context(env_file) _start_services(env_file, *STREAM_SERVICES) _wait_http( @@ -387,6 +408,7 @@ def execute_rebuild(env_file: Path, *, timeout_seconds: float) -> dict[str, obje "apply": True, "status": "PASS", "canonical_lag": lag, + "replay_bootstrap": replay_bootstrap, "redis_keys": final_size, } diff --git a/tests/test_phaseb_stable_rebuild.py b/tests/test_phaseb_stable_rebuild.py index 66c988d..b26fb8c 100644 --- a/tests/test_phaseb_stable_rebuild.py +++ b/tests/test_phaseb_stable_rebuild.py @@ -3,6 +3,7 @@ import subprocess import tempfile import unittest +from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch @@ -15,11 +16,12 @@ PROJECT_NAME, PROJECTOR_GROUP, QUERY_SERVICES, - REPLAY_TAIL_RECORDS_PER_PARTITION, + MAX_REPLAY_BOOTSTRAP_RECORDS, + REPLAY_LOOKBACK_SECONDS, STOP_SERVICES, STREAM_SERVICES, _env_value, - _reset_projector_to_bounded_tail, + _reset_projector_to_bounded_window, _stable_client_ssl_context, _start_services, _validate_project, @@ -41,9 +43,10 @@ def test_plan_is_exact_isolated_and_v1_safe(self): self.assertEqual(plan["delete_files"], list(CACHE_FILES)) self.assertEqual(plan["reset_group"], PROJECTOR_GROUP) self.assertEqual(plan["reset_topic"], CANONICAL_TOPIC) + self.assertEqual(plan["replay_lookback_seconds"], REPLAY_LOOKBACK_SECONDS) self.assertEqual( - plan["replay_tail_records_per_partition"], - REPLAY_TAIL_RECORDS_PER_PARTITION, + plan["max_replay_bootstrap_records"], + MAX_REPLAY_BOOTSTRAP_RECORDS, ) self.assertEqual( plan["lag_gate"]["expected_partitions"], @@ -114,27 +117,59 @@ def test_recovery_starts_roles_without_dependency_traversal(self): with self.assertRaisesRegex(ValueError, "at least one"): _start_services(env) - def test_bounded_tail_reset_matches_spool_partition_retention(self): + def test_bounded_time_window_covers_sparse_feeds_and_caps_records(self): env = Path("/tmp/stable.env") + describe = """GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID +stable-projector-v1 md.canonical.v2 0 10 20 10 - - - +stable-projector-v1 md.canonical.v2 1 10 20 10 - - - +stable-projector-v1 md.canonical.v2 2 10 20 10 - - - +stable-projector-v1 md.canonical.v2 3 10 20 10 - - - +stable-projector-v1 md.canonical.v2 4 10 20 10 - - - +stable-projector-v1 md.canonical.v2 5 10 20 10 - - - +""" + now = datetime(2026, 8, 20, 17, 30, tzinfo=timezone.utc) with patch( - "scripts.rebuild_v2_stable_projection_cache._kafka_group" + "scripts.rebuild_v2_stable_projection_cache._kafka_group", + side_effect=["", describe], ) as kafka_group: - _reset_projector_to_bounded_tail(env) + result = _reset_projector_to_bounded_window(env, now=now) kafka_group.assert_any_call( env, "--group", PROJECTOR_GROUP, "--topic", CANONICAL_TOPIC, - "--reset-offsets", "--to-latest", "--execute", + "--reset-offsets", "--to-datetime", + "2026-08-20T17:15:00.000", "--execute", ) kafka_group.assert_any_call( - env, - "--group", PROJECTOR_GROUP, - "--topic", CANONICAL_TOPIC, - "--reset-offsets", "--shift-by", - f"-{REPLAY_TAIL_RECORDS_PER_PARTITION}", "--execute", + env, "--group", PROJECTOR_GROUP, "--describe" ) + self.assertEqual(result["records"], 60) + self.assertEqual(result["partitions"], EXPECTED_CANONICAL_PARTITIONS) self.assertEqual(kafka_group.call_count, 2) + def test_bounded_time_window_rejects_missing_partition_or_oversized_replay(self): + env = Path("/tmp/stable.env") + line = "stable-projector-v1 md.canonical.v2 {partition} 0 {lag} {lag} - - -" + missing = "\n".join( + line.format(partition=index, lag=1) for index in range(5) + ) + oversized_lag = (MAX_REPLAY_BOOTSTRAP_RECORDS // 6) + 1 + oversized = "\n".join( + line.format(partition=index, lag=oversized_lag) for index in range(6) + ) + for output, message in ( + (missing, "every canonical partition"), + (oversized, "bounded event budget"), + ): + with self.subTest(message=message), patch( + "scripts.rebuild_v2_stable_projection_cache._kafka_group", + side_effect=["", output], + ): + with self.assertRaisesRegex(RuntimeError, message): + _reset_projector_to_bounded_window( + env, now=datetime(2026, 8, 20, tzinfo=timezone.utc) + ) + def test_lag_parser_requires_real_canonical_partitions(self): output = """GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID stable-projector-v1 md.canonical.v2 0 10 12 2 - - - From 655d2106d01f1c665a0b3fd31a091490e6944e40 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 17:38:57 +0000 Subject: [PATCH 4/9] fix(security): cover V2 stream ingress aliases --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 21 +++++++++++++++++++++ scripts/phase80_generate_tls.sh | 2 +- tests/test_phaseb_stable_deployment.py | 12 ++++++++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 0b6aa07..5480fc7 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6585,6 +6585,27 @@ untouched. independently of trade density while keeping recovery bounded. Exact warmup, source authority and cursor continuity gates remain unchanged. +**2026-08-20 container-network TLS finding: `BLOCKED BEFORE CONSUMER CUTOVER`:** + +- Host-port mTLS/JWT acceptance passed for Binance and OKX, but the same SDK + call over `executor_network` rejected gRPC hostname verification. Compose uses + `qdl-v2-stream-a` and `qdl-v2-stream-b`; the generated stream certificate + covered only `stream_v2_active`, `stream_v2_passive` and `qdl-v2-stream`. +- Add both published aliases to the certificate SAN contract, test the generator, + regenerate a private bundle and rotate the isolated V2 stack atomically. REST + query alias remains valid. Trading System stays V1 until container-network + query/stream acceptance passes with the exact production endpoint names. + +**2026-08-20 ingress SAN closure: `PASS / CERT ROTATION PENDING`:** + +- Stable stream certificate generation now covers `qdl-v2-stream-a` and + `qdl-v2-stream-b` in addition to internal role names and localhost. Query SAN + contract is unchanged. +- TLS/deployment contract tests passed 21/21, including published-alias + regression, common workload identity, RS256 rotation and duplicate-target + fail-closed behavior. No runtime or secret changed during tests. + + **2026-08-20 sparse-feed recovery closure: `PASS / RUNTIME REBUILD PENDING`:** - Recovery now derives a UTC broker timestamp exactly 15 minutes behind apply diff --git a/scripts/phase80_generate_tls.sh b/scripts/phase80_generate_tls.sh index b875ccb..1b94686 100755 --- a/scripts/phase80_generate_tls.sh +++ b/scripts/phase80_generate_tls.sh @@ -56,7 +56,7 @@ issue_certificate stable-trading-system-jwt stable-trading-system-jwt openssl pkey -in "${OUTPUT_DIR}/stable-trading-system-jwt.key" -pubout \ -out "${OUTPUT_DIR}/stable-trading-system-jwt.public.pem" >/dev/null 2>&1 issue_certificate stable-query query_v2_1 "DNS:query_v2_1,DNS:query_v2_2,DNS:qdl-v2-query" -issue_certificate stable-stream stream_v2_active "DNS:stream_v2_active,DNS:stream_v2_passive,DNS:qdl-v2-stream" +issue_certificate stable-stream stream_v2_active "DNS:stream_v2_active,DNS:stream_v2_passive,DNS:qdl-v2-stream,DNS:qdl-v2-stream-a,DNS:qdl-v2-stream-b" printf '%s\n' "${PASSWORD}" >"${OUTPUT_DIR}/key.password" printf '%s\n' "${PASSWORD}" >"${OUTPUT_DIR}/store.password" diff --git a/tests/test_phaseb_stable_deployment.py b/tests/test_phaseb_stable_deployment.py index a21cc30..1d0fd37 100644 --- a/tests/test_phaseb_stable_deployment.py +++ b/tests/test_phaseb_stable_deployment.py @@ -49,6 +49,18 @@ def setUp(self) -> None: effective_at_ns=time.time_ns(), ) + def test_tls_generator_covers_all_published_ingress_aliases(self): + script = (ROOT / "scripts/phase80_generate_tls.sh").read_text( + encoding="utf-8" + ) + for alias in ( + "qdl-v2-query", + "qdl-v2-stream-a", + "qdl-v2-stream-b", + ): + with self.subTest(alias=alias): + self.assertIn(f"DNS:{alias}", script) + def test_initial_authority_scope_is_explicit_and_excludes_dnse(self): expected = { item.binding_id From 4a605fbfe2783507d64819cdfdb1c930833b97d6 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 17:57:22 +0000 Subject: [PATCH 5/9] fix(sdk): fail over standby stream targets --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 43 +++++++ qdl_sdk/transport.py | 92 +++++++++----- tests/test_fund_phase5_stream_sdk.py | 141 ++++++++++++++++++++++ 3 files changed, 245 insertions(+), 31 deletions(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 5480fc7..3eaee44 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6605,6 +6605,49 @@ untouched. regression, common workload identity, RS256 rotation and duplicate-target fail-closed behavior. No runtime or secret changed during tests. +**2026-08-20 active/passive SDK failover finding: `BLOCKED BEFORE CONSUMER CUTOVER`:** + +- The rotated certificate and exact `executor_network` aliases now pass mTLS. + Governed recovery replayed 212,165 real-provider records across all six + canonical partitions, rebuilt 47 Redis keys and converged to total lag 36 for + three samples without touching V1. +- Direct acceptance through the replica currently holding the gateway lease + passed for Binance USD-M and OKX Swap: five final 1m bars, authoritative + source/complete coverage, replay-to-live controls, persistent cursor and exact + `+1` resume. Shared spool timestamps also continued advancing from real + provider records. +- Multi-target acceptance using the frozen order `qdl-v2-stream-a,b` timed out + when `a` was the standby owner. The transport rotates its target after gRPC + `UNAVAILABLE` but currently propagates that retryable error to the outer + session first; this can consume a bounded caller timeout before `b` is opened. +- Close this as an SDK transport defect: retry each unique target at most once + inside the same subscribe generation, preserve the exact cursor and auth + metadata, and expose a retryable dependency error only after every target + fails. Add standby-first, all-target-failed and no-duplicate/no-gap tests, + rerun the full network-off suite, then repeat exact-network real-provider + acceptance. Public V2 schemas and V1 remain unchanged. + +**2026-08-20 active/passive SDK failover closure: `PASS / IMMUTABLE REBUILD PENDING`:** + +- `GrpcStreamTransport` now retries each unique target at most once only when + gRPC returns `UNAVAILABLE` before any response was observed. It preserves the + original cursor, requirement and JWT metadata. Once any control/data response + has been observed, it rotates the preferred target but returns a retryable + error to the session layer so recovery uses the last acknowledged cursor. +- Real gRPC regression starts a standby endpoint before an active endpoint and + proves `REPLAYING -> offsets 1,2 -> LIVE` without duplicate or gap. A second + regression proves two standby endpoints are each attempted exactly once before + `DEPENDENCY_UNAVAILABLE` is exposed. +- Targeted SDK/security tests passed 19/19. The full network-off Python suite + passed 555 tests with six explicit environment skips using disposable tmpfs + logs. An initial full-suite invocation failed only because the read-only source + mount did not provide a writable log path; rerunning with the governed tmpfs + test mount passed completely. +- No running image, V1 route, provider, Kafka record, Redis/DB production state + or Trading System consumer was changed by this code slice. Build an immutable + Python image from the resulting commit and rerun exact-network real-provider + acceptance before consumer cutover. + **2026-08-20 sparse-feed recovery closure: `PASS / RUNTIME REBUILD PENDING`:** diff --git a/qdl_sdk/transport.py b/qdl_sdk/transport.py index 92059f7..95e4caa 100644 --- a/qdl_sdk/transport.py +++ b/qdl_sdk/transport.py @@ -212,37 +212,67 @@ async def subscribe( ("x-qdl-consumer-id", consumer_id), ("x-qdl-purpose", RestQueryTransport._purpose(requirement)), ) - try: - subscribe = self._subscribes[self._target_index] - async for response in subscribe(request, metadata=metadata): - record = response.record - payload = record.WhichOneof("payload") - if payload == "control": - yield ControlEvent( - record.control.code, - record.control.detail, - {"high_watermark": record.control.high_watermark}, - ) - continue - if payload == "event": - yield StreamEvent(record.logical_offset, record.resume_token, record.event) - except grpc.aio.AioRpcError as error: - detail = error.details() or "gRPC stream failed" - if error.code() is grpc.StatusCode.UNAVAILABLE and len(self.targets) > 1: - self._target_index = (self._target_index + 1) % len(self.targets) - self.target = self.targets[self._target_index] - if error.code() is grpc.StatusCode.OUT_OF_RANGE: - raise CursorExpiredError("CURSOR_EXPIRED", detail, retryable=False) from error - if error.code() is grpc.StatusCode.RESOURCE_EXHAUSTED: - raise SlowConsumerError("RATE_LIMITED", detail, retryable=True) from error - if error.code() is grpc.StatusCode.INVALID_ARGUMENT: - raise DataLayerError("CURSOR_INVALID", detail, retryable=False) from error - if error.code() is grpc.StatusCode.PERMISSION_DENIED: - raise DataLayerError("SOURCE_NOT_ALLOWED", detail, retryable=False) from error - if error.code() is grpc.StatusCode.FAILED_PRECONDITION: - code = detail.partition(":")[0] - raise DataLayerError(code or "DATA_NOT_READY", detail, retryable=False) from error - raise DataLayerError("DEPENDENCY_UNAVAILABLE", detail, retryable=True) from error + start_index = self._target_index + for attempt in range(len(self.targets)): + target_index = (start_index + attempt) % len(self.targets) + self._target_index = target_index + self.target = self.targets[target_index] + responses_seen = False + try: + subscribe = self._subscribes[target_index] + async for response in subscribe(request, metadata=metadata): + responses_seen = True + record = response.record + payload = record.WhichOneof("payload") + if payload == "control": + yield ControlEvent( + record.control.code, + record.control.detail, + {"high_watermark": record.control.high_watermark}, + ) + continue + if payload == "event": + yield StreamEvent( + record.logical_offset, + record.resume_token, + record.event, + ) + return + except grpc.aio.AioRpcError as error: + detail = error.details() or "gRPC stream failed" + if ( + error.code() is grpc.StatusCode.UNAVAILABLE + and len(self.targets) > 1 + ): + next_index = (target_index + 1) % len(self.targets) + self._target_index = next_index + self.target = self.targets[next_index] + if not responses_seen and attempt + 1 < len(self.targets): + continue + if error.code() is grpc.StatusCode.OUT_OF_RANGE: + raise CursorExpiredError( + "CURSOR_EXPIRED", detail, retryable=False + ) from error + if error.code() is grpc.StatusCode.RESOURCE_EXHAUSTED: + raise SlowConsumerError( + "RATE_LIMITED", detail, retryable=True + ) from error + if error.code() is grpc.StatusCode.INVALID_ARGUMENT: + raise DataLayerError( + "CURSOR_INVALID", detail, retryable=False + ) from error + if error.code() is grpc.StatusCode.PERMISSION_DENIED: + raise DataLayerError( + "SOURCE_NOT_ALLOWED", detail, retryable=False + ) from error + if error.code() is grpc.StatusCode.FAILED_PRECONDITION: + code = detail.partition(":")[0] + raise DataLayerError( + code or "DATA_NOT_READY", detail, retryable=False + ) from error + raise DataLayerError( + "DEPENDENCY_UNAVAILABLE", detail, retryable=True + ) from error async def close(self) -> None: for channel in self._channels: diff --git a/tests/test_fund_phase5_stream_sdk.py b/tests/test_fund_phase5_stream_sdk.py index cb7051f..c307172 100644 --- a/tests/test_fund_phase5_stream_sdk.py +++ b/tests/test_fund_phase5_stream_sdk.py @@ -36,6 +36,7 @@ ) from qdl.query.v2 import query_pb2 from qdl.replay import GapFreeHandoff, SignedHandoffCursorCodec +from qdl.runtime import GatewayFenced from qdl.consumer import UsageTelemetry from qdl.stream import ( DurableStreamGateway, @@ -361,6 +362,146 @@ async def test_slow_consumer_is_disconnected_without_durable_loss_or_peer_block( await slow.close() await peer.close() + async def test_grpc_multitarget_fails_over_from_standby_without_gap(self): + class StandbyAuthority: + current_epoch = None + + def __init__(self): + self.attempts = 0 + + def assert_active(self, expected_epoch=None): + del expected_epoch + self.attempts += 1 + raise GatewayFenced("test standby") + + standby_authority = StandbyAuthority() + standby_gateway = DurableStreamGateway( + handoff=self.handoff, + sink=self.spool, + max_buffer_events=2, + authority=standby_authority, + ) + standby_server = create_grpc_server( + GrpcMarketDataService( + gateway=standby_gateway, + query_service=None, + snapshot_loader=SnapshotLoader(self.record, self.token), + ), + identity_service=self.identity, + ) + active_server = create_grpc_server( + GrpcMarketDataService( + gateway=self.gateway, + query_service=None, + snapshot_loader=SnapshotLoader(self.record, self.token), + ), + identity_service=self.identity, + ) + standby_port = standby_server.add_insecure_port("127.0.0.1:0") + active_port = active_server.add_insecure_port("127.0.0.1:0") + await standby_server.start() + await active_server.start() + await self.gateway.publish(durable(self.record, 1)) + transport = GrpcStreamTransport( + (f"127.0.0.1:{standby_port}", f"127.0.0.1:{active_port}"), + allow_insecure_loopback=True, + credential_provider=self.credential, + ) + requirement = DataRequirement( + self.record.instrument_uid, + Feed.BAR, + Grade.ALPHA, + "alpha_binance_v1", + interval="1m", + warmup_limit=1, + ) + events = transport.subscribe( + requirement, + consumer_id="alpha-shadow", + cursor_token=self.token, + max_buffer_events=2, + ).__aiter__() + try: + self.assertEqual((await events.__anext__()).code, "REPLAYING") + first = await events.__anext__() + self.assertEqual(first.logical_offset, 1) + self.assertEqual((await events.__anext__()).code, "LIVE") + await self.gateway.publish(durable(self.record, 2)) + second = await events.__anext__() + self.assertEqual(second.logical_offset, 2) + self.assertEqual( + transport.target, + f"127.0.0.1:{active_port}", + ) + self.assertEqual(standby_authority.attempts, 1) + finally: + await transport.close() + await standby_server.stop(grace=0) + await active_server.stop(grace=0) + + async def test_grpc_multitarget_fails_after_each_standby_once(self): + class StandbyAuthority: + current_epoch = None + + def __init__(self): + self.attempts = 0 + + def assert_active(self, expected_epoch=None): + del expected_epoch + self.attempts += 1 + raise GatewayFenced("test standby") + + authorities = (StandbyAuthority(), StandbyAuthority()) + servers = [] + ports = [] + for authority in authorities: + gateway = DurableStreamGateway( + handoff=self.handoff, + sink=self.spool, + max_buffer_events=2, + authority=authority, + ) + server = create_grpc_server( + GrpcMarketDataService( + gateway=gateway, + query_service=None, + snapshot_loader=SnapshotLoader(self.record, self.token), + ), + identity_service=self.identity, + ) + ports.append(server.add_insecure_port("127.0.0.1:0")) + servers.append(server) + await server.start() + transport = GrpcStreamTransport( + tuple(f"127.0.0.1:{port}" for port in ports), + allow_insecure_loopback=True, + credential_provider=self.credential, + ) + requirement = DataRequirement( + self.record.instrument_uid, + Feed.BAR, + Grade.ALPHA, + "alpha_binance_v1", + interval="1m", + warmup_limit=1, + ) + events = transport.subscribe( + requirement, + consumer_id="alpha-shadow", + cursor_token=self.token, + max_buffer_events=2, + ).__aiter__() + try: + with self.assertRaises(DataLayerError) as raised: + await events.__anext__() + self.assertEqual(raised.exception.code, "DEPENDENCY_UNAVAILABLE") + self.assertTrue(raised.exception.retryable) + self.assertEqual(tuple(item.attempts for item in authorities), (1, 1)) + finally: + await transport.close() + for server in servers: + await server.stop(grace=0) + async def test_grpc_emits_backpressure_control_before_slow_consumer_disconnect(self): grpc_service = GrpcMarketDataService( gateway=self.gateway, From f1ad494a45ee5f78d6e4c2c452ccacd849d55b56 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 18:03:29 +0000 Subject: [PATCH 6/9] docs(runtime): certify V2 ingress cutover gate --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 34 +++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 3eaee44..c3e15f9 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6648,6 +6648,40 @@ untouched. Python image from the resulting commit and rerun exact-network real-provider acceptance before consumer cutover. +**2026-08-20 C.5 Data Layer ingress runtime acceptance: `PASS / CONSUMER CUTOVER READY`:** + +- Immutable Python image `qdl-v2-python:2.0.0-4a605fbfe278` is + `sha256:9ba5f4a3419c9b5a71bf9dbc8dc65817956054c8bf6b29be6ff6affb45d9601b`, + runs as `qdl:qdl`, and carries exact revision + `4a605fbfe2783507d64819cdfdb1c930833b97d6` with version `2.0.0`. + The unchanged Rust image remains + `sha256:ab57e015da2fb96ef6e4b2180676e0a41b2cc45b64080e820d6a8f29cdab180a`. +- Only the six Python V2 roles were recreated with the new immutable image. + Kafka, three Rust cores, stable Redis, all durable volumes, active certificate + set, V1 and Trading System were preserved. All six roles report restart count + zero; both query replicas are READY and exactly one stream replica is READY + while its peer is STANDBY. +- Final acceptance ran from the immutable image itself over + `executor_network` and the frozen target order `qdl-v2-stream-a,b`, with + `a` deliberately standby. Binance USD-M and OKX Swap each returned five + authoritative final 1m bars, live provider events, persistent cursors and + exact `+1` resume. Status was PASS; no synthetic event was used. +- A request without a client certificate failed the TLS handshake + (`curl rc=52`). V1 remained HTTP 200 at `/v1/health`, all 16 V1 Binance + shards stayed connected, recent queue-drop delta and Redis publish errors were + zero. Post-acceptance canonical lag was 191 across six partitions, within the + configured 250 bound; bounded Python-role logs contained no new error, + critical, exception, traceback or failure. +- Resource snapshot remained bounded: Python roles used about 39-69 MiB each, + stable Redis about 4 MiB, Rust roles about 25-44 MiB and Kafka replicas about + 435-471 MiB each. No container or persistent volume was deleted. +- A generated but undeployed bundle was rejected because it rotated cursor/HMAC/ + DB secrets during an image-only patch. It was verified unreferenced and removed + exactly; the active tested identity bundle remains intact. Trading System may + now proceed with the separately approved `V2_PRIMARY` plus V1 fallback + market-data cutover. This does not promote `RUST_SHADOW` to `RUST_PRIMARY` + and does not authorize DNSE migration. + **2026-08-20 sparse-feed recovery closure: `PASS / RUNTIME REBUILD PENDING`:** From 192c71b3198f170da42cee7a5c66dab49014f459 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 18:41:50 +0000 Subject: [PATCH 7/9] fix(core): quarantine non-positive trades --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 51 +++++++++++++++++++++ qdl/canonical/trade.py | 11 ++++- rust/qdl-core/src/canonical.rs | 47 +++++++++++++++++-- rust/qdl-realtime-core/src/lib.rs | 31 ++++++++++++- tests/test_v2_stable_multivenue_contract.py | 21 +++++++++ 5 files changed, 154 insertions(+), 7 deletions(-) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index c3e15f9..3f98335 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6703,3 +6703,54 @@ untouched. - Added exact command/plan regression. Targeted recovery/transport tests passed 26/26; full network-off Python passed 551 with six explicit skips. No runtime state changed during these tests. + +#### C.6 Trading System V2 Primary Canonical-Trade Closure + +**2026-08-20 status: `APPROVED / ROOT CAUSE CONFIRMED`:** + +- The immutable Trading System consumer passed authenticated V2 query for + Binance USD-M and OKX Swap, but its long-running TRADE streams repeatedly + failed closed. A bounded isolated probe using the same immutable image, + workload identity and real provider stream reproduced the failure without + sharing runtime cursor state. +- OKX passed 100/100 concurrent trade events. Binance reproduced an + authoritative/execution-eligible canonical trade carrying exact + `price=0` and `quantity=0`; the Trading System projector correctly rejected + it. This proves the defect is canonical semantic validation, not mTLS, JWT, + route quota, Redis, provider availability or consumer retry policy. +- Close the defect at the source-owned domain boundary in both the Python + oracle and Rust core: trade price and quantity must be finite canonical + decimals strictly greater than zero. Invalid provider records must enter the + existing bounded `SEMANTIC_INVALID` quarantine path and must never reach the + canonical topic as execution-eligible data. Do not weaken Trading System + validation or fabricate replacement values. +- Required gates are Python/Rust unit parity for zero and negative values, + Rust realtime-core quarantine behavior, existing golden/contract tests, full + network-off suites, a new immutable Rust image, bounded real-provider + Binance/OKX concurrent stream acceptance and unchanged V1 health. Recreate + only isolated V2 Rust roles necessary to apply the core fix; preserve Kafka, + Redis, projection state, certificates, V1 and every execution service. +- Any post-fix zero/negative canonical trade, continuity gap, duplicate, + restart loop or V1 impact blocks the Trading System cutover. Rust authority + remains `RUST_SHADOW`; this closure does not authorize authority promotion. + + +**2026-08-20 canonical semantic source closure: `PASS / IMMUTABLE BUILD PENDING`:** + +- Python canonical oracle and Rust canonical core now require exact finite trade + price and quantity strictly greater than zero for every venue. Existing BAR, + QUOTE, decimal encoding, event identity and frozen bytes are unchanged. +- Rust realtime-core maps a non-positive provider trade through the existing + atomic `SEMANTIC_INVALID` quarantine path and publishes no canonical record. + No downstream projector validation was weakened and no replacement price or + quantity is generated. +- Targeted evidence passed: Python multivenue contract 8/8, Rust `qdl-core` + 16/16 and Rust `qdl-realtime-core` 11/11. The full network-off Python suite + passed 556 tests with six explicit skips. The full locked Rust workspace test + completed with no failure. The first Python full-suite invocation had four + harness-only permission errors because `/app/logs` was root-owned; rerunning + the unchanged source with the runtime UID/GID-owned tmpfs passed completely. +- Build one immutable Python/Rust image pair from the resulting commit, recreate + only the isolated V2 roles required by the changed core, then require bounded + concurrent real-provider streams with zero invalid trade projection before + resuming the Trading System acceptance drill. diff --git a/qdl/canonical/trade.py b/qdl/canonical/trade.py index 0b5c6e4..c5131b2 100644 --- a/qdl/canonical/trade.py +++ b/qdl/canonical/trade.py @@ -65,6 +65,13 @@ def _decimal(value: Any) -> common_pb2.DecimalValue: return message +def _positive_trade_decimal(value: Any, *, field: str) -> common_pb2.DecimalValue: + parsed = CanonicalDecimal.from_text(str(value)) + if parsed.as_decimal() <= 0: + raise ValueError(f"{field} must be positive") + return _decimal(value) + + def _required(raw: Mapping[str, Any], field: str) -> Any: value = raw.get(field) if value is None or value == "": @@ -164,8 +171,8 @@ def _trade_envelope( raw_capture_id=context.raw_capture_id, trade=market_data_pb2.Trade( native_trade_id=native_trade_id, - price=_decimal(price), - quantity=_decimal(quantity), + price=_positive_trade_decimal(price, field="trade price"), + quantity=_positive_trade_decimal(quantity, field="trade quantity"), aggressor_side=side, is_block_trade=False, is_buyer_maker=is_buyer_maker, diff --git a/rust/qdl-core/src/canonical.rs b/rust/qdl-core/src/canonical.rs index bd68960..a387470 100644 --- a/rust/qdl-core/src/canonical.rs +++ b/rust/qdl-core/src/canonical.rs @@ -1,6 +1,7 @@ use prost::Message; use qdl_contracts::qdl::common::v1::{ - AggressorSide, BarOrigin, BookSide, QualityFlag, QuantityUnit, SourceRole, + decimal_value, AggressorSide, BarOrigin, BookSide, DecimalValue, QualityFlag, QuantityUnit, + SourceRole, }; use qdl_contracts::qdl::marketdata::v2::{ event_envelope, Bar, BarLifecycle, BookLevel, EventEnvelope, OrderBookSnapshot, Quote, Trade, @@ -745,6 +746,21 @@ struct TradeInput { identity_kind: TradeIdentityKind, } +fn parse_positive_trade_decimal(source: &str, field: &str) -> Result { + let value = parse_decimal(source)?; + let positive = match value.coefficient.as_ref() { + Some(decimal_value::Coefficient::Mantissa(value)) => *value > 0, + Some(decimal_value::Coefficient::MantissaText(value)) => { + value != "0" && !value.starts_with('-') + } + None => false, + }; + if !positive { + return Err(format!("{field} must be positive")); + } + Ok(value) +} + fn build_trade(fixture: &TradeFixture, trade: TradeInput) -> Result { let context = &fixture.context; validate_shadow_context(context)?; @@ -802,8 +818,11 @@ fn build_trade(fixture: &TradeFixture, trade: TradeInput) -> Result Result, String> { mod tests { use super::{canonical_bytes, TradeFixture}; + #[test] + fn non_positive_trade_price_and_quantity_fail_closed() { + let fixture_path = format!( + "{}/../../tests/fixtures/phase2/binance_usdm_trade.json", + env!("CARGO_MANIFEST_DIR") + ); + let template: TradeFixture = + serde_json::from_slice(&std::fs::read(fixture_path).expect("read fixture")) + .expect("decode fixture"); + for (field, value, message) in [ + ("p", "0", "trade price must be positive"), + ("p", "-0.01", "trade price must be positive"), + ("q", "0", "trade quantity must be positive"), + ("q", "-0.01", "trade quantity must be positive"), + ] { + let mut fixture = template.clone(); + fixture.raw[field] = serde_json::Value::String(value.into()); + let error = canonical_bytes(&fixture).expect_err("non-positive trade must fail"); + assert_eq!(error, message); + } + } + #[test] fn provider_fixtures_match_python_golden_bytes() { for (fixture_name, golden_name) in [ diff --git a/rust/qdl-realtime-core/src/lib.rs b/rust/qdl-realtime-core/src/lib.rs index 9ca861a..b93d55f 100644 --- a/rust/qdl-realtime-core/src/lib.rs +++ b/rust/qdl-realtime-core/src/lib.rs @@ -549,8 +549,8 @@ mod tests { use qdl_contracts::qdl::common::v1::{QuantityUnit, SourceRole}; use qdl_contracts::qdl::marketdata::v2::{event_envelope, EventEnvelope, TradeIdentityKind}; use qdl_contracts::qdl::provider::v1::{ - CaptureBoundary, QuarantineRecord, RawProviderEnvelope, TransportCompression, - TransportProtocol, + CaptureBoundary, QuarantineReason, QuarantineRecord, RawProviderEnvelope, + TransportCompression, TransportProtocol, }; use qdl_venue_core::ordering::SequencePolicy; use sha2::{Digest, Sha256}; @@ -665,6 +665,33 @@ mod tests { assert_eq!(repeated.duplicates, 1); } + #[test] + fn non_positive_trade_is_quarantined_before_canonical_publish() { + let binding = binding(( + "BINANCE_DIRECT", + "BINANCE", + "USDM", + "PERPETUAL", + "BTCUSDT", + "trade", + "binance_usdm_trade", + "PRIMARY", + SequencePolicy::Monotonic, + )); + for frame in [ + br#"{"s":"BTCUSDT","t":10,"p":"0","q":"0.01","T":3,"m":false}"#.as_slice(), + br#"{"s":"BTCUSDT","t":11,"p":"60000.1","q":"-0.01","T":4,"m":false}"#.as_slice(), + ] { + let mut core = core(binding.clone(), true); + let result = core.process(raw(&binding, frame, 1), 10).unwrap(); + assert!(result.canonical.is_empty()); + assert_eq!(result.quarantines.len(), 1); + let record = + QuarantineRecord::decode(result.quarantines[0].payload.as_slice()).unwrap(); + assert_eq!(record.reason, QuarantineReason::SemanticInvalid as i32); + } + } + #[test] fn transport_replay_is_byte_deterministic_across_fresh_cores() { let binding = binding(( diff --git a/tests/test_v2_stable_multivenue_contract.py b/tests/test_v2_stable_multivenue_contract.py index 0d3053f..15529fc 100644 --- a/tests/test_v2_stable_multivenue_contract.py +++ b/tests/test_v2_stable_multivenue_contract.py @@ -141,6 +141,27 @@ def test_provider_role_and_unknown_identity_fail_closed(self): with self.assertRaisesRegex(ValueError, "quantity unit is undefined"): resolve_quantity_unit(venue="UNKNOWN", market="X", product_type="Y") + def test_non_positive_trade_price_and_quantity_fail_closed_for_all_venues(self): + cases = ( + ("binance_usdm_trade.json", canonicalize_binance_usdm_trade, "p", "q"), + ("okx_trade.json", canonicalize_okx_trade, "px", "sz"), + ("dnse_derivative_trade.json", canonicalize_dnse_trade, "price", "quantity"), + ) + for fixture_name, build, price_field, quantity_field in cases: + fixture = json.loads((FIXTURES / fixture_name).read_text()) + context = TradeContext(**fixture["context"]) + for field, value, message in ( + (price_field, "0", "trade price must be positive"), + (price_field, "-0.01", "trade price must be positive"), + (quantity_field, "0", "trade quantity must be positive"), + (quantity_field, "-0.01", "trade quantity must be positive"), + ): + raw = dict(fixture["raw"]) + raw[field] = value + with self.subTest(fixture=fixture_name, field=field, value=value): + with self.assertRaisesRegex(ValueError, message): + build(raw, context) + class StablePublicPayloadUnitTests(unittest.TestCase): decimal = DecimalValue(coefficient="1", scale=0, source_text="1") From ba26f56dba2364a4f6cb37db29704e50322e9487 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 19:05:10 +0000 Subject: [PATCH 8/9] docs(runtime): record canonical trade closure --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 30 +++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 3f98335..043a6e4 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6754,3 +6754,33 @@ untouched. only the isolated V2 roles required by the changed core, then require bounded concurrent real-provider streams with zero invalid trade projection before resuming the Trading System acceptance drill. + + +**2026-08-20 canonical semantic runtime closure: `PASS`:** + +- Tested commit `192c71bd57e44231cc4386c5969d54515d3d9490` produced immutable + Rust image `sha256:60832a3a6b7fbe0d5eb50de92306905380084e3e9c99d66e78e2343bff93339a` + and Python image + `sha256:45044af0fc771291e99543e039100c8d4321b87e0e80a0d8b59f26e1a05eb475`; + labels carry the exact revision/version and both images run non-root. +- The three realtime Rust cores were rolling-recreated one at a time on the new + digest. Every replica is running with restart count zero. Kafka, ingestors, + projector, Redis, TLS, durable volumes, V1 and Trading System execution roles + were preserved. Rust authority remains `RUST_SHADOW`. +- Real Binance/OKX traffic continued after the rollout. The owner core + quarantined 166 semantically invalid provider records during the observed + window while publishing tens of thousands of valid canonical records; no + invalid trade reached the Trading System after the fix. Other replicas had + zero quarantine for their assigned slices. +- Trading System V2 cursors advanced from Binance TRADE 348395 to 359954, OKX + TRADE 111114 to 114662 and both BAR streams from 88 to 94 across soak and the + rollback drill. Projected trades were authoritative and sub-second fresh; + final 1m BARs remained closed and inside the 180-second execution freshness + bound. +- V1 never restarted and returned HTTP 200 throughout. Building/recreating on + the same host caused one transient V1 recent queue-drop observation of 4,504; + the queue stayed at zero, Redis publish errors stayed zero and the subsequent + five-minute metric returned to recent-drop zero. This is recorded as capacity + evidence; future image builds should remain outside a latency-sensitive + cutover window. Broad-universe V1 health remains non-strict/degraded for its + previously documented unused feeds, while demanded-feed failures are zero. From 0198a9a40e109dbdb2d58fe1d8a50346376e0591 Mon Sep 17 00:00:00 2001 From: BobbyAxerol Date: Thu, 20 Aug 2026 20:08:28 +0000 Subject: [PATCH 9/9] docs(runtime): record V2 consumer backpressure closure --- DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md | 28 +++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 043a6e4..9de44de 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6784,3 +6784,31 @@ untouched. evidence; future image builds should remain outside a latency-sensitive cutover window. Broad-universe V1 health remains non-strict/degraded for its previously documented unused feeds, while demanded-feed failures are zero. + + +#### C.7 Trading System Consumer Backpressure Ownership Closure + +**2026-08-20 status: `PASS / NO DATA LAYER BEHAVIOR CHANGE`:** + +- Trading System bounded diagnostics classified the intermittent post-cutover + failure as `DATA_STALE`, not sequence gap, source transition, mTLS, quota or + canonical semantic corruption. +- Read-only inspection of the latest 10,000 Binance USD-M BTCUSDT canonical + trades measured source-to-receive p99 33.379 ms, canonical projection p99 + 1,095.165 ms and maximum 1,174.830 ms, with zero canonical records over five + seconds. Kafka projector lag was 34 records across six partitions. V1 health + remained `ok`. +- The owner was the Trading System consumer: per-event Redis projection and + per-event durable cursor replacement could not absorb provider bursts. The + consumer now preserves ordered events in bounded 64-item/20 ms Redis batches + and checkpoints only the final offset after successful projection. +- Real-provider acceptance then ran six minutes plus a three-minute + post-rollback soak with zero continuity/reconnect warning. Binance and OKX + projected cache ages stayed below the unchanged five-second execution + contract, and the audited V2 -> V1 -> V2 service-only drill passed. +- No Data Layer source policy, freshness threshold, public contract, Kafka + topic, stable cache, authority record or provider adapter was changed for this + issue. Rust remains `RUST_SHADOW`; DNSE remains V1. The Data Layer Python + desired-image pin still differs from the already accepted running Python-role + image and requires a separate operator packet if those roles are to be + recreated; it is not part of this consumer closure.