Use permanent job IDs, explicit settings and compatible SSH with fewer idle writes (#486, #495, #496, #497)

api/activity_log.py

@@ -7,10 +7,12 @@
 from __future__ import annotations
 
 import codecs
+import json
 import os
 import re
 import stat
 from contextlib import contextmanager
+from functools import lru_cache
 from pathlib import Path
 
 WINDOW_BYTES = 65536
@@ -19,9 +21,12 @@
 _RUN = re.compile(r"^[A-Za-z0-9_.-]{8,96}$")
 
 
-def activity_log_path(directory: Path, job_key: str, run_id: str) -> Path:
+def activity_log_path(directory: Path, job_key: str, run_id: str, *, job_name: str = "", location: str = "") -> Path:
     if not _KEY.fullmatch(job_key) or not _RUN.fullmatch(run_id):
         raise ValueError("Invalid activity log identity")
+    if job_name or location:
+        from job_identity import job_log_filename, job_run_date_tag
+        return directory / job_log_filename(job_name, location, job_key, job_run_date_tag(run_id))
     return directory / f"Borg-Backup_{job_key}--activity-{run_id}.log"
 
 
@@ -33,6 +38,53 @@ def open_activity_file(path: Path):
     return os.fdopen(fd, "rb")
 
 
+def _saved_activity_run(config: dict, job_key: str, run_id: str) -> tuple[Path, dict] | None:
+    """Resolve a saved run by its status metadata after RAM state is gone."""
+    from jobs_api import _runtime_log_dir
+    status_dir = str(config.get("STATUS_DIR") or "")
+    archive_dir = str(config.get("STATUS_ARCHIVE_DIR") or (Path(status_dir) / "archive" if status_dir else ""))
+    try:
+        return _find_saved_activity_run(status_dir, archive_dir, _runtime_log_dir(config), job_key, run_id)
+    except FileNotFoundError:
+        return None
+
+
+@lru_cache(maxsize=128)
+def _find_saved_activity_run(status_dir: str, archive_dir: str, log_dir: Path,
+                             job_key: str, run_id: str) -> tuple[Path, dict]:
+    # Completed run mappings are immutable. Cache only successful lookups so
+    # each bounded log window does not reread all historical status files.
+    matches = {}
+    for directory in {status_dir, archive_dir} - {""}:
+        for status_file in Path(directory).glob(f"*_{job_key}.status"):
+            try:
+                data = json.loads(status_file.read_text(encoding="utf-8"))
+            except (OSError, ValueError):
+                continue
+            if not isinstance(data, dict) or data.get("job_id") != job_key or data.get("run_id") != run_id:
+                continue
+            if data.get("file_activity") is not True or not isinstance(data.get("log_file"), str) or not data["log_file"]:
+                continue
+            matches[Path(data["log_file"])] = data
+    if not matches:
+        # A runner can fail before writing status (for example while loading
+        # its job). Its complete retained log must still reopen after reboot.
+        from job_identity import job_log_paths
+        marker = f"INFO File activity run: job_id={job_key} run_id={run_id}\n".encode()
+        for candidate in job_log_paths(log_dir, job_key):
+            try:
+                with open_activity_file(candidate) as handle:
+                    if handle.readline(256) == marker:
+                        matches[candidate] = {}
+            except (OSError, ValueError):
+                continue
+    if len(matches) > 1:
+        raise ValueError("Multiple logs found for this job run")
+    if not matches:
+        raise FileNotFoundError("No saved file-activity log found for this job run")
+    return next(iter(matches.items()))
+
+
 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
     from job_control import read_control_state
@@ -53,6 +105,18 @@ def resolve_activity_run(config: dict, job_key: str, run_id: str = "") -> tuple[
         # 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}
+        from job_identity import job_log_paths
+        matches = [candidate for candidate in job_log_paths(path.parent, job_key)
+                   if candidate.name.endswith(f"--activity-{run_id}.log")]
+        if len(matches) > 1:
+            raise ValueError("Multiple logs found for this job run")
+        if matches:
+            path = matches[0]
+        else:
+            saved = _saved_activity_run(config, job_key, run_id)
+            if saved:
+                path, status = saved
+                state["exit_code"] = status.get("exit_code")
     control = read_control_state(run_id)
     if control.get("job_key") == job_key:
         state["phase"] = control.get("phase", "")

api/activity_log_capture.py

@@ -63,10 +63,10 @@ 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_key: str, run_id: str, destination: Path, *, job_name: str = "", location: str = "") -> tuple[Path, Path]:
     from activity_log import activity_log_path
 
-    retained = activity_log_path(destination, job_key, run_id)
+    retained = activity_log_path(destination, job_key, run_id, job_name=job_name, location=location)
     active = activity_log_path(CAPTURE_ROOT / run_id, job_key, run_id)
     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:
@@ -157,6 +157,10 @@ def supervise(record_path: Path, command: list[str]) -> int:
     record["pid"] = os.getpid()
     record["process_start"] = process_token(os.getpid())
     write_record(record_path, record)
+    if Path(record["retained_file"]).name.startswith("BBUI-"):
+        # The visible filename is the same as for a normal run. Keep the exact
+        # run identity in the log even if startup fails before status is saved.
+        print(f"INFO File activity run: job_id={record['job_key']} run_id={record['run_id']}", flush=True)
     try:
         process = subprocess.Popen(command)
         code = process.wait()

api/archive_browser.py

@@ -17,6 +17,13 @@
 _CACHE_LOCK = threading.Lock()
 
 
+class ArchiveNotFoundError(ValueError):
+    """The archive selection no longer exists in the current repository."""
+
+    api_code = "restore_archive_unavailable"
+    api_status = 404
+
+
 def _prune_expired_cache_entries(now: float) -> None:
     expired = [key for key, value in _CACHE.items() if float(value.get("expires", 0)) <= now]
     for key in expired:
@@ -46,6 +53,10 @@ def build_archive_index(repo: str, archive: str, env: dict[str, str]) -> dict[st
     except subprocess.TimeoutExpired as exc:
         raise TimeoutError("borg archive listing timed out") from exc
     if result.returncode != 0:
+        # Borg's C-locale error identifies an absent archive, not an unavailable
+        # repository or another list failure. Keep the original detail for logs.
+        if f"Archive {archive} does not exist" in result.stderr.splitlines():
+            raise ArchiveNotFoundError(f"borg list failed: {result.stderr.strip()}")
         raise RuntimeError(f"borg list failed: {result.stderr.strip()}")
 
     index: dict[str, dict[str, dict[str, Any]]] = {}

api/archive_prefix.py

@@ -4,7 +4,42 @@
 from typing import Iterable
 
 
-_ARCHIVE_PREFIX_RX = re.compile(r"^[A-Za-z0-9_.-]+-backup$")
+_ARCHIVE_PREFIX_RX = re.compile(r"^[A-Za-z0-9_.-]+$")
+
+
+def validate_archive_prefix(value: object) -> str:
+    prefix = str(value or "").strip()
+    if not prefix or not _ARCHIVE_PREFIX_RX.fullmatch(prefix):
+        raise ValueError("Archive prefix may contain only letters, digits, dots, underscores and hyphens")
+    return prefix
+
+
+def archive_prefix_from_metadata(meta: dict) -> str:
+    return validate_archive_prefix(meta.get("archive_prefix"))
+
+
+def job_archive_prefixes(meta: dict) -> list[str]:
+    return list(dict.fromkeys([
+        archive_prefix_from_metadata(meta),
+        *(validate_archive_prefix(value) for value in meta.get("archive_prefixes", [])),
+    ]))
+
+
+def validate_prefix_ownership(candidate: dict, other_jobs: Iterable[dict]) -> None:
+    """A Borg '<prefix>-*' selection belongs to one job per repository."""
+    prefixes = job_archive_prefixes(candidate)
+    for other in other_jobs:
+        if other.get("job_id") == candidate.get("job_id"):
+            continue
+        if other.get("repository_key") != candidate.get("repository_key"):
+            continue
+        for prefix in prefixes:
+            for owned in job_archive_prefixes(other):
+                if prefix == owned or prefix.startswith(owned + "-") or owned.startswith(prefix + "-"):
+                    raise ValueError(
+                        f"Archive prefix '{prefix}' overlaps with '{owned}' of job "
+                        f"'{other.get('name') or other.get('job_id')}' in the selected repository"
+                    )
 
 
 def archive_prefix_from_backup_type(backup_type: str) -> str:

api/borg_ssh.py

@@ -25,7 +25,7 @@
     ("LogLevel", "ERROR"),
     ("WarnWeakCrypto", "no"),
 )
-_MANAGED_OPTION_NAMES = {name.lower() for name, _ in _SSH_OPTIONS}
+_MANAGED_OPTION_NAMES = {name.lower() for name, _ in _SSH_OPTIONS} | {"ignoreunknown"}
 _SSH_INTERRUPTION_MARKERS = (
     "connection reset by peer",
     "broken pipe",
@@ -38,7 +38,8 @@
 
 
 def _option_name(value: str) -> str:
-    return str(value or "").split("=", 1)[0].strip().lower()
+    parts = str(value or "").replace("=", " ", 1).split(maxsplit=1)
+    return parts[0].lower() if parts else ""
 
 
 def build_borg_rsh(existing: str = "", identity_file: str = "") -> str:
@@ -52,21 +53,32 @@ def build_borg_rsh(existing: str = "", identity_file: str = "") -> str:
 
     identity = str(identity_file or "").strip()
     cleaned: list[str] = []
+    ignore_unknown_seen = False
     index = 0
     while index < len(tokens):
         token = tokens[index]
+        option = None
+        option_length = 1
         if token == "-o" and index + 1 < len(tokens):
-            value = tokens[index + 1]
-            if _option_name(value) in _MANAGED_OPTION_NAMES:
-                index += 2
-                continue
-            cleaned.extend((token, value))
-            index += 2
+            option = tokens[index + 1]
+            option_length = 2
+        elif token.startswith("-o") and len(token) > 2:
+            option = token[2:]
+        if option is not None:
+            name = _option_name(option)
+            if name == "ignoreunknown" and not ignore_unknown_seen:
+                # SSH uses the first list. Extend it in place so any custom
+                # options following it still have their original exemption.
+                parts = option.replace("=", " ", 1).split(maxsplit=1)
+                patterns = parts[1].split(",") if len(parts) > 1 else []
+                if "warnweakcrypto" not in {pattern.lower() for pattern in patterns}:
+                    patterns.append("WarnWeakCrypto")
+                cleaned.extend(("-o", "IgnoreUnknown=" + ",".join(patterns)))
+                ignore_unknown_seen = True
+            elif name not in _MANAGED_OPTION_NAMES:
+                cleaned.extend(tokens[index:index + option_length])
+            index += option_length
             continue
-        if token.startswith("-o") and len(token) > 2:
-            if _option_name(token[2:]) in _MANAGED_OPTION_NAMES:
-                index += 1
-                continue
         if identity and token == "-i" and index + 1 < len(tokens):
             index += 2
             continue
@@ -76,6 +88,9 @@ def build_borg_rsh(existing: str = "", identity_file: str = "") -> str:
         cleaned.append(token)
         index += 1
 
+    # Older clients must see this exemption before WarnWeakCrypto is parsed.
+    if not ignore_unknown_seen:
+        cleaned.extend(("-o", "IgnoreUnknown=WarnWeakCrypto"))
     if identity:
         cleaned.extend(("-i", identity))
     for name, value in _SSH_OPTIONS:

api/check_api.py

@@ -208,7 +208,9 @@ def _repository_command(
         if not selected_job_key:
             raise ValueError("Prune requires a backup job with a retention policy")
         retention = self._job_retention(config, selected_job_key)
-        archive_prefix = _archive_prefix_from_job_key(selected_job_key)
+        from archive_prefix import archive_prefix_from_metadata
+        from repository_context import load_job_metadata
+        archive_prefix = archive_prefix_from_metadata(load_job_metadata(config, selected_job_key))
         cmd = [
             "borg", "prune", "--lock-wait", self._LOCK_WAIT_SECONDS,
             "--list", "--progress",

api/config_api.py

@@ -795,7 +795,6 @@ def get_settings_data(ui_config: dict, include_storagebox_setup: bool = True) ->
             "RESTORE_TEST_LEVEL":         conf.get("RESTORE_TEST_LEVEL", "2"),
             "RESTORE_TEST_INTERVAL_DAYS": conf.get("RESTORE_TEST_INTERVAL_DAYS", "30"),
             "RESTORE_TEST_LOCATION":      conf.get("RESTORE_TEST_LOCATION", "local"),
-            "RESTORE_TEST_FORCE_CHUNK_TYPES": conf.get("RESTORE_TEST_FORCE_CHUNK_TYPES", "vms,photos"),
             "RESTORE_TEST_FULL_DRYRUN_MAX_ARCHIVE_GB": conf.get("RESTORE_TEST_FULL_DRYRUN_MAX_ARCHIVE_GB", "500"),
             "RESTORE_TEST_MIN_COVERAGE": conf.get("RESTORE_TEST_MIN_COVERAGE", "5"),
             "RESTORE_TEST_MAX_ENTRIES": conf.get("RESTORE_TEST_MAX_ENTRIES", "1000"),
@@ -1088,7 +1087,12 @@ def _is_required_storage_mount_available(mount_path: Path) -> bool:
             return False
 
 
-def ensure_data_dirs(global_data_dir: str) -> dict:
+def ensure_data_dirs(global_data_dir: str, *, read_only: bool = False) -> dict:
+    """Check storage, with a read-only mode for routine status requests.
+
+    Routine status reads inspect existing directories and access permissions.
+    Setup and runtime writers retain the actual write probe by default.
+    """
     root = (global_data_dir or "").strip()
     if not root:
         raise ValueError("GLOBAL_DATA_DIR is not set")
@@ -1103,8 +1107,16 @@ def ensure_data_dirs(global_data_dir: str) -> dict:
     created = []
     for key in ("base", "logs", "status", "restore_status", "cache", "remotes"):
         p = Path(paths[key])
-        p.mkdir(parents=True, exist_ok=True)
-        created.append(str(p))
+        if read_only:
+            if not p.is_dir():
+                raise RuntimeError(f"Required data directory is missing or not a directory: {p}")
+            if not os.access(p, os.W_OK | os.X_OK):
+                raise RuntimeError(f"Required data directory is not writable or accessible: {p}")
+        else:
+            p.mkdir(parents=True, exist_ok=True)
+            created.append(str(p))
+    if read_only:
+        return {"ok": True, "paths": paths, "created": created}
     # write test in status dir
     probe = Path(paths["status"]) / ".borg-ui-write-test"
     probe.write_text("ok\n", encoding="utf-8")
@@ -1167,7 +1179,7 @@ def validate_runtime_config(ui_config: dict) -> dict:
             })
         else:
             try:
-                ensure_data_dirs(data_dir)
+                ensure_data_dirs(data_dir, read_only=True)
             except Exception as exc:
                 errors.append({
                     "key": "GLOBAL_DATA_DIR",

api/history_api.py

@@ -42,24 +42,33 @@ def get_history_data(config: dict, filters: dict | None = None) -> dict:
         per_page = 20
 
     entries = []
+    from jobs_api import discover_jobs, resolve_data_root, resolve_scripts_dir
+    jobs = {job.key: job for job in discover_jobs(resolve_scripts_dir(config), resolve_data_root(config))} if config.get("BACKUP_SCRIPTS_DIR") else {}
     location_counts = {location: 0 for location in ("storagebox", "usb", "smb", "local")}
     known_types = {"flash", "appdata", "photos", "vms", "sonstiges"}
     for f in sorted(status_dir.glob("*.status"), reverse=True):
         # Filename: YYYY-MM-DD_HH-MM-SS_type_location.status
         stem = f.stem
-        parts = stem.split("_")
-        if len(parts) < 4:
+        parts = stem.split("_", 2)
+        if len(parts) < 3:
             continue
         date_part = parts[0]          # 2026-03-01
         time_part = parts[1]          # 02-15-43
-        backup_type = parts[2]        # flash / appdata / …
-        location = "_".join(parts[3:]) # local / usb / storagebox
 
         try:
             raw = json.loads(f.read_text(encoding="utf-8"))
         except (OSError, json.JSONDecodeError):
             continue
 
+        backup_type = str(raw.get("backup_type") or "unknown")
+        location = str(raw.get("location") or "unknown")
+        job_id = str(raw.get("job_id") or "")
+        job = jobs.get(job_id)
+        if job is None:
+            continue
+        if filters.get("job_key") and filters["job_key"] != job_id:
+            continue
+
         status = raw.get("status", "unknown")
         exit_code = raw.get("borg_exit_code", raw.get("exit_code"))
 
@@ -83,6 +92,9 @@ def get_history_data(config: dict, filters: dict | None = None) -> dict:
 
         entries.append({
             "entry_kind": "backup_run",
+            "job_id": job_id,
+            "job_key": job_id,
+            "job_name": (job.name or job.display_name) if job else backup_type,
             "filename": f.name,
             "date": date_part,
             "time": time_part.replace("-", ":"),
@@ -129,6 +141,7 @@ def _ts_key(entry: dict):
     end = start + per_page
 
     return {
+        "jobs": [{"job_id": job.key, "name": job.name or job.display_name} for job in jobs.values()],
         "entries": entries[start:end],
         "total": total,
         "page": page,