Factor unit validation out of runway

This required quite a lot of other changes as well, so this is quite
large. It also looks... not ideal and in places _more_ convoluted than
beforehand, but completely untangling all this in a single commit would
require it to be even larger.

consensus/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aleph-bft"
-version = "0.36.1"
+version = "0.36.2"
 edition = "2021"
 authors = ["Cardinal Cryptography"]
 categories = ["algorithms", "data-structures", "cryptography", "database"]

consensus/src/alerts/handler.rs

@@ -1,5 +1,6 @@
 use crate::{
     alerts::{Alert, AlertMessage, ForkProof, ForkingNotification},
+    units::Unit,
     Data, Hasher, Keychain, MultiKeychain, Multisigned, NodeIndex, PartialMultisignature,
     Recipient, SessionId, Signature, Signed, UncheckedSigned,
 };

consensus/src/alerts/mod.rs

@@ -1,6 +1,7 @@
 use crate::{
-    units::UncheckedSignedUnit, Data, Hasher, Index, Keychain, MultiKeychain, Multisigned,
-    NodeIndex, PartialMultisignature, Signable, Signature, UncheckedSigned,
+    units::{UncheckedSignedUnit, Unit},
+    Data, Hasher, Index, Keychain, MultiKeychain, Multisigned, NodeIndex, PartialMultisignature,
+    Signable, Signature, UncheckedSigned,
 };
 use aleph_bft_rmc::Message as RmcMessage;
 use codec::{Decode, Encode};

consensus/src/alerts/service.rs

@@ -7,7 +7,7 @@ use crate::{
 };
 use aleph_bft_rmc::{DoublingDelayScheduler, Message as RmcMessage};
 use futures::{FutureExt, StreamExt};
-use log::{debug, error, warn};
+use log::{debug, error, trace, warn};
 use std::time::Duration;
 
 const LOG_TARGET: &str = "AlephBFT-alerter";
@@ -147,6 +147,7 @@ impl<H: Hasher, D: Data, MK: MultiKeychain> Service<H, D, MK> {
     }
 
     fn handle_alert_from_runway(&mut self, alert: Alert<H, D, MK::Signature>) {
+        trace!(target: LOG_TARGET, "Handling alert {:?}.", alert);
         let (message, recipient, hash) = self.handler.on_own_alert(alert.clone());
         self.send_message_for_network(message, recipient);
         if let Some(multisigned) = self.rmc_service.start_rmc(hash) {

consensus/src/backup/loader.rs

@@ -10,7 +10,7 @@ use futures::{channel::oneshot, AsyncRead, AsyncReadExt};
 use log::{error, info, warn};
 
 use crate::{
-    units::{UncheckedSignedUnit, UnitCoord},
+    units::{UncheckedSignedUnit, Unit, UnitCoord},
     Data, Hasher, NodeIndex, Round, SessionId, Signature,
 };
 
@@ -246,7 +246,7 @@ mod tests {
     use crate::{
         backup::BackupLoader,
         units::{
-            create_preunits, creator_set, preunit_to_unchecked_signed_unit, preunit_to_unit,
+            create_preunits, creator_set, preunit_to_full_unit, preunit_to_unchecked_signed_unit,
             UncheckedSignedUnit as GenericUncheckedSignedUnit,
         },
         NodeCount, NodeIndex, Round, SessionId,
@@ -277,14 +277,14 @@ mod tests {
 
             let units: Vec<_> = pre_units
                 .iter()
-                .map(|(pre_unit, _)| preunit_to_unit(pre_unit.clone(), session_id))
+                .map(|pre_unit| preunit_to_full_unit(pre_unit.clone(), session_id))
                 .collect();
             for creator in creators.iter_mut() {
                 creator.add_units(&units);
             }
 
             let mut unchecked_signed_units = Vec::with_capacity(pre_units.len());
-            for ((pre_unit, _), keychain) in pre_units.into_iter().zip(keychains.iter()) {
+            for (pre_unit, keychain) in pre_units.into_iter().zip(keychains.iter()) {
                 unchecked_signed_units.push(preunit_to_unchecked_signed_unit(
                     pre_unit, session_id, keychain,
                 ))

consensus/src/backup/saver.rs

@@ -140,7 +140,7 @@ mod tests {
         let units: Vec<TestUnit> = (0..5)
             .map(|k| {
                 preunit_to_unchecked_signed_unit(
-                    creators[k].create_unit(0).unwrap().0,
+                    creators[k].create_unit(0).unwrap(),
                     0,
                     &keychains[k],
                 )

consensus/src/consensus.rs

@@ -7,20 +7,20 @@ use log::{debug, error};
 
 use crate::{
     config::Config,
-    creation::{self, SignedUnitWithParents},
-    extension::Service as Extender,
-    handle_task_termination,
-    reconstruction::Service as ReconstructionService,
-    runway::{NotificationIn, NotificationOut},
+    creation, handle_task_termination,
+    reconstruction::{ReconstructedUnit, Request, Service as ReconstructionService},
+    runway::ExplicitParents,
+    units::SignedUnit,
     Data, DataProvider, Hasher, MultiKeychain, Receiver, Round, Sender, SpawnHandle, Terminator,
 };
 
 pub struct IO<H: Hasher, D: Data, MK: MultiKeychain, DP: DataProvider<D>> {
-    pub incoming_notifications: Receiver<NotificationIn<H>>,
-    pub outgoing_notifications: Sender<NotificationOut<H>>,
-    pub units_for_runway: Sender<SignedUnitWithParents<H, D, MK>>,
+    pub units_from_runway: Receiver<SignedUnit<H, D, MK>>,
+    pub parents_from_runway: Receiver<ExplicitParents<H>>,
+    pub units_for_runway: Sender<ReconstructedUnit<SignedUnit<H, D, MK>>>,
+    pub requests_for_runway: Sender<Request<H>>,
+    pub new_units_for_runway: Sender<SignedUnit<H, D, MK>>,
     pub data_provider: DP,
-    pub ordered_batch_tx: Sender<Vec<H::Hash>>,
     pub starting_round: oneshot::Receiver<Option<Round>>,
 }
 
@@ -33,30 +33,22 @@ pub async fn run<H: Hasher, D: Data, MK: MultiKeychain, DP: DataProvider<D>>(
 ) {
     debug!(target: "AlephBFT", "{:?} Starting all services...", conf.node_ix());
     let IO {
-        incoming_notifications,
-        outgoing_notifications,
+        units_from_runway,
+        parents_from_runway,
         units_for_runway,
+        requests_for_runway,
+        new_units_for_runway,
         data_provider,
-        ordered_batch_tx,
         starting_round,
     } = io;
 
     let index = conf.node_ix();
 
-    let (electors_tx, electors_rx) = mpsc::unbounded();
-    let extender = Extender::<H>::new(index, electors_rx, ordered_batch_tx);
-    let extender_terminator = terminator.add_offspring_connection("AlephBFT-extender");
-    let mut extender_handle = spawn_handle
-        .spawn_essential("consensus/extender", async move {
-            extender.run(extender_terminator).await
-        })
-        .fuse();
-
     let (parents_for_creator, parents_from_dag) = mpsc::unbounded();
 
     let creator_terminator = terminator.add_offspring_connection("creator");
     let io = creation::IO {
-        outgoing_units: units_for_runway,
+        outgoing_units: new_units_for_runway,
         incoming_parents: parents_from_dag,
         data_provider,
     };
@@ -75,10 +67,11 @@ pub async fn run<H: Hasher, D: Data, MK: MultiKeychain, DP: DataProvider<D>>(
     };
 
     let reconstruction = ReconstructionService::new(
-        incoming_notifications,
-        outgoing_notifications,
+        units_from_runway,
+        parents_from_runway,
+        requests_for_runway,
+        units_for_runway,
         parents_for_creator,
-        electors_tx,
     );
 
     let reconstruction_terminator = terminator.add_offspring_connection("reconstruction");
@@ -97,9 +90,6 @@ pub async fn run<H: Hasher, D: Data, MK: MultiKeychain, DP: DataProvider<D>>(
         _ = creator_panic_handle.fuse() => {
             error!(target: "AlephBFT-consensus", "{:?} creator task terminated early with its task being dropped.", index);
         },
-        _ = extender_handle => {
-            debug!(target: "AlephBFT-consensus", "{:?} extender task terminated early.", index);
-        }
     }
     debug!(target: "AlephBFT", "{:?} All services stopping.", index);
 
@@ -114,7 +104,6 @@ pub async fn run<H: Hasher, D: Data, MK: MultiKeychain, DP: DataProvider<D>>(
     )
     .await;
     handle_task_termination(creator_handle, "AlephBFT-consensus", "Creator", index).await;
-    handle_task_termination(extender_handle, "AlephBFT-consensus", "Extender", index).await;
 
     debug!(target: "AlephBFT", "{:?} All services stopped.", index);
 }