From 727014c907655f07998d6d7b67b1878268f8c077 Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 05:21:37 -0700 Subject: [PATCH 1/6] feat: expose bounded operator metrics --- Cargo.lock | 2 +- crates/celld/Cargo.toml | 2 +- crates/celld/lib.rs | 1 + crates/celld/main.rs | 58 ++++++- crates/celld/metrics.rs | 332 ++++++++++++++++++++++++++++++++++++++++ crates/logic/lib.rs | 36 ++++- docs/security.md | 3 + docs/telemetry.md | 17 +- 8 files changed, 442 insertions(+), 9 deletions(-) create mode 100644 crates/celld/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index e9505e2a2..b4f17b329 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -288,7 +288,7 @@ dependencies = [ [[package]] name = "celld" -version = "0.2.1-coderabbit.2" +version = "0.2.1-coderabbit.3" dependencies = [ "aes", "aes-gcm", diff --git a/crates/celld/Cargo.toml b/crates/celld/Cargo.toml index 33a3912b9..62bc949d1 100644 --- a/crates/celld/Cargo.toml +++ b/crates/celld/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "celld" -version = "0.2.1-coderabbit.2" +version = "0.2.1-coderabbit.3" edition = "2021" rust-version = "1.94.1" license = "Apache-2.0" diff --git a/crates/celld/lib.rs b/crates/celld/lib.rs index 9131568f2..767163167 100644 --- a/crates/celld/lib.rs +++ b/crates/celld/lib.rs @@ -26,6 +26,7 @@ pub mod fleet; pub mod js; pub mod ltx_repl; pub mod memory; +pub mod metrics; mod otlp; pub mod ownership_store; pub mod peer_auth; diff --git a/crates/celld/main.rs b/crates/celld/main.rs index 75ef5e03f..6c8519d2e 100644 --- a/crates/celld/main.rs +++ b/crates/celld/main.rs @@ -375,6 +375,9 @@ enum Message { Snapshot { reply: oneshot::Sender, }, + Metrics { + reply: oneshot::Sender, + }, Health { reply: oneshot::Sender, }, @@ -880,6 +883,16 @@ impl AppHandle { .unwrap_or_else(|_| "{\"error\":\"actor_stopped\"}".into()) } + async fn metrics(&self) -> String { + let (reply, receive) = oneshot::channel(); + if self.tx.send(Message::Metrics { reply }).is_err() { + return "# celld actor stopped\n".into(); + } + receive + .await + .unwrap_or_else(|_| "# celld actor stopped\n".into()) + } + fn is_draining(&self) -> bool { self.draining.load(std::sync::atomic::Ordering::Relaxed) } @@ -1506,6 +1519,9 @@ impl Actor { Message::Snapshot { reply } => { let _ = reply.send(self.state_json()); } + Message::Metrics { reply } => { + let _ = reply.send(self.metrics_text()); + } Message::Health { reply } => { let _ = reply.send(self.state.ready_to_serve()); } @@ -2323,6 +2339,33 @@ impl Actor { ) } + fn metrics_text(&self) -> String { + let memory = celld::memory::sample(); + let phases = self.state.phase_census(); + let max_resident = self.state.max_resident(); + celld::metrics::render(&celld::metrics::NodeMetrics { + runtime_version: env!("CARGO_PKG_VERSION"), + region: &self.region, + ownership: self.ownership.name(), + serving: self.state.ready_to_serve(), + occupied: self.state.occupied(), + resident_limit: (max_resident != usize::MAX).then_some(max_resident), + evicting: self.state.evicting(), + restoring: self.state.activation_backlog(), + activating: self.state.activating(), + activation_waiting: self.state.activation_waiting().len(), + capacity_waiting: self.state.waiting().len(), + phases: &phases, + shed_reason: self.state.shed_reason(), + rss_bytes: memory.rss_bytes, + in_use_bytes: memory.in_use_bytes, + output_gate_pending: self.gated_responses.len() + self.ws_gated.len(), + publishes: self.publishes, + stops: self.stops, + activity: self.state.activity_snapshot(), + }) + } + fn begin_route_if_cold(&mut self, cell: &str) { if !matches!( self.state.phase(cell), @@ -2463,6 +2506,18 @@ fn response(status: StatusCode, body: impl Into) -> HttpReply { .expect("static HTTP response") } +fn metrics_response(body: impl Into) -> HttpReply { + Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/plain; version=0.0.4; charset=utf-8") + .body( + Full::new(body.into()) + .map_err(|never| match never {}) + .boxed_unsync(), + ) + .expect("static metrics response") +} + fn asset_response(response: axum::response::Response) -> HttpReply { response.map(|body| body.map_err(std::io::Error::other).boxed_unsync()) } @@ -3786,7 +3841,7 @@ async fn handle_internal( // but diagnostics is refused, and `Connection: close` tears the // keep-alive down so the drain loop can finish instead of holding every // idle connection open until the deadline. - if draining && !matches!(path.as_str(), "/__celld/probe" | "/state") { + if draining && !matches!(path.as_str(), "/__celld/probe" | "/state" | "/metrics") { let mut refused = response( StatusCode::SERVICE_UNAVAILABLE, "{\"ok\":false,\"draining\":true}", @@ -3807,6 +3862,7 @@ async fn handle_internal( let result = match path.as_str() { "/__celld/probe" => internal_probe(request, app).await, "/state" => response(StatusCode::OK, app.snapshot().await), + "/metrics" => metrics_response(app.metrics().await), "/shutdown" if request.method() != hyper::Method::POST => { response(StatusCode::METHOD_NOT_ALLOWED, "method not allowed") } diff --git a/crates/celld/metrics.rs b/crates/celld/metrics.rs new file mode 100644 index 000000000..460787a94 --- /dev/null +++ b/crates/celld/metrics.rs @@ -0,0 +1,332 @@ +// Copyright 2026 Deno Land Inc. Apache-2.0 license. + +//! Bounded Prometheus projection of the node decision core. +//! +//! The operator endpoint deliberately exposes no cell, request, tenant, or +//! bucket identifiers. Every label has a runtime-owned bounded vocabulary +//! except `region` and the build version supplied by the operator binary. + +use celld_logic::{ + pressure::{SHED_MEMORY, SHED_RSS_HARD}, + ActivitySnapshot, STABLE_PHASE_NAMES, +}; +use std::fmt::Write; + +pub struct NodeMetrics<'a> { + pub runtime_version: &'a str, + pub region: &'a str, + pub ownership: &'a str, + pub serving: bool, + pub occupied: usize, + pub resident_limit: Option, + pub evicting: usize, + pub restoring: usize, + pub activating: usize, + pub activation_waiting: usize, + pub capacity_waiting: usize, + pub phases: &'a [(&'static str, usize)], + pub shed_reason: Option<&'static str>, + pub rss_bytes: u64, + pub in_use_bytes: u64, + pub output_gate_pending: usize, + pub publishes: u64, + pub stops: u64, + pub activity: ActivitySnapshot, +} + +fn label_value(value: &str) -> String { + value + .replace('\\', "\\\\") + .replace('\n', "\\n") + .replace('"', "\\\"") +} + +fn help_and_type(output: &mut String, name: &str, help: &str, metric_type: &str) { + let _ = writeln!(output, "# HELP {name} {help}"); + let _ = writeln!(output, "# TYPE {name} {metric_type}"); +} + +pub fn render(metrics: &NodeMetrics<'_>) -> String { + let mut output = String::with_capacity(4096); + + help_and_type( + &mut output, + "celld_build_info", + "Build and deployment identity for this celld process.", + "gauge", + ); + let _ = writeln!( + output, + "celld_build_info{{version=\"{}\",region=\"{}\"}} 1", + label_value(metrics.runtime_version), + label_value(metrics.region), + ); + + help_and_type( + &mut output, + "celld_node_serving", + "Whether the node decision core currently permits request service.", + "gauge", + ); + let _ = writeln!(output, "celld_node_serving {}", u8::from(metrics.serving)); + + help_and_type( + &mut output, + "celld_node_ownership_state", + "Current node ownership backend state.", + "gauge", + ); + let _ = writeln!( + output, + "celld_node_ownership_state{{state=\"{}\"}} 1", + label_value(metrics.ownership), + ); + + for (name, help, value) in [ + ( + "celld_node_occupied_cells", + "Resident cells plus activation reservations on this node.", + metrics.occupied, + ), + ( + "celld_node_evicting_cells", + "Cells with an ownership eviction in flight.", + metrics.evicting, + ), + ( + "celld_node_restoring_cells", + "Cold routes either activating or queued for activation.", + metrics.restoring, + ), + ( + "celld_node_activating_cells", + "Cold routes currently holding an activation permit.", + metrics.activating, + ), + ( + "celld_node_activation_waiting_cells", + "Cold routes queued behind the activation ceiling.", + metrics.activation_waiting, + ), + ( + "celld_node_capacity_waiting_cells", + "Cold routes queued behind the resident-cell ceiling.", + metrics.capacity_waiting, + ), + ( + "celld_node_output_gate_pending_writes", + "HTTP and WebSocket writes waiting for durable replication.", + metrics.output_gate_pending, + ), + ] { + help_and_type(&mut output, name, help, "gauge"); + let _ = writeln!(output, "{name} {value}"); + } + + help_and_type( + &mut output, + "celld_node_phase_cells", + "Cells in each stable decision-core phase.", + "gauge", + ); + for phase in STABLE_PHASE_NAMES { + let count = metrics + .phases + .iter() + .find_map(|(candidate, count)| (*candidate == *phase).then_some(count)) + .copied() + .unwrap_or_default(); + let _ = writeln!( + output, + "celld_node_phase_cells{{phase=\"{}\"}} {count}", + label_value(phase), + ); + } + + if let Some(limit) = metrics.resident_limit { + help_and_type( + &mut output, + "celld_node_resident_limit_cells", + "Configured hard resident-cell admission limit.", + "gauge", + ); + let _ = writeln!(output, "celld_node_resident_limit_cells {limit}"); + help_and_type( + &mut output, + "celld_node_resident_headroom_cells", + "Remaining resident-cell admission headroom.", + "gauge", + ); + let _ = writeln!( + output, + "celld_node_resident_headroom_cells {}", + limit.saturating_sub(metrics.occupied), + ); + } + + help_and_type( + &mut output, + "celld_node_shedding", + "Whether the node is currently shedding or refusing ownership.", + "gauge", + ); + let _ = writeln!( + output, + "celld_node_shedding {}", + u8::from(metrics.shed_reason.is_some()), + ); + help_and_type( + &mut output, + "celld_node_shedding_reason", + "Current bounded decision-core shedding reason.", + "gauge", + ); + for reason in [SHED_MEMORY, SHED_RSS_HARD] { + let _ = writeln!( + output, + "celld_node_shedding_reason{{reason=\"{}\"}} {}", + label_value(reason), + u8::from(metrics.shed_reason == Some(reason)), + ); + } + + for (name, help, value) in [ + ( + "celld_process_resident_memory_bytes", + "Resident set size sampled from the process allocator.", + metrics.rss_bytes, + ), + ( + "celld_process_in_use_memory_bytes", + "Bytes the allocator reports as currently in use.", + metrics.in_use_bytes, + ), + ] { + help_and_type(&mut output, name, help, "gauge"); + let _ = writeln!(output, "{name} {value}"); + } + + for (name, help, value) in [ + ( + "celld_node_publishes_total", + "Runtime publications completed by this process.", + metrics.publishes, + ), + ( + "celld_node_stops_total", + "Runtime stops completed by this process.", + metrics.stops, + ), + ( + "celld_node_ownership_acquired_total", + "Ownership acquisitions accepted by the decision core.", + metrics.activity.acquired, + ), + ( + "celld_node_proxied_requests_total", + "Requests routed to a remote owner.", + metrics.activity.proxied, + ), + ( + "celld_node_expired_owner_leases_total", + "Expired owner leases observed by the decision core.", + metrics.activity.expired_owner_leases, + ), + ( + "celld_node_restores_total", + "Cell restores completed by the decision core.", + metrics.activity.restored, + ), + ( + "celld_node_authority_epochs_advanced_total", + "Ownership epochs advanced by the decision core.", + metrics.activity.advanced_epochs, + ), + ] { + help_and_type(&mut output, name, help, "counter"); + let _ = writeln!(output, "{name} {value}"); + } + + output +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn renders_bounded_node_metrics_without_identifiers() { + let output = render(&NodeMetrics { + runtime_version: "0.2.1-coderabbit.test", + region: "us-central1\"\\\n", + ownership: "bucket", + serving: true, + occupied: 7, + resident_limit: Some(10), + evicting: 1, + restoring: 2, + activating: 1, + activation_waiting: 1, + capacity_waiting: 3, + phases: &[("resident", 5), ("restoring", 2)], + shed_reason: Some(SHED_MEMORY), + rss_bytes: 1024, + in_use_bytes: 768, + output_gate_pending: 4, + publishes: 8, + stops: 2, + activity: ActivitySnapshot { + acquired: 11, + proxied: 12, + expired_owner_leases: 13, + restored: 14, + advanced_epochs: 15, + }, + }); + + assert!(output.contains("celld_build_info{version=\"0.2.1-coderabbit.test\",region=\"us-central1\\\"\\\\\\n\"} 1")); + assert!(output.contains("celld_node_resident_headroom_cells 3")); + assert!(output.contains("celld_node_phase_cells{phase=\"resident\"} 5")); + assert!(output.contains("celld_node_phase_cells{phase=\"inactive\"} 0")); + assert!(output.contains("celld_node_shedding_reason{reason=\"memory\"} 1")); + assert!(output.contains("celld_node_shedding_reason{reason=\"rss-hard\"} 0")); + assert!(output.contains("celld_node_output_gate_pending_writes 4")); + assert!(output.contains("celld_node_authority_epochs_advanced_total 15")); + assert!(!output.contains("cell_id")); + assert!(!output.contains("request_id")); + assert!(!output.contains("s3://")); + assert!(!output.contains("bucket_name")); + } + + #[test] + fn omits_unbounded_limit_and_zeroes_bounded_series_when_inactive() { + let output = render(&NodeMetrics { + runtime_version: "test", + region: "local", + ownership: "memory", + serving: true, + occupied: 0, + resident_limit: None, + evicting: 0, + restoring: 0, + activating: 0, + activation_waiting: 0, + capacity_waiting: 0, + phases: &[], + shed_reason: None, + rss_bytes: 0, + in_use_bytes: 0, + output_gate_pending: 0, + publishes: 0, + stops: 0, + activity: ActivitySnapshot::default(), + }); + + assert!(!output.contains("celld_node_resident_limit_cells")); + assert!(!output.contains("celld_node_resident_headroom_cells")); + assert!(output.contains("celld_node_phase_cells{phase=\"resident\"} 0")); + assert!(output.contains("celld_node_shedding_reason{reason=\"memory\"} 0")); + assert!(output.contains("celld_node_shedding_reason{reason=\"rss-hard\"} 0")); + assert!(output.contains("celld_node_shedding 0")); + } +} diff --git a/crates/logic/lib.rs b/crates/logic/lib.rs index 7391c1f9f..263a262e6 100644 --- a/crates/logic/lib.rs +++ b/crates/logic/lib.rs @@ -156,8 +156,29 @@ pub enum Phase { Fenced, } -/// The reported name of a phase. Stable across internal renames, because -/// `/state` and `celld diagnose` publish it. +/// The complete reported phase vocabulary. Names stay stable across internal +/// renames because `/state`, `celld diagnose`, and operator metrics publish +/// them. +pub const STABLE_PHASE_NAMES: &[&str] = &[ + "inactive", + "waiting_activation", + "reading_owner", + "reading_node_lease", + "reading_capacity", + "waiting_capacity", + "acquiring", + "reconciling_acquire", + "restoring", + "starting", + "publishing", + "ensuring_durability", + "cleaning", + "dormant", + "resident", + "remote", + "fenced", +]; + fn phase_name(phase: &Phase) -> &'static str { match phase { Phase::Inactive => "inactive", @@ -705,6 +726,17 @@ impl State { counts.into_iter().collect() } + /// Cumulative lifecycle decisions for bounded operator telemetry. + pub fn activity_snapshot(&self) -> ActivitySnapshot { + self.activity + } + + /// The hard resident-cell admission ceiling. `usize::MAX` means the + /// operator left the ceiling unbounded. + pub fn max_resident(&self) -> usize { + self.config.max_resident + } + /// Cold routes that have not finished: cells that hold an activation /// permit plus cells queued behind the activation ceiling. A capacity /// waiter already holds a permit, so this counts every cell once. A diff --git a/docs/security.md b/docs/security.md index fe365d352..4770bab94 100644 --- a/docs/security.md +++ b/docs/security.md @@ -73,6 +73,9 @@ alpha interface, so a release can change its paths or response formats. - `/state` reports the current occupancy, eviction, and restoration values. It remains available while a graceful shutdown drains existing work. +- `/metrics` reports the same node lifecycle state as bounded Prometheus text. + It includes no cell, request, tenant, or bucket identifiers and remains + available during graceful drain. - `/cell/NAME` resolves or activates a cell for an operator check. - `/evict/NAME` evicts a resident cell. - `/do/NAME` sends a direct Durable Object request. diff --git a/docs/telemetry.md b/docs/telemetry.md index 41a2766df..bd522dd2c 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -53,10 +53,19 @@ The sampler decides at the start of a request. An unsampled request records nothing and costs almost nothing. Under load, telemetry sheds before requests do, and celld counts what it sheds. -celld records no metrics yet. This is a known gap, not a silent one: -the spans carry the durations and the queue waits, so many questions -a metric answers have an answer in the traces, and a metrics signal -can come later without a change to the trace schema. +The private operator listener exposes bounded node lifecycle metrics at +`/metrics` in Prometheus text format. These gauges and counters cover serving, +ownership, residency, activation, eviction, phase census, capacity headroom, +memory, output-gate backlog, and cumulative ownership decisions. They contain +no cell, request, tenant, or bucket identifiers. Keep the internal listener +private and let a node-local collector scrape it. Every bounded phase and +shedding-reason label is emitted on every scrape, including explicit zeroes, +so a disappeared condition cannot look active through a stale series. + +Request latency, queue wait, bucket operation, and handler outcome remain trace +signals rather than metrics. The spans carry those durations and outcomes, so +derive their distributions in the OpenTelemetry backend without introducing a +second sampling decision. ## Query the bucket with DuckDB From 2235729815632c92281505add70d974788d7819d Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 09:13:11 -0700 Subject: [PATCH 2/6] feat: add fenced cell archive recovery --- .github/workflows/ci.yml | 4 +- crates/celld/cell_archive.rs | 648 ++++++++++++++++++++++++++++++++ crates/celld/lib.rs | 1 + crates/celld/ltx_repl.rs | 103 ++++- crates/celld/main.rs | 1 + crates/celld/main/cli.rs | 4 + crates/celld/ownership_store.rs | 14 + crates/celld/replication.rs | 18 +- crates/ltx/src/replica.rs | 3 + scripts/cell-archive-minio.sh | 171 +++++++++ 10 files changed, 961 insertions(+), 6 deletions(-) create mode 100644 crates/celld/cell_archive.rs create mode 100755 scripts/cell-archive-minio.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f736e5616..1a7de72cb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,4 +25,6 @@ jobs: - name: Build the tested production image run: | version=$(sed -n 's/^version = "\(.*\)"$/\1/p' crates/celld/Cargo.toml | head -1) - docker build --build-arg CELLD_COMMIT="$GITHUB_SHA" --build-arg CELLD_VERSION="$version" . + docker build --tag celld-ci --build-arg CELLD_COMMIT="$GITHUB_SHA" --build-arg CELLD_VERSION="$version" . + - name: Test cell disaster recovery against MinIO + run: ./scripts/cell-archive-minio.sh diff --git a/crates/celld/cell_archive.rs b/crates/celld/cell_archive.rs new file mode 100644 index 000000000..e4b176834 --- /dev/null +++ b/crates/celld/cell_archive.rs @@ -0,0 +1,648 @@ +// Copyright 2026 Deno Land Inc. Apache-2.0 license. + +//! Supported cell export and offline import commands. +//! +//! Export is a point-in-time view of the newest durable LTX epoch and does not +//! claim the cell. Import creates a brand-new lineage only. Its staging marker +//! makes upgraded nodes fail closed while LTX is being built; the offline gate +//! is still mandatory because older nodes do not understand that marker. + +use crate::bucket::Bucket; +use crate::fleet; +use crate::ltx_repl::LtxRepl; +use crate::ownership_store::{now_ms, BucketOwnership, NodeLeaseWire}; +use anyhow::{bail, Context}; +use celld_logic::CasOutcome; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::path::{Path, PathBuf}; + +const ARCHIVE_VERSION: u32 = 1; +const IMPORT_EPOCH: u64 = 1; +const IMPORT_ATTEMPT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30 * 60); +const IMPORT_ATTEMPT_LEASE_MS: u64 = 31 * 60 * 1_000; + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +enum ImportPhase { + Staging, + Ready, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +struct ImportMarker { + version: u32, + phase: ImportPhase, + source_sha256: String, + attempt_id: String, + #[serde(default)] + attempt_expires_ms: Option, + #[serde(default)] + durable_txid: Option, +} + +#[derive(Debug, Serialize)] +struct ExportManifest<'a> { + version: u32, + cell: &'a str, + source_epoch: u64, + source_txid: u64, + database_sha256: String, +} + +#[derive(Debug)] +struct StorageOptions { + bucket: String, + endpoint: Option, + region: String, +} + +#[derive(Debug)] +enum Command { + Export { + cell: String, + output: PathBuf, + storage: StorageOptions, + }, + Import { + cell: String, + input: PathBuf, + storage: StorageOptions, + offline: bool, + resume: bool, + }, + Help, +} + +pub async fn ensure_import_ready(bucket: &Bucket, cell: &str) -> anyhow::Result<()> { + let key = format!("cells/{cell}/import.json"); + let Some((bytes, _)) = bucket.get(&key).await? else { + return Ok(()); + }; + let marker: ImportMarker = serde_json::from_slice(&bytes) + .with_context(|| format!("decode import marker for {cell}"))?; + validate_marker(&marker, cell)?; + anyhow::ensure!( + matches!(marker.phase, ImportPhase::Ready), + "cell {cell} has an incomplete offline import; resume the import before activation" + ); + Ok(()) +} + +fn validate_marker(marker: &ImportMarker, cell: &str) -> anyhow::Result<()> { + anyhow::ensure!( + marker.version == ARCHIVE_VERSION, + "cell {cell} has unsupported import marker version {}", + marker.version + ); + anyhow::ensure!( + marker.source_sha256.len() == 64 + && marker + .source_sha256 + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')), + "cell {cell} import marker has an invalid source SHA-256" + ); + anyhow::ensure!( + marker.attempt_id.len() == 32 + && marker + .attempt_id + .bytes() + .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f')), + "cell {cell} import marker has an invalid attempt ID" + ); + match marker.phase { + ImportPhase::Staging => { + anyhow::ensure!( + marker.durable_txid.is_none() && marker.attempt_expires_ms.is_some(), + "cell {cell} staging import marker has invalid attempt state" + ); + } + ImportPhase::Ready => { + anyhow::ensure!( + marker.durable_txid.is_some_and(|txid| txid > 0) + && marker.attempt_expires_ms.is_none(), + "cell {cell} ready import marker has invalid durable state" + ); + } + } + Ok(()) +} + +pub async fn run(arguments: Vec) -> anyhow::Result<()> { + match parse(arguments)? { + Command::Help => { + print_help(); + Ok(()) + } + Command::Export { + cell, + output, + storage, + } => export(&cell, &output, &storage).await, + Command::Import { + cell, + input, + storage, + offline, + resume, + } => import(&cell, &input, &storage, offline, resume).await, + } +} + +fn parse(arguments: Vec) -> anyhow::Result { + if arguments.is_empty() || matches!(arguments[0].as_str(), "help" | "-h" | "--help") { + return Ok(Command::Help); + } + let operation = arguments[0].clone(); + if !matches!(operation.as_str(), "export" | "import") { + bail!("unknown cell command {operation:?}; run `celld cell --help` for usage"); + } + let cell = arguments + .get(1) + .filter(|value| !value.starts_with('-')) + .cloned() + .context("cell export/import requires CELL")?; + anyhow::ensure!( + celld_logic::cell::valid_cell_scope(&cell), + "invalid cell scope {cell:?}" + ); + + let mut bucket = None; + let mut endpoint = None; + let mut region = None; + let mut input = None; + let mut output = None; + let mut offline = false; + let mut resume = false; + let mut args = arguments.into_iter().skip(2); + while let Some(argument) = args.next() { + match argument.as_str() { + "--bucket" => bucket = Some(next_value(&mut args, "--bucket")?), + "--endpoint" => endpoint = Some(next_value(&mut args, "--endpoint")?), + "--region" => region = Some(next_value(&mut args, "--region")?), + "--input" => input = Some(PathBuf::from(next_value(&mut args, "--input")?)), + "--output" => output = Some(PathBuf::from(next_value(&mut args, "--output")?)), + "--offline" => offline = true, + "--resume" => resume = true, + other => bail!("unknown cell {operation} option {other:?}"), + } + } + let env = |name: &str| { + std::env::var(name) + .ok() + .filter(|value| !value.trim().is_empty()) + }; + let storage = StorageOptions { + bucket: bucket + .or_else(|| env("CELLD_BUCKET")) + .context("cell export/import requires --bucket or CELLD_BUCKET")?, + endpoint: endpoint.or_else(|| env("S3_ENDPOINT")), + region: region + .or_else(|| env("AWS_REGION")) + .or_else(|| env("AWS_DEFAULT_REGION")) + .unwrap_or_else(|| "us-east-1".to_string()), + }; + match operation.as_str() { + "export" => Ok(Command::Export { + cell, + output: output.context("cell export requires --output DATABASE")?, + storage, + }), + "import" => Ok(Command::Import { + cell, + input: input.context("cell import requires --input DATABASE")?, + storage, + offline, + resume, + }), + _ => unreachable!(), + } +} + +fn next_value(args: &mut impl Iterator, option: &str) -> anyhow::Result { + args.next() + .with_context(|| format!("{option} requires a value")) +} + +fn print_help() { + println!( + r#"celld cell — portable cell disaster recovery + +USAGE: + celld cell export CELL --output DATABASE --bucket [s3://|gs://]NAME[/PREFIX] + celld cell import CELL --input DATABASE --bucket [s3://|gs://]NAME[/PREFIX] --offline [--resume] + +Export reads the newest durable LTX epoch without claiming the cell and writes +DATABASE plus DATABASE.manifest.json. Import creates only a brand-new cell. It +requires every node from an older release to be stopped, rejects live node +leases, and is crash-resumable with the same input plus --resume after the +staging attempt's bounded lease expires."# + ); +} + +fn open_bucket(storage: &StorageOptions) -> anyhow::Result { + fleet::bucket_client( + &storage.bucket, + storage.endpoint.as_deref(), + &storage.region, + ) +} + +fn start_replication( + bucket: &Bucket, + storage: &StorageOptions, + watch: &Path, +) -> anyhow::Result { + LtxRepl::start( + watch, + bucket.backend(), + bucket.name.clone(), + bucket.prefix.clone(), + storage.endpoint.clone(), + storage.region.clone(), + None, + ) +} + +async fn export(cell: &str, output: &Path, storage: &StorageOptions) -> anyhow::Result<()> { + anyhow::ensure!( + !output.exists(), + "refusing to overwrite {}", + output.display() + ); + let manifest_path = manifest_path(output); + anyhow::ensure!( + !manifest_path.exists(), + "refusing to overwrite {}", + manifest_path.display() + ); + let bucket = open_bucket(storage)?; + fleet::validate_bucket(&bucket).await?; + let work = tempfile::tempdir()?; + let replication = start_replication(&bucket, storage, work.path())?; + let snapshot = replication + .restore_snapshot(cell) + .await? + .with_context(|| format!("cell {cell} has no durable snapshot"))?; + crate::replication::sqlite_snapshot(snapshot.path(), output)?; + validate_sqlite(output)?; + let manifest = ExportManifest { + version: ARCHIVE_VERSION, + cell, + source_epoch: snapshot.epoch, + source_txid: snapshot + .txid + .context("durable snapshot did not report its transaction")?, + database_sha256: sha256_file(output)?, + }; + write_private_new(&manifest_path, &serde_json::to_vec_pretty(&manifest)?)?; + println!( + "exported {cell} epoch {} to {}", + snapshot.epoch, + output.display() + ); + println!("manifest {}", manifest_path.display()); + Ok(()) +} + +async fn import( + cell: &str, + input: &Path, + storage: &StorageOptions, + offline: bool, + resume: bool, +) -> anyhow::Result<()> { + anyhow::ensure!(offline, "cell import requires --offline acknowledgement"); + anyhow::ensure!( + input.is_file(), + "SQLite archive does not exist: {}", + input.display() + ); + // Hash and replicate one private backup, not the caller's main database + // file. The backup API includes its committed WAL and prevents a writer + // from changing the bytes between the import identity and LTX capture. + let normalized_dir = tempfile::tempdir()?; + let normalized = normalized_dir.path().join("database.sqlite"); + crate::replication::sqlite_snapshot(input, &normalized) + .context("create consistent import archive")?; + validate_sqlite(&normalized)?; + let source_sha256 = sha256_file(&normalized)?; + let bucket = open_bucket(storage)?; + fleet::validate_bucket(&bucket).await?; + ensure_no_live_nodes(&bucket).await?; + + let marker_key = format!("cells/{cell}/import.json"); + let owner_key = format!("cells/{cell}/own.json"); + let existing_marker = bucket.get(&marker_key).await?; + let (claimed_marker, marker_token) = match existing_marker { + Some((bytes, token)) => { + let marker: ImportMarker = serde_json::from_slice(&bytes)?; + validate_marker(&marker, cell)?; + anyhow::ensure!( + marker.version == ARCHIVE_VERSION && marker.source_sha256 == source_sha256, + "cell {cell} has an import from a different archive; refusing to replace it" + ); + if matches!(marker.phase, ImportPhase::Ready) { + println!("cell {cell} import is already ready"); + return Ok(()); + } + anyhow::ensure!( + resume, + "cell {cell} has an incomplete import; retry with --resume after its attempt lease expires" + ); + anyhow::ensure!( + marker + .attempt_expires_ms + .is_some_and(|expires| expires <= now_ms()), + "cell {cell} import attempt {} is still active", + marker.attempt_id + ); + let claimed = staging_marker(source_sha256.clone()); + let token = bucket + .put_cas(&marker_key, serde_json::to_vec(&claimed)?, Some(&token)) + .await? + .with_context(|| format!("another process resumed cell {cell} first"))?; + (claimed, token) + } + None => { + anyhow::ensure!(!resume, "cell {cell} has no incomplete import to resume"); + anyhow::ensure!( + bucket.get(&owner_key).await?.is_none(), + "cell {cell} already has an owner lineage" + ); + anyhow::ensure!( + !bucket.list_any(&format!("cells/{cell}/ltx/")).await?, + "cell {cell} already has replicated history" + ); + let marker = staging_marker(source_sha256.clone()); + let token = bucket + .put_cas(&marker_key, serde_json::to_vec(&marker)?, None) + .await? + .with_context(|| format!("another import raced for cell {cell}"))?; + (marker, token) + } + }; + + ensure_no_live_nodes(&bucket).await?; + let owner = BucketOwnership::new( + bucket.clone(), + bucket.clone(), + "cell-import".into(), + String::new(), + ); + if let Some((bytes, _)) = bucket.get(&owner_key).await? { + #[derive(Deserialize)] + struct Owner { + node: String, + epoch: u64, + } + let owner: Owner = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + owner.node.is_empty() && owner.epoch == IMPORT_EPOCH, + "cell {cell} became owned during import" + ); + } else { + let work = tempfile::tempdir()?; + let replication = start_replication(&bucket, storage, work.path())?; + let txid = tokio::time::timeout( + IMPORT_ATTEMPT_TIMEOUT, + replication.seed_import_epoch(cell, IMPORT_EPOCH, &normalized), + ) + .await + .with_context(|| format!("cell {cell} import exceeded its 30-minute attempt limit"))??; + ensure_no_live_nodes(&bucket).await?; + anyhow::ensure!( + claimed_marker + .attempt_expires_ms + .is_some_and(|expires| expires > now_ms()), + "cell {cell} import attempt lease expired before publication" + ); + let publication_token = bucket + .put_cas( + &marker_key, + serde_json::to_vec(&claimed_marker)?, + Some(&marker_token), + ) + .await? + .with_context(|| format!("cell {cell} import attempt lost its staging claim"))?; + anyhow::ensure!( + owner.create_import_owner(cell, IMPORT_EPOCH).await? == CasOutcome::Applied, + "cell {cell} acquired an owner lineage while import was staging" + ); + let ready = ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Ready, + source_sha256, + attempt_id: claimed_marker.attempt_id, + attempt_expires_ms: None, + durable_txid: Some(txid), + }; + anyhow::ensure!( + bucket + .put_cas( + &marker_key, + serde_json::to_vec(&ready)?, + Some(&publication_token), + ) + .await? + .is_some(), + "import marker changed while publishing cell {cell}" + ); + println!("imported {cell} at epoch {IMPORT_EPOCH}, durable txid {txid}"); + return Ok(()); + } + + let txid = { + let work = tempfile::tempdir()?; + let replication = start_replication(&bucket, storage, work.path())?; + replication.epoch_max_txid(cell, IMPORT_EPOCH).await? + }; + anyhow::ensure!( + txid.is_some(), + "cell {cell} owner exists but imported LTX cannot be restored" + ); + let ready = ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Ready, + source_sha256, + attempt_id: claimed_marker.attempt_id, + attempt_expires_ms: None, + durable_txid: txid, + }; + anyhow::ensure!( + bucket + .put_cas( + &marker_key, + serde_json::to_vec(&ready)?, + Some(&marker_token), + ) + .await? + .is_some(), + "import marker changed while resuming cell {cell}" + ); + println!("resumed import for {cell}"); + Ok(()) +} + +fn staging_marker(source_sha256: String) -> ImportMarker { + ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Staging, + source_sha256, + attempt_id: format!("{:032x}", rand::random::()), + attempt_expires_ms: Some(now_ms().saturating_add(IMPORT_ATTEMPT_LEASE_MS)), + durable_txid: None, + } +} + +async fn ensure_no_live_nodes(bucket: &Bucket) -> anyhow::Result<()> { + let mut live = Vec::new(); + for object in bucket.list("nodes/").await? { + let key = object.location.as_ref(); + let Some((bytes, _)) = bucket.get(key).await? else { + continue; + }; + let lease: NodeLeaseWire = + serde_json::from_slice(&bytes).with_context(|| format!("decode node lease {key}"))?; + if lease.expires_ms > now_ms() { + live.push(lease.node); + } + } + live.sort(); + anyhow::ensure!( + live.is_empty(), + "offline import refused: live celld node lease(s): {}", + live.join(", ") + ); + Ok(()) +} + +fn validate_sqlite(path: &Path) -> anyhow::Result<()> { + anyhow::ensure!( + path.is_file(), + "SQLite archive does not exist: {}", + path.display() + ); + let connection = + rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?; + let integrity: String = connection.query_row("PRAGMA integrity_check", [], |row| row.get(0))?; + anyhow::ensure!( + integrity == "ok", + "SQLite integrity check failed: {integrity}" + ); + Ok(()) +} + +fn sha256_file(path: &Path) -> anyhow::Result { + let bytes = std::fs::read(path)?; + Ok(format!("{:x}", Sha256::digest(bytes))) +} + +fn manifest_path(database: &Path) -> PathBuf { + let mut path = database.as_os_str().to_os_string(); + path.push(".manifest.json"); + PathBuf::from(path) +} + +fn write_private_new(path: &Path, bytes: &[u8]) -> anyhow::Result<()> { + use std::io::Write; + let mut options = std::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options.open(path)?; + file.write_all(bytes)?; + file.sync_all()?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_import_only_with_explicit_paths() { + let command = parse(vec![ + "import".into(), + "Org:test".into(), + "--input".into(), + "archive.sqlite".into(), + "--bucket".into(), + "bucket/fleet".into(), + "--offline".into(), + "--resume".into(), + ]) + .unwrap(); + assert!(matches!( + command, + Command::Import { + offline: true, + resume: true, + .. + } + )); + } + + #[test] + fn rejects_invalid_cell_before_storage_access() { + let error = parse(vec![ + "export".into(), + "../escape".into(), + "--output".into(), + "archive.sqlite".into(), + "--bucket".into(), + "bucket".into(), + ]) + .unwrap_err(); + assert!(error.to_string().contains("invalid cell scope")); + } + + #[test] + fn validates_and_hashes_sqlite() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("archive.sqlite"); + let connection = rusqlite::Connection::open(&path).unwrap(); + connection + .execute_batch("CREATE TABLE values_ (value TEXT); INSERT INTO values_ VALUES ('ok');") + .unwrap(); + drop(connection); + validate_sqlite(&path).unwrap(); + assert_eq!(sha256_file(&path).unwrap().len(), 64); + } + + #[test] + fn import_markers_fail_closed_until_a_durable_transaction_is_ready() { + let hash = "a".repeat(64); + let staging = ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Staging, + source_sha256: hash.clone(), + attempt_id: "1".repeat(32), + attempt_expires_ms: Some(1), + durable_txid: None, + }; + validate_marker(&staging, "Knowledge:test").unwrap(); + let incomplete = ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Ready, + source_sha256: hash.clone(), + attempt_id: "1".repeat(32), + attempt_expires_ms: None, + durable_txid: None, + }; + assert!(validate_marker(&incomplete, "Knowledge:test").is_err()); + let ready = ImportMarker { + version: ARCHIVE_VERSION, + phase: ImportPhase::Ready, + source_sha256: hash, + attempt_id: "1".repeat(32), + attempt_expires_ms: None, + durable_txid: Some(1), + }; + validate_marker(&ready, "Knowledge:test").unwrap(); + } +} diff --git a/crates/celld/lib.rs b/crates/celld/lib.rs index 767163167..51e37b846 100644 --- a/crates/celld/lib.rs +++ b/crates/celld/lib.rs @@ -9,6 +9,7 @@ pub mod assets; pub mod asyncrt; pub mod bucket; +pub mod cell_archive; pub mod control_plane; pub mod dead_node_gc; pub mod deploy; diff --git a/crates/celld/ltx_repl.rs b/crates/celld/ltx_repl.rs index 8f6ef8512..540fbf976 100644 --- a/crates/celld/ltx_repl.rs +++ b/crates/celld/ltx_repl.rs @@ -27,6 +27,7 @@ use std::time::Duration; use std::time::Instant; use anyhow::anyhow; +use anyhow::Context; use celld_ltx::object_store::ObjectStore; use celld_ltx::replica; use celld_ltx::replica_compactor::ReplicaCompactor; @@ -632,7 +633,7 @@ impl LtxRepl { std::fs::create_dir_all(&directory)?; let path = directory.join("db.sqlite"); sqlite_snapshot(&source, &path)?; - Ok(Some(RestoredSnapshot::new(epoch, path, directory))) + Ok(Some(RestoredSnapshot::new(epoch, None, path, directory))) } /// Restore the newest durable replica into a private snapshot without @@ -645,7 +646,7 @@ impl LtxRepl { let _ = std::fs::remove_dir_all(&directory); std::fs::create_dir_all(&directory)?; let path = directory.join("db.sqlite"); - replica::restore_with_download_slots( + let stats = replica::restore_with_download_slots( &self.client_for(cell, epoch), &path, TXID(0), @@ -653,7 +654,103 @@ impl LtxRepl { ) .await .map_err(|error| anyhow!("restore snapshot {cell} e{epoch}: {error}"))?; - Ok(Some(RestoredSnapshot::new(epoch, path, directory))) + Ok(Some(RestoredSnapshot::new( + epoch, + Some(stats.max_txid), + path, + directory, + ))) + } + + /// Highest durable transaction in one epoch, or `None` when it has no + /// restorable LTX. Used to finish a crash-interrupted import marker from + /// the lineage that was already validated before its owner CAS. + pub(crate) async fn epoch_max_txid( + &self, + cell: &str, + epoch: u64, + ) -> anyhow::Result> { + let plan = replica::calc_restore_plan(&self.client_for(cell, epoch), TXID(0)) + .await + .with_context(|| format!("plan durable position for {cell} e{epoch}"))?; + Ok(plan.iter().map(|info| info.max_txid.0).max()) + } + + /// Replace the private epoch used by an offline import, capture the input + /// as LTX, and prove that the uploaded lineage restores cleanly. + /// + /// Authority policy lives in `cell_archive`: this primitive is called only + /// while a CAS-created staging marker blocks activation and before an owner + /// record exists. Clearing the epoch makes a retry after a process crash + /// deterministic instead of appending to an unknown partial upload. + pub(crate) async fn seed_import_epoch( + &self, + cell: &str, + epoch: u64, + source: &Path, + ) -> anyhow::Result { + use celld_ltx::object_store::path::Path as ObjPath; + + let remote_prefix = format!("{}cells/{cell}/ltx/e{epoch}", self.prefix); + let remote = ObjPath::from(remote_prefix.clone()); + let mut listed = self.store.list(Some(&remote)); + while let Some(object) = futures_util::StreamExt::next(&mut listed).await { + let object = object.context("list partial import epoch")?; + self.store + .delete(&object.location) + .await + .with_context(|| format!("clear partial import object {}", object.location))?; + } + + let directory = self.watch.join(format!(".import-{cell}-e{epoch}")); + let _ = std::fs::remove_dir_all(&directory); + std::fs::create_dir_all(&directory)?; + let path = directory.join("db.sqlite"); + sqlite_snapshot(source, &path).context("create consistent import snapshot")?; + + let client = self.client_for(cell, epoch); + let path_for_capture = path.clone(); + let mut replica = tokio::task::spawn_blocking(move || -> anyhow::Result<_> { + let mut db = Db::open(&path_for_capture).context("open import snapshot for LTX")?; + db.sync().context("capture import snapshot as LTX")?; + Ok(Replica::new(db, client)) + }) + .await??; + replica.sync().await.context("upload import snapshot LTX")?; + let txid = replica.pos().txid.0; + anyhow::ensure!(txid > 0, "import produced no durable LTX transaction"); + drop(replica); + + let restored = directory.join("roundtrip.sqlite"); + replica::restore_with_download_slots( + &self.client_for(cell, epoch), + &restored, + TXID(0), + self.restore_slots.clone(), + ) + .await + .context("round-trip imported LTX")?; + let connection = rusqlite::Connection::open_with_flags( + &restored, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + )?; + let integrity: String = + connection.query_row("PRAGMA integrity_check", [], |row| row.get(0))?; + anyhow::ensure!( + integrity == "ok", + "import round-trip integrity check failed: {integrity}" + ); + drop(connection); + let expected = directory.join("expected.sqlite"); + let actual = directory.join("actual.sqlite"); + sqlite_snapshot(&path, &expected).context("normalize captured import for comparison")?; + sqlite_snapshot(&restored, &actual).context("normalize restored import for comparison")?; + anyhow::ensure!( + std::fs::read(&expected)? == std::fs::read(&actual)?, + "import LTX round trip does not match the staged SQLite database" + ); + let _ = std::fs::remove_dir_all(&directory); + Ok(txid) } pub fn prune_local_cache(&self, max_bytes: u64) -> (usize, usize, u64) { diff --git a/crates/celld/main.rs b/crates/celld/main.rs index 6c8519d2e..78cdad266 100644 --- a/crates/celld/main.rs +++ b/crates/celld/main.rs @@ -4167,6 +4167,7 @@ async fn async_main(telemetry_config: Option) -> anyho celld::memory::tune_allocator(); let mut settings = match action_from_process()? { Action::Deploy(arguments) => return fleet::run_deploy(arguments).await, + Action::Cell(arguments) => return celld::cell_archive::run(arguments).await, Action::Connect(arguments) => { return celld::control_plane::handle_connect_command(arguments).await } diff --git a/crates/celld/main/cli.rs b/crates/celld/main/cli.rs index e6df71acb..bf7eccac2 100644 --- a/crates/celld/main/cli.rs +++ b/crates/celld/main/cli.rs @@ -35,6 +35,7 @@ pub(crate) enum Action { read_only: bool, }, Deploy(Vec), + Cell(Vec), Connect(Vec), Credentials(Vec), Token(Vec), @@ -49,6 +50,7 @@ pub(crate) fn action_from_process() -> anyhow::Result { let arguments = arguments[1..].to_vec(); match action { "deploy" => return Ok(Action::Deploy(arguments)), + "cell" => return Ok(Action::Cell(arguments)), "connect" => return Ok(Action::Connect(arguments)), "credentials" => return Ok(Action::Credentials(arguments)), "token" => return Ok(Action::Token(arguments)), @@ -224,6 +226,8 @@ pub(crate) fn print_help() { USAGE: celld --bucket [s3://|gs://]NAME[/PREFIX] [OPTIONS] celld deploy [PROJECT] --bucket [s3://|gs://]NAME[/PREFIX] [OPTIONS] + celld cell export CELL --output DATABASE --bucket [s3://|gs://]NAME[/PREFIX] + celld cell import CELL --input DATABASE --bucket [s3://|gs://]NAME[/PREFIX] --offline [--resume] celld diagnose --bucket [s3://|gs://]NAME[/PREFIX] [OPTIONS] [--peer NODE_ID]... Production install: celld --bucket s3://NAME [OPTIONS] diff --git a/crates/celld/ownership_store.rs b/crates/celld/ownership_store.rs index b0199cb97..0437110e4 100644 --- a/crates/celld/ownership_store.rs +++ b/crates/celld/ownership_store.rs @@ -184,6 +184,7 @@ impl BucketOwnership { } pub async fn read_owner(&self, cell: &str) -> anyhow::Result> { + crate::cell_archive::ensure_import_ready(&self.bucket, cell).await?; let key = format!("cells/{cell}/own.json"); let Some((owner, etag)) = load_json::(&self.bucket, &key).await? else { return Ok(None); @@ -195,6 +196,19 @@ impl BucketOwnership { })) } + /// Publish the initial unowned record for a fully staged import. + /// + /// This is deliberately absent-only. Import must never replace, rewind, + /// or join an existing authority lineage. + pub async fn create_import_owner(&self, cell: &str, epoch: u64) -> anyhow::Result { + let key = format!("cells/{cell}/own.json"); + let body = serde_json::to_vec(&OwnerWire { node: "", epoch })?; + match self.bucket.put_cas(&key, body, None).await? { + Some(_) => Ok(CasOutcome::Applied), + None => Ok(CasOutcome::Rejected), + } + } + pub async fn read_node_lease(&self, owner: &str) -> anyhow::Result> { load_node_lease(&self.bucket, owner).await } diff --git a/crates/celld/replication.rs b/crates/celld/replication.rs index 188dfa61c..289670107 100644 --- a/crates/celld/replication.rs +++ b/crates/celld/replication.rs @@ -22,6 +22,7 @@ pub enum SyncWait { pub struct RestoredSnapshot { pub epoch: u64, + pub txid: Option, path: PathBuf, directory: PathBuf, } @@ -33,9 +34,10 @@ impl RestoredSnapshot { /// Construct a snapshot whose `directory` is removed on drop, handing the /// caller an inspection copy with RAII cleanup. - pub(crate) fn new(epoch: u64, path: PathBuf, directory: PathBuf) -> Self { + pub(crate) fn new(epoch: u64, txid: Option, path: PathBuf, directory: PathBuf) -> Self { Self { epoch, + txid, path, directory, } @@ -138,12 +140,24 @@ pub(crate) fn sqlite_snapshot( source: &std::path::Path, destination: &std::path::Path, ) -> anyhow::Result<()> { + let mut options = std::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + drop(options.open(destination)?); + let result = (|| -> anyhow::Result<()> { let source = Connection::open_with_flags(source, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?; let mut destination = Connection::open(destination)?; let backup = rusqlite::backup::Backup::new(&source, &mut destination)?; backup.run_to_completion(64, Duration::from_millis(5), None)?; + Ok(()) + })(); + if result.is_err() { + let _ = std::fs::remove_file(destination); } - Ok(()) + result } diff --git a/crates/ltx/src/replica.rs b/crates/ltx/src/replica.rs index 63218934a..c30bf8163 100644 --- a/crates/ltx/src/replica.rs +++ b/crates/ltx/src/replica.rs @@ -404,6 +404,8 @@ pub async fn restore( pub struct RestorePlanStats { pub objects: usize, pub bytes: u64, + /// Highest transaction included in the fixed restore plan. + pub max_txid: u64, /// Object count per compaction level, ordered by level. pub by_level: BTreeMap, } @@ -458,6 +460,7 @@ pub async fn restore_with_download_slots( let mut stats = RestorePlanStats { objects: infos.len(), bytes: infos.iter().map(|info| info.size.max(0) as u64).sum(), + max_txid: infos.iter().map(|info| info.max_txid.0).max().unwrap_or(0), by_level: BTreeMap::new(), }; for info in &infos { diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh new file mode 100755 index 000000000..e02ad7e66 --- /dev/null +++ b/scripts/cell-archive-minio.sh @@ -0,0 +1,171 @@ +#!/usr/bin/env bash + +set -euo pipefail + +readonly MINIO_IMAGE='minio/minio@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e' +readonly MC_IMAGE='minio/mc@sha256:a7fe349ef4bd8521fb8497f55c6042871b2ae640607cf99d9bede5e9bdf11727' +readonly CELLD_IMAGE="${CELLD_ARCHIVE_IMAGE:-celld-ci}" +readonly RUN_ID="cell-archive-${RANDOM}-$$" +readonly NETWORK="${RUN_ID}-network" +readonly MINIO="${RUN_ID}-minio" +readonly TEST_ROOT="$(mktemp -d /tmp/celld-archive.XXXXXX)" +readonly BUCKET='celld-archive/fleet' +readonly ENDPOINT='http://minio:9000' +readonly ACCESS_KEY='celldtest' +readonly SECRET_KEY='celldtestsecret' +export TEST_ROOT + +cleanup() { + docker rm -f "$MINIO" >/dev/null 2>&1 || true + docker network rm "$NETWORK" >/dev/null 2>&1 || true + rm -rf "$TEST_ROOT" +} +trap cleanup EXIT + +docker network create "$NETWORK" >/dev/null +docker run -d --name "$MINIO" --network "$NETWORK" --network-alias minio \ + -e "MINIO_ROOT_USER=$ACCESS_KEY" -e "MINIO_ROOT_PASSWORD=$SECRET_KEY" \ + "$MINIO_IMAGE" server /data >/dev/null + +mc() { + docker run --rm -i --network "$NETWORK" --entrypoint /bin/sh "$MC_IMAGE" -c \ + "mc alias set local $ENDPOINT $ACCESS_KEY $SECRET_KEY >/dev/null && $*" +} + +for attempt in $(seq 1 40); do + if mc 'mc ready local >/dev/null' 2>/dev/null; then + break + fi + if [[ "$attempt" == 40 ]]; then + echo 'MinIO did not become ready' >&2 + exit 1 + fi + sleep 0.25 +done +mc 'mc mb local/celld-archive >/dev/null' + +python3 <<'PY' +import os +import sqlite3 + +path = os.path.join(os.environ["TEST_ROOT"], "source.sqlite") +connection = sqlite3.connect(path) +connection.execute("PRAGMA journal_mode=WAL") +connection.execute("CREATE TABLE facts(id INTEGER PRIMARY KEY, body TEXT NOT NULL)") +connection.executemany( + "INSERT INTO facts(body) VALUES (?)", + [("durable knowledge",), ("second fact",)], +) +connection.commit() +connection.close() +PY + +celld() { + docker run --rm --network "$NETWORK" -v "$TEST_ROOT:/archive" \ + -e "AWS_ACCESS_KEY_ID=$ACCESS_KEY" -e "AWS_SECRET_ACCESS_KEY=$SECRET_KEY" \ + -e AWS_REGION=us-east-1 "$CELLD_IMAGE" "$@" +} + +celld cell import Knowledge:test --input /archive/source.sqlite \ + --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline +celld cell export Knowledge:test --output /archive/export.sqlite \ + --bucket "$BUCKET" --endpoint "$ENDPOINT" + +python3 <<'PY' +import hashlib +import json +import os +import sqlite3 +import stat + +root = os.environ["TEST_ROOT"] +database = os.path.join(root, "export.sqlite") +manifest_path = database + ".manifest.json" +connection = sqlite3.connect(f"file:{database}?mode=ro", uri=True) +rows = connection.execute("SELECT body FROM facts ORDER BY id").fetchall() +connection.close() +assert rows == [("durable knowledge",), ("second fact",)] +with open(database, "rb") as file: + digest = hashlib.sha256(file.read()).hexdigest() +with open(manifest_path, encoding="utf-8") as file: + manifest = json.load(file) +assert manifest == { + "version": 1, + "cell": "Knowledge:test", + "source_epoch": 1, + "source_txid": 1, + "database_sha256": digest, +} +assert stat.S_IMODE(os.stat(database).st_mode) == 0o600 +assert stat.S_IMODE(os.stat(manifest_path).st_mode) == 0o600 +PY + +# Recreate a crash after the owner CAS but before the ready-marker CAS. The +# second command must derive the durable TXID from epoch 1 and finish safely. +ready_marker="$(mc 'mc cat local/celld-archive/fleet/cells/Knowledge:test/import.json')" +source_hash="$(jq -r .source_sha256 <<<"$ready_marker")" +mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:test/import.json >/dev/null' </dev/null + +# A crash before the owner CAS can leave partial LTX. An expired claimant may +# resume, clears that private epoch, and republishes a fully verified lineage. +mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:partial/import.json >/dev/null' </dev/null' <<'PARTIAL' +not-an-ltx-file +PARTIAL +celld cell import Knowledge:partial --input /archive/source.sqlite \ + --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline --resume +partial_marker="$(mc 'mc cat local/celld-archive/fleet/cells/Knowledge:partial/import.json')" +jq -e '.phase == "ready" and .durable_txid == 1' <<<"$partial_marker" >/dev/null + +# A still-live staging claimant cannot be stolen by a concurrent retry. +future_attempt="$(( $(date +%s) * 1000 + 60000 ))" +mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:active/import.json >/dev/null' <"$TEST_ROOT/active.log" 2>&1; then + echo 'a concurrent retry unexpectedly stole an active import attempt' >&2 + exit 1 +fi +grep -F 'is still active' "$TEST_ROOT/active.log" >/dev/null + +# A different archive cannot replace the imported lineage. +python3 <<'PY' +import os +import sqlite3 + +path = os.path.join(os.environ["TEST_ROOT"], "different.sqlite") +connection = sqlite3.connect(path) +connection.execute("CREATE TABLE facts(id INTEGER PRIMARY KEY, body TEXT NOT NULL)") +connection.execute("INSERT INTO facts(body) VALUES ('different')") +connection.commit() +connection.close() +PY +if celld cell import Knowledge:test --input /archive/different.sqlite \ + --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline >"$TEST_ROOT/different.log" 2>&1; then + echo 'a different archive unexpectedly replaced the imported lineage' >&2 + exit 1 +fi +grep -F 'different archive' "$TEST_ROOT/different.log" >/dev/null + +# A live fleet also fails closed before creating a target marker or lineage. +future_ms="$(( $(date +%s) * 1000 + 60000 ))" +mc 'mc pipe local/celld-archive/fleet/nodes/live-test.json >/dev/null' <"$TEST_ROOT/live.log" 2>&1; then + echo 'import unexpectedly ran while a node lease was live' >&2 + exit 1 +fi +grep -F 'live celld node lease(s): live-test' "$TEST_ROOT/live.log" >/dev/null + +echo 'cell archive MinIO test passed' From e44495cd8aca6e426b110540196633c61497fd21 Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 09:26:12 -0700 Subject: [PATCH 3/6] test: preserve archive ownership across runtimes --- scripts/cell-archive-minio.sh | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh index e02ad7e66..3ac11cf4f 100755 --- a/scripts/cell-archive-minio.sh +++ b/scripts/cell-archive-minio.sh @@ -61,7 +61,8 @@ connection.close() PY celld() { - docker run --rm --network "$NETWORK" -v "$TEST_ROOT:/archive" \ + docker run --rm --network "$NETWORK" --user "$(id -u):$(id -g)" \ + -v "$TEST_ROOT:/archive" \ -e "AWS_ACCESS_KEY_ID=$ACCESS_KEY" -e "AWS_SECRET_ACCESS_KEY=$SECRET_KEY" \ -e AWS_REGION=us-east-1 "$CELLD_IMAGE" "$@" } From 9fbbb9b601d8e59515ed99d7d51caf61bdd0fecc Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 09:50:00 -0700 Subject: [PATCH 4/6] test: qualify cell archives against GCS --- docs/README.md | 36 ++++++++ scripts/cell-archive-minio.sh | 162 +++++++++++++++++++++++++--------- 2 files changed, 156 insertions(+), 42 deletions(-) diff --git a/docs/README.md b/docs/README.md index a9036e872..55f5002f0 100644 --- a/docs/README.md +++ b/docs/README.md @@ -123,6 +123,42 @@ The bucket credentials give full control of the fleet. Keep them safe. The bucket contains the deployments, the SQLite replicas, the ownership records, the node leases, and the peer-authentication secret. +### Export and restore a cell + +`cell export` reconstructs one cell at its latest durable transaction and +writes a mode-`0600` SQLite archive plus a JSON manifest containing the cell, +source epoch, source transaction ID, and database SHA-256: + +```sh +celld cell export Knowledge:example \ + --bucket "$CELLD_BUCKET" \ + --output ./knowledge.sqlite +``` + +`cell import` creates a new cell lineage from an archive. It is an offline +operator operation: it fails while any Celld node lease is live, refuses to +replace an existing or differently sourced lineage, and uses an expiring +staging claim so an interrupted import can be resumed safely: + +```sh +celld cell import Knowledge:example \ + --bucket "$CELLD_BUCKET" \ + --input ./knowledge.sqlite \ + --offline + +celld cell import Knowledge:example \ + --bucket "$CELLD_BUCKET" \ + --input ./knowledge.sqlite \ + --offline \ + --resume +``` + +Stop the whole fleet and wait for every node lease to expire before importing. +Keep the same archive for `--resume`; a different archive cannot take over a +staged or completed import. Start nodes only after the command reports the +durable epoch and transaction ID. S3-compatible stores additionally use the +same `--endpoint` and `--region` arguments as other Celld commands. + A bucket value can add a key prefix: `s3://YOUR-BUCKET/PREFIX`. Every object of the fleet then goes below `PREFIX/`, so two fleets can share one bucket. A bucket value without a prefix keeps the objects at the root of diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh index 3ac11cf4f..453af2021 100755 --- a/scripts/cell-archive-minio.sh +++ b/scripts/cell-archive-minio.sh @@ -1,48 +1,114 @@ #!/usr/bin/env bash +# The default CI lane uses an isolated MinIO container. Set +# CELLD_ARCHIVE_BACKEND=gcs, CELLD_ARCHIVE_GCS_PROJECT, and +# CELLD_ARCHIVE_BINARY to exercise the same recovery contract against a unique +# GCS bucket with a host binary and Application Default Credentials. + set -euo pipefail readonly MINIO_IMAGE='minio/minio@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e' readonly MC_IMAGE='minio/mc@sha256:a7fe349ef4bd8521fb8497f55c6042871b2ae640607cf99d9bede5e9bdf11727' readonly CELLD_IMAGE="${CELLD_ARCHIVE_IMAGE:-celld-ci}" +readonly BACKEND="${CELLD_ARCHIVE_BACKEND:-minio}" readonly RUN_ID="cell-archive-${RANDOM}-$$" readonly NETWORK="${RUN_ID}-network" readonly MINIO="${RUN_ID}-minio" -readonly TEST_ROOT="$(mktemp -d /tmp/celld-archive.XXXXXX)" -readonly BUCKET='celld-archive/fleet' +TEST_ROOT="$(mktemp -d /tmp/celld-archive.XXXXXX)" +readonly TEST_ROOT readonly ENDPOINT='http://minio:9000' readonly ACCESS_KEY='celldtest' readonly SECRET_KEY='celldtestsecret' export TEST_ROOT +if [[ "$BACKEND" != 'minio' && "$BACKEND" != 'gcs' ]]; then + echo 'CELLD_ARCHIVE_BACKEND must be minio or gcs' >&2 + exit 1 +fi + +gcs_bucket_created=false +if [[ "$BACKEND" == 'gcs' ]]; then + readonly GCS_PROJECT="${CELLD_ARCHIVE_GCS_PROJECT:-}" + if [[ ! "$GCS_PROJECT" =~ ^[a-z][a-z0-9-]{4,28}[a-z0-9]$ ]]; then + echo 'CELLD_ARCHIVE_GCS_PROJECT must be an explicit GCP project ID' >&2 + exit 1 + fi + if [[ -z "${CELLD_ARCHIVE_BINARY:-}" || ! -x "$CELLD_ARCHIVE_BINARY" ]]; then + echo 'CELLD_ARCHIVE_BINARY must name an executable host Celld binary' >&2 + exit 1 + fi + project_hash="$(printf '%s' "$GCS_PROJECT" | shasum -a 256 | cut -c1-10)" + run_hash="$(printf '%s' "$RUN_ID" | shasum -a 256 | cut -c1-16)" + readonly GCS_BUCKET="cr-celld-archive-$project_hash-$run_hash" + readonly BUCKET="gs://$GCS_BUCKET/fleet" + gcloud auth application-default print-access-token >/dev/null +else + readonly BUCKET='celld-archive/fleet' +fi + cleanup() { docker rm -f "$MINIO" >/dev/null 2>&1 || true docker network rm "$NETWORK" >/dev/null 2>&1 || true + if [[ "$gcs_bucket_created" == true ]]; then + gcloud storage rm --recursive "gs://$GCS_BUCKET/**" >/dev/null 2>&1 || true + gcloud storage buckets delete "gs://$GCS_BUCKET" --quiet >/dev/null 2>&1 || true + fi rm -rf "$TEST_ROOT" } trap cleanup EXIT -docker network create "$NETWORK" >/dev/null -docker run -d --name "$MINIO" --network "$NETWORK" --network-alias minio \ - -e "MINIO_ROOT_USER=$ACCESS_KEY" -e "MINIO_ROOT_PASSWORD=$SECRET_KEY" \ - "$MINIO_IMAGE" server /data >/dev/null +if [[ "$BACKEND" == 'minio' ]]; then + docker network create "$NETWORK" >/dev/null + docker run -d --name "$MINIO" --network "$NETWORK" --network-alias minio \ + -e "MINIO_ROOT_USER=$ACCESS_KEY" -e "MINIO_ROOT_PASSWORD=$SECRET_KEY" \ + "$MINIO_IMAGE" server /data >/dev/null +else + gcloud storage buckets create "gs://$GCS_BUCKET" \ + --project "$GCS_PROJECT" --location us-central1 \ + --uniform-bucket-level-access >/dev/null + gcs_bucket_created=true +fi mc() { docker run --rm -i --network "$NETWORK" --entrypoint /bin/sh "$MC_IMAGE" -c \ "mc alias set local $ENDPOINT $ACCESS_KEY $SECRET_KEY >/dev/null && $*" } -for attempt in $(seq 1 40); do - if mc 'mc ready local >/dev/null' 2>/dev/null; then - break +object_cat() { + local key="$1" + if [[ "$BACKEND" == 'minio' ]]; then + mc "mc cat local/celld-archive/fleet/$key" + else + gcloud storage cat "gs://$GCS_BUCKET/fleet/$key" fi - if [[ "$attempt" == 40 ]]; then - echo 'MinIO did not become ready' >&2 - exit 1 +} + +object_put() { + local key="$1" + if [[ "$BACKEND" == 'minio' ]]; then + mc "mc pipe local/celld-archive/fleet/$key >/dev/null" + else + local input_file + input_file="$(mktemp "$TEST_ROOT/object.XXXXXX")" + cat > "$input_file" + gcloud storage cp "$input_file" "gs://$GCS_BUCKET/fleet/$key" >/dev/null + rm -f "$input_file" fi - sleep 0.25 -done -mc 'mc mb local/celld-archive >/dev/null' +} + +if [[ "$BACKEND" == 'minio' ]]; then + for attempt in $(seq 1 40); do + if mc 'mc ready local >/dev/null' 2>/dev/null; then + break + fi + if [[ "$attempt" == 40 ]]; then + echo 'MinIO did not become ready' >&2 + exit 1 + fi + sleep 0.25 + done + mc 'mc mb local/celld-archive >/dev/null' +fi python3 <<'PY' import os @@ -61,16 +127,28 @@ connection.close() PY celld() { - docker run --rm --network "$NETWORK" --user "$(id -u):$(id -g)" \ - -v "$TEST_ROOT:/archive" \ - -e "AWS_ACCESS_KEY_ID=$ACCESS_KEY" -e "AWS_SECRET_ACCESS_KEY=$SECRET_KEY" \ - -e AWS_REGION=us-east-1 "$CELLD_IMAGE" "$@" + if [[ "$BACKEND" == 'minio' ]]; then + docker run --rm --network "$NETWORK" --user "$(id -u):$(id -g)" \ + -v "$TEST_ROOT:/archive" \ + -e "AWS_ACCESS_KEY_ID=$ACCESS_KEY" -e "AWS_SECRET_ACCESS_KEY=$SECRET_KEY" \ + -e AWS_REGION=us-east-1 "$CELLD_IMAGE" "$@" + else + "$CELLD_ARCHIVE_BINARY" "$@" + fi } -celld cell import Knowledge:test --input /archive/source.sqlite \ - --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline -celld cell export Knowledge:test --output /archive/export.sqlite \ - --bucket "$BUCKET" --endpoint "$ENDPOINT" +if [[ "$BACKEND" == 'minio' ]]; then + archive_root='/archive' + storage_args=(--bucket "$BUCKET" --endpoint "$ENDPOINT") +else + archive_root="$TEST_ROOT" + storage_args=(--bucket "$BUCKET") +fi + +celld cell import Knowledge:test --input "$archive_root/source.sqlite" \ + "${storage_args[@]}" --offline +celld cell export Knowledge:test --output "$archive_root/export.sqlite" \ + "${storage_args[@]}" python3 <<'PY' import hashlib @@ -103,36 +181,36 @@ PY # Recreate a crash after the owner CAS but before the ready-marker CAS. The # second command must derive the durable TXID from epoch 1 and finish safely. -ready_marker="$(mc 'mc cat local/celld-archive/fleet/cells/Knowledge:test/import.json')" +ready_marker="$(object_cat 'cells/Knowledge:test/import.json')" source_hash="$(jq -r .source_sha256 <<<"$ready_marker")" -mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:test/import.json >/dev/null' </dev/null # A crash before the owner CAS can leave partial LTX. An expired claimant may # resume, clears that private epoch, and republishes a fully verified lineage. -mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:partial/import.json >/dev/null' </dev/null' <<'PARTIAL' +object_put 'cells/Knowledge:partial/ltx/e1/partial' <<'PARTIAL' not-an-ltx-file PARTIAL -celld cell import Knowledge:partial --input /archive/source.sqlite \ - --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline --resume -partial_marker="$(mc 'mc cat local/celld-archive/fleet/cells/Knowledge:partial/import.json')" +celld cell import Knowledge:partial --input "$archive_root/source.sqlite" \ + "${storage_args[@]}" --offline --resume +partial_marker="$(object_cat 'cells/Knowledge:partial/import.json')" jq -e '.phase == "ready" and .durable_txid == 1' <<<"$partial_marker" >/dev/null # A still-live staging claimant cannot be stolen by a concurrent retry. future_attempt="$(( $(date +%s) * 1000 + 60000 ))" -mc 'mc pipe local/celld-archive/fleet/cells/Knowledge:active/import.json >/dev/null' <"$TEST_ROOT/active.log" 2>&1; then +if celld cell import Knowledge:active --input "$archive_root/source.sqlite" \ + "${storage_args[@]}" --offline --resume >"$TEST_ROOT/active.log" 2>&1; then echo 'a concurrent retry unexpectedly stole an active import attempt' >&2 exit 1 fi @@ -150,8 +228,8 @@ connection.execute("INSERT INTO facts(body) VALUES ('different')") connection.commit() connection.close() PY -if celld cell import Knowledge:test --input /archive/different.sqlite \ - --bucket "$BUCKET" --endpoint "$ENDPOINT" --offline >"$TEST_ROOT/different.log" 2>&1; then +if celld cell import Knowledge:test --input "$archive_root/different.sqlite" \ + "${storage_args[@]}" --offline >"$TEST_ROOT/different.log" 2>&1; then echo 'a different archive unexpectedly replaced the imported lineage' >&2 exit 1 fi @@ -159,14 +237,14 @@ grep -F 'different archive' "$TEST_ROOT/different.log" >/dev/null # A live fleet also fails closed before creating a target marker or lineage. future_ms="$(( $(date +%s) * 1000 + 60000 ))" -mc 'mc pipe local/celld-archive/fleet/nodes/live-test.json >/dev/null' <"$TEST_ROOT/live.log" 2>&1; then +if celld cell import Knowledge:blocked --input "$archive_root/source.sqlite" \ + "${storage_args[@]}" --offline >"$TEST_ROOT/live.log" 2>&1; then echo 'import unexpectedly ran while a node lease was live' >&2 exit 1 fi grep -F 'live celld node lease(s): live-test' "$TEST_ROOT/live.log" >/dev/null -echo 'cell archive MinIO test passed' +echo "cell archive $BACKEND test passed" From edfff68e21a5275be606b50da5a7433c62682459 Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 10:03:32 -0700 Subject: [PATCH 5/6] fix: harden cell archive recovery paths --- crates/celld/bucket.rs | 12 +++++ crates/celld/cell_archive.rs | 80 +++++++++++++++++++++++++++------ crates/celld/ltx_repl.rs | 34 ++++++++++++-- crates/celld/main.rs | 15 ++++++- crates/celld/ownership_store.rs | 57 ++++++++++++++++++++++- crates/ltx/src/replica.rs | 2 +- scripts/cell-archive-minio.sh | 19 +++++++- 7 files changed, 196 insertions(+), 23 deletions(-) diff --git a/crates/celld/bucket.rs b/crates/celld/bucket.rs index adcc1ec38..7f341bcd3 100644 --- a/crates/celld/bucket.rs +++ b/crates/celld/bucket.rs @@ -154,6 +154,18 @@ fn split_spec(spec: &str) -> (StorageBackend, &str, String) { } impl Bucket { + #[cfg(test)] + pub(crate) fn memory_for_test() -> Bucket { + let store: Arc = Arc::new(object_store::memory::InMemory::new()); + Bucket { + store: store.clone(), + cas_store: store, + backend: StorageBackend::S3, + name: "memory".to_string(), + prefix: String::new(), + } + } + /// `bucket` is `[s3://|gs://]NAME[/PREFIX]`. With a PREFIX every key /// this client reads or writes lives under `PREFIX/`, so several /// fleets can share one bucket without colliding. diff --git a/crates/celld/cell_archive.rs b/crates/celld/cell_archive.rs index e4b176834..26efbcf33 100644 --- a/crates/celld/cell_archive.rs +++ b/crates/celld/cell_archive.rs @@ -15,6 +15,7 @@ use anyhow::{bail, Context}; use celld_logic::CasOutcome; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use std::io::{BufReader, Read}; use std::path::{Path, PathBuf}; const ARCHIVE_VERSION: u32 = 1; @@ -204,18 +205,27 @@ fn parse(arguments: Vec) -> anyhow::Result { .unwrap_or_else(|| "us-east-1".to_string()), }; match operation.as_str() { - "export" => Ok(Command::Export { - cell, - output: output.context("cell export requires --output DATABASE")?, - storage, - }), - "import" => Ok(Command::Import { - cell, - input: input.context("cell import requires --input DATABASE")?, - storage, - offline, - resume, - }), + "export" => { + anyhow::ensure!( + input.is_none() && !offline && !resume, + "cell export does not accept --input, --offline, or --resume" + ); + Ok(Command::Export { + cell, + output: output.context("cell export requires --output DATABASE")?, + storage, + }) + } + "import" => { + anyhow::ensure!(output.is_none(), "cell import does not accept --output"); + Ok(Command::Import { + cell, + input: input.context("cell import requires --input DATABASE")?, + storage, + offline, + resume, + }) + } _ => unreachable!(), } } @@ -535,8 +545,17 @@ fn validate_sqlite(path: &Path) -> anyhow::Result<()> { } fn sha256_file(path: &Path) -> anyhow::Result { - let bytes = std::fs::read(path)?; - Ok(format!("{:x}", Sha256::digest(bytes))) + let mut reader = BufReader::new(std::fs::File::open(path)?); + let mut digest = Sha256::new(); + let mut buffer = [0_u8; 64 * 1024]; + loop { + let read = reader.read(&mut buffer)?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + Ok(format!("{:x}", digest.finalize())) } fn manifest_path(database: &Path) -> PathBuf { @@ -601,6 +620,39 @@ mod tests { assert!(error.to_string().contains("invalid cell scope")); } + #[test] + fn rejects_options_owned_by_the_other_archive_operation() { + let export_error = parse(vec![ + "export".into(), + "Org:test".into(), + "--output".into(), + "archive.sqlite".into(), + "--bucket".into(), + "bucket".into(), + "--resume".into(), + ]) + .unwrap_err(); + assert!(export_error + .to_string() + .contains("cell export does not accept")); + + let import_error = parse(vec![ + "import".into(), + "Org:test".into(), + "--input".into(), + "archive.sqlite".into(), + "--output".into(), + "ignored.sqlite".into(), + "--bucket".into(), + "bucket".into(), + "--offline".into(), + ]) + .unwrap_err(); + assert!(import_error + .to_string() + .contains("cell import does not accept --output")); + } + #[test] fn validates_and_hashes_sqlite() { let directory = tempfile::tempdir().unwrap(); diff --git a/crates/celld/ltx_repl.rs b/crates/celld/ltx_repl.rs index 540fbf976..722ada4b7 100644 --- a/crates/celld/ltx_repl.rs +++ b/crates/celld/ltx_repl.rs @@ -15,6 +15,7 @@ //! one bucket would replicate over each other. use std::collections::HashMap; +use std::io::{BufReader, Read}; use std::path::Path; use std::path::PathBuf; use std::sync::atomic::AtomicBool; @@ -670,9 +671,14 @@ impl LtxRepl { cell: &str, epoch: u64, ) -> anyhow::Result> { - let plan = replica::calc_restore_plan(&self.client_for(cell, epoch), TXID(0)) - .await - .with_context(|| format!("plan durable position for {cell} e{epoch}"))?; + let plan = match replica::calc_restore_plan(&self.client_for(cell, epoch), TXID(0)).await { + Ok(plan) => plan, + Err(celld_ltx::Error::TxNotAvailable) => return Ok(None), + Err(error) => { + return Err(anyhow!(error)) + .with_context(|| format!("plan durable position for {cell} e{epoch}")); + } + }; Ok(plan.iter().map(|info| info.max_txid.0).max()) } @@ -746,7 +752,7 @@ impl LtxRepl { sqlite_snapshot(&path, &expected).context("normalize captured import for comparison")?; sqlite_snapshot(&restored, &actual).context("normalize restored import for comparison")?; anyhow::ensure!( - std::fs::read(&expected)? == std::fs::read(&actual)?, + files_equal(&expected, &actual)?, "import LTX round trip does not match the staged SQLite database" ); let _ = std::fs::remove_dir_all(&directory); @@ -856,6 +862,26 @@ impl LtxRepl { } } +fn files_equal(left: &Path, right: &Path) -> std::io::Result { + if std::fs::metadata(left)?.len() != std::fs::metadata(right)?.len() { + return Ok(false); + } + let mut left = BufReader::new(std::fs::File::open(left)?); + let mut right = BufReader::new(std::fs::File::open(right)?); + let mut left_buffer = [0_u8; 64 * 1024]; + let mut right_buffer = [0_u8; 64 * 1024]; + loop { + let left_read = left.read(&mut left_buffer)?; + let right_read = right.read(&mut right_buffer)?; + if left_read != right_read || left_buffer[..left_read] != right_buffer[..right_read] { + return Ok(false); + } + if left_read == 0 { + return Ok(true); + } + } +} + /// One capture+upload for a cell: advance its durable position on success and /// wake its waiters. Everything committed before the capture is durable once /// uploaded, so the target is read before `db.sync`. The `rusqlite` handle is diff --git a/crates/celld/main.rs b/crates/celld/main.rs index 78cdad266..067cb9d8a 100644 --- a/crates/celld/main.rs +++ b/crates/celld/main.rs @@ -115,6 +115,19 @@ enum Ownership { } impl Ownership { + async fn read_owner_for_activation(&self, cell: &str) -> Result, Failure> { + match self { + Self::Memory(memory) => Ok(memory.lock().await.owners.get(cell).cloned()), + Self::Bucket(bucket) => bucket + .read_owner_for_activation(cell) + .await + .map_err(|error| { + eprintln!("celld ownership activation read failed: {error:#}"); + Failure::Definite + }), + } + } + async fn read_owner(&self, cell: &str) -> Result, Failure> { match self { Self::Memory(memory) => Ok(memory.lock().await.owners.get(cell).cloned()), @@ -1799,7 +1812,7 @@ impl Actor { let timing_cell = cell.clone(); in_flight.push(Box::pin(async move { let started = Instant::now(); - let result = ownership.read_owner(&cell).await; + let result = ownership.read_owner_for_activation(&cell).await; CompletedEffect::timed( Event::OwnerRead { op, diff --git a/crates/celld/ownership_store.rs b/crates/celld/ownership_store.rs index 0437110e4..5c385d7f0 100644 --- a/crates/celld/ownership_store.rs +++ b/crates/celld/ownership_store.rs @@ -184,7 +184,6 @@ impl BucketOwnership { } pub async fn read_owner(&self, cell: &str) -> anyhow::Result> { - crate::cell_archive::ensure_import_ready(&self.bucket, cell).await?; let key = format!("cells/{cell}/own.json"); let Some((owner, etag)) = load_json::(&self.bucket, &key).await? else { return Ok(None); @@ -196,6 +195,17 @@ impl BucketOwnership { })) } + /// Read authority while admitting a cold activation or reconciling an + /// acquisition. Import markers are checked only on this path so ordinary + /// output-gate proofs and releases do not add an object-store request. + pub async fn read_owner_for_activation( + &self, + cell: &str, + ) -> anyhow::Result> { + crate::cell_archive::ensure_import_ready(&self.bucket, cell).await?; + self.read_owner(cell).await + } + /// Publish the initial unowned record for a fully staged import. /// /// This is deliberately absent-only. Import must never replace, rewind, @@ -436,3 +446,48 @@ fn process_load(live: &LiveLoad) -> NodeLoadWire { restoring: live.restoring.load(Ordering::Relaxed), } } + +#[cfg(test)] +mod tests { + use super::BucketOwnership; + use crate::bucket::Bucket; + + #[tokio::test] + async fn staging_import_marker_blocks_activation_only() { + let bucket = Bucket::memory_for_test(); + bucket + .put( + "cells/Knowledge:test/import.json", + br#"{ + "version": 1, + "phase": "staging", + "source_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "attempt_id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "attempt_expires_ms": 1 + }"# + .to_vec(), + ) + .await + .unwrap(); + let ownership = BucketOwnership::new( + bucket.clone(), + bucket, + "node.test".to_string(), + "probe-key".to_string(), + ); + + assert!(ownership + .read_owner("Knowledge:test") + .await + .unwrap() + .is_none()); + let error = ownership + .read_owner_for_activation("Knowledge:test") + .await + .unwrap_err(); + assert!( + error.to_string().contains("incomplete offline import"), + "unexpected error: {error:#}" + ); + } +} diff --git a/crates/ltx/src/replica.rs b/crates/ltx/src/replica.rs index c30bf8163..147a9edf4 100644 --- a/crates/ltx/src/replica.rs +++ b/crates/ltx/src/replica.rs @@ -460,7 +460,7 @@ pub async fn restore_with_download_slots( let mut stats = RestorePlanStats { objects: infos.len(), bytes: infos.iter().map(|info| info.size.max(0) as u64).sum(), - max_txid: infos.iter().map(|info| info.max_txid.0).max().unwrap_or(0), + max_txid: slice_max_txid(&infos).0, by_level: BTreeMap::new(), }; for info in &infos { diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh index 453af2021..7eebb5a4c 100755 --- a/scripts/cell-archive-minio.sh +++ b/scripts/cell-archive-minio.sh @@ -17,8 +17,10 @@ readonly MINIO="${RUN_ID}-minio" TEST_ROOT="$(mktemp -d /tmp/celld-archive.XXXXXX)" readonly TEST_ROOT readonly ENDPOINT='http://minio:9000' -readonly ACCESS_KEY='celldtest' -readonly SECRET_KEY='celldtestsecret' +ACCESS_KEY="celld$(printf '%s' "$RUN_ID-access" | shasum -a 256 | cut -c1-12)" +readonly ACCESS_KEY +SECRET_KEY="$(printf '%s' "$RUN_ID-secret" | shasum -a 256 | cut -c1-32)" +readonly SECRET_KEY export TEST_ROOT if [[ "$BACKEND" != 'minio' && "$BACKEND" != 'gcs' ]]; then @@ -96,6 +98,15 @@ object_put() { fi } +object_prefix_exists() { + local prefix="$1" + if [[ "$BACKEND" == 'minio' ]]; then + [[ -n "$(mc "mc find local/celld-archive/fleet/$prefix" 2>/dev/null)" ]] + else + gcloud storage ls --recursive "gs://$GCS_BUCKET/fleet/$prefix/**" >/dev/null 2>&1 + fi +} + if [[ "$BACKEND" == 'minio' ]]; then for attempt in $(seq 1 40); do if mc 'mc ready local >/dev/null' 2>/dev/null; then @@ -246,5 +257,9 @@ if celld cell import Knowledge:blocked --input "$archive_root/source.sqlite" \ exit 1 fi grep -F 'live celld node lease(s): live-test' "$TEST_ROOT/live.log" >/dev/null +if object_prefix_exists 'cells/Knowledge:blocked'; then + echo 'failed live-fleet import created target marker, owner, or LTX state' >&2 + exit 1 +fi echo "cell archive $BACKEND test passed" From ad991a92413a0a83305b8cc4618f6d220ef1f72c Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Mon, 17 Aug 2026 19:40:40 -0700 Subject: [PATCH 6/6] test: fail archive checks on storage errors --- scripts/cell-archive-minio.sh | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh index 7eebb5a4c..79fc1d1cb 100755 --- a/scripts/cell-archive-minio.sh +++ b/scripts/cell-archive-minio.sh @@ -100,11 +100,34 @@ object_put() { object_prefix_exists() { local prefix="$1" + local output + local root + local status if [[ "$BACKEND" == 'minio' ]]; then - [[ -n "$(mc "mc find local/celld-archive/fleet/$prefix" 2>/dev/null)" ]] + root='local/celld-archive/fleet' + if output="$(mc "mc find $root" 2>&1)"; then + : + else + status=$? + printf '%s\n' "$output" >&2 + exit "$status" + fi else - gcloud storage ls --recursive "gs://$GCS_BUCKET/fleet/$prefix/**" >/dev/null 2>&1 + root="gs://$GCS_BUCKET/fleet" + if output="$(gcloud storage ls --recursive "$root/**" 2>&1)"; then + : + else + status=$? + printf '%s\n' "$output" >&2 + exit "$status" + fi fi + while IFS= read -r object; do + if [[ "$object" == "$root/$prefix" || "$object" == "$root/$prefix/"* ]]; then + return 0 + fi + done <<< "$output" + return 1 } if [[ "$BACKEND" == 'minio' ]]; then