ab/tsunagi
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Change network protocol selections at runtime

Browse Source 
Protocol selection previously belonged to tsng up and therefore to the
service command line. A blocked or degraded carrier could only be removed
by editing the service and restarting the whole agent. Make the ordered
selection a local per-network setting instead.

Add Agent::set_protocols and a local control request shared by CLI and tray.
Keep tsng protocols without arguments as the protocol catalog; naming a
network shows its selection and supplying a comma-separated list replaces
it. Accept exact names, full ids and unique prefixes using shared selector
rules; ambiguous selectors and invalid lists change nothing. An explicit
none selects control-only operation. Offline edits take the directory lock
and apply on the next start; live edits never write behind the daemon.

Register every built-in plugin and transport in the CLI. Treat up --protocol
as ordered initial defaults for networks without an override, allowing a
protocol absent from the unit to be enabled later. Retain the targeting of
unqualified startup options and allow explicitly scoped options to configure
initially disabled protocols. Persist NULL for inherited preferences and an
empty selection distinctly in SQLite schema 8; migration preserves existing
identities, secrets, signed records and counters.

Serialize changes with network lifecycle operations and commit preferences
before applying or announcing them. Retire only the affected network's data
links and E2E channels, cancel pending opens, and reject obsolete completion
results by attempt id. Clear peer data channels when authenticated protocol
capabilities change while retaining control sessions and topology. Replay
allocations and peer identities after asynchronously prepared plugins become
ready. A control-only start defers creation of the agent's single interface
until data is enabled; plugins receive a sink that can use it later. Automatic
carrier fallback continues to preserve E2E state; explicit policy changes
renegotiate it. Identity derivation and peer wire formats are unchanged.

Add a collapsible protocols row to each tray network tile, with a catalog
provided by the daemon, descriptions, immediately applied checkboxes and
preference-order buttons. Keep editing disabled until the worker refreshes
status so a second click cannot overwrite an edit using a stale snapshot.
Local IPC wire version is 20 for the new request and report fields; upgrade
and restart agent, CLI and tray together. Crate release versions remain unchanged; the requested RC tag is v0.1.0-rc.30.

Validation on Linux through nix develop:
- cargo fmt --all -- --check
- cargo clippy --locked --workspace --all-targets -- -D warnings
- cargo test --locked --workspace --all-targets: 509 passed, 2 existing ignored
  tests, with inherited TSUNAGI_CONTROL_SOCKET and TSUNAGI_CONTROL_GROUP removed

Tests use real iroh membership, SQLite, UDP, AWG and WireGuard encryption,
QUIC carriers and memory TUNs. Cover either endpoint-id role, full-size packet
delivery after switching, an unaffected second active network, saved selection
after restart, empty/invalid selections, first enablement without restart,
rapid changes racing old opens, live and offline CLI changes, startup-absent
protocols and preservation of signed state through migration. Earlier compile
and focused-test iterations caught initializer mistakes, a short test secret,
asynchronous plugin allocation readiness, and an inherited system socket
path; corrected before the successful final checks. Plain cargo was unavailable
in the host shell, so validation used the Nix toolchain. Manual tray interaction,
macOS/Windows, privileged host TUNs and public-network traversal were not run.
This commit is contained in:
Aleksandr Bogomiakov committed 2026-10-12 16:19:18 +01:00
1 parent e701986a39
commit d941cdac7a
18 files changed
+1263 -124
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/tsunagi-core/src/agent/mod.rs
+132 -29
View File
@@ -58,6 +58,8 @@ use shutdown::Shutdown;
/// Summary of a configured network, whether or not it is running.
#[derive(Debug, Clone)]
pub struct ConfiguredNetwork {
/// Enabled local data-plane protocols, in preference order.
pub protocols: Vec<String>,
/// Saved local broadcast participation.
pub broadcast: bool,
/// Public network identifier.
@@ -104,6 +106,37 @@ struct Inner {
}
impl Inner {
/// Called under the lifecycle lock after startup: an initially
/// control-only agent creates its one interface on first enablement.
async fn ensure_interface(self: &Arc<Self>) {
if self.interface.get().is_some() {
return;
}
if let Some(factory) = self.config.tun_factory.clone() {
let carrier = Arc::new(Carrier(Arc::downgrade(self))) as Arc<dyn PacketCarrier>;
match crate::overlay::Interface::start(
factory,
self.config.interface_name.clone(),
self.config.interface_mtu,
Arc::clone(&self.routes),
carrier,
)
.await
{
Ok(interface) => {
let _ = self.interface.set(Arc::new(interface));
}
Err(err) => {
let _ = self.events.send(Event::PluginError {
network: NetworkId::from_bytes([0u8; 32]),
protocol: "overlay".into(),
reason: err.to_string(),
});
}
}
}
}
/// The name this agent currently answers to.
fn read_hostname(&self) -> String {
match self.hostname.read() {
@@ -265,12 +298,90 @@ impl TransportContext for TransportCtx {
}
impl Agent {
fn effective_protocols(&self, saved: Option<&Vec<String>>) -> Vec<String> {
saved
.or(self.inner.config.default_protocols.as_ref())
.cloned()
.unwrap_or_else(|| {
self.inner
.config
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect()
})
}
fn selected_plugins(&self, protocols: &[String]) -> Vec<crate::dataplane::SharedPlugin> {
protocols
.iter()
.filter_map(|name| {
self.inner
.config
.plugins
.iter()
.find(|p| p.protocol_id() == name)
.cloned()
})
.collect()
}
/// Saves and immediately applies a network's ordered local protocol set.
/// An empty set disables data traffic while retaining control sessions.
pub async fn set_protocols(&self, network_id: NetworkId, protocols: Vec<String>) -> Result<()> {
crate::dataplane::validate_protocols(&protocols, self.inner.limits.max_capabilities)?;
for name in &protocols {
if !self
.inner
.config
.plugins
.iter()
.any(|p| p.protocol_id() == name)
{
return Err(Error::InvalidProtocols(format!(
"this agent has no `{name}` plugin"
)));
}
}
let _lifecycle = self.inner.lifecycle.lock().await;
self.inner
.storage
.set_protocols(network_id, protocols.clone())
.await?;
if self.is_active(network_id).await {
let (reply, receive) = oneshot::channel();
if !protocols.is_empty() {
self.inner.ensure_interface().await;
}
self.command(
network_id,
NetCommand::SetProtocols {
protocols,
interface: self.inner.interface.get().cloned(),
reply,
},
)
.await?;
receive.await.map_err(|_| Error::Stopped)?;
}
Ok(())
}
/// 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> {
if let Some(protocols) = &config.default_protocols {
crate::dataplane::validate_protocols(protocols, config.limits.max_capabilities)?;
for name in protocols {
if !config.plugins.iter().any(|p| p.protocol_id() == name) {
return Err(Error::InvalidProtocols(format!(
"this agent has no `{name}` plugin"
)));
}
}
}
let storage = Storage::open(&config.paths)?;
let identity = storage.device_identity().await?;
let adapter = EndpointAdapter::bind(&config, &identity).await?;
@@ -303,31 +414,13 @@ impl Agent {
config,
});
// One interface for the agent, if it was given a way to make one.
// Failing to is reported and not fatal: the control plane works, and
// so do the protocols, they just have nowhere to put packets.
if let Some(factory) = inner.config.tun_factory.clone() {
let carrier = Arc::new(Carrier(Arc::downgrade(&inner))) as Arc<dyn PacketCarrier>;
match crate::overlay::Interface::start(
factory,
inner.config.interface_name.clone(),
inner.config.interface_mtu,
Arc::clone(&inner.routes),
carrier,
)
.await
{
Ok(interface) => {
let _ = inner.interface.set(Arc::new(interface));
}
Err(err) => {
let _ = inner.events.send(Event::PluginError {
network: NetworkId::from_bytes([0u8; 32]),
protocol: "overlay".into(),
reason: err.to_string(),
});
}
}
if inner
.config
.default_protocols
.as_ref()
.is_none_or(|names| !names.is_empty())
{
inner.ensure_interface().await;
}
// The data plane rides on iroh too, which is where it gets hole
@@ -364,8 +457,9 @@ impl Agent {
if !inner.config.plugins.is_empty() {
let (plugin_tx, plugin_rx) = mpsc::channel(64);
let sink = inner
.interface
.get()
.config
.tun_factory
.as_ref()
.map(|_| Arc::new(Sink(Arc::downgrade(&inner))) as Arc<dyn PacketSink>);
let context = PluginContext::new(plugin_tx, inner.identity.endpoint_id(), sink);
for plugin in &inner.config.plugins {
@@ -824,6 +918,12 @@ impl Agent {
};
let broadcast = stored.as_ref().is_none_or(|stored| stored.broadcast);
let protocols =
self.effective_protocols(stored.as_ref().and_then(|s| s.protocols.as_ref()));
let plugins = self.selected_plugins(&protocols);
if !plugins.is_empty() {
self.inner.ensure_interface().await;
}
let exit_offer = stored.as_ref().is_some_and(|stored| stored.exit_node);
let exit_via = stored
.as_ref()
@@ -842,7 +942,8 @@ impl Agent {
discovery,
discovery_interval: self.inner.config.discovery_interval,
discovery_policy: self.inner.config.discovery_policy.clone(),
plugins: self.inner.config.plugins.clone(),
available_plugins: self.inner.config.plugins.clone(),
plugins: plugins.clone(),
routes: Arc::clone(&self.inner.routes),
interface: self.inner.interface.get().cloned(),
hostname: self.inner.read_hostname(),
@@ -856,7 +957,7 @@ impl Agent {
networks.insert(network_id, handle);
drop(networks);
for plugin in &self.inner.config.plugins {
for plugin in &plugins {
plugin.on_network_activated(network_id);
}
@@ -997,6 +1098,7 @@ impl Agent {
.await?
.into_iter()
.map(|stored| ConfiguredNetwork {
protocols: self.effective_protocols(stored.protocols.as_ref()),
broadcast: stored.broadcast,
network_id: stored.network_id,
name: stored.name,
@@ -1075,6 +1177,7 @@ impl Agent {
let range = self.desired_range(stored.network_id).await?;
let range_conflicts = self.range_conflicts(stored.network_id, range).await?;
networks.push(NetworkStatus {
protocols: self.effective_protocols(stored.protocols.as_ref()),
auto_start: stored.auto_start,
broadcast: stored.broadcast,
exit_offer: stored.exit_node,
crates/tsunagi-core/src/agent/network.rs
+157 -19
View File
@@ -72,6 +72,11 @@ pub(crate) enum NetCommand {
enabled: bool,
reply: oneshot::Sender<()>,
},
SetProtocols {
interface: Option<Arc<crate::overlay::Interface>>,
protocols: Vec<String>,
reply: oneshot::Sender<()>,
},
/// Offer this agent as an exit node in this network, or stop.
SetExitOffer {
enabled: bool,
@@ -122,6 +127,7 @@ impl std::fmt::Debug for NetCommand {
NetCommand::Status { .. } => f.write_str("Status"),
NetCommand::Recheck => f.write_str("Recheck"),
NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"),
NetCommand::SetProtocols { protocols, .. } => write!(f, "SetProtocols({protocols:?})"),
NetCommand::SetExitOffer { enabled, .. } => write!(f, "SetExitOffer({enabled})"),
NetCommand::SetExitVia { via, .. } => write!(f, "SetExitVia({})", via.is_some()),
NetCommand::Reannounce => f.write_str("Reannounce"),
@@ -171,6 +177,7 @@ pub(crate) struct RuntimeParams {
pub(crate) discovery_interval: Duration,
pub(crate) discovery_policy: DiscoveryPolicy,
pub(crate) plugins: Vec<SharedPlugin>,
pub(crate) available_plugins: Vec<SharedPlugin>,
/// Who holds which overlay address, shared with every other network.
pub(crate) routes: Arc<crate::overlay::RoutingTable>,
/// The one interface, when the agent has one.
@@ -192,6 +199,7 @@ pub(crate) struct RuntimeParams {
/// Outcome of one attempt to open a data plane link.
struct LinkOutcome {
generation: u64,
peer: EndpointId,
protocol: String,
result: Result<SharedLink, String>,
@@ -332,6 +340,9 @@ struct Runtime {
announced_reach: HashSet<crate::proto::message::ReachableLink>,
/// Links currently being opened, so we do not start two.
opening: HashSet<(EndpointId, String)>,
link_tasks: HashMap<(EndpointId, String), tokio::task::AbortHandle>,
link_attempts: HashMap<(EndpointId, String), u64>,
next_link_attempt: u64,
link_results_tx: mpsc::Sender<LinkOutcome>,
link_results_rx: mpsc::Receiver<LinkOutcome>,
/// Protocol dial failure count and backoff cooldown per peer and protocol.
@@ -425,6 +436,9 @@ impl Runtime {
local_capabilities: Vec::new(),
announced_reach: HashSet::new(),
opening: HashSet::new(),
link_tasks: HashMap::new(),
link_attempts: HashMap::new(),
next_link_attempt: 0,
link_results_tx,
link_results_rx,
protocol_dial_states: HashMap::new(),
@@ -511,6 +525,10 @@ impl Runtime {
link.close();
}
self.links.clear();
for (_, task) in self.link_tasks.drain() {
task.abort();
}
self.opening.clear();
self.pending_health.clear();
self.last_failed_protocol.clear();
self.protocol_dial_states.clear();
@@ -553,6 +571,60 @@ impl Runtime {
NetCommand::Status { reply } => {
let _ = reply.send(Box::new(self.status()));
}
NetCommand::SetProtocols {
protocols,
interface,
reply,
} => {
self.params.interface = interface;
let current: Vec<_> = self
.params
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect();
if current != protocols {
let peers: Vec<_> = self.sessions.keys().copied().collect();
for peer in &peers {
self.reset_peer_data(*peer, "local protocol selection changed");
}
for plugin in &self.params.plugins {
if !protocols.iter().any(|name| name == plugin.protocol_id()) {
plugin.on_network_deactivated(self.network_id);
}
}
self.params.plugins = protocols
.iter()
.filter_map(|name| {
self.params
.available_plugins
.iter()
.find(|p| p.protocol_id() == name)
.cloned()
})
.collect();
for plugin in &self.params.plugins {
if !current.iter().any(|name| name == plugin.protocol_id()) {
plugin.on_network_activated(self.network_id);
}
}
self.hub.set_priority(protocols);
self.publish_allocations().await;
for peer in peers {
let caps = self
.sessions
.get(&peer)
.map(|s| s.capabilities.clone())
.unwrap_or_default();
self.dispatch_capabilities(peer, &caps);
}
self.reannounce();
self.ensure_links();
self.update_paths();
self.announce_reach(true);
}
let _ = reply.send(());
}
NetCommand::SetBroadcast { enabled, reply } => {
self.params.broadcast = enabled;
self.update_broadcast();
@@ -584,7 +656,22 @@ impl Runtime {
let _ = reply.send(outcome);
}
NetCommand::Recheck => self.discovery_round().await,
NetCommand::Reannounce => self.reannounce(),
NetCommand::Reannounce => {
// A newly enabled plugin may prepare its key asynchronously.
// Replay allocations and peer identities once it is ready.
self.publish_allocations().await;
let peers: Vec<_> = self
.sessions
.iter()
.map(|(peer, s)| (*peer, s.capabilities.clone()))
.collect();
for (peer, caps) in peers {
self.dispatch_capabilities(peer, &caps);
}
self.reannounce();
self.ensure_links();
self.update_paths();
}
NetCommand::Release { reply } => {
// A tombstone, so the address and the name this agent held
// are freed for somebody else instead of staying reserved
@@ -1578,13 +1665,16 @@ impl Runtime {
if self.links.contains_key(&key) || self.opening.contains(&key) {
continue;
}
self.opening.insert(key);
self.opening.insert(key.clone());
let results = self.link_results_tx.clone();
let transport = Arc::clone(&transport);
let network = self.network_id;
let shutdown = self.shutdown.clone();
tokio::spawn(async move {
self.next_link_attempt = self.next_link_attempt.wrapping_add(1);
let generation = self.next_link_attempt;
self.link_attempts.insert(key.clone(), generation);
let task = tokio::spawn(async move {
let result = tokio::select! {
biased;
_ = shutdown.wait() => Err("network deactivated".to_string()),
@@ -1594,12 +1684,14 @@ impl Runtime {
};
let _ = results
.send(LinkOutcome {
generation,
peer,
protocol,
result,
})
.await;
});
self.link_tasks.insert(key, task.abort_handle());
}
}
@@ -1834,7 +1926,15 @@ impl Runtime {
fn handle_link_result(&mut self, outcome: LinkOutcome) {
let key = (outcome.peer, outcome.protocol.clone());
if self.link_attempts.get(&key) != Some(&outcome.generation) {
if let Ok(link) = outcome.result {
link.close();
}
return;
}
self.link_attempts.remove(&key);
self.opening.remove(&key);
self.link_tasks.remove(&key);
match outcome.result {
Ok(link) => {
self.record_protocol_success(outcome.peer, &outcome.protocol);
@@ -1903,12 +2003,17 @@ impl Runtime {
link.close();
return;
}
if !self
.params
.plugins
.iter()
.any(|plugin| plugin.protocol_id() == protocol)
{
if !self.params.plugins.iter().any(|plugin| {
plugin.protocol_id() == protocol
&& self.sessions.get(&peer).is_some_and(|s| {
s.capabilities.iter().any(|c| {
c.enabled
&& c.protocol == protocol
&& c.version == plugin.protocol_version()
})
})
}) {
link.close();
return;
}
@@ -1966,14 +2071,25 @@ impl Runtime {
/// Drops every link to a peer.
fn drop_links_for(&mut self, peer: EndpointId) {
// The peer itself is going, not just a path to it, so the
// protocol's link goes with it.
self.reset_peer_data(peer, "peer session ended");
self.hub.remove_peer(peer);
self.reachable.remove(&peer);
self.protocol_dial_states.retain(|(p, _), _| *p != peer);
self.last_failed_protocol.remove(&peer);
self.pending_health.retain(|(p, _), _| *p != peer);
let keys: Vec<(EndpointId, String)> = self
}
/// Retire only the data plane; authenticated membership and topology remain.
fn reset_peer_data(&mut self, peer: EndpointId, reason: &str) {
self.link_attempts.retain(|(id, _), _| *id != peer);
self.opening.retain(|(id, _)| *id != peer);
self.link_tasks.retain(|(id, _), task| {
if *id == peer {
task.abort();
false
} else {
true
}
});
self.hub.clear_peer_links(peer);
let keys: Vec<_> = self
.links
.keys()
.filter(|(id, _)| *id == peer)
@@ -1987,9 +2103,15 @@ impl Runtime {
network: self.network_id,
peer,
protocol: key.1,
reason: "peer session ended".into(),
reason: reason.into(),
});
}
self.protocol_dial_states.retain(|(id, _), _| *id != peer);
self.last_failed_protocol.remove(&peer);
self.pending_health.retain(|(id, _), _| *id != peer);
for plugin in &self.params.plugins {
plugin.on_peer_gone(self.network_id, peer);
}
}
// --------------------------------------------------------------- sessions
@@ -2182,9 +2304,6 @@ impl Runtime {
self.watch_exit_via();
self.update_paths();
self.announce_reach(false);
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
@@ -2277,6 +2396,18 @@ impl Runtime {
}
ControlMessage::Announce(announcement) => {
let capabilities = announcement.capabilities.clone();
let changed = self.sessions.get(&peer).is_some_and(|session| {
let selection = |caps: &[PluginCapability]| {
caps.iter()
.filter(|c| c.enabled)
.map(|c| (c.protocol.clone(), c.version))
.collect::<Vec<_>>()
};
selection(&session.capabilities) != selection(&capabilities)
});
if changed {
self.reset_peer_data(peer, "peer protocol selection changed");
}
if let Some(session) = self.sessions.get_mut(&peer) {
session.hostname = Some(announcement.hostname.clone());
session.broadcast = announcement.broadcast;
@@ -2289,6 +2420,7 @@ impl Runtime {
self.dispatch_capabilities(peer, &capabilities);
self.ensure_links();
self.update_paths();
self.announce_reach(false);
}
ControlMessage::Ping { seq, payload } => {
let pong = ControlMessage::Pong {
@@ -2522,6 +2654,12 @@ impl Runtime {
members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id);
NetworkStatus {
protocols: self
.params
.plugins
.iter()
.map(|p| p.protocol_id().to_owned())
.collect(),
auto_start: true,
broadcast: self.params.broadcast,
exit_offer: self.params.exit_offer,
crates/tsunagi-core/src/agent/status.rs
+2
View File
@@ -223,6 +223,8 @@ pub struct RangeConflict {
/// Status of one network.
#[derive(Debug, Clone)]
pub struct NetworkStatus {
/// Enabled local data-plane protocols, in preference order.
pub protocols: Vec<String>,
/// Whether this network is requested to start or resume automatically.
pub auto_start: bool,
/// Local broadcast participation in this network.