261 lines
9 KiB
Rust
261 lines
9 KiB
Rust
//! Tier-2 SWIM-internal scrape, end to end (`DIAGNOSTICS_PLAN.md` T2.6 + T2.7).
|
|
//!
|
|
//! Boots two iroh drivers with `RelayMode::Disabled`, installs a full
|
|
//! diagnostics aggregator wired to an [`InMemorySink`], lets them join
|
|
//! and pump until each side sees the other Alive, then asserts that
|
|
//! the snapshot's tier-2 `swim` block carries:
|
|
//!
|
|
//! - self-describing protocol config (probe interval, suspicion
|
|
//! timeout, gossip fanout, ...),
|
|
//! - a per-peer SWIM entry for the other node, with state Alive and
|
|
//! at least one ping/ack timestamp populated,
|
|
//! - a non-empty recent-messages ring buffer that holds the ping/ack
|
|
//! exchange we just observed.
|
|
//!
|
|
//! T2.7 — discovery-resolve event emission — is exercised by the
|
|
//! second scenario, which forces a bare-key dial against a
|
|
//! deterministically-keyed unreachable peer and asserts that the
|
|
//! aggregator's sink saw the `discovery_resolve_started` /
|
|
//! `discovery_resolve_completed` Custom event pair.
|
|
|
|
#![cfg(feature = "iroh")]
|
|
|
|
mod common;
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use common::iroh::*;
|
|
use distribution::diagnostics::{
|
|
Aggregator, Event, Identity, InMemorySink, Role, SnapshotTrigger,
|
|
};
|
|
use distribution::diagnostics::iroh_introspect::IntrospectConfig;
|
|
use distribution::iroh_driver::IrohDriver;
|
|
use distribution::types::NodeId;
|
|
use iroh::{EndpointAddr, PublicKey, SecretKey};
|
|
|
|
fn install_full_diagnostics(
|
|
driver: &mut IrohDriver,
|
|
run_id: &str,
|
|
) -> (Arc<InMemorySink>, Arc<Aggregator<Arc<InMemorySink>>>) {
|
|
let sink = Arc::new(InMemorySink::new());
|
|
let identity = Identity::new(driver.node_id(), Role::stage(), run_id);
|
|
let aggregator = Arc::new(Aggregator::new(identity, sink.clone()));
|
|
driver.install_diagnostics_with_config(
|
|
aggregator.clone(),
|
|
IntrospectConfig {
|
|
scrape_interval: Duration::from_millis(50),
|
|
},
|
|
);
|
|
(sink, aggregator)
|
|
}
|
|
|
|
fn node_hex(id: &NodeId) -> String {
|
|
let mut s = String::with_capacity(64);
|
|
for b in id.0 {
|
|
s.push_str(&format!("{:02x}", b));
|
|
}
|
|
s
|
|
}
|
|
|
|
#[test]
|
|
fn two_node_cluster_produces_tier2_swim_snapshot_block() {
|
|
let mut a = make_driver();
|
|
let mut b = make_driver();
|
|
|
|
let (_sink_a, agg_a) = install_full_diagnostics(&mut a, "run-tier2-swim");
|
|
let (_sink_b, agg_b) = install_full_diagnostics(&mut b, "run-tier2-swim");
|
|
|
|
let b_addr = b.endpoint_addr();
|
|
let a_id = a.node_id();
|
|
let b_id = b.node_id();
|
|
a.join(&[b_addr]);
|
|
|
|
let converged = pump_until_pair(
|
|
&mut a,
|
|
&mut b,
|
|
Duration::from_secs(5),
|
|
|a, b| {
|
|
let a_key = PublicKey::from_bytes(&a.node_id().0).unwrap();
|
|
let b_key = PublicKey::from_bytes(&b.node_id().0).unwrap();
|
|
sees_alive(a, &b_key) && sees_alive(b, &a_key)
|
|
},
|
|
);
|
|
assert!(converged, "nodes did not converge within 5s");
|
|
|
|
// Drive a couple more pumps so the membership round-trip (ping/ack)
|
|
// and the message ring buffer pick up at least one observation
|
|
// beyond the join handshake. Twenty ticks at the default pace is
|
|
// enough to land a ping in each direction even when the suspect
|
|
// timer hasn't started.
|
|
for _ in 0..20 {
|
|
pump_one(&mut a);
|
|
pump_one(&mut b);
|
|
}
|
|
|
|
let snap_a = agg_a.snapshot(SnapshotTrigger::OnDemand);
|
|
let snap_b = agg_b.snapshot(SnapshotTrigger::OnDemand);
|
|
|
|
a.shutdown();
|
|
b.shutdown();
|
|
|
|
let swim_a = snap_a
|
|
.body
|
|
.swim
|
|
.as_ref()
|
|
.expect("A's snapshot must include the tier-2 SWIM block");
|
|
let swim_b = snap_b
|
|
.body
|
|
.swim
|
|
.as_ref()
|
|
.expect("B's snapshot must include the tier-2 SWIM block");
|
|
|
|
// Config block is self-describing — probe parameters land in the
|
|
// snapshot itself so the bundle reader doesn't have to assume
|
|
// anything about the running version.
|
|
for (label, swim) in [("A", swim_a), ("B", swim_b)] {
|
|
assert!(
|
|
swim.config.probe_interval_ticks > 0,
|
|
"{label}: probe_interval_ticks should be > 0; got {}",
|
|
swim.config.probe_interval_ticks,
|
|
);
|
|
assert!(
|
|
swim.config.suspicion_timeout_ticks > 0,
|
|
"{label}: suspicion_timeout_ticks should be > 0",
|
|
);
|
|
assert!(
|
|
swim.config.gossip_fanout_lambda > 0,
|
|
"{label}: gossip_fanout_lambda should be > 0",
|
|
);
|
|
assert!(
|
|
!swim.config.probe_mode.is_empty(),
|
|
"{label}: probe_mode should be a non-empty string",
|
|
);
|
|
}
|
|
|
|
// Self id hex reflects the node owning the snapshot.
|
|
assert_eq!(swim_a.self_node_id_hex, node_hex(&a_id));
|
|
assert_eq!(swim_b.self_node_id_hex, node_hex(&b_id));
|
|
|
|
// Each side has a SWIM entry for the other.
|
|
let b_hex = node_hex(&b_id);
|
|
let a_hex = node_hex(&a_id);
|
|
let a_view_of_b = swim_a
|
|
.peers
|
|
.iter()
|
|
.find(|p| p.peer_node_id_hex == b_hex)
|
|
.expect("A's tier-2 SWIM peers must include B");
|
|
let b_view_of_a = swim_b
|
|
.peers
|
|
.iter()
|
|
.find(|p| p.peer_node_id_hex == a_hex)
|
|
.expect("B's tier-2 SWIM peers must include A");
|
|
|
|
use distribution::diagnostics::PeerState;
|
|
assert_eq!(
|
|
a_view_of_b.state,
|
|
PeerState::Alive,
|
|
"A should see B as Alive; got {:?}",
|
|
a_view_of_b.state,
|
|
);
|
|
assert_eq!(
|
|
b_view_of_a.state,
|
|
PeerState::Alive,
|
|
"B should see A as Alive; got {:?}",
|
|
b_view_of_a.state,
|
|
);
|
|
|
|
// At least one of the ping/ack timestamps should be populated on
|
|
// each side after a converged exchange. Which specific one fires
|
|
// first depends on probe scheduling, so just require *some*
|
|
// observation.
|
|
let touched = |p: &distribution::diagnostics::Tier2SwimPeer| {
|
|
p.last_ping_sent_at_ms.is_some()
|
|
|| p.last_ack_received_at_ms.is_some()
|
|
|| p.last_ping_received_at_ms.is_some()
|
|
};
|
|
assert!(
|
|
touched(a_view_of_b),
|
|
"A's peer entry for B should have at least one ping/ack timestamp; got {a_view_of_b:#?}",
|
|
);
|
|
assert!(
|
|
touched(b_view_of_a),
|
|
"B's peer entry for A should have at least one ping/ack timestamp; got {b_view_of_a:#?}",
|
|
);
|
|
|
|
// The recent-messages ring buffer picks up at least one
|
|
// ping/ack/join observation per side. The exact mix depends on
|
|
// who probes first.
|
|
let known_kinds = ["ping", "ack", "ping_req", "indirect_ack", "join_request", "join_response"];
|
|
let saw_known = |swim: &distribution::diagnostics::Tier2SwimState| {
|
|
swim.recent_messages
|
|
.iter()
|
|
.any(|m| known_kinds.contains(&m.kind.as_str()))
|
|
};
|
|
assert!(
|
|
saw_known(swim_a),
|
|
"A's recent_messages should contain a known SWIM kind; got {:?}",
|
|
swim_a.recent_messages.iter().map(|m| &m.kind).collect::<Vec<_>>(),
|
|
);
|
|
assert!(
|
|
saw_known(swim_b),
|
|
"B's recent_messages should contain a known SWIM kind; got {:?}",
|
|
swim_b.recent_messages.iter().map(|m| &m.kind).collect::<Vec<_>>(),
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn bare_key_dial_emits_discovery_resolve_events() {
|
|
// Force a bare-key dial: build an `EndpointAddr` from a
|
|
// deterministic key with no relay and no direct addresses. With
|
|
// `RelayMode::Disabled` iroh has nothing to discover and the dial
|
|
// will fail — but the diagnostics layer should still emit the
|
|
// discovery_resolve_started / discovery_resolve_completed Custom
|
|
// events around the attempt.
|
|
let mut a = make_driver();
|
|
let (sink, _agg) = install_full_diagnostics(&mut a, "run-tier2-discovery");
|
|
|
|
let unreachable_secret = SecretKey::from_bytes(&[0xcd; 32]);
|
|
let unreachable_key = unreachable_secret.public();
|
|
let unreachable_addr = EndpointAddr::new(unreachable_key);
|
|
a.join(&[unreachable_addr]);
|
|
|
|
// Pump for a short while so the join task actually issues at
|
|
// least one dial attempt. We're not waiting for success — just
|
|
// for the event pair to appear in the sink.
|
|
let start = std::time::Instant::now();
|
|
let saw_pair = loop {
|
|
pump_one(&mut a);
|
|
let records = sink.records();
|
|
let started = records.iter().any(|r| match &r.event {
|
|
Event::Custom { kind, .. } => kind == "discovery_resolve_started",
|
|
_ => false,
|
|
});
|
|
let completed = records.iter().any(|r| match &r.event {
|
|
Event::Custom { kind, .. } => kind == "discovery_resolve_completed",
|
|
_ => false,
|
|
});
|
|
if started && completed {
|
|
break true;
|
|
}
|
|
if start.elapsed() > Duration::from_secs(15) {
|
|
break false;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(20));
|
|
};
|
|
|
|
a.shutdown();
|
|
|
|
assert!(
|
|
saw_pair,
|
|
"bare-key dial should emit both discovery_resolve_started and \
|
|
discovery_resolve_completed Custom events; got {:#?}",
|
|
sink.records()
|
|
.iter()
|
|
.filter_map(|r| match &r.event {
|
|
Event::Custom { kind, .. } => Some(kind.clone()),
|
|
_ => None,
|
|
})
|
|
.collect::<Vec<_>>(),
|
|
);
|
|
}
|