Fix performance issues when peers spam us

bitcoin/script.c

@@ -239,8 +239,7 @@ u8 *bitcoin_scriptsig_redeem(const tal_t *ctx,
 	script_push_bytes(&script, redeemscript,
 			  tal_count(redeemscript));
 
-	if (taken(redeemscript))
-		tal_free(redeemscript);
+	tal_free_if_taken(redeemscript);
 
 	return script;
 }

bitcoin/tx.c

@@ -359,8 +359,7 @@ void bitcoin_tx_input_set_witness(struct bitcoin_tx *tx, int innum,
 	wally_psbt_input_set_final_witness(&tx->psbt->inputs[innum], stack);
 	tal_wally_end(tx->psbt);
 
-	if (taken(witness))
-		tal_free(witness);
+	tal_free_if_taken(witness);
 }
 
 void bitcoin_tx_input_set_script(struct bitcoin_tx *tx, int innum, u8 *script)

common/cryptomsg.c

@@ -4,6 +4,7 @@
 #include <ccan/crypto/hkdf_sha256/hkdf_sha256.h>
 #include <ccan/mem/mem.h>
 #include <common/cryptomsg.h>
+#include <common/utils.h>
 #include <sodium/crypto_aead_chacha20poly1305.h>
 #include <wire/wire_io.h>
 
@@ -232,7 +233,6 @@ u8 *cryptomsg_encrypt_msg(const tal_t *ctx,
 
 	maybe_rotate_key(&cs->sn, &cs->sk, &cs->s_ck);
 
-	if (taken(msg))
-		tal_free(msg);
+	tal_free_if_taken(msg);
 	return out;
 }

common/features.c

@@ -225,8 +225,7 @@ bool feature_set_or(struct feature_set *a,
 		for (size_t j = 0; j < tal_bytelen(b->bits[i])*8; j++) {
 			if (feature_is_set(b->bits[i], j)
 			    && feature_offered(a->bits[i], j)) {
-				if (taken(b))
-					tal_free(b);
+				tal_free_if_taken(b);
 				return false;
 			}
 		}
@@ -239,8 +238,7 @@ bool feature_set_or(struct feature_set *a,
 		}
 	}
 
-	if (taken(b))
-		tal_free(b);
+	tal_free_if_taken(b);
 	return true;
 }
 
@@ -252,8 +250,7 @@ bool feature_set_sub(struct feature_set *a,
 		for (size_t j = 0; j < tal_bytelen(b->bits[i])*8; j++) {
 			if (feature_is_set(b->bits[i], j)
 			    && !feature_offered(a->bits[i], j)) {
-				if (taken(b))
-					tal_free(b);
+				tal_free_if_taken(b);
 				return false;
 			}
 		}
@@ -268,8 +265,7 @@ bool feature_set_sub(struct feature_set *a,
 	}
 
 
-	if (taken(b))
-		tal_free(b);
+	tal_free_if_taken(b);
 	return true;
 }
 
@@ -534,8 +530,7 @@ u8 *featurebits_or(const tal_t *ctx, const u8 *f1 TAKES, const u8 *f2 TAKES)
 		result[l1 - l2 + i] |= f2[i];
 
 	/* Cleanup the featurebits if we were told to do so. */
-	if (taken(f2))
-		tal_free(f2);
+	tal_free_if_taken(f2);
 
 	return result;
 }

common/json_stream.c

@@ -190,8 +190,7 @@ void json_add_primitive(struct json_stream *js,
 			const char *val TAKES)
 {
 	json_add_primitive_fmt(js, fieldname, "%s", val);
-	if (taken(val))
-		tal_free(val);
+	tal_free_if_taken(val);
 }
 
 void json_add_string(struct json_stream *js,
@@ -200,8 +199,7 @@ void json_add_string(struct json_stream *js,
 {
 	if (json_filter_ok(js->filter, fieldname))
 		json_out_addstr(js->jout, fieldname, str);
-	if (taken(str))
-		tal_free(str);
+	tal_free_if_taken(str);
 }
 
 static char *json_member_direct(struct json_stream *js,
@@ -304,8 +302,7 @@ void json_add_stringn(struct json_stream *result, const char *fieldname,
 		      const char *value TAKES, size_t value_len)
 {
 	json_add_str_fmt(result, fieldname, "%.*s", (int)value_len, value);
-	if (taken(value))
-		tal_free(value);
+	tal_free_if_taken(value);
 }
 
 void json_add_bool(struct json_stream *result, const char *fieldname, bool value)
@@ -345,8 +342,7 @@ void json_add_escaped_string(struct json_stream *result, const char *fieldname,
 		memcpy(dest + 1, esc->s, strlen(esc->s));
 		dest[1+strlen(esc->s)] = '"';
 	}
-	if (taken(esc))
-		tal_free(esc);
+	tal_free_if_taken(esc);
 }
 
 void json_add_timeabs(struct json_stream *result, const char *fieldname,
@@ -607,8 +603,7 @@ void json_add_psbt(struct json_stream *stream,
 	const char *psbt_b64;
 	psbt_b64 = fmt_wally_psbt(NULL, psbt);
 	json_add_string(stream, fieldname, take(psbt_b64));
-	if (taken(psbt))
-		tal_free(psbt);
+	tal_free_if_taken(psbt);
 }
 
 void json_add_amount_msat(struct json_stream *result,

common/read_peer_msg.c

@@ -18,8 +18,7 @@ bool handle_peer_error_or_warning(struct per_peer_state *pps,
 	/* Simply log incoming warnings */
 	err = is_peer_warning(tmpctx, msg);
 	if (err) {
-		if (taken(msg))
-			tal_free(msg);
+		tal_free_if_taken(msg);
 		status_info("Received %s", err);
 		return true;
 	}

common/sphinx.c

@@ -165,8 +165,7 @@ void sphinx_add_hop(struct sphinx_path *path, const struct pubkey *pubkey,
 	size_t len = tal_bytelen(payload);
 	towire_bigsize(&with_len, len);
 	towire_u8_array(&with_len, payload, len);
-	if (taken(payload))
-		tal_free(payload);
+	tal_free_if_taken(payload);
 
 	if (!sphinx_add_hop_has_length(path, pubkey, take(with_len)))
 		abort();

common/status.h

@@ -50,6 +50,23 @@ void status_io(enum log_level iodir,
 #define status_broken( ...)			\
 	status_fmt(LOG_BROKEN, NULL, __VA_ARGS__)
 
+/* Common case of logging once, based on bool flag */
+#define status_unusual_once(flag, ...)					\
+	do {								\
+		if (*(flag) != true) {					\
+			status_fmt(LOG_UNUSUAL, NULL, __VA_ARGS__);	\
+			(*flag) = true;					\
+		}							\
+	} while(0)
+
+#define status_broken_once(flag, ...)					\
+	do {								\
+		if (*(flag) != true) {					\
+			status_fmt(LOG_BROKEN, NULL, __VA_ARGS__);	\
+			(*flag) = true;					\
+		}							\
+	} while(0)
+
 /* For daemons which handle multiple peers */
 #define status_peer_trace(peer, ...)			\
 	status_fmt(LOG_TRACE, (peer), __VA_ARGS__)

common/utils.c

@@ -171,8 +171,7 @@ char *utf8_str(const tal_t *ctx, const u8 *buf TAKES, size_t buflen)
 	char *ret;
 
 	if (!utf8_check(buf, buflen)) {
-		if (taken(buf))
-			tal_free(buf);
+		tal_free_if_taken(buf);
 		return NULL;
 	}
 

common/utils.h

@@ -97,6 +97,13 @@ void tal_arr_remove_(void *p, size_t elemsize, size_t n);
 		(*(p))[n] = (v);					\
 	} while(0)
 
+/* Helper to free an ptr if it's taken() */
+static inline void tal_free_if_taken(const tal_t *p)
+{
+	if (taken(p))
+		tal_free(p);
+}
+
 /* Check for valid UTF-8 */
 bool utf8_check(const void *buf, size_t buflen);
 

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);
@@ -646,11 +650,9 @@ static struct io_plan *connection_in(struct io_conn *conn,
 	/* Did we fail to accept? */
 	if (!conn) {
 		static bool accept_logged = false;
-		if (!accept_logged) {
-			status_broken("accepting incoming fd failed: %s",
-				      strerror(errno));
-			accept_logged = true;
-		}
+		status_broken_once(&accept_logged,
+				   "accepting incoming fd failed: %s",
+				   strerror(errno));
 		/* Maybe free up some fds by closing something. */
 		close_random_connection(daemon);
 		return NULL;
@@ -1753,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;
@@ -2567,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

@@ -428,8 +428,7 @@ static u8 *process_batch_elements(const tal_t *ctx, struct peer *peer, const u8
 
 	} while(plen);
 
-	if (taken(msg))
-		tal_free(msg);
+	tal_free_if_taken(msg);
 
 	return ret;
 }
@@ -1396,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);
@@ -1512,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
@@ -1627,12 +1664,10 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd)
 	 * (subd will see immediate hangup). */
 	if (fd == -1) {
 		static bool recvfd_logged = false;
-		if (!recvfd_logged) {
-			status_broken("receiving lightningd fd failed for %s: %s",
-				      fmt_node_id(tmpctx, &id),
-				      strerror(errno));
-			recvfd_logged = true;
-		}
+		status_broken_once(&recvfd_logged,
+				   "receiving lightningd fd failed for %s: %s",
+				   fmt_node_id(tmpctx, &id),
+				   strerror(errno));
 		/* Maybe free up some fds by closing something. */
 		close_random_connection(daemon);
 		return;

db/utils.c

@@ -200,8 +200,7 @@ void db_exec_prepared_v2(struct db_stmt *stmt TAKES)
 		db_fatal(stmt->db, "Error executing statement: %s", stmt->error);
 	}
 
-	if (taken(stmt))
-	    tal_free(stmt);
+	tal_free_if_taken(stmt);
 }
 
 size_t db_count_changes(struct db_stmt *stmt)
@@ -221,8 +220,7 @@ u64 db_last_insert_id_v2(struct db_stmt *stmt TAKES)
 	assert(stmt->executed);
 	id = stmt->db->config->last_insert_id_fn(stmt);
 
-	if (taken(stmt))
-		tal_free(stmt);
+	tal_free_if_taken(stmt);
 
 	return id;
 }

devtools/gossmap-compress.c

@@ -304,8 +304,7 @@ static void write_msg_to_gstore(int outfd, const u8 *msg TAKES)
 	    || !write_all(outfd, msg, tal_bytelen(msg))) {
 		err(1, "Writing gossip_store");
 	}
-	if (taken(msg))
-		tal_free(msg);
+	tal_free_if_taken(msg);
 }
 
 /* BOLT #7: