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(crate) use crate::document::replica::Epoch; // 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(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(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 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 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(crate) enum HeadsSummary { /// Sorted. Heads(Vec), 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, }, } /// Where the next sample starts (`HeadsSummary::of`). static SAMPLE_OFFSET: AtomicUsize = AtomicUsize::new(0); impl HeadsSummary { pub(crate) fn of(mut heads: Vec) -> Self { heads.sort_unstable(); if heads.len() <= MAX_LISTED_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(crate) fn digest(&self) -> [u8; 32] { match self { Self::Heads(heads) => digest_of(heads), Self::Digest { digest, .. } => *digest, } } /// The heads, if all are listed. pub(crate) fn heads(&self) -> Option<&[ChangeHash]> { match self { Self::Heads(heads) => Some(heads), 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, } } } impl PartialEq for HeadsSummary { fn eq(&self, other: &Self) -> bool { match (self, other) { (Self::Heads(a), Self::Heads(b)) => a == b, _ => self.digest() == other.digest(), } } } impl Eq for HeadsSummary {} /// SHA-256 over sorted heads. fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { let mut hasher = Sha256::new(); for head in sorted_heads { hasher.update(head.0); } hasher.finalize().into() } /// 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. #[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(crate) epoch_seq: u32, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub(crate) ready: bool, /// The sender would sync with the recipient: counts it as an active member on its /// own epoch. pub(crate) syncs_with_you: bool, } /// Cheap to clone: large parts are shared (`bytes::Bytes`). #[derive(Debug, Clone)] pub(crate) enum ParsedMessage { Hello(HelloRequest), HelloReply(HelloReply), Sync(SyncEnvelope), 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, Serialize, Deserialize)] pub(crate) enum BlobId { /// An Automerge change, by its hash: e.g. an epoch's snapshot change. Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash), /// Bytes by their BLAKE3 hash: e.g. the content of a remake. 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:{}", 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, Serialize, Deserialize)] pub(crate) struct BlobMessage { /// The sender's epoch (every message carries it). #[serde(with = "crate::transport::wire::codec::epoch")] pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub(crate) epoch_seq: u32, pub(crate) id: BlobId, pub(crate) body: BlobBody, } #[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, delta: Option, }, /// 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. Missing, } /// Sizes instead of contents: logged messages stay short. impl std::fmt::Debug for BlobBody { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { BlobBody::Request { chunks, delta } => f .debug_struct("Request") .field("chunks", chunks) .field("delta", &delta.is_some()) .finish(), BlobBody::Manifest { size, chunk_size, hashes, } => f .debug_struct("Manifest") .field("size", size) .field("chunk_size", chunk_size) .field("chunks", &hashes.len()) .finish(), BlobBody::Chunk { index, data } => f .debug_struct("Chunk") .field("index", index) .field("bytes", &data.len()) .finish(), BlobBody::Missing => f.write_str("Missing"), } } } /// What a message is, without its contents. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum MessageKind { Hello, HelloReply, Sync, Status, StatusReply, Blob, Who, Me, Ping, Pong, } impl ParsedMessage { 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, } } /// 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) => 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(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(crate) swarm_id: Option, #[serde(with = "crate::transport::wire::codec::epoch")] pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub(crate) epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. #[serde(with = "serde_bytes")] pub(crate) auth: Vec, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub(crate) ready: bool, /// Random per run of the sender; the reply echoes it. #[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, } /// 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, } /// 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: `:`. 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, /// Every address it proved itself at, best first, as the paths layer measures them /// (`transport::paths`). pub(crate) paths: Vec, } /// 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, #[serde(with = "crate::transport::wire::codec::epoch")] pub(crate) epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub(crate) epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. #[serde(with = "serde_bytes")] pub(crate) auth: Vec, /// The sender is ready. 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. #[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(crate) away_ms: Option, /// The sender's own run (as its HELLOs carry it): a new one means it restarted. #[serde(with = "crate::transport::wire::codec::run_id")] pub(crate) sender_run_id: RunId, } /// 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. #[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(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(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, /// Sorted. #[serde(with = "crate::transport::wire::codec::change_hashes")] pub(crate) to: Vec, } impl DeltaSpec { pub(crate) fn new(mut from: Vec, mut to: Vec) -> 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::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(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(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>, } /// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`). pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin> + Send>> + Send + Sync + 'static; pub(crate) type MessageHandler = Box; #[async_trait] 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 { 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 { 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 { 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 { 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 { 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 { 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 { let _ = peer; 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 { 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 { 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; fn encode(&self, peer: &str, message: &ParsedMessage) -> Result; async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>; /// Sends an outbound message payload to a targeted peer on a specific topic. async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> { let encoded = self.encode(peer, message)?; self.send_encoded(&encoded).await } /// Starts the long-polling inbound event loops, driving messages to the provided handler closure. async fn start_consumer(&self, handler: MessageHandler) -> Result<()>; /// Asynchronously monitors for fatal network client or sub-process failures. async fn wait_for_exit(&self) -> std::io::Result; } #[cfg(test)] mod tests { use super::*; use crate::testrand::for_seeds; /// Two summaries are equal exactly when the heads are, in any order, whichever form /// (listed or digest) each took. #[test] fn heads_summary_equality_is_set_equality() { for_seeds(300, |rng| { let n = rng.between(0, 2 * MAX_LISTED_HEADS + 2); let heads: Vec = (0..n).map(|_| ChangeHash(rng.bytes32())).collect(); let summary = HeadsSummary::of(heads.clone()); assert_eq!(summary.heads().is_some(), n <= MAX_LISTED_HEADS); let mut shuffled = heads.clone(); rng.shuffle(&mut shuffled); assert_eq!(summary, HeadsSummary::of(shuffled)); 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) { 1 if !other.is_empty() => { let i = rng.below(other.len()); other.remove(i); } 2 if !other.is_empty() => { let i = rng.below(other.len()); other[i] = ChangeHash(rng.bytes32()); } _ => other.push(ChangeHash(rng.bytes32())), } assert_ne!(summary, HeadsSummary::of(other)); }); } /// Successive samples cover every head. #[test] fn samples_rotate_over_all_heads() { let heads: Vec = (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()); } }