Copy db consistency check to reth-optimism-storage

[emhane]

Describe the feature

Copy check to reth-optimism-storage to remove optimism feature gates. Somehow needs to be injectable via NodeTypes::Storage.

reth/crates/storage/provider/src/providers/static_file/manager.rs

Lines 627 to 790 in ef033ab

	/// Ensures that any broken invariants which cannot be healed on the spot return a pipeline
	/// target to unwind to.
	///
	/// Two types of consistency checks are done for:
	///
	/// 1) When a static file fails to commit but the underlying data was changed.
	/// 2) When a static file was committed, but the required database transaction was not.
	///
	/// For 1) it can self-heal if `self.access.is_read_only()` is set to `false`. Otherwise, it
	/// will return an error.
	/// For 2) the invariants below are checked, and if broken, might require a pipeline unwind
	/// to heal.
	///
	/// For each static file segment:
	/// * the corresponding database table should overlap or have continuity in their keys
	/// ([`TxNumber`] or [`BlockNumber`]).
	/// * its highest block should match the stage checkpoint block number if it's equal or higher
	/// than the corresponding database table last entry.
	///
	/// Returns a [`Option`] of [`PipelineTarget::Unwind`] if any healing is further required.
	///
	/// WARNING: No static file writer should be held before calling this function, otherwise it
	/// will deadlock.
	#[allow(clippy::while_let_loop)]
	pub fn check_consistency<Provider>(
	&self,
	provider: &Provider,
	has_receipt_pruning: bool,
	) -> ProviderResult<Option<PipelineTarget>>
	where
	Provider: DBProvider + BlockReader + StageCheckpointReader + ChainSpecProvider,
	{
	// OVM historical import is broken and does not work with this check. It's importing
	// duplicated receipts resulting in having more receipts than the expected transaction
	// range.
	//
	// If we detect an OVM import was done (block #1 <https://optimistic.etherscan.io/block/1>), skip it.
	// More on [#11099](https://github.com/paradigmxyz/reth/pull/11099).
	#[cfg(feature = "optimism")]
	if reth_chainspec::EthChainSpec::chain(&provider.chain_spec()) ==
	reth_chainspec::Chain::optimism_mainnet() &&
	provider
	.block_number(reth_optimism_primitives::bedrock::OVM_HEADER_1_HASH)?
	.is_some()
	{
	info!(target: "reth::cli",
	"Skipping storage verification for OP mainnet, expected inconsistency in OVM chain"
	);
	return Ok(None)
	}
	info!(target: "reth::cli", "Verifying storage consistency.");
	let mut unwind_target: Option<BlockNumber> = None;
	let mut update_unwind_target = |new_target: BlockNumber| {
	if let Some(target) = unwind_target.as_mut() {
	*target = (*target).min(new_target);
	} else {
	unwind_target = Some(new_target);
	}
	};
	for segment in StaticFileSegment::iter() {
	if has_receipt_pruning && segment.is_receipts() {
	// Pruned nodes (including full node) do not store receipts as static files.
	continue
	}
	let initial_highest_block = self.get_highest_static_file_block(segment);
	// File consistency is broken if:
	//
	// * appending data was interrupted before a config commit, then data file will be
	// truncated according to the config.
	//
	// * pruning data was interrupted before a config commit, then we have deleted data that
	// we are expected to still have. We need to check the Database and unwind everything
	// accordingly.
	if self.access.is_read_only() {
	self.check_segment_consistency(segment)?;
	} else {
	// Fetching the writer will attempt to heal any file level inconsistency.
	self.latest_writer(segment)?;
	}
	// Only applies to block-based static files. (Headers)
	//
	// The updated `highest_block` may have decreased if we healed from a pruning
	// interruption.
	let mut highest_block = self.get_highest_static_file_block(segment);
	if initial_highest_block != highest_block {
	info!(
	target: "reth::providers::static_file",
	?initial_highest_block,
	unwind_target = highest_block,
	?segment,
	"Setting unwind target."
	);
	update_unwind_target(highest_block.unwrap_or_default());
	}
	// Only applies to transaction-based static files. (Receipts & Transactions)
	//
	// Make sure the last transaction matches the last block from its indices, since a heal
	// from a pruning interruption might have decreased the number of transactions without
	// being able to update the last block of the static file segment.
	let highest_tx = self.get_highest_static_file_tx(segment);
	if let Some(highest_tx) = highest_tx {
	let mut last_block = highest_block.unwrap_or_default();
	loop {
	if let Some(indices) = provider.block_body_indices(last_block)? {
	if indices.last_tx_num() <= highest_tx {
	break
	}
	} else {
	// If the block body indices can not be found, then it means that static
	// files is ahead of database, and the `ensure_invariants` check will fix
	// it by comparing with stage checkpoints.
	break
	}
	if last_block == 0 {
	break
	}
	last_block -= 1;
	info!(
	target: "reth::providers::static_file",
	highest_block = self.get_highest_static_file_block(segment),
	unwind_target = last_block,
	?segment,
	"Setting unwind target."
	);
	highest_block = Some(last_block);
	update_unwind_target(last_block);
	}
	}
	if let Some(unwind) = match segment {
	StaticFileSegment::Headers => self.ensure_invariants::<_, tables::Headers>(
	provider,
	segment,
	highest_block,
	highest_block,
	)?,
	StaticFileSegment::Transactions => self
	.ensure_invariants::<_, tables::Transactions>(
	provider,
	segment,
	highest_tx,
	highest_block,
	)?,
	StaticFileSegment::Receipts => self.ensure_invariants::<_, tables::Receipts>(
	provider,
	segment,
	highest_tx,
	highest_block,
	)?,
	} {
	update_unwind_target(unwind);
	}
	}
	Ok(unwind_target.map(PipelineTarget::Unwind))
	}

Additional context

Ref #11242