From b2cc0918bc25689b322cbe3da094c93b1ccb8997 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 23 Aug 2026 01:15:48 -0400 Subject: [PATCH] Home relay failure is a redial, not endpoint death MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A home-relay wire failure permanently bricked the endpoint (issue #88): the pump treated a receive or send error on the home relay as endpoint death, so any transient loss of the websocket — a mobile browser backgrounding the page, a network blip, a relay restart — left the endpoint refusing every operation, and the application had to rebuild it from scratch. Upstream iroh reconnects the relay with backoff and never kills the endpoint over it; the death ruling was an accumulated divergence with no artifact. The pump now retires the failed home conn and redials it: jittered exponential backoff from 10ms, unbounded attempts, immediate-and-reset after a connection that had lived >= 10s (upstream's established rule, approximated by uptime). The cap is 5s where upstream uses 16s — recovery latency after a browser page returns to foreground dominates this design's use, and the divergence is recorded on REDIAL_MAX_DELAY. Transmits to the home relay while down are dropped; QUIC loss recovery owns them. Reconnect reoccupies pool key HOME_RELAY, so routes carry over unchanged. Stale futures from a replaced conn are gated by Rc::ptr_eq against the pool's current entry: their datagrams are still handled, but they neither re-arm against the successor nor trigger a spurious redial. A retired conn is closed at retirement so its pending receive resolves and the teardown drain terminates. The redial dial is deliberately unbounded: settling a select race by dropping an in-flight import inside a live task wedges the polyengine host, so a dial timeout is not expressible here — the constraint and its consequence are recorded on the pump and redial_home docs. With no caller left, the endpoint-death machinery (dead, mark_dead) is gone. Same decision, adjacent defect: a dead foreign relay now retires its relay_keys URL mapping along with its pool entry, so a later dial through that relay reconnects instead of returning a key whose transmits blackhole. The deltic exam gains scenario 7, relay outage: dial and echo over an owned relay, stop the relay, hold 8s of outage, restart it, and require the same connection to survive and echo (the outage plus the 5s backoff cap stays inside the 30s idle window), then a fresh dial on the same endpoints — the anti-brick assertion — and finally a close with redials in flight, which must resolve without a trap. Blocked, not failed, when the relay was adopted rather than spawned. Falsified against the unfixed endpoint: the connections read closed after the hold and the scenario fails. Gates: just check, just build, just exam-polyengine (8/8), just matrix (13/13 pairings). Fixes #88 --- endpoint/src/endpoint_impl.rs | 331 +++++++++++++++++++++------- endpoint/src/webrtc.rs | 2 +- host-polyengine/README.md | 21 +- host-polyengine/src/repo.ts | 12 +- host-polyengine/src/run-endpoint.ts | 237 +++++++++++++++++++- justfile | 5 +- 6 files changed, 508 insertions(+), 100 deletions(-) diff --git a/endpoint/src/endpoint_impl.rs b/endpoint/src/endpoint_impl.rs index 059572b..796233b 100644 --- a/endpoint/src/endpoint_impl.rs +++ b/endpoint/src/endpoint_impl.rs @@ -159,8 +159,11 @@ pub(crate) struct State { /// The relay pool: the home relay (key `HOME_RELAY`) plus foreign /// relays opened for dialing or signaling, keyed by their /// normalized URL in `relay_keys`. A dead foreign relay leaves the - /// pool; routes naming it drop transmits until their connections - /// idle out. + /// pool and its URL mapping ([`State::retire_relay`]), so the next + /// dial through that URL opens a fresh connection; routes naming + /// the retired key drop transmits until their connections idle out. + /// The home relay's entry is replaced in place by a redial and its + /// URL mapping never leaves. relay_pool: HashMap>, relay_keys: HashMap, /// URLs an in-flight `connect` is currently opening; a second @@ -190,9 +193,6 @@ pub(crate) struct State { next_channel_id: u32, closed: bool, closed_at: Option, - /// Set when the relay connection died; every operation fails from - /// then on. - dead: Option, /// Wakers parked by `wait_until` futures, drained and fired by /// `wake_waiters`. waiters: Vec, @@ -210,10 +210,31 @@ enum RouteWire { Channel(u32), } -/// The home relay's pool key; its death kills the endpoint, where a -/// foreign relay's death only starves the routes that named it. +/// The home relay's pool key. A wire failure there is redialed by the +/// pump under [`redial_delay`]'s backoff and the key is reoccupied, +/// where a foreign relay's failure retires the relay and starves the +/// routes that named it until a later dial reopens it. const HOME_RELAY: u32 = 0; +/// The first delay before a home-relay redial, doubled at every failed +/// attempt up to [`REDIAL_MAX_DELAY`]. +const REDIAL_MIN_DELAY: Duration = Duration::from_millis(10); + +/// The ceiling the redial backoff doubles up to; attempts are +/// unbounded. Upstream iroh's relay actor caps at 16s +/// (`src/socket/transports/relay/actor.rs`, `build_backoff`); this +/// endpoint deliberately narrows the cap, because the outage it must +/// recover from quickly is a browser page returning to the foreground — +/// recovery latency there is what the cap buys, and the long outages a +/// larger cap serves are not what this design optimizes for. +const REDIAL_MAX_DELAY: Duration = Duration::from_secs(5); + +/// How long a home-relay connection must have been registered for its +/// failure to count as a sound connection lost, rather than a relay +/// that cannot be reached: an older connection is redialed at once and +/// restarts the backoff, a younger one consumes the next delay. +const REDIAL_ESTABLISHED: Duration = Duration::from_secs(10); + struct ChannelEntry { wire: Rc, /// The relay-authenticated peer the channel was signaled with; @@ -310,7 +331,6 @@ impl State { next_channel_id: 0, closed: false, closed_at: None, - dead: None, waiters: Vec::new(), pump_waker: None, pump_kicked: false, @@ -370,8 +390,8 @@ impl State { /// True once no operation can succeed anymore; signaling sessions /// poll this to abandon their dance. - pub(crate) fn is_closed_or_dead(&self) -> bool { - self.closed || self.dead.is_some() + pub(crate) fn is_closed(&self) -> bool { + self.closed } /// Claim the signaling slot for `peer`, recording the relay the @@ -418,7 +438,7 @@ impl State { peer: [u8; 32], wire: Rc, ) -> Result<(), Error> { - if self.is_closed_or_dead() { + if self.closed { wire.close(); return Err(Error::Closed); } @@ -440,6 +460,16 @@ impl State { Ok(()) } + /// Retire a relay: it leaves the pool AND its URL mapping, so a + /// later `ensure_relay` for that URL opens a fresh connection + /// instead of handing out a key with no pool entry behind it. + /// Returns the retired connection, for the caller to close. + fn retire_relay(&mut self, key: u32) -> Option> { + let conn = self.relay_pool.remove(&key); + self.relay_keys.retain(|_, mapped| *mapped != key); + conn + } + /// Retire a dead channel; a peer routed over it moves back to the /// relay it was signaled through (its connections survive the move /// or idle out). @@ -455,15 +485,6 @@ impl State { } } - fn mark_dead(&mut self, reason: &str) { - self.dead = Some(reason.to_string()); - for entry in self.conns.values_mut() { - if entry.error.is_none() { - entry.error = Some(Error::Closed); - } - } - } - /// Drain endpoint-bound events, application events, and transmits /// until quiescent; returns whether anything progressed, so the /// pump wakes parked method futures exactly when the state they @@ -760,17 +781,38 @@ fn on_event( } /// The endpoint's I/O task: relayed datagrams in and out, noq's timers, -/// and the flush after every kick. The two long-lived import futures stay -/// pinned across iterations and are resolved before the task returns (an -/// in-flight import is a component-model subtask; jco traps on cancelling -/// one — the teardown discipline). Channel receives live in a -/// persistent set with the same discipline: closing every channel -/// resolves them before the task returns. -async fn pump(shared: Shared, udp: Option>) { +/// the flush after every kick, and the home relay's redial. The two +/// long-lived import futures stay pinned across iterations and are +/// resolved before the task returns (an in-flight import is a +/// component-model subtask; jco traps on cancelling one — the teardown +/// discipline). Channel receives live in a persistent set with the same +/// discipline: closing every channel resolves them before the task +/// returns. The redial slot is the exception: an in-flight relay dial is +/// dropped at teardown rather than awaited. +/// +/// The two kinds of drop are not the same hazard. Dropping an in-flight +/// import to settle a select race INSIDE a live task wedges the host, +/// which sees an activation that never parks, finishes, or traps (see +/// [`redial_home`]). Dropping one as the whole task ends is the +/// host-driven cancellation `ensure_relay` already takes when its call +/// is cancelled, and is sound. +/// +/// A failed home relay retires its connection from the pool and arms a +/// redial; the pump ends only on close, never on a wire failure. +/// `home_url` and `identity` are what a redial re-runs `RelayConn::connect` +/// with. +async fn pump(shared: Shared, udp: Option>, home_url: String, identity: Rc) { let mut udp_recv = pin!(udp_receive(udp.clone()).fuse()); let mut tick = pin!(monotonic_clock::wait_for(TICK_NS).fuse()); let mut channel_recvs: FuturesUnordered = FuturesUnordered::new(); let mut relay_recvs: FuturesUnordered = FuturesUnordered::new(); + // The home relay's redial, armed only while the pool has no home + // entry; terminated is the healthy state. + let mut redial: futures::future::Fuse = futures::future::Fuse::terminated(); + let mut redial_attempt: u32 = 0; + // When the pool's home entry was registered: `bind` connected it + // immediately before spawning this task. + let mut home_since: Option = Some(Instant::now()); // Set by the tick arm: wake the waiters even without drain progress, // so deadline conditions are re-checked at tick granularity. let mut force_wake = false; @@ -788,14 +830,21 @@ async fn pump(shared: Shared, udp: Option>) { match item { Some((key, peer, datagram)) => { let conn = shared.borrow().relay_pool.get(&key).cloned(); - // A retired relay's queued transmits are lost, as is - // a failed foreign send; the home relay's failure is - // the endpoint's death. + // The datagram is lost either way — a retired + // relay has no wire and a failed send delivered + // nothing — and QUIC's loss recovery owns the + // retransmit. A home-relay send failure additionally + // retires the connection and arms the redial, unless + // the pool has already moved on from it. if let Some(conn) = conn { - if conn.send_datagram(&peer, &datagram).await.is_err() && key == HOME_RELAY + if conn.send_datagram(&peer, &datagram).await.is_err() + && key == HOME_RELAY + && retire_home(&shared, &conn) { - shared.borrow_mut().mark_dead("relay send failed"); - break 'pump; + let delay = next_redial_delay(&mut redial_attempt, &mut home_since); + redial = redial_home(home_url.clone(), identity.clone(), delay) + .boxed_local() + .fuse(); } } } @@ -852,9 +901,6 @@ async fn pump(shared: Shared, udp: Option>) { { let st = shared.borrow(); - if st.dead.is_some() { - break 'pump; - } if st.closed { let all_drained = st.conns.values().all(|e| e.drained || e.error.is_some()); let lingered = st @@ -869,13 +915,21 @@ async fn pump(shared: Shared, udp: Option>) { select_biased! { event = next_relay_event(&mut relay_recvs).fuse() => { - let (key, result) = event; + let (key, conn, result) = event; + // Whether the connection this receive read is still the + // pool's: a redial replaces the home entry under the + // same key, so a future armed against the OLD connection + // may resolve afterwards. + let current = is_pool_current(&shared, key, &conn); match result { Ok(datagram) => { // Re-arm before handling so the wire keeps - // flowing; a retired relay stays retired. - let conn = shared.borrow().relay_pool.get(&key).cloned(); - if let Some(conn) = conn { + // flowing. Only the pool's current connection is + // re-armed — a second receive on a replaced one + // would double up on its successor's key — while + // the datagram itself is genuine, relay-source- + // authenticated data and is handled either way. + if current { relay_recvs.push(Box::pin(relay_receive(key, conn))); } if datagram.payload.first() == Some(&SIGNAL_PREFIX) { @@ -886,14 +940,26 @@ async fn pump(shared: Shared, udp: Option>) { .handle_relay_datagram(key, datagram.source, datagram.payload); } } - Err(err) => { - if key == HOME_RELAY { - shared.borrow_mut().mark_dead(&err); - break 'pump; + // A failure on a connection the pool has already + // moved on from is stale: the replacement is live, + // or a redial is already in flight. + Err(_) if !current => {} + Err(_) if key == HOME_RELAY => { + if retire_home(&shared, &conn) { + let delay = next_redial_delay(&mut redial_attempt, &mut home_since); + redial = redial_home(home_url.clone(), identity.clone(), delay) + .boxed_local() + .fuse(); } + } + Err(_) => { // A foreign relay died; routes naming it starve - // and their connections idle out. - shared.borrow_mut().relay_pool.remove(&key); + // and their connections idle out, and the URL + // mapping goes with the pool entry so a later + // dial through it reconnects. + if let Some(conn) = shared.borrow_mut().retire_relay(key) { + conn.close(); + } } } }, @@ -935,6 +1001,37 @@ async fn pump(shared: Shared, udp: Option>) { } } }, + opened = redial => { + match opened { + Ok(conn) => { + // The home URL already maps to HOME_RELAY in + // `relay_keys`; reoccupying the pool key is the + // whole reconnection, and every route naming it + // resumes unchanged. + let conn = Rc::new(conn); + let mut st = shared.borrow_mut(); + if st.closed { + drop(st); + conn.close(); + } else { + st.relay_pool.insert(HOME_RELAY, conn.clone()); + st.wake_waiters(); + drop(st); + relay_recvs.push(Box::pin(relay_receive(HOME_RELAY, conn))); + home_since = Some(Instant::now()); + redial_attempt = 0; + } + } + Err(_) => { + if !shared.borrow().closed { + let delay = next_redial_delay(&mut redial_attempt, &mut home_since); + redial = redial_home(home_url.clone(), identity.clone(), delay) + .boxed_local() + .fuse(); + } + } + } + }, _ = kicked(&shared).fuse() => { // A method mutated state needing a flush; the loop top // drains and transmits it. @@ -947,8 +1044,8 @@ async fn pump(shared: Shared, udp: Option>) { } } - // No waiter sleeps past the pump: every break path set its terminal - // state (dead, or closed and drained/lingered) before arriving here. + // No waiter sleeps past the pump: the only break path set its + // terminal state (closed and drained/lingered) before arriving here. shared.borrow_mut().wake_waiters(); // Resolve the pinned imports before the task ends: close every pool @@ -986,35 +1083,27 @@ async fn pump(shared: Shared, udp: Option>) { } } -type RelayRecvFuture = futures::future::LocalBoxFuture< - 'static, - ( - u32, - Result, - ), ->; - -/// One relay receive, tagged with the relay's pool key. -async fn relay_receive( - key: u32, - conn: Rc, -) -> ( +type RelayRecvResult = ( u32, + Rc, Result, -) { +); + +type RelayRecvFuture = futures::future::LocalBoxFuture<'static, RelayRecvResult>; + +/// One relay receive, tagged with the relay's pool key AND the +/// connection it read: a redial replaces the pool's home entry under +/// the same key, so the key alone does not say which connection a +/// completed receive speaks for. +async fn relay_receive(key: u32, conn: Rc) -> RelayRecvResult { let result = conn.recv_datagram().await; - (key, result) + (key, conn, result) } /// The next completed relay receive, or pending-forever while the pool -/// is empty (only during teardown; the home relay is armed before the -/// first select). The set owns the in-flight import futures. -async fn next_relay_event( - set: &mut FuturesUnordered, -) -> ( - u32, - Result, -) { +/// is empty (during teardown, or while the home relay is being +/// redialed). The set owns the in-flight import futures. +async fn next_relay_event(set: &mut FuturesUnordered) -> RelayRecvResult { if set.is_empty() { std::future::pending().await } else { @@ -1022,6 +1111,89 @@ async fn next_relay_event( } } +/// Whether `conn` is still the connection the pool holds under `key`. +fn is_pool_current(shared: &Shared, key: u32, conn: &Rc) -> bool { + shared + .borrow() + .relay_pool + .get(&key) + .is_some_and(|pooled| Rc::ptr_eq(pooled, conn)) +} + +/// Take the failed home relay out of the pool and close it, so its +/// pending receive resolves and leaves the pump's set. Its `relay_keys` +/// entry stays: a redial reoccupies `HOME_RELAY`. Returns whether this +/// call is the one that retired it, i.e. whether the caller owns arming +/// the redial. +fn retire_home(shared: &Shared, conn: &Rc) -> bool { + if !is_pool_current(shared, HOME_RELAY, conn) { + return false; + } + shared.borrow_mut().relay_pool.remove(&HOME_RELAY); + conn.close(); + true +} + +/// The delay before the redial attempt numbered `attempt` (zero-based): +/// `REDIAL_MIN_DELAY` doubled per attempt, capped at +/// `REDIAL_MAX_DELAY`, plus up to as much again in jitter (also capped), +/// so endpoints that lost one relay together do not redial in lockstep. +/// A random source that refuses yields the unjittered delay. +fn redial_delay(attempt: u32) -> Duration { + let base = REDIAL_MIN_DELAY + .checked_mul(1u32.checked_shl(attempt.min(31)).unwrap_or(u32::MAX)) + .unwrap_or(REDIAL_MAX_DELAY) + .min(REDIAL_MAX_DELAY); + let mut bytes = [0u8; 2]; + let spread = match getrandom::fill(&mut bytes) { + Ok(()) => u16::from_le_bytes(bytes) as u64, + Err(_) => 0, + }; + let jitter = Duration::from_nanos(base.as_nanos() as u64 / u64::from(u16::MAX) * spread); + (base + jitter).min(REDIAL_MAX_DELAY) +} + +/// The delay the next redial waits, consuming the backoff state. A home +/// relay that had been registered at least `REDIAL_ESTABLISHED` before +/// it failed is redialed at once and restarts the backoff; anything +/// younger takes the next delay. +fn next_redial_delay(attempt: &mut u32, home_since: &mut Option) -> Duration { + if home_since + .take() + .is_some_and(|at| at.elapsed() >= REDIAL_ESTABLISHED) + { + *attempt = 0; + return Duration::ZERO; + } + let delay = redial_delay(*attempt); + *attempt = attempt.saturating_add(1); + delay +} + +type RedialFuture = futures::future::LocalBoxFuture<'static, Result>; + +/// One home-relay redial: wait out the backoff, then connect and +/// authenticate. +/// +/// The dial carries no timeout of its own, and must not: a timeout is a +/// second future racing the dial, so every attempt drops whichever of +/// the two loses — the in-task select-race drop [`pump`]'s doc rules +/// out. Dropping the timer that way wedges the polyengine host, which +/// reports it as a resumed activation whose claim is never released. A +/// relay that accepts the socket and then stalls the handshake +/// therefore holds this slot until the socket resolves; nothing else in +/// the endpoint waits on it. +async fn redial_home( + url: String, + identity: Rc, + delay: Duration, +) -> Result { + if !delay.is_zero() { + monotonic_clock::wait_for(delay.as_nanos() as u64).await; + } + RelayConn::connect(&url, &identity).await +} + type ChannelRecvFuture = futures::future::LocalBoxFuture<'static, (u32, Result>, String>)>; @@ -1053,7 +1225,7 @@ fn handle_signal(shared: &Shared, via: u32, source: [u8; 32], payload: &[u8]) { const INBOX_CAP: usize = 64; let spawn_answerer = { let mut st = shared.borrow_mut(); - if !st.webrtc_enabled || st.is_closed_or_dead() { + if !st.webrtc_enabled || st.closed { return; } // Claim the slot synchronously with the decision, so a second @@ -1360,7 +1532,7 @@ impl EndpointRes { if let Some(key) = st.relay_keys.get(&normalized) { return Ok(*key); } - if st.is_closed_or_dead() { + if st.closed { return Err(Error::Closed); } st.relay_opening.insert(normalized.clone()) @@ -1394,7 +1566,7 @@ impl EndpointRes { "a concurrent open of this relay failed".into(), ))); } - if st.is_closed_or_dead() { + if st.closed { return Some(Err(Error::Closed)); } if started.elapsed() > Duration::from_secs(30) { @@ -1514,7 +1686,12 @@ impl GuestEndpoint for EndpointRes { shared .borrow_mut() .register_relay(&relay_url, Rc::new(relay)); - wit_bindgen::spawn_local(pump(shared.clone(), udp.clone())); + wit_bindgen::spawn_local(pump( + shared.clone(), + udp.clone(), + relay_url.clone(), + identity.clone(), + )); Ok(Endpoint::new(EndpointRes { shared, @@ -1589,7 +1766,7 @@ impl GuestEndpoint for EndpointRes { let (handle, epoch) = { let mut st = self.shared.borrow_mut(); - if st.dead.is_some() || st.closed { + if st.closed { return Err(Error::Closed); } let remote = match direct { @@ -1676,7 +1853,7 @@ impl GuestEndpoint for EndpointRes { return Some(Ok((handle, epoch))); } } - if st.dead.is_some() || st.closed { + if st.closed { return Some(Err(Error::Closed)); } None diff --git a/endpoint/src/webrtc.rs b/endpoint/src/webrtc.rs index d96d38f..a80b5f9 100644 --- a/endpoint/src/webrtc.rs +++ b/endpoint/src/webrtc.rs @@ -141,7 +141,7 @@ async fn next_signal( ) -> Result, Error> { loop { let payload = wait_until(shared, move |st| { - if st.is_closed_or_dead() { + if st.is_closed() { return Some(Err(Error::Closed)); } if let Some(payload) = st.pop_signal_inbox(peer) { diff --git a/host-polyengine/README.md b/host-polyengine/README.md index f91df20..3adc3fc 100644 --- a/host-polyengine/README.md +++ b/host-polyengine/README.md @@ -29,21 +29,12 @@ just exam-polyengine builds the endpoint component and the stock relay, installs the leg's pinned module graph + the `node-datachannel` addon -(`just polyengine-setup`, idempotent), fetches the sha256-pinned translator -release asset, and runs `src/run-endpoint.ts` — five scenarios: - -1. **bind + identity** — `identity-generate` → `new - EndpointOptions(identity)` → `Endpoint.bind`; the Ed25519 identity - minted through `polymorph:webcrypto`; three export calls against the - live detached pump (the lann/jco#11 shape). -2. **relay echo** — two endpoint instances, QUIC handshake and an - authenticated echo over a stock `iroh-relay --dev`. -3. **WebRTC upgrade** — a relay-dialed connection moves onto the data - channel; `connection.path` reports the move. -4. **concurrency proof points** — 40 export calls against two live - pumps (jco#11) and `accept` parked before the dial and woken by the - pump (jco#13): issue #10's rows as passing assertions. -5. **teardown** — idempotent close, no guest traps, the relay reaped. +(`just polyengine-setup`, idempotent), and runs the scenarios in +`src/run-endpoint.ts` — the endpoint lifecycle (bind + identity through +idempotent teardown), the relay and WebRTC wires, the issue #10 +concurrency rows, and the liveness/recovery rows (idle survival, relay +outage). Each scenario names its assertions where it lives; the exam's +summary line is the inventory. The exam retries the handshake-shaped scenarios a bounded number of times: `endpoint/src/endpoint_impl.rs`'s shared state has a RefCell diff --git a/host-polyengine/src/repo.ts b/host-polyengine/src/repo.ts index 850004f..03c8348 100644 --- a/host-polyengine/src/repo.ts +++ b/host-polyengine/src/repo.ts @@ -50,6 +50,12 @@ export async function endpointComponentBytes(): Promise { export interface Relay { readonly url: string; + /** + * Whether this run spawned the relay process. An adopted relay + * (`false`) is somebody else's: `stop()` does nothing, so a scenario + * that needs the relay to actually go away must skip. + */ + readonly owned: boolean; stop(): Promise; } @@ -67,12 +73,13 @@ async function portOpen(port: number): Promise { * Spawn `iroh-relay --dev` (ws on 127.0.0.1:3340) and wait for it to accept. * * If something is already listening on the port we adopt it rather than - * racing a second binder. + * racing a second binder. Each call spawns its own process, so a stopped + * relay is restarted by calling this again. */ export async function startRelay(): Promise { if (await portOpen(RELAY_PORT)) { console.error(`relay: adopting an already-listening 127.0.0.1:${RELAY_PORT}`); - return { url: RELAY_URL, stop: () => Promise.resolve() }; + return { url: RELAY_URL, owned: false, stop: () => Promise.resolve() }; } let child: Deno.ChildProcess; try { @@ -101,6 +108,7 @@ export async function startRelay(): Promise { console.error(`relay: iroh-relay --dev listening on ${RELAY_URL} (pid ${child.pid})`); return { url: RELAY_URL, + owned: true, stop: async () => { try { child.kill("SIGTERM"); diff --git a/host-polyengine/src/run-endpoint.ts b/host-polyengine/src/run-endpoint.ts index 6331f98..dd2cce1 100644 --- a/host-polyengine/src/run-endpoint.ts +++ b/host-polyengine/src/run-endpoint.ts @@ -117,6 +117,21 @@ const DATAGRAM_CEILING = 3900; // this hold. Real wall time — the exam has no virtual clock. const IDLE_HOLD_MS = 35_000; +// How long the relay-outage scenario (issue #88) keeps the relay down. +// Real wall time, and deliberately well inside noq's 30s idle timeout: +// the QUIC connection must SURVIVE the outage, so the hold plus the +// redial backoff's ceiling (endpoint_impl.rs `REDIAL_MAX_DELAY`) plus +// the handshake has to stay under it. +const OUTAGE_HOLD_MS = 8_000; + +// The budget for the post-outage echo: one redial's backoff ceiling plus +// the reconnect handshake plus the round trip, generously. +const OUTAGE_RECOVERY_MS = 20_000; + +// How long redials are left in flight before the endpoints are closed, +// so teardown happens with a relay dial pending. +const OUTAGE_TEARDOWN_MS = 2_000; + /** * Bounded retries around the RefCell borrow hazard (see the header). The * budget is per-shape because the shapes lose the race at very different @@ -470,11 +485,168 @@ async function echoRoundtrip(conn: Connection, sconn: Connection, what: string): return await deadline(readAll(crecv), 30_000, `${what}: client read echo`); } +// --- the relay-outage probe (issue #88) -------------------------------------- + +/** + * The relay lifetime the outage probe drives. `stop`/`start` are the + * harness's own relay process; `url` is stable across a restart (the + * relay always returns on `RELAY_PORT`). + */ +interface RelayControl { + url(): string; + stop(): Promise; + start(): Promise; +} + +interface OutageReport { + readonly prePath: PathKind; + readonly preEcho: string; + /** Whether the port really stopped accepting while the relay was down. */ + readonly wentDown: boolean; + readonly clientState: ConnectionState; + readonly serverState: ConnectionState; + readonly postEcho: string; + readonly recoveryMs: number; + /** The echo on a connection dialed AFTER the outage. */ + readonly freshEcho: string; + /** Guest traps raised while the endpoints closed with redials pending. */ + readonly teardownPanics: string[]; + /** Guest traps raised before teardown (the RefCell borrow hazard). */ + readonly priorPanics: number; +} + +/** + * One relay-outage probe: establish a relay-carried connection, take the + * relay away for `OUTAGE_HOLD_MS`, bring it back, and require that the + * connection resumed, that the endpoint still dials, and that closing it + * with redials in flight raises no trap. A bricked endpoint comes back + * as a report whose fields fail the scenario's checks; only host-side + * noise throws. + */ +async function outageProbeOnce(control: RelayControl): Promise { + const server = await newEndpointInstance({ label: "outage-server" }); + const client = await newEndpointInstance({ label: "outage-client" }); + const bindOptions = { alpns: [ALPN], relayUrl: control.url(), webrtc: false }; + const sep = await deadline(bindEndpoint(server, bindOptions), 30_000, "server bind"); + const cep = await deadline(bindEndpoint(client, bindOptions), 30_000, "client bind"); + try { + return await outageProbeBody(control, sep, cep); + } catch (err) { + // An attempt that throws still owns two bound endpoints, and their + // pumps would go on redialing through every later scenario. + await closeQuietly(cep, "client close after a failed outage attempt"); + await closeQuietly(sep, "server close after a failed outage attempt"); + throw err; + } +} + +/** Close an endpoint without letting its own failure mask another. */ +async function closeQuietly(ep: Endpoint, what: string): Promise { + try { + await deadline(ep.close(), 15_000, what); + } catch { /* the endpoint may already be dead; the caller is unwinding */ } +} + +async function outageProbeBody( + control: RelayControl, + sep: Endpoint, + cep: Endpoint, +): Promise { + const serverId = await sep.id(); + const addrs: TransportAddr[] = [{ kind: "relay", value: control.url() }]; + const conn = await deadline( + cep.connect({ endpointId: serverId, addrs }, ALPN), + 60_000, + "connect", + ); + const sconn = await deadline(sep.accept(), 60_000, "accept"); + const preEcho = await echoRoundtrip(conn, sconn, "pre-outage echo"); + const prePath = await conn.path(); + + await control.stop(); + let wentDown = false; + for (let i = 0; i < 100; i++) { + if (!await portListening(RELAY_PORT)) { + wentDown = true; + break; + } + await settle(100); + } + + // The outage proper: no export call in flight, so the guests' pumps + // alone drive the redials, and no keep-alive can reach either peer. + await settle(OUTAGE_HOLD_MS); + await control.start(); + + const clientState = await conn.state(); + const serverState = await sconn.state(); + const alive = clientState === "open" && serverState === "open"; + const startedRecovery = performance.now(); + // The echo does not wait for the reconnection: it is written into a + // connection whose transmits are being dropped, and QUIC's loss + // recovery delivers it once the redial lands. + const postEcho = alive + ? await deadline( + echoRoundtrip(conn, sconn, "post-outage echo"), + OUTAGE_RECOVERY_MS, + "post-outage echo", + ) + : ""; + const recoveryMs = performance.now() - startedRecovery; + + // The anti-brick regression: a fresh dial on the SAME endpoints. + let freshEcho = ""; + if (alive) { + const fresh = await deadline( + cep.connect({ endpointId: serverId, addrs }, ALPN), + 60_000, + "post-outage connect", + ); + const freshServer = await deadline(sep.accept(), 60_000, "post-outage accept"); + freshEcho = await echoRoundtrip(fresh, freshServer, "post-outage fresh echo"); + await fresh.close(CLOSE_CODE, CLOSE_REASON); + await conn.close(CLOSE_CODE, CLOSE_REASON); + } + + // Teardown with redials pending: take the relay away again, let the + // backoff arm a dial, and close both endpoints on top of it. + await control.stop(); + await settle(OUTAGE_TEARDOWN_MS); + const priorPanics = takeGuestPanics().length; + await deadline(cep.close(), 15_000, "client close mid-redial"); + await deadline(sep.close(), 15_000, "server close mid-redial"); + await settle(500); + const teardownPanics = takeGuestPanics(); + await control.start(); + + return { + prePath, + preEcho, + wentDown, + clientState, + serverState, + postEcho, + recoveryMs, + freshEcho, + teardownPanics, + priorPanics, + }; +} + async function main(): Promise { installPanicWatchdog(); console.log("iroh endpoint exam (polyengine / stock Deno)"); - const relay = await startRelay(); + let relay = await startRelay(); + // The outage scenario replaces the relay process; every later use + // reads this binding, and the URL is the same across a restart. + const relayControl: RelayControl = { + url: () => relay.url, + stop: () => relay.stop(), + start: async () => { + relay = await startRelay(); + }, + }; try { // -- 1 ------------------------------------------------------------------- await scenario(1, "bind + identity (webcrypto ed25519 path)", async (v) => { @@ -785,9 +957,70 @@ async function main(): Promise { ); // -- 7 ------------------------------------------------------------------- + await scenario( + 7, + "relay outage: the connection survives it and the endpoint stays usable", + async (v) => { + if (!relay.owned) { + v.status = "BLOCKED"; + v.detail = "the relay was pre-existing and adopted, so this run cannot stop it"; + return; + } + let r: OutageReport | undefined; + let lastError = ""; + // Two attempts only: each costs the full outage in wall time, + // and the two handshakes it runs lose the RefCell race rarely. + const attempts = 2; + for (let attempt = 1; attempt <= attempts && !r; attempt++) { + takeGuestPanics(); + try { + r = await outageProbeOnce(relayControl); + } catch (err) { + lastError = describeError(err); + console.log(` attempt ${attempt}/${attempts} failed: ${lastError}`); + // The relay is left running whatever the attempt did with it. + if (!await portListening(RELAY_PORT)) await relayControl.start(); + await settle(100); + } + } + if (!r) throw new Error(`no attempt completed; last: ${lastError}`); + if (r.priorPanics > 0) { + v.notes.push(`${r.priorPanics} guest panic(s) before teardown (RefCell borrow hazard)`); + } + check(v, r.preEcho === MESSAGE.toUpperCase(), "a pre-outage echo round-trip completed"); + check(v, r.prePath === "relay", "the connection rode the relay wire"); + check(v, r.wentDown, `the relay stopped accepting on ${RELAY_PORT}`); + check( + v, + r.clientState === "open" && r.serverState === "open", + `both connections outlived ${OUTAGE_HOLD_MS} ms without a relay ` + + `(client ${r.clientState}, server ${r.serverState})`, + ); + check( + v, + r.postEcho === MESSAGE.toUpperCase(), + `an echo completed after the relay returned (${r.recoveryMs.toFixed(0)} ms)`, + ); + check( + v, + r.freshEcho === MESSAGE.toUpperCase(), + "a FRESH dial on the same endpoints succeeded after the outage", + ); + check( + v, + r.teardownPanics.length === 0, + `no guest trap closing the endpoints with redials in flight ` + + `(${r.teardownPanics.join("; ")})`, + ); + v.detail = `survived ${(OUTAGE_HOLD_MS / 1000).toFixed(0)} s without a relay; ` + + `echo back ${r.recoveryMs.toFixed(0)} ms after it returned, then a fresh dial`; + }, + ); + + // -- 8 ------------------------------------------------------------------- // Last by necessity: this scenario stops the relay every later // scenario would need. - await scenario(7, "teardown: close + wait-closed, relay reaped", async (v) => { + await scenario(8, "teardown: close + wait-closed, relay reaped", async (v) => { const inst = await newEndpointInstance({ label: "teardown" }); const ep = await deadline( bindEndpoint(inst, { alpns: [ALPN], relayUrl: relay.url, webrtc: false }), diff --git a/justfile b/justfile index 5ac0e0c..d767a34 100644 --- a/justfile +++ b/justfile @@ -82,9 +82,8 @@ polyengine-setup: cd host-polyengine && deno install --frozen --allow-scripts=npm:node-datachannel # The endpoint exam on the polyengine host: the endpoint component -# runtime-linked under stock Deno — bind + identity, relay echo, WebRTC -# upgrade, the issue #10 concurrency rows, teardown. See -# host-polyengine/README.md. +# runtime-linked under stock Deno — lifecycle, wires, concurrency, and +# liveness/recovery scenarios. See host-polyengine/README.md. exam-polyengine: build-components polyengine-setup #!/usr/bin/env bash set -euo pipefail