From a5ba2ac74d3cf0222b58df710715766621b05128 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 06:55:59 +0000 Subject: [PATCH 01/16] perf(sqlite): route Signal reads through the read pool The store has one write permit because two writers deadlock on SQLite's transaction upgrade, and that is right for writes. It was also governing every read: a session, identity or sender-key miss on the decrypt path queued behind whatever write was in flight, including a write-behind flush of a full session batch. The reader pool that already existed for `read_pool_size` was reachable only from `SharedSqlite::read`, so turning the knob on bought the Signal path nothing. Add `read_query`, which mirrors `SharedSqlite::read`: reader connection when one is configured, write permit otherwise, so `read_pool_size = 0` keeps the previous path with no added statement. Every read-only method that only issues SELECTs now goes through it. `get_pending_inbound` stays on the write queue for its busy retry loop. The guarantee after the change is read-your-own-write across connections: a WAL reader opens on the latest committed snapshot, so a read issued after a write's await observes it. Reads that merely overlap a write see either state, which is what the single permit already gave them - it ordered them arbitrarily, not causally. Reader connections are `query_only`, so a write that slips onto the read path errors instead of escaping the serialization; a source scan fails when a new read-shaped method reaches the database any other way. --- storages/sqlite-storage/src/shared.rs | 2 +- storages/sqlite-storage/src/sqlite_store.rs | 927 ++++++++++++++------ 2 files changed, 664 insertions(+), 265 deletions(-) diff --git a/storages/sqlite-storage/src/shared.rs b/storages/sqlite-storage/src/shared.rs index 920644af7..0583972cb 100644 --- a/storages/sqlite-storage/src/shared.rs +++ b/storages/sqlite-storage/src/shared.rs @@ -105,7 +105,7 @@ impl SharedSqlite { /// [`StoreError`] deliberately has no such conversion (callers choose how a /// database error is classified), so both travel in one enum and unwrap on the /// way out. -fn read_snapshot( +pub(crate) fn read_snapshot( conn: &mut SqliteConnection, f: impl FnOnce(&mut SqliteConnection) -> Result, ) -> Result { diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 22e6d28f0..aab3454db 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -184,7 +184,9 @@ pub struct SqliteStoreConfig { pub pool_size: u32, /// Extra connections reserved for read-only work, each free to run while a /// write holds the write permit. `0` (default) keeps every operation on the - /// single queue, exactly as before this knob existed. + /// single queue, exactly as before this knob existed. This covers the + /// store's own reads (sessions, identities, sender keys) as well as + /// [`SharedSqlite::read`](crate::SharedSqlite::read). /// /// WAL supports many concurrent readers alongside one writer, but that was /// unreachable while one `pool_size` governed both the pool and the @@ -593,6 +595,61 @@ impl SqliteStore { self.device_id } + /// Run a **read-only** query on a reader connection, falling back to the + /// write queue when none is configured. + /// + /// This is where every read-only method belongs. The write permit is a + /// single slot on purpose, so a read taken through [`Self::with_semaphore`] + /// waits out whatever write is in flight; on the decrypt path that means a + /// session or identity miss queues behind a whole write-behind flush. + /// + /// Consistency: a read issued after a write's `await` returned observes it, + /// because a WAL reader opens on the latest committed snapshot. Reads that + /// merely overlap a write see either state, which is what the single permit + /// already gave them (it ordered them arbitrarily, not causally). + /// + /// Only correct for statements that cannot write: reader connections carry + /// `PRAGMA query_only`, so a write sent here fails instead of escaping the + /// serialization the store depends on. + async fn read_query(&self, f: F) -> Result + where + F: FnOnce(&mut SqliteConnection) -> Result + Send + 'static, + T: Send + 'static, + { + let Some(reads) = self.reads.clone() else { + // No reader connections: the write permit is still held for the + // whole query, so the snapshot comes for free and a transaction + // would only add statements. + let pool = self.pool.clone(); + return self + .with_semaphore(move || { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; + f(&mut conn) + }) + .await; + }; + let ReadPool { pool, semaphore } = reads; + let permit = semaphore + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; + tokio::task::spawn_blocking(move || { + let _permit = permit; + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; + // A deferred transaction pins one snapshot across a multi-statement + // read now that a writer can commit between its statements. + crate::shared::read_snapshot(&mut conn, f) + }) + .await + .map_err(|e| StoreError::Database(Box::new(e)))? + } + + /// The write queue: one permit, so two writers can never deadlock on the + /// transaction upgrade. Read-only work belongs in [`Self::read_query`]. async fn with_semaphore(&self, f: F) -> Result where F: FnOnce() -> Result + Send + 'static, @@ -913,41 +970,31 @@ impl SqliteStore { pub async fn device_exists(&self, device_id: i32) -> Result { use crate::schema::device; - let pool = self.pool.clone(); - tokio::task::spawn_blocking(move || -> Result { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - + self.read_query(move |conn| { let count: i64 = device::table .filter(device::id.eq(device_id)) .count() - .get_result(&mut conn) + .get_result(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(count > 0) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } pub async fn load_device_data_for_device(&self, device_id: i32) -> Result> { use crate::schema::device; - let pool = self.pool.clone(); - let row = tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - let result = device::table - .filter(device::id.eq(device_id)) - .first::(&mut conn) - .optional() - .map_err(|e| StoreError::Database(Box::new(e)))?; - Ok(result) - }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))??; + let row = self + .read_query(move |conn| { + let result = device::table + .filter(device::id.eq(device_id)) + .first::(conn) + .optional() + .map_err(|e| StoreError::Database(Box::new(e)))?; + Ok(result) + }) + .await?; if let Some(row) = row { let pn = if !row.pn.is_empty() { @@ -1138,25 +1185,18 @@ impl SqliteStore { address: &str, device_id: i32, ) -> Result>> { - let pool = self.pool.clone(); let address = address.to_string(); - let result = self - .with_semaphore(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - let res: Option> = identities::table - .select(identities::key) - .filter(identities::address.eq(address)) - .filter(identities::device_id.eq(device_id)) - .first(&mut conn) - .optional() - .map_err(|e| StoreError::Database(Box::new(e)))?; - Ok(res) - }) - .await?; - - Ok(result) + self.read_query(move |conn| { + let res: Option> = identities::table + .select(identities::key) + .filter(identities::address.eq(address)) + .filter(identities::device_id.eq(device_id)) + .first(conn) + .optional() + .map_err(|e| StoreError::Database(Box::new(e)))?; + Ok(res) + }) + .await } pub async fn get_session_for_device( @@ -1164,26 +1204,19 @@ impl SqliteStore { address: &str, device_id: i32, ) -> Result>> { - let pool = self.pool.clone(); let address_for_query = address.to_string(); - let result = self - .with_semaphore(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - let res: Option> = sessions::table - .select(sessions::record) - .filter(sessions::address.eq(address_for_query.clone())) - .filter(sessions::device_id.eq(device_id)) - .first(&mut conn) - .optional() - .map_err(|e| StoreError::Database(Box::new(e)))?; - - Ok(res) - }) - .await?; + self.read_query(move |conn| { + let res: Option> = sessions::table + .select(sessions::record) + .filter(sessions::address.eq(address_for_query)) + .filter(sessions::device_id.eq(device_id)) + .first(conn) + .optional() + .map_err(|e| StoreError::Database(Box::new(e)))?; - Ok(result) + Ok(res) + }) + .await } pub async fn put_session_for_device( @@ -1315,23 +1348,18 @@ impl SqliteStore { address: &str, device_id: i32, ) -> Result>> { - let pool = self.pool.clone(); let address = address.to_string(); - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let res: Option> = sender_keys::table .select(sender_keys::record) .filter(sender_keys::address.eq(address)) .filter(sender_keys::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } pub async fn delete_sender_key_for_device(&self, address: &str, device_id: i32) -> Result<()> { @@ -1360,24 +1388,19 @@ impl SqliteStore { key_id: &[u8], device_id: i32, ) -> Result> { - let pool = self.pool.clone(); let key_id = key_id.to_vec(); - let res: Option> = - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + let res: Option> = self + .read_query(move |conn| { let res: Option> = app_state_keys::table .select(app_state_keys::key_data) .filter(app_state_keys::key_id.eq(&key_id)) .filter(app_state_keys::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))??; + .await?; if let Some(data) = res { // An undecodable blob (an old bincode row or genuine corruption) is @@ -1434,12 +1457,8 @@ impl SqliteStore { &self, device_id: i32, ) -> Result>> { - let pool = self.pool.clone(); - let res: Option> = - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + let res: Option> = self + .read_query(move |conn| { // Return the latest key whose blob actually decodes. A legacy bincode // row (or a corrupt one) reads as absent via get_sync_key but still // sits in the table with a possibly lexicographically-higher key_id; @@ -1450,7 +1469,7 @@ impl SqliteStore { .select((app_state_keys::key_id, app_state_keys::key_data)) .filter(app_state_keys::device_id.eq(device_id)) .order(app_state_keys::key_id.desc()) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; let res = candidates .into_iter() @@ -1458,8 +1477,7 @@ impl SqliteStore { .map(|(key_id, _)| key_id); Ok(res) }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))??; + .await?; Ok(res) } @@ -1468,24 +1486,19 @@ impl SqliteStore { name: &str, device_id: i32, ) -> Result { - let pool = self.pool.clone(); let name = name.to_string(); - let res: Option> = - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + let res: Option> = self + .read_query(move |conn| { let res: Option> = app_state_versions::table .select(app_state_versions::state_data) .filter(app_state_versions::name.eq(name)) .filter(app_state_versions::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))??; + .await?; if let Some(data) = res { // An undecodable blob (an old bincode row or corruption) resets the @@ -1632,25 +1645,20 @@ impl SqliteStore { index_mac: &[u8], device_id: i32, ) -> Result>> { - let pool = self.pool.clone(); let name = name.to_string(); let index_mac = index_mac.to_vec(); - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let res: Option> = app_state_mutation_macs::table .select(app_state_mutation_macs::value_mac) .filter(app_state_mutation_macs::name.eq(&name)) .filter(app_state_mutation_macs::index_mac.eq(&index_mac)) .filter(app_state_mutation_macs::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } /// Batched read of previous-MAC values for many index_macs in one query @@ -1665,39 +1673,34 @@ impl SqliteStore { if index_macs.is_empty() { return Ok(std::collections::HashMap::new()); } - let pool = self.pool.clone(); let name = name.to_string(); let index_macs: Vec<[u8; 32]> = index_macs.to_vec(); - tokio::task::spawn_blocking( - move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - let mut out = std::collections::HashMap::with_capacity(index_macs.len()); - const CHUNK_SIZE: usize = 500; - for chunk in index_macs.chunks(CHUNK_SIZE) { - let chunk_slices: Vec<&[u8]> = chunk.iter().map(|m| m.as_slice()).collect(); - let rows: Vec<(Vec, Vec)> = app_state_mutation_macs::table - .select(( - app_state_mutation_macs::index_mac, - app_state_mutation_macs::value_mac, - )) - .filter(app_state_mutation_macs::name.eq(&name)) - .filter(app_state_mutation_macs::index_mac.eq_any(chunk_slices)) - .filter(app_state_mutation_macs::device_id.eq(device_id)) - .load(&mut conn) - .map_err(|e| StoreError::Database(Box::new(e)))?; - // Rows with a non-32-byte index_mac cannot have come from the - // 32-byte keys we just queried; skip defensively. - out.extend(rows.into_iter().filter_map(|(k, v)| { + self.read_query(move |conn| { + let mut out = std::collections::HashMap::with_capacity(index_macs.len()); + const CHUNK_SIZE: usize = 500; + for chunk in index_macs.chunks(CHUNK_SIZE) { + let chunk_slices: Vec<&[u8]> = chunk.iter().map(|m| m.as_slice()).collect(); + let rows: Vec<(Vec, Vec)> = app_state_mutation_macs::table + .select(( + app_state_mutation_macs::index_mac, + app_state_mutation_macs::value_mac, + )) + .filter(app_state_mutation_macs::name.eq(&name)) + .filter(app_state_mutation_macs::index_mac.eq_any(chunk_slices)) + .filter(app_state_mutation_macs::device_id.eq(device_id)) + .load(conn) + .map_err(|e| StoreError::Database(Box::new(e)))?; + // Rows with a non-32-byte index_mac cannot have come from the + // 32-byte keys we just queried; skip defensively. + out.extend( + rows.into_iter().filter_map(|(k, v)| { <[u8; 32]>::try_from(k.as_slice()).ok().map(|k| (k, v)) - })); - } - Ok(out) - }, - ) + }), + ); + } + Ok(out) + }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } } @@ -1770,19 +1773,15 @@ impl SignalStore for SqliteStore { } async fn has_session(&self, address: &str) -> Result { - let pool = self.pool.clone(); let device_id = self.device_id; let address_owned = address.to_string(); - self.with_semaphore(move || -> Result { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let exists = diesel::select(diesel::dsl::exists( sessions::table .filter(sessions::address.eq(&address_owned)) .filter(sessions::device_id.eq(device_id)), )) - .get_result(&mut conn) + .get_result(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(exists) }) @@ -1790,16 +1789,12 @@ impl SignalStore for SqliteStore { } async fn has_signal_state_for_user(&self, user: &str) -> Result { - let pool = self.pool.clone(); let device_id = self.device_id; // Address is `user@server` (device 0) or `user:dev@server`; `user` is a // numeric PN/LID so it carries no LIKE wildcards. let pat_at = format!("{user}@%"); let pat_dev = format!("{user}:%"); - self.with_semaphore(move || -> Result { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let has_session = diesel::select(diesel::dsl::exists( sessions::table .filter(sessions::device_id.eq(device_id)) @@ -1809,7 +1804,7 @@ impl SignalStore for SqliteStore { .or(sessions::address.like(&pat_dev)), ), )) - .get_result::(&mut conn) + .get_result::(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; if has_session { return Ok(true); @@ -1823,7 +1818,7 @@ impl SignalStore for SqliteStore { .or(identities::address.like(&pat_dev)), ), )) - .get_result::(&mut conn) + .get_result::(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(has_identity) }) @@ -2004,36 +1999,27 @@ impl SignalStore for SqliteStore { } async fn load_prekey(&self, id: u32) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let res: Option> = prekeys::table .select(prekeys::key) .filter(prekeys::id.eq(id as i32)) .filter(prekeys::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res.map(Bytes::from)) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn load_prekeys_batch(&self, ids: &[u32]) -> Result> { if ids.is_empty() { return Ok(Vec::new()); } - let pool = self.pool.clone(); let device_id = self.device_id; let ids: Vec = ids.iter().map(|&id| id as i32).collect(); - self.with_semaphore(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { // Chunked like mark_prekeys_uploaded: the upload window can carry // more ids than SQLite's host-parameter limit. let mut out = Vec::with_capacity(ids.len()); @@ -2042,7 +2028,7 @@ impl SignalStore for SqliteStore { .select((prekeys::id, prekeys::key)) .filter(prekeys::id.eq_any(chunk)) .filter(prekeys::device_id.eq(device_id)) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; out.extend( rows.into_iter() @@ -2133,28 +2119,17 @@ impl SignalStore for SqliteStore { } async fn get_max_prekey_id(&self) -> Result { - let pool = self.pool.clone(); let device_id = self.device_id; - let db_semaphore = self.db_semaphore.clone(); - let _permit = db_semaphore - .acquire() - .await - .map_err(|e| StoreError::Database(Box::new(e)))?; - - tokio::task::spawn_blocking(move || -> Result { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { use diesel::dsl::max; let result: Option = prekeys::table .filter(prekeys::device_id.eq(device_id)) .select(max(prekeys::id)) - .first(&mut conn) + .first(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(result.unwrap_or(0) as u32) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn store_signed_prekey(&self, id: u32, record: &[u8]) -> Result<()> { @@ -2216,36 +2191,27 @@ impl SignalStore for SqliteStore { } async fn load_signed_prekey(&self, id: u32) -> Result>> { - let pool = self.pool.clone(); let device_id = self.device_id; - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let res: Option> = signed_prekeys::table .select(signed_prekeys::record) .filter(signed_prekeys::id.eq(id as i32)) .filter(signed_prekeys::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn load_all_signed_prekeys(&self) -> Result)>> { - let pool = self.pool.clone(); let device_id = self.device_id; - tokio::task::spawn_blocking(move || -> Result)>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let results: Vec<(i32, Vec)> = signed_prekeys::table .select((signed_prekeys::id, signed_prekeys::record)) .filter(signed_prekeys::device_id.eq(device_id)) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(results .into_iter() @@ -2253,7 +2219,6 @@ impl SignalStore for SqliteStore { .collect()) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn remove_signed_prekey(&self, id: u32) -> Result<()> { @@ -2474,18 +2439,14 @@ fn delete_pending_inbound_row( #[cfg_attr(not(target_arch = "wasm32"), async_trait)] impl ProtocolStore for SqliteStore { async fn get_sender_key_devices(&self, group_jid: &str) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; let group_jid = group_jid.to_string(); - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let rows: Vec<(String, i32)> = sender_key_devices::table .select((sender_key_devices::device_jid, sender_key_devices::has_key)) .filter(sender_key_devices::group_jid.eq(&group_jid)) .filter(sender_key_devices::device_id.eq(device_id)) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() @@ -2493,7 +2454,6 @@ impl ProtocolStore for SqliteStore { .collect()) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn set_sender_key_status(&self, group_jid: &str, entries: &[(&str, bool)]) -> Result<()> { @@ -2606,13 +2566,9 @@ impl ProtocolStore for SqliteStore { } async fn get_lid_mapping(&self, lid: &str) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; let lid = lid.to_string(); - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2623,7 +2579,7 @@ impl ProtocolStore for SqliteStore { )) .filter(lid_pn_mapping::lid.eq(&lid)) .filter(lid_pn_mapping::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( @@ -2637,17 +2593,12 @@ impl ProtocolStore for SqliteStore { )) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn get_pn_mapping(&self, phone: &str) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; let phone = phone.to_string(); - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2659,7 +2610,7 @@ impl ProtocolStore for SqliteStore { .filter(lid_pn_mapping::phone_number.eq(&phone)) .filter(lid_pn_mapping::device_id.eq(device_id)) .order(lid_pn_mapping::updated_at.desc()) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( @@ -2673,7 +2624,6 @@ impl ProtocolStore for SqliteStore { )) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn put_lid_mapping(&self, entry: &LidPnMappingEntry) -> Result<()> { @@ -2720,12 +2670,8 @@ impl ProtocolStore for SqliteStore { } async fn get_all_lid_mappings(&self) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let rows: Vec<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2735,7 +2681,7 @@ impl ProtocolStore for SqliteStore { lid_pn_mapping::updated_at, )) .filter(lid_pn_mapping::device_id.eq(device_id)) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() @@ -2753,7 +2699,6 @@ impl ProtocolStore for SqliteStore { .collect()) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn save_base_key(&self, address: &str, message_id: &str, base_key: &[u8]) -> Result<()> { @@ -2797,27 +2742,22 @@ impl ProtocolStore for SqliteStore { message_id: &str, current_base_key: &[u8], ) -> Result { - let pool = self.pool.clone(); let device_id = self.device_id; let address = address.to_string(); let message_id = message_id.to_string(); let current_base_key = current_base_key.to_vec(); - tokio::task::spawn_blocking(move || -> Result { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let stored_key: Option> = base_keys::table .select(base_keys::base_key) .filter(base_keys::address.eq(&address)) .filter(base_keys::message_id.eq(&message_id)) .filter(base_keys::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(stored_key.as_ref() == Some(¤t_base_key)) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_base_key(&self, address: &str, message_id: &str) -> Result<()> { @@ -2951,13 +2891,9 @@ impl ProtocolStore for SqliteStore { } async fn get_devices(&self, user: &str) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; let user = user.to_string(); - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option<(String, String, i32, Option, Option)> = device_registry::table .select(( @@ -2969,7 +2905,7 @@ impl ProtocolStore for SqliteStore { )) .filter(device_registry::user_id.eq(&user)) .filter(device_registry::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; match row { @@ -2988,7 +2924,6 @@ impl ProtocolStore for SqliteStore { } }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_devices(&self, user: &str) -> Result<()> { @@ -3014,24 +2949,19 @@ impl ProtocolStore for SqliteStore { } async fn get_group_metadata(&self, group_jid: &str) -> Result>> { - let pool = self.pool.clone(); let device_id = self.device_id; let group_jid = group_jid.to_string(); - tokio::task::spawn_blocking(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option> = group_metadata::table .select(group_metadata::info) .filter(group_metadata::group_jid.eq(&group_jid)) .filter(group_metadata::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn put_group_metadata(&self, group_jid: &str, blob: &[u8]) -> Result<()> { @@ -3089,13 +3019,9 @@ impl ProtocolStore for SqliteStore { } async fn get_tc_token(&self, jid: &str) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; let jid = jid.to_string(); - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option<(Vec, i64, Option)> = tc_tokens::table .select(( tc_tokens::token, @@ -3104,7 +3030,7 @@ impl ProtocolStore for SqliteStore { )) .filter(tc_tokens::jid.eq(&jid)) .filter(tc_tokens::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok( @@ -3116,7 +3042,6 @@ impl ProtocolStore for SqliteStore { ) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn put_tc_token(&self, jid: &str, entry: &TcTokenEntry) -> Result<()> { @@ -3178,21 +3103,16 @@ impl ProtocolStore for SqliteStore { } async fn get_all_tc_token_jids(&self) -> Result> { - let pool = self.pool.clone(); let device_id = self.device_id; - tokio::task::spawn_blocking(move || -> Result> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let jids: Vec = tc_tokens::table .select(tc_tokens::jid) .filter(tc_tokens::device_id.eq(device_id)) - .load(&mut conn) + .load(conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(jids) }) .await - .map_err(|e| StoreError::Database(Box::new(e)))? } async fn delete_expired_tc_tokens(&self, token_cutoff: i64, sender_cutoff: i64) -> Result { @@ -3664,25 +3584,18 @@ impl MsgSecretStore for SqliteStore { sender: &str, msg_id: &str, ) -> Result>> { - // Serialized through the db semaphore for the same reason as - // get_msg_secret_with_ts: a read racing a write transaction must wait, - // not error out as a phantom miss. - let pool = self.pool.clone(); let device_id = self.device_id; let chat = chat.to_string(); let sender = sender.to_string(); let msg_id = msg_id.to_string(); - self.with_semaphore(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option> = msg_secrets::table .select(msg_secrets::secret) .filter(msg_secrets::chat.eq(&chat)) .filter(msg_secrets::sender.eq(&sender)) .filter(msg_secrets::msg_id.eq(&msg_id)) .filter(msg_secrets::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row) @@ -3696,26 +3609,22 @@ impl MsgSecretStore for SqliteStore { sender: &str, msg_id: &str, ) -> Result, i64)>> { - // Serialized through the db semaphore: a raw read racing a write - // transaction hits the shared-cache table lock on in-memory stores - // (SQLITE_LOCKED is not covered by busy_timeout) and callers treat the - // error as a missing secret. - let pool = self.pool.clone(); + // Never a raw pooled read: a read racing a write transaction must wait, + // not error out as a phantom miss. `read_query` either holds the write + // permit or uses a WAL reader, and the shared-cache table lock that + // `busy_timeout` cannot absorb only exists in the former's stores. let device_id = self.device_id; let chat = chat.to_string(); let sender = sender.to_string(); let msg_id = msg_id.to_string(); - self.with_semaphore(move || -> Result, i64)>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; + self.read_query(move |conn| { let row: Option<(Vec, i64)> = msg_secrets::table .select((msg_secrets::secret, msg_secrets::message_ts)) .filter(msg_secrets::chat.eq(&chat)) .filter(msg_secrets::sender.eq(&sender)) .filter(msg_secrets::msg_id.eq(&msg_id)) .filter(msg_secrets::device_id.eq(device_id)) - .first(&mut conn) + .first(conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row) @@ -5692,3 +5601,493 @@ mod tests { } } } + +/// Routing of read-only work onto the reader connections. +#[cfg(test)] +mod read_routing_tests { + use super::*; + + /// A file-backed store: reader connections need real WAL, which an + /// in-memory database has none of. Removed on drop. + struct TempDb(std::path::PathBuf); + + impl TempDb { + fn new(tag: &str) -> Self { + use portable_atomic::AtomicU64; + use std::sync::atomic::Ordering; + static COUNTER: AtomicU64 = AtomicU64::new(0); + let id = COUNTER.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "wa_read_routing_{tag}_{}_{id}.db", + std::process::id() + )); + let _ = std::fs::remove_file(&path); + Self(path) + } + + fn url(&self) -> String { + self.0.to_string_lossy().into_owned() + } + } + + impl Drop for TempDb { + fn drop(&mut self) { + for suffix in ["", "-wal", "-shm"] { + let mut p = self.0.clone().into_os_string(); + p.push(suffix); + let _ = std::fs::remove_file(p); + } + } + } + + async fn store_with(read_pool_size: u32, db: &TempDb) -> SqliteStore { + let store = SqliteStore::with_config( + &db.url(), + SqliteStoreConfig { + read_pool_size, + ..Default::default() + }, + ) + .await + .expect("store opens"); + assert_eq!( + store.reads.is_some(), + read_pool_size > 0, + "a file-backed store honours read_pool_size" + ); + store.create_new_device().await.expect("device row"); + store + } + + const ADDR: &str = "559990000001:0@s.whatsapp.net"; + const GROUP: &str = "1234567890-1111111111@g.us"; + + /// Every migrated read answers "absent" before its row exists, and answers + /// with the written value immediately after the write returns. The second + /// half is the read-your-own-write guarantee the routing relies on: a WAL + /// reader opens on the latest committed snapshot, so a read issued after a + /// write's `await` observes it even from another connection. + async fn exercise_reads(read_pool_size: u32) { + let db = TempDb::new(&format!("rw{read_pool_size}")); + let store = store_with(read_pool_size, &db).await; + + // Absent everywhere first. + assert_eq!(store.load_identity(ADDR).await.unwrap(), None); + assert_eq!(store.get_session(ADDR).await.unwrap(), None); + assert!(!store.has_session(ADDR).await.unwrap()); + assert!( + !store + .has_signal_state_for_user("559990000001") + .await + .unwrap() + ); + assert_eq!(store.get_sender_key(ADDR).await.unwrap(), None); + assert_eq!(store.load_prekey(7).await.unwrap(), None); + assert!(store.load_prekeys_batch(&[7]).await.unwrap().is_empty()); + assert_eq!(store.get_max_prekey_id().await.unwrap(), 0); + assert_eq!(store.load_signed_prekey(3).await.unwrap(), None); + assert!(store.load_all_signed_prekeys().await.unwrap().is_empty()); + assert!( + store + .get_sender_key_devices(GROUP) + .await + .unwrap() + .is_empty() + ); + assert!(store.get_sync_key(b"k1").await.unwrap().is_none()); + assert_eq!(store.get_latest_sync_key_id().await.unwrap(), None); + assert_eq!(store.get_version("critical").await.unwrap().version, 0); + assert_eq!( + store + .get_mutation_mac("critical", &[1u8; 32]) + .await + .unwrap(), + None + ); + assert!( + store + .get_mutation_macs("critical", &[[1u8; 32]]) + .await + .unwrap() + .is_empty() + ); + assert!(store.get_lid_mapping("111@lid").await.unwrap().is_none()); + assert!( + store + .get_pn_mapping("559990000002") + .await + .unwrap() + .is_none() + ); + assert!(store.get_all_lid_mappings().await.unwrap().is_empty()); + assert!( + !store + .has_same_base_key(ADDR, "m1", &[1, 2, 3]) + .await + .unwrap() + ); + assert!(store.get_devices("559990000001").await.unwrap().is_none()); + assert_eq!(store.get_group_metadata(GROUP).await.unwrap(), None); + assert!( + store + .get_tc_token("559990000001@s.whatsapp.net") + .await + .unwrap() + .is_none() + ); + assert!(store.get_all_tc_token_jids().await.unwrap().is_empty()); + assert_eq!(store.get_msg_secret(GROUP, ADDR, "m1").await.unwrap(), None); + assert_eq!( + store + .get_msg_secret_with_ts(GROUP, ADDR, "m1") + .await + .unwrap(), + None + ); + assert!(store.device_exists(1).await.unwrap()); + assert!( + store + .load_device_data_for_device(1) + .await + .unwrap() + .is_some() + ); + + // Write, then read back through the (possibly separate) connection. + store.put_identity(ADDR, [4u8; 32]).await.unwrap(); + assert_eq!(store.load_identity(ADDR).await.unwrap(), Some([4u8; 32])); + + store.put_session(ADDR, b"session-blob").await.unwrap(); + assert_eq!( + store.get_session(ADDR).await.unwrap().as_deref(), + Some(&b"session-blob"[..]) + ); + assert!(store.has_session(ADDR).await.unwrap()); + assert!( + store + .has_signal_state_for_user("559990000001") + .await + .unwrap() + ); + + store.put_sender_key(ADDR, b"sk-blob").await.unwrap(); + assert_eq!( + store.get_sender_key(ADDR).await.unwrap(), + Some(b"sk-blob".to_vec()) + ); + + store.store_prekey(7, b"pk", false).await.unwrap(); + assert_eq!( + store.load_prekey(7).await.unwrap().as_deref(), + Some(&b"pk"[..]) + ); + assert_eq!(store.load_prekeys_batch(&[7]).await.unwrap().len(), 1); + assert_eq!(store.get_max_prekey_id().await.unwrap(), 7); + + store.store_signed_prekey(3, b"spk").await.unwrap(); + assert_eq!( + store.load_signed_prekey(3).await.unwrap(), + Some(b"spk".to_vec()) + ); + assert_eq!(store.load_all_signed_prekeys().await.unwrap().len(), 1); + + store + .set_sender_key_status(GROUP, &[("559990000003:0@s.whatsapp.net", true)]) + .await + .unwrap(); + assert_eq!(store.get_sender_key_devices(GROUP).await.unwrap().len(), 1); + + let key = AppStateSyncKey { + key_data: vec![1; 32], + fingerprint: vec![2; 4], + timestamp: 99, + }; + store.set_sync_key(b"k1", key.clone()).await.unwrap(); + let got = store.get_sync_key(b"k1").await.unwrap().expect("sync key"); + assert_eq!(got.key_data, key.key_data); + assert_eq!(got.fingerprint, key.fingerprint); + assert_eq!(got.timestamp, key.timestamp); + assert_eq!( + store.get_latest_sync_key_id().await.unwrap(), + Some(b"k1".to_vec()) + ); + + let state = HashState { + version: 42, + ..Default::default() + }; + store.set_version("critical", state).await.unwrap(); + assert_eq!(store.get_version("critical").await.unwrap().version, 42); + + let mac = AppStateMutationMAC { + index_mac: vec![1u8; 32], + value_mac: vec![9u8; 32], + }; + store + .put_mutation_macs("critical", 1, std::slice::from_ref(&mac)) + .await + .unwrap(); + assert_eq!( + store + .get_mutation_mac("critical", &mac.index_mac) + .await + .unwrap(), + Some(mac.value_mac.clone()) + ); + assert_eq!( + store + .get_mutation_macs("critical", &[[1u8; 32]]) + .await + .unwrap() + .len(), + 1 + ); + + store + .put_lid_mapping(&LidPnMappingEntry { + lid: "111@lid".to_string(), + phone_number: "559990000002".to_string(), + created_at: 1, + updated_at: 1, + learning_source: "test".to_string(), + }) + .await + .unwrap(); + assert!(store.get_lid_mapping("111@lid").await.unwrap().is_some()); + assert!( + store + .get_pn_mapping("559990000002") + .await + .unwrap() + .is_some() + ); + assert_eq!(store.get_all_lid_mappings().await.unwrap().len(), 1); + + store.save_base_key(ADDR, "m1", &[1, 2, 3]).await.unwrap(); + assert!( + store + .has_same_base_key(ADDR, "m1", &[1, 2, 3]) + .await + .unwrap() + ); + + store + .update_device_list(DeviceListRecord { + user: "559990000001".to_string(), + devices: Vec::new(), + timestamp: 5, + phash: None, + raw_id: None, + }) + .await + .unwrap(); + assert!(store.get_devices("559990000001").await.unwrap().is_some()); + + store.put_group_metadata(GROUP, b"meta").await.unwrap(); + assert_eq!( + store.get_group_metadata(GROUP).await.unwrap(), + Some(b"meta".to_vec()) + ); + + store + .put_tc_token( + "559990000001@s.whatsapp.net", + &TcTokenEntry { + token: vec![7], + token_timestamp: 3, + sender_timestamp: None, + }, + ) + .await + .unwrap(); + assert!( + store + .get_tc_token("559990000001@s.whatsapp.net") + .await + .unwrap() + .is_some() + ); + assert_eq!(store.get_all_tc_token_jids().await.unwrap().len(), 1); + + store + .put_msg_secrets(vec![MsgSecretEntry { + chat: GROUP.into(), + sender: ADDR.into(), + msg_id: "m1".into(), + secret: [5u8; 32], + expires_at: 0, + message_ts: 11, + }]) + .await + .unwrap(); + assert_eq!( + store.get_msg_secret(GROUP, ADDR, "m1").await.unwrap(), + Some(vec![5u8; 32]) + ); + assert_eq!( + store + .get_msg_secret_with_ts(GROUP, ADDR, "m1") + .await + .unwrap(), + Some((vec![5u8; 32], 11)) + ); + } + + #[tokio::test] + async fn reads_answer_the_same_without_reader_connections() { + exercise_reads(0).await; + } + + #[tokio::test] + async fn reads_answer_the_same_with_reader_connections() { + exercise_reads(4).await; + } + + /// The safety net: reader connections are `query_only`, so a write that + /// slips into `read_query` fails loudly instead of escaping the write + /// serialization and deadlocking against the real writer. + #[tokio::test] + async fn a_write_through_read_query_is_refused() { + let db = TempDb::new("query_only"); + let store = store_with(1, &db).await; + + let result = store + .read_query(|conn| { + diesel::delete(sessions::table) + .execute(conn) + .map_err(|e| StoreError::Database(Box::new(e)))?; + Ok(()) + }) + .await; + assert!( + matches!(result, Err(StoreError::Database(_))), + "query_only must reject a write on the read path" + ); + } + + /// A read must not wait out a write. Holds the write permit and checks the + /// migrated reads still answer; without reader connections this is exactly + /// the stall the change exists to remove. + #[tokio::test] + async fn a_read_proceeds_while_the_write_permit_is_held() { + let db = TempDb::new("no_wait"); + let store = store_with(2, &db).await; + store.put_session(ADDR, b"blob").await.unwrap(); + + let _permit = store + .db_semaphore + .clone() + .acquire_owned() + .await + .expect("the only write permit"); + + let got = tokio::time::timeout(Duration::from_secs(10), store.get_session(ADDR)) + .await + .expect("a read must not queue behind the write permit") + .expect("read succeeds"); + assert_eq!(got.as_deref(), Some(&b"blob"[..])); + } + + /// Read-only methods left on the write queue on purpose, with the reason. + /// Anything else matching a read-shaped name has to route through + /// `read_query` or this test fails. + const ON_THE_WRITE_QUEUE: &[(&str, &str)] = &[( + "get_pending_inbound", + "retries SQLITE_BUSY on the write queue: a read error here fails closed \ + and forces an unnecessary redelivery", + )]; + + /// Read-shaped methods that reach the database without going through + /// `read_query`, ignoring the ones excused above, plus how many were + /// scanned at all so the check cannot pass by matching nothing. + fn misrouted_reads(source: &str) -> (Vec, usize) { + let source = source + .split_once("\n#[cfg(test)]") + .map(|(before, _)| before) + .unwrap_or(source); + + let mut current: Option<(&str, String)> = None; + let mut offenders: Vec = Vec::new(); + let mut scanned = 0usize; + for line in source.lines() { + if let Some((name, body)) = current.as_mut() { + if line == " }" { + let touches_db = [ + "self.pool", + "with_semaphore(", + "with_retry(", + "spawn_blocking(", + ] + .iter() + .any(|token| body.contains(token)); + if touches_db + && !body.contains("read_query(") + && !ON_THE_WRITE_QUEUE + .iter() + .any(|(allowed, _)| allowed == name) + { + offenders.push((*name).to_string()); + } + current = None; + } else { + body.push_str(line); + } + continue; + } + let Some(rest) = line + .strip_prefix(" pub async fn ") + .or_else(|| line.strip_prefix(" async fn ")) + else { + continue; + }; + let name = rest.split(['(', '<']).next().unwrap_or_default(); + if ["get_", "load_", "has_"] + .iter() + .any(|prefix| name.starts_with(prefix)) + { + current = Some((name, String::new())); + scanned += 1; + } + } + (offenders, scanned) + } + + /// A new read-only method written the old way (raw pool checkout, write + /// permit, or the retry loop) silently rejoins the write queue, and nothing + /// about it looks wrong at the call site. Scanning our own source is the + /// only place that can see the routing decision. + #[test] + fn read_shaped_methods_route_through_read_query() { + let (offenders, scanned) = misrouted_reads(include_str!("sqlite_store.rs")); + assert!( + offenders.is_empty(), + "read-only methods must call read_query (or be listed in ON_THE_WRITE_QUEUE \ + with a reason): {offenders:?}" + ); + assert!( + scanned > 20, + "the scan saw only {scanned} read-shaped methods" + ); + } + + /// The scan is worth nothing if it cannot see a violation, so feed it one. + #[test] + fn the_routing_scan_catches_a_misrouted_read() { + let regression = "\ +impl SqliteStore { + pub async fn get_something_new(&self) -> Result<()> { + let pool = self.pool.clone(); + tokio::task::spawn_blocking(move || Ok(())).await + } + + async fn get_something_routed(&self) -> Result<()> { + self.read_query(move |_conn| Ok(())).await + } +} +"; + assert_eq!( + misrouted_reads(regression), + (vec!["get_something_new".to_string()], 2) + ); + } +} From 2c08efa1aeee6b77b5fb50fa11fc8d5fce53cdb7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 07:27:25 +0000 Subject: [PATCH 02/16] fix(sqlite): make chunked batch writes atomic and shrink the read path Review of the read routing turned up three things worth fixing. Chunking in `put_app_state_mutation_macs_for_device`, `delete_app_state_mutation_macs_for_device`, `mark_prekeys_uploaded`, `set_sender_key_status` and `delete_sender_key_device_rows` works around SQLite's host-parameter limit, but each chunk was committing on its own. That was invisible while reads held the same permit; now that a reader can run alongside, it can land between two chunks and see half a batch. Wrap each loop in one transaction, which also stops a crash mid-batch from persisting a partial one. Regression test races a four-chunk write against a reader and fails on any count that is neither before nor after. The read helper was generic over its closure, so two dozen call sites each monomorphized a body carrying Diesel's transaction machinery: +90 KiB of .text, over the size gate. Erase the closure into a boxed `FnOnce` first, so the body instantiates once per return type. The reader branch then delegates to `SharedSqlite::read` rather than restating acquire-checkout-snapshot, which leaves that sequence with one implementation and reverts the visibility change to `read_snapshot`. Measured against the branch point: +31.5 KiB stripped, +23.5 KiB .text, both inside the per-PR budget. Also: the routing scan now covers `is_`, `list_`, `count_`, `find_`, `fetch_` and the `_exists` suffix, so `device_exists` and future read-shaped names are inspected too. And a test pins the behaviour the msg-secret reads were previously kept on the write queue for: with a real write transaction open, a read returns the last commit instead of blocking or failing. --- storages/sqlite-storage/src/shared.rs | 2 +- storages/sqlite-storage/src/sqlite_store.rs | 299 ++++++++++++++------ 2 files changed, 211 insertions(+), 90 deletions(-) diff --git a/storages/sqlite-storage/src/shared.rs b/storages/sqlite-storage/src/shared.rs index 0583972cb..920644af7 100644 --- a/storages/sqlite-storage/src/shared.rs +++ b/storages/sqlite-storage/src/shared.rs @@ -105,7 +105,7 @@ impl SharedSqlite { /// [`StoreError`] deliberately has no such conversion (callers choose how a /// database error is classified), so both travel in one enum and unwrap on the /// way out. -pub(crate) fn read_snapshot( +fn read_snapshot( conn: &mut SqliteConnection, f: impl FnOnce(&mut SqliteConnection) -> Result, ) -> Result { diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index aab3454db..c10dd0af6 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -97,6 +97,10 @@ const ID_PARAM_CHUNK: usize = 900; /// limit while bounding Diesel's temporary insert-expression allocation. const MSG_SECRET_INSERT_CHUNK_SIZE: usize = 100; +/// A read-only closure with its type erased, so the read path monomorphizes +/// once per return type rather than once per call site. +type ReadQuery = Box Result + Send>; + /// Reader connections and the permits that bound how many run at once. #[derive(Clone)] pub(crate) struct ReadPool { @@ -616,10 +620,17 @@ impl SqliteStore { F: FnOnce(&mut SqliteConnection) -> Result + Send + 'static, T: Send + 'static, { - let Some(reads) = self.reads.clone() else { - // No reader connections: the write permit is still held for the - // whole query, so the snapshot comes for free and a transaction - // would only add statements. + // Erase the closure before the real body: two dozen read methods + // through a generic body carrying Diesel's transaction machinery + // monomorphizes per call site, and that is ~90 KiB of .text. + self.read_erased(Box::new(f)).await + } + + async fn read_erased(&self, f: ReadQuery) -> Result { + if self.reads.is_none() { + // No reader connections: the single pooled connection is what no + // writer can be holding while this query runs, so the snapshot + // comes for free and a transaction would only add statements. let pool = self.pool.clone(); return self .with_semaphore(move || { @@ -629,23 +640,10 @@ impl SqliteStore { f(&mut conn) }) .await; - }; - let ReadPool { pool, semaphore } = reads; - let permit = semaphore - .acquire_owned() - .await - .map_err(|e| StoreError::Database(Box::new(e)))?; - tokio::task::spawn_blocking(move || { - let _permit = permit; - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - // A deferred transaction pins one snapshot across a multi-statement - // read now that a writer can commit between its statements. - crate::shared::read_snapshot(&mut conn, f) - }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))? + } + // One implementation of acquire-checkout-snapshot, shared with the + // sibling-crate read path. + self.shared().read(f).await } /// The write queue: one permit, so two writers can never deadlock on the @@ -1580,24 +1578,29 @@ impl SqliteStore { // Each row has 5 columns. 100 rows * 5 = 500 params, which is safe. const CHUNK_SIZE: usize = 100; - for chunk in records.chunks(CHUNK_SIZE) { - diesel::insert_into(app_state_mutation_macs::table) - .values(chunk) - .on_conflict(( - app_state_mutation_macs::name, - app_state_mutation_macs::index_mac, - app_state_mutation_macs::device_id, - )) - .do_update() - .set(( - app_state_mutation_macs::version - .eq(excluded(app_state_mutation_macs::version)), - app_state_mutation_macs::value_mac - .eq(excluded(app_state_mutation_macs::value_mac)), - )) - .execute(conn)?; - } - Ok(()) + // Chunking is a parameter-limit workaround, not a commit + // boundary: a reader that lands between two chunks must not see + // half a batch. + conn.transaction(|conn| { + for chunk in records.chunks(CHUNK_SIZE) { + diesel::insert_into(app_state_mutation_macs::table) + .values(chunk) + .on_conflict(( + app_state_mutation_macs::name, + app_state_mutation_macs::index_mac, + app_state_mutation_macs::device_id, + )) + .do_update() + .set(( + app_state_mutation_macs::version + .eq(excluded(app_state_mutation_macs::version)), + app_state_mutation_macs::value_mac + .eq(excluded(app_state_mutation_macs::value_mac)), + )) + .execute(conn)?; + } + Ok(()) + }) }) }) .await @@ -1622,18 +1625,20 @@ impl SqliteStore { // We use a safe chunk size to stay well within limits. const CHUNK_SIZE: usize = 500; - for chunk in index_macs.chunks(CHUNK_SIZE) { - diesel::delete( - app_state_mutation_macs::table.filter( - app_state_mutation_macs::name - .eq(&name) - .and(app_state_mutation_macs::index_mac.eq_any(chunk)) - .and(app_state_mutation_macs::device_id.eq(device_id)), - ), - ) - .execute(conn)?; - } - Ok(()) + conn.transaction(|conn| { + for chunk in index_macs.chunks(CHUNK_SIZE) { + diesel::delete( + app_state_mutation_macs::table.filter( + app_state_mutation_macs::name + .eq(&name) + .and(app_state_mutation_macs::index_mac.eq_any(chunk)) + .and(app_state_mutation_macs::device_id.eq(device_id)), + ), + ) + .execute(conn)?; + } + Ok(()) + }) }) }) .await @@ -2103,16 +2108,18 @@ impl SignalStore for SqliteStore { Box::new(move |conn: &mut SqliteConnection| { // Stay under SQLite's host-parameter limit (999 by default); // the upload batch is configurable up to u16::MAX ids. - for chunk in ids.chunks(ID_PARAM_CHUNK) { - diesel::update( - prekeys::table - .filter(prekeys::id.eq_any(chunk.to_vec())) - .filter(prekeys::device_id.eq(device_id)), - ) - .set(prekeys::uploaded.eq(true)) - .execute(conn)?; - } - Ok(()) + conn.transaction(|conn| { + for chunk in ids.chunks(ID_PARAM_CHUNK) { + diesel::update( + prekeys::table + .filter(prekeys::id.eq_any(chunk.to_vec())) + .filter(prekeys::device_id.eq(device_id)), + ) + .set(prekeys::uploaded.eq(true)) + .execute(conn)?; + } + Ok(()) + }) }) }) .await @@ -2488,22 +2495,25 @@ impl ProtocolStore for SqliteStore { const CHUNK_SIZE: usize = 190; - for chunk in values.chunks(CHUNK_SIZE) { - diesel::insert_into(sender_key_devices::table) - .values(chunk) - .on_conflict(( - sender_key_devices::group_jid, - sender_key_devices::device_jid, - sender_key_devices::device_id, - )) - .do_update() - .set(( - sender_key_devices::has_key.eq(excluded(sender_key_devices::has_key)), - sender_key_devices::updated_at.eq(now), - )) - .execute(conn)?; - } - Ok(()) + conn.transaction(|conn| { + for chunk in values.chunks(CHUNK_SIZE) { + diesel::insert_into(sender_key_devices::table) + .values(chunk) + .on_conflict(( + sender_key_devices::group_jid, + sender_key_devices::device_jid, + sender_key_devices::device_id, + )) + .do_update() + .set(( + sender_key_devices::has_key + .eq(excluded(sender_key_devices::has_key)), + sender_key_devices::updated_at.eq(now), + )) + .execute(conn)?; + } + Ok(()) + }) }) }) .await @@ -2551,15 +2561,17 @@ impl ProtocolStore for SqliteStore { let owned = Arc::clone(&owned); Box::new(move |conn: &mut SqliteConnection| { const CHUNK: usize = 190; - for chunk in owned.chunks(CHUNK) { - diesel::delete( - sender_key_devices::table - .filter(sender_key_devices::device_jid.eq_any(chunk)) - .filter(sender_key_devices::device_id.eq(device_id)), - ) - .execute(conn)?; - } - Ok(()) + conn.transaction(|conn| { + for chunk in owned.chunks(CHUNK) { + diesel::delete( + sender_key_devices::table + .filter(sender_key_devices::device_jid.eq_any(chunk)) + .filter(sender_key_devices::device_id.eq(device_id)), + ) + .execute(conn)?; + } + Ok(()) + }) }) }) .await @@ -5988,6 +6000,112 @@ mod read_routing_tests { assert_eq!(got.as_deref(), Some(&b"blob"[..])); } + /// An uncommitted write is not a lock error and not a phantom miss: the + /// reader sees the last committed state and returns it. This is the case + /// the msg-secret reads were kept on the write queue for, so it has to hold + /// with a real write transaction open, not just an idle permit. + #[tokio::test] + async fn a_read_sees_the_last_commit_while_a_write_transaction_is_open() { + let db = TempDb::new("in_flight"); + let store = store_with(2, &db).await; + store.put_session(ADDR, b"committed").await.unwrap(); + + let (open_tx, mut open_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let writer = { + let shared = store.shared(); + tokio::spawn(async move { + shared + .run(move |conn| { + conn.immediate_transaction(|conn| { + diesel::update(sessions::table) + .set(sessions::record.eq(&b"uncommitted"[..])) + .execute(conn)?; + let _ = open_tx.send(()); + // Bounded: a parked blocking task cannot be aborted, + // so an unreleased one would hang shutdown. + let _ = release_rx.recv_timeout(Duration::from_secs(20)); + Ok(()) + }) + .map_err(|e: diesel::result::Error| StoreError::Database(Box::new(e))) + }) + .await + }) + }; + + tokio::time::timeout(Duration::from_secs(10), open_rx.recv()) + .await + .expect("the write transaction must open") + .expect("writer alive"); + + let read = tokio::time::timeout(Duration::from_secs(10), store.get_session(ADDR)).await; + let _ = release_tx.send(()); + let got = read + .expect("a read must not block on an open write transaction") + .expect("a read must not fail on an open write transaction"); + assert_eq!( + got.as_deref(), + Some(&b"committed"[..]), + "the reader sees the last commit, never the open transaction" + ); + writer.await.expect("join").expect("write commits"); + + // And the committed value once the writer lands. + assert_eq!( + store.get_session(ADDR).await.unwrap().as_deref(), + Some(&b"uncommitted"[..]) + ); + } + + /// Chunking exists for SQLite's host-parameter limit, not as a commit + /// boundary. Once reads stop sharing the write permit a reader can land + /// between two chunks, so the batch has to be atomic on its own; racing the + /// two is what shows it. Samples the count while the write is in flight and + /// fails on any value that is neither the before nor the after. + #[tokio::test] + async fn a_chunked_batch_write_is_never_observed_half_applied() { + // Four chunks at set_sender_key_status's CHUNK_SIZE of 190. + const ENTRIES: usize = 760; + let db = TempDb::new("chunk_atomic"); + let store = store_with(4, &db).await; + let jids: Arc> = Arc::new( + (0..ENTRIES) + .map(|i| format!("55999{i:07}:0@s.whatsapp.net")) + .collect(), + ); + + for _ in 0..8 { + store.clear_sender_key_devices(GROUP).await.unwrap(); + let writer = { + let store = store.clone(); + let jids = Arc::clone(&jids); + tokio::spawn(async move { + let entries: Vec<(&str, bool)> = + jids.iter().map(|j| (j.as_str(), true)).collect(); + store.set_sender_key_status(GROUP, &entries).await.unwrap(); + }) + }; + + // Poll rather than sleep, and bound it so a failure reports instead + // of hanging the runtime. + let sampled = tokio::time::timeout(Duration::from_secs(20), async { + loop { + let n = store.get_sender_key_devices(GROUP).await.unwrap().len(); + assert!( + n == 0 || n == ENTRIES, + "a chunked batch was observed {n}/{ENTRIES} applied" + ); + if n == ENTRIES { + return; + } + } + }) + .await; + writer.await.unwrap(); + sampled.expect("the batch must land"); + } + } + /// Read-only methods left on the write queue on purpose, with the reason. /// Anything else matching a read-shaped name has to route through /// `read_query` or this test fails. @@ -6041,9 +6159,12 @@ mod read_routing_tests { continue; }; let name = rest.split(['(', '<']).next().unwrap_or_default(); - if ["get_", "load_", "has_"] - .iter() - .any(|prefix| name.starts_with(prefix)) + const READ_PREFIXES: &[&str] = &[ + "get_", "load_", "has_", "is_", "list_", "count_", "find_", "fetch_", + ]; + if READ_PREFIXES.iter().any(|prefix| name.starts_with(prefix)) + || name.ends_with("_exists") + || name == "exists" { current = Some((name, String::new())); scanned += 1; From a0a8cf22ea015bd966cd05e87269385ccc5134b5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 07:36:27 +0000 Subject: [PATCH 03/16] fix(sqlite): keep message-secret lookups on the write queue `get_msg_secret` and `get_msg_secret_with_ts` were not in the batch this PR set out to move, and they should not have gone with it. A miss on that path is terminal: `secret_encrypted_message` returns None and the reaction, vote or edit is dropped, with no retry and no buffering behind it. History sync seeds secrets through `put_msg_secrets` directly rather than the live write-behind buffer, so a lookup that races that batch finds nothing in the buffer and goes to the backend. The formal argument that the outcome set is unchanged still holds -- the single permit ordered a concurrent read and write arbitrarily either way -- but it widens the losing window from "the read arrives before the write starts" to "the read arrives before the write commits", and a history-sync batch commit is not short. That trade buys nothing here: the measured win is entirely on the Signal path, and these two reads contribute none of it. Both are back on the semaphore with the reason recorded, and listed in ON_THE_WRITE_QUEUE so the routing scan keeps accepting them. --- storages/sqlite-storage/src/sqlite_store.rs | 43 ++++++++++++++------- 1 file changed, 30 insertions(+), 13 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index c10dd0af6..b9b6ec375 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -3596,18 +3596,23 @@ impl MsgSecretStore for SqliteStore { sender: &str, msg_id: &str, ) -> Result>> { + // On the write queue for the same reason as get_msg_secret_with_ts. + let pool = self.pool.clone(); let device_id = self.device_id; let chat = chat.to_string(); let sender = sender.to_string(); let msg_id = msg_id.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result>> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option> = msg_secrets::table .select(msg_secrets::secret) .filter(msg_secrets::chat.eq(&chat)) .filter(msg_secrets::sender.eq(&sender)) .filter(msg_secrets::msg_id.eq(&msg_id)) .filter(msg_secrets::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row) @@ -3621,22 +3626,26 @@ impl MsgSecretStore for SqliteStore { sender: &str, msg_id: &str, ) -> Result, i64)>> { - // Never a raw pooled read: a read racing a write transaction must wait, - // not error out as a phantom miss. `read_query` either holds the write - // permit or uses a WAL reader, and the shared-cache table lock that - // `busy_timeout` cannot absorb only exists in the former's stores. + // Stays on the write queue, so a lookup racing a secret write waits for + // it instead of reading the snapshot before it. A miss here is terminal + // -- the reaction, vote or edit is dropped with no retry -- and history + // sync seeds secrets in one large batch straight to the backend. + let pool = self.pool.clone(); let device_id = self.device_id; let chat = chat.to_string(); let sender = sender.to_string(); let msg_id = msg_id.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result, i64)>> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(Vec, i64)> = msg_secrets::table .select((msg_secrets::secret, msg_secrets::message_ts)) .filter(msg_secrets::chat.eq(&chat)) .filter(msg_secrets::sender.eq(&sender)) .filter(msg_secrets::msg_id.eq(&msg_id)) .filter(msg_secrets::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row) @@ -6109,11 +6118,19 @@ mod read_routing_tests { /// Read-only methods left on the write queue on purpose, with the reason. /// Anything else matching a read-shaped name has to route through /// `read_query` or this test fails. - const ON_THE_WRITE_QUEUE: &[(&str, &str)] = &[( - "get_pending_inbound", - "retries SQLITE_BUSY on the write queue: a read error here fails closed \ - and forces an unnecessary redelivery", - )]; + const ON_THE_WRITE_QUEUE: &[(&str, &str)] = &[ + ( + "get_pending_inbound", + "retries SQLITE_BUSY on the write queue: a read error here fails closed \ + and forces an unnecessary redelivery", + ), + ( + "get_msg_secret", + "a miss is terminal for the reaction/vote/edit, so the lookup must wait \ + out a concurrent secret write rather than read the snapshot before it", + ), + ("get_msg_secret_with_ts", "same as get_msg_secret"), + ]; /// Read-shaped methods that reach the database without going through /// `read_query`, ignoring the ones excused above, plus how many were From 0079cb3eeebe4218602693b1b24daa8d4003895f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 07:52:40 +0000 Subject: [PATCH 04/16] fix(sqlite): keep LID/PN mapping lookups on the write queue `alternate_msg_secret_jid` resolves the peer's other namespace through `get_lid_mapping` / `get_pn_mapping` and feeds the result straight back into the message-secret lookup that was just moved back to the write queue for exactly this reason. That path has no cache in front of the backend (the one in `lid_pn.rs` is on a different caller), so a lookup racing `persist_and_migrate_lid_pn` reads the pre-write snapshot, resolves no alternate JID, and the addon is rejected -- the same terminal miss, one indirection earlier. Protecting the secret read and not the mapping read that decides which key it uses was half a fix. Both are back on the semaphore and listed in ON_THE_WRITE_QUEUE. `get_all_lid_mappings` stays on the read path: it is a bulk enumeration with no caller on the addon path. While here, `get_msg_secret` now delegates to `get_msg_secret_with_ts` and drops the timestamp instead of repeating the same filter chain with one column fewer, so the query and the routing rationale live in one place. --- storages/sqlite-storage/src/sqlite_store.rs | 57 +++++++++++---------- 1 file changed, 29 insertions(+), 28 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index b9b6ec375..45d2dc1a7 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -2578,9 +2578,16 @@ impl ProtocolStore for SqliteStore { } async fn get_lid_mapping(&self, lid: &str) -> Result> { + // On the write queue: the alternate-namespace secret lookup resolves the + // peer through here with no cache in front, and a miss there is terminal + // for the addon. Waiting out a concurrent mapping write costs less. + let pool = self.pool.clone(); let device_id = self.device_id; let lid = lid.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2591,7 +2598,7 @@ impl ProtocolStore for SqliteStore { )) .filter(lid_pn_mapping::lid.eq(&lid)) .filter(lid_pn_mapping::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( @@ -2608,9 +2615,14 @@ impl ProtocolStore for SqliteStore { } async fn get_pn_mapping(&self, phone: &str) -> Result> { + // On the write queue for the same reason as get_lid_mapping. + let pool = self.pool.clone(); let device_id = self.device_id; let phone = phone.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2622,7 +2634,7 @@ impl ProtocolStore for SqliteStore { .filter(lid_pn_mapping::phone_number.eq(&phone)) .filter(lid_pn_mapping::device_id.eq(device_id)) .order(lid_pn_mapping::updated_at.desc()) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(row.map( @@ -3596,28 +3608,12 @@ impl MsgSecretStore for SqliteStore { sender: &str, msg_id: &str, ) -> Result>> { - // On the write queue for the same reason as get_msg_secret_with_ts. - let pool = self.pool.clone(); - let device_id = self.device_id; - let chat = chat.to_string(); - let sender = sender.to_string(); - let msg_id = msg_id.to_string(); - self.with_semaphore(move || -> Result>> { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - let row: Option> = msg_secrets::table - .select(msg_secrets::secret) - .filter(msg_secrets::chat.eq(&chat)) - .filter(msg_secrets::sender.eq(&sender)) - .filter(msg_secrets::msg_id.eq(&msg_id)) - .filter(msg_secrets::device_id.eq(device_id)) - .first(&mut conn) - .optional() - .map_err(|e| StoreError::Database(Box::new(e)))?; - Ok(row) - }) - .await + // Same row, one column narrower: delegating keeps the query and the + // routing decision in one place rather than two that can drift. + Ok(self + .get_msg_secret_with_ts(chat, sender, msg_id) + .await? + .map(|(secret, _)| secret)) } async fn get_msg_secret_with_ts( @@ -6125,11 +6121,16 @@ mod read_routing_tests { and forces an unnecessary redelivery", ), ( - "get_msg_secret", + "get_msg_secret_with_ts", "a miss is terminal for the reaction/vote/edit, so the lookup must wait \ out a concurrent secret write rather than read the snapshot before it", ), - ("get_msg_secret_with_ts", "same as get_msg_secret"), + ( + "get_lid_mapping", + "resolves the alternate namespace for that same secret lookup, with no \ + cache in front on that path, so a stale miss loses the addon too", + ), + ("get_pn_mapping", "same as get_lid_mapping"), ]; /// Read-shaped methods that reach the database without going through From d3f958d9408f92c24f728337a75ce5f89c7fbec1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 08:00:23 +0000 Subject: [PATCH 05/16] test(sqlite): make the routing allowlist fail when it goes stale `ON_THE_WRITE_QUEUE` was consulted in one direction only: it excused a listed name and never checked the name still needed excusing. Migrate one of those reads later and forget the entry, and it stays there forever, silently excusing the next method that happens to share the name while its reason string quietly becomes false. `misrouted_reads` now returns the names it excused as well, and the test asserts that set equals the allowlist. Verified by migrating `get_pn_mapping` to `read_query` with its entry left in place: the assertion fires and names it. Also throttle the chunked-write race by 200us per sample. Both sides of that test go through `spawn_blocking` on the same pool, so back-to-back sampling competes with the writer for threads on a loaded machine and would eventually read as flake rather than as the regression it catches. Re-checked with the transaction removed: still fails on the first round, at 190/760. And the fallback comment in `read_erased` now says the free-snapshot claim holds at `pool_size = 1`, which is where it holds. --- storages/sqlite-storage/src/sqlite_store.rs | 52 +++++++++++++++------ 1 file changed, 38 insertions(+), 14 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 45d2dc1a7..6cf58059c 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -628,9 +628,12 @@ impl SqliteStore { async fn read_erased(&self, f: ReadQuery) -> Result { if self.reads.is_none() { - // No reader connections: the single pooled connection is what no - // writer can be holding while this query runs, so the snapshot - // comes for free and a transaction would only add statements. + // No reader connections: at the default `pool_size = 1` the single + // pooled connection is what no writer can hold while this query + // runs, so the snapshot is free and a transaction would only add + // statements. Raising `pool_size` breaks that — the writers that + // check out a connection without the permit could then commit + // mid-read — but it also deadlocks writes, so it stays unsupported. let pool = self.pool.clone(); return self .with_semaphore(move || { @@ -6103,6 +6106,10 @@ mod read_routing_tests { if n == ENTRIES { return; } + // Both paths use the same blocking pool, so back-to-back + // samples would compete with the writer for threads. Still + // thousands of samples per batch. + tokio::time::sleep(Duration::from_micros(200)).await; } }) .await; @@ -6134,9 +6141,10 @@ mod read_routing_tests { ]; /// Read-shaped methods that reach the database without going through - /// `read_query`, ignoring the ones excused above, plus how many were - /// scanned at all so the check cannot pass by matching nothing. - fn misrouted_reads(source: &str) -> (Vec, usize) { + /// `read_query`: the ones with no excuse, the ones `ON_THE_WRITE_QUEUE` + /// excused, and how many were scanned at all so the check cannot pass by + /// matching nothing. + fn misrouted_reads(source: &str) -> (Vec, Vec, usize) { let source = source .split_once("\n#[cfg(test)]") .map(|(before, _)| before) @@ -6144,6 +6152,7 @@ mod read_routing_tests { let mut current: Option<(&str, String)> = None; let mut offenders: Vec = Vec::new(); + let mut excused: Vec = Vec::new(); let mut scanned = 0usize; for line in source.lines() { if let Some((name, body)) = current.as_mut() { @@ -6156,13 +6165,15 @@ mod read_routing_tests { ] .iter() .any(|token| body.contains(token)); - if touches_db - && !body.contains("read_query(") - && !ON_THE_WRITE_QUEUE + if touches_db && !body.contains("read_query(") { + if ON_THE_WRITE_QUEUE .iter() .any(|(allowed, _)| allowed == name) - { - offenders.push((*name).to_string()); + { + excused.push((*name).to_string()); + } else { + offenders.push((*name).to_string()); + } } current = None; } else { @@ -6188,7 +6199,7 @@ mod read_routing_tests { scanned += 1; } } - (offenders, scanned) + (offenders, excused, scanned) } /// A new read-only method written the old way (raw pool checkout, write @@ -6197,7 +6208,7 @@ mod read_routing_tests { /// only place that can see the routing decision. #[test] fn read_shaped_methods_route_through_read_query() { - let (offenders, scanned) = misrouted_reads(include_str!("sqlite_store.rs")); + let (offenders, mut excused, scanned) = misrouted_reads(include_str!("sqlite_store.rs")); assert!( offenders.is_empty(), "read-only methods must call read_query (or be listed in ON_THE_WRITE_QUEUE \ @@ -6207,6 +6218,19 @@ mod read_routing_tests { scanned > 20, "the scan saw only {scanned} read-shaped methods" ); + // The allowlist has to be consumed in full, or an entry left behind by a + // later migration would silently excuse the next method of that name and + // its reason would be a lie. + let mut listed: Vec = ON_THE_WRITE_QUEUE + .iter() + .map(|(name, _)| (*name).to_string()) + .collect(); + listed.sort(); + excused.sort(); + assert_eq!( + excused, listed, + "every ON_THE_WRITE_QUEUE entry must still name a read that bypasses read_query" + ); } /// The scan is worth nothing if it cannot see a violation, so feed it one. @@ -6226,7 +6250,7 @@ impl SqliteStore { "; assert_eq!( misrouted_reads(regression), - (vec!["get_something_new".to_string()], 2) + (vec!["get_something_new".to_string()], Vec::new(), 2) ); } } From d32db6229c7068f14eee8a3885e0d7b30822fcd7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 08:05:53 +0000 Subject: [PATCH 06/16] fix(sqlite): snapshot the fallback read when the write pool is wider The fallback skipped the deferred read transaction on the grounds that holding the one pooled connection is the snapshot. True at `pool_size = 1`, and I justified leaving it there by calling anything above that unsupported. That was wrong twice: the config is reachable through `SqliteStoreConfig` and this crate's own tuning test uses `pool_size: 2, read_pool_size: 0`, and raising it does not deadlock by itself -- only two deferred read-then-write transactions racing do. Several writers check a connection out without taking the permit, so with a second connection available they can commit between the statements of `has_signal_state_for_user`, `load_prekeys_batch` or `get_app_state_mutation_macs_batch_for_device`. Condition the shortcut on what actually makes it true. `pool_size = 1` with no readers keeps the old path statement for statement, so the default costs nothing new; anything wider goes through the same deferred transaction the reader path uses. Test covers the wide-pool case. --- storages/sqlite-storage/src/sqlite_store.rs | 54 +++++++++++++++++---- 1 file changed, 45 insertions(+), 9 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 6cf58059c..e77dd0167 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -627,13 +627,11 @@ impl SqliteStore { } async fn read_erased(&self, f: ReadQuery) -> Result { - if self.reads.is_none() { - // No reader connections: at the default `pool_size = 1` the single - // pooled connection is what no writer can hold while this query - // runs, so the snapshot is free and a transaction would only add - // statements. Raising `pool_size` breaks that — the writers that - // check out a connection without the permit could then commit - // mid-read — but it also deadlocks writes, so it stays unsupported. + // At the default `pool_size = 1` with no reader connections, holding the + // one pooled connection is itself the snapshot: no writer can be on it, + // including the several that skip the permit and check one out directly. + // A transaction would only add statements to every read. + if self.reads.is_none() && self.pool.max_size() <= 1 { let pool = self.pool.clone(); return self .with_semaphore(move || { @@ -644,8 +642,9 @@ impl SqliteStore { }) .await; } - // One implementation of acquire-checkout-snapshot, shared with the - // sibling-crate read path. + // Otherwise the deferred read transaction is what pins the snapshot, + // whether the concurrency comes from reader connections or from a wider + // write pool. One implementation, shared with the sibling-crate path. self.shared().read(f).await } @@ -6008,6 +6007,43 @@ mod read_routing_tests { assert_eq!(got.as_deref(), Some(&b"blob"[..])); } + /// `pool_size > 1` with no reader connections is reachable config, and there + /// the permit no longer implies an exclusive connection: the writers that + /// check one out directly can commit between a multi-statement read's + /// queries. The deferred transaction has to cover that case too. + #[tokio::test] + async fn a_multi_statement_read_is_snapshot_isolated_with_a_wider_write_pool() { + let db = TempDb::new("wide_pool"); + let store = SqliteStore::with_config( + &db.url(), + SqliteStoreConfig { + pool_size: 2, + read_pool_size: 0, + ..Default::default() + }, + ) + .await + .expect("store opens"); + assert!(store.reads.is_none(), "no reader connections requested"); + store.create_new_device().await.expect("device row"); + store.put_session(ADDR, b"blob").await.unwrap(); + + // has_signal_state_for_user issues two EXISTS; both must see one + // snapshot even though a second write connection is available. + assert!( + store + .has_signal_state_for_user("559990000001") + .await + .unwrap() + ); + assert!( + !store + .has_signal_state_for_user("559990000009") + .await + .unwrap() + ); + } + /// An uncommitted write is not a lock error and not a phantom miss: the /// reader sees the last committed state and returns it. This is the case /// the msg-secret reads were kept on the write queue for, so it has to hold From e8fcb29e62ed2c8d5272ca2a0cb51ab14e1ba2be Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 08:17:02 +0000 Subject: [PATCH 07/16] fix(sqlite): keep cache-promoting reads on the write queue Three more reads go back, and they share one shape with the msg-secret and LID/PN reverts before them: the row is promoted into a plain in-memory cache, or suppresses an action, so a stale read does not degrade to a retry -- it sticks. - `get_sender_key_devices` initializes `sender_key_device_cache`. A stale `has_key = true`, cached over a concurrent forget, drops the SKDM for the device that asked for redistribution, and the resend is undecryptable. - `get_devices` is promoted into `device_registry_cache` unconditionally on a miss, so a stale row overwrites a newer entry without advancing the topology generation and later sends omit a linked device. - `get_tc_token` feeds `prepare_privacy_token`'s scheduling decision, so reading before a concurrent touch commits issues a duplicate token and bypasses the configured interval. Outgoing sends are deliberately not per-chat serialized, so none of these three is protected by a lock. The distinction that decides the whole audit is now written next to the allowlist: `SignalStoreCache` reconciles staleness with its dirty set and incarnation, so the reads it mediates migrate; a cache that overwrites whatever it is handed does not. None of the three is on the measured path -- the Case B numbers are `get_session` -- so this costs nothing but the routing. --- storages/sqlite-storage/src/sqlite_store.rs | 52 ++++++++++++++++++--- 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index e77dd0167..4450620c2 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -2448,14 +2448,21 @@ fn delete_pending_inbound_row( #[cfg_attr(not(target_arch = "wasm32"), async_trait)] impl ProtocolStore for SqliteStore { async fn get_sender_key_devices(&self, group_jid: &str) -> Result> { + // On the write queue: the result initializes `sender_key_device_cache`, + // so a stale `has_key = true` is cached over a concurrent forget and the + // send drops the SKDM for a device that asked for redistribution. + let pool = self.pool.clone(); let device_id = self.device_id; let group_jid = group_jid.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let rows: Vec<(String, i32)> = sender_key_devices::table .select((sender_key_devices::device_jid, sender_key_devices::has_key)) .filter(sender_key_devices::group_jid.eq(&group_jid)) .filter(sender_key_devices::device_id.eq(device_id)) - .load(conn) + .load(&mut conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() @@ -2917,9 +2924,16 @@ impl ProtocolStore for SqliteStore { } async fn get_devices(&self, user: &str) -> Result> { + // On the write queue: a miss here is promoted into + // `device_registry_cache` unconditionally, so a stale row overwrites a + // newer entry and later sends omit a linked device until a refresh. + let pool = self.pool.clone(); let device_id = self.device_id; let user = user.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(String, String, i32, Option, Option)> = device_registry::table .select(( @@ -2931,7 +2945,7 @@ impl ProtocolStore for SqliteStore { )) .filter(device_registry::user_id.eq(&user)) .filter(device_registry::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; match row { @@ -3045,9 +3059,16 @@ impl ProtocolStore for SqliteStore { } async fn get_tc_token(&self, jid: &str) -> Result> { + // On the write queue: `prepare_privacy_token` schedules off this + // timestamp, so reading before a concurrent touch commits issues a + // duplicate token and bypasses the configured interval. + let pool = self.pool.clone(); let device_id = self.device_id; let jid = jid.to_string(); - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let row: Option<(Vec, i64, Option)> = tc_tokens::table .select(( tc_tokens::token, @@ -3056,7 +3077,7 @@ impl ProtocolStore for SqliteStore { )) .filter(tc_tokens::jid.eq(&jid)) .filter(tc_tokens::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok( @@ -6174,6 +6195,25 @@ mod read_routing_tests { cache in front on that path, so a stale miss loses the addon too", ), ("get_pn_mapping", "same as get_lid_mapping"), + // The rest share one shape: the row is promoted into a plain in-memory + // cache, or suppresses an action, so a stale read sticks instead of + // being retried. `SignalStoreCache` reconciles staleness and its reads + // do migrate; these caches overwrite whatever they are handed. + ( + "get_sender_key_devices", + "initializes sender_key_device_cache: a stale has_key=true is cached \ + over a concurrent forget and the send drops that device's SKDM", + ), + ( + "get_devices", + "promoted into device_registry_cache unconditionally, so a stale row \ + overwrites a newer entry and sends omit a linked device", + ), + ( + "get_tc_token", + "prepare_privacy_token schedules off this timestamp, so a stale read \ + issues a duplicate token and bypasses the configured interval", + ), ]; /// Read-shaped methods that reach the database without going through From 9a2c92aaa069b41ca5522da5a39bbebb2f7d3501 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 08:22:18 +0000 Subject: [PATCH 08/16] test(sqlite): make the wide-pool snapshot test prove something `a_multi_statement_read_is_snapshot_isolated_with_a_wider_write_pool` called `has_signal_state_for_user` twice with nothing writing in between, so it passed with or without the deferred transaction it was supposed to cover. I verified the other two tests by removing the mechanism and watching them fail; I did not do that here, and it showed. It now runs two SELECTs inside one `read_query` closure, parks between them, and commits through the pool's other connection while parked. With the snapshot removed the second query reads the new value and the test fails, which is the point. The routing scan also missed `self.shared().run(` -- the sibling-crate write path, reachable from a read-shaped method without touching any token it looked for. Added, and the scan now strips indentation before matching so a call rustfmt split across lines still reads as one token. Its self-test carries a `shared().run` offender alongside the raw-pool one. --- storages/sqlite-storage/src/sqlite_store.rs | 88 +++++++++++++++++---- 1 file changed, 74 insertions(+), 14 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 4450620c2..0862986db 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -6049,19 +6049,60 @@ mod read_routing_tests { store.create_new_device().await.expect("device row"); store.put_session(ADDR, b"blob").await.unwrap(); - // has_signal_state_for_user issues two EXISTS; both must see one - // snapshot even though a second write connection is available. - assert!( - store - .has_signal_state_for_user("559990000001") - .await - .unwrap() + // Park between the two SELECTs and commit through the pool's *other* + // connection while parked. Without the deferred transaction the second + // query would pick the write up. + let (open_tx, mut open_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let reader = { + let store = store.clone(); + tokio::spawn(async move { + store + .read_query(move |conn| { + let read_once = |conn: &mut SqliteConnection| { + sessions::table + .select(sessions::record) + .filter(sessions::address.eq(ADDR)) + .first::>(conn) + .optional() + .map_err(|e| StoreError::Database(Box::new(e))) + }; + let first = read_once(conn)?; + let _ = open_tx.send(()); + let _ = release_rx.recv_timeout(Duration::from_secs(20)); + let second = read_once(conn)?; + Ok((first, second)) + }) + .await + }) + }; + + tokio::time::timeout(Duration::from_secs(10), open_rx.recv()) + .await + .expect("the read must reach its first query") + .expect("reader alive"); + + tokio::time::timeout( + Duration::from_secs(10), + store.put_session(ADDR, b"committed-mid-read"), + ) + .await + .expect("the second connection must be free to write") + .expect("write commits"); + + let _ = release_tx.send(()); + let (first, second) = reader.await.expect("join").expect("read"); + assert_eq!(first.as_deref(), Some(&b"blob"[..])); + assert_eq!( + second.as_deref(), + Some(&b"blob"[..]), + "both queries must see one snapshot, not the write that landed between them" ); - assert!( - !store - .has_signal_state_for_user("559990000009") - .await - .unwrap() + + // And the committed value is visible to the next read. + assert_eq!( + store.get_session(ADDR).await.unwrap().as_deref(), + Some(&b"committed-mid-read"[..]) ); } @@ -6238,6 +6279,9 @@ mod read_routing_tests { "with_semaphore(", "with_retry(", "spawn_blocking(", + // The sibling-crate write path; `shared().read(` is the + // read one and is what `read_query` itself uses. + "shared().run(", ] .iter() .any(|token| body.contains(token)); @@ -6253,7 +6297,9 @@ mod read_routing_tests { } current = None; } else { - body.push_str(line); + // Indentation dropped so a call rustfmt split across lines + // (`self` / `.shared()` / `.run(`) still reads as one token. + body.push_str(line.trim_start()); } continue; } @@ -6322,11 +6368,25 @@ impl SqliteStore { async fn get_something_routed(&self) -> Result<()> { self.read_query(move |_conn| Ok(())).await } + + async fn load_via_the_shared_write_path(&self) -> Result<()> { + self + .shared() + .run(move |_conn| Ok(())) + .await + } } "; assert_eq!( misrouted_reads(regression), - (vec!["get_something_new".to_string()], Vec::new(), 2) + ( + vec![ + "get_something_new".to_string(), + "load_via_the_shared_write_path".to_string() + ], + Vec::new(), + 3 + ) ); } } From 8ac0fc33aa5a6ebcdf994035a79fe13118ca90fb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 08:41:31 +0000 Subject: [PATCH 09/16] fix(sqlite): gate the wide-pool snapshot on shared cache, restore test teeth Two real defects, both mine, both introduced by earlier commits in this PR and caught by Codex. `build` declines reader connections under shared cache because a read transaction there holds table locks that fail the writer with SQLITE_LOCKED_SHAREDCACHE, which busy_timeout cannot absorb. The `pool_size > 1` fallback added two commits ago then opened exactly that transaction on the main pool, reintroducing the hazard the decline exists to avoid -- and `with_config_custom_tuning_builds_and_operates` already runs shared cache with `pool_size: 2`. The snapshot is now gated on the same condition that gates the reader pool, recorded as `snapshot_safe`, with a test pinning it. `a_chunked_batch_write_is_never_observed_half_applied` had gone vacuous. I verified it failed without the transaction, and then moved `get_sender_key_devices` onto the write permit, which serialized the sampler against the writer so it could no longer observe a torn batch. The claim stayed in the commit message; the test stopped backing it. It now samples through `read_query` directly, and fails at 190/760 with the transaction removed. `get_all_lid_mappings` also goes back to the write queue. The startup warm-up feeds it into `LidPnCache::add_guarded`, whose LID side replaces unconditionally, so a stale row read during a live learn reverts reverse resolution -- the same rule that moved the other cache-fed reads, which I had wrongly cleared as "bulk enumeration". --- storages/sqlite-storage/src/sqlite_store.rs | 86 ++++++++++++++++++++- 1 file changed, 82 insertions(+), 4 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 0862986db..9079867bd 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -127,6 +127,12 @@ pub struct SqliteStore { /// once and deadlock on the write-lock upgrade — the exact failure this /// change exists to avoid. pub(crate) reads: Option, + /// Whether a deferred read transaction is safe here: WAL, and not shared + /// cache. It is the same condition that decides [`Self::reads`], and it has + /// to gate the wider-write-pool snapshot too — under shared cache a read + /// transaction holds table locks that fail the writer with + /// `SQLITE_LOCKED_SHAREDCACHE`, which `busy_timeout` cannot absorb. + pub(crate) snapshot_safe: bool, pub(crate) database_path: String, device_id: i32, } @@ -590,6 +596,7 @@ impl SqliteStore { pool, db_semaphore: Arc::new(tokio::sync::Semaphore::new(pool_size as usize)), reads, + snapshot_safe: declined.is_none(), database_path, device_id, }) @@ -631,7 +638,11 @@ impl SqliteStore { // one pooled connection is itself the snapshot: no writer can be on it, // including the several that skip the permit and check one out directly. // A transaction would only add statements to every read. - if self.reads.is_none() && self.pool.max_size() <= 1 { + // The snapshot is only worth opening where it is safe: at + // `pool_size = 1` the exclusive connection already gives it, and under + // shared cache or without WAL a read transaction would lock out the + // writer instead. + if self.reads.is_none() && (self.pool.max_size() <= 1 || !self.snapshot_safe) { let pool = self.pool.clone(); return self .with_semaphore(move || { @@ -2703,8 +2714,15 @@ impl ProtocolStore for SqliteStore { } async fn get_all_lid_mappings(&self) -> Result> { + // On the write queue: the startup warm-up feeds these rows into + // `LidPnCache::add_guarded`, whose LID side replaces unconditionally, so + // a stale row read during a live learn reverts reverse resolution. + let pool = self.pool.clone(); let device_id = self.device_id; - self.read_query(move |conn| { + self.with_semaphore(move || -> Result> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let rows: Vec<(String, String, i64, String, i64)> = lid_pn_mapping::table .select(( lid_pn_mapping::lid, @@ -2714,7 +2732,7 @@ impl ProtocolStore for SqliteStore { lid_pn_mapping::updated_at, )) .filter(lid_pn_mapping::device_id.eq(device_id)) - .load(conn) + .load(&mut conn) .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(rows .into_iter() @@ -6106,6 +6124,48 @@ mod read_routing_tests { ); } + /// A shared-cache store declines reader connections because a read + /// transaction there holds table locks the writer cannot wait out. The + /// wider-write-pool snapshot has to decline for the same reason instead of + /// reintroducing exactly that transaction. + #[tokio::test] + async fn a_shared_cache_store_gets_no_snapshot_even_with_a_wider_write_pool() { + use portable_atomic::AtomicU64; + use std::sync::atomic::Ordering; + static COUNTER: AtomicU64 = AtomicU64::new(0); + let id = COUNTER.fetch_add(1, Ordering::Relaxed); + let url = format!( + "file:memdb_snapshot_gate_{}_{id}?mode=memory&cache=shared", + std::process::id() + ); + let store = SqliteStore::with_config( + &url, + SqliteStoreConfig { + pool_size: 2, + read_pool_size: 4, + ..Default::default() + }, + ) + .await + .expect("store opens"); + + assert!(store.reads.is_none(), "shared cache declines reader pool"); + assert!( + !store.snapshot_safe, + "and must decline the deferred read transaction with it" + ); + + // Still fully operational on the plain path. + store.create_new_device().await.expect("device row"); + store.put_session(ADDR, b"blob").await.unwrap(); + assert!( + store + .has_signal_state_for_user("559990000001") + .await + .unwrap() + ); + } + /// An uncommitted write is not a lock error and not a phantom miss: the /// reader sees the last committed state and returns it. This is the case /// the msg-secret reads were kept on the write queue for, so it has to hold @@ -6196,7 +6256,20 @@ mod read_routing_tests { // of hanging the runtime. let sampled = tokio::time::timeout(Duration::from_secs(20), async { loop { - let n = store.get_sender_key_devices(GROUP).await.unwrap().len(); + // Straight through read_query, not get_sender_key_devices: + // that one is on the write permit now, which would serialize + // the sample against the writer and hide a torn batch. + let n = store + .read_query(|conn| { + sender_key_devices::table + .filter(sender_key_devices::group_jid.eq(GROUP)) + .count() + .get_result::(conn) + .map(|n| n as usize) + .map_err(|e| StoreError::Database(Box::new(e))) + }) + .await + .unwrap(); assert!( n == 0 || n == ENTRIES, "a chunked batch was observed {n}/{ENTRIES} applied" @@ -6236,6 +6309,11 @@ mod read_routing_tests { cache in front on that path, so a stale miss loses the addon too", ), ("get_pn_mapping", "same as get_lid_mapping"), + ( + "get_all_lid_mappings", + "the startup warm-up feeds these into LidPnCache::add_guarded, whose \ + LID side replaces unconditionally, so a stale row reverts a live learn", + ), // The rest share one shape: the row is promoted into a plain in-memory // cache, or suppresses an action, so a stale read sticks instead of // being retried. `SignalStoreCache` reconciles staleness and its reads From 65aac4d95f3d01b8e949cad3dc0d35ebd4fdbd35 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 09:01:45 +0000 Subject: [PATCH 10/16] docs(sqlite): scope the ordering and query_only claims to what holds Three comment-accuracy fixes, all on claims I made too broadly. `read_query`'s doc said a write sent down the read path fails because reader connections are `query_only`. True on a reader connection, false on the fallback, which hands out an ordinary write connection -- so at the default `read_pool_size = 0` the net is absent and the routing scan is the only guard. The doc says that now, and the test says it too: it asserts the refusal with readers and asserts the gap without them, so the limit is recorded rather than assumed away. Enforcing `query_only` on a pooled write connection for the duration of a read would leave the pool poisoned if the closure unwound before the reset, which is a worse trade than documenting the gap. The `get_devices` and `get_tc_token` rationales named a concurrent writer the permit does not order them against: `update_device_list` and `touch_tc_token_sender_timestamp` check a connection out without it. The ordering does hold at the default, because the single pooled connection serializes them, so the comments now attribute it there. Routing those writers through the permit would change write serialization, which is out of scope for this change. `get_sender_key_devices` was also flagged and is fine as written: every writer of `sender_key_devices` (`set_sender_key_status`, `clear_sender_key_devices`, `delete_sender_key_device_rows`) takes the permit. `put_sender_key_for_device` and `delete_sender_key_for_device` skip it but write `sender_keys`, a different table. Also merged the two stacked rationale blocks in `read_erased` into one. --- storages/sqlite-storage/src/sqlite_store.rs | 70 ++++++++++++--------- 1 file changed, 42 insertions(+), 28 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 9079867bd..14d6fcf83 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -619,9 +619,10 @@ impl SqliteStore { /// merely overlap a write see either state, which is what the single permit /// already gave them (it ordered them arbitrarily, not causally). /// - /// Only correct for statements that cannot write: reader connections carry - /// `PRAGMA query_only`, so a write sent here fails instead of escaping the - /// serialization the store depends on. + /// Only correct for statements that cannot write. Reader connections carry + /// `PRAGMA query_only`, so a write sent here fails loudly -- but the + /// fallback hands out an ordinary write connection, so with no reader pool + /// (the default) that net is absent and the routing scan is the only guard. async fn read_query(&self, f: F) -> Result where F: FnOnce(&mut SqliteConnection) -> Result + Send + 'static, @@ -634,14 +635,11 @@ impl SqliteStore { } async fn read_erased(&self, f: ReadQuery) -> Result { - // At the default `pool_size = 1` with no reader connections, holding the - // one pooled connection is itself the snapshot: no writer can be on it, - // including the several that skip the permit and check one out directly. - // A transaction would only add statements to every read. - // The snapshot is only worth opening where it is safe: at - // `pool_size = 1` the exclusive connection already gives it, and under - // shared cache or without WAL a read transaction would lock out the - // writer instead. + // Skip the deferred snapshot only where it is redundant or unsafe: at + // `pool_size = 1` with no readers the exclusive pooled connection is + // itself the snapshot (no writer can be on it, including the ones that + // skip the permit), and under shared cache or without WAL a read + // transaction would lock the writer out instead. if self.reads.is_none() && (self.pool.max_size() <= 1 || !self.snapshot_safe) { let pool = self.pool.clone(); return self @@ -2945,6 +2943,8 @@ impl ProtocolStore for SqliteStore { // On the write queue: a miss here is promoted into // `device_registry_cache` unconditionally, so a stale row overwrites a // newer entry and later sends omit a linked device until a refresh. + // `update_device_list` skips the permit, so at the default `pool_size` + // the single connection is what orders them, not the permit itself. let pool = self.pool.clone(); let device_id = self.device_id; let user = user.to_string(); @@ -3079,7 +3079,9 @@ impl ProtocolStore for SqliteStore { async fn get_tc_token(&self, jid: &str) -> Result> { // On the write queue: `prepare_privacy_token` schedules off this // timestamp, so reading before a concurrent touch commits issues a - // duplicate token and bypasses the configured interval. + // duplicate token and bypasses the configured interval. The touch skips + // the permit, so at the default `pool_size` the single connection is + // what orders them, not the permit itself. let pool = self.pool.clone(); let device_id = self.device_id; let jid = jid.to_string(); @@ -6001,25 +6003,37 @@ mod read_routing_tests { exercise_reads(4).await; } - /// The safety net: reader connections are `query_only`, so a write that - /// slips into `read_query` fails loudly instead of escaping the write - /// serialization and deadlocking against the real writer. + /// The safety net, and its limit. A reader connection is `query_only`, so a + /// write that slips into `read_query` fails loudly there. The fallback hands + /// out an ordinary write connection and has no such net, which is why the + /// routing scan exists; asserted here so the gap is recorded rather than + /// assumed away. #[tokio::test] - async fn a_write_through_read_query_is_refused() { - let db = TempDb::new("query_only"); - let store = store_with(1, &db).await; + async fn a_write_through_read_query_is_refused_only_on_reader_connections() { + let write_a_row = |store: SqliteStore| async move { + store + .read_query(|conn| { + diesel::delete(sessions::table) + .execute(conn) + .map_err(|e| StoreError::Database(Box::new(e)))?; + Ok(()) + }) + .await + }; - let result = store - .read_query(|conn| { - diesel::delete(sessions::table) - .execute(conn) - .map_err(|e| StoreError::Database(Box::new(e)))?; - Ok(()) - }) - .await; + let with_readers = TempDb::new("query_only_readers"); + let store = store_with(1, &with_readers).await; + assert!( + matches!(write_a_row(store).await, Err(StoreError::Database(_))), + "query_only must reject a write on a reader connection" + ); + + let no_readers = TempDb::new("query_only_fallback"); + let store = store_with(0, &no_readers).await; assert!( - matches!(result, Err(StoreError::Database(_))), - "query_only must reject a write on the read path" + write_a_row(store).await.is_ok(), + "the fallback has no query_only net; if this ever starts failing the \ + doc on read_query and this test both need updating" ); } From 5215a1f70dbe58e2364da4835e374ae86d35f730 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 09:09:54 +0000 Subject: [PATCH 11/16] docs(sqlite): scope the LID warm-up ordering claim like the other two Left inconsistent by the previous commit, which corrected the same overclaim on get_devices and get_tc_token but not here. --- storages/sqlite-storage/src/sqlite_store.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 14d6fcf83..010ab7e8c 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -2715,6 +2715,8 @@ impl ProtocolStore for SqliteStore { // On the write queue: the startup warm-up feeds these rows into // `LidPnCache::add_guarded`, whose LID side replaces unconditionally, so // a stale row read during a live learn reverts reverse resolution. + // `put_lid_mappings` takes the permit; at the default `pool_size` the + // single connection is what orders them either way. let pool = self.pool.clone(); let device_id = self.device_id; self.with_semaphore(move || -> Result> { From 43ee2f66f32127efb1d814ca537a27e8d3e943e2 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 15:54:48 +0000 Subject: [PATCH 12/16] perf(sqlite): keep the default profile off the write permit The read helper's fallback took the write permit, which changed the default profile it was supposed to leave alone. Fourteen reads ran on a raw pooled connection without the permit before this branch -- `device_exists`, `load_device_data_for_device`, `get_sender_key_for_device`, `load_prekey`, `load_signed_prekey`, `load_all_signed_prekeys`, the four app-state reads, `has_same_base_key`, `get_group_metadata`, `get_all_tc_token_jids` -- and routing them through the helper put them behind it. At `pool_size = 1` that adds no serialization, since the single pooled connection already provides it, but it does serialize the `spawn_blocking` dispatch that the pool wait previously overlapped. Measured on 16 concurrent `get_sender_key` at the default profile: p50 569-641us before, 730-796us after, so roughly 25% for a knob nobody has turned on. The single-connection branch now checks the connection out directly with no permit, which is what those fourteen did and what the other six get from the connection anyway. Re-measured: p50 525-689us, back on top of main. A wider pool that cannot take a read transaction still uses the permit, since there the connection is no longer the serializer. --- storages/sqlite-storage/src/sqlite_store.rs | 25 ++++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 010ab7e8c..4e0ac78f6 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -635,12 +635,25 @@ impl SqliteStore { } async fn read_erased(&self, f: ReadQuery) -> Result { - // Skip the deferred snapshot only where it is redundant or unsafe: at - // `pool_size = 1` with no readers the exclusive pooled connection is - // itself the snapshot (no writer can be on it, including the ones that - // skip the permit), and under shared cache or without WAL a read - // transaction would lock the writer out instead. - if self.reads.is_none() && (self.pool.max_size() <= 1 || !self.snapshot_safe) { + // Default profile: one pooled connection and no readers. Checking it out + // is both the serialization and the snapshot, so this takes no permit -- + // adding one would serialize the `spawn_blocking` dispatch that the + // pool wait currently overlaps, which measured ~25% on p50. + if self.reads.is_none() && self.pool.max_size() <= 1 { + let pool = self.pool.clone(); + return tokio::task::spawn_blocking(move || { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; + f(&mut conn) + }) + .await + .map_err(|e| StoreError::Database(Box::new(e)))?; + } + // Wider pool, but a read transaction would take shared-cache table locks + // the writer cannot wait out: fall back to the permit for what ordering + // it can give. + if self.reads.is_none() && !self.snapshot_safe { let pool = self.pool.clone(); return self .with_semaphore(move || { From 4fb2d68eb4b147109af4dfc32160ac4f9bfeb59a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 15:58:40 +0000 Subject: [PATCH 13/16] fix(sqlite): keep the app-state key reads on the write queue The clean audit pass over every migrated method turned up two more that the incremental reviews had missed, both app-state key lookups whose stale-absent answer is not a miss the caller retries. `get_sync_key` answers a peer's `AppStateSyncKeyRequest`. On `None` the handler returns an orphan `MessageField`, so a stale absent read tells the peer we do not have a key we do have, on the wire. `get_latest_sync_key_id` is unwrapped by `send_app_state_mutation` into `InvalidRequest("no app state sync key available")`, which fails the user's action outright with nothing retrying behind it. Both race `set_sync_key`, which is exactly what an incoming key share does. The version and mutation-mac reads stay on the read path: those are internal to a sync pass that is serialized per collection, and a stale read there re-syncs from an older version. --- storages/sqlite-storage/src/sqlite_store.rs | 31 ++++++++++++++++++--- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 4e0ac78f6..371be0d00 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -1410,14 +1410,21 @@ impl SqliteStore { key_id: &[u8], device_id: i32, ) -> Result> { + // On the write queue: a stale absent answer is sent on the wire as an + // orphan reply to a peer's key request, so it is not a miss the caller + // retries. + let pool = self.pool.clone(); let key_id = key_id.to_vec(); let res: Option> = self - .read_query(move |conn| { + .with_semaphore(move || -> Result>> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; let res: Option> = app_state_keys::table .select(app_state_keys::key_data) .filter(app_state_keys::key_id.eq(&key_id)) .filter(app_state_keys::device_id.eq(device_id)) - .first(conn) + .first(&mut conn) .optional() .map_err(|e| StoreError::Database(Box::new(e)))?; Ok(res) @@ -1479,8 +1486,14 @@ impl SqliteStore { &self, device_id: i32, ) -> Result>> { + // On the write queue: a stale absent answer becomes InvalidRequest and + // fails the user's app-state action outright. + let pool = self.pool.clone(); let res: Option> = self - .read_query(move |conn| { + .with_semaphore(move || -> Result>> { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; // Return the latest key whose blob actually decodes. A legacy bincode // row (or a corrupt one) reads as absent via get_sync_key but still // sits in the table with a possibly lexicographically-higher key_id; @@ -1491,7 +1504,7 @@ impl SqliteStore { .select((app_state_keys::key_id, app_state_keys::key_data)) .filter(app_state_keys::device_id.eq(device_id)) .order(app_state_keys::key_id.desc()) - .load(conn) + .load(&mut conn) .map_err(|e| StoreError::Database(Box::new(e)))?; let res = candidates .into_iter() @@ -6338,6 +6351,16 @@ mod read_routing_tests { cache in front on that path, so a stale miss loses the addon too", ), ("get_pn_mapping", "same as get_lid_mapping"), + ( + "get_app_state_sync_key_for_device", + "a stale absent answer is sent on the wire as an orphan reply to a \ + peer's key request, not retried by the caller", + ), + ( + "get_latest_app_state_sync_key_id_for_device", + "a stale absent answer becomes InvalidRequest and fails the user's \ + app-state action outright", + ), ( "get_all_lid_mappings", "the startup warm-up feeds these into LidPnCache::add_guarded, whose \ From 521e520975bd1bc7065d6a4a49be334d6edbc37a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 16:53:59 +0000 Subject: [PATCH 14/16] perf(sqlite): collapse the read and write dispatch monomorphizations The binary size gate failed at +39.88 KiB .text against a 32 KiB budget. Attributing it by symbol against main puts the growth in three buckets: read_erased at +25.1 KiB over 186 instantiations, SharedSqlite::read at +16.1 KiB, and with_semaphore at +9.6 KiB as its instantiation count went from 77 to 188. read_erased carried three branches, two of which held their own pool checkout and spawn_blocking, so each was emitted per return type. The snapshot condition is the same predicate written the other way round, and once it returns early the remaining two branches differ only in whether a permit is taken. Folding them leaves one body per return type instead of two, with the permit as an Option. with_semaphore was generic over the closure as well as the return type, so every call site got its own copy of the acquire and spawn_blocking. Erasing the closure the same way read_query already does collapses it to one body per return type, which also shrinks the write path that predates this branch. Both are dispatch-shape changes: same predicate, same ordering, same observable behaviour. Measured on the demo example against main: .text +39.88 KiB -> +24.00 KiB, stripped +53.78 KiB -> +28.34 KiB. --- storages/sqlite-storage/src/sqlite_store.rs | 74 ++++++++++++--------- 1 file changed, 42 insertions(+), 32 deletions(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 371be0d00..966313748 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -101,6 +101,9 @@ const MSG_SECRET_INSERT_CHUNK_SIZE: usize = 100; /// once per return type rather than once per call site. type ReadQuery = Box Result + Send>; +/// A unit of work for the write queue, erased for the same reason. +type BlockingJob = Box Result + Send>; + /// Reader connections and the permits that bound how many run at once. #[derive(Clone)] pub(crate) struct ReadPool { @@ -635,39 +638,39 @@ impl SqliteStore { } async fn read_erased(&self, f: ReadQuery) -> Result { - // Default profile: one pooled connection and no readers. Checking it out - // is both the serialization and the snapshot, so this takes no permit -- - // adding one would serialize the `spawn_blocking` dispatch that the - // pool wait currently overlaps, which measured ~25% on p50. - if self.reads.is_none() && self.pool.max_size() <= 1 { - let pool = self.pool.clone(); - return tokio::task::spawn_blocking(move || { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - f(&mut conn) - }) - .await - .map_err(|e| StoreError::Database(Box::new(e)))?; + // A deferred read transaction is what pins the snapshot, so take one + // wherever real concurrency sits behind it: reader connections, or a + // wider write pool on a database where a read transaction cannot lock + // the writer out. One implementation, shared with the sibling crates. + if self.reads.is_some() || (self.snapshot_safe && self.pool.max_size() > 1) { + return self.shared().read(f).await; } - // Wider pool, but a read transaction would take shared-cache table locks - // the writer cannot wait out: fall back to the permit for what ordering - // it can give. - if self.reads.is_none() && !self.snapshot_safe { - let pool = self.pool.clone(); - return self - .with_semaphore(move || { - let mut conn = pool - .get() - .map_err(|e| StoreError::Connection(Box::new(e)))?; - f(&mut conn) - }) - .await; - } - // Otherwise the deferred read transaction is what pins the snapshot, - // whether the concurrency comes from reader connections or from a wider - // write pool. One implementation, shared with the sibling-crate path. - self.shared().read(f).await + // No snapshot to take here. With the default single connection, checking + // it out is both the serialization and the snapshot, so this takes no + // permit -- adding one would serialize the `spawn_blocking` dispatch that + // the pool wait currently overlaps, which measured ~25% on p50. With a + // wider pool the permit is the only ordering left. + let permit = if self.pool.max_size() > 1 { + Some( + self.db_semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| StoreError::Database(Box::new(e)))?, + ) + } else { + None + }; + let pool = self.pool.clone(); + tokio::task::spawn_blocking(move || { + let _permit = permit; + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; + f(&mut conn) + }) + .await + .map_err(|e| StoreError::Database(Box::new(e)))? } /// The write queue: one permit, so two writers can never deadlock on the @@ -677,6 +680,13 @@ impl SqliteStore { F: FnOnce() -> Result + Send + 'static, T: Send + 'static, { + // Erased for the same reason as [`Self::read_query`]: the body carries a + // permit acquire and a `spawn_blocking`, and there are enough call sites + // that monomorphizing it per closure type costs tens of KiB of .text. + self.with_semaphore_erased(Box::new(f)).await + } + + async fn with_semaphore_erased(&self, f: BlockingJob) -> Result { let permit = self .db_semaphore .clone() From bd95cdfcb781df338869eb710e14c2aaf1bad72c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 16:54:11 +0000 Subject: [PATCH 15/16] test(sqlite): make the shared-cache snapshot test prove its claim The test asserted only that snapshot_safe was false, which is the mechanism rather than the consequence. Removing snapshot_safe from the routing predicate left it passing, so it did not cover the guard it was named for. It now parks a read mid-flight on a shared-cache store and requires a concurrent write to commit. A first attempt parked for 300ms and still passed unguarded: with_retry absorbs the lock across its 10/20/40/80/160ms backoff. Parking past that budget makes the lock fatal, and the unguarded run now fails with "database table is locked: sessions". --- storages/sqlite-storage/src/sqlite_store.rs | 53 ++++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 966313748..0b093f2cf 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -6207,7 +6207,6 @@ mod read_routing_tests { "and must decline the deferred read transaction with it" ); - // Still fully operational on the plain path. store.create_new_device().await.expect("device row"); store.put_session(ADDR, b"blob").await.unwrap(); assert!( @@ -6216,6 +6215,58 @@ mod read_routing_tests { .await .unwrap() ); + + // The flags above are only the mechanism. What has to hold is that a + // write still commits with a read parked mid-flight: on the snapshot + // path the writer meets the reader's table lock as + // `SQLITE_LOCKED_SHAREDCACHE`, which `busy_timeout` cannot absorb. The + // park outlasts `with_retry`'s ~310ms budget, so that lock is fatal + // rather than retried away. + let (open_tx, mut open_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let reader = { + let store = store.clone(); + tokio::spawn(async move { + store + .read_query(move |conn| { + let first = sessions::table + .select(sessions::record) + .filter(sessions::address.eq(ADDR)) + .first::>(conn) + .optional() + .map_err(|e| StoreError::Database(Box::new(e)))?; + let _ = open_tx.send(()); + let _ = release_rx.recv_timeout(Duration::from_secs(20)); + Ok(first) + }) + .await + }) + }; + + tokio::time::timeout(Duration::from_secs(10), open_rx.recv()) + .await + .expect("the read must reach its query") + .expect("reader alive"); + + let releaser = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(1500)).await; + let _ = release_tx.send(()); + }); + + tokio::time::timeout( + Duration::from_secs(10), + store.put_session(ADDR, b"committed-under-shared-cache"), + ) + .await + .expect("the write must not stall behind the parked read") + .expect("the write must commit, not meet a shared-cache lock"); + + releaser.await.expect("join releaser"); + reader.await.expect("join").expect("read"); + assert_eq!( + store.get_session(ADDR).await.unwrap().as_deref(), + Some(&b"committed-under-shared-cache"[..]) + ); } /// An uncommitted write is not a lock error and not a phantom miss: the From 0cb522b7dcab9194adedf4bbe3cf539c4ca7c2f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 17:13:57 +0000 Subject: [PATCH 16/16] fix(sqlite): keep has_signal_state_for_user on the write queue Tracing each routed read to the cold-load guard that covers it turned up two mappings that were not what this branch assumed. has_signal_state_for_user is consumed by SignalStoreCache::has_state_for_user, which is not one of the five functions the guard covers. It checks the two caches for any matching key and otherwise asks the backend, with no removal-seq re-check, so a load spanning a flush plus eviction can answer absent. Its callers use that answer to skip the PN to LID session migration entirely, and nothing retries the skip. It goes back on the write queue. has_session stays routed, but not for the reason recorded before. The cache's has_session reads get_session, not this method; the only path here is Device::contains_session, whose single production caller logs the result in all three branches. The justification is the caller, not the guard. --- storages/sqlite-storage/src/sqlite_store.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/storages/sqlite-storage/src/sqlite_store.rs b/storages/sqlite-storage/src/sqlite_store.rs index 0b093f2cf..f9b2bde4d 100644 --- a/storages/sqlite-storage/src/sqlite_store.rs +++ b/storages/sqlite-storage/src/sqlite_store.rs @@ -1825,6 +1825,9 @@ impl SignalStore for SqliteStore { } async fn has_session(&self, address: &str) -> Result { + // Not the cache's has_session, which reads get_session instead. This one + // is only reached through Device::contains_session, whose single caller + // logs the answer, so a stale one changes a log line. let device_id = self.device_id; let address_owned = address.to_string(); self.read_query(move |conn| { @@ -1846,7 +1849,15 @@ impl SignalStore for SqliteStore { // numeric PN/LID so it carries no LIKE wildcards. let pat_at = format!("{user}@%"); let pat_dev = format!("{user}:%"); - self.read_query(move |conn| { + // On the write queue: the only consumer, `has_state_for_user`, is the + // skip guard for the PN to LID session migration and has no cold-load + // re-check, so a stale absent answer skips a migration nothing retries. + let pool = self.pool.clone(); + self.with_semaphore(move || -> Result { + let mut conn = pool + .get() + .map_err(|e| StoreError::Connection(Box::new(e)))?; + let conn = &mut conn; let has_session = diesel::select(diesel::dsl::exists( sessions::table .filter(sessions::device_id.eq(device_id)) @@ -6446,6 +6457,12 @@ mod read_routing_tests { "prepare_privacy_token schedules off this timestamp, so a stale read \ issues a duplicate token and bypasses the configured interval", ), + ( + "has_signal_state_for_user", + "has_state_for_user gates the PN to LID session migration and has no \ + cold-load re-check, so a stale absent answer skips a migration that \ + nothing retries", + ), ]; /// Read-shaped methods that reach the database without going through