From b5a5acb44a0ad14fcd1392f16b7851da8cfebd5e Mon Sep 17 00:00:00 2001 From: niftynei Date: Mon, 24 Aug 2026 19:05:58 -0500 Subject: [PATCH 01/20] ccan/io: retain registrations during poll dispatch A callback can close a connection while readiness from the current poll call is still being dispatched. Deleting that connection compacts the fd table, and registering a replacement can restore its previous length with different occupants. A snapshot tied to mutable table positions can consequently skip an original event or deliver stale readiness to the replacement. Give each fd a stable registration object and retain that registration while a poll result refers to it. Deletion clears the registration's fd pointer, so pending readiness for a retired connection is ignored, while a replacement receives a distinct registration and cannot inherit the old event. Snapshot only entries with nonzero revents and preserve the IO_ALWAYS marker in the existing fairness order. Reuse a growable snapshot buffer between loop iterations and dispatch directly through the retained registration, avoiding both a fresh allocation on every poll and the repeated linear lookup required by the earlier generation-based approach. Register one process-exit cleanup for that reusable buffer. Retaining it for the daemon lifetime preserves allocation reuse, while releasing it after event dispatch can no longer be active avoids a reachable allocation in Valgrind and LeakSanitizer runs. Changelog-Fixed: io: replacing a file descriptor during poll dispatch no longer delivers stale readiness events to the replacement connection. --- ccan/ccan/io/backend.h | 2 + ccan/ccan/io/io.h | 5 ++ ccan/ccan/io/poll.c | 169 +++++++++++++++++++++++++++++++++++++++- connectd/connectd.c | 1 + lightningd/lightningd.c | 1 + 5 files changed, 177 insertions(+), 1 deletion(-) 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/connectd/connectd.c b/connectd/connectd.c index cb95ef9b0cac..9e96a31ec8f9 100644 --- a/connectd/connectd.c +++ b/connectd/connectd.c @@ -2532,6 +2532,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/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); From aedb1455934fe7dde24a33042f7566e180592465 Mon Sep 17 00:00:00 2001 From: niftynei Date: Mon, 24 Aug 2026 17:12:02 -0500 Subject: [PATCH 02/20] ccan/io: test fd replacement during poll dispatch Closing a connection while dispatching poll results compacts the fd table. If the callback immediately registers a replacement, the table can return to its original length with a different connection occupying a slot represented in the current poll result. Dispatching readiness by mutable table position can then deliver an old event to the replacement or skip another ready connection. Use a fake poll implementation to make three original connections ready. The connection selected first closes itself and installs a replacement while the other readiness events remain pending. Verify that every original event is delivered exactly once, that the replacement does not inherit stale readiness, and that it is handled normally by the following poll call. Register an exit assertion before enabling protection so atexit LIFO ordering checks that the backend releases its reusable snapshot buffer at process exit. This keeps leak-sensitive unit runs from accepting a permanently reachable allocation. Add the regression to check-units so this CCAN behavior is covered by the project's normal unit-test suite. Changelog-None --- ccan/ccan/io/test/run-49-stale-fd-readiness.c | 142 ++++++++++++++++++ common/test/Makefile | 9 +- 2 files changed, 148 insertions(+), 3 deletions(-) create mode 100644 ccan/ccan/io/test/run-49-stale-fd-readiness.c 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/%) From aedac156a26371691499fa4668a7b6d0da829c94 Mon Sep 17 00:00:00 2001 From: Vincenzo Palazzo Date: Wed, 25 Mar 2026 14:51:40 +0100 Subject: [PATCH 03/20] tests: add regression test for #8902 dual-fund disconnect Add test_inflight_disconnect_commitment_v2 which triggers a disconnect at +WIRE_COMMITMENT_SIGNED during a dual-funded open. Changelog-None --- tests/test_opening.py | 42 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index e23283f7251c..f67ea053ab7b 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 @@ -3220,3 +3220,43 @@ 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') +@pytest.mark.xfail( + reason="Issue #8902 is not fixed at this commit", + strict=True, +) +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') + opener.rpc.connect(funder.info['id'], 'localhost', funder.port) + fut.result(timeout=TIMEOUT) From 556a082a8c3fb9fd9fa9c8b0eda4a3f78ced83fa Mon Sep 17 00:00:00 2001 From: niftynei Date: Sat, 8 Aug 2026 16:35:57 -0500 Subject: [PATCH 04/20] dualopend: Parse error messages returned from lightningd correctly The error message in #8902 indicates that we're failing to correctly parse an error message from lightningd lightningd-2 2026-02-16T00:50:21.721Z **BROKEN** 038194b5f32bdf0aa59812c86c4ef7ad2f294104fa027d1ace9b469bb6f88cf37b-dualopend-chan#2: STATUS_FAIL_MASTER_IO: Error parsing 7011: 1b5b50656572206572726f7220776974682050534254207369676e6174757265732e00 The openchannel2_sign_hook_cb in lightningd can return error messages, not just the DUALOPEND_SEND_TX_SIGS message at this point. We handle this here. Changelog-Fixed: dualopend: dual-funding signing-hook errors are now reported correctly instead of causing a `dualopend` master-reply parse failure. --- openingd/dualopend.c | 66 ++++++++++++++++++++++++++++++-------------- 1 file changed, 46 insertions(+), 20 deletions(-) diff --git a/openingd/dualopend.c b/openingd/dualopend.c index 2f1849773496..978d8c95f304 100644 --- a/openingd/dualopend.c +++ b/openingd/dualopend.c @@ -552,6 +552,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 +2331,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 +2760,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); @@ -3441,19 +3461,25 @@ 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)); } From 392c3633dff8455848969a245ff6d9d679272f2c Mon Sep 17 00:00:00 2001 From: niftynei Date: Sat, 8 Aug 2026 18:11:43 -0500 Subject: [PATCH 05/20] funder: remove listener for peer disconnects Issue #8902 demonstrates that there are races conditions ocurring when we use the peer disconnection notifications. In theory, we don't actually need to listen for peer disconnects, as we're already listening for open attempt failures with both the state_change and the channel_open_failed notifications. Changelog-Fixed: funder: peer disconnects no longer race channel-open failure handling when cleaning up pending dual-funded opens. --- plugins/funder.c | 41 ----------------------------------------- tests/test_opening.py | 4 ---- 2 files changed, 45 deletions(-) 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_opening.py b/tests/test_opening.py index f67ea053ab7b..5fac530779d3 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -3224,10 +3224,6 @@ def test_no_retransmit_confirmed_funding(node_factory): @unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need') @pytest.mark.openchannel('v2') -@pytest.mark.xfail( - reason="Issue #8902 is not fixed at this commit", - strict=True, -) def test_inflight_disconnect_commitment_v2(node_factory, bitcoind): """Disconnect during dual-fund commitment signing should not trigger spurious BROKEN messages. """ From 79bfe7c5cbe30f5c95435da68c0e999c3e6898e0 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:09:45 -0500 Subject: [PATCH 06/20] dualopend: free penalty base on commit failure opener_commits allocates a temporary penalty base before negotiating the commitment and validating the remote signature. Both failure paths revert the channel state and return without transferring ownership, so they must release that allocation explicitly. Changelog-Fixed: dualopend: failed commitment negotiations no longer leak their temporary penalty-base allocation. --- openingd/dualopend.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/openingd/dualopend.c b/openingd/dualopend.c index 978d8c95f304..169311273e08 100644 --- a/openingd/dualopend.c +++ b/openingd/dualopend.c @@ -2928,6 +2928,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; } @@ -2937,6 +2938,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; } From d470471ea0518af295a5fb22aebab099c06c5076 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:09:59 -0500 Subject: [PATCH 07/20] dualopend: report HSM transport failures explicitly fetch_per_commitment_point previously ignored a failed write and passed a NULL read result to the wire decoder. An HSM disconnect could therefore appear as a malformed reply or crash instead of exposing the underlying transport failure. Check both sides of the synchronous HSM exchange and fail with STATUS_FAIL_HSM_IO at the point of failure. Clear errno before reading so a NULL result reports EOF rather than an unrelated stale error. Changelog-Fixed: dualopend: HSM transport failures are now reported directly instead of appearing as malformed replies or crashes. --- openingd/dualopend.c | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/openingd/dualopend.c b/openingd/dualopend.c index 169311273e08..3d1023cacdd8 100644 --- a/openingd/dualopend.c +++ b/openingd/dualopend.c @@ -4316,9 +4316,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)) From eb39e182ef1a7d7767a89d82f122bbcc68201994 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:10:26 -0500 Subject: [PATCH 08/20] lightningd: retire openingd before starting channeld hsmd permits only one client for a channel database ID. Starting channeld before the completed openingd owner has exited can therefore race creation of the replacement HSM client, particularly when several v1 opens finish at the same time. Release openingd after preserving its peer endpoint and before asking channeld to acquire the channel's HSM client. This makes owner exit an explicit barrier without blocking the handling of other completed opens. Changelog-Fixed: openingd: completed v1 channel opens no longer race `openingd` teardown against `channeld` HSM client creation. --- lightningd/opening_control.c | 14 ++++++++++++++ 1 file changed, 14 insertions(+) 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)); From 5a863821d2669072a8207648aa666eb04027f779 Mon Sep 17 00:00:00 2001 From: niftynei Date: Sat, 8 Aug 2026 20:50:28 -0500 Subject: [PATCH 09/20] dualopend: delay re-start of dualopend after an abort to prevent HSM race Flake in test_rbf_reconnect_tx_construct indicates that there's a race in the teardown of the old dualopend/HSM code, which causes mismatches in the inflights and BROKEN HSM messages. **BROKEN** dualopend-chan#1: STATUS_FAIL_HSM_IO: Bad get_per_commitment_point_reply Changelog-Fixed: dualopend: restarting `dualopend` after an RBF abort now waits for the previous HSM client to exit, avoiding commitment-point reply mismatches. --- lightningd/dual_open_control.c | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/lightningd/dual_open_control.c b/lightningd/dual_open_control.c index 451cf1f94246..2bfe56643131 100644 --- a/lightningd/dual_open_control.c +++ b/lightningd/dual_open_control.c @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -27,6 +28,7 @@ #include #include #include +#include #include #include #include @@ -2477,6 +2479,20 @@ static char *restart_dualopend(const tal_t *ctx, const struct lightningd *ld, return NULL; } +static void restart_dualopend_after_abort(struct channel *channel) +{ + char *err; + + /* A disconnect or another restart may have happened after the abort. */ + if (channel->owner || channel->peer->connected != PEER_CONNECTED) + return; + + err = restart_dualopend(tmpctx, channel->peer->ld, channel, true); + if (err) + log_broken(channel->log, + "Unable to restart dualopend after abort: %s", err); +} + struct openchannel_bump_info { struct command *cmd; struct channel_id *cid; @@ -4063,7 +4079,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 +4086,13 @@ 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); + /* We are called while the old dualopend's terminal + * status is still being handled. Defer replacement + * until it has been destroyed and its HSM client has + * gone away. */ + new_reltimer(channel->peer->ld->timers, + channel, time_from_msec(0), + restart_dualopend_after_abort, channel); } return; From 9670b19fe0042cc38bbecbacbf18930813ef69d7 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:10:11 -0500 Subject: [PATCH 10/20] connectd: acknowledge replacement subdaemon routes After a dualopend route receives tx_abort, connectd hides it from normal dispatch and schedules it to close. If lightningd starts a replacement owner during that interval, it previously had no way to know when connectd had installed the replacement route. The new owner could begin its peer protocol while the route was absent or still belonged to the retiring owner. Add a tracked attachment request and defer its reply until io_new_conn has installed the replacement duplex plan. Validate the peer generation and connection state before accepting it, and report rejection to lightningd so it does not start an owner against a half-live route. A replacement may instead inherit the existing endpoint that was quarantined after tx_abort. Add a resume request which finds that exact route, cancels its close timer, and makes it eligible for dispatch again. Keep ordinary subdaemon attachment on the existing unacknowledged path, so the synchronization barrier applies only to replacement dualopend routes. Changelog-Fixed: connectd: replacement `dualopend` routes are now activated only after `connectd` confirms installation, preventing reconnects from using missing or retiring routes. --- connectd/connectd.c | 10 +- connectd/connectd_wire.csv | 22 +++++ connectd/multiplex.c | 188 ++++++++++++++++++++++++++++++++----- connectd/multiplex.h | 6 +- 4 files changed, 201 insertions(+), 25 deletions(-) diff --git a/connectd/connectd.c b/connectd/connectd.c index 9e96a31ec8f9..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: 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); From 05456c242be89b7eb9051cadaefa75ffe3f38708 Mon Sep 17 00:00:00 2001 From: niftynei Date: Tue, 25 Aug 2026 11:51:22 -0500 Subject: [PATCH 11/20] dualopend: resume saved opens across disconnects Persist enough dual-open state to recreate owners and continue outstanding commands when their peer reconnects. Serialize replay behind retirement of the previous owner and acknowledgement of the replacement connectd route so stale destructors and transport generations cannot displace the new owner. Retain serialized RPC steps once the channel is durable, replay them only after channel_reestablish completes, and bound open attempts while their peer remains disconnected. Preserve completed RBF state across tx_abort so a replacement attempt can resume without losing an inflight the peer may know. Changelog-Fixed: dualopend: saved dual-funded opens and RBF attempts now resume their pending RPCs after peer reconnection instead of becoming stranded. --- lightningd/channel.c | 36 +- lightningd/channel.h | 85 +++ lightningd/connect_control.c | 6 + lightningd/dual_open_control.c | 704 +++++++++++++++++--- lightningd/dual_open_control.h | 13 + lightningd/lightningd.h | 3 + lightningd/options.c | 10 + lightningd/peer_control.c | 58 +- lightningd/test/run-invoice-select-inchan.c | 14 + openingd/dualopend.c | 77 ++- openingd/dualopend_wire.csv | 14 + wallet/test/run-wallet.c | 14 + 12 files changed, 927 insertions(+), 107 deletions(-) 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 2bfe56643131..2aa738f99715 100644 --- a/lightningd/dual_open_control.c +++ b/lightningd/dual_open_control.c @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -71,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); + + /* 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); +} - log_broken(channel->log, "%s", errmsg); - channel_disconnect(channel, LOG_INFORM, true, errmsg); +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, @@ -969,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 @@ -989,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" @@ -997,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, @@ -1025,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; } @@ -1127,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; } @@ -2453,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; @@ -2470,27 +2547,25 @@ 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 restart_dualopend_after_abort(struct channel *channel) +static void send_pending_open_attempt(struct channel *channel) { - char *err; + struct open_attempt *oa = channel->open_attempt; - /* A disconnect or another restart may have happened after the abort. */ - if (channel->owner || channel->peer->connected != PEER_CONNECTED) + 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; - err = restart_dualopend(tmpctx, channel->peer->ld, channel, true); - if (err) - log_broken(channel->log, - "Unable to restart dualopend after abort: %s", err); + /* 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 { @@ -2527,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 */ @@ -2811,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); } @@ -2894,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)); } @@ -2982,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, @@ -2998,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" @@ -3158,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. */ @@ -3744,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); @@ -3809,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: @@ -3827,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; } @@ -3970,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. */ @@ -4021,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, @@ -4028,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, @@ -4086,13 +4462,15 @@ static void dualopen_errmsg(struct channel *channel, delete_channel(channel, false); return; } - /* We are called while the old dualopend's terminal - * status is still being handled. Defer replacement - * until it has been destroyed and its HSM client has - * gone away. */ - new_reltimer(channel->peer->ld->timers, - channel, time_from_msec(0), - restart_dualopend_after_abort, channel); + /* 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; @@ -4191,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, @@ -4204,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) { @@ -4244,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, @@ -4278,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, @@ -4292,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) { @@ -4302,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, @@ -4378,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.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/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 3d1023cacdd8..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); @@ -3483,6 +3521,18 @@ static void rbf_wrap_up(struct state *state, handle_send_tx_sigs(state, msg); } 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) @@ -3641,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, @@ -3816,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); @@ -4125,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; @@ -4150,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: @@ -4364,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); @@ -4513,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; @@ -4525,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. */ @@ -4546,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/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, From 9f269ac98253671844ee9067640c1c1778f4380b Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:11:11 -0500 Subject: [PATCH 12/20] tests: cover half-saved dual-open reconnection A disconnect after the opener sends its final tx_complete can leave only that side with a saved channel. On reconnect it sends channel_reestablish, while the peer correctly reports the unknown channel. The opener must consume that error, discard the safe-to-forget half-open, fail the retained fundchannel RPC, and leave the peer connection usable instead of entering a reconnect loop. Cover both outcomes for the saved opener: without reconnection, a short grace period bounds the RPC and channel lifetime; with reconnection, the unknown- channel response clears the stale state. Also stop treating the final '=' directive as an expected disconnect: it merely consumes the directive and was only failing because an earlier half-open remained stranded by #8822. These cases make the original opener-disconnect test deterministic enough to remove its flaky marker. Changelog-None --- tests/test_connection.py | 72 ++++++++++++++++++++++++++++++++++------ 1 file changed, 61 insertions(+), 11 deletions(-) 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') From fc568f28576eb0348b6915463f1c5a8850bbd41b Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:11:39 -0500 Subject: [PATCH 13/20] tests: wait for dual-open disconnect before restart Stopping the peer returns before the surviving node necessarily processes the transport loss. Mutating the stopped peer's remembered funding txid and restarting it immediately can therefore overlap teardown of the old dualopend route with installation of its replacement. Wait until the surviving node observes the disconnect before restarting the peer. This keeps the test focused on reconciling the deliberate next-funding mismatch rather than an unrelated old-versus-new transport race. Changelog-None --- tests/test_opening.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_opening.py b/tests/test_opening.py index 5fac530779d3..9c574c65b5cd 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -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() From a1dfd1263feb3ede9f294fe89ad7536214bef382 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:11:59 -0500 Subject: [PATCH 14/20] tests: retry zeroconf reconnect after route failure connect() returns once the peer_connected hook completes, before replacement channeld routes have necessarily exchanged channel_reestablish. A transport can fail during that interval and leave one connectd reporting a surviving half-connection, causing the following close RPC to wait on a route that was never fully installed. Require both channelds to receive channel_reestablish before proceeding. If a connection generation fails or stalls, force-disconnect any surviving half, wait until both nodes observe the disconnect, and retry with a fresh generation. Keep the retry bounded because this test is about closing a withheld zeroconf channel. Repeated route-generation failures are exercised separately by the dedicated reconnect tests. Changelog-None --- tests/test_opening.py | 44 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index 9c574c65b5cd..1c25d5e6d746 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -3077,7 +3077,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. From 2d01ee860ce57b0a3dcce16a8664d268d3931f0b Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:12:12 -0500 Subject: [PATCH 15/20] tests: wait for injected commitment disconnect Seeing the dev_disconnect log only proves that the injected fault fired; it does not mean both connectds have finished retiring that transport generation. Reconnecting immediately can let a delayed disconnect from the old generation interfere with the replacement dualopend route and turn this into a route lifecycle race. Wait until both peers report disconnected before reconnecting. This keeps the test focused on resuming the outstanding fundchannel RPC after a disconnect during commitment signing. Changelog-None --- tests/test_opening.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_opening.py b/tests/test_opening.py index 1c25d5e6d746..9c6991c27a7c 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -3297,5 +3297,8 @@ def test_inflight_disconnect_commitment_v2(node_factory, bitcoind): 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) From 9e8b73613922ef96293faeb5c6cc65182152c4eb Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:12:32 -0500 Subject: [PATCH 16/20] tests: expect dual-open RPCs to survive reconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A transport failure after dual-open state has been persisted does not mean the operation failed. Returning RpcError at that point leaves callers unable to distinguish a safely retryable request from one whose channel or RBF attempt already has durable commitment state, and retrying can start conflicting work. Expect fundchannel and openchannel_signed to remain pending while replacement dualopends reestablish and retransmit the required messages. Where automatic reconnect is not guaranteed, reconnect explicitly and require the original RPC future—not a replacement call—to complete successfully. Changelog-None --- tests/test_opening.py | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index 9c6991c27a7c..391237067e1a 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -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') @@ -370,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') @@ -465,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]) @@ -1087,14 +1095,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') From 8fb55e20060280fa7c3e2dbdd80cebadb7609156 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:13:24 -0500 Subject: [PATCH 17/20] tests: exercise RBF reconnect wire faults independently Running several disconnect directives through one RBF attempt makes each fault depend on protocol state left by the earlier generations. A passing test then does not show clearly that every individual wire boundary can recover, and a failure does not identify which boundary is broken. Parameterize tx_init_rbf, tx_ack_rbf, tx_add_input, and tx_add_output so each direction is exercised in a fresh channel. Consume the initial channel-open TX_ADD messages before arming those RBF faults, and leave tx_complete and commitment_signed to the construction test. For each case, wait for the injected generation to retire, reconnect, and require the original openchannel_bump RPC to complete. Use a short retained- command timeout so a broken recovery path fails promptly instead of stalling the suite. Changelog-None --- tests/test_opening.py | 140 ++++++++++++++++++++++++++++++++---------- 1 file changed, 107 insertions(+), 33 deletions(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index 391237067e1a..cc435c2da232 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -817,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 @@ -851,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 @@ -899,26 +912,87 @@ 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 - # This should succeed + +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} + ]) + + l1.rpc.connect(l2.info['id'], 'localhost', l2.port) + 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) - 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_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', From 91170e33f4241bd513abd0f459a7d9a062ed8134 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:13:50 -0500 Subject: [PATCH 18/20] tests: cover repeated dual-open owner generations Single-disconnect cases prove that one replacement owner can resume an RBF, but they do not expose state which survives long enough to interfere with a later generation. A stale owner destructor, disconnect notification, route, or connection counter can incorrectly retire generation N+2 even though the first recovery succeeded. Inject the same tx_add_input disconnect through four complete transport and owner generations, waiting for both peers to observe each teardown before reconnecting. Require the original openchannel_bump future to complete after the final replacement, without issuing a caller retry. Keep this repeated-generation regression separate from the parameterized single-wire-fault cases so failures distinguish lifecycle reuse from a problem at one particular message boundary. Changelog-None --- tests/test_opening.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/test_opening.py b/tests/test_opening.py index cc435c2da232..ee46d287d504 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -988,6 +988,32 @@ def test_rbf_reconnect_tx_add(node_factory, bitcoind, chainparams, 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): From 2aa996b44a836d783974bf3a9b856620b1cfd195 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:14:00 -0500 Subject: [PATCH 19/20] tests: wait for RBF reconnect transitions to complete The construction reconnect test deliberately carries one RBF attempt through several partially persisted states: one-sided tx_complete, mutual remembered state, and commitment_signed retransmission. Starting the next fault before the previous transport and owner generation has retired turns those state checks into teardown races and can strand the replacement route before its scratch commitment is saved. Run blocking RPCs in futures, wait for both peers to observe each disconnect, and consume the expected RPC result before advancing. After the final commitment_signed fault, wait for connectivity to converge and for the selected inflight's scratch_txid to appear before retrying openchannel_update. Treat injected transport loss as clean owner retirement rather than accepting a BROKEN log. This keeps the test focused on which RBF state survives each construction boundary. Changelog-None --- tests/test_opening.py | 74 +++++++++++++++++++++++++++++-------------- 1 file changed, 50 insertions(+), 24 deletions(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index ee46d287d504..3a601fc2116e 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -1027,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 @@ -1060,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 .*') @@ -1086,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'] @@ -1118,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']) From 17b7f8df5a6160d0f7430f9c65ef7c1bf08c4931 Mon Sep 17 00:00:00 2001 From: niftynei Date: Fri, 21 Aug 2026 21:14:04 -0500 Subject: [PATCH 20/20] tests: shorten dual-open disconnect timeout test_rbf_to_chain_before_commit deliberately disconnects a dual-open attempt before the commitment exchange completes, then exercises recovery when a funding transaction reaches the chain. Waiting for the production 600-second disconnect timeout would instead leave the test stalled until its own timeout. Set the retaining node's timeout to three seconds so the expected expiration is prompt and deterministic. This is local test configuration and does not change the production default. Changelog-None --- tests/test_opening.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_opening.py b/tests/test_opening.py index 3a601fc2116e..f854f69956a5 100644 --- a/tests/test_opening.py +++ b/tests/test_opening.py @@ -1249,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}])