[{"_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/file.rs","oid","fb040a5c73db5e8a7e09178daee565c938f0ff4c","size",7226,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","// The stand-alone file replica is an external ReplicaClient backend. Celld\n// production uses the injected local node filesystem and an object client.\n#![allow(clippy::disallowed_methods)]\n\n//! client::file — file-backed `ReplicaClient`.\n//!\n//! Ported from litestream@v0.5.11 `file/replica_client.go` (the v0.5 methods\n//! only; the `*V3` legacy generation shim is not included). On-disk\n//! layout matches upstream: `/ltx//-.ltx`, the\n//! same tree the golden fixtures were captured from.\n\nuse crate::error::{Error, Result};\nuse crate::ltx::{self, FileInfo};\nuse crate::{ltx_file_path, ltx_level_dir, TXID};\nuse async_trait::async_trait;\nuse std::time::{Duration, UNIX_EPOCH};\n\nuse super::ReplicaClient;\n\n/// A `ReplicaClient` that stores LTX files on the local filesystem.\n#[derive(Debug, Clone)]\npub struct FileReplicaClient {\n path: String,\n}\n\nimpl FileReplicaClient {\n /// Creates a client rooted at `path` (the replica destination directory).\n pub fn new(path: impl Into) -> Self {\n FileReplicaClient { path: path.into() }\n }\n}\n\n#[async_trait]\nimpl ReplicaClient for FileReplicaClient {\n async fn ltx_files(&self, level: i32, seek: TXID) -> Result> {\n let dir = ltx_level_dir(&self.path, level as u32);\n let mut rd = match tokio::fs::read_dir(&dir).await {\n Ok(rd) => rd,\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),\n Err(e) => return Err(e.into()),\n };\n\n let mut infos = Vec::new();\n while let Some(entry) = rd.next_entry().await? {\n let name = entry.file_name().to_string_lossy().into_owned();\n // ModTime is the timestamp set at write time; skip non-LTX names.\n let (min_txid, max_txid) = match ltx::parse_filename(&name) {\n Ok(t) => t,\n Err(_) => continue,\n };\n if min_txid < seek {\n continue;\n }\n let meta = entry.metadata().await?;\n infos.push(FileInfo {\n level,\n min_txid,\n max_txid,\n size: meta.len() as i64,\n created_at: meta.modified().ok(),\n ..Default::default()\n });\n }\n\n // Iterator contract: ascending by (level, min_txid, max_txid)\n // (ltx.NewFileInfoSliceIterator sorts the slice).\n infos.sort_by(|a, b| {\n (a.level, a.min_txid.0, a.max_txid.0).cmp(&(b.level, b.min_txid.0, b.max_txid.0))\n });\n Ok(infos)\n }\n\n async fn open_ltx_file(&self, level: i32, min_txid: TXID, max_txid: TXID) -> Result> {\n let path = ltx_file_path(&self.path, level as u32, min_txid, max_txid);\n // NotFound is preserved so callers can classify auto-recoverable errors.\n tokio::fs::read(&path).await.map_err(Error::Io)\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 // Peek the LTX header timestamp (preserved as the file's creation time).\n let header = ltx::Header::parse(data)?;\n let created_at = Some(UNIX_EPOCH + Duration::from_millis(header.timestamp.max(0) as u64));\n\n let filename = ltx_file_path(&self.path, level as u32, min_txid, max_txid);\n let dir = ltx_level_dir(&self.path, level as u32);\n tokio::fs::create_dir_all(&dir).await?;\n\n // Write to a temp file then atomically rename; clean the temp on error.\n let tmp = format!(\"{filename}.tmp\");\n if let Err(e) = write_then_rename(&tmp, &filename, data).await {\n let _ = tokio::fs::remove_file(&tmp).await;\n return Err(e);\n }\n\n Ok(FileInfo {\n level,\n min_txid,\n max_txid,\n size: data.len() as i64,\n created_at,\n ..Default::default()\n })\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 use tokio::io::{AsyncReadExt, AsyncSeekExt};\n let mut file =\n tokio::fs::File::open(ltx_file_path(&self.path, level as u32, min, max)).await?;\n file.seek(std::io::SeekFrom::Start(offset)).await?;\n let mut bytes = Vec::new();\n file.take(len).read_to_end(&mut bytes).await?;\n Ok(bytes)\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 use tokio::io::AsyncWriteExt;\n let (mut file, first, size) = super::read_upload_chunk(file, &host, 0, 1 << 20).await?;\n let header = ltx::Header::parse(&first)?;\n let filename = ltx_file_path(&self.path, level as u32, min_txid, max_txid);\n tokio::fs::create_dir_all(ltx_level_dir(&self.path, level as u32)).await?;\n // The named temporary file owns cleanup on errors and cancellation.\n let temporary = tempfile::NamedTempFile::new_in(ltx_level_dir(&self.path, level as u32))?;\n let mut output = tokio::fs::File::from_std(temporary.reopen()?);\n output.write_all(&first).await?;\n let mut offset = first.len() as u64;\n drop(first);\n while offset < size {\n let (next_file, bytes, _) =\n super::read_upload_chunk(file, &host, offset, 1 << 20).await?;\n file = next_file;\n output.write_all(&bytes).await?;\n offset += bytes.len() as u64;\n }\n output.sync_all().await?;\n drop(output);\n temporary\n .persist(&filename)\n .map_err(|error| Error::Io(error.error))?;\n Ok(FileInfo {\n level,\n min_txid,\n max_txid,\n size: size as i64,\n created_at: Some(UNIX_EPOCH + Duration::from_millis(header.timestamp.max(0) as u64)),\n ..Default::default()\n })\n }\n\n async fn delete_ltx_files(&self, files: &[FileInfo]) -> Result<()> {\n for info in files {\n let path = ltx_file_path(&self.path, info.level as u32, info.min_txid, info.max_txid);\n match tokio::fs::remove_file(&path).await {\n Ok(()) => {}\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}\n Err(e) => return Err(e.into()),\n }\n }\n Ok(())\n }\n\n async fn delete_all(&self) -> Result<()> {\n match tokio::fs::remove_dir_all(&self.path).await {\n Ok(()) => Ok(()),\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),\n Err(e) => Err(e.into()),\n }\n }\n}\n\n/// Writes `data` to `tmp`, fsyncs it, and renames it onto `final_path`.\nasync fn write_then_rename(tmp: &str, final_path: &str, data: &[u8]) -> Result<()> {\n use tokio::io::AsyncWriteExt;\n let mut f = tokio::fs::File::create(tmp).await?;\n f.write_all(data).await?;\n f.sync_all().await?;\n drop(f);\n tokio::fs::rename(tmp, final_path).await?;\n Ok(())\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",1794467212279,"sidebar","expanded"]