Ops/v2 stable runtime cutover

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,

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")
 

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:

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<DecimalValue, String> {
+    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<EventEnvelope, String> {
     let context = &fixture.context;
     validate_shadow_context(context)?;
@@ -802,8 +818,11 @@ fn build_trade(fixture: &TradeFixture, trade: TradeInput) -> Result<EventEnvelop
         raw_capture_id: context.raw_capture_id.clone(),
         payload: Some(event_envelope::Payload::Trade(Trade {
             native_trade_id: trade.native_trade_id,
-            price: Some(parse_decimal(&trade.price)?),
-            quantity: Some(parse_decimal(&trade.quantity)?),
+            price: Some(parse_positive_trade_decimal(&trade.price, "trade price")?),
+            quantity: Some(parse_positive_trade_decimal(
+                &trade.quantity,
+                "trade quantity",
+            )?),
             aggressor_side: trade.side as i32,
             is_block_trade: false,
             is_buyer_maker: trade.is_buyer_maker,
@@ -823,6 +842,28 @@ pub fn canonical_bytes(fixture: &TradeFixture) -> Result<Vec<u8>, 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 [

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((

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"

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,6 +22,8 @@
 KAFKA_ADMIN_CONFIG = "/etc/kafka/secrets/admin.properties"
 EXPECTED_CANONICAL_PARTITIONS = 6
 MAX_ACCEPTED_LAG = 250
+REPLAY_LOOKBACK_SECONDS = 15 * 60
+MAX_REPLAY_BOOTSTRAP_RECORDS = 1_000_000
 REQUIRED_BOUNDED_LAG_SAMPLES = 3
 STOP_SERVICES = (
     "projector_v2",
@@ -64,6 +67,8 @@ def rebuild_plan(env_file: Path) -> dict[str, object]:
         "flush_service": "stable_redis",
         "reset_group": PROJECTOR_GROUP,
         "reset_topic": CANONICAL_TOPIC,
+        "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,
@@ -158,6 +163,42 @@ def _kafka_group(env_file: Path, *arguments: str) -> str:
     return result.stdout
 
 
+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,
+        "--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:
     if not env_file.is_file():
         raise FileNotFoundError(f"stable env file does not exist: {env_file}")
@@ -320,16 +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")
 
-    _kafka_group(
-        env_file,
-        "--group",
-        PROJECTOR_GROUP,
-        "--topic",
-        CANONICAL_TOPIC,
-        "--reset-offsets",
-        "--to-earliest",
-        "--execute",
-    )
+    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(
@@ -376,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,
     }