From b67f94da2ce846e784e45999a2650f0988d7aee7 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:12:32 +0100 Subject: [PATCH 1/6] fix(replication): measure the paid near range to the 60th peer A node's near range now reaches its 60th closest routing-table peer with no margin, instead of its 20th with two extra bits. It covers about the same number of nodes per key, and reads local peer density from more peers. With that range and evenly spread IDs, a simulation of 500 to 10,000 nodes found no key this node or a neighbour needs outside it, so a sync hint build now reads only near entries at any list size. The 100,000-entry limit is gone. A node knowing fewer than 60 peers still counts every key as near. Co-Authored-By: Claude Opus 5.5 --- docs/REPLICATION_DESIGN.md | 2 +- docs/adr/ADR-0028-pressure-gated-retention.md | 31 +++-- src/replication/config.rs | 23 ++-- src/replication/neighbor_sync.rs | 119 ++++-------------- src/replication/paid_list.rs | 98 +++++++++++---- src/replication/pruning.rs | 2 +- tests/e2e/replication.rs | 14 +-- 7 files changed, 137 insertions(+), 152 deletions(-) diff --git a/docs/REPLICATION_DESIGN.md b/docs/REPLICATION_DESIGN.md index 3aaf117c..d0e68112 100644 --- a/docs/REPLICATION_DESIGN.md +++ b/docs/REPLICATION_DESIGN.md @@ -167,7 +167,7 @@ Rules: 13. Only admitted unknown replica keys that pass presence quorum or paid-list authorization are queued for fetch. 14. Receiver processes unknown paid hints via Section 7.2 majority checks in a paid-list pipeline: successful checks may update `PaidForList(self)`. If the receiver is also storage-responsible for `K`, successful verification may queue record fetch using verified `Present` sources from the same round. If the same key is also present in replica hints, rule 10 drops the paid-hint duplicate and fetch behavior is governed by the replica-hint pipeline. 15. Sync payloads MUST NOT include PoP material; PoP remains fresh-replication-only. -16. Nodes SHOULD use ongoing neighbor sync rounds to re-announce paid hints for locally paid keys to improve paid-list convergence. Above `PAID_HINT_FULL_SCAN_ENTRIES` entries (ADR-0028), a sender reads only the entries near self for these hints, and can miss a key far from self. +16. Nodes SHOULD use ongoing neighbor sync rounds to re-announce paid hints for locally paid keys to improve paid-list convergence. A sender reads only the entries near self for these hints (ADR-0028): those within its `PAID_LIST_NEAR_RANK`th closest peer's distance, or every entry until it knows that many peers. 17. After each round, node sets `NeighborSyncCursor(self)` to the position after the last scanned peer in the (possibly shrunk) snapshot. Peers removed during scanning (cooldown or unreachable) do not occupy cursor positions — the cursor reflects the snapshot's state after removals. Priority syncs do not advance the cursor unless the peer was also removed from the snapshot. 18. When `PriorityNeighborSyncOrder(self)` is empty and `NeighborSyncCursor(self) >= |NeighborSyncOrder(self)|`, the cycle is complete (`NeighborSyncCycleComplete(self)`). Node MUST execute post-cycle responsibility pruning (Section 11), then recompute `CloseNeighbors(self)` from current `LocalRT(self)`, take a fresh snapshot, and reset the cursor to `0` to begin the next cycle. diff --git a/docs/adr/ADR-0028-pressure-gated-retention.md b/docs/adr/ADR-0028-pressure-gated-retention.md index 888dc49b..cd600bc9 100644 --- a/docs/adr/ADR-0028-pressure-gated-retention.md +++ b/docs/adr/ADR-0028-pressure-gated-retention.md @@ -177,16 +177,19 @@ store shares that partition. - **Above it:** each prune pass evicts down to the threshold, at most 100,000 entries, farthest from this node first. Eviction is local, with no confirmations. - **What stays even then:** entries near this node, and entries of records it still - stores. Near means within four to eight times the distance to its farthest paid-group - peer, where it or a neighbour may still have a vote. While it knows too few peers to - say, every entry counts as near. - - **Sync hints:** read every entry up to 100,000 entries, and only the entries near this - node above that, so their cost stops growing with the list. Each entry read costs a - routing-table lookup: measured at 4.7 µs with 4 peers and 14.5 µs with 24, which - extrapolates to about 100 µs with 200. A node builds hints every round and for every - inbound sync. Where few peers are known near a key far from this node, this node or a - neighbour can still be in its paid close group, so above that size such a hint can be - missed. With evenly spread IDs, a simulation of 100 to 2,000 nodes found none. + stores. A key is near when its distance has no more significant bits than the + distance to this node's 60th closest routing-table peer (`PAID_LIST_NEAR_RANK`). With + evenly spread IDs and at most 20 peers per ring, about 120 nodes count a given key as + near, from a simulation of 4,000 nodes. While a node knows fewer than 60 peers, every + entry counts as near. + - **Sync hints:** read only the entries near this node, so their cost doesn't grow with + entries outside that range. Each entry read costs a routing-table lookup: measured at + 4.7 µs with 4 peers and 14.5 µs with 24, which extrapolates to about 100 µs with 200. + A node builds hints every round and for every inbound sync. With evenly spread IDs, a + simulation of 500 to 10,000 nodes found no key outside the near range that this node + or one of its 20 closest neighbours is among the closest 20 or 40 for, in about 48,000 + sampled keys. A node knowing fewer than 60 peers reads every entry, which covers small + networks where a neighbour can be in the paid close group of a key far from this node. - **Store size:** the LMDB map grows from 256 MiB to 1 GiB, which holds about 17.6 million entries, measured at about 61 bytes each with single-key commits. The file grows as entries do, on every platform. A full store logs one error, since new paid @@ -289,14 +292,15 @@ store shares that partition. - a clock surviving "in range" during bootstrap; - paid-list eviction: farthest first, up to its limit, never what it keeps, never near; - near ranges, and reading only the keys in one; - - a far key read for hints while the list can be read whole, and only near keys above that; + - a far key read for hints until the node knows 60 peers, and only near keys after that; - the store's file not taking the whole map up front; - removing a paid entry leaving its record's clock; - a pointer needing all but one of its group to prove it. - **End-to-end tests:** - `out_of_range_records_stay_until_the_disk_needs_the_room`: with room, a record the whole group holds is kept and its group never asked; under pressure it goes. - - `a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node`: a small list is read whole. + - `a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node`: in a small network, a peer + far from a node still gets the hint for a key it is closest to. - `paid_entries_stay_below_the_threshold_and_near_this_node`: disk pressure evicts no paid entry, and an entry near this node stays over the threshold. - The existing chunk prune tests run under a simulated nearly full disk. @@ -308,7 +312,8 @@ store shares that partition. - it stops at 15% free; - below the threshold, `paid ... evicted=0` holds on every node; - with a test build's low threshold, only far entries without a stored record go; - - hint-build times stay flat as `paid total=` grows past 100,000; + - on a network large enough for routing tables to pass 60 peers, hint-build times follow + the entries near a node rather than `paid total=`; - how fast a node under pressure erases, against how fast it receives in-range data; - after restarting a node under pressure, `candidates=` doesn't fall to 0 once bootstrap has drained; diff --git a/src/replication/config.rs b/src/replication/config.rs index 8c2e1da9..bc0ab06f 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -101,21 +101,14 @@ const _: () = assert!( /// Most paid-list entries one prune pass evicts. pub const MAX_PAID_EVICTIONS_PER_PASS: usize = 100_000; -/// Bits beyond the distance to this node's farthest paid-close-group peer within -/// which a key still counts as near this node. -/// -/// Two bits is at least four times that distance, which covers the difference -/// in peer density between here and the key. -pub const PAID_LIST_NEAR_MARGIN_BITS: usize = 2; - -/// Paid-list entries up to which a sync hint build reads every entry. -/// -/// A neighbour can need a hint for a key far from this node, where few peers are -/// known, so reading only near keys can miss one. But every entry read costs a -/// routing table lookup, entries are no longer pruned by rank, and a node builds -/// hints every round and for every inbound sync. So above this size a build reads -/// only near keys (ADR-0028). -pub const PAID_HINT_FULL_SCAN_ENTRIES: u64 = 100_000; +/// Which routing-table peer, counting out from this node, sets how far its near +/// range reaches. +/// +/// A key is near when its distance has no more significant bits than the distance +/// to this peer. Reading this far out gives a steadier measure of local peer +/// density than the closest few peers do, and reaches past every key this node or +/// a neighbour is in the paid close group of when IDs are evenly spread (ADR-0028). +pub const PAID_LIST_NEAR_RANK: usize = 60; /// Number of closest peers to self eligible for neighbor sync. pub const NEIGHBOR_SYNC_SCOPE: usize = 20; diff --git a/src/replication/neighbor_sync.rs b/src/replication/neighbor_sync.rs index 550c9b35..f6575e1a 100644 --- a/src/replication/neighbor_sync.rs +++ b/src/replication/neighbor_sync.rs @@ -15,10 +15,8 @@ use saorsa_core::P2PNode; use crate::ant_protocol::XorName; use crate::client::xor_distance; use crate::error::Result; -use crate::replication::config::{ - ReplicationConfig, PAID_HINT_FULL_SCAN_ENTRIES, REPLICATION_PROTOCOL_ID, -}; -use crate::replication::paid_list::{near_bits_for_radius, PaidList, XORNAME_BITS}; +use crate::replication::config::{ReplicationConfig, PAID_LIST_NEAR_RANK, REPLICATION_PROTOCOL_ID}; +use crate::replication::paid_list::{near_bits_for_peer_distances, PaidList}; use crate::replication::protocol::{ NeighborSyncRequest, NeighborSyncResponse, ReplicationMessage, ReplicationMessageBody, }; @@ -108,52 +106,32 @@ pub(crate) async fn build_replica_hints_for_peer_with_close_groups( } /// How many low bits of a key may differ from this node's for the key to count -/// as near it, from the distance to its farthest paid-close-group peer. Every -/// key counts as near until this node knows a full paid close group. +/// as near it, from the distance to its [`PAID_LIST_NEAR_RANK`]th closest peer. /// /// With evenly spread IDs, entries beyond this carry no vote and need no hint. -/// Where few peers are known near a key, this node or a neighbour can still be in -/// its paid close group. -pub(crate) async fn paid_near_bits(p2p_node: &Arc, group_size: usize) -> usize { +pub(crate) async fn paid_near_bits(p2p_node: &Arc) -> usize { let self_xor: XorName = *p2p_node.peer_id().as_bytes(); - let closest = p2p_node + let distances: Vec = p2p_node .dht_manager() - .find_closest_nodes_local(&self_xor, group_size) - .await; - if closest.len() < group_size { - return XORNAME_BITS; - } - closest + .find_closest_nodes_local(&self_xor, PAID_LIST_NEAR_RANK) + .await .iter() .map(|node| xor_distance(&self_xor, node.peer_id.as_bytes())) - .max() - .map_or(XORNAME_BITS, |radius| near_bits_for_radius(&radius)) -} - -/// How many low bits of a key may differ from this node's for a hint build to read -/// it: all of them while the list is small enough to read whole. -fn paid_hint_scan_bits(entries: u64, near_bits: usize) -> usize { - if entries <= PAID_HINT_FULL_SCAN_ENTRIES { - XORNAME_BITS - } else { - near_bits - } + .collect(); + near_bits_for_peer_distances(&distances) } -/// The paid-list keys a hint build considers. +/// The paid-list keys a hint build considers: those near this node. async fn paid_hint_candidates( paid_list: &PaidList, p2p_node: &Arc, - group_size: usize, ) -> Result> { let self_xor: XorName = *p2p_node.peer_id().as_bytes(); - let near_bits = paid_near_bits(p2p_node, group_size).await; - let scan_bits = paid_hint_scan_bits(paid_list.count()?, near_bits); - paid_list.keys_near(&self_xor, scan_bits) + paid_list.keys_near(&self_xor, paid_near_bits(p2p_node).await) } /// Build outbound hint sets for a batch of peers with one scan over local -/// storage and one scan over the paid-list entries it can afford to read. +/// storage and one scan over the paid-list entries near this node. pub(crate) async fn build_sync_hints_for_peers( peers: &[PeerId], storage: &Arc, @@ -199,14 +177,13 @@ pub(crate) async fn build_sync_hints_for_peers( } } - let paid_keys = - match paid_hint_candidates(paid_list, p2p_node, paid_list_close_group_size).await { - Ok(keys) => keys, - Err(e) => { - warn!("Failed to read PaidForList for batch hint construction: {e}"); - Vec::new() - } - }; + let paid_keys = match paid_hint_candidates(paid_list, p2p_node).await { + Ok(keys) => keys, + Err(e) => { + warn!("Failed to read PaidForList for batch hint construction: {e}"); + Vec::new() + } + }; let paid_key_count = paid_keys.len(); for key in paid_keys { @@ -267,14 +244,13 @@ pub async fn build_paid_hints_for_peer( p2p_node: &Arc, paid_list_close_group_size: usize, ) -> Vec { - let paid_keys = - match paid_hint_candidates(paid_list, p2p_node, paid_list_close_group_size).await { - Ok(keys) => keys, - Err(e) => { - warn!("Failed to read PaidForList for hint construction: {e}"); - return Vec::new(); - } - }; + let paid_keys = match paid_hint_candidates(paid_list, p2p_node).await { + Ok(keys) => keys, + Err(e) => { + warn!("Failed to read PaidForList for hint construction: {e}"); + return Vec::new(); + } + }; let dht = p2p_node.dht_manager(); let mut hints = Vec::new(); @@ -618,49 +594,6 @@ mod tests { PeerId::from_bytes(bytes) } - /// A key whose last byte is `b`, the rest zero. - fn key_ending_in(b: u8) -> XorName { - let mut key = [0u8; 32]; - key[31] = b; - key - } - - // -- paid hint candidates ------------------------------------------------ - - #[tokio::test] - async fn a_far_key_is_read_for_hints_until_the_list_is_too_large_to_read_whole() { - // This node is 0, its neighbours are 1 and 2, and a group is two, so its - // group reaches distance 2. Key 129 is far from all three, yet its two - // closest are neighbour 1 (distance 128) and this node (129). - const THIS_NODE: u8 = 0; - const GROUP_RADIUS: u8 = 2; - const FAR_KEY: u8 = 129; - let dir = tempfile::tempdir().expect("tempdir"); - let paid_list = PaidList::new(dir.path()).await.expect("paid list"); - paid_list - .insert(&key_ending_in(FAR_KEY)) - .await - .expect("insert"); - let near_bits = near_bits_for_radius(&key_ending_in(GROUP_RADIUS)); - let read = |entries| { - paid_list - .keys_near( - &key_ending_in(THIS_NODE), - paid_hint_scan_bits(entries, near_bits), - ) - .expect("keys") - }; - - assert_eq!( - read(PAID_HINT_FULL_SCAN_ENTRIES), - vec![key_ending_in(FAR_KEY)] - ); - assert!( - read(PAID_HINT_FULL_SCAN_ENTRIES + 1).is_empty(), - "above the limit only near keys are read" - ); - } - // -- select_sync_batch --------------------------------------------------- #[test] diff --git a/src/replication/paid_list.rs b/src/replication/paid_list.rs index 827def6d..bc6c7948 100644 --- a/src/replication/paid_list.rs +++ b/src/replication/paid_list.rs @@ -20,7 +20,7 @@ use crate::ant_protocol::XorName; use crate::error::{Error, Result}; use crate::logging::{debug, error, trace, warn}; -use crate::replication::config::{PAID_LIST_MAP_SIZE, PAID_LIST_NEAR_MARGIN_BITS}; +use crate::replication::config::{PAID_LIST_MAP_SIZE, PAID_LIST_NEAR_RANK}; use crate::replication::out_of_range_clocks::OutOfRangeClocks; use heed::types::Bytes; use heed::{Database, Env, EnvOpenOptions, MdbError, RoTxn}; @@ -43,22 +43,31 @@ pub const XORNAME_BITS: usize = XORNAME_LEN * BITS_PER_BYTE; const RECORD_CLOCKS_FILE: &str = "record_out_of_range.bin"; /// How many low bits of a key may differ from this node's for the key to count -/// as near it, given the distance to its farthest paid-close-group peer. +/// as near it, given its distances to the closest peers it knows. /// -/// The margin covers the usual difference in peer density between here and the -/// key. With evenly spread IDs, neither this node nor any of its neighbours is in -/// the paid close group of a key beyond it. Where few peers are known near a key, -/// one of them still can be. +/// Every key counts as near until it knows [`PAID_LIST_NEAR_RANK`] peers: in a +/// network that small, this node or a neighbour can be in the paid close group of +/// any key. #[must_use] -pub fn near_bits_for_radius(radius: &XorName) -> usize { - let significant = radius +pub fn near_bits_for_peer_distances(distances: &[XorName]) -> usize { + if distances.len() < PAID_LIST_NEAR_RANK { + return XORNAME_BITS; + } + distances + .iter() + .max() + .map_or(XORNAME_BITS, significant_bits) +} + +/// How many bits of `distance` count, from its highest set bit down. +fn significant_bits(distance: &XorName) -> usize { + distance .iter() .enumerate() .find(|(_, byte)| **byte != 0) .map_or(0, |(index, byte)| { (XORNAME_LEN - index) * BITS_PER_BYTE - byte.leading_zeros() as usize - }); - (significant + PAID_LIST_NEAR_MARGIN_BITS).min(XORNAME_BITS) + }) } /// Mutable access to bit `bit` of `key`, counted from the least significant: @@ -544,24 +553,69 @@ mod tests { key } + fn key_ending_in(last: u8) -> XorName { + let mut key = [0u8; 32]; + key[31] = last; + key + } + + /// As many distances as a near range needs, all very small, then `farthest`. + fn distances_out_to(farthest: XorName) -> Vec { + let mut distances = vec![key_ending_in(1); PAID_LIST_NEAR_RANK - 1]; + distances.push(farthest); + distances + } + #[test] - fn near_bits_follow_the_radius_plus_the_margin() { - let mut radius = [0u8; 32]; - assert_eq!(near_bits_for_radius(&radius), PAID_LIST_NEAR_MARGIN_BITS); + fn significant_bits_run_from_the_highest_set_bit() { + let mut distance = [0u8; 32]; + assert_eq!(significant_bits(&distance), 0); - radius[31] = 0b0000_0001; - assert_eq!( - near_bits_for_radius(&radius), - 1 + PAID_LIST_NEAR_MARGIN_BITS - ); + distance[31] = 0b0000_0001; + assert_eq!(significant_bits(&distance), 1); + + distance[30] = 0b0000_0100; + assert_eq!(significant_bits(&distance), 11); + + assert_eq!(significant_bits(&[0xFF; 32]), XORNAME_BITS); + } + + #[test] + fn every_key_is_near_until_enough_peers_are_known() { + assert_eq!(near_bits_for_peer_distances(&[]), XORNAME_BITS); + + let one_short = vec![key_ending_in(1); PAID_LIST_NEAR_RANK - 1]; + assert_eq!(near_bits_for_peer_distances(&one_short), XORNAME_BITS); + } + + #[test] + fn the_near_range_reaches_the_farthest_counted_peer_however_close_the_rest() { + const FARTHEST_BITS: usize = 7; + let distances = distances_out_to(key_ending_in(0b0100_0000)); + assert_eq!(near_bits_for_peer_distances(&distances), FARTHEST_BITS); + } + + #[tokio::test] + async fn a_far_key_is_read_until_enough_peers_are_known() { + // This node is 0 and its neighbours are 1 and 2. Key 129 is far from all + // three, yet with a group of two its closest are neighbour 1 and this node. + const FAR_KEY: u8 = 129; + let (pl, _temp) = create_test_paid_list().await; + pl.insert(&key_ending_in(FAR_KEY)).await.expect("insert"); + let this_node = key_ending_in(0); - radius[30] = 0b0000_0100; + let neighbours = [key_ending_in(1), key_ending_in(2)]; + let few_known = near_bits_for_peer_distances(&neighbours); assert_eq!( - near_bits_for_radius(&radius), - 11 + PAID_LIST_NEAR_MARGIN_BITS + pl.keys_near(&this_node, few_known).expect("keys"), + vec![key_ending_in(FAR_KEY)] ); - assert_eq!(near_bits_for_radius(&[0xFF; 32]), XORNAME_BITS); + let many_known = near_bits_for_peer_distances(&distances_out_to(key_ending_in(2))); + assert!(pl + .keys_near(&this_node, many_known) + .expect("keys") + .is_empty()); } #[test] diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 729d8f3e..eb804e93 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -701,7 +701,7 @@ async fn evict_excess_paid_entries(ctx: &PrunePassContext<'_>) -> (usize, usize) if excess == 0 { return (count, 0); } - let near_bits = paid_near_bits(ctx.p2p_node, ctx.config.paid_list_close_group_size).await; + let near_bits = paid_near_bits(ctx.p2p_node).await; let storage = Arc::clone(ctx.storage); let evicted = ctx .paid_list diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index 83b16f85..f334721b 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -2025,16 +2025,16 @@ async fn prune_marks_immediately_and_candidacy_waits_for_hysteresis() { harness.teardown().await.expect("teardown"); } -/// A key far from a node can still have that node's neighbour among its closest, -/// so a paid list small enough to read whole is read whole. +/// In a small network, a node still sends a peer the hint for a key far from the node +/// when that peer is the key's closest. `PaidList`'s unit tests cover the near range +/// itself: a network this small can't set one. #[tokio::test(flavor = "multi_thread")] #[serial] async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { const ONLY_THE_CLOSEST: usize = 1; const EVERY_KNOWN_PEER: usize = 100; const LAST_KEY_BYTE: usize = 31; - // A full network: with only a few peers, every key counts as near this node. - let harness = TestHarness::setup().await.expect("setup"); + let harness = TestHarness::setup_minimal().await.expect("setup"); harness.warmup_dht().await.expect("warmup"); let node = harness.test_node(0).expect("node"); @@ -2044,10 +2044,10 @@ async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { .dht_manager() .find_closest_nodes_local(&self_xor, EVERY_KNOWN_PEER) .await; + assert!(peers.len() < ant_node::replication::config::PAID_LIST_NEAR_RANK); let farthest = peers.last().expect("a known peer").peer_id; - // Beside the farthest peer, so that peer alone is its closest, well outside - // this node's distance to its own closest peer. + // Beside the farthest peer, so that peer alone is its closest. let mut key: XorName = *farthest.as_bytes(); key[LAST_KEY_BYTE] ^= 1; let paid_dir = tempfile::tempdir().expect("tempdir"); @@ -2073,7 +2073,7 @@ async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { /// Paid entries are evicted only once the list passes its threshold, and disk /// pressure alone never evicts one. Even over the threshold, an entry near this /// node stays. `PaidList`'s unit tests cover evicting far entries: in a network -/// this small, a node's near range is most of the key space. +/// this small, every key counts as near. #[tokio::test] #[serial] async fn paid_entries_stay_below_the_threshold_and_near_this_node() { From 8baf7e3abe4726c7a78bffb1c82d657c3f0f1abd Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:32:22 +0100 Subject: [PATCH 2/6] docs(adr): ADR-0028 measured hint lookup cost in a release build MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 100 µs figure was extrapolated from test builds, which this repo compiles at opt-level 1. Replaces it with a release-build measurement, and names the hardware it was taken on. Co-Authored-By: Claude Opus 5.5 --- docs/adr/ADR-0028-pressure-gated-retention.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/adr/ADR-0028-pressure-gated-retention.md b/docs/adr/ADR-0028-pressure-gated-retention.md index cd600bc9..49881820 100644 --- a/docs/adr/ADR-0028-pressure-gated-retention.md +++ b/docs/adr/ADR-0028-pressure-gated-retention.md @@ -183,8 +183,10 @@ store shares that partition. near, from a simulation of 4,000 nodes. While a node knows fewer than 60 peers, every entry counts as near. - **Sync hints:** read only the entries near this node, so their cost doesn't grow with - entries outside that range. Each entry read costs a routing-table lookup: measured at - 4.7 µs with 4 peers and 14.5 µs with 24, which extrapolates to about 100 µs with 200. + entries outside that range. Each entry read costs a local routing-table lookup. In a + release build on an Apple M3 Ultra core, with 101 to 107 peers known, one took 10–11 µs + for a group of 7 and 15–16 µs for a group of 20, so 150,000 took about 2 s. Typical + node hardware is slower, roughly in proportion to its single-core speed. A node builds hints every round and for every inbound sync. With evenly spread IDs, a simulation of 500 to 10,000 nodes found no key outside the near range that this node or one of its 20 closest neighbours is among the closest 20 or 40 for, in about 48,000 From 3e7caa6c0de6bb23ed66d271ca5dea0b33c65067 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:43:56 +0100 Subject: [PATCH 3/6] perf(replication): build sync hints from close groups per key range A hint build looked up every stored record and every paid entry it read in the routing table, one key at a time, and each lookup clones and sorts the whole table. A build now takes one copy of the routing table and works out each key's close group once per key range, then reads only the ranges where the peer it builds for can be in the group, from the record store and the paid list. Each key gets the group a lookup for it would give, so the hints don't change. Because each key's group is exact, a peer gets the paid hint for a key however far the key is from this node, so hints no longer read only near entries. The near range now only decides eviction, and its helper moves to pruning. Inbound syncs use the same batch builder as outbound ones, so they take one routing-table copy instead of two and are logged. The log line reports candidates=, stored_keys_read= and paid_keys_read=. PaidList::keys_near is replaced by a crate-private keys_in_ranges, and ChunkStore gains a crate-private keys_in_ranges. Co-Authored-By: Claude Opus 5.5 --- docs/REPLICATION_DESIGN.md | 3 +- docs/adr/ADR-0028-pressure-gated-retention.md | 55 ++- src/replication/close_group_map.rs | 452 ++++++++++++++++++ src/replication/config.rs | 7 +- src/replication/mod.rs | 1 + src/replication/neighbor_sync.rs | 313 +++++++----- src/replication/paid_list.rs | 85 ++-- src/replication/pruning.rs | 21 +- src/storage/chunk_store.rs | 49 ++ tests/e2e/replication.rs | 114 ++++- 10 files changed, 911 insertions(+), 189 deletions(-) create mode 100644 src/replication/close_group_map.rs diff --git a/docs/REPLICATION_DESIGN.md b/docs/REPLICATION_DESIGN.md index d0e68112..c2800252 100644 --- a/docs/REPLICATION_DESIGN.md +++ b/docs/REPLICATION_DESIGN.md @@ -160,6 +160,7 @@ Rules: 8. In each session, sender-side hint construction uses peer-targeted sets: - `ReplicaHintsForPeer`: keys the sender believes the receiver should hold (receiver is among the `CLOSE_GROUP_SIZE` nearest to `K` in sender's `SelfInclusiveRT`). - `PaidHintsForPeer`: keys the sender believes the receiver should track in `PaidForList` (receiver is among the `PAID_LIST_CLOSE_GROUP_SIZE` nearest to `K` in sender's `SelfInclusiveRT`). + - A sender MAY evaluate both from one `SelfInclusiveRT` snapshot per build, once per key range rather than once per key, reading only the ranges where the receiver can be in the group (ADR-0028). The sets MUST be the ones a per-key evaluation over that snapshot gives. 9. Transport-level chunking/fragmentation is implementation detail and out of scope for replication logic. 10. Receiver treats hint sets as unordered collections and deduplicates repeated keys. If a key appears in both `ReplicaHintsForPeer` and `PaidHintsForPeer` in the same session, receiver MUST keep only the replica-hint entry and drop the paid-hint duplicate (single-pipeline processing). 11. Receiver diffs replica hints against local store and pending sets, then runs per-key admission rules before quorum logic. @@ -167,7 +168,7 @@ Rules: 13. Only admitted unknown replica keys that pass presence quorum or paid-list authorization are queued for fetch. 14. Receiver processes unknown paid hints via Section 7.2 majority checks in a paid-list pipeline: successful checks may update `PaidForList(self)`. If the receiver is also storage-responsible for `K`, successful verification may queue record fetch using verified `Present` sources from the same round. If the same key is also present in replica hints, rule 10 drops the paid-hint duplicate and fetch behavior is governed by the replica-hint pipeline. 15. Sync payloads MUST NOT include PoP material; PoP remains fresh-replication-only. -16. Nodes SHOULD use ongoing neighbor sync rounds to re-announce paid hints for locally paid keys to improve paid-list convergence. A sender reads only the entries near self for these hints (ADR-0028): those within its `PAID_LIST_NEAR_RANK`th closest peer's distance, or every entry until it knows that many peers. +16. Nodes SHOULD use ongoing neighbor sync rounds to re-announce paid hints for locally paid keys to improve paid-list convergence. A sender includes every local paid key whose `PaidCloseGroup` holds the receiver, however far the key is from self. 17. After each round, node sets `NeighborSyncCursor(self)` to the position after the last scanned peer in the (possibly shrunk) snapshot. Peers removed during scanning (cooldown or unreachable) do not occupy cursor positions — the cursor reflects the snapshot's state after removals. Priority syncs do not advance the cursor unless the peer was also removed from the snapshot. 18. When `PriorityNeighborSyncOrder(self)` is empty and `NeighborSyncCursor(self) >= |NeighborSyncOrder(self)|`, the cycle is complete (`NeighborSyncCycleComplete(self)`). Node MUST execute post-cycle responsibility pruning (Section 11), then recompute `CloseNeighbors(self)` from current `LocalRT(self)`, take a fresh snapshot, and reset the cursor to `0` to begin the next cycle. diff --git a/docs/adr/ADR-0028-pressure-gated-retention.md b/docs/adr/ADR-0028-pressure-gated-retention.md index 49881820..27b78ee0 100644 --- a/docs/adr/ADR-0028-pressure-gated-retention.md +++ b/docs/adr/ADR-0028-pressure-gated-retention.md @@ -182,16 +182,26 @@ store shares that partition. evenly spread IDs and at most 20 peers per ring, about 120 nodes count a given key as near, from a simulation of 4,000 nodes. While a node knows fewer than 60 peers, every entry counts as near. - - **Sync hints:** read only the entries near this node, so their cost doesn't grow with - entries outside that range. Each entry read costs a local routing-table lookup. In a - release build on an Apple M3 Ultra core, with 101 to 107 peers known, one took 10–11 µs - for a group of 7 and 15–16 µs for a group of 20, so 150,000 took about 2 s. Typical - node hardware is slower, roughly in proportion to its single-core speed. - A node builds hints every round and for every inbound sync. With evenly spread IDs, a - simulation of 500 to 10,000 nodes found no key outside the near range that this node - or one of its 20 closest neighbours is among the closest 20 or 40 for, in about 48,000 - sampled keys. A node knowing fewer than 60 peers reads every entry, which covers small - networks where a neighbour can be in the paid close group of a key far from this node. + - **Sync hints:** each build takes one copy of the routing table and works out each + key's close group once per key range, rather than once per key + (`src/replication/close_group_map.rs`). It then reads only the ranges where a peer it + builds for can be in the group, from the record store and the paid list, which are + both kept sorted by key. Each key gets the group a lookup for it would give, so a peer + gets the hint for a key however far the key is from this node. The near range plays no + part in hints. + - **Why:** before this, every stored record and every paid entry read cost a local + routing-table lookup, in every build. In a release build on an Apple M3 Ultra core, + with 101 to 107 peers known, one took 10–11 µs for a group of 7 and 15–16 µs for a + group of 20, so 150,000 took about 2 s. Typical node hardware is slower, roughly in + proportion to its single-core speed. A node builds hints every round and for every + inbound sync, a neighbour waits 15 s for an answer, and the records and paid entries + this ADR keeps add to every build. + - **Measured:** in a release build on the same machine, on a 300-node local network + where the node knew 91 peers, building hints for its nearest neighbour over 20,000 + paid entries took 0.9 ms instead of 265–272 ms, and over 20,000 stored records + 0.35 ms instead of 179–195 ms, with the same hints. Those keys were spread over the + whole key space, so the neighbour's ranges held under a tenth of them. A real node's + keys cluster around it, so a build reads a larger share, at well under 1 µs a key. - **Store size:** the LMDB map grows from 256 MiB to 1 GiB, which holds about 17.6 million entries, measured at about 61 bytes each with single-key commits. The file grows as entries do, on every platform. A full store logs one error, since new paid @@ -204,6 +214,7 @@ store shares that partition. - `src/replication/retention.rs`: the pressure state, thresholds and erase order. - `src/replication/out_of_range_clocks.rs`: saving and restoring the clocks. - `ChunkStore::disk_space`: the measurement. +- `src/replication/close_group_map.rs`: close groups per key range, for hint builds. ### Why this holds @@ -265,13 +276,16 @@ store shares that partition. ### Neutral / Operational - **The prune log line gains `disk_pressure=` and `retained=`.** +- **The hint-build log line** reports `candidates=`, `stored_keys_read=` and + `paid_keys_read=` instead of `stored_keys=` and `paid_keys=`, and inbound syncs are + logged too. - **Public Rust API breaks:** - `PrunePassContext` gains `disk_pressure`. - `ReplicationConfig` gains `paid_list_evict_start_entries`. - `PruneResult`'s paid counters become `paid_entries_evicted`, and the prune log line's paid part becomes `paid total= evicted=`. - - `PaidList` loses its paid timers, its prune cursor and `all_keys`, and gains `keys_near` - and `evict_far`. Its record timestamps are `SystemTime`s. + - `PaidList` loses its paid timers, its prune cursor and `all_keys`, and gains + `evict_far`. Its record timestamps are `SystemTime`s. - `PointerReplication::prune_pass` takes the pressure state. - `NeighborSyncState::prune_cursor` is removed. - PR #266's removals are included. @@ -294,15 +308,22 @@ store shares that partition. - a clock surviving "in range" during bootstrap; - paid-list eviction: farthest first, up to its limit, never what it keeps, never near; - near ranges, and reading only the keys in one; - - a far key read for hints until the node knows 60 peers, and only near keys after that; + - a far paid entry kept until the node knows 60 peers; + - close-group maps matching a sort of every candidate by distance, for 0 to 250 + candidates and groups of 1 to 20, with IDs evenly spread, clustered beside one + candidate, or sharing their top bits, and ranges covering every key once; + - only keys whose group holds a peer being hinted; + - reading key ranges from the record store and the paid list; - the store's file not taking the whole map up front; - removing a paid entry leaving its record's clock; - a pointer needing all but one of its group to prove it. - **End-to-end tests:** - `out_of_range_records_stay_until_the_disk_needs_the_room`: with room, a record the whole group holds is kept and its group never asked; under pressure it goes. - - `a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node`: in a small network, a peer - far from a node still gets the hint for a key it is closest to. + - `a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node`: a peer far from a node + still gets the hint for a key it is closest to. + - `hint_builds_match_a_lookup_per_key`: on 25 nodes, every peer gets exactly the paid + and replica hints a lookup per key gives it. - `paid_entries_stay_below_the_threshold_and_near_this_node`: disk pressure evicts no paid entry, and an entry near this node stays over the threshold. - The existing chunk prune tests run under a simulated nearly full disk. @@ -314,8 +335,8 @@ store shares that partition. - it stops at 15% free; - below the threshold, `paid ... evicted=0` holds on every node; - with a test build's low threshold, only far entries without a stored record go; - - on a network large enough for routing tables to pass 60 peers, hint-build times follow - the entries near a node rather than `paid total=`; + - hint builds stay in milliseconds as `paid total=` and retained records grow, from + `Neighbor-sync hint build ... elapsed_ms=`; - how fast a node under pressure erases, against how fast it receives in-range data; - after restarting a node under pressure, `candidates=` doesn't fall to 0 once bootstrap has drained; diff --git a/src/replication/close_group_map.rs b/src/replication/close_group_map.rs new file mode 100644 index 00000000..a85f884c --- /dev/null +++ b/src/replication/close_group_map.rs @@ -0,0 +1,452 @@ +//! Close groups worked out once per key range, instead of once per key. +//! +//! The peers closest to a key change only at a few points in the key space. Of two +//! peers, the one closer to a key is decided by the first bit where the two peers +//! differ: the one matching the key's bit there is closer, whatever the lower bits. +//! So walking down a key's bits from the top, the peers still competing for its +//! group split at each bit into a closer side and a farther side: +//! +//! - if the closer side can fill the group, the farther side is out; +//! - if not, the whole closer side is in, and the rest of the group comes from the +//! farther side. +//! +//! The walk stops once every key under a prefix has the same group, or once the +//! peers still competing don't differ at the next bit. Those few are then compared +//! for each key. A routing table of a few hundred peers gives tens to a few hundred +//! ranges. +//! +//! Each key's group is the one sorting every candidate by distance to it would +//! give: the map applies the same order, once per range. Hint builds use it to read +//! only the ranges where a peer they build for can be in the group. + +use std::collections::HashSet; +use std::ops::Range; + +use saorsa_core::identity::PeerId; + +use crate::ant_protocol::{XorName, XORNAME_LEN}; +use crate::client::xor_distance; + +#[cfg(test)] +use rand::{rngs::StdRng, Rng, SeedableRng}; + +/// Bits in one byte of a key. +const BITS_PER_BYTE: usize = 8; + +/// Bits in a key. +const KEY_BITS: usize = XORNAME_LEN * BITS_PER_BYTE; + +/// A byte's highest bit, which comes first in a key's bit order. +const HIGHEST_BIT_IN_BYTE: u8 = 0x80; + +/// Keys from `low` to `high` whose close groups hold the same settled candidates, +/// plus the closest `need` of the open ones, picked for each key. +#[derive(Debug, Clone)] +pub struct GroupRange { + low: XorName, + high: XorName, + /// Candidates in the group of every key in the range, as indices. + settled: Vec, + /// Candidates still competing, as a run of indices. + open: Range, + /// How many of the open candidates each key's group takes. + need: usize, +} + +impl GroupRange { + /// The range's first and last key. + pub fn bounds(&self) -> (XorName, XorName) { + (self.low, self.high) + } +} + +/// The close group of every key, for one set of candidates and one group size. +#[derive(Debug)] +pub struct CloseGroupMap { + /// Every candidate, sorted by ID. + candidates: Vec, + /// Ranges in key order, covering every key once. + ranges: Vec, +} + +impl CloseGroupMap { + /// Map each key's closest `group_size` of `candidates`. + pub fn new(candidates: impl IntoIterator, group_size: usize) -> Self { + let mut candidates: Vec = candidates.into_iter().collect(); + candidates.sort_unstable_by(|a, b| a.as_bytes().cmp(b.as_bytes())); + candidates.dedup(); + let mut walk = Walk { + candidates: &candidates, + settled: Vec::new(), + ranges: Vec::new(), + }; + walk.split( + [0; XORNAME_LEN], + 0, + 0..candidates.len(), + group_size.min(candidates.len()), + ); + let ranges = walk.ranges; + Self { candidates, ranges } + } + + /// Every range, in key order. + pub fn ranges(&self) -> &[GroupRange] { + &self.ranges + } + + /// Whether any of `peers` is in the group of some key in `range`. + pub fn may_include_any(&self, range: &GroupRange, peers: &HashSet) -> bool { + let open = if range.need == 0 { + &[][..] + } else { + self.candidates.get(range.open.clone()).unwrap_or(&[]) + }; + range + .settled + .iter() + .filter_map(|&index| self.candidates.get(index)) + .chain(open) + .any(|peer| peers.contains(peer)) + } + + /// The close group of `key`, which must lie in `range`. + pub fn group_in(&self, range: &GroupRange, key: &XorName) -> Vec { + let mut group: Vec = range + .settled + .iter() + .filter_map(|&index| self.candidates.get(index)) + .copied() + .collect(); + if range.need == 0 { + return group; + } + let mut open: Vec<(XorName, PeerId)> = self + .candidates + .get(range.open.clone()) + .unwrap_or(&[]) + .iter() + .map(|peer| (xor_distance(key, peer.as_bytes()), *peer)) + .collect(); + if range.need < open.len() { + open.select_nth_unstable_by_key(range.need - 1, |(distance, _)| *distance); + open.truncate(range.need); + } + group.extend(open.into_iter().map(|(_, peer)| peer)); + group + } + + /// The close group of any key. + #[cfg(test)] + pub fn group_of(&self, key: &XorName) -> Vec { + let index = self.ranges.partition_point(|range| range.high < *key); + self.ranges + .get(index) + .map_or_else(Vec::new, |range| self.group_in(range, key)) + } +} + +/// The state of one map's walk down the key bits. +struct Walk<'a> { + candidates: &'a [PeerId], + /// Candidates in the group of every key under the current prefix. + settled: Vec, + ranges: Vec, +} + +impl Walk<'_> { + /// Map the keys sharing the first `depth` bits of `prefix`, whose groups hold + /// the settled candidates and the closest `need` of `open`. Every candidate in + /// `open` shares those bits with the others. + fn split(&mut self, prefix: XorName, depth: usize, open: Range, need: usize) { + if need == 0 || open.len() == need { + let mut settled = self.settled.clone(); + if need > 0 { + settled.extend(open); + } + self.push(prefix, depth, settled, 0..0, 0); + return; + } + // Sorted by ID and sharing every bit above this one, the open candidates + // run from those with this bit clear to those with it set. + let run = self.candidates.get(open.clone()).unwrap_or(&[]); + let set_from = open.start + run.partition_point(|peer| !bit(peer.as_bytes(), depth)); + if depth >= KEY_BITS || set_from == open.start || set_from == open.end { + // No open candidate differs from another at this bit, so a key's bit + // here can't change which of them are closer: each key decides. + let settled = self.settled.clone(); + self.push(prefix, depth, settled, open, need); + return; + } + for key_bit in [false, true] { + let (closer, farther) = if key_bit { + (set_from..open.end, open.start..set_from) + } else { + (open.start..set_from, set_from..open.end) + }; + let child = with_bit(prefix, depth, key_bit); + if closer.len() >= need { + self.split(child, depth + 1, closer, need); + } else { + let mark = self.settled.len(); + let taken = closer.len(); + self.settled.extend(closer); + self.split(child, depth + 1, farther, need - taken); + self.settled.truncate(mark); + } + } + } + + fn push( + &mut self, + prefix: XorName, + depth: usize, + settled: Vec, + open: Range, + need: usize, + ) { + let mut high = prefix; + for free in depth..KEY_BITS { + high = with_bit(high, free, true); + } + self.ranges.push(GroupRange { + low: prefix, + high, + settled, + open, + need, + }); + } +} + +/// Bit `depth` of `id`, counting from its highest bit. +fn bit(id: &XorName, depth: usize) -> bool { + id.get(depth / BITS_PER_BYTE) + .is_some_and(|byte| byte & (HIGHEST_BIT_IN_BYTE >> (depth % BITS_PER_BYTE)) != 0) +} + +/// `id` with bit `depth`, counting from its highest bit, set to `value`. +fn with_bit(mut id: XorName, depth: usize, value: bool) -> XorName { + let mask = HIGHEST_BIT_IN_BYTE >> (depth % BITS_PER_BYTE); + if let Some(byte) = id.get_mut(depth / BITS_PER_BYTE) { + if value { + *byte |= mask; + } else { + *byte &= !mask; + } + } + id +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + + const SEED: u64 = 0x5EED_C105; + const KEYS_PER_CASE: usize = 400; + const RECORD_GROUP: usize = 7; + const PAID_GROUP: usize = 20; + const GROUP_SIZES: [usize; 4] = [1, 2, RECORD_GROUP, PAID_GROUP]; + const MEDIUM_TABLE: usize = 60; + const LARGE_TABLE: usize = 250; + /// Around and at the group sizes, where the edge cases are. + const TABLE_SIZES: [usize; 8] = [ + 0, + 1, + 2, + RECORD_GROUP, + PAID_GROUP, + PAID_GROUP + 1, + MEDIUM_TABLE, + LARGE_TABLE, + ]; + const EXAMPLE_GROUP: usize = 2; + /// Bytes a clustered ID shares with the candidate it sits beside. + const CLUSTERED_SHARED_BYTES: usize = 3; + const CLUSTERED_IDS: usize = 40; + + fn peer(bytes: XorName) -> PeerId { + PeerId::from_bytes(bytes) + } + + /// A key whose first byte is `first` and the rest zero. + fn key_starting(first: u8) -> XorName { + let mut key = [0u8; XORNAME_LEN]; + key[0] = first; + key + } + + /// The closest `group_size` candidates to `key`, by sorting them all. + fn brute_force(candidates: &[PeerId], key: &XorName, group_size: usize) -> Vec { + let mut sorted = candidates.to_vec(); + sorted.sort_by_key(|peer| xor_distance(key, peer.as_bytes())); + sorted.truncate(group_size); + sorted + } + + fn sorted(mut peers: Vec) -> Vec { + peers.sort_unstable_by(|a, b| a.as_bytes().cmp(b.as_bytes())); + peers + } + + /// Keys that test the edges: the ends of the key space, each candidate's own + /// ID and its neighbours one bit away, then random keys. + fn probe_keys(candidates: &[PeerId], rng: &mut StdRng) -> Vec { + let mut keys = vec![[0u8; XORNAME_LEN], [0xFF; XORNAME_LEN]]; + for candidate in candidates { + let id = *candidate.as_bytes(); + keys.push(id); + keys.push(with_bit(id, KEY_BITS - 1, !bit(&id, KEY_BITS - 1))); + keys.push(with_bit(id, 0, !bit(&id, 0))); + } + keys.extend((0..KEYS_PER_CASE).map(|_| rng.gen::())); + keys + } + + fn assert_matches_brute_force(candidates: &[PeerId], group_size: usize, rng: &mut StdRng) { + let map = CloseGroupMap::new(candidates.iter().copied(), group_size); + for key in probe_keys(candidates, rng) { + assert_eq!( + sorted(map.group_of(&key)), + sorted(brute_force(candidates, &key, group_size)), + "{} candidates, group {group_size}, key {}", + candidates.len(), + hex::encode(key) + ); + } + } + + #[test] + fn every_group_matches_sorting_every_candidate_by_distance() { + let mut rng = StdRng::seed_from_u64(SEED); + for table in TABLE_SIZES { + let candidates: Vec = (0..table).map(|_| peer(rng.gen())).collect(); + for group_size in GROUP_SIZES { + assert_matches_brute_force(&candidates, group_size, &mut rng); + } + } + } + + #[test] + fn groups_match_with_ids_clustered_beside_one_candidate() { + let mut rng = StdRng::seed_from_u64(SEED); + let mut candidates: Vec = (0..MEDIUM_TABLE).map(|_| peer(rng.gen())).collect(); + let beside = *candidates[0].as_bytes(); + for _ in 0..CLUSTERED_IDS { + let mut id: XorName = rng.gen(); + id[..CLUSTERED_SHARED_BYTES].copy_from_slice(&beside[..CLUSTERED_SHARED_BYTES]); + candidates.push(peer(id)); + } + for group_size in GROUP_SIZES { + assert_matches_brute_force(&candidates, group_size, &mut rng); + } + } + + #[test] + fn groups_match_when_every_candidate_shares_its_top_bits() { + let mut rng = StdRng::seed_from_u64(SEED); + let candidates: Vec = (0..MEDIUM_TABLE) + .map(|_| { + let mut id: XorName = rng.gen(); + id[0] = 0; + peer(id) + }) + .collect(); + for group_size in GROUP_SIZES { + assert_matches_brute_force(&candidates, group_size, &mut rng); + } + } + + #[test] + fn ranges_cover_every_key_once_in_order() { + let mut rng = StdRng::seed_from_u64(SEED); + let candidates: Vec = (0..LARGE_TABLE).map(|_| peer(rng.gen())).collect(); + for group_size in GROUP_SIZES { + let map = CloseGroupMap::new(candidates.iter().copied(), group_size); + let bounds: Vec<(XorName, XorName)> = + map.ranges().iter().map(GroupRange::bounds).collect(); + assert_eq!(bounds.first().map(|b| b.0), Some([0u8; XORNAME_LEN])); + assert_eq!(bounds.last().map(|b| b.1), Some([0xFF; XORNAME_LEN])); + for pair in bounds.windows(2) { + let (_, high) = pair[0]; + let (low, _) = pair[1]; + assert!(high < low, "ranges overlap or are out of order"); + let mut next = high; + for byte in next.iter_mut().rev() { + let (sum, carried) = byte.overflowing_add(1); + *byte = sum; + if !carried { + break; + } + } + assert_eq!(next, low, "a gap between ranges"); + } + } + } + + #[test] + fn the_worked_example_splits_into_three_ranges() { + // Peers 0000, 0001, 0101, 1000 and 1110 in the top four bits, a group of two. + // For keys starting 01, peer 0101 is in, and 0000 and 0001 differ only in + // their last bit, so each key picks one of them. + let ids = [ + 0b0000_0000, + 0b0001_0000, + 0b0101_0000, + 0b1000_0000, + 0b1110_0000, + ]; + let candidates: Vec = ids.iter().map(|first| peer(key_starting(*first))).collect(); + let map = CloseGroupMap::new(candidates.iter().copied(), EXAMPLE_GROUP); + let starts: Vec = map.ranges().iter().map(|range| range.low[0]).collect(); + assert_eq!(starts, vec![0b0000_0000, 0b0100_0000, 0b1000_0000]); + + let settled = |range: &GroupRange| -> Vec { + sorted(range.settled.iter().map(|&i| map.candidates[i]).collect()) + }; + let ranges = map.ranges(); + assert_eq!( + settled(&ranges[0]), + sorted(vec![candidates[0], candidates[1]]) + ); + assert_eq!(settled(&ranges[1]), vec![candidates[2]]); + assert_eq!(ranges[1].need, 1, "0000 and 0001 are compared for each key"); + assert_eq!( + map.group_of(&key_starting(0b0100_0000)), + vec![candidates[2], candidates[0]] + ); + assert_eq!( + map.group_of(&key_starting(0b0101_0000)), + vec![candidates[2], candidates[1]] + ); + assert_eq!( + settled(&ranges[2]), + sorted(vec![candidates[3], candidates[4]]) + ); + + let neighbour = HashSet::from([candidates[1]]); + let read: Vec = ranges + .iter() + .map(|range| map.may_include_any(range, &neighbour)) + .collect(); + assert_eq!( + read, + vec![true, true, false], + "keys from 1000 on are never read" + ); + } + + #[test] + fn with_no_more_candidates_than_the_group_every_key_gets_them_all() { + let mut rng = StdRng::seed_from_u64(SEED); + let candidates: Vec = (0..RECORD_GROUP).map(|_| peer(rng.gen())).collect(); + let map = CloseGroupMap::new(candidates.iter().copied(), PAID_GROUP); + assert_eq!(map.ranges().len(), 1); + assert_eq!(sorted(map.group_of(&rng.gen())), sorted(candidates)); + + let empty = CloseGroupMap::new(Vec::new(), PAID_GROUP); + assert_eq!(empty.ranges().len(), 1); + assert!(empty.group_of(&rng.gen()).is_empty()); + } +} diff --git a/src/replication/config.rs b/src/replication/config.rs index bc0ab06f..e3abd07f 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -105,9 +105,10 @@ pub const MAX_PAID_EVICTIONS_PER_PASS: usize = 100_000; /// range reaches. /// /// A key is near when its distance has no more significant bits than the distance -/// to this peer. Reading this far out gives a steadier measure of local peer -/// density than the closest few peers do, and reaches past every key this node or -/// a neighbour is in the paid close group of when IDs are evenly spread (ADR-0028). +/// to this peer, and a paid entry near this node is never evicted. Reading this +/// far out gives a steadier measure of local peer density than the closest few +/// peers do, and reaches past every key this node or a neighbour is in the paid +/// close group of when IDs are evenly spread (ADR-0028). pub const PAID_LIST_NEAR_RANK: usize = 60; /// Number of closest peers to self eligible for neighbor sync. diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 3e66f9bd..acd4b940 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -19,6 +19,7 @@ pub mod audit; pub mod audit_coordinator; pub(crate) mod audit_metrics; pub mod bootstrap; +pub(crate) mod close_group_map; pub mod commitment; pub mod commitment_state; pub mod config; diff --git a/src/replication/neighbor_sync.rs b/src/replication/neighbor_sync.rs index f6575e1a..c773d3d7 100644 --- a/src/replication/neighbor_sync.rs +++ b/src/replication/neighbor_sync.rs @@ -13,10 +13,10 @@ use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; use crate::ant_protocol::XorName; -use crate::client::xor_distance; use crate::error::Result; -use crate::replication::config::{ReplicationConfig, PAID_LIST_NEAR_RANK, REPLICATION_PROTOCOL_ID}; -use crate::replication::paid_list::{near_bits_for_peer_distances, PaidList}; +use crate::replication::close_group_map::{CloseGroupMap, GroupRange}; +use crate::replication::config::{ReplicationConfig, REPLICATION_PROTOCOL_ID}; +use crate::replication::paid_list::PaidList; use crate::replication::protocol::{ NeighborSyncRequest, NeighborSyncResponse, ReplicationMessage, ReplicationMessageBody, }; @@ -60,78 +60,122 @@ pub(crate) struct PeerSyncHints { /// Returns keys that we believe the peer should hold (peer is among the /// `CLOSE_GROUP_SIZE` nearest to `K` in our `SelfInclusiveRT`). /// -/// This intentionally scans all locally stored records, not only records for -/// which this node is still responsible. Out-of-range records retained by -/// prune hysteresis therefore continue to repair their new close group before -/// this node is allowed to delete them. +/// This considers every locally stored record, not only those this node is still +/// responsible for: an out-of-range record it keeps still repairs its new close +/// group. Only the key ranges where the peer can be in the group are read. pub async fn build_replica_hints_for_peer( peer: &PeerId, storage: &Arc, p2p_node: &Arc, close_group_size: usize, ) -> Vec { - build_replica_hints_for_peer_with_close_groups(peer, storage, p2p_node, close_group_size) - .await + let targets = HashSet::from([*peer]); + let groups = CloseGroupMap::new(close_group_candidates(p2p_node).await, close_group_size); + let (mut hints, _) = replica_hints(&groups, storage, &targets); + hints + .remove(peer) + .unwrap_or_default() .into_iter() .map(|hint| hint.key) .collect() } -pub(crate) async fn build_replica_hints_for_peer_with_close_groups( - peer: &PeerId, - storage: &Arc, - p2p_node: &Arc, - close_group_size: usize, -) -> Vec { - let all_keys = match storage.all_keys().await { - Ok(keys) => keys, - Err(e) => { - warn!("Failed to read stored keys for hint construction: {e}"); - return Vec::new(); - } - }; +/// Every candidate for a key's close group in this node's view: its routing table +/// and itself. One snapshot serves a whole build. +async fn close_group_candidates(p2p_node: &Arc) -> Vec { + let self_id = *p2p_node.peer_id(); + let mut candidates: Vec = p2p_node + .dht_manager() + .routing_table_peers() + .await + .into_iter() + .map(|node| node.peer_id) + .filter(|peer| *peer != self_id) + .collect(); + candidates.push(self_id); + candidates +} - let dht = p2p_node.dht_manager(); - let mut hints = Vec::new(); - for key in all_keys { - let closest = dht - .find_closest_nodes_local_with_self(&key, close_group_size) - .await; - let close_peers = closest.iter().map(|n| n.peer_id).collect::>(); - if close_peers.contains(peer) { - hints.push(SentReplicaHint { key, close_peers }); +/// Call `hint` with each key `read` returns whose close group in `groups` holds +/// one of `targets`, and with that group. `read` is asked only for the ranges +/// where one of them can be in the group. Returns how many keys were read. +fn for_each_hinted_key( + groups: &CloseGroupMap, + targets: &HashSet, + read: impl FnOnce(&[(XorName, XorName)]) -> Result>>, + mut hint: impl FnMut(XorName, Vec), +) -> Result { + let ranges: Vec<&GroupRange> = groups + .ranges() + .iter() + .filter(|range| groups.may_include_any(range, targets)) + .collect(); + let bounds: Vec<(XorName, XorName)> = ranges.iter().map(|range| range.bounds()).collect(); + let mut keys_read = 0; + for (range, keys) in ranges.into_iter().zip(read(&bounds)?) { + keys_read += keys.len(); + for key in keys { + let group = groups.group_in(range, &key); + if group.iter().any(|peer| targets.contains(peer)) { + hint(key, group); + } } } - hints + Ok(keys_read) } -/// How many low bits of a key may differ from this node's for the key to count -/// as near it, from the distance to its [`PAID_LIST_NEAR_RANK`]th closest peer. -/// -/// With evenly spread IDs, entries beyond this carry no vote and need no hint. -pub(crate) async fn paid_near_bits(p2p_node: &Arc) -> usize { - let self_xor: XorName = *p2p_node.peer_id().as_bytes(); - let distances: Vec = p2p_node - .dht_manager() - .find_closest_nodes_local(&self_xor, PAID_LIST_NEAR_RANK) - .await - .iter() - .map(|node| xor_distance(&self_xor, node.peer_id.as_bytes())) - .collect(); - near_bits_for_peer_distances(&distances) +/// Replica hints from the stored records for each of `targets`, and how many +/// stored keys were read. +fn replica_hints( + groups: &CloseGroupMap, + storage: &ChunkStore, + targets: &HashSet, +) -> (HashMap>, usize) { + let mut hints: HashMap> = HashMap::new(); + let keys_read = for_each_hinted_key( + groups, + targets, + |ranges| Ok(storage.keys_in_ranges(ranges)), + |key, group| { + let close_peers: HashSet = group.into_iter().collect(); + for peer in close_peers.intersection(targets) { + hints.entry(*peer).or_default().push(SentReplicaHint { + key, + close_peers: close_peers.clone(), + }); + } + }, + ) + .unwrap_or_default(); + (hints, keys_read) } -/// The paid-list keys a hint build considers: those near this node. -async fn paid_hint_candidates( +/// Paid hints from the paid list for each of `targets`, and how many entries +/// were read. +fn paid_hints( + groups: &CloseGroupMap, paid_list: &PaidList, - p2p_node: &Arc, -) -> Result> { - let self_xor: XorName = *p2p_node.peer_id().as_bytes(); - paid_list.keys_near(&self_xor, paid_near_bits(p2p_node).await) + targets: &HashSet, +) -> Result<(HashMap>, usize)> { + let mut hints: HashMap> = HashMap::new(); + let keys_read = for_each_hinted_key( + groups, + targets, + |ranges| paid_list.keys_in_ranges(ranges), + |key, group| { + for peer in group.iter().filter(|peer| targets.contains(*peer)) { + hints.entry(*peer).or_default().push(key); + } + }, + )?; + Ok((hints, keys_read)) } -/// Build outbound hint sets for a batch of peers with one scan over local -/// storage and one scan over the paid-list entries near this node. +/// Build hint sets for a batch of peers from one routing-table snapshot. +/// +/// Each store is read only in the key ranges where one of the peers can be in +/// the close group, and each key's group is the one a lookup for that key would +/// give, so a peer gets the hint for a key however far the key is from this node. pub(crate) async fn build_sync_hints_for_peers( peers: &[PeerId], storage: &Arc, @@ -141,7 +185,7 @@ pub(crate) async fn build_sync_hints_for_peers( paid_list_close_group_size: usize, ) -> HashMap { let started = Instant::now(); - let target_peers = peers.iter().copied().collect::>(); + let targets = peers.iter().copied().collect::>(); let mut hints_by_peer = peers .iter() .copied() @@ -152,52 +196,32 @@ pub(crate) async fn build_sync_hints_for_peers( return hints_by_peer; } - let all_keys = match storage.all_keys().await { - Ok(keys) => keys, - Err(e) => { - warn!("Failed to read stored keys for batch hint construction: {e}"); - Vec::new() - } - }; - let stored_key_count = all_keys.len(); - - let dht = p2p_node.dht_manager(); - for key in all_keys { - let closest = dht - .find_closest_nodes_local_with_self(&key, close_group_size) - .await; - let close_peers = closest.iter().map(|n| n.peer_id).collect::>(); - for peer in close_peers.intersection(&target_peers) { - if let Some(peer_hints) = hints_by_peer.get_mut(peer) { - peer_hints.sent_replica_hints.push(SentReplicaHint { - key, - close_peers: close_peers.clone(), - }); - } + let candidates = close_group_candidates(p2p_node).await; + let candidate_count = candidates.len(); + let record_groups = CloseGroupMap::new(candidates.iter().copied(), close_group_size); + let paid_groups = CloseGroupMap::new(candidates, paid_list_close_group_size); + + let (replica, stored_keys_read) = replica_hints(&record_groups, storage, &targets); + for (peer, hints) in replica { + if let Some(peer_hints) = hints_by_peer.get_mut(&peer) { + peer_hints.sent_replica_hints = hints; } } - let paid_keys = match paid_hint_candidates(paid_list, p2p_node).await { - Ok(keys) => keys, + let paid_keys_read = match paid_hints(&paid_groups, paid_list, &targets) { + Ok((paid, keys_read)) => { + for (peer, hints) in paid { + if let Some(peer_hints) = hints_by_peer.get_mut(&peer) { + peer_hints.paid_hints = hints; + } + } + keys_read + } Err(e) => { warn!("Failed to read PaidForList for batch hint construction: {e}"); - Vec::new() + 0 } }; - let paid_key_count = paid_keys.len(); - - for key in paid_keys { - let closest = dht - .find_closest_nodes_local_with_self(&key, paid_list_close_group_size) - .await; - for node in closest { - if target_peers.contains(&node.peer_id) { - if let Some(peer_hints) = hints_by_peer.get_mut(&node.peer_id) { - peer_hints.paid_hints.push(key); - } - } - } - } let replica_hint_count = hints_by_peer .values() @@ -212,22 +236,14 @@ pub(crate) async fn build_sync_hints_for_peers( if elapsed_ms >= HINT_BUILD_SLOW_LOG_MS { info!( target: "ant_node::replication::neighbor_sync", - "Slow neighbor-sync hint build: peers={}, stored_keys={}, paid_keys={}, replica_hints={}, paid_hints={}, elapsed_ms={elapsed_ms}", + "Slow neighbor-sync hint build: peers={}, candidates={candidate_count}, stored_keys_read={stored_keys_read}, paid_keys_read={paid_keys_read}, replica_hints={replica_hint_count}, paid_hints={paid_hint_count}, elapsed_ms={elapsed_ms}", peers.len(), - stored_key_count, - paid_key_count, - replica_hint_count, - paid_hint_count, ); } else { debug!( target: "ant_node::replication::neighbor_sync", - "Neighbor-sync hint build: peers={}, stored_keys={}, paid_keys={}, replica_hints={}, paid_hints={}, elapsed_ms={elapsed_ms}", + "Neighbor-sync hint build: peers={}, candidates={candidate_count}, stored_keys_read={stored_keys_read}, paid_keys_read={paid_keys_read}, replica_hints={replica_hint_count}, paid_hints={paid_hint_count}, elapsed_ms={elapsed_ms}", peers.len(), - stored_key_count, - paid_key_count, - replica_hint_count, - paid_hint_count, ); } @@ -244,25 +260,18 @@ pub async fn build_paid_hints_for_peer( p2p_node: &Arc, paid_list_close_group_size: usize, ) -> Vec { - let paid_keys = match paid_hint_candidates(paid_list, p2p_node).await { - Ok(keys) => keys, + let targets = HashSet::from([*peer]); + let groups = CloseGroupMap::new( + close_group_candidates(p2p_node).await, + paid_list_close_group_size, + ); + match paid_hints(&groups, paid_list, &targets) { + Ok((mut hints, _)) => hints.remove(peer).unwrap_or_default(), Err(e) => { warn!("Failed to read PaidForList for hint construction: {e}"); - return Vec::new(); - } - }; - - let dht = p2p_node.dht_manager(); - let mut hints = Vec::new(); - for key in paid_keys { - let closest = dht - .find_closest_nodes_local_with_self(&key, paid_list_close_group_size) - .await; - if closest.iter().any(|n| n.peer_id == *peer) { - hints.push(key); + Vec::new() } } - hints } /// Take a fresh snapshot of close neighbors for a new round-robin cycle. @@ -545,24 +554,23 @@ pub(crate) async fn handle_sync_request_with_proofs( let sender_in_rt = p2p_node.dht_manager().is_in_routing_table(sender).await; // Build outbound hints (always sent, even to non-RT peers). - let sent_replica_hints = build_replica_hints_for_peer_with_close_groups( - sender, + let hints = build_sync_hints_for_peers( + std::slice::from_ref(sender), storage, + paid_list, p2p_node, config.close_group_size, + config.paid_list_close_group_size, ) - .await; + .await + .remove(sender) + .unwrap_or_default(); + let sent_replica_hints = hints.sent_replica_hints; let replica_hints = sent_replica_hints .iter() .map(|hint| hint.key) .collect::>(); - let paid_hints = build_paid_hints_for_peer( - sender, - paid_list, - p2p_node, - config.paid_list_close_group_size, - ) - .await; + let paid_hints = hints.paid_hints; let response = NeighborSyncResponse { replica_hints, @@ -594,6 +602,53 @@ mod tests { PeerId::from_bytes(bytes) } + // -- hint builds ---------------------------------------------------------- + + #[test] + fn only_keys_whose_group_holds_a_target_are_hinted() { + // Peers 0000, 0001, 0101, 1000 and 1110 in the top four bits, a group of + // two. Keys starting 01 take 0101 and whichever of 0000 and 0001 is closer. + const GROUP: usize = 2; + let candidates = [ + 0b0000_0000, + 0b0001_0000, + 0b0101_0000, + 0b1000_0000, + 0b1110_0000, + ] + .map(peer_id_from_byte); + let groups = CloseGroupMap::new(candidates, GROUP); + let targets = HashSet::from([candidates[1]]); + let keys = [0b0000_0000, 0b0100_0000, 0b0101_0000, 0b1100_0000] + .map(|first| *peer_id_from_byte(first).as_bytes()); + + let mut hinted = Vec::new(); + let keys_read = for_each_hinted_key( + &groups, + &targets, + |ranges| { + Ok(ranges + .iter() + .map(|(low, high)| { + keys.iter() + .copied() + .filter(|key| low <= key && key <= high) + .collect() + }) + .collect()) + }, + |key, _| hinted.push(key), + ) + .expect("hints"); + + assert_eq!( + hinted, + vec![keys[0], keys[2]], + "0100 goes to 0000, not 0001" + ); + assert_eq!(keys_read, 3, "keys from 1000 on aren't read"); + } + // -- select_sync_batch --------------------------------------------------- #[test] diff --git a/src/replication/paid_list.rs b/src/replication/paid_list.rs index bc6c7948..256187ba 100644 --- a/src/replication/paid_list.rs +++ b/src/replication/paid_list.rs @@ -365,26 +365,33 @@ impl PaidList { Ok(entries as u64) } - /// The keys sharing all but the low `near_bits` bits with `center`. - /// - /// Reads only those keys, however long the list is. [`XORNAME_BITS`] gives - /// every key. + /// The keys in each of `ranges`, from its first key to its last, in ascending + /// order. Reads only those keys, however long the list is. /// /// # Errors /// /// Returns an error if the LMDB read transaction or iteration fails. - pub fn keys_near(&self, center: &XorName, near_bits: usize) -> Result> { + pub(crate) fn keys_in_ranges( + &self, + ranges: &[(XorName, XorName)], + ) -> Result>> { let rtxn = self .env .read_txn() .map_err(|e| Error::Storage(format!("Failed to create read txn: {e}")))?; - let (low, high) = prefix_range(center, near_bits); - let mut keys = Vec::new(); - visit_range(self.db, &rtxn, (&low, &high), |key| { - keys.push(key); - true - })?; - Ok(keys) + ranges + .iter() + .map(|(low, high)| { + let mut keys = Vec::new(); + if low <= high { + visit_range(self.db, &rtxn, (low, high), |key| { + keys.push(key); + true + })?; + } + Ok(keys) + }) + .collect() } /// Remove up to `limit` entries outside the `near_bits` around `center`, @@ -596,26 +603,29 @@ mod tests { } #[tokio::test] - async fn a_far_key_is_read_until_enough_peers_are_known() { + async fn a_far_key_is_kept_until_enough_peers_are_known() { // This node is 0 and its neighbours are 1 and 2. Key 129 is far from all // three, yet with a group of two its closest are neighbour 1 and this node. const FAR_KEY: u8 = 129; + const EVICT_UP_TO: usize = 10; let (pl, _temp) = create_test_paid_list().await; pl.insert(&key_ending_in(FAR_KEY)).await.expect("insert"); let this_node = key_ending_in(0); - let neighbours = [key_ending_in(1), key_ending_in(2)]; - let few_known = near_bits_for_peer_distances(&neighbours); - assert_eq!( - pl.keys_near(&this_node, few_known).expect("keys"), - vec![key_ending_in(FAR_KEY)] - ); + let few_known = near_bits_for_peer_distances(&[key_ending_in(1), key_ending_in(2)]); + let evicted = pl + .evict_far(this_node, few_known, EVICT_UP_TO, |_| false) + .await + .expect("evict"); + assert_eq!(evicted, 0, "too few peers known to call any key far"); let many_known = near_bits_for_peer_distances(&distances_out_to(key_ending_in(2))); - assert!(pl - .keys_near(&this_node, many_known) - .expect("keys") - .is_empty()); + let evicted = pl + .evict_far(this_node, many_known, EVICT_UP_TO, |_| false) + .await + .expect("evict"); + assert_eq!(evicted, 1); + assert!(!pl.contains(&key_ending_in(FAR_KEY)).expect("contains")); } #[test] @@ -647,15 +657,23 @@ mod tests { } #[tokio::test] - async fn keys_near_reads_only_the_keys_around_the_center() { + async fn keys_in_ranges_reads_each_range_with_both_ends_included() { let (pl, _temp) = create_test_paid_list().await; - let center = key_with_first_byte(0x40); - let (near, far) = (key_with_first_byte(0x41), key_with_first_byte(0xC0)); - pl.insert(&near).await.expect("insert"); - pl.insert(&far).await.expect("insert"); + let keys = [0x10, 0x20, 0x30, 0x40, 0xC0].map(key_with_first_byte); + for key in &keys { + pl.insert(key).await.expect("insert"); + } - let near_bits = XORNAME_BITS - 4; - assert_eq!(pl.keys_near(¢er, near_bits).expect("keys"), vec![near]); + let ranges = [ + (key_with_first_byte(0x20), key_with_first_byte(0x30)), + (key_with_first_byte(0x41), key_with_first_byte(0xBF)), + (key_with_first_byte(0xC0), [0xFF; 32]), + (key_with_first_byte(0x30), key_with_first_byte(0x20)), + ]; + assert_eq!( + pl.keys_in_ranges(&ranges).expect("keys"), + vec![vec![keys[1], keys[2]], vec![], vec![keys[4]], vec![]] + ); } #[tokio::test] @@ -810,7 +828,7 @@ mod tests { } #[tokio::test] - async fn keys_near_with_every_bit_free_gives_every_key() { + async fn keys_in_the_whole_key_space_are_every_key() { let (pl, _temp) = create_test_paid_list().await; let key_a: XorName = [0x01; 32]; @@ -821,7 +839,10 @@ mod tests { pl.insert(&key_b).await.expect("insert 2"); pl.insert(&key_c).await.expect("insert 3"); - let mut keys = pl.keys_near(&[0u8; 32], XORNAME_BITS).expect("keys"); + let mut keys = pl + .keys_in_ranges(&[([0u8; 32], [0xFF; 32])]) + .expect("keys") + .concat(); keys.sort_unstable(); let mut expected = vec![key_a, key_b, key_c]; diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index eb804e93..dbe27b4d 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -78,17 +78,17 @@ use saorsa_core::P2PNode; use tokio::sync::RwLock; use crate::ant_protocol::XorName; +use crate::client::xor_distance; use crate::replication::audit_coordinator::AuditChallengeCoordinator; use crate::replication::audit_metrics::{self, AuditFailureClass, AuditType}; use crate::replication::commitment_state::ResponderCommitmentState; use crate::replication::config::{ storage_admission_width, ReplicationConfig, AUDIT_FAILURE_TRUST_WEIGHT, MAX_PAID_EVICTIONS_PER_PASS, MAX_PRUNE_AUDIT_CANDIDATES_PER_PASS, - MAX_PRUNE_AUDIT_REQUESTS_PER_PASS, REPLICATION_PROTOCOL_ID, + MAX_PRUNE_AUDIT_REQUESTS_PER_PASS, PAID_LIST_NEAR_RANK, REPLICATION_PROTOCOL_ID, }; -use crate::replication::neighbor_sync::paid_near_bits; use crate::replication::out_of_range_clocks::elapsed; -use crate::replication::paid_list::PaidList; +use crate::replication::paid_list::{near_bits_for_peer_distances, PaidList}; use crate::replication::protocol::{ compute_audit_digest, AuditChallenge, AuditResponse, ReplicationMessage, ReplicationMessageBody, ABSENT_KEY_DIGEST, @@ -682,6 +682,21 @@ fn select_record_prune_audits( } } +/// How many low bits of a key may differ from this node's for its paid entry to +/// count as near, from the distance to its [`PAID_LIST_NEAR_RANK`]th closest +/// peer. Near entries are never evicted. +async fn paid_near_bits(p2p_node: &Arc) -> usize { + let self_xor: XorName = *p2p_node.peer_id().as_bytes(); + let distances: Vec = p2p_node + .dht_manager() + .find_closest_nodes_local(&self_xor, PAID_LIST_NEAR_RANK) + .await + .iter() + .map(|node| xor_distance(&self_xor, node.peer_id.as_bytes())) + .collect(); + near_bits_for_peer_distances(&distances) +} + /// Evict `PaidForList` entries once the list has passed its threshold, and /// return how many entries it now holds and how many went. /// diff --git a/src/storage/chunk_store.rs b/src/storage/chunk_store.rs index d6029349..e7bdfe0a 100644 --- a/src/storage/chunk_store.rs +++ b/src/storage/chunk_store.rs @@ -1644,6 +1644,28 @@ impl ChunkStore { .collect()) } + /// The stored keys in each of `ranges`, from its first key to its last, in + /// ascending order, leaving out those [`Self::all_keys`] leaves out. + pub(crate) fn keys_in_ranges(&self, ranges: &[(XorName, XorName)]) -> Vec> { + // Copied out first, as in `all_keys`, so no two of these locks are held at once. + let mut unservable: HashSet = self.suspect.read().clone(); + unservable.extend(self.known_wrong.read().iter().copied()); + let index = self.index.read(); + ranges + .iter() + .map(|(low, high)| { + if low > high { + return Vec::new(); + } + index + .range(*low..=*high) + .filter(|key| !unservable.contains(*key)) + .copied() + .collect() + }) + .collect() + } + /// Stop answering for a chunk this store could not read. /// /// The file stays. It may be perfectly good and unreadable only for the moment, and @@ -3544,6 +3566,33 @@ mod tests { assert_eq!(reopened.all_keys().await.expect("all_keys"), keys); } + #[tokio::test] + async fn keys_in_ranges_reads_each_range_and_leaves_out_unservable_keys() { + const STORED: usize = 64; + let (store, _dir) = test_store().await; + for i in 0..STORED { + let (addr, content) = addressed(&format!("ranged-{i}")); + store.put(&addr, &content).await.expect("put"); + } + let all = store.all_keys().await.expect("all_keys"); + let (first_half, second_half) = all.split_at(STORED / 2); + let suspect = second_half[0]; + store.mark_suspect(&suspect); + + let ranges = [ + (first_half[0], first_half[first_half.len() - 1]), + (second_half[0], [0xFF; 32]), + (second_half[1], first_half[0]), + ]; + let read = store.keys_in_ranges(&ranges); + assert_eq!(read[0], first_half, "both ends are included"); + assert_eq!(read[1], second_half[1..], "a suspect key is left out"); + assert!( + read[2].is_empty(), + "a range ending before it starts is empty" + ); + } + #[tokio::test] async fn get_raw_skips_verification() { let (store, _dir) = test_store().await; diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index f334721b..5f8701ed 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -2025,9 +2025,8 @@ async fn prune_marks_immediately_and_candidacy_waits_for_hysteresis() { harness.teardown().await.expect("teardown"); } -/// In a small network, a node still sends a peer the hint for a key far from the node -/// when that peer is the key's closest. `PaidList`'s unit tests cover the near range -/// itself: a network this small can't set one. +/// A node sends a peer the hint for a key far from the node when that peer is the +/// key's closest. #[tokio::test(flavor = "multi_thread")] #[serial] async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { @@ -2044,7 +2043,6 @@ async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { .dht_manager() .find_closest_nodes_local(&self_xor, EVERY_KNOWN_PEER) .await; - assert!(peers.len() < ant_node::replication::config::PAID_LIST_NEAR_RANK); let farthest = peers.last().expect("a known peer").peer_id; // Beside the farthest peer, so that peer alone is its closest. @@ -2070,6 +2068,114 @@ async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() { harness.teardown().await.expect("teardown"); } +/// The keys for which a lookup puts `peer` among the closest `group_size`, in order. +async fn keys_by_lookup( + p2p: &Arc, + keys: &[XorName], + peer: &PeerId, + group_size: usize, +) -> Vec { + let mut hinted = Vec::new(); + for key in keys { + let closest = p2p + .dht_manager() + .find_closest_nodes_local_with_self(key, group_size) + .await; + if closest.iter().any(|node| node.peer_id == *peer) { + hinted.push(*key); + } + } + hinted +} + +/// Every peer gets exactly the paid and replica hints a lookup per key would +/// give it. +#[tokio::test(flavor = "multi_thread")] +#[serial] +async fn hint_builds_match_a_lookup_per_key() { + const PAID_KEYS: usize = 400; + const STORED_CHUNKS: usize = 150; + const EVERY_KNOWN_PEER: usize = 100; + // The routing table can change while expected hints are worked out, so a peer + // is checked again until two lookups in a row agree. + const ATTEMPTS: usize = 3; + let harness = TestHarness::setup().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let node = harness.test_node(0).expect("node"); + let p2p = Arc::clone(node.p2p_node.as_ref().expect("p2p")); + let storage = node.ant_protocol.as_ref().expect("protocol").storage(); + let config = ReplicationConfig::default(); + + let paid_dir = tempfile::tempdir().expect("tempdir"); + let paid_list = Arc::new( + ant_node::replication::paid_list::PaidList::new(paid_dir.path()) + .await + .expect("paid list"), + ); + let mut paid_keys: Vec = (0..PAID_KEYS).map(|_| rand::random()).collect(); + paid_keys.sort_unstable(); + for key in &paid_keys { + paid_list.insert(key).await.expect("insert"); + } + for i in 0..STORED_CHUNKS { + let content = format!("hint-equivalence-{i}").into_bytes(); + storage + .put(&compute_address(&content), &content) + .await + .expect("put"); + } + let stored_keys = storage.all_keys().await.expect("stored keys"); + + let self_xor: XorName = *p2p.peer_id().as_bytes(); + let peers = p2p + .dht_manager() + .find_closest_nodes_local(&self_xor, EVERY_KNOWN_PEER) + .await; + assert!(!peers.is_empty()); + let (mut paid_hinted, mut replica_hinted) = (0, 0); + for peer in peers.iter().map(|node| node.peer_id) { + let mut agreed = false; + for _ in 0..ATTEMPTS { + let expected_paid = + keys_by_lookup(&p2p, &paid_keys, &peer, config.paid_list_close_group_size).await; + let expected_replica = + keys_by_lookup(&p2p, &stored_keys, &peer, config.close_group_size).await; + let paid = ant_node::replication::neighbor_sync::build_paid_hints_for_peer( + &peer, + &paid_list, + &p2p, + config.paid_list_close_group_size, + ) + .await; + let replica = ant_node::replication::neighbor_sync::build_replica_hints_for_peer( + &peer, + &storage, + &p2p, + config.close_group_size, + ) + .await; + let still_paid = + keys_by_lookup(&p2p, &paid_keys, &peer, config.paid_list_close_group_size).await; + let still_replica = + keys_by_lookup(&p2p, &stored_keys, &peer, config.close_group_size).await; + if still_paid != expected_paid || still_replica != expected_replica { + continue; + } + assert_eq!(paid, expected_paid, "paid hints for {peer}"); + assert_eq!(replica, expected_replica, "replica hints for {peer}"); + paid_hinted += paid.len(); + replica_hinted += replica.len(); + agreed = true; + break; + } + assert!(agreed, "the routing table kept changing"); + } + assert!(paid_hinted > 0 && replica_hinted > 0, "nothing was hinted"); + + harness.teardown().await.expect("teardown"); +} + /// Paid entries are evicted only once the list passes its threshold, and disk /// pressure alone never evicts one. Even over the threshold, an entry near this /// node stays. `PaidList`'s unit tests cover evicting far entries: in a network From 3b42f75ab7df3ea3752a23a78f3d33639ace1ed1 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:56:48 +0100 Subject: [PATCH 4/6] perf(replication): use the close-group map in prune passes and hint admission A prune pass looked up every stored record in the routing table, including the out-of-range records now kept until disk pressure, and hint admission looked up every hinted key this node didn't already know. Each now builds one map per pass or batch: the retention width for pruning, and the paid close group for admission. CloseGroupMap::group_of returns a group closest first, as a lookup does, so the strict group is still the head of the retention group. The check just before an erasure still does a fresh lookup, since the routing table may have changed while the proofs were gathered. Both widths are pinned by unit tests: mapping either at the wrong width passed every existing test. Co-Authored-By: Claude Opus 5.5 --- docs/adr/ADR-0028-pressure-gated-retention.md | 14 +++- src/replication/admission.rs | 50 ++++++++----- src/replication/close_group_map.rs | 53 +++++++++++--- src/replication/neighbor_sync.rs | 18 +---- src/replication/pruning.rs | 72 +++++++++++++++++-- 5 files changed, 155 insertions(+), 52 deletions(-) diff --git a/docs/adr/ADR-0028-pressure-gated-retention.md b/docs/adr/ADR-0028-pressure-gated-retention.md index 27b78ee0..3cd3cb86 100644 --- a/docs/adr/ADR-0028-pressure-gated-retention.md +++ b/docs/adr/ADR-0028-pressure-gated-retention.md @@ -202,6 +202,11 @@ store shares that partition. 0.35 ms instead of 179–195 ms, with the same hints. Those keys were spread over the whole key space, so the neighbour's ranges held under a tenth of them. A real node's keys cluster around it, so a build reads a larger share, at well under 1 µs a key. + - **Prune passes and hint admission** use the same map in place of a lookup per key: a + prune pass checks every stored record, including the out-of-range ones this ADR keeps, + and admission checks every hinted key this node doesn't already know. The check just + before an erasure still does a fresh lookup, since the routing table may have changed + while the proofs were gathered. - **Store size:** the LMDB map grows from 256 MiB to 1 GiB, which holds about 17.6 million entries, measured at about 61 bytes each with single-key commits. The file grows as entries do, on every platform. A full store logs one error, since new paid @@ -214,7 +219,8 @@ store shares that partition. - `src/replication/retention.rs`: the pressure state, thresholds and erase order. - `src/replication/out_of_range_clocks.rs`: saving and restoring the clocks. - `ChunkStore::disk_space`: the measurement. -- `src/replication/close_group_map.rs`: close groups per key range, for hint builds. +- `src/replication/close_group_map.rs`: close groups per key range, for hint builds, + prune passes and hint admission. ### Why this holds @@ -309,10 +315,12 @@ store shares that partition. - paid-list eviction: farthest first, up to its limit, never what it keeps, never near; - near ranges, and reading only the keys in one; - a far paid entry kept until the node knows 60 peers; - - close-group maps matching a sort of every candidate by distance, for 0 to 250 - candidates and groups of 1 to 20, with IDs evenly spread, clustered beside one + - close-group maps matching, in order, a sort of every candidate by distance, for 0 to + 250 candidates and groups of 1 to 20, with IDs evenly spread, clustered beside one candidate, or sharing their top bits, and ranges covering every key once; - only keys whose group holds a peer being hinted; + - a map of the retention width giving a prune pass the retention and strict groups, in + the order a lookup gives them; - reading key ranges from the record store and the paid list; - the store's file not taking the whole map up front; - removing a paid entry leaving its record's clock; diff --git a/src/replication/admission.rs b/src/replication/admission.rs index 2fb21098..a9d58a49 100644 --- a/src/replication/admission.rs +++ b/src/replication/admission.rs @@ -15,6 +15,7 @@ use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; use crate::ant_protocol::XorName; +use crate::replication::close_group_map::{close_group_candidates, CloseGroupMap}; use crate::replication::config::ReplicationConfig; use crate::replication::paid_list::PaidList; use crate::storage::ChunkStore; @@ -77,22 +78,26 @@ pub async fn is_in_paid_close_group( /// Storage responsibility is a *different* question, asked later against live /// routing state at the point of download. Keeping the two apart is what makes /// the sender's replica/paid labelling unable to influence what we store. -async fn is_relevant( +fn is_relevant( self_id: &PeerId, key: &XorName, - p2p_node: &Arc, - config: &ReplicationConfig, - storage: &Arc, - paid_list: &Arc, + paid_groups: &CloseGroupMap, + storage: &ChunkStore, + paid_list: &PaidList, pending_keys: &HashSet, ) -> bool { // Fast paths: we already hold the key, already track it, or already know it // is paid for. Each means the relevance question is settled -- no - // routing-table lookup needed. + // close-group check needed. storage.exists(key).unwrap_or(false) || pending_keys.contains(key) || paid_list.contains(key).unwrap_or(false) - || is_in_paid_close_group(self_id, key, p2p_node, config.paid_list_close_group_size).await + || paid_groups.includes(key, self_id) +} + +/// Every key's `PaidCloseGroup`, for a whole batch of hints. +fn paid_group_map(candidates: Vec, config: &ReplicationConfig) -> CloseGroupMap { + CloseGroupMap::new(candidates, config.paid_list_close_group_size) } /// Admit neighbor-sync hints per Section 7.1 rules. @@ -122,6 +127,11 @@ pub async fn admit_hints( paid_only_keys: Vec::new(), rejected_keys: Vec::new(), }; + if replica_hints.is_empty() && paid_hints.is_empty() { + return result; + } + // One routing-table snapshot for the whole batch, instead of a lookup per key. + let paid_groups = paid_group_map(close_group_candidates(p2p_node).await, config); let mut seen_replica = HashSet::new(); for &key in replica_hints { @@ -131,14 +141,11 @@ pub async fn admit_hints( if is_relevant( self_id, &key, - p2p_node, - config, + &paid_groups, storage, paid_list, pending_keys, - ) - .await - { + ) { result.replica_keys.push(key); } else { result.rejected_keys.push(key); @@ -159,14 +166,11 @@ pub async fn admit_hints( if is_relevant( self_id, &key, - p2p_node, - config, + &paid_groups, storage, paid_list, pending_keys, - ) - .await - { + ) { result.paid_only_keys.push(key); } else { result.rejected_keys.push(key); @@ -199,6 +203,18 @@ mod tests { [b; 32] } + #[test] + fn a_node_ranked_just_inside_the_paid_close_group_finds_a_key_relevant() { + const CANDIDATES: u8 = 30; + let config = ReplicationConfig::default(); + // A key of zeroes, so each candidate's rank is its first byte. + let candidates: Vec = (1..=CANDIDATES).map(peer_id_from_byte).collect(); + let rank = |r: usize| peer_id_from_byte(u8::try_from(r).expect("rank fits")); + let map = paid_group_map(candidates, &config); + assert!(map.includes(&[0u8; 32], &rank(config.paid_list_close_group_size))); + assert!(!map.includes(&[0u8; 32], &rank(config.paid_list_close_group_size + 1))); + } + // ----------------------------------------------------------------------- // AdmissionResult construction helpers for pure-logic tests // diff --git a/src/replication/close_group_map.rs b/src/replication/close_group_map.rs index a85f884c..3826a8f0 100644 --- a/src/replication/close_group_map.rs +++ b/src/replication/close_group_map.rs @@ -17,12 +17,14 @@ //! //! Each key's group is the one sorting every candidate by distance to it would //! give: the map applies the same order, once per range. Hint builds use it to read -//! only the ranges where a peer they build for can be in the group. +//! only the ranges where a peer they build for can be in the group. Prune passes +//! and hint admission use it in place of a routing-table lookup per key. use std::collections::HashSet; use std::ops::Range; use saorsa_core::identity::PeerId; +use saorsa_core::P2PNode; use crate::ant_protocol::{XorName, XORNAME_LEN}; use crate::client::xor_distance; @@ -39,6 +41,23 @@ const KEY_BITS: usize = XORNAME_LEN * BITS_PER_BYTE; /// A byte's highest bit, which comes first in a key's bit order. const HIGHEST_BIT_IN_BYTE: u8 = 0x80; +/// Every candidate for a key's close group in this node's view: its routing table +/// and itself, the set a self-inclusive lookup sorts. Take it once for a whole +/// pass or build. +pub async fn close_group_candidates(p2p_node: &P2PNode) -> Vec { + let self_id = *p2p_node.peer_id(); + let mut candidates: Vec = p2p_node + .dht_manager() + .routing_table_peers() + .await + .into_iter() + .map(|node| node.peer_id) + .filter(|peer| *peer != self_id) + .collect(); + candidates.push(self_id); + candidates +} + /// Keys from `low` to `high` whose close groups hold the same settled candidates, /// plus the closest `need` of the open ones, picked for each key. #[derive(Debug, Clone)] @@ -136,13 +155,25 @@ impl CloseGroupMap { group } - /// The close group of any key. - #[cfg(test)] + /// The close group of `key`, closest first, as a lookup returns it. pub fn group_of(&self, key: &XorName) -> Vec { + let mut group = self + .range_of(key) + .map_or_else(Vec::new, |range| self.group_in(range, key)); + group.sort_unstable_by_key(|peer| xor_distance(key, peer.as_bytes())); + group + } + + /// Whether `peer` is in the close group of `key`. + pub fn includes(&self, key: &XorName, peer: &PeerId) -> bool { + self.range_of(key) + .is_some_and(|range| self.group_in(range, key).contains(peer)) + } + + /// The range holding `key`. + fn range_of(&self, key: &XorName) -> Option<&GroupRange> { let index = self.ranges.partition_point(|range| range.high < *key); - self.ranges - .get(index) - .map_or_else(Vec::new, |range| self.group_in(range, key)) + self.ranges.get(index) } } @@ -262,6 +293,8 @@ mod tests { LARGE_TABLE, ]; const EXAMPLE_GROUP: usize = 2; + /// Candidates checked for membership of each key's group. + const MEMBERSHIP_CHECKS: usize = 3; /// Bytes a clustered ID shares with the candidate it sits beside. const CLUSTERED_SHARED_BYTES: usize = 3; const CLUSTERED_IDS: usize = 40; @@ -307,13 +340,17 @@ mod tests { fn assert_matches_brute_force(candidates: &[PeerId], group_size: usize, rng: &mut StdRng) { let map = CloseGroupMap::new(candidates.iter().copied(), group_size); for key in probe_keys(candidates, rng) { + let expected = brute_force(candidates, &key, group_size); assert_eq!( - sorted(map.group_of(&key)), - sorted(brute_force(candidates, &key, group_size)), + map.group_of(&key), + expected, "{} candidates, group {group_size}, key {}", candidates.len(), hex::encode(key) ); + for candidate in candidates.iter().take(MEMBERSHIP_CHECKS) { + assert_eq!(map.includes(&key, candidate), expected.contains(candidate)); + } } } diff --git a/src/replication/neighbor_sync.rs b/src/replication/neighbor_sync.rs index c773d3d7..c190749b 100644 --- a/src/replication/neighbor_sync.rs +++ b/src/replication/neighbor_sync.rs @@ -14,7 +14,7 @@ use saorsa_core::P2PNode; use crate::ant_protocol::XorName; use crate::error::Result; -use crate::replication::close_group_map::{CloseGroupMap, GroupRange}; +use crate::replication::close_group_map::{close_group_candidates, CloseGroupMap, GroupRange}; use crate::replication::config::{ReplicationConfig, REPLICATION_PROTOCOL_ID}; use crate::replication::paid_list::PaidList; use crate::replication::protocol::{ @@ -80,22 +80,6 @@ pub async fn build_replica_hints_for_peer( .collect() } -/// Every candidate for a key's close group in this node's view: its routing table -/// and itself. One snapshot serves a whole build. -async fn close_group_candidates(p2p_node: &Arc) -> Vec { - let self_id = *p2p_node.peer_id(); - let mut candidates: Vec = p2p_node - .dht_manager() - .routing_table_peers() - .await - .into_iter() - .map(|node| node.peer_id) - .filter(|peer| *peer != self_id) - .collect(); - candidates.push(self_id); - candidates -} - /// Call `hint` with each key `read` returns whose close group in `groups` holds /// one of `targets`, and with that group. `read` is asked only for the ranges /// where one of them can be in the group. Returns how many keys were read. diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index dbe27b4d..1b110dc8 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -81,6 +81,7 @@ use crate::ant_protocol::XorName; use crate::client::xor_distance; use crate::replication::audit_coordinator::AuditChallengeCoordinator; use crate::replication::audit_metrics::{self, AuditFailureClass, AuditType}; +use crate::replication::close_group_map::{close_group_candidates, CloseGroupMap}; use crate::replication::commitment_state::ResponderCommitmentState; use crate::replication::config::{ storage_admission_width, ReplicationConfig, AUDIT_FAILURE_TRUST_WEIGHT, @@ -378,8 +379,11 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune commitment_state: ctx.commitment_state, }; + let retention_groups = + retention_group_map(close_group_candidates(ctx.p2p_node).await, ctx.config); for key in &stored_keys { - let groups = record_prune_lookup_groups(key, ctx.p2p_node, ctx.config).await; + let groups = + split_record_prune_groups(retention_groups.group_of(key), ctx.config.close_group_size); let outcome = evaluate_record_prune_key(&deps, key, &groups.retention, &groups.strict, now); if let Some(candidate) = tally_record_prune_outcome(&mut stats, outcome, key) { @@ -501,7 +505,6 @@ async fn record_prune_lookup_groups( p2p_node: &Arc, config: &ReplicationConfig, ) -> RecordPruneLookupGroups { - // Closest first, so the strict group is the head of the retention group. let retention: Vec = p2p_node .dht_manager() .find_closest_nodes_local_with_self(key, storage_admission_width(config.close_group_size)) @@ -509,11 +512,20 @@ async fn record_prune_lookup_groups( .iter() .map(|node| node.peer_id) .collect(); - let strict = retention - .iter() - .take(config.close_group_size) - .copied() - .collect(); + split_record_prune_groups(retention, config.close_group_size) +} + +/// Every stored record's storage-retention group, for a whole pass. +fn retention_group_map(candidates: Vec, config: &ReplicationConfig) -> CloseGroupMap { + CloseGroupMap::new(candidates, storage_admission_width(config.close_group_size)) +} + +/// The retention group, closest first, with its head as the strict close group. +fn split_record_prune_groups( + retention: Vec, + close_group_size: usize, +) -> RecordPruneLookupGroups { + let strict = retention.iter().take(close_group_size).copied().collect(); RecordPruneLookupGroups { retention, strict } } @@ -1622,6 +1634,52 @@ mod tests { PeerId::from_bytes(bytes) } + #[test] + fn the_strict_group_is_the_closest_part_of_the_retention_group() { + const RETENTION: u8 = 9; + const STRICT: usize = 7; + let retention: Vec = (1..=RETENTION).map(peer_id_from_byte).collect(); + let groups = split_record_prune_groups(retention.clone(), STRICT); + assert_eq!(groups.strict, retention[..STRICT]); + assert_eq!(groups.retention, retention); + } + + #[test] + fn a_node_ranked_just_inside_the_retention_width_is_in_range_but_not_strict() { + const CANDIDATES: u8 = 30; + let config = ReplicationConfig::default(); + let width = storage_admission_width(config.close_group_size); + // A key of zeroes, so each candidate's rank is its first byte. + let candidates: Vec = (1..=CANDIDATES).map(peer_id_from_byte).collect(); + let this_node = peer_id_from_byte(u8::try_from(width).expect("width fits")); + let map = retention_group_map(candidates, &config); + let groups = split_record_prune_groups(map.group_of(&[0u8; 32]), config.close_group_size); + assert!(groups.retention.contains(&this_node)); + assert!(!groups.strict.contains(&this_node)); + } + + #[test] + fn a_retention_map_gives_the_groups_a_lookup_would() { + const CANDIDATES: u8 = 40; + const STRICT: usize = 7; + let candidates: Vec = (0..CANDIDATES) + .map(|b| peer_id_from_byte(b.wrapping_mul(37))) + .collect(); + let map = CloseGroupMap::new(candidates.iter().copied(), storage_admission_width(STRICT)); + for first in 0..=u8::MAX { + let mut key = [0u8; 32]; + key[0] = first; + let mut by_distance = candidates.clone(); + by_distance.sort_by_key(|peer| crate::client::xor_distance(&key, peer.as_bytes())); + let groups = split_record_prune_groups(map.group_of(&key), STRICT); + assert_eq!( + groups.retention, + by_distance[..storage_admission_width(STRICT)] + ); + assert_eq!(groups.strict, by_distance[..STRICT]); + } + } + fn key_from_byte(b: u8) -> XorName { [b; 32] } From 7b689c6e2a548fc7fd5bb9d4352d1aa40229d5a6 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:59:15 +0100 Subject: [PATCH 5/6] test(replication): compare the empty group with assert_eq clippy 1.99 rejects assert!(x.is_empty()) under -D warnings (assert_is_empty). Co-Authored-By: Claude Opus 5.5 --- src/replication/close_group_map.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/replication/close_group_map.rs b/src/replication/close_group_map.rs index 3826a8f0..42fbc324 100644 --- a/src/replication/close_group_map.rs +++ b/src/replication/close_group_map.rs @@ -484,6 +484,6 @@ mod tests { let empty = CloseGroupMap::new(Vec::new(), PAID_GROUP); assert_eq!(empty.ranges().len(), 1); - assert!(empty.group_of(&rng.gen()).is_empty()); + assert_eq!(empty.group_of(&rng.gen()), Vec::::new()); } } From ca44084b8d81fe0a6286ec388c2be8bc52c1aac5 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:24:20 +0100 Subject: [PATCH 6/6] fix(replication): bound close-group maps and yield in long key loops A map now stops splitting after 1,024 ranges. Zero plus one ID per high bit splits into C(n, G) ranges, about 30 million for 30 IDs and a group of 20, where routing tables gave 13 to 202. Past the limit, each prefix still pending becomes one range whose open candidates each key compares, so groups stay exact. The loops over keys in hint builds, prune passes, admission and commitment rotation call consume_budget, so a large store doesn't hold a runtime worker the way the per-key lookups' awaits used to prevent. Commitment rotation uses one map instead of a lookup per stored key and pointer. The near range is three paid close groups out (PAID_LIST_NEAR_GROUPS), so it follows a configured group size; the 60th peer by default. A sync from a peer outside the routing table returns before building maps, and hint helpers read their store outside the shared loop, so the record-store read no longer goes through an artificial Result. Docs: the design doc's definition of near, and ADR-0028's lists. Co-Authored-By: Claude Opus 5.5 --- docs/REPLICATION_DESIGN.md | 2 +- docs/adr/ADR-0028-pressure-gated-retention.md | 29 ++-- src/replication/admission.rs | 3 + src/replication/close_group_map.rs | 42 ++++-- src/replication/config.rs | 9 +- src/replication/mod.rs | 9 +- src/replication/neighbor_sync.rs | 135 +++++++++--------- src/replication/paid_list.rs | 39 +++-- src/replication/pruning.rs | 18 ++- tests/e2e/replication.rs | 2 +- 10 files changed, 177 insertions(+), 111 deletions(-) diff --git a/docs/REPLICATION_DESIGN.md b/docs/REPLICATION_DESIGN.md index c2800252..981d34be 100644 --- a/docs/REPLICATION_DESIGN.md +++ b/docs/REPLICATION_DESIGN.md @@ -115,7 +115,7 @@ Parameter safety constraints (MUST hold): - Under disk pressure (ADR-0028), audit scheduling orders eligible candidates farthest from self first (candidates whose proofs failed last, oldest failure first), groups a bounded window of them by target peer and sends multi-key requests, split at `max(floor(sqrt(local_in_range_keys)), 1)` keys per request, sized from the records self is in range for so retained records never make a batch too large for a peer to accept. Without pressure, no prune-confirmation audit is sent. The pass is bounded by actual requests and complete candidate proof sets, not candidate-to-peer edges; no partial candidate audit is carried across passes. A responder's size rejection is not renegotiated in-band: since its acceptance limit already carries a 2x margin over the sender size, an honestly-sized peer rejects only once close-group store spread exceeds 4x, so under the assumed ~2x spread a size rejection is graded as an attributable failure (distinct `size_reject` label), never a possession signal. - Immediately before audited deletion, revalidate against the current routing table: the key must still be outside the retention width, must not be held by an answerable retained commitment, and the positive reports must still satisfy the CURRENT strict close group (membership churn invalidates stale reports). - Prune-confirmation failures emit trust penalties only after fresh responsibility confirmation; first-time bootstrap claims use the same one-time `BOOTSTRAP_CLAIM_GRACE_PERIOD` tracking before abuse penalties apply. - `PaidForList` entries stay at or below `paid_list_evict_start_entries` (about 85% of the store's capacity, ADR-0028), whatever the disk. Above it, a pass evicts entries down to the threshold, locally and with no confirmations: those farthest from self first, never those near self (within four to eight times the distance to self's farthest paid-close-group peer) or those whose record self still stores. + `PaidForList` entries stay at or below `paid_list_evict_start_entries` (about 85% of the store's capacity, ADR-0028), whatever the disk. Above it, a pass evicts entries down to the threshold, locally and with no confirmations: those farthest from self first, never those near self (within the distance to self's closest peer three paid close groups out, `PAID_LIST_NEAR_GROUPS`) or those whose record self still stores. 23. Peers claiming bootstrap status are skipped for sync and audit without penalty for up to `BOOTSTRAP_CLAIM_GRACE_PERIOD` from first observation. After the grace period, each continued bootstrap claim emits `BootstrapClaimAbuse` evidence to `TrustEngine` (via `report_trust_event` with `ApplicationFailure(weight)`). If a peer stops claiming bootstrap and later claims bootstrap again, the repeated claim emits `BootstrapClaimAbuse` immediately; the grace period is not reset. 24. Audit trust-penalty signals require responsibility confirmation: on audit failure, challenger MUST perform fresh local RT closest-peer lookups for each challenged key and only penalize the peer for keys where it is confirmed responsible. diff --git a/docs/adr/ADR-0028-pressure-gated-retention.md b/docs/adr/ADR-0028-pressure-gated-retention.md index 3cd3cb86..05fda2bf 100644 --- a/docs/adr/ADR-0028-pressure-gated-retention.md +++ b/docs/adr/ADR-0028-pressure-gated-retention.md @@ -178,10 +178,11 @@ store shares that partition. farthest from this node first. Eviction is local, with no confirmations. - **What stays even then:** entries near this node, and entries of records it still stores. A key is near when its distance has no more significant bits than the - distance to this node's 60th closest routing-table peer (`PAID_LIST_NEAR_RANK`). With - evenly spread IDs and at most 20 peers per ring, about 120 nodes count a given key as - near, from a simulation of 4,000 nodes. While a node knows fewer than 60 peers, every - entry counts as near. + distance to this node's closest routing-table peer three paid close groups out + (`PAID_LIST_NEAR_GROUPS`): the 60th with the default group of 20. With evenly spread + IDs and at most 20 peers per ring, about 120 nodes count a given key as near, from a + simulation of 4,000 nodes. While a node knows fewer peers than that, every entry + counts as near. - **Sync hints:** each build takes one copy of the routing table and works out each key's close group once per key range, rather than once per key (`src/replication/close_group_map.rs`). It then reads only the ranges where a peer it @@ -202,11 +203,16 @@ store shares that partition. 0.35 ms instead of 179–195 ms, with the same hints. Those keys were spread over the whole key space, so the neighbour's ranges held under a tenth of them. A real node's keys cluster around it, so a build reads a larger share, at well under 1 µs a key. - - **Prune passes and hint admission** use the same map in place of a lookup per key: a - prune pass checks every stored record, including the out-of-range ones this ADR keeps, - and admission checks every hinted key this node doesn't already know. The check just - before an erasure still does a fresh lookup, since the routing table may have changed - while the proofs were gathered. + - **Prune passes, hint admission and commitment rotation** use the same map in place of + a lookup per key: a prune pass and a rotation check every stored record, including the + out-of-range ones this ADR keeps, and admission checks every hinted key this node + doesn't already know. The check just before an erasure still does a fresh lookup, + since the routing table may have changed while the proofs were gathered. + - **Bounds:** a map stops splitting after 1,024 ranges, where routing tables gave 13 to + 202. A skewed set of peer IDs could otherwise need millions. Past that, each remaining + range compares its competing peers for each key, which gives the same groups at more + cost a key. These loops let other tasks run between keys, as the per-key lookups' + awaits did. - **Store size:** the LMDB map grows from 256 MiB to 1 GiB, which holds about 17.6 million entries, measured at about 61 bytes each with single-key commits. The file grows as entries do, on every platform. A full store logs one error, since new paid @@ -313,11 +319,14 @@ store shares that partition. - record clocks surviving a reopen of the paid list; - a clock surviving "in range" during bootstrap; - paid-list eviction: farthest first, up to its limit, never what it keeps, never near; - - near ranges, and reading only the keys in one; + - near ranges reaching the counted peer, and every key near until enough peers are + known; - a far paid entry kept until the node knows 60 peers; - close-group maps matching, in order, a sort of every candidate by distance, for 0 to 250 candidates and groups of 1 to 20, with IDs evenly spread, clustered beside one candidate, or sharing their top bits, and ranges covering every key once; + - a skewed set of IDs that would split into millions of ranges staying within the limit, + with the same groups; - only keys whose group holds a peer being hinted; - a map of the retention width giving a prune pass the retention and strict groups, in the order a lookup gives them; diff --git a/src/replication/admission.rs b/src/replication/admission.rs index a9d58a49..1173ea40 100644 --- a/src/replication/admission.rs +++ b/src/replication/admission.rs @@ -13,6 +13,7 @@ use std::sync::Arc; use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; +use tokio::task::coop::consume_budget; use crate::ant_protocol::XorName; use crate::replication::close_group_map::{close_group_candidates, CloseGroupMap}; @@ -138,6 +139,7 @@ pub async fn admit_hints( if !seen_replica.insert(key) { continue; } + consume_budget().await; if is_relevant( self_id, &key, @@ -163,6 +165,7 @@ pub async fn admit_hints( if seen_replica.contains(&key) { continue; } + consume_budget().await; if is_relevant( self_id, &key, diff --git a/src/replication/close_group_map.rs b/src/replication/close_group_map.rs index 42fbc324..1a0e1fcd 100644 --- a/src/replication/close_group_map.rs +++ b/src/replication/close_group_map.rs @@ -13,7 +13,7 @@ //! The walk stops once every key under a prefix has the same group, or once the //! peers still competing don't differ at the next bit. Those few are then compared //! for each key. A routing table of a few hundred peers gives tens to a few hundred -//! ranges. +//! ranges, and the walk stops splitting past [`MAX_SPLIT_RANGES`] whatever the peers. //! //! Each key's group is the one sorting every candidate by distance to it would //! give: the map applies the same order, once per range. Hint builds use it to read @@ -28,6 +28,7 @@ use saorsa_core::P2PNode; use crate::ant_protocol::{XorName, XORNAME_LEN}; use crate::client::xor_distance; +use crate::replication::paid_list::XORNAME_BITS; #[cfg(test)] use rand::{rngs::StdRng, Rng, SeedableRng}; @@ -35,12 +36,14 @@ use rand::{rngs::StdRng, Rng, SeedableRng}; /// Bits in one byte of a key. const BITS_PER_BYTE: usize = 8; -/// Bits in a key. -const KEY_BITS: usize = XORNAME_LEN * BITS_PER_BYTE; - /// A byte's highest bit, which comes first in a key's bit order. const HIGHEST_BIT_IN_BYTE: u8 = 0x80; +/// Ranges after which a map stops splitting. Routing tables gave 13 to 202, but a +/// skewed set of candidates could need millions. Past this, each prefix left is one +/// range whose open candidates each key compares: the same groups, at more cost a key. +const MAX_SPLIT_RANGES: usize = 1_024; + /// Every candidate for a key's close group in this node's view: its routing table /// and itself, the set a self-inclusive lookup sorts. Take it once for a whole /// pass or build. @@ -202,9 +205,14 @@ impl Walk<'_> { // run from those with this bit clear to those with it set. let run = self.candidates.get(open.clone()).unwrap_or(&[]); let set_from = open.start + run.partition_point(|peer| !bit(peer.as_bytes(), depth)); - if depth >= KEY_BITS || set_from == open.start || set_from == open.end { + if depth >= XORNAME_BITS + || set_from == open.start + || set_from == open.end + || self.ranges.len() >= MAX_SPLIT_RANGES + { // No open candidate differs from another at this bit, so a key's bit - // here can't change which of them are closer: each key decides. + // here can't change which of them are closer, or the map has split + // enough: each key decides. let settled = self.settled.clone(); self.push(prefix, depth, settled, open, need); return; @@ -237,7 +245,7 @@ impl Walk<'_> { need: usize, ) { let mut high = prefix; - for free in depth..KEY_BITS { + for free in depth..XORNAME_BITS { high = with_bit(high, free, true); } self.ranges.push(GroupRange { @@ -330,7 +338,7 @@ mod tests { for candidate in candidates { let id = *candidate.as_bytes(); keys.push(id); - keys.push(with_bit(id, KEY_BITS - 1, !bit(&id, KEY_BITS - 1))); + keys.push(with_bit(id, XORNAME_BITS - 1, !bit(&id, XORNAME_BITS - 1))); keys.push(with_bit(id, 0, !bit(&id, 0))); } keys.extend((0..KEYS_PER_CASE).map(|_| rng.gen::())); @@ -395,6 +403,24 @@ mod tests { } } + #[test] + fn a_skewed_set_of_candidates_stays_within_the_range_limit() { + // Zero, then one candidate with each successive high bit set. Without the + // limit, a group of 20 would split into C(30, 20), about 30 million, ranges. + const ONE_HOT_CANDIDATES: usize = 29; + let mut candidates = vec![peer([0u8; XORNAME_LEN])]; + for depth in 0..ONE_HOT_CANDIDATES { + candidates.push(peer(with_bit([0u8; XORNAME_LEN], depth, true))); + } + let mut rng = StdRng::seed_from_u64(SEED); + for group_size in GROUP_SIZES { + let map = CloseGroupMap::new(candidates.iter().copied(), group_size); + // Past the limit, each prefix still pending on the walk adds one range. + assert!(map.ranges().len() <= MAX_SPLIT_RANGES + XORNAME_BITS); + assert_matches_brute_force(&candidates, group_size, &mut rng); + } + } + #[test] fn ranges_cover_every_key_once_in_order() { let mut rng = StdRng::seed_from_u64(SEED); diff --git a/src/replication/config.rs b/src/replication/config.rs index e3abd07f..25bb5d3d 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -101,15 +101,16 @@ const _: () = assert!( /// Most paid-list entries one prune pass evicts. pub const MAX_PAID_EVICTIONS_PER_PASS: usize = 100_000; -/// Which routing-table peer, counting out from this node, sets how far its near -/// range reaches. +/// How many paid close groups of routing-table peers, counting out from this +/// node, set how far its near range reaches: with the default group of 20, to the +/// 60th closest peer. /// /// A key is near when its distance has no more significant bits than the distance -/// to this peer, and a paid entry near this node is never evicted. Reading this +/// to that peer, and a paid entry near this node is never evicted. Reading this /// far out gives a steadier measure of local peer density than the closest few /// peers do, and reaches past every key this node or a neighbour is in the paid /// close group of when IDs are evenly spread (ADR-0028). -pub const PAID_LIST_NEAR_RANK: usize = 60; +pub const PAID_LIST_NEAR_GROUPS: usize = 3; /// Number of closest peers to self eligible for neighbor sync. pub const NEIGHBOR_SYNC_SCOPE: usize = 20; diff --git a/src/replication/mod.rs b/src/replication/mod.rs index acd4b940..90d7da1c 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -61,6 +61,7 @@ use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; use tokio::sync::broadcast::error::RecvError; use tokio::sync::{mpsc, Notify, RwLock, Semaphore}; +use tokio::task::coop::consume_budget; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; @@ -79,6 +80,7 @@ use crate::replication::audit_metrics::{ AuditResponderClass, AuditResponderDropReason, AuditResponderMetrics, ReplicationResponderClass, ResponderAdmissionCeiling, }; +use crate::replication::close_group_map::{close_group_candidates, CloseGroupMap}; use crate::replication::commitment::{commitment_hash, StorageCommitment}; use crate::replication::commitment_state::{ PeerCommitmentRecord, PersistedRetention, ResponderCommitmentState, GOSSIP_ANSWERABILITY_TTL, @@ -10346,9 +10348,12 @@ async fn rebuild_and_rotate_commitment( // forever (the rebuild reads all_keys, so a retained-on-disk key would be // re-committed and re-gossiped every rotation — a permanent pin). let self_id = *p2p.peer_id(); + // One routing-table snapshot for every key, instead of a lookup per key. + let groups = CloseGroupMap::new(close_group_candidates(p2p).await, config.close_group_size); let mut keys = Vec::with_capacity(stored_keys.len()); for k in stored_keys { - if admission::is_responsible(&self_id, &k, p2p, config.close_group_size).await { + consume_budget().await; + if groups.includes(&k, &self_id) { keys.push(k); } } @@ -10356,7 +10361,7 @@ async fn rebuild_and_rotate_commitment( // commitment, and the pruner reclaims it once no retained slot holds it. let mut pointer_keys = Vec::new(); for state in pointers.map(PointerStore::held_states).unwrap_or_default() { - if admission::is_responsible(&self_id, &state.address, p2p, config.close_group_size).await { + if groups.includes(&state.address, &self_id) { pointer_keys.push(state.address); } } diff --git a/src/replication/neighbor_sync.rs b/src/replication/neighbor_sync.rs index c190749b..bb1bb3b8 100644 --- a/src/replication/neighbor_sync.rs +++ b/src/replication/neighbor_sync.rs @@ -11,6 +11,7 @@ use crate::logging::{debug, info, warn}; use rand::Rng; use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; +use tokio::task::coop::consume_budget; use crate::ant_protocol::XorName; use crate::error::Result; @@ -71,7 +72,7 @@ pub async fn build_replica_hints_for_peer( ) -> Vec { let targets = HashSet::from([*peer]); let groups = CloseGroupMap::new(close_group_candidates(p2p_node).await, close_group_size); - let (mut hints, _) = replica_hints(&groups, storage, &targets); + let (mut hints, _) = replica_hints(&groups, storage, &targets).await; hints .remove(peer) .unwrap_or_default() @@ -80,78 +81,82 @@ pub async fn build_replica_hints_for_peer( .collect() } -/// Call `hint` with each key `read` returns whose close group in `groups` holds -/// one of `targets`, and with that group. `read` is asked only for the ranges -/// where one of them can be in the group. Returns how many keys were read. -fn for_each_hinted_key( - groups: &CloseGroupMap, +/// The ranges where one of `targets` can be in the close group, with their bounds. +fn ranges_holding<'a>( + groups: &'a CloseGroupMap, targets: &HashSet, - read: impl FnOnce(&[(XorName, XorName)]) -> Result>>, - mut hint: impl FnMut(XorName, Vec), -) -> Result { +) -> (Vec<&'a GroupRange>, Vec<(XorName, XorName)>) { let ranges: Vec<&GroupRange> = groups .ranges() .iter() .filter(|range| groups.may_include_any(range, targets)) .collect(); - let bounds: Vec<(XorName, XorName)> = ranges.iter().map(|range| range.bounds()).collect(); - let mut keys_read = 0; - for (range, keys) in ranges.into_iter().zip(read(&bounds)?) { - keys_read += keys.len(); + let bounds = ranges.iter().map(|range| range.bounds()).collect(); + (ranges, bounds) +} + +/// Call `hint` with each key of `keys`, read range by range from `ranges`, whose +/// close group in `groups` holds one of `targets`, and with that group. +async fn for_each_hinted_key( + groups: &CloseGroupMap, + ranges: Vec<&GroupRange>, + keys: Vec>, + targets: &HashSet, + mut hint: impl FnMut(XorName, Vec), +) { + for (range, keys) in ranges.into_iter().zip(keys) { for key in keys { + // A large store is a long loop: let other tasks run in between. + consume_budget().await; let group = groups.group_in(range, &key); if group.iter().any(|peer| targets.contains(peer)) { hint(key, group); } } } - Ok(keys_read) } /// Replica hints from the stored records for each of `targets`, and how many /// stored keys were read. -fn replica_hints( +async fn replica_hints( groups: &CloseGroupMap, storage: &ChunkStore, targets: &HashSet, ) -> (HashMap>, usize) { + let (ranges, bounds) = ranges_holding(groups, targets); + let keys = storage.keys_in_ranges(&bounds); + let keys_read = keys.iter().map(Vec::len).sum(); let mut hints: HashMap> = HashMap::new(); - let keys_read = for_each_hinted_key( - groups, - targets, - |ranges| Ok(storage.keys_in_ranges(ranges)), - |key, group| { - let close_peers: HashSet = group.into_iter().collect(); - for peer in close_peers.intersection(targets) { - hints.entry(*peer).or_default().push(SentReplicaHint { - key, - close_peers: close_peers.clone(), - }); - } - }, - ) - .unwrap_or_default(); + for_each_hinted_key(groups, ranges, keys, targets, |key, group| { + let close_peers: HashSet = group.into_iter().collect(); + for peer in close_peers.intersection(targets) { + hints.entry(*peer).or_default().push(SentReplicaHint { + key, + close_peers: close_peers.clone(), + }); + } + }) + .await; (hints, keys_read) } /// Paid hints from the paid list for each of `targets`, and how many entries /// were read. -fn paid_hints( +async fn paid_hints( groups: &CloseGroupMap, paid_list: &PaidList, targets: &HashSet, ) -> Result<(HashMap>, usize)> { + let (ranges, bounds) = ranges_holding(groups, targets); + let keys = paid_list.keys_in_ranges(&bounds)?; + let keys_read = keys.iter().map(Vec::len).sum(); let mut hints: HashMap> = HashMap::new(); - let keys_read = for_each_hinted_key( - groups, - targets, - |ranges| paid_list.keys_in_ranges(ranges), - |key, group| { - for peer in group.iter().filter(|peer| targets.contains(*peer)) { - hints.entry(*peer).or_default().push(key); - } - }, - )?; + for_each_hinted_key(groups, ranges, keys, targets, |key, group| { + for peer in group.iter().filter(|peer| targets.contains(*peer)) { + hints.entry(*peer).or_default().push(key); + } + }) + .await; Ok((hints, keys_read)) } @@ -181,18 +186,22 @@ pub(crate) async fn build_sync_hints_for_peers( } let candidates = close_group_candidates(p2p_node).await; + // A peer outside the routing table is in no group this node can see. + if !candidates.iter().any(|peer| targets.contains(peer)) { + return hints_by_peer; + } let candidate_count = candidates.len(); let record_groups = CloseGroupMap::new(candidates.iter().copied(), close_group_size); let paid_groups = CloseGroupMap::new(candidates, paid_list_close_group_size); - let (replica, stored_keys_read) = replica_hints(&record_groups, storage, &targets); + let (replica, stored_keys_read) = replica_hints(&record_groups, storage, &targets).await; for (peer, hints) in replica { if let Some(peer_hints) = hints_by_peer.get_mut(&peer) { peer_hints.sent_replica_hints = hints; } } - let paid_keys_read = match paid_hints(&paid_groups, paid_list, &targets) { + let paid_keys_read = match paid_hints(&paid_groups, paid_list, &targets).await { Ok((paid, keys_read)) => { for (peer, hints) in paid { if let Some(peer_hints) = hints_by_peer.get_mut(&peer) { @@ -249,7 +258,7 @@ pub async fn build_paid_hints_for_peer( close_group_candidates(p2p_node).await, paid_list_close_group_size, ); - match paid_hints(&groups, paid_list, &targets) { + match paid_hints(&groups, paid_list, &targets).await { Ok((mut hints, _)) => hints.remove(peer).unwrap_or_default(), Err(e) => { warn!("Failed to read PaidForList for hint construction: {e}"); @@ -588,8 +597,8 @@ mod tests { // -- hint builds ---------------------------------------------------------- - #[test] - fn only_keys_whose_group_holds_a_target_are_hinted() { + #[tokio::test] + async fn only_keys_whose_group_holds_a_target_are_hinted() { // Peers 0000, 0001, 0101, 1000 and 1110 in the top four bits, a group of // two. Keys starting 01 take 0101 and whichever of 0000 and 0001 is closer. const GROUP: usize = 2; @@ -606,31 +615,29 @@ mod tests { let keys = [0b0000_0000, 0b0100_0000, 0b0101_0000, 0b1100_0000] .map(|first| *peer_id_from_byte(first).as_bytes()); + let (ranges, bounds) = ranges_holding(&groups, &targets); + let read: Vec> = bounds + .iter() + .map(|(low, high)| { + keys.iter() + .copied() + .filter(|key| low <= key && key <= high) + .collect() + }) + .collect(); + assert_eq!( + read.iter().map(Vec::len).sum::(), + 3, + "keys from 1000 on aren't read" + ); let mut hinted = Vec::new(); - let keys_read = for_each_hinted_key( - &groups, - &targets, - |ranges| { - Ok(ranges - .iter() - .map(|(low, high)| { - keys.iter() - .copied() - .filter(|key| low <= key && key <= high) - .collect() - }) - .collect()) - }, - |key, _| hinted.push(key), - ) - .expect("hints"); + for_each_hinted_key(&groups, ranges, read, &targets, |key, _| hinted.push(key)).await; assert_eq!( hinted, vec![keys[0], keys[2]], "0100 goes to 0000, not 0001" ); - assert_eq!(keys_read, 3, "keys from 1000 on aren't read"); } // -- select_sync_batch --------------------------------------------------- diff --git a/src/replication/paid_list.rs b/src/replication/paid_list.rs index 256187ba..ae9d7a35 100644 --- a/src/replication/paid_list.rs +++ b/src/replication/paid_list.rs @@ -20,7 +20,7 @@ use crate::ant_protocol::XorName; use crate::error::{Error, Result}; use crate::logging::{debug, error, trace, warn}; -use crate::replication::config::{PAID_LIST_MAP_SIZE, PAID_LIST_NEAR_RANK}; +use crate::replication::config::PAID_LIST_MAP_SIZE; use crate::replication::out_of_range_clocks::OutOfRangeClocks; use heed::types::Bytes; use heed::{Database, Env, EnvOpenOptions, MdbError, RoTxn}; @@ -43,14 +43,13 @@ pub const XORNAME_BITS: usize = XORNAME_LEN * BITS_PER_BYTE; const RECORD_CLOCKS_FILE: &str = "record_out_of_range.bin"; /// How many low bits of a key may differ from this node's for the key to count -/// as near it, given its distances to the closest peers it knows. +/// as near it, given its distances to the closest `near_rank` peers it knows. /// -/// Every key counts as near until it knows [`PAID_LIST_NEAR_RANK`] peers: in a -/// network that small, this node or a neighbour can be in the paid close group of -/// any key. +/// Every key counts as near until it knows `near_rank` peers: in a network that +/// small, this node or a neighbour can be in the paid close group of any key. #[must_use] -pub fn near_bits_for_peer_distances(distances: &[XorName]) -> usize { - if distances.len() < PAID_LIST_NEAR_RANK { +pub fn near_bits_for_peer_distances(distances: &[XorName], near_rank: usize) -> usize { + if distances.len() < near_rank { return XORNAME_BITS; } distances @@ -541,6 +540,10 @@ impl PaidList { mod tests { use super::*; + /// The default near rank: three paid close groups out. + const NEAR_RANK: usize = crate::replication::config::PAID_LIST_CLOSE_GROUP_SIZE + * crate::replication::config::PAID_LIST_NEAR_GROUPS; + /// Short probe used to prove `wait_idle` is still blocked on a parked op. const WAIT_IDLE_BLOCKED_PROBE: std::time::Duration = std::time::Duration::from_millis(200); /// Generous ceiling for `wait_idle` to complete once the op is released. @@ -568,7 +571,7 @@ mod tests { /// As many distances as a near range needs, all very small, then `farthest`. fn distances_out_to(farthest: XorName) -> Vec { - let mut distances = vec![key_ending_in(1); PAID_LIST_NEAR_RANK - 1]; + let mut distances = vec![key_ending_in(1); NEAR_RANK - 1]; distances.push(farthest); distances } @@ -589,17 +592,23 @@ mod tests { #[test] fn every_key_is_near_until_enough_peers_are_known() { - assert_eq!(near_bits_for_peer_distances(&[]), XORNAME_BITS); + assert_eq!(near_bits_for_peer_distances(&[], NEAR_RANK), XORNAME_BITS); - let one_short = vec![key_ending_in(1); PAID_LIST_NEAR_RANK - 1]; - assert_eq!(near_bits_for_peer_distances(&one_short), XORNAME_BITS); + let one_short = vec![key_ending_in(1); NEAR_RANK - 1]; + assert_eq!( + near_bits_for_peer_distances(&one_short, NEAR_RANK), + XORNAME_BITS + ); } #[test] fn the_near_range_reaches_the_farthest_counted_peer_however_close_the_rest() { const FARTHEST_BITS: usize = 7; let distances = distances_out_to(key_ending_in(0b0100_0000)); - assert_eq!(near_bits_for_peer_distances(&distances), FARTHEST_BITS); + assert_eq!( + near_bits_for_peer_distances(&distances, NEAR_RANK), + FARTHEST_BITS + ); } #[tokio::test] @@ -612,14 +621,16 @@ mod tests { pl.insert(&key_ending_in(FAR_KEY)).await.expect("insert"); let this_node = key_ending_in(0); - let few_known = near_bits_for_peer_distances(&[key_ending_in(1), key_ending_in(2)]); + let few_known = + near_bits_for_peer_distances(&[key_ending_in(1), key_ending_in(2)], NEAR_RANK); let evicted = pl .evict_far(this_node, few_known, EVICT_UP_TO, |_| false) .await .expect("evict"); assert_eq!(evicted, 0, "too few peers known to call any key far"); - let many_known = near_bits_for_peer_distances(&distances_out_to(key_ending_in(2))); + let many_known = + near_bits_for_peer_distances(&distances_out_to(key_ending_in(2)), NEAR_RANK); let evicted = pl .evict_far(this_node, many_known, EVICT_UP_TO, |_| false) .await diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 1b110dc8..aa08f8a6 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -76,6 +76,7 @@ use rand::Rng; use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; use tokio::sync::RwLock; +use tokio::task::coop::consume_budget; use crate::ant_protocol::XorName; use crate::client::xor_distance; @@ -86,7 +87,7 @@ use crate::replication::commitment_state::ResponderCommitmentState; use crate::replication::config::{ storage_admission_width, ReplicationConfig, AUDIT_FAILURE_TRUST_WEIGHT, MAX_PAID_EVICTIONS_PER_PASS, MAX_PRUNE_AUDIT_CANDIDATES_PER_PASS, - MAX_PRUNE_AUDIT_REQUESTS_PER_PASS, PAID_LIST_NEAR_RANK, REPLICATION_PROTOCOL_ID, + MAX_PRUNE_AUDIT_REQUESTS_PER_PASS, PAID_LIST_NEAR_GROUPS, REPLICATION_PROTOCOL_ID, }; use crate::replication::out_of_range_clocks::elapsed; use crate::replication::paid_list::{near_bits_for_peer_distances, PaidList}; @@ -382,6 +383,8 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune let retention_groups = retention_group_map(close_group_candidates(ctx.p2p_node).await, ctx.config); for key in &stored_keys { + // A large store is a long loop: let other tasks run in between. + consume_budget().await; let groups = split_record_prune_groups(retention_groups.group_of(key), ctx.config.close_group_size); @@ -695,18 +698,19 @@ fn select_record_prune_audits( } /// How many low bits of a key may differ from this node's for its paid entry to -/// count as near, from the distance to its [`PAID_LIST_NEAR_RANK`]th closest -/// peer. Near entries are never evicted. -async fn paid_near_bits(p2p_node: &Arc) -> usize { +/// count as near: up to its closest peer [`PAID_LIST_NEAR_GROUPS`] paid close +/// groups out. Near entries are never evicted. +async fn paid_near_bits(p2p_node: &Arc, paid_group_size: usize) -> usize { + let near_rank = paid_group_size.saturating_mul(PAID_LIST_NEAR_GROUPS); let self_xor: XorName = *p2p_node.peer_id().as_bytes(); let distances: Vec = p2p_node .dht_manager() - .find_closest_nodes_local(&self_xor, PAID_LIST_NEAR_RANK) + .find_closest_nodes_local(&self_xor, near_rank) .await .iter() .map(|node| xor_distance(&self_xor, node.peer_id.as_bytes())) .collect(); - near_bits_for_peer_distances(&distances) + near_bits_for_peer_distances(&distances, near_rank) } /// Evict `PaidForList` entries once the list has passed its threshold, and @@ -728,7 +732,7 @@ async fn evict_excess_paid_entries(ctx: &PrunePassContext<'_>) -> (usize, usize) if excess == 0 { return (count, 0); } - let near_bits = paid_near_bits(ctx.p2p_node).await; + let near_bits = paid_near_bits(ctx.p2p_node, ctx.config.paid_list_close_group_size).await; let storage = Arc::clone(ctx.storage); let evicted = ctx .paid_list diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index 5f8701ed..c503262b 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -2098,7 +2098,7 @@ async fn hint_builds_match_a_lookup_per_key() { const EVERY_KNOWN_PEER: usize = 100; // The routing table can change while expected hints are worked out, so a peer // is checked again until two lookups in a row agree. - const ATTEMPTS: usize = 3; + const ATTEMPTS: usize = 5; let harness = TestHarness::setup().await.expect("setup"); harness.warmup_dht().await.expect("warmup");