diff --git a/app/main.py b/app/main.py index 3603e0360f..e8ee348e72 100644 --- a/app/main.py +++ b/app/main.py @@ -137,6 +137,31 @@ def _ensure_web_asset_mime_types() -> None: _ensure_web_asset_mime_types() +async def run_http_bridge_heartbeat_maintenance(proxy_service: Any) -> None: + """Per-replica bridge upkeep driven by the ring heartbeat. + + Both passes are request-independent by design: durable ownership must be + reconciled even on a replica nothing is routing to, and the idle sweep is + otherwise only reached from ``_get_or_create_http_bridge_session``, so a + replica that stops taking bridge requests would keep its idle sessions' + upstream WebSockets open until restart (issue #1354). Each pass is isolated + so one failing cannot skip the other or stop the heartbeat. + """ + if proxy_service is None: + return + for attribute, failure_message in ( + ("reconcile_durable_http_bridge_ownership", "HTTP bridge durable ownership reconciliation failed"), + ("prune_idle_http_bridge_sessions", "HTTP bridge idle sweep failed"), + ): + pass_callable = getattr(proxy_service, attribute, None) + if pass_callable is None: + continue + try: + await pass_callable() + except Exception: + logger.warning(failure_message, exc_info=True) + + def _log_abandoned_lease_release(task: asyncio.Task[None]) -> None: if task.cancelled(): return @@ -543,12 +568,7 @@ async def _heartbeat_only(svc: RingMembershipService, iid: str) -> None: await svc.heartbeat(iid, endpoint_base_url=bridge_endpoint_base_url) except Exception: logger.warning("Ring heartbeat failed", exc_info=True) - proxy_service = getattr(app.state, "proxy_service", None) - if proxy_service is not None and hasattr(proxy_service, "reconcile_durable_http_bridge_ownership"): - try: - await proxy_service.reconcile_durable_http_bridge_ownership() - except Exception: - logger.warning("HTTP bridge durable ownership reconciliation failed", exc_info=True) + await run_http_bridge_heartbeat_maintenance(getattr(app.state, "proxy_service", None)) await refresh_cap_partition(svc.list_active, iid) async def _register_and_heartbeat(svc: RingMembershipService, iid: str) -> None: diff --git a/app/modules/proxy/_service/http_bridge/session_registry.py b/app/modules/proxy/_service/http_bridge/session_registry.py index 58cd3bacd9..7e310f651d 100644 --- a/app/modules/proxy/_service/http_bridge/session_registry.py +++ b/app/modules/proxy/_service/http_bridge/session_registry.py @@ -3,6 +3,7 @@ import asyncio import logging from collections.abc import Mapping +from typing import Any from app.core.clients.proxy import ProxyResponseError from app.core.config.settings import Settings @@ -73,6 +74,21 @@ def _requires_durable_recovery_alias_serialization(session: _HTTPBridgeSession) class _HTTPBridgeSessionRegistryMixin: + async def prune_idle_http_bridge_sessions(self: Any) -> int: + """Run the idle sweep off the request path (issue #1354). + + The sweep is otherwise reached only from + ``_get_or_create_http_bridge_session``, so a replica that stops taking + bridge requests keeps idle sessions' upstream WebSockets open until + restart. Heartbeat-driven, so it runs without traffic or leadership. + """ + async with self._http_bridge_lock: + pruned_sessions = self._prune_http_bridge_sessions_locked() + if not pruned_sessions: + return 0 + self._schedule_http_bridge_session_closes(pruned_sessions, reason="idle_sweep") + return len(pruned_sessions) + async def _register_http_bridge_turn_state( self: _HTTPBridgeServiceProtocol, session: _HTTPBridgeSession, diff --git a/openspec/changes/sweep-idle-bridge-sessions-off-request-path/proposal.md b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/proposal.md new file mode 100644 index 0000000000..ec7e732aac --- /dev/null +++ b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/proposal.md @@ -0,0 +1,21 @@ +## Why + +The HTTP-bridge idle sweep (`_prune_http_bridge_sessions_locked`) is reached from exactly one place: `_get_or_create_http_bridge_session`. It is therefore request-driven, so a replica that stops receiving bridge requests never evicts its idle sessions and holds their upstream WebSockets, registry entries, and durable claims until the process restarts. + +This is the residual leg of issue #1354. Fix (a) (#1476) already made account stream leases turn-scoped, so an idle session releases its cap slot as soon as its last turn detaches — the cap-exhaustion symptom is addressed there. What remains is resource hygiene on a quiet replica: in a multi-replica deployment, traffic moving off one replica leaves its warm sessions pinned indefinitely. + +## What Changes + +- Expose the existing sweep as `prune_idle_http_bridge_sessions()`: take the bridge lock, run the same `_prune_http_bridge_sessions_locked` selection the request path uses, and schedule the pruned sessions' closes through the existing bounded close scheduler. +- Drive it from the per-replica ring heartbeat, alongside the durable-ownership reconcile that already runs there, so the sweep happens on every replica regardless of traffic, leadership, or durable-row cleanup. +- No new selection logic and no new timing: eligibility, the idle TTL that protects a session freshly handed to a request, and the close path are unchanged. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `sticky-session-operations`: idle bridge-session eviction no longer depends on request traffic reaching the replica. diff --git a/openspec/changes/sweep-idle-bridge-sessions-off-request-path/specs/sticky-session-operations/spec.md b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/specs/sticky-session-operations/spec.md new file mode 100644 index 0000000000..99eec97f93 --- /dev/null +++ b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/specs/sticky-session-operations/spec.md @@ -0,0 +1,28 @@ +## ADDED Requirements + +### Requirement: Idle bridge sessions are swept without request traffic + +The system MUST evict idle HTTP-bridge sessions on every replica independently of whether that replica is receiving bridge requests. The sweep MUST reuse the same eligibility the request path applies — a session with pending or queued work, an admission waiter, a handoff in progress, or an unanchored reservation, and a session still inside its idle TTL, MUST NOT be evicted — and MUST close evicted sessions through the existing bounded close path so a slow upstream-reader cancellation cannot block the caller. A sweep failure MUST NOT interrupt the loop that drives it, and MUST NOT prevent the other per-replica bridge upkeep that shares that loop from running. + +#### Scenario: A replica with no bridge traffic still evicts idle sessions + +- **GIVEN** a replica holds an idle bridge session past its idle TTL and receives no further bridge requests +- **WHEN** the sweep runs +- **THEN** the session is detached from the registry and closed, releasing its upstream WebSocket + +#### Scenario: Sweep eligibility matches the request path + +- **GIVEN** a session with pending work whose idle TTL has elapsed, and a session used moments ago +- **WHEN** the sweep runs +- **THEN** neither session is evicted + +#### Scenario: One failing upkeep pass does not skip the other + +- **GIVEN** the durable-ownership reconcile raises on a heartbeat tick +- **WHEN** that tick runs +- **THEN** the idle sweep still runs and the heartbeat loop continues + +#### Scenario: Sweeping an empty registry does nothing + +- **WHEN** the sweep runs with no registered bridge sessions +- **THEN** no session is closed and no cleanup work is scheduled diff --git a/openspec/changes/sweep-idle-bridge-sessions-off-request-path/tasks.md b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/tasks.md new file mode 100644 index 0000000000..1f6968b9ad --- /dev/null +++ b/openspec/changes/sweep-idle-bridge-sessions-off-request-path/tasks.md @@ -0,0 +1,18 @@ +## 1. Sweep entry point + +- [x] 1.1 Add `prune_idle_http_bridge_sessions()` to the bridge session-registry mixin (mixin.py is at its architecture line ratchet): take the bridge lock, reuse `_prune_http_bridge_sessions_locked`, and schedule closes via `_schedule_http_bridge_session_closes` with reason `idle_sweep` + +## 2. Heartbeat wiring + +- [x] 2.1 Extract the heartbeat's bridge upkeep into `run_http_bridge_heartbeat_maintenance()` in `app/main.py` and call the sweep there beside the durable-ownership reconcile, isolating each pass so one failing cannot skip the other or stop the heartbeat + +## 3. Tests + +- [x] 3.1 Idle session is evicted with no request traffic; a freshly-used session is spared +- [x] 3.2 A session with pending work is spared even past its idle TTL +- [x] 3.3 Empty registry is a no-op and schedules no cleanup task +- [x] 3.4 Heartbeat maintenance runs both passes, isolates a failing one, and tolerates a missing service — so removing the wiring fails a test rather than silently restoring the leak + +## 4. Spec + +- [x] 4.1 Record that idle eviction does not depend on request traffic reaching the replica diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 64f3a24077..0b6b78cc77 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -26975,3 +26975,89 @@ async def test_http_bridge_eventless_timeout_signal_drains_after_repeated_sessio await http_bridge_upstream_events_module._record_http_bridge_account_timeout_signal(service, session) await http_bridge_upstream_events_module._record_http_bridge_account_timeout_signal(service, session) record_errors.assert_awaited_once_with(account, 2) + + +@pytest.mark.asyncio +async def test_prune_idle_http_bridge_sessions_evicts_without_request_traffic() -> None: + """The idle sweep is otherwise only reached from the request path, so a + replica that stops taking bridge requests would keep idle sessions' + upstream WebSockets open until restart (issue #1354).""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + idle = _make_bridge_session(key_value="idle-no-traffic") + idle.last_used_at = time.monotonic() - (idle.idle_ttl_seconds + 60.0) + fresh = _make_bridge_session(key_value="fresh-no-traffic") + fresh.last_used_at = time.monotonic() + service._http_bridge_sessions[idle.key] = idle + service._http_bridge_sessions[fresh.key] = fresh + + pruned_count = await service.prune_idle_http_bridge_sessions() + await asyncio.gather(*service._background_cleanup_tasks) + + assert pruned_count == 1 + assert idle.key not in service._http_bridge_sessions + assert fresh.key in service._http_bridge_sessions + assert fresh.closed is False + + +@pytest.mark.asyncio +async def test_prune_idle_http_bridge_sessions_spares_sessions_with_pending_work() -> None: + """The sweep reuses _prune_http_bridge_sessions_locked, so a session with + in-flight work keeps its own lifecycle even past the idle TTL.""" + service = proxy_service.ProxyService(cast(Any, nullcontext())) + busy = _make_bridge_session(key_value="busy-no-traffic", queued_request_count=1) + busy.last_used_at = time.monotonic() - (busy.idle_ttl_seconds + 60.0) + service._http_bridge_sessions[busy.key] = busy + + assert await service.prune_idle_http_bridge_sessions() == 0 + assert busy.key in service._http_bridge_sessions + assert busy.closed is False + + +@pytest.mark.asyncio +async def test_prune_idle_http_bridge_sessions_is_a_noop_on_an_empty_registry() -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + + assert await service.prune_idle_http_bridge_sessions() == 0 + assert not service._background_cleanup_tasks + + +@pytest.mark.asyncio +async def test_heartbeat_maintenance_runs_both_bridge_passes() -> None: + """The ring heartbeat is what makes these request-independent. Asserting the + sweep only through a direct call would still pass if the wiring were + removed, leaving the quiet-replica leak (issue #1354).""" + from app.main import run_http_bridge_heartbeat_maintenance + + proxy_service_double = SimpleNamespace( + reconcile_durable_http_bridge_ownership=AsyncMock(return_value=0), + prune_idle_http_bridge_sessions=AsyncMock(return_value=0), + ) + + await run_http_bridge_heartbeat_maintenance(proxy_service_double) + + proxy_service_double.reconcile_durable_http_bridge_ownership.assert_awaited_once() + proxy_service_double.prune_idle_http_bridge_sessions.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_heartbeat_maintenance_isolates_a_failing_pass() -> None: + """A failing reconcile must not skip the sweep, and neither may stop the + heartbeat loop.""" + from app.main import run_http_bridge_heartbeat_maintenance + + proxy_service_double = SimpleNamespace( + reconcile_durable_http_bridge_ownership=AsyncMock(side_effect=RuntimeError("durable read failed")), + prune_idle_http_bridge_sessions=AsyncMock(return_value=0), + ) + + await run_http_bridge_heartbeat_maintenance(proxy_service_double) + + proxy_service_double.prune_idle_http_bridge_sessions.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_heartbeat_maintenance_tolerates_a_missing_service_or_pass() -> None: + from app.main import run_http_bridge_heartbeat_maintenance + + await run_http_bridge_heartbeat_maintenance(None) + await run_http_bridge_heartbeat_maintenance(SimpleNamespace())