feat: proxy backend exclusion, materialized view backfill, config defaults

config.toml.example

@@ -69,7 +69,7 @@ replicate_receiver_queue_depth = 1024
 replicate_receiver_workers = 1
 replication_truncate_stale_peer_multiplier = 2
 # Raft consensus settings (used when cluster is enabled)
-# raft_heartbeat_interval_ms = 300
+# raft_heartbeat_interval_ms = 1000
 # raft_election_timeout_ms = 1000
 # raft_snapshot_threshold = 1000
 

deploy/examples/three-node.yaml

@@ -37,7 +37,7 @@ spec:
     heartbeatIntervalSecs: 2
     heartbeatMissThreshold: 5
     replicationMaxRetries: 5
-    raftHeartbeatIntervalMs: 300
+    raftHeartbeatIntervalMs: 1000
     raftElectionTimeoutMs: 1000
     replication:
       mode: async

deploy/kind/kind-config.yaml

@@ -7,13 +7,13 @@ nodes:
     # hyperbytedb-proxy NodePort → 30086 → localhost:8086
     extraPortMappings:
       - containerPort: 30086
-        hostPort: 8086
+        hostPort: 18086
         protocol: TCP
       - containerPort: 30090
-        hostPort: 9090
+        hostPort: 19090
         protocol: TCP
       - containerPort: 30000
-        hostPort: 3000
+        hostPort: 13000
         protocol: TCP
   - role: worker
     extraMounts:

deploy/kind/manifests/hyperbytedb-cr.yaml

@@ -1,6 +1,6 @@
 # HyperbyteDB cluster for kind — tuned for a 4-CPU / high-memory node (16 GiB pod limit).
 # Matches docker-compose tuning: chDB pool = cores, larger flush batches, faster WAL drain.
-apiVersion: hyperbytedb.hyperbytedb.io/v1alpha1
+apiVersion: hyperbytedb.hyperbyte.cloud/v1alpha1
 kind: HyperbytedbCluster
 metadata:
   name: hyperbytedb
@@ -48,6 +48,7 @@ spec:
     raftHeartbeatIntervalMs: 1000
     raftElectionTimeoutMs: 5000
     raftSnapshotThreshold: 1000
+    drainWaitSecs: 30
     replication:
       mode: async
       ackTimeoutMs: 5000

deploy/kind/manifests/operator.yaml

@@ -28,13 +28,13 @@ rules:
 - apiGroups: [policy]
   resources: [poddisruptionbudgets]
   verbs: [get, list, watch, create, update, patch, delete]
-- apiGroups: [hyperbytedb.hyperbytedb.io]
+- apiGroups: [hyperbytedb.hyperbyte.cloud]
   resources: [hyperbytedbclusters, hyperbytedbbackups, hyperbytedbrestores]
   verbs: [get, list, watch, create, update, patch, delete]
-- apiGroups: [hyperbytedb.hyperbytedb.io]
+- apiGroups: [hyperbytedb.hyperbyte.cloud]
   resources: [hyperbytedbclusters/status, hyperbytedbbackups/status, hyperbytedbrestores/status]
   verbs: [get, update, patch]
-- apiGroups: [hyperbytedb.hyperbytedb.io]
+- apiGroups: [hyperbytedb.hyperbyte.cloud]
   resources: [hyperbytedbclusters/finalizers, hyperbytedbbackups/finalizers, hyperbytedbrestores/finalizers]
   verbs: [update]
 - apiGroups: [coordination.k8s.io]

docs/deep-dive/deep-dive-clustering.md

@@ -296,7 +296,7 @@ Raft messages are exchanged via HTTP:
 
 | Parameter | Default | Description |
 |-----------|---------|-------------|
-| `raft_heartbeat_interval_ms` | 300 | Raft heartbeat interval |
+| `raft_heartbeat_interval_ms` | 1000 | Raft heartbeat interval |
 | `raft_election_timeout_ms` | 1000 | Raft election timeout |
 | `raft_snapshot_threshold` | 1000 | Log entries before snapshot |
 
@@ -564,7 +564,7 @@ Step 5: Set node state to Leaving
 | `raft_dir` | `"./raft"` | RocksDB directory for Raft state |
 | `sync_max_concurrent_files` | `4` | Max concurrent file downloads during sync |
 | `replication_max_retries` | `5` | Max retries for failed replications |
-| `raft_heartbeat_interval_ms` | `300` | Raft heartbeat interval (milliseconds) |
+| `raft_heartbeat_interval_ms` | `1000` | Raft heartbeat interval (milliseconds) |
 | `raft_election_timeout_ms` | `1000` | Raft election timeout (milliseconds) |
 | `raft_snapshot_threshold` | `1000` | Log entries before Raft snapshot |
 | `replication.mode` | `"async"` | Coordinator replication mode: `"async"` (default) or `"sync_quorum"` |

docs/user-guide/advanced-features.md

@@ -105,6 +105,17 @@ CREATE MATERIALIZED VIEW "mv_cpu_1h" ON "mydb"
 AS SELECT mean("usage_idle") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), *
 ```
 
+By default, `CREATE` installs the destination tables and ClickHouse materialized views only — rollups begin with the next write to the source. Historical backfill is **disabled by default** because it can scan the entire source measurement and take minutes on large datasets.
+
+To also aggregate existing source history on create, add `WITH BACKFILL`:
+
+```sql
+CREATE MATERIALIZED VIEW "mv_cpu_1h" ON "mydb" WITH BACKFILL
+AS SELECT mean("usage_idle") INTO "cpu_1h" FROM "cpu" GROUP BY time(1h), *
+```
+
+A `WHERE time > ...` clause in the SELECT limits the backfill scan when `WITH BACKFILL` is used.
+
 Requirements (same as `SELECT INTO` / continuous queries):
 
 - `INTO` destination measurement is required
@@ -116,7 +127,7 @@ On `CREATE`, HyperbyteDB:
 1. Registers the destination measurement in metadata
 2. Creates destination MergeTree tables in chDB
 3. Installs fact and series ClickHouse materialized views
-4. Backfills historical data from the source
+4. Backfills historical data from the source **only when `WITH BACKFILL` is specified**
 
 ### Manage materialized views
 
@@ -133,7 +144,7 @@ DROP MATERIALIZED VIEW "mv_cpu_1h" ON "mydb"
 |--|------------------|-------------------|
 | Trigger | 10s scheduler | Each flush to source |
 | Latency | Up to resample interval | Near real-time |
-| Backfill | Re-scans window each run | One-time on CREATE; then incremental |
+| Backfill | Re-scans window each run | Opt-in on CREATE (`WITH BACKFILL`); then incremental |
 | Engine | WAL writeback | ClickHouse MV |
 
 ---

docs/user-guide/configuration.md

@@ -130,7 +130,7 @@ Master-master peer-to-peer clustering with Raft consensus for schema mutations.
 | `replicate_receiver_queue_depth` | integer | `1024` | Bounded apply queue on the replicate receiver |
 | `replicate_receiver_workers` | integer | `1` | **Ignored.** Receiver uses a single ordered worker |
 | `replication_truncate_stale_peer_multiplier` | integer | `2` | When >0, peers with ack 0 and stale heartbeats are omitted from truncate barrier (× heartbeat interval) |
-| `raft_heartbeat_interval_ms` | — | *unset* | Optional Raft heartbeat (ms); uses internal default if omitted |
+| `raft_heartbeat_interval_ms` | — | *unset* | Optional Raft heartbeat (ms); defaults to 1000 if omitted |
 | `raft_election_timeout_ms` | — | *unset* | Optional Raft election timeout (ms) |
 | `raft_snapshot_threshold` | — | *unset* | Optional log entries before Raft snapshot |
 

docs/user-guide/operator/cluster.md

@@ -86,7 +86,7 @@ spec:
     heartbeatIntervalSecs: 1
     heartbeatMissThreshold: 3
     replicationMaxRetries: 10
-    raftHeartbeatIntervalMs: 200
+    raftHeartbeatIntervalMs: 1000
     raftElectionTimeoutMs: 800
     raftSnapshotThreshold: 500
     replication:
@@ -306,7 +306,7 @@ Tuning parameters for multi-replica cluster behavior. These only take effect whe
 | `replicationMaxCoalesceBodyBytes` | int64 | `8388608` | Max bytes for coalescing consecutive WAL batches |
 | `replicateReceiverQueueDepth` | int32 | `1024` | Bounded apply queue on the replicate receiver |
 | `replicationTruncateStalePeerMultiplier` | int64 | `2` | Omit stale peers from WAL truncate barrier |
-| `raftHeartbeatIntervalMs` | int32 | `300` | Raft leader heartbeat interval |
+| `raftHeartbeatIntervalMs` | int32 | `1000` | Raft leader heartbeat interval |
 | `raftElectionTimeoutMs` | int32 | `1000` | Raft election timeout |
 | `raftSnapshotThreshold` | int32 | `1000` | Log entries before Raft snapshot |
 | `replication.mode` | string | `async` | `async` or `sync_quorum` |

hyperbytedb-proxy/Dockerfile

@@ -43,6 +43,7 @@ RUN mkdir -p hyperbytedb/benches hyperbytedb/benches/support \
     && echo "fn main(){}" > hyperbytedb/benches/ingestion_line_protocol.rs \
     && echo "fn main(){}" > hyperbytedb/benches/query_fixed_dataset.rs \
     && echo "fn main(){}" > hyperbytedb/benches/flush_service.rs \
+    && echo "fn main(){}" > hyperbytedb/benches/ingestion_prepared.rs \
     && echo "" > hyperbytedb/benches/support/mod.rs \
     && echo "fn main(){}" > hyperbytedb-proxy/benches/routing.rs \
     && echo "" > hyperbytedb-proxy/benches/support/mod.rs

hyperbytedb-proxy/src/admin.rs

@@ -2,17 +2,18 @@
 //! itself.
 //!
 //! - `GET /healthz`         — liveness, always 200 once the process is up.
-//! - `GET /readyz`          — readiness, 200 only when ≥1 backend is `Active`.
+//! - `GET /readyz`          — readiness, 200 only when ≥1 backend is routable (Active and not excluded).
 //! - `GET /metrics`         — Prometheus exposition.
 //! - `GET /admin/backends`  — JSON dump of the current pool, for debugging.
 //!
 //! Routes are chosen so they can be allowlisted before the catch-all proxy
 //! handler, with no risk of colliding with a hyperbytedb path.
 
+use std::net::IpAddr;
 use std::sync::Arc;
 
 use axum::Json;
-use axum::extract::State;
+use axum::extract::{Path, State};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use serde::Serialize;
@@ -32,7 +33,10 @@ pub async fn healthz() -> Response {
 
 pub async fn readyz(State(state): State<AdminState>) -> Response {
     let snap = state.pool.snapshot().await;
-    let active = snap.iter().filter(|b| b.health() == Health::Active).count();
+    let active = snap
+        .iter()
+        .filter(|b| b.health() == Health::Active && !b.is_excluded())
+        .count();
     if active > 0 {
         (
             StatusCode::OK,
@@ -87,3 +91,68 @@ pub async fn list_backends(State(state): State<AdminState>) -> Response {
         .collect();
     Json(body).into_response()
 }
+
+// ---------------------------------------------------------------------------
+// Operator-driven backend exclusion endpoints
+// ---------------------------------------------------------------------------
+
+#[derive(Serialize)]
+struct ExcludeResponse {
+    status: &'static str,
+    ip: String,
+}
+
+/// `POST /admin/backends/{ip}/exclude` — Tell the proxy to stop routing to
+/// this backend. Called by the operator before killing a pod during rolling
+/// upgrades.
+pub async fn exclude_backend(State(state): State<AdminState>, Path(ip): Path<IpAddr>) -> Response {
+    match state.pool.exclude_backend(ip).await {
+        Ok(true) => (
+            StatusCode::OK,
+            Json(ExcludeResponse {
+                status: "excluded",
+                ip: ip.to_string(),
+            }),
+        )
+            .into_response(),
+        Ok(false) => (
+            StatusCode::CONFLICT,
+            Json(ExcludeResponse {
+                status: "already_excluded",
+                ip: ip.to_string(),
+            }),
+        )
+            .into_response(),
+        Err(e) => (
+            StatusCode::NOT_FOUND,
+            Json(serde_json::json!({"status": "error", "message": e.to_string()})),
+        )
+            .into_response(),
+    }
+}
+
+/// `POST /admin/backends/{ip}/include` — Clear the exclusion flag so the
+/// proxy may route to this backend again. Called by the operator after the
+/// replacement pod is healthy.
+pub async fn include_backend(State(state): State<AdminState>, Path(ip): Path<IpAddr>) -> Response {
+    let was_excluded = state.pool.include_backend(ip).await;
+    let status = if was_excluded {
+        "included"
+    } else {
+        "not_excluded"
+    };
+    (
+        StatusCode::OK,
+        Json(ExcludeResponse {
+            status,
+            ip: ip.to_string(),
+        }),
+    )
+        .into_response()
+}
+
+/// `GET /admin/pool` — Full pool status including exclusion flags.
+pub async fn pool_status(State(state): State<AdminState>) -> Response {
+    let statuses = state.pool.pool_status().await;
+    Json(statuses).into_response()
+}

hyperbytedb-proxy/src/backend.rs

@@ -2,7 +2,7 @@
 //! counters are atomic so the routing hot path is lock-free.
 
 use std::net::IpAddr;
-use std::sync::atomic::{AtomicI64, AtomicU8, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicUsize, Ordering};
 use std::time::SystemTime;
 
 /// Health classification observed by the most recent probe.
@@ -62,6 +62,10 @@ pub struct Backend {
     /// Consecutive probe failures. Used to log transitions cleanly and (in
     /// future) to back-off probes for backends that are deeply broken.
     consecutive_failures: AtomicUsize,
+    /// Operator-driven exclusion flag. When true, `pick_active` skips this
+    /// backend entirely — set by the operator before killing a pod during
+    /// rolling upgrades.
+    excluded: AtomicBool,
 }
 
 impl Backend {
@@ -75,6 +79,7 @@ impl Backend {
             inflight: AtomicUsize::new(0),
             last_probe_unix: AtomicI64::new(0),
             consecutive_failures: AtomicUsize::new(0),
+            excluded: AtomicBool::new(false),
         }
     }
 
@@ -110,6 +115,15 @@ impl Backend {
         self.inflight.load(Ordering::Relaxed)
     }
 
+    pub fn is_excluded(&self) -> bool {
+        self.excluded.load(Ordering::Acquire)
+    }
+
+    /// Set the operator-driven exclusion flag. Returns the previous value.
+    pub fn set_excluded(&self, val: bool) -> bool {
+        self.excluded.swap(val, Ordering::AcqRel)
+    }
+
     /// RAII guard that increments `inflight` on construction and decrements
     /// on drop, even on panic. Use for the lifetime of one proxied request.
     pub fn enter(self: &std::sync::Arc<Self>) -> InflightGuard {

hyperbytedb-proxy/src/lib.rs

@@ -14,7 +14,7 @@ use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use axum::Router;
-use axum::routing::{any, get};
+use axum::routing::{any, get, post};
 use metrics_exporter_prometheus::PrometheusBuilder;
 use tokio::net::TcpListener;
 use tokio::signal::unix::{SignalKind, signal};
@@ -60,6 +60,9 @@ pub async fn run() -> Result<()> {
         .route("/readyz", get(admin::readyz))
         .route("/metrics", get(admin::metrics_endpoint))
         .route("/admin/backends", get(admin::list_backends))
+        .route("/admin/backends/{ip}/exclude", post(admin::exclude_backend))
+        .route("/admin/backends/{ip}/include", post(admin::include_backend))
+        .route("/admin/pool", get(admin::pool_status))
         .with_state(admin_state);
 
     // Order matters: admin routes first, then the catch-all proxy fallback.