From 18faffcff72f546aa1588736ae30a8bbf0d21856 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:29:16 -0400 Subject: [PATCH 01/22] Surface example stderr on any non-zero exit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docs-example harness only re-raises captured subproc stderr when the LAST line contains 'Error', but a `tractor` root-actor crash always ends stderr with the strict-EG collapse note `( ^^^ this exc was collapsed from a group ^^^ )` — so every possible crash is swallowed down to a bare `assert 1 == 0`, exactly what the macOS CI leg shows for the UDS example in GH #473. - raise with the FULL stderr (+stdout) whenever the example subproc exits non-zero, regardless of stderr shape. - keep the legacy last-line 'Error' check for zero-rc runs which still emit error-ish output. First task-bullet of GH #473. Prompt-IO: ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tests/test_docs_examples.py | 32 +++++++++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index cce85292..1692ae63 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -178,10 +178,11 @@ def test_example( code = ex.read() with run_example_in_subproc(code) as proc: + out = None err = None try: if not proc.poll(): - _, err = proc.communicate(timeout=timeout) + out, err = proc.communicate(timeout=timeout) except subprocess.TimeoutExpired as e: test_log.exception( @@ -190,9 +191,34 @@ def test_example( proc.kill() err = e.stderr + errmsg: str = err.decode() if err else '' + + # XXX, ALWAYS surface the subproc's full stderr + # whenever it exits non-zero! + # + # The prior impl only raised when the LAST stderr + # line contained 'Error', swallowing any crash whose + # traceback ends in a non-`XxxError:` line; in + # particular EVERY `tractor` root-actor crash ends + # with the strict-EG collapse note, + # '( ^^^ this exc was collapsed from a group ^^^ )', + # so ALL such failures were reduced to a bare + # `assert 1 == 0` in CI logs.. see GH #473. + rc: int|None = proc.returncode + if rc: + outmsg: str = out.decode() if out else '' + raise Exception( + f'Example script exited with rc={rc} !?\n' + f'\n' + f'stdout:\n' + f'{outmsg}\n' + f'\n' + f'stderr:\n' + f'{errmsg}\n' + ) + # if we get some gnarly output let's aggregate and raise - if err: - errmsg = err.decode() + if errmsg: errlines = errmsg.splitlines() last_error = errlines[-1] if ( From bceca74eb7c23403c2a78564fe00dc823a1cf4be Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:30:27 -0400 Subject: [PATCH 02/22] Fix UDS addr corruption sans-autobind (macOS) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MsgpackUDSStream.get_stream_addrs()` matches the `(peername, sockname)` pair by type to find the listener's fs-path, but the `(str, str)` arm unconditionally takes `peername`: on platforms without linux's `SO_PASSCRED`-triggered autobind (macOS!) the accept side's `getpeername()` is `''`, so every accepted conn gets garbage `Path('')` laddr/raddr structs. Proven on linux by disabling `SO_PASSCRED` (no autobind -> same `''` shape as darwin): the `uds_transport_actor_tree.py` example reports `listener sock file: .` pre-fix and the real registry sockpath post-fix. - pick the non-empty name in the `(str, str)` arm: `peername` on the connect side, `sockname` on the accept side; raise `ValueError` on an (unexpected) empty pair. - document the linux-autobind origin of the `bytes` arms which the original impl noted as "unclear". - `start_listener()`: create the bindspace dir with `parents=True, exist_ok=True` (nested custom `filedir`s + racing actors). - example docstring: peer-pid comes via `SO_PEERCRED` on linux but `LOCAL_PEERPID` on macOS. May not be the (only) macOS crasher for GH #473 — it is non-fatal on the linux sim — but with stderr surfacing now in place the next macOS CI run pins any remaining layer. Prompt-IO: ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- examples/uds_transport_actor_tree.py | 5 +-- tractor/ipc/_uds.py | 48 +++++++++++++++++++++++----- 2 files changed, 43 insertions(+), 10 deletions(-) diff --git a/examples/uds_transport_actor_tree.py b/examples/uds_transport_actor_tree.py index 62ee55f4..93a17626 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -6,7 +6,8 @@ subactor inherits the preference. Every channel address is a filesystem socket path (no TCP port in sight!) and, as a kernel-provided bonus, the peer's pid is -exchanged for free via `SO_PEERCRED`. +exchanged for free via `SO_PEERCRED` on linux, +`LOCAL_PEERPID` on macOS. ''' import os @@ -42,7 +43,7 @@ async def main() -> None: # (named for the root registrar) this channel rode in # on, NOT a per-child path; the child-specific identity # we get for free is the kernel-reported peer pid (via - # `SO_PEERCRED`). + # `SO_PEERCRED` on linux, `LOCAL_PEERPID` on macOS). print( f'portal chan tpt proto: {raddr.proto_key!r}\n' f'listener sock file: {raddr.sockpath}\n' diff --git a/tractor/ipc/_uds.py b/tractor/ipc/_uds.py index 6dae10ec..afd28848 100644 --- a/tractor/ipc/_uds.py +++ b/tractor/ipc/_uds.py @@ -307,7 +307,16 @@ async def start_listener( f'>{{\n' f'|_{bs!r}\n' ) - bs.mkdir() + bs.mkdir( + # ensure the full ancestor tree for any nested + # (custom `filedir`) bindspace; the default + # `get_rt_dir()` space is pre-created but a custom + # one may have missing parents. + parents=True, + # avoid `FileExistsError` from racing actors, same + # guard as in `get_rt_dir()`. + exist_ok=True, + ) with _reraise_as_connerr( src_excs=( @@ -560,11 +569,18 @@ class MsgpackUDSStream(MsgpackTransport): ]: sock: trio.socket.socket = stream.socket - # NOTE XXX, it's unclear why one or the other ends up being - # `bytes` versus the socket-file-path, i presume it's - # something to do with who is the server (called `.listen()`)? - # maybe could be better implemented using another info-query - # on the socket like, + # NOTE, the `bytes` case is a linux-only artifact: setting + # `SO_PASSCRED` (see `open_unix_socket_w_passcred()`) + # causes the kernel to *autobind* the un-named client + # sock to an abstract-namespace addr which python + # delivers as `bytes`; the listener-bound end is always + # the fs-path `str`. On platforms WITHOUT autobind + # (macOS et al) the un-bound end instead reports as an + # empty `str` so BOTH names arrive as `str`s and the + # real fs-path is whichever is non-empty: `peername` on + # the connect side, `sockname` on the accept side. + # + # for socket-api deats see, # https://beej.us/guide/bgnet/html/split-wide/system-calls-or-bust.html#gethostnamewho-am-i sockname: str|bytes = sock.getsockname() # https://beej.us/guide/bgnet/html/split-wide/system-calls-or-bust.html#getpeernamewho-are-you @@ -576,8 +592,24 @@ class MsgpackUDSStream(MsgpackTransport): case (bytes(), str()): sock_path: Path = Path(sockname) - case (str(), str()): # XXX, likely macOS - sock_path: Path = Path(peername) + # XXX, no-autobind case (macOS): the un-bound end + # is `''`, NOT a `bytes` abstract-ns addr; taking + # `peername` unconditionally (as prior impl did) + # delivers garbage `Path('')` addrs on the accept + # side! + case (str(), str()): + bound_name: str = ( + peername + or + sockname + ) + if not bound_name: + raise ValueError( + f'Empty UDS (peername, sockname) pair ??\n' + f'peername: {peername!r}\n' + f'sockname: {sockname!r}\n' + ) + sock_path: Path = Path(bound_name) case _: raise TypeError( From 3d02a8569e05b3cfbb75b53752b8d40f7811b7a0 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:30:27 -0400 Subject: [PATCH 03/22] Run UDS-on-macOS CI leg, un-skip the example UDS-on-macOS is otherwise un-exercised: the matrix explicitly excludes the `macos-latest` + `uds` combo, so the `uds_transport_actor_tree.py` example (skipped on macOS CI since PR #460) is the only thing that ever touches that path. - drop the matrix `exclude` so the full suite runs with `--tpt-proto=uds` on `macos-latest`. - un-skip the example on macOS+CI; with example-stderr surfacing in place a still-red run now yields the full traceback GH #473 asks for instead of a bare returncode assert. Task-bullets 3 + 4 of GH #473. Prompt-IO: ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md (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 | 5 ----- tests/test_docs_examples.py | 15 --------------- 2 files changed, 20 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8bb1297c..529f8b11 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -118,11 +118,6 @@ jobs: 'tcp', 'uds', ] - # https://github.com/orgs/community/discussions/26253#discussioncomment-3250989 - exclude: - # don't do UDS run on macOS (for now) - - os: macos-latest - tpt_proto: 'uds' steps: - uses: actions/checkout@v4 diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index 1692ae63..90081a49 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -145,21 +145,6 @@ def test_example( 'This test does run just fine "in person" however..' ) - if ( - 'uds_transport_actor_tree' in ex_file - and - _friggin_macos - and - ci_env - ): - pytest.skip( - 'UDS-transport example reliably fails on macOS CI.\n' - 'UDS-on-macOS is otherwise un-exercised by the matrix\n' - '(no `tpt_proto=uds` macOS job), so this new example is\n' - 'the first to surface it; the macOS UDS path needs\n' - 'root-causing. Passes on Linux.' - ) - from .conftest import cpu_perf_headroom timeout: float = ( From 451e0acf8a754f0323a1182f2894433316325802 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:30:27 -0400 Subject: [PATCH 04/22] Add prompt-io log for GH #473 UDS-on-macOS work Provenance entry (+ unedited raw output) for the root-cause session behind the prior three patches, per the NLNet generative-AI policy tracked under `ai/prompt-io/`. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- .../20260702T155006Z_65bf9df5_prompt_io.md | 77 +++++++++++++ ...20260702T155006Z_65bf9df5_prompt_io.raw.md | 101 ++++++++++++++++++ 2 files changed, 178 insertions(+) create mode 100644 ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md create mode 100644 ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.raw.md diff --git a/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md b/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md new file mode 100644 index 00000000..fa7c7d8d --- /dev/null +++ b/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.md @@ -0,0 +1,77 @@ +--- +model: claude-fable-5 +service: claude +session: b8ec2ffe-1e47-4db5-8bc3-5bebc5feaaf6 +timestamp: 2026-07-02T15:50:06Z +git_ref: 65bf9df5 +scope: code +substantive: true +raw_file: 20260702T155006Z_65bf9df5_prompt_io.raw.md +--- + +## Prompt + +From `ai/prompt-io/prompts/issue_473.md`: + +> NOTE: you MUST pause this work at 12:50PM EST (BEFORE +> your weekly limit reset) for review by a human! +> +> resolve GH #473: https://github.com/goodboy/tractor/issues/474 +> do it with /open-wkt. + +(The prompt's issue-number/URL mismatch was resolved in +favor of #473 — file name, session name and prompt text +all say #473; the `/474` URL is a typo.) + +GH #473: "Root-cause UDS-on-macOS — +`uds_transport_actor_tree` example fails on CI". + +## Response summary + +Worked all four #473 next-step checkboxes in wkt +`uds_macos_473`: + +1. **stderr surfacing**: `test_docs_examples.py` now + raises with the full subproc stderr (+stdout) on ANY + non-zero exit. The old last-line-'Error' check could + never fire for `tractor` crashes since root-actor + stderr always ends with the strict-EG collapse note + `( ^^^ this exc was collapsed from a group ^^^ )` — + proven against the real PR #460 macOS CI log (bare + `assert 1 == 0`, no traceback). +2. **root-cause (linux-provable layer)**: macOS-only + addr corruption in + `MsgpackUDSStream.get_stream_addrs()` — no + `SO_PASSCRED`/autobind on darwin means the accept + side's `getpeername()` is `''`, and the + `(str(), str())` arm took `peername` unconditionally → + `Path('')` garbage addrs on every accepted conn. + Proven + fixed via linux no-autobind simulation. + Possibly not the final macOS crasher (non-fatal on + linux-sim); the diagnostic patch guarantees the next + macOS CI run shows any remaining layer. +3. **CI matrix**: removed the `macos-latest`+`uds` + exclude. +4. **un-skip**: dropped the macOS+CI skip of the example. + +Also: `start_listener()` bindspace mkdir hardened +(`parents=True, exist_ok=True`), example docstring +peer-pid mechanism corrected for macOS. + +## Files changed + +- `tests/test_docs_examples.py` — surface full stderr on + non-zero exit; remove macOS skip of the UDS example +- `tractor/ipc/_uds.py` — fix no-autobind + `get_stream_addrs()` arm; harden bindspace mkdir; + document autobind semantics +- `.github/workflows/ci.yml` — add macOS UDS leg (remove + matrix exclude) +- `examples/uds_transport_actor_tree.py` — platform- + correct peer-pid docs + +## Human edits + +None yet — pending human review (hard-stop review +checkpoint at 12:50PM EST per prompt); commit staging and +any edits are the human's. diff --git a/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.raw.md b/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.raw.md new file mode 100644 index 00000000..9eaa9457 --- /dev/null +++ b/ai/prompt-io/claude/20260702T155006Z_65bf9df5_prompt_io.raw.md @@ -0,0 +1,101 @@ +--- +model: claude-fable-5 +service: claude +timestamp: 2026-07-02T15:50:06Z +git_ref: 65bf9df5 +diff_cmd: git diff main..wkt/uds_macos_473 +--- + +# Raw output — GH #473 UDS-on-macOS root-cause session + +NOTE: code output is diff-referenced (not copied) per +prompt-io diff-ref mode; all generated code is on branch +`wkt/uds_macos_473` relative to `main` (65bf9df5). + +## Diagnostic narrative (verbatim) + +Root-cause work proceeded by linux-side simulation since no +macOS host is available: + +1. Pulled the actual macOS CI failure log from PR #460 run + 28343878933 (job 84115107992): the failure is a bare + `assert 1 == 0` on `proc.returncode` with NO stderr + surfaced — the harness only re-raised when the LAST + stderr line contained 'Error'. + +2. Reproduced the stderr-swallow mechanism on linux: any + `tractor` root-actor crash ends stderr with the + strict-EG collapse note + `( ^^^ this exc was collapsed from a group ^^^ )` which + never matches 'Error' — so EVERY possible crash was + swallowed. (Verified by sabotaging the runtime dir via + an over-long `XDG_RUNTIME_DIR` → `OSError: AF_UNIX path + too long` → rc=1 + swallowed.) + +3. Found + proved a macOS-only addr-corruption bug in + `MsgpackUDSStream.get_stream_addrs()`: the + `(str(), str())` match-arm unconditionally took + `peername`, but on no-autobind platforms (macOS lacks + linux's `SO_PASSCRED`-triggered autobind) the accept + side's `getpeername()` is `''` → `Path('')` garbage + laddr/raddr on EVERY accepted UDS conn. Simulated on + linux by nulling `SO_PASSCRED` (no autobind → same `''` + shape): pre-fix the example printed + `listener sock file: .`; post-fix it prints the real + registry sockpath. Non-fatal on linux-sim (rc=0), so + possibly not the final macOS crasher — the diagnostic + patch guarantees the next macOS CI run reveals any + remaining layer. + +4. Falsified the missing-parent-dir theory: + `get_rt_dir()` already `mkdir(parents=True, + exist_ok=True)`s at import (and macOS TCP CI passes), + so `~/Library/Caches/TemporaryItems` absence cannot be + the crasher. Hardened `start_listener()`'s bindspace + mkdir anyway (custom `filedir` case + racing actors). + +## Generated changes (diff pointers) + +> `git diff main..wkt/uds_macos_473 -- tests/test_docs_examples.py` + +- always raise with FULL subproc stderr (+stdout) on any + non-zero example exit; keep legacy last-line 'Error' + check for zero-rc cases; drop the macOS+CI skip of + `uds_transport_actor_tree.py` (GH #473 next-step). + +> `git diff main..wkt/uds_macos_473 -- tractor/ipc/_uds.py` + +- `get_stream_addrs()`: document the autobind semantics + (bytes = linux abstract-ns autobind artifact), add + no-autobind `(str, str)` arm picking the non-empty name + (`peername` connect-side, `sockname` accept-side) with + an empty-pair `ValueError` guard. +- `start_listener()`: `bs.mkdir(parents=True, + exist_ok=True)`. + +> `git diff main..wkt/uds_macos_473 -- .github/workflows/ci.yml` + +- remove the `macos-latest`+`uds` matrix exclude so + UDS-on-macOS is exercised by CI (GH #473 next-step). + +> `git diff main..wkt/uds_macos_473 -- examples/uds_transport_actor_tree.py` + +- docs nit: peer-pid mechanism is `SO_PEERCRED` on linux, + `LOCAL_PEERPID` on macOS. + +## Verification (verbatim summary) + +- macOS-shape sim (no autobind) + fix: example rc=0 with + correct listener sockpath. +- native linux post-fix: example rc=0, autobind arms + unchanged. +- sabotage smoke-test: diagnostic patch surfaces the full + EG traceback incl. collapse-note line. +- `pytest tests/ipc/ tests/test_2way.py tests/discovery/ + --tpt-proto uds`: 61 passed, 1 `TooSlowError` flake + (test_simple_rpc) that passes solo in 2.2s — attributed + to a concurrent full-suite run (other session) loading + the host. +- full `tests/test_docs_examples.py` run pending at + raw-file write time (queued behind the concurrent + session's suite). From 634d914161d4db9db7a153d40f23ae785a1ea1d0 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 16:39:32 -0400 Subject: [PATCH 05/22] Use short Darwin UDS runtime paths `platformdirs` can place the runtime dir deep below a pytest temp home, pushing `registry@1616.sock` past Darwin's 104-byte `AF_UNIX` limit. Use a compact `/tmp/-` root on Darwin and secure it before allocating socket paths: - require a real, current-user-owned dir via `lstat()` - tighten existing roots to mode `0700` - reject symlinks and unsafe nested dirs Cover the path budget, mode, and symlink rejection. Caught-during: review remediation Found-via: `/run-tests` test_macos_rt_dir_fits_uds_path_limit Review: PR #480 (goodboy,copilot-pull-request-reviewer[bot]) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 68 +++++++++++++++++++++++++++++++++++- tractor/runtime/_state.py | 70 ++++++++++++++++++++++++++++++++++---- 2 files changed, 131 insertions(+), 7 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 5d1fdea3..d7908cae 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -3,14 +3,17 @@ Unit-ish tests for specific IPC transport protocol backends. ''' from __future__ import annotations +import os from pathlib import Path +import stat +import sys import pytest import trio import tractor from tractor import Actor -from tractor.runtime import _state from tractor.discovery import _addr +from tractor.runtime import _state @pytest.fixture @@ -31,6 +34,69 @@ def bindspace_dir_str() -> str: bs_dir.rmdir() +def test_macos_rt_dir_fits_uds_path_limit( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Keep the default Darwin UDS bindpath below its 104-byte limit. + + `platformdirs` normally places the runtime directory below the + long `~/Library/Caches/TemporaryItems` path. Pytest also assigns + a deeply nested temporary home, so appending a registry socket + name made every macOS UDS listener fail with `AF_UNIX path too + long`. This test simulates Darwin and an intentionally long + platformdirs result, then proves `get_rt_dir()` uses the short + system temporary directory and leaves room for the socket name. + + ''' + long_rt_dir: Path = tmp_path / ('long' * 30) + monkeypatch.setattr(sys, 'platform', 'darwin') + monkeypatch.setattr( + 'platformdirs.user_runtime_dir', + lambda appname: str(long_rt_dir / appname), + ) + monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path) + rt_dir: Path = _state.get_rt_dir() + sockpath: Path = ( + Path('/tmp') + / f'tractor-{os.getuid()}' + / 'registry@1616.sock' + ) + + assert rt_dir == tmp_path / f'tractor-{os.getuid()}' + assert len(os.fsencode(sockpath)) < 104 + assert stat.S_IMODE(rt_dir.stat().st_mode) == 0o700 + + +def test_macos_rt_dir_rejects_symlink( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Reject a pre-created symlink at the Darwin runtime path. + + Darwin uses the predictable `/tmp/tractor-` path to stay + below its `AF_UNIX` limit. A hostile local user could otherwise + point that path at a victim-owned directory and make + `get_rt_dir()` chmod or place sockets in the symlink target. The + test replaces `/tmp` with a controlled directory, installs the + malicious link, and proves non-following validation rejects it. + + ''' + runtime_link: Path = tmp_path / f'tractor-{os.getuid()}' + target_dir: Path = tmp_path / 'target' + target_dir.mkdir(mode=0o755) + runtime_link.symlink_to(target_dir, target_is_directory=True) + monkeypatch.setattr(sys, 'platform', 'darwin') + monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path) + + with pytest.raises(PermissionError, match='Unsafe Darwin'): + _state.get_rt_dir() + + assert stat.S_IMODE(target_dir.stat().st_mode) == 0o755 + + def test_uds_bindspace_created_implicitly( debug_mode: bool, bindspace_dir_str: str, diff --git a/tractor/runtime/_state.py b/tractor/runtime/_state.py index 0020fd8d..df012cca 100644 --- a/tractor/runtime/_state.py +++ b/tractor/runtime/_state.py @@ -22,7 +22,10 @@ from __future__ import annotations from contextvars import ( ContextVar, ) +import os from pathlib import Path +import stat +import sys from typing import ( Any, Callable, @@ -42,6 +45,9 @@ if TYPE_CHECKING: from .._context import Context +_DARWIN_TMPDIR: Path = Path('/tmp') + + # default IPC transport protocol settings TransportProtocolKey = Literal[ 'tcp', @@ -335,11 +341,44 @@ def get_rt_dir( # `import tractor` path (gh #470). import platformdirs - rt_dir: Path = Path( - platformdirs.user_runtime_dir( - appname=appname, - ), - ) + rt_root: Path|None = None + if sys.platform == 'darwin': + # Darwin's AF_UNIX path limit is 104 bytes. The standard + # platformdirs path can consume that before the sock name. + rt_root = ( + _DARWIN_TMPDIR + / f'{appname}-{os.getuid()}' + ) + try: + rt_stat: os.stat_result = rt_root.lstat() + except FileNotFoundError: + try: + rt_root.mkdir(mode=0o700) + except FileExistsError: + pass + rt_stat = rt_root.lstat() + + if ( + not stat.S_ISDIR(rt_stat.st_mode) + or + rt_stat.st_uid != os.getuid() + ): + raise PermissionError( + f'Unsafe Darwin runtime directory!\n' + f'path: {rt_root}\n' + f'owner uid: {rt_stat.st_uid}\n' + f'mode: {stat.filemode(rt_stat.st_mode)}\n' + ) + if stat.S_IMODE(rt_stat.st_mode) != 0o700: + rt_root.chmod(0o700) + + rt_dir: Path = rt_root + else: + rt_dir = Path( + platformdirs.user_runtime_dir( + appname=appname, + ), + ) # Normalize and validate that `subdir` is a relative path # without any parent-directory ("..") components, to prevent @@ -363,11 +402,30 @@ def get_rt_dir( rt_dir: Path = rt_dir / subdir_path - if not rt_dir.is_dir(): + try: + dir_stat: os.stat_result = rt_dir.lstat() + except FileNotFoundError: rt_dir.mkdir( + mode=0o700, parents=True, exist_ok=True, # avoid `FileExistsError` from conc calls ) + dir_stat = rt_dir.lstat() + + if rt_root is not None: + if ( + not stat.S_ISDIR(dir_stat.st_mode) + or + dir_stat.st_uid != os.getuid() + ): + raise PermissionError( + f'Unsafe Darwin runtime directory!\n' + f'path: {rt_dir}\n' + f'owner uid: {dir_stat.st_uid}\n' + f'mode: {stat.filemode(dir_stat.st_mode)}\n' + ) + if rt_dir != rt_root: + rt_dir.relative_to(rt_root) return rt_dir From 0a3b0efcc6af2e9b107d106ce4c78260bf620024 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 16:40:05 -0400 Subject: [PATCH 06/22] Reap timed-out docs example trees Always drain docs-example pipes, even when the process exited before the first status check, and decode malformed output with replacement so diagnostics preserve the original failure. Run POSIX examples in dedicated sessions and kill the full process group on timeout. Reap the leader in every exit path, with bounded Windows cleanup that cannot wait forever on descendant-held pipes. Cover fast non-zero exits, invalid output bytes, process-group termination, and post-timeout reaping. Review: PR #480 (goodboy,copilot-pull-request-reviewer[bot]) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_docs_examples.py | 248 +++++++++++++++++++++++++++--------- 1 file changed, 189 insertions(+), 59 deletions(-) diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index 90081a49..f23fc926 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -5,11 +5,13 @@ Let's make sure them docs work yah? from contextlib import contextmanager import itertools import os +import signal import sys import subprocess import platform import shutil from typing import Callable +from unittest.mock import Mock import pytest import tractor @@ -21,6 +23,181 @@ _non_linux: bool = platform.system() != 'Linux' _friggin_macos: bool = platform.system() == 'Darwin' +def _kill_proc_tree(proc: subprocess.Popen) -> None: + ''' + Terminate an example process and its POSIX descendants. + + ''' + try: + if platform.system() == 'Windows': + proc.kill() + else: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +def _reap_killed_proc( + proc: subprocess.Popen, +) -> tuple[bytes, bytes]: + ''' + Reap a killed process without waiting on Windows descendants. + + ''' + if platform.system() != 'Windows': + return proc.communicate() + + proc.wait(timeout=5) + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() + return b'', b'' + + +def _wait_for_proc( + proc: subprocess.Popen, + timeout: float, + test_log: tractor.log.StackLevelAdapter, +) -> None: + ''' + Wait for an example process and surface its captured output. + + ''' + try: + out, err = proc.communicate(timeout=timeout) + + except subprocess.TimeoutExpired as timeout_exc: + test_log.exception( + f'Example failed to finish within {timeout}s ??\n' + ) + _kill_proc_tree(proc) + out, err = _reap_killed_proc(proc) + if platform.system() == 'Windows': + out = timeout_exc.output or b'' + err = timeout_exc.stderr or b'' + + errmsg: str = err.decode(errors='replace') + + # XXX, ALWAYS surface the subproc's full stderr + # whenever it exits non-zero! + # + # The prior impl only raised when the LAST stderr + # line contained 'Error', swallowing any crash whose + # traceback ends in a non-`XxxError:` line; in + # particular EVERY `tractor` root-actor crash ends + # with the strict-EG collapse note, + # '( ^^^ this exc was collapsed from a group ^^^ )', + # so ALL such failures were reduced to a bare + # `assert 1 == 0` in CI logs.. see GH #473. + rc: int|None = proc.returncode + if rc: + outmsg: str = out.decode(errors='replace') + raise Exception( + f'Example script exited with rc={rc} !?\n' + f'\n' + f'stdout:\n' + f'{outmsg}\n' + f'\n' + f'stderr:\n' + f'{errmsg}\n' + ) + + # if we get some gnarly output let's aggregate and raise + if errmsg: + errlines = errmsg.splitlines() + last_error = errlines[-1] + if ( + 'Error' in last_error + + # XXX: currently we print this to console, but maybe + # shouldn't eventually once we figure out what's + # a better way to be explicit about aio side + # cancels? + and + 'asyncio.exceptions.CancelledError' not in last_error + ): + raise Exception(errmsg) + + assert proc.returncode == 0 + + +def test_wait_for_failed_example_captures_output(): + ''' + Preserve diagnostics from a subprocess which already exited. + + The previous `poll()` guard skipped `communicate()` when a fast + failure returned a non-zero status before the parent checked it. + Its stdout and stderr were therefore reported as empty. This + fake process begins with `returncode=1` and returns non-UTF-8 + output, proving the helper always drains both pipes and replaces + undecodable bytes without hiding the original process failure. + + ''' + proc = Mock() + proc.returncode = 1 + proc.communicate.return_value = ( + b'stdout\xff', + b'stderr\xff', + ) + + with pytest.raises(Exception) as exc_info: + _wait_for_proc( + proc=proc, + timeout=1, + test_log=Mock(), + ) + + proc.communicate.assert_called_once_with(timeout=1) + errmsg: str = str(exc_info.value) + assert 'stdout\ufffd' in errmsg + assert 'stderr\ufffd' in errmsg + + +@pytest.mark.skipif( + platform.system() == 'Windows', + reason='POSIX process groups are unavailable on Windows', +) +def test_wait_for_timed_out_example_reaps_group( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Kill the example process group and reap its leader on timeout. + + The old timeout branch killed only the immediate process and + never drained it. Actor descendants could retain the capture + pipes while the leader remained unreaped, hanging CI until its + job timeout. This fake process raises `TimeoutExpired` on the + timed wait and completes on the second `communicate()` call; + the assertions prove group-directed `SIGKILL` precedes that + final drain and leaves a concrete non-zero return code. + + ''' + proc = Mock() + proc.pid = 1234 + + def communicate(timeout=None): + if timeout is not None: + raise subprocess.TimeoutExpired('example', timeout) + proc.returncode = -signal.SIGKILL + return b'', b'timed out' + + proc.communicate.side_effect = communicate + killpg = Mock() + monkeypatch.setattr(os, 'killpg', killpg) + + with pytest.raises(Exception, match='timed out'): + _wait_for_proc( + proc=proc, + timeout=.01, + test_log=Mock(), + ) + + killpg.assert_called_once_with(1234, signal.SIGKILL) + assert proc.communicate.call_count == 2 + assert proc.returncode == -signal.SIGKILL + + @pytest.fixture def run_example_in_subproc( loglevel: str, @@ -61,6 +238,7 @@ def run_example_in_subproc( ] else: script_file = testdir.makefile('.py', script_code) + kwargs['start_new_session'] = True cmdargs = [ sys.executable, str(script_file), @@ -77,9 +255,12 @@ def run_example_in_subproc( **kwargs, ) assert not proc.returncode - yield proc - proc.wait() - assert proc.returncode == 0 + try: + yield proc + finally: + if proc.poll() is None: + _kill_proc_tree(proc) + _reap_killed_proc(proc) yield run @@ -163,59 +344,8 @@ def test_example( code = ex.read() with run_example_in_subproc(code) as proc: - out = None - err = None - try: - if not proc.poll(): - out, err = proc.communicate(timeout=timeout) - - except subprocess.TimeoutExpired as e: - test_log.exception( - f'Example failed to finish within {timeout}s ??\n' - ) - proc.kill() - err = e.stderr - - errmsg: str = err.decode() if err else '' - - # XXX, ALWAYS surface the subproc's full stderr - # whenever it exits non-zero! - # - # The prior impl only raised when the LAST stderr - # line contained 'Error', swallowing any crash whose - # traceback ends in a non-`XxxError:` line; in - # particular EVERY `tractor` root-actor crash ends - # with the strict-EG collapse note, - # '( ^^^ this exc was collapsed from a group ^^^ )', - # so ALL such failures were reduced to a bare - # `assert 1 == 0` in CI logs.. see GH #473. - rc: int|None = proc.returncode - if rc: - outmsg: str = out.decode() if out else '' - raise Exception( - f'Example script exited with rc={rc} !?\n' - f'\n' - f'stdout:\n' - f'{outmsg}\n' - f'\n' - f'stderr:\n' - f'{errmsg}\n' - ) - - # if we get some gnarly output let's aggregate and raise - if errmsg: - errlines = errmsg.splitlines() - last_error = errlines[-1] - if ( - 'Error' in last_error - - # XXX: currently we print this to console, but maybe - # shouldn't eventually once we figure out what's - # a better way to be explicit about aio side - # cancels? - and - 'asyncio.exceptions.CancelledError' not in last_error - ): - raise Exception(errmsg) - - assert proc.returncode == 0 + _wait_for_proc( + proc=proc, + timeout=timeout, + test_log=test_log, + ) From 64e820e18e10b640b75a60532f9bff0756a44e4a Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 18:30:44 -0400 Subject: [PATCH 07/22] Normalize peer resets in `.send()` A raw UDS readiness client can disconnect before the actor handshake. Darwin reports the first server write as `ECONNRESET`, wrapped in `trio.BrokenResourceError`; letting it escape cancels the daemon's shared IPC nursery and makes later roots elect themselves registrar. Walk the exception chain for `EPIPE` or `ECONNRESET` and translate either into the existing `TransportClosed` boundary. Also handle argument-less resource errors without raising `IndexError`. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_server.py | 46 +++++++++++++++++++++++++++++++++++++ tractor/ipc/_transport.py | 48 +++++++++++++++++++++++++++++++-------- 2 files changed, 84 insertions(+), 10 deletions(-) diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py index 1d63bd1b..01ddb222 100644 --- a/tests/ipc/test_server.py +++ b/tests/ipc/test_server.py @@ -3,6 +3,7 @@ High-level `.ipc._server` unit tests. ''' from __future__ import annotations +import errno import pytest import trio @@ -14,6 +15,8 @@ from tractor import ( from tractor._testing.addr import ( get_rando_addr, ) +from tractor._exceptions import TransportClosed +from tractor.ipc._transport import MsgpackTransport # TODO, use/check-roundtripping with some of these wrapper types? # # from .._addr import Address @@ -23,6 +26,49 @@ from tractor._testing.addr import ( # from ._tcp import TCPAddress +def test_send_normalizes_peer_reset(): + ''' + Normalize Darwin's pre-handshake peer reset as transport closure. + + A raw UDS readiness client connects and immediately disconnects. + Darwin reports the server's first handshake write as + `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing + that raw error to escape cancels the daemon's shared IPC nursery. + This fake stream reproduces the exact exception chain and proves + `.send()` raises the expected `TransportClosed` boundary instead. + + ''' + class ResetStream: + async def send_all(self, data: bytes) -> None: + try: + raise OSError( + errno.ECONNRESET, + 'Connection reset by peer', + ) + except OSError as reset_err: + raise trio.BrokenResourceError from reset_err + + async def main(): + transport = object.__new__(MsgpackTransport) + transport.stream = ResetStream() + transport._send_lock = trio.StrictFIFOLock() + transport._laddr = 'local' + transport._raddr = 'remote' + transport._task = trio.lowlevel.current_task() + + with pytest.raises(TransportClosed) as exc_info: + await transport.send( + {'probe': True}, + strict_types=False, + ) + + assert exc_info.value.src_exc.__cause__.errno == ( + errno.ECONNRESET + ) + + trio.run(main) + + @pytest.mark.parametrize( '_tpt_proto', ['uds', 'tcp'] diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index dd3be179..48692f79 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -33,6 +33,7 @@ from collections.abc import ( AsyncGenerator, AsyncIterator, ) +import errno import struct import trio @@ -61,6 +62,33 @@ if TYPE_CHECKING: log = get_logger() +def _peer_closed_errno(exc: BaseException) -> int|None: + ''' + Find a peer-close errno in a transport exception chain. + + ''' + seen: set[int] = set() + while ( + exc + and + id(exc) not in seen + ): + seen.add(id(exc)) + if ( + isinstance(exc, OSError) + and + exc.errno in { + errno.ECONNRESET, + errno.EPIPE, + } + ): + return exc.errno + + exc = exc.__cause__ or exc.__context__ + + return None + + # (codec, transport) MsgTransportKey = tuple[str, str] @@ -443,23 +471,23 @@ class MsgpackTransport(MsgTransport): trans_err = _re tpt_name: str = f'{type(self).__name__!r}' - trans_err_msg: str = trans_err.args[0] + trans_err_msg: str = ( + str(trans_err.args[0]) + if trans_err.args + else '' + ) by_whom: str = { 'another task closed this fd': 'locally', 'this socket was already closed': 'by peer', }.get(trans_err_msg) match trans_err: - # XXX, specifc to UDS transport and its, - # well, "speediness".. XD - # |_ likely todo with races related to how fast - # the socket is setup/torn-down on linux - # as it pertains to rando pings from the - # `.discovery` subsys and protos. + # UDS peers can disconnect before handshake. + # Linux normally reports `EPIPE`; Darwin reports + # `ECONNRESET` for the same expected closure. case trio.BrokenResourceError() if ( - '[Errno 32] Broken pipe' - in - trans_err_msg + _peer_closed_errno(trans_err) + is not None ): tpt_closed = TransportClosed.from_src_exc( message=( From 340d5069403404f665764a2551d56b97423cada2 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 18:40:55 -0400 Subject: [PATCH 08/22] Bound generated UDS socket paths Actor names and custom runtime subdirs can otherwise produce unsafe or overlong pathname sockets after moving Darwin's bindspace to `/tmp`. Deats, - hash unsafe or over-budget actor names while retaining `@pid.sock` - share deterministic naming with post-kill socket cleanup - enforce Linux and Darwin `sun_path` byte budgets - restore non-Darwin dir validation and mode `0700` - validate every nested Darwin runtime-dir component Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 137 +++++++++++++++++++++++++++++++++++++ tractor/ipc/_uds.py | 70 ++++++++++++++++++- tractor/runtime/_state.py | 65 +++++++++++------- tractor/spawn/_reap.py | 7 +- 4 files changed, 250 insertions(+), 29 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index d7908cae..9c6d99bd 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -97,6 +97,143 @@ def test_macos_rt_dir_rejects_symlink( assert stat.S_IMODE(target_dir.stat().st_mode) == 0o755 +def test_rt_dir_rejects_non_directory( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Preserve the non-Darwin runtime-directory type contract. + + Replacing `Path.is_dir()` with unguarded `lstat()` briefly made + existing files look like valid runtime directories on Linux. + This test points `platformdirs` at a regular file and proves + `get_rt_dir()` rejects it during initialization. + + ''' + rt_file: Path = tmp_path / 'runtime-file' + rt_file.touch() + monkeypatch.setattr(sys, 'platform', 'linux') + monkeypatch.setattr( + 'platformdirs.user_runtime_dir', + lambda appname: str(rt_file), + ) + + with pytest.raises(FileExistsError): + _state.get_rt_dir() + + new_rt_dir: Path = tmp_path / 'new-runtime-dir' + monkeypatch.setattr( + 'platformdirs.user_runtime_dir', + lambda appname: str(new_rt_dir), + ) + assert _state.get_rt_dir() == new_rt_dir + assert stat.S_IMODE(new_rt_dir.stat().st_mode) == 0o700 + + +def test_macos_rt_dir_rejects_intermediate_symlink( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Reject symlinks in nested Darwin runtime subdirectories. + + The earlier final-component check allowed `link/child` to follow + an intermediate symlink and create `child` outside the secured + runtime root. This test installs that link and proves traversal + stops before anything is created in its target. + + ''' + rt_root: Path = tmp_path / f'tractor-{os.getuid()}' + target_dir: Path = tmp_path / 'target' + rt_root.mkdir(mode=0o700) + target_dir.mkdir() + (rt_root / 'link').symlink_to( + target_dir, + target_is_directory=True, + ) + monkeypatch.setattr(sys, 'platform', 'darwin') + monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path) + + with pytest.raises(PermissionError, match='Unsafe Darwin'): + _state.get_rt_dir(subdir='link/child') + + assert not (target_dir / 'child').exists() + + +@pytest.mark.parametrize( + ('platform_name', 'path_limit'), + [ + ('darwin', 104), + ('linux', 108), + ], +) +def test_uds_sockname_compaction( + monkeypatch: pytest.MonkeyPatch, + platform_name: str, + path_limit: int, +): + ''' + Keep generated actor sockets safe and below Darwin's byte limit. + + Actor names are unrestricted identity strings. A long, multibyte, + or path-like name previously produced overlong or escaping socket + paths. These cases prove `UDSAddress.get_sockname()` preserves a + short legacy name, deterministically compacts unsafe names, keeps + the reaper's `@pid.sock` suffix, and stays within Darwin's byte + limit. + + ''' + from tractor.ipc._uds import UDSAddress + + bindspace: Path = Path('/tmp/tractor-501') + pid: int = 12345 + from tractor.ipc import _uds + + monkeypatch.setattr(sys, 'platform', platform_name) + monkeypatch.setattr(_uds, '_SUN_PATH_LIMIT', path_limit) + + short: Path = UDSAddress.get_sockname( + name='worker', + pid=pid, + bindspace=bindspace, + ) + long_name: str = 'actor-' + ('\u00e9' * 100) + compact: Path = UDSAddress.get_sockname( + name=long_name, + pid=pid, + bindspace=bindspace, + ) + unsafe: Path = UDSAddress.get_sockname( + name='../worker', + pid=pid, + bindspace=bindspace, + ) + + assert short == Path(f'worker@{pid}.sock') + assert compact == UDSAddress.get_sockname( + name=long_name, + pid=pid, + bindspace=bindspace, + ) + assert compact.name.endswith(f'@{pid}.sock') + assert unsafe.parent == Path('.') + assert '..' not in unsafe.name + assert len(os.fsencode(bindspace / compact)) < path_limit + + with pytest.raises(ValueError) as exc_info: + UDSAddress.get_sockname( + name=long_name, + pid=pid, + bindspace=Path('/tmp') / ('x' * 90), + ) + + errmsg: str = str(exc_info.value) + assert 'leaves no room' in errmsg + assert 'name was unsafe: False' in errmsg + assert 'name was over budget: True' in errmsg + assert f'AF_UNIX path limit: {path_limit}' in errmsg + + def test_uds_bindspace_created_implicitly( debug_mode: bool, bindspace_dir_str: str, diff --git a/tractor/ipc/_uds.py b/tractor/ipc/_uds.py index afd28848..f074820d 100644 --- a/tractor/ipc/_uds.py +++ b/tractor/ipc/_uds.py @@ -21,6 +21,7 @@ from __future__ import annotations from contextlib import ( contextmanager as cm, ) +import hashlib from pathlib import Path import os import sys @@ -96,6 +97,12 @@ else: log = get_logger() +_SUN_PATH_LIMIT: int = ( + 108 + if sys.platform == 'linux' + else 104 +) + def unwrap_sockpath( sockpath: Path, @@ -196,7 +203,7 @@ class UDSAddress( err_on_no_runtime=False, ) if actor: - sockname: str = f'{actor.aid.name}@{pid}' + sockname: str = actor.aid.name # XXX, orig version which broke both macOS (file-name # length) and `multiaddrs` ('::' invalid separator). # sockname: str = '::'.join(actor.uid) + f'@{pid}' @@ -222,15 +229,72 @@ class UDSAddress( # `(?P.+)@(?P\d+)\.sock` regex, and the # `spawn._reap` `{name}@{pid}.sock` reconstruction. token: str = uuid4().hex[:8] - sockname: str = f'{prefix}.{token}@{pid}' + sockname = f'{prefix}.{token}' - sockpath: Path = Path(f'{sockname}.sock') + sockpath: Path = cls.get_sockname( + name=sockname, + pid=pid, + bindspace=filedir, + ) return UDSAddress( filedir=filedir, filename=sockpath, maybe_pid=pid, ) + @classmethod + def get_sockname( + cls, + name: str, + pid: int, + bindspace: Path, + ) -> Path: + ''' + Build a safe, deterministic UDS socket filename. + + ''' + suffix: str = f'@{pid}.sock' + filename: str = f'{name}{suffix}' + unsafe: bool = ( + '\0' in name + or + '/' in name + or + bool(os.altsep and os.altsep in name) + or + Path(filename).is_absolute() + ) + too_long: bool = ( + len(os.fsencode(bindspace / filename)) + >= _SUN_PATH_LIMIT + ) + if ( + unsafe + or + too_long + ): + digest: str = hashlib.blake2s( + os.fsencode(name), + digest_size=16, + ).hexdigest() + filename = f'actor.{digest}{suffix}' + + sockpath: Path = bindspace / filename + path_nbytes: int = len(os.fsencode(sockpath)) + if path_nbytes >= _SUN_PATH_LIMIT: + raise ValueError( + f'UDS bindspace leaves no room for an AF_UNIX ' + f'socket filename!\n' + f'bindspace: {bindspace}\n' + f'name was unsafe: {unsafe}\n' + f'name was over budget: {too_long}\n' + f'compacted filename: {filename}\n' + f'encoded path bytes: {path_nbytes}\n' + f'AF_UNIX path limit: {_SUN_PATH_LIMIT}\n' + ) + + return Path(filename) + @classmethod def get_root(cls) -> UDSAddress: def_uds_filename: Path = 'registry@1616.sock' diff --git a/tractor/runtime/_state.py b/tractor/runtime/_state.py index df012cca..154266bb 100644 --- a/tractor/runtime/_state.py +++ b/tractor/runtime/_state.py @@ -383,6 +383,7 @@ def get_rt_dir( # Normalize and validate that `subdir` is a relative path # without any parent-directory ("..") components, to prevent # escaping the runtime directory. + subdir_path: Path|None = None if subdir: subdir_path = ( subdir @@ -400,32 +401,46 @@ def get_rt_dir( f'{subdir!r}\n' ) - rt_dir: Path = rt_dir / subdir_path - - try: - dir_stat: os.stat_result = rt_dir.lstat() - except FileNotFoundError: - rt_dir.mkdir( - mode=0o700, - parents=True, - exist_ok=True, # avoid `FileExistsError` from conc calls - ) - dir_stat = rt_dir.lstat() - - if rt_root is not None: - if ( - not stat.S_ISDIR(dir_stat.st_mode) - or - dir_stat.st_uid != os.getuid() - ): - raise PermissionError( - f'Unsafe Darwin runtime directory!\n' - f'path: {rt_dir}\n' - f'owner uid: {dir_stat.st_uid}\n' - f'mode: {stat.filemode(dir_stat.st_mode)}\n' + if rt_root is None: + if subdir_path is not None: + rt_dir = rt_dir / subdir_path + if not rt_dir.is_dir(): + rt_dir.mkdir( + # Runtime dirs hold IPC sockets; owner-only access + # prevents other users from traversing the bindspace. + mode=0o700, + parents=True, + exist_ok=True, ) - if rt_dir != rt_root: - rt_dir.relative_to(rt_root) + return rt_dir + + if subdir_path is not None: + for part in subdir_path.parts: + rt_dir = rt_dir / part + try: + dir_stat: os.stat_result = rt_dir.lstat() + except FileNotFoundError: + try: + # Every Darwin component is private so no other + # user can replace descendants below `rt_root`. + rt_dir.mkdir(mode=0o700) + except FileExistsError: + pass + dir_stat = rt_dir.lstat() + + if ( + not stat.S_ISDIR(dir_stat.st_mode) + or + dir_stat.st_uid != os.getuid() + ): + raise PermissionError( + f'Unsafe Darwin runtime directory!\n' + f'path: {rt_dir}\n' + f'owner uid: {dir_stat.st_uid}\n' + f'mode: {stat.filemode(dir_stat.st_mode)}\n' + ) + if stat.S_IMODE(dir_stat.st_mode) != 0o700: + rt_dir.chmod(0o700) return rt_dir diff --git a/tractor/spawn/_reap.py b/tractor/spawn/_reap.py index bf2c104d..f8ab4518 100644 --- a/tractor/spawn/_reap.py +++ b/tractor/spawn/_reap.py @@ -71,6 +71,7 @@ fd leak. Different bug class but same broader theme of from __future__ import annotations import os +from pathlib import Path from typing import TYPE_CHECKING import trio @@ -154,7 +155,11 @@ def unlink_uds_bind_addrs( and subactor is not None and proc.pid is not None ): - sockname: str = f'{subactor.aid.name}@{proc.pid}.sock' + sockname: Path = UDSAddress.get_sockname( + name=subactor.aid.name, + pid=proc.pid, + bindspace=UDSAddress.def_bindspace, + ) sockpath: str = str( UDSAddress.def_bindspace / sockname ) From bd38204fde928bd26307d29c81a5cf7e2bd9d0d1 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 18:41:20 -0400 Subject: [PATCH 09/22] Preserve docs-example body failures Best-effort subprocess teardown must not replace the exception raised by the test body. Suppress cleanup errors only while propagating that active failure; keep raising teardown errors on normal body exit. Also close the Windows stdin pipe after its bounded leader reap. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_docs_examples.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index f23fc926..7d9b6418 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -48,6 +48,8 @@ def _reap_killed_proc( return proc.communicate() proc.wait(timeout=5) + if proc.stdin: + proc.stdin.close() if proc.stdout: proc.stdout.close() if proc.stderr: @@ -257,7 +259,15 @@ def run_example_in_subproc( assert not proc.returncode try: yield proc - finally: + except BaseException: + if proc.poll() is None: + try: + _kill_proc_tree(proc) + _reap_killed_proc(proc) + except Exception: + pass + raise + else: if proc.poll() is None: _kill_proc_tree(proc) _reap_killed_proc(proc) From 0d6d7c2a63c1c41d482db4aaaf7eb1cb823a5353 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 18:57:13 -0400 Subject: [PATCH 10/22] Keep UDS post-kill cleanup best-effort `unlink_uds_bind_addrs()` reconstructs self-assigned socket paths after a hard kill. An over-budget bindspace can make `UDSAddress.get_sockname()` raise before the guarded `os.unlink()`, replacing the original supervision outcome after the child is gone. Catch and report reconstruction failures, then skip cleanup without raising. Cover the overflow path and prove no unlink is attempted. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 40 ++++++++++++++++++++++++++++++++++++++ tractor/spawn/_reap.py | 20 ++++++++++++++----- 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index 9c6d99bd..f27fed5d 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -7,6 +7,8 @@ import os from pathlib import Path import stat import sys +from types import SimpleNamespace +from unittest.mock import Mock import pytest import trio @@ -234,6 +236,44 @@ def test_uds_sockname_compaction( assert f'AF_UNIX path limit: {path_limit}' in errmsg +def test_uds_reaper_ignores_unreconstructable_path( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Keep post-kill UDS cleanup best-effort on path overflow. + + `unlink_uds_bind_addrs()` reconstructs a self-assigned socket from + the dead actor's name and PID. An over-budget bindspace makes that + naming helper raise before `os.unlink()`; propagating the error + would replace the original supervision outcome after the child was + already killed. This test forces overflow and proves cleanup skips + reconstruction without attempting an unlink or raising. + + ''' + from tractor.ipc import _uds + from tractor.spawn import _reap + + long_bindspace: Path = Path('/tmp') / ('x' * 120) + proc = SimpleNamespace(pid=12345) + subactor = SimpleNamespace( + aid=SimpleNamespace(name='worker'), + ) + unlink = Mock() + monkeypatch.setattr( + _uds.UDSAddress, + 'def_bindspace', + long_bindspace, + ) + monkeypatch.setattr(_reap.os, 'unlink', unlink) + + _reap.unlink_uds_bind_addrs( + proc=proc, + subactor=subactor, + ) + + unlink.assert_not_called() + + def test_uds_bindspace_created_implicitly( debug_mode: bool, bindspace_dir_str: str, diff --git a/tractor/spawn/_reap.py b/tractor/spawn/_reap.py index f8ab4518..5dfff2e7 100644 --- a/tractor/spawn/_reap.py +++ b/tractor/spawn/_reap.py @@ -155,11 +155,21 @@ def unlink_uds_bind_addrs( and subactor is not None and proc.pid is not None ): - sockname: Path = UDSAddress.get_sockname( - name=subactor.aid.name, - pid=proc.pid, - bindspace=UDSAddress.def_bindspace, - ) + try: + sockname: Path = UDSAddress.get_sockname( + name=subactor.aid.name, + pid=proc.pid, + bindspace=UDSAddress.def_bindspace, + ) + except Exception: + log.exception( + f'Failed to reconstruct UDS sock-file for ' + f'post-kill cleanup — skipping\n' + f' |_{proc}\n' + f' |_{subactor.aid}\n' + ) + return + sockpath: str = str( UDSAddress.def_bindspace / sockname ) From 5724c0516a87a63db51e45799906cec9edf95458 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 20:04:48 -0400 Subject: [PATCH 11/22] Probe registrar capability in actor handshakes Transport connect alone can select a foreign, stalled, or ordinary actor endpoint as the registry. On macOS UDS this also exercises a fragile connect-and-bail path before every root election. Extend `Aid` with backward-compatible probe and registrar capability fields, require a bounded typed handshake, and classify addresses as absent, occupied, or confirmed registrars. Reject occupied endpoints instead of binding over them. Also, - close `_connect_chan()` in `finally` - bypass normal peer tracking for election probes - preserve legacy registrar handshakes with unknown capability - cover foreign listeners and idle registrar peer state Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/discovery/test_tpt_bind_addrs.py | 82 ++++++++++++++++++++++++++ tractor/_root.py | 75 +++++++++++++++++++---- tractor/ipc/_chan.py | 32 ++++++++-- tractor/ipc/_server.py | 10 +++- tractor/msg/types.py | 2 + tractor/runtime/_runtime.py | 1 + 6 files changed, 183 insertions(+), 19 deletions(-) diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index ae3b3437..423e041a 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -12,6 +12,7 @@ bind-address selection in `_root.py`: import pytest import trio import tractor +from tractor import _root from tractor.discovery._addr import ( wrap_address, ) @@ -19,6 +20,87 @@ from tractor.discovery._multiaddr import mk_maddr from tractor._testing.addr import get_rando_addr +def test_transport_only_listener_is_not_registrar(): + ''' + Require a Tractor handshake before accepting a registry address. + + The old election probe marked an address live after transport + connect alone. A non-Tractor listener, or a registrar still + failing its initial handshake, was therefore selected as the + remote registry. This test accepts the probe and closes it without + replying, then proves `open_root_actor()` ignores that endpoint and + elects the local actor registrar instead. + + ''' + async def transport_only_handler( + stream: trio.SocketStream, + ) -> None: + await stream.aclose() + + async def main(): + listeners = await trio.open_tcp_listeners(0) + listener = listeners[0] + sockname = listener.socket.getsockname() + reg_addr: tuple[str, int] = ( + sockname[0], + sockname[1], + ) + + async with trio.open_nursery() as tn: + tn.start_soon( + trio.serve_listeners, + transport_only_handler, + listeners, + ) + with pytest.raises( + RuntimeError, + match='occupied but did not answer', + ): + async with tractor.open_root_actor( + registry_addrs=[reg_addr], + enable_transports=['tcp'], + ): + pytest.fail('foreign listener selected as registrar') + + tn.cancel_scope.cancel() + + trio.run(main) + + +def test_registry_probe_preserves_no_peers_state( + reg_addr: tuple, + tpt_proto: str, +): + ''' + Keep an idle registrar peer-free after an election probe. + + Probe handshakes exchange registrar capability but must not enter + `IPCServer._peers`. Resetting `_no_more_peers` before identifying a + probe left an idle registrar reporting phantom peers and delayed + shutdown. This test probes the live local registrar and proves its + peer map and no-peers event remain unchanged afterward. + + ''' + async def main(): + async with tractor.open_root_actor( + registry_addrs=[reg_addr], + enable_transports=[tpt_proto], + ): + actor = tractor.current_actor() + server = actor.ipc_server + + probe_status = await _root._probe_registry( + addr=wrap_address(reg_addr), + ) + assert probe_status == 'registrar' + + await trio.sleep(0) + assert not server._peers + assert server._no_more_peers.is_set() + + trio.run(main) + + # ------------------------------------------------------------------ # helpers # ------------------------------------------------------------------ diff --git a/tractor/_root.py b/tractor/_root.py index c123631b..3813b31d 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -31,6 +31,7 @@ import sys from typing import ( Any, Callable, + Literal, ) import warnings @@ -63,7 +64,9 @@ from .trionics import ( ) from ._exceptions import ( RuntimeFailure, + TransportClosed, ) +from .msg.types import Aid logger = log.get_logger('tractor') @@ -83,6 +86,45 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = ( ) +async def _probe_registry( + addr: Address, + timeout: float = 1, +) -> Literal[ + 'absent', + 'occupied', + 'registrar', +]: + ''' + Confirm an address serves the Tractor actor handshake. + + ''' + try: + with trio.move_on_after(timeout) as cs: + async with _connect_chan(addr.unwrap()) as chan: + peer_aid: Aid = await chan._do_handshake( + aid=Aid( + name='registry-probe', + uuid=mk_uuid(), + pid=os.getpid(), + is_probe=True, + ), + timeout=timeout, + ) + if peer_aid.is_registrar is not False: + return 'registrar' + return 'occupied' + + if cs.cancelled_caught: + return 'occupied' + + except OSError: + return 'absent' + except TransportClosed: + return 'occupied' + + return 'occupied' + + # TODO: stick this in a `@acm` defined in `devx.debug`? # -[ ] also maybe consider making this a `wrapt`-deco to # save an indent level? @@ -453,6 +495,7 @@ async def open_root_actor( # closed into below ping task-func ponged_addrs: list[Address] = [] + occupied_addrs: list[Address] = [] async def ping_tpt_socket( addr: Address, @@ -467,18 +510,15 @@ async def open_root_actor( server is listening at that addr. ''' - try: - # TODO: this connect-and-bail forces us to have to - # carefully rewrap TCP 104-connection-reset errors as - # EOF so as to avoid propagating cancel-causing errors - # to the channel-msg loop machinery. Likely it would - # be better to eventually have a "discovery" protocol - # with basic handshake instead? - with trio.move_on_after(timeout): - async with _connect_chan(addr.unwrap()): - ponged_addrs.append(addr) - - except OSError: + probe_status = await _probe_registry( + addr=addr, + timeout=timeout, + ) + if probe_status == 'registrar': + ponged_addrs.append(addr) + elif probe_status == 'occupied': + occupied_addrs.append(addr) + else: # ?TODO, make this a "discovery" log level? logger.info( f'No root-actor registry found @ {addr!r}\n' @@ -496,6 +536,17 @@ async def open_root_actor( addr, ) + if ( + not ponged_addrs + and + occupied_addrs + ): + raise RuntimeError( + f'Registry address(es) are occupied but did not ' + f'answer as Tractor registrars!\n' + f'occupied_addrs: {occupied_addrs!r}\n' + ) + if tpt_bind_addrs is None: tpt_bind_addrs: list[Address] = [] else: diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index a2450585..a4a4f941 100644 --- a/tractor/ipc/_chan.py +++ b/tractor/ipc/_chan.py @@ -495,6 +495,7 @@ class Channel: async def _do_handshake( self, aid: Aid, + timeout: float|None = None, ) -> Aid: ''' @@ -505,8 +506,27 @@ class Channel: "actor model" parlance. ''' - await self.send(aid) - peer_aid: Aid = await self.recv() + try: + with trio.fail_after( + timeout if timeout is not None else float('inf') + ): + await self.send(aid) + peer_aid: Aid = await self.recv() + if not isinstance(peer_aid, Aid): + raise TypeError( + f'Expected {Aid!r}, received {peer_aid!r}' + ) + except ( + MsgTypeError, + TypeError, + UnicodeDecodeError, + trio.TooSlowError, + ) as handshake_err: + raise TransportClosed( + message='Peer sent an invalid actor handshake!\n', + src_exc=handshake_err, + loglevel='warning', + ) from handshake_err log.runtime( f'Received hanshake with peer\n' f'<= {peer_aid.reprol(sin_uuid=False)}\n' @@ -529,6 +549,8 @@ async def _connect_chan( ''' chan = await Channel.from_addr(addr) - yield chan - with trio.CancelScope(shield=True): - await chan.aclose() + try: + yield chan + finally: + with trio.CancelScope(shield=True): + await chan.aclose() diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index 31f4c6b0..b430dfeb 100644 --- a/tractor/ipc/_server.py +++ b/tractor/ipc/_server.py @@ -316,8 +316,6 @@ async def handle_stream_from_peer( ) ''' - server._no_more_peers = trio.Event() # unset by making new - # TODO, debug_mode tooling for when hackin this lower layer? # with debug.maybe_open_crash_handler( # pdb=True, @@ -335,6 +333,7 @@ async def handle_stream_from_peer( if actor := _state.current_actor(): peer_aid: msgtypes.Aid = await chan._do_handshake( aid=actor.aid, + timeout=1, ) except ( TransportClosed, @@ -364,6 +363,13 @@ async def handle_stream_from_peer( ) return + # Registry election probes need only the server's `Aid` capability + # response; never register them as ordinary RPC peers. + if peer_aid.is_probe: + return + + server._no_more_peers = trio.Event() # unset by making new + uid: tuple[str, str] = ( peer_aid.name, peer_aid.uuid, diff --git a/tractor/msg/types.py b/tractor/msg/types.py index 4f3e33cc..334458c0 100644 --- a/tractor/msg/types.py +++ b/tractor/msg/types.py @@ -144,6 +144,8 @@ class Aid( name: str uuid: str pid: int|None = None + is_registrar: bool|None = None + is_probe: bool = False # TODO? can/should we extend this field set? # -[ ] use built-in support for UUIDs? `uuid.UUID` which has diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index 695090e1..67cae017 100644 --- a/tractor/runtime/_runtime.py +++ b/tractor/runtime/_runtime.py @@ -259,6 +259,7 @@ class Actor: name=name, uuid=uuid, pid=os.getpid(), + is_registrar=self.is_registrar, ) self._task: trio.Task|None = None From 6e424d4696d1af4be7d87ab9373fe05b8ada26b0 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 20:38:22 -0400 Subject: [PATCH 12/22] Use a daemon-ready sentinel in discovery tests The discovery `daemon` fixture probed UDS readiness by connecting and immediately closing. That entered Tractor's actor handshake without an `Aid` payload and destabilized the remote registrar on macOS before test roots attempted discovery. Run the child through a small `open_root_actor()` wrapper and publish a filesystem sentinel only after runtime startup completes. Poll that sentinel with process-liveness checks and guaranteed setup-failure cleanup, without touching the transport socket. Cover transport-free readiness and deterministic polling backoff. Caught-during: review remediation Found-via: `/run-tests` discovery daemon fixture consumers Cause: initial sentinel drafts mishandled `pytest.Testdir`, emitted invalid `python -c` syntax, and misplaced return-code logging. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/discovery/conftest.py | 174 ++++++++++++------------- tests/discovery/test_daemon_fixture.py | 72 ++++++++++ 2 files changed, 152 insertions(+), 94 deletions(-) create mode 100644 tests/discovery/test_daemon_fixture.py diff --git a/tests/discovery/conftest.py b/tests/discovery/conftest.py index 73749a6d..9ed318a0 100644 --- a/tests/discovery/conftest.py +++ b/tests/discovery/conftest.py @@ -13,9 +13,8 @@ under `tests/discovery/` automatically picks this up. ''' from __future__ import annotations -import os +from pathlib import Path import platform -import socket import subprocess import sys import time @@ -31,27 +30,22 @@ from ..conftest import ( def _wait_for_daemon_ready( - reg_addr: tuple, - tpt_proto: str, + ready_path: Path, *, deadline: float = 10.0, poll_interval: float = 0.05, proc: subprocess.Popen|None = None, ) -> None: ''' - Active-poll the daemon's bind address until it - accepts a connection (proving it has called - `bind() + listen()` and is ready to handle IPC). + Poll until the daemon reports completed actor startup. Replaces the historical blind `time.sleep()` in the `daemon` fixture which was racy under load — see `ai/conc-anal/test_register_duplicate_name_daemon_connect_race_issue.md`. - Uses stdlib `socket` directly (no trio runtime - bootstrap cost) — sufficient because - `tractor.run_daemon()` doesn't return from - bootstrap until the runtime is fully ready to - accept IPC. + The child writes `ready_path` only after entering + `open_root_actor()`, which guarantees all transport listeners are + serving without requiring a raw connection probe. Raises `TimeoutError` on `deadline` exceeded. If `proc` is given, ALSO raises early if the daemon @@ -70,43 +64,25 @@ def _wait_for_daemon_ready( if proc is not None and proc.poll() is not None: raise RuntimeError( f'Daemon proc exited (rc={proc.returncode}) ' - f'before becoming ready to accept on ' - f'{reg_addr!r}' + f'before reporting ready at {ready_path!r}' ) try: - if tpt_proto == 'tcp': - # `socket.create_connection` does the - # `socket() + connect()` dance with a - # builtin timeout — perfect primitive - # for a one-shot probe. - with socket.create_connection( - reg_addr, - timeout=poll_interval, - ): - return - else: - # UDS — `reg_addr` is a `(filedir, sockname)` - # tuple per `tractor.ipc._uds.UDSAddress.unwrap`. - sockpath: str = os.path.join(*reg_addr) - sock = socket.socket(socket.AF_UNIX) - try: - sock.settimeout(poll_interval) - sock.connect(sockpath) - return - finally: - sock.close() + if ready_path.is_file(): + if proc is not None and proc.poll() is not None: + raise RuntimeError( + f'Daemon proc exited (rc={proc.returncode}) ' + f'after reporting ready at {ready_path!r}' + ) + return except ( - ConnectionRefusedError, FileNotFoundError, OSError, - socket.timeout, ) as exc: last_exc = exc - time.sleep(poll_interval) + time.sleep(poll_interval) raise TimeoutError( - f'Daemon never accepted on {reg_addr!r} within ' - f'{deadline}s (last connect-attempt exc: ' - f'{last_exc!r})' + f'Daemon never reported ready at {ready_path!r} within ' + f'{deadline}s (last sentinel-state exc: {last_exc!r})' ) @@ -136,18 +112,27 @@ def daemon( ) loglevel: str = 'info' + ready_path: Path = ( + Path(str(testdir.tmpdir)) + / 'daemon-ready' + ) + ready_path.unlink(missing_ok=True) code: str = ( - "import tractor; " - "tractor.run_daemon([], " - "registry_addrs={reg_addrs}, " - "enable_transports={enable_tpts}, " - "debug_mode={debug_mode}, " - "loglevel={ll})" - ).format( - reg_addrs=str([reg_addr]), - enable_tpts=str([tpt_proto]), - ll="'{}'".format(loglevel) if loglevel else None, - debug_mode=debug_mode, + f'from pathlib import Path\n' + f'import tractor\n' + f'import trio\n' + f'\n' + f'async def main():\n' + f' async with tractor.open_root_actor(\n' + f' registry_addrs={[reg_addr]!r},\n' + f' enable_transports={[tpt_proto]!r},\n' + f' debug_mode={debug_mode!r},\n' + f' loglevel={loglevel!r},\n' + f' ):\n' + f' Path({str(ready_path)!r}).touch()\n' + f' await trio.sleep_forever()\n' + f'\n' + f'trio.run(main)\n' ) cmd: list[str] = [ sys.executable, @@ -176,48 +161,49 @@ def daemon( 15.0 if (_non_linux and ci_env) else 10.0 ) - _wait_for_daemon_ready( - reg_addr=reg_addr, - tpt_proto=tpt_proto, - deadline=deadline, - proc=proc, - ) - - assert not proc.returncode - yield proc - sig_prog(proc, _INT_SIGNAL) - - # XXX! yeah.. just be reaaal careful with this bc - # sometimes it can lock up on the `_io.BufferedReader` - # and hang.. - # - # NB, drain happens at TEARDOWN (post-yield), so the - # test body has its chance to read `proc.stderr` - # FIRST. Reading here AFTER would silently swallow - # the daemon's stderr output and break tests that - # assert on it (e.g. `test_abort_on_sigint`). - stderr: str = proc.stderr.read().decode() - stdout: str = proc.stdout.read().decode() - if ( - stderr - or - stdout - ): - print( - f'Daemon actor tree produced output:\n' - f'{proc.args}\n' - f'\n' - f'stderr: {stderr!r}\n' - f'stdout: {stdout!r}\n' + try: + _wait_for_daemon_ready( + ready_path=ready_path, + deadline=deadline, + proc=proc, ) - if (rc := proc.returncode) != -2: - msg: str = ( - f'Daemon actor tree was not cancelled !?\n' - f'proc.args: {proc.args!r}\n' - f'proc.returncode: {rc!r}\n' - ) - if rc < 0: - raise RuntimeError(msg) + assert not proc.returncode + yield proc + finally: + if proc.poll() is None: + sig_prog(proc, _INT_SIGNAL) - test_log.error(msg) + # XXX! yeah.. just be reaaal careful with this bc + # sometimes it can lock up on the `_io.BufferedReader` + # and hang.. + # + # NB, drain happens at TEARDOWN (post-yield), so the + # test body has its chance to read `proc.stderr` + # FIRST. Reading here AFTER would silently swallow + # the daemon's stderr output and break tests that + # assert on it (e.g. `test_abort_on_sigint`). + stderr: str = proc.stderr.read().decode() + stdout: str = proc.stdout.read().decode() + if ( + stderr + or + stdout + ): + print( + f'Daemon actor tree produced output:\n' + f'{proc.args}\n' + f'\n' + f'stderr: {stderr!r}\n' + f'stdout: {stdout!r}\n' + ) + + if (rc := proc.returncode) != -2: + msg: str = ( + f'Daemon actor tree was not cancelled !?\n' + f'proc.args: {proc.args!r}\n' + f'proc.returncode: {rc!r}\n' + ) + if rc < 0: + raise RuntimeError(msg) + test_log.error(msg) diff --git a/tests/discovery/test_daemon_fixture.py b/tests/discovery/test_daemon_fixture.py new file mode 100644 index 00000000..bf53758c --- /dev/null +++ b/tests/discovery/test_daemon_fixture.py @@ -0,0 +1,72 @@ +''' +Discovery daemon fixture regressions. + +''' +from unittest.mock import ( + call, + Mock, +) + +from .conftest import _wait_for_daemon_ready + + +def test_daemon_ready_check_does_not_connect( + monkeypatch, + tmp_path, +): + ''' + Detect a listening UDS daemon without creating a raw connection. + + The old UDS readiness helper connected and immediately closed. That + entered Tractor's actor-handshake handler with no `Aid` payload and + destabilized the remote registrar on macOS before discovery tests + started. This test creates the child sentinel, forbids all socket + construction and connection helpers, then proves readiness returns + without touching the transport layer. + + ''' + ready_path = tmp_path / 'daemon-ready' + ready_path.touch() + socket_ctor = Mock(side_effect=AssertionError('socket opened')) + connect = Mock(side_effect=AssertionError('socket connected')) + monkeypatch.setattr('socket.socket', socket_ctor) + monkeypatch.setattr('socket.create_connection', connect) + + _wait_for_daemon_ready( + ready_path=ready_path, + deadline=.1, + poll_interval=.01, + ) + + socket_ctor.assert_not_called() + connect.assert_not_called() + + +def test_daemon_ready_check_backs_off(monkeypatch): + ''' + Back off while waiting for the child startup sentinel. + + The sentinel may appear after several parent polling intervals. + A deterministic false/false/true path sequence proves the helper + sleeps between unsuccessful observations instead of hot-spinning + and starving a booting daemon on constrained CI workers. + + ''' + ready_path = Mock() + ready_path.is_file.side_effect = [False, False, True] + sleep = Mock() + monotonic = Mock(side_effect=[0, 0, 0, 0]) + monkeypatch.setattr('time.sleep', sleep) + monkeypatch.setattr('time.monotonic', monotonic) + + _wait_for_daemon_ready( + ready_path=ready_path, + deadline=.2, + poll_interval=.01, + ) + + assert ready_path.is_file.call_count == 3 + assert sleep.call_args_list == [ + call(.01), + call(.01), + ] From 0b63af020e82569054ab5a421f5d84e047c662f3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 23:40:12 -0400 Subject: [PATCH 13/22] Retry transient registrar handshakes A loaded runner can accept a registry transport while delaying its actor handshake beyond the first one-second attempt. Treating that timeout as final classifies a healthy daemon as occupied and cascades into unrelated discovery failures. Retry connected handshake failures on fresh channels under a shared three-second budget, while returning immediately for truly absent listeners. Bound every connect-plus-handshake attempt, use incremental backoff, and retain the one-second unauthenticated server limit. Also bound shielded probe-channel cleanup to 200ms and cover the real timeout, reconnection, backoff, fresh-channel, and stalled-close paths. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/discovery/test_tpt_bind_addrs.py | 104 +++++++++++++++++++++++++ tractor/_root.py | 69 ++++++++++------ tractor/ipc/_chan.py | 15 +++- tractor/ipc/_server.py | 4 +- 4 files changed, 166 insertions(+), 26 deletions(-) diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index 423e041a..df696888 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -9,6 +9,13 @@ bind-address selection in `_root.py`: 3. Explicit bind given -> wraps via `wrap_address()` and uses them ''' +from contextlib import asynccontextmanager as acm +from unittest.mock import ( + AsyncMock, + call, + Mock, +) + import pytest import trio import tractor @@ -20,6 +27,103 @@ from tractor.discovery._multiaddr import mk_maddr from tractor._testing.addr import get_rando_addr +def test_registry_probe_retries_transient_handshake( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Retry a connected registrar after transient handshake timeout. + + Loaded macOS runners can accept the transport while delaying the + actor handshake beyond one second. Treating that first timeout as + final makes a healthy remote daemon look occupied and cascades into + discovery failures. This deterministic fake fails once, succeeds + on the second complete handshake, and proves one bounded backoff. + + ''' + async def stall_handshake(**kwargs): + await trio.sleep_forever() + + first_handshake = AsyncMock(side_effect=stall_handshake) + second_handshake = AsyncMock( + return_value=tractor.msg.Aid( + name='registrar', + uuid='registrar-uuid', + pid=1234, + is_registrar=True, + ), + ) + chans = [ + Mock(_do_handshake=first_handshake), + Mock(_do_handshake=second_handshake), + ] + closed: list[object] = [] + + @acm + async def connect_chan(addr, close_timeout): + assert close_timeout == .2 + chan = chans[len(closed)] + try: + yield chan + finally: + closed.append(chan) + + sleep = AsyncMock() + monkeypatch.setattr(_root, '_connect_chan', connect_chan) + monkeypatch.setattr(_root.trio, 'sleep', sleep) + + async def main(): + status = await _root._probe_registry( + addr=wrap_address(('127.0.0.1', 1616)), + timeout=.3, + attempt_timeout=.1, + max_attempts=3, + retry_delay=.01, + ) + assert status == 'registrar' + + trio.run(main) + + first_handshake.assert_awaited_once() + second_handshake.assert_awaited_once() + assert first_handshake.await_args.kwargs['timeout'] == .1 + assert second_handshake.await_args.kwargs['timeout'] == .1 + assert closed == chans + sleep.assert_has_awaits([call(.01)]) + + +def test_probe_channel_close_is_bounded( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Bound shielded channel cleanup after a registry probe. + + `_connect_chan()` shields `.aclose()` so cancellation cannot leak + ordinary channels. A stalled close previously let registry probing + exceed every connect and handshake deadline. This fake close never + completes; the explicit cleanup allowance must still return control + to the caller without cancelling its surrounding task. + + ''' + chan = Mock() + chan.aclose = AsyncMock(side_effect=trio.sleep_forever) + monkeypatch.setattr( + tractor.Channel, + 'from_addr', + AsyncMock(return_value=chan), + ) + + async def main(): + with trio.fail_after(.5): + async with _root._connect_chan( + ('127.0.0.1', 1616), + close_timeout=.01, + ): + pass + + trio.run(main) + chan.aclose.assert_awaited_once() + + def test_transport_only_listener_is_not_registrar(): ''' Require a Tractor handshake before accepting a registry address. diff --git a/tractor/_root.py b/tractor/_root.py index 3813b31d..a8316f49 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -88,7 +88,11 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = ( async def _probe_registry( addr: Address, - timeout: float = 1, + timeout: float = 3, + attempt_timeout: float = 1, + max_attempts: int = 3, + retry_delay: float = .05, + close_timeout: float = .2, ) -> Literal[ 'absent', 'occupied', @@ -97,30 +101,49 @@ async def _probe_registry( ''' Confirm an address serves the Tractor actor handshake. + Connection and handshake work share `timeout`; each attempt gets + `attempt_timeout`. Shielded channel cleanup may consume at most one + additional `close_timeout` after either deadline fires. + ''' - try: - with trio.move_on_after(timeout) as cs: - async with _connect_chan(addr.unwrap()) as chan: - peer_aid: Aid = await chan._do_handshake( - aid=Aid( - name='registry-probe', - uuid=mk_uuid(), - pid=os.getpid(), - is_probe=True, - ), - timeout=timeout, + connected_once: bool = False + with trio.move_on_after(timeout): + for attempt in range(max_attempts): + try: + with trio.move_on_after(attempt_timeout) as attempt_cs: + async with _connect_chan( + addr.unwrap(), + close_timeout=close_timeout, + ) as chan: + connected_once = True + peer_aid: Aid = await chan._do_handshake( + aid=Aid( + name='registry-probe', + uuid=mk_uuid(), + pid=os.getpid(), + is_probe=True, + ), + timeout=attempt_timeout, + ) + if peer_aid.is_registrar is not False: + return 'registrar' + return 'occupied' + + if attempt_cs.cancelled_caught: + if not connected_once: + return 'absent' + + except OSError: + return ( + 'occupied' + if connected_once + else 'absent' ) - if peer_aid.is_registrar is not False: - return 'registrar' - return 'occupied' + except TransportClosed: + pass - if cs.cancelled_caught: - return 'occupied' - - except OSError: - return 'absent' - except TransportClosed: - return 'occupied' + if attempt + 1 < max_attempts: + await trio.sleep(retry_delay * (attempt + 1)) return 'occupied' @@ -499,7 +522,7 @@ async def open_root_actor( async def ping_tpt_socket( addr: Address, - timeout: float = 1, + timeout: float = 3, ) -> None: ''' Attempt temporary connection to see if a registry is diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index a4a4f941..bf9a05f0 100644 --- a/tractor/ipc/_chan.py +++ b/tractor/ipc/_chan.py @@ -538,7 +538,8 @@ class Channel: @acm async def _connect_chan( - addr: UnwrappedAddress + addr: UnwrappedAddress, + close_timeout: float|None = None, ) -> typing.AsyncGenerator[Channel, None]: ''' Create and connect a `Channel` to the provided `addr`, disconnect @@ -553,4 +554,14 @@ async def _connect_chan( yield chan finally: with trio.CancelScope(shield=True): - await chan.aclose() + if close_timeout is None: + await chan.aclose() + else: + with trio.move_on_after(close_timeout) as close_cs: + await chan.aclose() + if close_cs.cancelled_caught: + log.warning( + f'Timed out closing channel after ' + f'{close_timeout}s\n' + f'|_{chan}\n' + ) diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index b430dfeb..b0ada242 100644 --- a/tractor/ipc/_server.py +++ b/tractor/ipc/_server.py @@ -72,6 +72,8 @@ if TYPE_CHECKING: log = log.get_logger() +_PRE_REG_HANDSHAKE_TIMEOUT: float = 1 + async def maybe_wait_on_canced_subs( uid: tuple[str, str], @@ -333,7 +335,7 @@ async def handle_stream_from_peer( if actor := _state.current_actor(): peer_aid: msgtypes.Aid = await chan._do_handshake( aid=actor.aid, - timeout=1, + timeout=_PRE_REG_HANDSHAKE_TIMEOUT, ) except ( TransportClosed, From 1a9ce915f3530c69b8aea7602abef18182fb0c22 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 10:34:59 -0400 Subject: [PATCH 14/22] Harden inbound actor handshakes Registry probes need short retry deadlines, but applying their one-second budget to every portal and child connection can terminate a valid delayed actor with no client retry path. Give ordinary pre-registration handshakes an independent ten-second deadline. Normalize raw `msgspec.DecodeError` frames to `TransportClosed` so malformed peers cannot cancel the shared IPC nursery with decoder internals. Cover malformed frames and the ordinary-vs-probe timeout distinction. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_server.py | 96 +++++++++++++++++++++++++++++++++++++++- tractor/ipc/_chan.py | 2 + tractor/ipc/_server.py | 11 ++--- 3 files changed, 100 insertions(+), 9 deletions(-) diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py index 01ddb222..6186454e 100644 --- a/tests/ipc/test_server.py +++ b/tests/ipc/test_server.py @@ -4,7 +4,12 @@ High-level `.ipc._server` unit tests. ''' from __future__ import annotations import errno +from unittest.mock import ( + AsyncMock, + Mock, +) +import msgspec import pytest import trio from tractor import ( @@ -16,7 +21,10 @@ from tractor._testing.addr import ( get_rando_addr, ) from tractor._exceptions import TransportClosed +from tractor.ipc._chan import Channel +from tractor.ipc import _server from tractor.ipc._transport import MsgpackTransport +from tractor.msg.types import Aid # TODO, use/check-roundtripping with some of these wrapper types? # # from .._addr import Address @@ -30,8 +38,8 @@ def test_send_normalizes_peer_reset(): ''' Normalize Darwin's pre-handshake peer reset as transport closure. - A raw UDS readiness client connects and immediately disconnects. - Darwin reports the server's first handshake write as + A UDS peer may disconnect before completing the actor handshake. + Darwin can report the server's first handshake write as `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing that raw error to escape cancels the daemon's shared IPC nursery. This fake stream reproduces the exact exception chain and proves @@ -69,6 +77,90 @@ def test_send_normalizes_peer_reset(): trio.run(main) +def test_handshake_normalizes_decode_error(): + ''' + Keep malformed pre-handshake frames out of the service nursery. + + A non-msgpack peer can trigger `msgspec.DecodeError` before a + remote `Aid` exists. Letting that decoder error escape the inbound + handler cancels the actor's shared IPC nursery. This fake channel + proves `_do_handshake()` presents only `TransportClosed` upward. + + ''' + chan = object.__new__(Channel) + chan.send = AsyncMock() + chan.recv = AsyncMock( + side_effect=msgspec.DecodeError('malformed handshake'), + ) + + async def main(): + with pytest.raises(TransportClosed) as exc_info: + await chan._do_handshake( + aid=Aid( + name='local', + uuid='local-uuid', + pid=1234, + ), + timeout=.1, + ) + + assert isinstance( + exc_info.value.src_exc, + msgspec.DecodeError, + ) + + trio.run(main) + + +def test_server_uses_independent_handshake_timeout( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Give ordinary actor handshakes a distinct, generous deadline. + + Registry probes use short retries, but ordinary portal and child + connections do not retry. Applying the probe's one-second timeout + in the server can terminate a valid delayed child and leave its + parent blocked in `IPCServer.wait_for_peer()`. This handler fake + proves the server uses its separate pre-registration budget. + + ''' + handshake = AsyncMock( + side_effect=TransportClosed(message='stop after assertion'), + ) + chan = Mock(_do_handshake=handshake) + actor = Mock( + aid=Aid( + name='local', + uuid='local-uuid', + pid=1234, + ), + ) + monkeypatch.setattr( + Channel, + 'from_stream', + Mock(return_value=chan), + ) + monkeypatch.setattr( + _server._state, + 'current_actor', + Mock(return_value=actor), + ) + + async def main(): + await _server.handle_stream_from_peer( + stream=Mock(), + server=Mock(), + ) + + trio.run(main) + handshake.assert_awaited_once_with( + aid=actor.aid, + timeout=_server._PRE_REG_HANDSHAKE_TIMEOUT, + ) + assert _server._PRE_REG_HANDSHAKE_TIMEOUT == 10 + + @pytest.mark.parametrize( '_tpt_proto', ['uds', 'tcp'] diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index bf9a05f0..6eadf381 100644 --- a/tractor/ipc/_chan.py +++ b/tractor/ipc/_chan.py @@ -33,6 +33,7 @@ from typing import ( ) import warnings +import msgspec import trio from ._types import ( @@ -518,6 +519,7 @@ class Channel: ) except ( MsgTypeError, + msgspec.DecodeError, TypeError, UnicodeDecodeError, trio.TooSlowError, diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index b0ada242..80f85841 100644 --- a/tractor/ipc/_server.py +++ b/tractor/ipc/_server.py @@ -72,7 +72,7 @@ if TYPE_CHECKING: log = log.get_logger() -_PRE_REG_HANDSHAKE_TIMEOUT: float = 1 +_PRE_REG_HANDSHAKE_TIMEOUT: float = 10 async def maybe_wait_on_canced_subs( @@ -352,12 +352,9 @@ async def handle_stream_from_peer( # "kinda-error" that we expect to tolerate during # discovery-sys related pings, queires, DoS etc. ): - # XXX: This may propagate up from `Channel._aiter_recv()` - # and `MsgpackStream._inter_packets()` on a read from the - # stream particularly when the runtime is first starting up - # inside `open_root_actor()` where there is a check for - # a bound listener on the registrar addr. the reset will be - # because the handshake was never meant took place. + # `TransportClosed` is expected when a peer disconnects or + # fails the initial typed handshake, including foreign clients + # and probes racing shutdown. log.runtime( con_status + From 16dd876b0c457ffa2e665e64ebdaf91f73413b10 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 10:36:27 -0400 Subject: [PATCH 15/22] Make UDS reaping platform-aware Moving Darwin sockets to `/tmp/tractor-` left pytest and the standalone reaper searching only `XDG_RUNTIME_DIR`. Enabling that shared bindspace naively would also let automatic session cleanup unlink an independent live `registry@1616.sock`. Resolve the runtime's actual default UDS bindspace on every platform. Exclude the pid-less registry sentinel from automatic cleanup while retaining explicit CLI removal, and document that destructive choice. Cover bindspace resolution and automatic-vs-explicit sentinel policy. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- docs/guide/testing.rst | 6 ++-- scripts/tractor-reap | 16 +++++---- tests/ipc/test_each_tpt.py | 69 ++++++++++++++++++++++++++++++++++++++ tractor/_testing/_reap.py | 46 +++++++++++-------------- 4 files changed, 102 insertions(+), 35 deletions(-) diff --git a/docs/guide/testing.rst b/docs/guide/testing.rst index d07415b8..8a99a637 100644 --- a/docs/guide/testing.rst +++ b/docs/guide/testing.rst @@ -185,8 +185,10 @@ first with a bounded grace window — so actor runtimes can run their ``trio`` teardown paths — escalating to ``SIGKILL`` only as a last resort. The ``--shm`` sweep unlinks ``/dev/shm/`` segments that no live process has open (it leans on psutil_, already in -your dev venv, to check live mappings and fds) and ``--uds`` -clears socket files whose binder pid is dead. +your dev venv, to check live mappings and fds) and ``--uds`` clears +dead-binder sockets from Tractor's platform-specific runtime dir. It +also unconditionally removes ``registry@1616.sock``; do not run the UDS +sweep while a live registrar is serving from that default address. Testing your own ``tractor`` app -------------------------------- diff --git a/scripts/tractor-reap b/scripts/tractor-reap index 11ad8e09..60db0470 100755 --- a/scripts/tractor-reap +++ b/scripts/tractor-reap @@ -23,10 +23,10 @@ Two cleanup phases (run in order when both are enabled): hard-crashing actor leaves leaked segments that nothing else GCs. -3. **UDS sweep** (`--uds` / `--uds-only`) — unlinks - `${XDG_RUNTIME_DIR}/tractor/@.sock` files - whose binder pid is dead (or the `1616` registry - sentinel). Needed because the IPC server's +3. **UDS sweep** (`--uds` / `--uds-only`) — unlinks socket + files from Tractor's platform-specific default bindspace whose + binder pid is dead (or the `1616` registry sentinel). Needed + because the IPC server's `os.unlink()` cleanup lives in a `finally:` block that doesn't always run on hard exits (SIGKILL, escaped `KeyboardInterrupt`, etc.) — see issue #452. @@ -137,8 +137,8 @@ def main() -> int: action='store_true', help=( 'after process reap, also unlink orphaned ' - '${XDG_RUNTIME_DIR}/tractor/*.sock files ' - 'whose binder pid is dead (or the 1616 ' + 'sockets from Tractor\'s platform default ' + 'bindspace whose binder pid is dead (or the 1616 ' 'registry sentinel). See issue #452.' ), ) @@ -212,7 +212,9 @@ def main() -> int: # --- phase 3: UDS sweep (opt-in) --- if args.uds or args.uds_only: - leaked_uds: list[str] = find_orphaned_uds() + leaked_uds: list[str] = find_orphaned_uds( + include_registry_sentinel=True, + ) if not leaked_uds: print( '[tractor-reap] no orphaned UDS sock-files ' diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index f27fed5d..c52dc3cd 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -5,6 +5,7 @@ Unit-ish tests for specific IPC transport protocol backends. from __future__ import annotations import os from pathlib import Path +import socket import stat import sys from types import SimpleNamespace @@ -99,6 +100,74 @@ def test_macos_rt_dir_rejects_symlink( assert stat.S_IMODE(target_dir.stat().st_mode) == 0o755 +def test_reaper_uses_default_uds_bindspace( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Sweep the same platform-specific bindspace used by UDS actors. + + The reaper previously consulted only `XDG_RUNTIME_DIR`, missing + Darwin sockets after the runtime moved to `/tmp/tractor-`. + This test replaces `UDSAddress.def_bindspace` and proves the test + harness resolves that shared transport default directly. + + ''' + from tractor._testing import _reap + from tractor.ipc._uds import UDSAddress + + monkeypatch.setattr( + UDSAddress, + 'def_bindspace', + tmp_path, + ) + + assert _reap.get_uds_dir() == str(tmp_path) + + +def test_automatic_reaper_preserves_registry_sentinel( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Reserve unconditional registry cleanup for the explicit CLI. + + The `registry@1616.sock` suffix does not encode its binder PID, so + automatic pytest cleanup cannot distinguish a leak from another + live registrar. This test creates registry and actor sockets, + proves the default sweep selects only the dead actor, then proves + explicit sentinel inclusion retains the CLI's documented behavior. + + ''' + from tractor._testing import _reap + + registry_path: Path = tmp_path / 'registry@1616.sock' + actor_path: Path = tmp_path / 'worker@1234.sock' + socks: list[socket.socket] = [] + for path in (registry_path, actor_path): + sock = socket.socket(socket.AF_UNIX) + sock.bind(str(path)) + socks.append(sock) + + monkeypatch.setattr(_reap, '_is_alive', lambda pid: False) + try: + assert _reap.find_orphaned_uds( + uds_dir=str(tmp_path), + ) == [str(actor_path)] + assert set( + _reap.find_orphaned_uds( + uds_dir=str(tmp_path), + include_registry_sentinel=True, + ) + ) == { + str(registry_path), + str(actor_path), + } + finally: + for sock in socks: + sock.close() + + def test_rt_dir_rejects_non_directory( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, diff --git a/tractor/_testing/_reap.py b/tractor/_testing/_reap.py index e9511e96..bb7ba0ec 100644 --- a/tractor/_testing/_reap.py +++ b/tractor/_testing/_reap.py @@ -111,14 +111,13 @@ SHM_DIR: str = '/dev/shm' # UDS-socket leak sweep — see `find_orphaned_uds()` / # `reap_uds()` below. Tractor's UDS transport -# (`tractor.ipc._uds`) creates sock files under -# `${XDG_RUNTIME_DIR}/tractor/@.sock`; a +# (`tractor.ipc._uds`) creates sock files in its platform-specific +# default bindspace; a # crash / SIGKILL / mid-cancel teardown can leave the # file behind because `os.unlink()` lives in the # `_serve_ipc_eps` `finally:` block which doesn't always # get to run on hard exits. The reaper here is best-effort # cleanup for the test harness + the `tractor-reap` CLI. -_UDS_SUBDIR: str = 'tractor' # `@.sock` — pid is the binder's pid at # creation time. Special sentinel: `registry@1616.sock` # uses the magic `1616` not a real pid (the root @@ -738,19 +737,16 @@ def reap_shm( def get_uds_dir() -> str|None: ''' - Path of tractor's per-user UDS sock-file dir - (`${XDG_RUNTIME_DIR}/tractor/`). + Path of Tractor's platform-specific default UDS bindspace. - Returns `None` when `XDG_RUNTIME_DIR` is unset (e.g. - non-systemd hosts, or inside a container without the - var plumbed through). Caller should treat that as - "no UDS leaks possible to detect — skip". + Returns `None` only when the bindspace cannot be resolved. ''' - xdg: str|None = os.environ.get('XDG_RUNTIME_DIR') - if not xdg: + try: + from tractor.ipc._uds import UDSAddress + return str(UDSAddress.def_bindspace) + except Exception: return None - return os.path.join(xdg, _UDS_SUBDIR) def _parse_uds_name(filename: str) -> tuple[str, int]|None: @@ -768,16 +764,16 @@ def _parse_uds_name(filename: str) -> tuple[str, int]|None: def find_orphaned_uds( *, uds_dir: str|None = None, + include_registry_sentinel: bool = False, ) -> list[str]: ''' `/*.sock` paths whose binder pid is no - longer alive (orphaned). Includes the - `registry@1616.sock` sentinel — `1616` is a magic - sentinel pid (not a real one) so the file's - presence alone signals a leak from a dead session. + longer alive (orphaned). Explicit callers may include the + `registry@1616.sock` sentinel; automatic pytest cleanup excludes + it because binder liveness cannot be inferred from magic `1616`. - Returns `[]` on platforms without `XDG_RUNTIME_DIR` - or when the dir doesn't exist. Files whose name + Returns `[]` when the platform bindspace cannot be resolved or the + dir doesn't exist. Files whose name doesn't match the `@.sock` pattern are skipped (we don't unlink things we don't recognize). @@ -811,10 +807,8 @@ def find_orphaned_uds( continue _name, pid = parsed if pid == _UDS_REGISTRY_SENTINEL_PID: - # sentinel — never a real pid; if the file - # exists nobody live is "owning" it via - # /proc lookup, so always orphaned - leaked.append(path) + if include_registry_sentinel: + leaked.append(path) continue if not _is_alive(pid): leaked.append(path) @@ -933,8 +927,8 @@ def track_orphaned_uds_per_test(): teardown that flakifies sibling tests via sock-file rebind races). - Snapshots `${XDG_RUNTIME_DIR}/tractor/` before and - after each test; any `@.sock` files + Snapshots Tractor's platform-specific default UDS bindspace before + and after each test; any `@.sock` files created during the test that survive teardown AND whose creator pid is dead are surfaced as a loud warning AND reaped, so the next test starts with a @@ -950,8 +944,8 @@ def track_orphaned_uds_per_test(): it (vs. blanket session-end sweep) makes blame obvious + prevents cascade flakiness. - Cheap: 2x `os.listdir` + a few `os.stat`s per test. - Skips silently when `XDG_RUNTIME_DIR` isn't set. + Cheap: 2x `os.listdir` + a few `os.stat`s per test. Skips silently + when the platform bindspace cannot be resolved. ''' uds_dir: str|None = get_uds_dir() From 584ea4e9add1f084a4fca03bf43b3251dfb3f925 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 10:37:19 -0400 Subject: [PATCH 16/22] Document stable macOS UDS operation The remediation grew beyond the original no-autobind fix, leaving public docs and nearby comments describing connect-only discovery, XDG-only socket paths, raw readiness probes, and old cleanup naming. Document typed registrar probing, occupied-address rejection, Darwin's short runtime directory, platform-aware socket cleanup, sentinel readiness, and subprocess output draining. Add the GH #473 bugfix fragment and update regression rationale without changing task states. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- docs/explain/architecture.rst | 7 ++++--- docs/guide/discovery.rst | 20 +++++++++++-------- nooz/473.bugfix.rst | 4 ++++ tests/discovery/conftest.py | 27 ++++++++++++-------------- tests/discovery/test_daemon_fixture.py | 2 +- tests/discovery/test_tpt_bind_addrs.py | 4 ++-- tests/test_docs_examples.py | 10 +++++----- tractor/_root.py | 12 +++++------- tractor/ipc/_uds.py | 2 +- tractor/runtime/_state.py | 6 +++--- tractor/spawn/_reap.py | 14 ++++++------- 11 files changed, 56 insertions(+), 52 deletions(-) create mode 100644 nooz/473.bugfix.rst diff --git a/docs/explain/architecture.rst b/docs/explain/architecture.rst index 97d9fcdc..091647f3 100644 --- a/docs/explain/architecture.rst +++ b/docs/explain/architecture.rst @@ -130,9 +130,10 @@ UDS: same-host, creds included Pass ``enable_transports=['uds']`` and actors instead talk over unix-domain sockets, with socket files placed in the per-user -runtime dir (``$XDG_RUNTIME_DIR/tractor/`` on linux, the -``platformdirs`` equivalent elsewhere). Two perks over tcp on a -single host: +runtime dir: ``$XDG_RUNTIME_DIR/tractor/`` on linux, a short +owner-only ``/tmp/tractor-`` dir on Darwin, and the +``platformdirs`` equivalent elsewhere. Two perks over tcp on a single +host: - no ports to fight over; addrs are just file paths, - the kernel snitches on your peer for free: the listening side diff --git a/docs/guide/discovery.rst b/docs/guide/discovery.rst index cc76a7f3..ea993b52 100644 --- a/docs/guide/discovery.rst +++ b/docs/guide/discovery.rst @@ -44,8 +44,9 @@ clan shares one registry with zero config on your part. The bootstrap rule inside ``open_root_actor()`` is delightfully simple: -- on boot, ping every socket addr in ``registry_addrs``; when none - are passed the per-transport defaults are used: for TCP the +- on boot, probe every addr in ``registry_addrs`` with a bounded + Tractor ``Aid`` handshake; when none are passed the per-transport + defaults are used: for TCP the loopback ``('127.0.0.1', 1616)``, for UDS a ``registry@1616.sock`` file, @@ -53,9 +54,11 @@ simple: actor and register with the *existing* registry; your own IPC server binds random same-transport addrs instead, -- if **nothing answers, congratulations: you just became the - registrar**. Your transport server binds the registry addrs - themselves and you start serving lookups for everyone else. +- if every address is absent, congratulations: you just became the + registrar. Your transport server binds the registry addrs + themselves and you start serving lookups for everyone else, +- if no registrar answers but an address is occupied by a foreign or + non-responsive endpoint, startup fails instead of binding over it. Pass ``ensure_registry=True`` when your program *requires* being the one-and-only registrar; boot then fails loudly with a @@ -196,9 +199,10 @@ the existing registrar: trio.run(main) -Per the bootstrap rules above, if the registrar at those addrs is -*not* reachable this process simply becomes its own (registrar) -root — so the same code works standalone and as a tree-joiner. +Per the bootstrap rules above, if those addrs are absent this process +becomes its own registrar root, so the same code works standalone and +as a tree-joiner. An occupied address that does not complete a Tractor +registrar handshake fails startup instead of being rebound. "Arbiter"? A legacy naming note ------------------------------- diff --git a/nooz/473.bugfix.rst b/nooz/473.bugfix.rst new file mode 100644 index 00000000..1736add6 --- /dev/null +++ b/nooz/473.bugfix.rst @@ -0,0 +1,4 @@ +Fix Unix-domain-socket actor trees and registrar discovery on macOS. +Runtime sockets now use a short, owner-only runtime directory, +generated socket names remain within platform limits, and transient +or reset pre-handshake connections no longer destabilize discovery. diff --git a/tests/discovery/conftest.py b/tests/discovery/conftest.py index 9ed318a0..2a7320c6 100644 --- a/tests/discovery/conftest.py +++ b/tests/discovery/conftest.py @@ -1,12 +1,10 @@ ''' -Discovery-suite fixtures, including the `daemon` -remote-registrar subprocess used by the multi-program -discovery tests. +Discovery-suite fixtures, including the `daemon` remote-registrar +subprocess used by the multi-program discovery tests. Lives here (vs. the parent `tests/conftest.py`) -because `daemon` is a discovery-protocol primitive — -boots a separate `tractor.run_daemon()` process whose -sole purpose is to serve as a registrar peer for +because `daemon` is a discovery-protocol primitive: it boots a child +that enters `open_root_actor()` and waits as a registrar peer for discovery-roundtrip tests. Pytest fixtures inherit DOWNWARD through conftest hierarchy, so anything under `tests/discovery/` automatically picks this up. @@ -49,9 +47,8 @@ def _wait_for_daemon_ready( Raises `TimeoutError` on `deadline` exceeded. If `proc` is given, ALSO raises early if the daemon - process exits non-zero before the deadline (catches - daemon-startup-crash that the blind sleep used to - silently mask). + process exits before the deadline (catches a daemon startup crash + that the blind sleep used to silently mask). ''' end: float = time.monotonic() + deadline @@ -148,9 +145,9 @@ def daemon( **kwargs, ) - # Active-poll the daemon's bind address until it's - # ready to accept connections — replaces the legacy - # blind `time.sleep(2.2)` which was racy under load + # Poll the child's ready sentinel, published after actor startup, + # instead of connecting to its transport socket. This replaces + # the legacy blind `time.sleep(2.2)` which was racy under load # (see # `ai/conc-anal/test_register_duplicate_name_daemon_connect_race_issue.md`). # @@ -174,9 +171,9 @@ def daemon( if proc.poll() is None: sig_prog(proc, _INT_SIGNAL) - # XXX! yeah.. just be reaaal careful with this bc - # sometimes it can lock up on the `_io.BufferedReader` - # and hang.. + # NOTE: these blocking reads can hang when descendants retain + # inherited pipe descriptors. Keep teardown signaling above + # them and avoid adding subprocesses outside the actor tree. # # NB, drain happens at TEARDOWN (post-yield), so the # test body has its chance to read `proc.stderr` diff --git a/tests/discovery/test_daemon_fixture.py b/tests/discovery/test_daemon_fixture.py index bf53758c..d8899e42 100644 --- a/tests/discovery/test_daemon_fixture.py +++ b/tests/discovery/test_daemon_fixture.py @@ -15,7 +15,7 @@ def test_daemon_ready_check_does_not_connect( tmp_path, ): ''' - Detect a listening UDS daemon without creating a raw connection. + Observe completed daemon startup without a raw connection. The old UDS readiness helper connected and immediately closed. That entered Tractor's actor-handshake handler with no `Aid` payload and diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index df696888..ca65587e 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -132,8 +132,8 @@ def test_transport_only_listener_is_not_registrar(): connect alone. A non-Tractor listener, or a registrar still failing its initial handshake, was therefore selected as the remote registry. This test accepts the probe and closes it without - replying, then proves `open_root_actor()` ignores that endpoint and - elects the local actor registrar instead. + replying, then proves `open_root_actor()` rejects that occupied + endpoint instead of selecting it or binding over it. ''' async def transport_only_handler( diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index 7d9b6418..27ec4146 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -81,8 +81,9 @@ def _wait_for_proc( errmsg: str = err.decode(errors='replace') - # XXX, ALWAYS surface the subproc's full stderr - # whenever it exits non-zero! + # NOTE: always include captured stdout and stderr for a non-zero + # exit. Depending on the final stderr line previously hid grouped + # exception diagnostics; see GH #473. # # The prior impl only raised when the LAST stderr # line contained 'Error', swallowing any crash whose @@ -246,9 +247,8 @@ def run_example_in_subproc( str(script_file), ] - # XXX: BE FOREVER WARNED: if you enable lots of tractor logging - # in the subprocess it may cause infinite blocking on the pipes - # due to backpressure!!! + # Captured pipes are drained by `_wait_for_proc()` while the + # example runs. proc = testdir.popen( cmdargs, stdin=subprocess.PIPE, diff --git a/tractor/_root.py b/tractor/_root.py index a8316f49..28696918 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -102,8 +102,8 @@ async def _probe_registry( Confirm an address serves the Tractor actor handshake. Connection and handshake work share `timeout`; each attempt gets - `attempt_timeout`. Shielded channel cleanup may consume at most one - additional `close_timeout` after either deadline fires. + `attempt_timeout`. Shielded cleanup may add up to `close_timeout` + per attempted channel. ''' connected_once: bool = False @@ -525,12 +525,10 @@ async def open_root_actor( timeout: float = 3, ) -> None: ''' - Attempt temporary connection to see if a registry is - listening at the requested address by a tranport layer - ping. + Probe with a bounded Tractor actor handshake. - If a connection can't be made quickly we assume none no - server is listening at that addr. + Classify the address as a registrar, occupied by a + non-registrar, or absent. ''' probe_status = await _probe_registry( diff --git a/tractor/ipc/_uds.py b/tractor/ipc/_uds.py index f074820d..1986ce57 100644 --- a/tractor/ipc/_uds.py +++ b/tractor/ipc/_uds.py @@ -656,7 +656,7 @@ class MsgpackUDSStream(MsgpackTransport): case (bytes(), str()): sock_path: Path = Path(sockname) - # XXX, no-autobind case (macOS): the un-bound end + # NOTE, no-autobind case (macOS): the un-bound end # is `''`, NOT a `bytes` abstract-ns addr; taking # `peername` unconditionally (as prior impl did) # delivers garbage `Path('')` addrs on the accept diff --git a/tractor/runtime/_state.py b/tractor/runtime/_state.py index 154266bb..38b64260 100644 --- a/tractor/runtime/_state.py +++ b/tractor/runtime/_state.py @@ -332,9 +332,9 @@ def get_rt_dir( userspace apps stick their IPC and cache related system util-files. - On linux we use a `${XDG_RUNTIME_DIR}/tractor/` subdir by - default, but equivalents are mapped for each platform using - the lovely `platformdirs` lib. + Linux uses `${XDG_RUNTIME_DIR}/tractor/`; Darwin uses a short, + owner-only `/tmp/tractor-` path; other platforms use the + lovely `platformdirs` lib. ''' # lazy-imported to keep it off the eager diff --git a/tractor/spawn/_reap.py b/tractor/spawn/_reap.py index 5dfff2e7..09358bdd 100644 --- a/tractor/spawn/_reap.py +++ b/tractor/spawn/_reap.py @@ -35,14 +35,14 @@ Future-work TODO — authoritative UDS bind-addr tracking `unlink_uds_bind_addrs()` currently has two cleanup paths: 1. Explicit `bind_addrs` (when parent set them at spawn time) -2. **Convention-based reconstruction** — - `/tractor/@.sock` — for the +2. **Convention-based reconstruction** in the platform default UDS + bindspace — for the common case where the subactor self-assigned a random sock via `UDSAddress.get_random()`. -Path (2) hardcodes the `@.sock` convention from -`tractor.ipc._uds.UDSAddress`. If that convention ever -changes — or the subactor binds to a non-default +Path (2) delegates filename reconstruction to +`tractor.ipc._uds.UDSAddress.get_sockname()`. If the subactor binds to +a non-default `bindspace`/`filedir` — we'll silently fail to unlink. A more authoritative approach would be: @@ -105,7 +105,7 @@ def unlink_uds_bind_addrs( `_serve_ipc_eps` `finally:` block (which normally calls `os.unlink(addr.sockpath)`) never runs. Without this parent-side cleanup, the dead subactor's - `${XDG_RUNTIME_DIR}/tractor/@.sock` file + platform-default UDS socket file accumulates on the filesystem (see issue #454 + the autouse `_track_orphaned_uds_per_test` fixture). @@ -119,7 +119,7 @@ def unlink_uds_bind_addrs( picked its own random sock via `UDSAddress.get_random()`), reconstruct the path from `(subactor.aid.name, proc.pid)` using the - same `@.sock` convention. We can do this + same `UDSAddress.get_sockname()` helper. We can do this because the subactor uses its OWN `os.getpid()` at bind time, which equals `proc.pid` from the parent's view. From ac61d7a5bfcd85acbea2996fa3688bf631fcd7ab Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 10:59:07 -0400 Subject: [PATCH 17/22] Use a short UDS reaper test bindspace Darwin's pytest `tmp_path` can already exceed the 104-byte AF_UNIX budget before appending either synthetic socket filename. That made the new sentinel-policy regression fail identically in both macOS matrix legs without exercising reaper behavior. Allocate the test's real socket files under a short `/tmp/tractor-reap-*` directory and retain scoped cleanup. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 55 +++++++++++++++++++++----------------- 1 file changed, 30 insertions(+), 25 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index c52dc3cd..b62e2341 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -8,6 +8,7 @@ from pathlib import Path import socket import stat import sys +import tempfile from types import SimpleNamespace from unittest.mock import Mock @@ -127,7 +128,6 @@ def test_reaper_uses_default_uds_bindspace( def test_automatic_reaper_preserves_registry_sentinel( monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, ): ''' Reserve unconditional registry cleanup for the explicit CLI. @@ -141,31 +141,36 @@ def test_automatic_reaper_preserves_registry_sentinel( ''' from tractor._testing import _reap - registry_path: Path = tmp_path / 'registry@1616.sock' - actor_path: Path = tmp_path / 'worker@1234.sock' - socks: list[socket.socket] = [] - for path in (registry_path, actor_path): - sock = socket.socket(socket.AF_UNIX) - sock.bind(str(path)) - socks.append(sock) + with tempfile.TemporaryDirectory( + prefix='tractor-reap-', + dir='/tmp', + ) as tmpdir: + bindspace: Path = Path(tmpdir) + registry_path: Path = bindspace / 'registry@1616.sock' + actor_path: Path = bindspace / 'worker@1234.sock' + socks: list[socket.socket] = [] + for path in (registry_path, actor_path): + sock = socket.socket(socket.AF_UNIX) + sock.bind(str(path)) + socks.append(sock) - monkeypatch.setattr(_reap, '_is_alive', lambda pid: False) - try: - assert _reap.find_orphaned_uds( - uds_dir=str(tmp_path), - ) == [str(actor_path)] - assert set( - _reap.find_orphaned_uds( - uds_dir=str(tmp_path), - include_registry_sentinel=True, - ) - ) == { - str(registry_path), - str(actor_path), - } - finally: - for sock in socks: - sock.close() + monkeypatch.setattr(_reap, '_is_alive', lambda pid: False) + try: + assert _reap.find_orphaned_uds( + uds_dir=str(bindspace), + ) == [str(actor_path)] + assert set( + _reap.find_orphaned_uds( + uds_dir=str(bindspace), + include_registry_sentinel=True, + ) + ) == { + str(registry_path), + str(actor_path), + } + finally: + for sock in socks: + sock.close() def test_rt_dir_rejects_non_directory( From ebf2258b4fcb1d48b380c401c5c22d8fc38f6fd8 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 12:28:56 -0400 Subject: [PATCH 18/22] Keep accepted RPCs alive on `TransportClosed` An RPC caller can close its channel after the callee creates the endpoint coro but before its `StartAck` or final response lands. Treat those response-send failures as terminal delivery failures so the accepted endpoint still runs and application errors stay local instead of cancelling the shared service nursery. Register each cancellable RPC in `Actor._rpc_tasks` before publishing its `Context` through `TaskStatus.started()`. This closes the checkpoint-free completion race without a cross-task handoff and keeps `Actor._ongoing_rpc_tasks` balanced through existing cleanup. Add regressions for caller disconnects at `StartAck` and error shipment, plus registration-before-execution and cleanup checks. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/test_rpc.py | 127 ++++++++++++++++++++++++++++++++++++++++ tractor/runtime/_rpc.py | 92 +++++++++++++++++++++-------- 2 files changed, 195 insertions(+), 24 deletions(-) diff --git a/tests/test_rpc.py b/tests/test_rpc.py index 6e2b414c..440e14c6 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -4,11 +4,18 @@ related API and error checks. ''' import itertools +from unittest.mock import ( + AsyncMock, + Mock, +) import pytest import tractor import trio +from tractor._exceptions import TransportClosed +from tractor.runtime import _rpc + async def sleep_back_actor( actor_name, @@ -46,6 +53,126 @@ async def short_sleep(): await trio.sleep(0) +def test_rpc_runs_after_startack_disconnect(): + ''' + Complete an accepted RPC when its caller closes before `StartAck`. + + Registrar teardown opens short-lived `unregister_actor` RPCs. A + loaded caller can close its channel while the registrar sends the + acknowledgement; normalized `TransportClosed` previously escaped + into the shared service nursery before the already-created + coroutine was awaited. This fake fails the first response send and + proves the RPC side effect still runs with no later send attempt. + + ''' + async def main(): + rpc_ran = trio.Event() + + chan = Mock() + chan.send = AsyncMock( + side_effect=TransportClosed( + message='caller closed before StartAck', + ), + ) + chan.connected.return_value = False + ctx = Mock( + chan=chan, + cid='rpc-cid', + _scope=None, + _task='rpc-task', + ) + actor = Mock() + actor.get_context.return_value = ctx + actor._rpc_tasks = {} + actor._ongoing_rpc_tasks = trio.Event() + actor._ongoing_rpc_tasks.set() + + async def rpc_func(): + assert (chan, ctx.cid) in actor._rpc_tasks + rpc_ran.set() + + async def invoke(task_status): + await _rpc._invoke( + actor=actor, + cid=ctx.cid, + chan=chan, + func=rpc_func, + kwargs={}, + task_status=task_status, + ) + + async with trio.open_nursery() as nursery: + started_ctx = await nursery.start(invoke) + assert started_ctx is ctx + await rpc_ran.wait() + + assert rpc_ran.is_set() + assert not actor._rpc_tasks + assert actor._ongoing_rpc_tasks.is_set() + chan.send.assert_awaited_once() + + trio.run(main) + + +def test_error_shipment_ignores_closed_response_channel(monkeypatch): + ''' + Preserve an application error when its response channel is closed. + + A caller can disconnect after submitting an RPC but before its + error response. Normalized `TransportClosed` from that final send + is terminal response failure, not a new actor-wide service error. + This test proves error shipment logs and returns without replacing + the original application exception. + + ''' + chan = Mock() + chan.send = AsyncMock( + side_effect=[ + None, + TransportClosed( + message='caller closed before Error response', + ), + ], + ) + error_msg = Mock(boxed_type_str='ValueError') + monkeypatch.setattr( + _rpc, + 'pack_error', + Mock(return_value=error_msg), + ) + ctx = Mock( + chan=chan, + cid='rpc-cid', + _scope=None, + _task='rpc-task', + ) + actor = Mock() + actor.get_context.return_value = ctx + actor._rpc_tasks = {} + actor._ongoing_rpc_tasks = trio.Event() + actor._ongoing_rpc_tasks.set() + + async def failing_rpc(): + raise ValueError('application failure') + + async def main(): + async with trio.open_nursery() as nursery: + started_ctx = await nursery.start( + _rpc._invoke, + actor, + ctx.cid, + chan, + failing_rpc, + {}, + ) + assert started_ctx is ctx + + trio.run(main) + assert chan.send.await_count == 2 + assert not actor._rpc_tasks + assert actor._ongoing_rpc_tasks.is_set() + + @pytest.mark.parametrize( 'to_call', [ ([], 'short_sleep', tractor.RemoteActorError), diff --git a/tractor/runtime/_rpc.py b/tractor/runtime/_rpc.py index d2c9ea30..d90cb658 100644 --- a/tractor/runtime/_rpc.py +++ b/tractor/runtime/_rpc.py @@ -94,6 +94,33 @@ if TYPE_CHECKING: log = get_logger('tractor') +def _register_rpc_task( + actor: Actor, + chan: Channel, + func: Callable, + is_rpc: bool, + task_status: TaskStatus[ + Context | BaseException + ], + ctx: Context, +) -> None: + ''' + Register an RPC task before publishing it to `Nursery.start()`. + + ''' + if is_rpc: + if not actor._rpc_tasks: + actor._ongoing_rpc_tasks = trio.Event() + + actor._rpc_tasks[(chan, ctx.cid)] = ( + ctx, + func, + trio.Event(), + ) + + task_status.started(ctx) + + # ?TODO? move to a `tractor.lowlevel._rpc` with the below # func-type-cases implemented "on top of" `@context` defs: # -[ ] std async func helper decorated with `@rpc_func`? @@ -142,7 +169,14 @@ async def _invoke_non_context( # is propagated! with cancel_scope as cs: ctx._scope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) async with aclosing(coro) as agen: async for item in agen: # TODO: can we send values back in here? @@ -178,7 +212,14 @@ async def _invoke_non_context( ) with cancel_scope as cs: ctx._scope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) await coro if not cs.cancelled_caught: @@ -202,23 +243,29 @@ async def _invoke_non_context( ) await chan.send(ack) except ( + TransportClosed, trio.ClosedResourceError, trio.BrokenResourceError, BrokenPipeError, ) as ipc_err: failed_resp = True - if is_rpc: - raise ipc_err - else: - log.exception( - f'Failed to ack runtime RPC request\n\n' - f'{func} x=> {ctx.chan}\n\n' - f'{ack}\n' - ) + log.warning( + f'Failed to ack runtime RPC request\n\n' + f'{func} x=> {ctx.chan}\n\n' + f'{ack}\n' + f' |_{ipc_err!r}\n' + ) with cancel_scope as cs: ctx._scope: CancelScope = cs - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) result = await coro fname: str = func.__name__ @@ -247,6 +294,7 @@ async def _invoke_non_context( ) await chan.send(ret_msg) except ( + TransportClosed, BrokenPipeError, trio.BrokenResourceError, ): @@ -673,7 +721,14 @@ async def _invoke( ): ctx._scope_nursery = tn rpc_ctx_cs = ctx._scope = tn.cancel_scope - task_status.started(ctx) + _register_rpc_task( + actor, + chan, + func, + is_rpc, + task_status, + ctx, + ) # invoke user endpoint fn. res: Any|PayloadT = await coro @@ -920,6 +975,7 @@ async def try_ship_error_to_remote( # downward should be mostly wrapping such cases in a # tpt-closed; the `.critical()` usage is warranted. except ( + TransportClosed, trio.ClosedResourceError, trio.BrokenResourceError, BrokenPipeError, @@ -1221,18 +1277,6 @@ async def process_messages( ) continue - else: - # mark our global state with ongoing rpc tasks - actor._ongoing_rpc_tasks = trio.Event() - - # store cancel scope such that the rpc task can be - # cancelled gracefully if requested - actor._rpc_tasks[(chan, cid)] = ( - ctx, - func, - trio.Event(), - ) - # XXX RUNTIME-SCOPED! remote (likely internal) error # (^- bc no `Error.cid` -^) # From d0cc06815f5f7db4b7a7cafeaaf9022e8ea05dcf Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 12:49:40 -0400 Subject: [PATCH 19/22] Accept raced `TransportClosed` diagnostics The `@context` debugger E2E intentionally closes its channel. Teardown can surface from either local error shipment or the peer receive task. The RPC response fix makes local close win under CI, while the test required both scheduler-dependent diagnostics. Keep the common debugger and cancellation assertions, then accept either transport-close report. This preserves real actor-tree teardown coverage without depending on task scheduling order. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/devx/test_debugger.py | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/tests/devx/test_debugger.py b/tests/devx/test_debugger.py index dfcf36d8..5830cf94 100644 --- a/tests/devx/test_debugger.py +++ b/tests/devx/test_debugger.py @@ -1307,18 +1307,30 @@ def test_ctxep_pauses_n_maybe_ipc_breaks( if _non_linux: tpt: str = 'TCP' - assert_before( + before: str = assert_before( child, ['peer IPC channel closed abruptly?', 'another task closed this fd', 'Debug lock request was CANCELLED?', - f"'Msgpack{tpt}Stream' was already closed locally?", - f"TransportClosed: 'Msgpack{tpt}Stream' was already closed 'by peer'?", - ] + ] - # XXX races on whether these show/hit? - # 'Failed to REPl via `_pause()` You called `tractor.pause()` from an already cancelled scope!', - # 'AssertionError', + # XXX races on whether these show/hit? + # 'Failed to REPl via `_pause()` You called `tractor.pause()` from an already cancelled scope!', + # 'AssertionError', + ) + + # Error shipment and peer receive race after local close. + # Either diagnostic proves the transport was torn down. + closed_locally: str = ( + f"'Msgpack{tpt}Stream' was already closed locally?" + ) + closed_by_peer: str = ( + f"TransportClosed: 'Msgpack{tpt}Stream' was " + f"already closed 'by peer'?" + ) + assert ( + closed_locally in before + or closed_by_peer in before ) # OSc(ancel) the hanging tree do_ctlc( From f454cefe56f96b502f8b0c754b6dfe977851b2c3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 18:40:23 -0400 Subject: [PATCH 20/22] Enforce `get_rt_dir()` ownership on POSIX Linux previously accepted a pre-existing runtime bindspace without checking its owner or mode, even though Darwin enforced both. Share the POSIX directory guard so every managed root and subdir rejects non-directories and foreign UIDs before changing permissions. Normalize owner-controlled bindspaces to `0o700` and add Linux regressions for mode repair, foreign ownership, and non-directory paths. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_each_tpt.py | 67 ++++++++++++++++++++- tractor/runtime/_state.py | 115 +++++++++++++++++++++---------------- 2 files changed, 130 insertions(+), 52 deletions(-) diff --git a/tests/ipc/test_each_tpt.py b/tests/ipc/test_each_tpt.py index b62e2341..512300a6 100644 --- a/tests/ipc/test_each_tpt.py +++ b/tests/ipc/test_each_tpt.py @@ -194,7 +194,10 @@ def test_rt_dir_rejects_non_directory( lambda appname: str(rt_file), ) - with pytest.raises(FileExistsError): + with pytest.raises( + PermissionError, + match='Unsafe POSIX', + ): _state.get_rt_dir() new_rt_dir: Path = tmp_path / 'new-runtime-dir' @@ -206,6 +209,68 @@ def test_rt_dir_rejects_non_directory( assert stat.S_IMODE(new_rt_dir.stat().st_mode) == 0o700 +def test_linux_rt_dir_secures_existing_path( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Enforce owner-only access on an existing Linux runtime directory. + + Linux previously accepted any existing directory returned by + `platformdirs`, without checking ownership or correcting a + traversable mode. This test creates an owner-controlled `0o755` + directory and proves `get_rt_dir()` normalizes the managed + bindspace to `0o700` before returning it. + + ''' + rt_dir: Path = tmp_path / 'tractor' + rt_dir.mkdir(mode=0o755) + monkeypatch.setattr(sys, 'platform', 'linux') + monkeypatch.setattr( + 'platformdirs.user_runtime_dir', + lambda appname: str(rt_dir), + ) + + assert _state.get_rt_dir() == rt_dir + assert stat.S_IMODE(rt_dir.stat().st_mode) == 0o700 + + +def test_linux_rt_dir_rejects_foreign_owner( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +): + ''' + Reject an existing Linux runtime directory owned by another UID. + + A pre-created bindspace must never be made private with `chmod` + until ownership is verified. This test makes the current process + appear to have a different UID and proves `get_rt_dir()` rejects + the directory without changing its original mode. + + ''' + rt_dir: Path = tmp_path / 'tractor' + rt_dir.mkdir(mode=0o755) + original_mode: int = stat.S_IMODE(rt_dir.stat().st_mode) + monkeypatch.setattr(sys, 'platform', 'linux') + monkeypatch.setattr( + 'platformdirs.user_runtime_dir', + lambda appname: str(rt_dir), + ) + monkeypatch.setattr( + os, + 'getuid', + lambda: rt_dir.stat().st_uid + 1, + ) + + with pytest.raises( + PermissionError, + match='Unsafe POSIX', + ): + _state.get_rt_dir() + + assert stat.S_IMODE(rt_dir.stat().st_mode) == original_mode + + def test_macos_rt_dir_rejects_intermediate_symlink( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, diff --git a/tractor/runtime/_state.py b/tractor/runtime/_state.py index 38b64260..5caa92e3 100644 --- a/tractor/runtime/_state.py +++ b/tractor/runtime/_state.py @@ -323,6 +323,60 @@ def current_ipc_ctx( +def _ensure_owner_only_posix_dir( + path: Path, + *, + parents: bool = False, +) -> None: + ''' + Create or validate a UID-owned POSIX runtime directory. + + Pre-existing directories are accepted only when owned by the + current user. Their mode is normalized to `0o700` because runtime + directories hold IPC sockets and are private bindspaces. + + ''' + # TODO: https://github.com/goodboy/tractor/issues/494 + # Research having the actor-tree root process choose and create + # this bindspace, then propagate it to every subactor. On + # Linux, a private mount namespace could isolate it while letting + # spawned subactors inherit access; independently launched + # discovery clients would need an explicit join or fallback path. + # POSIX metadata alone records UID/GID ownership, so other systems + # still need explicit runtime metadata and lifecycle management. + try: + dir_stat: os.stat_result = path.lstat() + except FileNotFoundError: + try: + path.mkdir( + mode=0o700, + parents=parents, + ) + except FileExistsError: + pass + dir_stat = path.lstat() + + if ( + not stat.S_ISDIR(dir_stat.st_mode) + or + dir_stat.st_uid != os.getuid() + ): + platform_name: str = ( + 'Darwin' + if sys.platform == 'darwin' + else 'POSIX' + ) + raise PermissionError( + f'Unsafe {platform_name} runtime directory!\n' + f'path: {path}\n' + f'owner uid: {dir_stat.st_uid}\n' + f'mode: {stat.filemode(dir_stat.st_mode)}\n' + ) + + if stat.S_IMODE(dir_stat.st_mode) != 0o700: + path.chmod(0o700) + + def get_rt_dir( subdir: str|Path|None = None, appname: str = 'tractor', @@ -332,9 +386,9 @@ def get_rt_dir( userspace apps stick their IPC and cache related system util-files. - Linux uses `${XDG_RUNTIME_DIR}/tractor/`; Darwin uses a short, - owner-only `/tmp/tractor-` path; other platforms use the - lovely `platformdirs` lib. + Linux uses an owner-only `${XDG_RUNTIME_DIR}/tractor/`; Darwin + uses a short, owner-only `/tmp/tractor-` path; other + platforms use the lovely `platformdirs` lib. ''' # lazy-imported to keep it off the eager @@ -349,29 +403,6 @@ def get_rt_dir( _DARWIN_TMPDIR / f'{appname}-{os.getuid()}' ) - try: - rt_stat: os.stat_result = rt_root.lstat() - except FileNotFoundError: - try: - rt_root.mkdir(mode=0o700) - except FileExistsError: - pass - rt_stat = rt_root.lstat() - - if ( - not stat.S_ISDIR(rt_stat.st_mode) - or - rt_stat.st_uid != os.getuid() - ): - raise PermissionError( - f'Unsafe Darwin runtime directory!\n' - f'path: {rt_root}\n' - f'owner uid: {rt_stat.st_uid}\n' - f'mode: {stat.filemode(rt_stat.st_mode)}\n' - ) - if stat.S_IMODE(rt_stat.st_mode) != 0o700: - rt_root.chmod(0o700) - rt_dir: Path = rt_root else: rt_dir = Path( @@ -401,7 +432,7 @@ def get_rt_dir( f'{subdir!r}\n' ) - if rt_root is None: + if os.name != 'posix': if subdir_path is not None: rt_dir = rt_dir / subdir_path if not rt_dir.is_dir(): @@ -414,33 +445,15 @@ def get_rt_dir( ) return rt_dir + _ensure_owner_only_posix_dir( + rt_dir, + parents=(rt_root is None), + ) + if subdir_path is not None: for part in subdir_path.parts: rt_dir = rt_dir / part - try: - dir_stat: os.stat_result = rt_dir.lstat() - except FileNotFoundError: - try: - # Every Darwin component is private so no other - # user can replace descendants below `rt_root`. - rt_dir.mkdir(mode=0o700) - except FileExistsError: - pass - dir_stat = rt_dir.lstat() - - if ( - not stat.S_ISDIR(dir_stat.st_mode) - or - dir_stat.st_uid != os.getuid() - ): - raise PermissionError( - f'Unsafe Darwin runtime directory!\n' - f'path: {rt_dir}\n' - f'owner uid: {dir_stat.st_uid}\n' - f'mode: {stat.filemode(dir_stat.st_mode)}\n' - ) - if stat.S_IMODE(dir_stat.st_mode) != 0o700: - rt_dir.chmod(0o700) + _ensure_owner_only_posix_dir(rt_dir) return rt_dir From 75cda1933c98e6eefca550393c483bc5fcb9ec13 Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 18:43:36 -0400 Subject: [PATCH 21/22] Move registrar probes to `discovery._api` Registrar election and multi-address probing are discovery-protocol concerns, but their implementation lived in root-runtime ignition. Move the bounded handshake probe and its concurrent address classifier into `discovery._api`, leaving `open_root_actor()` to consume the classified results. Update probe tests for the canonical module and clarify that the daemon-fixture regressions directly exercise their sibling `conftest` plugin. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/discovery/test_daemon_fixture.py | 4 + tests/discovery/test_tpt_bind_addrs.py | 16 ++-- tractor/_root.py | 115 ++----------------------- tractor/discovery/_api.py | 115 +++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 115 deletions(-) diff --git a/tests/discovery/test_daemon_fixture.py b/tests/discovery/test_daemon_fixture.py index d8899e42..84a80ca3 100644 --- a/tests/discovery/test_daemon_fixture.py +++ b/tests/discovery/test_daemon_fixture.py @@ -1,6 +1,10 @@ ''' Discovery daemon fixture regressions. +This module imports private helpers from the sibling +`tests.discovery.conftest` plugin to exercise that fixture machinery +directly, rather than testing a production `tractor` API. + ''' from unittest.mock import ( call, diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index ca65587e..538b4e8b 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -2,7 +2,8 @@ `open_root_actor(tpt_bind_addrs=...)` test suite. Verify all three runtime code paths for explicit IPC-server -bind-address selection in `_root.py`: +bind-address selection in `_root.py` and registry probing in +`discovery._api`: 1. Non-registrar, no explicit bind -> random addrs from registry proto 2. Registrar, no explicit bind -> binds to registry_addrs @@ -19,11 +20,12 @@ from unittest.mock import ( import pytest import trio import tractor -from tractor import _root +from tractor.discovery import _api from tractor.discovery._addr import ( wrap_address, ) from tractor.discovery._multiaddr import mk_maddr +from tractor.ipc import _connect_chan from tractor._testing.addr import get_rando_addr @@ -68,11 +70,11 @@ def test_registry_probe_retries_transient_handshake( closed.append(chan) sleep = AsyncMock() - monkeypatch.setattr(_root, '_connect_chan', connect_chan) - monkeypatch.setattr(_root.trio, 'sleep', sleep) + monkeypatch.setattr(_api, '_connect_chan', connect_chan) + monkeypatch.setattr(_api.trio, 'sleep', sleep) async def main(): - status = await _root._probe_registry( + status = await _api._probe_registry( addr=wrap_address(('127.0.0.1', 1616)), timeout=.3, attempt_timeout=.1, @@ -114,7 +116,7 @@ def test_probe_channel_close_is_bounded( async def main(): with trio.fail_after(.5): - async with _root._connect_chan( + async with _connect_chan( ('127.0.0.1', 1616), close_timeout=.01, ): @@ -193,7 +195,7 @@ def test_registry_probe_preserves_no_peers_state( actor = tractor.current_actor() server = actor.ipc_server - probe_status = await _root._probe_registry( + probe_status = await _api._probe_registry( addr=wrap_address(reg_addr), ) assert probe_status == 'registrar' diff --git a/tractor/_root.py b/tractor/_root.py index 28696918..a598526b 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -31,7 +31,6 @@ import sys from typing import ( Any, Callable, - Literal, ) import warnings @@ -48,9 +47,7 @@ from .devx import ( from .spawn import _spawn from .runtime import _state from . import log -from .ipc import ( - _connect_chan, -) +from .discovery._api import _probe_registry_addrs from .discovery._addr import ( Address, UnwrappedAddress, @@ -64,9 +61,7 @@ from .trionics import ( ) from ._exceptions import ( RuntimeFailure, - TransportClosed, ) -from .msg.types import Aid logger = log.get_logger('tractor') @@ -86,68 +81,6 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = ( ) -async def _probe_registry( - addr: Address, - timeout: float = 3, - attempt_timeout: float = 1, - max_attempts: int = 3, - retry_delay: float = .05, - close_timeout: float = .2, -) -> Literal[ - 'absent', - 'occupied', - 'registrar', -]: - ''' - Confirm an address serves the Tractor actor handshake. - - Connection and handshake work share `timeout`; each attempt gets - `attempt_timeout`. Shielded cleanup may add up to `close_timeout` - per attempted channel. - - ''' - connected_once: bool = False - with trio.move_on_after(timeout): - for attempt in range(max_attempts): - try: - with trio.move_on_after(attempt_timeout) as attempt_cs: - async with _connect_chan( - addr.unwrap(), - close_timeout=close_timeout, - ) as chan: - connected_once = True - peer_aid: Aid = await chan._do_handshake( - aid=Aid( - name='registry-probe', - uuid=mk_uuid(), - pid=os.getpid(), - is_probe=True, - ), - timeout=attempt_timeout, - ) - if peer_aid.is_registrar is not False: - return 'registrar' - return 'occupied' - - if attempt_cs.cancelled_caught: - if not connected_once: - return 'absent' - - except OSError: - return ( - 'occupied' - if connected_once - else 'absent' - ) - except TransportClosed: - pass - - if attempt + 1 < max_attempts: - await trio.sleep(retry_delay * (attempt + 1)) - - return 'occupied' - - # TODO: stick this in a `@acm` defined in `devx.debug`? # -[ ] also maybe consider making this a `wrapt`-deco to # save an indent level? @@ -516,46 +449,12 @@ async def open_root_actor( from .devx._stackscope import enable_stack_on_sig enable_stack_on_sig() - # closed into below ping task-func - ponged_addrs: list[Address] = [] - occupied_addrs: list[Address] = [] - - async def ping_tpt_socket( - addr: Address, - timeout: float = 3, - ) -> None: - ''' - Probe with a bounded Tractor actor handshake. - - Classify the address as a registrar, occupied by a - non-registrar, or absent. - - ''' - probe_status = await _probe_registry( - addr=addr, - timeout=timeout, - ) - if probe_status == 'registrar': - ponged_addrs.append(addr) - elif probe_status == 'occupied': - occupied_addrs.append(addr) - else: - # ?TODO, make this a "discovery" log level? - logger.info( - f'No root-actor registry found @ {addr!r}\n' - ) - - # !TODO, this is basically just another (abstract) - # happy-eyeballs, so we should try for formalize it somewhere - # in a `.[_]discovery` ya? - # - async with trio.open_nursery() as tn: - for uw_addr in uw_reg_addrs: - addr: Address = wrap_address(uw_addr) - tn.start_soon( - ping_tpt_socket, - addr, - ) + ponged_addrs: list[Address] + occupied_addrs: list[Address] + ( + ponged_addrs, + occupied_addrs, + ) = await _probe_registry_addrs(uw_reg_addrs) if ( not ponged_addrs diff --git a/tractor/discovery/_api.py b/tractor/discovery/_api.py index ec559baa..1691e2c8 100644 --- a/tractor/discovery/_api.py +++ b/tractor/discovery/_api.py @@ -21,14 +21,18 @@ management of (service) actors. """ from __future__ import annotations import ipaddress +import os import socket from typing import ( AsyncGenerator, AsyncContextManager, + Literal, TYPE_CHECKING, ) from contextlib import asynccontextmanager as acm +import trio + from tractor.log import get_logger from ..trionics import ( gather_contexts, @@ -40,6 +44,7 @@ from ..ipc._uds import UDSAddress from ._addr import ( UnwrappedAddress, Address, + mk_uuid, wrap_address, ) from ..runtime._portal import ( @@ -52,6 +57,7 @@ from ..runtime._state import ( _runtime_vars, _def_tpt_proto, ) +from ..msg.types import Aid if TYPE_CHECKING: from ..runtime._runtime import Actor @@ -60,6 +66,115 @@ if TYPE_CHECKING: log = get_logger() +async def _probe_registry( + addr: Address, + timeout: float = 3, + attempt_timeout: float = 1, + max_attempts: int = 3, + retry_delay: float = .05, + close_timeout: float = .2, +) -> Literal[ + 'absent', + 'occupied', + 'registrar', +]: + ''' + Confirm an address serves the Tractor actor handshake. + + Connection and handshake work share `timeout`; each attempt gets + `attempt_timeout`. Shielded cleanup may add up to `close_timeout` + per attempted channel. + + ''' + from .._exceptions import TransportClosed + + connected_once: bool = False + with trio.move_on_after(timeout): + for attempt in range(max_attempts): + try: + with trio.move_on_after(attempt_timeout) as attempt_cs: + async with _connect_chan( + addr.unwrap(), + close_timeout=close_timeout, + ) as chan: + connected_once = True + peer_aid: Aid = await chan._do_handshake( + aid=Aid( + name='registry-probe', + uuid=mk_uuid(), + pid=os.getpid(), + is_probe=True, + ), + timeout=attempt_timeout, + ) + if peer_aid.is_registrar is not False: + return 'registrar' + return 'occupied' + + if attempt_cs.cancelled_caught: + if not connected_once: + return 'absent' + + except OSError: + return ( + 'occupied' + if connected_once + else 'absent' + ) + except TransportClosed: + pass + + if attempt + 1 < max_attempts: + await trio.sleep(retry_delay * (attempt + 1)) + + return 'occupied' + + +async def _probe_registry_addrs( + addrs: list[UnwrappedAddress], + timeout: float = 3, +) -> tuple[ + list[Address], + list[Address], +]: + ''' + Concurrently classify candidate registrar addresses. + + Return confirmed registrar addresses followed by addresses occupied + by non-registrar or unresponsive Tractor peers. + + ''' + registrar_addrs: list[Address] = [] + occupied_addrs: list[Address] = [] + + async def probe_addr(addr: Address) -> None: + probe_status = await _probe_registry( + addr=addr, + timeout=timeout, + ) + if probe_status == 'registrar': + registrar_addrs.append(addr) + elif probe_status == 'occupied': + occupied_addrs.append(addr) + else: + # ?TODO, make this a "discovery" log level? + log.info( + f'No root-actor registry found @ {addr!r}\n' + ) + + async with trio.open_nursery() as nursery: + for unwrapped_addr in addrs: + nursery.start_soon( + probe_addr, + wrap_address(unwrapped_addr), + ) + + return ( + registrar_addrs, + occupied_addrs, + ) + + def _is_local_addr(addr: Address) -> bool: ''' Determine whether `addr` is reachable on the From f75c9cdeab681db5b576cb972bb14be9150dab5b Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 18:54:56 -0400 Subject: [PATCH 22/22] Traverse `BaseExceptionGroup` peer-close errors `_peer_closed_errno()` followed cause and context links but did not descend through grouped exceptions. A reset below a group could therefore escape as `trio.BrokenResourceError` instead of the normalized `TransportClosed` boundary. Walk the exception tree with cycle protection, requiring every group branch to represent peer closure before normalization. Extend the `MsgpackTransport.send()` regression to prove all-transport and mixed-failure behavior. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_server.py | 68 ++++++++++++++++++++++++++++----------- tractor/ipc/_transport.py | 61 +++++++++++++++++++++++++++-------- 2 files changed, 98 insertions(+), 31 deletions(-) diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py index 6186454e..7689aeff 100644 --- a/tests/ipc/test_server.py +++ b/tests/ipc/test_server.py @@ -34,31 +34,49 @@ from tractor.msg.types import Aid # from ._tcp import TCPAddress -def test_send_normalizes_peer_reset(): +def test_send_normalizes_only_grouped_peer_resets(): ''' - Normalize Darwin's pre-handshake peer reset as transport closure. + Normalize only all-peer-close grouped transport failures. A UDS peer may disconnect before completing the actor handshake. Darwin can report the server's first handshake write as - `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing - that raw error to escape cancels the daemon's shared IPC nursery. - This fake stream reproduces the exact exception chain and proves - `.send()` raises the expected `TransportClosed` boundary instead. + `ECONNRESET`, wrapped by `trio.BrokenResourceError` and potentially + nested in an `ExceptionGroup`. This fake stream first groups reset + and broken-pipe branches, proving `.send()` normalizes a complete + peer-close tree to `TransportClosed`. It then groups a reset with + an unrelated `ValueError`, proving the mixed failure remains a + `trio.BrokenResourceError` instead of hiding the application error. ''' - class ResetStream: - async def send_all(self, data: bytes) -> None: + def broken_resource(err_no: int) -> trio.BrokenResourceError: + try: + raise OSError( + err_no, + 'Peer closed', + ) + except OSError as peer_err: try: - raise OSError( - errno.ECONNRESET, - 'Connection reset by peer', - ) - except OSError as reset_err: - raise trio.BrokenResourceError from reset_err + raise trio.BrokenResourceError from peer_err + except trio.BrokenResourceError as broken_err: + return broken_err + + class GroupedFailureStream: + def __init__(self, exceptions: list[Exception]) -> None: + self.exceptions = exceptions + + async def send_all(self, data: bytes) -> None: + grouped_err = ExceptionGroup( + 'concurrent send failures', + self.exceptions, + ) + raise trio.BrokenResourceError from grouped_err async def main(): transport = object.__new__(MsgpackTransport) - transport.stream = ResetStream() + transport.stream = GroupedFailureStream([ + broken_resource(errno.ECONNRESET), + broken_resource(errno.EPIPE), + ]) transport._send_lock = trio.StrictFIFOLock() transport._laddr = 'local' transport._raddr = 'remote' @@ -70,9 +88,23 @@ def test_send_normalizes_peer_reset(): strict_types=False, ) - assert exc_info.value.src_exc.__cause__.errno == ( - errno.ECONNRESET - ) + grouped_err = exc_info.value.src_exc.__cause__ + assert isinstance(grouped_err, ExceptionGroup) + assert len(grouped_err.exceptions) == 2 + + transport.stream = GroupedFailureStream([ + ValueError('unrelated failure'), + broken_resource(errno.ECONNRESET), + ]) + with pytest.raises(trio.BrokenResourceError) as exc_info: + await transport.send( + {'probe': True}, + strict_types=False, + ) + + grouped_err = exc_info.value.__cause__ + assert isinstance(grouped_err, ExceptionGroup) + assert isinstance(grouped_err.exceptions[0], ValueError) trio.run(main) diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index 48692f79..dfa36696 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -64,29 +64,64 @@ log = get_logger() def _peer_closed_errno(exc: BaseException) -> int|None: ''' - Find a peer-close errno in a transport exception chain. + Classify a complete transport exception tree as peer closure. + + Follow explicit cause/context links. For a `BaseExceptionGroup`, + require every child branch to resolve to a peer-close errno so an + unrelated concurrent failure is never hidden as `TransportClosed`. ''' - seen: set[int] = set() - while ( - exc - and - id(exc) not in seen - ): - seen.add(id(exc)) + def find_peer_errno( + current_exc: BaseException, + ancestors: set[int], + ) -> int|None: + exc_id: int = id(current_exc) + if exc_id in ancestors: + return None + + ancestors = ancestors | {exc_id} if ( - isinstance(exc, OSError) + isinstance(current_exc, OSError) and - exc.errno in { + current_exc.errno in { errno.ECONNRESET, errno.EPIPE, } ): - return exc.errno + return current_exc.errno - exc = exc.__cause__ or exc.__context__ + if isinstance(current_exc, BaseExceptionGroup): + child_errnos: list[int|None] = [ + find_peer_errno( + child_exc, + ancestors, + ) + for child_exc in current_exc.exceptions + ] + if all( + child_errno is not None + for child_errno in child_errnos + ): + return child_errnos[0] + return None - return None + chained_exc: BaseException|None = ( + current_exc.__cause__ + or + current_exc.__context__ + ) + if chained_exc is not None: + return find_peer_errno( + chained_exc, + ancestors, + ) + + return None + + return find_peer_errno( + exc, + set(), + ) # (codec, transport)