connectd: don't stall peer reads when lightningd attaches no subd

When connectd reads a peer message whose channel_id has no attached
subd, it creates a subd with conn == NULL and sends connectd_peer_spoke
to lightningd, asking it to start up a subdaemon and respond with
connectd_peer_connect_subd and the file descriptor that should be
assigned to conn.  While connectd is waiting for lightningd's response,
it stops reading *all* messages from the peer.

There were six cases in lightningd's handle_peer_spoke where lightningd
would fail to respond, so when those occurred connectd would end up
waiting indefinitely while never reading or responding to the peer's
messages.  Essentially the connection would become a "zombie" until
either the next gossip flush (if any) or the ping timeout caused the
connection to be dropped.

Add a new connectd_peer_no_subd message for lightningd to send connectd
when it will not be creating a subdaemon, so connectd can then continue
servicing the connection as usual.

Notably the connectd_peer_no_subd message uses a new spoke_id to
uniquely identify the connectd_peer_spoke message that it is responding
to, since multiple connectd_peer_spoke messages may be in flight at a
time and because the channel_id (which is the more natural choice) can
be changed by connectd before lightningd's response is received, thereby
causing the wrong subd to be freed.

Fixes: #9369
Changelog-Fixed: connectd: no longer stop servicing a peer connection for ~80s when a channel fails.

Author: morehouse

connectd/connectd.c

@@ -2016,6 +2016,27 @@ static void peer_disconnect(struct daemon *daemon, const u8 *msg)
 	disconnect_peer(peer);
 }
 
+/* lightningd tells us it's not attaching a subd for a peer_spoke we sent. */
+static void peer_no_subd(struct daemon *daemon, const u8 *msg)
+{
+	struct node_id id;
+	u64 counter, spoke_id;
+	struct peer *peer;
+
+	if (!fromwire_connectd_peer_no_subd(msg, &id, &counter, &spoke_id))
+		master_badmsg(WIRE_CONNECTD_PEER_NO_SUBD, msg);
+
+	peer = peer_htable_get(daemon->peers, &id);
+	if (!peer)
+		return;
+
+	/* If it's reconnected already, that subd is long gone. */
+	if (peer->counter != counter)
+		return;
+
+	resume_read_without_subd(peer, spoke_id);
+}
+
 /* lightningd tells us a peer is no longer "important". */
 static void peer_downgrade(struct daemon *daemon, const u8 *msg)
 {
@@ -2408,6 +2429,10 @@ static struct io_plan *recv_req(struct io_conn *conn,
 		return daemon_conn_read_with_fd(conn, daemon->master,
 						recv_peer_connect_subd, daemon);
 
+	case WIRE_CONNECTD_PEER_NO_SUBD:
+		peer_no_subd(daemon, msg);
+		goto out;
+
 	case WIRE_CONNECTD_START_SHUTDOWN:
 		start_shutdown(daemon, msg);
 		goto out;
@@ -2537,6 +2562,7 @@ int main(int argc, char *argv[])
 	daemon = tal(NULL, struct daemon);
 	daemon->developer = developer;
 	daemon->connection_counter = 1;
+	daemon->spoke_counter = 1;
 	/* htable_new is our helper which allocates a htable, initializes it
 	 * and set up the memleak callback so our memleak code can see objects
 	 * inside it */

connectd/connectd.h

@@ -291,6 +291,10 @@ struct daemon {
 	/* Counter from which we derive connection identifiers. */
 	u64 connection_counter;
 
+	/* Counter from which we derive connectd_peer_spoke identifiers, so
+	 * lightningd's answer can be matched to the subd it is about. */
+	u64 spoke_counter;
+
 	/* Base for timeout timers, and how long to wait for init msg */
 	struct timers timers;
 	u32 timeout_secs;

connectd/connectd_wire.csv

@@ -116,6 +116,8 @@ msgdata,connectd_peer_connect_subd,channel_id,channel_id,
 msgtype,connectd_peer_spoke,2005
 msgdata,connectd_peer_spoke,id,node_id,
 msgdata,connectd_peer_spoke,counter,u64,
+# Identifies the subd we made for this: quote it back in connectd_peer_no_subd.
+msgdata,connectd_peer_spoke,spoke_id,u64,
 msgdata,connectd_peer_spoke,msgtype,u16,
 msgdata,connectd_peer_spoke,channel_id,channel_id,
 # If msgtype == WIRE_ERROR, this is the string.
@@ -130,6 +132,13 @@ msgtype,connectd_disconnect_peer,2016
 msgdata,connectd_disconnect_peer,id,node_id,
 msgdata,connectd_disconnect_peer,counter,u64,
 
+# master -> connectd: no subd is coming for the connectd_peer_spoke you sent,
+# so discard the one you created and resume reading from the peer.
+msgtype,connectd_peer_no_subd,2017
+msgdata,connectd_peer_no_subd,id,node_id,
+msgdata,connectd_peer_no_subd,counter,u64,
+msgdata,connectd_peer_no_subd,spoke_id,u64,
+
 # master -> connectd: give message to peer.
 msgtype,connectd_peer_send_msg,2003
 msgdata,connectd_peer_send_msg,id,node_id,

connectd/multiplex.c

@@ -44,6 +44,11 @@ struct subd {
 	/* The actual connection to talk to it (NULL if it's not connected yet)  */
 	struct io_conn *conn;
 
+	/* Identifies the connectd_peer_spoke we sent for this subd, so
+	 * lightningd's connectd_peer_no_subd names this exact subd rather
+	 * than whatever happens to own channel_id by then. */
+	u64 spoke_id;
+
 	/* Input buffer */
 	u8 *in;
 
@@ -52,8 +57,19 @@ struct subd {
 
 	/* After we've told it to tx_abort, we don't send anything else. */
 	bool rcvd_tx_abort;
+
+	/* Lightningd answered our peer_spoke without attaching a conn.  We
+	 * keep queueing (it may still attach one later, and then it wants
+	 * these messages), but we count them: see MAX_UNDELIVERED_MSGS. */
+	bool no_conn_coming;
+	size_t undelivered;
 };
 
+/* Until a conn is attached to a subd, nothing reads the messages queued to
+ * it, so limit the number of messages queued.  Hang up if the limit is
+ * reached. */
+#define MAX_UNDELIVERED_MSGS 100
+
 /* FIXME: reorder! */
 static bool is_urgent(enum peer_wire type);
 static void destroy_connected_subd(struct subd *subd);
@@ -1383,6 +1399,44 @@ static void destroy_connected_subd(struct subd *subd)
 	}
 }
 
+/* Lightningd answered the connectd_peer_spoke we sent for spoke_id with "no
+ * subd for you".  Start reading from the peer again so that we don't cause
+ * an unnecessary disconnect due to ping timeout.
+ *
+ * We deliberately keep the subd and everything queued on it.  Lightningd may
+ * yet attach a conn via a different path (e.g., connect_activate_subd()),
+ * and we can then deliver the queued messages to the subd.
+ *
+ * We match on spoke_id, not channel_id: by now maybe_update_channelid() may
+ * have re-keyed this subd, or a later message may have created a different
+ * subd for the same channel_id. */
+void resume_read_without_subd(struct peer *peer, u64 spoke_id)
+{
+	/* If we're tearing down, we're not reading from the peer anyway. */
+	if (!peer->to_peer || peer->draining_state != NOT_DRAINING)
+		return;
+
+	for (size_t i = 0; i < tal_count(peer->subds); i++) {
+		struct subd *subd = peer->subds[i];
+
+		if (subd->spoke_id != spoke_id)
+			continue;
+
+		/* It got a conn after all: lightningd had already sent a
+		 * connectd_peer_connect_subd, and that reached us first, so
+		 * write_to_subd() wakes us in the normal way. */
+		if (subd->conn)
+			return;
+
+		status_peer_debug(&peer->id,
+				  "No subd for %s: resuming read",
+				  fmt_channel_id(tmpctx, &subd->channel_id));
+		subd->no_conn_coming = true;
+		io_wake(&peer->peer_in);
+		return;
+	}
+}
+
 static struct subd *new_subd(struct peer *peer,
 			     const struct channel_id *channel_id)
 {
@@ -1396,6 +1450,9 @@ static struct subd *new_subd(struct peer *peer,
 	subd->opener_revocation_basepoint = NULL;
 	subd->conn = NULL;
 	subd->rcvd_tx_abort = false;
+	subd->spoke_id = peer->daemon->spoke_counter++;
+	subd->no_conn_coming = false;
+	subd->undelivered = 0;
 
 	/* Connect it to the peer */
 	tal_arr_expand(&peer->subds, subd);
@@ -1506,10 +1563,13 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
 	       status_peer_debug(&peer->id, "Activating for message %s",
 				 peer_wire_name(t));
 	       subd = new_subd(peer, &channel_id);
-	       /* We tell lightningd to fire up a subdaemon to handle this! */
+	       /* We tell lightningd to fire up a subdaemon to handle this!
+		* It must answer with either connectd_peer_connect_subd or
+		* connectd_peer_no_subd: until it does, we stop reading. */
 	       daemon_conn_send(peer->daemon->master,
 				take(towire_connectd_peer_spoke(NULL, &peer->id,
 								peer->counter,
+								subd->spoke_id,
 								t,
 								&channel_id,
 								is_peer_error(tmpctx, decrypted))));
@@ -1521,6 +1581,17 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
        /* Tell them to write. */
        msg_enqueue(subd->outq, take(decrypted));
 
+	/* Disconnect if too many messages queue up on a subd without a conn
+	 * attached.  Otherwise the queue can grow unbounded. */
+	if (subd->no_conn_coming && ++subd->undelivered > MAX_UNDELIVERED_MSGS) {
+		status_peer_unusual(&peer->id,
+				    CI_UNEXPECTED "%zu messages queued with no"
+				    " subd: hanging up", subd->undelivered);
+		/* This frees subd: don't touch it again. */
+		disconnect_peer(peer);
+		return io_wait(peer_conn, &peer->peer_in, next_read, peer);
+	}
+
        /* Is this a tx_abort?  Ignore from now on, and close after sending! */
        if (type == WIRE_TX_ABORT) {
 	       subd->rcvd_tx_abort = true;
@@ -1611,6 +1682,11 @@ static struct io_plan *subd_conn_init(struct io_conn *subd_conn,
 {
 	subd->conn = subd_conn;
 
+	/* A conn arrived after all, so stop counting against
+	 * MAX_UNDELIVERED_MSGS. */
+	subd->no_conn_coming = false;
+	subd->undelivered = 0;
+
 	/* subd is a child of the conn: free when it closes! */
 	tal_steal(subd->conn, subd);
 	tal_add_destructor(subd, destroy_connected_subd);

connectd/multiplex.h

@@ -37,6 +37,10 @@ void set_custommsgs(struct daemon *daemon, const u8 *msg);
 /* Lightningd wants to talk to you. */
 void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd);
 
+/* Lightningd isn't attaching a subd for the peer_spoke we sent with this
+ * spoke_id, so start reading from the peer again. */
+void resume_read_without_subd(struct peer *peer, u64 spoke_id);
+
 /* Disconnect peer: give outgoing msgs time to drain though. */
 void disconnect_peer(struct peer *peer);
 

lightningd/connect_control.c

@@ -548,6 +548,7 @@ static unsigned connectd_msg(struct subd *connectd, const u8 *msg, const int *fd
 	case WIRE_CONNECTD_DEV_REPORT_FDS:
 	case WIRE_CONNECTD_PEER_SEND_MSG:
 	case WIRE_CONNECTD_PEER_CONNECT_SUBD:
+	case WIRE_CONNECTD_PEER_NO_SUBD:
 	case WIRE_CONNECTD_PING:
 	case WIRE_CONNECTD_SEND_ONIONMSG:
 	case WIRE_CONNECTD_INJECT_ONIONMSG:

lightningd/peer_control.c

@@ -2021,8 +2021,9 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 	int other_fd;
 	struct peer_fd *pfd;
 	char *errmsg;
+	u64 spoke_id;
 
-	if (!fromwire_connectd_peer_spoke(msg, msg, &id, &connectd_counter, &msgtype, &channel_id, &errmsg))
+	if (!fromwire_connectd_peer_spoke(msg, msg, &id, &connectd_counter, &spoke_id, &msgtype, &channel_id, &errmsg))
 		fatal("Connectd gave bad CONNECTD_PEER_SPOKE message %s",
 		      tal_hex(msg, msg));
 
@@ -2068,7 +2069,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 			if (errmsg) {
 				channel_fail_permanent(channel, REASON_REMOTE,
 						       "They sent %s", errmsg);
-				return;
+				goto no_subd;
 			}
 
 			/* If channel is active there are two possibilities:
@@ -2077,7 +2078,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 			 * 2. subd exited */
 			if (channel->owner) {
 				/* We raced... */
-				return;
+				goto no_subd;
 			}
 
 			if (msgtype == WIRE_CHANNEL_REESTABLISH
@@ -2093,7 +2094,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 					      take(towire_connectd_peer_send_msg(NULL, &peer->id,
 										 peer->connectd_counter,
 										 error)));
-				return;
+				goto no_subd;
 			}
 
 			log_debug(channel->log, "channel already active");
@@ -2106,7 +2107,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 				/* FIXME: Send informative error? */
 				close(other_fd);
 			}
-			return;
+			goto no_subd;
 		}
 
 		/* Send generic error. */
@@ -2142,7 +2143,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 			goto tell_connectd;
 		/* FIXME: Send informative error? */
 		close(other_fd);
-		return;
+		goto no_subd;
 
 	case WIRE_OPEN_CHANNEL2:
 		if (!dual_fund) {
@@ -2162,7 +2163,7 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 			goto tell_connectd;
 		/* FIXME: Send informative error? */
 		close(other_fd);
-		return;
+		goto no_subd;
 
 	case WIRE_CHANNEL_REESTABLISH:
 		/* Maybe a previously closed channel? */
@@ -2202,6 +2203,17 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 							peer->connectd_counter)));
 	return;
 
+no_subd:
+	/* connectd made a subd for this message and stopped reading from the
+	 * peer until we answer.  We're not attaching one, so tell it: nothing
+	 * else will ever wake it.  Use our own copies of id/counter, since
+	 * channel_fail_permanent() above may have freed peer and channel. */
+	subd_send_msg(ld->connectd,
+		      take(towire_connectd_peer_no_subd(NULL, &id,
+							connectd_counter,
+							spoke_id)));
+	return;
+
 tell_connectd:
 	subd_send_msg(ld->connectd,
 		      take(towire_connectd_peer_connect_subd(NULL, &id,

lightningd/test/run-invoice-select-inchan.c

@@ -291,7 +291,7 @@ bool fromwire_connectd_peer_disconnected(const void *p UNNEEDED, struct node_id
 bool fromwire_connectd_peer_reconnected(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, struct node_id *id UNNEEDED, u64 *prev_counter UNNEEDED, u64 *counter UNNEEDED, struct wireaddr_internal *addr UNNEEDED, struct wireaddr **remote_addr UNNEEDED, bool *incoming UNNEEDED, u8 **features UNNEEDED, u64 *connected_time_nsec UNNEEDED)
 { fprintf(stderr, "fromwire_connectd_peer_reconnected called!\n"); abort(); }
 /* Generated stub for fromwire_connectd_peer_spoke */
-bool fromwire_connectd_peer_spoke(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, struct node_id *id UNNEEDED, u64 *counter UNNEEDED, u16 *msgtype UNNEEDED, struct channel_id *channel_id UNNEEDED, wirestring **error UNNEEDED)
+bool fromwire_connectd_peer_spoke(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, struct node_id *id UNNEEDED, u64 *counter UNNEEDED, u64 *spoke_id UNNEEDED, u16 *msgtype UNNEEDED, struct channel_id *channel_id UNNEEDED, wirestring **error UNNEEDED)
 { fprintf(stderr, "fromwire_connectd_peer_spoke called!\n"); abort(); }
 /* Generated stub for fromwire_dualopend_dev_memleak_reply */
 bool fromwire_dualopend_dev_memleak_reply(const void *p UNNEEDED, bool *leak UNNEEDED)
@@ -636,6 +636,9 @@ u8 *towire_connectd_disconnect_peer(const tal_t *ctx UNNEEDED, const struct node
 /* Generated stub for towire_connectd_peer_connect_subd */
 u8 *towire_connectd_peer_connect_subd(const tal_t *ctx UNNEEDED, const struct node_id *id UNNEEDED, u64 counter UNNEEDED, const struct channel_id *channel_id UNNEEDED)
 { fprintf(stderr, "towire_connectd_peer_connect_subd called!\n"); abort(); }
+/* Generated stub for towire_connectd_peer_no_subd */
+u8 *towire_connectd_peer_no_subd(const tal_t *ctx UNNEEDED, const struct node_id *id UNNEEDED, u64 counter UNNEEDED, u64 spoke_id UNNEEDED)
+{ fprintf(stderr, "towire_connectd_peer_no_subd called!\n"); abort(); }
 /* Generated stub for towire_connectd_peer_send_msg */
 u8 *towire_connectd_peer_send_msg(const tal_t *ctx UNNEEDED, const struct node_id *id UNNEEDED, u64 counter UNNEEDED, const u8 *msg UNNEEDED)
 { fprintf(stderr, "towire_connectd_peer_send_msg called!\n"); abort(); }

tests/test_connection.py

@@ -3157,6 +3157,62 @@ def test_dataloss_protection(node_factory, bitcoind):
     assert (closetxid, "confirmed") in set([(o['txid'], o['status']) for o in l2.rpc.listfunds()['outputs']])
 
 
+@unittest.skipIf(os.getenv('TEST_DB_PROVIDER', 'sqlite3') != 'sqlite3', "sqlite3-specific DB rollback")
+@pytest.mark.openchannel('v1')
+@pytest.mark.openchannel('v2')
+def test_dataloss_protection_still_reads(node_factory, bitcoind, executor):
+    """Ensure we keep servicing the peer connection after data-loss recovery.
+
+    l2 comes back with an old database, so l1's channeld sends a warning and
+    deliberately exits *without* disconnecting so that l2 can send an error and
+    have l1 force close for them.  By the time l2 sends the error, l1 has no
+    subd for the channel, so connectd creates one, sends connectd_peer_spoke and
+    parks the peer read until lightningd answers.  lightningd's
+    handle_peer_spoke() must fail the channel and respond with
+    connectd_peer_no_subd, so that connectd continues to service the peer
+    connection."""
+    l1 = node_factory.get_node(may_reconnect=True, allow_warning=True,
+                               feerates=(7500, 7500, 7500, 7500),
+                               options={'dev-force-features': -7})
+    l2 = node_factory.get_node(may_reconnect=True,
+                               broken_log='Cannot broadcast our commitment tx: they have a future one',
+                               feerates=(7500, 7500, 7500, 7500),
+                               options={'dev-force-features': -7})
+
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+    l1.fundchannel(l2, 10**6)
+    l2.stop()
+
+    # Save copy of the db, then move on.
+    dbpath = os.path.join(l2.daemon.lightning_dir, TEST_NETWORK, "lightningd.sqlite3")
+    orig_db = Path(dbpath).read_bytes()
+    l2.start()
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+    l1.pay(l2, 200000000)
+
+    # Make sure both sides consider it completely settled.
+    l1.daemon.wait_for_logs(["peer_in WIRE_REVOKE_AND_ACK"] * 2)
+    l2.daemon.wait_for_logs(["peer_in WIRE_REVOKE_AND_ACK"] * 2)
+
+    # Now, move l2 back in time.
+    l2.stop()
+    Path(dbpath).write_bytes(orig_db)
+    l2.start()
+    l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
+
+    # l2 freaks out and sends an error; l1 takes handle_peer_spoke()'s errmsg
+    # exit and never creates the requested subdaemon.
+    l1.daemon.wait_for_log("They sent ERROR.*Awaiting unilateral close")
+
+    # connectd made a subd for that error and is waiting on lightningd's
+    # response.
+    assert l1.daemon.is_in_log('Activating for message WIRE_ERROR')
+
+    # l1 must still be reading from l2. Ensure l1 responds to l2's ping message.
+    fut = executor.submit(l2.rpc.ping, l1.info['id'])
+    fut.result(20)
+
+
 @unittest.skipIf(os.getenv('TEST_DB_PROVIDER', 'sqlite3') != 'sqlite3', "sqlite3-specific DB rollback")
 @pytest.mark.openchannel('v1')
 @pytest.mark.openchannel('v2')