diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 9de44de..0d154b6 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -6812,3 +6812,66 @@ untouched. 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. + +#### C.8 Post-Merge Long-Soak Stream Session Recovery Closure + +**2026-08-21 status: `SOURCE PASS / IMMUTABLE RUNTIME ACCEPTANCE PENDING`:** + +- Source branches merged cleanly into Data Layer `origin/dev` commit `6b6b345` + and Trading System `origin/dev` commit `be80256`; both merge trees are byte + identical to their tested feature heads. No `dev -> main` release is allowed + by this fact alone. +- Read-only inspection after approximately eight hours of runtime found the + Trading System V2 consumer repeatedly receiving `DATA_STALE`, then + `RATE_LIMITED: consumer request quota is exhausted` for Binance USD-M and OKX + Swap. The shared quota reached 602-604 requests against its unchanged + 600/minute limit. `market_data_service` remained running but grew to about + 212 MiB and no longer held stable long-lived TRADE sessions. +- The source-owned SDK session replaces its current transport iterator during + retry and cursor replacement, while `warmup_then_stream` closes only the + iterator created at initial entry. A replacement iterator can therefore lose + cleanup ownership. Immediate bounded SDK retries then amplify a stream fault + into quota pressure. Raising quota/freshness, dropping events or weakening + Trading System validation is forbidden. +- Approved hotfix scope is limited to explicit iterator ownership in + `WarmupStreamSession`: close the current iterator before replacement, close + the current iterator on context exit, make close idempotent and retain the + last acknowledged cursor. Public V2 models, protobuf, endpoint, provider, + source policy, freshness, quota and V1 contracts stay frozen. +- Required gates are deterministic retry/cursor-replacement/context-exit close + tests; no duplicate/gap and no acknowledgment before consumer commit; the + existing stream SDK/transport/security suite; full network-off Python and + locked Rust regressions; deterministic `qdl_sdk==2.0.0` rebuild and consumer + repin; then an immutable runtime test with bounded real Binance/OKX streams, + request rate below quota, stable memory, fresh cache and V1 fallback intact. +- Runtime rollback remains the already exercised Trading System service-only + `V2_PRIMARY -> V1` route. Source verification does not authorize a container + recreation, Redis mutation, authority CAS, DNSE promotion, alpha startup or + `dev -> main` release. + +**2026-08-21 source hotfix result: `PASS`:** + +- `WarmupStreamSession` now owns exactly one current iterator. Cursor expiry and + retry close the old iterator before replacement; terminal errors close it + before propagating; context exit closes the current replacement and repeated + `aclose()` is idempotent. Cursor restoration and acknowledgment ordering are + unchanged. +- Deterministic recovery regression proves three generations (expired cursor, + transient reconnect and final live iterator) are each closed exactly once and + that an additional close is a no-op. The complete stream SDK suite passed + 15/15, including real gRPC handoff, standby failover, slow-consumer recovery, + signed cursor scope and bar revision behavior. +- Full network-off Data Layer Python regression passed 556 tests with six + explicit environment skips. Production SDK lint and `git diff --check` + passed. Rust source is byte-identical to merged `dev`; no Rust authority, + canonical or provider behavior changed in this SDK-only slice. +- Two independent SDK builds were byte-identical at SHA-256 + `6c1e374153756d1918be03c7efeac2d36c68ef235e46f12035ee59afa462a19a`; + source digest is + `1535f7f5cfb50050dc300a3b65471508ca9e10d4f3bcff0d9a9a9108cc23737e`. + The corresponding release manifest and SBOM are the only artifacts eligible + for the Trading System repin. +- No running container, quota key, cursor, Kafka record, Redis projection, + PostgreSQL row, V1 route or authority state was changed. Runtime acceptance + remains blocked on an immutable image/recreation packet and a bounded + real-provider soak. diff --git a/qdl_sdk/client.py b/qdl_sdk/client.py index ce34d73..4121643 100644 --- a/qdl_sdk/client.py +++ b/qdl_sdk/client.py @@ -160,14 +160,18 @@ def __init__( self._reconnect_attempts = 0 self._telemetry = telemetry self.state_restored = state_restored + self._closed = False def __aiter__(self): return self async def __anext__(self) -> StreamEvent | ControlEvent: + if self._closed or self._events is None: + raise StopAsyncIteration try: event = await self._events.__anext__() except CursorExpiredError: + await self._close_events() self.warmup = await self._fresh_snapshot() self._last_seen_offset = self.warmup.watermark_offset self._events = self._subscribe( @@ -180,6 +184,7 @@ async def __anext__(self) -> StreamEvent | ControlEvent: self.warmup, ) except DataLayerError as error: + await self._close_events() if not error.retryable or self._reconnect_attempts >= self._max_reconnect_attempts: raise self._reconnect_attempts += 1 @@ -227,6 +232,19 @@ def acknowledge(self, event: StreamEvent) -> None: cursor_offset=event.logical_offset, ) + async def aclose(self) -> None: + if self._closed: + return + self._closed = True + await self._close_events() + + async def _close_events(self) -> None: + events = self._events + self._events = None + close = getattr(events, "aclose", None) + if close is not None: + await close() + def _subscribe(self, token: str): return self._stream_transport.subscribe( self.requirement, @@ -482,9 +500,7 @@ async def warmup_then_stream( try: yield session finally: - close = getattr(events, "aclose", None) - if close is not None: - await close() + await session.aclose() async def close(self) -> None: await self.stream_transport.close() diff --git a/tests/test_fund_phase5_stream_sdk.py b/tests/test_fund_phase5_stream_sdk.py index c307172..c572502 100644 --- a/tests/test_fund_phase5_stream_sdk.py +++ b/tests/test_fund_phase5_stream_sdk.py @@ -244,6 +244,7 @@ def load(self, requirement, *, consumer_id): class ScriptedIterator: def __init__(self, values): self.values = list(values) + self.close_calls = 0 def __aiter__(self): return self @@ -257,18 +258,21 @@ async def __anext__(self): return value async def aclose(self): - return None + self.close_calls += 1 class ScriptedStreamTransport: def __init__(self, scripts): self.scripts = list(scripts) self.tokens = [] + self.iterators = [] def subscribe(self, requirement, **kwargs): del requirement self.tokens.append(kwargs["cursor_token"]) - return ScriptedIterator(self.scripts.pop(0)) + iterator = ScriptedIterator(self.scripts.pop(0)) + self.iterators.append(iterator) + return iterator async def close(self): return None @@ -761,6 +765,15 @@ async def test_cursor_expiration_rebuilds_snapshot_and_transient_error_reconnect stream.tokens, ["snapshot-token", "snapshot-token", "token-1"], ) + self.assertEqual( + [iterator.close_calls for iterator in stream.iterators], + [1, 1, 1], + ) + await session.aclose() + self.assertEqual( + [iterator.close_calls for iterator in stream.iterators], + [1, 1, 1], + ) async def test_warmup_applies_realtime_quality_only_to_tail_watermark(self): requirement = DataRequirement(