diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 823411056..0180d37be 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -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}; diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index d8d63072c..a2a637eec 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -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 = 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, - attempt: usize, - ) -> Result { - 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 diff --git a/rust/lancedb/src/table/checkpoint.rs b/rust/lancedb/src/table/checkpoint.rs index 43b4105ba..069a252e2 100644 --- a/rust/lancedb/src/table/checkpoint.rs +++ b/rust/lancedb/src/table/checkpoint.rs @@ -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, -} - -/// 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, -} - -/// 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, - /// 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>, -} - -impl BucketStats { - /// The newest flushed generation, or `None` when L0 is empty. - pub(crate) fn newest_generation(&self) -> Option { - 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, -} - -/// 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, -} +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 { /// 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 { +fn checkpoint_fault(e: Error, attempt: usize) -> Result { #[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 = 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, + attempt: usize, +) -> Result { + 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 = [&[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()); - } } diff --git a/rust/lancedb/src/table/lsm_stats.rs b/rust/lancedb/src/table/lsm_stats.rs new file mode 100644 index 000000000..28c2cd3e2 --- /dev/null +++ b/rust/lancedb/src/table/lsm_stats.rs @@ -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, +} + +/// 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, +} + +/// 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, + /// 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>, +} + +impl BucketStats { + /// The newest flushed generation, or `None` when L0 is empty. + pub(crate) fn newest_generation(&self) -> Option { + 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, +} + +/// 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, +} + +#[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 = [&[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()); + } +}