diff --git a/fs/ocfs2/cluster/heartbeat.c b/fs/ocfs2/cluster/heartbeat.c
index c25aa2ff3ebc..834210af4e4b 100644
--- a/fs/ocfs2/cluster/heartbeat.c
+++ b/fs/ocfs2/cluster/heartbeat.c
@@ -1492,13 +1492,38 @@ void o2hb_init(void)
 	o2hb_debug_init();
 }
 
-/* if we're already in a callback then we're already serialized by the sem */
-static void o2hb_fill_node_map_from_callback(unsigned long *map,
-					     unsigned int bits)
+static void __o2hb_fill_node_map(unsigned long *map, unsigned int bits)
 {
 	bitmap_copy(map, o2hb_live_node_bitmap, bits);
 }
 
+void o2hb_callback_read_lock(void)
+{
+	down_read(&o2hb_callback_sem);
+}
+
+void o2hb_callback_read_unlock(void)
+{
+	up_read(&o2hb_callback_sem);
+}
+
+void o2hb_synchronize_callbacks(void)
+{
+	down_write(&o2hb_callback_sem);
+	up_write(&o2hb_callback_sem);
+}
+
+/*
+ * Callers must already hold o2hb_callback_sem for read or write so the copy
+ * stays serialized with callback delivery.
+ */
+void o2hb_fill_node_map_locked(unsigned long *map, unsigned int bits)
+{
+	spin_lock(&o2hb_live_lock);
+	__o2hb_fill_node_map(map, bits);
+	spin_unlock(&o2hb_live_lock);
+}
+
 /*
  * get a map of all nodes that are heartbeating in any regions
  */
@@ -1506,11 +1531,9 @@ void o2hb_fill_node_map(unsigned long *map, unsigned int bits)
 {
 	/* callers want to serialize this map and callbacks so that they
 	 * can trust that they don't miss nodes coming to the party */
-	down_read(&o2hb_callback_sem);
-	spin_lock(&o2hb_live_lock);
-	o2hb_fill_node_map_from_callback(map, bits);
-	spin_unlock(&o2hb_live_lock);
-	up_read(&o2hb_callback_sem);
+	o2hb_callback_read_lock();
+	o2hb_fill_node_map_locked(map, bits);
+	o2hb_callback_read_unlock();
 }
 EXPORT_SYMBOL_GPL(o2hb_fill_node_map);
 
@@ -2609,7 +2632,7 @@ int o2hb_check_node_heartbeating_no_sem(u8 node_num)
 	unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
 
 	spin_lock(&o2hb_live_lock);
-	o2hb_fill_node_map_from_callback(testing_map, O2NM_MAX_NODES);
+	__o2hb_fill_node_map(testing_map, O2NM_MAX_NODES);
 	spin_unlock(&o2hb_live_lock);
 	if (!test_bit(node_num, testing_map)) {
 		mlog(ML_HEARTBEAT,
@@ -2626,7 +2649,7 @@ int o2hb_check_node_heartbeating_from_callback(u8 node_num)
 {
 	unsigned long testing_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
 
-	o2hb_fill_node_map_from_callback(testing_map, O2NM_MAX_NODES);
+	o2hb_fill_node_map_locked(testing_map, O2NM_MAX_NODES);
 	if (!test_bit(node_num, testing_map)) {
 		mlog(ML_HEARTBEAT,
 		     "node (%u) does not have heartbeating enabled.\n",
diff --git a/fs/ocfs2/cluster/heartbeat.h b/fs/ocfs2/cluster/heartbeat.h
index 8ef8c1b9eeb7..2ca2b657583c 100644
--- a/fs/ocfs2/cluster/heartbeat.h
+++ b/fs/ocfs2/cluster/heartbeat.h
@@ -58,6 +58,11 @@ int o2hb_register_callback(const char *region_uuid,
 			   struct o2hb_callback_func *hc);
 void o2hb_unregister_callback(const char *region_uuid,
 			      struct o2hb_callback_func *hc);
+void o2hb_callback_read_lock(void);
+void o2hb_callback_read_unlock(void);
+void o2hb_synchronize_callbacks(void);
+void o2hb_fill_node_map_locked(unsigned long *map,
+			       unsigned int bits);
 void o2hb_fill_node_map(unsigned long *map,
 			unsigned int bits);
 void o2hb_exit(void);
diff --git a/fs/ocfs2/cluster/nodemanager.c b/fs/ocfs2/cluster/nodemanager.c
index ebdf0bdbb8c6..e08850a5d736 100644
--- a/fs/ocfs2/cluster/nodemanager.c
+++ b/fs/ocfs2/cluster/nodemanager.c
@@ -326,6 +326,7 @@ static ssize_t o2nm_node_local_store(struct config_item *item, const char *page,
 	struct o2nm_node *node = to_o2nm_node(item);
 	struct o2nm_cluster *cluster;
 	unsigned long tmp;
+	bool starting = false;
 	char *p = (char *)page;
 	ssize_t ret;
 
@@ -362,6 +363,7 @@ static ssize_t o2nm_node_local_store(struct config_item *item, const char *page,
 		ret = o2net_start_listening(node);
 		if (ret)
 			goto out;
+		starting = true;
 	}
 
 	if (!tmp && cluster->cl_has_local &&
@@ -375,6 +377,8 @@ static ssize_t o2nm_node_local_store(struct config_item *item, const char *page,
 	if (node->nd_local) {
 		cluster->cl_has_local = tmp;
 		cluster->cl_local_node = node->nd_num;
+		if (starting)
+			o2net_complete_start_listening(node);
 	}
 
 	ret = count;
diff --git a/fs/ocfs2/cluster/tcp.c b/fs/ocfs2/cluster/tcp.c
index 09a1f3b77bb8..e62c1ef8223b 100644
--- a/fs/ocfs2/cluster/tcp.c
+++ b/fs/ocfs2/cluster/tcp.c
@@ -105,6 +105,8 @@ static struct socket *o2net_listen_sock;
  * destroying the work queue.
  */
 static struct workqueue_struct *o2net_wq;
+/* Heartbeat callbacks stay registered across local-node off/on. */
+static bool o2net_listening;
 static struct work_struct o2net_listen_work;
 
 static struct o2hb_callback_func o2net_hb_up, o2net_hb_down;
@@ -1692,6 +1694,19 @@ static void o2net_still_up(struct work_struct *work)
 
 /* ------------------------------------------------------------ */
 
+static void o2net_hb_node_up(struct o2net_node *nn)
+{
+	/* ensure an immediate connect attempt */
+	nn->nn_last_connect_attempt = jiffies -
+		(msecs_to_jiffies(o2net_reconnect_delay()) + 1);
+
+	spin_lock(&nn->nn_lock);
+	atomic_set(&nn->nn_timeout, 0);
+	if (nn->nn_persistent_error)
+		o2net_set_nn_state(nn, NULL, 0, 0);
+	spin_unlock(&nn->nn_lock);
+}
+
 void o2net_disconnect_node(struct o2nm_node *node)
 {
 	struct o2net_node *nn = o2net_nn_from_num(node->nd_num);
@@ -1713,41 +1728,39 @@ void o2net_disconnect_node(struct o2nm_node *node)
 static void o2net_hb_node_down_cb(struct o2nm_node *node, int node_num,
 				  void *data)
 {
+	u8 this_node;
+
 	o2quo_hb_down(node_num);
 
 	if (!node)
-		return;
+		goto out;
 
-	if (node_num != o2nm_this_node())
+	this_node = o2nm_this_node();
+	if (!READ_ONCE(o2net_listening) || this_node == O2NM_MAX_NODES)
+		goto out;
+
+	if (node_num != this_node)
 		o2net_disconnect_node(node);
 
+out:
 	BUG_ON(atomic_read(&o2net_connected_peers) < 0);
 }
 
 static void o2net_hb_node_up_cb(struct o2nm_node *node, int node_num,
 				void *data)
 {
-	struct o2net_node *nn = o2net_nn_from_num(node_num);
+	u8 this_node;
 
 	o2quo_hb_up(node_num);
 
 	BUG_ON(!node);
 
-	/* ensure an immediate connect attempt */
-	nn->nn_last_connect_attempt = jiffies -
-		(msecs_to_jiffies(o2net_reconnect_delay()) + 1);
+	this_node = o2nm_this_node();
+	if (!READ_ONCE(o2net_listening) || this_node == O2NM_MAX_NODES)
+		return;
 
-	if (node_num != o2nm_this_node()) {
-		/* believe it or not, accept and node heartbeating testing
-		 * can succeed for this node before we got here.. so
-		 * only use set_nn_state to clear the persistent error
-		 * if that hasn't already happened */
-		spin_lock(&nn->nn_lock);
-		atomic_set(&nn->nn_timeout, 0);
-		if (nn->nn_persistent_error)
-			o2net_set_nn_state(nn, NULL, 0, 0);
-		spin_unlock(&nn->nn_lock);
-	}
+	if (node_num != this_node)
+		o2net_hb_node_up(o2net_nn_from_num(node_num));
 }
 
 void o2net_unregister_hb_callbacks(void)
@@ -1756,6 +1769,37 @@ void o2net_unregister_hb_callbacks(void)
 	o2hb_unregister_callback(NULL, &o2net_hb_down);
 }
 
+/*
+ * Delay heartbeat-driven network work until the local node is fully published
+ * through o2nm_this_node(), then replay the nodes that are already live while
+ * callback delivery stays blocked.
+ */
+void o2net_complete_start_listening(struct o2nm_node *node)
+{
+	unsigned long live_nodes[BITS_TO_LONGS(O2NM_MAX_NODES)];
+	unsigned long node_num;
+	u8 local_node;
+
+	local_node = o2nm_this_node();
+	if (WARN_ON_ONCE(local_node == O2NM_MAX_NODES))
+		return;
+	if (WARN_ON_ONCE(local_node != node->nd_num))
+		return;
+	if (WARN_ON_ONCE(!o2net_wq))
+		return;
+
+	o2hb_callback_read_lock();
+	WRITE_ONCE(o2net_listening, true);
+	o2hb_fill_node_map_locked(live_nodes, O2NM_MAX_NODES);
+	for_each_set_bit(node_num, live_nodes, O2NM_MAX_NODES) {
+		if (node_num == local_node)
+			continue;
+
+		o2net_hb_node_up(o2net_nn_from_num(node_num));
+	}
+	o2hb_callback_read_unlock();
+}
+
 int o2net_register_hb_callbacks(void)
 {
 	int ret;
@@ -2034,6 +2078,8 @@ int o2net_start_listening(struct o2nm_node *node)
 {
 	int ret = 0;
 
+	if (WARN_ON_ONCE(READ_ONCE(o2net_listening)))
+		return -EBUSY;
 	BUG_ON(o2net_wq != NULL);
 	BUG_ON(o2net_listen_sock != NULL);
 
@@ -2065,6 +2111,9 @@ void o2net_stop_listening(struct o2nm_node *node)
 	BUG_ON(o2net_wq == NULL);
 	BUG_ON(o2net_listen_sock == NULL);
 
+	WRITE_ONCE(o2net_listening, false);
+	o2hb_synchronize_callbacks();
+
 	/* stop the listening socket from generating work */
 	write_lock_bh(&sock->sk->sk_callback_lock);
 	sock->sk->sk_data_ready = sock->sk->sk_user_data;
diff --git a/fs/ocfs2/cluster/tcp.h b/fs/ocfs2/cluster/tcp.h
index a75b551d31c7..2e86d42b5faf 100644
--- a/fs/ocfs2/cluster/tcp.h
+++ b/fs/ocfs2/cluster/tcp.h
@@ -96,6 +96,7 @@ struct o2nm_node;
 int o2net_register_hb_callbacks(void);
 void o2net_unregister_hb_callbacks(void);
 int o2net_start_listening(struct o2nm_node *node);
+void o2net_complete_start_listening(struct o2nm_node *node);
 void o2net_stop_listening(struct o2nm_node *node);
 void o2net_disconnect_node(struct o2nm_node *node);
 int o2net_num_connected_peers(void);