diff --git a/src/cognitive_agent/application/firmware_workbench.py b/src/cognitive_agent/application/firmware_workbench.py index ebebcf5..7a6c070 100644 --- a/src/cognitive_agent/application/firmware_workbench.py +++ b/src/cognitive_agent/application/firmware_workbench.py @@ -45,6 +45,7 @@ FIRMWARE_VERSION = "SMV8-r15-b16" FINAL_RELEASE_MESSAGE = "已发布，待车机端下载升级" FIRMWARE_NOTIFICATION_TOPIC = "from_pc/firmware/notification" FIRMWARE_NOTIFICATION_FRAME_COUNT = 3 +_ATOMIC_REPLACE_RETRY_DELAYS = (0.02, 0.04, 0.08, 0.16, 0.32, 0.5, 0.5) BuildRunner = Callable[[str, Callable[[str, str], None], threading.Event], Path] @@ -69,6 +70,35 @@ def _safe_file_id(value: str, kind: str) -> str: return normalized +def _atomic_write_json(path: Path, document: Mapping[str, Any]) -> None: + """Keep the previous JSON intact while retrying brief Windows file locks.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + try: + with temporary.open("x", encoding="utf-8") as stream: + json.dump(document, stream, ensure_ascii=False, indent=2) + stream.flush() + os.fsync(stream.fileno()) + for attempt in range(len(_ATOMIC_REPLACE_RETRY_DELAYS) + 1): + try: + # The temporary handle must be closed before a Windows rename. + temporary.replace(path) + return + except OSError as exc: + retryable = isinstance(exc, PermissionError) or getattr( + exc, "winerror", None + ) in {5, 32, 33} + if not retryable or attempt == len(_ATOMIC_REPLACE_RETRY_DELAYS): + raise + time.sleep(_ATOMIC_REPLACE_RETRY_DELAYS[attempt]) + finally: + try: + temporary.unlink(missing_ok=True) + except OSError: + # Cleanup must not hide the original write/replace failure. + pass + + class FirmwareWorkbenchService: """Execute a real IAR build and publish its package into an OTA repository.""" @@ -122,30 +152,33 @@ class FirmwareWorkbenchService: return try: data = json.loads(self._state_file.read_text(encoding="utf-8")) - if isinstance(data, dict): - for key in ("tasks", "artifacts", "releases"): - if not isinstance(data.get(key), list): - data[key] = [] - self._state = data - recovered = False - for task in self._state["tasks"]: - if task.get("status") in ACTIVE_TASK_STATES: - task["status"] = "blocked" - task["finished_at"] = _now() - task["message"] = "服务重启中断了任务，请点击重试。" - self._finish_running_stage(task, "blocked", task["message"]) - self._append_log_locked( - task, - "error", - "任务恢复", - task["message"], - stage="pipeline", - ) - recovered = True - if recovered: - self._save_locked() - except (OSError, ValueError, TypeError): + except (ValueError, TypeError): self._state = {"tasks": [], "artifacts": [], "releases": []} + return + if isinstance(data, dict): + for key in ("tasks", "artifacts", "releases"): + if not isinstance(data.get(key), list): + data[key] = [] + self._state = data + recovered = False + for task in self._state["tasks"]: + if task.get("status") in ACTIVE_TASK_STATES: + task["status"] = "blocked" + task["finished_at"] = _now() + task["message"] = "服务重启中断了任务，请点击重试。" + self._finish_running_stage(task, "blocked", task["message"]) + self._append_log_locked( + task, + "error", + "任务恢复", + task["message"], + stage="pipeline", + ) + recovered = True + # An I/O error is not corrupt JSON. Do not reset history if recovery + # cannot be persisted, or the baseline registration could overwrite it. + if recovered: + self._save_locked() def _ensure_time_only_baseline(self) -> None: """Register the immutable timed-control firmware as the workbench baseline.""" @@ -229,12 +262,7 @@ class FirmwareWorkbenchService: self._save_locked() def _save_locked(self) -> None: - self.artifact_root.mkdir(parents=True, exist_ok=True) - temporary = self._state_file.with_suffix(".tmp") - temporary.write_text( - json.dumps(self._state, ensure_ascii=False, indent=2), encoding="utf-8" - ) - temporary.replace(self._state_file) + _atomic_write_json(self._state_file, self._state) def _synchronize_public_release_urls(self) -> None: """Apply the configured vehicle-facing base URL to persisted releases.""" @@ -258,17 +286,11 @@ class FirmwareWorkbenchService: changed = True if not changed: return - latest_temp = self.ota_root / "latest.tmp" if self._state["releases"]: - latest_temp.write_text( - json.dumps( - self._public_release_payload(self._state["releases"][0]), - ensure_ascii=False, - indent=2, - ), - encoding="utf-8", + _atomic_write_json( + self.ota_root / "latest.json", + self._public_release_payload(self._state["releases"][0]), ) - latest_temp.replace(self.ota_root / "latest.json") self._save_locked() def _find_task_locked(self, task_id: str) -> dict[str, Any]: @@ -646,12 +668,7 @@ class FirmwareWorkbenchService: latest_path.unlink(missing_ok=True) else: latest_manifest = self._public_release_payload(self._state["releases"][0]) - latest_temp = self.ota_root / "latest.tmp" - latest_temp.write_text( - json.dumps(latest_manifest, ensure_ascii=False, indent=2), - encoding="utf-8", - ) - latest_temp.replace(latest_path) + _atomic_write_json(latest_path, latest_manifest) self._save_locked() return { "deleted": True, @@ -1396,11 +1413,7 @@ class FirmwareWorkbenchService: ) release_root.parent.mkdir(parents=True, exist_ok=True) staging_root.replace(release_root) - latest_temp = self.ota_root / "latest.tmp" - latest_temp.write_text( - json.dumps(release_manifest, ensure_ascii=False, indent=2), encoding="utf-8" - ) - latest_temp.replace(self.ota_root / "latest.json") + _atomic_write_json(self.ota_root / "latest.json", release_manifest) final_package = release_root / str(artifact["firmware_filename"]) release = { **release_manifest, diff --git a/tests/unit/test_firmware_workbench.py b/tests/unit/test_firmware_workbench.py index 3ccc3f7..acaf60d 100644 --- a/tests/unit/test_firmware_workbench.py +++ b/tests/unit/test_firmware_workbench.py @@ -2,6 +2,7 @@ from __future__ import annotations import hashlib import json +import os import threading import time import zipfile @@ -13,6 +14,7 @@ from cognitive_agent.application.firmware_workbench import ( FINAL_RELEASE_MESSAGE, FirmwareWorkbenchError, FirmwareWorkbenchService, + _atomic_write_json, ) from cognitive_agent.infrastructure.can.gateway import CanGatewayError @@ -339,3 +341,184 @@ def test_rocking_horse_build_failure_does_not_publish(tmp_path: Path) -> None: assert all(item.get("is_baseline") for item in service.snapshot()["artifacts"]) finally: service.close() + + +def test_atomic_state_write_retries_locks_without_touching_other_temporary_files( + tmp_path: Path, monkeypatch, +) -> None: + path = tmp_path / "state.json" + previous = '{"tasks": [{"id": "previous-task"}]}' + path.write_text(previous, encoding="utf-8") + legacy_temporary = tmp_path / "state.tmp" + legacy_temporary.write_text("another writer", encoding="utf-8") + original_replace = Path.replace + attempts = [] + + def blocked_replace(source, destination): + attempts.append(source) + assert path.read_text(encoding="utf-8") == previous + assert json.loads(source.read_text(encoding="utf-8")) == {"tasks": ["新任务"]} + if len(attempts) <= 2: + raise PermissionError(13, "file temporarily in use", str(destination)) + return original_replace(source, destination) + + monkeypatch.setattr(Path, "replace", blocked_replace) + _atomic_write_json(path, {"tasks": ["新任务"]}) + assert len(attempts) == 3 + assert attempts[0] != legacy_temporary + assert len(set(attempts)) == 1 + assert legacy_temporary.read_text(encoding="utf-8") == "another writer" + assert json.loads(path.read_text(encoding="utf-8")) == {"tasks": ["新任务"]} + assert list(tmp_path.glob(".state.json.*.tmp")) == [] + + +@pytest.mark.parametrize("error,expected_attempts", [ + (PermissionError(13, "access denied"), 8), + (OSError(28, "disk full"), 1), +]) +def test_atomic_write_failure_preserves_history_and_cleans_its_temporary_file( + tmp_path: Path, monkeypatch, error, expected_attempts, +) -> None: + path = tmp_path / "state.json" + previous = '{"tasks": [{"id": "preserved-task"}]}' + path.write_text(previous, encoding="utf-8") + attempts = [] + + def reject_replace(source, destination): + attempts.append(source) + raise error + + monkeypatch.setattr(Path, "replace", reject_replace) + with pytest.raises(type(error)): + _atomic_write_json(path, {"tasks": []}) + assert len(attempts) == expected_attempts + assert path.read_text(encoding="utf-8") == previous + assert list(tmp_path.glob(".state.json.*.tmp")) == [] + + +def test_build_and_publish_survive_transient_state_and_manifest_locks( + tmp_path: Path, monkeypatch, +) -> None: + image = make_image(tmp_path) + building = threading.Event() + failures = {"state.json": 2, "latest.json": 2} + original_replace = Path.replace + + def replace_with_locks(source, destination): + name = Path(destination).name + if building.is_set() and failures.get(name, 0): + failures[name] -= 1 + raise PermissionError(13, "temporary Windows file lock", str(destination)) + return original_replace(source, destination) + + def build_runner(task_id, log, cancel): + building.set() + log("info", "IAR build: 0 errors, 0 warnings") + return image + + service = FirmwareWorkbenchService( + ROOT, artifact_root=tmp_path / "workbench", ota_root=tmp_path / "ota", + build_runner=build_runner, + ) + monkeypatch.setattr(Path, "replace", replace_with_locks) + try: + task = wait_for_terminal(service, service.start_task("增加动态木马功能")["task"]["id"]) + assert task["status"] == "success", task.get("failure_reason") + assert [stage["status"] for stage in task["stages"]] == ["success"] * 8 + assert failures == {"state.json": 0, "latest.json": 0} + stored = json.loads((tmp_path / "workbench/state.json").read_text(encoding="utf-8")) + assert stored["tasks"][0]["status"] == "success" + latest = json.loads((tmp_path / "ota/latest.json").read_text(encoding="utf-8")) + assert latest["release_id"] == task["release_id"] + finally: + service.close() + + +def test_recovery_save_failure_does_not_reinitialize_existing_history(tmp_path: Path, monkeypatch): + path = tmp_path / "workbench/state.json" + path.parent.mkdir() + original = {"tasks": [{"id": "fw-task-recovery", "status": "running", "stages": []}], + "artifacts": [], "releases": []} + path.write_text(json.dumps(original), encoding="utf-8") + save = FirmwareWorkbenchService._save_locked + attempts = [] + + def fail_first_save(service): + attempts.append(True) + if len(attempts) == 1: + raise PermissionError(13, "state is locked") + save(service) + + monkeypatch.setattr(FirmwareWorkbenchService, "_save_locked", fail_first_save) + with pytest.raises(PermissionError): + FirmwareWorkbenchService(ROOT, artifact_root=path.parent, ota_root=tmp_path / "ota") + assert len(attempts) == 1 + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_unreadable_state_is_not_overwritten_by_baseline_registration(tmp_path: Path, monkeypatch): + path = tmp_path / "workbench/state.json" + path.parent.mkdir() + previous = '{"tasks": [{"id": "preserved-task"}], "artifacts": [], "releases": []}' + path.write_text(previous, encoding="utf-8") + read = Path.read_text + + def deny_state_read(source, *args, **kwargs): + if source == path: + raise PermissionError(13, "state read denied") + return read(source, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", deny_state_read) + with pytest.raises(PermissionError): + FirmwareWorkbenchService(ROOT, artifact_root=path.parent, ota_root=tmp_path / "ota") + assert path.read_bytes() == previous.encode("utf-8") + + +@pytest.mark.skipif(os.name != "nt", reason="Requires a real Windows file-sharing lock") +def test_atomic_write_recovers_from_a_real_windows_reader_lock(tmp_path: Path, monkeypatch): + import ctypes + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, + wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] + kernel.CreateFileW.restype = wintypes.HANDLE + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + path = tmp_path / "state.json" + path.write_text('{"before": true}', encoding="utf-8") + # Permit read/write, but intentionally omit FILE_SHARE_DELETE to reproduce WinError 5/32. + handle = kernel.CreateFileW(str(path), 0x80000000, 0x00000003, None, 3, 0x80, None) + assert handle != ctypes.c_void_p(-1).value, ctypes.get_last_error() + blocked = threading.Event() + errors = [] + observed_winerrors = [] + replace = Path.replace + + def observe_replace(source, destination): + try: + return replace(source, destination) + except PermissionError as exc: + observed_winerrors.append(exc.winerror) + blocked.set() + raise + + def write(): + try: + _atomic_write_json(path, {"after": True}) + except Exception as exc: + errors.append(exc) + + monkeypatch.setattr(Path, "replace", observe_replace) + worker = threading.Thread(target=write) + worker.start() + try: + assert blocked.wait(timeout=2), "Windows did not reproduce the sharing conflict" + assert json.loads(path.read_text(encoding="utf-8")) == {"before": True} + finally: + kernel.CloseHandle(handle) + worker.join(timeout=4) + assert not worker.is_alive() + assert not errors + assert observed_winerrors and set(observed_winerrors) <= {5, 32, 33} + assert json.loads(path.read_text(encoding="utf-8")) == {"after": True}