Introduce immutable job IDs and guided migration (#447)

api/activity_log.py

@@ -15,14 +15,9 @@
 
 WINDOW_BYTES = 65536
 SEARCH_BYTES = 1024 * 1024
-_KEY = re.compile(r"^[A-Za-z0-9_.-]{1,128}$")
-_RUN = re.compile(r"^[A-Za-z0-9_.-]{8,96}$")
-
-
-def activity_log_path(directory: Path, job_key: str, run_id: str) -> Path:
-    if not _KEY.fullmatch(job_key) or not _RUN.fullmatch(run_id):
-        raise ValueError("Invalid activity log identity")
-    return directory / f"Borg-Backup_{job_key}--activity-{run_id}.log"
+def activity_log_path(directory: Path, job_id: str, run_id: str, name: str = "job") -> Path:
+    from job_runs import log_filename
+    return directory / log_filename(job_id, run_id, name)
 
 
 def open_activity_file(path: Path):
@@ -33,49 +28,48 @@ def open_activity_file(path: Path):
     return os.fdopen(fd, "rb")
 
 
-def resolve_activity_run(config: dict, job_key: str, run_id: str = "") -> tuple[Path, dict]:
-    from jobs_api import JobManager, durable_running_states, _runtime_log_dir
+def resolve_activity_run(config: dict, job_id: str, run_id: str = "") -> tuple[Path, dict]:
+    from jobs_api import JobManager, durable_running_states
     from job_control import read_control_state
-
-    if not _KEY.fullmatch(job_key):
-        raise ValueError("Invalid job key")
-    memory = JobManager.get().get_state(job_key)
-    current = memory if memory.get("running") else durable_running_states(config).get(job_key, memory)
-    run_id = run_id or str(current.get("run_id") or "")
-    path = activity_log_path(_runtime_log_dir(config), job_key, run_id)
+    from job_model import validate_job_id
+    from job_runs import find_run_status, validate_run_id
+    validate_job_id(job_id)
+    validate_run_id(run_id)
+    memory = JobManager.get().get_state(job_id)
+    current = memory if memory.get("running") else durable_running_states(config).get(job_id, memory)
     if current.get("run_id") == run_id:
-        if not current.get("file_activity"):
-            raise ValueError("File activity is not enabled for this run")
-        # The configured directory may have changed since this run started.
-        path = Path(current["log_file"])
         state = dict(current)
     else:
-        # Exact run filenames allow reconnecting after completion or a UI
-        # restart, without accepting arbitrary filesystem paths from clients.
-        state = {"running": False, "exit_code": None, "run_id": run_id}
+        state = find_run_status(config, job_id, run_id)
     control = read_control_state(run_id)
-    if control.get("job_key") == job_key:
+    if control.get("job_id") == job_id:
+        if not state:
+            state = {**control, "running": not control.get("finished")}
         state["phase"] = control.get("phase", "")
         if control.get("finished") and not state.get("running"):
             state["exit_code"] = control.get("exit_code")
     from activity_log_capture import capture_record, capture_path
-    capture = capture_record(job_key, run_id)
+    capture = capture_record(job_id, run_id)
     if capture:
-        path = capture_path(capture)
+        state.setdefault("run_id", run_id)
+        state["file_activity"] = True
+        state["log_file"] = str(capture_path(capture))
         state["capture"] = capture
         if capture.get("status") in {"saved", "failed"}:
             state["running"] = False
             state["exit_code"] = capture.get("exit_code")
         state["log_persistence_failed"] = capture.get("status") == "failed" or (capture.get("status") == "running" and not state.get("running"))
-    return path, state
+    if not state or not state.get("file_activity") or not state.get("log_file"):
+        raise FileNotFoundError("File activity is not available for this run")
+    return Path(state["log_file"]), state
 
 
 @contextmanager
-def open_activity_run(config: dict, job_key: str, run_id: str = ""):
+def open_activity_run(config: dict, job_id: str, run_id: str = ""):
     # RAM can be released between resolving its path and opening it. The saved
     # location is published first, allowing a retry without resetting cursors.
     for attempt in range(2):
-        path, state = resolve_activity_run(config, job_key, run_id)
+        path, state = resolve_activity_run(config, job_id, run_id)
         try:
             handle = open_activity_file(path)
             break
@@ -119,8 +113,8 @@ def read_window(handle, start: int, end: int, *, running: bool = False, align_st
 
 
 def get_activity_window(config: dict, qs: dict) -> dict:
-    job = (qs.get("job") or [""])[0]
-    run = (qs.get("run") or [""])[0]
+    job = (qs.get("job_id") or [""])[0]
+    run = (qs.get("run_id") or [""])[0]
     with open_activity_run(config, job, run) as (_path, state, handle):
         info = os.fstat(handle.fileno())
         size = info.st_size

api/activity_log_capture.py

@@ -20,9 +20,13 @@ def file_identity(info) -> str:
 def read_record(path: Path) -> dict:
     try:
         data = json.loads(path.read_text(encoding="utf-8"))
-        required = ("job_key", "run_id", "active_file", "retained_file", "active_file_id", "started_at")
+        required = ("job_id", "run_id", "active_file", "retained_file", "active_file_id", "started_at")
         if not isinstance(data, dict) or any(not isinstance(data.get(key), str) or not data[key] for key in required):
             return {}
+        from job_model import validate_job_id
+        from job_runs import validate_run_id
+        validate_job_id(data["job_id"])
+        validate_run_id(data["run_id"])
         return data if data.get("status") in {"running", "saved", "failed"} else {}
     except (OSError, ValueError):
         return {}
@@ -38,12 +42,12 @@ def write_record(path: Path, data: dict) -> None:
         temporary.unlink(missing_ok=True)
 
 
-def capture_record(job_key: str, run_id: str) -> dict:
+def capture_record(job_id: str, run_id: str) -> dict:
     from activity_log import activity_log_path
 
-    activity_log_path(CAPTURE_ROOT, job_key, run_id)  # Validate both components.
+    activity_log_path(CAPTURE_ROOT, job_id, run_id)  # Validate both components.
     record = read_record(CAPTURE_ROOT / run_id / "capture.json")
-    if record.get("job_key") != job_key or record.get("run_id") != run_id:
+    if record.get("job_id") != job_id or record.get("run_id") != run_id:
         return {}
     return record
 
@@ -63,17 +67,17 @@ def open_capture_file(record_path: Path):
                 raise
 
 
-def prepare_capture(job_key: str, run_id: str, destination: Path) -> tuple[Path, Path]:
+def prepare_capture(job_id: str, run_id: str, destination: Path, *, name: str = "job") -> tuple[Path, Path]:
     from activity_log import activity_log_path
 
-    retained = activity_log_path(destination, job_key, run_id)
-    active = activity_log_path(CAPTURE_ROOT / run_id, job_key, run_id)
+    retained = activity_log_path(destination, job_id, run_id, name)
+    active = activity_log_path(CAPTURE_ROOT / run_id, job_id, run_id, name)
     active.parent.mkdir(parents=True, mode=0o700)
     with os.fdopen(os.open(active, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as handle:
         identity = file_identity(os.fstat(handle.fileno()))
     record_path = active.parent / "capture.json"
     write_record(record_path, {
-        "job_key": job_key, "run_id": run_id, "status": "running",
+        "job_id": job_id, "run_id": run_id, "job_name_snapshot": name, "status": "running",
         "active_file": str(active), "retained_file": str(retained),
         "active_file_id": identity, "started_at": datetime.now(timezone.utc).isoformat(),
     })
@@ -141,7 +145,7 @@ def running_captures() -> list[dict]:
         except (TypeError, ValueError):
             continue
         rows.append({
-            "job_key": record["job_key"], "run_id": record["run_id"],
+            "job_id": record["job_id"], "run_id": record["run_id"],
             "running": True, "exit_code": None, "file_activity": True,
             "log_file": record["active_file"], "pid": record["pid"],
             "start_time": record["started_at"], "source": "activity_capture",
@@ -150,6 +154,17 @@ def running_captures() -> list[dict]:
 
 
 def supervise(record_path: Path, command: list[str]) -> int:
+    from migration_barrier import MigrationBlocked, writer_lease
+    config = {"BACKUP_SCRIPTS_DIR": os.environ.get("BORG_SCRIPT_DIR") or "/boot/config/borg-backup"}
+    try:
+        with writer_lease(config):
+            return _supervise_admitted(record_path, command)
+    except MigrationBlocked as exc:
+        print(f"Backup capture start blocked: {exc.reason}", flush=True)
+        return 2
+
+
+def _supervise_admitted(record_path: Path, command: list[str]) -> int:
     # This small process owns persistence independently of the WebUI process.
     # The runner and Borg inherit stdout/stderr pointing directly at the RAM
     # file, so browser speed cannot block them and no Python log list grows.