dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 07:14:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add forgotten column

Browse Source
This commit is contained in:
Eric Eastwood
2024-08-21 21:54:22 -05:00
parent 0a938b137a
commit ee2ef0b4d9
4 changed files with 459 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/storage/databases/main/events_bg_updates.py
+15 -5
View File
@@ -1984,19 +1984,27 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
# lists
insert_keys = insert_map.keys()
insert_values = insert_map.values()
# We don't need to do anything `ON CONFLICT` because we never partially
# insert/update the snapshots
# We don't need to update the state `ON CONFLICT` because we never
# partially insert/update the snapshots and anything already there is
# up-to-date EXCEPT for the `forgotten` field since that is updated out
# of band from the membership changes.
#
# We need to find the `forgotten` value during the transaction because
# we can't risk inserting stale data.
txn.execute(
f"""
INSERT INTO sliding_sync_membership_snapshots
(room_id, user_id, sender, membership_event_id, membership, event_stream_ordering
(room_id, user_id, sender, membership_event_id, membership, forgotten, event_stream_ordering
{("," + ", ".join(insert_keys)) if insert_keys else ""})
VALUES (
?, ?, ?, ?, ?, ?
?, ?, ?, ?, ?,
(SELECT forgotten FROM room_memberships WHERE room_id = ? AND user_id = ?),
?
{("," + ", ".join("?" for _ in insert_values)) if insert_values else ""}
)
ON CONFLICT (room_id, user_id)
DO NOTHING
DO UPDATE SET
forgotten = EXCLUDED.forgotten
""",
[
room_id,
@@ -2004,6 +2012,8 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
sender,
membership_event_id,
membership,
room_id,
user_id,
membership_event_stream_ordering,
]
+ list(insert_values),
synapse/storage/databases/main/roommember.py
+6
View File
@@ -1337,6 +1337,12 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
keyvalues={"user_id": user_id, "room_id": room_id},
updatevalues={"forgotten": 1},
)
self.db_pool.simple_update_txn(
txn,
table="sliding_sync_membership_snapshots",
keyvalues={"user_id": user_id, "room_id": room_id},
updatevalues={"forgotten": 1},
)
self._invalidate_cache_and_stream(txn, self.did_forget, (user_id, room_id))
self._invalidate_cache_and_stream(
synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql
+3
View File
@@ -80,6 +80,9 @@ CREATE TABLE IF NOT EXISTS sliding_sync_membership_snapshots(
sender TEXT NOT NULL,
membership_event_id TEXT NOT NULL REFERENCES events(event_id),
membership TEXT NOT NULL,
-- This is an integer just to match `room_memberships` and also means we don't need
-- to do any casting.
forgotten INTEGER DEFAULT 0 NOT NULL,
-- `stream_ordering` of the `membership_event_id`
event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering),
-- For remote invites/knocks that don't include any stripped state, we want to be
tests/storage/test_events.py
+435 -11
View File
@@ -527,6 +527,9 @@ class _SlidingSyncMembershipSnapshotResult:
room_name: Optional[str]
is_encrypted: bool
tombstone_successor_room_id: Optional[str]
# Make this default to "not forgotten" because it doesn't apply to many tests and we
# don't want to force all of the tests to deal with it.
forgotten: bool = False
class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
@@ -598,7 +601,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
Mapping from the (room_id, user_id) to _SlidingSyncMembershipSnapshotResult.
"""
rows = cast(
List[Tuple[str, str, str, str, str, int, bool, str, str, bool, str]],
List[Tuple[str, str, str, str, str, int, int, bool, str, str, bool, str]],
self.get_success(
self.store.db_pool.simple_select_list(
"sliding_sync_membership_snapshots",
@@ -609,6 +612,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
"sender",
"membership_event_id",
"membership",
"forgotten",
"event_stream_ordering",
"has_known_state",
"room_type",
@@ -627,12 +631,13 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
sender=row[2],
membership_event_id=row[3],
membership=row[4],
event_stream_ordering=row[5],
has_known_state=bool(row[6]),
room_type=row[7],
room_name=row[8],
is_encrypted=bool(row[9]),
tombstone_successor_room_id=row[10],
forgotten=bool(row[5]),
event_stream_ordering=row[6],
has_known_state=bool(row[7]),
room_type=row[8],
room_name=row[9],
is_encrypted=bool(row[10]),
tombstone_successor_room_id=row[11],
)
for row in rows
}
@@ -3356,7 +3361,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
column="room_id",
iterable=(room_id_no_info, room_id_with_info, space_room_id),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_invite",
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_local_invite",
)
)
@@ -3574,7 +3579,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
space_room_id,
),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_invite",
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_remote_invite",
)
)
@@ -3808,7 +3813,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
space_room_id,
),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_invite",
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_remote_invite_rejections_and_retractions",
)
)
@@ -4069,7 +4074,7 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
column="room_id",
iterable=(room_id_no_info, room_id_with_info, space_room_id),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_invite",
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_historical_state",
)
)
@@ -4184,3 +4189,422 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase):
tombstone_successor_room_id=None,
),
)
def test_membership_snapshots_background_update_forgotten_missing(self) -> None:
"""
Test that a new row is inserted into `sliding_sync_membership_snapshots` when it
doesn't exist in the table yet.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
# User1 joins the room
self.helper.join(room_id, user1_id, tok=user1_tok)
# User1 leaves the room (we have to leave in order to forget the room)
self.helper.leave(room_id, user1_id, tok=user1_tok)
state_map = self.get_success(
self.storage_controllers.state.get_current_state(room_id)
)
# Forget the room
channel = self.make_request(
"POST",
f"/_matrix/client/r0/rooms/{room_id}/forget",
content={},
access_token=user1_tok,
)
self.assertEqual(channel.code, 200, channel.result)
# Clean-up the `sliding_sync_membership_snapshots` table as if the inserts did not
# happen during event creation.
self.get_success(
self.store.db_pool.simple_delete_many(
table="sliding_sync_membership_snapshots",
column="room_id",
iterable=(room_id,),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_forgotten_missing",
)
)
# We shouldn't find anything in the table because we just deleted them in
# preparation for the test.
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
set(),
exact=True,
)
# Insert and run the background update.
self.get_success(
self.store.db_pool.simple_insert(
"background_updates",
{
"update_name": _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BACKFILL,
"progress_json": "{}",
},
)
)
self.store.db_pool.updates._all_done = False
self.wait_for_background_updates()
# Make sure the table is populated
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
{
(room_id, user1_id),
(room_id, user2_id),
},
exact=True,
)
# Holds the info according to the current state when the user joined
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
_SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user1_id,
sender=user1_id,
membership_event_id=state_map[(EventTypes.Member, user1_id)].event_id,
membership=Membership.LEAVE,
event_stream_ordering=state_map[
(EventTypes.Member, user1_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
# Room is forgotten
forgotten=True,
),
)
# Holds the info according to the current state when the user joined
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user2_id)),
_SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user2_id,
sender=user2_id,
membership_event_id=state_map[(EventTypes.Member, user2_id)].event_id,
membership=Membership.JOIN,
event_stream_ordering=state_map[
(EventTypes.Member, user2_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
),
)
def test_membership_snapshots_background_update_forgotten_partial(self) -> None:
"""
Test an existing `sliding_sync_membership_snapshots` row is updated with the
latest `forgotten` status after the background update passes over it.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
# User1 joins the room
self.helper.join(room_id, user1_id, tok=user1_tok)
# User1 leaves the room (we have to leave in order to forget the room)
self.helper.leave(room_id, user1_id, tok=user1_tok)
state_map = self.get_success(
self.storage_controllers.state.get_current_state(room_id)
)
# Forget the room
channel = self.make_request(
"POST",
f"/_matrix/client/r0/rooms/{room_id}/forget",
content={},
access_token=user1_tok,
)
self.assertEqual(channel.code, 200, channel.result)
# Clean-up the `sliding_sync_joined_rooms` table as if the forgotten status
# never made it into the table.
self.get_success(
self.store.db_pool.simple_update(
table="sliding_sync_membership_snapshots",
keyvalues={"room_id": room_id},
updatevalues={"forgotten": 0},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_forgotten_partial",
)
)
# We should see the partial row that we made in preparation for the test.
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
{
(room_id, user1_id),
(room_id, user2_id),
},
exact=True,
)
user1_snapshot = _SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user1_id,
sender=user1_id,
membership_event_id=state_map[(EventTypes.Member, user1_id)].event_id,
membership=Membership.LEAVE,
event_stream_ordering=state_map[
(EventTypes.Member, user1_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
# Room is *not* forgotten because of our test preparation
forgotten=False,
)
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
user1_snapshot,
)
user2_snapshot = _SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user2_id,
sender=user2_id,
membership_event_id=state_map[(EventTypes.Member, user2_id)].event_id,
membership=Membership.JOIN,
event_stream_ordering=state_map[
(EventTypes.Member, user2_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
)
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user2_id)),
user2_snapshot,
)
# Insert and run the background update.
self.get_success(
self.store.db_pool.simple_insert(
"background_updates",
{
"update_name": _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BACKFILL,
"progress_json": "{}",
},
)
)
self.store.db_pool.updates._all_done = False
self.wait_for_background_updates()
# Make sure the table is populated
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
{
(room_id, user1_id),
(room_id, user2_id),
},
exact=True,
)
# Forgotten status is now updated
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
attr.evolve(user1_snapshot, forgotten=True),
)
# Holds the info according to the current state when the user joined
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user2_id)),
user2_snapshot,
)
def test_membership_snapshot_forget(self) -> None:
"""
Test forgetting a room will update `sliding_sync_membership_snapshots`
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
# User1 joins the room
self.helper.join(room_id, user1_id, tok=user1_tok)
# User1 leaves the room (we have to leave in order to forget the room)
self.helper.leave(room_id, user1_id, tok=user1_tok)
state_map = self.get_success(
self.storage_controllers.state.get_current_state(room_id)
)
# Check on the `sliding_sync_membership_snapshots` table (nothing should be
# forgotten yet)
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
{
(room_id, user1_id),
(room_id, user2_id),
},
exact=True,
)
# Holds the info according to the current state when the user joined
user1_snapshot = _SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user1_id,
sender=user1_id,
membership_event_id=state_map[(EventTypes.Member, user1_id)].event_id,
membership=Membership.LEAVE,
event_stream_ordering=state_map[
(EventTypes.Member, user1_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
# Room is not forgotten
forgotten=False,
)
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
user1_snapshot,
)
# Holds the info according to the current state when the user joined
user2_snapshot = _SlidingSyncMembershipSnapshotResult(
room_id=room_id,
user_id=user2_id,
sender=user2_id,
membership_event_id=state_map[(EventTypes.Member, user2_id)].event_id,
membership=Membership.JOIN,
event_stream_ordering=state_map[
(EventTypes.Member, user2_id)
].internal_metadata.stream_ordering,
has_known_state=True,
room_type=None,
room_name=None,
is_encrypted=False,
tombstone_successor_room_id=None,
)
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user2_id)),
user2_snapshot,
)
# Forget the room
channel = self.make_request(
"POST",
f"/_matrix/client/r0/rooms/{room_id}/forget",
content={},
access_token=user1_tok,
)
self.assertEqual(channel.code, 200, channel.result)
# Check on the `sliding_sync_membership_snapshots` table
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
{
(room_id, user1_id),
(room_id, user2_id),
},
exact=True,
)
# Room is now forgotten for user1
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
attr.evolve(user1_snapshot, forgotten=True),
)
# Nothing changed for user2
self.assertEqual(
sliding_sync_membership_snapshots_results.get((room_id, user2_id)),
user2_snapshot,
)
def test_membership_snapshot_missing_forget(
self,
) -> None:
"""
Test forgetting a room with no existing row in `sliding_sync_membership_snapshots`.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
# User1 joins the room
self.helper.join(room_id, user1_id, tok=user1_tok)
# User1 leaves the room (we have to leave in order to forget the room)
self.helper.leave(room_id, user1_id, tok=user1_tok)
# Clean-up the `sliding_sync_membership_snapshots` table as if the inserts did not
# happen during event creation.
self.get_success(
self.store.db_pool.simple_delete_many(
table="sliding_sync_membership_snapshots",
column="room_id",
iterable=(room_id,),
keyvalues={},
desc="sliding_sync_membership_snapshots.test_membership_snapshots_background_update_forgotten_missing",
)
)
# We shouldn't find anything in the table because we just deleted them in
# preparation for the test.
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
set(),
exact=True,
)
# Forget the room
channel = self.make_request(
"POST",
f"/_matrix/client/r0/rooms/{room_id}/forget",
content={},
access_token=user1_tok,
)
self.assertEqual(channel.code, 200, channel.result)
# It doesn't explode
# We still shouldn't find anything in the table because nothing has re-created them
sliding_sync_membership_snapshots_results = (
self._get_sliding_sync_membership_snapshots()
)
self.assertIncludes(
set(sliding_sync_membership_snapshots_results.keys()),
set(),
exact=True,
)