Hand a quiescent handle's descriptor back to the host (#35)

DESIGN.md

@@ -258,6 +258,10 @@ There are two ways a host drives a loop:
 5. **Handle transfer:**
    - **Between loops:** `Loop::detach(h) -> Detached` (`Send`) and `Loop::attach(Detached, token) -> Handle`. For sockets, pipes and servers across threads or workers, it cancels in-flight ops with the usual exactly-once completions before detaching.
    - **Between processes:** fd passing via `SCM_RIGHTS` on Unix and `WSADuplicateSocketW` / `DuplicateHandle` on Windows, exposed on pipe handles so `child.send(msg, handle)` and cluster round-robin can move sockets (today `emitter.rs:430` drops the handle).
+   - **Out of turnloop entirely (host handoff):** the same `detach` followed by `Detached::into_fd()` on Unix, or `into_socket()` / `into_handle()` on Windows. This is Node's mid-stream `socket.upgradeToTLS` — PostgreSQL's `SSLRequest` hands a live, already-connected socket to a TLS layer — so the class of socket that *might* later be upgraded no longer has to choose its transport at creation. `detach` already proves quiescence, so the guarantee the host gets is total: no operation, no buffer, no registration, no completion, ever again, for that transport. Conversion restores what the backend changed on adoption (Unix status flags and termios, Windows console mode) and hands the descriptor over in the mode the loop held it: non-blocking for a loop-created socket.
+     - **Windows:** a handle's IOCP association is permanent and inescapable. Windows cannot dissociate one, rejects a second `CreateIoCompletionPort` with `ERROR_INVALID_PARAMETER`, and neither `WSADuplicateSocketW` nor `DuplicateHandle` escapes it — both produce another descriptor for the *same* socket or file object, which is where the association lives. Quiescence makes it inert (no packet can ever arrive for it), and the receiving host drives the transport with synchronous/non-blocking Winsock calls or with overlapped calls whose `OVERLAPPED.hEvent` has its low-order bit set, which suppresses the completion packet. A named-pipe instance keeps `FILE_FLAG_OVERLAPPED`, so for a pipe the tagged-`hEvent` form is the only one. The route back to completion-port-driven I/O is `attach`: an imported association is exactly what the IOCP backend's overlapped-event routing exists for.
+     - **WASI 0.2/0.3 and web:** `Unsupported`. A WASI socket is a component-model resource handle in the component's own table, not a descriptor, and there is no interface that hands one to the embedder; a browser resource is a host JS object. Neither has an identity a host could act on.
+   - **Reporting only:** `Loop::raw_transport(h) -> RawTransport` reports a live transport's native identity for Node's `socket._handle.fd`. The loop keeps ownership; the value is valid until the handle is closed or detached, and is for reporting and read-only queries, never for I/O, closing, mode changes or registration elsewhere.
 6. **Multi-threaded accept:**
    - **Kernel-balanced:** where the kernel balances load (`SO_REUSEPORT` on Linux/FreeBSD), each loop gets its own listener with `ListenOpts::reuse_port`.
    - **Everywhere else** (macOS doesn't balance, and on Windows a socket can join only one completion port): one accepting loop hands connections to other loops with `detach`/`attach`. The policy (round-robin, least-loaded) belongs to the host.
@@ -311,10 +315,15 @@ impl Loop {
     pub fn poster(&self) -> Poster;             // Send + Sync + Clone: post (token, payload) to THIS loop
     pub fn integration(&mut self) -> io::Result<Integration>; // Fd | Event | HostCallback | RuntimeOwned
 
-    // multithreading (§5a)
+    // multithreading and host handoff (§5a)
     pub fn detach(&mut self, h: Handle) -> io::Result<Detached>;          // Detached: Send
     pub fn attach(&mut self, d: Detached, tok: Token) -> io::Result<Handle>;
     pub fn send_handle(&mut self, pipe: Handle, h: Handle, tok: Token) -> io::Result<OpId>; // SCM_RIGHTS / DuplicateHandle
+    pub fn raw_transport(&self, h: Handle) -> io::Result<RawTransport>;   // borrowed, reporting only: socket._handle.fd
+    // and, on the transport a detach returned, ownership leaves for good:
+    //   unix:    Detached::into_fd(self) -> OwnedFd
+    //   windows: Detached::into_socket(self) -> io::Result<OwnedSocket>
+    //            Detached::into_handle(self) -> io::Result<OwnedHandle>
 
     // timers
     pub fn timer(&mut self, at: Instant, repeat: Option<Duration>, tok: Token) -> Handle;
@@ -441,6 +450,8 @@ Two backends, because both versions matter now:
 | Outbound HTTP | protocol crate | protocol crate | protocol crate | protocol crate or `wasi:http` | protocol crate or `wasi:http` | host `fetch` |
 | WebSocket | protocol crate | protocol crate | protocol crate | protocol crate | protocol crate | host `WebSocket` |
 | Local IPC | AF_UNIX | AF_UNIX | named pipes (overlapped) | unsupported | unsupported | `postMessage` |
+| Descriptor handoff (§5a) | `Detached::into_fd` | `Detached::into_fd` | `into_socket`/`into_handle`; the IOCP association is permanent and inescapable, so synchronous calls or a tagged `hEvent` | unsupported (resource handle, not a descriptor) | unsupported (resource handle, not a descriptor) | unsupported (host object) |
+| Native identity reporting (`_handle.fd`) | `RawTransport::Fd` | `RawTransport::Fd` | `RawTransport::Socket`/`Handle` | unsupported | unsupported | unsupported |
 | 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 |

crates/turnloop-contract/Cargo.toml

@@ -23,7 +23,7 @@ workspace = true
 libc.workspace = true
 
 [target.'cfg(windows)'.dependencies]
-windows-sys = { version = "=0.61.2", features = ["Wdk_Foundation", "Win32_Foundation", "Win32_UI_WindowsAndMessaging", "Win32_System_Threading", "Win32_System_Console", "Win32_Storage_FileSystem", "Win32_Security", "Win32_System_IO", "Win32_System_Pipes"] }
+windows-sys = { version = "=0.61.2", features = ["Wdk_Foundation", "Win32_Foundation", "Win32_Networking_WinSock", "Win32_UI_WindowsAndMessaging", "Win32_System_Threading", "Win32_System_Console", "Win32_Storage_FileSystem", "Win32_Security", "Win32_System_IO", "Win32_System_Pipes"] }
 
 [package.metadata.turnloop-ci]
 role = "contract"

crates/turnloop-contract/src/handoff.rs

@@ -0,0 +1,235 @@
+//! Handing a transport back to the host (DESIGN §5a; issue #35).
+//!
+//! These are the loop-side halves of the contract: which requests are refused,
+//! and what the loop still knows about a transport once the host owns it. The
+//! halves that need the descriptor itself — bytes flowing over it after the
+//! handoff, a TLS handshake on it, the Windows completion-port rules — are
+//! platform code and live in `tests/handoff.rs`, because a descriptor cannot be
+//! used through a generic `B::Detached`.
+use super::*;
+
+/// The error kind of a refused handoff. `expect_err` needs `Debug`, which a
+/// backend's transport type is not required to implement.
+fn refused<B: Backend>(result: Result<B::Detached>, what: &str) -> ErrorKind {
+    match result {
+        Ok(_) => panic!("{what} was handed out"),
+        Err(e) => e.kind,
+    }
+}
+
+/// Drive the loop until every handle is gone, so a fixture cannot leak one.
+fn close_all<B: Backend>(l: &mut Driver<B>, handles: &[Handle]) {
+    let mut out = Completions::default();
+    for (i, h) in handles.iter().enumerate() {
+        if l.close(*h, Token(900 + i as u64)).is_err() {
+            continue;
+        }
+    }
+    let until = l.now() + Duration::from_secs(5);
+    while l.alive() {
+        assert!(l.now() < until, "close never completed");
+        l.turn(Timeout::Until(until), &mut out).expect("turn");
+        out.drain();
+    }
+}
+
+/// A handle with work in flight is never handed out, and cancellation is never
+/// silent: the refusal is `WouldBlock`, the host turns, collects the terminal
+/// completion it is owed, and only then owns the transport.
+pub fn pending_operations_refuse_handoff<B: Backend>() {
+    let mut l = Driver::<B>::new(Config::default()).expect("loop");
+    let (server, client, conn) = pair(&mut l);
+    let mut bytes = [0x3c; 24];
+    // SAFETY: the region stays fixed and untouched until Cancelled is delivered.
+    let provided = unsafe { IoBufMut::from_raw_parts(bytes.as_mut_ptr(), bytes.len()) };
+    let read = l
+        .read(conn, ReadBuf::Provided(provided), Token(1))
+        .expect("pending read");
+    assert_eq!(
+        refused::<B>(l.detach(conn), "a busy handle"),
+        ErrorKind::WouldBlock,
+        "a handle with an outstanding operation must not be handed out"
+    );
+    // The refusal must not have cancelled anything behind the host's back either:
+    // the identity is still live and still refuses, turn after turn.
+    let mut out = Completions::default();
+    assert_eq!(
+        refused::<B>(l.detach(conn), "a busy handle"),
+        ErrorKind::WouldBlock
+    );
+    l.turn(Timeout::Now, &mut out).expect("cancel turn");
+    assert_eq!(out.len(), 1, "exactly one terminal completion is owed");
+    assert_eq!(out[0].op, Some(read));
+    assert!(matches!(out[0].result, OpResult::Cancelled));
+    assert_eq!(bytes, [0x3c; 24], "the loop wrote into a cancelled buffer");
+    out.drain();
+
+    let transport = l.detach(conn).expect("quiescent handoff");
+    assert_eq!(
+        refused::<B>(l.detach(conn), "an already handed-off transport"),
+        ErrorKind::NotFound,
+        "a transport cannot be handed out twice"
+    );
+    assert_eq!(
+        l.raw_transport(conn).expect_err("gone").kind,
+        ErrorKind::NotFound
+    );
+    assert!(
+        l.read(conn, ReadBuf::Pooled, Token(2)).is_err(),
+        "the loop still accepts operations on a transport it gave away"
+    );
+    // Nothing for that handle can arrive any more, however long the host turns.
+    let until = l.now() + Duration::from_millis(60);
+    while l.now() < until {
+        l.turn(Timeout::After(Duration::from_millis(10)), &mut out)
+            .expect("idle turn");
+        assert_eq!(out.len(), 0, "a handed-off transport produced {out:?}");
+    }
+    drop(transport);
+    close_all(&mut l, &[client, server]);
+}
+
+/// Close is the other owner of a transport's end of life; the two never overlap.
+pub fn closing_and_closed_handles_refuse_handoff<B: Backend>() {
+    let mut l = Driver::<B>::new(Config::default()).expect("loop");
+    let (server, client, conn) = pair(&mut l);
+    l.read_start(conn, Token(1)).expect("multishot read");
+    l.close(conn, Token(2)).expect("close");
+    assert!(l.is_closing(conn), "close has not finished yet");
+    assert_eq!(
+        refused::<B>(l.detach(conn), "a closing transport"),
+        ErrorKind::InvalidInput,
+        "a closing transport must not be handed out"
+    );
+    let mut out = Completions::default();
+    let until = l.now() + Duration::from_secs(5);
+    let mut closed = false;
+    while !closed {
+        assert!(l.now() < until, "close never completed");
+        l.turn(Timeout::Until(until), &mut out).expect("turn");
+        for c in out.drain() {
+            closed |= matches!(c.result, OpResult::Closed);
+        }
+    }
+    assert_eq!(
+        refused::<B>(l.detach(conn), "a closed transport"),
+        ErrorKind::NotFound
+    );
+    assert_eq!(
+        l.raw_transport(conn).expect_err("closed handle").kind,
+        ErrorKind::NotFound
+    );
+    close_all(&mut l, &[client, server]);
+}
+
+/// Both ends of an accept are ordinary transports: the listener and the
+/// connection it produced are each handed over, and the loop keeps nothing.
+pub fn listeners_and_accepted_sockets_are_handed_off<B: Backend>() {
+    let mut l = Driver::<B>::new(Config::default()).expect("loop");
+    let (server, client, conn) = pair(&mut l);
+    let accepted = l.detach(conn).expect("accepted socket");
+    let listener = l.detach(server).expect("listener");
+    assert!(
+        l.local_addr(server).is_err() && l.local_addr(conn).is_err(),
+        "the loop still answers for transports it gave away"
+    );
+    // The client is the only handle left, so liveness proves the other two are
+    // gone from the core's accounting and not merely unregistered natively.
+    let mut out = Completions::default();
+    l.turn(Timeout::Now, &mut out).expect("idle turn");
+    assert_eq!(out.len(), 0);
+    assert!(l.alive(), "the remaining client still counts");
+    close_all(&mut l, &[client]);
+    assert!(!l.alive());
+    drop(accepted);
+    drop(listener);
+}
+
+/// Reporting a descriptor is not owning it: the value is stable while the loop
+/// holds the transport, and every non-transport handle says `Unsupported`.
+pub fn raw_transport_reports_live_transports<B: Backend>() {
+    let mut l = Driver::<B>::new(Config::default()).expect("loop");
+    let (server, client, conn) = pair(&mut l);
+    let first = l.raw_transport(conn).expect("live transport");
+    assert_eq!(first, l.raw_transport(conn).expect("stable"));
+    assert_ne!(
+        first,
+        l.raw_transport(client).expect("client"),
+        "two live transports reported the same identity"
+    );
+    assert_eq!(
+        l.raw_transport(server).expect("listener"),
+        l.raw_transport(server).expect("stable listener")
+    );
+    // Reporting is read-only: the socket is still fully usable afterwards.
+    l.write(conn, WriteBuf::Owned(b"report".to_vec()), Token(1))
+        .expect("write");
+    let mut out = Completions::default();
+    let until = l.now() + Duration::from_secs(5);
+    let mut wrote = 0;
+    while wrote == 0 {
+        assert!(l.now() < until, "write after reporting never completed");
+        l.turn(Timeout::Until(until), &mut out).expect("turn");
+        for c in out.drain() {
+            if let OpResult::Wrote(n) = c.result {
+                assert_eq!(n, 6);
+                wrote += 1;
+            }
+        }
+    }
+    assert_eq!(l.raw_transport(conn).expect("still live"), first);
+
+    let timer = l
+        .timer(l.now() + Duration::from_secs(30), None, Token(2))
+        .expect("timer");
+    assert_eq!(
+        l.raw_transport(timer).expect_err("timer").kind,
+        ErrorKind::Unsupported,
+        "a timer is not a transport and has no descriptor"
+    );
+    assert_eq!(
+        refused::<B>(l.detach(timer), "a timer"),
+        ErrorKind::InvalidInput
+    );
+    close_all(&mut l, &[timer, client, conn, server]);
+}
+
+/// A platform with no descriptor to give says so, rather than inventing one.
+/// WASI 0.2 and 0.3 sockets are component-model resource handles and web
+/// resources are host objects: neither has an identity a host could act on.
+pub fn handoff_is_unsupported<B: Backend>() {
+    let mut l = Driver::<B>::new(Config::default()).expect("loop");
+    let (server, client, conn) = pair(&mut l);
+    for h in [server, client, conn] {
+        assert_eq!(
+            refused::<B>(l.detach(h), "a transport on a platform without descriptors"),
+            ErrorKind::Unsupported
+        );
+        assert_eq!(
+            l.raw_transport(h).expect_err("identity").kind,
+            ErrorKind::Unsupported
+        );
+    }
+    // The refusals changed nothing: the connection still works.
+    l.write(conn, WriteBuf::Owned(b"still here".to_vec()), Token(1))
+        .expect("write");
+    l.read(client, ReadBuf::Pooled, Token(2)).expect("read");
+    let mut out = Completions::default();
+    let until = l.now() + Duration::from_secs(5);
+    let mut read = 0;
+    while read == 0 {
+        assert!(l.now() < until, "refused handoff broke the connection");
+        l.turn(Timeout::Until(until), &mut out).expect("turn");
+        for c in out.drain() {
+            if let OpResult::Read {
+                n,
+                lease: Some(data),
+            } = c.result
+            {
+                assert_eq!(&data.as_slice()[..n], b"still here");
+                read += 1;
+            }
+        }
+    }
+    close_all(&mut l, &[client, conn, server]);
+}

crates/turnloop-contract/src/lib.rs

@@ -1608,6 +1608,7 @@ pub fn no_spin<B: Backend>() {
 
 #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
 pub mod filesystem;
+pub mod handoff;
 pub mod native_surface;
 pub mod sockopts;
 

crates/turnloop-contract/tests/allocations.rs

@@ -519,6 +519,80 @@ fn steady_socket_options_allocate_nothing() {
     turnloop_contract::sockopts::close_all(&mut l, &[client, conn, server, udp]);
 }
 
+/// Handing a transport to the host and taking it back allocates nothing: the
+/// identity is read out of the backend's own table, `detach` only unregisters,
+/// and the conversion to an owned descriptor moves the resource it already held.
+#[cfg(any(
+    target_vendor = "apple",
+    target_os = "linux",
+    target_os = "android",
+    target_os = "freebsd",
+    windows
+))]
+#[test]
+fn steady_handoff_allocates_nothing() {
+    let mut l = Loop::new(Config::default()).expect("loop");
+    let (server, client, conn) =
+        turnloop_contract::sockopts::plain_pair(&mut l, &ListenOpts::default());
+    let mut h = conn;
+    let mut total = 0;
+    let mut cycles = 0;
+    for i in 0..101u32 {
+        ALLOCS.with(|n| n.set(0));
+        ACTIVE.with(|v| v.set(i != 0));
+        let reported = l.raw_transport(h).expect("identity");
+        let transport = l.detach(h).expect("handoff");
+        #[cfg(unix)]
+        let (owned, back) = {
+            use std::os::fd::AsRawFd;
+            let fd = transport.into_fd();
+            let owned = RawTransport::Fd(fd.as_raw_fd());
+            (owned, Detached::from_fd(fd).expect("re-adopt"))
+        };
+        #[cfg(windows)]
+        let (owned, back) = {
+            use std::os::windows::io::AsRawSocket;
+            let socket = transport.into_socket().expect("socket transport");
+            let owned = RawTransport::Socket(socket.as_raw_socket() as usize);
+            (owned, Detached::from_socket(socket).expect("re-adopt"))
+        };
+        h = l.attach(back, Token(0)).expect("attach");
+        ACTIVE.with(|v| v.set(false));
+        assert_eq!(
+            owned, reported,
+            "the host received a different transport from the reported one"
+        );
+        if i != 0 {
+            cycles += 1;
+            total += ALLOCS.with(|n| n.get());
+        }
+    }
+    assert_eq!(cycles, 100, "the handoff subject ran");
+    assert_eq!(total, 0, "steady handoff allocations");
+    // The socket survived a hundred round trips: the cycle was not vacuous.
+    let mut out = Completions::default();
+    l.write(h, WriteBuf::Owned(b"round trip".to_vec()), Token(1))
+        .expect("write");
+    l.read(client, ReadBuf::Pooled, Token(2)).expect("read");
+    let until = l.now() + Duration::from_secs(5);
+    let mut read = 0;
+    while read == 0 {
+        assert!(l.now() < until, "the re-attached socket stopped working");
+        l.turn(Timeout::Until(until), &mut out).expect("turn");
+        for c in out.drain() {
+            if let OpResult::Read {
+                n,
+                lease: Some(data),
+            } = c.result
+            {
+                assert_eq!(&data.as_slice()[..n], b"round trip");
+                read += 1;
+            }
+        }
+    }
+    turnloop_contract::sockopts::close_all(&mut l, &[client, h, server]);
+}
+
 #[test]
 fn steady_deadline_poll_allocate_nothing() {
     let mut l = Loop::new(Config::default()).expect("loop");