diff --git a/client/api/src/backend.rs b/client/api/src/backend.rs index f6a5583bcb..8a0406883e 100644 --- a/client/api/src/backend.rs +++ b/client/api/src/backend.rs @@ -52,6 +52,9 @@ pub trait Backend: Send + Sync { fn is_indexed(&self) -> bool { self.log_indexer().is_indexed() } + + /// Get the hash of the latest substrate block fully indexed by the backend. + async fn latest_block_hash(&self) -> Result; } #[derive(Debug, Eq, PartialEq)] diff --git a/client/cli/src/frontier_db_cmd/mapping_db.rs b/client/cli/src/frontier_db_cmd/mapping_db.rs index 3e2b64fb4f..fc6b269caf 100644 --- a/client/cli/src/frontier_db_cmd/mapping_db.rs +++ b/client/cli/src/frontier_db_cmd/mapping_db.rs @@ -40,22 +40,21 @@ pub enum MappingKey { EthBlockOrTransactionHash(H256), } -pub struct MappingDb<'a, C, B: BlockT> { +pub struct MappingDb<'a, B: BlockT, C> { cmd: &'a FrontierDbCmd, client: Arc, - backend: Arc>, + backend: Arc>, } -impl<'a, C, B: BlockT> MappingDb<'a, C, B> +impl<'a, B: BlockT, C> MappingDb<'a, B, C> where - C: ProvideRuntimeApi, + C: HeaderBackend + ProvideRuntimeApi, C::Api: EthereumRuntimeRPCApi, - C: HeaderBackend, { pub fn new( cmd: &'a FrontierDbCmd, client: Arc, - backend: Arc>, + backend: Arc>, ) -> Self { Self { cmd, @@ -176,4 +175,4 @@ where } } -impl<'a, C, B: BlockT> FrontierDbMessage for MappingDb<'a, C, B> {} +impl<'a, B: BlockT, C: HeaderBackend> FrontierDbMessage for MappingDb<'a, B, C> {} diff --git a/client/cli/src/frontier_db_cmd/meta_db.rs b/client/cli/src/frontier_db_cmd/meta_db.rs index a37024c773..c94716c7fa 100644 --- a/client/cli/src/frontier_db_cmd/meta_db.rs +++ b/client/cli/src/frontier_db_cmd/meta_db.rs @@ -24,6 +24,7 @@ use std::{ use ethereum_types::H256; use serde::Deserialize; +use sp_blockchain::HeaderBackend; // Substrate use sp_runtime::traits::Block as BlockT; @@ -57,13 +58,13 @@ impl FromStr for MetaKey { } } -pub struct MetaDb<'a, B: BlockT> { +pub struct MetaDb<'a, B: BlockT, C> { cmd: &'a FrontierDbCmd, - backend: Arc>, + backend: Arc>, } -impl<'a, B: BlockT> MetaDb<'a, B> { - pub fn new(cmd: &'a FrontierDbCmd, backend: Arc>) -> Self { +impl<'a, B: BlockT, C: HeaderBackend> MetaDb<'a, B, C> { + pub fn new(cmd: &'a FrontierDbCmd, backend: Arc>) -> Self { Self { cmd, backend } } @@ -151,4 +152,4 @@ impl<'a, B: BlockT> MetaDb<'a, B> { } } -impl<'a, B: BlockT> FrontierDbMessage for MetaDb<'a, B> {} +impl<'a, B: BlockT, C: HeaderBackend> FrontierDbMessage for MetaDb<'a, B, C> {} diff --git a/client/cli/src/frontier_db_cmd/mod.rs b/client/cli/src/frontier_db_cmd/mod.rs index 16c5403d71..a88194e714 100644 --- a/client/cli/src/frontier_db_cmd/mod.rs +++ b/client/cli/src/frontier_db_cmd/mod.rs @@ -98,15 +98,14 @@ pub enum DbValue { } impl FrontierDbCmd { - pub fn run( + pub fn run( &self, client: Arc, - backend: Arc>, + backend: Arc>, ) -> sc_cli::Result<()> where - C: ProvideRuntimeApi, + C: HeaderBackend + ProvideRuntimeApi, C::Api: fp_rpc::EthereumRuntimeRPCApi, - C: HeaderBackend, { match self.column { Column::Meta => { diff --git a/client/cli/src/frontier_db_cmd/tests.rs b/client/cli/src/frontier_db_cmd/tests.rs index 54854db948..e02c33a1bd 100644 --- a/client/cli/src/frontier_db_cmd/tests.rs +++ b/client/cli/src/frontier_db_cmd/tests.rs @@ -49,8 +49,8 @@ type OpaqueBlock = pub fn open_frontier_backend>( client: Arc, path: PathBuf, -) -> Result>, String> { - Ok(Arc::new(fc_db::kv::Backend::::new( +) -> Result>, String> { + Ok(Arc::new(fc_db::kv::Backend::::new( client, &fc_db::kv::DatabaseSettings { source: sc_client_db::DatabaseSource::RocksDb { diff --git a/client/db/src/kv/mod.rs b/client/db/src/kv/mod.rs index 9ccbaf933e..47f85704dd 100644 --- a/client/db/src/kv/mod.rs +++ b/client/db/src/kv/mod.rs @@ -62,14 +62,15 @@ pub mod static_keys { } #[derive(Clone)] -pub struct Backend { +pub struct Backend { + client: Arc, meta: Arc>, mapping: Arc>, log_indexer: LogIndexerBackend, } #[async_trait::async_trait] -impl fc_api::Backend for Backend { +impl> fc_api::Backend for Backend { async fn block_hash( &self, ethereum_block_hash: &H256, @@ -88,6 +89,10 @@ impl fc_api::Backend for Backend { fn log_indexer(&self) -> &dyn fc_api::LogIndexerBackend { &self.log_indexer } + + async fn latest_block_hash(&self) -> Result { + Ok(self.client.info().best_hash) + } } #[derive(Clone, Default)] @@ -115,8 +120,8 @@ pub fn frontier_database_dir(db_config_dir: &Path, db_path: &str) -> PathBuf { db_config_dir.join("frontier").join(db_path) } -impl Backend { - pub fn open>( +impl> Backend { + pub fn open( client: Arc, database: &DatabaseSource, db_config_dir: &Path, @@ -148,13 +153,11 @@ impl Backend { ) } - pub fn new>( - client: Arc, - config: &DatabaseSettings, - ) -> Result { - let db = utils::open_database::(client, config)?; + pub fn new(client: Arc, config: &DatabaseSettings) -> Result { + let db = utils::open_database::(client.clone(), config)?; Ok(Self { + client, mapping: Arc::new(MappingDb { db: db.clone(), write_lock: Arc::new(Mutex::new(())), diff --git a/client/db/src/kv/upgrade.rs b/client/db/src/kv/upgrade.rs index fd802e1727..3eb1af4775 100644 --- a/client/db/src/kv/upgrade.rs +++ b/client/db/src/kv/upgrade.rs @@ -348,8 +348,10 @@ mod tests { pub fn open_frontier_backend>( client: Arc, setting: &crate::kv::DatabaseSettings, - ) -> Result>, String> { - Ok(Arc::new(crate::kv::Backend::::new(client, setting)?)) + ) -> Result>, String> { + Ok(Arc::new(crate::kv::Backend::::new( + client, setting, + )?)) } #[cfg_attr(not(feature = "rocksdb"), ignore)] diff --git a/client/db/src/lib.rs b/client/db/src/lib.rs index cabc1a30d2..eea7f396df 100644 --- a/client/db/src/lib.rs +++ b/client/db/src/lib.rs @@ -18,6 +18,8 @@ #![warn(unused_crate_dependencies)] +use std::sync::Arc; + // Substrate pub use sc_client_db::DatabaseSource; use sp_runtime::traits::Block as BlockT; @@ -27,8 +29,8 @@ pub mod kv; pub mod sql; #[derive(Clone)] -pub enum Backend { - KeyValue(kv::Backend), +pub enum Backend { + KeyValue(Arc>), #[cfg(feature = "sql")] - Sql(sql::Backend), + Sql(Arc>), } diff --git a/client/db/src/sql/mod.rs b/client/db/src/sql/mod.rs index 73b0c8b6e9..5de8d96980 100644 --- a/client/db/src/sql/mod.rs +++ b/client/db/src/sql/mod.rs @@ -96,7 +96,6 @@ pub enum BackendConfig<'a> { pub struct Backend { /// The Sqlite connection. pool: SqlitePool, - /// The additional overrides for the logs handler. storage_override: Arc>, @@ -656,7 +655,7 @@ where } /// Retrieve the block hash for the last indexed canon block. - pub async fn get_last_indexed_canon_block(&self) -> Result { + pub async fn last_indexed_canon_block(&self) -> Result { let row = sqlx::query( "SELECT b.substrate_block_hash FROM blocks AS b INNER JOIN sync_status AS s @@ -823,6 +822,15 @@ impl> fc_api::Backend for Backend { fn log_indexer(&self) -> &dyn fc_api::LogIndexerBackend { self } + + async fn latest_block_hash(&self) -> Result { + // Retrieves the block hash for the latest indexed block, maybe it's not canon. + sqlx::query("SELECT substrate_block_hash FROM blocks ORDER BY block_number DESC LIMIT 1") + .fetch_one(self.pool()) + .await + .map(|row| H256::from_slice(&row.get::, _>(0)[..])) + .map_err(|e| format!("Failed to fetch best hash: {}", e)) + } } #[async_trait::async_trait] diff --git a/client/mapping-sync/src/kv/mod.rs b/client/mapping-sync/src/kv/mod.rs index 0ac044bcd2..d49f579977 100644 --- a/client/mapping-sync/src/kv/mod.rs +++ b/client/mapping-sync/src/kv/mod.rs @@ -37,9 +37,9 @@ use fp_rpc::EthereumRuntimeRPCApi; use crate::{EthereumBlockNotification, EthereumBlockNotificationSinks, SyncStrategy}; -pub fn sync_block( +pub fn sync_block>( storage_override: Arc>, - backend: &fc_db::kv::Backend, + backend: &fc_db::kv::Backend, header: &Block::Header, ) -> Result<(), String> { let substrate_block_hash = header.hash(); @@ -100,11 +100,11 @@ pub fn sync_block( pub fn sync_genesis_block( client: &C, - backend: &fc_db::kv::Backend, + backend: &fc_db::kv::Backend, header: &Block::Header, ) -> Result<(), String> where - C: ProvideRuntimeApi, + C: HeaderBackend + ProvideRuntimeApi, C::Api: EthereumRuntimeRPCApi, { let substrate_block_hash = header.hash(); @@ -148,7 +148,7 @@ pub fn sync_one_block( client: &C, substrate_backend: &BE, storage_override: Arc>, - frontier_backend: &fc_db::kv::Backend, + frontier_backend: &fc_db::kv::Backend, sync_from: ::Number, strategy: SyncStrategy, sync_oracle: Arc, @@ -237,7 +237,7 @@ pub fn sync_blocks( client: &C, substrate_backend: &BE, storage_override: Arc>, - frontier_backend: &fc_db::kv::Backend, + frontier_backend: &fc_db::kv::Backend, limit: usize, sync_from: ::Number, strategy: SyncStrategy, @@ -271,13 +271,14 @@ where Ok(synced_any) } -pub fn fetch_header( +pub fn fetch_header( substrate_backend: &BE, - frontier_backend: &fc_db::kv::Backend, + frontier_backend: &fc_db::kv::Backend, checking_tip: Block::Hash, sync_from: ::Number, ) -> Result, String> where + C: HeaderBackend, BE: HeaderBackend, { if frontier_backend.mapping().is_synced(&checking_tip)? { diff --git a/client/mapping-sync/src/kv/worker.rs b/client/mapping-sync/src/kv/worker.rs index 352e9c3b59..b9695382e1 100644 --- a/client/mapping-sync/src/kv/worker.rs +++ b/client/mapping-sync/src/kv/worker.rs @@ -47,7 +47,7 @@ pub struct MappingSyncWorker { client: Arc, substrate_backend: Arc, storage_override: Arc>, - frontier_backend: Arc>, + frontier_backend: Arc>, have_next: bool, retry_times: usize, @@ -68,7 +68,7 @@ impl MappingSyncWorker { client: Arc, substrate_backend: Arc, storage_override: Arc>, - frontier_backend: Arc>, + frontier_backend: Arc>, retry_times: usize, sync_from: ::Number, strategy: SyncStrategy, @@ -250,7 +250,7 @@ mod tests { let storage_override = Arc::new(SchemaV3StorageOverride::new(client.clone())); let frontier_backend = Arc::new( - fc_db::kv::Backend::::new( + fc_db::kv::Backend::::new( client.clone(), &fc_db::kv::DatabaseSettings { source: sc_client_db::DatabaseSource::RocksDb { @@ -392,7 +392,7 @@ mod tests { let storage_override = Arc::new(SchemaV3StorageOverride::new(client.clone())); let frontier_backend = Arc::new( - fc_db::kv::Backend::::new( + fc_db::kv::Backend::::new( client.clone(), &fc_db::kv::DatabaseSettings { source: sc_client_db::DatabaseSource::RocksDb { diff --git a/client/mapping-sync/src/sql/mod.rs b/client/mapping-sync/src/sql/mod.rs index b78885ad2a..86c5b2db80 100644 --- a/client/mapping-sync/src/sql/mod.rs +++ b/client/mapping-sync/src/sql/mod.rs @@ -89,7 +89,7 @@ where match cmd { WorkerCommand::ResumeSync => { // Attempt to resume from last indexed block. If there is no data in the db, sync genesis. - match indexer_backend.get_last_indexed_canon_block().await.ok() { + match indexer_backend.last_indexed_canon_block().await.ok() { Some(last_block_hash) => { log::debug!(target: "frontier-sql", "Resume from last block {last_block_hash:?}"); if let Some(parent_hash) = client diff --git a/client/rpc/src/lib.rs b/client/rpc/src/lib.rs index f1cce8b83d..28b8ed8a8a 100644 --- a/client/rpc/src/lib.rs +++ b/client/rpc/src/lib.rs @@ -205,7 +205,13 @@ pub mod frontier_backend_client { } } BlockNumberOrHash::Num(number) => Some(BlockId::Number(number.unique_saturated_into())), - BlockNumberOrHash::Latest => Some(BlockId::Hash(client.info().best_hash)), + BlockNumberOrHash::Latest => match backend.latest_block_hash().await { + Ok(hash) => Some(BlockId::Hash(hash)), + Err(e) => { + log::warn!(target: "rpc", "Failed to get latest block hash from the sql db: {:?}", e); + Some(BlockId::Hash(client.info().best_hash)) + } + }, BlockNumberOrHash::Earliest => Some(BlockId::Hash(client.info().genesis_hash)), BlockNumberOrHash::Pending => None, BlockNumberOrHash::Safe => Some(BlockId::Hash(client.info().finalized_hash)), @@ -363,8 +369,8 @@ mod tests { fn open_frontier_backend>( client: Arc, path: PathBuf, - ) -> Result>, String> { - Ok(Arc::new(fc_db::kv::Backend::::new( + ) -> Result>, String> { + Ok(Arc::new(fc_db::kv::Backend::::new( client, &fc_db::kv::DatabaseSettings { source: sc_client_db::DatabaseSource::RocksDb { diff --git a/template/node/src/command.rs b/template/node/src/command.rs index 62acedf395..7622729451 100644 --- a/template/node/src/command.rs +++ b/template/node/src/command.rs @@ -219,7 +219,7 @@ pub fn run() -> sc_cli::Result<()> { let (client, _, _, _, frontier_backend) = service::new_chain_ops(&mut config, &cli.eth)?; let frontier_backend = match frontier_backend { - fc_db::Backend::KeyValue(kv) => std::sync::Arc::new(kv), + fc_db::Backend::KeyValue(kv) => kv, _ => panic!("Only fc_db::Backend::KeyValue supported"), }; cmd.run(client, frontier_backend) diff --git a/template/node/src/eth.rs b/template/node/src/eth.rs index a3bcb3dbe0..228e7113ab 100644 --- a/template/node/src/eth.rs +++ b/template/node/src/eth.rs @@ -23,7 +23,7 @@ use frontier_template_runtime::opaque::Block; use crate::client::{FullBackend, FullClient}; /// Frontier DB backend type. -pub type FrontierBackend = fc_db::Backend; +pub type FrontierBackend = fc_db::Backend; pub fn db_config_dir(config: &Configuration) -> PathBuf { config.base_path.config_dir(config.chain_spec.id()) @@ -127,7 +127,7 @@ pub async fn spawn_frontier_tasks( task_manager: &TaskManager, client: Arc>, backend: Arc, - frontier_backend: FrontierBackend, + frontier_backend: Arc>>, filter_pool: Option, storage_override: Arc>, fee_history_cache: FeeHistoryCache, @@ -145,7 +145,7 @@ pub async fn spawn_frontier_tasks( Executor: NativeExecutionDispatch + 'static, { // Spawn main mapping sync worker background task. - match frontier_backend { + match &*frontier_backend { fc_db::Backend::KeyValue(b) => { task_manager.spawn_essential_handle().spawn( "frontier-mapping-sync-worker", @@ -156,7 +156,7 @@ pub async fn spawn_frontier_tasks( client.clone(), backend, storage_override.clone(), - Arc::new(b), + b.clone(), 3, 0, fc_mapping_sync::SyncStrategy::Normal, @@ -173,10 +173,10 @@ pub async fn spawn_frontier_tasks( fc_mapping_sync::sql::SyncWorker::run( client.clone(), backend, - Arc::new(b), + b.clone(), client.import_notification_stream(), fc_mapping_sync::sql::SyncWorkerConfig { - read_notification_timeout: Duration::from_secs(10), + read_notification_timeout: Duration::from_secs(30), check_indexed_blocks_interval: Duration::from_secs(60), }, fc_mapping_sync::SyncStrategy::Parachain, diff --git a/template/node/src/service.rs b/template/node/src/service.rs index 69c0605f59..5f5308144c 100644 --- a/template/node/src/service.rs +++ b/template/node/src/service.rs @@ -60,7 +60,7 @@ pub fn new_partial( Option, BoxBlockImport, GrandpaLinkHalf>, - FrontierBackend, + FrontierBackend>, Arc>, ), >, @@ -119,11 +119,11 @@ where let storage_override = Arc::new(StorageOverrideHandler::new(client.clone())); let frontier_backend = match eth_config.frontier_backend_type { - BackendType::KeyValue => FrontierBackend::KeyValue(fc_db::kv::Backend::open( + BackendType::KeyValue => FrontierBackend::KeyValue(Arc::new(fc_db::kv::Backend::open( Arc::clone(&client), &config.database, &db_config_dir(config), - )?), + )?)), BackendType::Sql => { let db_path = db_config_dir(config).join("sql"); std::fs::create_dir_all(&db_path).expect("failed creating sql db directory"); @@ -143,7 +143,7 @@ where storage_override.clone(), )) .unwrap_or_else(|err| panic!("failed creating sql backend: {:?}", err)); - FrontierBackend::Sql(backend) + FrontierBackend::Sql(Arc::new(backend)) } }; @@ -350,6 +350,7 @@ where let role = config.role.clone(); let force_authoring = config.force_authoring; let name = config.network.node_name.clone(); + let frontier_backend = Arc::new(frontier_backend); let enable_grandpa = !config.disable_grandpa && sealing.is_none(); let prometheus_registry = config.prometheus_registry().cloned(); @@ -415,9 +416,9 @@ where enable_dev_signer, network: network.clone(), sync: sync_service.clone(), - frontier_backend: match frontier_backend.clone() { - fc_db::Backend::KeyValue(b) => Arc::new(b), - fc_db::Backend::Sql(b) => Arc::new(b), + frontier_backend: match &*frontier_backend { + fc_db::Backend::KeyValue(b) => b.clone(), + fc_db::Backend::Sql(b) => b.clone(), }, storage_override: storage_override.clone(), block_data_cache: block_data_cache.clone(), @@ -711,7 +712,7 @@ pub fn new_chain_ops( Arc, BasicQueue, TaskManager, - FrontierBackend, + FrontierBackend, ), ServiceError, > { diff --git a/ts-tests/tests/test-contract-methods.ts b/ts-tests/tests/test-contract-methods.ts index d8da2273f2..9eb5c526a9 100644 --- a/ts-tests/tests/test-contract-methods.ts +++ b/ts-tests/tests/test-contract-methods.ts @@ -74,6 +74,7 @@ describeWithFrontier("Frontier RPC (Contract Methods)", (context) => { let number = (await context.web3.eth.getBlock("latest")).number; let last = number + BLOCK_HASH_COUNT; for (let i = number; i <= last; i++) { + await new Promise((resolve) => setTimeout(resolve, 60)); let hash = (await context.web3.eth.getBlock("latest")).hash; expect(await contract.methods.blockHash(i).call()).to.eq(hash); await createAndFinalizeBlockNowait(context.web3);