mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-06-12 13:31:10 +00:00
DB-based blockchain data cache for light nodes (#251)
* use db in light clients * fixed comment * fixed grumbles * blockchain data cache * fixed grumbles * fix compilation * keep the last entry in the cache * fixed grumbles * fixed reporting of pruned entries
This commit is contained in:
committed by
Gav Wood
parent
fb2a8ada37
commit
859a420ae9
@@ -189,7 +189,7 @@ pub trait Proposer<B: Block> {
|
||||
/// Block import trait.
|
||||
pub trait BlockImport<B: Block> {
|
||||
/// Import a block alongside its corresponding justification.
|
||||
fn import_block(&self, block: B, justification: Justification<B::Hash>);
|
||||
fn import_block(&self, block: B, justification: Justification<B::Hash>, authorities: &[AuthorityId]);
|
||||
}
|
||||
|
||||
/// Trait for getting the authorities at a given block.
|
||||
@@ -308,7 +308,8 @@ impl<B, P, I, InStream, OutSink> Future for BftFuture<B, P, I, InStream, OutSink
|
||||
info!(target: "bft", "Importing block #{} ({}) directly from BFT consensus",
|
||||
justified_block.header().number(), justified_block.hash());
|
||||
|
||||
self.import.import_block(justified_block, committed.justification)
|
||||
self.import.import_block(justified_block, committed.justification,
|
||||
&self.inner.context().authorities)
|
||||
}
|
||||
|
||||
Ok(Async::Ready(()))
|
||||
@@ -649,7 +650,7 @@ mod tests {
|
||||
}
|
||||
|
||||
impl BlockImport<TestBlock> for FakeClient {
|
||||
fn import_block(&self, block: TestBlock, _justification: Justification<H256>) {
|
||||
fn import_block(&self, block: TestBlock, _justification: Justification<H256>, _authorities: &[AuthorityId]) {
|
||||
assert!(self.imported_heights.lock().insert(block.header.number))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,447 @@
|
||||
// Copyright 2017 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Polkadot.
|
||||
|
||||
// Polkadot 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.
|
||||
|
||||
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! DB-backed cache of blockchain data.
|
||||
|
||||
use std::sync::Arc;
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use kvdb::{KeyValueDB, DBTransaction};
|
||||
|
||||
use client::blockchain::Cache as BlockchainCache;
|
||||
use client::error::Result as ClientResult;
|
||||
use codec::{Codec, Encode, Decode, Input, Output};
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::generic::BlockId;
|
||||
use runtime_primitives::traits::{Block as BlockT, As, NumberFor};
|
||||
use utils::{COLUMN_META, BlockKey, db_err, meta_keys, read_id, db_key_to_number, number_to_db_key};
|
||||
|
||||
/// Database-backed cache of blockchain data.
|
||||
pub struct DbCache<Block: BlockT> {
|
||||
db: Arc<KeyValueDB>,
|
||||
block_index_column: Option<u32>,
|
||||
authorities_at: DbCacheList<Block, Vec<AuthorityId>>,
|
||||
}
|
||||
|
||||
impl<Block> DbCache<Block>
|
||||
where
|
||||
Block: BlockT,
|
||||
NumberFor<Block>: As<u64>,
|
||||
{
|
||||
/// Create new cache.
|
||||
pub fn new(
|
||||
db: Arc<KeyValueDB>,
|
||||
block_index_column: Option<u32>,
|
||||
authorities_column: Option<u32>
|
||||
) -> ClientResult<Self> {
|
||||
Ok(DbCache {
|
||||
db: db.clone(),
|
||||
block_index_column,
|
||||
authorities_at: DbCacheList::new(db, meta_keys::BEST_AUTHORITIES, authorities_column)?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Get authorities_cache.
|
||||
pub fn authorities_at_cache(&self) -> &DbCacheList<Block, Vec<AuthorityId>> {
|
||||
&self.authorities_at
|
||||
}
|
||||
}
|
||||
|
||||
impl<Block> BlockchainCache<Block> for DbCache<Block>
|
||||
where
|
||||
Block: BlockT,
|
||||
NumberFor<Block>: As<u64>,
|
||||
{
|
||||
fn authorities_at(&self, at: BlockId<Block>) -> Option<Vec<AuthorityId>> {
|
||||
let authorities_at = read_id(&*self.db, self.block_index_column, at).and_then(|at| match at {
|
||||
Some(at) => self.authorities_at.value_at_key(at),
|
||||
None => Ok(None),
|
||||
});
|
||||
|
||||
match authorities_at {
|
||||
Ok(authorities) => authorities,
|
||||
Err(error) => {
|
||||
warn!("Trying to read authorities from db cache has failed with: {}", error);
|
||||
None
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Database-backed blockchain cache which holds its entries as a list.
|
||||
/// The meta column holds the pointer to the best known cache entry and
|
||||
/// every entry points to the previous entry.
|
||||
/// New entry appears when the set of authorities changes in block, so the
|
||||
/// best entry here means the entry that is valid for the best block (and
|
||||
/// probably for its ascendants).
|
||||
pub struct DbCacheList<Block: BlockT, T: Clone> {
|
||||
db: Arc<KeyValueDB>,
|
||||
meta_key: &'static [u8],
|
||||
column: Option<u32>,
|
||||
/// Best entry at the moment. None means that cache has no entries at all.
|
||||
best_entry: RwLock<Option<Entry<NumberFor<Block>, T>>>,
|
||||
}
|
||||
|
||||
/// Single cache entry.
|
||||
#[derive(Clone)]
|
||||
#[cfg_attr(test, derive(Debug, PartialEq))]
|
||||
pub struct Entry<N, T: Clone> {
|
||||
/// first block, when this value became actual
|
||||
valid_from: N,
|
||||
/// None means that we do not know the value starting from `valid_from` block
|
||||
value: Option<T>,
|
||||
}
|
||||
|
||||
/// Internal representation of the single cache entry. The entry points to the
|
||||
/// previous entry in the cache, allowing us to traverse back in time in list-style.
|
||||
#[cfg_attr(test, derive(Debug, PartialEq))]
|
||||
struct StorageEntry<N, T> {
|
||||
/// None if valid from the beginning
|
||||
prev_valid_from: Option<N>,
|
||||
/// None means that we do not know the value starting from `valid_from` block
|
||||
value: Option<T>,
|
||||
}
|
||||
|
||||
impl<Block, T> DbCacheList<Block, T>
|
||||
where
|
||||
Block: BlockT,
|
||||
NumberFor<Block>: As<u64>,
|
||||
T: Clone + PartialEq + Codec,
|
||||
{
|
||||
/// Creates new cache list.
|
||||
fn new(db: Arc<KeyValueDB>, meta_key: &'static [u8], column: Option<u32>) -> ClientResult<Self> {
|
||||
let best_entry = RwLock::new(db.get(COLUMN_META, meta_key)
|
||||
.map_err(db_err)
|
||||
.and_then(|block| match block {
|
||||
Some(block) => {
|
||||
let valid_from = db_key_to_number(&block)?;
|
||||
read_storage_entry::<Block, T>(&*db, column, valid_from)
|
||||
.map(|entry| Some(Entry {
|
||||
valid_from,
|
||||
value: entry
|
||||
.expect("meta entry references the entry at the block; storage entry at block exists when referenced; qed")
|
||||
.value,
|
||||
}))
|
||||
},
|
||||
None => Ok(None),
|
||||
})?);
|
||||
|
||||
Ok(DbCacheList {
|
||||
db,
|
||||
column,
|
||||
meta_key,
|
||||
best_entry,
|
||||
})
|
||||
}
|
||||
|
||||
/// Gets the best known entry.
|
||||
pub fn best_entry(&self) -> Option<Entry<NumberFor<Block>, T>> {
|
||||
self.best_entry.read().clone()
|
||||
}
|
||||
|
||||
/// Commits the new best pending value to the database. Returns Some if best entry must
|
||||
/// be updated after transaction is committed.
|
||||
pub fn commit_best_entry(
|
||||
&self,
|
||||
transaction: &mut DBTransaction,
|
||||
valid_from: NumberFor<Block>,
|
||||
pending_value: Option<T>
|
||||
) -> Option<Entry<NumberFor<Block>, T>> {
|
||||
let best_entry = self.best_entry();
|
||||
let update_best_entry = match (
|
||||
best_entry.as_ref().and_then(|a| a.value.as_ref()),
|
||||
pending_value.as_ref()
|
||||
) {
|
||||
(Some(best_value), Some(pending_value)) => best_value != pending_value,
|
||||
(None, Some(_)) | (Some(_), None) => true,
|
||||
(None, None) => false,
|
||||
};
|
||||
if !update_best_entry {
|
||||
return None;
|
||||
}
|
||||
|
||||
let valid_from_key = number_to_db_key(valid_from);
|
||||
transaction.put(COLUMN_META, self.meta_key, &valid_from_key);
|
||||
transaction.put(self.column, &valid_from_key, &StorageEntry {
|
||||
prev_valid_from: best_entry.map(|b| b.valid_from),
|
||||
value: pending_value.clone(),
|
||||
}.encode());
|
||||
|
||||
Some(Entry {
|
||||
valid_from,
|
||||
value: pending_value,
|
||||
})
|
||||
}
|
||||
|
||||
/// Updates the best in-memory cache entry. Must be called after transaction with changes
|
||||
/// from commit_best_entry has been committed.
|
||||
pub fn update_best_entry(&self, best_entry: Option<Entry<NumberFor<Block>, T>>) {
|
||||
*self.best_entry.write() = best_entry;
|
||||
}
|
||||
|
||||
/// Prune all entries from the beginning up to the block (including entry at the number). Returns
|
||||
/// the number of pruned entries. Pruning never deletes the latest entry in the cache.
|
||||
pub fn prune_entries(
|
||||
&self,
|
||||
transaction: &mut DBTransaction,
|
||||
last_to_prune: NumberFor<Block>
|
||||
) -> ClientResult<usize> {
|
||||
// find the last entry we want to keep
|
||||
let mut last_entry_to_keep = match self.best_entry() {
|
||||
Some(best_entry) => best_entry.valid_from,
|
||||
None => return Ok(0),
|
||||
};
|
||||
let mut first_entry_to_remove = last_entry_to_keep;
|
||||
while first_entry_to_remove > last_to_prune {
|
||||
last_entry_to_keep = first_entry_to_remove;
|
||||
|
||||
let entry = read_storage_entry::<Block, T>(&*self.db, self.column, first_entry_to_remove)?
|
||||
.expect("entry referenced from the next entry; entry exists when referenced; qed");
|
||||
// if we have reached the first list entry
|
||||
// AND all list entries are for blocks that are later than last_to_prune
|
||||
// => nothing to prune
|
||||
first_entry_to_remove = match entry.prev_valid_from {
|
||||
Some(prev_valid_from) => prev_valid_from,
|
||||
None => return Ok(0),
|
||||
}
|
||||
}
|
||||
|
||||
// remove all entries, starting from entry_to_remove
|
||||
let mut pruned = 0;
|
||||
let mut entry_to_remove = Some(first_entry_to_remove);
|
||||
while let Some(current_entry) = entry_to_remove {
|
||||
let entry = read_storage_entry::<Block, T>(&*self.db, self.column, current_entry)?
|
||||
.expect("referenced entry exists; entry_to_remove is a reference to the entry; qed");
|
||||
|
||||
if current_entry != last_entry_to_keep {
|
||||
transaction.delete(self.column, &number_to_db_key(current_entry));
|
||||
pruned += 1;
|
||||
}
|
||||
entry_to_remove = entry.prev_valid_from;
|
||||
}
|
||||
|
||||
let mut entry = read_storage_entry::<Block, T>(&*self.db, self.column, last_entry_to_keep)?
|
||||
.expect("last_entry_to_keep >= first_entry_to_remove; that means that we're leaving this entry in the db; qed");
|
||||
entry.prev_valid_from = None;
|
||||
transaction.put(self.column, &number_to_db_key(last_entry_to_keep), &entry.encode());
|
||||
|
||||
Ok(pruned)
|
||||
}
|
||||
|
||||
/// Reads the cached value, actual at given block. Returns None if the value was not cached
|
||||
/// or if it has been pruned.
|
||||
fn value_at_key(&self, key: BlockKey) -> ClientResult<Option<T>> {
|
||||
let at = db_key_to_number::<NumberFor<Block>>(&key)?;
|
||||
let best_valid_from = match self.best_entry() {
|
||||
// there are entries in cache
|
||||
Some(best_entry) => {
|
||||
// we're looking for the best value
|
||||
if at >= best_entry.valid_from {
|
||||
return Ok(best_entry.value);
|
||||
}
|
||||
|
||||
// we're looking for the value of older blocks
|
||||
best_entry.valid_from
|
||||
},
|
||||
// there are no entries in the cache
|
||||
None => return Ok(None),
|
||||
};
|
||||
|
||||
let mut entry = read_storage_entry::<Block, T>(&*self.db, self.column, best_valid_from)?
|
||||
.expect("self.best_entry().is_some() if there's entry for best_valid_from; qed");
|
||||
loop {
|
||||
let prev_valid_from = match entry.prev_valid_from {
|
||||
Some(prev_valid_from) => prev_valid_from,
|
||||
None => return Ok(None),
|
||||
};
|
||||
|
||||
let prev_entry = read_storage_entry::<Block, T>(&*self.db, self.column, prev_valid_from)?
|
||||
.expect("entry referenced from the next entry; entry exists when referenced; qed");
|
||||
if at >= prev_valid_from {
|
||||
return Ok(prev_entry.value);
|
||||
}
|
||||
|
||||
entry = prev_entry;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads the entry at the block with given number.
|
||||
fn read_storage_entry<Block, T>(
|
||||
db: &KeyValueDB,
|
||||
column: Option<u32>,
|
||||
number: NumberFor<Block>
|
||||
) -> ClientResult<Option<StorageEntry<NumberFor<Block>, T>>>
|
||||
where
|
||||
Block: BlockT,
|
||||
NumberFor<Block>: As<u64>,
|
||||
T: Codec,
|
||||
{
|
||||
db.get(column, &number_to_db_key(number))
|
||||
.and_then(|entry| match entry {
|
||||
Some(entry) => Ok(StorageEntry::<NumberFor<Block>, T>::decode(&mut &entry[..])),
|
||||
None => Ok(None),
|
||||
})
|
||||
.map_err(db_err)
|
||||
}
|
||||
|
||||
impl<N: Encode, T: Encode> Encode for StorageEntry<N, T> {
|
||||
fn encode_to<O: Output>(&self, dest: &mut O) {
|
||||
dest.push(&self.prev_valid_from);
|
||||
dest.push(&self.value);
|
||||
}
|
||||
}
|
||||
|
||||
impl<N: Decode, T: Decode> Decode for StorageEntry<N, T> {
|
||||
fn decode<I: Input>(input: &mut I) -> Option<Self> {
|
||||
Some(StorageEntry {
|
||||
prev_valid_from: Decode::decode(input)?,
|
||||
value: Decode::decode(input)?,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use runtime_primitives::testing::Block as RawBlock;
|
||||
use light::{AUTHORITIES_ENTRIES_TO_KEEP, columns, LightStorage};
|
||||
use light::tests::insert_block;
|
||||
use super::*;
|
||||
|
||||
type Block = RawBlock<u64>;
|
||||
|
||||
#[test]
|
||||
fn authorities_storage_entry_serialized() {
|
||||
let test_cases: Vec<StorageEntry<u64, Vec<AuthorityId>>> = vec![
|
||||
StorageEntry { prev_valid_from: Some(42), value: Some(vec![[1u8; 32].into()]) },
|
||||
StorageEntry { prev_valid_from: None, value: Some(vec![[1u8; 32].into(), [2u8; 32].into()]) },
|
||||
StorageEntry { prev_valid_from: None, value: None },
|
||||
];
|
||||
|
||||
for expected in test_cases {
|
||||
let serialized = expected.encode();
|
||||
let deserialized = StorageEntry::decode(&mut &serialized[..]).unwrap();
|
||||
assert_eq!(expected, deserialized);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn best_authorities_are_updated() {
|
||||
let db = LightStorage::new_test();
|
||||
let authorities_at: Vec<(usize, Option<Entry<u64, Vec<AuthorityId>>>)> = vec![
|
||||
(0, None),
|
||||
(0, None),
|
||||
(1, Some(Entry { valid_from: 1, value: Some(vec![[2u8; 32].into()]) })),
|
||||
(1, Some(Entry { valid_from: 1, value: Some(vec![[2u8; 32].into()]) })),
|
||||
(2, Some(Entry { valid_from: 3, value: Some(vec![[4u8; 32].into()]) })),
|
||||
(2, Some(Entry { valid_from: 3, value: Some(vec![[4u8; 32].into()]) })),
|
||||
(3, Some(Entry { valid_from: 5, value: None })),
|
||||
(3, Some(Entry { valid_from: 5, value: None })),
|
||||
];
|
||||
|
||||
// before any block, there are no entries in cache
|
||||
assert!(db.cache().authorities_at_cache().best_entry().is_none());
|
||||
assert_eq!(db.db().iter(columns::AUTHORITIES).count(), 0);
|
||||
|
||||
// insert blocks and check that best_authorities() returns correct result
|
||||
let mut prev_hash = Default::default();
|
||||
for number in 0..authorities_at.len() {
|
||||
let authorities_at_number = authorities_at[number].1.clone().and_then(|e| e.value);
|
||||
prev_hash = insert_block(&db, &prev_hash, number as u64, authorities_at_number);
|
||||
assert_eq!(db.cache().authorities_at_cache().best_entry(), authorities_at[number].1);
|
||||
assert_eq!(db.db().iter(columns::AUTHORITIES).count(), authorities_at[number].0);
|
||||
}
|
||||
|
||||
// check that authorities_at() returns correct results for all retrospective blocks
|
||||
for number in 1..authorities_at.len() + 1 {
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(number as u64)),
|
||||
authorities_at.get(number + 1)
|
||||
.or_else(|| authorities_at.last())
|
||||
.unwrap().1.clone().and_then(|e| e.value));
|
||||
}
|
||||
|
||||
// now check that cache entries are pruned when new blocks are inserted
|
||||
let mut current_entries_count = authorities_at.last().unwrap().0;
|
||||
let pruning_starts_at = AUTHORITIES_ENTRIES_TO_KEEP as usize;
|
||||
for number in authorities_at.len()..authorities_at.len() + pruning_starts_at {
|
||||
prev_hash = insert_block(&db, &prev_hash, number as u64, None);
|
||||
if number > pruning_starts_at {
|
||||
let prev_entries_count = authorities_at[number - pruning_starts_at].0;
|
||||
let entries_count = authorities_at.get(number - pruning_starts_at + 1).map(|e| e.0)
|
||||
.unwrap_or_else(|| authorities_at.last().unwrap().0);
|
||||
current_entries_count -= entries_count - prev_entries_count;
|
||||
}
|
||||
|
||||
// there's always at least 1 entry in the cache (after first insertion)
|
||||
assert_eq!(db.db().iter(columns::AUTHORITIES).count(), ::std::cmp::max(current_entries_count, 1));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn best_authorities_are_pruned() {
|
||||
let db = LightStorage::<Block>::new_test();
|
||||
let mut transaction = DBTransaction::new();
|
||||
|
||||
// insert first entry at block#100
|
||||
db.cache().authorities_at_cache().update_best_entry(
|
||||
db.cache().authorities_at_cache().commit_best_entry(&mut transaction, 100, Some(vec![[1u8; 32].into()])));
|
||||
db.db().write(transaction).unwrap();
|
||||
|
||||
// no entries are pruned, since there's only one entry in the cache
|
||||
let mut transaction = DBTransaction::new();
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 50).unwrap(), 0);
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 100).unwrap(), 0);
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 150).unwrap(), 0);
|
||||
|
||||
// insert second entry at block#200
|
||||
let mut transaction = DBTransaction::new();
|
||||
db.cache().authorities_at_cache().update_best_entry(
|
||||
db.cache().authorities_at_cache().commit_best_entry(&mut transaction, 200, Some(vec![[2u8; 32].into()])));
|
||||
db.db().write(transaction).unwrap();
|
||||
|
||||
let mut transaction = DBTransaction::new();
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 50).unwrap(), 0);
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 100).unwrap(), 1);
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 150).unwrap(), 1);
|
||||
// still only 1 entry is removed since pruning never deletes the last entry
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 200).unwrap(), 1);
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 250).unwrap(), 1);
|
||||
|
||||
// physically remove entry for block#100 from db
|
||||
let mut transaction = DBTransaction::new();
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 150).unwrap(), 1);
|
||||
db.db().write(transaction).unwrap();
|
||||
|
||||
assert_eq!(db.cache().authorities_at_cache().best_entry().unwrap().value, Some(vec![[2u8; 32].into()]));
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(50)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(100)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(150)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(200)), Some(vec![[2u8; 32].into()]));
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(250)), Some(vec![[2u8; 32].into()]));
|
||||
|
||||
// try to delete last entry => failure (no entries are removed)
|
||||
let mut transaction = DBTransaction::new();
|
||||
assert_eq!(db.cache().authorities_at_cache().prune_entries(&mut transaction, 300).unwrap(), 0);
|
||||
db.db().write(transaction).unwrap();
|
||||
|
||||
assert_eq!(db.cache().authorities_at_cache().best_entry().unwrap().value, Some(vec![[2u8; 32].into()]));
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(50)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(100)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(150)), None);
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(200)), Some(vec![[2u8; 32].into()]));
|
||||
assert_eq!(db.cache().authorities_at(BlockId::Number(250)), Some(vec![[2u8; 32].into()]));
|
||||
}
|
||||
}
|
||||
@@ -38,6 +38,7 @@ extern crate kvdb_memorydb;
|
||||
|
||||
pub mod light;
|
||||
|
||||
mod cache;
|
||||
mod utils;
|
||||
|
||||
use std::sync::Arc;
|
||||
@@ -47,7 +48,7 @@ use codec::{Decode, Encode};
|
||||
use kvdb::{KeyValueDB, DBTransaction};
|
||||
use memorydb::MemoryDB;
|
||||
use parking_lot::RwLock;
|
||||
use primitives::H256;
|
||||
use primitives::{H256, AuthorityId};
|
||||
use runtime_primitives::generic::BlockId;
|
||||
use runtime_primitives::bft::Justification;
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, As, Hash, HashFor, NumberFor, Zero};
|
||||
@@ -204,6 +205,10 @@ impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> {
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
fn cache(&self) -> Option<&client::blockchain::Cache<Block>> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Database transaction
|
||||
@@ -231,6 +236,10 @@ impl<Block: BlockT> client::backend::BlockImportOperation<Block> for BlockImport
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn update_authorities(&mut self, _authorities: Vec<AuthorityId>) {
|
||||
// currently authorities are not cached on full nodes
|
||||
}
|
||||
|
||||
fn update_storage(&mut self, update: MemoryDB) -> Result<(), client::error::Error> {
|
||||
self.updates = update;
|
||||
Ok(())
|
||||
|
||||
@@ -21,14 +21,15 @@ use parking_lot::RwLock;
|
||||
|
||||
use kvdb::{KeyValueDB, DBTransaction};
|
||||
|
||||
use client::blockchain::{BlockStatus, HeaderBackend as BlockchainHeaderBackend,
|
||||
Info as BlockchainInfo};
|
||||
use client::blockchain::{BlockStatus, Cache as BlockchainCache,
|
||||
HeaderBackend as BlockchainHeaderBackend, Info as BlockchainInfo};
|
||||
use client::error::{ErrorKind as ClientErrorKind, Result as ClientResult};
|
||||
use client::light::blockchain::Storage as LightBlockchainStorage;
|
||||
use codec::{Decode, Encode};
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::generic::BlockId;
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Hash, HashFor, Zero};
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Hash, HashFor, Zero, As};
|
||||
use cache::DbCache;
|
||||
use utils::{meta_keys, Meta, db_err, number_to_db_key, open_database, read_db, read_id, read_meta};
|
||||
use DatabaseSettings;
|
||||
|
||||
@@ -36,12 +37,17 @@ pub(crate) mod columns {
|
||||
pub const META: Option<u32> = ::utils::COLUMN_META;
|
||||
pub const BLOCK_INDEX: Option<u32> = Some(1);
|
||||
pub const HEADER: Option<u32> = Some(2);
|
||||
pub const AUTHORITIES: Option<u32> = Some(3);
|
||||
}
|
||||
|
||||
/// Keep authorities for last 'AUTHORITIES_ENTRIES_TO_KEEP' blocks.
|
||||
pub(crate) const AUTHORITIES_ENTRIES_TO_KEEP: u64 = 2048;
|
||||
|
||||
/// Light blockchain storage. Stores most recent headers + CHTs for older headers.
|
||||
pub struct LightStorage<Block: BlockT> {
|
||||
db: Arc<KeyValueDB>,
|
||||
meta: RwLock<Meta<<<Block as BlockT>::Header as HeaderT>::Number, Block::Hash>>,
|
||||
cache: DbCache<Block>,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Debug)]
|
||||
@@ -73,14 +79,26 @@ impl<Block> LightStorage<Block>
|
||||
}
|
||||
|
||||
fn from_kvdb(db: Arc<KeyValueDB>) -> ClientResult<Self> {
|
||||
let cache = DbCache::new(db.clone(), columns::BLOCK_INDEX, columns::AUTHORITIES)?;
|
||||
let meta = RwLock::new(read_meta::<Block>(&*db, columns::HEADER)?);
|
||||
|
||||
Ok(LightStorage {
|
||||
db,
|
||||
meta,
|
||||
cache,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn db(&self) -> &Arc<KeyValueDB> {
|
||||
&self.db
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn cache(&self) -> &DbCache<Block> {
|
||||
&self.cache
|
||||
}
|
||||
|
||||
fn update_meta(&self, hash: Block::Hash, number: <<Block as BlockT>::Header as HeaderT>::Number, is_best: bool) {
|
||||
if is_best {
|
||||
let mut meta = self.meta.write();
|
||||
@@ -139,7 +157,7 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
|
||||
where
|
||||
Block: BlockT,
|
||||
{
|
||||
fn import_header(&self, is_new_best: bool, header: Block::Header) -> ClientResult<()> {
|
||||
fn import_header(&self, is_new_best: bool, header: Block::Header, authorities: Option<Vec<AuthorityId>>) -> ClientResult<()> {
|
||||
let mut transaction = DBTransaction::new();
|
||||
|
||||
let hash = header.hash();
|
||||
@@ -149,16 +167,39 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
|
||||
transaction.put(columns::HEADER, &key, &header.encode());
|
||||
transaction.put(columns::BLOCK_INDEX, hash.as_ref(), &key);
|
||||
|
||||
if is_new_best {
|
||||
let best_authorities = if is_new_best {
|
||||
transaction.put(columns::META, meta_keys::BEST_BLOCK, &key);
|
||||
}
|
||||
|
||||
// cache authorities for previous block
|
||||
let number: u64 = number.as_();
|
||||
let previous_number = number.checked_sub(1);
|
||||
let best_authorities = previous_number
|
||||
.and_then(|previous_number| self.cache.authorities_at_cache()
|
||||
.commit_best_entry(&mut transaction, As::sa(previous_number), authorities));
|
||||
|
||||
// prune authorities from 'ancient' blocks
|
||||
if let Some(ancient_number) = number.checked_sub(AUTHORITIES_ENTRIES_TO_KEEP) {
|
||||
self.cache.authorities_at_cache().prune_entries(&mut transaction, As::sa(ancient_number))?;
|
||||
}
|
||||
|
||||
best_authorities
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
debug!("Light DB Commit {:?} ({})", hash, number);
|
||||
self.db.write(transaction).map_err(db_err)?;
|
||||
self.update_meta(hash, number, is_new_best);
|
||||
if let Some(best_authorities) = best_authorities {
|
||||
self.cache.authorities_at_cache().update_best_entry(Some(best_authorities));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn cache(&self) -> Option<&BlockchainCache<Block>> {
|
||||
Some(&self.cache)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -168,7 +209,12 @@ pub(crate) mod tests {
|
||||
|
||||
type Block = RawBlock<u32>;
|
||||
|
||||
pub fn insert_block(db: &LightStorage<Block>, parent: &Hash, number: u32) -> Hash {
|
||||
pub fn insert_block(
|
||||
db: &LightStorage<Block>,
|
||||
parent: &Hash,
|
||||
number: u64,
|
||||
authorities: Option<Vec<AuthorityId>>
|
||||
) -> Hash {
|
||||
let header = Header {
|
||||
number: number.into(),
|
||||
parent_hash: *parent,
|
||||
@@ -178,14 +224,14 @@ pub(crate) mod tests {
|
||||
};
|
||||
|
||||
let hash = header.hash();
|
||||
db.import_header(true, header).unwrap();
|
||||
db.import_header(true, header, authorities).unwrap();
|
||||
hash
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn returns_known_header() {
|
||||
let db = LightStorage::new_test();
|
||||
let known_hash = insert_block(&db, &Default::default(), 0);
|
||||
let known_hash = insert_block(&db, &Default::default(), 0, None);
|
||||
let header_by_hash = db.header(BlockId::Hash(known_hash)).unwrap().unwrap();
|
||||
let header_by_number = db.header(BlockId::Number(0)).unwrap().unwrap();
|
||||
assert_eq!(header_by_hash, header_by_number);
|
||||
@@ -201,12 +247,12 @@ pub(crate) mod tests {
|
||||
#[test]
|
||||
fn returns_info() {
|
||||
let db = LightStorage::new_test();
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0);
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0, None);
|
||||
let info = db.info().unwrap();
|
||||
assert_eq!(info.best_hash, genesis_hash);
|
||||
assert_eq!(info.best_number, 0);
|
||||
assert_eq!(info.genesis_hash, genesis_hash);
|
||||
let best_hash = insert_block(&db, &genesis_hash, 1);
|
||||
let best_hash = insert_block(&db, &genesis_hash, 1, None);
|
||||
let info = db.info().unwrap();
|
||||
assert_eq!(info.best_hash, best_hash);
|
||||
assert_eq!(info.best_number, 1);
|
||||
@@ -216,7 +262,7 @@ pub(crate) mod tests {
|
||||
#[test]
|
||||
fn returns_block_status() {
|
||||
let db = LightStorage::new_test();
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0);
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0, None);
|
||||
assert_eq!(db.status(BlockId::Hash(genesis_hash)).unwrap(), BlockStatus::InChain);
|
||||
assert_eq!(db.status(BlockId::Number(0)).unwrap(), BlockStatus::InChain);
|
||||
assert_eq!(db.status(BlockId::Hash(1.into())).unwrap(), BlockStatus::Unknown);
|
||||
@@ -226,7 +272,7 @@ pub(crate) mod tests {
|
||||
#[test]
|
||||
fn returns_block_hash() {
|
||||
let db = LightStorage::new_test();
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0);
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0, None);
|
||||
assert_eq!(db.hash(0).unwrap(), Some(genesis_hash));
|
||||
assert_eq!(db.hash(1).unwrap(), None);
|
||||
}
|
||||
@@ -235,11 +281,11 @@ pub(crate) mod tests {
|
||||
fn import_header_works() {
|
||||
let db = LightStorage::new_test();
|
||||
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0);
|
||||
let genesis_hash = insert_block(&db, &Default::default(), 0, None);
|
||||
assert_eq!(db.db.iter(columns::HEADER).count(), 1);
|
||||
assert_eq!(db.db.iter(columns::BLOCK_INDEX).count(), 1);
|
||||
|
||||
let _ = insert_block(&db, &genesis_hash, 1);
|
||||
let _ = insert_block(&db, &genesis_hash, 1, None);
|
||||
assert_eq!(db.db.iter(columns::HEADER).count(), 2);
|
||||
assert_eq!(db.db.iter(columns::BLOCK_INDEX).count(), 2);
|
||||
}
|
||||
|
||||
@@ -41,6 +41,8 @@ pub mod meta_keys {
|
||||
pub const TYPE: &[u8; 4] = b"type";
|
||||
/// Best block key.
|
||||
pub const BEST_BLOCK: &[u8; 4] = b"best";
|
||||
/// Best authorities block key.
|
||||
pub const BEST_AUTHORITIES: &[u8; 4] = b"auth";
|
||||
}
|
||||
|
||||
/// Database metadata.
|
||||
@@ -69,6 +71,17 @@ pub fn number_to_db_key<N>(n: N) -> BlockKey where N: As<u64> {
|
||||
]
|
||||
}
|
||||
|
||||
/// Convert block key into block number.
|
||||
pub fn db_key_to_number<N>(key: &[u8]) -> client::error::Result<N> where N: As<u64> {
|
||||
match key.len() {
|
||||
4 => Ok((key[0] as u64) << 24
|
||||
| (key[1] as u64) << 16
|
||||
| (key[2] as u64) << 8
|
||||
| (key[3] as u64)).map(As::sa),
|
||||
_ => Err(client::error::ErrorKind::Backend("Invalid block key".into()).into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps database error to client error
|
||||
pub fn db_err(err: kvdb::Error) -> client::error::Error {
|
||||
use std::error::Error;
|
||||
|
||||
@@ -18,6 +18,7 @@
|
||||
|
||||
use state_machine::backend::Backend as StateBackend;
|
||||
use error;
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::bft::Justification;
|
||||
use runtime_primitives::traits::{Block as BlockT, NumberFor};
|
||||
use runtime_primitives::generic::BlockId;
|
||||
@@ -38,6 +39,9 @@ pub trait BlockImportOperation<Block: BlockT> {
|
||||
is_new_best: bool
|
||||
) -> error::Result<()>;
|
||||
|
||||
/// Append authorities set to the transaction. This is a set of parent block (set which
|
||||
/// has been used to check justification of this block).
|
||||
fn update_authorities(&mut self, authorities: Vec<AuthorityId>);
|
||||
/// Inject storage data into the database.
|
||||
fn update_storage(&mut self, update: <Self::State as StateBackend>::Transaction) -> error::Result<()>;
|
||||
/// Inject storage data into the database replacing any existing data.
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
//! Polkadot blockchain trait
|
||||
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
|
||||
use runtime_primitives::generic::BlockId;
|
||||
use runtime_primitives::bft::Justification;
|
||||
@@ -40,6 +41,15 @@ pub trait Backend<Block: BlockT>: HeaderBackend<Block> {
|
||||
fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<<Block as BlockT>::Extrinsic>>>;
|
||||
/// Get block justification. Returns `None` if justification does not exist.
|
||||
fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification<Block::Hash>>>;
|
||||
|
||||
/// Returns data cache reference, if it is enabled on this backend.
|
||||
fn cache(&self) -> Option<&Cache<Block>>;
|
||||
}
|
||||
|
||||
/// Blockchain optional data cache.
|
||||
pub trait Cache<Block: BlockT>: Send + Sync {
|
||||
/// Returns the set of authorities, that was active at given block or None if there's no entry in the cache.
|
||||
fn authorities_at(&self, block: BlockId<Block>) -> Option<Vec<AuthorityId>>;
|
||||
}
|
||||
|
||||
/// Block import outcome
|
||||
|
||||
@@ -133,12 +133,13 @@ pub struct BlockImportNotification<Block: BlockT> {
|
||||
pub struct JustifiedHeader<Block: BlockT> {
|
||||
header: <Block as BlockT>::Header,
|
||||
justification: ::bft::Justification<Block::Hash>,
|
||||
authorities: Vec<AuthorityId>,
|
||||
}
|
||||
|
||||
impl<Block: BlockT> JustifiedHeader<Block> {
|
||||
/// Deconstruct the justified header into parts.
|
||||
pub fn into_inner(self) -> (<Block as BlockT>::Header, ::bft::Justification<Block::Hash>) {
|
||||
(self.header, self.justification)
|
||||
pub fn into_inner(self) -> (<Block as BlockT>::Header, ::bft::Justification<Block::Hash>, Vec<AuthorityId>) {
|
||||
(self.header, self.justification, self.authorities)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -213,9 +214,12 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
|
||||
/// Get the set of authorities at a given block.
|
||||
pub fn authorities_at(&self, id: &BlockId<Block>) -> error::Result<Vec<AuthorityId>> {
|
||||
self.executor.call(id, "authorities", &[])
|
||||
.and_then(|r| Vec::<AuthorityId>::decode(&mut &r.return_data[..])
|
||||
.ok_or(error::ErrorKind::AuthLenInvalid.into()))
|
||||
match self.backend.blockchain().cache().and_then(|cache| cache.authorities_at(*id)) {
|
||||
Some(cached_value) => Ok(cached_value),
|
||||
None => self.executor.call(id, "authorities",&[])
|
||||
.and_then(|r| Vec::<AuthorityId>::decode(&mut &r.return_data[..])
|
||||
.ok_or(error::ErrorKind::AuthLenInvalid.into()))
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the RuntimeVersion at a given block.
|
||||
@@ -281,6 +285,7 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
Ok(JustifiedHeader {
|
||||
header,
|
||||
justification: just,
|
||||
authorities,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -291,7 +296,7 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
header: JustifiedHeader<Block>,
|
||||
body: Option<Vec<<Block as BlockT>::Extrinsic>>,
|
||||
) -> error::Result<ImportResult> {
|
||||
let (header, justification) = header.into_inner();
|
||||
let (header, justification, authorities) = header.into_inner();
|
||||
let parent_hash = header.parent_hash().clone();
|
||||
match self.backend.blockchain().status(BlockId::Hash(parent_hash))? {
|
||||
blockchain::BlockStatus::InChain => {},
|
||||
@@ -301,7 +306,7 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
let _import_lock = self.import_lock.lock();
|
||||
let height: u64 = header.number().as_();
|
||||
*self.importing_block.write() = Some(hash);
|
||||
let result = self.execute_and_import_block(origin, hash, header, justification, body);
|
||||
let result = self.execute_and_import_block(origin, hash, header, justification, body, authorities);
|
||||
*self.importing_block.write() = None;
|
||||
telemetry!("block.import";
|
||||
"height" => height,
|
||||
@@ -318,6 +323,7 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
header: Block::Header,
|
||||
justification: bft::Justification<Block::Hash>,
|
||||
body: Option<Vec<Block::Extrinsic>>,
|
||||
authorities: Vec<AuthorityId>,
|
||||
) -> error::Result<ImportResult> {
|
||||
let parent_hash = header.parent_hash().clone();
|
||||
match self.backend.blockchain().status(BlockId::Hash(hash))? {
|
||||
@@ -362,6 +368,7 @@ impl<B, E, Block> Client<B, E, Block> where
|
||||
trace!("Imported {}, (#{}), best={}, origin={:?}", hash, header.number(), is_new_best, origin);
|
||||
let unchecked: bft::UncheckedJustification<_> = justification.uncheck().into();
|
||||
transaction.set_block_data(header.clone(), body, Some(unchecked.into()), is_new_best)?;
|
||||
transaction.update_authorities(authorities);
|
||||
if let Some(storage_update) = storage_update {
|
||||
transaction.update_storage(storage_update)?;
|
||||
}
|
||||
@@ -467,11 +474,17 @@ impl<B, E, Block> bft::BlockImport<Block> for Client<B, E, Block>
|
||||
E: CallExecutor<Block>,
|
||||
Block: BlockT,
|
||||
{
|
||||
fn import_block(&self, block: Block, justification: ::bft::Justification<Block::Hash>) {
|
||||
fn import_block(
|
||||
&self,
|
||||
block: Block,
|
||||
justification: ::bft::Justification<Block::Hash>,
|
||||
authorities: &[AuthorityId]
|
||||
) {
|
||||
let (header, extrinsics) = block.deconstruct();
|
||||
let justified_header = JustifiedHeader {
|
||||
header: header,
|
||||
justification,
|
||||
authorities: authorities.to_vec(),
|
||||
};
|
||||
|
||||
let _ = self.import_block(BlockOrigin::ConsensusBroadcast, justified_header, Some(extrinsics));
|
||||
@@ -533,6 +546,7 @@ mod tests {
|
||||
use keyring::Keyring;
|
||||
use test_client::{self, TestClient};
|
||||
use test_client::client::BlockOrigin;
|
||||
use test_client::client::backend::Backend as TestBackend;
|
||||
use test_client::runtime as test_runtime;
|
||||
use test_client::runtime::{Transfer, Extrinsic};
|
||||
|
||||
@@ -593,6 +607,18 @@ mod tests {
|
||||
assert_eq!(client.using_environment(|| test_runtime::system::balance_of(Keyring::Ferdie.to_raw_public().into())).unwrap(), 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_uses_authorities_from_blockchain_cache() {
|
||||
let client = test_client::new();
|
||||
test_client::client::in_mem::cache_authorities_at(
|
||||
client.backend().blockchain(),
|
||||
Default::default(),
|
||||
Some(vec![[1u8; 32].into()]));
|
||||
assert_eq!(client.authorities_at(
|
||||
&BlockId::Hash(Default::default())).unwrap(),
|
||||
vec![[1u8; 32].into()]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_builder_does_not_include_invalid() {
|
||||
let client = test_client::new();
|
||||
|
||||
@@ -22,6 +22,7 @@ use parking_lot::RwLock;
|
||||
use error;
|
||||
use backend;
|
||||
use light;
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::generic::BlockId;
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Zero, NumberFor, As};
|
||||
use runtime_primitives::bft::Justification;
|
||||
@@ -88,9 +89,27 @@ struct BlockchainStorage<Block: BlockT> {
|
||||
}
|
||||
|
||||
/// In-memory blockchain. Supports concurrent reads.
|
||||
#[derive(Clone)]
|
||||
pub struct Blockchain<Block: BlockT> {
|
||||
storage: Arc<RwLock<BlockchainStorage<Block>>>,
|
||||
cache: Cache<Block>,
|
||||
}
|
||||
|
||||
struct Cache<Block: BlockT> {
|
||||
storage: Arc<RwLock<BlockchainStorage<Block>>>,
|
||||
authorities_at: RwLock<HashMap<Block::Hash, Option<Vec<AuthorityId>>>>,
|
||||
}
|
||||
|
||||
impl<Block: BlockT + Clone> Clone for Blockchain<Block> {
|
||||
fn clone(&self) -> Self {
|
||||
let storage = Arc::new(RwLock::new(self.storage.read().clone()));
|
||||
Blockchain {
|
||||
storage: storage.clone(),
|
||||
cache: Cache {
|
||||
storage,
|
||||
authorities_at: RwLock::new(self.cache.authorities_at.read().clone()),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Block: BlockT> Blockchain<Block> {
|
||||
@@ -113,7 +132,11 @@ impl<Block: BlockT> Blockchain<Block> {
|
||||
genesis_hash: Default::default(),
|
||||
}));
|
||||
Blockchain {
|
||||
storage: storage,
|
||||
storage: storage.clone(),
|
||||
cache: Cache {
|
||||
storage: storage,
|
||||
authorities_at: Default::default(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,19 +220,37 @@ impl<Block: BlockT> blockchain::Backend<Block> for Blockchain<Block> {
|
||||
b.justification().map(|x| x.clone()))
|
||||
))
|
||||
}
|
||||
|
||||
fn cache(&self) -> Option<&blockchain::Cache<Block>> {
|
||||
Some(&self.cache)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Block: BlockT> light::blockchain::Storage<Block> for Blockchain<Block> {
|
||||
fn import_header(&self, is_new_best: bool, header: Block::Header) -> error::Result<()> {
|
||||
fn import_header(
|
||||
&self,
|
||||
is_new_best: bool,
|
||||
header: Block::Header,
|
||||
authorities: Option<Vec<AuthorityId>>
|
||||
) -> error::Result<()> {
|
||||
let hash = header.hash();
|
||||
let parent_hash = *header.parent_hash();
|
||||
self.insert(hash, header, None, None, is_new_best);
|
||||
if is_new_best {
|
||||
self.cache.insert(parent_hash, authorities);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn cache(&self) -> Option<&blockchain::Cache<Block>> {
|
||||
Some(&self.cache)
|
||||
}
|
||||
}
|
||||
|
||||
/// In-memory operation.
|
||||
pub struct BlockImportOperation<Block: BlockT> {
|
||||
pending_block: Option<PendingBlock<Block>>,
|
||||
pending_authorities: Option<Vec<AuthorityId>>,
|
||||
old_state: InMemory,
|
||||
new_state: Option<InMemory>,
|
||||
}
|
||||
@@ -236,6 +277,10 @@ impl<Block: BlockT> backend::BlockImportOperation<Block> for BlockImportOperatio
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn update_authorities(&mut self, authorities: Vec<AuthorityId>) {
|
||||
self.pending_authorities = Some(authorities);
|
||||
}
|
||||
|
||||
fn update_storage(&mut self, update: <InMemory as StateBackend>::Transaction) -> error::Result<()> {
|
||||
self.new_state = Some(self.old_state.update(update));
|
||||
Ok(())
|
||||
@@ -282,6 +327,7 @@ impl<Block> backend::Backend<Block> for Backend<Block> where
|
||||
|
||||
Ok(BlockImportOperation {
|
||||
pending_block: None,
|
||||
pending_authorities: None,
|
||||
old_state: state,
|
||||
new_state: None,
|
||||
})
|
||||
@@ -292,9 +338,14 @@ impl<Block> backend::Backend<Block> for Backend<Block> where
|
||||
let old_state = &operation.old_state;
|
||||
let (header, body, justification) = pending_block.block.into_inner();
|
||||
let hash = header.hash();
|
||||
let parent_hash = *header.parent_hash();
|
||||
|
||||
self.states.write().insert(hash, operation.new_state.unwrap_or_else(|| old_state.clone()));
|
||||
self.blockchain.insert(hash, header, justification, body, pending_block.is_best);
|
||||
// dumb implementation - store value for each block
|
||||
if pending_block.is_best {
|
||||
self.blockchain.cache.insert(parent_hash, operation.pending_authorities);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -316,3 +367,29 @@ impl<Block> backend::Backend<Block> for Backend<Block> where
|
||||
}
|
||||
|
||||
impl<Block: BlockT> backend::LocalBackend<Block> for Backend<Block> {}
|
||||
|
||||
impl<Block: BlockT> Cache<Block> {
|
||||
fn insert(&self, at: Block::Hash, authorities: Option<Vec<AuthorityId>>) {
|
||||
self.authorities_at.write().insert(at, authorities);
|
||||
}
|
||||
}
|
||||
|
||||
impl<Block: BlockT> blockchain::Cache<Block> for Cache<Block> {
|
||||
fn authorities_at(&self, block: BlockId<Block>) -> Option<Vec<AuthorityId>> {
|
||||
let hash = match block {
|
||||
BlockId::Hash(hash) => hash,
|
||||
BlockId::Number(number) => self.storage.read().hashes.get(&number).cloned()?,
|
||||
};
|
||||
|
||||
self.authorities_at.read().get(&hash).cloned().unwrap_or(None)
|
||||
}
|
||||
}
|
||||
|
||||
/// Insert authorities entry into in-memory blockchain cache. Extracted as a separate function to use it in tests.
|
||||
pub fn cache_authorities_at<Block: BlockT>(
|
||||
blockchain: &Blockchain<Block>,
|
||||
at: Block::Hash,
|
||||
authorities: Option<Vec<AuthorityId>>
|
||||
) {
|
||||
blockchain.cache.insert(at, authorities);
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
|
||||
use std::sync::{Arc, Weak};
|
||||
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::{bft::Justification, generic::BlockId};
|
||||
use runtime_primitives::traits::{Block as BlockT, NumberFor};
|
||||
use state_machine::{Backend as StateBackend, TrieBackend as StateTrieBackend,
|
||||
@@ -39,6 +40,7 @@ pub struct Backend<S, F> {
|
||||
pub struct ImportOperation<Block: BlockT, F> {
|
||||
is_new_best: bool,
|
||||
header: Option<Block::Header>,
|
||||
authorities: Option<Vec<AuthorityId>>,
|
||||
_phantom: ::std::marker::PhantomData<F>,
|
||||
}
|
||||
|
||||
@@ -69,13 +71,14 @@ impl<S, F, Block> ClientBackend<Block> for Backend<S, F> where Block: BlockT, S:
|
||||
Ok(ImportOperation {
|
||||
is_new_best: false,
|
||||
header: None,
|
||||
authorities: None,
|
||||
_phantom: Default::default(),
|
||||
})
|
||||
}
|
||||
|
||||
fn commit_operation(&self, operation: Self::BlockImportOperation) -> ClientResult<()> {
|
||||
let header = operation.header.expect("commit is called after set_block_data; set_block_data sets header; qed");
|
||||
self.blockchain.storage().import_header(operation.is_new_best, header)
|
||||
self.blockchain.storage().import_header(operation.is_new_best, header, operation.authorities)
|
||||
}
|
||||
|
||||
fn blockchain(&self) -> &Blockchain<S, F> {
|
||||
@@ -121,6 +124,10 @@ impl<F, Block> BlockImportOperation<Block> for ImportOperation<Block, F> where B
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn update_authorities(&mut self, authorities: Vec<AuthorityId>) {
|
||||
self.authorities = Some(authorities);
|
||||
}
|
||||
|
||||
fn update_storage(&mut self, _update: <Self::State as StateBackend>::Transaction) -> ClientResult<()> {
|
||||
// we're not storing anything locally => ignore changes
|
||||
Ok(())
|
||||
|
||||
@@ -20,10 +20,11 @@
|
||||
use std::sync::Weak;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use primitives::AuthorityId;
|
||||
use runtime_primitives::{bft::Justification, generic::BlockId};
|
||||
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
|
||||
|
||||
use blockchain::{Backend as BlockchainBackend, BlockStatus,
|
||||
use blockchain::{Backend as BlockchainBackend, BlockStatus, Cache as BlockchainCache,
|
||||
HeaderBackend as BlockchainHeaderBackend, Info as BlockchainInfo};
|
||||
use error::Result as ClientResult;
|
||||
use light::fetcher::Fetcher;
|
||||
@@ -31,7 +32,15 @@ use light::fetcher::Fetcher;
|
||||
/// Light client blockchain storage.
|
||||
pub trait Storage<Block: BlockT>: BlockchainHeaderBackend<Block> {
|
||||
/// Store new header.
|
||||
fn import_header(&self, is_new_best: bool, header: Block::Header) -> ClientResult<()>;
|
||||
fn import_header(
|
||||
&self,
|
||||
is_new_best: bool,
|
||||
header: Block::Header,
|
||||
authorities: Option<Vec<AuthorityId>>
|
||||
) -> ClientResult<()>;
|
||||
|
||||
/// Get storage cache.
|
||||
fn cache(&self) -> Option<&BlockchainCache<Block>>;
|
||||
}
|
||||
|
||||
/// Light client blockchain.
|
||||
@@ -92,4 +101,8 @@ impl<S, F, Block> BlockchainBackend<Block> for Blockchain<S, F> where Block: Blo
|
||||
fn justification(&self, _id: BlockId<Block>) -> ClientResult<Option<Justification<Block::Hash>>> {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
fn cache(&self) -> Option<&BlockchainCache<Block>> {
|
||||
self.storage.cache()
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user