fix(replication): drain priority sync queue and recover from routing-event lag#165
Open
mickvandijke wants to merge 18 commits into
Open
fix(replication): drain priority sync queue and recover from routing-event lag#165mickvandijke wants to merge 18 commits into
mickvandijke wants to merge 18 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 { |
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, 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, or left holding LMDB/P2P resources after engine shutdown.
The branch contains eighteen 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 hintsProblems 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.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
Audit concurrency and observability
AuditChallengeCoordinatoracross responsible, prune-confirmation, and possession audits.Event-stream and detached-task lifecycle
Commit Details For Today's Follow-ups
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
Expected Effect
SemVer
Patch.