Files
pezkuwi-subxt/substrate/core/offchain/src/lib.rs
T
David Craven a3d19baea3 Refactors the offchain worker api (#3150)
* Update offchain primitives.

* Update offchain worker.

* Update im-online.

* Update service.

* Update node and node-template.

* Update runtime version.

* Fix build.

* Fix offchain worker tests.

* Generalize authority_pubkey.

* Add test.

* Update lib.rs
2019-07-22 17:20:57 +08:00

239 lines
6.4 KiB
Rust

// Copyright 2019 Parity Technologies (UK) Ltd.
// This file is part of Substrate.
// Substrate is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Substrate is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
//! Substrate offchain workers.
//!
//! The offchain workers is a special function of the runtime that
//! gets executed after block is imported. During execution
//! it's able to asynchronously submit extrinsics that will either
//! be propagated to other nodes added to the next block
//! produced by the node as unsigned transactions.
//!
//! Offchain workers can be used for computation-heavy tasks
//! that are not feasible for execution during regular block processing.
//! It can either be tasks that no consensus is required for,
//! or some form of consensus over the data can be built on-chain
//! for instance via:
//! 1. Challenge period for incorrect computations
//! 2. Majority voting for results
//! 3. etc
#![warn(missing_docs)]
use std::{
fmt,
marker::PhantomData,
sync::Arc,
};
use client::runtime_api::ApiExt;
use log::{debug, warn};
use primitives::{
ExecutionContext,
crypto,
};
use runtime_primitives::{
generic::BlockId,
traits::{self, ProvideRuntimeApi},
};
use futures::future::Future;
use transaction_pool::txpool::{Pool, ChainApi};
use network::NetworkStateInfo;
mod api;
pub mod testing;
pub use offchain_primitives::OffchainWorkerApi;
/// Provides currently configured authority key.
pub trait AuthorityKeyProvider<Block: traits::Block>: Clone + 'static {
/// The crypto used by the block authoring algorithm.
type ConsensusPair: crypto::Pair;
/// The crypto used by the finality gadget.
type FinalityPair: crypto::Pair;
/// Returns currently configured authority key.
fn authority_key(&self, block_id: &BlockId<Block>) -> Option<Self::ConsensusPair>;
/// Returns currently configured finality gadget authority key.
fn fg_authority_key(&self, block_id: &BlockId<Block>) -> Option<Self::FinalityPair>;
}
/// An offchain workers manager.
pub struct OffchainWorkers<
Client,
Storage,
KeyProvider,
Block: traits::Block,
> {
client: Arc<Client>,
db: Storage,
authority_key: KeyProvider,
keys_password: crypto::Protected<String>,
_block: PhantomData<Block>,
}
impl<Client, Storage, KeyProvider, Block: traits::Block> OffchainWorkers<
Client,
Storage,
KeyProvider,
Block,
> {
/// Creates new `OffchainWorkers`.
pub fn new(
client: Arc<Client>,
db: Storage,
authority_key: KeyProvider,
keys_password: crypto::Protected<String>,
) -> Self {
Self {
client,
db,
authority_key,
keys_password,
_block: PhantomData,
}
}
}
impl<Client, Storage, KeyProvider, Block: traits::Block> fmt::Debug for OffchainWorkers<
Client,
Storage,
KeyProvider,
Block,
> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_tuple("OffchainWorkers").finish()
}
}
impl<Client, Storage, KeyProvider, Block> OffchainWorkers<
Client,
Storage,
KeyProvider,
Block,
> where
Block: traits::Block,
Client: ProvideRuntimeApi,
Client::Api: OffchainWorkerApi<Block>,
KeyProvider: AuthorityKeyProvider<Block>,
Storage: client::backend::OffchainStorage + 'static,
{
/// Start the offchain workers after given block.
#[must_use]
pub fn on_block_imported<A>(
&self,
number: &<Block::Header as traits::Header>::Number,
pool: &Arc<Pool<A>>,
network_state: Arc<dyn NetworkStateInfo + Send + Sync>,
) -> impl Future<Item = (), Error = ()> where
A: ChainApi<Block=Block> + 'static,
{
let runtime = self.client.runtime_api();
let at = BlockId::number(*number);
let has_api = runtime.has_api::<dyn OffchainWorkerApi<Block>>(&at);
debug!("Checking offchain workers at {:?}: {:?}", at, has_api);
if has_api.unwrap_or(false) {
let (api, runner) = api::AsyncApi::new(
pool.clone(),
self.db.clone(),
self.keys_password.clone(),
self.authority_key.clone(),
at.clone(),
network_state.clone(),
);
debug!("Running offchain workers at {:?}", at);
let api = Box::new(api);
runtime.offchain_worker_with_context(&at, ExecutionContext::OffchainWorker(api), *number).unwrap();
futures::future::Either::A(runner.process())
} else {
futures::future::Either::B(futures::future::ok(()))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::Future;
use primitives::{ed25519, sr25519};
use network::{Multiaddr, PeerId};
struct MockNetworkStateInfo();
impl NetworkStateInfo for MockNetworkStateInfo {
fn external_addresses(&self) -> Vec<Multiaddr> {
Vec::new()
}
fn peer_id(&self) -> PeerId {
PeerId::random()
}
}
#[derive(Clone)]
pub(crate) struct TestProvider<Block> {
_marker: PhantomData<Block>,
pub(crate) sr_key: Option<sr25519::Pair>,
pub(crate) ed_key: Option<ed25519::Pair>,
}
impl<Block: traits::Block> Default for TestProvider<Block> {
fn default() -> Self {
Self {
_marker: PhantomData,
sr_key: None,
ed_key: None,
}
}
}
impl<Block: traits::Block> AuthorityKeyProvider<Block> for TestProvider<Block> {
type ConsensusPair = ed25519::Pair;
type FinalityPair = sr25519::Pair;
fn authority_key(&self, _: &BlockId<Block>) -> Option<Self::ConsensusPair> {
self.ed_key.clone()
}
fn fg_authority_key(&self, _: &BlockId<Block>) -> Option<Self::FinalityPair> {
self.sr_key.clone()
}
}
#[test]
fn should_call_into_runtime_and_produce_extrinsic() {
// given
let _ = env_logger::try_init();
let runtime = tokio::runtime::Runtime::new().unwrap();
let client = Arc::new(test_client::new());
let pool = Arc::new(Pool::new(Default::default(), ::transaction_pool::ChainApi::new(client.clone())));
let db = client_db::offchain::LocalStorage::new_test();
let mock = Arc::new(MockNetworkStateInfo());
// when
let offchain = OffchainWorkers::new(client, db, TestProvider::default(), "".to_owned().into());
runtime.executor().spawn(offchain.on_block_imported(&0u64, &pool, mock.clone()));
// then
runtime.shutdown_on_idle().wait().unwrap();
assert_eq!(pool.status().ready, 1);
assert_eq!(pool.ready().next().unwrap().is_propagateable(), false);
}
}