From 49dfdce1069a6a92ce19fa5bad88436d9508f88b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 26 Jun 2026 12:34:33 +0000 Subject: [PATCH 1/2] Add startup retention reclaim to recover from a full disk Ongoing retention removes a segment's manifest entry before destroying its backing data. When the process starts against an already-full disk that order cannot make progress: deleting the manifest row is a SQLite write that itself needs free space to journal, so it panics with "database or disk is full" before any space is reclaimed. Add Catalog::reclaim, a one-shot check run before the steady-state checkpoint/retention loops. When over the limit it removes the oldest segment data-first (destroy backing data, then remove the manifest entry), freeing space so the regular retention loop can take over. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Jkk2NoeuqoKS8XsxZSu7gq --- catalog/src/catalog.rs | 70 ++++++++++++++++++++++++++++++++++++++++ catalog/src/partition.rs | 56 ++++++++++++++++++++++++++++---- server/src/lib.rs | 6 ++++ 3 files changed, 125 insertions(+), 7 deletions(-) diff --git a/catalog/src/catalog.rs b/catalog/src/catalog.rs index bc58fee..23ae1b5 100644 --- a/catalog/src/catalog.rs +++ b/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 { diff --git a/catalog/src/partition.rs b/catalog/src/partition.rs index c6fa8cb..8c56737 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(a: Range, b: Range) -> Range { 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", diff --git a/server/src/lib.rs b/server/src/lib.rs index 93aabfa..f7cd4d9 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; { From 620e7592950e1f0bd51c07875963e06452704e05 Mon Sep 17 00:00:00 2001 From: Cyril Plisko Date: Tue, 30 Jun 2026 19:04:49 +0300 Subject: [PATCH 2/2] Align Rust version with platform --- .github/workflows/rust.yml | 2 +- Makefile | 2 +- rust-toolchain.toml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index f7ef4bf..7cb28ec 100644 --- a/.github/workflows/rust.yml +++ b/.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: diff --git a/Makefile b/Makefile index 91d6ced..54e9c6b 100644 --- a/Makefile +++ b/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 diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 1be126d..f35f369 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,3 @@ [toolchain] -channel = "1.89.0" +channel = "1.90.0" components = ["clippy", "rustfmt"]