Add per-catalog active partition count limit

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;
 
@@ -41,6 +42,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 +57,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 +324,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 +351,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;
@@ -418,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<Option<SegmentIndex>> {
+        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()
@@ -457,6 +487,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 +896,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(())
+    }
 }

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<SegmentIndex>),
+    /// 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<SegmentIndex>, 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<SegmentData> = 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.

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<Option<SegmentIndex>> {
+        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));