From b45bbd9cdc87da1b16e819f485080da885b26563 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 7 Jul 2026 17:13:54 +0000 Subject: [PATCH 1/2] Bound sliding sync per-connection state under position replay Clients may resend the same sliding sync position many times, e.g. if they repeatedly fail to process our responses. Each such request records a new connection position, but the clean-up of old positions only ran the first time a position was used: it lives in `get_and_clear_connection_positions`, whose result is cached, so replayed positions never triggered it. A wedged client replaying one position could thus accumulate unboundedly many rows in `sliding_sync_connection_positions` (and, via the per-position state, in `sliding_sync_connection_streams`/`_room_configs`). Seen in the wild: ~1k positions on one connection over a few hours. Fix by deleting all other positions for the connection whenever we persist a new one: only the position the request was based on (which the client may retry from, not yet having seen our response) and the newly created position remain valid. Also update the connection's `last_used_ts` when persisting, since for replayed positions the cached read no longer does so; otherwise a connection in constant use from the same position would be expired as idle. Co-Authored-By: Claude Fable 5 --- .../storage/databases/main/sliding_sync.py | 51 ++++- .../sliding_sync/test_connection_tracking.py | 180 ++++++++++++++++++ 2 files changed, 229 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index dbb9efc2acf..fe9efd979c8 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/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` diff --git a/tests/rest/client/sliding_sync/test_connection_tracking.py b/tests/rest/client/sliding_sync/test_connection_tracking.py index 44e7fa47262..b5a5fada4c2 100644 --- a/tests/rest/client/sliding_sync/test_connection_tracking.py +++ b/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) From db200aae58af31284070a03b3be11ae323099ccc Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 7 Jul 2026 18:22:33 +0100 Subject: [PATCH 2/2] Newsfile --- changelog.d/19924.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19924.bugfix diff --git a/changelog.d/19924.bugfix b/changelog.d/19924.bugfix new file mode 100644 index 00000000000..ae0c94d0f8a --- /dev/null +++ b/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.