diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml index 3f3e928c9..38f8f42db 100644 --- a/.github/workflows/master.yml +++ b/.github/workflows/master.yml @@ -77,6 +77,15 @@ jobs: - name: Set up the environment uses: ./.github/actions/setup-python-env + - name: Check Windows worker types + run: >- + uv run --locked --no-sync ty check --python-platform win32 + src/powercontext/builtin/runtime/artifact_processing.py + tests/builtin/runtime/test_artifact_processing.py + + - name: Verify spawned worker lifecycle on Windows + run: uv run --locked --no-sync pytest -q tests/builtin/runtime/test_artifact_processing.py + - name: Exercise fixtures without Windows long-path support shell: pwsh run: | diff --git a/src/powercontext/builtin/runtime/artifact_processing.py b/src/powercontext/builtin/runtime/artifact_processing.py index 35ca87a57..350cfa708 100644 --- a/src/powercontext/builtin/runtime/artifact_processing.py +++ b/src/powercontext/builtin/runtime/artifact_processing.py @@ -29,7 +29,6 @@ from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from enum import StrEnum -from multiprocessing.connection import Connection from multiprocessing.context import SpawnProcess from multiprocessing.process import BaseProcess from random import SystemRandom @@ -43,8 +42,10 @@ from typing_extensions import override if sys.platform == "win32": + from multiprocessing.connection import PipeConnection as Connection from multiprocessing.popen_spawn_win32 import Popen as SpawnPopen else: + from multiprocessing.connection import Connection from multiprocessing.popen_spawn_posix import Popen as SpawnPopen from powercontext._logging import log_safely diff --git a/tests/builtin/runtime/test_artifact_processing.py b/tests/builtin/runtime/test_artifact_processing.py index ea87f3858..0ff810d82 100644 --- a/tests/builtin/runtime/test_artifact_processing.py +++ b/tests/builtin/runtime/test_artifact_processing.py @@ -15,8 +15,10 @@ from __future__ import annotations import asyncio +import ctypes import os import signal +import sys import threading import time from pathlib import Path @@ -81,6 +83,35 @@ async def _wait_until(predicate, *, timeout_seconds: float = 3.0) -> None: await asyncio.sleep(0.01) +def _assert_child_exited(pid: int) -> None: + if sys.platform != "win32": + with pytest.raises(ProcessLookupError): + os.kill(pid, 0) + return + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel32.OpenProcess.restype = wintypes.HANDLE + kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel32.WaitForSingleObject.restype = wintypes.DWORD + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + handle = kernel32.OpenProcess(0x00100000, False, pid) # SYNCHRONIZE; never terminate during observation. + if not handle: + error = ctypes.get_last_error() + if error == 87: # ERROR_INVALID_PARAMETER: PID no longer exists. + return + raise ctypes.WinError(error) + try: + result = kernel32.WaitForSingleObject(handle, 0) + if result == 0xFFFFFFFF: + raise ctypes.WinError(ctypes.get_last_error()) + assert result == 0 + finally: + kernel32.CloseHandle(handle) + + def test_spawn_launcher_runs_a_real_child_process() -> None: async def scenario() -> None: assignment = ArtifactProcessingWorkAssignment( @@ -138,12 +169,7 @@ async def child_is_ready() -> bool: assert child_pid is not None await asyncio.wait_for(supervisor.close(), timeout=SPAWN_TEST_TIMEOUT_SECONDS) - child_reaped = False - try: - os.kill(child_pid, 0) - except ProcessLookupError: - child_reaped = True - assert child_reaped + _assert_child_exited(child_pid) async with ArtifactProcessingSupervisor( database=profile.database, @@ -164,6 +190,7 @@ async def child_is_ready() -> bool: asyncio.run(scenario()) +@pytest.mark.skipif(sys.platform == "win32", reason="Injects failure into POSIX Popen._launch") def test_supervisor_close_owns_cancelled_post_spawn_cleanup(tmp_path, monkeypatch) -> None: async def scenario() -> None: config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'spawn-cancel-close.db'}") @@ -209,12 +236,7 @@ def stalled_launch(popen, process): finally: release_start.set() await asyncio.wait_for(close_task, timeout=SPAWN_TEST_TIMEOUT_SECONDS) - child_reaped = False - try: - os.kill(child_pid, 0) - except ProcessLookupError: - child_reaped = True - assert child_reaped + _assert_child_exited(child_pid) async with ArtifactProcessingSupervisor( database=profile.database, @@ -231,6 +253,7 @@ def stalled_launch(popen, process): asyncio.run(scenario()) +@pytest.mark.skipif(sys.platform == "win32", reason="Injects failure into POSIX Popen._launch") def test_supervisor_close_waits_for_cleanup_after_worker_start_timeout(tmp_path, monkeypatch) -> None: async def scenario() -> None: config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'spawn-timeout-close.db'}") @@ -308,8 +331,9 @@ async def cleanup_started() -> bool: ] with pytest.raises(ProcessLookupError): os.kill(child_pid, 0) - with pytest.raises(ChildProcessError): - await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG) + if sys.platform != "win32": + with pytest.raises(ChildProcessError): + await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG) async with ArtifactProcessingSupervisor( database=profile.database, @@ -327,6 +351,7 @@ async def cleanup_started() -> bool: @pytest.mark.parametrize("cancel_start", [False, True], ids=["ordinary-failure", "cancelled-failure"]) +@pytest.mark.skipif(sys.platform == "win32", reason="Injects failure into POSIX Popen._launch") def test_spawn_launcher_reaps_child_when_popen_raises_before_publication(tmp_path, monkeypatch, cancel_start) -> None: async def scenario() -> None: ready = tmp_path / f"worker-ready-{cancel_start}" @@ -378,7 +403,55 @@ def failing_launch(popen, process): assert child_pid is not None with pytest.raises(ProcessLookupError): os.kill(child_pid, 0) - with pytest.raises(ChildProcessError): - await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG) + if sys.platform != "win32": + with pytest.raises(ChildProcessError): + await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG) + + asyncio.run(scenario()) + + +def test_cancelled_spawn_start_cleans_up_child_before_returning(tmp_path, monkeypatch) -> None: + async def scenario() -> None: + ready = tmp_path / "cancelled-start-ready" + launched = threading.Event() + release = threading.Event() + child_pid: int | None = None + real_start = artifact_processing_module._start_owned_spawn_process + + def delayed_start(process, owner): + nonlocal child_pid + real_start(process, owner) + child_pid = process.pid + launched.set() + release.wait(timeout=SPAWN_TEST_TIMEOUT_SECONDS) + + monkeypatch.setattr(artifact_processing_module, "_start_owned_spawn_process", delayed_start) + assignment = ArtifactProcessingWorkAssignment( + binding_name=BINDING, + scope_id=str(ready), + artifact_family="topic-memory", + claimed_request_generation=1, + fence=ArtifactProcessingFence( + supervisor_group="global", + holder_id="holder-a", + supervisor_generation=1, + lease_mode="single-process", + ), + worker_id="00000000-0000-4000-8000-000000000001", + ) + start_task = asyncio.create_task(SpawnArtifactProcessingWorkerLauncher(_ignore_sigterm).start(assignment)) + try: + assert await asyncio.to_thread(launched.wait, SPAWN_TEST_TIMEOUT_SECONDS) + start_task.cancel() + await asyncio.sleep(0.05) + assert not start_task.done() + finally: + release.set() + if not start_task.done() and not start_task.cancelling(): + start_task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(start_task, timeout=SPAWN_TEST_TIMEOUT_SECONDS) + assert child_pid is not None + _assert_child_exited(child_pid) asyncio.run(scenario())