Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118//! `happyview.delete-collection` — delete every record in a collection.//!//! Batched deliberately. One unbounded `DELETE` is a single transaction whose//! every dirtied page accumulates in the WAL until it commits, which is how a//! large delete inflates disk usage instead of reducing it. Committing per batch//! releases the write lock (so ingest can interleave) and lets the WAL//! checkpoint between batches.
use crate::AppState;use crate::db::{DatabaseBackend, adapt_sql};
use super::super::{Job, db};use super::NativeOutcome;
/// Rows per transaction.const BATCH_SIZE: i64 = 5000;
pub async fn run(state: &AppState, job: &Job) -> NativeOutcome { let collection = match job.input.get("collection").and_then(|v| v.as_str()) { Some(c) if !c.is_empty() => c.to_string(), _ => return NativeOutcome::Failed("input.collection is required".into()), }; let backend = state.db_backend;
let count_sql = adapt_sql( "SELECT COUNT(*) FROM happyview_records WHERE collection = ?", backend, ); let total: i64 = match crate::db::query_as::<(i64,)>(&count_sql) .bind(&collection) .fetch_one(&state.db) .await { Ok((n,)) => n, Err(e) => return NativeOutcome::Failed(format!("failed to count records: {e}")), };
let delete_sql = adapt_sql( "DELETE FROM happyview_records WHERE uri IN \ (SELECT uri FROM happyview_records WHERE collection = ? LIMIT ?)", backend, );
let mut deleted: i64 = 0; let mut stopped: Option<&'static str> = None;
loop { if let Some(reason) = db::should_stop(state, &job.id).await { tracing::info!(job_id = %job.id, reason, deleted, "delete-collection stopping"); stopped = Some(reason); break; }
let affected = match crate::db::query(&delete_sql) .bind(&collection) .bind(BATCH_SIZE) .execute(&state.db) .await { Ok(r) => r.rows_affected(), Err(e) => return NativeOutcome::Failed(format!("failed to delete batch: {e}")), }; deleted += affected as i64;
let _ = db::update_progress( state, &job.id, &serde_json::json!({ "deleted": deleted, "total": total }), ) .await;
// A batch smaller than BATCH_SIZE means the collection, as it stood // when this statement ran, has been drained. Waiting for a batch of // exactly zero instead would never happen on a collection still // receiving inserts (e.g. from Jetstream) — the loop would spin // indefinitely, holding the write lock and starving every other // queued job behind this one. Records that arrive after a job was // asked to delete a collection are out of scope for that job. if (affected as i64) < BATCH_SIZE { break; } }
if let Some(reason) = stopped { // finalize()'s pause/cancel arms only set status — they don't persist // a result, so this is the only place the partial count and the // reason it stopped are recorded. let _ = db::update_progress( state, &job.id, &serde_json::json!({ "deleted": deleted, "total": total, "stopped": reason }), ) .await; }
// Incremental auto-vacuum makes reclamation possible but never automatic; // this is what actually returns the freed pages to the filesystem. It is a // no-op on instances that have not yet run the one-time vacuum. Runs // whether the loop drained the collection or was interrupted — a cancel is // the operator's escape hatch from a slow delete, and it should reclaim // whatever was freed up to that point rather than reclaiming nothing. if backend == DatabaseBackend::Sqlite && let Err(e) = crate::db::query("PRAGMA incremental_vacuum") .execute(&state.db) .await { tracing::warn!(error = %e, "incremental_vacuum after delete failed"); }
match stopped { Some(reason) => NativeOutcome::Completed(serde_json::json!({ "deleted": deleted, "stopped": reason, })), None => NativeOutcome::Completed(serde_json::json!({ "deleted": deleted })), }}