fix(whatsapp): retry WhatsApp SQLite writes on database-is-locked (#2077) (#2098)

Co-authored-by: Cursor <cursoragent@cursor.com>

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;
 

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::<rusqlite::Error>()
+    {
+        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<T, F>(op_name: &str, mut f: F) -> Result<T>
+where
+    F: FnMut() -> Result<T>,
+{
+    let mut last_err: Option<anyhow::Error> = 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);
+    }
+}

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> {
-        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<Connection> {
+        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<String, ChatMeta>,
+    ) -> Result<usize> {
         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<usize> {
         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<u64> {
+        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<u64> {
         let conn = self.open_conn()?;
         let now = Self::now_secs();
 

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]