Fix typing being reset causing infinite syncs

synapse/app/synchrotron.py

@@ -236,6 +236,13 @@ def stream_positions(self):
         return {"typing": self._latest_room_serial}
 
     def process_replication_rows(self, token, rows):
+        if self._latest_room_serial > token:
+            # The master has gone backwards. To prevent inconsistent data, just
+            # clear everything.
+            self._room_serials = {}
+            self._room_typing = {}
+
+        # Set the latest serial token to whatever the server gave us.
         self._latest_room_serial = token
 
         for row in rows:

tests/rest/client/v2_alpha/test_sync.py

@@ -15,6 +15,7 @@
 
 from mock import Mock
 
+from synapse.rest.client.v1 import admin, login, room
 from synapse.rest.client.v2_alpha import sync
 
 from tests import unittest
@@ -65,3 +66,102 @@ def test_sync_presence_disabled(self):
                 ["next_batch", "rooms", "account_data", "to_device", "device_lists"]
             ).issubset(set(channel.json_body.keys()))
         )
+
+
+class SyncTypingTests(unittest.HomeserverTestCase):
+
+    servlets = [
+        admin.register_servlets,
+        room.register_servlets,
+        login.register_servlets,
+        sync.register_servlets
+    ]
+    user_id = True
+    hijack_auth = False
+
+    def test_sync_backwards_typing(self):
+
+        # Register the user who gets notified
+        user_id = self.register_user("user", "pass")
+        access_token = self.login("user", "pass")
+
+        # Register the user who sends the message
+        other_user_id = self.register_user("otheruser", "pass")
+        other_access_token = self.login("otheruser", "pass")
+
+        # Create a room
+        room = self.helper.create_room_as(user_id, tok=access_token)
+
+        # Invite the other person
+        self.helper.invite(room=room, src=user_id, tok=access_token, targ=other_user_id)
+
+        # The other user joins
+        self.helper.join(room=room, user=other_user_id, tok=other_access_token)
+
+        # The other user sends some messages
+        self.helper.send(room, body="Hi!", tok=other_access_token)
+        self.helper.send(room, body="There!", tok=other_access_token)
+
+        request, channel = self.make_request(
+            "PUT",
+            "/rooms/%s/typing/%s?access_token=%s" % (room, other_user_id, other_access_token),
+            b'{"typing": true, "timeout": 30000}',
+        )
+        self.render(request)
+        self.assertEquals(200, channel.code)
+
+        request, channel = self.make_request("GET", "/sync?access_token=%s" % (access_token,))
+        self.render(request)
+        self.assertEquals(200, channel.code)
+        next_batch = channel.json_body["next_batch"]
+
+        request, channel = self.make_request(
+            "PUT",
+            "/rooms/%s/typing/%s?access_token=%s" % (room, other_user_id, other_access_token),
+            b'{"typing": false}',
+        )
+        self.render(request)
+        self.assertEquals(200, channel.code)
+
+
+        request, channel = self.make_request(
+            "PUT",
+            "/rooms/%s/typing/%s?access_token=%s" % (room, other_user_id, other_access_token),
+            b'{"typing": true, "timeout": 30000}',
+        )
+        self.render(request)
+        self.assertEquals(200, channel.code)
+
+        # Should return immediately
+        request, channel = self.make_request("GET", "/sync?timeout=300000&access_token=%s&since=%s" % (access_token, next_batch))
+        self.render(request)
+        self.assertEquals(200, channel.code)
+        next_batch = channel.json_body["next_batch"]
+
+        # Reset typing serial back to 0, as if the master had.
+        typing = self.hs.get_typing_handler()
+        typing._latest_room_serial = 0
+
+        # Since it checks the state token, we need some state to update to
+        # invalidate the stream token.
+        self.helper.send(room, body="There!", tok=other_access_token)
+
+        request, channel = self.make_request("GET", "/sync?timeout=3000000&access_token=%s&since=%s" % (access_token, next_batch))
+        self.render(request)
+        self.assertEquals(200, channel.code)
+        next_batch = channel.json_body["next_batch"]
+
+        # This should time out! But it does not, because our stream token is ahead.
+        request, channel = self.make_request("GET", "/sync?timeout=3000000&access_token=%s&since=%s" % (access_token, next_batch))
+        self.render(request)
+        self.assertEquals(200, channel.code)
+        next_batch = channel.json_body["next_batch"]
+
+        # Clear the typing information, so that it doesn't think everything is
+        # in the future.
+        typing._room_serials = {}
+        typing._room_typing = {}
+
+        # Now it SHOULD fail as it never completes!
+        request, channel = self.make_request("GET", "/sync?timeout=3000000&access_token=%s&since=%s" % (access_token, next_batch))
+        self.assertRaises(Exception, self.render, request)