task: metadata: install: pip install -e .[dev]
patch: diff --git a/changelog.d/12354.misc b/changelog.d/12354.misc
new file mode 100644
index 000000000..e3b8950fa
--- /dev/null
+++ b/changelog.d/12354.misc
@@ -0,0 +1,1 @@
+Update docstrings for `ReadWriteLock` tests.
diff --git a/changelog.d/12358.misc b/changelog.d/12358.misc
new file mode 100644
index 000000000..fcacbcba5
--- /dev/null
+++ b/changelog.d/12358.misc
@@ -0,0 +1,1 @@
+Fix a long-standing bug where `Linearizer`s could get stuck if a cancellation were to happen at the wrong time.
diff --git a/changelog.d/12366.misc b/changelog.d/12366.misc
new file mode 100644
index 000000000..33d8e6c71
--- /dev/null
+++ b/changelog.d/12366.misc
@@ -0,0 +1,1 @@
+Make `StreamToken.from_string` and `RoomStreamToken.parse` propagate cancellations instead of replacing them with `SynapseError`s.
diff --git a/synapse/types.py b/synapse/types.py
index 6bbefb6fa..9ac688b23 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -39,6 +39,7 @@ from typing_extensions import TypedDict
 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 @@ class 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 @@ class 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")
 
diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py
index 4b2a16a6a..650e44de2 100644
--- a/synapse/util/async_helpers.py
+++ b/synapse/util/async_helpers.py
@@ -453,7 +453,11 @@ class Linearizer:
         #
         # 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
 

problem_statement: `Linearizer` can get stuck due to cancellation
Once a `Linearizer` task acquires the lock, it does a 0-duration sleep to reset the call stack:
https://github.com/matrix-org/synapse/blob/605d161d7d585847fd1bb98d14d5281daeac8e86/synapse/util/async_helpers.py#L466

In the rare event that the task is cancelled while sleeping, the lock is not released and the `Linearizer` becomes stuck.

Can be reproduced by modifying the cancellation test slightly. Cancelling the 2nd task after 1st task releases the lock (instead of before) will make the test get stuck indefinitely:
```py
        with cm1:
            pass

        d2.cancel()
```
instead of 
https://github.com/matrix-org/synapse/blob/release-v1.53/tests/util/test_linearizer.py#L159-L162

repo: matrix-org/synapse
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/tests/util/test_linearizer.py b/tests/util/test_linearizer.py
index c2a209e63..47a1cfbdc 100644
--- a/tests/util/test_linearizer.py
+++ b/tests/util/test_linearizer.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 synapse.util.async_helpers import Linearizer
 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 @@ class LinearizerTestCase(unittest.TestCase):
 
         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 @@ class LinearizerTestCase(unittest.TestCase):
         )
         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)
diff --git a/tests/util/test_rwlock.py b/tests/util/test_rwlock.py
index 0c8422619..5da04362a 100644
--- a/tests/util/test_rwlock.py
+++ b/tests/util/test_rwlock.py
@@ -40,8 +40,8 @@ class ReadWriteLockTestCase(unittest.TestCase):
 
         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 @@ class ReadWriteLockTestCase(unittest.TestCase):
 
         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 @@ class ReadWriteLockTestCase(unittest.TestCase):
 
         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(