rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-14 14:46:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(table): one file per LSM operation

Browse Source 
`table/checkpoint.rs` held the stats types, the fault taxonomy, and the
constants, while the checkpoint loop itself lived in `table.rs` — the only
substantial body in a file where every other `Table` method is a delegate.

Split along the operation boundary the rest of the crate already uses:
`table/lsm_stats.rs` owns the `get_lsm_stats` shape and its tests,
`table/checkpoint.rs` owns the loop plus the fault classification it acts
on, and `Table::checkpoint_lsm` becomes a one-line delegate like its
neighbors. The loop's helper constants drop to private now that nothing
outside the module reads them.

No behavior change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Rammer
2026-08-03 13:34:20 -05:00
co-authored by Claude Opus 5
parent 0bd7d04079
commit feeedc9d25
4 changed files with 305 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/lancedb/src/remote/table.rs
+2 -1
View File
@@ -27,7 +27,8 @@ use crate::table::MergeBranchResult;
use crate::table::MergeResult;
use crate::table::Tags;
use crate::table::UpdateResult;
use crate::table::checkpoint::{CODE_INVALID_TABLE_STATE, GetLsmStatsResponse, classify_lsm_fault};
use crate::table::checkpoint::{CODE_INVALID_TABLE_STATE, classify_lsm_fault};
use crate::table::lsm_stats::GetLsmStatsResponse;
use crate::table::merge::MergeFilter;
use crate::table::query::create_multi_vector_plan;
use crate::table::{AlterColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult};
rust/lancedb/src/table.rs
+5 -113
View File
@@ -68,6 +68,7 @@ mod create_index;
pub mod datafusion;
pub(crate) mod dataset;
pub mod delete;
pub mod lsm_stats;
pub mod merge;
pub mod optimize;
mod primary_key;
@@ -83,11 +84,7 @@ pub use branch_merge::{
BranchDiff, ColumnChange, ColumnSummary, IndexSummary, MergeBlocker, MergeBlockerCode,
MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary,
};
pub use checkpoint::{BucketStats, GenerationStats, LsmFault, LsmStats, MemtableStats};
use checkpoint::{
CheckpointControl, CheckpointOutcome, MAX_IDLE_POLLS, MAX_REISSUES, POLL_INTERVAL, backoff,
checkpoint_fault,
};
pub use checkpoint::LsmFault;
pub use chrono::Duration;
pub use delete::DeleteResult;
use futures::future::join_all;
@@ -95,6 +92,7 @@ pub use lance::dataset::refs::{BranchContents, Ref, TagContents, Tags as LanceTa
pub use lance::dataset::scanner::DatasetRecordBatchStream;
use lance::dataset::statistics::DatasetStatisticsExt;
pub use lance_index::optimize::OptimizeOptions;
pub use lsm_stats::{BucketStats, GenerationStats, LsmStats, MemtableStats};
pub use optimize::{CompactionOptions, OptimizeAction, OptimizeStats};
pub use schema_evolution::{
AddColumnsResult, AlterColumnsResult, DropColumnsResult, FieldMetadataUpdate,
@@ -1668,7 +1666,7 @@ impl Table {
/// already-converged table costs two round trips and triggers nothing.
///
/// **No deadline.** It returns when the target generations are gone, or
/// errors after [`MAX_IDLE_POLLS`] polls with neither progress nor a
/// errors after a bounded number of polls with neither progress nor a
/// running pass — a fenced bucket, a saturated compactor pool, a
/// failing merge. Wrap it in `tokio::time::timeout` if you need a
/// wall-clock bound.
@@ -1685,113 +1683,7 @@ impl Table {
/// # }
/// ```
pub async fn checkpoint_lsm(&self) -> Result<()> {
for attempt in 0..=MAX_REISSUES {
// The seal is what turns everything written before this call
// into a generation, so the watermark has to be read after it.
// It is also idempotent: sealing an empty active memtable is a
// no-op, so a re-issue does not churn empty generations.
match self.flush_lsm().await {
Ok(()) => {}
Err(e) => match checkpoint_fault(e, attempt)? {
CheckpointControl::Retry | CheckpointControl::ReissueFromFlush => {
backoff(attempt).await;
continue;
}
},
}
let Some(stats) = self.get_lsm_stats(false).await? else {
// Not WAL-backed: nothing to converge, and `flush_lsm`
// would have errored first on every path but a race.
return Ok(());
};
let targets: HashMap<String, u64> = stats
.buckets
.iter()
.filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
.collect();
if targets.is_empty() {
return Ok(());
}
match self.drain_to_targets(&targets, attempt).await? {
CheckpointOutcome::Done => return Ok(()),
CheckpointOutcome::ReissueFromFlush => continue,
}
}
Err(Error::Runtime {
message: "checkpoint_lsm: the owning node kept losing its claim; \
re-issued from flush the maximum number of times"
.into(),
})
}
/// Trigger and poll until no bucket holds a generation at or below its
/// target. Split out so the flush re-issue path above stays readable.
async fn drain_to_targets(
&self,
targets: &HashMap<String, u64>,
attempt: usize,
) -> Result<CheckpointOutcome> {
let mut idle_polls = 0;
let mut last_outstanding = usize::MAX;
loop {
let Some(stats) = self.get_lsm_stats(false).await? else {
return Ok(CheckpointOutcome::Done);
};
let mut outstanding = 0;
let mut buckets_left = 0;
let mut all_compacting = true;
for b in &stats.buckets {
let Some(target) = targets.get(&b.shard_id) else {
continue;
};
let n = b.outstanding_generations(*target);
if n > 0 {
outstanding += n;
buckets_left += 1;
all_compacting &= b.compacting;
}
}
if outstanding == 0 {
return Ok(CheckpointOutcome::Done);
}
// Progress, not time, is the bound. A pass already running on
// every outstanding bucket counts as progress: piling on would
// only collect 429s, and the latch it would contend for is the
// one doing the work.
if outstanding < last_outstanding || all_compacting {
idle_polls = 0;
} else {
idle_polls += 1;
if idle_polls >= MAX_IDLE_POLLS {
return Err(Error::Runtime {
message: format!(
"checkpoint_lsm: {outstanding} generation(s) across {buckets_left} \
bucket(s) stopped making progress with no compaction running; the \
compactor pool may be saturated or the writer fenced"
),
});
}
}
last_outstanding = outstanding;
if !all_compacting {
match self.compact_lsm().await {
Ok(()) => {}
Err(e) => match checkpoint_fault(e, attempt)? {
// Every bucket busy (429) is the state the poll
// above already handles; fall through and re-read.
CheckpointControl::Retry => {}
CheckpointControl::ReissueFromFlush => {
return Ok(CheckpointOutcome::ReissueFromFlush);
}
},
}
}
tokio::time::sleep(POLL_INTERVAL).await;
}
checkpoint::checkpoint_lsm(self).await
}
/// Seal every bucket's active memtable into L0 without touching the
rust/lancedb/src/table/checkpoint.rs
+123 -171
View File
@@ -34,100 +34,7 @@
use std::collections::HashMap;
use std::time::Duration;
use serde::Deserialize;
/// One flushed L0 generation.
#[derive(Debug, Clone, Deserialize)]
pub struct GenerationStats {
pub generation: u64,
pub bytes: u64,
/// Present only when `include_generation_rows` was requested. Off by
/// default because each count opens an uncached Lance dataset, and the
/// checkpoint loop polls this route needing only generation numbers.
#[serde(default)]
pub rows: Option<u64>,
}
/// One in-memory memtable.
#[derive(Debug, Clone, Deserialize)]
pub struct MemtableStats {
pub generation: u64,
pub rows: u64,
pub bytes: u64,
pub batches: u64,
/// Names of the indexes this memtable carries. An absent name is the
/// whole answer to "why is my fresh-tier search on that column
/// brute-force"; cross-reference `list_indices` for its kind and column.
pub indexes: Vec<String>,
}
/// Live state of one bucket. A table is N buckets on one node; flattening
/// to a single number hides the one hot bucket that is usually why someone
/// opened this endpoint.
///
/// Nothing here is derived — sums and differences (total L0 bytes, WAL lag)
/// are the caller's to compute from measured fields.
#[derive(Debug, Clone, Deserialize)]
pub struct BucketStats {
pub shard_id: String,
/// `Active` | `Sealed` (drop-table 2PC in flight).
pub status: String,
pub writer_epoch: u64,
pub manifest_version: u64,
pub current_generation: u64,
pub replay_after_wal_entry_position: u64,
pub wal_entry_position_last_seen: u64,
pub generations: Vec<GenerationStats>,
/// Whether a pass owns this bucket's compaction latch right now. Says
/// *a* driver is running, not *whose*: the server's periodic trigger
/// competes for the same latch, so a caller that dispatched a compact
/// cannot read this as "mine is progressing", only as "do not pile on".
pub compacting: bool,
/// Oldest first, active last. Absent for a `Sealed` bucket, whose
/// in-memory state is torn down.
#[serde(default)]
pub memtables: Option<Vec<MemtableStats>>,
}
impl BucketStats {
/// The newest flushed generation, or `None` when L0 is empty.
pub(crate) fn newest_generation(&self) -> Option<u64> {
self.generations.iter().map(|g| g.generation).max()
}
/// How many generations at or below `target` are still in L0.
///
/// A *count*, not a boolean: it is the checkpoint's progress metric,
/// and one pass drains a bounded prefix rather than the whole target
/// set. A per-bucket boolean would read as "no progress" for every
/// pass but the last, so a bucket needing more passes than the idle
/// bound would fail a checkpoint that was working correctly.
/// Compaction drains oldest-first, so this decreases monotonically.
pub(crate) fn outstanding_generations(&self, target: u64) -> usize {
self.generations
.iter()
.filter(|g| g.generation <= target)
.count()
}
}
/// Live LSM state, one entry per bucket.
///
/// Every field is measured. There is no "WAL is off" shape here — that
/// case is `None` from [`crate::Table::get_lsm_stats`], because a struct of
/// zeros would read as measurements.
#[derive(Debug, Clone, Deserialize)]
pub struct LsmStats {
pub buckets: Vec<BucketStats>,
}
/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
/// the table has no LSM write path.
#[derive(Debug, Deserialize)]
pub(crate) struct GetLsmStatsResponse {
#[serde(default)]
pub lsm_stats: Option<LsmStats>,
}
use crate::{Error, Result, Table};
/// How a failed LSM request should be handled by the checkpoint loop.
///
@@ -196,28 +103,28 @@ fn body_code(body: &str) -> Option<u64> {
/// How long the loop waits between `get_lsm_stats` polls. A pass over a
/// `prefix_max` prefix takes seconds, so a shorter interval only adds
/// round trips.
pub(crate) const POLL_INTERVAL: Duration = Duration::from_secs(1);
const POLL_INTERVAL: Duration = Duration::from_secs(1);
/// Consecutive polls tolerated with no progress *and* no pass running
/// before the checkpoint gives up. This is the bound in place of a
/// deadline: a slow table waits (a running pass resets the counter) while
/// a stuck one — a fenced bucket, a saturated compactor pool, a failing
/// merge — fails loudly instead of spinning.
pub(crate) const MAX_IDLE_POLLS: usize = 10;
const MAX_IDLE_POLLS: usize = 10;
/// Cap on re-issues from `flush` after a 404. A node in a crash loop would
/// otherwise turn flush → compact → 404 → flush into a spin.
pub(crate) const MAX_REISSUES: usize = 3;
const MAX_REISSUES: usize = 3;
/// Base backoff between retries, doubled per consecutive retry up to
/// [`RETRY_BACKOFF_MAX`]. Latch contention clears in about the time one
/// compaction pass takes, so starting small is right; a saturated pool
/// wants the ceiling.
pub(crate) const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100);
pub(crate) const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5);
const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(100);
const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5);
/// Sleep before re-issuing a retryable request.
pub(crate) async fn backoff(attempt: usize) {
async fn backoff(attempt: usize) {
let delay = RETRY_BACKOFF_BASE
.saturating_mul(1u32 << attempt.min(8) as u32)
.min(RETRY_BACKOFF_MAX);
@@ -226,7 +133,7 @@ pub(crate) async fn backoff(attempt: usize) {
/// What the checkpoint loop should do next after a failed request.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointControl {
enum CheckpointControl {
/// Re-issue the same call after a backoff.
Retry,
/// Restart from `flush`, which is what re-claims and replays.
@@ -235,7 +142,7 @@ pub(crate) enum CheckpointControl {
/// Whether the drain loop finished or needs the table re-claimed first.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointOutcome {
enum CheckpointOutcome {
Done,
ReissueFromFlush,
}
@@ -253,13 +160,10 @@ pub(crate) enum CheckpointOutcome {
/// Draining is terminal by design: the drain gate is a one-way latch that
/// never releases until the process restarts, so retrying against that
/// node spins forever when the truthful answer was in the first response.
pub(crate) fn checkpoint_fault(
e: crate::Error,
attempt: usize,
) -> crate::Result<CheckpointControl> {
fn checkpoint_fault(e: Error, attempt: usize) -> Result<CheckpointControl> {
#[cfg(feature = "remote")]
let fault = match &e {
crate::Error::LsmRoute { fault, .. } => Some(*fault),
Error::LsmRoute { fault, .. } => Some(*fault),
_ => None,
};
#[cfg(not(feature = "remote"))]
@@ -274,6 +178,118 @@ pub(crate) fn checkpoint_fault(
}
}
/// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark
/// from the resulting L0, then trigger and poll until it drains.
pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
for attempt in 0..=MAX_REISSUES {
// The seal is what turns everything written before this call into a
// generation, so the watermark has to be read after it. It is also
// idempotent: sealing an empty active memtable is a no-op, so a
// re-issue does not churn empty generations.
match table.flush_lsm().await {
Ok(()) => {}
Err(e) => match checkpoint_fault(e, attempt)? {
CheckpointControl::Retry | CheckpointControl::ReissueFromFlush => {
backoff(attempt).await;
continue;
}
},
}
let Some(stats) = table.get_lsm_stats(false).await? else {
// Not WAL-backed: nothing to converge, and `flush_lsm` would
// have errored first on every path but a race.
return Ok(());
};
let targets: HashMap<String, u64> = stats
.buckets
.iter()
.filter_map(|b| Some((b.shard_id.clone(), b.newest_generation()?)))
.collect();
if targets.is_empty() {
return Ok(());
}
match drain_to_targets(table, &targets, attempt).await? {
CheckpointOutcome::Done => return Ok(()),
CheckpointOutcome::ReissueFromFlush => continue,
}
}
Err(Error::Runtime {
message: "checkpoint_lsm: the owning node kept losing its claim; \
re-issued from flush the maximum number of times"
.into(),
})
}
/// Trigger and poll until no bucket holds a generation at or below its
/// target. Split out so the flush re-issue path above stays readable.
async fn drain_to_targets(
table: &Table,
targets: &HashMap<String, u64>,
attempt: usize,
) -> Result<CheckpointOutcome> {
let mut idle_polls = 0;
let mut last_outstanding = usize::MAX;
loop {
let Some(stats) = table.get_lsm_stats(false).await? else {
return Ok(CheckpointOutcome::Done);
};
let mut outstanding = 0;
let mut buckets_left = 0;
let mut all_compacting = true;
for b in &stats.buckets {
let Some(target) = targets.get(&b.shard_id) else {
continue;
};
let n = b.outstanding_generations(*target);
if n > 0 {
outstanding += n;
buckets_left += 1;
all_compacting &= b.compacting;
}
}
if outstanding == 0 {
return Ok(CheckpointOutcome::Done);
}
// Progress, not time, is the bound. A pass already running on
// every outstanding bucket counts as progress: piling on would
// only collect 429s, and the latch it would contend for is the
// one doing the work.
if outstanding < last_outstanding || all_compacting {
idle_polls = 0;
} else {
idle_polls += 1;
if idle_polls >= MAX_IDLE_POLLS {
return Err(Error::Runtime {
message: format!(
"checkpoint_lsm: {outstanding} generation(s) across {buckets_left} \
bucket(s) stopped making progress with no compaction running; the \
compactor pool may be saturated or the writer fenced"
),
});
}
}
last_outstanding = outstanding;
if !all_compacting {
match table.compact_lsm().await {
Ok(()) => {}
Err(e) => match checkpoint_fault(e, attempt)? {
// Every bucket busy (429) is the state the poll above
// already handles; fall through and re-read.
CheckpointControl::Retry => {}
CheckpointControl::ReissueFromFlush => {
return Ok(CheckpointOutcome::ReissueFromFlush);
}
},
}
}
tokio::time::sleep(POLL_INTERVAL).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -317,68 +333,4 @@ mod tests {
assert_eq!(classify_lsm_fault(503, "gateway timeout"), LsmFault::Retry);
assert_eq!(classify_lsm_fault(503, r#"{"code":"19"}"#), LsmFault::Retry);
}
fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats {
BucketStats {
shard_id: shard.into(),
status: "Active".into(),
writer_epoch: 1,
manifest_version: 1,
current_generation: generations.iter().max().copied().unwrap_or(0) + 1,
replay_after_wal_entry_position: 0,
wal_entry_position_last_seen: 0,
generations: generations
.iter()
.map(|g| GenerationStats {
generation: *g,
bytes: 1,
rows: None,
})
.collect(),
compacting,
memtables: None,
}
}
/// The target watermark is the newest generation at the start, and a
/// generation created *after* it must not hold the loop open — that is
/// the whole reason the predicate terminates under write load.
#[test]
fn newer_generations_do_not_extend_the_target() {
let start = bucket("b0", &[7, 8], false);
let target = start.newest_generation().expect("L0 is non-empty");
assert_eq!(target, 8);
// Compaction drained 7 and 8; 9 and 10 arrived while it ran.
let later = bucket("b0", &[9, 10], false);
assert_eq!(
later.outstanding_generations(target),
0,
"generations above the target are somebody else's problem"
);
// Still holding 8 means still outstanding.
assert_eq!(
bucket("b0", &[8, 9], false).outstanding_generations(target),
1
);
}
/// The progress metric counts generations, not buckets. A pass drains
/// a bounded prefix, so one bucket going 3 → 2 → 1 → 0 must read as
/// three steps of progress, not three idle polls.
#[test]
fn progress_is_measured_in_generations() {
let target = 3;
let counts: Vec<usize> = [&[1u64, 2, 3][..], &[2, 3][..], &[3][..], &[][..]]
.iter()
.map(|gens| bucket("b0", gens, false).outstanding_generations(target))
.collect();
assert_eq!(counts, vec![3, 2, 1, 0]);
}
#[test]
fn empty_l0_has_no_target() {
assert!(bucket("b0", &[], false).newest_generation().is_none());
}
}
rust/lancedb/src/table/lsm_stats.rs
+175
View File
@@ -0,0 +1,175 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Live per-bucket LSM state — the shape [`crate::Table::get_lsm_stats`]
//! returns and the one [`super::checkpoint`] polls for its completion
//! predicate.
//!
//! Nothing here is derived. Sums and differences (total L0 bytes, WAL lag)
//! are the caller's to compute from measured fields, and there is no
//! "WAL is off" shape — that case is `None`, because a struct of zeros
//! would read as measurements.
use serde::Deserialize;
/// One flushed L0 generation.
#[derive(Debug, Clone, Deserialize)]
pub struct GenerationStats {
pub generation: u64,
pub bytes: u64,
/// Present only when `include_generation_rows` was requested. Off by
/// default because each count opens an uncached Lance dataset, and the
/// checkpoint loop polls this route needing only generation numbers.
#[serde(default)]
pub rows: Option<u64>,
}
/// One in-memory memtable.
#[derive(Debug, Clone, Deserialize)]
pub struct MemtableStats {
pub generation: u64,
pub rows: u64,
pub bytes: u64,
pub batches: u64,
/// Names of the indexes this memtable carries. An absent name is the
/// whole answer to "why is my fresh-tier search on that column
/// brute-force"; cross-reference `list_indices` for its kind and column.
pub indexes: Vec<String>,
}
/// Live state of one bucket. A table is N buckets on one node; flattening
/// to a single number hides the one hot bucket that is usually why someone
/// opened this endpoint.
///
/// Nothing here is derived — sums and differences (total L0 bytes, WAL lag)
/// are the caller's to compute from measured fields.
#[derive(Debug, Clone, Deserialize)]
pub struct BucketStats {
pub shard_id: String,
/// `Active` | `Sealed` (drop-table 2PC in flight).
pub status: String,
pub writer_epoch: u64,
pub manifest_version: u64,
pub current_generation: u64,
pub replay_after_wal_entry_position: u64,
pub wal_entry_position_last_seen: u64,
pub generations: Vec<GenerationStats>,
/// Whether a pass owns this bucket's compaction latch right now. Says
/// *a* driver is running, not *whose*: the server's periodic trigger
/// competes for the same latch, so a caller that dispatched a compact
/// cannot read this as "mine is progressing", only as "do not pile on".
pub compacting: bool,
/// Oldest first, active last. Absent for a `Sealed` bucket, whose
/// in-memory state is torn down.
#[serde(default)]
pub memtables: Option<Vec<MemtableStats>>,
}
impl BucketStats {
/// The newest flushed generation, or `None` when L0 is empty.
pub(crate) fn newest_generation(&self) -> Option<u64> {
self.generations.iter().map(|g| g.generation).max()
}
/// How many generations at or below `target` are still in L0.
///
/// A *count*, not a boolean: it is the checkpoint's progress metric,
/// and one pass drains a bounded prefix rather than the whole target
/// set. A per-bucket boolean would read as "no progress" for every
/// pass but the last, so a bucket needing more passes than the idle
/// bound would fail a checkpoint that was working correctly.
/// Compaction drains oldest-first, so this decreases monotonically.
pub(crate) fn outstanding_generations(&self, target: u64) -> usize {
self.generations
.iter()
.filter(|g| g.generation <= target)
.count()
}
}
/// Live LSM state, one entry per bucket.
///
/// Every field is measured. There is no "WAL is off" shape here — that
/// case is `None` from [`crate::Table::get_lsm_stats`], because a struct of
/// zeros would read as measurements.
#[derive(Debug, Clone, Deserialize)]
pub struct LsmStats {
pub buckets: Vec<BucketStats>,
}
/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
/// the table has no LSM write path.
#[derive(Debug, Deserialize)]
pub(crate) struct GetLsmStatsResponse {
#[serde(default)]
pub lsm_stats: Option<LsmStats>,
}
#[cfg(test)]
mod tests {
use super::*;
fn bucket(shard: &str, generations: &[u64], compacting: bool) -> BucketStats {
BucketStats {
shard_id: shard.into(),
status: "Active".into(),
writer_epoch: 1,
manifest_version: 1,
current_generation: generations.iter().max().copied().unwrap_or(0) + 1,
replay_after_wal_entry_position: 0,
wal_entry_position_last_seen: 0,
generations: generations
.iter()
.map(|g| GenerationStats {
generation: *g,
bytes: 1,
rows: None,
})
.collect(),
compacting,
memtables: None,
}
}
/// The target watermark is the newest generation at the start, and a
/// generation created *after* it must not hold the loop open — that is
/// the whole reason the predicate terminates under write load.
#[test]
fn newer_generations_do_not_extend_the_target() {
let start = bucket("b0", &[7, 8], false);
let target = start.newest_generation().expect("L0 is non-empty");
assert_eq!(target, 8);
// Compaction drained 7 and 8; 9 and 10 arrived while it ran.
let later = bucket("b0", &[9, 10], false);
assert_eq!(
later.outstanding_generations(target),
0,
"generations above the target are somebody else's problem"
);
// Still holding 8 means still outstanding.
assert_eq!(
bucket("b0", &[8, 9], false).outstanding_generations(target),
1
);
}
/// The progress metric counts generations, not buckets. A pass drains
/// a bounded prefix, so one bucket going 3 → 2 → 1 → 0 must read as
/// three steps of progress, not three idle polls.
#[test]
fn progress_is_measured_in_generations() {
let target = 3;
let counts: Vec<usize> = [&[1u64, 2, 3][..], &[2, 3][..], &[3][..], &[][..]]
.iter()
.map(|gens| bucket("b0", gens, false).outstanding_generations(target))
.collect();
assert_eq!(counts, vec![3, 2, 1, 0]);
}
#[test]
fn empty_l0_has_no_target() {
assert!(bucket("b0", &[], false).newest_generation().is_none());
}
}