diff --git a/docs/REPLICATION_DESIGN.md b/docs/REPLICATION_DESIGN.md
index 3aaf117c..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.
 
@@ -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. 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 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 888dc49b..05fda2bf 100644
--- a/docs/adr/ADR-0028-pressure-gated-retention.md
+++ b/docs/adr/ADR-0028-pressure-gated-retention.md
@@ -177,16 +177,42 @@ 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 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
+    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.
+  - **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
@@ -199,6 +225,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,
+  prune passes and hint admission.
 
 ### Why this holds
 
@@ -260,13 +288,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.
@@ -288,15 +319,28 @@ 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;
-  - a far key read for hints while the list can be read whole, and only near keys above that;
+  - 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;
+  - 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`: a small list is read whole.
+  - `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.
@@ -308,7 +352,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;
+  - 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/admission.rs b/src/replication/admission.rs
index 2fb21098..1173ea40 100644
--- a/src/replication/admission.rs
+++ b/src/replication/admission.rs
@@ -13,8 +13,10 @@ 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};
 use crate::replication::config::ReplicationConfig;
 use crate::replication::paid_list::PaidList;
 use crate::storage::ChunkStore;
@@ -77,22 +79,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<P2PNode>,
-    config: &ReplicationConfig,
-    storage: &Arc<ChunkStore>,
-    paid_list: &Arc<PaidList>,
+    paid_groups: &CloseGroupMap,
+    storage: &ChunkStore,
+    paid_list: &PaidList,
     pending_keys: &HashSet<XorName>,
 ) -> 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<PeerId>, config: &ReplicationConfig) -> CloseGroupMap {
+    CloseGroupMap::new(candidates, config.paid_list_close_group_size)
 }
 
 /// Admit neighbor-sync hints per Section 7.1 rules.
@@ -122,23 +128,26 @@ 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 {
         if !seen_replica.insert(key) {
             continue;
         }
+        consume_budget().await;
         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);
@@ -156,17 +165,15 @@ pub async fn admit_hints(
         if seen_replica.contains(&key) {
             continue;
         }
+        consume_budget().await;
         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 +206,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<PeerId> = (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
new file mode 100644
index 00000000..1a0e1fcd
--- /dev/null
+++ b/src/replication/close_group_map.rs
@@ -0,0 +1,515 @@
+//! 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, 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
+//! 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;
+use crate::replication::paid_list::XORNAME_BITS;
+
+#[cfg(test)]
+use rand::{rngs::StdRng, Rng, SeedableRng};
+
+/// Bits in one byte of a key.
+const BITS_PER_BYTE: usize = 8;
+
+/// 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.
+pub async fn close_group_candidates(p2p_node: &P2PNode) -> Vec<PeerId> {
+    let self_id = *p2p_node.peer_id();
+    let mut candidates: Vec<PeerId> = 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)]
+pub struct GroupRange {
+    low: XorName,
+    high: XorName,
+    /// Candidates in the group of every key in the range, as indices.
+    settled: Vec<usize>,
+    /// Candidates still competing, as a run of indices.
+    open: Range<usize>,
+    /// 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<PeerId>,
+    /// Ranges in key order, covering every key once.
+    ranges: Vec<GroupRange>,
+}
+
+impl CloseGroupMap {
+    /// Map each key's closest `group_size` of `candidates`.
+    pub fn new(candidates: impl IntoIterator<Item = PeerId>, group_size: usize) -> Self {
+        let mut candidates: Vec<PeerId> = 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<PeerId>) -> 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<PeerId> {
+        let mut group: Vec<PeerId> = 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 `key`, closest first, as a lookup returns it.
+    pub fn group_of(&self, key: &XorName) -> Vec<PeerId> {
+        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)
+    }
+}
+
+/// 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<usize>,
+    ranges: Vec<GroupRange>,
+}
+
+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<usize>, 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 >= 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, or the map has split
+            // enough: 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<usize>,
+        open: Range<usize>,
+        need: usize,
+    ) {
+        let mut high = prefix;
+        for free in depth..XORNAME_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;
+    /// 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;
+
+    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<PeerId> {
+        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<PeerId>) -> Vec<PeerId> {
+        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<XorName> {
+        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, 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::<XorName>()));
+        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) {
+            let expected = brute_force(candidates, &key, group_size);
+            assert_eq!(
+                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));
+            }
+        }
+    }
+
+    #[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<PeerId> = (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<PeerId> = (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<PeerId> = (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 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);
+        let candidates: Vec<PeerId> = (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<PeerId> = ids.iter().map(|first| peer(key_starting(*first))).collect();
+        let map = CloseGroupMap::new(candidates.iter().copied(), EXAMPLE_GROUP);
+        let starts: Vec<u8> = map.ranges().iter().map(|range| range.low[0]).collect();
+        assert_eq!(starts, vec![0b0000_0000, 0b0100_0000, 0b1000_0000]);
+
+        let settled = |range: &GroupRange| -> Vec<PeerId> {
+            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<bool> = 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<PeerId> = (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_eq!(empty.group_of(&rng.gen()), Vec::<PeerId>::new());
+    }
+}
diff --git a/src/replication/config.rs b/src/replication/config.rs
index 8c2e1da9..25bb5d3d 100644
--- a/src/replication/config.rs
+++ b/src/replication/config.rs
@@ -101,21 +101,16 @@ 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;
+/// 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 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_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 3e66f9bd..90d7da1c 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;
@@ -60,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;
@@ -78,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,
@@ -10345,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);
         }
     }
@@ -10355,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 550c9b35..bb1bb3b8 100644
--- a/src/replication/neighbor_sync.rs
+++ b/src/replication/neighbor_sync.rs
@@ -11,14 +11,13 @@ 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::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::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::{
     NeighborSyncRequest, NeighborSyncResponse, ReplicationMessage, ReplicationMessageBody,
 };
@@ -62,98 +61,110 @@ 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<ChunkStore>,
     p2p_node: &Arc<P2PNode>,
     close_group_size: usize,
 ) -> Vec<XorName> {
-    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).await;
+    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<ChunkStore>,
-    p2p_node: &Arc<P2PNode>,
-    close_group_size: usize,
-) -> Vec<SentReplicaHint> {
-    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();
-        }
-    };
-
-    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::<HashSet<_>>();
-        if close_peers.contains(peer) {
-            hints.push(SentReplicaHint { key, close_peers });
-        }
-    }
-    hints
+/// The ranges where one of `targets` can be in the close group, with their bounds.
+fn ranges_holding<'a>(
+    groups: &'a CloseGroupMap,
+    targets: &HashSet<PeerId>,
+) -> (Vec<&'a GroupRange>, Vec<(XorName, XorName)>) {
+    let ranges: Vec<&GroupRange> = groups
+        .ranges()
+        .iter()
+        .filter(|range| groups.may_include_any(range, targets))
+        .collect();
+    let bounds = ranges.iter().map(|range| range.bounds()).collect();
+    (ranges, bounds)
 }
 
-/// 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.
-///
-/// 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<P2PNode>, group_size: usize) -> usize {
-    let self_xor: XorName = *p2p_node.peer_id().as_bytes();
-    let closest = p2p_node
-        .dht_manager()
-        .find_closest_nodes_local(&self_xor, group_size)
-        .await;
-    if closest.len() < group_size {
-        return XORNAME_BITS;
+/// 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<Vec<XorName>>,
+    targets: &HashSet<PeerId>,
+    mut hint: impl FnMut(XorName, Vec<PeerId>),
+) {
+    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);
+            }
+        }
     }
-    closest
-        .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
-    }
+/// Replica hints from the stored records for each of `targets`, and how many
+/// stored keys were read.
+async fn replica_hints(
+    groups: &CloseGroupMap,
+    storage: &ChunkStore,
+    targets: &HashSet<PeerId>,
+) -> (HashMap<PeerId, Vec<SentReplicaHint>>, 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<PeerId, Vec<SentReplicaHint>> = HashMap::new();
+    for_each_hinted_key(groups, ranges, keys, targets, |key, group| {
+        let close_peers: HashSet<PeerId> = 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)
 }
 
-/// The paid-list keys a hint build considers.
-async fn paid_hint_candidates(
+/// Paid hints from the paid list for each of `targets`, and how many entries
+/// were read.
+async fn paid_hints(
+    groups: &CloseGroupMap,
     paid_list: &PaidList,
-    p2p_node: &Arc<P2PNode>,
-    group_size: usize,
-) -> Result<Vec<XorName>> {
-    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)
+    targets: &HashSet<PeerId>,
+) -> Result<(HashMap<PeerId, Vec<XorName>>, 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<PeerId, Vec<XorName>> = HashMap::new();
+    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))
 }
 
-/// 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.
+/// 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<ChunkStore>,
@@ -163,7 +174,7 @@ pub(crate) async fn build_sync_hints_for_peers(
     paid_list_close_group_size: usize,
 ) -> HashMap<PeerId, PeerSyncHints> {
     let started = Instant::now();
-    let target_peers = peers.iter().copied().collect::<HashSet<_>>();
+    let targets = peers.iter().copied().collect::<HashSet<_>>();
     let mut hints_by_peer = peers
         .iter()
         .copied()
@@ -174,53 +185,36 @@ 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::<HashSet<_>>();
-        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;
+    // 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).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 =
-        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_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 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) {
+                    peer_hints.paid_hints = hints;
                 }
             }
+            keys_read
         }
-    }
+        Err(e) => {
+            warn!("Failed to read PaidForList for batch hint construction: {e}");
+            0
+        }
+    };
 
     let replica_hint_count = hints_by_peer
         .values()
@@ -235,22 +229,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,
         );
     }
 
@@ -267,26 +253,18 @@ pub async fn build_paid_hints_for_peer(
     p2p_node: &Arc<P2PNode>,
     paid_list_close_group_size: usize,
 ) -> Vec<XorName> {
-    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 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);
+    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).await {
+        Ok((mut hints, _)) => hints.remove(peer).unwrap_or_default(),
+        Err(e) => {
+            warn!("Failed to read PaidForList for hint construction: {e}");
+            Vec::new()
         }
     }
-    hints
 }
 
 /// Take a fresh snapshot of close neighbors for a new round-robin cycle.
@@ -569,24 +547,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::<Vec<_>>();
-    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,
@@ -618,46 +595,48 @@ 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 ------------------------------------------------
+    // -- hint builds ----------------------------------------------------------
 
     #[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")
-        };
-
+    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;
+        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 (ranges, bounds) = ranges_holding(&groups, &targets);
+        let read: Vec<Vec<XorName>> = bounds
+            .iter()
+            .map(|(low, high)| {
+                keys.iter()
+                    .copied()
+                    .filter(|key| low <= key && key <= high)
+                    .collect()
+            })
+            .collect();
         assert_eq!(
-            read(PAID_HINT_FULL_SCAN_ENTRIES),
-            vec![key_ending_in(FAR_KEY)]
+            read.iter().map(Vec::len).sum::<usize>(),
+            3,
+            "keys from 1000 on aren't read"
         );
-        assert!(
-            read(PAID_HINT_FULL_SCAN_ENTRIES + 1).is_empty(),
-            "above the limit only near keys are read"
+        let mut hinted = Vec::new();
+        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"
         );
     }
 
diff --git a/src/replication/paid_list.rs b/src/replication/paid_list.rs
index 827def6d..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_MARGIN_BITS};
+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,22 +43,30 @@ 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 `near_rank` 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 `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], near_rank: usize) -> usize {
+    if distances.len() < 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:
@@ -356,26 +364,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<Vec<XorName>> {
+    pub(crate) fn keys_in_ranges(
+        &self,
+        ranges: &[(XorName, XorName)],
+    ) -> Result<Vec<Vec<XorName>>> {
         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`,
@@ -525,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.
@@ -544,24 +563,80 @@ 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<XorName> {
+        let mut distances = vec![key_ending_in(1); 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);
+
+        distance[31] = 0b0000_0001;
+        assert_eq!(significant_bits(&distance), 1);
 
-        radius[31] = 0b0000_0001;
+        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(&[], NEAR_RANK), XORNAME_BITS);
+
+        let one_short = vec![key_ending_in(1); NEAR_RANK - 1];
         assert_eq!(
-            near_bits_for_radius(&radius),
-            1 + PAID_LIST_NEAR_MARGIN_BITS
+            near_bits_for_peer_distances(&one_short, NEAR_RANK),
+            XORNAME_BITS
         );
+    }
 
-        radius[30] = 0b0000_0100;
+    #[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_radius(&radius),
-            11 + PAID_LIST_NEAR_MARGIN_BITS
+            near_bits_for_peer_distances(&distances, NEAR_RANK),
+            FARTHEST_BITS
         );
+    }
+
+    #[tokio::test]
+    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 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");
 
-        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)), NEAR_RANK);
+        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]
@@ -593,15 +668,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(&center, 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]
@@ -756,7 +839,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];
@@ -767,7 +850,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 729d8f3e..aa08f8a6 100644
--- a/src/replication/pruning.rs
+++ b/src/replication/pruning.rs
@@ -76,19 +76,21 @@ 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;
 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,
     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_GROUPS, 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,
@@ -378,8 +380,13 @@ 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;
+        // 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);
 
         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 +508,6 @@ async fn record_prune_lookup_groups(
     p2p_node: &Arc<P2PNode>,
     config: &ReplicationConfig,
 ) -> RecordPruneLookupGroups {
-    // Closest first, so the strict group is the head of the retention group.
     let retention: Vec<PeerId> = p2p_node
         .dht_manager()
         .find_closest_nodes_local_with_self(key, storage_admission_width(config.close_group_size))
@@ -509,11 +515,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<PeerId>, 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<PeerId>,
+    close_group_size: usize,
+) -> RecordPruneLookupGroups {
+    let strict = retention.iter().take(close_group_size).copied().collect();
     RecordPruneLookupGroups { retention, strict }
 }
 
@@ -682,6 +697,22 @@ 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: 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<P2PNode>, 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<XorName> = p2p_node
+        .dht_manager()
+        .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_rank)
+}
+
 /// Evict `PaidForList` entries once the list has passed its threshold, and
 /// return how many entries it now holds and how many went.
 ///
@@ -1607,6 +1638,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<PeerId> = (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<PeerId> = (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<PeerId> = (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]
     }
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<Vec<XorName>> {
+        // Copied out first, as in `all_keys`, so no two of these locks are held at once.
+        let mut unservable: HashSet<XorName> = 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 83b16f85..c503262b 100644
--- a/tests/e2e/replication.rs
+++ b/tests/e2e/replication.rs
@@ -2025,16 +2025,15 @@ 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.
+/// 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() {
     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");
@@ -2046,8 +2045,7 @@ async fn a_neighbour_gets_a_paid_hint_for_a_key_far_from_this_node() {
         .await;
     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");
@@ -2070,10 +2068,118 @@ 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<P2PNode>,
+    keys: &[XorName],
+    peer: &PeerId,
+    group_size: usize,
+) -> Vec<XorName> {
+    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 = 5;
+    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<XorName> = (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
-/// 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() {