Release v2.2.2: bounded recovery and execution proof convergence

.github/workflows/ci.yml

@@ -280,6 +280,7 @@ jobs:
       - name: Start isolated Kafka (KRaft, plaintext, no ACLs)
         run: |
           docker run -d --name kn-kafka --hostname localhost \
+            --label qdl.test.namespace=kn-native-integration \
             -e KAFKA_HEAP_OPTS="-Xmx512m -Xms256m" \
             apache/kafka@sha256:9516fb7634bad307d17c33b589fde9023003b0cb761374f500002b980a3149b9
           for attempt in $(seq 1 60); do
@@ -301,13 +302,41 @@ jobs:
           done
 
       - name: Committed reads, no-commit reader, projector stage A/B, market cache (Kafka + Redis)
+        timeout-minutes: 30
         run: |
+          fault_dir=$(mktemp -d)
+          broker_id=$(docker inspect --format '{{.Id}}' kn-kafka)
+          test "$(docker inspect --format '{{index .Config.Labels "qdl.test.namespace"}}' "$broker_id")" = kn-native-integration
+          # This controller targets the captured immutable CI container ID only.
+          (
+            trap 'docker unpause "$broker_id" >/dev/null 2>&1 || true' EXIT
+            for attempt in $(seq 1 18000); do
+              if test -f "$fault_dir/ready"; then
+                docker pause "$broker_id"
+                printf 'isolated CI broker paused\n' > "$fault_dir/paused"
+                sleep 10
+                docker unpause "$broker_id"
+                printf 'isolated CI broker restored\n' > "$fault_dir/restored"
+                exit 0
+              fi
+              sleep 0.1
+            done
+            echo 'Recovery test did not reach the fault boundary' >&2
+            exit 1
+          ) &
+          controller=$!
+          trap 'kill "$controller" 2>/dev/null || true; wait "$controller" 2>/dev/null || true; docker unpause "$broker_id" >/dev/null 2>&1 || true' EXIT
           docker run --rm --network container:kn-kafka \
             -e QDL_KN_TEST_KAFKA=localhost:9092 \
             -e QDL_KN_TEST_REDIS=redis://localhost:6379/0 \
             -e QDL_KN_TEST_REDIS_EXCLUSIVE=redis://localhost:6380/0 \
+            -e QDL_RECOVERY_FAULT_DIR=/fault \
+            -v "$fault_dir:/fault" \
             -v "$PWD:/workspace" -w /workspace rust:1.82-slim \
             sh -c 'apt-get update -qq && apt-get install -y --no-install-recommends clang cmake libclang-dev libcurl4-openssl-dev libssl-dev libzstd-dev make pkg-config zlib1g-dev >/dev/null && cargo test -p qdl-stream-gateway --test committed_reads --locked -- --ignored && cargo test -p qdl-projector --locked -- --ignored --test-threads=1'
+          wait "$controller"
+          test -s "$fault_dir/receipt.json"
+          cat "$fault_dir/receipt.json"
 
       - name: Remove isolated Kafka and Redis
         if: always()

docker-compose.v2-stable.yml

@@ -55,7 +55,8 @@ x-kafka-env: &kafka-env
   KAFKA_LOG_RETENTION_MS: 86400000
   KAFKA_LOG_SEGMENT_BYTES: 16777216
   KAFKA_LOG_DIRS: /var/lib/kafka/data
-  KAFKA_HEAP_OPTS: -Xms256m -Xmx256m
+  KAFKA_HEAP_OPTS: -Xms512m -Xmx768m
+  KAFKA_JVM_PERFORMANCE_OPTS: -XX:+UseG1GC -XX:MaxGCPauseMillis=100
 
 x-kafka: &kafka
   logging: *logging
@@ -72,14 +73,15 @@ x-kafka: &kafka
     timeout: 5s
     retries: 36
     start_period: 20s
-  # Coordinator recovery exceeded 512 MiB despite a 256 MiB JVM heap. Keep
-  # native/direct-buffer headroom bounded without enlarging the Java heap.
+  # Retain native/direct-buffer headroom outside the bounded 768 MiB heap.
+  # Explicit G1 avoids small-container Serial GC ergonomics.
   # R1.25 (2026-09-17): at 768 MiB the kernel memcg OOM-killer killed kafka1's
   # JVM twice in 70 s once raw ingest reached ~930 rec/s (anon-rss 391 MB, the
   # rest page cache and mmap'd segment indexes, which the memcg charges). Docker
   # showed oom_killed=false because PID 1 is the entrypoint script, so the
   # crash loop was invisible to every health check. 1536 MiB is 2x the
-  # measured working set at 2x the morning's load; the heap stays 256 MiB.
+  # measured working set at 2x the morning's load. The 2026-10-01 heap
+  # correction uses at most half this limit after observed Full-GC pressure.
   mem_limit: 1536m
   cpus: 0.75
 

qdl/data_quality/execution_mark_index.py

@@ -1,93 +1,8 @@
-"""Pure validation for the explicit execution MARK/INDEX quiet contract."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Mapping
-
-
-_VALID_RECENCY_MODES = frozenset({
-    "STRICT_EVENT_SESSION_LIVE",
-    "COMPONENT_SESSION_LIVE",
-})
-_MIN_COMPONENT_CADENCE_MS = 250
-_MAX_COMPONENT_CADENCE_MS = 120_000
-
-
-@dataclass(frozen=True, slots=True)
-class QuietExecutionMarkIndexEvidence:
-    """Bounded session/component ages derived from immutable response labels."""
-
-    session_liveness_ms: int
-    session_checked_age_ms: int
-    component_mark_age_ms: int
-    component_index_age_ms: int
-    component_mark_quiet_after_ms: int
-    component_index_quiet_after_ms: int
-
-
-def validate_quiet_execution_mark_index_evidence(
-    labels: Mapping[str, str],
-    *,
-    at_ns: int,
-    max_session_liveness_ms: int | None,
-) -> QuietExecutionMarkIndexEvidence:
-    """Validate the signed live-view proof without rewriting event lineage.
-
-    Only an execution MARK/INDEX request already selected by its caller reaches
-    this helper. It therefore validates the current session and paired-component
-    cadence that make an old, unchanged source event observable. Missing or
-    malformed evidence always fails closed.
-    """
-
-    if max_session_liveness_ms is None or max_session_liveness_ms < 1:
-        raise ValueError("quiet execution MARK/INDEX contract is incomplete")
-    if (
-        labels.get("event_recency_policy") != "OBSERVE"
-        or labels.get("recency_mode") not in _VALID_RECENCY_MODES
-        or labels.get("provider_session_state") != "LIVE"
-    ):
-        raise ValueError("quiet execution MARK/INDEX session evidence is not live")
-    try:
-        session_liveness_ms = int(labels["provider_session_liveness_ms"])
-        session_checked_at_ns = int(labels["provider_session_checked_at_ns"])
-        component_values = tuple(
-            (
-                name,
-                int(labels[f"component_{name.lower()}_received_at_ns"]),
-                int(labels[f"component_{name.lower()}_quiet_after_ms"]),
-            )
-            for name in ("MARK", "INDEX")
-        )
-    except (KeyError, TypeError, ValueError) as error:
-        raise ValueError("quiet execution MARK/INDEX evidence is malformed") from error
-    if (
-        session_liveness_ms < 0
-        or session_checked_at_ns <= 0
-        or session_checked_at_ns > at_ns
-    ):
-        raise ValueError("quiet execution MARK/INDEX session clock is invalid")
-    session_checked_age_ms = (at_ns - session_checked_at_ns) // 1_000_000
-    if session_liveness_ms + session_checked_age_ms > max_session_liveness_ms:
-        raise ValueError("quiet execution MARK/INDEX provider session exceeded its SLA")
-
-    component_ages: dict[str, tuple[int, int]] = {}
-    for name, receipt_ns, cadence_ms in component_values:
-        if (
-            receipt_ns <= 0
-            or not _MIN_COMPONENT_CADENCE_MS <= cadence_ms <= _MAX_COMPONENT_CADENCE_MS
-            or receipt_ns > at_ns
-        ):
-            raise ValueError("quiet execution MARK/INDEX component evidence is invalid")
-        age_ms = (at_ns - receipt_ns) // 1_000_000
-        if age_ms > cadence_ms:
-            raise ValueError("quiet execution MARK/INDEX component exceeded its cadence")
-        component_ages[name] = (age_ms, cadence_ms)
-    return QuietExecutionMarkIndexEvidence(
-        session_liveness_ms=session_liveness_ms,
-        session_checked_age_ms=int(session_checked_age_ms),
-        component_mark_age_ms=component_ages["MARK"][0],
-        component_index_age_ms=component_ages["INDEX"][0],
-        component_mark_quiet_after_ms=component_ages["MARK"][1],
-        component_index_quiet_after_ms=component_ages["INDEX"][1],
-    )
+"""Compatibility import for the SDK-shared execution MARK/INDEX proof validator."""
+
+from qdl_sdk.execution_quality import (
+    QuietExecutionMarkIndexEvidence as QuietExecutionMarkIndexEvidence,
+)
+from qdl_sdk.execution_quality import (
+    validate_quiet_execution_mark_index_evidence as validate_quiet_execution_mark_index_evidence,
+)

qdl/query/__init__.py

@@ -58,18 +58,6 @@
     QueryBackendError,
     SourceMetadata,
 )
-from qdl.query.service import (
-    BatchItemResult,
-    BatchQueryResult,
-    QueryResult,
-    QueryServiceError,
-    ReferenceBatchItemResult,
-    ReferenceBatchQueryResult,
-    ReadinessItemResult,
-    ReadinessResult,
-    V2QueryService,
-    WarmupResult,
-)
 from qdl.warmup.contracts import (
     IntervalSourcePolicy,
     WarmupSpecification,
@@ -132,3 +120,25 @@
     "WarmupSpecification",
     "WarmupTimeRange",
 ]
+
+
+# SDK contracts also import this package; service loading must not create a cycle.
+_SERVICE_EXPORTS = frozenset({
+    "BatchItemResult", "BatchQueryResult", "QueryResult", "QueryServiceError",
+    "ReferenceBatchItemResult", "ReferenceBatchQueryResult", "ReadinessItemResult",
+    "ReadinessResult", "V2QueryService", "WarmupResult",
+})
+
+
+def __getattr__(name):
+    if name not in _SERVICE_EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    from importlib import import_module
+
+    value = getattr(import_module("qdl.query.service"), name)
+    globals()[name] = value
+    return value
+
+
+def __dir__():
+    return sorted(set(globals()) | _SERVICE_EXPORTS)

qdl/reference/execution_live.py

@@ -311,12 +311,19 @@ def _result_from_response(
         ):
             raise ValueError("execution MARK/INDEX live view identity/provenance mismatch")
         fields = self._fields(request, envelope)
+        component_lineage = paired_mark_index_lineage(envelope)
         observed_at_ns = int(envelope.source_event_time_ns)
         labels = [
             ("native_symbol", request.instrument.native_symbol),
             ("execution_view", "STABLE_STREAM_GATEWAY"),
             ("freshness_basis", freshness_basis),
             ("source_event_time_ns", str(observed_at_ns)),
+            ("component_mark_source_event_time_ns", str(component_lineage.mark_source_event_time_ms * 1_000_000)),
+            ("component_index_source_event_time_ns", str(component_lineage.index_source_event_time_ms * 1_000_000)),
+            ("component_mark_received_at_ns", str(component_lineage.mark_received_at_ns)),
+            ("component_index_received_at_ns", str(component_lineage.index_received_at_ns)),
+            ("component_mark_capture_id", component_lineage.mark_capture_id.hex()),
+            ("component_index_capture_id", component_lineage.index_capture_id.hex()),
             ("provider_confirmation_ns", str(int(envelope.received_at_ns))),
             ("connection_generation", str(int(envelope.connection_generation))),
             ("gateway_lease_epoch", str(int(payload["lease_epoch"]))),
@@ -328,11 +335,11 @@ def _result_from_response(
             ),
         ]
         if event_recency_policy is StalePolicy.OBSERVE:
-            labels.extend(self._quiet_labels(
+            labels = list({**dict(labels), **dict(self._quiet_labels(
                 envelope,
                 response,
                 max_session_liveness_ms=max_session_liveness_ms,
-            ))
+            ))}.items())
         observation = ReferenceObservation(
             instrument_uid=request.instrument.instrument_uid,
             instrument_revision=request.instrument.metadata_revision,

qdl/reference/local_mark_index.py

@@ -114,9 +114,22 @@ def attach_cache(self, *, backend, relax_to_alpha: bool = True) -> "CacheRefresh
         return self
 
     async def read(self, *, instrument_uid: str, **kwargs):
+        result = await self._read_cache(instrument_uid=instrument_uid, **kwargs)
+        if (result.record is None and getattr(self._backend, "hot_client", None) is not None
+                and result.reason in {"STALE", "COMPONENT_STALE", "NOT_READY",
+                    "MARKET_CACHE_NOT_READY", "MARKET_CACHE_UNAVAILABLE", "MARKET_CACHE_BACKUP_UNAVAILABLE"}):
+            # Exactly one bounded attempt; identity/fence/gap/session refusals
+            # are not healed by remembering an older view from another replica.
+            return await self._read_cache(instrument_uid=instrument_uid, prefer_hot=True, **kwargs)
+        return result
+
+    async def _read_cache(self, *, instrument_uid: str, prefer_hot: bool = False, **kwargs):
         binding = self._bindings.get(instrument_uid)
         if binding is not None:
-            stored, state = await asyncio.to_thread(self._backend.latest_stored_event, binding)
+            if prefer_hot:
+                stored, state = await asyncio.to_thread(self._backend.latest_stored_event, binding, prefer_hot=True)
+            else:
+                stored, state = await asyncio.to_thread(self._backend.latest_stored_event, binding)
             reason = None if state == "OK" else f"MARKET_CACHE_{state}"
             if stored is not None:
                 try: