Proof-of-concept mesh agent library over iroh
Working library with real iroh connections, not an interface sketch: - persistent device identity in state.sqlite, stable across restarts - deterministic network space derived from name + secret via HKDF-SHA256, with frozen labels and unambiguous length-prefixed encoding - replaceable discovery returning unverified candidates only; static bootstrap, in-memory test backend and a composite - real iroh connections plus an explicit mutual membership proof: HMAC-SHA256 over a role-separated transcript bound to the TLS exporter, the network id and both endpoint identities - small versioned control protocol: handshake, announcement, ping/pong - multiple networks per agent with enforced isolation - automatic reconnect with bounded backoff and jitter - mandatory state vs disposable cache, with a real directory ownership lock - status snapshots, event stream and honest diagnostics 47 integration and unit tests cover the required scenarios offline on loopback. Snapshots, revocations and WireGuard are designed for and documented, not implemented. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
Claude Opus 5
@@ -0,0 +1,112 @@
|
||||
//! Events published by a running agent.
|
||||
//!
|
||||
//! Events are delivered through a bounded [`tokio::sync::broadcast`] channel.
|
||||
//! A slow subscriber is lagged, never allowed to stall the runtime.
|
||||
//!
|
||||
//! Nothing here ever carries a secret, a derived key or a handshake proof.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use iroh::EndpointId;
|
||||
|
||||
use crate::identity::NetworkId;
|
||||
use crate::net::TransportKind;
|
||||
use crate::proto::ControlMessage;
|
||||
use crate::proto::handshake::Role;
|
||||
|
||||
/// Something that happened inside the agent.
|
||||
#[derive(Debug, Clone)]
|
||||
#[non_exhaustive]
|
||||
pub enum Event {
|
||||
/// A network was activated locally.
|
||||
NetworkActivated {
|
||||
/// The network.
|
||||
network: NetworkId,
|
||||
},
|
||||
/// A network was deactivated locally.
|
||||
///
|
||||
/// This is a local deactivation only. It is not a signed revocation of
|
||||
/// membership, and it says nothing about the network's other participants.
|
||||
NetworkDeactivated {
|
||||
/// The network.
|
||||
network: NetworkId,
|
||||
},
|
||||
/// A peer completed the handshake and has an authenticated session.
|
||||
PeerConnected {
|
||||
/// The network the session belongs to.
|
||||
network: NetworkId,
|
||||
/// Authenticated peer endpoint id.
|
||||
peer: EndpointId,
|
||||
/// Which side this agent played.
|
||||
role: Role,
|
||||
/// How the connection currently reaches the peer.
|
||||
transport: TransportKind,
|
||||
/// RTT of the selected path, if iroh reported one.
|
||||
rtt: Option<Duration>,
|
||||
},
|
||||
/// A peer's session ended.
|
||||
PeerDisconnected {
|
||||
/// The network the session belonged to.
|
||||
network: NetworkId,
|
||||
/// The peer.
|
||||
peer: EndpointId,
|
||||
/// Why the session ended. Free of secrets.
|
||||
reason: String,
|
||||
},
|
||||
/// A control message arrived on an authenticated session.
|
||||
MessageReceived {
|
||||
/// The network.
|
||||
network: NetworkId,
|
||||
/// The peer that sent it.
|
||||
peer: EndpointId,
|
||||
/// The message.
|
||||
message: ControlMessage,
|
||||
},
|
||||
/// An outbound dial failed.
|
||||
///
|
||||
/// A dead candidate produces these and nothing else; other peers keep
|
||||
/// connecting normally.
|
||||
DialFailed {
|
||||
/// The network.
|
||||
network: NetworkId,
|
||||
/// The candidate that could not be reached.
|
||||
peer: EndpointId,
|
||||
/// Why it failed.
|
||||
reason: String,
|
||||
},
|
||||
/// A handshake was rejected.
|
||||
///
|
||||
/// The network is `None` when the failure happened before the peer's
|
||||
/// requested network could be resolved.
|
||||
HandshakeRejected {
|
||||
/// The network, when known.
|
||||
network: Option<NetworkId>,
|
||||
/// The peer, when known.
|
||||
peer: Option<EndpointId>,
|
||||
/// Why it was rejected.
|
||||
reason: String,
|
||||
},
|
||||
/// A message or session was rejected for violating the protocol.
|
||||
ProtocolViolation {
|
||||
/// The network, when known.
|
||||
network: Option<NetworkId>,
|
||||
/// The peer, when known.
|
||||
peer: Option<EndpointId>,
|
||||
/// What was wrong.
|
||||
reason: String,
|
||||
},
|
||||
/// The disposable cache was discarded and recreated at startup.
|
||||
CacheReset {
|
||||
/// Why it was discarded. Free of secrets.
|
||||
reason: String,
|
||||
},
|
||||
/// An IP plugin reported an error. Never fatal.
|
||||
PluginError {
|
||||
/// The network the call was scoped to.
|
||||
network: NetworkId,
|
||||
/// Plugin protocol id.
|
||||
protocol: String,
|
||||
/// The reported error.
|
||||
reason: String,
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,611 @@
|
||||
//! The agent runtime.
|
||||
//!
|
||||
//! An [`Agent`] owns one persistent identity, one iroh endpoint, one state
|
||||
//! directory and any number of networks. It is started explicitly with
|
||||
//! [`Agent::spawn`] and stopped explicitly with [`Agent::shutdown`]; it starts
|
||||
//! no runtime of its own, installs no global logger, handles no signals and
|
||||
//! never calls `process::exit`. Several agents can therefore run side by side in
|
||||
//! one process, which is exactly what the integration tests do.
|
||||
//!
|
||||
//! # Local readiness
|
||||
//!
|
||||
//! [`Agent::spawn`] returns as soon as the local agent is ready. It never waits
|
||||
//! for other participants to appear or for a relay to become reachable.
|
||||
//!
|
||||
//! # Failure containment
|
||||
//!
|
||||
//! A bad signature, a wrong secret, a malformed frame or an unknown version
|
||||
//! rejects that message or that session. It never stops another network and
|
||||
//! never stops the agent. There is no global, irreversible error flag.
|
||||
|
||||
mod events;
|
||||
mod network;
|
||||
mod session;
|
||||
mod shutdown;
|
||||
mod status;
|
||||
|
||||
pub use events::Event;
|
||||
pub use status::{
|
||||
AgentStatus, CandidateStatus, NetworkMetrics, NetworkState, NetworkStatus, PeerStatus,
|
||||
};
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Weak};
|
||||
|
||||
use iroh::{EndpointAddr, EndpointId};
|
||||
use tokio::sync::{RwLock, broadcast, mpsc, oneshot};
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::config::{AgentConfig, Limits};
|
||||
use crate::error::{Error, Result};
|
||||
use crate::identity::{DeviceIdentity, NetworkId, NetworkKeys, NetworkName, NetworkSecret};
|
||||
use crate::net::EndpointAdapter;
|
||||
use crate::proto::handshake;
|
||||
use crate::proto::message::ControlMessage;
|
||||
use crate::storage::{CacheOutcome, Storage};
|
||||
|
||||
use network::{InboundSession, NetCommand, NetworkHandle, RuntimeParams};
|
||||
use shutdown::Shutdown;
|
||||
|
||||
/// Summary of a configured network, whether or not it is running.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ConfiguredNetwork {
|
||||
/// Public network identifier.
|
||||
pub network_id: NetworkId,
|
||||
/// Network name.
|
||||
pub name: NetworkName,
|
||||
/// Whether it is activated automatically at startup.
|
||||
pub auto_start: bool,
|
||||
/// Whether it is currently running.
|
||||
pub active: bool,
|
||||
}
|
||||
|
||||
/// A running agent.
|
||||
///
|
||||
/// Cloning gives another handle to the same agent.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Agent {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Inner {
|
||||
config: AgentConfig,
|
||||
limits: Arc<Limits>,
|
||||
storage: Storage,
|
||||
identity: DeviceIdentity,
|
||||
adapter: EndpointAdapter,
|
||||
hostname: String,
|
||||
events: broadcast::Sender<Event>,
|
||||
networks: RwLock<HashMap<NetworkId, NetworkHandle>>,
|
||||
shutdown: Shutdown,
|
||||
accept_task: std::sync::Mutex<Option<JoinHandle<()>>>,
|
||||
}
|
||||
|
||||
impl Drop for Inner {
|
||||
fn drop(&mut self) {
|
||||
// Nothing here awaits; this is only a safety net for a handle that was
|
||||
// dropped without an explicit shutdown.
|
||||
self.shutdown.trigger();
|
||||
if let Ok(mut guard) = self.accept_task.lock()
|
||||
&& let Some(task) = guard.take()
|
||||
{
|
||||
task.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
/// Starts an agent.
|
||||
///
|
||||
/// Opens the state store (taking its ownership lock), restores the
|
||||
/// persistent device identity, binds the iroh endpoint and activates every
|
||||
/// configured network whose auto-start flag is set.
|
||||
pub async fn spawn(config: AgentConfig) -> Result<Self> {
|
||||
let storage = Storage::open(&config.paths)?;
|
||||
let identity = storage.device_identity().await?;
|
||||
let adapter = EndpointAdapter::bind(&config, &identity).await?;
|
||||
|
||||
let hostname = resolve_hostname(&config, &storage, identity.endpoint_id())?;
|
||||
storage.set_hostname(hostname.clone()).await?;
|
||||
|
||||
let (events, _) = broadcast::channel(config.limits.event_buffer);
|
||||
let limits = Arc::new(config.limits.clone());
|
||||
|
||||
let inner = Arc::new(Inner {
|
||||
limits,
|
||||
storage,
|
||||
identity,
|
||||
adapter,
|
||||
hostname,
|
||||
events,
|
||||
networks: RwLock::new(HashMap::new()),
|
||||
shutdown: Shutdown::new(),
|
||||
accept_task: std::sync::Mutex::new(None),
|
||||
config,
|
||||
});
|
||||
|
||||
if let CacheOutcome::Reset(reason) = inner.storage.cache_outcome().clone() {
|
||||
let _ = inner.events.send(Event::CacheReset { reason });
|
||||
}
|
||||
|
||||
let accept = tokio::spawn(accept_loop(Arc::downgrade(&inner)));
|
||||
if let Ok(mut guard) = inner.accept_task.lock() {
|
||||
*guard = Some(accept);
|
||||
}
|
||||
|
||||
let agent = Self { inner };
|
||||
|
||||
for stored in agent.inner.storage.list_networks().await? {
|
||||
if stored.auto_start {
|
||||
let keys = NetworkKeys::derive(&stored.name, &stored.secret);
|
||||
agent.activate_with_keys(keys).await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(agent)
|
||||
}
|
||||
|
||||
/// This device's persistent endpoint id.
|
||||
pub fn endpoint_id(&self) -> EndpointId {
|
||||
self.inner.identity.endpoint_id()
|
||||
}
|
||||
|
||||
/// This endpoint's dialable address as iroh currently reports it.
|
||||
pub fn endpoint_addr(&self) -> EndpointAddr {
|
||||
self.inner.adapter.addr()
|
||||
}
|
||||
|
||||
/// An address containing only the locally bound sockets.
|
||||
///
|
||||
/// Handy when relays and address lookup are disabled and peers must be
|
||||
/// handed literal addresses, as in the test suite.
|
||||
pub fn local_addr(&self) -> EndpointAddr {
|
||||
self.inner.adapter.loopback_addr()
|
||||
}
|
||||
|
||||
/// The hostname announced to peers.
|
||||
pub fn hostname(&self) -> &str {
|
||||
&self.inner.hostname
|
||||
}
|
||||
|
||||
/// The underlying iroh endpoint, for callers that need more detail.
|
||||
pub fn endpoint(&self) -> &iroh::Endpoint {
|
||||
self.inner.adapter.endpoint()
|
||||
}
|
||||
|
||||
/// Subscribes to agent events.
|
||||
///
|
||||
/// The channel is bounded; a subscriber that falls behind is lagged rather
|
||||
/// than allowed to stall the runtime.
|
||||
pub fn subscribe(&self) -> broadcast::Receiver<Event> {
|
||||
self.inner.events.subscribe()
|
||||
}
|
||||
|
||||
/// Adds a network to the persistent configuration and activates it.
|
||||
///
|
||||
/// The same `(name, secret)` always produces the same [`NetworkId`], on
|
||||
/// every device.
|
||||
pub async fn join_network(
|
||||
&self,
|
||||
name: &NetworkName,
|
||||
secret: &NetworkSecret,
|
||||
) -> Result<NetworkId> {
|
||||
let keys = NetworkKeys::derive(name, secret);
|
||||
let network_id = keys.network_id();
|
||||
self.inner
|
||||
.storage
|
||||
.upsert_network(network_id, name.clone(), secret.clone(), true)
|
||||
.await?;
|
||||
self.activate_with_keys(keys).await?;
|
||||
Ok(network_id)
|
||||
}
|
||||
|
||||
/// Activates a configured network that is currently inactive.
|
||||
pub async fn activate_network(&self, network_id: NetworkId) -> Result<()> {
|
||||
let stored = self
|
||||
.inner
|
||||
.storage
|
||||
.list_networks()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|stored| stored.network_id == network_id)
|
||||
.ok_or(Error::NetworkUnknown(network_id))?;
|
||||
let keys = NetworkKeys::derive(&stored.name, &stored.secret);
|
||||
self.inner.storage.set_auto_start(network_id, true).await?;
|
||||
self.activate_with_keys(keys).await
|
||||
}
|
||||
|
||||
async fn activate_with_keys(&self, keys: NetworkKeys) -> Result<()> {
|
||||
if self.inner.shutdown.is_triggered() {
|
||||
return Err(Error::Stopped);
|
||||
}
|
||||
let network_id = keys.network_id();
|
||||
let mut networks = self.inner.networks.write().await;
|
||||
if networks.contains_key(&network_id) {
|
||||
return Err(Error::NetworkAlreadyActive(network_id));
|
||||
}
|
||||
|
||||
let handle = network::spawn(RuntimeParams {
|
||||
keys,
|
||||
adapter: self.inner.adapter.clone(),
|
||||
storage: self.inner.storage.clone(),
|
||||
events: self.inner.events.clone(),
|
||||
limits: Arc::clone(&self.inner.limits),
|
||||
reconnect: self.inner.config.reconnect.clone(),
|
||||
discovery: self.inner.config.discovery.clone(),
|
||||
discovery_interval: self.inner.config.discovery_interval,
|
||||
plugins: self.inner.config.plugins.clone(),
|
||||
hostname: self.inner.hostname.clone(),
|
||||
});
|
||||
networks.insert(network_id, handle);
|
||||
drop(networks);
|
||||
|
||||
let _ = self.inner.events.send(Event::NetworkActivated {
|
||||
network: network_id,
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Deactivates a running network, leaving its configuration in place.
|
||||
///
|
||||
/// This is a local action. It is not a signed revocation of membership and
|
||||
/// it does not remove this agent from anyone else's view of the network.
|
||||
/// Other networks keep running.
|
||||
pub async fn deactivate_network(&self, network_id: NetworkId) -> Result<()> {
|
||||
let handle = {
|
||||
let mut networks = self.inner.networks.write().await;
|
||||
networks
|
||||
.remove(&network_id)
|
||||
.ok_or(Error::NetworkNotActive(network_id))?
|
||||
};
|
||||
handle.stop().await;
|
||||
for plugin in &self.inner.config.plugins {
|
||||
plugin.on_network_deactivated(network_id);
|
||||
}
|
||||
self.inner.storage.set_auto_start(network_id, false).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Deactivates a network if it is running and removes it from the state
|
||||
/// store together with its cached hints.
|
||||
pub async fn forget_network(&self, network_id: NetworkId) -> Result<()> {
|
||||
if self.is_active(network_id).await {
|
||||
self.deactivate_network(network_id).await?;
|
||||
}
|
||||
self.inner.storage.remove_network(network_id).await
|
||||
}
|
||||
|
||||
/// Whether a network is currently running.
|
||||
pub async fn is_active(&self, network_id: NetworkId) -> bool {
|
||||
self.inner.networks.read().await.contains_key(&network_id)
|
||||
}
|
||||
|
||||
/// Lists configured networks and whether each is running.
|
||||
pub async fn list_networks(&self) -> Result<Vec<ConfiguredNetwork>> {
|
||||
let active: Vec<NetworkId> = self.inner.networks.read().await.keys().copied().collect();
|
||||
Ok(self
|
||||
.inner
|
||||
.storage
|
||||
.list_networks()
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|stored| ConfiguredNetwork {
|
||||
network_id: stored.network_id,
|
||||
name: stored.name,
|
||||
auto_start: stored.auto_start,
|
||||
active: active.contains(&stored.network_id),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Sends a control message to one authenticated peer in one network.
|
||||
///
|
||||
/// Fails if that network is not active or if there is no authenticated
|
||||
/// session with that peer *in that network*. Being authenticated in network
|
||||
/// A never grants the right to send into network B.
|
||||
pub async fn send(
|
||||
&self,
|
||||
network_id: NetworkId,
|
||||
peer: EndpointId,
|
||||
message: ControlMessage,
|
||||
) -> Result<()> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
self.command(
|
||||
network_id,
|
||||
NetCommand::Send {
|
||||
peer,
|
||||
message,
|
||||
reply: reply_tx,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
reply_rx.await.map_err(|_| Error::Stopped)?
|
||||
}
|
||||
|
||||
/// Sends a control message to every authenticated peer in a network.
|
||||
///
|
||||
/// Returns how many sessions accepted it into their outbound queue.
|
||||
pub async fn broadcast(&self, network_id: NetworkId, message: ControlMessage) -> Result<usize> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
self.command(
|
||||
network_id,
|
||||
NetCommand::Broadcast {
|
||||
message,
|
||||
reply: reply_tx,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
reply_rx.await.map_err(|_| Error::Stopped)
|
||||
}
|
||||
|
||||
/// Status of one running network.
|
||||
pub async fn network_status(&self, network_id: NetworkId) -> Result<NetworkStatus> {
|
||||
let (reply_tx, reply_rx) = oneshot::channel();
|
||||
self.command(network_id, NetCommand::Status { reply: reply_tx })
|
||||
.await?;
|
||||
reply_rx
|
||||
.await
|
||||
.map(|boxed| *boxed)
|
||||
.map_err(|_| Error::Stopped)
|
||||
}
|
||||
|
||||
/// Status of the whole agent, including configured but inactive networks.
|
||||
pub async fn status(&self) -> Result<AgentStatus> {
|
||||
let endpoint = self.inner.adapter.snapshot();
|
||||
let active: Vec<NetworkId> = self.inner.networks.read().await.keys().copied().collect();
|
||||
|
||||
let mut networks = Vec::new();
|
||||
for stored in self.inner.storage.list_networks().await? {
|
||||
if active.contains(&stored.network_id)
|
||||
&& let Ok(status) = self.network_status(stored.network_id).await
|
||||
{
|
||||
networks.push(status);
|
||||
continue;
|
||||
}
|
||||
let keys = NetworkKeys::derive(&stored.name, &stored.secret);
|
||||
networks.push(NetworkStatus {
|
||||
descriptor: keys.descriptor(),
|
||||
name: stored.name,
|
||||
network_id: stored.network_id,
|
||||
state: NetworkState::Inactive,
|
||||
peers: Vec::new(),
|
||||
candidates: Vec::new(),
|
||||
metrics: NetworkMetrics::default(),
|
||||
});
|
||||
}
|
||||
|
||||
Ok(AgentStatus {
|
||||
endpoint_id: endpoint.endpoint_id,
|
||||
hostname: self.inner.hostname.clone(),
|
||||
bound_sockets: endpoint.bound_sockets,
|
||||
observed_addrs: endpoint.observed_addrs,
|
||||
endpoint_addr: self.inner.adapter.addr(),
|
||||
cache_outcome: self.inner.storage.cache_outcome().clone(),
|
||||
cache_healthy: self.inner.storage.cache_healthy(),
|
||||
networks,
|
||||
})
|
||||
}
|
||||
|
||||
/// Asks one network to re-run discovery and re-evaluate dials right now.
|
||||
///
|
||||
/// Call this when the host's network environment changed. Platform wake-up
|
||||
/// notifications can be wired to it later.
|
||||
pub async fn recheck_network(&self, network_id: NetworkId) -> Result<()> {
|
||||
self.command(network_id, NetCommand::Recheck).await
|
||||
}
|
||||
|
||||
/// Asks every running network to re-run discovery right now.
|
||||
pub async fn recheck(&self) {
|
||||
let senders: Vec<mpsc::Sender<NetCommand>> = self
|
||||
.inner
|
||||
.networks
|
||||
.read()
|
||||
.await
|
||||
.values()
|
||||
.map(|handle| handle.commands.clone())
|
||||
.collect();
|
||||
for sender in senders {
|
||||
let _ = sender.send(NetCommand::Recheck).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Stops every network, the accept loop and the endpoint.
|
||||
///
|
||||
/// After this returns, the state directory can be opened by another agent.
|
||||
pub async fn shutdown(&self) {
|
||||
self.inner.shutdown.trigger();
|
||||
|
||||
let handles: Vec<NetworkHandle> = {
|
||||
let mut networks = self.inner.networks.write().await;
|
||||
networks.drain().map(|(_, handle)| handle).collect()
|
||||
};
|
||||
for handle in handles {
|
||||
handle.stop().await;
|
||||
}
|
||||
|
||||
self.inner.adapter.close().await;
|
||||
|
||||
let task = self
|
||||
.inner
|
||||
.accept_task
|
||||
.lock()
|
||||
.ok()
|
||||
.and_then(|mut guard| guard.take());
|
||||
if let Some(task) = task {
|
||||
let _ = task.await;
|
||||
}
|
||||
|
||||
// Release the directory so another instance can claim it right away.
|
||||
self.inner.storage.release_ownership_lock();
|
||||
}
|
||||
|
||||
async fn command(&self, network_id: NetworkId, command: NetCommand) -> Result<()> {
|
||||
let sender = {
|
||||
let networks = self.inner.networks.read().await;
|
||||
networks
|
||||
.get(&network_id)
|
||||
.map(|handle| handle.commands.clone())
|
||||
.ok_or(Error::NetworkNotActive(network_id))?
|
||||
};
|
||||
sender.send(command).await.map_err(|_| Error::Stopped)
|
||||
}
|
||||
}
|
||||
|
||||
/// Accepts inbound connections and routes authenticated sessions to networks.
|
||||
///
|
||||
/// Holds only a weak reference, so dropping every [`Agent`] handle lets the
|
||||
/// runtime state be released and this loop exit.
|
||||
async fn accept_loop(weak: Weak<Inner>) {
|
||||
let Some(inner) = weak.upgrade() else {
|
||||
return;
|
||||
};
|
||||
let endpoint = inner.adapter.endpoint().clone();
|
||||
let shutdown = inner.shutdown.clone();
|
||||
let permits = Arc::new(tokio::sync::Semaphore::new(
|
||||
inner.limits.max_inbound_handshakes,
|
||||
));
|
||||
drop(inner);
|
||||
|
||||
loop {
|
||||
let incoming = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => break,
|
||||
incoming = endpoint.accept() => match incoming {
|
||||
Some(incoming) => incoming,
|
||||
None => break,
|
||||
},
|
||||
};
|
||||
|
||||
let Some(inner) = weak.upgrade() else {
|
||||
break;
|
||||
};
|
||||
let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else {
|
||||
// Too many handshakes in flight: refuse cheaply instead of queueing.
|
||||
incoming.refuse();
|
||||
continue;
|
||||
};
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _permit = permit;
|
||||
handle_incoming(inner, incoming).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_incoming(inner: Arc<Inner>, incoming: iroh::endpoint::Incoming) {
|
||||
let connecting = match incoming.accept() {
|
||||
Ok(connecting) => connecting,
|
||||
Err(err) => {
|
||||
tracing::debug!(%err, "inbound connection could not be accepted");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let conn = match connecting.await {
|
||||
Ok(conn) => conn,
|
||||
Err(err) => {
|
||||
tracing::debug!(%err, "inbound connection failed during setup");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let peer = conn.remote_id();
|
||||
|
||||
let (mut send, mut recv) = match conn.accept_bi().await {
|
||||
Ok(streams) => streams,
|
||||
Err(err) => {
|
||||
tracing::debug!(%err, "peer did not open a control stream");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Snapshot the active networks so the handshake's lookup stays synchronous.
|
||||
let known: HashMap<NetworkId, NetworkKeys> = inner
|
||||
.networks
|
||||
.read()
|
||||
.await
|
||||
.iter()
|
||||
.map(|(id, handle)| (*id, handle.keys.clone()))
|
||||
.collect();
|
||||
|
||||
let local_id = inner.identity.endpoint_id();
|
||||
let outcome = handshake::respond(
|
||||
&conn,
|
||||
&mut send,
|
||||
&mut recv,
|
||||
local_id,
|
||||
&inner.limits,
|
||||
|network_id| known.get(&network_id).cloned(),
|
||||
)
|
||||
.await;
|
||||
|
||||
let outcome = match outcome {
|
||||
Ok(outcome) => outcome,
|
||||
Err(err) => {
|
||||
// The network id is deliberately not reported here: before a
|
||||
// successful handshake the peer's claim is unverified.
|
||||
let network = None;
|
||||
conn.close(2u32.into(), b"handshake rejected");
|
||||
let _ = inner.events.send(Event::HandshakeRejected {
|
||||
network,
|
||||
peer: Some(peer),
|
||||
reason: err.to_string(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let sender = {
|
||||
let networks = inner.networks.read().await;
|
||||
networks
|
||||
.get(&outcome.network_id)
|
||||
.map(|handle| handle.commands.clone())
|
||||
};
|
||||
let Some(sender) = sender else {
|
||||
// The network was deactivated while the handshake ran.
|
||||
conn.close(3u32.into(), b"network no longer active");
|
||||
return;
|
||||
};
|
||||
|
||||
let inbound = InboundSession {
|
||||
conn,
|
||||
send,
|
||||
recv,
|
||||
outcome,
|
||||
};
|
||||
if sender
|
||||
.send(NetCommand::Inbound(Box::new(inbound)))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::debug!("network runtime stopped before the session could be installed");
|
||||
}
|
||||
}
|
||||
|
||||
/// Picks the hostname to announce.
|
||||
///
|
||||
/// Order: explicit configuration, then what the state store already holds, then
|
||||
/// a best-effort environment variable, then a stable fallback derived from the
|
||||
/// endpoint id. The library does not shell out to discover a hostname.
|
||||
fn resolve_hostname(
|
||||
config: &AgentConfig,
|
||||
storage: &Storage,
|
||||
endpoint_id: EndpointId,
|
||||
) -> Result<String> {
|
||||
if let Some(hostname) = &config.hostname {
|
||||
return Ok(hostname.clone());
|
||||
}
|
||||
if let Some(stored) = storage.hostname_blocking()?
|
||||
&& !stored.is_empty()
|
||||
{
|
||||
return Ok(stored);
|
||||
}
|
||||
for key in ["HOSTNAME", "COMPUTERNAME"] {
|
||||
if let Ok(value) = std::env::var(key) {
|
||||
let value = value.trim();
|
||||
if !value.is_empty() {
|
||||
return Ok(value.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(format!("tsunagi-{}", endpoint_id.fmt_short()))
|
||||
}
|
||||
@@ -0,0 +1,856 @@
|
||||
//! The per-network runtime.
|
||||
//!
|
||||
//! One of these runs for every locally active network. It owns that network's
|
||||
//! sessions, its dial loop and its counters. Everything it touches carries an
|
||||
//! explicit [`NetworkId`], so deactivating or breaking one network cannot
|
||||
//! disturb another and cannot stop the agent.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use iroh::EndpointId;
|
||||
use iroh::endpoint::{Connection, RecvStream, SendStream};
|
||||
use tokio::sync::{broadcast, mpsc, oneshot};
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::config::{Limits, ReconnectPolicy};
|
||||
use crate::dataplane::{PluginCapability, SharedPlugin};
|
||||
use crate::discovery::{Candidate, CandidateSource, NetworkDiscovery};
|
||||
use crate::error::{Error, Result};
|
||||
use crate::identity::{NetworkId, NetworkKeys};
|
||||
use crate::net::{EndpointAdapter, PathAddr, snapshot_connection};
|
||||
use crate::proto::handshake::{self, HandshakeOutcome, Role};
|
||||
use crate::proto::message::{Announcement, ControlMessage, Envelope, encode, kind};
|
||||
use crate::storage::Storage;
|
||||
|
||||
use super::events::Event;
|
||||
use super::session::{self, Session, SessionEvent};
|
||||
use super::shutdown::Shutdown;
|
||||
use super::status::{CandidateStatus, NetworkMetrics, NetworkState, NetworkStatus, PeerStatus};
|
||||
|
||||
/// An inbound connection that already passed the handshake.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct InboundSession {
|
||||
pub(crate) conn: Connection,
|
||||
pub(crate) send: SendStream,
|
||||
pub(crate) recv: RecvStream,
|
||||
pub(crate) outcome: HandshakeOutcome,
|
||||
}
|
||||
|
||||
/// Commands accepted by a network runtime.
|
||||
pub(crate) enum NetCommand {
|
||||
Inbound(Box<InboundSession>),
|
||||
Send {
|
||||
peer: EndpointId,
|
||||
message: ControlMessage,
|
||||
reply: oneshot::Sender<Result<()>>,
|
||||
},
|
||||
Broadcast {
|
||||
message: ControlMessage,
|
||||
reply: oneshot::Sender<usize>,
|
||||
},
|
||||
Status {
|
||||
reply: oneshot::Sender<Box<NetworkStatus>>,
|
||||
},
|
||||
Recheck,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for NetCommand {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
NetCommand::Inbound(_) => f.write_str("Inbound"),
|
||||
NetCommand::Send { peer, message, .. } => {
|
||||
write!(f, "Send({}, {})", peer.fmt_short(), kind(message))
|
||||
}
|
||||
NetCommand::Broadcast { message, .. } => write!(f, "Broadcast({})", kind(message)),
|
||||
NetCommand::Status { .. } => f.write_str("Status"),
|
||||
NetCommand::Recheck => f.write_str("Recheck"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle to a running network runtime.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct NetworkHandle {
|
||||
pub(crate) keys: NetworkKeys,
|
||||
pub(crate) commands: mpsc::Sender<NetCommand>,
|
||||
shutdown: Shutdown,
|
||||
task: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl NetworkHandle {
|
||||
/// Stops the runtime and waits for its task to finish.
|
||||
pub(crate) async fn stop(self) {
|
||||
self.shutdown.trigger();
|
||||
let _ = self.task.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything a network runtime needs to run.
|
||||
pub(crate) struct RuntimeParams {
|
||||
pub(crate) keys: NetworkKeys,
|
||||
pub(crate) adapter: EndpointAdapter,
|
||||
pub(crate) storage: Storage,
|
||||
pub(crate) events: broadcast::Sender<Event>,
|
||||
pub(crate) limits: Arc<Limits>,
|
||||
pub(crate) reconnect: ReconnectPolicy,
|
||||
pub(crate) discovery: Option<Arc<dyn NetworkDiscovery>>,
|
||||
pub(crate) discovery_interval: Duration,
|
||||
pub(crate) plugins: Vec<SharedPlugin>,
|
||||
pub(crate) hostname: String,
|
||||
}
|
||||
|
||||
/// Outcome of one outbound dial.
|
||||
enum DialOutcome {
|
||||
Established(Box<InboundSession>),
|
||||
Failed {
|
||||
peer: EndpointId,
|
||||
reason: String,
|
||||
during_handshake: bool,
|
||||
},
|
||||
}
|
||||
|
||||
/// Backoff bookkeeping for one candidate.
|
||||
#[derive(Debug)]
|
||||
struct DialState {
|
||||
consecutive_failures: u32,
|
||||
next_attempt: Instant,
|
||||
in_flight: bool,
|
||||
source: CandidateSource,
|
||||
}
|
||||
|
||||
impl DialState {
|
||||
fn new(source: CandidateSource) -> Self {
|
||||
Self {
|
||||
consecutive_failures: 0,
|
||||
next_attempt: Instant::now(),
|
||||
in_flight: false,
|
||||
source,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts a network runtime.
|
||||
pub(crate) fn spawn(params: RuntimeParams) -> NetworkHandle {
|
||||
let keys = params.keys.clone();
|
||||
let shutdown = Shutdown::new();
|
||||
let (commands_tx, commands_rx) = mpsc::channel(64);
|
||||
|
||||
let runtime_shutdown = shutdown.clone();
|
||||
let task = tokio::spawn(async move {
|
||||
let mut runtime = Runtime::new(params, runtime_shutdown);
|
||||
runtime.run(commands_rx).await;
|
||||
});
|
||||
|
||||
NetworkHandle {
|
||||
keys,
|
||||
commands: commands_tx,
|
||||
shutdown,
|
||||
task,
|
||||
}
|
||||
}
|
||||
|
||||
struct Runtime {
|
||||
params: RuntimeParams,
|
||||
network_id: NetworkId,
|
||||
local_id: EndpointId,
|
||||
shutdown: Shutdown,
|
||||
sessions: HashMap<EndpointId, Session>,
|
||||
dial_states: HashMap<EndpointId, DialState>,
|
||||
candidate_addrs: HashMap<EndpointId, iroh::EndpointAddr>,
|
||||
metrics: NetworkMetrics,
|
||||
session_events_tx: mpsc::Sender<SessionEvent>,
|
||||
session_events_rx: mpsc::Receiver<SessionEvent>,
|
||||
dial_results_tx: mpsc::Sender<DialOutcome>,
|
||||
dial_results_rx: mpsc::Receiver<DialOutcome>,
|
||||
}
|
||||
|
||||
impl Runtime {
|
||||
fn new(params: RuntimeParams, shutdown: Shutdown) -> Self {
|
||||
let network_id = params.keys.network_id();
|
||||
let local_id = params.adapter.endpoint_id();
|
||||
let (session_events_tx, session_events_rx) = mpsc::channel(256);
|
||||
let (dial_results_tx, dial_results_rx) = mpsc::channel(64);
|
||||
Self {
|
||||
params,
|
||||
network_id,
|
||||
local_id,
|
||||
shutdown,
|
||||
sessions: HashMap::new(),
|
||||
dial_states: HashMap::new(),
|
||||
candidate_addrs: HashMap::new(),
|
||||
metrics: NetworkMetrics::default(),
|
||||
session_events_tx,
|
||||
session_events_rx,
|
||||
dial_results_tx,
|
||||
dial_results_rx,
|
||||
}
|
||||
}
|
||||
|
||||
fn emit(&self, event: Event) {
|
||||
// A broadcast with no subscribers is not an error.
|
||||
let _ = self.params.events.send(event);
|
||||
}
|
||||
|
||||
async fn run(&mut self, mut commands: mpsc::Receiver<NetCommand>) {
|
||||
let mut ticker = tokio::time::interval(self.params.discovery_interval);
|
||||
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = self.shutdown.wait() => break,
|
||||
command = commands.recv() => match command {
|
||||
Some(command) => self.handle_command(command).await,
|
||||
None => break,
|
||||
},
|
||||
event = self.session_events_rx.recv() => {
|
||||
if let Some(event) = event {
|
||||
self.handle_session_event(event).await;
|
||||
}
|
||||
}
|
||||
result = self.dial_results_rx.recv() => {
|
||||
if let Some(result) = result {
|
||||
self.handle_dial_result(result).await;
|
||||
}
|
||||
}
|
||||
_ = ticker.tick() => self.discovery_round().await,
|
||||
}
|
||||
}
|
||||
|
||||
self.teardown().await;
|
||||
}
|
||||
|
||||
async fn teardown(&mut self) {
|
||||
// Stop accepting session events first: nothing is going to act on them
|
||||
// any more, and a sender blocked on a full queue would stall shutdown.
|
||||
self.session_events_rx.close();
|
||||
if let Some(discovery) = &self.params.discovery {
|
||||
let _ = discovery
|
||||
.unpublish(self.params.keys.discovery_key(), self.local_id)
|
||||
.await;
|
||||
}
|
||||
let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
|
||||
for peer in peers {
|
||||
if let Some(session) = self.sessions.remove(&peer) {
|
||||
session.stop().await;
|
||||
}
|
||||
}
|
||||
self.emit(Event::NetworkDeactivated {
|
||||
network: self.network_id,
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- commands
|
||||
|
||||
async fn handle_command(&mut self, command: NetCommand) {
|
||||
match command {
|
||||
NetCommand::Inbound(inbound) => {
|
||||
self.install_session(*inbound).await;
|
||||
}
|
||||
NetCommand::Send {
|
||||
peer,
|
||||
message,
|
||||
reply,
|
||||
} => {
|
||||
let _ = reply.send(self.send_to(peer, message));
|
||||
}
|
||||
NetCommand::Broadcast { message, reply } => {
|
||||
let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
|
||||
let mut delivered = 0;
|
||||
for peer in peers {
|
||||
if self.send_to(peer, message.clone()).is_ok() {
|
||||
delivered += 1;
|
||||
}
|
||||
}
|
||||
let _ = reply.send(delivered);
|
||||
}
|
||||
NetCommand::Status { reply } => {
|
||||
let _ = reply.send(Box::new(self.status()));
|
||||
}
|
||||
NetCommand::Recheck => self.discovery_round().await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Queues a message without blocking the runtime loop.
|
||||
///
|
||||
/// The envelope is encoded here so that the exact number of control bytes
|
||||
/// handed to the transport is known and can be reported honestly.
|
||||
///
|
||||
/// A full queue is backpressure: the send fails rather than stalling every
|
||||
/// other peer in this network.
|
||||
fn send_to(&mut self, peer: EndpointId, message: ControlMessage) -> Result<()> {
|
||||
let network = self.network_id;
|
||||
let envelope = Envelope {
|
||||
network_id: *network.as_bytes(),
|
||||
message,
|
||||
};
|
||||
let encoded = encode(&envelope)?;
|
||||
let bytes = encoded.len() as u64;
|
||||
|
||||
let session = self.sessions.get_mut(&peer).ok_or(Error::NoSuchPeer {
|
||||
network,
|
||||
peer: peer.fmt_short().to_string(),
|
||||
})?;
|
||||
|
||||
match session.outbound.try_send(encoded) {
|
||||
Ok(()) => {
|
||||
session.messages_sent += 1;
|
||||
session.bytes_sent += bytes;
|
||||
self.metrics.control_messages_sent += 1;
|
||||
self.metrics.control_bytes_sent += bytes;
|
||||
Ok(())
|
||||
}
|
||||
Err(mpsc::error::TrySendError::Full(_)) => Err(Error::Storage(format!(
|
||||
"outbound queue for peer {} is full",
|
||||
peer.fmt_short()
|
||||
))),
|
||||
Err(mpsc::error::TrySendError::Closed(_)) => Err(Error::NoSuchPeer {
|
||||
network,
|
||||
peer: peer.fmt_short().to_string(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------- discovery
|
||||
|
||||
async fn discovery_round(&mut self) {
|
||||
if self.shutdown.is_triggered() {
|
||||
return;
|
||||
}
|
||||
|
||||
let mut candidates: Vec<Candidate> = Vec::new();
|
||||
|
||||
if let Some(discovery) = self.params.discovery.clone() {
|
||||
let key = self.params.keys.discovery_key();
|
||||
// Publishing every round keeps a restarted agent reachable at its
|
||||
// new local port without any special case.
|
||||
if let Err(err) = discovery.publish(key, self.params.adapter.addr()).await {
|
||||
tracing::debug!(%err, "discovery publish failed");
|
||||
}
|
||||
if let Err(err) = discovery
|
||||
.publish(key, self.params.adapter.loopback_addr())
|
||||
.await
|
||||
{
|
||||
tracing::debug!(%err, "discovery publish of bound sockets failed");
|
||||
}
|
||||
match discovery.resolve(key).await {
|
||||
Ok(found) => candidates.extend(found),
|
||||
Err(err) => tracing::debug!(%err, "discovery resolve failed"),
|
||||
}
|
||||
}
|
||||
|
||||
// A stale or missing cache only changes which candidates we try first.
|
||||
// It never bypasses authentication.
|
||||
for hint in self.params.storage.hints_for_network(self.network_id).await {
|
||||
let Ok(endpoint_id) = EndpointId::from_bytes(&hint.endpoint_id) else {
|
||||
continue;
|
||||
};
|
||||
let Some(addr) = decode_hint(endpoint_id, &hint.addr) else {
|
||||
continue;
|
||||
};
|
||||
candidates.push(Candidate::new(addr, CandidateSource::Cache));
|
||||
}
|
||||
|
||||
for candidate in candidates {
|
||||
let peer = candidate.endpoint_id();
|
||||
if peer == self.local_id {
|
||||
continue;
|
||||
}
|
||||
self.candidate_addrs
|
||||
.entry(peer)
|
||||
.and_modify(|existing| merge_addr(existing, &candidate.addr))
|
||||
.or_insert_with(|| candidate.addr.clone());
|
||||
self.dial_states
|
||||
.entry(peer)
|
||||
.or_insert_with(|| DialState::new(candidate.source));
|
||||
}
|
||||
|
||||
self.start_dials();
|
||||
}
|
||||
|
||||
fn start_dials(&mut self) {
|
||||
let now = Instant::now();
|
||||
let in_flight = self
|
||||
.dial_states
|
||||
.values()
|
||||
.filter(|state| state.in_flight)
|
||||
.count();
|
||||
let mut budget = self
|
||||
.params
|
||||
.limits
|
||||
.max_concurrent_dials
|
||||
.saturating_sub(in_flight);
|
||||
if budget == 0 || self.sessions.len() >= self.params.limits.max_sessions_per_network {
|
||||
return;
|
||||
}
|
||||
|
||||
let ready: Vec<EndpointId> = self
|
||||
.dial_states
|
||||
.iter()
|
||||
.filter(|(peer, state)| {
|
||||
!state.in_flight && state.next_attempt <= now && !self.sessions.contains_key(*peer)
|
||||
})
|
||||
.map(|(peer, _)| *peer)
|
||||
.collect();
|
||||
|
||||
for peer in ready {
|
||||
if budget == 0 {
|
||||
break;
|
||||
}
|
||||
let Some(addr) = self.candidate_addrs.get(&peer).cloned() else {
|
||||
continue;
|
||||
};
|
||||
if let Some(state) = self.dial_states.get_mut(&peer) {
|
||||
state.in_flight = true;
|
||||
}
|
||||
budget -= 1;
|
||||
self.metrics.dial_attempts += 1;
|
||||
|
||||
let adapter = self.params.adapter.clone();
|
||||
let keys = self.params.keys.clone();
|
||||
let limits = Arc::clone(&self.params.limits);
|
||||
let results = self.dial_results_tx.clone();
|
||||
let local_id = self.local_id;
|
||||
let shutdown = self.shutdown.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let outcome = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => DialOutcome::Failed {
|
||||
peer,
|
||||
reason: "network deactivated".into(),
|
||||
during_handshake: false,
|
||||
},
|
||||
outcome = dial(adapter, addr, keys, limits, local_id, peer) => outcome,
|
||||
};
|
||||
let _ = results.send(outcome).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_dial_result(&mut self, result: DialOutcome) {
|
||||
match result {
|
||||
DialOutcome::Established(inbound) => {
|
||||
let peer = inbound.outcome.peer;
|
||||
if let Some(state) = self.dial_states.get_mut(&peer) {
|
||||
state.in_flight = false;
|
||||
state.consecutive_failures = 0;
|
||||
state.next_attempt = Instant::now();
|
||||
}
|
||||
self.install_session(*inbound).await;
|
||||
}
|
||||
DialOutcome::Failed {
|
||||
peer,
|
||||
reason,
|
||||
during_handshake,
|
||||
} => {
|
||||
self.metrics.dial_failures += 1;
|
||||
if during_handshake {
|
||||
self.metrics.handshake_failures += 1;
|
||||
}
|
||||
let give_up = {
|
||||
let policy = &self.params.reconnect;
|
||||
let state = self
|
||||
.dial_states
|
||||
.entry(peer)
|
||||
.or_insert_with(|| DialState::new(CandidateSource::Discovery));
|
||||
state.in_flight = false;
|
||||
state.consecutive_failures = state.consecutive_failures.saturating_add(1);
|
||||
let delay = policy.delay_for(state.consecutive_failures);
|
||||
state.next_attempt = Instant::now() + delay;
|
||||
policy
|
||||
.max_consecutive_failures
|
||||
.is_some_and(|max| state.consecutive_failures >= max)
|
||||
};
|
||||
if give_up {
|
||||
// Keep the backoff entry but push it far out; discovery
|
||||
// seeing the peer again resets it.
|
||||
if let Some(state) = self.dial_states.get_mut(&peer) {
|
||||
state.next_attempt = Instant::now() + self.params.reconnect.max_delay;
|
||||
}
|
||||
}
|
||||
self.emit(Event::DialFailed {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------- sessions
|
||||
|
||||
async fn install_session(&mut self, inbound: InboundSession) {
|
||||
let InboundSession {
|
||||
conn,
|
||||
send,
|
||||
recv,
|
||||
outcome,
|
||||
} = inbound;
|
||||
let peer = outcome.peer;
|
||||
|
||||
if self.sessions.len() >= self.params.limits.max_sessions_per_network
|
||||
&& !self.sessions.contains_key(&peer)
|
||||
{
|
||||
conn.close(1u32.into(), b"session limit reached");
|
||||
self.emit(Event::ProtocolViolation {
|
||||
network: Some(self.network_id),
|
||||
peer: Some(peer),
|
||||
reason: "session limit for this network reached".into(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Two agents may dial each other at the same time. Both sides apply the
|
||||
// same deterministic rule, so they converge on the same session.
|
||||
if let Some(existing) = self.sessions.get(&peer) {
|
||||
let existing_initiator = initiator_of(existing.role, self.local_id, peer);
|
||||
let new_initiator = initiator_of(outcome.role, self.local_id, peer);
|
||||
if existing_initiator.as_bytes() <= new_initiator.as_bytes() {
|
||||
conn.close(0u32.into(), b"duplicate session");
|
||||
return;
|
||||
}
|
||||
if let Some(old) = self.sessions.remove(&peer) {
|
||||
old.abort();
|
||||
old.conn
|
||||
.close(0u32.into(), b"replaced by preferred session");
|
||||
}
|
||||
}
|
||||
|
||||
self.record_hints(&conn).await;
|
||||
|
||||
let session = session::spawn(
|
||||
self.network_id,
|
||||
peer,
|
||||
outcome.role,
|
||||
conn.clone(),
|
||||
send,
|
||||
recv,
|
||||
Arc::clone(&self.params.limits),
|
||||
self.session_events_tx.clone(),
|
||||
self.shutdown.clone(),
|
||||
);
|
||||
|
||||
let snapshot = snapshot_connection(&conn);
|
||||
self.sessions.insert(peer, session);
|
||||
self.metrics.sessions_established += 1;
|
||||
|
||||
// Announce ourselves straight away so the peer learns our hostname and
|
||||
// capabilities without another round of discovery.
|
||||
let announcement = ControlMessage::Announce(self.local_announcement());
|
||||
if let Err(err) = self.send_to(peer, announcement) {
|
||||
tracing::debug!(%err, "could not queue initial announcement");
|
||||
}
|
||||
|
||||
self.emit(Event::PeerConnected {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
role: outcome.role,
|
||||
transport: snapshot.transport,
|
||||
rtt: snapshot.rtt,
|
||||
});
|
||||
}
|
||||
|
||||
fn local_announcement(&mut self) -> Announcement {
|
||||
let mut capabilities = Vec::new();
|
||||
let mut errors = Vec::new();
|
||||
for plugin in &self.params.plugins {
|
||||
match plugin.local_capability(self.network_id) {
|
||||
Ok(Some(capability)) => capabilities.push(capability),
|
||||
Ok(None) => {}
|
||||
Err(err) => errors.push((plugin.protocol_id().to_string(), err.to_string())),
|
||||
}
|
||||
}
|
||||
for (protocol, reason) in errors {
|
||||
self.metrics.plugin_errors += 1;
|
||||
self.emit(Event::PluginError {
|
||||
network: self.network_id,
|
||||
protocol,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
capabilities.truncate(self.params.limits.max_capabilities);
|
||||
Announcement {
|
||||
hostname: self.params.hostname.clone(),
|
||||
capabilities,
|
||||
}
|
||||
}
|
||||
|
||||
async fn record_hints(&self, conn: &Connection) {
|
||||
let snapshot = snapshot_connection(conn);
|
||||
let peer_bytes = *snapshot.remote_id.as_bytes();
|
||||
for path in &snapshot.paths {
|
||||
if let Some(encoded) = encode_hint(&path.remote) {
|
||||
self.params
|
||||
.storage
|
||||
.record_hint(
|
||||
self.network_id,
|
||||
peer_bytes,
|
||||
encoded,
|
||||
self.params.limits.max_hints_per_peer,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_session_event(&mut self, event: SessionEvent) {
|
||||
match event {
|
||||
SessionEvent::Message {
|
||||
session_id,
|
||||
peer,
|
||||
message,
|
||||
bytes,
|
||||
} => {
|
||||
let current = self.sessions.get(&peer).map(|session| session.id);
|
||||
if current != Some(session_id) {
|
||||
return;
|
||||
}
|
||||
self.metrics.control_messages_received += 1;
|
||||
self.metrics.control_bytes_received += bytes as u64;
|
||||
if let Some(session) = self.sessions.get_mut(&peer) {
|
||||
session.messages_received += 1;
|
||||
session.bytes_received += bytes as u64;
|
||||
}
|
||||
self.dispatch_message(peer, message);
|
||||
}
|
||||
SessionEvent::Violation {
|
||||
session_id,
|
||||
peer,
|
||||
error,
|
||||
} => {
|
||||
let current = self.sessions.get(&peer).map(|session| session.id);
|
||||
if current != Some(session_id) {
|
||||
return;
|
||||
}
|
||||
self.metrics.protocol_violations += 1;
|
||||
self.emit(Event::ProtocolViolation {
|
||||
network: Some(self.network_id),
|
||||
peer: Some(peer),
|
||||
reason: error.to_string(),
|
||||
});
|
||||
}
|
||||
SessionEvent::Closed {
|
||||
session_id,
|
||||
peer,
|
||||
reason,
|
||||
} => {
|
||||
let current = self.sessions.get(&peer).map(|session| session.id);
|
||||
if current != Some(session_id) {
|
||||
return;
|
||||
}
|
||||
if let Some(session) = self.sessions.remove(&peer) {
|
||||
session.abort();
|
||||
session.conn.close(0u32.into(), b"session ended");
|
||||
}
|
||||
self.metrics.disconnects += 1;
|
||||
for plugin in &self.params.plugins {
|
||||
plugin.on_peer_gone(self.network_id, peer);
|
||||
}
|
||||
// Retry promptly, then back off if it keeps failing.
|
||||
let policy = &self.params.reconnect;
|
||||
let state = self
|
||||
.dial_states
|
||||
.entry(peer)
|
||||
.or_insert_with(|| DialState::new(CandidateSource::Discovery));
|
||||
state.in_flight = false;
|
||||
state.next_attempt = Instant::now() + policy.initial_delay;
|
||||
self.emit(Event::PeerDisconnected {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_message(&mut self, peer: EndpointId, message: ControlMessage) {
|
||||
match &message {
|
||||
ControlMessage::Announce(announcement) => {
|
||||
let capabilities = announcement.capabilities.clone();
|
||||
if let Some(session) = self.sessions.get_mut(&peer) {
|
||||
session.hostname = Some(announcement.hostname.clone());
|
||||
session.capabilities = capabilities.clone();
|
||||
}
|
||||
self.dispatch_capabilities(peer, &capabilities);
|
||||
}
|
||||
ControlMessage::Ping { seq, payload } => {
|
||||
let pong = ControlMessage::Pong {
|
||||
seq: *seq,
|
||||
payload: payload.clone(),
|
||||
};
|
||||
if let Err(err) = self.send_to(peer, pong) {
|
||||
tracing::debug!(%err, "could not queue pong");
|
||||
}
|
||||
}
|
||||
ControlMessage::Pong { .. } | ControlMessage::Bye { .. } => {}
|
||||
}
|
||||
|
||||
self.emit(Event::MessageReceived {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
message,
|
||||
});
|
||||
}
|
||||
|
||||
fn dispatch_capabilities(&mut self, peer: EndpointId, capabilities: &[PluginCapability]) {
|
||||
let mut errors = Vec::new();
|
||||
for capability in capabilities {
|
||||
for plugin in &self.params.plugins {
|
||||
if plugin.protocol_id() != capability.protocol {
|
||||
continue;
|
||||
}
|
||||
// The core hands the opaque payload over without interpreting it.
|
||||
if let Err(err) = plugin.on_peer_capability(self.network_id, peer, capability) {
|
||||
errors.push((plugin.protocol_id().to_string(), err.to_string()));
|
||||
}
|
||||
}
|
||||
}
|
||||
for (protocol, reason) in errors {
|
||||
self.metrics.plugin_errors += 1;
|
||||
self.emit(Event::PluginError {
|
||||
network: self.network_id,
|
||||
protocol,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------- status
|
||||
|
||||
fn status(&self) -> NetworkStatus {
|
||||
let mut peers: Vec<PeerStatus> = self
|
||||
.sessions
|
||||
.values()
|
||||
.map(|session| {
|
||||
let snapshot = snapshot_connection(&session.conn);
|
||||
PeerStatus {
|
||||
endpoint_id: session.peer,
|
||||
role: session.role,
|
||||
hostname: session.hostname.clone(),
|
||||
capabilities: session.capabilities.clone(),
|
||||
connected_for: session.established.elapsed(),
|
||||
paths: snapshot.paths,
|
||||
transport: snapshot.transport,
|
||||
rtt: snapshot.rtt,
|
||||
connection: snapshot.counters,
|
||||
control_messages_sent: session.messages_sent,
|
||||
control_messages_received: session.messages_received,
|
||||
control_bytes_sent: session.bytes_sent,
|
||||
control_bytes_received: session.bytes_received,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
peers.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes()));
|
||||
|
||||
let mut candidates: Vec<CandidateStatus> = self
|
||||
.dial_states
|
||||
.iter()
|
||||
.map(|(peer, state)| CandidateStatus {
|
||||
endpoint_id: *peer,
|
||||
source: state.source,
|
||||
consecutive_failures: state.consecutive_failures,
|
||||
})
|
||||
.collect();
|
||||
candidates.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes()));
|
||||
|
||||
NetworkStatus {
|
||||
descriptor: self.params.keys.descriptor(),
|
||||
name: self.params.keys.name().clone(),
|
||||
network_id: self.network_id,
|
||||
state: NetworkState::Active,
|
||||
peers,
|
||||
candidates,
|
||||
metrics: self.metrics.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Which side dialled, given a role and the two identities.
|
||||
fn initiator_of(role: Role, local: EndpointId, peer: EndpointId) -> EndpointId {
|
||||
match role {
|
||||
Role::Initiator => local,
|
||||
Role::Responder => peer,
|
||||
}
|
||||
}
|
||||
|
||||
/// Performs one dial and handshake.
|
||||
async fn dial(
|
||||
adapter: EndpointAdapter,
|
||||
addr: iroh::EndpointAddr,
|
||||
keys: NetworkKeys,
|
||||
limits: Arc<Limits>,
|
||||
local_id: EndpointId,
|
||||
peer: EndpointId,
|
||||
) -> DialOutcome {
|
||||
let connect = tokio::time::timeout(limits.dial_timeout, adapter.connect(addr)).await;
|
||||
let (conn, mut send, mut recv) = match connect {
|
||||
Ok(Ok(parts)) => parts,
|
||||
Ok(Err(err)) => {
|
||||
return DialOutcome::Failed {
|
||||
peer,
|
||||
reason: err.to_string(),
|
||||
during_handshake: false,
|
||||
};
|
||||
}
|
||||
Err(_) => {
|
||||
return DialOutcome::Failed {
|
||||
peer,
|
||||
reason: "dial timed out".into(),
|
||||
during_handshake: false,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
match handshake::initiate(&conn, &mut send, &mut recv, local_id, &keys, &limits).await {
|
||||
Ok(outcome) => DialOutcome::Established(Box::new(InboundSession {
|
||||
conn,
|
||||
send,
|
||||
recv,
|
||||
outcome,
|
||||
})),
|
||||
Err(err) => {
|
||||
conn.close(2u32.into(), b"handshake failed");
|
||||
DialOutcome::Failed {
|
||||
peer,
|
||||
reason: err.to_string(),
|
||||
during_handshake: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Encodes a path address as a cache hint.
|
||||
fn encode_hint(addr: &PathAddr) -> Option<String> {
|
||||
match addr {
|
||||
PathAddr::Ip(socket) => Some(format!("ip:{socket}")),
|
||||
PathAddr::Relay(url) => Some(format!("relay:{url}")),
|
||||
PathAddr::Other(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Decodes a cache hint back into an address. Malformed hints are ignored.
|
||||
fn decode_hint(endpoint_id: EndpointId, hint: &str) -> Option<iroh::EndpointAddr> {
|
||||
if let Some(rest) = hint.strip_prefix("ip:") {
|
||||
let socket: std::net::SocketAddr = rest.parse().ok()?;
|
||||
return Some(iroh::EndpointAddr::new(endpoint_id).with_ip_addr(socket));
|
||||
}
|
||||
if let Some(rest) = hint.strip_prefix("relay:") {
|
||||
let url: iroh::RelayUrl = rest.parse().ok()?;
|
||||
return Some(iroh::EndpointAddr::new(endpoint_id).with_relay_url(url));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// Folds newly learned addresses into a known candidate address.
|
||||
fn merge_addr(existing: &mut iroh::EndpointAddr, incoming: &iroh::EndpointAddr) {
|
||||
if existing.id != incoming.id {
|
||||
*existing = incoming.clone();
|
||||
return;
|
||||
}
|
||||
for addr in &incoming.addrs {
|
||||
existing.addrs.insert(addr.clone());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,337 @@
|
||||
//! One authenticated session with one peer, in one network.
|
||||
//!
|
||||
//! A session owns a reader task and a writer task over a single QUIC
|
||||
//! bidirectional stream. Splitting them keeps both halves simple and avoids
|
||||
//! cancelling a partially consumed frame, which stream reads do not tolerate.
|
||||
//!
|
||||
//! Every inbound message is re-checked against the session's network id, so an
|
||||
//! authenticated session for one network can never deliver into another.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
use iroh::EndpointId;
|
||||
use iroh::endpoint::{Connection, RecvStream, SendStream};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::config::Limits;
|
||||
use crate::dataplane::PluginCapability;
|
||||
use crate::error::ProtocolError;
|
||||
use crate::identity::NetworkId;
|
||||
use crate::proto::handshake::Role;
|
||||
use crate::proto::message::{ControlMessage, Envelope, decode, validate};
|
||||
use crate::proto::{read_frame, write_frame};
|
||||
|
||||
use super::shutdown::Shutdown;
|
||||
|
||||
/// What a session task reports back to its network runtime.
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum SessionEvent {
|
||||
/// A valid control message arrived.
|
||||
Message {
|
||||
/// Which session instance produced this.
|
||||
session_id: u64,
|
||||
/// The peer.
|
||||
peer: EndpointId,
|
||||
/// The decoded message.
|
||||
message: ControlMessage,
|
||||
/// Payload bytes read off the wire.
|
||||
bytes: usize,
|
||||
},
|
||||
/// The peer sent something the protocol does not allow.
|
||||
Violation {
|
||||
/// Which session instance produced this.
|
||||
session_id: u64,
|
||||
/// The peer.
|
||||
peer: EndpointId,
|
||||
/// What was wrong.
|
||||
error: ProtocolError,
|
||||
},
|
||||
/// The session ended.
|
||||
Closed {
|
||||
/// Which session instance ended.
|
||||
session_id: u64,
|
||||
/// The peer.
|
||||
peer: EndpointId,
|
||||
/// Why it ended.
|
||||
reason: String,
|
||||
},
|
||||
}
|
||||
|
||||
/// A live session, as held by the network runtime.
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct Session {
|
||||
pub(crate) id: u64,
|
||||
pub(crate) peer: EndpointId,
|
||||
pub(crate) role: Role,
|
||||
pub(crate) established: Instant,
|
||||
pub(crate) conn: Connection,
|
||||
/// Already encoded frame payloads, so the runtime can account for the exact
|
||||
/// number of control bytes it queues.
|
||||
pub(crate) outbound: mpsc::Sender<Vec<u8>>,
|
||||
pub(crate) hostname: Option<String>,
|
||||
pub(crate) capabilities: Vec<PluginCapability>,
|
||||
pub(crate) messages_sent: u64,
|
||||
pub(crate) messages_received: u64,
|
||||
pub(crate) bytes_sent: u64,
|
||||
pub(crate) bytes_received: u64,
|
||||
reader: JoinHandle<()>,
|
||||
writer: JoinHandle<()>,
|
||||
shutdown: Shutdown,
|
||||
}
|
||||
|
||||
impl Session {
|
||||
/// Signals both tasks to stop and waits for them, with a bounded grace
|
||||
/// period.
|
||||
///
|
||||
/// A peer that stops reading must not be able to hold up shutdown, so the
|
||||
/// tasks are aborted if they do not wind down in time.
|
||||
pub(crate) async fn stop(self) {
|
||||
let Session {
|
||||
conn,
|
||||
reader,
|
||||
writer,
|
||||
shutdown,
|
||||
..
|
||||
} = self;
|
||||
shutdown.trigger();
|
||||
// Closing the connection unblocks a reader parked on the stream.
|
||||
conn.close(0u32.into(), b"session stopped by local agent");
|
||||
|
||||
let reader_abort = reader.abort_handle();
|
||||
let writer_abort = writer.abort_handle();
|
||||
let joined = tokio::time::timeout(STOP_GRACE, async move {
|
||||
let _ = reader.await;
|
||||
let _ = writer.await;
|
||||
})
|
||||
.await;
|
||||
if joined.is_err() {
|
||||
tracing::debug!("session tasks did not wind down in time; aborting them");
|
||||
reader_abort.abort();
|
||||
writer_abort.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// Aborts both tasks without waiting. Used on replacement.
|
||||
pub(crate) fn abort(&self) {
|
||||
self.shutdown.trigger();
|
||||
self.reader.abort();
|
||||
self.writer.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// How long a stopping session may take to wind down before its tasks are
|
||||
/// aborted. Shutdown must be bounded even if a peer stops reading.
|
||||
const STOP_GRACE: std::time::Duration = std::time::Duration::from_millis(500);
|
||||
|
||||
/// Source of monotonically increasing session instance ids.
|
||||
static NEXT_SESSION_ID: AtomicU64 = AtomicU64::new(1);
|
||||
|
||||
/// Starts the reader and writer tasks for an authenticated stream.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) fn spawn(
|
||||
network_id: NetworkId,
|
||||
peer: EndpointId,
|
||||
role: Role,
|
||||
conn: Connection,
|
||||
send: SendStream,
|
||||
recv: RecvStream,
|
||||
limits: Arc<Limits>,
|
||||
events: mpsc::Sender<SessionEvent>,
|
||||
parent_shutdown: Shutdown,
|
||||
) -> Session {
|
||||
let id = NEXT_SESSION_ID.fetch_add(1, Ordering::Relaxed);
|
||||
let shutdown = Shutdown::new();
|
||||
let (outbound_tx, outbound_rx) = mpsc::channel(limits.session_send_queue);
|
||||
|
||||
let writer = tokio::spawn(writer_task(
|
||||
send,
|
||||
outbound_rx,
|
||||
Arc::clone(&limits),
|
||||
shutdown.clone(),
|
||||
parent_shutdown.clone(),
|
||||
));
|
||||
|
||||
let reader = tokio::spawn(reader_task(
|
||||
id,
|
||||
network_id,
|
||||
peer,
|
||||
recv,
|
||||
limits,
|
||||
events,
|
||||
shutdown.clone(),
|
||||
parent_shutdown,
|
||||
));
|
||||
|
||||
Session {
|
||||
id,
|
||||
peer,
|
||||
role,
|
||||
established: Instant::now(),
|
||||
conn,
|
||||
outbound: outbound_tx,
|
||||
hostname: None,
|
||||
capabilities: Vec::new(),
|
||||
messages_sent: 0,
|
||||
messages_received: 0,
|
||||
bytes_sent: 0,
|
||||
bytes_received: 0,
|
||||
reader,
|
||||
writer,
|
||||
shutdown,
|
||||
}
|
||||
}
|
||||
|
||||
async fn writer_task(
|
||||
mut send: SendStream,
|
||||
mut outbound: mpsc::Receiver<Vec<u8>>,
|
||||
limits: Arc<Limits>,
|
||||
shutdown: Shutdown,
|
||||
parent: Shutdown,
|
||||
) {
|
||||
loop {
|
||||
let encoded = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => break,
|
||||
_ = parent.wait() => break,
|
||||
encoded = outbound.recv() => match encoded {
|
||||
Some(encoded) => encoded,
|
||||
None => break,
|
||||
},
|
||||
};
|
||||
|
||||
// The write must stay cancellable: shutting the session down cannot
|
||||
// wait for a peer that has stopped reading. Abandoning a half written
|
||||
// frame is fine, because the session is going away with it.
|
||||
let write = tokio::time::timeout(
|
||||
limits.write_timeout,
|
||||
write_frame(&mut send, &encoded, limits.max_frame_len),
|
||||
);
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => break,
|
||||
_ = parent.wait() => break,
|
||||
result = write => result,
|
||||
};
|
||||
match result {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(err)) => {
|
||||
tracing::debug!(%err, "control stream write failed");
|
||||
break;
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::debug!("control stream write timed out");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = send.finish();
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn reader_task(
|
||||
session_id: u64,
|
||||
network_id: NetworkId,
|
||||
peer: EndpointId,
|
||||
mut recv: RecvStream,
|
||||
limits: Arc<Limits>,
|
||||
events: mpsc::Sender<SessionEvent>,
|
||||
shutdown: Shutdown,
|
||||
parent: Shutdown,
|
||||
) {
|
||||
let reason = loop {
|
||||
let frame = tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.wait() => break "stopped locally".to_string(),
|
||||
_ = parent.wait() => break "network deactivated".to_string(),
|
||||
frame = read_frame(&mut recv, limits.max_frame_len) => frame,
|
||||
};
|
||||
|
||||
let payload = match frame {
|
||||
Ok(payload) => payload,
|
||||
Err(ProtocolError::StreamClosed) => break "peer closed the control stream".to_string(),
|
||||
Err(ProtocolError::Stream(err)) => {
|
||||
// The transport went away. That is a disconnect, not a peer
|
||||
// misbehaving, so it ends the session without being counted as
|
||||
// a protocol violation.
|
||||
break format!("control stream error: {err}");
|
||||
}
|
||||
Err(err) => {
|
||||
// A framing violation ends this session. Framing errors are not
|
||||
// recoverable mid-stream: the next bytes have no known meaning.
|
||||
let text = err.to_string();
|
||||
let _ = events
|
||||
.send(SessionEvent::Violation {
|
||||
session_id,
|
||||
peer,
|
||||
error: err,
|
||||
})
|
||||
.await;
|
||||
break text;
|
||||
}
|
||||
};
|
||||
|
||||
let bytes = payload.len();
|
||||
let envelope: Envelope = match decode(&payload) {
|
||||
Ok(envelope) => envelope,
|
||||
Err(err) => {
|
||||
let _ = events
|
||||
.send(SessionEvent::Violation {
|
||||
session_id,
|
||||
peer,
|
||||
error: err,
|
||||
})
|
||||
.await;
|
||||
break "malformed control frame".to_string();
|
||||
}
|
||||
};
|
||||
|
||||
// Network isolation: a session authenticated for one network must never
|
||||
// deliver a message belonging to another.
|
||||
if envelope.network_id != *network_id.as_bytes() {
|
||||
let _ = events
|
||||
.send(SessionEvent::Violation {
|
||||
session_id,
|
||||
peer,
|
||||
error: ProtocolError::NetworkMismatch,
|
||||
})
|
||||
.await;
|
||||
break "network id mismatch on an authenticated session".to_string();
|
||||
}
|
||||
|
||||
if let Err(err) = validate(&envelope.message, &limits) {
|
||||
let _ = events
|
||||
.send(SessionEvent::Violation {
|
||||
session_id,
|
||||
peer,
|
||||
error: err,
|
||||
})
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
|
||||
if events
|
||||
.send(SessionEvent::Message {
|
||||
session_id,
|
||||
peer,
|
||||
message: envelope.message,
|
||||
bytes,
|
||||
})
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
break "network runtime stopped".to_string();
|
||||
}
|
||||
};
|
||||
|
||||
// Best effort: the runtime may already have stopped draining this channel
|
||||
// while it tears the network down, and a closing session must not block on
|
||||
// that.
|
||||
let _ = events.try_send(SessionEvent::Closed {
|
||||
session_id,
|
||||
peer,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
//! A minimal cancellation primitive.
|
||||
//!
|
||||
//! Kept local so the crate does not pull in a utility dependency for one type,
|
||||
//! and so that no global state is involved: every agent and every network
|
||||
//! runtime owns its own token.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use tokio::sync::watch;
|
||||
|
||||
/// A clonable cancellation token.
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct Shutdown {
|
||||
tx: Arc<watch::Sender<bool>>,
|
||||
rx: watch::Receiver<bool>,
|
||||
}
|
||||
|
||||
impl Shutdown {
|
||||
/// Creates an untriggered token.
|
||||
pub(crate) fn new() -> Self {
|
||||
let (tx, rx) = watch::channel(false);
|
||||
Self {
|
||||
tx: Arc::new(tx),
|
||||
rx,
|
||||
}
|
||||
}
|
||||
|
||||
/// Triggers cancellation. Idempotent.
|
||||
pub(crate) fn trigger(&self) {
|
||||
let _ = self.tx.send(true);
|
||||
}
|
||||
|
||||
/// Whether cancellation has been triggered.
|
||||
pub(crate) fn is_triggered(&self) -> bool {
|
||||
*self.rx.borrow()
|
||||
}
|
||||
|
||||
/// Resolves once cancellation has been triggered.
|
||||
pub(crate) async fn wait(&self) {
|
||||
let mut rx = self.rx.clone();
|
||||
{
|
||||
if *rx.borrow() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
while rx.changed().await.is_ok() {
|
||||
if *rx.borrow() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
// The sender is gone, which for our purposes means "stop".
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
//! Status snapshots.
|
||||
//!
|
||||
//! Metrics are reported at the level they actually belong to. Values that
|
||||
//! genuinely cannot be attributed to a single network — everything the iroh
|
||||
//! endpoint aggregates, for instance — stay at the endpoint level rather than
|
||||
//! being split between networks with invented precision.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use iroh::{EndpointAddr, EndpointId};
|
||||
|
||||
use crate::dataplane::PluginCapability;
|
||||
use crate::discovery::CandidateSource;
|
||||
use crate::identity::{NetworkDescriptor, NetworkId, NetworkName};
|
||||
use crate::net::{ConnectionCounters, PathAddr, PathInfo, TransportKind};
|
||||
use crate::proto::handshake::Role;
|
||||
use crate::storage::CacheOutcome;
|
||||
|
||||
/// Whether a configured network is running locally.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum NetworkState {
|
||||
/// Configured and running.
|
||||
Active,
|
||||
/// Configured but not running.
|
||||
Inactive,
|
||||
}
|
||||
|
||||
/// An unverified candidate as seen by a network runtime.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CandidateStatus {
|
||||
/// Candidate endpoint id.
|
||||
pub endpoint_id: EndpointId,
|
||||
/// Where it came from.
|
||||
pub source: CandidateSource,
|
||||
/// Consecutive failed dial attempts since the last success.
|
||||
pub consecutive_failures: u32,
|
||||
}
|
||||
|
||||
/// Status of one authenticated session.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PeerStatus {
|
||||
/// Authenticated endpoint id.
|
||||
pub endpoint_id: EndpointId,
|
||||
/// Which side this agent played in the handshake.
|
||||
pub role: Role,
|
||||
/// Hostname the peer announced, if it has announced one yet.
|
||||
///
|
||||
/// A mutable binding, not an identity.
|
||||
pub hostname: Option<String>,
|
||||
/// Capabilities the peer announced. Payloads stay opaque.
|
||||
pub capabilities: Vec<PluginCapability>,
|
||||
/// How long the session has been up.
|
||||
pub connected_for: Duration,
|
||||
/// Verified paths of the underlying connection.
|
||||
pub paths: Vec<PathInfo>,
|
||||
/// How the connection currently reaches the peer.
|
||||
pub transport: TransportKind,
|
||||
/// RTT of the selected path, when iroh reported one.
|
||||
pub rtt: Option<Duration>,
|
||||
/// Per-connection counters.
|
||||
pub connection: ConnectionCounters,
|
||||
/// Control messages sent on this session.
|
||||
pub control_messages_sent: u64,
|
||||
/// Control messages received on this session.
|
||||
pub control_messages_received: u64,
|
||||
/// Control payload bytes queued for this session.
|
||||
pub control_bytes_sent: u64,
|
||||
/// Control payload bytes read from this session.
|
||||
pub control_bytes_received: u64,
|
||||
}
|
||||
|
||||
/// Counters scoped to one logical network.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct NetworkMetrics {
|
||||
/// Outbound dial attempts started.
|
||||
pub dial_attempts: u64,
|
||||
/// Outbound dials that failed before or during the handshake.
|
||||
pub dial_failures: u64,
|
||||
/// Handshakes rejected, in either direction.
|
||||
pub handshake_failures: u64,
|
||||
/// Sessions that reached the authenticated state.
|
||||
pub sessions_established: u64,
|
||||
/// Sessions that ended.
|
||||
pub disconnects: u64,
|
||||
/// Control messages sent in this network.
|
||||
pub control_messages_sent: u64,
|
||||
/// Control messages received in this network.
|
||||
pub control_messages_received: u64,
|
||||
/// Control bytes sent in this network, payload only.
|
||||
pub control_bytes_sent: u64,
|
||||
/// Control bytes received in this network, payload only.
|
||||
pub control_bytes_received: u64,
|
||||
/// Messages or sessions rejected for protocol violations.
|
||||
pub protocol_violations: u64,
|
||||
/// Errors reported by IP plugins. Never fatal.
|
||||
pub plugin_errors: u64,
|
||||
}
|
||||
|
||||
/// Status of one network.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct NetworkStatus {
|
||||
/// Immutable deterministic description of the network space.
|
||||
pub descriptor: NetworkDescriptor,
|
||||
/// Network name, for convenience.
|
||||
pub name: NetworkName,
|
||||
/// Public network identifier.
|
||||
pub network_id: NetworkId,
|
||||
/// Whether the network is running locally.
|
||||
pub state: NetworkState,
|
||||
/// Authenticated sessions.
|
||||
pub peers: Vec<PeerStatus>,
|
||||
/// Unverified candidates currently known. Not peers.
|
||||
pub candidates: Vec<CandidateStatus>,
|
||||
/// Per-network counters.
|
||||
pub metrics: NetworkMetrics,
|
||||
}
|
||||
|
||||
impl NetworkStatus {
|
||||
/// Endpoint ids of peers with an authenticated session.
|
||||
pub fn connected_peers(&self) -> Vec<EndpointId> {
|
||||
self.peers.iter().map(|peer| peer.endpoint_id).collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Status of the whole agent.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AgentStatus {
|
||||
/// This device's persistent endpoint id.
|
||||
pub endpoint_id: EndpointId,
|
||||
/// Hostname announced to peers.
|
||||
pub hostname: String,
|
||||
/// Sockets actually bound.
|
||||
pub bound_sockets: Vec<std::net::SocketAddr>,
|
||||
/// Addresses iroh believes this endpoint has. Observed, not verified.
|
||||
pub observed_addrs: Vec<PathAddr>,
|
||||
/// The dialable address of this endpoint, as iroh currently reports it.
|
||||
pub endpoint_addr: EndpointAddr,
|
||||
/// What happened to the disposable cache at startup.
|
||||
pub cache_outcome: CacheOutcome,
|
||||
/// Whether the cache is currently usable.
|
||||
pub cache_healthy: bool,
|
||||
/// Per-network status, including configured but inactive networks.
|
||||
pub networks: Vec<NetworkStatus>,
|
||||
}
|
||||
|
||||
impl AgentStatus {
|
||||
/// Looks up one network's status.
|
||||
pub fn network(&self, network_id: &NetworkId) -> Option<&NetworkStatus> {
|
||||
self.networks
|
||||
.iter()
|
||||
.find(|status| &status.network_id == network_id)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user