[{"_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/client/bundle.rs","oid","1ef61628107f826721e8f7f58ad116d1233fd107","size",6876,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! A [`ReplicaClient`] overlay that makes bundle-resident L0 segments\n//! visible beside a real per-cell client.\n//!\n//! celld-original, not a Litestream port, and self-contained like\n//! `crate::bundle`: nothing else in this crate depends on it. The overlay\n//! is how the additive compactor drains bundles — it lists and opens\n//! bundle rows as if they were ordinary per-cell L0 files (the bytes ARE\n//! ordinary L0 bytes, verbatim), and its writes pass through to the inner\n//! client, so compaction output is pure per-cell layout. The host supplies\n//! a [`BundleFetcher`] that knows where this cell-epoch's un-drained rows\n//! live and how to read them; with no fetcher the overlay is the inner\n//! client, exactly.\n\nuse std::future::Future;\nuse std::pin::Pin;\nuse std::sync::Arc;\n\nuse async_trait::async_trait;\n\nuse crate::bundle::BundleRow;\nuse crate::client::ReplicaClient;\nuse crate::error::Result;\nuse crate::ltx::FileInfo;\nuse crate::TXID;\n\n/// A bundle row plus the opaque source (for celld: the bundle object key)\n/// it can be fetched from.\n#[derive(Clone, Debug)]\npub struct LocatedRow {\n pub source: String,\n pub row: BundleRow,\n}\n\n/// Where this cell-epoch's un-drained bundle rows live. Object-safe by\n/// boxed futures, because hosts hold it behind `Arc`.\npub trait BundleFetcher: Send + Sync {\n /// The rows for this client's cell-epoch, any order, duplicates\n /// allowed (the overlay dedupes by TXID).\n fn rows<'a>(&'a self) -> Pin>> + Send + 'a>>;\n /// One row's verbatim L0 bytes.\n fn fetch<'a>(\n &'a self,\n located: &'a LocatedRow,\n ) -> Pin>> + Send + 'a>>;\n\n /// Reads a range within one row. Hosts with remote bundles override this\n /// method so oversized compaction does not buffer the complete bundle.\n fn fetch_range<'a>(\n &'a self,\n located: &'a LocatedRow,\n offset: u64,\n len: u64,\n ) -> Pin>> + Send + 'a>> {\n Box::pin(async move {\n let bytes = self.fetch(located).await?;\n let start = offset.min(bytes.len() as u64) as usize;\n let end = offset.saturating_add(len).min(bytes.len() as u64) as usize;\n Ok(bytes[start..end].to_vec())\n })\n }\n}\n\npub struct BundleOverlayClient {\n inner: C,\n fetcher: Option>,\n}\n\nimpl BundleOverlayClient {\n pub fn new(inner: C, fetcher: Option>) -> Self {\n Self { inner, fetcher }\n }\n\n async fn overlay_rows(&self) -> Result> {\n match &self.fetcher {\n Some(fetcher) => fetcher.rows().await,\n None => Ok(Vec::new()),\n }\n }\n}\n\n#[async_trait]\nimpl ReplicaClient for BundleOverlayClient {\n async fn ltx_files(&self, level: i32, seek: TXID) -> Result> {\n let mut files = self.inner.ltx_files(level, seek).await?;\n if level == 0 {\n for located in self.overlay_rows().await? {\n let txid = located.row.txid();\n if txid < seek\n || files\n .iter()\n .any(|f| f.min_txid == txid && f.max_txid == txid)\n {\n continue;\n }\n files.push(FileInfo {\n level: 0,\n min_txid: txid,\n max_txid: txid,\n pre_apply_checksum: Default::default(),\n post_apply_checksum: Default::default(),\n size: located.row.len as i64,\n created_at: None,\n });\n }\n // A single row and a later direct drain can share a minimum txid.\n // Only identical ranges are duplicates; hiding the wider range\n // loses its tail when the individual bundle rows leave the index.\n files.sort_by_key(|f| (f.min_txid, f.max_txid));\n files.dedup_by_key(|f| (f.min_txid, f.max_txid));\n }\n Ok(files)\n }\n\n async fn open_ltx_file(&self, level: i32, min_txid: TXID, max_txid: TXID) -> Result> {\n // The drained per-cell copy is byte-identical when it exists, so\n // try the inner client first and fall back to the bundle.\n match self.inner.open_ltx_file(level, min_txid, max_txid).await {\n Ok(bytes) => Ok(bytes),\n Err(inner_error) => {\n if level == 0 && min_txid == max_txid {\n for located in self.overlay_rows().await? {\n if located.row.txid() == min_txid {\n if let Some(fetcher) = &self.fetcher {\n return fetcher.fetch(&located).await;\n }\n }\n }\n }\n Err(inner_error)\n }\n }\n }\n\n async fn write_ltx_file(\n &self,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n data: &[u8],\n ) -> Result {\n self.inner\n .write_ltx_file(level, min_txid, max_txid, data)\n .await\n }\n\n async fn read_range(\n &self,\n level: i32,\n min: TXID,\n max: TXID,\n offset: u64,\n len: u64,\n ) -> Result> {\n match self.inner.read_range(level, min, max, offset, len).await {\n Ok(bytes) => Ok(bytes),\n Err(inner_error) => {\n if level == 0 && min == max {\n for located in self.overlay_rows().await? {\n if located.row.txid() == min {\n if let Some(fetcher) = &self.fetcher {\n return fetcher.fetch_range(&located, offset, len).await;\n }\n }\n }\n }\n Err(inner_error)\n }\n }\n }\n\n async fn write_ltx_file_from_file(\n &self,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n file: crate::host::HostFile,\n host: crate::LtxHost,\n ) -> Result {\n self.inner\n .write_ltx_file_from_file(level, min_txid, max_txid, file, host)\n .await\n }\n\n async fn delete_ltx_files(&self, files: &[FileInfo]) -> Result<()> {\n // Bundle rows cannot be carved out of their object; deleting the\n // per-cell twins (when present) is the inner client's business, and\n // a row-only \"file\" simply has nothing to delete.\n self.inner.delete_ltx_files(files).await\n }\n\n async fn delete_all(&self) -> Result<()> {\n self.inner.delete_all().await\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",1794436913952,"sidebar","expanded"]