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

Allow multiple workers to write to receipts stream. #16432

Merged
merged 15 commits into from
Oct 25, 2023
Merged

Allow multiple workers to write to receipts stream. #16432

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
c7d6f75
Add concrete MultiWriterStreamToken class
erikjohnston Oct 4, 2023
7378d52
Use new class for receipts in `StreamToken`
erikjohnston Sep 29, 2023
f099895
Change function types
erikjohnston Oct 3, 2023
b012c88
Fix DB layer
erikjohnston Oct 4, 2023
342d634
Fix appservice
erikjohnston Oct 5, 2023
e7acc24
Allow multiple receipt writers
erikjohnston Oct 5, 2023
bc37349
Newsfile
erikjohnston Oct 5, 2023
0ae622f
Fix missing column on sqlite
erikjohnston Oct 5, 2023
f3462e6
Fix unit tests
erikjohnston Oct 5, 2023
0484455
Fix replication
erikjohnston Oct 6, 2023
98ae86a
Fix StreamToken.get_fields typing
erikjohnston Oct 6, 2023
51c670d
Merge remote-tracking branch 'origin/develop' into erikj/multi_receipts2
erikjohnston Oct 6, 2023
91218f9
Add tests
erikjohnston Oct 12, 2023
68f57cd
Merge remote-tracking branch 'origin/develop' into erikj/multi_receipts2
erikjohnston Oct 23, 2023
a5d4d96
Updated schema delta version
erikjohnston Oct 23, 2023
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
40 changes: 40 additions & 0 deletions synapse/notifier.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,13 @@
Dict,
Iterable,
List,
Literal,
Optional,
Set,
Tuple,
TypeVar,
Union,
overload,
)

import attr
Expand Down Expand Up @@ -453,6 +455,44 @@ def _notify_pusher_pool(self, max_room_stream_token: RoomStreamToken) -> None:
except Exception:
logger.exception("Error pusher pool of event")

@overload
def on_new_event(
self,
stream_key: Literal[StreamKeyType.ROOM],
new_token: RoomStreamToken,
users: Optional[Collection[Union[str, UserID]]] = None,
rooms: Optional[StrCollection] = None,
) -> None:
...
Comment on lines +458 to +466
Copy link
Member

Choose a reason for hiding this comment

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

I'm surprised we didn't already need this overload! Does it help avoid any asserts or anything?

Copy link
Member Author

Choose a reason for hiding this comment

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

It doesn't reduce assertions, as it only limits what is accepted as arguments (it doesn't change the return type). I added it to catch the case where we passed in an int when using StreamKeyType.RECEIPT.


@overload
def on_new_event(
self,
stream_key: Literal[StreamKeyType.RECEIPT],
new_token: MultiWriterStreamToken,
users: Optional[Collection[Union[str, UserID]]] = None,
rooms: Optional[StrCollection] = None,
) -> None:
...

@overload
def on_new_event(
self,
stream_key: Literal[
StreamKeyType.ACCOUNT_DATA,
StreamKeyType.DEVICE_LIST,
StreamKeyType.PRESENCE,
StreamKeyType.PUSH_RULES,
StreamKeyType.TO_DEVICE,
StreamKeyType.TYPING,
StreamKeyType.UN_PARTIAL_STATED_ROOMS,
],
new_token: int,
users: Optional[Collection[Union[str, UserID]]] = None,
rooms: Optional[StrCollection] = None,
) -> None:
...

def on_new_event(
self,
stream_key: StreamKeyType,
Expand Down
3 changes: 2 additions & 1 deletion synapse/replication/tcp/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,8 +126,9 @@ async def on_rdata(
StreamKeyType.ACCOUNT_DATA, token, users=[row.user_id for row in rows]
)
elif stream_name == ReceiptsStream.NAME:
new_token = self.store.get_max_receipt_stream_id()
self.notifier.on_new_event(
StreamKeyType.RECEIPT, token, rooms=[row.room_id for row in rows]
StreamKeyType.RECEIPT, new_token, rooms=[row.room_id for row in rows]
clokep marked this conversation as resolved.
Show resolved Hide resolved
)
await self._pusher_pool.on_new_receipts({row.user_id for row in rows})
elif stream_name == ToDeviceStream.NAME:
Expand Down