mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 10 Wiki Activity

test: regression coverage for transport-driven reconnect eviction

Browse Source 
Extract the conn-event handling into a testable ConnEventHandler (grace window
as a field) and add tests/reconnect_eviction.rs covering:
- second outage after a reconnect still evicts (the bug fixed in bc1a0a2;
  proven to fail when the transport-arming is disabled)
- a reconnect within grace is not evicted
- a first-ever dial is not given an eviction timer
- a graceful leave scrubs seen_connected so a later rejoin dials cleanly

Behavior-preserving refactor: the conn-event task now builds a ConnEventHandler
and forwards each event to it; arm_grace_timer takes the grace Duration as a
param (production passes RECONNECT_GRACE). All existing transport/reconnect
integration tests still pass; clippy clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-06-01 05:01:03 -04:00
co-authored by Claude Opus 4.8
parent bc1a0a2b23
commit 3b7e6e076e
2 changed files with 233 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/mod.rs
+92 -46
View File
@@ -83,6 +83,7 @@ fn arm_grace_timer(
transport: &Arc<IrohTransport>,
jitter: &Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
ui_tx: &mpsc::Sender<UiEvent>,
grace: Duration,
peer_id: EndpointId,
) {
let mut timers_guard = timers.lock().unwrap();
@@ -95,7 +96,7 @@ fn arm_grace_timer(
let timers_evict = timers.clone();
let seen_evict = seen_connected.clone();
let handle = tokio::spawn(async move {
tokio::time::sleep(RECONNECT_GRACE).await;
tokio::time::sleep(grace).await;
crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
transport_evict.disconnect_peer(peer_id).await;
jitter_evict.lock().await.remove(&peer_id);
@@ -106,6 +107,87 @@ fn arm_grace_timer(
timers_guard.insert(peer_id, handle);
}
/// Handles the transport's per-peer link-state stream (`ConnEvent`): arms/cancels
/// reconnect grace timers, tracks which peers we've linked with, and forwards
/// link state to the UI. Pulled out of the conn-event task as a unit so the
/// reconnect-eviction behavior can be tested without standing up a full session.
/// Exposed (with a `grace` override) for that purpose; not part of the public API.
pub struct ConnEventHandler {
ui_tx: mpsc::Sender<UiEvent>,
grace_timers: GraceTimers,
seen_connected: SeenConnected,
transport: Arc<IrohTransport>,
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
grace: Duration,
}
impl ConnEventHandler {
pub fn new(
ui_tx: mpsc::Sender<UiEvent>,
grace_timers: GraceTimers,
seen_connected: SeenConnected,
transport: Arc<IrohTransport>,
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
) -> Self {
Self {
ui_tx,
grace_timers,
seen_connected,
transport,
jitter,
grace: RECONNECT_GRACE,
}
}
/// Override the eviction grace window. For tests that can't wait 45s.
pub fn with_grace(mut self, grace: Duration) -> Self {
self.grace = grace;
self
}
pub async fn handle(&self, event: ConnEvent) {
match event {
ConnEvent::Connecting(id) => {
// A reconnect (we've linked with this peer before): arm an eviction
// timer so a peer that never comes back is cleared even when gossip
// doesn't re-report the drop — the transport reliably re-emits this
// on every outage, gossip's NeighborDown does not. A first-ever dial
// (not yet in seen_connected) gets no timer; Connected cancels it on
// recovery.
if self.seen_connected.lock().unwrap().contains(&id) {
arm_grace_timer(
&self.grace_timers,
&self.seen_connected,
&self.transport,
&self.jitter,
&self.ui_tx,
self.grace,
id,
);
}
let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
}
ConnEvent::Connected(id) => {
// The audio link came back — the peer recovered within the grace
// window, so cancel its eviction.
cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().insert(id);
let _ = self.ui_tx.send(UiEvent::PeerConnected { id }).await;
}
ConnEvent::Left(id) => {
// The peer closed its link gracefully (intentional leave) — evict
// immediately, like a PeerLeft, instead of leaving it "reconnecting"
// until the grace timer or the slow gossip Leave.
cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().remove(&id);
self.transport.disconnect_peer(id).await;
self.jitter.lock().await.remove(&id);
let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
}
}
}
}
struct ActiveSession {
endpoint: Endpoint,
router: Router,
@@ -549,6 +631,7 @@ async fn run_core_loop(
&transport_events,
&jitter_events,
&ui_tx_events,
RECONNECT_GRACE,
peer_id,
);
}
@@ -565,53 +648,16 @@ async fn run_core_loop(
continue;
}
};
let ui_tx_conn = ui_tx.clone();
let grace_timers_conn = grace_timers.clone();
let seen_connected_conn = seen_connected.clone();
let transport_conn = transport.clone();
let jitter_conn = jitter.clone();
let conn_handler = ConnEventHandler::new(
ui_tx.clone(),
grace_timers.clone(),
seen_connected.clone(),
transport.clone(),
jitter.clone(),
);
let conn_event_task = tokio::spawn(async move {
while let Some(event) = conn_events.recv().await {
match event {
ConnEvent::Connecting(id) => {
// A reconnect (we've linked with this peer before):
// arm an eviction timer so a peer that never comes
// back is cleared even when gossip doesn't re-report
// the drop — the transport reliably re-emits this on
// every outage, gossip's NeighborDown does not. A
// first-ever dial (not yet in seen_connected) gets no
// timer; ConnEvent::Connected cancels it on recovery.
if seen_connected_conn.lock().unwrap().contains(&id) {
arm_grace_timer(
&grace_timers_conn,
&seen_connected_conn,
&transport_conn,
&jitter_conn,
&ui_tx_conn,
id,
);
}
let _ = ui_tx_conn.send(UiEvent::PeerConnecting { id }).await;
}
ConnEvent::Connected(id) => {
// The audio link came back — the peer recovered
// within the grace window, so cancel its eviction.
cancel_grace_timer(&grace_timers_conn, &id);
seen_connected_conn.lock().unwrap().insert(id);
let _ = ui_tx_conn.send(UiEvent::PeerConnected { id }).await;
}
ConnEvent::Left(id) => {
// The peer closed its link gracefully (intentional
// leave) — evict immediately, like a PeerLeft,
// instead of leaving it "reconnecting" until the
// grace timer or the slow gossip Leave.
cancel_grace_timer(&grace_timers_conn, &id);
seen_connected_conn.lock().unwrap().remove(&id);
transport_conn.disconnect_peer(id).await;
jitter_conn.lock().await.remove(&id);
let _ = ui_tx_conn.send(UiEvent::PeerLeft { id }).await;
}
}
conn_handler.handle(event).await;
}
});
tests/reconnect_eviction.rs
+141
View File
@@ -0,0 +1,141 @@
//! Regression tests for the reconnect-eviction logic in `core`'s conn-event
//! handler. The bug these guard: after a peer reconnects, a *second* sustained
//! outage never evicted it, because the grace timer was armed only by the gossip
//! `PeerConnectionLost` path — which goes silent once gossip's neighbor state is
//! stale — while the transport keeps reliably reporting the drop. The fix arms the
//! grace timer from the transport `ConnEvent::Connecting` path too. These tests
//! drive `ConnEventHandler` directly with a short grace so the second-outage
//! eviction (and the cases that must NOT evict) are verified without a live link.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use iroh::endpoint::presets;
use iroh::{Endpoint, EndpointId, RelayMode, SecretKey};
use peerspeak::core::ConnEventHandler;
use peerspeak::core::messages::UiEvent;
use peerspeak::network::ConnEvent;
use peerspeak::network::iroh_impl::IrohTransport;
/// A real (but unconnected) transport. The handler's eviction path calls
/// `disconnect_peer`, which is a safe no-op for a peer we never linked to.
async fn make_transport() -> Arc<IrohTransport> {
let endpoint = Endpoint::builder(presets::Minimal)
.secret_key(SecretKey::generate())
.relay_mode(RelayMode::Disabled)
.bind()
.await
.expect("bind endpoint");
let (transport, _audio_proto) = IrohTransport::new(endpoint);
Arc::new(transport)
}
/// A peer identity with no live endpoint — all we need to key the handler's maps.
fn fake_peer() -> EndpointId {
SecretKey::generate().public()
}
fn make_handler(
ui_tx: mpsc::Sender<UiEvent>,
transport: Arc<IrohTransport>,
grace: Duration,
) -> ConnEventHandler {
let grace_timers = Arc::new(std::sync::Mutex::new(HashMap::new()));
let seen_connected = Arc::new(std::sync::Mutex::new(HashSet::new()));
let jitter = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
ConnEventHandler::new(ui_tx, grace_timers, seen_connected, transport, jitter).with_grace(grace)
}
/// Drain UI events until a `PeerConnectionFailed` for `peer` (eviction) arrives,
/// or `within` elapses. Returns whether the eviction fired.
async fn evicted_within(
ui_rx: &mut mpsc::Receiver<UiEvent>,
peer: EndpointId,
within: Duration,
) -> bool {
let deadline = tokio::time::Instant::now() + within;
loop {
match tokio::time::timeout_at(deadline, ui_rx.recv()).await {
Ok(Some(UiEvent::PeerConnectionFailed { id })) if id == peer => return true,
Ok(Some(_)) => continue, // ignore PeerConnecting / PeerConnected / PeerLeft
Ok(None) => return false, // channel closed
Err(_) => return false, // timed out — no eviction
}
}
}
const GRACE: Duration = Duration::from_millis(200);
/// The regression: connect, drop, reconnect, then drop again — the second outage
/// must still arm a grace timer and evict. (Before the fix, only the first did.)
#[tokio::test]
async fn second_outage_after_reconnect_still_evicts() {
let (ui_tx, mut ui_rx) = mpsc::channel(100);
let h = make_handler(ui_tx, make_transport().await, GRACE);
let peer = fake_peer();
h.handle(ConnEvent::Connected(peer)).await; // initial link up
h.handle(ConnEvent::Connecting(peer)).await; // first drop -> arms
h.handle(ConnEvent::Connected(peer)).await; // reconnects -> cancels
h.handle(ConnEvent::Connecting(peer)).await; // SECOND drop -> must re-arm
assert!(
evicted_within(&mut ui_rx, peer, Duration::from_secs(2)).await,
"second outage after a reconnect must evict the peer"
);
}
/// A peer that recovers within the grace window must NOT be evicted.
#[tokio::test]
async fn reconnect_within_grace_is_not_evicted() {
let (ui_tx, mut ui_rx) = mpsc::channel(100);
let h = make_handler(ui_tx, make_transport().await, GRACE);
let peer = fake_peer();
h.handle(ConnEvent::Connected(peer)).await;
h.handle(ConnEvent::Connecting(peer)).await; // drop -> arms
h.handle(ConnEvent::Connected(peer)).await; // recovers -> cancels
assert!(
!evicted_within(&mut ui_rx, peer, GRACE * 3).await,
"a peer that reconnects within the grace window must not be evicted"
);
}
/// A first-ever dial (never connected) must not be handed an eviction timer.
#[tokio::test]
async fn first_ever_dial_is_not_evicted() {
let (ui_tx, mut ui_rx) = mpsc::channel(100);
let h = make_handler(ui_tx, make_transport().await, GRACE);
let peer = fake_peer();
// Connecting with no prior Connected: the peer isn't in `seen_connected`.
h.handle(ConnEvent::Connecting(peer)).await;
assert!(
!evicted_within(&mut ui_rx, peer, GRACE * 3).await,
"a first-ever dial must not arm an eviction timer"
);
}
/// A graceful leave must scrub `seen_connected`, so if that identity later rejoins,
/// its initial dial isn't mistaken for a reconnect and given an eviction clock.
#[tokio::test]
async fn graceful_leave_lets_a_later_rejoin_dial_cleanly() {
let (ui_tx, mut ui_rx) = mpsc::channel(100);
let h = make_handler(ui_tx, make_transport().await, GRACE);
let peer = fake_peer();
h.handle(ConnEvent::Connected(peer)).await; // linked once
h.handle(ConnEvent::Left(peer)).await; // graceful leave -> scrub seen
h.handle(ConnEvent::Connecting(peer)).await; // later rejoin's initial dial
assert!(
!evicted_within(&mut ui_rx, peer, GRACE * 3).await,
"a rejoin's initial dial after a graceful leave must not arm eviction"
);
}