[light-client] Supervisor (#302)

* Remove Peers struct, only supply one peer to each light client instance

* Remove fork detector from light client instance

* Add Debug instance to LightClient

* Require all components and their deps to be clonable

* Start work on supervisor and fork detection

* Allowing supplying the state to work over to the light client

* Remove unnecessary instances of operations traits

* Remove unnecessary Clone and DynClone instances

* Improve docs

* Small refactor in supervisor

* Finalize fork detection

* Fix wrong uses of internal light client state instead of the supplied one

* Remove light client's internal state

* Remove outdated comments

* Ensure the supervisor can be sent across thread boundaries

* Rename Handler to Handle

* Update comment

* Update example to invoke supervisor

* Formatting

* Working supervisor example

* Remove light store committed by error

* Compare headers hashes instead of for equality in fork detector

* Bubble up I/O errors that may occur during fork detection

* Inject fork detector into supervisor

* Deduplicate Handle::verify_to_highest/verify_to_target

* Move PeerList and its builder into their own module

* Record error which caused peer to be deemed faulty

* Remove prelude module in favor of explicit imports

* Hoist primary hash computation outside of for loop

* Throw an error if there are no witnesses when doing fork detection

* Rename leftover secondaries to witnesses

* Set clock_drift parameter to 10 seconds to match Go default

* Add configurable timeouts to Io component

* Add the primary node as a witness in the example CLI

* Remove witness from peers when request times out during fork detection

* Fix PeerList::swap_primary

* Improve PeerListBuilder API

* Add doc comments

* Use enum instead of 0 to denote latest height

light-client/Cargo.toml

@@ -8,16 +8,18 @@ edition = "2018"
 tendermint = { path = "../tendermint" }
 
 anomaly = { version = "0.2.0", features = ["serializer"] }
+contracts = "0.4.0"
+crossbeam-channel = "0.4.2"
 derive_more = "0.99.5"
+futures = "0.3.4"
+prost-amino = "0.5.0"
 serde = "1.0.106"
+serde_cbor = "0.11.1"
 serde_derive = "1.0.106"
+sled = "0.31.0"
+static_assertions = "1.1.0"
 thiserror = "1.0.15"
-futures = "0.3.4"
 tokio = "0.2.20"
-prost-amino = "0.5.0"
-contracts = "0.4.0"
-sled = "0.31.0"
-serde_cbor = "0.11.1"
 
 [dev-dependencies]
 serde_json = "1.0.51"

light-client/examples/light_client.rs

@@ -1,8 +1,26 @@
+use tendermint_light_client::{
+    components::{
+        clock::SystemClock,
+        io::{AtHeight, Io, ProdIo},
+        scheduler,
+        verifier::ProdVerifier,
+    },
+    fork_detector::ProdForkDetector,
+    light_client::{self, LightClient},
+    peer_list::PeerList,
+    state::State,
+    store::{sled::SledStore, LightStore, VerifiedStatus},
+    supervisor::{Instance, Supervisor},
+    types::{Height, PeerId, Time, TrustThreshold},
+};
+
 use gumdrop::Options;
-use tendermint_light_client::prelude::Height;
 
 use std::collections::HashMap;
-use std::path::PathBuf;
+use std::{
+    path::{Path, PathBuf},
+    time::Duration,
+};
 
 #[derive(Debug, Options)]
 struct CliOptions {
@@ -57,46 +75,47 @@ fn main() {
     }
 }
 
-fn sync_cmd(opts: SyncOpts) {
-    use tendermint_light_client::components::scheduler;
-    use tendermint_light_client::prelude::*;
-
-    let primary_addr = opts.address;
-    let primary: PeerId = "BADFADAD0BEFEEDC0C0ADEADBEEFC0FFEEFACADE".parse().unwrap();
-
+fn make_instance(
+    peer_id: PeerId,
+    addr: tendermint::net::Address,
+    db_path: impl AsRef<Path>,
+    opts: &SyncOpts,
+) -> Instance {
     let mut peer_map = HashMap::new();
-    peer_map.insert(primary, primary_addr);
+    peer_map.insert(peer_id, addr);
 
-    let mut io = ProdIo::new(peer_map);
+    let timeout = Duration::from_secs(10);
+    let io = ProdIo::new(peer_map, Some(timeout));
 
-    let db = sled::open(opts.db_path).unwrap_or_else(|e| {
+    let db = sled::open(db_path).unwrap_or_else(|e| {
         println!("[ error ] could not open database: {}", e);
         std::process::exit(1);
     });
 
     let mut light_store = SledStore::new(db);
 
     if let Some(height) = opts.trusted_height {
-        let trusted_state = io.fetch_light_block(primary, height).unwrap_or_else(|e| {
-            println!("[ error ] could not retrieve trusted header: {}", e);
-            std::process::exit(1);
-        });
+        let trusted_state = io
+            .fetch_light_block(peer_id, AtHeight::At(height))
+            .unwrap_or_else(|e| {
+                println!("[ error ] could not retrieve trusted header: {}", e);
+                std::process::exit(1);
+            });
 
         light_store.insert(trusted_state, VerifiedStatus::Verified);
+    } else {
+        if light_store.latest(VerifiedStatus::Verified).is_none() {
+            println!("[ error ] no trusted state in database, please specify a trusted header");
+            std::process::exit(1);
+        }
     }
 
-    let peers = Peers {
-        primary,
-        witnesses: Vec::new(),
-    };
-
     let state = State {
-        peers,
         light_store: Box::new(light_store),
         verification_trace: HashMap::new(),
     };
 
-    let options = Options {
+    let options = light_client::Options {
         trust_threshold: TrustThreshold {
             numerator: 1,
             denominator: 3,
@@ -109,27 +128,44 @@ fn sync_cmd(opts: SyncOpts) {
     let verifier = ProdVerifier::default();
     let clock = SystemClock;
     let scheduler = scheduler::basic_bisecting_schedule;
-    let fork_detector = ProdForkDetector::default();
 
-    let mut light_client = LightClient::new(
-        state,
-        options,
-        clock,
-        scheduler,
-        verifier,
-        fork_detector,
-        io,
-    );
+    let light_client = LightClient::new(peer_id, options, clock, scheduler, verifier, io);
+
+    Instance::new(light_client, state)
+}
+
+fn sync_cmd(opts: SyncOpts) {
+    let addr = opts.address.clone();
+
+    let primary: PeerId = "BADFADAD0BEFEEDC0C0ADEADBEEFC0FFEEFACADE".parse().unwrap();
+    let witness: PeerId = "CEFEEDBADFADAD0C0CEEFACADE0ADEADBEEFC0FF".parse().unwrap();
+
+    let primary_path = opts.db_path.clone().join(primary.to_string());
+    let witness_path = opts.db_path.clone().join(witness.to_string());
+
+    let primary_instance = make_instance(primary, addr.clone(), primary_path, &opts);
+    let witness_instance = make_instance(witness, addr.clone(), witness_path, &opts);
+
+    let peer_list = PeerList::builder()
+        .primary(primary, primary_instance)
+        .witness(witness, witness_instance)
+        .build();
+
+    let mut supervisor = Supervisor::new(peer_list, ProdForkDetector::default());
+    let mut handle = supervisor.handle();
+
+    std::thread::spawn(|| supervisor.run());
 
     loop {
-        match light_client.verify_to_highest() {
+        handle.verify_to_highest_async(|result| match result {
             Ok(light_block) => {
                 println!("[ info  ] synced to block {}", light_block.height());
             }
             Err(e) => {
                 println!("[ error ] sync failed: {}", e);
             }
-        }
+        });
+
         std::thread::sleep(Duration::from_millis(800));
     }
 }

light-client/src/callback.rs

@@ -0,0 +1,26 @@
+use std::fmt;
+
+/// A boxed `FnOnce(A) -> () + Send`.
+pub struct Callback<A> {
+    inner: Box<dyn FnOnce(A) -> () + Send>,
+}
+
+impl<A> fmt::Debug for Callback<A> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Callback").finish()
+    }
+}
+
+impl<A> Callback<A> {
+    /// Box the given closure in a `Callback`.
+    pub fn new(inner: impl FnOnce(A) -> () + Send + 'static) -> Self {
+        Self {
+            inner: Box::new(inner),
+        }
+    }
+
+    /// Call the underlying closure on `result`.
+    pub fn call(self, result: A) -> () {
+        (self.inner)(result);
+    }
+}

light-client/src/components.rs

@@ -1,7 +1,6 @@
 //! Components used by the Light Client.
 
 pub mod clock;
-pub mod fork_detector;
 pub mod io;
 pub mod scheduler;
 pub mod verifier;

light-client/src/components/clock.rs

@@ -1,12 +1,13 @@
-use crate::prelude::*;
+use crate::types::Time;
 
 /// Abstracts over the current time.
-pub trait Clock {
+pub trait Clock: Send {
     /// Get the current time.
     fn now(&self) -> Time;
 }
 
 /// Provides the current wall clock time.
+#[derive(Copy, Clone)]
 pub struct SystemClock;
 impl Clock for SystemClock {
     fn now(&self) -> Time {