Compare commits

..

No commits in common. "7de9d8358956a30c327f9e5a24055940fe0417fe" and "0f30c4b9213b154bfb87ba089db33912184e75a6" have entirely different histories.

14 changed files with 9 additions and 276 deletions

View File

@ -80,9 +80,7 @@ run_group(group_id)
→ create group_run row (status=running) → create group_run row (status=running)
→ for each enabled group_member in run_order: call run_module(group_run_id=...) → for each enabled group_member in run_order: call run_module(group_run_id=...)
→ continues past individual module failures (all members run) → continues past individual module failures (all members run)
→ but a group cancel breaks the loop: remaining members never start → final status: dry_run if all dry_run, error if any errored, success otherwise
→ final status: cancelled if cancelled, else dry_run if all dry_run,
error if any errored, success otherwise
→ finish_group_run(status) → finish_group_run(status)
``` ```

View File

@ -19,7 +19,7 @@ import os
from .. import __version__, db, jrunner, repo from .. import __version__, db, jrunner, repo
from ..web import mount_web from ..web import mount_web
from .routes import connections, groups, introspect, modules, runs, system from .routes import connections, introspect, modules, runs, system
_log = logging.getLogger("pipekit.web") _log = logging.getLogger("pipekit.web")
@ -167,7 +167,6 @@ def create_app() -> FastAPI:
app.include_router(system.router) app.include_router(system.router)
app.include_router(connections.router, prefix="/api") app.include_router(connections.router, prefix="/api")
app.include_router(introspect.router, prefix="/api") app.include_router(introspect.router, prefix="/api")
app.include_router(groups.router, prefix="/api")
app.include_router(modules.router, prefix="/api") app.include_router(modules.router, prefix="/api")
app.include_router(runs.router, prefix="/api") app.include_router(runs.router, prefix="/api")
mount_web(app) mount_web(app)

View File

@ -1,84 +0,0 @@
"""Groups + group runs. The JSON counterpart to the web UI's group pages.
The web routes (web/app.py) already expose group runs, but behind session-cookie auth and
returning HTML/redirects, so they are only usable from a browser. External schedulers need
Basic auth and JSON -- hence these. Same async contract as POST /modules/{id}/run: the
group_run_id comes back immediately and the work continues in the background, so callers poll
GET /group-runs/{id} for the terminal status.
"""
from __future__ import annotations
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from ... import engine, repo
from ..auth import require_auth
router = APIRouter(tags=["groups"], dependencies=[Depends(require_auth)])
@router.get("/groups")
def list_groups() -> list[dict]:
return repo.list_groups()
@router.get("/groups/{group_id}")
def get_group(group_id: int) -> dict:
g = repo.get_group(group_id)
if g is None:
raise HTTPException(404, f"group id={group_id} not found")
return {**g, "members": repo.list_group_members(group_id)}
@router.post("/groups/{group_id}/run")
def run_group(group_id: int, background: BackgroundTasks,
dry_run: bool = False) -> dict:
"""Kick off a group run. Returns group_run_id immediately."""
g = repo.get_group(group_id)
if g is None:
raise HTTPException(404, f"group id={group_id} not found")
group_run_id = repo.create_group_run(group_id, triggered_by="api")
background.add_task(_run_in_background, group_id, group_run_id, dry_run)
return {"group_run_id": group_run_id}
def _run_in_background(group_id: int, group_run_id: int, dry_run: bool) -> None:
try:
engine.run_group(group_id, group_run_id=group_run_id, dry_run=dry_run)
except Exception as e: # noqa: BLE001
# run_group already records per-module failures; this is the group row itself
# failing (or a LockBusy escaping), which must not leave it stuck at 'running'.
repo.finish_group_run(group_run_id, status="error")
raise e from None
@router.get("/group-runs/{group_run_id}")
def get_group_run(group_run_id: int) -> dict:
gr = repo.get_group_run(group_run_id)
if gr is None:
raise HTTPException(404, f"group_run id={group_run_id} not found")
return {**gr, "runs": repo.list_runs_for_group_run(group_run_id)}
@router.post("/group-runs/{group_run_id}/cancel")
def cancel_group_run(group_run_id: int) -> dict:
"""Stop an in-flight group run.
Kills the module currently running and prevents the remaining members from
starting. The group run lands with status ``cancelled``; the engine's own
finally-blocks release the module lock and write the run_log rows, so this
handler deliberately does not touch either.
"""
gr = repo.get_group_run(group_run_id)
if gr is None:
raise HTTPException(404, f"group_run id={group_run_id} not found")
if gr["status"] != "running":
raise HTTPException(409, f"group run {group_run_id} is {gr['status']}, not running")
return engine.request_group_cancel(group_run_id)
@router.get("/groups/{group_id}/group-runs")
def list_group_runs(group_id: int, limit: int = 20) -> list[dict]:
if repo.get_group(group_id) is None:
raise HTTPException(404, f"group id={group_id} not found")
return repo.list_group_runs(group_id, limit=limit)

View File

@ -139,26 +139,6 @@ def cmd_run(args) -> int:
return 0 if outcome.status == "success" else 1 return 0 if outcome.status == "success" else 1
def cmd_run_group(args) -> int:
group = repo.get_group_by_name(args.group)
if group is None:
print(f"error: group {args.group!r} not found")
return 1
try:
outcome = engine.run_group(group["id"], dry_run=args.dry_run)
except engine.LockBusy as e:
print(f"busy: {e}")
return 75 # EX_TEMPFAIL — retryable, distinct from a real failure
tag = "DRY RUN — no jrunner calls made" if args.dry_run else ""
print(f"group_run_id={outcome.group_run_id} status={outcome.status} "
f"modules={len(outcome.module_outcomes)} {tag}".rstrip())
for o in outcome.module_outcomes:
line = f" run_id={o.run_id} status={o.status} rows={o.row_count}"
print(f"{line} error={o.error}" if o.error else line)
return 0 if outcome.status in ("success", "dry_run") else 1
def cmd_serve(args) -> int: def cmd_serve(args) -> int:
import uvicorn import uvicorn
from .api import create_app from .api import create_app
@ -371,13 +351,6 @@ def main(argv: list[str] | None = None) -> int:
help="build SQL but do not invoke jrunner") help="build SQL but do not invoke jrunner")
p_run.set_defaults(func=cmd_run) p_run.set_defaults(func=cmd_run)
p_rg = sub.add_parser("run-group",
help="run all enabled modules in a group (synchronous)")
p_rg.add_argument("group", help="group name")
p_rg.add_argument("--dry-run", action="store_true",
help="build SQL but do not invoke jrunner")
p_rg.set_defaults(func=cmd_run_group)
p_exp = sub.add_parser( p_exp = sub.add_parser(
"export", help="dump config (drivers/connections/modules/groups) to text files") "export", help="dump config (drivers/connections/modules/groups) to text files")
p_exp.add_argument("--dir", help="target dir (default <repo>/config)") p_exp.add_argument("--dir", help="target dir (default <repo>/config)")

View File

@ -48,31 +48,6 @@ def _apply_migrations(conn: sqlite3.Connection) -> None:
if "last_fired_at" not in sc_cols: if "last_fired_at" not in sc_cols:
conn.execute("ALTER TABLE schedule ADD COLUMN last_fired_at TEXT") conn.execute("ALTER TABLE schedule ADD COLUMN last_fired_at TEXT")
# group_run.status predates 'dry_run'. schema.sql carries the correct CHECK, but
# CREATE TABLE IF NOT EXISTS never re-applies it to an existing DB, so a group
# dry run raised IntegrityError in finish_group_run. CHECK constraints need a
# table rebuild (SQLite has no ALTER for them); run_log got the same treatment
# earlier. Idempotent: keyed on the constraint text itself.
gr_ddl = conn.execute(
"SELECT sql FROM sqlite_master WHERE type='table' AND name='group_run'"
).fetchone()
if gr_ddl and "dry_run" not in gr_ddl[0]:
conn.executescript("""
CREATE TABLE group_run_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
group_id INTEGER NOT NULL REFERENCES grp(id),
started_at TEXT DEFAULT (datetime('now')),
finished_at TEXT,
status TEXT NOT NULL DEFAULT 'running'
CHECK (status IN ('running','success','error','cancelled','dry_run')),
triggered_by TEXT
);
INSERT INTO group_run_new (id, group_id, started_at, finished_at, status, triggered_by)
SELECT id, group_id, started_at, finished_at, status, triggered_by FROM group_run;
DROP TABLE group_run;
ALTER TABLE group_run_new RENAME TO group_run;
""")
@contextmanager @contextmanager
def connect(db_path: Path | None = None): def connect(db_path: Path | None = None):

View File

@ -1,6 +1,5 @@
from .cancel import request_cancel from .cancel import request_cancel
from .runner import (GroupRunOutcome, LockBusy, RunOutcome, request_group_cancel, from .runner import GroupRunOutcome, LockBusy, RunOutcome, run_group, run_module
run_group, run_module)
__all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel", __all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel",
"request_group_cancel", "run_group", "run_module"] "run_group", "run_module"]

View File

@ -17,7 +17,6 @@ import threading
_lock = threading.Lock() _lock = threading.Lock()
_active: dict[int, object] = {} # run_id -> subprocess.Popen (live jrunner proc) _active: dict[int, object] = {} # run_id -> subprocess.Popen (live jrunner proc)
_cancelled: set[int] = set() # run_ids asked to stop _cancelled: set[int] = set() # run_ids asked to stop
_cancelled_groups: set[int] = set() # group_run_ids asked to stop
class RunCancelled(Exception): class RunCancelled(Exception):
@ -30,27 +29,6 @@ def arm(run_id: int) -> None:
_cancelled.discard(run_id) _cancelled.discard(run_id)
def arm_group(group_run_id: int) -> None:
"""Clear any stale group cancel flag before a group run starts."""
with _lock:
_cancelled_groups.discard(group_run_id)
def flag_group(group_run_id: int) -> None:
"""Mark a group run as cancelled so run_group stops launching members.
Only sets the flag; terminating the module that is already in flight is the
caller's job (it needs the DB to find the live run_id).
"""
with _lock:
_cancelled_groups.add(group_run_id)
def is_group_cancelled(group_run_id: int) -> bool:
with _lock:
return group_run_id in _cancelled_groups
def register(run_id: int, proc: object) -> None: def register(run_id: int, proc: object) -> None:
"""Record the live subprocess for a run so cancel can reach it.""" """Record the live subprocess for a run so cancel can reach it."""
with _lock: with _lock:

View File

@ -38,7 +38,7 @@ class RunOutcome:
@dataclass @dataclass
class GroupRunOutcome: class GroupRunOutcome:
group_run_id: int group_run_id: int
status: str # success | error | dry_run | cancelled status: str # success | error | dry_run
module_outcomes: list module_outcomes: list
@ -208,17 +208,10 @@ def run_group(group_id: int, *, dry_run: bool = False,
if group_run_id is None: if group_run_id is None:
group_run_id = repo.create_group_run(group_id, triggered_by="manual") group_run_id = repo.create_group_run(group_id, triggered_by="manual")
cancel.arm_group(group_run_id)
members = [m for m in repo.list_group_members(group_id) if m["module_enabled"]] members = [m for m in repo.list_group_members(group_id) if m["module_enabled"]]
outcomes: list[RunOutcome] = [] outcomes: list[RunOutcome] = []
stopped = False
for member in members: for member in members:
# Unlike a module failure (which the group runs past), a cancel means stop:
# don't start any remaining member.
if cancel.is_group_cancelled(group_run_id):
stopped = True
break
try: try:
outcome = run_module( outcome = run_module(
member["module_id"], member["module_id"],
@ -231,10 +224,7 @@ def run_group(group_id: int, *, dry_run: bool = False,
outcome = RunOutcome(run_id, "error", None, str(e), None, None) outcome = RunOutcome(run_id, "error", None, str(e), None, None)
outcomes.append(outcome) outcomes.append(outcome)
# 'cancelled' outranks 'error': the error is usually the killed jrunner proc. if not outcomes:
if stopped or any(o.status == "cancelled" for o in outcomes):
final = "cancelled"
elif not outcomes:
final = "dry_run" if dry_run else "success" final = "dry_run" if dry_run else "success"
elif all(o.status == "dry_run" for o in outcomes): elif all(o.status == "dry_run" for o in outcomes):
final = "dry_run" final = "dry_run"
@ -247,23 +237,6 @@ def run_group(group_id: int, *, dry_run: bool = False,
return GroupRunOutcome(group_run_id, final, outcomes) return GroupRunOutcome(group_run_id, final, outcomes)
def request_group_cancel(group_run_id: int) -> dict:
"""Stop a group run: flag it, and terminate whichever module is in flight.
Lives here rather than in cancel.py because finding the live run needs the DB.
The flag is what stops *remaining* members; the per-run cancel is what stops
the current one. Returns the runs that were signalled (may be empty if the
group is between members this instant -- the flag still lands).
"""
cancel.flag_group(group_run_id)
signalled = [
{"run_id": r["id"], "result": cancel.request_cancel(r["id"])}
for r in repo.list_runs_for_group_run(group_run_id)
if r["status"] == "running"
]
return {"group_run_id": group_run_id, "signalled": signalled}
def _run_hooks(module_id: int, *, fail_fast: bool, run_on_set: set[str]) -> str: def _run_hooks(module_id: int, *, fail_fast: bool, run_on_set: set[str]) -> str:
"""Run hooks whose ``run_on`` is in run_on_set. Returns a text log.""" """Run hooks whose ``run_on`` is in run_on_set. Returns a text log."""
hooks = [h for h in repo.list_hooks(module_id) if h["run_on"] in run_on_set] hooks = [h for h in repo.list_hooks(module_id) if h["run_on"] in run_on_set]

View File

@ -866,31 +866,13 @@ def _schedules_with_next(schedules: list[dict]) -> list[dict]:
s = dict(s) s = dict(s)
try: try:
cron = croniter(s["cron_expr"], now) cron = croniter(s["cron_expr"], now)
nxt = cron.get_next(datetime) s["next_fire_at"] = cron.get_next(datetime).strftime("%Y-%m-%d %H:%M %Z")
s["next_fire_at"] = nxt.strftime("%Y-%m-%d %H:%M")
s["next_fire_in"] = _humanize_delta((nxt - now).total_seconds())
except CroniterBadCronError: except CroniterBadCronError:
s["next_fire_at"] = "invalid expression" s["next_fire_at"] = "invalid expression"
s["next_fire_in"] = None
result.append(s) result.append(s)
return result return result
def _humanize_delta(seconds: float) -> str:
"""'in 3h 12m' style relative label for a positive second count."""
secs = int(max(seconds, 0))
days, rem = divmod(secs, 86400)
hours, rem = divmod(rem, 3600)
mins = rem // 60
if days:
return f"in {days}d {hours}h"
if hours:
return f"in {hours}h {mins}m"
if mins:
return f"in {mins}m"
return "in <1m"
def _sanitize_identifier(name: str) -> str: def _sanitize_identifier(name: str) -> str:
"""Lower-case a source column name and replace characters that aren't """Lower-case a source column name and replace characters that aren't
valid in an unquoted identifier with underscores.""" valid in an unquoted identifier with underscores."""
@ -1356,7 +1338,6 @@ def groups_index(request: Request):
g["last_run_at"] = None g["last_run_at"] = None
g["last_status"] = None g["last_status"] = None
g["last_duration_s"] = None g["last_duration_s"] = None
g["schedules"] = _schedules_with_next(repo.list_schedules_for_group(g["id"]))
return _templates.TemplateResponse( return _templates.TemplateResponse(
request, "groups.html", request, "groups.html",
_ctx(groups=groups), _ctx(groups=groups),
@ -1508,28 +1489,6 @@ def group_run_live_fragment(request: Request, group_run_id: int):
) )
@_router.post("/group-runs/{group_run_id}/cancel", response_class=HTMLResponse)
def group_run_cancel(request: Request, group_run_id: int):
"""Stop the whole group: kill the module in flight, skip the rest.
Distinct from /runs/{id}/cancel, which cancels one member and lets the group
carry on to the next.
"""
group_run = repo.get_group_run(group_run_id)
if group_run is None:
raise HTTPException(404, f"group run id={group_run_id} not found")
if group_run["status"] == "running":
engine.request_group_cancel(group_run_id)
# Re-read: the engine writes the terminal status from its own thread, so the
# fragment may still show 'running' for a tick until the poll catches up.
group_run = repo.get_group_run(group_run_id)
module_runs = repo.list_runs_for_group_run(group_run_id)
return _templates.TemplateResponse(
request, "_group_run_live.html",
_ctx(group_run=group_run, module_runs=module_runs),
)
def _save_group_schedules(form, group_id: int) -> None: def _save_group_schedules(form, group_id: int) -> None:
"""Process sched_* parallel arrays from a group form POST.""" """Process sched_* parallel arrays from a group form POST."""
from croniter import croniter as _croniter from croniter import croniter as _croniter

View File

@ -147,8 +147,6 @@ table.grid tr:hover td { background: #1c2128; }
.pill.running { color: var(--accent); } .pill.running { color: var(--accent); }
.pill.disabled { color: var(--text-muted); } .pill.disabled { color: var(--text-muted); }
.pill.warning { color: var(--warning); } .pill.warning { color: var(--warning); }
.pill.cancelled { color: var(--warning); }
.pill.dry_run { color: var(--text-muted); }
/* Group membership tags */ /* Group membership tags */
.tag { .tag {
@ -164,8 +162,6 @@ table.grid tr:hover td { background: #1c2128; }
} }
.tag:hover { color: var(--accent); } .tag:hover { color: var(--accent); }
.muted { color: var(--text-muted); }
/* Labeled key-value rows (used in detail views) */ /* Labeled key-value rows (used in detail views) */
dl.keyval { dl.keyval {
display: grid; display: grid;

View File

@ -9,12 +9,6 @@
<header> <header>
Module runs Module runs
<span style="margin-left:auto"> <span style="margin-left:auto">
{% if group_run.status == 'running' %}
<button class="btn ghost" style="margin-right:0.5rem"
hx-post="/group-runs/{{ group_run.id }}/cancel"
hx-target="#group-run-live" hx-swap="outerHTML"
hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
{% endif %}
<span class="pill {{ group_run.status }}">{{ group_run.status }}</span> <span class="pill {{ group_run.status }}">{{ group_run.status }}</span>
{% if group_run.duration_s is not none %}&nbsp;· {{ group_run.duration_s | duration }}{% endif %} {% if group_run.duration_s is not none %}&nbsp;· {{ group_run.duration_s | duration }}{% endif %}
</span> </span>

View File

@ -76,7 +76,7 @@
{% for s in schedules %} {% for s in schedules %}
<tr> <tr>
<td class="mono">{{ s.cron_expr }}</td> <td class="mono">{{ s.cron_expr }}</td>
<td class="mono">{% if s.enabled %}{{ s.next_fire_at }}{% if s.next_fire_in %} <span class="muted">({{ s.next_fire_in }})</span>{% endif %}{% else %}—{% endif %}</td> <td class="mono">{% if s.enabled %}{{ s.next_fire_at }}{% else %}—{% endif %}</td>
<td class="mono">{{ s.last_fired_at | localtime }}</td> <td class="mono">{{ s.last_fired_at | localtime }}</td>
<td> <td>
{% if s.enabled %} {% if s.enabled %}

View File

@ -31,17 +31,7 @@
hx-swap="outerHTML" hx-swap="outerHTML"
{% endif %}> {% endif %}>
<div class="panel"> <div class="panel">
<header> <header>Module runs</header>
Module runs
{% if group_run.status == 'running' %}
<span style="margin-left:auto">
<button class="btn ghost"
hx-post="/group-runs/{{ group_run.id }}/cancel"
hx-target="#group-run-live" hx-swap="outerHTML"
hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
</span>
{% endif %}
</header>
<div class="body tight"> <div class="body tight">
{% if module_runs %} {% if module_runs %}
<table class="grid"> <table class="grid">

View File

@ -18,8 +18,6 @@
<tr> <tr>
<th>name</th> <th>name</th>
<th>members</th> <th>members</th>
<th>schedule</th>
<th>next run</th>
<th>last run</th> <th>last run</th>
<th style="width:7em;text-align:right">duration</th> <th style="width:7em;text-align:right">duration</th>
<th style="width:9em">status</th> <th style="width:9em">status</th>
@ -31,21 +29,6 @@
<tr> <tr>
<td><a href="/groups/{{ g.id }}"><strong>{{ g.name }}</strong></a></td> <td><a href="/groups/{{ g.id }}"><strong>{{ g.name }}</strong></a></td>
<td class="mono">{{ g.member_count }}</td> <td class="mono">{{ g.member_count }}</td>
{% set scheds = g.schedules if g.schedules is iterable and g.schedules is not string else [] %}
<td class="mono">
{% for s in scheds %}
<div{% if not s.enabled %} class="muted"{% endif %}>{{ s.cron_expr }}{% if not s.enabled %} (disabled){% endif %}</div>
{% else %}
<span class="muted"></span>
{% endfor %}
</td>
<td class="mono">
{% for s in scheds %}
<div>{% if s.enabled %}{{ s.next_fire_at }}{% if s.next_fire_in %} <span class="muted">({{ s.next_fire_in }})</span>{% endif %}{% else %}<span class="muted"></span>{% endif %}</div>
{% else %}
<span class="muted"></span>
{% endfor %}
</td>
<td class="mono">{{ g.last_run_at | localtime }}</td> <td class="mono">{{ g.last_run_at | localtime }}</td>
<td class="mono" style="text-align:right">{{ g.last_duration_s | duration }}</td> <td class="mono" style="text-align:right">{{ g.last_duration_s | duration }}</td>
<td> <td>