Skip to content

Commit

Permalink
Use fully-qualified PersistedEventPosition when returning `RoomsFor…
Browse files Browse the repository at this point in the history
…User` (#17265)

Use fully-qualified `PersistedEventPosition` (`instance_name` and `stream_ordering`) when returning `RoomsForUser` to facilitate proper comparisons and `RoomStreamToken` generation.

Spawning from #17187 where we want to utilize this change
  • Loading branch information
MadLittleMods authored Jun 4, 2024
1 parent eab0b54 commit 7d8f0ef
Show file tree
Hide file tree
Showing 11 changed files with 85 additions and 75 deletions.
1 change: 1 addition & 0 deletions changelog.d/17265.misc
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Use fully-qualified `PersistedEventPosition` when returning `RoomsForUser` to facilitate proper comparisons and `RoomStreamToken` generation.
4 changes: 2 additions & 2 deletions synapse/federation/federation_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -674,7 +674,7 @@ async def on_make_join_request(
# This is in addition to the HS-level rate limiting applied by
# BaseFederationServlet.
# type-ignore: mypy doesn't seem able to deduce the type of the limiter(!?)
await self._room_member_handler._join_rate_per_room_limiter.ratelimit( # type: ignore[has-type]
await self._room_member_handler._join_rate_per_room_limiter.ratelimit(
requester=None,
key=room_id,
update=False,
Expand Down Expand Up @@ -717,7 +717,7 @@ async def on_send_join_request(
SynapseTags.SEND_JOIN_RESPONSE_IS_PARTIAL_STATE,
caller_supports_partial_state,
)
await self._room_member_handler._join_rate_per_room_limiter.ratelimit( # type: ignore[has-type]
await self._room_member_handler._join_rate_per_room_limiter.ratelimit(
requester=None,
key=room_id,
update=False,
Expand Down
10 changes: 2 additions & 8 deletions synapse/handlers/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,13 +126,7 @@ async def export_user_data(self, user_id: str, writer: "ExfiltrationWriter") ->
# Get all rooms the user is in or has been in
rooms = await self._store.get_rooms_for_local_user_where_membership_is(
user_id,
membership_list=(
Membership.JOIN,
Membership.LEAVE,
Membership.BAN,
Membership.INVITE,
Membership.KNOCK,
),
membership_list=Membership.LIST,
)

# We only try and fetch events for rooms the user has been in. If
Expand Down Expand Up @@ -179,7 +173,7 @@ async def export_user_data(self, user_id: str, writer: "ExfiltrationWriter") ->
if room.membership == Membership.JOIN:
stream_ordering = self._store.get_room_max_stream_ordering()
else:
stream_ordering = room.stream_ordering
stream_ordering = room.event_pos.stream

from_key = RoomStreamToken(topological=0, stream=0)
to_key = RoomStreamToken(stream=stream_ordering)
Expand Down
2 changes: 1 addition & 1 deletion synapse/handlers/initial_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ async def handle_room(event: RoomsForUser) -> None:
)
elif event.membership == Membership.LEAVE:
room_end_token = RoomStreamToken(
stream=event.stream_ordering,
stream=event.event_pos.stream,
)
deferred_room_state = run_in_background(
self._state_storage_controller.get_state_for_events,
Expand Down
3 changes: 2 additions & 1 deletion synapse/handlers/pagination.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
from synapse.api.errors import SynapseError
from synapse.api.filtering import Filter
from synapse.events.utils import SerializeEventConfig
from synapse.handlers.room import ShutdownRoomParams, ShutdownRoomResponse
from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
from synapse.logging.opentracing import trace
from synapse.metrics.background_process_metrics import run_as_background_process
Expand All @@ -38,6 +37,8 @@
JsonMapping,
Requester,
ScheduledTask,
ShutdownRoomParams,
ShutdownRoomResponse,
StreamKeyType,
TaskStatus,
)
Expand Down
60 changes: 2 additions & 58 deletions synapse/handlers/room.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@
)

import attr
from typing_extensions import TypedDict

import synapse.events.snapshot
from synapse.api.constants import (
Expand Down Expand Up @@ -81,6 +80,8 @@
RoomAlias,
RoomID,
RoomStreamToken,
ShutdownRoomParams,
ShutdownRoomResponse,
StateMap,
StrCollection,
StreamKeyType,
Expand Down Expand Up @@ -1780,63 +1781,6 @@ def get_current_key_for_room(self, room_id: str) -> Awaitable[RoomStreamToken]:
return self.store.get_current_room_stream_token_for_room_id(room_id)


class ShutdownRoomParams(TypedDict):
"""
Attributes:
requester_user_id:
User who requested the action. Will be recorded as putting the room on the
blocking list.
new_room_user_id:
If set, a new room will be created with this user ID
as the creator and admin, and all users in the old room will be
moved into that room. If not set, no new room will be created
and the users will just be removed from the old room.
new_room_name:
A string representing the name of the room that new users will
be invited to. Defaults to `Content Violation Notification`
message:
A string containing the first message that will be sent as
`new_room_user_id` in the new room. Ideally this will clearly
convey why the original room was shut down.
Defaults to `Sharing illegal content on this server is not
permitted and rooms in violation will be blocked.`
block:
If set to `true`, this room will be added to a blocking list,
preventing future attempts to join the room. Defaults to `false`.
purge:
If set to `true`, purge the given room from the database.
force_purge:
If set to `true`, the room will be purged from database
even if there are still users joined to the room.
"""

requester_user_id: Optional[str]
new_room_user_id: Optional[str]
new_room_name: Optional[str]
message: Optional[str]
block: bool
purge: bool
force_purge: bool


class ShutdownRoomResponse(TypedDict):
"""
Attributes:
kicked_users: An array of users (`user_id`) that were kicked.
failed_to_kick_users:
An array of users (`user_id`) that that were not kicked.
local_aliases:
An array of strings representing the local aliases that were
migrated from the old room to the new.
new_room_id: A string representing the room ID of the new room.
"""

kicked_users: List[str]
failed_to_kick_users: List[str]
local_aliases: List[str]
new_room_id: Optional[str]


class RoomShutdownHandler:
DEFAULT_MESSAGE = (
"Sharing illegal content on this server is not permitted and rooms in"
Expand Down
2 changes: 1 addition & 1 deletion synapse/handlers/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -2805,7 +2805,7 @@ async def _get_room_changes_for_initial_sync(
continue

leave_token = now_token.copy_and_replace(
StreamKeyType.ROOM, RoomStreamToken(stream=event.stream_ordering)
StreamKeyType.ROOM, RoomStreamToken(stream=event.event_pos.stream)
)
room_entries.append(
RoomSyncResultBuilder(
Expand Down
14 changes: 12 additions & 2 deletions synapse/storage/databases/main/roommember.py
Original file line number Diff line number Diff line change
Expand Up @@ -476,7 +476,7 @@ def _get_rooms_for_local_user_where_membership_is_txn(
)

sql = """
SELECT room_id, e.sender, c.membership, event_id, e.stream_ordering, r.room_version
SELECT room_id, e.sender, c.membership, event_id, e.instance_name, e.stream_ordering, r.room_version
FROM local_current_membership AS c
INNER JOIN events AS e USING (room_id, event_id)
INNER JOIN rooms AS r USING (room_id)
Expand All @@ -488,7 +488,17 @@ def _get_rooms_for_local_user_where_membership_is_txn(
)

txn.execute(sql, (user_id, *args))
results = [RoomsForUser(*r) for r in txn]
results = [
RoomsForUser(
room_id=room_id,
sender=sender,
membership=membership,
event_id=event_id,
event_pos=PersistedEventPosition(instance_name, stream_ordering),
room_version_id=room_version,
)
for room_id, sender, membership, event_id, instance_name, stream_ordering, room_version in txn
]

return results

Expand Down
2 changes: 1 addition & 1 deletion synapse/storage/roommember.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ class RoomsForUser:
sender: str
membership: str
event_id: str
stream_ordering: int
event_pos: PersistedEventPosition
room_version_id: str


Expand Down
57 changes: 57 additions & 0 deletions synapse/types/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1279,3 +1279,60 @@ class ScheduledTask:
result: Optional[JsonMapping]
# Optional error that should be assigned a value when the status is FAILED
error: Optional[str]


class ShutdownRoomParams(TypedDict):
"""
Attributes:
requester_user_id:
User who requested the action. Will be recorded as putting the room on the
blocking list.
new_room_user_id:
If set, a new room will be created with this user ID
as the creator and admin, and all users in the old room will be
moved into that room. If not set, no new room will be created
and the users will just be removed from the old room.
new_room_name:
A string representing the name of the room that new users will
be invited to. Defaults to `Content Violation Notification`
message:
A string containing the first message that will be sent as
`new_room_user_id` in the new room. Ideally this will clearly
convey why the original room was shut down.
Defaults to `Sharing illegal content on this server is not
permitted and rooms in violation will be blocked.`
block:
If set to `true`, this room will be added to a blocking list,
preventing future attempts to join the room. Defaults to `false`.
purge:
If set to `true`, purge the given room from the database.
force_purge:
If set to `true`, the room will be purged from database
even if there are still users joined to the room.
"""

requester_user_id: Optional[str]
new_room_user_id: Optional[str]
new_room_name: Optional[str]
message: Optional[str]
block: bool
purge: bool
force_purge: bool


class ShutdownRoomResponse(TypedDict):
"""
Attributes:
kicked_users: An array of users (`user_id`) that were kicked.
failed_to_kick_users:
An array of users (`user_id`) that that were not kicked.
local_aliases:
An array of strings representing the local aliases that were
migrated from the old room to the new.
new_room_id: A string representing the room ID of the new room.
"""

kicked_users: List[str]
failed_to_kick_users: List[str]
local_aliases: List[str]
new_room_id: Optional[str]
5 changes: 4 additions & 1 deletion tests/replication/storage/test_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,10 @@ def test_invites(self) -> None:
USER_ID,
"invite",
event.event_id,
event.internal_metadata.stream_ordering,
PersistedEventPosition(
self.hs.get_instance_name(),
event.internal_metadata.stream_ordering,
),
RoomVersions.V1.identifier,
)
],
Expand Down

0 comments on commit 7d8f0ef

Please sign in to comment.