fix(replication): drain priority sync queue and recover from routing-event lag#165
Open
mickvandijke wants to merge 23 commits into
Open
fix(replication): drain priority sync queue and recover from routing-event lag#165mickvandijke wants to merge 23 commits into
mickvandijke wants to merge 23 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR improves replication convergence under churn by (a) ensuring neighbor sync drains queued priority peers without waiting for periodic ticks and (b) recovering from lagged DHT routing-table broadcast events. It also expands verification/fetch pipeline behavior (batching caps, retry/defer semantics, paid-list edge-voter quorum handling) and adds an e2e scenario covering paid-list-authorized repair below storage quorum.
Changes:
- Neighbor-sync loop now drains the priority queue back-to-back and resyncs close peers on
RecvError::Lagged. - Verification/fetch pipeline enhancements: bounded verification batches, fetch→verification retry metadata, and deferred re-verification scheduling.
- Paid-list quorum evaluation updated with “edge voter” handling; adds a deterministic paid-list repair e2e scenario and bumps crate version.
Reviewed changes
Copilot reviewed 11 out of 12 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/poc_d1_bounded_queues.rs | Updates test VerificationEntry construction for new timing fields. |
| tests/poc_bootstrap_stall.rs | Updates test VerificationEntry construction for new timing fields. |
| tests/e2e/replication.rs | Adds deterministic e2e scenario for paid-list-majority repair under low storage quorum. |
| src/replication/types.rs | Adds next_verify_at, fetch retry metadata, and NeighborSyncState::has_priority_peers + test. |
| src/replication/scheduling.rs | Adds deferred pending scheduling, fetch retry→verification requeue, and returns evicted keys. |
| src/replication/quorum.rs | Adds edge-aware paid-list vote summary and splits verification requests into capped batches. |
| src/replication/mod.rs | Implements lag recovery, neighbor-sync drain-before-park, verification request caps, and retry/defer integration. |
| src/replication/config.rs | Introduces PAID_LIST_FLEX_EDGE_COUNT and MAX_VERIFICATION_KEYS_PER_REQUEST. |
| src/replication/bootstrap.rs | Updates test VerificationEntry construction for new timing fields. |
| src/replication/admission.rs | Adjusts cross-set dedup/admission so paid hints can survive replica rejection under churn. |
| Cargo.toml | Bumps version to 0.14.3. |
| Cargo.lock | Updates locked crate version to 0.14.3. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
2543
to
2547
| results.push(protocol::KeyVerificationResult { | ||
| key: *key, | ||
| present, | ||
| present: cached.present.unwrap_or(false), | ||
| paid, | ||
| }); |
| } | ||
| continue; | ||
| } | ||
| Err(RecvError::Closed) => continue, |
Comment on lines
+495
to
+497
| let handles = | ||
| spawn_verification_batch_tasks(targets, p2p_node, config.verification_request_timeout); | ||
| collect_verification_batch_results(handles, targets, &mut evidence).await; |
Comment on lines
+43
to
+47
| /// Number of furthest paid-list close-group peers treated as churny edge | ||
| /// voters. | ||
| /// | ||
| /// Once the paid-list close group reaches [`PAID_LIST_CLOSE_GROUP_SIZE`], edge | ||
| /// peers are queried, but a negative edge paid-list response does not count |
mickvandijke
force-pushed
the
fix/neighbor-sync-drain-window
branch
from
1a6fe63 to
fc8d724
Compare
Comment on lines
+1284
to
+1288
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; |
Comment on lines
+2837
to
2841
| if cached.present.is_none() && paid.is_none() { | ||
| continue; | ||
| } | ||
|
|
||
| results.push(protocol::KeyVerificationResult { |
Comment on lines
18
to
21
| pub mod audit; | ||
| pub mod audit_coordinator; | ||
| pub(crate) mod audit_metrics; | ||
| pub mod bootstrap; |
Comment on lines
+132
to
+134
| fn fresh_offer_key_lock_index(key: &XorName) -> usize { | ||
| usize::from(key[0]) % FRESH_OFFER_KEY_LOCK_SHARDS | ||
| } |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
SemVer: patch
…event lag Neighbor-sync paid-list hint propagation could stall for tens of minutes on the furthest close-group members during correlated topology change (partition heal, mass join). Two root causes: - The neighbor-sync loop parked on the periodic 10-20 min tick after every single round, and `sync_trigger` is a coalescing `Notify` that collapses a burst of entrant wakeups into one. A churn burst that queued many priority peers therefore drained only one batch (<=4) promptly; the rest waited for subsequent ticks. Park only when `priority_order` is empty and otherwise run rounds back-to-back, draining the durable queue at round-trip speed. The drain terminates because `select_next_sync_peer` pops each priority peer unconditionally and `new_cycle` only refills under `is_cycle_complete()`. - The DHT event handler discarded broadcast `RecvError::Lagged`, silently dropping `KClosestPeersChanged` events under load -- exactly when churn is heaviest -- so missed entrants were never queued. On lag, resynchronize from ground truth: snapshot the current close-peer set, queue every member for priority sync, and fire the trigger. Add `NeighborSyncState::has_priority_peers` and a regression test covering the loop's drain/termination contract. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SemVer: patch Consume dequeued fetch candidates through reservation-aware queue APIs and requeue no-source retry candidates for verification. Centralize pending_verify insertion bookkeeping so per-sender counters stay in lockstep.
SemVer: patch
mickvandijke
force-pushed
the
fix/neighbor-sync-drain-window
branch
from
74c990e to
b8d490d
Compare
Comment on lines
+631
to
+633
| self.insert_pending_unchecked(key, verification); | ||
| self.release_retry_slot(&sender); | ||
| true |
Comment on lines
+53
to
+54
| #[cfg(feature = "logging")] | ||
| impl AuditType { |
Comment on lines
+77
to
+78
| #[cfg(feature = "logging")] | ||
| impl AuditResponderClass { |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
Comment on lines
+204
to
+209
| if let Some(retry) = &mut candidate.retry_verification { | ||
| retry | ||
| .replica_hint_sources | ||
| .extend(entry.replica_hint_sources); | ||
| retry.hint_sources.extend(entry.hint_sources); | ||
| } |
Comment on lines
+222
to
+227
| if let Some(retry) = &mut in_flight.retry_verification { | ||
| retry | ||
| .replica_hint_sources | ||
| .extend(entry.replica_hint_sources); | ||
| retry.hint_sources.extend(entry.hint_sources); | ||
| } |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
A closed Tokio broadcast receiver is immediately ready with RecvError::Closed on every recv(), and both the P2P and DHT event branches responded by continuing the select! loop — spinning a core (and, on the P2P branch, flooding logs) until shutdown. A closed broadcast channel can never yield again, so treat Closed as terminal: handle_replication_event_recv_error now returns ControlFlow and both branches break the loop, which also drops replication_tx and cascades a clean shutdown of the serial handler. Also hoist mid-file imports flagged by the rust-conventions hook: the saorsa_pqc sig import moves to the top block and the tests module switches to `use super::*;` with module-qualified paths. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The broadcast-lag recovery path requeued the current close-peer snapshot but never called retain_sync_peers, unlike the normal KClosestPeersChanged path. Peers that left the close set during the lost event window stayed in priority_order ahead of genuine entrants, each burning a request timeout before current peers could sync. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Comment on lines
+110
to
+114
| enum PruneAuditChallengeResult { | ||
| Response(Box<ReplicationMessage>), | ||
| NoResponse(AuditFailureClass), | ||
| MalformedResponse, | ||
| } |
Comment on lines
+1183
to
+1187
| PruneAuditChallengeResult::Response(decoded) => *decoded, | ||
| PruneAuditChallengeResult::NoResponse(class) => { | ||
| // No response means an immediate audit failure, but keep the local | ||
| // class split so timeout metrics are not polluted by pre-delivery | ||
| // failures. |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
Comment on lines
+321
to
+326
| if let Some(keys) = self.pending_keys_by_source.get_mut(source) { | ||
| keys.remove(key); | ||
| if keys.is_empty() { | ||
| self.pending_keys_by_source.remove(source); | ||
| } | ||
| } |
Comment on lines
+115
to
+121
| let Some(entry) = targets.get_mut(&peer) else { | ||
| return; | ||
| }; | ||
| entry.references = entry.references.saturating_sub(1); | ||
| if entry.references == 0 { | ||
| targets.remove(&peer); | ||
| } |
Comment on lines
+53
to
+54
| #[cfg(feature = "logging")] | ||
| impl AuditType { |
Comment on lines
+77
to
+78
| #[cfg(feature = "logging")] | ||
| impl AuditResponderClass { |
Once all four worker permits were held, dispatch_fresh_offer handled the next offer inline — an on-chain payment verification and a multi-MiB LMDB write — on the serial non-audit loop. Stalling that loop backs up the 256-slot inbound queue, at which point the P2P event loop starts handling messages inline too, and the broadcast receiver lags and drops replication messages wholesale. The per-key shard locks made that easy to reach. The index was key[0] % 64, but a node only receives fresh offers for keys close to its own ID, so the accepted key set shares a long prefix and nearly every offer landed on one shard. Unrelated keys serialized behind a single mutex, keeping the four workers occupied and making the inline path the steady state rather than the exception. The ordering those locks preserved has no meaning here: the key is the content address of the data, so same key implies same bytes. Dispatch now only claims the key, takes an admission permit, and spawns. The worker permit is awaited inside the task, so handle_fresh_offer has a single caller and no inline path exists to fall back to. A new admission semaphore bounds outstanding offers — each holds its payload until it completes — at 16, a 64 MiB ceiling. The shard locks are replaced by an exact per-key in-flight set behind an RAII guard, so unrelated keys never contend and concurrent duplicates collapse onto the first claimant rather than repeating its verification. Behaviour change: past the admission bound an offer is refused rather than queued or handled inline. A refused offer is not stored and is therefore penalized as absent by the delayed possession check, the same as any other declined replica. The previous behaviour was not penalty-free either — a stalled loop drops offers through broadcast lag, with the same result and less predictability. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A replica hint is a possession claim made by the sending peer. The receiver treated it as authorization: `HintPipeline::Replica` skipped the `is_responsible(storage_admission_width)` check that `PaidOnly` keys had to pass, so whoever labelled the hint decided what we store. Two messages were enough to exploit it. A paid hint for key K admits at `paid_list_close_group_size` (20) and lands in `pending_verify` as `PaidOnly`. A second message re-advertising K as a replica hint hits the `already_pending` fast path in `admit_hints`, which skips admission checks for keys already queued, and reaches `add_pending_verify` as `Replica` — escalating the live entry. Both fetch paths then honoured the tag and downloaded without ever asking whether this node was in the top-9 for K. Nodes ranked 10-20 could be conscripted into fetching and storing any key an attacker could name; pruning reclaims it, but the bandwidth and disk pressure are spent, and honest routing skew triggers the same path. Derive `pipeline` from `replica_hint_sources` instead of storing it. The tag *is* "did any peer claim to hold this", so a stored copy could only drift from its own definition — `remove_hint_source` already recomputed it on peer departure. Deriving deletes the escalation and both demotion sites. Then gate downloads where the decision is actually made, on live routing state, in both places the pipelines diverged: the local paid-list fast path and the post-verification path. `pipeline` keeps its real jobs — selecting fetch sources and scoping sole-source trust penalties — and loses the one it should never have had. This narrows replica repair from the sender's view to our own: a node with a transiently skewed routing table now declines to repair a key it ranks outside the top-9. That matches pruning, which already evicts at the same width, so the fetch would have been undone anyway. Fresh PUTs keep their wider accept window (see `handle_fresh_replication_offer`), where rejecting risks losing data a client is writing now. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Admission asked two different questions depending on which list the sender put a key on: replica hints were gated at storage_admission_width (9), paid hints at paid_list_close_group_size (20). The sender chose which gate applied to its own hint. Ask one question instead. Admission decides only relevance -- should we learn this key exists and is paid for -- which is the 20-wide paid close group, since that is the width across which nodes track payment validity. Storage responsibility stays where the previous commit put it: at the point of download, against live routing state. Mislabelling a hint now gains an attacker nothing, because both labels reach the same gate. The hint set only records whether the sender claimed possession, which makes it a candidate fetch source. The set of admissible keys is unchanged -- a paid hint for any key a replica hint could carry was already admissible at 20 -- so this widens no attack surface. It does recover information the old gate discarded: a replica hint for a key we rank 10-20 for was rejected outright, even though that band exists precisely to track payment validity. The two-gate dance this removes (rejected_replica rescued by admitted_paid) existed only because the gates disagreed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… heap A hint for a key already in the fetch queue only needs its advertiser merged into that candidate's source set -- a field the heap never orders on. The old path took the whole heap, scanned it linearly for the key, and re-heapified, costing O(n) per duplicate. Under source aggregation a batch of m duplicates ran O(m*n) while holding the global queue write lock, which a neighbor could trigger deliberately by re-hinting keys it knows are queued. Split FetchCandidate into FetchOrder (key + distance, the only fields the Ord impl reads) held in the heap, and FetchPayload (sources + retry metadata) held in a key-indexed map that also serves as the membership index. Merging a source is now an O(1) map lookup that never touches the heap; the ordering invariant is immutable-by-construction rather than restored by a rebuild. The departed-peer path edits payloads in place and rebuilds only when a peer actually orphaned a candidate. Measured per-key merge cost goes from 302us at a 50k-deep queue to a flat ~400ns independent of depth. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Comment on lines
+153
to
157
| // Cross-set precedence: the replica pass already ruled on this key, | ||
| // under the same gate this pass would apply. Re-asking cannot change | ||
| // the answer, and re-recording it would double-count the rejection. | ||
| if seen_replica.contains(&key) { | ||
| continue; |
Comment on lines
+1283
to
+1289
| let Some(_slot) = audit_challenge_coordinator.acquire(*peer).await else { | ||
| warn!( | ||
| "Prune audit challenge for {} against {peer} could not acquire coordinator slot", | ||
| hex::encode(key) | ||
| ); | ||
| return PruneAuditChallengeResult::MalformedResponse; | ||
| }; |
ReplicationEngine::shutdown() promises that no background work still holds the LMDB environment when it returns, but several paths race a select! on the shutdown token against futures awaiting a spawn_blocking LMDB transaction (fetch storage.put, prune storage.delete / paid_list.remove_batch, verification paid_list.insert). Dropping the losing future does not cancel spawn_blocking: the closure keeps running with a cloned Env, so reopening the same environment after shutdown was undefined behavior. Track the blocking tasks at the storage layer instead of shielding call sites: LmdbStorage and PaidList each route every spawn_blocking through their own TaskTracker and expose wait_idle(); shutdown() awaits both after its existing detached-task drain. Per-fetch tasks are additionally spawned on the detached tracker so a dropped in_flight set can no longer leak Arc<LmdbStorage> past shutdown. Cancellation semantics are unchanged — network I/O still aborts promptly; only the bounded in-flight transaction is awaited. Regression coverage: storage- and paid-list-level wait_idle tests park a write inside its blocking closure and drop the awaiter; a new poc_shutdown_lmdb_drain suite proves shutdown() blocks until the detached write commits and both environments reopen cleanly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR hardens replication repair under churn, load, queue pressure, event-stream failure, and shutdown. It expands the original neighbor-sync drain fix into a broader set of paid-list repair, verification, bootstrap accounting, audit concurrency, fresh-offer dispatch, admission gating, and async lifecycle fixes so legitimate repair work is not silently dropped, converted into false audit failures, delayed behind stale peers, left spinning on closed broadcasts, serialized behind a saturated worker pool and dropped through broadcast lag, used to conscript out-of-range nodes into storing arbitrary keys, or left holding LMDB/P2P resources — including detached LMDB blocking transactions — after engine shutdown.
The branch contains twenty-three commits:
19bac10-fix(replication): harden paid-list verification repair0a5f078-fix(replication): tolerate paid-list edge churn2576f7a-fix(replication): drain priority sync queue and recover from routing-event lag95a8cdf-fix(replication): preserve verification retry capacity0c7bd9a-fix(replication): eliminate false audit challenge timeouts3800f9a-fix(replication): preserve dequeued retry reservations95fb3d9-chore(replication): fix audit admission clippy6ccb9bf-fix(replication): release cancelled async workb8d490d-fix(replication): silence no-logging audit label warnings77aa87c-fix(replication): unblock bootstrap when rejected peer leavesc896fa4-refactor(replication): prioritize source-aware hints5accdfb-refactor(replication): aggregate bootstrap hint batches1545551-fix(replication): aggregate verification per peer6b701c6-fix(replication): drain fresh offers through LMDB writesd2dc3f5-fix(replication): track detached audit workf81d28e-fix(replication): stop message handler when event streams close4859608-fix(replication): prune departed peers during DHT lag recovery50f6d7e-fix(replication): penalize rejected singleton replica hintsb251ba2-fix(replication): keep fresh offers off the serial message loopde8dcd7-fix(replication): gate replica downloads on storage responsibilitya0d63ae-refactor(replication): admit hints through a single relevance gate3d83004-perf(replication): merge duplicate hints without rebuilding the fetch heap52d19ef-fix(replication): drain detached LMDB blocking ops on shutdownProblems Addressed
Notifycoalescing allowed queued priority peers to wait for later 10-20 minute periodic ticks, while lagged DHT broadcasts could hide entrants entirely or leave departed peers ahead of current neighbors, each consuming a request timeout.spawn_blockingwaiter does not cancel the blocking transaction, so shutdown could report drained while LMDB still owned the environment.RecvError::Closedcould consume a core, flood P2P warnings, and prevent the replication pipeline from shutting down cleanly.key[0] % 64) collapsed the close-prefix accepted-key set onto a single shard, making the inline path the steady state, backing up the 256-slot inbound queue, and dropping replication messages wholesale once the P2P event loop lagged its broadcast receiver.HintPipeline::Replicaskipped theis_responsible(storage_admission_width)gate that paid keys had to pass, and a stored pipeline tag let a second replica message escalate a queued paid entry through thealready_pendingfast path, so a peer could name any key and force nodes ranked 10-20 for it to fetch and store it.select!on the shutdown token against futures awaiting aspawn_blockingLMDB transaction — the fetch worker'sstorage.put, the prune pass'sstorage.delete/paid_list.remove_batch, the verification worker'spaid_list.insert. Dropping the losing future does not cancel the blocking closure, which keeps running with a clonedEnv; per-fetch tasks were also bare-spawned outside both trackers, so a droppedin_flightset could leakArc<LmdbStorage>pastshutdown(). Reopening the same environment afterwards was undefined behavior.Changes
Paid-list verification and repair
pending_verify, fetch queue, and in-flight fetch state.Paid-list edge churn
Neighbor sync and bootstrap lifecycle
Source-aware bounded verification
Fresh-offer dispatch, admission gating, and fetch-queue merges
handle_fresh_offerhas a single caller and no inline verification/LMDB path on the serial loop.key[0] % 64shard locks with an exact per-key in-flight set behind an RAII guard, so unrelated keys never contend and concurrent duplicates collapse onto the first claimant instead of repeating its verification.replica_hint_sourcesinstead of storing the tag, deleting the paid→replica escalation and both demotion sites.is_responsible(storage_admission_width)at both fetch sites (local paid-list fast path and post-verification path), matching pruning width; fresh PUTs keep their wider accept window.paid_list_close_group_size(20), so mislabelling a hint gains nothing and the two-gate rescue dance (rejected_replicarescued byadmitted_paid) is removed; the admissible key set is unchanged.FetchCandidateintoFetchOrder(key + distance, the only fields theOrdimpl reads) held in the heap andFetchPayload(sources + retry metadata) held in a key-indexed map that also serves as the membership index, making a duplicate-source merge an O(1) map lookup and rebuilding the heap only when a departed peer actually orphans a candidate.Audit concurrency and observability
AuditChallengeCoordinatoracross responsible, prune-confirmation, and possession audits.Event-stream and detached-task lifecycle
LmdbStorageandPaidListroute everyspawn_blockingthrough per-instanceTaskTrackers and exposewait_idle(), so a dropped async awaiter no longer untracks a live transaction (constructor-time opens stay untracked — they cannot outlive the constructor).shutdown()awaitswait_idle()on both environments after the detached-task drain, strengthening its contract: when it returns, no LMDB blocking operation is still running and no engine-spawned task holdsArc<LmdbStorage>/Arc<PaidList>.tokio::spawn; thein_flightplumbing and prompt network-I/O cancellation are unchanged — only the bounded in-flight LMDB transaction is awaited.Latest Commit Details
b251ba2- keep fresh offers off the serial message loopOnce all four worker permits were held,
dispatch_fresh_offerhandled the next offer — an on-chain payment verification and a multi-MiB LMDB write — inline on the serial non-audit loop, backing up the 256-slot inbound queue until the P2P event loop also handled messages inline and its broadcast receiver lagged and dropped replication messages wholesale. The per-key shard locks (key[0] % 64) made that the steady state rather than the exception: a node only receives fresh offers for keys close to its own ID, so the accepted key set shares a long prefix and nearly every offer landed on one shard. The ordering those locks preserved has no meaning here, since the key is the content address of the data. Dispatch now only claims the key, takes an admission permit, and spawns; the worker permit is awaited inside the task sohandle_fresh_offerhas a single caller and no inline fallback exists. A 16-permit admission semaphore bounds outstanding offers at a 64 MiB ceiling, and an exact per-key in-flight set behind an RAII guard replaces the shard locks so unrelated keys never contend and concurrent duplicates collapse onto the first claimant. Past the bound an offer is refused — not stored, and therefore penalized as absent by the delayed possession check — rather than queued or handled inline.de8dcd7- gate replica downloads on storage responsibilityA replica hint is a possession claim, but the receiver treated it as authorization:
HintPipeline::Replicaskipped theis_responsible(storage_admission_width)check thatPaidOnlykeys had to pass, so whoever labelled the hint decided what we store. Two messages were enough to exploit it — a paid hint queued key K asPaidOnly, then a replica re-advertisement hit thealready_pendingfast path inadmit_hintsand escalated the live entry toReplica, and both fetch paths downloaded without ever asking whether this node was in the top-9 for K. The tag is now derived fromreplica_hint_sourcesinstead of stored (the tag is "did any peer claim to hold this"), which deletes the escalation and both demotion sites, and downloads are gated on live routing state at both diverging fetch sites. This narrows replica repair from the sender's view to our own — a transiently skewed routing table now declines to repair a key it ranks outside the top-9, matching pruning, which already evicts at the same width — while fresh PUTs keep their wider accept window.a0d63ae- admit hints through a single relevance gateAdmission previously asked two questions depending on the sender's label — replica hints gated at
storage_admission_width(9), paid hints atpaid_list_close_group_size(20) — letting the sender choose its own gate. Admission now decides only relevance (should we learn this key exists and is paid for) at the 20-wide paid close group, since that is the width across which nodes track payment validity; storage responsibility stays at download time against live routing state. Mislabelling a hint gains an attacker nothing because both labels reach the same gate; the admissible key set is unchanged (a paid hint for any key a replica hint could carry was already admissible at 20), and the two-gate rescue dance is removed. It does recover information the old gate discarded: a replica hint for a key we rank 10-20 for was previously rejected outright even though that band exists precisely to track payment validity.3d83004- merge duplicate hints without rebuilding the fetch heapA hint for a key already in the fetch queue only needs its advertiser merged into that candidate's source set — a field the heap never orders on — but the old path took the whole heap, scanned it linearly for the key, and re-heapified, costing O(n) per duplicate; under source aggregation a batch of m duplicates ran O(m·n) while holding the global queue write lock, which a neighbor could trigger by re-hinting queued keys.
FetchCandidateis split intoFetchOrder(key + distance, the only fields theOrdimpl reads) held in the heap andFetchPayload(sources + retry metadata) held in a key-indexed map that also serves as the membership index; merging a source is now an O(1) map lookup that never touches the heap, and the departed-peer path edits payloads in place and rebuilds only when a peer actually orphaned a candidate. Measured per-key merge cost goes from 302µs at a 50k-deep queue to a flat ~400ns independent of depth.52d19ef- drain detached LMDB blocking ops on shutdownReplicationEngine::shutdown()promises that when it returns no background work still holds the LMDB environment, but several paths race aselect!on the shutdown token against futures awaiting aspawn_blockingLMDB transaction — the fetch worker'sstorage.put, the neighbor-sync prune pass'sstorage.deleteandpaid_list.remove_batch(thePaidListis a second LMDB environment with the same pattern), and the verification worker'spaid_list.insert. Dropping the losing future does not cancelspawn_blocking: the closure keeps running on the blocking pool with a clonedEnv, so reopening the same environment after shutdown was undefined behavior. Rather than shielding every call site, the blocking tasks are now tracked at the storage layer:LmdbStorageandPaidListroute everyspawn_blockingthrough their ownTaskTrackerand exposewait_idle(), whichshutdown()awaits after its existing detached-task drain. Per-fetch tasks are additionally spawned on the detached tracker, so an aborted worker's droppedin_flightset can no longer leakArc<LmdbStorage>past shutdown. Cancellation semantics are unchanged — network I/O still aborts promptly, and shutdown waits only for the bounded in-flight transaction (milliseconds). Regression coverage parks a write inside its blocking closure, drops the awaiter, and proveswait_idle()/shutdown()block until it commits and both environments reopen cleanly.Preceding Follow-up Commit Details
f81d28e- stop the message handler when event streams closeMakes P2P lag remain recoverable but treats a closed P2P broadcast as terminal, matching the new terminal handling for a closed DHT broadcast. The replication loop now exits instead of spinning indefinitely on an immediately ready closed receiver; dropping its sender also lets the serial handler finish. The commit additionally hoists imports required by the Rust conventions hook.
4859608- prune departed peers during DHT lag recoveryBrings lag recovery in line with the normal
KClosestPeersChangedpath by retaining only the current close-peer set before requeueing its members. Stale peers from missed departure events no longer sit ahead of genuine entrants and burn one request timeout each.50f6d7e- penalize rejected singleton replica hintsPenalizes a sole replica advertiser when either the close group definitively rejects the key or that advertiser explicitly denies possessing it. This closes the free-replication abuse path while keeping inconclusive rounds without a direct contradiction, paid-only advertisements, and corroborated replica hints non-penalizing. Trust reports remain bounded per peer and verification cycle.
Earlier Follow-up Commit Details
77aa87c- unblock bootstrap when a rejected peer leavesRetires capacity-rejection debt on
PeerRemovedand immediately reruns bootstrap drain detection, so progress no longer depends on an unrelated later pipeline event.c896fa4- prioritise source-aware hintsReplaces the single hint sender with live source sets, preserves replica claimants separately, prioritises corroborated work, bounds verification cycles and network concurrency, and simplifies stale proof-of-concept tests around the production queue implementation.
5accdfb- aggregate bootstrap hint batchesMakes a bootstrap neighbor batch an atomic source-aggregation unit: sync requests run concurrently, response metadata is processed first, admitted hints are queued together, and verification waits until batch publication and drain accounting are complete.
1545551- aggregate verification per peerSends one bounded request per target peer for the cycle, aligns the incoming limit with the cycle bound, and rejects oversized batches rather than processing a misleading prefix.
6b701c6- drain fresh offers through LMDB writesRemoves cancellation around a started fresh-offer handler and makes its tracker drain unconditional, preventing a dropped async waiter from detaching a live blocking LMDB transaction.
d2dc3f5- track detached audit workGeneralises the fresh-offer tracker into a shared detached-task lifecycle barrier and applies it to storage/P2P-capable audit responders, delayed possession checks, and audit launches.
Coverage Added Or Updated
wait_idleregression tests: a write parked inside its blocking closure with a dropped awaiter keepswait_idleblocked, commits after release, and leaves the store usable.poc_shutdown_lmdb_drain):shutdown()blocks until a detached LMDB write commits, after which both environments reopen cleanly.Expected Effect
shutdown()returns, no LMDB blocking operation is still running on either environment and no engine-spawned task holds the storage, so the same LMDB files can be reopened safely.SemVer
Patch.