mirror of
https://github.com/mountain-loop/yaak.git
synced 2026-08-24 20:34:05 +02:00
Fix UI freezes with large workspaces (#536)
This commit is contained in:
@@ -79,6 +79,15 @@ impl BlobContext {
|
||||
Ok(chunks)
|
||||
}
|
||||
|
||||
/// List all distinct body IDs in the blob database.
|
||||
pub fn list_body_ids(&self) -> Result<Vec<String>> {
|
||||
let mut stmt = self.conn.prepare("SELECT DISTINCT body_id FROM body_chunks")?;
|
||||
let ids = stmt
|
||||
.query_map([], |row| row.get(0))?
|
||||
.collect::<std::result::Result<Vec<String>, _>>()?;
|
||||
Ok(ids)
|
||||
}
|
||||
|
||||
/// Delete all chunks for a body.
|
||||
pub fn delete_chunks(&self, body_id: &str) -> Result<()> {
|
||||
self.conn.execute("DELETE FROM body_chunks WHERE body_id = ?1", params![body_id])?;
|
||||
|
||||
@@ -65,6 +65,21 @@ impl<'a> ClientDb<'a> {
|
||||
Ok(self.ctx.find_many(col, value, limit)?)
|
||||
}
|
||||
|
||||
/// Bulk-delete all rows matching a column value WITHOUT recording model
|
||||
/// changes or emitting events. Only use for cascades whose deletion is
|
||||
/// implied by a recorded parent delete (e.g. workspace children — see
|
||||
/// [`ModelChangeEvent::Delete`]).
|
||||
pub(crate) fn delete_many_untracked<M>(
|
||||
&self,
|
||||
col: impl IntoColumnRef,
|
||||
value: impl Into<SimpleExpr>,
|
||||
) -> Result<usize>
|
||||
where
|
||||
M: UpsertModelInfo,
|
||||
{
|
||||
Ok(self.ctx.delete_many::<M>(col, value)?)
|
||||
}
|
||||
|
||||
// --- Write operations (with event recording) ---
|
||||
|
||||
pub(crate) fn upsert<M>(&self, model: &M, source: &UpdateSource) -> Result<M>
|
||||
|
||||
@@ -44,6 +44,56 @@ impl<'a> ClientDb<'a> {
|
||||
Ok(count)
|
||||
}
|
||||
|
||||
/// 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
|
||||
/// directory, so orphans accumulate; this runs in the background at startup.
|
||||
///
|
||||
/// Safe against in-flight sends: the response row is created before its
|
||||
/// body file or chunks are written.
|
||||
///
|
||||
/// Returns the number of orphaned bodies deleted.
|
||||
pub fn delete_orphaned_response_bodies(
|
||||
&self,
|
||||
blobs: &BlobManager,
|
||||
responses_dir: &std::path::Path,
|
||||
) -> Result<usize> {
|
||||
let mut deleted = 0;
|
||||
|
||||
// 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) {
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
if !path.is_file() {
|
||||
continue;
|
||||
}
|
||||
let Some(response_id) = path.file_name().and_then(|n| n.to_str()) else {
|
||||
continue;
|
||||
};
|
||||
if self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_some()
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if fs::remove_file(&path).is_ok() {
|
||||
deleted += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(deleted)
|
||||
}
|
||||
|
||||
/// Returns the number of responses deleted.
|
||||
pub fn delete_all_http_responses_for_workspace(
|
||||
&self,
|
||||
@@ -118,3 +168,68 @@ impl<'a> ClientDb<'a> {
|
||||
if response.id.is_empty() { Ok(response.clone()) } else { self.upsert(response, source) }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::blob_manager::BodyChunk;
|
||||
use crate::init_in_memory;
|
||||
use crate::models::{HttpRequest, HttpResponse, Workspace};
|
||||
use crate::util::UpdateSource;
|
||||
|
||||
#[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 source = &UpdateSource::Background;
|
||||
let workspace = db
|
||||
.upsert_workspace(&Workspace { name: "GC Test".to_string(), ..Default::default() }, source)
|
||||
.expect("Failed to upsert workspace");
|
||||
let request = db
|
||||
.upsert_http_request(
|
||||
&HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() },
|
||||
source,
|
||||
)
|
||||
.expect("Failed to upsert request");
|
||||
|
||||
let live = 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_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_body_id, 0, b"live".to_vec())).unwrap();
|
||||
blob_ctx.insert_chunk(&BodyChunk::new("rs_gone.request", 0, b"dead".to_vec())).unwrap();
|
||||
}
|
||||
|
||||
let dir = std::env::temp_dir().join(format!("yaak-blob-gc-test-{}", live.id));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
std::fs::write(dir.join(&live.id), b"live").unwrap();
|
||||
std::fs::write(dir.join("rs_gone"), b"dead").unwrap();
|
||||
|
||||
let deleted = db
|
||||
.delete_orphaned_response_bodies(&blob_manager, &dir)
|
||||
.expect("Failed to GC response bodies");
|
||||
assert_eq!(deleted, 2);
|
||||
|
||||
// Live data survives, orphans are gone
|
||||
let blob_ctx = blob_manager.connect();
|
||||
assert!(blob_ctx.body_exists(&live_body_id).unwrap());
|
||||
assert!(!blob_ctx.body_exists("rs_gone.request").unwrap());
|
||||
assert!(dir.join(&live.id).exists());
|
||||
assert!(!dir.join("rs_gone").exists());
|
||||
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,7 +103,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn records_model_changes_for_upsert_and_delete() {
|
||||
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
|
||||
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
|
||||
let db = query_manager.connect();
|
||||
|
||||
let workspace = db
|
||||
@@ -128,7 +128,7 @@ mod tests {
|
||||
));
|
||||
assert!(matches!(created_changes[0].payload.update_source, UpdateSource::Sync));
|
||||
|
||||
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync)
|
||||
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
|
||||
.expect("Failed to delete workspace");
|
||||
|
||||
let all_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
|
||||
@@ -178,7 +178,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn list_model_changes_since_uses_timestamp_with_id_tiebreaker() {
|
||||
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
|
||||
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
|
||||
let db = query_manager.connect();
|
||||
|
||||
let workspace = db
|
||||
@@ -192,7 +192,7 @@ mod tests {
|
||||
&UpdateSource::Sync,
|
||||
)
|
||||
.expect("Failed to upsert workspace");
|
||||
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync)
|
||||
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
|
||||
.expect("Failed to delete workspace");
|
||||
|
||||
let all = db.list_model_changes_after(0, 10).expect("Failed to list changes");
|
||||
|
||||
@@ -1,10 +1,17 @@
|
||||
use crate::blob_manager::BlobManager;
|
||||
use crate::client_db::ClientDb;
|
||||
use crate::error::Result;
|
||||
use crate::models::{
|
||||
AnyModel, EnvironmentIden, FolderIden, GrpcRequestIden, HttpRequestHeader, HttpRequestIden,
|
||||
ResolvedHttpRequestSettings, ResolvedSetting, WebsocketRequestIden, Workspace, WorkspaceIden,
|
||||
AnyModel, CookieJar, CookieJarIden, Environment, EnvironmentIden, Folder, FolderIden,
|
||||
GraphQlIntrospection, GraphQlIntrospectionIden, GrpcConnection, GrpcConnectionIden, GrpcEvent,
|
||||
GrpcEventIden, GrpcRequest, GrpcRequestIden, HttpRequest, HttpRequestHeader, HttpRequestIden,
|
||||
HttpResponse, HttpResponseEvent, HttpResponseEventIden, HttpResponseIden,
|
||||
ResolvedHttpRequestSettings, ResolvedSetting, SyncState, SyncStateIden, WebsocketConnection,
|
||||
WebsocketConnectionIden, WebsocketEvent, WebsocketEventIden, WebsocketRequest,
|
||||
WebsocketRequestIden, Workspace, WorkspaceIden, WorkspaceMeta, WorkspaceMetaIden,
|
||||
};
|
||||
use crate::util::UpdateSource;
|
||||
use log::warn;
|
||||
use serde_json::Value;
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
@@ -32,37 +39,99 @@ impl<'a> ClientDb<'a> {
|
||||
Ok(workspaces)
|
||||
}
|
||||
|
||||
/// Delete a workspace and everything in it.
|
||||
///
|
||||
/// Children are bulk-deleted with one statement per table and are NOT
|
||||
/// individually recorded in model_changes or emitted as events — the single
|
||||
/// workspace delete event implies the subtree (see [`ModelChangeEvent::Delete`]).
|
||||
/// This keeps huge workspaces (thousands of requests) fast and avoids
|
||||
/// flooding event consumers.
|
||||
pub fn delete_workspace(
|
||||
&self,
|
||||
workspace: &Workspace,
|
||||
source: &UpdateSource,
|
||||
blobs: &BlobManager,
|
||||
) -> Result<Workspace> {
|
||||
for m in self.find_many(HttpRequestIden::WorkspaceId, &workspace.id, None)? {
|
||||
self.delete_http_request(&m, source)?;
|
||||
let wid = workspace.id.as_str();
|
||||
|
||||
// Collect response cleanup targets before their rows disappear. The actual
|
||||
// cleanup runs at the end: response bodies live on disk and in the blob DB,
|
||||
// which don't participate in this transaction, so removing them must wait
|
||||
// until every statement that could fail (and roll back the rows) is done.
|
||||
let responses = self.find_many::<HttpResponse>(HttpResponseIden::WorkspaceId, wid, None)?;
|
||||
|
||||
// Sync and the CLI call this on a plain connection where each statement
|
||||
// would otherwise commit on its own, leaving a partially-deleted workspace
|
||||
// if one fails. A savepoint makes the cascade atomic there, and nests
|
||||
// harmlessly inside the interactive path's transaction.
|
||||
let conn = self.conn().resolve();
|
||||
conn.execute_batch("SAVEPOINT delete_workspace")?;
|
||||
|
||||
let result: Result<Workspace> = (|| {
|
||||
self.delete_many_untracked::<HttpResponseEvent>(
|
||||
HttpResponseEventIden::WorkspaceId,
|
||||
wid,
|
||||
)?;
|
||||
self.delete_many_untracked::<HttpResponse>(HttpResponseIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<HttpRequest>(HttpRequestIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<GrpcEvent>(GrpcEventIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<GrpcConnection>(GrpcConnectionIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<GrpcRequest>(GrpcRequestIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<WebsocketEvent>(WebsocketEventIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<WebsocketConnection>(
|
||||
WebsocketConnectionIden::WorkspaceId,
|
||||
wid,
|
||||
)?;
|
||||
self.delete_many_untracked::<WebsocketRequest>(WebsocketRequestIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<GraphQlIntrospection>(
|
||||
GraphQlIntrospectionIden::WorkspaceId,
|
||||
wid,
|
||||
)?;
|
||||
self.delete_many_untracked::<Folder>(FolderIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<Environment>(EnvironmentIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<CookieJar>(CookieJarIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<SyncState>(SyncStateIden::WorkspaceId, wid)?;
|
||||
self.delete_many_untracked::<WorkspaceMeta>(WorkspaceMetaIden::WorkspaceId, wid)?;
|
||||
self.delete(workspace, source)
|
||||
})();
|
||||
|
||||
let deleted = match result {
|
||||
Ok(deleted) => {
|
||||
conn.execute_batch("RELEASE delete_workspace")?;
|
||||
deleted
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = conn
|
||||
.execute_batch("ROLLBACK TO delete_workspace; RELEASE delete_workspace");
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
// Best-effort cleanup of response bodies (disk files and blob chunks).
|
||||
// Failures only orphan unreferenced data, and are logged.
|
||||
let blob_ctx = blobs.connect();
|
||||
for m in responses {
|
||||
if let Some(p) = m.body_path {
|
||||
if let Err(e) = std::fs::remove_file(&p) {
|
||||
warn!("Failed to delete response body file {p:?}: {e}");
|
||||
}
|
||||
}
|
||||
if let Err(e) = blob_ctx.delete_chunks_like(&format!("{}.%", m.id)) {
|
||||
warn!("Failed to delete blobs for response {}: {e}", m.id);
|
||||
}
|
||||
}
|
||||
|
||||
for m in self.find_many(GrpcRequestIden::WorkspaceId, &workspace.id, None)? {
|
||||
self.delete_grpc_request(&m, source)?;
|
||||
}
|
||||
|
||||
for m in self.find_many(WebsocketRequestIden::FolderId, &workspace.id, None)? {
|
||||
self.delete_websocket_request(&m, source)?;
|
||||
}
|
||||
|
||||
for m in self.find_many(FolderIden::WorkspaceId, &workspace.id, None)? {
|
||||
self.delete_folder(&m, source)?;
|
||||
}
|
||||
|
||||
for m in self.find_many(EnvironmentIden::WorkspaceId, &workspace.id, None)? {
|
||||
self.delete_environment(&m, source)?;
|
||||
}
|
||||
|
||||
self.delete(workspace, source)
|
||||
Ok(deleted)
|
||||
}
|
||||
|
||||
pub fn delete_workspace_by_id(&self, id: &str, source: &UpdateSource) -> Result<Workspace> {
|
||||
pub fn delete_workspace_by_id(
|
||||
&self,
|
||||
id: &str,
|
||||
source: &UpdateSource,
|
||||
blobs: &BlobManager,
|
||||
) -> Result<Workspace> {
|
||||
let workspace = self.get_workspace(id)?;
|
||||
self.delete_workspace(&workspace, source)
|
||||
self.delete_workspace(&workspace, source, blobs)
|
||||
}
|
||||
|
||||
pub fn upsert_workspace(&self, w: &Workspace, source: &UpdateSource) -> Result<Workspace> {
|
||||
|
||||
Reference in New Issue
Block a user