fix(telegram): recover when polling updater stops while process stays alive (#55769) - #59
Conversation
|
Review Complete Files Reviewed: 1 By Severity:
PR #59 refactors the Telegram adapter's disconnect/cancellation logic, introducing three concurrency bugs (a race condition in media-group flush task management, text-batch tasks that continue dispatching after disconnect, and a removed guard that allowed drop-on-disconnect behavior) plus two test-breaking regressions. Files Reviewed (1 files) |
![tenki-reviewer[bot]](https://avatars.githubusercontent.com/in/2900641?s=60&v=4){kind=small}
There was a problem hiding this comment.
Risk: 🔴 Critical (88/100) — 2 high findings · 116 LOC across 1 file
Summary
PR #59 refactors plugins/platforms/telegram/adapter.py, inlining the _cancel_pending_delivery_tasks() method and removing the _drop_delayed_deliveries mechanism. The changes introduce three high-severity concurrency bugs and two test-breaking regressions.
Critical Issues
1. Test breakage: _cancel_pending_delivery_tasks removed but still called
tests/gateway/test_telegram_text_batching.py:297 — The test calls adapter._cancel_pending_delivery_tasks(), which was deleted from the adapter. This will raise AttributeError at test time.
2. Text-batch flush tasks leak past disconnect
plugins/platforms/telegram/adapter.py:2940 — The inline cancellation in disconnect() only handles photo-batch tasks (_photo_batch_task). Text-batch flush tasks created via asyncio.create_task(self._flush_text_batch()) are never cancelled, so they continue dispatching messages into a torn-down session.
3. Media-group task race: unconditional pop erases replacements
plugins/platforms/telegram/adapter.py:7308 — The finally block in _flush_media_group_event changed from a guarded pop (only when _media_group_tasks[id] is current_task) to an unconditional pop, creating a race where a newly scheduled replacement task is erased.
4. Six tests expect removed drop-on-disconnect behavior
tests/gateway/test_telegram_text_batching.py:185 — The _drop_delayed_deliveries attribute, _mark_connected, and _mark_disconnected overrides were removed without updating the six tests that depend on this behavior.
Recommendation
Do not merge. All four findings are high severity. The concurrency bugs are subtle data races that will cause message loss or corruption in production; the test regressions indicate incomplete refactoring. Restore the cancelled-task safety invariants and update all affected tests before re-submitting.
| @@ -3025,6 +2957,13 @@ async def disconnect(self) -> None: | |||
| logger.warning("[%s] Error during Telegram disconnect: %s", self.name, e, exc_info=True) | |||
| self._release_platform_lock() | |||
|
|
|||
| 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.
🟠 Text-batch flush tasks not cancelled during disconnect — can dispatch messages into torn-down session (bug)
The PR refactored disconnect() in plugins/platforms/telegram/adapter.py, removing the centralized _cancel_pending_delivery_tasks() method and replacing it with inline cancellation. The inline code at lines 2940–2964 only handles _media_group_tasks and _pending_photo_batch_tasks. _pending_text_batch_tasks receives no cancellation, awaiting, or clearing. Likewise, _pending_text_batches is never cleared. The _should_drop_delayed_delivery() guard that previously short-circuited _flush_text_batch and _enqueue_text_event was also removed entirely.
A _flush_text_batch task (line 6901) sleeping behind await asyncio.sleep(delay) at line 6932 can wake up after disconnect() has shut down the app (line 2955), released the platform lock (line 2958), and set self._app = None / self._bot = None (lines 2967–2968). It then pops its event from _pending_text_batches and calls self.handle_message(event) at line 6940, spawning agent logic on a torn-down adapter. The old code's _should_drop_delayed_delivery docstring explicitly warned: "If disconnect wins the race, dispatching them spawns an agent on a torn-down session, producing stale/duplicate deliveries."
💡 Suggestion: Add text-batch task cancellation inside disconnect() before or after the photo-batch cancellation block. Either restore the _should_drop_delayed_delivery() guard in _flush_text_batch as defense-in-depth, or ensure text-batch tasks are always cancelled and their dicts cleared.
📋 Prompt for AI Agents
In plugins/platforms/telegram/adapter.py, in the disconnect() method, add text-batch task cleanup after the photo-batch cancellation block (around line 2964):
# Cancel text-batch flush tasks to prevent stale deliveries
for task in self._pending_text_batch_tasks.values():
if task and not task.done():
task.cancel()
self._pending_text_batch_tasks.clear()
self._pending_text_batches.clear()| 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 unconditionally pops _media_group_tasks entry, erasing replacement task (bug)
In _flush_media_group_event (line 7299–7308), the finally block was changed from a guarded pop (if self._media_group_tasks.get(media_group_id) is current_task) to an unconditional pop (self._media_group_tasks.pop(media_group_id, None)).
When a follow-up album update arrives for the same media_group_id, _queue_media_group_event (line 7274) cancels the prior flush task and creates a replacement stored under the same key — all synchronously, before the cancelled task's CancelledError is delivered. The cancelled original's finally block then unconditionally pops the dict entry, removing the replacement task, not the original.
Consequences:
- The replacement task is no longer tracked in
_media_group_tasks, sodisconnect()cannot find and cancel it. - Subsequent album updates won't see the existing task to cancel/replace it, creating multiple concurrent flushes for the same album.
- An orphaned task may still dispatch
handle_messagewith partially-aggregated events.
This same guarded-pop pattern is correctly preserved in _flush_text_batch (line 6942) and _flush_photo_batch (line 6973), confirming the removal was unintended.
💡 Suggestion: Restore the current_task identity check in the finally block of _flush_media_group_event, matching the pattern still present in _flush_text_batch (line 6942) and _flush_photo_batch (line 6973).
📋 Prompt for AI Agents
In plugins/platforms/telegram/adapter.py, in _flush_media_group_event (line 7299), add back current_task = asyncio.current_task() as the first line of the method (before the try block). Then change line 7308 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)This matches the identical guarded-pop pattern used in _flush_text_batch (line 6942) and _flush_photo_batch (line 6973).
1 participant
Summary
The Telegram gateway can silently stop receiving messages while the process stays alive (systemd green, threads healthy, send path fine) — no logs, no errors — until manually restarted (NousResearch#55769). This makes the gateway self-heal from that state.
Root cause: the polling heartbeat's
_probe_pending_updatesalready handled a wedged-but-running long-poll consumer, but treated a fully stopped updater (running == False, no reconnect in flight) as "someone else's job" — it reset its counter and returned. Becauseget_me()on the general request path stays healthy, neither PTB'serror_callbacknor the connectivity heartbeat ever fires. Result: process alive, send path fine, polling dead, indefinitely.Changes
plugins/platforms/telegram/adapter.py—_probe_pending_updatesnow detectsupdater.running == Falseand feeds it into the existing_handle_polling_network_errorrecovery ladder (stop → drain pool →start_polling). Debounced over two consecutive probes via a new_polling_not_running_count. The in-flight-reconnect guard is moved ahead of the updater check so the reconnect's own transientstop()→start_polling()window (whererunningis briefly False) can't false-trip. No new restart machinery, no new config keys, no new env vars.tests/gateway/test_telegram_pending_update_probe.py— replaced the test that encoded the old buggy "stopped updater = no-op" assumption with coverage for: single stopped probe does not escalate, two consecutive stopped probes trigger recovery, a recovered (running) updater resets the counter, and an in-flight reconnect suppresses escalation.Validation
running=False)Salvaged from @PRATHAMESH75's PR NousResearch#55789, cherry-picked onto current
mainwith authorship preserved.Infographic
Nous Research
Mirror-of: NousResearch#55921
NousResearch#55921