refactor(master): drop RocksDB backend, etcd-only scheduler

crates/lance-context-master/Cargo.toml

@@ -26,7 +26,6 @@ futures = "0.3"
 lance = "7.0.0"
 metrics = "0.24"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
-rocksdb = { version = "0.24", default-features = false }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "sync"] }

crates/lance-context-master/src/config.rs

@@ -1,15 +1,6 @@
 //! Command-line / environment configuration for the master control-plane.
 
-use clap::{Parser, ValueEnum};
-
-/// Durable scheduler state backend.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
-pub enum TaskStoreBackend {
-    /// Single-master RocksDB on a dedicated local PVC.
-    Rocksdb,
-    /// Shared etcd state with lease-based task claims for HA masters.
-    Etcd,
-}
+use clap::Parser;
 
 /// Control-plane (master) process for lance-context rollout stores.
 #[derive(Debug, Clone, Parser)]
@@ -65,29 +56,9 @@ pub struct MasterConfig {
     #[arg(long, env = "TASK_CONCURRENCY", default_value_t = 4)]
     pub task_concurrency: usize,
 
-    /// Scheduler state backend. `rocksdb` requires exactly one master and a
-    /// dedicated PVC; `etcd` supports multiple stateless master replicas.
-    #[arg(
-        long,
-        env = "TASK_STORE_BACKEND",
-        value_enum,
-        default_value_t = TaskStoreBackend::Rocksdb
-    )]
-    pub task_store_backend: TaskStoreBackend,
-
-    /// Local RocksDB directory for durable scheduler task state. This must be
-    /// backed by persistent local storage in production; it must not be an
-    /// object-store URI or a volume shared by multiple master processes. Used
-    /// only when `TASK_STORE_BACKEND=rocksdb`.
-    #[arg(
-        long,
-        env = "TASK_DB_PATH",
-        default_value = "./master-data/tasks.rocksdb"
-    )]
-    pub task_db_path: String,
-
-    /// Comma-separated etcd v3 endpoints. Required when
-    /// `TASK_STORE_BACKEND=etcd`.
+    /// Comma-separated etcd v3 endpoints. Scheduler state (task queue,
+    /// lease-based claims, per-experiment write locks) lives in etcd so several
+    /// stateless master replicas can share one queue. Required.
     #[arg(long, env = "ETCD_ENDPOINTS", value_delimiter = ',')]
     pub etcd_endpoints: Vec<String>,
 
@@ -120,7 +91,7 @@ pub struct MasterConfig {
     #[arg(long, env = "ETCD_LEASE_TTL_SECS", default_value_t = 30)]
     pub etcd_lease_ttl_secs: i64,
 
-    /// Maximum number of terminal scheduler tasks retained in RocksDB and the
+    /// Maximum number of terminal scheduler tasks retained in etcd and the
     /// queue UI. Queued and running tasks are never pruned, and at least one
     /// terminal task is retained for status polling.
     #[arg(long, env = "TASK_HISTORY_LIMIT", default_value_t = 1_000)]

crates/lance-context-master/src/routes.rs

@@ -422,9 +422,9 @@ fn blob_filename(record: &lance_context_core::RolloutRecord) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::{MasterConfig, TaskStoreBackend};
+    use crate::config::MasterConfig;
     use chrono::Utc;
-    use lance_context_core::{RolloutRecord, RolloutRegistry, RolloutStore};
+    use lance_context_core::{generate_id, RolloutRecord, RolloutRegistry, RolloutStore};
     use serde_json::json;
     use tempfile::TempDir;
 
@@ -440,25 +440,25 @@ mod tests {
             target_rows_per_fragment: 1_048_576,
             worker_endpoints: vec![],
             task_concurrency: 4,
-            task_store_backend: TaskStoreBackend::Rocksdb,
-            task_db_path: dir
-                .path()
-                .join("master-tasks")
-                .to_string_lossy()
-                .to_string(),
-            etcd_endpoints: vec![],
-            etcd_prefix: "/test".to_string(),
+            etcd_endpoints: test_etcd_endpoints(),
+            etcd_prefix: format!("/lance-context/test/{}", generate_id()),
             etcd_username: None,
             etcd_password: None,
             etcd_ca_cert: None,
             etcd_client_cert: None,
             etcd_client_key: None,
-            etcd_lease_ttl_secs: 30,
+            etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
             ui_dir: None,
         }
     }
 
+    fn test_etcd_endpoints() -> Vec<String> {
+        std::env::var("ETCD_TEST_ENDPOINTS")
+            .map(|value| value.split(',').map(str::to_string).collect())
+            .unwrap_or_default()
+    }
+
     fn test_record(id: &str, with_blob: bool) -> RolloutRecord {
         let bytes = with_blob.then(|| b"master-blob".to_vec());
         RolloutRecord {
@@ -499,6 +499,7 @@ mod tests {
     /// Create N experiments via core, register them, scan, and assert the stats
     /// table + list endpoint reflect them.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn scan_populates_and_list_returns_experiments() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -550,6 +551,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn scan_sees_registry_commits_from_another_handle() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -570,6 +572,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn scan_reconciles_removed_experiments() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -594,6 +597,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn get_experiment_not_found() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -607,6 +611,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn fresh_detail_refreshes_only_requested_experiment() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -650,6 +655,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn records_endpoint_filters_pages_and_rejects_unknown_experiment() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();
@@ -716,6 +722,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn blob_endpoint_sets_download_headers_and_returns_bytes() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(test_config(&dir)).await.unwrap();

crates/lance-context-master/src/scheduler.rs

@@ -1,8 +1,8 @@
 //! Unified task scheduler.
 //!
-//! Each master polls one durable queue with **bounded concurrency**. RocksDB
-//! supports one master; etcd uses atomic lease-backed claims so multiple
-//! stateless masters can drain the same queue. It executes three kinds of task
+//! Each master polls one durable queue with **bounded concurrency**. The queue
+//! lives in etcd, which uses atomic lease-backed claims so multiple stateless
+//! masters can drain the same queue. It executes three kinds of task
 //! ([`TaskKind`]):
 //!
 //! - **Compact** — rewrites an experiment's base-table fragments. Two `Rewrite`s
@@ -362,8 +362,9 @@ pub fn spawn_scheduler(state: &Arc<MasterState>) -> JoinHandle<()> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::{MasterConfig, TaskStoreBackend};
+    use crate::config::MasterConfig;
     use lance_context_api::TaskState;
+    use lance_context_core::generate_id;
     use tempfile::TempDir;
 
     fn config(dir: &TempDir) -> MasterConfig {
@@ -379,20 +380,16 @@ mod tests {
             target_rows_per_fragment: 1_048_576,
             worker_endpoints: vec![],
             task_concurrency: 4,
-            task_store_backend: TaskStoreBackend::Rocksdb,
-            task_db_path: dir
-                .path()
-                .join("master-tasks")
-                .to_string_lossy()
-                .to_string(),
-            etcd_endpoints: vec![],
-            etcd_prefix: "/test".to_string(),
+            etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS")
+                .map(|value| value.split(',').map(str::to_string).collect())
+                .unwrap_or_default(),
+            etcd_prefix: format!("/lance-context/test/{}", generate_id()),
             etcd_username: None,
             etcd_password: None,
             etcd_ca_cert: None,
             etcd_client_cert: None,
             etcd_client_key: None,
-            etcd_lease_ttl_secs: 30,
+            etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
             ui_dir: None,
         }
@@ -414,6 +411,7 @@ mod tests {
     /// Manual enqueue -> dispatcher compacts -> task reaches Done and the stats
     /// table records a compaction.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn manual_compaction_runs_and_updates_stats() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();
@@ -466,6 +464,7 @@ mod tests {
     /// Manual enqueue of an `IndexId` task -> dispatcher builds the ZoneMap
     /// index -> task reaches Done with the expected detail summary.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn index_id_task_builds_index_and_reaches_done() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();
@@ -499,6 +498,7 @@ mod tests {
 
     /// Enqueuing the same experiment twice while queued de-dupes to one task.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn enqueue_dedupes_queued_compactions() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();
@@ -511,6 +511,7 @@ mod tests {
 
     /// A MergeWal task with no configured endpoints fails fast with a clear msg.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn merge_wal_without_endpoints_fails() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();
@@ -525,6 +526,7 @@ mod tests {
     /// MergeWal fans out to every configured worker endpoint and sums the
     /// reclaimed counts. Uses a tiny in-process stub server per "worker".
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn merge_wal_broadcasts_and_sums_reclaimed() {
         use axum::{routing::post, Json, Router};
 
@@ -561,6 +563,7 @@ mod tests {
     /// `index_id` depending on a `compact` must start after compaction finishes,
     /// so the two never contend for the shared per-experiment base-table gate.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn dependent_task_runs_after_dependency_done() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();
@@ -612,6 +615,7 @@ mod tests {
     /// A dependent whose dependency `Failed` is skipped (marked `Failed`) rather
     /// than run.
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn dependent_skipped_when_dependency_fails() {
         let dir = TempDir::new().unwrap();
         let state = MasterState::new(config(&dir)).await.unwrap();

crates/lance-context-master/src/state.rs

@@ -26,8 +26,8 @@ pub struct MasterState {
     pub base_uri: String,
     /// Effective configuration.
     pub config: MasterConfig,
-    /// Durable scheduler queue. RocksDB is single-master; etcd provides shared
-    /// CAS/lease semantics for stateless HA master replicas.
+    /// Durable scheduler queue in etcd, providing shared CAS/lease semantics
+    /// for stateless HA master replicas.
     pub task_store: TaskStore,
     /// Shared HTTP client for fanning `MergeWal` tasks out to worker endpoints.
     pub http: reqwest::Client,
@@ -79,9 +79,8 @@ impl MasterState {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::TaskStoreBackend;
     use lance_context_api::{TaskKind, TaskState};
-    use lance_context_core::RolloutStore;
+    use lance_context_core::{generate_id, RolloutStore};
     use tempfile::TempDir;
 
     fn test_config(dir: &TempDir) -> MasterConfig {
@@ -96,26 +95,23 @@ mod tests {
             target_rows_per_fragment: 1_048_576,
             worker_endpoints: vec![],
             task_concurrency: 4,
-            task_store_backend: TaskStoreBackend::Rocksdb,
-            task_db_path: dir
-                .path()
-                .join("master-tasks")
-                .to_string_lossy()
-                .to_string(),
-            etcd_endpoints: vec![],
-            etcd_prefix: "/test".to_string(),
+            etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS")
+                .map(|value| value.split(',').map(str::to_string).collect())
+                .unwrap_or_default(),
+            etcd_prefix: format!("/lance-context/test/{}", generate_id()),
             etcd_username: None,
             etcd_password: None,
             etcd_ca_cert: None,
             etcd_client_cert: None,
             etcd_client_key: None,
-            etcd_lease_ttl_secs: 30,
+            etcd_lease_ttl_secs: 5,
             task_history_limit: 1_000,
             ui_dir: None,
         }
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn startup_backfills_legacy_rollout_datasets() {
         let dir = TempDir::new().unwrap();
         let uri = dir.path().join("legacy.rollout.lance");
@@ -130,6 +126,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[ignore = "requires ETCD_TEST_ENDPOINTS"]
     async fn startup_requeues_interrupted_local_tasks() {
         let dir = TempDir::new().unwrap();
         let config = test_config(&dir);