From 56b919ec4489c56c4d64320c32a0de3176cb18ba Mon Sep 17 00:00:00 2001 From: Rahul Agrawal <12agrawalrahul@gmail.com> Date: Thu, 3 Sep 2026 03:21:11 +0530 Subject: [PATCH] feat(imports): recover an import whose background job was killed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A worker killed mid-job left the Import in a running status forever: nothing clears it, the UI keeps showing an in-flight job, and the only way back was editing `status` by hand. - parse / remediate / generate clear `error` when a run starts, so a traceback from an earlier run isn't displayed against a fresh run; - `retry_import` restarts a Failed or stale-running import — re-enqueuing the parse when nothing was parsed, else handing it back to the stage before the lost job. A live job is never interrupted; - `wikify.tasks.fail_stuck_imports` (cron, every 15 minutes) fails an import that has been in a running status with no progress for 30 minutes. Jobs publish progress on every page, so a long-running pass is never touched. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_0126A9jCM6LTxnmCPH3hWbTR --- wikify/api/imports.py | 39 +++++++ wikify/hooks.py | 7 ++ wikify/jobs/generate.py | 1 + wikify/jobs/parse.py | 1 + wikify/jobs/remediate.py | 1 + wikify/tasks.py | 49 +++++++++ wikify/tests/test_job_recovery.py | 175 ++++++++++++++++++++++++++++++ 7 files changed, 273 insertions(+) create mode 100644 wikify/tasks.py create mode 100644 wikify/tests/test_job_recovery.py diff --git a/wikify/api/imports.py b/wikify/api/imports.py index d355aac..6441de1 100644 --- a/wikify/api/imports.py +++ b/wikify/api/imports.py @@ -4,7 +4,9 @@ import frappe +from wikify import tasks from wikify.engine import preview_wiki as _preview_wiki +from wikify.jobs._util import publish_progress from wikify.seed import seed_uncategorized_project #: Guard against a runaway drag-and-drop — one worker chews through these serially. @@ -89,6 +91,43 @@ def trigger_remediation(import_name: str, scope: str = "flagged") -> str: return import_name +@frappe.whitelist() +def retry_import(import_name: str) -> str: + """Restart an Import that failed, or whose worker died mid-job. + + Runs from `Failed`, or from a running status that has gone stale (no progress for + `tasks.STALE_MINUTES`) — a live job is never interrupted. The stale traceback is + cleared; an import with nothing parsed yet gets its parse job re-enqueued, and one + that already has a document is handed back to the stage before the one that died, + so remediation or wiki generation can be re-run. + """ + imp = frappe.get_doc("Wikify Import", import_name) + if imp.status not in ("Failed", *tasks.RUNNING_STATUSES): + frappe.throw(f"Nothing to retry — the import is in {imp.status}.") + if imp.status != "Failed" and not tasks.is_stale(imp): + frappe.throw(f"This import is still running ({imp.stage_label or imp.status}).") + + # The transition is persisted before it is broadcast — a realtime hiccup must not + # leave the import stuck in the status it is being rescued from. + if not imp.source_document: + imp.db_set({"status": "Queued", "error": None}) + publish_progress(import_name, 0, "Queued for retry", status="Queued") + frappe.enqueue( + "wikify.jobs.parse.run", + queue="long", + timeout=3600, + import_name=import_name, + ) + return import_name + + # The parse result (and any approved tree) is intact — hand the import back to the + # stage it was in before the lost job so the user can re-run it from the UI. + resume_status = "Graphed" if imp.status == "Generating Wiki" else "Review" + imp.db_set({"status": resume_status, "error": None}) + publish_progress(import_name, 100, f"Ready to retry from {resume_status}", status=resume_status) + return import_name + + @frappe.whitelist() def reclassify(import_name: str) -> str: """Re-tag the doc's Source Sections after manual tree edits. diff --git a/wikify/hooks.py b/wikify/hooks.py index 5cf6050..cc6cea6 100644 --- a/wikify/hooks.py +++ b/wikify/hooks.py @@ -157,6 +157,13 @@ # Scheduled Tasks # --------------- +scheduler_events = { + "cron": { + # Frequent enough that a lost worker surfaces while the user is still watching. + "*/15 * * * *": ["wikify.tasks.fail_stuck_imports"], + }, +} + # scheduler_events = { # "all": [ # "wikify.tasks.all" diff --git a/wikify/jobs/generate.py b/wikify/jobs/generate.py index be179e2..56a3824 100644 --- a/wikify/jobs/generate.py +++ b/wikify/jobs/generate.py @@ -23,6 +23,7 @@ def run( imp = frappe.get_doc("Wikify Import", import_name) try: imp.db_set("status", "Generating Wiki") + imp.db_set("error", None) publish_progress(import_name, 0, "Generating wiki", status="Generating Wiki") log(import_name, "info", "generate", f"Generating wiki for {imp.import_title}") diff --git a/wikify/jobs/parse.py b/wikify/jobs/parse.py index a1c851d..93ff14e 100644 --- a/wikify/jobs/parse.py +++ b/wikify/jobs/parse.py @@ -19,6 +19,7 @@ def run(import_name: str) -> None: imp = frappe.get_doc("Wikify Import", import_name) try: imp.db_set("status", "Parsing") + imp.db_set("error", None) imp.db_set("started_at", now_datetime()) publish_progress(import_name, 0, "Starting parse", status="Parsing") log(import_name, "info", "parse", f"Starting parse of {imp.import_title}") diff --git a/wikify/jobs/remediate.py b/wikify/jobs/remediate.py index 89ceef0..fa094fe 100644 --- a/wikify/jobs/remediate.py +++ b/wikify/jobs/remediate.py @@ -17,6 +17,7 @@ def run(import_name: str, scope: str = "flagged", instruction: str = "") -> None imp = frappe.get_doc("Wikify Import", import_name) try: imp.db_set("status", "Remediating") + imp.db_set("error", None) publish_progress(import_name, 0, f"Starting remediation ({scope})", status="Remediating") log(import_name, "info", "remediate", f"Remediating {scope} pages of {imp.import_title}") diff --git a/wikify/tasks.py b/wikify/tasks.py new file mode 100644 index 0000000..369d177 --- /dev/null +++ b/wikify/tasks.py @@ -0,0 +1,49 @@ +"""Scheduled maintenance for the Imports flow.""" + +from __future__ import annotations + +import frappe +from frappe.utils import add_to_date, get_datetime, now_datetime + +from wikify.jobs._util import log, publish_progress + +#: Statuses that mean a background job owns the import right now. +RUNNING_STATUSES = ("Queued", "Parsing", "Remediating", "Generating Wiki") + +#: A running import whose progress hasn't moved for this long has lost its worker. +#: Jobs write progress every few seconds (a 400-page remediation runs for the better +#: part of an hour but publishes on every page), so this keys off time-since-last- +#: progress and never touches a legitimately slow pass. +STALE_MINUTES = 30 + + +def stale_cutoff(): + """Progress older than this means the job is gone, not slow.""" + return add_to_date(now_datetime(), minutes=-STALE_MINUTES) + + +def is_stale(import_doc) -> bool: + """Is this import sitting in a running status with no progress written since the cutoff?""" + return import_doc.status in RUNNING_STATUSES and get_datetime(import_doc.modified) < stale_cutoff() + + +def fail_stuck_imports() -> None: + """Fail imports whose worker died mid-job. + + Nothing else moves an import out of a running status, so a killed worker (OOM, + deploy, hard timeout) leaves the UI showing an in-flight job forever with no way + back except editing `status` by hand. + """ + stuck = frappe.get_all( + "Wikify Import", + filters={"status": ("in", RUNNING_STATUSES), "modified": ("<", stale_cutoff())}, + fields=["name", "status"], + ) + for imp in stuck: + message = ( + f"No progress for {STALE_MINUTES} minutes while {imp.status} — the worker running " + f"this import was lost. Retry the import to restart it." + ) + frappe.db.set_value("Wikify Import", imp.name, {"status": "Failed", "error": message}) + publish_progress(imp.name, 100, "Worker lost", status="Failed") + log(imp.name, "error", "scheduler", message) diff --git a/wikify/tests/test_job_recovery.py b/wikify/tests/test_job_recovery.py new file mode 100644 index 0000000..c355ccd --- /dev/null +++ b/wikify/tests/test_job_recovery.py @@ -0,0 +1,175 @@ +# Copyright (c) 2026, BWH and contributors +# For license information, please see license.txt + +"""Recovering an Import whose background job died: the stale error is cleared when a +new run starts, `retry_import` restarts a failed or stuck import, and the scheduled +`fail_stuck_imports` fails one whose progress stopped. + +The progress/log helpers are patched out throughout — they `frappe.db.commit()`, which +would break the test rollback and leak rows into the site. +""" + +import tempfile +from pathlib import Path +from unittest.mock import DEFAULT, patch + +import frappe +from frappe.tests.utils import FrappeTestCase +from frappe.utils import add_to_date, now_datetime +from frappe.utils.file_manager import save_file + +from wikify import tasks +from wikify.api import imports as imports_api +from wikify.engine import store +from wikify.jobs import generate as generate_job +from wikify.jobs import remediate as remediate_job +from wikify.tests.test_parse_pipeline import _make_sample_pdf + +OLD_ERROR = "Traceback (most recent call last):\n RuntimeError: from a run in August" + + +def _without_realtime(module): + helpers = {name: DEFAULT for name in ("publish_progress", "log") if hasattr(module, name)} + return patch.multiple(module, **helpers) + + +class TestJobRecovery(FrappeTestCase): + def setUp(self): + self.project = frappe.get_doc( + { + "doctype": "Wikify Project", + "project_name": f"Job Recovery {frappe.generate_hash(length=6)}", + } + ).insert(ignore_permissions=True) + + def _import(self, status: str, error: str | None = OLD_ERROR, with_document: bool = False) -> str: + imp = frappe.get_doc( + { + "doctype": "Wikify Import", + "import_title": "Handbook", + "project": self.project.name, + "pdf": "/private/files/handbook.pdf", + "status": status, + "error": error, + "source_document": store.create_document("Handbook") if with_document else None, + } + ).insert(ignore_permissions=True) + return imp.name + + def _age(self, import_name: str, minutes: int) -> None: + frappe.db.set_value( + "Wikify Import", + import_name, + "modified", + add_to_date(now_datetime(), minutes=-minutes), + update_modified=False, + ) + + def _read(self, import_name: str) -> dict: + return frappe.db.get_value("Wikify Import", import_name, ["status", "error"], as_dict=True) + + def test_remediate_job_clears_the_error_of_an_earlier_run(self): + import_name = self._import("Review", with_document=True) + path = Path(tempfile.mkdtemp()) / "handbook.pdf" + _make_sample_pdf(str(path)) + pdf = save_file("handbook.pdf", path.read_bytes(), "Wikify Import", import_name, is_private=1) + frappe.db.set_value("Wikify Import", import_name, "pdf", pdf.file_url) + summary = {"targets": 1, "adopted": 1, "canonical_mean": 0.9, "sections": 2, "cost": 0} + + with ( + _without_realtime(remediate_job), + patch.object(remediate_job, "remediate_pdf", return_value=summary), + ): + remediate_job.run(import_name, scope="flagged") + + self.assertIsNone(self._read(import_name).error) + + def test_generate_job_clears_the_error_of_an_earlier_run(self): + import_name = self._import("Graphed", with_document=True) + result = {"space": "Handbook", "space_route": "handbook", "pages": 2, "groups": 1} + result.update({"deleted": 0, "links": 0}) + + with ( + _without_realtime(generate_job), + patch.object(generate_job, "generate_wiki", return_value=result), + ): + generate_job.run(import_name, wiki_space="Handbook") + + row = self._read(import_name) + self.assertIsNone(row.error) + self.assertEqual(row.status, "Completed") + + def test_retry_hands_a_failed_import_back_to_review(self): + import_name = self._import("Failed", with_document=True) + + with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue: + self.assertEqual(imports_api.retry_import(import_name), import_name) + enqueue.assert_not_called() + + row = self._read(import_name) + self.assertEqual(row.status, "Review") + self.assertIsNone(row.error) + + def test_retry_re_enqueues_the_parse_when_nothing_was_parsed(self): + import_name = self._import("Failed") + + with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue: + imports_api.retry_import(import_name) + + self.assertEqual(enqueue.call_args.args[0], "wikify.jobs.parse.run") + self.assertEqual(enqueue.call_args.kwargs["import_name"], import_name) + self.assertIsNone(self._read(import_name).error) + + def test_retry_recovers_an_import_whose_worker_was_lost(self): + import_name = self._import("Remediating", with_document=True) + self._age(import_name, tasks.STALE_MINUTES + 5) + + with _without_realtime(imports_api), patch.object(frappe, "enqueue"): + imports_api.retry_import(import_name) + + self.assertEqual(self._read(import_name).status, "Review") + + def test_retry_refuses_a_job_that_is_still_running(self): + import_name = self._import("Remediating", with_document=True) + + with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue: + with self.assertRaises(frappe.ValidationError): + imports_api.retry_import(import_name) + enqueue.assert_not_called() + + self.assertEqual(self._read(import_name).status, "Remediating") + + def test_retry_refuses_a_completed_import(self): + import_name = self._import("Completed", error=None, with_document=True) + + with _without_realtime(imports_api), self.assertRaises(frappe.ValidationError): + imports_api.retry_import(import_name) + + def test_a_running_import_with_no_progress_is_failed(self): + import_name = self._import("Remediating", error=None, with_document=True) + self._age(import_name, tasks.STALE_MINUTES + 5) + + with _without_realtime(tasks): + tasks.fail_stuck_imports() + + row = self._read(import_name) + self.assertEqual(row.status, "Failed") + self.assertIn("worker running this import was lost", row.error) + + def test_a_slow_job_still_reporting_progress_is_left_alone(self): + import_name = self._import("Remediating", error=None, with_document=True) + self._age(import_name, tasks.STALE_MINUTES - 5) + + with _without_realtime(tasks): + tasks.fail_stuck_imports() + + self.assertEqual(self._read(import_name).status, "Remediating") + + def test_an_import_at_rest_is_left_alone(self): + import_name = self._import("Review", error=None, with_document=True) + self._age(import_name, tasks.STALE_MINUTES * 4) + + with _without_realtime(tasks): + tasks.fail_stuck_imports() + + self.assertEqual(self._read(import_name).status, "Review")