Reclaim hotblocks disk via DeleteFilesInRange + range deletes (NET-819, NET-798)

Replace the per-key tombstone purge with a two-phase table cleanup that can
actually reclaim disk space, including at a full disk.

Phase 1 (logical, snapshot-safe): cleanup() drops each deleted table with a
single range tombstone (OptimisticTransactionDB::delete_range_cf) instead of
millions of point deletes, then marks it reclaim-pending. Range tombstones
respect snapshots, so in-flight queries are unaffected.

Phase 2 (physical): reclaim_disk_space(grace) unlinks whole SST files below the
live watermark (min live TableId across all chunks + dirty tables) via
DeleteFilesInRange. It performs no writes and needs no scratch space, so it
makes progress even at 100% disk. A per-table deletion timestamp in
CF_DELETED_TABLES gates the unlink behind a grace period -- the file unlink
ignores snapshots, so grace must exceed the max query/snapshot lifetime.

Also:
- TABLES CF: compact-on-deletion collector + 24h periodic compaction so
  compaction finds tombstone-heavy / boundary files (NET-819).
- hotblocks: the cleanup loop runs both phases each tick and backs off on
  error instead of busy-looping failing writes; startup reclaims unconfigured
  datasets' files before serving with a zero grace, since no readers exist yet
  (NET-798). New --reclaim-grace-secs flag (default 15m).

Tests: logical-delete snapshot safety, physical reclaim after grace, watermark
pinning by a live table, idempotency, end-to-end delete_dataset + reclaim, and
a value-codec/migration unit test.

Supersedes the sync_dataset_cleanup flag (draft PR #79).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mo4islona

crates/hotblocks/src/cli.rs

@@ -61,6 +61,13 @@ pub struct CLI {
     #[arg(long, value_name = "N", default_value = "10")]
     pub rocksdb_keep_log_file_num: usize,
 
+    /// Grace period, in seconds, before a deleted table's data files are
+    /// physically unlinked by the background cleanup. MUST exceed the longest
+    /// query/snapshot lifetime: file unlink ignores snapshots, so a shorter
+    /// value can silently drop rows from an in-flight query. Default: 15 min.
+    #[arg(long, value_name = "SECS", default_value = "900")]
+    pub reclaim_grace_secs: u64,
+
     /// Known client IDs for metrics labeling. Client IDs not in this list
     /// will be reported as "unknown" to prevent metrics cardinality abuse.
     #[arg(long = "known-client", value_name = "ID")]

crates/hotblocks/src/main.rs

@@ -36,7 +36,23 @@ fn main() -> anyhow::Result<()> {
         .block_on(async {
             let app = args.build_app().await?;
 
-            tokio::spawn(db_cleanup_task(app.db.clone()));
+            // Reclaim disk from unconfigured datasets deleted during startup,
+            // before serving. No in-flight queries exist yet, so a zero grace is
+            // safe and frees the most -- this replaces the old blocking purge and
+            // works even at a full disk (file unlink, no writes).
+            {
+                let db = app.db.clone();
+                match tokio::task::spawn_blocking(move || db.reclaim_disk_space(Duration::ZERO)).await {
+                    Ok(Ok(n)) => debug!("startup reclaim freed {n} table(s)"),
+                    Ok(Err(err)) => error!(error =? err, "startup reclaim failed"),
+                    Err(_) => error!("startup reclaim panicked")
+                }
+            }
+
+            tokio::spawn(db_cleanup_task(
+                app.db.clone(),
+                Duration::from_secs(args.reclaim_grace_secs)
+            ));
 
             let api = build_api(app);
 
@@ -89,24 +105,41 @@ async fn shutdown_signal() {
     }
 }
 
+const CLEANUP_INTERVAL: Duration = Duration::from_secs(10);
+/// Backoff after a failed cleanup tick, so a persistent error (e.g. a full disk)
+/// doesn't busy-loop failing writes.
+const CLEANUP_ERROR_BACKOFF: Duration = Duration::from_secs(30);
+
 #[instrument(name = "db_cleanup", skip_all)]
-async fn db_cleanup_task(db: DBRef) {
-    tokio::time::sleep(Duration::from_secs(10)).await;
+async fn db_cleanup_task(db: DBRef, reclaim_grace: Duration) {
+    tokio::time::sleep(CLEANUP_INTERVAL).await;
     loop {
-        debug!("db cleanup started");
         let db = db.clone();
-        let result = tokio::task::spawn_blocking(move || db.cleanup()).await;
+        let result = tokio::task::spawn_blocking(move || {
+            // Phase 1: logical, snapshot-safe purge (one range tombstone per table).
+            let purged = db.cleanup()?;
+            // Phase 2: physically unlink files whose grace has elapsed. No writes
+            // / scratch space, so it makes progress even at a full disk.
+            let reclaimed = db.reclaim_disk_space(reclaim_grace)?;
+            anyhow::Ok((purged, reclaimed))
+        })
+        .await;
+
         match result {
-            Ok(Ok(deleted)) => {
-                if deleted > 0 {
-                    debug!("purged {} tables", deleted)
-                } else {
-                    debug!("nothing to purge, pausing cleanup for 10 seconds");
-                    tokio::time::sleep(Duration::from_secs(10)).await;
+            Ok(Ok((purged, reclaimed))) => {
+                if purged > 0 || reclaimed > 0 {
+                    debug!("cleanup: purged {purged} tables, reclaimed {reclaimed} files");
                 }
+                tokio::time::sleep(CLEANUP_INTERVAL).await;
+            }
+            Ok(Err(err)) => {
+                error!(error =? err, "database cleanup failed; backing off");
+                tokio::time::sleep(CLEANUP_ERROR_BACKOFF).await;
+            }
+            Err(_) => {
+                error!("database cleanup task panicked; backing off");
+                tokio::time::sleep(CLEANUP_ERROR_BACKOFF).await;
             }
-            Ok(Err(err)) => error!(error =? err, "database cleanup task failed"),
-            Err(_) => error!("database cleanup task panicked")
         }
     }
 }

crates/storage/src/db/db.rs

@@ -12,7 +12,7 @@ use super::{
 use crate::db::{
     ops::{perform_dataset_compaction, CompactionStatus},
     read::datasets::list_all_datasets,
-    write::{ops::deleted_deleted_tables, table_builder::TableBuilder, tx::Tx},
+    write::{ops as cleanup_ops, table_builder::TableBuilder, tx::Tx},
     Chunk, DatasetUpdate
 };
 
@@ -36,7 +36,8 @@ pub struct DatabaseSettings {
     direct_io: bool,
     cache_index_and_filter_blocks: bool,
     max_log_file_size: usize,
-    keep_log_file_num: usize
+    keep_log_file_num: usize,
+    auto_compactions: bool
 }
 
 impl Default for DatabaseSettings {
@@ -48,7 +49,8 @@ impl Default for DatabaseSettings {
             direct_io: false,
             cache_index_and_filter_blocks: false,
             max_log_file_size: 10,
-            keep_log_file_num: 10
+            keep_log_file_num: 10,
+            auto_compactions: true
         }
     }
 }
@@ -91,6 +93,14 @@ impl DatabaseSettings {
         self
     }
 
+    /// Enable/disable RocksDB background auto-compaction of the table data.
+    /// Defaults to `true`; mainly for tests that need deterministic control over
+    /// when compaction runs (manual `compact_tables`/reclaim still work).
+    pub fn with_auto_compactions(mut self, yes: bool) -> Self {
+        self.auto_compactions = yes;
+        self
+    }
+
     fn db_options(&self) -> RocksOptions {
         let mut options = RocksOptions::default();
         options.create_if_missing(true);
@@ -141,6 +151,16 @@ impl DatabaseSettings {
         let mut options = RocksOptions::default();
         options.set_block_based_table_factory(&block_based_table_factory);
         options.set_compression_type(rocksdb::DBCompressionType::Lz4);
+        // Help compaction find tombstone-heavy SSTs (table deletes leave range
+        // tombstones), and bound staleness so no dead file lingers uncompacted.
+        // A lone range tombstone over a whole table has low deletion *density*,
+        // so periodic compaction is the real backstop here; the collector mostly
+        // catches denser boundary files. Thresholds are provisional.
+        options.add_compact_on_deletion_collector_factory(128 * 1024, 64 * 1024, 0.5);
+        options.set_periodic_compaction_seconds(24 * 60 * 60);
+        if !self.auto_compactions {
+            options.set_disable_auto_compactions(true);
+        }
         options
     }
 
@@ -287,8 +307,38 @@ impl Database {
         Ok(())
     }
 
+    /// Phase 1 -- logically purge deleted tables (snapshot-safe range tombstones).
+    /// Returns the number of tables logically deleted by this call.
     pub fn cleanup(&self) -> anyhow::Result<usize> {
-        deleted_deleted_tables(&self.db)
+        cleanup_ops::logical_cleanup(&self.db)
+    }
+
+    /// Phase 2 -- physically reclaim disk space from tables deleted longer than
+    /// `grace` ago, by unlinking whole SST files below the live watermark.
+    ///
+    /// Performs no writes and needs no scratch space, so it works at a full disk.
+    /// `grace` MUST exceed the maximum in-flight query/snapshot lifetime: file
+    /// unlinks ignore snapshots, so a shorter grace can drop rows from a running
+    /// query. Use [`Duration::ZERO`] only where no readers exist (e.g. startup).
+    /// Returns the number of deleted-table records reclaimed.
+    pub fn reclaim_disk_space(&self, grace: std::time::Duration) -> anyhow::Result<usize> {
+        cleanup_ops::reclaim_disk_space(&self.db, grace)
+    }
+
+    /// Flush the table-data column family's memtable to SST files. Useful before
+    /// a reclaim/shutdown so freshly written data is on disk as files.
+    pub fn flush(&self) -> anyhow::Result<()> {
+        self.db.flush_cf(self.db.cf_handle(CF_TABLES).unwrap())?;
+        Ok(())
+    }
+
+    /// Force a full compaction of the table-data column family. Unlike
+    /// [`Database::reclaim_disk_space`] this *writes* (needs scratch space, so it
+    /// is not safe at a full disk); it exists to push data into the bottom level
+    /// and to rewrite tombstone-heavy boundary files that file-unlink can't reach.
+    pub fn compact_tables(&self) {
+        let cf = self.db.cf_handle(CF_TABLES).unwrap();
+        self.db.compact_range_cf(cf, None::<&[u8]>, None::<&[u8]>);
     }
 
     pub fn get_statistics(&self) -> Option<String> {