From 2214575bd1c1f03ee36f8d7fc6eec1f8e7c10042 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Mon, 9 Nov 2026 10:54:09 +0100 Subject: [PATCH] fix(automation): atomic state file writing and Windows PermissionError retry (T-20261107-192001888) - implement _atomic_write_bytes and _atomic_write_json with temp file in same directory and os.replace - preserve original file mode when target exists - retry os.replace with exponential backoff on transient Windows PermissionError - write paused-by-carecenter.json atomically - flush append buffer for events.jsonl - add unit tests for atomic state writing, transient retry and permanent failure cleanup --- .../automation_control.py | 61 ++++++++-- tests/test_automation_control.py | 107 +++++++++++++++++- 2 files changed, 156 insertions(+), 12 deletions(-) diff --git a/src/codex_logdatenbank_wartung/automation_control.py b/src/codex_logdatenbank_wartung/automation_control.py index 7daf397..ecb392b 100644 --- a/src/codex_logdatenbank_wartung/automation_control.py +++ b/src/codex_logdatenbank_wartung/automation_control.py @@ -185,17 +185,29 @@ def _read_automation(path: Path, *, match_folder: bool = True) -> tomlkit.TOMLDo return data -def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None: +def _atomic_write_bytes( + path: Path, + payload: bytes, + *, + retries: int = 5, + retry_delay: float = 0.05, +) -> None: """Schreibt atomar: erst vollstaendig kodieren, dann tmp im selben Ordner, fsync, ``os.replace``. Bricht der Vorgang ab, bleibt die Originaldatei unveraendert und die tmp-Datei wird entfernt (auch eine schreibgeschuetzte); der urspruengliche Fehler wird nicht durch einen Aufraeumfehler ersetzt. - ``mkstemp`` legt die Datei mit 0600 an: die Rechte des Originals kommen vor dem Austausch auf die tmp-Datei. + ``mkstemp`` legt die Datei mit 0600 an: die Rechte des Originals kommen vor dem Austausch auf die tmp-Datei, + sofern das Original bereits existiert. Ist ``path`` ein Link, wird sein Ziel ersetzt und der Link bleibt bestehen. + Unter Windows kann ``os.replace`` bei transienten Dateisperren (z. B. Virenscanner, Backup, Indexer) einen + ``PermissionError`` werfen. Dieser wird mit kurzem Retry und exponentiellem Backoff abgefangen. """ - payload = tomlkit.dumps(document).encode("utf-8") + path.parent.mkdir(parents=True, exist_ok=True) target = Path(os.path.realpath(path)) - mode = stat.S_IMODE(target.stat().st_mode) + try: + mode: int | None = stat.S_IMODE(target.stat().st_mode) + except OSError: + mode = None descriptor, tmp_name = tempfile.mkstemp(dir=target.parent, prefix=f"{target.name}.", suffix=".tmp") temporary = Path(tmp_name) try: @@ -203,8 +215,16 @@ def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None: handle.write(payload) handle.flush() os.fsync(handle.fileno()) - os.chmod(temporary, mode) - os.replace(temporary, target) + if mode is not None: + os.chmod(temporary, mode) + for attempt in range(max(1, retries)): + try: + os.replace(temporary, target) + break + except PermissionError: + if attempt >= retries - 1: + raise + time.sleep(retry_delay * (2 ** attempt)) except BaseException: # Die tmp-Datei kann das Nur-Lesen-Attribut des Originals tragen (Windows): ohne chmod scheitert unlink, # laesst die Datei liegen und ueberdeckt den eigentlichen Fehler. Dessen Text bleibt der, der zaehlt. @@ -216,6 +236,25 @@ def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None: raise +def _atomic_write_json( + path: Path, + data: object, + *, + indent: int = 2, + retries: int = 5, + retry_delay: float = 0.05, +) -> None: + """Schreibt ein JSON-Objekt atomar via ``_atomic_write_bytes``.""" + payload = (json.dumps(data, ensure_ascii=False, indent=indent) + "\n").encode("utf-8") + _atomic_write_bytes(path, payload, retries=retries, retry_delay=retry_delay) + + +def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None: + """Schreibt atomar: erst vollstaendig kodieren, dann tmp im selben Ordner, fsync, ``os.replace``.""" + payload = tomlkit.dumps(document).encode("utf-8") + _atomic_write_bytes(path, payload) + + def _quoted_string(value: object) -> str: return str(value or "") @@ -343,13 +382,12 @@ def load_carecenter_paused_ids(config: MaintenanceConfig) -> list[str]: def _save_carecenter_paused_ids(config: MaintenanceConfig, ids: set[str]) -> None: path = control_state_path(config) - path.parent.mkdir(parents=True, exist_ok=True) payload = { "tool": "CareCenter for Codex", "updated_at": _now_iso(), "paused_by_carecenter_ids": sorted(ids), } - path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + _atomic_write_json(path, payload, indent=2) def _append_event(config: MaintenanceConfig, result: AutomationControlResult) -> None: @@ -361,6 +399,7 @@ def _append_event(config: MaintenanceConfig, result: AutomationControlResult) -> } with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(payload, ensure_ascii=False) + "\n") + handle.flush() def _set_status(record: AutomationRecord, status: str) -> bool: @@ -438,7 +477,8 @@ def pause_active_automations(config: MaintenanceConfig) -> AutomationControlResu status="ok", target_count=len(active_records), ) - carecenter_ids = set(load_carecenter_paused_ids(config)) + initial_ids = set(load_carecenter_paused_ids(config)) + carecenter_ids = set(initial_ids) for record in active_records: try: if _set_status(record, PAUSED): @@ -451,7 +491,8 @@ def pause_active_automations(config: MaintenanceConfig) -> AutomationControlResu _record_unparseable(result, unparseable) if result.errors: result.status = "partial" if result.changed_ids else "failed" - _save_carecenter_paused_ids(config, carecenter_ids) + if result.changed_ids or carecenter_ids != initial_ids: + _save_carecenter_paused_ids(config, carecenter_ids) _append_event(config, result) return result diff --git a/tests/test_automation_control.py b/tests/test_automation_control.py index 0dbfad8..1108f88 100644 --- a/tests/test_automation_control.py +++ b/tests/test_automation_control.py @@ -373,8 +373,9 @@ def spy_replace(src: object, dst: object) -> None: pause_active_automations(config) - assert [name for name, _ in events] == ["fsync", "replace"] + assert [name for name, _ in events] == ["fsync", "replace", "fsync", "replace"] assert events[1][1] == (path.parent, path) + assert events[3][1] == (control_state_dir(config), control_state_path(config)) assert read_status(path) == "PAUSED" @@ -537,7 +538,7 @@ def spy_replace(src: object, dst: object) -> None: pause_active_automations(config) - assert calls == [("chmod", original_mode, True), ("replace", 0, False)] + assert calls[:2] == [("chmod", original_mode, True), ("replace", 0, False)] @pytest.mark.skipif(os.name != "nt", reason="nur unter Windows scheitert os.replace an einer schreibgeschuetzten Datei") @@ -615,3 +616,105 @@ def fake_realpath(value: object, *args: object, **kwargs: object) -> str: assert read_status(target) == "PAUSED" assert link_path.read_bytes() == link_before, "der Link-Pfad selbst bleibt unberuehrt" assert sorted(item.name for item in real_dir.iterdir()) == ["job.toml"] + + +# --- Atomare Zustandsdateien & PermissionError-Retry (T-20261107-192001888) ------------------------- + + +def test_save_carecenter_paused_ids_is_atomic_and_syncs_in_same_folder( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = make_config(tmp_path) + state_file = control_state_path(config) + state_dir = control_state_dir(config) + events: list[tuple[str, object]] = [] + real_fsync, real_replace = os.fsync, os.replace + + def spy_fsync(fd: int) -> None: + events.append(("fsync", fd)) + real_fsync(fd) + + def spy_replace(src: object, dst: object) -> None: + events.append(("replace", (Path(str(src)).parent, Path(str(dst))))) + real_replace(src, dst) + + monkeypatch.setattr(automation_control.os, "fsync", spy_fsync) + monkeypatch.setattr(automation_control.os, "replace", spy_replace) + + automation_control._save_carecenter_paused_ids(config, {"job-1", "job-2"}) + + assert [name for name, _ in events] == ["fsync", "replace"] + assert events[1][1] == (state_dir, state_file) + assert load_carecenter_paused_ids(config) == ["job-1", "job-2"] + # Es darf keine verwaiste temporaere Datei im Verzeichnis verbleiben + assert sorted(item.name for item in state_dir.iterdir()) == ["paused-by-carecenter.json"] + + +def test_save_carecenter_paused_ids_retries_on_transient_permission_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = make_config(tmp_path) + attempts = 0 + real_replace = os.replace + + def flaky_replace(src: object, dst: object) -> None: + nonlocal attempts + attempts += 1 + if attempts < 3: + raise PermissionError("Access is denied (transient lock)") + real_replace(src, dst) + + monkeypatch.setattr(automation_control.os, "replace", flaky_replace) + monkeypatch.setattr(automation_control.time, "sleep", lambda _s: None) + + automation_control._save_carecenter_paused_ids(config, {"resilient-job"}) + + assert attempts == 3 + assert load_carecenter_paused_ids(config) == ["resilient-job"] + assert sorted(item.name for item in control_state_dir(config).iterdir()) == ["paused-by-carecenter.json"] + + +def test_save_carecenter_paused_ids_fails_cleanly_on_permanent_permission_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = make_config(tmp_path) + state_file = control_state_path(config) + attempts = 0 + + def blocked_replace(src: object, dst: object) -> None: + nonlocal attempts + attempts += 1 + raise PermissionError("Access is denied (permanent lock)") + + monkeypatch.setattr(automation_control.os, "replace", blocked_replace) + monkeypatch.setattr(automation_control.time, "sleep", lambda _s: None) + + with pytest.raises(PermissionError, match="permanent lock"): + automation_control._save_carecenter_paused_ids(config, {"failed-job"}) + + assert attempts == 5 + # Tmp-Datei muss aufgeraeumt sein, kein Muell + state_dir = control_state_dir(config) + assert not state_file.exists() + assert list(state_dir.iterdir()) == [] + + +def test_save_carecenter_paused_ids_interrupted_leaves_previous_state_intact( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + config = make_config(tmp_path) + automation_control._save_carecenter_paused_ids(config, {"initial-job"}) + previous_bytes = control_state_path(config).read_bytes() + + def crash_replace(src: object, dst: object) -> None: + raise OSError("unexpected crash during replace") + + monkeypatch.setattr(automation_control.os, "replace", crash_replace) + + with pytest.raises(OSError, match="unexpected crash"): + automation_control._save_carecenter_paused_ids(config, {"overwritten-job"}) + + assert control_state_path(config).read_bytes() == previous_bytes + assert load_carecenter_paused_ids(config) == ["initial-job"] + assert sorted(item.name for item in control_state_dir(config).iterdir()) == ["paused-by-carecenter.json"] +