src/genro_asgi/tasks/scheduler.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """TaskScheduler — the loop that runs due schedules (◆D22).16 17 One asyncio task owned by the ``TaskManager`` (started with the manager on18 lifespan startup, dies in shutdown), tick ~30s (cron granularity is the minute).19 Each tick:20 21 - **scan**: walk every mounted app's routing tree (``nodes(lazy=True,22 forbidden=True)`` — a gated route is still a declared task) collecting entries23 whose metadata carries ``task``. The tree IS the live registry: no register()24 API. A task_name declared twice is an ERROR — both are excluded, no silent25 pick. A record whose task_name nobody declares is an ORPHAN: never executed,26 never deleted (a UI concern).27 - **sync defaults**: a declared ``task_every``/``task_cron`` auto-creates the28 missing code-default record (code = task_name); an existing record always wins29 (store override) — deleting it resets to the code default at the next scan.30 - **run**: for each due record resolve the live callable and spawn it — async31 handlers on the loop, sync ones through ``server.run_sync`` (the D2 pool). No32 overlap: a schedule whose previous run is still in flight is skipped. Runs are33 SYSTEM calls: no middleware chain, no auth filters.34 35 On completion the record's ``last_*``/``next_run_ts`` update and a JSONL line is36 appended to the task's capped log. ``run_now`` fires a schedule immediately (from37 another thread when the loop is up, inline otherwise) with the same no-overlap38 guard and the same outcome path.39 40 The store lives on the ``TaskManager`` (``manager.task_store``), not the server;41 the scheduler reaches the live server through ``manager.server``. Store I/O and42 sync task bodies are synchronous by construction (core 1b) and dispatched via43 ``server.run_sync`` (a zero-arg closure — ``run_sync`` takes no kwargs).44 """45 46 from __future__ import annotations47 48 import asyncio49 import contextlib50 import logging51 import time52 from typing import TYPE_CHECKING, Any53 54 from .schedule import TaskCadence55 56 if TYPE_CHECKING:57 from .manager import TaskManager58 59 __all__ = ["TaskScheduler", "TICK_SECONDS"]60 61 TICK_SECONDS = 30.062 63 64 class TaskScheduler:65 """The scheduling loop bound to its manager (dual relationship)."""66 67 __slots__ = ("manager", "tick_seconds", "_running", "_loop", "_loop_task")68 69 def __init__(self, manager: TaskManager) -> None:70 """Bind the scheduler to the manager owning the store, server and loop.71 72 Args:73 manager: The TaskManager; the store is ``manager.task_store``, the74 live server is ``manager.server``, the live registry is the75 mounted apps' routing trees.76 """77 self.manager = manager78 self.tick_seconds = TICK_SECONDS79 self._running: set[str] = set()80 self._loop: asyncio.AbstractEventLoop | None = None81 self._loop_task: asyncio.Task[None] | None = None82 83 @property84 def server(self) -> Any:85 """The live server (via the manager)."""86 return self.manager.server87 88 @property89 def store(self) -> Any:90 """The manager's TaskStore."""91 return self.manager.task_store92 93 @property94 def running(self) -> set[str]:95 """The codes with a run currently in flight (a copy)."""96 return set(self._running)97 98 # -- lifecycle --99 100 def start(self) -> None:101 """Start the tick loop on the running event loop (lifespan startup)."""102 self._loop = asyncio.get_running_loop()103 self._loop_task = self._loop.create_task(self._run_loop())104 logging.getLogger(__name__).info("task scheduler started (tick %ss)", self.tick_seconds)105 106 async def stop(self) -> None:107 """Cancel the tick loop (lifespan shutdown); in-flight runs finish."""108 if self._loop_task is not None:109 self._loop_task.cancel()110 with contextlib.suppress(asyncio.CancelledError):111 await self._loop_task112 self._loop_task = None113 logging.getLogger(__name__).info("task scheduler stopped")114 115 async def _run_loop(self) -> None:116 """Tick forever; a failing tick is logged and never kills the loop."""117 while True:118 try:119 await self.tick()120 except Exception:121 logging.getLogger(__name__).exception("task scheduler tick failed")122 await asyncio.sleep(self.tick_seconds)123 124 # -- the live registry --125 126 def _apps(self) -> Any:127 """Every application the server serves, in registration order."""128 return self.server.applications.values()129 130 def scan(self) -> dict[str, dict[str, Any]]:131 """task_name -> {"callable", "metadata"} from every mounted app's tree.132 133 Duplicates (the same task_name declared by two routes) are an explicit134 error: logged and EXCLUDED — no silent pick.135 """136 found: dict[str, dict[str, Any]] = {}137 duplicates: set[str] = set()138 for app in self._apps():139 router = getattr(app, "route", None)140 if router is None:141 continue142 for entry in self._walk(router):143 metadata = entry.get("metadata") or {}144 task_name = metadata.get("task")145 if not task_name:146 continue147 if task_name in found:148 duplicates.add(task_name)149 continue150 found[task_name] = {"callable": entry["callable"], "metadata": metadata}151 for task_name in duplicates:152 found.pop(task_name, None)153 logging.getLogger(__name__).error(154 "duplicate task name %r: declared by more than one route", task_name155 )156 return found157 158 def _walk(self, router: Any) -> Any:159 """Yield every handler entry in a router tree (structural view)."""160 tree = router.nodes(lazy=True, forbidden=True)161 yield from (tree.get("entries") or {}).values()162 for sub_router in (tree.get("routers") or {}).values():163 yield from self._walk(sub_router)164 165 def sync_defaults(self, registry: dict[str, dict[str, Any]], now: float) -> None:166 """Auto-create the code-default record for declared default schedules."""167 for task_name, info in registry.items():168 metadata = info["metadata"]169 every, cron = metadata.get("task_every"), metadata.get("task_cron")170 if every and cron:171 logging.getLogger(__name__).error(172 "task %r declares both task_every and task_cron", task_name173 )174 continue175 if not every and not cron:176 continue # schedulable, but only through store records177 kind, spec = ("every", every) if every else ("cron", cron)178 try:179 first_run = TaskCadence(kind, spec).get_next_run(now)180 except ValueError:181 logging.getLogger(__name__).exception(182 "task %r: invalid default schedule %r", task_name, spec183 )184 continue185 self.store.upsert_default(186 {187 "code": task_name,188 "task_name": task_name,189 "target_kind": "task",190 "kwargs": {},191 "kind": kind,192 "spec": spec,193 "enabled": True,194 "next_run_ts": first_run,195 "last_run_ts": None,196 "last_outcome": None,197 "last_error": None,198 "last_duration": None,199 }200 )201 202 # -- ticking and running --203 204 async def tick(self) -> None:205 """One pass: scan, sync defaults, spawn every due schedule.206 207 Store I/O (``sync_defaults`` writes, ``due_rows`` reads every record)208 runs on the server pool via ``run_sync``, never on the loop.209 """210 now = time.time()211 registry = self.scan()212 await self.server.run_sync(lambda: self.sync_defaults(registry, now))213 loop = asyncio.get_running_loop()214 due = await self.server.run_sync(lambda: self.store.due_rows(now))215 for row in due:216 if row.get("target_kind", "task") != "task":217 continue # reserved for the future privileged mode218 code = row["code"]219 if code in self._running:220 continue # no overlap221 info = registry.get(row["task_name"])222 if info is None:223 continue # orphan: never runs it (a UI concern)224 self._running.add(code)225 loop.create_task(self._execute(row, info["callable"]))226 227 async def _execute(self, row: dict[str, Any], task_callable: Any) -> None:228 """Run one schedule and settle its outcome (record + JSONL log).229 230 A sync task body runs through ``server.run_sync`` (the D2 pool — same231 rule as the executor); the store writes go there too.232 """233 code, task_name = row["code"], row["task_name"]234 kwargs = row.get("kwargs") or {}235 started = time.time()236 outcome, error = "ok", None237 try:238 if asyncio.iscoroutinefunction(task_callable):239 await task_callable(**kwargs)240 else:241 await self.server.run_sync(lambda: task_callable(**kwargs))242 except Exception as exc:243 outcome, error = "error", f"{type(exc).__name__}: {exc}"244 logging.getLogger(__name__).exception(245 "task %r (schedule %r) failed", task_name, code246 )247 duration = round(time.time() - started, 3)248 try:249 next_ts = TaskCadence(row["kind"], row["spec"]).get_next_run(started)250 except ValueError as exc:251 next_ts = None252 outcome, error = "error", error or f"invalid schedule: {exc}"253 await self.server.run_sync(254 lambda: self.store.update_run(255 code,256 last_run_ts=started,257 last_outcome=outcome,258 last_error=error,259 last_duration=duration,260 next_run_ts=next_ts,261 )262 )263 await self.server.run_sync(264 lambda: self.store.append_log(265 task_name,266 {"ts": started, "code": code, "outcome": outcome, "error": error,267 "duration": duration},268 )269 )270 self._running.discard(code)271 272 def run_now(self, code: str) -> str:273 """Fire a schedule immediately (the UI's Run now). Thread-safe.274 275 Returns:276 ``"started"`` (fired on the live loop), ``"done"`` (executed277 inline — no loop running, e.g. in tests), or ``"running"``278 (skipped: the previous run is still in flight).279 280 Raises:281 LookupError: If the code is unknown or its task_name is an orphan.282 """283 row = self.store.get(code)284 if row is None:285 raise LookupError(f"schedule not found: {code}")286 info = self.scan().get(row["task_name"])287 if info is None:288 raise LookupError(f"orphan task: {row['task_name']} (no mounted route declares it)")289 if code in self._running:290 return "running"291 self._running.add(code)292 if self._loop is not None and self._loop.is_running():293 asyncio.run_coroutine_threadsafe(self._execute(row, info["callable"]), self._loop)294 return "started"295 asyncio.run(self._execute(row, info["callable"]))296 return "done"