From 840b7de3abac5199cabdae526482484b8520fd33 Mon Sep 17 00:00:00 2001
From: dawn <90008@klbr.net>
Date: Mon, 22 Jun 2026 02:45:19 +0300
Subject: [PATCH] [api] guard db compact route against concurrent compactions

---
 src/api/db.rs  | 12 +++++++-----
 src/db/mod.rs  | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
 src/db/open.rs |  1 +
 3 files changed, 58 insertions(+), 5 deletions(-)

diff --git a/src/api/db.rs b/src/api/db.rs
index f0b2dfb..d0760c5 100644
--- a/src/api/db.rs
+++ b/src/api/db.rs
@@ -21,10 +21,12 @@ pub async fn handle_train_dict(
 pub async fn handle_compact(
     State(hydrant): State<Hydrant>,
 ) -> Result<StatusCode, (StatusCode, String)> {
-    hydrant
-        .db
-        .compact()
-        .await
-        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    if let Err(e) = hydrant.db.compact().await {
+        let err_msg = e.to_string();
+        if err_msg.contains("already in progress") {
+            return Err((StatusCode::CONFLICT, err_msg));
+        }
+        return Err((StatusCode::INTERNAL_SERVER_ERROR, err_msg));
+    }
     Ok(StatusCode::OK)
 }
diff --git a/src/db/mod.rs b/src/db/mod.rs
index 3903b05..f677ab9 100644
--- a/src/db/mod.rs
+++ b/src/db/mod.rs
@@ -94,6 +94,7 @@ pub struct Db {
     count_delta_in_flight: Arc<Mutex<BTreeSet<u64>>>,
     #[cfg(feature = "indexer")]
     lifecycle_count_lock: Arc<Mutex<()>>,
+    pub(crate) compaction_running: Arc<std::sync::atomic::AtomicBool>,
 }
 
 impl Db {
@@ -107,6 +108,22 @@ impl Db {
     }
 
     pub async fn compact(&self) -> Result<()> {
+        use std::sync::atomic::Ordering;
+        if self
+            .compaction_running
+            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+            .is_err()
+        {
+            return Err(miette::miette!("compaction already in progress"));
+        }
+        struct Guard(Arc<std::sync::atomic::AtomicBool>);
+        impl Drop for Guard {
+            fn drop(&mut self) {
+                self.0.store(false, Ordering::Release);
+            }
+        }
+        let _guard = Guard(self.compaction_running.clone());
+
         let compact = |ks: Keyspace| async move {
             tokio::task::spawn_blocking(move || ks.major_compact().into_diagnostic())
                 .await
@@ -301,3 +318,36 @@ pub fn load_persisted_firehose_sources(
     }
     Ok(sources)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::Config;
+    use std::sync::atomic::Ordering;
+
+    #[tokio::test]
+    async fn test_db_compact_concurrency_guard() -> Result<()> {
+        let tmp = tempfile::tempdir().into_diagnostic()?;
+        let mut cfg = Config::default();
+        cfg.database_path = tmp.path().to_path_buf();
+
+        let db = Db::open(&cfg)?;
+
+        // Manually mark compaction as running
+        db.compaction_running.store(true, Ordering::SeqCst);
+
+        // Attempting to compact should now fail
+        let res = db.compact().await;
+        assert!(res.is_err());
+        assert!(res.unwrap_err().to_string().contains("already in progress"));
+
+        // Release the lock
+        db.compaction_running.store(false, Ordering::SeqCst);
+
+        // Compacting should now succeed
+        let res = db.compact().await;
+        assert!(res.is_ok());
+
+        Ok(())
+    }
+}
diff --git a/src/db/open.rs b/src/db/open.rs
index e7b7807..ef1021a 100644
--- a/src/db/open.rs
+++ b/src/db/open.rs
@@ -411,6 +411,7 @@ impl Db {
             count_delta_in_flight: Arc::new(Mutex::new(BTreeSet::new())),
             #[cfg(feature = "indexer")]
             lifecycle_count_lock: Arc::new(Mutex::new(())),
+            compaction_running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
         };
 
         migration::run(&this)?;
-- 
2.51.2