Prune abandoned forks (#916)

* Address compiler warning

* Prune abandoned fork choice forks

* New approach to pruning

* Wrap some block hashes in a newtype pattern

For increased type safety.

* Add Graphviz chain dump emitter for debugging

* Fix broken test case

* Make prunes_abandoned_forks use real DiskStore

* Mark finalized blocks in the GraphViz output

* Refine debug stringification of Slot and Epoch

Before this commit: print!("{:?}", Slot(123)) == "Slot(\n123\n)".
After this commit: print!("{:?", Slot(123)) == "Slot(123)".

* Simplify build_block()

* Rewrite test case using more composable test primitives

* Working rewritten test case

* Tighten fork prunning test checks

* Add another pruning test case

* Bugfix: Finalized blocks weren't always properly detected

* Pruning: Add pruning_does_not_touch_blocks_prior_to_finalization test case

* Tighten pruning tests: check if heads are tracked properly

* Add a failing test case for a buggy scenario

* Change name of function to a more accurate one

* Fix failing test case

* Test case: Were skipped slots' states pruned?

* Style fix: Simplify dereferencing

* Tighten pruning tests: check if abandoned states are deleted

* Towards atomicity of db ops

* Correct typo

* Prune also skipped slots' states

* New logic for handling skipped states

* Make skipped slots test pass

* Post conflict resolution fixes

* Formatting fixes

* Tests passing

* Block hashes in Graphviz node labels

* Removed unused changes

* Fix bug with states having < SlotsPerHistoricalRoot roots

* Consolidate State/BlockRootsIterator for pruning

* Address review feedback

* Fix a bug in pruning tests

* Detach prune_abandoned_forks() from its object

* Move migrate.rs from store to beacon_chain

* Move forks pruning onto a background thread

* Bugfix: Heads weren't pruned when prune set contained only the head

* Rename: freeze_to_state() -> process_finalization()

* Eliminate redundant function parameter

Co-authored-by: Michael Sproul <michael@sigmaprime.io>

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -4,6 +4,7 @@ use crate::events::{EventHandler, EventKind};
 use crate::fork_choice::{Error as ForkChoiceError, ForkChoice};
 use crate::head_tracker::HeadTracker;
 use crate::metrics;
+use crate::migrate::Migrate;
 use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::shuffling_cache::ShufflingCache;
 use crate::snapshot_cache::SnapshotCache;
@@ -26,14 +27,16 @@ use state_processing::{
 use std::borrow::Cow;
 use std::cmp::Ordering;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::fs;
 use std::io::prelude::*;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use store::iter::{
-    BlockRootsIterator, ReverseBlockRootIterator, ReverseStateRootIterator, StateRootsIterator,
+    BlockRootsIterator, ParentRootBlockIterator, ReverseBlockRootIterator,
+    ReverseStateRootIterator, StateRootsIterator,
 };
-use store::{Error as DBError, Migrate, StateBatch, Store};
+use store::{Error as DBError, StateBatch, Store};
 use tree_hash::TreeHash;
 use types::*;
 
@@ -170,7 +173,7 @@ pub struct HeadInfo {
 
 pub trait BeaconChainTypes: Send + Sync + 'static {
     type Store: store::Store<Self::EthSpec>;
-    type StoreMigrator: store::Migrate<Self::Store, Self::EthSpec>;
+    type StoreMigrator: Migrate<Self::Store, Self::EthSpec>;
     type SlotClock: slot_clock::SlotClock;
     type Eth1Chain: Eth1ChainBackend<Self::EthSpec, Self::Store>;
     type EthSpec: types::EthSpec;
@@ -202,7 +205,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// A handler for events generated by the beacon chain.
     pub event_handler: T::EventHandler,
     /// Used to track the heads of the beacon chain.
-    pub(crate) head_tracker: HeadTracker,
+    pub(crate) head_tracker: Arc<HeadTracker>,
     /// A cache dedicated to block processing.
     pub(crate) block_processing_cache: TimeoutRwLock<SnapshotCache<T::EthSpec>>,
     /// Caches the shuffling for a given epoch and state root.
@@ -498,6 +501,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.head_tracker.heads()
     }
 
+    pub fn knows_head(&self, block_hash: &SignedBeaconBlockHash) -> bool {
+        self.head_tracker.contains_head((*block_hash).into())
+    }
+
     /// Returns the `BeaconState` at the given slot.
     ///
     /// Returns `None` when the state is not found in the database or there is an error skipping
@@ -1796,6 +1803,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let beacon_block_root = self.fork_choice.find_head(&self)?;
 
         let current_head = self.head_info()?;
+        let old_finalized_root = current_head.finalized_checkpoint.root;
 
         if beacon_block_root == current_head.block_root {
             return Ok(());
@@ -1923,7 +1931,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             });
 
         if new_finalized_epoch != old_finalized_epoch {
-            self.after_finalization(old_finalized_epoch, finalized_root)?;
+            self.after_finalization(
+                old_finalized_epoch,
+                finalized_root,
+                old_finalized_root.into(),
+            )?;
         }
 
         let _ = self.event_handler.register(EventKind::BeaconHeadChanged {
@@ -1942,6 +1954,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         old_finalized_epoch: Epoch,
         finalized_block_root: Hash256,
+        old_finalized_root: SignedBeaconBlockHash,
     ) -> Result<(), Error> {
         let finalized_block = self
             .store
@@ -1981,10 +1994,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
             // TODO: configurable max finality distance
             let max_finality_distance = 0;
-            self.store_migrator.freeze_to_state(
+            self.store_migrator.process_finalization(
                 finalized_block.state_root,
                 finalized_state,
                 max_finality_distance,
+                Arc::clone(&self.head_tracker),
+                old_finalized_root,
+                finalized_block_root.into(),
             );
 
             let _ = self.event_handler.register(EventKind::BeaconFinalization {
@@ -2052,6 +2068,100 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         Ok(dump)
     }
+
+    pub fn dump_as_dot<W: Write>(&self, output: &mut W) {
+        let canonical_head_hash = self
+            .canonical_head
+            .try_read_for(HEAD_LOCK_TIMEOUT)
+            .ok_or_else(|| Error::CanonicalHeadLockTimeout)
+            .unwrap()
+            .beacon_block_root;
+        let mut visited: HashSet<Hash256> = HashSet::new();
+        let mut finalized_blocks: HashSet<Hash256> = HashSet::new();
+
+        let genesis_block_hash = Hash256::zero();
+        write!(output, "digraph beacon {{\n").unwrap();
+        write!(output, "\t_{:?}[label=\"genesis\"];\n", genesis_block_hash).unwrap();
+
+        // Canonical head needs to be processed first as otherwise finalized blocks aren't detected
+        // properly.
+        let heads = {
+            let mut heads = self.heads();
+            let canonical_head_index = heads
+                .iter()
+                .position(|(block_hash, _)| *block_hash == canonical_head_hash)
+                .unwrap();
+            let (canonical_head_hash, canonical_head_slot) =
+                heads.swap_remove(canonical_head_index);
+            heads.insert(0, (canonical_head_hash, canonical_head_slot));
+            heads
+        };
+
+        for (head_hash, _head_slot) in heads {
+            for (block_hash, signed_beacon_block) in
+                ParentRootBlockIterator::new(&*self.store, head_hash)
+            {
+                if visited.contains(&block_hash) {
+                    break;
+                }
+                visited.insert(block_hash);
+
+                if signed_beacon_block.slot() % T::EthSpec::slots_per_epoch() == 0 {
+                    let block = self.get_block(&block_hash).unwrap().unwrap();
+                    let state = self
+                        .get_state(&block.state_root(), Some(block.slot()))
+                        .unwrap()
+                        .unwrap();
+                    finalized_blocks.insert(state.finalized_checkpoint.root);
+                }
+
+                if block_hash == canonical_head_hash {
+                    write!(
+                        output,
+                        "\t_{:?}[label=\"{} ({})\" shape=box3d];\n",
+                        block_hash,
+                        block_hash,
+                        signed_beacon_block.slot()
+                    )
+                    .unwrap();
+                } else if finalized_blocks.contains(&block_hash) {
+                    write!(
+                        output,
+                        "\t_{:?}[label=\"{} ({})\" shape=Msquare];\n",
+                        block_hash,
+                        block_hash,
+                        signed_beacon_block.slot()
+                    )
+                    .unwrap();
+                } else {
+                    write!(
+                        output,
+                        "\t_{:?}[label=\"{} ({})\" shape=box];\n",
+                        block_hash,
+                        block_hash,
+                        signed_beacon_block.slot()
+                    )
+                    .unwrap();
+                }
+                write!(
+                    output,
+                    "\t_{:?} -> _{:?};\n",
+                    block_hash,
+                    signed_beacon_block.parent_root()
+                )
+                .unwrap();
+            }
+        }
+
+        write!(output, "}}\n").unwrap();
+    }
+
+    // Used for debugging
+    #[allow(dead_code)]
+    pub fn dump_dot_file(&self, file_name: &str) {
+        let mut file = std::fs::File::create(file_name).unwrap();
+        self.dump_as_dot(&mut file);
+    }
 }
 
 impl<T: BeaconChainTypes> Drop for BeaconChain<T> {

beacon_node/beacon_chain/src/builder.rs

@@ -5,6 +5,7 @@ use crate::eth1_chain::{CachingEth1Backend, SszEth1};
 use crate::events::NullEventHandler;
 use crate::fork_choice::SszForkChoice;
 use crate::head_tracker::HeadTracker;
+use crate::migrate::Migrate;
 use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::shuffling_cache::ShufflingCache;
 use crate::snapshot_cache::{SnapshotCache, DEFAULT_SNAPSHOT_CACHE_SIZE};
@@ -47,7 +48,7 @@ impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
     for Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
@@ -96,7 +97,7 @@ impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
     >
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
@@ -431,7 +432,7 @@ where
             event_handler: self
                 .event_handler
                 .ok_or_else(|| "Cannot build without an event handler".to_string())?,
-            head_tracker: self.head_tracker.unwrap_or_default(),
+            head_tracker: Arc::new(self.head_tracker.unwrap_or_default()),
             block_processing_cache: TimeoutRwLock::new(SnapshotCache::new(
                 DEFAULT_SNAPSHOT_CACHE_SIZE,
                 canonical_head,
@@ -463,7 +464,7 @@ impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
     >
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
@@ -533,7 +534,7 @@ impl<TStore, TStoreMigrator, TSlotClock, TEthSpec, TEventHandler>
     >
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -571,7 +572,7 @@ impl<TStore, TStoreMigrator, TEth1Backend, TEthSpec, TEventHandler>
     >
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
     TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -610,7 +611,7 @@ impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec>
     >
 where
     TStore: Store<TEthSpec> + 'static,
-    TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
+    TStoreMigrator: Migrate<TStore, TEthSpec> + 'static,
     TSlotClock: SlotClock + 'static,
     TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
     TEthSpec: EthSpec + 'static,
@@ -642,12 +643,12 @@ fn genesis_block<T: EthSpec>(
 #[cfg(test)]
 mod test {
     use super::*;
+    use crate::migrate::{MemoryStore, NullMigrator};
     use eth2_hashing::hash;
     use genesis::{generate_deterministic_keypairs, interop_genesis_state};
     use sloggers::{null::NullLoggerBuilder, Build};
     use ssz::Encode;
     use std::time::Duration;
-    use store::{migrate::NullMigrator, MemoryStore};
     use tempfile::tempdir;
     use types::{EthSpec, MinimalEthSpec, Slot};
 

beacon_node/beacon_chain/src/head_tracker.rs

@@ -25,11 +25,22 @@ impl HeadTracker {
     /// the upstream user.
     pub fn register_block<E: EthSpec>(&self, block_root: Hash256, block: &BeaconBlock<E>) {
         let mut map = self.0.write();
-
         map.remove(&block.parent_root);
         map.insert(block_root, block.slot);
     }
 
+    /// Removes abandoned head.
+    pub fn remove_head(&self, block_root: Hash256) {
+        let mut map = self.0.write();
+        debug_assert!(map.contains_key(&block_root));
+        map.remove(&block_root);
+    }
+
+    /// Returns true iff `block_root` is a recognized head.
+    pub fn contains_head(&self, block_root: Hash256) -> bool {
+        self.0.read().contains_key(&block_root)
+    }
+
     /// Returns the list of heads in the chain.
     pub fn heads(&self) -> Vec<(Hash256, Slot)> {
         self.0

beacon_node/beacon_chain/src/lib.rs

@@ -11,6 +11,7 @@ pub mod events;
 mod fork_choice;
 mod head_tracker;
 mod metrics;
+pub mod migrate;
 mod persisted_beacon_chain;
 mod shuffling_cache;
 mod snapshot_cache;

beacon_node/beacon_chain/src/migrate.rs

@@ -0,0 +1,349 @@
+use crate::errors::BeaconChainError;
+use crate::head_tracker::HeadTracker;
+use parking_lot::Mutex;
+use slog::{debug, warn, Logger};
+use std::collections::{HashMap, HashSet};
+use std::iter::FromIterator;
+use std::mem;
+use std::sync::mpsc;
+use std::sync::Arc;
+use std::thread;
+use store::iter::{ParentRootBlockIterator, RootsIterator};
+use store::{hot_cold_store::HotColdDBError, Error, SimpleDiskStore, Store};
+pub use store::{DiskStore, MemoryStore};
+use types::*;
+use types::{BeaconState, EthSpec, Hash256, Slot};
+
+/// Trait for migration processes that update the database upon finalization.
+pub trait Migrate<S: Store<E>, E: EthSpec>: Send + Sync + 'static {
+    fn new(db: Arc<S>, log: Logger) -> Self;
+
+    fn process_finalization(
+        &self,
+        _state_root: Hash256,
+        _new_finalized_state: BeaconState<E>,
+        _max_finality_distance: u64,
+        _head_tracker: Arc<HeadTracker>,
+        _old_finalized_block_hash: SignedBeaconBlockHash,
+        _new_finalized_block_hash: SignedBeaconBlockHash,
+    ) {
+    }
+
+    /// Traverses live heads and prunes blocks and states of chains that we know can't be built
+    /// upon because finalization would prohibit it.  This is a optimisation intended to save disk
+    /// space.
+    ///
+    /// Assumptions:
+    ///  * It is called after every finalization.
+    fn prune_abandoned_forks(
+        store: Arc<S>,
+        head_tracker: Arc<HeadTracker>,
+        old_finalized_block_hash: SignedBeaconBlockHash,
+        new_finalized_block_hash: SignedBeaconBlockHash,
+        new_finalized_slot: Slot,
+    ) -> Result<(), BeaconChainError> {
+        let old_finalized_slot = store
+            .get_block(&old_finalized_block_hash.into())?
+            .ok_or_else(|| BeaconChainError::MissingBeaconBlock(old_finalized_block_hash.into()))?
+            .slot();
+
+        // Collect hashes from new_finalized_block back to old_finalized_block (inclusive)
+        let mut found_block = false; // hack for `take_until`
+        let newly_finalized_blocks: HashMap<SignedBeaconBlockHash, Slot> = HashMap::from_iter(
+            ParentRootBlockIterator::new(&*store, new_finalized_block_hash.into())
+                .take_while(|(block_hash, _)| {
+                    if found_block {
+                        false
+                    } else {
+                        found_block |= *block_hash == old_finalized_block_hash.into();
+                        true
+                    }
+                })
+                .map(|(block_hash, block)| (block_hash.into(), block.slot())),
+        );
+
+        // We don't know which blocks are shared among abandoned chains, so we buffer and delete
+        // everything in one fell swoop.
+        let mut abandoned_blocks: HashSet<SignedBeaconBlockHash> = HashSet::new();
+        let mut abandoned_states: HashSet<(Slot, BeaconStateHash)> = HashSet::new();
+        let mut abandoned_heads: HashSet<Hash256> = HashSet::new();
+
+        for (head_hash, head_slot) in head_tracker.heads() {
+            let mut potentially_abandoned_head: Option<Hash256> = Some(head_hash);
+            let mut potentially_abandoned_blocks: Vec<(
+                Slot,
+                Option<SignedBeaconBlockHash>,
+                Option<BeaconStateHash>,
+            )> = Vec::new();
+
+            let head_state_hash = store
+                .get_block(&head_hash)?
+                .ok_or_else(|| BeaconStateError::MissingBeaconBlock(head_hash.into()))?
+                .state_root();
+
+            let iterator = std::iter::once((head_hash, head_state_hash, head_slot))
+                .chain(RootsIterator::from_block(Arc::clone(&store), head_hash)?);
+            for (block_hash, state_hash, slot) in iterator {
+                if slot < old_finalized_slot {
+                    // We must assume here any candidate chains include old_finalized_block_hash,
+                    // i.e. there aren't any forks starting at a block that is a strict ancestor of
+                    // old_finalized_block_hash.
+                    break;
+                }
+                match newly_finalized_blocks.get(&block_hash.into()).copied() {
+                    // Block is not finalized, mark it and its state for deletion
+                    None => {
+                        potentially_abandoned_blocks.push((
+                            slot,
+                            Some(block_hash.into()),
+                            Some(state_hash.into()),
+                        ));
+                    }
+                    Some(finalized_slot) => {
+                        // Block root is finalized, and we have reached the slot it was finalized
+                        // at: we've hit a shared part of the chain.
+                        if finalized_slot == slot {
+                            // The first finalized block of a candidate chain lies after (in terms
+                            // of slots order) the newly finalized block.  It's not a candidate for
+                            // prunning.
+                            if finalized_slot == new_finalized_slot {
+                                potentially_abandoned_blocks.clear();
+                                potentially_abandoned_head.take();
+                            }
+
+                            break;
+                        }
+                        // Block root is finalized, but we're at a skip slot: delete the state only.
+                        else {
+                            potentially_abandoned_blocks.push((
+                                slot,
+                                None,
+                                Some(state_hash.into()),
+                            ));
+                        }
+                    }
+                }
+            }
+
+            abandoned_heads.extend(potentially_abandoned_head.into_iter());
+            if !potentially_abandoned_blocks.is_empty() {
+                abandoned_blocks.extend(
+                    potentially_abandoned_blocks
+                        .iter()
+                        .filter_map(|(_, maybe_block_hash, _)| *maybe_block_hash),
+                );
+                abandoned_states.extend(potentially_abandoned_blocks.iter().filter_map(
+                    |(slot, _, maybe_state_hash)| match maybe_state_hash {
+                        None => None,
+                        Some(state_hash) => Some((*slot, *state_hash)),
+                    },
+                ));
+            }
+        }
+
+        // XXX Should be performed atomically, see
+        // https://github.com/sigp/lighthouse/issues/692
+        for block_hash in abandoned_blocks.into_iter() {
+            store.delete_block(&block_hash.into())?;
+        }
+        for (slot, state_hash) in abandoned_states.into_iter() {
+            store.delete_state(&state_hash.into(), slot)?;
+        }
+        for head_hash in abandoned_heads.into_iter() {
+            head_tracker.remove_head(head_hash);
+        }
+
+        Ok(())
+    }
+}
+
+/// Migrator that does nothing, for stores that don't need migration.
+pub struct NullMigrator;
+
+impl<E: EthSpec> Migrate<SimpleDiskStore<E>, E> for NullMigrator {
+    fn new(_: Arc<SimpleDiskStore<E>>, _: Logger) -> Self {
+        NullMigrator
+    }
+}
+
+impl<E: EthSpec> Migrate<MemoryStore<E>, E> for NullMigrator {
+    fn new(_: Arc<MemoryStore<E>>, _: Logger) -> Self {
+        NullMigrator
+    }
+}
+
+/// Migrator that immediately calls the store's migration function, blocking the current execution.
+///
+/// Mostly useful for tests.
+pub struct BlockingMigrator<S> {
+    db: Arc<S>,
+}
+
+impl<E: EthSpec, S: Store<E>> Migrate<S, E> for BlockingMigrator<S> {
+    fn new(db: Arc<S>, _: Logger) -> Self {
+        BlockingMigrator { db }
+    }
+
+    fn process_finalization(
+        &self,
+        state_root: Hash256,
+        new_finalized_state: BeaconState<E>,
+        _max_finality_distance: u64,
+        head_tracker: Arc<HeadTracker>,
+        old_finalized_block_hash: SignedBeaconBlockHash,
+        new_finalized_block_hash: SignedBeaconBlockHash,
+    ) {
+        if let Err(e) = S::process_finalization(self.db.clone(), state_root, &new_finalized_state) {
+            // This migrator is only used for testing, so we just log to stderr without a logger.
+            eprintln!("Migration error: {:?}", e);
+        }
+
+        if let Err(e) = Self::prune_abandoned_forks(
+            self.db.clone(),
+            head_tracker,
+            old_finalized_block_hash,
+            new_finalized_block_hash,
+            new_finalized_state.slot,
+        ) {
+            eprintln!("Pruning error: {:?}", e);
+        }
+    }
+}
+
+type MpscSender<E> = mpsc::Sender<(
+    Hash256,
+    BeaconState<E>,
+    Arc<HeadTracker>,
+    SignedBeaconBlockHash,
+    SignedBeaconBlockHash,
+    Slot,
+)>;
+
+/// Migrator that runs a background thread to migrate state from the hot to the cold database.
+pub struct BackgroundMigrator<E: EthSpec> {
+    db: Arc<DiskStore<E>>,
+    tx_thread: Mutex<(MpscSender<E>, thread::JoinHandle<()>)>,
+    log: Logger,
+}
+
+impl<E: EthSpec> Migrate<DiskStore<E>, E> for BackgroundMigrator<E> {
+    fn new(db: Arc<DiskStore<E>>, log: Logger) -> Self {
+        let tx_thread = Mutex::new(Self::spawn_thread(db.clone(), log.clone()));
+        Self { db, tx_thread, log }
+    }
+
+    /// Perform the freezing operation on the database,
+    fn process_finalization(
+        &self,
+        finalized_state_root: Hash256,
+        new_finalized_state: BeaconState<E>,
+        max_finality_distance: u64,
+        head_tracker: Arc<HeadTracker>,
+        old_finalized_block_hash: SignedBeaconBlockHash,
+        new_finalized_block_hash: SignedBeaconBlockHash,
+    ) {
+        if !self.needs_migration(new_finalized_state.slot, max_finality_distance) {
+            return;
+        }
+
+        let (ref mut tx, ref mut thread) = *self.tx_thread.lock();
+
+        let new_finalized_slot = new_finalized_state.slot;
+        if let Err(tx_err) = tx.send((
+            finalized_state_root,
+            new_finalized_state,
+            head_tracker,
+            old_finalized_block_hash,
+            new_finalized_block_hash,
+            new_finalized_slot,
+        )) {
+            let (new_tx, new_thread) = Self::spawn_thread(self.db.clone(), self.log.clone());
+
+            drop(mem::replace(tx, new_tx));
+            let old_thread = mem::replace(thread, new_thread);
+
+            // Join the old thread, which will probably have panicked, or may have
+            // halted normally just now as a result of us dropping the old `mpsc::Sender`.
+            if let Err(thread_err) = old_thread.join() {
+                warn!(
+                    self.log,
+                    "Migration thread died, so it was restarted";
+                    "reason" => format!("{:?}", thread_err)
+                );
+            }
+
+            // Retry at most once, we could recurse but that would risk overflowing the stack.
+            let _ = tx.send(tx_err.0);
+        }
+    }
+}
+
+impl<E: EthSpec> BackgroundMigrator<E> {
+    /// Return true if a migration needs to be performed, given a new `finalized_slot`.
+    fn needs_migration(&self, finalized_slot: Slot, max_finality_distance: u64) -> bool {
+        let finality_distance = finalized_slot - self.db.get_split_slot();
+        finality_distance > max_finality_distance
+    }
+
+    /// Spawn a new child thread to run the migration process.
+    ///
+    /// Return a channel handle for sending new finalized states to the thread.
+    fn spawn_thread(
+        db: Arc<DiskStore<E>>,
+        log: Logger,
+    ) -> (
+        mpsc::Sender<(
+            Hash256,
+            BeaconState<E>,
+            Arc<HeadTracker>,
+            SignedBeaconBlockHash,
+            SignedBeaconBlockHash,
+            Slot,
+        )>,
+        thread::JoinHandle<()>,
+    ) {
+        let (tx, rx) = mpsc::channel();
+        let thread = thread::spawn(move || {
+            while let Ok((
+                state_root,
+                state,
+                head_tracker,
+                old_finalized_block_hash,
+                new_finalized_block_hash,
+                new_finalized_slot,
+            )) = rx.recv()
+            {
+                match DiskStore::process_finalization(db.clone(), state_root, &state) {
+                    Ok(()) => {}
+                    Err(Error::HotColdDBError(HotColdDBError::FreezeSlotUnaligned(slot))) => {
+                        debug!(
+                            log,
+                            "Database migration postponed, unaligned finalized block";
+                            "slot" => slot.as_u64()
+                        );
+                    }
+                    Err(e) => {
+                        warn!(
+                            log,
+                            "Database migration failed";
+                            "error" => format!("{:?}", e)
+                        );
+                    }
+                };
+
+                match Self::prune_abandoned_forks(
+                    db.clone(),
+                    head_tracker,
+                    old_finalized_block_hash,
+                    new_finalized_block_hash,
+                    new_finalized_slot,
+                ) {
+                    Ok(()) => {}
+                    Err(e) => warn!(log, "Block pruning failed: {:?}", e),
+                }
+            }
+        });
+
+        (tx, thread)
+    }
+}

beacon_node/beacon_chain/src/test_utils.rs

@@ -1,6 +1,7 @@
 pub use crate::beacon_chain::{
     BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY,
 };
+use crate::migrate::{BlockingMigrator, Migrate, NullMigrator};
 pub use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::{
     builder::{BeaconChainBuilder, Witness},
@@ -15,16 +16,16 @@ use sloggers::{null::NullLoggerBuilder, Build};
 use slot_clock::TestingSlotClock;
 use state_processing::per_slot_processing;
 use std::borrow::Cow;
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
-use store::{
-    migrate::{BlockingMigrator, NullMigrator},
-    DiskStore, MemoryStore, Migrate, Store,
-};
+use store::{DiskStore, MemoryStore, Store};
 use tempfile::{tempdir, TempDir};
+use tree_hash::TreeHash;
 use types::{
-    AggregateSignature, Attestation, BeaconState, ChainSpec, Domain, EthSpec, Hash256, Keypair,
-    SecretKey, Signature, SignedBeaconBlock, SignedRoot, Slot,
+    AggregateSignature, Attestation, BeaconState, BeaconStateHash, ChainSpec, Domain, EthSpec,
+    Hash256, Keypair, SecretKey, Signature, SignedBeaconBlock, SignedBeaconBlockHash, SignedRoot,
+    Slot,
 };
 
 pub use types::test_utils::generate_deterministic_keypairs;
@@ -136,7 +137,10 @@ impl<E: EthSpec> BeaconChainHarness<DiskHarnessType<E>> {
             .logger(log.clone())
             .custom_spec(spec.clone())
             .store(store.clone())
-            .store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(store))
+            .store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(
+                store,
+                log.clone(),
+            ))
             .data_dir(data_dir.path().to_path_buf())
             .genesis_state(
                 interop_genesis_state::<E>(&keypairs, HARNESS_GENESIS_TIME, &spec)
@@ -176,7 +180,10 @@ impl<E: EthSpec> BeaconChainHarness<DiskHarnessType<E>> {
             .logger(log.clone())
             .custom_spec(spec)
             .store(store.clone())
-            .store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(store))
+            .store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(
+                store,
+                log.clone(),
+            ))
             .data_dir(data_dir.path().to_path_buf())
             .resume_from_db()
             .expect("should resume beacon chain from db")
@@ -278,6 +285,127 @@ where
         head_block_root.expect("did not produce any blocks")
     }
 
+    /// Returns current canonical head slot
+    pub fn get_chain_slot(&self) -> Slot {
+        self.chain.slot().unwrap()
+    }
+
+    /// Returns current canonical head state
+    pub fn get_head_state(&self) -> BeaconState<E> {
+        self.chain.head().unwrap().beacon_state
+    }
+
+    /// Adds a single block (synchronously) onto either the canonical chain (block_strategy ==
+    /// OnCanonicalHead) or a fork (block_strategy == ForkCanonicalChainAt).
+    pub fn add_block(
+        &self,
+        state: &BeaconState<E>,
+        block_strategy: BlockStrategy,
+        slot: Slot,
+        validators: &[usize],
+    ) -> (SignedBeaconBlockHash, BeaconState<E>) {
+        while self.chain.slot().expect("should have a slot") < slot {
+            self.advance_slot();
+        }
+
+        let (block, new_state) = self.build_block(state.clone(), slot, block_strategy);
+
+        let outcome = self
+            .chain
+            .process_block(block)
+            .expect("should not error during block processing");
+
+        self.chain.fork_choice().expect("should find head");
+
+        if let BlockProcessingOutcome::Processed { block_root } = outcome {
+            let attestation_strategy = AttestationStrategy::SomeValidators(validators.to_vec());
+            self.add_free_attestations(&attestation_strategy, &new_state, block_root, slot);
+            (block_root.into(), new_state)
+        } else {
+            panic!("block should be successfully processed: {:?}", outcome);
+        }
+    }
+
+    /// `add_block()` repeated `num_blocks` times.
+    pub fn add_blocks(
+        &self,
+        mut state: BeaconState<E>,
+        mut slot: Slot,
+        num_blocks: usize,
+        attesting_validators: &[usize],
+        block_strategy: BlockStrategy,
+    ) -> (
+        HashMap<Slot, SignedBeaconBlockHash>,
+        HashMap<Slot, BeaconStateHash>,
+        Slot,
+        SignedBeaconBlockHash,
+        BeaconState<E>,
+    ) {
+        let mut blocks: HashMap<Slot, SignedBeaconBlockHash> = HashMap::with_capacity(num_blocks);
+        let mut states: HashMap<Slot, BeaconStateHash> = HashMap::with_capacity(num_blocks);
+        for _ in 0..num_blocks {
+            let (new_root_hash, new_state) =
+                self.add_block(&state, block_strategy, slot, attesting_validators);
+            blocks.insert(slot, new_root_hash);
+            states.insert(slot, new_state.tree_hash_root().into());
+            state = new_state;
+            slot += 1;
+        }
+        let head_hash = blocks[&(slot - 1)];
+        (blocks, states, slot, head_hash, state)
+    }
+
+    /// A wrapper on `add_blocks()` to avoid passing enums explicitly.
+    pub fn add_canonical_chain_blocks(
+        &self,
+        state: BeaconState<E>,
+        slot: Slot,
+        num_blocks: usize,
+        attesting_validators: &[usize],
+    ) -> (
+        HashMap<Slot, SignedBeaconBlockHash>,
+        HashMap<Slot, BeaconStateHash>,
+        Slot,
+        SignedBeaconBlockHash,
+        BeaconState<E>,
+    ) {
+        let block_strategy = BlockStrategy::OnCanonicalHead;
+        self.add_blocks(
+            state,
+            slot,
+            num_blocks,
+            attesting_validators,
+            block_strategy,
+        )
+    }
+
+    /// A wrapper on `add_blocks()` to avoid passing enums explicitly.
+    pub fn add_stray_blocks(
+        &self,
+        state: BeaconState<E>,
+        slot: Slot,
+        num_blocks: usize,
+        attesting_validators: &[usize],
+    ) -> (
+        HashMap<Slot, SignedBeaconBlockHash>,
+        HashMap<Slot, BeaconStateHash>,
+        Slot,
+        SignedBeaconBlockHash,
+        BeaconState<E>,
+    ) {
+        let block_strategy = BlockStrategy::ForkCanonicalChainAt {
+            previous_slot: slot,
+            first_slot: slot + 2,
+        };
+        self.add_blocks(
+            state,
+            slot + 2,
+            num_blocks,
+            attesting_validators,
+            block_strategy,
+        )
+    }
+
     /// Returns a newly created block, signed by the proposer for the given slot.
     fn build_block(
         &self,