Mute denied nodes

backend/src/aggregator.rs

@@ -1,8 +1,9 @@
 use std::collections::{HashMap, HashSet};
 use actix::prelude::*;
+use actix_web_actors::ws::{CloseReason, CloseCode};
 use lazy_static::lazy_static;
 
-use crate::node::connector::Initialize;
+use crate::node::connector::{Initialize, Mute};
 use crate::feed::connector::{FeedConnector, Connected, FeedId};
 use crate::util::DenseMap;
 use crate::feed::{self, FeedMessageSerializer};
@@ -30,7 +31,7 @@ pub struct ChainEntry {
 }
 
 lazy_static! {
-    /// Labels of chains we consider "first party". These chains are allowed any
+    /// Labels of chains we consider "first party". These chains allow any
     /// number of nodes to connect.
     static ref FIRST_PARTY_NETWORKS: HashSet<&'static str> = {
         let mut set = HashSet::new();
@@ -42,7 +43,7 @@ lazy_static! {
     };
 }
 /// Max number of nodes allowed to connect to the telemetry server.
-const THIRD_PARTY_NETWORKS_MAX_NODES: usize = 500;
+const THIRD_PARTY_NETWORKS_MAX_NODES: usize = 2;
 
 impl Aggregator {
     pub fn new(denylist: HashSet<String>) -> Self {
@@ -131,6 +132,8 @@ pub struct AddNode {
     pub conn_id: ConnId,
     /// Recipient for the initialization message
     pub rec: Recipient<Initialize>,
+    /// Recipient for the mute message
+    pub mute: Recipient<Mute>,
 }
 
 /// Message sent from the Chain to the Aggregator when the Chain loses all nodes
@@ -196,10 +199,13 @@ impl Handler<AddNode> for Aggregator {
 
     fn handle(&mut self, msg: AddNode, ctx: &mut Self::Context) {
         if self.denylist.contains(&*msg.node.chain) {
-            log::debug!(target: "Aggregator::AddNode", "'{}' is on the denylist.", msg.node.chain);
+            log::warn!(target: "Aggregator::AddNode", "'{}' is on the denylist.", msg.node.chain);
+            let AddNode { mute, .. } = msg;
+            let reason = CloseReason{ code: CloseCode::Abnormal, description: Some("Denied".into()) };
+            let _ = mute.do_send(Mute { reason });
             return;
         }
-        let AddNode { node, conn_id, rec } = msg;
+        let AddNode { node, conn_id, rec, mute } = msg;
         log::trace!(target: "Aggregator::AddNode", "New node connected. Chain '{}'", node.chain);
 
         let cid = self.lazy_chain(&node.chain, ctx);
@@ -212,6 +218,8 @@ impl Handler<AddNode> for Aggregator {
             });
         } else {
             log::warn!(target: "Aggregator::AddNode", "Chain {} is over quota ({})", chain.label, chain.max_nodes);
+            let reason = CloseReason{ code: CloseCode::Again, description: Some("Overquota".into()) };
+            let _ = mute.do_send(Mute { reason });
         }
     }
 }

backend/src/chain.rs

@@ -284,7 +284,7 @@ impl Chain {
         if node.update_block(*block) {
             if block.height > self.best.height {
                 self.best = *block;
-                log::info!(
+                log::debug!(
                     "[{}] [nodes={}/feeds={}] new best block={}/{:?}",
                     self.label.0,
                     nodes_len,

backend/src/node/connector.rs

@@ -5,7 +5,7 @@ use std::mem;
 
 use bytes::{Bytes, BytesMut};
 use actix::prelude::*;
-use actix_web_actors::ws;
+use actix_web_actors::ws::{self, CloseReason};
 use actix_http::ws::Item;
 use crate::aggregator::{Aggregator, AddNode};
 use crate::chain::{Chain, UpdateNode, RemoveNode};
@@ -24,7 +24,7 @@ const CONT_BUF_LIMIT: usize = 10 * 1024 * 1024;
 pub struct NodeConnector {
     /// Multiplexing connections by id
     multiplex: BTreeMap<ConnId, ConnMultiplex>,
-    /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
+    /// Client must send ping at least once every 60 seconds (CLIENT_TIMEOUT),
     hb: Instant,
     /// Aggregator actor address
     aggregator: Addr<Aggregator>,
@@ -109,8 +109,6 @@ impl NodeConnector {
             ConnMultiplex::Waiting { backlog } => {
                 if let Payload::SystemConnected(connected) = msg.payload() {
                     let mut node = connected.node.clone();
-                    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(),
@@ -123,7 +121,10 @@ impl NodeConnector {
                         _ => ()
                     }
 
-                    self.aggregator.do_send(AddNode { node, conn_id, rec });
+                    let rec = ctx.address().recipient();
+                    let mute = ctx.address().recipient();
+
+                    self.aggregator.do_send(AddNode { node, conn_id, rec, mute });
                 } else {
                     if backlog.len() >= 10 {
                         backlog.remove(0);
@@ -157,6 +158,23 @@ impl NodeConnector {
     }
 }
 
+#[derive(Message)]
+#[rtype(result = "()")]
+pub struct Mute {
+    pub reason: CloseReason,
+}
+
+impl Handler<Mute> for NodeConnector {
+    type Result = ();
+    fn handle(&mut self, msg: Mute, ctx: &mut Self::Context) {
+        let Mute { reason } = msg;
+        log::debug!(target: "NodeConnector::Mute", "Muting a node. Reason: {:?}", reason.description);
+
+        ctx.close(Some(reason));
+        ctx.stop();
+    }
+}
+
 #[derive(Message)]
 #[rtype(result = "()")]
 pub struct Initialize {
@@ -203,7 +221,8 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for NodeConnector {
             Ok(ws::Message::Pong(_)) => return,
             Ok(ws::Message::Text(text)) => text.into_bytes(),
             Ok(ws::Message::Binary(data)) => data,
-            Ok(ws::Message::Close(_)) => {
+            Ok(ws::Message::Close(reason)) => {
+                ctx.close(reason);
                 ctx.stop();
                 return;
             }