[{"_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",{"_40":41,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_33":-7,"_34":35,"_36":37,"_38":39,"_40":41,"_42":31,"_43":31,"_44":31,"_45":31,"_46":31,"_47":31,"_48":49,"_50":49,"_51":31,"_52":-7,"_53":-7,"_54":55,"_56":57,"_58":59,"_60":-5,"_61":62,"_63":64,"_65":66,"_67":68},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":14,"_22":23,"_24":-5,"_25":26,"_27":28,"_28":29,"_30":31,"_32":14},"ref","90b43017241f81189453d326d05948f388b34652","path","crates/ltx/src/replica.rs","oid","f53fc2622ae39da6815a4299f7c678a506444e4a","size",40897,"commit","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","//! replica.rs — single-replica sync loop + restore orchestration.\n//! Ported from litestream@v0.5.11 `replica.go`.\n//!\n//! A [`Replica`] connects a managed [`crate::db::Db`] to a replication\n//! destination via a [`ReplicaClient`]. It drives two halves of replication:\n//!\n//! * **sync** (`replica.go:132-180`): copy each new L0 LTX file the capture loop\n//! wrote locally up to the remote replica, advancing the replicated position\n//! one TXID at a time. `calc_pos` (`replica.go:208-214`) recovers the starting\n//! position from the newest file already on the replica.\n//! * **restore** (`replica.go:533-725`): download the LTX files from the replica\n//! in TXID order, merge them (the compactor in `ltx.NewCompactor`,\n//! ltx@v0.5.1 compactor.go), and reconstruct the SQLite database file the way\n//! `Decoder.DecodeDatabaseTo` (ltx@v0.5.1 decoder.go:223-268) does — pages\n//! `1..=commit`, lock page zero-filled, written verbatim — to a temp file that\n//! is fsync'd and atomically renamed into place.\n//!\n//! ## Scope: L0-only, single replica\n//! The real `litestream v0.5.11` L0-only architecture stores **everything at\n//! level 0** — the snapshot (MinTXID==1) and every incremental — under\n//! `ltx/0/`. The snapshot level (`SnapshotLevel = 9`,\n//! compaction_level.go:9) is empty without compaction. [`calc_restore_plan`] is\n//! ported faithfully (snapshot anchor at `SnapshotLevel` + per-level cursors so\n//! adding compaction later \"just works\"), but in this scope the plan is the\n//! contiguous L0 chain `1..=N`.\n//!\n//! ## Deferred work that needs a background runtime or additional scope\n//! * **Follow mode** (`replica.go:730-987`, `applyLTXFile`/`fillFollowGap`): the\n//! continuous tail-restore loop. The current API performs one restore.\n//! * **The background monitor goroutine + backoff** (`replica.go:326-441`) and\n//! `Start`/`Stop`: need a Tokio task owning the `Db`; the\n//! synchronous `sync()` primitive it would call is implemented here.\n//! * **V3 (v0.3.x generation) restore** (`RestoreV3`, replica.go:990-1096) is not\n//! included because celld has no earlier replica generation to support.\n//! * **Timestamp / `-txid` targeted restore plumbing through the public API**:\n//! [`calc_restore_plan`] honors a target TXID, but the\n//! timestamp path and `RestoreOptions` surface stay minimal for the one-shot.\n\n// Restore phase durations are diagnostics in this stand-alone public crate.\n// celld injects its scheduling facilities at the engine boundary.\n#![allow(clippy::disallowed_methods)]\n\nuse crate::client::ReplicaClient;\nuse crate::db::Db;\nuse crate::error::{new_ltx_error, Error, Result};\nuse crate::ltx::{self, FileInfo};\nuse crate::{Pos, TXID};\nuse futures_util::stream;\nuse futures_util::StreamExt;\nuse futures_util::TryStreamExt;\nuse std::collections::{BTreeMap, HashMap, HashSet};\nuse std::path::{Path, PathBuf};\nuse std::sync::Arc;\nuse std::time::Instant;\nuse tokio::sync::Semaphore;\n\n/// Keep enough reads in flight to hide an object store's round-trip latency,\n/// while the caller's semaphore bounds aggregate restore traffic.\nconst RESTORE_DOWNLOAD_CONCURRENCY: usize = 64;\n\n/// The compaction level which full snapshots are held at.\n///\n/// Ported from `SnapshotLevel` (compaction_level.go:9). In the L0-only one-shot\n/// scope no files land here (snapshots are written at L0), but\n/// [`calc_restore_plan`] still probes it first so the algorithm stays a faithful\n/// port that works the moment compaction is added.\npub use crate::compaction_level::SNAPSHOT_LEVEL;\n\n/// The number of compaction levels that one restore-plan calculation can list\n/// concurrently. A caller-supplied semaphore still bounds all remote requests.\nconst RESTORE_PLAN_CONCURRENCY: usize = SNAPSHOT_LEVEL as usize + 1;\n\n/// Whether a sync error means the cached replica position can no longer be\n/// trusted and must be re-derived from the store on the next sync.\n///\n/// True only for errors that say the local/replica state itself diverged. A\n/// transient store or IO error is not one of these: a fenced single writer's\n/// position is still valid, and its re-upload is an idempotent overwrite. See\n/// [`Replica::sync`] for why celld narrows Litestream's clear-on-any-error.\n///\n/// The walk follows `source()` the way Go's `errors.Is` follows `%w`, and it\n/// also steps into a [`SyncStepError`], which hides its cause from `source()`.\n/// Matching only the outer variant made the reset unreachable: every sync\n/// error was already `Other`.\nfn pos_untrustworthy(err: &Error) -> bool {\n let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(err);\n while let Some(error) = cause {\n if matches!(\n error.downcast_ref::(),\n Some(\n Error::NoSnapshots\n | Error::ChecksumMismatch\n | Error::LTXCorrupted\n | Error::LTXMissing\n | Error::TxNotAvailable\n )\n ) {\n return true;\n }\n cause = match error.downcast_ref::() {\n Some(step) => Some(&step.cause),\n None => error.source(),\n };\n }\n false\n}\n\n/// A sync step's error, named for the step and keeping the original error so\n/// that [`pos_untrustworthy`] can classify it. Formatting the original into a\n/// string, as the port first did, discarded that variant.\n///\n/// The message already prints the cause, so `source()` does not return it.\n/// Returning it too makes a reporter that walks the chain print the cause a\n/// second time, after the step line.\n#[derive(Debug, thiserror::Error)]\n#[error(\"{step}: {cause}\")]\nstruct SyncStepError {\n step: &'static str,\n cause: Error,\n}\n\n/// Wraps an error from the named sync step, the Go `fmt.Errorf(\"step: %w\")`.\nfn sync_step(step: &'static str) -> impl FnOnce(Error) -> Error {\n move |cause| Error::Other(Box::new(SyncStepError { step, cause }))\n}\n\n/// Connects a database to a replication destination via a [`ReplicaClient`].\n///\n/// Ported from `Replica` (replica.go:30-59). The Go type also owns the\n/// background-monitor machinery (`wg`/`cancel`/`f`) and tunables\n/// (`SyncInterval`/`MonitorEnabled`); those drive the deferred monitor loop and\n/// are intentionally omitted here (see module docs).\npub struct Replica {\n /// The database being replicated. `None` for a restore-only replica (the Go\n /// `NewReplicaWithClient(nil, client)` restore-only shape).\n db: Option,\n\n /// Client used to connect to the remote replica.\n pub client: C,\n\n /// Current replicated position (`replica.go:33-34` `pos`).\n pos: Pos,\n\n /// Whether `pos` is known and needs no `calc_pos` listing to re-derive.\n /// Seeded true at activation (see [`Self::seed_pos`]); reset false only when\n /// a divergence error says the position can no longer be trusted. Upstream\n /// has no equivalent -- it lists whenever `pos` is zero -- which is the\n /// listing celld's fencing lets it skip.\n pos_known: bool,\n}\n\nimpl Replica {\n /// Creates a replica that owns `db` and replicates through `client`.\n ///\n /// Ported from `NewReplicaWithClient` (replica.go:73-77).\n pub fn new(db: Db, client: C) -> Self {\n Replica {\n db: Some(db),\n client,\n pos: Pos::ZERO,\n pos_known: false,\n }\n }\n\n /// Creates a restore-only replica with no attached database (Go's\n /// `NewReplicaWithClient(nil, client)`).\n pub fn new_client_only(client: C) -> Self {\n Replica {\n db: None,\n client,\n pos: Pos::ZERO,\n pos_known: false,\n }\n }\n\n /// Returns a reference to the attached database, if any.\n /// Ported from `Replica.DB` (replica.go:89).\n pub fn db(&self) -> Option<&Db> {\n self.db.as_ref()\n }\n\n /// Consumes the replica and returns the attached database, if any. Lets a\n /// host perform a clean [`Db::close`] (which consumes the `Db`) after a final\n /// sync — the shutdown ordering the deferred monitor loop would otherwise own\n /// (replica.go `Stop` → `db.Close`).\n pub fn into_db(self) -> Option {\n self.db\n }\n\n /// Returns a mutable reference to the attached database, if any. Lets a\n /// caller drive `db.sync()` (the local capture half) before `replica.sync()`\n /// (the upload half) — the `SyncAndWait` ordering (db.go:500-512).\n pub fn db_mut(&mut self) -> Option<&mut Db> {\n self.db.as_mut()\n }\n\n /// The current replicated position. Ported from `Replica.Pos`\n /// (replica.go:237-241).\n pub fn pos(&self) -> Pos {\n self.pos\n }\n\n /// Sets the current replicated position. Ported from `Replica.SetPos`\n /// (replica.go:244-248).\n fn set_pos(&mut self, pos: Pos) {\n self.pos = pos;\n }\n\n /// Seed the replicated position from the caller's known-durable state and\n /// mark it known, so the next sync skips the `calc_pos` listing. A fresh\n /// cell seeds 0; a just-restored cell seeds its restored max. celld always\n /// knows this at activation -- local equals remote under epoch fencing --\n /// so the listing that only existed to re-discover the position, and that\n /// storms a rate-limiting store, is not issued on the activation path.\n pub fn seed_pos(&mut self, pos: Pos) {\n self.pos = pos;\n self.pos_known = true;\n }\n\n /// Copies new L0 LTX files from the local capture directory to the replica.\n ///\n /// Ported from `Replica.Sync` (replica.go:132-180). A divergence error clears\n /// the cached position so the next sync recomputes it from the replica; a\n /// transient error keeps it (see [`pos_untrustworthy`]). Requires an\n /// attached database.\n pub async fn sync(&mut self) -> Result<()> {\n match self.sync_inner().await {\n Ok(()) => Ok(()),\n Err(e) => {\n // Litestream clears the cached position on *any* error so the\n // next sync re-derives it with a `calc_pos` listing\n // (replica.go:137-143) -- safe there only because it is single-\n // node with no fencing. celld fences one writer per epoch\n // through the store, so within an epoch a transient store/IO\n // error does not invalidate our own position; re-uploads are\n // idempotent overwrites. Forget the position only when the\n // error says the state itself diverged, else a rate-limiting or\n // slow store turns every write into a listing.\n if pos_untrustworthy(&e) {\n self.pos = Pos::ZERO;\n self.pos_known = false;\n }\n Err(e)\n }\n }\n }\n\n async fn sync_inner(&mut self) -> Result<()> {\n // Re-derive the replica position with a listing only when it is not\n // already known (replica.go:146-152). celld seeds it at activation from\n // local state that equals the remote under epoch fencing, so a listing\n // -- which only ever re-discovered a position the owner already knows --\n // never runs on the hot path, and a fresh cell's pointless list of an\n // empty prefix is skipped entirely.\n if !self.pos_known {\n let pos = self.calc_pos().await.map_err(sync_step(\"calc pos\"))?;\n self.set_pos(pos);\n self.pos_known = true;\n }\n\n // Find current position of the database (replica.go:155-160).\n let dpos = {\n let db = self\n .db\n .as_mut()\n .ok_or_else(|| Error::Other(\"no database attached to replica\".into()))?;\n db.pos()\n .map_err(sync_step(\"cannot determine current position\"))?\n };\n if dpos.is_zero() {\n return Err(Error::Other(\"no position, waiting for data\".into()));\n }\n\n // Replicate all L0 LTX files since the last replica position\n // (replica.go:169-174). Each successful upload advances pos by one TXID;\n // re-reading `self.pos()` each iteration mirrors the Go loop exactly.\n loop {\n let tx_id = TXID(self.pos().txid.0 + 1);\n if tx_id > dpos.txid {\n break;\n }\n self.upload_ltx_file(0, tx_id, tx_id).await?;\n self.set_pos(Pos::new(tx_id, 0));\n }\n\n Ok(())\n }\n\n /// Uploads a single local LTX file to the replica.\n ///\n /// Ported from `Replica.uploadLTXFile` (replica.go:182-205). A failure to\n /// open the local file is wrapped as an `LTXError{op:\"open\"}` so the monitor\n /// can classify auto-recoverable corruption (replica.go:186). The write\n /// itself is delegated to the client.\n async fn upload_ltx_file(&mut self, level: i32, min_txid: TXID, max_txid: TXID) -> Result<()> {\n let db = self\n .db\n .as_ref()\n .ok_or_else(|| Error::Other(\"no database attached to replica\".into()))?;\n let filename = db.ltx_path(level as u32, min_txid, max_txid);\n\n let data = match db.read_ltx_file(level as u32, min_txid, max_txid) {\n Ok(b) => b,\n Err(e) => {\n return Err(Error::Ltx(Box::new(new_ltx_error(\n \"open\", &filename, level, min_txid.0, max_txid.0, e,\n ))));\n }\n };\n\n self.client\n .write_ltx_file(level, min_txid, max_txid, &data)\n .await\n .map_err(sync_step(\"write ltx file\"))?;\n\n Ok(())\n }\n\n /// Returns the last position saved to the replica for level 0.\n ///\n /// Ported from `Replica.calcPos` (replica.go:208-214) + `MaxLTXFileInfo`\n /// (replica.go:218-233): scans the L0 listing for the highest `max_txid`.\n async fn calc_pos(&self) -> Result {\n let info = self\n .max_ltx_file_info(0)\n .await\n .map_err(sync_step(\"max ltx file\"))?;\n Ok(Pos::new(info.max_txid, info.post_apply_checksum))\n }\n\n /// Metadata about the last LTX file for a given level (highest `max_txid`),\n /// or a zero `FileInfo` if none exist. Ported from `Replica.MaxLTXFileInfo`\n /// (replica.go:218-233).\n async fn max_ltx_file_info(&self, level: i32) -> Result {\n let files = self.client.ltx_files(level, TXID(0)).await?;\n let mut info = FileInfo::default();\n for item in files {\n if item.max_txid > info.max_txid {\n info = item;\n }\n }\n Ok(info)\n }\n\n /// Restores the database from this replica's client into `output_path`.\n ///\n /// Convenience wrapper over [`restore`] using this replica's client and the\n /// most-recent state (no target TXID). Mirrors the common\n /// `Replica.Restore(ctx, opt)` call with `OutputPath` set and no TXID\n /// (replica.go:533).\n pub async fn restore(&self, output_path: impl AsRef) -> Result {\n restore(&self.client, output_path, TXID(0)).await\n }\n\n /// Detects when the local database has been restored to an earlier state\n /// than the replica (a lower TXID) and, if so, seeds the local L0 directory\n /// with the replica's newest L0 file so the next sync snapshots forward.\n ///\n /// Ported from `DB.checkDatabaseBehindReplica` (db.go:1211-1294), issue #781.\n /// In upstream this runs inside `DB.init()` because the `DB` owns its\n /// `Replica`; our synchronous `Db` does not, so the orchestration lives here\n /// on `Replica` and a host calls it once after [`Db::open`], before the first\n /// sync. The file-writing tail is [`Db::seed_l0_baseline`].\n ///\n /// Without this, a hard recovery (restore an old snapshot, reopen, write new\n /// data) would silently drop the new writes: the fresh local DB snapshots at\n /// TXID 1, but [`Replica::sync`] computes the replica position from the\n /// remote's higher `MaxTXID`, so its upload loop (`pos+1 ..= db.pos`) never\n /// runs (`pos+1` already exceeds the local DB's TXID). Seeding the remote\n /// baseline makes the next [`Db::sync`] see a continuity break and snapshot at\n /// the current (post-restore-plus-writes) state, which then uploads.\n ///\n /// No-op (returns `Ok`) when there is no remote data, when the database is at\n /// or ahead of the replica, or when there is no attached database.\n pub async fn check_database_behind_replica(&mut self) -> Result<()> {\n // Replica position from remote (db.go:1224-1230). Done first so a\n // restore-only replica with no DB is a clean no-op.\n let replica_info = self.max_ltx_file_info(0).await?;\n if replica_info.max_txid == TXID(0) {\n return Ok(()); // no remote replica data yet\n }\n\n let db = match self.db.as_mut() {\n Some(db) => db,\n None => return Ok(()),\n };\n\n // Database position from local L0 files (db.go:1218-1222).\n let db_pos = db\n .pos()\n .map_err(|e| Error::Other(format!(\"get database position: {e}\").into()))?;\n\n // If the database is ahead or equal, nothing to do (db.go:1232-1235).\n if db_pos.txid >= replica_info.max_txid {\n return Ok(());\n }\n\n // Fetch the latest L0 LTX file from the replica (db.go:1251-1257).\n let min_txid = replica_info.min_txid;\n let max_txid = replica_info.max_txid;\n let data = self\n .client\n .open_ltx_file(0, min_txid, max_txid)\n .await\n .map_err(|e| Error::Other(format!(\"open remote L0 file: {e}\").into()))?;\n\n // Seed it as the local baseline (db.go:1259-1293).\n db.seed_l0_baseline(min_txid, max_txid, &data)?;\n\n // Drop the now-stale cached replica position so the next sync recomputes\n // it from the remote (mirrors clearing pos on a state change).\n self.pos = Pos::ZERO;\n Ok(())\n }\n}\n\n/// Restores a database from `client` into `output_path`, optionally up to a\n/// target `txid` (`TXID(0)` = most recent state).\n///\n/// Ported from `Replica.Restore` (replica.go:533-725), LTX path only. Steps:\n/// 1. refuse to overwrite an existing output (replica.go:591-595);\n/// 2. [`calc_restore_plan`] → the ordered snapshot+incremental file list;\n/// 3. download + merge them (compactor semantics) and reconstruct the database\n/// image the way `Decoder.DecodeDatabaseTo` does;\n/// 4. write to `.tmp`, fsync, and atomically rename (replica.go:657-694).\n///\n/// The V3 generation path and follow mode are out of scope (see module docs).\npub async fn restore(\n client: &C,\n output_path: impl AsRef,\n txid: TXID,\n) -> Result {\n restore_with_host_and_download_slots(\n client,\n output_path,\n txid,\n crate::LtxHost::default(),\n Arc::new(Semaphore::new(1)),\n )\n .await\n}\n\n/// What a restore read: the plan's shape, for the caller's telemetry.\n#[derive(Debug, Clone)]\npub struct RestorePlanStats {\n pub objects: usize,\n pub bytes: u64,\n /// Object count per compaction level, ordered by level.\n pub by_level: BTreeMap,\n}\n\n/// Wall-clock phases for one successful restore.\n#[derive(Debug, Clone)]\npub struct RestoreTimingStats {\n pub plan: RestorePlanStats,\n pub plan_us: u64,\n pub download_us: u64,\n pub apply_us: u64,\n}\n\n/// Restores a database like [`restore`], but downloads independent LTX files\n/// concurrently. All restores that share `download_slots` share one hard I/O\n/// ceiling, so a cold cohort cannot multiply the limit per cell.\npub async fn restore_with_download_slots(\n client: &C,\n output_path: impl AsRef,\n txid: TXID,\n download_slots: Arc,\n) -> Result {\n restore_with_host_and_download_slots(\n client,\n output_path,\n txid,\n crate::LtxHost::default(),\n download_slots,\n )\n .await\n}\n\n/// Restores a database through an injected local filesystem.\npub async fn restore_with_host_and_download_slots(\n client: &C,\n output_path: impl AsRef,\n txid: TXID,\n host: crate::LtxHost,\n download_slots: Arc,\n) -> Result {\n Ok(\n restore_timed_with_host_and_download_slots(client, output_path, txid, host, download_slots)\n .await?\n .plan,\n )\n}\n\n/// Restores a database and reports the plan, download, and local apply times.\npub async fn restore_timed_with_download_slots(\n client: &C,\n output_path: impl AsRef,\n txid: TXID,\n download_slots: Arc,\n) -> Result {\n restore_timed_with_host_and_download_slots(\n client,\n output_path,\n txid,\n crate::LtxHost::default(),\n download_slots,\n )\n .await\n}\n\n/// Restores a database through an injected local filesystem and reports the\n/// plan, download, and local apply times.\npub async fn restore_timed_with_host_and_download_slots(\n client: &C,\n output_path: impl AsRef,\n txid: TXID,\n host: crate::LtxHost,\n download_slots: Arc,\n) -> Result {\n let output_path = output_path.as_ref();\n ensure_restore_output_absent(&host, output_path)?;\n let plan_started = Instant::now();\n let infos = calc_restore_plan_with_slots(client, txid, download_slots.clone()).await?;\n let plan_us = plan_started.elapsed().as_micros() as u64;\n restore_from_plan_inner(client, output_path, infos, host, download_slots, plan_us).await\n}\n\n/// Restores a database from a plan which [`calc_restore_plan`] produced.\n///\n/// This path lets a caller reuse the exact plan that fixed an epoch seal. It\n/// therefore avoids a second remote listing before it downloads the files.\npub async fn restore_from_plan_with_download_slots(\n client: &C,\n output_path: impl AsRef,\n plan: Vec,\n download_slots: Arc,\n) -> Result {\n let output_path = output_path.as_ref();\n let host = crate::LtxHost::default();\n ensure_restore_output_absent(&host, output_path)?;\n restore_from_plan_inner(client, output_path, plan, host, download_slots, 0).await\n}\n\nfn ensure_restore_output_absent(host: &crate::LtxHost, output_path: &Path) -> Result<()> {\n // Ensure output path does not already exist (replica.go:591-595).\n match host.metadata(output_path) {\n Ok(_) => {\n return Err(Error::Other(\n format!(\n \"cannot restore, output path already exists: {}\",\n output_path.display()\n )\n .into(),\n ));\n }\n Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}\n Err(e) => return Err(e.into()),\n }\n Ok(())\n}\n\n/// Maps a download failure during restore. A planned object that is gone\n/// becomes [`Error::LTXMissing`], so a caller that owns the listing can build\n/// a fresh plan: epoch GC can delete an object between plan selection and its\n/// download. Every other failure stays opaque and aborts the restore.\nfn classify_restore_open_error(e: Error) -> Error {\n match &e {\n Error::Io(io) if io.kind() == std::io::ErrorKind::NotFound => Error::LTXMissing,\n _ => Error::Other(format!(\"open ltx file: {e}\").into()),\n }\n}\n\nasync fn restore_from_plan_inner(\n client: &C,\n output_path: &Path,\n infos: Vec,\n host: crate::LtxHost,\n download_slots: Arc,\n plan_us: u64,\n) -> Result {\n // Validate the whole plan before starting I/O. An invalid later entry must\n // not cause a partial download burst before the restore fails.\n for info in &infos {\n if info.size < ltx::HEADER_SIZE as i64 {\n return Err(Error::Other(\n format!(\n \"invalid ltx file: level={} min={} max={} has size {} bytes (minimum {})\",\n info.level,\n info.min_txid,\n info.max_txid,\n info.size,\n ltx::HEADER_SIZE\n )\n .into(),\n ));\n }\n }\n\n let mut stats = RestorePlanStats {\n objects: infos.len(),\n bytes: infos.iter().map(|info| info.size.max(0) as u64).sum(),\n by_level: BTreeMap::new(),\n };\n for info in &infos {\n *stats.by_level.entry(info.level).or_insert(0) += 1;\n }\n\n // Files are independent reads. `buffered` overlaps them but yields them in\n // plan order, which preserves the compactor input contract. The shared\n // semaphore is the node-level ceiling across every concurrent restore.\n let download_started = Instant::now();\n let files: Vec> = stream::iter(infos)\n .map(|info| {\n let download_slots = download_slots.clone();\n async move {\n let _permit = download_slots\n .acquire_owned()\n .await\n .expect(\"restore download semaphore closed\");\n client\n .open_ltx_file(info.level, info.min_txid, info.max_txid)\n .await\n .map_err(classify_restore_open_error)\n }\n })\n .buffered(RESTORE_DOWNLOAD_CONCURRENCY)\n .try_collect()\n .await?;\n let download_us = download_started.elapsed().as_micros() as u64;\n\n if files.is_empty() {\n return Err(Error::Other(\"no matching backup files available\".into()));\n }\n\n // Merge the files (compactor) and reconstruct the SQLite database image.\n let apply_started = Instant::now();\n let image = build_database_image(&files)?;\n\n // Create the parent directory if needed (replica.go:649-655).\n if let Some(parent) = output_path.parent() {\n if !parent.as_os_str().is_empty() {\n host.create_dir_all(parent)?;\n }\n }\n\n // Output to a temp file & atomically rename (replica.go:657-694).\n let tmp_output_path = append_ext(output_path, \"tmp\");\n write_file_atomic(&host, &tmp_output_path, output_path, &image)?;\n let apply_us = apply_started.elapsed().as_micros() as u64;\n\n Ok(RestoreTimingStats {\n plan: stats,\n plan_us,\n download_us,\n apply_us,\n })\n}\n\n/// Returns the ordered list of LTX files needed to restore the database at\n/// `txid` (`TXID(0)` = latest).\n///\n/// Ported from `CalcRestorePlan` (replica.go:1419-1536). Anchors on the latest\n/// snapshot at or before the target, then walks compaction levels\n/// `SnapshotLevel-1 ..= 0` with per-level cursors, repeatedly choosing the file\n/// that extends the longest contiguous TXID range. In the L0-only scope the\n/// snapshot anchor lives at L0 (MinTXID==1) and the only cursor with files is\n/// level 0, so the result is the contiguous chain `1..=N` (capped at `txid`).\npub async fn calc_restore_plan(client: &C, txid: TXID) -> Result> {\n calc_restore_plan_with_slots(\n client,\n txid,\n Arc::new(Semaphore::new(RESTORE_PLAN_CONCURRENCY)),\n )\n .await\n}\n\n/// Calculates a restore plan while a shared semaphore bounds remote listings.\npub async fn calc_restore_plan_with_slots(\n client: &C,\n txid: TXID,\n request_slots: Arc,\n) -> Result> {\n let mut infos: Vec = Vec::new();\n\n // Every level is independent. List them together instead of paying ten\n // object-store round trips in series. The shared request slots keep a cold\n // cohort from multiplying this fan-out by its activation count.\n let levels: Vec<(i32, Vec)> = stream::iter((0..=SNAPSHOT_LEVEL).rev())\n .map(|level| {\n let request_slots = request_slots.clone();\n async move {\n let _permit = request_slots\n .acquire_owned()\n .await\n .expect(\"restore request semaphore closed\");\n Ok::<_, Error>((level, client.ltx_files(level, TXID(0)).await?))\n }\n })\n .buffered(RESTORE_PLAN_CONCURRENCY)\n .try_collect()\n .await?;\n let mut levels = levels.into_iter().collect::>();\n\n // Start with the latest snapshot before the target TXID (replica.go:1430-1452).\n let snapshot_files = levels.remove(&SNAPSHOT_LEVEL).unwrap_or_default();\n let mut snapshot: Option = None;\n for info in snapshot_files {\n if txid != TXID(0) && info.max_txid > txid {\n continue;\n }\n snapshot = Some(info);\n }\n if let Some(s) = snapshot {\n infos.push(s);\n }\n\n // Collect candidates across all compaction levels and pick the next file\n // from any level that extends the longest contiguous TXID range\n // (replica.go:1454-1515).\n let max_level = SNAPSHOT_LEVEL - 1;\n let start_txid = slice_max_txid(&infos);\n let mut current_max = start_txid;\n if txid != TXID(0) && current_max >= txid {\n return Ok(infos);\n }\n\n // Build a cursor per level, highest level first (replica.go:1463-1473).\n let mut cursors: Vec = Vec::with_capacity((max_level + 1) as usize);\n for level in (0..=max_level).rev() {\n let files = levels.remove(&level).unwrap_or_default();\n cursors.push(RestoreLevelCursor::new(files));\n }\n\n loop {\n // Choose the best candidate across all level cursors (replica.go:1483-1494).\n let mut next_idx: Option = None;\n for i in 0..cursors.len() {\n cursors[i].refresh(current_max, txid);\n if cursors[i].candidate.is_none() {\n continue;\n }\n match next_idx {\n None => next_idx = Some(i),\n Some(ni) => {\n let cand = cursors[i].candidate.as_ref().unwrap();\n let best = cursors[ni].candidate.as_ref().unwrap();\n if restore_candidate_better(best, cand) {\n next_idx = Some(i);\n }\n }\n }\n }\n\n let ni = match next_idx {\n Some(ni) => ni,\n None => break,\n };\n\n // Take the chosen candidate (replica.go:1500-1510).\n let cand = cursors[ni].candidate.take().unwrap();\n if cand.max_txid <= current_max {\n continue;\n }\n current_max = cand.max_txid;\n infos.push(cand);\n\n if txid != TXID(0) && current_max >= txid {\n break;\n }\n }\n\n // For a latest/most-recent restore, verify the tail is contiguous\n // (replica.go:1517-1526).\n if !infos.is_empty() && txid == TXID(0) {\n for cursor in cursors.iter_mut() {\n cursor.ensure_current();\n if let Some(cur) = &cursor.current {\n if cur.min_txid.0 > current_max.0 + 1 {\n return Err(Error::Other(\n format!(\n \"non-contiguous ltx files: have up to {} but next file starts at {}\",\n current_max, cur.min_txid\n )\n .into(),\n ));\n }\n }\n }\n }\n\n if infos.is_empty() {\n return Err(Error::TxNotAvailable);\n }\n if txid != TXID(0) && slice_max_txid(&infos) < txid {\n return Err(Error::TxNotAvailable);\n }\n\n Ok(infos)\n}\n\n/// A single level's streaming view during restore planning.\n///\n/// Ported from `restoreLevelCursor` (replica.go:1538-1603). The Go version\n/// streams from a `FileIterator`; here the client already returns a sorted\n/// `Vec`, so the \"iterator\" is the slice plus a read index.\nstruct RestoreLevelCursor {\n /// Files for this level, sorted ascending (the `LTXFiles` contract).\n files: Vec,\n /// Read index into `files` (the iterator cursor).\n idx: usize,\n /// Last item read but not yet evaluated (`current`, replica.go:1542).\n current: Option,\n /// Best eligible file at this level for the current `current_max`\n /// (`candidate`, replica.go:1544).\n candidate: Option,\n /// True once the iterator is exhausted (`done`, replica.go:1546).\n done: bool,\n}\n\nimpl RestoreLevelCursor {\n fn new(files: Vec) -> Self {\n RestoreLevelCursor {\n files,\n idx: 0,\n current: None,\n candidate: None,\n done: false,\n }\n }\n\n /// Advances the iterator while files could be contiguous with `current_max`,\n /// keeping the best eligible candidate. Ported from `refresh`\n /// (replica.go:1549-1587).\n fn refresh(&mut self, current_max: TXID, txid: TXID) {\n if self.done {\n return;\n }\n if let Some(c) = &self.candidate {\n if c.max_txid <= current_max {\n self.candidate = None;\n }\n }\n\n loop {\n self.ensure_current();\n if self.done {\n return;\n }\n\n let info = self.current.clone().unwrap();\n if info.min_txid.0 > current_max.0 + 1 {\n return;\n }\n self.current = None;\n\n if info.max_txid <= current_max {\n continue;\n }\n if txid != TXID(0) && info.max_txid > txid {\n continue;\n }\n\n match &self.candidate {\n None => self.candidate = Some(info),\n Some(c) => {\n if restore_candidate_better(c, &info) {\n self.candidate = Some(info);\n }\n }\n }\n }\n }\n\n /// Populates `current` with the next item, or marks `done`. Ported from\n /// `ensureCurrent` (replica.go:1589-1603).\n fn ensure_current(&mut self) {\n if self.done || self.current.is_some() {\n return;\n }\n if self.idx >= self.files.len() {\n self.done = true;\n return;\n }\n self.current = Some(self.files[self.idx].clone());\n self.idx += 1;\n }\n}\n\n/// True if `next` is a strictly better restore candidate than `curr`: longer\n/// reach first (`MaxTXID`), then a smaller `MinTXID` (more coverage), then a\n/// higher level (more compacted), then an earlier `created_at`.\n///\n/// Ported from `restoreCandidateBetter` (replica.go:1605-1616).\nfn restore_candidate_better(curr: &FileInfo, next: &FileInfo) -> bool {\n if next.max_txid != curr.max_txid {\n return next.max_txid > curr.max_txid;\n }\n if next.min_txid != curr.min_txid {\n return next.min_txid < curr.min_txid;\n }\n if next.level != curr.level {\n return next.level > curr.level;\n }\n // CreatedAt.Before(curr): an unknown (None) timestamp is treated as not\n // earlier, matching Go's zero-time comparison being false here.\n match (next.created_at, curr.created_at) {\n (Some(n), Some(c)) => n < c,\n _ => false,\n }\n}\n\n/// Maximum `max_txid` across a slice, `TXID(0)` if empty. Mirrors\n/// `FileInfoSlice.MaxTXID` (ltx.go:612-619); the slice here is already in plan\n/// (ascending) order, but we scan defensively rather than assume.\nfn slice_max_txid(infos: &[FileInfo]) -> TXID {\n infos.iter().map(|f| f.max_txid).max().unwrap_or(TXID(0))\n}\n\n/// Merges the LTX `files` (a snapshot followed by incrementals, in plan order)\n/// and reconstructs the full SQLite database image.\n///\n/// This replaces Go's `ltx.NewCompactor(pw, rdrs)` + `Decoder.DecodeDatabaseTo`\n/// (replica.go:667-682). The compactor's contract (ltx@v0.5.1 compactor.go):\n/// * page sizes must match across inputs;\n/// * the **last** input's `Commit` is the final database size;\n/// * for each page number, the **latest** input that carries it wins\n/// (compactor.go:198-228 iterates inputs newest-first);\n/// * pages numbered beyond the final `Commit` are dropped (truncation).\n///\n/// `DecodeDatabaseTo` then writes pages `1..=Commit`, zero-filling the lock page\n/// (decoder.go:236-254). Every input is fully decoded and checksum-verified, so\n/// a corrupt download is caught here.\nfn build_database_image(files: &[Vec]) -> Result> {\n let mut page_size = None;\n let mut commit = None;\n let mut merged: HashMap> = HashMap::new();\n\n for data in files {\n let (decoded, pages) = ltx::decode_file_with_pages(data)?;\n let header = decoded.header;\n if let Some(expected) = page_size {\n if header.page_size != expected {\n return Err(Error::Other(\n format!(\n \"input files have mismatched page sizes: {} != {}\",\n expected, header.page_size\n )\n .into(),\n ));\n }\n } else {\n page_size = Some(header.page_size);\n }\n let previous_commit = commit.unwrap_or(0);\n if header.commit > previous_commit {\n let present: HashSet = pages.iter().map(|(pgno, _)| *pgno).collect();\n let lock = ltx::lock_pgno(header.page_size);\n for pgno in (previous_commit + 1)..=header.commit {\n if pgno != lock && !present.contains(&pgno) && !merged.contains_key(&pgno) {\n return Err(Error::Other(\n format!(\n \"missing newly committed page {pgno} in LTX transaction {} (database grew from {previous_commit} to {} pages)\",\n header.max_txid, header.commit,\n )\n .into(),\n ));\n }\n }\n }\n\n // Inputs arrive oldest to newest, so moving each page into the map makes\n // the latest input win without a second staging collection or clone.\n for (pgno, data) in pages {\n merged.insert(pgno, data);\n }\n commit = Some(header.commit);\n }\n\n let (Some(page_size), Some(commit)) = (page_size, commit) else {\n return Err(Error::Other(\n \"cannot build a database from no LTX files\".into(),\n ));\n };\n merged.retain(|pgno, _| *pgno <= commit);\n\n let page_size_usize = page_size as usize;\n let lock = ltx::lock_pgno(page_size);\n\n // Reconstruct the database image: pages 1..=commit, lock page zero-filled\n // (decoder.go:236-254). A non-lock page missing from the merge means the\n // backup chain is incomplete.\n let mut image = Vec::with_capacity(commit as usize * page_size_usize);\n let zero_page = vec![0u8; page_size_usize];\n for pgno in 1..=commit {\n if pgno == lock {\n image.extend_from_slice(&zero_page);\n continue;\n }\n match merged.get(&pgno) {\n Some(data) => image.extend_from_slice(data),\n None => {\n return Err(Error::Other(\n format!(\"missing page {pgno} in restore plan (incomplete backup)\").into(),\n ));\n }\n }\n }\n\n Ok(image)\n}\n\n/// Appends `ext` to `path` as a literal suffix (`p` → `p.tmp`), the way Go's\n/// `opt.OutputPath + \".tmp\"` does (NOT `Path::set_extension`, which would replace\n/// an existing extension).\nfn append_ext(path: &Path, ext: &str) -> PathBuf {\n let mut s = path.as_os_str().to_owned();\n s.push(\".\");\n s.push(ext);\n PathBuf::from(s)\n}\n\n/// Writes `data` to `tmp_path`, fsyncs, then renames onto `final_path` — the\n/// crash-consistent atomic-write idiom (replica.go:661-694). On any failure the\n/// temp file is removed.\nfn write_file_atomic(\n host: &crate::LtxHost,\n tmp_path: &Path,\n final_path: &Path,\n data: &[u8],\n) -> Result<()> {\n let result = (|| -> Result<()> {\n let mut file = host.create(tmp_path)?;\n file.write_all(data)?;\n file.sync_all()?;\n drop(file);\n host.rename(tmp_path, final_path)?;\n Ok(())\n })();\n if result.is_err() {\n let _ = host.remove_file(tmp_path);\n }\n result\n}\n\n#[doc(hidden)]\npub mod internal {\n use super::*;\n\n pub const RESTORE_PLAN_CONCURRENCY: usize = super::RESTORE_PLAN_CONCURRENCY;\n\n pub async fn upload_ltx_file(\n replica: &mut Replica,\n level: i32,\n min_txid: TXID,\n max_txid: TXID,\n ) -> Result<()> {\n replica.upload_ltx_file(level, min_txid, max_txid).await\n }\n\n pub fn build_database_image(files: &[Vec]) -> Result> {\n super::build_database_image(files)\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",{"_54":55,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_42":31,"_89":82,"_90":31,"_48":49,"_50":49},"branches",[82],"tags",[69,70,71,72,73,74,75,76,77,78,79,80,81],"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","main","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":49,"_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,"_52":125,"_126":31,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_52":125,"_126":31,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_52":125,"_126":31,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_52":125,"_126":31,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_52":125,"_126":31,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_52":125,"_126":31,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_52":31,"_126":31,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_52":31,"_126":31,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_52":31,"_126":31,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_52":31,"_126":31,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_52":31,"_126":31,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_52":31,"_126":31,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_52":31,"_126":31,"_127":31},{"_119":153,"_121":154,"_15":153,"_123":155,"_52":31,"_126":31,"_127":31},{"_119":150,"_121":151,"_15":150,"_123":152,"_52":31,"_126":31,"_127":31},{"_119":147,"_121":148,"_15":147,"_123":149,"_52":31,"_126":31,"_127":31},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794434635396,"sidebar","expanded"]