From c6648c3ca448671a60f486b0cfc4301eb658aa50 Mon Sep 17 00:00:00 2001 From: Matt Morehouse Date: Sat, 1 Aug 2026 20:52:36 -0500 Subject: [PATCH] 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 remove the subd and 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. --- connectd/connectd.c | 26 ++++++++++ connectd/connectd.h | 4 ++ connectd/connectd_wire.csv | 9 ++++ connectd/multiplex.c | 46 ++++++++++++++++- connectd/multiplex.h | 4 ++ lightningd/connect_control.c | 1 + lightningd/peer_control.c | 33 +++++++++--- lightningd/test/run-invoice-select-inchan.c | 5 +- tests/test_connection.py | 56 +++++++++++++++++++++ wallet/test/run-wallet.c | 5 +- 10 files changed, 179 insertions(+), 10 deletions(-) diff --git a/connectd/connectd.c b/connectd/connectd.c index cb95ef9b0cac..1838ba99d4e6 100644 --- a/connectd/connectd.c +++ b/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 */ diff --git a/connectd/connectd.h b/connectd/connectd.h index 11665cc2440d..889066ec7c96 100644 --- a/connectd/connectd.h +++ b/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; diff --git a/connectd/connectd_wire.csv b/connectd/connectd_wire.csv index c1361a597556..ece5294cf56f 100644 --- a/connectd/connectd_wire.csv +++ b/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, diff --git a/connectd/multiplex.c b/connectd/multiplex.c index 1051645f8cec..ce4be97b2251 100644 --- a/connectd/multiplex.c +++ b/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)))); diff --git a/connectd/multiplex.h b/connectd/multiplex.h index f42f7cb76ea2..b3662b79dab4 100644 --- a/connectd/multiplex.h +++ b/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); diff --git a/lightningd/connect_control.c b/lightningd/connect_control.c index 9c8afa9676a9..f6b5ff75e483 100644 --- a/lightningd/connect_control.c +++ b/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: diff --git a/lightningd/peer_control.c b/lightningd/peer_control.c index bfee7c084e6e..81c7403dfcce 100644 --- a/lightningd/peer_control.c +++ b/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, diff --git a/lightningd/test/run-invoice-select-inchan.c b/lightningd/test/run-invoice-select-inchan.c index 0b42cb1c7178..14715b6e3b16 100644 --- a/lightningd/test/run-invoice-select-inchan.c +++ b/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(); } diff --git a/tests/test_connection.py b/tests/test_connection.py index 6d5d736862bd..186af6ae9085 100644 --- a/tests/test_connection.py +++ b/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') diff --git a/wallet/test/run-wallet.c b/wallet/test/run-wallet.c index c44307710943..01e9e3e0ba42 100644 --- a/wallet/test/run-wallet.c +++ b/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(); }