diff --git a/src/composed/mod.rs b/src/composed/mod.rs index cdc2dfb..76ad5b2 100644 --- a/src/composed/mod.rs +++ b/src/composed/mod.rs @@ -207,6 +207,7 @@ pub struct ComposedEgress { /// The shared inbound queue's message: a tunnel's datagram (keyed by /// the table key — the reply address rides the entry) or a relay's /// end marker (the tunnel's EOF drained). +#[derive(Debug)] enum Inbound { Datagram(String, Bytes), Ended, @@ -860,3 +861,103 @@ fn peer_to_socket_addr(peer: &SubstrateAddr) -> SocketAddr { fn sentinel_peer() -> SocketAddr { SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0) } + +#[cfg(test)] +mod egress_unit_tests { + use super::*; + use crate::rfc1928::util::target_addr::ToTargetAddr; + use tokio::io::AsyncWriteExt; + + /// The reply-oversize guard fires (review 003 D-2): a reply whose + /// inner bytes exceed the joint bound is a `General` egress error + /// — the guard exists although the tunnel codec's u16 datagram + /// bound upstream should keep every real reply under it. + #[test] + fn inner_bytes_oversize_guard_fires() { + let addr = ("10.0.0.1", 53u16).to_target_addr().unwrap(); + let payload = vec![0u8; MAX_INNER_LEN]; + let err = inner_bytes(&addr, &payload).expect_err("the joint bound trips"); + assert!( + matches!(err, EgressError::General(ref m) if m.contains("oversize")), + "the oversize guard: {err}" + ); + let ok = inner_bytes( + &addr, + &payload[..MAX_INNER_LEN - addr.to_be_bytes().unwrap().len()], + ); + assert!(ok.is_ok(), "the bound-edge reply passes"); + } + + /// The relay task's sender-closed exit (review 003 D-2): the + /// shared queue closed with the egress — the stage exited + /// mid-relay — ends the relay's datagram pump silently (the + /// `tx.send(..).is_err()` early return), distinct from the EOF + /// exit that still reports `Ended`. + #[tokio::test] + async fn relay_read_exits_silently_when_the_queue_is_closed() { + let (tx, rx) = tokio::sync::mpsc::channel::(4); + drop(rx); + let (client_side, mut server_side) = tokio::io::duplex(1024); + server_side + .write_all(&alktunnels::wire::frame_datagram(b"queued").unwrap()) + .await + .unwrap(); + server_side.shutdown().await.unwrap(); + drop(server_side); + let handle = tokio::spawn(relay_read(client_side, tx, "k".into())); + // The relay exits on the closed-queue send (silently) — it + // joins promptly with no panic. + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("the relay exits (the closed queue ends the pump)") + .expect("no panic"); + } + + /// The relay task's EOF exit still reports `Ended` (the count's + /// decrement path — the shape the integration tests observe + /// indirectly, pinned here directly). + #[tokio::test] + async fn relay_read_reports_ended_on_eof() { + let (tx, mut rx) = tokio::sync::mpsc::channel::(4); + let (client_side, mut server_side) = tokio::io::duplex(1024); + server_side + .write_all(&alktunnels::wire::frame_datagram(b"one").unwrap()) + .await + .unwrap(); + server_side + .write_all(&alktunnels::wire::frame_datagram(b"two").unwrap()) + .await + .unwrap(); + server_side.shutdown().await.unwrap(); + drop(server_side); + let handle = tokio::spawn(relay_read(client_side, tx, "k".into())); + let first = rx.recv().await.expect("a datagram arrives"); + assert!( + matches!(first, Inbound::Datagram(ref key, ref dg) if key == "k" && (&dg[..] == b"one" || &dg[..] == b"two")), + "a queued datagram: {first:?}" + ); + // The second datagram (the batched frame) then the EOF + // marker — drain until the `Ended` marker arrives. + let mut saw_end = false; + for _ in 0..3 { + match rx.recv().await.expect("the queue delivers") { + Inbound::Datagram(key, dg) => { + assert_eq!(key, "k"); + assert!( + &dg[..] == b"one" || &dg[..] == b"two", + "a queued datagram: {dg:?}" + ); + } + Inbound::Ended => { + saw_end = true; + break; + } + } + } + assert!(saw_end, "the EOF exit reports Ended"); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("the relay exits after the EOF") + .expect("no panic"); + } +} diff --git a/tests/composed-egress.rs b/tests/composed-egress.rs index 87ec2d4..87ef2cf 100644 --- a/tests/composed-egress.rs +++ b/tests/composed-egress.rs @@ -431,3 +431,77 @@ fn composed_shapes_compile() { assert_send::(); assert_send::>>(); } + +/// The eviction-drives-association-end composition (review 003 D-2): +/// a cap-1 table whose only entry is evicted by a send whose +/// replacement dial FAILS — the table ends with zero entries and the +/// evicted tunnel's EOF drains its relay to `Ended`, so the parked +/// `next_reply` hits the closed-queue condition +/// (`ever_opened && live_relays == 0`) and returns the +/// association-end `Err` — the eviction path's own way to reach the +/// shape `closed_queue_yields_err_not_a_hot_none` pins by far-side +/// dial. Also pinned: the count's `saturating_sub` discipline (the +/// stage keeps serving a *live* relay beside the ended one — the +/// failing send does not poison the association). +#[tokio::test] +async fn eviction_to_a_failed_dial_ends_the_association_when_the_relay_ends() { + // The dial's target echoes everything (a live tunnel) until the + // consumer side EOFs; the SECOND dial fails outright (the + // replacement the eviction can never establish). + let dial: TunnelDialFn = { + let inner = counting_udp_echo_dial(Arc::new(AtomicUsize::new(0))); + let first = Arc::new(std::sync::atomic::AtomicBool::new(true)); + Arc::new(move |substrate: Substrate, backing: &str| { + let ok = first.swap(false, Ordering::SeqCst); + let inner = Arc::clone(&inner); + let backing = backing.to_string(); + Box::pin(async move { + if !ok { + return Err(TunnelEstablishError::DialFailed( + "the replacement dial fails".into(), + )); + } + inner(substrate, &backing).await + }) + }) + }; + let topo = wire_direct(dial).await; + let mut egress = egress_for(&topo, 1).await; + + // One live destination: the single tunnel (cap 1). The relay is + // live; the echo works. + round_trip(egress.as_mut(), &dest(53), b"alive").await; + + // The association still serves while the relay is live (the + // saturating_sub discipline: a live relay keeps the stage in the + // queue branch — a second datagram round-trips on the same + // tunnel). + round_trip(egress.as_mut(), &dest(53), b"again").await; + + // Evict the only entry with a send whose dial fails: the LRU + // drop closes the tunnel (the entry's session + write half go), + // the dial fails after the eviction, and the table is left + // EMPTY with the evicted relay draining its EOF. The send + // surfaces the dial failure (the datagram was not delivered). + let err = egress + .send(dest(9999), Bytes::from_static(b"evict")) + .await + .expect_err("the replacement dial fails"); + assert!( + matches!(err, EgressError::Unreachable(ref m) if m.contains("tunnel open"), + ), + "the failed replacement dial: {err}" + ); + + // The evicted relay's EOF drains → `Ended` → the count hits 0 → + // the parked `next_reply` returns the association-end `Err` + // (awaited to prove it is the Err shape, never a hot `Ok(None)`). + let result = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("the association-end surfaces (the evicted relay's EOF drains)") + .expect_err("the all-relays-ended association end"); + assert!( + matches!(result, EgressError::General(ref m) if m.contains("relays ended")), + "the eviction-driven association end: {result}" + ); +}