pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-13 09:21:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid querying the local validator in availability recovery (#2792)

Browse Source 
* guide: don't request availability data from ourselves

* add QueryAllChunks message

* implement QueryAllChunks

* remove unused relay_parent from StoreChunk

* test QueryAllChunks

* fast paths make short roads

* test early exit behavior
This commit is contained in:
Robert Habermeier
2021-04-01 15:57:41 +02:00
committed by GitHub
parent a960e2ff6d
commit 5da762e728
10 changed files with 375 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polkadot/node/network/availability-recovery/src/lib.rs
+48
View File
@@ -334,6 +334,34 @@ impl RequestChunksPhase {
params: &InteractionParams,
sender: &mut impl SubsystemSender,
) -> Result<AvailableData, RecoveryError> {
// First query the store for any chunks we've got.
{
let (tx, rx) = oneshot::channel();
sender.send_message(
AvailabilityStoreMessage::QueryAllChunks(params.candidate_hash, tx).into()
).await;
match rx.await {
Ok(chunks) => {
// This should either be length 1 or 0. If we had the whole data,
// we wouldn't have reached this stage.
let chunk_indices: Vec<_> = chunks.iter().map(|c| c.index).collect();
self.shuffling.retain(|i| !chunk_indices.contains(i));
for chunk in chunks {
self.received_chunks.insert(chunk.index, chunk);
}
}
Err(oneshot::Canceled) => {
tracing::warn!(
target: LOG_TARGET,
candidate_hash = ?params.candidate_hash,
"Failed to reach the availability store"
);
}
}
}
loop {
if self.is_unavailable(&params) {
tracing::debug!(
@@ -431,6 +459,26 @@ fn reconstructed_data_matches_root(
impl<S: SubsystemSender> Interaction<S> {
async fn run(mut self) -> Result<AvailableData, RecoveryError> {
// First just see if we have the data available locally.
{
let (tx, rx) = oneshot::channel();
self.sender.send_message(
AvailabilityStoreMessage::QueryAvailableData(self.params.candidate_hash, tx).into()
).await;
match rx.await {
Ok(Some(data)) => return Ok(data),
Ok(None) => {}
Err(oneshot::Canceled) => {
tracing::warn!(
target: LOG_TARGET,
candidate_hash = ?self.params.candidate_hash,
"Failed to reach the availability store",
)
}
}
}
loop {
// These only fail if we cannot reach the underlying subsystem, which case there is nothing
// meaningful we can do.
polkadot/node/network/availability-recovery/src/tests.rs
+173
View File
@@ -207,6 +207,46 @@ impl TestState {
);
}
async fn respond_to_available_data_query(
&self,
virtual_overseer: &mut VirtualOverseer,
with_data: bool,
) {
assert_matches!(
overseer_recv(virtual_overseer).await,
AllMessages::AvailabilityStore(
AvailabilityStoreMessage::QueryAvailableData(_, tx)
) => {
let _ = tx.send(if with_data {
Some(self.available_data.clone())
} else {
println!("SENDING NONE");
None
});
}
)
}
async fn respond_to_query_all_request(
&self,
virtual_overseer: &mut VirtualOverseer,
send_chunk: impl Fn(usize) -> bool
) {
assert_matches!(
overseer_recv(virtual_overseer).await,
AllMessages::AvailabilityStore(
AvailabilityStoreMessage::QueryAllChunks(_, tx)
) => {
let v = self.chunks.iter()
.filter(|c| send_chunk(c.index.0 as usize))
.cloned()
.collect();
let _ = tx.send(v);
}
)
}
async fn test_chunk_requests(
&self,
candidate_hash: CandidateHash,
@@ -430,6 +470,9 @@ fn availability_is_recovered_from_chunks_if_no_group_provided() {
let candidate_hash = test_state.candidate.hash();
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -459,6 +502,9 @@ fn availability_is_recovered_from_chunks_if_no_group_provided() {
test_state.test_runtime_api(&mut virtual_overseer).await;
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
new_candidate.hash(),
&mut virtual_overseer,
@@ -506,6 +552,9 @@ fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunk
let candidate_hash = test_state.candidate.hash();
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -535,6 +584,9 @@ fn availability_is_recovered_from_chunks_even_if_backing_group_supplied_if_chunk
test_state.test_runtime_api(&mut virtual_overseer).await;
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
new_candidate.hash(),
&mut virtual_overseer,
@@ -589,6 +641,9 @@ fn bad_merkle_path_leads_to_recovery_error() {
test_state.chunks[3].chunk = vec![3; 32];
test_state.chunks[4].chunk = vec![4; 32];
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -642,6 +697,9 @@ fn wrong_chunk_index_leads_to_recovery_error() {
test_state.chunks[3] = test_state.chunks[0].clone();
test_state.chunks[4] = test_state.chunks[0].clone();
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -705,6 +763,9 @@ fn invalid_erasure_coding_leads_to_invalid_error() {
test_state.test_runtime_api(&mut virtual_overseer).await;
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -757,6 +818,8 @@ fn fast_path_backing_group_recovers() {
_ => Has::No,
};
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.test_full_data_requests(
candidate_hash,
&mut virtual_overseer,
@@ -808,12 +871,17 @@ fn no_answers_in_fast_path_causes_chunk_requests() {
0 | 3 => Has::No,
_ => Has::timeout(),
};
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.test_full_data_requests(
candidate_hash,
&mut virtual_overseer,
who_has,
).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -905,6 +973,9 @@ fn chunks_retry_until_all_nodes_respond() {
let candidate_hash = test_state.candidate.hash();
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |_| false).await;
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
@@ -925,3 +996,105 @@ fn chunks_retry_until_all_nodes_respond() {
assert_eq!(rx.await.unwrap().unwrap_err(), RecoveryError::Unavailable);
});
}
#[test]
fn returns_early_if_we_have_the_data() {
let test_state = TestState::default();
test_harness_chunks_only(|test_harness| async move {
let TestHarness { mut virtual_overseer } = test_harness;
overseer_signal(
&mut virtual_overseer,
OverseerSignal::ActiveLeaves(ActiveLeavesUpdate {
activated: smallvec![ActivatedLeaf {
hash: test_state.current.clone(),
number: 1,
span: Arc::new(jaeger::Span::Disabled),
}],
deactivated: smallvec![],
}),
).await;
let (tx, rx) = oneshot::channel();
overseer_send(
&mut virtual_overseer,
AvailabilityRecoveryMessage::RecoverAvailableData(
test_state.candidate.clone(),
test_state.session_index,
None,
tx,
)
).await;
test_state.test_runtime_api(&mut virtual_overseer).await;
test_state.respond_to_available_data_query(&mut virtual_overseer, true).await;
assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
});
}
#[test]
fn does_not_query_local_validator() {
let test_state = TestState::default();
test_harness_chunks_only(|test_harness| async move {
let TestHarness { mut virtual_overseer } = test_harness;
overseer_signal(
&mut virtual_overseer,
OverseerSignal::ActiveLeaves(ActiveLeavesUpdate {
activated: smallvec![ActivatedLeaf {
hash: test_state.current.clone(),
number: 1,
span: Arc::new(jaeger::Span::Disabled),
}],
deactivated: smallvec![],
}),
).await;
let (tx, rx) = oneshot::channel();
overseer_send(
&mut virtual_overseer,
AvailabilityRecoveryMessage::RecoverAvailableData(
test_state.candidate.clone(),
test_state.session_index,
None,
tx,
)
).await;
test_state.test_runtime_api(&mut virtual_overseer).await;
test_state.respond_to_available_data_query(&mut virtual_overseer, false).await;
test_state.respond_to_query_all_request(&mut virtual_overseer, |i| i == 0).await;
let candidate_hash = test_state.candidate.hash();
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
test_state.validators.len(),
|i| if i == 0 {
panic!("requested from local validator")
} else {
Has::timeout()
},
).await;
// second round, make sure it uses the local chunk.
test_state.test_chunk_requests(
candidate_hash,
&mut virtual_overseer,
test_state.threshold() - 1,
|i| if i == 0 {
panic!("requested from local validator")
} else {
Has::Yes
},
).await;
assert_eq!(rx.await.unwrap().unwrap(), test_state.available_data);
});
}