Reclaim hotblocks disk: point-delete cleanup + startup DeleteFilesInRange

Cargo.toml

@@ -14,6 +14,7 @@ members = [
     "crates/primitives",
     "crates/query",
     "crates/query-example",
+    "crates/reclaim-measure",
     "crates/storage"
 ]
 resolver = "2"

Dockerfile

@@ -13,12 +13,15 @@ ADD crates crates
 
 
 FROM builder AS hotblocks-builder
-RUN cargo build -p sqd-hotblocks --release
+RUN cargo build -p sqd-hotblocks -p reclaim-measure --release
 
 
 FROM rust AS hotblocks
 WORKDIR /app
 COPY --from=hotblocks-builder /app/target/release/sqd-hotblocks .
+# Read-only probe: opens the live db as a RocksDB secondary instance and reports how much
+# --startup-disk-reclaim would free. Run it in the pod before turning that flag on.
+COPY --from=hotblocks-builder /app/target/release/reclaim-measure .
 ENTRYPOINT ["/app/sqd-hotblocks"]
 
 

crates/hotblocks/src/cli.rs

@@ -61,6 +61,29 @@ pub struct CLI {
     #[arg(long, value_name = "N", default_value = "10")]
     pub rocksdb_keep_log_file_num: usize,
 
+    /// Concurrent RocksDB background flush + compaction jobs.
+    /// Defaults to the core count, clamped to 2..=8.
+    #[arg(long, value_name = "N")]
+    pub rocksdb_max_background_jobs: Option<usize>,
+
+    /// Rewrite every table SST older than this, collecting dead data that never made a file
+    /// tombstone-dense enough for the deletion collector. Lower means faster reclaim and
+    /// proportionally more write amplification. 0 disables it, leaving RocksDB's 30-day
+    /// `ttl` as the only backstop.
+    #[arg(long, value_name = "SECS", default_value_t = sqd_storage::db::DEFAULT_PERIODIC_COMPACTION_SECS)]
+    pub rocksdb_periodic_compaction_secs: u64,
+
+    /// Reclaim dead disk at startup: purge orphaned dirty-table markers, then unlink whole
+    /// SST files below the live-table watermark. Off by default -- the unlink ignores
+    /// snapshots, so it is only safe before any query exists. Use the `reclaim-measure`
+    /// binary (shipped in this image) to size the win first.
+    ///
+    /// Note that this runs *after* the database opens, and opening replays the WAL and
+    /// flushes it to L0. A volume at literally zero free bytes needs a few hundred MB
+    /// cleared by hand before the process can get far enough to reclaim anything.
+    #[arg(long)]
+    pub startup_disk_reclaim: bool,
+
     /// 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")]
@@ -80,12 +103,19 @@ impl CLI {
     pub async fn build_app(&self) -> anyhow::Result<App> {
         let datasets = DatasetConfig::read_config_file(&self.datasets).context("failed to read datasets config")?;
 
-        let db = DatabaseSettings::default()
+        let mut settings = DatabaseSettings::default()
             .with_data_cache_size(self.data_cache_size)
             .with_rocksdb_stats(self.rocksdb_stats)
             .with_direct_io(!self.rocksdb_disable_direct_io)
             .with_max_log_file_size(self.rocksdb_max_log_file_size)
             .with_keep_log_file_num(self.rocksdb_keep_log_file_num)
+            .with_periodic_compaction_secs(self.rocksdb_periodic_compaction_secs);
+
+        if let Some(jobs) = self.rocksdb_max_background_jobs {
+            settings = settings.with_max_background_jobs(jobs);
+        }
+
+        let db = settings
             .open(&self.database_dir)
             .map(Arc::new)
             .context("failed to open rocksdb database")?;
@@ -101,7 +131,9 @@ impl CLI {
             .filter_map(|(id, cfg)| (cfg.retention_strategy == RetentionConfig::Api).then_some(*id))
             .collect();
 
-        let data_service = DataService::start(db.clone(), datasets).await.map(Arc::new)?;
+        let data_service = DataService::start(db.clone(), datasets, self.startup_disk_reclaim)
+            .await
+            .map(Arc::new)?;
 
         let query_service = {
             let mut builder = QueryService::builder(db.clone());

crates/hotblocks/src/data_service.rs

@@ -7,7 +7,7 @@ use anyhow::{Context, anyhow};
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use sqd_data_client::reqwest::ReqwestDataClient;
 use sqd_storage::db::DatasetId;
-use tracing::{error, info};
+use tracing::{debug, error, info, warn};
 
 use crate::{
     dataset_config::{DatasetConfig, RetentionConfig},
@@ -23,14 +23,24 @@ pub struct DataService {
 }
 
 impl DataService {
-    pub async fn start(db: DBRef, datasets: BTreeMap<DatasetId, DatasetConfig>) -> anyhow::Result<Self> {
-        let all_datasets = db.get_all_datasets()?;
-        for dataset in all_datasets {
-            if !datasets.contains_key(&dataset.id) {
-                info!("deleting unconfigured dataset {}", dataset.id);
-                if let Err(err) = db.delete_dataset(dataset.id) {
-                    error!("failed to delete dataset {}: {}", dataset.id, err);
-                }
+    pub async fn start(
+        db: DBRef,
+        datasets: BTreeMap<DatasetId, DatasetConfig>,
+        disk_reclaim: bool
+    ) -> anyhow::Result<Self> {
+        let unconfigured: Vec<DatasetId> = db
+            .get_all_datasets()?
+            .into_iter()
+            .filter(|ds| !datasets.contains_key(&ds.id))
+            .map(|ds| ds.id)
+            .collect();
+
+        // Must run before any controller spawns -- see `startup_disk_recovery`.
+        {
+            let db = db.clone();
+            let recovery = tokio::task::spawn_blocking(move || startup_disk_recovery(&db, &unconfigured, disk_reclaim));
+            if let Err(err) = recovery.await {
+                error!(error =? err, "startup disk recovery panicked");
             }
         }
 
@@ -80,3 +90,54 @@ impl DataService {
             .ok_or(UnknownDataset { dataset_id })
     }
 }
+
+/// Startup-only disk recovery; must run before any ingest or query exists (the file
+/// unlink ignores snapshots, and the orphan purge treats every dirty marker as an orphan
+/// from a dead build).
+///
+/// `disk_reclaim` gates both reclaim steps (`--startup-disk-reclaim`, off by default);
+/// deleting unconfigured datasets always runs. Ordering matters on a near-full disk:
+/// 1. an unlink pass first -- it needs no scratch space, so it frees below-watermark space
+///    where every write below would fail with ENOSPC;
+/// 2. bookkeeping writes that lift the reclaim watermark: purge orphan dirty markers,
+///    delete unconfigured datasets;
+/// 3. a second unlink pass to free whatever step 2 unpinned.
+///
+/// Every step is best-effort: a failure leaves the watermark pinned until a later startup
+/// succeeds, but never blocks startup.
+///
+/// Does not rescue a volume at literally zero free bytes: the database has already opened
+/// by the time we get here, and opening replays the WAL and flushes it to L0.
+fn startup_disk_recovery(db: &DBRef, unconfigured: &[DatasetId], disk_reclaim: bool) {
+    if disk_reclaim {
+        if let Err(err) = db.reclaim_disk_space() {
+            error!(error =? err, "startup disk reclaim (first pass) failed");
+        }
+
+        match db.purge_orphan_dirty_tables() {
+            Ok(0) => {}
+            Ok(n) => info!("purged {n} orphan dirty table(s) left by an interrupted build"),
+            Err(err) => warn!(error =? err, "failed to purge orphan dirty tables")
+        }
+    } else {
+        // FUTURE: ungate the orphan purge once the rollout measurement is done. It is safe
+        // without the unlink, and while gated an interrupted build leaks its data for good;
+        // it shares the flag only so `reclaim-measure` can still contrast the two watermarks.
+        info!("startup disk reclaim is off; enable with --startup-disk-reclaim");
+    }
+
+    for dataset_id in unconfigured {
+        info!("deleting unconfigured dataset {dataset_id}");
+        if let Err(err) = db.delete_dataset(*dataset_id) {
+            error!("failed to delete dataset {dataset_id}: {err}; its chunks keep pinning the reclaim watermark");
+        }
+    }
+
+    if disk_reclaim {
+        if let Err(err) = db.reclaim_disk_space() {
+            error!(error =? err, "startup disk reclaim (second pass) failed");
+        }
+    }
+
+    debug!("startup disk recovery complete");
+}

crates/hotblocks/src/main.rs

@@ -36,6 +36,10 @@ fn main() -> anyhow::Result<()> {
         .block_on(async {
             let app = args.build_app().await?;
 
+            // NB: startup disk recovery (orphan purge + file unlink, gated by
+            // --startup-disk-reclaim) already ran inside `build_app` -> `DataService::start`,
+            // before any controller spawned.
+
             tokio::spawn(db_cleanup_task(app.db.clone()));
 
             let api = build_api(app);
@@ -89,24 +93,46 @@ async fn shutdown_signal() {
     }
 }
 
+const CLEANUP_INTERVAL: Duration = Duration::from_secs(10);
+/// Longer pause after a failed tick, so a persistent error (e.g. full disk) doesn't
+/// busy-loop failing writes.
+const CLEANUP_ERROR_BACKOFF: Duration = Duration::from_secs(30);
+
+/// Routine Phase-1 cleanup: point-delete deleted tables' data every tick so compaction
+/// can reclaim their space -- in normal operation the entire runtime reclaim path.
+///
+/// The physical file unlink (`Database::reclaim_disk_space`) is NOT run here: it ignores
+/// snapshots and could break an in-flight query, so it runs only at startup. FUTURE:
+/// trigger it here under disk pressure too, accepting that risk.
 #[instrument(name = "db_cleanup", skip_all)]
 async fn db_cleanup_task(db: DBRef) {
-    tokio::time::sleep(Duration::from_secs(10)).await;
+    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;
-        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;
+
+        let failed = match result {
+            Ok(Ok(purged)) => {
+                if purged > 0 {
+                    debug!("cleanup: logically purged {purged} table(s)");
                 }
+                false
             }
-            Ok(Err(err)) => error!(error =? err, "database cleanup task failed"),
-            Err(_) => error!("database cleanup task panicked")
-        }
+            Ok(Err(err)) => {
+                error!(error =? err, "database cleanup failed");
+                true
+            }
+            Err(_) => {
+                error!("database cleanup task panicked");
+                true
+            }
+        };
+
+        tokio::time::sleep(if failed {
+            CLEANUP_ERROR_BACKOFF
+        } else {
+            CLEANUP_INTERVAL
+        })
+        .await;
     }
 }

crates/reclaim-measure/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "reclaim-measure"
+version = "0.0.0"
+edition = "2021"
+publish = false
+
+[[bin]]
+name = "reclaim-measure"
+path = "src/main.rs"
+
+[dependencies]
+sqd-storage = { path = "../storage" }
+rocksdb = { version = "0.24.0", features = ["jemalloc"] }
+borsh = { workspace = true }
+anyhow = { workspace = true }
+
+[lints]
+workspace = true