diff --git a/README.md b/README.md index c0ff2b0..3ef18d9 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,15 @@ Agents do not deploy daemons directly. They configure Idunn's target catalog, release targets, migration commands, and command boundaries so Idunn can run the shared rollout primitive and leave typed witnesses behind. +Deployment consequence has one final owner: immediately before every migration, +deploy, or restart process spawn, Idunn re-opens the root-owned deployment-brake +CultCache and the dedicated operator public anchor. A released record must bind +the canonical runtime (`yggdrasil` for `yggdrasil-local`), exact source revision +(or `restart:`), exact request id, operator, signature, and short expiry. +Missing, corrupt, engaged, expired, foreign, wrong-scope, substituted, or +badly-signed state fails closed. Planning and lifecycle-command publication do +not bypass or cache this decision. + Local package surfaces: - Organ contract: `docs/idunn.md` diff --git a/crates/idunn-daemon/src/main.rs b/crates/idunn-daemon/src/main.rs index 53877a1..486bb9b 100644 --- a/crates/idunn-daemon/src/main.rs +++ b/crates/idunn-daemon/src/main.rs @@ -5,16 +5,18 @@ use cultcache_rs::{ }; use cultmesh_rs::{CultMesh, CultMeshNode, CultMeshNodeOptions}; use cultnet_rs::{ - CultNetDocumentBinding, CultNetDocumentRegistry, CultNetMessage, CultNetRawDocumentRecord, - CultNetRawPayloadEncoding, CultNetReadOnlySnapshotPolicy, CultNetRudpPacketType, - CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, - CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, - GameCultProviderHealthIdentity, IdunnAuthenticatedProviderHealthProjectionPurpose, - IdunnServiceIdentity, IdunnSignedDaemonHealthPurpose, ServiceIdentitySignature, - ServiceIdentitySigner, decode_cultnet_message_from_slice, decode_rudp_packet, - derive_service_identity_id, encode_cultnet_message_to_vec, encode_rudp_packet, - open_service_identity_at, serve_read_only_raw_snapshot, - verify_service_identity_signature_with_public_key, + CultNetDocumentBinding, CultNetDocumentRegistry, CultNetMessage, CultNetRawDocumentRecord, + CultNetRawPayloadEncoding, CultNetReadOnlySnapshotPolicy, CultNetRudpPacketType, + CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, + CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, + GameCultProviderHealthIdentity, IDUNN_DEPLOYMENT_BRAKE_SCHEMA, + IdunnAuthenticatedProviderHealthProjectionPurpose, IdunnDeploymentBrakeObservation, + IdunnDeploymentBrakeOperatorIdentity, IdunnDeploymentBrakeRecord, IdunnServiceIdentity, + IdunnSignedDaemonHealthPurpose, ServiceIdentityProfile, ServiceIdentitySignature, + ServiceIdentitySigner, ServiceIdentityTrustAnchor, decode_cultnet_message_from_slice, + decode_rudp_packet, derive_service_identity_id, encode_cultnet_message_to_vec, + encode_rudp_packet, evaluate_idunn_deployment_brake, open_service_identity_at, + serve_read_only_raw_snapshot, verify_service_identity_signature_with_public_key, }; use ed25519_dalek::{Signature, Verifier, VerifyingKey}; use odin_core::{ @@ -35,7 +37,7 @@ use odin_core::{ authenticated_provider_health_reason_code, plan_keepalive, }; use sha2::{Digest, Sha256}; -use std::collections::{BTreeSet, HashMap}; +use std::collections::{BTreeSet, HashMap}; use std::env; use std::fs::{self, File}; use std::net::{SocketAddr, UdpSocket}; @@ -49,9 +51,9 @@ use std::time::{Instant, SystemTime, UNIX_EPOCH}; const CULTNET_RUDP_PROTOCOL_ID: &str = "cultnet.transport.rudp.v0"; const IDUNN_HEALTH_RUDP_CONNECTION_ID: u32 = 0x1d0d_0001; -const IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID: u32 = 0x1d0d_0002; -const IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID: &str = "idunn-daemon"; -const IDUNN_PUBLIC_HEALTH_QUERY_ROLE: &str = "authenticated-provider-health-projector"; +const IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID: u32 = 0x1d0d_0002; +const IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID: &str = "idunn-daemon"; +const IDUNN_PUBLIC_HEALTH_QUERY_ROLE: &str = "authenticated-provider-health-projector"; const EPIPHANY_SIGNED_RUNTIME_HEALTH_TYPE: &str = "epiphany.idunn_signed_runtime_health"; const EPIPHANY_SIGNED_RUNTIME_HEALTH_SCHEMA_VERSION: &str = "epiphany.idunn_signed_runtime_health.v0"; @@ -67,10 +69,10 @@ struct TargetActuationGate { impl TargetActuationGate { fn new() -> Self { - Self { - lock: Mutex::new(()), - reserved: AtomicBool::new(false), - } + Self { + lock: Mutex::new(()), + reserved: AtomicBool::new(false), + } } } const HOST_IDENTITY_TYPE: &str = "epiphany.host_identity_trust_anchor.v0"; @@ -504,13 +506,16 @@ fn locally_supervised_health_contract(id: &str, default_failure_state: &str) -> struct CommonOptions { store_path: PathBuf, release_authority_store_path: Option, + deployment_brake_store_path: Option, + deployment_brake_operator_anchor_path: Option, + deployment_brake_runtime_id: Option, operator_alarm_command: Option, rudp_health_bind: Option, trusted_epiphany_health_identity_store: Option, daemon_health_trust_store_path: Option, service_identity_store_path: Option, public_health_store_path: Option, - public_health_query_bind: Option, + public_health_query_bind: Option, execute: bool, command_timeout_seconds: u64, } @@ -732,7 +737,7 @@ fn run_swarm( ); } publish_runtime_transport_check(common, &store_lock, &now)?; - start_public_health_query_listener(common, &targets)?; + start_public_health_query_listener(common, &targets)?; start_rudp_health_ingress(common, &store_lock, &now)?; publish_surgery_plans(&options.profile, &targets, common, &store_lock, &now)?; @@ -742,12 +747,12 @@ fn run_swarm( let actuation_gates = Arc::new( targets .iter() - .map(|target| { - ( - target.daemon_id.clone(), - Arc::new(TargetActuationGate::new()), - ) - }) + .map(|target| { + ( + target.daemon_id.clone(), + Arc::new(TargetActuationGate::new()), + ) + }) .collect::>(), ); @@ -986,12 +991,12 @@ fn run_target_cycle( if plan.deployment_request.is_some() || plan.restart_request.is_some() { schedule_automatic_actuation( - target.clone(), - options.clone(), - Arc::clone(store_lock), - actuation_gate, - plan, - now, + target.clone(), + options.clone(), + Arc::clone(store_lock), + actuation_gate, + plan, + now, ); return Ok(()); } @@ -1018,21 +1023,21 @@ fn schedule_target_actuation( daemon_id: String, action: impl FnOnce() -> Result<()> + Send + 'static, ) { - if gate - .reserved - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() - { + if gate + .reserved + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { return; } thread::spawn(move || { let result = with_target_actuation_gate(&gate.lock, action); gate.reserved.store(false, Ordering::Release); if let Err(error) = result { - eprintln!( - "Idunn automatic actuation failed for {}: {}", - daemon_id, error - ); + eprintln!( + "Idunn automatic actuation failed for {}: {}", + daemon_id, error + ); } }); } @@ -1055,7 +1060,7 @@ fn finish_target_cycle( target .release .as_ref() - .and_then(|release| run_state_migration(target, release, now, options)) + .and_then(|release| run_state_migration(target, release, request, now, options)) } else { None }; @@ -1413,33 +1418,33 @@ fn persist_current_deployment_request( current_head_request_envelope, current_head_result_envelope, current_request_envelope, - ) = with_store_node(options, store_lock, |node| { - let head = node.get::(&request.daemon_id)?; - Ok(( - head.clone(), - node.cache() - .get_envelope::(&request.daemon_id)?, - head.as_ref() - .map(|head| { - node.cache() - .get_envelope::(&head.request_id) - }) - .transpose()? - .flatten(), - head.as_ref() - .map(|head| { - node.cache() - .get_envelope::(&format!( + ) = with_store_node(options, store_lock, |node| { + let head = node.get::(&request.daemon_id)?; + Ok(( + head.clone(), + node.cache() + .get_envelope::(&request.daemon_id)?, + head.as_ref() + .map(|head| { + node.cache() + .get_envelope::(&head.request_id) + }) + .transpose()? + .flatten(), + head.as_ref() + .map(|head| { + node.cache() + .get_envelope::(&format!( "result:{}", head.request_id )) - }) - .transpose()? - .flatten(), - node.cache() - .get_envelope::(&request.request_id)?, - )) - })?; + }) + .transpose()? + .flatten(), + node.cache() + .get_envelope::(&request.request_id)?, + )) + })?; if let Some(existing) = current_request_envelope.as_ref() { let decoded: IdunnDeploymentRequestRecord = rmp_serde::from_slice(&existing.payload)?; if decoded != *request { @@ -1478,10 +1483,10 @@ fn persist_current_deployment_request( if prior_result.result_id != format!("result:{}", current.request_id) || prior_result.request_id != current.request_id || prior_result.daemon_id != current.daemon_id - || !matches!( - prior_result.state.as_str(), - "succeeded" | "failed" | "skipped" - ) + || !matches!( + prior_result.state.as_str(), + "succeeded" | "failed" | "skipped" + ) { return Err(anyhow!( "current deployment request lacks an exact terminal result" @@ -1569,7 +1574,7 @@ fn terminalize_interrupted_deployment_requests( let request_matches = opening .iter() .filter(|entry| { - entry.r#type == IdunnDeploymentRequestRecord::TYPE && entry.key == head.request_id + entry.r#type == IdunnDeploymentRequestRecord::TYPE && entry.key == head.request_id }) .collect::>(); if request_matches.len() != 1 { @@ -2197,98 +2202,98 @@ struct IdunnProjectionPublisher { write_lock: Mutex<()>, } -#[derive(Clone)] -struct IdunnPublicHealthSnapshotServer { - public_store_path: PathBuf, - allowed_record_keys: BTreeSet, -} - -impl IdunnPublicHealthSnapshotServer { - fn new(public_store_path: PathBuf, targets: &[DaemonTarget]) -> Result { - let allowed_record_keys = targets - .iter() - .map(|target| { - provider_health_projection_key(&target.daemon_id, &target.health_contract.id) - }) - .collect::>(); - if allowed_record_keys.len() != targets.len() { - return Err(anyhow!( - "public health query target catalog contains a duplicate daemon/contract pair" - )); - } - let server = Self { - public_store_path, - allowed_record_keys, - }; - // Startup observes the exact public store before any worker starts. - // An absent store is an empty public projection, not private fallback. - server.raw_snapshot()?; - Ok(server) - } - - fn serve(&self, request: &CultNetMessage) -> Result { - let registry = public_health_document_registry(); - let mut policy = CultNetReadOnlySnapshotPolicy::new(); - for key in &self.allowed_record_keys { - policy.allow(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA, key)?; - } - serve_read_only_raw_snapshot(®istry, &policy, &self.raw_snapshot()?, request) - } - - fn raw_snapshot(&self) -> Result> { - let registry = public_health_document_registry(); - let entries = SingleFileMessagePackBackingStore::new(&self.public_store_path) - .pull_all_read_only_snapshot()?; - let mut records = Vec::with_capacity(entries.len()); - for entry in entries { - if entry.r#type != IdunnAuthenticatedProviderHealthProjectionRecord::TYPE - || entry.schema_id.as_deref() - != Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA) - { - return Err(anyhow!( - "public health query store contains a non-projection envelope" - )); - } - let projection: IdunnAuthenticatedProviderHealthProjectionRecord = - rmp_serde::from_slice(&entry.payload) - .context("decoding public health query projection")?; - if rmp_serde::to_vec(&projection)? != entry.payload - || projection.projection_id != entry.key - { - return Err(anyhow!( - "public health query projection is noncanonical or has a mismatched key" - )); - } - projection.validate()?; - let mut record = registry.raw_document_record_from_envelope(&entry)?; - record.source_runtime_id = Some(IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID.into()); - record.source_agent_id = None; - record.source_role = Some(IDUNN_PUBLIC_HEALTH_QUERY_ROLE.into()); - record.tags = Some(vec!["public-health".into()]); - records.push(record); - } - Ok(records) - } -} - -fn public_health_document_registry() -> CultNetDocumentRegistry { - let mut registry = CultNetDocumentRegistry::new(); - registry.register(CultNetDocumentBinding { - schema_id: IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), - document_type: IdunnAuthenticatedProviderHealthProjectionRecord::TYPE.into(), - mutation_contract: None, - payload_schema_version: Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into()), - }); - registry -} - -fn provider_health_projection_key(daemon_id: &str, health_contract: &str) -> String { - format!( - "provider-health:{:x}", - Sha256::digest([daemon_id.as_bytes(), b"\0", health_contract.as_bytes()].concat()) - ) -} - +#[derive(Clone)] +struct IdunnPublicHealthSnapshotServer { + public_store_path: PathBuf, + allowed_record_keys: BTreeSet, +} + +impl IdunnPublicHealthSnapshotServer { + fn new(public_store_path: PathBuf, targets: &[DaemonTarget]) -> Result { + let allowed_record_keys = targets + .iter() + .map(|target| { + provider_health_projection_key(&target.daemon_id, &target.health_contract.id) + }) + .collect::>(); + if allowed_record_keys.len() != targets.len() { + return Err(anyhow!( + "public health query target catalog contains a duplicate daemon/contract pair" + )); + } + let server = Self { + public_store_path, + allowed_record_keys, + }; + // Startup observes the exact public store before any worker starts. + // An absent store is an empty public projection, not private fallback. + server.raw_snapshot()?; + Ok(server) + } + + fn serve(&self, request: &CultNetMessage) -> Result { + let registry = public_health_document_registry(); + let mut policy = CultNetReadOnlySnapshotPolicy::new(); + for key in &self.allowed_record_keys { + policy.allow(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA, key)?; + } + serve_read_only_raw_snapshot(®istry, &policy, &self.raw_snapshot()?, request) + } + + fn raw_snapshot(&self) -> Result> { + let registry = public_health_document_registry(); + let entries = SingleFileMessagePackBackingStore::new(&self.public_store_path) + .pull_all_read_only_snapshot()?; + let mut records = Vec::with_capacity(entries.len()); + for entry in entries { + if entry.r#type != IdunnAuthenticatedProviderHealthProjectionRecord::TYPE + || entry.schema_id.as_deref() + != Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA) + { + return Err(anyhow!( + "public health query store contains a non-projection envelope" + )); + } + let projection: IdunnAuthenticatedProviderHealthProjectionRecord = + rmp_serde::from_slice(&entry.payload) + .context("decoding public health query projection")?; + if rmp_serde::to_vec(&projection)? != entry.payload + || projection.projection_id != entry.key + { + return Err(anyhow!( + "public health query projection is noncanonical or has a mismatched key" + )); + } + projection.validate()?; + let mut record = registry.raw_document_record_from_envelope(&entry)?; + record.source_runtime_id = Some(IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID.into()); + record.source_agent_id = None; + record.source_role = Some(IDUNN_PUBLIC_HEALTH_QUERY_ROLE.into()); + record.tags = Some(vec!["public-health".into()]); + records.push(record); + } + Ok(records) + } +} + +fn public_health_document_registry() -> CultNetDocumentRegistry { + let mut registry = CultNetDocumentRegistry::new(); + registry.register(CultNetDocumentBinding { + schema_id: IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), + document_type: IdunnAuthenticatedProviderHealthProjectionRecord::TYPE.into(), + mutation_contract: None, + payload_schema_version: Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into()), + }); + registry +} + +fn provider_health_projection_key(daemon_id: &str, health_contract: &str) -> String { + format!( + "provider-health:{:x}", + Sha256::digest([daemon_id.as_bytes(), b"\0", health_contract.as_bytes()].concat()) + ) +} + #[derive(Clone, Debug, PartialEq, Eq)] struct AuthenticatedProviderHealthSource { health: IdunnDaemonHealthRecord, @@ -2315,8 +2320,8 @@ fn initialize_projection_publisher( options.service_identity_store_path.as_deref(), options.public_health_store_path.as_deref(), ) else { - if options.service_identity_store_path.is_some() - || options.public_health_store_path.is_some() + if options.service_identity_store_path.is_some() + || options.public_health_store_path.is_some() { return Err(anyhow!( "--service-identity-store and --public-health-store must be configured together" @@ -2345,11 +2350,11 @@ fn initialize_projection_publisher( }))) } -fn reject_store_aliases( - options: &CommonOptions, - identity_path: &Path, - public_path: &Path, -) -> Result<()> { +fn reject_store_aliases( + options: &CommonOptions, + identity_path: &Path, + public_path: &Path, +) -> Result<()> { let identity = normalized_store_path(identity_path)?; let public = normalized_store_path(public_path)?; let mut private_paths = vec![normalized_store_path(&options.store_path)?, identity]; @@ -2363,12 +2368,12 @@ fn reject_store_aliases( { private_paths.push(normalized_store_path(path)?); } - if private_paths.iter().enumerate().any(|(index, path)| { - private_paths - .iter() - .skip(index + 1) - .any(|other| other == path) - }) { + if private_paths.iter().enumerate().any(|(index, path)| { + private_paths + .iter() + .skip(index + 1) + .any(|other| other == path) + }) { return Err(anyhow!( "service identity, operational, trust, and release stores must not alias" )); @@ -2411,12 +2416,12 @@ impl IdunnProjectionPublisher { ) -> Result<()> { let evaluated_at_unix_millis = parse_timestamp_millis(evaluated_at) .ok_or_else(|| anyhow!("projection evaluation time is invalid"))?; - let current = - read_fresh_daemon_published_health(options, store_lock, desired, evaluated_at)? - .and_then(|managed| managed.projection_source) - .ok_or_else(|| { - anyhow!("authenticated provider health vanished before projection") - })?; + let current = + read_fresh_daemon_published_health(options, store_lock, desired, evaluated_at)? + .and_then(|managed| managed.projection_source) + .ok_or_else(|| { + anyhow!("authenticated provider health vanished before projection") + })?; if ¤t != source { return Err(anyhow!( "authenticated provider health changed before projection publication" @@ -2430,13 +2435,13 @@ impl IdunnProjectionPublisher { .observed_at_unix_millis .saturating_add(u64::from(desired.max_silence_seconds).saturating_mul(1000)); if expires_at_unix_millis <= evaluated_at_unix_millis { - return Err(anyhow!( - "authenticated provider health expired before projection" - )); + return Err(anyhow!( + "authenticated provider health expired before projection" + )); } - let projection_key = provider_health_projection_key( - &source.health.daemon_id, - &source.health.health_contract, + let projection_key = provider_health_projection_key( + &source.health.daemon_id, + &source.health.health_contract, ); let _write_guard = self .write_lock @@ -2450,41 +2455,41 @@ impl IdunnProjectionPublisher { || entry.schema_id.as_deref() != Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA) }) { - return Err(anyhow!( - "public health store contains a non-projection envelope" - )); + return Err(anyhow!( + "public health store contains a non-projection envelope" + )); } for entry in &entries { self.verify_existing_projection(entry)?; } - let current_envelope = entries - .iter() - .find(|entry| entry.key == projection_key) - .cloned(); + let current_envelope = entries + .iter() + .find(|entry| entry.key == projection_key) + .cloned(); let current_projection = current_envelope .as_ref() - .map(|entry| { - rmp_serde::from_slice::( - &entry.payload, - ) - }) + .map(|entry| { + rmp_serde::from_slice::( + &entry.payload, + ) + }) .transpose() .context("decoding existing public health projection")?; if let Some(existing) = current_projection.as_ref() { existing.validate()?; - if current_envelope - .as_ref() - .is_some_and(|entry| entry.key != existing.projection_id) + if current_envelope + .as_ref() + .is_some_and(|entry| entry.key != existing.projection_id) || existing.projection_id != projection_key { - return Err(anyhow!( - "public projection envelope key does not match its payload" - )); + return Err(anyhow!( + "public projection envelope key does not match its payload" + )); } if existing.idunn_signer_identity_id != self.signer.entry().identity_id { - return Err(anyhow!( - "public projection signer identity changed without store rotation" - )); + return Err(anyhow!( + "public projection signer identity changed without store rotation" + )); } if existing.projection_incarnation_id == self.incarnation_id && existing.signed_health_sha256 == source.statement_sha256 @@ -2502,15 +2507,15 @@ impl IdunnProjectionPublisher { } let mut projection = IdunnAuthenticatedProviderHealthProjectionRecord { schema_version: IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), - projection_id: projection_key.clone(), - daemon_id: source.health.daemon_id.clone(), + projection_id: projection_key.clone(), + daemon_id: source.health.daemon_id.clone(), health_contract: source.health.health_contract.clone(), - provider_state: source.health.state.clone(), - reason_code: reason_code.into(), + provider_state: source.health.state.clone(), + reason_code: reason_code.into(), provider_observed_at_unix_millis: source.admission.observed_at_unix_millis, admitted_at_unix_millis: source.admission.admitted_at_unix_millis, - evaluated_at_unix_millis, - trust_binding_id: source.binding.binding_id.clone(), + evaluated_at_unix_millis, + trust_binding_id: source.binding.binding_id.clone(), trust_binding_sha256: source.binding_sha256.clone(), signed_health_sha256: source.statement_sha256.clone(), authenticated_admission_sha256: source.admission_sha256.clone(), @@ -2523,25 +2528,25 @@ impl IdunnProjectionPublisher { deployment_id: source.admission.deployment_id.clone(), idunn_runtime_id: "idunn-daemon".into(), idunn_signer_identity_id: self.signer.entry().identity_id.clone(), - projection_incarnation_id: self.incarnation_id.clone(), - projection_sequence: sequence, - signature_algorithm: "ed25519".into(), - signature: Vec::new(), - private_state_exposed: false, - expires_at_unix_millis, + projection_incarnation_id: self.incarnation_id.clone(), + projection_sequence: sequence, + signature_algorithm: "ed25519".into(), + signature: Vec::new(), + private_state_exposed: false, + expires_at_unix_millis, }; let unsigned = rmp_serde::to_vec(&projection)?; - projection.signature = self - .signer - .sign::(&unsigned) - .signature; + projection.signature = self + .signer + .sign::(&unsigned) + .signature; projection.validate()?; self.require_incarnation_trust_snapshot(options)?; let replacement = CultCacheEnvelope { key: projection_key.clone(), r#type: IdunnAuthenticatedProviderHealthProjectionRecord::TYPE.into(), - payload: rmp_serde::to_vec(&projection)?, - stored_at: evaluated_at.into(), + payload: rmp_serde::to_vec(&projection)?, + stored_at: evaluated_at.into(), schema_id: Some(IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into()), }; let expected = CultCacheExpectedEnvelope { @@ -2549,19 +2554,19 @@ impl IdunnProjectionPublisher { r#type: IdunnAuthenticatedProviderHealthProjectionRecord::TYPE.into(), current: current_envelope, }; - if backing.compare_exchange(&[expected], &[replacement])? { - return Ok(()); - } + if backing.compare_exchange(&[expected], &[replacement])? { + return Ok(()); + } } - Err(anyhow!( - "public health projection lost repeated cross-process contention" - )) + Err(anyhow!( + "public health projection lost repeated cross-process contention" + )) } fn require_incarnation_trust_snapshot(&self, options: &CommonOptions) -> Result<()> { - let path = options - .daemon_health_trust_store_path - .as_deref() + let path = options + .daemon_health_trust_store_path + .as_deref() .ok_or_else(|| anyhow!("projection publisher lost its root trust store"))?; if trust_store_snapshot_sha256(path)? != self.root_trust_snapshot_sha256 { return Err(anyhow!( @@ -2576,9 +2581,9 @@ impl IdunnProjectionPublisher { rmp_serde::from_slice(&entry.payload) .context("decoding public health projection during store audit")?; if rmp_serde::to_vec(&projection)? != entry.payload { - return Err(anyhow!( - "public health projection is not canonical MessagePack" - )); + return Err(anyhow!( + "public health projection is not canonical MessagePack" + )); } projection.validate()?; if entry.key != projection.projection_id @@ -2605,19 +2610,19 @@ impl IdunnProjectionPublisher { } fn record_sha256(value: &T) -> Result { - Ok(format!( - "sha256-{:x}", - Sha256::digest(rmp_serde::to_vec(value)?) - )) + Ok(format!( + "sha256-{:x}", + Sha256::digest(rmp_serde::to_vec(value)?) + )) } fn publish_authenticated_provider_health( - publisher: Option<&IdunnProjectionPublisher>, - options: &CommonOptions, - store_lock: &Arc>, - desired: &IdunnDesiredDaemonRecord, - source: Option<&AuthenticatedProviderHealthSource>, - evaluated_at: &str, + publisher: Option<&IdunnProjectionPublisher>, + options: &CommonOptions, + store_lock: &Arc>, + desired: &IdunnDesiredDaemonRecord, + source: Option<&AuthenticatedProviderHealthSource>, + evaluated_at: &str, ) -> Result<()> { if let (Some(publisher), Some(source)) = (publisher, source) { publisher.publish_if_current(options, store_lock, desired, source, evaluated_at)?; @@ -2844,177 +2849,177 @@ fn start_rudp_health_ingress( Ok(()) } -fn start_public_health_query_listener( - options: &CommonOptions, - targets: &[DaemonTarget], -) -> Result> { - let Some(bind_address) = options.public_health_query_bind else { - return Ok(None); - }; - let public_store_path = options - .public_health_store_path - .clone() - .ok_or_else(|| anyhow!("--public-health-query-bind requires --public-health-store"))?; - if options.service_identity_store_path.is_none() { - return Err(anyhow!( - "--public-health-query-bind requires --service-identity-store projection authority" - )); - } - let server = IdunnPublicHealthSnapshotServer::new(public_store_path, targets)?; - let socket = UdpSocket::bind(bind_address) - .with_context(|| format!("binding public health query listener at {bind_address}"))?; - let local_addr = socket.local_addr()?; - socket.set_read_timeout(Some(Duration::from_millis(250)))?; - thread::spawn(move || { - if let Err(error) = run_public_health_query_listener_loop(socket, server, None) { - eprintln!("Idunn public health query listener stopped: {error:#}"); - } - }); - println!( - "Idunn public health query listener active at {local_addr} for {} target projection key(s).", - targets.len() - ); - Ok(Some(local_addr)) -} - -fn run_public_health_query_listener_loop( - socket: UdpSocket, - server: IdunnPublicHealthSnapshotServer, - stop: Option>, -) -> Result<()> { - let mut sessions: HashMap = HashMap::new(); - let mut buffer = vec![0_u8; 65_535]; - loop { - if stop - .as_ref() - .is_some_and(|flag| flag.load(Ordering::Acquire)) - { - return Ok(()); - } - match socket.recv_from(&mut buffer) { - Ok((size, source)) => { - if let Err(error) = handle_public_health_query_datagram( - &socket, - &mut sessions, - &server, - &buffer[..size], - source, - ) { - eprintln!( - "Idunn public health query listener refused datagram from {source}: {error:#}" - ); - } - } - Err(error) - if matches!( - error.kind(), - std::io::ErrorKind::WouldBlock - | std::io::ErrorKind::TimedOut - | std::io::ErrorKind::ConnectionReset - | std::io::ErrorKind::ConnectionAborted - ) => {} - Err(error) => return Err(error.into()), - } - let now = unix_epoch_millis()?; - let mut expired = Vec::new(); - for (peer, session) in &mut sessions { - for packet in session.due_resends(now) { - socket.send_to(&encode_rudp_packet(&packet)?, peer)?; - } - if session.check_timeout(now, 60_000) { - expired.push(*peer); - } - } - for peer in expired { - sessions.remove(&peer); - } - } -} - -fn handle_public_health_query_datagram( - socket: &UdpSocket, - sessions: &mut HashMap, - server: &IdunnPublicHealthSnapshotServer, - wire: &[u8], - source: SocketAddr, -) -> Result<()> { - let packet = decode_rudp_packet(wire)?; - if packet.connection_id != IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID { - return Err(anyhow!( - "unexpected public health query RUDP connection id {:08x}", - packet.connection_id - )); - } - let now = unix_epoch_millis()?; - if packet.packet_type == CultNetRudpPacketType::Connect { - let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { - connection_id: IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID, - initial_sequence: 1, - resend_delay_ms: 100, - max_pending_reliable_packets: Some(64), - }); - let accept = session.accept_connect(&packet, now, Vec::new())?; - socket.send_to(&encode_rudp_packet(&accept)?, source)?; - sessions.insert(source, session); - return Ok(()); - } - let session = sessions - .get_mut(&source) - .ok_or_else(|| anyhow!("public health query peer has no accepted RUDP session"))?; - let result = session.receive(&packet, now)?; - if let Some(reply) = result.reply { - socket.send_to(&encode_rudp_packet(&reply)?, source)?; - } - if result.disconnected { - sessions.remove(&source); - return Ok(()); - } - if packet.packet_type == CultNetRudpPacketType::Accept || !result.delivered.is_empty() { - socket.send_to(&encode_rudp_packet(&session.create_ack())?, source)?; - } - for frame in result.delivered { - let response = if frame.channel_id != "schema" { - CultNetMessage::Error { - error: "public health queries require the CultNet schema channel".into(), - } - } else { - match decode_cultnet_message_from_slice( - &frame.payload, - CultNetWireContract::CultNetSchemaV0, - ) { - Ok(request) => { - server - .serve(&request) - .unwrap_or_else(|error| CultNetMessage::Error { - error: format!("public health query refused: {error}"), - }) - } - Err(error) => CultNetMessage::Error { - error: format!("public health query message is malformed: {error}"), - }, - } - }; - let payload = - encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0)?; - let packets = session.send_many( - "schema", - payload, - CultNetRudpSendOptions { - reliable: true, - ordered: true, - sequenced: false, - now_ms: now, - reliable_expire_after_ms: None, - }, - Some(1200), - )?; - for packet in packets { - socket.send_to(&encode_rudp_packet(&packet)?, source)?; - } - } - Ok(()) -} - +fn start_public_health_query_listener( + options: &CommonOptions, + targets: &[DaemonTarget], +) -> Result> { + let Some(bind_address) = options.public_health_query_bind else { + return Ok(None); + }; + let public_store_path = options + .public_health_store_path + .clone() + .ok_or_else(|| anyhow!("--public-health-query-bind requires --public-health-store"))?; + if options.service_identity_store_path.is_none() { + return Err(anyhow!( + "--public-health-query-bind requires --service-identity-store projection authority" + )); + } + let server = IdunnPublicHealthSnapshotServer::new(public_store_path, targets)?; + let socket = UdpSocket::bind(bind_address) + .with_context(|| format!("binding public health query listener at {bind_address}"))?; + let local_addr = socket.local_addr()?; + socket.set_read_timeout(Some(Duration::from_millis(250)))?; + thread::spawn(move || { + if let Err(error) = run_public_health_query_listener_loop(socket, server, None) { + eprintln!("Idunn public health query listener stopped: {error:#}"); + } + }); + println!( + "Idunn public health query listener active at {local_addr} for {} target projection key(s).", + targets.len() + ); + Ok(Some(local_addr)) +} + +fn run_public_health_query_listener_loop( + socket: UdpSocket, + server: IdunnPublicHealthSnapshotServer, + stop: Option>, +) -> Result<()> { + let mut sessions: HashMap = HashMap::new(); + let mut buffer = vec![0_u8; 65_535]; + loop { + if stop + .as_ref() + .is_some_and(|flag| flag.load(Ordering::Acquire)) + { + return Ok(()); + } + match socket.recv_from(&mut buffer) { + Ok((size, source)) => { + if let Err(error) = handle_public_health_query_datagram( + &socket, + &mut sessions, + &server, + &buffer[..size], + source, + ) { + eprintln!( + "Idunn public health query listener refused datagram from {source}: {error:#}" + ); + } + } + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::WouldBlock + | std::io::ErrorKind::TimedOut + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + ) => {} + Err(error) => return Err(error.into()), + } + let now = unix_epoch_millis()?; + let mut expired = Vec::new(); + for (peer, session) in &mut sessions { + for packet in session.due_resends(now) { + socket.send_to(&encode_rudp_packet(&packet)?, peer)?; + } + if session.check_timeout(now, 60_000) { + expired.push(*peer); + } + } + for peer in expired { + sessions.remove(&peer); + } + } +} + +fn handle_public_health_query_datagram( + socket: &UdpSocket, + sessions: &mut HashMap, + server: &IdunnPublicHealthSnapshotServer, + wire: &[u8], + source: SocketAddr, +) -> Result<()> { + let packet = decode_rudp_packet(wire)?; + if packet.connection_id != IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID { + return Err(anyhow!( + "unexpected public health query RUDP connection id {:08x}", + packet.connection_id + )); + } + let now = unix_epoch_millis()?; + if packet.packet_type == CultNetRudpPacketType::Connect { + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID, + initial_sequence: 1, + resend_delay_ms: 100, + max_pending_reliable_packets: Some(64), + }); + let accept = session.accept_connect(&packet, now, Vec::new())?; + socket.send_to(&encode_rudp_packet(&accept)?, source)?; + sessions.insert(source, session); + return Ok(()); + } + let session = sessions + .get_mut(&source) + .ok_or_else(|| anyhow!("public health query peer has no accepted RUDP session"))?; + let result = session.receive(&packet, now)?; + if let Some(reply) = result.reply { + socket.send_to(&encode_rudp_packet(&reply)?, source)?; + } + if result.disconnected { + sessions.remove(&source); + return Ok(()); + } + if packet.packet_type == CultNetRudpPacketType::Accept || !result.delivered.is_empty() { + socket.send_to(&encode_rudp_packet(&session.create_ack())?, source)?; + } + for frame in result.delivered { + let response = if frame.channel_id != "schema" { + CultNetMessage::Error { + error: "public health queries require the CultNet schema channel".into(), + } + } else { + match decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) { + Ok(request) => { + server + .serve(&request) + .unwrap_or_else(|error| CultNetMessage::Error { + error: format!("public health query refused: {error}"), + }) + } + Err(error) => CultNetMessage::Error { + error: format!("public health query message is malformed: {error}"), + }, + } + }; + let payload = + encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0)?; + let packets = session.send_many( + "schema", + payload, + CultNetRudpSendOptions { + reliable: true, + ordered: true, + sequenced: false, + now_ms: now, + reliable_expire_after_ms: None, + }, + Some(1200), + )?; + for packet in packets { + socket.send_to(&encode_rudp_packet(&packet)?, source)?; + } + } + Ok(()) +} + fn rudp_health_ingress_record( ingress_id: &str, bind_address: SocketAddr, @@ -3393,9 +3398,9 @@ fn authenticate_generic_signed_health( || document.source_runtime_id.as_deref() != Some(statement.source_runtime_id.as_str()) || document.source_role.as_deref() != Some("daemon-health-publisher") { - return Err(anyhow!( - "signed daemon health transport identity is invalid" - )); + return Err(anyhow!( + "signed daemon health transport identity is invalid" + )); } let (binding, binding_sha256) = load_daemon_health_trust_binding(options, &statement)?; if binding.release_binding_required { @@ -3404,9 +3409,9 @@ fn authenticate_generic_signed_health( || statement.source_commit.is_none() || statement.deployment_id.is_none() { - return Err(anyhow!( - "signed daemon health lacks required release binding" - )); + return Err(anyhow!( + "signed daemon health lacks required release binding" + )); } } else if statement.release_id.is_some() || statement.release_witness_sha256.is_some() @@ -3417,8 +3422,8 @@ fn authenticate_generic_signed_health( "signed daemon health carries release authority not declared by its trust binding" )); } - let expected_identity = - derive_service_identity_id::(&binding.signer_public_key)?; + let expected_identity = + derive_service_identity_id::(&binding.signer_public_key)?; if binding.signer_identity_id != expected_identity || statement.signer_identity_id != binding.signer_identity_id { @@ -3450,9 +3455,9 @@ fn authenticate_generic_signed_health( || admitted_at_unix_millis.saturating_sub(statement.observed_at_unix_millis) > EPIPHANY_ADMISSION_MAX_AGE_SECONDS.saturating_mul(1000) { - return Err(anyhow!( - "signed daemon health observation is future or stale" - )); + return Err(anyhow!( + "signed daemon health observation is future or stale" + )); } let signed_health_sha256 = format!("sha256-{:x}", Sha256::digest(&document.payload)); let observed_at = chrono::DateTime::from_timestamp_millis( @@ -3539,8 +3544,8 @@ fn admit_generic_signed_health( for _ in 0..8 { let (expected_health, expected_admission, expected_statement) = with_store_node(options, store_lock, |node| { - if let Some(existing) = - node.get::(&admission.daemon_id)? + if let Some(existing) = + node.get::(&admission.daemon_id)? && !generic_signed_health_advances(&existing, &admission) { return Err(anyhow!("signed daemon health is replayed or regressed")); @@ -3553,9 +3558,9 @@ fn admit_generic_signed_health( .get_envelope::( &admission.daemon_id, )?, - node.cache().get_envelope::( - &admission.signed_health_sha256, - )?, + node.cache().get_envelope::( + &admission.signed_health_sha256, + )?, )) })?; let expected = [ @@ -3631,17 +3636,17 @@ fn validate_generic_release_binding( || request.release_authority_id.trim().is_empty() || request.release_authority_envelope_sha256.trim().is_empty() { - return Err(anyhow!( - "signed daemon health deployment authority is invalid" - )); + return Err(anyhow!( + "signed daemon health deployment authority is invalid" + )); } let current = node .get::(&admission.daemon_id)? .ok_or_else(|| anyhow!("signed daemon health has no current deployment head"))?; if current.request_id != request.request_id || current.sequence == 0 { - return Err(anyhow!( - "signed daemon health names a superseded deployment" - )); + return Err(anyhow!( + "signed daemon health names a superseded deployment" + )); } let requested_at = parse_timestamp_millis(&request.requested_at) .ok_or_else(|| anyhow!("deployment request timestamp is invalid"))?; @@ -3707,8 +3712,8 @@ fn decode_health_from_rudp_message( if document.record_key != wire.daemon_id { return Err(anyhow!( "record key {} does not match health daemon_id {}", - document.record_key, - wire.daemon_id + document.record_key, + wire.daemon_id )); } let received_at_unix_millis = parse_timestamp_millis(admitted_at) @@ -5325,13 +5330,16 @@ impl Options { }); let mut store_path = PathBuf::from("scratch/idunn/idunn.keepalive.cc"); let mut release_authority_store_path = None; + let mut deployment_brake_store_path = None; + let mut deployment_brake_operator_anchor_path = None; + let mut deployment_brake_runtime_id = None; let mut operator_alarm_command = None; let mut rudp_health_bind = None; let mut trusted_epiphany_health_identity_store = None; let mut daemon_health_trust_store_path = None; let mut service_identity_store_path = None; let mut public_health_store_path = None; - let mut public_health_query_bind = None; + let mut public_health_query_bind = None; let mut execute = false; let mut command_timeout_seconds = 30; let mut daemon_id = None; @@ -5370,6 +5378,22 @@ impl Options { "--release-authority-store", )?)) } + "--deployment-brake-store" => { + deployment_brake_store_path = Some(PathBuf::from(take_value( + &mut args, + "--deployment-brake-store", + )?)) + } + "--deployment-brake-operator-anchor" => { + deployment_brake_operator_anchor_path = Some(PathBuf::from(take_value( + &mut args, + "--deployment-brake-operator-anchor", + )?)) + } + "--deployment-brake-runtime-id" => { + deployment_brake_runtime_id = + Some(take_value(&mut args, "--deployment-brake-runtime-id")?) + } "--repository-full-name" => { validation_repository_full_name = Some(take_value(&mut args, "--repository-full-name")?) @@ -5454,15 +5478,15 @@ impl Options { "--public-health-store", )?)); } - "--public-health-query-bind" => { - public_health_query_bind = Some( - take_value(&mut args, "--public-health-query-bind")? - .parse() - .with_context( - || "--public-health-query-bind must be a socket address", - )?, - ); - } + "--public-health-query-bind" => { + public_health_query_bind = Some( + take_value(&mut args, "--public-health-query-bind")? + .parse() + .with_context( + || "--public-health-query-bind must be a socket address", + )?, + ); + } "--disabled" => enabled = false, "--execute" => execute = true, "--interval-seconds" => { @@ -5502,13 +5526,19 @@ impl Options { let common = CommonOptions { store_path, release_authority_store_path, + deployment_brake_store_path, + deployment_brake_operator_anchor_path, + deployment_brake_runtime_id: deployment_brake_runtime_id.or_else(|| { + (swarm_profile.as_deref() == Some("yggdrasil-local")) + .then(|| "yggdrasil".to_string()) + }), operator_alarm_command, rudp_health_bind, trusted_epiphany_health_identity_store, daemon_health_trust_store_path, service_identity_store_path, public_health_store_path, - public_health_query_bind, + public_health_query_bind, execute, command_timeout_seconds, }; @@ -5576,6 +5606,14 @@ impl Options { "--trusted-epiphany-health-identity-store is required for yggdrasil-local" )); } + if swarm_profile.as_deref() == Some("yggdrasil-local") + && (common.deployment_brake_store_path.is_none() + || common.deployment_brake_operator_anchor_path.is_none()) + { + return Err(anyhow!( + "--deployment-brake-store and --deployment-brake-operator-anchor are required for yggdrasil-local" + )); + } if let Some(action) = lifecycle_action { if swarm_profile.is_some() { @@ -5634,20 +5672,20 @@ impl Options { } }; - if common.public_health_query_bind.is_some() && !matches!(mode, Mode::Swarm(_)) { - return Err(anyhow!( - "--public-health-query-bind requires --swarm-profile" - )); - } - if common.public_health_query_bind.is_some() - && (common.public_health_store_path.is_none() - || common.service_identity_store_path.is_none()) - { - return Err(anyhow!( - "--public-health-query-bind requires --public-health-store and --service-identity-store" - )); - } - + if common.public_health_query_bind.is_some() && !matches!(mode, Mode::Swarm(_)) { + return Err(anyhow!( + "--public-health-query-bind requires --swarm-profile" + )); + } + if common.public_health_query_bind.is_some() + && (common.public_health_store_path.is_none() + || common.service_identity_store_path.is_none()) + { + return Err(anyhow!( + "--public-health-query-bind requires --public-health-store and --service-identity-store" + )); + } + Ok(Self { common, mode }) } } @@ -5695,7 +5733,13 @@ fn run_restart( options: &CommonOptions, ) -> IdunnRestartResultRecord { let result_id = format!("result:{}", request.request_id); - match run_shell(&request.command, options, None) { + match run_braked_shell( + &request.command, + options, + &format!("restart:{}", request.daemon_id), + &request.request_id, + None, + ) { Ok(output) if output.status.success() => IdunnRestartResultRecord { result_id, request_id: request.request_id.clone(), @@ -5743,7 +5787,13 @@ fn run_deployment( completed_at: requested_at.to_string(), }; } - match run_shell(&request.command, options, Some(request)) { + match run_braked_shell( + &request.command, + options, + &request.source_revision, + &request.request_id, + Some(request), + ) { Ok(output) if output.status.success() => IdunnDeploymentResultRecord { result_id, request_id: request.request_id.clone(), @@ -5814,6 +5864,7 @@ fn truncate_detail(value: &str, max_chars: usize) -> String { fn run_state_migration( target: &DaemonTarget, release: &ReleaseTarget, + deployment: &IdunnDeploymentRequestRecord, requested_at: &str, options: &CommonOptions, ) -> Option { @@ -5829,7 +5880,15 @@ fn run_state_migration( completed_at: requested_at.to_string(), }); }; - match run_shell(command, options, None) { + // Migration is consequential deployment work. It consumes the same exact + // release/request grant as the deployment spawn that follows it. + match run_braked_shell( + command, + options, + &deployment.source_revision, + &deployment.request_id, + Some(deployment), + ) { Ok(output) if output.status.success() => Some(IdunnStateMigrationResultRecord { result_id, plan_id, @@ -5891,6 +5950,8 @@ fn rollout_result_record( fn run_shell( command: &str, options: &CommonOptions, + release_id: &str, + deployment_id: &str, deployment: Option<&IdunnDeploymentRequestRecord>, ) -> Result { let mut process = if cfg!(windows) { @@ -5946,6 +6007,9 @@ fn run_shell( .with_context(|| format!("creating {}", stderr_path.display()))?; configure_command_lifetime(&mut process); + // This is deliberately the last fallible authority read before spawn. + // Preparation above has no consequence outside Idunn's temporary files. + verify_deployment_brake(options, release_id, deployment_id)?; let child = process .stdout(Stdio::from(stdout_file)) .stderr(Stdio::from(stderr_file)) @@ -5967,6 +6031,95 @@ fn run_shell( }) } +/// The single final consequence gate. Callers may plan and persist before this, +/// but every migration, deployment, and restart re-opens both root-owned files +/// here, immediately before `Command::spawn`. +fn run_braked_shell( + command: &str, + options: &CommonOptions, + release_id: &str, + deployment_id: &str, + deployment: Option<&IdunnDeploymentRequestRecord>, +) -> Result { + run_shell(command, options, release_id, deployment_id, deployment) +} + +fn verify_deployment_brake( + options: &CommonOptions, + release_id: &str, + deployment_id: &str, +) -> Result<()> { + let store_path = options + .deployment_brake_store_path + .as_ref() + .ok_or_else(|| anyhow!("--deployment-brake-store is required for actuation"))?; + let anchor_path = options + .deployment_brake_operator_anchor_path + .as_ref() + .ok_or_else(|| anyhow!("--deployment-brake-operator-anchor is required for actuation"))?; + let runtime_id = options + .deployment_brake_runtime_id + .as_deref() + .ok_or_else(|| anyhow!("deployment brake runtime id is not configured"))?; + + let entries = SingleFileMessagePackBackingStore::new(store_path) + .pull_all_read_only_snapshot() + .with_context(|| format!("reading deployment brake {}", store_path.display()))?; + let observation = match entries.as_slice() { + [entry] + if entry.r#type == IdunnDeploymentBrakeRecord::TYPE + && entry.key == cultnet_rs::IDUNN_DEPLOYMENT_BRAKE_ID + && entry.schema_id.as_deref() == Some(IDUNN_DEPLOYMENT_BRAKE_SCHEMA) => + { + match rmp_serde::from_slice::(&entry.payload) { + Ok(record) + if rmp_serde::to_vec(&record).ok().as_deref() + == Some(entry.payload.as_slice()) => + { + record + } + _ => return Err(anyhow!("deployment brake is corrupt")), + } + } + [] => return Err(anyhow!("deployment brake is missing")), + _ => return Err(anyhow!("deployment brake store is corrupt or ambiguous")), + }; + let anchor_entries = SingleFileMessagePackBackingStore::new(anchor_path) + .pull_all_read_only_snapshot() + .with_context(|| { + format!( + "reading deployment brake operator anchor {}", + anchor_path.display() + ) + })?; + let [anchor_entry] = anchor_entries.as_slice() else { + return Err(anyhow!( + "deployment brake operator anchor is missing or ambiguous" + )); + }; + if anchor_entry.r#type != IdunnDeploymentBrakeOperatorIdentity::TRUST_ANCHOR_TYPE + || anchor_entry.key != IdunnDeploymentBrakeOperatorIdentity::TRUST_ANCHOR_KEY + || anchor_entry.schema_id.as_deref() + != Some(IdunnDeploymentBrakeOperatorIdentity::TRUST_ANCHOR_SCHEMA) + { + return Err(anyhow!("deployment brake operator anchor is foreign")); + } + let anchor: ServiceIdentityTrustAnchor = rmp_serde::from_slice(&anchor_entry.payload) + .context("decoding deployment brake operator anchor")?; + if rmp_serde::to_vec(&anchor)? != anchor_entry.payload { + return Err(anyhow!("deployment brake operator anchor is noncanonical")); + } + evaluate_idunn_deployment_brake( + IdunnDeploymentBrakeObservation::Present(&observation), + &anchor, + runtime_id, + release_id, + deployment_id, + unix_epoch_millis()?, + ) + .map_err(|denial| anyhow!("deployment brake denied actuation: {denial:?}")) +} + fn actuator_idunn_rudp_health_endpoint(options: &CommonOptions) -> Option { let bind = options.rudp_health_bind?; let host = if bind.ip().is_unspecified() { @@ -6179,7 +6332,7 @@ fn unix_epoch_millis() -> Result { } fn help_text() -> &'static str { - "Usage: idunn --daemon [--name ] [--verse ] [--store ] [--release-authority-store ] [--daemon-health-trust-store ] [--deploy-command ] [--restart-command ] [--operator-alarm-command ] [--rudp-health-bind ] [--trusted-epiphany-health-identity-store ] [--execute] [--interval-seconds ] [--command-timeout-seconds ] [--repo-root ] [--swarm-profile ] [--service-identity-store ] [--public-health-store ] [--public-health-query-bind ]\n idunn restart --daemon [--store ] [--requested-by ] [--detail ]\n idunn redeploy --daemon [--store ] [--requested-by ] [--detail ]\n idunn validate-health-admission --store --daemon --deployment-request-id --release-id --release-witness-sha256 --source-revision \n\nIdunn supervises owner-authenticated CultNet/RUDP health with --daemon, or a built-in swarm supervisor with --swarm-profile starfire-local or yggdrasil-local. Generic signed health requires a root-owned CultCache trust store via --daemon-health-trust-store. Unsigned idunn.daemon_health packets are diagnostic-only and cannot satisfy lifecycle health. Yggdrasil release targets require an explicit Bifrost CultCache path via --release-authority-store. The Epiphany v0 migration path additionally requires its pinned host identity. RUDP health ingress is disabled unless --rudp-health-bind is supplied. The read-only outward projection listener is disabled unless --public-health-query-bind is supplied with the dedicated public store and Idunn service identity; it serves only target-catalog authenticated-provider projection keys and never opens private Idunn state. The restart/redeploy verbs publish typed idunn.lifecycle_command.v1 records; the running supervisor claims them and executes only through its configured command boundary." + "Usage: idunn --daemon [--name ] [--verse ] [--store ] [--release-authority-store ] [--daemon-health-trust-store ] [--deploy-command ] [--restart-command ] [--operator-alarm-command ] [--rudp-health-bind ] [--trusted-epiphany-health-identity-store ] [--execute] [--interval-seconds ] [--command-timeout-seconds ] [--repo-root ] [--swarm-profile ] [--service-identity-store ] [--public-health-store ] [--public-health-query-bind ]\n idunn restart --daemon [--store ] [--requested-by ] [--detail ]\n idunn redeploy --daemon [--store ] [--requested-by ] [--detail ]\n idunn validate-health-admission --store --daemon --deployment-request-id --release-id --release-witness-sha256 --source-revision \n\nIdunn supervises owner-authenticated CultNet/RUDP health with --daemon, or a built-in swarm supervisor with --swarm-profile starfire-local or yggdrasil-local. Generic signed health requires a root-owned CultCache trust store via --daemon-health-trust-store. Unsigned idunn.daemon_health packets are diagnostic-only and cannot satisfy lifecycle health. Yggdrasil release targets require an explicit Bifrost CultCache path via --release-authority-store. The Epiphany v0 migration path additionally requires its pinned host identity. RUDP health ingress is disabled unless --rudp-health-bind is supplied. The read-only outward projection listener is disabled unless --public-health-query-bind is supplied with the dedicated public store and Idunn service identity; it serves only target-catalog authenticated-provider projection keys and never opens private Idunn state. The restart/redeploy verbs publish typed idunn.lifecycle_command.v1 records; the running supervisor claims them and executes only through its configured command boundary." } #[cfg(test)] @@ -6213,7 +6366,7 @@ mod tests { .unwrap(); }); assert!( - entered_rx.recv_timeout(Duration::from_millis(100)).is_err(), + entered_rx.recv_timeout(Duration::from_millis(100)).is_err(), "automatic request owner entered while manual actuation was live" ); release_tx.send(()).unwrap(); @@ -6233,11 +6386,11 @@ mod tests { Ok(()) }); - assert!( - actuation_rx - .recv_timeout(Duration::from_millis(100)) - .is_err() - ); + assert!( + actuation_rx + .recv_timeout(Duration::from_millis(100)) + .is_err() + ); // This is the target loop's independent signed-health admission lane: // it does not acquire or await the consequence gate. let admitted_health = "accepted"; @@ -6271,8 +6424,8 @@ mod tests { #[cfg(unix)] #[test] fn status_timeout_terminates_command_descendants() { - let marker = - env::temp_dir().join(format!("idunn-status-descendant-{}", std::process::id())); + let marker = + env::temp_dir().join(format!("idunn-status-descendant-{}", std::process::id())); let _ = fs::remove_file(&marker); let error = wait_for_child_status_with_timeout( descendant_marker_command(&marker), @@ -6282,17 +6435,17 @@ mod tests { .expect_err("command should time out"); assert!(error.to_string().contains("timed out")); thread::sleep(Duration::from_millis(1200)); - assert!( - !marker.exists(), - "timed-out descendant survived to write marker" - ); + assert!( + !marker.exists(), + "timed-out descendant survived to write marker" + ); } #[cfg(unix)] #[test] fn output_timeout_terminates_command_descendants() { - let marker = - env::temp_dir().join(format!("idunn-output-descendant-{}", std::process::id())); + let marker = + env::temp_dir().join(format!("idunn-output-descendant-{}", std::process::id())); let _ = fs::remove_file(&marker); let error = wait_for_child_with_timeout( descendant_marker_command(&marker), @@ -6302,10 +6455,10 @@ mod tests { .expect_err("command should time out"); assert!(error.to_string().contains("timed out")); thread::sleep(Duration::from_millis(1200)); - assert!( - !marker.exists(), - "timed-out descendant survived to write marker" - ); + assert!( + !marker.exists(), + "timed-out descendant survived to write marker" + ); } use cultnet_rs::{CultNetRawDocumentRecord, CultNetRawPayloadEncoding}; use ed25519_dalek::{Signer, SigningKey}; @@ -6551,13 +6704,16 @@ mod tests { let options = CommonOptions { store_path: env::temp_dir().join("unused-idunn-store.cc"), release_authority_store_path: Some(path.clone()), + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: true, command_timeout_seconds: 1, }; @@ -7502,11 +7658,11 @@ mod tests { )); assert!(actuator.contains("deploy:epiphany|restart:epiphany")); assert!(actuator.contains("/usr/local/bin/idunn validate-release-authority")); - assert!( - actuator.contains( - "epiphany|bifrost-persona-feedback) target_requires_bifrost_authority=true" - ) - ); + assert!( + actuator.contains( + "epiphany|bifrost-persona-feedback) target_requires_bifrost_authority=true" + ) + ); assert!(actuator.contains( "voidbot|heimdall|repixelizer|streampixels) target_requires_bifrost_authority=false" )); @@ -7572,9 +7728,9 @@ mod tests { } fn deployed_revision(&self, _release: &ReleaseTarget) -> Result { - Err( - std::io::Error::new(std::io::ErrorKind::NotFound, "deployment witness is absent") - .into(), + Err( + std::io::Error::new(std::io::ErrorKind::NotFound, "deployment witness is absent") + .into(), ) } } @@ -7695,11 +7851,11 @@ mod tests { assert_eq!(health.state, "stale-deployment"); assert_eq!(health.transport, "idunn.release-not-deployed"); - assert!( - health - .detail - .contains("no deployed revision witness exists") - ); + assert!( + health + .detail + .contains("no deployed revision witness exists") + ); assert!( plan_keepalive(&desired, &health, "unix:100") .deployment_request @@ -7806,18 +7962,18 @@ mod tests { ); assert_eq!( target.deploy_command.as_deref(), - Some( - "sudo -n /usr/local/libexec/idunn-yggdrasil deploy bifrost-persona-feedback \"$IDUNN_SOURCE_COMMIT\" \"$IDUNN_REPOSITORY_FULL_NAME\" \"$IDUNN_UPSTREAM_REF\" \"$BIFROST_RELEASE_AUTHORITY_ID\" \"$BIFROST_RELEASE_AUTHORITY_SHA256\" \"$IDUNN_DEPLOYMENT_REQUEST_ID\" \"$IDUNN_REQUIRES_BIFROST_AUTHORITY\"" - ) + Some( + "sudo -n /usr/local/libexec/idunn-yggdrasil deploy bifrost-persona-feedback \"$IDUNN_SOURCE_COMMIT\" \"$IDUNN_REPOSITORY_FULL_NAME\" \"$IDUNN_UPSTREAM_REF\" \"$BIFROST_RELEASE_AUTHORITY_ID\" \"$BIFROST_RELEASE_AUTHORITY_SHA256\" \"$IDUNN_DEPLOYMENT_REQUEST_ID\" \"$IDUNN_REQUIRES_BIFROST_AUTHORITY\"" + ) ); assert!(target.restart_command.is_none()); let release = target.release.as_ref().expect("release target"); assert!(release.requires_bifrost_authority); assert_eq!(release.repository_full_name, "GameCult/Bifrost"); - assert_eq!( - release.repo_path, - PathBuf::from("/srv/build/Bifrost-persona-feedback") - ); + assert_eq!( + release.repo_path, + PathBuf::from("/srv/build/Bifrost-persona-feedback") + ); let desired = IdunnDesiredDaemonRecord { daemon_id: target.daemon_id.clone(), @@ -7907,13 +8063,16 @@ mod tests { let options = CommonOptions { store_path: store_path.clone(), release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }; @@ -7983,13 +8142,16 @@ mod tests { let options = CommonOptions { store_path: store_path.clone(), release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }; @@ -8549,27 +8711,30 @@ mod tests { let options = CommonOptions { store_path: root.path().join("idunn.cc"), release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }; let lock = Arc::new(Mutex::new(())); - let outcome = - admit_health_from_rudp_message(&message, &options, &lock, "2026-06-15T00:00:01Z") - .unwrap(); + let outcome = + admit_health_from_rudp_message(&message, &options, &lock, "2026-06-15T00:00:01Z") + .unwrap(); assert_eq!(outcome.authority, "diagnostic-only"); assert!(health_from_rudp_message(&message, &options).is_err()); with_store_node(&options, &lock, |node| { - assert!( - node.get::(&health.daemon_id)? - .is_none() - ); + assert!( + node.get::(&health.daemon_id)? + .is_none() + ); let diagnostic = node .get::(&format!( "diagnostic:{}", @@ -8618,9 +8783,9 @@ mod tests { SingleFileMessagePackBackingStore::new(&trust_path) .push(&typed_envelope(&binding.binding_id, &binding, "unix:1784483100").unwrap()) .unwrap(); - let observed_at_unix_millis = chrono::DateTime::parse_from_rfc3339("2026-07-19T12:00:00Z") - .unwrap() - .timestamp_millis() as u64; + let observed_at_unix_millis = chrono::DateTime::parse_from_rfc3339("2026-07-19T12:00:00Z") + .unwrap() + .timestamp_millis() as u64; let mut statement = IdunnSignedDaemonHealthRecord { schema_version: odin_core::IDUNN_SIGNED_DAEMON_HEALTH_SCHEMA.into(), daemon_id: binding.daemon_id.clone(), @@ -8633,7 +8798,7 @@ mod tests { publisher_sequence: sequence, observed_at_unix_millis, release_id: release_bound.then(|| "release-test-1".into()), - release_witness_sha256: release_bound.then(|| format!("sha256-{}", "a".repeat(64))), + release_witness_sha256: release_bound.then(|| format!("sha256-{}", "a".repeat(64))), source_commit: release_bound.then(|| "b".repeat(40)), deployment_id: release_bound.then(|| "deploy-test-1".into()), signature_algorithm: "ed25519".into(), @@ -8691,13 +8856,16 @@ mod tests { CommonOptions { store_path, release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: Some(trust_path), service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }, @@ -8854,233 +9022,233 @@ mod tests { rmp_serde::from_slice(&entries[0].payload).unwrap() } - fn public_query_target(desired: &IdunnDesiredDaemonRecord) -> DaemonTarget { - DaemonTarget { - daemon_id: desired.daemon_id.clone(), - verse_id: desired.verse_id.clone(), - name: desired.name.clone(), - health_contract: health_contract(&desired.health_contract, "failed"), - deploy_command: None, - restart_command: None, - release: None, - enabled: true, - interval_seconds: 30, - } - } - - fn connect_public_query_client( - server_addr: SocketAddr, - ) -> CultNetRudpSocketTransportConnection { - let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); - socket - .set_read_timeout(Some(Duration::from_millis(100))) - .unwrap(); - let mut client = - CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::client( - "epiphany-test-consumer", - socket, - server_addr, - IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID, - )) - .unwrap(); - client.connect(Vec::new()).unwrap(); - let deadline = Instant::now() + Duration::from_secs(2); - while !client.connected() { - client.receive_once().unwrap(); - client.poll_resends().unwrap(); - assert!( - Instant::now() < deadline, - "public query handshake timed out" - ); - } - client - } - - fn receive_public_query_message( - client: &mut CultNetRudpSocketTransportConnection, - ) -> CultNetMessage { - let deadline = Instant::now() + Duration::from_secs(2); - loop { - if let Some(frame) = client.receive_once().unwrap() { - assert_eq!(frame.channel_id, "schema"); - return decode_cultnet_message_from_slice( - &frame.payload, - CultNetWireContract::CultNetSchemaV0, - ) - .unwrap(); - } - client.poll_resends().unwrap(); - assert!(Instant::now() < deadline, "public query response timed out"); - } - } - - #[test] - fn public_query_server_returns_exact_allowlisted_bytes_and_correlation() { - let (options, lock, desired, source, publisher, _root) = projection_fixture(false); - publisher - .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:02Z") - .unwrap(); - let public_path = options.public_health_store_path.as_ref().unwrap(); - let stored = SingleFileMessagePackBackingStore::new(public_path) - .pull_all_read_only_snapshot() - .unwrap() - .remove(0); - let server = IdunnPublicHealthSnapshotServer::new( - public_path.clone(), - &[public_query_target(&desired)], - ) - .unwrap(); - let request = CultNetMessage::SnapshotRequest { - message_id: "status-correlation-7".into(), - schema_ids: Some(vec![ - IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), - ]), - record_keys: Some(vec![stored.key.clone()]), - }; - let CultNetMessage::SnapshotResponseRaw { - message_id, - documents, - } = server.serve(&request).unwrap() - else { - panic!("expected raw public health snapshot"); - }; - assert_eq!(message_id, "status-correlation-7"); - assert_eq!(documents.len(), 1); - assert_eq!(documents[0].payload, stored.payload); - assert_eq!( - documents[0].source_runtime_id.as_deref(), - Some(IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID) - ); - assert_eq!( - documents[0].source_role.as_deref(), - Some(IDUNN_PUBLIC_HEALTH_QUERY_ROLE) - ); - - for request in [ - CultNetMessage::SnapshotRequest { - message_id: "unknown-schema".into(), - schema_ids: Some(vec!["private.idunn.state.v0".into()]), - record_keys: None, - }, - CultNetMessage::SnapshotRequest { - message_id: "unknown-key".into(), - schema_ids: None, - record_keys: Some(vec!["provider-health:not-in-target-catalog".into()]), - }, - ] { - let CultNetMessage::SnapshotResponseRaw { documents, .. } = - server.serve(&request).unwrap() - else { - panic!("expected bounded empty response"); - }; - assert!(documents.is_empty()); - } - } - - #[test] - fn public_query_store_contamination_is_fatal_and_private_type_is_never_served() { - let (options, _lock, desired, _source, _publisher, root) = projection_fixture(false); - let private_path = root.path().join("private-in-public.cc"); - let private = IdunnDaemonHealthRecord { - daemon_id: desired.daemon_id.clone(), - state: "active".into(), - detail: "private managed judgment".into(), - observed_at: "2026-07-19T12:00:02Z".into(), - health_contract: desired.health_contract.clone(), - publication_source: "daemon-authenticated".into(), - transport: CULTNET_RUDP_PROTOCOL_ID.into(), - }; - SingleFileMessagePackBackingStore::new(&private_path) - .push(&typed_envelope(&private.daemon_id, &private, "2026-07-19T12:00:02Z").unwrap()) - .unwrap(); - assert!( - IdunnPublicHealthSnapshotServer::new(private_path, &[public_query_target(&desired)]) - .is_err() - ); - - let public_path = options.public_health_store_path.as_ref().unwrap(); - let server = IdunnPublicHealthSnapshotServer::new( - public_path.clone(), - &[public_query_target(&desired)], - ) - .unwrap(); - SingleFileMessagePackBackingStore::new(public_path) - .push(&typed_envelope(&private.daemon_id, &private, "2026-07-19T12:00:03Z").unwrap()) - .unwrap(); - assert!( - server - .serve(&CultNetMessage::SnapshotRequest { - message_id: "private-probe".into(), - schema_ids: None, - record_keys: None, - }) - .is_err() - ); - } - - #[test] - fn public_query_rudp_is_multi_peer_and_survives_malformed_request() { - let (options, lock, desired, source, publisher, _root) = projection_fixture(false); - publisher - .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:02Z") - .unwrap(); - let server = IdunnPublicHealthSnapshotServer::new( - options.public_health_store_path.clone().unwrap(), - &[public_query_target(&desired)], - ) - .unwrap(); - let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); - let server_addr = socket.local_addr().unwrap(); - socket - .set_read_timeout(Some(Duration::from_millis(100))) - .unwrap(); - let stop = Arc::new(AtomicBool::new(false)); - let worker_stop = Arc::clone(&stop); - let worker = thread::spawn(move || { - run_public_health_query_listener_loop(socket, server, Some(worker_stop)) - }); - - let mut hostile = connect_public_query_client(server_addr); - hostile.send("schema", vec![0xc1]).unwrap(); - assert!(matches!( - receive_public_query_message(&mut hostile), - CultNetMessage::Error { .. } - )); - - let mut first = connect_public_query_client(server_addr); - let mut second = connect_public_query_client(server_addr); - for (client, id) in [(&mut first, "first-peer"), (&mut second, "second-peer")] { - let request = CultNetMessage::SnapshotRequest { - message_id: id.into(), - schema_ids: Some(vec![ - IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), - ]), - record_keys: Some(vec![provider_health_projection_key( - &desired.daemon_id, - &desired.health_contract, - )]), - }; - client - .send( - "schema", - encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0) - .unwrap(), - ) - .unwrap(); - let CultNetMessage::SnapshotResponseRaw { - message_id, - documents, - } = receive_public_query_message(client) - else { - panic!("expected peer snapshot response"); - }; - assert_eq!(message_id, id); - assert_eq!(documents.len(), 1); - } - stop.store(true, Ordering::Release); - worker.join().unwrap().unwrap(); - } - + fn public_query_target(desired: &IdunnDesiredDaemonRecord) -> DaemonTarget { + DaemonTarget { + daemon_id: desired.daemon_id.clone(), + verse_id: desired.verse_id.clone(), + name: desired.name.clone(), + health_contract: health_contract(&desired.health_contract, "failed"), + deploy_command: None, + restart_command: None, + release: None, + enabled: true, + interval_seconds: 30, + } + } + + fn connect_public_query_client( + server_addr: SocketAddr, + ) -> CultNetRudpSocketTransportConnection { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let mut client = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::client( + "epiphany-test-consumer", + socket, + server_addr, + IDUNN_PUBLIC_HEALTH_QUERY_CONNECTION_ID, + )) + .unwrap(); + client.connect(Vec::new()).unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + while !client.connected() { + client.receive_once().unwrap(); + client.poll_resends().unwrap(); + assert!( + Instant::now() < deadline, + "public query handshake timed out" + ); + } + client + } + + fn receive_public_query_message( + client: &mut CultNetRudpSocketTransportConnection, + ) -> CultNetMessage { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + if let Some(frame) = client.receive_once().unwrap() { + assert_eq!(frame.channel_id, "schema"); + return decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) + .unwrap(); + } + client.poll_resends().unwrap(); + assert!(Instant::now() < deadline, "public query response timed out"); + } + } + + #[test] + fn public_query_server_returns_exact_allowlisted_bytes_and_correlation() { + let (options, lock, desired, source, publisher, _root) = projection_fixture(false); + publisher + .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:02Z") + .unwrap(); + let public_path = options.public_health_store_path.as_ref().unwrap(); + let stored = SingleFileMessagePackBackingStore::new(public_path) + .pull_all_read_only_snapshot() + .unwrap() + .remove(0); + let server = IdunnPublicHealthSnapshotServer::new( + public_path.clone(), + &[public_query_target(&desired)], + ) + .unwrap(); + let request = CultNetMessage::SnapshotRequest { + message_id: "status-correlation-7".into(), + schema_ids: Some(vec![ + IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), + ]), + record_keys: Some(vec![stored.key.clone()]), + }; + let CultNetMessage::SnapshotResponseRaw { + message_id, + documents, + } = server.serve(&request).unwrap() + else { + panic!("expected raw public health snapshot"); + }; + assert_eq!(message_id, "status-correlation-7"); + assert_eq!(documents.len(), 1); + assert_eq!(documents[0].payload, stored.payload); + assert_eq!( + documents[0].source_runtime_id.as_deref(), + Some(IDUNN_PUBLIC_HEALTH_QUERY_RUNTIME_ID) + ); + assert_eq!( + documents[0].source_role.as_deref(), + Some(IDUNN_PUBLIC_HEALTH_QUERY_ROLE) + ); + + for request in [ + CultNetMessage::SnapshotRequest { + message_id: "unknown-schema".into(), + schema_ids: Some(vec!["private.idunn.state.v0".into()]), + record_keys: None, + }, + CultNetMessage::SnapshotRequest { + message_id: "unknown-key".into(), + schema_ids: None, + record_keys: Some(vec!["provider-health:not-in-target-catalog".into()]), + }, + ] { + let CultNetMessage::SnapshotResponseRaw { documents, .. } = + server.serve(&request).unwrap() + else { + panic!("expected bounded empty response"); + }; + assert!(documents.is_empty()); + } + } + + #[test] + fn public_query_store_contamination_is_fatal_and_private_type_is_never_served() { + let (options, _lock, desired, _source, _publisher, root) = projection_fixture(false); + let private_path = root.path().join("private-in-public.cc"); + let private = IdunnDaemonHealthRecord { + daemon_id: desired.daemon_id.clone(), + state: "active".into(), + detail: "private managed judgment".into(), + observed_at: "2026-07-19T12:00:02Z".into(), + health_contract: desired.health_contract.clone(), + publication_source: "daemon-authenticated".into(), + transport: CULTNET_RUDP_PROTOCOL_ID.into(), + }; + SingleFileMessagePackBackingStore::new(&private_path) + .push(&typed_envelope(&private.daemon_id, &private, "2026-07-19T12:00:02Z").unwrap()) + .unwrap(); + assert!( + IdunnPublicHealthSnapshotServer::new(private_path, &[public_query_target(&desired)]) + .is_err() + ); + + let public_path = options.public_health_store_path.as_ref().unwrap(); + let server = IdunnPublicHealthSnapshotServer::new( + public_path.clone(), + &[public_query_target(&desired)], + ) + .unwrap(); + SingleFileMessagePackBackingStore::new(public_path) + .push(&typed_envelope(&private.daemon_id, &private, "2026-07-19T12:00:03Z").unwrap()) + .unwrap(); + assert!( + server + .serve(&CultNetMessage::SnapshotRequest { + message_id: "private-probe".into(), + schema_ids: None, + record_keys: None, + }) + .is_err() + ); + } + + #[test] + fn public_query_rudp_is_multi_peer_and_survives_malformed_request() { + let (options, lock, desired, source, publisher, _root) = projection_fixture(false); + publisher + .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:02Z") + .unwrap(); + let server = IdunnPublicHealthSnapshotServer::new( + options.public_health_store_path.clone().unwrap(), + &[public_query_target(&desired)], + ) + .unwrap(); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let server_addr = socket.local_addr().unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let stop = Arc::new(AtomicBool::new(false)); + let worker_stop = Arc::clone(&stop); + let worker = thread::spawn(move || { + run_public_health_query_listener_loop(socket, server, Some(worker_stop)) + }); + + let mut hostile = connect_public_query_client(server_addr); + hostile.send("schema", vec![0xc1]).unwrap(); + assert!(matches!( + receive_public_query_message(&mut hostile), + CultNetMessage::Error { .. } + )); + + let mut first = connect_public_query_client(server_addr); + let mut second = connect_public_query_client(server_addr); + for (client, id) in [(&mut first, "first-peer"), (&mut second, "second-peer")] { + let request = CultNetMessage::SnapshotRequest { + message_id: id.into(), + schema_ids: Some(vec![ + IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SCHEMA.into(), + ]), + record_keys: Some(vec![provider_health_projection_key( + &desired.daemon_id, + &desired.health_contract, + )]), + }; + client + .send( + "schema", + encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0) + .unwrap(), + ) + .unwrap(); + let CultNetMessage::SnapshotResponseRaw { + message_id, + documents, + } = receive_public_query_message(client) + else { + panic!("expected peer snapshot response"); + }; + assert_eq!(message_id, id); + assert_eq!(documents.len(), 1); + } + stop.store(true, Ordering::Release); + worker.join().unwrap().unwrap(); + } + #[test] fn projection_is_signed_canonical_public_state_without_provider_detail() { let (options, lock, desired, source, publisher, _root) = projection_fixture(false); @@ -9225,23 +9393,23 @@ mod tests { payload: rmp_serde::to_vec(&corrupted).unwrap(), ..current_envelope.clone() }; - assert!( - backing - .compare_exchange( - &[CultCacheExpectedEnvelope { - key: current_envelope.key.clone(), - r#type: current_envelope.r#type.clone(), - current: Some(current_envelope), - }], - &[corrupted_envelope], - ) - .unwrap() - ); - assert!( - second - .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:04Z") - .is_err() - ); + assert!( + backing + .compare_exchange( + &[CultCacheExpectedEnvelope { + key: current_envelope.key.clone(), + r#type: current_envelope.r#type.clone(), + current: Some(current_envelope), + }], + &[corrupted_envelope], + ) + .unwrap() + ); + assert!( + second + .publish_if_current(&options, &lock, &desired, &source, "2026-07-19T12:00:04Z") + .is_err() + ); let contaminated_path = root.path().join("contaminated-public.cc"); SingleFileMessagePackBackingStore::new(&contaminated_path) @@ -9456,13 +9624,16 @@ mod tests { let options = CommonOptions { store_path, release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: Some(identity_path), daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }; @@ -9752,13 +9923,16 @@ mod tests { let options = CommonOptions { store_path: root.path().join("idunn.cc"), release_authority_store_path: None, + deployment_brake_store_path: None, + deployment_brake_operator_anchor_path: None, + deployment_brake_runtime_id: None, operator_alarm_command: None, rudp_health_bind: None, trusted_epiphany_health_identity_store: None, daemon_health_trust_store_path: None, service_identity_store_path: None, public_health_store_path: None, - public_health_query_bind: None, + public_health_query_bind: None, execute: false, command_timeout_seconds: 1, }; diff --git a/crates/idunn-daemon/src/provisioning.rs b/crates/idunn-daemon/src/provisioning.rs index 965aa3c..b046f58 100644 --- a/crates/idunn-daemon/src/provisioning.rs +++ b/crates/idunn-daemon/src/provisioning.rs @@ -3,8 +3,12 @@ use cultcache_rs::{ CultCacheEnvelope, CultCacheExpectedEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, }; use cultnet_rs::{ - GameCultProviderHealthIdentity, IdunnServiceIdentity, ServiceIdentityProfile, - ServiceIdentityTrustAnchor, derive_service_identity_id, enroll_service_identity_at, + GameCultProviderHealthIdentity, IDUNN_DEPLOYMENT_BRAKE_AUTHORITY, IDUNN_DEPLOYMENT_BRAKE_ID, + IDUNN_DEPLOYMENT_BRAKE_SCHEMA, IDUNN_DEPLOYMENT_BRAKE_SCOPE, IDUNN_DEPLOYMENT_RELEASE_PURPOSE, + IdunnDeploymentBrakeObservation, IdunnDeploymentBrakeOperatorIdentity, + IdunnDeploymentBrakeRecord, IdunnDeploymentBrakeReleasePurpose, IdunnServiceIdentity, + ServiceIdentityProfile, ServiceIdentityTrustAnchor, derive_service_identity_id, + enroll_service_identity_at, evaluate_idunn_deployment_brake, export_service_identity_trust_anchor, open_service_identity_at, }; use odin_core::{ @@ -50,6 +54,26 @@ pub fn run(args: impl IntoIterator) -> Result<()> { let signer = open_service_identity_at::(&private)?; export_service_identity_trust_anchor(&signer, &public)?; } + "enroll-deployment-brake-operator" => { + require_only(&options, &["private-store"])?; + enroll_service_identity_at::(&path( + &options, + "private-store", + )?)?; + } + "export-deployment-brake-operator-anchor" => { + require_only(&options, &["private-store", "public-anchor"])?; + let private = path(&options, "private-store")?; + let public = path(&options, "public-anchor")?; + reject_alias(&private, &public)?; + refuse_existing(&public, "deployment brake operator anchor")?; + let signer = + open_service_identity_at::(&private)?; + export_service_identity_trust_anchor(&signer, &public)?; + } + "deployment-brake-engage" => deployment_brake_engage(&options)?, + "deployment-brake-release" => deployment_brake_release(&options)?, + "deployment-brake-status" => deployment_brake_status(&options)?, "create-daemon-health-trust-binding" => create_health_binding(&options)?, "add-daemon-health-trust-binding" => add_health_binding(&options)?, "validate-daemon-health-trust-binding" => { @@ -63,6 +87,204 @@ pub fn run(args: impl IntoIterator) -> Result<()> { Ok(()) } +fn deployment_brake_engage(options: &BTreeMap) -> Result<()> { + require_only( + options, + &[ + "store", + "runtime-id", + "owner", + "reason", + "observed-at-unix-millis", + ], + )?; + let observed = parse_u64(options, "observed-at-unix-millis")?; + let record = IdunnDeploymentBrakeRecord { + schema_version: IDUNN_DEPLOYMENT_BRAKE_SCHEMA.into(), + brake_id: IDUNN_DEPLOYMENT_BRAKE_ID.into(), + authority: IDUNN_DEPLOYMENT_BRAKE_AUTHORITY.into(), + runtime_id: required(options, "runtime-id")?.into(), + status: "engaged".into(), + scope: IDUNN_DEPLOYMENT_BRAKE_SCOPE.into(), + reason: required(options, "reason")?.into(), + observed_at_unix_millis: observed, + expires_at_unix_millis: None, + authorization_id: None, + authorization_purpose: None, + authorized_release_id: None, + authorized_deployment_id: None, + authorized_by: None, + authorization_issued_at_unix_millis: None, + authorization_expires_at_unix_millis: None, + signature_algorithm: None, + signature: None, + private_state_exposed: false, + updated_by: required(options, "owner")?.into(), + }; + record.validate()?; + replace_brake(&path(options, "store")?, record, observed) +} + +fn deployment_brake_release(options: &BTreeMap) -> Result<()> { + require_only( + options, + &[ + "store", + "private-store", + "runtime-id", + "owner", + "reason", + "authorization-id", + "release-id", + "deployment-id", + "issued-at-unix-millis", + "expires-at-unix-millis", + ], + )?; + let issued = parse_u64(options, "issued-at-unix-millis")?; + let expires = parse_u64(options, "expires-at-unix-millis")?; + let signer = open_service_identity_at::(&path( + options, + "private-store", + )?)?; + let mut record = IdunnDeploymentBrakeRecord { + schema_version: IDUNN_DEPLOYMENT_BRAKE_SCHEMA.into(), + brake_id: IDUNN_DEPLOYMENT_BRAKE_ID.into(), + authority: IDUNN_DEPLOYMENT_BRAKE_AUTHORITY.into(), + runtime_id: required(options, "runtime-id")?.into(), + status: "released".into(), + scope: IDUNN_DEPLOYMENT_BRAKE_SCOPE.into(), + reason: required(options, "reason")?.into(), + observed_at_unix_millis: issued, + expires_at_unix_millis: Some(expires), + authorization_id: Some(required(options, "authorization-id")?.into()), + authorization_purpose: Some(IDUNN_DEPLOYMENT_RELEASE_PURPOSE.into()), + authorized_release_id: Some(required(options, "release-id")?.into()), + authorized_deployment_id: Some(required(options, "deployment-id")?.into()), + authorized_by: Some(signer.trust_anchor()?.identity_id), + authorization_issued_at_unix_millis: Some(issued), + authorization_expires_at_unix_millis: Some(expires), + signature_algorithm: Some("ed25519".into()), + signature: None, + private_state_exposed: false, + updated_by: required(options, "owner")?.into(), + }; + record.signature = Some( + signer + .sign::(&rmp_serde::to_vec(&record)?) + .signature, + ); + record.validate()?; + replace_brake(&path(options, "store")?, record, issued) +} + +fn deployment_brake_status(options: &BTreeMap) -> Result<()> { + allow_only( + options, + &[ + "store", + "operator-anchor", + "runtime-id", + "release-id", + "deployment-id", + "now-unix-millis", + ], + )?; + let record = read_brake(&path(options, "store")?)?; + if record.runtime_id != required(options, "runtime-id")? { + bail!("deployment brake belongs to another runtime") + } + if record.status == "engaged" { + println!( + "engaged runtime={} scope={} owner={} reason={}", + record.runtime_id, record.scope, record.updated_by, record.reason + ); + return Ok(()); + } + let anchor = read_operator_anchor(&path(options, "operator-anchor")?)?; + evaluate_idunn_deployment_brake( + IdunnDeploymentBrakeObservation::Present(&record), + &anchor, + required(options, "runtime-id")?, + required(options, "release-id")?, + required(options, "deployment-id")?, + parse_u64(options, "now-unix-millis")?, + ) + .map_err(|d| anyhow!("deployment brake denied actuation: {d:?}"))?; + println!( + "released runtime={} release={} deployment={} owner={}", + record.runtime_id, + record.authorized_release_id.unwrap(), + record.authorized_deployment_id.unwrap(), + record.updated_by + ); + Ok(()) +} + +fn replace_brake(path: &Path, record: IdunnDeploymentBrakeRecord, millis: u64) -> Result<()> { + let store = SingleFileMessagePackBackingStore::new(path); + let entries = store.pull_all_read_only_snapshot()?; + let current = match entries.as_slice() { + [] => None, + [entry] + if entry.r#type == IdunnDeploymentBrakeRecord::TYPE + && entry.key == IDUNN_DEPLOYMENT_BRAKE_ID => + { + Some(entry.clone()) + } + _ => bail!("deployment brake store is corrupt or ambiguous"), + }; + let next = typed_envelope( + IDUNN_DEPLOYMENT_BRAKE_ID, + &record, + IDUNN_DEPLOYMENT_BRAKE_SCHEMA, + millis, + )?; + if !store.compare_exchange( + &[CultCacheExpectedEnvelope { + r#type: IdunnDeploymentBrakeRecord::TYPE.into(), + key: IDUNN_DEPLOYMENT_BRAKE_ID.into(), + current, + }], + &[next], + )? { + bail!("deployment brake changed during transition"); + } + Ok(()) +} + +fn read_brake(path: &Path) -> Result { + let entries = SingleFileMessagePackBackingStore::new(path).pull_all_read_only_snapshot()?; + let [entry] = entries.as_slice() else { + bail!("deployment brake is missing or ambiguous") + }; + if entry.r#type != IdunnDeploymentBrakeRecord::TYPE + || entry.key != IDUNN_DEPLOYMENT_BRAKE_ID + || entry.schema_id.as_deref() != Some(IDUNN_DEPLOYMENT_BRAKE_SCHEMA) + { + bail!("deployment brake is foreign") + } + let record: IdunnDeploymentBrakeRecord = rmp_serde::from_slice(&entry.payload)?; + if rmp_serde::to_vec(&record)? != entry.payload { + bail!("deployment brake is noncanonical") + } + record.validate()?; + Ok(record) +} + +fn read_operator_anchor(path: &Path) -> Result { + let entries = SingleFileMessagePackBackingStore::new(path).pull_all_read_only_snapshot()?; + let [entry] = entries.as_slice() else { + bail!("operator anchor is missing or ambiguous") + }; + if entry.r#type != IdunnDeploymentBrakeOperatorIdentity::TRUST_ANCHOR_TYPE + || entry.key != IdunnDeploymentBrakeOperatorIdentity::TRUST_ANCHOR_KEY + { + bail!("operator anchor is foreign") + } + Ok(rmp_serde::from_slice(&entry.payload)?) +} + fn validate_health_binding_store(path: &Path) -> Result<()> { let entries = SingleFileMessagePackBackingStore::new(path).pull_all_read_only_snapshot()?; if entries.is_empty() { @@ -395,6 +617,13 @@ fn require_only(options: &BTreeMap, names: &[&str]) -> Result<() Ok(()) } +fn allow_only(options: &BTreeMap, names: &[&str]) -> Result<()> { + if let Some(name) = options.keys().find(|name| !names.contains(&name.as_str())) { + bail!("unsupported option --{name}"); + } + Ok(()) +} + fn required<'a>(options: &'a BTreeMap, name: &str) -> Result<&'a str> { options .get(name) @@ -474,7 +703,7 @@ fn normalized(path: &Path) -> Result { } fn usage() -> &'static str { - "Usage: idunn-provision enroll-idunn-identity --private-store \n idunn-provision export-idunn-public-anchor --private-store --public-anchor \n idunn-provision create-daemon-health-trust-binding --output --binding-id --daemon --health-contract --source-runtime --signer-public-key-hex --bound-at-unix-millis --release-binding-required \n idunn-provision add-daemon-health-trust-binding --output --binding-id --daemon --health-contract --source-runtime --signer-public-key-hex --bound-at-unix-millis --release-binding-required \n idunn-provision validate-daemon-health-trust-binding --input \n idunn-provision create-provider-projection-trust-anchor --output --trust-anchor-id --runtime-id --idunn-public-anchor --bound-at-unix-millis --expires-at-unix-millis \n idunn-provision validate-provider-projection-trust-anchor --input --idunn-public-anchor " + "Usage: idunn-provision enroll-idunn-identity --private-store \n idunn-provision export-idunn-public-anchor --private-store --public-anchor \n idunn-provision enroll-deployment-brake-operator --private-store \n idunn-provision export-deployment-brake-operator-anchor --private-store --public-anchor \n idunn-provision deployment-brake-engage --store --runtime-id --owner --reason --observed-at-unix-millis \n idunn-provision deployment-brake-release --store --private-store --runtime-id --owner --reason --authorization-id --release-id --deployment-id --issued-at-unix-millis --expires-at-unix-millis \n idunn-provision deployment-brake-status --store --operator-anchor --runtime-id [--release-id --deployment-id --now-unix-millis ]\n idunn-provision create-daemon-health-trust-binding --output --binding-id --daemon --health-contract --source-runtime --signer-public-key-hex --bound-at-unix-millis --release-binding-required \n idunn-provision add-daemon-health-trust-binding --output --binding-id --daemon --health-contract --source-runtime --signer-public-key-hex --bound-at-unix-millis --release-binding-required \n idunn-provision validate-daemon-health-trust-binding --input \n idunn-provision create-provider-projection-trust-anchor --output --trust-anchor-id --runtime-id --idunn-public-anchor --bound-at-unix-millis --expires-at-unix-millis \n idunn-provision validate-provider-projection-trust-anchor --input --idunn-public-anchor " } #[cfg(test)] @@ -610,16 +839,45 @@ mod tests { let temp = TempDir::new()?; let store = temp.path().join("trust.cc"); let path = store.to_str().unwrap().to_string(); - invoke(&binding_args("create-daemon-health-trust-binding", &path, "one", "daemon-one", "runtime-one"))?; + invoke(&binding_args( + "create-daemon-health-trust-binding", + &path, + "one", + "daemon-one", + "runtime-one", + ))?; let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); - let workers = [("two", "daemon-two", "runtime-two"), ("three", "daemon-three", "runtime-three")].map(|(id, daemon, runtime)| { - let path = path.clone(); let barrier = barrier.clone(); - std::thread::spawn(move || { barrier.wait(); invoke(&binding_args("add-daemon-health-trust-binding", &path, id, daemon, runtime)) }) + let workers = [ + ("two", "daemon-two", "runtime-two"), + ("three", "daemon-three", "runtime-three"), + ] + .map(|(id, daemon, runtime)| { + let path = path.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + barrier.wait(); + invoke(&binding_args( + "add-daemon-health-trust-binding", + &path, + id, + daemon, + runtime, + )) + }) }); barrier.wait(); - let wins = workers.into_iter().map(|worker| worker.join().unwrap()).filter(Result::is_ok).count(); + let wins = workers + .into_iter() + .map(|worker| worker.join().unwrap()) + .filter(Result::is_ok) + .count(); assert!((1..=2).contains(&wins)); - assert_eq!(SingleFileMessagePackBackingStore::new(&store).pull_all_read_only_snapshot()?.len(), 1 + wins); + assert_eq!( + SingleFileMessagePackBackingStore::new(&store) + .pull_all_read_only_snapshot()? + .len(), + 1 + wins + ); Ok(()) } @@ -877,4 +1135,148 @@ mod tests { partial.extend(["--release-binding-required", "sometimes"]); assert!(invoke(&partial).is_err()); } + + #[test] + fn brake_operator_can_engage_then_issue_one_exact_bounded_release() -> Result<()> { + let temp = TempDir::new()?; + let private = temp.path().join("operator-private.cc"); + let anchor = temp.path().join("operator-public.cc"); + let store = temp.path().join("brake.cc"); + invoke(&[ + "enroll-deployment-brake-operator", + "--private-store", + private.to_str().unwrap(), + ])?; + invoke(&[ + "export-deployment-brake-operator-anchor", + "--private-store", + private.to_str().unwrap(), + "--public-anchor", + anchor.to_str().unwrap(), + ])?; + invoke(&[ + "deployment-brake-engage", + "--store", + store.to_str().unwrap(), + "--runtime-id", + "yggdrasil", + "--owner", + "operator/metacrat", + "--reason", + "sleep mode", + "--observed-at-unix-millis", + "100", + ])?; + assert_eq!(read_brake(&store)?.status, "engaged"); + invoke(&[ + "deployment-brake-release", + "--store", + store.to_str().unwrap(), + "--private-store", + private.to_str().unwrap(), + "--runtime-id", + "yggdrasil", + "--owner", + "operator/metacrat", + "--reason", + "one deployment", + "--authorization-id", + "auth/r4", + "--release-id", + "commit-4", + "--deployment-id", + "request-4", + "--issued-at-unix-millis", + "200", + "--expires-at-unix-millis", + "800", + ])?; + invoke(&[ + "deployment-brake-status", + "--store", + store.to_str().unwrap(), + "--operator-anchor", + anchor.to_str().unwrap(), + "--runtime-id", + "yggdrasil", + "--release-id", + "commit-4", + "--deployment-id", + "request-4", + "--now-unix-millis", + "500", + ])?; + assert!( + invoke(&[ + "deployment-brake-status", + "--store", + store.to_str().unwrap(), + "--operator-anchor", + anchor.to_str().unwrap(), + "--runtime-id", + "yggdrasil", + "--release-id", + "substituted", + "--deployment-id", + "request-4", + "--now-unix-millis", + "500" + ]) + .is_err() + ); + Ok(()) + } + + #[test] + fn brake_compare_exchange_rejects_stale_transition_snapshot() -> Result<()> { + let temp = TempDir::new()?; + let store_path = temp.path().join("brake.cc"); + let base = IdunnDeploymentBrakeRecord { + schema_version: IDUNN_DEPLOYMENT_BRAKE_SCHEMA.into(), + brake_id: IDUNN_DEPLOYMENT_BRAKE_ID.into(), + authority: IDUNN_DEPLOYMENT_BRAKE_AUTHORITY.into(), + runtime_id: "yggdrasil".into(), + status: "engaged".into(), + scope: IDUNN_DEPLOYMENT_BRAKE_SCOPE.into(), + reason: "first".into(), + observed_at_unix_millis: 100, + expires_at_unix_millis: None, + authorization_id: None, + authorization_purpose: None, + authorized_release_id: None, + authorized_deployment_id: None, + authorized_by: None, + authorization_issued_at_unix_millis: None, + authorization_expires_at_unix_millis: None, + signature_algorithm: None, + signature: None, + private_state_exposed: false, + updated_by: "operator/a".into(), + }; + replace_brake(&store_path, base.clone(), 100)?; + let stale = SingleFileMessagePackBackingStore::new(&store_path) + .pull_all_read_only_snapshot()? + .remove(0); + let mut changed = base.clone(); + changed.reason = "racing writer".into(); + changed.observed_at_unix_millis = 101; + replace_brake(&store_path, changed, 101)?; + let candidate = typed_envelope( + IDUNN_DEPLOYMENT_BRAKE_ID, + &base, + IDUNN_DEPLOYMENT_BRAKE_SCHEMA, + 102, + )?; + assert!( + !SingleFileMessagePackBackingStore::new(&store_path).compare_exchange( + &[CultCacheExpectedEnvelope { + r#type: IdunnDeploymentBrakeRecord::TYPE.into(), + key: IDUNN_DEPLOYMENT_BRAKE_ID.into(), + current: Some(stale) + }], + &[candidate] + )? + ); + Ok(()) + } } diff --git a/scripts/linux/idunn-yggdrasil.service b/scripts/linux/idunn-yggdrasil.service index d17ad1c..252526b 100644 --- a/scripts/linux/idunn-yggdrasil.service +++ b/scripts/linux/idunn-yggdrasil.service @@ -10,7 +10,7 @@ User=idunn Group=idunn # Operational cancellation ceiling only. Target daemons own readiness through # typed health; this duration must never be interpreted as deployment success. -ExecStart=/usr/local/bin/idunn --swarm-profile yggdrasil-local --repo-root /srv/odin/source --store /var/lib/gamecult/idunn/yggdrasil.keepalive.cc --release-authority-store /srv/bifrost/state/repository-release-authority.cc --rudp-health-bind 10.77.0.1:17870 --trusted-epiphany-health-identity-store /etc/gamecult/idunn/epiphany-health-identity.ccmp --command-timeout-seconds 21600 --execute +ExecStart=/usr/local/bin/idunn --swarm-profile yggdrasil-local --repo-root /srv/odin/source --store /var/lib/gamecult/idunn/yggdrasil.keepalive.cc --release-authority-store /srv/bifrost/state/repository-release-authority.cc --deployment-brake-store /var/lib/gamecult/idunn/deployment-brake.cc --deployment-brake-operator-anchor /etc/gamecult/idunn/deployment-brake-operator-anchor.cc --rudp-health-bind 10.77.0.1:17870 --trusted-epiphany-health-identity-store /etc/gamecult/idunn/epiphany-health-identity.ccmp --command-timeout-seconds 21600 --execute Restart=always RestartSec=5 TimeoutStopSec=20 diff --git a/vendor/cultnet-rs b/vendor/cultnet-rs index a1001dc..402f6e7 160000 --- a/vendor/cultnet-rs +++ b/vendor/cultnet-rs @@ -1 +1 @@ -Subproject commit a1001dc855aa4de04859d01d8be4c966f510f557 +Subproject commit 402f6e7f0b938a251fc9443d734a6c86d5c1b40e