diff --git a/src/pinky_daemon/agent_registry.py b/src/pinky_daemon/agent_registry.py index 615140d0..c2486931 100644 --- a/src/pinky_daemon/agent_registry.py +++ b/src/pinky_daemon/agent_registry.py @@ -3567,7 +3567,7 @@ def update_buzz_inbound_health( self, agent_name: str, *, - status: str, + status: str | None = None, last_error: str = "", connected_at: float | None = None, liveness_at: float | None = None, @@ -3576,15 +3576,16 @@ def update_buzz_inbound_health( connected = float(connected_at) if connected_at is not None else None liveness = float(liveness_at) if liveness_at is not None else None event = float(event_at) if event_at is not None else None + status_value = str(status or "unknown")[:40] if status is not None else None self._db.execute( """UPDATE buzz_inbound_policies - SET status=?, last_error=?, + SET status=COALESCE(?, status), last_error=?, last_connect_at=COALESCE(MAX(last_connect_at, ?), last_connect_at), last_liveness_at=COALESCE(MAX(last_liveness_at, ?), last_liveness_at), last_event_at=COALESCE(MAX(last_event_at, ?), last_event_at) WHERE agent=?""", ( - str(status or "unknown")[:40], + status_value, str(last_error or "")[:160], connected, liveness, diff --git a/src/pinky_daemon/buzz_inbound.py b/src/pinky_daemon/buzz_inbound.py index 7bbd5845..a505677c 100644 --- a/src/pinky_daemon/buzz_inbound.py +++ b/src/pinky_daemon/buzz_inbound.py @@ -302,7 +302,6 @@ async def process_event( self._registry.mark_buzz_inbound_event_delivered(self.agent_name, event_id) self._registry.update_buzz_inbound_health( self.agent_name, - status="connected", event_at=self._clock(), ) self.stats["delivered"] += 1 @@ -499,25 +498,30 @@ async def _run_authenticated_subscription(self, ws, channels: list[str]) -> None raise BuzzRelayProtocolError("relay_auth_refused") await self._processor.replay_pending() - main_sub = f"pinky-{self._agent_name}-{secrets.token_hex(8)}" since = self._registry.get_buzz_subscription_since(self._agent_name) - await self._send_frame( - ws, - [ - "REQ", - main_sub, - # The production Buzz relay does not live-fanout events to a - # subscription whose filter carries ``since``. Keep the wire - # subscription open-ended and enforce this exact floor in the - # client before any cache, authorization gate, or durable write. - {"kinds": [9, 20002], "#h": channels}, - ], - ) + main_subscription_since: dict[str, int] = {} + connection_token = secrets.token_hex(8) + for index, channel_id in enumerate(channels): + main_sub = f"pinky-{self._agent_name}-{connection_token}-{index}" + main_subscription_since[main_sub] = since + await self._send_frame( + ws, + [ + "REQ", + main_sub, + # The production Buzz relay does not live-fanout events to + # subscriptions carrying ``since`` or multiple ``#h`` + # values. Keep each live REQ open-ended and channel-local, + # then enforce the shared floor in the client before any + # cache, authorization gate, or durable write. + {"kinds": [9, 20002], "#h": [channel_id]}, + ], + ) await self._wait_for_eose( ws, - subscription_id=main_sub, + subscription_ids=set(main_subscription_since), timeout=self._liveness_timeout, - subscription_since={main_sub: since}, + subscription_since=main_subscription_since, ) now = time.time() self._status = "connected" @@ -541,8 +545,7 @@ async def _run_authenticated_subscription(self, ws, channels: list[str]) -> None await self._active_liveness_probe( ws, channels, - main_subscription=main_sub, - main_since=since, + main_subscription_since=main_subscription_since, ) next_heartbeat = loop.time() + self._heartbeat_interval await self._processor.replay_pending() @@ -555,7 +558,7 @@ async def _run_authenticated_subscription(self, ws, channels: list[str]) -> None await self._handle_frame( ws, frame, - subscription_since={main_sub: since}, + subscription_since=main_subscription_since, ) async def _active_liveness_probe( @@ -563,8 +566,7 @@ async def _active_liveness_probe( ws, channels: list[str], *, - main_subscription: str, - main_since: int, + main_subscription_since: dict[str, int], ) -> None: heartbeat_sub = f"pinky-live-{secrets.token_hex(8)}" heartbeat_since = int(time.time()) @@ -584,10 +586,10 @@ async def _active_liveness_probe( try: await self._wait_for_eose( ws, - subscription_id=heartbeat_sub, + subscription_ids={heartbeat_sub}, timeout=self._liveness_timeout, subscription_since={ - main_subscription: main_since, + **main_subscription_since, heartbeat_sub: heartbeat_since, }, ) @@ -611,18 +613,20 @@ async def _wait_for_eose( self, ws, *, - subscription_id: str, + subscription_ids: set[str], timeout: float, subscription_since: dict[str, int], ) -> None: + pending_eose = set(subscription_ids) deadline = asyncio.get_running_loop().time() + timeout - while True: + while pending_eose: remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0: raise asyncio.TimeoutError frame = await self._receive_frame(ws, timeout=remaining) - if len(frame) >= 2 and frame[0] == "EOSE" and frame[1] == subscription_id: - return + if len(frame) >= 2 and frame[0] == "EOSE" and frame[1] in pending_eose: + pending_eose.remove(frame[1]) + continue await self._handle_frame( ws, frame, diff --git a/tests/test_buzz_inbound_poller.py b/tests/test_buzz_inbound_poller.py index dc8eca88..7d9fab5e 100644 --- a/tests/test_buzz_inbound_poller.py +++ b/tests/test_buzz_inbound_poller.py @@ -17,6 +17,7 @@ OWNER = BuzzNostrSigner(bytes.fromhex("33" * 32)) USER = BuzzNostrSigner(bytes.fromhex("44" * 32)) CHANNEL = "00000000-0000-4000-8000-000000000001" +OTHER_CHANNEL = "00000000-0000-4000-8000-000000000002" COMMUNITY = "example" @@ -31,7 +32,7 @@ async def dispatch_pre_authorized(self, agent_name, message): # noqa: ANN001 return True -def _registry(tmp_path, relay_url: str): +def _registry(tmp_path, relay_url: str, *, channels: list[dict] | None = None): store = AgentRegistry( str(tmp_path / "agents.db"), buzz_device_key_path=str(tmp_path / "identity" / ".device_key"), @@ -48,7 +49,7 @@ def _registry(tmp_path, relay_url: str): store.configure_buzz_inbound_owner_control( "barsik", owner_pubkey=OWNER.pubkey, - channels=[{"channel_id": CHANNEL, "label": "#general"}], + channels=channels or [{"channel_id": CHANNEL, "label": "#general"}], approved_users=[{"pubkey": USER.pubkey, "display_name": "Brad"}], owner_actor="ui:admin", ) @@ -70,39 +71,78 @@ async def _authenticate(ws, relay_url: str, captured: list[list]) -> list: return event -class SinceBlindLiveRelayRig: - """Model Buzz: stored queries work, but wire-``since`` disables live push.""" +class LiveFanoutQuirkRelayRig: + """Model Buzz live suppression for wire-``since`` or multi-``#h`` REQs.""" - def __init__(self, *, stored_events: list[dict], live_event: dict) -> None: + def __init__( + self, + *, + channels: list[str], + stored_events: list[dict], + live_events: list[dict], + hold_final_eose: bool = False, + ) -> None: self.relay_url = "" + self.channels = channels self.stored_events = stored_events - self.live_event = live_event + self.live_events = live_events + self.hold_final_eose = hold_final_eose self.captured: list[list] = [] - self.main_filter: dict = {} + self.main_requests: list[list] = [] + self.main_requests_received = asyncio.Event() + self.partial_eose_sent = asyncio.Event() + self.release_final_eose = asyncio.Event() + self.all_eose_sent = asyncio.Event() self.stored_batch_sent = asyncio.Event() - self.release_live_event = asyncio.Event() - self.live_event_pushed = asyncio.Event() + self.release_live_events = asyncio.Event() + self.live_events_pushed = asyncio.Event() + + @staticmethod + def _channel_id(event: dict) -> str: + return next(tag[1] for tag in event["tags"] if tag[0] == "h") async def handler(self, ws) -> None: # noqa: ANN001 await _authenticate(ws, self.relay_url, self.captured) - main = json.loads(await ws.recv()) - self.captured.append(main) - assert main[0] == "REQ" - self.main_filter.update(main[2]) + for _ in self.channels: + main = json.loads(await ws.recv()) + self.captured.append(main) + assert main[0] == "REQ" + self.main_requests.append(main) + self.main_requests_received.set() # Stored-query delivery works for every filter shape on the real - # relay, including filters with ``since``. A no-since main REQ can - # therefore receive the channel's full history before EOSE. + # relay, including filters with ``since`` or multiple ``#h`` values. for event in self.stored_events: - await ws.send(json.dumps(["EVENT", main[1], event])) - await ws.send(json.dumps(["EOSE", main[1]])) + channel_id = self._channel_id(event) + for main in self.main_requests: + if channel_id in main[2].get("#h", []): + await ws.send(json.dumps(["EVENT", main[1], event])) + eose_requests = self.main_requests + if self.hold_final_eose: + eose_requests = self.main_requests[:-1] + for main in eose_requests: + await ws.send(json.dumps(["EOSE", main[1]])) + if self.hold_final_eose: + self.partial_eose_sent.set() + await self.release_final_eose.wait() + await ws.send(json.dumps(["EOSE", self.main_requests[-1][1]])) + self.all_eose_sent.set() self.stored_batch_sent.set() - await self.release_live_event.wait() - # This is the production quirk isolated by the live probe matrix. - if "since" not in self.main_filter: - await ws.send(json.dumps(["EVENT", main[1], self.live_event])) - self.live_event_pushed.set() + await self.release_live_events.wait() + pushed = 0 + for main in self.main_requests: + subscription_filter = main[2] + # These are the two independent production quirks isolated by the + # controlled live probe matrices: either shape suppresses fan-out. + if "since" in subscription_filter or len(subscription_filter.get("#h", [])) != 1: + continue + for event in self.live_events: + if self._channel_id(event) in subscription_filter["#h"]: + await ws.send(json.dumps(["EVENT", main[1], event])) + pushed += 1 + if pushed: + self.live_events_pushed.set() try: while True: @@ -415,7 +455,11 @@ async def test_since_blind_relay_live_push_survives_large_stale_eose_burst( content="live event after EOSE", created_at=subscription_floor, ) - rig = SinceBlindLiveRelayRig(stored_events=stored_burst, live_event=live_event) + rig = LiveFanoutQuirkRelayRig( + channels=[CHANNEL], + stored_events=stored_burst, + live_events=[live_event], + ) async with websockets.serve(rig.handler, "127.0.0.1", 0) as server: port = server.sockets[0].getsockname()[1] @@ -448,7 +492,7 @@ async def notify(_agent, _message): await asyncio.sleep(0.01) assert poller.health["status"] == "connected" - assert rig.main_filter == {"kinds": [9, 20002], "#h": [CHANNEL]} + assert rig.main_requests[0][2] == {"kinds": [9, 20002], "#h": [CHANNEL]} assert poller.health["rejected"] == len(stored_burst) assert poller._processor._recent_ids == set() assert broker.calls == [] @@ -459,8 +503,8 @@ async def notify(_agent, _message): "SELECT last_seen_at FROM buzz_inbound_principals WHERE agent='barsik'" ).fetchall() == [(0.0,), (0.0,)] - rig.release_live_event.set() - await asyncio.wait_for(rig.live_event_pushed.wait(), timeout=2) + rig.release_live_events.set() + await asyncio.wait_for(rig.live_events_pushed.wait(), timeout=2) await asyncio.wait_for(broker.delivered.wait(), timeout=2) poller.stop() await asyncio.wait_for(task, timeout=2) @@ -475,6 +519,108 @@ async def notify(_agent, _message): store.close() +@pytest.mark.asyncio +async def test_live_fanout_uses_one_req_per_channel_and_waits_for_every_eose(tmp_path): + stored_event = USER.sign_event( + kind=9, + tags=[["h", CHANNEL]], + content="stored event before all EOSE", + ) + live_events = [ + USER.sign_event( + kind=9, + tags=[["h", CHANNEL]], + content="live event in general", + ), + USER.sign_event( + kind=9, + tags=[["h", OTHER_CHANNEL]], + content="live event in support", + ), + ] + rig = LiveFanoutQuirkRelayRig( + channels=[CHANNEL, OTHER_CHANNEL], + stored_events=[stored_event], + live_events=live_events, + hold_final_eose=True, + ) + + async with websockets.serve(rig.handler, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + rig.relay_url = f"ws://127.0.0.1:{port}" + store = _registry( + tmp_path, + rig.relay_url, + channels=[ + {"channel_id": CHANNEL, "label": "#general"}, + {"channel_id": OTHER_CHANNEL, "label": "#support"}, + ], + ) + broker = FakeBroker() + + async def notify(_agent, _message): + return True + + poller = BrokerBuzzPoller( + store.get_buzz_signing_material("barsik"), + broker, + store, + notify, + heartbeat_interval=10, + liveness_timeout=1, + ) + task = asyncio.create_task(poller.start()) + await asyncio.wait_for(rig.main_requests_received.wait(), timeout=2) + await asyncio.wait_for(rig.partial_eose_sent.wait(), timeout=2) + await asyncio.wait_for(broker.delivered.wait(), timeout=2) + for _ in range(200): + if poller.poll_count >= 4: + break + await asyncio.sleep(0.01) + + assert poller.poll_count >= 4 + assert poller.health["status"] == "starting" + assert store.get_buzz_inbound_policy("barsik")["status"] != "connected" + assert len(rig.main_requests) == 2 + assert len({frame[1] for frame in rig.main_requests}) == 2 + assert {tuple(frame[2]["#h"]) for frame in rig.main_requests} == { + (CHANNEL,), + (OTHER_CHANNEL,), + } + assert all(frame[2]["kinds"] == [9, 20002] for frame in rig.main_requests) + assert all("since" not in frame[2] for frame in rig.main_requests) + + rig.release_final_eose.set() + await asyncio.wait_for(rig.all_eose_sent.wait(), timeout=2) + for _ in range(200): + if poller.health["status"] == "connected": + break + await asyncio.sleep(0.01) + assert poller.health["status"] == "connected" + assert store.get_buzz_inbound_policy("barsik")["status"] == "connected" + + rig.release_live_events.set() + await asyncio.wait_for(rig.live_events_pushed.wait(), timeout=2) + for _ in range(200): + if len(broker.calls) == len(live_events) + 1: + break + await asyncio.sleep(0.01) + poller.stop() + await asyncio.wait_for(task, timeout=2) + + delivered_ids = [call[1].message_id for call in broker.calls] + assert delivered_ids[0] == stored_event["id"] + assert all(delivered_ids.count(event["id"]) == 1 for event in live_events) + assert len(delivered_ids) == len(set(delivered_ids)) == 3 + assert store._db.execute( + "SELECT event_id, delivery_status FROM buzz_inbound_events ORDER BY event_id" + ).fetchall() == sorted( + (event["id"], "delivered") for event in [stored_event, *live_events] + ) + assert not any(frame[0] == "EVENT" for frame in rig.captured) + store.close() + + @pytest.mark.asyncio async def test_restart_replays_overlap_but_dedupes_delivered_event(tmp_path): relay_url = ""