Merge branch 'develop' into dmr/add-importlib_metadata-dependency

Author: David Robertson

.github/workflows/docs.yaml

@@ -22,7 +22,7 @@ jobs:
       - name: Setup mdbook
         uses: peaceiris/actions-mdbook@4b5ef36b314c2599664ca107bb8c02412548d79d # v1.1.14
         with:
-          mdbook-version: '0.4.9'
+          mdbook-version: '0.4.17'
 
       - name: Build the documentation
         # mdbook will only create an index.html if we're including docs/README.md in SUMMARY.md.

changelog.d/12320.misc

@@ -0,0 +1 @@
+Upgrade the version of `mdbook` in CI to 0.4.17.

changelog.d/12339.doc

@@ -0,0 +1 @@
+Update docstrings for `ReadWriteLock` tests.

changelog.d/12354.misc

@@ -0,0 +1 @@
+Fix a long-standing bug where `Linearizer`s could get stuck if a cancellation were to happen at the wrong time.

changelog.d/12358.misc

@@ -0,0 +1 @@
+Make `StreamToken.from_string` and `RoomStreamToken.parse` propagate cancellations instead of replacing them with `SynapseError`s.

changelog.d/12366.misc

@@ -0,0 +1 @@
+Remove temporary pin of signedjson<=1.1.1 that was added in Synapse 1.56.0.

changelog.d/12379.misc

@@ -48,7 +48,7 @@
     "unpaddedbase64>=1.1.0",
     "canonicaljson>=1.4.0",
     # we use the type definitions added in signedjson 1.1.
-    "signedjson>=1.1.0,<=1.1.1",
+    "signedjson>=1.1.0",
     "pynacl>=1.2.1",
     "idna>=2.5",
     # validating SSL certs for IP addresses requires service_identity 18.1.

synapse/python_dependencies.py

@@ -39,6 +39,7 @@
 from unpaddedbase64 import decode_base64
 from zope.interface import Interface
 
+from twisted.internet.defer import CancelledError
 from twisted.internet.interfaces import (
     IReactorCore,
     IReactorPluggableNameResolver,
@@ -540,6 +541,8 @@ async def parse(cls, store: "PurgeEventsStore", string: str) -> "RoomStreamToken
                     stream=stream,
                     instance_map=frozendict(instance_map),
                 )
+        except CancelledError:
+            raise
         except Exception:
             pass
         raise SynapseError(400, "Invalid room stream token %r" % (string,))
@@ -705,6 +708,8 @@ async def from_string(cls, store: "DataStore", string: str) -> "StreamToken":
             return cls(
                 await RoomStreamToken.parse(store, keys[0]), *(int(k) for k in keys[1:])
             )
+        except CancelledError:
+            raise
         except Exception:
             raise SynapseError(400, "Invalid stream token")
 

synapse/types.py

@@ -453,7 +453,11 @@ async def _acquire_lock(self, key: Hashable) -> _LinearizerEntry:
         #
         # This needs to happen while we hold the lock. We could put it on the
         # exit path, but that would slow down the uncontended case.
-        await self._clock.sleep(0)
+        try:
+            await self._clock.sleep(0)
+        except CancelledError:
+            self._release_lock(key, entry)
+            raise
 
         return entry
 

synapse/util/async_helpers.py

@@ -13,7 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Callable, Hashable, Tuple
+from typing import Hashable, Tuple
+
+from typing_extensions import Protocol
 
 from twisted.internet import defer, reactor
 from twisted.internet.base import ReactorBase
@@ -25,10 +27,15 @@
 from tests import unittest
 
 
+class UnblockFunction(Protocol):
+    def __call__(self, pump_reactor: bool = True) -> None:
+        ...
+
+
 class LinearizerTestCase(unittest.TestCase):
     def _start_task(
         self, linearizer: Linearizer, key: Hashable
-    ) -> Tuple["Deferred[None]", "Deferred[None]", Callable[[], None]]:
+    ) -> Tuple["Deferred[None]", "Deferred[None]", UnblockFunction]:
         """Starts a task which acquires the linearizer lock, blocks, then completes.
 
         Args:
@@ -52,11 +59,12 @@ async def task() -> None:
 
         d = defer.ensureDeferred(task())
 
-        def unblock() -> None:
+        def unblock(pump_reactor: bool = True) -> None:
             unblock_d.callback(None)
             # The next task, if it exists, will acquire the lock and require a kick of
             # the reactor to advance.
-            self._pump()
+            if pump_reactor:
+                self._pump()
 
         return d, acquired_d, unblock
 
@@ -212,3 +220,38 @@ def test_cancellation(self) -> None:
         )
         unblock3()
         self.successResultOf(d3)
+
+    def test_cancellation_during_sleep(self) -> None:
+        """Tests cancellation during the sleep just after waiting for a `Linearizer`."""
+        linearizer = Linearizer()
+
+        key = object()
+
+        d1, acquired_d1, unblock1 = self._start_task(linearizer, key)
+        self.assertTrue(acquired_d1.called)
+
+        # Create a second task, waiting for the first task.
+        d2, acquired_d2, _ = self._start_task(linearizer, key)
+        self.assertFalse(acquired_d2.called)
+
+        # Create a third task, waiting for the second task.
+        d3, acquired_d3, unblock3 = self._start_task(linearizer, key)
+        self.assertFalse(acquired_d3.called)
+
+        # Once the first task completes, cancel the waiting second task while it is
+        # sleeping just after acquiring the lock.
+        unblock1(pump_reactor=False)
+        self.successResultOf(d1)
+        d2.cancel()
+        self._pump()
+
+        self.assertTrue(d2.called)
+        self.failureResultOf(d2, CancelledError)
+
+        # The third task should continue running.
+        self.assertTrue(
+            acquired_d3.called,
+            "Third task did not get the lock after the second task was cancelled",
+        )
+        unblock3()
+        self.successResultOf(d3)

tests/util/test_linearizer.py

@@ -40,8 +40,8 @@ def _start_reader_or_writer(
 
         Returns:
             A tuple of three `Deferred`s:
-             * A `Deferred` that resolves with `return_value` once the reader or writer
-               completes successfully.
+             * A cancellable `Deferred` for the entire read or write operation that
+               resolves with `return_value` on successful completion.
              * A `Deferred` that resolves once the reader or writer acquires the lock.
              * A `Deferred` that blocks the reader or writer. Must be resolved by the
                caller to allow the reader or writer to release the lock and complete.
@@ -87,8 +87,8 @@ def _start_nonblocking_reader(
 
         Returns:
             A tuple of two `Deferred`s:
-             * A `Deferred` that resolves with `return_value` once the reader completes
-               successfully.
+             * A cancellable `Deferred` for the entire read operation that resolves with
+               `return_value` on successful completion.
              * A `Deferred` that resolves once the reader acquires the lock.
         """
         d, acquired_d, unblock_d = self._start_reader_or_writer(
@@ -106,8 +106,8 @@ def _start_nonblocking_writer(
 
         Returns:
             A tuple of two `Deferred`s:
-             * A `Deferred` that resolves with `return_value` once the writer completes
-               successfully.
+             * A cancellable `Deferred` for the entire write operation that resolves
+               with `return_value` on successful completion.
              * A `Deferred` that resolves once the writer acquires the lock.
         """
         d, acquired_d, unblock_d = self._start_reader_or_writer(