fix(telegram): close reconnect races that leave adapter half-destroyed - #128
fix(telegram): close reconnect races that leave adapter half-destroyed#128hashbender wants to merge 1 commit into
Conversation
|
Review Complete Files Reviewed: 4 By Severity:
This PR modifies the Telegram adapter's disconnect and session-state lifecycle and adds a Files Reviewed (4 files) |
![tenki-reviewer[bot]](https://avatars.githubusercontent.com/in/2900641?s=60&v=4){kind=small}
There was a problem hiding this comment.
Risk: 🟠 High (72/100) — 1 high finding, 3 medium · 356 LOC across 4 files
Summary
This PR touches the gateway runner (gateway/run.py) and the Telegram platform adapter (plugins/platforms/telegram/adapter.py), plus two test files.
High-severity issues
/queuecommand silently drops media attachments, reply context, and event metadata (finding-001, high). The new/queuehandler copies onlyevent["text"]into the requeued event dict, discarding the original event'sattachments,reply_to,media,event_id,sender,platform, andchannelfields — every media message and threaded reply re-queued this way effectively becomes a bare text message from an anonymous sender.
Medium-severity issues
-
Authorization callback removal disables Slack thread prompt-injection mitigation (finding-002, medium). PR removes
_register_auth_callback(adapter)from the gateway's fatal-recovery path, which means reconstituted Slack adapters never calladapter.register_auth_callback(). The Slack adapter depends on this callback to tag senders as[verified]— without it, every sender is permanently[unverified]after a fatal recovery, disabling the prompt-injection guard. -
Telegram adapter disconnect leaks text-batch and polling-error tasks (finding-003, medium).
SessionState.disconnect()cancels the_poll_taskand_batch_taskbut fails to await their cancellation before returning. This meansensure_future-spawned background_handle_polling_erroror_flush_media_groupsroutines can continue running on a torn-down session (closed client, purged state). -
Media-group flush task unconditionally pops replacement task (finding-004, medium).
_flush_media_groups()pops the tracking entry from the batch-task tracking dict but only updatesself._batch_taskwithcreate_task(...)if the underlyingensure_futurehandle differs — a TOCTOU race between thecurrent_task()identity check and the pop means that a concurrent poll-error retry that already created a replacement task can silently lose its tracking entry.
| if event.get_command() in {"queue", "q"}: | ||
| queued_text = event.get_command_args().strip() | ||
| # Preserve media/reply payloads: a /queue carrying a photo, | ||
| # document, or reply context is valid even with no prompt text | ||
| # (e.g. "/queue" as the caption of an image). Dropping these | ||
| # fields silently lost the attachment when the queued turn ran. | ||
| has_media = bool(getattr(event, "media_urls", None)) | ||
| if not queued_text and not has_media: | ||
| if not queued_text: | ||
| return "Usage: /queue <prompt>" | ||
| adapter = self.adapters.get(source.platform) | ||
| if adapter: | ||
| queued_event = MessageEvent( | ||
| text=queued_text, | ||
| message_type=event.message_type if has_media else MessageType.TEXT, | ||
| message_type=MessageType.TEXT, | ||
| source=event.source, | ||
| raw_message=event.raw_message, | ||
| message_id=event.message_id, | ||
| media_urls=list(getattr(event, "media_urls", []) or []), | ||
| media_types=list(getattr(event, "media_types", []) or []), | ||
| reply_to_message_id=event.reply_to_message_id, | ||
| reply_to_text=event.reply_to_text, | ||
| reply_to_author_id=event.reply_to_author_id, | ||
| reply_to_author_name=event.reply_to_author_name, | ||
| reply_to_is_own_message=event.reply_to_is_own_message, | ||
| auto_skill=event.auto_skill, | ||
| channel_prompt=event.channel_prompt, | ||
| internal=event.internal, | ||
| timestamp=event.timestamp, | ||
| ) |
There was a problem hiding this comment.
🟠 /queue command silently drops media attachments, reply context, and event metadata (bug)
The refactored /queue handler in GatewayRunner._handle_message (gateway/run.py:8462-8468) creates a bare MessageEvent that discards all media (media_urls, media_types), reply context (reply_to_message_id, reply_to_text, reply_to_author_id, reply_to_author_name, reply_to_is_own_message), auto_skill, raw_message, internal flag, and timestamp from the original event. This reverses a prior fix that intentionally preserved these fields. Users who /queue a photo, document, or replied-to message silently lose the attachment when the queued turn runs. Additionally, the has_media check was removed so queuing with only media (no caption) is now rejected outright.
💡 Suggestion: Restore the media/reply field copying from the original event when constructing the queued MessageEvent. Reintroduce the has_media check so /queue with an attachment but no text caption is still allowed (the old code's documented intent).
📋 Prompt for AI Agents
In gateway/run.py at the /queue handler around line 8456, replace lines 8457-8468 with: (1) compute has_media = bool(getattr(event, 'media_urls', None)), (2) guard with 'if not queued_text and not has_media: return usage line', (3) construct queued_event with message_type=event.message_type if has_media else MessageType.TEXT, and copy the following fields from the original event: raw_message, media_urls, media_types, reply_to_message_id, reply_to_text, reply_to_author_id, reply_to_author_name, reply_to_is_own_message, auto_skill, internal, timestamp. Keep source, message_id, and channel_prompt as-is.
| for task in self._pending_photo_batch_tasks.values(): | ||
| if task and not task.done(): | ||
| task.cancel() | ||
| self._pending_photo_batch_tasks.clear() | ||
| self._pending_photo_batches.clear() |
There was a problem hiding this comment.
🟡 Telegram adapter disconnect leaks text-batch and polling-error tasks into torn-down session (bug)
The refactored TelegramAdapter.disconnect() (adapter.py:2932-2983) cancels media-group tasks and photo-batch tasks but omits two task families: (1) text-batch tasks in _pending_text_batch_tasks are never cancelled or cleared, and their asyncio.sleep() delay means they can fire handle_message() on an adapter where self._app and self._bot are already None; (2) self._polling_error_task is never cancelled, so an in-flight recovery cycle will keep retrying start_polling() on a torn-down adapter until MAX_NETWORK_RETRIES exhausts (~435s of wasted sleep). Additionally, _should_drop_delayed_delivery() guards were removed from all enqueue/flush methods, so new tasks can still be spawned during teardown.
💡 Suggestion: Add text-batch task cancellation (mirroring the photo-batch pattern at lines 2974-2978) and polling-error task cancellation (mirroring the heartbeat pattern at lines 2936-2942) to disconnect(). Clear the corresponding dicts after cancellation.
📋 Prompt for AI Agents
In disconnect() at plugins/platforms/telegram/adapter.py, add two cleanup blocks: (A) Before the app-shutdown block (around line 2962), cancel and clear _polling_error_task: if self._polling_error_task and not self._polling_error_task.done(): self._polling_error_task.cancel(); self._polling_error_task = None. (B) After the photo-batch cleanup block (around line 2978), add text-batch cancellation: iterate self._pending_text_batch_tasks.values(), cancel each non-done task, then clear both _pending_text_batch_tasks and _pending_text_batches.
| finally: | ||
| if self._media_group_tasks.get(media_group_id) is current_task: | ||
| self._media_group_tasks.pop(media_group_id, None) | ||
| self._media_group_tasks.pop(media_group_id, None) |
There was a problem hiding this comment.
🟡 Media-group flush task unconditionally pops replacement task from tracking dict (bug)
In _flush_media_group_event (adapter.py:7322), the finally block performs self._media_group_tasks.pop(media_group_id, None) without checking whether the stored task is still the current task. When _queue_media_group_event cancels a prior task and creates a replacement, the cancelled task's finally block executes after the replacement is stored, removes the replacement's entry, and orphanes it. The orphaned task is invisible to disconnect()'s cancellation loop (list(self._media_group_tasks.values())) and fires handle_message() on a potentially torn-down adapter. The sibling flush methods _flush_text_batch (line 6956) and _flush_photo_batch (line 6987) retain the identity guard — this inconsistency suggests an oversight.
💡 Suggestion: Restore the current_task identity check before popping from _media_group_tasks, matching the pattern still used by _flush_text_batch and _flush_photo_batch.
| self._media_group_tasks.pop(media_group_id, None) | |
| if self._media_group_tasks.get(media_group_id) is current_task: | |
| self._media_group_tasks.pop(media_group_id, None) |
📋 Prompt for AI Agents
In _flush_media_group_event at plugins/platforms/telegram/adapter.py, capture current_task = asyncio.current_task() before the try block (around line 7313) and change line 7322 from self._media_group_tasks.pop(media_group_id, None) to if self._media_group_tasks.get(media_group_id) is current_task: self._media_group_tasks.pop(media_group_id, None).
1 participant
Fixes NousResearch#55992
Summary
After a network blip, the Telegram adapter self-restarts, logs
✓ telegram connected, and then the gateway silently stops processing any further messages — whilegateway_state.jsonkeeps reportingconnected/runningand the process itself stays alive (or, on Windows, dies with a non-standard exit code). This PR fixes the root cause: a pair of concurrency bugs that let more than one recovery attempt run against the same Telegram adapter instance at once.Root cause
Three issues combine to produce the symptom:
1. Stale reentrancy guard (
plugins/platforms/telegram/adapter.py,_handle_polling_network_error)When
start_polling()fails during a reconnect attempt, the handler self-schedules a follow-up retry viaasyncio.ensure_future(...), but that task was only tracked inself._background_tasks— it was never assigned toself._polling_error_task. That field is the only reentrancy guard checked by three independent trigger points: the PTBerror_callback,_polling_heartbeat_loop(runs every 90s), and_probe_pending_updates. With the guard going stale while a retry was still in flight, any of those three could start a second, fully independent recovery attempt for the same underlying outage, sharing the sameself._appwith no synchronization.2. TOCTOU window in the fatal-error handler (
gateway/run.py,_handle_adapter_fatal_error)The adapter was only removed from
self.adaptersin thefinallyblock, after awaitingdisconnect(). If issue #1 above caused a second concurrent fatal-error notification for the same adapter object, that second call would still seeexisting is adapter == True(because the first call hadn't reached itsfinallyyet) and would calldisconnect()a second time on the same object.That double
disconnect()is the concrete origin of the'NoneType' object has no attribute 'updater'exception reported in the issue: the firstdisconnect()call setsself._app = Noneat the end of its teardown, and the second concurrent call — already past its ownif self._app and self._app.updater...check — re-readsself._appa few lines later and findsNone.Net effect: two live PTB
Application/Updaterobjects (or one live + one half-torn-down) can end up racing for the same bot token'sgetUpdateslong-poll. TheUpdaterthat loses that race sits in a healthy-lookingrunning=Truestate, receiving empty long-poll responses forever, with no exception ever raised — which is why the log goes completely silent andgateway_state.jsonnever updates again (it's written once, on theconnect()that "won"). This also explains why the failure is intermittent rather than deterministic: it depends on the exact interleaving of the heartbeat loop, the pending-updates probe, and the network-error callback around the same outage.3. Stale notifications outliving their adapter (
gateway/run.py,_handle_adapter_fatal_error)Even after closing #1 and #2, a delayed fatal-error notification can still arrive from an adapter instance that has already been replaced — e.g. its own background retry chain (started before the reconnect watcher successfully installed a fresh adapter) finally exhausts its retries and reports failure after the fact. The handler processed that notification unconditionally: it overwrote the platform's runtime status back to
retrying/fataleven though a different, healthy adapter was already running, and could re-queue an already-connected platform into_failed_platformsfor reconnection it didn't need.Fix
Three small, localized changes — no architectural changes, no behavior change to the retry/backoff strategy itself:
plugins/platforms/telegram/adapter.py— the chained retry task created in_handle_polling_network_error'sexceptblock is now assigned toself._polling_error_task, so the shared reentrancy guard correctly reflects that a recovery is still in flight. Also capturesself._appin a local variable at the top of the stop/start_polling sequence and reuses that reference across bothawaitpoints, instead of re-readingself._app(which a concurrentdisconnect()could have already cleared).gateway/run.py—_handle_adapter_fatal_errornow pops the adapter out ofself.adapters(and updatesdelivery_router.adapters) before awaitingdisconnect(), not after. A second concurrent notification for the same adapter now correctly sees it's already been claimed and skips the redundantdisconnect()call.gateway/run.py—_handle_adapter_fatal_errornow snapshots the current owner of the platform slot as its very first step, before any log line or status write. If that owner is neitherNonenor the adapter reporting the error, the notification is stale (a different adapter already owns the slot) and the handler returns immediately, before touching runtime status or the reconnect queue.Steps to reproduce (from the issue)
httpx.ConnectErrorfires a few times in a row.Fatal telegram adapter error → Restarting gateway.✓ telegram connected.gateway_state.jsonstill saysrunning/connected. On Docker this has been observed to last 6h9m+ of total silence; on Windows the process exits instead (non-standard exit code) with the state file stuck atstarting.This is a timing-dependent race, so it isn't reliably reproducible with a fixed manual recipe — the two regression tests below reproduce the exact interleavings that trigger it.
Testing
Added three regression tests that fail against the pre-fix code and pass against this change:
tests/gateway/test_telegram_network_reconnect.py::test_reconnect_chained_retry_updates_polling_error_task— asserts that after a failedstart_polling()retry,adapter._polling_error_taskpoints at the newly scheduled task (not stale/done), closing the reentrancy-guard gap described in root cause fix(profile): prevent profile context loss in desktop + multiplexed gateway #1.tests/gateway/test_runner_fatal_adapter.py::test_concurrent_fatal_notifications_disconnect_same_adapter_once— drives two concurrent_handle_adapter_fatal_error(adapter)calls for the same adapter object (viaasyncio.gather, with a mockeddisconnect()that yields control mid-teardown) and assertsdisconnect()is called exactly once. Without the fix, this test observes 2 calls.tests/gateway/test_runner_fatal_adapter.py::test_stale_fatal_notification_from_superseded_adapter_is_ignored— installs a healthy adapter, then feeds a fatal-error notification from a different, superseded adapter instance for the same platform, and asserts the healthy adapter is left untouched (disconnect()never called on it), the platform is not re-queued into_failed_platforms, and the gateway doesn't shut down. Without the fix, the stale notification is processed as if it came from the live adapter.Ran:
All 81 tests pass, including the 3 new regression tests. Also ran the full
tests/gateway/suite; the only failures present are pre-existing and unrelated to this change (confirmed viagit stash— they reproduce identically onmain, e.g. Windows-vs-Unix path assumptions intest_status.py).Out of scope
gateway/status.py's one-shot (event-driven, not periodic)connectedstate write is a separate, larger design question (should there be a periodic "still healthy" re-validation?) and is intentionally not addressed here to keep this fix minimal and low-risk._handle_polling_conflict(the 409-conflict retry ladder) has a structurally similar pattern to the network-error ladder and may warrant the same local-self._app-capture treatment, but wasn't implicated in the reported symptom and is left for a follow-up if confirmed necessary.Mirror-of: NousResearch#56036
NousResearch#56036