diff --git a/pipekit/api/app.py b/pipekit/api/app.py index 06b6562..6bfc018 100644 --- a/pipekit/api/app.py +++ b/pipekit/api/app.py @@ -124,6 +124,18 @@ async def _lifespan(app: FastAPI): _log.warning("cleared %d stale module lock(s) at startup", n) except Exception: # noqa: BLE001 _log.exception("stale-lock cleanup failed") + # Finalise run_log / group_run rows abandoned by a process that died + # mid-run — in a fresh process nothing legitimate is running yet, so any + # `running` row is a leftover that would otherwise poll forever. + try: + runs, group_runs = repo.reconcile_abandoned_runs( + "run abandoned — pipekit process exited before completion " + "(reconciled at startup)") + if runs or group_runs: + _log.warning("reconciled %d abandoned run(s) and %d group run(s) " + "at startup", runs, group_runs) + except Exception: # noqa: BLE001 + _log.exception("abandoned-run reconciliation failed") from ..scheduler import start_scheduler start_scheduler() yield diff --git a/pipekit/db.py b/pipekit/db.py index e1a7730..dfa48ff 100644 --- a/pipekit/db.py +++ b/pipekit/db.py @@ -37,6 +37,8 @@ def _apply_migrations(conn: sqlite3.Connection) -> None: conn.execute("ALTER TABLE module ADD COLUMN columns_json TEXT") if "dest_description" not in cols: conn.execute("ALTER TABLE module ADD COLUMN dest_description TEXT") + if "description" not in cols: + conn.execute("ALTER TABLE module ADD COLUMN description TEXT") rl_cols = {r[1] for r in conn.execute("PRAGMA table_info(run_log)")} if "live_log" not in rl_cols: diff --git a/pipekit/repo.py b/pipekit/repo.py index c9071b6..784808f 100644 --- a/pipekit/repo.py +++ b/pipekit/repo.py @@ -164,18 +164,19 @@ def create_module(*, name: str, source_connection_id: int, merge_strategy: str = "full", merge_key: str | None = None, staging_table: str | None = None, columns: list[dict] | None = None, - dest_description: str | None = None) -> dict: + dest_description: str | None = None, + description: str | None = None) -> dict: staging = staging_table or f"pipekit_staging.{name}" cols_json = json.dumps(columns) if columns else None with db.connect() as c: cur = c.execute( "INSERT INTO module (name, source_connection_id, dest_connection_id, " "dest_table, staging_table, source_query, merge_strategy, merge_key, " - "columns_json, dest_description) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "columns_json, dest_description, description) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (name, source_connection_id, dest_connection_id, dest_table, staging, source_query, merge_strategy, merge_key, cols_json, - dest_description), + dest_description, description), ) return _row(c.execute( "SELECT * FROM module WHERE id=?", (cur.lastrowid,)).fetchone()) @@ -207,6 +208,7 @@ def update_module(module_id: int, *, name: str | None = None, merge_strategy: str | None = None, merge_key: str | None = None, dest_description: str | None = None, + description: str | None = None, enabled: int | None = None) -> dict | None: fields: list[str] = [] values: list = [] @@ -219,6 +221,7 @@ def update_module(module_id: int, *, name: str | None = None, ("merge_strategy", merge_strategy), ("merge_key", merge_key), ("dest_description", dest_description), + ("description", description), ("enabled", enabled)): if val is not None: fields.append(f"{col}=?") @@ -426,6 +429,31 @@ def clear_stale_locks(max_age_hours: int = 24, live_pids: set[int] | None = None return cleared +def reconcile_abandoned_runs(reason: str) -> tuple[int, int]: + """Mark run_log / group_run rows still ``running`` as ``error``. + + Meant to be called once at process startup, before any run can begin: a + hard kill (or ``systemctl stop`` mid-run) skips the engine's finally, so + the run_log row (and any owning group_run) stays ``running`` forever — the + live-log poller would spin and the runs list would show a phantom run. In a + fresh process no run has started yet, so every ``running`` row is a leftover + and is safe to finalise. Returns (runs_reconciled, group_runs_reconciled). + """ + with db.connect() as c: + runs = c.execute( + "UPDATE run_log SET finished_at=datetime('now'), status='error', " + "error=CASE WHEN error IS NULL OR error='' THEN ? " + "ELSE error || char(10) || ? END " + "WHERE status='running'", + (reason, reason), + ).rowcount + group_runs = c.execute( + "UPDATE group_run SET finished_at=datetime('now'), status='error' " + "WHERE status='running'", + ).rowcount + return runs, group_runs + + # --------------------------------------------------------------------------- # Run log # --------------------------------------------------------------------------- diff --git a/pipekit/schema.sql b/pipekit/schema.sql index c3b3ecb..c651796 100644 --- a/pipekit/schema.sql +++ b/pipekit/schema.sql @@ -29,6 +29,7 @@ CREATE TABLE IF NOT EXISTS connection ( CREATE TABLE IF NOT EXISTS module ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, + description TEXT, -- free-text note about the module, shown in the UI source_connection_id INTEGER NOT NULL REFERENCES connection(id), dest_connection_id INTEGER NOT NULL REFERENCES connection(id), dest_table TEXT NOT NULL, diff --git a/pipekit/web/app.py b/pipekit/web/app.py index c132383..dbcfc2f 100644 --- a/pipekit/web/app.py +++ b/pipekit/web/app.py @@ -147,10 +147,12 @@ def home(request: Request): m["last_run_at"] = last["started_at"] m["last_status"] = last["status"] m["last_row_count"] = last["row_count"] + m["last_duration_s"] = last["duration_s"] else: m["last_run_at"] = None m["last_status"] = None m["last_row_count"] = None + m["last_duration_s"] = None m["groups"] = groups_by_module.get(m["id"], []) # group by source connection @@ -287,6 +289,7 @@ async def module_update(request: Request, module_id: int): merge_strategy=merge_strategy, merge_key=(form.get("merge_key") or "").strip() or None, dest_description=new_description, + description=(form.get("description") or "").strip(), enabled=1 if form.get("enabled") == "1" else 0, ) _save_inline_watermarks(form, module_id) @@ -647,6 +650,7 @@ async def wizard_create(request: Request): merge_key = (form.get("merge_key") or "").strip() or None staging_table = (form.get("staging_table") or "").strip() or None dest_description = (form.get("dest_description") or "").strip() or None + description = (form.get("description") or "").strip() or None picked = form.getlist("col") if repo.get_module_by_name(module_name) is not None: @@ -809,6 +813,7 @@ async def wizard_create(request: Request): staging_table=staging_table, columns=chosen, dest_description=dest_description, + description=description, ) _save_inline_watermarks(form, module["id"]) return RedirectResponse(url=f"/modules/{module['id']}", status_code=303) diff --git a/pipekit/web/static/style.css b/pipekit/web/static/style.css index 3e2b550..7fdd93e 100644 --- a/pipekit/web/static/style.css +++ b/pipekit/web/static/style.css @@ -66,6 +66,7 @@ header.topbar nav a:hover { header.topbar .right { margin-left: auto; color: var(--text-muted); font-size: 12px; } main { + max-width: 1400px; margin: 1rem auto; padding: 0 1.2rem; } diff --git a/pipekit/web/templates/_module_status_pill.html b/pipekit/web/templates/_module_status_pill.html index 9697935..0a687ac 100644 --- a/pipekit/web/templates/_module_status_pill.html +++ b/pipekit/web/templates/_module_status_pill.html @@ -1,6 +1,7 @@ -{# Partial: status cell for one module row on the index page. - Swaps itself (outerHTML) every 3s while running; stops when idle. #} - (lives inside the compact "last run" cell). Swaps itself + (outerHTML) every 3s while running; stops when idle. #} +never ran {% endif %} - + diff --git a/pipekit/web/templates/module_detail.html b/pipekit/web/templates/module_detail.html index 7ae8500..0708b40 100644 --- a/pipekit/web/templates/module_detail.html +++ b/pipekit/web/templates/module_detail.html @@ -36,6 +36,9 @@
+ {% if module.description %} +
description
{{ module.description }}
+ {% endif %}
source
{{ source_conn.name }} ({{ source_conn.jdbc_url }})
destination
{{ dest_conn.name }} ({{ dest_conn.jdbc_url }})
dest table
{{ module.dest_table }}
diff --git a/pipekit/web/templates/module_form.html b/pipekit/web/templates/module_form.html index 9972fb2..df27639 100644 --- a/pipekit/web/templates/module_form.html +++ b/pipekit/web/templates/module_form.html @@ -26,6 +26,12 @@ must be unique; also used as the default staging table suffix + +