From 6355ef1ad32db195a625085c40e38f9857a74492 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Tue, 25 Aug 2026 21:32:29 -0400 Subject: [PATCH] Fix #123: every subduction call gets a real deadline, and a silent handshake closes its own wire NeverTimeout -- the Timeout impl that returned Ok(fut.await), no bound ever -- is retired. It co-conspired in the #113 hang (a sync call parked forever on a dead-but-unfailable transport; #122 removed that corpse) and still converted every reachable-but-silent peer into wedged-forever: the page's own 30s waits masked the UX, but every abandoned sync left an engine task parked for the life of the page, holding Arc and connection Rcs. MonotonicTimeout races each call against wasi:clocks/monotonic-clock @0.3.0's wait-for (the async track; 0.2.9's pollable shape is not awaitable from this guest). The composite already imported the interface -- iroh.wit pulls it, and @polyengine/wasi serves waitFor on the 0.3 union provider in both hosts -- so the guest world import (engine.wit) is the only surface change. Every call site already passed CallTimeout::Default = the crate's 30s roundtrip bound; smoke-measured 100ms->109ms, 250ms->252ms; happy-path overhead not measurable (rebind-sync settle 0.20s, cross 0.03s, unchanged). Abort is safe by subduction's own design: the multiplexer's PendingGuard removes the pending entry when the call future drops, and ingestion/subscription happen only in the success arm (cited at the impl, verified against the pinned source). A TimedOut outcome is NOT 'gone:' -- the wire is alive, the peer is silent -- and the page keeper's leave-alone arm deliberately does not re-dial on it (#78's double-dial discipline; one-sentence notes in solo.ts and relay-partition.ts's wave-3 history). The keyhive handshake sat outside subduction's Timeout and parked the same way; the initiate side now carries its own 30s bound (the responder's 300s anti-replay window is a different fact, untouched). A timed-out dial CLOSES its connection explicitly -- this is by definition the case where the peer is alive, so nothing else ever would: the close resolves wait-closed, which wakes conn_gone_monitor, which closes the conn_inbound queues and finds the named timeout Err already in conn_results and leaves it (a named error outranks gone:). One teardown mechanism, the ordinary one. And the leak the fix would have created: with a real bound every abandoned handle INSERTS an outcome nobody will read (sync-status is one-shot; the page polls 30s then gives up). SYNCS_CAP=256 with an insertion-order deque evicts oldest -- eviction horizon ~5min at the demo's 48 settles/min against the page's 30s read horizon. demo/host/timeout-check.ts (+ just timeout-check): control settles 0.20s; a blackholed peer produces a BOUNDED named outcome at ~30s. The probe's banner is honest that it cannot discriminate the call bound from the QUIC idle timeout (both 30s; measured 30.03s fixed vs 30.20s reverted) -- true live-wire-silent-peer needs a cooperating mute peer, which would mean test-only WIT surface on a shipping world. The property it pins is bounded-not-parked, which is what #123 names. Gates: engine just check (-D warnings) clean; conn-gone + rebind-sync unchanged (x3); timeout-check x3; full e2e 34/34 (relay-partition converge 5.3s, no regression); devstore ALL REQUIRED ROWS PASS; soak seed 2/25 green. Independently reviewed; should-fix applied. --- demo/e2e/scenarios/relay-partition.ts | 7 +- demo/host/solo.ts | 6 + demo/host/timeout-check.ts | 305 ++++++++++++++++++++++++++ demo/justfile | 11 + engine/guest/src/lib.rs | 239 ++++++++++++++++++-- engine/guest/wit/engine.wit | 15 ++ 6 files changed, 561 insertions(+), 22 deletions(-) create mode 100644 demo/host/timeout-check.ts 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