mosaik 0.3.17

//! Bonds — persistent peer-to-peer connections within a consensus group.
//!
//! A **bond** is a long-lived, bidirectional connection between the local node
//! and a single remote peer that belongs to the same group. Every node in a
//! group maintains one bond with every other group member, forming an
//! all-to-all mesh of connections over which group-internal traffic flows.
//!
//! # Relationship to the Groups subsystem
//!
//! Groups (see [`super`]) provide fault-tolerant coordination among a set of
//! trusted peers. Bonds are the transport layer that makes this possible:
//! they carry Raft consensus messages, heartbeat liveness probes,
//! discovery-entry  updates, and notifications about newly joined peers.
//! Without active bonds a group cannot reach consensus or detect failures.
//!
//! # Lifecycle
//!
//! 1. **Establishment** — When a new peer is discovered in the group's
//!    discovery catalog, [`Bond::create`] opens a QUIC connection and performs
//!    a mutual handshake. Both sides prove knowledge of the shared group secret
//!    by exchanging proofs derived from the TLS session secrets and the group
//!    key. Incoming connections follow the mirror path through
//!    [`Bond::accept`].
//!
//! 2. **Steady state** — Once established, a [`BondWorker`](worker::BondWorker)
//!    runs an event loop that multiplexes
//!    [`BondMessage`](protocol::BondMessage) traffic (Raft messages, peer-entry
//!    updates, bond-formed notifications) with periodic
//!    [`Heartbeat`](heartbeat::Heartbeat) pings. The associated [`Bond`] handle
//!    exposes a command channel so the group worker can send messages or close
//!    the connection.
//!
//! 3. **Failure & reconnection** — If heartbeats are missed beyond a configured
//!    threshold the bond is considered failed and torn down. The group worker
//!    might attempt to re-establish the bond if the peer is still present in
//!    the discovery catalog, which would involve going through the
//!    establishment phase again.
//!
//! # Key types
//!
//! | Type | Role |
//! |------|------|
//! | [`Bond`] | Clonable handle used by the group to send commands to a bond |
//! | [`Bonds`] | Watchable, ordered collection of all active bonds in a group |
//! | [`BondId`] | Deterministic identifier derived from the shared session random, identical on both sides |
//! | [`BondEvent`] | Events emitted by a bond back to the group worker (connected, raft message, terminated) |

use {
	crate::{
		Digest,
		PeerId,
		discovery::SignedPeerEntry,
		groups::{
			Error,
			Groups,
			StateMachine,
			bond::worker::WorkerCommand,
			error::{InvalidProof, NotAllowed},
			raft,
			state::WorkerState,
		},
		network::{
			CloseReason,
			UnexpectedClose,
			UnknownPeer,
			link::{Link, RecvError},
		},
		primitives::{EncodeError, Short, ShortFmtExt, encoding::try_serialize},
	},
	bytes::Bytes,
	core::fmt,
	iroh::endpoint::ApplicationClose,
	protocol::{BondMessage, HandshakeEnd},
	std::sync::Arc,
	tokio::{
		sync::{
			mpsc::{UnboundedReceiver, UnboundedSender},
			watch,
		},
		time::timeout,
	},
	worker::BondWorker,
};

mod heartbeat;
mod protocol;
mod worker;

pub(super) use protocol::{Acceptor, HandshakeStart};

/// A unique identifier for a bond connection between the local node and a
/// remote peer in the group. This value is derived from the shared random
/// between the two peers during bond establishment and is used to correlate the
/// bond connection with the corresponding peer entry in the discovery catalog
/// and to identify the bond in logs and metrics.
///
/// The bond id should be identical on both sides of the bond connection.
pub type BondId = Digest;

pub enum BondEvent<M: StateMachine> {
	/// The bond connection to the remote peer has been successfully established.
	Connected,

	/// Received a raft-protocol message from the remote peer.
	Raft(raft::Message<M>),

	/// The bond connection to the remote peer has been closed. The provided
	/// reason hints at why the bond was closed.
	Terminated(ApplicationClose),
}

pub type BondEvents<M> = UnboundedReceiver<BondEvent<M>>;

/// Handle to a bond with another peer in a group.
pub struct Bond<M: StateMachine> {
	/// A unique identifier for this bond connection.
	///
	/// This value is identical on both sides of the bond connection.
	id: BondId,

	/// Channel for sending commands to the bond worker loop.
	commands_tx: UnboundedSender<WorkerCommand>,

	/// Watch channel for observing when the bond worker has terminated and the
	/// reason for termination.
	terminated_rx: watch::Receiver<Option<ApplicationClose>>,

	/// Watch channel for observing updates to the remote peer's discovery entry.
	peer: watch::Receiver<SignedPeerEntry>,

	#[doc(hidden)]
	_p: core::marker::PhantomData<M>,
}

impl<M: StateMachine> Clone for Bond<M> {
	fn clone(&self) -> Self {
		Self {
			id: self.id,
			commands_tx: self.commands_tx.clone(),
			terminated_rx: self.terminated_rx.clone(),
			peer: self.peer.clone(),
			_p: core::marker::PhantomData,
		}
	}
}

impl<M: StateMachine> fmt::Debug for Bond<M> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_struct("Bond")
			.field("id", &self.id)
			.field("peer_id", self.peer.borrow().id())
			.finish_non_exhaustive()
	}
}

impl<M: StateMachine> fmt::Display for Bond<M> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(
			f,
			"Bond(id={}, peer={})",
			Short(self.id),
			Short(self.peer.borrow().id()),
		)
	}
}

/// Public API
impl<M: StateMachine> Bond<M> {
	/// Closes the bond connection to the remote peer with the provided
	/// application-level close reason and waits for the bond worker to terminate.
	pub async fn close(self, reason: impl CloseReason) {
		let _ = self.commands_tx.send(WorkerCommand::Close(reason.into()));
		self.terminated().await;
	}

	/// Returns the most recent peer entry for the remote peer in the bond.
	pub fn peer(&self) -> SignedPeerEntry {
		self.peer.borrow().clone()
	}

	/// Returns the unique identifier for this bond connection.
	/// This value is derived from the shared random between the two peers during
	/// bond establishment and is used to correlate the bond connection with the
	/// corresponding peer entry in the discovery catalog and to identify the bond
	/// in logs and metrics.
	pub const fn id(&self) -> BondId {
		self.id
	}

	/// Returns `true` if the bond connection has been terminated.
	pub fn is_terminated(&self) -> bool {
		self.terminated_rx.borrow().is_some()
	}

	/// Returns a future that resolves when the bond connection has been
	/// terminated and provides the reason for termination if available.
	pub fn terminated(
		&self,
	) -> impl Future<Output = ApplicationClose> + Send + Sync + 'static {
		let mut rx = self.terminated_rx.clone();
		async move {
			rx.wait_for(|v| v.is_some()).await.map_or_else(
				|_| UnexpectedClose.into(),
				|reason| reason.clone().unwrap_or_else(|| UnexpectedClose.into()),
			)
		}
	}
}

/// Internal API for creating and accepting bonds
impl<M: StateMachine> Bond<M> {
	/// Initiates the process of creating a new bond connection to a remote
	/// peer in the group.
	///
	/// This happens in response to discovering a new peer in the group via
	/// the discovery catalog. This method is called only for peers that are
	/// already known in the discovery catalog.
	pub(super) async fn create(
		group: Arc<WorkerState<M>>,
		peer: SignedPeerEntry,
	) -> Result<(Self, BondEvents<M>), Error> {
		if group.config.authorize_peer(&peer).is_err() {
			tracing::debug!(
				network = %group.network_id().short(),
				peer = %peer.id().short(),
				group = %group.group_id().short(),
				"bonding failed: unauthorized",
			);

			return Err(Error::Unauthorized);
		}

		// attempt to establish a new wire link to the remote peer
		let mut link = group
			.local
			.connect_with_cancel::<Groups>(
				peer.address().clone(),
				group.cancel.child_token(),
			)
			.await
			.map_err(|e| Error::Link(e.into()))?;

		// prepare a handshake message with our proof of knowledge of the group
		// secret and send it to the remote peer over the newly established link
		link
			.send(&HandshakeStart {
				network_id: *group.network_id(),
				group_id: *group.group_id(),
				proof: group.generate_key_proof(&link),
				bonds: group.bonds.iter().map(|b| b.peer()).collect(),
			})
			.await
			.map_err(|e| Error::Link(e.into()))?;

		// After sending our handshake with a proof of knowledge of the group
		// secret, wait for the accepting peer to respond with its own proof of
		// knowledge of the group secret.
		let Ok(recv_result) = timeout(
			group.global_config.handshake_timeout,
			link.recv::<HandshakeEnd>(),
		)
		.await
		else {
			tracing::debug!(
				network = %group.network_id().short(),
				peer = %peer.id().short(),
				group = %group.group_id().short(),
				"handshake timeout waiting for bond confirmation",
			);

			return Err(Error::Link(RecvError::Cancelled.into()));
		};

		let confirm = match recv_result {
			Ok(resp) => resp,
			Err(e) => match e.close_reason() {
				// the remote peer closed the link during handshake because the local
				// node is not known in its discovery catalog. Trigger full discovery
				// catalog sync.
				Some(reason) if reason == UnknownPeer => {
					// trigger full catalog sync and retry bonding
					if let Err(e) =
						group.discovery.sync_with(peer.address().clone()).await
					{
						link.close(UnknownPeer).await.ok();
						return Err(Error::Discovery(e));
					}

					// retry creating the bond after syncing the catalog
					return Box::pin(Self::create(group, peer)).await;
				}
				// The remote peer rejected our authentication proof or missing tickets.
				// This is an unrecoverable error.
				Some(reason) if reason == InvalidProof || reason == NotAllowed => {
					tracing::warn!(
						network = %group.network_id().short(),
						peer = %peer.id().short(),
						group = %group.group_id().short(),
						"remote peer rejected: unauthorized",
					);

					link
						.close(reason.clone())
						.await
						.map_err(|e| Error::Link(e.into()))?;

					return Err(Error::InvalidGroupKeyProof);
				}
				// Bonding failed for some other reason.
				_ => return Err(Error::Link(e.into())),
			},
		};

		// validate the accepting peer's proof of knowledge of the group secret
		if !group.validate_key_proof(&link, confirm.proof) {
			tracing::warn!(
				network = %group.network_id().short(),
				peer = %peer.id().short(),
				group = %group.group_id().short(),
				"remote peer provided invalid group secret proof",
			);

			link
				.close(InvalidProof)
				.await
				.map_err(|e| Error::Link(e.into()))?;

			return Err(Error::InvalidGroupKeyProof);
		}

		// attempt to connect to all existing peers in the group known to the
		// accepting node
		for peer in confirm.bonds {
			group.bond_with(peer);
		}

		Ok(BondWorker::spawn(group, peer, link))
	}

	/// Accepts an incoming bond connection for this group.
	///
	/// This is called by the group's protocol handler when a new connection
	/// is established  in [`bond::Acceptor::accept`].
	///
	/// By the time this method is called:
	/// - The network id has already been verified to match the local node's
	///   network id.
	/// - The group id has already been verified to match this group's id.
	/// - The presence of the remote peer in the local discovery catalog is
	///   verified.
	/// - The authentication proof has not been verified yet.
	pub(super) async fn accept(
		group: Arc<WorkerState<M>>,
		link: Link<Groups>,
		peer: SignedPeerEntry,
		handshake: HandshakeStart,
	) -> Result<(Self, BondEvents<M>), Error> {
		let mut link = link;

		if group.config.authorize_peer(&peer).is_err() {
			tracing::debug!(
				network = %group.network_id().short(),
				peer = %peer.id().short(),
				group = %group.group_id().short(),
				"rejecting bond: unauthorized",
			);

			link
				.close(NotAllowed)
				.await
				.map_err(|e| Error::Link(e.into()))?;

			return Err(Error::Unauthorized);
		}

		// verify the remote peer's proof of knowledge of the group secret
		if !group.validate_key_proof(&link, handshake.proof) {
			tracing::warn!(
				network = %group.network_id().short(),
				peer = %peer.id().short(),
				group = %group.group_id().short(),
				"remote peer provided invalid group secret proof",
			);

			link
				.close(InvalidProof)
				.await
				.map_err(|e| Error::Link(e.into()))?;

			return Err(Error::InvalidGroupKeyProof);
		}

		// After verifying the remote peer's proof of knowledge of the group secret,
		// respond with our own proof of knowledge of the group secret.
		let proof = group.generate_key_proof(&link);
		let existing = group.bonds.iter().map(|b| b.peer()).collect();
		let resp = HandshakeEnd {
			proof,
			bonds: existing,
		};
		link.send(&resp).await.map_err(|e| Error::Link(e.into()))?;

		// attempt to connect to all existing peers in the group known to the
		// initiating node
		for peer in handshake.bonds {
			group.bond_with(peer);
		}

		Ok(BondWorker::spawn(group, peer, link))
	}
}

/// Internal Bond API used by the groups module.
impl<M: StateMachine> Bond<M> {
	/// Sends a wire message over the bond connection to the remote peer.
	#[allow(clippy::needless_pass_by_value)]
	pub(super) fn send_message(
		&self,
		message: BondMessage<M>,
	) -> Result<(), EncodeError> {
		let serialized = try_serialize(&message)?;
		unsafe { self.send_raw_message(serialized) };
		Ok(())
	}

	/// Sends a raw pre-encoded message over the bond connection to the remote
	/// peer.
	///
	/// SAFETY:
	/// This method is private and is used only by specialized internal methods in
	/// the `Bonds` type that serialize a `BondMessage` once and then send the
	/// same encoded bytes buffer to all bonded peers. External APIs cannot send
	/// arbitrary bytes buffers and can only send structured `BondMessage` values
	/// via the `send_message` method, which ensures that only valid `BondMessage`
	/// values are sent.
	unsafe fn send_raw_message(&self, message: Bytes) {
		let _ = self
			.commands_tx
			.send(WorkerCommand::SendRawMessage(message));
	}
}

/// A watchable collection of currently active bonds in a group.
///
/// This type allows observing changes to the set of active bonds.
///
/// This type is cheap to clone and can be freely passed around, all clones of
/// this type will always reflect the same up to date set of active bonds in the
/// group.
#[derive(Debug)]
pub struct Bonds<M: StateMachine>(
	pub(super) watch::Sender<im::OrdMap<PeerId, Bond<M>>>,
);

impl<M: StateMachine> Clone for Bonds<M> {
	fn clone(&self) -> Self {
		Self(self.0.clone())
	}
}

/// Public API
impl<M: StateMachine> Bonds<M> {
	/// Returns the number of active bonds in the group.
	pub fn len(&self) -> usize {
		self.0.borrow().len()
	}

	/// Returns `true` if there are no active bonds in the group.
	pub fn is_empty(&self) -> bool {
		self.0.borrow().is_empty()
	}

	/// Returns `true` if there is an active bond to the specified peer.
	pub fn contains_peer(&self, peer_id: &PeerId) -> bool {
		self.0.borrow().contains_key(peer_id)
	}

	/// Returns an iterator over all active bonds in the group ordered by their
	/// peer ids at the time of calling this method.
	pub fn iter(&self) -> impl Iterator<Item = Bond<M>> {
		let bonds = self.0.borrow().clone();
		bonds.into_iter().map(|(_, bond)| bond)
	}

	/// Returns a future that resolves when there is a change to the active
	/// bonds in the group.
	pub async fn changed(&self) {
		let _ = self.0.subscribe().changed().await;
	}

	/// Returns the bond to the specified peer if it exists.
	pub fn get(&self, peer_id: &PeerId) -> Option<Bond<M>> {
		self.0.borrow().get(peer_id).cloned()
	}
}

/// Internal API
impl<M: StateMachine> Default for Bonds<M> {
	fn default() -> Self {
		Self(watch::Sender::new(im::OrdMap::new()))
	}
}

/// Internal API
impl<M: StateMachine> Bonds<M> {
	pub(super) fn update_with(
		&self,
		f: impl FnOnce(&mut im::OrdMap<PeerId, Bond<M>>),
	) {
		self.0.send_if_modified(|active| {
			let before = active.len();
			f(active);
			active.len() != before
		});
	}

	/// Notifies all active bonds that the local peer's discovery entry has
	/// been updated.
	pub(super) fn notify_local_info_update(&self, entry: &SignedPeerEntry) {
		self
			.broadcast(&BondMessage::PeerEntryUpdate(Box::new(entry.clone())), &[])
			.expect("infallible serialization");
	}

	/// Notifies all active bonds that a new bond has been formed with the
	/// specified peer and sends its latest known discovery entry.
	pub(super) fn notify_bond_formed(&self, with: &SignedPeerEntry) {
		self
			.broadcast(&BondMessage::BondFormed(Box::new(with.clone())), &[
				*with.id()
			])
			.expect("infallible serialization");
	}

	/// Notifies all active bonds that the bond with the specified peer has
	/// been terminated voluntarily by the remote peer, so they don't need to wait
	/// for heartbeat timeouts to detect that the peer has left and can adjust
	/// their quorums.
	pub(super) fn notify_departure(&self) {
		self
			.broadcast(&BondMessage::Departure, &[])
			.expect("infallible serialization");
	}

	/// Sends a raft protocol message to all bonded peers.
	///
	/// Fails if the message serialization fail.
	pub(super) fn broadcast_raft(
		&self,
		message: raft::Message<M>,
	) -> Result<Vec<PeerId>, EncodeError> {
		let message = BondMessage::Raft(message);
		self.broadcast(&message, &[])
	}

	/// Sends a raft protocol message to the specified bonded peer.
	///
	/// Fails if the message serialization fail.
	pub(super) fn send_raft_to(
		&self,
		message: raft::Message<M>,
		to: PeerId,
	) -> Result<(), EncodeError> {
		let Some(bond) = self.get(&to) else {
			tracing::warn!(
				peer = %Short(to),
				"attempted to send raft message to non-bonded peer",
			);
			return Ok(());
		};

		bond.send_message(BondMessage::Raft(message))
	}

	/// Broadcast a message to all connected peers in the group. The `except`
	/// parameter specifies a list of peer ids to exclude from the broadcast.
	/// Returns the list of peer ids to which the message was sent.
	fn broadcast(
		&self,
		message: &BondMessage<M>,
		except: &[PeerId],
	) -> Result<Vec<PeerId>, EncodeError> {
		// serialize once and reuse a pointer to the same encoded message bytes
		// buffer for all bonds
		let encoded = try_serialize(message)?;

		let mut sent_to = Vec::new();
		for bond in self.iter() {
			if !except.contains(bond.peer().id()) {
				unsafe { bond.send_raw_message(encoded.clone()) };
				sent_to.push(*bond.peer().id());
			}
		}
		Ok(sent_to)
	}
}