diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 54d23ff..9c97c00 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -97,7 +97,7 @@ jobs:
 
   build-multi-arm64:
     runs-on:
-      - buildjet-4vcpu-ubuntu-2204-arm
+      - 4x16g-arm
     steps:
       - name: Prepare
         run: |
diff --git a/catalog/src/catalog.rs b/catalog/src/catalog.rs
index bc58fee..a90a79f 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;
 
@@ -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(),
         }
     }
 }
@@ -69,8 +73,22 @@ impl CatalogRetention {
         match self {
             Self::Fixed(r) => Ok(r.clone()),
             Self::RootMountTotal => {
-                let mount_size = storage::path_mount_stat(root.to_owned()).await?.total;
-                info!(?root, %mount_size, "resolved size for retention");
+                let stat = storage::path_mount_stat(root.to_owned()).await?;
+                // `total` includes blocks reserved for root (typically ~5% on
+                // ext4), which are never usable by plateau. Use avail + used
+                // (i.e. total - reserved) so retention fires before the storage
+                // monitor's min_available threshold.
+                let used = stat.total.0.saturating_sub(stat.free.0);
+                let mount_size = ByteSize(stat.avail.0.saturating_add(used));
+                info!(
+                    ?root,
+                    total = %stat.total,
+                    free = %stat.free,
+                    avail = %stat.avail,
+                    %used,
+                    %mount_size,
+                    "resolved size for retention"
+                );
 
                 Ok(Retention {
                     max_bytes: mount_size,
@@ -233,6 +251,36 @@ impl Catalog {
             .await;
     }
 
+    /// One-shot retention reclaim, run once at startup *before* the
+    /// steady-state checkpoint/retention loops begin.
+    ///
+    /// [retain] removes a segment's manifest entry before destroying its
+    /// backing data. When the process starts against a disk that is already
+    /// full that order cannot make progress: deleting the manifest row is a
+    /// SQLite write that itself needs free space to journal, so it fails
+    /// ("database or disk is full") before any data is freed.
+    ///
+    /// To break out of that state, perform a single retention check and, if we
+    /// are over the limit, destroy the backing data of the oldest segment
+    /// *first* and only then remove its manifest entry. Reclaiming that space
+    /// leaves room for the manifest update to succeed, after which the regular
+    /// retention loop can make progress on its own.
+    pub async fn reclaim(&self) {
+        if !self.over_retention_limit().await {
+            return;
+        }
+
+        let Some(oldest) = self.manifest.get_oldest_segment(None).await else {
+            error!("over retention limit at startup but no segment to reclaim");
+            return;
+        };
+
+        info!("startup reclaim: removing oldest segment {:?}", oldest);
+        let topic = self.get_topic(oldest.topic()).await;
+        let partition = topic.get_partition(oldest.partition()).await;
+        partition.reclaim_oldest().await;
+    }
+
     pub async fn retain(&self) {
         trace!("begin global retention check");
         self.gauge_topics().await;
@@ -290,18 +338,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;
             };
@@ -313,7 +365,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;
@@ -388,6 +444,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()
@@ -427,6 +501,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;
 
@@ -671,6 +751,46 @@ mod test {
         Ok(())
     }
 
+    #[test_log::test(tokio::test)]
+    async fn test_reclaim() -> Result<()> {
+        let (_root, mut catalog) = catalog().await;
+        catalog.config.retain.max_bytes = ByteSize::b(8000 + catalog.manifest.db_bytes() as u64);
+        catalog.config.headroom = ByteSize::b(0);
+
+        let data = "x".to_string().repeat(500);
+
+        // build several sealed segments so there is a clear "oldest" to reclaim
+        let records = build_records((0..10).map(|_| (0, data.clone())));
+        let topic = catalog.get_topic("topic").await;
+        let partition = topic.get_partition("default").await;
+        for _ in 0..6 {
+            partition.extend_records(&records).await?;
+            partition.compact().await;
+        }
+        partition.extend_records(&records).await?;
+
+        let oldest_before = catalog.manifest.get_oldest_segment(None).await.unwrap();
+        let size_before = catalog.byte_size().await;
+        assert!(size_before > catalog.total_byte_limit());
+
+        // reclaim removes exactly one (the oldest) segment, freeing space so the
+        // steady-state loop can take over.
+        catalog.reclaim().await;
+
+        let size_after = catalog.byte_size().await;
+        assert!(size_after < size_before);
+
+        let oldest_after = catalog.manifest.get_oldest_segment(None).await.unwrap();
+        assert_eq!(oldest_after.partition(), oldest_before.partition());
+        assert!(oldest_after.segment > oldest_before.segment);
+
+        // a second reclaim continues making progress one segment at a time
+        catalog.reclaim().await;
+        assert!(catalog.byte_size().await < size_after);
+
+        Ok(())
+    }
+
     #[test_log::test(tokio::test)]
     async fn test_max_open_topics() -> Result<()> {
         let (_root, catalog) = catalog_config(Config {
@@ -790,4 +910,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(())
+    }
 }
diff --git a/catalog/src/partition.rs b/catalog/src/partition.rs
index 4a6c56a..060aa25 100644
--- a/catalog/src/partition.rs
+++ b/catalog/src/partition.rs
@@ -79,6 +79,19 @@ pub struct State {
     fin: oneshot::Receiver<()>,
 }
 
+/// Order in which a segment's manifest entry and backing data are removed.
+#[derive(Clone, Copy, Debug)]
+enum RemovalOrder {
+    /// Remove the manifest entry first, then destroy the backing data. The
+    /// steady-state order used by ongoing retention.
+    ManifestFirst,
+    /// Destroy the backing data first, then remove the manifest entry. Used by
+    /// the startup reclaim when the disk is already full: the manifest update
+    /// is a SQLite write that itself needs free space, so removing it first
+    /// fails ("database or disk is full") before any space has been reclaimed.
+    DataFirst,
+}
+
 fn merge_ranges<T: Ord>(a: Range<T>, b: Range<T>) -> Range<T> {
     std::cmp::min(a.start, b.start)..std::cmp::max(a.end, b.end)
 }
@@ -111,7 +124,11 @@ impl Partition {
         let (commit_writer, commits) = watch::channel(record);
         let commit_manifest = manifest.clone();
         let commit_id = id.clone();
-        let (sealed_tx, sealed_ix) = watch::channel(None);
+        // All manifest segments up to and including the max are sealed on
+        // attach: find_starting_index always starts the new active segment at
+        // max+1, so the previous max is definitionally closed.
+        let initial_sealed = segment.prev();
+        let (sealed_tx, sealed_ix) = watch::channel(initial_sealed);
         tokio::spawn(async move {
             while let Some(r) = writes.recv().await {
                 trace!("{} checkpoint: {:?}", commit_id, &r);
@@ -397,6 +414,16 @@ impl Partition {
         state.remove_oldest(self).await;
     }
 
+    /// Reclaim the oldest segment by destroying its backing data before
+    /// removing its manifest entry. Used by the startup reclaim pass to make
+    /// progress against a full disk; see [RemovalOrder::DataFirst].
+    pub(crate) async fn reclaim_oldest(&self) {
+        let state = self.state.read().await;
+        state
+            .remove_oldest_in_order(self, RemovalOrder::DataFirst)
+            .await;
+    }
+
     pub(crate) async fn close(self) {
         self.state.into_inner().close().await;
     }
@@ -588,14 +615,33 @@ impl State {
     }
 
     async fn remove_oldest(&self, partition: &Partition) {
+        self.remove_oldest_in_order(partition, RemovalOrder::ManifestFirst)
+            .await;
+    }
+
+    async fn remove_oldest_in_order(&self, partition: &Partition, order: RemovalOrder) {
         if let Some(ix) = partition.manifest.get_min_segment(&partition.id).await {
-            // TODO ensure we handle failure if this call
-            partition
-                .manifest
-                .remove_segment(ix.to_id(&partition.id))
-                .await;
-            // succeeds but this does not complete e.g. due to node failure
-            self.messages.destroy(ix).expect("segment destroyed");
+            match order {
+                RemovalOrder::ManifestFirst => {
+                    // TODO ensure we handle failure if this call
+                    partition
+                        .manifest
+                        .remove_segment(ix.to_id(&partition.id))
+                        .await;
+                    // succeeds but this does not complete e.g. due to node failure
+                    self.messages.destroy(ix).expect("segment destroyed");
+                }
+                RemovalOrder::DataFirst => {
+                    // Free the backing data before touching the manifest so the
+                    // manifest update (which itself needs free space to journal)
+                    // cannot fail before any disk has been reclaimed.
+                    self.messages.destroy(ix).expect("segment destroyed");
+                    partition
+                        .manifest
+                        .remove_segment(ix.to_id(&partition.id))
+                        .await;
+                }
+            }
             info!("retain {}: destroyed {:?}", partition.id, ix);
             counter!(
                 "partition_segments_destroyed",
diff --git a/catalog/src/reconcile.rs b/catalog/src/reconcile.rs
index 7ab30e8..22b775e 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<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
@@ -751,16 +768,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
@@ -776,11 +801,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")
@@ -800,11 +825,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,
@@ -1002,15 +1026,13 @@ impl ReconcileJob {
         {
             if part_path.exists() {
                 tracked_files.insert(part_path.clone());
-                if part_path != segment_file.cache_path() {
-                    match fs::metadata(&part_path).await {
-                        Ok(metadata) => {
-                            total_actual_size += metadata.len() as usize;
-                            debug!("Part {:?} size: {}", part_path, metadata.len());
-                        }
-                        Err(e) => {
-                            warn!("Error getting metadata for part {:?}: {:?}", part_path, e);
-                        }
+                match fs::metadata(&part_path).await {
+                    Ok(metadata) => {
+                        total_actual_size += metadata.len() as usize;
+                        debug!("Part {:?} size: {}", part_path, metadata.len());
+                    }
+                    Err(e) => {
+                        warn!("Error getting metadata for part {:?}: {:?}", part_path, e);
                     }
                 }
             } else {
@@ -1663,6 +1685,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 7fe01f1..1710648 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<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));
diff --git a/data/src/segment.rs b/data/src/segment.rs
index 09fa103..80b0936 100644
--- a/data/src/segment.rs
+++ b/data/src/segment.rs
@@ -221,14 +221,15 @@ impl Segment {
     }
 
     /// Return an estimate of the on-disk size of the corresponding file(s),
-    /// excluding the active chunk cache.
+    /// including the active chunk cache if present.
     pub fn size_estimate(&self) -> Result<usize> {
         let main_size = fs::metadata(&self.path).map(|p| p.len()).unwrap_or(0);
         let part_size: u64 = self
             .parts()
             .map(|part| fs::metadata(part).map(|p| p.len()).unwrap_or(0))
             .sum();
-        Ok(usize::try_from(main_size + part_size)?)
+        let cache_size = fs::metadata(self.cache_path()).map(|p| p.len()).unwrap_or(0);
+        Ok(usize::try_from(main_size + part_size + cache_size)?)
     }
 }
 
@@ -373,9 +374,7 @@ impl Writer {
 
     /// Return an estimate of the on-disk size of the corresponding file(s).
     pub fn size_estimate(&self) -> Result<usize> {
-        let segment_size = self.segment.size_estimate()?;
-        let cache_size = self.cache.size() as usize;
-        Ok(segment_size + cache_size)
+        self.segment.size_estimate()
     }
 
     pub fn close(self) -> Result<usize> {
diff --git a/server/src/lib.rs b/server/src/lib.rs
index ecd2d44..0e65f20 100644
--- a/server/src/lib.rs
+++ b/server/src/lib.rs
@@ -76,6 +76,12 @@ pub async fn task_from_catalog_config(
     config: config::PlateauConfig,
     stop: future::BoxFuture<'_, ()>,
 ) -> bool {
+    // Before the steady-state checkpoint/retention loops start, run a one-shot
+    // reclaim. If the process is starting against a full disk, ongoing
+    // retention (which removes the manifest entry before the backing data)
+    // cannot free space, so reclaim the oldest segment data-first to unblock it.
+    catalog.reclaim().await;
+
     let (addr, end_tx, server) = http::serve(config.clone(), catalog.clone()).await;
 
     {