Queue the Rust persister's writes

The persist transaction sent its statements one at a time: about forty
round trips for a new room, about twenty for a message.

Every write whose result nothing reads is now queued: the batch inserts
and upserts, the deletes and updates, the current state and membership
batches, the sliding sync rows, threads and sticky events. Queued writes
go out together ahead of the next read, or with the COMMIT, in the order
the code issues them, so the statements run in the same order as before.
What is left are the reads, each a round trip carrying the writes before
it: the room's share lock, room depth, outliers, redactions, backward
extremities and the cache stream's IDs.

The partial state insert stays immediate: it turns a constraint
violation into `PartialStateConflict`, so the queue is sent first, and a
queued write's own violation cannot be taken for it. `upsert_one`, for
out-of-band memberships, also stays immediate.

`OwnedColumn::copy` and `owned_params!` copy borrowed values into ones
a queued statement can own (`ToOwnedSql`), and `execute_batch_later`
queues a statement per row.

The new room's write transaction goes from 42 round trips to about 12.
The E2EE monitor's room on jki.re goes from 102 to 79 round trips
overall, a plain room from 90 to 63.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PVLsNh6akt4ifEizY4D1ta

rust/src/database/postgres/write.rs

@@ -41,11 +41,12 @@
 
 use std::{
     cell::RefCell,
+    collections::HashMap,
     future::Future,
     pin::Pin,
     sync::{
         atomic::{AtomicBool, AtomicU8, Ordering},
-        Mutex,
+        Arc, Mutex,
     },
     task::Poll,
     time::{Duration, Instant},
@@ -258,13 +259,104 @@ impl OwnedColumn {
     }
 }
 
+impl OwnedColumn {
+    /// A column of copies of `values`, for values borrowed from data that
+    /// does not outlive the call, such as `&str`s of an event.
+    pub fn copy<T: ToOwnedSql>(name: &'static str, sql_type: &'static str, values: &[T]) -> Self
+    where
+        Vec<T::Owned>: ToSql + Sync + Send + 'static,
+    {
+        OwnedColumn::new(name, sql_type, values.iter().map(T::to_owned_sql).collect())
+    }
+}
+
+/// A value that can be copied into one a queued statement can own.
+pub trait ToOwnedSql {
+    type Owned;
+    fn to_owned_sql(&self) -> Self::Owned;
+}
+
+impl ToOwnedSql for str {
+    type Owned = String;
+    fn to_owned_sql(&self) -> String {
+        self.to_owned()
+    }
+}
+
+impl<T: ToOwnedSql + ?Sized> ToOwnedSql for &T {
+    type Owned = T::Owned;
+    fn to_owned_sql(&self) -> T::Owned {
+        (**self).to_owned_sql()
+    }
+}
+
+impl<T: ToOwnedSql> ToOwnedSql for Vec<T> {
+    type Owned = Vec<T::Owned>;
+    fn to_owned_sql(&self) -> Vec<T::Owned> {
+        self.iter().map(T::to_owned_sql).collect()
+    }
+}
+
+impl<T: ToOwnedSql> ToOwnedSql for [T] {
+    type Owned = Vec<T::Owned>;
+    fn to_owned_sql(&self) -> Vec<T::Owned> {
+        self.iter().map(T::to_owned_sql).collect()
+    }
+}
+
+impl ToOwnedSql for Option<&str> {
+    type Owned = Option<String>;
+    fn to_owned_sql(&self) -> Option<String> {
+        self.map(str::to_owned)
+    }
+}
+
+impl ToOwnedSql for String {
+    type Owned = String;
+    fn to_owned_sql(&self) -> String {
+        self.clone()
+    }
+}
+
+impl ToOwnedSql for Option<String> {
+    type Owned = Option<String>;
+    fn to_owned_sql(&self) -> Option<String> {
+        self.clone()
+    }
+}
+
+macro_rules! copy_to_owned_sql {
+    ($($t:ty),*) => {
+        $(impl ToOwnedSql for $t {
+            type Owned = $t;
+            fn to_owned_sql(&self) -> $t {
+                *self
+            }
+        })*
+    };
+}
+
+copy_to_owned_sql!(i32, i64, bool, Option<i32>, Option<i64>, Option<bool>);
+
+/// The parameters of a queued statement, copied from borrowed values: for
+/// [`WriteTxn::execute_later`].
+#[macro_export]
+macro_rules! owned_params {
+    ($($param:expr),* $(,)?) => {
+        vec![$(Box::new(
+            $crate::database::postgres::write::ToOwnedSql::to_owned_sql(&$param),
+        ) as $crate::database::postgres::write::OwnedParam),*]
+    };
+}
+
 /// How to turn the error of a queued statement into the one its caller
 /// wants, e.g. a constraint violation into a domain error.
 pub type MapError = fn(WriteError) -> WriteError;
 
 /// A statement queued by one of the `*_later` methods.
 struct Queued {
-    sql: String,
+    /// Shared by the rows of an [`WriteTxn::execute_batch_later`].
+    sql: Arc<str>,
     params: Vec<OwnedParam>,
     map_error: Option<MapError>,
 }
@@ -379,18 +471,19 @@ impl WriteTxn {
         let queued = self.take_queued();
         let mut begin = self.take_begin();
 
-        let sqls: Vec<&str> = queued.iter().map(|q| q.sql.as_str()).chain(sql).collect();
-        let mut statements: Vec<Option<tokio_postgres::Statement>> =
-            sqls.iter().map(|sql| conn.cached(sql)).collect();
+        let sqls: Vec<&str> = queued.iter().map(|q| &*q.sql).chain(sql).collect();
+        // Each distinct statement once: a batch queues one row per statement.
+        let mut by_sql: HashMap<&str, Option<tokio_postgres::Statement>> = HashMap::new();
+        for sql in &sqls {
+            by_sql.entry(sql).or_insert_with(|| conn.cached(sql));
+        }
 
-        let mut missing: Vec<&str> = sqls
+        let mut missing: Vec<&str> = by_sql
             .iter()
-            .zip(&statements)
             .filter(|(_, statement)| statement.is_none())
             .map(|(sql, _)| *sql)
             .collect();
         missing.sort_unstable();
-        missing.dedup();
         if !missing.is_empty() {
             let prepares = join_in_order(missing.iter().map(|sql| conn.prepare_cached(sql, false)));
             let prepared = match begin.take() {
@@ -408,15 +501,13 @@ impl WriteTxn {
                 let statement = result.map_err(|e| {
                     self.statement_failed(&format!("preparing {}", describe(sql)), e)
                 })?;
-                for (slot, slot_sql) in statements.iter_mut().zip(&sqls) {
-                    if slot_sql == sql {
-                        *slot = Some(statement.clone());
-                    }
-                }
+                by_sql.insert(sql, Some(statement));
             }
         }
 
-        let mut statements = statements.into_iter().map(|s| s.expect("prepared above"));
+        let mut statements = sqls
+            .iter()
+            .map(|sql| by_sql[sql].clone().expect("cached or prepared above"));
         let queued_statements: Vec<tokio_postgres::Statement> =
             statements.by_ref().take(queued.len()).collect();
         let last_statement = statements.next();
@@ -533,8 +624,11 @@ impl WriteTxn {
     /// Queue a write whose result nobody reads. It goes out ahead of the
     /// next statement that is not queued, or of the `COMMIT`, and a failure
     /// surfaces there.
+    ///
+    /// Queued statements are all prepared before the first of them runs, so
+    /// one must not depend on a schema change an earlier queued one makes.
     pub fn execute_later(&self, sql: impl Into<String>, params: Vec<OwnedParam>) {
-        self.queue(sql.into(), params, None);
+        self.queue(Arc::from(sql.into()), params, None);
     }
 
     /// As [`WriteTxn::execute_later`], with `map_error` applied to the
@@ -545,10 +639,19 @@ impl WriteTxn {
         params: Vec<OwnedParam>,
         map_error: MapError,
     ) {
-        self.queue(sql.into(), params, Some(map_error));
+        self.queue(Arc::from(sql.into()), params, Some(map_error));
     }
 
-    fn queue(&self, sql: String, params: Vec<OwnedParam>, map_error: Option<MapError>) {
+    /// As [`WriteTxn::execute_batch`], queued as [`WriteTxn::execute_later`]:
+    /// the statement once per row of parameters.
+    pub fn execute_batch_later(&self, sql: &str, rows: Vec<Vec<OwnedParam>>) {
+        let sql: Arc<str> = Arc::from(sql);
+        for params in rows {
+            self.queue(Arc::clone(&sql), params, None);
+        }
+    }
+
+    fn queue(&self, sql: Arc<str>, params: Vec<OwnedParam>, map_error: Option<MapError>) {
         self.queued.lock().expect("not poisoned").push(Queued {
             sql,
             params,

rust/src/persist/txn/caches.rs

@@ -55,19 +55,18 @@ pub async fn send_invalidation_to_replication(
     keys: &[String],
 ) -> Result<(), WriteError> {
     let stream_id = input.cache_stream.next_id(txn).await?;
-    txn.execute(
+    txn.execute_later(
         "INSERT INTO cache_invalidation_stream_by_instance
             (stream_id, instance_name, cache_func, keys, invalidation_ts)
         VALUES ($1, $2, $3, $4, $5)",
-        &[
-            &stream_id,
-            &input.instance_name,
-            &cache_name,
-            &keys,
-            &input.now_ms()?,
+        crate::owned_params![
+            stream_id,
+            input.instance_name,
+            cache_name,
+            keys,
+            input.now_ms()?
         ],
-    )
-    .await?;
+    );
     effects.add_cache_stream_id(stream_id);
     effects.set_new_replication_data();
     Ok(())

rust/src/persist/txn/current_state.rs

@@ -21,14 +21,13 @@ use std::collections::BTreeSet;
 
 use anyhow::anyhow;
 use serde_json::Value;
-use tokio_postgres::types::ToSql;
 
 use super::{
     caches::{invalidate_cache_and_stream, send_invalidation_to_replication},
     profile, TxnInput,
 };
 use crate::{
-    database::postgres::write::{WriteError, WriteTxn},
+    database::postgres::write::{OwnedParam, WriteError, WriteTxn},
     events::constants::event_type::{M_ROOM_CREATE, M_ROOM_MEMBER},
     persist::{
         effects::{PersistEffects, StreamChangeCache},
@@ -101,15 +100,14 @@ pub async fn update_current_state(
         // room version is recorded first, and the delta stream told what
         // went.
         upsert_room_version(txn, room_id).await?;
-        txn.execute(
+        txn.execute_later(
             "INSERT INTO current_state_delta_stream
                 (stream_id, instance_name, room_id, type, state_key, event_id, prev_event_id)
             SELECT $1, $2, room_id, type, state_key, NULL, event_id
             FROM current_state_events
             WHERE room_id = $3",
-            &[&stream_id, &input.instance_name, &room_id],
-        )
-        .await?;
+            crate::owned_params![stream_id, input.instance_name, room_id],
+        );
 
         let rows = txn
             .query(
@@ -123,73 +121,63 @@ pub async fn update_current_state(
             members.insert(user_id);
         }
 
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM current_state_events WHERE room_id = $1",
-            &[&room_id],
-        )
-        .await?;
-        txn.execute(
+            crate::owned_params![room_id],
+        );
+        txn.execute_later(
             "DELETE FROM sliding_sync_joined_rooms WHERE room_id = $1",
-            &[&room_id],
-        )
-        .await?;
+            crate::owned_params![room_id],
+        );
     } else {
         // The delta stream reads each key's event before it changes.
-        let delta_rows: Vec<Vec<&(dyn ToSql + Sync)>> = changed
+        let delta_rows: Vec<Vec<OwnedParam>> = changed
             .iter()
             .map(|change| {
-                vec![
-                    &stream_id as &(dyn ToSql + Sync),
-                    &input.instance_name,
-                    &room_id,
-                    &change.event_type,
-                    &change.state_key,
-                    &change.event_id,
+                crate::owned_params![
+                    stream_id,
+                    input.instance_name,
+                    room_id,
+                    change.event_type,
+                    change.state_key,
+                    change.event_id,
                 ]
             })
             .collect();
-        txn.execute_batch(
+        txn.execute_batch_later(
             "INSERT INTO current_state_delta_stream
                 (stream_id, instance_name, room_id, type, state_key, event_id, prev_event_id)
             SELECT $1::bigint, $2::text, $3::text, $4::text, $5::text, $6::text, (
                 SELECT event_id FROM current_state_events
                 WHERE room_id = $3::text AND type = $4::text AND state_key = $5::text
             )",
-            &delta_rows,
-        )
-        .await?;
+            delta_rows,
+        );
 
-        let key_rows: Vec<Vec<&(dyn ToSql + Sync)>> = changed
+        let key_rows: Vec<Vec<OwnedParam>> = changed
             .iter()
-            .map(|change| {
-                vec![
-                    &room_id as &(dyn ToSql + Sync),
-                    &change.event_type,
-                    &change.state_key,
-                ]
-            })
+            .map(|change| crate::owned_params![room_id, change.event_type, change.state_key,])
             .collect();
-        txn.execute_batch(
+        txn.execute_batch_later(
             "DELETE FROM current_state_events
             WHERE room_id = $1 AND type = $2 AND state_key = $3",
-            &key_rows,
-        )
-        .await?;
+            key_rows,
+        );
 
         // The membership comes from `room_memberships`, which the members
         // step has filled.
-        let insert_rows: Vec<Vec<&(dyn ToSql + Sync)>> = to_insert
+        let insert_rows: Vec<Vec<OwnedParam>> = to_insert
             .iter()
             .map(|change| {
-                vec![
-                    &room_id as &(dyn ToSql + Sync),
-                    &change.event_type,
-                    &change.state_key,
-                    &change.event_id,
+                crate::owned_params![
+                    room_id,
+                    change.event_type,
+                    change.state_key,
+                    change.event_id,
                 ]
             })
             .collect();
-        txn.execute_batch(
+        txn.execute_batch_later(
             "INSERT INTO current_state_events
                 (room_id, type, state_key, event_id, membership, event_stream_ordering)
             VALUES (
@@ -197,9 +185,8 @@ pub async fn update_current_state(
                 (SELECT membership FROM room_memberships WHERE event_id = $4),
                 (SELECT stream_ordering FROM events WHERE event_id = $4)
             )",
-            &insert_rows,
-        )
-        .await?;
+            insert_rows,
+        );
 
         update_joined_room(txn, room_id, changes).await?;
     }
@@ -217,24 +204,17 @@ pub async fn update_current_state(
             .iter()
             .map(|change| change.state_key)
             .collect();
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM local_current_membership WHERE room_id = $1 AND user_id = ANY($2)",
-            &[&room_id, &user_ids],
-        )
-        .await?;
+            crate::owned_params![room_id, user_ids],
+        );
     }
-    let membership_rows: Vec<Vec<&(dyn ToSql + Sync)>> = local_members
+    let membership_rows: Vec<Vec<OwnedParam>> = local_members
         .iter()
         .filter(|change| change.event_id.is_some())
-        .map(|change| {
-            vec![
-                &room_id as &(dyn ToSql + Sync),
-                &change.state_key,
-                &change.event_id,
-            ]
-        })
+        .map(|change| crate::owned_params![room_id, change.state_key, change.event_id,])
         .collect();
-    txn.execute_batch(
+    txn.execute_batch_later(
         "INSERT INTO local_current_membership
             (room_id, user_id, event_id, membership, event_stream_ordering)
         VALUES (
@@ -242,9 +222,8 @@ pub async fn update_current_state(
             (SELECT membership FROM room_memberships WHERE event_id = $3),
             (SELECT stream_ordering FROM events WHERE event_id = $3)
         )",
-        &membership_rows,
-    )
-    .await?;
+        membership_rows,
+    );
 
     update_membership_snapshots(txn, room_id, changes).await?;
 
@@ -319,13 +298,12 @@ async fn upsert_room_version(txn: &WriteTxn, room_id: &str) -> Result<(), WriteE
     let json: &str = row.try_get(0).map_err(anyhow::Error::new)?;
     let event: Value = serde_json::from_str(json).map_err(anyhow::Error::new)?;
     let (room_version, creator) = room_version_and_creator(&event)?;
-    txn.execute(
+    txn.execute_later(
         "INSERT INTO rooms (room_id, room_version, is_public, creator)
         VALUES ($1, $2, FALSE, $3)
         ON CONFLICT (room_id) DO UPDATE SET room_version = EXCLUDED.room_version",
-        &[&room_id, &room_version, &creator],
-    )
-    .await?;
+        crate::owned_params![room_id, room_version, creator],
+    );
     Ok(())
 }
 
@@ -358,23 +336,27 @@ fn room_version_and_creator(event: &Value) -> anyhow::Result<(&str, Option<&str>
 
 /// The columns of a [`SlidingSyncStateValues`] that are set, as Python's
 /// `SlidingSyncStateInsertValues` dict holds them.
-fn state_value_columns(
-    values: &SlidingSyncStateValues,
-) -> Vec<(&'static str, Box<dyn ToSql + Sync + Send + '_>)> {
-    let mut columns: Vec<(&'static str, Box<dyn ToSql + Sync + Send + '_>)> = Vec::new();
+fn state_value_columns(values: &SlidingSyncStateValues) -> Vec<(&'static str, OwnedParam)> {
+    let mut columns: Vec<(&'static str, OwnedParam)> = Vec::new();
     if let Some(room_type) = &values.room_type {
-        columns.push(("room_type", Box::new(room_type.as_deref())));
+        columns.push((
+            "room_type",
+            Box::new(room_type.as_deref().map(str::to_owned)),
+        ));
     }
     if let Some(is_encrypted) = values.is_encrypted {
         columns.push(("is_encrypted", Box::new(is_encrypted)));
     }
     if let Some(room_name) = &values.room_name {
-        columns.push(("room_name", Box::new(room_name.as_deref())));
+        columns.push((
+            "room_name",
+            Box::new(room_name.as_deref().map(str::to_owned)),
+        ));
     }
     if let Some(successor) = &values.tombstone_successor_room_id {
         columns.push((
             "tombstone_successor_room_id",
-            Box::new(successor.as_deref()),
+            Box::new(successor.as_deref().map(str::to_owned)),
         ));
     }
     columns
@@ -416,14 +398,9 @@ async fn update_joined_room(
         placeholders.join(", "),
         updates.join(", ")
     );
-    let mut params: Vec<&(dyn ToSql + Sync)> =
-        vec![&room_id, &changes.joined_room_bump_stamp_to_fully_insert];
-    params.extend(
-        columns
-            .iter()
-            .map(|(_, value)| &**value as &(dyn ToSql + Sync)),
-    );
-    txn.execute(&sql, &params).await?;
+    let mut params = crate::owned_params![room_id, changes.joined_room_bump_stamp_to_fully_insert];
+    params.extend(columns.into_iter().map(|(_, value)| value));
+    txn.execute_later(sql, params);
     Ok(())
 }
 
@@ -441,12 +418,11 @@ async fn update_membership_snapshots(
             .iter()
             .map(|user_id| &**user_id)
             .collect();
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM sliding_sync_membership_snapshots
             WHERE room_id = $1 AND user_id = ANY($2)",
-            &[&room_id, &user_ids],
-        )
-        .await?;
+            crate::owned_params![room_id, user_ids],
+        );
     }
 
     if changes.to_insert_membership_snapshots.is_empty() {
@@ -487,33 +463,28 @@ async fn update_membership_snapshots(
     );
     // A new membership is not forgotten.
     let forgotten = 0i32;
-    // As `&str`s, which the rows borrow: an `Arc<str>` is not `ToSql`.
-    let snapshots: Vec<[&str; 4]> = changes
+    let rows: Vec<Vec<OwnedParam>> = changes
         .to_insert_membership_snapshots
         .iter()
         .map(|info| {
-            [
-                &*info.user_id,
-                &*info.sender,
-                &*info.membership_event_id,
-                &*info.membership,
-            ]
-        })
-        .collect();
-    let rows: Vec<Vec<&(dyn ToSql + Sync)>> = snapshots
-        .iter()
-        .map(|[user_id, sender, event_id, membership]| {
-            let mut row: Vec<&(dyn ToSql + Sync)> =
-                vec![&room_id, user_id, sender, event_id, membership, &forgotten];
-            row.extend(
-                columns
-                    .iter()
-                    .map(|(_, value)| &**value as &(dyn ToSql + Sync)),
-            );
+            let mut row = crate::owned_params![
+                room_id,
+                *info.user_id,
+                *info.sender,
+                *info.membership_event_id,
+                *info.membership,
+                forgotten,
+            ];
+            // Each row owns its copy of the shared values.
+            let mut shared_columns = state_value_columns(&shared.state);
+            if let Some(has_known_state) = shared.has_known_state {
+                shared_columns.push(("has_known_state", Box::new(has_known_state)));
+            }
+            row.extend(shared_columns.into_iter().map(|(_, value)| value));
             row
         })
         .collect();
-    txn.execute_batch(&sql, &rows).await?;
+    txn.execute_batch_later(&sql, rows);
     Ok(())
 }
 
@@ -548,11 +519,10 @@ async fn handle_potentially_left_users(
             continue;
         }
         // `mark_remote_user_device_list_as_unsubscribed_txn`.
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM device_lists_remote_extremeties WHERE user_id = $1",
-            &[&user_id],
-        )
-        .await?;
+            crate::owned_params![user_id],
+        );
         invalidate_cache_and_stream(
             txn,
             input,

rust/src/persist/txn/edges.rs

@@ -23,7 +23,7 @@ use anyhow::anyhow;
 
 use super::Persisting;
 use crate::{
-    database::postgres::write::{Column, WriteError, WriteTxn},
+    database::postgres::write::{OwnedColumn, WriteError, WriteTxn},
     events::{constants::event_type, Event},
 };
 
@@ -35,35 +35,32 @@ pub async fn update_forward_extremities(
     new_forward_extremities: &HashSet<Arc<str>>,
     max_stream_order: i64,
 ) -> Result<(), WriteError> {
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM event_forward_extremities WHERE room_id = $1",
-        &[&room_id],
-    )
-    .await?;
+        crate::owned_params![room_id],
+    );
 
     let event_ids: Vec<&str> = new_forward_extremities.iter().map(|id| &**id).collect();
     let room_ids = vec![room_id; event_ids.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "event_forward_extremities",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     let stream_orderings = vec![max_stream_order; event_ids.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "stream_ordering_to_exterm",
-        &[
-            Column::new("room_id", "text", &room_ids),
-            Column::new("event_id", "text", &event_ids),
-            Column::new("stream_ordering", "bigint", &stream_orderings),
+        vec![
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("stream_ordering", "bigint", &stream_orderings),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -74,22 +71,20 @@ pub async fn set_state_dag_extremities(
     room_id: &str,
     new_extremities: &HashSet<Arc<str>>,
 ) -> Result<(), WriteError> {
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM msc4242_state_dag_forward_extremities WHERE room_id = $1",
-        &[&room_id],
-    )
-    .await?;
+        crate::owned_params![room_id],
+    );
     let event_ids: Vec<&str> = new_extremities.iter().map(|id| &**id).collect();
     let room_ids = vec![room_id; event_ids.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "msc4242_state_dag_forward_extremities",
-        &[
-            Column::new("room_id", "text", &room_ids),
-            Column::new("event_id", "text", &event_ids),
+        vec![
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("event_id", "text", &event_ids),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -144,15 +139,14 @@ pub async fn update_backward_extremities(
     if !potential.is_empty() {
         let event_ids: Vec<&str> = potential.iter().copied().collect();
         let room_ids = vec![room_id; event_ids.len()];
-        txn.upsert_many(
+        txn.upsert_many_later(
             "event_backward_extremities",
-            &[
-                Column::new("room_id", "text", &room_ids),
-                Column::new("event_id", "text", &event_ids),
+            vec![
+                OwnedColumn::copy("room_id", "text", &room_ids),
+                OwnedColumn::copy("event_id", "text", &event_ids),
             ],
-            &[],
-        )
-        .await?;
+            vec![],
+        )?;
 
         // Record the stream orderings where we have new gaps.
         let gap_orderings: Vec<i64> = events
@@ -163,16 +157,15 @@ pub async fn update_backward_extremities(
             .collect();
         let room_ids = vec![room_id; gap_orderings.len()];
         let instance_names = vec![instance_name; gap_orderings.len()];
-        txn.insert_many(
+        txn.insert_many_later(
             "timeline_gaps",
-            &[
-                Column::new("room_id", "text", &room_ids),
-                Column::new("instance_name", "text", &instance_names),
-                Column::new("stream_ordering", "bigint", &gap_orderings),
+            vec![
+                OwnedColumn::copy("room_id", "text", &room_ids),
+                OwnedColumn::copy("instance_name", "text", &instance_names),
+                OwnedColumn::copy("stream_ordering", "bigint", &gap_orderings),
             ],
             None,
-        )
-        .await?;
+        )?;
     }
 
     // An event with no prev events will never have anything to backfill
@@ -184,16 +177,14 @@ pub async fn update_backward_extremities(
         .map(|(e, _)| e.event.event_id())
         .collect();
     if !to_remove.is_empty() {
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM event_backward_extremities WHERE event_id = ANY($1) AND room_id = $2",
-            &[&to_remove, &room_id],
-        )
-        .await?;
-        txn.execute(
+            crate::owned_params![to_remove, room_id],
+        );
+        txn.execute_later(
             "DELETE FROM event_failed_pull_attempts WHERE event_id = ANY($1) AND room_id = $2",
-            &[&to_remove, &room_id],
-        )
-        .await?;
+            crate::owned_params![to_remove, room_id],
+        );
     }
     Ok(())
 }
@@ -220,15 +211,14 @@ pub async fn handle_mult_prev_events(
             stream_ordering: persisting.stream_ordering(),
         });
     }
-    txn.insert_many(
+    txn.insert_many_later(
         "event_edges",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("prev_event_id", "text", &prev_event_ids),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("prev_event_id", "text", &prev_event_ids),
         ],
         None,
-    )
-    .await?;
+    )?;
     update_backward_extremities(txn, instance_name, &backward).await
 }
 
@@ -271,15 +261,14 @@ pub async fn store_state_dag_edges(
     // Python inserts the create event's row and upserts the rest. Only the
     // create event has a NULL edge, and the key of a NULL never conflicts,
     // so one upsert does both.
-    txn.upsert_many(
+    txn.upsert_many_later(
         "msc4242_state_dag_edges",
-        &[
-            Column::new("room_id", "text", &room_ids),
-            Column::new("event_id", "text", &event_ids),
-            Column::new("prev_state_event_id", "text", &prev_state_event_ids),
+        vec![
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("prev_state_event_id", "text", &prev_state_event_ids),
         ],
-        &[],
-    )
-    .await?;
+        vec![],
+    )?;
     Ok(())
 }

rust/src/persist/txn/events.rs

@@ -31,7 +31,7 @@ use super::{
     PartialStateConflict, Persisting, TxnInput,
 };
 use crate::{
-    database::postgres::write::{Column, WriteError, WriteTxn},
+    database::postgres::write::{Column, OwnedColumn, WriteError, WriteTxn},
     event_auth::get_domain_from_id,
     events::{constants::event_type, redact_event_py},
     persist::{
@@ -83,12 +83,11 @@ pub async fn update_room_depths(
     if min_depth.is_some_and(|min_depth| depth_update >= min_depth) {
         return Ok(());
     }
-    txn.upsert_many(
+    txn.upsert_many_later(
         "room_depth",
-        &[Column::new("room_id", "text", &vec![room_id])],
-        &[Column::new("min_depth", "bigint", &vec![depth_update])],
-    )
-    .await?;
+        vec![OwnedColumn::copy("room_id", "text", &[room_id])],
+        vec![OwnedColumn::copy("min_depth", "bigint", &[depth_update])],
+    )?;
     Ok(())
 }
 
@@ -145,26 +144,24 @@ pub async fn update_outliers<'a>(
         );
         store_state_mappings(txn, std::slice::from_ref(&event), effects).await?;
 
-        txn.insert_many(
+        txn.insert_many_later(
             "ex_outlier_stream",
-            &[
-                Column::new(
+            vec![
+                OwnedColumn::copy(
                     "event_stream_ordering",
                     "bigint",
-                    &vec![event.stream_ordering()],
+                    &[event.stream_ordering()],
                 ),
-                Column::new("event_id", "text", &vec![event.event().event_id()]),
-                Column::new("state_group", "bigint", &vec![event.context().state_group]),
-                Column::new("instance_name", "text", &vec![input.instance_name]),
+                OwnedColumn::copy("event_id", "text", &[event.event().event_id()]),
+                OwnedColumn::copy("state_group", "bigint", &[event.context().state_group]),
+                OwnedColumn::copy("instance_name", "text", &[input.instance_name]),
             ],
             None,
-        )
-        .await?;
-        txn.execute(
+        )?;
+        txn.execute_later(
             "UPDATE events SET outlier = FALSE WHERE event_id = $1",
-            &[&event.event().event_id()],
-        )
-        .await?;
+            crate::owned_params![event.event().event_id()],
+        );
         update_backward_extremities(
             txn,
             input.instance_name,
@@ -395,13 +392,12 @@ pub async fn delete_sticky_events(
     let event_ids: Vec<&str> = event_ids.iter().map(String::as_str).collect();
     // `sticky_events` has no index on `event_id`, but has one on the
     // stream ordering.
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM sticky_events WHERE event_stream_ordering IN (
             SELECT stream_ordering FROM events WHERE event_id = ANY($1)
         )",
-        &[&event_ids],
-    )
-    .await?;
+        crate::owned_params![event_ids],
+    );
     Ok(())
 }
 
@@ -471,50 +467,47 @@ pub async fn store_events(
         }
     }
 
-    txn.insert_many(
+    txn.insert_many_later(
         "event_json",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("internal_metadata", "text", &internal_metadata),
-            Column::new("json", "text", &jsons),
-            Column::new("format_version", "bigint", &format_versions),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("internal_metadata", "text", &internal_metadata),
+            OwnedColumn::copy("json", "text", &jsons),
+            OwnedColumn::copy("format_version", "bigint", &format_versions),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     let processed = vec![true; n];
-    txn.insert_many(
+    txn.insert_many_later(
         "events",
-        &[
-            Column::new("instance_name", "text", &instance_names),
-            Column::new("stream_ordering", "bigint", &stream_orderings),
-            Column::new("topological_ordering", "bigint", &depths),
-            Column::new("depth", "bigint", &depths),
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("type", "text", &types),
-            Column::new("processed", "boolean", &processed),
-            Column::new("outlier", "boolean", &outliers),
-            Column::new("origin_server_ts", "bigint", &origin_server_tss),
-            Column::new("received_ts", "bigint", &received_tss),
-            Column::new("sender", "text", &senders),
-            Column::new("contains_url", "boolean", &contains_urls),
-            Column::new("state_key", "text", &state_keys),
-            Column::new("rejection_reason", "text", &rejection_reasons),
+        vec![
+            OwnedColumn::copy("instance_name", "text", &instance_names),
+            OwnedColumn::copy("stream_ordering", "bigint", &stream_orderings),
+            OwnedColumn::copy("topological_ordering", "bigint", &depths),
+            OwnedColumn::copy("depth", "bigint", &depths),
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("type", "text", &types),
+            OwnedColumn::copy("processed", "boolean", &processed),
+            OwnedColumn::copy("outlier", "boolean", &outliers),
+            OwnedColumn::copy("origin_server_ts", "bigint", &origin_server_tss),
+            OwnedColumn::copy("received_ts", "bigint", &received_tss),
+            OwnedColumn::copy("sender", "text", &senders),
+            OwnedColumn::copy("contains_url", "boolean", &contains_urls),
+            OwnedColumn::copy("state_key", "text", &state_keys),
+            OwnedColumn::copy("rejection_reason", "text", &rejection_reasons),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     // A redaction of an event we now have unredacted needs to censor it.
     if !unredacted.is_empty() {
-        txn.execute(
+        txn.execute_later(
             "UPDATE redactions SET have_censored = FALSE WHERE redacts = ANY($1)",
-            &[&unredacted],
-        )
-        .await?;
+            crate::owned_params![unredacted],
+        );
     }
 
     let state: Vec<_> = events.iter().filter(|e| e.event().is_state()).collect();
@@ -522,17 +515,16 @@ pub async fn store_events(
     let room_ids: Vec<&str> = state.iter().map(|e| e.event().room_id()).collect();
     let types: Vec<&str> = state.iter().map(|e| e.event().type_()).collect();
     let state_keys: Vec<Option<&str>> = state.iter().map(|e| e.event().state_key()).collect();
-    txn.insert_many(
+    txn.insert_many_later(
         "state_events",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("type", "text", &types),
-            Column::new("state_key", "text", &state_keys),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("type", "text", &types),
+            OwnedColumn::copy("state_key", "text", &state_keys),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -565,19 +557,18 @@ pub async fn persist_transaction_ids(
     let device_ids: Vec<&str> = rows.iter().map(|(_, d, _)| d.as_str()).collect();
     let txn_ids: Vec<&str> = rows.iter().map(|(_, _, t)| t.as_str()).collect();
     let inserted_tss = vec![inserted_ts; rows.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "event_txn_id_device_id",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("user_id", "text", &user_ids),
-            Column::new("device_id", "text", &device_ids),
-            Column::new("txn_id", "text", &txn_ids),
-            Column::new("inserted_ts", "bigint", &inserted_tss),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("user_id", "text", &user_ids),
+            OwnedColumn::copy("device_id", "text", &device_ids),
+            OwnedColumn::copy("txn_id", "text", &txn_ids),
+            OwnedColumn::copy("inserted_ts", "bigint", &inserted_tss),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -631,6 +622,11 @@ pub async fn store_state_mappings(
         .collect();
     let event_ids: Vec<&str> = partial.iter().map(|e| e.event().event_id()).collect();
     let room_ids: Vec<&str> = partial.iter().map(|e| e.event().room_id()).collect();
+    if !partial.is_empty() {
+        // Send the queued writes first, so that their failures are not taken
+        // for this insert's constraint violation below.
+        txn.flush().await?;
+    }
     let inserted = txn
         .insert_many(
             "partial_state_events",
@@ -653,16 +649,16 @@ pub async fn store_state_mappings(
 
     let event_ids: Vec<&str> = state_groups.iter().map(|(id, _)| *id).collect();
     let groups: Vec<Option<i64>> = state_groups.iter().map(|(_, group)| *group).collect();
-    txn.upsert_many(
+    txn.upsert_many_later(
         "event_to_state_groups",
-        &[Column::new("event_id", "text", &event_ids)],
-        &[Column::new("state_group", "bigint", &groups)],
-    )
-    .await?;
+        vec![OwnedColumn::copy("event_id", "text", &event_ids)],
+        vec![OwnedColumn::copy("state_group", "bigint", &groups)],
+    )?;
 
     for (event_id, group) in state_groups {
-        // Python would prefill a missing group too, but the insert above
-        // has failed by then.
+        // Python would prefill a missing group too, but the queued insert
+        // above fails when it is sent, and the effects go with the
+        // transaction.
         if let Some(group) = group {
             effects.add_state_group_prefill(event_id.to_owned(), group);
         }
@@ -686,21 +682,19 @@ pub async fn persist_auth_chain(
             .values()
             .map(|l| l.sequence_number)
             .collect();
-        txn.insert_many(
+        txn.insert_many_later(
             "event_auth_chains",
-            &[
-                Column::new("event_id", "text", &event_ids),
-                Column::new("chain_id", "bigint", &chain_ids),
-                Column::new("sequence_number", "bigint", &sequence_numbers),
+            vec![
+                OwnedColumn::copy("event_id", "text", &event_ids),
+                OwnedColumn::copy("chain_id", "bigint", &chain_ids),
+                OwnedColumn::copy("sequence_number", "bigint", &sequence_numbers),
             ],
             None,
-        )
-        .await?;
-        txn.execute(
+        )?;
+        txn.execute_later(
             "DELETE FROM event_auth_chain_to_calculate WHERE event_id = ANY($1)",
-            &[&event_ids],
-        )
-        .await?;
+            crate::owned_params![event_ids],
+        );
 
         let (mut origin_chains, mut origin_seqs) = (Vec::new(), Vec::new());
         let (mut target_chains, mut target_seqs) = (Vec::new(), Vec::new());
@@ -712,17 +706,16 @@ pub async fn persist_auth_chain(
                 target_seqs.push(seq);
             }
         }
-        txn.insert_many(
+        txn.insert_many_later(
             "event_auth_chain_links",
-            &[
-                Column::new("origin_chain_id", "bigint", &origin_chains),
-                Column::new("origin_sequence_number", "bigint", &origin_seqs),
-                Column::new("target_chain_id", "bigint", &target_chains),
-                Column::new("target_sequence_number", "bigint", &target_seqs),
+            vec![
+                OwnedColumn::copy("origin_chain_id", "bigint", &origin_chains),
+                OwnedColumn::copy("origin_sequence_number", "bigint", &origin_seqs),
+                OwnedColumn::copy("target_chain_id", "bigint", &target_chains),
+                OwnedColumn::copy("target_sequence_number", "bigint", &target_seqs),
             ],
             None,
-        )
-        .await?;
+        )?;
     }
 
     let mut event_ids = Vec::new();
@@ -743,16 +736,15 @@ pub async fn persist_auth_chain(
             auth_ids.push(auth_id);
         }
     }
-    txn.insert_many(
+    txn.insert_many_later(
         "event_auth",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("auth_id", "text", &auth_ids),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("auth_id", "text", &auth_ids),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -775,15 +767,14 @@ pub async fn store_rejections<'a>(
         .collect();
     // `last_check` is a text column.
     let last_checks = vec![input.now_ms()?.to_string(); rejected.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "rejections",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("reason", "text", &reasons),
-            Column::new("last_check", "text", &last_checks),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("reason", "text", &reasons),
+            OwnedColumn::copy("last_check", "text", &last_checks),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(left)
 }

rust/src/persist/txn/metadata.rs

@@ -31,7 +31,7 @@ use crate::events::constants::event_type::{
     M_ROOM_TOPIC as TOPIC,
 };
 use crate::{
-    database::postgres::write::{Column, WriteError, WriteTxn},
+    database::postgres::write::{OwnedColumn, OwnedParam, WriteError, WriteTxn},
     events::{
         constants::{event_type, relation_type},
         json_object::JsonObject,
@@ -66,11 +66,10 @@ pub async fn update_metadata_tables(
         let Some(redacted) = redacts(event) else {
             continue;
         };
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM event_push_actions WHERE room_id = $1 AND event_id = $2",
-            &[&event.room_id(), &redacted],
-        )
-        .await?;
+            crate::owned_params![event.room_id(), redacted],
+        );
         handle_redact_relations(txn, input, effects, redacted).await?;
     }
 
@@ -146,7 +145,7 @@ async fn set_push_actions(
         }
     }
     if !event_ids.is_empty() {
-        txn.execute(
+        txn.execute_later(
             "INSERT INTO event_push_actions (
                 room_id, event_id, user_id, actions, stream_ordering,
                 topological_ordering, notif, highlight, unread, thread_id
@@ -156,9 +155,8 @@ async fn set_push_actions(
             FROM unnest($2::text[], $3::bigint[], $4::bigint[])
                 AS e(event_id, stream_ordering, depth)
             INNER JOIN event_push_actions_staging AS s USING (event_id)",
-            &[&input.room_id, &event_ids, &stream_orderings, &depths],
-        )
-        .await?;
+            crate::owned_params![input.room_id, event_ids, stream_orderings, depths],
+        );
     }
 
     let mut staged = Vec::new();
@@ -173,11 +171,10 @@ async fn set_push_actions(
         }
     }
     if !staged.is_empty() {
-        txn.execute(
+        txn.execute_later(
             "DELETE FROM event_push_actions_staging WHERE event_id = ANY($1)",
-            &[&staged],
-        )
-        .await?;
+            crate::owned_params![staged],
+        );
     }
     Ok(())
 }
@@ -203,11 +200,10 @@ async fn handle_redact_relations(
     };
     let relates_to: String = row.try_get(0).map_err(anyhow::Error::new)?;
     let rel_type: String = row.try_get(1).map_err(anyhow::Error::new)?;
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM event_relations WHERE event_id = $1",
-        &[&redacted_event_id],
-    )
-    .await?;
+        crate::owned_params![redacted_event_id],
+    );
 
     // Any relation information for the related event must be cleared.
     let mut invalidations = vec![(
@@ -252,7 +248,7 @@ async fn handle_redact_relations(
         let event_id: String = latest.try_get(0).map_err(anyhow::Error::new)?;
         let topological_ordering: i64 = latest.try_get(1).map_err(anyhow::Error::new)?;
         let stream_ordering: i64 = latest.try_get(2).map_err(anyhow::Error::new)?;
-        txn.execute(
+        txn.execute_later(
             "INSERT INTO threads
                 (room_id, thread_id, latest_event_id, topological_ordering, stream_ordering)
             VALUES ($1, $2, $3, $4, $5)
@@ -260,15 +256,14 @@ async fn handle_redact_relations(
                 latest_event_id = EXCLUDED.latest_event_id,
                 topological_ordering = EXCLUDED.topological_ordering,
                 stream_ordering = EXCLUDED.stream_ordering",
-            &[
-                &room_id,
-                &relates_to,
-                &event_id,
-                &topological_ordering,
-                &stream_ordering,
+            crate::owned_params![
+                room_id,
+                relates_to,
+                event_id,
+                topological_ordering,
+                stream_ordering
             ],
-        )
-        .await?;
+        );
     } else {
         // The thread no longer exists. As in Python's
         // `simple_delete_one_txn`, by thread ID alone, and exactly one row.
@@ -315,23 +310,22 @@ async fn store_search(
     if event_ids.is_empty() {
         return Ok(());
     }
-    txn.execute(
+    txn.execute_later(
         "INSERT INTO event_search
             (event_id, room_id, key, vector, stream_ordering, origin_server_ts)
         SELECT event_id, room_id, key, to_tsvector('english', value),
             stream_ordering, origin_server_ts
         FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::bigint[], $6::bigint[])
             AS t(event_id, room_id, key, value, stream_ordering, origin_server_ts)",
-        &[
-            &event_ids,
-            &room_ids,
-            &keys,
-            &values,
-            &stream_orderings,
-            &origin_server_tss,
+        crate::owned_params![
+            event_ids,
+            room_ids,
+            keys,
+            values,
+            stream_orderings,
+            origin_server_tss
         ],
-    )
-    .await?;
+    );
     Ok(())
 }
 
@@ -416,16 +410,15 @@ async fn store_redactions(
         return Ok(());
     }
     let received_tss = vec![input.now_ms()?; event_ids.len()];
-    txn.upsert_many(
+    txn.upsert_many_later(
         "redactions",
-        &[Column::new("event_id", "text", &event_ids)],
-        &[
-            Column::new("redacts", "text", &redacted),
-            Column::new("received_ts", "bigint", &received_tss),
-            Column::new("recheck", "boolean", &rechecks),
+        vec![OwnedColumn::copy("event_id", "text", &event_ids)],
+        vec![
+            OwnedColumn::copy("redacts", "text", &redacted),
+            OwnedColumn::copy("received_ts", "bigint", &received_tss),
+            OwnedColumn::copy("recheck", "boolean", &rechecks),
         ],
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -453,17 +446,16 @@ async fn store_retention_policy(
     ) else {
         return Ok(());
     };
-    txn.execute(
+    txn.execute_later(
         "INSERT INTO room_retention (room_id, event_id, min_lifetime, max_lifetime)
         VALUES ($1, $2, $3, $4)",
-        &[
-            &event.room_id(),
-            &event.event_id(),
-            &min_lifetime,
-            &max_lifetime,
+        crate::owned_params![
+            event.room_id(),
+            event.event_id(),
+            min_lifetime,
+            max_lifetime
         ],
-    )
-    .await?;
+    );
     invalidate_cache_and_stream(
         txn,
         input,
@@ -519,34 +511,27 @@ async fn store_relations(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<()
             ));
         }
     }
-    txn.insert_many(
+    txn.insert_many_later(
         "event_relations",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("relates_to_id", "text", &parent_ids),
-            Column::new("relation_type", "text", &rel_types),
-            Column::new("aggregation_key", "text", &aggregation_keys),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("relates_to_id", "text", &parent_ids),
+            OwnedColumn::copy("relation_type", "text", &rel_types),
+            OwnedColumn::copy("aggregation_key", "text", &aggregation_keys),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     // Only a later event becomes the latest. The stream ordering is always
     // later, unless the event was backfilled. In order, as each reply
     // compares with the one before.
-    let thread_rows: Vec<Vec<&(dyn ToSql + Sync)>> = replies
+    let thread_rows: Vec<Vec<OwnedParam>> = replies
         .iter()
         .map(|(room_id, thread_id, event_id, depth, stream_ordering)| {
-            vec![
-                room_id as &(dyn ToSql + Sync),
-                thread_id,
-                event_id,
-                depth,
-                stream_ordering,
-            ]
+            crate::owned_params![room_id, thread_id, event_id, depth, stream_ordering]
         })
         .collect();
-    txn.execute_batch(
+    txn.execute_batch_later(
         "INSERT INTO threads
             (room_id, thread_id, latest_event_id, topological_ordering, stream_ordering)
         VALUES ($1, $2, $3, $4, $5)
@@ -557,9 +542,8 @@ async fn store_relations(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<()
         WHERE
             threads.topological_ordering <= EXCLUDED.topological_ordering AND
             threads.stream_ordering < EXCLUDED.stream_ordering",
-        &thread_rows,
-    )
-    .await?;
+        thread_rows,
+    );
     Ok(())
 }
 
@@ -586,17 +570,16 @@ async fn store_labels(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<(), W
             depths.push(event.depth());
         }
     }
-    txn.insert_many(
+    txn.insert_many_later(
         "event_labels",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("label", "text", &labels),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("topological_ordering", "bigint", &depths),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("label", "text", &labels),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("topological_ordering", "bigint", &depths),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -622,15 +605,14 @@ async fn store_expiry(txn: &WriteTxn, events: &[Persisting<'_>]) -> Result<(), W
         event_ids.push(event.event_id());
         expiry_tss.push(expiry_ts);
     }
-    txn.insert_many(
+    txn.insert_many_later(
         "event_expiry",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("expiry_ts", "bigint", &expiry_tss),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("expiry_ts", "bigint", &expiry_tss),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -678,21 +660,20 @@ async fn store_room_members(
         .iter()
         .map(|p| non_null_str(p.event().content().get_field("avatar_url")))
         .collect();
-    txn.insert_many(
+    txn.insert_many_later(
         "room_memberships",
-        &[
-            Column::new("event_id", "text", &event_ids),
-            Column::new("event_stream_ordering", "bigint", &stream_orderings),
-            Column::new("user_id", "text", &user_ids),
-            Column::new("sender", "text", &senders),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("membership", "text", &memberships),
-            Column::new("display_name", "text", &display_names),
-            Column::new("avatar_url", "text", &avatar_urls),
+        vec![
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("event_stream_ordering", "bigint", &stream_orderings),
+            OwnedColumn::copy("user_id", "text", &user_ids),
+            OwnedColumn::copy("sender", "text", &senders),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("membership", "text", &memberships),
+            OwnedColumn::copy("display_name", "text", &display_names),
+            OwnedColumn::copy("avatar_url", "text", &avatar_urls),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     if input.inhibit_local_membership_updates {
         return Ok(());
@@ -726,7 +707,7 @@ async fn store_room_members(
             }
         };
 
-        txn.execute(
+        txn.execute_later(
             "INSERT INTO local_current_membership
                 (room_id, user_id, event_id, event_stream_ordering, membership)
             VALUES ($1, $2, $3, $4, $5)
@@ -734,15 +715,14 @@ async fn store_room_members(
                 event_id = EXCLUDED.event_id,
                 event_stream_ordering = EXCLUDED.event_stream_ordering,
                 membership = EXCLUDED.membership",
-            &[
-                &event.room_id(),
-                &user_id,
-                &event.event_id(),
-                &persisting.stream_ordering(),
-                &membership,
+            crate::owned_params![
+                event.room_id(),
+                user_id,
+                event.event_id(),
+                persisting.stream_ordering(),
+                membership
             ],
-        )
-        .await?;
+        );
 
         let snapshot_state = stripped_state
             .map(|state| snapshot_values_from_stripped_state(state.as_deref()))

rust/src/persist/txn/profile.rs

@@ -22,7 +22,7 @@ use std::collections::BTreeSet;
 
 use super::TxnInput;
 use crate::{
-    database::postgres::write::{Column, WriteError, WriteTxn},
+    database::postgres::write::{OwnedColumn, WriteError, WriteTxn},
     persist::{
         effects::PersistEffects, extremities::is_mine_id, sliding_sync::SlidingSyncTableChanges,
     },
@@ -129,23 +129,21 @@ async fn record_profile_updates_for_user_left_room(
     }
     let no_longer_sharing_ids: Vec<&str> = no_longer_sharing.iter().map(String::as_str).collect();
 
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM profile_updates_per_user
         WHERE user_id = ANY($1)
             AND stream_id IN (SELECT stream_id FROM profile_updates WHERE user_id = $2)",
-        &[&no_longer_sharing_ids, &user_id],
-    )
-    .await?;
+        crate::owned_params![no_longer_sharing_ids, user_id],
+    );
     let user = BTreeSet::from([user_id]);
     record_profile_updates(txn, input, &user, LEFT_ROOM, &no_longer_sharing, effects).await?;
 
-    txn.execute(
+    txn.execute_later(
         "DELETE FROM profile_updates_per_user
         WHERE user_id = $1
             AND stream_id IN (SELECT stream_id FROM profile_updates WHERE user_id = ANY($2))",
-        &[&user_id, &no_longer_sharing_ids],
-    )
-    .await?;
+        crate::owned_params![user_id, no_longer_sharing_ids],
+    );
     let no_longer_sharing: BTreeSet<&str> = no_longer_sharing_ids.into_iter().collect();
     let target = BTreeSet::from([user_id.to_owned()]);
     record_profile_updates(txn, input, &no_longer_sharing, LEFT_ROOM, &target, effects).await
@@ -184,18 +182,17 @@ async fn record_profile_updates(
     let actions = vec![action; updated.len()];
     // `affected_fields` is left NULL: it is set only for profile changes.
     let inserted = vec![inserted_ts; updated.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "profile_updates",
-        &[
-            Column::new("stream_id", "bigint", &stream_ids),
-            Column::new("instance_name", "text", &instance_names),
-            Column::new("user_id", "text", &updated),
-            Column::new("action", "text", &actions),
-            Column::new("inserted_ts", "bigint", &inserted),
+        vec![
+            OwnedColumn::copy("stream_id", "bigint", &stream_ids),
+            OwnedColumn::copy("instance_name", "text", &instance_names),
+            OwnedColumn::copy("user_id", "text", &updated),
+            OwnedColumn::copy("action", "text", &actions),
+            OwnedColumn::copy("inserted_ts", "bigint", &inserted),
         ],
         None,
-    )
-    .await?;
+    )?;
 
     let mut per_user_stream_ids = Vec::new();
     let mut per_user_ids = Vec::new();
@@ -208,15 +205,14 @@ async fn record_profile_updates(
         }
     }
     let inserted = vec![inserted_ts; per_user_ids.len()];
-    txn.insert_many(
+    txn.insert_many_later(
         "profile_updates_per_user",
-        &[
-            Column::new("stream_id", "bigint", &per_user_stream_ids),
-            Column::new("user_id", "text", &per_user_ids),
-            Column::new("inserted_ts", "bigint", &inserted),
+        vec![
+            OwnedColumn::copy("stream_id", "bigint", &per_user_stream_ids),
+            OwnedColumn::copy("user_id", "text", &per_user_ids),
+            OwnedColumn::copy("inserted_ts", "bigint", &inserted),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }

rust/src/persist/txn/sliding_sync.rs

@@ -48,7 +48,7 @@ pub async fn update_sliding_sync_joined_room(
         .find(|e| DEFAULT_BUMP_EVENT_TYPES.contains(&e.event().type_()))
         .map(Persisting::stream_ordering);
 
-    txn.execute(
+    txn.execute_later(
         "UPDATE sliding_sync_joined_rooms
         SET
             event_stream_ordering = CASE
@@ -62,8 +62,7 @@ pub async fn update_sliding_sync_joined_room(
                 ELSE bump_stamp
             END
         WHERE room_id = $3",
-        &[&max_stream_ordering, &max_bump_stamp, &room_id],
-    )
-    .await?;
+        crate::owned_params![max_stream_ordering, max_bump_stamp, room_id],
+    );
     Ok(())
 }

rust/src/persist/txn/sticky.rs

@@ -19,11 +19,10 @@
 //! redacted events is in [`super::events`].
 
 use log::info;
-use tokio_postgres::types::ToSql;
 
 use super::{metadata_error, TxnInput};
 use crate::{
-    database::postgres::write::{Column, WriteError, WriteTxn},
+    database::postgres::write::{OwnedColumn, OwnedParam, WriteError, WriteTxn},
     events::Event,
     persist::effects::PersistEffects,
 };
@@ -80,20 +79,19 @@ pub async fn insert_sticky_events(
     let expiries: Vec<i64> = sticky.iter().map(|&(_, _, at)| at).collect();
     // `stream_id` and `event_stream_ordering` are `INTEGER`, which a
     // `bigint` casts to on assignment.
-    txn.insert_many(
+    txn.insert_many_later(
         "sticky_events",
-        &[
-            Column::new("instance_name", "text", &instance_names),
-            Column::new("stream_id", "bigint", &stream_ids),
-            Column::new("room_id", "text", &room_ids),
-            Column::new("event_id", "text", &event_ids),
-            Column::new("event_stream_ordering", "bigint", &stream_orderings),
-            Column::new("sender", "text", &senders),
-            Column::new("expires_at", "bigint", &expiries),
+        vec![
+            OwnedColumn::copy("instance_name", "text", &instance_names),
+            OwnedColumn::copy("stream_id", "bigint", &stream_ids),
+            OwnedColumn::copy("room_id", "text", &room_ids),
+            OwnedColumn::copy("event_id", "text", &event_ids),
+            OwnedColumn::copy("event_stream_ordering", "bigint", &stream_orderings),
+            OwnedColumn::copy("sender", "text", &senders),
+            OwnedColumn::copy("expires_at", "bigint", &expiries),
         ],
         None,
-    )
-    .await?;
+    )?;
     Ok(())
 }
 
@@ -110,15 +108,14 @@ pub async fn un_soft_fail_sticky_events(
     if event_ids.is_empty() {
         return Ok(());
     }
-    txn.execute(
+    txn.execute_later(
         "UPDATE event_json
         SET internal_metadata = (
             jsonb_set(internal_metadata::jsonb, '{soft_failed}', 'false'::jsonb)
         )::text
         WHERE event_id = ANY($1)",
-        &[&event_ids],
-    )
-    .await?;
+        crate::owned_params![event_ids],
+    );
     for event_id in event_ids {
         effects.add_event_cache_invalidation((*event_id).to_owned());
     }
@@ -145,23 +142,16 @@ pub async fn un_soft_fail_sticky_events(
 
     let stream_ids = input.sticky_stream.next_ids(txn, event_ids.len()).await?;
     effects.add_sticky_stream_ids(&stream_ids);
-    let rows: Vec<Vec<&(dyn ToSql + Sync)>> = stream_ids
+    let rows: Vec<Vec<OwnedParam>> = stream_ids
         .iter()
         .zip(&ordered)
-        .map(|(stream_id, event_id)| {
-            vec![
-                stream_id as &(dyn ToSql + Sync),
-                &input.instance_name,
-                event_id,
-            ]
-        })
+        .map(|(stream_id, event_id)| crate::owned_params![stream_id, input.instance_name, event_id])
         .collect();
     // `sticky_events` has no index on `event_id`, as in Python.
-    txn.execute_batch(
+    txn.execute_batch_later(
         "UPDATE sticky_events SET stream_id = $1::bigint, instance_name = $2
         WHERE event_id = $3",
-        &rows,
-    )
-    .await?;
+        rows,
+    );
     Ok(())
 }