Removal of execution strategies (#14387)

* Start * More work! * Moar * More changes * More fixes * More worrk * More fixes * More fixes to make it compile * Adds `NoOffchainStorage` * Pass the extensions * Small basti making small progress * Fix merge errors and remove `ExecutionContext` * Move registration of `ReadRuntimeVersionExt` to `ExecutionExtension` Instead of registering `ReadRuntimeVersionExt` in `sp-state-machine` it is moved to `ExecutionExtension` which provides the default extensions. * Fix compilation * Register the global extensions inside runtime api instance * Fixes * Fix `generate_initial_session_keys` by passing the keystore extension * Fix the grandpa tests * Fix more tests * Fix more tests * Don't set any heap pages if there isn't an override * Fix small fallout * FMT * Fix tests * More tests * Offchain worker custom extensions * More fixes * Make offchain tx pool creation reusable Introduces an `OffchainTransactionPoolFactory` for creating offchain transactions pools that can be registered in the runtime externalities context. This factory will be required for a later pr to make the creation of offchain transaction pools easier. * Fixes * Fixes * Set offchain transaction pool in BABE before using it in the runtime * Add the `offchain_tx_pool` to Grandpa as well * Fix the nodes * Print some error when using the old warnings * Fix merge issues * Fix compilation * Rename `babe_link` * Rename to `offchain_tx_pool_factory` * Cleanup * FMT * Fix benchmark name * Fix `try-runtime` * Remove `--execution` CLI args * Make clippy happy * Forward bls functions * Fix docs * Update UI tests * Update client/api/src/execution_extensions.rs Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Koute <koute@users.noreply.github.com> * Update client/cli/src/params/import_params.rs Co-authored-by: Koute <koute@users.noreply.github.com> * Update client/api/src/execution_extensions.rs Co-authored-by: Koute <koute@users.noreply.github.com> * Pass the offchain storage to the MMR RPC * Update client/api/src/execution_extensions.rs Co-authored-by: Sebastian Kunert <skunert49@gmail.com> * Review comments * Fixes --------- Co-authored-by: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Co-authored-by: Koute <koute@users.noreply.github.com> Co-authored-by: Sebastian Kunert <skunert49@gmail.com>
2026-07-18 22:15:40 +00:00 · 2023-07-11 16:21:38 +02:00
parent a2b01c061b
commit 5eb816d7a6
96 changed files with 1175 additions and 1499 deletions
@@ -25,8 +25,8 @@ pub use http::SharedClient;
 use libp2p::{Multiaddr, PeerId};
 use sp_core::{
 	offchain::{
-		self, HttpError, HttpRequestId, HttpRequestStatus, OffchainStorage, OpaqueMultiaddr,
-		OpaqueNetworkState, StorageKind, Timestamp,
+		self, HttpError, HttpRequestId, HttpRequestStatus, OpaqueMultiaddr, OpaqueNetworkState,
+		Timestamp,
 	},
 	OpaquePeerId,
 };
@@ -36,110 +36,6 @@ mod http;

 mod timestamp;

-fn unavailable_yet<R: Default>(name: &str) -> R {
-	tracing::error!(
-		target: super::LOG_TARGET,
-		"The {:?} API is not available for offchain workers yet. Follow \
-		https://github.com/paritytech/substrate/issues/1458 for details",
-		name
-	);
-	Default::default()
-}
-
-const LOCAL_DB: &str = "LOCAL (fork-aware) DB";
-
-/// Offchain DB reference.
-#[derive(Debug, Clone)]
-pub struct Db<Storage> {
-	/// Persistent storage database.
-	persistent: Storage,
-}
-
-impl<Storage: OffchainStorage> Db<Storage> {
-	/// Create new instance of Offchain DB.
-	pub fn new(persistent: Storage) -> Self {
-		Self { persistent }
-	}
-
-	/// Create new instance of Offchain DB, backed by given backend.
-	pub fn factory_from_backend<Backend, Block>(
-		backend: &Backend,
-	) -> Option<Box<dyn sc_client_api::execution_extensions::DbExternalitiesFactory>>
-	where
-		Backend: sc_client_api::Backend<Block, OffchainStorage = Storage>,
-		Block: sp_runtime::traits::Block,
-		Storage: 'static,
-	{
-		sc_client_api::Backend::offchain_storage(backend).map(|db| Box::new(Self::new(db)) as _)
-	}
-}
-
-impl<Storage: OffchainStorage> offchain::DbExternalities for Db<Storage> {
-	fn local_storage_set(&mut self, kind: StorageKind, key: &[u8], value: &[u8]) {
-		tracing::debug!(
-			target: "offchain-worker::storage",
-			?kind,
-			key = ?array_bytes::bytes2hex("", key),
-			value = ?array_bytes::bytes2hex("", value),
-			"Write",
-		);
-		match kind {
-			StorageKind::PERSISTENT => self.persistent.set(STORAGE_PREFIX, key, value),
-			StorageKind::LOCAL => unavailable_yet(LOCAL_DB),
-		}
-	}
-
-	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
-		tracing::debug!(
-			target: "offchain-worker::storage",
-			?kind,
-			key = ?array_bytes::bytes2hex("", key),
-			"Clear",
-		);
-		match kind {
-			StorageKind::PERSISTENT => self.persistent.remove(STORAGE_PREFIX, key),
-			StorageKind::LOCAL => unavailable_yet(LOCAL_DB),
-		}
-	}
-
-	fn local_storage_compare_and_set(
-		&mut self,
-		kind: StorageKind,
-		key: &[u8],
-		old_value: Option<&[u8]>,
-		new_value: &[u8],
-	) -> bool {
-		tracing::debug!(
-			target: "offchain-worker::storage",
-			?kind,
-			key = ?array_bytes::bytes2hex("", key),
-			new_value = ?array_bytes::bytes2hex("", new_value),
-			old_value = ?old_value.as_ref().map(|s| array_bytes::bytes2hex("", s)),
-			"CAS",
-		);
-		match kind {
-			StorageKind::PERSISTENT =>
-				self.persistent.compare_and_set(STORAGE_PREFIX, key, old_value, new_value),
-			StorageKind::LOCAL => unavailable_yet(LOCAL_DB),
-		}
-	}
-
-	fn local_storage_get(&mut self, kind: StorageKind, key: &[u8]) -> Option<Vec<u8>> {
-		let result = match kind {
-			StorageKind::PERSISTENT => self.persistent.get(STORAGE_PREFIX, key),
-			StorageKind::LOCAL => unavailable_yet(LOCAL_DB),
-		};
-		tracing::debug!(
-			target: "offchain-worker::storage",
-			?kind,
-			key = ?array_bytes::bytes2hex("", key),
-			result = ?result.as_ref().map(|s| array_bytes::bytes2hex("", s)),
-			"Read",
-		);
-		result
-	}
-}
-
 /// Asynchronous offchain API.
 ///
 /// NOTE this is done to prevent recursive calls into the runtime
@@ -329,7 +225,7 @@ mod tests {
 		config::MultiaddrWithPeerId, types::ProtocolName, NetworkPeers, NetworkStateInfo,
 		ReputationChange,
 	};
-	use sp_core::offchain::{DbExternalities, Externalities};
+	use sp_core::offchain::{storage::OffchainDb, DbExternalities, Externalities, StorageKind};
 	use std::time::SystemTime;

 	pub(super) struct TestNetwork();
@@ -418,8 +314,8 @@ mod tests {
 		AsyncApi::new(mock, false, shared_client)
 	}

-	fn offchain_db() -> Db<LocalStorage> {
-		Db::new(LocalStorage::new_test())
+	fn offchain_db() -> OffchainDb<LocalStorage> {
+		OffchainDb::new(LocalStorage::new_test())
 	}

 	#[test]
@@ -35,22 +35,26 @@

 #![warn(missing_docs)]

-use std::{fmt, marker::PhantomData, sync::Arc};
+use std::{fmt, sync::Arc};

 use futures::{
 	future::{ready, Future},
 	prelude::*,
 };
 use parking_lot::Mutex;
+use sc_client_api::BlockchainEvents;
 use sc_network::{NetworkPeers, NetworkStateInfo};
+use sc_transaction_pool_api::OffchainTransactionPoolFactory;
 use sp_api::{ApiExt, ProvideRuntimeApi};
-use sp_core::{offchain, traits::SpawnNamed, ExecutionContext};
+use sp_core::{offchain, traits::SpawnNamed};
+use sp_externalities::Extension;
+use sp_keystore::{KeystoreExt, KeystorePtr};
 use sp_runtime::traits::{self, Header};
 use threadpool::ThreadPool;

 mod api;

-pub use api::Db as OffchainDb;
+pub use sp_core::offchain::storage::OffchainDb;
 pub use sp_offchain::{OffchainWorkerApi, STORAGE_PREFIX};

 const LOG_TARGET: &str = "offchain-worker";
@@ -61,65 +65,160 @@ pub trait NetworkProvider: NetworkStateInfo + NetworkPeers {}

 impl<T> NetworkProvider for T where T: NetworkStateInfo + NetworkPeers {}

+/// Special type that implements [`OffchainStorage`](offchain::OffchainStorage).
+///
+/// This type can not be constructed and should only be used when passing `None` as `offchain_db` to
+/// [`OffchainWorkerOptions`] to make the compiler happy.
+#[derive(Clone)]
+pub enum NoOffchainStorage {}
+
+impl offchain::OffchainStorage for NoOffchainStorage {
+	fn set(&mut self, _: &[u8], _: &[u8], _: &[u8]) {
+		unimplemented!("`NoOffchainStorage` can not be constructed!")
+	}
+
+	fn remove(&mut self, _: &[u8], _: &[u8]) {
+		unimplemented!("`NoOffchainStorage` can not be constructed!")
+	}
+
+	fn get(&self, _: &[u8], _: &[u8]) -> Option<Vec<u8>> {
+		unimplemented!("`NoOffchainStorage` can not be constructed!")
+	}
+
+	fn compare_and_set(&mut self, _: &[u8], _: &[u8], _: Option<&[u8]>, _: &[u8]) -> bool {
+		unimplemented!("`NoOffchainStorage` can not be constructed!")
+	}
+}
+
 /// Options for [`OffchainWorkers`]
-pub struct OffchainWorkerOptions {
+pub struct OffchainWorkerOptions<RA, Block: traits::Block, Storage, CE> {
+	/// Provides access to the runtime api.
+	pub runtime_api_provider: Arc<RA>,
+	/// Provides access to the keystore.
+	pub keystore: Option<KeystorePtr>,
+	/// Provides access to the offchain database.
+	///
+	/// Use [`NoOffchainStorage`] as type when passing `None` to have some type that works.
+	pub offchain_db: Option<Storage>,
+	/// Provides access to the transaction pool.
+	pub transaction_pool: Option<OffchainTransactionPoolFactory<Block>>,
+	/// Provides access to network information.
+	pub network_provider: Arc<dyn NetworkProvider + Send + Sync>,
+	/// Is the node running as validator?
+	pub is_validator: bool,
 	/// Enable http requests from offchain workers?
 	///
 	/// If not enabled, any http request will panic.
 	pub enable_http_requests: bool,
+	/// Callback to create custom [`Extension`]s that should be registered for the
+	/// `offchain_worker` runtime call.
+	///
+	/// These [`Extension`]s are registered along-side the default extensions and are accessible in
+	/// the host functions.
+	///
+	/// # Example:
+	///
+	/// ```nocompile
+	/// custom_extensions: |block_hash| {
+	///     vec![MyCustomExtension::new()]
+	/// }
+	/// ```
+	pub custom_extensions: CE,
 }

 /// An offchain workers manager.
-pub struct OffchainWorkers<Client, Block: traits::Block> {
-	client: Arc<Client>,
-	_block: PhantomData<Block>,
+pub struct OffchainWorkers<RA, Block: traits::Block, Storage> {
+	runtime_api_provider: Arc<RA>,
 	thread_pool: Mutex<ThreadPool>,
 	shared_http_client: api::SharedClient,
-	enable_http: bool,
+	enable_http_requests: bool,
+	keystore: Option<KeystorePtr>,
+	offchain_db: Option<OffchainDb<Storage>>,
+	transaction_pool: Option<OffchainTransactionPoolFactory<Block>>,
+	network_provider: Arc<dyn NetworkProvider + Send + Sync>,
+	is_validator: bool,
+	custom_extensions: Box<dyn Fn(Block::Hash) -> Vec<Box<dyn Extension>> + Send>,
 }

-impl<Client, Block: traits::Block> OffchainWorkers<Client, Block> {
+impl<RA, Block: traits::Block, Storage> OffchainWorkers<RA, Block, Storage> {
 	/// Creates new [`OffchainWorkers`].
-	pub fn new(client: Arc<Client>) -> Self {
-		Self::new_with_options(client, OffchainWorkerOptions { enable_http_requests: true })
-	}
-
-	/// Creates new [`OffchainWorkers`] using the given `options`.
-	pub fn new_with_options(client: Arc<Client>, options: OffchainWorkerOptions) -> Self {
+	pub fn new<CE: Fn(Block::Hash) -> Vec<Box<dyn Extension>> + Send + 'static>(
+		OffchainWorkerOptions {
+			runtime_api_provider,
+			keystore,
+			offchain_db,
+			transaction_pool,
+			network_provider,
+			is_validator,
+			enable_http_requests,
+			custom_extensions,
+		}: OffchainWorkerOptions<RA, Block, Storage, CE>,
+	) -> Self {
 		Self {
-			client,
-			_block: PhantomData,
+			runtime_api_provider,
 			thread_pool: Mutex::new(ThreadPool::with_name(
 				"offchain-worker".into(),
 				num_cpus::get(),
 			)),
 			shared_http_client: api::SharedClient::new(),
-			enable_http: options.enable_http_requests,
+			enable_http_requests,
+			keystore,
+			offchain_db: offchain_db.map(OffchainDb::new),
+			transaction_pool,
+			is_validator,
+			network_provider,
+			custom_extensions: Box::new(custom_extensions),
 		}
 	}
 }

-impl<Client, Block: traits::Block> fmt::Debug for OffchainWorkers<Client, Block> {
+impl<RA, Block: traits::Block, Storage: offchain::OffchainStorage> fmt::Debug
+	for OffchainWorkers<RA, Block, Storage>
+{
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 		f.debug_tuple("OffchainWorkers").finish()
 	}
 }

-impl<Client, Block> OffchainWorkers<Client, Block>
+impl<RA, Block, Storage> OffchainWorkers<RA, Block, Storage>
 where
 	Block: traits::Block,
-	Client: ProvideRuntimeApi<Block> + Send + Sync + 'static,
-	Client::Api: OffchainWorkerApi<Block>,
+	RA: ProvideRuntimeApi<Block> + Send + Sync + 'static,
+	RA::Api: OffchainWorkerApi<Block>,
+	Storage: offchain::OffchainStorage + 'static,
 {
+	/// Run the offchain workers on every block import.
+	pub async fn run<BE: BlockchainEvents<Block>>(
+		self,
+		import_events: Arc<BE>,
+		spawner: impl SpawnNamed,
+	) {
+		import_events
+			.import_notification_stream()
+			.for_each(move |n| {
+				if n.is_new_best {
+					spawner.spawn(
+						"offchain-on-block",
+						Some("offchain-worker"),
+						self.on_block_imported(&n.header).boxed(),
+					);
+				} else {
+					tracing::debug!(
+						target: LOG_TARGET,
+						"Skipping offchain workers for non-canon block: {:?}",
+						n.header,
+					)
+				}
+
+				ready(())
+			})
+			.await;
+	}
+
 	/// Start the offchain workers after given block.
 	#[must_use]
-	pub fn on_block_imported(
-		&self,
-		header: &Block::Header,
-		network_provider: Arc<dyn NetworkProvider + Send + Sync>,
-		is_validator: bool,
-	) -> impl Future<Output = ()> {
-		let runtime = self.client.runtime_api();
+	fn on_block_imported(&self, header: &Block::Header) -> impl Future<Output = ()> {
+		let runtime = self.runtime_api_provider.runtime_api();
 		let hash = header.hash();
 		let has_api_v1 = runtime.has_api_with::<dyn OffchainWorkerApi<Block>, _>(hash, |v| v == 1);
 		let has_api_v2 = runtime.has_api_with::<dyn OffchainWorkerApi<Block>, _>(hash, |v| v == 2);
@@ -140,36 +239,59 @@ where
 		};
 		tracing::debug!(
 			target: LOG_TARGET,
-			"Checking offchain workers at {:?}: version:{}",
-			hash,
-			version
+			"Checking offchain workers at {hash:?}: version: {version}",
 		);
+
 		let process = (version > 0).then(|| {
-			let (api, runner) =
-				api::AsyncApi::new(network_provider, is_validator, self.shared_http_client.clone());
-			tracing::debug!(target: LOG_TARGET, "Spawning offchain workers at {:?}", hash);
+			let (api, runner) = api::AsyncApi::new(
+				self.network_provider.clone(),
+				self.is_validator,
+				self.shared_http_client.clone(),
+			);
+			tracing::debug!(target: LOG_TARGET, "Spawning offchain workers at {hash:?}");
 			let header = header.clone();
-			let client = self.client.clone();
+			let client = self.runtime_api_provider.clone();

 			let mut capabilities = offchain::Capabilities::all();
+			capabilities.set(offchain::Capabilities::HTTP, self.enable_http_requests);
+
+			let keystore = self.keystore.clone();
+			let db = self.offchain_db.clone();
+			let tx_pool = self.transaction_pool.clone();
+			let custom_extensions = (*self.custom_extensions)(hash);

-			capabilities.set(offchain::Capabilities::HTTP, self.enable_http);
 			self.spawn_worker(move || {
-				let runtime = client.runtime_api();
+				let mut runtime = client.runtime_api();
 				let api = Box::new(api);
-				tracing::debug!(target: LOG_TARGET, "Running offchain workers at {:?}", hash);
+				tracing::debug!(target: LOG_TARGET, "Running offchain workers at {hash:?}");
+
+				if let Some(keystore) = keystore {
+					runtime.register_extension(KeystoreExt(keystore.clone()));
+				}
+
+				if let Some(pool) = tx_pool {
+					runtime.register_extension(pool.offchain_transaction_pool(hash));
+				}
+
+				if let Some(offchain_db) = db {
+					runtime.register_extension(offchain::OffchainDbExt::new(
+						offchain::LimitedExternalities::new(capabilities, offchain_db.clone()),
+					));
+				}
+
+				runtime.register_extension(offchain::OffchainWorkerExt::new(
+					offchain::LimitedExternalities::new(capabilities, api),
+				));
+
+				custom_extensions.into_iter().for_each(|ext| runtime.register_extension(ext));

-				let context = ExecutionContext::OffchainCall(Some((api, capabilities)));
 				let run = if version == 2 {
-					runtime.offchain_worker_with_context(hash, context, &header)
+					runtime.offchain_worker(hash, &header)
 				} else {
 					#[allow(deprecated)]
-					runtime.offchain_worker_before_version_2_with_context(
-						hash,
-						context,
-						*header.number(),
-					)
+					runtime.offchain_worker_before_version_2(hash, *header.number())
 				};
+
 				if let Err(e) = run {
 					tracing::error!(
 						target: LOG_TARGET,
@@ -201,44 +323,6 @@ where
 	}
 }

-/// Inform the offchain worker about new imported blocks
-pub async fn notification_future<Client, Block, Spawner>(
-	is_validator: bool,
-	client: Arc<Client>,
-	offchain: Arc<OffchainWorkers<Client, Block>>,
-	spawner: Spawner,
-	network_provider: Arc<dyn NetworkProvider + Send + Sync>,
-) where
-	Block: traits::Block,
-	Client:
-		ProvideRuntimeApi<Block> + sc_client_api::BlockchainEvents<Block> + Send + Sync + 'static,
-	Client::Api: OffchainWorkerApi<Block>,
-	Spawner: SpawnNamed,
-{
-	client
-		.import_notification_stream()
-		.for_each(move |n| {
-			if n.is_new_best {
-				spawner.spawn(
-					"offchain-on-block",
-					Some("offchain-worker"),
-					offchain
-						.on_block_imported(&n.header, network_provider.clone(), is_validator)
-						.boxed(),
-				);
-			} else {
-				tracing::debug!(
-					target: LOG_TARGET,
-					"Skipping offchain workers for non-canon block: {:?}",
-					n.header,
-				)
-			}
-
-			ready(())
-		})
-		.await;
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -348,8 +432,17 @@ mod tests {
 		let header = client.header(client.chain_info().genesis_hash).unwrap().unwrap();

 		// when
-		let offchain = OffchainWorkers::new(client);
-		futures::executor::block_on(offchain.on_block_imported(&header, network, false));
+		let offchain = OffchainWorkers::new(OffchainWorkerOptions {
+			runtime_api_provider: client,
+			keystore: None,
+			offchain_db: None::<NoOffchainStorage>,
+			transaction_pool: Some(OffchainTransactionPoolFactory::new(pool.clone())),
+			network_provider: network,
+			is_validator: false,
+			enable_http_requests: false,
+			custom_extensions: |_| Vec::new(),
+		});
+		futures::executor::block_on(offchain.on_block_imported(&header));

 		// then
 		assert_eq!(pool.status().ready, 1);