Adjust to new versions

aggregator/Cargo.toml

@@ -8,7 +8,7 @@ license = "Apache 2.0"
 
 [dependencies]
 aleph-bft-rmc = "0.6"
-aleph-bft-types = "0.7"
+aleph-bft-types = "0.8"
 
 async-trait = "0.1"
 futures = "0.3"

finality-aleph/Cargo.toml

@@ -7,15 +7,15 @@ license = "Apache 2.0"
 
 [dependencies]
 # fixed version to 'freeze' some types used in abft, mainly `SignatureSet` used in justification and signature aggregation
-aleph-bft-crypto = "0.4"
+aleph-bft-crypto = "0.5"
 
 current-aleph-bft = { package = "aleph-bft", version = "0.20" }
 current-aleph-bft-rmc = { package = "aleph-bft-rmc", version = "0.6" }
 legacy-aleph-bft = { package = "aleph-bft", version = "0.19" }
 legacy-aleph-bft-rmc = { package = "aleph-bft-rmc", version = "0.5" }
 
 aleph-primitives = { package = "primitives", path = "../primitives" }
-legacy-aleph-aggregator = { package = "aggregator", git = "https://github.com/Cardinal-Cryptography/aleph-node.git", tag = "aggregator-v0.2.0" }
+legacy-aleph-aggregator = { package = "aggregator", git = "https://github.com/Cardinal-Cryptography/aleph-node.git", tag = "aggregator-v0.2.1" }
 current-aleph-aggregator = { path = "../aggregator", package = "aggregator" }
 
 async-trait = "0.1"

finality-aleph/src/abft/common.rs

@@ -25,23 +25,39 @@ fn exponential_slowdown(
 }
 
 pub type DelaySchedule = Arc<dyn Fn(usize) -> Duration + Sync + Send + 'static>;
+pub type RecipientCountSchedule = Arc<dyn Fn(usize) -> usize + Sync + Send + 'static>;
 
 pub fn unit_creation_delay_fn(unit_creation_delay: UnitCreationDelay) -> DelaySchedule {
-    Arc::new(move |t| {
-        if t == 0 {
-            Duration::from_millis(2000)
-        } else {
-            exponential_slowdown(t, unit_creation_delay.0 as f64, 5000, 1.005)
-        }
+    Arc::new(move |t| match t {
+        0 => Duration::from_millis(2000),
+        _ => exponential_slowdown(t, unit_creation_delay.0 as f64, 5000, 1.005),
+    })
+}
+
+pub fn coord_request_delay_fn() -> DelaySchedule {
+    Arc::new(|t| match t {
+        0 => Duration::from_millis(0),
+        1 => Duration::from_millis(50),
+        2 => Duration::from_millis(1000),
+        _ => Duration::from_millis(3000 * (t as u64 - 2)),
     })
 }
 
+pub fn coord_request_recipients_fn() -> RecipientCountSchedule {
+    Arc::new(|t| if t <= 2 { 3 } else { 1 })
+}
+
 pub struct DelayConfig {
     pub tick_interval: Duration,
     pub requests_interval: Duration,
     pub unit_rebroadcast_interval_min: Duration,
     pub unit_rebroadcast_interval_max: Duration,
     pub unit_creation_delay: DelaySchedule,
+    pub coord_request_delay: DelaySchedule,
+    pub coord_request_recipients: RecipientCountSchedule,
+    pub parent_request_delay: DelaySchedule,
+    pub parent_request_recipients: RecipientCountSchedule,
+    pub newest_request_delay: DelaySchedule,
 }
 
 pub struct AlephConfig {
@@ -83,10 +99,14 @@ impl From<DelayConfig> for current_aleph_bft::DelayConfig {
     fn from(cfg: DelayConfig) -> Self {
         Self {
             tick_interval: cfg.tick_interval,
-            requests_interval: cfg.requests_interval,
             unit_rebroadcast_interval_max: cfg.unit_rebroadcast_interval_max,
             unit_rebroadcast_interval_min: cfg.unit_rebroadcast_interval_min,
             unit_creation_delay: cfg.unit_creation_delay,
+            coord_request_delay: cfg.coord_request_delay,
+            coord_request_recipients: cfg.coord_request_recipients,
+            parent_request_delay: cfg.parent_request_delay,
+            parent_request_recipients: cfg.parent_request_recipients,
+            newest_request_delay: cfg.newest_request_delay,
         }
     }
 }

finality-aleph/src/abft/current.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use current_aleph_bft::{Config, LocalIO, Terminator};
 use log::debug;
@@ -7,7 +7,10 @@ use sp_runtime::traits::Block;
 
 use crate::{
     abft::{
-        common::{unit_creation_delay_fn, AlephConfig, DelayConfig},
+        common::{
+            coord_request_delay_fn, coord_request_recipients_fn, unit_creation_delay_fn,
+            AlephConfig, DelayConfig,
+        },
         NetworkWrapper, SpawnHandleT,
     },
     crypto::Signature,
@@ -76,11 +79,16 @@ pub fn create_aleph_config(
     unit_creation_delay: UnitCreationDelay,
 ) -> Config {
     let delay_config = DelayConfig {
-        tick_interval: Duration::from_millis(100),
+        tick_interval: Duration::from_millis(10),
         requests_interval: Duration::from_millis(3000),
         unit_rebroadcast_interval_min: Duration::from_millis(15000),
         unit_rebroadcast_interval_max: Duration::from_millis(20000),
         unit_creation_delay: unit_creation_delay_fn(unit_creation_delay),
+        coord_request_delay: coord_request_delay_fn(),
+        coord_request_recipients: coord_request_recipients_fn(),
+        parent_request_delay: Arc::new(|_| Duration::from_millis(3000)),
+        parent_request_recipients: Arc::new(|_| 1),
+        newest_request_delay: Arc::new(|_| Duration::from_millis(3000)),
     };
 
     AlephConfig::new(delay_config, n_members, node_id, session_id).into()

finality-aleph/src/abft/legacy.rs

@@ -1,13 +1,16 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
-use legacy_aleph_bft::{Config, LocalIO};
+use legacy_aleph_bft::{Config, LocalIO, Terminator};
 use log::debug;
 use sp_blockchain::HeaderBackend;
 use sp_runtime::traits::Block;
 
 use crate::{
     abft::{
-        common::{unit_creation_delay_fn, AlephConfig, DelayConfig},
+        common::{
+            coord_request_delay_fn, coord_request_recipients_fn, unit_creation_delay_fn,
+            AlephConfig, DelayConfig,
+        },
         NetworkWrapper, SpawnHandleT,
     },
     data_io::{AlephData, OrderedDataInterpreter},
@@ -41,6 +44,7 @@ pub fn run_member<
         session_id,
     } = subtask_common;
     let (stop, exit) = oneshot::channel();
+    let member_terminator = Terminator::create_root(exit, "member");
     let local_io = LocalIO::new(data_provider, ordered_data_interpreter, backup.0, backup.1);
 
     let task = {
@@ -53,7 +57,7 @@ pub fn run_member<
                 network,
                 multikeychain,
                 spawn_handle,
-                exit,
+                member_terminator,
             )
             .await;
             debug!(target: "aleph-party", "Member task stopped for {:?}", session_id);
@@ -71,11 +75,16 @@ pub fn create_aleph_config(
     unit_creation_delay: UnitCreationDelay,
 ) -> Config {
     let delay_config = DelayConfig {
-        tick_interval: Duration::from_millis(100),
+        tick_interval: Duration::from_millis(10),
         requests_interval: Duration::from_millis(3000),
         unit_rebroadcast_interval_min: Duration::from_millis(15000),
         unit_rebroadcast_interval_max: Duration::from_millis(20000),
         unit_creation_delay: unit_creation_delay_fn(unit_creation_delay),
+        coord_request_delay: coord_request_delay_fn(),
+        coord_request_recipients: coord_request_recipients_fn(),
+        parent_request_delay: Arc::new(|_| Duration::from_millis(3000)),
+        parent_request_recipients: Arc::new(|_| 1),
+        newest_request_delay: Arc::new(|_| Duration::from_millis(3000)),
     };
 
     AlephConfig::new(delay_config, n_members, node_id, session_id).into()

finality-aleph/src/aggregation/mod.rs

@@ -220,7 +220,7 @@ where
     fn send(
         &self,
         data: D,
-        recipient: legacy_aleph_bft::Recipient,
+        recipient: current_aleph_bft::Recipient,
     ) -> Result<(), CurrentNetworkError> {
         self.0.send(data, recipient.into()).map_err(|e| match e {
             SendError::SendFailed => CurrentNetworkError::SendFail,