dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-14 18:01:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bound sliding sync per-connection state under position replay

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Erik Johnston
2026-07-07 17:14:09 +00:00
co-authored by Claude Fable 5
parent ec6c53e7df
commit b45bbd9cdc
2 changed files with 229 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/storage/databases/main/sliding_sync.py
+49 -2
View File
@@ -216,7 +216,7 @@ class SlidingSyncStore(SQLBaseStore):
# 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 @@ class SlidingSyncStore(SQLBaseStore):
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 @@ class SlidingSyncStore(SQLBaseStore):
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
+180
View File
@@ -503,3 +503,183 @@ class SlidingSyncConnectionTrackingTestCase(SlidingSyncBase):
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)