Fix bug where typing replication breaks (#17252)

This can happen on restarts of the service, due to old rooms being
pruned.

changelog.d/17252.bugfix

@@ -0,0 +1 @@
+Fix bug where typing updates would not be sent when using workers after a restart.

synapse/handlers/typing.py

@@ -477,9 +477,9 @@ async def get_all_typing_updates(
 
         rows = []
         for room_id in changed_rooms:
-            serial = self._room_serials[room_id]
-            if last_id < serial <= current_id:
-                typing = self._room_typing[room_id]
+            serial = self._room_serials.get(room_id)
+            if serial and last_id < serial <= current_id:
+                typing = self._room_typing.get(room_id, set())
                 rows.append((serial, [room_id, list(typing)]))
         rows.sort()
 

tests/handlers/test_typing.py

@@ -32,7 +32,7 @@
 from synapse.api.constants import EduTypes
 from synapse.api.errors import AuthError
 from synapse.federation.transport.server import TransportLayerServer
-from synapse.handlers.typing import TypingWriterHandler
+from synapse.handlers.typing import FORGET_TIMEOUT, TypingWriterHandler
 from synapse.http.federation.matrix_federation_agent import MatrixFederationAgent
 from synapse.server import HomeServer
 from synapse.types import JsonDict, Requester, StreamKeyType, UserID, create_requester
@@ -501,3 +501,54 @@ def test_typing_timeout(self) -> None:
                 }
             ],
         )
+
+    def test_prune_typing_replication(self) -> None:
+        """Regression test for `get_all_typing_updates` breaking when we prune
+        old updates
+        """
+        self.room_members = [U_APPLE, U_BANANA]
+
+        instance_name = self.hs.get_instance_name()
+
+        self.get_success(
+            self.handler.started_typing(
+                target_user=U_APPLE,
+                requester=create_requester(U_APPLE),
+                room_id=ROOM_ID,
+                timeout=10000,
+            )
+        )
+
+        rows, _, _ = self.get_success(
+            self.handler.get_all_typing_updates(
+                instance_name=instance_name,
+                last_id=0,
+                current_id=self.handler.get_current_token(),
+                limit=100,
+            )
+        )
+        self.assertEqual(rows, [(1, [ROOM_ID, [U_APPLE.to_string()]])])
+
+        self.reactor.advance(20000)
+
+        rows, _, _ = self.get_success(
+            self.handler.get_all_typing_updates(
+                instance_name=instance_name,
+                last_id=1,
+                current_id=self.handler.get_current_token(),
+                limit=100,
+            )
+        )
+        self.assertEqual(rows, [(2, [ROOM_ID, []])])
+
+        self.reactor.advance(FORGET_TIMEOUT)
+
+        rows, _, _ = self.get_success(
+            self.handler.get_all_typing_updates(
+                instance_name=instance_name,
+                last_id=1,
+                current_id=self.handler.get_current_token(),
+                limit=100,
+            )
+        )
+        self.assertEqual(rows, [])