diff --git a/routstr/core/terminal_outcome_writer.py b/routstr/core/terminal_outcome_writer.py index 1586d82d..3fca381b 100644 --- a/routstr/core/terminal_outcome_writer.py +++ b/routstr/core/terminal_outcome_writer.py @@ -167,18 +167,8 @@ class TerminalOutcomeWriter: self._task = self._new_task() return True - async def stop(self, *, timeout: float = 5.0, close_coverage: bool = False) -> bool: - closed_day = _utc_day_from_ms(self._now_ms()) - timedelta(days=1) + async def stop(self, *, timeout: float = 5.0) -> bool: if not self._enabled: - if close_coverage: - try: - await self._recover_unattended_coverage(self._now_ms()) - await self._close_coverage(closed_day) - except Exception: - logger.critical( - "Terminal outcome coverage closure failed", exc_info=True - ) - return False return True self._stopping = True self._accepting = False @@ -204,8 +194,6 @@ class TerminalOutcomeWriter: except asyncio.CancelledError: pass try: - if close_coverage: - await self._close_coverage(closed_day) clean = drained and not self._loss_pending and await self._close_run() except (TimeoutError, asyncio.CancelledError): pass @@ -349,15 +337,6 @@ class TerminalOutcomeWriter: await session.commit() return bool(result.rowcount == 1) - async def _close_coverage(self, closed_day: date) -> None: - async with self._session_factory() as session: - await session.exec( # type: ignore[call-overload] - update(TerminalOutcomeEpoch) - .where(col(TerminalOutcomeEpoch.current_slot) == 1) - .values(coverage_end_day=closed_day, current_slot=None) - ) - await session.commit() - async def _close_run(self) -> bool: now = self._now_ms() return await self._update_run( diff --git a/routstr/core/terminal_outcomes.py b/routstr/core/terminal_outcomes.py index 7ea8509a..a1470324 100644 --- a/routstr/core/terminal_outcomes.py +++ b/routstr/core/terminal_outcomes.py @@ -142,9 +142,5 @@ async def start_terminal_outcome_writer(*, serving: bool = False) -> bool: return await terminal_outcome_writer.start(serving=serving) -async def stop_terminal_outcome_writer( - *, timeout: float = 5.0, close_coverage: bool = False -) -> bool: - return await terminal_outcome_writer.stop( - timeout=timeout, close_coverage=close_coverage - ) +async def stop_terminal_outcome_writer(*, timeout: float = 5.0) -> bool: + return await terminal_outcome_writer.stop(timeout=timeout) diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index d7e7f8fe..b88ea3ee 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -254,31 +254,6 @@ async def _track_generic_terminal_stream( raise -async def _track_x_cashu_generic_stream( - stream: AsyncIterator[bytes], - state: _TerminalOutcomeState, - *, - amount: int, - unit: str, -) -> AsyncGenerator[bytes, None]: - try: - async for chunk in stream: - yield chunk - state.mark_success() - except BaseException: - state.mark_transport_failure() - raise - finally: - terminal_context = state.settlement_context(require_success=True) - if terminal_context is not None: - _record_x_cashu_terminal_outcome( - terminal_context, - None, - amount=amount, - unit=unit, - ) - - def _observe_terminal_sse_bytes( state: _TerminalOutcomeState, buffered: bytes, @@ -5169,22 +5144,8 @@ class BaseUpstreamProvider: extra={"path": path, "status_code": response.status_code}, ) - outcome_state = _TerminalOutcomeState( - _terminal_outcome_context( - getattr(request.state, "request_id", None), model_obj - ) - ) return ClosingStreamingResponse( - OwnedUpstreamStream( - _track_x_cashu_generic_stream( - response.aiter_bytes(), - outcome_state, - amount=amount, - unit=unit, - ), - response, - client, - ), + OwnedUpstreamStream(response.aiter_bytes(), response, client), status_code=response.status_code, headers=dict(response.headers), ) @@ -5488,22 +5449,8 @@ class BaseUpstreamProvider: extra={"path": path, "status_code": response.status_code}, ) - outcome_state = _TerminalOutcomeState( - _terminal_outcome_context( - getattr(request.state, "request_id", None), model_obj - ) - ) return ClosingStreamingResponse( - OwnedUpstreamStream( - _track_x_cashu_generic_stream( - response.aiter_bytes(), - outcome_state, - amount=amount, - unit=unit, - ), - response, - client, - ), + OwnedUpstreamStream(response.aiter_bytes(), response, client), status_code=response.status_code, headers=dict(response.headers), ) diff --git a/tests/unit/test_terminal_outcomes.py b/tests/unit/test_terminal_outcomes.py index b1506cae..4b6dbe10 100644 --- a/tests/unit/test_terminal_outcomes.py +++ b/tests/unit/test_terminal_outcomes.py @@ -537,7 +537,7 @@ def test_cashu_retained_msats_uses_exact_persisted_units( assert losses == ["invalid Cashu retained value"] -async def test_collection_pause_excludes_disabled_days_after_restart( +async def test_clean_stop_then_downtime_excludes_unattended_days( ledger: tuple[AsyncEngine, SessionFactory], ) -> None: _, sessions = ledger @@ -546,7 +546,7 @@ async def test_collection_pause_excludes_disabled_days_after_restart( writer = TerminalOutcomeWriter(session_factory=sessions, clock=clock) assert await writer.start() clock.value = _timestamp(day + timedelta(days=3)) - assert await writer.stop(timeout=1, close_coverage=True) + assert await writer.stop(timeout=1) clock.value = _timestamp(day + timedelta(days=6)) assert await writer.start() async with sessions() as session: @@ -746,23 +746,6 @@ async def test_failed_writer_start_cannot_backfill_missed_days_as_zero( assert await writer.stop(timeout=1) -async def test_disable_closes_coverage_after_background_rotation_is_stopped( - ledger: tuple[AsyncEngine, SessionFactory], - monkeypatch: pytest.MonkeyPatch, -) -> None: - _, sessions = ledger - writer = TerminalOutcomeWriter(session_factory=sessions) - assert await writer.start() - original = writer._close_coverage - - async def close_without_racing_writer(day: date) -> None: - assert not writer.running - await original(day) - - monkeypatch.setattr(writer, "_close_coverage", close_without_racing_writer) - assert await writer.stop(timeout=1, close_coverage=True) - - async def test_restart_alongside_live_writer_preserves_continuous_coverage( ledger: tuple[AsyncEngine, SessionFactory], ) -> None: