diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e0906a02f..32cf42fa9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1282,6 +1282,17 @@ jobs: --test s3_cluster \ -- --nocapture --test-threads=1 2>&1 | tee -a "$test_log" + # Keep the same package feature graph for this admitted-startup unit + # fixture; selecting only server would rebuild Lance and object_store. + cargo test --locked \ + -p omnigraph-engine \ + -p omnigraph-server \ + -p omnigraph-cluster \ + -p omnigraph-cli \ + --lib \ + settings::tests::s3_bootstrap_handoff_serves_empty_then_deploys_first_graph \ + -- --exact --nocapture --test-threads=1 2>&1 | tee -a "$test_log" + if grep -Eiq \ "SKIP public_physical_ref_token_rejects_s3_same_version_aba|skipping s3 " \ "$test_log" @@ -1300,6 +1311,7 @@ jobs: public_physical_ref_token_rejects_s3_same_version_aba \ s3::server_opens_s3_graph_directly_and_serves_snapshot_and_read \ s3::server_boots_cluster_from_bare_storage_uri_and_serves_query \ + settings::tests::s3_bootstrap_handoff_serves_empty_then_deploys_first_graph \ s3_cluster_full_lifecycle_import_apply_serve_evolve_delete \ cli_cluster_e2e::cluster_e2e_s3_live_apply_without_restart \ system_local::local_cli_s3_end_to_end_init_load_read_flow diff --git a/Cargo.lock b/Cargo.lock index 16e4197b7..25b48166f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5620,6 +5620,7 @@ dependencies = [ name = "omnigraph-cluster" version = "0.13.0" dependencies = [ + "async-trait", "omnigraph-compiler", "omnigraph-engine", "omnigraph-policy", diff --git a/changelog.d/bootstrap-serving-handoff.added.md b/changelog.d/bootstrap-serving-handoff.added.md new file mode 100644 index 000000000..f36b5f941 --- /dev/null +++ b/changelog.d/bootstrap-serving-handoff.added.md @@ -0,0 +1,3 @@ +- A fresh, empty S3 cluster initialized through `omnigraph-cluster::bootstrap_serving` can transfer its retained ownership to the first server with `--bootstrap-handoff FILE`. The bounded, exact receipt is checked before listening; failed or replayed claims never fall back to ordinary startup. See [first boot from a bootstrap receipt][bootstrap-serving-handoff]. + +[bootstrap-serving-handoff]: ../docs/user/clusters/index.md#first-boot-from-a-bootstrap-receipt diff --git a/crates/omnigraph-cluster/Cargo.toml b/crates/omnigraph-cluster/Cargo.toml index ea7268821..f5fb98005 100644 --- a/crates/omnigraph-cluster/Cargo.toml +++ b/crates/omnigraph-cluster/Cargo.toml @@ -35,6 +35,7 @@ tokio = { workspace = true } ulid = { workspace = true } [dev-dependencies] +async-trait = { workspace = true } serial_test = "3" tempfile = { workspace = true } tokio = { workspace = true } diff --git a/crates/omnigraph-cluster/src/admission.rs b/crates/omnigraph-cluster/src/admission.rs index 040dc6f45..9528175cc 100644 --- a/crates/omnigraph-cluster/src/admission.rs +++ b/crates/omnigraph-cluster/src/admission.rs @@ -48,6 +48,30 @@ struct AdmissionOwner { } impl ClusterAdmission { + /// Only the private policy-only bootstrap protocol can transfer an + /// already-held lock into an admission. Neither owner is ever unlocked. + pub(crate) fn from_bootstrap_lock( + store: ClusterStore, + guard: StateLockGuard, + serving_deployment: Option, + ) -> Self { + let purpose = if serving_deployment.is_some() { + ClusterAdmissionPurpose::Serve + } else { + ClusterAdmissionPurpose::Deployment + }; + Self(Arc::new(AdmissionOwner { + canonical_root: store + .canonical_root() + .expect("bootstrap has a validated S3 root"), + store, + guard, + schema_contracts: BTreeMap::new(), + serving_deployment, + purpose, + })) + } + pub fn canonical_root(&self) -> &str { &self.0.canonical_root } @@ -149,6 +173,12 @@ impl ClusterAdmission { /// outstanding clone, or abandonment retains the persisted lock. Operators /// must exclude concurrent force-unlock/release; the backend has no /// conditional-delete guarantee. + /// + /// After bootstrap transfer, this obligation includes earlier accepted + /// bootstrap control I/O, not only the current serving process. Otherwise + /// deleting this lock could let a delayed bootstrap create-if-absent + /// recreate its old lock. The transfer proves those attempts harmless + /// only while the lock remains continuously present. pub async fn release_after_settlement(self) -> Result<(), Diagnostic> { let owner = Arc::try_unwrap(self.0).map_err(|owner| { Diagnostic::error( diff --git a/crates/omnigraph-cluster/src/deployment.rs b/crates/omnigraph-cluster/src/deployment.rs index 5ec9d6a77..ffd39453c 100644 --- a/crates/omnigraph-cluster/src/deployment.rs +++ b/crates/omnigraph-cluster/src/deployment.rs @@ -12,6 +12,11 @@ use omnigraph::db::{ mod execution; pub use execution::*; +mod bootstrap; +pub use bootstrap::{ + BootstrapServingReceipt, MAX_BOOTSTRAP_SERVING_RECEIPT_BYTES, bootstrap_serving, + claim_bootstrap_serving, +}; pub(crate) const MAX_LEDGER_BYTES: usize = 16 * 1024 * 1024; pub const MAX_BUNDLE_BYTES: usize = 16 * 1024 * 1024; diff --git a/crates/omnigraph-cluster/src/deployment/bootstrap.rs b/crates/omnigraph-cluster/src/deployment/bootstrap.rs new file mode 100644 index 000000000..0ee474c0e --- /dev/null +++ b/crates/omnigraph-cluster/src/deployment/bootstrap.rs @@ -0,0 +1,520 @@ +//! A single initial policy deployment with no graph/native-I/O effects. +//! The retained lock is transferred by CAS, never removed or reconstructed. + +use super::*; +use crate::admission::ClusterAdmission; +use crate::state_lock::StateLockGuard; +use omnigraph_storage::StorageKind; + +#[cfg(test)] +#[path = "bootstrap_fault_tests.rs"] +mod fault_tests; + +pub const MAX_BOOTSTRAP_SERVING_RECEIPT_BYTES: usize = 16 * 1024; + +/// Exact native authority projected for the first serving owner. This is not +/// a bearer credential: claiming requires storage access and revalidates every +/// binding against the existing native ledger, immutable input and lock. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BootstrapServingReceipt { + pub version: u32, + pub canonical_root: String, + pub bootstrap_lock_id: String, + pub bootstrap_lock_version: String, + pub ledger_id: String, + pub state_revision: u64, + pub state_cas: String, + pub deployment_id: String, + pub input_digest: String, + pub config_digest: String, + pub result_revision: u64, +} + +impl BootstrapServingReceipt { + fn validate(&self) -> Result<(), Diagnostic> { + if self.version != 1 + || self.canonical_root.len() > 4096 + || self.bootstrap_lock_version.is_empty() + || self.bootstrap_lock_version.len() > 1024 + || self.bootstrap_lock_version.chars().any(char::is_control) + || Ulid::from_string(&self.bootstrap_lock_id).is_err() + || Ulid::from_string(&self.ledger_id).is_err() + || !authorization::valid_digest(&self.input_digest) + || !authorization::valid_digest(&self.config_digest) + || !self + .state_cas + .strip_prefix("sha256:") + .is_some_and(authorization::valid_digest) + || self.state_revision != 3 + || self.result_revision != 1 + || encoded_size(self)? > MAX_BOOTSTRAP_SERVING_RECEIPT_BYTES + { + return Err(refusal( + "bootstrap_receipt_invalid", + "invalid bootstrap handoff receipt", + )); + } + let id = DeploymentId::parse(&self.deployment_id)?; + if id.ledger != self.ledger_id || id.sequence != 1 { + return Err(refusal( + "bootstrap_receipt_invalid", + "receipt is not the first deployment", + )); + } + Ok(()) + } +} + +/// Initialize one fresh S3 cluster with cluster-scoped management policy only. +/// The private bootstrap owner cannot execute another deployment. Failure or +/// cancellation after acquisition retains its lock; repeat initialization +/// refuses. There is no unlock, deployment replay, graph open or background +/// engine work here. Transport retries keep their original preconditions. +pub async fn bootstrap_serving( + config_dir: impl AsRef, + caller: &DeploymentCaller, +) -> Result { + let bundle = capture_deployment(config_dir)?; + require_s3(&bundle.canonical_root)?; + let store = ClusterStore::for_storage_root(&bundle.canonical_root)?; + bootstrap_in_store(&store, &bundle, caller).await +} + +fn require_s3(root: &str) -> Result<(), Diagnostic> { + if omnigraph_storage::storage_kind_for_uri(root).ok() != Some(StorageKind::S3) { + return Err(refusal( + "bootstrap_handoff_backend_unsupported", + "bootstrap serving handoff requires an S3 backend with native conditional updates", + )); + } + Ok(()) +} + +fn validate_bootstrap_bundle( + bundle: &DeploymentBundle, + caller: &DeploymentCaller, +) -> Result<(), Diagnostic> { + execution::validate_bundle(bundle, &bundle.canonical_root)?; + caller.authority()?; + if bundle.resources.is_empty() + || bundle.resources.iter().any(|(address, resource)| { + !matches!(resource_kind(address), ResourceKind::Policy(_)) + || resource.applies_to.as_deref() != Some(&["cluster".to_owned()][..]) + || resource.embedding_provider.is_some() + || resource.embedding_profile.is_some() + || resource.external_blob_policy.is_some() + }) + { + return Err(refusal( + "bootstrap_policy_only", + "fresh serving bootstrap accepts cluster-scoped management policies only", + )); + } + execution::authorize_bootstrap_bundle(bundle, caller) +} + +async fn bootstrap_in_store( + store: &ClusterStore, + bundle: &DeploymentBundle, + caller: &DeploymentCaller, +) -> Result { + validate_bootstrap_bundle(bundle, caller)?; + store.require_fresh_bootstrap(false).await?; + let (guard, lock_version) = store.acquire_bootstrap_lock().await?; + // The lock is retained from this point, even if the second freshness read + // or any later validation, write or response is interrupted. + store.require_fresh_bootstrap(true).await?; + let state = execution::empty_ledger(); + store + .write_state(&state, None, &mut store.observations()) + .await?; + let admission = ClusterAdmission::from_bootstrap_lock(store.clone(), guard, None); + let mut effects_started = false; + let result = execution::execute_captured_deployment_in_store( + store, + bundle, + None, + caller, + &admission, + &BTreeMap::new(), + |_, _, _| {}, + |_| {}, + &mut effects_started, + ) + .await?; + let DeploymentLookup::Complete { result } = result else { + return Err(refusal( + "bootstrap_outcome_unknown", + "initial deployment is not terminal", + )); + }; + let (state, state_cas) = execution::read_existing(store).await?; + let receipt = BootstrapServingReceipt { + version: 1, + canonical_root: bundle.canonical_root.clone(), + bootstrap_lock_id: admission.lock_id().to_owned(), + bootstrap_lock_version: lock_version, + ledger_id: state.ledger_id.clone().unwrap(), + state_revision: state.state_revision, + state_cas, + deployment_id: result.id, + input_digest: result.input_digest, + config_digest: bundle.config_digest.clone(), + result_revision: result.result_revision, + }; + receipt.validate()?; + validate_terminal(store, &receipt).await?; + store + .verify_bootstrap_lock(&receipt.bootstrap_lock_id, &receipt.bootstrap_lock_version) + .await?; + // No effects follow terminal verification. Dropping this private owner + // retains the original lock; only the exact claim below can replace it. + Ok(receipt) +} + +/// Claim a retained fresh bootstrap lock for this process and return its exact +/// admitted serving snapshot. A failed/lost CAS acknowledgement grants no +/// admission. Replaying a receipt never adopts an already claimed lock. +pub async fn claim_bootstrap_serving( + root: &str, + receipt: &BootstrapServingReceipt, +) -> Result> { + let result = async { + receipt.validate()?; + require_s3(root)?; + let store = ClusterStore::for_storage_root(root)?; + if store.canonical_root()? != receipt.canonical_root { + return Err(refusal( + "bootstrap_root_mismatch", + "selected root differs from the handoff receipt", + )); + } + claim_in_store(&store, receipt).await + } + .await; + result.map_err(|error| vec![error]) +} + +async fn claim_in_store( + store: &ClusterStore, + receipt: &BootstrapServingReceipt, +) -> Result { + receipt.validate()?; + let result = validate_terminal(store, receipt).await?; + let guard: StateLockGuard = store + .claim_bootstrap_lock(&receipt.bootstrap_lock_id, &receipt.bootstrap_lock_version) + .await?; + // There is intentionally no read/adopt fallback after an uncertain CAS. + let admission = ClusterAdmission::from_bootstrap_lock(store.clone(), guard, Some(result)); + validate_terminal(store, receipt).await?; + let snapshot = serve::read_snapshot_with_store(store) + .await + .map_err(|mut errors| errors.remove(0))?; + if !snapshot.graphs.is_empty() + || !snapshot.applied_graphs.is_empty() + || !snapshot.queries.is_empty() + || !snapshot.diagnostics.is_empty() + || snapshot.state_revision != receipt.state_revision + || snapshot.state_cas.as_deref() != Some(receipt.state_cas.as_str()) + || snapshot.config_digest.as_deref() != Some(receipt.config_digest.as_str()) + { + return Err(refusal( + "bootstrap_snapshot_mismatch", + "claimed bootstrap snapshot changed", + )); + } + Ok(AdmittedServingSnapshot::from_bootstrap( + snapshot, + receipt.canonical_root.clone(), + admission, + )) +} + +async fn validate_terminal( + store: &ClusterStore, + receipt: &BootstrapServingReceipt, +) -> Result { + if store.canonical_root()? != receipt.canonical_root { + return Err(refusal("bootstrap_root_mismatch", "bootstrap root changed")); + } + let (state, cas) = execution::read_existing(store).await?; + let result = state + .deployment_results + .as_ref() + .filter(|results| results.len() == 1) + .and_then(|results| results.first()) + .ok_or_else(|| { + refusal( + "bootstrap_state_mismatch", + "expected exactly one terminal bootstrap result", + ) + })?; + if state.version != 2 + || state.ledger_id.as_deref() != Some(receipt.ledger_id.as_str()) + || state.state_revision != receipt.state_revision + || cas != receipt.state_cas + || state.next_sequence != Some(2) + || state.outstanding.is_some() + || !state + .applied_revision + .schema_contracts + .as_ref() + .is_some_and(BTreeMap::is_empty) + || state.applied_revision.result_revision != Some(receipt.result_revision) + || state.applied_revision.config_digest.as_deref() != Some(receipt.config_digest.as_str()) + || result.id != receipt.deployment_id + || result.input_digest != receipt.input_digest + || result.config_digest.as_deref() != Some(receipt.config_digest.as_str()) + || result.result_revision != receipt.result_revision + || !result.converged + || !result.graphs.is_empty() + || !result.recovery_executors.is_empty() + || result.base.result_revision != 0 + || !result.base.resource_digests.is_empty() + || !result.base.schema_contracts.is_empty() + || !state.recovery_records.is_empty() + || !state.approval_records.is_empty() + { + return Err(refusal( + "bootstrap_state_mismatch", + "native state differs from the exact initial policy deployment", + )); + } + let bundle = store.read_deployment_bundle(&receipt.input_digest).await?; + // The result's authority remains attribution, not a fresh identity. Input + // authorization occurred before bootstrap; claiming storage ownership + // does not execute or impersonate that initiating actor. + validate_bootstrap_bundle(&bundle, &DeploymentCaller::storage_owner(None))?; + if bundle.config_digest != receipt.config_digest + || bundle.canonical_root != receipt.canonical_root + || serde_json::to_value(&bundle.resources).unwrap() + != serde_json::to_value(&state.applied_revision.resources).unwrap() + { + return Err(refusal( + "bootstrap_state_mismatch", + "bootstrap input and achieved policy differ", + )); + } + authorization::AppliedPolicies::load(store, &state).await?; + store.require_no_bootstrap_graphs_or_recovery().await?; + Ok(result.clone()) +} + +#[cfg(test)] +mod tests { + use super::*; + use omnigraph_storage::StorageAdapter; + use std::sync::Arc; + + pub(super) const ROOT: &str = "s3://bootstrap-tests/root"; + const POLICY: &str = "version: 1\ngroups:\n owners: [principal:owner]\nrules:\n - id: configure\n allow: {actors: {group: owners}, actions: [config_manage]}\n"; + + pub(super) fn fixture() -> ( + tempfile::TempDir, + CapturedDeployment, + ClusterStore, + Arc, + ) { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("policy.yaml"), POLICY).unwrap(); + fs::write(dir.path().join(CLUSTER_CONFIG_FILE), format!( + "version: 1\nstorage: {ROOT}\npolicies:\n management:\n file: policy.yaml\n applies_to: [cluster]\n" + )).unwrap(); + let bundle = capture_deployment(dir.path()).unwrap(); + let adapter = super::fault_tests::memory_adapter(); + let store = ClusterStore::for_storage_root(ROOT) + .unwrap() + .with_bootstrap_test_adapter(adapter.clone()); + (dir, bundle, store, adapter) + } + + pub(super) fn owner() -> DeploymentCaller { + DeploymentCaller::storage_owner(Some("principal:owner".into())) + } + + pub(super) async fn bytes(adapter: &Arc, name: &str) -> String { + adapter.read_text(&format!("{ROOT}/{name}")).await.unwrap() + } + + #[tokio::test] + async fn bootstrap_claim_preserves_exact_result_and_state_without_unlock() { + let (_dir, bundle, store, adapter) = fixture(); + adapter + .write_text(&format!("{ROOT}/application/opaque.json"), "opaque") + .await + .unwrap(); + adapter + .write_text( + &format!("{ROOT}/__cluster/application_metadata.json"), + "opaque", + ) + .await + .unwrap(); + let receipt = bootstrap_in_store(&store, &bundle, &owner()).await.unwrap(); + assert_eq!(receipt.input_digest, bundle.input_digest().unwrap()); + let state = bytes(&adapter, CLUSTER_STATE_FILE).await; + let before = bytes(&adapter, CLUSTER_LOCK_FILE).await; + assert_eq!( + serde_json::from_str::(&before).unwrap()["operation"], + "bootstrap_serving" + ); + let captured = claim_in_store(&store, &receipt).await.unwrap(); + let (snapshot, root, admission) = captured.into_parts(); + assert_eq!(root, ROOT); + assert!(snapshot.graphs.is_empty()); + assert_eq!( + snapshot.state_cas.as_deref(), + Some(receipt.state_cas.as_str()) + ); + let admission = admission.unwrap(); + admission.validate_serving().unwrap(); + assert_ne!(admission.lock_id(), receipt.bootstrap_lock_id); + assert_eq!(bytes(&adapter, CLUSTER_STATE_FILE).await, state); + let lock = bytes(&adapter, CLUSTER_LOCK_FILE).await; + assert_ne!(lock, before); + drop(admission); + assert_eq!(bytes(&adapter, CLUSTER_LOCK_FILE).await, lock); + assert!(claim_in_store(&store, &receipt).await.is_err()); + assert!(bootstrap_in_store(&store, &bundle, &owner()).await.is_err()); + assert_eq!(bytes(&adapter, CLUSTER_STATE_FILE).await, state); + assert_eq!(bytes(&adapter, CLUSTER_LOCK_FILE).await, lock); + } + + #[tokio::test] + async fn receipt_mismatch_and_native_residue_refuse_without_claiming() { + let (_dir, bundle, store, adapter) = fixture(); + let receipt = bootstrap_in_store(&store, &bundle, &owner()).await.unwrap(); + let before = bytes(&adapter, CLUSTER_LOCK_FILE).await; + for changed in [ + BootstrapServingReceipt { + canonical_root: "s3://bootstrap-tests/other".into(), + ..receipt.clone() + }, + BootstrapServingReceipt { + input_digest: "0".repeat(64), + ..receipt.clone() + }, + BootstrapServingReceipt { + bootstrap_lock_version: "unknown".into(), + ..receipt.clone() + }, + BootstrapServingReceipt { + state_revision: 4, + ..receipt.clone() + }, + BootstrapServingReceipt { + version: 2, + ..receipt.clone() + }, + ] { + assert!(claim_in_store(&store, &changed).await.is_err()); + assert_eq!(bytes(&adapter, CLUSTER_LOCK_FILE).await, before); + } + adapter + .write_text( + &format!("{ROOT}/graphs/unrecorded.omni/__manifest/data"), + "residue", + ) + .await + .unwrap(); + assert!(claim_in_store(&store, &receipt).await.is_err()); + assert_eq!(bytes(&adapter, CLUSTER_LOCK_FILE).await, before); + let mut value = serde_json::to_value(&receipt).unwrap(); + value["extra"] = serde_json::json!(true); + assert!(serde_json::from_value::(value).is_err()); + } + + #[tokio::test] + async fn bootstrap_rejects_existing_native_authority_and_unauthorized_identity() { + for path in [ + CLUSTER_STATE_FILE, + CLUSTER_LOCK_FILE, + "__cluster/resources/unknown", + "__cluster/recoveries/old.json", + "graphs/old.omni/data", + "__manifest/_versions/1.manifest", + ] { + let (_dir, bundle, store, adapter) = fixture(); + adapter + .write_text(&format!("{ROOT}/{path}"), "existing") + .await + .unwrap(); + assert!( + bootstrap_in_store(&store, &bundle, &owner()).await.is_err(), + "{path}" + ); + assert_eq!( + adapter.read_text(&format!("{ROOT}/{path}")).await.unwrap(), + "existing" + ); + if path != CLUSTER_LOCK_FILE { + assert!( + !adapter + .exists(&format!("{ROOT}/{CLUSTER_LOCK_FILE}")) + .await + .unwrap() + ); + } + } + let (dir, bundle, store, adapter) = fixture(); + let denied = DeploymentCaller::AuthenticatedIdentity( + IdentityAuthorization::bootstrap_config_dir("principal:denied", dir.path()).unwrap(), + ); + assert!(bootstrap_in_store(&store, &bundle, &denied).await.is_err()); + let allowed = DeploymentCaller::AuthenticatedIdentity( + IdentityAuthorization::bootstrap_config_dir("principal:owner", dir.path()).unwrap(), + ); + validate_bootstrap_bundle(&bundle, &allowed).unwrap(); + fs::write( + dir.path().join(CLUSTER_CONFIG_FILE), + format!("version: 1\nstorage: {ROOT}\n"), + ) + .unwrap(); + let empty = capture_deployment(dir.path()).unwrap(); + assert_eq!( + bootstrap_in_store(&store, &empty, &owner()) + .await + .unwrap_err() + .code, + "bootstrap_policy_only" + ); + fs::write( + dir.path().join("people.pg"), + "node Person { name: String @key }\n", + ) + .unwrap(); + fs::write(dir.path().join(CLUSTER_CONFIG_FILE), format!( + "version: 1\nstorage: {ROOT}\ngraphs:\n people:\n schema: people.pg\npolicies:\n management:\n file: policy.yaml\n applies_to: [cluster]\n" + )).unwrap(); + let graphs = capture_deployment(dir.path()).unwrap(); + assert_eq!( + bootstrap_in_store(&store, &graphs, &owner()) + .await + .unwrap_err() + .code, + "bootstrap_policy_only" + ); + assert!(!adapter.exists(&format!("{ROOT}/__cluster")).await.unwrap()); + assert!(!adapter.exists(&format!("{ROOT}/graphs")).await.unwrap()); + assert!(require_s3("file:///tmp/bootstrap").is_err()); + assert!(require_s3("az://container/root").is_err()); + } + + #[tokio::test] + async fn changed_policy_payload_refuses_before_lock_claim() { + let (_dir, bundle, store, adapter) = fixture(); + let receipt = bootstrap_in_store(&store, &bundle, &owner()).await.unwrap(); + let lock = bytes(&adapter, CLUSTER_LOCK_FILE).await; + let digest = &bundle.resources["policy.management"].digest; + adapter + .write_text( + &format!("{ROOT}/__cluster/resources/policy/management/{digest}.yaml"), + "corrupt policy payload", + ) + .await + .unwrap(); + assert!(claim_in_store(&store, &receipt).await.is_err()); + assert_eq!(bytes(&adapter, CLUSTER_LOCK_FILE).await, lock); + } +} diff --git a/crates/omnigraph-cluster/src/deployment/bootstrap_fault_tests.rs b/crates/omnigraph-cluster/src/deployment/bootstrap_fault_tests.rs new file mode 100644 index 000000000..decaaa86b --- /dev/null +++ b/crates/omnigraph-cluster/src/deployment/bootstrap_fault_tests.rs @@ -0,0 +1,465 @@ +//! The in-memory backend supplies real atomic create/CAS semantics. This test +//! adapter records the actual requests and injects lost acknowledgements only +//! after the backend committed them; no production failpoint substitutes for +//! an uncertain storage response. +use super::*; +use omnigraph_storage::{ + ListDirBounds, ObjectStorageAdapter, Result as StorageResult, StorageAdapter, +}; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, +}; +use tokio::sync::{Barrier, Notify}; + +const URI_PREFIX: &str = "s3://bootstrap-tests/"; +const ROOT: &str = super::tests::ROOT; + +#[derive(Clone, Debug)] +struct Write { + uri: String, + payload: String, + expected_version: Option, + predecessor: Option<(String, String)>, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Point { + LockCreate, + StateCreate, + StateAccepted, + StateTerminal, + Claim, +} + +#[derive(Clone, Copy, Debug)] +enum Fault { + LostResponse, + CancelAfterCommit, +} + +#[derive(Debug)] +struct Controlled { + inner: ObjectStorageAdapter, + allow_fixture_setup: bool, + writes: Mutex>, + forbidden_calls: AtomicUsize, + fault: Mutex>, + committed: Notify, + claim_barrier: Mutex>>, + claim_attempts: AtomicUsize, +} + +impl Controlled { + fn new() -> Self { + Self { + inner: ObjectStorageAdapter::in_memory(), + allow_fixture_setup: false, + writes: Mutex::default(), + forbidden_calls: AtomicUsize::new(0), + fault: Mutex::default(), + committed: Notify::new(), + claim_barrier: Mutex::default(), + claim_attempts: AtomicUsize::new(0), + } + } + + fn key<'a>(&self, uri: &'a str) -> &'a str { + uri.strip_prefix(URI_PREFIX) + .expect("test-only S3 namespace") + } + + fn point(uri: &str, payload: &str, cas: bool) -> Option { + if uri.ends_with("/__cluster/lock.json") { + Some(if cas { Point::Claim } else { Point::LockCreate }) + } else if uri.ends_with("/__cluster/state.json") { + if !cas { + Some(Point::StateCreate) + } else { + match serde_json::from_str::(payload).unwrap()["state_revision"] + .as_u64() + { + Some(2) => Some(Point::StateAccepted), + Some(3) => Some(Point::StateTerminal), + _ => None, + } + } + } else { + None + } + } + + async fn after_commit(&self, point: Option) -> StorageResult<()> { + let fault = { + let mut active = self.fault.lock().unwrap(); + if active.is_some_and(|(expected, _)| Some(expected) == point) { + active.take().map(|(_, fault)| fault) + } else { + None + } + }; + match fault { + Some(Fault::LostResponse) => Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "injected response lost after backend commit", + ) + .into()), + Some(Fault::CancelAfterCommit) => { + self.committed.notify_one(); + std::future::pending::>().await + } + None => Ok(()), + } + } + + fn forbid(&self, operation: &str) -> StorageResult { + self.forbidden_calls.fetch_add(1, Ordering::SeqCst); + Err(std::io::Error::other(format!( + "forbidden bootstrap storage operation: {operation}" + )) + .into()) + } + + async fn read(&self, name: &str) -> String { + self.read_text(&format!("{ROOT}/{name}")).await.unwrap() + } + + fn arm(&self, point: Point, fault: Fault) { + *self.fault.lock().unwrap() = Some((point, fault)); + } + + async fn assert_no_forbidden_calls(&self) { + // A regression to asynchronous Drop cleanup must run before asserting + // the DELETE trap. In-memory operations otherwise need never yield. + tokio::task::yield_now().await; + assert_eq!(self.forbidden_calls.load(Ordering::SeqCst), 0); + } +} + +pub(super) fn memory_adapter() -> Arc { + let mut adapter = Controlled::new(); + adapter.allow_fixture_setup = true; + Arc::new(adapter) +} + +#[async_trait::async_trait] +impl StorageAdapter for Controlled { + async fn read_text(&self, uri: &str) -> StorageResult { + self.inner.read_text(self.key(uri)).await + } + async fn read_text_if_exists(&self, uri: &str) -> StorageResult> { + self.inner.read_text_if_exists(self.key(uri)).await + } + async fn read_text_if_exists_bounded( + &self, + uri: &str, + max_bytes: u64, + ) -> StorageResult> { + self.inner + .read_text_if_exists_bounded(self.key(uri), max_bytes) + .await + } + async fn read_bytes_if_exists_bounded( + &self, + uri: &str, + max_bytes: u64, + ) -> StorageResult>> { + self.inner + .read_bytes_if_exists_bounded(self.key(uri), max_bytes) + .await + } + async fn exists(&self, uri: &str) -> StorageResult { + self.inner.exists(self.key(uri)).await + } + async fn read_text_versioned(&self, uri: &str) -> StorageResult<(String, String)> { + self.inner.read_text_versioned(self.key(uri)).await + } + async fn read_text_versioned_if_exists_bounded( + &self, + uri: &str, + max_bytes: u64, + ) -> StorageResult> { + self.inner + .read_text_versioned_if_exists_bounded(self.key(uri), max_bytes) + .await + } + async fn list_dir(&self, uri: &str) -> StorageResult> { + self.inner.list_dir(self.key(uri)).await.map(|entries| { + entries + .into_iter() + .map(|entry| format!("{URI_PREFIX}{entry}")) + .collect() + }) + } + async fn list_dir_bounded( + &self, + uri: &str, + suffix: &str, + bounds: ListDirBounds, + ) -> StorageResult> { + self.inner + .list_dir_bounded(self.key(uri), suffix, bounds) + .await + .map(|entries| { + entries + .into_iter() + .map(|entry| format!("{URI_PREFIX}{entry}")) + .collect() + }) + } + async fn write_text(&self, uri: &str, contents: &str) -> StorageResult<()> { + if self.allow_fixture_setup { + self.inner.write_text(self.key(uri), contents).await + } else { + self.forbid("unconditional text PUT") + } + } + async fn write_bytes(&self, uri: &str, contents: &[u8]) -> StorageResult<()> { + if self.allow_fixture_setup { + self.inner.write_bytes(self.key(uri), contents).await + } else { + self.forbid("unconditional bytes PUT") + } + } + async fn rename_text(&self, _from: &str, _to: &str) -> StorageResult<()> { + self.forbid("rename") + } + async fn delete(&self, _uri: &str) -> StorageResult<()> { + self.forbid("DELETE") + } + async fn delete_prefix(&self, _uri: &str) -> StorageResult<()> { + self.forbid("prefix DELETE") + } + + async fn write_text_if_absent(&self, uri: &str, contents: &str) -> StorageResult { + let created = self + .inner + .write_text_if_absent(self.key(uri), contents) + .await?; + if created { + self.writes.lock().unwrap().push(Write { + uri: uri.into(), + payload: contents.into(), + expected_version: None, + predecessor: None, + }); + self.after_commit(Self::point(uri, contents, false)).await?; + } + Ok(created) + } + + async fn write_text_if_match( + &self, + uri: &str, + contents: &str, + expected: &str, + ) -> StorageResult> { + let predecessor = self.inner.read_text_versioned(self.key(uri)).await?; + if uri.ends_with("/__cluster/lock.json") { + self.claim_attempts.fetch_add(1, Ordering::SeqCst); + let barrier = self.claim_barrier.lock().unwrap().clone(); + if let Some(barrier) = barrier { + barrier.wait().await; + } + } + let version = self + .inner + .write_text_if_match(self.key(uri), contents, expected) + .await?; + if version.is_some() { + assert_eq!( + predecessor.1, expected, + "record the real backend predecessor" + ); + self.writes.lock().unwrap().push(Write { + uri: uri.into(), + payload: contents.into(), + expected_version: Some(expected.into()), + predecessor: Some(predecessor), + }); + self.after_commit(Self::point(uri, contents, true)).await?; + } + Ok(version) + } +} + +fn fixture() -> ( + tempfile::TempDir, + CapturedDeployment, + ClusterStore, + Arc, +) { + let (directory, bundle, store, _) = super::tests::fixture(); + let adapter = Arc::new(Controlled::new()); + let store = store.with_bootstrap_test_adapter(adapter.clone()); + (directory, bundle, store, adapter) +} + +#[tokio::test] +async fn backend_recorded_late_conditional_writes_cannot_replace_claimed_state() { + let (_directory, bundle, store, adapter) = fixture(); + let receipt = bootstrap_in_store(&store, &bundle, &super::tests::owner()) + .await + .unwrap(); + let bootstrap_writes = adapter.writes.lock().unwrap().clone(); + let state_writes: Vec<_> = bootstrap_writes + .iter() + .filter(|write| write.uri.ends_with(CLUSTER_STATE_FILE)) + .collect(); + assert_eq!(state_writes.len(), 3); + assert!(state_writes[0].expected_version.is_none()); + assert!( + state_writes[1..] + .iter() + .all(|write| write.predecessor.is_some()) + ); + let admitted = claim_in_store(&store, &receipt).await.unwrap(); + let state = adapter.read(CLUSTER_STATE_FILE).await; + let lock = adapter.read(CLUSTER_LOCK_FILE).await; + for write in &bootstrap_writes { + if let Some(expected) = &write.expected_version { + // Replay the exact old PUT, including the real predecessor token, + // directly against the backend. There is no fresh store pre-read. + assert!( + adapter + .inner + .write_text_if_match(adapter.key(&write.uri), &write.payload, expected) + .await + .unwrap() + .is_none() + ); + } else { + assert!( + !adapter + .inner + .write_text_if_absent(adapter.key(&write.uri), &write.payload) + .await + .unwrap() + ); + } + } + assert_eq!(adapter.read(CLUSTER_STATE_FILE).await, state); + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + drop(admitted); + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + adapter.assert_no_forbidden_calls().await; +} + +#[tokio::test] +async fn both_claimants_reach_cas_with_the_same_old_etag_but_only_one_wins() { + let (_directory, bundle, store, adapter) = fixture(); + let receipt = bootstrap_in_store(&store, &bundle, &super::tests::owner()) + .await + .unwrap(); + *adapter.claim_barrier.lock().unwrap() = Some(Arc::new(Barrier::new(2))); + let (one, two) = tokio::time::timeout(std::time::Duration::from_secs(5), async { + tokio::join!( + claim_in_store(&store, &receipt), + claim_in_store(&store, &receipt) + ) + }) + .await + .expect("both claimants must reach their exact-version CAS"); + *adapter.claim_barrier.lock().unwrap() = None; + assert_eq!(adapter.claim_attempts.load(Ordering::SeqCst), 2); + assert_eq!(usize::from(one.is_ok()) + usize::from(two.is_ok()), 1); + let lock = adapter.read(CLUSTER_LOCK_FILE).await; + drop((one, two)); + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + adapter.assert_no_forbidden_calls().await; +} + +#[tokio::test] +async fn committed_bootstrap_writes_with_lost_responses_retain_exclusion() { + for point in [ + Point::LockCreate, + Point::StateCreate, + Point::StateAccepted, + Point::StateTerminal, + ] { + let (_directory, bundle, store, adapter) = fixture(); + adapter.arm(point, Fault::LostResponse); + assert!( + bootstrap_in_store(&store, &bundle, &super::tests::owner()) + .await + .is_err() + ); + assert!( + adapter.fault.lock().unwrap().is_none(), + "fault must run after a real write" + ); + let lock = adapter.read(CLUSTER_LOCK_FILE).await; + assert!( + bootstrap_in_store(&store, &bundle, &super::tests::owner()) + .await + .is_err() + ); + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + adapter.assert_no_forbidden_calls().await; + } +} + +#[tokio::test] +async fn committed_claim_with_lost_response_grants_no_replay_admission() { + let (_directory, bundle, store, adapter) = fixture(); + let receipt = bootstrap_in_store(&store, &bundle, &super::tests::owner()) + .await + .unwrap(); + adapter.arm(Point::Claim, Fault::LostResponse); + assert!(claim_in_store(&store, &receipt).await.is_err()); + assert!(adapter.fault.lock().unwrap().is_none()); + let lock = adapter.read(CLUSTER_LOCK_FILE).await; + assert_ne!( + serde_json::from_str::(&lock).unwrap()["lock_id"], + receipt.bootstrap_lock_id + ); + assert!(claim_in_store(&store, &receipt).await.is_err()); + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + adapter.assert_no_forbidden_calls().await; +} + +#[tokio::test] +async fn cancellation_after_backend_commit_never_releases_or_adopts_ownership() { + for point in [ + Point::LockCreate, + Point::StateCreate, + Point::StateAccepted, + Point::StateTerminal, + Point::Claim, + ] { + let (_directory, bundle, store, adapter) = fixture(); + let caller = super::tests::owner(); + let receipt = if point == Point::Claim { + Some(bootstrap_in_store(&store, &bundle, &caller).await.unwrap()) + } else { + None + }; + adapter.arm(point, Fault::CancelAfterCommit); + { + let operation = async { + if let Some(receipt) = &receipt { + claim_in_store(&store, receipt).await.map(|_| ()) + } else { + bootstrap_in_store(&store, &bundle, &caller) + .await + .map(|_| ()) + } + }; + tokio::pin!(operation); + tokio::select! { + result = &mut operation => panic!("operation must wait after commit: {result:?}"), + () = adapter.committed.notified() => {}, + () = tokio::time::sleep(std::time::Duration::from_secs(5)) => panic!("injected storage commit was not reached"), + } + // Leaving the scope cancels the actual suspended storage future. + } + let lock = adapter.read(CLUSTER_LOCK_FILE).await; + if let Some(receipt) = &receipt { + assert!(claim_in_store(&store, receipt).await.is_err()); + } else { + assert!(bootstrap_in_store(&store, &bundle, &caller).await.is_err()); + } + assert_eq!(adapter.read(CLUSTER_LOCK_FILE).await, lock); + adapter.assert_no_forbidden_calls().await; + } +} diff --git a/crates/omnigraph-cluster/src/deployment/execution.rs b/crates/omnigraph-cluster/src/deployment/execution.rs index 0273b73c1..a9b5dc166 100644 --- a/crates/omnigraph-cluster/src/deployment/execution.rs +++ b/crates/omnigraph-cluster/src/deployment/execution.rs @@ -49,7 +49,7 @@ async fn policies( Ok(policies) } -fn authorize_bootstrap_bundle( +pub(super) fn authorize_bootstrap_bundle( bundle: &DeploymentBundle, caller: &DeploymentCaller, ) -> Result<(), Diagnostic> { @@ -601,7 +601,7 @@ fn bundle_from_capture( } } -fn validate_bundle(bundle: &DeploymentBundle, root: &str) -> Result<(), Diagnostic> { +pub(super) fn validate_bundle(bundle: &DeploymentBundle, root: &str) -> Result<(), Diagnostic> { validate_bundle_with_root(bundle, root, |declared| { ClusterStore::for_storage_root(declared)?.canonical_root() }) @@ -1619,6 +1619,32 @@ async fn execute_captured_deployment( effects_started: &mut bool, ) -> Result { let store = ClusterStore::for_storage_root(bundle.canonical_root())?; + execute_captured_deployment_in_store( + &store, + bundle, + requested_id, + caller, + admission, + live_graphs, + report_id, + on_accepted, + effects_started, + ) + .await +} + +#[allow(clippy::too_many_arguments)] // Same executor under the private bootstrap store/owner. +pub(super) async fn execute_captured_deployment_in_store( + store: &ClusterStore, + bundle: &CapturedDeployment, + requested_id: Option<&str>, + caller: &DeploymentCaller, + admission: &ClusterAdmission, + live_graphs: &BTreeMap>, + report_id: impl FnOnce(&str, &str, &str), + on_accepted: impl FnOnce(DeploymentLookup), + effects_started: &mut bool, +) -> Result { let root = store.canonical_root()?; validate_bundle(bundle, &root)?; if admission.canonical_root() != root { @@ -1631,12 +1657,12 @@ async fn execute_captured_deployment( let input_digest = bundle.input_digest()?; // Existing identities are lookup-only, including while their owner holds // the lock. Input equality is required before exposing the original result. - let (before, before_cas) = read_existing(&store).await?; + let (before, before_cas) = read_existing(store).await?; require_v2(&before)?; if let Some(id) = requested_id { let existing = lookup(&before, id)?; if !matches!(existing, DeploymentLookup::NotRecorded) { - lookup_policies(&store, &before, caller, Some(id)).await?; + lookup_policies(store, &before, caller, Some(id)).await?; let recorded_digest = before .outstanding .as_ref() @@ -1660,7 +1686,7 @@ async fn execute_captured_deployment( return Ok(existing); } } - deployment_policies(&store, &before, bundle, caller).await?; + deployment_policies(store, &before, bundle, caller).await?; let sequence = before.next_sequence.unwrap(); let id = requested_id.map(str::to_owned).unwrap_or_else(|| { format!( @@ -1685,7 +1711,7 @@ async fn execute_captured_deployment( policy, .. } = match prepare_deployment( - &store, + store, bundle, caller, (before, before_cas), @@ -1708,10 +1734,10 @@ async fn execute_captured_deployment( )); } seams::fail(&DEPLOYMENT_BEFORE_ACCEPTANCE)?; - cas = replace(&store, &mut state, &cas).await?; + cas = replace(store, &mut state, &cas).await?; on_accepted(lookup(&state, &id)?); seams::fail(&DEPLOYMENT_AFTER_ACCEPTANCE)?; - install_catalog_payloads(&store, &state, bundle).await?; + install_catalog_payloads(store, &state, bundle).await?; for (graph, entry) in state.outstanding.as_ref().unwrap().graphs.clone() { let result = if let Some(delete) = &entry.delete { state @@ -1722,9 +1748,9 @@ async fn execute_captured_deployment( .get_mut(&graph) .unwrap() .state = GraphDeploymentState::Started; - cas = replace(&store, &mut state, &cas).await?; + cas = replace(store, &mut state, &cas).await?; seams::fail(&DEPLOYMENT_AFTER_STARTED)?; - complete_graph_deletion(&store, &graph, delete, &id, admission).await?; + complete_graph_deletion(store, &graph, delete, &id, admission).await?; seams::fail(&DEPLOYMENT_AFTER_SCHEMA)?; GraphDeploymentResult::Deleted { contract: delete.contract.clone(), @@ -1738,7 +1764,7 @@ async fn execute_captured_deployment( .get_mut(&graph) .unwrap() .state = GraphDeploymentState::Started; - cas = replace(&store, &mut state, &cas).await?; + cas = replace(store, &mut state, &cas).await?; seams::fail(&DEPLOYMENT_AFTER_STARTED)?; let db = Omnigraph::apply_prepared_graph_create(create).await .map_err(|error| refusal("deployment_outcome_unknown", format!("deployment {id} graph creation remains outstanding under admission {}: {error}", admission.lock_id())))?; @@ -1755,7 +1781,7 @@ async fn execute_captured_deployment( let db = match live_graphs.get(&graph) { Some(db) => db.clone(), None => std::sync::Arc::new( - open_graph(&store, &graph, &policy, false) + open_graph(store, &graph, &policy, false) .await .map_err(|error| retained_error(error, admission.lock_id()))?, ), @@ -1768,7 +1794,7 @@ async fn execute_captured_deployment( .get_mut(&graph) .unwrap() .state = GraphDeploymentState::Started; - cas = replace(&store, &mut state, &cas).await?; + cas = replace(store, &mut state, &cas).await?; seams::fail(&DEPLOYMENT_AFTER_STARTED)?; let applied = db.apply_prepared_schema_as(intent, caller.actor()).await .map_err(|error| refusal("deployment_outcome_unknown", format!("deployment {id} remains outstanding under admission {}; schema invocation failed: {error}; establish quiescence and reconcile its original identity", admission.lock_id())))?; @@ -1802,9 +1828,9 @@ async fn execute_captured_deployment( .state = GraphDeploymentState::Settled { result: Box::new(result), }; - cas = replace(&store, &mut state, &cas).await?; + cas = replace(store, &mut state, &cas).await?; } - let result = finish(&store, &mut state, &cas, bundle).await?; + let result = finish(store, &mut state, &cas, bundle).await?; // Exact publication is not a proof that all accepted native/control I/O // has stopped. Retain admission until explicit operator quiescence/unlock. Ok(DeploymentLookup::Complete { result }) diff --git a/crates/omnigraph-cluster/src/serve.rs b/crates/omnigraph-cluster/src/serve.rs index b3c4b4b81..3705bb0da 100644 --- a/crates/omnigraph-cluster/src/serve.rs +++ b/crates/omnigraph-cluster/src/serve.rs @@ -100,6 +100,18 @@ pub struct AdmittedServingSnapshot { } impl AdmittedServingSnapshot { + pub(crate) fn from_bootstrap( + snapshot: ServingSnapshot, + canonical_root: String, + admission: crate::admission::ClusterAdmission, + ) -> Self { + Self { + snapshot, + canonical_root, + admission: Some(admission), + } + } + pub fn canonical_root(&self) -> &str { &self.canonical_root } diff --git a/crates/omnigraph-cluster/src/state_lock.rs b/crates/omnigraph-cluster/src/state_lock.rs index 38196307e..d543db2cf 100644 --- a/crates/omnigraph-cluster/src/state_lock.rs +++ b/crates/omnigraph-cluster/src/state_lock.rs @@ -45,6 +45,18 @@ pub(crate) struct StateLockFile { } impl StateLockFile { + fn new(operation: &str) -> Self { + Self { + version: LOCK_VERSION, + lock_id: Ulid::new().to_string(), + operation: operation.to_owned(), + created_at: OffsetDateTime::now_utc() + .format(&Rfc3339) + .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_owned()), + pid: process::id(), + } + } + pub(crate) fn parse(text: &str) -> Result { let lock: Self = serde_json::from_str(text).map_err(StateLockError::LockParse)?; if lock.version != LOCK_VERSION { @@ -80,7 +92,8 @@ pub(crate) enum StateLockAcquire { /// Exclusive persisted cluster-state lock. /// /// Private fields and the absence of `Clone` make ownership non-forgeable. -/// The only constructor performs the backend's atomic create-if-absent. +/// Constructors perform atomic create-if-absent or the bounded initial +/// bootstrap owner's exact-version conditional transfer. #[derive(Debug)] pub(crate) struct StateLockGuard { adapter: Arc, @@ -160,15 +173,7 @@ pub(crate) async fn acquire_state_lock( ) -> Result { let _validated_cluster_root = cluster_root_from_lock_uri(lock_uri)?; let adapter = storage.adapter(); - let lock = StateLockFile { - version: LOCK_VERSION, - lock_id: Ulid::new().to_string(), - operation: operation.to_string(), - created_at: OffsetDateTime::now_utc() - .format(&Rfc3339) - .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_string()), - pid: process::id(), - }; + let lock = StateLockFile::new(operation); let payload = serde_json::to_string_pretty(&lock).map_err(StateLockError::LockEncode)?; if adapter.write_text_if_absent(lock_uri, &payload).await? { return Ok(StateLockAcquire::Acquired(StateLockGuard { @@ -182,6 +187,115 @@ pub(crate) async fn acquire_state_lock( Ok(StateLockAcquire::Held) } +const BOOTSTRAP_OPERATION: &str = "bootstrap_serving"; + +/// No release path exists in initial bootstrap, including a lost write +/// acknowledgement before a guard could be constructed. +pub(crate) async fn acquire_bootstrap_lock( + adapter: Arc, + kind: StorageKind, + uri: &str, +) -> Result<(StateLockGuard, String), StateLockError> { + require_conditional_bootstrap(kind, uri)?; + let lock = StateLockFile::new(BOOTSTRAP_OPERATION); + let payload = serde_json::to_string_pretty(&lock).map_err(StateLockError::LockEncode)?; + if !adapter.write_text_if_absent(uri, &payload).await? { + return Err(StateLockError::InvalidBinding( + "bootstrap lock is already held".into(), + )); + } + let guard = StateLockGuard { + adapter: adapter.clone(), + uri: uri.into(), + kind, + lock, + release_on_drop: false, + }; + let (observed, version) = read_bootstrap_lock(&adapter, uri).await?; + if observed.lock_id != guard.lock_id() { + return Err(StateLockError::InvalidBinding( + "bootstrap owner changed".into(), + )); + } + Ok((guard, version)) +} + +fn require_conditional_bootstrap(kind: StorageKind, uri: &str) -> Result<(), StateLockError> { + cluster_root_from_lock_uri(uri)?; + if kind != StorageKind::S3 { + return Err(StateLockError::InvalidBinding( + "bootstrap handoff requires S3 conditional updates".into(), + )); + } + Ok(()) +} + +async fn read_bootstrap_lock( + adapter: &Arc, + uri: &str, +) -> Result<(StateLockFile, String), StateLockError> { + let (text, version) = adapter + .read_text_versioned_if_exists_bounded(uri, 64 * 1024) + .await? + .ok_or_else(|| StateLockError::InvalidBinding("bootstrap lock is absent".into()))?; + let lock = StateLockFile::parse(&text)?; + if lock.operation != BOOTSTRAP_OPERATION + || version.is_empty() + || version.len() > 1024 + || version.chars().any(char::is_control) + { + return Err(StateLockError::InvalidBinding( + "not a retained bootstrap lock and version".into(), + )); + } + Ok((lock, version)) +} + +pub(crate) async fn verify_bootstrap_lock( + adapter: &Arc, + uri: &str, + lock_id: &str, + expected_version: &str, +) -> Result<(), StateLockError> { + let (lock, version) = read_bootstrap_lock(adapter, uri).await?; + if lock.lock_id != lock_id || version != expected_version { + return Err(StateLockError::InvalidBinding( + "bootstrap lock identity or version changed".into(), + )); + } + Ok(()) +} + +pub(crate) async fn claim_bootstrap_lock( + adapter: Arc, + kind: StorageKind, + uri: &str, + lock_id: &str, + expected_version: &str, +) -> Result { + require_conditional_bootstrap(kind, uri)?; + verify_bootstrap_lock(&adapter, uri, lock_id, expected_version).await?; + let lock = StateLockFile::new("serve"); + let payload = serde_json::to_string_pretty(&lock).map_err(StateLockError::LockEncode)?; + // Never refresh the token or adopt a winner after an uncertain response. + if adapter + .write_text_if_match(uri, &payload, expected_version) + .await? + .is_none() + { + return Err(StateLockError::InvalidBinding( + "bootstrap lock claim lost its exact-version CAS".into(), + )); + } + Ok(StateLockGuard { + adapter, + uri: uri.into(), + kind, + lock, + release_on_drop: false, + }) +} + fn cluster_root_from_lock_uri(lock_uri: &str) -> Result { let suffix = format!("/{CLUSTER_LOCK_FILE}"); let root = lock_uri.strip_suffix(&suffix).ok_or_else(|| { diff --git a/crates/omnigraph-cluster/src/store.rs b/crates/omnigraph-cluster/src/store.rs index c82500dbd..ef988b9c0 100644 --- a/crates/omnigraph-cluster/src/store.rs +++ b/crates/omnigraph-cluster/src/store.rs @@ -48,6 +48,10 @@ pub(crate) struct StateSnapshot { pub(crate) state_cas: Option, } +fn bootstrap_lock_error(error: StateLockError) -> Diagnostic { + Diagnostic::error("bootstrap_lock_error", CLUSTER_LOCK_FILE, error.to_string()) +} + /// Only explicit stopped-ledger conversion may remove obsolete runtime fields. /// The remaining ledger is decoded by the ordinary strict types and validator. fn decode_ledger(text: &str, upgrade: bool) -> Result<(ClusterState, bool), Diagnostic> { @@ -211,6 +215,97 @@ impl ClusterStore { self.storage.kind() } + #[cfg(test)] + pub(crate) fn with_bootstrap_test_adapter(mut self, adapter: Arc) -> Self { + self.adapter = adapter; + self + } + + async fn require_absent_native_paths(&self, paths: &[&str]) -> Result<(), Diagnostic> { + for path in paths { + if self + .adapter + .exists(&self.uri(path)) + .await + .map_err(|error| { + Diagnostic::error("bootstrap_freshness_unknown", *path, error.to_string()) + })? + { + return Err(Diagnostic::error( + "bootstrap_root_not_fresh", + *path, + "existing native authority or graph data prevents initial bootstrap", + )); + } + } + Ok(()) + } + + pub(crate) async fn require_fresh_bootstrap(&self, own_lock: bool) -> Result<(), Diagnostic> { + self.require_no_bootstrap_graphs_or_recovery().await?; + self.require_absent_native_paths(&[CLUSTER_STATE_FILE, CLUSTER_RESOURCES_DIR]) + .await?; + if !own_lock { + self.require_absent_native_paths(&[CLUSTER_LOCK_FILE]) + .await?; + } + Ok(()) + } + + pub(crate) async fn require_no_bootstrap_graphs_or_recovery(&self) -> Result<(), Diagnostic> { + self.require_absent_native_paths(&[ + crate::CLUSTER_GRAPHS_DIR, + "__manifest", + "__history", + CLUSTER_RECOVERIES_DIR, + "__cluster/approvals", + ]) + .await + } + + pub(crate) async fn acquire_bootstrap_lock( + &self, + ) -> Result<(StateLockGuard, String), Diagnostic> { + crate::state_lock::acquire_bootstrap_lock( + self.adapter.clone(), + self.kind(), + &self.uri(CLUSTER_LOCK_FILE), + ) + .await + .map_err(bootstrap_lock_error) + } + + pub(crate) async fn verify_bootstrap_lock( + &self, + id: &str, + version: &str, + ) -> Result<(), Diagnostic> { + crate::state_lock::verify_bootstrap_lock( + &self.adapter, + &self.uri(CLUSTER_LOCK_FILE), + id, + version, + ) + .await + .map_err(bootstrap_lock_error) + } + + pub(crate) async fn claim_bootstrap_lock( + &self, + id: &str, + version: &str, + ) -> Result { + crate::state_lock::claim_bootstrap_lock( + self.adapter.clone(), + self.kind(), + &self.uri(CLUSTER_LOCK_FILE), + id, + version, + ) + .await + .map_err(bootstrap_lock_error) + } + /// Canonical identity of the same store used for the serving snapshot. /// This is metadata only; it does not reread the config or ledger. pub(crate) fn canonical_root(&self) -> Result { diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index 941e16ac2..d45e05bac 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -16,7 +16,8 @@ use handlers::*; use settings::*; pub use settings::{ ServerRuntimeState, classify_server_runtime_state, load_server_settings, - load_server_settings_with_data_token_trust, load_server_settings_with_identity_trust, + load_server_settings_with_bootstrap_handoff, load_server_settings_with_data_token_trust, + load_server_settings_with_identity_trust, }; pub mod auth; pub mod data_tokens; @@ -228,11 +229,11 @@ pub struct ServerConfig { pub cluster_admission: Option, } -/// Applied server settings paired with already validated offline token trust. +/// Applied server settings bound to a root and optional validated identity trust. /// -/// Constructed only by [`load_server_settings_with_data_token_trust`]. The -/// settings are exposed read-only so their graphs cannot be replaced after the -/// canonical serving root has been checked against the trust document. +/// Constructed by the identity-trust and bootstrap-handoff settings loaders. The +/// settings are exposed read-only so their graphs cannot be replaced after +/// ownership or identity trust has been bound to the canonical serving root. #[derive(Debug, Clone)] pub struct ManagedServerConfig { config: ServerConfig, @@ -2779,8 +2780,8 @@ pub async fn serve_with_mcp_settings(config: ServerConfig, mcp: McpSettings) -> serve_config(config, None, None, mcp).await } -/// Serve settings whose offline data-token trust was validated against their -/// applied snapshot's canonical root before any graph engine open. +/// Serve root-bound settings whose enabled identity trust was validated against +/// the applied snapshot's canonical root before any graph engine open. pub async fn serve_with_data_token_trust(config: ManagedServerConfig) -> Result<()> { serve_config(config.config, config.trust, config.oidc_trust, config.mcp).await } diff --git a/crates/omnigraph-server/src/main.rs b/crates/omnigraph-server/src/main.rs index 4fbe49d88..3f051ec59 100644 --- a/crates/omnigraph-server/src/main.rs +++ b/crates/omnigraph-server/src/main.rs @@ -3,8 +3,9 @@ use std::path::PathBuf; use clap::Parser; use color_eyre::eyre::Result; use omnigraph_server::{ - McpSettings, init_tracing, load_server_settings, load_server_settings_with_identity_trust, - resolve_shutdown_grace, serve_with_data_token_trust, serve_with_mcp_settings, + McpSettings, init_tracing, load_server_settings, load_server_settings_with_bootstrap_handoff, + load_server_settings_with_identity_trust, resolve_shutdown_grace, serve_with_data_token_trust, + serve_with_mcp_settings, }; #[derive(Debug, Parser)] @@ -22,6 +23,11 @@ struct Cli { /// The server's only boot source (RFC-011 cluster-only). #[arg(long)] cluster: Option, + /// Claim the exact receipt of a fresh empty S3 bootstrap. Administrator-only + /// boot input, at most 16 KiB of strict JSON. Not a restart or recovery path; + /// any failure retains ownership and requires operator investigation. + #[arg(long, requires = "cluster")] + bootstrap_handoff: Option, #[arg(long)] bind: Option, /// Public JSON trust for offline signed data credentials (RFC 0053). @@ -78,6 +84,22 @@ async fn main() -> Result<()> { cli.mcp_allow_writes, cli.mcp_allowed_hosts, )?; + if let Some(receipt_path) = cli.bootstrap_handoff.as_deref() { + let shutdown_grace = resolve_shutdown_grace(cli.shutdown_grace_seconds)?; + let settings = load_server_settings_with_bootstrap_handoff( + cli.cluster.as_ref(), + cli.bind, + cli.unauthenticated, + cli.require_all_graphs, + receipt_path, + cli.data_token_trust.as_deref(), + cli.oidc_identity_trust.as_deref(), + ) + .await? + .with_shutdown_grace(shutdown_grace) + .with_mcp_settings(mcp); + return serve_with_data_token_trust(settings).await; + } match (cli.data_token_trust, cli.oidc_identity_trust) { (data, oidc) if data.is_some() || oidc.is_some() => { let settings = load_server_settings_with_identity_trust( @@ -106,3 +128,35 @@ async fn main() -> Result<()> { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bootstrap_handoff_requires_an_explicit_cluster() { + assert!( + Cli::try_parse_from(["omnigraph-server", "--bootstrap-handoff", "receipt.json"]) + .is_err() + ); + let cli = Cli::try_parse_from([ + "omnigraph-server", + "--cluster", + "s3://example/cluster", + "--bootstrap-handoff", + "receipt.json", + "--data-token-trust", + "trust.json", + ]) + .unwrap(); + assert_eq!(cli.bootstrap_handoff, Some(PathBuf::from("receipt.json"))); + assert_eq!(cli.cluster, Some(PathBuf::from("s3://example/cluster"))); + assert_eq!(cli.data_token_trust, Some(PathBuf::from("trust.json"))); + assert!( + Cli::try_parse_from(["omnigraph-server", "--cluster", "./cluster"]) + .unwrap() + .bootstrap_handoff + .is_none() + ); + } +} diff --git a/crates/omnigraph-server/src/settings.rs b/crates/omnigraph-server/src/settings.rs index 5d20f975c..d97d060d2 100644 --- a/crates/omnigraph-server/src/settings.rs +++ b/crates/omnigraph-server/src/settings.rs @@ -3,6 +3,7 @@ //! modularization). use super::*; +use std::io::Read; use std::path::Path; /// Build serving settings from a cluster directory's applied revision @@ -364,6 +365,89 @@ pub async fn load_server_settings_with_identity_trust( let bound = omnigraph_cluster::admit_serving_snapshot(&cluster_arg) .await .map_err(|diagnostics| serving_snapshot_error(cluster_dir, &diagnostics))?; + settings_from_admitted( + cluster_dir, + cli_bind, + cli_allow_unauthenticated, + cli_require_all_graphs, + bound, + (data_trust_path, oidc_trust_path), + SettingsRefusal::ReleaseReadOnlyAdmission, + ) + .await +} + +/// Claim an exact fresh, empty S3 bootstrap receipt before constructing serving +/// settings. This is an administrator-supplied boot input, never HTTP authority. +/// Failure does not retry, fall back to ordinary admission, or remove the lock. +/// Optional identity profiles retain their ordinary canonical-root validation. +pub async fn load_server_settings_with_bootstrap_handoff( + cli_cluster: Option<&PathBuf>, + cli_bind: Option, + cli_allow_unauthenticated: bool, + cli_require_all_graphs: bool, + receipt_path: &Path, + data_trust_path: Option<&Path>, + oidc_trust_path: Option<&Path>, +) -> Result { + let cluster_dir = required_cluster(cli_cluster)?; + let receipt = read_bootstrap_handoff(receipt_path)?; + let bound = + omnigraph_cluster::claim_bootstrap_serving(&cluster_dir.to_string_lossy(), &receipt) + .await + .map_err(|diagnostics| serving_snapshot_error(cluster_dir, &diagnostics))?; + settings_from_admitted( + cluster_dir, + cli_bind, + cli_allow_unauthenticated, + cli_require_all_graphs, + bound, + (data_trust_path, oidc_trust_path), + SettingsRefusal::RetainClaimedAdmission, + ) + .await +} + +fn read_bootstrap_handoff(path: &Path) -> Result { + let mut options = fs::OpenOptions::new(); + options.read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + // A mistaken FIFO path must not block boot waiting for a writer. + options.custom_flags(libc::O_NONBLOCK); + } + let file = options + .open(path) + .wrap_err("failed to open bootstrap handoff receipt")?; + if !file.metadata()?.is_file() { + bail!("bootstrap handoff receipt must be a regular file"); + } + let limit = omnigraph_cluster::MAX_BOOTSTRAP_SERVING_RECEIPT_BYTES; + let mut bytes = Vec::new(); + file.take(limit as u64 + 1) + .read_to_end(&mut bytes) + .wrap_err("failed to read bootstrap handoff receipt")?; + if bytes.len() > limit { + bail!("bootstrap handoff receipt exceeds {limit} bytes"); + } + serde_json::from_slice(&bytes).wrap_err("invalid bootstrap handoff receipt JSON") +} + +enum SettingsRefusal { + ReleaseReadOnlyAdmission, + RetainClaimedAdmission, +} + +async fn settings_from_admitted( + cluster_dir: &Path, + cli_bind: Option, + cli_allow_unauthenticated: bool, + cli_require_all_graphs: bool, + bound: omnigraph_cluster::AdmittedServingSnapshot, + (data_trust_path, oidc_trust_path): (Option<&Path>, Option<&Path>), + refusal: SettingsRefusal, +) -> Result { let canonical_root = bound.canonical_root().to_string(); let (snapshot, _, admission) = bound.into_parts(); let validated: Result<_> = (|| { @@ -384,7 +468,16 @@ pub async fn load_server_settings_with_identity_trust( })(); let (mut config, trust, oidc_trust) = match validated { Ok(validated) => validated, - Err(error) => return Err(release_settings_refusal(error, admission).await), + Err(error) => { + return Err(match refusal { + SettingsRefusal::ReleaseReadOnlyAdmission => { + release_settings_refusal(error, admission).await + } + // Even a completed settings refusal must retain a claimed lock: + // delayed bootstrap attempts rely on this object never vanishing. + SettingsRefusal::RetainClaimedAdmission => error, + }); + } }; config.cluster_admission = admission; Ok(ManagedServerConfig { @@ -573,6 +666,344 @@ mod tests { use std::sync::atomic::AtomicBool; use tempfile::tempdir; + #[test] + fn bootstrap_handoff_reader_is_strict_and_bounded() { + let temp = tempdir().unwrap(); + let path = temp.path().join("receipt.json"); + let value = serde_json::json!({ + "version": 1, "canonical_root": "s3://example/cluster", + "bootstrap_lock_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV", + "bootstrap_lock_version": "\"etag\"", + "ledger_id": "01ARZ3NDEKTSV4RRFFQ69G5FAW", "state_revision": 3, + "state_cas": format!("sha256:{}", "a".repeat(64)), + "deployment_id": "01ARZ3NDEKTSV4RRFFQ69G5FAW:1:01ARZ3NDEKTSV4RRFFQ69G5FAX", + "input_digest": "b".repeat(64), "config_digest": "c".repeat(64), + "result_revision": 1 + }); + let mut exact = serde_json::to_vec(&value).unwrap(); + exact.resize(omnigraph_cluster::MAX_BOOTSTRAP_SERVING_RECEIPT_BYTES, b' '); + fs::write(&path, &exact).unwrap(); + assert_eq!( + super::read_bootstrap_handoff(&path).unwrap().canonical_root, + "s3://example/cluster" + ); + exact.push(b' '); + fs::write(&path, exact).unwrap(); + assert!( + super::read_bootstrap_handoff(&path) + .unwrap_err() + .to_string() + .contains("exceeds 16384 bytes") + ); + for invalid in [ + "{".to_string(), + "{}".to_string(), + format!("{value} {{}}"), + value.to_string().replacen('{', "{\"version\":1,", 1), + value + .to_string() + .replacen('{', "{\"unrecognized\":true,", 1), + ] { + fs::write(&path, invalid).unwrap(); + assert!( + super::read_bootstrap_handoff(&path) + .unwrap_err() + .to_string() + .contains("invalid bootstrap handoff receipt JSON") + ); + } + assert!(super::read_bootstrap_handoff(temp.path()).is_err()); + } + + #[tokio::test] + async fn malformed_bootstrap_handoff_never_falls_back_to_ordinary_boot() { + let temp = tempdir().unwrap(); + let cluster = temp.path().join("must-not-be-created"); + let receipt = temp.path().join("receipt.json"); + fs::write(&receipt, b"{}").unwrap(); + let error = super::load_server_settings_with_bootstrap_handoff( + Some(&cluster), + None, + true, + false, + &receipt, + None, + None, + ) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("invalid bootstrap handoff receipt JSON"), + "{error}" + ); + assert!(!cluster.exists()); + } + + /// Uses the same admitted startup and HTTP deployment path as `serve_config`. + /// S3 qualification is independently gated from the local settings suite. + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn s3_bootstrap_handoff_serves_empty_then_deploys_first_graph() { + use crate::api::{HTTP_API_CONTRACT, HTTP_API_CONTRACT_HEADER}; + use axum::body::{Body, to_bytes}; + use axum::http::{Request, StatusCode}; + use serde_json::{Value, json}; + use tower::ServiceExt; + + let Ok(bucket) = env::var("OMNIGRAPH_S3_TEST_BUCKET") else { + eprintln!("skipping S3 bootstrap serving test: OMNIGRAPH_S3_TEST_BUCKET is not set"); + return; + }; + async fn response( + app: &axum::Router, + method: &str, + path: &str, + body: Value, + ) -> (StatusCode, Value) { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(HTTP_API_CONTRACT_HEADER, HTTP_API_CONTRACT) + .header("authorization", "Bearer bootstrap-secret") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let code = response.status(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + (code, serde_json::from_slice(&bytes).unwrap()) + } + let unique = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let root = format!( + "s3://{bucket}/server-bootstrap/{}-{unique}", + std::process::id() + ); + let temp = tempdir().unwrap(); + let source = format!( + "version: 1\nstorage: {root}\npolicies:\n management:\n file: ./management.yaml\n applies_to: [cluster]\n" + ); + fs::write(temp.path().join("management.yaml"), "version: 1\ngroups:\n operators: [operator]\nrules:\n - id: manage\n allow:\n actors: {group: operators}\n actions: [config_manage]\n - id: discover\n allow:\n actors: {group: operators}\n actions: [graph_list]\n").unwrap(); + fs::write(temp.path().join("cluster.yaml"), &source).unwrap(); + let caller = omnigraph_cluster::DeploymentCaller::storage_owner(None); + let receipt = omnigraph_cluster::bootstrap_serving(temp.path(), &caller) + .await + .unwrap(); + let receipt_path = temp.path().join("receipt.json"); + fs::write(&receipt_path, serde_json::to_vec(&receipt).unwrap()).unwrap(); + let root_path = PathBuf::from(&root); + let storage = omnigraph::storage::storage_for_uri(&root).unwrap(); + let lock_uri = format!("{root}/__cluster/lock.json"); + let bootstrap_lock = storage.read_text(&lock_uri).await.unwrap(); + // A receipt may not choose a different root than the administrator did. + let wrong = PathBuf::from(format!("{root}-wrong")); + assert!( + super::load_server_settings_with_bootstrap_handoff( + Some(&wrong), + None, + false, + false, + &receipt_path, + None, + None, + ) + .await + .is_err() + ); + assert_eq!(storage.read_text(&lock_uri).await.unwrap(), bootstrap_lock); + for name in ["lock.json", "state.json"] { + assert!( + storage + .read_text_if_exists(&format!("{root}-wrong/__cluster/{name}")) + .await + .unwrap() + .is_none() + ); + } + + let settings = super::load_server_settings_with_bootstrap_handoff( + Some(&root_path), + None, + false, + false, + &receipt_path, + None, + None, + ) + .await + .unwrap(); + let config = settings.config; + let owner = config.cluster_admission.as_ref().unwrap(); + assert_ne!(owner.lock_id(), receipt.bootstrap_lock_id); + assert_eq!(owner.canonical_root(), receipt.canonical_root); + assert_eq!(config.witness.state_revision, receipt.state_revision); + assert_eq!( + config.witness.state_cas.as_deref(), + Some(receipt.state_cas.as_str()) + ); + let serving_lock = storage.read_text(&lock_uri).await.unwrap(); + assert!( + super::load_server_settings_with_bootstrap_handoff( + Some(&root_path), + None, + false, + false, + &receipt_path, + None, + None, + ) + .await + .is_err(), + "a receipt must be single-claim" + ); + assert_eq!(storage.read_text(&lock_uri).await.unwrap(), serving_lock); + + let ServerConfigMode::Multi { + graphs, + config_path, + server_policy, + } = config.mode; + assert!(graphs.is_empty()); + let state = super::open_multi_graph_state_admitted( + graphs, + vec![("operator".into(), "bootstrap-secret".into())], + server_policy.as_ref(), + config_path, + false, + config.cluster_admission, + ) + .await + .unwrap() + .with_boot_witness( + config.witness, + Arc::new(AtomicBool::new(false)), + DEFAULT_SHUTDOWN_GRACE, + ); + super::deployment::initialize_boot_activation(&state); + let app = super::build_app(state); + let (code, ready) = response(&app, "GET", "/readyz", Value::Null).await; + assert_eq!(code, StatusCode::OK, "{ready}"); + assert_eq!(ready["ready_graph_count"], 0); + let (code, inventory) = response(&app, "GET", "/graphs", Value::Null).await; + assert_eq!(code, StatusCode::OK, "{inventory}"); + assert!(inventory["graphs"].as_array().unwrap().is_empty()); + let (code, status) = response(&app, "GET", "/cluster/deployments", Value::Null).await; + assert_eq!(code, StatusCode::OK, "{status}"); + let status: omnigraph_cluster::DeploymentStatus = + serde_json::from_value(status["status"].clone()).unwrap(); + let id = status.next_deployment_id(); + fs::write( + temp.path().join("people.pg"), + "node Person { name: String @key }\n", + ) + .unwrap(); + fs::write( + temp.path().join("cluster.yaml"), + format!("{source}graphs:\n people:\n schema: ./people.pg\n"), + ) + .unwrap(); + let candidate = omnigraph_cluster::capture_deployment(temp.path()).unwrap(); + let (code, accepted) = response( + &app, + "POST", + "/cluster/deployments", + json!({"deployment_id":id,"deployment":candidate}), + ) + .await; + assert!( + matches!(code, StatusCode::OK | StatusCode::ACCEPTED), + "{code}: {accepted}" + ); + let completed = tokio::time::timeout(std::time::Duration::from_secs(60), async { + loop { + let (code, result) = response( + &app, + "GET", + &format!("/cluster/deployments/{id}"), + Value::Null, + ) + .await; + assert_eq!(code, StatusCode::OK, "{result}"); + if result["in_progress"] == false { + break result; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + }) + .await + .expect("first graph deployment must complete"); + assert_eq!(completed["active"], true, "{completed}"); + assert_eq!( + completed["deployment"]["result"]["converged"], true, + "{completed}" + ); + let (code, inventory) = response(&app, "GET", "/graphs", Value::Null).await; + assert_eq!(code, StatusCode::OK, "{inventory}"); + assert_eq!(inventory["graphs"][0]["graph_id"], "people"); + let (code, snapshot) = response(&app, "GET", "/graphs/people/snapshot", Value::Null).await; + assert_eq!(code, StatusCode::OK, "{snapshot}"); + assert!(snapshot["datasets"].is_array(), "{snapshot}"); + assert_eq!(storage.read_text(&lock_uri).await.unwrap(), serving_lock); + drop(app); + assert_eq!(storage.read_text(&lock_uri).await.unwrap(), serving_lock); + + // Invalid identity trust after a successful claim must retain the fresh + // owner too, even though startup never constructs a router/listener. + let refused_root = format!("{root}-trust-refused"); + fs::write( + temp.path().join("cluster.yaml"), + source.replace(&root, &refused_root), + ) + .unwrap(); + let receipt = omnigraph_cluster::bootstrap_serving(temp.path(), &caller) + .await + .unwrap(); + fs::write(&receipt_path, serde_json::to_vec(&receipt).unwrap()).unwrap(); + let trust = temp.path().join("bad-trust.json"); + fs::write(&trust, b"{}").unwrap(); + let refused_path = PathBuf::from(&refused_root); + assert!( + super::load_server_settings_with_bootstrap_handoff( + Some(&refused_path), + None, + false, + false, + &receipt_path, + Some(&trust), + None, + ) + .await + .is_err() + ); + let refused_uri = format!("{refused_root}/__cluster/lock.json"); + let retained = storage.read_text(&refused_uri).await.unwrap(); + let retained_json: Value = serde_json::from_str(&retained).unwrap(); + assert_ne!(retained_json["lock_id"], receipt.bootstrap_lock_id); + assert!( + super::load_server_settings_with_bootstrap_handoff( + Some(&refused_path), + None, + false, + false, + &receipt_path, + None, + None, + ) + .await + .is_err() + ); + assert_eq!(storage.read_text(&refused_uri).await.unwrap(), retained); + } + /// `authorize` returns the allow/deny **decision** (`Authz`) and reserves /// `Err` for operational failures, so the invoke handler can hide a denial /// as 404 without also masking a 401/500. Pins each outcome. diff --git a/docs/dev/control-plane.md b/docs/dev/control-plane.md index eff5aee66..b1964b12b 100644 --- a/docs/dev/control-plane.md +++ b/docs/dev/control-plane.md @@ -186,6 +186,26 @@ doors remain operator-excluded across outstanding work and recovery. No tags or additional retention store protect deployment evidence, and the executor never runs cleanup. +Fresh S3 initialization has one narrower exception to exact-ID unlock: +`bootstrap_serving` executes only the initial cluster-scoped policy deployment +under a private retained bootstrap lock. It opens no graphs and returns a strict +16 KiB maximum receipt bound to the exact root, input, native ledger/result, +state CAS and original lock version. Native authority, graph or recovery +residue refuses initialization; unrelated opaque metadata grants no authority. +`claim_bootstrap_serving` validates those bindings and conditionally replaces +that same lock version with a new process-owned serving lock before capturing +the empty serving snapshot. It never deletes the lock or adopts an earlier +claim after an uncertain acknowledgement. Errors, cancellation and lost +responses retain exclusion. Already-issued bootstrap state updates use spent +CAS predecessors; immutable payload retries cannot alter achieved input. This +is only fresh zero-graph bootstrap, not native-I/O settlement, existing-writer +replacement or expiry-based takeover. Later serving owners retain the ordinary +settled-release obligations above, including all earlier accepted bootstrap +control I/O. Settling only the current serving process is insufficient: deleting +its lock could let an older bootstrap create-if-absent request recreate the +bootstrap lock. Handoff itself proves those attempts harmless only while the +continuously retained lock exists. + Do not bypass the cluster API with direct filesystem writes, edit `state.json`, or derive a second mutable inventory. Content digests and live observations are recomputed from the declared and durable authorities. The distributed support boundary remains one mutation-capable writer process. diff --git a/docs/rfcs/2026-09-29-server-runtime-and-online-deployment.md b/docs/rfcs/2026-09-29-server-runtime-and-online-deployment.md index 8570af270..aea4db579 100644 --- a/docs/rfcs/2026-09-29-server-runtime-and-online-deployment.md +++ b/docs/rfcs/2026-09-29-server-runtime-and-online-deployment.md @@ -7,7 +7,7 @@ implementation: in-progress authors: - OmniGraph maintainers created: 2026-09-29 -updated: 2026-10-07 +updated: 2026-10-08 discussion: https://github.com/ModernRelay/omnigraph/pull/799 supersedes: - "0034" @@ -669,6 +669,62 @@ explicit managed-root deletion including history. They write no ledger or plan resource and consume no deployment sequence. Apply checks its fresh accepted base; an observed plan is neither a reservation nor permission to replay work. +### Fresh empty-cluster bootstrap handoff + +An opt-in storage-owner bootstrap may initialize a fresh S3 root with only +cluster-scoped management policies, then transfer its native lock to the first +serving process without deleting the lock. This is not recovery of a previous +writer, engine settlement, general restart, graph initialization or an expiring +lease. Local and Azure backends remain unsupported by this handoff: local CAS +is emulated, and Azure requires its separate admission qualification. + +`bootstrap_serving(config_dir, caller)` captures and validates the ordinary +bounded configuration and existing bootstrap authorization. It refuses graphs, +schemas, queries, providers, Blob bindings, existing ledgers, graph residue and +unknown native control state. Freshness covers native-owned authority and graph +namespaces; unrelated opaque application metadata is neither adopted nor used +as execution authority. It acquires one private bootstrap admission and retains +it across the initial ledger and the existing policy deployment protocol. +That capability never escapes to callers or permits a second deployment. No +graph engine opens. Every state transition increments the native revision and +uses its captured CAS; immutable inputs and policies use content-addressed +create-if-absent writes. No success, error, cancellation or destructor deletes +this admission. Only exact terminal convergence can produce the receipt. + +`BootstrapServingReceipt` is strict, versioned JSON bounded to 16 KiB. It binds +canonical root, bootstrap lock identity and exact backend version, ledger +identity, terminal state revision and content CAS, deployment identity, input +digest, applied configuration digest and result revision. It is a projection +of existing native authority, not another mutable ledger or bearer credential. +Loss before receipt delivery retains uncertainty; a repeat bootstrap refuses. + +`claim_bootstrap_serving(root, receipt)` checks the explicitly selected root, +bounded receipt, original immutable input, exact terminal zero-graph result, +management policy and bootstrap lock. It conditionally replaces that exact +lock version with a fresh, unpredictable serving-process lock identity. Two +claimants cannot both win; an error or lost acknowledgement never grants a +serving capability. A later process cannot adopt an observed serving lock or +repeat a claim by using its bytes. After confirmed CAS, the claimant revalidates +the terminal state and returns its admitted serving snapshot under the same +new guard. Any subsequent failure retains that guard. Server startup accepts +an explicit `--bootstrap-handoff FILE` alongside its selected `--cluster`; +ordinary startup, authentication and policy behavior stay unchanged. + +The old bootstrap can have no new transition after its terminal result. Late +retries of earlier ledger writes use already-spent CAS versions; immutable +payload retries cannot change content; a delayed lock create cannot replace an +existing lock. This narrow proof permits the ownership transfer without a +general native-I/O drain capability. Any unconditional write, delete, graph +effect, reused revision or retry against a refreshed CAS invalidates it. +Raw/older writers and administrative lock removal remain externally excluded. + +Qualification extends cluster admission/deployment tests with two claimants, +delayed spent CAS, lost claim acknowledgement, cancellation, exact input/root/ +policy binding, malformed receipts and refusal of repeated initialization. +S3-compatible backend tests establish actual conditional-write behavior; +provider deployment evidence remains separate. General S3 replacement and +undo retain the existing stopped-writer and accepted-I/O settlement boundary. + ## Online deployment `POST /cluster/plan` accepts a bounded captured bundle and returns the observed diff --git a/docs/user/clusters/index.md b/docs/user/clusters/index.md index 31babbc3b..bf99b1e62 100644 --- a/docs/user/clusters/index.md +++ b/docs/user/clusters/index.md @@ -78,6 +78,32 @@ through `cluster.yaml`; a root URI boots directly from applied resources. Editin local files alone never changes serving behavior. See [HTTP server](../operations/server.md) for authentication and routes. +### First boot from a bootstrap receipt + +Administrators using the `omnigraph-cluster` library can prepare a fresh S3 +cluster with `bootstrap_serving`, containing only its initial cluster management +policy. After that initializer has stopped, supply its exact receipt at first boot: + +```bash +OMNIGRAPH_SERVER_BEARER_TOKENS_JSON='{"act-alice":"secret"}' \ + omnigraph-server --cluster s3://company-data/company-brain \ + --bootstrap-handoff /run/omnigraph/bootstrap-receipt.json +``` + +The receipt must be a regular file containing strict JSON, at most 16 KiB. +It must match the selected root and its completed empty bootstrap. The server +claims ownership once before listening; replay, mismatched state and uncertain +claims refuse. Any failure after the claim retains ownership, including invalid +trust or server settings. Investigate that exact attempt; do not retry by +removing the lock or omitting the flag. + +The receipt supplies no user permissions. Configure static tokens or identity +trust normally; the applied policy controls who can create the first graph +through `cluster apply --server`. Protect the receipt and boot configuration as +administrator inputs. This path supports only first boot of a fresh S3 cluster; +it is not a restart, recovery or writer-fencing mechanism. Local and Azure roots +are unsupported, and other writers must remain excluded. + ## Deploy without restarting Edit and validate the bundle, then submit it to the running owner: diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 8735ac6ea..376faef13 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -37,6 +37,11 @@ start; see [cluster deployments](../clusters/index.md). An unapplied resource edit does not activate it, although changing or breaking the directory's config can change where boot looks for applied state. +For first boot of a fresh, empty S3 cluster initialized by the library, use +`--bootstrap-handoff FILE` with its exact receipt. This is a one-time ownership +claim, not a restart or recovery path; see +[first boot from a bootstrap receipt](../clusters/index.md#first-boot-from-a-bootstrap-receipt). + ### Logs The server logs to standard output at `info` unless `RUST_LOG` sets other