Split presence out of master

scripts/synapse_port_db

@@ -634,8 +634,11 @@ class Porter(object):
                 "device_inbox_sequence", ("device_inbox", "device_federation_outbox")
             )
             await self._setup_sequence(
-                "account_data_sequence", ("room_account_data", "room_tags_revisions", "account_data"))
-            await self._setup_sequence("receipts_sequence", ("receipts_linearized", ))
+                "account_data_sequence",
+                ("room_account_data", "room_tags_revisions", "account_data"),
+            )
+            await self._setup_sequence("receipts_sequence", ("receipts_linearized",))
+            await self._setup_sequence("presence_stream_sequence", ("presence_stream",))
             await self._setup_auth_chain_sequence()
 
             # Step 3. Get tables.

synapse/config/workers.py

@@ -64,10 +64,14 @@ class WriterLocations:
     Attributes:
         events: The instances that write to the event and backfill streams.
         typing: The instance that writes to the typing stream.
-        to_device: The instance that writes to the to_device stream.
-        account_data: The instance that writes to the account data streams.
-        receipts: The instance that writes to the receipts stream.
-        presence: The instance that writes to the presence stream.
+        to_device: The instances that write to the to_device stream. Currently
+            can only be a single instance.
+        account_data: The instances that write to the account data streams. Currently
+            can only be a single instance.
+        receipts: The instances that write to the receipts stream. Currently
+            can only be a single instance.
+        presence: The instances that write to the presence stream. Currently
+            can only be a single instance.
     """
 
     events = attr.ib(

synapse/replication/http/_base.py

@@ -158,7 +158,10 @@ async def _handle_request(self, request, **kwargs):
     def make_client(cls, hs):
         """Create a client that makes requests.
 
-        Returns a callable that accepts the same parameters as `_serialize_payload`.
+        Returns a callable that accepts the same parameters as
+        `_serialize_payload`, but also accepts an extra `instance_name`
+        parameter to specify which instance to hit (the instance must be in
+        the `instance_map` config).
         """
         clock = hs.get_clock()
         client = hs.get_simple_http_client()

synapse/storage/databases/main/presence.py

@@ -43,10 +43,6 @@ def __init__(
         )
 
         if isinstance(database.engine, PostgresEngine):
-            self._can_persist_presence = (
-                self._instance_name in hs.config.worker.writers.presence
-            )
-
             self._presence_id_gen = MultiWriterIdGenerator(
                 db_conn=db_conn,
                 db=database,
@@ -57,7 +53,6 @@ def __init__(
                 writers=hs.config.worker.writers.to_device,
             )
         else:
-            self._can_persist_presence = True
             self._presence_id_gen = StreamIdGenerator(
                 db_conn, "presence_stream", "stream_id"
             )

synapse/storage/databases/main/schema/delta/59/12presence_stream_instance.sql

@@ -13,4 +13,6 @@
  * limitations under the License.
  */
 
+--- Add a column to specify which instance wrote the row. Historic rows have
+--  `NULL`, which indicates that the master instance wrote them.
 ALTER TABLE presence_stream ADD COLUMN instance_name TEXT;