Offchain execution extensions (#4145)

* Pass Extensions instead of individual objects.

* Move TransactionPool to a separate ExternalitiesExtension.

* Fix compilation.?

* Clean up.

* Refactor testing utilities.

* Add docs, fix tests.

* Fix doctest.

* Fix formatting and add some logs.

* Add some docs.

* Remove unused files.
This commit is contained in:
Tomasz Drwięga
2019-11-22 17:10:23 +01:00
committed by Gavin Wood
parent f000392cc0
commit 86b6ac5571
39 changed files with 554 additions and 360 deletions
@@ -23,6 +23,31 @@ use runtime_interface::pass_by::{PassByCodec, PassByInner, PassByEnum};
pub use crate::crypto::KeyTypeId;
#[cfg(feature = "std")]
pub mod storage;
#[cfg(feature = "std")]
pub mod testing;
/// Offchain workers local storage.
pub trait OffchainStorage: Clone + Send + Sync {
/// Persist a value in storage under given key and prefix.
fn set(&mut self, prefix: &[u8], key: &[u8], value: &[u8]);
/// Retrieve a value from storage under given key and prefix.
fn get(&self, prefix: &[u8], key: &[u8]) -> Option<Vec<u8>>;
/// Replace the value in storage if given old_value matches the current one.
///
/// Returns `true` if the value has been set and false otherwise.
fn compare_and_set(
&mut self,
prefix: &[u8],
key: &[u8],
old_value: Option<&[u8]>,
new_value: &[u8],
) -> bool;
}
/// A type of supported crypto.
#[derive(Clone, Copy, PartialEq, Eq, Encode, Decode, RuntimeDebug, PassByEnum)]
#[repr(C)]
@@ -296,10 +321,6 @@ pub trait Externalities: Send {
/// Even if this function returns `true`, it does not mean that any keys are configured
/// and that the validator is registered in the chain.
fn is_validator(&self) -> bool;
/// Submit transaction.
///
/// The transaction will end up in the pool and be propagated to others.
fn submit_transaction(&mut self, extrinsic: Vec<u8>) -> Result<(), ()>;
/// Returns information about the local node's network state.
fn network_state(&self) -> Result<OpaqueNetworkState, ()>;
@@ -466,10 +487,6 @@ impl<T: Externalities + ?Sized> Externalities for Box<T> {
(& **self).is_validator()
}
fn submit_transaction(&mut self, ex: Vec<u8>) -> Result<(), ()> {
(&mut **self).submit_transaction(ex)
}
fn network_state(&self) -> Result<OpaqueNetworkState, ()> {
(& **self).network_state()
}
@@ -569,11 +586,6 @@ impl<T: Externalities> Externalities for LimitedExternalities<T> {
self.externalities.is_validator()
}
fn submit_transaction(&mut self, ex: Vec<u8>) -> Result<(), ()> {
self.check(Capability::TransactionPool, "submit_transaction");
self.externalities.submit_transaction(ex)
}
fn network_state(&self) -> Result<OpaqueNetworkState, ()> {
self.check(Capability::NetworkState, "network_state");
self.externalities.network_state()
@@ -670,6 +682,34 @@ impl OffchainExt {
}
}
/// Abstraction over transaction pool.
///
/// This trait is currently used within the `ExternalitiesExtension`
/// to provide offchain calls with access to the transaction pool without
/// tight coupling with any pool implementation.
#[cfg(feature = "std")]
pub trait TransactionPool {
/// Submit transaction.
///
/// The transaction will end up in the pool and be propagated to others.
fn submit_transaction(&mut self, extrinsic: Vec<u8>) -> Result<(), ()>;
}
#[cfg(feature = "std")]
externalities::decl_extension! {
/// An externalities extension to submit transactions to the pool.
pub struct TransactionPoolExt(Box<dyn TransactionPool + Send>);
}
#[cfg(feature = "std")]
impl TransactionPoolExt {
/// Create a new instance of `TransactionPoolExt`.
pub fn new<O: TransactionPool + Send + 'static>(pool: O) -> Self {
Self(Box::new(pool))
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -0,0 +1,60 @@
// 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/>.
//! In-memory implementation of offchain workers database.
use std::collections::hash_map::{HashMap, Entry};
use crate::offchain::OffchainStorage;
/// In-memory storage for offchain workers.
#[derive(Debug, Clone, Default)]
pub struct InMemOffchainStorage {
storage: HashMap<Vec<u8>, Vec<u8>>,
}
impl OffchainStorage for InMemOffchainStorage {
fn set(&mut self, prefix: &[u8], key: &[u8], value: &[u8]) {
let key = prefix.iter().chain(key).cloned().collect();
self.storage.insert(key, value.to_vec());
}
fn get(&self, prefix: &[u8], key: &[u8]) -> Option<Vec<u8>> {
let key: Vec<u8> = prefix.iter().chain(key).cloned().collect();
self.storage.get(&key).cloned()
}
fn compare_and_set(
&mut self,
prefix: &[u8],
key: &[u8],
old_value: Option<&[u8]>,
new_value: &[u8],
) -> bool {
let key = prefix.iter().chain(key).cloned().collect();
match self.storage.entry(key) {
Entry::Vacant(entry) => if old_value.is_none() {
entry.insert(new_value.to_vec());
true
} else { false },
Entry::Occupied(ref mut entry) if Some(entry.get().as_slice()) == old_value => {
entry.insert(new_value.to_vec());
true
},
_ => false,
}
}
}
@@ -0,0 +1,339 @@
// 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/>.
//! Utilities for offchain calls testing.
//!
//! Namely all ExecutionExtensions that allow mocking
//! the extra APIs.
use std::{
collections::BTreeMap,
sync::Arc,
};
use crate::offchain::{
self,
storage::InMemOffchainStorage,
HttpError,
HttpRequestId as RequestId,
HttpRequestStatus as RequestStatus,
Timestamp,
StorageKind,
OpaqueNetworkState,
TransactionPool,
OffchainStorage,
};
use parking_lot::RwLock;
/// Pending request.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct PendingRequest {
/// HTTP method
pub method: String,
/// URI
pub uri: String,
/// Encoded Metadata
pub meta: Vec<u8>,
/// Request headers
pub headers: Vec<(String, String)>,
/// Request body
pub body: Vec<u8>,
/// Has the request been sent already.
pub sent: bool,
/// Response body
pub response: Option<Vec<u8>>,
/// Number of bytes already read from the response body.
pub read: usize,
/// Response headers
pub response_headers: Vec<(String, String)>,
}
/// Internal state of the externalities.
///
/// This can be used in tests to respond or assert stuff about interactions.
#[derive(Debug, Default)]
pub struct OffchainState {
/// A list of pending requests.
pub requests: BTreeMap<RequestId, PendingRequest>,
expected_requests: BTreeMap<RequestId, PendingRequest>,
/// Persistent local storage
pub persistent_storage: InMemOffchainStorage,
/// Local storage
pub local_storage: InMemOffchainStorage,
}
impl OffchainState {
/// Asserts that pending request has been submitted and fills it's response.
pub fn fulfill_pending_request(
&mut self,
id: u16,
expected: PendingRequest,
response: impl Into<Vec<u8>>,
response_headers: impl IntoIterator<Item=(String, String)>,
) {
match self.requests.get_mut(&RequestId(id)) {
None => {
panic!("Missing pending request: {:?}.\n\nAll: {:?}", id, self.requests);
}
Some(req) => {
assert_eq!(
*req,
expected,
);
req.response = Some(response.into());
req.response_headers = response_headers.into_iter().collect();
}
}
}
fn fulfill_expected(&mut self, id: u16) {
if let Some(mut req) = self.expected_requests.remove(&RequestId(id)) {
let response = req.response.take().expect("Response checked while added.");
let headers = std::mem::replace(&mut req.response_headers, vec![]);
self.fulfill_pending_request(id, req, response, headers);
}
}
/// Add expected HTTP request.
///
/// This method can be used to initialize expected HTTP requests and their responses
/// before running the actual code that utilizes them (for instance before calling into runtime).
/// Expected request has to be fulfilled before this struct is dropped,
/// the `response` and `response_headers` fields will be used to return results to the callers.
pub fn expect_request(&mut self, id: u16, expected: PendingRequest) {
if expected.response.is_none() {
panic!("Expected request needs to have a response.");
}
self.expected_requests.insert(RequestId(id), expected);
}
}
impl Drop for OffchainState {
fn drop(&mut self) {
// If we panic! while we are already in a panic, the test dies with an illegal instruction.
if !self.expected_requests.is_empty() && !std::thread::panicking() {
panic!("Unfulfilled expected requests: {:?}", self.expected_requests);
}
}
}
/// Implementation of offchain externalities used for tests.
#[derive(Clone, Default, Debug)]
pub struct TestOffchainExt(pub Arc<RwLock<OffchainState>>);
impl TestOffchainExt {
/// Create new `TestOffchainExt` and a reference to the internal state.
pub fn new() -> (Self, Arc<RwLock<OffchainState>>) {
let ext = Self::default();
let state = ext.0.clone();
(ext, state)
}
}
impl offchain::Externalities for TestOffchainExt {
fn is_validator(&self) -> bool {
unimplemented!("not needed in tests so far")
}
fn network_state(&self) -> Result<OpaqueNetworkState, ()> {
Ok(OpaqueNetworkState {
peer_id: Default::default(),
external_addresses: vec![],
})
}
fn timestamp(&mut self) -> Timestamp {
unimplemented!("not needed in tests so far")
}
fn sleep_until(&mut self, _deadline: Timestamp) {
unimplemented!("not needed in tests so far")
}
fn random_seed(&mut self) -> [u8; 32] {
unimplemented!("not needed in tests so far")
}
fn local_storage_set(&mut self, kind: StorageKind, key: &[u8], value: &[u8]) {
let mut state = self.0.write();
match kind {
StorageKind::LOCAL => &mut state.local_storage,
StorageKind::PERSISTENT => &mut state.persistent_storage,
}.set(b"", key, value);
}
fn local_storage_compare_and_set(
&mut self,
kind: StorageKind,
key: &[u8],
old_value: Option<&[u8]>,
new_value: &[u8]
) -> bool {
let mut state = self.0.write();
match kind {
StorageKind::LOCAL => &mut state.local_storage,
StorageKind::PERSISTENT => &mut state.persistent_storage,
}.compare_and_set(b"", key, old_value, new_value)
}
fn local_storage_get(&mut self, kind: StorageKind, key: &[u8]) -> Option<Vec<u8>> {
let state = self.0.read();
match kind {
StorageKind::LOCAL => &state.local_storage,
StorageKind::PERSISTENT => &state.persistent_storage,
}.get(b"", key)
}
fn http_request_start(&mut self, method: &str, uri: &str, meta: &[u8]) -> Result<RequestId, ()> {
let mut state = self.0.write();
let id = RequestId(state.requests.len() as u16);
state.requests.insert(id.clone(), PendingRequest {
method: method.into(),
uri: uri.into(),
meta: meta.into(),
..Default::default()
});
Ok(id)
}
fn http_request_add_header(
&mut self,
request_id: RequestId,
name: &str,
value: &str,
) -> Result<(), ()> {
let mut state = self.0.write();
if let Some(req) = state.requests.get_mut(&request_id) {
req.headers.push((name.into(), value.into()));
Ok(())
} else {
Err(())
}
}
fn http_request_write_body(
&mut self,
request_id: RequestId,
chunk: &[u8],
_deadline: Option<Timestamp>
) -> Result<(), HttpError> {
let mut state = self.0.write();
let sent = {
let req = state.requests.get_mut(&request_id).ok_or(HttpError::IoError)?;
req.body.extend(chunk);
if chunk.is_empty() {
req.sent = true;
}
req.sent
};
if sent {
state.fulfill_expected(request_id.0);
}
Ok(())
}
fn http_response_wait(
&mut self,
ids: &[RequestId],
_deadline: Option<Timestamp>,
) -> Vec<RequestStatus> {
let state = self.0.read();
ids.iter().map(|id| match state.requests.get(id) {
Some(req) if req.response.is_none() =>
panic!("No `response` provided for request with id: {:?}", id),
None => RequestStatus::Invalid,
_ => RequestStatus::Finished(200),
}).collect()
}
fn http_response_headers(&mut self, request_id: RequestId) -> Vec<(Vec<u8>, Vec<u8>)> {
let state = self.0.read();
if let Some(req) = state.requests.get(&request_id) {
req.response_headers
.clone()
.into_iter()
.map(|(k, v)| (k.into_bytes(), v.into_bytes()))
.collect()
} else {
Default::default()
}
}
fn http_response_read_body(
&mut self,
request_id: RequestId,
buffer: &mut [u8],
_deadline: Option<Timestamp>
) -> Result<usize, HttpError> {
let mut state = self.0.write();
if let Some(req) = state.requests.get_mut(&request_id) {
let response = req.response
.as_mut()
.expect(&format!("No response provided for request: {:?}", request_id));
if req.read >= response.len() {
// Remove the pending request as per spec.
state.requests.remove(&request_id);
Ok(0)
} else {
let read = std::cmp::min(buffer.len(), response[req.read..].len());
buffer[0..read].copy_from_slice(&response[req.read..read]);
req.read += read;
Ok(read)
}
} else {
Err(HttpError::IoError)
}
}
}
/// The internal state of the fake transaction pool.
#[derive(Default)]
pub struct PoolState {
/// A vector of transactions submitted from the runtime.
pub transactions: Vec<Vec<u8>>,
}
/// Implementation of transaction pool used for test.
///
/// Note that this implementation does not verify correctness
/// of sent extrinsics. It's meant to be used in contexts
/// where an actual runtime is not known.
///
/// It's advised to write integration tests that include the
/// actual transaction pool to make sure the produced
/// transactions are valid.
#[derive(Default)]
pub struct TestTransactionPoolExt(Arc<RwLock<PoolState>>);
impl TestTransactionPoolExt {
/// Create new `TestTransactionPoolExt` and a reference to the internal state.
pub fn new() -> (Self, Arc<RwLock<PoolState>>) {
let ext = Self::default();
let state = ext.0.clone();
(ext, state)
}
}
impl TransactionPool for TestTransactionPoolExt {
fn submit_transaction(&mut self, extrinsic: Vec<u8>) -> Result<(), ()> {
self.0.write().transactions.push(extrinsic);
Ok(())
}
}
+6 -2
View File
@@ -18,7 +18,11 @@
use crate::{crypto::KeyTypeId, ed25519, sr25519};
use std::{fmt::{Debug, Display}, panic::UnwindSafe};
use std::{
fmt::{Debug, Display},
panic::UnwindSafe,
sync::Arc,
};
pub use externalities::{Externalities, ExternalitiesExt};
@@ -68,7 +72,7 @@ pub trait BareCryptoStore: Send + Sync {
}
/// A pointer to the key store.
pub type BareCryptoStorePtr = std::sync::Arc<parking_lot::RwLock<dyn BareCryptoStore>>;
pub type BareCryptoStorePtr = Arc<parking_lot::RwLock<dyn BareCryptoStore>>;
externalities::decl_extension! {
/// The keystore extension to register/retrieve from the externalities.
+7 -3
View File
@@ -35,7 +35,10 @@ use rstd::ops::Deref;
#[cfg(feature = "std")]
use primitives::{
crypto::Pair, traits::KeystoreExt, offchain::OffchainExt, hexdisplay::HexDisplay,
crypto::Pair,
traits::KeystoreExt,
offchain::{OffchainExt, TransactionPoolExt},
hexdisplay::HexDisplay,
storage::ChildStorageKey,
};
@@ -424,8 +427,9 @@ pub trait Offchain {
///
/// The transaction will end up in the pool.
fn submit_transaction(&mut self, data: Vec<u8>) -> Result<(), ()> {
self.extension::<OffchainExt>()
.expect("submit_transaction can be called only in the offchain worker context")
self.extension::<TransactionPoolExt>()
.expect("submit_transaction can be called only in the offchain call context with
TransactionPool capabilities enabled")
.submit_transaction(data)
}
@@ -21,7 +21,6 @@ inherents = { package = "substrate-inherents", path = "../inherents", default-fe
[dev-dependencies]
serde_json = "1.0.41"
rand = "0.7.2"
substrate-offchain = { path = "../../client/offchain" }
[features]
bench = []
@@ -516,8 +516,10 @@ impl<'a> HeadersIterator<'a> {
mod tests {
use super::*;
use runtime_io::TestExternalities;
use substrate_offchain::testing;
use primitives::offchain::OffchainExt;
use primitives::offchain::{
OffchainExt,
testing,
};
#[test]
fn should_send_a_basic_request_and_get_response() {
@@ -14,6 +14,29 @@
// You should have received a copy of the GNU General Public License
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
//! A collection of higher lever helpers for offchain workers.
//! A collection of higher lever helpers for offchain calls.
use crate::{
traits,
generic::BlockId,
};
pub mod http;
/// An abstraction for transaction pool.
///
/// This trait is used by offchain calls to be able to submit transactions.
/// The main use case is for offchain workers, to feed back the results of computations,
/// but since the transaction pool access is a separate `ExternalitiesExtension` it can
/// be also used in context of other offchain calls. For one may generate and submit
/// a transaction for some misbehavior reports (say equivocation).
pub trait TransactionPool<Block: traits::Block>: Send + Sync {
/// Submit transaction.
///
/// The transaction will end up in the pool and be propagated to others.
fn submit_at(
&self,
at: &BlockId<Block>,
extrinsic: Block::Extrinsic,
) -> Result<(), ()>;
}
+19 -26
View File
@@ -23,7 +23,7 @@ use log::{warn, trace};
use hash_db::Hasher;
use codec::{Decode, Encode};
use primitives::{
storage::well_known_keys, NativeOrEncoded, NeverNativeValue, offchain::OffchainExt,
storage::well_known_keys, NativeOrEncoded, NeverNativeValue,
traits::{KeystoreExt, CodeExecutor}, hexdisplay::HexDisplay, hash::H256,
};
use overlayed_changes::OverlayedChangeSet;
@@ -68,7 +68,8 @@ pub use error::{Error, ExecutionError};
type CallResult<R, E> = Result<NativeOrEncoded<R>, E>;
type DefaultHandler<R, E> = fn(CallResult<R, E>, CallResult<R, E>) -> CallResult<R, E>;
/// Default handler of the execution manager.
pub type DefaultHandler<R, E> = fn(CallResult<R, E>, CallResult<R, E>) -> CallResult<R, E>;
/// Type of changes trie transaction.
pub type ChangesTrieTransaction<H, N> = (
@@ -185,23 +186,12 @@ impl<'a, B, H, N, T, Exec> StateMachine<'a, B, H, N, T, Exec> where
pub fn new(
backend: &'a B,
changes_trie_storage: Option<&'a T>,
offchain_ext: Option<OffchainExt>,
overlay: &'a mut OverlayedChanges,
exec: &'a Exec,
method: &'a str,
call_data: &'a [u8],
keystore: Option<KeystoreExt>,
extensions: Extensions,
) -> Self {
let mut extensions = Extensions::new();
if let Some(keystore) = keystore {
extensions.register(keystore);
}
if let Some(offchain) = offchain_ext {
extensions.register(offchain);
}
Self {
backend,
exec,
@@ -497,8 +487,12 @@ where
Exec: CodeExecutor,
{
let proving_backend = proving_backend::ProvingBackend::new(trie_backend);
let mut extensions = Extensions::new();
if let Some(keystore) = keystore {
extensions.register(keystore);
}
let mut sm = StateMachine::<_, H, _, InMemoryChangesTrieStorage<H, u64>, Exec>::new(
&proving_backend, None, None, overlay, exec, method, call_data, keystore,
&proving_backend, None, overlay, exec, method, call_data, extensions,
);
let (result, _, _) = sm.execute_using_consensus_failure_handler::<_, NeverNativeValue, fn() -> _>(
@@ -542,8 +536,12 @@ where
H: Hasher<Out=H256>,
Exec: CodeExecutor,
{
let mut extensions = Extensions::new();
if let Some(keystore) = keystore {
extensions.register(keystore);
}
let mut sm = StateMachine::<_, H, _, InMemoryChangesTrieStorage<H, u64>, Exec>::new(
trie_backend, None, None, overlay, exec, method, call_data, keystore,
trie_backend, None, overlay, exec, method, call_data, extensions,
);
sm.execute_using_consensus_failure_handler::<_, NeverNativeValue, fn() -> _>(
@@ -820,7 +818,6 @@ mod tests {
let mut state_machine = StateMachine::new(
&backend,
Some(&changes_trie_storage),
None,
&mut overlayed_changes,
&DummyCodeExecutor {
change_changes_trie_config: false,
@@ -830,7 +827,7 @@ mod tests {
},
"test",
&[],
None,
Default::default(),
);
assert_eq!(
@@ -849,7 +846,6 @@ mod tests {
let mut state_machine = StateMachine::new(
&backend,
Some(&changes_trie_storage),
None,
&mut overlayed_changes,
&DummyCodeExecutor {
change_changes_trie_config: false,
@@ -859,7 +855,7 @@ mod tests {
},
"test",
&[],
None,
Default::default(),
);
assert_eq!(state_machine.execute(ExecutionStrategy::NativeElseWasm).unwrap().0, vec![66]);
@@ -875,7 +871,6 @@ mod tests {
let mut state_machine = StateMachine::new(
&backend,
Some(&changes_trie_storage),
None,
&mut overlayed_changes,
&DummyCodeExecutor {
change_changes_trie_config: false,
@@ -885,7 +880,7 @@ mod tests {
},
"test",
&[],
None,
Default::default(),
);
assert!(
@@ -1085,7 +1080,6 @@ mod tests {
let mut state_machine = StateMachine::new(
&backend,
Some(&changes_trie_storage),
None,
&mut overlayed_changes,
&DummyCodeExecutor {
change_changes_trie_config: true,
@@ -1095,7 +1089,7 @@ mod tests {
},
"test",
&[],
None,
Default::default(),
);
assert!(state_machine.execute(ExecutionStrategy::NativeWhenPossible).is_err());
@@ -1110,7 +1104,6 @@ mod tests {
let mut state_machine = StateMachine::new(
&backend,
Some(&changes_trie_storage),
None,
&mut overlayed_changes,
&DummyCodeExecutor {
change_changes_trie_config: true,
@@ -1120,7 +1113,7 @@ mod tests {
},
"test",
&[],
None,
Default::default(),
);
assert!(state_machine.execute(ExecutionStrategy::NativeElseWasm).is_err());