From f9376c9fc510ee0e26e97a5195bf8591fd560c18 Mon Sep 17 00:00:00 2001
From: Maciej Hirsz <hello@maciej.codes>
Date: Fri, 30 Oct 2020 19:03:17 +0100
Subject: [PATCH 1/5] Handle continuation frames

---
 backend/src/node/connector.rs | 46 ++++++++++++++++++++++++++++++++---
 1 file changed, 42 insertions(+), 4 deletions(-)

diff --git a/backend/src/node/connector.rs b/backend/src/node/connector.rs
index d6f09bb93..cc09c3c3f 100644
--- a/backend/src/node/connector.rs
+++ b/backend/src/node/connector.rs
@@ -1,9 +1,11 @@
 use std::time::{Duration, Instant};
 use std::net::Ipv4Addr;
+use std::mem;
 
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use actix::prelude::*;
 use actix_web_actors::ws;
+use actix_http::ws::Item;
 use crate::aggregator::{Aggregator, AddNode};
 use crate::chain::{Chain, UpdateNode, RemoveNode};
 use crate::node::NodeId;
@@ -14,6 +16,8 @@ use crate::util::LocateRequest;
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20);
 /// How long before lack of client response causes a timeout
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(60);
+/// Continuation buffer limit, 1mb
+const CONT_BUF_LIMIT: usize = 1024 * 1024;
 
 pub struct NodeConnector {
     /// Id of the node this connector is responsible for handling
@@ -30,6 +34,8 @@ pub struct NodeConnector {
     ip: Option<Ipv4Addr>,
     /// Actix address of location services
     locator: Recipient<LocateRequest>,
+    /// Buffer for constructing continuation messages
+    contbuf: BytesMut,
 }
 
 impl Actor for NodeConnector {
@@ -57,6 +63,7 @@ impl NodeConnector {
             backlog: Vec::new(),
             ip,
             locator,
+            contbuf: BytesMut::new(),
         }
     }
 
@@ -104,6 +111,27 @@ impl NodeConnector {
             self.backlog.push(msg);
         }
     }
+
+    fn start_frame(&mut self, bytes: &[u8]) {
+        if !self.contbuf.is_empty() {
+            log::error!("Unused continuation buffer");
+            self.contbuf.clear();
+        }
+        self.continue_frame(bytes);
+    }
+
+    fn continue_frame(&mut self, bytes: &[u8]) {
+        if self.contbuf.len() + bytes.len() <= CONT_BUF_LIMIT {
+            self.contbuf.extend_from_slice(&bytes);
+        } else {
+            log::error!("Continuation buffer overflow");
+            self.contbuf = BytesMut::new();
+        }
+    }
+
+    fn finish_frame(&mut self) -> Bytes {
+        mem::replace(&mut self.contbuf, BytesMut::new()).freeze()
+    }
 }
 
 #[derive(Message)]
@@ -148,9 +176,19 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for NodeConnector {
                 return;
             }
             Ok(ws::Message::Nop) => return,
-            Ok(ws::Message::Continuation(_)) => {
-                log::error!("Continuation not supported");
-                return;
+            Ok(ws::Message::Continuation(cont)) => match cont {
+                Item::FirstText(bytes) | Item::FirstBinary(bytes) => {
+                    self.start_frame(&bytes);
+                    return;
+                }
+                Item::Continue(bytes) => {
+                    self.continue_frame(&bytes);
+                    return;
+                }
+                Item::Last(bytes) => {
+                    self.continue_frame(&bytes);
+                    self.finish_frame()
+                }
             }
             Err(error) => {
                 log::error!("{:?}", error);

From e373221147fef93c702905754591a1951571f385 Mon Sep 17 00:00:00 2001
From: Maciej Hirsz <hello@maciej.codes>
Date: Fri, 30 Oct 2020 19:46:25 +0100
Subject: [PATCH 2/5] Parachain multiplexing MVP

---
 backend/src/aggregator.rs     | 20 +++++--------
 backend/src/chain.rs          | 14 ++++++---
 backend/src/node/connector.rs | 55 ++++++++++++++++++++---------------
 backend/src/node/message.rs   |  3 +-
 backend/src/types.rs          |  1 +
 5 files changed, 53 insertions(+), 40 deletions(-)

diff --git a/backend/src/aggregator.rs b/backend/src/aggregator.rs
index ff8ab4765..f13466019 100644
--- a/backend/src/aggregator.rs
+++ b/backend/src/aggregator.rs
@@ -6,7 +6,7 @@ use crate::feed::connector::{FeedConnector, Connected, FeedId};
 use crate::util::DenseMap;
 use crate::feed::{self, FeedMessageSerializer};
 use crate::chain::{self, Chain, ChainId, Label, GetNodeNetworkState};
-use crate::types::{NodeDetails, NodeId};
+use crate::types::{ConnId, NodeDetails, NodeId};
 
 pub struct Aggregator {
     labels: HashMap<Label, ChainId>,
@@ -106,8 +106,11 @@ impl Actor for Aggregator {
 #[derive(Message)]
 #[rtype(result = "()")]
 pub struct AddNode {
+    /// Details of the node being added to the aggregator
     pub node: NodeDetails,
-    pub network_id: Option<Label>,
+    /// Connection id used by the node connector for multiplexing parachains
+    pub conn_id: ConnId,
+    /// Recipient for the initialization message
     pub rec: Recipient<Initialize>,
 }
 
@@ -173,21 +176,14 @@ impl Handler<AddNode> for Aggregator {
     type Result = ();
 
     fn handle(&mut self, msg: AddNode, ctx: &mut Self::Context) {
-        let AddNode { node, network_id, rec } = msg;
+        let AddNode { node, conn_id, rec } = msg;
 
-        let cid = self.lazy_chain(&node.chain, &network_id, ctx);
+        let cid = self.lazy_chain(&node.chain, &None, ctx);
         let chain = self.chains.get_mut(cid).expect("Entry just created above; qed");
 
-        if let Some(network_id) = network_id {
-            // Attach network id to the chain if it was not done yet
-            if chain.network_id.is_none() {
-                chain.network_id = Some(network_id.clone());
-                self.networks.insert(network_id, cid);
-            }
-        }
-
         chain.addr.do_send(chain::AddNode {
             node,
+            conn_id,
             rec,
         });
     }
diff --git a/backend/src/chain.rs b/backend/src/chain.rs
index 03172827f..51d423867 100644
--- a/backend/src/chain.rs
+++ b/backend/src/chain.rs
@@ -9,7 +9,7 @@ use crate::node::{Node, connector::Initialize, message::{NodeMessage, Details}};
 use crate::feed::connector::{FeedId, FeedConnector, Subscribed, Unsubscribed};
 use crate::feed::{self, FeedMessageSerializer};
 use crate::util::{DenseMap, NumStats, now};
-use crate::types::{NodeId, NodeDetails, NodeLocation, Block, Timestamp, BlockNumber};
+use crate::types::{ConnId, NodeId, NodeDetails, NodeLocation, Block, Timestamp, BlockNumber};
 
 const STALE_TIMEOUT: u64 = 2 * 60 * 1000; // 2 minutes
 
@@ -194,7 +194,11 @@ impl Actor for Chain {
 #[derive(Message)]
 #[rtype(result = "()")]
 pub struct AddNode {
+    /// Details of the node being added to the aggregator
     pub node: NodeDetails,
+    /// Connection id used by the node connector for multiplexing parachains
+    pub conn_id: ConnId,
+    /// Recipient for the initialization message
     pub rec: Recipient<Initialize>,
 }
 
@@ -248,11 +252,13 @@ impl Handler<AddNode> for Chain {
     type Result = ();
 
     fn handle(&mut self, msg: AddNode, ctx: &mut Self::Context) {
-        self.increment_label_count(&msg.node.chain);
+        let AddNode { node, conn_id, rec } = msg;
+        self.increment_label_count(&node.chain);
 
-        let nid = self.nodes.add(Node::new(msg.node));
+        let nid = self.nodes.add(Node::new(node));
+        let chain = ctx.address();
 
-        if let Err(_) = msg.rec.do_send(Initialize(nid, ctx.address())) {
+        if let Err(_) = rec.do_send(Initialize { nid, conn_id, chain }) {
             self.nodes.remove(nid);
         } else if let Some(node) = self.nodes.get(nid) {
             self.serializer.push(feed::AddedNode(nid, node));
diff --git a/backend/src/node/connector.rs b/backend/src/node/connector.rs
index cc09c3c3f..68859fd12 100644
--- a/backend/src/node/connector.rs
+++ b/backend/src/node/connector.rs
@@ -1,3 +1,4 @@
+use std::collections::BTreeMap;
 use std::time::{Duration, Instant};
 use std::net::Ipv4Addr;
 use std::mem;
@@ -11,13 +12,14 @@ use crate::chain::{Chain, UpdateNode, RemoveNode};
 use crate::node::NodeId;
 use crate::node::message::{NodeMessage, Details, SystemConnected};
 use crate::util::LocateRequest;
+use crate::types::ConnId;
 
 /// How often heartbeat pings are sent
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20);
 /// How long before lack of client response causes a timeout
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(60);
-/// Continuation buffer limit, 1mb
-const CONT_BUF_LIMIT: usize = 1024 * 1024;
+/// Continuation buffer limit, 10mb
+const CONT_BUF_LIMIT: usize = 10 * 1024 * 1024;
 
 pub struct NodeConnector {
     /// Id of the node this connector is responsible for handling
@@ -26,10 +28,11 @@ pub struct NodeConnector {
     hb: Instant,
     /// Aggregator actor address
     aggregator: Addr<Aggregator>,
-    /// Chain actor address
-    chain: Option<Addr<Chain>>,
+    /// Mapping message connection id to addresses of chains for multiplexing
+    /// a node running multiple parachains
+    chains: BTreeMap<ConnId, Addr<Chain>>,
     /// Backlog of messages to be sent once we get a recipient handle to the chain
-    backlog: Vec<NodeMessage>,
+    backlogs: BTreeMap<ConnId, Vec<NodeMessage>>,
     /// IP address of the node this connector is responsible for
     ip: Option<Ipv4Addr>,
     /// Actix address of location services
@@ -46,7 +49,7 @@ impl Actor for NodeConnector {
     }
 
     fn stopped(&mut self, _: &mut Self::Context) {
-        if let Some(chain) = self.chain.as_ref() {
+        for chain in self.chains.values() {
             chain.do_send(RemoveNode(self.nid));
         }
     }
@@ -59,8 +62,8 @@ impl NodeConnector {
             nid: !0,
             hb: Instant::now(),
             aggregator,
-            chain: None,
-            backlog: Vec::new(),
+            chains: BTreeMap::new(),
+            backlogs: BTreeMap::new(),
             ip,
             locator,
             contbuf: BytesMut::new(),
@@ -78,7 +81,9 @@ impl NodeConnector {
     }
 
     fn handle_message(&mut self, msg: NodeMessage, data: Bytes, ctx: &mut <Self as Actor>::Context) {
-        if let Some(chain) = self.chain.as_ref() {
+        let conn_id = msg.id.unwrap_or(0);
+
+        if let Some(chain) = self.chains.get(&conn_id) {
             chain.do_send(UpdateNode {
                 nid: self.nid,
                 msg,
@@ -92,23 +97,22 @@ impl NodeConnector {
             let SystemConnected { network_id: _, mut node } = connected;
             let rec = ctx.address().recipient();
 
-            // FIXME: mergin chains by network_id is not the way to do it.
-            // This will at least force all CC3 nodes to be aggregated with
-            // the rest.
-            let network_id = None; // network_id.map(Into::into);
+            // FIXME: Use genesis hash instead of names to avoid this mess
             match &*node.chain {
                 "Kusama CC3" => node.chain = "Kusama".into(),
                 "Polkadot CC1" => node.chain = "Polkadot".into(),
                 _ => (),
             }
 
-            self.aggregator.do_send(AddNode { rec, network_id, node });
+            self.aggregator.do_send(AddNode { rec, conn_id, node });
         } else {
-            if self.backlog.len() >= 10 {
-                self.backlog.remove(0);
+            let backlog = self.backlogs.entry(conn_id).or_default();
+
+            if backlog.len() >= 10 {
+                backlog.remove(0);
             }
 
-            self.backlog.push(msg);
+            backlog.push(msg);
         }
     }
 
@@ -136,21 +140,26 @@ impl NodeConnector {
 
 #[derive(Message)]
 #[rtype(result = "()")]
-pub struct Initialize(pub NodeId, pub Addr<Chain>);
+pub struct Initialize {
+    pub nid: NodeId,
+    pub conn_id: ConnId,
+    pub chain: Addr<Chain>,
+}
 
 impl Handler<Initialize> for NodeConnector {
     type Result = ();
 
     fn handle(&mut self, msg: Initialize, _: &mut Self::Context) {
-        let Initialize(nid, chain) = msg;
-        let backlog = std::mem::replace(&mut self.backlog, Vec::new());
+        let Initialize { nid, conn_id, chain } = msg;
 
-        for msg in backlog {
-            chain.do_send(UpdateNode { nid, msg, raw: None });
+        if let Some(backlog) = self.backlogs.remove(&conn_id) {
+            for msg in backlog {
+                chain.do_send(UpdateNode { nid, msg, raw: None });
+            }
         }
 
         self.nid = nid;
-        self.chain = Some(chain.clone());
+        self.chains.insert(conn_id, chain.clone());
 
         // Acquire the node's physical location
         if let Some(ip) = self.ip {
diff --git a/backend/src/node/message.rs b/backend/src/node/message.rs
index 80e56b151..c230a118b 100644
--- a/backend/src/node/message.rs
+++ b/backend/src/node/message.rs
@@ -3,13 +3,14 @@ use chrono::{DateTime, Utc};
 use serde::Deserialize;
 use serde::de::IgnoredAny;
 use crate::node::NodeDetails;
-use crate::types::{Block, BlockNumber, BlockHash};
+use crate::types::{Block, BlockNumber, BlockHash, ConnId};
 
 #[derive(Deserialize, Debug, Message)]
 #[rtype(result = "()")]
 pub struct NodeMessage {
     pub level: Level,
     pub ts: DateTime<Utc>,
+    pub id: Option<ConnId>,
     #[serde(flatten)]
     pub details: Details,
 }
diff --git a/backend/src/types.rs b/backend/src/types.rs
index 585cf9887..b7840ff11 100644
--- a/backend/src/types.rs
+++ b/backend/src/types.rs
@@ -4,6 +4,7 @@ use serde::Deserialize;
 use crate::util::{MeanList, now};
 
 pub type NodeId = usize;
+pub type ConnId = u64;
 pub type BlockNumber = u64;
 pub type Timestamp = u64;
 pub type Address = Box<str>;

From 1baff2809b10c030288bd2062ff7469caac0035f Mon Sep 17 00:00:00 2001
From: Maciej Hirsz <hello@maciej.codes>
Date: Wed, 4 Nov 2020 16:01:49 +0100
Subject: [PATCH 3/5] Better multiplexing

---
 backend/src/node/connector.rs | 76 ++++++++++++++++++++++-------------
 1 file changed, 48 insertions(+), 28 deletions(-)

diff --git a/backend/src/node/connector.rs b/backend/src/node/connector.rs
index 68859fd12..2c75abcf2 100644
--- a/backend/src/node/connector.rs
+++ b/backend/src/node/connector.rs
@@ -22,17 +22,12 @@ const CLIENT_TIMEOUT: Duration = Duration::from_secs(60);
 const CONT_BUF_LIMIT: usize = 10 * 1024 * 1024;
 
 pub struct NodeConnector {
-    /// Id of the node this connector is responsible for handling
-    nid: NodeId,
+    /// Multiplexing connections by id
+    multiplex: BTreeMap<ConnId, NodeMultiplex>,
     /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
     hb: Instant,
     /// Aggregator actor address
     aggregator: Addr<Aggregator>,
-    /// Mapping message connection id to addresses of chains for multiplexing
-    /// a node running multiple parachains
-    chains: BTreeMap<ConnId, Addr<Chain>>,
-    /// Backlog of messages to be sent once we get a recipient handle to the chain
-    backlogs: BTreeMap<ConnId, Vec<NodeMessage>>,
     /// IP address of the node this connector is responsible for
     ip: Option<Ipv4Addr>,
     /// Actix address of location services
@@ -41,6 +36,27 @@ pub struct NodeConnector {
     contbuf: BytesMut,
 }
 
+enum NodeMultiplex {
+    Connected {
+        /// Id of the node this multiplex connector is responsible for handling
+        nid: NodeId,
+        /// Chain address to which this multiplex connector is delegating messages
+        chain: Addr<Chain>,
+    },
+    Waiting {
+        /// Backlog of messages to be sent once we get a recipient handle to the chain
+        backlog: Vec<NodeMessage>,
+    }
+}
+
+impl Default for NodeMultiplex {
+    fn default() -> Self {
+        NodeMultiplex::Waiting {
+            backlog: Vec::new(),
+        }
+    }
+}
+
 impl Actor for NodeConnector {
     type Context = ws::WebsocketContext<Self>;
 
@@ -49,8 +65,10 @@ impl Actor for NodeConnector {
     }
 
     fn stopped(&mut self, _: &mut Self::Context) {
-        for chain in self.chains.values() {
-            chain.do_send(RemoveNode(self.nid));
+        for mx in self.multiplex.values() {
+            if let NodeMultiplex::Connected { chain, nid } = mx {
+                chain.do_send(RemoveNode(*nid));
+            }
         }
     }
 }
@@ -58,12 +76,9 @@ impl Actor for NodeConnector {
 impl NodeConnector {
     pub fn new(aggregator: Addr<Aggregator>, locator: Recipient<LocateRequest>, ip: Option<Ipv4Addr>) -> Self {
         Self {
-            // Garbage id, will be replaced by the Initialize message
-            nid: !0,
+            multiplex: BTreeMap::new(),
             hb: Instant::now(),
             aggregator,
-            chains: BTreeMap::new(),
-            backlogs: BTreeMap::new(),
             ip,
             locator,
             contbuf: BytesMut::new(),
@@ -83,15 +98,18 @@ impl NodeConnector {
     fn handle_message(&mut self, msg: NodeMessage, data: Bytes, ctx: &mut <Self as Actor>::Context) {
         let conn_id = msg.id.unwrap_or(0);
 
-        if let Some(chain) = self.chains.get(&conn_id) {
-            chain.do_send(UpdateNode {
-                nid: self.nid,
-                msg,
-                raw: Some(data)
-            });
+        let backlog = match self.multiplex.entry(conn_id).or_default() {
+            NodeMultiplex::Connected { nid, chain } => {
+                chain.do_send(UpdateNode {
+                    nid: *nid,
+                    msg,
+                    raw: Some(data),
+                });
 
-            return;
-        }
+                return;
+            }
+            NodeMultiplex::Waiting { backlog } => backlog,
+        };
 
         if let Details::SystemConnected(connected) = msg.details {
             let SystemConnected { network_id: _, mut node } = connected;
@@ -106,8 +124,6 @@ impl NodeConnector {
 
             self.aggregator.do_send(AddNode { rec, conn_id, node });
         } else {
-            let backlog = self.backlogs.entry(conn_id).or_default();
-
             if backlog.len() >= 10 {
                 backlog.remove(0);
             }
@@ -152,14 +168,18 @@ impl Handler<Initialize> for NodeConnector {
     fn handle(&mut self, msg: Initialize, _: &mut Self::Context) {
         let Initialize { nid, conn_id, chain } = msg;
 
-        if let Some(backlog) = self.backlogs.remove(&conn_id) {
-            for msg in backlog {
+        let mx = self.multiplex.entry(conn_id).or_default();
+
+        if let NodeMultiplex::Waiting { backlog } = mx {
+            for msg in backlog.drain(..) {
                 chain.do_send(UpdateNode { nid, msg, raw: None });
             }
-        }
 
-        self.nid = nid;
-        self.chains.insert(conn_id, chain.clone());
+            *mx = NodeMultiplex::Connected {
+                nid,
+                chain: chain.clone(),
+            };
+        };
 
         // Acquire the node's physical location
         if let Some(ip) = self.ip {

From 92eae0fa521aa8d7a9f12e2e1e1f76aae759769a Mon Sep 17 00:00:00 2001
From: Maciej Hirsz <hello@maciej.codes>
Date: Wed, 4 Nov 2020 16:04:54 +0100
Subject: [PATCH 4/5] Rename NodeMultiplex -> ConnMultiplex

---
 backend/src/node/connector.rs | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/backend/src/node/connector.rs b/backend/src/node/connector.rs
index 2c75abcf2..790413ef8 100644
--- a/backend/src/node/connector.rs
+++ b/backend/src/node/connector.rs
@@ -23,7 +23,7 @@ const CONT_BUF_LIMIT: usize = 10 * 1024 * 1024;
 
 pub struct NodeConnector {
     /// Multiplexing connections by id
-    multiplex: BTreeMap<ConnId, NodeMultiplex>,
+    multiplex: BTreeMap<ConnId, ConnMultiplex>,
     /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
     hb: Instant,
     /// Aggregator actor address
@@ -36,7 +36,7 @@ pub struct NodeConnector {
     contbuf: BytesMut,
 }
 
-enum NodeMultiplex {
+enum ConnMultiplex {
     Connected {
         /// Id of the node this multiplex connector is responsible for handling
         nid: NodeId,
@@ -49,9 +49,9 @@ enum NodeMultiplex {
     }
 }
 
-impl Default for NodeMultiplex {
+impl Default for ConnMultiplex {
     fn default() -> Self {
-        NodeMultiplex::Waiting {
+        ConnMultiplex::Waiting {
             backlog: Vec::new(),
         }
     }
@@ -66,7 +66,7 @@ impl Actor for NodeConnector {
 
     fn stopped(&mut self, _: &mut Self::Context) {
         for mx in self.multiplex.values() {
-            if let NodeMultiplex::Connected { chain, nid } = mx {
+            if let ConnMultiplex::Connected { chain, nid } = mx {
                 chain.do_send(RemoveNode(*nid));
             }
         }
@@ -99,7 +99,7 @@ impl NodeConnector {
         let conn_id = msg.id.unwrap_or(0);
 
         let backlog = match self.multiplex.entry(conn_id).or_default() {
-            NodeMultiplex::Connected { nid, chain } => {
+            ConnMultiplex::Connected { nid, chain } => {
                 chain.do_send(UpdateNode {
                     nid: *nid,
                     msg,
@@ -108,7 +108,7 @@ impl NodeConnector {
 
                 return;
             }
-            NodeMultiplex::Waiting { backlog } => backlog,
+            ConnMultiplex::Waiting { backlog } => backlog,
         };
 
         if let Details::SystemConnected(connected) = msg.details {
@@ -170,12 +170,12 @@ impl Handler<Initialize> for NodeConnector {
 
         let mx = self.multiplex.entry(conn_id).or_default();
 
-        if let NodeMultiplex::Waiting { backlog } = mx {
+        if let ConnMultiplex::Waiting { backlog } = mx {
             for msg in backlog.drain(..) {
                 chain.do_send(UpdateNode { nid, msg, raw: None });
             }
 
-            *mx = NodeMultiplex::Connected {
+            *mx = ConnMultiplex::Connected {
                 nid,
                 chain: chain.clone(),
             };

From 54d92e94b610974e68b53def78724fd5ac06bc33 Mon Sep 17 00:00:00 2001
From: Maciej Hirsz <hello@maciej.codes>
Date: Wed, 4 Nov 2020 16:18:35 +0100
Subject: [PATCH 5/5] Cleaner `handle_message`

---
 backend/src/node/connector.rs | 44 +++++++++++++++++------------------
 1 file changed, 21 insertions(+), 23 deletions(-)

diff --git a/backend/src/node/connector.rs b/backend/src/node/connector.rs
index 790413ef8..cf999a26c 100644
--- a/backend/src/node/connector.rs
+++ b/backend/src/node/connector.rs
@@ -98,37 +98,35 @@ impl NodeConnector {
     fn handle_message(&mut self, msg: NodeMessage, data: Bytes, ctx: &mut <Self as Actor>::Context) {
         let conn_id = msg.id.unwrap_or(0);
 
-        let backlog = match self.multiplex.entry(conn_id).or_default() {
+        match self.multiplex.entry(conn_id).or_default() {
             ConnMultiplex::Connected { nid, chain } => {
                 chain.do_send(UpdateNode {
                     nid: *nid,
                     msg,
                     raw: Some(data),
                 });
-
-                return;
-            }
-            ConnMultiplex::Waiting { backlog } => backlog,
-        };
-
-        if let Details::SystemConnected(connected) = msg.details {
-            let SystemConnected { network_id: _, mut node } = connected;
-            let rec = ctx.address().recipient();
-
-            // FIXME: Use genesis hash instead of names to avoid this mess
-            match &*node.chain {
-                "Kusama CC3" => node.chain = "Kusama".into(),
-                "Polkadot CC1" => node.chain = "Polkadot".into(),
-                _ => (),
             }
-
-            self.aggregator.do_send(AddNode { rec, conn_id, node });
-        } else {
-            if backlog.len() >= 10 {
-                backlog.remove(0);
+            ConnMultiplex::Waiting { backlog } => {
+                if let Details::SystemConnected(connected) = msg.details {
+                    let SystemConnected { network_id: _, mut node } = connected;
+                    let rec = ctx.address().recipient();
+
+                    // FIXME: Use genesis hash instead of names to avoid this mess
+                    match &*node.chain {
+                        "Kusama CC3" => node.chain = "Kusama".into(),
+                        "Polkadot CC1" => node.chain = "Polkadot".into(),
+                        _ => (),
+                    }
+
+                    self.aggregator.do_send(AddNode { rec, conn_id, node });
+                } else {
+                    if backlog.len() >= 10 {
+                        backlog.remove(0);
+                    }
+
+                    backlog.push(msg);
+                }
             }
-
-            backlog.push(msg);
         }
     }