diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index 119f8363d6..5d0a516b1a 100644 --- a/gateway/platforms/webhook.py +++ b/gateway/platforms/webhook.py @@ -657,6 +657,9 @@ class WebhookAdapter(BasePlatformAdapter): route_name, route_config, profile, error_response = self._resolve_route(request) if error_response is None: raw_body, error_response = await self._read_authenticated_body(request, route_name, route_config) + if error_response is None: # a parked profile is retried, before any delivery id is recorded + from gateway.platforms.webhook_ingress import parked_profile_refusal + error_response = parked_profile_refusal(self.gateway_runner, profile) if error_response is not None: return error_response # Rate limiting (after auth) @@ -735,15 +738,18 @@ class WebhookAdapter(BasePlatformAdapter): "delivery_id": delivery_id}, status=202) def _spawn_agent_run(self, payload: Any, prompt: str, delivery_id: str, now: float, *, route_config: dict, - route_name: str, profile, event_type: str) -> asyncio.Task: + route_name: str, profile, event_type: str, superseded=()) -> asyncio.Task: """Coalesced-path dispatch (settled timer / disconnect flush): admit the merged event as a task so ``WebhookCoalescer.flush`` can await the hand-off. The HTTP ack for these deliveries was - already ``coalesced``; an admission failure is logged, never retried by the producer.""" + already ``coalesced``: an admission failure is logged and gives back the batch's delivery ids + (*superseded* too), so a resend of any of them is processed, not answered as a duplicate.""" async def _admit_or_log(): outcome = await self._admit_agent_run(payload, prompt, delivery_id, now, route_config=route_config, route_name=route_name, profile=profile, event_type=event_type) if outcome is None: logger.error("[webhook] Coalesced delivery %s on route %s was not admitted", delivery_id, route_name) + for released in (*superseded, delivery_id): + self._seen_deliveries.pop(_WebhookDeliveryIdentity.from_parts(profile, route_name, released), None) task = asyncio.create_task(_admit_or_log()) self._background_tasks.add(task) task.add_done_callback(self._background_tasks.discard) diff --git a/gateway/platforms/webhook_coalesce.py b/gateway/platforms/webhook_coalesce.py index 9739ef0d9e..803b8598ce 100644 --- a/gateway/platforms/webhook_coalesce.py +++ b/gateway/platforms/webhook_coalesce.py @@ -54,6 +54,7 @@ class PendingEvent: first_at: float count: int dispatch_kwargs: dict[str, Any] + superseded: tuple[str, ...] = () # delivery ids of the earlier events this one replaced def prompt_with_note(self) -> str: if self.count <= 1: @@ -101,8 +102,10 @@ class WebhookCoalescer: first_at = existing.first_at if existing else now count = existing.count + 1 if existing else 1 dispatch_kwargs = {**dispatch_kwargs, "route_name": route_name, "event_type": event_type} + superseded = (*existing.superseded, existing.delivery_id) if existing else () self._pending[group_key] = PendingEvent(payload=payload, prompt=prompt, delivery_id=delivery_id, - first_at=first_at, count=count, dispatch_kwargs=dispatch_kwargs) + first_at=first_at, count=count, dispatch_kwargs=dispatch_kwargs, + superseded=superseded) if existing is not None: logger.info("[webhook] coalesced delivery %s superseded by %s (group=%s, %d events)", existing.delivery_id, delivery_id, group_key, count) @@ -129,7 +132,7 @@ class WebhookCoalescer: logger.info("[webhook] coalesce settled group=%s events=%d delivery=%s", group_key, entry.count, entry.delivery_id) return self._dispatch(entry.payload, entry.prompt_with_note(), entry.delivery_id, time.time(), - **entry.dispatch_kwargs) + **entry.dispatch_kwargs, superseded=entry.superseded) async def flush(self) -> None: """Dispatch every pending group now and wait until each run is handed to the runner, so an adapter diff --git a/gateway/platforms/webhook_ingress.py b/gateway/platforms/webhook_ingress.py index 9d5e464653..57ac01c3c3 100644 --- a/gateway/platforms/webhook_ingress.py +++ b/gateway/platforms/webhook_ingress.py @@ -27,18 +27,42 @@ async def producer_scope(adapter, event): else: runtime_home = runner._admit_primary_source(source, home) if runtime_home is None: + _refuse_parked(runner, getattr(source, '_rejected_route_profile', None)) source.profile_route_rejected = True runtime_home = home source._authorization_profile_home = home # The routed profile's own ledger admits the delivery; an unserved route is refused. authority = authority_for_home(runner, runtime_home) if authority is None: + _refuse_parked(runner, getattr(source, 'profile', None) or profile) raise RuntimeStoreError('profile_mismatch') with native_callback(runner, event, home, profile): async with _async_profile_runtime_scope(runtime_home): yield authority +def _refuse_parked(runner, profile): + """A delivery routed to a parked profile is retried, never refused for good: a rescan serves the + profile again, so it is ``runtime_draining`` (no relay ACK; the connector redelivers), not the + permanent ``profile_mismatch`` of a route to a profile this gateway does not serve.""" + from gateway.run_runtime import profile_parked + if profile_parked(runner, profile): + raise RuntimeStoreError('runtime_draining') + + +def parked_profile_refusal(runner, profile): + """A retryable 503 for a webhook to a parked ``/p//``, else None. Answered before the + delivery id is recorded: coalesced, ``deliver_only`` and ``cron_job`` routes act before (or + without) an admission, so the in-memory id is their only replay guard, and a recorded id would + answer the provider's retry as a duplicate once the profile is served again.""" + from gateway.run_runtime import profile_parked + if not profile_parked(runner, profile): + return None + from aiohttp import web + return web.json_response({'error': f'Profile {profile} is not being served; retry this delivery'}, + status=503, headers={'Retry-After': '30'}) + + async def admit_producer(adapter, event): """Commit a producer delivery; ``None`` when a plugin skipped or consumed it. diff --git a/gateway/profile_routing.py b/gateway/profile_routing.py index a468ab4135..e59e06e22d 100644 --- a/gateway/profile_routing.py +++ b/gateway/profile_routing.py @@ -49,7 +49,14 @@ def _whatsapp_user_chat_ids_match(platform: str, left: Optional[str], right: Opt class ProfileRouteRejected(RuntimeError): - """An explicit route matched a profile this gateway does not serve.""" + """An explicit route matched a profile this gateway does not serve (``profile``, when known).""" + + @property + def profile(self): + return self.args[1] if len(self.args) > 1 else None + + def __str__(self): + return str(self.args[0]) if self.args else '' @dataclass(frozen=True) diff --git a/gateway/run.py b/gateway/run.py index 96e9c7fc39..91b382ba0d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4205,7 +4205,7 @@ class GatewayRunner( logger.warning( "Rejecting profile route %r: target profile %r is not served", matched.name, matched.profile) - raise ProfileRouteRejected(matched.name) + raise ProfileRouteRejected(matched.name, matched.profile) return matched.profile logger.debug( "No profile route matched: platform=%s chat_id=%s thread_id=%s parent_chat_id=%s", diff --git a/gateway/run_runtime.py b/gateway/run_runtime.py index bd17863af7..78204639e1 100644 --- a/gateway/run_runtime.py +++ b/gateway/run_runtime.py @@ -81,6 +81,13 @@ def unpark_profile(runner, name): _record_parked_profiles(parked) +def profile_parked(runner, name): + """True when *name* is parked: not served now, but a rescan serves it again, so its ingress is + temporarily unavailable (retry it), never an invalid route (refuse it for good).""" + descriptor = getattr(runner, 'session_runtime_descriptor', None) or {} + return bool(name) and name in (descriptor.get('parked_profiles') or {}) + + def parked_profile_map(runner): descriptor = getattr(runner, 'session_runtime_descriptor', None) if descriptor is None: diff --git a/gateway/session_identity.py b/gateway/session_identity.py index f462f6f120..ee04005296 100644 --- a/gateway/session_identity.py +++ b/gateway/session_identity.py @@ -98,7 +98,7 @@ def clear_identity(source: Any) -> None: guild's cached voice source is shared by every speaker, and ``profile_routes[].user_id`` routes per speaker). ``source.profile`` is the previous event's routing *result*; left in place it short-circuits the route and the new speaker runs as the old one.""" - for attr in (_IDENTITY_ATTR, "profile_route_rejected", "_authorization_profile_home"): + for attr in (_IDENTITY_ATTR, "profile_route_rejected", "_authorization_profile_home", "_rejected_route_profile"): with suppress(AttributeError): delattr(source, attr) with suppress(AttributeError): @@ -257,6 +257,7 @@ def resolve_identity( else runner._profile_name_for_source(source, adapter_profile=adapter_profile)) except ProfileRouteRejected as exc: source.profile_route_rejected = True + source._rejected_route_profile = exc.profile # a parked target is retried (producer_scope) raise IdentityUnresolved(f"{where}: {exc}") from exc source.profile = routed or adapter_profile diff --git a/tests/gateway/test_parked_profile_ingress.py b/tests/gateway/test_parked_profile_ingress.py new file mode 100644 index 0000000000..342d9b9d76 --- /dev/null +++ b/tests/gateway/test_parked_profile_ingress.py @@ -0,0 +1,258 @@ +"""Ingress for a parked secondary is never acknowledged and then dropped. + +A parked profile (here: its startup Bot recovery failed) comes back through a rescan, so its +traffic is temporarily unavailable, not an invalid route. A relay frame is not acknowledged, so the +connector redelivers it. A webhook gets a retryable 503 before its delivery id is recorded, so the +provider's retry is processed, not answered as a duplicate. +""" +import asyncio +import json +import sqlite3 +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.run import GatewayRunner +from gateway.runtime_ownership import process_ownership +from tests.waiting import async_wait_until + + +def _homes(tmp_path, monkeypatch): + root = tmp_path / '.hermes' + root.mkdir(mode=0o700) + monkeypatch.setattr(Path, 'home', lambda: tmp_path) + monkeypatch.setenv('HERMES_HOME', str(root)) + import hermes_state + monkeypatch.setattr(hermes_state, 'DEFAULT_DB_PATH', hermes_state._IMPORT_DEFAULT_DB_PATH) + beta = root / 'profiles' / 'beta' + beta.mkdir(parents=True, mode=0o700) + (beta / 'config.yaml').write_text('{}') + return [('default', root), ('beta', beta)] + + +def _admissions(home): + with sqlite3.connect(Path(home) / 'state.db') as conn: + return conn.execute('SELECT COUNT(*) FROM session_admissions').fetchone()[0] + + +async def _runner_with_beta_parked(tmp_path, monkeypatch, homes, **config_fields): + """Multiplexed runner whose secondary ``beta`` fails startup Bot recovery once and is parked.""" + from gateway import run + from gateway import session_bot + from gateway.run_runtime import initialize_gateway_runtime, parked_profile_map, start_gateway_runtime_api + monkeypatch.setattr(run, '_load_gateway_config', lambda *a: {'platform_toolsets': {'cli': []}}) + config = GatewayConfig(sessions_dir=tmp_path / 'sessions') + config.multiplex_profiles = True + config._runtime_profile_homes = tuple(homes) + for key, value in config_fields.items(): + setattr(config, key, value) + runner = GatewayRunner(config) + beta, real, failed = homes[1][1].resolve(), session_bot.recover_bot_deliveries, [] + + async def recover_once_failing(authority): + if Path(authority.profile_id).resolve() == beta and not failed: + failed.append(authority) + raise OSError('mailbox write failed') + return await real(authority) + + async def api(_runner): + return SimpleNamespace(api_origin='http://127.0.0.1:1') + + monkeypatch.setattr(session_bot, 'recover_bot_deliveries', recover_once_failing) + monkeypatch.setattr('gateway.run_api.start_gateway_api', api) + process_ownership.reserve([home for _, home in homes]) + await initialize_gateway_runtime(runner) + await start_gateway_runtime_api(runner) + assert 'beta' in parked_profile_map(runner) + return runner + + +async def _serve_beta_again(runner, beta): + """The control-socket rescan serves beta again and lifts its park.""" + from gateway.run_runtime import unpark_profile + assert await runner._serve_profile_runtime('beta', beta) is None + unpark_profile(runner, 'beta') + + +def _close(runner, homes): + for authority in list(getattr(runner, 'session_authorities', None) or []): + if authority is not None: + authority.db.close() + for _, home in homes: + process_ownership.release(home) + + +@pytest.mark.asyncio +async def test_a_relay_frame_for_a_parked_profile_is_redelivered_not_acked(tmp_path, monkeypatch): + websockets = pytest.importorskip('websockets') + from gateway.profile_routing import ProfileRoute + from gateway.relay.adapter import RelayAdapter + from gateway.relay.descriptor import CapabilityDescriptor + from gateway.relay.ws_transport import WebSocketRelayTransport + homes = _homes(tmp_path, monkeypatch) + runner = await _runner_with_beta_parked( + tmp_path, monkeypatch, homes, + profile_routes=[ProfileRoute(name='to-beta', platform='telegram', profile='beta', chat_id='C')]) + descriptor = CapabilityDescriptor(1, 'telegram', 'Fixture', 4096, False, True, False, 'plain', 'chars') + outgoing, ready, adapter = [], asyncio.Queue(), None + frame = json.dumps({'type': 'inbound', 'bufferId': 'buf-1', 'event': { + 'text': 'please do this', 'message_id': 'm1', + 'source': {'platform': 'telegram', 'chat_id': 'C', 'chat_type': 'group', 'user_id': 'U'}}}) + '\n' + + async def connector(ws): + await ws.recv() # hello + await ws.send(json.dumps({'type': 'descriptor', 'descriptor': json.loads(descriptor.to_json())}) + '\n') + await ready.put(ws) + async for line in ws: + outgoing.append(json.loads(line)) + + from gateway.platforms import webhook_ingress + real_admit, attempts = webhook_ingress.admit_producer, [] + + async def recorded(adapter_, event): + try: + receipt = await real_admit(adapter_, event) + except Exception as exc: + attempts.append(exc) + raise + attempts.append(receipt) + return receipt + + monkeypatch.setattr(webhook_ingress, 'admit_producer', recorded) + acks = lambda: [f for f in outgoing if f.get('type') == 'inbound_ack' and f.get('bufferId') == 'buf-1'] + try: + async with websockets.serve(connector, '127.0.0.1', 0) as relay: + transport = WebSocketRelayTransport( + f'ws://127.0.0.1:{relay.sockets[0].getsockname()[1]}/relay', 'telegram', 'fixture-bot', + gateway_id='fixture-gateway', upgrade_secret='fixture-secret', reconnect=True, + reconnect_backoff_s=.05) + adapter = RelayAdapter(PlatformConfig(enabled=True), descriptor, transport) + adapter.gateway_runner = runner + runner.adapters[Platform.RELAY] = adapter + runner._wire_adapter_handlers(adapter) + assert await adapter.connect() + ws = await asyncio.wait_for(ready.get(), 5) + await ws.send(frame) + await async_wait_until(lambda: attempts, what='the parked delivery reached admission') + from gateway.relay.adapter import _PERMANENT_REFUSALS + refused = attempts[0] + assert getattr(refused, 'reason', None) not in _PERMANENT_REFUSALS, ( + f'a parked profile was refused for good (the frame is acknowledged and lost): {refused!r}') + # The profile is served again and the connector redelivers the unacknowledged frame. + await _serve_beta_again(runner, homes[1][1].resolve()) + await ws.send(frame) + await async_wait_until(acks, what='buf-1 acknowledged after the redelivery') + assert len(acks()) == 1, 'the parked delivery was acknowledged too' + assert _admissions(homes[1][1]) == 1 + finally: + if adapter is not None: + await adapter.disconnect() + _close(runner, homes) + + +def _webhook_request(delivery_id, body, route='events'): + raw = json.dumps(body).encode() + request = MagicMock() + request.headers = {'X-GitHub-Delivery': delivery_id} + request.content_length = len(raw) + request.match_info = {'route_name': route, 'profile': 'beta'} + request.method = 'POST' + + async def read(): + return raw + + request.read = read + return request + + +def _webhook(runner, routes): + from gateway.platforms.webhook import WebhookAdapter + adapter = WebhookAdapter(PlatformConfig(enabled=True, extra={ + 'host': '127.0.0.1', 'port': 0, 'routes': routes, 'rate_limit': 100})) + adapter.gateway_runner = runner + runner.adapters[Platform.WEBHOOK] = adapter + runner._wire_adapter_handlers(adapter) + return adapter + + +def _cancel(adapter): + for task in list(adapter._background_tasks): + task.cancel() + + +@pytest.mark.asyncio +async def test_a_coalesced_webhook_for_a_parked_profile_is_retried_not_lost(tmp_path, monkeypatch): + from gateway.platforms.webhook import _INSECURE_NO_AUTH + homes = _homes(tmp_path, monkeypatch) + beta = homes[1][1].resolve() + runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes) + adapter = _webhook(runner, {'events': {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta', + 'coalesce': {'key': 'id', 'window_seconds': 0.1}}}) + settled = lambda: not adapter._coalescer.pending and not adapter._background_tasks + try: + first = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17})) + assert first.status == 503 and first.headers.get('Retry-After'), (first.status, first.text) + await _serve_beta_again(runner, beta) + retry = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17})) + await async_wait_until(settled, what='the retried batch dispatched') + assert json.loads(retry.text)['status'] == 'coalesced', retry.text + assert _admissions(beta) == 1, 'delivery-1 was never admitted' + finally: + _cancel(adapter) + _close(runner, homes) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('mode', ['deliver_only', 'cron_job']) +async def test_a_non_durable_webhook_for_a_parked_profile_is_refused_before_its_id_is_recorded( + tmp_path, monkeypatch, mode): + """``deliver_only`` and ``cron_job`` act without an admission: the in-memory delivery id is their + only replay guard, so a parked profile must refuse before recording it.""" + from gateway.platforms.webhook import _INSECURE_NO_AUTH + homes = _homes(tmp_path, monkeypatch) + runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes) + route = {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta', mode: True} + if mode == 'deliver_only': + route['deliver'] = 'log' + adapter = _webhook(runner, {'events': route}) + try: + refused = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17})) + assert refused.status == 503 and refused.headers.get('Retry-After'), (refused.status, refused.text) + assert adapter._seen_deliveries == {}, 'the refused delivery id was recorded' + finally: + _cancel(adapter) + _close(runner, homes) + + +@pytest.mark.asyncio +async def test_a_coalesced_batch_whose_admission_fails_releases_its_delivery_ids(tmp_path, monkeypatch): + """The provider got 202 ``coalesced`` for every delivery of the batch. If the flush cannot admit + it, a resend of any of them must be processed, not answered as a duplicate.""" + from gateway.platforms import webhook_ingress + from gateway.platforms.webhook import _INSECURE_NO_AUTH, _WebhookDeliveryIdentity + homes = _homes(tmp_path, monkeypatch) + runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes) + await _serve_beta_again(runner, homes[1][1].resolve()) + adapter = _webhook(runner, {'events': {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta', + 'coalesce': {'key': 'id', 'window_seconds': 0.1}}}) + + async def store_busy(adapter_, event): + raise sqlite3.OperationalError('database is locked') + + monkeypatch.setattr(webhook_ingress, 'admit_producer', store_busy) + settled = lambda: not adapter._coalescer.pending and not adapter._background_tasks + try: + for delivery in ('delivery-1', 'delivery-2'): # one entity: delivery-2 supersedes delivery-1 + accepted = await adapter._handle_webhook(_webhook_request(delivery, {'id': 17})) + assert json.loads(accepted.text)['status'] == 'coalesced', accepted.text + await async_wait_until(settled, what='the batch flushed') + held = [identity for identity in adapter._seen_deliveries + if identity in {_WebhookDeliveryIdentity.from_parts('beta', 'events', d) + for d in ('delivery-1', 'delivery-2')}] + assert held == [], f'a failed batch still holds its delivery ids: {held}' + finally: + _cancel(adapter) + _close(runner, homes)