Skip to content

feat(core): sample and warn on MemWAL pending generations per shard #261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
beinan merged 1 commit into from
Sep 22, 2026
Merged

feat(core): sample and warn on MemWAL pending generations per shard #261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions crates/lance-context-core/src/datagen_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>,
/// 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<usize>,
/// Periodically merge this writer's pending generations. `None` or zero
/// disables the timer.
pub cleanup_interval_secs: Option<u64>,
Expand Down Expand Up @@ -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
Expand Down
55 changes: 55 additions & 0 deletions crates/lance-context-core/src/generic_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>,
/// 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<usize>,
/// Shared, capacity-bounded Lance session.
pub session: Option<Arc<Session>>,
/// Whether [`GenericStore::add`] seals before returning, making the rows it
Expand Down Expand Up @@ -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`
Expand Down Expand Up @@ -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<f64> = 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.
Expand Down
31 changes: 30 additions & 1 deletion crates/lance-context-core/src/metrics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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
Expand Down Expand Up @@ -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,
};
27 changes: 27 additions & 0 deletions crates/lance-context-core/src/rollout_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>,
/// 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<usize>,
/// 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).
Expand Down Expand Up @@ -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(
Expand All @@ -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(),
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -2806,6 +2814,7 @@ mod tests {
merge_after_generations: Some(0),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -2850,6 +2859,7 @@ mod tests {
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
};

{
Expand Down Expand Up @@ -2899,6 +2909,7 @@ mod tests {
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -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"))
Expand Down Expand Up @@ -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()
},
)
Expand Down Expand Up @@ -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()
},
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -3391,6 +3406,7 @@ mod tests {
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3430,6 +3446,7 @@ mod tests {
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3566,6 +3583,7 @@ mod tests {
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3652,6 +3670,7 @@ mod tests {
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3699,6 +3718,7 @@ mod tests {
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3768,6 +3788,7 @@ mod tests {
merge_after_generations: Some(3),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3805,6 +3826,7 @@ mod tests {
merge_after_generations: Some(3),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -3863,6 +3885,7 @@ mod tests {
merge_after_generations: Some(2),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
},
)
Expand Down Expand Up @@ -4501,6 +4526,7 @@ mod tests {
merge_after_generations: None, // disabled
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down Expand Up @@ -4533,6 +4559,7 @@ mod tests {
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
pending_generations_warn: None,
},
)
.await
Expand Down
8 changes: 8 additions & 0 deletions crates/lance-context-core/src/store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>,
/// 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<usize>,
/// Whether [`ContextStore::add`] seals the memtable before returning, so the
/// rows it wrote are immediately readable.
///
Expand Down Expand Up @@ -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,
}
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading