feat(security): auth-gate cluster routes, enforce GRANT/REVOKE, cap decompression, redact credentials

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

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.

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:**

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

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,

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<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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(),
     }
 }

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<Arc<rate_limit::EndpointRateLimiters>>,
+    /// Set when WAL group-commit is enabled; `false` after the writer task exits.
+    pub wal_batcher_alive: Option<Arc<AtomicBool>>,
+    /// Set when `[disk] enabled`; `true` when free space is below readonly threshold.
+    pub disk_read_only: Option<Arc<AtomicBool>>,
 }
 
 pub fn build_router(state: Arc<AppState>) -> Router {

hyperbytedb/src/adapters/http/write.rs

@@ -36,6 +36,16 @@ pub async fn handle_write(
     Query(params): Query<WriteParams>,
     body: axum::body::Bytes,
 ) -> Result<Response, HyperbytedbError> {
+    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;

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<BatchRequest>,
     enqueue_timeout: Option<Duration>,
-    _writer: JoinHandle<()>,
+    writer_alive: Arc<AtomicBool>,
     inner: Arc<RocksDbWal>,
 }
 
@@ -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<AtomicBool> {
+        self.writer_alive.clone()
+    }
+
+    async fn supervise_writer(handle: JoinHandle<()>, writer_alive: Arc<AtomicBool>) {
+        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<BatchRequest>,
         wal: Arc<RocksDbWal>,
@@ -142,6 +162,12 @@ impl BatchingWal {
     }
 
     async fn enqueue(&self, bundle: WalAppendBundle) -> Result<u64, HyperbytedbError> {
+        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)
+        );
+    }
 }

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<chrono::Utc>,
+    ) -> Result<(), HyperbytedbError> {
+        self.run_pending_queries(now).await
+    }
+
+    async fn run_pending_queries(
+        &self,
+        now: chrono::DateTime<chrono::Utc>,
+    ) -> 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");