diff --git a/ccan/ccan/io/backend.h b/ccan/ccan/io/backend.h index 714972d15ea4..5934e154cd2c 100644 --- a/ccan/ccan/io/backend.h +++ b/ccan/ccan/io/backend.h @@ -11,6 +11,8 @@ struct fd { /* We could put these in io_plan, but they pack nicely here */ bool exclusive[2]; size_t backend_info; + /* Stable while a poll result still refers to this registration. */ + struct io_fd_registration *registration; }; /* Listeners create connections. */ diff --git a/ccan/ccan/io/io.h b/ccan/ccan/io/io.h index 5d084828b96a..b4eecb3cfd77 100644 --- a/ccan/ccan/io/io.h +++ b/ccan/ccan/io/io.h @@ -816,6 +816,11 @@ struct timemono (*io_time_override(struct timemono (*now)(void)))(void); */ int (*io_poll_override(int (*poll)(struct pollfd *fds, nfds_t nfds, int timeout)))(struct pollfd *, nfds_t, int); +/* Protect daemons which replace connections inside readiness callbacks from + * delivering an old poll result to a newly registered fd object. */ +void io_poll_protect_stale_fds(void); + + /** * io_have_fd - do we own this file descriptor? * @fd: the file descriptor. diff --git a/ccan/ccan/io/poll.c b/ccan/ccan/io/poll.c index c4cbaee85678..a6b3b45941e2 100644 --- a/ccan/ccan/io/poll.c +++ b/ccan/ccan/io/poll.c @@ -11,13 +11,69 @@ #include #include +struct io_fd_registration { + struct fd *fd; + size_t refs; +}; + +struct ready_fd { + /* NULL is the position at which to service an IO_ALWAYS plan. */ + struct io_fd_registration *registration; + short revents; +}; + static size_t num_fds = 0, max_fds = 0, num_waiting = 0, num_always = 0, max_always = 0, num_exclusive = 0; static struct pollfd *pollfds = NULL; static struct fd **fds = NULL; +static bool protect_stale_fds = false; +static struct ready_fd *ready_fds = NULL; +static size_t ready_fds_capacity = 0; static struct io_plan **always = NULL; static struct timemono (*nowfn)(void) = time_mono; static int (*pollfn)(struct pollfd *fds, nfds_t nfds, int timeout) = poll; +static void cleanup_ready_fds(void) +{ + free(ready_fds); + ready_fds = NULL; + ready_fds_capacity = 0; +} + +static struct io_fd_registration *new_registration(struct fd *fd) +{ + struct io_fd_registration *registration = malloc(sizeof(*registration)); + + if (!registration) + return NULL; + registration->fd = fd; + registration->refs = 1; + return registration; +} + +static void registration_put(struct io_fd_registration *registration) +{ + assert(registration->refs != 0); + registration->refs--; + if (registration->refs == 0) + free(registration); +} + +void io_poll_protect_stale_fds(void) +{ + if (protect_stale_fds) + return; + if (atexit(cleanup_ready_fds) != 0) + abort(); + + /* This can be enabled after callers have already registered fds. */ + for (size_t i = 0; i < num_fds; i++) { + fds[i]->registration = new_registration(fds[i]); + if (!fds[i]->registration) + abort(); + } + protect_stale_fds = true; +} + struct timemono (*io_time_override(struct timemono (*now)(void)))(void) { struct timemono (*old)(void) = nowfn; @@ -54,6 +110,12 @@ static bool add_fd(struct fd *fd, short events) return false; max_fds = num; } + if (protect_stale_fds) { + fd->registration = new_registration(fd); + if (!fd->registration) + return false; + } else + fd->registration = NULL; pollfds[num_fds].events = events; /* In case it's idle. */ @@ -75,6 +137,7 @@ static bool add_fd(struct fd *fd, short events) static void del_fd(struct fd *fd) { size_t n = fd->backend_info; + struct io_fd_registration *registration = fd->registration; assert(n != -1); assert(n < num_fds); @@ -98,6 +161,12 @@ static void del_fd(struct fd *fd) } num_fds--; fd->backend_info = -1; + if (registration) { + assert(registration->fd == fd); + registration->fd = NULL; + fd->registration = NULL; + registration_put(registration); + } if (fd->exclusive[IO_IN]) num_exclusive--; @@ -369,6 +438,26 @@ static void restore_pollfds(void) } } +static void append_ready_fd(size_t *num_ready, + struct io_fd_registration *registration, + short revents) +{ + if (*num_ready == ready_fds_capacity) { + size_t capacity = ready_fds_capacity ? ready_fds_capacity * 2 : 8; + struct ready_fd *new_ready_fds; + + new_ready_fds = realloc(ready_fds, + sizeof(*ready_fds) * capacity); + if (!new_ready_fds) + abort(); + ready_fds = new_ready_fds; + ready_fds_capacity = capacity; + } + ready_fds[*num_ready].registration = registration; + ready_fds[*num_ready].revents = revents; + (*num_ready)++; +} + /* This is the main loop. */ void *io_loop(struct timers *timers, struct timer **expired) { @@ -431,6 +520,85 @@ void *io_loop(struct timers *timers, struct timer **expired) break; } + if (protect_stale_fds) { + size_t num_polled = num_fds; + size_t num_ready = 0; + + if (r == 0) { + handle_always(); + continue; + } + + /* Preserve the old fairness order, but retain only ready fds and + * the point where IO_ALWAYS work was interleaved. */ + fairness_counter++; + for (size_t rotation = 0; rotation < num_polled; rotation++) { + struct io_fd_registration *registration; + + i = (rotation + fairness_counter) % num_polled; + if (i == 0) + append_ready_fd(&num_ready, NULL, 0); + + if (!pollfds[i].revents) + continue; + registration = fds[i]->registration; + assert(registration); + registration->refs++; + append_ready_fd(&num_ready, registration, + pollfds[i].revents); + } + for (size_t n = 0; n < num_polled; n++) + pollfds[n].revents = 0; + + for (size_t n = 0; n < num_ready && !io_loop_return; n++) { + socklen_t errno_len = sizeof(errno); + struct io_fd_registration *registration; + struct fd *fd; + struct io_conn *c; + int events; + + registration = ready_fds[n].registration; + if (!registration) { + if (handle_always()) + break; + continue; + } + fd = registration->fd; + if (!fd) + continue; + events = ready_fds[n].revents; + c = (void *)fd; + if (fd->listener) { + struct io_listener *l = (void *)fd; + if (events & POLLIN) { + accept_conn(l); + r--; + } else if (events & (POLLHUP|POLLNVAL|POLLERR)) { + r--; + errno = EBADF; + io_close_listener(l); + } + } else if (events & (POLLIN|POLLOUT)) { + r--; + io_ready(c, events); + } else if (events & (POLLHUP|POLLNVAL|POLLERR)) { + r--; + if (getsockopt(fd->fd, SOL_SOCKET, + SO_ERROR, &errno, + &errno_len) == -1) + errno = EBADF; + io_close(c); + } + } + /* Callbacks can retire registrations, so release snapshot + * references only after dispatch has stopped using the buffer. */ + for (size_t n = 0; n < num_ready; n++) { + if (ready_fds[n].registration) + registration_put(ready_fds[n].registration); + } + continue; + } + fairness_counter++; for (size_t rotation = 0; rotation < num_fds && !io_loop_return; rotation++) { socklen_t errno_len = sizeof(errno); @@ -486,7 +654,6 @@ void *io_loop(struct timers *timers, struct timer **expired) } } } - ret = io_loop_return; io_loop_return = NULL; diff --git a/ccan/ccan/io/test/run-49-stale-fd-readiness.c b/ccan/ccan/io/test/run-49-stale-fd-readiness.c new file mode 100644 index 000000000000..6c5618424d5d --- /dev/null +++ b/ccan/ccan/io/test/run-49-stale-fd-readiness.c @@ -0,0 +1,142 @@ +#include "config.h" +#include +#include +/* Include the C files directly to make each readiness delivery observable. */ +#include +#include +#include + +static int replacement_fd; +static unsigned int poll_calls, actor_events, victim_events, + survivor_events, replacement_events; +static char completed; + +static void check_ready_fds_cleaned(void) +{ + assert(ready_fds == NULL); + assert(ready_fds_capacity == 0); +} + +static int replacement_event(int fd, struct io_plan_arg *arg) +{ + replacement_events++; + return 1; +} + +static struct io_plan *replacement_ready(struct io_conn *conn, void *unused) +{ + io_break(&completed); + return io_close(conn); +} + +static struct io_plan *replacement_init(struct io_conn *conn, void *unused) +{ + io_plan_arg(conn, IO_IN); + return io_set_plan(conn, IO_IN, replacement_event, + replacement_ready, NULL); +} + +static int actor_event(int fd, struct io_plan_arg *arg) +{ + actor_events++; + return 1; +} + +static struct io_plan *actor_ready(struct io_conn *conn, void *unused) +{ + struct io_plan *closed = io_close(conn); + + /* Closing the current fd compacts the table. Adding its replacement + * restores the old table length, but must not make it part of the poll + * result currently being dispatched. */ + if (!io_new_conn(NULL, replacement_fd, replacement_init, NULL)) + abort(); + return closed; +} + +static struct io_plan *actor_init(struct io_conn *conn, void *unused) +{ + io_plan_arg(conn, IO_IN); + return io_set_plan(conn, IO_IN, actor_event, actor_ready, NULL); +} + +static int victim_event(int fd, struct io_plan_arg *arg) +{ + victim_events++; + return 1; +} + +static struct io_plan *victim_init(struct io_conn *conn, void *unused) +{ + io_plan_arg(conn, IO_IN); + return io_set_plan(conn, IO_IN, victim_event, io_close_cb, NULL); +} + +static int survivor_event(int fd, struct io_plan_arg *arg) +{ + survivor_events++; + return 1; +} + +static struct io_plan *survivor_init(struct io_conn *conn, void *unused) +{ + io_plan_arg(conn, IO_IN); + return io_set_plan(conn, IO_IN, survivor_event, io_close_cb, NULL); +} + +static int fake_poll(struct pollfd *fds, nfds_t nfds, int timeout) +{ + poll_calls++; + for (size_t i = 0; i < nfds; i++) + fds[i].revents = 0; + + if (poll_calls == 1) { + /* Fairness rotation handles slot 1 first. That actor replaces + * itself while slots 0 and 2 still have readiness pending. */ + assert(nfds == 3); + for (size_t i = 0; i < nfds; i++) + fds[i].revents = POLLIN; + return 3; + } + + assert(poll_calls == 2); + for (size_t i = 0; i < nfds; i++) { + if (fds[i].fd != replacement_fd) + continue; + fds[i].revents = POLLIN; + return 1; + } + abort(); +} + +int main(void) +{ + int actor_pipe[2], victim_pipe[2], survivor_pipe[2], replacement_pipe[2]; + + assert(pipe(actor_pipe) == 0); + assert(pipe(victim_pipe) == 0); + assert(pipe(survivor_pipe) == 0); + assert(pipe(replacement_pipe) == 0); + replacement_fd = replacement_pipe[0]; + + assert(io_poll_override(fake_poll) == poll); + assert(io_new_conn(NULL, victim_pipe[0], victim_init, NULL)); + assert(io_new_conn(NULL, actor_pipe[0], actor_init, NULL)); + assert(io_new_conn(NULL, survivor_pipe[0], survivor_init, NULL)); + + /* Registered first so this runs after the backend's LIFO cleanup. */ + assert(atexit(check_ready_fds_cleaned) == 0); + io_poll_protect_stale_fds(); + assert(io_loop(NULL, NULL) == &completed); + assert(poll_calls == 2); + assert(actor_events == 1); + assert(victim_events == 1); + assert(survivor_events == 1); + assert(replacement_events == 1); + + close(actor_pipe[1]); + close(victim_pipe[1]); + close(survivor_pipe[1]); + close(replacement_pipe[1]); + return 0; +} diff --git a/common/test/Makefile b/common/test/Makefile index 71f9dc17d981..dd165bef0f45 100644 --- a/common/test/Makefile +++ b/common/test/Makefile @@ -1,7 +1,10 @@ +CCAN_IO_TEST_SRC := ccan/ccan/io/test/run-49-stale-fd-readiness.c COMMON_TEST_SRC := $(wildcard common/test/run-*.c) COMMON_TEST_OBJS := $(COMMON_TEST_SRC:.c=.o) COMMON_TEST_PROGRAMS := $(COMMON_TEST_OBJS:.o=) +CCAN_IO_TEST_OBJS := $(CCAN_IO_TEST_SRC:.c=.o) +CCAN_IO_TEST_PROGRAMS := $(CCAN_IO_TEST_OBJS:.o=) COMMON_TEST_COMMON_OBJS := \ common/autodata.o \ @@ -13,8 +16,8 @@ COMMON_TEST_COMMON_OBJS := \ $(COMMON_TEST_PROGRAMS): $(COMMON_TEST_COMMON_OBJS) $(BITCOIN_OBJS) $(COMMON_TEST_OBJS): $(COMMON_HEADERS) $(WIRE_HEADERS) $(COMMON_SRC) common/test/Makefile -ALL_C_SOURCES += $(COMMON_TEST_SRC) -ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS) +ALL_C_SOURCES += $(COMMON_TEST_SRC) $(CCAN_IO_TEST_SRC) +ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS) $(CCAN_IO_TEST_PROGRAMS) # Make them all depend on common/ files, for simplicity (they directly #include some) $(COMMON_TEST_OBJS): $(COMMON_SRC) @@ -178,4 +181,4 @@ common/test/run-close_tx: \ wire/fromwire.o \ wire/towire.o -check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%) +check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%) $(CCAN_IO_TEST_PROGRAMS:%=unittest/%) diff --git a/connectd/connectd.c b/connectd/connectd.c index cb95ef9b0cac..3caafe66a765 100644 --- a/connectd/connectd.c +++ b/connectd/connectd.c @@ -2354,7 +2354,9 @@ static struct io_plan *recv_peer_connect_subd(struct io_conn *conn, int fd, struct daemon *daemon) { - peer_connect_subd(daemon, msg, fd); + peer_connect_subd(daemon, msg, fd, + fromwire_peektype(msg) + == WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED); return daemon_conn_read_next(conn, daemon->master); } @@ -2404,10 +2406,15 @@ static struct io_plan *recv_req(struct io_conn *conn, goto out; case WIRE_CONNECTD_PEER_CONNECT_SUBD: + case WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED: /* This comes with an fd */ return daemon_conn_read_with_fd(conn, daemon->master, recv_peer_connect_subd, daemon); + case WIRE_CONNECTD_PEER_RESUME_SUBD: + peer_resume_subd(daemon, msg); + goto out; + case WIRE_CONNECTD_START_SHUTDOWN: start_shutdown(daemon, msg); goto out; @@ -2470,6 +2477,7 @@ static struct io_plan *recv_req(struct io_conn *conn, case WIRE_CONNECTD_CUSTOMMSG_IN: case WIRE_CONNECTD_PEER_DISCONNECTED: case WIRE_CONNECTD_PEER_RECONNECTED: + case WIRE_CONNECTD_PEER_CONNECT_SUBD_REPLY: case WIRE_CONNECTD_START_SHUTDOWN_REPLY: case WIRE_CONNECTD_INJECT_ONIONMSG_REPLY: case WIRE_CONNECTD_ONIONMSG_FORWARD_FAIL: @@ -2532,6 +2540,7 @@ int main(int argc, char *argv[]) /* Common subdaemon setup code. */ developer = subdaemon_setup(argc, argv); + io_poll_protect_stale_fds(); /* Allocate and set up our simple top-level structure. */ daemon = tal(NULL, struct daemon); diff --git a/connectd/connectd_wire.csv b/connectd/connectd_wire.csv index c1361a597556..a95e6ed9e90c 100644 --- a/connectd/connectd_wire.csv +++ b/connectd/connectd_wire.csv @@ -112,6 +112,28 @@ msgdata,connectd_peer_connect_subd,id,node_id, msgdata,connectd_peer_connect_subd,counter,u64, msgdata,connectd_peer_connect_subd,channel_id,channel_id, +# Master -> connectd: attach a replacement dualopend and acknowledge only once +# its route is active (+ fd to subd). +msgtype,connectd_peer_connect_subd_tracked,2018 +msgdata,connectd_peer_connect_subd_tracked,id,node_id, +msgdata,connectd_peer_connect_subd_tracked,counter,u64, +msgdata,connectd_peer_connect_subd_tracked,channel_id,channel_id, + +# Connectd -> master: the subd route was accepted or rejected. A restarted +# dualopend does not begin its peer protocol until the route is installed. +msgtype,connectd_peer_connect_subd_reply,2104 +msgdata,connectd_peer_connect_subd_reply,id,node_id, +msgdata,connectd_peer_connect_subd_reply,counter,u64, +msgdata,connectd_peer_connect_subd_reply,channel_id,channel_id, +msgdata,connectd_peer_connect_subd_reply,accepted,bool, + +# Master -> connectd: a replacement subd has inherited the endpoint returned +# after tx_abort. Reactivate that exact route and cancel its abort timer. +msgtype,connectd_peer_resume_subd,2017 +msgdata,connectd_peer_resume_subd,id,node_id, +msgdata,connectd_peer_resume_subd,counter,u64, +msgdata,connectd_peer_resume_subd,channel_id,channel_id, + # Connectd -> master: peer said something interesting msgtype,connectd_peer_spoke,2005 msgdata,connectd_peer_spoke,id,node_id, diff --git a/connectd/multiplex.c b/connectd/multiplex.c index 1051645f8cec..e609e7c85920 100644 --- a/connectd/multiplex.c +++ b/connectd/multiplex.c @@ -43,6 +43,11 @@ struct subd { /* The actual connection to talk to it (NULL if it's not connected yet) */ struct io_conn *conn; + /* A tracked io_new_conn is waiting for subd_conn_init_tracked. */ + bool conn_pending; + /* Identify that request when acknowledging the active connection. */ + u64 pending_counter; + struct channel_id pending_channel_id; /* Input buffer */ u8 *in; @@ -52,6 +57,7 @@ struct subd { /* After we've told it to tx_abort, we don't send anything else. */ bool rcvd_tx_abort; + struct oneshot *abort_close_timer; }; /* FIXME: reorder! */ @@ -95,6 +101,7 @@ static void close_peer_io_timeout(struct peer *peer) static void close_subd_timeout(struct subd *subd) { + subd->abort_close_timer = NULL; status_peer_broken(&subd->peer->id, "Subd did not close, forcing close"); io_close(subd->conn); } @@ -1395,7 +1402,9 @@ static struct subd *new_subd(struct peer *peer, subd->temporary_channel_id = NULL; subd->opener_revocation_basepoint = NULL; subd->conn = NULL; + subd->conn_pending = false; subd->rcvd_tx_abort = false; + subd->abort_close_timer = NULL; /* Connect it to the peer */ tal_arr_expand(&peer->subds, subd); @@ -1525,9 +1534,11 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn, if (type == WIRE_TX_ABORT) { subd->rcvd_tx_abort = true; /* In case it doesn't close by itself */ - notleak(new_reltimer(&peer->daemon->timers, subd, - time_from_sec(5), - close_subd_timeout, subd)); + subd->abort_close_timer + = notleak(new_reltimer(&peer->daemon->timers, + subd, time_from_sec(5), + close_subd_timeout, + subd)); } /* Wait for them to wake us */ @@ -1606,16 +1617,62 @@ static struct io_plan *read_hdr_from_peer(struct io_conn *peer_conn, read_body_from_peer, peer); } -static struct io_plan *subd_conn_init(struct io_conn *subd_conn, - struct subd *subd) +static void send_connect_subd_reply(struct daemon *daemon, + const struct node_id *id, + u64 counter, + const struct channel_id *channel_id, + bool accepted) +{ + daemon_conn_send(daemon->master, + take(towire_connectd_peer_connect_subd_reply( + NULL, id, counter, channel_id, accepted))); +} + +static struct io_plan *subd_conn_ready(struct io_conn *subd_conn, + struct subd *subd) +{ + bool accepted; + + accepted = subd->peer->to_peer != NULL + && subd->peer->counter == subd->pending_counter + && subd->peer->draining_state == NOT_DRAINING; + send_connect_subd_reply(subd->peer->daemon, &subd->peer->id, + subd->pending_counter, + &subd->pending_channel_id, accepted); + if (!accepted) + return io_close(subd_conn); + return read_from_subd(subd_conn, subd); +} + +/* The long-standing attachment path for ordinary subdaemons. Keep this + * independent of the replacement-dualopend acknowledgement barrier. */ +static struct io_plan *subd_conn_init_untracked(struct io_conn *subd_conn, + struct subd *subd) { subd->conn = subd_conn; + tal_steal(subd->conn, subd); + tal_add_destructor(subd, destroy_connected_subd); + return io_duplex(subd_conn, + read_from_subd(subd_conn, subd), + write_to_subd(subd_conn, subd)); +} + +static struct io_plan *subd_conn_init_tracked(struct io_conn *subd_conn, + struct subd *subd) +{ + assert(subd->conn_pending); + subd->conn_pending = false; + subd->conn = subd_conn; /* subd is a child of the conn: free when it closes! */ tal_steal(subd->conn, subd); tal_add_destructor(subd, destroy_connected_subd); + + /* Defer the tracked acknowledgement until ccan/io has installed this + * duplex plan, so lightningd cannot start a replacement owner against a + * half-live route. */ return io_duplex(subd_conn, - read_from_subd(subd_conn, subd), + io_always(subd_conn, subd_conn_ready, subd), write_to_subd(subd_conn, subd)); } @@ -1631,8 +1688,9 @@ static void destroy_peer_conn(struct io_conn *peer_conn, struct peer *peer) /* Wake subds: give them 5 seconds to flush. */ for (size_t i = 0; i < tal_count(peer->subds); i++) { - /* Might not be connected yet (no destructor, simple) */ - if (!peer->subds[i]->conn) { + /* A pending io_new_conn still owns an fd and must unwind normally. */ + if (!peer->subds[i]->conn + && !peer->subds[i]->conn_pending) { tal_arr_remove(&peer->subds, i); i--; continue; @@ -1682,7 +1740,8 @@ struct io_plan *multiplex_peer_setup(struct io_conn *peer_conn, write_to_peer(peer_conn, peer)); } -void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd) +void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd, + bool tracked) { struct node_id id; u64 counter; @@ -1690,7 +1749,14 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd) struct channel_id channel_id; struct subd *subd; - if (!fromwire_connectd_peer_connect_subd(msg, &id, &counter, &channel_id)) + if (tracked) { + if (!fromwire_connectd_peer_connect_subd_tracked(msg, &id, + &counter, + &channel_id)) + master_badmsg(WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED, + msg); + } else if (!fromwire_connectd_peer_connect_subd(msg, &id, &counter, + &channel_id)) master_badmsg(WIRE_CONNECTD_PEER_CONNECT_SUBD, msg); /* If receiving fd failed, fd will be -1. Log and ignore @@ -1703,24 +1769,59 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd) strerror(errno)); /* Maybe free up some fds by closing something. */ close_random_connection(daemon); - return; + goto reply; } /* Races can happen: this might be gone by now (or reconnected!). */ peer = peer_htable_get(daemon->peers, &id); if (!peer || peer->counter != counter) { close(fd); - return; + goto reply; } /* Could be disconnecting now */ if (!peer->to_peer || peer->draining_state != NOT_DRAINING) { close(fd); - return; + goto reply; } /* If peer said something, we created this and queued msg. */ subd = find_subd(peer, &channel_id); + if (!tracked) { + if (!subd) { + subd = new_subd(peer, &channel_id); + release_one_waiting_connection( + peer->daemon, + tal_fmt(tmpctx, "%s given a subd", + fmt_node_id(tmpctx, &id))); + } + if (subd->conn) { + status_peer_debug(&id, + "Already have a subd for channel_id %s: ignoring", + fmt_channel_id(tmpctx, &channel_id)); + close(fd); + return; + } + io_new_conn(peer, fd, subd_conn_init_untracked, subd); + return; + } + + /* lightningd serializes owner installation. Thus an attached endpoint + * here belongs to the retiring owner and this request is its handover. */ + if (subd && subd->conn) { + status_peer_debug(&id, + "Replacing attached subd for channel_id %s", + fmt_channel_id(tmpctx, &channel_id)); + io_close(subd->conn); + subd = NULL; + } else if (subd && subd->conn_pending) { + status_peer_debug(&id, + "Subd attachment already pending for channel_id %s", + fmt_channel_id(tmpctx, &channel_id)); + close(fd); + goto reply; + } + if (!subd) { subd = new_subd(peer, &channel_id); /* Implies lightningd is ready for another peer. */ @@ -1728,20 +1829,61 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd) tal_fmt(tmpctx, "%s given a subd", fmt_node_id(tmpctx, &id))); } + /* subd_conn_ready sends the reply after the duplex route is active. */ + subd->conn_pending = true; + subd->pending_counter = counter; + subd->pending_channel_id = channel_id; + io_new_conn(peer, fd, subd_conn_init_tracked, subd); + return; - /* We only keep one connection per channel_id. If one is already - * attached for this channel_id, drop this fd rather than replacing - * it. */ - if (subd->conn) { - status_peer_debug(&id, - "Already have a subd for channel_id %s: ignoring", - fmt_channel_id(tmpctx, &channel_id)); - close(fd); +reply: + if (!tracked) return; + send_connect_subd_reply(daemon, &id, counter, &channel_id, false); +} + +void peer_resume_subd(struct daemon *daemon, const u8 *msg) +{ + struct node_id id; + u64 counter; + struct channel_id channel_id; + struct peer *peer; + struct subd *subd = NULL; + bool accepted = false; + + if (!fromwire_connectd_peer_resume_subd(msg, &id, &counter, + &channel_id)) + master_badmsg(WIRE_CONNECTD_PEER_RESUME_SUBD, msg); + + peer = peer_htable_get(daemon->peers, &id); + if (!peer || peer->counter != counter || !peer->to_peer + || peer->draining_state != NOT_DRAINING) + goto reply; + + /* find_subd deliberately hides a route once it has delivered tx_abort. + * Resume must find that exact attached route, not create another one. */ + for (size_t i = 0; i < tal_count(peer->subds); i++) { + struct subd *candidate = peer->subds[i]; + + if (!candidate->rcvd_tx_abort || !candidate->conn) + continue; + if (channel_id_eq(&candidate->channel_id, &channel_id) + || (candidate->temporary_channel_id + && channel_id_eq(candidate->temporary_channel_id, + &channel_id))) { + subd = candidate; + break; + } } + if (!subd) + goto reply; + + subd->abort_close_timer = tal_free(subd->abort_close_timer); + subd->rcvd_tx_abort = false; + accepted = true; - /* This sets subd->conn inside subd_conn_init, and reparents subd! */ - io_new_conn(peer, fd, subd_conn_init, subd); +reply: + send_connect_subd_reply(daemon, &id, counter, &channel_id, accepted); } /* Lightningd says to send a ping */ diff --git a/connectd/multiplex.h b/connectd/multiplex.h index f42f7cb76ea2..c59a0392d3d8 100644 --- a/connectd/multiplex.h +++ b/connectd/multiplex.h @@ -35,7 +35,11 @@ void custommsg_completed(struct daemon *daemon, const u8 *msg); 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); +void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd, + bool tracked); + +/* A replacement owner inherited an existing route after tx_abort. */ +void peer_resume_subd(struct daemon *daemon, const u8 *msg); /* Disconnect peer: give outgoing msgs time to drain though. */ void disconnect_peer(struct peer *peer); diff --git a/lightningd/channel.c b/lightningd/channel.c index 692d81980150..e0b594ff3ff0 100644 --- a/lightningd/channel.c +++ b/lightningd/channel.c @@ -22,6 +22,21 @@ void channel_set_owner(struct channel *channel, struct subd *owner) { struct subd *old_owner = channel->owner; channel->owner = owner; + if (owner) { + if (channel->open_attempt) + channel->open_attempt->disconnect_timer + = tal_free(channel->open_attempt->disconnect_timer); + } else if (channel->open_attempt + && channel->open_attempt->cmd + && channel->open_attempt->open_msg) { + /* A command queued to the old owner was not necessarily delivered to + * the peer. Its replacement must replay the latest serialized step + * after channel_reestablish. */ + channel->open_attempt->open_msg_state = OPEN_ATTEMPT_MSG_UNSENT; + } + if (!owner) + channel->dualopend_connectd_fd + = tal_free(channel->dualopend_connectd_fd); if (old_owner) subd_release_channel(old_owner, channel); @@ -252,6 +267,8 @@ struct open_attempt *new_channel_open_attempt(struct channel *channel) oa->cmd = NULL; oa->aborted = false; oa->open_msg = NULL; + oa->open_msg_state = OPEN_ATTEMPT_MSG_UNSENT; + oa->disconnect_timer = NULL; return oa; } @@ -367,6 +384,15 @@ struct channel *new_unsaved_channel(struct peer *peer, channel->our_config.id = 0; channel->open_attempt = NULL; + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel->dualopend_restart_mode = DUALOPEND_RESTART_REESTABLISH; + channel->dualopend_owner_connectd_counter = 0; + channel->dualopend_owner_install_retries = 0; + channel->owner_reinit_msg = NULL; + channel->dualopend_connectd_fd = NULL; + channel->dualopend_restart_peer_fd = NULL; + channel->dualopend_abort_reason = NULL; + channel->dualopend_depth_needs_recheck = false; channel->last_htlc_sigs = NULL; channel->remote_channel_ready = false; @@ -580,6 +606,15 @@ struct channel *new_channel(struct peer *peer, u64 dbid, channel->reestablished = false; channel->error = NULL; channel->open_attempt = NULL; + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel->dualopend_restart_mode = DUALOPEND_RESTART_REESTABLISH; + channel->dualopend_owner_connectd_counter = 0; + channel->dualopend_owner_install_retries = 0; + channel->owner_reinit_msg = NULL; + channel->dualopend_connectd_fd = NULL; + channel->dualopend_restart_peer_fd = NULL; + channel->dualopend_abort_reason = NULL; + channel->dualopend_depth_needs_recheck = false; channel->openchannel_signed_cmd = NULL; if (their_shachain) channel->their_shachain = *their_shachain; @@ -1350,4 +1385,3 @@ const u8 *channel_update_for_error(const tal_t *ctx, return channel_gossip_update_for_error(ctx, channel); } - diff --git a/lightningd/channel.h b/lightningd/channel.h index 8e3a499ad09c..5b5833b08f0e 100644 --- a/lightningd/channel.h +++ b/lightningd/channel.h @@ -15,8 +15,37 @@ #include struct uncommitted_channel; +struct oneshot; +struct peer_fd; struct wally_psbt; +enum open_attempt_msg_state { + OPEN_ATTEMPT_MSG_UNSENT, + OPEN_ATTEMPT_MSG_SENT, +}; + +/* Local lifecycle of the subdaemon which owns a dual-funded open. This is + * deliberately separate from enum channel_state: the latter describes the + * persisted wire protocol, while this describes installing its local owner. */ +enum dualopend_owner_state { + DUALOPEND_OWNER_NONE, + DUALOPEND_OWNER_RESTART_PENDING, + DUALOPEND_OWNER_ROUTE_PENDING, + /* from_abort owners can finish initialization before connectd replies. */ + DUALOPEND_OWNER_READY_PENDING_ROUTE, + DUALOPEND_OWNER_REESTABLISHING, + DUALOPEND_OWNER_READY, + DUALOPEND_OWNER_RETIRING, +}; + +/* How a replacement dualopend joins the protocol on the current transport. + * This survives owner replacement: only a peer transport change resets an + * AFTER_ABORT continuation to a full channel_reestablish. */ +enum dualopend_restart_mode { + DUALOPEND_RESTART_REESTABLISH, + DUALOPEND_RESTART_AFTER_ABORT, +}; + /* FIXME: Define serialization primitive for this? */ struct channel_info { struct channel_config their_config; @@ -119,6 +148,10 @@ struct open_attempt { /* First msg to send to dualopend (to make it create channel) */ const u8 *open_msg; + enum open_attempt_msg_state open_msg_state; + + /* Bounds how long cmd and its PSBT survive a continuous disconnect. */ + struct oneshot *disconnect_timer; }; /* Statistics for a channel */ @@ -151,6 +184,33 @@ struct channel { /* Open attempt */ struct open_attempt *open_attempt; + /* Installation state and startup mode for the dualopend owner. */ + enum dualopend_owner_state dualopend_owner_state; + enum dualopend_restart_mode dualopend_restart_mode; + + /* Transport on which the current dualopend route was installed. This is + * intentionally owner state, not peer state: a replacement TCP connection + * can reach connectd before lightningd handles PEER_RECONNECTED. */ + u64 dualopend_owner_connectd_counter; + /* At most one local owner-install retry is allowed per transport. */ + u8 dualopend_owner_install_retries; + + /* Held while DUALOPEND_OWNER_ROUTE_PENDING. */ + const u8 *owner_reinit_msg; + /* Connectd's socketpair endpoint is held until the replacement dualopend + * confirms that it initialized with the peer endpoint open. */ + struct peer_fd *dualopend_connectd_fd; + /* A tx_abort returns the existing connectd endpoint. Retain it until the + * retiring owner has exited, then give this same route to its replacement. */ + struct peer_fd *dualopend_restart_peer_fd; + /* Set while an in-place tx_abort waits for connectd to reactivate the + * existing route. The aborting RPC completes only after that barrier. */ + char *dualopend_abort_reason; + + /* A funding-depth callback ran before the replacement dualopend was + * ready. Re-evaluate against the current chain when it becomes ready. */ + bool dualopend_depth_needs_recheck; + /* Database ID: 0 == not in db yet */ u64 dbid; @@ -721,6 +781,31 @@ static inline bool channel_state_uncommitted(enum channel_state state) abort(); } +/* A dual-funded open can resume its pending command after reconnection once + * the channel has been saved. */ +static inline bool channel_state_saved_dualopend(enum channel_state state) +{ + switch (state) { + case DUALOPEND_OPEN_COMMIT_READY: + case DUALOPEND_OPEN_COMMITTED: + return true; + case DUALOPEND_OPEN_INIT: + case DUALOPEND_AWAITING_LOCKIN: + case CHANNELD_AWAITING_LOCKIN: + case CHANNELD_NORMAL: + case CHANNELD_AWAITING_SPLICE: + case CLOSINGD_SIGEXCHANGE: + case CHANNELD_SHUTTING_DOWN: + case CLOSINGD_COMPLETE: + case AWAITING_UNILATERAL: + case FUNDING_SPEND_SEEN: + case ONCHAIN: + case CLOSED: + return false; + } + abort(); +} + /* Established enough, that we could reach out to peer to discuss */ static inline bool channel_state_wants_peercomms(enum channel_state state) { diff --git a/lightningd/connect_control.c b/lightningd/connect_control.c index 9c8afa9676a9..0ad69f1c86c1 100644 --- a/lightningd/connect_control.c +++ b/lightningd/connect_control.c @@ -548,6 +548,8 @@ 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_CONNECT_SUBD_TRACKED: + case WIRE_CONNECTD_PEER_RESUME_SUBD: case WIRE_CONNECTD_PING: case WIRE_CONNECTD_SEND_ONIONMSG: case WIRE_CONNECTD_INJECT_ONIONMSG: @@ -566,6 +568,10 @@ static unsigned connectd_msg(struct subd *connectd, const u8 *msg, const int *fd case WIRE_CONNECTD_INJECT_ONIONMSG_REPLY: break; + case WIRE_CONNECTD_PEER_CONNECT_SUBD_REPLY: + dual_open_owner_route_result(connectd->ld, msg); + break; + case WIRE_CONNECTD_PEER_CONNECTED: case WIRE_CONNECTD_PEER_RECONNECTED: handle_peer_connected(connectd->ld, msg); diff --git a/lightningd/dual_open_control.c b/lightningd/dual_open_control.c index 451cf1f94246..2aa738f99715 100644 --- a/lightningd/dual_open_control.c +++ b/lightningd/dual_open_control.c @@ -11,8 +11,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -27,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -69,21 +72,67 @@ void channel_unsaved_close_conn(struct channel *channel, const char *why) delete_channel(channel, false); } -static void channel_saved_err_broken_reconn(struct channel *channel, - const char *fmt, ...) +bool dual_open_attempt_waiting_for_owner(const struct channel *channel) { - va_list ap; - const char *errmsg; + return channel->open_attempt + && channel->open_attempt->cmd + && channel->open_attempt->open_msg; +} - /* We only reconnect to 'saved' channel peers */ - assert(!channel_state_uncommitted(channel->state)); +static void open_attempt_disconnect_timeout(struct channel *channel) +{ + struct open_attempt *oa = channel->open_attempt; + const char *why = "Peer did not reconnect before dual-open timeout"; - va_start(ap, fmt); - errmsg = tal_vfmt(tmpctx, fmt, ap); - va_end(ap); + /* Installing a replacement owner cancels the timer. These guards also + * make expiry harmless if another completion path won the race. */ + if (!oa) + return; + oa->disconnect_timer = NULL; + if (!oa->cmd || channel->owner) + return; + + log_unusual(channel->log, "%s", why); + channel_cleanup_commands(channel, why); - log_broken(channel->log, "%s", errmsg); - channel_disconnect(channel, LOG_INFORM, true, errmsg); + /* No commitment signatures were exchanged in OPEN_COMMIT_READY, so this + * persisted half-open is safe to forget. Later states may have a + * publishable inflight: fail only the RPC and retain the channel there. */ + if (channel->state == DUALOPEND_OPEN_COMMIT_READY) + channel_fail_permanent(channel, REASON_LOCAL, "%s", why); +} + +void dual_open_attempt_peer_disconnected(struct channel *channel) +{ + struct open_attempt *oa = channel->open_attempt; + + if (!channel->owner + && channel->peer->connected != PEER_CONNECTED + && channel->dualopend_owner_state != DUALOPEND_OWNER_RETIRING) { + channel->owner_reinit_msg = tal_free(channel->owner_reinit_msg); + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel->dualopend_restart_mode + = DUALOPEND_RESTART_REESTABLISH; + } + + /* FIXME: This only bounds open_attempt RPCs while the peer is continuously + * disconnected and no owner exists. openchannel_signed_cmd and owners + * stuck installing a route or reestablishing have no lifecycle deadline. + * Use one timeout covering the entire retained dual-open recovery instead. + * Any serialized dual-open command can be stranded while its peer is + * absent, including an RBF request in AWAITING_LOCKIN. */ + if (!oa || !oa->cmd || oa->disconnect_timer || channel->owner + || channel->peer->connected == PEER_CONNECTED) + return; + + log_debug(channel->log, + "Arming dual-open disconnect timeout for %u seconds", + channel->peer->ld->config.dual_open_disconnect_timeout_secs); + oa->disconnect_timer + = new_reltimer(channel->peer->ld->timers, oa, + time_from_sec(channel->peer->ld->config + .dual_open_disconnect_timeout_secs), + open_attempt_disconnect_timeout, channel); } static void channel_err_broken(struct channel *channel, @@ -967,16 +1016,9 @@ static void dualopend_tell_depth(struct channel *channel, const u8 *msg; u32 to_go; - if (!channel->owner) { - log_debug(channel->log, - "Funding tx %s confirmed, but peer disconnected", - fmt_bitcoin_txid(tmpctx, txid)); - return; - } - log_debug(channel->log, - "Funding tx %s confirmed, telling peer", - fmt_bitcoin_txid(tmpctx, txid)); + "Funding tx %s confirmed at depth %u", + fmt_bitcoin_txid(tmpctx, txid), depth); if (depth < channel->minimum_depth) { to_go = channel->minimum_depth - depth; } else @@ -987,6 +1029,17 @@ static void dualopend_tell_depth(struct channel *channel, assert(channel->scid); assert(bitcoin_txid_eq(&channel->funding.txid, txid)); + if (!channel->owner + || channel->dualopend_owner_state + != DUALOPEND_OWNER_READY) { + log_debug(channel->log, + "Funding depth reached before dualopend owner ready"); + channel->dualopend_depth_needs_recheck = true; + return; + } + + channel->dualopend_depth_needs_recheck = false; + channel_set_billboard(channel, false, tal_fmt(tmpctx, "Funding depth reached" " %d confirmations, alerting peer" @@ -995,11 +1048,33 @@ static void dualopend_tell_depth(struct channel *channel, msg = towire_dualopend_depth_reached(NULL, depth); subd_send_msg(channel->owner, take(msg)); - } else + } else { + channel->dualopend_depth_needs_recheck = false; channel_set_billboard(channel, false, tal_fmt(tmpctx, "Funding needs %d more" " confirmations to be ready.", to_go)); + } +} + +static void dualopend_recheck_depth(struct channel *channel) +{ + u32 current_height, funding_height; + + if (!channel->dualopend_depth_needs_recheck) + return; + + channel->dualopend_depth_needs_recheck = false; + if (!channel->scid) + return; + + funding_height = short_channel_id_blocknum(*channel->scid); + current_height = get_block_height(channel->peer->ld->topology); + if (current_height < funding_height) + return; + + dualopend_tell_depth(channel, &channel->funding.txid, + current_height - funding_height + 1); } static enum watch_result opening_depth_cb(struct lightningd *ld, @@ -1023,6 +1098,7 @@ static enum watch_result opening_reorged_cb(struct lightningd *ld, struct channe { /* Reorged out? OK, we're not committed yet. */ log_info(inflight->channel->log, "Candidate funding tx was in a block, now reorged out"); + inflight->channel->dualopend_depth_needs_recheck = false; return DELETE_WATCH; } @@ -1125,9 +1201,12 @@ openchannel2_sign_hook_cb(struct openchannel2_psbt_payload *payload STEALS) send_msg: /* Peer's gone away, let's try reconnecting */ if (!payload->dualopend) { - channel_saved_err_broken_reconn(channel, - "dualopend daemon died" - " before signed PSBT returned"); + /* The signed PSBT was saved above. Owner retirement while the + * asynchronous plugin hook was running is recoverable: its replacement + * reloads this inflight during reinit. The owner's error path is + * responsible for retiring or reconnecting the peer transport. */ + log_debug(channel->log, + "Signed PSBT returned after dualopend owner retired"); tal_free(msg); return; } @@ -2451,8 +2530,8 @@ json_openchannel_abort(struct command *cmd, return command_still_pending(cmd); } -static char *restart_dualopend(const tal_t *ctx, const struct lightningd *ld, - struct channel *channel, bool from_abort) +static char *restart_dualopend(const tal_t *ctx, struct channel *channel, + bool from_abort) { struct peer_fd *pfd; int other_fd; @@ -2468,15 +2547,27 @@ static char *restart_dualopend(const tal_t *ctx, const struct lightningd *ld, close(other_fd); return tal_fmt(ctx, "Peer not connected"); } - subd_send_msg(ld->connectd, - take(towire_connectd_peer_connect_subd(NULL, - &channel->peer->id, - channel->peer->connectd_counter, - &channel->cid))); - subd_send_fd(ld->connectd, other_fd); + assert(!channel->dualopend_connectd_fd); + channel->dualopend_connectd_fd = new_peer_fd(channel, other_fd); return NULL; } +static void send_pending_open_attempt(struct channel *channel) +{ + struct open_attempt *oa = channel->open_attempt; + + if (!oa || !oa->cmd || !oa->open_msg || !channel->owner + || channel->dualopend_owner_state != DUALOPEND_OWNER_READY + || oa->open_msg_state == OPEN_ATTEMPT_MSG_SENT) + return; + + /* Record this before queueing the message: teardown can run while + * servicing the queue and must not cause a duplicate send. */ + oa->open_msg_state = OPEN_ATTEMPT_MSG_SENT; + channel->dualopend_restart_mode = DUALOPEND_RESTART_REESTABLISH; + subd_send_msg(channel->owner, oa->open_msg); +} + struct openchannel_bump_info { struct command *cmd; struct channel_id *cid; @@ -2511,26 +2602,47 @@ static struct command_result *openchannel_bump(struct openchannel_bump_info *inf "secured, see `openchannel_signed`"); } - /* It's possible that the last open failed/was aborted. - * So now we restart the attempt! */ - if (!channel->owner) { - char *err = restart_dualopend(cmd, cmd->ld, channel, false); - if (err) - return command_fail(cmd, FUNDING_PEER_NOT_CONNECTED, - "%s", err); - } + if (channel->open_attempt && channel->open_attempt->cmd) + return command_fail(cmd, FUNDING_STATE_INVALID, + "Another openchannel command is pending"); + /* Serialize the request before deciding whether an owner is usable. It + * survives owner replacement and is sent only in OWNER_READY. */ channel->open_attempt = oa = new_channel_open_attempt(channel); oa->funding = *info->amount; oa->cmd = info->cmd; oa->our_upfront_shutdown_script = channel->shutdown_scriptpubkey[LOCAL]; + oa->open_msg = towire_dualopend_rbf_init(oa, *info->amount, + *info->feerate_per_kw_funding, + info->psbt); - subd_send_msg(channel->owner, - take(towire_dualopend_rbf_init(NULL, *info->amount, - *info->feerate_per_kw_funding, - info->psbt))); - return command_still_pending(cmd); + switch (channel->dualopend_owner_state) { + case DUALOPEND_OWNER_READY: + send_pending_open_attempt(channel); + return command_still_pending(cmd); + + case DUALOPEND_OWNER_NONE: { + bool from_abort = channel->dualopend_restart_mode + == DUALOPEND_RESTART_AFTER_ABORT; + char *err = restart_dualopend(cmd, channel, from_abort); + if (err) { + oa->cmd = NULL; + channel->open_attempt = tal_free(channel->open_attempt); + return command_fail(cmd, FUNDING_PEER_NOT_CONNECTED, + "%s", err); + } + return command_still_pending(cmd); + } + + case DUALOPEND_OWNER_RESTART_PENDING: + case DUALOPEND_OWNER_ROUTE_PENDING: + case DUALOPEND_OWNER_READY_PENDING_ROUTE: + case DUALOPEND_OWNER_REESTABLISHING: + case DUALOPEND_OWNER_RETIRING: + return command_still_pending(cmd); + } + abort(); } /* sync_waiter must return void, so we use a simple wrapper */ @@ -2795,18 +2907,19 @@ json_openchannel_signed(struct command *cmd, channel->funding_psbt = clone_psbt(channel, inflight->funding_psbt); wallet_channel_save(cmd->ld->wallet, channel); - /* Only after we've updated/saved our psbt do we check - * for peer connected */ - if (!channel->owner) - return command_fail(cmd, FUNDING_PEER_NOT_CONNECTED, - "Peer not connected"); + /* Keep the RPC after the signed PSBT is durable. A replacement + * dualopend loads it from reinit and the channel_reestablish dance + * retransmits tx_signatures when required. */ + channel->openchannel_signed_cmd = tal_steal(channel, cmd); + if (!channel->owner + || channel->dualopend_owner_state != DUALOPEND_OWNER_READY) + return command_still_pending(cmd); /* Send our tx_sigs to the peer */ subd_send_msg(channel->owner, take(towire_dualopend_send_tx_sigs(NULL, inflight->funding_psbt))); - channel->openchannel_signed_cmd = tal_steal(channel, cmd); return command_still_pending(cmd); } @@ -2878,10 +2991,17 @@ static void validate_input_unspent(struct bitcoind *bitcoind, static void openchannel_update_valid_psbt(struct psbt_validator *pv) { + struct open_attempt *oa = pv->channel->open_attempt; u8 *msg; + assert(pv->cmd); - pv->channel->open_attempt->cmd = pv->cmd; + oa->cmd = pv->cmd; + /* A PSBT update cannot be replayed blindly after channel_reestablish: the + * peer may resolve a next_funding mismatch with tx_abort first. Clear the + * earlier rbf_init replay slot so disconnect cleanup fails this command and + * leaves reconciliation to the reestablish/abort path. */ + oa->open_msg = tal_free(oa->open_msg); msg = towire_dualopend_psbt_updated(NULL, pv->psbt); subd_send_msg(pv->channel->owner, take(msg)); } @@ -2966,13 +3086,11 @@ static struct command_result *json_openchannel_update(struct command *cmd, return command_fail(cmd, FUNDING_UNKNOWN_CHANNEL, "Unknown channel %s", fmt_channel_id(tmpctx, cid)); - if (!channel->owner) - return command_fail(cmd, FUNDING_PEER_NOT_CONNECTED, - "Peer not connected"); - if (!channel->open_attempt) { - /* Check if the last inflight for this matches? */ + /* Check if the last inflight for this matches. This is an + * idempotent lookup and does not need a live owner: the result was + * already made durable before the previous owner disconnected. */ inflight = find_inprogress_inflight(channel, psbt); if (inflight) { return command_success(cmd, @@ -2982,6 +3100,10 @@ static struct command_result *json_openchannel_update(struct command *cmd, "Channel open not in progress"); } + if (!channel->owner) + return command_fail(cmd, FUNDING_PEER_NOT_CONNECTED, + "Peer not connected"); + if (channel->open_attempt->cmd) return command_fail(cmd, FUNDING_STATE_INVALID, "Another openchannel command" @@ -3142,6 +3264,7 @@ static struct command_result *openchannel_init(struct command *cmd, } /* Go! */ + channel->open_attempt->open_msg_state = OPEN_ATTEMPT_MSG_SENT; subd_send_msg(channel->owner, channel->open_attempt->open_msg); /* Tell connectd connect this to this channel id. */ @@ -3728,12 +3851,92 @@ static void handle_dualopend_got_announcement(struct subd *dualopend, const u8 * &remote_ann_bitcoin_sig); } +static void handle_dualopend_abort_complete(struct subd *dualopend, + const u8 *msg) +{ + struct channel *channel = dualopend->channel; + char *reason; + + if (!fromwire_dualopend_abort_complete(tmpctx, msg, &reason)) { + channel_internal_error(channel, + "bad dualopend_abort_complete %s", + tal_hex(tmpctx, msg)); + return; + } + + /* dualopend retained its peer and HSM descriptors and restored the last + * completed funding attempt. Mirror that rollback in the wallet and + * ask connectd to make the quarantined route usable again. Do not finish + * the RPC until connectd acknowledges that barrier: callers commonly begin + * their retry immediately after openchannel_abort returns. */ + if (maybe_cleanup_last_inflight(channel)) + log_debug(channel->log, "Cleaned up incomplete inflight"); + channel->dualopend_abort_reason = tal_steal(channel, reason); + channel->dualopend_owner_state = DUALOPEND_OWNER_READY_PENDING_ROUTE; + subd_send_msg(channel->peer->ld->connectd, + take(towire_connectd_peer_resume_subd( + NULL, &channel->peer->id, + channel->dualopend_owner_connectd_counter, + &channel->cid))); +} + static unsigned int dual_opend_msg(struct subd *dualopend, const u8 *msg, const int *fds) { enum dualopend_wire t = fromwire_peektype(msg); struct channel *channel = dualopend->channel; + if (t != WIRE_DUALOPEND_READY && t != WIRE_DUALOPEND_INIT_READY) + channel->dualopend_restart_mode + = DUALOPEND_RESTART_REESTABLISH; + + if (t == WIRE_DUALOPEND_INIT_READY) { + struct peer_fd *connectd_fd = channel->dualopend_connectd_fd; + int fd; + + /* Abort continuations inherit an already-attached route. */ + if (!connectd_fd) + return 0; + channel->dualopend_connectd_fd = NULL; + fd = connectd_fd->fd; + connectd_fd->fd = -1; + subd_send_msg(channel->peer->ld->connectd, + take(towire_connectd_peer_connect_subd_tracked( + NULL, &channel->peer->id, + channel->dualopend_owner_connectd_counter, + &channel->cid))); + subd_send_fd(channel->peer->ld->connectd, fd); + tal_free(connectd_fd); + return 0; + } + + if (t == WIRE_DUALOPEND_READY) { + /* This signal is emitted only after initialization and, for normal + * reconnects, after channel_reestablish has completed. An abort + * restart is no longer special once it reaches this point: retaining + * AFTER_ABORT would make a later transport failure look like a failed + * abort installation and could start a competing owner. */ + channel->dualopend_restart_mode + = DUALOPEND_RESTART_REESTABLISH; + if (channel->dualopend_owner_state + == DUALOPEND_OWNER_ROUTE_PENDING) { + channel->dualopend_owner_state + = DUALOPEND_OWNER_READY_PENDING_ROUTE; + return 0; + } + if (channel->dualopend_owner_state + != DUALOPEND_OWNER_REESTABLISHING) { + log_broken(channel->log, + "dualopend ready in owner state %u", + channel->dualopend_owner_state); + return 0; + } + channel->dualopend_owner_state = DUALOPEND_OWNER_READY; + dualopend_recheck_depth(channel); + send_pending_open_attempt(channel); + return 0; + } + switch (t) { case WIRE_DUALOPEND_GOT_OFFER: accepter_got_offer(dualopend, dualopend->channel, msg); @@ -3793,6 +3996,9 @@ static unsigned int dual_opend_msg(struct subd *dualopend, case WIRE_DUALOPEND_GOT_ANNOUNCEMENT: handle_dualopend_got_announcement(dualopend, msg); return 0; + case WIRE_DUALOPEND_ABORT_COMPLETE: + handle_dualopend_abort_complete(dualopend, msg); + return 0; /* Messages we send */ case WIRE_DUALOPEND_INIT: case WIRE_DUALOPEND_REINIT: @@ -3811,6 +4017,8 @@ static unsigned int dual_opend_msg(struct subd *dualopend, case WIRE_DUALOPEND_DEPTH_REACHED: case WIRE_DUALOPEND_DEV_MEMLEAK: case WIRE_DUALOPEND_DEV_MEMLEAK_REPLY: + case WIRE_DUALOPEND_INIT_READY: + case WIRE_DUALOPEND_READY: break; } @@ -3954,6 +4162,7 @@ static struct command_result *json_queryrates(struct command *cmd, } /* Go! */ + channel->open_attempt->open_msg_state = OPEN_ATTEMPT_MSG_SENT; subd_send_msg(channel->owner, channel->open_attempt->open_msg); /* Tell connectd connect this to this channel id. */ @@ -4005,6 +4214,116 @@ AUTODATA(json_command, &openchannel_signed_command); AUTODATA(json_command, &openchannel_bump_command); AUTODATA(json_command, &openchannel_abort_command); +static void restart_dualopend_after_owner_exit(struct channel *channel) +{ + char *err; + bool from_abort; + + /* Owner destruction can race either a same-transport abort continuation or + * a new peer transport. Decide the startup protocol only after both the + * old owner and the peer connection have reached their current states. */ + if (channel->dualopend_owner_state != DUALOPEND_OWNER_RETIRING + || channel->owner) { + return; + } + + if (channel->peer->connected != PEER_CONNECTED) { + channel->dualopend_restart_peer_fd + = tal_free(channel->dualopend_restart_peer_fd); + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + dual_open_attempt_peer_disconnected(channel); + return; + } + + from_abort = channel->dualopend_restart_mode + == DUALOPEND_RESTART_AFTER_ABORT + && channel->dualopend_owner_connectd_counter + == channel->peer->connectd_counter + && channel->dualopend_restart_peer_fd != NULL; + if (!from_abort) { + channel->dualopend_restart_peer_fd + = tal_free(channel->dualopend_restart_peer_fd); + channel->dualopend_restart_mode + = DUALOPEND_RESTART_REESTABLISH; + } + + if (from_abort) { + /* Keep the endpoint returned by the old owner. connectd still owns + * the other end of this exact abort-marked route; resume_subd is the + * barrier which cancels its close timer and makes it routable again. */ + channel->dualopend_owner_state = DUALOPEND_OWNER_RESTART_PENDING; + subd_send_msg(channel->peer->ld->connectd, + take(towire_connectd_peer_resume_subd( + NULL, &channel->peer->id, + channel->peer->connectd_counter, + &channel->cid))); + err = NULL; + } else { + err = restart_dualopend(tmpctx, channel, false); + } + if (err) { + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel_cleanup_commands(channel, + tal_fmt(tmpctx, + "Unable to restart dualopend after owner exit: %s", + err)); + return; + } +} + +/* A lost peer fd is different from tx_abort: connectd's transport-loss + * notification can arrive after the owning subdaemon has exited. Never + * restart on the transport which just produced EOF merely because lightningd + * still labels it connected. Whichever side of the owner-exit/disconnect + * race runs second completes retirement and arms any retained RPC timeout. */ +static void finish_dualopend_transport_loss(struct channel *channel) +{ + if (channel->dualopend_owner_state != DUALOPEND_OWNER_RETIRING + || channel->owner) + return; + + channel->dualopend_restart_mode = DUALOPEND_RESTART_REESTABLISH; + if (channel->peer->connected == PEER_CONNECTED) { + if (channel->dualopend_owner_connectd_counter + == channel->peer->connectd_counter) { + channel->dualopend_owner_state + = DUALOPEND_OWNER_RESTART_PENDING; + return; + } + restart_dualopend_after_owner_exit(channel); + return; + } + + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + dual_open_attempt_peer_disconnected(channel); +} + +/* A zero-length timer from dualopen_errmsg is not an owner-exit barrier: the + * timer can run before ccan/io has closed and destroyed the old subd. In that + * case the replacement steals both the connectd route and the exclusive hsmd + * client while the old process is still unwinding. Schedule the restart from + * the old subd's destructor instead, so both of those resources are genuinely + * released first. */ +static void restart_dualopend_after_owner_exit_cb(struct subd *old_owner UNUSED, + struct channel *channel) +{ + restart_dualopend_after_owner_exit(channel); +} + +void dual_open_owner_begin_retirement(struct channel *channel, + struct subd *retiring_owner) +{ + /* The first path which notices retirement owns the destructor callback. + * Later EOF/status callbacks from this same subdaemon must be inert. */ + if (channel->dualopend_owner_state == DUALOPEND_OWNER_RETIRING) + return; + + channel->dualopend_owner_state = DUALOPEND_OWNER_RETIRING; + tal_add_destructor2(retiring_owner, + restart_dualopend_after_owner_exit_cb, + channel); +} + static void dualopen_errmsg(struct channel *channel, struct peer_fd *peer_fd, const char *desc, @@ -4012,46 +4331,119 @@ static void dualopen_errmsg(struct channel *channel, bool disconnect, bool warning) { - /* Clean up any in-progress open attempts */ - channel_cleanup_commands(channel, desc); - if (channel_state_uncommitted(channel->state)) { + channel_cleanup_commands(channel, desc); log_info(channel->log, "%s", "Unsaved peer failed." " Deleting channel."); delete_channel(channel, false); return; } - if ((warning || disconnect) && channel_state_open_uncommitted(channel->state)) { - log_info(channel->log, "%s", "Commit ready peer failed." - " Deleting channel."); - delete_channel(channel, false); - return; - } - /* Do we have an error to send? */ - if (err_for_them && !channel->error && !warning) - channel->error = tal_dup_talarr(channel, u8, err_for_them); - - /* No peer_fd means a subd crash or disconnection. */ + /* No peer_fd means dualopend died or the transport disappeared. A + * saved dual-funded open can resume after reconnect, so retain its + * command and PSBT. */ if (!peer_fd) { - if (!warning && disconnect) + struct subd *retiring_owner = channel->owner; + + /* peer_channels_cleanup may already have retired this owner. Its + * eventual destructor, not this late status callback, owns restart. */ + if (channel->dualopend_owner_state == DUALOPEND_OWNER_RETIRING) + return; + + if (!warning && disconnect) { + channel->owner_reinit_msg + = tal_free(channel->owner_reinit_msg); + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel_cleanup_commands(channel, desc); channel_fail_permanent(channel, err_for_them ? REASON_LOCAL : REASON_PROTOCOL, "%s: %s ERROR %s", channel->owner->name, err_for_them ? "sent" : "received", desc); - else - /* If the channel is unsaved, we forget it */ - channel_fail_transient(channel, disconnect, "%s: %s", + } else { + u64 owner_connectd_counter + = channel->dualopend_owner_connectd_counter; + log_debug(channel->log, + "Retiring failed dualopend transport counter %"PRIu64 + " (peer current %"PRIu64")", + owner_connectd_counter, + channel->peer->connectd_counter); + + channel->owner_reinit_msg + = tal_free(channel->owner_reinit_msg); + assert(retiring_owner); + /* This callback is made by destroy_subd() after it has detached + * itself from the channel. It is too late to add another + * destructor to retiring_owner: schedule the restart for the next + * event-loop turn, after destroy_subd() has returned. */ + channel->dualopend_owner_state = DUALOPEND_OWNER_RETIRING; + /* The dev-memleak RPC can run before this next-turn continuation. + * It remains intentionally reachable from the timer list and is + * cancelled automatically if the channel is freed. */ + notleak(new_reltimer(channel->peer->ld->timers, channel, + time_from_msec(0), + finish_dualopend_transport_loss, + channel)); + if (!channel_state_saved_dualopend(channel->state) + && !dual_open_attempt_waiting_for_owner(channel) + && !channel->openchannel_signed_cmd) + channel_cleanup_commands(channel, desc); + /* No peer_fd means the owner's route to this transport has + * disappeared. Reusing that transport is unsafe: the old owner + * may already have completed channel_reestablish, in which case a + * replacement would wait forever for a second one. Retire this + * transport and let peer_connected install the replacement on the + * next connection. */ + channel_fail_transient(channel, false, "%s: %s", channel->owner->name, desc); + + /* The peer may already have reconnected inside connectd, before + * lightningd has processed PEER_RECONNECTED. Disconnect only the + * transport which owned this route; connectd ignores a stale + * counter instead of tearing down its replacement. */ + if (owner_connectd_counter != 0) + subd_send_msg(channel->peer->ld->connectd, + take(towire_connectd_disconnect_peer(NULL, + &channel->peer->id, + owner_connectd_counter))); + /* The transport-loss notification follows this callback. Keep + * RETIRING until peer_channels_cleanup observes PEER_DISCONNECTED; + * otherwise a zero-length timer can install a replacement on the + * transport which is still shutting down. */ + dual_open_attempt_peer_disconnected(channel); + } return; } + /* Peer errors and explicit aborts finish any in-progress attempt. */ + channel_cleanup_commands(channel, desc); + + if ((warning || disconnect) && channel_state_open_uncommitted(channel->state)) { + log_info(channel->log, "%s", "Commit ready peer failed." + " Deleting channel."); + delete_channel(channel, false); + return; + } + + /* Do we have an error to send? */ + if (err_for_them && !channel->error && !warning) + channel->error = tal_dup_talarr(channel, u8, err_for_them); + /* Other implementations chose to ignore errors early on. Not * surprisingly, they now spew out spurious errors frequently, * and we would close the channel on them. We now support warnings * for this case. */ if (warning || !disconnect) { + struct subd *retiring_owner = channel->owner; + + /* Claim the retiring state before releasing the old owner. All other + * owner installers defer while this transition is in progress. */ + if (!disconnect && !channel_state_open_uncommitted(channel->state)) { + channel->dualopend_restart_mode + = DUALOPEND_RESTART_AFTER_ABORT; + dual_open_owner_begin_retirement(channel, retiring_owner); + } + /* We *don't* hang up if they aborted: that's fine! */ channel_fail_transient(channel, disconnect, "%s %s: %s", channel->owner->name, @@ -4063,7 +4455,6 @@ static void dualopen_errmsg(struct channel *channel, if (maybe_cleanup_last_inflight(channel)) log_debug(channel->log, "Cleaned up incomplete inflight"); - if (!disconnect) { if (channel_state_open_uncommitted(channel->state)) { log_info(channel->log, "%s", "Commit ready peer can't reconnect." @@ -4071,13 +4462,15 @@ static void dualopen_errmsg(struct channel *channel, delete_channel(channel, false); return; } - char *err = restart_dualopend(tmpctx, - channel->peer->ld, - channel, true); - if (err) - log_broken(channel->log, - "Unable to restart dualopend" - " after abort: %s", err); + /* tx_abort hands the existing connectd endpoint back to lightningd. + * Keep it for the replacement owner; the old owner's destructor is + * the barrier which makes handing it on safe. */ + assert(channel->dualopend_owner_state + == DUALOPEND_OWNER_RETIRING); + assert(retiring_owner); + assert(!channel->dualopend_restart_peer_fd); + channel->dualopend_restart_peer_fd + = tal_steal(channel, peer_fd); } return; @@ -4176,9 +4569,8 @@ bool peer_start_dualopend(struct peer *peer, strerror(errno)); return false; } - dev_old_seed = dev_setup_dualopend_seed(tmpctx, peer->ld); - channel->owner = new_channel_subd(channel, + channel_set_owner(channel, new_channel_subd(channel, peer->ld, "lightning_dualopend", channel, @@ -4189,7 +4581,7 @@ bool peer_start_dualopend(struct peer *peer, dualopen_errmsg, channel_set_billboard, take(&peer_fd->fd), - take(&hsmfd), NULL); + take(&hsmfd), NULL)); dev_restore_seed(dev_old_seed); if (!channel->owner) { @@ -4229,9 +4621,127 @@ bool peer_start_dualopend(struct peer *peer, *channel->alias[LOCAL], peer->ld->dev_any_channel_type); subd_send_msg(channel->owner, take(msg)); + /* Initial dualopend does not perform a reconnect dance. Its init is + * already queued before any later master request. */ + channel->dualopend_owner_state = DUALOPEND_OWNER_READY; + channel->dualopend_restart_mode = DUALOPEND_RESTART_REESTABLISH; + channel->dualopend_owner_connectd_counter = peer->connectd_counter; return true; } +void dual_open_owner_route_result(struct lightningd *ld, const u8 *msg) +{ + struct node_id peer_id; + struct channel_id cid; + struct channel *channel; + u64 connectd_counter; + bool accepted; + + if (!fromwire_connectd_peer_connect_subd_reply(msg, &peer_id, + &connectd_counter, + &cid, + &accepted)) + fatal("Bad connectd_peer_connect_subd_reply: %s", + tal_hex(msg, msg)); + + channel = channel_by_cid(ld, &cid); + if (!channel || !node_id_eq(&channel->peer->id, &peer_id)) + return; + + /* For an in-place tx_abort, this acknowledgement means connectd has + * reactivated the abort-marked route. Only now is it safe to attach the + * returned endpoint to the replacement owner. */ + if (channel->peer->connectd_counter == connectd_counter + && channel->dualopend_owner_state + == DUALOPEND_OWNER_RESTART_PENDING + && channel->dualopend_restart_mode + == DUALOPEND_RESTART_AFTER_ABORT) { + struct peer_fd *peer_fd; + + if (!accepted) { + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + dual_open_attempt_peer_disconnected(channel); + return; + } + peer_fd = channel->dualopend_restart_peer_fd; + channel->dualopend_restart_peer_fd = NULL; + if (!peer_fd + || !peer_restart_dualopend(channel->peer, peer_fd, + channel, true)) { + tal_free(peer_fd); + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + channel_cleanup_commands(channel, + "Unable to restart dualopend after abort"); + return; + } + tal_free(peer_fd); + /* The route acknowledgement preceded owner creation, so initialize + * immediately instead of waiting for a connect_subd reply. */ + assert(channel->owner_reinit_msg); + subd_send_msg(channel->owner, take(channel->owner_reinit_msg)); + channel->owner_reinit_msg = NULL; + channel->dualopend_owner_state = DUALOPEND_OWNER_REESTABLISHING; + return; + } + + /* Initial owners do not wait on this reply, and a disconnect can retire a + * route-pending owner before connectd's response is handled. */ + if (channel->peer->connectd_counter != connectd_counter + || (channel->dualopend_owner_state + != DUALOPEND_OWNER_ROUTE_PENDING + && channel->dualopend_owner_state + != DUALOPEND_OWNER_READY_PENDING_ROUTE)) + return; + + if (!accepted) { + log_debug(channel->log, "Connectd rejected dualopend owner route"); + /* The owner cannot service peer traffic without a connectd route. + * Retire it now; the peer-disconnected/reconnected path will install + * the owner for the next transport. */ + channel->owner_reinit_msg = tal_free(channel->owner_reinit_msg); + channel->dualopend_owner_state = DUALOPEND_OWNER_NONE; + if (channel->dualopend_abort_reason) { + channel_cleanup_commands(channel, + channel->dualopend_abort_reason); + channel->dualopend_abort_reason + = tal_free(channel->dualopend_abort_reason); + } + channel_fail_transient(channel, false, + "Connectd rejected owner route"); + return; + } + + if (!channel->owner) + return; + + log_debug(channel->log, "Connectd accepted dualopend owner route"); + if (channel->dualopend_owner_state + == DUALOPEND_OWNER_READY_PENDING_ROUTE) { + channel->dualopend_owner_state = DUALOPEND_OWNER_READY; + if (channel->dualopend_abort_reason) { + log_info(channel->log, "RBF negotiation aborted: %s", + channel->dualopend_abort_reason); + channel_cleanup_commands(channel, + channel->dualopend_abort_reason); + channel->dualopend_abort_reason + = tal_free(channel->dualopend_abort_reason); + } + dualopend_recheck_depth(channel); + send_pending_open_attempt(channel); + } else { + /* Normal reconnect owners are initialized before route attachment, so + * they cannot receive a queued channel_reestablish while still waiting + * for their master init. Abort continuations retain their init until + * connectd has reactivated the inherited route. */ + if (channel->owner_reinit_msg) { + subd_send_msg(channel->owner, + take(channel->owner_reinit_msg)); + channel->owner_reinit_msg = NULL; + } + channel->dualopend_owner_state = DUALOPEND_OWNER_REESTABLISHING; + } +} + bool peer_restart_dualopend(struct peer *peer, struct peer_fd *peer_fd, struct channel *channel, @@ -4263,7 +4773,6 @@ bool peer_restart_dualopend(struct peer *peer, "Failed to get hsm fd for dualopend"); return false; } - dev_old_seed = dev_setup_dualopend_seed(tmpctx, peer->ld); channel_set_owner(channel, new_channel_subd(channel, peer->ld, @@ -4277,6 +4786,8 @@ bool peer_restart_dualopend(struct peer *peer, channel_set_billboard, take(&peer_fd->fd), take(&hsmfd), NULL)); + assert(peer_fd->fd == -1); + assert(hsmfd == -1); dev_restore_seed(dev_old_seed); if (!channel->owner) { @@ -4287,6 +4798,14 @@ bool peer_restart_dualopend(struct peer *peer, "Failed to create dualopend"); return false; } + channel->dualopend_owner_state = DUALOPEND_OWNER_ROUTE_PENDING; + if (channel->dualopend_owner_connectd_counter + != peer->connectd_counter) + channel->dualopend_owner_install_retries = 0; + channel->dualopend_owner_connectd_counter = peer->connectd_counter; + if (from_abort) + channel->dualopend_restart_mode + = DUALOPEND_RESTART_AFTER_ABORT; /* Find the max self delay and min htlc capacity */ channel_config(peer->ld, &unused_config, @@ -4363,6 +4882,16 @@ bool peer_restart_dualopend(struct peer *peer, channel->req_confirmed_ins[REMOTE], *channel->alias[LOCAL]); - subd_send_msg(channel->owner, take(msg)); + if (from_abort) { + /* An inherited route remains abort-marked until connectd explicitly + * resumes it. */ + channel->owner_reinit_msg = tal_steal(channel, msg); + } else { + /* Initialize before attaching the route. The peer fd buffers our + * channel_reestablish until connectd installs its endpoint, and an + * already-queued peer reestablish cannot race ahead of master init. */ + subd_send_msg(channel->owner, take(msg)); + channel->owner_reinit_msg = NULL; + } return true; } diff --git a/lightningd/dual_open_control.h b/lightningd/dual_open_control.h index c9d4c4d72952..f8bacfc88453 100644 --- a/lightningd/dual_open_control.h +++ b/lightningd/dual_open_control.h @@ -20,6 +20,19 @@ void watch_opening_inflight(struct lightningd *ld, /* Close connection to an unsaved channel */ void channel_unsaved_close_conn(struct channel *channel, const char *why); +/* Bound retention of a pending saved dual-open while its peer is absent. */ +void dual_open_attempt_peer_disconnected(struct channel *channel); + +/* Serialize replacement behind destruction of the current dualopend. */ +void dual_open_owner_begin_retirement(struct channel *channel, + struct subd *retiring_owner); + +/* True when an RPC has been serialized but not delivered to a ready owner. */ +bool dual_open_attempt_waiting_for_owner(const struct channel *channel); + +/* Connectd accepted (or rejected) the pending dualopend peer route. */ +void dual_open_owner_route_result(struct lightningd *ld, const u8 *msg); + void NO_NULL_ARGS json_add_unsaved_channel(struct command *cmd, struct json_stream *response, const struct channel *channel, diff --git a/lightningd/lightningd.c b/lightningd/lightningd.c index 44ed3fa66425..dfd159efa51d 100644 --- a/lightningd/lightningd.c +++ b/lightningd/lightningd.c @@ -1199,6 +1199,7 @@ int main(int argc, char *argv[]) /*~ What happens in strange locales should stay there. */ setup_locale(); + io_poll_protect_stale_fds(); /*~ This handles --dev-debug-self really early, which we otherwise ignore */ daemon_developer_mode(argv); diff --git a/lightningd/lightningd.h b/lightningd/lightningd.h index 6d778c929e95..994d5a558e5c 100644 --- a/lightningd/lightningd.h +++ b/lightningd/lightningd.h @@ -73,6 +73,9 @@ struct config { /* How long before we give up waiting for INIT msg */ u32 connection_timeout_secs; + /* How long a pending saved dual-open may wait for its peer to reconnect. */ + u32 dual_open_disconnect_timeout_secs; + /* Allow dust reserves (including 0) when being called via * `fundchannel` or in the `openchannel` hook. This is a * slight spec incompatibility, but implementations do this diff --git a/lightningd/opening_control.c b/lightningd/opening_control.c index 6cc3d82f1e61..eae76b982a6d 100644 --- a/lightningd/opening_control.c +++ b/lightningd/opening_control.c @@ -453,6 +453,13 @@ static void opening_funder_finished(struct subd *openingd, const u8 *resp, goto cleanup; } + /* The peer endpoint is safely held by peer_fd now, and openingd has + * completed its work. Retire it before peer_start_channeld() asks hsmd + * for another client with this dbid: hsmd deliberately will not create + * that client until the previous owner has gone away. */ + fc->uc->open_daemon = NULL; + subd_release_channel(openingd, fc->uc); + /* Watch for funding confirms */ channel_watch_funding(ld, channel); @@ -567,6 +574,13 @@ static void opening_fundee_finished(struct subd *openingd, goto failed; } + /* Establish an explicit owner-exit barrier before requesting channeld's + * HSM client. This is particularly important when several v1 opens + * finish concurrently: waiting synchronously for one lingering openingd + * must not delay processing another openingd's returned peer endpoint. */ + uc->open_daemon = NULL; + subd_release_channel(openingd, uc); + log_debug(channel->log, "Watching funding tx %s", fmt_bitcoin_txid(reply, &channel->funding.txid)); diff --git a/lightningd/options.c b/lightningd/options.c index 42ee3a6e43f3..0bf067dac5a4 100644 --- a/lightningd/options.c +++ b/lightningd/options.c @@ -1017,6 +1017,9 @@ static const struct config testnet_config = { /* 1 minute should be enough for anyone! */ .connection_timeout_secs = 60, + /* Longer than connectd's maximum reconnect backoff. */ + .dual_open_disconnect_timeout_secs = 600, + .allowdustreserve = false, .require_confirmed_inputs = false, @@ -1095,6 +1098,9 @@ static const struct config mainnet_config = { /* 1 minute should be enough for anyone! */ .connection_timeout_secs = 60, + /* Longer than connectd's maximum reconnect backoff. */ + .dual_open_disconnect_timeout_secs = 600, + .allowdustreserve = false, .require_confirmed_inputs = false, @@ -1536,6 +1542,10 @@ static void register_opts(struct lightningd *ld) clnopt_witharg("--funding-confirms", OPT_SHOWINT, opt_set_u32, opt_show_u32, &ld->config.funding_confirms, "Confirmations required for funding transaction"); + clnopt_witharg("--dual-open-disconnect-timeout", OPT_SHOWINT, + opt_set_u32, opt_show_u32, + &ld->config.dual_open_disconnect_timeout_secs, + "Seconds to retain a pending v2 open while its peer is disconnected"); clnopt_witharg("--require-confirmed-inputs", OPT_SHOWBOOL, opt_set_bool_arg, opt_show_bool, &ld->config.require_confirmed_inputs, diff --git a/lightningd/peer_control.c b/lightningd/peer_control.c index 1134bca48d70..649c0e62d50c 100644 --- a/lightningd/peer_control.c +++ b/lightningd/peer_control.c @@ -172,8 +172,23 @@ void peer_channels_cleanup(struct peer *peer) for (size_t i = 0; i < tal_count(channels); i++) { c = channels[i]; if (channel_state_wants_peercomms(c->state)) { - channel_cleanup_commands(c, "Disconnected"); + struct subd *retiring_owner = c->owner; + + if (retiring_owner + && (c->state == DUALOPEND_OPEN_COMMIT_READY + || c->state == DUALOPEND_OPEN_COMMITTED + || c->state == DUALOPEND_AWAITING_LOCKIN)) + dual_open_owner_begin_retirement(c, + retiring_owner); + /* Durable dual-funded operations are resumable. Keep their + * commands and PSBTs until dualopend reestablishes with the + * peer. */ + if (!channel_state_saved_dualopend(c->state) + && !dual_open_attempt_waiting_for_owner(c) + && !c->openchannel_signed_cmd) + channel_cleanup_commands(c, "Disconnected"); channel_fail_transient(c, true, "Disconnected"); + dual_open_attempt_peer_disconnected(c); } else if (channel_state_uncommitted(c->state)) { channel_unsaved_close_conn(c, "Disconnected"); } @@ -1399,6 +1414,15 @@ static void connect_activate_subd(struct lightningd *ld, struct channel *channel struct peer_fd *pfd; int other_fd; + /* The old dualopend can still deliver a terminal callback until its subd + * object is destroyed. Its destructor will install the replacement on + * whichever peer transport is current at that point. */ + if (channel->dualopend_owner_state == DUALOPEND_OWNER_RETIRING) { + log_debug(channel->log, + "Deferring dualopend install until retiring owner exits"); + return; + } + /* If we have a canned error for this channel, send it now */ if (channel->error) { error = channel->error; @@ -1430,8 +1454,14 @@ static void connect_activate_subd(struct lightningd *ld, struct channel *channel if (peer_restart_dualopend(channel->peer, pfd, - channel, false)) - goto tell_connectd; + channel, + channel->dualopend_restart_mode + == DUALOPEND_RESTART_AFTER_ABORT)) { + assert(!channel->dualopend_connectd_fd); + channel->dualopend_connectd_fd + = new_peer_fd(channel, other_fd); + return; + } close(other_fd); return; @@ -2071,6 +2101,20 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg) return; } + /* connectd retains this message until an owner is attached. While + * the peer_connected hooks are running, connect_activate_subd is + * the designated owner installer; the abort path can likewise have + * a selected replacement pending. In either window, leave the + * message queued instead of spawning a competing owner here. */ + if (!channel->owner + && (peer->connected == PEER_CONNECTING + || channel->dualopend_owner_state + != DUALOPEND_OWNER_NONE)) { + log_debug(channel->log, + "Deferring peer message for pending owner install"); + return; + } + /* If channel is active there are two possibilities: * 1. We have started subd, but channeld hasn't processed * the connectd_peer_connect_subd message yet. @@ -2101,7 +2145,9 @@ void handle_peer_spoke(struct lightningd *ld, const u8 *msg) pfd = sockpair(tmpctx, channel, &other_fd, &error); if (!pfd) goto send_error; - if (peer_restart_dualopend(peer, pfd, channel, false)) + if (peer_restart_dualopend(peer, pfd, channel, + channel->dualopend_restart_mode + == DUALOPEND_RESTART_AFTER_ABORT)) goto tell_connectd; /* FIXME: Send informative error? */ close(other_fd); @@ -2247,8 +2293,10 @@ static void peer_disconnected(struct lightningd *ld, /* Note: we don't force subds to stop. They should exit soon, * but if we get a (re)connection we'll force them to stop */ - list_for_each(&p->channels, channel, list) + list_for_each(&p->channels, channel, list) { channel_gossip_channel_disconnect(channel); + dual_open_attempt_peer_disconnected(channel); + } } /* If you were trying to connect, it failed. */ diff --git a/lightningd/test/run-invoice-select-inchan.c b/lightningd/test/run-invoice-select-inchan.c index 0b42cb1c7178..211e51b2d0dc 100644 --- a/lightningd/test/run-invoice-select-inchan.c +++ b/lightningd/test/run-invoice-select-inchan.c @@ -128,6 +128,20 @@ const char *channel_state_str(enum channel_state state UNNEEDED) /* Generated stub for channel_unsaved_close_conn */ void channel_unsaved_close_conn(struct channel *channel UNNEEDED, const char *why UNNEEDED) { fprintf(stderr, "channel_unsaved_close_conn called!\n"); abort(); } +/* Generated stub for dual_open_attempt_peer_disconnected */ +void dual_open_attempt_peer_disconnected(struct channel *channel UNNEEDED) +{ +} +/* Generated stub for dual_open_attempt_waiting_for_owner */ +bool dual_open_attempt_waiting_for_owner(const struct channel *channel UNNEEDED) +{ + return false; +} +/* Generated stub for dual_open_owner_begin_retirement */ +void dual_open_owner_begin_retirement(struct channel *channel UNNEEDED, + struct subd *retiring_owner UNNEEDED) +{ +} /* Generated stub for channel_update_feerates */ void channel_update_feerates(struct lightningd *ld UNNEEDED, const struct channel *channel UNNEEDED) { fprintf(stderr, "channel_update_feerates called!\n"); abort(); } diff --git a/openingd/dualopend.c b/openingd/dualopend.c index 2f1849773496..d8b9d4f7ad55 100644 --- a/openingd/dualopend.c +++ b/openingd/dualopend.c @@ -215,6 +215,13 @@ struct state { /* Did we send tx-abort? */ const char *aborted_err; + /* Set for the remainder of the current top-level dispatch after a + * tx_abort handshake completes in place. Nested callers must not start a + * second abort while unwinding their old NORETURN assumptions. */ + bool abort_completed; + /* While constructing an RBF, retain the last completed attempt so a + * negotiated tx_abort can roll back without restarting this daemon. */ + struct tx_state *previous_tx_state; /* State of inflight funding transaction attempt */ struct tx_state *tx_state; @@ -339,8 +346,34 @@ static bool shutdown_complete(const struct state *state) /* They failed the open with us */ static void negotiation_aborted(struct state *state, const char *why, bool aborted) { + u8 *msg; + status_debug("aborted opening negotiation: %s", why); + /* A saved channel can continue on the same peer connection after the + * tx_abort acknowledgement. Rolling back here avoids handing both the + * peer and HSM descriptors through a replacement process. */ + /* A daemon created from an incomplete reconnect candidate has no prior + * tx_state to restore. If that candidate also lacks the peer's funding + * signatures, let the established replacement path reload the preceding + * durable inflight after lightningd removes the incomplete one. */ + if (aborted && state->channel + && (state->previous_tx_state + || state->tx_state->remote_funding_sigs_rcvd)) { + msg = towire_dualopend_abort_complete(NULL, why); + + if (state->previous_tx_state) { + tal_free(state->tx_state); + state->tx_state = state->previous_tx_state; + state->previous_tx_state = NULL; + } + state->aborted_err = tal_free(state->aborted_err); + state->abort_completed = true; + wire_sync_write(REQ_FD, take(msg)); + peer_billboard(false, "RBF aborted, awaiting next attempt"); + return; + } + /* Tell master that funding failed (don't disconnect if we aborted) */ peer_failed_received_errmsg(state->pps, !aborted, why); } @@ -355,6 +388,11 @@ static void open_abort(struct state *state, const char *errmsg; u8 *msg; + /* A nested negotiation helper may still be unwinding after + * negotiation_aborted() completed the handshake in place. */ + if (state->abort_completed) + return; + va_start(ap, fmt); errmsg = tal_vfmt(NULL, fmt, ap); va_end(ap); @@ -552,6 +590,32 @@ static void handle_failure_fatal(struct state *state, u8 *msg) open_err_fatal(state, "%s", err); } +static bool check_accepter_error(struct state *state, + u8 *msg, + char *err_reason) +{ + if (!msg) { + if (err_reason) + negotiation_failed(state, "%s", err_reason); + else + /* FIXME: what do we do here?? */ + return false; + } + + /* `msg` could be a failure message */ + if (fromwire_peektype(msg) == WIRE_DUALOPEND_FAIL) { + handle_failure_fatal(state, msg); + return false; + } + + if (fromwire_peektype(msg) != WIRE_DUALOPEND_SEND_TX_SIGS) { + master_badmsg(WIRE_DUALOPEND_SEND_TX_SIGS, msg); + return false; + } + + return true; +} + static void check_channel_id(struct state *state, struct channel_id *id_in, struct channel_id *orig_id) @@ -2305,9 +2369,6 @@ static u8 *accepter_commits(struct state *state, wire_sync_write(REQ_FD, take(msg)); msg = wire_sync_read(tmpctx, REQ_FD); - if (fromwire_peektype(msg) != WIRE_DUALOPEND_SEND_TX_SIGS) - master_badmsg(WIRE_DUALOPEND_SEND_TX_SIGS, msg); - return msg; } @@ -2737,11 +2798,8 @@ static void accepter_start(struct state *state, const u8 *oc2_msg) } msg = accepter_commits(state, tx_state, total, &err_reason); - if (!msg) { - if (err_reason) - negotiation_failed(state, "%s", err_reason); - return; - } + if (!check_accepter_error(state, msg, err_reason)) + return; /* Finally, send our funding tx sigs */ handle_send_tx_sigs(state, msg); @@ -2908,6 +2966,7 @@ static u8 *opener_commits(struct state *state, msg = opening_negotiate_msg(tmpctx, state); if (!msg) { *err_reason = NULL; + tal_free(pbase); revert_channel_state(state); return NULL; } @@ -2917,6 +2976,7 @@ static u8 *opener_commits(struct state *state, &remote_sig); if (error) { *err_reason = tal_fmt(tmpctx, "Commit sig error: %s", error); + tal_free(pbase); revert_channel_state(state); return NULL; } @@ -3441,20 +3501,38 @@ static void rbf_wrap_up(struct state *state, else msg = opener_commits(state, tx_state, total, &err_reason); - if (!msg) { - if (err_reason) - open_abort(state, "%s", err_reason); - else - open_abort(state, "%s", "Unable to commit"); - /* We need to 'reset' the channel to what it - * was before we did this. */ - return; - } - - if (state->our_role == TX_ACCEPTER) + /* in TX_ACCEPTER case, `msg` could be a failure message */ + if (msg && (fromwire_peektype(msg) == WIRE_DUALOPEND_FAIL)) { + if (fromwire_dualopend_fail(msg, msg, &err_reason)) + msg = tal_free(msg); + } + + if (!msg) { + if (err_reason) + open_abort(state, "%s", err_reason); + else + open_abort(state, "%s", "Unable to commit"); + /* We need to 'reset' the channel to what it + * was before we did this. */ + return; + } + + if (state->our_role == TX_ACCEPTER) { handle_send_tx_sigs(state, msg); - else + } else wire_sync_write(REQ_FD, take(msg)); + + /* The replacement attempt is now committed. It is no longer possible + * to return to the previous funding state with tx_abort. */ + state->previous_tx_state = tal_free(state->previous_tx_state); +} + +static void promote_rbf_tx_state(struct state *state, + struct tx_state *tx_state) +{ + assert(!state->previous_tx_state); + state->previous_tx_state = state->tx_state; + state->tx_state = tal_steal(state, tx_state); } static void rbf_local_start(struct state *state, u8 *msg) @@ -3613,9 +3691,9 @@ static void rbf_local_start(struct state *state, u8 *msg) return; } - /* Promote tx_state */ - tal_free(state->tx_state); - state->tx_state = tal_steal(state, tx_state); + /* Promote the candidate, but retain the last completed state until this + * RBF commits: tx_abort must be able to return to it in place. */ + promote_rbf_tx_state(state, tx_state); /* Notify lightningd about require_confirmed state */ msg = towire_dualopend_update_require_confirmed(NULL, @@ -3788,9 +3866,9 @@ static void rbf_remote_start(struct state *state, const u8 *rbf_msg) peer_write(state->pps, msg); peer_billboard(false, "channel rbf: ack sent, waiting for reply"); - /* Promote tx_state */ - tal_free(state->tx_state); - state->tx_state = tal_steal(state, tx_state); + /* Promote the candidate, but retain the last completed state until this + * RBF commits: tx_abort must be able to return to it in place. */ + promote_rbf_tx_state(state, tx_state); /* We merge with RBF's we've initiated now */ rbf_wrap_up(state, tx_state, total); @@ -4097,6 +4175,8 @@ static u8 *handle_master_in(struct state *state) case WIRE_DUALOPEND_RBF_INIT: rbf_local_start(state, msg); return NULL; + case WIRE_DUALOPEND_READY: + break; case WIRE_DUALOPEND_SEND_TX_SIGS: handle_send_tx_sigs(state, msg); return NULL; @@ -4122,6 +4202,8 @@ static u8 *handle_master_in(struct state *state) case WIRE_DUALOPEND_VALIDATE_INPUTS_REPLY: /* Messages we send */ + case WIRE_DUALOPEND_INIT_READY: + case WIRE_DUALOPEND_ABORT_COMPLETE: case WIRE_DUALOPEND_COMMIT_READY: case WIRE_DUALOPEND_GOT_OFFER: case WIRE_DUALOPEND_GOT_RBF_OFFER: @@ -4288,9 +4370,18 @@ static void fetch_per_commitment_point(u32 point_count, u8 *msg; struct secret *none; - wire_sync_write(HSM_FD, - take(towire_hsmd_get_per_commitment_point(NULL, point_count))); + if (!wire_sync_write(HSM_FD, + take(towire_hsmd_get_per_commitment_point(NULL, + point_count)))) + status_failed(STATUS_FAIL_HSM_IO, + "Writing get_per_commitment_point: %s", + strerror(errno)); + errno = 0; msg = wire_sync_read(tmpctx, HSM_FD); + if (!msg) + status_failed(STATUS_FAIL_HSM_IO, + "Reading get_per_commitment_point reply: %s", + errno == 0 ? "EOF" : strerror(errno)); if (!fromwire_hsmd_get_per_commitment_point_reply(tmpctx, msg, commit_point, &none)) @@ -4327,6 +4418,8 @@ int main(int argc, char *argv[]) /* Init state to not aborted */ state->aborted_err = NULL; + state->abort_completed = false; + state->previous_tx_state = NULL; /*~ The very first thing we read from lightningd is our init msg */ msg = wire_sync_read(tmpctx, REQ_FD); @@ -4476,6 +4569,9 @@ int main(int argc, char *argv[]) * so we might as well get the hsm daemon to generate it now. */ fetch_per_commitment_point(0, &state->first_per_commitment_point[LOCAL]); fetch_per_commitment_point(1, &state->second_per_commitment_point[LOCAL]); + if (state->channel) + wire_sync_write(REQ_FD, + take(towire_dualopend_init_ready(NULL))); /*~ We manually run a little poll() loop here. With only two fds */ pollfd[0].fd = REQ_FD; @@ -4488,6 +4584,8 @@ int main(int argc, char *argv[]) do_reconnect_dance(state); state->reconnected = true; } + if (state->channel) + wire_sync_write(REQ_FD, take(towire_dualopend_ready(NULL))); /* We exit when we get a conclusion to write to lightningd: either * opening_funder_reply or opening_fundee. */ @@ -4509,6 +4607,10 @@ int main(int argc, char *argv[]) else if (pollfd[1].revents & POLLIN) msg = handle_peer_in(state); + /* In-place abort completion is only a guard while the current nested + * handler unwinds. A later event starts a fresh negotiation. */ + state->abort_completed = false; + /* If we've shutdown, we're done */ if (shutdown_complete(state)) msg = towire_dualopend_shutdown_complete(state); diff --git a/openingd/dualopend_wire.csv b/openingd/dualopend_wire.csv index d6b157495437..6ea2d4337e4d 100644 --- a/openingd/dualopend_wire.csv +++ b/openingd/dualopend_wire.csv @@ -81,6 +81,20 @@ msgdata,dualopend_reinit,we_require_confirmed_inputs,bool, msgdata,dualopend_reinit,they_require_confirmed_inputs,bool, msgdata,dualopend_reinit,local_alias,short_channel_id, +# dualopend->master: initialization and any reconnect dance are complete, so +# master requests may now be delivered to this owner. +msgtype,dualopend_ready,7032 + +# dualopend->master: a saved channel's tx_abort handshake is complete. The +# daemon has rolled back to the last completed funding attempt and remains the +# owner of the peer and HSM connections. +msgtype,dualopend_abort_complete,7035 +msgdata,dualopend_abort_complete,reason,wirestring, + +# dualopend->master: reinit and HSM setup are complete and the peer endpoint is +# open in this process, so master may attach connectd's socketpair endpoint. +msgtype,dualopend_init_ready,7034 + # dualopend->master: they offered channel, should we continue? msgtype,dualopend_got_offer,7005 msgdata,dualopend_got_offer,channel_id,channel_id, diff --git a/plugins/funder.c b/plugins/funder.c index 6ce91b6959de..b172e72ce3be 100644 --- a/plugins/funder.c +++ b/plugins/funder.c @@ -117,17 +117,6 @@ static struct command_result *unreserve_psbt(struct command *cmd, return command_still_pending(aux); } -static void cleanup_peer_pending_opens(struct command *cmd, - const struct node_id *id) -{ - struct pending_open *i, *next; - list_for_each_safe(&pending_opens, i, next, list) { - if (node_id_eq(&i->peer_id, id)) { - unreserve_psbt(cmd, i); - } - } -} - static struct command_result * command_hook_cont_psbt(struct command *cmd, struct wally_psbt *psbt) { @@ -1086,32 +1075,6 @@ json_rbf_channel_call(struct command *cmd, return send_outreq(req); } -static struct command_result *json_disconnect(struct command *cmd, - const char *buf, - const jsmntok_t *params) -{ - struct node_id id; - const char *err; - - err = json_scan(tmpctx, buf, params, - "{disconnect:{id:%}}", - JSON_SCAN(json_to_node_id, &id)); - if (err) - plugin_err(cmd->plugin, - "`disconnect` notification payload did not" - " scan %s: %.*s", - err, json_tok_full_len(params), - json_tok_full(buf, params)); - - plugin_log(cmd->plugin, LOG_DBG, - "Cleaning up inflights for peer id %s", - fmt_node_id(tmpctx, &id)); - - cleanup_peer_pending_opens(cmd, &id); - - return notification_handled(cmd); -} - static struct command_result * delete_channel_from_datastore(struct command *cmd, struct channel_id *cid) @@ -1552,10 +1515,6 @@ const struct plugin_notification notifs[] = { "channel_open_failed", json_channel_open_failed, }, - { - "disconnect", - json_disconnect, - }, { "channel_state_changed", json_channel_state_changed, diff --git a/tests/test_connection.py b/tests/test_connection.py index 4b07bda708c9..4356df89c014 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -478,13 +478,12 @@ def test_disconnect(node_factory): @pytest.mark.openchannel('v1') @pytest.mark.openchannel('v2') -# FIXME: https://github.com/ElementsProject/lightning/issues/8822 -@pytest.mark.flaky(reruns=1) def test_disconnect_opener(node_factory): # Now error on opener side during channel open. disconnects = ['-WIRE_OPEN_CHANNEL', '+WIRE_OPEN_CHANNEL', '-WIRE_FUNDING_CREATED'] + failing_disconnects = disconnects if EXPERIMENTAL_DUAL_FUND: disconnects = ['-WIRE_OPEN_CHANNEL2', '+WIRE_OPEN_CHANNEL2', @@ -494,16 +493,20 @@ def test_disconnect_opener(node_factory): '+WIRE_TX_ADD_OUTPUT', '-WIRE_TX_COMPLETE', '=WIRE_TX_COMPLETE'] + # '=' consumes the directive without disconnecting. It used to fail + # only because the preceding half-open was stranded by #8822. + failing_disconnects = disconnects[:-1] l1 = node_factory.get_node(disconnect=disconnects, may_reconnect=EXPERIMENTAL_DUAL_FUND, - options={'dev-no-reconnect': None}) + options={'dev-no-reconnect': None, + 'dual-open-disconnect-timeout': 1}) l2 = node_factory.get_node(may_reconnect=EXPERIMENTAL_DUAL_FUND, options={'dev-no-reconnect': None}) l1.fundwallet(2000000) - for d in disconnects: + for d in failing_disconnects: l1.rpc.connect(l2.info['id'], 'localhost', l2.port) with pytest.raises(RpcError): l1.rpc.fundchannel(l2.info['id'], CHANNEL_SIZE) @@ -621,22 +624,69 @@ def test_disconnect_half_signed(node_factory): @pytest.mark.openchannel('v2') def test_disconnect_half_signed_v2(node_factory): - # Now, these are the corner cases. - # L1 remembers the channel, L2 doesn't + # L1 remembers the channel and pending RPC during a reconnect grace + # period; L2 never received the final tx_complete and forgets it. disconnects = ['-WIRE_TX_COMPLETE'] - l1 = node_factory.get_node(disconnect=disconnects) + l1 = node_factory.get_node( + disconnect=disconnects, + options={'dual-open-disconnect-timeout': 1}) l2 = node_factory.get_node() l1.fundwallet(2000000) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - with pytest.raises(RpcError): - l1.rpc.fundchannel(l2.info['id'], CHANNEL_SIZE) + fund = node_factory.executor.submit(l1.rpc.fundchannel, + l2.info['id'], CHANNEL_SIZE) # Opener remembers, peer doesn't. - wait_for(lambda: l2.rpc.listpeers(l1.info['id'])['peers'] == []) - wait_for(lambda: only_one(l1.rpc.listpeers(l2.info['id'])['peers'])['connected'] is False) + l1.daemon.wait_for_log('to DUALOPEND_OPEN_COMMIT_READY') + wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) assert len(l1.rpc.listpeerchannels(l2.info['id'])['channels']) == 1 + assert l2.rpc.listpeerchannels(l1.info['id'])['channels'] == [] + + # With reconnection disabled by the harness, the grace period must bound + # both the RPC lifetime and retention of the safe-to-forget channel. + with pytest.raises(RpcError, match='did not reconnect'): + fund.result(timeout=TIMEOUT) + wait_for(lambda: l1.rpc.listpeerchannels(l2.info['id'])['channels'] == []) + + +@pytest.mark.openchannel('v2') +def test_disconnect_half_signed_v2_reconnect(node_factory): + """The saved opener must escape an unknown-channel reconnect loop (#8822).""" + l1, l2 = node_factory.get_nodes(2, opts=[ + {'disconnect': ['-WIRE_TX_COMPLETE'], + 'may_reconnect': True, + 'dev-no-reconnect': None}, + {'may_reconnect': True, + 'dev-no-reconnect': None} + ]) + + l1.fundwallet(2000000) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + fund = node_factory.executor.submit(l1.rpc.fundchannel, + l2.info['id'], CHANNEL_SIZE) + + # l1 sent the final tx_complete and saved the channel; l2 did not receive + # it and discarded its unsaved channel when the connection failed. + l1.daemon.wait_for_log('to DUALOPEND_OPEN_COMMIT_READY') + wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) + assert len(l1.rpc.listpeerchannels(l2.info['id'])['channels']) == 1 + assert l2.rpc.listpeerchannels(l1.info['id'])['channels'] == [] + + # l1 now sends channel_reestablish for a channel unknown to l2. The error + # must reach l1 so it can forget the stale channel, rather than both peers + # repeatedly reconnecting and immediately disconnecting. + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + l2.daemon.wait_for_log(r'Unknown channel .* for WIRE_CHANNEL_REESTABLISH') + wait_for(lambda: l1.rpc.listpeerchannels(l2.info['id'])['channels'] == [], + timeout=10) + + # Once the stale channel is gone, a fresh transport should remain usable. + with pytest.raises(RpcError): + fund.result(timeout=TIMEOUT) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + assert l1.rpc.getpeer(l2.info['id'])['connected'] @pytest.mark.openchannel('v1') diff --git a/tests/test_opening.py b/tests/test_opening.py index e23283f7251c..f854f69956a5 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -2,7 +2,7 @@ from fixtures import TEST_NETWORK from pyln.client import RpcError, Millisatoshi from utils import ( - only_one, wait_for, sync_blockheight, first_channel_id, calc_lease_fee, check_coin_moves + TIMEOUT, only_one, wait_for, sync_blockheight, first_channel_id, calc_lease_fee, check_coin_moves ) from pyln.testing.utils import FUNDAMOUNT @@ -157,9 +157,9 @@ def test_v2_open_sigs_reconnect_2(node_factory, bitcoind): # Wait for it to arrive. wait_for(lambda: len(l1.rpc.listfunds()['outputs']) > 0) - # Fund the channel, should disconnect after getting l2's sigs - with pytest.raises(RpcError): - l1.rpc.fundchannel(l2.info['id'], chan_amount) + # Fund the channel. The transport drops after getting l2's sigs, but the + # RPC is retained while dualopend reconnects and resends our signatures. + l1.rpc.fundchannel(l2.info['id'], chan_amount) # peer reconnects, and we resend our sigs l1.daemon.wait_for_log('Peer has reconnected, state DUALOPEND_OPEN_COMMITTED') @@ -216,6 +216,7 @@ def _fund(): # Corrupt l2's stored funding txid so it disagrees with l1's on reconnect. l2.stop() + wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) l2.db_manip("UPDATE channel_funding_inflights SET funding_tx_id = X'{}'".format('01' * 32)) l2.start() @@ -369,9 +370,9 @@ def test_v2_open_sigs_restart_while_dead(node_factory, bitcoind): # Wait for it to arrive. wait_for(lambda: len(l1.rpc.listfunds()['outputs']) > 0) - # Make a channel happen, with multiple disconnects! - with pytest.raises(RpcError): - l1.rpc.fundchannel(l2.info['id'], chan_amount) + # Make a channel happen, with multiple disconnects. The RPC survives the + # reconnects and completes once the funding transaction is broadcast. + l1.rpc.fundchannel(l2.info['id'], chan_amount) l1.daemon.wait_for_log('Broadcasting funding tx') l1.daemon.wait_for_log('sendrawtx exit 0') @@ -464,7 +465,15 @@ def test_v2_rbf_single(node_factory, bitcoind, chainparams): update = l1.rpc.openchannel_update(chan_id, bump['psbt']) assert update['commitments_secured'] signed_psbt = l1.rpc.signpsbt(update['psbt'])['signed_psbt'] - l1.rpc.openchannel_signed(chan_id, signed_psbt) + signed_fut = node_factory.executor.submit(l1.rpc.openchannel_signed, + chan_id, signed_psbt) + wait_for(lambda: + signed_fut.done() + or (not l1.rpc.getpeer(l2.info['id'])['connected'] + and not l2.rpc.getpeer(l1.info['id'])['connected'])) + if not signed_fut.done(): + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + signed_fut.result(timeout=TIMEOUT) bitcoind.generate_block(1) sync_blockheight(bitcoind, [l1]) @@ -808,14 +817,19 @@ def test_v2_rbf_multi(node_factory, bitcoind, chainparams): @unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') @pytest.mark.openchannel('v2') -def test_rbf_reconnect_init(node_factory, bitcoind, chainparams): - disconnects = ['-WIRE_TX_INIT_RBF', - '+WIRE_TX_INIT_RBF'] +@pytest.mark.parametrize('disconnect', [ + '-WIRE_TX_INIT_RBF', + '+WIRE_TX_INIT_RBF', +]) +def test_rbf_reconnect_init(node_factory, bitcoind, chainparams, disconnect): l1, l2 = node_factory.get_nodes(2, - opts=[{'disconnect': disconnects, - 'may_reconnect': True}, - {'may_reconnect': True}]) + opts=[{'disconnect': [disconnect], + 'may_reconnect': True, + 'dev-no-reconnect': None, + 'dual-open-disconnect-timeout': 3}, + {'may_reconnect': True, + 'dev-no-reconnect': None}]) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) amount = 2**24 @@ -842,28 +856,36 @@ def test_rbf_reconnect_init(node_factory, bitcoind, chainparams): prev_utxos, reservedok=True, excess_as_change=True) - # Do the bump!? - for d in disconnects: - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - with pytest.raises(RpcError): - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) - assert l1.rpc.getpeer(l2.info['id']) is not None - - # This should succeed + # The serialized bump remains pending while dualopend and its transport + # are replaced. Reconnect before the retained-command timeout and expect + # this same RPC, rather than a caller retry, to complete. + bump_fut = node_factory.executor.submit(l1.rpc.openchannel_bump, + chan_id, chan_amount, + initpsbt['psbt']) + l1.daemon.wait_for_log(r'dev_disconnect: ' + re.escape(disconnect)) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) + bump = bump_fut.result(timeout=TIMEOUT) + assert bump['channel_id'] == chan_id @unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') @pytest.mark.openchannel('v2') -def test_rbf_reconnect_ack(node_factory, bitcoind, chainparams): - disconnects = ['-WIRE_TX_ACK_RBF', - '+WIRE_TX_ACK_RBF'] +@pytest.mark.parametrize('disconnect', [ + '-WIRE_TX_ACK_RBF', + '+WIRE_TX_ACK_RBF', +]) +def test_rbf_reconnect_ack(node_factory, bitcoind, chainparams, disconnect): l1, l2 = node_factory.get_nodes(2, - opts=[{'may_reconnect': True}, - {'disconnect': disconnects, - 'may_reconnect': True}]) + opts=[{'may_reconnect': True, + 'dev-no-reconnect': None, + 'dual-open-disconnect-timeout': 3}, + {'disconnect': [disconnect], + 'may_reconnect': True, + 'dev-no-reconnect': None}]) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) amount = 2**24 @@ -890,26 +912,113 @@ def test_rbf_reconnect_ack(node_factory, bitcoind, chainparams): prev_utxos, reservedok=True, excess_as_change=True) - # Do the bump!? - for d in disconnects: - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - with pytest.raises(RpcError): - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) - assert l1.rpc.getpeer(l2.info['id']) is not None + bump_fut = node_factory.executor.submit(l1.rpc.openchannel_bump, + chan_id, chan_amount, + initpsbt['psbt']) + l2.daemon.wait_for_log(r'dev_disconnect: ' + re.escape(disconnect)) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + bump = bump_fut.result(timeout=TIMEOUT) + assert bump['channel_id'] == chan_id + + +def _setup_rbf_reconnect_tx_add(node_factory, bitcoind, disconnects): + """Open a channel and prepare an RBF for TX_ADD reconnect tests.""" + l1, l2 = node_factory.get_nodes(2, opts=[ + {'disconnect': disconnects, + 'may_reconnect': True, + 'dev-no-reconnect': None, + 'dual-open-disconnect-timeout': 3}, + {'may_reconnect': True, + 'dev-no-reconnect': None} + ]) - # This should succeed l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) + amount = 2**24 + chan_amount = 100000 + bitcoind.rpc.sendtoaddress(l1.rpc.newaddr()['p2tr'], + amount / 10**8 + 0.01) + bitcoind.generate_block(1) + wait_for(lambda: len(l1.rpc.listfunds()['outputs']) > 0) + + res = l1.rpc.fundchannel(l2.info['id'], chan_amount) + chan_id = res['channel_id'] + vin = only_one(bitcoind.rpc.decoderawtransaction(res['tx'])['vin']) + prev_utxos = ["{}:{}".format(vin['txid'], vin['vout'])] + l1.daemon.wait_for_log(' to DUALOPEND_AWAITING_LOCKIN') + + rate = int(find_next_feerate(l1, l2)[:-5]) + initpsbt = l1.rpc.utxopsbt(chan_amount, '{}perkw'.format(rate * 4), + 42 + 172, prev_utxos, reservedok=True, + excess_as_change=True) + + return l1, l2, chan_amount, chan_id, initpsbt + + +@unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') +@pytest.mark.openchannel('v2') +@pytest.mark.parametrize('tx_add_disconnect', [ + '-WIRE_TX_ADD_INPUT', + '+WIRE_TX_ADD_INPUT', + '-WIRE_TX_ADD_OUTPUT', + '+WIRE_TX_ADD_OUTPUT', +]) +def test_rbf_reconnect_tx_add(node_factory, bitcoind, chainparams, + tx_add_disconnect): + """A retained bump survives a disconnect at each interactive-tx add.""" + # Consume the initial funding transaction's add messages before arming the + # fault for the later RBF construction. + disconnects = ['=WIRE_TX_ADD_INPUT', '=WIRE_TX_ADD_OUTPUT*2', + tx_add_disconnect] + l1, l2, chan_amount, chan_id, initpsbt = \ + _setup_rbf_reconnect_tx_add(node_factory, bitcoind, disconnects) + + bump_fut = node_factory.executor.submit(l1.rpc.openchannel_bump, + chan_id, chan_amount, + initpsbt['psbt']) + l1.daemon.wait_for_log(r'dev_disconnect: ' + + re.escape(tx_add_disconnect)) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + bump = bump_fut.result(timeout=TIMEOUT) + assert bump['channel_id'] == chan_id + + +@unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') +@pytest.mark.openchannel('v2') +def test_rbf_reconnect_tx_add_repeated_generations(node_factory, bitcoind, + chainparams): + """A retained bump survives repeated owner and connection generations.""" + generations = 4 + fault = '-WIRE_TX_ADD_INPUT' + disconnects = ['=WIRE_TX_ADD_INPUT', '=WIRE_TX_ADD_OUTPUT*2'] \ + + [fault] * generations + l1, l2, chan_amount, chan_id, initpsbt = \ + _setup_rbf_reconnect_tx_add(node_factory, bitcoind, disconnects) + + bump_fut = node_factory.executor.submit(l1.rpc.openchannel_bump, + chan_id, chan_amount, + initpsbt['psbt']) + for _ in range(generations): + l1.daemon.wait_for_log(r'dev_disconnect: ' + re.escape(fault)) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + + bump = bump_fut.result(timeout=TIMEOUT) + assert bump['channel_id'] == chan_id @unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') @pytest.mark.openchannel('v2') def test_rbf_reconnect_tx_construct(node_factory, bitcoind, chainparams): disconnects = ['=WIRE_TX_ADD_INPUT', # Initial funding succeeds - '-WIRE_TX_ADD_INPUT', - '+WIRE_TX_ADD_INPUT', - '-WIRE_TX_ADD_OUTPUT', - '+WIRE_TX_ADD_OUTPUT', + '=WIRE_TX_COMPLETE', '-WIRE_TX_COMPLETE', '+WIRE_TX_COMPLETE', '-WIRE_COMMITMENT_SIGNED', @@ -918,10 +1027,10 @@ def test_rbf_reconnect_tx_construct(node_factory, bitcoind, chainparams): l1, l2 = node_factory.get_nodes(2, opts=[{'disconnect': disconnects, 'may_reconnect': True, - 'dev-no-reconnect': None}, - {'may_reconnect': True, 'dev-no-reconnect': None, - 'broken_log': 'dualopend daemon died before signed PSBT returned'}]) + 'dual-open-disconnect-timeout': 3}, + {'may_reconnect': True, + 'dev-no-reconnect': None}]) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) amount = 2**24 @@ -951,20 +1060,17 @@ def test_rbf_reconnect_tx_construct(node_factory, bitcoind, chainparams): prev_utxos, reservedok=True, excess_as_change=True) - # Run through TX_ADD wires - for d in disconnects[1:-4]: - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - with pytest.raises(RpcError): - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) - wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) + bump = l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) # The first TX_COMPLETE breaks + update_fut = node_factory.executor.submit(l1.rpc.openchannel_update, + chan_id, bump['psbt']) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - bump = l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) with pytest.raises(RpcError): - update = l1.rpc.openchannel_update(chan_id, bump['psbt']) - wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + update_fut.result(timeout=TIMEOUT) # l1 should remember, l2 has forgotten # l2 should send tx-abort, to reset l2.daemon.wait_for_log(r'tx-abort: Sent next_funding_txid .* doesn\'t match ours .*') @@ -977,29 +1083,42 @@ def test_rbf_reconnect_tx_construct(node_factory, bitcoind, chainparams): # The next TX_COMPLETE break (both remember) + they break on the # COMMITMENT_SIGNED during the reconnect bump = l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) - with pytest.raises(RpcError): - update = l1.rpc.openchannel_update(chan_id, bump['psbt']) - wait_for(lambda: l1.rpc.getpeer(l2.info['id'])['connected'] is False) + update_fut = node_factory.executor.submit(l1.rpc.openchannel_update, + chan_id, bump['psbt']) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + with pytest.raises(RpcError): + update_fut.result(timeout=TIMEOUT) l2.daemon.wait_for_logs([r'Got dualopend reestablish', r'No commitment, not sending our sigs']) l1.daemon.wait_for_logs([r'Got dualopend reestablish', r'No commitment, not sending our sigs', r'dev_disconnect: -WIRE_COMMITMENT_SIGNED', 'peer_disconnected']) - assert not l1.rpc.getpeer(l2.info['id'])['connected'] + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) l1.rpc.connect(l2.info['id'], 'localhost', l2.port) # COMMITMENT_SIGNED disconnects *during* the reconnect # We can't bump because the last negotiation is in the wrong state + bump_fut = node_factory.executor.submit(l1.rpc.openchannel_bump, + chan_id, chan_amount, + initpsbt['psbt']) with pytest.raises(RpcError, match=r'Funding sigs for this channel not secured'): - l1.rpc.openchannel_bump(chan_id, chan_amount, initpsbt['psbt']) + bump_fut.result(timeout=TIMEOUT) # l2 reconnects, but doesn't have l1's commitment + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] is False + and l2.rpc.getpeer(l1.info['id'])['connected'] is False) + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) l2.daemon.wait_for_logs([r'Got dualopend reestablish', r'No commitment, not sending our sigs', - # This is a BROKEN log, it's expected! - r'dualopend daemon died before signed PSBT returned|dualopend: Owning subdaemon dualopend died', - r'Owning subdaemon dualopend died']) + # The injected transport failure retires this + # owner cleanly; it is not a BROKEN condition. + r'dualopend: Owning subdaemon dualopend died']) # If we received their commitment_signed first, we *will* have scratch! inflights = only_one(l1.rpc.listpeerchannels()['channels'])['inflight'] @@ -1009,9 +1128,25 @@ def test_rbf_reconnect_tx_construct(node_factory, bitcoind, chainparams): else: assert 'scratch_txid' not in inflights[1] - # After reconnecting, we have a scratch txid! - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) - wait_for(lambda: 'scratch_txid' in only_one(l1.rpc.listpeerchannels()['channels'])['inflight'][1]) + # Depending on which commitment_signed arrives first, that reestablish + # owner may produce scratch state before its injected transport failure + # has finished disconnecting. Wait for that owner to retire completely + # before installing its replacement. + def have_scratch_txid(): + inflights = only_one(l1.rpc.listpeerchannels()['channels'])['inflight'] + return len(inflights) > 1 and 'scratch_txid' in inflights[1] + + l1.daemon.wait_for_logs([r'dev_disconnect: \+WIRE_COMMITMENT_SIGNED', + 'peer_disconnected']) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] + == l2.rpc.getpeer(l1.info['id'])['connected']) + if not l1.rpc.getpeer(l2.info['id'])['connected']: + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + wait_for(lambda: + l1.rpc.getpeer(l2.info['id'])['connected'] + and l2.rpc.getpeer(l1.info['id'])['connected']) + wait_for(have_scratch_txid) # We can call update again! It should short-circuit this time :) update = l1.rpc.openchannel_update(chan_id, bump['psbt']) @@ -1086,14 +1221,13 @@ def test_rbf_reconnect_tx_sigs(node_factory, bitcoind, chainparams): # Sign our inputs, and continue signed_psbt = l1.rpc.signpsbt(update['psbt'])['signed_psbt'] - # First time we error when we send our sigs - with pytest.raises(RpcError): - l1.rpc.openchannel_signed(chan_id, signed_psbt) + # Sending our signatures triggers the disconnect sequence. The RPC is + # retained while dualopend reconnects and completes the exchange. + l1.rpc.openchannel_signed(chan_id, signed_psbt) # Absolute chaos ensues as these guys disconnect/reconnect # when sending tx-sigs. By the end, both should have # broadcast a funding tx. - l1.rpc.connect(l2.info['id'], 'localhost', l2.port) l1.daemon.wait_for_log('Broadcasting funding tx') l2.daemon.wait_for_log('Broadcasting funding tx') @@ -1115,7 +1249,8 @@ def test_rbf_to_chain_before_commit(node_factory, bitcoind, chainparams): '-WIRE_COMMITMENT_SIGNED'] l1, l2 = node_factory.get_nodes(2, opts=[{'may_reconnect': True, - 'dev-no-reconnect': None}, + 'dev-no-reconnect': None, + 'dual-open-disconnect-timeout': 3}, {'disconnect': disconnects, 'may_reconnect': True, 'dev-no-reconnect': None}]) @@ -3076,7 +3211,49 @@ def test_zeroconf_withhold(node_factory, bitcoind, stay_withheld, mutual_close): assert bitcoind.rpc.getrawmempool() == [] if mutual_close: - l1.connect(l2) + # connect() only waits for the peer_connected hook. A transport can + # still fail while the replacement channelds exchange reestablish, so + # retry that connection generation rather than queueing shutdown on a + # half-installed route. + established = False + for _ in range(3): + last_l1 = len(l1.daemon.logs) + last_l2 = len(l2.daemon.logs) + + def reestablished(): + return ( + l1.daemon.is_in_log( + r'peer_in WIRE_CHANNEL_REESTABLISH', start=last_l1 + ) + and l2.daemon.is_in_log( + r'peer_in WIRE_CHANNEL_REESTABLISH', start=last_l2 + ) + ) + + l1.connect(l2) + try: + wait_for(lambda: reestablished() + or not l1.rpc.getpeer(l2.info['id'])['connected'] + or not l2.rpc.getpeer(l1.info['id'])['connected'], + timeout=5) + except ValueError: + # A failed route can leave both transports nominally + # connected without either channeld receiving reestablish. + pass + if reestablished(): + established = True + break + # One connectd can retain its half of a failed generation after + # the other side has dropped. Retire it before retrying, or it + # will reject/absorb the replacement transport. + if l1.rpc.getpeer(l2.info['id'])['connected']: + l1.rpc.disconnect(l2.info['id'], force=True) + if l2.rpc.getpeer(l1.info['id'])['connected']: + l2.rpc.disconnect(l1.info['id'], force=True) + wait_for(lambda: + not l1.rpc.getpeer(l2.info['id'])['connected'] + and not l2.rpc.getpeer(l1.info['id'])['connected']) + assert established if not stay_withheld: # sendpsbt marks it as no longer withheld. @@ -3220,3 +3397,42 @@ def test_no_retransmit_confirmed_funding(node_factory): # Should not have attempted (and failed) to re-broadcast the funding tx. assert not l1.daemon.is_in_log('Failed to re-transmit funding tx') assert not l1.daemon.is_in_log('Successfully rexmitted funding tx') + + +@unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') +@pytest.mark.openchannel('v2') +def test_inflight_disconnect_commitment_v2(node_factory, bitcoind): + """Disconnect during dual-fund commitment signing should not trigger spurious BROKEN messages. + """ + disconnects = ["+WIRE_COMMITMENT_SIGNED"] + + opts = [{'experimental-dual-fund': None, 'dev-no-reconnect': None, + 'may_reconnect': True, 'disconnect': disconnects}, + {'experimental-dual-fund': None, 'dev-no-reconnect': None, + 'may_reconnect': True}] + + opener, funder = node_factory.get_nodes(2, opts=opts) + + amount = 500000 + opener.fundwallet(20000000) + funder.fundwallet(20000000) + + funder.rpc.call('funderupdate', + {'policy': 'available', + 'policy_mod': 100, + 'per_channel_max_msat': '1btc', + 'reserve_tank_msat': '0msat', + 'fund_probability': 100, + 'fuzz_percent': 0, + 'leases_only': False}) + + opener.rpc.connect(funder.info['id'], 'localhost', funder.port) + fut = node_factory.executor.submit(opener.rpc.fundchannel, + funder.info['id'], amount) + + opener.daemon.wait_for_log(r'dev_disconnect: .WIRE_COMMITMENT_SIGNED') + wait_for(lambda: + opener.rpc.getpeer(funder.info['id'])['connected'] is False + and funder.rpc.getpeer(opener.info['id'])['connected'] is False) + opener.rpc.connect(funder.info['id'], 'localhost', funder.port) + fut.result(timeout=TIMEOUT) diff --git a/wallet/test/run-wallet.c b/wallet/test/run-wallet.c index c44307710943..50052f06163f 100644 --- a/wallet/test/run-wallet.c +++ b/wallet/test/run-wallet.c @@ -110,6 +110,20 @@ u64 channel_mvt_index_created(struct lightningd *ld UNNEEDED, /* Generated stub for channel_unsaved_close_conn */ void channel_unsaved_close_conn(struct channel *channel UNNEEDED, const char *why UNNEEDED) { fprintf(stderr, "channel_unsaved_close_conn called!\n"); abort(); } +/* Generated stub for dual_open_attempt_peer_disconnected */ +void dual_open_attempt_peer_disconnected(struct channel *channel UNNEEDED) +{ +} +/* Generated stub for dual_open_attempt_waiting_for_owner */ +bool dual_open_attempt_waiting_for_owner(const struct channel *channel UNNEEDED) +{ + return false; +} +/* Generated stub for dual_open_owner_begin_retirement */ +void dual_open_owner_begin_retirement(struct channel *channel UNNEEDED, + struct subd *retiring_owner UNNEEDED) +{ +} /* Generated stub for channel_update_details */ bool channel_update_details(const u8 *channel_update UNNEEDED, u32 *timestamp UNNEEDED,