Bound sliding sync per-connection state under position replay

changelog.d/19924.bugfix

@@ -0,0 +1 @@
+Fix unbounded growth of the `sliding_sync_connection_positions` table when a client repeatedly retries requests from the same sliding sync position.

synapse/storage/databases/main/sliding_sync.py

@@ -216,7 +216,7 @@ def persist_per_connection_state_txn(
             # The `previous_connection_position` is a user-supplied value, so we
             # need to make sure that the one they supplied is actually theirs.
             sql = f"""
-                SELECT connection_key
+                SELECT connection_key, last_used_ts
                 FROM sliding_sync_connection_positions
                 INNER JOIN sliding_sync_connections USING (connection_key)
                 WHERE
@@ -231,7 +231,25 @@ def persist_per_connection_state_txn(
             if row is None:
                 raise SlidingSyncUnknownPosition()
 
-            (connection_key,) = row
+            (connection_key, last_used_ts) = row
+
+            # Update the `last_used_ts` if it's due to be updated. This is
+            # also done when a position is first used (in
+            # `_get_and_clear_connection_positions_txn`), but that read is
+            # cached, so for a client that repeatedly retries from the same
+            # position this is the only place the connection gets marked as
+            # in use (and thus protected from expiry).
+            now = self.clock.time_msec()
+            if (
+                last_used_ts is None
+                or now - last_used_ts > UPDATE_INTERVAL_LAST_USED_TS.as_millis()
+            ):
+                self.db_pool.simple_update_txn(
+                    txn,
+                    table="sliding_sync_connections",
+                    keyvalues={"connection_key": connection_key},
+                    updatevalues={"last_used_ts": now},
+                )
         else:
             # We're restarting the connection, so we clear the previous existing data we
             # used to track it. We do this here to ensure that if we get lots of
@@ -272,6 +290,35 @@ def persist_per_connection_state_txn(
             returning=("connection_position",),
         )
 
+        if previous_connection_position is not None:
+            # Delete all other positions for this connection. The only
+            # positions the client can validly send from now on are the
+            # previous position (it hasn't seen the response containing the
+            # new position yet, so may retry from it) and the new position.
+            #
+            # We can't rely on `_get_and_clear_connection_positions_txn` to do
+            # this: it only runs when a position is *first* used (the result is
+            # cached), so a client that repeatedly retries from the same
+            # position (e.g. because it never successfully processes our
+            # responses) would accumulate unboundedly many positions, plus
+            # their associated per-room state in the dependent tables.
+            #
+            # Note this means that if two requests concurrently fork from the
+            # same position, the later one to persist wins and the other fork
+            # dies before it is ever used. A client that somehow processed the
+            # losing response will get `SlidingSyncUnknownPosition` and start a
+            # new connection, which was already the outcome once one of the
+            # forks got used.
+            sql = """
+                DELETE FROM sliding_sync_connection_positions
+                WHERE connection_key = ?
+                    AND connection_position != ? AND connection_position != ?
+            """
+            txn.execute(
+                sql,
+                (connection_key, previous_connection_position, connection_position),
+            )
+
         # We need to deduplicate the `required_state` JSON. We do this by
         # fetching all JSON associated with the connection and comparing that
         # with the updates to `required_state`

tests/rest/client/sliding_sync/test_connection_tracking.py

@@ -503,3 +503,183 @@ def test_sliding_sync_connection_expires_after_time(self) -> None:
         channel = self.make_sync_request(sync_body, since=from_token, tok=user1_tok)
         self.assertEqual(channel.code, 400)
         self.assertEqual(channel.json_body["errcode"], Codes.UNKNOWN_POS)
+
+    def _get_connection_positions_in_db(self) -> set[int]:
+        """Fetch all rows in `sliding_sync_connection_positions`."""
+        rows = self.get_success(
+            self.store.db_pool.simple_select_list(
+                table="sliding_sync_connection_positions",
+                keyvalues={},
+                retcols=("connection_position",),
+                desc="_get_connection_positions_in_db",
+            )
+        )
+        return {connection_position for (connection_position,) in rows}
+
+    @staticmethod
+    def _connection_position_from_token(token: str) -> int:
+        """Extract the connection position from a sliding sync token."""
+        return int(token.split("/", 1)[0])
+
+    def test_replaying_connection_position_does_not_accumulate_state(self) -> None:
+        """
+        Test that repeatedly syncing from the same connection position does not
+        accumulate per-connection state in the database.
+
+        Clients are allowed to resend a position many times (e.g. if they never
+        successfully process our responses), and each such request records a
+        new connection position. We check that the old, unused positions get
+        deleted.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id, user1_id, tok=user1_tok)
+
+        sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [],
+                    "timeline_limit": 1,
+                }
+            }
+        }
+
+        # Initial sync to create the connection.
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Repeatedly sync from the same position. We use a higher
+        # `timeline_limit` than the state persisted at the replayed position,
+        # which counts as a per-connection state update, ensuring that every
+        # request has something to persist (and so records a new connection
+        # position). Otherwise, requests without updates simply return the
+        # position they were given.
+        replay_sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [],
+                    "timeline_limit": 2,
+                }
+            }
+        }
+        first_replay_token: str | None = None
+        latest_token = from_token
+        for _ in range(5):
+            self.helper.send(room_id, "msg", tok=user2_tok)
+            _, latest_token = self.do_sync(
+                replay_sync_body, since=from_token, tok=user1_tok
+            )
+            if first_replay_token is None:
+                first_replay_token = latest_token
+        assert first_replay_token is not None
+
+        # Sanity check that the replayed requests did indeed create new
+        # connection positions.
+        self.assertNotEqual(
+            self._connection_position_from_token(latest_token),
+            self._connection_position_from_token(from_token),
+        )
+        self.assertNotEqual(
+            self._connection_position_from_token(latest_token),
+            self._connection_position_from_token(first_replay_token),
+        )
+
+        # Only the position the client keeps sending and the most recently
+        # issued position should remain (rather than one position per request).
+        self.assertEqual(
+            self._get_connection_positions_in_db(),
+            {
+                self._connection_position_from_token(from_token),
+                self._connection_position_from_token(latest_token),
+            },
+        )
+
+        # The per-position state should not have accumulated either.
+        stream_rows = self.get_success(
+            self.store.db_pool.simple_select_list(
+                table="sliding_sync_connection_streams",
+                keyvalues={},
+                retcols=("connection_position",),
+                desc="get_connection_streams",
+            )
+        )
+        self.assertLessEqual(
+            {connection_position for (connection_position,) in stream_rows},
+            self._get_connection_positions_in_db(),
+        )
+
+        # Positions from replayed requests that the client never used are gone.
+        channel = self.make_sync_request(
+            replay_sync_body, since=first_replay_token, tok=user1_tok
+        )
+        self.assertEqual(channel.code, 400)
+        self.assertEqual(channel.json_body["errcode"], Codes.UNKNOWN_POS)
+
+        # The client can continue from the most recently issued position, and
+        # doing so cleans up the position it had been replaying.
+        _, next_token = self.do_sync(
+            replay_sync_body, since=latest_token, tok=user1_tok
+        )
+        self.assertEqual(
+            self._get_connection_positions_in_db(),
+            {
+                self._connection_position_from_token(latest_token),
+                self._connection_position_from_token(next_token),
+            },
+        )
+
+    def test_replaying_connection_position_does_not_expire_connection(self) -> None:
+        """
+        Test that a connection that is in constant use, but only ever from the
+        same connection position, does not get expired as unused.
+
+        The `last_used_ts` of a connection is normally updated when a position
+        is first used, but that read is cached, so replayed requests must
+        update it when persisting their state instead.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id, user1_id, tok=user1_tok)
+
+        sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [],
+                    "timeline_limit": 1,
+                }
+            }
+        }
+
+        # Initial sync to create the connection.
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Keep replaying the same position, with a higher `timeline_limit` so
+        # that each request has per-connection state updates to persist. The
+        # gap between any two requests is well below CONNECTION_EXPIRY, but the
+        # total time elapsed is well above it, so the connection only survives
+        # if the replayed requests keep `last_used_ts` up to date.
+        replay_sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [],
+                    "timeline_limit": 2,
+                }
+            }
+        }
+        for _ in range(15):
+            self.reactor.advance(CONNECTION_EXPIRY.as_secs() / 10)
+            self.helper.send(room_id, "msg", tok=user2_tok)
+            self.do_sync(replay_sync_body, since=from_token, tok=user1_tok)