diff --git a/crates/idunn-daemon/src/main.rs b/crates/idunn-daemon/src/main.rs index 0f28d2b..77e6af7 100644 --- a/crates/idunn-daemon/src/main.rs +++ b/crates/idunn-daemon/src/main.rs @@ -673,6 +673,12 @@ fn run_swarm(options: &SwarmOptions, common: &CommonOptions) -> Result<()> { let store_lock = Arc::new(Mutex::new(())); let now = timestamp()?; + let recovered = terminalize_interrupted_deployment_requests(common, &store_lock, &now)?; + if recovered > 0 { + println!( + "Idunn terminalized {recovered} deployment request(s) interrupted by the prior daemon incarnation." + ); + } publish_runtime_transport_check(common, &store_lock, &now)?; start_rudp_health_ingress(common, &store_lock, &now)?; publish_surgery_plans(&options.profile, &targets, common, &store_lock, &now)?; @@ -1434,6 +1440,123 @@ fn persist_current_deployment_request( )) } +/// On daemon startup no deployment actuator from the prior process can still +/// be an Idunn-owned child. Close any exact current request that lacks a result +/// before this incarnation launches workers. The current head remains durable +/// history; its exact failed result permits ordinary CAS supersession. +fn terminalize_interrupted_deployment_requests( + options: &CommonOptions, + store_lock: &Arc>, + completed_at: &str, +) -> Result { + let _store_guard = store_lock + .lock() + .map_err(|_| anyhow!("Idunn store lock is poisoned"))?; + let backing = SingleFileMessagePackBackingStore::new(&options.store_path); + let opening = backing.pull_all()?; + let heads = opening + .iter() + .filter(|entry| entry.r#type == IdunnCurrentDeploymentRequestRecord::TYPE) + .collect::>(); + let mut recovered = 0; + for head_envelope in heads { + let head: IdunnCurrentDeploymentRequestRecord = + rmp_serde::from_slice(&head_envelope.payload)?; + if head.daemon_id.trim().is_empty() + || head.request_id.trim().is_empty() + || head.sequence == 0 + || head_envelope.key != head.daemon_id + { + return Err(anyhow!( + "current deployment request head is invalid during startup recovery" + )); + } + let request_matches = opening + .iter() + .filter(|entry| { + entry.r#type == IdunnDeploymentRequestRecord::TYPE + && entry.key == head.request_id + }) + .collect::>(); + if request_matches.len() != 1 { + return Err(anyhow!( + "current deployment request {} lacks one exact request during startup recovery", + head.request_id + )); + } + let request_envelope = request_matches[0]; + let request: IdunnDeploymentRequestRecord = + rmp_serde::from_slice(&request_envelope.payload)?; + if request.request_id != head.request_id || request.daemon_id != head.daemon_id { + return Err(anyhow!( + "current deployment request is substituted during startup recovery" + )); + } + let result_key = format!("result:{}", head.request_id); + let result_matches = opening + .iter() + .filter(|entry| { + entry.r#type == IdunnDeploymentResultRecord::TYPE && entry.key == result_key + }) + .collect::>(); + if result_matches.len() > 1 { + return Err(anyhow!( + "current deployment request has duplicate results during startup recovery" + )); + } + if let Some(result_envelope) = result_matches.first() { + let result: IdunnDeploymentResultRecord = + rmp_serde::from_slice(&result_envelope.payload)?; + if result.result_id != result_key + || result.request_id != head.request_id + || result.daemon_id != head.daemon_id + || !matches!(result.state.as_str(), "succeeded" | "failed" | "skipped") + { + return Err(anyhow!( + "current deployment request lacks an exact terminal result during startup recovery" + )); + } + continue; + } + let result = IdunnDeploymentResultRecord { + result_id: result_key.clone(), + request_id: head.request_id.clone(), + daemon_id: head.daemon_id.clone(), + state: "failed".to_string(), + detail: "deployment actuator was interrupted when the owning Idunn daemon incarnation stopped" + .to_string(), + completed_at: completed_at.to_string(), + }; + let expected = [ + CultCacheExpectedEnvelope { + key: head.daemon_id.clone(), + r#type: IdunnCurrentDeploymentRequestRecord::TYPE.into(), + current: Some(head_envelope.clone()), + }, + CultCacheExpectedEnvelope { + key: head.request_id.clone(), + r#type: IdunnDeploymentRequestRecord::TYPE.into(), + current: Some(request_envelope.clone()), + }, + CultCacheExpectedEnvelope { + key: result_key.clone(), + r#type: IdunnDeploymentResultRecord::TYPE.into(), + current: None, + }, + ]; + if !backing.compare_exchange( + &expected, + &[typed_envelope(&result_key, &result, completed_at)?], + )? { + return Err(anyhow!( + "interrupted deployment request changed during startup recovery" + )); + } + recovered += 1; + } + Ok(recovered) +} + fn release_authority_id( repository_full_name: &str, upstream_ref: &str, @@ -7423,6 +7546,57 @@ mod tests { assert!(health_from_rudp_message(&forged, &forged_options).is_err()); } + #[test] + fn startup_terminalizes_interrupted_deployment_before_supersession() { + let (_message, options, _root) = signed_epiphany_health_fixture(1); + let lock = Arc::new(Mutex::new(())); + let prior = with_store_node(&options, &lock, |node| { + node.get::( + "deploy:yggdrasil-epiphany:fixture", + )? + .ok_or_else(|| anyhow!("fixture request missing")) + }) + .unwrap(); + + assert_eq!( + terminalize_interrupted_deployment_requests( + &options, + &lock, + "2026-07-16T00:02:00Z", + ) + .unwrap(), + 1 + ); + let result = with_store_node(&options, &lock, |node| { + node.get::( + "result:deploy:yggdrasil-epiphany:fixture", + )? + .ok_or_else(|| anyhow!("recovered result missing")) + }) + .unwrap(); + assert_eq!(result.request_id, prior.request_id); + assert_eq!(result.daemon_id, prior.daemon_id); + assert_eq!(result.state, "failed"); + assert_eq!(result.completed_at, "2026-07-16T00:02:00Z"); + assert!(result.detail.contains("owning Idunn daemon incarnation stopped")); + assert_eq!( + terminalize_interrupted_deployment_requests( + &options, + &lock, + "2026-07-16T00:03:00Z", + ) + .unwrap(), + 0 + ); + + let mut successor = prior; + successor.request_id = "deploy:yggdrasil-epiphany:successor".into(); + successor.requested_at = "2026-07-16T00:04:00Z".into(); + let head = persist_current_deployment_request(&options, &lock, &successor).unwrap(); + assert_eq!(head.request_id, successor.request_id); + assert_eq!(head.sequence, 2); + } + #[test] fn pinned_epiphany_health_identity_load_is_read_only() { let (_message, options, _root) = signed_epiphany_health_fixture(1);