fix: notify watcher after releasing the enforcer lock (#408)

Author: hsluoyz

casbin/synced_enforcer.py

@@ -35,15 +35,123 @@ def value(self, value):
             self._value = value
 
 
+class _DeferredNotifier:
+    """Queues the watcher notifications raised inside a locked section and delivers them
+    once the lock has been released.
+
+    Watchers have a lock of their own, and some of them (the redis watcher, for one) hold
+    it while running the update callback, which calls back into the enforcer to reload the
+    policy. Notifying from inside the enforcer lock would let the two threads take those
+    two locks in opposite orders and deadlock:
+    https://github.com/casbin/pycasbin/issues/408
+    """
+
+    def __init__(self):
+        self._local = threading.local()
+
+    def _stack(self):
+        stack = getattr(self._local, "stack", None)
+        if stack is None:
+            stack = []
+            self._local.stack = stack
+        return stack
+
+    def defer(self, func, args, kwargs):
+        """queues a call, and returns whether it was queued. False means no lock is held
+        by this thread, so the caller has to run the call itself."""
+        stack = self._stack()
+        if not stack:
+            return False
+        stack[-1].append((func, args, kwargs))
+        return True
+
+    def begin(self):
+        """starts collecting, called right after the lock has been taken."""
+        self._stack().append([])
+
+    def end(self, suppress_errors=False):
+        """delivers what was collected, called right after the lock has been released."""
+        stack = self._stack()
+        queued = stack.pop()
+        if stack:
+            # an outer lock is still held, so hand the calls over to it
+            stack[-1].extend(queued)
+            return
+
+        for func, args, kwargs in queued:
+            try:
+                func(*args, **kwargs)
+            except Exception:
+                # the caller is already unwinding an exception of its own, and hiding it
+                # behind a notification failure would only make the original harder to find
+                if not suppress_errors:
+                    raise
+
+
+class _NotifyAfterUnlock:
+    """wraps one of the enforcer locks so that _DeferredNotifier collects notifications
+    for as long as the lock is held."""
+
+    def __init__(self, lock, notifier):
+        self._lock = lock
+        self._notifier = notifier
+
+    def __enter__(self):
+        self._lock.__enter__()
+        self._notifier.begin()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        try:
+            self._lock.__exit__(exc_type, exc_value, traceback)
+        finally:
+            self._notifier.end(suppress_errors=exc_type is not None)
+        return False
+
+
+class _DeferredWatcher:
+    """stands in for the user's watcher on the wrapped enforcer, and routes every call it
+    makes through a _DeferredNotifier."""
+
+    def __init__(self, watcher, notifier):
+        self._watcher = watcher
+        self._notifier = notifier
+
+    def __bool__(self):
+        return bool(self._watcher)
+
+    def __getattr__(self, name):
+        watcher = self.__dict__.get("_watcher")
+        if watcher is None:
+            raise AttributeError(name)
+
+        attr = getattr(watcher, name)
+        if not callable(attr):
+            return attr
+
+        notifier = self._notifier
+
+        def deferred(*args, **kwargs):
+            # a deferred call has no result to hand back yet, but the enforcer ignores
+            # what the watcher returns anyway
+            if notifier.defer(attr, args, kwargs):
+                return None
+            return attr(*args, **kwargs)
+
+        return deferred
+
+
 class SyncedEnforcer:
     """SyncedEnforcer wraps Enforcer and provides synchronized access.
     It's also a drop-in replacement for Enforcer"""
 
     def __init__(self, model=None, adapter=None):
         self._e = self._new_enforcer(model, adapter)
         self._rwlock = RWLockWrite()
-        self._rl = self._rwlock.gen_rlock()
-        self._wl = self._rwlock.gen_wlock()
+        self._notifier = _DeferredNotifier()
+        self._rl = _NotifyAfterUnlock(self._rwlock.gen_rlock(), self._notifier)
+        self._wl = _NotifyAfterUnlock(self._rwlock.gen_wlock(), self._notifier)
+        self._watcher = None
         self._auto_loading = AtomicBool(False)
         self._auto_loading_thread = None
 
@@ -115,9 +223,19 @@ def set_adapter(self, adapter):
             self._e.set_adapter(adapter)
 
     def set_watcher(self, watcher):
-        """sets the current watcher."""
+        """sets the current watcher.
+
+        It is notified only once the enforcer lock has been released, so a watcher that
+        holds a lock of its own while running the update callback cannot deadlock against
+        the enforcer."""
         with self._wl:
-            self._e.set_watcher(watcher)
+            self._watcher = watcher
+            self._e.set_watcher(None if watcher is None else _DeferredWatcher(watcher, self._notifier))
+
+    def get_watcher(self):
+        """gets the watcher that was passed to set_watcher()."""
+        with self._rl:
+            return self._watcher
 
     def set_effector(self, eft):
         """sets the current effector."""

tests/test_synced_enforcer.py

@@ -0,0 +1,143 @@
+# Copyright 2026 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+from unittest import TestCase
+
+import casbin
+from tests.test_enforcer import get_examples
+
+
+class LockingWatcher:
+    """mimics the redis watcher, which holds a lock of its own both while notifying the
+    other nodes and while running the update callback."""
+
+    def __init__(self):
+        self.mutex = threading.Lock()
+        self.callback = None
+        self.updates = []
+
+    def set_update_callback(self, callback):
+        self.callback = callback
+
+    def update(self):
+        with self.mutex:
+            self.updates.append("update")
+        return True
+
+    def close(self):
+        pass
+
+
+class TestSyncedEnforcerWatcher(TestCase):
+    """the lock the enforcer takes and the lock the watcher takes have to be taken in the
+    same order everywhere, or the two deadlock. See
+    https://github.com/casbin/pycasbin/issues/408"""
+
+    def get_enforcer(self, adapter):
+        return casbin.SyncedEnforcer(get_examples("basic_model.conf"), adapter)
+
+    def test_watcher_is_notified(self):
+        w = LockingWatcher()
+        e = self.get_enforcer(casbin.persist.Adapter())
+        e.set_watcher(w)
+
+        e.add_policy("alice", "data1", "read")
+
+        # the notification is deferred, but only until the lock is released: it still has
+        # to have been delivered by the time add_policy() returns
+        self.assertEqual(w.updates, ["update"])
+        self.assertTrue(e.enforce("alice", "data1", "read"))
+
+    def test_get_watcher_returns_the_watcher_that_was_set(self):
+        w = LockingWatcher()
+        e = self.get_enforcer(casbin.persist.Adapter())
+        e.set_watcher(w)
+
+        self.assertIs(e.get_watcher(), w)
+
+    def test_watcher_is_notified_outside_the_enforcer_lock(self):
+        w = LockingWatcher()
+        e = self.get_enforcer(casbin.persist.Adapter())
+        e.set_watcher(w)
+
+        reader_finished = threading.Event()
+
+        def update():
+            # a reader can only get through here if the write lock has been released
+            reader = threading.Thread(target=lambda: (e.get_policy(), reader_finished.set()), daemon=True)
+            reader.start()
+            reader.join(timeout=5)
+
+        w.update = update
+        e.add_policy("alice", "data1", "read")
+
+        self.assertTrue(reader_finished.is_set(), "the watcher was notified while the enforcer was still locked")
+
+    def test_save_policy_notifies_outside_the_enforcer_lock(self):
+        """save_policy() notifies the watcher too, and it holds the read lock while doing
+        so, so the same deferral has to apply there."""
+        w = LockingWatcher()
+        e = self.get_enforcer(casbin.persist.Adapter())
+        e.set_watcher(w)
+
+        writer_finished = threading.Event()
+
+        def update():
+            # a writer can only get through here if the read lock has been released
+            writer = threading.Thread(target=lambda: (e.load_policy(), writer_finished.set()), daemon=True)
+            writer.start()
+            writer.join(timeout=5)
+
+        w.update = update
+        e.save_policy()
+
+        self.assertTrue(writer_finished.is_set(), "the watcher was notified while the enforcer was still locked")
+
+    def test_callback_holding_the_watcher_lock_does_not_deadlock(self):
+        """reproduces the issue: the subscribe thread holds the watcher lock and wants the
+        enforcer lock, while the thread changing the policy holds the enforcer lock and
+        wants the watcher lock."""
+        w = LockingWatcher()
+        enforcer_locked = threading.Event()
+        callback_reached = threading.Event()
+
+        class SlowAdapter(casbin.persist.Adapter):
+            def add_policy(self, sec, ptype, rule):
+                # the write lock is held for as long as we stay in here, so this is where
+                # the subscribe thread gets its chance to ask for it
+                enforcer_locked.set()
+                callback_reached.wait(timeout=5)
+
+        e = self.get_enforcer(SlowAdapter())
+        e.set_watcher(w)
+        w.set_update_callback(e.load_policy)
+
+        def subscribe():
+            with w.mutex:
+                enforcer_locked.wait(timeout=5)
+                callback_reached.set()
+                w.callback()
+
+        subscriber = threading.Thread(target=subscribe, daemon=True)
+        subscriber.start()
+
+        added = threading.Event()
+        writer = threading.Thread(target=lambda: (e.add_policy("alice", "data1", "read"), added.set()), daemon=True)
+        writer.start()
+
+        self.assertTrue(added.wait(timeout=10), "add_policy deadlocked against the watcher lock")
+        subscriber.join(timeout=10)
+        self.assertFalse(subscriber.is_alive(), "the update callback deadlocked against the enforcer lock")
+        self.assertEqual(w.updates, ["update"])