From 556a178771276235bd8b44447b61da0bc23b45d8 Mon Sep 17 00:00:00 2001 From: ajayr Date: Sat, 22 Aug 2026 09:34:45 +0100 Subject: [PATCH] fix(buzz): drop replayed events across reconnects with a persistent seen-id store (#4888) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(buzz): drop replayed events across reconnects with a persistent seen-id store The Buzz connector's resubscribe filter replays by design: 'since' is the created_at of the last processed event and NIP-01 'since' is inclusive, so every relay reconnect redelivers at least that event. The guard against re-running the agent on it was the manager's inbound dedupe, whose default store is in-process with a 10-minute TTL — so any reconnect more than ten minutes after a channel's last message (or any gateway restart) re-answered that message. Users saw the agent respond to an old question after every relay restart. Fix: persist the ids of fully processed events per channel (BuzzSeenEventStore, JSON under {base_dir}/channels/, atomic writes) and drop redelivered ids in _handle_chat_event before the /connect branch — a replayed /connect would otherwise be re-answered with a spurious 'code invalid or expired'. Matching is by exact event id only, never timestamp, so a genuinely new event (same-second or clock-skewed author) can never be skipped, preserving the connector's fail-toward-replay invariant. Only fully processed events are recorded, mirroring the watermark rule: a gated drop or failed publish stays replayable. Fail-open in both directions: an unreadable store loads empty (costs one replayed reply, the previous behavior) and a failed write is logged and retried on the next record. Id lists and the channel map are bounded like the connector's other remote-fed maps. The persistent path is wired in ChannelService (like channel_store); directly constructed channels get a memory-only store so tests and tooling stay free of filesystem side effects. Co-Authored-By: Claude Fable 5 * fix(buzz): coalesce seen-store writes, clean up temp files, harden docs and coverage Address review on the seen-event store: - record() now marks the store dirty and coalesces persistence to one write per FLUSH_DELAY_SECONDS on the event loop, so a reconnect backlog burst pays one O(store) file write instead of one per event; sync callers (no running loop) keep immediate writes, and BuzzChannel.stop() flushes so a clean shutdown loses nothing. A crash inside the window only costs replay, never a skip. - _save() unlinks its temp file on failure (ChannelStore parity), so a persistently unwritable path no longer accumulates *.tmp litter. - Module docstring now documents that restart protection is bounded to the newest MAX_IDS_PER_CHANNEL ids per channel (and to raise it if a relay ever serves a deeper default backlog), and pins the single-event-loop assumption that makes the class safe without a lock. - New tests: MAX_CHANNELS LRU eviction, coalescing behavior, flush idempotence, temp-file cleanup, stop() flushing, and the ChannelService wiring that injects seen_event_store_path (the line that makes real deployments durable). Co-Authored-By: Claude Fable 5 * fix(buzz): reschedule the coalesced flush when the pending timer's loop is gone A pending flush handle pinned to a since-closed event loop kept _flush_handle non-None forever, so later record() calls on a new loop never scheduled a timer and the store silently stopped persisting until an explicit flush(). Track the scheduling loop (TimerHandle has no public get_loop()) and reschedule when it differs from the running one. Unreachable in production (one loop per process, stop() flushes), but now hardened and tested. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- backend/app/channels/buzz.py | 33 ++- backend/app/channels/buzz_seen_events.py | 207 ++++++++++++++ backend/app/channels/service.py | 9 + backend/tests/test_buzz_seen_events.py | 347 +++++++++++++++++++++++ 4 files changed, 595 insertions(+), 1 deletion(-) create mode 100644 backend/app/channels/buzz_seen_events.py create mode 100644 backend/tests/test_buzz_seen_events.py diff --git a/backend/app/channels/buzz.py b/backend/app/channels/buzz.py index fa65def29..ff3cdf154 100644 --- a/backend/app/channels/buzz.py +++ b/backend/app/channels/buzz.py @@ -50,6 +50,7 @@ from urllib.parse import urlparse from app.channels import buzz_nostr from app.channels.base import Channel +from app.channels.buzz_seen_events import BuzzSeenEventStore from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import InboundMessage, InboundMessageType, InboundQueueFullError, MessageBus, OutboundMessage @@ -273,6 +274,14 @@ class BuzzChannel(Channel): self._last_requester: dict[tuple[str, str | None], str] = {} self._pending_auth_challenge: str | None = None # set from an AUTH relay frame; consumed by the NIP-42 flow in _session self._seen_created_at: dict[str, int] = {} # channel id -> high-water mark of PROCESSED created_at (see _advance_watermark) + # Persistent processed-event-id record. The resubscribe filter replays + # by design (inclusive ``since``), and the manager's inbound dedupe has + # a 10-minute in-process TTL — so without durable ids, every relay + # reconnect (or gateway restart) re-answers the last message in each + # channel. Dedupe here is by exact event id only, never timestamp, so + # it cannot skip a genuinely new event. ``seen_event_store`` is a test + # seam; the default persists under ``{base_dir}/channels/``. + self._seen_events: BuzzSeenEventStore = config.get("seen_event_store") or BuzzSeenEventStore(config.get("seen_event_store_path")) self._chat_subscriptions: set[str] = set() # channel ids with a live per-channel REQ on the CURRENT connection self._resubscribe_attempts: dict[str, int] = {} # sub id -> CLOSED-recovery attempts spent on the current connection/auth epoch self._auth_completed = False # has THIS socket's NIP-42 handshake been ACKNOWLEDGED by the relay? (see _handle_auth_ok, _AUTH_REQUIRED_CLOSE_PREFIX) @@ -369,6 +378,9 @@ class BuzzChannel(Channel): self._pending_auth_event_id = None self._session_started_at = None self._pending_auth_challenge = None + # Seen-id persistence is coalesced (FLUSH_DELAY_SECONDS); a clean stop + # must not lose records still inside that window to a replay on restart. + self._seen_events.flush() logger.info("[buzz] channel stopped") # -- subscriptions ------------------------------------------------------ @@ -1231,6 +1243,21 @@ class BuzzChannel(Channel): channel_id = channel_id_values[0] created_at = int(ev.get("created_at", 0)) text = str(ev.get("content", "")) + event_id = str(ev.get("id", "")) + + # Drop redeliveries of events this connector already fully processed. + # The resubscribe filter replays the watermark event on every reconnect + # (inclusive ``since``), and after a long disconnect or cursor eviction + # the relay's default backlog comes back too. The manager's inbound + # dedupe only covers a 10-minute in-process window, so without this a + # relay reconnect re-answers the last message in each channel. Matching + # is by exact event id, so a new event — whatever its author-chosen + # created_at — is never affected. Sits before the /connect branch + # deliberately: a replayed /connect would otherwise be re-answered with + # a spurious "code invalid or expired" reply on every reconnect. + if self._seen_events.seen(channel_id, event_id): + logger.debug("[buzz] dropped replayed event id=%s in channel %s", event_id, channel_id) + return # /connect must be consulted before the allowlist gate (framework # ordering rule — see Channel._pending_connect_code) so a not-yet-bound @@ -1252,6 +1279,7 @@ class BuzzChannel(Channel): # cursor: leaving it behind would replay this /connect on every reconnect # and answer each replay with a spurious "code invalid or expired" reply. self._advance_watermark(channel_id, created_at) + self._seen_events.record(channel_id, event_id) return if author not in self._allowed_users: # Deny-by-default is intentional (see start()'s empty-allowlist warning), @@ -1283,8 +1311,11 @@ class BuzzChannel(Channel): await self._publish(inbound) # Only a fully accepted-and-published event advances the cursor, and only # after the publish actually succeeded. A dropped event must never move it - # (that was the DoS), and a failed publish must leave it replayable. + # (that was the DoS), and a failed publish must leave it replayable. The + # same rule applies to the persistent seen-id record: recording before a + # failed publish would turn "replayable" into "silently skipped". self._advance_watermark(channel_id, created_at) + self._seen_events.record(channel_id, event_id) # -- outbound -------------------------------------------------------------- diff --git a/backend/app/channels/buzz_seen_events.py b/backend/app/channels/buzz_seen_events.py new file mode 100644 index 000000000..615c855b4 --- /dev/null +++ b/backend/app/channels/buzz_seen_events.py @@ -0,0 +1,207 @@ +"""Persistent record of Buzz chat events that were fully processed. + +Why this exists +--------------- +The Buzz connector's resubscribe filter deliberately replays rather than skips: +``since`` is the created_at of the last processed event and NIP-01 ``since`` is +inclusive, so every reconnect redelivers at least that event (see +``BuzzChannel._chat_filter``). The manager's inbound dedupe absorbs those +redeliveries — but its default store is in-process with a 10-minute TTL, so a +reconnect more than 10 minutes after the last message (or any gateway restart) +re-runs the agent on an already-answered message. + +This store closes that gap at the connector: the ids of fully processed events +are persisted per channel, and a redelivered id is dropped before it reaches the +bus. Dedupe is by exact event id only — never by timestamp — so a genuinely new +event (which always has a fresh id, whatever its author-chosen created_at) can +never be skipped, preserving the connector's fail-toward-replay invariant. + +Failure policy is fail-open in both directions: an unreadable file loads as +empty (costing at most one replayed answer, the pre-existing behavior) and a +failed write is logged and retried on the next flush (costing replay, never a +skip). The id lists are bounded per channel and the channel map is bounded like +the connector's other remote-fed maps. That per-channel bound also bounds the +restart protection itself: after a gateway restart ``_seen_created_at`` is +empty, the resubscribe REQ carries no ``since``, and the relay's default +backlog replays — only the newest ``MAX_IDS_PER_CHANNEL`` processed ids per +channel are dropped, so a relay backlog deeper than that would re-answer the +tail. If a relay ever serves a deeper default backlog, raise +``MAX_IDS_PER_CHANNEL`` here. + +Writes are coalesced: ``record()`` marks the store dirty and schedules one +flush per ``FLUSH_DELAY_SECONDS`` on the running event loop, so a reconnect +backlog burst pays one O(store) file write instead of one per event. With no +running loop (tests, tooling) ``record()`` flushes synchronously, and +``BuzzChannel.stop()`` flushes pending state on shutdown. This class is not +thread-safe by design: everything runs on the single channel event loop +(mutation in ``_handle_chat_event``, the coalesced flush via ``call_later`` on +that same loop). Anyone adding an off-loop user — e.g. a threaded flusher — +must add a lock around ``_ids``/``_sets`` first, as ChannelStore does. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import tempfile +from collections import OrderedDict, deque +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Coalescing window for persisting the store. Losing this window's records in a +# crash only costs replay (fail-open), never a skip. +FLUSH_DELAY_SECONDS = 1.0 + +# Ids retained per channel. Reconnect replay is normally the single watermark +# event; the deep case is a channel whose cursor was evicted, which replays the +# relay's default backlog window. Both are far below this bound. +MAX_IDS_PER_CHANNEL = 512 +# Channel-map cap, mirroring the connector's other remote-fed maps +# (channel ids arrive in remote ``h`` tags). +MAX_CHANNELS = 512 + + +class BuzzSeenEventStore: + """Bounded, JSON-persisted map of channel id -> recently processed event ids.""" + + def __init__(self, path: str | Path | None = None) -> None: + # ``path=None`` means memory-only: no file is read or written, which is + # exactly the pre-existing (non-durable) behavior. The channel service + # wires the persistent path for real deployments; constructing a + # channel directly (tests, tooling) must not create directories or + # files as a side effect. + self._path = Path(path) if path is not None else None + self._ids: OrderedDict[str, deque[str]] = OrderedDict() + self._sets: dict[str, set[str]] = {} + self._loaded = False + self._dirty = False + self._flush_handle: asyncio.TimerHandle | None = None + # The loop the pending handle was scheduled on. TimerHandle has no + # public get_loop(), so it is tracked here to detect a stale handle. + self._flush_loop: asyncio.AbstractEventLoop | None = None + + # -- persistence --------------------------------------------------------- + + def _ensure_loaded(self) -> None: + if self._loaded: + return + self._loaded = True + if self._path is None: + return + try: + if not self._path.exists(): + return + raw = json.loads(self._path.read_text(encoding="utf-8")) + except Exception: + logger.warning("[buzz] unreadable seen-event store, starting fresh (costs at most one replayed reply)", exc_info=True) + return + if not isinstance(raw, dict): + return + for channel_id, ids in raw.items(): + if not isinstance(ids, list): + continue + clean = deque((str(i) for i in ids if i), maxlen=MAX_IDS_PER_CHANNEL) + self._ids[str(channel_id)] = clean + self._sets[str(channel_id)] = set(clean) + self._enforce_channel_cap() + + def _save(self) -> None: + if self._path is None: + return + tmp_name: str | None = None + try: + path = self._path + path.parent.mkdir(parents=True, exist_ok=True) + payload = {channel: list(ids) for channel, ids in self._ids.items()} + # Atomic same-directory replace, matching ChannelStore._save: a + # crash mid-write must never truncate the store (a truncated store + # would fail open into replay on the next start, which is + # recoverable — but there is no reason to accept even that). + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, suffix=".tmp", delete=False, encoding="utf-8") as fh: + tmp_name = fh.name + json.dump(payload, fh) + Path(tmp_name).replace(path) + self._dirty = False + except Exception: + # Mirror ChannelStore._save: never leave the temp file behind, or a + # persistently failing write accumulates one *.tmp per attempt. + if tmp_name is not None: + Path(tmp_name).unlink(missing_ok=True) + logger.warning("[buzz] failed to persist seen-event store (will retry on next flush)", exc_info=True) + + def _request_flush(self) -> None: + """Coalesce persistence: at most one write per FLUSH_DELAY_SECONDS. + + With no running event loop (tests, tooling) the write happens + synchronously, preserving the immediate-durability semantics direct + callers had before coalescing existed. + """ + self._dirty = True + try: + loop = asyncio.get_running_loop() + except RuntimeError: + self.flush() + return + # A pending handle pinned to a since-closed loop would otherwise block + # scheduling forever, silently stopping persistence until an explicit + # flush() (only reachable when callers span loops, e.g. tests). + if self._flush_handle is None or self._flush_loop is not loop: + if self._flush_handle is not None: + self._flush_handle.cancel() + self._flush_loop = loop + self._flush_handle = loop.call_later(FLUSH_DELAY_SECONDS, self._flush_scheduled) + + def _flush_scheduled(self) -> None: + self._flush_handle = None + if self._dirty: + self._save() + + def flush(self) -> None: + """Persist pending records now (no-op when nothing is dirty). + + Called on channel stop so a clean shutdown never loses records to the + coalescing window; a crash inside the window only costs replay. + """ + if self._flush_handle is not None: + self._flush_handle.cancel() + self._flush_handle = None + if self._dirty: + self._save() + + def _enforce_channel_cap(self) -> None: + while len(self._ids) > MAX_CHANNELS: + evicted, _ = self._ids.popitem(last=False) + self._sets.pop(evicted, None) + + # -- api ---------------------------------------------------------------- + + def seen(self, channel_id: str, event_id: str) -> bool: + """True if *event_id* was already fully processed in *channel_id*.""" + if not event_id: + return False + self._ensure_loaded() + return event_id in self._sets.get(channel_id, ()) + + def record(self, channel_id: str, event_id: str) -> None: + """Record a fully processed event and schedule a coalesced persist.""" + if not channel_id or not event_id: + return + self._ensure_loaded() + ids = self._ids.get(channel_id) + if ids is None: + ids = deque(maxlen=MAX_IDS_PER_CHANNEL) + self._ids[channel_id] = ids + self._sets[channel_id] = set() + id_set = self._sets[channel_id] + if event_id in id_set: + return + if len(ids) == ids.maxlen: + id_set.discard(ids[0]) + ids.append(event_id) + id_set.add(event_id) + # Move the channel to the back so the channel-cap eviction is LRU-ish. + self._ids.move_to_end(channel_id) + self._enforce_channel_cap() + self._request_flush() diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index 6a27710de..ea0a36f17 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -7,6 +7,7 @@ import logging import math import os from collections.abc import Callable +from pathlib import Path from typing import TYPE_CHECKING, Any from app.channels.base import Channel @@ -392,6 +393,14 @@ class ChannelService: try: config = dict(config) config["channel_store"] = self.store + if name == "buzz" and "seen_event_store_path" not in config: + # Durable processed-event ids for the Buzz connector's replay + # guard. Wired here (like channel_store) rather than defaulted + # inside the connector so that directly constructed channels + # (tests, tooling) stay free of filesystem side effects. + from deerflow.config.paths import get_paths + + config["seen_event_store_path"] = str(Path(get_paths().base_dir) / "channels" / "buzz_seen_events.json") if self._connection_repo is not None: config["connection_repo"] = self._connection_repo channel = channel_cls(bus=self.bus, config=config) diff --git a/backend/tests/test_buzz_seen_events.py b/backend/tests/test_buzz_seen_events.py new file mode 100644 index 000000000..830f21528 --- /dev/null +++ b/backend/tests/test_buzz_seen_events.py @@ -0,0 +1,347 @@ +"""Tests for the Buzz connector's persistent replay guard (seen-event store).""" + +import asyncio +import json + +import pytest + +pytest.importorskip("coincurve") + +from app.channels import buzz_nostr +from app.channels.buzz import BuzzChannel +from app.channels.buzz_seen_events import MAX_IDS_PER_CHANNEL, BuzzSeenEventStore +from app.channels.message_bus import MessageBus + +# Keypairs mirror tests/test_buzz_channel.py: inbound events are +# signature-verified, so fixture authors must be real keypairs. +SK3_HEX = "0000000000000000000000000000000000000000000000000000000000000003" +PK3_HEX = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9" +SK_OWNER = "0000000000000000000000000000000000000000000000000000000000000005" +OWNER = buzz_nostr.parse_private_key(SK_OWNER).pubkey_hex +CHANNEL = "136852ee-63e1-49c2-8927-413b5ee8e5f7" + + +def _event(*, sk=SK_OWNER, kind=9, content="@DeerFlow hello", channel=CHANNEL, mentions=(PK3_HEX,), reply_to=None, created_at=1700000100): + tags = [["h", channel]] + if reply_to: + tags.append(["e", reply_to]) + tags.extend(["p", m] for m in mentions) + return buzz_nostr.sign_event(buzz_nostr.parse_private_key(sk), kind, tags, content, created_at) + + +# -- store ------------------------------------------------------------------ + + +def test_memory_only_store_records_without_touching_disk(tmp_path): + store = BuzzSeenEventStore(None) + store.record(CHANNEL, "e1") + assert store.seen(CHANNEL, "e1") + assert not store.seen(CHANNEL, "e2") + assert list(tmp_path.iterdir()) == [] # nothing written anywhere we can observe + + +def test_persistent_store_round_trips_across_instances(tmp_path): + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + store.record(CHANNEL, "e1") + store.record(CHANNEL, "e2") + reloaded = BuzzSeenEventStore(path) + assert reloaded.seen(CHANNEL, "e1") and reloaded.seen(CHANNEL, "e2") + assert not reloaded.seen(CHANNEL, "e3") + assert not reloaded.seen("other-channel", "e1") + + +def test_per_channel_id_cap_evicts_oldest(tmp_path): + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + for i in range(MAX_IDS_PER_CHANNEL + 5): + store.record(CHANNEL, f"id-{i}") + assert not store.seen(CHANNEL, "id-0") + assert store.seen(CHANNEL, f"id-{MAX_IDS_PER_CHANNEL + 4}") + # The persisted form respects the cap too. + persisted = json.loads(path.read_text()) + assert len(persisted[CHANNEL]) == MAX_IDS_PER_CHANNEL + + +def test_corrupt_store_file_fails_open_to_empty(tmp_path): + path = tmp_path / "seen.json" + path.write_text("{not json", encoding="utf-8") + store = BuzzSeenEventStore(path) + assert not store.seen(CHANNEL, "e1") + store.record(CHANNEL, "e1") # recovers by overwriting on next record + assert BuzzSeenEventStore(path).seen(CHANNEL, "e1") + + +def test_empty_event_id_is_never_seen_or_recorded(tmp_path): + store = BuzzSeenEventStore(tmp_path / "seen.json") + store.record(CHANNEL, "") + assert not store.seen(CHANNEL, "") + + +def test_unwritable_path_fails_open(tmp_path): + blocker = tmp_path / "blocker" + blocker.write_text("") # a file where the store expects a parent directory + store = BuzzSeenEventStore(blocker / "seen.json") + store.record(CHANNEL, "e1") # write fails, logged, no raise + assert store.seen(CHANNEL, "e1") # in-memory state still works + + +# -- connector integration --------------------------------------------------- + + +def _started_with_store(store: BuzzSeenEventStore): + ch = BuzzChannel( + bus=MessageBus(), + config={ + "relay_url": "wss://buzz.example.com", + "private_key": SK3_HEX, + "allowed_users": [OWNER], + "seen_event_store": store, + }, + ) + ch._keys = buzz_nostr.parse_private_key(SK3_HEX) + captured = [] + + async def publish(msg): + captured.append(msg) + + ch._publish = publish + return ch, captured + + +def _dispatch(ch, ev): + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", "sub1", ev]))) + + +def test_redelivered_event_is_dropped_within_one_process(tmp_path): + ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json")) + ev = _event() + _dispatch(ch, ev) + assert len(captured) == 1 + _dispatch(ch, ev) # relay replay of the same event (inclusive `since`) + assert len(captured) == 1 + + +def test_redelivered_event_is_dropped_across_restart(tmp_path): + """The bug: a relay reconnect after a gateway restart re-answered the last message.""" + path = tmp_path / "seen.json" + ev = _event() + store1 = BuzzSeenEventStore(path) + ch1, captured1 = _started_with_store(store1) + _dispatch(ch1, ev) + assert len(captured1) == 1 + store1.flush() # what BuzzChannel.stop() does on a clean shutdown + + # Simulated restart: a fresh channel instance, fresh store object, same file. + ch2, captured2 = _started_with_store(BuzzSeenEventStore(path)) + _dispatch(ch2, ev) + assert captured2 == [] + + +def test_new_event_at_same_created_at_is_still_processed(tmp_path): + """Dedupe is by id only: a same-second new event must never be skipped.""" + ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json")) + _dispatch(ch, _event(content="@DeerFlow first", created_at=1700000100)) + _dispatch(ch, _event(content="@DeerFlow second", created_at=1700000100)) + assert len(captured) == 2 + + +def test_older_created_at_new_event_is_still_processed(tmp_path): + """A clock-skewed author's new event (older timestamp) must never be skipped.""" + ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json")) + _dispatch(ch, _event(content="@DeerFlow newer", created_at=1700000200)) + _dispatch(ch, _event(content="@DeerFlow older-clock", created_at=1700000100)) + assert len(captured) == 2 + + +def test_dropped_event_is_not_recorded_and_stays_replayable(tmp_path): + """Only fully processed events are recorded — a gated drop must not poison the id.""" + store = BuzzSeenEventStore(tmp_path / "seen.json") + ch, captured = _started_with_store(store) + ev = _event(content="no mention here", mentions=()) # dropped by the mention gate + _dispatch(ch, ev) + assert captured == [] + assert not store.seen(CHANNEL, str(ev["id"])) + + +def test_replayed_connect_is_not_reprocessed(tmp_path): + """A replayed /connect must not be re-answered with 'code invalid or expired'.""" + store = BuzzSeenEventStore(tmp_path / "seen.json") + ch, captured = _started_with_store(store) + replies = [] + + async def fake_bind(code, author, channel_id): + replies.append(code) + + ch._bind_connection = fake_bind + ch._connection_repo = object() # /connect is only consulted when connections are configured + ev = _event(sk=SK_OWNER, content="/connect abc123", mentions=(PK3_HEX,)) + _dispatch(ch, ev) + assert replies == ["abc123"] + _dispatch(ch, ev) # replay + assert replies == ["abc123"] + + +def test_default_config_uses_memory_only_store(): + """Direct construction (no service wiring) must not persist anywhere.""" + ch = BuzzChannel( + bus=MessageBus(), + config={"relay_url": "wss://buzz.example.com", "private_key": SK3_HEX, "allowed_users": [OWNER]}, + ) + assert isinstance(ch._seen_events, BuzzSeenEventStore) + assert ch._seen_events._path is None + + +def test_channel_cap_evicts_least_recently_recorded(tmp_path): + from app.channels.buzz_seen_events import MAX_CHANNELS + + store = BuzzSeenEventStore(tmp_path / "seen.json") + for i in range(MAX_CHANNELS): + store.record(f"chan-{i}", f"id-{i}") + # Touch the oldest channel so LRU ordering (move_to_end) protects it. + store.record("chan-0", "id-0b") + store.record("chan-new", "id-new") # overflows the map by one + assert store.seen("chan-0", "id-0") # refreshed -> survived + assert store.seen("chan-new", "id-new") + assert not store.seen("chan-1", "id-1") # now the least recently used -> evicted + # The persisted form respects the cap too. + persisted = json.loads((tmp_path / "seen.json").read_text()) + assert len(persisted) == MAX_CHANNELS + assert "chan-1" not in persisted + + +def test_saves_are_coalesced_under_a_running_loop(tmp_path): + """One timer (and at most one file write) per burst, not one write per record.""" + from app.channels import buzz_seen_events + + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + saves = [] + original_save = store._save + + def counting_save(): + saves.append(1) + original_save() + + store._save = counting_save + + async def burst(): + for i in range(50): + store.record(CHANNEL, f"id-{i}") + assert saves == [] # nothing written yet: flush is pending on the loop + assert store._flush_handle is not None + await asyncio.sleep(buzz_seen_events.FLUSH_DELAY_SECONDS + 0.1) + + asyncio.run(burst()) + assert len(saves) == 1 + persisted = json.loads(path.read_text()) + assert len(persisted[CHANNEL]) == 50 + + +def test_flush_persists_pending_records_and_is_idempotent(tmp_path): + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + + async def record_only(): + store.record(CHANNEL, "e1") + assert not path.exists() # still inside the coalescing window + + asyncio.run(record_only()) + store.flush() + assert BuzzSeenEventStore(path).seen(CHANNEL, "e1") + mtime = path.stat().st_mtime_ns + store.flush() # nothing dirty -> no rewrite + assert path.stat().st_mtime_ns == mtime + + +def test_failed_save_leaves_no_tmp_files(tmp_path, monkeypatch): + """A persistent write failure must not accumulate *.tmp files (ChannelStore parity).""" + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + + def boom(*args, **kwargs): + raise OSError("disk full") + + monkeypatch.setattr("app.channels.buzz_seen_events.json.dump", boom) + for i in range(3): + store.record(CHANNEL, f"id-{i}") # sync context -> each record attempts a save + assert not path.exists() + assert [p.name for p in tmp_path.iterdir()] == [] # no *.tmp litter + monkeypatch.undo() + store.record(CHANNEL, "id-3") # recovery: next flush persists everything + assert json.loads(path.read_text())[CHANNEL] == ["id-0", "id-1", "id-2", "id-3"] + + +def test_channel_stop_flushes_pending_records(tmp_path): + """A clean stop must not lose records still inside the coalescing window.""" + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + ch, captured = _started_with_store(store) + ev = _event() + + async def dispatch_then_stop(): + await ch.handle_relay_frame(json.dumps(["EVENT", "sub1", ev])) + assert not path.exists() # flush still pending + ch._running = True # stop() is a no-op on a never-started channel + ch.bus.subscribe_outbound(ch._on_outbound) + await ch.stop() + + asyncio.run(dispatch_then_stop()) + assert len(captured) == 1 + assert BuzzSeenEventStore(path).seen(CHANNEL, str(ev["id"])) + + +def test_service_wiring_injects_persistent_store_path(tmp_path, monkeypatch): + """The _start_channel wiring is what makes real deployments durable — a + silent regression there would revert to memory-only with all unit tests green.""" + from app.channels.service import ChannelService + + captured_config = {} + + class StubChannel: + def __init__(self, bus, config): + captured_config.update(config) + self.is_running = True + + async def start(self): + pass + + class StubPaths: + base_dir = str(tmp_path) + + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: StubPaths()) + monkeypatch.setattr("deerflow.reflection.resolve_class", lambda path, base_class=None: StubChannel) + + service = ChannelService(channels_config={}) + started = asyncio.run(service._start_channel("buzz", {"relay_url": "wss://x", "private_key": SK3_HEX})) + assert started + expected = str(tmp_path / "channels" / "buzz_seen_events.json") + assert captured_config["seen_event_store_path"] == expected + # An explicitly configured path must win over the default wiring. + captured_config.clear() + asyncio.run(service._start_channel("buzz", {"relay_url": "wss://x", "private_key": SK3_HEX, "seen_event_store_path": "/custom/seen.json"})) + assert captured_config["seen_event_store_path"] == "/custom/seen.json" + + +def test_stale_timer_from_closed_loop_does_not_block_rescheduling(tmp_path): + """A pending flush timer pinned to a since-closed loop must not stop the + store from scheduling on a new loop (it would silently stop persisting).""" + from app.channels import buzz_seen_events + + path = tmp_path / "seen.json" + store = BuzzSeenEventStore(path) + + async def record_and_abandon(): + store.record(CHANNEL, "e1") # schedules a timer on THIS loop... + + asyncio.run(record_and_abandon()) # ...which closes before the timer fires + assert store._flush_handle is not None # the stale handle survives the loop + + async def record_on_new_loop(): + store.record(CHANNEL, "e2") + assert store._flush_loop is asyncio.get_running_loop() + await asyncio.sleep(buzz_seen_events.FLUSH_DELAY_SECONDS + 0.1) + + asyncio.run(record_on_new_loop()) + persisted = json.loads(path.read_text()) + assert persisted[CHANNEL] == ["e1", "e2"] # both records made it to disk