github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-09-09 21:49:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(buzz): move seen-event persistence off event loop (#5103)

Browse Source 
* fix(buzz): move seen-event persistence off event loop

* fix(buzz): address seen-event persistence review

* fix(buzz): replace stale scheduled flush tasks

* fix(buzz): harden final seen-event flush

* fix(buzz): make seen-event shutdown retryable

* fix(buzz): quiesce persistence after channel stop

* fix(buzz): drain late events on repeated stop

---------

Co-authored-by: zaoshangduziteng <309590849+zaoshangduziteng@users.noreply.github.com>
This commit is contained in:
早上肚子疼 2026-08-31 23:18:34 +08:00 committed by GitHub
parent 1af79c7bcf
commit 3b601922ff
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 672 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/app/channels/AGENTS.md
View File

File diff suppressed because one or more lines are too long

33
backend/app/channels/buzz.py
View File

@ -289,6 +289,12 @@ class BuzzChannel(Channel):
self._session_started_at: int | None = None # wall clock at which the CURRENT socket opened; anchors the live membership filter
self._transport: Any = None
self._task: asyncio.Task | None = None
# Transport admission and cleanup completion are separate lifecycle
# states: Gateway may cancel stop() after ``_running`` is cleared, and
# ChannelService retains this instance specifically so cleanup can be
# retried.
self._stop_complete = True
self._seen_events.quiesce()
self._publish = self.bus.publish_inbound # test seam (discord.py idiom)
@property
@ -308,9 +314,11 @@ class BuzzChannel(Channel):
# message looks exactly like a broken relay to the operator. Say so
# once, loudly, at the only point where it is actionable.
logger.warning("[buzz] channels.buzz.allowed_users is empty: EVERY inbound chat message will be dropped (Buzz denies by default). Add member pubkeys (hex or npub) to enable the channel.")
self._stop_complete = False
self.bus.subscribe_outbound(self._on_outbound)
self._spawn_connection()
self._running = True
self._seen_events.resume()
logger.info("[buzz] channel started (relay=%s pubkey=%s allowed_users=%d)", self._relay_url, self._keys.pubkey_hex, len(self._allowed_users))
def _spawn_connection(self) -> None:
@ -343,9 +351,23 @@ class BuzzChannel(Channel):
still subscribed to anything -- from a previous process lifetime. The
per-channel replay cursors (``_seen_created_at``) deliberately survive, so
a restart resumes where it left off instead of replaying every channel.
``_running`` closes transport admission at the start of teardown, while
``_stop_complete`` is set only after the final seen-event flush. Keeping
those states separate lets ChannelService retry this same instance when
its outer shutdown timeout cancels ``stop()`` mid-cleanup.
Seen-event scheduling is quiesced before even the already-stopped guard.
A relay task abandoned after the bounded wait can therefore record late
ids as dirty state, but cannot attach new callbacks to a channel the
service may remove. A repeated ``stop()`` drains that dirty state, while
``start()`` explicitly resumes automatic scheduling.
"""
if not self._running:
self._seen_events.quiesce()
if not self._running and self._stop_complete:
await self._seen_events.aflush()
return
self._stop_complete = False
self._running = False
self.bus.unsubscribe_outbound(self._on_outbound)
if self._task is not None:
@ -380,7 +402,8 @@ class BuzzChannel(Channel):
self._pending_auth_challenge = None
# Seen-id persistence is coalesced (FLUSH_DELAY_SECONDS); a clean stop
# must not lose records still inside that window to a replay on restart.
self._seen_events.flush()
await self._seen_events.aflush()
self._stop_complete = True
logger.info("[buzz] channel stopped")
# -- subscriptions ------------------------------------------------------
@ -1255,7 +1278,7 @@ class BuzzChannel(Channel):
# created_at — is never affected. Sits before the /connect branch
# deliberately: a replayed /connect would otherwise be re-answered with
# a spurious "code invalid or expired" reply on every reconnect.
if self._seen_events.seen(channel_id, event_id):
if await self._seen_events.aseen(channel_id, event_id):
logger.debug("[buzz] dropped replayed event id=%s in channel %s", event_id, channel_id)
return
@ -1279,7 +1302,7 @@ class BuzzChannel(Channel):
# cursor: leaving it behind would replay this /connect on every reconnect
# and answer each replay with a spurious "code invalid or expired" reply.
self._advance_watermark(channel_id, created_at)
self._seen_events.record(channel_id, event_id)
await self._seen_events.arecord(channel_id, event_id)
return
if author not in self._allowed_users:
# Deny-by-default is intentional (see start()'s empty-allowlist warning),
@ -1315,7 +1338,7 @@ class BuzzChannel(Channel):
# same rule applies to the persistent seen-id record: recording before a
# failed publish would turn "replayable" into "silently skipped".
self._advance_watermark(channel_id, created_at)
self._seen_events.record(channel_id, event_id)
await self._seen_events.arecord(channel_id, event_id)
# -- outbound --------------------------------------------------------------

248
backend/app/channels/buzz_seen_events.py
View File

@ -28,15 +28,24 @@ channel are dropped, so a relay backlog deeper than that would re-answer the
tail. If a relay ever serves a deeper default backlog, raise
``MAX_IDS_PER_CHANNEL`` here.
Writes are coalesced: ``record()`` marks the store dirty and schedules one
Writes are coalesced: ``arecord()`` marks the store dirty and schedules one
flush per ``FLUSH_DELAY_SECONDS`` on the running event loop, so a reconnect
backlog burst pays one O(store) file write instead of one per event. With no
running loop (tests, tooling) ``record()`` flushes synchronously, and
``BuzzChannel.stop()`` flushes pending state on shutdown. This class is not
thread-safe by design: everything runs on the single channel event loop
(mutation in ``_handle_chat_event``, the coalesced flush via ``call_later`` on
that same loop). Anyone adding an off-loop user — e.g. a threaded flusher —
must add a lock around ``_ids``/``_sets`` first, as ChannelStore does.
backlog burst pays one O(store) file write instead of one per event. The timer
captures an immutable payload on the event loop and writes it through
``asyncio.to_thread``; a generation counter keeps records that arrive during
that write dirty for the next flush. ``aseen()`` likewise offloads the initial
file load, and ``BuzzChannel.stop()`` awaits ``aflush()`` so clean shutdown is
durable before it returns. That final flush is bounded: it waits for an existing
write and attempts at most one newer snapshot, leaving any still-moving
generation dirty for fail-open replay rather than hanging shutdown. The
in-flight worker write is shielded and retained if Gateway cancellation
interrupts shutdown, so a retried stop awaits it instead of racing it with a
second snapshot; every final attempt also removes its coalescing timer before
returning. ``BuzzChannel`` quiesces automatic scheduling before stop and resumes
it after start, so a timed-out relay task that records after the stop boundary
leaves data dirty for fail-open replay without creating detached file work. The
synchronous ``seen()`` / ``record()`` / ``flush()`` methods remain for tests and
tooling that run outside an event loop.
"""
from __future__ import annotations
@ -45,6 +54,7 @@ import asyncio
import json
import logging
import tempfile
import threading
from collections import OrderedDict, deque
from pathlib import Path
@ -64,7 +74,11 @@ MAX_CHANNELS = 512
class BuzzSeenEventStore:
"""Bounded, JSON-persisted map of channel id -> recently processed event ids."""
"""Bounded, JSON-persisted map of channel id -> recently processed event ids.
Gateway event-loop callers must use ``aseen()``, ``arecord()``, and
``aflush()`` so filesystem access stays on a worker thread.
"""
def __init__(self, path: str | Path | None = None) -> None:
# ``path=None`` means memory-only: no file is read or written, which is
@ -76,8 +90,12 @@ class BuzzSeenEventStore:
self._ids: OrderedDict[str, deque[str]] = OrderedDict()
self._sets: dict[str, set[str]] = {}
self._loaded = False
self._load_lock = threading.Lock()
self._dirty = False
self._generation = 0
self._quiesced = False
self._flush_handle: asyncio.TimerHandle | None = None
self._flush_task: asyncio.Task[bool] | None = None
# The loop the pending handle was scheduled on. TimerHandle has no
# public get_loop(), so it is tracked here to detect a stale handle.
self._flush_loop: asyncio.AbstractEventLoop | None = None
@ -85,36 +103,43 @@ class BuzzSeenEventStore:
# -- persistence ---------------------------------------------------------
def _ensure_loaded(self) -> None:
if self._loaded:
return
self._loaded = True
if self._path is None:
return
try:
if not self._path.exists():
with self._load_lock:
if self._loaded:
return
raw = json.loads(self._path.read_text(encoding="utf-8"))
except Exception:
logger.warning("[buzz] unreadable seen-event store, starting fresh (costs at most one replayed reply)", exc_info=True)
return
if not isinstance(raw, dict):
return
for channel_id, ids in raw.items():
if not isinstance(ids, list):
continue
clean = deque((str(i) for i in ids if i), maxlen=MAX_IDS_PER_CHANNEL)
self._ids[str(channel_id)] = clean
self._sets[str(channel_id)] = set(clean)
self._enforce_channel_cap()
try:
if self._path is None:
return
try:
if not self._path.exists():
return
raw = json.loads(self._path.read_text(encoding="utf-8"))
except Exception:
logger.warning("[buzz] unreadable seen-event store, starting fresh (costs at most one replayed reply)", exc_info=True)
return
if not isinstance(raw, dict):
return
for channel_id, ids in raw.items():
if not isinstance(ids, list):
continue
clean = deque((str(i) for i in ids if i), maxlen=MAX_IDS_PER_CHANNEL)
self._ids[str(channel_id)] = clean
self._sets[str(channel_id)] = set(clean)
self._enforce_channel_cap()
finally:
# Publish only after every loaded entry is visible. Async hot
# paths may read this flag without taking ``_load_lock``.
self._loaded = True
def _save(self) -> None:
def _snapshot(self) -> dict[str, list[str]]:
return {channel: list(ids) for channel, ids in self._ids.items()}
def _write_snapshot(self, payload: dict[str, list[str]]) -> bool:
if self._path is None:
return
return True
tmp_name: str | None = None
try:
path = self._path
path.parent.mkdir(parents=True, exist_ok=True)
payload = {channel: list(ids) for channel, ids in self._ids.items()}
# Atomic same-directory replace, matching ChannelStore._save: a
# crash mid-write must never truncate the store (a truncated store
# would fail open into replay on the next start, which is
@ -123,13 +148,27 @@ class BuzzSeenEventStore:
tmp_name = fh.name
json.dump(payload, fh)
Path(tmp_name).replace(path)
self._dirty = False
return True
except Exception:
# Mirror ChannelStore._save: never leave the temp file behind, or a
# persistently failing write accumulates one *.tmp per attempt.
if tmp_name is not None:
Path(tmp_name).unlink(missing_ok=True)
logger.warning("[buzz] failed to persist seen-event store (will retry on next flush)", exc_info=True)
return False
def _save(self) -> None:
if self._write_snapshot(self._snapshot()):
self._dirty = False
async def _flush_once(self) -> bool:
if not self._dirty:
return True
generation = self._generation
saved = await asyncio.to_thread(self._write_snapshot, self._snapshot())
if saved and generation == self._generation:
self._dirty = False
return saved
def _request_flush(self) -> None:
"""Coalesce persistence: at most one write per FLUSH_DELAY_SECONDS.
@ -139,6 +178,8 @@ class BuzzSeenEventStore:
callers had before coalescing existed.
"""
self._dirty = True
if self._quiesced:
return
try:
loop = asyncio.get_running_loop()
except RuntimeError:
@ -153,10 +194,127 @@ class BuzzSeenEventStore:
self._flush_loop = loop
self._flush_handle = loop.call_later(FLUSH_DELAY_SECONDS, self._flush_scheduled)
def quiesce(self) -> None:
"""Disable automatic persistence while retaining late records as dirty."""
self._quiesced = True
if self._flush_handle is not None:
self._flush_handle.cancel()
self._flush_handle = None
def resume(self) -> None:
"""Resume automatic persistence and schedule any retained dirty state."""
if not self._quiesced:
return
self._quiesced = False
if self._dirty:
self._request_flush()
def _flush_task_on_current_loop(self) -> asyncio.Task[bool] | None:
"""Return the pending flush only when it belongs to the running loop."""
pending = self._flush_task
if pending is not None and pending.get_loop() is not asyncio.get_running_loop():
self._flush_task = None
return None
return pending
def _flush_scheduled(self) -> None:
self._flush_handle = None
if self._dirty:
self._save()
if not self._dirty:
return
pending = self._flush_task_on_current_loop()
if pending is not None and not pending.done():
return
self._flush_task = asyncio.create_task(self._flush_once())
self._flush_task.add_done_callback(self._flush_finished)
def _flush_finished(self, task: asyncio.Task[bool]) -> None:
if self._flush_task is task:
self._flush_task = None
try:
saved = task.result()
except asyncio.CancelledError:
return
except Exception:
logger.warning("[buzz] unexpected seen-event flush failure", exc_info=True)
return
# A successful snapshot may be stale when another event arrived while
# the worker thread was writing it. Schedule that newer generation for
# the next coalescing window. A failed write deliberately waits for the
# next record or explicit flush, preserving the store's fail-open retry
# policy instead of spinning on a broken filesystem every second.
if saved and self._dirty:
self._request_flush()
def _final_flush_finished(self, task: asyncio.Task[bool]) -> None:
"""Retire a stop-owned write without scheduling work after stop."""
if self._flush_task is task:
self._flush_task = None
try:
task.result()
except asyncio.CancelledError:
return
except Exception:
logger.warning("[buzz] unexpected final seen-event flush failure", exc_info=True)
async def aflush(self) -> None:
"""Make one bounded final persist without blocking the event loop."""
if self._flush_handle is not None:
self._flush_handle.cancel()
self._flush_handle = None
try:
pending = self._flush_task_on_current_loop()
if pending is not None and pending is not asyncio.current_task():
if not pending.cancelled():
# This scheduled write now belongs to channel teardown.
# Its normal callback would arm another timer when the
# snapshot is stale, possibly after a cancelled aflush()
# has already run its timer cleanup.
pending.remove_done_callback(self._flush_finished)
pending.remove_done_callback(self._final_flush_finished)
pending.add_done_callback(self._final_flush_finished)
try:
# Cancelling channel stop must not cancel the worker
# write: the thread keeps running regardless, and a
# retried stop must await that same write rather than
# start a concurrent stale snapshot.
await asyncio.shield(pending)
except asyncio.CancelledError:
current = asyncio.current_task()
if current is not None and current.cancelling():
raise
logger.warning("[buzz] in-flight seen-event flush was cancelled during final persist; retrying")
except Exception:
logger.warning("[buzz] in-flight seen-event flush failed during final persist; retrying", exc_info=True)
if self._flush_task is pending and pending.done():
self._flush_task = None
# One final snapshot keeps shutdown bounded even if an abandoned
# relay task is still recording. Track and shield it so a Gateway
# timeout leaves one retryable write instead of an untracked worker
# thread that can race the next stop attempt.
if self._dirty:
final = asyncio.create_task(self._flush_once())
self._flush_task = final
final.add_done_callback(self._final_flush_finished)
try:
await asyncio.shield(final)
except asyncio.CancelledError:
current = asyncio.current_task()
if current is not None and current.cancelling():
raise
logger.warning("[buzz] final seen-event flush was cancelled; leaving store dirty")
except Exception:
logger.warning("[buzz] final seen-event flush failed; leaving store dirty", exc_info=True)
if self._flush_task is final and final.done():
self._flush_task = None
finally:
# ``aflush`` is the channel-stop boundary: never leave a timer
# owned by a stopped channel. A record that raced the final
# snapshot remains dirty so a retried stop can persist it.
if self._flush_handle is not None:
self._flush_handle.cancel()
self._flush_handle = None
def flush(self) -> None:
"""Persist pending records now (no-op when nothing is dirty).
@ -184,11 +342,30 @@ class BuzzSeenEventStore:
self._ensure_loaded()
return event_id in self._sets.get(channel_id, ())
async def aseen(self, channel_id: str, event_id: str) -> bool:
"""Async counterpart of :meth:`seen` for Gateway event-loop callers."""
if not event_id:
return False
if not self._loaded:
await asyncio.to_thread(self._ensure_loaded)
return event_id in self._sets.get(channel_id, ())
def record(self, channel_id: str, event_id: str) -> None:
"""Record a fully processed event and schedule a coalesced persist."""
if not channel_id or not event_id:
return
self._ensure_loaded()
self._record_loaded(channel_id, event_id)
async def arecord(self, channel_id: str, event_id: str) -> None:
"""Record an event without performing file IO on the event loop."""
if not channel_id or not event_id:
return
if not self._loaded:
await asyncio.to_thread(self._ensure_loaded)
self._record_loaded(channel_id, event_id)
def _record_loaded(self, channel_id: str, event_id: str) -> None:
ids = self._ids.get(channel_id)
if ids is None:
ids = deque(maxlen=MAX_IDS_PER_CHANNEL)
@ -204,4 +381,5 @@ class BuzzSeenEventStore:
# Move the channel to the back so the channel-cap eviction is LRU-ish.
self._ids.move_to_end(channel_id)
self._enforce_channel_cap()
self._generation += 1
self._request_flush()

174
backend/tests/blocking_io/test_buzz_channel_seen_events.py Normal file
View File

@ -0,0 +1,174 @@
"""Regression coverage for Buzz replay persistence at the channel boundary."""
from __future__ import annotations
import asyncio
import json
import threading
from pathlib import Path
import pytest
from app.channels import buzz_nostr, buzz_seen_events
from app.channels.buzz import BuzzChannel
from app.channels.buzz_seen_events import BuzzSeenEventStore
from app.channels.message_bus import MessageBus
pytestmark = pytest.mark.asyncio
_BOT_PUBLIC = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9"
_OWNER_PUBLIC = "11" * 32
_CHANNEL_ID = "136852ee-63e1-49c2-8927-413b5ee8e5f7"
def _event() -> dict:
tags = [["h", _CHANNEL_ID], ["p", _BOT_PUBLIC]]
created_at = 1_700_000_100
content = "@DeerFlow hello"
return {
"id": buzz_nostr.event_id(_OWNER_PUBLIC, created_at, 9, tags, content),
"pubkey": _OWNER_PUBLIC,
"created_at": created_at,
"kind": 9,
"tags": tags,
"content": content,
# Signature verification is patched below. Keeping a correctly shaped
# event makes this test independent of the optional ``buzz`` extra, so
# the default blocking-I/O CI job cannot silently skip the regression.
"sig": "00" * 64,
}
def _channel(path: Path, *, seen_events: BuzzSeenEventStore | None = None) -> tuple[BuzzChannel, list]:
channel = BuzzChannel(
bus=MessageBus(),
config={
"relay_url": "wss://buzz.example.com",
"private_key": "unused-by-this-test",
"allowed_users": [_OWNER_PUBLIC],
"seen_event_store": seen_events or BuzzSeenEventStore(path),
},
)
channel._keys = buzz_nostr.NostrKeys(secret=b"", pubkey_hex=_BOT_PUBLIC)
published = []
async def publish(message) -> None:
published.append(message)
channel._publish = publish
return channel, published
async def test_channel_stop_persists_replay_guard_without_blocking(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A clean stop suppresses the same relay event after channel restart."""
monkeypatch.setattr(buzz_nostr, "verify_event", lambda _event: True)
path = tmp_path / "buzz-seen-events.json"
event_frame = json.dumps(["EVENT", "buzz-chat", _event()])
first, first_messages = _channel(path)
await first.handle_relay_frame(event_frame)
assert len(first_messages) == 1
first._running = True
first.bus.subscribe_outbound(first._on_outbound)
await first.stop()
restarted, restarted_messages = _channel(path)
await restarted.handle_relay_frame(event_frame)
assert restarted_messages == []
async def test_channel_stop_retries_seen_event_flush_after_cancellation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A Gateway timeout must leave seen-event cleanup retryable."""
path = tmp_path / "buzz-seen-events.json"
seen_events = BuzzSeenEventStore(path)
channel, _ = _channel(path, seen_events=seen_events)
write_started = threading.Event()
release_write = threading.Event()
retry_write_started = threading.Event()
write_snapshot = seen_events._write_snapshot
write_count = 0
def fail_cancelled_write(payload: dict[str, list[str]]) -> bool:
nonlocal write_count
write_count += 1
if write_count == 1:
write_started.set()
assert release_write.wait(timeout=2)
return False
retry_write_started.set()
return write_snapshot(payload)
seen_events._write_snapshot = fail_cancelled_write
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 60)
await seen_events.arecord(_CHANNEL_ID, "event-1")
channel._running = True
channel.bus.subscribe_outbound(channel._on_outbound)
first_stop = asyncio.create_task(channel.stop())
assert await asyncio.to_thread(write_started.wait, 2)
first_stop.cancel()
with pytest.raises(asyncio.CancelledError):
await first_stop
retry_stop = asyncio.create_task(channel.stop())
assert not await asyncio.to_thread(retry_write_started.wait, 0.05)
release_write.set()
await retry_stop
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-1")
async def test_abandoned_relay_records_are_drained_by_retried_stop_or_restart(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Late relay records stay unscheduled but remain reachable to cleanup."""
path = tmp_path / "buzz-seen-events.json"
seen_events = BuzzSeenEventStore(path)
channel, _ = _channel(path, seen_events=seen_events)
relay_started = asyncio.Event()
release_abandoned_relay = asyncio.Event()
async def cancellation_resistant_relay() -> None:
relay_started.set()
try:
await asyncio.Future()
except asyncio.CancelledError:
await release_abandoned_relay.wait()
await seen_events.arecord(_CHANNEL_ID, "late-event")
monkeypatch.setattr("app.channels.buzz.STOP_TIMEOUT_SECONDS", 0.01)
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01)
channel._task = asyncio.create_task(cancellation_resistant_relay())
channel._running = True
channel.bus.subscribe_outbound(channel._on_outbound)
await relay_started.wait()
abandoned_relay = channel._task
assert abandoned_relay is not None
await channel.stop()
release_abandoned_relay.set()
await abandoned_relay
assert await seen_events.aseen(_CHANNEL_ID, "late-event")
await asyncio.sleep(0.05)
stopped_view = BuzzSeenEventStore(path)
assert not await stopped_view.aseen(_CHANNEL_ID, "late-event")
await channel.stop()
retried_stop_view = BuzzSeenEventStore(path)
assert await retried_stop_view.aseen(_CHANNEL_ID, "late-event")
await seen_events.arecord(_CHANNEL_ID, "restart-event")
await asyncio.sleep(0.05)
still_stopped_view = BuzzSeenEventStore(path)
assert not await still_stopped_view.aseen(_CHANNEL_ID, "restart-event")
monkeypatch.setattr(buzz_nostr, "parse_private_key", lambda _value: buzz_nostr.NostrKeys(secret=b"", pubkey_hex=_BOT_PUBLIC))
channel._spawn_connection = lambda: None
await channel.start()
await asyncio.sleep(0.05)
restarted_view = BuzzSeenEventStore(path)
assert await restarted_view.aseen(_CHANNEL_ID, "restart-event")
await channel.stop()

250
backend/tests/blocking_io/test_buzz_seen_event_store.py Normal file
View File

@ -0,0 +1,250 @@
"""Regression coverage for Buzz replay-guard persistence on async paths.
Buzz receives and stops channels on the Gateway event loop. Persisting the
seen-event replay guard must therefore stay off that loop while still making a
clean stop durable before it returns.
"""
from __future__ import annotations
import asyncio
import threading
from pathlib import Path
import pytest
from app.channels import buzz_seen_events
from app.channels.buzz_seen_events import BuzzSeenEventStore
pytestmark = pytest.mark.asyncio
_CHANNEL_ID = "136852ee-63e1-49c2-8927-413b5ee8e5f7"
async def test_async_replay_guard_round_trips_without_blocking_the_event_loop(tmp_path: Path) -> None:
"""An async flush makes a recorded event visible to a fresh store."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
await store.arecord(_CHANNEL_ID, "event-1")
await store.aflush()
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-1")
async def test_loaded_store_does_not_dispatch_hot_path_lookups_to_a_thread(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Only the cold file load needs the shared worker pool."""
store = BuzzSeenEventStore(tmp_path / "buzz-seen-events.json")
assert not await store.aseen(_CHANNEL_ID, "event-1")
async def unexpected_to_thread(*_args, **_kwargs):
raise AssertionError("loaded seen-event store used the worker pool")
monkeypatch.setattr(asyncio, "to_thread", unexpected_to_thread)
await store.arecord(_CHANNEL_ID, "event-1")
assert await store.aseen(_CHANNEL_ID, "event-1")
async def test_event_recorded_during_a_final_write_remains_retryable(tmp_path: Path) -> None:
"""An in-flight snapshot must not mark a newer generation as clean."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
write_started = threading.Event()
release_write = threading.Event()
snapshots: list[dict[str, list[str]]] = []
write_snapshot = store._write_snapshot
def pause_first_write(payload: dict[str, list[str]]) -> bool:
snapshots.append(payload)
if len(snapshots) == 1:
write_started.set()
assert release_write.wait(timeout=2)
return write_snapshot(payload)
store._write_snapshot = pause_first_write
await store.arecord(_CHANNEL_ID, "event-1")
flush = asyncio.create_task(store.aflush())
assert await asyncio.to_thread(write_started.wait, 2)
await store.arecord(_CHANNEL_ID, "event-2")
release_write.set()
await flush
assert await store.aseen(_CHANNEL_ID, "event-2")
# The final flush deliberately cancels its trailing timer, but the newer
# generation stays dirty so a retried cleanup can make it durable.
await store.aflush()
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-1")
assert await restarted.aseen(_CHANNEL_ID, "event-2")
async def test_final_flush_is_bounded_while_records_keep_arriving(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Shutdown must return instead of chasing an active producer forever."""
store = BuzzSeenEventStore(tmp_path / "buzz-seen-events.json")
first_write_started = threading.Event()
release_first_write = threading.Event()
second_write_started = threading.Event()
release_second_write = threading.Event()
snapshots: list[dict[str, list[str]]] = []
write_snapshot = store._write_snapshot
def pause_bounded_writes(payload: dict[str, list[str]]) -> bool:
snapshots.append(payload)
if len(snapshots) == 1:
first_write_started.set()
assert release_first_write.wait(timeout=2)
elif len(snapshots) == 2:
second_write_started.set()
assert release_second_write.wait(timeout=2)
return write_snapshot(payload)
store._write_snapshot = pause_bounded_writes
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01)
await store.arecord(_CHANNEL_ID, "event-0")
assert await asyncio.to_thread(first_write_started.wait, 2)
# Leave enough time to finish the bounded flush before a timer armed by a
# late record can fire. Once aflush() returns, no third write may remain
# scheduled for a store whose owning channel has stopped.
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.05)
final_flush = asyncio.create_task(store.aflush())
await store.arecord(_CHANNEL_ID, "late-1")
release_first_write.set()
assert await asyncio.to_thread(second_write_started.wait, 2)
await store.arecord(_CHANNEL_ID, "late-2")
release_second_write.set()
await final_flush
await asyncio.sleep(0.1)
assert len(snapshots) == 2
assert await store.aseen(_CHANNEL_ID, "late-2")
async def test_final_flush_ignores_a_task_from_a_closed_event_loop(tmp_path: Path) -> None:
"""A stale loop-owned task must not abort shutdown on the current loop."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
def leave_cancelled_task_from_closed_loop() -> None:
async def seed() -> None:
await store.arecord(_CHANNEL_ID, "event-1")
store._flush_task = asyncio.create_task(asyncio.sleep(60))
asyncio.run(seed())
await asyncio.to_thread(leave_cancelled_task_from_closed_loop)
await store.aflush()
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-1")
async def test_final_flush_recovers_when_an_in_flight_flush_task_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A background persistence failure must not escape channel shutdown."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
write_started = threading.Event()
release_write = threading.Event()
write_snapshot = store._write_snapshot
write_count = 0
def fail_first_write(payload: dict[str, list[str]]) -> bool:
nonlocal write_count
write_count += 1
if write_count == 1:
write_started.set()
assert release_write.wait(timeout=2)
raise RuntimeError("worker pool is shutting down")
return write_snapshot(payload)
store._write_snapshot = fail_first_write
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01)
await store.arecord(_CHANNEL_ID, "event-1")
assert await asyncio.to_thread(write_started.wait, 2)
final_flush = asyncio.create_task(store.aflush())
await asyncio.sleep(0)
release_write.set()
await final_flush
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-1")
async def test_cancelled_final_flush_does_not_rearm_timer_from_existing_write(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A retained scheduled write must not create new work after stop."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
write_started = threading.Event()
release_write = threading.Event()
snapshots: list[dict[str, list[str]]] = []
write_snapshot = store._write_snapshot
def pause_first_write(payload: dict[str, list[str]]) -> bool:
snapshots.append(payload)
if len(snapshots) == 1:
write_started.set()
assert release_write.wait(timeout=2)
return write_snapshot(payload)
store._write_snapshot = pause_first_write
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.01)
await store.arecord(_CHANNEL_ID, "event-1")
assert await asyncio.to_thread(write_started.wait, 2)
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0.05)
await store.arecord(_CHANNEL_ID, "event-2")
final_flush = asyncio.create_task(store.aflush())
await asyncio.sleep(0)
final_flush.cancel()
with pytest.raises(asyncio.CancelledError):
await final_flush
release_write.set()
await asyncio.sleep(0.1)
assert len(snapshots) == 1
assert await store.aseen(_CHANNEL_ID, "event-2")
await store.aflush()
restarted = BuzzSeenEventStore(path)
assert await restarted.aseen(_CHANNEL_ID, "event-2")
async def test_scheduled_flush_replaces_a_pending_task_from_a_closed_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A stale task must not disable normal coalesced persistence on a new loop."""
path = tmp_path / "buzz-seen-events.json"
store = BuzzSeenEventStore(path)
def pending_task_from_closed_loop() -> asyncio.Task:
loop = asyncio.new_event_loop()
async def remain_pending() -> None:
await asyncio.sleep(60)
task = loop.create_task(remain_pending())
loop.run_until_complete(asyncio.sleep(0))
task._log_destroy_pending = False
loop.close()
return task
stale_task = await asyncio.to_thread(pending_task_from_closed_loop)
store._flush_task = stale_task
monkeypatch.setattr(buzz_seen_events, "FLUSH_DELAY_SECONDS", 0)
try:
await store.arecord(_CHANNEL_ID, "event-1")
deadline = asyncio.get_running_loop().time() + 1
while asyncio.get_running_loop().time() < deadline:
if await BuzzSeenEventStore(path).aseen(_CHANNEL_ID, "event-1"):
break
await asyncio.sleep(0.01)
assert await BuzzSeenEventStore(path).aseen(_CHANNEL_ID, "event-1")
finally:
if store._flush_task is stale_task:
store._flush_task = None

10
backend/tests/test_buzz_seen_events.py
View File

@ -130,7 +130,7 @@ def test_redelivered_event_is_dropped_across_restart(tmp_path):
ch1, captured1 = _started_with_store(store1)
_dispatch(ch1, ev)
assert len(captured1) == 1
store1.flush() # what BuzzChannel.stop() does on a clean shutdown
store1.flush() # synchronous test equivalent of BuzzChannel.stop()'s aflush()
# Simulated restart: a fresh channel instance, fresh store object, same file.
ch2, captured2 = _started_with_store(BuzzSeenEventStore(path))
@ -217,13 +217,13 @@ def test_saves_are_coalesced_under_a_running_loop(tmp_path):
path = tmp_path / "seen.json"
store = BuzzSeenEventStore(path)
saves = []
original_save = store._save
original_write_snapshot = store._write_snapshot
def counting_save():
def counting_write_snapshot(payload):
saves.append(1)
original_save()
return original_write_snapshot(payload)
store._save = counting_save
store._write_snapshot = counting_write_snapshot
async def burst():
for i in range(50):