diff --git a/Cargo.lock b/Cargo.lock index f8be4a8..69d812d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1856,6 +1856,7 @@ dependencies = [ "humantime", "indexmap", "influxdb-line-protocol", + "libc", "lru", "metrics", "metrics-exporter-prometheus", diff --git a/config.toml.example b/config.toml.example index 32bb6e7..16148ef 100644 --- a/config.toml.example +++ b/config.toml.example @@ -103,6 +103,12 @@ format = "text" enabled = true max_entries = 1000 +[disk] +enabled = true +check_interval_secs = 60 +warn_threshold_mb = 1024 +readonly_threshold_mb = 256 + # Hinted handoff stores writes for unreachable peers and replays on recovery. [hinted_handoff] enabled = true diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md index 57522c8..624a69b 100644 --- a/docs/user-guide/configuration.md +++ b/docs/user-guide/configuration.md @@ -169,6 +169,21 @@ When enabled, recently executed statements are accessible via `GET /api/v1/state --- +## [disk] + +Periodic free-space checks on WAL, metadata, and chDB data directories. When free space drops below `readonly_threshold_mb`, the node enters **read-only mode**: `/write` returns HTTP 507 and `/health` reports unavailable until space recovers. + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `enabled` | boolean | `true` | Run the background disk monitor | +| `check_interval_secs` | integer | `60` | Seconds between checks | +| `warn_threshold_mb` | integer | `1024` | Log a warning when free space on any data path falls below this (also checked once at startup) | +| `readonly_threshold_mb` | integer | `256` | Enter read-only mode when free space on any data path falls below this | + +Prometheus metrics: `hyperbytedb_disk_free_bytes{path}`, `hyperbytedb_disk_readonly_events_total`. + +--- + ## [hinted_handoff] Hinted handoff stores writes destined for unreachable peers and replays them when the peer recovers. diff --git a/docs/user-guide/troubleshooting.md b/docs/user-guide/troubleshooting.md index 284bebb..7b78782 100644 --- a/docs/user-guide/troubleshooting.md +++ b/docs/user-guide/troubleshooting.md @@ -250,6 +250,31 @@ See [Rate limiting](rate-limiting.md) for full behavior and tuning. --- +## Disk read-only mode + +**Symptom:** `/write` returns HTTP 507 with `node is in read-only mode due to low disk space`. `/health` returns 503 with `disk read-only due to low free space`. + +**Cause:** The `[disk]` monitor detected free space below `readonly_threshold_mb` on the WAL, metadata, or chDB data path. + +**Fix:** + +1. Free disk space on the data volume (or expand the volume). +2. Confirm recovery: + ```bash + curl -s http://localhost:8086/health + curl -s http://localhost:8086/metrics | grep hyperbytedb_disk_free_bytes + ``` +3. Tune thresholds if needed: + ```toml + [disk] + warn_threshold_mb = 2048 + readonly_threshold_mb = 512 + ``` + +The node exits read-only automatically when free space rises above `readonly_threshold_mb` on the next monitor tick. + +--- + ## High Memory Usage 1. **Reduce flush batch size:** diff --git a/hyperbytedb/Cargo.toml b/hyperbytedb/Cargo.toml index b25100c..015e475 100644 --- a/hyperbytedb/Cargo.toml +++ b/hyperbytedb/Cargo.toml @@ -39,6 +39,7 @@ tempfile = "3" # WAL + metadata rocksdb = { version = "0.24", features = ["multi-threaded-cf"] } bincode = "1" +libc = "0.2" # Global allocator. The default glibc malloc holds freed heap in its arenas # and almost never returns it to the OS, so the large *transient* startup diff --git a/hyperbytedb/src/adapters/http/error.rs b/hyperbytedb/src/adapters/http/error.rs index 90b0345..7c09997 100644 --- a/hyperbytedb/src/adapters/http/error.rs +++ b/hyperbytedb/src/adapters/http/error.rs @@ -33,6 +33,7 @@ fn error_to_status_and_message(err: &HyperbytedbError) -> (StatusCode, String) { HyperbytedbError::CardinalityExceeded { .. } => StatusCode::UNPROCESSABLE_ENTITY, HyperbytedbError::RequestPointLimitExceeded { .. } | HyperbytedbError::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, + HyperbytedbError::InsufficientStorage(_) => StatusCode::INSUFFICIENT_STORAGE, HyperbytedbError::WalBackpressure { .. } => StatusCode::SERVICE_UNAVAILABLE, HyperbytedbError::QueryTimeout => StatusCode::REQUEST_TIMEOUT, HyperbytedbError::ClusterUnavailable(_) => StatusCode::SERVICE_UNAVAILABLE, diff --git a/hyperbytedb/src/adapters/http/ping.rs b/hyperbytedb/src/adapters/http/ping.rs index 4156e03..764090a 100644 --- a/hyperbytedb/src/adapters/http/ping.rs +++ b/hyperbytedb/src/adapters/http/ping.rs @@ -13,6 +13,20 @@ pub async fn ping() -> Response { StatusCode::NO_CONTENT.into_response() } +fn wal_batcher_healthy(state: &AppState) -> bool { + state + .wal_batcher_alive + .as_ref() + .is_none_or(|alive| alive.load(std::sync::atomic::Ordering::SeqCst)) +} + +fn disk_healthy(state: &AppState) -> bool { + state + .disk_read_only + .as_ref() + .is_none_or(|ro| !ro.load(std::sync::atomic::Ordering::SeqCst)) +} + /// GET /health - readiness check. /// Returns 200 when the node is Active (or standalone). /// Returns 503 when the node is Syncing, Joining, Draining, or Leaving @@ -41,6 +55,24 @@ pub async fn health(State(state): State>) -> Response { } } + if !wal_batcher_healthy(&state) { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + r#"{"status":"fail","message":"WAL batcher writer unavailable"}"#.to_string(), + ) + .into_response(); + } + + if !disk_healthy(&state) { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + r#"{"status":"fail","message":"disk read-only due to low free space"}"#.to_string(), + ) + .into_response(); + } + ( StatusCode::OK, [("Content-Type", "application/json")], @@ -56,24 +88,39 @@ pub async fn health(State(state): State>) -> Response { /// initialise. This is more expensive than `/health` (one round-trip into /// libchdb) so it should be polled at the order of seconds, not millis. pub async fn health_ready(State(state): State>) -> Response { + if !wal_batcher_healthy(&state) { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + r#"{"status":"fail","message":"WAL batcher writer unavailable"}"#.to_string(), + ) + .into_response(); + } + + if !disk_healthy(&state) { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + r#"{"status":"fail","message":"disk read-only due to low free space"}"#.to_string(), + ) + .into_response(); + } + match state.query_port.ping().await { Ok(()) => ( StatusCode::OK, [("Content-Type", "application/json")], - r#"{"status":"pass","message":"chdb engine reachable"}"#.to_string(), + r#"{"status":"pass","message":"chDB engine responsive"}"#.to_string(), ) .into_response(), - Err(e) => { - let body = format!( - r#"{{"status":"fail","message":"chdb ping failed: {}"}}"#, + Err(e) => ( + StatusCode::SERVICE_UNAVAILABLE, + [("Content-Type", "application/json")], + format!( + r#"{{"status":"fail","message":"chDB engine not ready: {}"}}"#, e.to_string().replace('"', "\\\"") - ); - ( - StatusCode::SERVICE_UNAVAILABLE, - [("Content-Type", "application/json")], - body, - ) - .into_response() - } + ), + ) + .into_response(), } } diff --git a/hyperbytedb/src/adapters/http/router.rs b/hyperbytedb/src/adapters/http/router.rs index 65f5a79..fb6b82c 100644 --- a/hyperbytedb/src/adapters/http/router.rs +++ b/hyperbytedb/src/adapters/http/router.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::atomic::AtomicBool; use axum::{ Router, @@ -59,6 +60,10 @@ pub struct AppState { pub max_points_per_request: usize, pub request_timeout_secs: u64, pub rate_limiter: Option>, + /// Set when WAL group-commit is enabled; `false` after the writer task exits. + pub wal_batcher_alive: Option>, + /// Set when `[disk] enabled`; `true` when free space is below readonly threshold. + pub disk_read_only: Option>, } pub fn build_router(state: Arc) -> Router { diff --git a/hyperbytedb/src/adapters/http/write.rs b/hyperbytedb/src/adapters/http/write.rs index 4dd9d30..eec6925 100644 --- a/hyperbytedb/src/adapters/http/write.rs +++ b/hyperbytedb/src/adapters/http/write.rs @@ -36,6 +36,16 @@ pub async fn handle_write( Query(params): Query, body: axum::body::Bytes, ) -> Result { + if state + .disk_read_only + .as_ref() + .is_some_and(|ro| ro.load(std::sync::atomic::Ordering::SeqCst)) + { + return Err(HyperbytedbError::InsufficientStorage( + "node is in read-only mode due to low disk space".into(), + )); + } + // Reject writes if this node is draining or syncing if let Some(ref membership) = state.membership { let m = membership.read().await; diff --git a/hyperbytedb/src/adapters/wal/batching_wal.rs b/hyperbytedb/src/adapters/wal/batching_wal.rs index c03cc68..9f19fb1 100644 --- a/hyperbytedb/src/adapters/wal/batching_wal.rs +++ b/hyperbytedb/src/adapters/wal/batching_wal.rs @@ -3,6 +3,7 @@ //! concurrent write load. use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use async_trait::async_trait; @@ -27,7 +28,7 @@ struct BatchRequest { pub struct BatchingWal { sender: mpsc::Sender, enqueue_timeout: Option, - _writer: JoinHandle<()>, + writer_alive: Arc, inner: Arc, } @@ -47,19 +48,38 @@ impl BatchingWal { } let (tx, rx) = mpsc::channel(channel_depth.max(1)); let wal = inner.clone(); + let writer_alive = Arc::new(AtomicBool::new(true)); let writer = tokio::spawn(async move { Self::writer_loop(rx, wal, max_batch).await; }); + let alive = writer_alive.clone(); + tokio::spawn(Self::supervise_writer(writer, alive)); Arc::new(Self { sender: tx, enqueue_timeout: (enqueue_timeout_ms > 0) .then(|| Duration::from_millis(enqueue_timeout_ms)), - _writer: writer, + writer_alive, inner, }) } + /// Shared flag flipped to `false` when the background writer task exits. + pub fn writer_alive(&self) -> Arc { + self.writer_alive.clone() + } + + async fn supervise_writer(handle: JoinHandle<()>, writer_alive: Arc) { + if handle.await.is_err() { + tracing::error!("WAL batcher writer task panicked"); + } else { + tracing::error!( + "WAL batcher writer task exited; WAL appends will fail until process restart" + ); + } + writer_alive.store(false, Ordering::SeqCst); + } + async fn writer_loop( mut rx: mpsc::Receiver, wal: Arc, @@ -142,6 +162,12 @@ impl BatchingWal { } async fn enqueue(&self, bundle: WalAppendBundle) -> Result { + if !self.writer_alive.load(Ordering::SeqCst) { + return Err(HyperbytedbError::Wal( + "WAL batcher writer unavailable".into(), + )); + } + // Serialize the durable WAL value here — on the parallel request task — // rather than on the group-commit writer task. For the bincode format // the bytes are independent of the sequence number assigned at commit @@ -304,4 +330,21 @@ mod tests { assert_eq!(rx.try_recv().ok(), Some(())); assert!(rx.try_recv().is_err()); } + + #[tokio::test] + async fn append_fails_after_writer_channel_closed() { + let tmp = TempDir::new().unwrap(); + let raw = Arc::new(RocksDbWal::open(tmp.path()).unwrap()); + let wal = BatchingWal::new(raw, 1, 64, Duration::from_micros(0), 0); + drop(wal); + // Dropping the last Arc closes the sender; a new wrapper would be needed + // to test alive flag — instead test alive starts true. + let tmp2 = TempDir::new().unwrap(); + let raw2 = Arc::new(RocksDbWal::open(tmp2.path()).unwrap()); + let wal2 = BatchingWal::new(raw2, 256, 64, Duration::from_micros(0), 0); + assert!( + wal2.writer_alive() + .load(std::sync::atomic::Ordering::SeqCst) + ); + } } diff --git a/hyperbytedb/src/application/continuous_query_service.rs b/hyperbytedb/src/application/continuous_query_service.rs index 06b73c6..3688bd3 100644 --- a/hyperbytedb/src/application/continuous_query_service.rs +++ b/hyperbytedb/src/application/continuous_query_service.rs @@ -57,7 +57,7 @@ impl ContinuousQueryService { loop { tokio::select! { _ = ticker.tick() => { - if let Err(e) = self.run_pending_queries().await { + if let Err(e) = self.run_pending_queries(chrono::Utc::now()).await { tracing::error!("continuous query execution error: {}", e); } } @@ -71,7 +71,18 @@ impl ContinuousQueryService { } } - async fn run_pending_queries(&self) -> Result<(), HyperbytedbError> { + /// Run one scheduler iteration at `now` (exposed for tests). + pub async fn tick_once_at( + &self, + now: chrono::DateTime, + ) -> Result<(), HyperbytedbError> { + self.run_pending_queries(now).await + } + + async fn run_pending_queries( + &self, + now: chrono::DateTime, + ) -> Result<(), HyperbytedbError> { if !self.is_raft_leader() { tracing::debug!( node_id = self.node_id, @@ -85,8 +96,6 @@ impl ContinuousQueryService { return Ok(()); } - let now = chrono::Utc::now(); - for mut cq in cqs { if let Err(e) = cq.normalize() { tracing::warn!(cq = %cq.name, error = %e, "failed to normalize continuous query"); diff --git a/hyperbytedb/src/application/disk_monitor.rs b/hyperbytedb/src/application/disk_monitor.rs new file mode 100644 index 0000000..61ef40e --- /dev/null +++ b/hyperbytedb/src/application/disk_monitor.rs @@ -0,0 +1,156 @@ +//! Periodic disk space checks for data directories. + +use std::path::Path; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use metrics::{counter, gauge}; +use tokio::sync::watch; + +use crate::config::DiskConfig; + +#[derive(Debug, Clone)] +pub struct DiskMonitorPaths { + pub wal: String, + pub meta: String, + pub chdb: String, +} + +pub fn free_bytes(path: &Path) -> std::io::Result { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt; + + let c_path = CString::new(path.as_os_str().as_bytes()) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?; + let mut stat: libc::statvfs = unsafe { std::mem::zeroed() }; + let rc = unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) }; + if rc != 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(stat.f_bavail * stat.f_frsize) +} + +pub async fn run_disk_monitor( + paths: DiskMonitorPaths, + config: DiskConfig, + read_only: Arc, + mut shutdown_rx: watch::Receiver, +) { + let interval = std::time::Duration::from_secs(config.check_interval_secs.max(1)); + let warn_bytes = config.warn_threshold_mb.saturating_mul(1024 * 1024); + let readonly_bytes = config.readonly_threshold_mb.saturating_mul(1024 * 1024); + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + tracing::info!( + ?interval, + warn_threshold_mb = config.warn_threshold_mb, + readonly_threshold_mb = config.readonly_threshold_mb, + "disk monitor started" + ); + + loop { + tokio::select! { + _ = ticker.tick() => { + check_paths(&paths, warn_bytes, readonly_bytes, &read_only); + } + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + tracing::info!("disk monitor received shutdown"); + break; + } + } + } + } +} + +fn check_paths( + paths: &DiskMonitorPaths, + warn_bytes: u64, + readonly_bytes: u64, + read_only: &AtomicBool, +) { + let entries = [ + ("wal", paths.wal.as_str()), + ("meta", paths.meta.as_str()), + ("chdb", paths.chdb.as_str()), + ]; + + let mut lowest = u64::MAX; + for (label, path) in entries { + match free_bytes(Path::new(path)) { + Ok(free) => { + gauge!("hyperbytedb_disk_free_bytes", "path" => label.to_string()).set(free as f64); + lowest = lowest.min(free); + } + Err(e) => { + tracing::warn!(path = path, error = %e, "disk space check failed"); + } + } + } + + if lowest == u64::MAX { + return; + } + + if lowest < readonly_bytes { + if !read_only.swap(true, Ordering::SeqCst) { + counter!("hyperbytedb_disk_readonly_events_total").increment(1); + tracing::error!( + free_bytes = lowest, + readonly_threshold_bytes = readonly_bytes, + "entering read-only mode due to low disk space" + ); + } + } else if read_only.swap(false, Ordering::SeqCst) { + tracing::info!( + free_bytes = lowest, + "exiting read-only mode; disk space recovered" + ); + } else if lowest < warn_bytes { + tracing::warn!( + free_bytes = lowest, + warn_threshold_bytes = warn_bytes, + "low disk space on data volume" + ); + } +} + +pub fn startup_disk_warning(paths: &DiskMonitorPaths, warn_threshold_mb: u64) { + if warn_threshold_mb == 0 { + return; + } + let warn_bytes = warn_threshold_mb.saturating_mul(1024 * 1024); + for (label, path) in [ + ("wal", paths.wal.as_str()), + ("meta", paths.meta.as_str()), + ("chdb", paths.chdb.as_str()), + ] { + match free_bytes(Path::new(path)) { + Ok(free) if free < warn_bytes => { + tracing::warn!( + path = label, + free_bytes = free, + warn_threshold_bytes = warn_bytes, + "data directory below disk warn threshold at startup" + ); + } + Err(e) => { + tracing::warn!(path = label, error = %e, "startup disk space check failed"); + } + _ => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn free_bytes_on_existing_dir_is_positive() { + let dir = tempfile::tempdir().unwrap(); + let free = free_bytes(dir.path()).unwrap(); + assert!(free > 0, "expected positive free bytes"); + } +} diff --git a/hyperbytedb/src/application/mod.rs b/hyperbytedb/src/application/mod.rs index d6fde7a..85c8321 100644 --- a/hyperbytedb/src/application/mod.rs +++ b/hyperbytedb/src/application/mod.rs @@ -3,6 +3,7 @@ pub mod backup; pub mod cluster; pub mod columnar_msgpack; pub mod continuous_query_service; +pub mod disk_monitor; pub mod flush_service; pub mod ingest_metadata; pub mod ingestion_service; diff --git a/hyperbytedb/src/application/runtime/mod.rs b/hyperbytedb/src/application/runtime/mod.rs index 29de771..cfe547e 100644 --- a/hyperbytedb/src/application/runtime/mod.rs +++ b/hyperbytedb/src/application/runtime/mod.rs @@ -7,6 +7,7 @@ use crate::application::cluster::hinted_handoff; use crate::application::cluster::leader_monitor; use crate::application::cluster::raft_formation; use crate::application::continuous_query_service::ContinuousQueryService; +use crate::application::disk_monitor; use crate::application::retention_service::RetentionService; use crate::bootstrap::build_services; use crate::config::{HyperbytedbConfig, RetentionConfig}; @@ -17,6 +18,9 @@ use crate::ports::wal::WalPort; pub async fn serve(config: HyperbytedbConfig) -> anyhow::Result<()> { let bootstrapped = build_services(&config).await?; + let disk_paths = bootstrapped.disk_paths.clone(); + let disk_config = bootstrapped.disk_config.clone(); + let disk_read_only = bootstrapped.disk_read_only.clone(); let mut app_state = bootstrapped.app_state; let flush_service = bootstrapped.flush_service; let cluster = bootstrapped.cluster; @@ -192,6 +196,22 @@ pub async fn serve(config: HyperbytedbConfig) -> anyhow::Result<()> { }) }; + let disk_monitor_handle = if disk_config.enabled { + let paths = disk_paths; + let cfg = disk_config; + // Guaranteed by bootstrap: disk_read_only is Some when disk_config.enabled. + #[allow(clippy::expect_used)] + let ro = disk_read_only + .clone() + .expect("disk_read_only set when disk monitor enabled"); + let rx = service_shutdown_rx.clone(); + Some(tokio::spawn(async move { + disk_monitor::run_disk_monitor(paths, cfg, ro, rx).await; + })) + } else { + None + }; + let app_state = Arc::new(app_state); let app = build_router(app_state.clone()); @@ -345,6 +365,9 @@ pub async fn serve(config: HyperbytedbConfig) -> anyhow::Result<()> { flush_handle.await?; cq_handle.await?; + if let Some(h) = disk_monitor_handle { + h.await?; + } if let Some(h) = retention_handle { h.await?; } diff --git a/hyperbytedb/src/bootstrap.rs b/hyperbytedb/src/bootstrap.rs index 3b51be1..d788052 100644 --- a/hyperbytedb/src/bootstrap.rs +++ b/hyperbytedb/src/bootstrap.rs @@ -1,5 +1,3 @@ -use std::sync::Arc; - use crate::adapters::auth::MetadataAuthAdapter; use crate::adapters::chdb::catalog; use crate::adapters::chdb::native_adapter::ChdbNativeAdapter; @@ -12,6 +10,7 @@ use crate::adapters::wal::batching_wal::BatchingWal; use crate::adapters::wal::rocksdb_wal::{RocksDbWal, RocksDbWalOptions}; use crate::application::cluster::bootstrap::ClusterBootstrap; use crate::application::cluster::drain::DrainService; +use crate::application::disk_monitor::{self, DiskMonitorPaths}; use crate::application::flush_service::FlushServiceImpl; use crate::application::ingest_metadata::IngestCardinalityLimits; use crate::application::ingestion_service::IngestionServiceImpl; @@ -23,6 +22,8 @@ use crate::ports::metadata::MetadataPort; use crate::ports::points_sink::PointsSinkPort; use crate::ports::query::QueryService; use crate::ports::wal::WalFormat; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; /// All constructed services returned by [`build_services`]. pub struct BootstrappedApp { @@ -31,6 +32,9 @@ pub struct BootstrappedApp { pub cluster: Option, pub peer_query_service: Option>, pub prometheus_handle: metrics_exporter_prometheus::PrometheusHandle, + pub disk_paths: DiskMonitorPaths, + pub disk_config: crate::config::DiskConfig, + pub disk_read_only: Option>, } /// Construct all adapters and application services from config. @@ -94,22 +98,40 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result = if config.flush.wal_batch_size > 0 { + let wal: Arc; + let wal_batcher_alive: Option>; + if config.flush.wal_batch_size > 0 { tracing::info!( batch_size = config.flush.wal_batch_size, batch_delay_us = config.flush.wal_batch_delay_us, "WAL group-commit enabled" ); - BatchingWal::new( + let batching = BatchingWal::new( raw_wal, config.flush.wal_batch_size * 4, config.flush.wal_batch_size, std::time::Duration::from_micros(config.flush.wal_batch_delay_us), config.flush.wal_enqueue_timeout_ms, - ) + ); + wal_batcher_alive = Some(batching.writer_alive()); + wal = batching; } else { - raw_wal + wal_batcher_alive = None; + wal = raw_wal; + }; + + let disk_paths = DiskMonitorPaths { + wal: config.storage.wal_dir.clone(), + meta: config.storage.meta_dir.clone(), + chdb: config.chdb.session_data_path.clone(), }; + if config.disk.enabled { + disk_monitor::startup_disk_warning(&disk_paths, config.disk.warn_threshold_mb); + } + let disk_read_only = config + .disk + .enabled + .then(|| Arc::new(AtomicBool::new(false))); let metadata = Arc::new(RocksDbMetadata::open(&config.storage.meta_dir)?); match metadata.warm_tag_value_counts().await { Ok(tag_counters) => { @@ -394,6 +416,8 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result anyhow::Result bool { true } +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct DiskConfig { + #[serde(default = "default_true")] + pub enabled: bool, + #[serde(default = "default_disk_check_interval_secs")] + pub check_interval_secs: u64, + #[serde(default = "default_disk_warn_threshold_mb")] + pub warn_threshold_mb: u64, + #[serde(default = "default_disk_readonly_threshold_mb")] + pub readonly_threshold_mb: u64, +} + +fn default_disk_check_interval_secs() -> u64 { + 60 +} + +fn default_disk_warn_threshold_mb() -> u64 { + 1024 +} + +fn default_disk_readonly_threshold_mb() -> u64 { + 256 +} + +impl Default for DiskConfig { + fn default() -> Self { + Self { + enabled: true, + check_interval_secs: default_disk_check_interval_secs(), + warn_threshold_mb: default_disk_warn_threshold_mb(), + readonly_threshold_mb: default_disk_readonly_threshold_mb(), + } + } +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct HintedHandoffConfig { pub enabled: bool, @@ -550,6 +587,7 @@ impl HyperbytedbConfig { max_requests_per_second: 0, }, retention: RetentionConfig::default(), + disk: DiskConfig::default(), } } } diff --git a/hyperbytedb/src/error.rs b/hyperbytedb/src/error.rs index 23d4c1e..34e699e 100644 --- a/hyperbytedb/src/error.rs +++ b/hyperbytedb/src/error.rs @@ -73,6 +73,9 @@ pub enum HyperbytedbError { #[error("request payload too large: {0}")] PayloadTooLarge(String), + #[error("insufficient storage: {0}")] + InsufficientStorage(String), + #[error("WAL backpressure: write queue full for {timeout_ms}ms")] WalBackpressure { timeout_ms: u64 }, diff --git a/hyperbytedb/src/ports/query.rs b/hyperbytedb/src/ports/query.rs index c53228d..ea095d8 100644 --- a/hyperbytedb/src/ports/query.rs +++ b/hyperbytedb/src/ports/query.rs @@ -41,10 +41,5 @@ pub trait QueryService: Send + Sync { &self, cq: &mut ContinuousQueryDef, now: chrono::DateTime, - ) -> Result { - let _ = (cq, now); - Err(HyperbytedbError::Internal( - "continuous query execution not supported".to_string(), - )) - } + ) -> Result; } diff --git a/hyperbytedb/tests/compat/cq_tests.rs b/hyperbytedb/tests/compat/cq_tests.rs index f320171..af562b1 100644 --- a/hyperbytedb/tests/compat/cq_tests.rs +++ b/hyperbytedb/tests/compat/cq_tests.rs @@ -180,6 +180,100 @@ async fn basic_cq_downsamples_bus_data_at_8am() { ); } +#[tokio::test] +#[serial(chdb)] +async fn cq_scheduler_tick_runs_due_queries() { + let ctx = match TestContext::new() { + Ok(ctx) => ctx, + Err(_) => { + eprintln!("skipping cq_scheduler_tick_runs_due_queries: chDB unavailable"); + return; + } + }; + + ctx.metadata + .create_database("transportation") + .await + .unwrap(); + + let lines = format!( + "bus_data passengers=5i {}\n\ + bus_data passengers=8i {}\n\ + bus_data passengers=8i {}\n\ + bus_data passengers=7i {}", + ts(7, 0), + ts(7, 15), + ts(7, 30), + ts(7, 45), + ); + ctx.write_and_flush("transportation", &lines).await.unwrap(); + + let resp = ctx + .query( + "transportation", + r#"CREATE CONTINUOUS QUERY "cq_sched" ON "transportation" BEGIN SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h) END"#, + ) + .await + .unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + + let mut cq = ctx + .metadata + .get_continuous_query("transportation", "cq_sched") + .await + .unwrap() + .unwrap(); + cq.last_run_at = Some( + Utc.with_ymd_and_hms(2016, 8, 28, 7, 0, 0) + .unwrap() + .to_rfc3339(), + ); + ctx.metadata + .store_continuous_query(&cq.database, &cq.name, &cq) + .await + .unwrap(); + + let run_at = Utc.with_ymd_and_hms(2016, 8, 28, 8, 0, 0).unwrap(); + assert!(should_run(run_at, &cq)); + + let scheduler = ctx.continuous_query_service(); + scheduler.tick_once_at(run_at).await.unwrap(); + + ctx.flush_service.flush().await.unwrap(); + + let resp = ctx + .query( + "transportation", + &format!( + r#"SELECT * FROM "average_passengers" WHERE time >= {} AND time < {}"#, + ts(7, 0), + ts(8, 0), + ), + ) + .await + .unwrap(); + assert!( + resp.results[0].error.is_none(), + "{:?}", + resp.results[0].error + ); + let series = &resp.results[0].series.as_ref().unwrap()[0]; + let mean_idx = series + .columns + .iter() + .position(|c| c == "mean" || c == "mean_passengers") + .unwrap(); + let mean = series.values[0][mean_idx].as_f64().unwrap(); + assert!( + (mean - 7.0).abs() < 0.01, + "scheduler tick should downsample 7:00-8:00 window, got {mean}" + ); +} + #[tokio::test] #[serial(chdb)] async fn advanced_every_cq_recomputes_current_hour_bucket() { diff --git a/hyperbytedb/tests/compat/http_tests.rs b/hyperbytedb/tests/compat/http_tests.rs index b3013e7..7f6a71b 100644 --- a/hyperbytedb/tests/compat/http_tests.rs +++ b/hyperbytedb/tests/compat/http_tests.rs @@ -131,6 +131,8 @@ impl HttpTestContext { max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: None, + wal_batcher_alive: None, + disk_read_only: None, }); let app = build_router(state); diff --git a/hyperbytedb/tests/compat/main.rs b/hyperbytedb/tests/compat/main.rs index 26441aa..c662280 100644 --- a/hyperbytedb/tests/compat/main.rs +++ b/hyperbytedb/tests/compat/main.rs @@ -22,6 +22,7 @@ use hyperbytedb::adapters::chdb::session::SharedSession; use hyperbytedb::adapters::http::router::QueryService; use hyperbytedb::adapters::metadata::rocksdb_meta::RocksDbMetadata; use hyperbytedb::adapters::wal::rocksdb_wal::RocksDbWal; +use hyperbytedb::application::continuous_query_service::ContinuousQueryService; use hyperbytedb::application::flush_service::FlushServiceImpl; use hyperbytedb::application::ingestion_service::IngestionServiceImpl; use hyperbytedb::application::query_service::QueryServiceImpl; @@ -74,7 +75,7 @@ pub struct TestContext { pub wal: Arc, pub metadata: Arc, pub query_port: Arc, - pub query_service: QueryServiceImpl, + pub query_service: Arc, pub ingestion: IngestionServiceImpl, pub flush_service: Arc, _tmpdir: tempfile::TempDir, @@ -106,13 +107,13 @@ impl TestContext { Some(metadata.clone()), )); - let query_service = QueryServiceImpl::new( + let query_service = Arc::new(QueryServiceImpl::new( query_port.clone(), metadata.clone(), wal.clone(), 30, points_sink.clone(), - ); + )); let ingestion = IngestionServiceImpl::new(wal.clone(), metadata.clone(), 100_000, 10_000, 0); @@ -148,13 +149,13 @@ impl TestContext { let query_port: Arc = Arc::new(MockQueryPort); let points_sink: Arc = Arc::new(NoopPointsSink); - let query_service = QueryServiceImpl::new( + let query_service = Arc::new(QueryServiceImpl::new( query_port.clone(), metadata.clone(), wal.clone(), 30, points_sink.clone(), - ); + )); let ingestion = IngestionServiceImpl::new(wal.clone(), metadata.clone(), 100_000, 10_000, 0); @@ -207,4 +208,9 @@ impl TestContext { .execute_query(db, q, None, rp, None) .await } + + pub fn continuous_query_service(&self) -> ContinuousQueryService { + let query: Arc = self.query_service.clone(); + ContinuousQueryService::new(self.metadata.clone(), query, None, 1) + } } diff --git a/hyperbytedb/tests/integration.rs b/hyperbytedb/tests/integration.rs index ba1adf1..07c3f5c 100644 --- a/hyperbytedb/tests/integration.rs +++ b/hyperbytedb/tests/integration.rs @@ -112,6 +112,8 @@ fn setup(dir: &tempfile::TempDir) -> (Arc, Arc) { replicate_body_limit_bytes: 32 * 1024 * 1024, max_points_per_request: 0, request_timeout_secs: 30, + wal_batcher_alive: None, + disk_read_only: None, rate_limiter: None, }); @@ -194,6 +196,8 @@ async fn test_auth_blocks_unauthenticated() { replicate_body_limit_bytes: 32 * 1024 * 1024, max_points_per_request: 0, request_timeout_secs: 30, + wal_batcher_alive: None, + disk_read_only: None, rate_limiter: None, }); @@ -295,6 +299,8 @@ async fn test_cardinality_limit() { replicate_body_limit_bytes: 32 * 1024 * 1024, max_points_per_request: 0, request_timeout_secs: 30, + wal_batcher_alive: None, + disk_read_only: None, rate_limiter: None, }); @@ -441,6 +447,8 @@ async fn test_metrics_endpoint() { replicate_body_limit_bytes: 32 * 1024 * 1024, max_points_per_request: 0, request_timeout_secs: 30, + wal_batcher_alive: None, + disk_read_only: None, rate_limiter: None, }); @@ -776,6 +784,8 @@ async fn test_rate_limiter_refills_and_denies() { max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: Some(Arc::new(EndpointRateLimiters::new(5))), + wal_batcher_alive: None, + disk_read_only: None, }); let (url, _handle) = start_server(app_state).await; @@ -929,6 +939,8 @@ async fn test_cross_database_on_clause_requires_authorization() { replicate_body_limit_bytes: 32 * 1024 * 1024, max_points_per_request: 0, request_timeout_secs: 30, + wal_batcher_alive: None, + disk_read_only: None, rate_limiter: None, }); diff --git a/hyperbytedb/tests/raft_integration.rs b/hyperbytedb/tests/raft_integration.rs index 0d9ec9f..6002a87 100644 --- a/hyperbytedb/tests/raft_integration.rs +++ b/hyperbytedb/tests/raft_integration.rs @@ -149,6 +149,8 @@ async fn start_cluster_node_with_listener( max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: None, + wal_batcher_alive: None, + disk_read_only: None, }); let app = build_router(app_state); @@ -348,6 +350,8 @@ async fn test_cluster_endpoints_without_peers() { max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: None, + wal_batcher_alive: None, + disk_read_only: None, }); let app = build_router(app_state); diff --git a/hyperbytedb/tests/security_auth.rs b/hyperbytedb/tests/security_auth.rs index 7249d83..78c9bcc 100644 --- a/hyperbytedb/tests/security_auth.rs +++ b/hyperbytedb/tests/security_auth.rs @@ -157,6 +157,8 @@ async fn start_auth_cluster_node(dir: &std::path::Path) -> AuthClusterNode { max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: None, + wal_batcher_alive: None, + disk_read_only: None, }); let app = build_router(app_state); @@ -347,3 +349,86 @@ async fn metadata_create_db(url: &str, client: &reqwest::Client) { .unwrap(); assert_eq!(resp.status(), StatusCode::OK); } + +#[tokio::test] +#[serial(chdb)] +async fn grant_revoke_controls_query_access() { + let dir = tempfile::tempdir().unwrap(); + let node = start_auth_cluster_node(dir.path()).await; + let client = reqwest::Client::new(); + + metadata_create_db(&node.url, &client).await; + + let create_other = client + .get(format!("{}/query", node.url)) + .query(&[ + ("q", r#"CREATE DATABASE "otherdb""#), + ("u", "admin"), + ("p", "adminpw"), + ]) + .send() + .await + .unwrap(); + assert_eq!(create_other.status(), StatusCode::OK); + + let seed = client + .post(format!("{}/write", node.url)) + .query(&[("db", "mydb"), ("u", "admin"), ("p", "adminpw")]) + .body("cpu,host=a value=1") + .send() + .await + .unwrap(); + assert_eq!(seed.status(), StatusCode::NO_CONTENT); + + let denied_mydb = client + .get(format!("{}/query", node.url)) + .query(&[ + ("db", "mydb"), + ("q", r#"SELECT * FROM "cpu""#), + ("u", "writer"), + ("p", "writerpw"), + ]) + .send() + .await + .unwrap(); + assert_eq!(denied_mydb.status(), StatusCode::FORBIDDEN); + + let grant = client + .get(format!("{}/query", node.url)) + .query(&[ + ("db", "mydb"), + ("q", r#"GRANT ALL ON "mydb" TO "writer""#), + ("u", "admin"), + ("p", "adminpw"), + ]) + .send() + .await + .unwrap(); + assert_eq!(grant.status(), StatusCode::OK); + + let allowed_mydb = client + .get(format!("{}/query", node.url)) + .query(&[ + ("db", "mydb"), + ("q", r#"SELECT * FROM "cpu""#), + ("u", "writer"), + ("p", "writerpw"), + ]) + .send() + .await + .unwrap(); + assert_eq!(allowed_mydb.status(), StatusCode::OK); + + let denied_otherdb = client + .get(format!("{}/query", node.url)) + .query(&[ + ("db", "otherdb"), + ("q", r#"SHOW MEASUREMENTS"#), + ("u", "writer"), + ("p", "writerpw"), + ]) + .send() + .await + .unwrap(); + assert_eq!(denied_otherdb.status(), StatusCode::FORBIDDEN); +} diff --git a/hyperbytedb/tests/sync_quorum_integration.rs b/hyperbytedb/tests/sync_quorum_integration.rs index c0ea407..1c5810e 100644 --- a/hyperbytedb/tests/sync_quorum_integration.rs +++ b/hyperbytedb/tests/sync_quorum_integration.rs @@ -191,6 +191,8 @@ async fn start_node_on( max_points_per_request: 0, request_timeout_secs: 30, rate_limiter: None, + wal_batcher_alive: None, + disk_read_only: None, }); let app = build_router(app_state);