From 4f83e8d991ec156895921a4a2ba60af4e0efbcb1 Mon Sep 17 00:00:00 2001 From: Paul Trowbridge Date: Fri, 10 Jul 2026 04:06:24 -0400 Subject: [PATCH] feat: cancel an in-flight module run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds run cancellation, which had no implementation despite `cancelled` being a declared run status. Mechanism: API/web/scheduler runs execute as background tasks in the service process, so `running_pid`'s PID is the service itself — killing it is not an option. Instead, engine/cancel.py keeps an in-process registry of run_id -> live jrunner Popen. The cancel handler (same process) terminates that subprocess; migrate then fails and the engine records the run as `cancelled` (not `error`) and releases the lock in its finally. Checkpoints before migrate and before merge honor a cancel flag set between phases. - engine/cancel.py: registry (arm/register/unregister/is_cancelled/request_cancel) + RunCancelled - jrunner.migrate: optional proc_hook to expose the live Popen (no engine dependency) - runner: arm, register via proc_hook, two checkpoints, classify cancelled, unregister - POST /api/modules/{id}/cancel (extracts run_id from the lock) and POST /runs/{id}/cancel (web) - cancel button on run detail + live fragment, shown while running - api/app lifespan: clear stale locks (dead PID / >24h) at startup via the existing repo.clear_stale_locks, so a hard service kill no longer wedges `running=1` Verified: real subprocess terminate -> status=cancelled, lock released; endpoints return 409 (not running) / 404 (unknown) / 200 (fragment). Co-Authored-By: Claude Opus 4.8 --- pipekit/api/app.py | 22 +++++++++- pipekit/api/routes/modules.py | 22 ++++++++++ pipekit/engine/__init__.py | 4 +- pipekit/engine/cancel.py | 63 +++++++++++++++++++++++++++ pipekit/engine/runner.py | 26 +++++++++-- pipekit/jrunner.py | 6 +++ pipekit/web/app.py | 14 ++++++ pipekit/web/templates/_run_live.html | 6 +++ pipekit/web/templates/run_detail.html | 6 +++ 9 files changed, 164 insertions(+), 5 deletions(-) create mode 100644 pipekit/engine/cancel.py diff --git a/pipekit/api/app.py b/pipekit/api/app.py index e119b47..06b6562 100644 --- a/pipekit/api/app.py +++ b/pipekit/api/app.py @@ -15,7 +15,9 @@ from fastapi import FastAPI, Request from fastapi.exception_handlers import http_exception_handler from starlette.exceptions import HTTPException as StarletteHTTPException -from .. import __version__, db, jrunner +import os + +from .. import __version__, db, jrunner, repo from ..web import mount_web from .routes import connections, introspect, modules, runs, system @@ -102,8 +104,26 @@ class _RequestBodyCapture: await self.app(scope, replay, send) +def _live_pids() -> set[int]: + """PIDs currently alive on this host (Linux /proc). Empty set elsewhere.""" + try: + return {int(p) for p in os.listdir("/proc") if p.isdigit()} + except OSError: + return set() + + @asynccontextmanager async def _lifespan(app: FastAPI): + # Release module locks left behind by a previous process that died mid-run + # (a hard kill skips the engine's finally, so `running` stays 1 and future + # runs would refuse with "already running"). Clears locks held by a dead PID + # or older than 24h. + try: + n = repo.clear_stale_locks(live_pids=_live_pids()) + if n: + _log.warning("cleared %d stale module lock(s) at startup", n) + except Exception: # noqa: BLE001 + _log.exception("stale-lock cleanup failed") from ..scheduler import start_scheduler start_scheduler() yield diff --git a/pipekit/api/routes/modules.py b/pipekit/api/routes/modules.py index a9b7a60..736876e 100644 --- a/pipekit/api/routes/modules.py +++ b/pipekit/api/routes/modules.py @@ -93,6 +93,28 @@ def _run_in_background(module_id: int, run_id: int, dry_run: bool) -> None: repo.finish_run(run_id, status="error", error=str(e)) +@router.post("/modules/{module_id}/cancel") +def cancel_module_run(module_id: int) -> dict: + """Request cancellation of a module's in-flight run. + + Kills the live jrunner subprocess if one is active in this process; either + way the cancel flag is honored at the next engine checkpoint. The run lands + with status ``cancelled`` and the lock is released by the engine. + """ + m = repo.get_module(module_id) + if m is None: + raise HTTPException(404, f"module id={module_id} not found") + if not m["running"]: + raise HTTPException(409, f"module {m['name']!r} is not running") + # running_pid is ":" (see engine.runner) + _, _, run_id_str = (m["running_pid"] or "").partition(":") + if not run_id_str: + raise HTTPException(409, "no run_id recorded for the active lock") + run_id = int(run_id_str) + result = engine.request_cancel(run_id) + return {"module_id": module_id, "run_id": run_id, "status": result} + + # --------------------------------------------------------------------------- # Watermarks — scoped to a module # --------------------------------------------------------------------------- diff --git a/pipekit/engine/__init__.py b/pipekit/engine/__init__.py index 0882ccc..c63cd53 100644 --- a/pipekit/engine/__init__.py +++ b/pipekit/engine/__init__.py @@ -1,3 +1,5 @@ +from .cancel import request_cancel from .runner import GroupRunOutcome, LockBusy, RunOutcome, run_group, run_module -__all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "run_group", "run_module"] +__all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel", + "run_group", "run_module"] diff --git a/pipekit/engine/cancel.py b/pipekit/engine/cancel.py new file mode 100644 index 0000000..b6f0c21 --- /dev/null +++ b/pipekit/engine/cancel.py @@ -0,0 +1,63 @@ +"""In-process cancellation registry for running module jobs. + +API/web/scheduler runs all execute as background tasks *inside the pipekit +service process*, and so does the cancel HTTP handler — so no IPC or OS signals +are needed. This module holds a small in-memory map of ``run_id -> live jrunner +Popen`` plus a set of run_ids that have been asked to stop. The cancel endpoint +looks up the live subprocess and terminates it; the engine also checks the flag +at phase boundaries so a cancel lands promptly even between subprocess calls. + +Scope: this only reaches runs in *this* process. A ``pipekit run`` CLI run is a +separate process and isn't in the registry — cancel it with Ctrl-C. +""" +from __future__ import annotations + +import threading + +_lock = threading.Lock() +_active: dict[int, object] = {} # run_id -> subprocess.Popen (live jrunner proc) +_cancelled: set[int] = set() # run_ids asked to stop + + +class RunCancelled(Exception): + """Raised inside run_module when a cancel was requested at a checkpoint.""" + + +def arm(run_id: int) -> None: + """Clear any stale cancel flag before a run starts.""" + with _lock: + _cancelled.discard(run_id) + + +def register(run_id: int, proc: object) -> None: + """Record the live subprocess for a run so cancel can reach it.""" + with _lock: + _active[run_id] = proc + + +def unregister(run_id: int) -> None: + """Drop a run's subprocess handle (proc finished or run ended).""" + with _lock: + _active.pop(run_id, None) + + +def is_cancelled(run_id: int) -> bool: + with _lock: + return run_id in _cancelled + + +def request_cancel(run_id: int) -> str: + """Flag a run for cancellation and terminate its live subprocess if any. + + Returns ``"signalled"`` if a live subprocess was terminated, or + ``"flagged"`` if none was active this instant (the flag is honored at the + next engine checkpoint — e.g. the run is between phases or in another + process). + """ + with _lock: + _cancelled.add(run_id) + proc = _active.get(run_id) + if proc is not None and proc.poll() is None: # type: ignore[attr-defined] + proc.terminate() # type: ignore[attr-defined] + return "signalled" + return "flagged" diff --git a/pipekit/engine/runner.py b/pipekit/engine/runner.py index ebda5dd..413bd26 100644 --- a/pipekit/engine/runner.py +++ b/pipekit/engine/runner.py @@ -20,7 +20,8 @@ import traceback from dataclasses import dataclass from .. import drivers, jrunner, repo -from . import merge, watermark +from . import cancel, merge, watermark +from .cancel import RunCancelled @dataclass @@ -66,6 +67,8 @@ def run_module(module_id: int, *, group_run_id: int | None = None, repo.finish_run(run_id, status="error", error="already running") raise LockBusy(f"module {module['name']!r} is already running") + cancel.arm(run_id) + resolved_sql: str | None = None merge_sql: str | None = None row_count: int | None = None @@ -118,6 +121,9 @@ def run_module(module_id: int, *, group_run_id: int | None = None, jrunner.run_dest_sql(dest_conn, drop_sql) jrunner.run_dest_sql(dest_conn, like_sql) + if cancel.is_cancelled(run_id): + raise RunCancelled() + # 5. migrate source → staging. jrunner does its own `DELETE FROM staging` # before loading, so we don't need a separate TRUNCATE. migrate_result = jrunner.migrate( @@ -125,11 +131,15 @@ def run_module(module_id: int, *, group_run_id: int | None = None, sql=resolved_sql, dest_table=module["staging_table"], clear=False, progress_cb=lambda line: repo.append_run_live_log(run_id, line), + proc_hook=lambda p: cancel.register(run_id, p), ) row_count = migrate_result.row_count repo.log_run_output(run_id, jrunner_stdout=migrate_result.stdout, jrunner_stderr=migrate_result.stderr) + if cancel.is_cancelled(run_id): + raise RunCancelled() + # 7. merge jrunner.run_dest_sql(dest_conn, merge_sql) @@ -142,10 +152,19 @@ def run_module(module_id: int, *, group_run_id: int | None = None, return RunOutcome(run_id, status, row_count, None, resolved_sql, merge_sql) except Exception as e: # noqa: BLE001 - error = f"{type(e).__name__}: {e}\n{traceback.format_exc()}" + # A cancel request kills the jrunner subprocess, which surfaces as a + # JrunnerError — treat any failure under a set cancel flag (or an + # explicit checkpoint raise) as 'cancelled', not 'error'. + cancelled = isinstance(e, RunCancelled) or cancel.is_cancelled(run_id) if isinstance(e, jrunner.JrunnerError): repo.log_run_output(run_id, jrunner_stdout=e.stdout, jrunner_stderr=e.stderr) + if cancelled: + status = "cancelled" + error = "cancelled by user" + else: + status = "error" + error = f"{type(e).__name__}: {e}\n{traceback.format_exc()}" # Failure-path hooks, if any. Never let these mask the real error. try: hook_log = _run_hooks(module_id, fail_fast=False, @@ -154,9 +173,10 @@ def run_module(module_id: int, *, group_run_id: int | None = None, repo.log_run_output(run_id, hook_log=hook_log) except Exception: # noqa: BLE001, S110 pass - return RunOutcome(run_id, "error", row_count, error, resolved_sql, merge_sql) + return RunOutcome(run_id, status, row_count, error, resolved_sql, merge_sql) finally: + cancel.unregister(run_id) repo.finish_run(run_id, status=status, row_count=row_count, error=error) repo.release_module_lock(module_id) diff --git a/pipekit/jrunner.py b/pipekit/jrunner.py index 441efae..b72d91c 100644 --- a/pipekit/jrunner.py +++ b/pipekit/jrunner.py @@ -147,9 +147,13 @@ def migrate( trim: bool = True, timeout: int = 3600, progress_cb: "Callable[[str], None] | None" = None, + proc_hook: "Callable[[object], None] | None" = None, ) -> MigrateResult: """Stream `sql` results from source into `dest_table` via jrunner migration mode. + If ``proc_hook`` is provided it is called with the live ``Popen`` right after + spawn, letting a caller (the engine) register it for cancellation. + If ``progress_cb`` is provided it is called with each non-empty stdout line as jrunner produces it, enabling live progress reporting. """ @@ -180,6 +184,8 @@ def migrate( proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=_subprocess_env()) + if proc_hook is not None: + proc_hook(proc) stderr_buf: list[str] = [] diff --git a/pipekit/web/app.py b/pipekit/web/app.py index e8f227c..c132383 100644 --- a/pipekit/web/app.py +++ b/pipekit/web/app.py @@ -1071,6 +1071,20 @@ def run_live_fragment(request: Request, run_id: int): ) +@_router.post("/runs/{run_id}/cancel", response_class=HTMLResponse) +def run_cancel(request: Request, run_id: int): + run = repo.get_run(run_id) + if run is None: + raise HTTPException(404, f"run id={run_id} not found") + if run["status"] == "running": + engine.request_cancel(run_id) + # Return the (soon-to-update) live fragment so the button/status refresh. + run = repo.get_run(run_id) + module = repo.get_module(run["module_id"]) + run["module_name"] = module["name"] if module else "?" + return _templates.TemplateResponse(request, "_run_live.html", _ctx(run=run)) + + # --------------------------------------------------------------------------- # Watermarks — add/edit/delete forms on module detail # --------------------------------------------------------------------------- diff --git a/pipekit/web/templates/_run_live.html b/pipekit/web/templates/_run_live.html index 95cfabd..177336e 100644 --- a/pipekit/web/templates/_run_live.html +++ b/pipekit/web/templates/_run_live.html @@ -9,6 +9,12 @@
Progress + {% if run.status == 'running' %} + + {% endif %} {{ run.status }} {% if run.row_count is not none %} · {{ run.row_count }} rows{% endif %} diff --git a/pipekit/web/templates/run_detail.html b/pipekit/web/templates/run_detail.html index 5c64b1f..f020304 100644 --- a/pipekit/web/templates/run_detail.html +++ b/pipekit/web/templates/run_detail.html @@ -48,6 +48,12 @@
Progress + {% if run.status == 'running' %} + + {% endif %} {{ run.status }} {% if run.row_count is not none %} · {{ run.row_count }} rows{% endif %}