diff options
Diffstat (limited to 'src/transport/mod.rs')
| -rw-r--r-- | src/transport/mod.rs | 647 |
1 files changed, 534 insertions, 113 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 35e3499..25ad56d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -2,62 +2,102 @@ use anyhow::Result; use async_trait::async_trait; use automerge::ChangeHash; use automerge::sync::Message; +use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::boxed::Box; use std::future::Future; use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; -pub use crate::document::replica::Epoch; +pub(crate) use crate::document::replica::Epoch; -pub mod firewall; -pub mod mycelium; +// Used by the transports that don't authenticate their peers themselves (`tor`). +#[cfg_attr(not(feature = "tor"), allow(dead_code))] +pub(crate) mod challenge; +pub(crate) mod firewall; +pub(crate) mod identity; +#[cfg(feature = "libp2p")] +pub(crate) mod libp2p; +pub(crate) mod mycelium; +pub(crate) mod paths; #[cfg(test)] pub(crate) mod sim; -pub mod traffic; +pub(crate) mod stats; +#[cfg(feature = "tor")] +pub(crate) mod tor; +pub(crate) mod traffic; +pub(crate) mod wire; /// Random identifier of one run of a node (from start to exit). -pub type RunId = [u8; 16]; +pub(crate) type RunId = [u8; 16]; -/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on -/// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see -/// the size test in `mycelium`). There are rarely more heads than nodes that wrote -/// concurrently since the last merging write. -pub const MAX_LISTED_HEADS: usize = 32; +/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes +/// on the wire, so 32 of them keep a keepalive within one small packet (a Mycelium +/// chunk: 1300 bytes; see the size test in `mycelium`). There are rarely more heads +/// than nodes that wrote concurrently since the last merging write. +pub(crate) const MAX_LISTED_HEADS: usize = 32; /// A document's heads, compactly: the heads themselves while there are few, otherwise a -/// digest of them. Two summaries are equal exactly when the heads are, whatever form -/// each took (nodes with different limits still agree). +/// digest of them with some of them. Two summaries are equal exactly when the heads are, +/// whatever form each took (nodes with different limits still agree). #[derive(Debug, Clone)] -pub enum HeadsSummary { +pub(crate) enum HeadsSummary { /// Sorted. Heads(Vec<ChangeHash>), - /// SHA-256 over the sorted heads. - Digest([u8; 32]), + Digest { + /// SHA-256 over the sorted heads. + digest: [u8; 32], + /// `MAX_LISTED_HEADS` of the heads, another window of them each time: a receiver + /// pulls those it lacks, and over a few keepalives every head is listed, instead + /// of starting the sync over to find out what it lacks. + sample: Vec<ChangeHash>, + }, } +/// Where the next sample starts (`HeadsSummary::of`). +static SAMPLE_OFFSET: AtomicUsize = AtomicUsize::new(0); + impl HeadsSummary { - pub fn of(mut heads: Vec<ChangeHash>) -> Self { + pub(crate) fn of(mut heads: Vec<ChangeHash>) -> Self { heads.sort_unstable(); if heads.len() <= MAX_LISTED_HEADS { - Self::Heads(heads) - } else { - Self::Digest(digest_of(&heads)) + return Self::Heads(heads); + } + let start = SAMPLE_OFFSET.fetch_add(MAX_LISTED_HEADS, Ordering::Relaxed) % heads.len(); + let sample = heads + .iter() + .cycle() + .skip(start) + .take(MAX_LISTED_HEADS) + .copied() + .collect(); + Self::Digest { + digest: digest_of(&heads), + sample, } } /// The digest of the heads, computed if they are listed. - pub fn digest(&self) -> [u8; 32] { + pub(crate) fn digest(&self) -> [u8; 32] { match self { Self::Heads(heads) => digest_of(heads), - Self::Digest(digest) => *digest, + Self::Digest { digest, .. } => *digest, } } - /// The heads, if listed. - pub fn heads(&self) -> Option<&[ChangeHash]> { + /// The heads, if all are listed. + pub(crate) fn heads(&self) -> Option<&[ChangeHash]> { match self { Self::Heads(heads) => Some(heads), - Self::Digest(_) => None, + Self::Digest { .. } => None, + } + } + + /// The heads listed: all of them, or a sample. + pub(crate) fn listed(&self) -> &[ChangeHash] { + match self { + Self::Heads(heads) => heads, + Self::Digest { sample, .. } => sample, } } } @@ -82,79 +122,117 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { hasher.finalize().into() } -/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient. -#[derive(Debug, Clone)] -pub struct Keepalive { +/// STATUS and STATUS_REPLY. Addressed to one peer: `syncs_with_you` is about the recipient. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct Status { /// Only comparable between nodes on the same epoch. - pub heads: HeadsSummary, - pub epoch: Epoch, + #[serde(with = "crate::transport::wire::codec::heads")] + pub(crate) heads: HeadsSummary, + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. - pub epoch_seq: u32, + pub(crate) epoch_seq: u32, /// The sender is ready (same meaning as `SyncEnvelope::ready`). - pub ready: bool, + pub(crate) ready: bool, /// The sender would sync with the recipient: counts it as an active member on its /// own epoch. - pub syncs_with_you: bool, + pub(crate) syncs_with_you: bool, } -#[derive(Debug)] -pub enum ParsedMessage { +/// Cheap to clone: large parts are shared (`bytes::Bytes`). +#[derive(Debug, Clone)] +pub(crate) enum ParsedMessage { Hello(HelloRequest), HelloReply(HelloReply), Sync(SyncEnvelope), - Ping(Keepalive), - Pong(Keepalive), + Status(Status), + StatusReply(Status), Blob(BlobMessage), + /// Asks the recipient who it is, saying who the sender is (`transport::identity`). + Who(IdentityProof), + /// Answers a WHO. + Me(IdentityProof), + /// Asks for a PONG with this nonce, over the same path (`transport::paths`). + Ping(u64), + /// Answers a PING. + Pong(u64), +} + +/// A node's proof that it holds the key of its node ID (`identity`), sending from its +/// transport address to the recipient's: a signature over both addresses +/// (`transport::identity::signed`). WHO and ME end at the identity layer; the rest of +/// the node never sees them. +#[derive(Clone, Serialize, Deserialize)] +pub(crate) struct IdentityProof { + /// The sender's public node key. + #[serde(with = "crate::transport::wire::codec::bytes32")] + pub(crate) key: [u8; 32], + #[serde(with = "crate::transport::wire::codec::bytes64")] + pub(crate) signature: [u8; 64], +} + +impl std::fmt::Debug for IdentityProof { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("IdentityProof") + .field("key", &Hex(&self.key)) + .finish_non_exhaustive() + } } /// Content-addressed bytes that any node holding them can serve, in chunks /// (`swarm::blob`). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum BlobId { +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub(crate) enum BlobId { /// An Automerge change, by its hash: e.g. an epoch's snapshot change. - Change(ChangeHash), + Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash), /// Bytes by their BLAKE3 hash: e.g. the content of a remake. - Content([u8; 32]), + Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), + /// The changes of a `DeltaSpec`, by its ID (`DeltaSpec::id`). The requests carry + /// the spec, so any node holding its `to` can serve it. + Delta(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), } impl std::fmt::Display for BlobId { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { BlobId::Change(hash) => write!(f, "change:{hash}"), - BlobId::Content(hash) => { - write!(f, "content:")?; - hash.iter().try_for_each(|byte| write!(f, "{byte:02x}")) - } + BlobId::Content(hash) => write!(f, "content:{}", Hex(hash)), + BlobId::Delta(id) => write!(f, "delta:{}", Hex(id)), } } } /// A blob request or one of its answers. Each is a message of its own, so no message /// is larger than one chunk. -#[derive(Debug, Clone)] -pub struct BlobMessage { +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct BlobMessage { /// The sender's epoch (every message carries it). - pub epoch: Epoch, + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. - pub epoch_seq: u32, - pub id: BlobId, - pub body: BlobBody, + pub(crate) epoch_seq: u32, + pub(crate) id: BlobId, + pub(crate) body: BlobBody, } -#[derive(Clone)] -pub enum BlobBody { - /// Asks for these chunks; for the manifest if empty. +#[derive(Clone, Serialize, Deserialize)] +pub(crate) enum BlobBody { + /// Asks for these chunks; for the manifest if empty. `delta`: the spec of a + /// `BlobId::Delta`, which the ID alone doesn't say. Request { chunks: Vec<u32>, + delta: Option<DeltaSpec>, }, /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. Manifest { size: u64, chunk_size: u32, + #[serde(with = "crate::transport::wire::codec::hashes32")] hashes: Vec<[u8; 32]>, }, Chunk { index: u32, + #[serde(with = "crate::transport::wire::codec::shared_bytes")] data: bytes::Bytes, }, /// The sender doesn't have the blob. @@ -165,9 +243,11 @@ pub enum BlobBody { impl std::fmt::Debug for BlobBody { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - BlobBody::Request { chunks } => { - f.debug_struct("Request").field("chunks", chunks).finish() - } + BlobBody::Request { chunks, delta } => f + .debug_struct("Request") + .field("chunks", chunks) + .field("delta", &delta.is_some()) + .finish(), BlobBody::Manifest { size, chunk_size, @@ -190,125 +270,439 @@ impl std::fmt::Debug for BlobBody { /// What a message is, without its contents. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum MessageKind { +pub(crate) enum MessageKind { Hello, HelloReply, Sync, + Status, + StatusReply, + Blob, + Who, + Me, Ping, Pong, - Blob, } impl ParsedMessage { - pub fn kind(&self) -> MessageKind { + pub(crate) fn kind(&self) -> MessageKind { match self { ParsedMessage::Hello(_) => MessageKind::Hello, ParsedMessage::HelloReply(_) => MessageKind::HelloReply, ParsedMessage::Sync(_) => MessageKind::Sync, + ParsedMessage::Status(_) => MessageKind::Status, + ParsedMessage::StatusReply(_) => MessageKind::StatusReply, + ParsedMessage::Blob(_) => MessageKind::Blob, + ParsedMessage::Who(_) => MessageKind::Who, + ParsedMessage::Me(_) => MessageKind::Me, ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong(_) => MessageKind::Pong, - ParsedMessage::Blob(_) => MessageKind::Blob, - } - } - - /// Every message carries the sender's epoch. - pub fn epoch(&self) -> Epoch { - match self { - ParsedMessage::Hello(hello) => hello.epoch, - ParsedMessage::HelloReply(reply) => reply.epoch, - ParsedMessage::Sync(envelope) => envelope.epoch, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, - ParsedMessage::Blob(blob) => blob.epoch, } } - /// `SyncEnvelope::epoch_seq`, as every message carries it. - pub fn epoch_seq(&self) -> u32 { + /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them + /// but WHO, ME, PING and PONG (which the node never sees). + pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> { match self { - ParsedMessage::Hello(hello) => hello.epoch_seq, - ParsedMessage::HelloReply(reply) => reply.epoch_seq, - ParsedMessage::Sync(envelope) => envelope.epoch_seq, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq, - ParsedMessage::Blob(blob) => blob.epoch_seq, + ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)), + ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)), + ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)), + ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => { + Some((status.epoch, status.epoch_seq)) + } + ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)), + ParsedMessage::Who(_) + | ParsedMessage::Me(_) + | ParsedMessage::Ping(_) + | ParsedMessage::Pong(_) => None, } } } -#[derive(Debug, Clone)] -pub struct HelloRequest { +#[derive(Clone, Serialize, Deserialize)] +pub(crate) struct HelloRequest { /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in /// another swarm doesn't answer. - pub swarm_id: Option<String>, - pub epoch: Epoch, + pub(crate) swarm_id: Option<String>, + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. - pub epoch_seq: u32, + pub(crate) epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. - pub auth: Vec<u8>, + #[serde(with = "serde_bytes")] + pub(crate) auth: Vec<u8>, /// The sender is ready (same meaning as `SyncEnvelope::ready`). - pub ready: bool, + pub(crate) ready: bool, /// Random per run of the sender; the reply echoes it. - pub run_id: RunId, + #[serde(with = "crate::transport::wire::codec::run_id")] + pub(crate) run_id: RunId, + /// The handshake's number: the sender numbers the handshakes it starts in a run, + /// with all peers, from 1. A retry carries the same number, so the recipient resets + /// its sync session only for a new handshake, not for every resend or late copy. + pub(crate) handshake: u32, } -#[derive(Debug, Clone)] -pub struct HelloReply { - pub swarm_id: String, - pub peers: Vec<String>, - pub epoch: Epoch, +/// A peer a HELLO_REPLY lists: who it is, and where the sender reaches it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct ListedPeer { + pub(crate) id: String, + /// As `Transport::addresses_of` gives them; hints for the recipient. + pub(crate) addresses: Vec<String>, +} + +/// Where messages to a peer go now (`Transport::destination`). +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Destination { + /// As the identity layer keeps it: `<scheme>:<transport identity>`. + pub(crate) address: String, + /// The peer proved itself there; else it is only said to be there, and sends wait + /// for its proof. + pub(crate) proven: bool, + /// The transport's connection to it, if the transport has connections. + pub(crate) connection: Option<Connection>, + /// Every address it proved itself at, best first, as the paths layer measures them + /// (`transport::paths`). + pub(crate) paths: Vec<stats::PathInfo>, +} + +/// How a transport with connections (libp2p) is connected to a peer. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Connection { + Direct, + /// Only through a relay. + Relayed, +} + +impl Connection { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Relayed => "relayed", + } + } +} + +#[derive(Clone, Serialize, Deserialize)] +pub(crate) struct HelloReply { + pub(crate) swarm_id: String, + pub(crate) peers: Vec<ListedPeer>, + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. - pub epoch_seq: u32, + pub(crate) epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. - pub auth: Vec<u8>, + #[serde(with = "serde_bytes")] + pub(crate) auth: Vec<u8>, /// The sender is ready. - pub ready: bool, + pub(crate) ready: bool, /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the /// recipient (delivered late, or queued while it was down) is ignored. - pub run_id: RunId, + #[serde(with = "crate::transport::wire::codec::run_id")] + pub(crate) run_id: RunId, /// How long before this HELLO the sender had last heard from the recipient, if ever /// (in milliseconds): tells a restarted node how long it was away. - pub away_ms: Option<u64>, + pub(crate) away_ms: Option<u64>, /// The sender's own run (as its HELLOs carry it): a new one means it restarted. - pub sender_run_id: RunId, + #[serde(with = "crate::transport::wire::codec::run_id")] + pub(crate) sender_run_id: RunId, } -#[derive(Debug, Clone)] -pub struct SyncEnvelope { - pub swarm_id: String, +/// Bytes shown as lowercase hex, e.g. run IDs in logs. +pub(crate) struct Hex<'a>(pub(crate) &'a [u8]); + +impl std::fmt::Display for Hex<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.iter().try_for_each(|byte| write!(f, "{byte:02x}")) + } +} + +impl std::fmt::Debug for Hex<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + std::fmt::Display::fmt(self, f) + } +} + +/// Whether a handshake carries a swarm-key proof, not the proof: logged messages don't +/// hand out tags, and stay short. +fn auth_shown(auth: &[u8]) -> &'static str { + if auth.is_empty() { "none" } else { "present" } +} + +impl std::fmt::Debug for HelloRequest { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HelloRequest") + .field("swarm_id", &self.swarm_id) + .field("epoch", &self.epoch) + .field("epoch_seq", &self.epoch_seq) + .field("auth", &auth_shown(&self.auth)) + .field("ready", &self.ready) + .field("run_id", &Hex(&self.run_id)) + .field("handshake", &self.handshake) + .finish() + } +} + +impl std::fmt::Debug for HelloReply { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HelloReply") + .field("swarm_id", &self.swarm_id) + .field("peers", &self.peers.len()) + .field("epoch", &self.epoch) + .field("epoch_seq", &self.epoch_seq) + .field("auth", &auth_shown(&self.auth)) + .field("ready", &self.ready) + .field("run_id", &Hex(&self.run_id)) + .field("away_ms", &self.away_ms) + .field("sender_run_id", &Hex(&self.sender_run_id)) + .finish() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct SyncEnvelope { + pub(crate) swarm_id: String, /// Epoch of the document the message was generated from. - pub epoch: Epoch, + #[serde(with = "crate::transport::wire::codec::epoch")] + pub(crate) epoch: Epoch, /// How many times the sender changed epoch in its current run, when it sent this /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver /// ignores an epoch older than the newest it has seen in the same session. - pub epoch_seq: u32, - pub message: Message, + pub(crate) epoch_seq: u32, + pub(crate) body: SyncBody, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a /// catching-up node that it has everything; an empty document says nothing else. - pub ready: bool, + pub(crate) ready: bool, +} + +/// What a sync envelope carries. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) enum SyncBody { + /// An Automerge sync message. + Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload), + /// Instead of a message too large to send in one piece: fetch these changes as a + /// blob (`BlobId::Delta`), then sync on. + Delta(DeltaSpec), +} + +/// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of +/// a document at `to`. Every node holding `to` produces the same bytes for it +/// (`swarm::blob::delta_bytes`). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct DeltaSpec { + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) from: Vec<ChangeHash>, + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) to: Vec<ChangeHash>, +} + +impl DeltaSpec { + pub(crate) fn new(mut from: Vec<ChangeHash>, mut to: Vec<ChangeHash>) -> Self { + from.sort_unstable(); + from.dedup(); + to.sort_unstable(); + to.dedup(); + Self { from, to } + } + + /// Names the spec as a blob: SHA-256 over both lists. + pub(crate) fn id(&self) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(b"continuum/delta/1"); + for heads in [&self.from, &self.to] { + hasher.update((heads.len() as u32).to_be_bytes()); + for head in heads { + hasher.update(head.0); + } + } + hasher.finalize().into() + } +} + +/// An Automerge sync message in Automerge's own encoding. Decoded only where it is +/// applied (`decode`), after the sender proved to be a member and off the consumer: a +/// large message from anyone else costs nothing but its bytes. +#[derive(Clone)] +pub(crate) struct SyncPayload(pub(crate) bytes::Bytes); + +impl SyncPayload { + pub(crate) fn encode(message: Message) -> Self { + Self(message.encode().into()) + } + + pub(crate) fn decode(&self) -> Result<Message> { + Message::decode(&self.0) + .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}")) + } +} + +/// Sizes instead of contents: logged messages stay short. +impl std::fmt::Debug for SyncPayload { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "SyncPayload({} bytes)", self.0.len()) + } } /// A message serialized for the wire. Cloning is cheap (shared buffer). #[derive(Clone, Debug)] -pub struct EncodedMessage { - pub peer: String, - pub body: bytes::Bytes, - pub kind: MessageKind, +pub(crate) struct EncodedMessage { + pub(crate) peer: String, + /// Where the transport sends it: `peer` itself for a transport whose identities are + /// addresses, the peer's address under the identity layer (`transport::identity`). + pub(crate) address: String, + pub(crate) body: bytes::Bytes, + pub(crate) kind: MessageKind, /// Bytes the message takes on the wire (without the transport's own framing). - pub payload_len: usize, + pub(crate) payload_len: usize, + /// The message itself, under the paths layer (`transport::paths`): to encode it + /// again for another transport when a send there fails. None from a transport. + pub(crate) message: Option<std::sync::Arc<ParsedMessage>>, } /// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`). -pub type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> +pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send + Sync + 'static; -pub type MessageHandler = Box<HandlerFn>; +pub(crate) type MessageHandler = Box<HandlerFn>; + +/// Whether `message` is a PING or PONG. The paths layer times them, so a transport's +/// consumer hands them on as soon as they are read (`hand_on_probe`), not after the +/// messages read before them are handled (which can take a while: sync). +pub(crate) fn is_probe(message: &ParsedMessage) -> bool { + matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) +} + +/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own. +pub(crate) fn hand_on_probe( + handler: &std::sync::Arc<HandlerFn>, + sender: String, + message: ParsedMessage, + payload_len: usize, +) { + let handler = std::sync::Arc::clone(handler); + tokio::spawn(async move { + if let Err(error) = handler(sender.clone(), message, payload_len).await { + tracing::warn!(peer = %sender, error = %format!("{error:#}"), "Failed to process inbound message"); + } + }); +} #[async_trait] -pub trait Transport: Send + Sync { +pub(crate) trait Transport: Send + Sync { /// This node's identity on the transport (for Mycelium: its public key). fn local_id(&self) -> &str; + /// Whether `id` can be a node's identity on this transport (for Mycelium: a public + /// key). Identities other peers name to us (in a HELLO_REPLY) are checked with it. + fn is_peer_id(&self, id: &str) -> bool; + + /// How long the transport tries to deliver a message before dropping it: after that, + /// a message sent is delivered or lost for good. + fn delivery_limit(&self) -> std::time::Duration; + + /// Where this node can be reached, as addresses other nodes can use: published in + /// the member list (`swarm::members`). By default its identity, for a transport + /// whose identities are addresses (Mycelium: a public key). + fn local_addresses(&self) -> Vec<String> { + vec![self.local_id().to_owned()] + } + + /// For a transport under the identity layer: the transport identity at `target`, an + /// address of this transport without its scheme, noting where it is if `target` says + /// more than who (as a libp2p multiaddr ending in the peer ID does). None if + /// `target` isn't one of this transport's. By default `target` itself, for a + /// transport whose identities are addresses. + fn locate(&self, target: &str) -> Option<String> { + self.is_peer_id(target).then(|| target.to_owned()) + } + + /// For a transport under the identity layer: where the node with transport identity + /// `target` is, as addresses others can use (`locate` reads them). By default + /// `target` itself, as for `locate`. + fn locations(&self, target: &str) -> Vec<String> { + vec![target.to_owned()] + } + + /// Where `peer` can be reached, as far as this node knows: passed on to others (in + /// a HELLO_REPLY). By default its identity, as for `local_addresses`. + fn addresses_of(&self, peer: &str) -> Vec<String> { + vec![peer.to_owned()] + } + + /// The transports this node is on, as the schemes of their addresses + /// (`swarm::topology`), whether or not it has an address on each (one that only + /// connects out has none). By default the empty scheme, for a transport whose + /// identities are addresses. + fn transports(&self) -> Vec<String> { + vec![String::new()] + } + + /// Seeds the transport found by itself since last asked (libp2p: nodes on the LAN), + /// for a node that is joining. By default none. + fn discovered_seeds(&self) -> Vec<String> { + Vec::new() + } + + /// Whether this node has a way to reach `peer` at all: an address of it on one of + /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`). + /// By default yes, as for `local_addresses`. + fn can_reach(&self, peer: &str) -> bool { + let _ = peer; + true + } + + /// Where others say `peer` can be reached (the member list, a HELLO_REPLY): hints, + /// to be checked before they are relied on. By default nothing to learn, as for + /// `local_addresses`. + fn learn_addresses(&self, peer: &str, addresses: &[String]) { + let _ = (peer, addresses); + } + + /// Where messages to `peer` go now. None if nowhere, and by default, for a + /// transport whose identities are addresses (`peer` is where they go). + fn destination(&self, peer: &str) -> Option<Destination> { + let _ = peer; + None + } + + /// Where messages to `target` (a transport identity) go now, as an address that + /// names `target` too: the remote address of the connection in use, say. By default + /// nowhere but `target` itself (Mycelium). + fn location(&self, target: &str) -> Option<String> { + let _ = target; + None + } + + /// How this node is connected to `target` (a transport identity), if it is. By + /// default not at all, for a transport without connections (Mycelium). + fn connection(&self, target: &str) -> Option<Connection> { + let _ = target; + None + } + + /// The smoothed round-trip time of the path messages to `peer` go over, if + /// measured (`transport::paths`). By default none. + fn rtt(&self, peer: &str) -> Option<std::time::Duration> { + let _ = peer; + None + } + + /// Whether the transport answers PINGs with PONGs itself (the paths layer, + /// `transport::paths`); if not (a bare transport, in tests), the node does. By + /// default not. + fn answers_pings(&self) -> bool { + false + } + + /// Figures about the transport itself (`stats`). By default none. + fn stats(&self) -> stats::TransportStats { + stats::TransportStats::default() + } + /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's /// identity, as `local_id` gives it on that node. async fn resolve_seed(&self, seed: &str) -> Result<String>; @@ -347,7 +741,16 @@ mod tests { let mut shuffled = heads.clone(); rng.shuffle(&mut shuffled); assert_eq!(summary, HeadsSummary::of(shuffled)); - assert_eq!(summary, HeadsSummary::Digest(summary.digest())); + assert_eq!( + summary, + HeadsSummary::Digest { + digest: summary.digest(), + sample: Vec::new(), + } + ); + // A sample is some of the heads. + assert!(summary.listed().iter().all(|head| heads.contains(head))); + assert_eq!(summary.listed().len(), n.min(MAX_LISTED_HEADS)); let mut other = heads.clone(); match rng.below(3) { @@ -364,4 +767,22 @@ mod tests { assert_ne!(summary, HeadsSummary::of(other)); }); } + + /// Successive samples cover every head. + #[test] + fn samples_rotate_over_all_heads() { + let heads: Vec<ChangeHash> = (0..100u8).map(|n| ChangeHash([n; 32])).collect(); + let mut seen = std::collections::HashSet::new(); + for _ in 0..heads.len().div_ceil(MAX_LISTED_HEADS) + 1 { + seen.extend(HeadsSummary::of(heads.clone()).listed().iter().copied()); + } + // Other tests sample concurrently and move the window too, so allow a few more. + for _ in 0..100 { + if seen.len() == heads.len() { + break; + } + seen.extend(HeadsSummary::of(heads.clone()).listed().iter().copied()); + } + assert_eq!(seen.len(), heads.len()); + } } |