connectd: limit incoming traffic to 1MB per second.

Decryption is pretty efficient, but incoming traffic can bog down
connectd, especially on smaller nodes, so simply limit it to 1MB per
second.

This triggers in various tests, which is good: shows that it's working,
and that we continue to (slowly!) process traffic.

Changelog-Fixed: connectd: throttle incoming peers to give fairer peer handling under stress.
Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

Author: rustyrussell

connectd/connectd.c

@@ -128,6 +128,10 @@ static struct peer *new_peer(struct daemon *daemon,
 	peer->cs = *cs;
 	peer->subds = tal_arr(peer, struct subd *, 0);
 	peer->peer_in = NULL;
+	peer->bytes_rcvd_this_second = 0;
+	peer->bytes_rcvd_start_time = time_mono();
+	peer->recv_timer = NULL;
+	peer->throttle_warned = false;
 	membuf_init(&peer->encrypted_peer_out,
 		    tal_arr(peer, u8, 0), 0,
 		    membuf_tal_resize);
@@ -1751,8 +1755,10 @@ static void connect_init(struct daemon *daemon, const u8 *msg)
 	}
 
 	/* 500 bytes per second, not 1M per second */
-	if (dev_throttle_gossip)
+	if (dev_throttle_gossip) {
 		daemon->gossip_stream_limit = 500;
+		daemon->incoming_stream_limit = 500;
+	}
 
 	if (dev_limit_connections_inflight)
 		daemon->max_connect_in_flight = 1;
@@ -2565,6 +2571,7 @@ int main(int argc, char *argv[])
 	daemon->dev_keep_nagle = false;
 	/* We generally allow 1MB per second per peer, except for dev testing */
 	daemon->gossip_stream_limit = 1000000;
+	daemon->incoming_stream_limit = 1000000;
 	daemon->scid_htable = new_htable(daemon, scid_htable);
 
 	/* stdin == control */

connectd/connectd.h

@@ -83,6 +83,15 @@ struct peer {
 	/* Input buffer. */
 	u8 *peer_in;
 
+	/* Bytes received in the last second. */
+	size_t bytes_rcvd_this_second;
+	/* When that second starts */
+	struct timemono bytes_rcvd_start_time;
+	/* Timer when we're throttling input */
+	struct oneshot *recv_timer;
+	/* Only send message once if peer gets throttled */
+	bool throttle_warned;
+
 	/* Output buffer. */
 	struct msg_queue *peer_outq;
 
@@ -309,9 +318,12 @@ struct daemon {
 	/* Allow localhost to be considered "public", only with --developer */
 	bool dev_allow_localhost;
 
-	/* How much to gossip allow a peer every 60 seconds (bytes) */
+	/* How much to gossip allow a peer every second (bytes) */
 	size_t gossip_stream_limit;
 
+	/* How much incomign traffic do we allow per peer every second (bytes) */
+	size_t incoming_stream_limit;
+
 	/* We support use of a SOCKS5 proxy (e.g. Tor) */
 	struct addrinfo *proxyaddr;
 

connectd/multiplex.c

@@ -1395,6 +1395,8 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
 				 tal_bytelen(peer->peer_in));
                return io_close(peer_conn);
        }
+
+       peer->bytes_rcvd_this_second += tal_bytelen(peer->peer_in);
        tal_free(peer->peer_in);
 
        type = fromwire_peektype(decrypted);
@@ -1511,16 +1513,52 @@ static struct io_plan *read_body_from_peer(struct io_conn *peer_conn,
        if (!cryptomsg_decrypt_header(&peer->cs, peer->peer_in, &len))
                return io_close(peer_conn);
 
+       peer->bytes_rcvd_this_second += tal_bytelen(peer->peer_in);
+
        tal_resize(&peer->peer_in, (u32)len + CRYPTOMSG_BODY_OVERHEAD);
        return io_read(peer_conn, peer->peer_in, tal_count(peer->peer_in),
 		      read_body_from_peer_done, peer);
 }
 
+static void recv_throttle_timeout(struct peer *peer)
+{
+	peer->recv_timer = tal_free(peer->recv_timer);
+	io_wake(&peer->peer_in);
+}
+
 static struct io_plan *read_hdr_from_peer(struct io_conn *peer_conn,
 					  struct peer *peer)
 {
+	struct timemono now = time_mono();
 	assert(peer->to_peer == peer_conn);
 
+	/* If it's been over a second, make a fresh start. */
+	if (time_to_sec(timemono_between(now, peer->bytes_rcvd_start_time)) > 0) {
+		peer->bytes_rcvd_start_time = now;
+		peer->bytes_rcvd_this_second = 0;
+	}
+
+	/* You sent too much this second? */
+	if (peer->bytes_rcvd_this_second > peer->daemon->incoming_stream_limit) {
+		status_unusual_once(&peer->throttle_warned,
+				    CI_UNEXPECTED
+				    "Throttling incoming peer %s:"
+				    " too much traffic",
+				    fmt_node_id(tmpctx, &peer->id));
+
+		/* Set timer for next second (if not already) */
+		if (!peer->recv_timer) {
+			peer->recv_timer = new_abstimer(&peer->daemon->timers,
+							peer,
+							timemono_add(peer->bytes_rcvd_start_time,
+								     time_from_sec(1)),
+							recv_throttle_timeout,
+							peer);
+		}
+		return io_wait(peer_conn, &peer->peer_in,
+			       read_hdr_from_peer, peer);
+	}
+
 	/* BOLT #8:
 	 *
 	 * ### Receiving and Decrypting Messages

tests/test_askrene.py

@@ -1463,6 +1463,7 @@ def test_real_data(node_factory, bitcoind, executor):
                                      opts=[{'gossip_store_file': outfile.name,
                                             'allow_warning': True,
                                             'dev-throttle-gossip': None,
+                                            'broken_log': 'Throttling incoming peer',
                                             # This can be slow!
                                             'askrene-timeout': TIMEOUT},
                                            {'allow_warning': True}])
@@ -1573,6 +1574,7 @@ def test_real_biases(node_factory, bitcoind, executor):
     l1, l2 = node_factory.line_graph(2, fundamount=AMOUNT,
                                      opts=[{'gossip_store_file': outfile.name,
                                             'allow_warning': True,
+                                            'broken_log': 'Throttling incoming peer',
                                             'dev-throttle-gossip': None},
                                            {'allow_warning': True}])
 
@@ -1685,6 +1687,7 @@ def test_askrene_fake_channeld(node_factory, bitcoind):
                                      opts=[{'gossip_store_file': outfile.name,
                                             'subdaemon': 'channeld:../tests/plugins/channeld_fakenet',
                                             'allow_warning': True,
+                                            'broken_log': 'Throttling incoming peer',
                                             'dev-throttle-gossip': None},
                                            {'allow_bad_gossip': True}])
 

tests/test_gossip.py

@@ -2164,7 +2164,9 @@ def test_dump_own_gossip(node_factory):
 def test_gossip_throttle(node_factory, bitcoind, chainparams):
     """Make some gossip, test it gets throttled"""
     l1, l2, l3, l4 = node_factory.line_graph(4, wait_for_announce=True,
-                                             opts=[{}, {}, {}, {'dev-throttle-gossip': None}])
+                                             opts=[{}, {}, {},
+                                                   {'broken_log': 'Throttling incoming peer',
+                                                    'dev-throttle-gossip': None}])
 
     # We expect: self-advertizement (3 messages for l1 and l4) plus
     # 4 node announcements, 3 channel announcements and 6 channel updates.

tests/test_plugin.py

@@ -3160,7 +3160,7 @@ def test_commando(node_factory, executor):
 @pytest.mark.slow_test
 def test_commando_stress(node_factory, executor):
     """Stress test to slam commando with many large queries"""
-    nodes = node_factory.get_nodes(5)
+    nodes = node_factory.get_nodes(5, opts=[{'broken_log': 'Throttling incoming peer'}, {}, {}, {}, {}])
 
     rune = nodes[0].rpc.createrune()['rune']
     for n in nodes[1:]:

tests/test_xpay.py

@@ -241,6 +241,7 @@ def test_xpay_fake_channeld(node_factory, bitcoind, chainparams, slow_mode):
                                             'allow_warning': True,
                                             'dev-throttle-gossip': None,
                                             'log-level': 'info',
+                                            'broken_log': 'Throttling incoming peer',
                                             # xpay gets upset if it's aging when we remove cln-askrene!
                                             'dev-xpay-no-age': None,
                                             },