Extra child descriptors and session control at spawn (#38)

DESIGN.md

@@ -352,6 +352,10 @@ impl Loop {
 
     // processes and signals
     pub fn spawn(&mut self, spec: &ProcessSpec, tok: Token) -> io::Result<Process>;  // handles for stdio pipes
+    // Node's stdio tail: extra child descriptors at fixed numbers (fork()'s IPC
+    // channel at 3), their parent ends written into the caller's slice.
+    pub fn spawn_extra(&mut self, spec: &ProcessSpec, tok: Token,
+                       parents: &mut [Option<Handle>]) -> io::Result<Process>;
     pub fn kill(&mut self, p: Handle, sig: Signal) -> io::Result<()>;
     pub fn signal_start(&mut self, sig: Signal, tok: Token) -> io::Result<Handle>;
 
@@ -373,6 +377,9 @@ pub enum OpResult {
     Cancelled, Closed, Stopped, Err(Error),
 }
 pub struct Error { pub kind: ErrorKind, pub os: Option<i32> }   // host maps to ECONNRESET etc.
+
+pub struct ChildFd { pub number: u32, pub source: ChildFdSource }   // number in 3..=MAX_CHILD_FD
+pub enum ChildFdSource { Null, Pipe, Duplex, Handle(Handle) }       // Duplex is Node's 'pipe'/'ipc'
 ```
 
 ## 7. Platform design
@@ -398,7 +405,7 @@ pub struct Error { pub kind: ErrorKind, pub os: Option<i32> }   // host maps to
 - **Named pipes:** overlapped `ConnectNamedPipe`/`ReadFile`/`WriteFile`. This is the `pipe_listen`/`pipe_connect` transport (Node IPC uses named pipes on Windows).
 - **Stdio that isn't overlapped** (a handle inherited as a synchronous pipe or file): a dedicated reader thread per handle that posts completions. It isn't possible to reopen such a handle overlapped. libuv does the same.
 - **Console/TTY:** `ReadConsoleInputW` on a reader thread, with VT input and output modes (`ENABLE_VIRTUAL_TERMINAL_PROCESSING`/`_INPUT`). Resize events become a `Signal::WinCh` completion; Perry has no resize support on Windows today (`tty.rs:9–12`).
-- **Processes:** `CreateProcessW` with overlapped pipe handles, a Job Object for kill-tree semantics, and `RegisterWaitForSingleObject` on the process handle to post the exit completion.
+- **Processes:** `CreateProcessW` with overlapped pipe handles, a Job Object for kill-tree semantics, and `RegisterWaitForSingleObject` on the process handle to post the exit completion. Descriptors beyond stdin/stdout/stderr are published through the C run-time inherited-descriptor block in `STARTUPINFOW.lpReserved2` (count, per-descriptor flags from `GetFileType`, then the handles), plus `PROC_THREAD_ATTRIBUTE_HANDLE_LIST`. That is libuv's and Node's own convention, so a child descriptor number — `NODE_CHANNEL_FD=3` — means the same thing here as it does under Node. A child that does not use the C run-time inherits the handles but has no number for them.
 - **Signals:** `SetConsoleCtrlHandler` for CTRL_C, CTRL_BREAK and CTRL_CLOSE, mapped to `SIGINT`/`SIGBREAK`/`SIGHUP`. SIGTERM has no console equivalent; documented as such, matching Perry today (`os/signal.rs:441–531`).
 - **Wake:** `PostQueuedCompletionStatus` with a reserved completion key.
 - **Timer precision:** the default system tick is about 15.6 ms and `GetQueuedCompletionStatusEx` timeouts round to it. Use a **high-resolution waitable timer** (`CREATE_WAITABLE_TIMER_HIGH_RESOLUTION`, Windows 10 1803+) armed to the next deadline and associated with the port through dynamically resolved `NtAssociateWaitCompletionPacket`. `GetQueuedCompletionStatusEx` is nonalertable. Cancellation returning `STATUS_PENDING` requires dequeuing the generation-tagged timer packet before reuse. M1 prototyped both APC and NT packet routes; §15 question 3 records the NT packet decision and measured lateness. `timeBeginPeriod` is not an acceptable default because it changes the tick system-wide.
@@ -455,6 +462,8 @@ Two backends, because both versions matter now:
 | Stdio pipes | readiness | readiness | overlapped, or reader thread | `wasi:cli` streams | `wasi:cli` streams | unsupported |
 | TTY | termios + readiness | termios + readiness | console API reader thread, VT modes | size only | size only | unsupported |
 | Child processes | pidfd / SIGCHLD | EVFILT_PROC | RegisterWaitForSingleObject + Job Object | unsupported | unsupported | unsupported |
+| Child descriptors beyond stdio | one `pre_exec` `dup2` per number, sources relocated above every target first | same | C run-time inherited-descriptor block + handle-list attribute | unsupported | unsupported | unsupported |
+| Child session control | `setsid` (`detached`), `setpgid` (`new_process_group`), `TIOCSCTTY` (`controlling_terminal`) | same | detached process group + Job Object; no controlling terminal | unsupported | unsupported | unsupported |
 | Signals | sigaction + self-pipe | EVFILT_SIGNAL | SetConsoleCtrlHandler | unsupported | unsupported | unsupported |
 | Files | blocking pool | blocking pool | blocking pool | `wasi:filesystem` preopens, run in `turn` | `wasi:filesystem` async preopens, run in `turn` | unsupported (no OPFS host mapping defined) |
 | File watch | inotify (recursion by the host) | FSEvents for directories (recursive), kqueue for files | ReadDirectoryChangesW (recursive) | unsupported | unsupported | unsupported |
@@ -511,6 +520,10 @@ and work on any live socket handle, including one produced by `accept`.
 | Microtasks, nextTick | Perry | unchanged |
 | Error text and codes | Perry | Map `Error { kind, os }` to Node's `code`/`errno`/`syscall` |
 | Keep-alive for JS-level resources | Perry | Uses turnloop `alive()` plus its own JS timers until those move (P3) |
+| Child descriptor policy | Perry | Which number is the IPC channel, and the name and value of `NODE_CHANNEL_FD`; turnloop places the descriptor, never names it |
+| Program resolution and shell | Perry | `shell: true`, `execPath`, PATH policy and argv construction |
+| Pty allocation | Perry | `openpty` and its termios; turnloop owns only the child-side `setsid`/`TIOCSCTTY` through `ProcessSpec::controlling_terminal` |
+| Reaping a child turnloop did not spawn | Perry | turnloop reaps only its own children, always by their own identity (`waitpid(pid, …)`, never `-1`), so a host waiter in the same process keeps its own children's status. There is no `adopt_process`; see `docs/lanes/procspec.md` |
 | Thread-pool jobs touching JS | never | Pool jobs are `Send` Rust closures; results convert on the main thread |
 
 **Wiring for P0:**

crates/turnloop-contract/src/bin/native_child.rs

@@ -21,6 +21,12 @@ fn main() {
             println!("60 service timer expiries; no-spin bounds passed");
         }
         "exit" => std::process::exit(23),
+        "exit-with" => std::process::exit(
+            args[2]
+                .to_str()
+                .and_then(|code| code.parse().ok())
+                .expect("exit code"),
+        ),
         "sleep" => std::thread::sleep(Duration::from_secs(60)),
         "roundtrip" => {
             let mut stdout = std::io::stdout().lock();
@@ -89,6 +95,61 @@ fn main() {
         "handle" => native_handle(args.get(2).expect("pipe path")),
         #[cfg(any(target_vendor = "apple", target_os = "freebsd"))]
         "blocked-signal" => blocked_signal(),
+        // Both extra descriptors carry bytes in both directions. The channel
+        // number is read out of the environment exactly as Node reads
+        // NODE_CHANNEL_FD, so this proves the handoff, not a hard-coded 3.
+        #[cfg(any(unix, windows))]
+        "channel" => {
+            let channel = numbered_fd("NODE_CHANNEL_FD");
+            let extra = numbered_fd("TURNLOOP_EXTRA_FD");
+            let mut request = [0u8; 6];
+            read_exact(channel, &mut request);
+            assert_eq!(&request, b"ping-3", "channel request");
+            write_all(channel, b"pong-3");
+            read_exact(extra, &mut request);
+            assert_eq!(&request, b"ping-4", "extra request");
+            write_all(extra, b"pong-4");
+            print!("ok");
+            std::io::stdout().flush().expect("flush transcript");
+        }
+        // One-way pipe, null device and an adopted parent transport.
+        #[cfg(any(unix, windows))]
+        "extra-sources" => {
+            let (one_way, null, adopted) = (fd_stream(3), fd_stream(4), fd_stream(5));
+            write_all(one_way, b"three");
+            write_all(null, b"void");
+            let mut discard = [0u8; 4];
+            assert_eq!(read_once(null, &mut discard), 0, "null device reads EOF");
+            write_all(adopted, b"five");
+            print!("ok");
+            std::io::stdout().flush().expect("flush transcript");
+        }
+        #[cfg(unix)]
+        "tty-session" => {
+            // SAFETY: queries this process only; no pointer arguments or mutation.
+            let (pid, group, session) =
+                unsafe { (libc::getpid(), libc::getpgrp(), libc::getsid(0)) };
+            // SAFETY: constant device path with integer flags; -1 means no
+            // controlling terminal, which is the distinction under test.
+            let terminal = unsafe { libc::open(c"/dev/tty".as_ptr(), libc::O_RDWR) };
+            let foreground = if terminal < 0 {
+                -1
+            } else {
+                // SAFETY: live descriptor on this process's controlling terminal.
+                unsafe { libc::tcgetpgrp(terminal) }
+            };
+            if terminal >= 0 {
+                // SAFETY: closing the descriptor this branch just opened.
+                unsafe {
+                    libc::close(terminal);
+                }
+            }
+            println!("{pid}:{group}:{session}:{}", i32::from(terminal >= 0));
+            assert_eq!(
+                foreground, group,
+                "child leads its terminal foreground group"
+            );
+        }
         "copy" => {
             let mut bytes = Vec::new();
             std::io::stdin()
@@ -317,3 +378,115 @@ fn native_handle(path: &std::ffi::OsStr) {
 fn native_handle(_: &std::ffi::OsStr) {
     panic!("instantiate handle fixture on production native backend");
 }
+
+/// A child descriptor as this platform's stream identity: the number itself on
+/// Unix, and the C run-time's handle for it on Windows, which is precisely how
+/// libuv's `uv_pipe_open` turns Node's `NODE_CHANNEL_FD` into a usable stream.
+#[cfg(unix)]
+type FdStream = std::os::fd::RawFd;
+#[cfg(windows)]
+type FdStream = windows_sys::Win32::Foundation::HANDLE;
+
+#[cfg(windows)]
+unsafe extern "C" {
+    fn _get_osfhandle(fd: i32) -> isize;
+}
+
+#[cfg(any(unix, windows))]
+fn fd_stream(fd: i32) -> FdStream {
+    #[cfg(unix)]
+    {
+        fd
+    }
+    #[cfg(windows)]
+    {
+        // SAFETY: the C run-time owns its descriptor table; this only reads it.
+        let handle = unsafe { _get_osfhandle(fd) };
+        assert!(handle > 0, "descriptor {fd} is absent from the C run-time");
+        handle as FdStream
+    }
+}
+
+#[cfg(any(unix, windows))]
+fn numbered_fd(variable: &str) -> FdStream {
+    let value = std::env::var(variable).unwrap_or_else(|_| panic!("{variable} is unset"));
+    fd_stream(value.parse().expect("descriptor number"))
+}
+
+#[cfg(any(unix, windows))]
+fn read_once(stream: FdStream, buffer: &mut [u8]) -> usize {
+    #[cfg(unix)]
+    {
+        // SAFETY: live inherited descriptor and a writable buffer of that length.
+        let n = unsafe { libc::read(stream, buffer.as_mut_ptr().cast(), buffer.len()) };
+        assert!(n >= 0, "read: {}", std::io::Error::last_os_error());
+        n as usize
+    }
+    #[cfg(windows)]
+    {
+        use windows_sys::Win32::{Foundation::ERROR_BROKEN_PIPE, Storage::FileSystem::ReadFile};
+        let mut read = 0;
+        // SAFETY: live inherited handle, writable buffer of the stated length,
+        // and a synchronous handle so no OVERLAPPED is required.
+        let ok = unsafe {
+            ReadFile(
+                stream,
+                buffer.as_mut_ptr(),
+                buffer.len() as u32,
+                &mut read,
+                std::ptr::null_mut(),
+            )
+        };
+        if ok == 0 {
+            let error = std::io::Error::last_os_error();
+            assert_eq!(
+                error.raw_os_error(),
+                Some(ERROR_BROKEN_PIPE as i32),
+                "read: {error}"
+            );
+            return 0;
+        }
+        read as usize
+    }
+}
+
+#[cfg(any(unix, windows))]
+fn read_exact(stream: FdStream, buffer: &mut [u8]) {
+    let mut filled = 0;
+    while filled < buffer.len() {
+        let n = read_once(stream, &mut buffer[filled..]);
+        assert_ne!(n, 0, "premature end of stream after {filled} bytes");
+        filled += n;
+    }
+}
+
+#[cfg(any(unix, windows))]
+fn write_all(stream: FdStream, mut bytes: &[u8]) {
+    while !bytes.is_empty() {
+        #[cfg(unix)]
+        // SAFETY: live inherited descriptor and a readable buffer of that length.
+        let n = unsafe { libc::write(stream, bytes.as_ptr().cast(), bytes.len()) };
+        #[cfg(unix)]
+        assert!(n > 0, "write: {}", std::io::Error::last_os_error());
+        #[cfg(unix)]
+        let n = n as usize;
+        #[cfg(windows)]
+        let n = {
+            let mut wrote = 0;
+            // SAFETY: live inherited handle, readable buffer of the stated
+            // length, and a synchronous handle so no OVERLAPPED is required.
+            let ok = unsafe {
+                windows_sys::Win32::Storage::FileSystem::WriteFile(
+                    stream,
+                    bytes.as_ptr(),
+                    bytes.len() as u32,
+                    &mut wrote,
+                    std::ptr::null_mut(),
+                )
+            };
+            assert_ne!(ok, 0, "write: {}", std::io::Error::last_os_error());
+            wrote as usize
+        };
+        bytes = &bytes[n..];
+    }
+}

crates/turnloop-contract/tests/allocations.rs

@@ -1175,6 +1175,157 @@ fn executor_steady_io_poll_and_sleep_allocate_nothing() {
     assert_eq!(count, 1001);
 }
 
+/// Steady-state traffic on a child's extra descriptors, on both platforms.
+///
+/// Spawning is resource creation and is charged to setup, like any other open.
+/// What must be free is everything after it: the writes, the reads, the exit
+/// and the closes of a child holding two descriptors beyond its standard three.
+#[cfg(any(unix, windows))]
+#[test]
+fn extra_child_descriptor_traffic_allocates_nothing_after_spawn() {
+    const CHILDREN: usize = 8;
+    /// `ping-3` then `ping-4`, so one static region serves both writes.
+    static PINGS: [u8; 12] = *b"ping-3ping-4";
+    fn ping(extra: usize) -> WriteBuf {
+        // SAFETY: a static region outlives every completion and the loop itself.
+        WriteBuf::Provided(unsafe { IoBuf::from_raw_parts(PINGS[extra * 6..].as_ptr(), 6) })
+    }
+    /// Turn until the awaited handle has produced `want` bytes, folding every
+    /// other completion into the counters. Nothing here allocates.
+    fn collect(l: &mut Loop, out: &mut Completions, h: Handle, want: &[u8], seen: &mut [usize; 4]) {
+        let mut got = [0u8; 6];
+        let mut filled = 0;
+        while filled < want.len() {
+            l.read(h, ReadBuf::Pooled, Token(70)).expect("read");
+            let before = filled;
+            while filled == before {
+                let until = l.now() + Duration::from_secs(10);
+                assert!(l.now() < until);
+                l.turn(Timeout::Until(until), out).expect("turn");
+                for c in out.drain() {
+                    match c.result {
+                        OpResult::Read {
+                            n,
+                            lease: Some(bytes),
+                        } => {
+                            assert_eq!(c.handle, Some(h), "only one read is in flight");
+                            assert!(n > 0);
+                            got[filled..filled + n].copy_from_slice(bytes.as_slice());
+                            filled += n;
+                        }
+                        OpResult::Wrote(n) => {
+                            assert_eq!(n, 6);
+                            seen[0] += 1;
+                        }
+                        OpResult::Exited(status) => {
+                            assert_eq!(status.code, Some(0));
+                            seen[1] += 1;
+                        }
+                        OpResult::Eof => seen[2] += 1,
+                        OpResult::Closed => seen[3] += 1,
+                        other => panic!("unexpected {other:?}"),
+                    }
+                }
+            }
+        }
+        assert_eq!(filled, want.len(), "exchanges are fixed length");
+        assert_eq!(&got[..filled], want, "the child answered");
+    }
+    let mut l = Loop::new(Config::default()).expect("loop");
+    let mut spec = ProcessSpec::new(env!("CARGO_BIN_EXE_native_child"));
+    spec.windows_hide = true;
+    spec.args = vec!["channel".into()];
+    spec.stdio = [ProcessStdio::Null, ProcessStdio::Pipe, ProcessStdio::Null];
+    spec.extra = vec![
+        ChildFd {
+            number: 3,
+            source: ChildFdSource::Duplex,
+        },
+        ChildFd {
+            number: 4,
+            source: ChildFdSource::Duplex,
+        },
+    ];
+    spec.env = vec![
+        ("NODE_CHANNEL_FD".into(), "3".into()),
+        ("TURNLOOP_EXTRA_FD".into(), "4".into()),
+    ];
+    let mut ends = [[None; 2]; CHILDREN];
+    let children: [_; CHILDREN] = std::array::from_fn(|i| {
+        l.spawn_extra(&spec, Token(i as u64), &mut ends[i])
+            .expect("child with extra descriptors")
+    });
+    let mut out = Completions::default();
+    // Warm the notifier, services and pooled buffers, leaving the children live.
+    l.turn(Timeout::Now, &mut out).expect("warm services");
+    assert!(out.is_empty());
+    let mut seen = [0usize; 4];
+    ALLOCS.with(|n| n.set(0));
+    ACTIVE.with(|v| v.set(true));
+    for (i, child) in children.iter().enumerate() {
+        for (extra, end) in ends[i].iter().enumerate() {
+            let h = end.expect("parent end");
+            l.write(h, ping(extra), Token(71)).expect("write ping");
+            collect(
+                &mut l,
+                &mut out,
+                h,
+                if extra == 0 { b"pong-3" } else { b"pong-4" },
+                &mut seen,
+            );
+        }
+        collect(
+            &mut l,
+            &mut out,
+            child.stdout.expect("child stdout"),
+            b"ok",
+            &mut seen,
+        );
+    }
+    // Every child must report its own exit before anything is closed, so a
+    // Cancelled watch can never stand in for a delivered status.
+    let until = l.now() + Duration::from_secs(20);
+    while seen[1] < CHILDREN {
+        assert!(l.now() < until, "exit deadline");
+        l.turn(Timeout::Until(until), &mut out).expect("exits");
+        for c in out.drain() {
+            match c.result {
+                OpResult::Exited(status) => {
+                    assert_eq!(status.code, Some(0));
+                    seen[1] += 1;
+                }
+                OpResult::Eof => seen[2] += 1,
+                other => panic!("unexpected exit-phase completion {other:?}"),
+            }
+        }
+    }
+    for (i, child) in children.iter().enumerate() {
+        for h in [ends[i][0], ends[i][1], child.stdout, Some(child.handle)] {
+            l.close(h.expect("handle"), Token(72)).expect("close");
+        }
+    }
+    while l.alive() {
+        assert!(l.now() < until, "teardown deadline");
+        l.turn(Timeout::Until(until), &mut out).expect("teardown");
+        for c in out.drain() {
+            match c.result {
+                OpResult::Eof => seen[2] += 1,
+                OpResult::Closed => seen[3] += 1,
+                other => panic!("unexpected teardown {other:?}"),
+            }
+        }
+    }
+    ACTIVE.with(|v| v.set(false));
+    assert_eq!(
+        ALLOCS.with(Cell::get),
+        0,
+        "extra-descriptor writes, reads, exits and closes"
+    );
+    assert_eq!(seen[0], CHILDREN * 2, "every ping completed");
+    assert_eq!(seen[1], CHILDREN, "every child exited exactly once");
+    assert_eq!(seen[3], CHILDREN * 4, "every handle closed exactly once");
+}
+
 #[cfg(unix)]
 #[test]
 fn signal_exit_and_external_notification_delivery_allocate_nothing() {