BUG: make log rotation safe across Windows processes

xinference/deploy/test/test_log_rotation.py

@@ -15,12 +15,15 @@
 
 import logging
 import os
+import shutil
 import sys
+import threading
 import time
 from unittest import mock
 
 import pytest
 
+from .. import utils as deploy_utils
 from ..utils import SafeTimedAndSizeRotatingFileHandler, get_config_dict
 
 _skip_on_windows = pytest.mark.skipif(
@@ -450,6 +453,227 @@ def test_rolling_rename_on_size_trigger(self, tmp_path):
         assert "test.log.2" in files
 
 
+class _FakeMsvcrt:
+    LK_LOCK = 1
+    LK_UNLCK = 2
+
+    def __init__(self):
+        self.actions = []
+
+    def locking(self, fd, mode, count):
+        self.actions.append((fd, mode, count))
+
+
+class _BlockingFakeMsvcrt(_FakeMsvcrt):
+    def __init__(self):
+        super().__init__()
+        self.lock = threading.Lock()
+        self.acquire_attempted = threading.Event()
+
+    def locking(self, fd, mode, count):
+        super().locking(fd, mode, count)
+        if mode == self.LK_LOCK:
+            self.acquire_attempted.set()
+            self.lock.acquire()
+        else:
+            self.lock.release()
+
+
+class TestWindowsRotationFallback:
+    def test_lock_file_is_closed_when_locking_fails(self, tmp_path):
+        from ..utils import SafeRotatingFileHandler
+
+        class _FailingMsvcrt(_FakeMsvcrt):
+            def locking(self, fd, mode, count):
+                raise OSError("lock failed")
+
+        log_path = str(tmp_path / "test.log")
+        h = SafeRotatingFileHandler(log_path, maxBytes=100, backupCount=1)
+        opened = []
+        real_open = open
+
+        def tracking_open(*args, **kwargs):
+            lock_fd = real_open(*args, **kwargs)
+            opened.append(lock_fd)
+            return lock_fd
+
+        with (
+            mock.patch.object(deploy_utils, "msvcrt", _FailingMsvcrt()),
+            mock.patch("builtins.open", side_effect=tracking_open),
+        ):
+            with pytest.raises(OSError, match="lock failed"):
+                h._acquire_rotation_lock()
+        h.close()
+
+        assert len(opened) == 1
+        assert opened[0].closed
+
+    def test_msvcrt_lock_is_acquired_and_released(self, tmp_path):
+        from ..utils import SafeRotatingFileHandler
+
+        fake_msvcrt = _FakeMsvcrt()
+        log_path = str(tmp_path / "test.log")
+        with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt):
+            h = SafeRotatingFileHandler(log_path, maxBytes=100, backupCount=1)
+            fake_msvcrt.actions.clear()
+            lock_fd = h._acquire_rotation_lock()
+            h._release_rotation_lock(lock_fd)
+            h.close()
+
+        assert [action[1] for action in fake_msvcrt.actions] == [
+            fake_msvcrt.LK_LOCK,
+            fake_msvcrt.LK_UNLCK,
+        ]
+        assert all(action[2] == 1 for action in fake_msvcrt.actions)
+
+    def test_copy_truncate_fallback_preserves_live_file(self, tmp_path):
+        from ..utils import SafeRotatingFileHandler
+
+        fake_msvcrt = _FakeMsvcrt()
+        log_path = str(tmp_path / "test.log")
+        archive_path = str(tmp_path / "test.log.1")
+        with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt):
+            h = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=1)
+            h.stream.write("before rotation\n")
+            h.stream.flush()
+            with mock.patch(
+                "logging.handlers.BaseRotatingHandler.rotate",
+                side_effect=PermissionError(13, "file is in use"),
+            ):
+                h.doRollover()
+            h.close()
+
+        with open(archive_path, encoding="utf8") as archive:
+            assert archive.read() == "before rotation\n"
+        with open(log_path, encoding="utf8") as live:
+            assert live.read() == ""
+
+    def test_copy_truncate_blocks_concurrent_writes(self, tmp_path):
+        from ..utils import SafeRotatingFileHandler
+
+        fake_msvcrt = _BlockingFakeMsvcrt()
+        log_path = str(tmp_path / "test.log")
+        archive_path = tmp_path / "test.log.1"
+        copy_finished = threading.Event()
+        allow_truncate = threading.Event()
+        errors = []
+        real_copy2 = shutil.copy2
+
+        def pausing_copy(source, dest):
+            result = real_copy2(source, dest)
+            copy_finished.set()
+            if not allow_truncate.wait(timeout=5):
+                raise TimeoutError("test did not release the truncate step")
+            return result
+
+        def capture_errors(target):
+            try:
+                target()
+            except BaseException as exc:
+                errors.append(exc)
+
+        with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt):
+            h1 = SafeRotatingFileHandler(log_path, maxBytes=1000, backupCount=1)
+            h2 = SafeRotatingFileHandler(log_path, maxBytes=1000, backupCount=1)
+            h1.setFormatter(logging.Formatter("%(message)s"))
+            h2.setFormatter(logging.Formatter("%(message)s"))
+            h1.emit(_make_record("before rotation"))
+
+            with (
+                mock.patch(
+                    "logging.handlers.BaseRotatingHandler.rotate",
+                    side_effect=PermissionError(13, "file is in use"),
+                ),
+                mock.patch("shutil.copy2", side_effect=pausing_copy),
+            ):
+                rotate_thread = threading.Thread(
+                    target=capture_errors, args=(h1.doRollover,)
+                )
+                rotate_thread.start()
+                assert copy_finished.wait(timeout=5)
+
+                fake_msvcrt.acquire_attempted.clear()
+                write_thread = threading.Thread(
+                    target=capture_errors,
+                    args=(lambda: h2.emit(_make_record("during rotation")),),
+                )
+                write_thread.start()
+                assert fake_msvcrt.acquire_attempted.wait(timeout=5)
+                assert write_thread.is_alive()
+
+                allow_truncate.set()
+                rotate_thread.join(timeout=5)
+                write_thread.join(timeout=5)
+
+            h1.close()
+            h2.close()
+
+        assert not rotate_thread.is_alive()
+        assert not write_thread.is_alive()
+        assert errors == []
+        assert archive_path.read_text(encoding="utf8") == "before rotation\n"
+        assert (tmp_path / "test.log").read_text(encoding="utf8") == (
+            "during rotation\n"
+        )
+
+    def test_size_rotation_generation_prevents_duplicate_archive(self, tmp_path):
+        from ..utils import SafeRotatingFileHandler
+
+        fake_msvcrt = _FakeMsvcrt()
+        log_path = str(tmp_path / "test.log")
+        with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt):
+            h1 = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=2)
+            h2 = SafeRotatingFileHandler(log_path, maxBytes=1, backupCount=2)
+            h1.stream.write("one archive only\n")
+            h1.stream.flush()
+            with mock.patch(
+                "logging.handlers.BaseRotatingHandler.rotate",
+                side_effect=PermissionError(13, "file is in use"),
+            ):
+                h1.doRollover()
+                h2.doRollover()
+            h1.close()
+            h2.close()
+
+        assert (tmp_path / "test.log.1").read_text(encoding="utf8") == (
+            "one archive only\n"
+        )
+        assert not (tmp_path / "test.log.2").exists()
+
+    def test_timed_rotation_marker_prevents_archive_clobber(self, tmp_path):
+        fake_msvcrt = _FakeMsvcrt()
+        log_path = str(tmp_path / "test.log")
+        with mock.patch.object(deploy_utils, "msvcrt", fake_msvcrt):
+            h1 = SafeTimedAndSizeRotatingFileHandler(
+                log_path, when="midnight", backupCount=2, maxBytes=1000
+            )
+            h2 = SafeTimedAndSizeRotatingFileHandler(
+                log_path, when="midnight", backupCount=2, maxBytes=1000
+            )
+            scheduled = int(time.time()) - 1
+            h1.rolloverAt = scheduled
+            h2.rolloverAt = scheduled
+            h1.stream.write("keep this archive\n")
+            h1.stream.flush()
+            with mock.patch(
+                "logging.handlers.BaseRotatingHandler.rotate",
+                side_effect=PermissionError(13, "file is in use"),
+            ):
+                h1.doRollover()
+                h2.doRollover()
+            h1.close()
+            h2.close()
+
+        archives = [
+            path
+            for path in tmp_path.iterdir()
+            if path.name.startswith("test.log.") and not path.name.endswith(".lock")
+        ]
+        assert len(archives) == 1
+        assert archives[0].read_text(encoding="utf8") == "keep this archive\n"
+        assert h2.rolloverAt > time.time()
+
+
 def _mp_worker_size(proc_id, log_path, max_bytes, n_records):
     """Module-level worker for multiprocessing (size mode)."""
     from ..utils import SafeRotatingFileHandler