diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index eef29f1..27c4e00 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -40,6 +40,19 @@ pub struct MasterConfig { #[arg(long, env = "TARGET_ROWS_PER_FRAGMENT", default_value_t = 1_048_576)] pub target_rows_per_fragment: usize, + /// Interval, in seconds, between automatic WAL-merge sweeps. Each sweep + /// enqueues a `MergeWal` task for every experiment whose pending MemWAL + /// generation count crosses `merge_wal_min_generations`; the task fans out + /// to the configured worker endpoints. `0` disables automatic WAL merge + /// (manual triggers still work). + #[arg(long, env = "MERGE_WAL_INTERVAL_SECS", default_value_t = 600)] + pub merge_wal_interval_secs: u64, + + /// Minimum pending MemWAL generations (per experiment) before an automatic + /// `MergeWal` is enqueued. Read from the periodically-scanned stats table. + #[arg(long, env = "MERGE_WAL_MIN_GENERATIONS", default_value_t = 8)] + pub merge_wal_min_generations: i64, + /// Data-plane worker base URLs (comma-separated), e.g. /// `http://rollout-0:3000,http://rollout-1:3000`. A `MergeWal` task fans out /// to every endpoint so each worker merges its own MemWAL shard (the master diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 1713f4f..d79b960 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -438,6 +438,8 @@ mod tests { compaction_interval_secs: 0, min_fragments: 16, target_rows_per_fragment: 1_048_576, + merge_wal_interval_secs: 0, + merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, etcd_endpoints: test_etcd_endpoints(), diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 7181662..319f39c 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -54,11 +54,12 @@ fn kind_label(kind: TaskKind) -> &'static str { } } -/// Enqueue a task and return its record. For [`TaskKind::Compact`] and -/// [`TaskKind::IndexId`] this de-dupes against an existing non-terminal task for -/// the same target: if one is already `Queued` or `Running`, its record is -/// returned unchanged and nothing new is enqueued. `MergeWal` tasks are not -/// de-duped (each fan-out is independent). +/// Enqueue a task and return its record. For [`TaskKind::Compact`], +/// [`TaskKind::IndexId`], and depless [`TaskKind::MergeWal`] this de-dupes +/// against an existing non-terminal task for the same target: if one is already +/// `Queued` or `Running`, its record is returned unchanged and nothing new is +/// enqueued. A `MergeWal` that is part of a dependency chain (non-empty +/// `depends_on`) is not de-duped. pub async fn enqueue( state: &Arc, kind: TaskKind, @@ -305,9 +306,44 @@ fn in_quiet_hours(config: &CompactionConfig) -> bool { .any(|(start, end)| hour >= *start && hour < *end) } +/// Enqueue a `MergeWal` task for every experiment whose pending MemWAL +/// generation count is at or above the configured threshold, reading candidates +/// from the stats table. Coordinated across master replicas by a dedicated +/// task-store lock so only one replica sweeps at a time. Depless `MergeWal` +/// enqueues de-dupe, so a still-running fan-out is not re-queued. +pub async fn sweep_merge_wal_candidates(state: &Arc) -> lance::Result { + let Some(guard) = state + .task_store + .try_coordination_lock("merge-wal-sweep") + .await? + else { + return Ok(0); + }; + let result = sweep_merge_wal_inner(state).await; + let release = state.task_store.release_coordination_lock(guard).await; + match (result, release) { + (Ok(count), Ok(())) => Ok(count), + (Err(error), _) => Err(error), + (Ok(_), Err(error)) => Err(error), + } +} + +async fn sweep_merge_wal_inner(state: &Arc) -> lance::Result { + let threshold = state.config.merge_wal_min_generations; + let rows = state.stats.lock().await.list(None, usize::MAX, 0).await?; + let mut queued = 0; + for row in rows { + if row.pending_wal_generations >= threshold { + enqueue(state, TaskKind::MergeWal, &row.name).await?; + queued += 1; + } + } + Ok(queued) +} + /// Spawn the scheduler poller plus the optional periodic auto-sweep. pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { - // Optional periodic auto-sweep feeds the same queue. + // Optional periodic compaction auto-sweep feeds the same queue. let interval_secs = state.config.compaction_interval_secs; if interval_secs > 0 { let sweep_state = state.clone(); @@ -325,6 +361,26 @@ pub fn spawn_scheduler(state: &Arc) -> JoinHandle<()> { }); } + // Optional periodic WAL-merge auto-sweep, independent of compaction. + let merge_interval_secs = state.config.merge_wal_interval_secs; + if merge_interval_secs > 0 { + let sweep_state = state.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(Duration::from_secs(merge_interval_secs)); + ticker.tick().await; // skip immediate tick + loop { + ticker.tick().await; + match sweep_merge_wal_candidates(&sweep_state).await { + Ok(n) if n > 0 => { + tracing::info!(queued = n, "auto merge-wal sweep queued experiments") + } + Ok(_) => {} + Err(e) => tracing::warn!(error = %e, "auto merge-wal sweep failed"), + } + } + }); + } + let concurrency = state.config.task_concurrency.max(1); let sem = Arc::new(Semaphore::new(concurrency)); let dispatch_state = state.clone(); @@ -378,6 +434,8 @@ mod tests { // Low threshold so a handful of appends crosses it. min_fragments: 2, target_rows_per_fragment: 1_048_576, + merge_wal_interval_secs: 0, + merge_wal_min_generations: 2, worker_endpoints: vec![], task_concurrency: 4, etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS") @@ -636,6 +694,60 @@ mod tests { worker.abort(); } + /// The WAL-merge sweep enqueues a `MergeWal` only for experiments whose + /// pending generation count is at or above the threshold, and de-dupes so a + /// second sweep does not pile up a duplicate for the same target. + #[tokio::test] + #[ignore = "requires ETCD_TEST_ENDPOINTS"] + async fn sweep_merge_wal_enqueues_over_threshold_and_dedupes() { + use crate::stats_store::StatRow; + + let dir = TempDir::new().unwrap(); + let mut cfg = config(&dir); + cfg.merge_wal_min_generations = 3; + let state = MasterState::new(cfg).await.unwrap(); + + let seed = |name: &str, pending: i64| StatRow { + name: name.to_string(), + uri: state.rollout_uri(name), + row_count: 0, + fragment_count: 0, + last_updated: 0, + pending_wal_generations: pending, + last_compaction: StatRow::NO_COMPACTION, + total_compactions: 0, + scanned_at: 0, + }; + { + let mut stats = state.stats.lock().await; + stats.upsert(&seed("hot", 5)).await.unwrap(); // >= threshold + stats.upsert(&seed("cold", 1)).await.unwrap(); // < threshold + } + + let queued = sweep_merge_wal_candidates(&state).await.unwrap(); + assert_eq!(queued, 1, "only the over-threshold experiment is swept"); + + let tasks = state.task_store.list().await.unwrap(); + let merge_tasks: Vec<_> = tasks + .iter() + .filter(|t| t.kind == TaskKind::MergeWal) + .collect(); + assert_eq!(merge_tasks.len(), 1); + assert_eq!(merge_tasks[0].target, "hot"); + + // Second sweep must de-dupe against the still-queued MergeWal. + sweep_merge_wal_candidates(&state).await.unwrap(); + let merge_after = state + .task_store + .list() + .await + .unwrap() + .into_iter() + .filter(|t| t.kind == TaskKind::MergeWal) + .count(); + assert_eq!(merge_after, 1, "duplicate MergeWal is de-duped"); + } + /// Minimal rollout record builder for tests (the core struct has no /// `Default`). pub fn rollout_record(id: &str) -> lance_context_core::RolloutRecord { use chrono::TimeZone; diff --git a/crates/lance-context-master/src/state.rs b/crates/lance-context-master/src/state.rs index 8eab409..8d43bbb 100644 --- a/crates/lance-context-master/src/state.rs +++ b/crates/lance-context-master/src/state.rs @@ -93,6 +93,8 @@ mod tests { compaction_interval_secs: 0, min_fragments: 16, target_rows_per_fragment: 1_048_576, + merge_wal_interval_secs: 0, + merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, etcd_endpoints: std::env::var("ETCD_TEST_ENDPOINTS") diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 5479047..cfeeb29 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -86,9 +86,10 @@ impl TaskStore { Ok(store) } - /// Atomically enqueue a task. Standalone Compact and IndexId tasks use an - /// etcd dedupe key while queued or running. Tasks with dependencies are - /// always distinct because they belong to a specific ordered chain. + /// Atomically enqueue a task. Standalone Compact, IndexId, and depless + /// MergeWal tasks use an etcd dedupe key while queued or running. Tasks with + /// dependencies are always distinct because they belong to a specific + /// ordered chain. pub async fn enqueue( &self, kind: TaskKind, @@ -806,7 +807,15 @@ fn fail_dependency(task: &mut TaskRecord, dependency: &str) { } fn should_dedupe(kind: TaskKind, depends_on: &[String]) -> bool { - depends_on.is_empty() && matches!(kind, TaskKind::Compact | TaskKind::IndexId) + // MergeWal is deduped for depless enqueues (periodic auto-sweep and the + // manual "Merge WAL" button) so a slow fan-out cannot pile up duplicate + // broadcasts for the same experiment. A MergeWal that is part of an ordered + // Optimize chain carries `depends_on` and is intentionally not deduped. + depends_on.is_empty() + && matches!( + kind, + TaskKind::Compact | TaskKind::IndexId | TaskKind::MergeWal + ) } fn requires_target_lock(kind: TaskKind) -> bool { @@ -874,6 +883,8 @@ mod tests { compaction_interval_secs: 0, min_fragments: 16, target_rows_per_fragment: 1_048_576, + merge_wal_interval_secs: 0, + merge_wal_min_generations: 8, worker_endpoints: vec![], task_concurrency: 4, etcd_endpoints: vec![], diff --git a/deploy/kubernetes/README.md b/deploy/kubernetes/README.md index dc37c25..416772c 100644 --- a/deploy/kubernetes/README.md +++ b/deploy/kubernetes/README.md @@ -30,3 +30,17 @@ remain idempotent across crash recovery. The `_stats.rollout.lance` table remains in `DATA_DIR`. etcd coordinates its single-writer sections, while readers reload the latest Lance manifest so every master replica sees current stats. + +## Automatic maintenance + +Two periodic sweeps run on the master and feed the shared scheduler queue: + +- **Compaction** (`COMPACTION_INTERVAL_SECS`, `MIN_FRAGMENTS`) rewrites an + experiment's base-table fragments locally on the master. +- **WAL merge** (`MERGE_WAL_INTERVAL_SECS`, `MERGE_WAL_MIN_GENERATIONS`) enqueues + a `MergeWal` task for every experiment whose pending MemWAL generation count + (from the periodically-scanned stats table) crosses the threshold. The task + fans out to every `WORKER_ENDPOINTS` worker, each of which folds its own shard. + The master cannot merge a shard it does not own without fencing the live + writer, so the merge itself always runs on the owning worker. Set the interval + to `0` to disable it; the manual "Merge WAL" / "Optimize" UI actions still work. diff --git a/deploy/kubernetes/master.yaml b/deploy/kubernetes/master.yaml index a88ba52..8a9c3b3 100644 --- a/deploy/kubernetes/master.yaml +++ b/deploy/kubernetes/master.yaml @@ -60,6 +60,13 @@ spec: value: /etc/lance-context/etcd/tls.key - name: TASK_HISTORY_LIMIT value: "1000" + # Master periodically fans a WAL merge out to every worker for any + # experiment with at least this many pending MemWAL generations. + # Set MERGE_WAL_INTERVAL_SECS to 0 to disable auto WAL merge. + - name: MERGE_WAL_INTERVAL_SECS + value: "600" + - name: MERGE_WAL_MIN_GENERATIONS + value: "8" - name: MASTER_PORT value: "8090" - name: UI_DIR