🧪 test(fork): fork once the event loop has closed

tests/test_fork_coordination.py

@@ -780,13 +780,14 @@ def test_cancelled_async_acquire_unregisters_reused_descriptor(tmp_path: Path) -
 import os
 import sys
 import threading
+import time
 import warnings
 
 from filelock import AsyncFileLock
 
 warnings.filterwarnings("ignore", message=".*multi-threaded, use of fork.*", category=DeprecationWarning)
 
-async def main() -> int:
+async def main() -> tuple[int, tuple[int, int]]:
     callback_started = asyncio.Event()
     finish_callback = threading.Event()
     loop = asyncio.get_running_loop()
@@ -812,50 +813,55 @@ def block_after_acquire(fd: int) -> None:
     except asyncio.CancelledError:
         pass
     else:
-        return 1
+        raise SystemExit(1)
     if lock.is_locked or lock.lock_counter != 0:
-        return 2
-
-    path_stat = os.stat(sys.argv[1])
-    if (path_stat.st_dev, path_stat.st_ino) != identity:
-        return 3
-
-    occupant = os.open(os.devnull, os.O_RDONLY)
-    if occupant != descriptor:
-        os.dup2(occupant, descriptor)
-    source = os.open(sys.argv[1], os.O_RDWR)
-    if source == descriptor:
-        return 4
-    if (source_stat := os.fstat(source)).st_dev != identity[0] or source_stat.st_ino != identity[1]:
-        return 5
-    os.dup2(source, descriptor)
-    os.close(source)
-    if occupant != descriptor:
-        os.close(occupant)
-
-    # The canceled acquire ran the flock backend in the default executor, leaving an idle worker thread behind. Join it
-    # before forking so the child is single-threaded: NetBSD deadlocks a child forked from a multi-threaded process.
-    await loop.shutdown_default_executor()
+        raise SystemExit(2)
+    return descriptor, identity
+
+descriptor, identity = asyncio.run(main())
+
+path_stat = os.stat(sys.argv[1])
+if (path_stat.st_dev, path_stat.st_ino) != identity:
+    raise SystemExit(3)
+
+occupant = os.open(os.devnull, os.O_RDONLY)
+if occupant != descriptor:
+    os.dup2(occupant, descriptor)
+source = os.open(sys.argv[1], os.O_RDWR)
+if source == descriptor:
+    raise SystemExit(4)
+if (source_stat := os.fstat(source)).st_dev != identity[0] or source_stat.st_ino != identity[1]:
+    raise SystemExit(5)
+os.dup2(source, descriptor)
+os.close(source)
+if occupant != descriptor:
+    os.close(occupant)
+
+# The canceled acquire ran the flock backend in the default executor. asyncio.run() joins that worker on the way out,
+# but Thread.join() returns while the OS thread is still unwinding, and NetBSD deadlocks a child forked from a process
+# that still carries one (#707). So fork from module scope with the loop closed, leave the worker time to go, and
+# report a thread that is still counted rather than wedging the fork on it.
+time.sleep(0.5)
+if threading.active_count() != 1:
+    raise SystemExit(6)
 
-    child_pid = os.fork()
-    if child_pid == 0:
-        try:
-            replacement_stat = os.fstat(descriptor)
-        except OSError:
-            os._exit(1)
-        os._exit(0 if (replacement_stat.st_dev, replacement_stat.st_ino) == identity else 2)
-    _, status = os.waitpid(child_pid, 0)
-    os.close(descriptor)
-    return os.waitstatus_to_exitcode(status)
-
-raise SystemExit(asyncio.run(main()))
+child_pid = os.fork()
+if child_pid == 0:
+    try:
+        replacement_stat = os.fstat(descriptor)
+    except OSError:
+        os._exit(1)
+    os._exit(0 if (replacement_stat.st_dev, replacement_stat.st_ino) == identity else 2)
+_, status = os.waitpid(child_pid, 0)
+os.close(descriptor)
+raise SystemExit(os.waitstatus_to_exitcode(status))
 """
     result = subprocess.run(
         [sys.executable, "-c", script, str(tmp_path / "canceled.lock")],
         check=False,
         capture_output=True,
         text=True,
-        timeout=10,
+        timeout=30,
     )
 
     assert (result.returncode, result.stderr) == (0, "")