Lock Sliding Sync connections when inserting lazy members, to prevent repeated deadlocks. (#19826)

Got paged today for this. The sliding sync worker in question had loads
of deadlocks in the logs.
I restarted it and it got unwedged, but we should have a more robust
defence, which this PR proposes.

```
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 257324 waits for ShareLock on transaction 688227036; blocked by process 254908.
Process 254908 waits for ShareLock on transaction 688222971; blocked by process 256179.
Process 256179 waits for ExclusiveLock on tuple (302352,92) of relation 2962200779 of database 16403; blocked by process 257213.
Process 257213 waits for ShareLock on transaction 688225005; blocked by process 254905.
Process 254905 waits for ShareLock on transaction 688228814; blocked by process 257324.
HINT:  See server log for query details.
CONTEXT:  while inserting index tuple (183070,103) in relation "sliding_sync_connection_lazy_members"
```

I wonder if an unfortunate side effect is that these repeated attempts
leave a lot of dead tuples on the table,
which would then harm the performance of the next attempt to insert the
tuples,
I suspect making it more likely that they will deadlock again (?).

---

By acquring a `FOR NO KEY UPDATE` lock upfront before beginning work, we
can ensure that one
of the transactions gets queued behind the other one, meaning the first
one can succeed unimpeded.

`FOR NO KEY UPDATE` blocks other `FOR NO KEY UPDATE` locks and is the
weakest lock level that blocks itself.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>

Author: reivilibre

changelog.d/19826.bugfix

@@ -0,0 +1 @@
+Lock Sliding Sync connections when inserting lazy members, to prevent repeated deadlocks.

synapse/storage/databases/main/sliding_sync.py

@@ -186,15 +186,35 @@ def persist_per_connection_state_txn(
         # First we fetch (or create) the connection key associated with the
         # previous connection position.
         if previous_connection_position is not None:
+            lock_clause = ""
+            if isinstance(self.database_engine, PostgresEngine):
+                # Lock the sliding sync connection row for update upfront,
+                # to prevent deadlocks between concurrent transactions
+                # (which can retry again and again without making progress).
+                #
+                # (We don't need to explicitly lock in the other branch,
+                # where we re-create the connection, as that implies a lock
+                # anyway)
+                #
+                # Specifically, the statements seen to deadlock against
+                # each other were
+                # `INSERT INTO sliding_sync_connection_lazy_members`
+                # with conflicting tuples on
+                #     "sliding_sync_connection_lazy_members_idx" UNIQUE, btree
+                #     (connection_key, room_id, user_id)
+                # https://www.postgresql.org/docs/current/explicit-locking.html#LOCKING-ROWS
+                lock_clause = "FOR NO KEY UPDATE OF sliding_sync_connections"
+
             # The `previous_connection_position` is a user-supplied value, so we
             # need to make sure that the one they supplied is actually theirs.
-            sql = """
+            sql = f"""
                 SELECT connection_key
                 FROM sliding_sync_connection_positions
                 INNER JOIN sliding_sync_connections USING (connection_key)
                 WHERE
                     connection_position = ?
                     AND user_id = ? AND effective_device_id = ? AND conn_id = ?
+                {lock_clause}
             """
             txn.execute(
                 sql, (previous_connection_position, user_id, device_id, conn_id)