diff --git a/demo/e2e/scenarios/relay-partition.ts b/demo/e2e/scenarios/relay-partition.ts index 259c1757..a33e9c20 100644 --- a/demo/e2e/scenarios/relay-partition.ts +++ b/demo/e2e/scenarios/relay-partition.ts @@ -116,7 +116,12 @@ // index 1. // 4. And `sync_with_peer` walks that list IN ORDER, under this // engine's never-firing timeout — so it called the dead one first -// and parked there for ever, never reaching the live one. +// and parked there for ever, never reaching the live one. That +// timeout is CAPPED since #123 (subduction's 30s default, on the +// monotonic clock): the same stall would now end with a name +// instead of never — and a sync that times out means the wire is +// alive and the PEER IS SILENT, which is not `gone:` and must not +// be re-dialled on. // // THE ASYMMETRY EXPLAINED: only the side that CALLS `sync_with_peer` // walks the stale list. An acceptor answering an inbound request diff --git a/demo/host/solo.ts b/demo/host/solo.ts index 9071eb8c..47a195ae 100644 --- a/demo/host/solo.ts +++ b/demo/host/solo.ts @@ -4786,6 +4786,12 @@ async function startApp( // endpoint having died (`Closed`), and that is a fact about the // transport rather than about the peer: it is `rebindEndpoint`'s to // answer, not the re-dial path's. + // + // Since #123 that leave-alone arm also catches a TIMED-OUT sync + // outcome, and deliberately: a timeout says the wire is alive and + // the peer is silent — the one situation where re-dialling would + // mint the #78 double-dial against a connection that is working + // fine. Only `gone:` re-dials. if (dialledPeer) { const h = await probeConn(dialledPeer.conn); if (h.kind === "gone") { diff --git a/demo/host/timeout-check.ts b/demo/host/timeout-check.ts new file mode 100644 index 00000000..7034c85e --- /dev/null +++ b/demo/host/timeout-check.ts @@ -0,0 +1,305 @@ +// The gate for #123: a sync against a SILENT peer must end, and end +// with a name. +// +// deno run -A host/timeout-check.ts (or: just timeout-check) +// +// WHY A THIRD PROBE. `conn-gone-check.ts` proves the engine notices a +// DEAD wire; `rebind-sync-check.ts` proves it can USE a replacement. +// Neither says anything about the third shape, which is the one #123 is +// about: a wire nothing has killed, carrying a peer that has stopped +// answering. Until #123 the engine's subduction `Timeout` was +// `NeverTimeout` — literally `Ok(fut.await)` — so that shape did not +// produce a slow sync, it produced a sync that NEVER SETTLED, and the +// handle sat unread for the life of the page. +// +// WHAT IS ASSERTED, EXACTLY: BOUNDED, NOT PARKED. The claim under test +// is that `sync-start` against a silent peer REACHES A NAMED OUTCOME +// within a bound — not which layer produced it. Two mechanisms can +// legitimately win the race here: +// +// - THE CALL TIMEOUT (#123's `MonotonicTimeout`): subduction's +// `CallTimeout::Default` resolves to `DEFAULT_ROUNDTRIP_TIMEOUT`, +// 30s, and the roundtrip is dropped. `conn-status` still says LIVE: +// the wire is fine, the peer is silent. +// - THE QUIC IDLE TIMEOUT plus #122's teardown: the transport itself +// gives up on the blackholed path, `conn-status` flips to `gone:`, +// the connection is removed and the sync fails against a closed +// transport. +// +// Both are bounded outcomes and both are correct. The probe REPORTS +// which one won (it watches `conn-status` alongside the sync and prints +// the order the two landed in) and fails only on the third +// possibility — nothing settles at all — which is the pre-#123 +// behaviour. Pinning one mechanism would make this probe a test of the +// relay's idle timer as much as of the engine's bound. +// +// MEASURED, AND SAID PLAINLY: ON THIS RIG QUIC USUALLY WINS. Running +// this probe against a deliberately reverted guest (the `Ok(fut.await)` +// body put back) settles in 30.20s and reports `gone:`, against 30.03s +// and `gone:` with the bound in place. iroh's idle timeout and +// subduction's `DEFAULT_ROUNDTRIP_TIMEOUT` are BOTH 30s, and blackholing +// a path takes the QUIC keepalives down with the application traffic — +// so a headless rig cannot produce "live wire, silent peer" from the +// network side alone. That is a fact about transports, not a gap that +// more probe cleverness closes: app-level silence needs a cooperating +// peer engine that ignores requests, which would mean test-only WIT +// surface on a shipping world, and that trade is not worth it. +// +// WHAT THIS PROBE IS THEREFORE WORTH, precisely: it pins +// BOUNDED-NOT-PARKED end to end — the property #123 is about — and it +// would go red on any future path where the transport does NOT rescue +// the sync (which is the whole family the bound exists for, and exactly +// the shape #113 hit). It does not, on this rig, prove that the call +// bound specifically fired. What proves the bound's machinery works is +// the clock itself: `wasi:clocks/monotonic-clock@0.3.0`'s `wait-for` +// was measured through a temporary guest export during #123's bringup +// (asked 100ms → guest measured 109ms; asked 250ms → 252ms), and the +// export was removed rather than left as permanent test-only surface. +// +// THE ARRANGEMENT. Alice talks to the relay directly; BOB reaches it +// through the e2e suite's severable TCP proxy (e2e/proxy.ts, imported +// rather than re-implemented — it is harness code and its +// `blackhole()` is precisely this fault shape: bytes stop moving, no +// RST, no FIN, sockets stay open). Blackholing the proxy leaves bob's +// engine running and its sockets open while nothing it says reaches +// anyone — app-level silence, which is what a wedged remote engine +// looks like from across a healthy network. +// +// The healthy CONTROL beat runs first and is not a formality: a timeout +// impl that bounds the silent path by also slowing the working one is +// the wrong fix, so the control asserts a settle in well under a second +// on the same rig. + +import { type Engine, newEngine, unhex, until } from "../../runtime/engine.ts"; +import { probeNoNet } from "./probe-net.ts"; +import { startTcpProxy } from "../e2e/proxy.ts"; + +const ENVELOPE = new URL("../build/engine.plan.json", import.meta.url); +const WASM = new URL("../../engine/target/composed.wasm", import.meta.url); +const RELAY_BIN = new URL("../../engine/.deps/relay/bin/iroh-relay", import.meta.url).pathname; + +/// The healthy-path budget. `rebind-sync-check` measures ~0.2s on this +/// rig; a second is generous and still an order of magnitude under the +/// bound being tested, so the two beats can never be confused. +const CONTROL_MS = 1_000; +/// How long the blackholed sync gets before the probe calls it PARKED. +/// The engine's own bound is 30s (subduction's `DEFAULT_ROUNDTRIP_TIMEOUT`, +/// mirrored by the handshake constant in the guest); `sync_with_peer` +/// may spend that per connection it walks, so this is 30s plus room for +/// one such walk plus polling slack — a "long enough to believe never" +/// figure, not a performance budget. +const BOUND_MS = 90_000; +/// The `conn-status` dead-wire marker (engine.wit's `conn-status` +/// contract). Matched with `includes`: the host prefixes the guest's +/// message ("component error: gone: …"). +const GONE = "gone:"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +function freePort(): number { + const l = Deno.listen({ port: 0 }); + const port = (l.addr as Deno.NetAddr).port; + l.close(); + return port; +} + +/** conn-gone-check.ts's Relay, verbatim in behaviour: an ephemeral port + * of this probe's own (a shared :3340 would take sibling runs down with + * it), and `/generate_204` — the relay's own net-report endpoint — as + * the readiness signal. */ +class Relay { + #proc: Deno.ChildProcess | null = null; + #dir: string | null = null; + readonly url: string; + constructor(readonly port: number) { + this.url = `http://127.0.0.1:${port}`; + } + + async start(): Promise { + if (this.#proc) return; + try { + await Deno.stat(RELAY_BIN); + } catch { + console.error(`no iroh-relay at ${RELAY_BIN} — run \`cd engine && just relay-bin\``); + Deno.exit(2); + } + this.#dir ??= await Deno.makeTempDir({ prefix: "pm-timeout-relay." }); + const cfg = `${this.#dir}/relay.toml`; + await Deno.writeTextFile( + cfg, + `http_bind_addr = "127.0.0.1:${this.port}"\nenable_metrics = false\n`, + ); + this.#proc = new Deno.Command(RELAY_BIN, { + args: ["--dev", "--config-path", cfg], + stdout: "null", + stderr: "null", + }).spawn(); + for (let i = 0; i < 120; i++) { + try { + const r = await fetch(`${this.url}/generate_204`, { signal: AbortSignal.timeout(2_000) }); + await r.body?.cancel(); + if (r.status === 204 || r.ok) return; + } catch { /* not up yet */ } + await sleep(250); + } + throw new Error(`the local relay never answered on ${this.url}`); + } + + async dispose(): Promise { + const proc = this.#proc; + this.#proc = null; + if (proc) { + try { + proc.kill("SIGKILL"); + } catch { /* already dead */ } + await proc.status; + } + if (this.#dir) await Deno.remove(this.#dir, { recursive: true }).catch(() => {}); + } +} + +/** Whether the engine still calls this wire alive — read at the moment + * the sync settles, which is what tells the two winning mechanisms + * apart. */ +async function wireState(e: Engine, conn: number): Promise<"live" | "gone" | "error"> { + try { + return (await e.driver.connStatus(conn)) === undefined ? "error" : "live"; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + return message.includes(GONE) ? "gone" : "error"; + } +} + +/** A `sync-start` allowed NOT to settle: `until` throws on timeout, + * which is the right shape for the control and the wrong shape for the + * assertion — here the difference between "errored" and "never + * answered" is the entire result, so both are returned as data. */ +async function settle( + e: Engine, + peer: Uint8Array, + part: Uint8Array, + subscribe: boolean, + boundMs: number, + /** When given, `conn-status` is polled alongside the sync and the + * first moment it reported `gone:` comes back with the result — the + * ordering of the two is what names the mechanism. */ + watchConn?: number, +): Promise<{ ms: number; outcome: string; errored: boolean; goneMs: number | null } | null> { + const t0 = performance.now(); + const h = await e.driver.syncStart(peer, part, subscribe); + let goneMs: number | null = null; + const done = (outcome: string, errored: boolean) => ({ + ms: performance.now() - t0, + outcome, + errored, + goneMs, + }); + while (performance.now() - t0 < boundMs) { + if (watchConn !== undefined && goneMs === null) { + if (await wireState(e, watchConn) === "gone") goneMs = performance.now() - t0; + } + try { + const s = await e.driver.syncStatus(h); + if (s !== undefined) return done(s, false); + } catch (err) { + // An ERRORED sync is a SETTLED sync, and the outcome this probe + // most expects: a bound fired and said so. + return done(err instanceof Error ? err.message : String(err), true); + } + await sleep(200); + } + return null; +} + +async function main() { + const artifacts = { + envelope: await Deno.readTextFile(ENVELOPE), + bytes: await Deno.readFile(WASM), + }; + const relay = new Relay(freePort()); + await relay.start(); + // Bob's whole view of the relay goes through here. + const proxy = await startTcpProxy(relay.port); + console.log(`relay up on ${relay.url}; bob's path proxied via ${proxy.url}`); + + const alice = await newEngine("alice", artifacts, probeNoNet); + const bob = await newEngine("bob", artifacts, probeNoNet); + try { + const aliceId = unhex(await alice.driver.init(false)); + const bobId = unhex(await bob.driver.init(false)); + + await alice.driver.irohBind(relay.url); + const bobEp = await bob.driver.irohBind(proxy.url); + const cb = await bob.driver.irohStart(false, new Uint8Array(), proxy.url, new Uint8Array()); + const ca = await alice.driver.irohStart(true, unhex(bobEp), relay.url, bobId); + await until( + "handshake", + async () => (await alice.driver.connStatus(ca)) && (await bob.driver.connStatus(cb)), + ); + await until( + "contact cards over the bridge", + async () => + (await alice.driver.khKnowsAgent(bobId)) && (await bob.driver.khKnowsAgent(aliceId)), + ); + + const part = await alice.driver.createPartition(); + await alice.driver.khAddMember(part, bobId, "edit"); + await alice.driver.sealPartition(part); + await bob.driver.adoptPartition(part); + await until("bob's keyhive knows the doc", () => bob.driver.khKnowsAgent(part)); + + // Pull before subscribe: a subscribe=true FIRST sync reports commits + // received but does not store them (bringup.ts records the same + // ordering constraint). + const pullHandle = await bob.driver.syncStart(aliceId, part, false); + await until("bob pull", () => bob.driver.syncStatus(pullHandle)); + await until("bob decrypts creation", async () => (await bob.tasks.revision()) >= 1n); + + // --- beat 1: the CONTROL — a healthy sync settles FAST ------------ + const control = await settle(alice, bobId, part, true, CONTROL_MS + 5_000); + if (!control || control.errored || control.ms > CONTROL_MS) { + throw new Error( + `CONTROL: a healthy sync must settle in under ${CONTROL_MS}ms — got ` + + (control ? `${control.ms.toFixed(0)}ms (${control.outcome})` : "no settle at all") + + ` — the bound added in #123 must not cost the happy path anything`, + ); + } + console.log(`control: healthy sync settled in ${(control.ms / 1000).toFixed(2)}s`); + + // --- beat 2: bob goes SILENT (not dead) --------------------------- + proxy.blackhole(); + console.log("bob's relay path blackholed: sockets open, bytes stop moving"); + + // --- beat 3: THE ASSERTION — bounded, not parked ------------------ + const t0 = performance.now(); + const stuck = await settle(alice, bobId, part, true, BOUND_MS, ca); + if (!stuck) { + throw new Error( + `sync-start against a SILENT peer never settled in ${BOUND_MS / 1000}s — ` + + `this is the #123 shape exactly: the wire is not dead, the peer is not ` + + `answering, and the call has no bound`, + ); + } + const wire = await wireState(alice, ca); + const mechanism = wire === "gone" + ? `the QUIC idle timeout + #122 teardown (conn-status reported gone: at ` + + `${stuck.goneMs === null ? "settle time" : `${(stuck.goneMs / 1000).toFixed(2)}s`}) ` + + `— the expected winner on a blackholed path, see this file's banner` + : "the #123 call bound (conn-status still LIVE — live wire, silent peer)"; + console.log( + `\nBOUNDED in ${(stuck.ms / 1000).toFixed(2)}s ` + + `(${stuck.errored ? "errored" : "reported"}): ${stuck.outcome.slice(0, 200)}`, + ); + console.log(`mechanism that won the race: ${mechanism}`); + console.log(`elapsed since the blackhole: ${((performance.now() - t0) / 1000).toFixed(2)}s`); + console.log("timeout-check: OK"); + } finally { + await proxy.close(); + await relay.dispose(); + } +} + +if (import.meta.main) { + await main(); + Deno.exit(0); +} diff --git a/demo/justfile b/demo/justfile index f31a462b..ae2fe870 100644 --- a/demo/justfile +++ b/demo/justfile @@ -71,6 +71,17 @@ rebind-sync: translate deno check host/rebind-sync-check.ts deno run -A host/rebind-sync-check.ts +# #123: the third wire shape — alive, and silent. Bob's relay path is +# blackholed through the e2e severable proxy (bytes stop moving; no RST, +# no FIN), so his engine keeps running while nothing it says arrives. +# Alice's sync against him must reach a NAMED outcome instead of parking +# forever, which is what `NeverTimeout` used to guarantee it did not. +# The probe asserts bounded-not-parked and reports which mechanism won +# (the call bound, or the QUIC idle timeout racing it) — see its banner. +timeout-check: translate + deno check host/timeout-check.ts + deno run -A host/timeout-check.ts + # The #20 G5 kill-and-resume beat: the engine checkpoints into a real # directory (@polyengine/wasi's `filesystem-node` backend) and a FRESH # DENO PROCESS resumes from it. Needs no infra at all — the assertion is diff --git a/engine/guest/src/lib.rs b/engine/guest/src/lib.rs index dab3ef1e..31b376cb 100644 --- a/engine/guest/src/lib.rs +++ b/engine/guest/src/lib.rs @@ -42,7 +42,7 @@ mod usdoc; mod wordlist; use std::cell::{Cell, RefCell}; -use std::collections::{BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::rc::Rc; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -51,7 +51,7 @@ use automerge::transaction::Transactable; use automerge::{AutoCommit, Change, ObjType, ReadDoc, ScalarValue, Value, ROOT}; use ed25519_dalek::VerifyingKey as DalekVerifyingKey; use future_form::{FutureForm, Local}; -use futures::future::{AbortHandle, Abortable, LocalBoxFuture}; +use futures::future::{select, AbortHandle, Abortable, Either, LocalBoxFuture}; use polymorph_webcrypto_guest::{ aes_gcm::{self, AesVariant}, @@ -156,7 +156,7 @@ type Sd = Subduction< Hdl, Auth, WebcryptoSigner, - NeverTimeout, + MonotonicTimeout, WitSpawn, CountLeadingZeroBytes, 256, @@ -306,16 +306,84 @@ impl Spawn for WitSpawn { } } +/// Sleep `dur` on the host's monotonic clock. +/// +/// `wasi:clocks/monotonic-clock@0.3.0`'s `wait-for` is an `async func`, +/// so awaiting it is the whole implementation — no reactor, no polling +/// loop, no `pollable` bridging (which is why the world imports the +/// @0.3 track and not the @0.2.9 one that `std` already drags in; see +/// the import's note in wit/engine.wit). +/// +/// Saturating on the nanosecond conversion, not wrapping: a duration +/// past `u64::MAX` nanoseconds (~584 years) is nobody's real deadline, +/// and truncating one into a SHORT sleep would turn an absurd bound +/// into a spuriously firing one. +async fn sleep_for(dur: Duration) { + let nanos = u64::try_from(dur.as_nanos()).unwrap_or(u64::MAX); + wasi::clocks::monotonic_clock::wait_for(nanos).await; +} + +/// The bound on every subduction call (#123). +/// +/// WHAT IT BOUNDS. One whole request/response ROUNDTRIP — the send plus +/// the wait for that request's response — per call, not a sync round and +/// not an idle timer. `sync_with_peer` makes several calls; each gets +/// its own fresh `dur`. +/// +/// WHERE THE DURATION COMES FROM. Every call site here passes +/// `CallTimeout::Default`, which subduction resolves to its own +/// `DEFAULT_ROUNDTRIP_TIMEOUT` — 30s, subduction_core/src/multiplexer.rs:47, +/// deliberately the Erlang/OTP `GenServer.call` convention: calls are +/// BOUNDED BY DEFAULT even on a transport whose recv loop never notices +/// a byte-alive but protocol-silent peer. This impl supplies the clock +/// that constant was always assuming; the engine simply had none. +/// +/// WHY DROPPING THE LOSER IS SAFE. Subduction's calls are cancel-safe by +/// documented design: `ManagedConnection::call` registers its pending +/// response under an RAII `PendingGuard` +/// (subduction_core/src/connection/managed.rs:110-155) whose `Drop` +/// removes the multiplexer entry — named there as making the call safe +/// against exactly this, "an outer deadline elapsed, a `select!` arm +/// lost". Request IDs are never reused, so a straggling response can +/// never be mismatched onto a later call. Ingestion and subscription +/// happen only in the success arm, so a timed-out call cannot have +/// half-applied anything. +/// +/// WHAT A `TimedOut` OUTCOME MEANS, AND WHAT IT IS NOT. It means the +/// wire is alive and the PEER IS SILENT — a diagnosis in its own right, +/// distinct from `gone:` (the wire itself is dead, see `conn_status`). +/// The page must NOT re-dial on it: re-dialling a live wire is the #78 +/// double-dial hazard, and the wire-keeper (demo/host/solo.ts) keeps +/// timeouts in its leave-alone arm on purpose. +/// +/// RETIRED HERE: `NeverTimeout`, which returned `Ok(fut.await)` — no +/// bound, ever. It was a co-conspirator in the #113 stale-transport +/// hang (fixed in #122): `sync_with_peer` walks a peer's connections +/// serially, and a call parked on a dead-but-unfailable transport +/// wedged the sync forever where a real bound would have made it a 30s +/// stall with a name. #[derive(Clone, Debug, PartialEq)] -struct NeverTimeout; +struct MonotonicTimeout; -impl Timeout for NeverTimeout { +impl Timeout for MonotonicTimeout { fn timeout<'a, T2: 'a>( &'a self, - _dur: Duration, + dur: Duration, fut: ::Future<'a, T2>, ) -> ::Future<'a, Result> { - Box::pin(async move { Ok(fut.await) }) + // The shape upstream uses for the same job, with `wait-for` + // where it has `futures_timer::Delay` + // (subduction_websocket/src/timeout.rs's `FuturesTimerTimeout`, + // and subduction_wasm's `JsTimeout` over a `setTimeout` + // promise). `select` — not `select!` — so the winner is taken by + // value and the loser is DROPPED, which is what arms the + // `PendingGuard` above. + Box::pin(async move { + match select(fut, Box::pin(sleep_for(dur))).await { + Either::Left((val, _sleep)) => Ok(val), + Either::Right(((), _fut)) => Err(TimedOut), + } + }) } } @@ -618,6 +686,11 @@ struct State { conn_inbound: HashMap>>>, syncs: HashMap>, + /// Insertion order over `syncs`, so the oldest outcome can be + /// evicted when the table is over `SYNCS_CAP`. See + /// `record_sync_outcome` for why an outcome map that is read + /// one-shot still needs a cap. + sync_order: VecDeque, endpoint: Option>, /// The relay this endpoint bound to. Pairing has no relay hint in the /// code (PAIRING.md §1), so both sides use their configured one. @@ -665,6 +738,41 @@ fn with_state(f: impl FnOnce(&mut State) -> R) -> Result { }) } +/// How many sync outcomes may sit unread before the oldest is dropped. +/// +/// THE ABANDONED-HANDLE PATTERN (#123). `syncs` is one-shot — an entry +/// is REMOVED as it is read (see `sync_status`) — and that was enough +/// while a stuck sync simply never inserted anything: the task parked +/// forever and its handle's row never existed. A real bound reverses +/// that. Every sync the page gave up on (the page polls for ~30s and +/// then stops asking) now lands an outcome nobody will ever read, and +/// "removed on read" bounds nothing when the read never comes. The map +/// once grew without bound for the neighbouring reason and the demo +/// starts ~48 syncs/minute, forever; leaving the new leak in would be +/// re-earning the same bug from the other side. +/// +/// 256 is a ceiling, not a working set: the honest in-flight count is a +/// handful, so an eviction here means outcomes are being abandoned far +/// faster than read — worth noticing, never worth wedging over. +const SYNCS_CAP: usize = 256; + +/// Record a sync outcome, evicting the OLDEST if the table is over +/// `SYNCS_CAP`. Eviction costs the abandoning caller its outcome, which +/// it was never going to read; a live poller's entry is minutes newer +/// than anything at the front. +fn record_sync_outcome(s: &mut State, handle: u32, outcome: Result) { + s.syncs.insert(handle, outcome); + s.sync_order.push_back(handle); + // The DEQUE is what is capped, not the map: an entry already read + // still holds its slot until it reaches the front, so both stay + // bounded by `SYNCS_CAP` without a second pass to find the holes. + while s.sync_order.len() > SYNCS_CAP { + if let Some(oldest) = s.sync_order.pop_front() { + s.syncs.remove(&oldest); + } + } +} + fn now_ts() -> TimestampSeconds { let secs = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -759,6 +867,25 @@ async fn nudge_keyhive_sync() { // --- shared handshake + iroh pumps (unchanged from the skeleton) --- +/// How long the INITIATOR's keyhive/subduction handshake gets before it +/// is called silent (#123). +/// +/// Derived from the caller above it, not invented: the page's dial has a +/// 30s `until` on `conn-status` (demo/host/solo.ts's wire-keeper), so a +/// handshake still parked at 30s has already lost its audience — the +/// page has given up and moved on, and every second past that is a task +/// held open for nobody. Matching the page keeps one number in play +/// instead of two disagreeing ones, and it happens to coincide with +/// subduction's own `DEFAULT_ROUNDTRIP_TIMEOUT` (30s), which is the same +/// judgement about how long a live wire may stay silent. +/// +/// The RESPONDER side is deliberately not touched: `handshake::respond` +/// below already carries its own 300s accept deadline (an anti-replay +/// window over the peer's timestamp, not a liveness bound), and an +/// acceptor waiting on a dialler is the cheap direction — it holds a +/// queue, not a user. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); + #[allow(clippy::too_many_arguments)] async fn subduction_handshake( transport: QueueTransport, @@ -774,16 +901,73 @@ async fn subduction_handshake( let expected = arr32(&expected_peer, "expected peer")?; let audience = Audience::known(PeerId::new(expected)); let nonce = Nonce::from_bytes(rand::random::<[u8; 16]>()); - handshake::initiate::( - QueueHandshake(transport), - |h, _peer| (MessageTransport::new(h.0), ()), - &signer, - audience, - now, - nonce, + // BOUNDED (#123). The handshake runs BEFORE the connection is + // handed to subduction, so it is outside `MonotonicTimeout`'s + // reach entirely: `initiate` writes its hello and then awaits + // the peer's reply on a queue that a byte-alive but silent peer + // never feeds — parking this task, and the dial with it, + // forever. Same machinery as the call bound: race it against + // the monotonic clock and drop the loser. + // + // Dropping a half-finished `initiate` is safe in the way that + // matters here: nothing has been registered with subduction + // yet — the authenticated connection only reaches it via the + // `add_connection` below, in the success arm. + // + // THE WIRE, THOUGH, IS OURS TO END, and the timeout arm does it + // EXPLICITLY (see the `close` there). Nothing else would: this + // is by definition the case where the peer is ALIVE, so QUIC + // raises no error, `wait-closed` never resolves on its own, and + // `conn_gone_monitor` — which only ever runs after + // `wait-closed` — never gets to do its job. Left alone, a + // timed-out dial would leak the whole chain: an open connection + // the wire-keeper re-dials past every ~30s, plus an + // `iroh_reader` still shovelling frames into a `conn_inbound` + // queue whose only consumer died with the dropped handshake + // future. + // + // THE TEARDOWN CHAIN, once we close: `close(code, reason)` makes + // this connection's own `wait-closed` resolve, which wakes the + // `conn_gone_monitor` already spawned for it by the wiring task, + // which closes the registered `conn_inbound` senders (so the + // reader's writes fail and it exits) and then applies its rule 1 + // to `conn_results` — where it finds the timeout `Err` this + // function is about to return and leaves it in place, because a + // named error outranks `gone:`. One mechanism, the ordinary one; + // this arm only supplies the close the peer was never going to. + let conn_id = transport.id; + match select( + Box::pin(handshake::initiate::( + QueueHandshake(transport), + |h, _peer| (MessageTransport::new(h.0), ()), + &signer, + audience, + now, + nonce, + )), + Box::pin(sleep_for(HANDSHAKE_TIMEOUT)), ) .await - .map_err(|e| format!("initiate: {e:?}")) + { + Either::Left((res, _sleep)) => res.map_err(|e| format!("initiate: {e:?}")), + Either::Right(((), _initiate)) => { + // An honest application close: the code is ours (1 = + // "this side gave up on the handshake") and the reason + // is what the peer's `wait-closed` will report, so a + // remote engine debugging its own silence is told why + // rather than left with a bare transport close. + let _ = with_state(|s| { + if let Some(conn) = s.iroh_conns.get(&conn_id) { + conn.close(1, "handshake timed out"); + } + }); + Err(format!( + "initiate: handshake timed out after {}s (peer silent; the wire is not \ + reported dead)", + HANDSHAKE_TIMEOUT.as_secs() + )) + } + } } else { handshake::respond::( QueueHandshake(transport), @@ -908,12 +1092,18 @@ async fn iroh_reader(in_tx: async_channel::Sender>, recv: RecvStream, se /// alongside it), so after an endpoint rebind the peer owns a dead /// connection at index 0 and the live one at index 1. `sync_with_peer` /// walks that list IN ORDER (subduction.rs:2190) and calls on the dead -/// one first; with this engine's `NeverTimeout` (see it above — the -/// `Timeout` impl that never fires) that call parks forever and the -/// live connection is never reached. Net effect before this fix: after -/// a relay outage, `conn-status` said LIVE within a second and no sync +/// one first; with the `NeverTimeout` this engine carried at the time +/// (a `Timeout` impl that never fired — retired in #123, see +/// `MonotonicTimeout` above) that call parked forever and the live +/// connection was never reached. Net effect before this fix: after a +/// relay outage, `conn-status` said LIVE within a second and no sync /// ever completed again, for the life of the page. /// +/// The bound added in #123 does not replace this repair, and was never +/// an alternative to it: it would have turned the wedge into a 30s +/// stall per stale connection, which is a diagnosis, not a fix. Closing +/// the queues is still what removes the corpse. +/// /// The asymmetry that made it confusing: only the side that CALLS /// `sync_with_peer` walks the stale list. An acceptor answering an /// incoming request replies on the connection the message arrived on, @@ -3797,7 +3987,7 @@ fn finish_init( .signer(signer.clone()) .storage(sd_storage.clone(), policy) .spawner(WitSpawn) - .timer(NeverTimeout) + .timer(MonotonicTimeout) .build::(); wit_bindgen::spawn_local(async move { let _ = listener.await; @@ -3820,6 +4010,7 @@ fn finish_init( conn_results: HashMap::new(), conn_inbound: HashMap::new(), syncs: HashMap::new(), + sync_order: VecDeque::new(), endpoint: None, relay_url: None, iroh_identity: None, @@ -4940,7 +5131,7 @@ impl DriverGuest for Component { )), Err(e) => Err(format!("sync_with_peer: {e:?}")), }; - let _ = with_state(|s| s.syncs.insert(handle, outcome)); + let _ = with_state(|s| record_sync_outcome(s, handle, outcome)); }); Ok(handle) } @@ -4951,6 +5142,12 @@ impl DriverGuest for Component { // REMOVED as it is read, so the table holds only in-flight syncs. // Leaving completed entries in made the map grow without bound — // the demo starts ~48 syncs/minute, forever. + // + // One-shot is no longer the WHOLE bound: since #123 a sync can + // settle after its caller stopped polling, and an outcome that + // is never read is never removed here. `record_sync_outcome` + // caps the table for exactly that case; this stays the fast + // path for outcomes somebody is waiting on. match with_state(|s| s.syncs.remove(&handle))? { Some(Ok(summary)) => Ok(Some(summary)), Some(Err(e)) => Err(e), diff --git a/engine/guest/wit/engine.wit b/engine/guest/wit/engine.wit index b93daa79..4c1de956 100644 --- a/engine/guest/wit/engine.wit +++ b/engine/guest/wit/engine.wit @@ -915,6 +915,21 @@ interface store-fetch-types { } world engine { + /// The clock every BOUND in this engine is measured against (#123). + /// Before it, subduction calls ran under a `Timeout` impl that never + /// fired, so "slow or silent peer" meant "wedged forever" (the #113 + /// latch's co-conspirator). `wait-for` is the whole need: race it + /// against the call and the loser is dropped. + /// + /// The @0.3.0 track, not @0.2.9: only 0.3's `wait-for` is an `async + /// func` a guest can await directly (0.2 hands back a `pollable`, + /// which is a different — blocking — shape). The composite already + /// imported this exact interface for the iroh endpoint, so every + /// embedder that instantiates us today already satisfies it; + /// `@polyengine/wasi`'s batteries serve it from one union provider + /// (wasi/src/clocks.ts, the D-1 note), which is what both the Deno + /// host and the browser worker instantiate through. + import wasi:clocks/monotonic-clock@0.3.0; import polymorph:iroh/endpoint@0.1.0; import polymorph:iroh/identity-generate@0.1.0; /// The other half of the endpoint-identity story: minting an iroh