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

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;
+
+	discard_pending_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;
 
@@ -1383,6 +1388,41 @@ static void destroy_connected_subd(struct subd *subd)
 	}
 }
 
+/* Lightningd answered the connectd_peer_spoke we sent for spoke_id with "no
+ * subd for you".  Free the subd we optimistically created and start reading
+ * from the peer again.
+ *
+ * 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 discard_pending_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));
+		tal_arr_remove(&peer->subds, i);
+		tal_free(subd);
+		io_wake(&peer->peer_in);
+		return;
+	}
+}
+
 static struct subd *new_subd(struct peer *peer,
 			     const struct channel_id *channel_id)
 {
@@ -1396,6 +1436,7 @@ 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++;
 
 	/* Connect it to the peer */
 	tal_arr_expand(&peer->subds, subd);
@@ -1506,10 +1547,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))));

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 this spoke_id.  Drop the one we
+ * created, and start reading from the peer again. */
+void discard_pending_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,14 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg)
 				/* FIXME: Send informative error? */
 				close(other_fd);
 			}
-			return;
+
+			/* We start subds for a new connection once the
+			 * peer_connected hooks return, so if they haven't
+			 * yet, one is on its way and connectd should continue
+			 * to wait. */
+			if (peer->connected != PEER_CONNECTED)
+				return;
+			goto no_subd;
 		}
 
 		/* Send generic error. */
@@ -2142,7 +2150,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 +2170,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 +2210,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')

wallet/test/run-wallet.c

@@ -320,7 +320,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)
@@ -705,6 +705,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(); }