mirror of
https://github.com/element-hq/synapse
synced 2024-08-20 07:20:20 +00:00
Create a separate ReplicationNotifier
This commit is contained in:
parent
cf18fea9e1
commit
71472bac91
15 changed files with 89 additions and 13 deletions
|
@ -226,8 +226,7 @@ class Notifier:
|
|||
self.store = hs.get_datastores().main
|
||||
self.pending_new_room_events: List[_PendingRoomEventEntry] = []
|
||||
|
||||
# Called when there are new things to stream over replication
|
||||
self.replication_callbacks: List[Callable[[], None]] = []
|
||||
self._replication_notifier = hs.get_replication_notifier()
|
||||
self._new_join_in_room_callbacks: List[Callable[[str, str], None]] = []
|
||||
|
||||
self._federation_client = hs.get_federation_http_client()
|
||||
|
@ -279,7 +278,7 @@ class Notifier:
|
|||
it needs to do any asynchronous work, a background thread should be started and
|
||||
wrapped with run_as_background_process.
|
||||
"""
|
||||
self.replication_callbacks.append(cb)
|
||||
self._replication_notifier.add_replication_callback(cb)
|
||||
|
||||
def add_new_join_in_room_callback(self, cb: Callable[[str, str], None]) -> None:
|
||||
"""Add a callback that will be called when a user joins a room.
|
||||
|
@ -741,8 +740,7 @@ class Notifier:
|
|||
|
||||
def notify_replication(self) -> None:
|
||||
"""Notify the any replication listeners that there's a new event"""
|
||||
for cb in self.replication_callbacks:
|
||||
cb()
|
||||
self._replication_notifier.notify_replication()
|
||||
|
||||
def notify_user_joined_room(self, event_id: str, room_id: str) -> None:
|
||||
for cb in self._new_join_in_room_callbacks:
|
||||
|
@ -759,3 +757,26 @@ class Notifier:
|
|||
# Tell the federation client about the fact the server is back up, so
|
||||
# that any in flight requests can be immediately retried.
|
||||
self._federation_client.wake_destination(server)
|
||||
|
||||
|
||||
@attr.s
|
||||
class ReplicationNotifier:
|
||||
"""Tracks callbacks for things that need to know about stream changes.
|
||||
|
||||
This is separate from the notifier to avoid circular dependencies.
|
||||
"""
|
||||
|
||||
_replication_callbacks: List[Callable[[], None]] = attr.Factory(list)
|
||||
|
||||
def add_replication_callback(self, cb: Callable[[], None]) -> None:
|
||||
"""Add a callback that will be called when some new data is available.
|
||||
Callback is not given any arguments. It should *not* return a Deferred - if
|
||||
it needs to do any asynchronous work, a background thread should be started and
|
||||
wrapped with run_as_background_process.
|
||||
"""
|
||||
self._replication_callbacks.append(cb)
|
||||
|
||||
def notify_replication(self) -> None:
|
||||
"""Notify the any replication listeners that there's a new event"""
|
||||
for cb in self._replication_callbacks:
|
||||
cb()
|
||||
|
|
|
@ -107,7 +107,7 @@ from synapse.http.client import InsecureInterceptableContextFactory, SimpleHttpC
|
|||
from synapse.http.matrixfederationclient import MatrixFederationHttpClient
|
||||
from synapse.metrics.common_usage_metrics import CommonUsageMetricsManager
|
||||
from synapse.module_api import ModuleApi
|
||||
from synapse.notifier import Notifier
|
||||
from synapse.notifier import Notifier, ReplicationNotifier
|
||||
from synapse.push.bulk_push_rule_evaluator import BulkPushRuleEvaluator
|
||||
from synapse.push.pusherpool import PusherPool
|
||||
from synapse.replication.tcp.client import ReplicationDataHandler
|
||||
|
@ -389,6 +389,10 @@ class HomeServer(metaclass=abc.ABCMeta):
|
|||
def get_notifier(self) -> Notifier:
|
||||
return Notifier(self)
|
||||
|
||||
@cache_in_self
|
||||
def get_replication_notifier(self) -> ReplicationNotifier:
|
||||
return ReplicationNotifier()
|
||||
|
||||
@cache_in_self
|
||||
def get_auth(self) -> Auth:
|
||||
return Auth(self)
|
||||
|
|
|
@ -75,6 +75,7 @@ class AccountDataWorkerStore(PushRulesWorkerStore, CacheInvalidationWorkerStore)
|
|||
self._account_data_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="account_data",
|
||||
instance_name=self._instance_name,
|
||||
tables=[
|
||||
|
@ -95,6 +96,7 @@ class AccountDataWorkerStore(PushRulesWorkerStore, CacheInvalidationWorkerStore)
|
|||
# SQLite).
|
||||
self._account_data_id_gen = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"room_account_data",
|
||||
"stream_id",
|
||||
extra_tables=[("room_tags_revisions", "stream_id")],
|
||||
|
|
|
@ -75,6 +75,7 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
|
|||
self._cache_id_gen = MultiWriterIdGenerator(
|
||||
db_conn,
|
||||
database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="caches",
|
||||
instance_name=hs.get_instance_name(),
|
||||
tables=[
|
||||
|
|
|
@ -91,6 +91,7 @@ class DeviceInboxWorkerStore(SQLBaseStore):
|
|||
MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="to_device",
|
||||
instance_name=self._instance_name,
|
||||
tables=[("device_inbox", "instance_name", "stream_id")],
|
||||
|
@ -101,7 +102,7 @@ class DeviceInboxWorkerStore(SQLBaseStore):
|
|||
else:
|
||||
self._can_write_to_device = True
|
||||
self._device_inbox_id_gen = StreamIdGenerator(
|
||||
db_conn, "device_inbox", "stream_id"
|
||||
db_conn, hs.get_replication_notifier(), "device_inbox", "stream_id"
|
||||
)
|
||||
|
||||
max_device_inbox_id = self._device_inbox_id_gen.get_current_token()
|
||||
|
|
|
@ -92,6 +92,7 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
|
|||
# class below that is used on the main process.
|
||||
self._device_list_id_gen: AbstractStreamIdTracker = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"device_lists_stream",
|
||||
"stream_id",
|
||||
extra_tables=[
|
||||
|
|
|
@ -1181,7 +1181,10 @@ class EndToEndKeyStore(EndToEndKeyWorkerStore, SQLBaseStore):
|
|||
super().__init__(database, db_conn, hs)
|
||||
|
||||
self._cross_signing_id_gen = StreamIdGenerator(
|
||||
db_conn, "e2e_cross_signing_keys", "stream_id"
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"e2e_cross_signing_keys",
|
||||
"stream_id",
|
||||
)
|
||||
|
||||
async def set_e2e_device_keys(
|
||||
|
|
|
@ -191,6 +191,7 @@ class EventsWorkerStore(SQLBaseStore):
|
|||
self._stream_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="events",
|
||||
instance_name=hs.get_instance_name(),
|
||||
tables=[("events", "instance_name", "stream_ordering")],
|
||||
|
@ -200,6 +201,7 @@ class EventsWorkerStore(SQLBaseStore):
|
|||
self._backfill_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="backfill",
|
||||
instance_name=hs.get_instance_name(),
|
||||
tables=[("events", "instance_name", "stream_ordering")],
|
||||
|
@ -217,12 +219,14 @@ class EventsWorkerStore(SQLBaseStore):
|
|||
# SQLite).
|
||||
self._stream_id_gen = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"events",
|
||||
"stream_ordering",
|
||||
is_writer=hs.get_instance_name() in hs.config.worker.writers.events,
|
||||
)
|
||||
self._backfill_id_gen = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"events",
|
||||
"stream_ordering",
|
||||
step=-1,
|
||||
|
@ -300,6 +304,7 @@ class EventsWorkerStore(SQLBaseStore):
|
|||
self._un_partial_stated_events_stream_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="un_partial_stated_event_stream",
|
||||
instance_name=hs.get_instance_name(),
|
||||
tables=[
|
||||
|
@ -311,7 +316,10 @@ class EventsWorkerStore(SQLBaseStore):
|
|||
)
|
||||
else:
|
||||
self._un_partial_stated_events_stream_id_gen = StreamIdGenerator(
|
||||
db_conn, "un_partial_stated_event_stream", "stream_id"
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"un_partial_stated_event_stream",
|
||||
"stream_id",
|
||||
)
|
||||
|
||||
def get_un_partial_stated_events_token(self) -> int:
|
||||
|
|
|
@ -77,6 +77,7 @@ class PresenceStore(PresenceBackgroundUpdateStore, CacheInvalidationWorkerStore)
|
|||
self._presence_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="presence_stream",
|
||||
instance_name=self._instance_name,
|
||||
tables=[("presence_stream", "instance_name", "stream_id")],
|
||||
|
@ -85,7 +86,7 @@ class PresenceStore(PresenceBackgroundUpdateStore, CacheInvalidationWorkerStore)
|
|||
)
|
||||
else:
|
||||
self._presence_id_gen = StreamIdGenerator(
|
||||
db_conn, "presence_stream", "stream_id"
|
||||
db_conn, hs.get_replication_notifier(), "presence_stream", "stream_id"
|
||||
)
|
||||
|
||||
self.hs = hs
|
||||
|
|
|
@ -118,6 +118,7 @@ class PushRulesWorkerStore(
|
|||
# class below that is used on the main process.
|
||||
self._push_rules_stream_id_gen: AbstractStreamIdTracker = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"push_rules_stream",
|
||||
"stream_id",
|
||||
is_writer=hs.config.worker.worker_app is None,
|
||||
|
|
|
@ -62,6 +62,7 @@ class PusherWorkerStore(SQLBaseStore):
|
|||
# class below that is used on the main process.
|
||||
self._pushers_id_gen: AbstractStreamIdTracker = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"pushers",
|
||||
"id",
|
||||
extra_tables=[("deleted_pushers", "stream_id")],
|
||||
|
|
|
@ -73,6 +73,7 @@ class ReceiptsWorkerStore(SQLBaseStore):
|
|||
self._receipts_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="receipts",
|
||||
instance_name=self._instance_name,
|
||||
tables=[("receipts_linearized", "instance_name", "stream_id")],
|
||||
|
@ -91,6 +92,7 @@ class ReceiptsWorkerStore(SQLBaseStore):
|
|||
# SQLite).
|
||||
self._receipts_id_gen = StreamIdGenerator(
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"receipts_linearized",
|
||||
"stream_id",
|
||||
is_writer=hs.get_instance_name() in hs.config.worker.writers.receipts,
|
||||
|
|
|
@ -126,6 +126,7 @@ class RoomWorkerStore(CacheInvalidationWorkerStore):
|
|||
self._un_partial_stated_rooms_stream_id_gen = MultiWriterIdGenerator(
|
||||
db_conn=db_conn,
|
||||
db=database,
|
||||
notifier=hs.get_replication_notifier(),
|
||||
stream_name="un_partial_stated_room_stream",
|
||||
instance_name=self._instance_name,
|
||||
tables=[
|
||||
|
@ -137,7 +138,10 @@ class RoomWorkerStore(CacheInvalidationWorkerStore):
|
|||
)
|
||||
else:
|
||||
self._un_partial_stated_rooms_stream_id_gen = StreamIdGenerator(
|
||||
db_conn, "un_partial_stated_room_stream", "stream_id"
|
||||
db_conn,
|
||||
hs.get_replication_notifier(),
|
||||
"un_partial_stated_room_stream",
|
||||
"stream_id",
|
||||
)
|
||||
|
||||
async def store_room(
|
||||
|
|
|
@ -20,6 +20,7 @@ from collections import OrderedDict
|
|||
from contextlib import contextmanager
|
||||
from types import TracebackType
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
AsyncContextManager,
|
||||
ContextManager,
|
||||
Dict,
|
||||
|
@ -49,6 +50,9 @@ from synapse.storage.database import (
|
|||
from synapse.storage.types import Cursor
|
||||
from synapse.storage.util.sequence import PostgresSequenceGenerator
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from synapse.notifier import ReplicationNotifier
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
@ -182,6 +186,7 @@ class StreamIdGenerator(AbstractStreamIdGenerator):
|
|||
def __init__(
|
||||
self,
|
||||
db_conn: LoggingDatabaseConnection,
|
||||
notifier: "ReplicationNotifier",
|
||||
table: str,
|
||||
column: str,
|
||||
extra_tables: Iterable[Tuple[str, str]] = (),
|
||||
|
@ -205,6 +210,8 @@ class StreamIdGenerator(AbstractStreamIdGenerator):
|
|||
# The key and values are the same, but we never look at the values.
|
||||
self._unfinished_ids: OrderedDict[int, int] = OrderedDict()
|
||||
|
||||
self._notifier = notifier
|
||||
|
||||
def advance(self, instance_name: str, new_id: int) -> None:
|
||||
# Advance should never be called on a writer instance, only over replication
|
||||
if self._is_writer:
|
||||
|
@ -227,6 +234,8 @@ class StreamIdGenerator(AbstractStreamIdGenerator):
|
|||
with self._lock:
|
||||
self._unfinished_ids.pop(next_id)
|
||||
|
||||
self._notifier.notify_replication()
|
||||
|
||||
return _AsyncCtxManagerWrapper(manager())
|
||||
|
||||
def get_next_mult(self, n: int) -> AsyncContextManager[Sequence[int]]:
|
||||
|
@ -250,6 +259,8 @@ class StreamIdGenerator(AbstractStreamIdGenerator):
|
|||
for next_id in next_ids:
|
||||
self._unfinished_ids.pop(next_id)
|
||||
|
||||
self._notifier.notify_replication()
|
||||
|
||||
return _AsyncCtxManagerWrapper(manager())
|
||||
|
||||
def get_current_token(self) -> int:
|
||||
|
@ -296,6 +307,7 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
|
|||
self,
|
||||
db_conn: LoggingDatabaseConnection,
|
||||
db: DatabasePool,
|
||||
notifier: "ReplicationNotifier",
|
||||
stream_name: str,
|
||||
instance_name: str,
|
||||
tables: List[Tuple[str, str, str]],
|
||||
|
@ -304,6 +316,7 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
|
|||
positive: bool = True,
|
||||
) -> None:
|
||||
self._db = db
|
||||
self._notifier = notifier
|
||||
self._stream_name = stream_name
|
||||
self._instance_name = instance_name
|
||||
self._positive = positive
|
||||
|
@ -535,7 +548,9 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
|
|||
# Cast safety: the second argument to _MultiWriterCtxManager, multiple_ids,
|
||||
# controls the return type. If `None` or omitted, the context manager yields
|
||||
# a single integer stream_id; otherwise it yields a list of stream_ids.
|
||||
return cast(AsyncContextManager[int], _MultiWriterCtxManager(self))
|
||||
return cast(
|
||||
AsyncContextManager[int], _MultiWriterCtxManager(self, self._notifier)
|
||||
)
|
||||
|
||||
def get_next_mult(self, n: int) -> AsyncContextManager[List[int]]:
|
||||
# If we have a list of instances that are allowed to write to this
|
||||
|
@ -544,7 +559,10 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
|
|||
raise Exception("Tried to allocate stream ID on non-writer")
|
||||
|
||||
# Cast safety: see get_next.
|
||||
return cast(AsyncContextManager[List[int]], _MultiWriterCtxManager(self, n))
|
||||
return cast(
|
||||
AsyncContextManager[List[int]],
|
||||
_MultiWriterCtxManager(self, self._notifier, n),
|
||||
)
|
||||
|
||||
def get_next_txn(self, txn: LoggingTransaction) -> int:
|
||||
"""
|
||||
|
@ -563,6 +581,7 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
|
|||
|
||||
txn.call_after(self._mark_id_as_finished, next_id)
|
||||
txn.call_on_exception(self._mark_id_as_finished, next_id)
|
||||
txn.call_after(self._notifier.notify_replication)
|
||||
|
||||
# Update the `stream_positions` table with newly updated stream
|
||||
# ID (unless self._writers is not set in which case we don't
|
||||
|
@ -787,6 +806,7 @@ class _MultiWriterCtxManager:
|
|||
"""Async context manager returned by MultiWriterIdGenerator"""
|
||||
|
||||
id_gen: MultiWriterIdGenerator
|
||||
notifier: "ReplicationNotifier"
|
||||
multiple_ids: Optional[int] = None
|
||||
stream_ids: List[int] = attr.Factory(list)
|
||||
|
||||
|
@ -814,6 +834,8 @@ class _MultiWriterCtxManager:
|
|||
for i in self.stream_ids:
|
||||
self.id_gen._mark_id_as_finished(i)
|
||||
|
||||
self.notifier.notify_replication()
|
||||
|
||||
if exc_type is not None:
|
||||
return False
|
||||
|
||||
|
|
|
@ -52,6 +52,7 @@ class StreamIdGeneratorTestCase(HomeserverTestCase):
|
|||
def _create(conn: LoggingDatabaseConnection) -> StreamIdGenerator:
|
||||
return StreamIdGenerator(
|
||||
db_conn=conn,
|
||||
notifier=self.hs.get_replication_notifier(),
|
||||
table="foobar",
|
||||
column="stream_id",
|
||||
)
|
||||
|
@ -196,6 +197,7 @@ class MultiWriterIdGeneratorTestCase(HomeserverTestCase):
|
|||
return MultiWriterIdGenerator(
|
||||
conn,
|
||||
self.db_pool,
|
||||
notifier=self.hs.get_replication_notifier(),
|
||||
stream_name="test_stream",
|
||||
instance_name=instance_name,
|
||||
tables=[("foobar", "instance_name", "stream_id")],
|
||||
|
@ -630,6 +632,7 @@ class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
|
|||
return MultiWriterIdGenerator(
|
||||
conn,
|
||||
self.db_pool,
|
||||
notifier=self.hs.get_replication_notifier(),
|
||||
stream_name="test_stream",
|
||||
instance_name=instance_name,
|
||||
tables=[("foobar", "instance_name", "stream_id")],
|
||||
|
@ -766,6 +769,7 @@ class MultiTableMultiWriterIdGeneratorTestCase(HomeserverTestCase):
|
|||
return MultiWriterIdGenerator(
|
||||
conn,
|
||||
self.db_pool,
|
||||
notifier=self.hs.get_replication_notifier(),
|
||||
stream_name="test_stream",
|
||||
instance_name=instance_name,
|
||||
tables=[
|
||||
|
|
Loading…
Reference in a new issue