Skip `test_ringbuf` at collection off-linux

`tests/test_ringbuf.py` imports `tractor.ipc._ringbuf` at module
top, which pulls in `tractor.ipc._linux` whose module-level
`ffi.dlopen(None)` raises `OSError` on Windows (and any non-linux
host). That fires at COLLECTION, before the module's existing
`pytestmark = pytest.mark.skip` can apply, so it aborts the whole
pytest session — the new `windows-latest` CI leg never gets past
collection.

- guard the module with `pytest.skip(allow_module_level=True)`
  gated on `platform.system() != 'Linux'`, placed before the
  crashing import — same idiom as `tests/devx/test_debugger.py`.
- the `eventfd`-based ringbuf backend is linux-only by design, so
  macOS skips cleanly too (previously it only skipped incidentally
  via the absent `cffi` optional dep).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

.github/workflows/ci.yml

@@ -91,6 +91,11 @@ jobs:
     name: '${{ matrix.os }} Python${{ matrix.python-version }} spawn_backend=${{ matrix.spawn_backend }} tpt_proto=${{ matrix.tpt_proto }}'
     timeout-minutes: 16
     runs-on: ${{ matrix.os }}
+    # Windows support is nascent: keep its legs informational so a
+    # known-incomplete area doesn't block merges. The `import
+    # tractor` smoke step below is the hard signal for this job;
+    # promote the whole leg to required once the suite is green.
+    continue-on-error: ${{ matrix.os == 'windows-latest' }}
 
     strategy:
       fail-fast: false
@@ -98,6 +103,7 @@ jobs:
         os: [
           ubuntu-latest,
           macos-latest,
+          windows-latest,
         ]
         python-version: [
           '3.13',
@@ -123,6 +129,10 @@ jobs:
           # don't do UDS run on macOS (for now)
           - os: macos-latest
             tpt_proto: 'uds'
+          # UDS is POSIX-only; Windows has no `AF_UNIX` so the
+          # backend is intentionally unavailable there.
+          - os: windows-latest
+            tpt_proto: 'uds'
 
     steps:
       - uses: actions/checkout@v4
@@ -150,6 +160,12 @@ jobs:
       - name: List deps tree
         run: uv tree
 
+      # hard signal for the Windows import-safety fix: `import
+      # tractor` must succeed everywhere, and `HAS_UDS` reflects
+      # platform capability (False on Windows, True on POSIX).
+      - name: 'Smoke: import tractor'
+        run: uv run python -c "import tractor; from tractor.ipc._uds import HAS_UDS; print('import tractor OK | HAS_UDS=', HAS_UDS)"
+
       - name: Run tests
         run: >
           uv run

scripts/cpu-perf-check

@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+# tractor: distributed structured concurrency.
+# Copyright 2018-eternity Tyler Goodlet.
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+'''
+`cpu-perf-check` — sustained-load CPU throttle detector.
+
+Idle freq snapshots LIE. A laptop can read
+`governor=performance`, `EPP=performance`,
+`platform_profile=performance`, `scaling_max_freq=<full>`
+and momentarily clock a P-core at 5GHz — while a
+firmware/EC power cap (AMD PPT/STAPM and friends) clamps
+the whole package to ~1.5GHz the instant a sustained
+multi-core load lands. That throttle masquerades as a
+`trio`-backend test *regression*: a wave of `fail_after` /
+`TooSlowError` / `Cancelled(source='deadline')` deadline
+misses on spawn-heavy tests, on byte-identical code that
+was green yesterday.
+
+The existing `tests/conftest.py:cpu_scaling_factor()` only
+reads STATIC `scaling_max_freq` vs `*_pstate_max_freq`, so
+it returns `1.0` (no throttle) during exactly this failure
+— it can't see the cap. This script complements it by
+BURNING every core for a few seconds and sampling the
+ACHIEVED `scaling_cur_freq`, which is the only thing that
+exposes the clamp.
+
+Exit code: `0` if sustained perf looks restored, `1` if
+throttled — so it gates a test run:
+
+    py313/bin/python scripts/cpu-perf-check && pytest tests/ ...
+
+Tunables (env-overridable):
+    CPU_PERF_SECS        load duration       (default 4.0)
+    CPU_PERF_HEALTHY_FRAC sustained/max floor (default 0.45)
+
+'''
+from __future__ import annotations
+import glob
+import os
+import time
+import multiprocessing as mp
+
+
+def _read(path: str) -> str | None:
+    try:
+        with open(path) as f:
+            return f.read().strip()
+    except OSError:
+        return None
+
+
+def _cur_freqs_mhz() -> list[int]:
+    out: list[int] = []
+    for f in glob.glob(
+        '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
+    ):
+        if (v := _read(f)):
+            out.append(int(v) // 1000)
+    return out
+
+
+def _pkg_max_mhz() -> int:
+    '''
+    Highest per-core ceiling across the package — the
+    P-core max on hybrid parts.
+
+    '''
+    mxs: list[int] = []
+    for f in glob.glob(
+        '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
+    ):
+        if (v := _read(f)):
+            mxs.append(int(v) // 1000)
+    return max(mxs) if mxs else 0
+
+
+def _burn(stop: float) -> None:
+    # fixed-width (64-bit-masked) arithmetic keeps this a steady
+    # ALU load; an unmasked `x` grows ~x**2/iter into a huge bigint,
+    # degenerating into noisy alloc/mul-bound work (and needless
+    # memory) across N procs.
+    x: int = 1
+    while time.perf_counter() < stop:
+        x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF
+
+
+def main(
+    secs: float = float(os.environ.get('CPU_PERF_SECS', 4.0)),
+    # sustained aggregate must clear this fraction of the
+    # package max-freq ceiling. Throttled (~1.5GHz of 5.1GHz)
+    # ~= 0.29; a healthy all-core load easily clears 0.5.
+    healthy_frac: float = float(
+        os.environ.get('CPU_PERF_HEALTHY_FRAC', 0.45)
+    ),
+) -> int:
+    if not glob.glob('/sys/devices/system/cpu/cpu0/cpufreq'):
+        print('no cpufreq sysfs (non-linux?) — skipping, assume OK')
+        return 0
+
+    b: str = '/sys/devices/system/cpu/cpu0/cpufreq/'
+    pkg_max: int = _pkg_max_mhz()
+    print('=== static knobs (ALL can read fine while throttled) ===')
+    print(f'  governor          : {_read(b + "scaling_governor")}')
+    print(f'  EPP               : {_read(b + "energy_performance_preference")}')
+    print(f'  platform_profile  : '
+          f'{_read("/sys/firmware/acpi/platform_profile")}')
+    print(f'  pkg max freq      : {pkg_max} MHz')
+
+    ncpu: int = os.cpu_count() or 1
+    print(f'\n=== sustained {ncpu}-core load ({secs:.0f}s) — the real test ===')
+    stop: float = time.perf_counter() + secs
+    procs = [
+        mp.Process(target=_burn, args=(stop,), daemon=True)
+        for _ in range(ncpu)
+    ]
+    for p in procs:
+        p.start()
+
+    # skip the initial ~0.6s ramp, then sample steady-state.
+    # `try/finally` so a Ctrl-C / sampling error still reaps the
+    # burners instead of leaving stray CPU hogs (daemon=True is the
+    # backstop should we exit abnormally before the join).
+    samples: list[int] = []
+    try:
+        time.sleep(0.6)
+        while time.perf_counter() < stop - 0.2:
+            if (fr := _cur_freqs_mhz()):
+                samples.append(sum(fr) // len(fr))
+            time.sleep(0.3)
+    finally:
+        for p in procs:
+            p.terminate()
+        for p in procs:
+            p.join()
+
+    if not (samples and pkg_max):
+        print('  could not sample cur freq — assume OK')
+        return 0
+
+    sustained: int = sum(samples) // len(samples)
+    frac: float = sustained / pkg_max
+    print(f'  aggregate cur-freq samples: {samples}')
+    print(f'  sustained avg             : {sustained} MHz '
+          f'({frac * 100:.0f}% of {pkg_max} MHz max)')
+
+    if frac < healthy_frac:
+        print(
+            f'\n  ❌ THROTTLED — sustained {sustained}MHz is only '
+            f'{frac * 100:.0f}% of max (< {healthy_frac * 100:.0f}%).\n'
+            f'     Power cap (PPT/STAPM) still engaged. Fixes:\n'
+            f'       - bounce /sys/firmware/acpi/platform_profile\n'
+            f'         (balanced -> performance)\n'
+            f'       - unplug/replug USB-C to re-negotiate PD\n'
+            f'       - ryzenadj to lift STAPM/PPT\n'
+            f'       - else reboot\n'
+            f'     Do NOT bump test budgets — the box is slow, not the code.'
+        )
+        return 1
+
+    print(
+        f'\n  ✅ PERF OK — sustained {sustained}MHz holds '
+        f'{frac * 100:.0f}% of max. Cap looks lifted; safe to run tests.'
+    )
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

tests/conftest.py

@@ -134,6 +134,150 @@ def cpu_scaling_factor() -> float:
     return factor
 
 
+# session-cached sustained-load throttle multiplier — measured
+# once (lazily) on the first `cpu_perf_headroom()` call. `None`
+# = not-yet-measured.
+_sustained_headroom: float|None = None
+
+
+def _measure_sustained_headroom(
+    secs: float = 0.9,
+    # a healthy all-core sustained clock holds AT/ABOVE this
+    # fraction of the package single-core max ceiling (boost sags
+    # under full multi-core load even un-throttled, but not far);
+    # at/above it we assume no throttle and return 1.0.
+    throttle_gate: float = 0.6,
+    max_headroom: float = 4.,
+) -> float:
+    '''
+    One-shot all-core burn returning a latency multiplier
+    (>= 1.0) that reflects *sustained-load* CPU throttle.
+
+    Catches the firmware/EC power-cap clamp (AMD PPT/STAPM &
+    friends) that pins achieved `scaling_cur_freq` to a fraction
+    of the ceiling under multi-core load while EVERY static knob
+    (`governor`, `scaling_max_freq`, `EPP`, `platform_profile`)
+    still reads "full performance". That cap is INVISIBLE to
+    `cpu_scaling_factor()` and is the gremlin behind mass `trio`
+    deadline-miss failures on byte-identical code — see
+    `scripts/cpu-perf-check`.
+
+    Best-effort: returns 1.0 on non-linux / missing sysfs / any
+    error so it can never break a test run.
+
+    '''
+    import glob
+    import multiprocessing as mp
+
+    def _read_mhz(path: str) -> int|None:
+        try:
+            with open(path) as f:
+                return int(f.read()) // 1000
+        except OSError:
+            return None
+
+    try:
+        maxs: list[int] = [
+            v for f in glob.glob(
+                '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
+            )
+            if (v := _read_mhz(f)) is not None
+        ]
+        pkg_max: int = max(maxs) if maxs else 0
+        if not pkg_max:
+            return 1.
+
+        def _burn(stop: float) -> None:
+            # fixed-width (64-bit-masked) arithmetic keeps this a
+            # steady ALU load; an unmasked `x` grows ~x**2/iter into
+            # a huge bigint, degenerating into noisy alloc/mul-bound
+            # work (and needless memory) across N procs.
+            x: int = 1
+            while time.perf_counter() < stop:
+                x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF
+
+        # explicit `fork` ctx so we're immune to whatever global
+        # mp start-method tractor/the suite may have set (`spawn`
+        # would re-exec + re-import 24x — slow and pointless here).
+        ctx = mp.get_context('fork')
+        ncpu: int = os.cpu_count() or 1
+        stop: float = time.perf_counter() + secs
+        procs = [
+            ctx.Process(target=_burn, args=(stop,), daemon=True)
+            for _ in range(ncpu)
+        ]
+        for p in procs:
+            p.start()
+
+        # skip the ~0.4s boost window so we sample the steady
+        # state AFTER any power-cap has engaged.
+        samples: list[int] = []
+        time.sleep(0.4)
+        while time.perf_counter() < stop - 0.1:
+            curs: list[int] = [
+                v for f in glob.glob(
+                    '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
+                )
+                if (v := _read_mhz(f)) is not None
+            ]
+            if curs:
+                samples.append(sum(curs) // len(curs))
+            time.sleep(0.15)
+        for p in procs:
+            p.join()
+
+        if not samples:
+            return 1.
+        frac: float = (sum(samples) // len(samples)) / pkg_max
+        # below the gate we read it as a power-cap throttle. The
+        # spawn/IPC/fork-bound work these budgets guard slows ~1:1
+        # with the achieved-vs-max freq ratio, so compensate by the
+        # FULL inverse fraction (a boost-discounted factor
+        # under-shoots and still trips the marginal cases).
+        if frac >= throttle_gate:
+            return 1.
+        # a 0/parked-core freq read would `ZeroDivisionError` the
+        # inverse below — silently swallowed by the outer `except`
+        # into a 1.0 (no-throttle), defeating the probe on exactly
+        # the broken box it should flag; read 0 as max throttle.
+        if frac <= 0:
+            return max_headroom
+        return min(max_headroom, 1. / frac)
+
+    except Exception:
+        return 1.
+
+
+def cpu_perf_headroom() -> float:
+    '''
+    Latency-headroom multiplier (>= 1.0) covering BOTH cpu-perf
+    throttle classes — multiply a test's deadline by it, e.g.
+    `timeout *= cpu_perf_headroom()`:
+
+      - static cpu-freq scaling — via `cpu_scaling_factor()`
+        (governor/policy lowered the `scaling_max_freq` ceiling).
+
+      - sustained-load power-cap throttle — via
+        `_measure_sustained_headroom()` (firmware/EC PPT/STAPM
+        clamps achieved freq under load while every static knob
+        reads "performance"; INVISIBLE to the static check). This
+        is the gremlin behind mass `trio` deadline-miss failures
+        on unchanged code — see
+        `ai/conc-anal/trio_033_cancel_cascade_slowdown_depth3_issue.md`.
+
+    The sustained probe runs ONCE per session (cached); the cost
+    is a ~0.9s all-core burn on first call only.
+
+    '''
+    global _sustained_headroom
+    static: float = cpu_scaling_factor()
+    if _non_linux:
+        return static
+    if _sustained_headroom is None:
+        _sustained_headroom = _measure_sustained_headroom()
+    return max(static, _sustained_headroom)
+
+
 # NOTE, the `--ll`/`--tl` CLI flags + the `loglevel`, `test_log`
 # and `testing_pkg_name` fixtures have been factored into the
 # `tractor._testing.pytest` plugin (loaded via the `-p` entry in

tests/devx/test_debugger.py

@@ -794,6 +794,14 @@ def test_multi_nested_subactors_error_through_nurseries(
         loglevel='pdb',
     )
     last_send_char: str|None = None
+
+    # inflate pexpect waits under CPU throttle — incl. the
+    # sustained-load power-cap invisible to static freq reads — so
+    # a slow-to-boot child REPL doesn't trip a false `TIMEOUT`.
+    # See `scripts/cpu-perf-check`.
+    from ..conftest import cpu_perf_headroom
+    headroom: float = cpu_perf_headroom()
+
     for (
         i,
         send_char,
@@ -817,6 +825,9 @@ def test_multi_nested_subactors_error_through_nurseries(
         if is_forking_spawner:
             timeout += 4
 
+        if headroom != 1.:
+            timeout *= headroom
+
         try:
             child.expect(
                 PROMPT,

tests/test_advanced_streaming.py

@@ -189,11 +189,18 @@ def test_dynamic_pub_sub(
     # sits forever until external SIGINT. The `afk_alarm_w_trace`
     # outer guard below is the AFK-safety counterpart (SIGALRM
     # raises in the main thread regardless of trio scope state).
-    fail_after_s: int = (
+    fail_after_s: float = (
         8
         if is_forking_spawner
         else 20
     )
+    # inflate under CPU throttle — incl. the sustained-load
+    # power-cap invisible to static freq reads — so a slow box
+    # doesn't trip the deadline. See `scripts/cpu-perf-check`.
+    from .conftest import cpu_perf_headroom
+    headroom: float = cpu_perf_headroom()
+    if headroom != 1.:
+        fail_after_s *= headroom
 
     async def main():
         # bug-class-3 breadcrumb: tag each level of the cancel path

tests/test_cancellation.py

@@ -596,6 +596,15 @@ async def test_nested_multierrors(
     # depth=3, BOTH variants will reliably `xpass` and
     # pytest will yell — our signal to drop the marker. See
     # `ai/conc-anal/cancel_cascade_too_slow_under_main_thread_forkserver_issue.md`.
+    #
+    # Probe CPU throttle ONCE up-front (folds in the sustained-load
+    # power-cap that static freq reads miss): used BOTH to inflate
+    # the deadline budget below AND to xfail depth=3, whose failure
+    # mode under throttle is a runtime-internal reap deadline — not
+    # a test-budget miss. See `scripts/cpu-perf-check`.
+    from .conftest import cpu_perf_headroom
+    headroom: float = cpu_perf_headroom()
+
     if start_method == 'main_thread_forkserver':
         request.node.add_marker(
             pytest.mark.xfail(
@@ -609,6 +618,34 @@ async def test_nested_multierrors(
             )
         )
 
+    # Under CPU throttle (incl. the sustained-load power-cap that
+    # static freq reads miss) the DEEP depth=3 tree trips tractor's
+    # INTERNAL reap deadlines (`soft_kill`/`hard_kill`
+    # `move_on_after`/`terminate_after=1.6`) before slow subprocs
+    # exit, injecting a `Cancelled(source='deadline')` into the BEG
+    # — the SAME shape-mismatch class as the MTF xfail above, and
+    # NOT fixable by inflating the test-level budget (the Cancelled
+    # is minted inside the runtime, not by our `fail_after`).
+    # xfail(strict=False) so it auto-clears the moment the box is
+    # un-throttled (`headroom == 1.`); depth=1's shallow tree stays
+    # under those internal deadlines so it just rides the budget
+    # inflation below. See `scripts/cpu-perf-check`.
+    elif (
+        depth == 3
+        and
+        headroom != 1.
+    ):
+        request.node.add_marker(
+            pytest.mark.xfail(
+                strict=False,
+                reason=(
+                    'CPU throttled — tractor reap deadline injects '
+                    'Cancelled into BEG; see conc-anal/'
+                    'trio_033_cancel_cascade_slowdown_depth3_issue.md'
+                ),
+            )
+        )
+
     # Per-backend/-depth budgets: in the non-hang case the
     # whole spawn + cancel-cascade should complete in well
     # under these. On the borderline hang case the
@@ -636,11 +673,14 @@ async def test_nested_multierrors(
         case ('main_thread_forkserver', 3):
             timeout = 30
 
-    # headroom for CPU-freq scaling AND/OR slow CI so the inner
+    # inflate the budget by the throttle headroom probed above so
-    # snapshot-capturing budget doesn't fire spuriously on a
+    # a slow box doesn't masquerade as a deadline regression.
-    # sluggish runner; see `cpu_scaling_factor()`.
+    # NOTE, `headroom = cpu_perf_headroom()` (set above) is the
-    from .conftest import cpu_scaling_factor
+    # SUPERSET of `cpu_scaling_factor()` — it folds in the static
-    timeout *= cpu_scaling_factor()
+    # cpu-freq scaling + slow-CI bump AND the sustained-load
+    # throttle probe this depth-3 cascade was the poster child for.
+    if headroom != 1.:
+        timeout *= headroom
 
     async with fail_after_w_trace(timeout):
         try:
@@ -748,7 +788,7 @@ def test_cancel_via_SIGINT_other_task(
     started from a seperate ``trio`` child  task.
 
     '''
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
 
     pid: int = os.getpid()
     timeout: float = (
@@ -758,8 +798,9 @@ def test_cancel_via_SIGINT_other_task(
     if _friggin_windows:  # smh
         timeout += 1
 
-    # add latency headroom for CPU freq scaling (auto-cpufreq et al.)
+    # latency headroom for static cpu-freq scaling + sustained-load
-    headroom: float = cpu_scaling_factor()
+    # throttle + CI (auto-cpufreq et al.); see `cpu_perf_headroom()`.
+    headroom: float = cpu_perf_headroom()
     if headroom != 1.:
         timeout *= headroom
 
@@ -962,11 +1003,11 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
     if _friggin_windows:  # smh
         timeout += 1
 
-    # CPU-scaling / CI latency headroom — macOS CI especially is
+    # CPU-scaling / sustained-throttle / CI latency headroom — macOS
-    # slow for this graceful-vs-hard-reap timing race; see
+    # CI especially is slow for this graceful-vs-hard-reap timing
-    # `cpu_scaling_factor()`.
+    # race; see `cpu_perf_headroom()`.
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
-    timeout *= cpu_scaling_factor()
+    timeout *= cpu_perf_headroom()
 
     async def main():
         start = time.time()

tests/test_clustering.py

@@ -24,8 +24,14 @@ def test_empty_mngrs_input_raises(
             'actor-cluster teardown hangs intermittently on UDS'
         )
 
+    # inflate under CPU throttle — incl. the sustained-load
+    # power-cap invisible to static freq reads. See
+    # `scripts/cpu-perf-check`.
+    from .conftest import cpu_perf_headroom
+    fail_after_s: float = 3 * cpu_perf_headroom()
+
     async def main():
-        with trio.fail_after(3):
+        with trio.fail_after(fail_after_s):
             async with (
                 open_actor_cluster(
                     modules=[__name__],
@@ -93,6 +99,13 @@ async def test_streaming_to_actor_cluster(
         10 if is_forking_spawner
         else 6
     )
+    # inflate under CPU throttle — incl. the sustained-load
+    # power-cap invisible to static freq reads. See
+    # `scripts/cpu-perf-check`.
+    from .conftest import cpu_perf_headroom
+    headroom: float = cpu_perf_headroom()
+    if headroom != 1.:
+        delay *= headroom
     with trio.fail_after(delay):
         async with (
             open_actor_cluster(modules=[__name__]) as portals,

tests/test_docs_examples.py

@@ -160,7 +160,7 @@ def test_example(
             'root-causing. Passes on Linux.'
         )
 
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
 
     timeout: float = (
         60
@@ -168,8 +168,9 @@ def test_example(
         else 16
     )
 
-    # add latency headroom for CPU freq scaling (auto-cpufreq et al.)
+    # add latency headroom for CPU freq scaling/throttle
-    headroom: float = cpu_scaling_factor()
+    # (auto-cpufreq et al.)
+    headroom: float = cpu_perf_headroom()
     if headroom != 1.:
         timeout *= headroom
 

tests/test_inter_peer_cancellation.py

@@ -24,7 +24,7 @@ from tractor._testing import (
     expect_ctxc,
 )
 
-from .conftest import cpu_scaling_factor
+from .conftest import cpu_perf_headroom
 
 pytestmark = [
     pytest.mark.skipon_spawn_backend(
@@ -1280,12 +1280,12 @@ def test_peer_spawns_and_cancels_service_subactor(
 
 
     async def _main():
-        headroom: float = cpu_scaling_factor()
+        headroom: float = cpu_perf_headroom()
         this_fast_on_linux: float = 3
         this_fast = this_fast_on_linux * headroom
         if headroom != 1.:
             test_log.warning(
-                f'Adding latency headroom on linux bc CPU scaling,\n'
+                f'Adding latency headroom on linux bc CPU perf scaling/throttle,\n'
                 f'headroom: {headroom}\n'
                 f'this_fast_on_linux: {this_fast_on_linux} -> {this_fast}\n'
             )

tests/test_legacy_one_way_streaming.py

@@ -326,11 +326,11 @@ def time_quad_ex(
     ):
         timeout += 1
 
-    # inflate the cancel-deadline for CPU-freq scaling AND/OR CI
+    # inflate the cancel-deadline for static cpu-freq scaling +
-    # latency (see `cpu_scaling_factor()`) so the example isn't
+    # sustained-load throttle + CI latency (see `cpu_perf_headroom()`)
-    # cancelled mid-stream on a throttled/CI runner.
+    # so the example isn't cancelled mid-stream on a throttled/CI box.
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
-    timeout *= cpu_scaling_factor()
+    timeout *= cpu_perf_headroom()
 
     start: float = time.time()
     results: list[int] = trio.run(partial(
@@ -379,8 +379,8 @@ def test_a_quadruple_example(
     # https://github.com/AdnanHodzic/auto-cpufreq?tab=readme-ov-file#example-config-file-contents
     #
     # HENCE this below latency-headroom compensation logic..
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
-    headroom: float = cpu_scaling_factor()
+    headroom: float = cpu_perf_headroom()
     if headroom != 1.:
         this_fast = this_fast_on_linux * headroom
         test_log.warning(

tests/test_resource_cache.py

@@ -213,12 +213,12 @@ def test_open_local_sub_to_stream(
     N local tasks using `trionics.maybe_open_context()`.
 
     '''
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
     timeout: float = (
         4
         if not platform.system() == "Windows"
         else 10
-    ) * cpu_scaling_factor()
+    ) * cpu_perf_headroom()
 
     if debug_mode:
         timeout = 999

tests/test_ringbuf.py

@@ -1,10 +1,22 @@
 import time
+import platform
 
 import trio
 import pytest
 
 import tractor
 
+# `tractor.ipc._ringbuf` is built on linux `eventfd(2)`; importing
+# it pulls in `tractor.ipc._linux` whose module-level
+# `ffi.dlopen(None)` raises on non-linux. Skip the whole module at
+# COLLECTION before that crashing import runs (a `pytestmark` skip
+# is too late — markers apply only after the import succeeds).
+if platform.system() != 'Linux':
+    pytest.skip(
+        'ringbuf (eventfd) IPC is linux-only',
+        allow_module_level=True,
+    )
+
 # XXX `cffi` dun build on py3.14 yet..
 pytest.importorskip("cffi")
 

tests/test_spawning.py

@@ -153,9 +153,9 @@ async def test_most_beautiful_word(
     # actor spawn + IPC round-trip is comfortably sub-second on a
     # warm box, but slow/noisy CI runners (esp. macOS) blow a flat
     # 1s deadline. Scale for CI/CPU-throttle headroom — `== 1s`
-    # locally where `cpu_scaling_factor()` is `1.0`.
+    # locally where `cpu_perf_headroom()` is `1.0`.
-    from .conftest import cpu_scaling_factor
+    from .conftest import cpu_perf_headroom
-    with trio.fail_after(1 * cpu_scaling_factor()):
+    with trio.fail_after(1 * cpu_perf_headroom()):
         async with tractor.open_nursery(
             debug_mode=debug_mode,
         ) as an:

tractor/_testing/addr.py

@@ -50,9 +50,14 @@ def get_rando_addr(
     fight over the bind, cascading into a chain of
     "Address already in use" failures.
 
-    For UDS this concern doesn't apply: `UDSAddress.get_random()`
+    For UDS the *cross*-process concern is handled by keying
-    already builds socket paths from `os.getpid()` so each
+    socket paths off `os.getpid()` (each pytest process gets its
-    pytest process gets its own socket-path namespace.
+    own namespace). *Within* a process — where `get_random()` has
+    no live runtime to name from — it also appends a per-call
+    `uuid4` token, so two calls (e.g. a `reg_addr` + a distinct
+    bind addr in one test body) yield distinct sockpaths rather
+    than aliasing to the same `name@pid.sock` and tripping
+    `EADDRINUSE`.
 
     '''
     addr_type: Type[_addr.Addres] = _addr._address_types[tpt_proto]

tractor/devx/_stackscope.py

@@ -31,12 +31,21 @@ from threading import (
     RLock,
 )
 import multiprocessing as mp
+
+import platform
+
 from signal import (
     signal,
     getsignal,
-    SIGUSR1,
     SIGINT,
 )
+
+
+if platform.system() != "Windows":
+    from signal import SIGUSR1
+else:
+    SIGUSR1 = None
+
 # import traceback
 from types import ModuleType
 from typing import (
@@ -347,8 +356,8 @@ def dump_tree_on_sig(
 
 
 def enable_stack_on_sig(
-    sig: int = SIGUSR1,
+    sig: int|None = SIGUSR1,
-) -> ModuleType:
+) -> ModuleType|None:
     '''
     Enable `stackscope` tracing on reception of a signal; by
     default this is SIGUSR1.
@@ -367,6 +376,16 @@ def enable_stack_on_sig(
     >> pkill --signal SIGUSR1 -f <part-of-cmd: str>
 
     '''
+    # no `SIGUSR1` on this platform (e.g. Windows) -> nothing to
+    # wire up; degrade gracefully instead of crashing callers that
+    # only guard against a missing `stackscope` (`ImportError`).
+    if sig is None:
+        log.warning(
+            'No `SIGUSR1` on this platform;\n'
+            'skipping `stackscope` trace-on-signal setup!\n'
+        )
+        return None
+
     try:
         # NOTE, `stackscope._glue` does intentional async-gen type
         # introspection at import-time which trips

tractor/discovery/_addr.py

@@ -32,14 +32,16 @@ from ..runtime._state import (
     _def_tpt_proto,
 )
 from ..ipc._tcp import TCPAddress
-from ..ipc._uds import UDSAddress
+from ..ipc._uds import (
+    UDSAddress,
+    HAS_UDS,
+)
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
 
 log = get_logger()
 
-
 # TODO, maybe breakout the netns key to a struct?
 # class NetNs(Struct)[str, int]:
 #     ...
@@ -172,19 +174,18 @@ class Address(Protocol):
 
 _address_types: bidict[str, Type[Address]] = {
     'tcp': TCPAddress,
-    'uds': UDSAddress
 }
+if HAS_UDS:
+    _address_types['uds'] = UDSAddress
 
 
 # TODO! really these are discovery sys default addrs ONLY useful for
 # when none is provided to a root actor on first boot.
-_default_lo_addrs: dict[
+_default_lo_addrs: dict[str, UnwrappedAddress] = {
-    str,
-    UnwrappedAddress
-] = {
     'tcp': TCPAddress.get_root().unwrap(),
-    'uds': UDSAddress.get_root().unwrap(),
 }
+if HAS_UDS:
+    _default_lo_addrs['uds'] = UDSAddress.get_root().unwrap()
 
 
 def get_address_cls(name: str) -> Type[Address]:

tractor/ipc/_server.py

@@ -62,16 +62,17 @@ from .. import log
 from ..discovery._addr import Address
 from ._chan import Channel
 from ._transport import MsgTransport
-from ._uds import UDSAddress
+
-from ._tcp import TCPAddress
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
     from ..runtime._supervise import ActorNursery
 
 
-log = log.get_logger()
+from ._tcp import TCPAddress
+from ._uds import UDSAddress
 
+log = log.get_logger()
 
 async def maybe_wait_on_canced_subs(
     uid: tuple[str, str],

tractor/ipc/_types.py

@@ -18,17 +18,13 @@
 IPC subsys type-lookup helpers?
 
 '''
-from typing import (
+from typing import Type
-    Type,
-    # TYPE_CHECKING,
-)
-
-import trio
 import socket
+import trio
 
 from tractor.ipc._transport import (
     MsgTransportKey,
-    MsgTransport
+    MsgTransport,
 )
 from tractor.ipc._tcp import (
     TCPAddress,
@@ -37,87 +33,86 @@ from tractor.ipc._tcp import (
 from tractor.ipc._uds import (
     UDSAddress,
     MsgpackUDSStream,
+    HAS_UDS,
 )
 
-# if TYPE_CHECKING:
+# the UDS backend is importable everywhere but only *usable* where
-#     from tractor._addr import Address
+# `trio` reports `has_unix` (i.e. POSIX). On Windows / no-`AF_UNIX`
-
+# hosts `HAS_UDS` is `False` and the runtime registers TCP only.
-
 Address = TCPAddress|UDSAddress
 
 # manually updated list of all supported msg transport types
-_msg_transports = [
+_msg_transports: list[Type[MsgTransport]] = [
     MsgpackTCPStream,
-    MsgpackUDSStream
 ]
+if HAS_UDS:
+    _msg_transports.append(MsgpackUDSStream)
 
-
+# map a `MsgTransportKey` to its `MsgTransport` type
-# convert a MsgTransportKey to the corresponding transport type
+_key_to_transport: dict[MsgTransportKey, Type[MsgTransport]] = {
-_key_to_transport: dict[
-    MsgTransportKey,
-    Type[MsgTransport],
-] = {
     ('msgpack', 'tcp'): MsgpackTCPStream,
-    ('msgpack', 'uds'): MsgpackUDSStream,
 }
+if HAS_UDS:
+    _key_to_transport[('msgpack', 'uds')] = MsgpackUDSStream
 
-# convert an Address wrapper to its corresponding transport type
+# map an `Address`-wrapper to its `MsgTransport` type
-_addr_to_transport: dict[
+_addr_to_transport: dict[Type[Address], Type[MsgTransport]] = {
-    Type[TCPAddress|UDSAddress],
-    Type[MsgTransport]
-] = {
     TCPAddress: MsgpackTCPStream,
-    UDSAddress: MsgpackUDSStream,
 }
+if HAS_UDS:
+    _addr_to_transport[UDSAddress] = MsgpackUDSStream
 
 
+# ------------------------------------------------------------
+# Helpers
+# ------------------------------------------------------------
 def transport_from_addr(
     addr: Address,
-    codec_key: str = 'msgpack',
+    codec_key: str = "msgpack",
 ) -> Type[MsgTransport]:
-    '''
+    """
     Given a destination address and a desired codec, find the
     corresponding `MsgTransport` type.
-
+    """
-    '''
     try:
-        return _addr_to_transport[type(addr)]
+        return _addr_to_transport[type(addr)]  # type: ignore[call-arg]
-
     except KeyError:
         raise NotImplementedError(
-            f'No known transport for address {repr(addr)}'
+            f"No known transport for address {repr(addr)}"
         )
 
 
 def transport_from_stream(
     stream: trio.abc.Stream,
-    codec_key: str = 'msgpack'
+    codec_key: str = "msgpack",
 ) -> Type[MsgTransport]:
-    '''
+    """
     Given an arbitrary `trio.abc.Stream` and a desired codec,
     find the corresponding `MsgTransport` type.
-
+    """
-    '''
     transport = None
+
     if isinstance(stream, trio.SocketStream):
         sock: socket.socket = stream.socket
-        match sock.family:
+        fam = sock.family
-            case socket.AF_INET | socket.AF_INET6:
-                transport = 'tcp'
 
-            case socket.AF_UNIX:
+        if fam in (socket.AF_INET, getattr(socket, "AF_INET6", None)):
-                transport = 'uds'
+            transport = "tcp"
 
-            case _:
+        # only consider `AF_UNIX` when the UDS backend is active;
-                raise NotImplementedError(
+        # `HAS_UDS` short-circuits before `socket.AF_UNIX` so this
-                    f'Unsupported socket family: {sock.family}'
+        # stays safe on hosts where that constant is absent.
-                )
+        if transport is None and HAS_UDS and fam == socket.AF_UNIX:  # type: ignore[attr-defined]
+            transport = "uds"
+
+        if transport is None:
+            raise NotImplementedError(f"Unsupported socket family: {fam}")
 
     if not transport:
         raise NotImplementedError(
-            f'Could not figure out transport type for stream type {type(stream)}'
+            f"Could not figure out transport type for stream type {type(stream)}"
         )
 
     key = (codec_key, transport)
-
     return _key_to_transport[key]
+

tractor/ipc/_uds.py

@@ -25,17 +25,28 @@ from pathlib import Path
 import os
 import sys
 from socket import (
-    AF_UNIX,
     SOCK_STREAM,
     SOL_SOCKET,
     error as socket_error,
 )
+# NOTE, `AF_UNIX` is absent on Windows / any CPython built without
+# unix-domain-socket support. Keep this module importable
+# everywhere (so `UDSAddress` stays referenceable for type and
+# `isinstance()` checks plus registry lookups); the `AF_UNIX`-using
+# code paths below are runtime-only and are never reached when the
+# UDS backend is unusable (gated on `trio`'s `has_unix`, see
+# `HAS_UDS`).
+try:
+    from socket import AF_UNIX
+except ImportError:
+    AF_UNIX = None
 import struct
 from typing import (
     Type,
     TYPE_CHECKING,
     ClassVar,
 )
+from uuid import uuid4
 
 import msgspec
 import trio
@@ -90,6 +101,13 @@ else:
 log = get_logger()
 
 
+# single source of truth for "is the UDS backend usable on this
+# host?" — reuse `trio`'s `has_unix` (the same predicate that gates
+# `trio.open_unix_socket()`) rather than re-deriving an `AF_UNIX`
+# probe in every consumer module.
+HAS_UDS: bool = has_unix
+
+
 def unwrap_sockpath(
     sockpath: Path,
 ) -> tuple[Path, Path]:
@@ -204,7 +222,18 @@ class UDSAddress(
             else:
                 prefix: str = 'no_runtime_actor'
 
-            sockname: str = f'{prefix}@{pid}'
+            # XXX, no live actor -> no `Aid` to key off, so mix a
+            # per-CALL token into the NAME part for uniqueness:
+            # w/o a runtime the sockname is otherwise a pure fn of
+            # `(prefix, pid)`, so two `get_random()` calls in one
+            # proc alias to the SAME sockpath and the 2nd `.bind()`
+            # trips `EADDRINUSE`. Token goes BEFORE `@{pid}` so the
+            # canonical `...@{pid}.sock` suffix stays intact for the
+            # reapers keyed off it: `._testing._reap`'s
+            # `(?P<name>.+)@(?P<pid>\d+)\.sock` regex, and the
+            # `spawn._reap` `{name}@{pid}.sock` reconstruction.
+            token: str = uuid4().hex[:8]
+            sockname: str = f'{prefix}.{token}@{pid}'
 
         sockpath: Path = Path(f'{sockname}.sock')
         return UDSAddress(