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

On catchup, process each row with its own stream id #7286

Merged
merged 5 commits into from
Apr 20, 2020
Merged

On catchup, process each row with its own stream id #7286

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8f0870c
On catchup, process each row with its own stream id
richvdh Apr 16, 2020
c7c0fe3
Batch up the updates returned by get_updates_since
richvdh Apr 16, 2020
f1709e2
Fix _batch_updates for py37
richvdh Apr 17, 2020
dc35b68
Relax StreamChangeCache sanity-check
richvdh Apr 17, 2020
378d34e
Remove the sanity check
richvdh Apr 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/7286.misc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Move catchup of replication streams logic to worker.
6 changes: 2 additions & 4 deletions synapse/replication/tcp/handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,11 +268,9 @@ async def on_POSITION(self, cmd: PositionCommand):
missing_updates,
) = await stream.get_updates_since(current_token, cmd.token)

if updates:
for token, row in updates:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to worry about batched updates? I.e. updates with multiple entries per token?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was hoping you would know the answer to that :/

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like you do: https://github.com/matrix-org/synapse/blob/master/synapse/replication/tcp/resource.py#L198-L220

await self.on_rdata(
cmd.stream_name,
current_token,
[stream.parse_row(update[1]) for update in updates],
cmd.stream_name, token, [stream.parse_row(row)],
)

# We've now caught up to position sent to us, notify handler.
Expand Down
5 changes: 5 additions & 0 deletions synapse/util/caches/stream_change_cache.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,11 @@ def entity_has_changed(self, entity, stream_pos):
"""
assert type(stream_pos) is int

if stream_pos in self._cache:
raise NotImplementedError(
"more than one entity changing at a stream position"
)

if stream_pos > self._earliest_known_stream_pos:
old_pos = self._entity_to_key.get(entity, None)
if old_pos is not None:
Expand Down