mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
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-<uuid>` 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 <willem.jiang@gmail.com>
This commit is contained in:
FanouZeng-TT
GitHub
parent
28a06aa777
commit
064bb50602
@ -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 |
|
||||
|---------|-----------|------------|
|
||||
|
||||
File diff suppressed because one or more lines are too long
@ -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 "<unnamed>", 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 "<no reason>")
|
||||
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 "<no reason>")
|
||||
else:
|
||||
logger.info("[buzz] relay closed unrecognized subscription %s: %s", sub_id, reason or "<no reason>")
|
||||
|
||||
@ -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 "<no reason>")
|
||||
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 "<unnamed>"
|
||||
logger.warning("[buzz] relay closed the chat subscription for channel %s (%s), so it is now UNLISTENED: %s", name, channel_id, reason or "<no reason>")
|
||||
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
|
||||
|
||||
@ -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":["<our pubkey>"], "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":["<our pubkey>"], "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-<uuid>` | `{"kinds":[9], "#h":["<uuid>"], "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 `<unnamed>`), 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
|
||||
|
||||
|
||||
@ -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-<channel>` 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
|
||||
|
||||
@ -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):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user