Add startup retention reclaim to recover from a full disk

.github/workflows/rust.yml

@@ -14,7 +14,7 @@ on:
 env:
   CARGO_TERM_COLOR: always
   REGISTRY_IMAGE: ghcr.io/wallaroolabs/plateau
-  RUST_VERSION: 1.84.1
+  RUST_VERSION: 1.90.0
 
 jobs:
   check:

Makefile

@@ -6,7 +6,7 @@ PLATEAU_IMAGE  			  = $(IMAGE_ROOT)/plateau
 DOCKER                    = docker buildx
 REVISION                  = $(shell git describe --match="" --always --abbrev=40 --dirty)
 TIME                      = $(shell date -u +"%Y-%m-%dT%H:%M:%SZ")
-RUST_VERSION              = 1.84.1
+RUST_VERSION              = 1.90.0
 
 test:
 	cargo test --workspace --features batch,polars,replicate -- --nocapture

catalog/src/catalog.rs

@@ -233,6 +233,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;
@@ -671,6 +701,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 {

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)
 }
@@ -401,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;
     }
@@ -592,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",

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.89.0"
+channel = "1.90.0"
 components = ["clippy", "rustfmt"]

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;
 
     {