"""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()