diff --git a/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.md b/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.md new file mode 100644 index 00000000..8325d9d5 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.md @@ -0,0 +1,53 @@ +--- +model: gpt-5.6-sol +service: opencode +session: trio-bindspace-fd-transport +timestamp: 2026-08-28T17:29:43Z +git_ref: 2ca8c570 +scope: code +substantive: true +raw_file: 20260828T172943Z_2ca8c570_prompt_io.raw.md +--- + +## Prompt + +After committing generic pre-handshake child-death detection, the human +approved the proposed next Layer C slice: transport a live bindspace +capability through the Trio spawn backend. + +## Response summary + +Plumb an optional process-local `Bindspace` through actor spawn APIs. +Give a Trio exec child its own inherited namespace descriptor, preserve +existing caller `pass_fds`, transport the FD/inode pair through the CLI, +and close the temporary parent descriptor on every open-process outcome. +Reject unsupported multiprocessing transport explicitly. Exercise the +complete Trio path with a real actor that enters a distinct network +namespace and reports its namespace and inherited-FD inodes over RPC. + +## Files changed + +- `tractor/runtime/_supervise.py` - accept and relay a child bindspace. +- `tractor/to_actor/_api.py` - expose bindspace spawn configuration. +- `tractor/spawn/_spawn.py` - relay the bindspace to spawn backends. +- `tractor/spawn/_trio.py` - duplicate, transfer, and close the FD. +- `tractor/spawn/_mp.py` - reject unsupported MP transport. +- `tractor/_child.py` - parse and forward the atomic CLI capability. +- `tests/test_netns_spawn.py` - cover transport and cleanup schedules. + +## Human edits + +The human approved the Trio-first transport boundary after reviewing +the proposed dependency order, then supplied staged-diff comments across +multiple review passes. Human-directed revisions replaced abstract +ownership wording with concrete FD and process terminology, clarified +descriptor cleanup, preserved existing `pass_fds` explicitly, and +replaced the mocked success-path test with a real child actor, real +`setns()`, handshake, and RPC round trip inside unprivileged user/network +namespaces. Follow-up review added direct parent FD-table verification +that the temporary child-bound duplicate is closed after successful +spawn. + +Multiprocessing FD reduction and root-process namespace entry remain +separate follow-up slices. No direct manual source edits were observed; +the review-driven source revisions were applied by the agent. diff --git a/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.raw.md b/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.raw.md new file mode 100644 index 00000000..17969926 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.raw.md @@ -0,0 +1,34 @@ +--- +model: gpt-5.6-sol +service: opencode +timestamp: 2026-08-28T17:29:43Z +git_ref: 2ca8c570 +diff_cmd: git diff HEAD~1..HEAD +--- + +# Raw output - transport bindspace FD through Trio spawn + +The human approved the next WireGuard Layer C increment after landing +pre-handshake child-death detection. + +## Generated code + +> `git diff HEAD~1..HEAD -- tractor/_child.py tractor/runtime/_supervise.py tractor/spawn/_spawn.py tractor/spawn/_trio.py tractor/spawn/_mp.py tractor/to_actor/_api.py tests/test_netns_spawn.py` + +Thread an optional live `Bindspace` through actor spawn APIs. For the +Trio backend, duplicate its namespace FD, append that temporary +descriptor to existing `pass_fds`, and send the atomic descriptor/inode +pair through the child CLI. Close the parent temporary duplicate as +soon as `open_process()` returns, raises, or is cancelled, while the +child consumes its inherited descriptor before runtime startup. + +Reject bindspace transport explicitly in multiprocessing backends until +their descriptor-reduction slice is implemented. Add privilege-free +tests for CLI forwarding, merged pass-FD state, caller immutability, +source/duplicate ownership, exec failure, cancellation, absent live +authority, death-before-handshake cleanup, and MP rejection. + +## Scope boundary + +Multiprocessing FD reduction and root-process namespace entry remain +follow-up work. diff --git a/tests/test_netns_spawn.py b/tests/test_netns_spawn.py index 61eb6fc6..3ad0ccd2 100644 --- a/tests/test_netns_spawn.py +++ b/tests/test_netns_spawn.py @@ -8,6 +8,10 @@ import errno from functools import partial import os from pathlib import Path +import shutil +import subprocess +import sys +import tempfile from types import SimpleNamespace from typing import ( Any, @@ -15,17 +19,31 @@ from typing import ( ) import pytest +import trio +import tractor from tractor import _child from tractor.devx import _proctitle +from tractor.discovery._bindspace import ( + Bindspace, + BindspaceRef, + BindspaceSpec, +) +from tractor.msg import Aid from tractor.spawn import ( _entry, + _mp, _netns, _spawn, + _trio, ) from tractor.trionics import patches +_SELF_NETNS_PATH = Path('/proc/self/ns/net') +_SELF_FD_DIR = Path('/proc/self/fd') + + def _assert_fd_closed(namespace_fd: int) -> None: ''' Assert that bootstrap consumed its child-owned descriptor. @@ -37,6 +55,162 @@ def _assert_fd_closed(namespace_fd: int) -> None: assert exc_info.value.errno == errno.EBADF +def _fds_referencing( + reference_fd: int, +) -> set[int]: + ''' + Find this process's FDs for the same open kernel object. + + Snapshotting the matching descriptor numbers around a spawn lets + the E2E test detect a leaked `os.dup()` entry without replacing + `open_process()` or observing the child's descriptor table. + + ''' + reference_stat: os.stat_result = os.fstat(reference_fd) + matching_fds: set[int] = set() + fd_path: Path + for fd_path in _SELF_FD_DIR.iterdir(): + try: + open_fd: int = int(fd_path.name) + open_stat: os.stat_result = os.fstat(open_fd) + except (OSError, ValueError): + continue + + if ( + open_stat.st_dev == reference_stat.st_dev + and + open_stat.st_ino == reference_stat.st_ino + ): + matching_fds.add(open_fd) + + return matching_fds + + +def _bindspace_for_fd(namespace_fd: int) -> Bindspace: + ''' + Build one borrowed stand-in netns capability around a real FD. + + ''' + key: str = 'spawn-test-netns' + inode: int = os.fstat(namespace_fd).st_ino + return Bindspace( + spec=BindspaceSpec( + kind='netns', + key=key, + ), + ref=BindspaceRef( + kind='netns', + key=key, + inode=inode, + ), + namespace_fd=namespace_fd, + ownership='borrowed', + ) + + +class _MockIpcServer: + ''' + Provide the peer-event state used by `trio_proc()` tests. + + ''' + def __init__(self) -> None: + self._peer_connected: dict[ + tuple[str, str], + trio.Event, + ] = {} + + async def wait_for_peer( + self, + child_uid: tuple[str, str], + ) -> tuple[trio.Event, object]: + ''' + Model a child that dies or fails before its handshake. + + ''' + await trio.sleep_forever() + + +def _netns_bootstrap_from_cmd( + command: list[str], +) -> tuple[int, int]: + ''' + Parse the namespace tuple that the exec child would receive. + + ''' + arg_index: int = command.index('--netns_bootstrap') + return _child.parse_netns_bootstrap(command[arg_index + 1]) + + +class _SpawnTestNursery: + ''' + Track provisional Trio child publication during transport tests. + + ''' + def __init__(self) -> None: + self._actor = SimpleNamespace( + ipc_server=_MockIpcServer(), + ) + self._children: dict[ + tuple[str, str], + tuple, + ] = {} + + def _register_child( + self, + subactor: object, + proc: object, + portal: object|None, + ) -> tuple[trio.Event, trio.Event, bool]: + ''' + Publish one provisional child after its peer event exists. + + ''' + uid: tuple[str, str] = subactor.aid.uid + assert uid in self._actor.ipc_server._peer_connected + assert portal is None + self._children[uid] = (subactor, proc, portal) + return (trio.Event(), trio.Event(), False) + + +def _spawn_test_subactor(uid: tuple[str, str]) -> SimpleNamespace: + ''' + Build the actor fields reached before a failed Trio handshake. + + ''' + return SimpleNamespace( + aid=Aid( + name=uid[0], + uuid=uid[1], + ), + loglevel=None, + pformat=lambda: uid[0], + ) + + +async def _report_child_netns( + inherited_fd: int, +) -> tuple[int, int]: + ''' + Report namespace and inherited-FD inodes from a spawned actor. + + `_consume_netns_bootstrap()` has already entered the target netns + and closed its bootstrap FD before this RPC can run. The unrelated + `inherited_fd` must remain open because the caller included it in + `proc_kwargs['pass_fds']` before Trio appended the netns FD. + + ''' + child_netns_fd: int = os.open( + _SELF_NETNS_PATH, + os.O_RDONLY, + ) + try: + child_netns_inode: int = os.fstat(child_netns_fd).st_ino + inherited_inode: int = os.fstat(inherited_fd).st_ino + return child_netns_inode, inherited_inode + finally: + os.close(child_netns_fd) + + def test_enter_netns_rejects_mismatched_inherited_fd( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -268,6 +442,476 @@ def test_netns_entry_error_survives_close_failure( assert 'OverflowError' in entry_error.__notes__[0] +def test_trio_child_cli_forwards_netns_bootstrap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + The exec child must retain one atomic FD-and-inode capability. + + Supply the tuple exactly as the Trio parent emits it and replace + `_actor_child_main()` before any runtime work. The captured kwargs + prove argparse does not split, reorder, or drop either value while + forwarding the capability to the child-owned cleanup boundary. + + ''' + calls: list[dict[str, object]] = [] + uid: tuple[str, str] = ('cli-netns-child', 'test') + parent_addr: tuple[str, int] = ('127.0.0.1', 1616) + bootstrap: tuple[int, int] = (12, 3456) + + def fake_actor_child_main(**kwargs: object) -> None: + ''' + Capture parsed child-bootstrap arguments without starting Trio. + + ''' + calls.append(kwargs) + + monkeypatch.setattr( + _child, + '_actor_child_main', + fake_actor_child_main, + ) + + _child.main([ + '--uid', + str(uid), + '--parent_addr', + str(parent_addr), + '--netns_bootstrap', + str(bootstrap), + ]) + + assert calls == [{ + 'uid': uid, + 'loglevel': None, + 'parent_addr': parent_addr, + 'infect_asyncio': False, + 'spawn_method': 'trio', + 'netns_bootstrap': bootstrap, + }] + + +def test_trio_spawn_requires_live_bindspace_fd() -> None: + ''' + A `BindspaceRef` alone cannot let a child enter its namespace. + + Construct a valid `Bindspace` with its required identity metadata + but no open namespace FD. Calling `trio_proc()` must fail before + `open_process()` because an inode identifies a namespace but does + not provide an open handle that the child can inherit. + + ''' + key: str = 'missing-spawn-netns' + bindspace = Bindspace( + spec=BindspaceSpec( + kind='netns', + key=key, + ), + # A realized bindspace always retains identity metadata; this + # test isolates the missing live-FD condition. + ref=BindspaceRef( + kind='netns', + key=key, + inode=1, + ), + namespace_fd=None, + ownership='borrowed', + ) + uid: tuple[str, str] = ('missing-netns-fd', 'test') + + async def main() -> None: + ''' + Reject the ref-only capability before `open_process()`. + + ''' + with pytest.raises( + ValueError, + match='bindspace.namespace_fd.*required', + ): + await _trio.trio_proc( + name=uid[0], + actor_nursery=_SpawnTestNursery(), + subactor=_spawn_test_subactor(uid), + errors={}, + bind_addrs=[], + parent_addr=('127.0.0.1', 1616), + _runtime_vars={}, + bindspace=bindspace, + ) + + trio.run(main) + + +def test_trio_spawn_relays_bindspace_to_child_actor( + tmp_path: Path, + start_method: str, + tpt_proto: str, +) -> None: + ''' + Move a subactor into the relayed bindspace netns. + + Re-exec this one test inside an unprivileged user/net namespace, + then move the nested pytest parent into a second netns. The actor + initially inherits the second namespace but receives an FD for the + first. UDS keeps the parent handshake reachable across the netns + boundary. The child reports its resulting namespace inode and a + caller-supplied inherited FD over a real `Portal`, proving the exec + CLI, merged `pass_fds`, `setns()`, handshake, and RPC path. + + ''' + if start_method != 'trio': + pytest.skip('bindspace FD relay is implemented by Trio spawn') + + reexec_var: str = 'TRACTOR_TEST_NETNS_E2E_REEXEC' + if os.environ.get(reexec_var) != '1': + unshare_path: str|None = shutil.which('unshare') + if unshare_path is None: + pytest.skip('`unshare` is unavailable') + + # Give nested pytest `CAP_SYS_ADMIN` only inside a disposable + # user namespace. The first `--net` creates the target netns; + # the nested test retains its FD before creating a second netns + # for the parent and child to inherit at spawn. Probe separately + # so hosts disabling unprivileged user namespaces skip cleanly. + probe = subprocess.run( + [ + unshare_path, + '--user', + '--map-root-user', + '--net', + 'true', + ], + capture_output=True, + text=True, + check=False, + ) + if probe.returncode: + reason: str = probe.stderr.strip() + pytest.skip( + f'unprivileged user/net namespaces unavailable: ' + f'{reason}' + ) + + nested_env: dict[str, str] = dict(os.environ) + nested_env[reexec_var] = '1' + nested_env['VIRTUAL_ENV'] = sys.prefix + nested_rt_dir: Path = Path( + tempfile.mkdtemp(prefix='tne-') + ) + nested_env['XDG_RUNTIME_DIR'] = str(nested_rt_dir) + python_bin: str = str(Path(sys.executable).parent) + nested_env['PATH'] = ( + python_bin + + os.pathsep + + nested_env['PATH'] + ) + test_id: str = ( + 'tests/test_netns_spawn.py::' + 'test_trio_spawn_relays_bindspace_to_child_actor' + ) + try: + subprocess.run( + [ + unshare_path, + '--user', + '--map-root-user', + '--net', + sys.executable, + '-m', + 'pytest', + test_id, + '--spawn-backend=trio', + '--tpt-proto=uds', + '-x', + '--tb=short', + '--no-header', + '--timeout=30', + ], + env=nested_env, + check=True, + ) + finally: + shutil.rmtree(nested_rt_dir) + return + + assert start_method == 'trio' + assert tpt_proto == 'uds' + + # Namespace creation/realization is outside this transport slice: + # production spawn accepts an already-open `namespace_fd`. Build + # both disposable namespaces directly for this E2E boundary. + target_netns_fd: int = os.open( + _SELF_NETNS_PATH, + os.O_RDONLY, + ) + target_netns_inode: int = os.fstat(target_netns_fd).st_ino + initial_target_fds: set[int] = _fds_referencing( + target_netns_fd, + ) + assert target_netns_fd in initial_target_fds + + # Move the parent to a second netns after retaining an FD for the + # first. The child must use that FD to differ from its parent. + os.unshare(os.CLONE_NEWNET) + parent_netns_fd: int = os.open( + _SELF_NETNS_PATH, + os.O_RDONLY, + ) + parent_netns_inode: int = os.fstat(parent_netns_fd).st_ino + assert parent_netns_inode != target_netns_inode + + inherited_path: Path = tmp_path / 'caller-pass-fd' + inherited_path.touch() + inherited_fd: int = os.open(inherited_path, os.O_RDONLY) + bindspace: Bindspace = _bindspace_for_fd(target_netns_fd) + + async def main() -> None: + ''' + Start the child and receive its namespace observations by RPC. + + ''' + async with tractor.open_nursery() as actor_nursery: + portal: tractor.Portal = await actor_nursery.start_actor( + 'netns-bootstrap-child', + bindspace=bindspace, + enable_modules=[__name__], + proc_kwargs={ + 'pass_fds': (inherited_fd,), + }, + ) + report: tuple[int, int] = await portal.run( + _report_child_netns, + inherited_fd=inherited_fd, + ) + + ( + child_netns_inode, + inherited_inode, + ) = report + assert child_netns_inode == target_netns_inode + assert child_netns_inode != parent_netns_inode + assert inherited_inode == inherited_path.stat().st_ino + await portal.cancel_actor() + + try: + trio.run(main) + # Any `os.dup(target_netns_fd)` entry made for child exec must + # now be absent from the parent's descriptor table. + assert _fds_referencing(target_netns_fd) == initial_target_fds + # Both descriptors supplied by this parent remain open after + # the child exits; only the temporary `os.dup()` FD is closed. + assert os.fstat(target_netns_fd).st_ino == bindspace.ref.inode + assert os.fstat(inherited_fd).st_ino == inherited_path.stat().st_ino + finally: + os.close(parent_netns_fd) + os.close(target_netns_fd) + os.close(inherited_fd) + + +def test_trio_spawn_failure_closes_child_netns_fd_in_parent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A failed exec must close the child netns FD in the parent process. + + Raise one unique error from `open_process()` after capturing and + validating the FD made by `os.dup(Bindspace.namespace_fd)`. Since + exec fails, no child inherits it. The backend must close that + parent descriptor, preserve the original error, leave the original + bindspace FD open, and avoid removing a child record that was never + added to `ActorNursery._children`. + + ''' + namespace_path: Path = tmp_path / 'failed-trio-bindspace' + namespace_path.touch() + namespace_fd: int = os.open(namespace_path, os.O_RDONLY) + bindspace: Bindspace = _bindspace_for_fd(namespace_fd) + uid: tuple[str, str] = ('failed-netns-exec', 'test') + child_fds: list[int] = [] + open_error = OSError('could not exec child') + + async def fail_open_process( + command: list[str], + **kwargs: object, + ) -> trio.Process: + ''' + Fail after checking the child FD in `pass_fds` and the CLI. + + ''' + child_fd, expected_inode = _netns_bootstrap_from_cmd(command) + assert kwargs['pass_fds'] == (child_fd,) + assert os.fstat(child_fd).st_ino == expected_inode + child_fds.append(child_fd) + raise open_error + + monkeypatch.setattr( + _trio.trio.lowlevel, + 'open_process', + fail_open_process, + ) + actor_nursery = _SpawnTestNursery() + + async def main() -> None: + ''' + Exercise cleanup before Trio child publication. + + ''' + with pytest.raises(OSError) as exc_info: + await _trio.trio_proc( + name=uid[0], + actor_nursery=actor_nursery, + subactor=_spawn_test_subactor(uid), + errors={}, + bind_addrs=[], + parent_addr=('127.0.0.1', 1616), + _runtime_vars={}, + bindspace=bindspace, + ) + + assert exc_info.value is open_error + + try: + trio.run(main) + assert len(child_fds) == 1 + _assert_fd_closed(child_fds[0]) + assert os.fstat(namespace_fd).st_ino == bindspace.ref.inode + assert actor_nursery._children == {} + finally: + os.close(namespace_fd) + + +def test_trio_spawn_cancel_closes_child_netns_fd_in_parent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Cancellation during exec must close the parent-side child netns FD. + + Park `open_process()` after it receives the descriptor made by + `os.dup(Bindspace.namespace_fd)`, then cancel the task running + `trio_proc()`. The controlled event fixes the cancellation point + inside the open call. Cleanup must close that descriptor in the + parent while preserving the original bindspace FD; no process or + `ActorNursery._children` entry exists to reap at this schedule. + + ''' + namespace_path: Path = tmp_path / 'cancelled-trio-bindspace' + namespace_path.touch() + namespace_fd: int = os.open(namespace_path, os.O_RDONLY) + bindspace: Bindspace = _bindspace_for_fd(namespace_fd) + uid: tuple[str, str] = ('cancelled-netns-exec', 'test') + open_called = trio.Event() + child_fds: list[int] = [] + + async def park_open_process( + command: list[str], + **kwargs: object, + ) -> trio.Process: + ''' + Record the child FD, then signal that the open call is parked. + + ''' + child_fd, expected_inode = _netns_bootstrap_from_cmd(command) + assert kwargs['pass_fds'] == (child_fd,) + assert os.fstat(child_fd).st_ino == expected_inode + child_fds.append(child_fd) + open_called.set() + try: + await trio.sleep_forever() + except trio.Cancelled: + raise + + monkeypatch.setattr( + _trio.trio.lowlevel, + 'open_process', + park_open_process, + ) + actor_nursery = _SpawnTestNursery() + + async def run_spawn() -> None: + ''' + Keep cancellation propagation explicit at the backend task. + + ''' + try: + await _trio.trio_proc( + name=uid[0], + actor_nursery=actor_nursery, + subactor=_spawn_test_subactor(uid), + errors={}, + bind_addrs=[], + parent_addr=('127.0.0.1', 1616), + _runtime_vars={}, + bindspace=bindspace, + ) + except trio.Cancelled: + raise + + async def main() -> None: + ''' + Cancel only after `open_process()` owns the checkpoint. + + ''' + async with trio.open_nursery() as nursery: + nursery.start_soon(run_spawn) + await open_called.wait() + nursery.cancel_scope.cancel() + + try: + trio.run(main) + assert len(child_fds) == 1 + _assert_fd_closed(child_fds[0]) + assert os.fstat(namespace_fd).st_ino == bindspace.ref.inode + assert actor_nursery._children == {} + finally: + os.close(namespace_fd) + + +def test_mp_spawn_rejects_bindspace_transport( + tmp_path: Path, +) -> None: + ''' + Unimplemented MP FD transfer must fail before process creation. + + Supply one valid live bindspace directly to the multiprocessing + backend. Until spawn/forkserver reduction gives the child exclusive + descriptor ownership, both variants must raise the same actionable + error instead of silently booting an actor in the parent's netns. + + ''' + namespace_path: Path = tmp_path / 'mp-bindspace' + namespace_path.touch() + namespace_fd: int = os.open(namespace_path, os.O_RDONLY) + bindspace: Bindspace = _bindspace_for_fd(namespace_fd) + + async def main() -> None: + ''' + Invoke the backend before any multiprocessing context access. + + ''' + with pytest.raises( + NotImplementedError, + match='multiprocessing spawn backends', + ): + await _mp.mp_proc( + name='unsupported-netns-child', + actor_nursery=None, # type: ignore[arg-type] + subactor=None, # type: ignore[arg-type] + errors={}, + bind_addrs=[], + parent_addr=('127.0.0.1', 1616), + _runtime_vars={}, + bindspace=bindspace, + ) + + try: + trio.run(main) + assert os.fstat(namespace_fd).st_ino == bindspace.ref.inode + finally: + os.close(namespace_fd) + + @pytest.mark.parametrize('backend', ('mp', 'trio')) def test_child_entry_consumes_netns_before_runtime( backend: str, diff --git a/tractor/_child.py b/tractor/_child.py index d356f33d..3c5a1c30 100644 --- a/tractor/_child.py +++ b/tractor/_child.py @@ -49,6 +49,34 @@ def parse_ipaddr(arg): return arg +def parse_netns_bootstrap(arg: str) -> tuple[int, int]: + ''' + Parse one atomic inherited namespace capability. + + Descriptor and inode validation remains in + `_consume_netns_bootstrap()` so every valid descriptor-shaped + input reaches its exact child-owned cleanup boundary. + + ''' + try: + value: object = literal_eval(arg) + except (ValueError, SyntaxError) as exc: + raise argparse.ArgumentTypeError( + 'netns bootstrap must be an `(fd, inode)` tuple' + ) from exc + + if ( + not isinstance(value, tuple) + or + len(value) != 2 + ): + raise argparse.ArgumentTypeError( + 'netns bootstrap must be an `(fd, inode)` tuple' + ) + + return value + + def _actor_child_main( uid: tuple[str, str], loglevel: str | None, @@ -123,7 +151,11 @@ def _actor_child_main( ) -if __name__ == "__main__": +def main(argv: list[str]|None = None) -> None: + ''' + Parse Trio child-bootstrap arguments and enter actor runtime. + + ''' __tracebackhide__: bool = True parser = argparse.ArgumentParser() @@ -131,7 +163,11 @@ if __name__ == "__main__": parser.add_argument("--loglevel", type=str) parser.add_argument("--parent_addr", type=parse_ipaddr) parser.add_argument("--asyncio", action='store_true') - args = parser.parse_args() + parser.add_argument( + '--netns_bootstrap', + type=parse_netns_bootstrap, + ) + args = parser.parse_args(argv) _actor_child_main( uid=args.uid, @@ -139,4 +175,9 @@ if __name__ == "__main__": parent_addr=args.parent_addr, infect_asyncio=args.asyncio, spawn_method='trio', + netns_bootstrap=args.netns_bootstrap, ) + + +if __name__ == "__main__": + main() diff --git a/tractor/runtime/_supervise.py b/tractor/runtime/_supervise.py index 66f87357..546a31c4 100644 --- a/tractor/runtime/_supervise.py +++ b/tractor/runtime/_supervise.py @@ -65,6 +65,7 @@ from ..spawn import _spawn if TYPE_CHECKING: import multiprocessing as mp + from ..discovery._bindspace import Bindspace # from ..ipc._server import IPCServer from ..ipc import IPCServer from ..spawn._spawn import ProcessType @@ -419,6 +420,7 @@ class ActorNursery: *, bind_addrs: list[UnwrappedAddress]|None = None, + bindspace: 'Bindspace|None' = None, rpc_module_paths: list[str]|None = None, enable_transports: list[str] = [_state._def_tpt_proto], enable_modules: list[str]|None = None, @@ -504,6 +506,7 @@ class ActorNursery: bind_addrs, parent_addr, _rtv, # run time vars + bindspace=bindspace, infect_asyncio=infect_asyncio, proc_kwargs=proc_kwargs ) diff --git a/tractor/spawn/_mp.py b/tractor/spawn/_mp.py index b0a2b529..e1a62ceb 100644 --- a/tractor/spawn/_mp.py +++ b/tractor/spawn/_mp.py @@ -54,6 +54,7 @@ from ._spawn import ( if TYPE_CHECKING: + from tractor.discovery._bindspace import Bindspace from tractor.ipc import ( _server, ) @@ -73,12 +74,19 @@ async def mp_proc( parent_addr: UnwrappedAddress, _runtime_vars: dict[str, Any], # serialized and sent to _child *, + bindspace: Bindspace|None = None, infect_asyncio: bool = False, task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED, proc_kwargs: dict[str, any] = {} ) -> None: + if bindspace is not None: + raise NotImplementedError( + 'Network namespace bindspace transport is not yet ' + 'supported by multiprocessing spawn backends!' + ) + # uggh zone try: from multiprocessing import semaphore_tracker # type: ignore diff --git a/tractor/spawn/_spawn.py b/tractor/spawn/_spawn.py index e30afec3..1a66e23f 100644 --- a/tractor/spawn/_spawn.py +++ b/tractor/spawn/_spawn.py @@ -50,6 +50,7 @@ from tractor.msg import types as msgtypes if TYPE_CHECKING: + from tractor.discovery._bindspace import Bindspace from tractor.ipc import ( _server, Channel, @@ -439,6 +440,7 @@ async def new_proc( *, + bindspace: Bindspace|None = None, infect_asyncio: bool = False, task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED, proc_kwargs: dict[str, any] = {} @@ -460,6 +462,7 @@ async def new_proc( bind_addrs, parent_addr, _runtime_vars, # run time vars + bindspace=bindspace, infect_asyncio=infect_asyncio, task_status=task_status, proc_kwargs=proc_kwargs diff --git a/tractor/spawn/_trio.py b/tractor/spawn/_trio.py index 07b34b4b..956a8b8a 100644 --- a/tractor/spawn/_trio.py +++ b/tractor/spawn/_trio.py @@ -22,6 +22,7 @@ Spawns sub-actors as fresh OS processes driven by ''' from __future__ import annotations +import os import sys from typing import ( Any, @@ -56,6 +57,7 @@ from ._spawn import ( if TYPE_CHECKING: + from tractor.discovery._bindspace import Bindspace from tractor.ipc import ( _server, ) @@ -76,6 +78,7 @@ async def trio_proc( parent_addr: UnwrappedAddress, _runtime_vars: dict[str, Any], # serialized and sent to _child *, + bindspace: Bindspace|None = None, infect_asyncio: bool = False, task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED, proc_kwargs: dict[str, any] = {} @@ -119,16 +122,61 @@ async def trio_proc( if infect_asyncio: spawn_cmd.append('--asyncio') + child_netns_fd: int|None = None + if bindspace is not None: + if (namespace_fd := bindspace.namespace_fd) is None: + raise ValueError( + '`bindspace.namespace_fd` is required for ' + 'Trio child transport!' + ) + + # Snapshot caller-owned process options before duplicating the + # live `Bindspace.namespace_fd`. No checkpoint separates this + # setup from `open_process()` below. + inherited_fds: tuple[int, ...] = tuple( + proc_kwargs.get('pass_fds', ()) + ) + proc_kwargs = dict(proc_kwargs) + child_netns_fd = os.dup(namespace_fd) + try: + netns_bootstrap: tuple[int, int] = ( + # FD number retained in the child's descriptor table. + child_netns_fd, + # Namespace identity checked before the child enters it. + bindspace.ref.inode, + ) + spawn_cmd.extend(( + '--netns_bootstrap', + str(netns_bootstrap), + )) + + # Keep every descriptor requested by the caller and append + # the namespace FD needed during child bootstrap. + proc_kwargs['pass_fds'] = ( + *inherited_fds, + child_netns_fd, + ) + except BaseException: + os.close(child_netns_fd) + raise + cancelled_during_spawn: bool = False proc: trio.Process|None = None ipc_server: _server.Server = actor_nursery._actor.ipc_server peer_event: trio.Event|None = None + child_registered: bool = False try: try: - proc: trio.Process = await trio.lowlevel.open_process( - spawn_cmd, - **proc_kwargs, - ) + try: + proc: trio.Process = await trio.lowlevel.open_process( + spawn_cmd, + **proc_kwargs, + ) + finally: + if child_netns_fd is not None: + # The child now has its own descriptor-table entry. + # Close the temporary entry in the parent process. + os.close(child_netns_fd) log.runtime( f'Started new child subproc\n' f'(>\n' @@ -157,6 +205,7 @@ async def trio_proc( proc=proc, portal=None, ) + child_registered = True if cancel_during_registration: cancelled_during_spawn = True proc.kill() @@ -334,7 +383,11 @@ async def trio_proc( ): ipc_server._peer_connected.pop(subactor.aid.uid) - if not cancelled_during_spawn: + if ( + child_registered + and + not cancelled_during_spawn + ): # pop child entry to indicate we no longer managing this # subactor actor_nursery._children.pop(subactor.aid.uid) diff --git a/tractor/to_actor/_api.py b/tractor/to_actor/_api.py index 7e6de264..02b54946 100644 --- a/tractor/to_actor/_api.py +++ b/tractor/to_actor/_api.py @@ -61,6 +61,7 @@ from ..runtime._supervise import ( if TYPE_CHECKING: from ..discovery._addr import UnwrappedAddress + from ..discovery._bindspace import Bindspace from ..runtime._portal import Portal @@ -254,6 +255,7 @@ async def run( # is provided. name: str|None = None, bind_addrs: list[UnwrappedAddress]|None = None, + bindspace: Bindspace|None = None, enable_modules: list[str]|None = None, loglevel: str|None = None, debug_mode: bool|None = None, @@ -337,6 +339,7 @@ async def run( (enable_modules or []) ), bind_addrs=bind_addrs, + bindspace=bindspace, loglevel=loglevel, debug_mode=debug_mode, infect_asyncio=infect_asyncio,