Unverified Commit 70a4469d authored by asynchronous rob's avatar asynchronous rob Committed by GitHub
Browse files

Chain Selection Subsystem Logic (#3277)



* crate skeleton and type definitions

* add ChainSelectionMessage

* add error type

* run loop

* fix overseer

* simplify determine_new_blocks API

* write an overlay struct and fetch new blocks

* add new function to overlay

* more flow

* add leaves to overlay and add a strong type around leaves-set

* add is_parent_viable

* implement block import, ignoring reversions

* add stagnant-at to overlay

* add stagnant

* add revert consensus log

* flow for reversions

* extract and import block reversions

* recursively update viability

* remove redundant parameter from WriteBlockEntry

* do some removal of viable leaves

* address grumbles

* refactor

* address grumbles

* add comment about non-monotonicity

* extract backend to submodule

* begin the hunt for viable leaves

* viability pivots for updating the active leaves

* remove LeafSearchFrontier

* partially -> explicitly viable and untwist some booleans

* extract tree to submodule

* implement block finality update

* Implement block approval routine

* implement stagnant detection

* ensure blocks pruned on finality are removed from the active leaves set

* write down some planned test cases

* floww

* leaf loading

* implement best_leaf_containing

* write down a few more tests to do

* remove dependence of tree on header

* guide: ChainApiMessage::BlockWeight

* node: BlockWeight ChainAPI

* fix compile issue

* note a few TODOs for the future

* fetch block weight using new BlockWeight ChainAPI

* implement unimplemented

* sort leaves by block number after weight

* remove warnings and add more TODOs

* create test module

* storage for test backend

* wrap inner in mutex

* add write waker query to test backend

* Add OverseerSignal -> FromOverseer conversion

* add test harnes

* add no-op test

* add some more test helpers

* the first test

* more progress on tests

* test two subtrees

* determine-new-blocks: cleaner genesis avoidance and tighter ancestry requests

* don't make ancestry requests when asking for one block

* add a couple more tests

* add to AllMessages in guide

* remove bad spaces from bridge

* compact iterator

* test import with gaps

* more reversion tests

* test finalization pruning subtrees

* fixups

* test clobbering and fix bug in overlay

* exhaustive backend state after finalizaiton tested

* more finality tests

* leaf tests

* test approval

* test ChainSelectionMessage::Leaves thoroughly

* remove TODO

* avoid Ordering::is_ne so CI can build

* comment algorithmic complexity

* Update node/core/chain-selection/src/lib.rs
Co-authored-by: default avatarBernhard Schuster <bernhard@ahoi.io>
Co-authored-by: default avatarBernhard Schuster <bernhard@ahoi.io>
parent 897afffb
Pipeline #143544 passed with stages
in 39 minutes and 13 seconds
......@@ -6109,6 +6109,25 @@ dependencies = [
"tracing",
]
[[package]]
name = "polkadot-node-core-chain-selection"
version = "0.1.0"
dependencies = [
"assert_matches",
"futures 0.3.15",
"kvdb",
"parity-scale-codec",
"parking_lot 0.11.1",
"polkadot-node-primitives",
"polkadot-node-subsystem",
"polkadot-node-subsystem-test-helpers",
"polkadot-node-subsystem-util",
"polkadot-primitives",
"sp-core",
"thiserror",
"tracing",
]
[[package]]
name = "polkadot-node-core-dispute-coordinator"
version = "0.1.0"
......
......@@ -48,6 +48,7 @@ members = [
"node/core/bitfield-signing",
"node/core/candidate-validation",
"node/core/chain-api",
"node/core/chain-selection",
"node/core/dispute-coordinator",
"node/core/dispute-participation",
"node/core/parachains-inherent",
......
[package]
name = "polkadot-node-core-chain-selection"
description = "Chain Selection Subsystem"
version = "0.1.0"
authors = ["Parity Technologies <admin@parity.io>"]
edition = "2018"
[dependencies]
futures = "0.3.15"
tracing = "0.1.26"
polkadot-primitives = { path = "../../../primitives" }
polkadot-node-primitives = { path = "../../primitives" }
polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
polkadot-node-subsystem-util = { path = "../../subsystem-util" }
kvdb = "0.9.0"
thiserror = "1.0.23"
parity-scale-codec = "2"
[dev-dependencies]
polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }
sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
parking_lot = "0.11"
assert_matches = "1"
// Copyright 2021 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/>.
//! An abstraction over storage used by the chain selection subsystem.
//!
//! This provides both a [`Backend`] trait and an [`OverlayedBackend`]
//! struct which allows in-memory changes to be applied on top of a
//! [`Backend`], maintaining consistency between queries and temporary writes,
//! before any commit to the underlying storage is made.
use polkadot_primitives::v1::{BlockNumber, Hash};
use std::collections::HashMap;
use crate::{Error, LeafEntrySet, BlockEntry, Timestamp};
pub(super) enum BackendWriteOp {
WriteBlockEntry(BlockEntry),
WriteBlocksByNumber(BlockNumber, Vec<Hash>),
WriteViableLeaves(LeafEntrySet),
WriteStagnantAt(Timestamp, Vec<Hash>),
DeleteBlocksByNumber(BlockNumber),
DeleteBlockEntry(Hash),
DeleteStagnantAt(Timestamp),
}
/// An abstraction over backend storage for the logic of this subsystem.
pub(super) trait Backend {
/// Load a block entry from the DB.
fn load_block_entry(&self, hash: &Hash) -> Result<Option<BlockEntry>, Error>;
/// Load the active-leaves set.
fn load_leaves(&self) -> Result<LeafEntrySet, Error>;
/// Load the stagnant list at the given timestamp.
fn load_stagnant_at(&self, timestamp: Timestamp) -> Result<Vec<Hash>, Error>;
/// Load all stagnant lists up to and including the given unix timestamp
/// in ascending order.
fn load_stagnant_at_up_to(&self, up_to: Timestamp)
-> Result<Vec<(Timestamp, Vec<Hash>)>, Error>;
/// Load the earliest kept block number.
fn load_first_block_number(&self) -> Result<Option<BlockNumber>, Error>;
/// Load blocks by number.
fn load_blocks_by_number(&self, number: BlockNumber) -> Result<Vec<Hash>, Error>;
/// Atomically write the list of operations, with later operations taking precedence over prior.
fn write<I>(&mut self, ops: I) -> Result<(), Error>
where I: IntoIterator<Item = BackendWriteOp>;
}
/// An in-memory overlay over the backend.
///
/// This maintains read-only access to the underlying backend, but can be
/// converted into a set of write operations which will, when written to
/// the underlying backend, give the same view as the state of the overlay.
pub(super) struct OverlayedBackend<'a, B: 'a> {
inner: &'a B,
// `None` means 'deleted', missing means query inner.
block_entries: HashMap<Hash, Option<BlockEntry>>,
// `None` means 'deleted', missing means query inner.
blocks_by_number: HashMap<BlockNumber, Option<Vec<Hash>>>,
// 'None' means 'deleted', missing means query inner.
stagnant_at: HashMap<Timestamp, Option<Vec<Hash>>>,
// 'None' means query inner.
leaves: Option<LeafEntrySet>,
}
impl<'a, B: 'a + Backend> OverlayedBackend<'a, B> {
pub(super) fn new(backend: &'a B) -> Self {
OverlayedBackend {
inner: backend,
block_entries: HashMap::new(),
blocks_by_number: HashMap::new(),
stagnant_at: HashMap::new(),
leaves: None,
}
}
pub(super) fn load_block_entry(&self, hash: &Hash) -> Result<Option<BlockEntry>, Error> {
if let Some(val) = self.block_entries.get(&hash) {
return Ok(val.clone())
}
self.inner.load_block_entry(hash)
}
pub(super) fn load_blocks_by_number(&self, number: BlockNumber) -> Result<Vec<Hash>, Error> {
if let Some(val) = self.blocks_by_number.get(&number) {
return Ok(val.as_ref().map_or(Vec::new(), Clone::clone));
}
self.inner.load_blocks_by_number(number)
}
pub(super) fn load_leaves(&self) -> Result<LeafEntrySet, Error> {
if let Some(ref set) = self.leaves {
return Ok(set.clone())
}
self.inner.load_leaves()
}
pub(super) fn load_stagnant_at(&self, timestamp: Timestamp) -> Result<Vec<Hash>, Error> {
if let Some(val) = self.stagnant_at.get(&timestamp) {
return Ok(val.as_ref().map_or(Vec::new(), Clone::clone));
}
self.inner.load_stagnant_at(timestamp)
}
pub(super) fn write_block_entry(&mut self, entry: BlockEntry) {
self.block_entries.insert(entry.block_hash, Some(entry));
}
pub(super) fn delete_block_entry(&mut self, hash: &Hash) {
self.block_entries.insert(*hash, None);
}
pub(super) fn write_blocks_by_number(&mut self, number: BlockNumber, blocks: Vec<Hash>) {
if blocks.is_empty() {
self.blocks_by_number.insert(number, None);
} else {
self.blocks_by_number.insert(number, Some(blocks));
}
}
pub(super) fn delete_blocks_by_number(&mut self, number: BlockNumber) {
self.blocks_by_number.insert(number, None);
}
pub(super) fn write_leaves(&mut self, leaves: LeafEntrySet) {
self.leaves = Some(leaves);
}
pub(super) fn write_stagnant_at(&mut self, timestamp: Timestamp, hashes: Vec<Hash>) {
self.stagnant_at.insert(timestamp, Some(hashes));
}
pub(super) fn delete_stagnant_at(&mut self, timestamp: Timestamp) {
self.stagnant_at.insert(timestamp, None);
}
/// Transform this backend into a set of write-ops to be written to the
/// inner backend.
pub(super) fn into_write_ops(self) -> impl Iterator<Item = BackendWriteOp> {
let block_entry_ops = self.block_entries.into_iter().map(|(h, v)| match v {
Some(v) => BackendWriteOp::WriteBlockEntry(v),
None => BackendWriteOp::DeleteBlockEntry(h),
});
let blocks_by_number_ops = self.blocks_by_number.into_iter().map(|(n, v)| match v {
Some(v) => BackendWriteOp::WriteBlocksByNumber(n, v),
None => BackendWriteOp::DeleteBlocksByNumber(n),
});
let leaf_ops = self.leaves.into_iter().map(BackendWriteOp::WriteViableLeaves);
let stagnant_at_ops = self.stagnant_at.into_iter().map(|(n, v)| match v {
Some(v) => BackendWriteOp::WriteStagnantAt(n, v),
None => BackendWriteOp::DeleteStagnantAt(n),
});
block_entry_ops
.chain(blocks_by_number_ops)
.chain(leaf_ops)
.chain(stagnant_at_ops)
}
}
/// Attempt to find the given ancestor in the chain with given head.
///
/// If the ancestor is the most recently finalized block, and the `head` is
/// a known unfinalized block, this will return `true`.
///
/// If the ancestor is an unfinalized block and `head` is known, this will
/// return true if `ancestor` is in `head`'s chain.
///
/// If the ancestor is an older finalized block, this will return `false`.
fn contains_ancestor(
backend: &impl Backend,
head: Hash,
ancestor: Hash,
) -> Result<bool, Error> {
let mut current_hash = head;
loop {
if current_hash == ancestor { return Ok(true) }
match backend.load_block_entry(&current_hash)? {
Some(e) => { current_hash = e.parent_hash }
None => break
}
}
Ok(false)
}
/// This returns the best unfinalized leaf containing the required block.
///
/// If the required block is finalized but not the most recent finalized block,
/// this will return `None`.
///
/// If the required block is unfinalized but not an ancestor of any viable leaf,
/// this will return `None`.
//
// Note: this is O(N^2) in the depth of `required` and the number of leaves.
// We expect the number of unfinalized blocks to be small, as in, to not exceed
// single digits in practice, and exceedingly unlikely to surpass 1000.
//
// However, if we need to, we could implement some type of skip-list for
// fast ancestry checks.
pub(super) fn find_best_leaf_containing(
backend: &impl Backend,
required: Hash,
) -> Result<Option<Hash>, Error> {
let leaves = backend.load_leaves()?;
for leaf in leaves.into_hashes_descending() {
if contains_ancestor(backend, leaf, required)? {
return Ok(Some(leaf))
}
}
// If there are no viable leaves containing the ancestor
Ok(None)
}
// Copyright 2021 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/>.
//! Implements the Chain Selection Subsystem.
use polkadot_primitives::v1::{BlockNumber, Hash, Header, ConsensusLog};
use polkadot_node_primitives::BlockWeight;
use polkadot_subsystem::{
Subsystem, SubsystemContext, SubsystemError, SpawnedSubsystem,
OverseerSignal, FromOverseer,
messages::{ChainSelectionMessage, ChainApiMessage},
errors::ChainApiError,
};
use parity_scale_codec::Error as CodecError;
use futures::channel::oneshot;
use futures::prelude::*;
use std::time::{UNIX_EPOCH, SystemTime};
use crate::backend::{Backend, OverlayedBackend, BackendWriteOp};
mod backend;
mod tree;
#[cfg(test)]
mod tests;
const LOG_TARGET: &str = "parachain::chain-selection";
/// Timestamp based on the 1 Jan 1970 UNIX base, which is persistent across node restarts and OS reboots.
type Timestamp = u64;
#[derive(Debug, Clone)]
enum Approval {
// Approved
Approved,
// Unapproved but not stagnant
Unapproved,
// Unapproved and stagnant.
Stagnant,
}
impl Approval {
fn is_stagnant(&self) -> bool {
matches!(*self, Approval::Stagnant)
}
}
#[derive(Debug, Clone)]
struct ViabilityCriteria {
// Whether this block has been explicitly reverted by one of its descendants.
explicitly_reverted: bool,
// The approval state of this block specifically.
approval: Approval,
// The earliest unviable ancestor - the hash of the earliest unfinalized
// block in the ancestry which is explicitly reverted or stagnant.
earliest_unviable_ancestor: Option<Hash>,
}
impl ViabilityCriteria {
fn is_viable(&self) -> bool {
self.is_parent_viable() && self.is_explicitly_viable()
}
// Whether the current block is explicitly viable.
// That is, whether the current block is neither reverted nor stagnant.
fn is_explicitly_viable(&self) -> bool {
!self.explicitly_reverted && !self.approval.is_stagnant()
}
// Whether the parent is viable. This assumes that the parent
// descends from the finalized chain.
fn is_parent_viable(&self) -> bool {
self.earliest_unviable_ancestor.is_none()
}
}
// Light entries describing leaves of the chain.
//
// These are ordered first by weight and then by block number.
#[derive(Debug, Clone, PartialEq)]
struct LeafEntry {
weight: BlockWeight,
block_number: BlockNumber,
block_hash: Hash,
}
impl PartialOrd for LeafEntry {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
let ord = self.weight.cmp(&other.weight)
.then(self.block_number.cmp(&other.block_number));
if !matches!(ord, std::cmp::Ordering::Equal) { Some(ord) } else { None }
}
}
#[derive(Debug, Default, Clone)]
struct LeafEntrySet {
inner: Vec<LeafEntry>
}
impl LeafEntrySet {
fn remove(&mut self, hash: &Hash) -> bool {
match self.inner.iter().position(|e| &e.block_hash == hash) {
None => false,
Some(i) => {
self.inner.remove(i);
true
}
}
}
fn insert(&mut self, new: LeafEntry) {
let mut pos = None;
for (i, e) in self.inner.iter().enumerate() {
if e == &new { return }
if e < &new {
pos = Some(i);
break
}
}
match pos {
None => self.inner.push(new),
Some(i) => self.inner.insert(i, new),
}
}
fn into_hashes_descending(self) -> impl Iterator<Item = Hash> {
self.inner.into_iter().map(|e| e.block_hash)
}
}
#[derive(Debug, Clone)]
struct BlockEntry {
block_hash: Hash,
block_number: BlockNumber,
parent_hash: Hash,
children: Vec<Hash>,
viability: ViabilityCriteria,
weight: BlockWeight,
}
impl BlockEntry {
fn leaf_entry(&self) -> LeafEntry {
LeafEntry {
block_hash: self.block_hash,
block_number: self.block_number,
weight: self.weight,
}
}
fn non_viable_ancestor_for_child(&self) -> Option<Hash> {
if self.viability.is_viable() {
None
} else {
self.viability.earliest_unviable_ancestor.or(Some(self.block_hash))
}
}
}
#[derive(Debug, thiserror::Error)]
#[allow(missing_docs)]
pub enum Error {
#[error(transparent)]
ChainApi(#[from] ChainApiError),
#[error(transparent)]
Io(#[from] std::io::Error),
#[error(transparent)]
Oneshot(#[from] oneshot::Canceled),
#[error(transparent)]
Subsystem(#[from] SubsystemError),
#[error(transparent)]
Codec(#[from] CodecError),
}
impl Error {
fn trace(&self) {
match self {
// don't spam the log with spurious errors
Self::Oneshot(_) => tracing::debug!(target: LOG_TARGET, err = ?self),
// it's worth reporting otherwise
_ => tracing::warn!(target: LOG_TARGET, err = ?self),
}
}
}
fn timestamp_now() -> Timestamp {
// `SystemTime` is notoriously non-monotonic, so our timers might not work
// exactly as expected. Regardless, stagnation is detected on the order of minutes,
// and slippage of a few seconds in either direction won't cause any major harm.
//
// The exact time that a block becomes stagnant in the local node is always expected
// to differ from other nodes due to network asynchrony and delays in block propagation.
// Non-monotonicity exarcerbates that somewhat, but not meaningfully.
match SystemTime::now().duration_since(UNIX_EPOCH) {
Ok(d) => d.as_secs(),
Err(e) => {
tracing::warn!(
target: LOG_TARGET,
err = ?e,
"Current time is before unix epoch. Validation will not work correctly."
);
0
}
}
}
fn stagnant_timeout_from_now() -> Timestamp {
// If a block isn't approved in 120 seconds, nodes will abandon it
// and begin building on another chain.
const STAGNANT_TIMEOUT: Timestamp = 120;
timestamp_now() + STAGNANT_TIMEOUT
}
// TODO https://github.com/paritytech/polkadot/issues/3293:
//
// This is used just so we can have a public function that calls
// `run` and eliminates all the unused errors.
//
// Should be removed when the real implementation is done.
struct VoidBackend;
impl Backend for VoidBackend {
fn load_block_entry(&self, _: &Hash) -> Result<Option<BlockEntry>, Error> {
Ok(None)
}
fn load_leaves(&self) -> Result<LeafEntrySet, Error> {
Ok(LeafEntrySet::default())
}
fn load_stagnant_at(&self, _: Timestamp) -> Result<Vec<Hash>, Error> {
Ok(Vec::new())
}
fn load_stagnant_at_up_to(&self, _: Timestamp)
-> Result<Vec<(Timestamp, Vec<Hash>)>, Error>
{
Ok(Vec::new())
}
fn load_first_block_number(&self) -> Result<Option<BlockNumber>, Error> {
Ok(None)
}
fn load_blocks_by_number(&self, _: BlockNumber) -> Result<Vec<Hash>, Error> {
Ok(Vec::new())
}
fn write<I>(&mut self, _: I) -> Result<(), Error>
where I: IntoIterator<Item = BackendWriteOp>
{
Ok(())
}
}
/// The chain selection subsystem.
pub struct ChainSelectionSubsystem;
impl<Context> Subsystem<Context> for ChainSelectionSubsystem
where Context: SubsystemContext<Message = ChainSelectionMessage>
{
fn start(self, ctx: Context) -> SpawnedSubsystem {
let backend = VoidBackend;
SpawnedSubsystem {
future: run(ctx, backend).map(|()| Ok(())).boxed(),
name: "chain-selection-subsystem",
}
}
}
async fn run<Context, B>(mut ctx: Context, mut backend: B)
where
Context: SubsystemContext<Message = ChainSelectionMessage>,
B: Backend,
{
loop {
let res = run_iteration(&mut ctx, &mut backend).await;
match res {
Err(e) => {
e.trace();
if let Error::Subsystem(SubsystemError::Context(_)) = e {
break;
}
}
Ok(()) => {
tracing::info!(target: LOG_TARGET, "received `Conclude` signal, exiting");
break;
}