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(); }