diff --git a/src/control_plane.rs b/src/control_plane.rs index 5289c67..3d1b809 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -30,7 +30,8 @@ use sha2::{Digest, Sha256}; use uuid::Uuid; use crate::deployment::{ - DependencyKind, OperatorBinding, RouteBinding, WorkloadBinding, capability_compatible, + DependencyKind, OperatorBinding, RolloutStrategy, RouteBinding, WorkloadBinding, + capability_compatible, }; use crate::deployment_plan::{ CompiledDeploymentPlan, DependencyProviderAuthority, SealedRelease, compile_deployment_plan, @@ -44,7 +45,7 @@ use crate::drivers::{ }; use crate::host_actuator::{ HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver, - IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service, + HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service, }; const DEPLOYMENT_COMMAND_SCHEMA: &str = "idunn.deployment_command.v2"; @@ -721,6 +722,15 @@ struct DeploymentTransaction { } impl DeploymentTransaction { + /// Whether this transaction's binding declares stop-then-start. A + /// transaction without a plan yet cannot, so it reads as false. + fn rollout_stops_incumbent_first(&self) -> bool { + self.plan + .as_ref() + .and_then(|plan| plan.parsed_inputs().ok()) + .is_some_and(|(_, binding)| binding.rollout.strategy == RolloutStrategy::StopThenStart) + } + fn new( command: &DeploymentCommand, target: String, @@ -3265,8 +3275,15 @@ impl Engine { && stored.value.blocks_new_target_mutation() }); if blocker.is_some_and(|stored| { - stored.value.phase >= DeploymentPhase::Fencing - && stored.value.phase < DeploymentPhase::Complete + let owns_from = if stored.value.rollout_stops_incumbent_first() { + // The incumbent is stopped at Starting by design; a + // continuity restart in that window would fight the + // candidate for the same host resources. + DeploymentPhase::Starting + } else { + DeploymentPhase::Fencing + }; + stored.value.phase >= owns_from && stored.value.phase < DeploymentPhase::Complete }) { continue; } @@ -3392,6 +3409,18 @@ impl Engine { Some(observation) } } + Err(error) if error.downcast_ref::().is_some() => { + // The host cannot be asked right now (its actuator is + // between sessions, or Idunn just restarted). That is + // silence, not a death; counting it as one burned every + // continuity attempt in the reattach window on + // 2026-09-11 and left a healthy Muninn demoted. + eprintln!( + "Idunn cannot observe admitted {} this tick: {error:#}", + current.value.target + ); + continue; + } Err(error) => { operational_error = Some(error); None @@ -3914,6 +3943,14 @@ impl Engine { if current.value.workload.is_none() { let now = now_millis()?; let activation = required(¤t.value.activation, "activation")?; + if current.value.rollout_stops_incumbent_first() { + let snapshot = ControlSnapshot::read(&self.options.state_store)?; + if let Some(incumbent) = self.exact_incumbent(&snapshot, ¤t.value)? { + self.workload_for(&incumbent.value.plan)? + .stop(&incumbent.value.workload) + .context("stopping the incumbent before the candidate starts")?; + } + } let observation = self .workload_for(plan)? .start_prepared(plan, release, installed, expected, activation)?; diff --git a/src/deployment.rs b/src/deployment.rs index d499def..23ecac3 100644 --- a/src/deployment.rs +++ b/src/deployment.rs @@ -713,6 +713,12 @@ pub struct RolloutBinding { #[serde(rename_all = "kebab-case")] pub enum RolloutStrategy { CandidateThenPromote, + /// The admitted incumbent is stopped before the candidate starts. For a + /// workload that owns fixed host resources no route can multiplex (one + /// desktop, one capture device, fixed advertised ports), two incarnations + /// cannot overlap; the honest strategy says so instead of failing at + /// bind time. Only a route-less host-actuator workload may declare it. + StopThenStart, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -1194,7 +1200,13 @@ impl OperatorBinding { ); return self.validate_shared_tail(); } - WorkloadBinding::SystemdTransient(workload) => workload, + WorkloadBinding::SystemdTransient(workload) => { + ensure!( + self.rollout.strategy == RolloutStrategy::CandidateThenPromote, + "a systemd-transient workload rolls out candidate-then-promote" + ); + workload + } }; let needs_state_group = workload.state_root.is_some() || !workload.read_write_paths.is_empty(); diff --git a/src/host_actuator.rs b/src/host_actuator.rs index e017ec5..90592be 100644 --- a/src/host_actuator.rs +++ b/src/host_actuator.rs @@ -299,6 +299,37 @@ impl HostActuatorEnvelope { } } +/// The host could not be asked: nothing is attached under its name, or the +/// session dropped before it answered. This is not evidence about the +/// workload either way. Continuity must not count it as a death, and a +/// transaction must retry it, which the phase machine already does for +/// any error. +#[derive(Debug)] +pub struct HostUnobservable { + pub host: String, + pub reason: String, +} + +impl std::fmt::Display for HostUnobservable { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "host actuator {} is unobservable: {}", + self.host, self.reason + ) + } +} + +impl std::error::Error for HostUnobservable {} + +fn unobservable(host: &str, reason: impl Into) -> anyhow::Error { + HostUnobservable { + host: host.into(), + reason: reason.into(), + } + .into() +} + /// One attached host: the session that authenticated as it and the nonce /// its messages carry. A host that connects again replaces this wholesale. struct AttachedHost { @@ -506,10 +537,9 @@ impl HostActuatorHub { ) -> Result { self.service(anchors, signer)?; let (session, nonce, sequence) = { - let attached = self - .attached - .get_mut(host) - .with_context(|| format!("host actuator {host} is not attached"))?; + let Some(attached) = self.attached.get_mut(host) else { + return Err(unobservable(host, "not attached")); + }; let sequence = attached.next_sequence; attached.next_sequence += 1; (attached.session.clone(), attached.nonce.clone(), sequence) @@ -521,12 +551,17 @@ impl HostActuatorHub { loop { self.service(anchors, signer)?; let Some(attached) = self.attached.get_mut(host) else { - bail!("host actuator {host} detached while a request was outstanding"); + return Err(unobservable( + host, + "detached while a request was outstanding", + )); }; - ensure!( - attached.session.session_generation == session.session_generation, - "host actuator {host} reconnected while a request was outstanding" - ); + if attached.session.session_generation != session.session_generation { + return Err(unobservable( + host, + "reconnected while a request was outstanding", + )); + } if let Some(index) = attached .reports .iter() @@ -543,10 +578,9 @@ impl HostActuatorHub { attached .reports .retain(|(answered, _)| *answered > sequence); - ensure!( - Instant::now() < deadline, - "host actuator {host} did not answer within {timeout:?}" - ); + if Instant::now() >= deadline { + return Err(unobservable(host, format!("no answer within {timeout:?}"))); + } thread::sleep(REQUEST_POLL); } }