mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-06 17:38:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): stop channels before draining tasks

Browse Source
This commit is contained in:
chengyongru 2026-07-15 21:44:09 +08:00 committed by chengyongru
parent 681edfa6f3
commit d4e0294734
2 changed files with 9 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
nanobot/cli/commands.py
View File

@ -2119,6 +2119,9 @@ def _run_gateway(
await shutdown_task await shutdown_task
cron.stop() cron.stop()
agent.stop() agent.stop()
# Some SDKs swallow task cancellation while attempting to reconnect.
# Close channel transports before waiting for their runners to exit.
await channels.stop_all()
for task in tasks: for task in tasks:
if not task.done(): if not task.done():
task.cancel() task.cancel()
@ -2127,7 +2130,6 @@ def _run_gateway(
if runtime_tasks is not None and not runtime_tasks_drained: if runtime_tasks is not None and not runtime_tasks_drained:
with suppress(asyncio.CancelledError, Exception): with suppress(asyncio.CancelledError, Exception):
await runtime_tasks await runtime_tasks
await channels.stop_all()
# Flush all cached sessions to durable storage before exit. # Flush all cached sessions to durable storage before exit.
# This prevents data loss on filesystems with write-back # This prevents data loss on filesystems with write-back
# caching (rclone VFS, NFS, FUSE mounts, etc.). # caching (rclone VFS, NFS, FUSE mounts, etc.).

6
tests/cli/test_commands.py
View File

@ -3235,6 +3235,7 @@ def test_gateway_shutdown_event_exits_forever_runtime_tasks(
config = Config() config = Config()
config.gateway.port = 18791 config.gateway.port = 18791
seen: dict[str, object] = {} seen: dict[str, object] = {}
shutdown_order: list[str] = []
class _FakeSessionManager: class _FakeSessionManager:
def flush_all(self) -> int: def flush_all(self) -> int:
@ -3274,9 +3275,11 @@ def test_gateway_shutdown_event_exits_forever_runtime_tasks(
await asyncio.Event().wait() await asyncio.Event().wait()
finally: finally:
seen["channel_task_cleaned_up"] = True seen["channel_task_cleaned_up"] = True
shutdown_order.append("channel_task_cleaned_up")
async def stop_all(self) -> None: async def stop_all(self) -> None:
seen["channels_stopped"] = True seen["channels_stopped"] = True
shutdown_order.append("channels_stopped")
class _FakeCronService: class _FakeCronService:
def __init__(self, _store_path: Path) -> None: def __init__(self, _store_path: Path) -> None:
@ -3343,6 +3346,9 @@ def test_gateway_shutdown_event_exits_forever_runtime_tasks(
assert seen["channels_stopped"] is True assert seen["channels_stopped"] is True
assert seen["cron_stopped"] is True assert seen["cron_stopped"] is True
assert seen["shutdown_handlers_restored"] is True assert seen["shutdown_handlers_restored"] is True
# Channel cleanup must run before cancellation drains the manager task.
# DingTalk's stream SDK can otherwise swallow cancellation and reconnect.
assert shutdown_order == ["channels_stopped", "channel_task_cleaned_up"]
def test_serve_uses_api_config_defaults_and_workspace_override( def test_serve_uses_api_config_defaults_and_workspace_override(