review-003 D-2: the composed-egress failure matrix pinned (eviction→association-end, relay exits, oversize guard)

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::<Inbound>(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::<Inbound>(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");
+    }
+}

tests/composed-egress.rs

@@ -431,3 +431,77 @@ fn composed_shapes_compile() {
     assert_send::<ComposedEgress>();
     assert_send::<alksocks::composed::SessionOwner<Box<dyn AsyncRead + Send + Unpin>>>();
 }
+
+/// 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}"
+    );
+}