diff --git a/rust/src/database/postgres/write.rs b/rust/src/database/postgres/write.rs index b39464dc6f..714ff4d82d 100644 --- a/rust/src/database/postgres/write.rs +++ b/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(name: &'static str, sql_type: &'static str, values: &[T]) -> Self + where + Vec: 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 ToOwnedSql for &T { + type Owned = T::Owned; + fn to_owned_sql(&self) -> T::Owned { + (**self).to_owned_sql() + } +} + +impl ToOwnedSql for Vec { + type Owned = Vec; + fn to_owned_sql(&self) -> Vec { + self.iter().map(T::to_owned_sql).collect() + } +} + +impl ToOwnedSql for [T] { + type Owned = Vec; + fn to_owned_sql(&self) -> Vec { + self.iter().map(T::to_owned_sql).collect() + } +} + +impl ToOwnedSql for Option<&str> { + type Owned = Option; + fn to_owned_sql(&self) -> Option { + self.map(str::to_owned) + } +} + +impl ToOwnedSql for String { + type Owned = String; + fn to_owned_sql(&self) -> String { + self.clone() + } +} + +impl ToOwnedSql for Option { + type Owned = Option; + fn to_owned_sql(&self) -> Option { + 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, Option, Option); + +/// 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, params: Vec, map_error: Option, } @@ -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> = - 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> = 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 = 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, params: Vec) { - 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, 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, map_error: Option) { + /// 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>) { + let sql: Arc = Arc::from(sql); + for params in rows { + self.queue(Arc::clone(&sql), params, None); + } + } + + fn queue(&self, sql: Arc, params: Vec, map_error: Option) { self.queued.lock().expect("not poisoned").push(Queued { sql, params, diff --git a/rust/src/persist/txn/caches.rs b/rust/src/persist/txn/caches.rs index d723d88aaa..04d00cd7c4 100644 --- a/rust/src/persist/txn/caches.rs +++ b/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(()) diff --git a/rust/src/persist/txn/current_state.rs b/rust/src/persist/txn/current_state.rs index a86dc7b886..a118f5cc21 100644 --- a/rust/src/persist/txn/current_state.rs +++ b/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> = changed + let delta_rows: Vec> = 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> = changed + let key_rows: Vec> = 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> = to_insert + let insert_rows: Vec> = 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> = local_members + let membership_rows: Vec> = 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)> { - let mut columns: Vec<(&'static str, Box)> = 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, ¶ms).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` is not `ToSql`. - let snapshots: Vec<[&str; 4]> = changes + let rows: Vec> = changes .to_insert_membership_snapshots .iter() .map(|info| { - [ - &*info.user_id, - &*info.sender, - &*info.membership_event_id, - &*info.membership, - ] - }) - .collect(); - let rows: Vec> = 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, diff --git a/rust/src/persist/txn/edges.rs b/rust/src/persist/txn/edges.rs index 54d7454730..8f92f96b0b 100644 --- a/rust/src/persist/txn/edges.rs +++ b/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>, 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>, ) -> 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 = 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(()) } diff --git a/rust/src/persist/txn/events.rs b/rust/src/persist/txn/events.rs index bad5411db5..688e761309 100644 --- a/rust/src/persist/txn/events.rs +++ b/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> = 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> = 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) } diff --git a/rust/src/persist/txn/metadata.rs b/rust/src/persist/txn/metadata.rs index 53678ba1b7..1d79a0e4e2 100644 --- a/rust/src/persist/txn/metadata.rs +++ b/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> = replies + let thread_rows: Vec> = 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())) diff --git a/rust/src/persist/txn/profile.rs b/rust/src/persist/txn/profile.rs index fdaf787fee..9e0e356cc3 100644 --- a/rust/src/persist/txn/profile.rs +++ b/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(()) } diff --git a/rust/src/persist/txn/sliding_sync.rs b/rust/src/persist/txn/sliding_sync.rs index bef9319e2f..03518bec42 100644 --- a/rust/src/persist/txn/sliding_sync.rs +++ b/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(()) } diff --git a/rust/src/persist/txn/sticky.rs b/rust/src/persist/txn/sticky.rs index d09141de91..54256784d3 100644 --- a/rust/src/persist/txn/sticky.rs +++ b/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 = 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> = stream_ids + let rows: Vec> = 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(()) }