Page the slot tables so a capacity ceiling costs nothing idle

DESIGN.md

@@ -552,7 +552,7 @@ No other Perry change is needed for P0.
 
 **Hard rules**, enforced by tests and CI benchmarks:
 
-1. **Allocations:** zero heap allocations per read, write, timer or accept after warm-up (checked with a counting allocator in tests).
+1. **Allocations:** zero heap allocations per read, write, timer or accept after warm-up (checked with a counting allocator in tests). *Warm-up* includes reaching a high-water mark: `Config::max_handles` and `Config::max_operations` are ceilings whose slot storage is built in pages as the mark rises (§10a), so passing a new mark allocates a page and a loop at its mark allocates nothing. **Contract tests:** a loop built with a 1024x larger ceiling makes the same allocations, and requests the same bytes, as the smaller one; a loop pinned at a high-water mark 500 handles above page zero turns 1000 times without allocating.
 2. **Wake:** zero syscalls on `notify()` while the loop is running (checked with a syscall-counting harness: `strace -c` / `ktrace` / ETW in CI smoke tests).
 3. **OS waits and discovery** (amended by tl-i01b, see the rationale below):
    1. A `turn` makes at most one OS wait.
@@ -566,6 +566,44 @@ No other Perry change is needed for P0.
 4a. **No spin.** A turn with nothing ready and a future deadline blocks until that deadline, at the precision in §7.6. It never returns immediately and never degrades into a zero-timeout poll loop. Hosts must pass exact deadlines (`Instant`, not truncated milliseconds). The Linux A/B measured what happens otherwise: 37,607 turns for 50 timers, ~70× user instructions. A backend whose timeout is implemented by a private wakeup source — epoll's timerfd, the IOCP deadline packet, the WASI 0.2 deadline pollable, the WASI 0.3 deadline subtask — reports that wake as a **zero-event** wait, exactly as a timed OS wait that returned nothing; real I/O or notifier events arriving in the same call still make it non-empty. **Contract tests:** with an idle registered socket and a 0.5 ms / 2 ms / 10 ms timer, turns per expiry ≤ 2 and zero-event OS waits ≤ 1 per expiry, on every backend; and (tl-i02) each of the sixty expiries at those delays costs exactly one blocking wait that observed no native event, identically with the loop idle and with a registered-but-idle native operation, with no allocation, while a call that carries real bytes is not counted empty.
 5. **Instruction budgets per operation** (Linux, cgu=1, `perf stat -e instructions:u,instructions:k`): TCP read / write / accept, timer start + cancel, notify + turn round trip, blocking job round trip, idle turn. **Values to be set from the attribution run of today's tokio bridge**, with a target below the tokio-bridge cost and within X % of a hand-written epoll loop. The CI gate compares against a committed baseline, with a control probe that must not move.
 
+### 10a. Capacities are ceilings, not reservations
+
+`max_handles` and `max_operations` name the largest number of slots a loop may
+ever hold. They are not preallocations: slots live in fixed-size pages built on
+demand, page zero with the loop and another when the high-water mark crosses into
+it. An idle loop therefore costs one page of each structure whatever its ceiling,
+which is what makes a large default ceiling and a loop per agent (§5a) affordable
+at once — the alternative forced hosts to choose between refusing connections and
+paying for a loop they mostly do not use.
+
+Substituting pages for a flat vector is safe because of how slots are addressed.
+A slot is named by its index and reused under a generation, so **adding a page
+moves no existing slot and invalidates no handle or operation id**; pages are
+separately allocated, so a slot's *address* is stable too, which the backends that
+hand a slot's address to the kernel rely on. Indices are handed out
+lowest-free-first, so the materialised prefix tracks the loop's high-water mark,
+not its ceiling.
+
+Two things stay contiguous and sized by their ceiling. The Windows `OVERLAPPED`
+slab (`kernel`, and the `bridges` parallel to it) maps a completion packet's
+pointer back to an operation index by pointer arithmetic over one allocation;
+paging it needs a different reverse map, which is its own change. The cross-thread
+result rings are lock-free and index by a power-of-two mask, where the capacity is
+also the backpressure bound. `pooled_buffers` is a separate question, tracked in
+#43: it is the remaining per-loop cost that scales with configuration.
+
+**Reaching the ceiling is backpressure.** An operation whose completion creates a
+handle — an accept, a handle receive — reserves its handle slot when it is
+submitted. An accept the kernel has been asked to perform therefore always has
+somewhere to put its connection, and at the ceiling the *submission* is refused
+with `ResourceLimit` instead, before the kernel is asked. The pending connection
+stays in the listener's backlog, which is the queue meant to absorb it. This
+replaces accepting a connection and then destroying it for want of a slot, which
+is what a host at its ceiling did before, and is why it surfaced as a connection
+refusal rather than a delay. A multishot accept holds one such reservation, so it
+is protected for one connection at a time; bounding a whole batch needs a per-turn
+native event budget on `Backend::poll`, which is a separate change.
+
 **Rule 3 rationale (tl-i01b, spec-owner decision, 2026-09-15).** The original rule 3 ("none when completions are already queued") forbade even a zero-timeout poll. Backend revision 2 has no separate no-wait discovery primitive: `poll(Duration::ZERO)` is the only way to learn about fresh readiness or completions, and on Unix, after cached readiness reaches `EAGAIN`, only the poller marks a resource ready again. libuv and Node make the same trade: `uv_run` computes `uv_backend_timeout()`, which is zero while pending, idle or closing work exists, and still calls `uv__io_poll` with that zero timeout (see the libuv [loop API](https://docs.libuv.org/en/v1.x/loop.html) and [`src/unix/core.c`](https://github.com/libuv/libuv/blob/v1.x/src/unix/core.c)). The [tl-i01 probes](docs/lanes/tl-i01.md#evidence--specification-decision) showed that skipping discovery breaks the unchanged fairness contract: skipping the native step whenever work was queued failed the timer/I/O/post fairness test after two seconds, and a guard that only drained cached work delivered 64 posts with zero reads and failed the same fairness test. A separate no-wait collection API on all six backends is not justified before measurement. Rule 4a and its raw zero-event accounting are unchanged: an empty discovery poll still counts toward the same no-spin bound.
 
 **Methodology** (lessons already paid for):

crates/turnloop-contract/src/lib.rs

@@ -219,6 +219,18 @@ mod native {
         capacity_and_stale_ids::<B>();
     }
     #[test]
+    fn growth_preserves_handles() {
+        paged_growth_preserves_handles::<B>();
+    }
+    #[test]
+    fn accept_reserves_a_handle_slot() {
+        accept_reserves_its_handle_slot::<B>();
+    }
+    #[test]
+    fn an_armed_accept_keeps_its_reserved_slot() {
+        an_armed_accept_keeps_its_slot::<B>();
+    }
+    #[test]
     fn cancellation_close() {
         cancel_close_ordering::<B>();
     }
@@ -894,6 +906,273 @@ pub fn capacity_and_stale_ids<B: Backend>() {
     assert_eq!(count, 3);
 }
 
+/// A key taken before the handle table grew still names its own resource after.
+///
+/// `max_handles` is a ceiling served by pages built on demand, so a loop crosses
+/// page boundaries as its high-water mark rises. Growth must leave every live
+/// handle addressing what it addressed: an index means the same slot before and
+/// after, and no live slot is ever handed out twice.
+pub fn paged_growth_preserves_handles<B: Backend>() {
+    // A ceiling many pages wide, so reaching the high-water mark below grows the
+    // table repeatedly while every handle taken from an earlier page is live.
+    let config = Config {
+        max_handles: 4096,
+        max_operations: 4096,
+        events_per_turn: 64,
+        ..Config::default()
+    };
+    let mut l = Driver::<B>::new(config).expect("loop");
+    let at = l.now() + Duration::from_secs(30);
+    const LIVE: u64 = 600;
+    let mut handles = Vec::new();
+    for i in 0..LIVE {
+        handles.push(l.timer(at, None, Token(i)).expect("within the ceiling"));
+    }
+    let mut keys: Vec<u64> = handles.iter().map(|h| h.key()).collect();
+    keys.sort_unstable();
+    keys.dedup();
+    assert_eq!(
+        keys.len(),
+        handles.len(),
+        "growth handed out a slot that was still live"
+    );
+    let mut indices: Vec<usize> = handles.iter().map(|h| h.index()).collect();
+    indices.sort_unstable();
+    indices.dedup();
+    assert_eq!(
+        indices.len(),
+        handles.len(),
+        "two live handles share a slot"
+    );
+    // Slots are handed out lowest-free-first, so holding LIVE of them at once
+    // means the table really did grow through every page boundary below it —
+    // without this the rest of the assertions would pass on an ungrown table.
+    assert_eq!(
+        (indices[0], *indices.last().expect("handles")),
+        (0, LIVE as usize - 1),
+        "the high-water mark did not reach the slot count under test"
+    );
+    // Every handle still resolves, the first page's included, and each still
+    // names the timer it was created with rather than a neighbour's.
+    for (i, h) in handles.iter().enumerate() {
+        let op = l
+            .timer_op(*h)
+            .expect("a handle taken before growth still names its timer");
+        assert_eq!(op.owner(), h.owner());
+        l.set_ref(*h, true)
+            .expect("a live handle is still addressable");
+        assert_eq!(h.index(), indices[i], "a handle changed slot across growth");
+    }
+    // Closing every handle delivers exactly one Closed per handle, carrying the
+    // token that handle was closed with: growth aliased no slot onto another.
+    let mut out = Completions::with_capacity(64);
+    let mut closed = vec![false; LIVE as usize];
+    for h in &handles {
+        l.close(*h, Token(h.index() as u64)).expect("close");
+    }
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    while closed.iter().any(|done| !done) {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "Closed never delivered"
+        );
+        l.turn(Timeout::After(Duration::from_millis(50)), &mut out)
+            .expect("drain");
+        for c in out.drain() {
+            if matches!(c.result, OpResult::Closed) {
+                let slot = c.token.0 as usize;
+                assert!(!closed[slot], "slot {slot} delivered Closed twice");
+                closed[slot] = true;
+            }
+        }
+    }
+    // A stale key never reaches the slot it used to name, even after reuse.
+    let reused = l.timer(at, None, Token(9999)).expect("freed slots return");
+    assert!(
+        handles.iter().all(|h| h.key() != reused.key()),
+        "a retired key was handed out again"
+    );
+    assert!(
+        l.set_ref(handles[0], true).is_err(),
+        "stale handle rejected"
+    );
+}
+
+/// A loop at its handle ceiling refuses to *ask* for a connection, rather than
+/// accepting one and destroying it.
+///
+/// An accept reserves its handle slot when it is submitted, so an accept the
+/// kernel has been asked to perform always has somewhere to put its connection.
+/// At the ceiling the submission is refused instead, and the connection stays in
+/// the listener's backlog: the accept queue is the queue that absorbs it.
+pub fn accept_reserves_its_handle_slot<B: Backend>() {
+    // Two handles: the listener, and room for exactly one accepted connection.
+    let mut l = Driver::<B>::new(Config {
+        max_handles: 2,
+        max_operations: 8,
+        ..Config::default()
+    })
+    .expect("loop");
+    let listener = l
+        .tcp_listen("127.0.0.1:0".parse().expect("addr"), &ListenOpts::default())
+        .expect("listen");
+    let addr = l.local_addr(listener).expect("addr");
+    // The reserve is taken at submission, so a second concurrent accept is
+    // refused before it reaches the kernel even though nothing has connected.
+    let first = l.accept(listener, Token(1)).expect("one slot is free");
+    assert!(
+        matches!(
+            l.accept(listener, Token(2)),
+            Err(Error {
+                kind: ErrorKind::ResourceLimit,
+                ..
+            })
+        ),
+        "a second accept has no slot to reserve and must not reach the kernel"
+    );
+    // The reserved slot is honoured: the accept that was armed delivers.
+    let _client = std::net::TcpStream::connect(addr).expect("client");
+    let mut out = Completions::with_capacity(8);
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    let mut conn = None;
+    while conn.is_none() {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "accept never delivered"
+        );
+        l.turn(Timeout::After(Duration::from_millis(50)), &mut out)
+            .expect("turn");
+        for c in out.drain() {
+            match c.result {
+                OpResult::Accepted { conn: h, .. } => conn = Some(h),
+                other => panic!("unexpected {other:?}"),
+            }
+        }
+    }
+    let conn = conn.expect("accepted");
+    assert_eq!(l.timer_op(conn), None, "an accepted socket is not a timer");
+    let _ = first;
+    // Now at the ceiling: no slot to reserve, so no accept is armed at all.
+    assert!(
+        matches!(
+            l.accept(listener, Token(3)),
+            Err(Error {
+                kind: ErrorKind::ResourceLimit,
+                ..
+            })
+        ),
+        "at the ceiling an accept is refused before the kernel is asked"
+    );
+    // Traffic on handles that already exist is not gated by the reservation: a
+    // loop at its ceiling still serves the connections it has.
+    static HELLO: [u8; 5] = *b"hello";
+    // SAFETY: HELLO is static immutable memory, valid through the completion.
+    let buf = unsafe { IoBuf::from_raw_parts(HELLO.as_ptr(), HELLO.len()) };
+    l.write(conn, WriteBuf::Provided(buf), Token(6))
+        .expect("a loop at its ceiling still serves its open connections");
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    let mut wrote = 0;
+    while wrote == 0 {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "write never completed"
+        );
+        l.turn(Timeout::After(Duration::from_millis(50)), &mut out)
+            .expect("turn");
+        for c in out.drain() {
+            if let OpResult::Wrote(n) = c.result {
+                wrote = n;
+            }
+        }
+    }
+    assert_eq!(wrote, HELLO.len());
+    // Releasing the connection returns its slot, and accepts arm again.
+    l.close(conn, Token(4)).expect("close");
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    let mut released = false;
+    while !released {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "close never delivered"
+        );
+        l.turn(Timeout::After(Duration::from_millis(50)), &mut out)
+            .expect("turn");
+        for c in out.drain() {
+            released |= matches!(c.result, OpResult::Closed);
+        }
+    }
+    l.accept(listener, Token(5))
+        .expect("a freed slot makes the loop acceptive again");
+}
+
+/// A slot promised to an armed accept cannot be spent by anything else.
+///
+/// The reservation is only worth something if the loop refuses to hand that slot
+/// to a handle the host creates directly. Without that, a host can fill its
+/// ceiling while an accept is in flight, and the accept arrives with nowhere to
+/// put its connection — which is the failure this whole mechanism exists to
+/// remove, because the connection is destroyed rather than deferred.
+pub fn an_armed_accept_keeps_its_slot<B: Backend>() {
+    // Three slots: the listener, one held for the armed accept, one to spend.
+    let mut l = Driver::<B>::new(Config {
+        max_handles: 3,
+        max_operations: 16,
+        ..Config::default()
+    })
+    .expect("loop");
+    let listener = l
+        .tcp_listen("127.0.0.1:0".parse().expect("addr"), &ListenOpts::default())
+        .expect("listen");
+    let addr = l.local_addr(listener).expect("addr");
+    l.accept(listener, Token(1)).expect("one slot to reserve");
+    let at = l.now() + Duration::from_secs(30);
+    let spare = l.timer(at, None, Token(2)).expect("the unreserved slot");
+    assert!(
+        matches!(
+            l.timer(at, None, Token(3)),
+            Err(Error {
+                kind: ErrorKind::ResourceLimit,
+                ..
+            })
+        ),
+        "the armed accept's slot must not be available to a new handle"
+    );
+    // And the accept therefore still has somewhere to put its connection.
+    let _client = std::net::TcpStream::connect(addr).expect("client");
+    let mut out = Completions::with_capacity(8);
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    let mut conn = None;
+    while conn.is_none() {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "accept never delivered"
+        );
+        l.turn(Timeout::After(Duration::from_millis(50)), &mut out)
+            .expect("turn");
+        for c in out.drain() {
+            match c.result {
+                OpResult::Accepted { conn: h, .. } => conn = Some(h),
+                OpResult::Err(e) => panic!("the reserved slot was spent elsewhere: {e:?}"),
+                other => panic!("unexpected {other:?}"),
+            }
+        }
+    }
+    let conn = conn.expect("accepted");
+    // Every slot is now live, and the reservation is gone with the accept.
+    assert!(
+        matches!(
+            l.timer(at, None, Token(4)),
+            Err(Error {
+                kind: ErrorKind::ResourceLimit,
+                ..
+            })
+        ),
+        "the loop is at its ceiling"
+    );
+    l.close(conn, Token(5)).expect("close");
+    l.close(spare, Token(6)).expect("close");
+}
+
 pub fn pooled_lease_backpressure<B: Backend>() {
     let mut l = Driver::<B>::new(Config {
         pooled_buffers: 1,