From 0d88d1ae2802e47f08b0db72987b8d68ecfc16d8 Mon Sep 17 00:00:00 2001
From: Jarrod Johnson <jjohnson2@lenovo.com>
Date: Fri, 4 Sep 2026 12:43:18 -0400
Subject: [PATCH] Replace asyncio Locks with re-entrant behavior

If connect_to_leader calls itself, let it use it's own lock.
---
 .../confluent/collective/manager.py           | 24 +++++++++++++++----
 1 file changed, 20 insertions(+), 4 deletions(-)

diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py
index 89ea87d9..9db594fd 100644
--- a/confluent_server/confluent/collective/manager.py
+++ b/confluent_server/confluent/collective/manager.py
@@ -66,17 +66,33 @@ class ContextBool(object):
     def __init__(self):
         self.active = False
         self.mylock = None
+        self.owner = None
+        self.depth = 0
 
     async def __aenter__(self):
-        self.active = True
         if self.mylock is None:
             self.mylock = asyncio.Lock()
-        return await self.mylock.__aenter__()
+        currtask = asyncio.current_task()
+        if self.owner is not None and self.owner is currtask:
+            # reentrant acquisition by the same task, just track depth
+            self.depth += 1
+            return None
+        rv = await self.mylock.__aenter__()
+        self.owner = currtask
+        self.depth = 1
+        self.active = True
+        return rv
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        self.active = False
         if self.mylock is None:
             self.mylock = asyncio.Lock()
+        currtask = asyncio.current_task()
+        if self.owner is currtask and self.depth > 1:
+            self.depth -= 1
+            return False
+        self.owner = None
+        self.depth = 0
+        self.active = False
         return await self.mylock.__aexit__(exc_type, exc_val, exc_tb)
 
 connecting = ContextBool()
@@ -103,7 +119,7 @@ async def connect_to_leader(cert=None, name=None, leader=None, remote=None, isre
         return False
     async with connecting:
         if cfm._initlock is None:
-            cfm._initlock = asyncio.Lock()
+            cfm._initlock = ContextBool()
         async with cfm._initlock:
             # remote is a socket...
             banner = await tlvdata.recv(remote)  # the banner