From d1e0bfd16855af3b3bae45ac4e3bc9555e901268 Mon Sep 17 00:00:00 2001 From: CodeGhost21 <164498022+CodeGhost21@users.noreply.github.com> Date: Wed, 20 May 2026 08:20:53 +0530 Subject: [PATCH] fix(whatsapp): retry WhatsApp SQLite writes on database-is-locked (#2077) (#2098) Co-authored-by: Cursor --- src/openhuman/whatsapp_data/mod.rs | 1 + src/openhuman/whatsapp_data/sqlite_retry.rs | 135 ++++++++++++++++++++ src/openhuman/whatsapp_data/store.rs | 67 +++++++++- src/openhuman/whatsapp_data/store_tests.rs | 122 ++++++++++++++++++ 4 files changed, 322 insertions(+), 3 deletions(-) create mode 100644 src/openhuman/whatsapp_data/sqlite_retry.rs diff --git a/src/openhuman/whatsapp_data/mod.rs b/src/openhuman/whatsapp_data/mod.rs index 8ca2589bb..165616d75 100644 --- a/src/openhuman/whatsapp_data/mod.rs +++ b/src/openhuman/whatsapp_data/mod.rs @@ -19,6 +19,7 @@ pub mod global; pub mod ops; pub mod rpc; mod schemas; +mod sqlite_retry; pub mod store; pub mod types; diff --git a/src/openhuman/whatsapp_data/sqlite_retry.rs b/src/openhuman/whatsapp_data/sqlite_retry.rs new file mode 100644 index 000000000..dbaf91a81 --- /dev/null +++ b/src/openhuman/whatsapp_data/sqlite_retry.rs @@ -0,0 +1,135 @@ +//! SQLite busy/locked detection and retry-with-backoff for WhatsApp data writes. +//! +//! Modelled on [`crate::openhuman::memory::tree::jobs::worker::is_sqlite_busy`] — +//! the configured `busy_timeout` absorbs short waits inside rusqlite; this layer +//! catches residual `SQLITE_BUSY` / `SQLITE_LOCKED` after that window. + +use std::thread; +use std::time::Duration; + +use anyhow::{Context, Result}; + +/// Per-connection busy handler window (issue #2077). +pub const BUSY_TIMEOUT: Duration = Duration::from_millis(5000); + +/// Application-level retries after rusqlite's busy handler is exhausted. +const WRITE_RETRY_ATTEMPTS: u32 = 6; +const WRITE_RETRY_BASE_MS: u64 = 25; + +/// Returns true when `err` is transient SQLite write-lock contention. +pub fn is_sqlite_busy(err: &anyhow::Error) -> bool { + if let Some(rusqlite::Error::SqliteFailure(sqlite_err, _)) = + err.downcast_ref::() + { + return matches!( + sqlite_err.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + ); + } + let msg = format!("{err:#}").to_ascii_lowercase(); + msg.contains("database is locked") || msg.contains("database table is locked") +} + +/// Run `f` up to [`WRITE_RETRY_ATTEMPTS`] times when SQLite reports busy/locked. +pub fn retry_on_sqlite_busy(op_name: &str, mut f: F) -> Result +where + F: FnMut() -> Result, +{ + let mut last_err: Option = None; + + for attempt in 0..WRITE_RETRY_ATTEMPTS { + match f() { + Ok(val) => { + if attempt > 0 { + log::debug!("[whatsapp_data] {op_name} succeeded after {attempt} busy retries"); + } + return Ok(val); + } + Err(e) => { + if !is_sqlite_busy(&e) { + return Err(e); + } + if attempt + 1 == WRITE_RETRY_ATTEMPTS { + last_err = Some(e); + break; + } + let sleep_ms = WRITE_RETRY_BASE_MS + .saturating_mul(2u64.saturating_pow(attempt)) + .min(500); + log::warn!( + "[whatsapp_data] {op_name} SQLite busy/locked \ + (attempt {} of {WRITE_RETRY_ATTEMPTS}), retry in {sleep_ms}ms: {e:#}", + attempt + 1, + ); + thread::sleep(Duration::from_millis(sleep_ms)); + } + } + } + + Err(last_err.expect("WRITE_RETRY_ATTEMPTS > 0").context(format!( + "{op_name} failed after {WRITE_RETRY_ATTEMPTS} SQLite busy retries" + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn is_sqlite_busy_matches_database_busy_code() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, + }, + Some("database is locked".into()), + ); + let err = anyhow::Error::from(raw); + assert!(is_sqlite_busy(&err)); + } + + #[test] + fn is_sqlite_busy_does_not_match_constraint_violation() { + let raw = rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::ConstraintViolation, + extended_code: 19, + }, + Some("UNIQUE constraint failed".into()), + ); + let err = anyhow::Error::from(raw); + assert!(!is_sqlite_busy(&err)); + } + + #[test] + fn retry_on_sqlite_busy_succeeds_after_transient_busy() { + let mut calls = 0u32; + let result = retry_on_sqlite_busy("test_op", || { + calls += 1; + if calls < 3 { + Err(anyhow::Error::from(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, + }, + Some("database is locked".into()), + ))) + } else { + Ok(42usize) + } + }); + assert_eq!(result.unwrap(), 42); + assert_eq!(calls, 3); + } + + #[test] + fn retry_on_sqlite_busy_does_not_retry_non_busy_errors() { + let mut calls = 0u32; + let result: Result<()> = retry_on_sqlite_busy("test_op", || { + calls += 1; + anyhow::bail!("permanent failure"); + }); + assert!(result.is_err()); + assert_eq!(calls, 1); + } +} diff --git a/src/openhuman/whatsapp_data/store.rs b/src/openhuman/whatsapp_data/store.rs index 6d35df5ef..671d96fc6 100644 --- a/src/openhuman/whatsapp_data/store.rs +++ b/src/openhuman/whatsapp_data/store.rs @@ -7,10 +7,12 @@ use std::collections::HashMap; use std::path::Path; +use std::sync::Mutex; use anyhow::{Context, Result}; use rusqlite::{params, Connection}; +use crate::openhuman::whatsapp_data::sqlite_retry::{retry_on_sqlite_busy, BUSY_TIMEOUT}; use crate::openhuman::whatsapp_data::types::{ ChatMeta, IngestMessage, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest, WhatsAppChat, WhatsAppMessage, @@ -19,6 +21,9 @@ use crate::openhuman::whatsapp_data::types::{ /// SQLite-backed store for WhatsApp chats and messages. pub struct WhatsAppDataStore { db_path: std::path::PathBuf, + /// Serializes write paths (upsert + prune) so concurrent ingest RPCs do not + /// open competing writers on the same `whatsapp_data.db` file. + write_lock: Mutex<()>, } impl WhatsAppDataStore { @@ -31,7 +36,10 @@ impl WhatsAppDataStore { .with_context(|| format!("create whatsapp_data dir: {}", parent.display()))?; } log::debug!("[whatsapp_data] opening store at {}", db_path.display()); - let store = Self { db_path }; + let store = Self { + db_path, + write_lock: Mutex::new(()), + }; store.init_schema()?; Ok(store) } @@ -77,8 +85,27 @@ impl WhatsAppDataStore { } fn open_conn(&self) -> Result { - Connection::open(&self.db_path) - .with_context(|| format!("open whatsapp_data db: {}", self.db_path.display())) + let conn = Connection::open(&self.db_path) + .with_context(|| format!("open whatsapp_data db: {}", self.db_path.display()))?; + Self::configure_connection(&conn)?; + Ok(conn) + } + + /// Per-connection pragmas: busy handler + WAL (idempotent on existing DBs). + fn configure_connection(conn: &Connection) -> Result<()> { + conn.busy_timeout(BUSY_TIMEOUT) + .context("configure whatsapp_data busy_timeout")?; + if let Err(wal_err) = conn.execute_batch("PRAGMA journal_mode=WAL;") { + log::warn!( + "[whatsapp_data] failed to enable WAL (filesystem may not support it): {wal_err}" + ); + } + Ok(()) + } + + #[cfg(test)] + pub(crate) fn open_conn_for_test(&self) -> Result { + self.open_conn() } fn now_secs() -> i64 { @@ -97,6 +124,20 @@ impl WhatsAppDataStore { if chats.is_empty() { return Ok(0); } + let _write_guard = self + .write_lock + .lock() + .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?; + retry_on_sqlite_busy("upsert_chats", || { + self.upsert_chats_inner(account_id, chats) + }) + } + + fn upsert_chats_inner( + &self, + account_id: &str, + chats: &HashMap, + ) -> Result { let conn = self.open_conn()?; let now = Self::now_secs(); let mut count = 0usize; @@ -127,6 +168,16 @@ impl WhatsAppDataStore { if msgs.is_empty() { return Ok(0); } + let _write_guard = self + .write_lock + .lock() + .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?; + retry_on_sqlite_busy("upsert_messages", || { + self.upsert_messages_inner(account_id, msgs) + }) + } + + fn upsert_messages_inner(&self, account_id: &str, msgs: &[IngestMessage]) -> Result { let conn = self.open_conn()?; let now = Self::now_secs(); let mut count = 0usize; @@ -210,6 +261,16 @@ impl WhatsAppDataStore { /// `last_message_ts` for every chat that lost rows, so `list_chats` /// returns accurate counts and ordering immediately. pub fn prune_old_messages(&self, cutoff_ts: i64) -> Result { + let _write_guard = self + .write_lock + .lock() + .map_err(|e| anyhow::anyhow!("whatsapp_data write lock poisoned: {e}"))?; + retry_on_sqlite_busy("prune_old_messages", || { + self.prune_old_messages_inner(cutoff_ts) + }) + } + + fn prune_old_messages_inner(&self, cutoff_ts: i64) -> Result { let conn = self.open_conn()?; let now = Self::now_secs(); diff --git a/src/openhuman/whatsapp_data/store_tests.rs b/src/openhuman/whatsapp_data/store_tests.rs index 3e2e75a22..4ccfd77ef 100644 --- a/src/openhuman/whatsapp_data/store_tests.rs +++ b/src/openhuman/whatsapp_data/store_tests.rs @@ -4,11 +4,16 @@ //! `store.rs`. They focus on cross-account scoping guarantees: data written //! for one `account_id` must never appear in queries for a different account. +use super::super::sqlite_retry::BUSY_TIMEOUT; use super::super::types::{ ChatMeta, IngestMessage, ListChatsRequest, ListMessagesRequest, SearchMessagesRequest, }; use super::WhatsAppDataStore; use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::mpsc; +use std::thread; +use std::time::Duration; use tempfile::tempdir; fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) { @@ -17,6 +22,28 @@ fn make_store() -> (WhatsAppDataStore, tempfile::TempDir) { (store, tmp) } +fn db_path_for(tmp: &tempfile::TempDir) -> PathBuf { + tmp.path().join("whatsapp_data").join("whatsapp_data.db") +} + +/// Hold an immediate write transaction until the returned sender fires, then commit. +fn spawn_write_blocker(db_path: PathBuf) -> mpsc::Sender<()> { + let (hold_tx, hold_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + thread::spawn(move || { + let conn = rusqlite::Connection::open(&db_path).expect("blocker open"); + conn.busy_timeout(BUSY_TIMEOUT) + .expect("blocker busy_timeout"); + conn.execute_batch("BEGIN IMMEDIATE") + .expect("blocker BEGIN IMMEDIATE"); + hold_tx.send(()).expect("blocker ready signal"); + let _ = release_rx.recv(); + conn.execute_batch("COMMIT").expect("blocker COMMIT"); + }); + hold_rx.recv().expect("blocker must acquire write lock"); + release_tx +} + fn chat_meta(name: &str) -> ChatMeta { ChatMeta { name: Some(name.to_string()), @@ -338,6 +365,101 @@ fn prune_old_messages_refreshes_chat_stats() { ); } +/// Concurrent external write lock must not fail ingest — busy_timeout + +/// retry_on_sqlite_busy should wait for the blocker to commit. +#[test] +fn upsert_chats_succeeds_after_sqlite_busy_contention() { + let (store, tmp) = make_store(); + let workspace = tmp.path().to_path_buf(); + let db_path = db_path_for(&tmp); + let mut chats = HashMap::new(); + chats.insert("chat@c.us".to_string(), chat_meta("Alice")); + + let release = spawn_write_blocker(db_path); + let upsert_handle = thread::spawn(move || store.upsert_chats("acct1", &chats)); + // Let upsert reach SQLite busy wait, then release the external write lock. + thread::sleep(Duration::from_millis(50)); + release.send(()).expect("release blocker"); + + let count = upsert_handle + .join() + .expect("upsert thread") + .expect("upsert must succeed once blocker releases"); + assert_eq!(count, 1); + + let store = WhatsAppDataStore::new(&workspace).expect("reopen store"); + let rows = store + .list_chats(&ListChatsRequest { + account_id: Some("acct1".to_string()), + limit: None, + offset: None, + }) + .unwrap(); + assert_eq!(rows.len(), 1); +} + +#[test] +fn prune_old_messages_succeeds_after_sqlite_busy_contention() { + let (store, tmp) = make_store(); + let workspace = tmp.path().to_path_buf(); + let db_path = db_path_for(&tmp); + let mut chats = HashMap::new(); + chats.insert("chat@c.us".to_string(), chat_meta("Chat")); + store.upsert_chats("acct1", &chats).unwrap(); + store + .upsert_messages( + "acct1", + &[ + simple_message("old", "chat@c.us", "old", 1_000_000), + simple_message("new", "chat@c.us", "new", 2_000_000_000), + ], + ) + .unwrap(); + + let release = spawn_write_blocker(db_path); + let prune_handle = thread::spawn(move || store.prune_old_messages(1_500_000_000)); + thread::sleep(Duration::from_millis(50)); + release.send(()).expect("release blocker"); + + let pruned = prune_handle + .join() + .expect("prune thread") + .expect("prune must succeed once blocker releases"); + assert_eq!(pruned, 1); + + let store = WhatsAppDataStore::new(&workspace).expect("reopen store"); + let chats_after = store + .list_chats(&ListChatsRequest { + account_id: Some("acct1".to_string()), + limit: None, + offset: None, + }) + .unwrap(); + assert_eq!(chats_after[0].message_count, 1); +} + +#[test] +fn open_conn_configures_busy_timeout_and_wal() { + let (store, _tmp) = make_store(); + let conn = store.open_conn_for_test().expect("open"); + let busy_ms: i64 = conn + .pragma_query_value(None, "busy_timeout", |v| v.get(0)) + .expect("busy_timeout pragma"); + assert_eq!( + busy_ms, + BUSY_TIMEOUT.as_millis() as i64, + "busy_timeout must match configured window" + ); + let journal_mode: String = conn + .pragma_query_value(None, "journal_mode", |v| v.get(0)) + .expect("journal_mode pragma"); + assert_eq!( + journal_mode.to_ascii_lowercase(), + "wal", + "journal_mode must be WAL" + ); +} + /// Messages with an empty message_id or chat_id must be silently skipped, /// never causing a panic or spurious database error. #[test]