From 391b3c1116550a7205ea544fb06dbc4559821dbd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 26 Jun 2026 17:41:07 +0000 Subject: [PATCH 1/2] Add per-catalog active partition count limit Introduce `max_active_partitions` config (default 4096) enforced alongside the existing `max_partition_bytes` limit in `prune_topics`. When the number of active (in-memory, writable) partitions across the catalog exceeds the limit, the oldest active partitions are closed until the count is back within bounds, mirroring the byte-based pruning. Also fix the pruning `ages` map, which was keyed solely on segment end time: partitions sharing an end time collided and overwrote each other, hiding them from both the byte and active-count limits. The key now includes the topic and partition names as tiebreakers. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dcvvv5f1pihsE6gDaETRid --- catalog/src/catalog.rs | 68 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 63 insertions(+), 5 deletions(-) diff --git a/catalog/src/catalog.rs b/catalog/src/catalog.rs index 23ae1b5..077f4af 100644 --- a/catalog/src/catalog.rs +++ b/catalog/src/catalog.rs @@ -41,6 +41,8 @@ pub struct Config { #[serde(default = "Catalog::default_max_open_topics")] pub max_open_topics: usize, pub max_partition_bytes: ByteSize, + #[serde(default = "Catalog::default_max_active_partitions")] + pub max_active_partitions: usize, } impl Default for Config { @@ -54,6 +56,7 @@ impl Default for Config { storage: Default::default(), max_open_topics: Catalog::default_max_open_topics(), max_partition_bytes: ByteSize::mib(3500), + max_active_partitions: Catalog::default_max_active_partitions(), } } } @@ -320,18 +323,22 @@ impl Catalog { } let mut bytes = 0; + // Keyed on (end time, topic, partition) so partitions that share an end + // time stay distinct entries rather than colliding (which would hide + // them from both the byte and active-count limits below). let mut ages = BTreeMap::default(); for (topic_name, topic) in topics.iter() { for (partition_name, data) in topic.active_data().await { - ages.insert(*data.time.end(), (topic_name.clone(), partition_name)); + ages.insert((*data.time.end(), topic_name.clone(), partition_name), ()); bytes += data.size; } } let max_bytes = self.config.max_partition_bytes.as_u64() as usize; - trace!(bytes, max_bytes); - while bytes > max_bytes { - let Some((time, (topic_name, partition_name))) = ages.pop_first() else { + let max_active = self.config.max_active_partitions; + trace!(bytes, max_bytes, active = ages.len(), max_active); + while bytes > max_bytes || ages.len() > max_active { + let Some(((time, topic_name, partition_name), ())) = ages.pop_first() else { error!("ran out of topics while trying to prune"); return; }; @@ -343,7 +350,11 @@ impl Catalog { }; let age = Utc::now().signed_duration_since(time).to_std(); - info!("closing {topic_name}/{partition_name} (age {age:?}, {bytes} > {max_bytes})"); + info!( + "closing {topic_name}/{partition_name} (age {age:?}, {bytes} > {max_bytes} bytes, \ + {} > {max_active} active)", + ages.len() + 1 + ); let Some(data) = topic.close_partition(&partition_name).await else { error!("invalid partition {partition_name}"); continue; @@ -457,6 +468,12 @@ impl Catalog { 128 } + /// Default number of active (in-memory, writable) partitions to keep open + /// across the catalog. + pub fn default_max_active_partitions() -> usize { + 256 + } + pub async fn close(self) { let mut state = self.state.write().await; @@ -860,4 +877,45 @@ mod test { Ok(()) } + + #[test(tokio::test)] + async fn test_partition_active_count_limit() -> Result<()> { + // High byte limit so only the count limit is exercised. + let (_root, catalog) = catalog_config(Config { + max_active_partitions: 2, + ..Default::default() + }) + .await; + + let time = Utc::now() + .checked_sub_signed(TimeDelta::try_seconds(10).unwrap()) + .unwrap() + .with_nanosecond(0) + .unwrap(); + + let record = Record { + time, + message: "hello".bytes().collect(), + }; + + // Three distinct topics, each with one active partition. + for ix in 0..3 { + let name = format!("topic-{ix}"); + let topic = catalog.get_topic(&name).await; + let insert = topic + .extend_records("default", slice::from_ref(&record)) + .await?; + topic + .ensure_index("default", RecordIndex(insert.end.0 - 1)) + .await?; + } + + assert_eq!(catalog.active_partitions().await, 3); + + // Pruning closes the oldest active partition down to the count limit. + catalog.prune_topics().await; + assert_eq!(catalog.active_partitions().await, 2); + + Ok(()) + } } From 2ad79e9c214ffc3c34732b43b6e26918758f7dad Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 26 Jun 2026 19:52:50 +0000 Subject: [PATCH 2/2] Treat non-resident partitions' segments as sealed in reconcile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconcile classified segments using the in-memory `sealed_ix` watermark, reading it via `get_topic`/`get_partition` — which lazily *loads* the partition. A freshly loaded partition reports `sealed_ix == None`, so every one of its on-disk segments fell into the informational "active" bucket. This inflated the reported `active_segments` count: each cold or evicted partition contributed all of its segments, even though a closed partition's tail is immutable (a reopen begins a brand-new segment at `max_segment + 1`, so nothing ever appends to the persisted tail again). Classify each partition without loading it: a non-resident partition is quiescent and all its persisted segments are sealed, while a resident partition keeps the conservative watermark semantics so its live active tail (and any in-flight roll) still lands in the active bucket. Add a non-loading `Catalog::resident_sealed_ix` / `Topic::resident_sealed_ix` lookup and a `SealStatus` classification for the pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dcvvv5f1pihsE6gDaETRid --- catalog/src/catalog.rs | 19 ++++++ catalog/src/reconcile.rs | 135 ++++++++++++++++++++++++++++++++------- catalog/src/topic.rs | 17 +++++ 3 files changed, 148 insertions(+), 23 deletions(-) diff --git a/catalog/src/catalog.rs b/catalog/src/catalog.rs index 077f4af..0963623 100644 --- a/catalog/src/catalog.rs +++ b/catalog/src/catalog.rs @@ -24,6 +24,7 @@ use crate::manifest::Manifest; use crate::manifest::Scope; use crate::partition; use crate::reconcile::ReconcileReport; +use crate::slog::SegmentIndex; use crate::storage::{self, DiskMonitor}; use crate::topic::Topic; @@ -429,6 +430,24 @@ impl Catalog { } } + /// Read a partition's `sealed_ix` watermark *without* loading the topic or + /// partition into memory. + /// + /// Returns `None` when the partition (or its topic) is not currently + /// resident: a non-resident partition is quiescent — no writer can extend + /// its tail, and a future reopen begins a brand-new segment — so every + /// persisted segment is immutable. Returns `Some(watermark)` when resident, + /// where `watermark` is the in-memory `sealed_ix` (segments at or below it + /// are durably sealed). + pub async fn resident_sealed_ix( + &self, + topic: &str, + partition: &str, + ) -> Option> { + let state = self.state.read().await; + state.topics.get(topic)?.resident_sealed_ix(partition).await + } + /// Returns true if the Catalog is not accepting log writes. pub fn is_readonly(&self) -> bool { self.disk_monitor.is_readonly() diff --git a/catalog/src/reconcile.rs b/catalog/src/reconcile.rs index a916bc9..d8ab3c7 100644 --- a/catalog/src/reconcile.rs +++ b/catalog/src/reconcile.rs @@ -33,13 +33,30 @@ use crate::partition::Partition; use crate::slog::Slog; use crate::topic::Topic; -/// Whether a segment is sealed relative to a partition's `sealed_ix` watermark. +/// How a partition's segments are classified during a reconcile pass. +enum SealStatus { + /// The partition is resident in memory. Segments at or below the + /// `sealed_ix` watermark are durably sealed; anything above it is the live + /// active tail. A `None` watermark means no segment has sealed durably yet, + /// so every segment is the active tail. + Resident(Option), + /// The partition is not resident in memory. No writer can extend any of its + /// segments, and a reopen would begin a fresh segment, so every persisted + /// segment is immutable — treat them all as sealed. + Quiescent, +} + +/// Whether a segment is sealed (immutable on disk) for this pass. /// -/// A segment is sealed only when the partition has a watermark (`Some`) and the -/// segment index is at or below it. When the watermark is `None` (no segment has -/// sealed durably yet) every segment is treated as active. -fn is_sealed(sealed_ix: Option, index: SegmentIndex) -> bool { - sealed_ix.is_some_and(|watermark| index <= watermark) +/// A non-resident partition is quiescent, so all of its segments are sealed. +/// A resident partition is sealed only at or below its `sealed_ix` watermark; +/// the live active tail above it (or every segment, when the watermark is +/// `None`) is treated as active. +fn is_sealed(status: &SealStatus, index: SegmentIndex) -> bool { + match status { + SealStatus::Quiescent => true, + SealStatus::Resident(watermark) => watermark.is_some_and(|w| index <= w), + } } /// Parse the partition name and segment index out of a segment file *stem* of @@ -750,16 +767,24 @@ impl ReconcileJob { topic_name, partition_name, topic_path ); - // Read the sealed watermark once for this partition pass. Segments at or - // below it are durable and run through the strict sealed-diff pipeline; - // anything above it (or every segment, if the partition has not sealed - // one yet) is the currently active tail and goes to the informational - // active bucket instead. The watermark never moves backward, so this - // single read is a stable basis for the whole pass. - let sealed_ix = { - let topic = self.catalog.get_topic(topic_name).await; - let partition = topic.get_partition(partition_name).await; - partition.sealed_ix() + // Classify this partition's segments once for the whole pass, *without* + // loading it into memory. A partition that is not resident is quiescent: + // no writer can extend its tail, and a reopen would start a brand-new + // segment, so every persisted segment is immutable and runs through the + // strict sealed-diff pipeline. (Force-loading it would reset the + // in-memory watermark to `None` and wrongly bucket every segment as + // active.) A resident partition keeps the conservative watermark + // semantics: segments at or below `sealed_ix` are durable; anything + // above it is the live active tail and goes to the informational active + // bucket instead. The watermark never moves backward, so this single + // read is a stable basis for the whole pass. + let seal_status = match self + .catalog + .resident_sealed_ix(topic_name, partition_name) + .await + { + Some(watermark) => SealStatus::Resident(watermark), + None => SealStatus::Quiescent, }; // Create sets to track files @@ -775,11 +800,11 @@ impl ReconcileJob { let segments: Vec = segments_stream.collect().await; if let Some((start, end)) = segments.first().zip(segments.last()) { debug!( - "Fetched {} segments: {} ..= {} (sealed_ix={:?})", + "Fetched {} segments: {} ..= {} (resident={})", segments.len(), start.index.0, end.index.0, - sealed_ix, + matches!(seal_status, SealStatus::Resident(_)), ); } else { debug!("Found no segments") @@ -799,11 +824,10 @@ impl ReconcileJob { // detection phase does not false-positive on active segment files. tracked_files.insert(segment_path.clone()); - // A segment is "sealed" only when the partition has a watermark and - // this segment's index is at or below it. Everything else is the - // active tail (this includes the all-segments-active case when the - // partition has never sealed a segment, i.e. sealed_ix is None). - if is_sealed(sealed_ix, segment.index) { + // Sealed segments run the strict diff; the live active tail of a + // resident partition goes to the informational active bucket. See + // `SealStatus` for how non-resident partitions are handled. + if is_sealed(&seal_status, segment.index) { self.process_sealed_segment( &partition_id, &segment, @@ -1660,6 +1684,71 @@ mod tests { Ok(()) } + /// A partition evicted from memory (as the active-partition limit does) is + /// quiescent: its persisted tail can never be written to again — a reopen + /// starts a fresh segment — so reconcile must treat that tail as sealed + /// rather than force-loading the partition and counting every segment as + /// active. + #[test_log::test(tokio::test)] + async fn test_evicted_partition_segments_are_sealed() -> Result<()> { + // High roll threshold: while resident the single segment stays active + // and unsealed (sealed_ix == None) — the case that previously inflated + // the active bucket once the partition was force-loaded. + let (_tmpdir, catalog) = create_test_catalog_rolling(1000).await; + let topic_name = "evicted"; + let partition_name = "p0"; + + let topic = catalog.get_topic(topic_name).await; + topic + .extend_records(partition_name, &test_records(&["a", "b", "c"])) + .await?; + drop(topic); + + // Persist the active segment to the manifest without sealing it. + catalog.checkpoint().await; + catalog + .get_topic(topic_name) + .await + .ensure_index(partition_name, RecordIndex(3)) + .await?; + + // Evict the partition from memory, exactly as the active-partition limit + // does. Its tail is now immutable on disk. + let evicted = catalog + .get_topic(topic_name) + .await + .close_partition(partition_name) + .await; + assert!(evicted.is_some(), "partition should have been resident"); + assert_eq!( + catalog.resident_sealed_ix(topic_name, partition_name).await, + None, + "partition should no longer be resident" + ); + + let config = ReconcileConfig { + track_files: true, + ..Default::default() + }; + let mut reconciler = ReconcileJob::with_config(catalog.clone(), config); + assert!(reconciler.run(Some(100)).await?); + + let report = reconciler.report(); + // The evicted tail must NOT be counted as active... + assert!( + active_entries(report, topic_name, partition_name).is_empty(), + "evicted partition's segments must not be in the active bucket" + ); + // ...and reconcile must not have force-loaded the partition back in. + assert_eq!( + catalog.resident_sealed_ix(topic_name, partition_name).await, + None, + "reconcile must not force-load a non-resident partition" + ); + + Ok(()) + } + /// The UpdateManifestSizes fix must never apply to an active segment, even /// when its on-disk size is corrupted. The drift is reported (non-zero /// delta) but the manifest is left untouched. diff --git a/catalog/src/topic.rs b/catalog/src/topic.rs index 55289ec..9576867 100644 --- a/catalog/src/topic.rs +++ b/catalog/src/topic.rs @@ -15,6 +15,7 @@ use crate::manifest::{Manifest, PartitionId, Scope, SegmentData}; use crate::partition::Config as PartitionConfig; use crate::partition::Partition; +use crate::slog::SegmentIndex; use crate::data::index::{Ordering, RecordIndex}; use crate::transport::{PartitionFilter, PartitionSelector, SchemaChunk, TopicIterator}; @@ -142,6 +143,22 @@ impl Topic { .await } + /// Read a partition's `sealed_ix` watermark *without* loading it into + /// memory. Returns `None` when the partition is not currently resident + /// (so reconcile can treat its on-disk segments as immutable rather than + /// forcing a load, which would reset the in-memory watermark to `None`), + /// or `Some(watermark)` when resident. + pub(crate) async fn resident_sealed_ix( + &self, + partition_name: &str, + ) -> Option> { + self.partitions + .read() + .await + .get(partition_name) + .map(|partition| partition.sealed_ix()) + } + pub async fn get_partition(&self, partition_name: &str) -> RwLockReadGuard<'_, Partition> { let partitions = self.partitions.read().await; let current_partition = RwLockReadGuard::try_map(partitions, |map| map.get(partition_name));