fix(email): prevent orphaned poll tasks and replace deprecated utcnow in cron scheduling
linuztx committed
Mar 15, 2026 at 18:58 UTC
2211c4c14f0c489581bfe2ed7c4f3e2ee1310eef
2 files changed
+47
-57
plugins/_email_integration/extensions/python/job_loop/_10_email_poll.py
+41
-36
@@ -1,6 +1,8 @@
1
"""Per-handler email poll loop with configurable seconds/cron intervals."""
2
3
import asyncio
4
+from datetime import datetime, timezone
5
+from typing import Any
6
7
from crontab import CronTab
8
@@ -10,36 +12,46 @@ from helpers.print_style import PrintStyle
12
from helpers import plugins
13
14
13
-PLUGIN_NAME = "_email_integration"
14
-DEFAULT_INTERVAL = 15
15
-MIN_INTERVAL = 5
16
-
17
-_poll_tasks: dict[str, asyncio.Task] = {}
15
+PLUGIN_NAME: str = "_email_integration"
16
+DEFAULT_INTERVAL: int = 15
17
+MIN_INTERVAL: int = 5
18
19
20
# ------------------------------------------------------------------
21
-# Poll interval
21
+# Extension entry point
22
# ------------------------------------------------------------------
23
24
-def _get_sleep_seconds(handler_cfg: dict) -> float:
25
- mode = handler_cfg.get("poll_mode", "seconds")
26
- if mode == "cron":
27
- expr = handler_cfg.get("poll_interval_cron", "*/2 * * * *")
28
- try:
29
- return max(CronTab(expr).next(default_utc=True), MIN_INTERVAL)
30
- except Exception:
31
- return DEFAULT_INTERVAL
32
- return max(handler_cfg.get("poll_interval_seconds", DEFAULT_INTERVAL), MIN_INTERVAL)
24
+class EmailAutoPoll(Extension):
25
+
26
+ async def execute(self, **kwargs: Any) -> None:
27
+ # _poll_tasks lives in handler.py (persists across module reloads)
28
+ from plugins._email_integration.helpers.handler import _poll_tasks
29
+
30
+ config = plugins.get_plugin_config(PLUGIN_NAME) or {}
31
+ handlers = config.get("handlers", [])
32
+ enabled_names = {
33
+ h["name"] for h in handlers if h.get("enabled") and h.get("name")
34
+ }
35
+
36
+ for name in list(_poll_tasks):
37
+ if name not in enabled_names or _poll_tasks[name].done():
38
+ task = _poll_tasks.pop(name, None)
39
+ if task and not task.done():
40
+ task.cancel()
41
+
42
+ for name in enabled_names:
43
+ if name not in _poll_tasks or _poll_tasks[name].done():
44
+ _poll_tasks[name] = asyncio.create_task(_handler_poll_loop(name))
45
46
47
# ------------------------------------------------------------------
48
# Per-handler poll loop
49
# ------------------------------------------------------------------
50
39
-async def _handler_poll_loop(handler_name: str):
51
+async def _handler_poll_loop(handler_name: str) -> None:
52
from plugins._email_integration.helpers.handler import (
41
- _poll_single_handler,
53
_load_state,
54
+ _poll_single_handler,
55
_save_state,
56
_state_lock,
57
)
@@ -67,24 +79,17 @@ async def _handler_poll_loop(handler_name: str):
79
80
81
# ------------------------------------------------------------------
70
-# Extension entry point
82
+# Poll interval
83
# ------------------------------------------------------------------
84
73
-class EmailAutoPoll(Extension):
74
-
75
- async def execute(self, **kwargs):
76
- config = plugins.get_plugin_config(PLUGIN_NAME) or {}
77
- handlers = config.get("handlers", [])
78
- enabled_names = {
79
- h["name"] for h in handlers if h.get("enabled") and h.get("name")
80
- }
81
-
82
- for name in list(_poll_tasks):
83
- if name not in enabled_names or _poll_tasks[name].done():
84
- task = _poll_tasks.pop(name, None)
85
- if task and not task.done():
86
- task.cancel()
87
-
88
- for name in enabled_names:
89
- if name not in _poll_tasks or _poll_tasks[name].done():
90
- _poll_tasks[name] = asyncio.create_task(_handler_poll_loop(name))
85
+def _get_sleep_seconds(handler_cfg: dict) -> float:
86
+ mode = handler_cfg.get("poll_mode", "seconds")
87
+ if mode == "cron":
88
+ expr = handler_cfg.get("poll_interval_cron", "*/2 * * * *")
89
+ try:
90
+ cron = CronTab(expr)
91
+ next_sec = cron.next(now=datetime.now(timezone.utc)) # type: ignore[union-attr]
92
+ return max(next_sec, MIN_INTERVAL)
93
+ except Exception:
94
+ return DEFAULT_INTERVAL
95
+ return max(handler_cfg.get("poll_interval_seconds", DEFAULT_INTERVAL), MIN_INTERVAL)
plugins/_email_integration/helpers/handler.py
+6
-21
@@ -47,6 +47,11 @@ def _read_fw(filename: str, **kwargs: str) -> str:
47
48
_state_lock = asyncio.Lock()
49
50
+# Poll task registry — lives here (not in extension module) because
51
+# extension modules are re-executed on each job_loop tick (cache disabled),
52
+# which would reset module-level state and orphan running tasks.
53
+_poll_tasks: dict[str, asyncio.Task] = {} # type: ignore[type-arg]
54
+
55
def _load_state() -> dict:
56
path = files.get_abs_path(STATE_FILE)
57
if os.path.isfile(path):
@@ -64,29 +69,9 @@ def _save_state(state: dict):
69
70
71
# ------------------------------------------------------------------
67
-# Main auto-poll entry point (called from job_loop extension)
72
+# Single handler poll (called from per-handler poll loop)
73
# ------------------------------------------------------------------
74
70
-async def poll_all_handlers():
71
- config = plugins.get_plugin_config(PLUGIN_NAME) or {}
72
- handlers = config.get("handlers", [])
73
- enabled = [h for h in handlers if h.get("enabled", False)]
74
-
75
- if not enabled:
76
- return
77
-
78
- state = _load_state()
79
-
80
- for handler_cfg in enabled:
81
- try:
82
- await _poll_single_handler(handler_cfg, state)
83
- except Exception as e:
84
- name = handler_cfg.get("name", "?")
85
- PrintStyle.error(f"Email poll error ({name}): {format_error(e)}")
86
-
87
- _save_state(state)
88
-
89
-
75
async def _poll_single_handler(handler_cfg: dict, state: dict):
76
name = handler_cfg.get("name", "default")
77
account_type = handler_cfg.get("account_type", "imap")