fix: Retry session directory cleanup on Windows while handles are released

src/openjd/sessions/_session.py

@@ -603,7 +603,7 @@ def cleanup(self) -> None:
                     # Note: Blocking call until the process has exited
                     _subprocess.run()
 
-                self._working_dir.cleanup()
+                self._working_dir.cleanup(logger=self._logger)
             except RuntimeError as exc:
                 # Warn if we couldn't cleanup the temporary files for some reason.
                 self._logger.exception(

src/openjd/sessions/_tempdir.py

@@ -3,7 +3,8 @@
 import os
 import stat
 import sys
-from ._logging import LoggerAdapter, LogContent, LogExtraInfo
+import time
+from ._logging import LOG, LoggerAdapter, LogContent, LogExtraInfo
 from pathlib import Path
 from shutil import rmtree
 from ._embedded_files import chown_group
@@ -32,6 +33,69 @@
 mode does not depend on how the embedding application happens to be configured.
 """
 
+_CLEANUP_RETRY_DELAYS_SECONDS: tuple[float, ...] = (1.0, 2.0, 4.0, 8.0, 16.0)
+"""Seconds to wait before each retry of a failed Windows cleanup: about 31 s in total.
+
+Windows refuses to delete a file or directory while any process holds a handle to
+it, and a child process that has exited or been terminated can still hold handles
+for a while. POSIX does not retry: a failure there is a permission problem that
+waiting will not fix.
+"""
+
+_MAX_FILES_TO_INSPECT = 256
+"""Most files a failed cleanup inspects to find what is still holding them."""
+
+_MAX_FILES_TO_NAME = 3
+"""Most undeletable file names a failed-attempt log line lists."""
+
+
+def _find_lock_holders(files: list[str]) -> tuple[list[tuple[int, str]], Optional[str]]:
+    """The (pid, name) of each process holding any of ``files`` open, and the
+    reason the lookup failed if it did. Windows only.
+
+    Never raises: the result only feeds a log line, and an exception here would
+    abort the cleanup it is explaining.
+    """
+    try:
+        from ._win32._file_locks import processes_holding  # type: ignore
+
+        return [(h.pid, h.name) for h in processes_holding(files)], None
+    except Exception as err:
+        return [], f"{type(err).__name__}: {err}"
+
+
+def _describe_failed_attempt(
+    dir: Path,
+    error: BaseException,
+    files: list[str],
+    holders: list[tuple[int, str]],
+    holders_error: Optional[str],
+    retry_in: Optional[float],
+) -> str:
+    """The text of a failed-attempt log line. ``files`` are those that could not be deleted."""
+    reason = getattr(error, "strerror", None) or str(error)
+    msg = f"Could not delete temporary directory {dir}: {reason}."
+    if files:
+        names = []
+        for p in files[:_MAX_FILES_TO_NAME]:
+            try:
+                names.append(str(Path(p).relative_to(dir)))
+            except ValueError:
+                names.append(str(p))
+        msg += f" {len(files)} file(s) could not be deleted, including {', '.join(names)}."
+        if holders:
+            seen: dict[int, str] = {}
+            for pid, name in holders:
+                seen.setdefault(pid, name)
+            listed = ", ".join(f"{name} (pid {pid})" for pid, name in seen.items())
+            msg += f" Files are still locked open by: {listed}."
+        elif holders_error is not None:
+            msg += f" The processes holding them open could not be determined: {holders_error}."
+        else:
+            msg += " No running process was found holding them open."
+    msg += f" Retrying in {retry_in:g} s." if retry_in is not None else " Giving up."
+    return msg
+
 
 def _prepare_temp_dir_root_posix(temp_dir: str) -> None:
     """Validate the shared temporary root and set its mode, operating on a single
@@ -331,21 +395,76 @@ def __init__(
             rmtree(self.path, ignore_errors=True)
             raise
 
-    def cleanup(self) -> None:
+    def cleanup(self, logger: Optional[LoggerAdapter] = None) -> None:
         """Deletes the temporary directory and all of its contents.
+
+        On Windows a failed deletion is retried after each delay in
+        ``_CLEANUP_RETRY_DELAYS_SECONDS``, because a process that has just exited,
+        or been terminated, can hold its file handles for a while afterwards. Each
+        failed attempt is logged as a warning naming the files that could not be
+        deleted and the processes holding them open.
+
+        Args:
+            logger: Where to log failed attempts. Defaults to the module logger.
+
         Raises:
             RuntimeError - If not all files could be deleted. The message names
                 each path that could not be deleted along with the reason.
         """
-        failures: list[str] = []
+        existed = os.path.lexists(self.path)
+        failures = self._delete_tree()
+        retry_windows = is_windows() and existed
+        retry_delays = _CLEANUP_RETRY_DELAYS_SECONDS if retry_windows else ()
+        for delay in retry_delays:
+            if not failures:
+                break
+            self._log_failed_attempt(failures, delay, logger)
+            time.sleep(delay)
+            if not os.path.lexists(self.path):
+                # The pass that reported a failure still removed the directory.
+                failures = []
+                break
+            failures = self._delete_tree()
+
+        if failures:
+            if retry_windows:
+                self._log_failed_attempt(failures, None, logger)
+            raise RuntimeError(
+                f"Files within temporary directory {str(self.path)} could not be deleted.\n"
+                + "\n".join(f"{path}: {error!r}" for path, error in failures)
+            )
+
+    def _log_failed_attempt(
+        self,
+        failures: list[tuple[str, BaseException]],
+        retry_in: Optional[float],
+        logger: Optional[LoggerAdapter],
+    ) -> None:
+        files = [p for p, _ in failures if os.path.isfile(p)][:_MAX_FILES_TO_INSPECT]
+        holders, holders_error = _find_lock_holders(files) if files else ([], None)
+        message = _describe_failed_attempt(
+            self.path, failures[0][1], files, holders, holders_error, retry_in
+        )
+        (logger if logger is not None else LOG).warning(
+            message,
+            extra=LogExtraInfo(
+                openjd_log_content=LogContent.FILE_PATH
+                | LogContent.PROCESS_CONTROL
+                | LogContent.EXCEPTION_INFO
+            ),
+        )
+
+    def _delete_tree(self) -> list[tuple[str, BaseException]]:
+        """One recursive deletion pass. Returns each path that could not be deleted, with the error."""
+        failures: list[tuple[str, BaseException]] = []
 
         def _record(path: Any, error: BaseException) -> None:
             # R5-7 fix: keep the reason. Cleanup is exactly where the difference
             # between "permission denied" and "a process still holds this open"
             # decides what the operator should do next, and the old handler
             # accepted the exception and discarded it -- leaving a list of bare
             # paths with no cause.
-            failures.append(f"{path}: {error!r}")
+            failures.append((str(path), error))
 
         if sys.version_info >= (3, 12):
             # `onerror` is deprecated from 3.12 and slated for removal; `onexc`
@@ -367,8 +486,4 @@ def onerror(
 
             rmtree(self.path, onerror=onerror)
 
-        if failures:
-            raise RuntimeError(
-                f"Files within temporary directory {str(self.path)} could not be deleted.\n"
-                + "\n".join(failures)
-            )
+        return failures

src/openjd/sessions/_win32/_file_locks.py

@@ -0,0 +1,124 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""Find the processes holding files open, via the Windows Restart Manager.
+
+Used only to explain a failed cleanup in the log, so every failure here is
+raised as OSError for the caller to report, and never stops the cleanup.
+"""
+
+import sys
+from ctypes import POINTER, Structure, WinDLL, byref, c_int, create_unicode_buffer  # type: ignore
+from ctypes.wintypes import BOOL, DWORD, FILETIME, LPCWSTR, LPDWORD, LPWSTR, UINT, ULONG, WCHAR
+from typing import NamedTuple, Sequence
+
+import psutil
+
+# This assertion short-circuits mypy from type checking this module on platforms other than Windows
+# https://mypy.readthedocs.io/en/stable/common_issues.html#python-version-and-system-platform-checks
+assert sys.platform == "win32"
+
+CCH_RM_SESSION_KEY = 32
+CCH_RM_MAX_APP_NAME = 255
+CCH_RM_MAX_SVC_NAME = 63
+ERROR_SUCCESS = 0
+ERROR_MORE_DATA = 234
+
+
+class RM_UNIQUE_PROCESS(Structure):
+    _fields_ = [("dwProcessId", DWORD), ("ProcessStartTime", FILETIME)]
+
+
+class RM_PROCESS_INFO(Structure):
+    _fields_ = [
+        ("Process", RM_UNIQUE_PROCESS),
+        ("strAppName", WCHAR * (CCH_RM_MAX_APP_NAME + 1)),
+        ("strServiceShortName", WCHAR * (CCH_RM_MAX_SVC_NAME + 1)),
+        ("ApplicationType", c_int),
+        ("AppStatus", ULONG),
+        ("TSSessionId", DWORD),
+        ("bRestartable", BOOL),
+    ]
+
+
+rstrtmgr = WinDLL("rstrtmgr")
+
+# https://learn.microsoft.com/en-us/windows/win32/api/restartmanager/nf-restartmanager-rmstartsession
+rstrtmgr.RmStartSession.restype = DWORD
+rstrtmgr.RmStartSession.argtypes = [LPDWORD, DWORD, LPWSTR]
+# https://learn.microsoft.com/en-us/windows/win32/api/restartmanager/nf-restartmanager-rmregisterresources
+rstrtmgr.RmRegisterResources.restype = DWORD
+rstrtmgr.RmRegisterResources.argtypes = [
+    DWORD,
+    UINT,
+    POINTER(LPCWSTR),
+    UINT,
+    POINTER(RM_UNIQUE_PROCESS),
+    UINT,
+    POINTER(LPCWSTR),
+]
+# https://learn.microsoft.com/en-us/windows/win32/api/restartmanager/nf-restartmanager-rmgetlist
+rstrtmgr.RmGetList.restype = DWORD
+rstrtmgr.RmGetList.argtypes = [
+    DWORD,
+    POINTER(UINT),
+    POINTER(UINT),
+    POINTER(RM_PROCESS_INFO),
+    LPDWORD,
+]
+# https://learn.microsoft.com/en-us/windows/win32/api/restartmanager/nf-restartmanager-rmendsession
+rstrtmgr.RmEndSession.restype = DWORD
+rstrtmgr.RmEndSession.argtypes = [DWORD]
+
+
+class FileHolder(NamedTuple):
+    pid: int
+    name: str
+
+
+def processes_holding(files: Sequence[str]) -> list[FileHolder]:
+    """Returns the processes that hold any of ``files`` open.
+
+    Raises:
+        OSError: if the Restart Manager could not be queried.
+    """
+    if not files:
+        return []
+    handle = DWORD()
+    key = create_unicode_buffer(CCH_RM_SESSION_KEY + 1)
+    rc = rstrtmgr.RmStartSession(byref(handle), 0, key)
+    if rc != ERROR_SUCCESS:
+        raise OSError(f"RmStartSession failed ({rc})")
+    try:
+        return _list_holders(handle, files)
+    finally:
+        rstrtmgr.RmEndSession(handle)
+
+
+def _list_holders(handle: DWORD, files: Sequence[str]) -> list[FileHolder]:
+    names = (LPCWSTR * len(files))(*files)
+    rc = rstrtmgr.RmRegisterResources(handle, len(files), names, 0, None, 0, None)
+    if rc != ERROR_SUCCESS:
+        raise OSError(f"RmRegisterResources failed ({rc})")
+    infos = (RM_PROCESS_INFO * 0)()
+    # The list can grow between calls, so ask again a few times on ERROR_MORE_DATA.
+    for _ in range(3):
+        needed = UINT(0)
+        count = UINT(len(infos))
+        reasons = DWORD(0)
+        rc = rstrtmgr.RmGetList(handle, byref(needed), byref(count), infos, byref(reasons))
+        if rc == ERROR_SUCCESS:
+            return [_holder_from(infos[i]) for i in range(count.value)]
+        if rc != ERROR_MORE_DATA:
+            raise OSError(f"RmGetList failed ({rc})")
+        infos = (RM_PROCESS_INFO * needed.value)()
+    raise OSError("RmGetList kept reporting more processes")
+
+
+def _holder_from(info: RM_PROCESS_INFO) -> FileHolder:
+    pid = int(info.Process.dwProcessId)
+    try:
+        # The executable's file name, such as UnrealShaderCompileWorker.exe.
+        name = psutil.Process(pid).name()
+    except psutil.Error:
+        name = info.strAppName
+    return FileHolder(pid=pid, name=name)