Merge branch 'main' into L1-59

.github/scripts/run_consensus.sh

@@ -118,7 +118,7 @@ function run_containers() {
   for index in $(seq 0 "${authorities_count}"); do
     containers+=("Node${index}")
   done
-  docker-compose $(get_compose_file_list "${docker_compose_file}" "${override_file}") up -d "${containers[@]}"
+  docker compose $(get_compose_file_list "${docker_compose_file}" "${override_file}") up -d "${containers[@]}"
 }
 
 function archive_logs() {
@@ -133,7 +133,7 @@ function archive_logs() {
   pushd $(mktemp -d) > /dev/null
   for index in $(seq 0 "${node_count}"); do
     echo "Archiving "Node${index}" logs..."
-    docker-compose ${compose_file_list} logs --no-color --no-log-prefix "Node${index}" > "Node${index}.log"
+    docker compose ${compose_file_list} logs --no-color --no-log-prefix "Node${index}" > "Node${index}.log"
   done
   tar -czf "${tarball_output}" Node*
   popd > /dev/null
@@ -163,12 +163,12 @@ else
   exit 1
 fi
 
-if docker inspect ${NODE_IMAGE} > /dev/null; then
-  echo "aleph-node image tag ${NODE_IMAGE} found locally"
+if docker inspect ${CHAIN_BOOTSTRAPPER_IMAGE} > /dev/null; then
+  echo "chain-bootstrapper image tag ${CHAIN_BOOTSTRAPPER_IMAGE} found locally"
 else
-  echo "${NODE_IMAGE} not found locally."
+  echo "${CHAIN_BOOTSTRAPPER_IMAGE} not found locally."
   echo "Build image first with:"
-  echo "docker build -t ${NODE_IMAGE} -f docker/Dockerfile ."
+  echo "docker build -t ${CHAIN_BOOTSTRAPPER_IMAGE} -f ./bin/chain-bootstrapper/Dockerfile ."
   exit 1
 fi
 

.github/workflows/on-pull-request-commit.yml

@@ -75,3 +75,21 @@ jobs:
           test-case: finalization::finalization
           # yamllint disable-line rule:line-length
           aleph-e2e-client-image: ${{ needs.build-aleph-e2e-client-image.outputs.aleph-e2e-client-image }}
+
+  run-e2e-token-transfer-test:
+    name: Run e2e token transfer test
+    needs:
+      - build-test-node
+      - build-aleph-e2e-client-image
+      - build-chain-bootstrapper
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Checkout source code
+        uses: actions/checkout@v4
+
+      - name: Run e2e test
+        uses: ./.github/actions/run-e2e-test
+        with:
+          test-case: token_transfer
+          # yamllint disable-line rule:line-length
+          aleph-e2e-client-image: ${{ needs.build-aleph-e2e-client-image.outputs.aleph-e2e-client-image }}

aggregator/src/aggregator.rs

@@ -171,16 +171,12 @@ impl<
                         warn!(target: "aleph-aggregator", "failed broadcasting a message from rmc: {:?}", e);
                     }
                 }
-                message_from_network = self.network.next() => match message_from_network {
-                    Some(message) => {
-                        trace!(target: "aleph-aggregator", "Received message for rmc: {:?}", message);
-                        if let Some(multisigned) = self.rmc_service.process_message(message) {
-                            self.multisigned_events.push_back(multisigned);
-                        }
-                    },
-                    None => {
-                        // In case the network is down we can terminate (?).
-                        return Err(IOError::NetworkChannelClosed);
+                message_from_network = self.network.next() =>  {
+                    // In case the network is down we can terminate (?).
+                    let message = message_from_network.ok_or(IOError::NetworkChannelClosed)?;
+                    trace!(target: "aleph-aggregator", "Received message for rmc: {:?}", message);
+                    if let Some(multisigned) = self.rmc_service.process_message(message) {
+                        self.multisigned_events.push_back(multisigned);
                     }
                 }
             }

clique/src/lib.rs

@@ -24,7 +24,7 @@ mod testing;
 
 pub use crypto::{PublicKey, SecretKey};
 pub use rate_limiting::{RateLimitingDialer, RateLimitingListener};
-pub use service::{Service, SpawnHandleT};
+pub use service::{Service, SpawnHandleExt, SpawnHandleT};
 
 const LOG_TARGET: &str = "network-clique";
 /// A basic alias for properties we expect basic data to satisfy.

clique/src/service.rs

@@ -9,7 +9,7 @@ use futures::{
         mpsc::{self, UnboundedSender},
         oneshot,
     },
-    Future, StreamExt,
+    Future, StreamExt, TryFutureExt,
 };
 use log::{info, trace, warn};
 use substrate_prometheus_endpoint::Registry;
@@ -83,7 +83,9 @@ impl<PK: PublicKey, D: Data, A: Data> Network<PK, A, D> for ServiceInterface<PK,
 pub trait SpawnHandleT {
     /// Run task
     fn spawn(&self, name: &'static str, task: impl Future<Output = ()> + Send + 'static);
+}
 
+pub trait SpawnHandleExt: SpawnHandleT {
     /// Run an essential task
     fn spawn_essential(
         &self,
@@ -92,6 +94,26 @@ pub trait SpawnHandleT {
     ) -> Pin<Box<dyn Future<Output = Result<(), ()>> + Send>>;
 }
 
+impl<SH: SpawnHandleT> SpawnHandleExt for SH {
+    fn spawn_essential(
+        &self,
+        name: &'static str,
+        task: impl Future<Output = ()> + Send + 'static,
+    ) -> Pin<Box<dyn Future<Output = Result<(), ()>> + Send>> {
+        let (tx, rx) = oneshot::channel();
+        self.spawn(name, async move {
+            task.await;
+            let _ = tx.send(());
+        });
+        Box::pin(rx.map_err(move |_| {
+            warn!(
+                target: LOG_TARGET,
+                "Task '{name}' exited early."
+            )
+        }))
+    }
+}
+
 #[derive(Debug)]
 pub enum Error {
     Commands,

clique/src/testing/clique_network.rs

@@ -26,14 +26,6 @@ impl SpawnHandleT for Spawner {
     fn spawn(&self, name: &'static str, task: impl futures::Future<Output = ()> + Send + 'static) {
         SpawnHandle::spawn(self, name, task)
     }
-
-    fn spawn_essential(
-        &self,
-        name: &'static str,
-        task: impl futures::Future<Output = ()> + Send + 'static,
-    ) -> std::pin::Pin<Box<dyn futures::Future<Output = Result<(), ()>> + Send>> {
-        SpawnHandle::spawn_essential(self, name, task)
-    }
 }
 
 pub const LOG_TARGET: &str = "network-clique-test";

finality-aleph/src/abft/current/mod.rs

@@ -2,7 +2,7 @@ use std::time::Duration;
 
 use current_aleph_bft::{create_config, default_delay_config, Config, LocalIO, Terminator};
 use log::debug;
-use network_clique::SpawnHandleT;
+use network_clique::SpawnHandleExt;
 
 mod network;
 mod traits;

finality-aleph/src/abft/legacy/mod.rs

@@ -2,7 +2,7 @@ use std::{pin::Pin, time::Duration};
 
 use legacy_aleph_bft::{create_config, default_delay_config, Config, LocalIO, Terminator};
 use log::debug;
-use network_clique::SpawnHandleT;
+use network_clique::SpawnHandleExt;
 
 mod network;
 mod traits;

finality-aleph/src/abft/traits.rs

@@ -2,8 +2,8 @@
 
 use std::{cmp::Ordering, fmt::Debug, hash::Hash as StdHash, marker::PhantomData, pin::Pin};
 
-use futures::{channel::oneshot, Future, TryFutureExt};
-use network_clique::SpawnHandleT;
+use futures::{channel::oneshot, Future};
+use network_clique::{SpawnHandleExt, SpawnHandleT};
 use parity_scale_codec::{Codec, Decode, Encode};
 use sc_service::SpawnTaskHandle;
 use sp_runtime::traits::Hash as SpHash;
@@ -88,7 +88,7 @@ impl SpawnHandle {
             let result = task.await;
             let _ = tx.send(result);
         };
-        let result = <Self as SpawnHandleT>::spawn_essential(self, name, wrapped_task);
+        let result = <Self as SpawnHandleExt>::spawn_essential(self, name, wrapped_task);
         let wrapped_result = async move {
             let main_result = result.await;
             if main_result.is_err() {
@@ -111,19 +111,6 @@ impl SpawnHandleT for SpawnHandle {
     fn spawn(&self, name: &'static str, task: impl Future<Output = ()> + Send + 'static) {
         self.0.spawn(name, None, task)
     }
-
-    fn spawn_essential(
-        &self,
-        name: &'static str,
-        task: impl Future<Output = ()> + Send + 'static,
-    ) -> Pin<Box<dyn Future<Output = Result<(), ()>> + Send>> {
-        let (tx, rx) = oneshot::channel();
-        self.spawn(name, async move {
-            task.await;
-            let _ = tx.send(());
-        });
-        Box::pin(rx.map_err(|_| ()))
-    }
 }
 
 impl current_aleph_bft::SpawnHandle for SpawnHandle {
@@ -136,7 +123,7 @@ impl current_aleph_bft::SpawnHandle for SpawnHandle {
         name: &'static str,
         task: impl Future<Output = ()> + Send + 'static,
     ) -> current_aleph_bft::TaskHandle {
-        SpawnHandleT::spawn_essential(self, name, task)
+        SpawnHandleExt::spawn_essential(self, name, task)
     }
 }
 
@@ -150,6 +137,6 @@ impl legacy_aleph_bft::SpawnHandle for SpawnHandle {
         name: &'static str,
         task: impl Future<Output = ()> + Send + 'static,
     ) -> legacy_aleph_bft::TaskHandle {
-        SpawnHandleT::spawn_essential(self, name, task)
+        SpawnHandleExt::spawn_essential(self, name, task)
     }
 }

finality-aleph/src/network/data/component.rs

@@ -1,6 +1,6 @@
 use std::{fmt::Display, marker::PhantomData};
 
-use futures::{channel::mpsc, StreamExt};
+use futures::{channel::mpsc, stream::FusedStream, StreamExt};
 use log::warn;
 
 use crate::{
@@ -135,7 +135,11 @@ impl<D: Data> Sender<D> for mpsc::UnboundedSender<(D, Recipient)> {
 #[async_trait::async_trait]
 impl<D: Data> Receiver<D> for mpsc::UnboundedReceiver<D> {
     async fn next(&mut self) -> Option<D> {
-        StreamExt::next(self).await
+        if self.is_terminated() {
+            None
+        } else {
+            StreamExt::next(self).await
+        }
     }
 }
 

finality-aleph/src/network/data/split.rs

@@ -142,7 +142,7 @@ async fn forward_or_wait<
     RightData: Data,
     R: Receiver<Split<LeftData, RightData>>,
 >(
-    receiver: &Arc<Mutex<R>>,
+    receiver: &Mutex<R>,
     left_sender: &mpsc::UnboundedSender<LeftData>,
     right_sender: &mpsc::UnboundedSender<RightData>,
     name: &str,

finality-aleph/src/network/substrate.rs

@@ -149,16 +149,15 @@ impl<D: Data> GossipNetwork<D> for ProtocolNetwork {
             tokio::select! {
                 maybe_event = self.service.next_event() => {
                     let event = maybe_event.ok_or(Self::Error::NetworkStreamTerminated)?;
-                    if let Some((message, peer_id)) = self.handle_network_event(event) {
-                        match D::decode_all(&mut &message[..]) {
-                            Ok(message) => return Ok((message, peer_id)),
-                            Err(e) => {
-                                warn!(
-                                    target: LOG_TARGET,
-                                    "Error decoding message: {}", e
-                                )
-                            },
-                        }
+                    let Some((message, peer_id)) = self.handle_network_event(event) else { continue };
+                    match D::decode_all(&mut &message[..]) {
+                        Ok(message) => return Ok((message, peer_id)),
+                        Err(e) => {
+                            warn!(
+                                target: LOG_TARGET,
+                                "Error decoding message: {}", e
+                            )
+                        },
                     }
                 },
                 _ = status_ticker.tick() => {

finality-aleph/src/party/manager/mod.rs

@@ -3,7 +3,7 @@ use std::{collections::HashSet, marker::PhantomData, sync::Arc};
 use async_trait::async_trait;
 use futures::channel::oneshot;
 use log::{debug, info, trace, warn};
-use network_clique::SpawnHandleT;
+use network_clique::SpawnHandleExt;
 use pallet_aleph_runtime_api::AlephSessionApi;
 use sc_keystore::{Keystore, LocalKeystore};
 use sp_application_crypto::RuntimeAppPublic;

finality-aleph/src/party/manager/task.rs

@@ -2,7 +2,7 @@ use std::{boxed::Box, pin::Pin};
 
 use futures::channel::oneshot;
 use log::{debug, warn};
-use network_clique::SpawnHandleT;
+use network_clique::SpawnHandleExt;
 
 use crate::{Future, SpawnHandle};