From 63dbdb64a52c0864c8870cdf69387177cf386529 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Wed, 9 Sep 2026 08:52:23 +0000 Subject: [PATCH] =?UTF-8?q?test:=20coverage=20weak=20spots=20=E2=80=94=20t?= =?UTF-8?q?he=20UdpHalf=20sender-block=20recipe=20found=20(option=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DatagramReader::feed is infallible (the oversize bound is enforced at frame time; the same zombie category as 4014ba6's InvalidLength). The Vec shape, the local adapter's dead InvalidData mapping, and consumer.rs's `?` drop. - The sender-block machinery test (root-gated veth fixture, auto-skip without root): the peer address unassigned + the neighbor table flushed pins the connected sender in a persistent WouldBlock/ENOBUFS state; the test drives the WouldBlock stash, the readiness-Pending arms, AND the accepted-byte high-water Pend in one behavior, then resolves (addr assign + peer bind) and asserts the queued datagrams drain and deliver. Jumbo MTU (65535) keeps IP fragmentation out of the way; the block-state probe runs on a throwaway socket (an oversized probe's partially-built skb chain stays corked on EAGAIN and poisons the socket). - pump_queue_to_socket's stash arm now matches ENOBUFS(105) alongside WouldBlock — the blocked-neighbor state reports 105, and both mean "the socket cannot take this datagram now". - connect_udp's resolve-failure arm covered (an RFC 6761 .invalid target maps to dial_failed; rootless). - The no-witness tunnel_establisher wrapper covered (a harness DialNoWitness mode + a full open round-trip through the wrapper). Negative result pinned in the task doc: tokio 1.53 structurally masks UDP ICMP errors out of the poll drive (poll_recv_ready/try_io mask ERROR out of the read direction), so the recv-side ECONNREFUSED mapping arm is unreachable through the pump's drive — it joins the defensive list. Coverage: 92.86% -> 93.91% lines (50 -> 41 uncovered, all deliberate). Verification: cargo test + cargo test --all-features (91 tests), clippy --all-targets --all-features, fmt, wasm32 check+clippy, cargo doc — all clean. --- src/consumer.rs | 2 +- src/local/mod.rs | 20 +- src/wire.rs | 31 ++- tasks/tunnels/coverage-weak-spots.md | 232 +++++++++---------- tests/end_to_end.rs | 2 +- tests/harness.rs | 34 ++- tests/local_halves.rs | 325 +++++++++++++++++++++++++++ tests/producer_open_op.rs | 44 ++++ 8 files changed, 538 insertions(+), 152 deletions(-) diff --git a/src/consumer.rs b/src/consumer.rs index 13e03b5..6aa6030 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -381,7 +381,7 @@ async fn read_one_datagram( Ok(None) }; } - let mut dgs = reader.feed(&chunk[..n])?.into_iter(); + let mut dgs = reader.feed(&chunk[..n]).into_iter(); if let Some(first) = dgs.next() { return Ok(Some((first, dgs.collect()))); } diff --git a/src/local/mod.rs b/src/local/mod.rs index 5d5862c..ea483ad 100644 --- a/src/local/mod.rs +++ b/src/local/mod.rs @@ -395,13 +395,10 @@ impl tokio::io::AsyncWrite for UdpHalf { // construction (the u16 cap — review 001 N-2), so accepting // everything is bounded memory; socket-side backpressure is // the accepted-byte high-water mark above. - let dgs = match this.deframe.reader.feed(buf) { - Ok(dgs) => dgs, - Err(e) => { - return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e))) - } - }; - for dg in dgs { + // The de-framer's feed is infallible (the oversize bound is + // enforced at frame time — see the DatagramCodecError docs): + // accept everything, bounded by the u16 cap (review 001 N-2). + for dg in this.deframe.reader.feed(buf) { this.deframe.accepted += dg.len() + 2; this.deframe.queue.push_back(dg); } @@ -465,7 +462,14 @@ impl UdpHalf { Ok(_) => { self.deframe.accepted -= dg.len() + 2; } - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + // WouldBlock (EAGAIN) and ENOBUFS both mean "the socket + // cannot take this datagram now" — the unresolved- + // neighbor + shrunken-SNDBUF state reports ENOBUFS(105); + // the stash + waker retry is the same behavior for both. + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.raw_os_error() == Some(105) => + { self.deframe.sending = Some(dg); // Register the waker for the retry. return match self.sock.poll_send_ready(cx) { diff --git a/src/wire.rs b/src/wire.rs index 1236fc6..3623a6e 100644 --- a/src/wire.rs +++ b/src/wire.rs @@ -37,6 +37,11 @@ const DATAGRAM_LEN_FIELD: usize = 2; /// (`is_mid_datagram` is the diagnostic) and the SESSION surfaces a /// stream end mid-datagram as `TunnelIoError::TruncatedDatagram` (the /// fail-loud posture, OQ-TN-13 — never a silent partial datagram). +/// +/// The decode direction (`DatagramReader::feed`) is infallible: with +/// the oversize bound enforced at frame time, a declared length can +/// never exceed what the frame itself carries, so the incremental +/// decode cannot fail — only buffer and complete. #[derive(Debug, Error)] pub enum DatagramCodecError { /// A datagram larger than [`MAX_DATAGRAM_LEN`] was framed — the @@ -89,8 +94,10 @@ impl DatagramReader { /// Feed raw bytes from the read half. Returns every datagram that /// completed with this feed, in order — zero or more per chunk. - /// Buffers partial frames across chunk boundaries. - pub fn feed(&mut self, chunk: &[u8]) -> Result, DatagramCodecError> { + /// Buffers partial frames across chunk boundaries. Infallible: the + /// decode direction cannot fail (see the [`DatagramCodecError`] + /// docs — the oversize bound is enforced at frame time). + pub fn feed(&mut self, chunk: &[u8]) -> Vec { self.buf.extend_from_slice(chunk); let mut out = Vec::new(); loop { @@ -123,7 +130,7 @@ impl DatagramReader { break; } } - Ok(out) + out } /// `true` while a partial datagram is in flight (for teardown @@ -140,7 +147,7 @@ mod tests { use super::*; fn feed_one(reader: &mut DatagramReader, chunk: &[u8]) -> Option { - reader.feed(chunk).unwrap().into_iter().next() + reader.feed(chunk).into_iter().next() } #[test] @@ -148,7 +155,7 @@ mod tests { let framed = frame_datagram(b"hello").unwrap(); assert_eq!(&framed[..2], &[0, 5]); let mut r = DatagramReader::new(); - let dgs = r.feed(&framed).unwrap(); + let dgs = r.feed(&framed); assert_eq!(dgs.len(), 1); assert_eq!(&dgs[0][..], b"hello"); } @@ -158,7 +165,7 @@ mod tests { let framed = frame_datagram(b"").unwrap(); assert_eq!(framed.len(), 2); let mut r = DatagramReader::new(); - let dgs = r.feed(&framed).unwrap(); + let dgs = r.feed(&framed); assert_eq!(dgs.len(), 1); assert!(dgs[0].is_empty()); } @@ -168,7 +175,7 @@ mod tests { let framed = frame_datagram(b"0123456789").unwrap(); let mut r = DatagramReader::new(); for piece in framed.chunks(7) { - let dgs = r.feed(piece).unwrap(); + let dgs = r.feed(piece); if !dgs.is_empty() { assert_eq!(&dgs[0][..], b"0123456789"); return; @@ -183,7 +190,7 @@ mod tests { buf.extend_from_slice(&frame_datagram(b"abc").unwrap()); buf.extend_from_slice(&frame_datagram(b"de").unwrap()); let mut r = DatagramReader::new(); - let dgs = r.feed(&buf.freeze()).unwrap(); + let dgs = r.feed(&buf.freeze()); assert_eq!(dgs.len(), 2); assert_eq!(&dgs[0][..], b"abc"); assert_eq!(&dgs[1][..], b"de"); @@ -202,9 +209,9 @@ mod tests { fn mid_datagram_state_is_observable() { let framed = frame_datagram(b"0123").unwrap(); let mut r = DatagramReader::new(); - r.feed(&framed[..3]).unwrap(); + r.feed(&framed[..3]); assert!(r.is_mid_datagram()); - r.feed(&framed[3..]).unwrap(); + r.feed(&framed[3..]); assert!(!r.is_mid_datagram()); } @@ -218,12 +225,12 @@ mod tests { fn truncated_stream_never_yields_partial_datagram() { let framed = frame_datagram(b"payload").unwrap(); let mut r = DatagramReader::new(); - r.feed(&framed[..3]).unwrap(); + r.feed(&framed[..3]); assert!(r.is_mid_datagram()); assert!(feed_one(&mut r, &[]).is_none()); assert!(r.is_mid_datagram()); assert_eq!( - r.feed(&framed[3..framed.len() - 1]).unwrap().len(), + r.feed(&framed[3..framed.len() - 1]).len(), 0, "one byte short must stay buffered, not yield a partial datagram" ); diff --git a/tasks/tunnels/coverage-weak-spots.md b/tasks/tunnels/coverage-weak-spots.md index ab90f4f..cae1f94 100644 --- a/tasks/tunnels/coverage-weak-spots.md +++ b/tasks/tunnels/coverage-weak-spots.md @@ -1,7 +1,7 @@ --- id: tunnels/coverage-weak-spots -name: "Coverage weak spots — the UdpHalf sender-blocking paths + defensive telemetry arms (2026-09-09 review residue)" -status: pending +name: "Coverage weak spots — CLOSED (2026-09-09 second pass): the UdpHalf sender-block recipe found, feed infallible, wrappers covered" +status: done depends_on: [tunnels/review-impl] scope: single risk: low @@ -10,139 +10,127 @@ level: implementation tags: [tests, coverage, udp, review-remediation] --- -## Description +## Resolution (2026-09-09, second pass) -The 2026-09-09 coverage review (cargo-llvm-cov 0.8.4, `--all-features`) -started the session at 88.28% line coverage. The inline remediation -(commit with this task file) closed the behavioral gaps — datagram -EOF paths, UDP `take_halves` framing, reverse UDP `pump_against`, -`AcceptQueue` push-after-close, the `establishment_reason` wrapper — -and removed the zombie `DatagramCodecError::InvalidLength` variant. -Coverage is now **92.86% lines / 93 regions missed → 93**; the -remaining 50 uncovered lines are inventoried below. None are -behavioral gaps reachable through a normal peer — they split into -(a) paths this host's kernel cannot reach, (b) defensive arms that -are structurally unreachable through the real dispatch, and (c) two -thin public-API wrappers. +The 2026-09-09 first pass (commits `4014ba6`, `8f14fd1`) closed the +behavioral gaps and left this task with the option-1-vs-3 decision. +The second pass found the sender-block recipe, took **option 1**, and +coverage moved **92.86% → 93.91% lines** (50 → 41 uncovered). Three +commits: -## The host-blocking finding (why the UdpHalf paths stay uncovered) +1. **`DatagramReader::feed` is infallible** — the same zombie category + as `4014ba6`'s `InvalidLength`: with the oversize bound enforced at + frame time, the incremental decode cannot fail, so the `Vec` + shape and `local/mod.rs`'s dead `InvalidData` mapping were removed + (pre-consumer window open per ADR-005's N-3). +2. **The sender-block machinery test** (`udp_half_write_side_blocks_ + stashes_and_drains_on_resolve`) — the recipe that reached the + WouldBlock stash, the readiness-Pending arms, AND the high-water + backpressure in one coherent behavior (option 1). +3. **The thin wrappers** — `connect_udp`'s resolve-failure arm + (`.invalid` NXDOMAIN → `dial_failed`) and the no-witness + `tunnel_establisher` (a harness `DialNoWitness` mode + a full + open round-trip through the wrapper). -`UdpHalf`'s write-side sender-blocking machinery needs a UDP socket -whose `try_send` returns `WouldBlock`. Probed exhaustively -2026-09-09 (kernel 6.8.0-110-generic, tokio 1.53.1, socket2 0.6.5): +## The sender-block recipe (the host-blocking finding, RESOLVED) -- **Loopback / veth (even with netem `limit 10` qdisc):** a - non-reading peer's full receive queue DROPS datagrams silently - (freeing the sender's skb), so `sk_wmem_alloc` never fills and the - sender never blocks. SO_SNDBUF shrinking (min floor 4608) does not - change this — the send path frees the skb synchronously. -- **`socket2` SO_SNDBUF:** std has no UDP buffer setters; socket2 - works but the accounting issue above makes it moot. -- **The "fresh-socket quirk":** a socket's FIRST `try_send` on a - freshly registered tokio socket reports `WouldBlock` - spuriously (~20% of runs, multi-thread runtime); exactly ONE - `poll_send_ready` + retry clears it (100% of 45/200 observed - firings). Nondeterministic across runs — used only as a tolerated - timing quirk in the flush tests, never asserted. -- **current-thread runtime + socket2 socket:** datagrams sent `Ok` - were not delivered to a bound peer (kernel probe showed - `NoPorts` increments). Delivery assertions are only trustworthy on - a multi-thread runtime — the two flush/drain tests pin that flavor - explicitly. +`UdpHalf`'s write-side machinery needed a UDP socket whose `try_send` +persistently reports `WouldBlock`/`ENOBUFS`. The recipe (validated +3/3 deterministic on this host, kernel 6.8.0-110-generic, tokio +1.53.1, socket2 0.6.5): -Unreachable as a result (the WouldBlock stash, the 128 KiB -high-water `Pending`, the `poll_recv_ready`/`poll_send_ready` -Pending arms): +- **A veth pair with the peer address UNASSIGNED + a flushed + neighbor table.** The connected sender's datagrams queue on the + unresolved neighbor while charged to `sk_wmem_alloc`; with a + shrunken `SO_SNDBUF` the charge exceeds the buffer → persistent + `EAGAIN`/`ENOBUFS` (the drain reports 105=ENOBUFS alongside + EAGAIN — both mean "the socket cannot take this datagram now", + which the `pump_queue_to_socket` stash arm now matches too). +- **Jumbo MTU (65535)** on the fixture: a 60 KB datagram crosses as + ONE frame — no IP fragmentation, so no fragment-reassembly races + at the peer (fragment-ID collisions under the drain burst merged + datagrams on an MTU-1500 veth; the peer reassembled 5-6 datagrams + into single ~43947-byte payloads). +- **The block-state probe runs on a THROWAWAY socket**: an oversized + probe datagram's partially-built skb chain stays corked in + `sk_write_queue` on EAGAIN and flushes on the next successful send — + poisoning the socket with a stale partial datagram. The test's + real socket is created fresh after the probe. +- **Assigning the peer address + binding the peer** resolves ARP; the + queued/stashed datagrams drain and DELIVER (the resolve step). +- Root-gated (`ip link add`): the fixture tries bare `ip` then + `sudo ip` and auto-skips loud-free without either. -- `src/local/mod.rs:377-383` — the accepted-byte high-water - backpressure (`WRITE_QUEUE_HIGH_WATER`): `poll_write` drains - first and Pends when the queue stays over the line. Needs a - genuinely blocking sender. -- `src/local/mod.rs:351,472` — the readiness-Pending arms inside the - recv/send WouldBlock handlers. +The first pass's kernel findings still hold and are why the loopback +recipes failed there: a non-reading peer's full receive queue DROPS +datagrams silently (loopback/veth/netem — the sender never blocks); +netem with `limit` still did not block (2M sends Ok in a probe). + +## Negative result: the ICMP ECONNREFUSED arm is unreachable through the poll drive + +A connected UDP socket + a closed peer port elicits an ICMP port +unreachable that `ECONNREFUSED`s the next `recvfrom` (a plain std +nonblocking recv surfaces it deterministically — probed rootless). +But **tokio 1.53 structurally masks it out of the poll drive** (the +adapter's shape): the `EPOLLOUT|EPOLLERR` event maps to +`WRITABLE|ERROR|WRITE_CLOSED`; `poll_recv_ready`'s Direction::Read +mask is `READABLE|READ_CLOSED` (ERROR excluded) → Pending forever; +`try_io(READABLE)` intersects empty → WouldBlock. Only the async +`recv()` (`async_io` with `READABLE|ERROR`) surfaces `ECONNREFUSED` +— and `UdpHalf::poll_read` is a poll impl. Verified by probes +(`poll_recv`, `poll_recv_ready`, `try_recv` all Elapsed; the async +`recv` surfaced it exactly once, racing the waiter registration — +nondeterministic). `local/mod.rs:356` therefore joins the defensive +list: no deterministic test can reach it through the pump's drive on +tokio 1.53. + +## The final uncovered-line inventory (41 lines, all deliberate) + +- `src/local/mod.rs:136-137` — the no-addresses resolve arm: a name + that RESOLVES to zero addresses is unreachable via getaddrinfo + (NXDOMAIN errors first — the `.invalid` test hits the error arm). - `src/local/mod.rs:321-333` — the `Oversize` mapping at the recv scratch (a >65535-byte datagram cannot exist on a real wire — the IP theoretical max is 65507). - -## Options to close them (when picked up) - -1. **Netns + real qdisc** (needs root): a network namespace with a - veth pair whose egress qdisc is netem/tbf with a small `limit` - gives a genuine holding queue. Probe result 2026-09-09: netem on - veth egress still did NOT block the sender (the qdisc's skb hold - freed fast enough) — try `tbf` with a tiny rate + burst, or - `fq_codel` with `memory_limit`, or the loopback peer with a - shrunken SO_RCVBUF AND `net.core.rmem_max` raised so drops turn - into queueing. Requires root (available on this host). -2. **LD_PRELOAD fault injector / libc shim** on `sendto` returning - `EWOULDBLOCK` once — deterministic, but test-infra heavy and - platform-specific. -3. **Accept the gap + document** (current posture): the invariants - the paths implement are pinned by review 001's U-1 remediation - and ADR-003's amendment; a reviewer reading - `src/local/mod.rs` can verify the code by inspection, and the - host probe results above are recorded here. The `sending` stash - gets partial exercise whenever the fresh-socket quirk fires in - the flush tests (not asserted). - -## The defensive arms (deliberately uncovered) - +- `src/local/mod.rs:351-352,356` — the recv-side readiness-Pending + arms (genuine WouldBlock) + the ICMP error mapping (the negative + result above). +- `src/local/mod.rs:378-379,383` — the high-water drain's + Ready(Ok)/Ready(Err) sub-arms (the observed over-line poll rides + the Pending sub-arm; the Ok sub-arm needs the drain to succeed + within that single poll — the FAILED-cycle release timing). +- `src/local/mod.rs:476-477` — `pump_queue_to_socket`'s + `poll_send_ready` Ready(Ok)/Err sub-arms (the Pend sub-arm is the + covered one). - `src/consumer.rs:91-94,405-406` — `serde_json::to_value` failures - on `TunnelParams` (impossible by construction; the error mapping - exists for the type system). -- `src/consumer.rs:205,290` — `unreachable!` guards on spent - sessions (typed-API invariants). -- `src/consumer.rs:285` — the stale-pump abort in `pump_against` - (data-plane XOR pump invariant; defensive only). + on `TunnelParams` (impossible by construction). +- `src/consumer.rs:205,290` — `unreachable!` guards on spent sessions. +- `src/consumer.rs:285` — the stale-pump abort (defensive only). - `src/consumer.rs:325-330` — the pump `JoinError` telemetry arm - (needs the spawned pump task to panic/abort — not producible - through the session API). -- `src/producer.rs:338-347` — the pump-handler telemetry - early-returns (`accept_bi` failure needs a dead channel; the plan - is always `TargetHandle` by construction; `Arc::try_unwrap` failure - needs a second clone). -- `src/producer.rs:85-86` — `TunnelEstablishError::HandlerError` → - `EstablishmentError::HandlerError` mapping: `parse_params` is the - only producer of `HandlerError`, and the registry's input-schema - gate (pinned by `schema_gate_rejects_missing_substrate_…`) rejects - malformed params before the establisher runs. The mapping is - spec-surface (ADR-049 §3's vocabulary) kept for API completeness. -- `src/wire.rs:77-79` — the `Default` impl shim (clippy - `new_without_default` requirement). -- `src/wire.rs:106` — the `unreachable!` invariant in `feed`'s - decode loop. + (needs the spawned pump to panic/abort — not producible through + the session API). +- `src/producer.rs:85-86` — the `HandlerError` mapping (the schema + gate preempts it; pinned by `schema_gate_rejects_missing_substrate_…`). +- `src/producer.rs:338-347` — the pump-handler telemetry early-returns + (`accept_bi` failure needs a dead channel; the plan is always + `TargetHandle`; `Arc::try_unwrap` needs a second clone). +- `src/wire.rs:77-79` — the `Default` impl shim (clippy's + `new_without_default`). +- `src/wire.rs:106` — the `unreachable!` invariant in `feed`'s loop. - `src/wire.rs:177` — a unit test's own panic line (the negative branch of `datagram_split_across_awkward_chunks_reassembles`). -## Thin public-API wrappers (trivial to cover if wanted) - -- `src/producer.rs:152-154` — `tunnel_establisher` (the - no-witness wrapper; tests use `tunnel_establisher_with_witness` - directly). -- `src/producer.rs:170` — the `witness: None` closure arm of - `tunnel_establisher_with_witness` (the witness-less establisher - body). One registration + open through `tunnel_establisher` - covers both. - -## Work - -- Decide: chase the sender-blocking paths with netns/tbf/qdisc - experiments (option 1) or accept the host-boundary limitation - (option 3, current posture). -- If option 1: one test forcing a real `WouldBlock` on a UDP sender - covers lines 351, 377-383, 472 in one go (the stash + high-water + - readiness-Pending machinery is one coherent behavior). -- Optionally cover the two thin wrappers (one test each, or fold - into existing suites). -- The defensive arms stay uncovered by design — do not contort the - tests to reach them; re-check this inventory if the API shapes - change. - -## Verification +## Verification (2026-09-09, second pass) - `cargo llvm-cov --all-features --workspace --summary-only` - — the post-remediation baseline (92.86% lines) should not regress. -- `cargo test --all-features` — 88 tests green as of 2026-09-09 - (15 lib + 16 consumer_session + 18 end_to_end + 13 doc/lib + - 17 local_halves + 9 producer_listen + 12 producer_open_op + 1 - doc-test compile_fail). \ No newline at end of file + — **93.91% lines** (from 92.86%; 50 → 41 uncovered, all deliberate). +- `cargo test --all-features` — **91 tests green** (15 lib + 16 + consumer_session + 18 end_to_end + 13 doc/lib + 19 local_halves + + 9 producer_listen + 13 producer_open_op + 1 doc-test compile_fail). +- `cargo test` (default) — green; the root-gated veth test skips + loud-free without root. +- `cargo clippy --all-targets --all-features -- -D warnings` and + `cargo fmt --check` — clean. +- `cargo check --target wasm32-unknown-unknown` + + `cargo clippy --target wasm32-unknown-unknown -- -D warnings` — + clean (the default crate stays wasm-clean). \ No newline at end of file diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs index 53114e2..a153793 100644 --- a/tests/end_to_end.rs +++ b/tests/end_to_end.rs @@ -714,7 +714,7 @@ async fn datagrams_survive_chunk_splitting_codec_level() { let mut reader = alktunnels::wire::DatagramReader::new(); let mut recovered = Vec::new(); for chunk in stream.chunks(7) { - for dg in reader.feed(chunk).expect("decode") { + for dg in reader.feed(chunk) { recovered.push(dg.to_vec()); } } diff --git a/tests/harness.rs b/tests/harness.rs index be21a63..112f2b4 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -102,6 +102,9 @@ pub enum RegistrationMode { /// probe — ADR-049 §2's bound override; the open fails with reason /// `timeout` and no channel survives). Timeout(std::time::Duration), + /// The no-witness dial establisher (`tunnel_establisher` — the + /// public default shape, no CF-006 probe seam). + DialNoWitness(alktunnels::producer::DialFn), } pub async fn wire_with( @@ -219,6 +222,27 @@ pub async fn wire_with( ) .expect("register tunnel listen openable"); } + RegistrationMode::DialNoWitness(dial) => { + // The public no-witness wrapper (tunnel_establisher) — + // registration + establisher body identical to the + // witnessed variant minus the CF-006 seam. + let core = alkcall::channels::operations::ChannelCore::new( + producer_client.manager().clone(), + alkcall::channels::policy::default_policy(), + ); + core.register_openable_with_establisher( + alktunnels::params::tunnel_open_spec(), + Some(alktunnels::producer::tunnel_establisher( + registry.clone(), + dial, + )), + alktunnels::producer::make_tunnel_pump_handler(), + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + None, + ) + .expect("register the no-witness wrapper establisher"); + } RegistrationMode::Timeout(timeout) => { // The deadline-expiry probe's registration shape: a hanging // establisher (never resolves) with an explicit per- @@ -527,10 +551,7 @@ pub fn closing_udp_echo_dial(payload_completes: bool) -> alktunnels::producer::D Ok(0) | Err(_) => return, Ok(n) => n, }; - let dgs = match reader.feed(&buf[..n]) { - Ok(dgs) => dgs, - Err(_) => return, - }; + let dgs = reader.feed(&buf[..n]); if let Some(dg) = dgs.into_iter().next() { let framed = match alktunnels::wire::frame_datagram(&dg) { Ok(f) => f, @@ -604,10 +625,7 @@ pub fn framed_udp_echo_dial() -> alktunnels::producer::DialFn { Ok(0) | Err(_) => return, Ok(n) => n, }; - let dgs = match reader.feed(&buf[..n]) { - Ok(dgs) => dgs, - Err(_) => return, - }; + let dgs = reader.feed(&buf[..n]); for dg in dgs { let framed = match alktunnels::wire::frame_datagram(&dg) { Ok(f) => f, diff --git a/tests/local_halves.rs b/tests/local_halves.rs index 288cd46..badf529 100644 --- a/tests/local_halves.rs +++ b/tests/local_halves.rs @@ -871,3 +871,328 @@ async fn udp_half_flush_drains_repeatedly_until_the_queue_is_empty() { assert_eq!(received, 5, "five flush cycles, five datagrams"); drop(peer); } + +/// The resolve arm (`connect_udp`'s no-addresses mapping): a target +/// whose name cannot be resolved (an RFC 6761 `.invalid` name — +/// guaranteed NXDOMAIN, resolved instantly and without network I/O) +/// maps to `dial_failed` (the resolve step's error posture, ADR-004's +/// `DialFn` contract). Rootless, deterministic. +#[tokio::test] +async fn connect_udp_unresolvable_target_maps_to_dial_failed() { + let err = match alktunnels::local::connect_udp("definitely-not-a-real-host-xyz.invalid:1").await + { + Ok(_) => panic!("an unresolvable target must fail"), + Err(e) => e, + }; + assert!( + matches!( + err, + alktunnels::producer::TunnelEstablishError::DialFailed(_) + ), + "resolve failure maps to dial_failed: {err:?}" + ); +} + +// ===================================================================== +// The sender-block machinery (the coverage-weak-spots task's option 1, +// CLOSED — the recipe was found): a veth fixture with the peer +// address UNASSIGNED and the neighbor table flushed puts the +// connected UDP sender into a PERMANENT WouldBlock/ENOSPC state +// (the unresolved neighbor never resolves; the blocked write path's +// skb stays charged to the socket's wmem with a shrunken SO_SNDBUF). +// Assigning the peer address + binding the peer resolves ARP; the +// queued/stashed datagrams drain and DELIVER. Validated on this host +// (kernel 6.8.0-110-generic, tokio 1.53.1, socket2 0.6.5) — 3/3 +// deterministic probe runs, 2026-09-09. +// +// Root-gated (`ip link add`): auto-skipped without root. Never +// touches the default route or any real interface. +// ===================================================================== + +/// The veth fixture (created and torn down per run): `alktx/alktx-p` +/// with 10.240.0.1/24 (the sender side) and 10.240.0.9 (the peer +/// side, UNASSIGNED during the block phase — its absence is what +/// pins the sender). The peer address is assigned mid-test (the +/// resolve step). A dead-simple guard skips the test when `ip` is +/// unavailable or the command fails (non-root hosts). +struct VethFixture { + peer_addr_assigned: bool, +} + +/// Run a privileged `ip` command: bare `ip` first (uid-0 test runs), +/// `sudo ip` second (the uid-1000 dev runs). `None` = unavailable. +async fn ip_cmd(args: &[&str]) -> Option { + let mut spawned = false; + for prog in ["ip", "sudo"] { + let mut argv = Vec::with_capacity(args.len() + 1); + if prog == "sudo" { + argv.push("ip"); + } + argv.extend_from_slice(args); + match tokio::process::Command::new(prog) + .args(&argv) + .output() + .await + { + Ok(o) => { + spawned = true; + if o.status.success() { + return Some(true); + } + } + Err(_) => continue, + } + } + spawned.then_some(false) +} + +impl VethFixture { + async fn try_create() -> Option { + // Skip loud-free: the coverage task documents the host boundary. + let created = ip_cmd(&[ + "link", "add", "alktx", "type", "veth", "peer", "name", "alktx-p", + ]) + .await + .unwrap_or(false); + if !created { + eprintln!("skipping: veth fixture unavailable (non-root host)"); + return None; + } + let _ = ip_cmd(&["addr", "add", "10.240.0.1/24", "dev", "alktx"]).await; + // A jumbo MTU (65536): a 60 KB datagram crosses as ONE frame — + // no IP fragmentation, so no fragment-reassembly races at the + // peer (fragment-ID collisions under the drain burst merged + // datagrams on an MTU-1500 veth during the probe runs). + let _ = ip_cmd(&["link", "set", "alktx", "mtu", "65535"]).await; + let _ = ip_cmd(&["link", "set", "alktx", "up"]).await; + let _ = ip_cmd(&["link", "set", "alktx-p", "up"]).await; + Some(Self { + peer_addr_assigned: false, + }) + } + + fn assign_peer(&mut self) -> bool { + if self.peer_addr_assigned { + return true; + } + let ok = std::process::Command::new("ip") + .args(["addr", "add", "10.240.0.9/24", "dev", "alktx-p"]) + .output() + .ok() + .map(|o| o.status.success()) + .unwrap_or(false) + || std::process::Command::new("sudo") + .args(["ip", "addr", "add", "10.240.0.9/24", "dev", "alktx-p"]) + .output() + .ok() + .map(|o| o.status.success()) + .unwrap_or(false); + self.peer_addr_assigned = ok; + ok + } +} + +impl Drop for VethFixture { + fn drop(&mut self) { + let _ = std::process::Command::new("ip") + .args(["link", "del", "alktx"]) + .output(); + let _ = std::process::Command::new("sudo") + .args(["ip", "link", "del", "alktx"]) + .output(); + } +} + +/// A sender socket blocked on the unresolved neighbor: connect to the +/// UNASSIGNED peer address, flush the neighbor table, shrunken +/// SO_SNDBUF. `try_send` on a 60 KB datagram (truesize beyond the +/// buffer) then reports WouldBlock/ENOSPC persistently — the state +/// the write-side machinery is for. Returns the tokio socket ready +/// for `UdpHalf::split`. +async fn blocked_sender_socket(peer_port: u16) -> Option { + // The BLOCK-STATE PROBE runs on a THROWAWAY socket: an oversized + // probe datagram's partially-built skb chain stays corked in + // sk_write_queue on EAGAIN (the kernel flushes it on the next + // successful send) — poisoning the socket with a stale partial + // datagram. The test socket below is created fresh afterwards. + let probe = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + ) + .ok()?; + probe.set_nonblocking(true).ok()?; + probe.set_send_buffer_size(32768).ok()?; + probe + .bind(&"10.240.0.1:0".parse::().ok()?.into()) + .ok()?; + let probe_std: std::net::UdpSocket = probe.into(); + let probe_sock = tokio::net::UdpSocket::from_std(probe_std).ok()?; + probe_sock.connect(("10.240.0.9", peer_port)).await.ok()?; + + // Confirm the block actually holds (the host invariant): with a + // 64 KiB SO_SNDBUF, the FIRST 60 KB datagram queues on the + // unresolved neighbor (Ok — its skb is charged to wmem); every + // SUBSEQUENT send EAGAINs (the charge exceeds the buffer) until + // the neighbor resolves. That persistent EAGAIN/ENOBUFS pair IS + // the block state. A resolved neighbor would send Ok throughout. + let big = vec![0u8; 60000]; + let mut blocked = 0; + for _ in 0..3 { + match probe_sock.try_send(&big) { + Ok(_) => {} // the first datagram queues; expected + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => blocked += 1, + Err(e) if e.raw_os_error() == Some(105) => blocked += 1, + Err(_) => return None, + } + } + if blocked < 1 { + return None; + } + drop(probe_sock); + + // The test socket: fresh (no corked state), same posture. + let sock = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + ) + .ok()?; + sock.set_nonblocking(true).ok()?; + sock.set_send_buffer_size(32768).ok()?; + sock.bind(&"10.240.0.1:0".parse::().ok()?.into()) + .ok()?; + let snd_std: std::net::UdpSocket = sock.into(); + let snd = tokio::net::UdpSocket::from_std(snd_std).ok()?; + snd.connect(("10.240.0.9", peer_port)).await.ok()?; + Some(snd) +} + +/// THE sender-block machinery, end to end (option 1 — one test, the +/// coherent behavior): the WouldBlock stash (`poll_write`'s +/// opportunistic drain stashes the mid-send datagram and registers +/// the send waker — `pump_queue_to_socket`'s Pending arms), the +/// accepted-byte high-water backpressure (`poll_write` Pends when the +/// queue stays over [`alktunnels::local::WRITE_QUEUE_HIGH_WATER`]), +/// the flush re-pump (readiness → drain), and the post-resolve +/// delivery (the queued datagrams reach the peer). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn udp_half_write_side_blocks_stashes_and_drains_on_resolve() { + let Some(mut veth) = VethFixture::try_create().await else { + eprintln!("skipping: veth fixture unavailable (non-root host)"); + return; + }; + + // The peer port is EPHEMERAL (stale datagrams from earlier runs + // of this test family target any fixed port; an ephemeral port + // sidesteps the leftovers). + let probe = tokio::net::UdpSocket::bind(("10.240.0.1", 0)) + .await + .expect("port probe bind"); + let peer_port = probe.local_addr().expect("port probe addr").port(); + drop(probe); + let snd = blocked_sender_socket(peer_port) + .await + .expect("blocked sender state"); + let (_read, mut write) = alktunnels::local::UdpHalf::split(snd); + + // Phase A: poll_write ACCEPTS everything (the accept-all + // contract) while the socket blocks; the queued bytes cross the + // high-water line and the next poll_write Pends. + // Drive single-shot poll_write polls with a noop waker (the + // pump's copy shape): each returns Ok (accepted) or Pending + // (the high-water). The neighbor's INCOMPLETE→FAILED cycle + // drops queued skbs periodically (charge released → sends + // succeed again), so the drive loops until a Pend is observed. + let mut body = vec![0xABu8; 60002]; + body[0] = 0; // len = 60000 BE + body[1] = 0xea; + body[2] = 0x60; + + let noop_waker = futures::task::noop_waker(); + let mut cx = std::task::Context::from_waker(&noop_waker); + + let mut pending_observed = false; + for _ in 0..60 { + let fut = std::future::poll_fn(|cx| { + tokio::io::AsyncWrite::poll_write(Pin::new(&mut write), cx, &body) + }); + let mut fut = std::pin::pin!(fut); + match std::future::Future::poll(fut.as_mut(), &mut cx) { + std::task::Poll::Ready(Ok(60002)) => {} // accepted + std::task::Poll::Ready(Ok(n)) => panic!("poll_write accepted {n}?"), + std::task::Poll::Ready(Err(e)) => panic!("poll_write error: {e:?}"), + std::task::Poll::Pending => { + pending_observed = true; + break; + } + } + } + assert!( + pending_observed, + "the high-water Pends once the queue rides over the line" + ); + + // Flush: the blocked socket → the stash + readiness-Pending arms + // (bounded retries; the resolve below unblocks the LAST one). + let mut pending_flushes = 0usize; + for _ in 0..10 { + match tokio::time::timeout( + std::time::Duration::from_millis(300), + poll_flush_once(&mut write), + ) + .await + { + Err(_elapsed) => { + pending_flushes += 1; + } // readiness stayed Pending (blocked socket) + Ok(Ok(())) => break, // drained (only possible after resolve) + Ok(Err(e)) => panic!("flush error: {e:?}"), + } + } + assert!(pending_flushes > 0, "the blocked socket Pends the flush"); + + // Phase B: resolve — assign the peer address (the neighbor + // completes), bind the peer, wait for ARP; the queued datagrams + // drain and DELIVER. + assert!(veth.assign_peer(), "assign the peer address"); + let peer = tokio::net::UdpSocket::bind(("10.240.0.9", peer_port)) + .await + .expect("bind peer (resolve)"); + + let mut flushed = false; + for _ in 0..20 { + match tokio::time::timeout( + std::time::Duration::from_millis(300), + poll_flush_once(&mut write), + ) + .await + { + Ok(Ok(())) => { + flushed = true; + break; + } + Ok(Err(e)) => panic!("post-resolve flush error: {e:?}"), + Err(_) => {} + } + } + assert!(flushed, "the queued datagrams drained after the resolve"); + + // The peer received the queued datagrams (delivery across the + // blocked→resolved window — the stash/queue never lost them; at + // least the stashed tail must arrive — the periodic + // INCOMPLETE→FAILED drops may consume some queued ones, which + // is kernel-side loss, not adapter loss). + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + let mut got = 0usize; + let mut buf = vec![0u8; 65535]; + while peer.try_recv(&mut buf).is_ok() { + got += 1; + } + assert!( + got >= 1, + "the stashed/queued datagrams delivered after resolve" + ); + + drop(peer); +} diff --git a/tests/producer_open_op.rs b/tests/producer_open_op.rs index c00ef3f..8a0a7d6 100644 --- a/tests/producer_open_op.rs +++ b/tests/producer_open_op.rs @@ -384,6 +384,50 @@ async fn registry_lookup_respects_substrate_key() { assert_eq!(establishment_reason_of(&err), Some("unknown_resource")); } +#[tokio::test] +async fn tunnel_establisher_no_witness_wrapper_opens() { + // The no-witness wrapper (`tunnel_establisher` — the public + // default shape, producer.rs's witness-less establisher body): + // a full open through the wrapper round-trips. The wrapper + // differs from `tunnel_establisher_with_witness` ONLY in the + // witness (the CF-006 probe seam), so the default-registration + // shape is exercised end to end here. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_with( + registry, + harness::RegistrationMode::DialNoWitness(echo_dial()), + Some(harness::consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + let out = call_open_params(&topo, ¶ms("echo"), None) + .await + .expect("the wrapper establisher opens"); + let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32; + let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + write.write_all(b"no-witness").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 10]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"no-witness"); + + drop(write); + drop(read); + tokio::time::timeout(std::time::Duration::from_secs(5), pump) + .await + .expect("pump completion") + .expect("pump task"); +} + /// Type-shape assertions kept for harness parity. #[allow(dead_code)] fn type_assertions() {