diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index b64158811..c3bd9bc8b 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -13,7 +13,8 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()`: it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. - `service.py` - Manages lifecycle of all configured channels from `config.yaml`. Shutdown closes manager admission first and keeps transports alive until every manager worker/follow-up watcher has exited. A successful manager stop therefore owns no live handler; if the Gateway's outer timeout cancels shutdown, the service retains its channel objects and global singleton so unfinished resources are not detached and cleanup can be retried. - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages`; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) -- `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating, streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published +- `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating, streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published + Buzz seen-event shutdown stays bounded and retryable: `aflush()` awaits any in-flight write, attempts at most one final snapshot, leaves a still-changing generation dirty for fail-open replay, and returns without a live persistence timer. A Gateway cancellation does not cancel the underlying worker-thread write; `BuzzChannel.stop()` tracks cleanup completion separately from transport admission so ChannelService can retry the retained channel without racing a second snapshot against that write. The store is quiesced before stop (including the already-stopped guard), so a timed-out relay task that records after stop only marks data dirty and cannot schedule detached file work; a repeated `stop()` still drains that dirty state, while `BuzzChannel.start()` explicitly resumes scheduling and flushes it automatically. - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs - `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store diff --git a/backend/app/channels/buzz.py b/backend/app/channels/buzz.py index ff3cdf154..a768d7c61 100644 --- a/backend/app/channels/buzz.py +++ b/backend/app/channels/buzz.py @@ -289,6 +289,12 @@ class BuzzChannel(Channel): self._session_started_at: int | None = None # wall clock at which the CURRENT socket opened; anchors the live membership filter self._transport: Any = None self._task: asyncio.Task | None = None + # Transport admission and cleanup completion are separate lifecycle + # states: Gateway may cancel stop() after ``_running`` is cleared, and + # ChannelService retains this instance specifically so cleanup can be + # retried. + self._stop_complete = True + self._seen_events.quiesce() self._publish = self.bus.publish_inbound # test seam (discord.py idiom) @property @@ -308,9 +314,11 @@ class BuzzChannel(Channel): # message looks exactly like a broken relay to the operator. Say so # once, loudly, at the only point where it is actionable. logger.warning("[buzz] channels.buzz.allowed_users is empty: EVERY inbound chat message will be dropped (Buzz denies by default). Add member pubkeys (hex or npub) to enable the channel.") + self._stop_complete = False self.bus.subscribe_outbound(self._on_outbound) self._spawn_connection() self._running = True + self._seen_events.resume() logger.info("[buzz] channel started (relay=%s pubkey=%s allowed_users=%d)", self._relay_url, self._keys.pubkey_hex, len(self._allowed_users)) def _spawn_connection(self) -> None: @@ -343,9 +351,23 @@ class BuzzChannel(Channel): still subscribed to anything -- from a previous process lifetime. The per-channel replay cursors (``_seen_created_at``) deliberately survive, so a restart resumes where it left off instead of replaying every channel. + + ``_running`` closes transport admission at the start of teardown, while + ``_stop_complete`` is set only after the final seen-event flush. Keeping + those states separate lets ChannelService retry this same instance when + its outer shutdown timeout cancels ``stop()`` mid-cleanup. + + Seen-event scheduling is quiesced before even the already-stopped guard. + A relay task abandoned after the bounded wait can therefore record late + ids as dirty state, but cannot attach new callbacks to a channel the + service may remove. A repeated ``stop()`` drains that dirty state, while + ``start()`` explicitly resumes automatic scheduling. """ - if not self._running: + self._seen_events.quiesce() + if not self._running and self._stop_complete: + await self._seen_events.aflush() return + self._stop_complete = False self._running = False self.bus.unsubscribe_outbound(self._on_outbound) if self._task is not None: @@ -380,7 +402,8 @@ class BuzzChannel(Channel): 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() + await self._seen_events.aflush() + self._stop_complete = True logger.info("[buzz] channel stopped") # -- subscriptions ------------------------------------------------------ @@ -1255,7 +1278,7 @@ class BuzzChannel(Channel): # 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): + if await self._seen_events.aseen(channel_id, event_id): logger.debug("[buzz] dropped replayed event id=%s in channel %s", event_id, channel_id) return @@ -1279,7 +1302,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) + await self._seen_events.arecord(channel_id, event_id) return if author not in self._allowed_users: # Deny-by-default is intentional (see start()'s empty-allowlist warning), @@ -1315,7 +1338,7 @@ class BuzzChannel(Channel): # 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) + await self._seen_events.arecord(channel_id, event_id) # -- outbound -------------------------------------------------------------- diff --git a/backend/app/channels/buzz_seen_events.py b/backend/app/channels/buzz_seen_events.py index 615c855b4..855320e2c 100644 --- a/backend/app/channels/buzz_seen_events.py +++ b/backend/app/channels/buzz_seen_events.py @@ -28,15 +28,24 @@ 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 +Writes are coalesced: ``arecord()`` 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. +backlog burst pays one O(store) file write instead of one per event. The timer +captures an immutable payload on the event loop and writes it through +``asyncio.to_thread``; a generation counter keeps records that arrive during +that write dirty for the next flush. ``aseen()`` likewise offloads the initial +file load, and ``BuzzChannel.stop()`` awaits ``aflush()`` so clean shutdown is +durable before it returns. That final flush is bounded: it waits for an existing +write and attempts at most one newer snapshot, leaving any still-moving +generation dirty for fail-open replay rather than hanging shutdown. The +in-flight worker write is shielded and retained if Gateway cancellation +interrupts shutdown, so a retried stop awaits it instead of racing it with a +second snapshot; every final attempt also removes its coalescing timer before +returning. ``BuzzChannel`` quiesces automatic scheduling before stop and resumes +it after start, so a timed-out relay task that records after the stop boundary +leaves data dirty for fail-open replay without creating detached file work. The +synchronous ``seen()`` / ``record()`` / ``flush()`` methods remain for tests and +tooling that run outside an event loop. """ from __future__ import annotations @@ -45,6 +54,7 @@ import asyncio import json import logging import tempfile +import threading from collections import OrderedDict, deque from pathlib import Path @@ -64,7 +74,11 @@ MAX_CHANNELS = 512 class BuzzSeenEventStore: - """Bounded, JSON-persisted map of channel id -> recently processed event ids.""" + """Bounded, JSON-persisted map of channel id -> recently processed event ids. + + Gateway event-loop callers must use ``aseen()``, ``arecord()``, and + ``aflush()`` so filesystem access stays on a worker thread. + """ def __init__(self, path: str | Path | None = None) -> None: # ``path=None`` means memory-only: no file is read or written, which is @@ -76,8 +90,12 @@ class BuzzSeenEventStore: self._ids: OrderedDict[str, deque[str]] = OrderedDict() self._sets: dict[str, set[str]] = {} self._loaded = False + self._load_lock = threading.Lock() self._dirty = False + self._generation = 0 + self._quiesced = False self._flush_handle: asyncio.TimerHandle | None = None + self._flush_task: asyncio.Task[bool] | 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 @@ -85,36 +103,43 @@ class BuzzSeenEventStore: # -- 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(): + with self._load_lock: + if self._loaded: 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() + try: + 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() + finally: + # Publish only after every loaded entry is visible. Async hot + # paths may read this flag without taking ``_load_lock``. + self._loaded = True - def _save(self) -> None: + def _snapshot(self) -> dict[str, list[str]]: + return {channel: list(ids) for channel, ids in self._ids.items()} + + def _write_snapshot(self, payload: dict[str, list[str]]) -> bool: if self._path is None: - return + return True 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 @@ -123,13 +148,27 @@ class BuzzSeenEventStore: tmp_name = fh.name json.dump(payload, fh) Path(tmp_name).replace(path) - self._dirty = False + return True 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) + return False + + def _save(self) -> None: + if self._write_snapshot(self._snapshot()): + self._dirty = False + + async def _flush_once(self) -> bool: + if not self._dirty: + return True + generation = self._generation + saved = await asyncio.to_thread(self._write_snapshot, self._snapshot()) + if saved and generation == self._generation: + self._dirty = False + return saved def _request_flush(self) -> None: """Coalesce persistence: at most one write per FLUSH_DELAY_SECONDS. @@ -139,6 +178,8 @@ class BuzzSeenEventStore: callers had before coalescing existed. """ self._dirty = True + if self._quiesced: + return try: loop = asyncio.get_running_loop() except RuntimeError: @@ -153,10 +194,127 @@ class BuzzSeenEventStore: self._flush_loop = loop self._flush_handle = loop.call_later(FLUSH_DELAY_SECONDS, self._flush_scheduled) + def quiesce(self) -> None: + """Disable automatic persistence while retaining late records as dirty.""" + self._quiesced = True + if self._flush_handle is not None: + self._flush_handle.cancel() + self._flush_handle = None + + def resume(self) -> None: + """Resume automatic persistence and schedule any retained dirty state.""" + if not self._quiesced: + return + self._quiesced = False + if self._dirty: + self._request_flush() + + def _flush_task_on_current_loop(self) -> asyncio.Task[bool] | None: + """Return the pending flush only when it belongs to the running loop.""" + pending = self._flush_task + if pending is not None and pending.get_loop() is not asyncio.get_running_loop(): + self._flush_task = None + return None + return pending + def _flush_scheduled(self) -> None: self._flush_handle = None - if self._dirty: - self._save() + if not self._dirty: + return + pending = self._flush_task_on_current_loop() + if pending is not None and not pending.done(): + return + self._flush_task = asyncio.create_task(self._flush_once()) + self._flush_task.add_done_callback(self._flush_finished) + + def _flush_finished(self, task: asyncio.Task[bool]) -> None: + if self._flush_task is task: + self._flush_task = None + try: + saved = task.result() + except asyncio.CancelledError: + return + except Exception: + logger.warning("[buzz] unexpected seen-event flush failure", exc_info=True) + return + # A successful snapshot may be stale when another event arrived while + # the worker thread was writing it. Schedule that newer generation for + # the next coalescing window. A failed write deliberately waits for the + # next record or explicit flush, preserving the store's fail-open retry + # policy instead of spinning on a broken filesystem every second. + if saved and self._dirty: + self._request_flush() + + def _final_flush_finished(self, task: asyncio.Task[bool]) -> None: + """Retire a stop-owned write without scheduling work after stop.""" + if self._flush_task is task: + self._flush_task = None + try: + task.result() + except asyncio.CancelledError: + return + except Exception: + logger.warning("[buzz] unexpected final seen-event flush failure", exc_info=True) + + async def aflush(self) -> None: + """Make one bounded final persist without blocking the event loop.""" + if self._flush_handle is not None: + self._flush_handle.cancel() + self._flush_handle = None + + try: + pending = self._flush_task_on_current_loop() + if pending is not None and pending is not asyncio.current_task(): + if not pending.cancelled(): + # This scheduled write now belongs to channel teardown. + # Its normal callback would arm another timer when the + # snapshot is stale, possibly after a cancelled aflush() + # has already run its timer cleanup. + pending.remove_done_callback(self._flush_finished) + pending.remove_done_callback(self._final_flush_finished) + pending.add_done_callback(self._final_flush_finished) + try: + # Cancelling channel stop must not cancel the worker + # write: the thread keeps running regardless, and a + # retried stop must await that same write rather than + # start a concurrent stale snapshot. + await asyncio.shield(pending) + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None and current.cancelling(): + raise + logger.warning("[buzz] in-flight seen-event flush was cancelled during final persist; retrying") + except Exception: + logger.warning("[buzz] in-flight seen-event flush failed during final persist; retrying", exc_info=True) + if self._flush_task is pending and pending.done(): + self._flush_task = None + + # One final snapshot keeps shutdown bounded even if an abandoned + # relay task is still recording. Track and shield it so a Gateway + # timeout leaves one retryable write instead of an untracked worker + # thread that can race the next stop attempt. + if self._dirty: + final = asyncio.create_task(self._flush_once()) + self._flush_task = final + final.add_done_callback(self._final_flush_finished) + try: + await asyncio.shield(final) + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None and current.cancelling(): + raise + logger.warning("[buzz] final seen-event flush was cancelled; leaving store dirty") + except Exception: + logger.warning("[buzz] final seen-event flush failed; leaving store dirty", exc_info=True) + if self._flush_task is final and final.done(): + self._flush_task = None + finally: + # ``aflush`` is the channel-stop boundary: never leave a timer + # owned by a stopped channel. A record that raced the final + # snapshot remains dirty so a retried stop can persist it. + if self._flush_handle is not None: + self._flush_handle.cancel() + self._flush_handle = None def flush(self) -> None: """Persist pending records now (no-op when nothing is dirty). @@ -184,11 +342,30 @@ class BuzzSeenEventStore: self._ensure_loaded() return event_id in self._sets.get(channel_id, ()) + async def aseen(self, channel_id: str, event_id: str) -> bool: + """Async counterpart of :meth:`seen` for Gateway event-loop callers.""" + if not event_id: + return False + if not self._loaded: + await asyncio.to_thread(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() + self._record_loaded(channel_id, event_id) + + async def arecord(self, channel_id: str, event_id: str) -> None: + """Record an event without performing file IO on the event loop.""" + if not channel_id or not event_id: + return + if not self._loaded: + await asyncio.to_thread(self._ensure_loaded) + self._record_loaded(channel_id, event_id) + + def _record_loaded(self, channel_id: str, event_id: str) -> None: ids = self._ids.get(channel_id) if ids is None: ids = deque(maxlen=MAX_IDS_PER_CHANNEL) @@ -204,4 +381,5 @@ class BuzzSeenEventStore: # 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._generation += 1 self._request_flush() diff --git a/backend/tests/blocking_io/test_buzz_channel_seen_events.py b/backend/tests/blocking_io/test_buzz_channel_seen_events.py new file mode 100644 index 000000000..e7cdeb80e --- /dev/null +++ b/backend/tests/blocking_io/test_buzz_channel_seen_events.py @@ -0,0 +1,174 @@ +"""Regression coverage for Buzz replay persistence at the channel boundary.""" + +from __future__ import annotations + +import asyncio +import json +import threading +from pathlib import Path + +import pytest + +from app.channels import buzz_nostr, buzz_seen_events +from app.channels.buzz import BuzzChannel +from app.channels.buzz_seen_events import BuzzSeenEventStore +from app.channels.message_bus import MessageBus + +pytestmark = pytest.mark.asyncio + +_BOT_PUBLIC = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9" +_OWNER_PUBLIC = "11" * 32 +_CHANNEL_ID = "136852ee-63e1-49c2-8927-413b5ee8e5f7" + + +def _event() -> dict: + tags = [["h", _CHANNEL_ID], ["p", _BOT_PUBLIC]] + created_at = 1_700_000_100 + content = "@DeerFlow hello" + return { + "id": buzz_nostr.event_id(_OWNER_PUBLIC, created_at, 9, tags, content), + "pubkey": _OWNER_PUBLIC, + "created_at": created_at, + "kind": 9, + "tags": tags, + "content": content, + # Signature verification is patched below. Keeping a correctly shaped + # event makes this test independent of the optional ``buzz`` extra, so + # the default blocking-I/O CI job cannot silently skip the regression. + "sig": "00" * 64, + } + + +def _channel(path: Path, *, seen_events: BuzzSeenEventStore | None = None) -> tuple[BuzzChannel, list]: + channel = BuzzChannel( + bus=MessageBus(), + config={ + "relay_url": "wss://buzz.example.com", + "private_key": "unused-by-this-test", + "allowed_users": [_OWNER_PUBLIC], + "seen_event_store": seen_events or BuzzSeenEventStore(path), + }, + ) + channel._keys = buzz_nostr.NostrKeys(secret=b"", pubkey_hex=_BOT_PUBLIC) + published = [] + + async def publish(message) -> None: + published.append(message) + + channel._publish = publish + return channel, published + + +async def test_channel_stop_persists_replay_guard_without_blocking(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A clean stop suppresses the same relay event after channel restart.""" + monkeypatch.setattr(buzz_nostr, "verify_event", lambda _event: True) + path = tmp_path / "buzz-seen-events.json" + event_frame = json.dumps(["EVENT", "buzz-chat", _event()]) + + first, first_messages = _channel(path) + await first.handle_relay_frame(event_frame) + assert len(first_messages) == 1 + + first._running = True + first.bus.subscribe_outbound(first._on_outbound) + await first.stop() + + restarted, restarted_messages = _channel(path) + await restarted.handle_relay_frame(event_frame) + + assert restarted_messages == [] + + +async def test_channel_stop_retries_seen_event_flush_after_cancellation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A Gateway timeout must leave seen-event cleanup retryable.""" + path = tmp_path / "buzz-seen-events.json" + seen_events = BuzzSeenEventStore(path) + channel, _ = _channel(path, seen_events=seen_events) + write_started = threading.Event() + release_write = threading.Event() + retry_write_started = threading.Event() + write_snapshot = seen_events._write_snapshot + write_count = 0 + + def fail_cancelled_write(payload: dict[str, list[str]]) -> bool: + nonlocal write_count + write_count += 1 + if write_count == 1: + write_started.set() + assert release_write.wait(timeout=2) + return False + retry_write_started.set() + return write_snapshot(payload) + + seen_events._write_snapshot = fail_cancelled_write + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 60) + await seen_events.arecord(_CHANNEL_ID, "event-1") + channel._running = True + channel.bus.subscribe_outbound(channel._on_outbound) + + first_stop = asyncio.create_task(channel.stop()) + assert await asyncio.to_thread(write_started.wait, 2) + first_stop.cancel() + with pytest.raises(asyncio.CancelledError): + await first_stop + + retry_stop = asyncio.create_task(channel.stop()) + assert not await asyncio.to_thread(retry_write_started.wait, 0.05) + release_write.set() + await retry_stop + + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-1") + + +async def test_abandoned_relay_records_are_drained_by_retried_stop_or_restart(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Late relay records stay unscheduled but remain reachable to cleanup.""" + path = tmp_path / "buzz-seen-events.json" + seen_events = BuzzSeenEventStore(path) + channel, _ = _channel(path, seen_events=seen_events) + relay_started = asyncio.Event() + release_abandoned_relay = asyncio.Event() + + async def cancellation_resistant_relay() -> None: + relay_started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + await release_abandoned_relay.wait() + await seen_events.arecord(_CHANNEL_ID, "late-event") + + monkeypatch.setattr("app.channels.buzz.STOP_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01) + channel._task = asyncio.create_task(cancellation_resistant_relay()) + channel._running = True + channel.bus.subscribe_outbound(channel._on_outbound) + await relay_started.wait() + + abandoned_relay = channel._task + assert abandoned_relay is not None + await channel.stop() + release_abandoned_relay.set() + await abandoned_relay + assert await seen_events.aseen(_CHANNEL_ID, "late-event") + + await asyncio.sleep(0.05) + stopped_view = BuzzSeenEventStore(path) + assert not await stopped_view.aseen(_CHANNEL_ID, "late-event") + + await channel.stop() + retried_stop_view = BuzzSeenEventStore(path) + assert await retried_stop_view.aseen(_CHANNEL_ID, "late-event") + + await seen_events.arecord(_CHANNEL_ID, "restart-event") + await asyncio.sleep(0.05) + still_stopped_view = BuzzSeenEventStore(path) + assert not await still_stopped_view.aseen(_CHANNEL_ID, "restart-event") + + monkeypatch.setattr(buzz_nostr, "parse_private_key", lambda _value: buzz_nostr.NostrKeys(secret=b"", pubkey_hex=_BOT_PUBLIC)) + channel._spawn_connection = lambda: None + await channel.start() + await asyncio.sleep(0.05) + + restarted_view = BuzzSeenEventStore(path) + assert await restarted_view.aseen(_CHANNEL_ID, "restart-event") + await channel.stop() diff --git a/backend/tests/blocking_io/test_buzz_seen_event_store.py b/backend/tests/blocking_io/test_buzz_seen_event_store.py new file mode 100644 index 000000000..85e22b9d9 --- /dev/null +++ b/backend/tests/blocking_io/test_buzz_seen_event_store.py @@ -0,0 +1,250 @@ +"""Regression coverage for Buzz replay-guard persistence on async paths. + +Buzz receives and stops channels on the Gateway event loop. Persisting the +seen-event replay guard must therefore stay off that loop while still making a +clean stop durable before it returns. +""" + +from __future__ import annotations + +import asyncio +import threading +from pathlib import Path + +import pytest + +from app.channels import buzz_seen_events +from app.channels.buzz_seen_events import BuzzSeenEventStore + +pytestmark = pytest.mark.asyncio + +_CHANNEL_ID = "136852ee-63e1-49c2-8927-413b5ee8e5f7" + + +async def test_async_replay_guard_round_trips_without_blocking_the_event_loop(tmp_path: Path) -> None: + """An async flush makes a recorded event visible to a fresh store.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + + await store.arecord(_CHANNEL_ID, "event-1") + await store.aflush() + + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-1") + + +async def test_loaded_store_does_not_dispatch_hot_path_lookups_to_a_thread(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Only the cold file load needs the shared worker pool.""" + store = BuzzSeenEventStore(tmp_path / "buzz-seen-events.json") + assert not await store.aseen(_CHANNEL_ID, "event-1") + + async def unexpected_to_thread(*_args, **_kwargs): + raise AssertionError("loaded seen-event store used the worker pool") + + monkeypatch.setattr(asyncio, "to_thread", unexpected_to_thread) + + await store.arecord(_CHANNEL_ID, "event-1") + assert await store.aseen(_CHANNEL_ID, "event-1") + + +async def test_event_recorded_during_a_final_write_remains_retryable(tmp_path: Path) -> None: + """An in-flight snapshot must not mark a newer generation as clean.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + write_started = threading.Event() + release_write = threading.Event() + snapshots: list[dict[str, list[str]]] = [] + write_snapshot = store._write_snapshot + + def pause_first_write(payload: dict[str, list[str]]) -> bool: + snapshots.append(payload) + if len(snapshots) == 1: + write_started.set() + assert release_write.wait(timeout=2) + return write_snapshot(payload) + + store._write_snapshot = pause_first_write + await store.arecord(_CHANNEL_ID, "event-1") + flush = asyncio.create_task(store.aflush()) + + assert await asyncio.to_thread(write_started.wait, 2) + await store.arecord(_CHANNEL_ID, "event-2") + release_write.set() + await flush + assert await store.aseen(_CHANNEL_ID, "event-2") + + # The final flush deliberately cancels its trailing timer, but the newer + # generation stays dirty so a retried cleanup can make it durable. + await store.aflush() + + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-1") + assert await restarted.aseen(_CHANNEL_ID, "event-2") + + +async def test_final_flush_is_bounded_while_records_keep_arriving(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Shutdown must return instead of chasing an active producer forever.""" + store = BuzzSeenEventStore(tmp_path / "buzz-seen-events.json") + first_write_started = threading.Event() + release_first_write = threading.Event() + second_write_started = threading.Event() + release_second_write = threading.Event() + snapshots: list[dict[str, list[str]]] = [] + write_snapshot = store._write_snapshot + + def pause_bounded_writes(payload: dict[str, list[str]]) -> bool: + snapshots.append(payload) + if len(snapshots) == 1: + first_write_started.set() + assert release_first_write.wait(timeout=2) + elif len(snapshots) == 2: + second_write_started.set() + assert release_second_write.wait(timeout=2) + return write_snapshot(payload) + + store._write_snapshot = pause_bounded_writes + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01) + await store.arecord(_CHANNEL_ID, "event-0") + assert await asyncio.to_thread(first_write_started.wait, 2) + + # Leave enough time to finish the bounded flush before a timer armed by a + # late record can fire. Once aflush() returns, no third write may remain + # scheduled for a store whose owning channel has stopped. + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.05) + final_flush = asyncio.create_task(store.aflush()) + await store.arecord(_CHANNEL_ID, "late-1") + release_first_write.set() + + assert await asyncio.to_thread(second_write_started.wait, 2) + await store.arecord(_CHANNEL_ID, "late-2") + release_second_write.set() + await final_flush + await asyncio.sleep(0.1) + + assert len(snapshots) == 2 + assert await store.aseen(_CHANNEL_ID, "late-2") + + +async def test_final_flush_ignores_a_task_from_a_closed_event_loop(tmp_path: Path) -> None: + """A stale loop-owned task must not abort shutdown on the current loop.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + + def leave_cancelled_task_from_closed_loop() -> None: + async def seed() -> None: + await store.arecord(_CHANNEL_ID, "event-1") + store._flush_task = asyncio.create_task(asyncio.sleep(60)) + + asyncio.run(seed()) + + await asyncio.to_thread(leave_cancelled_task_from_closed_loop) + await store.aflush() + + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-1") + + +async def test_final_flush_recovers_when_an_in_flight_flush_task_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A background persistence failure must not escape channel shutdown.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + write_started = threading.Event() + release_write = threading.Event() + write_snapshot = store._write_snapshot + write_count = 0 + + def fail_first_write(payload: dict[str, list[str]]) -> bool: + nonlocal write_count + write_count += 1 + if write_count == 1: + write_started.set() + assert release_write.wait(timeout=2) + raise RuntimeError("worker pool is shutting down") + return write_snapshot(payload) + + store._write_snapshot = fail_first_write + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01) + await store.arecord(_CHANNEL_ID, "event-1") + assert await asyncio.to_thread(write_started.wait, 2) + + final_flush = asyncio.create_task(store.aflush()) + await asyncio.sleep(0) + release_write.set() + await final_flush + + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-1") + + +async def test_cancelled_final_flush_does_not_rearm_timer_from_existing_write(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A retained scheduled write must not create new work after stop.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + write_started = threading.Event() + release_write = threading.Event() + snapshots: list[dict[str, list[str]]] = [] + write_snapshot = store._write_snapshot + + def pause_first_write(payload: dict[str, list[str]]) -> bool: + snapshots.append(payload) + if len(snapshots) == 1: + write_started.set() + assert release_write.wait(timeout=2) + return write_snapshot(payload) + + store._write_snapshot = pause_first_write + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01) + await store.arecord(_CHANNEL_ID, "event-1") + assert await asyncio.to_thread(write_started.wait, 2) + + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.05) + await store.arecord(_CHANNEL_ID, "event-2") + final_flush = asyncio.create_task(store.aflush()) + await asyncio.sleep(0) + final_flush.cancel() + with pytest.raises(asyncio.CancelledError): + await final_flush + + release_write.set() + await asyncio.sleep(0.1) + assert len(snapshots) == 1 + assert await store.aseen(_CHANNEL_ID, "event-2") + + await store.aflush() + restarted = BuzzSeenEventStore(path) + assert await restarted.aseen(_CHANNEL_ID, "event-2") + + +async def test_scheduled_flush_replaces_a_pending_task_from_a_closed_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A stale task must not disable normal coalesced persistence on a new loop.""" + path = tmp_path / "buzz-seen-events.json" + store = BuzzSeenEventStore(path) + + def pending_task_from_closed_loop() -> asyncio.Task: + loop = asyncio.new_event_loop() + + async def remain_pending() -> None: + await asyncio.sleep(60) + + task = loop.create_task(remain_pending()) + loop.run_until_complete(asyncio.sleep(0)) + task._log_destroy_pending = False + loop.close() + return task + + stale_task = await asyncio.to_thread(pending_task_from_closed_loop) + store._flush_task = stale_task + monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0) + + try: + await store.arecord(_CHANNEL_ID, "event-1") + deadline = asyncio.get_running_loop().time() + 1 + while asyncio.get_running_loop().time() < deadline: + if await BuzzSeenEventStore(path).aseen(_CHANNEL_ID, "event-1"): + break + await asyncio.sleep(0.01) + + assert await BuzzSeenEventStore(path).aseen(_CHANNEL_ID, "event-1") + finally: + if store._flush_task is stale_task: + store._flush_task = None diff --git a/backend/tests/test_buzz_seen_events.py b/backend/tests/test_buzz_seen_events.py index 830f21528..2934f9ca8 100644 --- a/backend/tests/test_buzz_seen_events.py +++ b/backend/tests/test_buzz_seen_events.py @@ -130,7 +130,7 @@ def test_redelivered_event_is_dropped_across_restart(tmp_path): ch1, captured1 = _started_with_store(store1) _dispatch(ch1, ev) assert len(captured1) == 1 - store1.flush() # what BuzzChannel.stop() does on a clean shutdown + store1.flush() # synchronous test equivalent of BuzzChannel.stop()'s aflush() # Simulated restart: a fresh channel instance, fresh store object, same file. ch2, captured2 = _started_with_store(BuzzSeenEventStore(path)) @@ -217,13 +217,13 @@ def test_saves_are_coalesced_under_a_running_loop(tmp_path): path = tmp_path / "seen.json" store = BuzzSeenEventStore(path) saves = [] - original_save = store._save + original_write_snapshot = store._write_snapshot - def counting_save(): + def counting_write_snapshot(payload): saves.append(1) - original_save() + return original_write_snapshot(payload) - store._save = counting_save + store._write_snapshot = counting_write_snapshot async def burst(): for i in range(50):