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. 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)