mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-07 02:07:38 +00:00
Paul Chobert
8c11b13f63
This PR fixes the issue described as comment here: https://github.com/element-hq/synapse/issues/18793#issuecomment-3502202379 In Element Call, this shows up as ghost participants: someone who left the call keeps being displayed until a later state change refreshes the room. The bug is not specific to Element Call: any state event can be affected, RTC membership just changes often enough to make it visible. ## What happens Alice has a client syncing against a homeserver where events are persisted by one worker (the event persister) and `/sync` is served by another (the sync worker). Her client is parked in a long-poll: `GET /sync?since=s99&timeout=30000`. Bob joins a call at the same moment Carol sends a message. Carol's message reaches the persister first; Bob's `m.call.member` arrives while that write is still in flight, so the per-room persist queue groups them into one transaction: ``` events (each gets its own stream ordering): stream_ordering 100: m.room.message Carol stream_ordering 101: m.call.member Bob (state) current_state_delta_stream (how state_after finds state changes): stream_id 100 ────► (m.call.member, @bob) -> $bob_join_call ▲ └─ stamped with the batch MINIMUM (100), not the event's own 101 (see `_update_current_state_txn`) ``` The transaction commits: both events and the delta row are now in the database, atomically. The persister then announces the new events over replication, one RDATA token per stream ordering — rows are only merged into one token when they share a position, and 100 and 101 don't. So the sync worker's events-stream position steps 99 → 100 → 101, and on reaching 100 it pokes the notifier. Alice's long-poll wakes at exactly that moment. Her response is built at the worker's *current* position — `end = 100` — with RDATA 101 still in the queue: ``` Sync A (since=99, end=100): timeline: events 99 < ordering ≤ 100 → [Carol's message] state_after: deltas 99 < stream_id ≤ 100 → [$bob_join_call] ← delivered EARLY next_batch: s100 ← mid-batch token ``` No race on the client's side is needed: the server *hands out* the mid-batch token as `next_batch`. Alice's client re-polls with it, as every sync client does. The worker has meanwhile processed RDATA 101: ``` Sync B (since=100, end=101): timeline: events 100 < ordering ≤ 101 → [Bob's m.call.member @101] ✓ state_after: deltas 100 < stream_id ≤ 101 → [] row is stamped 100 ✗ ``` A state event in the timeline with an empty `state_after`. An MSC4222 client trusts `state_after` over timeline state events, so Alice's copy of Bob's call membership never updates from this response. On a single process this cannot happen: the batch's stream IDs are released as a whole, so the position visible to `/sync` jumps 99 → 101 and `s100` is never handed out. Only a process that learns its position from replication — any sync worker — ticks through the middle of a batch. ### Pull Request Checklist <!-- Please read https://element-hq.github.io/synapse/latest/development/contributing_guide.html before submitting your pull request --> * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))
635 lines
24 KiB
Python
635 lines
24 KiB
Python
#
|
|
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
|
#
|
|
# Copyright (C) 2023 New Vector, Ltd
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# See the GNU Affero General Public License for more details:
|
|
# <https://www.gnu.org/licenses/agpl-3.0.html>.
|
|
#
|
|
# Originally licensed under the Apache License, Version 2.0:
|
|
# <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# [This file includes modifications made by New Vector Limited]
|
|
#
|
|
#
|
|
|
|
from typing import Any
|
|
|
|
from parameterized import parameterized
|
|
|
|
from twisted.internet.testing import MemoryReactor
|
|
|
|
from synapse.api.constants import EventTypes, Membership
|
|
from synapse.events import EventBase
|
|
from synapse.replication.tcp.commands import RdataCommand
|
|
from synapse.replication.tcp.resource import _batch_updates
|
|
from synapse.replication.tcp.streams._base import _STREAM_UPDATE_TARGET_ROW_COUNT
|
|
from synapse.replication.tcp.streams.events import (
|
|
_MAX_STATE_UPDATES_PER_ROOM,
|
|
EventsStream,
|
|
EventsStreamAllStateRow,
|
|
EventsStreamCurrentStateRow,
|
|
EventsStreamEventRow,
|
|
EventsStreamRow,
|
|
)
|
|
from synapse.rest import admin
|
|
from synapse.rest.client import login, room
|
|
from synapse.server import HomeServer
|
|
from synapse.util.clock import Clock
|
|
|
|
from tests import unittest
|
|
from tests.replication._base import BaseStreamTestCase
|
|
from tests.test_utils.event_injection import (
|
|
inject_event,
|
|
inject_member_event,
|
|
persist_message_and_state_event_in_one_batch,
|
|
)
|
|
|
|
|
|
class EventsStreamTestCase(BaseStreamTestCase):
|
|
servlets = [
|
|
admin.register_servlets,
|
|
login.register_servlets,
|
|
room.register_servlets,
|
|
]
|
|
|
|
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
|
|
super().prepare(reactor, clock, hs)
|
|
self.user_id = self.register_user("u1", "pass")
|
|
self.user_tok = self.login("u1", "pass")
|
|
|
|
self.reconnect()
|
|
|
|
self.room_id = self.helper.create_room_as(tok=self.user_tok)
|
|
self.test_handler.received_rdata_rows.clear()
|
|
|
|
def test_update_function_event_row_limit(self) -> None:
|
|
"""Test replication with many non-state events
|
|
|
|
Checks that all events are correctly replicated when there are lots of
|
|
event rows to be replicated.
|
|
"""
|
|
# disconnect, so that we can stack up some changes
|
|
self.disconnect()
|
|
|
|
# generate lots of non-state events. We inject them using inject_event
|
|
# so that they are not send out over replication until we call self.replicate().
|
|
events = [
|
|
self._inject_test_event()
|
|
for _ in range(_STREAM_UPDATE_TARGET_ROW_COUNT + 1)
|
|
]
|
|
|
|
# also one state event
|
|
state_event = self._inject_state_event()
|
|
|
|
# check we're testing what we think we are: no rows should yet have been
|
|
# received
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
# now reconnect to pull the updates
|
|
self.reconnect()
|
|
self.replicate()
|
|
|
|
# we should have received all the expected rows in the right order (as
|
|
# well as various cache invalidation updates which we ignore)
|
|
#
|
|
# Filter the updates to only include event changes
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
for event in events:
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual(EventsStream.NAME, stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, event.event_id)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, state_event.event_id)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state")
|
|
self.assertIsInstance(row.data, EventsStreamCurrentStateRow)
|
|
self.assertEqual(row.data.event_id, state_event.event_id)
|
|
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
@parameterized.expand(
|
|
[(_STREAM_UPDATE_TARGET_ROW_COUNT, False), (_MAX_STATE_UPDATES_PER_ROOM, True)]
|
|
)
|
|
def test_update_function_huge_state_change(
|
|
self, num_state_changes: int, collapse_state_changes: bool
|
|
) -> None:
|
|
"""Test replication with many state events
|
|
|
|
Ensures that all events are correctly replicated when there are lots of
|
|
state change rows to be replicated.
|
|
|
|
Args:
|
|
num_state_changes: The number of state changes to create.
|
|
collapse_state_changes: Whether the state changes are expected to be
|
|
collapsed or not.
|
|
"""
|
|
|
|
# we want to generate lots of state changes at a single stream ID.
|
|
#
|
|
# We do this by having two branches in the DAG. On one, we have a moderator
|
|
# which that generates lots of state; on the other, we de-op the moderator,
|
|
# thus invalidating all the state.
|
|
|
|
OTHER_USER = "@other_user:localhost"
|
|
|
|
# have the user join
|
|
self.get_success(
|
|
inject_member_event(self.hs, self.room_id, OTHER_USER, Membership.JOIN)
|
|
)
|
|
|
|
# Update existing power levels with mod at PL50
|
|
pls = self.helper.get_state(
|
|
self.room_id, EventTypes.PowerLevels, tok=self.user_tok
|
|
)
|
|
pls["users"][OTHER_USER] = 50
|
|
self.helper.send_state(
|
|
self.room_id,
|
|
EventTypes.PowerLevels,
|
|
pls,
|
|
tok=self.user_tok,
|
|
)
|
|
|
|
# this is the point in the DAG where we make a fork
|
|
fork_point = self.get_success(
|
|
self.hs.get_datastores().main.get_latest_event_ids_in_room(self.room_id)
|
|
)
|
|
|
|
events = [
|
|
self._inject_state_event(sender=OTHER_USER)
|
|
for _ in range(num_state_changes)
|
|
]
|
|
|
|
self.replicate()
|
|
# all those events and state changes should have landed
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertGreaterEqual(len(received_event_rows), 2 * len(events))
|
|
|
|
# disconnect, so that we can stack up the changes
|
|
self.disconnect()
|
|
self.test_handler.received_rdata_rows.clear()
|
|
|
|
# a state event which doesn't get rolled back, to check that the state
|
|
# before the huge update comes through ok
|
|
state1 = self._inject_state_event()
|
|
|
|
# roll back all the state by de-modding the user
|
|
prev_events = fork_point
|
|
pls["users"][OTHER_USER] = 0
|
|
pl_event = self.get_success(
|
|
inject_event(
|
|
self.hs,
|
|
prev_event_ids=list(prev_events),
|
|
type=EventTypes.PowerLevels,
|
|
state_key="",
|
|
sender=self.user_id,
|
|
room_id=self.room_id,
|
|
content=pls,
|
|
)
|
|
)
|
|
|
|
# one more bit of state that doesn't get rolled back
|
|
state2 = self._inject_state_event()
|
|
|
|
# check we're testing what we think we are: no rows should yet have been
|
|
# received
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
# now reconnect to pull the updates
|
|
self.reconnect()
|
|
self.replicate()
|
|
|
|
# we should have received all the expected rows in the right order (as
|
|
# well as various cache invalidation updates which we ignore)
|
|
#
|
|
# we expect:
|
|
#
|
|
# - two rows for state1
|
|
# - the PL event row, plus state rows for the PL event and each
|
|
# of the states that got reverted.
|
|
# - two rows for state2
|
|
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
# first check the first two rows, which should be the state1 event.
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, state1.event_id)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state")
|
|
self.assertIsInstance(row.data, EventsStreamCurrentStateRow)
|
|
self.assertEqual(row.data.event_id, state1.event_id)
|
|
|
|
# now the last two rows, which should be the state2 event.
|
|
stream_name, token, row = received_event_rows.pop(-2)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, state2.event_id)
|
|
|
|
stream_name, token, row = received_event_rows.pop(-1)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state")
|
|
self.assertIsInstance(row.data, EventsStreamCurrentStateRow)
|
|
self.assertEqual(row.data.event_id, state2.event_id)
|
|
|
|
# Based on the number of
|
|
if collapse_state_changes:
|
|
# that should leave us with the rows for the PL event, the state changes
|
|
# get collapsed into a single row.
|
|
self.assertEqual(len(received_event_rows), 2)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, pl_event.event_id)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state-all")
|
|
self.assertIsInstance(row.data, EventsStreamAllStateRow)
|
|
self.assertEqual(row.data.room_id, state2.room_id)
|
|
|
|
else:
|
|
# that should leave us with the rows for the PL event
|
|
self.assertEqual(len(received_event_rows), len(events) + 2)
|
|
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, pl_event.event_id)
|
|
|
|
# the state rows are unsorted
|
|
state_rows: list[EventsStreamCurrentStateRow] = []
|
|
for stream_name, _, row in received_event_rows:
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state")
|
|
self.assertIsInstance(row.data, EventsStreamCurrentStateRow)
|
|
state_rows.append(row.data)
|
|
|
|
state_rows.sort(key=lambda r: r.state_key)
|
|
|
|
sr = state_rows.pop(0)
|
|
self.assertEqual(sr.type, EventTypes.PowerLevels)
|
|
self.assertEqual(sr.event_id, pl_event.event_id)
|
|
for sr in state_rows:
|
|
self.assertEqual(sr.type, "test_state_event")
|
|
# "None" indicates the state has been deleted
|
|
self.assertIsNone(sr.event_id)
|
|
|
|
def test_update_function_state_row_limit(self) -> None:
|
|
"""Test replication with many state events over several stream ids."""
|
|
|
|
# we want to generate lots of state changes, but for this test, we want to
|
|
# spread out the state changes over a few stream IDs.
|
|
#
|
|
# We do this by having two branches in the DAG. On one, we have four moderators,
|
|
# each of which that generates lots of state; on the other, we de-op the users,
|
|
# thus invalidating all the state.
|
|
|
|
NUM_USERS = 4
|
|
STATES_PER_USER = _STREAM_UPDATE_TARGET_ROW_COUNT // 4 + 1
|
|
|
|
user_ids = ["@user%i:localhost" % (i,) for i in range(NUM_USERS)]
|
|
|
|
# have the users join
|
|
for u in user_ids:
|
|
self.get_success(
|
|
inject_member_event(self.hs, self.room_id, u, Membership.JOIN)
|
|
)
|
|
|
|
# Update existing power levels with mod at PL50
|
|
pls = self.helper.get_state(
|
|
self.room_id, EventTypes.PowerLevels, tok=self.user_tok
|
|
)
|
|
pls["users"].update(dict.fromkeys(user_ids, 50))
|
|
self.helper.send_state(
|
|
self.room_id,
|
|
EventTypes.PowerLevels,
|
|
pls,
|
|
tok=self.user_tok,
|
|
)
|
|
|
|
# this is the point in the DAG where we make a fork
|
|
fork_point = self.get_success(
|
|
self.hs.get_datastores().main.get_latest_event_ids_in_room(self.room_id)
|
|
)
|
|
|
|
events: list[EventBase] = []
|
|
for user in user_ids:
|
|
events.extend(
|
|
self._inject_state_event(sender=user) for _ in range(STATES_PER_USER)
|
|
)
|
|
|
|
self.replicate()
|
|
|
|
# all those events and state changes should have landed
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertGreaterEqual(len(received_event_rows), 2 * len(events))
|
|
|
|
# disconnect, so that we can stack up the changes
|
|
self.disconnect()
|
|
self.test_handler.received_rdata_rows.clear()
|
|
|
|
# now roll back all that state by de-modding the users
|
|
prev_events = list(fork_point)
|
|
pl_events = []
|
|
for u in user_ids:
|
|
pls["users"][u] = 0
|
|
e = self.get_success(
|
|
inject_event(
|
|
self.hs,
|
|
prev_event_ids=prev_events,
|
|
type=EventTypes.PowerLevels,
|
|
state_key="",
|
|
sender=self.user_id,
|
|
room_id=self.room_id,
|
|
content=pls,
|
|
)
|
|
)
|
|
prev_events = [e.event_id]
|
|
pl_events.append(e)
|
|
|
|
# check we're testing what we think we are: no rows should yet have been
|
|
# received
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
# now reconnect to pull the updates
|
|
self.reconnect()
|
|
self.replicate()
|
|
|
|
# we should have received all the expected rows in the right order (as
|
|
# well as various cache invalidation updates which we ignore)
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertGreaterEqual(len(received_event_rows), len(events))
|
|
for i in range(NUM_USERS):
|
|
# for each user, we expect the PL event row, followed by state rows for
|
|
# the PL event and each of the states that got reverted.
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, pl_events[i].event_id)
|
|
|
|
# the state rows are unsorted
|
|
state_rows: list[EventsStreamCurrentStateRow] = []
|
|
for _ in range(STATES_PER_USER + 1):
|
|
stream_name, token, row = received_event_rows.pop(0)
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "state")
|
|
self.assertIsInstance(row.data, EventsStreamCurrentStateRow)
|
|
state_rows.append(row.data)
|
|
|
|
state_rows.sort(key=lambda r: r.state_key)
|
|
|
|
sr = state_rows.pop(0)
|
|
self.assertEqual(sr.type, EventTypes.PowerLevels)
|
|
self.assertEqual(sr.event_id, pl_events[i].event_id)
|
|
for sr in state_rows:
|
|
self.assertEqual(sr.type, "test_state_event")
|
|
# "None" indicates the state has been deleted
|
|
self.assertIsNone(sr.event_id)
|
|
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
def test_backwards_stream_id(self) -> None:
|
|
"""
|
|
Test that RDATA that comes after the current position should be discarded.
|
|
"""
|
|
# disconnect, so that we can stack up some changes
|
|
self.disconnect()
|
|
|
|
# Generate an events. We inject them using inject_event so that they are
|
|
# not send out over replication until we call self.replicate().
|
|
event = self._inject_test_event()
|
|
|
|
# check we're testing what we think we are: no rows should yet have been
|
|
# received
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertEqual([], received_event_rows)
|
|
|
|
# now reconnect to pull the updates
|
|
self.reconnect()
|
|
self.replicate()
|
|
|
|
# We should have received the expected single row (as well as various
|
|
# cache invalidation updates which we ignore).
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
|
|
# There should be a single received row.
|
|
self.assertEqual(len(received_event_rows), 1)
|
|
|
|
stream_name, token, row = received_event_rows[0]
|
|
self.assertEqual("events", stream_name)
|
|
self.assertIsInstance(row, EventsStreamRow)
|
|
self.assertEqual(row.type, "ev")
|
|
self.assertIsInstance(row.data, EventsStreamEventRow)
|
|
self.assertEqual(row.data.event_id, event.event_id)
|
|
|
|
# Reset the data.
|
|
self.test_handler.received_rdata_rows = []
|
|
|
|
# Save the current token for later.
|
|
worker_events_stream = self.worker_hs.get_replication_streams()["events"]
|
|
prev_token = worker_events_stream.current_token("master")
|
|
|
|
# Manually send an old RDATA command, which should get dropped. This
|
|
# re-uses the row from above, but with an earlier stream token.
|
|
self.hs.get_replication_command_handler().send_command(
|
|
RdataCommand("events", "master", 1, row)
|
|
)
|
|
|
|
# No updates have been received (because it was discard as old).
|
|
received_event_rows = [
|
|
row
|
|
for row in self.test_handler.received_rdata_rows
|
|
if row[0] == EventsStream.NAME
|
|
]
|
|
self.assertEqual(len(received_event_rows), 0)
|
|
|
|
# Ensure the stream has not gone backwards.
|
|
current_token = worker_events_stream.current_token("master")
|
|
self.assertGreaterEqual(current_token, prev_token)
|
|
|
|
event_count = 0
|
|
|
|
def _inject_test_event(
|
|
self, body: str | None = None, sender: str | None = None, **kwargs: Any
|
|
) -> EventBase:
|
|
if sender is None:
|
|
sender = self.user_id
|
|
|
|
if body is None:
|
|
body = "event %i" % (self.event_count,)
|
|
self.event_count += 1
|
|
|
|
return self.get_success(
|
|
inject_event(
|
|
self.hs,
|
|
room_id=self.room_id,
|
|
sender=sender,
|
|
type="test_event",
|
|
content={"body": body},
|
|
**kwargs,
|
|
)
|
|
)
|
|
|
|
def _inject_state_event(
|
|
self,
|
|
body: str | None = None,
|
|
state_key: str | None = None,
|
|
sender: str | None = None,
|
|
) -> EventBase:
|
|
if sender is None:
|
|
sender = self.user_id
|
|
|
|
if state_key is None:
|
|
state_key = "state_%i" % (self.event_count,)
|
|
self.event_count += 1
|
|
|
|
if body is None:
|
|
body = "state event %s" % (state_key,)
|
|
|
|
return self.get_success(
|
|
inject_event(
|
|
self.hs,
|
|
room_id=self.room_id,
|
|
sender=sender,
|
|
type="test_state_event",
|
|
state_key=state_key,
|
|
content={"body": body},
|
|
)
|
|
)
|
|
|
|
|
|
class EventsStreamBatchTokensTestCase(unittest.HomeserverTestCase):
|
|
"""Tests for the tokens `EventsStream` hands to readers, which need no
|
|
worker (so they run under SQLite too, unlike `EventsStreamTestCase`)."""
|
|
|
|
servlets = [
|
|
admin.register_servlets,
|
|
login.register_servlets,
|
|
room.register_servlets,
|
|
]
|
|
|
|
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
|
|
self.user_id = self.register_user("u1", "pass")
|
|
self.user_tok = self.login("u1", "pass")
|
|
self.room_id = self.helper.create_room_as(tok=self.user_tok)
|
|
|
|
def test_update_function_splits_persist_batches(self) -> None:
|
|
"""The events stream emits a distinct token for every event stream
|
|
ordering (`_batch_updates` only collapses rows that share a token), and
|
|
`_process_rdata` calls `on_rdata` -- and hence
|
|
`process_replication_position` -- once per token. So a reader worker
|
|
(e.g. a sync worker) advances its events-stream position *through* the
|
|
middle of a persist batch, one event at a time, while reading
|
|
`current_state_delta_stream` -- whose rows are stamped with the batch
|
|
minimum -- straight from the database.
|
|
|
|
This is what makes the hand-built `since` token in
|
|
`tests/rest/client/test_sync.py::SyncStateAfterTimelineStateTestCase`
|
|
one a real deployment hands out.
|
|
"""
|
|
store = self.hs.get_datastores().main
|
|
before = store.get_room_max_token().stream
|
|
message, state_event = self.get_success(
|
|
persist_message_and_state_event_in_one_batch(
|
|
self.hs, self.room_id, self.user_id
|
|
)
|
|
)
|
|
after = store.get_room_max_token().stream
|
|
|
|
message_pos = message.internal_metadata.stream_ordering
|
|
state_pos = state_event.internal_metadata.stream_ordering
|
|
assert message_pos is not None and state_pos is not None
|
|
self.assertLess(message_pos, state_pos)
|
|
|
|
stream = EventsStream(self.hs)
|
|
updates, _upto, _limited = self.get_success(
|
|
# A limit of 100 is comfortably more than the handful of rows this
|
|
# test persists, so the update batch is never truncated.
|
|
stream._update_function("master", before, after, 100)
|
|
)
|
|
|
|
# The tokens a reader will actually advance to, in order.
|
|
delivered = [token for token, _row in _batch_updates(updates) if token]
|
|
|
|
self.assertIn(
|
|
message_pos,
|
|
delivered,
|
|
"a reader worker advances to the batch minimum before it sees the "
|
|
"state event",
|
|
)
|
|
self.assertIn(state_pos, delivered)
|
|
self.assertLess(delivered.index(message_pos), delivered.index(state_pos))
|