diff --git a/src/control_plane.rs b/src/control_plane.rs index 28185cc..00a7775 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -59,6 +59,7 @@ const ADMITTED_GENERATION_SCHEMA_V2: &str = "idunn.admitted_generation.v2"; /// The capability whose declaration makes a target Odin-correlated. pub(crate) const ODIN_RENDEZVOUS_CAPABILITY: &str = "odin.verse-rendezvous"; const TARGET_SUPERVISION_SCHEMA: &str = "idunn.target_supervision.v1"; +const EXPIRY_REQUEST_SCHEMA: &str = "idunn.expiry_request.v1"; /// How many times continuity will restart one target inside one window before /// it concludes the target itself is the problem. More than one because a /// start can fail for a passing reason -- a port still held, a peer not yet up @@ -75,10 +76,13 @@ const CONTINUITY_RESTART_BACKOFF_MILLIS: u64 = 5_000; /// healthy target performs none; a legitimate deployment performs one or two. /// Only deployments count: a Survival actuation is recorded beside this /// ceiling, never counted against it and never refused by it, so a target -/// whose continuity spent its restarts can still be rescued by a deploy. Each -/// Survival path has its own bound instead: a continuity's actuations the -/// restart log, an admitted-route repair the challenge backoff, and a -/// deployment's rollback or withdrawal the Forward change it undoes. +/// whose continuity spent its restarts can still be rescued by a deploy. +/// Survival is bounded elsewhere: a continuity's actuations by the restart log +/// times what one transaction can do before its phase deadline ends it (none, +/// over its admitted route), and an admitted-route repair by the challenge +/// backoff. A withdrawal that keeps failing retries, reloading each time, +/// under the resume backoff alone: the abort has no deadline. Nothing bounds +/// reloads across the host. const ROUTE_ACTUATION_CEILING: usize = 12; const ROUTE_ACTUATION_WINDOW_MILLIS: u64 = 3_600_000; /// How many Survival route actuation times a target keeps, for `status` @@ -152,6 +156,51 @@ impl DeploymentCommand { } } +/// An operator's request that a live transaction's post-fencing phase end now +/// (`idunn expire`). The CLI only writes this record. The daemon's deadline +/// resolver reads it as a deadline that has passed, and the record is deleted +/// once its transaction no longer has a phase to end. No CLI writes a +/// transaction field. +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "idunn.expiry_request", schema = "idunn.expiry_request.v1")] +struct ExpiryRequest { + #[cultcache(key = 0)] + schema_version: String, + #[cultcache(key = 1)] + command_id: String, + #[cultcache(key = 2)] + requested_by: String, + #[cultcache(key = 3)] + requested_at_unix_millis: u64, +} + +impl ExpiryRequest { + fn validate(&self) -> Result<()> { + ensure!( + self.schema_version == EXPIRY_REQUEST_SCHEMA, + "expiry request schema is unsupported" + ); + require_id(&self.command_id, "expiry request command id")?; + require_value(&self.requested_by, "expiry requester")?; + ensure!( + self.requested_at_unix_millis > 0, + "expiry request has no request time" + ); + Ok(()) + } +} + +fn expiry_request_envelope(value: &ExpiryRequest, now: u64) -> Result { + value.validate()?; + typed_envelope( + &value.command_id, + ExpiryRequest::TYPE, + EXPIRY_REQUEST_SCHEMA, + value, + now, + ) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] #[repr(u8)] @@ -511,6 +560,15 @@ impl LeasingEvidence { } } + /// The lease Idunn issued, granted or only prepared. A prepared lease may + /// already be on disk: the physical grant runs before `Granted` is durable. + fn issued_lease(&self) -> Option<&IdunnProcessWriteLeaseRecord> { + match self { + Self::SkippedStateless => None, + Self::Prepared { lease, .. } | Self::Granted { lease, .. } => Some(lease), + } + } + fn prepared_lease(&self) -> Option<(&IdunnProcessWriteLeaseRecord, &str)> { match self { Self::Prepared { @@ -820,6 +878,22 @@ enum AdoptionSource { } impl LeaseAdoptionEvidence { + /// The adoption a stateful candidate's Ready proves: its signed presence + /// held exactly the granted lease. `None` for a stateless candidate. + fn with_ready( + granted_lease_sha256: Option<&str>, + signed_presence_sha256: &str, + source: AdoptionSource, + observed_at_unix_millis: u64, + ) -> Option { + granted_lease_sha256.map(|lease| Self { + write_lease_sha256: lease.to_owned(), + signed_presence_sha256: signed_presence_sha256.to_owned(), + source, + observed_at_unix_millis, + }) + } + /// Whether this evidence is about exactly the lease that was granted. fn names(&self, leasing: &LeasingEvidence) -> bool { leasing.lease_sha256() == Some(self.write_lease_sha256.as_str()) @@ -870,6 +944,17 @@ impl PhaseDeadline { } } +/// How the deadline resolver ends a post-fencing phase. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PhaseEnd { + /// Abort through the post-fencing path. `lease_not_adopted` names a + /// granted lease with no adoption evidence. + Abort { lease_not_adopted: bool }, + /// A stateful deployment whose candidate holds an issued lease: the phase + /// is reported and not resolved. + OperatorRequired, +} + /// How a post-fencing failure recovers. `RestoreIncumbent` is what a /// post-fencing abort has always done and is the default for records that /// predate the field. @@ -1488,6 +1573,46 @@ impl DeploymentTransaction { .and_then(|plan| PhaseDeadline::entering(phase, plan, now)); } + /// A live transaction past the fence that is not already being aborted: + /// the only kind whose phase a deadline, or an operator's expiry, ends. + fn resolves_by_deadline(&self) -> bool { + self.completion.is_none() + && self.pre_fencing_abort.is_none() + && self.post_fencing_abort.is_none() + && (DeploymentPhase::Fencing..=DeploymentPhase::Committing).contains(&self.phase) + } + + /// What an ended post-fencing phase resolves to (Q3). A candidate that was + /// never issued a write lease cannot have written, so the transaction + /// aborts to the incumbent through the post-fencing path, stateless or not. + /// A continuity always aborts: its candidate is the admitted release, and + /// supervision restarts that release under the continuity backoff. A + /// deployment whose candidate was issued a lease may have written under a + /// new state contract, and restoring the incumbent over that is not + /// Idunn's call: it is left for the operator. + fn phase_end(&self) -> PhaseEnd { + let leasing = self.leasing.as_ref(); + let granted = leasing.and_then(LeasingEvidence::lease).is_some(); + let issued = leasing.and_then(LeasingEvidence::issued_lease).is_some(); + match self.command_kind { + CommandKind::Deploy if issued => PhaseEnd::OperatorRequired, + _ => PhaseEnd::Abort { + lease_not_adopted: granted && self.lease_adoption.is_none(), + }, + } + } + + /// The recovery a completed post-fencing abort names. A continuity whose + /// candidate adopted its lease restarts the admitted release; everything + /// else restores the incumbent. + fn terminal_recovery(&self) -> TerminalRecovery { + if self.command_kind == CommandKind::Continuity && self.lease_adoption.is_some() { + TerminalRecovery::RestartAdmitted + } else { + TerminalRecovery::RestoreIncumbent + } + } + /// 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 { @@ -2444,6 +2569,11 @@ enum Command { command_id: String, requested_by: String, }, + Expire { + state_store: PathBuf, + command_id: String, + requested_by: String, + }, Validate { recipe: PathBuf, binding: Option, @@ -2475,6 +2605,11 @@ pub fn run(args: impl Iterator) -> Result<()> { command_id, requested_by, } => cancel(&state_store, &command_id, &requested_by), + Command::Expire { + state_store, + command_id, + requested_by, + } => expire(&state_store, &command_id, &requested_by), Command::Validate { recipe, binding } => validate(&recipe, binding.as_deref()), } } @@ -2526,6 +2661,7 @@ fn parse(args: impl Iterator) -> Result { "up" => parse_up(args), "status" => parse_status(args), "cancel" => parse_cancel(args), + "expire" => parse_expire(args), "validate" => parse_validate(args), "--help" | "-h" | "help" => bail!(usage()), _ => bail!("unknown Idunn command {command:?}\n\n{}", usage()), @@ -2642,6 +2778,68 @@ fn parse_status(mut args: impl Iterator) -> Result { }) } +fn parse_expire(mut args: impl Iterator) -> Result { + let command_id = args + .next() + .ok_or_else(|| anyhow!("idunn expire requires a command id"))?; + require_id(&command_id, "command id")?; + let mut state_store = RuntimeOptions::default().state_store; + let mut requested_by = env::var("SUDO_USER") + .or_else(|_| env::var("USER")) + .or_else(|_| env::var("USERNAME")) + .unwrap_or_else(|_| "operator".into()); + while let Some(argument) = args.next() { + match argument.as_str() { + "--state-store" => state_store = path_value(&mut args, &argument)?, + "--requested-by" => requested_by = string_value(&mut args, &argument)?, + "--help" | "-h" => bail!(usage()), + _ => bail!("unknown Idunn expire option {argument:?}"), + } + } + require_value(&requested_by, "expiry requester")?; + Ok(Command::Expire { + state_store, + command_id, + requested_by, + }) +} + +/// Ask the daemon to end a live transaction's post-fencing phase now. This +/// writes a request and nothing else: the daemon's deadline resolver ends the +/// phase exactly as it ends one whose deadline has passed. +fn expire(store_path: &Path, command_id: &str, requested_by: &str) -> Result<()> { + let snapshot = ControlSnapshot::read(store_path)?; + let mut live = snapshot + .transactions + .iter() + .filter(|stored| stored.value.command_id == command_id && !stored.value.is_terminal()); + let current = live + .next() + .context("command has no live transaction to expire")?; + ensure!( + live.next().is_none(), + "command has multiple live transactions" + ); + ensure!( + current.value.resolves_by_deadline(), + "only a live transaction past the fence, and not already aborting, can be expired; before the fence, use idunn cancel" + ); + let now = now_millis()?; + let request = ExpiryRequest { + schema_version: EXPIRY_REQUEST_SCHEMA.into(), + command_id: command_id.to_owned(), + requested_by: requested_by.to_owned(), + requested_at_unix_millis: now, + }; + ensure!( + SingleFileMessagePackBackingStore::new(store_path) + .insert_entry_if_absent(expiry_request_envelope(&request, now)?)?, + "an expiry request for this command is already waiting" + ); + println!("{command_id} expiry requested; Idunn ends its phase as an expired deadline"); + Ok(()) +} + fn parse_cancel(mut args: impl Iterator) -> Result { let command_id = args .next() @@ -2787,6 +2985,7 @@ struct ControlSnapshot { transactions: Vec>, admitted: Vec>, targets: Vec>, + expiry_requests: Vec>, } impl ControlSnapshot { @@ -2842,6 +3041,19 @@ impl ControlSnapshot { ); snapshot.targets.push(Stored { envelope, value }); } + ExpiryRequest::TYPE => { + ensure!( + envelope.schema_id.as_deref() == Some(EXPIRY_REQUEST_SCHEMA), + "Idunn control store contains an unsupported expiry request" + ); + let value: ExpiryRequest = decode_record(&envelope)?; + value.validate()?; + ensure!( + envelope.key == value.command_id, + "expiry request key is not its command" + ); + snapshot.expiry_requests.push(Stored { envelope, value }); + } _ => bail!("Idunn control store contains a foreign document"), } } @@ -5172,7 +5384,7 @@ impl Engine { /// own command, so one target brake cannot suspend unrelated continuity. fn resume_one_transaction(&self) -> Result { let snapshot = ControlSnapshot::read(&self.options.state_store)?; - let mut progressed = false; + let mut progressed = self.retire_spent_expiry_requests(&snapshot)?; let mut candidates = snapshot .transactions .iter() @@ -5212,6 +5424,9 @@ impl Engine { /// transaction would report progress every tick. fn resume_candidate(&self, current: &Stored) -> Result { let id = current.value.transaction_id.as_str(); + if let Some(moved) = self.resolve_phase_deadline(current)? { + return Ok(moved); + } if self .resume_backoff .lock() @@ -5273,6 +5488,8 @@ impl Engine { // finish. self.begin_post_fencing_abort(latest, error)?; } else { + // Resumable only until the phase deadline: the resolver ends + // the phase before the next step once it has passed. self.record_resumable_error(latest, &error)?; } } @@ -6801,8 +7018,15 @@ impl Engine { self.record_gate_wait(current, "candidate is still warming") } CandidateAnswer::Answered { evidence, .. } => { + let adoption = LeaseAdoptionEvidence::with_ready( + lease, + &evidence.canonical_sha256, + AdoptionSource::Direct, + evidence.admitted_at_unix_millis, + ); self.persist_same_phase(current, |next| { next.ready = Some(ReadinessEvidence::RouteProof { evidence }); + next.lease_adoption = adoption; Ok(()) }) } @@ -6831,6 +7055,20 @@ impl Engine { ensure!(semantic_ready, "stored Ready receipt changed meaning"); } else if semantic_ready { let evidence = latest.clone(); + // A Ready stateful correlation carries exactly the current + // lease (the authenticator refuses one that does not). + let adoption = authenticated + .record() + .signed_presence_sha256 + .as_deref() + .and_then(|presence| { + LeaseAdoptionEvidence::with_ready( + current_lease, + presence, + AdoptionSource::OdinTopology, + evidence.admitted_at_unix_millis, + ) + }); let _token = SequenceAdmittedReady { transaction_id: admitted.value.transaction_id.clone(), evidence: evidence.clone(), @@ -6839,6 +7077,7 @@ impl Engine { }; return self.persist_same_phase(&admitted, |next| { next.ready = Some(ReadinessEvidence::OdinCorrelated { evidence }); + next.lease_adoption = adoption; Ok(()) }); } else { @@ -6991,14 +7230,14 @@ impl Engine { .as_ref() .and_then(LeasingEvidence::lease_sha256) .map(str::to_owned); + // A newer sequence admitted here changed the record, and commit + // goes on from that record rather than returning: returning let + // Odin's next re-stamp abort every attempt (F16). let Some((ready_current, authenticated)) = self.admit_latest_topology(current, current_lease_sha256.as_deref())? else { return Ok(()); }; - if ready_current.envelope != current.envelope { - return Ok(()); - } ensure!( is_semantic_ready(&authenticated), "latest Odin observation is not Ready at admission commit" @@ -7071,14 +7310,13 @@ impl Engine { .as_ref() .and_then(LeasingEvidence::lease_sha256) .map(str::to_owned); + // The final proof's own presence makes Odin re-stamp, so this + // admission usually writes. Commit is from the record it wrote. let Some((commit_current, authenticated)) = self.admit_latest_topology(&ready_current, current_lease_sha256.as_deref())? else { return Ok(()); }; - if commit_current.envelope != ready_current.envelope { - return Ok(()); - } ensure!( is_semantic_ready(&authenticated), "latest Odin observation is not Ready after the final admission challenge" @@ -8329,6 +8567,90 @@ impl Engine { replace_transaction(&self.options.state_store, current, &next) } + /// Every post-fencing phase ends. Before a live transaction past the + /// fence runs its step, a phase whose deadline has passed, or whose + /// command an operator asked to expire, is resolved as `phase_end` says. + /// Returns `Some(moved)` when it decided the tick, `None` to run the step. + /// + /// The deadline is written only by `enter_phase`, so nothing here or in a + /// resume path extends it. A record lifted from before deadlines existed + /// has none; its phase is timed once, from the first time this sees it. + fn resolve_phase_deadline( + &self, + current: &Stored, + ) -> Result> { + let transaction = ¤t.value; + if !transaction.resolves_by_deadline() { + return Ok(None); + } + let now = now_millis()?; + let Some(deadline) = transaction.phase_deadline else { + let mut next = transaction.clone(); + next.enter_phase(transaction.phase, now); + if next.phase_deadline.is_none() { + return Ok(None); + } + replace_transaction(&self.options.state_store, current, &next)?; + return Ok(Some(true)); + }; + let requested = ControlSnapshot::read(&self.options.state_store)? + .expiry_requests + .iter() + .any(|request| request.value.command_id == transaction.command_id); + let cause = if now >= deadline.deadline_at_unix_millis { + format!( + "the {:?} phase passed its deadline at unix ms {}", + transaction.phase, deadline.deadline_at_unix_millis + ) + } else if requested { + format!("the {:?} phase was expired by the operator", transaction.phase) + } else { + return Ok(None); + }; + match transaction.phase_end() { + PhaseEnd::Abort { lease_not_adopted } => { + let error = if lease_not_adopted { + anyhow!("lease-not-adopted: {cause}, and the granted write lease has no adoption evidence") + } else { + anyhow!("{cause}") + }; + self.begin_post_fencing_abort(current, error)?; + Ok(Some(true)) + } + PhaseEnd::OperatorRequired => { + self.report_once( + &format!("deadline:{}", transaction.transaction_id), + format!( + "Idunn leaves {} to the operator: {cause}, and its stateful candidate was issued a write lease", + transaction.transaction_id + ), + ); + Ok(None) + } + } + } + + /// Delete expiry requests whose command has no phase left to end: it was + /// resolved, it finished, or its command is gone. + fn retire_spent_expiry_requests(&self, snapshot: &ControlSnapshot) -> Result { + let spent = snapshot + .expiry_requests + .iter() + .filter(|request| { + !snapshot.transactions.iter().any(|stored| { + stored.value.command_id == request.value.command_id + && stored.value.resolves_by_deadline() + }) + }) + .map(|request| request.envelope.clone()) + .collect::>(); + if spent.is_empty() { + return Ok(false); + } + SingleFileMessagePackBackingStore::new(&self.options.state_store) + .delete_batch_if_unchanged(&spent) + } + fn record_resumable_error( &self, current: &Stored, @@ -8427,11 +8749,14 @@ impl Engine { if abort.lease_withdrawal == CleanupEvidence::Pending { let expected = required(¤t.value.expected, "Expected projection")?; let activation = required(¤t.value.activation, "activation")?; + // A prepared lease is withdrawn too: its physical grant may have + // landed before `Granted` was durable. Revoking an absent lease, + // and withdrawing an absent projection, are no-ops. let lease = current .value .leasing .as_ref() - .and_then(LeasingEvidence::lease) + .and_then(LeasingEvidence::issued_lease) .context("post-fencing abort lost the lease it must withdraw")?; let binding = current.value.plan.as_ref().unwrap().parsed_inputs()?.1; let lease_path = binding @@ -8540,7 +8865,7 @@ impl Engine { next.last_error = Some(abort.error.clone()); next.completion = Some(TransactionCompletion::FailedAfterFencing { error: abort.error.clone(), - recovery: TerminalRecovery::RestoreIncumbent, + recovery: current.value.terminal_recovery(), }); replace_transaction(&self.options.state_store, current, &next) } @@ -8780,7 +9105,7 @@ fn post_fencing_abort_intent( lease_withdrawal: if transaction .leasing .as_ref() - .and_then(LeasingEvidence::lease) + .and_then(LeasingEvidence::issued_lease) .is_some() { CleanupEvidence::Pending @@ -9072,6 +9397,7 @@ fn usage() -> &'static str { idunn up [--state-store PATH] [--no-wait]\n\ idunn status [--state-store PATH] [--command ID]\n\ idunn cancel [--state-store PATH]\n\ + idunn expire [--state-store PATH]\n\ idunn validate --recipe PATH [--binding PATH]\n\n\ Recipes describe capability and process requirements. Idunn seals exact\n\ source and artifacts, admits one incarnation, and delegates execution to\n\ @@ -9215,6 +9541,7 @@ mod tests { struct RouteStubs { calls: PathBuf, reload_fails: PathBuf, + ufw_delete_fails: PathBuf, } #[cfg(unix)] @@ -9226,6 +9553,7 @@ mod tests { std::fs::create_dir_all(&dir)?; let calls = dir.join("calls"); let reload_fails = dir.join("reload-fails"); + let ufw_delete_fails = dir.join("ufw-delete-fails"); let program = |name: &str, refuse: &str| -> Result { let path = dir.join(name); std::fs::write( @@ -9248,10 +9576,23 @@ mod tests { reload_fails.display() ), )?, - ufw: program("ufw", "")?, + ufw: program( + "ufw", + &format!( + "if [ \"$1\" = delete ] && [ -e '{}' ]; then exit 1; fi\n", + ufw_delete_fails.display() + ), + )?, preflight_root: root.join("route-preflight"), }; - Ok((actuators, Self { calls, reload_fails })) + Ok(( + actuators, + Self { + calls, + reload_fails, + ufw_delete_fails, + }, + )) } fn refuse_reloads(&self) -> Result<()> { @@ -9259,6 +9600,12 @@ mod tests { Ok(()) } + /// `ufw delete` exits non-zero with an error that is not "non-existent". + fn refuse_ufw_deletes(&self) -> Result<()> { + std::fs::write(&self.ufw_delete_fails, b"x")?; + Ok(()) + } + /// How many logged calls begin with `prefix`, e.g. `systemctl reload`. fn count(&self, prefix: &str) -> usize { std::fs::read_to_string(&self.calls) @@ -10406,6 +10753,7 @@ mod tests { schema_id: Some(DEPLOYMENT_TRANSACTION_SCHEMA.into()), }; let snapshot = ControlSnapshot { + expiry_requests: Vec::new(), commands: Vec::new(), transactions: vec![ Stored { @@ -10821,6 +11169,7 @@ mod tests { let one = DeploymentTransaction::new(&command, "ghostlight".into(), 0, None, 100)?; let two = DeploymentTransaction::new(&command, "ghostlight".into(), 1, None, 100)?; let snapshot = ControlSnapshot { + expiry_requests: Vec::new(), commands: vec![Stored { envelope: command_envelope(&command, 100)?, value: command, @@ -10852,6 +11201,7 @@ mod tests { let unrelated = DeploymentTransaction::new(&unrelated_command, "huginn".into(), 0, None, 101)?; let snapshot = ControlSnapshot { + expiry_requests: Vec::new(), commands: Vec::new(), transactions: [&first, &second, &unrelated] .into_iter() @@ -11860,6 +12210,20 @@ mod tests { transaction: &DeploymentTransaction, sequence: u64, ready: bool, + ) -> Result> { + signed_correlation_as(world, transaction, sequence, ready, "active", None, b"presence") + } + + /// Odin's correlation about the transaction's incarnation, observing a + /// presence in `state` that holds `lease`, identified by `presence`. + fn signed_correlation_as( + world: &EngineFixture, + transaction: &DeploymentTransaction, + sequence: u64, + ready: bool, + state: &str, + lease: Option<&str>, + presence: &[u8], ) -> Result> { let expected = transaction.expected.as_ref().context("no Expected")?; let activation = transaction.activation.as_ref().context("no activation")?; @@ -11869,10 +12233,10 @@ mod tests { expected_projection_sha256: expected.canonical_sha256()?, expected: true, current_activation_sha256: Some(activation.canonical_sha256()?), - signed_presence_sha256: Some(sha256_id(b"presence")), - observed_presence_state: Some("active".into()), + signed_presence_sha256: Some(sha256_id(presence)), + observed_presence_state: Some(state.into()), observed_presence_publisher_sequence: Some(sequence), - observed_write_lease_sha256: None, + observed_write_lease_sha256: lease.map(str::to_owned), observed_capabilities: expected .capabilities .iter() @@ -12370,6 +12734,7 @@ mod tests { generation.odin_publisher_sequence_cursor = 88; let target = generation.target.clone(); let snapshot = ControlSnapshot { + expiry_requests: Vec::new(), commands: Vec::new(), transactions: Vec::new(), admitted: vec![Stored { @@ -12514,6 +12879,61 @@ mod tests { Ok(()) } + /// Q3 as a table: which ended phases abort, which name the unadopted + /// lease, which are left to the operator, and which recovery the terminal + /// record names. + #[test] + fn an_ended_phase_aborts_unless_a_deployment_candidate_was_issued_a_lease() -> Result<()> { + let (_, mut transaction) = fixture_transaction(FIXTURE_TRANSACTIONS[2].1)?; + let lease = write_lease(); + let sha256 = lease.canonical_sha256()?; + let granted = LeasingEvidence::Granted { + lease: lease.clone(), + lease_sha256: sha256.clone(), + }; + let prepared = LeasingEvidence::Prepared { + lease, + lease_sha256: sha256.clone(), + }; + let adoption = LeaseAdoptionEvidence::with_ready( + Some(&sha256), + &sha256_id(b"adopting-presence"), + AdoptionSource::Direct, + 1, + ); + assert!(adoption.is_some()); + assert_eq!( + LeaseAdoptionEvidence::with_ready(None, &sha256_id(b"p"), AdoptionSource::Direct, 1), + None + ); + let abort = |lease_not_adopted| PhaseEnd::Abort { lease_not_adopted }; + use CommandKind::{Continuity, Deploy}; + let cases = [ + (Deploy, None, None, abort(false)), + (Deploy, Some(LeasingEvidence::SkippedStateless), None, abort(false)), + (Deploy, Some(prepared.clone()), None, PhaseEnd::OperatorRequired), + (Deploy, Some(granted.clone()), None, PhaseEnd::OperatorRequired), + (Deploy, Some(granted.clone()), adoption.clone(), PhaseEnd::OperatorRequired), + (Continuity, Some(LeasingEvidence::SkippedStateless), None, abort(false)), + (Continuity, Some(prepared), None, abort(false)), + (Continuity, Some(granted.clone()), None, abort(true)), + (Continuity, Some(granted), adoption.clone(), abort(false)), + ]; + for (kind, leasing, lease_adoption, expected) in cases { + transaction.command_kind = kind; + transaction.leasing = leasing.clone(); + transaction.lease_adoption = lease_adoption.clone(); + assert_eq!(transaction.phase_end(), expected, "{kind:?} {leasing:?} {lease_adoption:?}"); + let recovery = if kind == Continuity && lease_adoption.is_some() { + TerminalRecovery::RestartAdmitted + } else { + TerminalRecovery::RestoreIncumbent + }; + assert_eq!(transaction.terminal_recovery(), recovery, "{kind:?} {lease_adoption:?}"); + } + Ok(()) + } + #[test] fn an_operator_required_recovery_must_say_why() -> Result<()> { let (_, mut failed) = fixture_transaction(FIXTURE_TRANSACTIONS[5].1)?; @@ -13814,14 +14234,47 @@ mod tests { world .engine .begin_pre_fencing_abort(&resident(&world)?, anyhow!("candidate refused"))?; + let attempts = || workload.0.load(std::sync::atomic::Ordering::SeqCst); + let before = now_millis()?; + world.engine.run_scheduler_tick()?; + let after = now_millis()?; + assert_eq!(attempts(), 1, "the step did not run"); + // The first failure waits one poll interval: the wait back_off wrote. + let poll = world.engine.options.poll_millis; + let written = world + .engine + .resume_backoff + .lock() + .unwrap() + .get(&resident(&world)?.value.transaction_id) + .map(|backoff| backoff.not_before_unix_millis) + .expect("the failure set no backoff"); + assert!( + (before + poll..=after + poll).contains(&written), + "back_off wrote {written}, not one poll ({poll} ms) after the failure" + ); + // The wait is set from the wall clock, and a loaded host can spend it + // inside a few ticks. So the test owns the wait: held open, no tick + // retries; once due, the next tick does. + let id = resident(&world)?.value.transaction_id; + let set_due = |due: u64| { + world + .engine + .resume_backoff + .lock() + .unwrap() + .get_mut(&id) + .expect("the failure set no backoff") + .not_before_unix_millis = due; + }; + set_due(now_millis()? + RESUME_BACKOFF_CEILING_MILLIS / 2); for _ in 0..10 { world.engine.run_scheduler_tick()?; } - assert_eq!( - workload.0.load(std::sync::atomic::Ordering::SeqCst), - 1, - "the wedged step was retried inside its backoff" - ); + assert_eq!(attempts(), 1, "the wedged step was retried inside its backoff"); + set_due(1); + world.engine.run_scheduler_tick()?; + assert_eq!(attempts(), 2, "the step was not retried once its backoff was due"); Ok(()) } @@ -15384,6 +15837,8 @@ mod tests { /// that picked up Idunn's grant reports. lease: Mutex>, reply: Mutex, + /// Runs when the stable listener accepts, before it answers. + on_stable_hit: Mutex>>, } /// How the runtime answers a well-formed challenge. @@ -15540,6 +15995,11 @@ mod tests { &stub.stable_hits }; hits.fetch_add(1, Ordering::SeqCst); + if !candidate + && let Some(hook) = stub.on_stable_hit.lock().unwrap().as_ref() + { + hook(); + } if !stub.hang_up.load(Ordering::SeqCst) { let _ = stub.serve_one(stream); } @@ -15800,6 +16260,7 @@ mod tests { hang_up: AtomicBool::new(false), lease: Mutex::new(None), reply: Mutex::new(Reply::Honest), + on_stable_hit: Mutex::new(None), }); stub.listen(candidate_listener, true); stub.listen(stable_listener, false); @@ -15807,7 +16268,17 @@ mod tests { if phase >= DeploymentPhase::Fencing && provides_odin { // Odin's own first word about the candidate, as for any // Odin-correlated target. - let warming = signed_correlation(&world, &transaction, 4, false)?; + // A stateful candidate is warming, not active, until it holds + // the lease Idunn grants after the fence. + let warming = signed_correlation_as( + &world, + &transaction, + 4, + false, + if stateful { "warming" } else { "active" }, + None, + b"presence", + )?; let authenticated = world.engine.authenticate_topology_bytes( &ControlSnapshot::read(&world.state_store)?, &transaction, @@ -15872,6 +16343,33 @@ mod tests { // valid correlation that says the target is not Ready. let store = &world.engine.options.odin_correlation_store; let poisoned = match odin { + _ if provides_odin && stateful => { + // A new warming word after the fence: what the lease + // grant waits for. + SingleFileMessagePackBackingStore::new(store).insert_entry_if_absent( + CultCacheEnvelope { + key: crate::drivers::incarnation_key_of( + &transaction.target, + &transaction.expected.as_ref().unwrap().canonical_sha256()?, + ), + r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), + payload: signed_correlation_as( + &world, + &transaction, + 5, + false, + "warming", + None, + b"presence-5", + )?, + stored_at: rfc3339_millis(now_millis()?)?, + schema_id: Some( + cultnet_rs::ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into(), + ), + }, + )?; + Vec::new() + } _ if provides_odin => { odin_reports_ready(&world, &transaction, 5)?; Vec::new() @@ -17787,7 +18285,7 @@ mod tests { #[cfg(unix)] #[test] - fn a_failing_reload_that_deletes_the_fragment_backs_the_route_off() -> Result<()> { + fn a_failing_reload_that_puts_the_drift_back_backs_the_route_off() -> Result<()> { let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; routed.admit()?; routed.world.route_stubs.refuse_reloads()?; @@ -17797,7 +18295,7 @@ mod tests { routed.tick()?; } assert_eq!(routed.reloads() - reloads, 1, "the route was reloaded again and again"); - assert!(!routed.fragment_path().exists()); + assert_eq!(std::fs::read(routed.fragment_path())?, b"drifted bytes\n"); assert!(admitted(&routed)?.route_supervision.unwrap().degraded_since_unix_millis.is_some()); Ok(()) } @@ -17858,5 +18356,604 @@ mod tests { assert_eq!(after.route_actuations.len(), 2, "preflight and install are the Forward"); Ok(()) } + + // ----------------------------------------------------------------- + // B5: every post-fencing phase ends. + // ----------------------------------------------------------------- + + impl RoutedWorld { + /// One scheduler pass over the transaction, as the daemon runs it: + /// through the deadline resolver, with the resume backoff spent. + fn resume(&self) -> Result<()> { + self.world.engine.resume_backoff.lock().unwrap().clear(); + self.world.engine.resume_one_transaction()?; + Ok(()) + } + + fn resume_until_terminal(&self) -> Result { + for _ in 0..12 { + if latest(&self.world)?.completion.is_some() { + break; + } + self.resume()?; + } + let finished = latest(&self.world)?; + ensure!(finished.completion.is_some(), "never finished: {:?}", finished.last_error); + Ok(finished) + } + } + + /// The live transaction's phase deadline is already behind it. + fn pass_deadline(world: &EngineFixture) -> Result<()> { + let current = resident(world)?; + let mut next = current.value.clone(); + let deadline = next.phase_deadline.as_mut().context("no deadline to pass")?; + deadline.entered_at_unix_millis = 1; + deadline.deadline_at_unix_millis = 2; + replace_transaction(&world.state_store, ¤t, &next) + } + + /// Soul's F1 probe, committed. A continuity's route changes are + /// Survival, so no ceiling refuses its install and rollback while its + /// route proof keeps failing. The Routing deadline ends it: no attempt + /// runs after the deadline. + #[cfg(unix)] + #[test] + fn a_continuity_whose_route_proof_fails_ends_at_its_routing_deadline() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.stub.set_state("active"); + routed.run_to_routing()?; + let deadline = routed.transaction()?.phase_deadline.context("Routing has no deadline")?; + routed.stub.hang_up.store(true, Ordering::SeqCst); + for _ in 0..3 { + routed.resume()?; + } + // Before the deadline each attempt installs and rolls back, and no + // resume moves the deadline. + assert_eq!(routed.reloads(), 6); + let waiting = routed.transaction()?; + assert_eq!(waiting.phase, DeploymentPhase::Routing); + assert!(waiting.post_fencing_abort.is_none(), "aborted before its deadline"); + assert!(waiting.last_error.is_some()); + assert_eq!(waiting.phase_deadline, Some(deadline), "a resume moved the deadline"); + + pass_deadline(&routed.world)?; + routed.resume()?; + let abort = routed + .transaction()? + .post_fencing_abort + .context("the passed deadline did not end the phase")?; + assert!(abort.error.contains("Routing phase passed its deadline"), "{}", abort.error); + assert!(!abort.error.contains("lease-not-adopted"), "{}", abort.error); + let finished = routed.resume_until_terminal()?; + assert!( + matches!( + finished.completion, + Some(TransactionCompletion::FailedAfterFencing { + recovery: TerminalRecovery::RestoreIncumbent, + .. + }) + ), + "{:?}", + finished.completion + ); + // After the deadline no attempt runs; the withdrawal writes and + // reloads once, whatever the rolled-back disk shows. + assert_eq!(routed.reloads(), 7, "an attempt ran after the deadline"); + Ok(()) + } + + /// The live shape of F1. A continuity reuses the admitted Expected, so + /// its membership is the admitted fragment already on disk: while its + /// proof fails, the install and the rollback write nothing and reload + /// nothing, and neither does the withdrawal after its deadline. + #[cfg(unix)] + #[test] + fn a_continuity_over_its_admitted_route_reloads_nothing_while_its_proof_fails() + -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + let current = resident(&routed.world)?; + let mut next = current.value.clone(); + let binding = next.plan.as_ref().unwrap().parsed_inputs()?.1; + let admitted = NginxRouteDriver::new(binding.route.context("no route binding")?) + .render(next.expected.as_ref().context("no Expected")?)?; + let preflight = next.route_preflight.as_mut().context("no preflight")?; + preflight.incumbent_runtime_instance_id = Some(sha256_id(b"admitted-instance")); + preflight.incumbent_membership_sha256 = Some(sha256_id(&admitted)); + preflight.incumbent_configuration = Some(admitted.clone()); + replace_transaction(&routed.world.state_store, ¤t, &next)?; + std::fs::write(routed.fragment_path(), &admitted)?; + + routed.stub.set_state("active"); + routed.run_to_routing()?; + routed.stub.hang_up.store(true, Ordering::SeqCst); + for _ in 0..5 { + routed.resume()?; + } + assert_eq!(routed.transaction()?.phase, DeploymentPhase::Routing); + assert!(routed.transaction()?.last_error.is_some(), "the proof did not fail"); + assert_eq!(routed.world.route_stubs.count(""), 0, "a failing continuity ran a host program"); + + pass_deadline(&routed.world)?; + routed.resume_until_terminal()?; + assert_eq!(routed.world.route_stubs.count(""), 0, "its withdrawal ran a host program"); + assert_eq!(std::fs::read(routed.fragment_path())?, admitted, "the admitted route moved"); + Ok(()) + } + + /// The abort's withdrawal wrote the baseline back and Idunn died before + /// the reload: the disk already shows the baseline, and the retry still + /// reloads, so nginx does not keep proxying to the stopped candidate. + #[cfg(unix)] + #[test] + fn an_abort_whose_reload_a_crash_cut_off_reloads_on_retry() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.stub.set_state("active"); + routed.run_to_routing()?; + routed.promote()?; + let installed = routed.reloads(); + routed + .world + .engine + .begin_post_fencing_abort(&resident(&routed.world)?, anyhow!("test"))?; + std::fs::remove_file(routed.fragment_path())?; + routed.resume()?; + assert_eq!( + routed.transaction()?.post_fencing_abort.context("no abort")?.route_restoration, + CleanupEvidence::Complete + ); + assert_eq!(routed.reloads(), installed + 1, "the withdrawal trusted unloaded bytes"); + Ok(()) + } + + /// Soul's F2 probe, committed: a post-fence withdrawal whose + /// `ufw delete` keeps failing withdraws the firewall allow first, so its + /// retries reload nginx not at all. + #[cfg(unix)] + #[test] + fn a_withdrawal_whose_firewall_step_keeps_failing_never_reloads() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + routed.stub.set_state("active"); + routed.run_to_routing()?; + routed.promote()?; + let installed = routed.reloads(); + routed.world.route_stubs.refuse_ufw_deletes()?; + routed + .world + .engine + .begin_post_fencing_abort(&resident(&routed.world)?, anyhow!("test"))?; + for _ in 0..20 { + routed.resume()?; + } + assert_eq!(routed.reloads(), installed, "a failing withdrawal reloaded nginx"); + assert_eq!(routed.world.route_stubs.count("ufw delete"), 20); + assert!(routed.fragment_path().exists(), "the route went before its firewall rule"); + Ok(()) + } + + /// A record lifted from before deadlines existed is timed once, from + /// the first time the resolver sees it, and never again. + #[test] + fn a_post_fence_record_without_a_deadline_is_timed_once() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + let current = resident(&routed.world)?; + let mut lifted = current.value.clone(); + lifted.phase_deadline = None; + replace_transaction(&routed.world.state_store, ¤t, &lifted)?; + let engine = &routed.world.engine; + + let before = now_millis()?; + assert_eq!(engine.resolve_phase_deadline(&resident(&routed.world)?)?, Some(true)); + let stamped = routed.transaction()?.phase_deadline.context("the record was not timed")?; + assert_eq!(stamped.phase, DeploymentPhase::Fencing); + assert!(stamped.entered_at_unix_millis >= before); + assert_eq!( + stamped.deadline_at_unix_millis - stamped.entered_at_unix_millis, + u64::from(lifted.plan.as_ref().unwrap().phase_deadlines().fencing_seconds) * 1000 + ); + assert_eq!(engine.resolve_phase_deadline(&resident(&routed.world)?)?, None); + assert_eq!(routed.transaction()?.phase_deadline, Some(stamped)); + Ok(()) + } + + /// `idunn expire` writes a request and nothing else; the resolver ends + /// the phase from it, and the spent request is deleted. + #[test] + fn an_operator_expiry_ends_the_phase_through_the_resolver() -> Result<()> { + let early = routed_world(Odin::Unreachable, 1, DeploymentPhase::Warming, false)?; + let refused = expire(&early.world.state_store, "continuity-service", "operator") + .expect_err("a transaction before the fence was expired"); + assert!(format!("{refused:#}").contains("idunn cancel"), "{refused:#}"); + + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + let untouched = resident(&routed.world)?.envelope; + expire(&routed.world.state_store, "continuity-service", "operator")?; + assert!(expire(&routed.world.state_store, "continuity-service", "operator").is_err()); + let snapshot = ControlSnapshot::read(&routed.world.state_store)?; + assert_eq!(snapshot.expiry_requests.len(), 1); + assert_eq!(resident(&routed.world)?.envelope, untouched, "the CLI wrote the transaction"); + + routed.resume()?; + let abort = routed + .transaction()? + .post_fencing_abort + .context("the expiry did not end the phase")?; + assert!(abort.error.contains("Fencing phase was expired by the operator"), "{}", abort.error); + routed.resume()?; + assert!( + ControlSnapshot::read(&routed.world.state_store)?.expiry_requests.is_empty(), + "the spent request stayed" + ); + Ok(()) + } + + /// The `expire` verb as the CLI parses and runs it: the options land + /// where they say, and it asks only about its own command's live + /// transaction. + #[test] + fn idunn_expire_parses_its_options_and_asks_only_about_its_own_live_transaction() + -> Result<()> { + let args = |words: &[&str]| words.iter().map(|word| (*word).to_owned()).collect::>(); + assert_eq!( + parse(args(&["expire", "up-service", "--requested-by", "alice", "--state-store", "/s/c.cc"]).into_iter())?, + Command::Expire { + state_store: PathBuf::from("/s/c.cc"), + command_id: "up-service".into(), + requested_by: "alice".into(), + } + ); + let help = parse(args(&["expire", "up-service", "--help"]).into_iter()).unwrap_err(); + assert!(format!("{help}").contains("idunn expire "), "{help}"); + assert!(parse(args(&["expire", "up-service", "--force"]).into_iter()).is_err()); + assert!(parse(args(&["expire"]).into_iter()).is_err()); + + // The verb runs: a store with nothing live refuses. + let empty = TempDir::new()?; + let store = empty.path().join("control.cc").display().to_string(); + let refused = run(args(&["expire", "up-service", "--state-store", &store]).into_iter()) + .expect_err("expired a command with no transaction"); + assert!(format!("{refused:#}").contains("no live transaction"), "{refused:#}"); + + // Another target's deployment in flight is not this command's. + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + let now = now_millis()?; + let other = DeploymentCommand { + schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(), + command_id: "up-other".into(), + kind: CommandKind::Deploy, + selector: "other".into(), + requested_by: "test".into(), + requested_at_unix_millis: now, + }; + let sealing = DeploymentTransaction::new(&other, "other".into(), 0, None, now)?; + assert!(SingleFileMessagePackBackingStore::new(&routed.world.state_store).compare_exchange( + &[ + CultCacheExpectedEnvelope { + r#type: DeploymentCommand::TYPE.into(), + key: other.command_id.clone(), + current: None, + }, + CultCacheExpectedEnvelope { + r#type: DeploymentTransaction::TYPE.into(), + key: sealing.transaction_id.clone(), + current: None, + }, + ], + &[command_envelope(&other, now)?, transaction_envelope(&sealing, now)?], + )?); + expire(&routed.world.state_store, "continuity-service", "operator")?; + let requests = ControlSnapshot::read(&routed.world.state_store)?.expiry_requests; + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].value.command_id, "continuity-service"); + Ok(()) + } + + /// The resolver reads a request as its own transaction's only: with two + /// live transactions, a request for the other leaves this one running. + #[test] + fn an_expiry_request_ends_only_its_own_commands_phase() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?; + let now = now_millis()?; + let other = DeploymentCommand { + schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(), + command_id: "up-other".into(), + kind: CommandKind::Deploy, + selector: "other".into(), + requested_by: "test".into(), + requested_at_unix_millis: now, + }; + let sealing = DeploymentTransaction::new(&other, "other".into(), 0, None, now)?; + // The request names a command whose id has this transaction's + // length and first bytes, and differs only in its last. + let near = "continuity-servicx"; + assert_eq!(near.len(), "continuity-service".len()); + let request = ExpiryRequest { + schema_version: EXPIRY_REQUEST_SCHEMA.into(), + command_id: near.into(), + requested_by: "operator".into(), + requested_at_unix_millis: now, + }; + let store = SingleFileMessagePackBackingStore::new(&routed.world.state_store); + assert!(store.compare_exchange( + &[ + CultCacheExpectedEnvelope { + r#type: DeploymentCommand::TYPE.into(), + key: other.command_id.clone(), + current: None, + }, + CultCacheExpectedEnvelope { + r#type: DeploymentTransaction::TYPE.into(), + key: sealing.transaction_id.clone(), + current: None, + }, + CultCacheExpectedEnvelope { + r#type: ExpiryRequest::TYPE.into(), + key: request.command_id.clone(), + current: None, + }, + ], + &[ + command_envelope(&other, now)?, + transaction_envelope(&sealing, now)?, + expiry_request_envelope(&request, now)?, + ], + )?); + let snapshot = ControlSnapshot::read(&routed.world.state_store)?; + assert_eq!(snapshot.transactions.iter().filter(|t| !t.value.is_terminal()).count(), 2); + let mine = snapshot + .transactions + .iter() + .find(|t| t.value.command_id == "continuity-service") + .context("no continuity")?; + assert_eq!(routed.world.engine.resolve_phase_deadline(mine)?, None); + let after = ControlSnapshot::read(&routed.world.state_store)?; + let unchanged = after + .transactions + .iter() + .find(|t| t.value.command_id == "continuity-service") + .context("no continuity")?; + assert_eq!(unchanged.envelope, mine.envelope, "another command's request ended this phase"); + Ok(()) + } + + /// Q3 at the deadline, through the Engine. A continuity whose candidate + /// was granted a lease and never proved it held it fails as + /// `lease-not-adopted`; one whose Ready proved it restarts the admitted + /// release. A deployment whose candidate holds a lease is left to the + /// operator: nothing is aborted and its step still runs. + #[test] + fn a_stateful_deadline_resolves_by_lease_adoption() -> Result<()> { + let granted = |routed: &RoutedWorld| -> Result { + routed.drive_until(|t| t.phase == DeploymentPhase::AwaitingReady)?; + Ok(routed + .transaction()? + .leasing + .as_ref() + .and_then(LeasingEvidence::lease_sha256) + .context("no granted lease")? + .to_owned()) + }; + + let unadopted = stateful_routed_world(Odin::Unreachable, DeploymentPhase::Fencing)?; + granted(&unadopted)?; + pass_deadline(&unadopted.world)?; + unadopted.resume()?; + let abort = unadopted.transaction()?.post_fencing_abort.context("not ended")?; + assert!(abort.error.starts_with("lease-not-adopted: "), "{}", abort.error); + assert!(abort.error.contains("AwaitingReady phase passed its deadline"), "{}", abort.error); + let finished = unadopted.resume_until_terminal()?; + assert!(matches!( + finished.completion, + Some(TransactionCompletion::FailedAfterFencing { + recovery: TerminalRecovery::RestoreIncumbent, + .. + }) + )); + + let adopted = stateful_routed_world(Odin::Unreachable, DeploymentPhase::Fencing)?; + let lease = granted(&adopted)?; + adopted.stub.set_state("active"); + *adopted.stub.lease.lock().unwrap() = Some(lease.clone()); + adopted.step()?; + let ready = adopted.transaction()?; + let Some(ReadinessEvidence::RouteProof { evidence }) = &ready.ready else { + bail!("the candidate was not Ready"); + }; + let adoption = ready.lease_adoption.clone().context("Ready recorded no adoption")?; + assert_eq!(adoption.write_lease_sha256, lease); + assert_eq!(adoption.signed_presence_sha256, evidence.canonical_sha256); + assert_eq!(adoption.source, AdoptionSource::Direct); + adopted.drive_until(|t| t.phase == DeploymentPhase::Routing)?; + pass_deadline(&adopted.world)?; + adopted.resume()?; + let abort = adopted.transaction()?.post_fencing_abort.context("not ended")?; + assert!(!abort.error.contains("lease-not-adopted"), "{}", abort.error); + let finished = adopted.resume_until_terminal()?; + assert!( + matches!( + finished.completion, + Some(TransactionCompletion::FailedAfterFencing { + recovery: TerminalRecovery::RestartAdmitted, + .. + }) + ), + "{:?}", + finished.completion + ); + + let deploy = + build_routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false, true, CommandKind::Deploy)?; + granted(&deploy)?; + pass_deadline(&deploy.world)?; + for _ in 0..3 { + deploy.resume()?; + } + let held = deploy.transaction()?; + assert!(held.post_fencing_abort.is_none(), "a leased deployment was aborted"); + assert_eq!(held.phase, DeploymentPhase::AwaitingReady); + assert_eq!(held.last_error.as_deref(), Some("candidate is still warming"), "its step did not run"); + Ok(()) + } + + /// B5 Soul A. The physical grant lands before `Granted` is durable. A + /// continuity that dies in that window and passes its deadline is + /// aborted from `Prepared`; the abort still withdraws the lease on + /// disk, so the next fencing of the target is not refused. + #[test] + fn an_abort_from_a_prepared_lease_withdraws_a_grant_already_on_disk() -> Result<()> { + let routed = stateful_routed_world(Odin::Unreachable, DeploymentPhase::Fencing)?; + routed.drive_until(|t| matches!(t.leasing, Some(LeasingEvidence::Granted { .. })))?; + let current = resident(&routed.world)?; + let mut crashed = current.value.clone(); + let Some(LeasingEvidence::Granted { lease, lease_sha256 }) = crashed.leasing.clone() else { + bail!("no granted lease"); + }; + crashed.leasing = Some(LeasingEvidence::Prepared { lease: lease.clone(), lease_sha256 }); + replace_transaction(&routed.world.state_store, ¤t, &crashed)?; + let driver = CultCacheWriteLeaseDriver::new( + "service", + routed.world.root.join("lease/process-write-lease.cc"), + ); + assert!(driver.observe_exact(&lease)?, "the grant is not on disk"); + + pass_deadline(&routed.world)?; + routed.resume()?; + assert!(routed.transaction()?.post_fencing_abort.is_some(), "the deadline did not end it"); + routed.resume_until_terminal()?; + assert!(driver.observe_empty()?, "the abort left the granted lease on disk"); + // What the next fencing does to a candidate's lease path. + driver.revoke_exact(None)?; + Ok(()) + } + + /// B5 Soul F. A stateful candidate Odin reports on: its Ready word + /// carries the granted lease, and that Ready records the adoption, + /// sourced from Odin's topology and naming the presence Odin observed. + #[test] + fn an_odin_ready_word_records_a_stateful_candidates_lease_adoption() -> Result<()> { + let routed = build_routed_world( + Odin::Unreachable, + 1, + DeploymentPhase::Fencing, + true, + true, + CommandKind::Continuity, + )?; + for _ in 0..8 { + if routed.transaction()?.phase == DeploymentPhase::AwaitingReady { + break; + } + routed.step()?; + } + let waiting = routed.transaction()?; + assert_eq!(waiting.phase, DeploymentPhase::AwaitingReady, "{:?}", waiting.last_error); + let lease = waiting + .leasing + .as_ref() + .and_then(LeasingEvidence::lease_sha256) + .context("no granted lease")? + .to_owned(); + assert_eq!(waiting.lease_adoption, None); + + let ready = CultCacheEnvelope { + key: crate::drivers::incarnation_key_of( + &waiting.target, + &waiting.expected.as_ref().unwrap().canonical_sha256()?, + ), + r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), + payload: signed_correlation_as( + &routed.world, + &waiting, + waiting.odin_publisher_sequence_cursor + 1, + true, + "active", + Some(&lease), + b"presence-ready", + )?, + stored_at: rfc3339_millis(now_millis()?)?, + schema_id: Some(cultnet_rs::ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into()), + }; + replace_odin_entry(&routed.world.engine.options.odin_correlation_store, ready)?; + // One step admits the new sequence, the next records Ready from it. + for _ in 0..3 { + if routed.transaction()?.ready.is_some() { + break; + } + routed.step()?; + } + let adopted = routed.transaction()?; + assert!( + matches!(adopted.ready, Some(ReadinessEvidence::OdinCorrelated { .. })), + "{:?}", + adopted.last_error + ); + let adoption = adopted.lease_adoption.context("Odin's Ready recorded no adoption")?; + assert_eq!(adoption.source, AdoptionSource::OdinTopology); + assert_eq!(adoption.write_lease_sha256, lease); + assert_eq!(adoption.signed_presence_sha256, sha256_id(b"presence-ready")); + Ok(()) + } + + /// F16: Commit goes on from the topology it admitted, so Odin + /// re-stamping before the step and again during the final proof does + /// not send it round again. + #[test] + fn commit_completes_while_odin_restamps_every_tick() -> Result<()> { + let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, true)?; + routed.stub.set_state("active"); + routed.run_to_routing()?; + routed.promote()?; + let transaction = routed.transaction()?; + let cursor = transaction.odin_publisher_sequence_cursor; + let store = routed.world.engine.options.odin_correlation_store.clone(); + replace_odin_entry(&store, odin_ready_envelope(&routed.world, &transaction, cursor + 1)?)?; + let later = odin_ready_envelope(&routed.world, &transaction, cursor + 2)?; + *routed.stub.on_stable_hit.lock().unwrap() = Some(Box::new(move || { + replace_odin_entry(&store, later.clone()).expect("Odin re-stamps"); + })); + routed.step()?; + let generation = admitted(&routed)?; + assert_eq!( + generation.latest_odin_observation.context("no Odin receipt")?.publisher_sequence, + cursor + 2, + "commit did not go on from the re-stamped topology" + ); + Ok(()) + } + + fn odin_ready_envelope( + world: &EngineFixture, + transaction: &DeploymentTransaction, + sequence: u64, + ) -> Result { + let expected = transaction.expected.as_ref().context("no Expected")?; + Ok(CultCacheEnvelope { + key: crate::drivers::incarnation_key_of(&expected.target, &expected.canonical_sha256()?), + r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), + payload: signed_correlation(world, transaction, sequence, true)?, + stored_at: rfc3339_millis(now_millis()?)?, + schema_id: Some(cultnet_rs::ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into()), + }) + } + + fn replace_odin_entry(store: &Path, envelope: CultCacheEnvelope) -> Result<()> { + let backing = SingleFileMessagePackBackingStore::new(store); + let current = backing + .pull_all_read_only_snapshot()? + .into_iter() + .find(|entry| entry.key == envelope.key && entry.r#type == envelope.r#type); + ensure!( + backing.compare_exchange( + &[CultCacheExpectedEnvelope { + r#type: envelope.r#type.clone(), + key: envelope.key.clone(), + current, + }], + &[envelope], + )?, + "the Odin store changed under the re-stamp" + ); + Ok(()) + } } } diff --git a/src/deployment_plan.rs b/src/deployment_plan.rs index 5210448..603c9d2 100644 --- a/src/deployment_plan.rs +++ b/src/deployment_plan.rs @@ -467,15 +467,17 @@ pub fn select_dependencies( /// overrides any of them; the resolved values are frozen into the plan. pub const DEFAULT_FENCING_DEADLINE_SECONDS: u32 = 120; pub const DEFAULT_LEASING_DEADLINE_SECONDS: u32 = 120; -// Unsafe for streampixels-service (startup = "create-or-open-after-write-lease"): -// its Postgres migrations run inside AwaitingReady. B5 requires an explicit -// binding value for stateful targets instead of this default. +// A stateful candidate may run migrations inside AwaitingReady (streampixels- +// service runs Postgres's there). When this passes, a deployment whose +// candidate was issued a write lease is left to the operator, not aborted; a +// continuity is aborted and its release restarted. pub const DEFAULT_AWAITING_READY_DEADLINE_SECONDS: u32 = 300; pub const DEFAULT_ROUTING_DEADLINE_SECONDS: u32 = 120; pub const DEFAULT_COMMITTING_DEADLINE_SECONDS: u32 = 120; -/// The resolved per-phase deadline durations a plan carries. Nothing reads -/// them at the cut that introduced them. +/// The resolved per-phase deadline durations a plan carries. Entering a +/// post-fencing phase stamps its deadline from them, and the deadline +/// resolver ends the phase once it passes. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct PhaseDeadlines { diff --git a/src/drivers.rs b/src/drivers.rs index 189779b..b8375e1 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -707,6 +707,14 @@ const ROUTE_HTTP_CHUNK_LINE_BYTES: usize = 128; const ROUTE_HTTP_SNAPSHOT_PATH: &str = "/cultnet/snapshot"; impl RoutePreflightReceipt { + /// Whether the candidate's membership differs from the baseline the + /// preflight found. A continuity over its admitted route does not: its + /// transaction has no route change to make or to undo. + pub fn changes_route(&self) -> bool { + self.incumbent_membership_sha256.as_deref() + != Some(self.candidate_membership_sha256.as_str()) + } + pub fn validate(&self) -> Result<()> { require_driver_id(&self.route_id, "route preflight")?; require_sha256_id( @@ -5332,31 +5340,31 @@ impl NginxRouteDriver { } } + /// Put `prior` back as the route's fragment, and always write and reload: + /// on a transaction path, bytes on disk that already equal `prior` may be + /// a write whose reload a crash cut off. A reload that fails puts the + /// previous fragment back, so the disk shows what nginx still serves. + /// + /// With no prior membership the stable endpoint routes to nothing, and + /// its firewall allow is withdrawn first: a failing `ufw delete` then + /// retries without touching nginx, and a crash after it is safe, because + /// a rule already gone reads as withdrawn. fn restore(&self, prior: Option<&[u8]>, gate: &dyn RouteActuationGate) -> Result<()> { gate.admit(RouteActuation::Survival)?; - self.write_fragment(prior)?; - self.reload()?; - // No prior membership means the stable endpoint no longer routes to - // anything; its firewall allow goes with the fragment. if prior.is_none() { self.withdraw_endpoint()?; } - Ok(()) - } - - fn fail_after_rollback( - &self, - prior: Option<&[u8]>, - failure: anyhow::Error, - context: &str, - gate: &dyn RouteActuationGate, - ) -> Result { - match self.restore(prior, gate) { - Ok(()) => Err(failure).context(context.to_owned()), - Err(rollback) => Err(failure).context(format!( - "{context}; route rollback also failed: {rollback:#}" - )), + let current = self.current_configuration()?; + self.write_fragment(prior)?; + if let Err(reload) = self.reload() { + return match self.write_fragment(current.as_deref()) { + Ok(()) => Err(reload).context("reloading the restored route"), + Err(undo) => Err(reload).context(format!( + "reloading the restored route; putting the previous fragment back also failed: {undo:#}" + )), + }; } + Ok(()) } pub fn preflight( @@ -5431,22 +5439,42 @@ impl NginxRouteDriver { && prior_sha256 == preflight.incumbent_membership_sha256), "route baseline changed after preflight" ); - gate.admit(RouteActuation::Forward)?; - if !candidate_already_written { - atomic_replace(&self.binding.config_path, &rendered)?; + // A candidate whose membership is the baseline's (a continuity over + // its admitted route) changes nothing, so nothing is written, and no + // crash can have cut a write of this transaction short. + if !preflight.changes_route() { + return Ok(sha256_id(&rendered)); } + // Otherwise bytes already on disk prove nothing: a crash may have cut + // off the reload after an earlier attempt's write. So it always + // writes and reloads. + gate.admit(RouteActuation::Forward)?; + atomic_replace(&self.binding.config_path, &rendered)?; if let Err(error) = self.admit_endpoint().and_then(|()| self.reload()) { - if rollback_allowed { - return self.fail_after_rollback( - preflight.incumbent_configuration.as_deref(), - error, - "candidate route validation or reload failed", - gate, - ); - } - return Err(error).context( - "candidate route reload failed after incumbent route authority was fenced; candidate fragment retained for retry", - ); + // nginx did not load the candidate, so it still serves the + // baseline. The disk is put back to match it, which is the whole + // rollback, and a retry installs and reloads again rather than + // mistaking unloaded bytes for a live route. Past the fence the + // firewall allow stays for that retry; before it, an allow the + // baseline never had goes too. + let context = if rollback_allowed { + "candidate route validation or reload failed" + } else { + "candidate route reload failed after incumbent route authority was fenced; the fragment was put back for retry" + }; + let undo = self.write_fragment(prior.as_deref()).and_then(|()| { + if rollback_allowed && prior.is_none() { + self.withdraw_endpoint() + } else { + Ok(()) + } + }); + return match undo { + Ok(()) => Err(error).context(context), + Err(undo) => Err(error).context(format!( + "{context}; putting the route back also failed: {undo:#}" + )), + }; } ensure!( self.current_configuration()?.as_deref() == Some(rendered.as_slice()), @@ -5468,12 +5496,6 @@ impl NginxRouteDriver { Ok(self.current_configuration()?.as_deref() == Some(rendered.as_slice())) } - /// Restore the exact membership owned by an admitted generation. Any - /// different fragment is drift, not a rollback baseline, so it is never - /// restored after a failed reload. A failed reload removes the newly - /// written fragment so the next continuity pass cannot mistake disk bytes - /// for an adopted route. The caller must still challenge the stable - /// listener to prove that nginx workers adopted this membership. /// Put the route back exactly as the candidate found it. /// /// The preflight receipt captured the incumbent's configuration before this @@ -5490,11 +5512,21 @@ impl NginxRouteDriver { preflight.route_id == self.binding.route_id, "route preflight receipt describes another route" ); + if !preflight.changes_route() { + return Ok(()); + } self.restore(preflight.incumbent_configuration.as_deref(), gate) } + /// Restore the exact membership owned by an admitted generation. A failed + /// reload puts back the bytes it replaced, so the disk still shows what + /// nginx serves and the next pass sees the drift again rather than taking + /// unloaded bytes for an adopted route. The caller must still challenge + /// the stable listener to prove that nginx workers adopted this membership. + /// /// A fragment already equal to the admitted membership is left alone: no - /// write, no `nginx -t`, no firewall call, no reload. + /// write, no `nginx -t`, no firewall call, no reload. Supervision is the + /// only path that skips on equal bytes. pub fn restore_admitted_membership( &self, expected: &IdunnExpectedIncarnationRecord, @@ -5511,12 +5543,13 @@ impl NginxRouteDriver { } gate.admit(RouteActuation::Survival)?; self.validate_candidate_in_private_mount(&rendered)?; + let replaced = self.current_configuration()?; atomic_replace(&self.binding.config_path, &rendered)?; if let Err(reload) = self.admit_endpoint().and_then(|()| self.reload()) { - return match self.write_fragment(None) { + return match self.write_fragment(replaced.as_deref()) { Ok(()) => Err(reload).context("reloading the exact admitted route membership"), - Err(cleanup) => Err(reload).context(format!( - "reloading the exact admitted route membership; removing its unproved fragment also failed: {cleanup:#}" + Err(undo) => Err(reload).context(format!( + "reloading the exact admitted route membership; putting the replaced fragment back also failed: {undo:#}" )), }; } @@ -5651,6 +5684,9 @@ impl NginxRouteDriver { self.current_configuration()?.as_deref() == Some(rendered.as_slice()), "candidate route changed before rollback" ); + if !preflight.changes_route() { + return Ok(()); + } self.restore(preflight.incumbent_configuration.as_deref(), gate) } } @@ -11817,6 +11853,226 @@ Content-Le".to_vec()), |_| {}), Ok(()) } + /// Stub host programs that log every call. `systemctl reload` fails while + /// `/systemctl.fail` exists, `ufw delete` while `/ufw.fail` does. + #[cfg(unix)] + fn failable_route_driver( + temp: &Path, + ) -> Result<(NginxRouteDriver, IdunnExpectedIncarnationRecord, PathBuf)> { + use std::os::unix::fs::PermissionsExt; + + let (mut driver, candidate, calls) = stubbed_route_driver(temp)?; + let failable = |name: &str, verb: &str| -> Result { + let path = temp.join(name); + fs::write( + &path, + format!( + "#!/bin/sh\necho \"{name} $*\" >> '{}'\nif [ \"$1\" = {verb} ] && [ -e \"$0.fail\" ]; then exit 1; fi\nexit 0\n", + calls.display() + ), + )?; + fs::set_permissions(&path, fs::Permissions::from_mode(0o755))?; + Ok(path) + }; + driver.systemctl_program = failable("systemctl", "reload")?; + driver.ufw_program = failable("ufw", "delete")?; + // `/ufw.absent`: the rule is already gone, as ufw reports it. + let ufw = fs::read_to_string(&driver.ufw_program)?.replace( + "exit 0\n", + "if [ \"$1\" = delete ] && [ -e \"$0.absent\" ]; then echo 'Could not delete non-existent rule' >&2; exit 1; fi\nexit 0\n", + ); + fs::write(&driver.ufw_program, ufw)?; + Ok((driver, candidate, calls)) + } + + #[cfg(unix)] + fn logged(calls: &Path, prefix: &str) -> usize { + fs::read_to_string(calls) + .unwrap_or_default() + .lines() + .filter(|line| line.starts_with(prefix)) + .count() + } + + /// Admits everything and records what it was asked. + struct Recording(std::cell::RefCell>); + + impl RouteActuationGate for Recording { + fn admit(&self, kind: RouteActuation) -> Result<()> { + self.0.borrow_mut().push(kind); + Ok(()) + } + } + + /// On a transaction path, bytes already on disk prove nothing: a crash may + /// have cut off the reload after the write. The install past the fence and + /// the withdrawal both write and reload whatever the disk shows, and the + /// install admits its firewall allow again. + #[cfg(unix)] + #[test] + fn a_crash_between_a_write_and_its_reload_is_reloaded_on_retry() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let receipt = preflight_receipt(&driver, &candidate)?; + let rendered = driver.render(&candidate)?; + + // The install wrote its fragment and Idunn died before the reload. + fs::write(&driver.binding.config_path, &rendered)?; + driver.install(&candidate, &digest('a'), &receipt, false, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), 1, "the unloaded install was not reloaded"); + assert_eq!(logged(&calls, "ufw allow"), 1, "the first deploy's allow was skipped"); + + // The withdrawal removed it and Idunn died before the reload. + fs::remove_file(&driver.binding.config_path)?; + driver.withdraw_candidate_membership(&receipt, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), 2, "the unloaded withdrawal was not reloaded"); + assert_eq!(logged(&calls, "ufw delete"), 1); + Ok(()) + } + + /// A redeploy over an existing route changes it: the candidate differs + /// from the incumbent's fragment, so it installs and reloads, and its + /// rollback and withdrawal put the incumbent back and reload. + #[cfg(unix)] + #[test] + fn a_deploy_over_an_incumbent_route_installs_and_reloads() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let incumbent = b"# incumbent membership\n".to_vec(); + let mut receipt = preflight_receipt(&driver, &candidate)?; + receipt.incumbent_runtime_instance_id = Some(digest('b')); + receipt.incumbent_membership_sha256 = Some(sha256_id(&incumbent)); + receipt.incumbent_configuration = Some(incumbent.clone()); + assert!(receipt.changes_route()); + fs::write(&driver.binding.config_path, &incumbent)?; + + driver.install(&candidate, &digest('a'), &receipt, true, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), 1, "the redeploy did not install"); + assert_eq!(fs::read(&driver.binding.config_path)?, driver.render(&candidate)?); + driver.rollback(&candidate, &digest('a'), &receipt, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), 2, "the rollback did not reload"); + assert_eq!(fs::read(&driver.binding.config_path)?, incumbent); + driver.install(&candidate, &digest('a'), &receipt, true, &Unmetered)?; + driver.withdraw_candidate_membership(&receipt, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), 4, "the withdrawal did not reload"); + assert_eq!(fs::read(&driver.binding.config_path)?, incumbent); + Ok(()) + } + + /// With nothing left to route, the withdrawal takes the firewall allow + /// first. A `ufw delete` that keeps failing retries without reloading + /// nginx; once it goes through, the fragment is removed and reloaded. + #[cfg(unix)] + #[test] + fn a_failing_firewall_withdrawal_retries_without_reloading() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let receipt = preflight_receipt(&driver, &candidate)?; + driver.install(&candidate, &digest('a'), &receipt, false, &Unmetered)?; + let installed = logged(&calls, "systemctl reload"); + + fs::write(temp.path().join("ufw.fail"), b"x")?; + for _ in 0..5 { + assert!(driver.withdraw_candidate_membership(&receipt, &Unmetered).is_err()); + } + assert_eq!(logged(&calls, "systemctl reload"), installed, "a failing ufw delete reloaded nginx"); + assert_eq!(logged(&calls, "ufw delete"), 5); + fs::remove_file(temp.path().join("ufw.fail"))?; + driver.withdraw_candidate_membership(&receipt, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), installed + 1); + assert!(!driver.binding.config_path.exists()); + Ok(()) + } + + /// A crash after the firewall rule went and before the fragment did: the + /// retry reads the gone rule as withdrawn and finishes the withdrawal. + #[cfg(unix)] + #[test] + fn a_withdrawal_resumed_after_its_firewall_rule_went_finishes() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let receipt = preflight_receipt(&driver, &candidate)?; + driver.install(&candidate, &digest('a'), &receipt, false, &Unmetered)?; + let installed = logged(&calls, "systemctl reload"); + + fs::write(temp.path().join("ufw.absent"), b"x")?; + driver.withdraw_candidate_membership(&receipt, &Unmetered)?; + assert!(!driver.binding.config_path.exists(), "the fragment outlived the withdrawal"); + assert_eq!(logged(&calls, "systemctl reload"), installed + 1, "nginx was not reloaded"); + Ok(()) + } + + /// A continuity over its admitted route has the baseline's membership: its + /// install, rollback and withdrawal change nothing and run nothing. + #[cfg(unix)] + #[test] + fn a_transaction_that_does_not_change_the_route_runs_nothing() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let rendered = driver.render(&candidate)?; + let mut receipt = preflight_receipt(&driver, &candidate)?; + receipt.incumbent_runtime_instance_id = Some(digest('b')); + receipt.incumbent_membership_sha256 = Some(sha256_id(&rendered)); + receipt.incumbent_configuration = Some(rendered.clone()); + assert!(!receipt.changes_route()); + fs::write(&driver.binding.config_path, &rendered)?; + let gate = Recording(Default::default()); + + driver.install(&candidate, &digest('a'), &receipt, false, &gate)?; + driver.rollback(&candidate, &digest('a'), &receipt, &gate)?; + driver.withdraw_candidate_membership(&receipt, &gate)?; + assert_eq!(logged(&calls, ""), 0, "an unchanged route ran a program"); + assert!(gate.0.borrow().is_empty(), "an unchanged route was charged"); + assert_eq!(fs::read(&driver.binding.config_path)?, rendered); + + // The first-route receipt does change it. + assert!(preflight_receipt(&driver, &candidate)?.changes_route()); + Ok(()) + } + + /// Bytes on disk are what nginx loaded: a write whose reload fails is + /// undone, so the retry writes and reloads again instead of taking the + /// unloaded bytes for a live route. + #[cfg(unix)] + #[test] + fn a_failed_reload_puts_the_fragment_back_so_the_retry_reloads() -> Result<()> { + let temp = tempfile::tempdir()?; + let (driver, candidate, calls) = failable_route_driver(temp.path())?; + let receipt = preflight_receipt(&driver, &candidate)?; + let rendered = driver.render(&candidate)?; + let fail = temp.path().join("systemctl.fail"); + + // Install, fenced (no rollback) and not. + for rollback_allowed in [false, true] { + fs::write(&fail, b"x")?; + assert!( + driver + .install(&candidate, &digest('a'), &receipt, rollback_allowed, &Unmetered) + .is_err() + ); + assert!(!driver.binding.config_path.exists(), "an unloaded candidate stayed on disk"); + fs::remove_file(&fail)?; + let before = logged(&calls, "systemctl reload"); + driver.install(&candidate, &digest('a'), &receipt, rollback_allowed, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), before + 1, "the retry did not reload"); + assert_eq!(fs::read(&driver.binding.config_path)?, rendered); + if !rollback_allowed { + fs::remove_file(&driver.binding.config_path)?; + } + } + + // Restore: the candidate is live; withdrawing it fails at the reload. + fs::write(&fail, b"x")?; + assert!(driver.withdraw_candidate_membership(&receipt, &Unmetered).is_err()); + assert_eq!(fs::read(&driver.binding.config_path)?, rendered, "the unloaded removal stayed"); + fs::remove_file(&fail)?; + let before = logged(&calls, "systemctl reload"); + driver.withdraw_candidate_membership(&receipt, &Unmetered)?; + assert_eq!(logged(&calls, "systemctl reload"), before + 1, "the retry did not reload"); + assert!(!driver.binding.config_path.exists()); + Ok(()) + } + /// Observation observes: restoring a membership whose fragment is already /// exact touches nothing, and drifted bytes are restored with one reload. #[cfg(unix)] @@ -12014,8 +12270,6 @@ Content-Le".to_vec()), |_| {}), assert!(!config.exists()); assert_eq!(fs::read_dir(&driver.preflight_root)?.count(), 0); - let candidate_bytes = driver.render(&candidate)?; - fs::write(&config, &candidate_bytes)?; let membership_sha256 = driver.install(&candidate, &digest('a'), &preflight, true, &Unmetered)?; assert_eq!( fs::read_to_string(systemctl.with_extension("calls"))?, @@ -12040,15 +12294,16 @@ Content-Le".to_vec()), |_| {}), let next_preflight = driver.preflight(&candidate, &digest('b'), Some(&admitted), &Unmetered)?; assert_eq!(fs::read(&config)?, admitted_bytes); + // The next candidate renders the bytes already live: installing it + // runs nothing, even while a reload would fail. fs::write(systemctl.with_extension("fail"), b"fail\n")?; - assert!( - driver - .install(&candidate, &digest('b'), &next_preflight, false, &Unmetered) - .is_err() - ); - assert_eq!(fs::read(&config)?, admitted_bytes); - fs::remove_file(systemctl.with_extension("fail"))?; driver.install(&candidate, &digest('b'), &next_preflight, false, &Unmetered)?; + assert_eq!(fs::read(&config)?, admitted_bytes); + assert_eq!( + fs::read_to_string(systemctl.with_extension("calls"))?, + "reload\n" + ); + fs::remove_file(systemctl.with_extension("fail"))?; fs::write(&config, b"foreign route\n")?; assert!( driver @@ -12066,7 +12321,9 @@ Content-Le".to_vec()), |_| {}), .restore_admitted_membership(&candidate, &admitted.membership_sha256, &Unmetered) .is_err() ); - assert!(!config.exists()); + // The failed reload puts back what it replaced: the disk shows what + // nginx still serves. + assert_eq!(fs::read(&config)?, b"foreign route after admission\n"); fs::remove_file(systemctl.with_extension("fail"))?; driver.restore_admitted_membership(&candidate, &admitted.membership_sha256, &Unmetered)?; assert_eq!(fs::read(&config)?, admitted_bytes);