fix: fix sync relayer collaboration

[driftluo]

What problem does this PR solve?

HeaderProcess uses the current snapshot to process the one-time message, During the entire processing process, the changes in the db itself are imperceptible. Normally, there is no problem here. But during this process, if a compact block message is received and processed successfully, the chain inserts a new block, and the block's header is in the previous header process message, there will be a concurrency problem.

I use a simple diagram to describe the time series problem here：

sync thread   |          process headers         |

relay thread    |  compact block process  |

The time series here is that the header process holds the old snapshot longer than the compact block process completion time.

This will cause the block status in the old snapshot to differ from the actual status.

ckb/sync/src/types/mod.rs

Lines 2315 to 2335 in f3efbf2

	pub fn get_block_status(&self, block_hash: &Byte32) -> BlockStatus {
	match self.shared().state().block_status_map.get(block_hash) {
	Some(status_ref) => *status_ref.value(),
	None => {
	if self.shared().state().header_map.contains_key(block_hash) {
	BlockStatus::HEADER_VALID
	} else {
	let verified = self
	.snapshot
	.get_block_ext(block_hash)
	.map(|block_ext| block_ext.verified);
	match verified {
	None => BlockStatus::UNKNOWN,
	Some(None) => BlockStatus::BLOCK_STORED,
	Some(Some(true)) => BlockStatus::BLOCK_VALID,
	Some(Some(false)) => BlockStatus::BLOCK_INVALID,
	}
	}
	}
	}
	}

The actual status should be BLOCK_VALID, but because the block cannot be seen in the old snapshot, the returned status will be UNKNOWN

ckb/sync/src/synchronizer/headers_process.rs

Lines 290 to 336 in f3efbf2

	if status.contains(BlockStatus::HEADER_VALID) {
	let header_index = shared
	.get_header_index_view(
	&self.header.hash(),
	status.contains(BlockStatus::BLOCK_STORED),
	)
	.expect("header with HEADER_VALID should exist")
	.as_header_index();
	state
	.peers()
	.may_set_best_known_header(self.peer, header_index);
	return result;
	}
	if self.prev_block_check(&mut result).is_err() {
	debug!(
	"HeadersProcess rejected invalid-parent header: {} {}",
	self.header.number(),
	self.header.hash(),
	);
	state.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
	return result;
	}
	if let Some(is_invalid) = self.non_contextual_check(&mut result).err() {
	debug!(
	"HeadersProcess rejected non-contextual header: {} {}",
	self.header.number(),
	self.header.hash(),
	);
	if is_invalid {
	state.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
	}
	return result;
	}
	if self.version_check(&mut result).is_err() {
	debug!(
	"HeadersProcess rejected invalid-version header: {} {}",
	self.header.number(),
	self.header.hash(),
	);
	state.insert_block_status(self.header.hash(), BlockStatus::BLOCK_INVALID);
	return result;
	}
	shared.insert_valid_header(self.peer, self.header);

The UNKNOWN status will cause the header to be reinserted into the header map, which will cause the block's status to become HEADER_VALID, the compact blocks received later will all be put into the orphan block pool due to this abnormal state. This will cause synchronization to pause until no compact block messages are received for a while. The sync protocol can then resume synchronization by re-downloading the abnormal block.

This PR forcefully reset the snapshot in get block status to obtain the latest status

Check List

Tests

• Unit test
• Integration test

Release note

Title Only: Include only the PR title in the release note.