[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_179":-5,"_182":183,"_184":185},"routes/shell",{"_41":42,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":32,"_44":32,"_45":32,"_46":32,"_47":32,"_48":32,"_49":50,"_51":50,"_52":32,"_53":-7,"_54":-7,"_55":56,"_57":58,"_59":60,"_61":-5,"_62":63,"_64":65,"_66":67,"_68":69},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":22,"_23":24,"_25":-5,"_26":27,"_28":29,"_29":30,"_31":32,"_33":14},"ref","main","path","crates/ltx/src/replica_compactor.rs","oid","37cd75e07049607749786089a314bd6491f8b456","size",16664,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! Additive LTX compaction through a [`ReplicaClient`].\n//!\n//! This module ports the storage-independent part of Litestream v0.5.16's\n//! `Compactor`. It creates a destination object but never deletes a source.\n\nuse crate::client::ReplicaClient;\nuse crate::compaction_level::SNAPSHOT_LEVEL;\nuse crate::compactor::Compactor;\nuse crate::error::{Error, Result};\nuse crate::ltx::{FileInfo, HEADER_FLAG_NO_CHECKSUM};\nuse crate::ltx_file_path;\nuse crate::LtxHost;\nuse crate::TXID;\nuse std::io::{BufReader, BufWriter, Cursor, Read, Write};\nuse std::path::Path;\nuse std::path::PathBuf;\n\n/// The immutable object and source volume from one compaction attempt.\n#[derive(Debug, Clone, PartialEq, Eq)]\npub struct CompactionOutput {\n /// The published destination object.\n pub info: FileInfo,\n /// The number of source objects in the merge.\n pub input_files: usize,\n /// The sum of the source object sizes.\n pub input_bytes: u64,\n /// The number of source objects read from the local LTX directory.\n pub local_input_files: usize,\n}\n\n/// Destination coverage proved by the same attempt that selects its output.\n#[derive(Debug, Clone, PartialEq, Eq)]\npub struct CompactionResult {\n pub covered_txid: TXID,\n pub output: Option,\n}\n\n/// Compacts LTX objects between two adjacent replica levels.\npub struct ReplicaCompactor<'a, C> {\n client: &'a C,\n verify: bool,\n max_files: usize,\n max_input_bytes: u64,\n /// The epoch's first txid. An epoch that continues a chain it paged in\n /// starts at its cut, not at 1, so an empty destination level continues\n /// from here rather than from the first txid that never existed here.\n base: TXID,\n local_path: Option,\n host: LtxHost,\n}\n\nimpl<'a, C: ReplicaClient> ReplicaCompactor<'a, C> {\n pub fn new(client: &'a C) -> Self {\n Self {\n client,\n verify: false,\n max_files: usize::MAX,\n max_input_bytes: u64::MAX,\n base: TXID(1),\n local_path: None,\n host: LtxHost::default(),\n }\n }\n\n /// The txid the epoch's chain starts at (see the `base` field).\n pub fn with_base(mut self, base: TXID) -> Self {\n self.base = base;\n self\n }\n\n /// Checks destination continuity again after publication. Every attempt\n /// checks the existing destination before it selects a source.\n pub fn with_verification(mut self, verify: bool) -> Self {\n self.verify = verify;\n self\n }\n\n /// Limits the source count and buffered input bytes. An indivisible source\n /// above the byte budget compacts alone through scratch files.\n pub fn with_limits(mut self, max_files: usize, max_input_bytes: u64) -> Self {\n self.max_files = max_files;\n self.max_input_bytes = max_input_bytes;\n self\n }\n\n /// Uses the local LTX directory before it reads an object from the replica.\n pub fn with_local_path(mut self, path: impl AsRef) -> Self {\n self.local_path = Some(path.as_ref().to_path_buf());\n self\n }\n\n /// Uses an injected clock and executor host.\n pub fn with_host(mut self, host: LtxHost) -> Self {\n self.host = host;\n self\n }\n\n /// Compacts one new object prefix from `destination_level - 1`.\n ///\n /// The method returns `Ok(None)` when the destination already covers every\n /// available source object. It publishes one immutable destination object\n /// and leaves every source object intact.\n pub async fn compact(&self, destination_level: i32) -> Result> {\n Ok(self.compact_with_progress(destination_level).await?.output)\n }\n\n /// Returns the proved destination bound even when no source is available.\n /// An empty listing does not prove coverage of a concurrently durable tail.\n pub async fn compact_with_progress(&self, destination_level: i32) -> Result {\n if !(1..SNAPSHOT_LEVEL).contains(&destination_level) {\n return Err(invalid(\"the destination compaction level is invalid\"));\n }\n if self.max_files == 0 || self.max_input_bytes == 0 {\n return Err(invalid(\"the compaction limits must be positive\"));\n }\n\n let destination = self.client.ltx_files(destination_level, TXID(0)).await?;\n self.verify_files(&destination)?;\n let previous_max = destination\n .iter()\n .map(|file| file.max_txid)\n .max()\n .unwrap_or(TXID(0));\n let Some(next) = previous_max.0.checked_add(1) else {\n return Ok(CompactionResult {\n covered_txid: previous_max,\n output: None,\n });\n };\n let seek = TXID(next.max(self.base.0));\n let source_level = destination_level - 1;\n let mut available = self\n .client\n .ltx_files_bounded(source_level, seek, self.max_files)\n .await?;\n // The listing offset compares minimum txids. A direct drain can start\n // before `seek` and still contain the only copy of its uncovered tail.\n // Keep the bounded listing for ordinary rows, but search behind the\n // offset before reporting a gap or no work.\n if available.first().is_none_or(|file| file.min_txid != seek) {\n available = self.client.ltx_files(source_level, TXID(0)).await?;\n }\n available.sort_by_key(|file| (file.min_txid, std::cmp::Reverse(file.max_txid)));\n let mut source = Vec::new();\n let mut input_bytes = 0u64;\n let mut next = seek;\n for file in available {\n if file.max_txid < next {\n continue;\n }\n if file.min_txid > next {\n if source.is_empty() {\n return Err(invalid(\n \"the compaction source does not continue the destination level\",\n ));\n }\n break;\n }\n let size = u64::try_from(file.size)\n .map_err(|_| invalid(\"a compaction source has a negative size\"))?;\n if source.len() == self.max_files {\n break;\n }\n let total = input_bytes\n .checked_add(size)\n .ok_or_else(|| invalid(\"the compaction source size overflows\"))?;\n if total > self.max_input_bytes {\n if source.is_empty() {\n let output = self\n .compact_large_source(destination_level, seek, file)\n .await?;\n return Ok(CompactionResult {\n covered_txid: output.info.max_txid,\n output: Some(output),\n });\n }\n break;\n }\n input_bytes = total;\n let reaches_end = file.max_txid == TXID(u64::MAX);\n next = TXID(file.max_txid.0.saturating_add(1));\n source.push(file);\n if reaches_end {\n break;\n }\n }\n if source.is_empty() {\n return Ok(CompactionResult {\n covered_txid: previous_max,\n output: None,\n });\n }\n\n let min_txid = seek;\n let max_txid = source\n .iter()\n .map(|file| file.max_txid)\n .max()\n .ok_or_else(|| invalid(\"the compaction source is empty\"))?;\n let ranges: Vec<_> = source\n .iter()\n .map(|file| (file.min_txid, file.max_txid))\n .collect();\n let mut readers = Vec::with_capacity(source.len());\n let mut local_input_files = 0usize;\n for file in &source {\n let bytes = match &self.local_path {\n Some(path) => {\n let filename = ltx_file_path(\n &path.to_string_lossy(),\n file.level as u32,\n file.min_txid,\n file.max_txid,\n );\n match self.host.read_file(filename).await {\n Ok(bytes) => {\n local_input_files += 1;\n bytes\n }\n Err(error) if error.kind() == std::io::ErrorKind::NotFound => {\n self.client\n .open_ltx_file(file.level, file.min_txid, file.max_txid)\n .await?\n }\n Err(error) => return Err(Error::Io(error)),\n }\n }\n None => {\n self.client\n .open_ltx_file(file.level, file.min_txid, file.max_txid)\n .await?\n }\n };\n readers.push(Cursor::new(bytes));\n }\n\n // The merge is pure CPU over as much as 64 MiB and must not hold a\n // runtime worker: a restart drain runs rounds back-to-back, and on a\n // 4-vCPU node two pegged workers starved co-owned cells' durable\n // writes (2026-08-12 fleet roll).\n let (header, output) = self\n .host\n .run_blocking(move || {\n let mut compactor = Compactor::new(Vec::new(), readers);\n compactor.header_flags = HEADER_FLAG_NO_CHECKSUM;\n compactor.compact_from(min_txid, &ranges)?;\n Ok::<_, Error>((compactor.header(), compactor.into_writer()))\n })\n .await\n .map_err(|_| invalid(\"the compaction merge task panicked\"))??;\n if header.min_txid != min_txid || header.max_txid != max_txid {\n return Err(invalid(\n \"a compaction source key does not match its LTX header\",\n ));\n }\n let info = self\n .client\n .write_ltx_file(destination_level, min_txid, max_txid, &output)\n .await?;\n\n if self.verify {\n self.verify_level(destination_level).await?;\n }\n Ok(CompactionResult {\n covered_txid: info.max_txid,\n output: Some(CompactionOutput {\n info,\n input_files: source.len(),\n input_bytes,\n local_input_files,\n }),\n })\n }\n\n // A direct drain can produce one object larger than the entire input\n // budget. Deferring it forever stalls the level; buffering it defeats the\n // node's memory limit. Spool remote bytes and the output to anonymous files\n // and merge one page at a time. The codec retains only its page index.\n async fn compact_large_source(\n &self,\n destination_level: i32,\n min: TXID,\n source: FileInfo,\n ) -> Result {\n let host = self.host.clone();\n let local = self.local_path.as_ref().map(|path| {\n PathBuf::from(ltx_file_path(\n &path.to_string_lossy(),\n source.level as u32,\n source.min_txid,\n source.max_txid,\n ))\n });\n let (mut input, local_input_files) = self\n .host\n .run_blocking(move || match local {\n Some(path) => match host.open(&path) {\n Ok(file) => Ok((Some(file), 1)),\n Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok((None, 0)),\n Err(error) => Err(error),\n },\n None => Ok((None, 0)),\n })\n .await\n .map_err(|_| invalid(\"the compaction file task panicked\"))??;\n if input.is_none() {\n let mut scratch = self.scratch_file().await?;\n let mut offset = 0;\n while offset < source.size as u64 {\n let len = (source.size as u64 - offset).min(1 << 20);\n let bytes = self\n .client\n .read_range(source.level, source.min_txid, source.max_txid, offset, len)\n .await?;\n if bytes.len() as u64 != len {\n return Err(invalid(\"a compaction source range is short\"));\n }\n scratch = self\n .host\n .run_blocking(move || {\n scratch.write_all(&bytes)?;\n Ok::<_, Error>(scratch)\n })\n .await\n .map_err(|_| invalid(\"the compaction file task panicked\"))??;\n offset += len;\n }\n input = Some(scratch);\n }\n let input = input.ok_or_else(|| invalid(\"the compaction source is empty\"))?;\n let output = self.scratch_file().await?;\n let expected = (source.min_txid, source.max_txid);\n let output = self\n .host\n .run_blocking(move || {\n let reader = ScratchReader::new(input)?;\n if reader.len != source.size as u64 {\n return Err(invalid(\"a compaction source size changed\"));\n }\n // Buffered adapters avoid a syscall for every codec field.\n let mut compactor = Compactor::new(\n BufWriter::new(ScratchWriter(output)),\n vec![BufReader::new(reader)],\n );\n compactor.header_flags = HEADER_FLAG_NO_CHECKSUM;\n compactor.compact_from(min, &[expected])?;\n let output = compactor\n .into_writer()\n .into_inner()\n .map_err(|error| Error::Io(error.into_error()))?;\n Ok::<_, Error>(output.0)\n })\n .await\n .map_err(|_| invalid(\"the compaction merge task panicked\"))??;\n let info = self\n .client\n .write_ltx_file_from_file(\n destination_level,\n min,\n source.max_txid,\n output,\n self.host.clone(),\n )\n .await?;\n if self.verify {\n self.verify_level(destination_level).await?;\n }\n Ok(CompactionOutput {\n info,\n input_files: 1,\n input_bytes: source.size as u64,\n local_input_files,\n })\n }\n\n async fn scratch_file(&self) -> Result {\n let filesystem = self.host.filesystem();\n let directory = self.local_path.clone();\n self.host\n .run_blocking(move || filesystem.temporary_file(directory.as_deref()))\n .await\n .map_err(|_| invalid(\"the compaction file task panicked\"))?\n .map_err(Error::Io)\n }\n\n /// Verifies that a destination level has neither gaps nor overlaps.\n pub async fn verify_level(&self, level: i32) -> Result<()> {\n let files = self.client.ltx_files(level, TXID(0)).await?;\n self.verify_files(&files)\n }\n\n fn verify_files(&self, files: &[FileInfo]) -> Result<()> {\n if files.first().is_some_and(|file| file.min_txid != self.base)\n || files.iter().any(|file| file.min_txid > file.max_txid)\n {\n return Err(invalid(\"the compaction level is not contiguous\"));\n }\n for pair in files.windows(2) {\n if pair[0].max_txid.0.checked_add(1) != Some(pair[1].min_txid.0) {\n return Err(invalid(\"the compaction level is not contiguous\"));\n }\n }\n Ok(())\n }\n}\n\nfn invalid(message: &'static str) -> Error {\n Error::Other(message.into())\n}\n\nstruct ScratchReader {\n file: crate::host::HostFile,\n offset: u64,\n len: u64,\n}\n\nimpl ScratchReader {\n fn new(mut file: crate::host::HostFile) -> std::io::Result {\n let len = file.file_len()?;\n Ok(Self {\n file,\n offset: 0,\n len,\n })\n }\n}\n\nimpl Read for ScratchReader {\n fn read(&mut self, bytes: &mut [u8]) -> std::io::Result {\n let len = (self.len - self.offset).min(bytes.len() as u64) as usize;\n if len == 0 {\n return Ok(0);\n }\n bytes[..len].copy_from_slice(&self.file.read_exact_at(self.offset, len)?);\n self.offset += len as u64;\n Ok(len)\n }\n}\n\nstruct ScratchWriter(crate::host::HostFile);\n\nimpl Write for ScratchWriter {\n fn write(&mut self, bytes: &[u8]) -> std::io::Result {\n self.0.write_all(bytes)?;\n Ok(bytes.len())\n }\n fn flush(&mut self) -> std::io::Result<()> {\n Ok(())\n }\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[134,135,136,137,138,139,140,141,142,143,144,145,146],"navigation",{"_55":56,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_43":32,"_89":14,"_90":32,"_49":50,"_51":50},"branches",[14],"tags",[70,71,72,73,74,75,76,77,78,79,80,81,82],"cloneUrl","https://git.nexthub.sh/denoland/celld.git","v0.0.1","v0.0.2","v0.1.0","v0.2.0","v0.2.1","v0.3.0","v0.4.0","v0.4.1","v0.5.0","v0.5.1","v0.6.0","v0.6.1","v0.6.2","description","self-hosted, distributed Durable Objects","owner","denoland","section","","head","empty","pins",[116,117,118],"documents",{"_111":112,"_113":114},"code",{"_108":109,"_110":104},"automations",{"_106":50,"_107":-5},"delivery",{"_103":104,"_105":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_115":114},"total",38,"issue",{"_119":110,"_121":132,"_15":110,"_123":133,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_53":125,"_126":32,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_53":125,"_126":32,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_53":125,"_126":32,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_53":125,"_126":32,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_53":32,"_126":32,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_53":32,"_126":32,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_53":32,"_126":32,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_53":32,"_126":32,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_53":32,"_126":32,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_53":32,"_126":32,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_53":32,"_126":32,"_127":32},{"_119":153,"_121":154,"_15":153,"_123":155,"_53":32,"_126":32,"_127":32},{"_119":150,"_121":151,"_15":150,"_123":152,"_53":32,"_126":32,"_127":32},{"_119":147,"_121":148,"_15":147,"_123":149,"_53":32,"_126":32,"_127":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794462225929,"sidebar","expanded"]