From 064bb50602bcda68b81f67bda7ac765fd07eb8bf Mon Sep 17 00:00:00 2001 From: FanouZeng-TT <124567600+FanouZeng-TT@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:59:46 +0800 Subject: [PATCH] fix(buzz): page a saturated history instead of skipping the oldest (#6792) * fix(buzz): page a saturated history instead of skipping the oldest buzz-relay serves at most RELAY_HISTORY_LIMIT (2000) stored events per subscription, NEWEST-FIRST, even when the filter carries a `since`. A channel that banked more than the cap across one disconnect therefore had its OLDEST unread messages neither sent nor marked: the live page delivered the newest 2000, the watermark advanced past them, and the rest were lost -- the one remaining path in this connector that could skip a message outright. A chat/backlog page's EOSE now drives a descending `until` walk. When a page comes back AT the cap it may have older matches held back, so the connector re-asks with `until` set to that page's oldest `created_at` -- on a SEPARATE `buzz-backlog-` subscription id, never the live one -- and repeats until a page comes back under the cap or reaches the `since` the live REQ was sent with, bounded by MAX_BACKLOG_PAGES (50, warned when hit). Delivery is charged to the subscription that carried it before any gate, so a page's saturation is measured against what the relay sent, not what the allowlist/mention gates kept. The walk is strictly additive: the live subscription, its filter, and the watermark are never touched; a CLOSED for a backlog page only ends that walk; re-opening the live subscription closes any superseded page rather than orphaning it; and every failure simply leaves the pre-existing behaviour for the next reconnect to retry. * fix(buzz): skip the backlog walk when the channel has no read floor Reviewer willem-bd noted the descending-until walk has no semantic stop when `floor is None`: that is exactly a channel's first connect (no stored watermark) and again after a MAX_CACHED_CHANNELS eviction of its cursor. On those connects the walk would fetch up to MAX_BACKLOG_PAGES * RELAY_HISTORY_LIMIT events per channel instead of the newest 2000 the connector used to take. Require a floor before opening any backlog page. The backlog-loss case this walk exists for always has a watermark (the cursor survives reconnects and restarts), so it always has a floor; first connect and an evicted cursor keep the previous behaviour. Covers the skip with a new test, retargets the page-bound test to a real walk above its floor, and documents the floor requirement. * fix(buzz): stop orphaned and stalled backlog walks --------- Co-authored-by: Willem Jiang --- README.md | 2 +- backend/app/channels/AGENTS.md | 2 +- backend/app/channels/buzz.py | 283 +++++++++++++++++++++-- backend/docs/IM_CHANNEL_CONNECTIONS.md | 4 +- backend/tests/test_buzz_channel.py | 296 +++++++++++++++++++++++++ backend/tests/test_buzz_seen_events.py | 15 +- 6 files changed, 569 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index 98c732f2e8..a4e95fb4d6 100644 --- a/README.md +++ b/README.md @@ -923,7 +923,7 @@ DeerFlow supports receiving tasks from messaging apps. Channels auto-start when Cancelling a channel restart discards its pending configuration reload, preserving newer runtime settings applied afterward. -DeerFlow can also expose user-owned IM channel connections in the workspace UI. When `channel_connections` is enabled, logged-in users can bind Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, WeCom, QQ, or Buzz from the sidebar / Settings > Channels. It reuses the existing outbound `channels.*` transports, so no public IP or provider callback URL is required. Incoming IM messages then run under the connected DeerFlow user account. See [IM Channel Connections](backend/docs/IM_CHANNEL_CONNECTIONS.md) for setup and security notes. +DeerFlow can also expose user-owned IM channel connections in the workspace UI. When `channel_connections` is enabled, logged-in users can bind Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, WeCom, QQ, or Buzz from the sidebar / Settings > Channels. It reuses the existing outbound `channels.*` transports, so no public IP or provider callback URL is required. Incoming IM messages then run under the connected DeerFlow user account. See [IM Channel Connections](backend/docs/IM_CHANNEL_CONNECTIONS.md) for setup and security notes. Buzz backlog paging stops with a warning if timestamp-based queries cannot advance, and channel removal closes both live and backlog subscriptions. | Channel | Transport | Difficulty | |---------|-----------|------------| diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 8bbb268190..76f6a2f617 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -21,7 +21,7 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors. - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`. - `service.py` - Drain manager work before transports; retain timed-out instances for retry. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu disarms lark's `_auto_reconnect` first, else its reconnect branch eats the stop; DingTalk runs its own copy of the SDK connect loop), then join; a live join or Slack cleanup raises `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup. Leader lease (`leadership.py`): only the `channel_leases` holder starts a non-GitHub transport (`_start_channel` checks `holds()`; callers acquire first); `_supervise` renews at a fixed rate, runs starts/stops per provider, stops a transport before its lease lapses. - `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` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies stay plain text and their newlines and `` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (first failed tick logs at WARNING, later ones at DEBUG); 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; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads: slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `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 (thread-follow reads via the manager's `connection_identity.lookup_thread_id`, so bound authors' repo-only threads count), 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 on a live relay: `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event, `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel. 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** — 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 (a one-off hiccup is the common case); later ones back off 1s then 2s, awaited inline in the read loop, not in a background task that could outlive its 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 a false operator-facing warning that discovery/membership tracking was DOWN while discovery then completed normally. 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), `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 so one can be replaced or closed without disturbing the others, 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:** 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: one shared cursor is the newest event seen in *any* channel, so a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect. 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 (thread-follow reads via the manager's `connection_identity.lookup_thread_id`, so bound authors' repo-only threads count), 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 on a live relay: `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event, `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel. 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** — 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 (a one-off hiccup is the common case); later ones back off 1s then 2s, awaited inline in the read loop, not in a background task that could outlive its 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 a false operator-facing warning that discovery/membership tracking was DOWN while discovery then completed normally. 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), `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 so one can be replaced or closed without disturbing the others, 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. **Backlog paging:** freeze the read floor and track `_backlog_until`; stop/warn on stalled timestamps, never skip a tied second. Removal/live close clears paging and closes backlog; ignore stale frames. Reset cursors per socket. Bounds: [subscription model](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model). **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: one shared cursor is the newest event seen in *any* channel, so a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect. 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 diff --git a/backend/app/channels/buzz.py b/backend/app/channels/buzz.py index 416217534f..6ef6974e02 100644 --- a/backend/app/channels/buzz.py +++ b/backend/app/channels/buzz.py @@ -29,13 +29,17 @@ relay's stated reason suggests re-issuing the same REQ could work at all (see WARNING, because "listening to nothing" must never again be indistinguishable from "nothing is being said". -KNOWN BOUND (documented, not fixed): the relay caps historical delivery at 2000 -events per subscription and serves them NEWEST-FIRST, even with a ``since``. So a -channel that accumulated more than 2000 unread messages across one disconnect -loses the oldest of them: the relay never sends them, and the watermark advances -past them as the newer ones are processed. This is the one bounded skip path that -remains, it needs a disconnect plus >2000 messages in a single channel to trigger, -and closing it would require paging the backlog with descending ``until`` queries. +KNOWN BOUND (CLOSED): the relay caps historical delivery at 2000 events per +subscription and serves them NEWEST-FIRST, even with a ``since``, so a channel +that accumulated more than 2000 unread messages across one disconnect used to +lose the oldest of them -- the relay never sent them and the watermark advanced +past them. The live page's EOSE now drives a descending ``until`` walk over +separate backlog subscription ids (``_on_chat_page_complete``) until the page +comes back under the cap or reaches the ``since`` the walk started from, so the +oldest unread events are fetched instead of skipped. The walk is best effort and +additive: it never touches the live subscription's filter or the watermark, and +any failure simply leaves the pre-existing behaviour in place for the next +reconnect to retry. """ from __future__ import annotations @@ -88,6 +92,10 @@ MAX_CHANNEL_SUBSCRIPTIONS = 256 DISCOVERY_SUB_ID = "buzz-discovery" MEMBERSHIP_SUB_ID = "buzz-membership" CHAT_SUB_PREFIX = "buzz-chat-" +# Backlog pages ride their OWN subscription ids, never the live chat one. A REQ that +# carries ``until`` bounds every event the relay will send on that id, so reusing the +# live id would silently stop live delivery the moment a page was opened. +BACKLOG_SUB_PREFIX = "buzz-backlog-" # How many times ONE subscription may be re-opened after the relay ``CLOSED`` it, # per connection and per auth epoch. Re-issuing a closed REQ immediately and @@ -106,6 +114,18 @@ MAX_RESUBSCRIBE_ATTEMPTS = 3 # REQ into a connection that no longer exists. RESUBSCRIBE_BASE_DELAY_SECONDS = 1.0 +# The relay answers ONE subscription REQ with at most this many stored events, +# NEWEST-FIRST, even when the filter carries a ``since``. A page that comes back at +# exactly this size is saturated: the relay may be holding back older matching +# events. See ``_on_chat_page_complete``. +RELAY_HISTORY_LIMIT = 2000 + +# Bound on how many descending ``until`` pages one channel's backlog is walked back +# on a single connection. At ``RELAY_HISTORY_LIMIT`` events per page this covers +# ~100k messages; the bound exists so a relay that keeps answering "saturated" can +# never pin the read loop. +MAX_BACKLOG_PAGES = 50 + # How far BEFORE the socket opened the live membership subscription starts. # # Without any ``since`` the relay replays its whole stored 44100/44101 history on @@ -320,6 +340,16 @@ class BuzzChannel(Channel): 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 + # Backlog paging state (see `_on_chat_page_complete`): per-sub accounting of how + # much history a REQ was answered with, plus the floor the walk must stop at. + # Keyed by sub id because a live page and a backlog page for the same channel + # are different queries with different answers, and an event must be charged + # to the subscription that actually delivered it. + self._page_seen: dict[str, int] = {} # sub id -> stored events delivered since its REQ + self._page_oldest: dict[str, int] = {} # sub id -> oldest created_at among them + self._backlog_floor: dict[str, int | None] = {} # channel id -> the `since` its live REQ was sent with + self._backlog_pages: dict[str, int] = {} # channel id -> descending pages issued this connection + self._backlog_until: dict[str, int] = {} # channel id -> boundary requested by the current backlog page self._auth_completed = False # has THIS socket's NIP-42 handshake been ACKNOWLEDGED by the relay? (see _handle_auth_ok, _AUTH_REQUIRED_CLOSE_PREFIX) self._pending_auth_event_id: str | None = None # id of the AUTH event most recently sent, awaiting a matching OK (see _handle_auth_ok / _confirm_auth_if_pending) self._session_started_at: int | None = None # wall clock at which the CURRENT socket opened; anchors the live membership filter @@ -432,6 +462,7 @@ class BuzzChannel(Channel): self._channel_meta.clear() self._chat_subscriptions.clear() self._resubscribe_attempts.clear() + self._reset_backlog_state() self._auth_completed = False self._pending_auth_event_id = None self._session_started_at = None @@ -448,6 +479,71 @@ class BuzzChannel(Channel): def _chat_sub_id(channel_id: str) -> str: return f"{CHAT_SUB_PREFIX}{channel_id}" + @staticmethod + def _backlog_sub_id(channel_id: str) -> str: + return f"{BACKLOG_SUB_PREFIX}{channel_id}" + + def _reset_backlog_state(self) -> None: + """Drop every backlog-paging cursor, wherever per-socket state is reset. + + A page cursor only means anything for the connection that opened its REQ: a + new socket re-sends every subscription, so it must not believe an unfinished + walk from a dead one is still in flight. + """ + self._page_seen.clear() + self._page_oldest.clear() + self._backlog_floor.clear() + self._backlog_pages.clear() + self._backlog_until.clear() + + def _record_page_delivery(self, sub_id: str, ev: dict) -> None: + """Count one stored event against the subscription that delivered it. + + Counted HERE, from the ``EVENT`` frame's own sub id and BEFORE any gate, so the + count measures what the RELAY sent rather than what we chose to keep -- a page + is saturated or not regardless of how many of its events the allowlist or + mention gates drop. Reading the sub id (not the ``h`` tag) is also what keeps a + live event from being charged to an in-flight backlog page for the same channel. + """ + if sub_id not in self._page_seen or ev.get("kind") != buzz_nostr.KIND_CHAT: + return + self._page_seen[sub_id] += 1 + created_at = ev.get("created_at") + if not isinstance(created_at, int): + return + current = self._page_oldest.get(sub_id) + if current is None or created_at < current: + self._page_oldest[sub_id] = created_at + + def _backlog_filter(self, channel_id: str, floor: int | None, until: int) -> dict: + """The NIP-01 filter for ONE descending backlog page of *channel_id*. + + ``until`` is the oldest ``created_at`` the previous page carried, and it is + INCLUSIVE on purpose: NIP-01 ``until`` is an upper bound, so a strict + ``until - 1`` would drop any sibling event sharing that second with the page + boundary that the relay's cap had cut off. Re-fetching the boundary second + instead costs a replay the seen-id record absorbs, which is the direction this + connector fails in everywhere else. + + ``floor`` is the ``since`` the live REQ was sent with -- frozen at that moment, + NOT re-read from ``_seen_created_at`` here: processing the newest page advances + the watermark, and using the moved cursor would exclude exactly the older + events this query exists to fetch. + """ + page_filter: dict = {"kinds": [buzz_nostr.KIND_CHAT], "#h": [channel_id], "until": until} + if floor: + page_filter["since"] = floor + return page_filter + + @staticmethod + def _page_channel(sub_id: str) -> str | None: + """The channel a page's EOSE belongs to, or ``None`` for a sub we never page.""" + if sub_id.startswith(CHAT_SUB_PREFIX): + return sub_id[len(CHAT_SUB_PREFIX) :] or None + if sub_id.startswith(BACKLOG_SUB_PREFIX): + return sub_id[len(BACKLOG_SUB_PREFIX) :] or None + return None + def _chat_filter(self, channel_id: str) -> dict: """The NIP-01 filter for ONE channel's chat subscription. @@ -473,14 +569,12 @@ class BuzzChannel(Channel): our ``event_id`` metadata within this workspace) is what keeps that from starting a second run. - KNOWN BOUND: ``since`` narrows the query but does not lift the relay's cap on - how much history one subscription may be served -- buzz-relay answers with at - most 2000 stored events, NEWEST-FIRST. So if a single channel accumulated more - than 2000 messages while we were disconnected, the relay simply never sends - the oldest of them, and the watermark advances past them as the newer ones are - processed. That is the one bounded skip path that remains after this fix - (everything else fails toward replay); closing it would mean paging the - backlog with descending ``until`` queries rather than one REQ per channel. + ``since`` narrows the query but does not lift the relay's cap on how much + history one subscription may be served -- buzz-relay answers with at most + ``RELAY_HISTORY_LIMIT`` stored events, NEWEST-FIRST. A page that comes back + saturated is therefore walked back with descending ``until`` queries by + ``_on_chat_page_complete`` (on ``BACKLOG_SUB_PREFIX`` ids, never this one), so + the oldest unread events are reached instead of being skipped. """ chat_filter: dict = {"kinds": [buzz_nostr.KIND_CHAT], "#h": [channel_id]} since = self._seen_created_at.get(channel_id) @@ -554,19 +648,33 @@ class BuzzChannel(Channel): if len(self._chat_subscriptions) >= MAX_CHANNEL_SUBSCRIPTIONS: logger.warning("[buzz] per-channel subscription limit reached (%d); not listening to channel %s", MAX_CHANNEL_SUBSCRIPTIONS, channel_id) return + live_sub_id = self._chat_sub_id(channel_id) try: - await transport.send(buzz_nostr.req_frame(self._chat_sub_id(channel_id), self._chat_filter(channel_id))) + await transport.send(buzz_nostr.req_frame(live_sub_id, self._chat_filter(channel_id))) except Exception: logger.warning("[buzz] failed to subscribe to channel %s; will retry", channel_id, exc_info=True) return + # A transient CLOSED re-opens the live sub while a backlog page for the same + # channel may still be open, and re-opening the live REQ supersedes that walk: + # its cursor means nothing against this REQ's window, so close the orphaned page + # rather than leave a bounded subscription alive on the relay for the socket. + if self._backlog_pages.get(channel_id) is not None: + await self._finish_backlog_walk(channel_id, self._backlog_sub_id(channel_id)) + # Freeze this REQ's window and start counting what it delivers, so its EOSE can + # tell a saturated page (more history behind it) from a complete one. + self._backlog_floor[channel_id] = self._seen_created_at.get(channel_id) + self._page_seen[live_sub_id] = 0 + self._page_oldest.pop(live_sub_id, None) self._chat_subscriptions.add(channel_id) logger.info("[buzz] listening to channel %s (%s)", self._channel_meta.get(channel_id, {}).get("name") or "", channel_id) async def _close_chat_subscription(self, channel_id: str) -> None: """Stop listening to one channel, leaving every other subscription on this socket alone.""" - if channel_id not in self._chat_subscriptions: - return + subscribed = channel_id in self._chat_subscriptions self._chat_subscriptions.discard(channel_id) + await self._finish_backlog_walk(channel_id, self._chat_sub_id(channel_id)) + if not subscribed: + return transport = self._transport if transport is None: return @@ -625,15 +733,28 @@ class BuzzChannel(Channel): from the live set, because it genuinely is not subscribed from this moment and the auth branch/discovery sweep is what re-opens it. """ - if _is_auth_required_close(reason) and not self._auth_completed: + if _is_auth_required_close(reason) and not self._auth_completed and not sub_id.startswith(BACKLOG_SUB_PREFIX): if sub_id.startswith(CHAT_SUB_PREFIX): - self._chat_subscriptions.discard(sub_id[len(CHAT_SUB_PREFIX) :]) + channel_id = sub_id[len(CHAT_SUB_PREFIX) :] + self._chat_subscriptions.discard(channel_id) + await self._finish_backlog_walk(channel_id, sub_id) logger.debug("[buzz] relay closed %s pending NIP-42 auth (%s); the auth handshake re-opens it", sub_id, reason or "") return if sub_id == DISCOVERY_SUB_ID or sub_id == MEMBERSHIP_SUB_ID: await self._recover_control_subscription(sub_id, reason) elif sub_id.startswith(CHAT_SUB_PREFIX): await self._recover_chat_subscription(sub_id[len(CHAT_SUB_PREFIX) :], reason) + elif sub_id.startswith(BACKLOG_SUB_PREFIX): + # A backlog page is best effort: losing it only ends THIS connection's walk. + # The live subscription is untouched and the next reconnect re-walks from its + # EOSE, so there is nothing to recover and nothing alarming to report. + channel_id = sub_id[len(BACKLOG_SUB_PREFIX) :] + self._backlog_pages.pop(channel_id, None) + self._backlog_floor.pop(channel_id, None) + self._backlog_until.pop(channel_id, None) + self._page_seen.pop(sub_id, None) + self._page_oldest.pop(sub_id, None) + logger.debug("[buzz] relay closed backlog page %s: %s; the live subscription is unaffected", sub_id, reason or "") else: logger.info("[buzz] relay closed unrecognized subscription %s: %s", sub_id, reason or "") @@ -699,6 +820,7 @@ class BuzzChannel(Channel): logger.info("[buzz] relay closed chat subscription for channel %s, which we were not subscribed to: %s", channel_id, reason or "") return self._chat_subscriptions.discard(channel_id) + await self._finish_backlog_walk(channel_id, self._chat_sub_id(channel_id)) name = self._channel_meta.get(channel_id, {}).get("name") or "" logger.warning("[buzz] relay closed the chat subscription for channel %s (%s), so it is now UNLISTENED: %s", name, channel_id, reason or "") if not _is_transient_close(reason): @@ -781,6 +903,105 @@ class BuzzChannel(Channel): else: logger.warning("[buzz] channel discovery returned no channels: this identity is not a member of any channel on %s (add it with `buzz channels add-member`)", self._relay_url) + async def _on_chat_page_complete(self, sub_id: str) -> None: + """EOSE for a chat or backlog page: walk the backlog back if the page was saturated. + + The relay serves at most ``RELAY_HISTORY_LIMIT`` stored events per REQ, + newest-first, so a channel that accumulated more than that while we were + disconnected has its OLDEST unread events simply never sent -- and the + watermark advances past them as the newer ones are processed. That was the one + remaining path in this connector that could skip a message outright. + + A page that came back at exactly the cap is the signal: there may be older + matching events the relay held back. This re-asks with ``until`` set to the + oldest ``created_at`` that page carried -- on a SEPARATE backlog subscription id, + so the bounded query can never replace the live one -- and repeats until a page + comes back under the cap (the history is exhausted) or reaches the ``since`` the + walk started from (everything at or before that is already processed). + + If a saturated backlog page does not descend past its requested ``until``, + stop and warn instead of repeating the same inclusive timestamp bucket. + Moving ``until`` back a second could skip unread events in that bucket. + A channel removal or live-subscription close also ends the walk; queued + backlog events and EOSEs cannot continue it after cleanup. + + A floor (the live REQ's ``since``, frozen in ``_backlog_floor``) is REQUIRED: + without one the walk has no semantic stop short of ``MAX_BACKLOG_PAGES``, which + is the case on a channel's first connect (no stored watermark) and again after a + ``MAX_CACHED_CHANNELS`` eviction of its cursor. Walking there would turn the + newest-2000 first-connect replay into up to ``MAX_BACKLOG_PAGES * + RELAY_HISTORY_LIMIT`` events per channel, so the walk is skipped and first-connect + behaviour stays exactly as it was. The backlog-loss case this method exists for + always has a watermark (``_seen_created_at`` survives reconnects and restarts), + so it always has a floor. + + Best effort by construction: every exit here (no transport, a raised ``send``, + no floor, the page bound, a page under the cap) only means no further page is + fetched -- exactly the behaviour that existed before this method did. Nothing + about the live subscription, its filter, or the watermark is touched, so a + failure can never make delivery worse than it already was. + """ + channel_id = self._page_channel(sub_id) + if channel_id is None or channel_id not in self._chat_subscriptions or sub_id not in self._page_seen: + return + oldest = self._page_oldest.get(sub_id) + saturated = oldest is not None and self._page_seen.get(sub_id, 0) >= RELAY_HISTORY_LIMIT + floor = self._backlog_floor.get(channel_id) + # No floor means no walk -- see the docstring: first connect and an evicted + # cursor have nothing to stop at, so they keep the newest-2000 behaviour. + if not saturated or floor is None or oldest <= floor: + await self._finish_backlog_walk(channel_id, sub_id) + return + previous_until = self._backlog_until.get(channel_id) + if sub_id == self._backlog_sub_id(channel_id) and previous_until is not None and oldest >= previous_until: + # NIP-01's inclusive `until` can return the same full timestamp bucket. + # Subtracting a second would skip unread events in that bucket. + logger.warning("[buzz] stopped paging channel %s backlog: no timestamp progress below created_at=%d; remaining history was not fetched on this connection", channel_id, previous_until) + await self._finish_backlog_walk(channel_id, sub_id) + return + pages = self._backlog_pages.get(channel_id, 0) + if pages >= MAX_BACKLOG_PAGES: + logger.warning("[buzz] stopped paging channel %s backlog after %d page(s); history older than created_at=%d was not fetched on this connection", channel_id, pages, oldest) + await self._finish_backlog_walk(channel_id, sub_id) + return + transport = self._transport + if transport is None: + return # no socket: the next connection re-walks this from the live page's EOSE + backlog_sub_id = self._backlog_sub_id(channel_id) + self._backlog_pages[channel_id] = pages + 1 + self._backlog_until[channel_id] = oldest + self._page_seen[backlog_sub_id] = 0 + self._page_oldest.pop(backlog_sub_id, None) + try: + await transport.send(buzz_nostr.req_frame(backlog_sub_id, self._backlog_filter(channel_id, floor, oldest))) + except Exception: + logger.warning("[buzz] failed to page channel %s backlog below created_at=%d; older history stays unread until the next reconnect", channel_id, oldest, exc_info=True) + await self._finish_backlog_walk(channel_id, sub_id) + return + logger.info("[buzz] paging back channel %s backlog (page %d) below created_at=%d", channel_id, self._backlog_pages[channel_id], oldest) + + async def _finish_backlog_walk(self, channel_id: str, sub_id: str) -> None: + """Clear a channel's walk state and close its backlog page if one is open. + + Closing matters: a backlog REQ carries ``until``, so leaving it open would keep + a bounded, non-live subscription alive on the relay for the rest of the socket's + life for no benefit. + """ + backlog_sub_id = self._backlog_sub_id(channel_id) + backlog_open = self._backlog_pages.pop(channel_id, None) is not None + self._backlog_floor.pop(channel_id, None) + self._backlog_until.pop(channel_id, None) + for tracked in {sub_id, self._chat_sub_id(channel_id), backlog_sub_id}: + self._page_seen.pop(tracked, None) + self._page_oldest.pop(tracked, None) + if backlog_open: + transport = self._transport + if transport is not None: + try: + await transport.send(buzz_nostr.close_frame(backlog_sub_id)) + except Exception: + logger.debug("[buzz] failed to close backlog page %s", backlog_sub_id, exc_info=True) + def _handle_auth_ok(self, event_id: str, accepted: bool, message: str) -> None: """React to a relay ``OK`` acknowledgment, specifically for our outstanding NIP-42 AUTH event. @@ -873,6 +1094,7 @@ class BuzzChannel(Channel): self._session_started_at = int(time.time()) self._chat_subscriptions.clear() self._resubscribe_attempts.clear() + self._reset_backlog_state() self._auth_completed = False self._pending_auth_event_id = None try: @@ -888,11 +1110,13 @@ class BuzzChannel(Channel): await ws.send(json.dumps(["AUTH", auth], separators=(",", ":"))) self._chat_subscriptions.clear() self._resubscribe_attempts.clear() + self._reset_backlog_state() await self._open_control_subscriptions(ws) finally: self._transport = None self._chat_subscriptions.clear() self._resubscribe_attempts.clear() + self._reset_backlog_state() self._auth_completed = False self._pending_auth_event_id = None self._session_started_at = None @@ -993,12 +1217,27 @@ class BuzzChannel(Channel): self._pending_auth_challenge = str(frame[1]) elif kind == "OK" and len(frame) >= 3: self._handle_auth_ok(str(frame[1]), frame[2] is True, str(frame[3]) if len(frame) >= 4 else "") - elif kind == "EOSE" and len(frame) >= 2 and frame[1] == DISCOVERY_SUB_ID: - await self._on_discovery_complete() + elif kind == "EOSE" and len(frame) >= 2: + sub_id = str(frame[1]) + if sub_id == DISCOVERY_SUB_ID: + await self._on_discovery_complete() + else: + # Only a chat/backlog page EOSE does anything here; the membership + # subscription's EOSE (and any unrecognized one) needs no action. + await self._on_chat_page_complete(sub_id) elif kind == "CLOSED" and len(frame) >= 2: await self._handle_closed(str(frame[1]), str(frame[2]) if len(frame) >= 3 else "") elif kind == "EVENT" and len(frame) >= 3 and isinstance(frame[2], dict): ev = frame[2] + sub_id = str(frame[1]) + if sub_id.startswith(BACKLOG_SUB_PREFIX): + channel_id = self._page_channel(sub_id) + if channel_id not in self._chat_subscriptions or sub_id not in self._page_seen: + return # queued frames from a closed or superseded backlog page + # Charge the delivery to the subscription that carried it BEFORE any gate + # (see `_record_page_delivery`), so a saturated page is recognised however + # many of its events the gates drop. + self._record_page_delivery(sub_id, ev) if not buzz_nostr.verify_event(ev): logger.warning("[buzz] dropped relay event with an invalid id/signature (claimed pubkey=%s kind=%s)", ev.get("pubkey"), ev.get("kind")) return diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 720a4cf100..24526b9c4a 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -420,7 +420,7 @@ Buzz's relay only delivers chat events to **channel-scoped** subscriptions, whic | Subscription | Filter | Purpose | |---|---|---| | `buzz-discovery` | `{"kinds":[39000]}` | Historical query listing exactly the channels this identity is a member of (one stored event each, then `EOSE`). Supplies each channel's name and type, which is also what the DM mention-exemption reads. Do **not** narrow it with `#p` — that matches nothing. | -| `buzz-membership` | `{"kinds":[44100,44101], "#p":[""], "since": …}` | **Live** membership notifications. `44100` (added) subscribes to the new channel immediately; `44101` (removed) closes that channel's subscription. This is what makes a newly added channel work without a restart. | +| `buzz-membership` | `{"kinds":[44100,44101], "#p":[""], "since": …}` | **Live** membership notifications. `44100` (added) subscribes to the new channel immediately; `44101` (removed) closes that channel's live and backlog subscriptions and clears its paging state. This is what makes a newly added channel work without a restart. | | `buzz-chat-` | `{"kinds":[9], "#h":[""], "since": …}` | One per discovered channel — the only shape that actually receives messages. | Consequences worth knowing operationally: @@ -429,7 +429,7 @@ Consequences worth knowing operationally: - **Membership is scoped to live events.** The relay *stores* 44100/44101 events, so without a `since` every connection replayed the whole membership history as if it had just happened — re-running channel discovery once per stored add (you would see several `channel discovery complete` lines for one connect, and channels logged as ``), re-subscribing channels you have since been removed from, and briefly unsubscribing channels you are still in. The subscription is therefore anchored at the moment the socket opened, minus 60s of slack so a membership change made *during* the connect/auth handshake — or a small relay clock skew — is still picked up. - **The number of channel subscriptions is capped** (256). The channel list comes off the wire, so it is bounded like any other remote-fed state. At the cap, new channels are refused and named in a `per-channel subscription limit reached` warning rather than an existing, working subscription being evicted. - **A subscription the relay closes is re-opened, up to 3 times per connection.** Every subscription on the socket fails *silently* when the relay drops it: a chat subscription deafens one channel, `buzz-membership` stops DeerFlow ever learning it was added to or removed from a channel, `buzz-discovery` kills the completeness sweep. So a `CLOSED` frame is recovered, not just noted, and the subscription that went quiet is always named at WARNING level. Recovery is skipped when the relay's stated reason says the subscription is not ours any more — a NIP-01/NIP-42 `auth-required:` / `restricted:` / `blocked:` / `invalid:` prefix, or buzz-relay's own revocation wording — because re-issuing the same REQ then just fights the relay. Any other reason (including a `CLOSED` with no reason at all) is treated as a hiccup and retried, with the 3-attempt budget as the backstop; after that it stays down until the next reconnect, which rebuilds everything from scratch. -- **Known bound: more than 2000 unread messages in one channel across a disconnect loses the oldest of them.** The relay caps historical delivery at 2000 events per subscription and serves them newest-first, even with a `since`. DeerFlow processes what it receives and the channel's watermark advances past the rest, so those older messages are never delivered and never retried. Every other gap in the design fails toward duplicate delivery (which is absorbed by inbound dedupe); this is the one remaining case that can skip, and it needs both a disconnect and a >2000-message backlog in a *single* channel to occur. +- **A backlog larger than the relay's 2000-per-subscription cap is now paged, not lost.** The relay serves at most 2000 stored events per subscription, newest-first, even with a `since`. When a channel's history page comes back *at* that cap, DeerFlow walks the history backwards with further `until`-bounded queries on a separate subscription until a page comes back under the cap or it reaches the point it had already read, so the oldest unread messages are fetched rather than skipped. The walk needs a read watermark to stop at, so it runs only for a channel DeerFlow has already read from; a channel's *first* connect (and any channel whose cursor was evicted) keeps the previous newest-2000 behaviour rather than paging back with nothing to reach. The walk is best-effort and additive. It stops with a warning at 50 pages or when a saturated page makes no timestamp progress. A full page of events at the inclusive `until` boundary cannot be advanced safely using timestamps alone: subtracting a second could skip unread events with that timestamp, so DeerFlow stops instead of repeatedly fetching the same page. Remaining history may be unfetched. Membership removal or a closed live subscription closes any open backlog page and clears its state; late backlog frames are ignored. ### Buzz trust model diff --git a/backend/tests/test_buzz_channel.py b/backend/tests/test_buzz_channel.py index 8ea20d79cf..31eabfcff0 100644 --- a/backend/tests/test_buzz_channel.py +++ b/backend/tests/test_buzz_channel.py @@ -1579,6 +1579,10 @@ def test_session_end_drops_chat_subscription_bookkeeping(): asyncio.run(ch._session(ws)) assert ch._chat_subscriptions == set() assert ch._transport is None + # Backlog page cursors are per-socket too: a new socket must not think an + # unfinished walk from a dead one is still in flight. + assert ch._backlog_pages == {} and ch._backlog_floor == {} + assert ch._page_seen == {} and ch._page_oldest == {} # -- RESILIENCE REVIEW: a CLOSED subscription must not deafen the connector -------- @@ -2181,3 +2185,295 @@ def test_ordinary_replies_that_merely_mention_the_words_are_still_published(): ch._transport = transport asyncio.run(ch.send(_outbound(ch, "I stored that in memory and in the durable context data.", is_final=True))) assert len(_events_of(transport)) == 1 + + +# -- BACKLOG PAGING: a saturated history page must not skip the OLDEST messages ---- +# +# buzz-relay serves stored events NEWEST-FIRST and caps a single subscription at +# RELAY_HISTORY_LIMIT (2000). A channel that banked more than the cap across one +# disconnect therefore had its OLDEST unread events neither sent nor marked: the +# live page delivered the newest 2000, the watermark advanced past them, and the +# rest were silently lost. The fix is additive -- the saturated live page's EOSE +# drives a descending `until` walk on SEPARATE `buzz-backlog-` sub ids +# until a page comes back under the cap or reaches the `since` the live REQ was +# sent with. The live subscription, its filter, and the watermark are never +# touched, so a failed walk can only ever leave the pre-existing behaviour. + + +def test_saturated_live_page_walks_back_and_delivers_the_previously_skipped_event(monkeypatch): + """A page at the cap pages back, and the older message it fetches is delivered. + + Both halves of the fix in one shot: the saturated live page issues a + descending `until` backlog query on its own sub id (windowed to the boundary + second, floored at the live REQ's `since`), and the event that query returns + -- older than anything the live page carried -- reaches publish instead of + being skipped. The live REQ itself is never re-issued or altered. + """ + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 2, raising=False) + ch, captured = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000000 # an established cursor: the walk's `since` floor + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + assert _reqs_by_sub(transport)[live_sub] == [{"kinds": [9], "#h": [CHANNEL], "since": 1700000000}] + + # The live page comes back at exactly the cap, newest-first. + newest = _event(created_at=1700000300) + boundary = _event(created_at=1700000200) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, newest]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, boundary]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + + # Saturated -> one descending backlog REQ on a SEPARATE sub id, `until` + # INCLUSIVE on the boundary second and floored at the live REQ's `since`. + backlog_sub = f"buzz-backlog-{CHANNEL}" + assert _reqs_by_sub(transport)[backlog_sub] == [{"kinds": [9], "#h": [CHANNEL], "until": 1700000200, "since": 1700000000}] + # ... and the live REQ was neither re-issued nor rewritten: the walk is additive. + assert len(_reqs_by_sub(transport)[live_sub]) == 1 + + # The previously-skipped older message arrives on the backlog page. + skipped = _event(created_at=1700000100) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, skipped]))) + assert [m.metadata["event_id"] for m in captured] == [newest["id"], boundary["id"], skipped["id"]] + + # A page under the cap ends the walk: state cleared, bounded sub closed. + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", backlog_sub]))) + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + assert backlog_sub in _closed_subs(transport) + + +def test_a_live_page_under_the_cap_ends_the_walk_with_no_extra_traffic(monkeypatch): + """A page below the cap proves the history is exhausted: no backlog query at all.""" + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 2, raising=False) + ch, _ = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000000 + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000200)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + + assert not [f for f in transport.sent if f[0] == "REQ" and str(f[1]).startswith("buzz-backlog-")] + assert CHANNEL not in ch._backlog_pages + + +def test_backlog_walk_stops_when_the_page_reaches_the_live_since_floor(monkeypatch): + """A saturated backlog page that touches the floor ends the walk -- nothing older exists.""" + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 2, raising=False) + ch, _ = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000200 # the floor the walk must stop at + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000400)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000300)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + backlog_sub = f"buzz-backlog-{CHANNEL}" + assert len(_reqs_for(transport, backlog_sub)) == 1 + + # The backlog page is saturated too, but its oldest lands ON the floor. + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, _event(created_at=1700000300)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, _event(created_at=1700000200)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", backlog_sub]))) + + # Reached the floor: no further page, state cleared, bounded sub closed. + assert len(_reqs_for(transport, backlog_sub)) == 1 + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + assert backlog_sub in _closed_subs(transport) + + +def test_backlog_walk_is_bounded_by_max_pages_and_says_what_it_dropped(monkeypatch, caplog): + """A real walk that never reaches its floor is bounded; the bound is logged, not silent. + + A relay that keeps answering "saturated" while the walk advances towards an + ever-lower ``until`` could otherwise be paged forever. With a floor present the + walk still needs a hard cap; hitting it ends the walk and names the oldest + ``created_at`` that was not fetched, so a truncated walk never reads as a + complete one. + """ + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 1, raising=False) + monkeypatch.setattr("app.channels.buzz.MAX_BACKLOG_PAGES", 2, raising=False) + caplog.set_level(logging.WARNING) + ch, _ = _started() + transport = FakeTransport() + ch._transport = transport + # A floor exists, but every page stays above it: only MAX_BACKLOG_PAGES stops the walk. + ch._seen_created_at[CHANNEL] = 1700000000 + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000500)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + + backlog_sub = f"buzz-backlog-{CHANNEL}" + for created_at in (1700000400, 1700000300): # pages 1 and 2, each saturated, never reaching the floor + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, _event(created_at=created_at)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", backlog_sub]))) + + # Exactly two pages were issued; the bound stopped a third. + assert len(_reqs_for(transport, backlog_sub)) == 2 + assert any("stopped paging" in r.getMessage() for r in caplog.records) + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + assert backlog_sub in _closed_subs(transport) + + +def test_no_floor_first_connect_never_walks_the_backlog(monkeypatch): + """A saturated page with NO floor is left alone: first connect keeps its old shape. + + Without a stored watermark there is nothing for the walk to stop at, so paging + would fetch up to MAX_BACKLOG_PAGES * RELAY_HISTORY_LIMIT events on a channel's + FIRST connect -- a large multiple of the newest-2000 replay the relay already + sends. The walk is skipped instead, and the saturated page's own events are still + delivered exactly as before. + """ + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 1, raising=False) + ch, captured = _started() + transport = FakeTransport() + ch._transport = transport + # No watermark -> floor is None -> no walk. + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + only = _event(created_at=1700000500) + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, only]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + + # The live page was saturated (it hit the cap), but no backlog page was opened ... + assert not [f for f in transport.sent if f[0] == "REQ" and str(f[1]).startswith("buzz-backlog-")] + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + # ... and its own events still reached publish, so first-connect delivery is unchanged. + assert [m.metadata["event_id"] for m in captured] == [only["id"]] + + +@pytest.mark.parametrize("reason", ["error: gone", "auth-required: authenticate first"]) +def test_a_closed_backlog_page_only_drops_the_walk_and_never_touches_the_live_subscription(monkeypatch, reason): + """A relay CLOSED for a backlog page is best effort: the live subscription survives.""" + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 1, raising=False) + ch, _ = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000000 # a floor, so the saturated live page does open a walk + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000500)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + backlog_sub = f"buzz-backlog-{CHANNEL}" + assert len(_reqs_for(transport, backlog_sub)) == 1 + + asyncio.run(ch.handle_relay_frame(json.dumps(["CLOSED", backlog_sub, reason]))) + + # The walk state is dropped, but the channel stays subscribed and is NOT + # re-subscribed (a backlog close is not a chat-subscription outage). + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + assert CHANNEL not in ch._backlog_until + assert CHANNEL in ch._chat_subscriptions + assert len(_reqs_by_sub(transport)[live_sub]) == 1 + + +def test_reopening_the_live_subscription_closes_an_in_flight_backlog_page(monkeypatch): + """A transient close mid-walk re-opens the live sub and must not orphan the page. + + The live REQ's window is what the walk pages against; when the relay closes the + live subscription the walk is superseded, so re-opening it closes the bounded + backlog page instead of leaving an orphaned subscription on the relay. + """ + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 1, raising=False) + ch, _ = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000000 # a floor, so the saturated live page does open a walk + + asyncio.run(ch._ensure_chat_subscription(CHANNEL)) + live_sub = f"buzz-chat-{CHANNEL}" + asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=1700000500)]))) + asyncio.run(ch.handle_relay_frame(json.dumps(["EOSE", live_sub]))) + backlog_sub = f"buzz-backlog-{CHANNEL}" + assert len(_reqs_for(transport, backlog_sub)) == 1 + assert CHANNEL in ch._backlog_pages + + # The relay transiently closes the live subscription while that page is open. + asyncio.run(ch.handle_relay_frame(json.dumps(["CLOSED", live_sub, "error: hiccup"]))) + + # The live REQ is re-issued, and the superseded backlog page is closed, not left open. + assert len(_reqs_for(transport, live_sub)) == 2 + assert backlog_sub in _closed_subs(transport) + assert CHANNEL not in ch._backlog_pages + # ... and the new live REQ starts a fresh window, keyed to the current cursor. + assert ch._backlog_floor[CHANNEL] == 1700000500 + + +async def _open_backlog_walk(monkeypatch): + monkeypatch.setattr("app.channels.buzz.RELAY_HISTORY_LIMIT", 2) + ch, captured = _started() + transport = FakeTransport() + ch._transport = transport + ch._seen_created_at[CHANNEL] = 1700000000 + await ch._ensure_chat_subscription(CHANNEL) + live_sub = ch._chat_sub_id(CHANNEL) + for created_at in (1700000500, 1700000400): + await ch.handle_relay_frame(json.dumps(["EVENT", live_sub, _event(created_at=created_at)])) + await ch.handle_relay_frame(json.dumps(["EOSE", live_sub])) + return ch, transport, captured + + +@pytest.mark.asyncio +@pytest.mark.parametrize("ending", ["membership", "restricted", "pre-auth"]) +async def test_channel_access_ending_closes_backlog_and_ignores_stale_frames(monkeypatch, ending): + ch, transport, captured = await _open_backlog_walk(monkeypatch) + await ch._ensure_chat_subscription(CHANNEL_B) + other_live_sub = ch._chat_sub_id(CHANNEL_B) + backlog_sub = ch._backlog_sub_id(CHANNEL) + live_sub = ch._chat_sub_id(CHANNEL) + if ending == "membership": + await ch.handle_relay_frame(json.dumps(["EVENT", "buzz-membership", _membership_event(buzz_nostr.KIND_MEMBER_REMOVED)])) + assert live_sub in _closed_subs(transport) + else: + reason = "restricted: not a member" if ending == "restricted" else "auth-required: authenticate first" + await ch.handle_relay_frame(json.dumps(["CLOSED", live_sub, reason])) + assert backlog_sub in _closed_subs(transport) + assert CHANNEL not in ch._chat_subscriptions + assert CHANNEL_B in ch._chat_subscriptions + assert other_live_sub not in _closed_subs(transport) + assert other_live_sub in ch._page_seen + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_floor + assert live_sub not in ch._page_seen and backlog_sub not in ch._page_seen + assert live_sub not in ch._page_oldest and backlog_sub not in ch._page_oldest + assert CHANNEL not in ch._backlog_until + before = len(captured) + for created_at in (1700000300, 1700000200): + await ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, _event(created_at=created_at)])) + for sub_id in (backlog_sub, live_sub): + await ch.handle_relay_frame(json.dumps(["EOSE", sub_id])) + assert len(captured) == before + assert len(_reqs_for(transport, backlog_sub)) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("created_at", [1700000400, 1700000450]) +async def test_backlog_without_timestamp_progress_stops_and_warns(monkeypatch, caplog, created_at): + ch, transport, _ = await _open_backlog_walk(monkeypatch) + backlog_sub = ch._backlog_sub_id(CHANNEL) + caplog.set_level(logging.WARNING, logger="app.channels.buzz") + # Equal timestamps saturate the inclusive boundary; newer timestamps violate + # the requested window. Neither response can advance a descending walk. + for i in range(2): + event = _event(created_at=created_at, content=f"@DeerFlow tied {i}") + await ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, event])) + await ch.handle_relay_frame(json.dumps(["EOSE", backlog_sub])) + assert len(_reqs_for(transport, backlog_sub)) == 1 + assert backlog_sub in _closed_subs(transport) + assert CHANNEL in ch._chat_subscriptions + assert CHANNEL not in ch._backlog_pages and CHANNEL not in ch._backlog_until + assert "no timestamp progress" in caplog.text + assert "1700000400" in caplog.text + # Late EOSE/events must not restart a finished walk. + await ch.handle_relay_frame(json.dumps(["EVENT", backlog_sub, _event(created_at=1700000300)])) + await ch.handle_relay_frame(json.dumps(["EOSE", backlog_sub])) + assert len(_reqs_for(transport, backlog_sub)) == 1 diff --git a/backend/tests/test_buzz_seen_events.py b/backend/tests/test_buzz_seen_events.py index 2934f9ca8d..92fd003508 100644 --- a/backend/tests/test_buzz_seen_events.py +++ b/backend/tests/test_buzz_seen_events.py @@ -291,7 +291,8 @@ def test_channel_stop_flushes_pending_records(tmp_path): assert BuzzSeenEventStore(path).seen(CHANNEL, str(ev["id"])) -def test_service_wiring_injects_persistent_store_path(tmp_path, monkeypatch): +@pytest.mark.parametrize("custom_path", [None, "/custom/seen.json"]) +def test_service_wiring_injects_persistent_store_path(tmp_path, monkeypatch, custom_path): """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 @@ -313,14 +314,14 @@ def test_service_wiring_injects_persistent_store_path(tmp_path, monkeypatch): 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})) + service._running = True # _start_channel is admitted only by a running service + config = {"relay_url": "wss://x", "private_key": SK3_HEX} + if custom_path is not None: + config["seen_event_store_path"] = custom_path + started = asyncio.run(service._start_channel("buzz", config)) assert started - expected = str(tmp_path / "channels" / "buzz_seen_events.json") + expected = custom_path or 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):