[{"_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_vfs.rs","oid","3f9b7751911aee988c3f99763af2e1de5b8335d2","size",28013,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! paged_vfs.rs — a SQLite VFS that faults main-db pages in from the bucket.\n//!\n//! The fault-in model of paged restore. The main db is a real local file\n//! that starts sparse, sized to the pinned cut. On a read of a page the\n//! local file does not hold yet, the VFS fetches that page's run from the\n//! bucket ([`PageSource`]), writes it into the local file, then lets SQLite\n//! read it normally. SQLite otherwise runs as usual over the local file —\n//! WAL, page-1 writes, checkpoint — and faulting is hydration.\n//!\n//! It is a thin wrapper over a base VFS (the host's named VFS, or SQLite's\n//! default): every method forwards, except the main db's `xRead`, which\n//! faults first, `xWrite` and `xTruncate`, which keep the hydration set\n//! honest, and `xOpen`, which sizes the sparse file. The invariants:\n//!\n//! - One registration serves one database. The hydration set is per\n//! registration, shared by every connection to that file, so a checkpoint\n//! or an earlier fault by one connection is never overwritten by another\n//! connection re-faulting the older cut page.\n//! - A page is faulted at most once, and only when the local file does not\n//! hold newer bytes: a page a connection wrote (a checkpoint) is marked\n//! hydrated, and a run that arrives with a fault only fills pages not yet\n//! hydrated.\n//! - A page absent from the cut past the commit reads as\n//! the sparse file's zeros, and is marked resolved so it never faults again.\n//! - Pages truncated away are forgotten, so a later regrowth faults or\n//! writes them afresh rather than trusting stale marks.\n//! - A registration serves one file, the cell's database; any other main\n//! database opened by its name is a plain file, so no second sparse view\n//! can mark the cell's hydration set for bytes the cell never received.\n//! - The fault runs on the calling thread and touches no runtime; a failure\n//! surfaces to SQLite as `SQLITE_IOERR_READ` on that read.\n//! - Unregistering frees the registration; open files keep their own\n//! references to the reader and the hydration set, and SQLite opens\n//! nothing new through an unregistered VFS.\n\nuse crate::ltx::lock_pgno;\nuse crate::paged::PageSource;\nuse rusqlite::ffi;\nuse std::collections::HashSet;\nuse std::ffi::{c_void, CStr, CString};\nuse std::os::raw::{c_char, c_int};\nuse std::path::Path;\nuse std::sync::atomic::Ordering;\nuse std::sync::{Arc, Mutex};\n\n/// Type-erased page reader so one VFS registration can serve any client type.\npub trait PageReader: Send + Sync {\n fn page_size(&self) -> u32;\n fn commit(&self) -> u32;\n /// The cut's bytes for `pgno` and the pages stored right after it that\n /// came along in the same read (first entry is `pgno`); empty if the cut\n /// has no such page.\n fn read_run(&self, pgno: u32) -> crate::error::Result)>>;\n /// Faults served so far, for the hydration count.\n fn faults(&self) -> u64 {\n 0\n }\n}\n\nimpl PageReader for PageSource {\n fn page_size(&self) -> u32 {\n PageSource::page_size(self)\n }\n fn commit(&self) -> u32 {\n PageSource::commit(self)\n }\n fn read_run(&self, pgno: u32) -> crate::error::Result)>> {\n self.read_run_blocking(pgno)\n }\n fn faults(&self) -> u64 {\n PageSource::faults(self)\n }\n}\n\n/// Per-registration state, hung off the VFS `pAppData`.\nstruct VfsAppData {\n base: *mut ffi::sqlite3_vfs,\n /// The cell's database, as the base VFS spells it: the one file the\n /// registration serves.\n path: CString,\n reader: Arc,\n /// Files open through this registration. Unregistering with files still\n /// open used to free the VFS under them: the pager's close deletes the\n /// WAL through `xDelete`, which reads `pAppData` of the freed struct.\n /// The registration is freed by whichever comes last, the unregister or\n /// the final close.\n open_files: std::sync::atomic::AtomicUsize,\n retired: std::sync::atomic::AtomicBool,\n /// Pages whose newest bytes the local main-db file already holds, shared\n /// by every connection opened through this registration. Per-connection\n /// sets would let a later connection re-fault the older cut page over a\n /// page another connection already checkpointed — silent data rollback.\n /// One registration serves one database, so registration scope is\n /// database scope.\n hydrated: Arc>>,\n}\n\n/// Our `sqlite3_file` subclass: SQLite reads `methods` (first field); the rest is\n/// ours. `base` is the underlying VFS's file; `state` its fault-in bookkeeping.\n#[repr(C)]\nstruct PagedFile {\n methods: *const ffi::sqlite3_io_methods,\n base: *mut ffi::sqlite3_file,\n state: *mut FileState,\n}\n\nstruct FileState {\n /// The registration, for the open-file count (see [`VfsAppData::open_files`]).\n vfs: *mut ffi::sqlite3_vfs,\n /// `Some` only for the main db; other files (WAL, journal) forward untouched.\n reader: Option>,\n /// The registration's shared hydration set (see [`VfsAppData::hydrated`]).\n hydrated: Arc>>,\n page_size: u32,\n}\n\nunsafe fn base_vfs(vfs: *mut ffi::sqlite3_vfs) -> *mut ffi::sqlite3_vfs {\n (*((*vfs).pAppData as *mut VfsAppData)).base\n}\n\nunsafe fn base_methods(file: *mut ffi::sqlite3_file) -> *const ffi::sqlite3_io_methods {\n let paged = file as *mut PagedFile;\n (*(*paged).base).pMethods\n}\n\n// ── file methods ────────────────────────────────────────────────────────────\n\n/// A panic here would unwind across `extern \"C\"` and abort the process; the\n/// fault path spawns threads and locks a mutex, both of which can panic.\n/// It is an I/O error to the query instead.\nunsafe extern \"C\" fn x_read(\n file: *mut ffi::sqlite3_file,\n buf: *mut c_void,\n amt: c_int,\n ofst: ffi::sqlite3_int64,\n) -> c_int {\n std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {\n x_read_inner(file, buf, amt, ofst)\n }))\n .unwrap_or(ffi::SQLITE_IOERR_READ)\n}\n\nunsafe fn x_read_inner(\n file: *mut ffi::sqlite3_file,\n buf: *mut c_void,\n amt: c_int,\n ofst: ffi::sqlite3_int64,\n) -> c_int {\n let paged = file as *mut PagedFile;\n let state = &*(*paged).state;\n if let Some(reader) = &state.reader {\n let ps = state.page_size.max(1) as i64;\n let first = (ofst / ps) as u32 + 1;\n let last = ((ofst + amt as i64 - 1) / ps) as u32 + 1;\n for pgno in first..=last {\n if state.hydrated.lock().unwrap().contains(&pgno) {\n continue;\n }\n // The fetch runs without the lock, so a fault by one connection\n // never stalls another's read of a hydrated page, and a\n // background hydration step never stalls a foreground fault.\n // Two faults of one run both fetch; the loser writes nothing.\n let run = match reader.read_run(pgno) {\n Ok(run) => run,\n // The fill can complete while this fetch is in flight, and\n // then the owner's epoch GC may delete the objects the fetch\n // reads. The page is in the local file by then; read it there\n // instead of failing the query.\n Err(_) if state.hydrated.lock().unwrap().contains(&pgno) => continue,\n Err(_) => return ffi::SQLITE_IOERR_READ,\n };\n let mut hydrated = state.hydrated.lock().unwrap();\n if hydrated.contains(&pgno) {\n continue;\n }\n // Absent in the cut. Past the commit (or the lock page) the\n // sparse file's zeros stand; mark resolved so we do not re-fault.\n // Inside it the map is broken — a plan starts with a snapshot,\n // and a snapshot carries every page — and zeros would go out as\n // data (a 2 GB opener two thirds holes, fleet 2026-09-02): fail\n // the read instead.\n if run.is_empty() {\n if pgno <= reader.commit() && pgno != lock_pgno(ps as u32) {\n return ffi::SQLITE_IOERR_READ;\n }\n hydrated.insert(pgno);\n continue;\n }\n // Hydrate the whole run, except a page that already holds newer\n // bytes (a checkpoint, or an earlier run) — the cut is older.\n for (pgno_, page) in run {\n if pgno_ != pgno && hydrated.contains(&pgno_) {\n continue;\n }\n let base = (*paged).base;\n let write = (*base_methods(file)).xWrite.unwrap();\n let rc = write(\n base,\n page.as_ptr() as *const c_void,\n page.len() as c_int,\n (pgno_ as i64 - 1) * ps,\n );\n if rc != ffi::SQLITE_OK {\n return rc;\n }\n hydrated.insert(pgno_);\n }\n }\n }\n let read = (*base_methods(file)).xRead.unwrap();\n read((*paged).base, buf, amt, ofst)\n}\n\nunsafe extern \"C\" fn x_write(\n file: *mut ffi::sqlite3_file,\n buf: *const c_void,\n amt: c_int,\n ofst: ffi::sqlite3_int64,\n) -> c_int {\n // A write (e.g. a checkpoint) establishes the newest bytes for its pages, so\n // mark them hydrated to keep a later read from faulting an older cut page\n // over them.\n let paged = file as *mut PagedFile;\n let state = &*(*paged).state;\n if state.reader.is_some() {\n let ps = state.page_size.max(1) as i64;\n let first = (ofst / ps) as u32 + 1;\n let last = ((ofst + amt as i64 - 1) / ps) as u32 + 1;\n let mut hydrated = state.hydrated.lock().unwrap();\n let write = (*base_methods(file)).xWrite.unwrap();\n let rc = write((*paged).base, buf, amt, ofst);\n // Only bytes that landed count: a mark on a failed write would turn\n // the page into a hole the fault path never fills.\n if rc == ffi::SQLITE_OK {\n hydrated.extend(first..=last);\n }\n return rc;\n }\n let write = (*base_methods(file)).xWrite.unwrap();\n write((*paged).base, buf, amt, ofst)\n}\n\nunsafe extern \"C\" fn x_close(file: *mut ffi::sqlite3_file) -> c_int {\n let paged = file as *mut PagedFile;\n let rc = match (*base_methods(file)).xClose {\n Some(close) => close((*paged).base),\n None => ffi::SQLITE_OK,\n };\n ffi::sqlite3_free((*paged).base as *mut c_void);\n let state = Box::from_raw((*paged).state);\n let vfs = state.vfs;\n drop(state);\n let app = &*((*vfs).pAppData as *mut VfsAppData);\n let last = app.open_files.fetch_sub(1, Ordering::AcqRel) == 1;\n if last && app.retired.load(Ordering::Acquire) {\n free_vfs(vfs);\n }\n rc\n}\n\n/// Frees a registration's allocations: after its unregister, once no file\n/// is open through it.\nunsafe fn free_vfs(vfs: *mut ffi::sqlite3_vfs) {\n drop(Box::from_raw((*vfs).pAppData as *mut VfsAppData));\n drop(CString::from_raw((*vfs).zName as *mut c_char));\n drop(Box::from_raw(vfs));\n}\n\n/// Forward an io-method to the base file verbatim.\nmacro_rules! forward_io {\n ($name:ident ( $($arg:ident : $ty:ty),* ) $(-> $ret:ty)? => $field:ident) => {\n unsafe extern \"C\" fn $name(file: *mut ffi::sqlite3_file $(, $arg: $ty)*) $(-> $ret)? {\n let paged = file as *mut PagedFile;\n let m = (*base_methods(file)).$field.unwrap();\n m((*paged).base $(, $arg)*)\n }\n };\n}\n\nunsafe extern \"C\" fn x_truncate(file: *mut ffi::sqlite3_file, size: ffi::sqlite3_int64) -> c_int {\n // Pages past the new size are gone from the local file; drop their marks\n // so a later regrowth faults or writes them afresh instead of reading\n // the base file's zeros as hydrated data.\n let paged = file as *mut PagedFile;\n let state = &*(*paged).state;\n if state.reader.is_some() {\n let ps = state.page_size.max(1) as i64;\n let keep = ((size + ps - 1) / ps).max(0) as u32;\n state.hydrated.lock().unwrap().retain(|pgno| *pgno <= keep);\n }\n let truncate = (*base_methods(file)).xTruncate.unwrap();\n truncate((*paged).base, size)\n}\nforward_io!(x_sync(flags: c_int) -> c_int => xSync);\nforward_io!(x_file_size(size: *mut ffi::sqlite3_int64) -> c_int => xFileSize);\nforward_io!(x_lock(lock: c_int) -> c_int => xLock);\nforward_io!(x_unlock(lock: c_int) -> c_int => xUnlock);\nforward_io!(x_check_reserved_lock(out: *mut c_int) -> c_int => xCheckReservedLock);\nforward_io!(x_file_control(op: c_int, arg: *mut c_void) -> c_int => xFileControl);\nforward_io!(x_sector_size() -> c_int => xSectorSize);\nforward_io!(x_device_characteristics() -> c_int => xDeviceCharacteristics);\nforward_io!(x_shm_map(pg: c_int, sz: c_int, ext: c_int, out: *mut *mut c_void) -> c_int => xShmMap);\nforward_io!(x_shm_lock(ofst: c_int, n: c_int, flags: c_int) -> c_int => xShmLock);\nforward_io!(x_shm_unmap(delete: c_int) -> c_int => xShmUnmap);\n\nunsafe extern \"C\" fn x_shm_barrier(file: *mut ffi::sqlite3_file) {\n if let Some(barrier) = (*base_methods(file)).xShmBarrier {\n let paged = file as *mut PagedFile;\n barrier((*paged).base);\n }\n}\n\nstatic PAGED_IO_METHODS: ffi::sqlite3_io_methods = ffi::sqlite3_io_methods {\n iVersion: 2,\n xClose: Some(x_close),\n xRead: Some(x_read),\n xWrite: Some(x_write),\n xTruncate: Some(x_truncate),\n xSync: Some(x_sync),\n xFileSize: Some(x_file_size),\n xLock: Some(x_lock),\n xUnlock: Some(x_unlock),\n xCheckReservedLock: Some(x_check_reserved_lock),\n xFileControl: Some(x_file_control),\n xSectorSize: Some(x_sector_size),\n xDeviceCharacteristics: Some(x_device_characteristics),\n xShmMap: Some(x_shm_map),\n xShmLock: Some(x_shm_lock),\n xShmBarrier: Some(x_shm_barrier),\n xShmUnmap: Some(x_shm_unmap),\n xFetch: None,\n xUnfetch: None,\n};\n\n// ── vfs methods ───────────────────────────────────────────────────────────────\n\n/// The path SQLite hands `xOpen` is the base VFS's full pathname (absolute,\n/// symlinks resolved on unix), so the registered path is spelled the same\n/// way, by the same VFS, before the comparison.\nunsafe fn full_pathname(base: *mut ffi::sqlite3_vfs, path: &Path) -> Option {\n let raw = CString::new(path.as_os_str().as_encoded_bytes()).ok()?;\n let mut out = vec![0u8; (*base).mxPathname as usize + 1];\n let full = (*base).xFullPathname?;\n let rc = full(\n base,\n raw.as_ptr(),\n out.len() as c_int,\n out.as_mut_ptr().cast(),\n );\n // SQLITE_OK_SYMLINK (OK in the low byte) reports a resolved symlink.\n if rc & 0xff != ffi::SQLITE_OK {\n return None;\n }\n CStr::from_bytes_until_nul(&out).ok().map(CStr::to_owned)\n}\n\nunsafe extern \"C\" fn x_open(\n vfs: *mut ffi::sqlite3_vfs,\n name: *const c_char,\n file: *mut ffi::sqlite3_file,\n flags: c_int,\n out_flags: *mut c_int,\n) -> c_int {\n let app = &*((*vfs).pAppData as *mut VfsAppData);\n let base = app.base;\n // A separate base file, sized as the base VFS expects.\n let base_file = ffi::sqlite3_malloc((*base).szOsFile) as *mut ffi::sqlite3_file;\n if base_file.is_null() {\n return ffi::SQLITE_NOMEM;\n }\n std::ptr::write_bytes(base_file as *mut u8, 0, (*base).szOsFile as usize);\n\n let open = (*base).xOpen.unwrap();\n let rc = open(base, name, base_file, flags, out_flags);\n if rc != ffi::SQLITE_OK {\n ffi::sqlite3_free(base_file as *mut c_void);\n return rc;\n }\n\n // The registration serves one file: the cell's database. Any other\n // main database opened by this VFS name (a backup destination, an\n // inspection copy, a neighbour's file) is a plain file, because a second\n // sparse view of the cut would share the hydration set and mark pages\n // the cell's file never received, so the cell reads holes\n // (CelldPagedVfs.tla, `BreakSecondFileShares`).\n let is_cell = flags & ffi::SQLITE_OPEN_MAIN_DB != 0\n && !name.is_null()\n && CStr::from_ptr(name) == app.path.as_c_str();\n let (reader, page_size) = if is_cell {\n (Some(app.reader.clone()), app.reader.page_size())\n } else {\n (None, 0)\n };\n let state = Box::into_raw(Box::new(FileState {\n vfs,\n reader,\n hydrated: app.hydrated.clone(),\n page_size,\n }));\n app.open_files.fetch_add(1, Ordering::AcqRel);\n\n let paged = file as *mut PagedFile;\n (*paged).methods = &PAGED_IO_METHODS;\n (*paged).base = base_file;\n (*paged).state = state;\n\n // Size an EMPTY local main-db file to the cut so its logical size is\n // right and un-faulted pages read as holes we intercept. A later open\n // (the actor's connection, an alarm read) finds a file a checkpoint may\n // have grown past the cut; sizing it again would cut those pages off\n // while their hydration marks survive (CelldPagedVfs.tla,\n // `BreakOpenTruncates`).\n if is_cell {\n let methods = &*base_methods(file);\n let mut size: ffi::sqlite3_int64 = 0;\n let rc = (methods.xFileSize.unwrap())(base_file, &mut size);\n if rc != ffi::SQLITE_OK {\n return rc;\n }\n if size == 0 {\n let bytes = i64::from(app.reader.page_size()) * i64::from(app.reader.commit());\n let rc = (methods.xTruncate.unwrap())(base_file, bytes);\n if rc != ffi::SQLITE_OK {\n return rc;\n }\n }\n }\n ffi::SQLITE_OK\n}\n\nmacro_rules! forward_vfs {\n ($name:ident ( $($arg:ident : $ty:ty),* ) $(-> $ret:ty)? => $field:ident) => {\n unsafe extern \"C\" fn $name(vfs: *mut ffi::sqlite3_vfs $(, $arg: $ty)*) $(-> $ret)? {\n let base = base_vfs(vfs);\n let m = (*base).$field.unwrap();\n m(base $(, $arg)*)\n }\n };\n}\n\nforward_vfs!(x_delete(name: *const c_char, sync: c_int) -> c_int => xDelete);\nforward_vfs!(x_access(name: *const c_char, flags: c_int, out: *mut c_int) -> c_int => xAccess);\nforward_vfs!(x_full_pathname(name: *const c_char, n: c_int, out: *mut c_char) -> c_int => xFullPathname);\nforward_vfs!(x_randomness(n: c_int, out: *mut c_char) -> c_int => xRandomness);\nforward_vfs!(x_sleep(micros: c_int) -> c_int => xSleep);\nforward_vfs!(x_current_time(out: *mut f64) -> c_int => xCurrentTime);\nforward_vfs!(x_get_last_error(n: c_int, out: *mut c_char) -> c_int => xGetLastError);\n\n/// A process-unique suffix for each registration, so re-activations never\n/// collide on a name: SQLite finds a VFS by name, process-wide, and an\n/// unregistered name may still have files open through it.\nstatic REGISTRATION_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);\n\n/// The next registration's name, unique in this process.\npub fn next_registration_name() -> String {\n format!(\n \"celld-paged-{}\",\n REGISTRATION_SEQ.fetch_add(1, Ordering::SeqCst)\n )\n}\n\n/// Registers a paged VFS named `name` that faults the main db's pages in from\n/// `reader`. celld opens the cell's database through this VFS name. Call\n/// [`unregister_paged_vfs`] when the cell closes.\n///\n/// `base` names the VFS the paged one wraps; `None` is SQLite's default.\n/// celld passes its own named VFS when it runs one, so the local file's I/O\n/// stays under that host's control — the deterministic simulation in\n/// particular, which cannot judge a crash image written past it.\npub fn register_paged_vfs(\n name: &str,\n base: Option<&str>,\n path: &Path,\n reader: Arc,\n) -> crate::error::Result<()> {\n let cname = CString::new(name).map_err(|_| crate::error::Error::LTXCorrupted)?;\n let base_name = base\n .map(CString::new)\n .transpose()\n .map_err(|_| crate::error::Error::LTXCorrupted)?;\n unsafe {\n let base =\n ffi::sqlite3_vfs_find(base_name.as_ref().map_or(std::ptr::null(), |n| n.as_ptr()));\n if base.is_null() {\n return Err(crate::error::Error::Other(\"no base sqlite VFS\".into()));\n }\n let Some(path) = full_pathname(base, path) else {\n return Err(crate::error::Error::Other(\"paged vfs path\".into()));\n };\n let app = Box::into_raw(Box::new(VfsAppData {\n base,\n path,\n reader,\n hydrated: Arc::new(Mutex::new(HashSet::new())),\n open_files: std::sync::atomic::AtomicUsize::new(0),\n retired: std::sync::atomic::AtomicBool::new(false),\n }));\n let vfs = Box::into_raw(Box::new(ffi::sqlite3_vfs {\n iVersion: 2,\n szOsFile: std::mem::size_of::() as c_int,\n mxPathname: (*base).mxPathname,\n pNext: std::ptr::null_mut(),\n zName: cname.into_raw(),\n pAppData: app as *mut c_void,\n xOpen: Some(x_open),\n xDelete: Some(x_delete),\n xAccess: Some(x_access),\n xFullPathname: Some(x_full_pathname),\n xDlOpen: (*base).xDlOpen,\n xDlError: (*base).xDlError,\n xDlSym: (*base).xDlSym,\n xDlClose: (*base).xDlClose,\n xRandomness: Some(x_randomness),\n xSleep: Some(x_sleep),\n xCurrentTime: Some(x_current_time),\n xGetLastError: Some(x_get_last_error),\n xCurrentTimeInt64: None,\n xSetSystemCall: None,\n xGetSystemCall: None,\n xNextSystemCall: None,\n }));\n let rc = ffi::sqlite3_vfs_register(vfs, 0);\n if rc != ffi::SQLITE_OK {\n return Err(crate::error::Error::Other(\n format!(\"register vfs {name}: {rc}\").into(),\n ));\n }\n }\n Ok(())\n}\n\n/// How much of the cut the cell's file holds.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct Hydration {\n /// Pages of the cut the file holds (the lock page never counts).\n pub hydrated: u32,\n /// Pages the cut has, less the lock page.\n pub total: u32,\n /// Faults the registration served so far, foreground and background.\n pub faults: u64,\n}\n\nimpl Hydration {\n pub fn complete(&self) -> bool {\n self.hydrated >= self.total\n }\n}\n\n/// The registration's hydration count, or `None` for an unknown name.\npub fn hydration(name: &str) -> Option {\n let cname = CString::new(name).ok()?;\n unsafe {\n let vfs = ffi::sqlite3_vfs_find(cname.as_ptr());\n if vfs.is_null() {\n return None;\n }\n let app = &*((*vfs).pAppData as *mut VfsAppData);\n let commit = app.reader.commit();\n let lock = lock_pgno(app.reader.page_size());\n let hydrated = app.hydrated.lock().unwrap();\n Some(Hydration {\n hydrated: (1..=commit)\n .filter(|p| *p != lock && hydrated.contains(p))\n .count() as u32,\n total: commit - u32::from(lock <= commit),\n faults: app.reader.faults(),\n })\n }\n}\n\n/// Faults the rest of the cut in, a run at a time, through the VFS's own\n/// read path: a connection on the cell's file whose reads are issued below\n/// the pager, so every page arrives the way a foreground fault brings it and\n/// nothing else changes. The caller paces the steps and stops them; a\n/// hydrated cell reads only its local file, and its faults are inert.\npub struct Hydrator {\n conn: rusqlite::Connection,\n name: String,\n cursor: u32,\n}\n\nimpl Hydrator {\n pub fn open(name: &str, path: &Path) -> crate::error::Result {\n let conn = rusqlite::Connection::open_with_flags_and_vfs(\n path,\n rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,\n name,\n )\n .map_err(|error| crate::error::Error::Other(format!(\"hydrator open: {error}\").into()))?;\n Ok(Self {\n conn,\n name: name.to_string(),\n cursor: 1,\n })\n }\n\n /// Faults in the next hole and the holes after it until at least `pages`\n /// pages arrived (a fault brings a run, so one read can bring many) or\n /// the cut is complete; returns the count afterwards.\n pub fn step(&mut self, pages: u32) -> crate::error::Result {\n let cname =\n CString::new(self.name.as_str()).map_err(|_| crate::error::Error::LTXCorrupted)?;\n let (page_size, commit, lock) = unsafe {\n let vfs = ffi::sqlite3_vfs_find(cname.as_ptr());\n if vfs.is_null() {\n return Err(crate::error::Error::Other(\"hydrator: vfs is gone\".into()));\n }\n let app = &*((*vfs).pAppData as *mut VfsAppData);\n let page_size = app.reader.page_size();\n let commit = app.reader.commit();\n (page_size, commit, lock_pgno(page_size))\n };\n let mut file: *mut ffi::sqlite3_file = std::ptr::null_mut();\n // SAFETY: `conn` is open, and FILE_POINTER hands out its main file.\n let rc = unsafe {\n ffi::sqlite3_file_control(\n self.conn.handle(),\n c\"main\".as_ptr(),\n ffi::SQLITE_FCNTL_FILE_POINTER,\n (&mut file as *mut *mut ffi::sqlite3_file).cast(),\n )\n };\n if rc != ffi::SQLITE_OK || file.is_null() {\n return Err(crate::error::Error::Other(\"hydrator: no main file\".into()));\n }\n let mut page = vec![0u8; page_size as usize];\n let gone = || crate::error::Error::Other(\"hydrator: vfs is gone\".into());\n let start = hydration(&self.name).ok_or_else(gone)?.hydrated;\n // The count is recomputed only after a read that faulted: a page the\n // file already holds costs one set lookup, never a walk of the cut,\n // or a whale's step would be quadratic in its pages (the fleet's fill\n // made no progress in ten minutes).\n while self.cursor <= commit {\n let pgno = self.cursor;\n self.cursor += 1;\n if pgno == lock {\n continue;\n }\n let held = unsafe {\n let app = &*((*ffi::sqlite3_vfs_find(cname.as_ptr())).pAppData as *mut VfsAppData);\n let hydrated = app.hydrated.lock().unwrap();\n hydrated.contains(&pgno)\n };\n if held {\n continue;\n }\n // SAFETY: `file` is the live main file and `page` holds a page.\n let rc = unsafe {\n let read = (*(*file).pMethods).xRead.expect(\"xRead\");\n read(\n file,\n page.as_mut_ptr().cast(),\n page_size as c_int,\n (i64::from(pgno) - 1) * i64::from(page_size),\n )\n };\n if rc != ffi::SQLITE_OK {\n return Err(crate::error::Error::Other(\n format!(\"hydrate page {pgno}: sqlite rc {rc}\").into(),\n ));\n }\n if hydration(&self.name).ok_or_else(gone)?.hydrated - start >= pages {\n break;\n }\n }\n hydration(&self.name).ok_or_else(gone)\n }\n}\n\n/// Unregisters and frees a paged VFS registered by [`register_paged_vfs`].\npub fn unregister_paged_vfs(name: &str) -> crate::error::Result<()> {\n let cname = CString::new(name).map_err(|_| crate::error::Error::LTXCorrupted)?;\n unsafe {\n let vfs = ffi::sqlite3_vfs_find(cname.as_ptr());\n if vfs.is_null() {\n return Ok(());\n }\n ffi::sqlite3_vfs_unregister(vfs);\n let app = &*((*vfs).pAppData as *mut VfsAppData);\n app.retired.store(true, Ordering::Release);\n if app.open_files.load(Ordering::Acquire) == 0 {\n free_vfs(vfs);\n }\n }\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",1794598034304,"sidebar","expanded"]