Sweep orphaned response bodies at browser startup (#576)

Cargo.lock

@@ -11277,6 +11277,7 @@ dependencies = [
  "yaak-grpc",
  "yaak-http",
  "yaak-license",
+ "yaak-lifecycle",
  "yaak-mac-window",
  "yaak-models",
  "yaak-plugins",
@@ -11342,6 +11343,7 @@ dependencies = [
  "yaak-core",
  "yaak-crypto",
  "yaak-http",
+ "yaak-lifecycle",
  "yaak-models",
  "yaak-plugins",
  "yaak-templates",
@@ -11526,6 +11528,14 @@ dependencies = [
  "yaak-models",
 ]
 
+[[package]]
+name = "yaak-lifecycle"
+version = "0.0.0"
+dependencies = [
+ "log 0.4.29",
+ "yaak-models",
+]
+
 [[package]]
 name = "yaak-mac-window"
 version = "0.1.0"
@@ -11745,6 +11755,8 @@ dependencies = [
  "sqlite-wasm-vfs",
  "wasm-bindgen",
  "wasm-bindgen-futures",
+ "web-sys",
+ "yaak-lifecycle",
  "yaak-models",
 ]
 

Cargo.toml

@@ -14,6 +14,7 @@ members = [
   "crates/yaak-git",
   "crates/yaak-grpc",
   "crates/yaak-http",
+  "crates/yaak-lifecycle",
   "crates/yaak-models",
   "crates/yaak-plugins",
   "crates/yaak-sse",
@@ -77,6 +78,7 @@ yaak-crypto = { path = "crates/yaak-crypto" }
 yaak-git = { path = "crates/yaak-git" }
 yaak-grpc = { path = "crates/yaak-grpc" }
 yaak-http = { path = "crates/yaak-http" }
+yaak-lifecycle = { path = "crates/yaak-lifecycle" }
 yaak-models = { path = "crates/yaak-models" }
 yaak-plugins = { path = "crates/yaak-plugins" }
 yaak-sse = { path = "crates/yaak-sse" }

crates-cli/yaak-cli/Cargo.toml

@@ -45,6 +45,7 @@ yaak-api = { workspace = true }
 yaak-core = { workspace = true }
 yaak-crypto = { workspace = true }
 yaak-http = { workspace = true }
+yaak-lifecycle = { workspace = true }
 yaak-models = { workspace = true }
 yaak-plugins = { workspace = true }
 yaak-templates = { workspace = true }

crates-cli/yaak-cli/src/context.rs

@@ -49,6 +49,14 @@ impl CliContext {
                     std::process::exit(1);
                 }
             };
+
+        // Guest: the desktop may have this DB open, so only what's safe beside a live session
+        let _ = yaak_lifecycle::on_launch(
+            &yaak_lifecycle::Host::guest(),
+            &query_manager.connect(),
+            &blob_manager,
+        );
+
         let encryption_manager = Arc::new(EncryptionManager::new(query_manager.clone(), app_id));
 
         Self {

crates-tauri/yaak-app-client/Cargo.toml

@@ -88,6 +88,7 @@ yaak-grpc = { workspace = true }
 yaak-http = { workspace = true }
 yaak-license = { workspace = true, optional = true }
 yaak-mac-window = { workspace = true }
+yaak-lifecycle = { workspace = true }
 yaak-models = { workspace = true }
 yaak-plugins = { workspace = true }
 yaak-sse = { workspace = true }

crates-tauri/yaak-app-client/src/lib.rs

@@ -1258,6 +1258,14 @@ pub fn run() {
 
     builder
         .setup(|app| {
+            let lifecycle_host = yaak_lifecycle::Host::owner()
+                .with_responses_dir(app.path().app_data_dir()?.join("responses"));
+            if let Err(e) =
+                yaak_lifecycle::on_launch(&lifecycle_host, &app.db(), &app.blob_manager())
+            {
+                error!("on_launch hook failed: {e:?}");
+            }
+
             // The RPC command registry — every frontend command dispatches
             // through this via the single `rpc` Tauri command
             app.manage(rpc_ext::build_rpc_router::<TauriRuntime>());
@@ -1357,15 +1365,6 @@ pub fn run() {
                         let info = history::get_or_upsert_launch_info(&h);
                         debug!("Launched Yaak {:?}", info);
                     });
-
-                    // Cancel pending requests
-                    let h = app_handle.clone();
-                    tauri::async_runtime::block_on(async move {
-                        let db = h.db();
-                        let _ = db.cancel_pending_http_responses();
-                        let _ = db.cancel_pending_grpc_connections();
-                        let _ = db.cancel_pending_websocket_connections();
-                    });
                 }
                 RunEvent::WindowEvent { event: WindowEvent::Focused(true), label, .. } => {
                     #[cfg(any(target_os = "linux", target_os = "macos"))]

crates-tauri/yaak-app-client/src/models_ext.rs

@@ -15,7 +15,6 @@ use yaak_models::error::Result;
 use yaak_models::query_manager::QueryManager;
 use yaak_models::util::{ModelPayload, UpdateSource};
 
-const MODEL_CHANGES_RETENTION_HOURS: i64 = 1;
 const MODEL_CHANGES_POLL_INTERVAL_MS: u64 = 1000;
 const MODEL_CHANGES_POLL_BATCH_SIZE: usize = 200;
 
@@ -152,30 +151,11 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
                     }
                 };
 
-            let db = query_manager.connect();
-            if let Err(err) = db.prune_model_changes_older_than_hours(MODEL_CHANGES_RETENTION_HOURS)
-            {
-                error!("Failed to prune model_changes rows on startup: {err:?}");
-            }
             // Only stream writes that happen after this app launch.
             let cursor = ModelChangeCursor::from_launch_time();
 
             let poll_query_manager = query_manager.clone();
 
-            // GC response bodies orphaned by cascade deletes, which historically
-            // didn't clean the blob DB or responses directory
-            let gc_query_manager = query_manager.clone();
-            let gc_blob_manager = blob_manager.clone();
-            let gc_responses_dir = app_path.join("responses");
-            tauri::async_runtime::spawn_blocking(move || {
-                let db = gc_query_manager.connect();
-                match db.delete_orphaned_response_bodies(&gc_blob_manager, &gc_responses_dir) {
-                    Ok(0) => {}
-                    Ok(n) => log::info!("Deleted {n} orphaned response bodies"),
-                    Err(e) => error!("Failed to delete orphaned response bodies: {e:?}"),
-                }
-            });
-
             app_handle.manage(query_manager);
             app_handle.manage(blob_manager);
 

crates/yaak-lifecycle/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "yaak-lifecycle"
+version = "0.0.0"
+edition = "2024"
+authors = ["Gregory Schier"]
+publish = false
+
+[dependencies]
+log = { workspace = true }
+yaak-models = { workspace = true }

crates/yaak-lifecycle/src/lib.rs

@@ -0,0 +1,130 @@
+//! Lifecycle hooks shared by every host (desktop, browser, CLI). The hooks say
+//! what happens at each moment; the host decides when and on which thread.
+//!
+//! Builds for wasm32, so it can depend on `yaak-models` but not on the send
+//! engine or plugin runtime.
+
+use log::info;
+use std::path::PathBuf;
+use yaak_models::blob_manager::BlobManager;
+use yaak_models::client_db::ClientDb;
+use yaak_models::error::Result;
+
+const MODEL_CHANGES_RETENTION_HOURS: i64 = 1;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Role {
+    /// Has the database for the life of the app (desktop, browser worker)
+    Owner,
+    /// Short-lived, and an owner may be using the database right now (CLI).
+    /// Must not touch anything in flight.
+    Guest,
+}
+
+/// Paths are `None` on hosts without a filesystem (the browser).
+#[derive(Debug, Clone)]
+pub struct Host {
+    pub role: Role,
+    pub responses_dir: Option<PathBuf>,
+}
+
+impl Host {
+    pub fn owner() -> Self {
+        Self { role: Role::Owner, responses_dir: None }
+    }
+
+    pub fn guest() -> Self {
+        Self { role: Role::Guest, responses_dir: None }
+    }
+
+    pub fn with_responses_dir(mut self, dir: impl Into<PathBuf>) -> Self {
+        self.responses_dir = Some(dir.into());
+        self
+    }
+}
+
+/// Run once after the database is open, before the host answers anything.
+pub fn on_launch(host: &Host, db: &ClientDb, blobs: &BlobManager) -> Result<()> {
+    db.prune_model_changes_older_than_hours(MODEL_CHANGES_RETENTION_HOURS)?;
+
+    if host.role == Role::Owner {
+        // Anything still in flight was left by the last session
+        db.cancel_pending_http_responses()?;
+        db.cancel_pending_grpc_connections()?;
+        db.cancel_pending_websocket_connections()?;
+
+        // Cascaded deletes never cleaned up response bodies
+        let deleted = match host.responses_dir.as_deref() {
+            Some(dir) => db.delete_orphaned_response_bodies(blobs, dir)?,
+            None => db.delete_orphaned_response_body_blobs(blobs)?,
+        };
+        if deleted > 0 {
+            info!("Deleted {deleted} orphaned response bodies");
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use yaak_models::blob_manager::BodyChunk;
+    use yaak_models::init_in_memory;
+    use yaak_models::models::{HttpRequest, HttpResponse, HttpResponseState, Workspace};
+    use yaak_models::util::UpdateSource;
+
+    #[test]
+    fn only_the_owner_closes_what_the_last_session_left_open() {
+        let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+        let source = &UpdateSource::Background;
+
+        let workspace = db
+            .upsert_workspace(
+                &Workspace { name: "Hooks".to_string(), ..Default::default() },
+                source,
+            )
+            .unwrap();
+        let request = db
+            .upsert_http_request(
+                &HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() },
+                source,
+            )
+            .unwrap();
+        let pending = db
+            .upsert_http_response(
+                &HttpResponse {
+                    request_id: request.id.clone(),
+                    workspace_id: workspace.id.clone(),
+                    state: HttpResponseState::Connected,
+                    ..Default::default()
+                },
+                source,
+                &blob_manager,
+            )
+            .unwrap();
+
+        on_launch(&Host::guest(), &db, &blob_manager).unwrap();
+        let response = db.get_http_response(&pending.id).unwrap();
+        assert!(matches!(response.state, HttpResponseState::Connected));
+
+        on_launch(&Host::owner(), &db, &blob_manager).unwrap();
+        let response = db.get_http_response(&pending.id).unwrap();
+        assert!(matches!(response.state, HttpResponseState::Closed));
+    }
+
+    #[test]
+    fn owner_without_a_filesystem_still_sweeps_blobs() {
+        let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+        {
+            let blob_ctx = blob_manager.connect();
+            blob_ctx.insert_chunk(&BodyChunk::new("rs_gone", 0, b"dead".to_vec())).unwrap();
+        }
+
+        on_launch(&Host::owner(), &db, &blob_manager).unwrap();
+
+        assert!(!blob_manager.connect().body_exists("rs_gone").unwrap());
+    }
+}

crates/yaak-models/src/queries/http_responses.rs

@@ -45,6 +45,31 @@ impl<'a> ClientDb<'a> {
         Ok(count)
     }
 
+    /// Delete blob-stored response bodies whose owning HTTP response row no
+    /// longer exists. Blob ids are keyed by the response that owns them —
+    /// "{response_id}" for a response body, "{response_id}.request" for the
+    /// request that produced it — so ownership is the id's first segment.
+    ///
+    /// The blob half of [`Self::delete_orphaned_response_bodies`], on its own
+    /// for hosts with no filesystem to hold body files. See `crate::hooks`.
+    ///
+    /// Returns the number of orphaned bodies deleted.
+    pub fn delete_orphaned_response_body_blobs(&self, blobs: &BlobManager) -> Result<usize> {
+        let mut deleted = 0;
+
+        let blob_ctx = blobs.connect();
+        for body_id in blob_ctx.list_body_ids()? {
+            let response_id = body_id.split('.').next().unwrap_or_default();
+            if self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_some() {
+                continue;
+            }
+            blob_ctx.delete_chunks(&body_id)?;
+            deleted += 1;
+        }
+
+        Ok(deleted)
+    }
+
     /// Delete response body data (blob chunks and body files) whose owning HTTP
     /// response row no longer exists. Cascaded deletes (request, folder,
     /// workspace) historically never cleaned the blob DB or the responses
@@ -59,18 +84,7 @@ impl<'a> ClientDb<'a> {
         blobs: &BlobManager,
         responses_dir: &std::path::Path,
     ) -> Result<usize> {
-        let mut deleted = 0;
+        let mut deleted = self.delete_orphaned_response_body_blobs(blobs)?;
-
-        // Blob chunks are keyed "{response_id}.request"
-        let blob_ctx = blobs.connect();
-        for body_id in blob_ctx.list_body_ids()? {
-            let response_id = body_id.split('.').next().unwrap_or_default();
-            if self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_some() {
-                continue;
-            }
-            blob_ctx.delete_chunks(&body_id)?;
-            deleted += 1;
-        }
 
         // Body files are stored as {responses_dir}/{response_id}
         if let Ok(entries) = fs::read_dir(responses_dir) {
@@ -172,19 +186,20 @@ impl<'a> ClientDb<'a> {
 
 #[cfg(test)]
 mod tests {
-    use crate::blob_manager::BodyChunk;
+    use crate::blob_manager::{BlobManager, BodyChunk};
+    use crate::client_db::ClientDb;
     use crate::init_in_memory;
     use crate::models::{HttpRequest, HttpResponse, Workspace};
     use crate::util::UpdateSource;
 
-    #[test]
+    /// A workspace, a request, and one response that still exists.
-    fn deletes_orphaned_response_bodies() {
+    fn seed_live_response(db: &ClientDb, blob_manager: &BlobManager) -> HttpResponse {
-        let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
-        let db = query_manager.connect();
-
         let source = &UpdateSource::Background;
         let workspace = db
-            .upsert_workspace(&Workspace { name: "GC Test".to_string(), ..Default::default() }, source)
+            .upsert_workspace(
+                &Workspace { name: "GC Test".to_string(), ..Default::default() },
+                source,
+            )
             .expect("Failed to upsert workspace");
         let request = db
             .upsert_http_request(
@@ -192,19 +207,57 @@ mod tests {
                 source,
             )
             .expect("Failed to upsert request");
+        db.upsert_http_response(
+            &HttpResponse {
+                request_id: request.id.clone(),
+                workspace_id: workspace.id.clone(),
+                ..Default::default()
+            },
+            source,
+            blob_manager,
+        )
+        .expect("Failed to upsert response")
+    }
 
-        let live = db
+    /// What a browser host runs: no filesystem, so bodies exist only as blob
-            .upsert_http_response(
+    /// chunks, under both id shapes the blob DB uses.
-                &HttpResponse {
+    #[test]
-                    request_id: request.id.clone(),
+    fn deletes_orphaned_response_body_blobs() {
-                    workspace_id: workspace.id.clone(),
+        let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
-                    ..Default::default()
+        let db = query_manager.connect();
-                },
-                source,
-                &blob_manager,
-            )
-            .expect("Failed to upsert response");
 
+        let live = seed_live_response(&db, &blob_manager);
+        let live_request_body_id = format!("{}.request", live.id);
+        {
+            // Scope the connection: the in-memory pool only has one, and the GC
+            // needs to take it
+            let blob_ctx = blob_manager.connect();
+            blob_ctx.insert_chunk(&BodyChunk::new(&live.id, 0, b"live".to_vec())).unwrap();
+            blob_ctx
+                .insert_chunk(&BodyChunk::new(&live_request_body_id, 0, b"live".to_vec()))
+                .unwrap();
+            blob_ctx.insert_chunk(&BodyChunk::new("rs_gone", 0, b"dead".to_vec())).unwrap();
+            blob_ctx.insert_chunk(&BodyChunk::new("rs_gone.request", 0, b"dead".to_vec())).unwrap();
+        }
+
+        let deleted = db
+            .delete_orphaned_response_body_blobs(&blob_manager)
+            .expect("Failed to GC response body blobs");
+        assert_eq!(deleted, 2);
+
+        let blob_ctx = blob_manager.connect();
+        assert!(blob_ctx.body_exists(&live.id).unwrap());
+        assert!(blob_ctx.body_exists(&live_request_body_id).unwrap());
+        assert!(!blob_ctx.body_exists("rs_gone").unwrap());
+        assert!(!blob_ctx.body_exists("rs_gone.request").unwrap());
+    }
+
+    #[test]
+    fn deletes_orphaned_response_bodies() {
+        let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        let live = seed_live_response(&db, &blob_manager);
         let live_body_id = format!("{}.request", live.id);
         {
             // Scope the connection: the in-memory pool only has one, and the GC

crates/yaak-web/Cargo.toml

@@ -25,6 +25,7 @@ crate-type = ["cdylib", "rlib"]
 log = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
+yaak-lifecycle = { workspace = true }
 yaak-models = { workspace = true }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
@@ -35,3 +36,4 @@ sqlite-wasm-rs = "0.5"
 sqlite-wasm-vfs = "0.2"
 wasm-bindgen = "0.2.100"
 wasm-bindgen-futures = "0.4"
+web-sys = { version = "0.3", features = ["console"] }

crates/yaak-web/pkg/yaak_web_bg.js

@@ -677,8 +677,11 @@ export function __wbg_versions_215a3ab1c9d5745a(arg0) {
     const ret = arg0.versions;
     return ret;
 }
+export function __wbg_warn_b6f36cac66fc96a4(arg0, arg1) {
+    console.warn(arg0, arg1);
+}
 export function __wbindgen_cast_0000000000000001(arg0, arg1) {
-    // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1102, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
+    // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1103, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
     const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4);
     return ret;
 }
@@ -688,8 +691,8 @@ export function __wbindgen_cast_0000000000000002(arg0, arg1) {
     return ret;
 }
 export function __wbindgen_cast_0000000000000003(arg0, arg1) {
-    // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 70, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
+    // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 177, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
-    const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9);
+    const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h400c17219073e521);
     return ret;
 }
 export function __wbindgen_cast_0000000000000004(arg0, arg1) {
@@ -746,8 +749,8 @@ function wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4(arg0, arg
     }
 }
 
-function wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9(arg0, arg1, arg2) {
+function wasm_bindgen__convert__closures_____invoke__h400c17219073e521(arg0, arg1, arg2) {
-    const ret = wasm.wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9(arg0, arg1, arg2);
+    const ret = wasm.wasm_bindgen__convert__closures_____invoke__h400c17219073e521(arg0, arg1, arg2);
     if (ret[1]) {
         throw takeFromExternrefTable0(ret[0]);
     }

crates/yaak-web/pkg/yaak_web_bg.wasm.d.ts

@@ -17,7 +17,7 @@ export const rust_sqlite_wasm_realloc: (a: number, b: number) => number;
 export const sqlite3_os_end: () => number;
 export const sqlite3_os_init: () => number;
 export const wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4: (a: number, b: number, c: any) => [number, number];
-export const wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9: (a: number, b: number, c: any) => [number, number];
+export const wasm_bindgen__convert__closures_____invoke__h400c17219073e521: (a: number, b: number, c: any) => [number, number];
 export const wasm_bindgen__convert__closures_____invoke__h2c72ca09e851b7f3: (a: number, b: number, c: any, d: any) => void;
 export const wasm_bindgen__convert__closures_____invoke__h7e06bb925b918fbb: (a: number, b: number, c: any) => void;
 export const wasm_bindgen__convert__closures_____invoke__h87640adb2bbfa2fc: (a: number, b: number) => void;

crates/yaak-web/src/lib.rs

@@ -43,6 +43,10 @@ struct Host {
     events: mpsc::Receiver<ModelPayload>,
 }
 
+fn lifecycle_host() -> yaak_lifecycle::Host {
+    yaak_lifecycle::Host::owner()
+}
+
 thread_local! {
     static HOST: RefCell<Option<Host>> = const { RefCell::new(None) };
 }
@@ -91,6 +95,10 @@ pub async fn boot() -> Result<()> {
     let (queries, blobs, events) =
         yaak_models::init_standalone(DB_NAME, BLOB_DB_NAME).map_err(js_error)?;
 
+    if let Err(e) = yaak_lifecycle::on_launch(&lifecycle_host(), &queries.connect(), &blobs) {
+        web_sys::console::warn_2(&"on_launch hook failed".into(), &js_error(e));
+    }
+
     HOST.with(|h| *h.borrow_mut() = Some(Host { queries, blobs, events }));
     Ok(())
 }