ci: make Windows green — skip POSIX-only paths and use msvcrt locks

- tests/test_shm.py: skip the whole module on win32. Shared-memory
  transport is POSIX-only (relies on mmap semantics that don't carry
  over to Windows page handling); the user does not need shm there.
- tests/test_benchmarks_java.py: replace `socket.AF_UNIX` with a
  runtime `getattr` so mypy on Windows stops choking. The benchmark
  itself never runs without an explicitly built worker, so the
  RuntimeError on Windows is acceptable.
- vgi_rpc/conformance/_types.py: factor `_probe_file_lock` so the
  cross-process cancel-probe counter file uses `fcntl.flock` on
  POSIX and `msvcrt.locking` on Windows. Same advisory semantics on
  both.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_benchmarks_java.py

@@ -57,10 +57,13 @@ def _wait_for_http(port: int, timeout: float = 10.0) -> None:
 
 
 def _wait_for_unix(path: str, timeout: float = 10.0) -> None:
+    af_unix = getattr(socket, "AF_UNIX", None)
+    if af_unix is None:
+        raise RuntimeError("Unix domain sockets are not supported on this platform")
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:
         try:
-            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
+            with socket.socket(af_unix, socket.SOCK_STREAM) as s:
                 s.connect(path)
                 return
         except (FileNotFoundError, ConnectionRefusedError):

tests/test_shm.py

@@ -8,6 +8,7 @@
 import gc
 import logging
 import struct
+import sys
 import threading
 from dataclasses import dataclass
 from typing import Protocol
@@ -42,6 +43,8 @@
 )
 from vgi_rpc.utils import ArrowSerializableDataclass
 
+pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="shared-memory transport is POSIX-only")
+
 # ---------------------------------------------------------------------------
 # Test schema & helpers
 # ---------------------------------------------------------------------------

vgi_rpc/conformance/_types.py

@@ -9,11 +9,15 @@
 
 from __future__ import annotations
 
+import contextlib
 import datetime as _dt
+import sys
+import time
+from collections.abc import Iterator
 from dataclasses import dataclass
 from decimal import Decimal
 from enum import Enum
-from typing import Annotated, Any, cast
+from typing import IO, Annotated, Any, cast
 
 import pyarrow as pa
 import pyarrow.compute as pc
@@ -411,6 +415,40 @@ def exchange(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext
 # ---------------------------------------------------------------------------
 
 
+@contextlib.contextmanager
+def _probe_file_lock(f: IO[str], *, exclusive: bool) -> Iterator[None]:
+    """Cross-platform advisory file lock for the conformance probe file.
+
+    Uses ``fcntl.flock`` on POSIX and ``msvcrt.locking`` on Windows so the
+    cross-process probe-counter file works on both platforms.
+    """
+    if sys.platform == "win32":
+        import msvcrt
+
+        f.seek(0)
+        while True:
+            try:
+                msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
+                break
+            except OSError:
+                time.sleep(0.01)
+        try:
+            yield
+        finally:
+            with contextlib.suppress(OSError):
+                f.seek(0)
+                msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+    else:
+        import fcntl
+
+        flag = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
+        fcntl.flock(f.fileno(), flag)
+        try:
+            yield
+        finally:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+
 class _CancelProbe:
     """Process-wide counters observed by cancel conformance tests.
 
@@ -440,41 +478,31 @@ def _bump(cls, field: str) -> None:
         if path is None:
             setattr(cls, field, getattr(cls, field) + 1)
             return
-        import fcntl
         import json
 
-        with open(path, "a+") as f:
-            fcntl.flock(f.fileno(), fcntl.LOCK_EX)
-            try:
-                f.seek(0)
-                raw = f.read()
-                data = json.loads(raw) if raw else {"produce_calls": 0, "exchange_calls": 0, "on_cancel_calls": 0}
-                data[field] = data.get(field, 0) + 1
-                f.seek(0)
-                f.truncate()
-                f.write(json.dumps(data))
-                f.flush()
-            finally:
-                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+        with open(path, "a+") as f, _probe_file_lock(f, exclusive=True):
+            f.seek(0)
+            raw = f.read()
+            data = json.loads(raw) if raw else {"produce_calls": 0, "exchange_calls": 0, "on_cancel_calls": 0}
+            data[field] = data.get(field, 0) + 1
+            f.seek(0)
+            f.truncate()
+            f.write(json.dumps(data))
+            f.flush()
 
     @classmethod
     def snapshot(cls) -> tuple[int, int, int]:
         """Return ``(produce, exchange, on_cancel)`` from the active store."""
         path = cls._shared_path()
         if path is None:
             return cls.produce_calls, cls.exchange_calls, cls.on_cancel_calls
-        import fcntl
         import json
         import os
 
         if not os.path.exists(path):
             return 0, 0, 0
-        with open(path) as f:
-            fcntl.flock(f.fileno(), fcntl.LOCK_SH)
-            try:
-                raw = f.read()
-            finally:
-                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+        with open(path) as f, _probe_file_lock(f, exclusive=False):
+            raw = f.read()
         if not raw:
             return 0, 0, 0
         data = json.loads(raw)
@@ -488,16 +516,11 @@ def reset(cls) -> None:
         cls.on_cancel_calls = 0
         path = cls._shared_path()
         if path is not None:
-            import fcntl
             import json
 
-            with open(path, "w") as f:
-                fcntl.flock(f.fileno(), fcntl.LOCK_EX)
-                try:
-                    f.write(json.dumps({"produce_calls": 0, "exchange_calls": 0, "on_cancel_calls": 0}))
-                    f.flush()
-                finally:
-                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+            with open(path, "w") as f, _probe_file_lock(f, exclusive=True):
+                f.write(json.dumps({"produce_calls": 0, "exchange_calls": 0, "on_cancel_calls": 0}))
+                f.flush()
 
 
 @dataclass