connectd: limit incoming traffic to 1MB per second.

diff --git a/connectd/connectd.c b/connectd/connectd.c
index 4689ff74..ad3f11ec 100644
--- a/connectd/connectd.c
+++ b/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 */
diff --git a/connectd/connectd.h b/connectd/connectd.h
index ea012f6a..102d7050 100644
--- a/connectd/connectd.h
+++ b/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;
 
diff --git a/connectd/multiplex.c b/connectd/multiplex.c
index 39404251..c219ba8d 100644
--- a/connectd/multiplex.c
+++ b/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
diff --git a/tests/test_askrene.py b/tests/test_askrene.py
index 513e9ec7..43ebf308 100644
--- a/tests/test_askrene.py
+++ b/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}])
 
diff --git a/tests/test_gossip.py b/tests/test_gossip.py
index 30f94174..c8c838e4 100644
--- a/tests/test_gossip.py
+++ b/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.
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 6815fb08..84b53f72 100644
--- a/tests/test_plugin.py
+++ b/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:]:
diff --git a/tests/test_xpay.py b/tests/test_xpay.py
index c5ad9160..9dd07c32 100644
--- a/tests/test_xpay.py
+++ b/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,
                                             },