From e894f67509827088f6bdc6fa1d530568a6fc4875 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 20 Aug 2020 15:35:23 +0100
Subject: Implement config and routing for multiple event writers

---
 synapse/storage/databases/__init__.py              | 2 +-
 synapse/storage/databases/main/event_federation.py | 2 +-
 synapse/storage/databases/main/events.py           | 2 +-
 synapse/storage/databases/main/events_worker.py    | 2 +-
 synapse/storage/databases/main/stream.py           | 2 +-
 5 files changed, 5 insertions(+), 5 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/databases/__init__.py b/synapse/storage/databases/__init__.py
index 0ac854aee2..c73d54fb67 100644
--- a/synapse/storage/databases/__init__.py
+++ b/synapse/storage/databases/__init__.py
@@ -68,7 +68,7 @@ class Databases(object):
 
                     # If we're on a process that can persist events also
                     # instantiate a `PersistEventsStore`
-                    if hs.config.worker.writers.events == hs.get_instance_name():
+                    if hs.get_instance_name() in hs.config.worker.writers.events:
                         persist_events = PersistEventsStore(hs, database, main)
 
                 if "state" in database_config.databases:
diff --git a/synapse/storage/databases/main/event_federation.py b/synapse/storage/databases/main/event_federation.py
index 4826be630c..16f09fe531 100644
--- a/synapse/storage/databases/main/event_federation.py
+++ b/synapse/storage/databases/main/event_federation.py
@@ -440,7 +440,7 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore, SQLBas
         """
 
         if stream_ordering <= self.stream_ordering_month_ago:
-            raise StoreError(400, "stream_ordering too old")
+            raise StoreError(400, "stream_ordering too old %s" % (stream_ordering,))
 
         sql = """
                 SELECT event_id FROM stream_ordering_to_exterm
diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py
index 989372193f..bf2afc3dd7 100644
--- a/synapse/storage/databases/main/events.py
+++ b/synapse/storage/databases/main/events.py
@@ -110,7 +110,7 @@ class PersistEventsStore:
 
         # This should only exist on instances that are configured to write
         assert (
-            hs.config.worker.writers.events == hs.get_instance_name()
+            hs.get_instance_name() in hs.config.worker.writers.events
         ), "Can only instantiate EventsStore on master"
 
     async def _persist_events_and_state_updates(
diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py
index 8f46e04fa6..bda189b923 100644
--- a/synapse/storage/databases/main/events_worker.py
+++ b/synapse/storage/databases/main/events_worker.py
@@ -79,7 +79,7 @@ class EventsWorkerStore(SQLBaseStore):
     def __init__(self, database: DatabasePool, db_conn, hs):
         super(EventsWorkerStore, self).__init__(database, db_conn, hs)
 
-        if hs.config.worker.writers.events == hs.get_instance_name():
+        if hs.get_instance_name() in hs.config.worker.writers.events:
             # We are the process in charge of generating stream ids for events,
             # so instantiate ID generators based on the database
             if isinstance(database.engine, PostgresEngine):
diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py
index 497f607703..1ed7b4b2f7 100644
--- a/synapse/storage/databases/main/stream.py
+++ b/synapse/storage/databases/main/stream.py
@@ -1084,4 +1084,4 @@ class StreamStore(StreamWorkerStore):
         return self._stream_id_gen.get_current_token()
 
     def get_room_min_stream_ordering(self) -> int:
-        return self._backfill_id_gen.get_current_token()
+        return -self._backfill_id_gen.get_current_token()
-- 
cgit 1.5.1