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 <ccan/time/time.h>
 #include <ccan/timer/timer.h>
 
+struct io_fd_registration {
+	struct fd *fd;
+	size_t refs;
+};
+
+struct ready_fd {
+	/* NULL is the position at which to service an IO_ALWAYS plan. */
+	struct io_fd_registration *registration;
+	short revents;
+};
+
 static size_t num_fds = 0, max_fds = 0, num_waiting = 0, num_always = 0, max_always = 0, num_exclusive = 0;
 static struct pollfd *pollfds = NULL;
 static struct fd **fds = NULL;
+static bool protect_stale_fds = false;
+static struct ready_fd *ready_fds = NULL;
+static size_t ready_fds_capacity = 0;
 static struct io_plan **always = NULL;
 static struct timemono (*nowfn)(void) = time_mono;
 static int (*pollfn)(struct pollfd *fds, nfds_t nfds, int timeout) = poll;
 
+static void cleanup_ready_fds(void)
+{
+	free(ready_fds);
+	ready_fds = NULL;
+	ready_fds_capacity = 0;
+}
+
+static struct io_fd_registration *new_registration(struct fd *fd)
+{
+	struct io_fd_registration *registration = malloc(sizeof(*registration));
+
+	if (!registration)
+		return NULL;
+	registration->fd = fd;
+	registration->refs = 1;
+	return registration;
+}
+
+static void registration_put(struct io_fd_registration *registration)
+{
+	assert(registration->refs != 0);
+	registration->refs--;
+	if (registration->refs == 0)
+		free(registration);
+}
+
+void io_poll_protect_stale_fds(void)
+{
+	if (protect_stale_fds)
+		return;
+	if (atexit(cleanup_ready_fds) != 0)
+		abort();
+
+	/* This can be enabled after callers have already registered fds. */
+	for (size_t i = 0; i < num_fds; i++) {
+		fds[i]->registration = new_registration(fds[i]);
+		if (!fds[i]->registration)
+			abort();
+	}
+	protect_stale_fds = true;
+}
+
 struct timemono (*io_time_override(struct timemono (*now)(void)))(void)
 {
 	struct timemono (*old)(void) = nowfn;
@@ -54,6 +110,12 @@ static bool add_fd(struct fd *fd, short events)
 			return false;
 		max_fds = num;
 	}
+	if (protect_stale_fds) {
+		fd->registration = new_registration(fd);
+		if (!fd->registration)
+			return false;
+	} else
+		fd->registration = NULL;
 
 	pollfds[num_fds].events = events;
 	/* In case it's idle. */
@@ -75,6 +137,7 @@ static bool add_fd(struct fd *fd, short events)
 static void del_fd(struct fd *fd)
 {
 	size_t n = fd->backend_info;
+	struct io_fd_registration *registration = fd->registration;
 
 	assert(n != -1);
 	assert(n < num_fds);
@@ -98,6 +161,12 @@ static void del_fd(struct fd *fd)
 	}
 	num_fds--;
 	fd->backend_info = -1;
+	if (registration) {
+		assert(registration->fd == fd);
+		registration->fd = NULL;
+		fd->registration = NULL;
+		registration_put(registration);
+	}
 
 	if (fd->exclusive[IO_IN])
 		num_exclusive--;
@@ -369,6 +438,26 @@ static void restore_pollfds(void)
 	}
 }
 
+static void append_ready_fd(size_t *num_ready,
+			    struct io_fd_registration *registration,
+			    short revents)
+{
+	if (*num_ready == ready_fds_capacity) {
+		size_t capacity = ready_fds_capacity ? ready_fds_capacity * 2 : 8;
+		struct ready_fd *new_ready_fds;
+
+		new_ready_fds = realloc(ready_fds,
+					  sizeof(*ready_fds) * capacity);
+		if (!new_ready_fds)
+			abort();
+		ready_fds = new_ready_fds;
+		ready_fds_capacity = capacity;
+	}
+	ready_fds[*num_ready].registration = registration;
+	ready_fds[*num_ready].revents = revents;
+	(*num_ready)++;
+}
+
 /* This is the main loop. */
 void *io_loop(struct timers *timers, struct timer **expired)
 {
@@ -431,6 +520,85 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			break;
 		}
 
+		if (protect_stale_fds) {
+			size_t num_polled = num_fds;
+			size_t num_ready = 0;
+
+			if (r == 0) {
+				handle_always();
+				continue;
+			}
+
+			/* Preserve the old fairness order, but retain only ready fds and
+			 * the point where IO_ALWAYS work was interleaved. */
+			fairness_counter++;
+			for (size_t rotation = 0; rotation < num_polled; rotation++) {
+				struct io_fd_registration *registration;
+
+				i = (rotation + fairness_counter) % num_polled;
+				if (i == 0)
+					append_ready_fd(&num_ready, NULL, 0);
+
+				if (!pollfds[i].revents)
+					continue;
+				registration = fds[i]->registration;
+				assert(registration);
+				registration->refs++;
+				append_ready_fd(&num_ready, registration,
+						pollfds[i].revents);
+			}
+			for (size_t n = 0; n < num_polled; n++)
+				pollfds[n].revents = 0;
+
+			for (size_t n = 0; n < num_ready && !io_loop_return; n++) {
+				socklen_t errno_len = sizeof(errno);
+				struct io_fd_registration *registration;
+				struct fd *fd;
+				struct io_conn *c;
+				int events;
+
+				registration = ready_fds[n].registration;
+				if (!registration) {
+					if (handle_always())
+						break;
+					continue;
+				}
+				fd = registration->fd;
+				if (!fd)
+					continue;
+				events = ready_fds[n].revents;
+				c = (void *)fd;
+				if (fd->listener) {
+					struct io_listener *l = (void *)fd;
+					if (events & POLLIN) {
+						accept_conn(l);
+						r--;
+					} else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
+						r--;
+						errno = EBADF;
+						io_close_listener(l);
+					}
+				} else if (events & (POLLIN|POLLOUT)) {
+					r--;
+					io_ready(c, events);
+				} else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
+					r--;
+					if (getsockopt(fd->fd, SOL_SOCKET,
+						       SO_ERROR, &errno,
+						       &errno_len) == -1)
+						errno = EBADF;
+					io_close(c);
+				}
+			}
+			/* Callbacks can retire registrations, so release snapshot
+			 * references only after dispatch has stopped using the buffer. */
+			for (size_t n = 0; n < num_ready; n++) {
+				if (ready_fds[n].registration)
+					registration_put(ready_fds[n].registration);
+			}
+			continue;
+		}
+
 		fairness_counter++;
 		for (size_t rotation = 0; rotation < num_fds && !io_loop_return; rotation++) {
 			socklen_t errno_len = sizeof(errno);
@@ -486,7 +654,6 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			}
 		}
 	}
-
 	ret = io_loop_return;
 	io_loop_return = NULL;
 
diff --git a/ccan/ccan/io/test/run-49-stale-fd-readiness.c b/ccan/ccan/io/test/run-49-stale-fd-readiness.c
new file mode 100644
index 000000000000..6c5618424d5d
--- /dev/null
+++ b/ccan/ccan/io/test/run-49-stale-fd-readiness.c
@@ -0,0 +1,142 @@
+#include "config.h"
+#include <assert.h>
+#include <ccan/io/io.h>
+/* Include the C files directly to make each readiness delivery observable. */
+#include <ccan/io/poll.c>
+#include <ccan/io/io.c>
+#include <unistd.h>
+
+static int replacement_fd;
+static unsigned int poll_calls, actor_events, victim_events,
+		    survivor_events, replacement_events;
+static char completed;
+
+static void check_ready_fds_cleaned(void)
+{
+	assert(ready_fds == NULL);
+	assert(ready_fds_capacity == 0);
+}
+
+static int replacement_event(int fd, struct io_plan_arg *arg)
+{
+	replacement_events++;
+	return 1;
+}
+
+static struct io_plan *replacement_ready(struct io_conn *conn, void *unused)
+{
+	io_break(&completed);
+	return io_close(conn);
+}
+
+static struct io_plan *replacement_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, replacement_event,
+			   replacement_ready, NULL);
+}
+
+static int actor_event(int fd, struct io_plan_arg *arg)
+{
+	actor_events++;
+	return 1;
+}
+
+static struct io_plan *actor_ready(struct io_conn *conn, void *unused)
+{
+	struct io_plan *closed = io_close(conn);
+
+	/* Closing the current fd compacts the table.  Adding its replacement
+	 * restores the old table length, but must not make it part of the poll
+	 * result currently being dispatched. */
+	if (!io_new_conn(NULL, replacement_fd, replacement_init, NULL))
+		abort();
+	return closed;
+}
+
+static struct io_plan *actor_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, actor_event, actor_ready, NULL);
+}
+
+static int victim_event(int fd, struct io_plan_arg *arg)
+{
+	victim_events++;
+	return 1;
+}
+
+static struct io_plan *victim_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, victim_event, io_close_cb, NULL);
+}
+
+static int survivor_event(int fd, struct io_plan_arg *arg)
+{
+	survivor_events++;
+	return 1;
+}
+
+static struct io_plan *survivor_init(struct io_conn *conn, void *unused)
+{
+	io_plan_arg(conn, IO_IN);
+	return io_set_plan(conn, IO_IN, survivor_event, io_close_cb, NULL);
+}
+
+static int fake_poll(struct pollfd *fds, nfds_t nfds, int timeout)
+{
+	poll_calls++;
+	for (size_t i = 0; i < nfds; i++)
+		fds[i].revents = 0;
+
+	if (poll_calls == 1) {
+		/* Fairness rotation handles slot 1 first.  That actor replaces
+		 * itself while slots 0 and 2 still have readiness pending. */
+		assert(nfds == 3);
+		for (size_t i = 0; i < nfds; i++)
+			fds[i].revents = POLLIN;
+		return 3;
+	}
+
+	assert(poll_calls == 2);
+	for (size_t i = 0; i < nfds; i++) {
+		if (fds[i].fd != replacement_fd)
+			continue;
+		fds[i].revents = POLLIN;
+		return 1;
+	}
+	abort();
+}
+
+int main(void)
+{
+	int actor_pipe[2], victim_pipe[2], survivor_pipe[2], replacement_pipe[2];
+
+	assert(pipe(actor_pipe) == 0);
+	assert(pipe(victim_pipe) == 0);
+	assert(pipe(survivor_pipe) == 0);
+	assert(pipe(replacement_pipe) == 0);
+	replacement_fd = replacement_pipe[0];
+
+	assert(io_poll_override(fake_poll) == poll);
+	assert(io_new_conn(NULL, victim_pipe[0], victim_init, NULL));
+	assert(io_new_conn(NULL, actor_pipe[0], actor_init, NULL));
+	assert(io_new_conn(NULL, survivor_pipe[0], survivor_init, NULL));
+
+	/* Registered first so this runs after the backend's LIFO cleanup. */
+	assert(atexit(check_ready_fds_cleaned) == 0);
+	io_poll_protect_stale_fds();
+	assert(io_loop(NULL, NULL) == &completed);
+	assert(poll_calls == 2);
+	assert(actor_events == 1);
+	assert(victim_events == 1);
+	assert(survivor_events == 1);
+	assert(replacement_events == 1);
+
+	close(actor_pipe[1]);
+	close(victim_pipe[1]);
+	close(survivor_pipe[1]);
+	close(replacement_pipe[1]);
+	return 0;
+}
diff --git a/common/test/Makefile b/common/test/Makefile
index 71f9dc17d981..dd165bef0f45 100644
--- a/common/test/Makefile
+++ b/common/test/Makefile
@@ -1,7 +1,10 @@
+CCAN_IO_TEST_SRC := ccan/ccan/io/test/run-49-stale-fd-readiness.c
 COMMON_TEST_SRC := $(wildcard common/test/run-*.c)
 
 COMMON_TEST_OBJS := $(COMMON_TEST_SRC:.c=.o)
 COMMON_TEST_PROGRAMS := $(COMMON_TEST_OBJS:.o=)
+CCAN_IO_TEST_OBJS := $(CCAN_IO_TEST_SRC:.c=.o)
+CCAN_IO_TEST_PROGRAMS := $(CCAN_IO_TEST_OBJS:.o=)
 
 COMMON_TEST_COMMON_OBJS :=				\
 	common/autodata.o				\
@@ -13,8 +16,8 @@ COMMON_TEST_COMMON_OBJS :=				\
 $(COMMON_TEST_PROGRAMS): $(COMMON_TEST_COMMON_OBJS) $(BITCOIN_OBJS)
 $(COMMON_TEST_OBJS): $(COMMON_HEADERS) $(WIRE_HEADERS) $(COMMON_SRC) common/test/Makefile
 
-ALL_C_SOURCES += $(COMMON_TEST_SRC)
-ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS)
+ALL_C_SOURCES += $(COMMON_TEST_SRC) $(CCAN_IO_TEST_SRC)
+ALL_TEST_PROGRAMS += $(COMMON_TEST_PROGRAMS) $(CCAN_IO_TEST_PROGRAMS)
 
 # Make them all depend on common/ files, for simplicity (they directly #include some)
 $(COMMON_TEST_OBJS): $(COMMON_SRC)
@@ -178,4 +181,4 @@ common/test/run-close_tx:			\
 	wire/fromwire.o					\
 	wire/towire.o
 
-check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%)
+check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%) $(CCAN_IO_TEST_PROGRAMS:%=unittest/%)
diff --git a/connectd/connectd.c b/connectd/connectd.c
index cb95ef9b0cac..3caafe66a765 100644
--- a/connectd/connectd.c
+++ b/connectd/connectd.c
@@ -2354,7 +2354,9 @@ static struct io_plan *recv_peer_connect_subd(struct io_conn *conn,
 					      int fd,
 					      struct daemon *daemon)
 {
-	peer_connect_subd(daemon, msg, fd);
+	peer_connect_subd(daemon, msg, fd,
+			  fromwire_peektype(msg)
+			  == WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED);
 	return daemon_conn_read_next(conn, daemon->master);
 }
 
@@ -2404,10 +2406,15 @@ static struct io_plan *recv_req(struct io_conn *conn,
 		goto out;
 
 	case WIRE_CONNECTD_PEER_CONNECT_SUBD:
+	case WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED:
 		/* This comes with an fd */
 		return daemon_conn_read_with_fd(conn, daemon->master,
 						recv_peer_connect_subd, daemon);
 
+	case WIRE_CONNECTD_PEER_RESUME_SUBD:
+		peer_resume_subd(daemon, msg);
+		goto out;
+
 	case WIRE_CONNECTD_START_SHUTDOWN:
 		start_shutdown(daemon, msg);
 		goto out;
@@ -2470,6 +2477,7 @@ static struct io_plan *recv_req(struct io_conn *conn,
 	case WIRE_CONNECTD_CUSTOMMSG_IN:
 	case WIRE_CONNECTD_PEER_DISCONNECTED:
 	case WIRE_CONNECTD_PEER_RECONNECTED:
+	case WIRE_CONNECTD_PEER_CONNECT_SUBD_REPLY:
 	case WIRE_CONNECTD_START_SHUTDOWN_REPLY:
 	case WIRE_CONNECTD_INJECT_ONIONMSG_REPLY:
 	case WIRE_CONNECTD_ONIONMSG_FORWARD_FAIL:
@@ -2532,6 +2540,7 @@ int main(int argc, char *argv[])
 
 	/* Common subdaemon setup code. */
 	developer = subdaemon_setup(argc, argv);
+	io_poll_protect_stale_fds();
 
 	/* Allocate and set up our simple top-level structure. */
 	daemon = tal(NULL, struct daemon);
diff --git a/connectd/connectd_wire.csv b/connectd/connectd_wire.csv
index c1361a597556..a95e6ed9e90c 100644
--- a/connectd/connectd_wire.csv
+++ b/connectd/connectd_wire.csv
@@ -112,6 +112,28 @@ msgdata,connectd_peer_connect_subd,id,node_id,
 msgdata,connectd_peer_connect_subd,counter,u64,
 msgdata,connectd_peer_connect_subd,channel_id,channel_id,
 
+# Master -> connectd: attach a replacement dualopend and acknowledge only once
+# its route is active (+ fd to subd).
+msgtype,connectd_peer_connect_subd_tracked,2018
+msgdata,connectd_peer_connect_subd_tracked,id,node_id,
+msgdata,connectd_peer_connect_subd_tracked,counter,u64,
+msgdata,connectd_peer_connect_subd_tracked,channel_id,channel_id,
+
+# Connectd -> master: the subd route was accepted or rejected.  A restarted
+# dualopend does not begin its peer protocol until the route is installed.
+msgtype,connectd_peer_connect_subd_reply,2104
+msgdata,connectd_peer_connect_subd_reply,id,node_id,
+msgdata,connectd_peer_connect_subd_reply,counter,u64,
+msgdata,connectd_peer_connect_subd_reply,channel_id,channel_id,
+msgdata,connectd_peer_connect_subd_reply,accepted,bool,
+
+# Master -> connectd: a replacement subd has inherited the endpoint returned
+# after tx_abort.  Reactivate that exact route and cancel its abort timer.
+msgtype,connectd_peer_resume_subd,2017
+msgdata,connectd_peer_resume_subd,id,node_id,
+msgdata,connectd_peer_resume_subd,counter,u64,
+msgdata,connectd_peer_resume_subd,channel_id,channel_id,
+
 # Connectd -> master: peer said something interesting
 msgtype,connectd_peer_spoke,2005
 msgdata,connectd_peer_spoke,id,node_id,
diff --git a/connectd/multiplex.c b/connectd/multiplex.c
index 1051645f8cec..e609e7c85920 100644
--- a/connectd/multiplex.c
+++ b/connectd/multiplex.c
@@ -43,6 +43,11 @@ struct subd {
 
 	/* The actual connection to talk to it (NULL if it's not connected yet)  */
 	struct io_conn *conn;
+	/* A tracked io_new_conn is waiting for subd_conn_init_tracked. */
+	bool conn_pending;
+	/* Identify that request when acknowledging the active connection. */
+	u64 pending_counter;
+	struct channel_id pending_channel_id;
 
 	/* Input buffer */
 	u8 *in;
@@ -52,6 +57,7 @@ struct subd {
 
 	/* After we've told it to tx_abort, we don't send anything else. */
 	bool rcvd_tx_abort;
+	struct oneshot *abort_close_timer;
 };
 
 /* FIXME: reorder! */
@@ -95,6 +101,7 @@ static void close_peer_io_timeout(struct peer *peer)
 
 static void close_subd_timeout(struct subd *subd)
 {
+	subd->abort_close_timer = NULL;
 	status_peer_broken(&subd->peer->id, "Subd did not close, forcing close");
 	io_close(subd->conn);
 }
@@ -1395,7 +1402,9 @@ static struct subd *new_subd(struct peer *peer,
 	subd->temporary_channel_id = NULL;
 	subd->opener_revocation_basepoint = NULL;
 	subd->conn = NULL;
+	subd->conn_pending = false;
 	subd->rcvd_tx_abort = false;
+	subd->abort_close_timer = NULL;
 
 	/* Connect it to the peer */
 	tal_arr_expand(&peer->subds, subd);
@@ -1525,9 +1534,11 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
        if (type == WIRE_TX_ABORT) {
 	       subd->rcvd_tx_abort = true;
 	       /* In case it doesn't close by itself */
-	       notleak(new_reltimer(&peer->daemon->timers, subd,
-				    time_from_sec(5),
-				    close_subd_timeout, subd));
+	       subd->abort_close_timer
+		       = notleak(new_reltimer(&peer->daemon->timers,
+					     subd, time_from_sec(5),
+					     close_subd_timeout,
+					     subd));
        }
 
        /* Wait for them to wake us */
@@ -1606,16 +1617,62 @@ static struct io_plan *read_hdr_from_peer(struct io_conn *peer_conn,
 		       read_body_from_peer, peer);
 }
 
-static struct io_plan *subd_conn_init(struct io_conn *subd_conn,
-				      struct subd *subd)
+static void send_connect_subd_reply(struct daemon *daemon,
+				    const struct node_id *id,
+				    u64 counter,
+				    const struct channel_id *channel_id,
+				    bool accepted)
+{
+	daemon_conn_send(daemon->master,
+			 take(towire_connectd_peer_connect_subd_reply(
+				 NULL, id, counter, channel_id, accepted)));
+}
+
+static struct io_plan *subd_conn_ready(struct io_conn *subd_conn,
+				       struct subd *subd)
+{
+	bool accepted;
+
+	accepted = subd->peer->to_peer != NULL
+		&& subd->peer->counter == subd->pending_counter
+		&& subd->peer->draining_state == NOT_DRAINING;
+	send_connect_subd_reply(subd->peer->daemon, &subd->peer->id,
+				subd->pending_counter,
+				&subd->pending_channel_id, accepted);
+	if (!accepted)
+		return io_close(subd_conn);
+	return read_from_subd(subd_conn, subd);
+}
+
+/* The long-standing attachment path for ordinary subdaemons.  Keep this
+ * independent of the replacement-dualopend acknowledgement barrier. */
+static struct io_plan *subd_conn_init_untracked(struct io_conn *subd_conn,
+						struct subd *subd)
 {
 	subd->conn = subd_conn;
+	tal_steal(subd->conn, subd);
+	tal_add_destructor(subd, destroy_connected_subd);
+	return io_duplex(subd_conn,
+			 read_from_subd(subd_conn, subd),
+			 write_to_subd(subd_conn, subd));
+}
+
+static struct io_plan *subd_conn_init_tracked(struct io_conn *subd_conn,
+					      struct subd *subd)
+{
+	assert(subd->conn_pending);
+	subd->conn_pending = false;
+	subd->conn = subd_conn;
 
 	/* subd is a child of the conn: free when it closes! */
 	tal_steal(subd->conn, subd);
 	tal_add_destructor(subd, destroy_connected_subd);
+
+	/* Defer the tracked acknowledgement until ccan/io has installed this
+	 * duplex plan, so lightningd cannot start a replacement owner against a
+	 * half-live route. */
 	return io_duplex(subd_conn,
-			 read_from_subd(subd_conn, subd),
+			 io_always(subd_conn, subd_conn_ready, subd),
 			 write_to_subd(subd_conn, subd));
 }
 
@@ -1631,8 +1688,9 @@ static void destroy_peer_conn(struct io_conn *peer_conn, struct peer *peer)
 
 	/* Wake subds: give them 5 seconds to flush. */
 	for (size_t i = 0; i < tal_count(peer->subds); i++) {
-		/* Might not be connected yet (no destructor, simple) */
-		if (!peer->subds[i]->conn) {
+		/* A pending io_new_conn still owns an fd and must unwind normally. */
+		if (!peer->subds[i]->conn
+		    && !peer->subds[i]->conn_pending) {
 			tal_arr_remove(&peer->subds, i);
 			i--;
 			continue;
@@ -1682,7 +1740,8 @@ struct io_plan *multiplex_peer_setup(struct io_conn *peer_conn,
 			 write_to_peer(peer_conn, peer));
 }
 
-void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd)
+void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd,
+		       bool tracked)
 {
 	struct node_id id;
 	u64 counter;
@@ -1690,7 +1749,14 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd)
 	struct channel_id channel_id;
 	struct subd *subd;
 
-	if (!fromwire_connectd_peer_connect_subd(msg, &id, &counter, &channel_id))
+	if (tracked) {
+		if (!fromwire_connectd_peer_connect_subd_tracked(msg, &id,
+							       &counter,
+							       &channel_id))
+			master_badmsg(WIRE_CONNECTD_PEER_CONNECT_SUBD_TRACKED,
+				      msg);
+	} else if (!fromwire_connectd_peer_connect_subd(msg, &id, &counter,
+						       &channel_id))
 		master_badmsg(WIRE_CONNECTD_PEER_CONNECT_SUBD, msg);
 
 	/* If receiving fd failed, fd will be -1.  Log and ignore
@@ -1703,24 +1769,59 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd)
 				   strerror(errno));
 		/* Maybe free up some fds by closing something. */
 		close_random_connection(daemon);
-		return;
+		goto reply;
 	}
 
 	/* Races can happen: this might be gone by now (or reconnected!). */
 	peer = peer_htable_get(daemon->peers, &id);
 	if (!peer || peer->counter != counter) {
 		close(fd);
-		return;
+		goto reply;
 	}
 
 	/* Could be disconnecting now */
 	if (!peer->to_peer || peer->draining_state != NOT_DRAINING) {
 		close(fd);
-		return;
+		goto reply;
 	}
 
 	/* If peer said something, we created this and queued msg. */
 	subd = find_subd(peer, &channel_id);
+	if (!tracked) {
+		if (!subd) {
+			subd = new_subd(peer, &channel_id);
+			release_one_waiting_connection(
+				peer->daemon,
+				tal_fmt(tmpctx, "%s given a subd",
+					fmt_node_id(tmpctx, &id)));
+		}
+		if (subd->conn) {
+			status_peer_debug(&id,
+					  "Already have a subd for channel_id %s: ignoring",
+					  fmt_channel_id(tmpctx, &channel_id));
+			close(fd);
+			return;
+		}
+		io_new_conn(peer, fd, subd_conn_init_untracked, subd);
+		return;
+	}
+
+	/* lightningd serializes owner installation.  Thus an attached endpoint
+	 * here belongs to the retiring owner and this request is its handover. */
+	if (subd && subd->conn) {
+		status_peer_debug(&id,
+				  "Replacing attached subd for channel_id %s",
+				  fmt_channel_id(tmpctx, &channel_id));
+		io_close(subd->conn);
+		subd = NULL;
+	} else if (subd && subd->conn_pending) {
+		status_peer_debug(&id,
+				  "Subd attachment already pending for channel_id %s",
+				  fmt_channel_id(tmpctx, &channel_id));
+		close(fd);
+		goto reply;
+	}
+
 	if (!subd) {
 		subd = new_subd(peer, &channel_id);
 		/* Implies lightningd is ready for another peer. */
@@ -1728,20 +1829,61 @@ void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd)
 					       tal_fmt(tmpctx, "%s given a subd",
 						       fmt_node_id(tmpctx, &id)));
 	}
+	/* subd_conn_ready sends the reply after the duplex route is active. */
+	subd->conn_pending = true;
+	subd->pending_counter = counter;
+	subd->pending_channel_id = channel_id;
+	io_new_conn(peer, fd, subd_conn_init_tracked, subd);
+	return;
 
-	/* We only keep one connection per channel_id.  If one is already
-	 * attached for this channel_id, drop this fd rather than replacing
-	 * it. */
-	if (subd->conn) {
-		status_peer_debug(&id,
-				  "Already have a subd for channel_id %s: ignoring",
-				  fmt_channel_id(tmpctx, &channel_id));
-		close(fd);
+reply:
+	if (!tracked)
 		return;
+	send_connect_subd_reply(daemon, &id, counter, &channel_id, false);
+}
+
+void peer_resume_subd(struct daemon *daemon, const u8 *msg)
+{
+	struct node_id id;
+	u64 counter;
+	struct channel_id channel_id;
+	struct peer *peer;
+	struct subd *subd = NULL;
+	bool accepted = false;
+
+	if (!fromwire_connectd_peer_resume_subd(msg, &id, &counter,
+					       &channel_id))
+		master_badmsg(WIRE_CONNECTD_PEER_RESUME_SUBD, msg);
+
+	peer = peer_htable_get(daemon->peers, &id);
+	if (!peer || peer->counter != counter || !peer->to_peer
+	    || peer->draining_state != NOT_DRAINING)
+		goto reply;
+
+	/* find_subd deliberately hides a route once it has delivered tx_abort.
+	 * Resume must find that exact attached route, not create another one. */
+	for (size_t i = 0; i < tal_count(peer->subds); i++) {
+		struct subd *candidate = peer->subds[i];
+
+		if (!candidate->rcvd_tx_abort || !candidate->conn)
+			continue;
+		if (channel_id_eq(&candidate->channel_id, &channel_id)
+		    || (candidate->temporary_channel_id
+			&& channel_id_eq(candidate->temporary_channel_id,
+					 &channel_id))) {
+			subd = candidate;
+			break;
+		}
 	}
+	if (!subd)
+		goto reply;
+
+	subd->abort_close_timer = tal_free(subd->abort_close_timer);
+	subd->rcvd_tx_abort = false;
+	accepted = true;
 
-	/* This sets subd->conn inside subd_conn_init, and reparents subd! */
-	io_new_conn(peer, fd, subd_conn_init, subd);
+reply:
+	send_connect_subd_reply(daemon, &id, counter, &channel_id, accepted);
 }
 
 /* Lightningd says to send a ping */
diff --git a/connectd/multiplex.h b/connectd/multiplex.h
index f42f7cb76ea2..c59a0392d3d8 100644
--- a/connectd/multiplex.h
+++ b/connectd/multiplex.h
@@ -35,7 +35,11 @@ void custommsg_completed(struct daemon *daemon, const u8 *msg);
 void set_custommsgs(struct daemon *daemon, const u8 *msg);
 
 /* Lightningd wants to talk to you. */
-void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd);
+void peer_connect_subd(struct daemon *daemon, const u8 *msg, int fd,
+		       bool tracked);
+
+/* A replacement owner inherited an existing route after tx_abort. */
+void peer_resume_subd(struct daemon *daemon, const u8 *msg);
 
 /* Disconnect peer: give outgoing msgs time to drain though. */
 void disconnect_peer(struct peer *peer);
diff --git a/lightningd/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 <common/json_command.h>
 #include <common/psbt_open.h>
 #include <common/shutdown_scriptpubkey.h>
+#include <common/timeout.h>
 #include <common/wire_error.h>
 #include <connectd/connectd_wiregen.h>
 #include <errno.h>
@@ -27,6 +28,7 @@
 #include <lightningd/hsm_control.h>
 #include <lightningd/notification.h>
 #include <lightningd/opening_common.h>
+#include <lightningd/peer_control.h>
 #include <lightningd/peer_fd.h>
 #include <lightningd/plugin_hook.h>
 #include <openingd/dualopend_wiregen.h>
@@ -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;
diff --git a/lightningd/lightningd.c b/lightningd/lightningd.c
index 44ed3fa66425..dfd159efa51d 100644
--- a/lightningd/lightningd.c
+++ b/lightningd/lightningd.c
@@ -1199,6 +1199,7 @@ int main(int argc, char *argv[])
 
 	/*~ What happens in strange locales should stay there. */
 	setup_locale();
+	io_poll_protect_stale_fds();
 
 	/*~ This handles --dev-debug-self really early, which we otherwise ignore */
 	daemon_developer_mode(argv);
diff --git a/lightningd/opening_control.c b/lightningd/opening_control.c
index 6cc3d82f1e61..eae76b982a6d 100644
--- a/lightningd/opening_control.c
+++ b/lightningd/opening_control.c
@@ -453,6 +453,13 @@ static void opening_funder_finished(struct subd *openingd, const u8 *resp,
 		goto cleanup;
 	}
 
+	/* The peer endpoint is safely held by peer_fd now, and openingd has
+	 * completed its work.  Retire it before peer_start_channeld() asks hsmd
+	 * for another client with this dbid: hsmd deliberately will not create
+	 * that client until the previous owner has gone away. */
+	fc->uc->open_daemon = NULL;
+	subd_release_channel(openingd, fc->uc);
+
 	/* Watch for funding confirms */
 	channel_watch_funding(ld, channel);
 
@@ -567,6 +574,13 @@ static void opening_fundee_finished(struct subd *openingd,
 		goto failed;
 	}
 
+	/* Establish an explicit owner-exit barrier before requesting channeld's
+	 * HSM client.  This is particularly important when several v1 opens
+	 * finish concurrently: waiting synchronously for one lingering openingd
+	 * must not delay processing another openingd's returned peer endpoint. */
+	uc->open_daemon = NULL;
+	subd_release_channel(openingd, uc);
+
 	log_debug(channel->log, "Watching funding tx %s",
 		  fmt_bitcoin_txid(reply,
 				 &channel->funding.txid));
diff --git a/openingd/dualopend.c b/openingd/dualopend.c
index 2f1849773496..3d1023cacdd8 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);
@@ -2908,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;
 	}
@@ -2917,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;
 	}
@@ -3441,19 +3463,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));
 }
 
@@ -4288,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))
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 e23283f7251c..5fac530779d3 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,39 @@ def test_no_retransmit_confirmed_funding(node_factory):
     # Should not have attempted (and failed) to re-broadcast the funding tx.
     assert not l1.daemon.is_in_log('Failed to re-transmit funding tx')
     assert not l1.daemon.is_in_log('Successfully rexmitted funding tx')
+
+
+@unittest.skipIf(TEST_NETWORK != 'regtest', 'elementsd doesnt yet support PSBT features we need')
+@pytest.mark.openchannel('v2')
+def test_inflight_disconnect_commitment_v2(node_factory, bitcoind):
+    """Disconnect during dual-fund commitment signing should not trigger spurious BROKEN messages.
+    """
+    disconnects = ["+WIRE_COMMITMENT_SIGNED"]
+
+    opts = [{'experimental-dual-fund': None, 'dev-no-reconnect': None,
+             'may_reconnect': True, 'disconnect': disconnects},
+            {'experimental-dual-fund': None, 'dev-no-reconnect': None,
+             'may_reconnect': True}]
+
+    opener, funder = node_factory.get_nodes(2, opts=opts)
+
+    amount = 500000
+    opener.fundwallet(20000000)
+    funder.fundwallet(20000000)
+
+    funder.rpc.call('funderupdate',
+                    {'policy': 'available',
+                     'policy_mod': 100,
+                     'per_channel_max_msat': '1btc',
+                     'reserve_tank_msat': '0msat',
+                     'fund_probability': 100,
+                     'fuzz_percent': 0,
+                     'leases_only': False})
+
+    opener.rpc.connect(funder.info['id'], 'localhost', funder.port)
+    fut = node_factory.executor.submit(opener.rpc.fundchannel,
+                                       funder.info['id'], amount)
+
+    opener.daemon.wait_for_log(r'dev_disconnect: .WIRE_COMMITMENT_SIGNED')
+    opener.rpc.connect(funder.info['id'], 'localhost', funder.port)
+    fut.result(timeout=TIMEOUT)