Release a lock directly while the server shuts down

A lock released in the background ran its delete as a background
process. Once shutdown has begun, starting one raises. So a persist
that had committed failed as it released the purge lock, and the lock's
row stayed until the reaper removed it. The delete is now awaited once
the server is shutting down, as it was before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PVLsNh6akt4ifEizY4D1ta

synapse/server.py

@@ -371,6 +371,11 @@ class HomeServer(metaclass=abc.ABCMeta):
         # For every server we spawn in the process, track it in the metrics
         synapse_server_name_info.labels(**{SERVER_NAME_LABEL: self.hostname}).set(1)
 
+    def is_shutting_down(self) -> bool:
+        """Whether `shutdown` has been called. Background processes can no
+        longer be started."""
+        return self._is_shutdown
+
     def run_as_background_process(
         self,
         desc: "LiteralString",

synapse/storage/databases/main/lock.py

@@ -532,7 +532,8 @@ class Lock:
             },
             desc="drop_lock",
         )
-        if wait:
+        if wait or self._hs.is_shutting_down():
+            # No background process can start once shutdown has begun.
             await delete
         else:
             self._hs.run_as_background_process("Lock.release", lambda: delete)

tests/handlers/test_worker_lock.py

@@ -59,6 +59,28 @@ class WorkerLockTestCase(unittest.HomeserverTestCase):
         self.get_success(d2)
         self.get_success(lock2.__aexit__(None, None, None))
 
+    def test_release_in_background_while_shutting_down(self) -> None:
+        """A lock released in the background once shutdown has begun is still
+        released, without failing the caller."""
+        lock = self.worker_lock_handler.acquire_read_write_lock(
+            "name", "key", write=False, release_in_background=True
+        )
+        self.get_success(lock.__aenter__())
+
+        # `shutdown` sets this first, then waits for the rest to stop.
+        self.hs._is_shutdown = True
+        try:
+            self.get_success(lock.__aexit__(None, None, None))
+        finally:
+            self.hs._is_shutdown = False
+
+        rows = self.get_success(
+            self.store.db_pool.simple_select_list(
+                "worker_read_write_locks", {"lock_name": "name"}, ["token"]
+            )
+        )
+        self.assertEqual(rows, [])
+
     def test_timeouts_for_lock_locally(self) -> None:
         """
         Test that we regularly retry to reacquire locks.