fix(desktop): offload archive sqlite work

Finding: L2 archive read/save-subscription/archive_events sqlite work could block the GUI IPC path or async workers under busy_timeout contention.

Move archive planning, commits, save-subscription merge/remove/list, and paginated reads through spawn_blocking with each rusqlite connection scoped inside the blocking closure. Relay query awaits remain between DB phases with no connection or transaction in scope.

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

desktop/src-tauri/src/archive/mod.rs

@@ -57,6 +57,19 @@ fn now_secs() -> i64 {
         .as_secs() as i64
 }
 
+async fn run_archive_db_task<T, F>(task: F) -> Result<T, String>
+where
+    T: Send + 'static,
+    F: FnOnce(&Connection) -> Result<T, String> + Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let conn = open_db()?;
+        task(&conn)
+    })
+    .await
+    .map_err(|e| format!("spawn_blocking failed: {e}"))?
+}
+
 // ── Scope type ───────────────────────────────────────────────────────────────
 
 /// The three supported archive scope discriminants.
@@ -118,9 +131,9 @@ pub struct ArchiveBatchResult {
 ///
 /// # Send-safety
 ///
-/// `rusqlite::Connection` is `!Send`. All DB work is bracketed in scoped
-/// `{ let conn = open_db()?; ... }` blocks that drop the connection before any
-/// `.await`, exactly matching the pattern in `managed_agents/persona_events.rs`.
+/// SQLite planning and commit run on the blocking pool. Each phase opens and
+/// drops its own `rusqlite::Connection` inside the blocking closure; no
+/// connection, transaction, or lock is held across the relay-query await.
 #[tauri::command]
 pub async fn archive_events(
     state: State<'_, AppState>,
@@ -130,37 +143,39 @@ pub async fn archive_events(
     let relay_url = relay_ws_url_with_override(&state);
     let now = now_secs();
 
-    // ── Phase 1: plan (sync) ─────────────────────────────────────────────────
-    // Read subscriptions and build relay filters. Connection dropped before
-    // any .await.
-    let plan = {
-        let conn = open_db()?;
-        plan_archive(candidates, &identity_pk, &relay_url, &conn)?
-        // conn drops here
-    };
+    // ── Phase 1: plan (blocking SQLite) ─────────────────────────────────────
+    let plan_identity_pk = identity_pk.clone();
+    let plan_relay_url = relay_url.clone();
+    let plan = run_archive_db_task(move |conn| {
+        plan_archive(candidates, &plan_identity_pk, &plan_relay_url, conn)
+    })
+    .await?;
 
     // ── Phase 2: relay queries (async) ───────────────────────────────────────
-    // No Connection in scope — future is Send.
     let state_ref: &AppState = &state;
     let bucket_results = query_buckets(plan.buckets, state_ref).await;
 
-    // ── Phase 3: persist (sync) ──────────────────────────────────────────────
-    let conn = open_db()?;
+    // ── Phase 3: persist (blocking SQLite) ──────────────────────────────────
     let owner_keys = {
         let keys_guard = state.keys.lock().map_err(|e| e.to_string())?;
         keys_guard.clone()
-        // guard drops here
+        // guard drops here, before awaiting the blocking commit task.
     };
-    commit_archive(
-        bucket_results,
-        plan.ephemeral,
-        plan.pre_dropped,
-        &identity_pk,
-        &relay_url,
-        &owner_keys,
-        now,
-        &conn,
-    )
+    let commit_identity_pk = identity_pk.clone();
+    let commit_relay_url = relay_url.clone();
+    run_archive_db_task(move |conn| {
+        commit_archive(
+            bucket_results,
+            plan.ephemeral,
+            plan.pre_dropped,
+            &commit_identity_pk,
+            &commit_relay_url,
+            &owner_keys,
+            now,
+            conn,
+        )
+    })
+    .await
 }
 
 /// Validate an ephemeral observer frame (kind 24200) against ALL local rules.
@@ -398,7 +413,10 @@ async fn probe_event_readable(state: &AppState, event_id: &str) -> Result<(), St
 /// `useAgentMetricArchiveSeed` (kind 44200) instead of the former
 /// list → merge-in-TS → create pattern.
 #[tauri::command]
-pub fn merge_save_subscription_kinds(state: State<'_, AppState>, kind: u32) -> Result<(), String> {
+pub async fn merge_save_subscription_kinds(
+    state: State<'_, AppState>,
+    kind: u32,
+) -> Result<(), String> {
     if kind > u32::from(u16::MAX) {
         return Err(format!("kind {kind} is out of the valid range 0..=65535"));
     }
@@ -406,8 +424,11 @@ pub fn merge_save_subscription_kinds(state: State<'_, AppState>, kind: u32) -> R
     let identity_pk = identity_pubkey(&state)?;
     let relay_url = relay_ws_url_with_override(&state);
     let now = now_secs();
-    let conn = open_db()?;
-    store::merge_owner_p_kinds(&conn, &identity_pk, &relay_url, &identity_pk, kind, now)
+    let owner_pk = identity_pk.clone();
+    run_archive_db_task(move |conn| {
+        store::merge_owner_p_kinds(conn, &identity_pk, &relay_url, &owner_pk, kind, now)
+    })
+    .await
 }
 
 // ── remove_save_subscription_kind ────────────────────────────────────────────
@@ -425,28 +446,34 @@ pub fn merge_save_subscription_kinds(state: State<'_, AppState>, kind: u32) -> R
 /// former TS-side read-modify-overwrite + whole-row `deleteSaveSubscription`
 /// which would drop the *other* kind if `subs` state was stale.
 #[tauri::command]
-pub fn remove_save_subscription_kind(state: State<'_, AppState>, kind: u32) -> Result<(), String> {
+pub async fn remove_save_subscription_kind(
+    state: State<'_, AppState>,
+    kind: u32,
+) -> Result<(), String> {
     if kind > u32::from(u16::MAX) {
         return Err(format!("kind {kind} is out of the valid range 0..=65535"));
     }
 
     let identity_pk = identity_pubkey(&state)?;
     let relay_url = relay_ws_url_with_override(&state);
-    let conn = open_db()?;
-    store::remove_owner_p_kind(&conn, &identity_pk, &relay_url, &identity_pk, kind)
+    let owner_pk = identity_pk.clone();
+    run_archive_db_task(move |conn| {
+        store::remove_owner_p_kind(conn, &identity_pk, &relay_url, &owner_pk, kind)
+    })
+    .await
 }
 
 // ── list_save_subscriptions ──────────────────────────────────────────────────
 
 /// List all save subscriptions for the current identity + relay.
 #[tauri::command]
-pub fn list_save_subscriptions(
+pub async fn list_save_subscriptions(
     state: State<'_, AppState>,
 ) -> Result<Vec<store::SaveSubscription>, String> {
     let identity_pk = identity_pubkey(&state)?;
     let relay_url = relay_ws_url_with_override(&state);
-    let conn = open_db()?;
-    store::list_save_subscriptions(&conn, &identity_pk, &relay_url)
+    run_archive_db_task(move |conn| store::list_save_subscriptions(conn, &identity_pk, &relay_url))
+        .await
 }
 
 // ── delete_save_subscription ─────────────────────────────────────────────────
@@ -496,7 +523,7 @@ const DEFAULT_READ_LIMIT: i64 = 50;
 /// caller doing `Event::from_json` on an unfiltered read must filter by kind
 /// first (today's only reader filters `kinds: [24200]`).
 #[tauri::command]
-pub fn read_archived_events(
+pub async fn read_archived_events(
     state: State<'_, AppState>,
     scope_type: ScopeType,
     scope_value: String,
@@ -507,18 +534,22 @@ pub fn read_archived_events(
 ) -> Result<Vec<String>, String> {
     let identity_pk = identity_pubkey(&state)?;
     let relay_url = relay_ws_url_with_override(&state);
-    let conn = open_db()?;
-    store::read_archived_events(
-        &conn,
-        &identity_pk,
-        &relay_url,
-        scope_type.as_str(),
-        &scope_value,
-        kinds.as_deref(),
-        before_created_at,
-        before_id.as_deref(),
-        limit.unwrap_or(DEFAULT_READ_LIMIT),
-    )
+    let scope_type_str = scope_type.as_str().to_string();
+    let read_limit = limit.unwrap_or(DEFAULT_READ_LIMIT);
+    run_archive_db_task(move |conn| {
+        store::read_archived_events(
+            conn,
+            &identity_pk,
+            &relay_url,
+            &scope_type_str,
+            &scope_value,
+            kinds.as_deref(),
+            before_created_at,
+            before_id.as_deref(),
+            read_limit,
+        )
+    })
+    .await
 }
 
 // ── Tests ────────────────────────────────────────────────────────────────────