diff --git a/crates/celld/ltx_repl.rs b/crates/celld/ltx_repl.rs index 8f6ef8512..1cda58479 100644 --- a/crates/celld/ltx_repl.rs +++ b/crates/celld/ltx_repl.rs @@ -35,6 +35,8 @@ use celld_ltx::ObjectStoreClient; use celld_ltx::ObjectStoreConfig; use celld_ltx::Replica; use celld_ltx::TXID; +use sha2::Digest; +use sha2::Sha256; use tokio::sync::mpsc; use tokio::sync::Notify; use tokio::sync::Semaphore; @@ -68,6 +70,18 @@ const COMPACTION_MAX_FILES: usize = 256; /// input set until the client gains a streaming read and write surface. const COMPACTION_MAX_INPUT_BYTES: u64 = 64 * 1024 * 1024; +const FORK_SEED_FORMAT: &str = "celld-sqlite-fork-seed-v1"; + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] +pub struct ForkSeedManifest { + pub format: String, + pub checkpoint_id: String, + pub source_cell: String, + pub source_epoch: u64, + pub sqlite_sha256: String, + pub sqlite_bytes: u64, +} + #[derive(Debug, Clone, Copy)] struct CompactionConfig { min_txids: u64, @@ -137,10 +151,32 @@ pub struct LtxRepl { compaction_min_txids: u64, } +fn snapshot_active_at( + watch: &Path, + source: &Path, + cell: &str, + epoch: u64, +) -> anyhow::Result> { + if !source.is_file() { + return Ok(None); + } + let directory = watch.join(format!( + ".inspect-{cell}-e{epoch}-{:032x}", + rand::random::() + )); + std::fs::create_dir_all(&directory)?; + let path = directory.join("db.sqlite"); + if let Err(error) = sqlite_snapshot(source, &path) { + let _ = std::fs::remove_dir_all(&directory); + return Err(error); + } + Ok(Some(RestoredSnapshot::new(epoch, path, directory))) +} + impl LtxRepl { /// Private-corpus constructor over an injected store, so the epoch-seal /// protocol runs against an in-memory bucket instead of S3. - #[cfg(all(test, celld_internal_tests))] + #[cfg(test)] pub fn start_with_store_for_test(watch: &Path, store: Arc) -> Self { let cells: Arc>> = Arc::default(); let dirty = Arc::new(Notify::new()); @@ -287,6 +323,288 @@ impl LtxRepl { format!("{}cells/{cell}/ltx/e{epoch}.seal.json", self.prefix) } + fn fork_seed_key(&self, cell: &str, name: &str) -> String { + format!("{}cells/{cell}/fork-seed/{name}", self.prefix) + } + + fn checkpoint_key(&self, cell: &str, checkpoint: &str, name: &str) -> String { + format!( + "{}cells/{cell}/checkpoints/{checkpoint}/{name}", + self.prefix + ) + } + + async fn put_fork_seed_object( + &self, + cell: &str, + name: &str, + bytes: Vec, + ) -> anyhow::Result<()> { + use celld_ltx::object_store::path::Path as ObjPath; + use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; + + let key = ObjPath::from(self.fork_seed_key(cell, name)); + let create = PutOptions { + mode: PutMode::Create, + ..Default::default() + }; + match self + .store + .put_opts(&key, PutPayload::from(bytes.clone()), create) + .await + { + Ok(_) => Ok(()), + Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { + let existing = self.store.get(&key).await?.bytes().await?; + anyhow::ensure!( + existing.as_ref() == bytes, + "fork seed target {cell} already contains a different {name}" + ); + Ok(()) + } + Err(error) => Err(anyhow!("publish fork seed {cell}/{name}: {error}")), + } + } + + /// Publish a content-verified immutable checkpoint of the active cell. + pub async fn publish_checkpoint( + &self, + source_cell: &str, + source_epoch: u64, + checkpoint_id: &str, + ) -> anyhow::Result { + anyhow::ensure!( + celld_logic::cell::valid_cell_scope(source_cell) + && celld_logic::cell::valid_cell_scope(checkpoint_id), + "invalid checkpoint coordinate" + ); + let source = self.db_path(source_cell, source_epoch); + let watch = self.watch.clone(); + let cell = source_cell.to_string(); + let sqlite = tokio::task::spawn_blocking(move || -> anyhow::Result>> { + let Some(snapshot) = snapshot_active_at(&watch, &source, &cell, source_epoch)? else { + return Ok(None); + }; + Ok(Some(std::fs::read(snapshot.path())?)) + }) + .await?? + .ok_or_else(|| anyhow!("fork source is not active on this node"))?; + let manifest = ForkSeedManifest { + format: FORK_SEED_FORMAT.to_string(), + checkpoint_id: checkpoint_id.to_string(), + source_cell: source_cell.to_string(), + source_epoch, + sqlite_sha256: format!("{:x}", Sha256::digest(&sqlite)), + sqlite_bytes: sqlite.len() as u64, + }; + let encoded_manifest = serde_json::to_vec(&manifest)?; + self.put_checkpoint_object(source_cell, checkpoint_id, "database.sqlite", sqlite) + .await?; + self.put_checkpoint_object( + source_cell, + checkpoint_id, + "manifest.json", + encoded_manifest, + ) + .await?; + Ok(manifest) + } + + async fn put_checkpoint_object( + &self, + cell: &str, + checkpoint: &str, + name: &str, + bytes: Vec, + ) -> anyhow::Result<()> { + use celld_ltx::object_store::path::Path as ObjPath; + use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; + + let key = ObjPath::from(self.checkpoint_key(cell, checkpoint, name)); + let create = PutOptions { + mode: PutMode::Create, + ..Default::default() + }; + match self + .store + .put_opts(&key, PutPayload::from(bytes.clone()), create) + .await + { + Ok(_) => Ok(()), + Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { + let existing = self.store.get(&key).await?.bytes().await?; + anyhow::ensure!( + existing.as_ref() == bytes, + "checkpoint {cell}/{checkpoint} already contains a different {name}" + ); + Ok(()) + } + Err(error) => Err(anyhow!( + "publish checkpoint {cell}/{checkpoint}/{name}: {error}" + )), + } + } + + async fn read_checkpoint( + &self, + source_cell: &str, + checkpoint_id: &str, + ) -> anyhow::Result<(ForkSeedManifest, Vec)> { + use celld_ltx::object_store::path::Path as ObjPath; + + let manifest_key = + ObjPath::from(self.checkpoint_key(source_cell, checkpoint_id, "manifest.json")); + let manifest: ForkSeedManifest = + serde_json::from_slice(&self.store.get(&manifest_key).await?.bytes().await?)?; + anyhow::ensure!( + manifest.format == FORK_SEED_FORMAT, + "unsupported checkpoint format" + ); + anyhow::ensure!( + manifest.source_cell == source_cell && manifest.checkpoint_id == checkpoint_id, + "checkpoint coordinates do not match its manifest" + ); + let database_key = + ObjPath::from(self.checkpoint_key(source_cell, checkpoint_id, "database.sqlite")); + let sqlite = self.store.get(&database_key).await?.bytes().await?.to_vec(); + anyhow::ensure!( + sqlite.len() as u64 == manifest.sqlite_bytes, + "checkpoint byte count mismatch" + ); + anyhow::ensure!( + format!("{:x}", Sha256::digest(&sqlite)) == manifest.sqlite_sha256, + "checkpoint hash mismatch" + ); + Ok((manifest, sqlite)) + } + + /// Seed a never-before-activated target from one immutable checkpoint. + /// The ready manifest is last so first activation fails closed if copying + /// is interrupted. Every write is create-or-verify for exact retry. + pub async fn publish_fork_seed_from_checkpoint( + &self, + source_cell: &str, + checkpoint_id: &str, + target_cell: &str, + target_active: bool, + ) -> anyhow::Result { + use celld_ltx::object_store::path::Path as ObjPath; + + anyhow::ensure!( + source_cell != target_cell, + "fork source and target must differ" + ); + anyhow::ensure!( + celld_logic::cell::valid_cell_scope(source_cell) + && celld_logic::cell::valid_cell_scope(checkpoint_id) + && celld_logic::cell::valid_cell_scope(target_cell), + "invalid fork coordinate" + ); + let (manifest, sqlite) = self.read_checkpoint(source_cell, checkpoint_id).await?; + let encoded_manifest = serde_json::to_vec(&manifest)?; + let ready = ObjPath::from(self.fork_seed_key(target_cell, "ready.json")); + let exact_retry = match self.store.get(&ready).await { + Ok(result) => { + let existing = result.bytes().await?; + anyhow::ensure!( + existing.as_ref() == encoded_manifest, + "fork seed target {target_cell} already contains a different ready.json" + ); + true + } + Err(celld_ltx::object_store::Error::NotFound { .. }) => false, + Err(error) => return Err(anyhow!("read fork seed for {target_cell}: {error}")), + }; + if exact_retry { + self.put_fork_seed_object(target_cell, "reserved.json", encoded_manifest.clone()) + .await?; + self.put_fork_seed_object(target_cell, "database.sqlite", sqlite) + .await?; + self.put_fork_seed_object(target_cell, "ready.json", encoded_manifest) + .await?; + return Ok(manifest); + } + anyhow::ensure!( + !target_active, + "fork target {target_cell} is already active" + ); + anyhow::ensure!( + self.highest_nonempty_epoch(target_cell).await?.is_none(), + "fork target {target_cell} already has a durable replica" + ); + self.put_fork_seed_object(target_cell, "reserved.json", encoded_manifest.clone()) + .await?; + self.put_fork_seed_object(target_cell, "database.sqlite", sqlite) + .await?; + self.put_fork_seed_object(target_cell, "ready.json", encoded_manifest) + .await?; + Ok(manifest) + } + + async fn restore_fork_seed(&self, cell: &str, destination: &Path) -> anyhow::Result { + use celld_ltx::object_store::path::Path as ObjPath; + + let reservation = ObjPath::from(self.fork_seed_key(cell, "reserved.json")); + match self.store.head(&reservation).await { + Ok(_) => {} + Err(celld_ltx::object_store::Error::NotFound { .. }) => return Ok(false), + Err(error) => return Err(anyhow!("read fork reservation for {cell}: {error}")), + } + let ready = ObjPath::from(self.fork_seed_key(cell, "ready.json")); + let manifest: ForkSeedManifest = match self.store.get(&ready).await { + Ok(result) => serde_json::from_slice(&result.bytes().await?)?, + Err(celld_ltx::object_store::Error::NotFound { .. }) => { + anyhow::bail!("fork seed for {cell} is reserved but incomplete") + } + Err(error) => return Err(anyhow!("read fork manifest for {cell}: {error}")), + }; + anyhow::ensure!( + manifest.format == FORK_SEED_FORMAT, + "unsupported fork seed format" + ); + let database = ObjPath::from(self.fork_seed_key(cell, "database.sqlite")); + let sqlite = self.store.get(&database).await?.bytes().await?; + anyhow::ensure!( + sqlite.len() as u64 == manifest.sqlite_bytes, + "fork seed byte count mismatch" + ); + anyhow::ensure!( + format!("{:x}", Sha256::digest(&sqlite)) == manifest.sqlite_sha256, + "fork seed hash mismatch" + ); + let temporary = destination.with_extension("fork-seed.tmp"); + let destination = destination.to_path_buf(); + let expected_bytes = manifest.sqlite_bytes; + let expected_sha256 = manifest.sqlite_sha256.clone(); + tokio::task::spawn_blocking(move || -> anyhow::Result<()> { + let _ = std::fs::remove_file(&temporary); + std::fs::write(&temporary, &sqlite)?; + // FTS5's integrity path may use SQLite's write machinery even though + // quick_check is logically read-only. Validate a private temporary + // copy with normal flags, then prove the main database bytes remain + // identical to the signed manifest before activation. + let connection = rusqlite::Connection::open(&temporary)?; + let quick_check: String = connection + .query_row("PRAGMA quick_check", [], |row| row.get(0)) + .map_err(|error| anyhow!("fork seed SQLite quick_check failed: {error}"))?; + anyhow::ensure!( + quick_check == "ok", + "fork seed SQLite quick_check failed: {quick_check}" + ); + drop(connection); + let validated = std::fs::read(&temporary)?; + anyhow::ensure!( + validated.len() as u64 == expected_bytes + && format!("{:x}", Sha256::digest(&validated)) == expected_sha256, + "fork seed SQLite validation changed the database bytes" + ); + std::fs::rename(temporary, destination)?; + Ok(()) + }) + .await??; + Ok(true) + } + /// Read the source epoch's seal, writing it first if this activation is /// the first to restore from `from` (Cellarium §5.6). /// @@ -435,6 +753,9 @@ impl LtxRepl { std::fs::rename(&snapshot, &dst)?; info!(cell, epoch, "reused local eviction snapshot"); restored = true; + } else if fresh && self.restore_fork_seed(cell, &dst).await? { + info!(cell, epoch, "restored immutable fork seed"); + restored = true; } else if !fresh { // Restore the newest durable epoch into this epoch's path — up to // its seal, never past it. Sealing before reading fixes the cut @@ -623,16 +944,7 @@ impl LtxRepl { cell: &str, epoch: u64, ) -> anyhow::Result> { - let source = self.db_path(cell, epoch); - if !source.is_file() { - return Ok(None); - } - let directory = self.watch.join(format!(".inspect-{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)?; - Ok(Some(RestoredSnapshot::new(epoch, path, directory))) + snapshot_active_at(&self.watch, &self.db_path(cell, epoch), cell, epoch) } /// Restore the newest durable replica into a private snapshot without @@ -1102,3 +1414,248 @@ fn node_config( part_size: 0, } } + +#[cfg(test)] +mod fork_seed_tests { + use super::*; + use celld_ltx::object_store::memory::InMemory; + use celld_ltx::object_store::path::Path as ObjPath; + use celld_ltx::object_store::PutPayload; + + fn activation<'a>(cell: &'a str, fresh: bool) -> ActivationOptions<'a> { + ActivationOptions { + cell, + epoch: 1, + fresh, + took_over: false, + resume_local: false, + } + } + + async fn plant_fork_seed( + store: &Arc, + cell: &str, + manifest: &ForkSeedManifest, + sqlite: Vec, + ) { + let encoded = serde_json::to_vec(manifest).unwrap(); + for (name, bytes) in [ + ("reserved.json", encoded.clone()), + ("database.sqlite", sqlite), + ("ready.json", encoded), + ] { + store + .put( + &ObjPath::from(format!("cells/{cell}/fork-seed/{name}")), + PutPayload::from(bytes), + ) + .await + .unwrap(); + } + } + + #[tokio::test] + async fn fork_seed_is_exact_create_only_and_independent() { + let directory = tempfile::tempdir().unwrap(); + let store: Arc = Arc::new(InMemory::new()); + let replication = LtxRepl::start_with_store_for_test(directory.path(), store); + let source = replication + .activate(activation("source", true)) + .await + .unwrap(); + { + let connection = rusqlite::Connection::open(&source.path).unwrap(); + connection + .execute_batch( + "CREATE TABLE state(key TEXT PRIMARY KEY, value TEXT NOT NULL);\n\ + INSERT INTO state VALUES ('phase', 'checkpointed');", + ) + .unwrap(); + } + replication.await_durable("source", 1, 1).await.unwrap(); + + let manifest = replication + .publish_checkpoint("source", 1, "checkpoint-1") + .await + .unwrap(); + replication + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", false) + .await + .unwrap(); + assert_eq!(manifest.source_cell, "source"); + assert_eq!(manifest.source_epoch, 1); + assert_eq!(manifest.checkpoint_id, "checkpoint-1"); + assert_eq!(manifest.sqlite_sha256.len(), 64); + assert_eq!( + replication + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", false,) + .await + .unwrap(), + manifest + ); + + let source_connection = rusqlite::Connection::open(&source.path).unwrap(); + source_connection + .execute( + "UPDATE state SET value = 'source-advanced' WHERE key = 'phase'", + [], + ) + .unwrap(); + drop(source_connection); + assert!(replication + .publish_checkpoint("source", 1, "checkpoint-1") + .await + .unwrap_err() + .to_string() + .contains("already contains a different database.sqlite")); + + let existing = replication + .activate(activation("existing", true)) + .await + .unwrap(); + { + let connection = rusqlite::Connection::open(&existing.path).unwrap(); + connection + .execute("CREATE TABLE occupied(value TEXT)", []) + .unwrap(); + } + replication.await_durable("existing", 1, 1).await.unwrap(); + let existing_target = replication + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "existing", false) + .await + .unwrap_err(); + assert!(existing_target + .to_string() + .contains("already has a durable replica")); + + let fork = replication + .activate(activation("fork", true)) + .await + .unwrap(); + assert!(fork.restored); + let connection = rusqlite::Connection::open(&fork.path).unwrap(); + let value: String = connection + .query_row("SELECT value FROM state WHERE key = 'phase'", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(value, "checkpointed"); + connection + .execute("UPDATE state SET value = 'forked' WHERE key = 'phase'", []) + .unwrap(); + drop(connection); + replication.await_durable("fork", 1, 1).await.unwrap(); + assert_eq!( + replication + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", true,) + .await + .unwrap(), + manifest + ); + + let source_connection = rusqlite::Connection::open(&source.path).unwrap(); + let source_value: String = source_connection + .query_row("SELECT value FROM state WHERE key = 'phase'", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(source_value, "source-advanced"); + } + + #[tokio::test] + async fn active_snapshots_use_independent_temporary_directories() { + let directory = tempfile::tempdir().unwrap(); + let store: Arc = Arc::new(InMemory::new()); + let replication = LtxRepl::start_with_store_for_test(directory.path(), store); + replication + .activate(activation("source", true)) + .await + .unwrap(); + + let first = replication.snapshot_active("source", 1).unwrap().unwrap(); + let second = replication.snapshot_active("source", 1).unwrap().unwrap(); + assert_ne!(first.path(), second.path()); + assert!(first.path().is_file()); + assert!(second.path().is_file()); + } + + #[tokio::test] + async fn incomplete_seed_never_activates_as_empty() { + let directory = tempfile::tempdir().unwrap(); + let store: Arc = Arc::new(InMemory::new()); + store + .put( + &ObjPath::from("cells/incomplete/fork-seed/reserved.json"), + PutPayload::from_static(b"{}"), + ) + .await + .unwrap(); + let replication = LtxRepl::start_with_store_for_test(directory.path(), store); + let error = match replication.activate(activation("incomplete", true)).await { + Ok(_) => panic!("incomplete seed activated"), + Err(error) => error, + }; + assert!(error.to_string().contains("reserved but incomplete")); + assert!(!directory + .path() + .join("incomplete/ltx/e1/db.sqlite") + .exists()); + } + + #[tokio::test] + async fn corrupt_seed_hash_never_activates() { + let directory = tempfile::tempdir().unwrap(); + let store: Arc = Arc::new(InMemory::new()); + let sqlite = b"not a sqlite database".to_vec(); + let manifest = ForkSeedManifest { + format: FORK_SEED_FORMAT.to_string(), + checkpoint_id: "checkpoint-corrupt-hash".to_string(), + source_cell: "source".to_string(), + source_epoch: 1, + sqlite_sha256: "0".repeat(64), + sqlite_bytes: sqlite.len() as u64, + }; + plant_fork_seed(&store, "corrupt-hash", &manifest, sqlite).await; + let replication = LtxRepl::start_with_store_for_test(directory.path(), store); + let error = match replication.activate(activation("corrupt-hash", true)).await { + Ok(_) => panic!("corrupt hash seed activated"), + Err(error) => error, + }; + assert!(error.to_string().contains("fork seed hash mismatch")); + assert!(!directory + .path() + .join("corrupt-hash/ltx/e1/db.sqlite") + .exists()); + } + + #[tokio::test] + async fn invalid_sqlite_seed_never_activates() { + let directory = tempfile::tempdir().unwrap(); + let store: Arc = Arc::new(InMemory::new()); + let sqlite = b"not a sqlite database".to_vec(); + let manifest = ForkSeedManifest { + format: FORK_SEED_FORMAT.to_string(), + checkpoint_id: "checkpoint-invalid-sqlite".to_string(), + source_cell: "source".to_string(), + source_epoch: 1, + sqlite_sha256: format!("{:x}", Sha256::digest(&sqlite)), + sqlite_bytes: sqlite.len() as u64, + }; + plant_fork_seed(&store, "invalid-sqlite", &manifest, sqlite).await; + let replication = LtxRepl::start_with_store_for_test(directory.path(), store); + let error = match replication + .activate(activation("invalid-sqlite", true)) + .await + { + Ok(_) => panic!("invalid SQLite seed activated"), + Err(error) => error, + }; + assert!(error + .to_string() + .contains("fork seed SQLite quick_check failed")); + assert!(!directory + .path() + .join("invalid-sqlite/ltx/e1/db.sqlite") + .exists()); + } +} diff --git a/crates/celld/main.rs b/crates/celld/main.rs index 75ef5e03f..12e062c7f 100644 --- a/crates/celld/main.rs +++ b/crates/celld/main.rs @@ -62,6 +62,15 @@ static NEXT_CORE_REQUEST: AtomicU64 = AtomicU64::new(1); /// the complete shutdown grace. const CONNECTION_DRAIN_GRACE: std::time::Duration = std::time::Duration::from_secs(2); +const CHECKPOINT_REQUEST_HEADER: &str = "x-celld-checkpoint-id"; +const CHECKPOINT_SOURCE_HEADER: &str = "x-celld-checkpoint-source"; +const CHECKPOINT_EPOCH_HEADER: &str = "x-celld-checkpoint-source-epoch"; +const CHECKPOINT_SHA256_HEADER: &str = "x-celld-checkpoint-sqlite-sha256"; +const CHECKPOINT_BYTES_HEADER: &str = "x-celld-checkpoint-sqlite-bytes"; +const FORK_SOURCE_HEADER: &str = "x-celld-fork-source"; +const FORK_CHECKPOINT_HEADER: &str = "x-celld-fork-checkpoint"; +const FORK_TARGET_HEADER: &str = "x-celld-fork-target"; + /// The lossy stdout writer's flush handle. Every exit path uses /// `std::process::exit`, which skips destructors — but the last lines before /// an exit are the fence forensics, exactly the lines that must survive. @@ -2475,6 +2484,85 @@ fn peer_response(mut response: HttpReply) -> HttpReply { response } +fn take_worker_header(response: &mut celld::js::HttpResponse, name: &str) -> Option { + let mut first = None; + response.headers.retain(|(candidate, value)| { + if candidate.eq_ignore_ascii_case(name) { + if first.is_none() { + first = Some(value.clone()); + } + false + } else { + true + } + }); + first +} + +async fn fulfill_checkpoint_request( + runtime: &RuntimeManager, + scope: &str, + response: &mut celld::js::HttpResponse, +) -> anyhow::Result<()> { + let Some(checkpoint_id) = take_worker_header(response, CHECKPOINT_REQUEST_HEADER) else { + return Ok(()); + }; + anyhow::ensure!( + (200..300).contains(&response.status), + "a failed Worker response cannot publish a checkpoint" + ); + anyhow::ensure!( + response.stream.is_none() && response.ws.is_none(), + "a checkpoint response must be buffered and non-WebSocket" + ); + let manifest = runtime.publish_checkpoint(scope, &checkpoint_id).await?; + response + .headers + .push((CHECKPOINT_SOURCE_HEADER.into(), manifest.source_cell)); + response.headers.push(( + CHECKPOINT_EPOCH_HEADER.into(), + manifest.source_epoch.to_string(), + )); + response + .headers + .push((CHECKPOINT_SHA256_HEADER.into(), manifest.sqlite_sha256)); + response.headers.push(( + CHECKPOINT_BYTES_HEADER.into(), + manifest.sqlite_bytes.to_string(), + )); + Ok(()) +} + +async fn fulfill_fork_request( + runtime: &RuntimeManager, + response: &mut celld::js::HttpResponse, +) -> anyhow::Result<()> { + let source = take_worker_header(response, FORK_SOURCE_HEADER); + let checkpoint = take_worker_header(response, FORK_CHECKPOINT_HEADER); + let target = take_worker_header(response, FORK_TARGET_HEADER); + if source.is_none() && checkpoint.is_none() && target.is_none() { + return Ok(()); + } + let (source, checkpoint, target) = match (source, checkpoint, target) { + (Some(source), Some(checkpoint), Some(target)) => (source, checkpoint, target), + _ => anyhow::bail!("incomplete Worker fork instruction"), + }; + anyhow::ensure!( + (200..300).contains(&response.status), + "a failed Worker response cannot seed a fork" + ); + anyhow::ensure!( + response.stream.is_none() && response.ws.is_none(), + "a fork response must be buffered and non-WebSocket" + ); + let source = runtime.cell_scope(&source)?; + let target = runtime.cell_scope(&target)?; + runtime + .publish_fork_seed_from_checkpoint(&source, &checkpoint, &target) + .await?; + Ok(()) +} + fn runtime_response(worker_response: celld::js::HttpResponse) -> HttpReply { let Ok(status) = StatusCode::from_u16(worker_response.status) else { return response(StatusCode::INTERNAL_SERVER_ERROR, "invalid Worker status"); @@ -2785,13 +2873,18 @@ async fn dispatch_do_call(app: AppHandle, call: DoCallReq) { // activity guard has not dropped), so it fails rather than // acknowledges a write the node cannot prove durable. let result = match result { - Ok(response) => match response.write_position.filter(|_| app.output_gate) { - Some(position) => match app.gate_write(request, position).await { - Ok(()) => Ok(response), - Err(error) => Err(anyhow::Error::new(RoutedRequestError(error))), - }, - None => Ok(response), - }, + Ok(mut response) => { + if let Some(position) = + response.write_position.filter(|_| app.output_gate) + { + app.gate_write(request, position).await.map_err(|error| { + anyhow::Error::new(RoutedRequestError(error)) + })?; + } + let runtime = app.runtime.as_ref().context("no cell runtime")?; + fulfill_checkpoint_request(runtime, &scope, &mut response).await?; + Ok(response) + } Err(error) => Err(error), }; if let Some(timing) = websocket_timing.as_mut() { @@ -3261,7 +3354,7 @@ async fn internal_do(request: Request, app: AppHandle, scope: String) }; match runtime .fetch_cell( - scope, + scope.clone(), name, RuntimeFetch { url, @@ -3278,7 +3371,7 @@ async fn internal_do(request: Request, app: AppHandle, scope: String) ) .await { - Ok(worker_response) => { + Ok(mut worker_response) => { abort.request_id = None; // Output gate (RPO=0): a peer-served handler that advanced // the cell's committed position holds its reply until the @@ -3297,6 +3390,14 @@ async fn internal_do(request: Request, app: AppHandle, scope: String) )); } } + if let Err(error) = + fulfill_checkpoint_request(runtime, &scope, &mut worker_response).await + { + return peer_response(response( + StatusCode::INTERNAL_SERVER_ERROR, + format!("checkpoint publication failed: {error:#}"), + )); + } if let Some(target) = &worker_response.ws { let kind = if celld::js::ws_hibernatable(target.id).unwrap_or(false) { WebSocketKind::Hibernatable @@ -3616,7 +3717,19 @@ async fn handle_ingress( .fetch_worker(url, method, body, headers, connection) .await { - Ok(worker_response) => runtime_response(worker_response), + Ok(mut worker_response) => { + let Some(runtime) = &app.runtime else { + return response(StatusCode::SERVICE_UNAVAILABLE, "no cell runtime"); + }; + if let Err(error) = fulfill_fork_request(runtime, &mut worker_response).await { + tracing::warn!(%error, "fork seed publication failed"); + return response( + StatusCode::INTERNAL_SERVER_ERROR, + "fork seed publication failed", + ); + } + runtime_response(worker_response) + } Err(error) => match error.downcast_ref::() { // Saturation is not a failure of the request. Answering it now // lets the caller retry or shed; holding the connection until its @@ -5125,3 +5238,33 @@ impl Drop for AbortPeerFetchOnHangUp { } } } + +#[cfg(test)] +mod worker_header_tests { + use super::*; + + #[test] + fn control_header_extraction_removes_every_duplicate() { + let mut response = celld::js::HttpResponse { + status: 200, + body: Vec::new(), + stream: None, + headers: vec![ + ("X-Celld-Checkpoint-Id".into(), "first".into()), + ("content-type".into(), "application/json".into()), + ("x-celld-checkpoint-id".into(), "second".into()), + ], + ws: None, + write_position: None, + }; + + assert_eq!( + take_worker_header(&mut response, CHECKPOINT_REQUEST_HEADER).as_deref(), + Some("first") + ); + assert_eq!( + response.headers, + vec![("content-type".into(), "application/json".into())] + ); + } +} diff --git a/crates/celld/runtime.rs b/crates/celld/runtime.rs index 50412368a..fcce16b74 100644 --- a/crates/celld/runtime.rs +++ b/crates/celld/runtime.rs @@ -14,8 +14,7 @@ use crate::wake::WakeFlusher; use anyhow::{anyhow, Context}; use futures_util::StreamExt as _; use serde::{Deserialize, Serialize}; -use std::collections::BTreeSet; -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::{Arc, Mutex, Once}; @@ -177,6 +176,51 @@ fn alarm_reporter(cells: &Arc>, observe: &AlarmObserver) -> struct CellRegistry { starting: HashMap, published: HashMap, + initializing: HashSet, +} + +impl CellRegistry { + fn reserve_initialization(&mut self, cell: &str) -> anyhow::Result<()> { + if self.starting.contains_key(cell) + || self.published.contains_key(cell) + || !self.initializing.insert(cell.to_string()) + { + return Err(anyhow!( + "cell runtime already exists or is initializing: {cell}" + )); + } + Ok(()) + } +} + +struct CellInitializationReservation { + cells: Arc>, + cell: String, +} + +impl CellInitializationReservation { + fn acquire(cells: &Arc>, cell: &str) -> anyhow::Result { + cells + .lock() + .expect("cell registry poisoned") + .reserve_initialization(cell)?; + Ok(Self { + cells: cells.clone(), + cell: cell.to_string(), + }) + } +} + +impl Drop for CellInitializationReservation { + fn drop(&mut self) { + let removed = self + .cells + .lock() + .expect("cell registry poisoned") + .initializing + .remove(&self.cell); + debug_assert!(removed, "cell initialization reservation was lost"); + } } #[derive(Clone)] @@ -330,6 +374,34 @@ impl Replication { self.ltx.restore_snapshot(cell).await } + async fn publish_checkpoint( + &self, + cell: &str, + epoch: u64, + checkpoint_id: &str, + ) -> anyhow::Result { + self.ltx + .publish_checkpoint(cell, epoch, checkpoint_id) + .await + } + + async fn publish_fork_seed_from_checkpoint( + &self, + source_cell: &str, + checkpoint_id: &str, + target_cell: &str, + target_active: bool, + ) -> anyhow::Result { + self.ltx + .publish_fork_seed_from_checkpoint( + source_cell, + checkpoint_id, + target_cell, + target_active, + ) + .await + } + async fn ensure_durable(&self, cell: &str, epoch: u64) -> anyhow::Result<()> { match self.sync_wait(cell, epoch).await { SyncWait::Durable => {} @@ -376,6 +448,35 @@ impl RuntimeManager { &self.region } + pub async fn publish_checkpoint( + &self, + cell: &str, + checkpoint_id: &str, + ) -> anyhow::Result { + let epoch = self + .published_epoch(cell) + .ok_or_else(|| anyhow!("checkpoint source is not published: {cell}"))?; + self.replication + .as_ref() + .ok_or_else(|| anyhow!("checkpointing requires durable replication"))? + .publish_checkpoint(cell, epoch, checkpoint_id) + .await + } + + pub async fn publish_fork_seed_from_checkpoint( + &self, + source_cell: &str, + checkpoint_id: &str, + target_cell: &str, + ) -> anyhow::Result { + let _target_reservation = CellInitializationReservation::acquire(&self.cells, target_cell)?; + self.replication + .as_ref() + .ok_or_else(|| anyhow!("forking requires durable replication"))? + .publish_fork_seed_from_checkpoint(source_cell, checkpoint_id, target_cell, false) + .await + } + /// A deployment with no Durable Object classes can never land a Worker fetch /// on a cell, so the core's round-robin routing always returns `None`. Lets /// the request path skip the core round-trip entirely for stateless workers. @@ -789,6 +890,8 @@ impl RuntimeManager { /// Materialize an isolate and retain it as non-routable until publication. pub async fn start_cell(&self, cell: String, epoch: u64, fresh: bool) -> anyhow::Result<()> { + let _initialization_reservation = + CellInitializationReservation::acquire(&self.cells, &cell)?; let db_path = self.db_path(&cell, epoch); let class = cell .split_once(':') @@ -853,9 +956,6 @@ impl RuntimeManager { { let mut cells = self.cells.lock().expect("cell registry poisoned"); - if cells.starting.contains_key(&cell) || cells.published.contains_key(&cell) { - return Err(anyhow!("cell runtime already exists: {cell}")); - } cells.starting.insert( cell.clone(), CellHandle { @@ -1949,3 +2049,54 @@ pub(crate) async fn drive_cell( fn path_text(path: &Path) -> &str { path.to_str().expect("celld data path must be UTF-8") } + +#[cfg(test)] +mod tests { + use super::{CellInitializationReservation, CellRegistry}; + use std::sync::{mpsc, Arc, Barrier, Mutex}; + use std::thread; + + #[test] + fn initialization_reservation_serializes_activation_and_fork_creation() { + let cells = Arc::new(Mutex::new(CellRegistry::default())); + let start = Arc::new(Barrier::new(3)); + let release_winner = Arc::new(Barrier::new(2)); + let (send, receive) = mpsc::channel(); + let mut contenders = Vec::new(); + for operation in ["activation", "fork"] { + let cells = cells.clone(); + let start = start.clone(); + let release_winner = release_winner.clone(); + let send = send.clone(); + contenders.push(thread::spawn(move || { + start.wait(); + match CellInitializationReservation::acquire(&cells, "Class:target") { + Ok(reservation) => { + send.send((operation, true)).expect("report winner"); + release_winner.wait(); + drop(reservation); + } + Err(error) => { + assert!(error + .to_string() + .contains("already exists or is initializing")); + send.send((operation, false)).expect("report loser"); + } + } + })); + } + drop(send); + start.wait(); + + let outcomes = [receive.recv().unwrap(), receive.recv().unwrap()]; + assert_eq!(outcomes.iter().filter(|(_, won)| *won).count(), 1); + assert_eq!(outcomes.iter().filter(|(_, won)| !*won).count(), 1); + release_winner.wait(); + for contender in contenders { + contender.join().expect("contender did not panic"); + } + + CellInitializationReservation::acquire(&cells, "Class:target") + .expect("reservation is released after publication"); + } +}