src/genro_asgi/tasks/spool.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 """TaskSpool — the file spool of batch tasks (folder-move model, ◆D22).16 17 A batch task is a FOLDER on storage; its STATE is its POSITION in the tree. The18 sender (an app handler, or the scheduler) creates the folder under ``pending/``;19 the manager MOVES it into the assigned worker's folder; the worker runs it and20 MOVES it to ``terminated/`` or ``aborted/``. A move is the only state transition21 — one mount, one ``StorageNode.move_to``.22 23 Layout (all on the plain ``site`` mount — these are service data, never secrets,24 so NO encryption, exactly like the task STORE alongside it)::25 26 batches/27 pending/<task_id>/ sender writes here; the manager polls it28 active/<worker_id>/<task_id>/ the manager moves it here to assign it29 terminated/<task_id>/ completed30 aborted/<task_id>/ interrupted (user cancel / error / crash)31 32 Inside a task folder::33 34 descriptor.json the TaskDescriptor (identity + node_path + status + outcome)35 params.pkl the call kwargs (pickled: may carry Python objects)36 progress.json the worker's latest progress snapshot (single-writer: the worker)37 cancel marker file: present == the user requested a stop38 result the batch result (written by the worker on completion)39 40 A ``batch_id`` is terminal (§5.7): it never resumes or relaunches itself. An41 orphan (a task left under ``active/`` whose worker died) is settled ``aborted``42 at boot; relaunch = a NEW id. There are no locks: safety is structural — one43 single writer per state directory (sender on pending, one worker per active44 subfolder), and a claim is an atomic rename.45 46 The spool is the shared object the sender, the manager and the worker all use,47 each with the methods that concern it. Storage is synchronous by construction48 (core 1b): async callers dispatch blocking spool calls via ``server.run_sync``.49 """50 51 from __future__ import annotations52 53 import json54 import pickle55 from datetime import datetime, timezone56 from typing import TYPE_CHECKING, Any57 58 if TYPE_CHECKING:59 from genro_storage import StorageManager, StorageNode60 61 __all__ = ["TaskSpool", "new_descriptor", "STATUSES"]62 63 MOUNT = "site" # plain mount: spool data are service data, never encrypted64 ROOT = "batches"65 PENDING = "pending"66 ACTIVE = "active"67 TERMINATED = "terminated"68 ABORTED = "aborted"69 STATUSES = (PENDING, ACTIVE, TERMINATED, ABORTED)70 71 DESCRIPTOR_FILE = "descriptor.json"72 PARAMS_FILE = "params.pkl"73 PROGRESS_FILE = "progress.json"74 CANCEL_FILE = "cancel"75 RESULT_FILE = "result"76 77 78 def _now() -> float:79 """Current epoch seconds (UTC-based)."""80 return datetime.now(timezone.utc).timestamp()81 82 83 def new_descriptor(84 task_id: str,85 owner: str,86 mount: str,87 node_path: str,88 session_id: str | None = None,89 ) -> dict[str, Any]:90 """Build a fresh TaskDescriptor dict in the ``pending`` state.91 92 The autosufficient shape a worker needs to run the task without the sender's93 context: identity, the owner (for ``list_by_owner`` and progress), and how to94 resolve the code — ``mount`` (the app) + ``node_path`` (resolved by the worker95 with ``Router.node(node_path)``, callable, no HTTP). ``session_id`` is the96 launching MCP session, the key the executor publishes progress events under97 (None when the sender has no push channel). ``params`` travel in a separate98 pickled file, not inline here.99 """100 return {101 "task_id": task_id,102 "owner": owner,103 "mount": mount,104 "node_path": node_path,105 "session_id": session_id,106 "status": PENDING,107 "worker_id": None,108 "created_ts": _now(),109 "started_ts": None,110 "ended_ts": None,111 "outcome": None,112 "error": None,113 }114 115 116 class TaskSpool:117 """The file spool: create in pending, move between states, read progress/cancel.118 119 Note:120 Holds the shared ``StorageManager`` (dual relationship: ``self.storage``),121 never raw paths. All I/O is plain synchronous storage calls.122 """123 124 __slots__ = ("storage",)125 126 def __init__(self, storage: StorageManager) -> None:127 """Bind the spool to the server's storage service (the ``site`` mount)."""128 self.storage = storage129 130 # -- folder addressing --131 132 def _task_node(self, status: str, task_id: str, worker_id: str | None = None) -> StorageNode:133 """The folder node of ``task_id`` in ``status`` (worker_id only for ACTIVE)."""134 if status == ACTIVE:135 return self.storage.node(f"{MOUNT}:{ROOT}/{ACTIVE}/{worker_id}/{task_id}")136 return self.storage.node(f"{MOUNT}:{ROOT}/{status}/{task_id}")137 138 def _state_dir(self, *parts: str) -> StorageNode:139 """A state directory node (e.g. ``pending/`` or ``active/<worker_id>/``)."""140 return self.storage.node(f"{MOUNT}:{ROOT}/" + "/".join(parts))141 142 # -- I/O helpers --143 144 def _read_json(self, node: StorageNode) -> dict[str, Any] | None:145 if not node.exists():146 return None147 return json.loads(node.read_text())148 149 def _write_json(self, node: StorageNode, data: dict[str, Any]) -> None:150 node.write_text(json.dumps(data, indent=2))151 152 def _find_folder(self, task_id: str) -> StorageNode | None:153 """Locate a task's folder in whatever state it currently sits (or None)."""154 for status in (PENDING, TERMINATED, ABORTED):155 node = self._task_node(status, task_id)156 if node.exists():157 return node158 active_root = self._state_dir(ACTIVE)159 if active_root.is_dir():160 for worker_dir in active_root.children():161 candidate = worker_dir.child(task_id)162 if candidate.exists():163 return candidate164 return None165 166 # -- creation (used by the SENDER: an app handler, or the scheduler) --167 168 def create(self, descriptor: dict[str, Any], params: dict[str, Any]) -> str:169 """Create a task folder under ``pending/`` with its descriptor and params.170 171 Writes ``descriptor.json`` and the pickled ``params.pkl``. Returns the172 ``task_id``. This is a plain storage fact done by whoever launches the173 batch; the manager only ever MOVES the folder afterwards.174 """175 task_id = descriptor["task_id"]176 folder = self._task_node(PENDING, task_id)177 self._write_json(folder.child(DESCRIPTOR_FILE), descriptor)178 folder.child(PARAMS_FILE).write_bytes(pickle.dumps(params))179 return task_id180 181 def read_params(self, task_id: str) -> dict[str, Any]:182 """Unpickle a task's params (the worker reads them to run the task)."""183 folder = self._find_folder(task_id)184 if folder is None:185 raise LookupError(f"task not found: {task_id}")186 return pickle.loads(folder.child(PARAMS_FILE).read_bytes())187 188 # -- discovery (the manager polls pending; a worker polls its own active) --189 190 def list_pending(self) -> list[dict[str, Any]]:191 """The descriptors of every task under ``pending/`` (the manager's queue)."""192 return self._list_descriptors(self._state_dir(PENDING))193 194 def list_active(self, worker_id: str) -> list[dict[str, Any]]:195 """The descriptors of ``worker_id``'s active tasks (that worker's queue)."""196 return self._list_descriptors(self._state_dir(ACTIVE, worker_id))197 198 def _list_descriptors(self, directory: StorageNode) -> list[dict[str, Any]]:199 if not directory.is_dir():200 return []201 result: list[dict[str, Any]] = []202 for folder in directory.children():203 if folder.is_dir():204 descriptor = self._read_json(folder.child(DESCRIPTOR_FILE))205 if descriptor is not None:206 result.append(descriptor)207 return result208 209 # -- state transitions = MOVE --210 211 def assign(self, task_id: str, worker_id: str) -> None:212 """Move ``pending/<id>`` -> ``active/<worker_id>/<id>`` (the manager's act).213 214 Stamps ``status=active``, ``worker_id`` and ``started_ts`` on the215 descriptor, then moves the folder. The move is atomic on the mount.216 217 Raises:218 LookupError: if the task is not in ``pending``.219 """220 src = self._task_node(PENDING, task_id)221 if not src.exists():222 raise LookupError(f"task not pending: {task_id}")223 descriptor = self._read_json(src.child(DESCRIPTOR_FILE)) or {}224 descriptor.update(status=ACTIVE, worker_id=worker_id, started_ts=_now())225 self._write_json(src.child(DESCRIPTOR_FILE), descriptor)226 src.move_to(self._task_node(ACTIVE, task_id, worker_id))227 228 def settle(229 self, task_id: str, worker_id: str, outcome: str, error: str | None = None230 ) -> None:231 """Move ``active/<worker_id>/<id>`` -> ``terminated/`` or ``aborted/``.232 233 ``outcome`` == "ok" settles under ``terminated/``; anything else (error,234 aborted, orphan) settles under ``aborted/``. Stamps the descriptor with235 ``ended_ts``/``outcome``/``error``. Terminal: a settled task never moves236 again — re-settling raises because the folder is no longer active.237 238 Raises:239 LookupError: if the task is not active on ``worker_id``.240 """241 src = self._task_node(ACTIVE, task_id, worker_id)242 if not src.exists():243 raise LookupError(f"task not active on {worker_id}: {task_id}")244 target_status = TERMINATED if outcome == "ok" else ABORTED245 descriptor = self._read_json(src.child(DESCRIPTOR_FILE)) or {}246 descriptor.update(247 status=target_status, ended_ts=_now(), outcome=outcome, error=error248 )249 self._write_json(src.child(DESCRIPTOR_FILE), descriptor)250 src.move_to(self._task_node(target_status, task_id))251 252 # -- progress / cancel (files inside the task folder) --253 254 def write_progress(self, task_id: str, worker_id: str, data: dict[str, Any]) -> None:255 """Write the worker's latest progress snapshot (single-writer: the worker)."""256 folder = self._task_node(ACTIVE, task_id, worker_id)257 self._write_json(folder.child(PROGRESS_FILE), data)258 259 def read_progress(self, task_id: str) -> dict[str, Any] | None:260 """Read a task's latest progress snapshot, or None (monitor / SSE baseline)."""261 folder = self._find_folder(task_id)262 if folder is None:263 return None264 return self._read_json(folder.child(PROGRESS_FILE))265 266 def request_cancel(self, task_id: str) -> None:267 """Drop the ``cancel`` marker in the task folder (the user's stop request)."""268 folder = self._find_folder(task_id)269 if folder is None:270 raise LookupError(f"task not found: {task_id}")271 folder.child(CANCEL_FILE).write_text("")272 273 def is_cancelled(self, task_id: str) -> bool:274 """True if a ``cancel`` marker is present (the worker checks at each tick)."""275 folder = self._find_folder(task_id)276 return folder is not None and folder.child(CANCEL_FILE).exists()277 278 def write_result(self, task_id: str, worker_id: str, data: Any) -> None:279 """Persist the batch result in the task folder (pickled)."""280 folder = self._task_node(ACTIVE, task_id, worker_id)281 folder.child(RESULT_FILE).write_bytes(pickle.dumps(data))282 283 def read_result(self, task_id: str) -> Any:284 """Unpickle a task's result, or None if not written yet."""285 folder = self._find_folder(task_id)286 if folder is None:287 return None288 node = folder.child(RESULT_FILE)289 if not node.exists():290 return None291 return pickle.loads(node.read_bytes())292 293 # -- queries (the page and the monitor) --294 295 def get(self, task_id: str) -> dict[str, Any] | None:296 """The descriptor of ``task_id`` in whatever state it sits, or None."""297 folder = self._find_folder(task_id)298 if folder is None:299 return None300 return self._read_json(folder.child(DESCRIPTOR_FILE))301 302 def list_by_owner(self, owner: str) -> list[dict[str, Any]]:303 """Every task of ``owner`` across ALL states (pending/active/terminated/aborted).304 305 The user's full picture: queued, running, finished, aborted. Reads all306 state directories; the monitor uses the raw states, the page filters by307 owner here.308 """309 return [d for d in self._all_descriptors() if d.get("owner") == owner]310 311 def list_by_status(self, status: str) -> list[dict[str, Any]]:312 """Every task in one state (the monitor's pending / per-worker views).313 314 For ``active`` this spans all worker subfolders; each descriptor carries its315 ``worker_id``, so the monitor can group per worker.316 """317 if status == ACTIVE:318 active_root = self._state_dir(ACTIVE)319 if not active_root.is_dir():320 return []321 result: list[dict[str, Any]] = []322 for worker_dir in active_root.children():323 if worker_dir.is_dir():324 result.extend(self._list_descriptors(worker_dir))325 return result326 return self._list_descriptors(self._state_dir(status))327 328 def _all_descriptors(self) -> list[dict[str, Any]]:329 result: list[dict[str, Any]] = []330 for status in STATUSES:331 result.extend(self.list_by_status(status))332 return result333 334 # -- housekeeping --335 336 def purge(self, task_id: str) -> bool:337 """Remove a settled task's folder entirely (tree-removal). True if removed."""338 folder = self._find_folder(task_id)339 if folder is None:340 return False341 folder.delete()342 return True