[{"_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/paged.rs","oid","1a97460f0744cd7f02a5fb60c903545f9a732cc7","size",21760,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! paged.rs — read-on-demand paging from replica objects.\n//!\n//! The read path of paged restore, ported from Litestream's\n//! VFS. A cold cell serves a page by fetching only that page's frame from the\n//! bucket instead of downloading its whole restore chain: build a page map from\n//! each planned object's trailing page index, then range-read and decode one\n//! frame per demanded page.\n//!\n//! An LTX file ends with `[page index][8-byte index length][16-byte trailer]`,\n//! and every object celld writes carries the index (`codec::Encoder::close`),\n//! so the map reads the index tail-first and never downloads the frames it does\n//! not need.\n\nuse crate::client::ReplicaClient;\nuse crate::codec::{decode_page_frame_of, decode_page_index};\nuse crate::error::{Error, Result};\nuse crate::ltx::{FileInfo, Header, HEADER_SIZE, TRAILER_SIZE};\nuse crate::ltx_file_path;\nuse crate::TXID;\nuse futures_util::StreamExt;\nuse std::collections::BTreeSet;\nuse std::collections::HashMap;\nuse std::io::Read;\nuse std::io::Seek;\nuse std::sync::Mutex;\n\n/// Bytes fetched from the object tail first, sized to hold the whole page index\n/// of an ordinary object in one read; a larger index needs a second read. A\n/// fold of thirty 1 MB rows indexes ~8,000 pages in ~70 KiB, so 32 KiB cost\n/// every object of the 5 GB whale a second round trip: its map took 12 s on\n/// the node. Bytes are cheap on a ranged read; rounds are the cost.\nconst TAIL_FETCH_BYTES: u64 = 256 * 1024;\n\n/// Width of the big-endian page-index-length field, between the index and the\n/// trailer.\nconst INDEX_LEN_FIELD: u64 = 8;\n\n/// The most frame bytes one fault fetches. A fault brings its page's whole\n/// run — the consecutive pages stored as adjacent frames of the same object,\n/// which is how a snapshot or compaction lays a b-tree's leaves and a blob's\n/// overflow chain out — so a sequential read costs one round-trip per run\n/// instead of one per 4KiB page. 64 pages caps the waste on a scattered\n/// point lookup, whose interior pages sit in runs it will not read.\npub const READAHEAD_BYTES: u64 = 1024 * 1024;\n\n/// Object page indexes fetched at once while building a map: a whale's plan\n/// of ~200 objects in a few rounds.\nconst INDEX_FETCH_CONCURRENCY: usize = 32;\n\n/// Fetches and decodes one object's page index with ranged reads, without\n/// downloading the object. It reads the tail, takes the index length from the\n/// 8-byte field before the trailer, and decodes the index — a second exact\n/// read only when the index is larger than the tail fetch.\n///\n/// Returns the `(pgno, frame_offset, frame_size)` entries: each page's whole\n/// frame lives at `frame_offset` in the file, `frame_size` bytes long.\npub async fn fetch_page_index(\n client: &C,\n info: &FileInfo,\n) -> Result> {\n let size = u64::try_from(info.size).map_err(|_| Error::LTXCorrupted)?;\n let footer = INDEX_LEN_FIELD + TRAILER_SIZE as u64;\n if size < footer {\n return Err(Error::LTXCorrupted);\n }\n\n let tail_len = TAIL_FETCH_BYTES.min(size);\n let tail_start = size - tail_len;\n let tail = client\n .read_range(\n info.level,\n info.min_txid,\n info.max_txid,\n tail_start,\n tail_len,\n )\n .await?;\n if (tail.len() as u64) < footer {\n return Err(Error::LTXCorrupted);\n }\n\n // The index-length field sits just before the trailer; the index precedes\n // it. Compute the index's file range from the object size, so a partial\n // tail read still locates it.\n let field_at = tail.len() - TRAILER_SIZE - INDEX_LEN_FIELD as usize;\n let index_len = u64::from_be_bytes(\n tail[field_at..field_at + INDEX_LEN_FIELD as usize]\n .try_into()\n .map_err(|_| Error::LTXCorrupted)?,\n );\n let index_end = size - footer;\n let index_start = index_end\n .checked_sub(index_len)\n .ok_or(Error::LTXCorrupted)?;\n\n if index_start >= tail_start {\n let lo = (index_start - tail_start) as usize;\n let hi = (index_end - tail_start) as usize;\n decode_page_index(&tail[lo..hi])\n } else {\n let index_bytes = client\n .read_range(\n info.level,\n info.min_txid,\n info.max_txid,\n index_start,\n index_len,\n )\n .await?;\n decode_page_index(&index_bytes)\n }\n}\n\n/// Where one page's frame lives: the object that holds it and the byte range of\n/// its frame within that object.\n#[derive(Debug, Clone)]\npub struct PageLocator {\n pub info: FileInfo,\n pub offset: u64,\n pub size: u64,\n}\n\n/// An immutable per-activation view of a restore cut: which object serves each\n/// page, plus the page size and the database's page count at the cut. Built\n/// once from a plan and never updated in place, so a faulted page always comes\n/// from the pinned cut even as newer objects land.\n#[derive(Debug)]\npub struct PageMap {\n pub page_size: u32,\n pub commit: u32,\n pub pages: HashMap,\n}\n\n/// Builds a page map over a restore `plan` (as [`crate::replica::calc_restore_plan`]\n/// returns it, oldest object first) without downloading the objects. It fetches\n/// each object's page index and inserts its pages, so a later object in the\n/// chain overrides an earlier one — last-writer-wins, the newest version of\n/// each page. The page size and commit come from the newest object's header.\npub async fn build_page_map(client: &C, plan: &[FileInfo]) -> Result {\n let newest = plan.last().ok_or(Error::TxNotAvailable)?;\n let mut pages: HashMap = HashMap::new();\n // The indexes are independent reads, so fetch them concurrently; a whale's\n // plan of two dozen objects otherwise costs two round-trips each, in\n // series, on the activation path. `buffered` keeps plan order for the\n // last-writer-wins insertion below.\n // Built as a Vec first: a closure held inside this future's state would\n // make it higher-ranked over the plan's borrow and unspawnable.\n let header_bytes = client\n .read_range(\n newest.level,\n newest.min_txid,\n newest.max_txid,\n 0,\n HEADER_SIZE as u64,\n )\n .await?;\n let header = Header::parse(&header_bytes)?;\n let fetches: Vec<_> = plan\n .iter()\n .map(|info| fetch_page_index(client, info))\n .collect();\n let mut indexes = futures_util::stream::iter(fetches).buffered(INDEX_FETCH_CONCURRENCY);\n let mut planned = plan.iter();\n while let Some(index) = indexes.next().await {\n let info = planned.next().expect(\"one index per planned object\");\n let index = index?;\n // An older object can hold pages past the newest commit (the\n // database shrank since); they are not part of the cut.\n for (pgno, offset, size) in index.into_iter().filter(|(p, _, _)| *p <= header.commit) {\n pages.insert(\n pgno,\n PageLocator {\n info: info.clone(),\n offset,\n size,\n },\n );\n }\n }\n Ok(PageMap {\n page_size: header.page_size,\n commit: header.commit,\n pages,\n })\n}\n\nimpl PageMap {\n /// Serves one page: range-reads its frame from the object the map points to\n /// and decodes it. `None` when the cut has no such page (a page above the\n /// commit, or one never written).\n pub async fn read_page(\n &self,\n client: &C,\n pgno: u32,\n ) -> Result>> {\n let Some(loc) = self.pages.get(&pgno) else {\n return Ok(None);\n };\n let frame = client\n .read_range(\n loc.info.level,\n loc.info.min_txid,\n loc.info.max_txid,\n loc.offset,\n loc.size,\n )\n .await?;\n Ok(Some(decode_page_frame_of(\n &frame,\n self.page_size as usize,\n pgno,\n )?))\n }\n\n /// The VFS fault path, on whatever thread SQLite called `xRead` from:\n /// `pgno`'s page plus its run (see [`READAHEAD_BYTES`]), in one ranged\n /// read of the object that holds them. Empty when the cut has no such\n /// page; otherwise the first entry is `pgno`.\n pub fn read_run_sync(\n &self,\n reader: &dyn RangeReader,\n pgno: u32,\n max_bytes: u64,\n ) -> Result)>> {\n let Some(first) = self.pages.get(&pgno) else {\n return Ok(Vec::new());\n };\n // The cap is in decoded page bytes: frames are LZ4-compressed, and a\n // cap on frame bytes let one fault of a compressible file bring the\n // whole database (a 13MB fixture in a single 1MiB \"run\").\n let max_pages = (max_bytes / u64::from(self.page_size)).max(1) as u32;\n let mut end = first.offset + first.size;\n let mut last = pgno;\n while let Some(next) = self.pages.get(&(last + 1)) {\n let adjacent = next.info == first.info && next.offset == end;\n if !adjacent || last + 1 - pgno >= max_pages {\n break;\n }\n end += next.size;\n last += 1;\n }\n let bytes = reader.read_range(&first.info, first.offset, end - first.offset)?;\n let mut run = Vec::with_capacity((last - pgno + 1) as usize);\n let mut at = 0;\n if (bytes.len() as u64) < end - first.offset {\n return Err(Error::LTXCorrupted);\n }\n for pgno_ in pgno..=last {\n let size = self.pages[&pgno_].size as usize;\n let page = decode_page_frame_of(&bytes[at..at + size], self.page_size as usize, pgno_)?;\n run.push((pgno_, page));\n at += size;\n }\n Ok(run)\n }\n}\n\n/// A blocking ranged read of one replica object: the fault path's I/O.\n///\n/// SQLite's `xRead` is synchronous and runs on whichever thread entered the\n/// connection — a blocking thread for the managed Db, but a tokio task's\n/// thread for the actor's own queries, since isolates are entered from tasks.\n/// Every bridge from that thread into the async client failed in one context\n/// (`block_on` on a worker, `block_in_place` on a current-thread runtime, a\n/// same-runtime spawn stuck in the worker's LIFO slot). So the fault path does\n/// its own I/O on the calling thread with a plain socket and never touches a\n/// runtime, which is correct from any thread by construction.\npub trait RangeReader: Send + Sync {\n /// `len` bytes of `info`'s object starting at `offset`.\n fn read_range(&self, info: &FileInfo, offset: u64, len: u64) -> Result>;\n}\n\n/// A [`RangeReader`] over the file replica layout (`ltx_file_path`).\npub struct FileRangeReader {\n root: String,\n}\n\nimpl FileRangeReader {\n pub fn new(root: impl Into) -> Self {\n Self { root: root.into() }\n }\n}\n\n// The file replica is a development and test backend, not node storage:\n// it reads the same directory the file client writes with tokio::fs, outside\n// the injected node filesystem the lint guards.\n#[allow(clippy::disallowed_methods)]\nimpl RangeReader for FileRangeReader {\n fn read_range(&self, info: &FileInfo, offset: u64, len: u64) -> Result> {\n let path = ltx_file_path(&self.root, info.level as u32, info.min_txid, info.max_txid);\n let mut file = std::fs::File::open(path)?;\n file.seek(std::io::SeekFrom::Start(offset))?;\n let mut out = vec![0; len as usize];\n file.read_exact(&mut out)?;\n Ok(out)\n }\n}\n\n/// Children a fault should bring along, read off the faulted page's b-tree\n/// structure. Round-trips are the whole cost of a fault (a 4KiB and a 1MiB\n/// range cost about the same), and SQLite asks for pages one at a time, so\n/// the only way below one round-trip per page is to read the structure: an\n/// interior page names every child. Fetching the children concurrently makes\n/// a scan cost one round per tree level instead of one per leaf. A misparse\n/// can only waste bytes: every served page still comes from the map.\n///\n/// A leaf's overflow chains are deliberately not here: a chain is stored as\n/// a run, so the sequential walk SQLite does gets it in one read after the\n/// first overflow fault, while fetching every cell's chain at the leaf fault\n/// pulls the neighbours' megabytes for a single-row read.\n#[derive(Debug, Default, PartialEq, Eq)]\npub struct Prefetch {\n /// Child pages of an interior page, in cell order.\n pub children: Vec,\n}\n\n/// Children fetched ahead of a fault on a child page, along the tree in\n/// page order: a scan then costs one round per window of leaves, and a\n/// point read pays for one window at most.\nconst SCAN_AHEAD: usize = 64;\n\n/// Concurrent range reads while fetching ahead.\nconst PREFETCH_WORKERS: usize = 64;\n\n/// Parses a decoded b-tree page for what to prefetch; page 1 carries the\n/// 100-byte file header. Anything malformed yields an empty plan.\npub fn prefetch_plan(pgno: u32, page: &[u8]) -> Prefetch {\n let mut plan = Prefetch::default();\n let base = if pgno == 1 { 100 } else { 0 };\n let Some(head) = page.get(base..base + 12) else {\n return plan;\n };\n // 0x05 interior table, 0x02 interior index; leaves have no children.\n if head[0] != 0x05 && head[0] != 0x02 {\n return plan;\n }\n let cells = u16::from_be_bytes([head[3], head[4]]) as usize;\n let ptrs = base + 12;\n for i in 0..cells {\n let Some(at) = page.get(ptrs + 2 * i..ptrs + 2 * i + 2) else {\n break;\n };\n let at = u16::from_be_bytes([at[0], at[1]]) as usize;\n let Some(child) = page.get(at..at + 4) else {\n break;\n };\n plan.children\n .push(u32::from_be_bytes([child[0], child[1], child[2], child[3]]));\n }\n plan.children\n .push(u32::from_be_bytes([head[8], head[9], head[10], head[11]]));\n plan\n}\n\n/// A synchronous page reader over a [`PageMap`] for the paged SQLite VFS:\n/// the pinned cut's map plus a blocking [`RangeReader`] for its objects.\n///\n/// celld's VFS faults a missing page into a local sparse main-db file, so\n/// SQLite then runs entirely normally over that file (WAL, page-1 writes,\n/// checkpoint) and the local file is the cache. This source therefore holds no\n/// cache and does not rewrite page 1: it just serves the pinned cut's bytes for\n/// a page the local file does not have yet.\npub struct PageSource {\n map: PageMap,\n reader: Box,\n /// B-tree child pages named by an interior page this source has decoded\n /// and not yet delivered, and the last one that faulted: a fault on one\n /// fetches it alone, and a fault that continues along the tree looks\n /// ahead (see [`PageSource::read_run_blocking`]).\n scan: Mutex,\n /// Faults served: every `read_run_blocking`, foreground or hydration.\n faults: std::sync::atomic::AtomicU64,\n}\n\n#[derive(Default)]\nstruct Scan {\n children: BTreeSet,\n last: Option,\n}\n\nimpl PageSource {\n pub fn new(map: PageMap, reader: Box) -> Self {\n Self {\n map,\n reader,\n scan: Mutex::new(Scan::default()),\n faults: std::sync::atomic::AtomicU64::new(0),\n }\n }\n\n /// Faults served so far. A hydrated cell's reads leave it unchanged.\n pub fn faults(&self) -> u64 {\n self.faults.load(std::sync::atomic::Ordering::Relaxed)\n }\n\n /// The database page size at the pinned cut.\n pub fn page_size(&self) -> u32 {\n self.map.page_size\n }\n\n /// The database's page count at the pinned cut. The VFS sizes the local file\n /// to `page_size * commit`.\n pub fn commit(&self) -> u32 {\n self.map.commit\n }\n\n /// Reads `pgno`'s run on the calling thread ([`PageMap::read_run_sync`]),\n /// then what the faulted page's structure says comes next\n /// ([`prefetch_plan`]), concurrently: every page returned is hydrated by\n /// the VFS, so a scan or a big-row read pays one round-trip per level.\n pub fn read_run_blocking(&self, pgno: u32) -> Result)>> {\n self.faults\n .fetch_add(1, std::sync::atomic::Ordering::Relaxed);\n // A page an interior page named is a b-tree child. Its run is its\n // neighbours in page order — for a leaf of a big-row table, the rows'\n // overflow chains, megabytes a scan never reads (the whale's\n // count(*) faulted 700 MB). Fetch it alone. When the fault continues\n // along the tree (the last child fault was the nearest one below),\n // it is a scan: bring the next children with it, a window at a time.\n // A point read walks one path, so it never looks ahead: its cost\n // stays one request per level.\n let ahead = {\n let mut scan = self.scan.lock().unwrap();\n scan.children.remove(&pgno).then(|| {\n let sequential = scan.last.is_some_and(|last| {\n last < pgno && scan.children.range(last + 1..pgno).next().is_none()\n });\n scan.last = Some(pgno);\n let next: Vec = if sequential {\n scan.children\n .range(pgno + 1..)\n .take(SCAN_AHEAD)\n .copied()\n .collect()\n } else {\n Vec::new()\n };\n for p in &next {\n scan.children.remove(p);\n }\n next\n })\n };\n let page = u64::from(self.map.page_size);\n let mut run = self.map.read_run_sync(\n &*self.reader,\n pgno,\n if ahead.is_some() {\n page\n } else {\n READAHEAD_BYTES\n },\n )?;\n if run.is_empty() {\n return Ok(run);\n }\n if let Some(wanted) = ahead.filter(|w| !w.is_empty()) {\n let fetched = std::thread::scope(|scope| {\n let workers: Vec<_> = wanted\n .chunks(wanted.len().div_ceil(PREFETCH_WORKERS).max(1))\n .map(|chunk| {\n scope.spawn(move || {\n // Consecutive children (a small-row table's\n // leaves) come as one run; the rest one by one.\n let mut pages = Vec::new();\n let mut i = 0;\n while i < chunk.len() {\n let first = chunk[i];\n let mut len = 1;\n while i + len < chunk.len() && chunk[i + len] == first + len as u32\n {\n len += 1;\n }\n let got = self\n .map\n .read_run_sync(&*self.reader, first, len as u64 * page)\n .unwrap_or_default();\n i += got.len().max(1);\n pages.extend(got);\n }\n pages\n })\n })\n .collect();\n workers\n .into_iter()\n .filter_map(|w| w.join().ok())\n .flatten()\n .collect::>()\n });\n // A child a failed fetch did not deliver goes back on the list;\n // dropped, its own fault would come as a run of its rows.\n let delivered: std::collections::HashSet =\n fetched.iter().map(|(p, _)| *p).collect();\n let mut scan = self.scan.lock().unwrap();\n scan.children\n .extend(wanted.iter().copied().filter(|p| !delivered.contains(p)));\n drop(scan);\n run.extend(fetched);\n }\n // Every page delivered is off the list; every interior page among\n // them names the children the next faults fetch alone and ahead.\n let mut scan = self.scan.lock().unwrap();\n for (p, page) in &run {\n scan.children.remove(p);\n scan.children.extend(prefetch_plan(*p, page).children);\n }\n Ok(run)\n }\n\n /// Reads one page on the calling thread. `None` when the cut has no such\n /// page.\n pub fn read_page_blocking(&self, pgno: u32) -> Result>> {\n Ok(self\n .read_run_blocking(pgno)?\n .into_iter()\n .next()\n .map(|(_, page)| page))\n }\n}\n\n/// Routes a fault to the epoch that holds the object, by the object's first\n/// txid, as [`crate::client::epochs::EpochChain`] routes the async reads.\npub struct EpochChainReader {\n spans: Vec<(TXID, Box)>,\n}\n\nimpl EpochChainReader {\n /// `spans` ascending by the first txid each reader serves.\n pub fn new(spans: Vec<(TXID, Box)>) -> Self {\n assert!(!spans.is_empty(), \"an epoch chain has a span\");\n Self { spans }\n }\n}\n\nimpl RangeReader for EpochChainReader {\n fn read_range(&self, info: &FileInfo, offset: u64, len: u64) -> Result> {\n let (_, reader) = self\n .spans\n .iter()\n .rev()\n .find(|(lo, _)| *lo <= info.min_txid)\n .unwrap_or(&self.spans[0]);\n reader.read_range(info, offset, len)\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",1794475127271,"sidebar","expanded"]