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= (#14817)
Browse files Browse the repository at this point in the history
Fixes #14814
  • Loading branch information
erikjohnston committed Jan 12, 2023
1 parent 3952297 commit 84ce93c
Show file tree
Hide file tree
Showing 2 changed files with 60 additions and 6 deletions.
1 change: 1 addition & 0 deletions changelog.d/14817.bugfix
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix race where calling `/members` or `/state` with an `at` parameter could fail for newly created rooms, when using multiple workers.
65 changes: 59 additions & 6 deletions synapse/storage/databases/main/stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -801,13 +801,66 @@ 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 * FROM (
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
) AS a
UNION
SELECT * FROM (
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
) AS b
"""
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 84ce93c

Please sign in to comment.