diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 2feb4798..dd47a1f8 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -23,7 +23,7 @@ jobs: # self-hosted host. Everything else (in-repo branches, push, dispatch) uses the faster # self-hosted runner with a warm cargo cache. Runner groups can't gate by trigger, so the # split has to live here in the workflow — not in the approval setting. - runs-on: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork) && fromJSON('["ubuntu-latest"]') || fromJSON('["self-hosted", "dev-server"]') }} + runs-on: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) && fromJSON('["ubuntu-latest"]') || fromJSON('["self-hosted", "dev-server"]') }} if: github.event_name == 'push' || !github.event.pull_request.draft steps: - uses: actions/checkout@v4 @@ -50,7 +50,7 @@ jobs: name: Test # Fork PRs stay on GitHub-hosted (untrusted code); trusted events use self-hosted. See the # clippy job above for the rationale. - runs-on: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork) && fromJSON('["ubuntu-latest"]') || fromJSON('["self-hosted", "dev-server"]') }} + runs-on: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) && fromJSON('["ubuntu-latest"]') || fromJSON('["self-hosted", "dev-server"]') }} if: github.event_name == 'push' || !github.event.pull_request.draft steps: - uses: actions/checkout@v4 diff --git a/Cargo.lock b/Cargo.lock index 4dc9d909..ae52490c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4892,6 +4892,7 @@ name = "sqd-hotblocks" version = "0.1.0" dependencies = [ "anyhow", + "arrow", "async-stream", "axum", "bytes", @@ -4914,6 +4915,7 @@ dependencies = [ "sqd-primitives", "sqd-query", "sqd-storage", + "tempfile", "tikv-jemallocator", "tokio", "tower-http", @@ -5033,6 +5035,7 @@ dependencies = [ "sqd-array", "sqd-primitives", "tempfile", + "tracing", "uuid", ] diff --git a/crates/hotblocks-harness/README.md b/crates/hotblocks-harness/README.md index 5eea9f91..716776c5 100644 --- a/crates/hotblocks-harness/README.md +++ b/crates/hotblocks-harness/README.md @@ -23,6 +23,8 @@ is what makes the crash/restart and shutdown classes expressible at all. cargo test -p sqd-hotblocks-harness # the harness's own unit tests (model, chain, simulator) cargo test -p sqd-hotblocks --test ct1_happy_path # CT-1 — the Phase 0 exit criterion cargo test -p sqd-hotblocks --test ct9_source_faults +# Explicit endurance lane (ignored by default; includes the reclaim convergence wait): +cargo test -p sqd-hotblocks --test ct7_stall_and_churn ct7_churn_soak -- --ignored ``` The CT tests live in `crates/hotblocks/tests/` because only a test inside that package gets @@ -127,11 +129,10 @@ Fixed in `crates/data-client/src/reqwest/lines.rs`; pinned by a unit test there normative CONFLICT recovery of 04 §7. What is missing is the scripts. - **CT-5 (error taxonomy)** — `Model::predict_query` returns the outcome class for any query; `Client::query` already classifies every response. What is missing is the request matrix. -- **CT-7 (soak/space)** — needs `OB-6`-style space accounting; the retention model transition - (`Model::retain`) is implemented but the comparator compares the first block *exactly*, which - is wrong once retention starts trimming: the service trims whole chunks, so its window may be - larger than the model's (legal under RS-3/RS-4, `P-RETENTION-SLACK`). Give the comparator that - tolerance before writing retention tests. +- **CT-7 (soak)** — the ignored S4 runner in `ct7_stall_and_churn` drives API-controlled + moving-window retention and samples `total-sst-files-size`, `estimate-live-data-size`, and + memtable bytes through the existing RocksDB property endpoint. Prometheus OB-2/3/6 series are + intentionally deferred to a separate change. - **CT-9 (fuzz)** — `SimFaults` is the injection point. ## Known open questions diff --git a/crates/hotblocks-harness/src/driver.rs b/crates/hotblocks-harness/src/driver.rs index a21f9dfc..23796107 100644 --- a/crates/hotblocks-harness/src/driver.rs +++ b/crates/hotblocks-harness/src/driver.rs @@ -79,6 +79,7 @@ pub struct StatusData { pub finalized_head: Option } +#[derive(Clone)] pub struct Client { http: reqwest::Client, base: String, @@ -178,6 +179,19 @@ impl Client { Ok(Metrics(out)) } + /// Reads the service's existing intrinsic RocksDB diagnostic property surface. + pub async fn rocksdb_property(&self, column_family: &str, name: &str) -> Result> { + let res = self + .http + .get(format!("{}/rocksdb/prop/{column_family}/{name}", self.base)) + .send() + .await?; + if res.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); + } + Ok(Some(res.error_for_status()?.text().await?.trim().parse()?)) + } + pub async fn query(&self, body: &Value) -> Result { self.query_at("stream", body).await } diff --git a/crates/hotblocks-harness/src/lib.rs b/crates/hotblocks-harness/src/lib.rs index a93dd4c2..c308e9e5 100644 --- a/crates/hotblocks-harness/src/lib.rs +++ b/crates/hotblocks-harness/src/lib.rs @@ -44,6 +44,7 @@ pub mod driver; pub mod harness; pub mod model; pub mod sim; +pub mod soak; pub mod sut; pub mod types; @@ -56,5 +57,6 @@ pub use driver::{Client, Emitted, FollowStep, Follower, Outcome}; pub use harness::{Harness, HarnessConfig}; pub use model::{Finalize, ForkResolution, Model, Predicted}; pub use sim::{Numbering, SourceSim}; +pub use soak::{ChurnSoakConfig, ChurnSoakReport, SpaceSample, StallProbe, StallReport}; pub use sut::{DatasetSpec, Retention, Sut, SutConfig}; pub use types::{Anchor, Block, BlockNumber, BlockRef}; diff --git a/crates/hotblocks-harness/src/soak.rs b/crates/hotblocks-harness/src/soak.rs new file mode 100644 index 00000000..aed75492 --- /dev/null +++ b/crates/hotblocks-harness/src/soak.rs @@ -0,0 +1,356 @@ +//! CT-7 liveness and endurance probes derived from black-box watermarks and the service's +//! existing RocksDB diagnostic-property binding. + +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, AtomicU64, Ordering} + }, + time::{Duration, Instant} +}; + +use anyhow::{Result, bail, ensure}; +use serde_json::json; + +use crate::{ + chain::Chain, + driver::{Client, Outcome}, + harness::Harness, + types::BlockNumber +}; + +#[derive(Clone, Debug)] +pub struct StallProbe { + pub poll: Duration, + pub timeout: Duration, + pub stall_budget: Duration +} + +impl Default for StallProbe { + fn default() -> Self { + Self { + poll: Duration::from_millis(100), + timeout: Duration::from_secs(30), + stall_budget: Duration::from_secs(5) + } + } +} + +#[derive(Clone, Debug, Default)] +pub struct StallReport { + pub longest_zero_commit: Duration, + pub intervals_over_one_second: u64, + pub crossed_budget: bool, + pub saw_storage_diagnostic_surface: bool +} + +impl StallProbe { + /// Observes a source-ahead interval until `target` becomes query-visible (SLI-1/SLI-9). + pub async fn observe_ingest(&self, client: &Client, target: BlockNumber) -> Result { + let deadline = Instant::now() + self.timeout; + let mut report = StallReport::default(); + let mut last_head = None; + let mut zero_since = Instant::now(); + let mut counted_interval = false; + loop { + let status = client.status().await?; + let head = status.data.as_ref().map(|data| data.last_block); + if head != last_head { + let interval = zero_since.elapsed(); + report.longest_zero_commit = report.longest_zero_commit.max(interval); + last_head = head; + zero_since = Instant::now(); + counted_interval = false; + } else { + let interval = zero_since.elapsed(); + report.longest_zero_commit = report.longest_zero_commit.max(interval); + if interval >= Duration::from_secs(1) && !counted_interval { + report.intervals_over_one_second = report.intervals_over_one_second.saturating_add(1); + counted_interval = true; + } + } + + if !report.saw_storage_diagnostic_surface { + report.saw_storage_diagnostic_surface = client + .rocksdb_property("TABLES", "rocksdb.is-write-stopped") + .await? + .is_some() + && client + .rocksdb_property("TABLES", "rocksdb.estimate-pending-compaction-bytes") + .await? + .is_some(); + } + + if head.is_some_and(|head| head >= target) { + break; + } + if Instant::now() >= deadline { + bail!("SLI-9: ingest did not reach block {target} within {:?}", self.timeout); + } + tokio::time::sleep(self.poll).await; + } + + report.crossed_budget = report.longest_zero_commit >= self.stall_budget; + Ok(report) + } +} + +#[derive(Clone, Debug)] +pub struct ChurnSoakConfig { + pub retention_blocks: u64, + pub initial_blocks: u32, + pub rounds: u32, + pub blocks_per_round: u32, + pub round_pause: Duration, + pub reclaim_settle: Duration, + /// Concurrent clients that continuously scan the moving retention window. + pub query_concurrency: usize, + pub stall_probe: StallProbe +} + +impl Default for ChurnSoakConfig { + fn default() -> Self { + Self { + retention_blocks: 100, + initial_blocks: 150, + rounds: 30, + blocks_per_round: 20, + round_pause: Duration::from_millis(200), + reclaim_settle: Duration::from_secs(22), + query_concurrency: 2, + stall_probe: StallProbe::default() + } + } +} + +#[derive(Clone, Debug, Default)] +pub struct SpaceSample { + pub disk_bytes: u64, + pub live_bytes: u64, + pub debt_bytes: u64 +} + +impl SpaceSample { + pub fn amplification(&self) -> Option { + (self.live_bytes > 0).then(|| self.disk_bytes as f64 / self.live_bytes as f64) + } +} + +#[derive(Clone, Debug, Default)] +pub struct ChurnSoakReport { + pub samples: Vec, + pub longest_zero_commit: Duration, + pub max_window_excess: u64, + pub pressure_queries_completed: u64, + pub pressure_query_failures: u64, + pub first_pressure_query_failure: Option +} + +impl Harness { + /// Runs S4 churn against an API-controlled dataset and samples RocksDB space properties. + pub async fn run_churn_soak(&mut self, cfg: &ChurnSoakConfig) -> Result { + let anchor = self.sim.anchor_hash(&self.dataset); + ensure!( + self.client + .set_retention(&json!({"FromBlock": {"number": self.start_block, "parent_hash": anchor}})) + .await? + == 200, + "CT-7 requires an API-controlled dataset" + ); + + self.produce(cfg.initial_blocks)?; + let initial_target = self.model.head().expect("initial blocks were produced").number; + cfg.stall_probe.observe_ingest(&self.client, initial_target).await?; + ensure!( + self.client + .set_retention(&json!({"Head": cfg.retention_blocks})) + .await? + == 200, + "failed to enable moving-window retention" + ); + + let query_window = Arc::new(QueryWindow { + from: AtomicU64::new(self.start_block), + to: AtomicU64::new(initial_target) + }); + let query_pressure = QueryPressure::start( + self.client.clone(), + Arc::clone(&self.chain), + Arc::clone(&query_window), + cfg.query_concurrency + ); + + let mut report = ChurnSoakReport::default(); + for _ in 0..cfg.rounds { + self.produce(cfg.blocks_per_round)?; + let head = self.model.head().expect("churn produced a head").number; + let desired_floor = head.saturating_add(1).saturating_sub(cfg.retention_blocks); + self.model.retain(desired_floor, None); + query_window.from.store(desired_floor, Ordering::Relaxed); + query_window.to.store(head, Ordering::Relaxed); + + let stall = cfg.stall_probe.observe_ingest(&self.client, head).await?; + report.longest_zero_commit = report.longest_zero_commit.max(stall.longest_zero_commit); + let status = self.client.status().await?; + let first = status + .data + .as_ref() + .expect("ingested dataset has status data") + .first_block; + ensure!( + first <= desired_floor, + "RS-3: retention dropped required data (first {first}, required floor {desired_floor})" + ); + report.max_window_excess = report.max_window_excess.max(desired_floor.saturating_sub(first)); + report.samples.push(sample_space(&self.client).await?); + tokio::time::sleep(cfg.round_pause).await; + } + + let before_settle = sample_space(&self.client).await?; + tokio::time::sleep(cfg.reclaim_settle).await; + let after_settle = sample_space(&self.client).await?; + ensure!( + after_settle.debt_bytes <= before_settle.debt_bytes, + "LIV-7: reclaim debt grew while churn was quiescent ({} -> {} bytes)", + before_settle.debt_bytes, + after_settle.debt_bytes + ); + report.samples.extend([before_settle, after_settle]); + ( + report.pressure_queries_completed, + report.pressure_query_failures, + report.first_pressure_query_failure + ) = query_pressure.finish().await; + Ok(report) + } +} + +struct QueryWindow { + from: AtomicU64, + to: AtomicU64 +} + +struct QueryPressure { + stop: Arc, + tasks: Vec)>> +} + +impl QueryPressure { + fn start(client: Client, chain: Arc, window: Arc, concurrency: usize) -> Self { + let stop = Arc::new(AtomicBool::new(false)); + let mut tasks = Vec::with_capacity(concurrency); + for _ in 0..concurrency { + let client = client.clone(); + let chain = Arc::clone(&chain); + let window = Arc::clone(&window); + let stop = Arc::clone(&stop); + tasks.push(tokio::spawn(async move { + let mut completed = 0u64; + let mut failed = 0u64; + let mut first_failure = None; + while !stop.load(Ordering::Relaxed) { + let from = window.from.load(Ordering::Relaxed); + let to = window.to.load(Ordering::Relaxed).max(from); + let query = chain.scan_query(from, Some(to), None); + if let Some(message) = pressure_query_failure(client.query(&query).await) { + failed = failed.saturating_add(1); + first_failure.get_or_insert(message); + } else { + completed = completed.saturating_add(1); + } + } + (completed, failed, first_failure) + })); + } + Self { stop, tasks } + } + + async fn finish(mut self) -> (u64, u64, Option) { + self.stop.store(true, Ordering::Relaxed); + let mut completed = 0u64; + let mut failed = 0u64; + let mut first_failure = None; + for task in self.tasks.drain(..) { + match task.await { + Ok((task_completed, task_failed, task_first_failure)) => { + completed = completed.saturating_add(task_completed); + failed = failed.saturating_add(task_failed); + if first_failure.is_none() { + first_failure = task_first_failure; + } + } + Err(err) => { + failed = failed.saturating_add(1); + first_failure.get_or_insert_with(|| format!("query pressure task failed: {err}")); + } + } + } + (completed, failed, first_failure) + } +} + +fn pressure_query_failure(result: Result) -> Option { + match result { + Ok(Outcome::Ok { .. } | Outcome::NoData { .. }) => None, + Ok(Outcome::Error { status, body }) => Some(format!("HTTP {status}: {body}")), + Ok(Outcome::Conflict { hints }) => Some(format!("HTTP 409 conflict: {hints:?}")), + Err(err) => Some(format!("query transport error: {err:#}")) + } +} + +impl Drop for QueryPressure { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + for task in &self.tasks { + task.abort(); + } + } +} + +async fn sample_space(client: &Client) -> Result { + let sst_bytes = client + .rocksdb_property("TABLES", "rocksdb.total-sst-files-size") + .await? + .unwrap_or(0); + let memtable_bytes = client + .rocksdb_property("TABLES", "rocksdb.cur-size-all-mem-tables") + .await? + .unwrap_or(0); + let live_sst_bytes = client + .rocksdb_property("TABLES", "rocksdb.estimate-live-data-size") + .await? + .unwrap_or(sst_bytes) + .min(sst_bytes); + let disk_bytes = sst_bytes.saturating_add(memtable_bytes); + let live_bytes = live_sst_bytes.saturating_add(memtable_bytes); + Ok(SpaceSample { + disk_bytes, + live_bytes, + debt_bytes: disk_bytes.saturating_sub(live_bytes) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::driver::Watermarks; + + #[test] + fn query_pressure_accepts_only_success_and_no_data() { + assert!( + pressure_query_failure(Ok(Outcome::NoData { + watermarks: Watermarks::default() + })) + .is_none() + ); + assert_eq!( + pressure_query_failure(Ok(Outcome::Error { + status: 500, + body: "broken reader".into() + })) + .as_deref(), + Some("HTTP 500: broken reader") + ); + assert!(pressure_query_failure(Ok(Outcome::Conflict { hints: Vec::new() })).is_some()); + } +} diff --git a/crates/hotblocks/Cargo.toml b/crates/hotblocks/Cargo.toml index 302aab88..5f3b010e 100644 --- a/crates/hotblocks/Cargo.toml +++ b/crates/hotblocks/Cargo.toml @@ -36,7 +36,9 @@ url = { workspace = true, features = ["serde"] } [dev-dependencies] anyhow = { workspace = true } +arrow = { workspace = true } sqd-hotblocks-harness = { path = "../hotblocks-harness" } +tempfile = { workspace = true } tokio = { workspace = true, features = ["full"] } [lints] diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index 1db864ae..334e2b1d 100644 --- a/crates/hotblocks/src/cli.rs +++ b/crates/hotblocks/src/cli.rs @@ -73,10 +73,10 @@ pub struct CLI { #[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. + /// Unlink dead SST files below the live-table watermark at startup. Off by default -- + /// the unlink ignores snapshots, so it is only safe before any query exists. Orphaned + /// dirty-table markers are purged at every startup regardless of this flag. Use the + /// `reclaim-measure` binary (shipped in this image) to size the unlink 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 @@ -84,6 +84,12 @@ pub struct CLI { #[arg(long)] pub startup_disk_reclaim: bool, + /// Periodically unlink dead table SST files that are older than every live RocksDB + /// snapshot. Queries continue normally; old queries pin only the tables they can still + /// observe. Set to 0 to disable runtime whole-file reclaim. + #[arg(long, value_name = "SECS", default_value = "300")] + pub disk_reclaim_interval_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")] @@ -109,7 +115,8 @@ impl CLI { .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); + .with_periodic_compaction_secs(self.rocksdb_periodic_compaction_secs) + .with_runtime_reclaim(self.disk_reclaim_interval_secs > 0); if let Some(jobs) = self.rocksdb_max_background_jobs { settings = settings.with_max_background_jobs(jobs); diff --git a/crates/hotblocks/src/data_service.rs b/crates/hotblocks/src/data_service.rs index 5cf33fdc..474ec1af 100644 --- a/crates/hotblocks/src/data_service.rs +++ b/crates/hotblocks/src/data_service.rs @@ -95,13 +95,14 @@ impl DataService { /// 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. +/// `disk_reclaim` gates only whole-file unlink (`--startup-disk-reclaim`, off by default). +/// Orphan cleanup and deleting unconfigured datasets always run. Ordering matters on a +/// near-full disk: +/// 1. an optional unlink pass first -- it needs no scratch space, so it can free room for +/// the bookkeeping writes below; +/// 2. bookkeeping writes that lift the watermark: purge orphan dirty markers and delete +/// unconfigured datasets; +/// 3. an optional 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. @@ -113,19 +114,16 @@ fn startup_disk_recovery(db: &DBRef, unconfigured: &[DatasetId], disk_reclaim: b 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"); } + 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") + } + for dataset_id in unconfigured { info!("deleting unconfigured dataset {dataset_id}"); if let Err(err) = db.delete_dataset(*dataset_id) { @@ -141,3 +139,26 @@ fn startup_disk_recovery(db: &DBRef, unconfigured: &[DatasetId], disk_reclaim: b debug!("startup disk recovery complete"); } + +#[cfg(test)] +mod tests { + use arrow::datatypes::{DataType, Field, Schema}; + use sqd_storage::db::DatabaseSettings; + + use super::*; + + #[test] + fn startup_always_purges_orphans_when_file_unlink_is_disabled() { + let dir = tempfile::tempdir().unwrap(); + { + let db = DatabaseSettings::default().open(dir.path()).unwrap(); + let schema = Arc::new(Schema::new(vec![Field::new("data", DataType::UInt32, true)])); + db.new_table_builder(schema).finish().unwrap(); + } + let db = Arc::new(DatabaseSettings::default().open(dir.path()).unwrap()); + + startup_disk_recovery(&db, &[], false); + + assert_eq!(db.purge_orphan_dirty_tables().unwrap(), 0); + } +} diff --git a/crates/hotblocks/src/main.rs b/crates/hotblocks/src/main.rs index 9fb6d64f..5f039a06 100644 --- a/crates/hotblocks/src/main.rs +++ b/crates/hotblocks/src/main.rs @@ -14,7 +14,7 @@ use std::time::Duration; use api::build_api; use clap::Parser; use cli::CLI; -use tracing::{debug, error, instrument}; +use tracing::{debug, error, instrument, warn}; use types::DBRef; #[global_allocator] @@ -37,11 +37,18 @@ 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. + // NB: startup orphan purge, plus optional file unlink gated by + // --startup-disk-reclaim, already ran inside `build_app` -> + // `DataService::start` before any controller spawned. + let runtime_reclaim_enabled = args.disk_reclaim_interval_secs > 0; tokio::spawn(db_cleanup_task(app.db.clone())); + if runtime_reclaim_enabled { + tokio::spawn(db_reclaim_task( + app.db.clone(), + Duration::from_secs(args.disk_reclaim_interval_secs) + )); + } let api = build_api(app); @@ -99,12 +106,8 @@ const CLEANUP_INTERVAL: Duration = Duration::from_secs(10); /// 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. +/// Routine Phase-1 cleanup: point-delete deleted tables' data. The database captures at +/// open time whether sequence metadata is needed by snapshot-aware physical reclaim. #[instrument(name = "db_cleanup", skip_all)] async fn db_cleanup_task(db: DBRef) { tokio::time::sleep(CLEANUP_INTERVAL).await; @@ -137,3 +140,72 @@ async fn db_cleanup_task(db: DBRef) { .await; } } + +/// Whole-file reclaim whose watermark includes every deleted table still visible to the +/// oldest RocksDB snapshot. It never waits for the global snapshot count to reach zero. +#[instrument(name = "db_reclaim", skip_all)] +async fn db_reclaim_task(db: DBRef, interval: Duration) { + tokio::time::sleep(interval).await; + loop { + let db = db.clone(); + let result = tokio::task::spawn_blocking(move || db.reclaim_disk_space_runtime()).await; + + let failed = match result { + Ok(Ok(report)) => { + // TODO: Export this report only when there is a concrete dashboard/alert + // contract; structured debug fields are sufficient for the current rollout. + debug!( + snapshots = report.snapshot_count, + oldest_snapshot_sequence = report.oldest_snapshot_sequence, + safe_deleted_tables = report.safe_deleted_tables, + unsafe_deleted_tables = report.unsafe_deleted_tables, + watermark =? report.watermark, + "snapshot-aware disk reclaim completed" + ); + if report.skipped_malformed_chunks > 0 { + warn!( + malformed_chunks = report.skipped_malformed_chunks, + "runtime reclaim ignored unreadable committed chunks" + ); + } + false + } + Ok(Err(err)) => { + error!(error =? err, "snapshot-aware disk reclaim failed"); + true + } + Err(err) => { + error!(error =? err, "snapshot-aware disk reclaim task panicked"); + true + } + }; + + tokio::time::sleep(db_reclaim_delay(interval, failed)).await; + } +} + +fn db_reclaim_delay(interval: Duration, failed: bool) -> Duration { + if failed { + interval.max(CLEANUP_ERROR_BACKOFF) + } else { + interval + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reclaim_errors_never_retry_faster_than_the_configured_interval() { + assert_eq!( + db_reclaim_delay(Duration::from_secs(300), true), + Duration::from_secs(300) + ); + assert_eq!(db_reclaim_delay(Duration::from_secs(2), true), CLEANUP_ERROR_BACKOFF); + assert_eq!( + db_reclaim_delay(Duration::from_secs(300), false), + Duration::from_secs(300) + ); + } +} diff --git a/crates/hotblocks/tests/ct7_stall_and_churn.rs b/crates/hotblocks/tests/ct7_stall_and_churn.rs new file mode 100644 index 00000000..94156b5b --- /dev/null +++ b/crates/hotblocks/tests/ct7_stall_and_churn.rs @@ -0,0 +1,45 @@ +//! CT-7 — moving-window churn (GAP-1/GAP-6). + +use std::{sync::Arc, time::Duration}; + +use anyhow::{Result, ensure}; +use sqd_hotblocks_harness::{ + ChurnSoakConfig, + chain::HlFills, + harness::{Harness, HarnessConfig}, + sut::Retention +}; + +const START: u64 = 1_000; + +/// The real S4 runner is intentionally ignored in the ordinary unit-test lane: its default +/// includes two cleanup periods so physical debt has a chance to converge. Run explicitly with +/// `cargo test -p sqd-hotblocks --test ct7_stall_and_churn ct7_churn_soak -- --ignored`. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "endurance scenario; run explicitly"] +async fn ct7_churn_soak() -> Result<()> { + let mut harness_cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(HlFills), START); + harness_cfg.retention = Retention::Api; + harness_cfg + .sut_args + .extend(["--disk-reclaim-interval-secs".into(), "2".into()]); + harness_cfg.rust_log = "sqd_hotblocks=debug".into(); + let mut h = Harness::start(harness_cfg).await?; + + let report = h.run_churn_soak(&ChurnSoakConfig::default()).await?; + ensure!(!report.samples.is_empty()); + ensure!(report.samples.iter().any(|sample| sample.live_bytes > 0)); + ensure!(report.longest_zero_commit < Duration::from_secs(5)); + ensure!(report.pressure_queries_completed > 0); + ensure!( + report.pressure_query_failures == 0, + "query pressure observed {} failure(s); first: {}", + report.pressure_query_failures, + report.first_pressure_query_failure.as_deref().unwrap_or("unknown") + ); + ensure!( + h.sut.log_tail(200).contains("snapshot-aware disk reclaim completed"), + "periodic runtime reclaim did not run under query pressure" + ); + Ok(()) +} diff --git a/crates/reclaim-measure/src/main.rs b/crates/reclaim-measure/src/main.rs index 480b271f..415f56c0 100644 --- a/crates/reclaim-measure/src/main.rs +++ b/crates/reclaim-measure/src/main.rs @@ -13,7 +13,7 @@ use std::collections::BTreeSet; use anyhow::{Context, Result}; use rocksdb::{Options, DB}; use sqd_storage::db::{ - reclaim::{reclaim_upper_bound, sst_is_unlinkable, watermark}, + reclaim::{deleted_table_record_kind, reclaim_upper_bound, sst_is_unlinkable, watermark, DeletedTableRecordKind}, Chunk, TableId, CF_CHUNKS, CF_DATASETS, CF_DELETED_TABLES, CF_DIRTY_TABLES, CF_TABLES }; @@ -96,13 +96,19 @@ fn main() -> Result<()> { it.status()?; } - // Pending logical deletes awaiting Phase-1 cleanup. - let mut pending_deletes = 0u64; + // Deletion lifecycle records: only DeleteRequested still awaits point tombstones. + let (mut pending_deletes, mut unstamped_purges, mut retained_purges, mut malformed_deletes) = + (0u64, 0u64, 0u64, 0u64); { let mut it = db.raw_iterator_cf(&cf_deleted); it.seek_to_first(); while it.valid() { - pending_deletes += 1; + match deleted_table_record_kind(it.value().context("missing deleted-table value")?) { + DeletedTableRecordKind::DeleteRequested => pending_deletes += 1, + DeletedTableRecordKind::PurgedUnstamped => unstamped_purges += 1, + DeletedTableRecordKind::Purged => retained_purges += 1, + DeletedTableRecordKind::Malformed => malformed_deletes += 1 + } it.next(); } it.status()?; @@ -168,6 +174,9 @@ fn main() -> Result<()> { println!("live tables : {}", live.len()); println!("dirty markers (orphans) : {}", dirty.len()); println!("pending deletes : {pending_deletes}"); + println!("unstamped purges : {unstamped_purges}"); + println!("retained purge records : {retained_purges}"); + println!("malformed delete records: {malformed_deletes}"); println!("TABLES sst files : {n_files}"); match wm_now { Some(w) => println!("watermark (live+dirty) : {w}"), diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index f2f6f744..dff8d610 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -14,6 +14,7 @@ rocksdb = { version = "0.24.0", features = ["jemalloc"] } uuid = { workspace = true, features = ["v7", "borsh"] } sqd-array = { path = "../array" } sqd-primitives = { path = "../primitives", features = ["borsh", "sid", "range"] } +tracing = { workspace = true } [dev-dependencies] proptest = { workspace = true } diff --git a/crates/storage/src/db/db.rs b/crates/storage/src/db/db.rs index 07736365..b57595b1 100644 --- a/crates/storage/src/db/db.rs +++ b/crates/storage/src/db/db.rs @@ -1,4 +1,4 @@ -use std::path::Path; +use std::{path::Path, sync::Arc}; use anyhow::ensure; use arrow::datatypes::SchemaRef; @@ -12,7 +12,8 @@ use super::{ use crate::db::{ ops::{perform_dataset_compaction, CompactionStatus}, read::datasets::list_all_datasets, - write::{ops as cleanup_ops, table_builder::TableBuilder, tx::Tx}, + reclaim::RuntimeReclaimResult, + write::{inflight::InflightTableRegistry, ops as cleanup_ops, table_builder::TableBuilder, tx::Tx}, Chunk, DatasetUpdate }; @@ -45,7 +46,8 @@ pub struct DatabaseSettings { keep_log_file_num: usize, auto_compactions: bool, max_background_jobs: usize, - periodic_compaction_secs: u64 + periodic_compaction_secs: u64, + runtime_reclaim_enabled: bool } /// RocksDB's default of 2 could not keep up with ingest during the NET-819 incident, but a @@ -69,7 +71,8 @@ impl Default for DatabaseSettings { keep_log_file_num: 10, auto_compactions: true, max_background_jobs: default_max_background_jobs(), - periodic_compaction_secs: DEFAULT_PERIODIC_COMPACTION_SECS + periodic_compaction_secs: DEFAULT_PERIODIC_COMPACTION_SECS, + runtime_reclaim_enabled: false } } } @@ -134,6 +137,13 @@ impl DatabaseSettings { self } + /// Retain snapshot-horizon deletion records for periodic runtime whole-file reclaim. + /// Defaults to `false`, preserving the standalone storage API's bounded-cleanup contract. + pub fn with_runtime_reclaim(mut self, enabled: bool) -> Self { + self.runtime_reclaim_enabled = enabled; + self + } + fn db_options(&self) -> RocksOptions { let mut options = RocksOptions::default(); options.create_if_missing(true); @@ -143,8 +153,8 @@ impl DatabaseSettings { options.set_max_log_file_size(self.max_log_file_size * 1024 * 1024); options.set_keep_log_file_num(self.keep_log_file_num); // Keep compaction ahead of ingest + deletion tombstones: the default 2 jobs could - // not keep up during the NET-819 incident, and routine reclaim now relies entirely - // on compaction (the file unlink is startup-only). + // not keep up during the NET-819 incident. Runtime whole-file reclaim is a safety + // valve, not a replacement for normal background compaction. options.set_max_background_jobs(self.max_background_jobs as i32); options.set_max_subcompactions((self.max_background_jobs as u32 / 2).max(1)); if self.with_rocksdb_stats { @@ -228,13 +238,20 @@ impl DatabaseSettings { ] )?; - Ok(Database { db, options }) + Ok(Database { + db, + options, + inflight_tables: Arc::new(InflightTableRegistry::default()), + runtime_reclaim_enabled: self.runtime_reclaim_enabled + }) } } pub struct Database { db: RocksDB, - options: RocksOptions + options: RocksOptions, + inflight_tables: Arc, + runtime_reclaim_enabled: bool } impl Database { @@ -278,7 +295,7 @@ impl Database { } pub fn new_table_builder(&self, schema: SchemaRef) -> TableBuilder<'_> { - TableBuilder::new(&self.db, schema) + TableBuilder::new(&self.db, schema, &self.inflight_tables) } pub fn insert_chunk(&self, dataset_id: DatasetId, chunk: &Chunk) -> anyhow::Result<()> { @@ -319,6 +336,7 @@ impl Database { ) -> anyhow::Result { perform_dataset_compaction( &self.db, + &self.inflight_tables, dataset_id, max_chunk_size, write_amplification_limit, @@ -349,9 +367,11 @@ impl Database { } /// Phase 1 -- logically purge deleted tables (snapshot-safe point deletes). + /// Snapshot-horizon metadata is retained only when configured through + /// [`DatabaseSettings::with_runtime_reclaim`]. /// Returns the number of tables logically deleted by this call. pub fn cleanup(&self) -> anyhow::Result { - cleanup_ops::logical_cleanup(&self.db) + cleanup_ops::logical_cleanup(&self.db, self.runtime_reclaim_enabled) } /// Physically reclaim disk by unlinking whole SST files below the live watermark @@ -363,6 +383,18 @@ impl Database { cleanup_ops::reclaim_disk_space(&self.db) } + /// Physically reclaim whole SST files while queries are running. Unlike the startup + /// variant, this pins tables whose point tombstones are newer than RocksDB's oldest + /// live snapshot; continuous newer queries do not hold old tables forever. + pub fn reclaim_disk_space_runtime(&self) -> anyhow::Result { + ensure!( + self.runtime_reclaim_enabled, + "runtime reclaim was not enabled in DatabaseSettings" + ); + let build_fence = self.inflight_tables.reclaim_fence(); + cleanup_ops::reclaim_disk_space_runtime(&self.db, build_fence) + } + /// Crash recovery: purge `DIRTY_TABLES` markers left by builds that died before /// commit -- an orphan pins the reclaim watermark forever. MUST run before any ingest /// starts: it treats every dirty marker as an orphan. Returns orphans purged. @@ -403,3 +435,86 @@ impl std::fmt::Debug for Database { f.debug_struct("Database").field("path", &self.db.path()).finish() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::{deleted_table::DeletedTableState, table_id::TableId}; + + #[test] + fn malformed_deleted_value_does_not_block_valid_cleanup() { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseSettings::default() + .with_runtime_reclaim(true) + .open(dir.path()) + .unwrap(); + let cf_deleted = db.db.cf_handle(CF_DELETED_TABLES).unwrap(); + let malformed = TableId::new(); + let pending = TableId::new(); + db.db.put_cf(cf_deleted, malformed, [99]).unwrap(); + db.db.put_cf(cf_deleted, pending, []).unwrap(); + let old_snapshot = db.db.snapshot(); + + assert_eq!(db.cleanup().unwrap(), 2); + for id in [malformed, pending] { + let value = db.db.get_cf(cf_deleted, id).unwrap().unwrap(); + assert!(matches!( + DeletedTableState::decode(&value).unwrap(), + DeletedTableState::Purged { .. } + )); + } + let report = db.reclaim_disk_space_runtime().unwrap(); + assert_eq!(report.unsafe_deleted_tables, 2); + drop(old_snapshot); + let report = db.reclaim_disk_space_runtime().unwrap(); + assert_eq!(report.safe_deleted_tables, 2); + } + + #[test] + fn partial_tombstone_failure_keeps_delete_requested_for_retry() { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseSettings::default() + .with_runtime_reclaim(true) + .open(dir.path()) + .unwrap(); + let id = TableId::new(); + let cf_deleted = db.db.cf_handle(CF_DELETED_TABLES).unwrap(); + let cf_tables = db.db.cf_handle(CF_TABLES).unwrap(); + let mut table_key = id.as_ref().to_vec(); + table_key.push(1); + db.db.put_cf(cf_tables, &table_key, [42]).unwrap(); + db.db.put_cf(cf_deleted, id, []).unwrap(); + + let mut fail_after_write = |batch| { + db.db.write(batch)?; + anyhow::bail!("injected failure after partial tombstone write") + }; + let err = + cleanup_ops::purge_deleted_table_with_writer(&db.db, &id, true, 1, &mut fail_after_write).unwrap_err(); + assert!(err.to_string().contains("injected failure")); + assert!(db.db.get_cf(cf_tables, &table_key).unwrap().is_none()); + assert_eq!(db.db.get_cf(cf_deleted, id).unwrap().unwrap(), Vec::::new()); + + assert_eq!(db.cleanup().unwrap(), 1); + let value = db.db.get_cf(cf_deleted, id).unwrap().unwrap(); + assert!(matches!( + DeletedTableState::decode(&value).unwrap(), + DeletedTableState::Purged { .. } + )); + } + + #[test] + fn runtime_reclaim_skips_malformed_chunks_but_startup_reclaim_fails_closed() { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseSettings::default() + .with_runtime_reclaim(true) + .open(dir.path()) + .unwrap(); + let cf_chunks = db.db.cf_handle(CF_CHUNKS).unwrap(); + db.db.put_cf(cf_chunks, [0u8; 56], [99]).unwrap(); + + let report = db.reclaim_disk_space_runtime().unwrap(); + assert_eq!(report.skipped_malformed_chunks, 1); + assert!(db.reclaim_disk_space().is_err()); + } +} diff --git a/crates/storage/src/db/deleted_table.rs b/crates/storage/src/db/deleted_table.rs new file mode 100644 index 00000000..63d694e5 --- /dev/null +++ b/crates/storage/src/db/deleted_table.rs @@ -0,0 +1,96 @@ +use anyhow::{bail, ensure}; + +/// Durable lifecycle of a table after its last chunk reference is deleted. +/// +/// Empty values predate this state machine and remain `DeleteRequested`, so databases +/// written by older releases upgrade without a migration. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum DeletedTableState { + DeleteRequested, + PurgedUnstamped, + Purged { sequence: u64 } +} + +impl DeletedTableState { + const PURGED_UNSTAMPED: u8 = 1; + const PURGED: u8 = 2; + + pub(crate) fn decode(bytes: &[u8]) -> anyhow::Result { + match bytes { + [] => Ok(Self::DeleteRequested), + [Self::PURGED_UNSTAMPED] => Ok(Self::PurgedUnstamped), + [Self::PURGED, sequence @ ..] => { + ensure!( + sequence.len() == size_of::(), + "invalid purged-table sequence length" + ); + let sequence = u64::from_le_bytes(sequence.try_into().unwrap()); + Ok(Self::Purged { sequence }) + } + [tag, ..] => bail!("unknown deleted-table state tag {tag}") + } + } + + pub(crate) fn encode(self) -> Vec { + match self { + Self::DeleteRequested => Vec::new(), + Self::PurgedUnstamped => vec![Self::PURGED_UNSTAMPED], + Self::Purged { sequence } => { + let mut value = Vec::with_capacity(1 + size_of::()); + value.push(Self::PURGED); + value.extend_from_slice(&sequence.to_le_bytes()); + value + } + } + } + + pub(crate) fn is_reclaim_safe(self, oldest_snapshot_sequence: Option) -> bool { + match (self, oldest_snapshot_sequence) { + (Self::Purged { .. }, None) => true, + (Self::Purged { sequence }, Some(oldest)) => sequence <= oldest, + (Self::DeleteRequested | Self::PurgedUnstamped, _) => false + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_legacy_value_is_delete_requested() { + assert_eq!( + DeletedTableState::decode(&[]).unwrap(), + DeletedTableState::DeleteRequested + ); + } + + #[test] + fn states_round_trip() { + for state in [ + DeletedTableState::DeleteRequested, + DeletedTableState::PurgedUnstamped, + DeletedTableState::Purged { sequence: 42 } + ] { + assert_eq!(DeletedTableState::decode(&state.encode()).unwrap(), state); + } + } + + #[test] + fn malformed_state_fails_closed() { + assert!(DeletedTableState::decode(&[DeletedTableState::PURGED, 1]).is_err()); + assert!(DeletedTableState::decode(&[99]).is_err()); + } + + #[test] + fn only_sequence_stamped_purges_can_be_reclaimed() { + assert!(!DeletedTableState::DeleteRequested.is_reclaim_safe(None)); + assert!(!DeletedTableState::PurgedUnstamped.is_reclaim_safe(None)); + + let purged = DeletedTableState::Purged { sequence: 42 }; + assert!(purged.is_reclaim_safe(None)); + assert!(!purged.is_reclaim_safe(Some(41))); + assert!(purged.is_reclaim_safe(Some(42))); + assert!(purged.is_reclaim_safe(Some(43))); + } +} diff --git a/crates/storage/src/db/mod.rs b/crates/storage/src/db/mod.rs index 90e06547..6084d31e 100644 --- a/crates/storage/src/db/mod.rs +++ b/crates/storage/src/db/mod.rs @@ -1,5 +1,6 @@ mod data; mod db; +mod deleted_table; pub mod ops; mod read; pub mod reclaim; diff --git a/crates/storage/src/db/ops/compaction.rs b/crates/storage/src/db/ops/compaction.rs index 4f081723..3e4e1567 100644 --- a/crates/storage/src/db/ops/compaction.rs +++ b/crates/storage/src/db/ops/compaction.rs @@ -1,6 +1,7 @@ use std::{ cmp::{max, min}, - collections::BTreeMap + collections::BTreeMap, + sync::Arc }; use arrow::datatypes::SchemaRef; @@ -10,7 +11,7 @@ use crate::db::{ db::RocksDB, ops::{schema_merge::can_merge_schemas, table_merge::TableMerge}, table_id::TableId, - write::tx::Tx, + write::{inflight::InflightTableRegistry, tx::Tx}, Chunk, ChunkReader, DatasetId, ReadSnapshot, TableBuilder }; @@ -32,8 +33,9 @@ pub struct MergedChunk { pub size: usize } -pub fn perform_dataset_compaction( +pub(crate) fn perform_dataset_compaction( db: &RocksDB, + inflight_tables: &Arc, dataset_id: DatasetId, max_chunk_size: Option, write_amplification_limit: Option, @@ -41,6 +43,7 @@ pub fn perform_dataset_compaction( ) -> anyhow::Result { DatasetCompaction { db, + inflight_tables, snapshot: &ReadSnapshot::new(db), dataset_id, merge: Vec::new(), @@ -53,6 +56,7 @@ pub fn perform_dataset_compaction( struct DatasetCompaction<'a> { db: &'a RocksDB, + inflight_tables: &'a Arc, snapshot: &'a ReadSnapshot<'a>, dataset_id: DatasetId, merge: Vec>, @@ -171,7 +175,7 @@ impl<'a> DatasetCompaction<'a> { .collect::>>()?; let src = TableMerge::prepare(&chunks)?; - let mut table_builder = TableBuilder::new(self.db, src.schema()); + let mut table_builder = TableBuilder::new(self.db, src.schema(), self.inflight_tables); table_builder.set_stats(src.columns_with_stats().iter().copied())?; src.write(&mut table_builder)?; let table_id = table_builder.finish()?; diff --git a/crates/storage/src/db/reclaim.rs b/crates/storage/src/db/reclaim.rs index be068b23..07a4d02a 100644 --- a/crates/storage/src/db/reclaim.rs +++ b/crates/storage/src/db/reclaim.rs @@ -3,7 +3,48 @@ //! (They cannot share a scan: one runs on a secondary `rocksdb::DB`, the other on an //! `OptimisticTransactionDB`, and `rocksdb::DBInner` is not exported.) -use crate::db::table_id::TableId; +use crate::db::{deleted_table::DeletedTableState, table_id::TableId}; + +/// Diagnostic classification of a `CF_DELETED_TABLES` value. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DeletedTableRecordKind { + /// Point tombstones have not been written yet. + DeleteRequested, + /// Point tombstones were written, but their conservative sequence was not persisted. + PurgedUnstamped, + /// Point tombstones and their conservative sequence were persisted. + Purged, + /// The value cannot be decoded as any supported deletion state. + Malformed +} + +/// Classify deletion bookkeeping without exposing its internal sequence-state encoding. +pub fn deleted_table_record_kind(value: &[u8]) -> DeletedTableRecordKind { + match DeletedTableState::decode(value) { + Ok(DeletedTableState::DeleteRequested) => DeletedTableRecordKind::DeleteRequested, + Ok(DeletedTableState::PurgedUnstamped) => DeletedTableRecordKind::PurgedUnstamped, + Ok(DeletedTableState::Purged { .. }) => DeletedTableRecordKind::Purged, + Err(_) => DeletedTableRecordKind::Malformed + } +} + +/// What a snapshot-aware runtime reclaim observed and retired. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct RuntimeReclaimResult { + /// RocksDB snapshots that existed when the safety horizon was sampled. + pub snapshot_count: u64, + /// Sequence of the oldest live snapshot, or `None` when there were no snapshots. + pub oldest_snapshot_sequence: Option, + /// Purged table records old enough for every live snapshot and removed this pass. + pub safe_deleted_tables: usize, + /// Deleted tables still awaiting point deletes/stamping or still needed by an old snapshot. + pub unsafe_deleted_tables: usize, + /// Undecodable committed chunks ignored by runtime reclaim. Their table data was already + /// inaccessible through the query API; startup reclaim remains fail-closed on them. + pub skipped_malformed_chunks: usize, + /// Smallest table id that the whole-file unlink was not allowed to cross. + pub watermark: Option +} /// Lower bound for `DeleteFilesInRange`. Keys are `table_id (16B) ++ tag ++ ..`. pub const RECLAIM_LOWER_BOUND: [u8; 16] = [0u8; 16]; @@ -92,4 +133,15 @@ mod tests { assert!(sst_is_unlinkable(3, Some(&RECLAIM_LOWER_BOUND), &bound)); } + + #[test] + fn deleted_table_records_are_classified_for_diagnostics() { + assert_eq!(deleted_table_record_kind(&[]), DeletedTableRecordKind::DeleteRequested); + assert_eq!(deleted_table_record_kind(&[1]), DeletedTableRecordKind::PurgedUnstamped); + assert_eq!( + deleted_table_record_kind(&[2, 0, 0, 0, 0, 0, 0, 0, 0]), + DeletedTableRecordKind::Purged + ); + assert_eq!(deleted_table_record_kind(&[99]), DeletedTableRecordKind::Malformed); + } } diff --git a/crates/storage/src/db/table_id.rs b/crates/storage/src/db/table_id.rs index d88fb1ae..657d9f42 100644 --- a/crates/storage/src/db/table_id.rs +++ b/crates/storage/src/db/table_id.rs @@ -15,6 +15,8 @@ impl AsRef<[u8]> for TableId { } impl TableId { + /// UUIDv7 generation is monotonic within one process. Runtime reclaim uses an unused + /// id as a fence: tables allocated after the fence sort above its unlink range. pub fn new() -> Self { Self { uuid: Uuid::now_v7() } } @@ -48,4 +50,14 @@ mod tests { assert_eq!(TableId::try_from_key(&[0u8; 15]), None); assert_eq!(TableId::try_from_key(&[0u8; 17]), None); } + + #[test] + fn new_ids_are_ordered_for_reclaim_fences() { + let mut previous = TableId::new(); + for _ in 0..1_000 { + let next = TableId::new(); + assert!(previous < next); + previous = next; + } + } } diff --git a/crates/storage/src/db/write/inflight.rs b/crates/storage/src/db/write/inflight.rs new file mode 100644 index 00000000..d891d72b --- /dev/null +++ b/crates/storage/src/db/write/inflight.rs @@ -0,0 +1,98 @@ +use std::{collections::BTreeSet, sync::Arc}; + +use parking_lot::Mutex; + +use crate::db::table_id::TableId; + +/// Immutable inputs that keep a runtime unlink below every table build concurrent with it. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct ReclaimFence { + /// Smallest build active when the fence was taken. It remains pinned even if that build + /// finishes before the later RocksDB metadata snapshot is created. + pub(crate) active_build_watermark: Option, + /// Unused process-monotonic id. Every build registered after the fence sorts above it. + pub(crate) boundary: TableId +} + +/// Tracks table ids from allocation until their data and dirty marker are durable. +/// +/// The mutex is held only for set updates and reclaim-fence creation; table construction and +/// reclaim scans never retain it. +#[derive(Default)] +pub(crate) struct InflightTableRegistry { + ids: Mutex> +} + +impl InflightTableRegistry { + pub(crate) fn register(self: &Arc) -> InflightTableGuard { + let mut ids = self.ids.lock(); + let table_id = TableId::new(); + let inserted = ids.insert(table_id); + debug_assert!(inserted); + InflightTableGuard { + registry: Arc::clone(self), + table_id + } + } + + pub(crate) fn reclaim_fence(&self) -> ReclaimFence { + let ids = self.ids.lock(); + let active_build_watermark = ids.first().copied(); + let boundary = TableId::new(); + debug_assert!(active_build_watermark.is_none_or(|id| id < boundary)); + ReclaimFence { + active_build_watermark, + boundary + } + } +} + +pub(crate) struct InflightTableGuard { + registry: Arc, + table_id: TableId +} + +impl InflightTableGuard { + pub(crate) fn table_id(&self) -> TableId { + self.table_id + } +} + +impl Drop for InflightTableGuard { + fn drop(&mut self) { + let removed = self.registry.ids.lock().remove(&self.table_id); + debug_assert!(removed); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fence_captures_active_builds_and_bounds_later_ones() { + let registry = Arc::new(InflightTableRegistry::default()); + let first_active = registry.register(); + let second_active = registry.register(); + + let fence = registry.reclaim_fence(); + assert_eq!(fence.active_build_watermark, Some(first_active.table_id())); + assert!(second_active.table_id() < fence.boundary); + + let later = registry.register(); + assert!(fence.boundary < later.table_id()); + } + + #[test] + fn fence_keeps_a_builder_that_finishes_before_the_metadata_scan() { + let registry = Arc::new(InflightTableRegistry::default()); + let active = registry.register(); + let active_id = active.table_id(); + + let fence = registry.reclaim_fence(); + drop(active); + + assert_eq!(fence.active_build_watermark, Some(active_id)); + assert!(registry.reclaim_fence().active_build_watermark.is_none()); + } +} diff --git a/crates/storage/src/db/write/mod.rs b/crates/storage/src/db/write/mod.rs index d5545b53..da04bb7e 100644 --- a/crates/storage/src/db/write/mod.rs +++ b/crates/storage/src/db/write/mod.rs @@ -1,4 +1,5 @@ pub mod dataset_update; +pub(crate) mod inflight; pub mod ops; mod storage; pub mod table_builder; diff --git a/crates/storage/src/db/write/ops.rs b/crates/storage/src/db/write/ops.rs index 2dbc6cd3..5e6e4607 100644 --- a/crates/storage/src/db/write/ops.rs +++ b/crates/storage/src/db/write/ops.rs @@ -1,28 +1,109 @@ +use anyhow::Context; +use tracing::warn; + use crate::{ db::{ - db::{RocksDB, RocksWriteBatch, CF_CHUNKS, CF_DELETED_TABLES, CF_DIRTY_TABLES, CF_TABLES}, - reclaim::{reclaim_upper_bound, watermark, RECLAIM_LOWER_BOUND}, + db::{ + RocksDB, RocksIterator, RocksSnapshot, RocksWriteBatch, CF_CHUNKS, CF_DELETED_TABLES, CF_DIRTY_TABLES, + CF_TABLES + }, + deleted_table::DeletedTableState, + reclaim::{reclaim_upper_bound, watermark, RuntimeReclaimResult, RECLAIM_LOWER_BOUND}, table_id::TableId, + write::{inflight::ReclaimFence, storage::WRITE_BATCH_SIZE_LIMIT}, Chunk }, table::key::TableKeyFactory }; +struct DeletedTableScan { + records: Vec<(TableId, DeletedTableState)>, + malformed_keys: Vec>, + recovered_values: usize +} + /// Phase 1 -- logical, snapshot-safe purge of deleted tables. /// -/// Point-deletes the `CF_TABLES` keys of every table in `CF_DELETED_TABLES`, then drops -/// its bookkeeping entry. The deletes are MVCC-versioned, so in-flight queries are -/// unaffected; the space is freed later by compaction. Idempotent. Returns tables purged. -pub(crate) fn logical_cleanup(db: &RocksDB) -> anyhow::Result { +/// Point-deletes the `CF_TABLES` keys of every newly requested table. When runtime reclaim +/// is enabled, records an upper bound on the tombstone batch's RocksDB sequence and retains +/// it until every live snapshot is new enough for whole-file reclaim. Otherwise removes the +/// record atomically with the point deletes. Idempotent. Returns tables newly point-deleted. +pub(crate) fn logical_cleanup(db: &RocksDB, retain_reclaim_metadata: bool) -> anyhow::Result { // Collect first, then mutate: writing mid-iteration disturbs the cursor. - let (pending, malformed) = scan_table_ids(db, CF_DELETED_TABLES)?; - drop_malformed_keys(db, CF_DELETED_TABLES, &malformed)?; + let scan = scan_deleted_tables(db)?; + drop_malformed_keys(db, CF_DELETED_TABLES, &scan.malformed_keys)?; + if scan.recovered_values > 0 { + warn!( + records = scan.recovered_values, + "recovering malformed deletion states as pending deletes" + ); + } - for id in &pending { - purge_table(db, id)?; + let mut purged = 0; + for (id, state) in scan.records { + match state { + DeletedTableState::DeleteRequested => { + purge_deleted_table(db, &id, retain_reclaim_metadata)?; + purged += 1; + } + DeletedTableState::PurgedUnstamped => { + if retain_reclaim_metadata { + stamp_purged_table(db, &id)? + } else { + drop_deleted_table_record(db, &id)? + } + } + DeletedTableState::Purged { .. } => { + if !retain_reclaim_metadata { + drop_deleted_table_record(db, &id)? + } + } + } + } + + Ok(purged) +} + +fn scan_deleted_tables(db: &RocksDB) -> anyhow::Result { + let cf = db.cf_handle(CF_DELETED_TABLES).unwrap(); + collect_deleted_tables(db.raw_iterator_cf(cf)) +} + +fn scan_deleted_tables_at(db: &RocksDB, snapshot: &RocksSnapshot<'_, RocksDB>) -> anyhow::Result { + let cf = db.cf_handle(CF_DELETED_TABLES).unwrap(); + collect_deleted_tables(snapshot.raw_iterator_cf(cf)) +} + +fn collect_deleted_tables(mut it: RocksIterator<'_, RocksDB>) -> anyhow::Result { + let mut records = Vec::new(); + let mut malformed_keys = Vec::new(); + let mut recovered_values = 0usize; + + it.seek_to_first(); + while it.valid() { + let key = it.key().unwrap(); + match TableId::try_from_key(key) { + Some(id) => match DeletedTableState::decode(it.value().unwrap()) { + Ok(state) => records.push((id, state)), + Err(_) => { + // A key in CF_DELETED_TABLES is no longer referenced by a committed + // chunk. Replaying its point deletes is idempotent and, unlike dropping + // the bad value, creates a fresh sequence pin for old snapshots. + records.push((id, DeletedTableState::DeleteRequested)); + recovered_values = recovered_values.saturating_add(1); + } + }, + None => malformed_keys.push(key.to_vec()) + } + it.next(); } + it.status()?; - Ok(pending.len()) + Ok(DeletedTableScan { + records, + malformed_keys, + recovered_values + }) } /// Scan `cf`, splitting keys into well-formed `TableId`s and malformed leftovers. @@ -30,10 +111,22 @@ pub(crate) fn logical_cleanup(db: &RocksDB) -> anyhow::Result { /// [`drop_malformed_keys`] -- left in place, every pass would rescan them forever. fn scan_table_ids(db: &RocksDB, cf: &str) -> anyhow::Result<(Vec, Vec>)> { let cf = db.cf_handle(cf).unwrap(); + collect_table_ids(db.raw_iterator_cf(cf)) +} + +fn scan_table_ids_at( + db: &RocksDB, + snapshot: &RocksSnapshot<'_, RocksDB>, + cf: &str +) -> anyhow::Result<(Vec, Vec>)> { + let cf = db.cf_handle(cf).unwrap(); + collect_table_ids(snapshot.raw_iterator_cf(cf)) +} + +fn collect_table_ids(mut it: RocksIterator<'_, RocksDB>) -> anyhow::Result<(Vec, Vec>)> { let mut ids = Vec::new(); let mut malformed = Vec::new(); - let mut it = db.raw_iterator_cf(cf); it.seek_to_first(); while it.valid() { let key = it.key().unwrap(); @@ -53,35 +146,113 @@ fn drop_malformed_keys(db: &RocksDB, cf: &str, keys: &[Vec]) -> anyhow::Resu if keys.is_empty() { return Ok(()); } - let cf = db.cf_handle(cf).unwrap(); + let cf_name = cf; + let cf = db.cf_handle(cf_name).unwrap(); let mut batch = RocksWriteBatch::default(); for key in keys { batch.delete_cf(cf, key); } db.write(batch)?; + warn!( + column_family = cf_name, + records = keys.len(), + "dropped malformed bookkeeping records" + ); Ok(()) } -/// Point-delete all of `id`'s `CF_TABLES` data and drop its bookkeeping entries. One write -/// batch per table, so a large purge never holds every table's deletes in memory at once. -/// -/// A table is normally in only one bookkeeping CF, but clearing both costs one tombstone -/// and guarantees a purge never leaves a stale marker pinning the watermark. +/// Point-delete all of `id`'s `CF_TABLES` data in bounded batches, retain an unstamped +/// deletion record in the final batch, then persist a conservative upper bound on the +/// tombstones' RocksDB sequence. /// /// Point deletes rather than a single `delete_range` tombstone: range deletions are not /// officially supported on the transactional `OptimisticTransactionDB`. -fn purge_table(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { - let cf_tables = db.cf_handle(CF_TABLES).unwrap(); +fn purge_deleted_table(db: &RocksDB, id: &TableId, retain_reclaim_metadata: bool) -> anyhow::Result<()> { + purge_deleted_table_impl(db, id, retain_reclaim_metadata, WRITE_BATCH_SIZE_LIMIT, &mut |batch| { + db.write(batch)?; + Ok(()) + }) +} + +fn purge_deleted_table_impl( + db: &RocksDB, + id: &TableId, + retain_reclaim_metadata: bool, + batch_size_limit: usize, + write_batch: &mut impl FnMut(RocksWriteBatch) -> anyhow::Result<()> +) -> anyhow::Result<()> { let cf_deleted = db.cf_handle(CF_DELETED_TABLES).unwrap(); let cf_dirty = db.cf_handle(CF_DIRTY_TABLES).unwrap(); + let mut batch = RocksWriteBatch::default(); + tombstone_table_data(db, id, &mut batch, batch_size_limit, write_batch)?; + if retain_reclaim_metadata { + batch.put_cf(cf_deleted, id, DeletedTableState::PurgedUnstamped.encode()); + } else { + // The point tombstones and record removal share a batch, so a crash cannot lose + // pending Phase 1 work even when runtime reclaim metadata is disabled. + batch.delete_cf(cf_deleted, id); + } + batch.delete_cf(cf_dirty, id); + write_batch(batch)?; + + if retain_reclaim_metadata { + stamp_purged_table(db, id)?; + } + + Ok(()) +} + +fn drop_deleted_table_record(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { + db.delete_cf(db.cf_handle(CF_DELETED_TABLES).unwrap(), id)?; + Ok(()) +} + +/// Recoverable second half of a purge. If the process dies between the point-delete batch +/// and this write, `PurgedUnstamped` remains and the next cleanup stamps the already-present +/// tombstones with a newer (therefore conservative) sequence. +fn stamp_purged_table(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { + let sequence = db.latest_sequence_number(); + let cf_deleted = db.cf_handle(CF_DELETED_TABLES).unwrap(); + db.put_cf(cf_deleted, id, DeletedTableState::Purged { sequence }.encode())?; + + Ok(()) +} + +/// Startup-only orphan cleanup. No query exists, so no sequence record has to survive. +fn purge_orphan_table(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { + let cf_deleted = db.cf_handle(CF_DELETED_TABLES).unwrap(); + let cf_dirty = db.cf_handle(CF_DIRTY_TABLES).unwrap(); + + let mut batch = RocksWriteBatch::default(); + tombstone_table_data(db, id, &mut batch, WRITE_BATCH_SIZE_LIMIT, &mut |batch| { + db.write(batch)?; + Ok(()) + })?; + batch.delete_cf(cf_deleted, id); + batch.delete_cf(cf_dirty, id); + db.write(batch)?; + + Ok(()) +} + +fn tombstone_table_data( + db: &RocksDB, + id: &TableId, + batch: &mut RocksWriteBatch, + batch_size_limit: usize, + write_batch: &mut impl FnMut(RocksWriteBatch) -> anyhow::Result<()> +) -> anyhow::Result<()> { + let cf_tables = db.cf_handle(CF_TABLES).unwrap(); + let mut start = TableKeyFactory::new(id); let mut end = TableKeyFactory::new(id); let start_key = start.start(); let end_key = end.end(); - let mut batch = RocksWriteBatch::default(); - let mut it = db.raw_iterator_cf(cf_tables); + // Keep iteration stable while partial tombstone batches are committed beneath it. + let snapshot = db.snapshot(); + let mut it = snapshot.raw_iterator_cf(cf_tables); it.seek(start_key); while it.valid() { let key = it.key().unwrap(); @@ -89,17 +260,31 @@ fn purge_table(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { break; } batch.delete_cf(cf_tables, key); + if let Some(full_batch) = take_full_tombstone_batch(batch, batch_size_limit) { + write_batch(full_batch)?; + } it.next(); } it.status()?; - batch.delete_cf(cf_deleted, id); - batch.delete_cf(cf_dirty, id); - db.write(batch)?; - Ok(()) } +fn take_full_tombstone_batch(batch: &mut RocksWriteBatch, batch_size_limit: usize) -> Option { + (batch.size_in_bytes() >= batch_size_limit).then(|| std::mem::take(batch)) +} + +#[cfg(test)] +pub(in crate::db) fn purge_deleted_table_with_writer( + db: &RocksDB, + id: &TableId, + retain_reclaim_metadata: bool, + batch_size_limit: usize, + write_batch: &mut impl FnMut(RocksWriteBatch) -> anyhow::Result<()> +) -> anyhow::Result<()> { + purge_deleted_table_impl(db, id, retain_reclaim_metadata, batch_size_limit, write_batch) +} + /// Physically reclaim disk by unlinking whole `CF_TABLES` SST files below the live /// watermark ([`min_live_table_id`]). Table ids are time-ordered UUIDv7s, so dead tables /// form a contiguous low range; boundary, level-0 and above-watermark garbage is left to @@ -111,9 +296,10 @@ fn purge_table(db: &RocksDB, id: &TableId) -> anyhow::Result<()> { /// /// SAFETY: the unlink IGNORES snapshots -- it can break an in-flight query reading a /// just-deleted table below the watermark. So it runs only at STARTUP, before any -/// controller/query exists. FUTURE: trigger under runtime disk pressure too, accepting that -/// read risk. That, not this, is the answer to a genuinely full disk -- getting here at all -/// requires the database to have opened, and opening replays the WAL and flushes it to L0. +/// controller/query exists. Do not use this path for runtime disk pressure unless query +/// semantics explicitly permit invalidating live readers; use snapshot-aware runtime reclaim +/// otherwise. Getting here at all requires the database to have opened, and opening replays +/// the WAL and flushes it to L0. pub(crate) fn reclaim_disk_space(db: &RocksDB) -> anyhow::Result<()> { let cf_tables = db.cf_handle(CF_TABLES).unwrap(); let hi = reclaim_upper_bound(min_live_table_id(db)?); @@ -121,6 +307,100 @@ pub(crate) fn reclaim_disk_space(db: &RocksDB) -> anyhow::Result<()> { Ok(()) } +const NUM_SNAPSHOTS_PROPERTY: &str = "rocksdb.num-snapshots"; +const OLDEST_SNAPSHOT_SEQUENCE_PROPERTY: &str = "rocksdb.oldest-snapshot-sequence"; + +/// Runtime whole-file reclaim guarded by RocksDB's MVCC sequence horizon. +/// +/// A purged table pins the watermark while any snapshot can predate its point tombstones. +/// Once the oldest live snapshot reaches the recorded purge sequence, that table is safe +/// forever: future snapshots can only be newer. Continuous query traffic therefore does +/// not require a moment with zero snapshots. A genuinely long-lived snapshot intentionally +/// pins physical deletion; storage does not expire a live query behind its caller's back. +pub(crate) fn reclaim_disk_space_runtime( + db: &RocksDB, + build_fence: ReclaimFence +) -> anyhow::Result { + // One RocksDB snapshot makes the three bookkeeping scans a consistent view. It is + // dropped before reading the safety horizon, so it does not count as a live reader. + let metadata = db.snapshot(); + // TODO: Cache the per-dataset minimum committed table id if this full chunk scan shows + // up in profiles. It is snapshot-consistent and no longer runs under a write gate. + let live_scan = min_committed_table_id_at(db, &metadata)?; + let live = live_scan.min; + let (dirty, _) = scan_table_ids_at(db, &metadata, CF_DIRTY_TABLES)?; + let deleted = scan_deleted_tables_at(db, &metadata)?; + drop(metadata); + + let cf_tables = db.cf_handle(CF_TABLES).unwrap(); + let snapshot_count = db + .property_int_value_cf(cf_tables, NUM_SNAPSHOTS_PROPERTY)? + .context("RocksDB did not expose rocksdb.num-snapshots")?; + let oldest_snapshot_sequence = if snapshot_count == 0 { + None + } else { + Some( + db.property_int_value_cf(cf_tables, OLDEST_SNAPSHOT_SEQUENCE_PROPERTY)? + .context("RocksDB did not expose rocksdb.oldest-snapshot-sequence")? + ) + }; + + let mut safe = Vec::new(); + let mut unsafe_watermark = None; + let mut unsafe_deleted_tables = 0usize; + for (id, state) in deleted.records { + if state.is_reclaim_safe(oldest_snapshot_sequence) { + safe.push(id); + } else { + unsafe_watermark = watermark(unsafe_watermark, Some(id)); + unsafe_deleted_tables += 1; + } + } + + // A build active when the fence was taken remains pinned even if it finishes between + // fence creation and the metadata snapshot. Builds registered later have ids strictly + // above the process-monotonic boundary. Neither class needs to retain a lock here. + let watermark = runtime_reclaim_watermark(live, dirty, unsafe_watermark, build_fence); + let hi = reclaim_upper_bound(watermark); + db.delete_file_in_range_cf(cf_tables, RECLAIM_LOWER_BOUND.as_slice(), hi.as_slice())?; + + // Safe is monotonic: no future snapshot can predate one that already passed the purge + // sequence, so these records need not pin later passes even if no SST was eligible now. + if !safe.is_empty() { + let cf_deleted = db.cf_handle(CF_DELETED_TABLES).unwrap(); + let mut batch = RocksWriteBatch::default(); + for id in &safe { + batch.delete_cf(cf_deleted, id); + } + db.write(batch)?; + } + + Ok(RuntimeReclaimResult { + snapshot_count, + oldest_snapshot_sequence, + safe_deleted_tables: safe.len(), + unsafe_deleted_tables, + skipped_malformed_chunks: live_scan.malformed, + watermark + }) +} + +fn runtime_reclaim_watermark( + live: Option, + dirty: impl IntoIterator, + unsafe_deleted: Option, + build_fence: ReclaimFence +) -> Option { + watermark( + live, + dirty + .into_iter() + .chain(unsafe_deleted) + .chain(build_fence.active_build_watermark) + .chain(Some(build_fence.boundary)) + ) +} + /// Crash recovery -- purge orphaned `CF_DIRTY_TABLES` markers. /// /// A dirty marker is written when a build starts and removed when its chunk commits. @@ -131,13 +411,19 @@ pub(crate) fn reclaim_disk_space(db: &RocksDB) -> anyhow::Result<()> { /// MUST run only with no build in flight (e.g. startup before ingest): it treats EVERY /// dirty marker as an orphan, so it would tombstone a live build's data. Returns orphans /// purged. +/// +/// TODO: If restart-free recovery becomes necessary, extend in-flight ownership through +/// chunk commit or add durable build state. Marker age plus the builder registry is not +/// sufficient: a finished table can still be waiting for its chunk transaction. Until then, +/// every startup attempts recovery before ingest; failures are logged and retried at the +/// next startup. pub(crate) fn purge_orphan_dirty_tables(db: &RocksDB) -> anyhow::Result { // Collect first, mutate after: writing while iterating disturbs the cursor. let (orphans, malformed) = scan_table_ids(db, CF_DIRTY_TABLES)?; drop_malformed_keys(db, CF_DIRTY_TABLES, &malformed)?; for id in &orphans { - purge_table(db, id)?; + purge_orphan_table(db, id)?; } Ok(orphans.len()) @@ -149,18 +435,7 @@ pub(crate) fn purge_orphan_dirty_tables(db: &RocksDB) -> anyhow::Result { /// skipping could lift the watermark over live data. The watermark is global, so one bad /// chunk anywhere disables the unlink for all datasets -- the safe failure mode. fn min_live_table_id(db: &RocksDB) -> anyhow::Result> { - let mut min: Option = None; - { - let cf_chunks = db.cf_handle(CF_CHUNKS).unwrap(); - let mut it = db.raw_iterator_cf(cf_chunks); - it.seek_to_first(); - while it.valid() { - let chunk: Chunk = borsh::from_slice(it.value().unwrap())?; - min = watermark(min, chunk.tables().values().copied()); - it.next(); - } - it.status()?; - } + let min = min_committed_table_id(db)?; // Malformed dirty keys pin no data, and the startup purge deletes them anyway. This // path must not write -- it serves reclaim on a disk with no room for writes. @@ -168,3 +443,94 @@ fn min_live_table_id(db: &RocksDB) -> anyhow::Result> { Ok(watermark(min, dirty)) } + +fn min_committed_table_id(db: &RocksDB) -> anyhow::Result> { + let cf_chunks = db.cf_handle(CF_CHUNKS).unwrap(); + let mut it = db.raw_iterator_cf(cf_chunks); + Ok(collect_min_committed_table_id(&mut it, MalformedChunkPolicy::Abort)?.min) +} + +fn min_committed_table_id_at( + db: &RocksDB, + snapshot: &RocksSnapshot<'_, RocksDB> +) -> anyhow::Result { + let cf_chunks = db.cf_handle(CF_CHUNKS).unwrap(); + let mut it = snapshot.raw_iterator_cf(cf_chunks); + collect_min_committed_table_id(&mut it, MalformedChunkPolicy::Skip) +} + +#[derive(Clone, Copy)] +enum MalformedChunkPolicy { + Abort, + Skip +} + +#[derive(Default)] +struct CommittedTableScan { + min: Option, + malformed: usize +} + +fn collect_min_committed_table_id( + it: &mut RocksIterator<'_, RocksDB>, + malformed_policy: MalformedChunkPolicy +) -> anyhow::Result { + let mut min = None; + let mut malformed = 0usize; + it.seek_to_first(); + while it.valid() { + match borsh::from_slice::(it.value().unwrap()) { + Ok(chunk) => min = watermark(min, chunk.tables().values().copied()), + Err(err) => match malformed_policy { + MalformedChunkPolicy::Abort => return Err(err).context("invalid committed chunk"), + MalformedChunkPolicy::Skip => malformed = malformed.saturating_add(1) + } + } + it.next(); + } + it.status()?; + Ok(CommittedTableScan { min, malformed }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn id(byte: u8) -> TableId { + TableId::try_from_key(&[byte; 16]).unwrap() + } + + #[test] + fn runtime_watermark_never_crosses_the_build_fence() { + let fence = ReclaimFence { + active_build_watermark: Some(id(5)), + boundary: id(7) + }; + + assert_eq!(runtime_reclaim_watermark(None, None, None, fence), Some(id(5))); + assert_eq!( + runtime_reclaim_watermark(Some(id(3)), [id(8)], None, fence), + Some(id(3)) + ); + + let fence = ReclaimFence { + active_build_watermark: None, + boundary: id(7) + }; + assert_eq!( + runtime_reclaim_watermark(Some(id(9)), [id(8)], None, fence), + Some(id(7)) + ); + } + + #[test] + fn full_tombstone_batches_are_taken_for_flushing() { + let empty_batch_size = RocksWriteBatch::default().size_in_bytes(); + let mut batch = RocksWriteBatch::default(); + batch.put(b"key", vec![0u8; WRITE_BATCH_SIZE_LIMIT]); + + let full = take_full_tombstone_batch(&mut batch, WRITE_BATCH_SIZE_LIMIT); + assert!(full.is_some()); + assert_eq!(batch.size_in_bytes(), empty_batch_size); + } +} diff --git a/crates/storage/src/db/write/storage.rs b/crates/storage/src/db/write/storage.rs index 644c2a86..4965586c 100644 --- a/crates/storage/src/db/write/storage.rs +++ b/crates/storage/src/db/write/storage.rs @@ -8,6 +8,8 @@ use crate::{ kv::KvWrite }; +pub(super) const WRITE_BATCH_SIZE_LIMIT: usize = 8 * 1024 * 1024; + pub struct TableStorage<'a> { write_batch: RocksWriteBatch, db: &'a RocksDB, @@ -49,7 +51,7 @@ impl<'a> TableStorage<'a> { impl<'a> KvWrite for TableStorage<'a> { fn put(&mut self, key: &[u8], value: &[u8]) -> anyhow::Result<()> { self.write_batch.put_cf(self.cf, key, value); - if self.byte_size() > 8 * 1024 * 1024 { + if self.byte_size() >= WRITE_BATCH_SIZE_LIMIT { self.flush()?; } Ok(()) diff --git a/crates/storage/src/db/write/table_builder.rs b/crates/storage/src/db/write/table_builder.rs index 1185e78a..04064737 100644 --- a/crates/storage/src/db/write/table_builder.rs +++ b/crates/storage/src/db/write/table_builder.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeSet; +use std::{collections::BTreeSet, sync::Arc}; use anyhow::{ensure, Context}; use arrow::{array::RecordBatch, datatypes::SchemaRef}; @@ -10,9 +10,11 @@ use sqd_array::{ use crate::{ db::{ db::{RocksDB, CF_TABLES}, - table_id::TableId, - write::storage::TableStorage, - ReadSnapshot + write::{ + inflight::{InflightTableGuard, InflightTableRegistry}, + storage::TableStorage + }, + ReadSnapshot, TableId }, table::{ key::TableKeyFactory, @@ -28,12 +30,14 @@ pub struct TableBuilder<'a> { schema: SchemaRef, columns_with_stats: BTreeSet, writer: TableWriter<'a>, - db: &'a RocksDB + db: &'a RocksDB, + _inflight: InflightTableGuard } impl<'a> TableBuilder<'a> { - pub fn new(db: &'a RocksDB, schema: SchemaRef) -> Self { - let table_id = TableId::new(); + pub(crate) fn new(db: &'a RocksDB, schema: SchemaRef, inflight: &Arc) -> Self { + let build_guard = inflight.register(); + let table_id = build_guard.table_id(); let mut storage = TableStorage::new(db); storage.mark_table_dirty(table_id); @@ -45,7 +49,8 @@ impl<'a> TableBuilder<'a> { schema, columns_with_stats: BTreeSet::new(), writer, - db + db, + _inflight: build_guard } } diff --git a/crates/storage/src/db/write/tx.rs b/crates/storage/src/db/write/tx.rs index 4fca2e80..f97dfe13 100644 --- a/crates/storage/src/db/write/tx.rs +++ b/crates/storage/src/db/write/tx.rs @@ -126,8 +126,8 @@ impl<'a> Tx<'a> { } pub fn delete_table(&self, table_id: &TableId) -> anyhow::Result<()> { - // Value unused; the key's presence is the signal. `ops::logical_cleanup` - // point-deletes the table's data and drops this entry. + // Empty is the backwards-compatible `DeleteRequested` state. `logical_cleanup` + // point-deletes the table and replaces it with a sequence-stamped purge record. self.transaction .put_cf(self.cf_handle(CF_DELETED_TABLES), table_id, [])?; Ok(()) diff --git a/crates/storage/tests/cleanup_reclaim.rs b/crates/storage/tests/cleanup_reclaim.rs index edb2a98e..1136870f 100644 --- a/crates/storage/tests/cleanup_reclaim.rs +++ b/crates/storage/tests/cleanup_reclaim.rs @@ -3,6 +3,9 @@ //! * Physical reclaim ([`Database::reclaim_disk_space`]) -- SST-file unlink below the //! live watermark. It ignores snapshots, so it runs only where there are no live //! readers (startup in production); fully decoupled from Phase 1. +//! * Runtime physical reclaim ([`Database::reclaim_disk_space_runtime`]) -- the same +//! unlink with deleted tables pinned until RocksDB's oldest snapshot passes their +//! point-tombstone sequence. mod mock_db; #[allow(dead_code)] // shared with the other integration tests; this one uses a subset @@ -189,6 +192,86 @@ fn reclaim_breaks_a_live_pre_deletion_reader() { ); } +/// Runtime reclaim pins a deleted table while a pre-deletion query snapshot can still read +/// it. Once that one snapshot is dropped, a continuously live *newer* snapshot does not +/// prevent the same files from being unlinked. +#[test] +fn runtime_reclaim_waits_only_for_pre_delete_snapshots() { + let mut db = MockDB::uncached(); + let t = db.commit_table(3000); + db.compact_to_bottom(); + let before = db.sst_size(); + assert!(before > 0); + + let old_reader = db.snapshot(); + db.delete(&t); + assert_eq!(db.cleanup(), 1); + db.flush(); + + let first = db.runtime_reclaim(); + assert_eq!(first.safe_deleted_tables, 0); + assert_eq!(first.unsafe_deleted_tables, 1); + assert!(first.oldest_snapshot_sequence.is_some()); + assert!(db.sst_size() >= before, "the old reader must pin its table files"); + assert_eq!(db.read(&old_reader, &t), t.rows); + + // This models constant query traffic: at least one newer snapshot remains live while + // the last pre-deletion snapshot goes away. + let newer_reader = db.snapshot(); + drop(old_reader); + + let second = db.runtime_reclaim(); + assert!(second.snapshot_count >= 1, "the newer query snapshot is still live"); + assert_eq!(second.safe_deleted_tables, 1); + assert_eq!(second.unsafe_deleted_tables, 0); + assert!( + db.sst_size() * 4 < before, + "newer snapshots must not pin pre-tombstone files: before={before} after={}", + db.sst_size() + ); + assert!(!db.has_visible_chunk()); + drop(newer_reader); +} + +/// A deletion request is not sequence-safe until Phase 1 has actually written its point +/// tombstones. Runtime reclaim therefore leaves both its files and cleanup record alone. +#[test] +fn runtime_reclaim_waits_for_logical_cleanup() { + let mut db = MockDB::new(); + let t = db.commit_table(2000); + db.compact_to_bottom(); + let before = db.sst_size(); + + db.delete(&t); + let report = db.runtime_reclaim(); + assert_eq!(report.safe_deleted_tables, 0); + assert_eq!(report.unsafe_deleted_tables, 1); + assert!(db.sst_size() >= before); + + assert_eq!(db.cleanup(), 1, "runtime reclaim must preserve pending Phase 1 work"); +} + +/// Disabling runtime reclaim also disables its sequence bookkeeping. Point deletes remain +/// snapshot-safe, but completed purge records must not accumulate forever. +#[test] +fn cleanup_drops_sequence_records_when_runtime_reclaim_is_disabled() { + let mut db = MockDB::without_runtime_reclaim(); + let first_deleted = db.commit_table(1000); + let newly_deleted = db.commit_table(1000); + let old_reader = db.snapshot(); + + db.delete(&first_deleted); + assert_eq!(db.cleanup(), 1); + db.delete(&newly_deleted); + assert_eq!(db.cleanup(), 1); + assert_eq!(db.read(&old_reader, &first_deleted), first_deleted.rows); + assert_eq!(db.read(&old_reader, &newly_deleted), newly_deleted.rows); + drop(old_reader); + + assert!(db.try_runtime_reclaim().is_err()); + assert_eq!(db.cleanup(), 0); +} + /// An orphaned `DIRTY_TABLES` marker -- left when a build dies before committing its /// chunk -- counts as a live table, pinning disk reclaim for every later table. Startup /// recovery drops it so the watermark lifts. diff --git a/crates/storage/tests/database_ops.rs b/crates/storage/tests/database_ops.rs index 357c0472..51d22257 100644 --- a/crates/storage/tests/database_ops.rs +++ b/crates/storage/tests/database_ops.rs @@ -43,6 +43,31 @@ fn create_dataset() { assert!(stats.is_some()); } +#[test] +fn runtime_reclaim_does_not_wait_for_an_active_table_build() { + let db_dir = tempfile::tempdir().unwrap(); + let db = DatabaseSettings::default() + .with_runtime_reclaim(true) + .open(db_dir.path()) + .unwrap(); + let db = Arc::new(db); + let schema = Arc::new(Schema::new(vec![Field::new("data", DataType::UInt32, true)])); + let builder = db.new_table_builder(schema); + let (completed_tx, completed_rx) = std::sync::mpsc::channel(); + + let reclaim_db = Arc::clone(&db); + let reclaim = std::thread::spawn(move || { + reclaim_db.reclaim_disk_space_runtime().unwrap(); + completed_tx.send(()).unwrap(); + }); + + let completed_without_waiting = completed_rx.recv_timeout(std::time::Duration::from_secs(5)).is_ok(); + drop(builder); + reclaim.join().unwrap(); + + assert!(completed_without_waiting, "runtime reclaim waited for a table build"); +} + #[test] fn basic_chunks_test() { let (db, dataset_id) = setup_db(); diff --git a/crates/storage/tests/mock_db/mod.rs b/crates/storage/tests/mock_db/mod.rs index b93167ef..82dbaf04 100644 --- a/crates/storage/tests/mock_db/mod.rs +++ b/crates/storage/tests/mock_db/mod.rs @@ -7,7 +7,9 @@ use std::{sync::Arc, time::Duration}; use arrow::datatypes::{DataType, Schema}; use sqd_storage::{ - db::{Chunk, Database, DatabaseSettings, DatasetId, DatasetKind, ReadSnapshot, TableId}, + db::{ + reclaim::RuntimeReclaimResult, Chunk, Database, DatabaseSettings, DatasetId, DatasetKind, ReadSnapshot, TableId + }, table::write::{use_small_buffers, RestoreBufferSizesGuard} }; use tempfile::TempDir; @@ -57,6 +59,7 @@ impl MockDB { DatabaseSettings::default() .with_rocksdb_stats(true) .with_auto_compactions(false) + .with_runtime_reclaim(true) ) } @@ -69,6 +72,7 @@ impl MockDB { .with_auto_compactions(false) .with_data_cache_size(0) .with_chunk_cache_size(0) + .with_runtime_reclaim(true) ) } @@ -167,12 +171,29 @@ impl MockDB { self.db.cleanup().unwrap() } + pub fn without_runtime_reclaim() -> Self { + Self::open( + DatabaseSettings::default() + .with_rocksdb_stats(true) + .with_auto_compactions(false) + ) + } + /// Physically unlink dead SST files below the live watermark. The caller is responsible /// for there being no live pre-deletion reader (as at startup in production). pub fn reclaim(&self) { self.db.reclaim_disk_space().unwrap() } + /// Snapshot-aware physical reclaim used during normal service operation. + pub fn runtime_reclaim(&self) -> RuntimeReclaimResult { + self.db.reclaim_disk_space_runtime().unwrap() + } + + pub fn try_runtime_reclaim(&self) -> anyhow::Result { + self.db.reclaim_disk_space_runtime() + } + pub fn purge_orphans(&self) -> usize { self.db.purge_orphan_dirty_tables().unwrap() }