diff --git a/runtime/src/runtime/main.py b/runtime/src/runtime/main.py index a18a6f5..adb228f 100644 --- a/runtime/src/runtime/main.py +++ b/runtime/src/runtime/main.py @@ -19,8 +19,11 @@ from runtime.process import ( JUPYTER_PROCESSES, get_workspace, list_workspaces, + reconcile_processes, scan_workspaces, + start_reaper, start_workspace, + stop_reaper, stop_workspace, ) @@ -37,12 +40,26 @@ async def lifespan(app: FastAPI): logger.info("Starting up Runtime Service...") start_rclone_mount() logger.info("Scanning workspaces") + # P1-3: clean up stale sidecar metadata before we adopt any + # workspace. Orphans from a previous runtime incarnation are not + # re-adopted — see reconcile_processes() docstring. + counters = reconcile_processes() + logger.info( + f"Reconcile: scanned={counters['scanned']} " + f"removed_meta={counters['removed_meta']} " + f"live_in_registry={counters['live_in_registry']}" + ) await scan_workspaces() + # P1-3: spawn background idle reaper. Stopped in the lifespan + # finally block; awaits the cancellation to avoid a leaked task. + start_reaper() logger.info("Runtime Service started") try: yield finally: + logger.info("Service is shutting down. Stopping reaper...") + await stop_reaper() logger.info( "Service is shutting down. Terminating all active Jupyter sub-processes..." ) diff --git a/runtime/src/runtime/process.py b/runtime/src/runtime/process.py index 61dc294..8c7d861 100644 --- a/runtime/src/runtime/process.py +++ b/runtime/src/runtime/process.py @@ -4,11 +4,37 @@ Owns the in-memory ``JUPYTER_PROCESSES`` dict, the ``STATE_LOCK`` that serializes mutations to it, and the per-workspace start/stop/list/get operations. Workspace discovery (startup scan) lives here because it is a thin wrapper over ``start_workspace``. + +Process reconciliation and idle reaping (P1-3): + +* **On startup** ``reconcile_processes()`` walks the workspaces root + and removes ``.runtime-meta.json`` files whose owning Jupyter + process is no longer tracked. The runtime only knows about + workspaces it has explicitly started in this lifetime; orphans from + a previous process incarnation are simply not reaped — the + workspace directory on disk is left alone, the runtime does not + attempt to re-launch stray PIDs. This is the safe default: we never + re-adopt a process we did not fork, because we cannot trust its + runtime state. + +* **Idle reaping** runs in a background task started by + ``runtime.main.lifespan``. Every 60s the reaper looks for processes + whose ``last_used_at`` is older than ``JUPYTER_IDLE_TIMEOUT_SECONDS`` + (default 30 minutes) and gracefully terminates them. ``last_used_at`` + is bumped on ``start_workspace``, ``get_workspace``, and any + Jupyter HTTP interaction routed through this registry. Because we + don't see individual kernel API calls (those go straight through + nginx), the reaper is conservative — a long-running notebook that + is actually busy will still be torn down at the idle threshold. A + future iteration could either (a) poll the Jupyter ``/api/status`` + endpoint to count active kernels, or (b) require the editor UI to + send a heartbeat. Both are out of scope here. """ from __future__ import annotations import asyncio +import json import os import secrets import subprocess @@ -24,6 +50,27 @@ from runtime.mount import WORKSPACES_ROOT PUBLIC_BASE_URL = settings.public_base_url +# P1-3: idle reaping tuning knobs. The default 30 minutes matches the +# "user stepped away" mental model; a power user can override via env +# if they need to keep notebooks warm longer. +JUPYTER_IDLE_TIMEOUT_SECONDS = int( + os.environ.get("JUPYTER_IDLE_TIMEOUT_SECONDS", str(30 * 60)) +) +JUPYTER_REAP_INTERVAL_SECONDS = int( + os.environ.get("JUPYTER_REAP_INTERVAL_SECONDS", "60") +) +JUPYTER_MAX_LIFETIME_SECONDS = int( + os.environ.get("JUPYTER_MAX_LIFETIME_SECONDS", str(24 * 3600)) +) + +# Sidecar metadata file written next to the workspace directory. Holds +# the runtime state we need to reconstruct the in-memory map after a +# crash. We intentionally store only the bits that are safe to recover: +# process pid, port, started_at. ``last_used_at`` is intentionally +# NOT persisted — a fresh process inherits the "just started" state +# and gets 30 minutes before the first reap. +RUNTIME_META_FILENAME = ".runtime-meta.json" + class JupyterProcessRecord(TypedDict): process: subprocess.Popen @@ -31,11 +78,14 @@ class JupyterProcessRecord(TypedDict): token: str base_url: str started_at: float + last_used_at: float + meta_path: str JUPYTER_PROCESSES: dict[str, JupyterProcessRecord] = {} WORKSPACE_LOCKS: dict[str, asyncio.Lock] = {} _LOCKS_REGISTRY = asyncio.Lock() +_REAPER_TASK: asyncio.Task[None] | None = None def get_workspace_lock(ws_id: str) -> asyncio.Lock: @@ -68,18 +118,60 @@ def _drop_workspace_lock(ws_id: str) -> None: WORKSPACE_LOCKS.pop(ws_id, None) +def _meta_path(ws_id: str) -> str: + """Return the sidecar metadata file path for ``ws_id``.""" + return str(WORKSPACES_ROOT / ws_id / RUNTIME_META_FILENAME) + + +def _write_meta(ws_id: str, record: JupyterProcessRecord) -> None: + """Persist the bits of the record we can safely recover. + + Best-effort — failures (read-only mount, vanished dir) are logged + and swallowed; the in-memory state is the source of truth. + """ + payload = { + "workspace_id": ws_id, + "pid": record["process"].pid, + "port": record["port"], + "started_at": record["started_at"], + } + try: + with open(record["meta_path"], "w", encoding="utf-8") as fp: + json.dump(payload, fp) + except Exception as exc: # pragma: no cover - defensive + logger.warning(f"failed to write runtime meta for {ws_id}: {exc}") + + +def _delete_meta(ws_id: str) -> None: + """Best-effort remove of the sidecar metadata file.""" + try: + os.unlink(_meta_path(ws_id)) + except FileNotFoundError: + pass + except Exception as exc: # pragma: no cover - defensive + logger.warning(f"failed to delete runtime meta for {ws_id}: {exc}") + + +def _bump_last_used(record: JupyterProcessRecord) -> None: + """Touch ``last_used_at`` so the idle reaper does not eat active ws.""" + record["last_used_at"] = time.time() + + async def start_workspace(ws_id: str) -> dict: async with get_workspace_lock(ws_id): workspace_path = WORKSPACES_ROOT / ws_id + workspace_path.mkdir(parents=True, exist_ok=True) if ws_id in JUPYTER_PROCESSES: p_info = JUPYTER_PROCESSES[ws_id] if p_info["process"].poll() is None: - if time.time() - p_info["started_at"] > 24 * 3600: + if time.time() - p_info["started_at"] > JUPYTER_MAX_LIFETIME_SECONDS: logger.warning( - f"Reusing Jupyter for {ws_id} older than 24h " + f"Reusing Jupyter for {ws_id} older than " + f"{JUPYTER_MAX_LIFETIME_SECONDS}s " f"(started_at={p_info['started_at']})" ) + _bump_last_used(p_info) return { "status": "running", "workspace_id": ws_id, @@ -92,6 +184,9 @@ async def start_workspace(ws_id: str) -> dict: f"?token={p_info['token']}" ), } + # Process died but we still hold a record — drop it and + # start a fresh one. + _delete_meta(ws_id) del JUPYTER_PROCESSES[ws_id] port = get_free_port() @@ -124,14 +219,18 @@ async def start_workspace(ws_id: str) -> dict: ) full_url = f"{PUBLIC_BASE_URL}:{port}{base_path}?token={token}" - + meta_path = _meta_path(ws_id) + now = time.time() JUPYTER_PROCESSES[ws_id] = { "process": process, "base_url": PUBLIC_BASE_URL, "port": port, "token": token, - "started_at": time.time(), + "started_at": now, + "last_used_at": now, + "meta_path": meta_path, } + _write_meta(ws_id, JUPYTER_PROCESSES[ws_id]) logger.info( f"Started Jupyter for workspace {ws_id} " @@ -175,6 +274,7 @@ async def stop_workspace(ws_id: str) -> dict: logger.error(f"Failed to kill Jupyter process for {ws_id}: {err}") del JUPYTER_PROCESSES[ws_id] + _delete_meta(ws_id) _drop_workspace_lock(ws_id) return { @@ -197,6 +297,8 @@ async def list_workspaces() -> dict: f"?token={info['token']}" ), "is_alive": info["process"].poll() is None, + "last_used_at": info["last_used_at"], + "started_at": info["started_at"], } for ws_id, info in snapshot.items() }, @@ -216,7 +318,9 @@ async def get_workspace(ws_id: str) -> dict: if not is_alive: del JUPYTER_PROCESSES[ws_id] + _delete_meta(ws_id) else: + _bump_last_used(p_info) return { "status": "running", "pid": p_info["process"].pid, @@ -229,6 +333,7 @@ async def get_workspace(ws_id: str) -> dict: f"?token={p_info['token']}" ), "started_at": p_info["started_at"], + "last_used_at": p_info["last_used_at"], } _drop_workspace_lock(ws_id) @@ -241,6 +346,152 @@ async def get_workspace(ws_id: str) -> dict: ) +def reconcile_processes() -> dict[str, int]: + """P1-3 startup sweep. + + Walks the workspaces root and inspects ``.runtime-meta.json`` + sidecars. A sidecar whose workspace directory has no live + ``JUPYTER_PROCESSES`` entry AND whose recorded pid is not + running on this host is treated as a stale artifact and removed + (the workspace directory itself is left intact — that is user + data we have no right to touch). + + We do NOT scan ``/proc`` to find orphan jupyter processes that + have no sidecar at all. The runtime has no way to associate + such a process with a workspace without the sidecar, and + adopting a foreign process is unsafe. Stale orphans from a + previous runtime incarnation will keep their port and + workspace files; an operator can ``docker compose restart + runtime`` to fully reset. + + Returns a small counter dict so the caller can log it. + """ + counters = {"scanned": 0, "removed_meta": 0, "live_in_registry": 0} + if not WORKSPACES_ROOT.exists(): + return counters + try: + entries = os.listdir(WORKSPACES_ROOT) + except Exception as exc: + logger.warning(f"reconcile_processes: cannot list workspaces: {exc}") + return counters + for entry in entries: + ws_path = WORKSPACES_ROOT / entry + if not ws_path.is_dir(): + continue + meta_file = ws_path / RUNTIME_META_FILENAME + if not meta_file.exists(): + continue + counters["scanned"] += 1 + if entry in JUPYTER_PROCESSES: + counters["live_in_registry"] += 1 + # Trust the in-memory record (this process is the one that + # wrote the sidecar most recently). Refresh it. + _write_meta(entry, JUPYTER_PROCESSES[entry]) + continue + # Stale: this process did not start the Jupyter in question. + # Verify the pid is dead; if it is, drop the sidecar so the + # next scan is clean. If a live jupyter process with that pid + # exists we leave the sidecar alone (it might be a sibling + # runtime; the next call to start_workspace will reuse the + # port if available). + try: + with open(meta_file, encoding="utf-8") as fp: + meta = json.load(fp) + pid = int(meta.get("pid", 0)) + except Exception as exc: + logger.info(f"reconcile: removing malformed meta for {entry}: {exc}") + _delete_meta(entry) + counters["removed_meta"] += 1 + continue + if pid <= 0: + _delete_meta(entry) + counters["removed_meta"] += 1 + continue + try: + os.kill(pid, 0) + alive = True + except ProcessLookupError: + alive = False + except PermissionError: + alive = True # someone else's process, leave alone + if not alive: + logger.info(f"reconcile: dropping stale sidecar for {entry} (pid {pid} dead)") + _delete_meta(entry) + counters["removed_meta"] += 1 + return counters + + +async def _reap_loop() -> None: + """Background task: idle + dead reaper. + + Runs every ``JUPYTER_REAP_INTERVAL_SECONDS``; stops on + :class:`asyncio.CancelledError`. Decisions: + * process is dead (poll() != None) → drop the record + meta + * process is alive but ``last_used_at`` is older than the + idle timeout → graceful stop + * process is alive but older than the max lifetime → graceful + stop (defence in depth: prevents a workspace from holding a + port forever) + """ + while True: + try: + await asyncio.sleep(JUPYTER_REAP_INTERVAL_SECONDS) + now = time.time() + victims: list[str] = [] + for ws_id, info in list(JUPYTER_PROCESSES.items()): + if info["process"].poll() is not None: + logger.info(f"reap: dead process for {ws_id}, cleaning up") + del JUPYTER_PROCESSES[ws_id] + _delete_meta(ws_id) + continue + age_idle = now - info["last_used_at"] + age_total = now - info["started_at"] + if age_idle > JUPYTER_IDLE_TIMEOUT_SECONDS: + logger.info( + f"reap: idle Jupyter for {ws_id} " + f"(idle={age_idle:.0f}s > {JUPYTER_IDLE_TIMEOUT_SECONDS}s)" + ) + victims.append(ws_id) + elif age_total > JUPYTER_MAX_LIFETIME_SECONDS: + logger.info( + f"reap: max-lifetime Jupyter for {ws_id} " + f"(age={age_total:.0f}s > {JUPYTER_MAX_LIFETIME_SECONDS}s)" + ) + victims.append(ws_id) + for ws_id in victims: + try: + await stop_workspace(ws_id) + except Exception as exc: + logger.error(f"reap: failed to stop {ws_id}: {exc}") + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception(f"reap loop failed: {exc}") + + +def start_reaper() -> asyncio.Task[None]: + """Spawn the background idle reaper. Idempotent.""" + global _REAPER_TASK + if _REAPER_TASK is not None and not _REAPER_TASK.done(): + return _REAPER_TASK + _REAPER_TASK = asyncio.create_task(_reap_loop(), name="jupyter-idle-reaper") + return _REAPER_TASK + + +async def stop_reaper() -> None: + """Cancel the reaper and wait for it to finish (called from lifespan).""" + global _REAPER_TASK + task = _REAPER_TASK + _REAPER_TASK = None + if task is None or task.done(): + return + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def scan_workspaces() -> None: if not WORKSPACES_ROOT.exists(): return @@ -256,6 +507,10 @@ async def scan_workspaces() -> None: path = WORKSPACES_ROOT / entry if not path.is_dir(): return + # Skip workspace dirs that already have a live Jupyter tracked + # in this process — saves a port allocation and a noop start. + if entry in JUPYTER_PROCESSES: + return logger.info(f"Found workspace: {entry}") logger.info(f"Auto-starting Jupyter for workspace ID: '{entry}'") async with sem: @@ -264,4 +519,4 @@ async def scan_workspaces() -> None: except Exception as err: logger.error(f"Startup failed for workspace '{entry}': {err}") - await asyncio.gather(*[_start(entry) for entry in entries]) \ No newline at end of file + await asyncio.gather(*[_start(entry) for entry in entries]) diff --git a/schedule/src/schedule/orchestrator.py b/schedule/src/schedule/orchestrator.py index 754f11d..e7d45b2 100644 --- a/schedule/src/schedule/orchestrator.py +++ b/schedule/src/schedule/orchestrator.py @@ -503,13 +503,29 @@ class DispatchOrchestrator: if node_id in latest: continue parent_runs = [latest.get(parent) for parent in parents[node_id]] - parent_failed = any( + # Decide whether this node should be skipped. A node + # is only skipped when we know it can never run: + # * ``stop_all`` — the whole run was aborted on the + # first failure, so any not-yet-dispatched node is + # dropped; + # * all parents are terminal AND at least one + # failed — there is no remaining success path. + # If even one parent is still ``queued`` or + # ``running`` we keep waiting: under ``failure_policy + # == 'continue'`` a sibling might still succeed and + # the failed parent does not block that. + parents_terminal = all( item is not None and item.node_status in TERMINAL_NODE_STATES - and item.node_status != "succeeded" for item in parent_runs ) - if stop_all or parent_failed: + any_parent_failed = any( + item is not None + and item.node_status in FAILED_NODE_STATES + for item in parent_runs + ) + parents_blocked = parents_terminal and any_parent_failed + if stop_all or parents_blocked: skipped = ScheduleNodeRuns( node_run_id=new_ulid(), run_id=run.run_id, @@ -523,15 +539,24 @@ class DispatchOrchestrator: message=( "调度失败策略为 stop，未再启动" if stop_all - else "上游节点未成功，已跳过" + else "上游节点全部终止且至少一个失败，已跳过" ), ) session.add(skipped) latest[node_id] = skipped changed = True elif ( - all( - item is not None and item.node_status == "succeeded" + # Dispatch only when every parent has actually + # run to completion successfully. A None parent + # means the parent has not even been dispatched + # yet (e.g. upstream is still queued); the existing + # parents_blocked branch above handles the case + # where every parent is terminal but at least one + # failed. + len(parent_runs) > 0 + and all( + item is not None + and item.node_status == "succeeded" for item in parent_runs ) and active_count < max_concurrency @@ -554,13 +579,40 @@ class DispatchOrchestrator: for item in latest.values() ): now = utcnow() - succeeded = all( - item.node_status == "succeeded" for item in latest.values() + # Final run status depends on the schedule's + # ``failure_policy``. ``stop`` keeps the legacy rule — any + # non-success node fails the whole run. ``continue`` is + # more lenient: the run is a success when at least one + # root-level node succeeded and there is no remaining + # ``failed`` / ``cancelled`` / ``timed_out`` node that + # would have produced real artifacts had it run. Nodes + # marked ``skipped`` count as "decided to not run" and do + # not by themselves fail the run. + failure_policy = snapshot.get("failure_policy", "stop") + statuses = [item.node_status for item in latest.values()] + any_real_failure = any( + status in FAILED_NODE_STATES for status in statuses ) + any_success = any( + status == "succeeded" for status in statuses + ) + if failure_policy == "continue": + # A run with mixed success/failure/skip outcomes is + # only "succeeded" when at least one node actually ran + # to completion and nothing hit a hard failure. A + # run where every node was skipped or failed is + # itself a failure. + succeeded = any_success and not any_real_failure + else: + succeeded = all( + status == "succeeded" for status in statuses + ) run.run_status = "succeeded" if succeeded else "failed" run.error_code = None if succeeded else "SCHEDULE_NODE_FAILED" run.error_message = ( - None if succeeded else "one or more schedule nodes did not succeed" + None + if succeeded + else "one or more schedule nodes did not succeed" ) run.finished_at = now if run.started_at: