Catch up MSC4354 Sticky Events to remote homeservers that have missed them.

changelog.d/20165.feature

@@ -0,0 +1 @@
+Catch up [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) to remote homeservers that have missed them.

changelog.d/20166.misc

@@ -0,0 +1 @@
+Refactor the federation transmission code to delineate transaction preparation and completion.

synapse/replication/tcp/streams/_base.py

@@ -19,22 +19,23 @@
 #
 #
 
 import heapq
 import logging
 from typing import (
     TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
     Sequence,
     Union,
+    NewType,
 )
 
 import attr
 
 from synapse.api.constants import AccountDataTypes, ProfileUpdateAction
 from synapse.replication.http.streams import ReplicationGetStreamUpdates
 from synapse.types import UserID
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -866,6 +867,13 @@
         return rows, rows[-1][0], len(updates) == limit
 
 
+StickyEventStreamPosition = NewType("StickyEventStreamPosition", int)
+"""
+Integer corresponding to the stream position (`stream_id`)
+of a sticky event in the `sticky_events` table.
+"""
+
+
 @attr.s(slots=True, auto_attribs=True)
 class StickyEventsStreamRow:
     """Stream to inform workers about changes to sticky events."""

synapse/storage/database.py

@@ -2748,3 +2748,16 @@ def make_tuple_comparison_clause(keys: list[tuple[str, KV]]) -> tuple[str, list[
         "(%s) > (%s)" % (",".join(k[0] for k in keys), ",".join("?" for _ in keys)),
         [k[1] for k in keys],
     )
+
+
+def user_is_local_like_pattern(hs: "HomeServer") -> str:
+    """
+    Returns a LIKE pattern that matches the User IDs of local users on this
+    homeserver.
+
+    The caller should bind this pattern to a parameter and use it in
+    a `user_id LIKE ?` clause.
+    """
+    # This is good enough as if you have silly characters in your own
+    # hostname then that's your own fault.
+    return f"@%:{hs.hostname}"

synapse/storage/databases/main/event_push_actions.py

@@ -100,6 +100,7 @@
     LoggingDatabaseConnection,
     LoggingTransaction,
     PostgresEngine,
+    user_is_local_like_pattern,
 )
 from synapse.storage.databases.main.receipts import ReceiptsWorkerStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
@@ -1442,7 +1443,7 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool:
 
         # We only want local users, so we add a dodgy filter to the above query
         # and recheck it below.
-        user_filter = "%:" + self.hs.hostname
+        user_filter = user_is_local_like_pattern(self.hs)
 
         txn.execute(
             sql,

synapse/storage/databases/main/metrics.py

@@ -30,6 +30,7 @@
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    user_is_local_like_pattern,
 )
 from synapse.storage.databases.main.event_push_actions import (
     EventPushActionsWorkerStore,
@@ -133,9 +134,7 @@ def _count_messages(txn: LoggingTransaction) -> int:
 
     async def count_daily_sent_e2ee_messages(self) -> int:
         def _count_messages(txn: LoggingTransaction) -> int:
-            # This is good enough as if you have silly characters in your own
-            # hostname then that's your own fault.
-            like_clause = "%:" + self.hs.hostname
+            like_clause = user_is_local_like_pattern(self.hs)
 
             sql = """
                 SELECT COUNT(*) FROM events
@@ -189,9 +188,7 @@ def _count_messages(txn: LoggingTransaction) -> int:
 
     async def count_daily_sent_messages(self) -> int:
         def _count_messages(txn: LoggingTransaction) -> int:
-            # This is good enough as if you have silly characters in your own
-            # hostname then that's your own fault.
-            like_clause = "%:" + self.hs.hostname
+            like_clause = user_is_local_like_pattern(self.hs)
 
             sql = """
                 SELECT COUNT(*) FROM events

synapse/storage/databases/main/purge_events.py

@@ -23,7 +23,7 @@
 from typing import Any, cast
 
 from synapse.api.errors import SynapseError
-from synapse.storage.database import LoggingTransaction
+from synapse.storage.database import LoggingTransaction, user_is_local_like_pattern
 from synapse.storage.databases.main import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.engines import PostgresEngine
@@ -213,7 +213,10 @@ def _purge_history_txn(
             should_delete_expr += " AND sender NOT LIKE ?"
 
             # We include the parameter twice since we use the expression twice
-            should_delete_params += ("%:" + self.hs.hostname, "%:" + self.hs.hostname)
+            should_delete_params += (
+                user_is_local_like_pattern(self.hs),
+                user_is_local_like_pattern(self.hs),
+            )
 
         should_delete_params += (room_id, token.topological)