tipc_publ_notify() walks a failed node's publication list after the node
lock has been released. Its list iterator is not protected by the name
table lock, which is only acquired inside tipc_publ_purge().

CPU A can save the next publication before entering tipc_publ_purge().
CPU B then takes nametbl_lock in tipc_named_rcv(), processes a WITHDRAWAL
for that publication, unlinks it with list_del_init(), and queues it for
freeing with kfree_rcu(). CPU A advances to the removed publication and
loops on its self-linked binding_node, stalling the CPU.

The kernel reported:

  rcu: INFO: rcu_sched self-detected stall on CPU
  Call Trace:
   tipc_publ_notify+0x3b5/0x650
   tipc_node_write_unlock+0x49d/0x5d0
   tipc_node_link_down+0x15c/0x4a0
   tipc_node_delete_links+0xfc/0x190
   bearer_disable+0x111/0x270
   __tipc_nl_bearer_disable+0x1db/0x2f0
   tipc_nl_bearer_disable+0x1c/0x30

Locking the entire traversal would also prevent the race, but would hold
nametbl_lock with bottom halves disabled while purging every publication.
A node with many publications could therefore cause excessive lock hold
times and delay other name-table operations.

Move the failed node's publications to a private list under nametbl_lock,
then select and unlink its first entry under the same lock before purging
it. Concurrent withdrawals can still unlink entries from this private
list, while publications arriving after node recovery stay on the node's
live list. No publication pointer is carried across an unlocked interval.

Unlink before the table lookup to make progress even if the lookup fails,
and acquire and release the lock for each publication. Check the private
list under the lock on every iteration, since concurrent withdrawals can
still remove entries. Keep withdrawal notifications and the final
rc_dests update in their existing order.

Remove the failed-node address argument from tipc_publ_notify() and its
purge helper since unlinking no longer needs a node lookup.

Fixes: 9db9fdd1983e ("tipc: avoid to asynchronously notify subscriptions")
Cc: stable@vger.kernel.org
Assisted-by: GPT-6-Astra
Signed-off-by: Chengfeng Ye <nicoyip.dev@gmail.com>
---
 net/tipc/name_distr.c | 34 +++++++++++++++++++++++-----------
 net/tipc/name_distr.h |  2 +-
 net/tipc/node.c       |  2 +-
 3 files changed, 25 insertions(+), 13 deletions(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index ba4f4906e13b..34867b69044c 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -227,38 +227,50 @@ void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
  * tipc_publ_purge - remove publication associated with a failed node
  * @net: the associated network namespace
  * @p: the publication to remove
- * @addr: failed node's address
  *
  * Invoked for each publication issued by a newly failed node.
  * Removes publication structure from name table & deletes it.
+ * The caller must hold nametbl_lock and unlink the node subscription.
  */
-static void tipc_publ_purge(struct net *net, struct publication *p, u32 addr)
+static void tipc_publ_purge(struct net *net, struct publication *p)
 {
-	struct tipc_net *tn = tipc_net(net);
 	struct publication *_p;
 	struct tipc_uaddr ua;
 
 	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, p->scope, p->sr.type,
 		   p->sr.lower, p->sr.upper);
-	spin_lock_bh(&tn->nametbl_lock);
 	_p = tipc_nametbl_remove_publ(net, &ua, &p->sk, p->key);
-	if (_p)
-		tipc_node_unsubscribe(net, &_p->binding_node, addr);
-	spin_unlock_bh(&tn->nametbl_lock);
 	if (_p)
 		kfree_rcu(_p, rcu);
 }
 
 void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
-		      u32 addr, u16 capabilities)
+		      u16 capabilities)
 {
 	struct name_table *nt = tipc_name_table(net);
 	struct tipc_net *tn = tipc_net(net);
 
-	struct publication *publ, *tmp;
+	struct publication *publ;
+	LIST_HEAD(purge_list);
+
+	spin_lock_bh(&tn->nametbl_lock);
+	/* Leave new publications on the node's list during the purge. */
+	list_splice_init(nsub_list, &purge_list);
+	spin_unlock_bh(&tn->nametbl_lock);
+
+	for (;;) {
+		spin_lock_bh(&tn->nametbl_lock);
+		if (list_empty(&purge_list)) {
+			spin_unlock_bh(&tn->nametbl_lock);
+			break;
+		}
+		publ = list_first_entry(&purge_list, struct publication,
+					binding_node);
+		list_del_init(&publ->binding_node);
+		tipc_publ_purge(net, publ);
+		spin_unlock_bh(&tn->nametbl_lock);
+	}
 
-	list_for_each_entry_safe(publ, tmp, nsub_list, binding_node)
-		tipc_publ_purge(net, publ, addr);
 	spin_lock_bh(&tn->nametbl_lock);
 	if (!(capabilities & TIPC_NAMED_BCAST))
 		nt->rc_dests--;
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index c677f6f082df..8debe23469b2 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -74,6 +74,6 @@ void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
 		    u16 *rcv_nxt, bool *open);
 void tipc_named_reinit(struct net *net);
 void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
-		      u32 addr, u16 capabilities);
+		      u16 capabilities);
 
 #endif
diff --git a/net/tipc/node.c b/net/tipc/node.c
index bd91378b7540..9a218d137c45 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -422,7 +422,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	write_unlock_bh(&n->lock);
 
 	if (flags & TIPC_NOTIFY_NODE_DOWN)
-		tipc_publ_notify(net, publ_list, node, n->capabilities);
+		tipc_publ_notify(net, publ_list, n->capabilities);
 
 	if (flags & TIPC_NOTIFY_NODE_UP)
 		tipc_named_node_up(net, node, n->capabilities);
-- 
2.43.0