Fix bug when running presence off master (matrix-org#10149)

Hopefully fixes matrix-org#10027.

Author: erikjohnston

changelog.d/10149.bugfix

@@ -0,0 +1 @@
+Fix a bug which caused  presence updates to stop working some time after restart, when using a presence writer worker.

synapse/storage/databases/main/presence.py

@@ -50,7 +50,7 @@ def __init__(
                 instance_name=self._instance_name,
                 tables=[("presence_stream", "instance_name", "stream_id")],
                 sequence_name="presence_stream_sequence",
-                writers=hs.config.worker.writers.to_device,
+                writers=hs.config.worker.writers.presence,
             )
         else:
             self._presence_id_gen = StreamIdGenerator(

synapse/storage/util/id_generators.py

@@ -397,6 +397,11 @@ def get_next(self):
                 # ... persist event ...
         """
 
+        # If we have a list of instances that are allowed to write to this
+        # stream, make sure we're in it.
+        if self._writers and self._instance_name not in self._writers:
+            raise Exception("Tried to allocate stream ID on non-writer")
+
         return _MultiWriterCtxManager(self)
 
     def get_next_mult(self, n: int):
@@ -406,6 +411,11 @@ def get_next_mult(self, n: int):
                 # ... persist events ...
         """
 
+        # If we have a list of instances that are allowed to write to this
+        # stream, make sure we're in it.
+        if self._writers and self._instance_name not in self._writers:
+            raise Exception("Tried to allocate stream ID on non-writer")
+
         return _MultiWriterCtxManager(self, n)
 
     def get_next_txn(self, txn: LoggingTransaction):
@@ -416,6 +426,11 @@ def get_next_txn(self, txn: LoggingTransaction):
             # ... persist event ...
         """
 
+        # If we have a list of instances that are allowed to write to this
+        # stream, make sure we're in it.
+        if self._writers and self._instance_name not in self._writers:
+            raise Exception("Tried to allocate stream ID on non-writer")
+
         next_id = self._load_next_id_txn(txn)
 
         with self._lock: