mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-10 14:08:52 +00:00
fix(channels): bound Discord outbound cross-loop awaits and restart dead clients, fixes #5226 (#5227)
* fix(channels): bound Discord outbound cross-loop awaits and restart dead clients, fixes #5226 * docs: reduce inherited agent guidance size * fix(channels): stop half-started channels before discarding them _start_channel now tears the instance down (stop + untrack) whenever start() raises or the channel never reaches is_running, so an outbound listener subscribed before the transport was confirmed cannot outlive its channel. Addresses the review on #5227. * fix(channels): retain half-started channels until failed-start cleanup completes Ownership in _stop_and_discard_channel now mirrors ChannelService.stop(): the instance is dropped only after its stop() completes. A cancellation arriving mid-cleanup (or a stop() that raises) leaves it tracked, so a retried readiness attempt stops it again before replacing it and service shutdown can still reach it — untracking first orphaned resources nobody could clean up. Addresses the round-3 review on #5227. * fix(channels): defer replacement when a retained channel fails to stop The pre-retry stop in ensure_channel_ready popped unconditionally, so a retained instance whose second stop() raised was untracked with its outbound listener still subscribed — the same orphan one hop later. restart_channel (del after failed stop) and remove_channel (pop before stop) had the same shape. All three now route through _stop_and_discard_channel and decline the operation for that round when the instance is retained, so _start_channel can never overwrite a still-listening channel. Addresses the review on #5227. * fix(channels): enforce the retention guarantee inside the readiness attempt loop A failed attempt whose cleanup retained the instance used to let the next attempt (attempts=2 is the production default) construct a fresh instance and overwrite the retained one via _start_channel's unconditional assignment — orphaning the first instance's subscribed listener one hop earlier than the cross-round guard covers. The guard now lives at the mechanism: _start_channel refuses to install while the name is still tracked, and ensure_channel_ready ends the loop on retention. The shared discard helper's log message is path-neutral. Addresses the review on #5227. * fix(channels): make the retained-instance guard message path-neutral The guard can fire for any still-tracked instance, not only failed cleanup, so the message must not assume the cause. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
hataa
GitHub
parent
0d4925305a
commit
37b03a3811
File diff suppressed because one or more lines are too long
@ -19,6 +19,19 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
_DISCORD_MAX_MESSAGE_LEN = 2000
|
||||
|
||||
# Bound for outbound work scheduled onto the Discord client's event loop.
|
||||
# Discord API calls normally return in well under a second; anything still
|
||||
# pending after this is a dead or wedged client, not a slow response.
|
||||
DISCORD_OUTBOUND_TIMEOUT_SECONDS = 30.0
|
||||
|
||||
# File uploads carry an unbounded-size payload with no per-channel size cap
|
||||
# (unlike Feishu/Telegram's send_file limits), so they get their own, larger
|
||||
# bound: a 50 MB artifact over a ~5 Mbps uplink takes ~80 s to push, and a
|
||||
# large 429 retry-after inside discord.py can extend that further. Cancelling
|
||||
# a healthy upload mid-transfer would report it as failed, so the bound here
|
||||
# only exists to convert a wedged client into a logged failure.
|
||||
DISCORD_UPLOAD_TIMEOUT_SECONDS = 120.0
|
||||
|
||||
|
||||
class DiscordChannel(Channel):
|
||||
"""Discord bot channel.
|
||||
@ -239,10 +252,46 @@ class DiscordChannel(Channel):
|
||||
self._discord_module = None
|
||||
logger.info("Discord channel stopped")
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
"""Running means the client thread is still alive, not just started.
|
||||
|
||||
``_run_client`` exits when discord.py gives up for good (invalidated
|
||||
token, unrecoverable close) while ``_running`` stays True, so the base
|
||||
flag alone would keep reporting a healthy channel forever. Mirrors
|
||||
``FeishuChannel.is_running`` so ``ChannelService.ensure_channel_ready``
|
||||
can restart the channel after its client thread dies.
|
||||
"""
|
||||
if not self._running:
|
||||
return False
|
||||
return self._thread is not None and self._thread.is_alive()
|
||||
|
||||
async def _run_on_discord_loop(self, coro, *, timeout: float = DISCORD_OUTBOUND_TIMEOUT_SECONDS):
|
||||
"""Schedule *coro* on the Discord loop and await it with a bound.
|
||||
|
||||
The Discord client runs on a dedicated thread whose loop is stopped but
|
||||
not closed when the client dies, so ``call_soon_threadsafe`` keeps
|
||||
queueing callbacks that never run and an unbounded ``wrap_future``
|
||||
await would hang a ChannelManager worker forever. ``stop()`` already
|
||||
bounds its identical cross-loop awaits with ``wait_for``; this extends
|
||||
that pattern to the outbound path. Failing fast when the loop is
|
||||
missing or not running turns a dead client into a logged send failure
|
||||
instead of a wedged worker.
|
||||
"""
|
||||
loop = self._discord_loop
|
||||
if loop is None or not loop.is_running():
|
||||
coro.close()
|
||||
raise RuntimeError("Discord client event loop is not running")
|
||||
future = asyncio.run_coroutine_threadsafe(coro, loop)
|
||||
try:
|
||||
return await asyncio.wait_for(asyncio.wrap_future(future), timeout=timeout)
|
||||
except TimeoutError:
|
||||
future.cancel()
|
||||
raise
|
||||
|
||||
async def send(self, msg: OutboundMessage) -> None:
|
||||
# Stop typing indicator once we're sending the response
|
||||
stop_future = asyncio.run_coroutine_threadsafe(self._stop_typing(msg.chat_id, msg.thread_ts), self._discord_loop)
|
||||
await asyncio.wrap_future(stop_future)
|
||||
await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts))
|
||||
|
||||
target = await self._resolve_target(msg)
|
||||
if target is None:
|
||||
@ -251,12 +300,10 @@ class DiscordChannel(Channel):
|
||||
|
||||
text = msg.text or ""
|
||||
for chunk in self._split_text(text):
|
||||
send_future = asyncio.run_coroutine_threadsafe(target.send(chunk), self._discord_loop)
|
||||
await asyncio.wrap_future(send_future)
|
||||
await self._run_on_discord_loop(target.send(chunk))
|
||||
|
||||
async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
|
||||
stop_future = asyncio.run_coroutine_threadsafe(self._stop_typing(msg.chat_id, msg.thread_ts), self._discord_loop)
|
||||
await asyncio.wrap_future(stop_future)
|
||||
await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts))
|
||||
|
||||
target = await self._resolve_target(msg)
|
||||
if target is None:
|
||||
@ -274,8 +321,7 @@ class DiscordChannel(Channel):
|
||||
# success and failure paths.
|
||||
data = await asyncio.to_thread(self._read_attachment_bytes, str(attachment.actual_path))
|
||||
file = self._discord_module.File(io.BytesIO(data), filename=attachment.filename)
|
||||
send_future = asyncio.run_coroutine_threadsafe(target.send(file=file), self._discord_loop)
|
||||
await asyncio.wrap_future(send_future)
|
||||
await self._run_on_discord_loop(target.send(file=file), timeout=DISCORD_UPLOAD_TIMEOUT_SECONDS)
|
||||
logger.info("[Discord] file uploaded: %s", attachment.filename)
|
||||
return True
|
||||
except Exception:
|
||||
@ -773,9 +819,8 @@ class DiscordChannel(Channel):
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
get_future = asyncio.run_coroutine_threadsafe(self._fetch_channel(target_id), self._discord_loop)
|
||||
try:
|
||||
return await asyncio.wrap_future(get_future)
|
||||
return await self._run_on_discord_loop(self._fetch_channel(target_id))
|
||||
except Exception:
|
||||
logger.exception("[Discord] failed to resolve target id=%s", raw_id)
|
||||
return None
|
||||
|
||||
@ -255,11 +255,15 @@ class ChannelService:
|
||||
return True
|
||||
|
||||
if channel is not None:
|
||||
try:
|
||||
await channel.stop()
|
||||
except Exception:
|
||||
logger.exception("Error stopping non-running channel before readiness retry")
|
||||
self._channels.pop(name, None)
|
||||
# Ownership-preserving cleanup: the instance is retained when
|
||||
# its stop() fails or is cancelled, and this round must NOT
|
||||
# start a replacement over it — _start_channel would overwrite
|
||||
# the tracked entry and orphan the still-subscribed listener
|
||||
# one hop later (the gap this closes from the review on 5227).
|
||||
await self._stop_and_discard_channel(name, channel)
|
||||
if self._channels.get(name) is channel:
|
||||
logger.warning("Readiness retry deferred: previous %s channel failed to stop and remains tracked", name)
|
||||
return False
|
||||
|
||||
max_attempts = max(1, attempts)
|
||||
for attempt in range(max_attempts):
|
||||
@ -267,6 +271,13 @@ class ChannelService:
|
||||
logger.info("Retrying channel startup after readiness check")
|
||||
if await self._start_channel(name, channel_config):
|
||||
return True
|
||||
# A failed attempt whose cleanup retained the instance ends
|
||||
# the loop for this round: the next attempt would be refused
|
||||
# by _start_channel's retained-instance guard anyway, and the
|
||||
# still-tracked channel must not be replaced one hop later.
|
||||
if self._channels.get(name) is not None:
|
||||
logger.warning("Readiness retries deferred: %s channel failed to clean up after a failed start and remains tracked", name)
|
||||
return False
|
||||
return False
|
||||
|
||||
async def stop(self) -> None:
|
||||
@ -329,11 +340,14 @@ class ChannelService:
|
||||
async def restart_channel(self, name: str, *, reload_config: bool = True) -> bool:
|
||||
"""Restart a specific channel. Returns True if successful."""
|
||||
if name in self._channels:
|
||||
try:
|
||||
await self._channels[name].stop()
|
||||
except Exception:
|
||||
logger.exception("Error stopping channel for restart")
|
||||
del self._channels[name]
|
||||
channel = self._channels[name]
|
||||
# Same ownership rule as readiness retries: retain an instance
|
||||
# whose stop() fails, and decline the restart rather than
|
||||
# overwriting a still-tracked (still-listening) channel.
|
||||
await self._stop_and_discard_channel(name, channel)
|
||||
if self._channels.get(name) is channel:
|
||||
logger.warning("Restart deferred: %s channel failed to stop and remains tracked", name)
|
||||
return False
|
||||
|
||||
if reload_config:
|
||||
# Reading config.yaml and the runtime store is disk IO; keep it
|
||||
@ -364,16 +378,58 @@ class ChannelService:
|
||||
async def remove_channel(self, name: str) -> bool:
|
||||
"""Remove runtime config for a channel and stop it if currently running."""
|
||||
self._config.pop(name, None)
|
||||
channel = self._channels.pop(name, None)
|
||||
channel = self._channels.get(name)
|
||||
if channel is None:
|
||||
return True
|
||||
# Stop-then-drop with the shared ownership rule: a channel whose
|
||||
# stop() fails stays tracked (and returns False) instead of being
|
||||
# popped first and leaking its subscribed listener on failure.
|
||||
await self._stop_and_discard_channel(name, channel)
|
||||
if self._channels.get(name) is channel:
|
||||
logger.warning("Removal incomplete: %s channel failed to stop and remains tracked", name)
|
||||
return False
|
||||
logger.info("Channel stopped and removed")
|
||||
return True
|
||||
|
||||
async def _stop_and_discard_channel(self, name: str, channel: Channel) -> None:
|
||||
"""Stop a channel and drop it only once its ``stop()`` has completed.
|
||||
|
||||
This is the single ownership-preserving cleanup every discard path
|
||||
routes through (failed startup, readiness retry, restart, removal).
|
||||
``start()`` subscribes the outbound listener before the transport is
|
||||
up, so an instance that never reached ``is_running`` — or a running
|
||||
one being torn down — must be ``stop()``-ed before it is discarded:
|
||||
otherwise the bus keeps a strong reference to the dead listener and
|
||||
every future outbound for this channel name fans out to it, while
|
||||
repeated attempts accumulate more stale listeners the service can no
|
||||
longer clean up (the instances are untracked by then). Discord's
|
||||
fail-fast ``is_running`` makes this reachable for a client thread that
|
||||
dies immediately (invalid token); the same hygiene applies to any
|
||||
channel that subscribes before its transport is confirmed.
|
||||
|
||||
Ownership mirrors ``ChannelService.stop()``: the instance is dropped
|
||||
only after its ``stop()`` actually completes. A cancellation arriving
|
||||
mid-cleanup (or a ``stop()`` that raises) leaves it tracked, so a
|
||||
retried readiness attempt stops it again before replacing it and
|
||||
service shutdown can still reach it — untracking first would orphan
|
||||
resources nobody can clean up anymore. Callers check for retention
|
||||
(``self._channels.get(name) is channel``) and defer starting or
|
||||
removing a replacement for that round, so startup cannot silently
|
||||
overwrite a still-listening retained instance; ``ensure_channel_ready``
|
||||
additionally serializes on the per-channel readiness lock.
|
||||
"""
|
||||
try:
|
||||
await channel.stop()
|
||||
logger.info("Channel stopped and removed")
|
||||
return True
|
||||
except asyncio.CancelledError:
|
||||
# Keep this transport owned by the service: the Gateway deadline
|
||||
# interrupted cleanup, so detaching it here would hide resources
|
||||
# that may still be in use (mirrors ChannelService.stop()).
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("Error stopping channel for removal")
|
||||
return False
|
||||
logger.exception("Error stopping channel %s during discard", name)
|
||||
return
|
||||
if self._channels.get(name) is channel:
|
||||
self._channels.pop(name, None)
|
||||
|
||||
async def _start_channel(self, name: str, config: dict[str, Any]) -> bool:
|
||||
"""Instantiate and start a single channel."""
|
||||
@ -382,6 +438,17 @@ class ChannelService:
|
||||
logger.warning("Unknown channel type")
|
||||
return False
|
||||
|
||||
# Never install a fresh instance over a retained one: a channel whose
|
||||
# failed cleanup kept it tracked still holds a subscribed outbound
|
||||
# listener, and overwriting the entry here is the one remaining way to
|
||||
# orphan it (nothing would be able to stop it afterwards). Callers
|
||||
# decline the operation when they see the name still tracked; this
|
||||
# guard makes the invariant hold at the mechanism itself.
|
||||
retained = self._channels.get(name)
|
||||
if retained is not None:
|
||||
logger.warning("Refusing to start %s: another channel instance is still tracked under this name (previous cleanup incomplete, or the instance is still running)", name)
|
||||
return False
|
||||
|
||||
try:
|
||||
from deerflow.reflection import resolve_class
|
||||
|
||||
@ -390,6 +457,7 @@ class ChannelService:
|
||||
logger.exception("Failed to import channel class")
|
||||
return False
|
||||
|
||||
channel: Channel | None = None
|
||||
try:
|
||||
config = dict(config)
|
||||
config["channel_store"] = self.store
|
||||
@ -407,14 +475,17 @@ class ChannelService:
|
||||
self._channels[name] = channel
|
||||
await channel.start()
|
||||
if not channel.is_running:
|
||||
self._channels.pop(name, None)
|
||||
logger.error("Channel did not enter a running state after start()")
|
||||
await self._stop_and_discard_channel(name, channel)
|
||||
return False
|
||||
logger.info("Channel started")
|
||||
return True
|
||||
except Exception:
|
||||
self._channels.pop(name, None)
|
||||
logger.exception("Failed to start channel")
|
||||
if channel is not None:
|
||||
await self._stop_and_discard_channel(name, channel)
|
||||
else:
|
||||
self._channels.pop(name, None)
|
||||
return False
|
||||
|
||||
def get_status(self) -> dict[str, Any]:
|
||||
|
||||
@ -7519,6 +7519,367 @@ class TestChannelService:
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_readiness_retry_defers_when_old_instance_fails_to_stop(self):
|
||||
"""A retained channel whose stop() fails must not be replaced this round.
|
||||
|
||||
The pre-retry cleanup retains the instance when stop() raises; the
|
||||
readiness attempt then declines instead of letting _start_channel
|
||||
overwrite the still-tracked, still-listening channel — the one-hop-
|
||||
later orphan shape from the review.
|
||||
"""
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
class FailingStopChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
|
||||
async def start(self):
|
||||
self._running = True
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self._running = False
|
||||
raise RuntimeError("stop boom")
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
async def _on_outbound(self, msg):
|
||||
raise AssertionError("a listener slated for cleanup must never receive outbounds")
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
stale = FailingStopChannel(bus=service.bus, config={})
|
||||
service._channels["telegram"] = stale
|
||||
|
||||
ready = await service.ensure_channel_ready("telegram", attempts=2)
|
||||
|
||||
assert ready is False
|
||||
assert service._channels.get("telegram") is stale # retained, not overwritten
|
||||
assert stale.stop_calls == 1 # each retry stops the retained instance again
|
||||
# The listener stays subscribed precisely because the instance is
|
||||
# retained: only a completed stop may unsubscribe it.
|
||||
assert any(getattr(listener, "__self__", None) is stale for listener in service.bus._outbound_listeners)
|
||||
|
||||
# Shutdown reports the retained channel's failing stop (ExceptionGroup)
|
||||
# instead of silently orphaning it — expected here by construction.
|
||||
with pytest.raises(Exception):
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_readiness_attempts_do_not_replace_retained_instance(self, monkeypatch):
|
||||
"""Within one ensure_channel_ready loop, a failed attempt whose cleanup
|
||||
retains the instance must end the loop instead of being overwritten.
|
||||
|
||||
The reviewer repro on #5227: with attempts=2 (the production default),
|
||||
a channel whose start() never reaches is_running AND whose stop()
|
||||
raises used to let attempt 2 construct a fresh instance and overwrite
|
||||
the retained one — returning True while the first instance's outbound
|
||||
listener stayed subscribed forever.
|
||||
"""
|
||||
import deerflow.reflection as reflection_module
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
created = []
|
||||
|
||||
class FailFastAndUncleanChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
created.append(self)
|
||||
|
||||
async def start(self):
|
||||
# Subscribe the listener, then report a client thread that
|
||||
# died before start() returned (the Discord invalid-token
|
||||
# shape).
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
self._running = True
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return False
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self._running = False
|
||||
raise RuntimeError("stop boom")
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
async def _on_outbound(self, msg):
|
||||
raise AssertionError("a listener slated for cleanup must never receive outbounds")
|
||||
|
||||
monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: FailFastAndUncleanChannel)
|
||||
|
||||
ready = await service.ensure_channel_ready("telegram", attempts=2)
|
||||
|
||||
assert ready is False
|
||||
assert len(created) == 1 # attempt 2 never constructed a replacement
|
||||
retained = created[0]
|
||||
assert service._channels.get("telegram") is retained
|
||||
assert retained.stop_calls == 1
|
||||
assert any(getattr(listener, "__self__", None) is retained for listener in service.bus._outbound_listeners)
|
||||
|
||||
with pytest.raises(Exception):
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_restart_and_remove_retain_channel_when_stop_fails(self):
|
||||
"""restart_channel and remove_channel defer instead of orphaning a failed stop."""
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
class FailingStopChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
|
||||
async def start(self):
|
||||
self._running = True
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self._running = False
|
||||
raise RuntimeError("stop boom")
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
async def _on_outbound(self, msg):
|
||||
raise AssertionError("a listener slated for cleanup must never receive outbounds")
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
original = FailingStopChannel(bus=service.bus, config={})
|
||||
service._channels["telegram"] = original
|
||||
|
||||
assert await service.restart_channel("telegram") is False
|
||||
assert service._channels.get("telegram") is original
|
||||
assert original.stop_calls == 1
|
||||
assert any(getattr(listener, "__self__", None) is original for listener in service.bus._outbound_listeners)
|
||||
|
||||
assert await service.remove_channel("telegram") is False
|
||||
assert service._channels.get("telegram") is original
|
||||
assert original.stop_calls == 2
|
||||
assert any(getattr(listener, "__self__", None) is original for listener in service.bus._outbound_listeners)
|
||||
|
||||
with pytest.raises(Exception):
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_failed_channel_startup_is_transactional(self, monkeypatch):
|
||||
"""A channel that never reaches is_running must be stopped before discard.
|
||||
|
||||
start() subscribes the outbound listener before the transport is
|
||||
confirmed up, so a client thread that dies immediately (the Discord
|
||||
invalid-token shape) must not leave a stale listener behind on the
|
||||
bus — repeated readiness attempts would otherwise accumulate dead
|
||||
listeners the service can no longer clean up.
|
||||
"""
|
||||
import deerflow.reflection as reflection_module
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
created = []
|
||||
|
||||
class DeadOnArrivalChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
created.append(self)
|
||||
|
||||
async def start(self):
|
||||
# Every adapter subscribes outbound before its transport is up.
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
self._running = True
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return False
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self._running = False
|
||||
self.bus.unsubscribe_outbound(self._on_outbound)
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
async def _on_outbound(self, msg):
|
||||
raise AssertionError("a dead listener must never receive outbounds")
|
||||
|
||||
monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: DeadOnArrivalChannel)
|
||||
|
||||
ready = await service.ensure_channel_ready("telegram", attempts=3)
|
||||
|
||||
assert ready is False
|
||||
assert len(created) == 3
|
||||
assert all(channel.stop_calls == 1 for channel in created)
|
||||
assert service.bus._outbound_listeners == []
|
||||
assert "telegram" not in service._channels
|
||||
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_cancelled_failed_start_cleanup_retains_channel_until_cleaned(self, monkeypatch):
|
||||
"""Cancellation during failed-start cleanup must not orphan the channel.
|
||||
|
||||
``_stop_and_discard_channel`` keeps the half-started instance tracked
|
||||
until its ``stop()`` completes: cancelling the readiness request
|
||||
mid-cleanup (review repro) leaves the instance reachable, so a later
|
||||
readiness retry stops it again before replacing it and service
|
||||
shutdown can still clean it up. Untracking first would leave the
|
||||
subscribed outbound listener owned by nobody — stop count stuck at
|
||||
one and the listener still registered after ``service.stop()``.
|
||||
"""
|
||||
import deerflow.reflection as reflection_module
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
release = asyncio.Event()
|
||||
created = []
|
||||
|
||||
class SuspendableStopChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
self.stop_entered = asyncio.Event()
|
||||
created.append(self)
|
||||
|
||||
async def start(self):
|
||||
# Every adapter subscribes outbound before its transport is up.
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
self._running = True
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return False
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self.stop_entered.set()
|
||||
if not release.is_set():
|
||||
await release.wait()
|
||||
self._running = False
|
||||
self.bus.unsubscribe_outbound(self._on_outbound)
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: SuspendableStopChannel)
|
||||
|
||||
# Phase 1: readiness cancelled while the failed-start cleanup is
|
||||
# suspended inside stop().
|
||||
readiness = asyncio.ensure_future(service.ensure_channel_ready("telegram", attempts=1))
|
||||
while not created:
|
||||
await asyncio.sleep(0.01)
|
||||
await created[0].stop_entered.wait()
|
||||
readiness.cancel()
|
||||
try:
|
||||
await readiness
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
retained = service._channels.get("telegram")
|
||||
assert retained is created[0] # retained for retry/shutdown, not orphaned
|
||||
assert retained.stop_calls == 1 # first cleanup was interrupted
|
||||
assert service.bus._outbound_listeners # listener still registered
|
||||
|
||||
# Phase 2: a later readiness retry stops the retained instance
|
||||
# before replacing it — never swaps an uncleaned channel out.
|
||||
release.set()
|
||||
ready = await service.ensure_channel_ready("telegram", attempts=1)
|
||||
assert ready is False
|
||||
assert len(created) == 2
|
||||
assert created[0].stop_calls == 2 # cleanup completed on retry
|
||||
assert created[1].stop_calls == 1 # replacement got its own teardown
|
||||
assert service.bus._outbound_listeners == []
|
||||
assert "telegram" not in service._channels
|
||||
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_start_channel_exception_stops_and_discards(self, monkeypatch):
|
||||
"""A start() that raises mid-way must also stop the half-started channel."""
|
||||
import deerflow.reflection as reflection_module
|
||||
from app.channels.base import Channel
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
async def go():
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}})
|
||||
await service.manager.start()
|
||||
service._running = True
|
||||
|
||||
created = []
|
||||
|
||||
class StartRaisesChannel(Channel):
|
||||
def __init__(self, bus, config):
|
||||
super().__init__(name="telegram", bus=bus, config=config)
|
||||
self.stop_calls = 0
|
||||
created.append(self)
|
||||
|
||||
async def start(self):
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
self._running = True
|
||||
raise RuntimeError("simulated invalid token")
|
||||
|
||||
async def stop(self):
|
||||
self.stop_calls += 1
|
||||
self._running = False
|
||||
self.bus.unsubscribe_outbound(self._on_outbound)
|
||||
|
||||
async def send(self, msg):
|
||||
raise NotImplementedError
|
||||
|
||||
async def _on_outbound(self, msg):
|
||||
raise AssertionError("a discarded listener must never receive outbounds")
|
||||
|
||||
monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: StartRaisesChannel)
|
||||
|
||||
ready = await service.ensure_channel_ready("telegram", attempts=2)
|
||||
|
||||
assert ready is False
|
||||
assert len(created) == 2
|
||||
assert all(channel.stop_calls == 1 for channel in created)
|
||||
assert service.bus._outbound_listeners == []
|
||||
assert "telegram" not in service._channels
|
||||
|
||||
await service.stop()
|
||||
|
||||
_run(go())
|
||||
|
||||
def test_session_config_is_forwarded_to_manager(self):
|
||||
from app.channels.service import ChannelService
|
||||
|
||||
|
||||
@ -433,3 +433,138 @@ async def test_stop_wiring_drains_ack_tasks_across_loops() -> None:
|
||||
assert not channel._ack_reaction_tasks
|
||||
finally:
|
||||
_stop_bg_loop(bg_loop, bg_thread)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dead-client fail-fast and is_running thread-aliveness
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_fails_fast_when_discord_loop_is_not_running() -> None:
|
||||
"""A stopped (not closed) Discord loop must fail the send, not hang the worker.
|
||||
|
||||
``_run_client`` leaves the loop stopped-but-unclosed when the client dies,
|
||||
which is exactly the state where ``call_soon_threadsafe`` queues callbacks
|
||||
that never run — the permanent-hang case this guards against.
|
||||
"""
|
||||
channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
|
||||
channel._discord_loop = asyncio.new_event_loop() # created, never run
|
||||
channel._running = True
|
||||
msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello")
|
||||
|
||||
try:
|
||||
with pytest.raises(RuntimeError, match="event loop is not running"):
|
||||
await channel.send(msg)
|
||||
finally:
|
||||
channel._discord_loop.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_outbound_loop_call_times_out_when_never_completes() -> None:
|
||||
"""Even on a live loop, an outbound call that never resolves is bounded by the timeout."""
|
||||
bg_loop, bg_thread = _start_bg_loop()
|
||||
try:
|
||||
channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
|
||||
channel._discord_loop = bg_loop
|
||||
|
||||
async def _never_completes() -> None:
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
with pytest.raises(TimeoutError):
|
||||
await channel._run_on_discord_loop(_never_completes(), timeout=0.1)
|
||||
|
||||
# The cancelled call leaves its task parked on the bg loop (concurrent
|
||||
# cancellation cannot reach a running run_coroutine_threadsafe task);
|
||||
# clean it up so stopping the loop has nothing pending.
|
||||
async def _cancel_leftovers() -> None:
|
||||
current = asyncio.current_task()
|
||||
for task in asyncio.all_tasks():
|
||||
if task is not current:
|
||||
task.cancel()
|
||||
|
||||
cleanup = asyncio.run_coroutine_threadsafe(_cancel_leftovers(), bg_loop)
|
||||
cleanup.result(timeout=5)
|
||||
finally:
|
||||
_stop_bg_loop(bg_loop, bg_thread)
|
||||
|
||||
|
||||
def test_is_running_tracks_thread_aliveness() -> None:
|
||||
"""``is_running`` reflects the client thread, so readiness can restart a dead channel."""
|
||||
channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
|
||||
assert channel.is_running is False # never started
|
||||
|
||||
channel._running = True
|
||||
assert channel.is_running is False # started but the thread object is gone
|
||||
|
||||
finished = threading.Thread(target=lambda: None)
|
||||
finished.start()
|
||||
finished.join()
|
||||
channel._thread = finished
|
||||
assert channel.is_running is False # dead client thread (fatal exit)
|
||||
|
||||
channel._thread = threading.current_thread()
|
||||
assert channel.is_running is True # live thread -> healthy
|
||||
|
||||
|
||||
def _close_unawaited_mock_coroutines(run_mock) -> None:
|
||||
"""Close the coroutines handed to an AsyncMock that never awaited them.
|
||||
|
||||
``_run_on_discord_loop`` receives already-created coroutine objects; an
|
||||
AsyncMock stand-in records them without awaiting, so they must be closed
|
||||
explicitly or GC warns about never-awaited coroutines.
|
||||
"""
|
||||
for call in run_mock.await_args_list:
|
||||
call.args[0].close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_file_upload_call_uses_the_dedicated_upload_timeout(tmp_path) -> None:
|
||||
"""Pin the upload call site to DISCORD_UPLOAD_TIMEOUT_SECONDS.
|
||||
|
||||
Regressing ``send_file``'s upload call to the 30 s control-plane default
|
||||
(the exact bug round 1 of this review caught) keeps every helper-level
|
||||
test green; only the call site's ``timeout=`` kwarg can guard it.
|
||||
"""
|
||||
from app.channels.discord import DISCORD_UPLOAD_TIMEOUT_SECONDS
|
||||
|
||||
channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
|
||||
channel._discord_module = SimpleNamespace(File=lambda fp, filename=None: fp)
|
||||
run_mock = AsyncMock(return_value=None)
|
||||
channel._run_on_discord_loop = run_mock # type: ignore[method-assign]
|
||||
channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro))
|
||||
|
||||
path = tmp_path / "upload.txt"
|
||||
path.write_bytes(b"hello")
|
||||
att = ResolvedAttachment("/mnt/user-data/outputs/upload.txt", path, "upload.txt", "text/plain", 5, False)
|
||||
msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="t")
|
||||
|
||||
try:
|
||||
assert await channel.send_file(msg, att) is True
|
||||
finally:
|
||||
_close_unawaited_mock_coroutines(run_mock)
|
||||
|
||||
assert len(run_mock.await_args_list) == 2 # stop_typing, then the upload
|
||||
stop_call, upload_call = run_mock.await_args_list
|
||||
assert "timeout" not in stop_call.kwargs # control-plane call keeps the 30 s default
|
||||
assert upload_call.kwargs.get("timeout") == DISCORD_UPLOAD_TIMEOUT_SECONDS
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_control_calls_keep_the_default_outbound_bound() -> None:
|
||||
"""``send``'s typing-stop and message sends rely on the 30 s default, not an override."""
|
||||
channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
|
||||
run_mock = AsyncMock(return_value=None)
|
||||
channel._run_on_discord_loop = run_mock # type: ignore[method-assign]
|
||||
channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro))
|
||||
|
||||
msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello")
|
||||
|
||||
try:
|
||||
await channel.send(msg)
|
||||
finally:
|
||||
_close_unawaited_mock_coroutines(run_mock)
|
||||
|
||||
assert len(run_mock.await_args_list) == 2 # stop_typing + one text chunk
|
||||
for call in run_mock.await_args_list:
|
||||
assert "timeout" not in call.kwargs
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user