pytest: fix flake in test_even_sendcustommsg

diff --git a/connectd/connectd.c b/connectd/connectd.c
index 93fea082..44f81b6c 100644
--- a/connectd/connectd.c
+++ b/connectd/connectd.c
@@ -2401,6 +2401,10 @@ static struct io_plan *recv_req(struct io_conn *conn,
 		add_scid_map(daemon, msg);
 		goto out;
 
+	case WIRE_CONNECTD_CUSTOMMSG_IN_COMPLETE:
+		custommsg_completed(daemon, msg);
+		goto out;
+
 	case WIRE_CONNECTD_DEV_MEMLEAK:
 		if (daemon->developer) {
 			dev_connect_memleak(daemon, msg);
diff --git a/connectd/connectd_wire.csv b/connectd/connectd_wire.csv
index 4b4d5457..7b149159 100644
--- a/connectd/connectd_wire.csv
+++ b/connectd/connectd_wire.csv
@@ -194,6 +194,10 @@ msgdata,connectd_custommsg_in,id,node_id,
 msgdata,connectd_custommsg_in,msg_len,u16,
 msgdata,connectd_custommsg_in,msg,u8,msg_len
 
+# We got that custommsg, thankyou, it was delightful.
+msgtype,connectd_custommsg_in_complete,2111
+msgdata,connectd_custommsg_in_complete,id,node_id,
+
 # A custom message that the lightningd tells us to send to the peer.
 msgtype,connectd_custommsg_out,2011
 msgdata,connectd_custommsg_out,id,node_id,
diff --git a/connectd/multiplex.c b/connectd/multiplex.c
index 203962c3..a59b7ad3 100644
--- a/connectd/multiplex.c
+++ b/connectd/multiplex.c
@@ -856,6 +856,22 @@ static bool handle_custommsg(struct daemon *daemon,
 	return true;
 }
 
+void custommsg_completed(struct daemon *daemon, const u8 *msg)
+{
+	struct node_id id;
+	const struct peer *peer;
+
+	if (!fromwire_connectd_custommsg_in_complete(msg, &id))
+		master_badmsg(WIRE_CONNECTD_CUSTOMMSG_IN_COMPLETE, msg);
+
+	/* If it's still around, log it. */
+	peer = peer_htable_get(daemon->peers, &id);
+	if (peer) {
+		status_peer_debug(&peer->id, "custommsg processing finished");
+		log_peer_io(peer, msg);
+	}
+}
+
 /* We handle pings and gossip messages. */
 static bool handle_message_locally(struct peer *peer, const u8 *msg)
 {
diff --git a/connectd/multiplex.h b/connectd/multiplex.h
index 1a20143a..f42f7cb7 100644
--- a/connectd/multiplex.h
+++ b/connectd/multiplex.h
@@ -28,6 +28,9 @@ void send_manual_ping(struct daemon *daemon, const u8 *msg);
 /* When lightningd says to send a custom message (from a plugin) */
 void send_custommsg(struct daemon *daemon, const u8 *msg);
 
+/* lightningd has finished with the custommsg */
+void custommsg_completed(struct daemon *daemon, const u8 *msg);
+
 /* When lightningd says what custom messages we can recv */
 void set_custommsgs(struct daemon *daemon, const u8 *msg);
 
diff --git a/lightningd/connect_control.c b/lightningd/connect_control.c
index ae0bde7a..09da14aa 100644
--- a/lightningd/connect_control.c
+++ b/lightningd/connect_control.c
@@ -351,6 +351,7 @@ void connect_succeeded(struct lightningd *ld, const struct peer *peer,
 }
 
 struct custommsg_payload {
+	struct lightningd *ld;
 	struct node_id peer_id;
 	u8 *msg;
 };
@@ -376,6 +377,11 @@ static bool custommsg_cb(struct custommsg_payload *payload,
 
 static void custommsg_final(struct custommsg_payload *payload STEALS)
 {
+	/* Note: on shutdown, ld->connectd can be NULL! */
+	if (payload->ld->connectd) {
+		subd_send_msg(payload->ld->connectd,
+			      take(towire_connectd_custommsg_in_complete(NULL, &payload->peer_id)));
+	}
 	tal_steal(tmpctx, payload);
 }
 
@@ -397,6 +403,7 @@ static void handle_custommsg_in(struct lightningd *ld, const u8 *msg)
 {
 	struct custommsg_payload *p = tal(NULL, struct custommsg_payload);
 
+	p->ld = ld;
 	if (!fromwire_connectd_custommsg_in(p, msg, &p->peer_id, &p->msg)) {
 		log_broken(ld->log, "Malformed custommsg: %s",
 			   tal_hex(tmpctx, msg));
@@ -550,6 +557,7 @@ static unsigned connectd_msg(struct subd *connectd, const u8 *msg, const int *fd
 	case WIRE_CONNECTD_DEV_EXHAUST_FDS:
 	case WIRE_CONNECTD_DEV_SET_MAX_SCIDS_ENCODE_SIZE:
 	case WIRE_CONNECTD_SCID_MAP:
+	case WIRE_CONNECTD_CUSTOMMSG_IN_COMPLETE:
 	/* This is a reply, so never gets through to here. */
 	case WIRE_CONNECTD_INIT_REPLY:
 	case WIRE_CONNECTD_ACTIVATE_REPLY:
diff --git a/tests/test_misc.py b/tests/test_misc.py
index 20705782..5baebbcd 100644
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -4719,6 +4719,8 @@ def test_even_sendcustommsg(node_factory):
     l1.rpc.sendcustommsg(l2.info['id'], msg)
     l2.daemon.wait_for_log(r'\[IN\] {}'.format(msg))
     l2.daemon.wait_for_log(r'allow_even_msgs.*Got message 43690')
+    # Make sure it *processes* before we remove plugin.
+    l2.daemon.wait_for_log(f"{l1.info['id']}-connectd: custommsg processing finished")
 
     # And nobody gets upset
     assert only_one(l1.rpc.listpeers(l2.info['id'])['peers'])['connected']