From 4bfdd8146c1f475b9c27f8316c04ea3508735b25 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Tue, 22 Sep 2026 20:48:29 +0000 Subject: [PATCH] feat(core): sample and warn on MemWAL pending generations per shard (#259) Every LSM read (wal_shard_snapshots) now records each shard's flushed_generations.len() into a rollout_wal_pending_generations histogram and warns once per shard per read when it reaches ROLLOUT_WAL_PENDING_WARN_GENERATIONS (default 256; 0 disables). Every pending generation is a separate dataset the read path must open, so this count is the read-amplification signal. One generic store reached ~12.6k pending generations across 20 shards before anyone noticed, by which point every read exhausted a 32 GiB worker; nothing in logs or metrics pointed at the store until the crash logs were read by hand. The warn carries the dataset URI and shard id on the span so it is grep-able the way the oversized-blob warn is. No behaviour change. The threshold is plumbed through every store kind's options and the server config, and the metrics crate gives the histogram count-scale buckets (8..4096) instead of the duration defaults. Co-Authored-By: Claude Fable 5 --- .../lance-context-core/src/datagen_store.rs | 6 ++ .../lance-context-core/src/generic_store.rs | 55 +++++++++++++++++++ crates/lance-context-core/src/metrics.rs | 31 ++++++++++- .../lance-context-core/src/rollout_store.rs | 27 +++++++++ crates/lance-context-core/src/store.rs | 8 +++ crates/lance-context-core/src/store_base.rs | 41 +++++++++++++- crates/lance-context-metrics/src/lib.rs | 15 +++++ crates/lance-context-server/src/config.rs | 13 +++++ .../src/routes/datagen.rs | 1 + .../src/routes/generic.rs | 1 + .../src/routes/rollouts.rs | 1 + crates/lance-context-server/src/state.rs | 7 +++ crates/lance-context/src/unified_datagen.rs | 1 + crates/lance-context/src/unified_generic.rs | 2 + 14 files changed, 207 insertions(+), 2 deletions(-) diff --git a/crates/lance-context-core/src/datagen_store.rs b/crates/lance-context-core/src/datagen_store.rs index 37b289bc..aa46a468 100644 --- a/crates/lance-context-core/src/datagen_store.rs +++ b/crates/lance-context-core/src/datagen_store.rs @@ -59,6 +59,11 @@ pub struct DatagenStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Periodically merge this writer's pending generations. `None` or zero /// disables the timer. pub cleanup_interval_secs: Option, @@ -103,6 +108,7 @@ impl DatagenStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: None, schema: Arc::new(datagen_log_schema()), // Datagen keys on `event_id`, not `id`: event ids are derived diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 873df0ec..5a202a6b 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -71,6 +71,12 @@ pub struct GenericStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. Every pending generation is a + /// separate dataset a read must open, so this is the read-amplification + /// alarm. `None` uses the crate default (256); `Some(0)` disables the warn. + /// The `rollout_wal_pending_generations` histogram is emitted regardless. + pub pending_generations_warn: Option, /// Shared, capacity-bounded Lance session. pub session: Option>, /// Whether [`GenericStore::add`] seals before returning, making the rows it @@ -174,6 +180,7 @@ impl GenericStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: options.session, schema: create_schema, // Always `id`: the LSM merge key, which `SchemaSpec::validate` @@ -794,6 +801,54 @@ mod tests { }); } + /// Every LSM read samples each shard's pending-generation count. This is + /// the signal that would have caught a store at 12k pending generations + /// days before it OOMKilled its readers, so a read must emit it. + #[test] + fn reads_sample_pending_generations_per_shard() { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::MetricKind; + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let rt = tokio::runtime::Runtime::new().unwrap(); + + metrics::with_local_recorder(&recorder, || { + rt.block_on(async { + let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap(); + // Three sealed adds leave three flushed generations pending on + // this instance's single shard. + for i in 0..3 { + store + .add(&[row(json!({"id": format!("r{i}")}))]) + .await + .unwrap(); + } + store.list(None, None).await.unwrap(); + }); + }); + + let samples: Vec = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| { + key.kind() == MetricKind::Histogram + && key.key().name() == crate::metrics::ROLLOUT_WAL_PENDING_GENERATIONS + }) + .flat_map(|(_, _, _, value)| match value { + DebugValue::Histogram(v) => v.into_iter().map(|s| s.into_inner()).collect(), + _ => Vec::new(), + }) + .collect(); + assert!( + samples.contains(&3.0), + "a read must sample the shard's 3 pending generations; got {samples:?}" + ); + } + #[test] fn duplicate_ids_dedup_to_the_newest_write() { // The LSM merge key is `id`, so a re-added id supersedes the old row. diff --git a/crates/lance-context-core/src/metrics.rs b/crates/lance-context-core/src/metrics.rs index 332f6ca6..2ee86eef 100644 --- a/crates/lance-context-core/src/metrics.rs +++ b/crates/lance-context-core/src/metrics.rs @@ -54,6 +54,17 @@ pub const ROLLOUT_WAL_MERGE_DURATION: &str = "rollout_wal_merge_duration_seconds /// aborts on the first failing phase, so this also identifies where it died. pub const ROLLOUT_WAL_MERGE_ERRORS: &str = "rollout_wal_merge_errors_total"; +/// Flushed MemWAL generations pending merge on one shard, sampled once per +/// shard on every LSM read (`wal_shard_snapshots`). Unlabelled: the store and +/// shard go on the accompanying `warn!` span, not on the series. +/// +/// This is the read-amplification signal. Every pending generation is a +/// separate dataset the read path must open, so a shard whose count climbs into +/// the hundreds means its owner's merge has stopped keeping up — the failure +/// that took a worker fleet down at ~12k pending generations across 20 shards +/// before it was noticed in crash logs. +pub const ROLLOUT_WAL_PENDING_GENERATIONS: &str = "rollout_wal_pending_generations"; + /// Emit a histogram sample in seconds. No-op without the `metrics` feature. #[cfg(feature = "metrics")] macro_rules! observe_duration { @@ -69,6 +80,22 @@ macro_rules! observe_duration { }}; } +/// Emit a histogram sample of a plain value (a count, not a duration). No-op +/// without the `metrics` feature. +#[cfg(feature = "metrics")] +macro_rules! observe_value { + ($name:expr, $value:expr $(, $k:expr => $v:expr)* $(,)?) => { + ::metrics::histogram!($name $(, $k => $v)*).record($value as f64) + }; +} + +#[cfg(not(feature = "metrics"))] +macro_rules! observe_value { + ($name:expr, $value:expr $(, $k:expr => $v:expr)* $(,)?) => {{ + let _ = &$value; + }}; +} + /// Increment a counter by one. No-op without the `metrics` feature. /// /// The disabled arm still expands to a unit *expression* rather than an empty @@ -144,4 +171,6 @@ macro_rules! observe_phase { }}; } -pub(crate) use {count, observe_duration, observe_phase, timer_elapsed, timer_start}; +pub(crate) use { + count, observe_duration, observe_phase, observe_value, timer_elapsed, timer_start, +}; diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index cf5e6202..2082cef4 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -383,6 +383,11 @@ pub struct RolloutStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Shared Lance [`Session`] used to open this store's base dataset (and, /// transitively, every flushed MemWAL generation it reads — those inherit /// the base dataset's session). @@ -469,6 +474,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, } = options; let mut base = StorageBase::open( @@ -479,6 +485,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema: Arc::new(rollout_schema()), key_column: "id".to_string(), @@ -2487,6 +2494,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, schema: legacy_schema.clone(), key_column: "id".to_string(), @@ -2806,6 +2814,7 @@ mod tests { merge_after_generations: Some(0), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -2850,6 +2859,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }; { @@ -2899,6 +2909,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -2950,6 +2961,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }; let instance_a = RolloutStore::open_with_options(&uri, options("rollout-0")) @@ -3171,6 +3183,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -3233,6 +3246,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -3333,6 +3347,7 @@ mod tests { merge_after_generations: None, // no merge → epoch never reclaimed merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3391,6 +3406,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3430,6 +3446,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3566,6 +3583,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3652,6 +3670,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3699,6 +3718,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3768,6 +3788,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3805,6 +3826,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3863,6 +3885,7 @@ mod tests { merge_after_generations: Some(2), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -3904,6 +3927,7 @@ mod tests { merge_after_generations: None, // count trigger off merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -4403,6 +4427,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, ..Default::default() }, ) @@ -4501,6 +4526,7 @@ mod tests { merge_after_generations: None, // disabled merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await @@ -4533,6 +4559,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, }, ) .await diff --git a/crates/lance-context-core/src/store.rs b/crates/lance-context-core/src/store.rs index ca6c313f..3dfc022e 100644 --- a/crates/lance-context-core/src/store.rs +++ b/crates/lance-context-core/src/store.rs @@ -302,6 +302,11 @@ pub struct ContextStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every read. `None` uses the crate default + /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` + /// histogram is emitted regardless. + pub pending_generations_warn: Option, /// Whether [`ContextStore::add`] seals the memtable before returning, so the /// rows it wrote are immediately readable. /// @@ -333,6 +338,7 @@ impl Default for ContextStoreOptions { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, // Read-your-write by default; see the field docs. seal_on_add: true, } @@ -614,6 +620,7 @@ impl ContextStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + pending_generations_warn: options.pending_generations_warn, session: None, schema: Arc::new(arrow_schema.clone()), key_column: "id".to_string(), @@ -2237,6 +2244,7 @@ impl ContextStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, // A compactor never appends, so the seal mode is irrelevant to it; // deferring keeps it from ever emitting a generation. seal_on_add: false, diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 0e75933c..22b893d8 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -67,7 +67,9 @@ use lance_index::IndexType; use tracing::{info, warn}; use uuid::Uuid; -use crate::metrics::{count, observe_duration, observe_phase, timer_elapsed, timer_start}; +use crate::metrics::{ + count, observe_duration, observe_phase, observe_value, timer_elapsed, timer_start, +}; use crate::store::{CompactionConfig, CompactionStats}; /// Number of shard manifest files to scan per batch when discovering the latest @@ -97,6 +99,14 @@ pub(crate) const DEFAULT_MERGE_MAX_GENERATIONS: usize = 8; /// Buffered Arrow array bytes per merge pass, checked at generation boundaries. pub(crate) const DEFAULT_MERGE_MAX_BYTES: usize = 1024 * 1024 * 1024; +/// Default per-shard pending-generation count at which an LSM read warns. +/// +/// Sized well above the count trigger (`merge_after_generations`, 50 in the +/// reference deployment) and the per-pass cap (`merge_max_generations`, 8), so a +/// healthy store never trips it, while sitting far below the counts that make a +/// read unaffordable: at ~1,500 per shard, reads exhausted a 32 GiB worker. +pub(crate) const DEFAULT_PENDING_GENERATIONS_WARN: usize = 256; + /// Execute only the first `max_source_fragments` from a Lance compaction plan. /// /// Lance's built-in `max_source_fragments` stops before a whole planned task @@ -226,6 +236,10 @@ pub(crate) struct StorageBaseOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Warn when a single shard has at least this many flushed generations + /// pending merge (sampled on every LSM read). `None` uses the crate default + /// (256); `Some(0)` disables the warning. The metric is always emitted. + pub pending_generations_warn: Option, /// Shared, capacity-bounded Lance session. `None` preserves Lance's /// per-open default (a fresh 6 GiB index + 1 GiB metadata session *per /// store*, which is the source of unbounded per-append RSS growth). @@ -287,6 +301,8 @@ pub(crate) struct StorageBase { merge_after_generations: usize, merge_max_generations: usize, merge_max_bytes: usize, + /// Per-shard pending-generation count at which reads warn; `0` disables. + pending_generations_warn: usize, /// Timestamp of the last successful [`Self::compact`] on this handle. last_compaction: Option>, /// Number of successful compactions performed by this handle. @@ -343,6 +359,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -373,6 +390,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -397,6 +415,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + pending_generations_warn, session, schema, key_column, @@ -422,6 +441,8 @@ impl StorageBase { merge_after_generations: merge_after_generations.unwrap_or(0), merge_max_generations: merge_max_generations.unwrap_or(DEFAULT_MERGE_MAX_GENERATIONS), merge_max_bytes: merge_max_bytes.unwrap_or(DEFAULT_MERGE_MAX_BYTES), + pending_generations_warn: pending_generations_warn + .unwrap_or(DEFAULT_PENDING_GENERATIONS_WARN), last_compaction: None, total_compactions: 0, last_compaction_error: None, @@ -1386,11 +1407,14 @@ impl StorageBase { let object_store = self.dataset.object_store(None).await?; let branch_path = self.dataset.branch_location().path.clone(); let shard_ids = self.dataset.list_mem_wal_latest_shard_ids().await?; + let warn_at = self.pending_generations_warn; + let uri: Arc = Arc::from(self.dataset.uri()); let snapshots: Vec> = stream::iter(shard_ids) .map(|shard_id| { let object_store = object_store.clone(); let branch_path = branch_path.clone(); + let uri = uri.clone(); async move { let manifest_store = ShardManifestStore::new( object_store, @@ -1401,6 +1425,21 @@ impl StorageBase { let Some(manifest) = manifest_store.read_latest().await? else { return Ok(None); }; + let pending = manifest.flushed_generations.len(); + observe_value!(crate::metrics::ROLLOUT_WAL_PENDING_GENERATIONS, pending); + if warn_at != 0 && pending >= warn_at { + // Every pending generation is one more dataset this read + // must open; past this point the shard's owner has stopped + // keeping up and reads are on their way to exhausting memory. + warn!( + uri = %uri, + shard = %shard_id, + pending, + warn_at, + "MemWAL shard has many flushed generations pending merge; \ + reads open every one of them" + ); + } let mut snapshot = ShardSnapshot::new(shard_id) .with_spec_id(manifest.shard_spec_id) .with_current_generation(manifest.current_generation); diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index 3c54d9c2..bf0b2919 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -62,6 +62,15 @@ const JOB_LATENCY_METRICS: &[&str] = &[ "rollout_wal_merge_lock_wait_seconds", ]; +/// Buckets (upper bounds, generation counts) for `rollout_wal_pending_generations`. +/// +/// Powers of two from the per-pass merge cap (8) up past the point where a read +/// is unaffordable (~1,500 per shard exhausted a 32 GiB worker), so both the +/// healthy band and the runaway band resolve. 9 buckets, same cardinality +/// reasoning as [`REQUEST_LATENCY_BUCKETS`]. +const PENDING_GENERATIONS_BUCKETS: &[f64] = + &[8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0, 1024.0, 4096.0]; + /// Handle used to render the Prometheus exposition text on demand, plus a /// process-resource collector that is refreshed on each scrape. #[derive(Clone)] @@ -92,6 +101,12 @@ pub fn install_recorder() -> MetricsHandle { .set_buckets_for_metric(Matcher::Full((*name).to_string()), JOB_LATENCY_BUCKETS) .expect("job latency buckets are non-empty"); } + builder = builder + .set_buckets_for_metric( + Matcher::Full("rollout_wal_pending_generations".to_string()), + PENDING_GENERATIONS_BUCKETS, + ) + .expect("pending generation buckets are non-empty"); let prometheus = builder .install_recorder() .expect("failed to install Prometheus recorder"); diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index b4ad4e4a..c4eb86b5 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -50,6 +50,18 @@ pub struct ServerConfig { #[arg(long, env = "ROLLOUT_MERGE_MAX_BYTES", default_value = "1073741824")] pub rollout_merge_max_bytes: usize, + /// Warn when one MemWAL shard has at least this many flushed generations + /// pending merge, sampled on every LSM read of any store kind. Every + /// pending generation is a separate dataset a read must open, so this is + /// the read-amplification alarm; `0` disables the warn. The + /// `rollout_wal_pending_generations` histogram is emitted regardless. + #[arg( + long, + env = "ROLLOUT_WAL_PENDING_WARN_GENERATIONS", + default_value = "256" + )] + pub rollout_wal_pending_warn_generations: usize, + /// Interval, in seconds, for the periodic per-shard WAL cleanup task. When /// non-zero, the global sweeper folds this instance's flushed MemWAL /// generations into the base table on a schedule — the *time* half of the @@ -213,6 +225,7 @@ mod tests { let config = ServerConfig::try_parse_from(["lance-context-server", "--data-dir", "/tmp/x"]) .expect("the documented minimal invocation must parse"); assert_eq!(config.rollout_merge_max_bytes, 1024 * 1024 * 1024); + assert_eq!(config.rollout_wal_pending_warn_generations, 256); assert_eq!(config.rollout_flush_interval_secs, 30); assert_eq!(config.rollout_cleanup_interval_secs, 0); } diff --git a/crates/lance-context-server/src/routes/datagen.rs b/crates/lance-context-server/src/routes/datagen.rs index bc6f1281..5895d519 100644 --- a/crates/lance-context-server/src/routes/datagen.rs +++ b/crates/lance-context-server/src/routes/datagen.rs @@ -48,6 +48,7 @@ pub async fn create_datagen_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), cleanup_interval_secs: None, }; let store = DatagenStore::open_with_options(&uri, options) diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index 69ccda2d..2e1cf153 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -59,6 +59,7 @@ pub async fn create_generic_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), session: None, seal_on_add: req.seal_on_add, }; diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index 0d27dba0..c9c9a160 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -199,6 +199,7 @@ pub async fn create_rollout_store( .then_some(state.rollout_merge_after_generations), merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), session: state.rollout_session.clone(), }; diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index ef6d15fb..5fd09292 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -121,6 +121,8 @@ pub struct AppState { pub rollout_merge_max_generations: usize, /// Buffered Arrow array byte budget per merge pass; `0` disables this cap. pub rollout_merge_max_bytes: usize, + /// Per-shard pending-generation count at which reads warn; `0` disables. + pub rollout_wal_pending_warn_generations: usize, /// Periodic per-shard WAL-cleanup interval in seconds; `0` disables the /// global sweeper. See [`Self::spawn_global_sweeper`]. pub rollout_cleanup_interval_secs: u64, @@ -312,6 +314,7 @@ impl AppState { rollout_merge_after_generations: config.rollout_merge_after_generations, rollout_merge_max_generations: config.rollout_merge_max_generations, rollout_merge_max_bytes: config.rollout_merge_max_bytes, + rollout_wal_pending_warn_generations: config.rollout_wal_pending_warn_generations, rollout_cleanup_interval_secs: config.rollout_cleanup_interval_secs, rollout_flush_interval_secs: config.rollout_flush_interval_secs, blob_budget, @@ -377,6 +380,7 @@ impl AppState { rollout_merge_after_generations: 0, rollout_merge_max_generations: 8, rollout_merge_max_bytes: 1024 * 1024 * 1024, + rollout_wal_pending_warn_generations: 256, rollout_cleanup_interval_secs: 0, rollout_flush_interval_secs: 0, blob_budget: None, @@ -412,6 +416,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), session: self.rollout_session.clone(), } } @@ -600,6 +605,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), cleanup_interval_secs: None, } } @@ -742,6 +748,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), session: self.rollout_session.clone(), seal_on_add, } diff --git a/crates/lance-context/src/unified_datagen.rs b/crates/lance-context/src/unified_datagen.rs index bf1615a2..1668bbce 100644 --- a/crates/lance-context/src/unified_datagen.rs +++ b/crates/lance-context/src/unified_datagen.rs @@ -38,6 +38,7 @@ impl DatagenStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, cleanup_interval_secs: None, }; let store = LocalStore::open_with_options(uri, options) diff --git a/crates/lance-context/src/unified_generic.rs b/crates/lance-context/src/unified_generic.rs index 92685542..70ba02ae 100644 --- a/crates/lance-context/src/unified_generic.rs +++ b/crates/lance-context/src/unified_generic.rs @@ -41,6 +41,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, seal_on_add, }; @@ -62,6 +63,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + pending_generations_warn: None, session: None, seal_on_add, };