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

Relay across any mix of links, one online answer for every client, first claim owns a hostname

Browse Source 
A tunnel no longer depends on the kind of link under it. The relay hub keeps
one forwarding table per network instead of one per protocol, the envelope
names the end-to-end protocol, and a neighbour reached over several kinds of
transport is sent to over the best open one. A member in the middle needs no
protocol in common with either end. Which protocol a pair runs is settled by
the lower endpoint id's announced order, so both ends agree without asking.
A direct link that comes up later is shorter than the path through others, so
traffic moves to it with the same logical link.

The agent now answers "is it online" once: a live control session or any data
path. Members and peers report how they are reached (direct over a named
transport, relayed through a named member, or no data path yet), and `tsng
status` and the tray both read that instead of a session list and a WireGuard
handshake respectively.

A hostname belongs to whoever claimed it first. A claim carries the time its
author first claimed the name, signed with the record, and replicas order by
(time, id) so every member names the same holder. A later claimant keeps
working by address, is not found by the name, and is told so in its status,
`tsng status`, the tray and a HostnameTaken event.

Wire change: control and data ALPN 5, control PROTOCOL_VERSION 5, envelope
version 2, signing domain v3, state schema 7, local control protocol 18.
Upgrade every member together.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Aleksandr BogomiakovandClaude committed 2026-10-05 11:39:33 +01:00
1 parent c184ef0914
commit 97377e3703
22 files changed
+1699 -292
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/tsunagi-core/src/agent/events.rs
+13
View File
@@ -129,6 +129,19 @@ pub enum Event {
/// Why it is not up.
reason: String,
},
/// The name this agent claimed was claimed first by another member.
///
/// Nothing stops working: this agent keeps its address and every
/// connection. It just is not found by that name until it is renamed or
/// the member that holds the name lets it go.
HostnameTaken {
/// The network.
network: NetworkId,
/// The name that is taken.
name: String,
/// The member that holds it.
holder: EndpointId,
},
/// An IP plugin reported an error. Never fatal.
PluginError {
/// The network the call was scoped to.
crates/tsunagi-core/src/agent/mod.rs
+3 -2
View File
@@ -26,8 +26,8 @@ mod status;
pub use events::Event;
pub use status::{
AgentStatus, CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus,
OverlayStatus, PeerStatus,
AgentStatus, CandidateStatus, DataPath, HostnameConflict, MemberStatus, NetworkMetrics,
NetworkState, NetworkStatus, OverlayStatus, PeerStatus,
};
use std::collections::HashMap;
@@ -981,6 +981,7 @@ impl Agent {
// An inactive network has no runtime to ask; the roster and
// the range come from one. Empty, not invented.
members: Vec::new(),
hostname_conflict: None,
range: None,
range_conflict: None,
metrics: NetworkMetrics::default(),
crates/tsunagi-core/src/agent/network.rs
+211 -70
View File
@@ -32,7 +32,8 @@ use super::events::Event;
use super::session::{self, Session, SessionEvent};
use super::shutdown::Shutdown;
use super::status::{
CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus, PeerStatus,
CandidateStatus, DataPath, HostnameConflict, MemberStatus, NetworkMetrics, NetworkState,
NetworkStatus, PeerStatus,
};
/// An inbound connection that already passed the handshake.
@@ -320,6 +321,10 @@ struct Runtime {
std::time::Instant,
),
>,
/// The protocols this agent last announced, in the order it announced
/// them. The order is its preference, and the lower of two ids settles
/// which protocol a pair runs.
local_capabilities: Vec<PluginCapability>,
/// What this agent last told its peers it could reach, so it is only
/// said again when it changed.
announced_reach: HashSet<crate::proto::message::ReachableLink>,
@@ -338,6 +343,9 @@ struct Runtime {
/// publish a claim afterwards — the periodic check, a peer's records
/// arriving — would silently take it back.
released: bool,
/// The name this agent last said was taken, so it is said once per
/// change rather than on every round.
reported_name_taken: Option<HostnameConflict>,
/// The address last reported as absent from the interface, so it is said
/// once rather than for ever.
reported_missing: Option<std::net::Ipv4Addr>,
@@ -366,6 +374,16 @@ impl Runtime {
let (link_results_tx, link_results_rx) = mpsc::channel(64);
let (candidate_tx, discovery_candidates) =
mpsc::channel(params.limits.max_discovery_candidates.max(1));
let hub = crate::dataplane::relay::RelayHub::new(network_id, local_id);
// The order the protocols were configured in is also how links are
// preferred when a neighbour can be reached over more than one.
hub.set_priority(
params
.plugins
.iter()
.map(|plugin| plugin.protocol_id().to_string())
.collect(),
);
let discovery_worker = params.discovery.as_ref().map(|backend| {
DiscoveryWorker::spawn(
backend.clone(),
@@ -393,8 +411,9 @@ impl Runtime {
dial_results_tx,
dial_results_rx,
links: HashMap::new(),
hub: crate::dataplane::relay::RelayHub::new(network_id, local_id),
hub,
reachable: HashMap::new(),
local_capabilities: Vec::new(),
announced_reach: HashSet::new(),
opening: HashSet::new(),
link_results_tx,
@@ -403,6 +422,7 @@ impl Runtime {
reported_mismatch: HashSet::new(),
released: false,
reported_missing: None,
reported_name_taken: None,
state: StateSet::new(),
pending_state: Vec::new(),
own_version: 0,
@@ -617,6 +637,8 @@ impl Runtime {
tracing::debug!(%err, "could not queue re-announcement");
}
}
// What was announced decides which protocol each pair runs.
self.update_paths();
}
/// Queues a message without blocking the runtime loop.
@@ -669,6 +691,7 @@ impl Runtime {
// alone, so something has to look again; this runs on a timer and
// the check is a comparison when nothing has changed.
self.ensure_own_claim().await;
self.report_hostname_taken();
// The same timer repairs the relay picture: reachability is soft
// state, repeated while it holds and forgotten when it stops.
self.update_paths();
@@ -1031,12 +1054,25 @@ impl Runtime {
}
};
// A name keeps the time it was first claimed for as long as this
// member keeps the name, so a later claimant cannot overtake it by
// the passage of time, and a new or changed name starts the clock
// afresh.
let hostname_since = wanted_hostname.as_ref().map(|name| {
let kept = self
.state
.get(&self.local_id)
.filter(|record| record.body.hostname() == Some(name.as_str()))
.and_then(|record| record.body.hostname_since());
kept.unwrap_or_else(unix_millis)
});
let body = RecordBody::Claim {
address: wanted_address,
// The range only travels with an address, so a member of an
// IPv6-only network does not assert one.
range: wanted_address.and(range),
hostname: wanted_hostname,
hostname_since,
};
// Nothing to say is not the same as saying nothing changed: a member
@@ -1100,6 +1136,7 @@ impl Runtime {
for (peer, records) in std::mem::take(&mut self.pending_state) {
self.receive_state(peer, records).await;
}
self.report_hostname_taken();
}
/// Sends everything we know to every peer.
@@ -1658,68 +1695,93 @@ impl Runtime {
}
}
/// The protocol this agent and a member run their tunnel in, if they have
/// one in common.
///
/// Both ends must name the same one without asking each other, and what
/// each side was configured to prefer can differ. So the member with the
/// lower id decides, by the order it announces its protocols in, and the
/// other side reads that order from the announcement it already has.
fn tunnel_protocol(&self, session: &Session) -> Option<String> {
let usable = |ours: &[PluginCapability], theirs: &[PluginCapability], name: &str| {
ours.iter().any(|capability| {
capability.enabled
&& capability.protocol == name
&& theirs.iter().any(|other| {
other.enabled
&& other.protocol == name
&& other.version == capability.version
})
})
};
let (decides, other) = if self.local_id.as_bytes() < session.peer.as_bytes() {
(&self.local_capabilities, &session.capabilities)
} else {
(&session.capabilities, &self.local_capabilities)
};
decides
.iter()
.find(|capability| usable(decides, other, &capability.protocol))
.map(|capability| capability.protocol.clone())
}
/// Builds the graph off the packet path. Every remote row comes from that
/// member's authenticated control session; stale or incompatible rows never
/// enter the graph. The hub replaces an immutable table only on changes.
/// member's authenticated control session, and a row says only who that
/// member has a link to, never what kind of link: a member in the middle
/// needs no protocol in common with either end. Stale rows never enter the
/// graph. The hub replaces an immutable table only on changes.
fn update_paths(&mut self) {
self.reachable.retain(|peer, (_, heard)| {
heard.elapsed() < REACH_EXPIRY && self.sessions.contains_key(peer)
});
for (protocol, version) in self.served_protocols() {
let peers: Vec<_> = self
.sessions
.values()
.filter(|session| {
session.capabilities.iter().any(|capability| {
capability.enabled
&& capability.protocol == protocol
&& capability.version == version
})
})
.map(|session| session.peer)
.collect();
let mut members: HashSet<_> = peers.iter().map(|peer| *peer.as_bytes()).collect();
members.insert(*self.local_id.as_bytes());
let graph = self
.reachable
.iter()
.filter(|(peer, _)| members.contains(peer.as_bytes()))
.map(|(peer, (links, _))| {
let mut neighbors: Vec<_> = links
.iter()
.filter(|link| {
link.protocol == protocol
&& members.contains(&link.peer)
&& link.peer != *peer.as_bytes()
})
.map(|link| link.peer)
.collect();
neighbors.sort_unstable();
(*peer.as_bytes(), neighbors)
})
.collect();
self.hub.set_topology(&protocol, graph, members);
for peer in peers {
if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer, &protocol) {
continue;
}
let Some(plugin) = self
.params
.plugins
let mut members: HashSet<_> = self.sessions.keys().map(|peer| *peer.as_bytes()).collect();
members.insert(*self.local_id.as_bytes());
let graph = self
.reachable
.iter()
.filter(|(peer, _)| members.contains(peer.as_bytes()))
.map(|(peer, (links, _))| {
let mut neighbors: Vec<_> = links
.iter()
.find(|plugin| plugin.protocol_id() == protocol)
.cloned()
else {
continue;
};
let link = self.hub.link(peer, &protocol);
let path = link.path_description();
let max_datagram = link.max_datagram_size();
plugin.on_peer_link(self.network_id, peer, link);
.filter(|link| members.contains(&link.peer) && link.peer != *peer.as_bytes())
.map(|link| link.peer)
.collect();
neighbors.sort_unstable();
neighbors.dedup();
(*peer.as_bytes(), neighbors)
})
.collect();
self.hub.set_topology(graph, members);
let wanted: Vec<(EndpointId, String)> = self
.sessions
.values()
.filter_map(|session| Some((session.peer, self.tunnel_protocol(session)?)))
.collect();
for (peer, protocol) in wanted {
if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer) {
continue;
}
let Some(plugin) = self
.params
.plugins
.iter()
.find(|plugin| plugin.protocol_id() == protocol)
.cloned()
else {
continue;
};
let link = self.hub.link(peer, &protocol);
let relayed = self.hub.route(peer).is_some_and(|route| route.hops > 1);
let path = link.path_description();
let max_datagram = link.max_datagram_size();
plugin.on_peer_link(self.network_id, peer, link);
// A direct link has already been reported as one.
if relayed {
self.emit(Event::DataLinkUp {
network: self.network_id,
peer,
protocol: protocol.clone(),
protocol,
path,
max_datagram,
});
@@ -1786,30 +1848,25 @@ impl Runtime {
/// Hands a link to the plugin that owns its protocol.
fn adopt_link(&mut self, peer: EndpointId, protocol: String, link: SharedLink) {
let Some(plugin) = self
if !self
.params
.plugins
.iter()
.find(|plugin| plugin.protocol_id() == protocol)
.cloned()
else {
.any(|plugin| plugin.protocol_id() == protocol)
{
return;
};
}
let path = link.path_description();
let max_datagram = link.max_datagram_size();
self.links
.insert((peer, protocol.clone()), Arc::clone(&link));
// The protocol is handed the hub's link, not this one: the same
// object for as long as the peer is a peer, whatever happens to
// the path under it. A tunnel is not torn down because a direct
// link came or went.
let first_time = !self.hub.has_link(peer, &protocol);
// What this is, is a way to reach the peer. Which protocol's
// tunnel runs over it is decided in `update_paths`, and that tunnel
// is handed the hub's link, not this one: the same object for as
// long as the peer is a peer, whatever happens to the path under
// it. A tunnel is not torn down because a direct link came or went.
self.hub.set_direct(peer, &protocol, link);
let shared = self.hub.link(peer, &protocol);
let max_datagram = shared.max_datagram_size();
if first_time {
plugin.on_peer_link(self.network_id, peer, shared);
}
self.metrics.data_links_established += 1;
self.emit(Event::DataLinkUp {
network: self.network_id,
@@ -1941,6 +1998,13 @@ impl Runtime {
});
}
capabilities.truncate(self.params.limits.max_capabilities);
self.local_capabilities = capabilities
.iter()
.map(|capability| PluginCapability {
data: Vec::new(),
..capability.clone()
})
.collect();
let exit_node = self.exit_ready();
self.announced_exit = exit_node;
Announcement {
@@ -2175,6 +2239,69 @@ impl Runtime {
// ----------------------------------------------------------------- status
/// Whether a member can be reached now, by any path.
///
/// The single definition every client reports: a live control session,
/// or a data path to it through other members. Being reached over a long
/// path is being online; how well is `data_path`'s to say.
fn is_online(&self, peer: EndpointId) -> bool {
peer == self.local_id || self.sessions.contains_key(&peer) || self.hub.reachable(peer)
}
/// How packets reach a member, which says nothing about whether it is
/// there.
fn data_path(&self, peer: EndpointId) -> DataPath {
match self.hub.route(peer) {
None => DataPath::None,
Some(route) if route.hops <= 1 => DataPath::Direct {
transport: self
.hub
.direct_kinds(peer)
.into_iter()
.next()
.unwrap_or_default(),
},
Some(route) => DataPath::Relayed {
hops: route.hops,
via: route.via,
},
}
}
/// Says so when this agent's own name is held by someone else, and
/// again if that changes. Never anything but a statement: the agent keeps
/// working by address either way.
fn report_hostname_taken(&mut self) {
let taken = self.hostname_conflict_of(&self.local_id);
if taken == self.reported_name_taken {
return;
}
if let Some(conflict) = &taken {
tracing::warn!(
network = %self.network_id,
name = %conflict.name,
holder = %conflict.holder.fmt_short(),
"this name was claimed first by another member; it stays reachable by \
address but is not found by this name until it is renamed"
);
self.emit(Event::HostnameTaken {
network: self.network_id,
name: conflict.name.clone(),
holder: conflict.holder,
});
}
self.reported_name_taken = taken;
}
fn hostname_conflict_of(&self, author: &EndpointId) -> Option<HostnameConflict> {
self.state
.hostname_taken(author)
.map(|(name, holder)| HostnameConflict {
name: name.to_string(),
holder,
})
}
fn status(&self) -> NetworkStatus {
let served = self.served_protocols();
let mut peers: Vec<PeerStatus> = self
@@ -2212,6 +2339,7 @@ impl Runtime {
control_messages_received: session.messages_received,
control_bytes_sent: session.bytes_sent,
control_bytes_received: session.bytes_received,
data_path: self.data_path(session.peer),
}
})
.collect();
@@ -2249,6 +2377,9 @@ impl Runtime {
overlay_address_v4: self.state.address_of(&endpoint_id),
hostname: self.state.hostname_of(&endpoint_id).map(str::to_string),
last_hostname: self.known_hostnames.get(&endpoint_id).cloned(),
hostname_conflict: self.hostname_conflict_of(&endpoint_id),
online: self.is_online(endpoint_id),
data_path: self.data_path(endpoint_id),
})
})
.collect();
@@ -2276,6 +2407,7 @@ impl Runtime {
peers,
candidates,
members,
hostname_conflict: self.hostname_conflict_of(&self.local_id),
range: self.effective_range(),
// Only worth reporting while it is actually stuck: with a
// fallback in play the configured range being another
@@ -2290,6 +2422,15 @@ impl Runtime {
}
}
/// Milliseconds since the Unix epoch, for the time a name was first claimed.
fn unix_millis() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(u64::MAX, |elapsed| {
u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
})
}
/// Which side dialled, given a role and the two identities.
fn initiator_of(role: Role, local: EndpointId, peer: EndpointId) -> EndpointId {
match role {
crates/tsunagi-core/src/agent/status.rs
+52
View File
@@ -59,6 +59,43 @@ pub struct OverlayStatus {
pub broadcast_rules: Option<crate::overlay::BroadcastRulesReport>,
}
/// How packets reach a member right now.
///
/// Being reachable is one thing and being reachable well is another, and
/// the second is never a reason to call a member away: a member that can only
/// be reached through somebody else is online, over a longer path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DataPath {
/// Nothing carries packets to it yet. The member is connected for
/// control only, which is the state of a data path being set up.
None,
/// A direct link, of the named kind of transport.
Direct {
/// The kind of transport carrying it, e.g. `wg`.
transport: String,
},
/// Through other members.
Relayed {
/// Number of links on the way, at least two.
hops: u8,
/// The member the first of them goes to.
via: EndpointId,
},
}
/// A name that a member claimed and somebody else holds.
///
/// A name belongs to whoever claimed it first. The member that claimed it
/// later is still a member and still reachable by its address; it is just
/// not found by that name, on any member of the network.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HostnameConflict {
/// The name that is taken.
pub name: String,
/// The member that holds it.
pub holder: EndpointId,
}
/// A member the signed state knows about, connected or not.
///
/// This is the durable roster: it comes from signed records, so a member that
@@ -85,6 +122,15 @@ pub struct MemberStatus {
/// Not signed, so it is a convenience for naming a member that is away
/// and never a claim: the zone is served from `hostname`.
pub last_hostname: Option<String>,
/// The name it claimed, when somebody else claimed it first.
pub hostname_conflict: Option<HostnameConflict>,
/// Whether this agent can reach it now, by any path: a live control
/// session, or a data path through other members. The one answer to "is
/// it online" for every client, so that none of them has to work it out
/// from a different part of the picture. This agent counts as online.
pub online: bool,
/// How packets reach it. [`DataPath::None`] while it is offline.
pub data_path: DataPath,
}
/// Status of one authenticated session.
@@ -128,6 +174,8 @@ pub struct PeerStatus {
pub control_bytes_sent: u64,
/// Control payload bytes read from this session.
pub control_bytes_received: u64,
/// How packets reach this peer right now.
pub data_path: DataPath,
}
/// Counters scoped to one logical network.
@@ -188,6 +236,10 @@ pub struct NetworkStatus {
pub candidates: Vec<CandidateStatus>,
/// Members the signed state knows about, whether connected or not.
pub members: Vec<MemberStatus>,
/// This agent's own name, when another member claimed it first and so
/// holds it. This agent keeps working by address; nobody finds it by
/// this name until it is renamed or the other member lets it go.
pub hostname_conflict: Option<HostnameConflict>,
/// The overlay range this network uses, once it has one.
pub range: Option<crate::state::Ipv4Range>,
/// The range it could not have, because another network on this agent
crates/tsunagi-core/src/dataplane/relay.rs
+259 -127
View File
@@ -67,10 +67,41 @@ impl Tally {
}
}
/// The links to one neighbour, best first. Which transport carries a frame
/// is decided here and nowhere else: the end-to-end protocol is in the
/// envelope and never depends on it.
#[derive(Debug)]
struct NextHop {
peer: PeerId,
link: SharedLink,
links: Box<[SharedLink]>,
}
impl NextHop {
fn usable(&self) -> bool {
self.links.iter().any(|link| !link.is_closed())
}
fn first_open(&self) -> Option<&SharedLink> {
self.links.iter().find(|link| !link.is_closed())
}
/// Hands a frame to the best open link that accepts it. A link that
/// cannot carry the frame (too large for its path, closed a moment ago)
/// leaves it to the next one; the frame is only copied when there is
/// another to try.
fn send(&self, frame: Bytes) -> Result<(), TransportError> {
let mut open = self.links.iter().filter(|link| !link.is_closed());
let Some(mut current) = open.next() else {
return Err(TransportError::Closed);
};
for next in open {
if current.send(frame.clone()).is_ok() {
return Ok(());
}
current = next;
}
current.send(frame)
}
}
#[derive(Debug)]
@@ -82,21 +113,22 @@ struct ForwardRoute {
impl ForwardRoute {
fn select(&self, source: &PeerId, flow: FlowId) -> Option<&NextHop> {
if self.next.len() == 1 {
return self.next.first().filter(|hop| !hop.link.is_closed());
return self.next.first().filter(|hop| hop.usable());
}
let seed = u64::from_le_bytes(source[..8].try_into().ok()?);
let start = mix64(flow ^ seed) as usize % self.next.len();
// Stable fallback if a link closed just before its table was replaced.
(0..self.next.len())
.map(|i| &self.next[(start + i) % self.next.len()])
.find(|hop| !hop.link.is_closed())
.find(|hop| hop.usable())
}
}
#[derive(Debug, Default)]
struct ForwardingTable {
routes: HashMap<PeerId, ForwardRoute>,
inboxes: HashMap<PeerId, mpsc::Sender<Bytes>>,
/// Local delivery, by who sent it and which end-to-end protocol it is.
inboxes: HashMap<(PeerId, u64), mpsc::Sender<Bytes>>,
members: HashSet<PeerId>,
}
@@ -121,7 +153,7 @@ impl Plane {
return;
}
if header.destination == self.local {
let Some(inbox) = table.inboxes.get(&header.source) else {
let Some(inbox) = table.inboxes.get(&(header.source, header.protocol)) else {
self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed);
return;
};
@@ -144,7 +176,7 @@ impl Plane {
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
return;
};
if hop.link.send(envelope::decrement(frame)).is_ok() {
if hop.send(envelope::decrement(frame)).is_ok() {
self.tally.forwarded.fetch_add(1, Ordering::Relaxed);
} else {
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
@@ -152,27 +184,65 @@ impl Plane {
}
}
/// One physical link to a neighbour, of whatever kind of transport.
#[derive(Debug)]
struct Protocol {
plane: Arc<Plane>,
graph: HashMap<PeerId, Vec<PeerId>>,
members: HashSet<PeerId>,
raw: HashMap<PeerId, SharedLink>,
readers: HashMap<PeerId, JoinHandle<()>>,
links: HashMap<PeerId, Arc<PeerLink>>,
struct RawLink {
kind: String,
link: SharedLink,
reader: JoinHandle<()>,
}
impl Protocol {
/// How a destination is reached right now.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RouteSummary {
/// Number of transport links to it. One is a direct link.
pub hops: u8,
/// The neighbour the first link goes to. For a direct link that is the
/// destination itself.
pub via: EndpointId,
}
#[derive(Debug)]
struct State {
plane: Arc<Plane>,
/// Transport kinds, best first. A neighbour with several links is
/// reached over the first of these it has.
priority: Vec<String>,
graph: HashMap<PeerId, Vec<PeerId>>,
members: HashSet<PeerId>,
raw: HashMap<PeerId, Vec<RawLink>>,
links: HashMap<(PeerId, String), Arc<PeerLink>>,
}
impl State {
fn rank(&self, kind: &str) -> usize {
self.priority
.iter()
.position(|known| known == kind)
.unwrap_or(usize::MAX)
}
/// A neighbour's links, best kind first.
fn ordered(&self, peer: &PeerId) -> Vec<&RawLink> {
let mut links: Vec<&RawLink> = self
.raw
.get(peer)
.map(|links| links.iter().collect())
.unwrap_or_default();
links.sort_by_key(|raw| (self.rank(&raw.kind), raw.kind.clone()));
links
}
fn publish(&self) {
let mut graph = self.graph.clone();
graph.insert(
self.plane.local,
self.raw
.iter()
.filter(|(_, link)| !link.is_closed())
.map(|(peer, _)| *peer)
.collect(),
);
let mut neighbours: Vec<PeerId> = self
.raw
.iter()
.filter(|(_, links)| links.iter().any(|raw| !raw.link.is_closed()))
.map(|(peer, _)| *peer)
.collect();
neighbours.sort_unstable();
graph.insert(self.plane.local, neighbours);
let routing = RoutingTable::build(self.plane.local, &graph, ROUTING_HOP_LIMIT);
let routes = routing
.iter()
@@ -181,10 +251,12 @@ impl Protocol {
.next_hops
.iter()
.filter_map(|id| {
self.raw.get(id).map(|link| NextHop {
peer: *id,
link: link.clone(),
})
let links: Box<[_]> = self
.ordered(id)
.into_iter()
.map(|raw| raw.link.clone())
.collect();
(!links.is_empty()).then_some(NextHop { peer: *id, links })
})
.collect();
(!next.is_empty()).then_some((
@@ -204,17 +276,21 @@ impl Protocol {
inboxes: self
.links
.iter()
.map(|(peer, link)| (*peer, link.inbox_tx.clone()))
.map(|((peer, _), link)| ((*peer, link.tag), link.inbox_tx.clone()))
.collect(),
}));
}
fn stop_readers(&mut self) {
for raw in self.raw.values().flatten() {
raw.reader.abort();
}
}
}
impl Drop for Protocol {
impl Drop for State {
fn drop(&mut self) {
for task in self.readers.values() {
task.abort();
}
self.stop_readers();
for link in self.links.values() {
link.close();
}
@@ -222,12 +298,17 @@ impl Drop for Protocol {
}
}
/// One network's transport links and independently published protocol tables.
/// One network's transport links and its published forwarding table.
///
/// There is one table for the whole network, whatever protocols run over it.
/// A link is a way to reach a neighbour; a protocol is an end-to-end channel
/// to a peer. The two meet only in the envelope, which names the protocol, so
/// a tunnel keeps working over any mix of transports and a middle member
/// needs no protocol in common with either end.
#[derive(Debug)]
pub struct RelayHub {
network: NetworkId,
local: PeerId,
protocols: Mutex<HashMap<String, Protocol>>,
state: Mutex<State>,
tally: Arc<Tally>,
changed: Arc<Notify>,
}
@@ -235,40 +316,38 @@ pub struct RelayHub {
impl RelayHub {
/// Creates the router at this network's local endpoint.
pub fn new(network: NetworkId, local: EndpointId) -> Arc<Self> {
let tally = Arc::new(Tally::default());
Arc::new(Self {
network,
local: *local.as_bytes(),
protocols: Mutex::default(),
tally: Arc::default(),
state: Mutex::new(State {
plane: Arc::new(Plane {
local: *local.as_bytes(),
table: ArcSwap::from_pointee(ForwardingTable::default()),
tally: tally.clone(),
}),
priority: Vec::new(),
graph: HashMap::new(),
members: HashSet::new(),
raw: HashMap::new(),
links: HashMap::new(),
}),
tally,
changed: Arc::default(),
})
}
fn protocols(&self) -> std::sync::MutexGuard<'_, HashMap<String, Protocol>> {
self.protocols
.lock()
.unwrap_or_else(|error| error.into_inner())
fn state(&self) -> std::sync::MutexGuard<'_, State> {
self.state.lock().unwrap_or_else(|error| error.into_inner())
}
fn protocol<'a>(
&self,
protocols: &'a mut HashMap<String, Protocol>,
name: &str,
) -> &'a mut Protocol {
protocols
.entry(name.to_owned())
.or_insert_with(|| Protocol {
plane: Arc::new(Plane {
local: self.local,
table: ArcSwap::from_pointee(ForwardingTable::default()),
tally: self.tally.clone(),
}),
graph: HashMap::new(),
members: HashSet::new(),
raw: HashMap::new(),
readers: HashMap::new(),
links: HashMap::new(),
})
/// Sets which kinds of transport are preferred, best first, when a
/// neighbour can be reached over more than one.
pub fn set_priority(&self, kinds: Vec<String>) {
let mut state = self.state();
if state.priority != kinds {
state.priority = kinds;
state.publish();
}
}
/// Wakes the control loop promptly when a transport reader ends.
@@ -282,14 +361,11 @@ impl RelayHub {
}
/// Publishes authenticated topology. Refreshing identical state is cheap.
pub fn set_topology(
&self,
protocol: &str,
graph: HashMap<PeerId, Vec<PeerId>>,
members: HashSet<PeerId>,
) {
let mut protocols = self.protocols();
let state = self.protocol(&mut protocols, protocol);
///
/// Every row of `graph` is one member's own account of who it has a link
/// to; what kind of link it is does not matter here.
pub fn set_topology(&self, graph: HashMap<PeerId, Vec<PeerId>>, members: HashSet<PeerId>) {
let mut state = self.state();
if state.graph != graph || state.members != members {
state.graph = graph;
state.members = members;
@@ -298,22 +374,44 @@ impl RelayHub {
}
/// Whether the current table has a path to this destination.
pub fn reachable(&self, peer: EndpointId, protocol: &str) -> bool {
self.protocols().get(protocol).is_some_and(|state| {
state
.plane
.table
.load()
.routes
.contains_key(peer.as_bytes())
pub fn reachable(&self, peer: EndpointId) -> bool {
self.state()
.plane
.table
.load()
.routes
.contains_key(peer.as_bytes())
}
/// How a destination is reached right now, if it is.
pub fn route(&self, peer: EndpointId) -> Option<RouteSummary> {
let table = self.state().plane.table.load_full();
let route = table.routes.get(peer.as_bytes())?;
let via = EndpointId::from_bytes(&route.next.first()?.peer).ok()?;
Some(RouteSummary {
hops: route.hops,
via,
})
}
/// Stable logical link, preserved across topology changes.
/// The kinds of transport with an open direct link to a peer, best
/// first.
pub fn direct_kinds(&self, peer: EndpointId) -> Vec<String> {
let state = self.state();
state
.ordered(peer.as_bytes())
.into_iter()
.filter(|raw| !raw.link.is_closed())
.map(|raw| raw.kind.clone())
.collect()
}
/// Stable end-to-end channel for a protocol, preserved across topology
/// and transport changes.
pub fn link(&self, peer: EndpointId, protocol: &str) -> Arc<PeerLink> {
let mut protocols = self.protocols();
let state = self.protocol(&mut protocols, protocol);
if let Some(link) = state.links.get(peer.as_bytes()) {
let mut state = self.state();
let key = (*peer.as_bytes(), protocol.to_owned());
if let Some(link) = state.links.get(&key) {
return link.clone();
}
let (inbox_tx, inbox_rx) = mpsc::channel(INBOX);
@@ -321,88 +419,118 @@ impl RelayHub {
let link = Arc::new(PeerLink {
network: self.network,
peer,
tag: envelope::protocol_tag(protocol),
plane: state.plane.clone(),
inbox_tx,
inbox_rx: tokio::sync::Mutex::new(inbox_rx),
closed,
});
state.links.insert(*peer.as_bytes(), link.clone());
state.links.insert(key, link.clone());
state.publish();
link
}
/// Whether the protocol already owns its logical link.
/// Whether the protocol already owns its end-to-end channel to a peer.
pub fn has_link(&self, peer: EndpointId, protocol: &str) -> bool {
self.protocols()
.get(protocol)
.is_some_and(|state| state.links.contains_key(peer.as_bytes()))
self.state()
.links
.contains_key(&(*peer.as_bytes(), protocol.to_owned()))
}
/// Installs a raw transport link and starts its independent ingress reader.
pub fn set_direct(&self, peer: EndpointId, protocol: &str, raw: SharedLink) {
///
/// A neighbour may have one link per kind of transport; installing a
/// second of the same kind replaces the first.
pub fn set_direct(&self, peer: EndpointId, kind: &str, raw: SharedLink) {
if raw.network() != self.network || raw.peer() != peer {
return;
}
let mut protocols = self.protocols();
let state = self.protocol(&mut protocols, protocol);
let mut state = self.state();
let id = *peer.as_bytes();
if let Some(old) = state.readers.remove(&id) {
old.abort();
}
state.raw.insert(id, raw.clone());
state.publish();
let plane = state.plane.clone();
let changed = self.changed.clone();
state.readers.insert(
id,
tokio::spawn(async move {
let mut batch = 0;
while let Some(frame) = raw.recv().await {
plane.receive(id, frame);
batch += 1;
if batch == 64 {
tokio::task::yield_now().await;
batch = 0;
}
let reader_link = raw.clone();
let reader = tokio::spawn(async move {
let mut batch = 0;
while let Some(frame) = reader_link.recv().await {
plane.receive(id, frame);
batch += 1;
if batch == 64 {
tokio::task::yield_now().await;
batch = 0;
}
changed.notify_one();
}),
);
}
changed.notify_one();
});
let links = state.raw.entry(id).or_default();
if let Some(position) = links.iter().position(|known| known.kind == kind) {
links.remove(position).reader.abort();
}
links.push(RawLink {
kind: kind.to_owned(),
link: raw,
reader,
});
state.publish();
}
/// Removes a physical link without tearing down end-to-end protocol state.
pub fn clear_direct(&self, peer: EndpointId, protocol: &str) {
if let Some(state) = self.protocols().get_mut(protocol) {
state.raw.remove(peer.as_bytes());
if let Some(task) = state.readers.remove(peer.as_bytes()) {
task.abort();
/// Removes one physical link without tearing down end-to-end protocol state.
pub fn clear_direct(&self, peer: EndpointId, kind: &str) {
let mut state = self.state();
let mut changed = false;
if let Some(links) = state.raw.get_mut(peer.as_bytes()) {
if let Some(position) = links.iter().position(|known| known.kind == kind) {
links.remove(position).reader.abort();
changed = true;
}
if links.is_empty() {
state.raw.remove(peer.as_bytes());
}
}
if changed {
state.publish();
}
}
/// Revokes all links and topology involving a departed authenticated peer.
pub fn remove_peer(&self, peer: EndpointId) {
for state in self.protocols().values_mut() {
state.raw.remove(peer.as_bytes());
if let Some(task) = state.readers.remove(peer.as_bytes()) {
task.abort();
let mut state = self.state();
if let Some(links) = state.raw.remove(peer.as_bytes()) {
for raw in links {
raw.reader.abort();
}
if let Some(link) = state.links.remove(peer.as_bytes()) {
}
let gone: Vec<_> = state
.links
.keys()
.filter(|(id, _)| id == peer.as_bytes())
.cloned()
.collect();
for key in gone {
if let Some(link) = state.links.remove(&key) {
link.close();
}
state.members.remove(peer.as_bytes());
state.graph.remove(peer.as_bytes());
for neighbors in state.graph.values_mut() {
neighbors.retain(|id| id != peer.as_bytes());
}
state.publish();
}
state.members.remove(peer.as_bytes());
state.graph.remove(peer.as_bytes());
for neighbors in state.graph.values_mut() {
neighbors.retain(|id| id != peer.as_bytes());
}
state.publish();
}
/// Stops every reader and releases all transport handles.
pub fn close(&self) {
self.protocols().clear();
let mut state = self.state();
state.stop_readers();
state.raw.clear();
for link in state.links.values() {
link.close();
}
state.links.clear();
state.graph.clear();
state.members.clear();
state.plane.table.store(Arc::default());
}
}
@@ -411,6 +539,7 @@ impl RelayHub {
pub struct PeerLink {
network: NetworkId,
peer: EndpointId,
tag: u64,
plane: Arc<Plane>,
inbox_tx: mpsc::Sender<Bytes>,
inbox_rx: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
@@ -454,10 +583,11 @@ impl PacketLink for PeerLink {
let hop = route
.select(&self.plane.local, flow)
.ok_or(TransportError::Closed)?;
hop.link.send(envelope::encode(
hop.send(envelope::encode(
self.plane.local,
*self.peer.as_bytes(),
flow,
self.tag,
&payload,
))?;
if route.hops > 1 {
@@ -489,7 +619,9 @@ impl PacketLink for PeerLink {
fn path_description(&self) -> String {
let table = self.plane.table.load();
match table.routes.get(self.peer.as_bytes()) {
Some(route) if route.hops == 1 => route.next[0].link.path_description(),
Some(route) if route.hops == 1 => route.next[0]
.first_open()
.map_or_else(|| "unreachable".into(), |link| link.path_description()),
Some(route) => format!(
"relay {} hops via {} ({} equal paths)",
route.hops,
crates/tsunagi-core/src/dataplane/relay_tests.rs
+147 -16
View File
@@ -16,6 +16,9 @@ fn peer(seed: u8) -> EndpointId {
fn id(seed: u8) -> PeerId {
*peer(seed).as_bytes()
}
fn tag() -> u64 {
envelope::protocol_tag("ip")
}
#[derive(Debug)]
struct Wire {
@@ -78,9 +81,8 @@ impl PacketLink for Wire {
}
}
fn topology(hub: &RelayHub, protocol: &str, edges: &[(u8, &[u8])], members: &[u8]) {
fn topology(hub: &RelayHub, edges: &[(u8, &[u8])], members: &[u8]) {
hub.set_topology(
protocol,
edges
.iter()
.map(|(a, bs)| (id(*a), bs.iter().map(|b| id(*b)).collect()))
@@ -96,7 +98,7 @@ async fn ecmp_is_per_flow_direct_wins_and_existing_tunnel_survives_changes() {
let c = Wire::new(3, true);
hub.set_direct(peer(2), "ip", b.clone());
hub.set_direct(peer(3), "ip", c.clone());
topology(&hub, "ip", &[(2, &[4]), (3, &[4])], &[1, 2, 3, 4]);
topology(&hub, &[(2, &[4]), (3, &[4])], &[1, 2, 3, 4]);
let link = hub.link(peer(4), "ip");
let mut chosen = HashMap::new();
for _ in 0..8 {
@@ -132,7 +134,7 @@ async fn ecmp_is_per_flow_direct_wins_and_existing_tunnel_survives_changes() {
}
assert_eq!(c.take().len(), 100);
hub.clear_direct(peer(2), "ip");
topology(&hub, "ip", &[(3, &[])], &[1, 2, 3, 4]);
topology(&hub, &[(3, &[])], &[1, 2, 3, 4]);
assert!(link.send(Bytes::new()).is_err());
assert!(!link.is_closed());
assert!(Arc::ptr_eq(&link, &hub.link(peer(4), "ip")));
@@ -152,8 +154,8 @@ async fn transport_reader_forwards_without_any_plugin_reader_and_isolates_protoc
let c = Wire::new(3, true);
hub.set_direct(peer(1), "ip", a.clone());
hub.set_direct(peer(3), "ip", c.clone());
topology(&hub, "ip", &[(3, &[4])], &[1, 2, 3, 4]);
let frame = envelope::encode(id(1), id(4), 91, b"end-to-end encrypted");
topology(&hub, &[(3, &[4])], &[1, 2, 3, 4]);
let frame = envelope::encode(id(1), id(4), 91, tag(), b"end-to-end encrypted");
let pointer = frame.as_ptr();
a.feed.send(frame).await.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(1), async {
@@ -170,12 +172,136 @@ async fn transport_reader_forwards_without_any_plugin_reader_and_isolates_protoc
ROUTING_HOP_LIMIT - 1
);
assert_eq!(&sent[0][RELAY_OVERHEAD..], b"end-to-end encrypted");
let isolated = hub.link(peer(4), "other-ip");
assert!(isolated.send(Bytes::new()).is_err());
assert_eq!(hub.counters().forwarded, 1);
hub.close();
}
#[tokio::test]
async fn a_frame_is_delivered_only_to_the_protocol_it_names() {
let hub = RelayHub::new(network(), peer(2));
hub.set_direct(peer(1), "ip", Wire::new(1, true));
let ip = hub.link(peer(1), "ip");
let other = hub.link(peer(1), "other-ip");
let plane = ip.plane.clone();
plane.receive(id(1), envelope::encode(id(1), id(2), 0, tag(), b"for ip"));
plane.receive(
id(1),
envelope::encode(
id(1),
id(2),
0,
envelope::protocol_tag("other-ip"),
b"for other",
),
);
plane.receive(
id(1),
envelope::encode(
id(1),
id(2),
0,
envelope::protocol_tag("nobody-runs-this"),
b"for nobody",
),
);
assert_eq!(ip.recv().await.unwrap(), Bytes::from_static(b"for ip"));
assert_eq!(
other.recv().await.unwrap(),
Bytes::from_static(b"for other")
);
assert_eq!(hub.counters().dropped_unknown, 1);
hub.close();
}
/// A middle member joins two neighbours over different transports, and the
/// two ends share a protocol the middle has never heard of.
#[tokio::test]
async fn a_tunnel_crosses_a_middle_member_whose_links_are_of_different_kinds() {
let a = RelayHub::new(network(), peer(1));
let c = RelayHub::new(network(), peer(3));
let b = RelayHub::new(network(), peer(2));
let a_to_c = Wire::new(3, true);
let c_to_a = Wire::new(1, true);
let c_to_b = Wire::new(2, true);
let b_to_c = Wire::new(3, true);
a.set_direct(peer(3), "wg", a_to_c.clone());
c.set_direct(peer(1), "wg", c_to_a);
c.set_direct(peer(2), "wg-quic", c_to_b.clone());
b.set_direct(peer(3), "wg-quic", b_to_c.clone());
topology(&a, &[(3, &[1, 2])], &[1, 2, 3]);
topology(&b, &[(3, &[1, 2])], &[1, 2, 3]);
topology(&c, &[(1, &[3]), (2, &[3])], &[1, 2, 3]);
let outgoing = a.link(peer(2), "tunnel");
let incoming = b.link(peer(1), "tunnel");
outgoing
.send_flow(Bytes::from_static(b"sealed"), 5)
.unwrap();
let frame = a_to_c.take().remove(0);
// The middle forwards it onto its other kind of link without looking
// inside.
c.state().plane.receive(id(1), frame);
let frame = c_to_b.take().remove(0);
assert_eq!(c.counters().forwarded, 1);
b.state().plane.receive(id(3), frame);
assert_eq!(
incoming.recv().await.unwrap(),
Bytes::from_static(b"sealed")
);
assert_eq!(b.counters().received_via, 1);
assert_eq!(
b.route(peer(1)),
Some(RouteSummary {
hops: 2,
via: peer(3)
})
);
a.close();
b.close();
c.close();
}
#[tokio::test]
async fn a_neighbour_is_reached_over_the_preferred_kind_and_falls_back_when_it_is_lost() {
let hub = RelayHub::new(network(), peer(1));
hub.set_priority(vec!["wg".into(), "tcp-tls".into(), "wg-quic".into()]);
let slow = Wire::new(2, true);
let fast = Wire::new(2, true);
hub.set_direct(peer(2), "wg-quic", slow.clone());
let link = hub.link(peer(2), "tunnel");
link.send(Bytes::from_static(b"one")).unwrap();
assert_eq!(slow.take().len(), 1);
assert_eq!(hub.direct_kinds(peer(2)), ["wg-quic"]);
// A better kind appears: traffic moves to it with nothing to redo.
hub.set_direct(peer(2), "wg", fast.clone());
assert_eq!(hub.direct_kinds(peer(2)), ["wg", "wg-quic"]);
link.send(Bytes::from_static(b"two")).unwrap();
assert_eq!(fast.take().len(), 1);
assert!(slow.take().is_empty());
// It dies: the worse one carries on, and the logical link is untouched.
fast.dead.store(true, Ordering::Relaxed);
link.send(Bytes::from_static(b"three")).unwrap();
assert_eq!(slow.take().len(), 1);
assert!(!link.is_closed());
assert_eq!(
hub.route(peer(2)),
Some(RouteSummary {
hops: 1,
via: peer(2)
})
);
hub.clear_direct(peer(2), "wg-quic");
hub.clear_direct(peer(2), "wg");
assert!(!hub.reachable(peer(2)));
assert!(Arc::ptr_eq(&link, &hub.link(peer(2), "tunnel")));
hub.close();
}
#[tokio::test]
async fn local_delivery_is_bounded_and_malformed_or_unknown_sources_are_dropped() {
let hub = RelayHub::new(network(), peer(2));
@@ -184,18 +310,21 @@ async fn local_delivery_is_bounded_and_malformed_or_unknown_sources_are_dropped(
let link = hub.link(peer(1), "ip");
let plane = link.plane.clone();
for _ in 0..INBOX + 1 {
plane.receive(id(1), envelope::encode(id(1), id(2), 0, b"ip"));
plane.receive(id(1), envelope::encode(id(1), id(2), 0, tag(), b"ip"));
}
assert_eq!(hub.counters().dropped_congested, 1);
for _ in 0..INBOX {
assert_eq!(link.recv().await.unwrap(), Bytes::from_static(b"ip"));
}
plane.receive(id(1), Bytes::from_static(b"broken"));
plane.receive(id(1), envelope::encode(id(9), id(2), 0, b"unknown member"));
plane.receive(
id(1),
envelope::encode(id(9), id(2), 0, tag(), b"unknown member"),
);
assert_eq!(hub.counters().dropped_unknown, 2);
plane.receive(
id(1),
envelope::encode(id(1), id(9), 0, b"unknown destination"),
envelope::encode(id(1), id(9), 0, tag(), b"unknown destination"),
);
assert_eq!(hub.counters().dropped_no_link, 1);
hub.close();
@@ -210,11 +339,11 @@ async fn inconsistent_tables_cannot_loop_beyond_the_hop_limit() {
a.set_direct(peer(2), "ip", ab.clone());
b.set_direct(peer(1), "ip", ba.clone());
// Deliberately inconsistent snapshots while updates are in flight.
topology(&a, "ip", &[(2, &[4])], &[1, 2, 3, 4]);
topology(&b, "ip", &[(1, &[4])], &[1, 2, 3, 4]);
topology(&a, &[(2, &[4])], &[1, 2, 3, 4]);
topology(&b, &[(1, &[4])], &[1, 2, 3, 4]);
let ap = a.link(peer(4), "ip").plane.clone();
let bp = b.link(peer(4), "ip").plane.clone();
let mut frame = envelope::encode(id(3), id(4), 7, b"transit");
let mut frame = envelope::encode(id(3), id(4), 7, tag(), b"transit");
for turn in 0..ROUTING_HOP_LIMIT {
let (plane, incoming, wire) = if turn % 2 == 0 {
(&ap, id(2), &ab)
@@ -253,7 +382,7 @@ async fn forwarding_benchmark() {
hub.set_direct(peer(3), "ip", wire.clone());
let alternate = Wire::new(5, false);
hub.set_direct(peer(5), "ip", alternate.clone());
topology(&hub, "ip", &[(3, &[4]), (5, &[4])], &[1, 2, 3, 4, 5]);
topology(&hub, &[(3, &[4]), (5, &[4])], &[1, 2, 3, 4, 5]);
let plane = hub.link(peer(4), "ip").plane.clone();
let source = id(1);
let destination = id(4);
@@ -264,7 +393,9 @@ async fn forwarding_benchmark() {
const BATCHES: usize = 100;
for _ in 0..BATCHES {
let frames: Vec<_> = (0..COUNT)
.map(|flow| envelope::encode(source, destination, flow as u64, &vec![42; size]))
.map(|flow| {
envelope::encode(source, destination, flow as u64, tag(), &vec![42; size])
})
.collect();
let start = std::time::Instant::now();
for frame in frames {
crates/tsunagi-core/src/dataplane/routing/envelope.rs
+25 -5
View File
@@ -3,8 +3,8 @@ use super::{FlowId, PeerId};
use crate::config::{MAX_DATA_DATAGRAM, ROUTING_HOP_LIMIT};
use bytes::{BufMut, Bytes, BytesMut};
pub(crate) const HEADER: usize = 2 + 32 + 32 + 8;
const VERSION: u8 = 1;
pub(crate) const HEADER: usize = 2 + 32 + 32 + 8 + 8;
const VERSION: u8 = 2;
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Header {
@@ -12,6 +12,15 @@ pub(crate) struct Header {
pub destination: PeerId,
pub remaining: u8,
pub flow: FlowId,
/// Which end-to-end protocol the payload belongs to, whatever links carry it.
pub protocol: u64,
}
/// A stable 64-bit tag for a protocol id (FNV-1a), the same on every member.
pub(crate) fn protocol_tag(name: &str) -> u64 {
name.bytes().fold(0xcbf2_9ce4_8422_2325, |hash, byte| {
(hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
})
}
pub(crate) fn decode(bytes: &[u8]) -> Option<Header> {
@@ -28,16 +37,24 @@ pub(crate) fn decode(bytes: &[u8]) -> Option<Header> {
destination: bytes[34..66].try_into().ok()?,
remaining: bytes[1],
flow: u64::from_be_bytes(bytes[66..74].try_into().ok()?),
protocol: u64::from_be_bytes(bytes[74..82].try_into().ok()?),
})
}
pub(crate) fn encode(source: PeerId, destination: PeerId, flow: FlowId, payload: &[u8]) -> Bytes {
pub(crate) fn encode(
source: PeerId,
destination: PeerId,
flow: FlowId,
protocol: u64,
payload: &[u8],
) -> Bytes {
let mut bytes = BytesMut::with_capacity(HEADER + payload.len());
bytes.put_u8(VERSION);
bytes.put_u8(ROUTING_HOP_LIMIT);
bytes.extend_from_slice(&source);
bytes.extend_from_slice(&destination);
bytes.put_u64(flow);
bytes.put_u64(protocol);
bytes.extend_from_slice(payload);
bytes.freeze()
}
@@ -58,7 +75,7 @@ mod tests {
use super::*;
#[test]
fn transit_reuses_the_buffer_and_rejects_invalid_headers() {
let frame = encode([1; 32], [2; 32], 17, &[42; 1280]);
let frame = encode([1; 32], [2; 32], 17, protocol_tag("wg"), &[42; 1280]);
let address = frame.as_ptr();
let frame = decrement(frame);
assert_eq!(
@@ -66,7 +83,10 @@ mod tests {
address,
"no payload copy for an owned buffer"
);
assert_eq!(decode(&frame).unwrap().remaining, ROUTING_HOP_LIMIT - 1);
let header = decode(&frame).unwrap();
assert_eq!(header.remaining, ROUTING_HOP_LIMIT - 1);
assert_eq!(header.protocol, protocol_tag("wg"));
assert_ne!(protocol_tag("wg"), protocol_tag("wg-quic"));
assert_eq!(&frame[HEADER..], &[42; 1280]);
for length in 0..HEADER {
assert!(decode(&frame[..length]).is_none());
crates/tsunagi-core/src/ipc/mod.rs
+44 -1
View File
@@ -484,11 +484,43 @@ pub struct NetworkReport {
pub range_conflict: Option<String>,
/// The overlay, when a protocol is running one.
pub overlay: Option<OverlayReport>,
/// This agent's own name, when another member claimed it first.
pub hostname_conflict: Option<HostnameConflictReport>,
/// Exit-node settings: whether this agent offers one, and whether it
/// uses one.
pub exit: ExitReport,
}
/// How packets reach a member right now.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum DataPathReport {
/// Nothing carries packets to it: it is away, or only connected for
/// control while a data path is being set up.
#[default]
None,
/// A direct link.
Direct {
/// The kind of transport carrying it, e.g. `wg`.
transport: String,
},
/// Through other members.
Relayed {
/// Number of links on the way, at least two.
hops: u8,
/// The member the first of them goes to, as an endpoint id.
via: String,
},
}
/// A name somebody else claimed first.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct HostnameConflictReport {
/// The name that is taken.
pub name: String,
/// The member that holds it, as an endpoint id.
pub holder: String,
}
/// One member of the network, from signed state.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemberReport {
@@ -500,6 +532,15 @@ pub struct MemberReport {
pub failed_dials: u32,
/// The last hostname this agent saw it announce, kept while it is away.
pub hostname: Option<String>,
/// Whether this agent can reach it now, by any path. The one answer
/// every client gives to "is it online": a member reached through
/// others is online, over a longer path.
pub online: bool,
/// How packets reach it.
pub path: DataPathReport,
/// The name it claimed, when somebody else claimed it first. It still
/// works by address; it is not found by that name.
pub hostname_conflict: Option<HostnameConflictReport>,
}
/// One control plane peer.
@@ -516,6 +557,8 @@ pub struct PeerReport {
pub rtt_ms: Option<u64>,
/// The peer offers itself as an exit node in this network.
pub exit_node: bool,
/// How packets reach it.
pub path: DataPathReport,
}
/// The WireGuard overlay of one network.
@@ -779,7 +822,7 @@ pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5);
///
/// Bump it whenever [`Request`], [`Response`] or anything they contain
/// changes shape.
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 17]);
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 18]);
/// Reads one request off an accepted stream, answers it, writes the response.
///
crates/tsunagi-core/src/proto/message.rs
+7 -5
View File
@@ -20,7 +20,7 @@ use crate::error::ProtocolError;
/// The version in the ALPN is the wire-compatibility version of the control
/// protocol. It is independent of the network identity scheme version, so
/// bumping it must not change any existing [`crate::NetworkId`].
pub const ALPN: &[u8] = b"tsunagi/ctrl/4";
pub const ALPN: &[u8] = b"tsunagi/ctrl/5";
/// ALPN of the tsunagi data plane.
///
@@ -30,9 +30,11 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/4";
/// transport underneath can be replaced without touching the control protocol.
///
/// Version 4 adds source, destination, hop limit and stable flow id above
/// transport fragmentation. All members must upgrade together. Persistent
/// identities and network configuration do not change.
pub const DATA_ALPN: &[u8] = b"tsunagi/data/4";
/// transport fragmentation. Version 5 adds the end-to-end protocol to that
/// envelope, so a tunnel no longer depends on the kind of link under it, and
/// a first-claimed time to a signed hostname claim. All members must upgrade
/// together. Persistent identities and network configuration do not change.
pub const DATA_ALPN: &[u8] = b"tsunagi/data/5";
/// Largest plugin protocol identifier accepted when opening a data channel.
pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
@@ -41,7 +43,7 @@ pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
pub const MAX_SIGNATURE_LEN: usize = 64;
/// Control protocol version carried inside the handshake.
pub const PROTOCOL_VERSION: u16 = 4;
pub const PROTOCOL_VERSION: u16 = 5;
/// First message of the handshake, sent by the initiator.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
crates/tsunagi-core/src/state/mod.rs
+173 -8
View File
@@ -150,7 +150,7 @@ pub fn derived_ipv4_range(network: NetworkId) -> Ipv4Range {
}
/// Frozen domain separator for the bytes a record signature covers.
pub const RECORD_DOMAIN: &str = "tsunagi-signed-record-v2";
pub const RECORD_DOMAIN: &str = "tsunagi-signed-record-v3";
/// Longest hostname a record may carry.
///
@@ -207,6 +207,18 @@ pub enum RecordBody {
range: Option<Ipv4Range>,
/// The name this author answers to.
hostname: Option<String>,
/// When this author first claimed that name, in milliseconds since
/// the Unix epoch, kept across every later version that keeps the
/// name. It is what makes a name belong to whoever claimed it first
/// rather than to whoever happens to sort lowest, so a member that
/// arrives later with the same name cannot take it over.
///
/// The author's own word: a member with a wrong clock can claim an
/// earlier time. That costs it nothing it could not do by other
/// means, since anybody who knows the network secret is a member,
/// and it cannot take an address or reach anything it was not
/// already entitled to.
hostname_since: Option<u64>,
},
/// This author gives up everything it claimed.
///
@@ -243,6 +255,14 @@ impl RecordBody {
}
}
/// When the author first claimed the hostname, if it has one.
pub fn hostname_since(&self) -> Option<u64> {
match self {
RecordBody::Claim { hostname_since, .. } => *hostname_since,
RecordBody::Release => None,
}
}
fn canonical(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(96);
match self {
@@ -250,6 +270,7 @@ impl RecordBody {
address,
range,
hostname,
hostname_since,
} => {
push_lp(&mut out, b"claim");
push_opt(
@@ -270,6 +291,13 @@ impl RecordBody {
.as_deref(),
);
push_opt(&mut out, hostname.as_deref().map(str::as_bytes));
push_opt(
&mut out,
hostname_since
.map(|since| since.to_be_bytes())
.as_ref()
.map(|bytes| &bytes[..]),
);
}
RecordBody::Release => {
push_lp(&mut out, b"release");
@@ -391,8 +419,12 @@ impl SignedRecord {
address,
range,
hostname,
hostname_since,
} = &self.body
{
if hostname.is_none() && hostname_since.is_some() {
return Err(StateError::Malformed("a time was claimed for no hostname"));
}
// Bounds before anything is believed, because all of this came
// off the network.
if let Some(range) = range {
@@ -585,25 +617,33 @@ impl StateSet {
/// Who currently holds each claimed hostname.
///
/// The same rule as addresses, for the same reason: a name is owned, two
/// members may claim one, and every replica has to reach the same answer
/// about who has it without asking anybody.
/// A name belongs to the member that claimed it first, and the others
/// that claim it keep working by address but are not found by that name.
/// "First" is the time each author signed into its own claim, with the
/// lower id deciding a tie and a claim with no time counting as last, so
/// every replica reaches the same answer about who has it without
/// asking anybody. It is a pure function of the records: nothing depends
/// on the order they arrived in, so there is no race to lose.
pub fn hostname_holders(&self) -> HashMap<String, EndpointId> {
let mut holders: HashMap<String, EndpointId> = HashMap::new();
let mut holders: HashMap<String, (u64, EndpointId)> = HashMap::new();
for (author, record) in &self.records {
let Some(hostname) = record.body.hostname() else {
continue;
};
let claim = (record.body.hostname_since().unwrap_or(u64::MAX), *author);
holders
.entry(hostname.to_string())
.and_modify(|held| {
if author.as_bytes() < held.as_bytes() {
*held = *author;
if (claim.0, claim.1.as_bytes()) < (held.0, held.1.as_bytes()) {
*held = claim;
}
})
.or_insert(*author);
.or_insert(claim);
}
holders
.into_iter()
.map(|(hostname, (_, author))| (hostname, author))
.collect()
}
/// The hostname an author holds, if it holds one uncontested.
@@ -612,6 +652,19 @@ impl StateSet {
(self.hostname_holders().get(hostname) == Some(author)).then_some(hostname)
}
/// The name an author claimed and the member that holds it instead, when
/// somebody who claimed it earlier does.
pub fn hostname_taken(&self, author: &EndpointId) -> Option<(&str, EndpointId)> {
let hostname = self.records.get(author)?.body.hostname()?;
let holder = *self.hostname_holders().get(hostname)?;
(holder != *author).then_some((hostname, holder))
}
/// When an author first claimed the name it holds now, if it has one.
pub fn hostname_since_of(&self, author: &EndpointId) -> Option<u64> {
self.records.get(author)?.body.hostname_since()
}
/// The address an author holds, if it holds one uncontested.
pub fn address_of(&self, author: &EndpointId) -> Option<Ipv4Addr> {
let address = self.records.get(author)?.body.claimed_address()?;
@@ -661,6 +714,7 @@ mod tests {
address: Some(address.parse().unwrap()),
range: Some(range()),
hostname: None,
hostname_since: None,
}
}
@@ -720,6 +774,7 @@ mod tests {
address: Some("10.99.0.1".parse().unwrap()),
range: Some(range()),
hostname: None,
hostname_since: None,
},
);
assert!(matches!(outside.verify(id), Err(StateError::Malformed(_))));
@@ -735,6 +790,7 @@ mod tests {
prefix_len: 31,
}),
hostname: None,
hostname_since: None,
},
);
assert!(matches!(no_hosts.verify(id), Err(StateError::Malformed(_))));
@@ -776,6 +832,7 @@ mod tests {
address: None,
range: None,
hostname: Some(bad.to_string()),
hostname_since: None,
},
);
assert!(
@@ -797,6 +854,7 @@ mod tests {
address: Some("10.13.37.5".parse().unwrap()),
range: None,
hostname: None,
hostname_since: None,
},
);
assert!(matches!(record.verify(id), Err(StateError::Malformed(_))));
@@ -818,6 +876,7 @@ mod tests {
address: None,
range: None,
hostname: Some("music".into()),
hostname_since: None,
},
)
};
@@ -860,6 +919,7 @@ mod tests {
address: None,
range: None,
hostname: Some(name.to_string()),
hostname_since: None,
},
)
};
@@ -879,6 +939,109 @@ mod tests {
assert_eq!(set.hostname_of(&secret.public()), Some("new"));
}
fn named_at(
secret: &SecretKey,
id: NetworkId,
version: u64,
name: &str,
since: u64,
) -> SignedRecord {
SignedRecord::sign(
secret,
id,
version,
RecordBody::Claim {
address: None,
range: None,
hostname: Some(name.into()),
hostname_since: Some(since),
},
)
}
#[test]
fn a_name_belongs_to_whoever_claimed_it_first_whatever_their_ids_or_arrival_order() {
let id = network("naming");
let mut keys = [SecretKey::generate(), SecretKey::generate()];
keys.sort_by_key(|key| *key.public().as_bytes());
let [low, high] = keys;
// The member with the higher id claimed first, so it holds the name,
// and the lower id does not take it back by sorting first.
let early = named_at(&high, id, 1, "music", 1_000);
let late = named_at(&low, id, 1, "music", 2_000);
for order in [[&early, &late], [&late, &early]] {
let mut set = StateSet::new();
for record in order {
set.merge(id, (*record).clone()).unwrap();
}
assert_eq!(set.hostname_of(&high.public()), Some("music"));
assert_eq!(set.hostname_of(&low.public()), None);
assert_eq!(
set.hostname_taken(&low.public()),
Some(("music", high.public()))
);
assert_eq!(set.hostname_taken(&high.public()), None);
}
// At the same instant the lower id decides, the same way everywhere.
let mut set = StateSet::new();
set.merge(id, named_at(&high, id, 1, "music", 5)).unwrap();
set.merge(id, named_at(&low, id, 1, "music", 5)).unwrap();
assert_eq!(set.hostname_of(&low.public()), Some("music"));
assert_eq!(set.hostname_of(&high.public()), None);
}
#[test]
fn a_name_that_loses_is_found_again_once_the_holder_lets_it_go() {
let id = network("naming");
let first = SecretKey::generate();
let second = SecretKey::generate();
let mut set = StateSet::new();
set.merge(id, named_at(&first, id, 1, "music", 1)).unwrap();
set.merge(id, named_at(&second, id, 1, "music", 2)).unwrap();
assert_eq!(set.hostname_of(&second.public()), None);
// Renaming starts the clock afresh, so it cannot be used to jump the
// queue for a name somebody else already holds.
set.merge(id, named_at(&first, id, 2, "other", 3)).unwrap();
assert_eq!(set.hostname_of(&second.public()), Some("music"));
set.merge(id, named_at(&first, id, 3, "music", 4)).unwrap();
assert_eq!(set.hostname_of(&second.public()), Some("music"));
assert_eq!(set.hostname_of(&first.public()), None);
// And a release hands it on.
set.merge(id, SignedRecord::sign(&second, id, 2, RecordBody::Release))
.unwrap();
assert_eq!(set.hostname_of(&first.public()), Some("music"));
}
#[test]
fn a_time_with_no_name_is_not_a_valid_claim() {
let id = network("naming");
let secret = SecretKey::generate();
let record = SignedRecord::sign(
&secret,
id,
1,
RecordBody::Claim {
address: None,
range: None,
hostname: None,
hostname_since: Some(1),
},
);
assert!(matches!(record.verify(id), Err(StateError::Malformed(_))));
// The time is signed: moving it breaks the signature, so nobody can
// improve another member's place in the queue.
let mut forged = named_at(&secret, id, 1, "music", 9_000);
if let RecordBody::Claim { hostname_since, .. } = &mut forged.body {
*hostname_since = Some(1);
}
assert_eq!(forged.verify(id).unwrap_err(), StateError::BadSignature);
}
#[test]
fn a_release_gives_up_the_name_as_well_as_the_address() {
let secret = SecretKey::generate();
@@ -894,6 +1057,7 @@ mod tests {
address: Some("10.13.37.5".parse().unwrap()),
range: Some(range()),
hostname: Some("music".into()),
hostname_since: None,
},
),
)
@@ -1057,6 +1221,7 @@ mod tests {
address: Some("10.99.0.7".parse().unwrap()),
range: Some(custom),
hostname: None,
hostname_since: None,
},
),
)
crates/tsunagi-core/src/storage/state.rs
+18 -1
View File
@@ -21,7 +21,7 @@ use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret};
use crate::state::{RecordBody, SignedRecord};
/// Schema version written by this build.
pub const SCHEMA_VERSION: i64 = 6;
pub const SCHEMA_VERSION: i64 = 7;
/// Key of the stored hostname setting.
const SETTING_HOSTNAME: &str = "hostname";
@@ -221,6 +221,23 @@ impl StateStore {
)
.map_err(|err| self.corrupt(format!("cannot migrate schema to 6: {err}")))?;
}
// Migration 6 -> 7: a claim gained the time its hostname was first
// claimed, which changed the signing domain. Records signed before
// no longer verify, so they are discarded rather than kept as rows
// every read has to reject; each member re-publishes its own claim on
// its next run and the others arrive again from their authors. The
// version counter is kept, so nothing is ever published twice at one
// version.
if found < 7 {
self.conn
.execute_batch(
"BEGIN;
DELETE FROM signed_records;
PRAGMA user_version = 7;
COMMIT;",
)
.map_err(|err| self.corrupt(format!("cannot migrate schema to 7: {err}")))?;
}
Ok(())
}