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