diff --git a/Cargo.lock b/Cargo.lock index eec2ff6..390719e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2480,6 +2480,7 @@ dependencies = [ "dirs 6.0.0", "futures", "hex", + "libc", "lru", "once_cell", "parking_lot", diff --git a/Cargo.toml b/Cargo.toml index ae07e11..e0e6809 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,9 @@ tokio-util = { version = "0.7", features = ["rt"] } # Fix wyz 0.5.0 compatibility issue with tap 1.0 (CI build failure) wyz = "=0.5.1" +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] tempfile = "3.17" diff --git a/docs/adr/ADR-017-routing-table-snapshot-across-restart.md b/docs/adr/ADR-017-routing-table-snapshot-across-restart.md new file mode 100644 index 0000000..84a6455 --- /dev/null +++ b/docs/adr/ADR-017-routing-table-snapshot-across-restart.md @@ -0,0 +1,76 @@ +# ADR-017: Persist the Whole Routing Table Across a Restart + +## Status + +Proposed + +## Context + +`saorsa-core` persists exactly one thing across a restart: the close-group cache, which holds the `k` peers nearest to self. Everything else in the routing table is discarded and rebuilt by periodic bucket refresh at two buckets per 7.5 to 12.5 minutes, a cadence this crate's own source comment describes as approximately once-per-day full-table maintenance. + +Consumers above this crate do not ask "who are my neighbours". They ask "am I among the `w` closest to key `K`", which `find_closest_nodes_local_with_self` answers from whatever the local table happens to hold. That predicate has no notion of confidence and no failure value: if the table cannot name `w` peers closer to `K` than self, the answer is yes. A node that has just restarted therefore answers yes for most of the keyspace, and keeps doing so for hours. + +The effect is measured in production rather than theoretical. On the Autonomi production fleet, two independent restarted services were ranked against the 806-peer fleet by XOR distance to 200 keys each was actively claiming: median true rank 27 and 24, with **0 of 200 keys placing them in the true closest 7 or the true closest 9**. Fleet-wide, restarted nodes reported 40 to 60% of their stored records as in range where a converged table reports about 1%. Those figures come from production telemetry held outside this repository; they are cited here as the motivation, and nothing in this ADR depends on their exact values. + +A close group cannot stand in for a routing table, and the reason is combinatorial rather than a matter of degree. For a key `K`, write `c = CPL(self, K)`. Any peer `p` with `CPL(p, self) == c` agrees with self on bits `[0, c)` and differs at bit `c`; `K` agrees with self on `[0, c)` and also differs at bit `c`; therefore `p` agrees with `K` at bit `c` and is strictly closer to `K` than self is. Every peer in bucket `c` answers the question. A node holding `w` of them always answers correctly; a node whose bucket `c` is empty cannot, however many neighbours it has. Preserving peers across every bucket is what makes a restored table answer as the original did. + +Simulated on an 891-node network against correct shares of 1.01% (width 9) and 2.24% (width 20). The simulation lives outside this repository, so these numbers are supporting evidence for the shape of the fix, not a claim this repository can reproduce: + +| restored table | width 9 claim | width 20 claim | +|---|---|---| +| converged table (steady state) | 1.05% | 2.01% | +| today: nearest 20 only | 41.4% | 95.6% | +| nine peers per bucket | 1.05% | 12.3% | +| whole capped table (~129 entries, ~25 KB) | 1.05% | 2.01% | + +## Decision + +Persist the whole capped routing table to its own snapshot file, and restore it as dial candidates at startup. + +1. **Separate file.** The snapshot is written alongside `close_group_cache.json`, not in place of it, so an older binary cannot mistake one for the other and a downgrade behaves exactly as today. +2. **Every bucket, not the nearest `k`.** The snapshot carries the table up to its per-bucket capacity, subject to a flat ceiling of 1,024 peers and 8 addresses each. Nine peers per bucket fixes the narrow width and leaves the wide one at 12.3%, so the shape is the whole table rather than a per-bucket floor. The ceiling is well above the ~129 entries a fleet of this size produces; a network large enough to reach it would restore a truncated table, which is still strictly better than the close group alone. +3. **Two bindings, both required.** The schema version, so an unrecognised version is discarded rather than coerced; and the owning node id, which is load-bearing rather than hygiene, because bucket indices are relative to the owner and another node's snapshot describes a different partition of the id space. A snapshot older than seven days, or dated more than five minutes in the future, is refused. That age bound is the snapshot's own, deliberately not the close-group cache's one hour: the cache's bound exists for trust scores on `k` neighbours, while this file answers which parts of the id space the node knew about, which does not rot on the same timescale, and a maintenance window must not cost a node its table. +4. **Bounded load.** The file is opened once and checked through that handle, never re-opened by path, so a file swapped in between the check and the read cannot bypass either. It is refused if it is not a regular file or exceeds 512 KB, the read itself is capped at the same ceiling, and the peer and per-peer address counts are re-applied after parsing, so a corrupt or hostile file cannot become an unbounded dial set. +5. **Restored peers are dial candidates only.** They are not inserted into the routing table and confer no authority until dialled and identity-verified through the ordinary path. Routing-table membership is an authorization fact for callers above this crate, and a file on disk must not be able to grant it. For the same reason the snapshot records **no trust scores**, unlike the close-group cache: pre-dial trust restoration is defensible for `k` vetted neighbours and not for hundreds of unverified peers. +6. **Restored peers do not seed DHT discovery.** A dialled, identity-verified peer is already admitted to the routing table by the connection path. Adding the restored set to the discovery seed list would issue a serial `FIND_NODE` per peer to rediscover the table just restored, and then serially dial everything those queries returned, so the phase's own bound would be defeated by the phase after it. +7. **Bounded cost, then today's behaviour.** Restoration dials at bounded concurrency (16, against 4 for bootstrap dials, because the set is an order of magnitude larger). A 20-second budget stops *new* peers being dialled; dials already in flight are allowed to finish, because cancelling a handshake mid-flight leaves the far side holding a half-open connection. The phase's bound is therefore the budget plus the last peer's attempts, which is why a snapshot peer is tried at no more than two addresses. Whatever is not restored refills through ordinary discovery, which is exactly the behaviour of a node that had no snapshot. +8. **Once per process, and not for clients.** A re-bootstrap does not replay the snapshot, since the table it would restore is the one the node already has. `NodeMode::Client` skips restoration entirely and keeps its existing six-peer startup bound: a client does not serve the DHT, so it never asks the question this repairs. +9. **A save cannot shrink the snapshot while the table is still restoring.** The periodic and shutdown saves write the file; the post-bootstrap save does not. Nothing is written until the restore step has decided how many peers it recovered, so a node stopped before that step — and a client, which never restores — cannot overwrite a good file with an empty or partial table. For an hour after that decision, a save carrying fewer peers than the restore recovered is skipped, because a node stopped mid-restore holds only what it has re-dialled so far and would otherwise shrink its own file a little further on every cycle. After that hour the live table is the node's best knowledge, so the floor stops applying and the file keeps being refreshed rather than ageing out. An empty table is never written. + +## Alternatives considered + +- **Keep the close-group cache and choose its 20 peers more cleverly.** Rejected on measurement: at that size composition is second-order. Nearest-20 claims 40.5% of the keyspace at width 9 and an arbitrary 20 claims 43.3%. Size dominates. +- **Persist a fixed floor of peers per bucket.** Rejected as insufficient rather than wrong. Nine per bucket answers width 9 correctly but leaves width 20 at 12.3%, and this crate does not know the widths its consumers use. +- **Bind the snapshot to a network fingerprint derived from the configured bootstrap list.** Rejected: it hashes mutable address spellings rather than a stable network identity, and routine seed rotation would invalidate every node's snapshot during exactly the rollout this repairs. A snapshot carried to another network costs failed dials inside the existing budget, which is the same cost as a cold start. +- **Checksum the payload.** Rejected: it provides no authenticity for a locally written file, while JSON parsing is already all-or-nothing, the write is atomic, and live identity verification governs what the contents can achieve. +- **A config kill switch for restoration.** Rejected: new public API for a behaviour that rolls back by shipping the previous version or deleting the file, and one more untested branch through startup. +- **Estimate network size locally and refuse keys beyond an inferred horizon.** Rejected on two grounds. It admits every key out to true rank `slack × width`, which at the studied slack is rank 36, above the rank 24 to 27 band the production over-claim actually occupied: it caught 14.3% of that band with 6 of 10 simulated nodes catching none. It is also punishable, because a refusal resurfaces as an absent answer at the requester's audit, and peers near a victim's id can shrink its estimate. +- **Wait out the transient by tuning the consumer's retention timers.** Rejected: it treats a wrong answer as a scheduling problem, and it makes nodes act fastest exactly when their routing table is least trustworthy. + +## Consequences + +### Positive + +- A restarted node is expected to answer responsibility questions as its converged self did, at both widths, instead of claiming most of the keyspace for hours. That is the simulated result and the intent of the design; it is not yet demonstrated against a live network, see Validation. +- The fix is combinatorial, so it does not depend on network size, key distribution, or an estimator an adversary could move. +- Consumers gain nothing new to configure. The predicate, its widths and its call sites are untouched, and there is no new public API. + +### Negative + +- A new on-disk artifact to version and keep compatible. +- Startup dials a larger candidate set. New peers stop being dialled at the 20-second budget, but attempts already in flight still run, so the phase can exceed the budget by one peer's two address attempts. +- A snapshot full of departed peers spends that budget and yields little. No new peer is dialled after it, though attempts already in flight still finish. +- Every periodic and shutdown close-group save now writes a second small file before returning. + +### Neutral + +- The close-group cache remains, unchanged, with its own trust import and its own validity rules. +- A brand-new node with no snapshot is unaffected and still refills at the ordinary refresh cadence. Accelerating that case is a separate change and is deliberately not in scope here. +- Restored peers are unverified until dialled, so the table refills behind the file's contents rather than instantly. + +## Validation + +- Unit tests in `src/bootstrap/routing_snapshot.rs` for the file contract: disk round trip, missing file treated as absence, truncated file reported, oversized file refused without being read, peer cap re-applied on load, foreign owner refused, unknown schema version refused, and staleness bounded on both sides of now. +- Unit tests in `src/network.rs` for the restore path's candidate selection: self excluded, addresses already queued by an earlier bootstrap priority not redialled, undialable addresses dropped, repeated peers deduplicated, the per-peer dial list bounded, and a full table producing one candidate per peer. +- Not covered by tests in this PR: the dial phase itself against a live transport, including budget expiry, client-mode exclusion, and the no-shrink rule on saves. Those need a multi-node harness. +- **No testnet or production measurement of this change exists.** The over-claim it targets is measured in production; the fix is evidenced by simulation and unit tests only. A dev testnet run is the next step, and nothing here claims fleet readiness. diff --git a/docs/adr/README.md b/docs/adr/README.md index 72a4d4e..17d6c4d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -43,6 +43,7 @@ An Architecture Decision Record (ADR) is a document that captures an important a | [ADR-007](./ADR-007-adaptive-networking.md) | Adaptive Networking with ML | Accepted | Machine learning for dynamic routing optimization | | [ADR-008](./ADR-008-bootstrap-delegation.md) | Bootstrap Peer Discovery Scope | Superseded | Historical peer discovery design replaced by configured peers plus DHT discovery | | [ADR-016](./ADR-016-close-group-cache-validity.md) | Age-Bounded, Periodically Refreshed Close-Group Cache | Proposed | Bound persisted close-group age and refresh it safely during normal running | +| [ADR-017](./ADR-017-routing-table-snapshot-across-restart.md) | Persist the Whole Routing Table Across a Restart | Proposed | Restore every bucket as dial candidates so a restarted node stops claiming most of the keyspace | ### Messaging diff --git a/src/bootstrap/mod.rs b/src/bootstrap/mod.rs index 2c18ca5..753ac84 100644 --- a/src/bootstrap/mod.rs +++ b/src/bootstrap/mod.rs @@ -10,8 +10,13 @@ // distributed under these licenses is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -//! Close-group cache for warm-loading trusted peers across restarts. +//! Persisted peer knowledge for warm-starting across restarts. +//! +//! Two files, written side by side: [`cache`] holds the `k` peers nearest to +//! self, and [`routing_snapshot`] holds the whole routing table, which is what +//! a node consults to decide whether anyone is closer to a given key than it is. pub mod cache; +pub(crate) mod routing_snapshot; pub use cache::{CachedCloseGroupPeer, CloseGroupCache}; diff --git a/src/bootstrap/routing_snapshot.rs b/src/bootstrap/routing_snapshot.rs new file mode 100644 index 0000000..af1afb1 --- /dev/null +++ b/src/bootstrap/routing_snapshot.rs @@ -0,0 +1,475 @@ +// Copyright 2024 Saorsa Labs Limited +// +// This software is licensed under the MIT license or the Apache License, Version 2.0 +// , at your +// option. This file may not be copied, modified, or distributed except +// according to those terms. +// +// Unless required by applicable law or agreed to in writing, software +// distributed under these licenses is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +//! The whole routing table, persisted across a restart. +//! +//! [`CloseGroupCache`](super::cache::CloseGroupCache) persists the `k` peers +//! nearest to self, which is the right shape for reconnecting to a close group +//! and the wrong shape for reconstructing a routing table: the peers that +//! answer "is anyone closer to this key than me?" for a *distant* key are +//! exactly the ones a close group leaves out. For a key `K` sharing `c` leading +//! bits with self, every peer in bucket `c` is strictly closer to `K` than self +//! is, so a node that kept those peers answers correctly and a node whose +//! bucket `c` is empty cannot, however many neighbours it has. +//! +//! Restored peers are **dial candidates only**. They are dialled and +//! identity-verified through the ordinary path before they can enter the +//! routing table, and the snapshot carries no trust scores, because a file on +//! disk must not grant routing-table membership or above-neutral trust. + +use std::io::Write as _; +use std::path::Path; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +use crate::PeerId; +use crate::address::MultiAddr; + +/// Filename for the routing snapshot. +/// +/// Distinct from `close_group_cache.json`, which is still written alongside it, +/// so an older binary never reads a whole-table snapshot as a close group. +pub(crate) const ROUTING_SNAPSHOT_FILENAME: &str = "routing_snapshot.json"; + +/// Schema version. An unrecognised version is discarded, never coerced. +const SCHEMA_VERSION: u32 = 1; + +/// Maximum age of a snapshot that is still worth dialling. +/// +/// Deliberately not the close-group cache's one hour. That bound exists for +/// trust scores on `k` neighbours; this file answers which parts of the id +/// space the node knew about, which does not rot on the same timescale. A +/// maintenance window or a host move must not cost a node its table, and a +/// snapshot of departed peers already costs nothing beyond failed dials. +const MAX_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60); + +/// Tolerated wall-clock skew for a timestamp in the future, so a clock jump +/// cannot make a snapshot look fresh indefinitely. +const MAX_FUTURE_TIMESTAMP_SKEW: Duration = Duration::from_secs(5 * 60); + +/// Largest snapshot file that will be read. A full table is ~25 KB; anything +/// past this is not a snapshot this node wrote. +const MAX_SNAPSHOT_BYTES: u64 = 512 * 1024; + +/// Hard caps applied when writing and after parsing, so a corrupt or hostile +/// file cannot turn into an unbounded dial set. +const MAX_SNAPSHOT_PEERS: usize = 1024; +const MAX_ADDRESSES_PER_PEER: usize = 8; + +/// A peer recorded in a snapshot: identity and addresses, nothing else. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct SnapshotPeer { + /// Peer identity, re-verified on dial before it can enter the routing table. + pub peer_id: PeerId, + /// Addresses last known to reach this peer. + pub addresses: Vec, +} + +/// Why a snapshot on disk was not used. +/// +/// A node that silently cold-starts looks identical to one that never had a +/// snapshot, so every rejection is reportable. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub(crate) enum SnapshotRejection { + /// The schema version is not one this build understands. + #[error("unknown schema version {found} (this build writes {expected})")] + UnknownSchemaVersion { + /// Version found in the file. + found: u32, + /// Version this build writes. + expected: u32, + }, + /// The snapshot was written by a different node. Bucket indices are + /// relative to the owner, so another node's snapshot is meaningless here. + #[error("written by a different node")] + ForeignOwner, + /// Older than [`MAX_AGE`], or dated implausibly far in the future. + #[error("stale or implausibly future-dated")] + Stale, + /// The file exists but could not be used. + #[error("unusable snapshot file: {0}")] + Unreadable(String), +} + +/// A persisted routing table. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct RoutingSnapshot { + /// Schema version of this file. + pub schema_version: u32, + /// Node that wrote it. Bucket indices are relative to this id. + pub owner: PeerId, + /// When it was written (seconds since UNIX epoch). + pub saved_at_epoch_secs: u64, + /// Every routing-table peer at the time of writing, across all buckets. + pub peers: Vec, +} + +impl RoutingSnapshot { + /// Build a snapshot of the current routing table, capped. + pub fn new(owner: PeerId, saved_at_epoch_secs: u64, mut peers: Vec) -> Self { + peers.truncate(MAX_SNAPSHOT_PEERS); + for peer in &mut peers { + peer.addresses.truncate(MAX_ADDRESSES_PER_PEER); + } + Self { + schema_version: SCHEMA_VERSION, + owner, + saved_at_epoch_secs, + peers, + } + } + + /// The peers this node may dial, or why the snapshot was not used. + /// + /// # Errors + /// + /// Returns the first binding that failed, so the caller can log why. + pub fn peers_for( + &self, + expected_owner: &PeerId, + now_epoch_secs: u64, + ) -> Result<&[SnapshotPeer], SnapshotRejection> { + if self.schema_version != SCHEMA_VERSION { + return Err(SnapshotRejection::UnknownSchemaVersion { + found: self.schema_version, + expected: SCHEMA_VERSION, + }); + } + if self.owner != *expected_owner { + return Err(SnapshotRejection::ForeignOwner); + } + if self.is_stale(now_epoch_secs) { + return Err(SnapshotRejection::Stale); + } + Ok(&self.peers) + } + + /// Older than [`MAX_AGE`], or dated implausibly far in the future. + fn is_stale(&self, now_epoch_secs: u64) -> bool { + let future_skew = self.saved_at_epoch_secs.saturating_sub(now_epoch_secs); + if future_skew > MAX_FUTURE_TIMESTAMP_SKEW.as_secs() { + return true; + } + now_epoch_secs.saturating_sub(self.saved_at_epoch_secs) > MAX_AGE.as_secs() + } + + /// Write the snapshot to `{dir}/routing_snapshot.json`. + /// + /// Written to a temporary file in the same directory and persisted by + /// rename, so a crash mid-write leaves either the previous snapshot or + /// none, never a half-written one. + /// + /// # Errors + /// + /// Returns an error if the directory cannot be created, or the file cannot + /// be serialized, written, or persisted. + pub async fn save_to_dir(&self, dir: &Path) -> anyhow::Result<()> { + tokio::fs::create_dir_all(dir).await.map_err(|e| { + anyhow::anyhow!( + "failed to create routing snapshot directory {}: {e}", + dir.display() + ) + })?; + + let path = dir.join(ROUTING_SNAPSHOT_FILENAME); + let json = serde_json::to_vec(self) + .map_err(|e| anyhow::anyhow!("failed to serialize routing snapshot: {e}"))?; + + let dir_owned = dir.to_path_buf(); + tokio::task::spawn_blocking(move || { + let mut tmp = tempfile::NamedTempFile::new_in(&dir_owned).map_err(|e| { + anyhow::anyhow!("failed to create temp file in {}: {e}", dir_owned.display()) + })?; + tmp.write_all(&json) + .map_err(|e| anyhow::anyhow!("failed to write routing snapshot: {e}"))?; + tmp.persist(&path).map_err(|e| { + anyhow::anyhow!( + "failed to persist routing snapshot to {}: {e}", + path.display() + ) + })?; + Ok(()) + }) + .await + .map_err(|e| anyhow::anyhow!("routing snapshot save task panicked: {e}"))? + } + + /// Read the snapshot from `{dir}/routing_snapshot.json`. + /// + /// Returns `Ok(None)` when there is no snapshot. A file that exists but is + /// oversized, not a regular file, a symlink, or unparseable is reported + /// rather than treated as absence, because the two need different operator + /// responses. + /// + /// # Errors + /// + /// Returns [`SnapshotRejection::Unreadable`] with the underlying reason. + pub async fn load_from_dir(dir: &Path) -> Result, SnapshotRejection> { + let path = dir.join(ROUTING_SNAPSHOT_FILENAME); + + // Open once and check the handle, not the path: checking the path and + // then reading it again is two different files if anything replaces it + // in between. The read is separately capped, so the size check cannot + // be sidestepped by a file that grows after it is opened. + let file = match open_snapshot_file(&path).await { + Ok(file) => file, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(e) => return Err(SnapshotRejection::Unreadable(e.to_string())), + }; + let metadata = file + .metadata() + .await + .map_err(|e| SnapshotRejection::Unreadable(e.to_string()))?; + if !metadata.is_file() { + return Err(SnapshotRejection::Unreadable("not a regular file".into())); + } + if metadata.len() > MAX_SNAPSHOT_BYTES { + return Err(SnapshotRejection::Unreadable(format!( + "{} bytes exceeds the {MAX_SNAPSHOT_BYTES} byte limit", + metadata.len() + ))); + } + + let mut bytes = Vec::new(); + let mut bounded = tokio::io::AsyncReadExt::take(file, MAX_SNAPSHOT_BYTES + 1); + tokio::io::AsyncReadExt::read_to_end(&mut bounded, &mut bytes) + .await + .map_err(|e| SnapshotRejection::Unreadable(e.to_string()))?; + if bytes.len() as u64 > MAX_SNAPSHOT_BYTES { + return Err(SnapshotRejection::Unreadable(format!( + "exceeds the {MAX_SNAPSHOT_BYTES} byte limit while reading" + ))); + } + let mut snapshot: Self = serde_json::from_slice(&bytes) + .map_err(|e| SnapshotRejection::Unreadable(e.to_string()))?; + + // Re-apply the write-side caps: the bound has to hold for a file this + // process did not write. + snapshot.peers.truncate(MAX_SNAPSHOT_PEERS); + for peer in &mut snapshot.peers { + peer.addresses.truncate(MAX_ADDRESSES_PER_PEER); + } + Ok(Some(snapshot)) + } +} + +/// Open the snapshot for reading without following a symlink and without +/// blocking on a special file. +/// +/// A plain `open` of a FIFO placed at this fixed path blocks until a writer +/// appears, which would stall bootstrap before the regular-file check on the +/// handle could reject it. `O_NONBLOCK` makes that open return immediately and +/// is inert for regular files, and `O_NOFOLLOW` refuses a symlink outright: +/// this node only ever writes a regular file here, by rename. +#[cfg(unix)] +async fn open_snapshot_file(path: &Path) -> std::io::Result { + use std::os::unix::fs::OpenOptionsExt as _; + + let path = path.to_path_buf(); + let file = tokio::task::spawn_blocking(move || { + std::fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK | libc::O_NOFOLLOW) + .open(path) + }) + .await + .map_err(|e| std::io::Error::other(format!("snapshot open task panicked: {e}")))??; + Ok(tokio::fs::File::from_std(file)) +} + +/// Non-Unix fallback: a plain open, with the non-regular-file rejection still +/// enforced on the opened handle by the caller. +#[cfg(not(unix))] +async fn open_snapshot_file(path: &Path) -> std::io::Result { + tokio::fs::File::open(path).await +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] +mod tests { + use super::*; + + fn peer() -> SnapshotPeer { + SnapshotPeer { + peer_id: PeerId::random(), + addresses: vec!["/ip4/10.0.1.1/udp/9000/quic".parse().unwrap()], + } + } + + fn snapshot(owner: PeerId, saved_at: u64, count: usize) -> RoutingSnapshot { + RoutingSnapshot::new(owner, saved_at, (0..count).map(|_| peer()).collect()) + } + + #[tokio::test] + async fn round_trips_through_disk() { + let dir = tempfile::tempdir().unwrap(); + let owner = PeerId::random(); + let original = snapshot(owner, 1_000, 130); + + original.save_to_dir(dir.path()).await.unwrap(); + let loaded = RoutingSnapshot::load_from_dir(dir.path()) + .await + .unwrap() + .expect("snapshot present"); + + assert_eq!(loaded.peers, original.peers); + assert_eq!(loaded.peers_for(&owner, 1_000).unwrap().len(), 130); + } + + #[tokio::test] + async fn a_missing_snapshot_is_absence_not_an_error() { + let dir = tempfile::tempdir().unwrap(); + assert!( + RoutingSnapshot::load_from_dir(dir.path()) + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn a_truncated_file_is_reported_not_silently_ignored() { + let dir = tempfile::tempdir().unwrap(); + let owner = PeerId::random(); + snapshot(owner, 1_000, 4) + .save_to_dir(dir.path()) + .await + .unwrap(); + + let path = dir.path().join(ROUTING_SNAPSHOT_FILENAME); + let json = std::fs::read_to_string(&path).unwrap(); + std::fs::write(&path, &json[..json.len() / 2]).unwrap(); + + assert!(matches!( + RoutingSnapshot::load_from_dir(dir.path()).await, + Err(SnapshotRejection::Unreadable(_)) + )); + } + + #[tokio::test] + async fn an_oversized_file_is_refused_without_reading_it() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(ROUTING_SNAPSHOT_FILENAME); + std::fs::write(&path, vec![b'x'; (MAX_SNAPSHOT_BYTES + 1) as usize]).unwrap(); + + assert!(matches!( + RoutingSnapshot::load_from_dir(dir.path()).await, + Err(SnapshotRejection::Unreadable(_)) + )); + } + + #[tokio::test] + async fn a_file_with_too_many_peers_is_capped_on_load() { + let dir = tempfile::tempdir().unwrap(); + let owner = PeerId::random(); + // Build past the cap by hand: `new` caps on the write side. + let oversized = RoutingSnapshot { + schema_version: SCHEMA_VERSION, + owner, + saved_at_epoch_secs: 1_000, + peers: (0..MAX_SNAPSHOT_PEERS + 50).map(|_| peer()).collect(), + }; + let path = dir.path().join(ROUTING_SNAPSHOT_FILENAME); + std::fs::write(&path, serde_json::to_vec(&oversized).unwrap()).unwrap(); + + let loaded = RoutingSnapshot::load_from_dir(dir.path()) + .await + .unwrap() + .expect("snapshot present"); + assert_eq!(loaded.peers.len(), MAX_SNAPSHOT_PEERS); + } + + /// Regression test: a plain blocking open of a FIFO at the snapshot path + /// hangs until a writer appears, so bootstrap never reached the + /// regular-file rejection. + #[cfg(unix)] + #[tokio::test] + async fn a_fifo_at_the_snapshot_path_is_refused_without_blocking() { + use std::os::unix::ffi::OsStrExt as _; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(ROUTING_SNAPSHOT_FILENAME); + let c_path = std::ffi::CString::new(path.as_os_str().as_bytes()).unwrap(); + assert_eq!(unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) }, 0); + + let result = tokio::time::timeout( + Duration::from_secs(10), + RoutingSnapshot::load_from_dir(dir.path()), + ) + .await + .expect("loading must not block on a FIFO"); + assert!(matches!(result, Err(SnapshotRejection::Unreadable(_)))); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_symlink_at_the_snapshot_path_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let owner = PeerId::random(); + let target = dir.path().join("elsewhere.json"); + std::fs::write( + &target, + serde_json::to_vec(&snapshot(owner, 1_000, 4)).unwrap(), + ) + .unwrap(); + let path = dir.path().join(ROUTING_SNAPSHOT_FILENAME); + std::os::unix::fs::symlink(&target, &path).unwrap(); + + assert!(matches!( + RoutingSnapshot::load_from_dir(dir.path()).await, + Err(SnapshotRejection::Unreadable(_)) + )); + } + + #[test] + fn another_nodes_snapshot_is_refused() { + let owner = PeerId::random(); + let someone_else = PeerId::random(); + assert_eq!( + snapshot(owner, 1_000, 4).peers_for(&someone_else, 1_000), + Err(SnapshotRejection::ForeignOwner) + ); + } + + #[test] + fn an_unknown_schema_version_is_refused_rather_than_guessed_at() { + let owner = PeerId::random(); + let mut snap = snapshot(owner, 1_000, 4); + snap.schema_version = SCHEMA_VERSION + 1; + assert!(matches!( + snap.peers_for(&owner, 1_000), + Err(SnapshotRejection::UnknownSchemaVersion { .. }) + )); + } + + #[test] + fn staleness_is_bounded_on_both_sides_of_now() { + let owner = PeerId::random(); + let saved_at = 1_000_000; + let snap = snapshot(owner, saved_at, 4); + + // Fresh, and still usable well past the close-group cache's one hour. + assert!(snap.peers_for(&owner, saved_at).is_ok()); + assert!(snap.peers_for(&owner, saved_at + 6 * 60 * 60).is_ok()); + // Older than the maximum age. + assert_eq!( + snap.peers_for(&owner, saved_at + MAX_AGE.as_secs() + 1), + Err(SnapshotRejection::Stale) + ); + // Dated further in the future than tolerated skew. + assert_eq!( + snap.peers_for(&owner, saved_at - MAX_FUTURE_TIMESTAMP_SKEW.as_secs() - 1), + Err(SnapshotRejection::Stale) + ); + } +} diff --git a/src/network.rs b/src/network.rs index 9d0dde9..5c0c895 100644 --- a/src/network.rs +++ b/src/network.rs @@ -19,6 +19,7 @@ use crate::PeerId; use crate::adaptive::trust::{TrustRecord, TrustSnapshot}; use crate::adaptive::{AdaptiveDHT, AdaptiveDhtConfig, TrustEngine, TrustEvent}; use crate::bootstrap::cache::{CachedCloseGroupPeer, CloseGroupCache}; +use crate::bootstrap::routing_snapshot::{RoutingSnapshot, SnapshotPeer}; use crate::dht::core_engine::AddressType; use crate::dht_network_manager::{DhtNetworkConfig, DhtNetworkEvent, DhtNetworkManager}; use crate::error::{NetworkError, P2PError, P2pResult as Result}; @@ -31,11 +32,11 @@ use dashmap::DashMap; use futures::StreamExt; use parking_lot::Mutex as ParkingMutex; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex as TokioMutex, RwLock, broadcast}; use tokio::time::Instant; @@ -178,6 +179,43 @@ const MAX_CONCURRENT_BOOTSTRAP_DIALS: usize = 4; /// dial every candidate so their routing table converges fully. const CLIENT_BOOTSTRAP_TARGET: usize = 6; +/// Maximum routing-snapshot peers dialled concurrently. +/// +/// Higher than [`MAX_CONCURRENT_BOOTSTRAP_DIALS`] because this set is an order +/// of magnitude larger — a full table rather than a handful of seeds — and the +/// whole value of restoring it is that the table is usable in seconds. Still +/// bounded, so a restart cannot open an unbounded number of simultaneous +/// QUIC+PQC handshakes. +const MAX_CONCURRENT_SNAPSHOT_DIALS: usize = 16; + +/// How long the restored peer count holds authority over saving. +/// +/// It exists to stop a node that is stopped mid-restore from writing the +/// fragment it has re-dialled so far over a complete snapshot. Once the table +/// has had this long to converge it is the node's best knowledge, and a smaller +/// table is a smaller network rather than an unfinished restore, so the floor +/// stops applying and the file keeps being refreshed. +const SNAPSHOT_FLOOR_ENFORCED_FOR: Duration = Duration::from_secs(60 * 60); + +/// The floor has not been decided yet: bootstrap has not reached the restore +/// step, or this is a client, which never restores and never writes one. +const SNAPSHOT_FLOOR_UNRESOLVED: usize = usize::MAX; + +/// Addresses tried per snapshot peer. +/// +/// The budget below stops new peers, so the phase's real bound is the budget +/// plus the last peer's attempts. Two keeps that tail short while still +/// covering a peer whose first address has gone stale. +const MAX_SNAPSHOT_ADDRESSES_DIALLED: usize = 2; + +/// Wall-clock budget for the routing-snapshot dial phase. +/// +/// Startup must not be held hostage to a snapshot full of departed peers. When +/// the budget expires the remaining candidates are abandoned and the table +/// refills through ordinary discovery, which is exactly the behaviour of a node +/// that had no snapshot at all. +const SNAPSHOT_RESTORE_BUDGET: Duration = Duration::from_secs(20); + /// Serde helper — returns `true`. const fn default_true() -> bool { true @@ -825,6 +863,16 @@ pub struct P2PNode { /// Dedicated cancellation token for periodic close-group-cache saves. /// Cancelled and joined before the authoritative shutdown snapshot. close_group_cache_save_shutdown: CancellationToken, + /// Whether the routing snapshot has already seeded a bootstrap in this + /// process, so a re-bootstrap does not replay it. + routing_snapshot_restored: AtomicBool, + /// How many peers the routing snapshot restored at startup, zero when there + /// was none to restore, and [`SNAPSHOT_FLOOR_UNRESOLVED`] until bootstrap + /// has decided. A save must not write a table smaller than this while the + /// floor still applies. Shared with the periodic save task. + routing_snapshot_floor: Arc, + /// When that floor was decided, in epoch seconds. Zero while unresolved. + routing_snapshot_floor_at: Arc, /// Periodic close-group-cache task, retained so shutdown can prevent a /// late periodic write from replacing the final snapshot. @@ -948,6 +996,9 @@ impl P2PNode { start_time: Instant::now(), shutdown: CancellationToken::new(), close_group_cache_save_shutdown: CancellationToken::new(), + routing_snapshot_restored: AtomicBool::new(false), + routing_snapshot_floor: Arc::new(AtomicUsize::new(SNAPSHOT_FLOOR_UNRESOLVED)), + routing_snapshot_floor_at: Arc::new(AtomicU64::new(0)), close_group_cache_save_handle: TokioMutex::new(None), adaptive_dht, is_bootstrapped: Arc::new(AtomicBool::new(false)), @@ -1382,6 +1433,8 @@ impl P2PNode { let peer_id = self.peer_id; let k_value = self.config.dht_config.k_value; let shutdown = self.close_group_cache_save_shutdown.clone(); + let snapshot_floor = Arc::clone(&self.routing_snapshot_floor); + let snapshot_floor_at = Arc::clone(&self.routing_snapshot_floor_at); *task = Some(tokio::spawn(periodic_close_group_cache_save( dht_manager, trust_engine, @@ -1389,6 +1442,8 @@ impl P2PNode { k_value, dir, interval, + snapshot_floor, + snapshot_floor_at, shutdown, ))); info!( @@ -1437,11 +1492,25 @@ impl P2PNode { warn!("Periodic close group cache task failed during shutdown: {error}"); } - // Save close group cache before tearing down the DHT and transport layers. - if let Some(ref dir) = self.config.close_group_cache_dir - && let Err(e) = self.save_close_group_cache(dir, "shutdown").await - { - warn!("Failed to save close group cache on shutdown: {e}"); + // Save close group cache and routing snapshot before tearing down the + // DHT and transport layers. + if let Some(ref dir) = self.config.close_group_cache_dir { + if let Err(e) = self.save_close_group_cache(dir, "shutdown").await { + warn!("Failed to save close group cache on shutdown: {e}"); + } + let now_epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()); + save_routing_snapshot( + self.dht_manager(), + self.peer_id, + now_epoch, + dir, + &self.routing_snapshot_floor, + &self.routing_snapshot_floor_at, + "shutdown", + ) + .await; } // Signal the run loop to exit @@ -2073,7 +2142,27 @@ impl P2PNode { } } - if serial_addr_sets.is_empty() && parallel_addr_sets.is_empty() { + // Priority 2: the routing snapshot, if one validated. Addresses already + // queued as close-group or configured candidates are skipped so a peer + // is never dialled twice. + // + // Restored once per process. A client keeps its existing six-peer + // startup bound: it does not serve the DHT, so it never asks whether it + // is responsible for a key, which is the only question this repairs. + // A re-bootstrap of a running node skips it too — the table it would + // restore is the table the node already has. + let snapshot_addr_sets = if self.config.mode == NodeMode::Client + || self.routing_snapshot_restored.swap(true, Ordering::Relaxed) + { + None + } else { + self.routing_snapshot_dial_sets(&mut seen_addresses).await + }; + + if serial_addr_sets.is_empty() + && parallel_addr_sets.is_empty() + && snapshot_addr_sets.is_none() + { info!("No bootstrap peers configured"); return Ok(()); } @@ -2136,11 +2225,74 @@ impl P2PNode { // before we proceed to the DHT discovery phase below. } + // Phase C: the routing snapshot. + // + // The close group reconnects a neighbourhood; this restores the rest of + // the table, which is what a node consults to decide whether anyone is + // closer to a given key than it is. Dialled last, because the close + // group and the configured peers are what connectivity depends on, and + // concurrently, because the set is an order of magnitude larger. + // + // Two bounds, both deliberate: + // + // - New dials stop once the budget expires; dials already in flight are + // allowed to finish, each bounded by `identity_timeout`. Cancelling a + // handshake mid-flight would leave the far side holding a half-open + // connection, which is a worse outcome than waiting out one timeout. + // - Successes are NOT added to `connected_peer_ids`. A dialled, + // identity-verified peer is already admitted to the routing table by + // the connection path, so seeding DHT discovery with all of them would + // issue a serial FIND_NODE per peer to rediscover the table just + // restored, and then serially dial everything those queries returned. + // + // Every peer here is dialled and identity-verified through the same path + // as any other candidate. Nothing from the file enters the routing table + // on the file's authority alone. + let mut snapshot_dial_candidates = 0usize; + let mut snapshot_dial_successes = 0usize; + if let Some(snapshot_sets) = snapshot_addr_sets { + snapshot_dial_candidates = snapshot_sets.len(); + let deadline = tokio::time::Instant::now() + SNAPSHOT_RESTORE_BUDGET; + let mut snapshot_stream = futures::stream::iter(snapshot_sets.into_iter().map( + |(expected_peer_id, addrs)| async move { + if tokio::time::Instant::now() >= deadline { + return None; + } + self.dial_bootstrap_addr_set( + &addrs, + identity_timeout, + "routing_snapshot", + Some(expected_peer_id), + ) + .await + }, + )) + .buffer_unordered(MAX_CONCURRENT_SNAPSHOT_DIALS); + + while let Some(outcome) = snapshot_stream.next().await { + if outcome.is_some() { + snapshot_dial_successes += 1; + successful_connections += 1; + } + } + + if snapshot_dial_successes < snapshot_dial_candidates { + debug!( + snapshot_dial_successes, + snapshot_dial_candidates, + "Some routing-snapshot peers were not restored; the rest is left to \ + ordinary discovery" + ); + } + } + info!( cache_dial_candidates, cache_dial_successes, configured_dial_candidates, configured_dial_successes, + snapshot_dial_candidates, + snapshot_dial_successes, outbound_bootstrap_successes = successful_connections, outbound_reachable = successful_connections > 0, "Bootstrap reachability summary" @@ -2302,6 +2454,79 @@ impl P2PNode { None } + /// Record how many peers the restore recovered from, and when. + /// + /// Until this is called a snapshot is never written, so a node stopped + /// before the restore step, and a client that never restores at all, cannot + /// overwrite a good file with an empty or partial table. + fn resolve_snapshot_floor(&self, restored_peers: usize) { + let now_epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()); + // Publish the timestamp first and the count second with `Release`, so a + // save that sees a resolved floor cannot also see the zero timestamp it + // was published with and conclude the floor has already expired. + self.routing_snapshot_floor_at + .store(now_epoch, Ordering::Relaxed); + self.routing_snapshot_floor + .store(restored_peers, Ordering::Release); + } + + /// Load the routing snapshot and turn it into dial candidates. + /// + /// Returns `None` when there is nothing usable — no configured directory, + /// no file, or a rejected one. A rejection is logged with its reason: a node + /// that quietly cold-starts on every restart looks exactly like one that + /// never had a snapshot, and an operator needs to tell those apart. + async fn routing_snapshot_dial_sets( + &self, + seen_addresses: &mut HashSet, + ) -> Option)>> { + let dir = self.config.close_group_cache_dir.as_ref()?; + + let snapshot = match RoutingSnapshot::load_from_dir(dir).await { + Ok(Some(snapshot)) => snapshot, + Ok(None) => { + self.resolve_snapshot_floor(0); + return None; + } + Err(rejection) => { + warn!(%rejection, "Discarding routing snapshot"); + self.resolve_snapshot_floor(0); + return None; + } + }; + + let now_epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()); + let peers = match snapshot.peers_for(&self.peer_id, now_epoch) { + Ok(peers) => peers, + Err(rejection) => { + warn!(%rejection, "Discarding routing snapshot"); + self.resolve_snapshot_floor(0); + return None; + } + }; + + // A later save must not write a table smaller than this one, for as + // long as the floor applies. + self.resolve_snapshot_floor(peers.len()); + + let sets = snapshot_dial_sets(peers, &self.peer_id, seen_addresses); + if sets.is_empty() { + return None; + } + + info!( + snapshot_peers = peers.len(), + new_candidates = sets.len(), + age_secs = now_epoch.saturating_sub(snapshot.saved_at_epoch_secs), + "Restoring routing table from snapshot" + ); + Some(sets) + } + /// Persist the current close group peers and their trust scores to disk. async fn save_close_group_cache( &self, @@ -2322,6 +2547,46 @@ impl P2PNode { // disconnect_all_peers and periodic_tasks are now in TransportHandle } +/// Turn snapshot peers into dial candidates, skipping self and anything an +/// earlier bootstrap priority already queued. +/// +/// Extends `seen_addresses` with what it returns, so a peer reachable through +/// the close-group cache and the snapshot is dialled once, not twice. Only +/// dialable (QUIC) addresses survive. +fn snapshot_dial_sets( + peers: &[SnapshotPeer], + self_id: &PeerId, + seen_addresses: &mut HashSet, +) -> Vec<(PeerId, Vec)> { + let mut sets: Vec<(PeerId, Vec)> = Vec::new(); + let mut seen_peers: HashSet = HashSet::new(); + for peer in peers { + if peer.peer_id == *self_id || !seen_peers.insert(peer.peer_id) { + continue; + } + let new_addresses: Vec = peer + .addresses + .iter() + .filter(|addr| { + addr.dialable_socket_addr() + .is_some_and(|socket| !seen_addresses.contains(&socket)) + }) + .take(MAX_SNAPSHOT_ADDRESSES_DIALLED) + .cloned() + .collect(); + if new_addresses.is_empty() { + continue; + } + seen_addresses.extend( + new_addresses + .iter() + .filter_map(MultiAddr::dialable_socket_addr), + ); + sets.push((peer.peer_id, new_addresses)); + } + sets +} + /// Persist a close-group snapshot using owned subsystem handles. /// /// Keeping this separate from `P2PNode` allows the periodic task to own every @@ -2373,9 +2638,78 @@ async fn save_close_group_cache_snapshot( peer_count, dir.display() ); + Ok(()) } +/// Persist the whole routing table as a [`RoutingSnapshot`]. +/// +/// Best-effort and non-fatal: the close-group cache is the established path and +/// must not start failing because a newer, unread file could not be written. +/// +/// Writes nothing until the restore step has decided what the floor is, and +/// then nothing smaller than the table it restored while that floor still +/// applies. Between opening and finishing its restore a node holds only what it +/// has re-dialled so far, and one stopped in that window would otherwise +/// replace a complete snapshot with a fragment, shrinking its own file a little +/// further on every cycle. A client never resolves a floor and so never writes +/// a snapshot it would not read back. +async fn save_routing_snapshot( + dht_manager: &DhtNetworkManager, + peer_id: PeerId, + now_epoch: u64, + dir: &Path, + floor: &AtomicUsize, + floor_at: &AtomicU64, + save_reason: &'static str, +) { + let floor_peers = floor.load(Ordering::Acquire); + if floor_peers == SNAPSHOT_FLOOR_UNRESOLVED { + debug!( + save_reason, + "Skipping routing snapshot save: the restore step has not run" + ); + return; + } + + let peers: Vec = dht_manager + .routing_table_peers() + .await + .into_iter() + .map(|node| SnapshotPeer { + peer_id: node.peer_id, + addresses: node.addresses, + }) + .collect(); + let peer_count = peers.len(); + + if peer_count == 0 { + debug!( + save_reason, + "Skipping routing snapshot save: the table is empty" + ); + return; + } + let floor_applies = now_epoch.saturating_sub(floor_at.load(Ordering::Relaxed)) + < SNAPSHOT_FLOOR_ENFORCED_FOR.as_secs(); + if floor_applies && peer_count < floor_peers { + debug!( + save_reason, + peer_count, + floor_peers, + "Skipping routing snapshot save: it would shrink the snapshot while the table is \ + still restoring" + ); + return; + } + + let snapshot = RoutingSnapshot::new(peer_id, now_epoch, peers); + match snapshot.save_to_dir(dir).await { + Ok(()) => debug!(save_reason, peer_count, "Saved routing snapshot"), + Err(error) => warn!(save_reason, %error, "Failed to save routing snapshot"), + } +} + /// Periodically persist the close group until cancelled. async fn periodic_close_group_cache_save( dht_manager: Arc, @@ -2384,6 +2718,8 @@ async fn periodic_close_group_cache_save( k_value: usize, dir: PathBuf, interval: Duration, + snapshot_floor: Arc, + snapshot_floor_at: Arc, shutdown: CancellationToken, ) { let start = tokio::time::Instant::now() + interval; @@ -2405,6 +2741,18 @@ async fn periodic_close_group_cache_save( ).await { warn!("Periodic close group cache save failed: {error}"); } + let now_epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()); + save_routing_snapshot( + &dht_manager, + peer_id, + now_epoch, + &dir, + &snapshot_floor, + &snapshot_floor_at, + "periodic", + ).await; } } } @@ -3703,4 +4051,97 @@ mod tests { "timestamp-only mutation on a signed message must fail signature verification" ); } + + fn snapshot_peer(addr: &str) -> SnapshotPeer { + SnapshotPeer { + peer_id: PeerId::random(), + addresses: vec![addr.parse().expect("valid multiaddr")], + } + } + + #[test] + fn snapshot_dial_sets_skips_self() { + let self_id = PeerId::random(); + let mut peers = vec![snapshot_peer("/ip4/10.0.0.1/udp/9000/quic")]; + peers[0].peer_id = self_id; + let mut seen = HashSet::new(); + + assert!(snapshot_dial_sets(&peers, &self_id, &mut seen).is_empty()); + assert!(seen.is_empty(), "self must not reserve an address"); + } + + #[test] + fn snapshot_dial_sets_skips_addresses_already_queued() { + let self_id = PeerId::random(); + let queued = snapshot_peer("/ip4/10.0.0.2/udp/9000/quic"); + let fresh = snapshot_peer("/ip4/10.0.0.3/udp/9000/quic"); + let mut seen: HashSet = queued + .addresses + .iter() + .filter_map(MultiAddr::dialable_socket_addr) + .collect(); + + let sets = snapshot_dial_sets(&[queued, fresh.clone()], &self_id, &mut seen); + + assert_eq!( + sets.len(), + 1, + "the already-queued peer must not be redialled" + ); + assert_eq!(sets[0].0, fresh.peer_id); + assert_eq!( + seen.len(), + 2, + "the new address is reserved for later phases" + ); + } + + #[test] + fn snapshot_dial_sets_drops_undialable_addresses() { + let self_id = PeerId::random(); + let peers = vec![snapshot_peer("/ip4/10.0.0.4/tcp/9000")]; + let mut seen = HashSet::new(); + + assert!( + snapshot_dial_sets(&peers, &self_id, &mut seen).is_empty(), + "only QUIC addresses are dialable today" + ); + } + + #[test] + fn snapshot_dial_sets_dedupes_repeated_peers_and_bounds_addresses() { + let self_id = PeerId::random(); + let mut peer = snapshot_peer("/ip4/10.0.0.5/udp/9000/quic"); + peer.addresses = (0..6) + .map(|i| { + format!("/ip4/10.0.0.5/udp/900{i}/quic") + .parse() + .expect("valid multiaddr") + }) + .collect(); + let mut seen = HashSet::new(); + + let sets = snapshot_dial_sets(&[peer.clone(), peer], &self_id, &mut seen); + + assert_eq!(sets.len(), 1, "the same peer twice is one candidate"); + assert_eq!( + sets[0].1.len(), + MAX_SNAPSHOT_ADDRESSES_DIALLED, + "the dial tail past the budget stays bounded" + ); + } + + #[test] + fn snapshot_dial_sets_returns_every_other_peer_once() { + let self_id = PeerId::random(); + let peers: Vec = (0..130) + .map(|i| snapshot_peer(&format!("/ip4/10.1.{}.{}/udp/9000/quic", i / 256, i % 256))) + .collect(); + let mut seen = HashSet::new(); + + let sets = snapshot_dial_sets(&peers, &self_id, &mut seen); + + assert_eq!(sets.len(), 130); + assert_eq!(seen.len(), 130); + } }