Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Fix race calling /members?at=
Browse files Browse the repository at this point in the history
Fixes #14814
  • Loading branch information
erikjohnston committed Jan 11, 2023
1 parent 7b3a8f2 commit 7d6a18c
Showing 1 changed file with 55 additions and 6 deletions.
61 changes: 55 additions & 6 deletions synapse/storage/databases/main/stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -801,13 +801,62 @@ async def get_last_event_in_room_before_stream_ordering(
before this stream ordering.
"""

last_row = await self.get_room_event_before_stream_ordering(
room_id=room_id,
stream_ordering=end_token.stream,
def get_last_event_in_room_before_stream_ordering_txn(
txn: LoggingTransaction,
) -> Optional[str]:
# We need to handle the fact that the stream tokens can be vector
# clocks. We do this by getting all rows between the minimum and
# maximum stream ordering in the token, plus one row less than the
# minimum stream ordering. We then filter the results against the
# token and return the first row that matches.

sql = """
SELECT instance_name, stream_ordering, topological_ordering, event_id
FROM events
LEFT JOIN rejections USING (event_id)
WHERE room_id = ?
AND ? < stream_ordering AND stream_ordering <= ?
AND NOT outlier
AND rejections.event_id IS NULL
ORDER BY stream_ordering DESC
UNION
SELECT instance_name, stream_ordering, topological_ordering, event_id
FROM events
LEFT JOIN rejections USING (event_id)
WHERE room_id = ?
AND stream_ordering <= ?
AND NOT outlier
AND rejections.event_id IS NULL
ORDER BY stream_ordering DESC
LIMIT 1
"""
txn.execute(
sql,
(
room_id,
end_token.stream,
end_token.get_max_stream_pos(),
room_id,
end_token.stream,
),
)

for instance_name, stream_ordering, topological_ordering, event_id in txn:
if _filter_results(
lower_token=None,
upper_token=end_token,
instance_name=instance_name,
topological_ordering=topological_ordering,
stream_ordering=stream_ordering,
):
return event_id

return None

return await self.db_pool.runInteraction(
"get_last_event_in_room_before_stream_ordering",
get_last_event_in_room_before_stream_ordering_txn,
)
if last_row:
return last_row[2]
return None

async def get_current_room_stream_token_for_room_id(
self, room_id: str
Expand Down

0 comments on commit 7d6a18c

Please sign in to comment.