fix: release 1.5.2 with resilient Windows task state writes

tests/unit/runtime/test_task_control.py

@@ -1,12 +1,256 @@
 """Paused tasks yield their device and resume with a fresh lease."""
 
 import asyncio
+import errno
+import json
+import logging
+import os
+from pathlib import Path
+import threading
 
 import pytest
 
 import artemis.runtime.task_control as task_control
 
 
+def _windows_access_error(code: int) -> PermissionError:
+    error = PermissionError(errno.EACCES, "state file is in use")
+    error.winerror = code
+    return error
+
+
+@pytest.mark.parametrize("winerror", [5, 32, 33])
+def test_running_state_recovers_from_temporary_windows_file_access(tmp_path, monkeypatch, winerror):
+    monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path)
+    path = task_control.control_dir("task-123") / "task_control_state.json"
+    task_control._write_atomic(path, {"status": "queued"})
+    replace = os.replace
+    attempts = []
+    delays = []
+
+    def busy_replace(source, target):
+        attempts.append(source)
+        if len(attempts) <= 2:
+            assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"}
+            raise _windows_access_error(winerror)
+        return replace(source, target)
+
+    monkeypatch.setattr(task_control.os, "replace", busy_replace)
+    monkeypatch.setattr(task_control.time, "sleep", delays.append)
+    task_control.TaskControl("task-123", object())._state("running")
+
+    assert task_control.read_state("task-123")["status"] == "running"
+    assert delays and sum(delays) < 2
+    assert not list(path.parent.glob("*.tmp"))
+
+
+def test_persistent_windows_access_error_is_bounded_and_preserves_old_state(tmp_path, monkeypatch):
+    path = tmp_path / "task_control_state.json"
+    task_control._write_atomic(path, {"status": "queued"})
+    error = _windows_access_error(5)
+    attempts = []
+    delays = []
+
+    def denied_replace(source, target):
+        attempts.append(source)
+        raise error
+
+    monkeypatch.setattr(task_control.os, "replace", denied_replace)
+    monkeypatch.setattr(task_control.time, "sleep", delays.append)
+    with pytest.raises(PermissionError) as raised:
+        task_control._write_atomic(path, {"status": "running"})
+
+    assert raised.value is error
+    assert 1 < len(attempts) < 20
+    assert sum(delays) < 2
+    assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"}
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+@pytest.mark.parametrize(
+    "error", [PermissionError(errno.EACCES, "denied"), OSError(errno.ENOSPC, "disk full")]
+)
+def test_unrelated_file_errors_fail_without_retry_and_clean_temporary_file(
+    tmp_path, monkeypatch, error
+):
+    path = tmp_path / "task_control_state.json"
+    task_control._write_atomic(path, {"status": "queued"})
+    attempts = []
+    delays = []
+
+    def failed_replace(source, target):
+        attempts.append(source)
+        raise error
+
+    monkeypatch.setattr(task_control.os, "replace", failed_replace)
+    monkeypatch.setattr(task_control.time, "sleep", delays.append)
+    with pytest.raises(OSError) as raised:
+        task_control._write_atomic(path, {"status": "running"})
+
+    assert raised.value is error
+    assert len(attempts) == 1
+    assert delays == []
+    assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"}
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_temporary_write_failure_cleans_partial_file(tmp_path, monkeypatch):
+    path = tmp_path / "task_control_state.json"
+    task_control._write_atomic(path, {"status": "queued"})
+    write_text = Path.write_text
+    error = OSError(errno.ENOSPC, "disk full")
+
+    def partial_write(target, content, **kwargs):
+        write_text(target, "{", **kwargs)
+        raise error
+
+    monkeypatch.setattr(Path, "write_text", partial_write)
+    with pytest.raises(OSError) as raised:
+        task_control._write_atomic(path, {"status": "running"})
+
+    assert raised.value is error
+    assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"}
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_busy_temporary_cleanup_preserves_original_write_error(tmp_path, monkeypatch, caplog):
+    path = tmp_path / "task_control_state.json"
+    task_control._write_atomic(path, {"status": "queued"})
+    error = OSError(errno.ENOSPC, "disk full")
+    unlink = Path.unlink
+
+    def failed_replace(source, target):
+        raise error
+
+    def busy_unlink(target, **kwargs):
+        if target.suffix == ".tmp":
+            raise _windows_access_error(32)
+        return unlink(target, **kwargs)
+
+    monkeypatch.setattr(task_control.os, "replace", failed_replace)
+    monkeypatch.setattr(Path, "unlink", busy_unlink)
+    monkeypatch.setattr(task_control.time, "sleep", lambda _: None)
+    try:
+        with caplog.at_level(logging.WARNING, logger=task_control.__name__):
+            with pytest.raises(OSError) as raised:
+                task_control._write_atomic(path, {"status": "running"})
+        assert raised.value is error
+        assert "Could not remove task control temporary file" in caplog.text
+        assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"}
+    finally:
+        for temporary in tmp_path.glob("*.tmp"):
+            unlink(temporary)
+
+
+@pytest.mark.parametrize(
+    "send_request,filename",
+    [
+        (task_control.request_pause, "pause.request"),
+        (task_control.request_resume, "resume.request"),
+    ],
+)
+def test_repeated_control_requests_recover_from_file_sharing(
+    tmp_path, monkeypatch, send_request, filename
+):
+    monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path)
+    send_request("task-123")
+    replace = os.replace
+    attempts = []
+
+    def busy_replace(source, target):
+        attempts.append(target)
+        if len(attempts) == 1:
+            raise _windows_access_error(32)
+        return replace(source, target)
+
+    monkeypatch.setattr(task_control.os, "replace", busy_replace)
+    monkeypatch.setattr(task_control.time, "sleep", lambda _: None)
+    send_request("task-123")
+
+    assert (
+        json.loads((task_control.control_dir("task-123") / filename).read_text(encoding="utf-8"))[
+            "created_at"
+        ]
+        > 0
+    )
+    assert not list(task_control.control_dir("task-123").glob("*.tmp"))
+
+
+@pytest.mark.parametrize("persistent", [False, True])
+def test_release_does_not_fail_task_when_state_cleanup_is_busy(
+    tmp_path, monkeypatch, caplog, persistent
+):
+    monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path)
+    released = []
+
+    class Lock:
+        def release(self):
+            released.append(True)
+
+    control = task_control.TaskControl("task-123", Lock())
+    control._state("running")
+    path = control.directory / "task_control_state.json"
+    unlink = Path.unlink
+    attempts = []
+
+    def busy_unlink(target, **kwargs):
+        if target == path:
+            attempts.append(target)
+            if persistent or len(attempts) == 1:
+                raise _windows_access_error(32)
+        return unlink(target, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", busy_unlink)
+    monkeypatch.setattr(task_control.time, "sleep", lambda _: None)
+    with caplog.at_level(logging.WARNING, logger=task_control.__name__):
+        control.release()
+
+    assert released == [True]
+    assert 1 < len(attempts) < 20
+    assert path.exists() == persistent
+    if persistent:
+        assert path.name in caplog.text
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows file sharing semantics")
+def test_running_state_survives_real_windows_reader_handle(tmp_path, monkeypatch):
+    monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path)
+    control = task_control.TaskControl("task-123", object())
+    control._state("queued")
+    path = control.directory / "task_control_state.json"
+    reader = path.open(encoding="utf-8")
+    blocked = threading.Event()
+    errors = []
+    replace = os.replace
+
+    def release_reader():
+        if blocked.wait(2):
+            reader.close()
+
+    def observe_replace(source, target):
+        try:
+            return replace(source, target)
+        except OSError as exc:
+            errors.append(exc.winerror)
+            blocked.set()
+            raise
+
+    releaser = threading.Thread(target=release_reader)
+    releaser.start()
+    monkeypatch.setattr(task_control.os, "replace", observe_replace)
+    try:
+        control._state("running")
+    finally:
+        blocked.set()
+        releaser.join(timeout=3)
+        reader.close()
+
+    assert not releaser.is_alive()
+    assert errors and set(errors).issubset({5, 32, 33})
+    assert task_control.read_state("task-123")["status"] == "running"
+    assert not list(control.directory.glob("*.tmp"))
+
+
 @pytest.mark.asyncio
 async def test_pause_releases_device_and_resume_waits_for_new_acquisition(tmp_path, monkeypatch):
     monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path)