tests/discovery/test_daemon_fixture.py

@@ -1,10 +1,6 @@
 '''
 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,

tests/discovery/test_tpt_bind_addrs.py

@@ -2,8 +2,7 @@
 `open_root_actor(tpt_bind_addrs=...)` test suite.
 
 Verify all three runtime code paths for explicit IPC-server
-bind-address selection in `_root.py` and registry probing in
+bind-address selection in `_root.py`:
-`discovery._api`:
 
 1. Non-registrar, no explicit bind -> random addrs from registry proto
 2. Registrar, no explicit bind -> binds to registry_addrs
@@ -20,12 +19,11 @@ from unittest.mock import (
 import pytest
 import trio
 import tractor
-from tractor.discovery import _api
+from tractor import _root
 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
 
 
@@ -70,11 +68,11 @@ def test_registry_probe_retries_transient_handshake(
             closed.append(chan)
 
     sleep = AsyncMock()
-    monkeypatch.setattr(_api, '_connect_chan', connect_chan)
+    monkeypatch.setattr(_root, '_connect_chan', connect_chan)
-    monkeypatch.setattr(_api.trio, 'sleep', sleep)
+    monkeypatch.setattr(_root.trio, 'sleep', sleep)
 
     async def main():
-        status = await _api._probe_registry(
+        status = await _root._probe_registry(
             addr=wrap_address(('127.0.0.1', 1616)),
             timeout=.3,
             attempt_timeout=.1,
@@ -116,7 +114,7 @@ def test_probe_channel_close_is_bounded(
 
     async def main():
         with trio.fail_after(.5):
-            async with _connect_chan(
+            async with _root._connect_chan(
                 ('127.0.0.1', 1616),
                 close_timeout=.01,
             ):
@@ -195,7 +193,7 @@ def test_registry_probe_preserves_no_peers_state(
             actor = tractor.current_actor()
             server = actor.ipc_server
 
-            probe_status = await _api._probe_registry(
+            probe_status = await _root._probe_registry(
                 addr=wrap_address(reg_addr),
             )
             assert probe_status == 'registrar'

tests/ipc/test_each_tpt.py

@@ -194,10 +194,7 @@ def test_rt_dir_rejects_non_directory(
         lambda appname: str(rt_file),
     )
 
-    with pytest.raises(
+    with pytest.raises(FileExistsError):
-        PermissionError,
-        match='Unsafe POSIX',
-    ):
         _state.get_rt_dir()
 
     new_rt_dir: Path = tmp_path / 'new-runtime-dir'
@@ -209,68 +206,6 @@ 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,

tests/ipc/test_server.py

@@ -34,49 +34,31 @@ from tractor.msg.types import Aid
 # from ._tcp import TCPAddress
 
 
-def test_send_normalizes_only_grouped_peer_resets():
+def test_send_normalizes_peer_reset():
     '''
-    Normalize only all-peer-close grouped transport failures.
+    Normalize Darwin's pre-handshake peer reset as transport closure.
 
     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` and potentially
+    `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing
-    nested in an `ExceptionGroup`. This fake stream first groups reset
+    that raw error to escape cancels the daemon's shared IPC nursery.
-    and broken-pipe branches, proving `.send()` normalizes a complete
+    This fake stream reproduces the exact exception chain and proves
-    peer-close tree to `TransportClosed`. It then groups a reset with
+    `.send()` raises the expected `TransportClosed` boundary instead.
-    an unrelated `ValueError`, proving the mixed failure remains a
-    `trio.BrokenResourceError` instead of hiding the application error.
 
     '''
-    def broken_resource(err_no: int) -> trio.BrokenResourceError:
+    class ResetStream:
-        try:
-            raise OSError(
-                err_no,
-                'Peer closed',
-            )
-        except OSError as peer_err:
-            try:
-                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(
+            try:
-                'concurrent send failures',
+                raise OSError(
-                self.exceptions,
+                    errno.ECONNRESET,
-            )
+                    'Connection reset by peer',
-            raise trio.BrokenResourceError from grouped_err
+                )
+            except OSError as reset_err:
+                raise trio.BrokenResourceError from reset_err
 
     async def main():
         transport = object.__new__(MsgpackTransport)
-        transport.stream = GroupedFailureStream([
+        transport.stream = ResetStream()
-            broken_resource(errno.ECONNRESET),
-            broken_resource(errno.EPIPE),
-        ])
         transport._send_lock = trio.StrictFIFOLock()
         transport._laddr = 'local'
         transport._raddr = 'remote'
@@ -88,23 +70,9 @@ def test_send_normalizes_only_grouped_peer_resets():
                 strict_types=False,
             )
 
-        grouped_err = exc_info.value.src_exc.__cause__
+        assert exc_info.value.src_exc.__cause__.errno == (
-        assert isinstance(grouped_err, ExceptionGroup)
+            errno.ECONNRESET
-        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)
 

tractor/_root.py

@@ -31,6 +31,7 @@ import sys
 from typing import (
     Any,
     Callable,
+    Literal,
 )
 import warnings
 
@@ -47,7 +48,9 @@ from .devx import (
 from .spawn import _spawn
 from .runtime import _state
 from . import log
-from .discovery._api import _probe_registry_addrs
+from .ipc import (
+    _connect_chan,
+)
 from .discovery._addr import (
     Address,
     UnwrappedAddress,
@@ -61,7 +64,9 @@ from .trionics import (
 )
 from ._exceptions import (
     RuntimeFailure,
+    TransportClosed,
 )
+from .msg.types import Aid
 
 
 logger = log.get_logger('tractor')
@@ -81,6 +86,68 @@ _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?
@@ -449,12 +516,46 @@ async def open_root_actor(
             from .devx._stackscope import enable_stack_on_sig
             enable_stack_on_sig()
 
-        ponged_addrs: list[Address]
+        # closed into below ping task-func
-        occupied_addrs: list[Address]
+        ponged_addrs: list[Address] = []
-        (
+        occupied_addrs: list[Address] = []
-            ponged_addrs,
+
-            occupied_addrs,
+        async def ping_tpt_socket(
-        ) = await _probe_registry_addrs(uw_reg_addrs)
+            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,
+                )
 
         if (
             not ponged_addrs

tractor/discovery/_api.py

@@ -21,18 +21,14 @@ 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,
@@ -44,7 +40,6 @@ from ..ipc._uds import UDSAddress
 from ._addr import (
     UnwrappedAddress,
     Address,
-    mk_uuid,
     wrap_address,
 )
 from ..runtime._portal import (
@@ -57,7 +52,6 @@ from ..runtime._state import (
     _runtime_vars,
     _def_tpt_proto,
 )
-from ..msg.types import Aid
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
@@ -66,115 +60,6 @@ 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

tractor/ipc/_transport.py

@@ -64,64 +64,29 @@ log = get_logger()
 
 def _peer_closed_errno(exc: BaseException) -> int|None:
     '''
-    Classify a complete transport exception tree as peer closure.
+    Find a peer-close errno in a transport exception chain.
-
-    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`.
 
     '''
-    def find_peer_errno(
+    seen: set[int] = set()
-        current_exc: BaseException,
+    while (
-        ancestors: set[int],
+        exc
-    ) -> int|None:
+        and
-        exc_id: int = id(current_exc)
+        id(exc) not in seen
-        if exc_id in ancestors:
+    ):
-            return None
+        seen.add(id(exc))
-
-        ancestors = ancestors | {exc_id}
         if (
-            isinstance(current_exc, OSError)
+            isinstance(exc, OSError)
             and
-            current_exc.errno in {
+            exc.errno in {
                 errno.ECONNRESET,
                 errno.EPIPE,
             }
         ):
-            return current_exc.errno
+            return exc.errno
 
-        if isinstance(current_exc, BaseExceptionGroup):
+        exc = exc.__cause__ or exc.__context__
-            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
 
-        chained_exc: BaseException|None = (
+    return 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)

tractor/runtime/_state.py

@@ -323,60 +323,6 @@ 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',
@@ -386,9 +332,9 @@ def get_rt_dir(
     userspace apps stick their IPC and cache related system
     util-files.
 
-    Linux uses an owner-only `${XDG_RUNTIME_DIR}/tractor/`; Darwin
+    Linux uses `${XDG_RUNTIME_DIR}/tractor/`; Darwin uses a short,
-    uses a short, owner-only `/tmp/tractor-<uid>` path; other
+    owner-only `/tmp/tractor-<uid>` path; other platforms use the
-    platforms use the lovely `platformdirs` lib.
+    lovely `platformdirs` lib.
 
     '''
     # lazy-imported to keep it off the eager
@@ -403,6 +349,29 @@ 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(
@@ -432,7 +401,7 @@ def get_rt_dir(
                 f'{subdir!r}\n'
             )
 
-    if os.name != 'posix':
+    if rt_root is None:
         if subdir_path is not None:
             rt_dir = rt_dir / subdir_path
         if not rt_dir.is_dir():
@@ -445,15 +414,33 @@ 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
-            _ensure_owner_only_posix_dir(rt_dir)
+            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