diff --git a/zakurad/src/components/sync.rs b/zakurad/src/components/sync.rs index 26bb8232e..4a77977b6 100644 --- a/zakurad/src/components/sync.rs +++ b/zakurad/src/components/sync.rs @@ -1297,8 +1297,9 @@ where ); let tip_network = self.tip_network.clone(); + let state = self.state.clone(); let tips = std::mem::take(&mut self.prospective_tips); - extend = Some(Box::pin(Self::build_extend(tip_network, tips))); + extend = Some(Box::pin(Self::build_extend(tip_network, state, tips))); } // Dispatch from the reserve while we're below the lookahead limit. @@ -1493,6 +1494,101 @@ where Ok(()) } + async fn first_unknown_in_find_blocks_response( + &mut self, + hashes: &[block::Hash], + ) -> Result, Report> { + let mut seen = HashSet::new(); + let mut first_unknown = None; + + for (index, &hash) in hashes.iter().enumerate() { + if !seen.insert(hash) { + debug!(?hash, "discarding FindBlocks response with duplicate hash"); + return Ok(None); + } + + if first_unknown.is_none() { + if !self.state_contains(hash).await? { + first_unknown = Some(index); + } + continue; + } + + if self.state_contains(hash).await? { + debug!( + ?hash, + "discarding FindBlocks response with known hash in advertised suffix" + ); + return Ok(None); + } + } + + Ok(first_unknown) + } + + async fn validate_extend_find_blocks_response<'a>( + state: &mut ZS, + hashes: &'a [block::Hash], + locator: block::Hash, + expected_next: block::Hash, + ) -> Result, Report> { + let matched_hashes = match hashes { + [expected_hash, _rest @ ..] if *expected_hash == expected_next => hashes, + [first_hash, expected_hash, _rest @ ..] if *expected_hash == expected_next => { + debug!(?first_hash, ?expected_next, ?locator, "unexpected first hash, but the second matches: using the hashes after the match"); + &hashes[1..] + } + [] => return Ok(None), + [single_hash] => { + debug!( + ?single_hash, + ?expected_next, + ?locator, + "discarding response containing a single unexpected hash" + ); + return Ok(None); + } + [first_hash, second_hash, rest @ ..] => { + debug!(?first_hash, ?second_hash, rest_len = ?rest.len(), ?expected_next, ?locator, "discarding response that starts with two unexpected hashes"); + return Ok(None); + } + }; + + // We use the last hash for the tip, and we want to avoid bad tips. + // So we discard the last hash before checking duplicates or known hashes. + let matched_hashes = match matched_hashes { + [] => unreachable!("matched FindBlocks response contains the expected hash"), + [rest @ .., _last] => rest, + }; + + let mut seen = HashSet::new(); + if matched_hashes.iter().any(|hash| !seen.insert(*hash)) { + debug!( + ?matched_hashes, + "discarding FindBlocks extension response with duplicate hash" + ); + return Ok(None); + } + + let unknown_hashes = match matched_hashes { + [expected_hash, rest @ ..] if *expected_hash == expected_next => rest, + [] => return Ok(Some(&[])), + _ => unreachable!("matched FindBlocks response starts with the expected hash"), + }; + + for &hash in unknown_hashes { + if Self::state_contains_service(state, hash).await? { + debug!( + ?hash, + "discarding FindBlocks extension response with known hash in advertised suffix" + ); + return Ok(None); + } + } + + Ok(Some(unknown_hashes)) + } + /// Given a block_locator list fan out request for subsequent hashes to /// multiple peers #[instrument(skip(self))] @@ -1588,13 +1684,7 @@ where continue; } - let mut first_unknown = None; - for (i, &hash) in hashes.iter().enumerate() { - if !self.state_contains(hash).await? { - first_unknown = Some(i); - break; - } - } + let first_unknown = self.first_unknown_in_find_blocks_response(hashes).await?; debug!(hashes.len = ?hashes.len(), ?first_unknown); @@ -1685,6 +1775,7 @@ where #[instrument(skip_all)] async fn build_extend( mut tip_network: Timeout, + mut state: ZS, tips: HashSet, ) -> Result<(IndexSet, HashSet, usize), Report> { let stage_start = std::time::Instant::now(); @@ -1723,47 +1814,15 @@ where // zcashd sometimes appends an unrelated hash at the // start or end of its response. Check the first hash // against the previous response, and discard mismatches. - let unknown_hashes = match hashes.as_slice() { - [expected_hash, rest @ ..] if expected_hash == &tip.expected_next => { - rest - } - // If the first hash doesn't match, retry with the second. - [first_hash, expected_hash, rest @ ..] - if expected_hash == &tip.expected_next => - { - debug!(?first_hash, - ?tip.expected_next, - ?tip.tip, - "unexpected first hash, but the second matches: using the hashes after the match"); - rest - } - // We ignore these responses - [] => continue, - [single_hash] => { - debug!(?single_hash, - ?tip.expected_next, - ?tip.tip, - "discarding response containing a single unexpected hash"); - continue; - } - [first_hash, second_hash, rest @ ..] => { - debug!(?first_hash, - ?second_hash, - rest_len = ?rest.len(), - ?tip.expected_next, - ?tip.tip, - "discarding response that starts with two unexpected hashes"); - continue; - } - }; - - // We use the last hash for the tip, and we want to avoid - // bad tips. So we discard the last hash. (We don't need - // to worry about missed downloads, because we will pick - // them up again in the next ExtendTips.) - let unknown_hashes = match unknown_hashes { - [] => continue, - [rest @ .., _last] => rest, + let Some(unknown_hashes) = Self::validate_extend_find_blocks_response( + &mut state, + hashes.as_slice(), + tip.tip, + tip.expected_next, + ) + .await? + else { + continue; }; let new_tip = if let Some(end) = unknown_hashes.rchunks_exact(2).next() { @@ -2189,8 +2248,11 @@ where /// Returns `true` if the hash is present in the state, and `false` /// if the hash is not present in the state. pub(crate) async fn state_contains(&mut self, hash: block::Hash) -> Result { - match self - .state + Self::state_contains_service(&mut self.state, hash).await + } + + async fn state_contains_service(state: &mut ZS, hash: block::Hash) -> Result { + match state .ready() .await .map_err(|e| eyre!(e))? diff --git a/zakurad/src/components/sync/tests/vectors.rs b/zakurad/src/components/sync/tests/vectors.rs index 492d78681..c8ca3d4f0 100644 --- a/zakurad/src/components/sync/tests/vectors.rs +++ b/zakurad/src/components/sync/tests/vectors.rs @@ -165,11 +165,15 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> { block3_hash, // (discarded - last hash, possibly incorrect) ])); - // State is checked for the first unknown block (block 1) + // State is checked for each candidate hash before it is queued. state_service .expect_request(zs::Request::KnownBlock(block1_hash)) .await .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(block2_hash)) + .await + .respond(zs::Response::KnownBlock(None)); // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { @@ -251,6 +255,13 @@ async fn sync_blocks_ok() -> Result<(), crate::BoxError> { block5_hash, // (discarded - last hash, possibly incorrect) ])); + for hash in [block3_hash, block4_hash] { + state_service + .expect_request(zs::Request::KnownBlock(hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } + // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { peer_set @@ -384,6 +395,10 @@ async fn incomplete_checkpoint_range_refreshes_tips_without_verifier_timeout( .expect_request(zs::Request::KnownBlock(hashes[1])) .await .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(hashes[2])) + .await + .respond(zs::Response::KnownBlock(None)); for _ in 0..(sync::FANOUT - 1) { peer_set .expect_request(zn::Request::FindBlocks { @@ -472,6 +487,12 @@ async fn incomplete_checkpoint_range_refreshes_tips_without_verifier_timeout( .expect_request(zs::Request::KnownBlock(hashes[1])) .await .respond(zs::Response::KnownBlock(None)); + for hash in &hashes[2..=4] { + state_service + .expect_request(zs::Request::KnownBlock(*hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } for _ in 0..(sync::FANOUT - 1) { peer_set .expect_request(zn::Request::FindBlocks { @@ -546,11 +567,11 @@ async fn incomplete_checkpoint_range_refreshes_tips_without_verifier_timeout( } /// Test that the syncer downloads genesis, blocks 1-2 using obtain_tips, and blocks 3-4 using extend_tips, -/// with duplicate block hashes. +/// with unrelated trailing hashes that are discarded. /// /// This test also makes sure that the syncer downloads blocks in order. #[tokio::test] -async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> { +async fn sync_blocks_trailing_hashes_ok() -> Result<(), crate::BoxError> { // Get services let ( chain_sync_future, @@ -638,18 +659,20 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> { }) .await .respond(zn::Response::BlockHashes(vec![ - block1_hash, - block1_hash, block1_hash, // tip block2_hash, // expected_next block3_hash, // (discarded - last hash, possibly incorrect) ])); - // State is checked for the first unknown block (block 1) + // State is checked for each candidate hash before it is queued. state_service .expect_request(zs::Request::KnownBlock(block1_hash)) .await .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(block2_hash)) + .await + .respond(zs::Response::KnownBlock(None)); // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { @@ -728,11 +751,16 @@ async fn sync_blocks_duplicate_hashes_ok() -> Result<(), crate::BoxError> { block2_hash, // tip (discarded - already fetched) block3_hash, // expected_next block4_hash, - block3_hash, - block4_hash, block5_hash, // (discarded - last hash, possibly incorrect) ])); + for hash in [block3_hash, block4_hash] { + state_service + .expect_request(zs::Request::KnownBlock(hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } + // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { peer_set @@ -947,11 +975,19 @@ async fn sync_block_too_high_obtain_tips() -> Result<(), crate::BoxError> { block3_hash, // (discarded - last hash, possibly incorrect) ])); - // State is checked for the first unknown block (block 982k) + // State is checked for each candidate hash before it is queued. state_service .expect_request(zs::Request::KnownBlock(block982k_hash)) .await .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(block1_hash)) + .await + .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(block2_hash)) + .await + .respond(zs::Response::KnownBlock(None)); // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { @@ -1124,11 +1160,15 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> { block3_hash, // (discarded - last hash, possibly incorrect) ])); - // State is checked for the first unknown block (block 1) + // State is checked for each candidate hash before it is queued. state_service .expect_request(zs::Request::KnownBlock(block1_hash)) .await .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(block2_hash)) + .await + .respond(zs::Response::KnownBlock(None)); // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { @@ -1211,6 +1251,13 @@ async fn sync_block_too_high_extend_tips() -> Result<(), crate::BoxError> { block5_hash, // (discarded - last hash, possibly incorrect) ])); + for hash in [block3_hash, block4_hash, block982k_hash] { + state_service + .expect_request(zs::Request::KnownBlock(hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } + // Clear remaining block locator requests for _ in 0..(sync::FANOUT - 1) { peer_set @@ -1654,7 +1701,7 @@ async fn build_extend_discovers_hashes_without_dispatching() -> Result<(), crate _sync_status, mut block_verifier_router, mut peer_set, - _state_service, + mut state_service, _mock_chain_tip_sender, ) = setup_chain_sync(); @@ -1683,7 +1730,11 @@ async fn build_extend_discovers_hashes_without_dispatching() -> Result<(), crate // `build_extend` owns a clone of the tip network so it can run without borrowing `self`. let tip_network = Timeout::new(peer_set.clone(), sync::TIPS_RESPONSE_TIMEOUT); - let extend_handle = tokio::spawn(TestChainSync::build_extend(tip_network, tips)); + let extend_handle = tokio::spawn(TestChainSync::build_extend( + tip_network, + state_service.clone(), + tips, + )); // One peer extends the tip. The response starts with the expected hash (the match anchor) and // ends with a possibly-incorrect trailing hash that the syncer discards. @@ -1700,6 +1751,13 @@ async fn build_extend_discovers_hashes_without_dispatching() -> Result<(), crate block5_hash, // (discarded - last hash, possibly incorrect) ])); + for hash in [block3_hash, block4_hash] { + state_service + .expect_request(zs::Request::KnownBlock(hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } + // The remaining fan-out requests fail and are ignored. for _ in 0..(sync::FANOUT - 1) { peer_set @@ -1746,6 +1804,232 @@ async fn build_extend_discovers_hashes_without_dispatching() -> Result<(), crate Ok(()) } +#[tokio::test] +async fn obtain_tips_ignores_known_hash_after_first_unknown() -> Result<(), crate::BoxError> { + let ( + mut chain_sync, + _sync_status, + mut block_verifier_router, + mut peer_set, + mut state_service, + _mock_chain_tip_sender, + ) = setup_chain_sync(); + + let locator = block::Hash::from([0x01; 32]); + let unknown = block::Hash::from([0x02; 32]); + let known_not_in_locator = block::Hash::from([0x03; 32]); + let trailing = block::Hash::from([0x04; 32]); + + let respond_to_requests = async { + state_service + .expect_request(zs::Request::BlockLocator) + .await + .respond(zs::Response::BlockLocator(vec![locator])); + + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![locator], + stop: None, + }) + .await + .respond(zn::Response::BlockHashes(vec![ + unknown, + known_not_in_locator, + trailing, + ])); + + state_service + .expect_request(zs::Request::KnownBlock(unknown)) + .await + .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(known_not_in_locator)) + .await + .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain))); + + for _ in 0..(sync::FANOUT - 1) { + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![locator], + stop: None, + }) + .await + .respond(Err(zn::BoxError::from("synthetic test obtain tips error"))); + } + + Ok::<_, crate::BoxError>(()) + }; + + let (extra_hashes, responded) = futures::join!(chain_sync.obtain_tips(), respond_to_requests); + responded?; + let extra_hashes = extra_hashes?; + assert!(extra_hashes.is_empty()); + + peer_set.expect_no_requests().await; + block_verifier_router.expect_no_requests().await; + state_service.expect_no_requests().await; + + Ok(()) +} + +#[tokio::test] +async fn build_extend_ignores_malformed_find_blocks_responses() -> Result<(), crate::BoxError> { + let ( + _chain_sync, + _sync_status, + mut block_verifier_router, + mut peer_set, + mut state_service, + _mock_chain_tip_sender, + ) = setup_chain_sync(); + + let tip = block::Hash::from([0x10; 32]); + let expected_next = block::Hash::from([0x11; 32]); + let unknown = block::Hash::from([0x12; 32]); + let known_suffix = block::Hash::from([0x13; 32]); + let trailing = block::Hash::from([0x14; 32]); + let random = block::Hash::from([0x15; 32]); + + let tips = HashSet::from([sync::CheckedTip { tip, expected_next }]); + let extend_handle = tokio::spawn(TestChainSync::build_extend( + Timeout::new(peer_set.clone(), sync::TIPS_RESPONSE_TIMEOUT), + state_service.clone(), + tips, + )); + + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![tip], + stop: None, + }) + .await + .respond(zn::Response::BlockHashes(vec![ + expected_next, + unknown, + known_suffix, + trailing, + ])); + state_service + .expect_request(zs::Request::KnownBlock(unknown)) + .await + .respond(zs::Response::KnownBlock(None)); + state_service + .expect_request(zs::Request::KnownBlock(known_suffix)) + .await + .respond(zs::Response::KnownBlock(Some(zs::KnownBlock::BestChain))); + + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![tip], + stop: None, + }) + .await + .respond(zn::Response::BlockHashes(vec![ + expected_next, + unknown, + unknown, + trailing, + ])); + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![tip], + stop: None, + }) + .await + .respond(zn::Response::BlockHashes(vec![random])); + + let (download_set, prospective_tips, discovered) = extend_handle + .await + .expect("build_extend task should not panic")?; + assert!(download_set.is_empty()); + assert!(prospective_tips.is_empty()); + assert_eq!(discovered, 0); + + peer_set.expect_no_requests().await; + block_verifier_router.expect_no_requests().await; + state_service.expect_no_requests().await; + + Ok(()) +} + +#[tokio::test] +async fn build_extend_ignores_known_trailing_find_blocks_hash() -> Result<(), crate::BoxError> { + let ( + _chain_sync, + _sync_status, + mut block_verifier_router, + mut peer_set, + mut state_service, + _mock_chain_tip_sender, + ) = setup_chain_sync(); + + let tip = block::Hash::from([0x20; 32]); + let expected_next = block::Hash::from([0x21; 32]); + let unknown_a = block::Hash::from([0x22; 32]); + let unknown_b = block::Hash::from([0x23; 32]); + + let tips = HashSet::from([sync::CheckedTip { tip, expected_next }]); + let extend_handle = tokio::spawn(TestChainSync::build_extend( + Timeout::new(peer_set.clone(), sync::TIPS_RESPONSE_TIMEOUT), + state_service.clone(), + tips, + )); + + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![tip], + stop: None, + }) + .await + .respond(zn::Response::BlockHashes(vec![ + expected_next, + unknown_a, + unknown_b, + tip, // zcashd can append an unrelated known hash here. + ])); + + for hash in [unknown_a, unknown_b] { + state_service + .expect_request(zs::Request::KnownBlock(hash)) + .await + .respond(zs::Response::KnownBlock(None)); + } + + for _ in 0..(sync::FANOUT - 1) { + peer_set + .expect_request(zn::Request::FindBlocks { + known_blocks: vec![tip], + stop: None, + }) + .await + .respond(Err(zn::BoxError::from("synthetic test extend tips error"))); + } + + let (download_set, prospective_tips, discovered) = extend_handle + .await + .expect("build_extend task should not panic")?; + + assert_eq!( + download_set.into_iter().collect::>(), + vec![unknown_a, unknown_b], + "build_extend should keep valid inner hashes and discard the trailing known hash", + ); + assert_eq!(discovered, 2); + assert_eq!( + prospective_tips, + HashSet::from([sync::CheckedTip { + tip: unknown_a, + expected_next: unknown_b, + }]), + ); + + peer_set.expect_no_requests().await; + block_verifier_router.expect_no_requests().await; + state_service.expect_no_requests().await; + + Ok(()) +} + /// A registry miss (every ready peer marked missing the block) within budget schedules a backoff /// retry instead of blocking the loop or restarting the round, and does not re-request the block /// inline — the retry is deferred to the sync loop's timer arm so peers can drain meanwhile.