diff --git a/synapse/server.py b/synapse/server.py
index fffc144aed..a731bb3066 100644
--- a/synapse/server.py
+++ b/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",
diff --git a/synapse/storage/databases/main/lock.py b/synapse/storage/databases/main/lock.py
index 3bbb2a5780..963ce9d1da 100644
--- a/synapse/storage/databases/main/lock.py
+++ b/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)
diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py
index 5563ec4683..fd209b6e25 100644
--- a/tests/handlers/test_worker_lock.py
+++ b/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.