Move persist_events out from main data store.

changelog.d/6240.misc

@@ -0,0 +1 @@
+Move `persist_events` out from main data store.

synapse/handlers/federation.py

@@ -109,6 +109,7 @@ def __init__(self, hs):
         self.hs = hs
 
         self.store = hs.get_datastore()
+        self.storage = hs.get_storage()
         self.federation_client = hs.get_federation_client()
         self.state_handler = hs.get_state_handler()
         self.server_name = hs.hostname
@@ -2648,7 +2649,7 @@ def persist_events_and_notify(self, event_and_contexts, backfilled=False):
                 backfilled=backfilled,
             )
         else:
-            max_stream_id = yield self.store.persist_events(
+            max_stream_id = yield self.storage.persistence.persist_events(
                 event_and_contexts, backfilled=backfilled
             )
 

synapse/handlers/message.py

@@ -234,6 +234,7 @@ def __init__(self, hs):
         self.hs = hs
         self.auth = hs.get_auth()
         self.store = hs.get_datastore()
+        self.storage = hs.get_storage()
         self.state = hs.get_state_handler()
         self.clock = hs.get_clock()
         self.validator = EventValidator()
@@ -868,7 +869,7 @@ def is_inviter_member_event(e):
             if prev_state_ids:
                 raise AuthError(403, "Changing the room create event is forbidden")
 
-        (event_stream_id, max_stream_id) = yield self.store.persist_event(
+        event_stream_id, max_stream_id = yield self.storage.persistence.persist_event(
             event, context=context
         )
 

synapse/server.py

@@ -95,6 +95,7 @@
     WorkerServerNoticesSender,
 )
 from synapse.state import StateHandler, StateResolutionHandler
+from synapse.storage import DataStores, Storage
 from synapse.streams.events import EventSources
 from synapse.util import Clock
 from synapse.util.distributor import Distributor
@@ -196,6 +197,7 @@ def build_DEPENDENCY(self)
         "account_validity_handler",
         "saml_handler",
         "event_client_serializer",
+        "storage",
     ]
 
     REQUIRED_ON_MASTER_STARTUP = ["user_directory_handler", "stats_handler"]
@@ -225,6 +227,7 @@ def __init__(self, hostname, reactor=None, **kwargs):
         self.registration_ratelimiter = Ratelimiter()
 
         self.datastore = None
+        self.datastores = None
 
         # Other kwargs are explicit dependencies
         for depname in kwargs:
@@ -234,6 +237,7 @@ def setup(self):
         logger.info("Setting up.")
         with self.get_db_conn() as conn:
             self.datastore = self.DATASTORE_CLASS(conn, self)
+            self.datastores = DataStores(self.datastore, conn, self)
             conn.commit()
         logger.info("Finished setting up.")
 
@@ -266,7 +270,7 @@ def get_clock(self):
         return self.clock
 
     def get_datastore(self):
-        return self.datastore
+        return self.datastores.main
 
     def get_config(self):
         return self.config
@@ -537,6 +541,9 @@ def build_saml_handler(self):
     def build_event_client_serializer(self):
         return EventClientSerializer(self)
 
+    def build_storage(self) -> Storage:
+        return Storage(self, self.datastores)
+
     def remove_pusher(self, app_id, push_key, user_id):
         return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
 

synapse/storage/__init__.py

@@ -27,7 +27,24 @@
 stored in `synapse.storage.schema`.
 """
 
-from synapse.storage.data_stores.main import DataStore  # noqa: F401
+from synapse.storage.data_stores import DataStores
+from synapse.storage.data_stores.main import DataStore
+from synapse.storage.persist_events import EventsPersistenceStore
+
+__all__ = ["DataStores", "DataStore"]
+
+
+class Storage(object):
+    """The high level interfaces for talking to various storage layers.
+    """
+
+    def __init__(self, hs, stores: DataStores):
+        # We include the main data store here mainly so that we don't have to
+        # rewrite all the existing code to split it into high vs low level
+        # interfaces.
+        self.main = stores.main
+
+        self.persistence = EventsPersistenceStore(hs, stores)
 
 
 def are_all_users_on_domain(txn, database_engine, domain):

synapse/storage/data_stores/__init__.py

@@ -12,3 +12,15 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
+
+class DataStores(object):
+    """The various data stores.
+
+    These are low level interfaces to physical databases.
+    """
+
+    def __init__(self, main_store, db_conn, hs):
+        # Note we pass in the main store here as workers use a different main
+        # store.
+        self.main = main_store