[{"_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/db.rs","oid","f4b26239848fd24ff9f3dfd12838f2878b59cd8c","size",110531,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! db.rs — SQLite database lifecycle: WAL-mode setup, the long-running read-lock\n//! checkpoint takeover, the WAL→LTX capture loop, and manual checkpointing.\n//! Ported from litestream@v0.5.11 `db.go`.\n//!\n//! The highest-risk interaction is a long-running read transaction with a manual\n//! `PRAGMA wal_checkpoint`, so the implementation uses independently tested steps.\n//!\n//! ## Async/blocking shape\n//! `rusqlite::Connection` is `!Sync` and a `rusqlite::Transaction` borrows its\n//! `Connection`. Rather than thread a borrow across `.await`, this module keeps\n//! the capture API **synchronous** and owns the connection directly: the\n//! long-running read transaction is held with raw `BEGIN`/`ROLLBACK` SQL plus a\n//! `read_lock_held` flag — exactly as Go does (`acquireReadLock` runs `BEGIN` +\n//! `SELECT COUNT(1)`, db.go:956-976; `releaseReadLock` rolls back, db.go:979-992).\n//! `Replica` drives `sync()` and `checkpoint()` from a blocking context\n//! (`spawn_blocking` or a dedicated DB thread). This sidesteps the !Sync/borrow\n//! problem entirely and makes the idempotent-release behavior (issue #934) a\n//! trivial flag check.\n//!\n//! ## What this implements (the functional capture path)\n//! - `open`/`init`: WAL-mode DSN, `wal_autocheckpoint(0)`, control tables, read\n//! page size, acquire the read lock, ensure the WAL has ≥1 frame.\n//! - `acquire_read_lock`/`release_read_lock`: the checkpoint takeover; release is\n//! idempotent (db.go:979-992 / issue #934).\n//! - `sync` → `verify` → `sync_inner` → `write_ltx_from_wal`/`write_ltx_from_db`:\n//! diff the real WAL against the last LTX position and write the next L0 LTX\n//! file (`db.go:1517-1723`), with atomic tmp→rename and the pos cache.\n//! - `verify`: the snapshot-on-continuity-break branch lattice\n//! (`db.go:1296-1436`), including the issue #900 / #927 edge cases.\n//! - `checkpoint`/`exec_checkpoint`: release→`PRAGMA wal_checkpoint()`→\n//! re-acquire (`db.go:1875-1919`) and the two-phase WAL-restart handling\n//! (`db.go:1808-1873`).\n//! - `checkpoint_if_needed`: the 3-tier policy + the three anti-feedback flags\n//! (`synced_since_checkpoint`/`synced_to_wal_end`/`last_synced_wal_offset`,\n//! issues #896/#927/#997).\n//! - Litestream #1292 / celld #150: seal passive checkpoints with a writer\n//! barrier, retain database-growth pages, and snapshot truncate boundaries.\n//! - `crc64`, `pos` (cached + `LTXError` mapping), `reset_local_state`,\n//! `snapshot_to_writer`.\n//!\n//! ## Deferred work outside the functional path\n//! - `setPersistWAL` (the `unsafe` `sqlite3_file_control(SQLITE_FCNTL_PERSIST_WAL)`\n//! FFI): only matters when *all* connections close and SQLite\n//! would delete the WAL; the capture path keeps its own connection open.\n//! - The background monitor loop, `ensure_exists`, `sync_status`, `sync_and_wait`,\n//! retention, and retry behavior are not implemented.\n//! - A `loom` model of the lock protocol.\n\nuse crate::error::{new_ltx_error, Error, Result};\nuse crate::ltx::{self, lock_pgno, Crc64};\nuse crate::wal::{WalError, WalReader};\nuse crate::{\n ltx_file_path, ltx_level_dir, Pos, CHECKPOINT_MODE_PASSIVE, CHECKPOINT_MODE_RESTART,\n CHECKPOINT_MODE_TRUNCATE, META_DIR_SUFFIX, TXID, WAL_FRAME_HEADER_SIZE, WAL_HEADER_SIZE,\n};\nuse rusqlite::ffi;\nuse rusqlite::Connection;\nuse rusqlite::OpenFlags;\nuse std::collections::HashMap;\nuse std::ffi::c_int;\nuse std::path::{Path, PathBuf};\nuse std::time::{Duration, SystemTime, UNIX_EPOCH};\n\n/// Maps a `rusqlite::Error` into the crate error type.\nfn sql_err(e: rusqlite::Error) -> Error {\n Error::Other(Box::new(e))\n}\n\nfn disable_lookaside(conn: &Connection) -> Result<()> {\n // SQLite's default lookaside arena reserves 48 KiB for each connection.\n // Every resident cell owns both connections below, so the unused arenas\n // repeat a fixed cost at fleet density. LTX prepares a fixed, small SQL\n // vocabulary, and the measured durable-write throughput stays flat when\n // general allocations replace these arenas.\n //\n // SAFETY: the caller passes a newly opened connection before its first\n // SQLite operation, so no lookaside slot can be in use.\n let result = unsafe {\n ffi::sqlite3_db_config(\n conn.handle(),\n ffi::SQLITE_DBCONFIG_LOOKASIDE,\n std::ptr::null_mut::(),\n 0,\n 0,\n )\n };\n if result != ffi::SQLITE_OK {\n return Err(Error::Other(\n format!(\"disable SQLite lookaside failed: {result}\").into(),\n ));\n }\n Ok(())\n}\n\n/// SQLite checkpoint mode. Replaces Go's stringly-typed `mode` param; `Display`\n/// interpolates straight into `PRAGMA wal_checkpoint()` (db.go:1905), so it\n/// must render exactly `PASSIVE`/`FULL`/`RESTART`/`TRUNCATE`.\n///\n/// Ported from litestream@v0.5.11 litestream.go:22-28. RESTART was removed from\n/// automatic use (issue #724) but is still callable (`crc64` forces it), so all\n/// four variants are kept.\n#[derive(Debug, Clone, Copy, PartialEq, Eq)]\npub enum CheckpointMode {\n /// Non-blocking checkpoint; skips if there are active transactions.\n Passive,\n /// Like RESTART but does not block new transactions before flushing.\n Full,\n /// Blocks new transactions, flushes, and restarts the WAL.\n Restart,\n /// Like RESTART plus truncates the WAL file to zero length.\n Truncate,\n}\n\n/// The three columns of `PRAGMA wal_checkpoint`: whether a lock stopped the\n/// checkpoint, the frames in the WAL, and the frames backfilled.\n#[derive(Clone, Copy, Debug)]\nstruct CheckpointPragma {\n busy: i64,\n wal_frames: i64,\n backfilled: i64,\n}\n\nimpl std::fmt::Display for CheckpointMode {\n fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {\n let s = match self {\n CheckpointMode::Passive => CHECKPOINT_MODE_PASSIVE,\n CheckpointMode::Full => crate::CHECKPOINT_MODE_FULL,\n CheckpointMode::Restart => CHECKPOINT_MODE_RESTART,\n CheckpointMode::Truncate => CHECKPOINT_MODE_TRUNCATE,\n };\n f.write_str(s)\n }\n}\n\n/// `verify()`'s decision result (db.go:1509-1515).\n#[derive(Debug, Clone, Default)]\nstruct SyncInfo {\n /// End of the previous LTX read (byte offset into the WAL).\n offset: i64,\n salt1: u32,\n salt2: u32,\n /// Database page count recorded by the previous LTX file.\n prev_commit: u32,\n /// If true, a full snapshot is required.\n snapshotting: bool,\n /// Reason for the snapshot, for logging and diagnostics.\n reason: String,\n}\n\n/// Databases larger than this many pages (4 MiB at 4 KiB) must outgrow\n/// themselves in the WAL before a TRUNCATE checkpoint, since the checkpoint\n/// ends in a boundary image of the whole database. Smaller ones keep the\n/// absolute `truncate_page_n` threshold.\nconst RELATIVE_TRUNCATE_PAGES: u32 = 1024;\n\n/// A SQLite database managed for replication.\n///\n/// Owns the connection (WAL mode, auto-checkpoint disabled) so litestream — not\n/// SQLite — decides when the WAL is checkpointed, and holds the long-running read\n/// transaction that takes over checkpointing.\n///\n/// Ported from `DB` (db.go:64-198). The synchronous `!Sync` connection stays on\n/// the thread that drives it.\n/// What `verify` reads back from the last L0 this instance wrote: the\n/// header fields, plus the final consumed WAL frame's page. The ported\n/// check decodes the whole L0 and searches its pages for that frame; the\n/// L0's entry for the final frame's page IS that frame (page-map keeps\n/// the last write per page, and nothing follows the final frame), so\n/// comparing against the cached frame answers the same question — and a\n/// spurious mismatch only forces a snapshot, the safe direction.\nfn snapshot_reason_code(reason: &str) -> u8 {\n match reason {\n \"\" => 1, // first sync leaves the default reason empty\n \"wal truncated by another process\" => 2,\n \"wal header salt reset, snapshotting\" => 3,\n \"last page does not exist in last ltx file, wal overwritten by another process\" => 4,\n \"full or restart checkpoint detected, snapshotting\" => 5,\n \"checkpoint boundary snapshot\" => 6,\n \"WAL restarted before passive checkpoint barrier\" => 8,\n _ => 7,\n }\n}\n\n#[derive(Clone, Debug)]\nstruct LastL0Header {\n wal_offset: i64,\n wal_size: i64,\n wal_salt1: u32,\n wal_salt2: u32,\n commit: u32,\n final_pgno: u32,\n final_page: Vec,\n}\n\n/// Per-phase wall time of one `Db::sync` call, in microseconds.\n#[derive(Clone, Copy, Debug, Default)]\npub struct SyncTiming {\n /// Control-table self-heal and WAL existence checks.\n pub prepare_us: u64,\n /// The two halves of `prepare_us`: the schema-version pragma (plus\n /// the DDL when it fires) and the WAL existence stat, so a field\n /// inflation names its syscall.\n pub schema_check_us: u64,\n pub wal_exists_us: u64,\n /// `verify`: the WAL scan and continuity validation.\n pub verify_us: u64,\n /// `sync_inner` minus the fsync: WAL page reads, LTX encode, the file\n /// write and rename.\n pub encode_write_us: u64,\n /// The sub-phases of `encode_write_us`, so a field inflation names its\n /// line instead of hiding in the aggregate: position resolution, the\n /// WAL read (tail or full), page-map plus page collection, the LTX\n /// encode, and the file write-and-rename.\n pub pos_us: u64,\n pub wal_read_us: u64,\n pub map_collect_us: u64,\n pub ltx_encode_us: u64,\n pub file_write_us: u64,\n /// The WAL length seen by this call, for scaling.\n pub wal_len_bytes: u64,\n /// True when this call snapshotted (the full-database path).\n pub snapshot: bool,\n /// Which WAL read ran: 0 valid tail, 1 full because snapshotting, 2 valid\n /// prefix because the offset sat at the WAL start, 3 full because the\n /// bounded read fell back. `wal_read_bytes` is what that read actually\n /// transferred, including a bounded probe past the valid checksum chain.\n pub wal_read_kind: u8,\n pub wal_read_bytes: u64,\n /// The bytes the sync allocated to hold what it read: the tail plus the\n /// header on the tail path, the file otherwise.\n pub wal_image_bytes: u64,\n /// Which verify branch forced the snapshot: 0 none, 1 first sync,\n /// 2 wal truncated by another process, 3 salt reset, 4 last page\n /// missing from the last L0, 5 full or restart checkpoint detected,\n /// 6 checkpoint boundary, 7 other, 8 WAL restarted before the passive\n /// checkpoint barrier.\n pub snapshot_reason: u8,\n /// The cut file's fsync (zero under lazy capture).\n pub fsync_us: u64,\n /// `checkpoint_if_needed`: SQLite's checkpoint, including its own\n /// writes and fsyncs through the SQLite VFS.\n pub checkpoint_us: u64,\n /// One when this call ran a checkpoint pragma or swallowed its busy\n /// error, else zero. Counters rather than flags, so the round ledger\n /// sums them across cells like every other field here. A Queue soak\n /// showed 14% of checkpoint rounds repeating on the next round with only\n /// the sealing frame captured, and `checkpoint_us` alone could not say\n /// whether the backfill was short, the pragma was busy, or the sealing\n /// write failed to restart the WAL.\n pub checkpoint_runs: u64,\n /// The frames the WAL held when the pragma ran and the frames it had\n /// backfilled when it returned: `PRAGMA wal_checkpoint`'s second and\n /// third columns. Equal for a complete backfill; a shortfall names a\n /// reader that pinned the WAL.\n pub checkpoint_wal_frames: u64,\n pub checkpoint_backfilled: u64,\n /// One when the pragma reported a lock it could not take (its first\n /// column), and one when the checkpoint path failed with `SQLITE_BUSY`\n /// and the sync swallowed the error.\n pub checkpoint_busy: u64,\n pub checkpoint_busy_errors: u64,\n /// One when the WAL header changed across the sealing write, so the\n /// logical WAL restarted at its header.\n pub checkpoint_restarts: u64,\n}\n\npub struct Db {\n host: crate::LtxHost,\n path: PathBuf,\n meta_path: PathBuf,\n /// Main connection: writes, PRAGMAs, checkpoints, page-size reads.\n conn: Connection,\n /// Dedicated connection that holds the long-running read transaction.\n ///\n /// Go's `db.db` is a `*sql.DB` **connection pool**: the read lock (`db.rtx`)\n /// lives on one pooled connection while every other `db.db.Exec` grabs a\n /// *different* connection from the pool. A single `rusqlite::Connection`\n /// cannot do that — an `INSERT` issued on the same connection that holds the\n /// read transaction would be buffered in that transaction and not flushed to\n /// the WAL (it would not write a fresh WAL page after a TRUNCATE checkpoint).\n /// So the read lock gets its own connection, mirroring the pool's separation.\n rtx_conn: Connection,\n page_size: u32,\n\n /// `true` while the long-running read transaction is open (db.go:70 `rtx`).\n /// We track a flag rather than holding a borrowing `rusqlite::Transaction`.\n read_lock_held: bool,\n\n /// The last `sync` call's per-phase wall time. Telemetry only: the\n /// closed-book capture ledger reads it; nothing branches on it.\n last_sync_timing: SyncTiming,\n\n // ── Tunables (db.go:131-197) ────────────────────────────────────────────\n /// PASSIVE checkpoint threshold, in pages (db.go:34 default 1000).\n pub min_checkpoint_page_n: u32,\n /// TRUNCATE checkpoint threshold, in pages; 0 disables (db.go:35 default).\n pub truncate_page_n: u32,\n /// Time-based PASSIVE checkpoint interval; 0 disables (db.go:32 default 60s).\n pub checkpoint_interval: Duration,\n /// Busy timeout for SQLite locks (db.go:33 default 1s).\n pub busy_timeout: Duration,\n\n // ── Anti-feedback-loop bookkeeping (issues #896/#927/#997) ──────────────\n /// True once data has synced since the last checkpoint (#896, db.go:80).\n synced_since_checkpoint: bool,\n /// True if the last sync reached the exact WAL EOF (#927, db.go:88).\n synced_to_wal_end: bool,\n /// Logical end of WAL content after the last sync = `WALOffset + WALSize`\n /// from the last LTX (#997, db.go:96). Used for checkpoint thresholds\n /// instead of file size (stale post-checkpoint frames inflate file size).\n last_synced_wal_offset: i64,\n /// The database's page count as the last capture saw it, for the\n /// truncate guard: a second stat per sync would break the syscall\n /// ledger, and the capture already read the size.\n last_db_pages: u32,\n /// Logical WAL offset through which the last checkpoint backfilled the\n /// database, in the current WAL's coordinates; `WAL_HEADER_SIZE` after\n /// a restart and 0 before any checkpoint. The passive threshold counts\n /// frames appended past this point. The port compared the whole logical\n /// size, which re-fired the checkpoint on every sync after a checkpoint\n /// whose sealing write could not restart the WAL: the backfilled frames\n /// kept counting, so a Queue owner paid a writer barrier, a sealing\n /// write, and a fresh LTX per round until a restart succeeded.\n checkpointed_wal_offset: i64,\n /// The schema version `ensure_control_tables` last verified. The\n /// self-heal exists for a swept `sqlite_schema`, and any sweep bumps\n /// SQLite's schema version, so an unchanged version proves the tables\n /// still stand without re-running DDL through the parser every capture\n /// (~0.8 ms per sync in the 2026-08-25 closed-book ledger).\n verified_schema_version: Option,\n /// The header fields of the last L0 this instance wrote, keyed by its\n /// TXID. `verify` needs exactly these to place the next incremental\n /// read; re-reading the whole L0 file to parse its header cost more\n /// than the header is long. A miss (restore, compaction, reopen)\n /// falls back to the file.\n last_l0_header: Option<(TXID, LastL0Header)>,\n\n /// Cached L0 position; `None` = invalid (db.go:106-109).\n pos_cache: Option,\n /// The L0 level directory exists: created on the first cut (or after a\n /// `NotFound` on a later one), not probed with `mkdir` on every sync.\n l0_dir_ready: bool,\n /// The WAL, opened once. Every sync used to open it three times and stat\n /// it five; each was a path walk on a directory the node fsyncs thousands\n /// of times a second. See [`Db::with_wal_file`].\n wal_file: Option,\n /// Last L0 `FileInfo` (db.go:99-102; only L0 is tracked in the one-shot).\n max_l0_file_info: Option,\n}\n\n/// DDL for the replication control tables managed inside each database:\n/// `_litestream_seq` forces WAL writes when empty; `_litestream_lock` forces\n/// a write lock during sync (db.go:857-864). The text is byte-visible:\n/// `sqlite_schema` stores it verbatim in a replicated page, so it must stay\n/// character-identical to litestream's, and one copy serves every creation\n/// site.\nconst CONTROL_TABLES_DDL: &str =\n \"CREATE TABLE IF NOT EXISTS _litestream_seq (id INTEGER PRIMARY KEY, seq INTEGER);\\\n CREATE TABLE IF NOT EXISTS _litestream_lock (id INTEGER);\";\n\n/// True for exactly the two control tables above, compared without case the\n/// way SQLite compares identifiers. Application-facing paths (deleteAll\n/// sweeps, the SQL authorizer) must exempt these names, so the predicate\n/// lives here with the tables. Exact names, not a `_litestream_` prefix:\n/// ltx is a port of a pinned litestream, and workerd reserves only `_cf_`,\n/// so any other name stays reachable by application SQL.\npub fn is_control_table(name: &str) -> bool {\n name.eq_ignore_ascii_case(\"_litestream_seq\") || name.eq_ignore_ascii_case(\"_litestream_lock\")\n}\n\nimpl Db {\n /// Default minimum-checkpoint page count (`DefaultMinCheckpointPageN`,\n /// db.go:34).\n pub const DEFAULT_MIN_CHECKPOINT_PAGE_N: u32 = 1000;\n /// Default truncate page count (`DefaultTruncatePageN`, db.go:35).\n pub const DEFAULT_TRUNCATE_PAGE_N: u32 = 121_359;\n\n /// Default checkpoint interval (`DefaultCheckpointInterval`, db.go:32).\n pub const DEFAULT_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(60);\n /// Default busy timeout (`DefaultBusyTimeout`, db.go:33).\n pub const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(1);\n\n /// Opens and initializes the database with litestream's WAL-mode setup and\n /// acquires the long-running read lock.\n ///\n /// The host must complete this call successfully before opening any\n /// other SQLite connection that can write to this database. After a\n /// crash, an earlier write can hide a torn WAL header, so resumed capture\n /// can skip new commits when existing LTX files retain the previous\n /// capture position. An incomplete recovery checkpoint returns an error,\n /// including when another connection keeps the checkpoint busy.\n ///\n /// Ported from `DB.init` (db.go:795-911) — the connection-setup half plus the\n /// read-lock acquire and `ensureWALExists`.\n pub fn open(path: impl AsRef) -> Result {\n Self::open_with_host(path, crate::LtxHost::default())\n }\n\n /// Opens the database with an injected clock and executor host.\n ///\n /// This call must succeed before another writer opens the database.\n /// See [`Db::open`] for the startup ordering requirement.\n pub fn open_with_host(path: impl AsRef, host: crate::LtxHost) -> Result {\n Self::open_with_host_and_optional_vfs(path, host, None)\n }\n\n /// Opens both managed connections through a named SQLite VFS. Used by the\n /// fault-injection VFS in tests and by paged restore's fault-in VFS.\n ///\n /// This call must succeed before another writer opens the database.\n /// See [`Db::open`] for the startup ordering requirement.\n pub fn open_with_host_and_vfs(\n path: impl AsRef,\n host: crate::LtxHost,\n vfs: &str,\n ) -> Result {\n Self::open_with_host_and_optional_vfs(path, host, Some(vfs))\n }\n\n fn open_with_host_and_optional_vfs(\n path: impl AsRef,\n host: crate::LtxHost,\n vfs: Option<&str>,\n ) -> Result {\n let path = path.as_ref().to_path_buf();\n let meta_path = Self::meta_path_for(&path);\n\n let open = |path: &Path| match vfs {\n Some(vfs) => Connection::open_with_flags_and_vfs(path, OpenFlags::default(), vfs),\n None => Connection::open(path),\n };\n let conn = open(&path).map_err(sql_err)?;\n disable_lookaside(&conn)?;\n // Cap the page cache at 64 KiB; SQLite's default (-2000) lets each\n // connection grow to 2 MiB, and every resident cell owns this\n // connection, so the default repeats a dead-weight cost at fleet\n // density. The connection only runs PRAGMAs, the sealing write to\n // the control tables, and checkpoints, and a checkpoint streams WAL\n // frames into the database without revisiting pages, so a larger\n // cache buys no reuse. 16 default-size pages still cover the\n // control tables and their schema pages.\n conn.pragma_update(None, \"cache_size\", -64)\n .map_err(sql_err)?;\n\n // DSN pragmas: busy_timeout + wal_autocheckpoint(0) (db.go:818).\n // autocheckpoint MUST be 0 because litestream owns checkpointing.\n // Per-connection, not per-file: the cell's own writer\n // connection keeps SQLite's default 1000-page autocheckpoint, which\n // is safe because this connection's long-running read transaction\n // pins a WAL read mark, so a PASSIVE checkpoint elsewhere can\n // backfill but never reset or truncate the WAL under the reader.\n conn.busy_timeout(Self::DEFAULT_BUSY_TIMEOUT)\n .map_err(sql_err)?;\n conn.pragma_update(None, \"wal_autocheckpoint\", 0)\n .map_err(sql_err)?;\n\n // Enable WAL; SQLite returns the new mode on success (db.go:849-853).\n let mode: String = conn\n .query_row(\"PRAGMA journal_mode=WAL\", [], |r| r.get(0))\n .map_err(sql_err)?;\n if mode != \"wal\" {\n return Err(Error::Other(\n format!(\"enable wal failed, mode={mode:?}\").into(),\n ));\n }\n\n // Dedicated read-lock connection (mirrors a second pooled connection).\n let rtx_conn = open(&path).map_err(sql_err)?;\n disable_lookaside(&rtx_conn)?;\n // This connection runs one tiny read to take the read lock and then\n // idles for the whole residency, so SQLite's default page-cache limit\n // (-2000 = up to 2 MiB) is dead weight repeated per resident cell.\n // Cap it at a few pages; no user query or checkpoint runs here, so a\n // smaller cache cannot slow either path.\n rtx_conn\n .pragma_update(None, \"cache_size\", -16)\n .map_err(sql_err)?;\n rtx_conn\n .busy_timeout(Self::DEFAULT_BUSY_TIMEOUT)\n .map_err(sql_err)?;\n\n let mut db = Db {\n host,\n path,\n meta_path,\n conn,\n rtx_conn,\n page_size: 0,\n read_lock_held: false,\n last_sync_timing: SyncTiming::default(),\n min_checkpoint_page_n: Self::DEFAULT_MIN_CHECKPOINT_PAGE_N,\n truncate_page_n: Self::DEFAULT_TRUNCATE_PAGE_N,\n checkpoint_interval: Self::DEFAULT_CHECKPOINT_INTERVAL,\n busy_timeout: Self::DEFAULT_BUSY_TIMEOUT,\n synced_since_checkpoint: false,\n synced_to_wal_end: false,\n last_synced_wal_offset: 0,\n last_db_pages: 0,\n checkpointed_wal_offset: 0,\n verified_schema_version: None,\n last_l0_header: None,\n pos_cache: None,\n l0_dir_ready: false,\n wal_file: None,\n max_l0_file_info: None,\n };\n\n // Recover before any DDL can rewrite a torn header with reused salts,\n // and before our long-lived reader can block the checkpoint.\n db.recover_wal_header()?;\n db.conn.execute_batch(CONTROL_TABLES_DDL).map_err(sql_err)?;\n\n // Start the long-running read transaction (db.go:867-871).\n db.acquire_read_lock()?;\n\n // Read page size (db.go:874-878).\n let page_size: i64 = db\n .conn\n .query_row(\"PRAGMA page_size\", [], |r| r.get(0))\n .map_err(sql_err)?;\n if page_size <= 0 {\n return Err(Error::Other(\n format!(\"invalid db page size: {page_size}\").into(),\n ));\n }\n // Validated > 0 above; SQLite page sizes are <= 65536.\n db.page_size = page_size as u32;\n\n // Ensure the meta directory exists (db.go:880-883).\n db.host.create_dir_all(&db.meta_path)?;\n\n // Clear crash-leftover temp files (db.go:576).\n remove_tmp_files(&db.host, &db.meta_path)?;\n\n // Ensure the WAL has at least one frame (db.go:886-888).\n db.ensure_wal_exists()?;\n\n Ok(db)\n }\n\n // ── Paths ──────────────────────────────────────────────────────────────\n\n /// Computes the litestream meta-directory path (`.-litestream`) for a\n /// database file path, without opening it. Public so a host can locate (e.g.\n /// to wipe, during a hard recovery) the meta dir of a database it no longer\n /// has an open [`Db`] handle for. Mirrors `DB.MetaPath` (db.go:292) at the\n /// path level.\n pub fn meta_path_for_path(path: impl AsRef) -> PathBuf {\n Self::meta_path_for(path.as_ref())\n }\n\n fn meta_path_for(path: &Path) -> PathBuf {\n // Go: filepath.Join(dir, \".\"+file+MetaDirSuffix) (db.go:206).\n let dir = path.parent();\n let file = path.file_name().map(|s| s.to_owned()).unwrap_or_default();\n let mut name = std::ffi::OsString::from(\".\");\n name.push(&file);\n name.push(META_DIR_SUFFIX);\n match dir {\n Some(d) if !d.as_os_str().is_empty() => d.join(name),\n _ => PathBuf::from(name),\n }\n }\n\n /// The database file path. Ported from `DB.Path` (db.go:275).\n pub fn path(&self) -> &Path {\n &self.path\n }\n\n /// Path to the litestream meta directory (`.-litestream`).\n /// Ported from `DB.MetaPath` (db.go:292).\n pub fn meta_path(&self) -> &Path {\n &self.meta_path\n }\n\n /// Path to SQLite's WAL sidecar file (`-wal`).\n /// Ported from `DB.WALPath` (db.go:287-289).\n pub fn wal_path(&self) -> PathBuf {\n let mut s = self.path.clone().into_os_string();\n s.push(\"-wal\");\n PathBuf::from(s)\n }\n\n /// Root LTX directory (`/ltx`). Ported from `DB.LTXDir` (db.go:302).\n fn ltx_dir(&self) -> String {\n crate::ltx_dir(&self.meta_path.to_string_lossy())\n }\n\n /// LTX level sub-directory. Ported from `DB.LTXLevelDir` (db.go:332).\n fn ltx_level_dir(&self, level: u32) -> String {\n ltx_level_dir(&self.meta_path.to_string_lossy(), level)\n }\n\n /// Local path of a single LTX file. Ported from `DB.LTXPath` (db.go:338).\n ///\n /// `pub` so the [`crate::replica::Replica`] can read the local L0 LTX files\n /// the capture loop wrote and upload them (the Go exported `DB.LTXPath`,\n /// used by `Replica.uploadLTXFile`, replica.go:183).\n pub fn ltx_path(&self, level: u32, min_txid: TXID, max_txid: TXID) -> String {\n ltx_file_path(&self.meta_path.to_string_lossy(), level, min_txid, max_txid)\n }\n\n /// Reads one local LTX file through the host filesystem.\n pub fn read_ltx_file(&self, level: u32, min_txid: TXID, max_txid: TXID) -> Result> {\n let path = self.ltx_path(level, min_txid, max_txid);\n Ok(self.host.read(Path::new(&path))?)\n }\n\n /// The SQLite page size, in bytes. Ported from `DB.PageSize` (db.go:445).\n pub fn page_size(&self) -> u32 {\n self.page_size\n }\n\n // ── Read-lock takeover (checkpoint prevention) ─────────────────────────\n\n /// Begins a long-running read transaction to prevent external checkpoints.\n ///\n /// Ported from `acquireReadLock` (db.go:956-976): `BEGIN` then `SELECT\n /// COUNT(1) FROM _litestream_seq` to obtain the SHARED read lock. Held with\n /// raw SQL + a flag instead of a borrowing `rusqlite::Transaction` (see\n /// module docs). Idempotent: a no-op if already held.\n fn acquire_read_lock(&mut self) -> Result<()> {\n if self.read_lock_held {\n return Ok(());\n }\n self.rtx_conn\n .prepare_cached(\"BEGIN\")\n .and_then(|mut statement| statement.execute([]))\n .map_err(sql_err)?;\n // Execute a read query to obtain the read lock. On failure, roll back.\n if let Err(e) = self\n .rtx_conn\n .query_row(\"SELECT COUNT(1) FROM _litestream_seq\", [], |r| {\n r.get::<_, i64>(0)\n })\n {\n let _ = self.rtx_conn.execute_batch(\"ROLLBACK\");\n return Err(sql_err(e));\n }\n self.read_lock_held = true;\n Ok(())\n }\n\n /// Rolls back the long-running read transaction.\n ///\n /// Ported from `releaseReadLock` (db.go:979-992). Uses the `rollback` helper\n /// semantics: a \"no transaction is active\" / \"already rolled back\" error is\n /// swallowed (issue #934) — a double release must return\n /// `Ok(())`. The `read_lock_held` flag is cleared regardless.\n fn release_read_lock(&mut self) -> Result<()> {\n if !self.read_lock_held {\n return Ok(());\n }\n self.read_lock_held = false;\n rollback(&self.rtx_conn)\n }\n\n // ── WAL bootstrap ──────────────────────────────────────────────────────\n\n /// Recreates the control tables if they are missing. They are created at\n /// open, but an application-level sweep of `sqlite_schema` can drop them\n /// (they do not carry the protected `_cf_` prefix); `CREATE TABLE IF NOT\n /// EXISTS` on an existing table is a no-op, so this is safe to run on\n /// every capture.\n fn ensure_control_tables(&mut self) -> Result<()> {\n // A swept control table is a schema change, and every schema change\n // bumps SQLite's schema version — so an unchanged version proves\n // the last verification still holds and the DDL (a full parse and\n // execute per statement) can be skipped on the hot capture path.\n // Through the statement cache: this guard runs on every sync, and a\n // fresh `PRAGMA` per sync was one SQL compilation per capture on the\n // fleet profile.\n let version: i64 = self\n .conn\n .prepare_cached(\"PRAGMA schema_version\")\n .and_then(|mut statement| statement.query_row([], |row| row.get(0)))\n .map_err(sql_err)?;\n if self.verified_schema_version == Some(version) {\n return Ok(());\n }\n self.conn\n .execute_batch(CONTROL_TABLES_DDL)\n .map_err(sql_err)?;\n let verified: i64 = self\n .conn\n .prepare_cached(\"PRAGMA schema_version\")\n .and_then(|mut statement| statement.query_row([], |row| row.get(0)))\n .map_err(sql_err)?;\n self.verified_schema_version = Some(verified);\n Ok(())\n }\n\n /// Runs `op` on the held WAL handle, opening it on first use. A WAL that\n /// SQLite recreated, or that a truncation shortened under a read, surfaces\n /// as `NotFound` or `UnexpectedEof`; the handle is dropped and `op` runs\n /// once more on a fresh open, so a stale handle costs one retry, never a\n /// wrong answer. Any other error also drops the handle.\n fn with_wal_file(\n &mut self,\n op: impl Fn(&mut crate::HostFile) -> std::io::Result,\n ) -> std::io::Result {\n let mut attempt = 0;\n loop {\n if self.wal_file.is_none() {\n self.wal_file = Some(self.host.open(&self.wal_path())?);\n }\n let file = self.wal_file.as_mut().expect(\"opened above\");\n match op(file) {\n Ok(value) => return Ok(value),\n Err(error) => {\n self.wal_file = None;\n let retry = attempt == 0\n && matches!(\n error.kind(),\n std::io::ErrorKind::NotFound | std::io::ErrorKind::UnexpectedEof\n );\n if !retry {\n return Err(error);\n }\n attempt += 1;\n }\n }\n }\n }\n\n /// The 32-byte WAL header through the held handle (`readWALHeader`,\n /// litestream.go:138-148: an error if the file is missing or short).\n fn wal_header_bytes(&mut self) -> Result<[u8; WAL_HEADER_SIZE]> {\n let bytes = self.with_wal_file(|file| file.read_exact_at(0, WAL_HEADER_SIZE))?;\n Ok(bytes\n .try_into()\n .expect(\"the exact read has the header size\"))\n }\n\n /// `n` bytes at `offset` of the WAL through the held handle\n /// (`readWALFileAt`, litestream.go:152-166: a short read is an error).\n fn wal_bytes_at(&mut self, offset: i64, n: i64) -> Result> {\n Ok(self.with_wal_file(|file| file.read_exact_at(offset as u64, n as usize))?)\n }\n\n /// Resets a WAL with an invalid header checksum before initialization writes.\n fn recover_wal_header(&mut self) -> Result<()> {\n let header = match self.wal_header_bytes() {\n Ok(header) => header,\n // The later ensure_wal_exists call initializes missing or\n // short WALs. Other I/O errors must remain visible to the caller.\n Err(Error::Io(error))\n if matches!(\n error.kind(),\n std::io::ErrorKind::NotFound | std::io::ErrorKind::UnexpectedEof\n ) =>\n {\n return Ok(())\n }\n Err(error) => return Err(error),\n };\n match WalReader::new(&header) {\n Ok(_) => return Ok(()),\n Err(WalError::Eof) => {}\n // SQLite rejects invalid magic before copying the salts or nCkpt.\n // With nCkpt still zero, its next write chooses fresh salts, so\n // verify detects the new generation without this checkpoint.\n // Capture still rejects the invalid header until that write.\n Err(WalError::InvalidMagic(_)) => return Ok(()),\n // SQLite normally rejects an unsupported version while enabling\n // WAL; keep any remaining header-reader errors visible.\n Err(error) => return Err(error.into()),\n }\n\n // The host must open capture before other SQLite writers. After a\n // crash, an earlier write can reuse salts and hide the torn header.\n // SQLite rejects the torn WAL contents but retains the header's\n // salts and checkpoint sequence. A bootstrap write alone can reuse\n // those salts. The stale tail then lets an existing LTX cursor skip\n // new commits. SQLite's TRUNCATE checkpoint changes the salts and\n // removes that tail even if this process stops before its first\n // capture. No volatile recovery flag can protect that gap.\n // If control-table DDL is a no-op, ensure_wal_exists supplies the\n // first write in the new generation.\n // Only open checks the header, so live syncs keep their cheap check.\n let checkpoint = self.run_checkpoint_pragma(CheckpointMode::Truncate)?;\n if checkpoint.busy != 0 {\n // A busy checkpoint leaves recovery incomplete, so opening must fail.\n return Err(sql_err(rusqlite::Error::SqliteFailure(\n ffi::Error::new(ffi::SQLITE_BUSY),\n Some(\"torn WAL recovery checkpoint is busy\".to_string()),\n )));\n }\n Ok(())\n }\n\n /// Ensures the real WAL exists and has a header.\n ///\n /// Ported from `ensureWALExists` (db.go:1199-1209): exit early if the WAL\n /// header is present; otherwise force a write to `_litestream_seq`.\n fn ensure_wal_exists(&mut self) -> Result<()> {\n if self.wal_file_size()? >= WAL_HEADER_SIZE as i64 {\n return Ok(());\n }\n self.conn\n .execute_batch(\n \"INSERT INTO _litestream_seq (id, seq) VALUES (1, 1) \\\n ON CONFLICT (id) DO UPDATE SET seq = seq + 1\",\n )\n .map_err(sql_err)?;\n Ok(())\n }\n\n /// Size of the WAL file in bytes, 0 if absent. Ported from `walFileSize`\n /// (db.go:1183-1191).\n fn wal_file_size(&mut self) -> Result {\n match self.with_wal_file(|file| file.file_len()) {\n Ok(len) => Ok(len as i64),\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),\n Err(e) => Err(e.into()),\n }\n }\n\n /// The size of the main database file in bytes, 0 if absent.\n fn db_file_size(&self) -> Result {\n match self.host.metadata(&self.path) {\n Ok(md) => Ok(md.len as i64),\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),\n Err(e) => Err(e.into()),\n }\n }\n\n // ── Position ───────────────────────────────────────────────────────────\n\n /// The highest `(min,max)` TXID pair in the L0 directory, `(0,0)` if none.\n /// Ported from `DB.MaxLTX` (db.go:363-380).\n fn max_ltx(&self) -> Result<(TXID, TXID)> {\n let dir = self.ltx_level_dir(0);\n let entries = match self.host.read_dir(Path::new(&dir)) {\n Ok(e) => e,\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((TXID(0), TXID(0))),\n Err(e) => return Err(e.into()),\n };\n let mut min_txid = TXID(0);\n let mut max_txid = TXID(0);\n for ent in entries {\n let name = ent.file_name;\n let name = name.to_string_lossy();\n if let Ok((mn, mx)) = ltx::parse_filename(&name) {\n if mx > max_txid {\n min_txid = mn;\n max_txid = mx;\n }\n }\n }\n Ok((min_txid, max_txid))\n }\n\n /// The current replication position (cached; recomputed from the max L0 file).\n ///\n /// Ported from `DB.Pos` (db.go:392-425). Wraps fs/decode failures in\n /// `LTXError` (db.go:412,418).\n pub fn pos(&mut self) -> Result {\n if let Some(p) = self.pos_cache {\n return Ok(p);\n }\n\n let (min_txid, max_txid) = self.max_ltx()?;\n if min_txid == TXID(0) {\n return Ok(Pos::ZERO); // no replication yet\n }\n\n let ltx_path = self.ltx_path(0, min_txid, max_txid);\n let bytes = match self.host.read(Path::new(<x_path)) {\n Ok(b) => b,\n Err(e) => {\n return Err(Error::Ltx(Box::new(new_ltx_error(\n \"open\",\n <x_path,\n 0,\n min_txid.0,\n max_txid.0,\n e.into(),\n ))));\n }\n };\n\n let decoded = match ltx::decode_file(&bytes) {\n Ok(d) => d,\n Err(_e) => {\n // Decode/verify failure indicates corruption (db.go:417-419).\n return Err(Error::Ltx(Box::new(new_ltx_error(\n \"verify\",\n <x_path,\n 0,\n min_txid.0,\n max_txid.0,\n Error::LTXCorrupted,\n ))));\n }\n };\n\n let pos = Pos::new(decoded.header.max_txid, decoded.trailer.post_apply_checksum);\n self.pos_cache = Some(pos);\n Ok(pos)\n }\n\n /// Clears the cached position so the next `pos()` recomputes from disk.\n /// Ported from `invalidatePosCache` (db.go:430-434).\n fn invalidate_pos_cache(&mut self) {\n self.pos_cache = None;\n }\n\n /// Removes local LTX files, forcing a fresh snapshot on the next sync.\n /// Ported from `DB.ResetLocalState` (db.go:309-328).\n pub fn reset_local_state(&mut self) -> Result<()> {\n match self.host.remove_dir_all(Path::new(&self.ltx_dir())) {\n Ok(()) => {}\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}\n Err(e) => return Err(e.into()),\n }\n self.max_l0_file_info = None;\n self.invalidate_pos_cache();\n Ok(())\n }\n\n /// Clears the local L0 directory and seeds it with a single baseline L0 LTX\n /// file (`data` for the `min_txid`..`max_txid` range), atomically. The next\n /// [`Db::sync`] then sees the baseline does not match the real WAL and writes\n /// a fresh snapshot at the current database state.\n ///\n /// This is the file-writing tail of `checkDatabaseBehindReplica`\n /// (db.go:1241-1293): clear L0, invalidate the pos cache, write the fetched\n /// remote L0 file to its local path via a temp-file + fsync + rename, then\n /// invalidate the cache again. The *detection* half (compare DB pos vs replica\n /// pos and fetch the bytes) lives on [`crate::replica::Replica`] because the\n /// synchronous `Db` has no `ReplicaClient` handle of its own. Used by\n /// [`crate::replica::Replica::check_database_behind_replica`] (issue #781).\n pub fn seed_l0_baseline(&mut self, min_txid: TXID, max_txid: TXID, data: &[u8]) -> Result<()> {\n // Clear local L0 files (db.go:1241-1249).\n let l0_dir = self.ltx_level_dir(0);\n match self.host.remove_dir_all(Path::new(&l0_dir)) {\n Ok(()) => {}\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}\n Err(e) => return Err(e.into()),\n }\n self.max_l0_file_info = None;\n self.invalidate_pos_cache();\n self.host.create_dir_all(Path::new(&l0_dir))?;\n\n // Write the baseline file atomically (db.go:1260-1286).\n let local_path = self.ltx_path(0, min_txid, max_txid);\n let tmp_path = format!(\"{local_path}.tmp\");\n let _ = write_file_atomic(&self.host, &tmp_path, &local_path, data)?;\n self.last_l0_header = None;\n self.invalidate_pos_cache();\n Ok(())\n }\n\n /// Continues a chain at `txid`, whose database has `commit` pages: the\n /// next sync captures a delta at `txid + 1`, not the whole database. A\n /// zero-page L0 at `txid` is the local baseline verify reads back; its\n /// WAL fields name the start of the WAL as it stands, so the first\n /// capture begins at the WAL header. The WAL must exist: an activation\n /// writes the control tables before this runs. Returns the baseline's\n /// bytes: the caller uploads the same object as the epoch's marker, so\n /// the epoch is never empty (CelldPersistencePaged.tla, `BreakNoMarker`).\n /// A whole-database opener is what a paged cell cannot afford: on the\n /// 2 GB fleet whale it took minutes through the fault path, the\n /// durability gate timed out, and every re-activation started it over.\n pub fn seed_continuation(&mut self, txid: TXID, commit: u32) -> Result> {\n let wal = self.wal_header_bytes()?;\n let header = ltx::Header {\n version: ltx::VERSION,\n flags: ltx::HEADER_FLAG_NO_CHECKSUM,\n page_size: self.page_size,\n commit,\n min_txid: txid,\n max_txid: txid,\n timestamp: self.host.now_unix_millis(),\n pre_apply_checksum: 0,\n wal_offset: WAL_HEADER_SIZE as i64,\n wal_size: 0,\n wal_salt1: be_u32(&wal[16..]),\n wal_salt2: be_u32(&wal[20..]),\n node_id: 0,\n };\n let baseline = ltx::encode_file(&header, &[], 0)?;\n self.seed_l0_baseline(txid, txid, &baseline)?;\n Ok(baseline)\n }\n\n // ── Capture loop ───────────────────────────────────────────────────────\n\n /// Where one `sync` call's wall time went, phase by phase. The phases\n /// sum to the call's wall time minus only dispatch overhead, so a\n /// capture ledger built on this closes its books. `checkpoint_us` is\n /// SQLite's own checkpoint (its writes and fsyncs run through the\n /// SQLite VFS, invisible to the LTX host filesystem), measured here\n /// because no other layer can see it.\n pub fn last_sync_timing(&self) -> SyncTiming {\n self.last_sync_timing\n }\n\n /// Copies pending data from the WAL into the next L0 LTX file and applies\n /// the checkpoint policy.\n ///\n /// Ported from `DB.Sync` (db.go:994-1056). The public entry point of the\n /// capture loop. This function is synchronous as described in the module docs.\n pub fn sync(&mut self) -> Result<()> {\n self.sync_with_verify_hook(None)\n }\n\n /// `sync` with an optional hook that runs between `verify` and the\n /// capture read — the window a concurrent WAL change would land in.\n /// Test-only through `internal::sync_with_verify_hook`; production\n /// passes `None`.\n fn sync_with_verify_hook(&mut self, hook: Option>) -> Result<()> {\n self.last_sync_timing = SyncTiming::default();\n let phase = crate::host::telemetry_us();\n // Self-heal: recreate the control tables if something swept them out\n // of `sqlite_schema` from under the replicator — without them every\n // capture fails until the database is reopened. A no-op when the\n // tables exist (no schema change, no WAL write).\n self.ensure_control_tables()?;\n let schema_done = crate::host::telemetry_us();\n self.last_sync_timing.schema_check_us = schema_done.saturating_sub(phase);\n\n // Ensure the WAL has at least one frame (db.go:1017-1020).\n self.ensure_wal_exists()?;\n self.last_sync_timing.wal_exists_us =\n crate::host::telemetry_us().saturating_sub(schema_done);\n self.last_sync_timing.prepare_us = crate::host::telemetry_us().saturating_sub(phase);\n\n let (orig_wal_size, new_wal_size, synced) = self.verify_and_sync(hook)?;\n\n // Track that data was synced for time-based checkpoint decisions.\n if synced {\n self.synced_since_checkpoint = true;\n }\n\n let phase = crate::host::telemetry_us();\n self.checkpoint_if_needed(orig_wal_size, new_wal_size)?;\n self.last_sync_timing.checkpoint_us = crate::host::telemetry_us().saturating_sub(phase);\n\n // Recompute the cached position (kept for parity with db.go:1037-1041).\n let _ = self.pos()?;\n\n Ok(())\n }\n\n /// Verifies the last sync against the current WAL, then syncs.\n ///\n /// Ported from `DB.verifyAndSync` (db.go:1058-1090). Returns\n /// `(orig_wal_size, new_wal_size, synced)` where the sizes are the **logical**\n /// WAL offset (`WALOffset+WALSize` of the last LTX), not file size (#997).\n fn verify_and_sync(&mut self, between: Option>) -> Result<(i64, i64, bool)> {\n // Use the last synced WAL offset as the logical size for checkpoint\n // decisions; on the first sync fall back to file size (db.go:1062-1069).\n let mut orig_wal_size = self.last_synced_wal_offset;\n if orig_wal_size == 0 {\n orig_wal_size = self.wal_file_size()?;\n }\n\n let phase = crate::host::telemetry_us();\n let info = self.verify()?;\n self.last_sync_timing.verify_us = crate::host::telemetry_us().saturating_sub(phase);\n if let Some(between) = between {\n between();\n }\n let phase = crate::host::telemetry_us();\n let synced = self.sync_inner(info)?;\n self.last_sync_timing.encode_write_us = crate::host::telemetry_us()\n .saturating_sub(phase)\n .saturating_sub(self.last_sync_timing.fsync_us);\n\n let new_wal_size = self.last_synced_wal_offset;\n Ok((orig_wal_size, new_wal_size, synced))\n }\n\n /// Ensures the LTX state matches where it left off from the real WAL.\n ///\n /// Ported branch-for-branch from `DB.verify` (db.go:1296-1436).\n /// This is the snapshot-on-continuity-break brain — do not refactor for\n /// elegance on pass one.\n fn verify(&mut self) -> Result {\n let frame_size = self.page_size as i64 + WAL_FRAME_HEADER_SIZE as i64;\n let mut info = SyncInfo {\n snapshotting: true,\n ..Default::default()\n };\n\n let pos = self.pos()?;\n if pos.txid == TXID(0) {\n info.offset = WAL_HEADER_SIZE as i64;\n return Ok(info); // first sync\n }\n\n // Determine the last WAL offset we saved from: the header of the\n // last L0 (db.go:1311-1326). This instance wrote that header one\n // sync ago, so the cache answers without re-reading the file; a\n // key miss (restore, compaction, reopen) reads and parses as the\n // port always did.\n let cache_hit = matches!(&self.last_l0_header, Some((txid, _)) if *txid == pos.txid);\n let (hdr, ltx_bytes) = if cache_hit {\n let (_, cached) = self.last_l0_header.as_ref().expect(\"matched above\");\n (cached.clone(), None)\n } else {\n let ltx_path = self.ltx_path(0, pos.txid, pos.txid);\n let ltx_bytes = match self.host.read(Path::new(<x_path)) {\n Ok(b) => b,\n Err(e) => {\n return Err(Error::Ltx(Box::new(new_ltx_error(\n \"open\",\n <x_path,\n 0,\n pos.txid.0,\n pos.txid.0,\n e.into(),\n ))));\n }\n };\n let parsed = match ltx::Header::parse(<x_bytes) {\n Ok(h) => h,\n Err(_) => {\n return Err(Error::Ltx(Box::new(new_ltx_error(\n \"decode\",\n <x_path,\n 0,\n pos.txid.0,\n pos.txid.0,\n Error::LTXCorrupted,\n ))));\n }\n };\n (\n LastL0Header {\n wal_offset: parsed.wal_offset,\n wal_size: parsed.wal_size,\n wal_salt1: parsed.wal_salt1,\n wal_salt2: parsed.wal_salt2,\n commit: parsed.commit,\n final_pgno: 0,\n final_page: Vec::new(),\n },\n Some(ltx_bytes),\n )\n };\n info.offset = hdr.wal_offset + hdr.wal_size;\n info.salt1 = hdr.wal_salt1;\n info.salt2 = hdr.wal_salt2;\n info.prev_commit = hdr.commit;\n\n // If the LTX WAL offset exceeds the real WAL size, the WAL was truncated.\n let wal_size = self.wal_file_size()?;\n if info.offset > wal_size {\n // If we previously synced to the exact WAL end, this truncation is an\n // expected checkpoint: reset to the header and continue incrementally\n // rather than snapshotting (issue #927, db.go:1335-1355).\n if self.synced_to_wal_end {\n self.synced_to_wal_end = false;\n\n let wal_hdr = self.wal_header_bytes()?;\n info.offset = WAL_HEADER_SIZE as i64;\n info.salt1 = be_u32(&wal_hdr[16..]);\n info.salt2 = be_u32(&wal_hdr[20..]);\n info.snapshotting = false;\n info.reason = String::new();\n return Ok(info);\n }\n\n info.reason = \"wal truncated by another process\".to_string();\n return Ok(info);\n }\n\n // Compare WAL headers; restart from the beginning of the WAL if different.\n let wal_hdr = self.wal_header_bytes()?;\n let salt1 = be_u32(&wal_hdr[16..]);\n let salt2 = be_u32(&wal_hdr[20..]);\n let salt_match = salt1 == hdr.wal_salt1 && salt2 == hdr.wal_salt2;\n\n // Edge case: LTX represents the start of the WAL (WALOffset=32, WALSize=0).\n // Handle this before computing prev_wal_offset to avoid underflow\n // (32 - 4120 = -4088). See issue #900 (db.go:1375-1383).\n if info.offset == WAL_HEADER_SIZE as i64 {\n if salt_match {\n info.snapshotting = false;\n return Ok(info);\n }\n info.reason = \"wal header salt reset, snapshotting\".to_string();\n return Ok(info);\n }\n\n // If the offset is at the start of the first page, we can't check the\n // previous page (db.go:1386-1399).\n let prev_wal_offset = info.offset - frame_size;\n if prev_wal_offset == WAL_HEADER_SIZE as i64 {\n if salt_match {\n info.snapshotting = false;\n return Ok(info);\n }\n info.reason = \"wal header salt reset, snapshotting\".to_string();\n return Ok(info);\n } else if prev_wal_offset < WAL_HEADER_SIZE as i64 {\n return Err(Error::Other(\n format!(\"prev WAL offset is less than the header size: {prev_wal_offset}\").into(),\n ));\n }\n\n // If we can't verify the last page is in the last LTX file, snapshot.\n let last_page_match = match <x_bytes {\n Some(bytes) => self.last_page_match(bytes, &hdr, prev_wal_offset, frame_size)?,\n None => self.last_page_match_cached(&hdr, prev_wal_offset, frame_size)?,\n };\n if !last_page_match {\n info.reason =\n \"last page does not exist in last ltx file, wal overwritten by another process\"\n .to_string();\n return Ok(info);\n }\n\n // Salt changed (possible FULL/RESTART checkpoint). With a last-page match\n // we assume the WAL was not overwritten (db.go:1412-1431).\n if !salt_match {\n info.offset = WAL_HEADER_SIZE as i64;\n info.salt1 = salt1;\n info.salt2 = salt2;\n\n let detected =\n self.detect_full_checkpoint(&[(salt1, salt2), (hdr.wal_salt1, hdr.wal_salt2)])?;\n if detected {\n info.reason = \"full or restart checkpoint detected, snapshotting\".to_string();\n } else {\n info.snapshotting = false;\n }\n return Ok(info);\n }\n\n info.snapshotting = false;\n Ok(info)\n }\n\n /// Checks whether the last page read in the WAL exists in the last LTX file.\n ///\n /// Ported from `DB.lastPageMatch` (db.go:1438-1475). Re-reads the last synced\n /// WAL frame and searches the last LTX file's pages for a matching\n /// `(pgno, data)` pair.\n fn last_page_match(\n &mut self,\n ltx_bytes: &[u8],\n hdr: &LastL0Header,\n prev_wal_offset: i64,\n frame_size: i64,\n ) -> Result {\n if prev_wal_offset <= WAL_HEADER_SIZE as i64 {\n return Ok(false);\n }\n\n let frame = self.wal_bytes_at(prev_wal_offset, frame_size)?;\n let pgno = be_u32(&frame[0..]);\n let fsalt1 = be_u32(&frame[8..]);\n let fsalt2 = be_u32(&frame[12..]);\n let data = &frame[WAL_FRAME_HEADER_SIZE..];\n\n if fsalt1 != hdr.wal_salt1 || fsalt2 != hdr.wal_salt2 {\n return Ok(false);\n }\n\n // Verify the last WAL page exists, byte-for-byte, in the last LTX file.\n // Decode the full file and compare the decompressed page bytes.\n let pages = ltx::decode_file_pages(ltx_bytes).map_err(|_| Error::LTXCorrupted)?;\n for (p, page_data) in &pages {\n if *p == pgno && page_data.as_slice() == data {\n return Ok(true);\n }\n }\n Ok(false)\n }\n\n /// The cache-backed page check: compares the WAL frame at\n /// `prev_wal_offset` against the final frame the previous sync\n /// consumed. Equivalent to the ported search when the WAL is\n /// unchanged (the L0's entry for that page IS that frame), and\n /// strictly more conservative when it is not — a mismatch forces a\n /// snapshot, never an incremental read over rewritten history.\n fn last_page_match_cached(\n &mut self,\n hdr: &LastL0Header,\n prev_wal_offset: i64,\n frame_size: i64,\n ) -> Result {\n if prev_wal_offset <= WAL_HEADER_SIZE as i64 || hdr.final_page.is_empty() {\n return Ok(false);\n }\n let frame = self.wal_bytes_at(prev_wal_offset, frame_size)?;\n let pgno = be_u32(&frame[0..]);\n let fsalt1 = be_u32(&frame[8..]);\n let fsalt2 = be_u32(&frame[12..]);\n let data = &frame[WAL_FRAME_HEADER_SIZE..];\n Ok(fsalt1 == hdr.wal_salt1\n && fsalt2 == hdr.wal_salt2\n && pgno == hdr.final_pgno\n && data == hdr.final_page.as_slice())\n }\n\n /// Detects whether a FULL or RESTART checkpoint occurred (we may have missed\n /// frames). Ported from `DB.detectFullCheckpoint` (db.go:1477-1507).\n fn detect_full_checkpoint(&self, known_salts: &[(u32, u32)]) -> Result {\n let wal_bytes = self.host.read(&self.wal_path())?;\n let rd = WalReader::new(&wal_bytes).map_err(Error::from)?;\n let last_known = known_salts.last().copied().unwrap_or((0, 0));\n let mut m = rd.frame_salts_until(last_known);\n for s in known_salts {\n m.remove(s);\n }\n Ok(!m.is_empty())\n }\n\n /// Read an incremental WAL image through the first invalid frame.\n ///\n /// SQLite can retain stale frames after a checkpoint restarts the logical\n /// WAL. The frame checksum chain is the format's end marker, so grow a\n /// prefix geometrically and stop when `WalReader` stops before the bytes\n /// that were read. The geometric window keeps parsing linear overall and\n /// reads at most one prior window past the valid end. The returned byte\n /// count includes that discarded probe for an honest I/O ledger.\n fn read_valid_wal_image(&mut self, info: &SyncInfo, start: usize) -> Result<(WalImage, usize)> {\n let frame_size = self.page_size as usize + WAL_FRAME_HEADER_SIZE;\n let offset = info.offset;\n let salt1 = info.salt1;\n let salt2 = info.salt2;\n Ok(self.with_wal_file(|file| {\n let file_len = file.file_len()? as usize;\n if file_len < WAL_HEADER_SIZE || start < WAL_HEADER_SIZE || start >= file_len {\n return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof));\n }\n let complete_end =\n WAL_HEADER_SIZE + ((file_len - WAL_HEADER_SIZE) / frame_size) * frame_size;\n if start >= complete_end || !(start - WAL_HEADER_SIZE).is_multiple_of(frame_size) {\n return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof));\n }\n\n let tail_base = if start == WAL_HEADER_SIZE { 0 } else { start };\n let mut bytes = file.read_exact_at(0, WAL_HEADER_SIZE)?;\n let mut read_bytes = bytes.len();\n let mut cursor = start;\n let mut target_frames = 1_usize;\n loop {\n let target_end = start\n .saturating_add(target_frames.saturating_mul(frame_size))\n .min(complete_end);\n if target_end > cursor {\n let chunk = file.read_exact_at(cursor as u64, target_end - cursor)?;\n read_bytes += chunk.len();\n bytes.extend_from_slice(&chunk);\n cursor = target_end;\n }\n\n let valid_end = {\n let parsed = if offset == WAL_HEADER_SIZE as i64 {\n WalReader::new(&bytes)\n } else if tail_base == 0 {\n WalReader::new_with_offset(&bytes, offset, salt1, salt2)\n } else {\n WalReader::new_with_offset_over_tail(\n &bytes,\n tail_base as i64,\n offset,\n salt1,\n salt2,\n )\n };\n let mut reader = parsed.map_err(|error| {\n std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string())\n })?;\n reader.page_map().map_err(|error| {\n std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string())\n })?;\n if reader.offset() == 0 {\n WAL_HEADER_SIZE\n } else {\n reader.offset() as usize + frame_size\n }\n };\n\n if valid_end < cursor {\n let keep = if tail_base == 0 {\n valid_end\n } else {\n WAL_HEADER_SIZE + valid_end.saturating_sub(tail_base)\n };\n bytes.truncate(keep);\n return Ok((\n WalImage {\n bytes,\n tail_base,\n file_len,\n },\n read_bytes,\n ));\n }\n if cursor == complete_end {\n return Ok((\n WalImage {\n bytes,\n tail_base,\n file_len,\n },\n read_bytes,\n ));\n }\n target_frames = target_frames.saturating_mul(2);\n }\n })?)\n }\n\n /// Copies pending bytes from the real WAL into a new L0 LTX file.\n ///\n /// Ported from `DB.sync` (db.go:1517-1723). Returns `true` if an LTX file was\n /// written (there were new pages or we were snapshotting). Atomic\n /// tmp→fsync→rename with the pos cache + anti-feedback flags updated after.\n fn sync_inner(&mut self, mut info: SyncInfo) -> Result {\n let phase = crate::host::telemetry_us();\n // A capture that starts at the WAL header reads a logical WAL with no\n // backfilled prefix: the first sync, a restart, or a boundary image.\n // The checkpoint trigger counts from the backfilled boundary, so it\n // must not carry an offset from the WAL that just ended, or the new\n // WAL would grow past that offset before its first checkpoint.\n if info.offset == WAL_HEADER_SIZE as i64 {\n self.checkpointed_wal_offset = WAL_HEADER_SIZE as i64;\n }\n let pos = self.pos()?;\n let tx_id = TXID(pos.txid.0 + 1);\n let filename = self.ltx_path(0, tx_id, tx_id);\n\n let db_size = self.db_file_size()?;\n let mut commit = (db_size / self.page_size as i64) as u32;\n self.last_db_pages = commit;\n self.last_sync_timing.pos_us = crate::host::telemetry_us().saturating_sub(phase);\n let phase = crate::host::telemetry_us();\n\n // The incremental path reads only the valid checksum chain: the\n // 32-byte WAL header, the previous frame when there is one, and the\n // frames from `info.offset` on. A passive or restart checkpoint can\n // leave a large physical suffix after SQLite restarts the logical WAL\n // at its beginning. Reading to the physical file length made every\n // Queue capture re-read that stale suffix because Queue cells cannot\n // use size-triggered TRUNCATE checkpoints. The progressive reader\n // doubles its valid-prefix window, then drops the bounded look-ahead\n // after the first salt or checksum failure. The sparse image keeps\n // every offset absolute, so the reader and page collectors remain\n // unchanged. An I/O race or a previous-frame mismatch falls back to\n // the full read the port always did.\n let frame_size_bytes = self.page_size as i64 + WAL_FRAME_HEADER_SIZE as i64;\n let mut wal = if info.snapshotting {\n self.last_sync_timing.wal_read_kind = 1;\n let bytes = self.host.read(&self.wal_path())?;\n self.last_sync_timing.wal_read_bytes = bytes.len() as u64;\n self.last_sync_timing.wal_len_bytes = bytes.len() as u64;\n WalImage::whole(bytes)\n } else {\n let start = if info.offset <= WAL_HEADER_SIZE as i64 + frame_size_bytes {\n WAL_HEADER_SIZE\n } else {\n (info.offset - frame_size_bytes) as usize\n };\n match self.read_valid_wal_image(&info, start) {\n Ok((image, read_bytes)) => {\n self.last_sync_timing.wal_read_kind =\n if start == WAL_HEADER_SIZE { 2 } else { 0 };\n self.last_sync_timing.wal_read_bytes = read_bytes as u64;\n self.last_sync_timing.wal_len_bytes = image.file_len as u64;\n image\n }\n Err(_) => {\n self.last_sync_timing.wal_read_kind = 3;\n let bytes = self.host.read(&self.wal_path())?;\n self.last_sync_timing.wal_read_bytes = bytes.len() as u64;\n self.last_sync_timing.wal_len_bytes = bytes.len() as u64;\n WalImage::whole(bytes)\n }\n }\n };\n\n self.last_sync_timing.wal_read_us = crate::host::telemetry_us().saturating_sub(phase);\n self.last_sync_timing.wal_image_bytes = wal.bytes.len() as u64;\n self.last_sync_timing.snapshot = info.snapshotting;\n self.last_sync_timing.snapshot_reason = if info.snapshotting {\n snapshot_reason_code(&info.reason)\n } else {\n 0\n };\n let phase = crate::host::telemetry_us();\n\n // Choose the WAL reader start: from the header, or seek to info.offset.\n // A previous-frame mismatch falls back to a full read (snapshot),\n // mirroring NewWALReaderWithOffset's PrevFrameMismatchError handling\n // (db.go:1565-1581).\n // A previous-frame mismatch restarts the read from the header. The\n // sparse tail image is zero-filled below `start`, so a from-header\n // reader over it would see no valid frame and report \"nothing to\n // capture\" — a silent miss the ship loop would then credit. The\n // mismatch path therefore re-reads the complete WAL first, which is\n // exactly the port's former full-read behavior on this branch.\n let mismatch = !(info.offset == WAL_HEADER_SIZE as i64)\n && matches!(\n wal.reader_at(info.offset, info.salt1, info.salt2),\n Err(crate::wal::WalError::PrevFrameMismatch)\n );\n if mismatch {\n info.offset = WAL_HEADER_SIZE as i64;\n if self.last_sync_timing.wal_read_kind == 0 {\n let bytes = self.host.read(&self.wal_path())?;\n self.last_sync_timing.wal_read_kind = 3;\n self.last_sync_timing.wal_read_bytes = bytes.len() as u64;\n self.last_sync_timing.wal_len_bytes = bytes.len() as u64;\n self.last_sync_timing.wal_image_bytes = bytes.len() as u64;\n wal = WalImage::whole(bytes);\n }\n }\n let mut rd = if info.offset == WAL_HEADER_SIZE as i64 {\n WalReader::new(&wal.bytes).map_err(Error::from)?\n } else {\n wal.reader_at(info.offset, info.salt1, info.salt2)\n .map_err(Error::from)?\n };\n\n let (page_map, max_offset, wal_commit) = rd.page_map().map_err(Error::from)?;\n if wal_commit > 0 {\n commit = wal_commit;\n }\n\n let sz = if max_offset > 0 {\n max_offset - info.offset\n } else {\n 0\n };\n if sz < 0 {\n return Err(Error::Other(\n format!(\n \"wal size must be positive: sz={sz}, maxOffset={max_offset}, info.offset={}\",\n info.offset\n )\n .into(),\n ));\n }\n\n // Exit if there are no new WAL pages and we are not snapshotting\n // (db.go:1603-1607).\n if !info.snapshotting && sz == 0 {\n return Ok(false);\n }\n\n let (rd_salt1, rd_salt2) = rd.salt();\n\n // Build the page set for the encoder.\n let pages: Vec<(u32, Vec)> = if info.snapshotting {\n self.collect_snapshot_pages(&wal, &page_map, commit)?\n } else {\n self.collect_wal_pages(&wal, &page_map, info.prev_commit, commit)?\n };\n self.last_sync_timing.map_collect_us = crate::host::telemetry_us().saturating_sub(phase);\n let phase = crate::host::telemetry_us();\n\n let header = ltx::Header {\n version: ltx::VERSION,\n flags: ltx::HEADER_FLAG_NO_CHECKSUM,\n page_size: self.page_size,\n commit,\n min_txid: tx_id,\n max_txid: tx_id,\n timestamp: self.host.now_unix_millis(),\n pre_apply_checksum: 0,\n wal_offset: info.offset,\n wal_size: sz,\n wal_salt1: rd_salt1,\n wal_salt2: rd_salt2,\n node_id: 0,\n };\n\n // Encode the LTX file (with HeaderFlagNoChecksum, so post-apply is 0).\n let encoded = ltx::encode_file(&header, &pages, 0)?;\n self.last_sync_timing.ltx_encode_us = crate::host::telemetry_us().saturating_sub(phase);\n let phase = crate::host::telemetry_us();\n\n // Atomic tmp → fsync → rename (db.go:1609-1685).\n let tmp_filename = format!(\"{filename}.tmp\");\n let parent = Path::new(&tmp_filename).parent().map(Path::to_path_buf);\n if !self.l0_dir_ready {\n if let Some(parent) = &parent {\n self.host.create_dir_all(parent)?;\n }\n self.l0_dir_ready = true;\n }\n // On rename failure, clear the L0 cache + invalidate pos\n // (db.go:1680-1684); the error path below does that. A directory\n // that vanished under a ready flag is recreated once and the cut\n // retried, so the flag saves a `mkdir` per sync without trusting it.\n self.last_sync_timing.fsync_us =\n match write_file_atomic(&self.host, &tmp_filename, &filename, &encoded) {\n Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {\n if let Some(parent) = &parent {\n self.host.create_dir_all(parent)?;\n }\n write_file_atomic(&self.host, &tmp_filename, &filename, &encoded)?\n }\n other => other?,\n };\n self.last_sync_timing.file_write_us = crate::host::telemetry_us()\n .saturating_sub(phase)\n .saturating_sub(self.last_sync_timing.fsync_us);\n // The next verify reads exactly these fields back; caching them —\n // plus the final consumed WAL frame for the page check — is what\n // spares it re-reading the file it just watched being written.\n let frame_size = self.page_size as i64 + WAL_FRAME_HEADER_SIZE as i64;\n let final_frame_offset = max_offset - frame_size;\n let (final_pgno, final_page) = match (final_frame_offset >= WAL_HEADER_SIZE as i64)\n .then(|| wal.slice(final_frame_offset, frame_size as usize))\n .flatten()\n {\n Some(frame) => (be_u32(&frame[0..]), frame[WAL_FRAME_HEADER_SIZE..].to_vec()),\n None => (0, Vec::new()),\n };\n self.last_l0_header = Some((\n tx_id,\n LastL0Header {\n wal_offset: info.offset,\n wal_size: sz,\n wal_salt1: rd_salt1,\n wal_salt2: rd_salt2,\n commit,\n final_pgno,\n final_page,\n },\n ));\n\n // Update the L0 file-info cache and the cached position (db.go:1687-1702).\n self.max_l0_file_info = Some(ltx::FileInfo {\n level: 0,\n min_txid: tx_id,\n max_txid: tx_id,\n pre_apply_checksum: 0,\n post_apply_checksum: 0,\n size: encoded.len() as i64,\n created_at: Some(system_time_from_unix_millis(self.host.now_unix_millis())),\n });\n // The encoder's post-apply pos: for a NoChecksum file the post-apply\n // checksum is 0; the position is (tx_id, 0).\n self.pos_cache = Some(Pos::new(tx_id, 0));\n\n // Track the logical end of WAL content for checkpoint decisions\n // (db.go:1704-1718, issues #997/#927).\n let final_offset = info.offset + sz;\n self.last_synced_wal_offset = final_offset;\n self.synced_to_wal_end = match self.wal_file_size() {\n Ok(wal_size) => final_offset == wal_size,\n Err(_) => false,\n };\n\n Ok(true)\n }\n\n /// Collects the page set for an incremental sync, in ascending page-number\n /// order. Pages absent from the WAL growth range come from the database file\n /// (Litestream #1292 / celld #150).\n fn collect_wal_pages(\n &self,\n wal: &WalImage,\n page_map: &HashMap,\n prev_commit: u32,\n commit: u32,\n ) -> Result)>> {\n let mut pgnos: Vec = page_map.keys().copied().collect();\n let lock = lock_pgno(self.page_size);\n if commit > prev_commit {\n for pgno in (prev_commit + 1)..=commit {\n if pgno != lock && !page_map.contains_key(&pgno) {\n pgnos.push(pgno);\n }\n }\n }\n pgnos.sort_unstable();\n\n let mut out = Vec::with_capacity(pgnos.len());\n for pgno in pgnos {\n let data = match page_map.get(&pgno) {\n Some(&offset) => wal.page(offset, self.page_size)?,\n None => self.read_db_page(pgno)?,\n };\n out.push((pgno, data));\n }\n Ok(out)\n }\n\n /// Collects the full page set for a snapshot: every page `1..=commit`\n /// (skipping the lock page), reading from the WAL where present, else the DB\n /// file. Ported from `DB.writeLTXFromDB` (db.go:1725-1770).\n fn collect_snapshot_pages(\n &self,\n wal: &WalImage,\n page_map: &HashMap,\n commit: u32,\n ) -> Result)>> {\n let lock = lock_pgno(self.page_size);\n let mut out = Vec::with_capacity(commit as usize);\n for pgno in (1..=commit).filter(|pgno| *pgno != lock) {\n let data = match page_map.get(&pgno) {\n Some(&offset) => wal.page(offset, self.page_size)?,\n None => self.read_db_page(pgno)?,\n };\n out.push((pgno, data));\n }\n Ok(out)\n }\n\n /// Reads one page of the main database through the connection's VFS file,\n /// never through a host read of the path. A paged cell's file is a sparse\n /// cache of its cut, and the host read returned every unfaulted page as\n /// zeros: each paged epoch's opener snapshot shipped holes, and the next\n /// epoch restored a database whose integrity_check named every overflow\n /// chain (fleet, 2026-09-02). The VFS faults the page in instead.\n fn read_db_page(&self, pgno: u32) -> Result> {\n let mut file: *mut ffi::sqlite3_file = std::ptr::null_mut();\n // SAFETY: `conn` is open, and FILE_POINTER hands out the main file,\n // which lives as long as the connection.\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(std::io::Error::other(\"no main database file\").into());\n }\n let n = self.page_size as usize;\n let mut page = vec![0u8; n];\n let offset = i64::from(pgno - 1) * i64::from(self.page_size);\n // SAFETY: `file` is the live main file and `page` holds `n` bytes.\n let rc = unsafe {\n let read = (*(*file).pMethods).xRead.expect(\"xRead\");\n read(file, page.as_mut_ptr().cast(), n as c_int, offset)\n };\n if rc != ffi::SQLITE_OK {\n return Err(std::io::Error::other(format!(\"read page {pgno}: sqlite rc {rc}\")).into());\n }\n Ok(page)\n }\n\n // ── Checkpointing ──────────────────────────────────────────────────────\n\n /// Performs a checkpoint based on the configured thresholds (3-tier policy).\n ///\n /// Ported from `DB.checkpointIfNeeded` (db.go:1092-1156). Checks\n /// in priority order: TruncatePageN (TRUNCATE, blocking) → MinCheckpointPageN\n /// (PASSIVE) → CheckpointInterval (PASSIVE, gated on `synced_since_checkpoint`).\n fn checkpoint_if_needed(&mut self, orig_wal_size: i64, new_wal_size: i64) -> Result<()> {\n if self.page_size == 0 {\n return Ok(());\n }\n\n // Priority 1: emergency TRUNCATE (blocking) on the *original* logical\n // size. A truncate ends in a boundary image of the whole database\n // (see `checkpoint`). For a small database that image is the cheap\n // price the threshold was tuned for, so below `RELATIVE_TRUNCATE_PAGES`\n // the threshold is absolute, as upstream's. Above it the WAL must also\n // have grown past the database before a truncate: the image then costs\n // at most what the WAL it replaces did, and the chain stays within 2x\n // of the writes. A fixed threshold made a 1MB-row whale pay a\n // database-sized capture every write, and its chain grew as the\n // square of its size.\n if self.truncate_page_n > 0 {\n let relative = if self.last_db_pages > RELATIVE_TRUNCATE_PAGES {\n self.last_db_pages\n } else {\n 0\n };\n let threshold = self.truncate_page_n.max(relative);\n if orig_wal_size >= calc_wal_size(self.page_size, threshold) {\n return self.checkpoint(CheckpointMode::Truncate);\n }\n }\n\n // Priority 2: PASSIVE once the frames appended since the last\n // backfill reach the threshold. See `checkpointed_wal_offset` for why\n // this is not the whole logical size: a checkpoint whose sealing\n // write could not restart the WAL must cost one retry at the next\n // threshold, not a checkpoint per sync.\n let backfilled_through = self.checkpointed_wal_offset.clamp(\n WAL_HEADER_SIZE as i64,\n new_wal_size.max(WAL_HEADER_SIZE as i64),\n );\n let threshold =\n calc_wal_size(self.page_size, self.min_checkpoint_page_n) - WAL_HEADER_SIZE as i64;\n if new_wal_size - backfilled_through >= threshold {\n return self.checkpoint_passive_swallowing_busy();\n }\n\n // Priority 3: time-based PASSIVE, gated on data synced since last\n // checkpoint (#896). Uses the DB-file mtime and a logical-size guard so an\n // idle DB does not spin LTX files (db.go:1133-1153).\n if self.checkpoint_interval > Duration::ZERO && self.synced_since_checkpoint {\n let elapsed = self.host.file_age(&self.path)?;\n if elapsed > self.checkpoint_interval && new_wal_size > calc_wal_size(self.page_size, 1)\n {\n return self.checkpoint_passive_swallowing_busy();\n }\n }\n\n Ok(())\n }\n\n /// PASSIVE checkpoint that swallows SQLITE_BUSY (log-and-continue, the one\n /// sanctioned best-effort case, db.go:1118-1124).\n fn checkpoint_passive_swallowing_busy(&mut self) -> Result<()> {\n match self.checkpoint(CheckpointMode::Passive) {\n Ok(()) => Ok(()),\n Err(e) if is_sqlite_busy_error(&e) => {\n self.last_sync_timing.checkpoint_runs = 1;\n self.last_sync_timing.checkpoint_busy_errors = 1;\n Ok(())\n }\n Err(e) => Err(e),\n }\n }\n\n /// Performs a checkpoint on the WAL.\n ///\n /// Ported from `DB.Checkpoint`/`DB.checkpoint` (db.go:1801-1873).\n /// A passive checkpoint seals the WAL under a short writer barrier before\n /// it runs. If any checkpoint restarts the WAL, a post-checkpoint write lock\n /// protects either an incremental sync or a full boundary snapshot.\n ///\n /// The upstream `chkMu.TryLock` snapshot-vs-checkpoint gate is unnecessary in\n /// the synchronous `Db`. `&mut self` serializes `sync`, `checkpoint`, and\n /// `snapshot`, and one blocking thread owns the `Db`.\n pub fn checkpoint(&mut self, mode: CheckpointMode) -> Result<()> {\n self.checkpoint_with_passive_hooks(mode, None, None, None)\n }\n\n fn checkpoint_with_passive_hooks(\n &mut self,\n mode: CheckpointMode,\n passive_hook: Option>,\n passive_unlocked_hook: Option>,\n post_barrier_hook: Option>,\n ) -> Result<()> {\n // Self-heal, as in `sync`: `checkpoint` writes to both control tables\n // and re-acquires the read lock through `_litestream_seq`, and the\n // invariant is that the tables exist before any control-table\n // statement — not only before a capture.\n self.ensure_control_tables()?;\n\n // Read the WAL header before the checkpoint to detect a restart.\n let hdr = self.wal_header_bytes()?;\n\n // Copy the end of the WAL before the checkpoint to capture as much as\n // possible (db.go:1823-1826).\n self.verify_and_sync(None)?;\n\n let frame_size = self.page_size as i64 + WAL_FRAME_HEADER_SIZE as i64;\n let pre_checkpoint_frame_n = if self.last_synced_wal_offset > WAL_HEADER_SIZE as i64 {\n (self.last_synced_wal_offset - WAL_HEADER_SIZE as i64) / frame_size\n } else {\n 0\n };\n\n // A passive checkpoint does not acquire SQLite's writer lock. Hold a\n // short write transaction on the dedicated read-lock connection, then\n // sync again to seal every commit before running the checkpoint on the\n // main connection. Keep the barrier until the checkpoint completes.\n let pragma = if mode == CheckpointMode::Passive {\n self.exec_passive_checkpoint_with_barrier(hdr, passive_hook, passive_unlocked_hook)?\n } else {\n self.exec_checkpoint(mode)?\n };\n self.last_sync_timing.checkpoint_runs = 1;\n self.last_sync_timing.checkpoint_wal_frames = pragma.wal_frames.max(0) as u64;\n self.last_sync_timing.checkpoint_backfilled = pragma.backfilled.max(0) as u64;\n self.last_sync_timing.checkpoint_busy = u64::from(pragma.busy != 0);\n // The backfilled boundary in this WAL's coordinates. A short backfill\n // (a reader pinned the WAL) leaves the remainder counting toward the\n // next threshold, so a pinned WAL retries at the threshold and an\n // unpinned one does not retry at all.\n self.checkpointed_wal_offset =\n WAL_HEADER_SIZE as i64 + pragma.backfilled.max(0) * frame_size;\n if let Some(hook) = post_barrier_hook {\n hook();\n }\n\n // Force a write so a restarted WAL has a new header and at least one\n // frame that verify can read.\n self.conn\n .execute_batch(\n \"INSERT INTO _litestream_seq (id, seq) VALUES (1, 1) \\\n ON CONFLICT (id) DO UPDATE SET seq = seq + 1\",\n )\n .map_err(sql_err)?;\n\n // If the WAL header is unchanged, the WAL did not restart — done.\n let other = self.wal_header_bytes()?;\n if hdr == other {\n self.synced_since_checkpoint = false;\n return Ok(());\n }\n self.last_sync_timing.checkpoint_restarts = 1;\n\n // The WAL restarted. Grab the write lock, then either copy the new WAL\n // tail or take a complete boundary image. TRUNCATE always needs the\n // boundary image because SQLite reports zero frames after resetting the\n // WAL. A forced checkpoint also needs one if it covered more frames than\n // the sealed pre-checkpoint sync observed.\n self.conn\n .prepare_cached(\"BEGIN\")\n .and_then(|mut statement| statement.execute([]))\n .map_err(sql_err)?;\n let post = (|| -> Result<()> {\n self.conn\n .prepare_cached(\"INSERT INTO _litestream_lock (id) VALUES (1)\")\n .and_then(|mut statement| statement.execute([]))\n .map_err(sql_err)?;\n if mode == CheckpointMode::Truncate\n || (mode != CheckpointMode::Passive && pragma.wal_frames > pre_checkpoint_frame_n)\n {\n let info = SyncInfo {\n offset: WAL_HEADER_SIZE as i64,\n salt1: be_u32(&other[16..]),\n salt2: be_u32(&other[20..]),\n snapshotting: true,\n reason: \"checkpoint boundary snapshot\".to_string(),\n ..Default::default()\n };\n self.sync_inner(info)?;\n } else {\n self.verify_and_sync(None)?;\n }\n Ok(())\n })();\n // Always roll back the write transaction (db.go:1849,1867).\n let rb = rollback(&self.conn);\n post?;\n rb?;\n\n self.synced_since_checkpoint = false;\n Ok(())\n }\n\n /// Seals and runs a passive checkpoint while holding SQLite's writer lock.\n ///\n /// The long-lived read transaction uses `rtx_conn`, so this method releases\n /// that read lock and temporarily reuses the same connection for the writer\n /// barrier. The checkpoint itself runs through `conn`, as SQLite does not\n /// permit a checkpoint on the connection that owns the write transaction.\n fn exec_passive_checkpoint_with_barrier(\n &mut self,\n pre_checkpoint_header: [u8; WAL_HEADER_SIZE],\n hook: Option>,\n unlocked_hook: Option>,\n ) -> Result {\n self.release_read_lock()?;\n if let Some(hook) = unlocked_hook {\n hook();\n }\n\n let result = (|| -> Result {\n self.rtx_conn\n .prepare_cached(\"BEGIN\")\n .and_then(|mut statement| statement.execute([]))\n .map_err(sql_err)?;\n self.rtx_conn\n .prepare_cached(\"INSERT INTO _litestream_lock (id) VALUES (1)\")\n .and_then(|mut statement| statement.execute([]))\n .map_err(sql_err)?;\n\n // Writers can cross their own autocheckpoint threshold after the\n // read lock is released but before this barrier wins SQLite's\n // writer lock. A changed WAL header means some of those commits\n // can already live only in the database file. The normal\n // synced-to-end shortcut cannot distinguish that race from our own\n // completed checkpoint, so an incremental LTX can omit the\n // checkpointed pages and produce a malformed restore. Seal the\n // complete boundary while the writer lock makes the database file\n // and the new WAL tail a stable pair. The cost is one database-size\n // LTX file only when another checkpoint wins this narrow gap.\n let barrier_header = self.wal_header_bytes()?;\n if barrier_header != pre_checkpoint_header {\n let info = SyncInfo {\n offset: WAL_HEADER_SIZE as i64,\n salt1: be_u32(&barrier_header[16..]),\n salt2: be_u32(&barrier_header[20..]),\n snapshotting: true,\n reason: \"WAL restarted before passive checkpoint barrier\".to_string(),\n ..Default::default()\n };\n self.sync_inner(info)?;\n } else {\n // Commits can land between the earlier sync and acquisition of\n // the barrier. This second sync seals them before the\n // checkpoint.\n self.verify_and_sync(None)?;\n }\n if let Some(hook) = hook {\n hook();\n }\n self.run_checkpoint_pragma(CheckpointMode::Passive)\n })();\n\n // Release the writer barrier before restoring the long-lived read lock.\n // Preserve the operation error if both the operation and cleanup fail.\n let rollback_result = rollback(&self.rtx_conn);\n let reacquire_result = self.acquire_read_lock();\n match result {\n Err(error) => Err(error),\n Ok(pragma) => {\n rollback_result?;\n reacquire_result?;\n Ok(pragma)\n }\n }\n }\n\n /// Releases the read lock, runs `PRAGMA wal_checkpoint()`, and\n /// re-acquires the read lock — re-acquiring even on error.\n ///\n /// Ported from `DB.execCheckpoint` (db.go:1875-1919). The exact\n /// release→checkpoint→re-acquire sequence is load-bearing.\n fn exec_checkpoint(&mut self, mode: CheckpointMode) -> Result {\n // Ensure the read lock is removed before the checkpoint; defer the\n // re-acquire so it runs even on early return.\n self.release_read_lock()?;\n\n let result = self.run_checkpoint_pragma(mode);\n\n // Re-acquire the read lock immediately after the checkpoint (the deferred\n // re-acquire in Go). If the pragma succeeded, propagate any re-acquire\n // error; otherwise surface the original pragma error.\n let reacquire = self.acquire_read_lock();\n match (result, reacquire) {\n (Ok(pragma), Ok(())) => Ok(pragma),\n (Ok(_), Err(e)) => Err(e),\n (Err(e), _) => Err(e),\n }\n }\n\n /// Runs the raw `PRAGMA wal_checkpoint()` and reads its 3-int result.\n fn run_checkpoint_pragma(&self, mode: CheckpointMode) -> Result {\n let sql = format!(\"PRAGMA wal_checkpoint({mode})\");\n self.conn\n .prepare_cached(&sql)\n .map_err(sql_err)?\n .query_row([], |row| {\n Ok(CheckpointPragma {\n busy: row.get::<_, i64>(0)?,\n wal_frames: row.get::<_, i64>(1)?,\n backfilled: row.get::<_, i64>(2)?,\n })\n })\n .map_err(sql_err)\n }\n\n // ── CRC64 ──────────────────────────────────────────────────────────────\n\n /// Returns a CRC-64/ISO checksum of the database file and its current\n /// position, after forcing a RESTART checkpoint so the DB sits at the WAL\n /// start. Ported from `DB.CRC64` (db.go:2329-2359).\n pub fn crc64(&mut self) -> Result<(u64, Pos)> {\n // Force a RESTART checkpoint to ensure the DB is at the start of the WAL.\n self.checkpoint(CheckpointMode::Restart)?;\n\n let pos = self.pos()?;\n\n // Checksum the whole database (CRC64-ISO), page by page through the\n // VFS like every other database read.\n let pages = self.db_file_size()? / i64::from(self.page_size);\n let mut h = Crc64::new();\n for pgno in 1..=u32::try_from(pages).expect(\"page count\") {\n h.update(&self.read_db_page(pgno)?);\n }\n Ok((h.sum64(), pos))\n }\n\n // ── Snapshot ───────────────────────────────────────────────────────────\n\n /// Writes a full database snapshot as an LTX file to `w` and returns the\n /// snapshot position. Ported from `DB.SnapshotReader` (db.go:1922-2021),\n /// buffered rather than streamed, as described in `client/mod.rs`.\n ///\n /// The snapshot spans `MinTXID=1 .. MaxTXID=pos.TXID` (db.go:1996-1997). Its\n /// page set is the full DB (lock page skipped), and — being a snapshot — the\n /// rolling post-apply checksum **is** tracked.\n pub fn snapshot_to_writer(&mut self, w: &mut W) -> Result {\n if self.page_size == 0 {\n return Err(Error::Other(\n \"db not ready: page size not initialized\".into(),\n ));\n }\n\n let pos = self.pos()?;\n\n let db_size = self.db_file_size()?;\n let mut commit = (db_size / self.page_size as i64) as u32;\n\n let wal = WalImage::whole(self.host.read(&self.wal_path())?);\n let mut rd = WalReader::new(&wal.bytes).map_err(Error::from)?;\n let (page_map, max_offset, wal_commit) = rd.page_map().map_err(Error::from)?;\n if wal_commit > 0 {\n commit = wal_commit;\n }\n let wal_offset = rd.offset();\n let sz = if max_offset > 0 {\n max_offset - wal_offset\n } else {\n 0\n };\n let (salt1, salt2) = rd.salt();\n\n let pages = self.collect_snapshot_pages(&wal, &page_map, commit)?;\n\n // A snapshot tracks the rolling post-apply checksum (MinTXID==1, no\n // NoChecksum flag) — compute it the way decode_file verifies it.\n let lock = lock_pgno(self.page_size);\n let mut rolling: crate::Checksum = crate::CHECKSUM_FLAG;\n for (p, d) in &pages {\n if *p != lock {\n rolling = crate::CHECKSUM_FLAG | (rolling ^ ltx::checksum_page(*p, d));\n }\n }\n\n let header = ltx::Header {\n version: ltx::VERSION,\n flags: 0,\n page_size: self.page_size,\n commit,\n min_txid: TXID(1),\n max_txid: pos.txid,\n timestamp: self.host.now_unix_millis(),\n pre_apply_checksum: 0,\n wal_offset,\n wal_size: sz,\n wal_salt1: salt1,\n wal_salt2: salt2,\n node_id: 0,\n };\n\n let encoded = ltx::encode_file(&header, &pages, rolling)?;\n w.write_all(&encoded)?;\n\n Ok(Pos::new(pos.txid, rolling))\n }\n\n /// Closes the database, releasing the read lock first so other processes can\n /// checkpoint. Ported from the read-lock-release + connection-close portion of\n /// `DB.Close` (db.go:623-647).\n pub fn close(mut self) -> Result<()> {\n self.release_read_lock()?;\n // The connection is dropped here, closing it.\n Ok(())\n }\n}\n\n// ── free functions ─────────────────────────────────────────────────────────\n\n/// Returns the size of the WAL for a given page size & count, in i64 math to\n/// avoid u32 overflow with large page sizes. Ported from `calcWALSize`\n/// (db.go:1193-1197).\nfn calc_wal_size(page_size: u32, page_n: u32) -> i64 {\n WAL_HEADER_SIZE as i64 + (WAL_FRAME_HEADER_SIZE as i64 + page_size as i64) * page_n as i64\n}\n\n/// `true` if the error indicates an SQLITE_BUSY condition. Ported from\n/// `isSQLiteBusyError` (db.go:1158-1167). Matches both the rusqlite busy code and\n/// the Go substrings.\nfn is_sqlite_busy_error(err: &Error) -> bool {\n // Prefer matching the rusqlite error code when present.\n if let Error::Other(b) = err {\n if let Some(re) = b.downcast_ref::() {\n if let Some(code) = re.sqlite_error_code() {\n if code == rusqlite::ErrorCode::DatabaseBusy\n || code == rusqlite::ErrorCode::DatabaseLocked\n {\n return true;\n }\n }\n }\n }\n let s = err.to_string();\n s.contains(\"database is locked\") || s.contains(\"SQLITE_BUSY\")\n}\n\n/// `true` if the error indicates disk space issues (ENOSPC/EDQUOT). Ported from\n/// `isDiskFullError` (db.go:1169-1180) — case-insensitive substring\n/// match for parity with the Go table. This classifier matches the background\n/// monitor's disk-full temporary-file cleanup behavior (db.go:2304-2309).\n#[cfg_attr(not(test), allow(dead_code))]\nfn is_disk_full_error(err_msg: &str) -> bool {\n let s = err_msg.to_lowercase();\n s.contains(\"no space left on device\")\n || s.contains(\"disk quota exceeded\")\n || s.contains(\"enospc\")\n || s.contains(\"edquot\")\n}\n\n/// Rolls back the connection's current transaction, swallowing the\n/// \"already rolled back\" / \"no transaction is active\" errors (issue #934).\n/// Ported from `rollback` (litestream.go:130-135), adapted to rusqlite's message.\nfn rollback(conn: &Connection) -> Result<()> {\n match conn\n .prepare_cached(\"ROLLBACK\")\n .and_then(|mut statement| statement.execute([]).map(|_| ()))\n {\n Ok(()) => Ok(()),\n Err(e) => {\n let msg = e.to_string();\n // Go swallows \"transaction has already been committed or rolled back\".\n // rusqlite/SQLite reports \"cannot rollback - no transaction is active\".\n if msg.contains(\"transaction has already been committed or rolled back\")\n || msg.contains(\"no transaction is active\")\n || msg.contains(\"cannot rollback\")\n {\n Ok(())\n } else {\n Err(sql_err(e))\n }\n }\n }\n}\n\n/// Reads the 32-byte WAL header. Ported from `readWALHeader`\n/// (litestream.go:138-148): returns the header bytes; errors if the file is\n/// missing or shorter than 32 bytes.\nfn read_wal_header(host: &crate::LtxHost, path: &Path) -> Result<[u8; WAL_HEADER_SIZE]> {\n let mut file = host.open(path)?;\n let bytes = file.read_exact_at(0, WAL_HEADER_SIZE)?;\n Ok(bytes\n .try_into()\n .expect(\"the exact read has the header size\"))\n}\n\n/// Reads one page's worth of bytes from an in-memory WAL buffer at the frame\n/// `offset` (the offset is the start of the *frame header*; the page data follows\n/// the 24-byte frame header). Mirrors `walFile.ReadAt(data, offset+WALFrameHeaderSize)`\n/// (db.go:1745,1787).\n/// What one sync read of the WAL: the whole file (`tail_base` 0), or the\n/// 32-byte header followed by the file's bytes from `tail_base` onward.\n/// Offsets stay absolute at every consumer; the image maps them.\nstruct WalImage {\n bytes: Vec,\n tail_base: usize,\n file_len: usize,\n}\n\nimpl WalImage {\n fn whole(bytes: Vec) -> Self {\n let file_len = bytes.len();\n Self {\n bytes,\n tail_base: 0,\n file_len,\n }\n }\n\n /// A reader positioned at `offset`, over whichever shape this image has.\n fn reader_at(\n &self,\n offset: i64,\n salt1: u32,\n salt2: u32,\n ) -> std::result::Result, crate::wal::WalError> {\n if self.tail_base == 0 {\n WalReader::new_with_offset(&self.bytes, offset, salt1, salt2)\n } else {\n WalReader::new_with_offset_over_tail(\n &self.bytes,\n self.tail_base as i64,\n offset,\n salt1,\n salt2,\n )\n }\n }\n\n /// `n` bytes at absolute file `offset`, or `None` when the image does not\n /// hold them (a short read, or the gap between the header and the tail).\n fn slice(&self, offset: i64, n: usize) -> Option<&[u8]> {\n if offset < 0 {\n return None;\n }\n let offset = offset as usize;\n let start = if self.tail_base == 0 || offset < WAL_HEADER_SIZE {\n offset\n } else {\n WAL_HEADER_SIZE + offset.checked_sub(self.tail_base)?\n };\n let end = start.checked_add(n)?;\n (end <= self.bytes.len()).then(|| &self.bytes[start..end])\n }\n\n /// One page's worth of bytes at frame `offset` (the offset is the start of\n /// the *frame header*; the page data follows it).\n fn page(&self, offset: i64, page_size: u32) -> Result> {\n self.slice(offset + WAL_FRAME_HEADER_SIZE as i64, page_size as usize)\n .map(<[u8]>::to_vec)\n .ok_or_else(|| {\n Error::Io(std::io::Error::new(\n std::io::ErrorKind::UnexpectedEof,\n format!(\"short read wal page @ {offset}\"),\n ))\n })\n }\n}\n\n/// Writes `data` to `tmp_path`, fsyncs, then renames to `final_path` — the\n/// crash-consistent atomic-write idiom. On any failure the temp\n/// file is removed.\nfn write_file_atomic(\n host: &crate::LtxHost,\n tmp_path: &str,\n final_path: &str,\n data: &[u8],\n) -> Result {\n let result = (|| -> Result {\n let mut file = host.create(Path::new(tmp_path))?;\n file.write_all(data)?;\n let fsync = crate::host::telemetry_us();\n file.sync_all()?;\n let fsync_us = crate::host::telemetry_us().saturating_sub(fsync);\n drop(file);\n host.rename(Path::new(tmp_path), Path::new(final_path))?;\n Ok(fsync_us)\n })();\n if result.is_err() {\n let _ = host.remove_file(Path::new(tmp_path));\n }\n result\n}\n\n/// Recursively removes `.tmp` files under `root`. Ported from `removeTmpFiles`\n/// (litestream.go:169-182): missing root / errored entries are skipped.\nfn remove_tmp_files(host: &crate::LtxHost, root: &Path) -> Result<()> {\n fn walk(host: &crate::LtxHost, dir: &Path) {\n let entries = match host.read_dir(dir) {\n Ok(e) => e,\n Err(_) => return,\n };\n for ent in entries {\n let path = ent.path;\n if ent.is_dir {\n walk(host, &path);\n } else if path.extension().and_then(|e| e.to_str()) == Some(\"tmp\") {\n let _ = host.remove_file(&path);\n }\n }\n }\n walk(host, root);\n Ok(())\n}\n\nfn system_time_from_unix_millis(milliseconds: i64) -> SystemTime {\n if milliseconds >= 0 {\n UNIX_EPOCH + Duration::from_millis(milliseconds as u64)\n } else {\n UNIX_EPOCH - Duration::from_millis(milliseconds.unsigned_abs())\n }\n}\n\n/// Big-endian `u32` from the first four bytes of `b`.\n#[inline]\nfn be_u32(b: &[u8]) -> u32 {\n u32::from_be_bytes([b[0], b[1], b[2], b[3]])\n}\n\n#[doc(hidden)]\npub mod internal {\n /// Returns active lookaside slots while SQLite compiles a statement on\n /// each connection owned by the managed database.\n pub fn lookaside_used_while_preparing(db: &super::Db) -> super::Result<(i32, i32)> {\n fn current(conn: &rusqlite::Connection) -> super::Result {\n let mut statement = std::ptr::null_mut();\n // SAFETY: the connection is live, SQLite owns the prepared\n // statement until the matching finalize, and the SQL includes its\n // trailing NUL.\n let result = unsafe {\n rusqlite::ffi::sqlite3_prepare_v2(\n conn.handle(),\n c\"SELECT 1\".as_ptr(),\n -1,\n &mut statement,\n std::ptr::null_mut(),\n )\n };\n if result != rusqlite::ffi::SQLITE_OK {\n return Err(super::Error::Other(\n \"failed to prepare lookaside probe\".into(),\n ));\n }\n\n let mut active = 0;\n let mut highwater = 0;\n // SAFETY: the connection and output pointers remain live for this\n // call. The status operation does not own either output value.\n let result = unsafe {\n rusqlite::ffi::sqlite3_db_status(\n conn.handle(),\n rusqlite::ffi::SQLITE_DBSTATUS_LOOKASIDE_USED,\n &mut active,\n &mut highwater,\n 0,\n )\n };\n // SAFETY: prepare returned this statement exclusively to this\n // helper, including when the status operation fails.\n unsafe { rusqlite::ffi::sqlite3_finalize(statement) };\n if result != rusqlite::ffi::SQLITE_OK {\n return Err(super::Error::Other(\n \"failed to read SQLite lookaside status\".into(),\n ));\n }\n Ok(active)\n }\n\n Ok((current(&db.conn)?, current(&db.rtx_conn)?))\n }\n\n /// Counts SQL compilations on the capture's connections: SQLite runs the\n /// authorizer once per compilation, so the counter moves for a fresh\n /// `prepare` and for a re-prepare, and stays still for a cached statement.\n /// (Installing the authorizer expires the statements compiled so far, so\n /// the first sync after this call compiles once more.)\n pub fn count_sql_compilations(db: &super::Db) -> std::sync::Arc {\n use rusqlite::hooks::{AuthContext, Authorization};\n let counter = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));\n for conn in [&db.conn, &db.rtx_conn] {\n let counter = counter.clone();\n conn.authorizer(Some(move |_: AuthContext<'_>| {\n counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);\n Authorization::Allow\n }));\n }\n counter\n }\n\n use super::*;\n\n pub struct VerifyInfo {\n inner: SyncInfo,\n pub offset: i64,\n pub prev_commit: u32,\n pub snapshotting: bool,\n pub reason: String,\n }\n\n pub fn calc_wal_size(page_size: u32, page_n: u32) -> i64 {\n super::calc_wal_size(page_size, page_n)\n }\n\n pub fn is_sqlite_busy_error(error: &Error) -> bool {\n super::is_sqlite_busy_error(error)\n }\n\n pub fn is_disk_full_error(message: &str) -> bool {\n super::is_disk_full_error(message)\n }\n\n pub fn read_lock_held(db: &Db) -> bool {\n db.read_lock_held\n }\n\n /// Run one `sync` with `hook` invoked between `verify` and the capture\n /// read — the only window in which a WAL change can reach the\n /// previous-frame check without `verify` seeing it first.\n pub fn sync_with_verify_hook(db: &mut Db, hook: impl FnOnce() + 'static) -> Result<()> {\n db.sync_with_verify_hook(Some(Box::new(hook)))\n }\n\n pub fn wal_journal_mode(db: &Db) -> Result {\n db.conn\n .query_row(\"PRAGMA journal_mode\", [], |row| row.get(0))\n .map_err(sql_err)\n }\n\n pub fn wal_autocheckpoint(db: &Db) -> Result {\n db.conn\n .query_row(\"PRAGMA wal_autocheckpoint\", [], |row| row.get(0))\n .map_err(sql_err)\n }\n\n /// The effective `PRAGMA cache_size` of the writer connection.\n pub fn writer_cache_size(db: &Db) -> Result {\n db.conn\n .query_row(\"PRAGMA cache_size\", [], |row| row.get(0))\n .map_err(sql_err)\n }\n\n pub fn read_lock_connection_cache_size(db: &Db) -> Result {\n db.rtx_conn\n .query_row(\"PRAGMA cache_size\", [], |row| row.get(0))\n .map_err(sql_err)\n }\n\n pub fn rollback_read_lock_connection(db: &Db) -> Result<()> {\n db.rtx_conn.execute_batch(\"ROLLBACK\").map_err(sql_err)\n }\n\n pub fn release_read_lock(db: &mut Db) -> Result<()> {\n db.release_read_lock()\n }\n\n pub fn acquire_read_lock(db: &mut Db) -> Result<()> {\n db.acquire_read_lock()\n }\n\n pub fn verify(db: &mut Db) -> Result {\n let inner = db.verify()?;\n Ok(VerifyInfo {\n offset: inner.offset,\n prev_commit: inner.prev_commit,\n snapshotting: inner.snapshotting,\n reason: inner.reason.clone(),\n inner,\n })\n }\n\n pub fn sync_inner(db: &mut Db, info: VerifyInfo) -> Result {\n db.sync_inner(info.inner)\n }\n\n pub fn read_wal_header(db: &Db) -> Result<[u8; WAL_HEADER_SIZE]> {\n super::read_wal_header(&db.host, &db.wal_path())\n }\n\n pub fn invalidate_pos_cache(db: &mut Db) {\n db.invalidate_pos_cache();\n }\n\n pub fn ltx_level_dir(db: &Db, level: u32) -> String {\n db.ltx_level_dir(level)\n }\n\n pub fn checkpoint_passive_with_barrier_hook(\n db: &mut Db,\n hook: Box,\n ) -> Result<()> {\n db.checkpoint_with_passive_hooks(CheckpointMode::Passive, Some(hook), None, None)\n }\n\n pub fn checkpoint_passive_with_unlocked_hook(\n db: &mut Db,\n hook: Box,\n ) -> Result<()> {\n db.checkpoint_with_passive_hooks(CheckpointMode::Passive, None, Some(hook), None)\n }\n\n /// Run a passive checkpoint with `hook` invoked after the writer barrier\n /// dropped and before the sealing write: the window in which another\n /// writer can take the lock and restart the WAL itself.\n pub fn checkpoint_passive_with_post_barrier_hook(\n db: &mut Db,\n hook: Box,\n ) -> Result<()> {\n db.checkpoint_with_passive_hooks(CheckpointMode::Passive, None, None, Some(hook))\n }\n\n pub fn be_u32(bytes: &[u8]) -> u32 {\n super::be_u32(bytes)\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",1794436122703,"sidebar","expanded"]