Fix unhandled PermissionError during file retention sweep on Windows

CHANGELOG.rst

@@ -1,6 +1,7 @@
 `Unreleased`_
 =============
 
+- Fix possible ``PermissionError`` raised on Windows during file retention sweep when another process keeps a log file open; the sweep now skips locked or concurrently deleted files and still deletes the others (`#1495 <https://github.com/Delgan/loguru/issues/1495>`_).
 - Change the default log format to include the timezone offset since it produces less ambiguous logs (`#856 <https://github.com/Delgan/loguru/pull/856>`_, thanks `@tim-x-y-z <https://github.com/tim-x-y-z>`_).
 - Add new ``logger.reinstall()`` method to automatically set up the ``logger`` in spawned child processes (`#818 <https://github.com/Delgan/loguru/issues/818>`_, thanks `@monchin <https://github.com/monchin>`_).
 - Add support for template strings used as log messages (`#1397 <https://github.com/Delgan/loguru/issues/1397>`_, thanks `@TurtleOrangina <https://github.com/TurtleOrangina>`_).

loguru/_file_sink.py

@@ -71,17 +71,35 @@ class Retention:
     @staticmethod
     def retention_count(logs, number):
         def key_log(log):
-            return (-os.stat(log).st_mtime, log)
+            try:
+                return (-os.stat(log).st_mtime, log)
+            except FileNotFoundError:
+                # The file disappeared between the glob and the sort, so it
+                # cannot be part of the retained set anyway.
+                return (0, log)
 
         for log in sorted(logs, key=key_log)[number:]:
-            os.remove(log)
+            try:
+                os.remove(log)
+            except (PermissionError, FileNotFoundError):
+                # Another process may hold the file open (Windows) or have
+                # deleted it concurrently. Skip it rather than aborting the
+                # whole retention sweep, which would also leave the other
+                # out-of-retention files undeleted.
+                continue
 
     @staticmethod
     def retention_age(logs, seconds):
         t = datetime.datetime.now().timestamp()
         for log in logs:
-            if os.stat(log).st_mtime <= t - seconds:
-                os.remove(log)
+            try:
+                if os.stat(log).st_mtime <= t - seconds:
+                    os.remove(log)
+            except (PermissionError, FileNotFoundError):
+                # Same as ``retention_count``: a file locked by another
+                # process (Windows) or concurrently deleted must not abort
+                # the sweep or bubble up during interpreter exit.
+                continue
 
 
 class Rotation:
@@ -174,7 +192,7 @@ def __init__(
         mode="a",
         buffering=1,
         encoding="utf8",
-        **kwargs
+        **kwargs,
     ):
         self.encoding = encoding
 

tests/test_filesink_retention.py

@@ -5,6 +5,7 @@
 import pytest
 
 from loguru import logger
+from loguru._file_sink import Retention
 
 from .conftest import check_dir
 
@@ -343,6 +344,86 @@ def test_exception_during_retention_at_remove(tmp_path, capsys, delay):
     assert out == err == ""
 
 
+def test_retention_count_skips_locked_file(tmp_path, monkeypatch):
+    # Regression test for #1495: a file locked by another process (Windows)
+    # must not abort the sweep or propagate during interpreter exit.
+    for i, mtime in enumerate([100.0, 200.0, 300.0]):
+        path = tmp_path.joinpath("test.%d.log" % i)
+        path.write_text("test")
+        os.utime(path, (mtime, mtime))
+
+    locked = str(tmp_path.joinpath("test.0.log"))
+    real_remove = os.remove
+
+    def fake_remove(path):
+        if os.fspath(path) == locked:
+            raise PermissionError("file is locked")
+        return real_remove(path)
+
+    monkeypatch.setattr(os, "remove", fake_remove)
+
+    Retention.retention_count(
+        [locked] + [str(tmp_path.joinpath("test.%d.log" % i)) for i in (1, 2)],
+        number=1,
+    )
+
+    # The sweep did not abort: the locked file remains but the other
+    # out-of-retention files were still deleted.
+    assert os.path.exists(locked)
+    check_dir(tmp_path, size=2)
+
+
+def test_retention_count_skips_concurrently_deleted_file(tmp_path, monkeypatch):
+    # A file deleted concurrently must not abort the sweep either.
+    for i, mtime in enumerate([100.0, 200.0, 300.0]):
+        path = tmp_path.joinpath("test.%d.log" % i)
+        path.write_text("test")
+        os.utime(path, (mtime, mtime))
+
+    gone = str(tmp_path.joinpath("test.0.log"))
+    os.remove(gone)
+
+    real_remove = os.remove
+
+    def fake_remove(path):
+        if os.fspath(path) == gone:
+            raise FileNotFoundError("already deleted")
+        return real_remove(path)
+
+    monkeypatch.setattr(os, "remove", fake_remove)
+
+    Retention.retention_count(
+        [gone] + [str(tmp_path.joinpath("test.%d.log" % i)) for i in (1, 2)], number=1
+    )
+
+    check_dir(tmp_path, size=1)
+
+
+def test_retention_age_skips_locked_file(tmp_path, monkeypatch):
+    # Same as above for the age-based retention.
+    now = datetime.datetime.now().timestamp()
+    for i, mtime in enumerate([now - 100, now - 50, now]):
+        path = tmp_path.joinpath("test.%d.log" % i)
+        path.write_text("test")
+        os.utime(path, (mtime, mtime))
+
+    locked = str(tmp_path.joinpath("test.0.log"))
+    real_remove = os.remove
+
+    def fake_remove(path):
+        if os.fspath(path) == locked:
+            raise PermissionError("file is locked")
+        return real_remove(path)
+
+    monkeypatch.setattr(os, "remove", fake_remove)
+
+    logs = [str(tmp_path.joinpath("test.%d.log" % i)) for i in range(3)]
+    Retention.retention_age(logs, seconds=10)
+
+    assert os.path.exists(locked)
+    check_dir(tmp_path, size=2)
+
+
 @pytest.mark.parametrize("retention", [datetime.time(12, 12, 12), os, object()])
 def test_invalid_retention_type(retention):
     with pytest.raises(TypeError):