diff --git a/lightningd/peer_control.c b/lightningd/peer_control.c
index 2ec8bea3689f..6c1216902b2f 100644
--- a/lightningd/peer_control.c
+++ b/lightningd/peer_control.c
@@ -1309,6 +1309,7 @@ struct peer_connected_hook_payload {
 	struct lightningd *ld;
 	struct wireaddr_internal addr;
 	struct wireaddr *remote_addr;
+	u8 *their_features;
 	bool incoming;
 	/* We don't keep a pointer to peer: it might be freed! */
 	struct node_id peer_id;
@@ -1327,10 +1328,7 @@ peer_connected_serialize(struct peer_connected_hook_payload *payload,
 	if (payload->remote_addr)
 		json_add_string(stream, "remote_addr",
 				fmt_wireaddr(tmpctx, payload->remote_addr));
-	/* Since this is start of hook, peer is always in table! */
-	json_add_hex_talarr(stream, "features",
-			    peer_by_id(payload->ld, &payload->peer_id)
-			    ->their_features);
+	json_add_hex_talarr(stream, "features", payload->their_features);
 	json_object_end(stream); /* .peer */
 }
 
@@ -1759,7 +1757,6 @@ REGISTER_PLUGIN_HOOK(peer_connected,
 void handle_peer_connected(struct lightningd *ld, const u8 *msg)
 {
 	struct node_id id;
-	u8 *their_features;
 	struct peer *peer;
 	struct peer_connected_hook_payload *hook_payload;
 	u64 connectd_counter;
@@ -1776,7 +1773,7 @@ void handle_peer_connected(struct lightningd *ld, const u8 *msg)
 					      &hook_payload->addr,
 					      &hook_payload->remote_addr,
 					      &hook_payload->incoming,
-					      &their_features,
+					      &hook_payload->their_features,
 					      &connect_reason,
 					      &connect_nsec)) {
 		u64 prev_connectd_counter, connected_time_nsec;
@@ -1786,7 +1783,7 @@ void handle_peer_connected(struct lightningd *ld, const u8 *msg)
 							&hook_payload->addr,
 							&hook_payload->remote_addr,
 							&hook_payload->incoming,
-							&their_features,
+							&hook_payload->their_features,
 							&connected_time_nsec)) {
 			fatal("Connectd gave bad CONNECT_PEER_(RE)CONNECTED message %s",
 			      tal_hex(msg, msg));
@@ -1830,10 +1827,12 @@ void handle_peer_connected(struct lightningd *ld, const u8 *msg)
 		/* If we connected to them, we know this is a good address. */
 		peer = new_peer(ld, 0, &id, &hook_payload->addr,
 				last_known_addr,
-				take(their_features), hook_payload->incoming);
+				hook_payload->their_features,
+				hook_payload->incoming);
 	} else {
 		tal_free(peer->their_features);
-		peer->their_features = tal_steal(peer, their_features);
+		peer->their_features = tal_dup_talarr(peer, u8,
+						      hook_payload->their_features);
 
 		/* Update known address. */
 		tal_free(peer->last_known_addr);
diff --git a/tests/plugins/peer_connected_logger_a.py b/tests/plugins/peer_connected_logger_a.py
index 5ded23068203..484bb5f2cacc 100755
--- a/tests/plugins/peer_connected_logger_a.py
+++ b/tests/plugins/peer_connected_logger_a.py
@@ -4,6 +4,8 @@
 """
 
 from pyln.client import Plugin
+import os
+import time
 
 plugin = Plugin()
 
@@ -11,7 +13,12 @@
 @plugin.hook('peer_connected')
 def on_connected(peer, plugin, **kwargs):
     print(f"peer_connected_logger_a {peer['id']} {peer}")
+    if plugin.get_option("logger_a_sleep") is True:
+        # Block until file appears
+        while not os.path.exists("unsleep"):
+            time.sleep(0.25)
     return {'result': 'continue'}
 
 
+plugin.add_option("logger_a_sleep", False, 'Block until unsleep file exists', opt_type='bool')
 plugin.run()
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 12a9d6e5ed5a..b61d1cd4c920 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -514,6 +514,29 @@ def check_disconnect():
     assert not l1.daemon.is_in_log(f"peer_connected_logger_b {l3id}")
 
 
+def test_plugin_connected_hook_disconnect_crash(node_factory, executor):
+    """A peer disconnnects between plugin hook invocations"""
+    opts = [{},
+            {'plugin':
+             [os.path.join(os.getcwd(),
+                           'tests/plugins/peer_connected_logger_a.py'),
+              os.path.join(os.getcwd(),
+                           'tests/plugins/peer_connected_logger_b.py')],
+             'logger_a_sleep': True},
+            ]
+
+    l1, l2 = node_factory.get_nodes(2, opts=opts)
+    executor.submit(l1.rpc.connect, l2.info['id'], 'localhost', l2.port)
+    l2.daemon.wait_for_log(f'plugin-peer_connected_logger_a.py: peer_connected_logger_a {l1.info["id"]}')
+    l1.stop()
+
+    # Now make first plugin continue...
+    open(os.path.join(l2.daemon.lightning_dir, TEST_NETWORK, "unsleep"), "w").close()
+
+    # Should get log from second
+    l2.daemon.wait_for_log(f'plugin-peer_connected_logger_b.py: peer_connected_logger_b {l1.info["id"]}')
+
+
 def test_peer_connected_remote_addr(node_factory):
     """This tests the optional tlv `remote_addr` being passed to a plugin.