From 903e1c8e631b49510b85d80022508b6d89d5cb32 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 10 Sep 2026 22:22:46 +0200 Subject: [PATCH] Key the projection by incarnation; publish Expected after the brake; consume commands once; add cancel The projection kept one Expected/activation/lease slot per target. A deployment's Sealing overwrote that slot with the candidate's Expected, the incumbent failed activation-expected-projection on its next heartbeat and exited, the deployment was aborted for having killed its incumbent, and the command -- left resident after its transaction was retired to history -- read as queued and was frozen again. Odin cycled on that every few minutes. - Topology driver records are keyed {target}@{expected sha}; the incumbent and the sealed candidate are projected side by side. One CAS mutation replaces the six bespoke loops. withdraw_incarnation removes only one incarnation; demote_to_expected_only replaces restore_admitted_expected_only and is used only for an incarnation whose process is gone. Legacy target-keyed records are retired on the first publish for that target. - receive() takes the incarnation; the read-as-absence compensator for a correlation about another incarnation is gone. - Candidate Expected is published as Starting's first step, after the brake admitted the transaction, not in Sealing before it. - A pre-fencing abort withdraws only the candidate; a post-fencing abort also demotes the stopped incumbent. - A consumed command is archived with its last transaction; status still answers for it from history. - An engaged lifecycle brake stops continuity being minted instead of parking a transaction on the target. - idunn cancel retires a queued command as an operator refusal. Verified with cargo test --lib in rust:1.95-bookworm on yggdrasil: 114 passed, 2 ignored. Co-Authored-By: Claude Fable 5.1 --- docs/authority-map.md | 9 +- docs/deployment-authority.md | 24 +- src/control_plane.rs | 523 ++++++++++----- src/drivers.rs | 1161 ++++++++++++++++++---------------- src/provisioning.rs | 4 +- 5 files changed, 1004 insertions(+), 717 deletions(-) diff --git a/docs/authority-map.md b/docs/authority-map.md index ea258ba..e1c0d70 100644 --- a/docs/authority-map.md +++ b/docs/authority-map.md @@ -26,10 +26,11 @@ The detailed contract is presence and health; Idunn's current observed-activation projection; and observations from narrow execution and route drivers. - **Outputs:** private compiled plans and sealed releases; a release-bound - sanitized Expected projection for Odin/CultMesh; one launch-scoped runtime - activation published only after driver observation; process write-lease and - route decisions; continuity actions; and explicit disagreement or failure - records. + sanitized Expected projection for Odin/CultMesh, keyed by incarnation so an + incumbent and its candidate are projected side by side; one launch-scoped + runtime activation published only after driver observation; process + write-lease and route decisions; continuity actions; and explicit + disagreement or failure records. - **Derived state:** dashboards, operator summaries, proxy displays, process listings, and Odin correlation views are projections. They do not establish deployment, write, route, presence, or readiness authority. diff --git a/docs/deployment-authority.md b/docs/deployment-authority.md index fafa1f2..c3bd2fe 100644 --- a/docs/deployment-authority.md +++ b/docs/deployment-authority.md @@ -100,6 +100,24 @@ Every Idunn-managed deployment has an Odin-visible expected projection before dependent services are promoted, and an Odin-observed runtime projection before they are considered ready. +### Projection keying + +The three truths are per incarnation, and so is the projection. Every +Expected, activation, and write-lease record Idunn publishes is keyed by +`{target}@{expected projection sha256}`; only the runtime presence trust +anchor is keyed by target, because it is provider lookup material shared by +every incarnation. Odin keys its presence and correlation records the same way +and reads Idunn's projection by that key. + +A target being replaced therefore has two incarnations projected side by side: +the admitted incumbent under its key, the sealed candidate under its own. +Publishing the candidate cannot change what the incumbent reads about itself, +withdrawing a failed candidate removes only the candidate's records, and a +correlation Odin holds about one incarnation is never mistaken for evidence +about another. A record of those three types keyed by the bare target is the +shape of an earlier single-slot projection; Idunn retires such records the +first time it publishes for that target, and Odin never reads them. + ## Source-owned deployment declaration Each deployable repository publishes one small, reviewable declaration at a @@ -227,8 +245,10 @@ For a singleton incarnation: root-owned actuation stage. The privileged driver never opens Git. 4. Build, test, and package inside the bound runners. Seal the artifact, declaration, binding, launch contract, and input digests. -5. Derive and publish the release-bound Expected incarnation from the validated - sealed release and its exact compiled plan. +5. Derive the release-bound Expected incarnation from the validated sealed + release and its exact compiled plan. Publish it only once the deployment + brake has admitted this exact transaction, under the candidate's own + incarnation key, beside whatever the target's admitted incarnation projects. 6. Issue one opaque runtime-instance activation bound to the Expected digest, pass the exact Expected and activation documents through the actuator's isolated credential mechanism, and start the candidate privately. It has no diff --git a/src/control_plane.rs b/src/control_plane.rs index 0b847ed..83a06b6 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -18,8 +18,7 @@ use cultnet_rs::{ IdunnRuntimeActivationRecord, IdunnServiceIdentity, OdinTopologyAuthenticationContext, OdinTopologyDisagreement, OdinTopologyIdentity, RuntimePresenceAuthenticationContext, ServiceIdentityProfile, ServiceIdentitySigner, ServiceIdentityTrustAnchor, - OdinRuntimeTopologyCorrelationRecord, authenticate_odin_runtime_topology_correlation, - authenticate_runtime_presence_claim, + authenticate_odin_runtime_topology_correlation, authenticate_runtime_presence_claim, correlate_runtime_presence_claim, derive_service_identity_id, evaluate_idunn_continuity_restart, evaluate_idunn_deployment_brake, open_service_identity_at, verify_idunn_deployment_brake_authorization, verify_runtime_authority, @@ -1059,7 +1058,10 @@ impl DeploymentTransaction { self.phase == DeploymentPhase::Complete, "terminal failure is not Complete" ); - let abort = required(&self.post_fencing_abort, "terminal post-fence abort evidence")?; + let abort = required( + &self.post_fencing_abort, + "terminal post-fence abort evidence", + )?; let TransactionCompletion::FailedAfterFencing { error } = self.completion.as_ref().unwrap() else { @@ -1104,10 +1106,16 @@ impl DeploymentTransaction { self.plan.is_some() && self.sealed_release.is_some() && self.installed_release.is_some() - && self.expected.is_some() - && self.expected_publication_sha256.is_some(), + && self.expected.is_some(), "Starting transaction lacks sealed release evidence" ); + // Expected is published as Starting's first step, after the + // brake; from the activation onward it must be there. + ensure!( + self.expected_publication_sha256.is_some() + || (self.activation.is_none() && self.phase == DeploymentPhase::Starting), + "started transaction has not published its Expected" + ); match self.command_kind { CommandKind::Deploy => ensure!( self.frozen_source.is_some() && self.deployment_authorization.is_some(), @@ -1470,6 +1478,11 @@ enum Command { state_store: PathBuf, command_id: Option, }, + Cancel { + state_store: PathBuf, + command_id: String, + requested_by: String, + }, Validate { recipe: PathBuf, binding: Option, @@ -1496,6 +1509,11 @@ pub fn run(args: impl Iterator) -> Result<()> { state_store, command_id, } => status(&state_store, command_id.as_deref()), + Command::Cancel { + state_store, + command_id, + requested_by, + } => cancel(&state_store, &command_id, &requested_by), Command::Validate { recipe, binding } => validate(&recipe, binding.as_deref()), } } @@ -1509,7 +1527,11 @@ fn validate(recipe_path: &Path, binding_path: Option<&Path>) -> Result<()> { .with_context(|| format!("reading recipe {}", recipe_path.display()))?; let recipe = crate::deployment::TargetDeclaration::parse(&recipe_text) .with_context(|| format!("validating recipe {}", recipe_path.display()))?; - println!("recipe ok: {} declares target {}", recipe_path.display(), recipe.target); + println!( + "recipe ok: {} declares target {}", + recipe_path.display(), + recipe.target + ); let Some(binding_path) = binding_path else { return Ok(()); @@ -1527,7 +1549,11 @@ fn validate(recipe_path: &Path, binding_path: Option<&Path>) -> Result<()> { binding .admit(&recipe) .with_context(|| format!("admitting {} against its recipe", binding.target))?; - println!("binding ok: {} admits target {}", binding_path.display(), binding.target); + println!( + "binding ok: {} admits target {}", + binding_path.display(), + binding.target + ); Ok(()) } @@ -1538,6 +1564,7 @@ fn parse(args: impl Iterator) -> Result { "serve" => parse_serve(args), "up" => parse_up(args), "status" => parse_status(args), + "cancel" => parse_cancel(args), "validate" => parse_validate(args), "--help" | "-h" | "help" => bail!(usage()), _ => bail!("unknown Idunn command {command:?}\n\n{}", usage()), @@ -1646,6 +1673,85 @@ fn parse_status(mut args: impl Iterator) -> Result { }) } +fn parse_cancel(mut args: impl Iterator) -> Result { + let command_id = args + .next() + .ok_or_else(|| anyhow!("idunn cancel requires a deployment command id"))?; + require_id(&command_id, "deployment 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 cancel option {argument:?}"), + } + } + require_value(&requested_by, "cancellation requester")?; + Ok(Command::Cancel { + state_store, + command_id, + requested_by, + }) +} + +/// Withdraw a queued deployment command before it is frozen. +/// +/// A command with a live transaction has already changed something and is +/// not cancellable from here; the transaction's own abort paths own that. A +/// queued one has changed nothing, so it is retired the way a rejected one +/// is: a terminal refusal transaction naming the requester goes to history +/// with the command, and `status` reports it as such rather than as a command +/// that never ran. +fn cancel(store_path: &Path, command_id: &str, requested_by: &str) -> Result<()> { + let snapshot = ControlSnapshot::read(store_path)?; + let command = snapshot + .commands + .iter() + .find(|stored| stored.value.command_id == command_id) + .context("deployment command is unknown or already retired")?; + ensure!( + command.value.kind == CommandKind::Deploy, + "only deployment commands can be cancelled; continuity is Idunn's own" + ); + ensure!( + snapshot.transaction_for_command(command_id).is_empty(), + "deployment command already has a live transaction; it is past cancellation" + ); + let now = now_millis()?; + let refusal = DeploymentTransaction::rejected( + &command.value, + anyhow!("cancelled by {requested_by} before freezing"), + now, + )?; + let envelope = transaction_envelope(&refusal, now)?; + ensure!( + SingleFileMessagePackBackingStore::new(store_path).compare_exchange( + &[ + CultCacheExpectedEnvelope { + r#type: DeploymentCommand::TYPE.into(), + key: command_id.to_owned(), + current: Some(command.envelope.clone()), + }, + CultCacheExpectedEnvelope { + r#type: DeploymentTransaction::TYPE.into(), + key: refusal.transaction_id.clone(), + current: None, + }, + ], + std::slice::from_ref(&envelope), + )?, + "deployment command changed before it could be cancelled" + ); + archive_terminal_transaction(store_path, &envelope)?; + println!("{command_id} cancelled"); + Ok(()) +} + fn parse_validate(mut args: impl Iterator) -> Result { let mut recipe = None; let mut binding = None; @@ -2071,6 +2177,30 @@ fn archive_terminal_transaction(state_store: &Path, envelope: &CultCacheEnvelope SingleFileMessagePackBackingStore::new(state_store) .delete_batch_if_unchanged(std::slice::from_ref(envelope)) .context("retiring a finished transaction from the control store")?; + // The command goes with its last transaction. A command is consumed by + // being frozen once; left resident after its transaction retired, it read + // as queued again and was frozen again on the next tick, which is how one + // `idunn up` became an unbounded series of deployment attempts. + let transaction = read_transaction_record(envelope)?; + let snapshot = ControlSnapshot::read(state_store)?; + if !snapshot + .transaction_for_command(&transaction.command_id) + .is_empty() + { + return Ok(()); + } + if let Some(command) = snapshot + .commands + .iter() + .find(|stored| stored.value.command_id == transaction.command_id) + { + SingleFileMessagePackBackingStore::new(&history_store_path(state_store)) + .insert_entry_if_absent(command.envelope.clone()) + .context("archiving a consumed command to history")?; + SingleFileMessagePackBackingStore::new(state_store) + .delete_batch_if_unchanged(std::slice::from_ref(&command.envelope)) + .context("retiring a consumed command from the control store")?; + } Ok(()) } @@ -2092,7 +2222,26 @@ fn read_history_transactions(state_store: &Path) -> Vec { envelopes .into_iter() .filter(|envelope| envelope.r#type == DeploymentTransaction::TYPE) - .filter_map(|envelope| rmp_serde::from_slice::(&envelope.payload).ok()) + .filter_map(|envelope| { + rmp_serde::from_slice::(&envelope.payload).ok() + }) + .collect() +} + +/// Consumed commands, read as leniently as their transactions. +fn read_history_commands(state_store: &Path) -> Vec { + let path = history_store_path(state_store); + if !path.exists() { + return Vec::new(); + } + let Ok(envelopes) = SingleFileMessagePackBackingStore::new(&path).pull_all_read_only_snapshot() + else { + return Vec::new(); + }; + envelopes + .into_iter() + .filter(|envelope| envelope.r#type == DeploymentCommand::TYPE) + .filter_map(|envelope| rmp_serde::from_slice::(&envelope.payload).ok()) .collect() } @@ -2339,14 +2488,19 @@ fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { // which is the operator surface R11 refused to trade away for a smaller // live set. let archived = read_history_transactions(store_path); - let mut commands = snapshot.commands.iter().collect::>(); - commands.sort_by_key(|stored| stored.value.requested_at_unix_millis); + let archived_commands = read_history_commands(store_path); + let mut commands = snapshot + .commands + .iter() + .map(|stored| &stored.value) + .chain(archived_commands.iter()) + .collect::>(); + commands.sort_by_key(|command| command.requested_at_unix_millis); if let Some(command_id) = command_id { - commands.retain(|stored| stored.value.command_id == command_id); + commands.retain(|command| command.command_id == command_id); ensure!(!commands.is_empty(), "deployment command is unknown"); } - for stored in commands { - let command = &stored.value; + for command in commands { let mut transactions = snapshot.transaction_for_command(&command.command_id); transactions.extend( archived @@ -2368,7 +2522,10 @@ fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { if command_id.is_some() { for transaction in transactions { println!(" transaction {}", transaction.transaction_id); - println!(" target {} phase {:?}", transaction.target, transaction.phase); + println!( + " target {} phase {:?}", + transaction.target, transaction.phase + ); if let Some(expected) = &transaction.expected { println!(" runtime {}", expected.runtime_id); println!(" release {}", expected.sealed_release_id); @@ -2565,11 +2722,10 @@ impl Engine { let record: IdunnDeploymentBrakeRecord = rmp_serde::from_slice(&authorization.canonical_brake_bytes)?; if operator_anchor.is_none() { - operator_anchor = Some(read_trust_anchor::< - IdunnDeploymentBrakeOperatorIdentity, - >( - &self.options.deployment_brake_operator_anchor - )?); + operator_anchor = + Some(read_trust_anchor::( + &self.options.deployment_brake_operator_anchor, + )?); } verify_idunn_deployment_brake_authorization( &record, @@ -2732,9 +2888,7 @@ impl Engine { fn run_scheduler_tick(&self) -> Result { let transaction_progress = self.resume_one_transaction()?; let continuity_progress = self.supervise_one_admitted_generation()?; - Ok(!transaction_progress - && !continuity_progress - && self.freeze_one_queued_command()?) + Ok(!transaction_progress && !continuity_progress && self.freeze_one_queued_command()?) } /// Startup and every later loop use the same order: unfinished ownership @@ -3098,13 +3252,10 @@ impl Engine { let demotion = (|| -> Result<()> { let topology = self.topology(); let provider_anchor = self.provider_anchor_for_plan(¤t.value.plan)?; - if !topology.projected_activation_is_present(¤t.value.target)? { + if !topology.projected_activation_is_present(¤t.value.expected)? { return Ok(()); } - topology.restore_admitted_expected_only( - ¤t.value.expected, - &provider_anchor, - Some(¤t.value.activation), + topology.demote_to_expected_only( ¤t.value.expected, &provider_anchor, ¤t.value.activation, @@ -3119,6 +3270,15 @@ impl Engine { ); } let now = now_millis()?; + // An engaged lifecycle brake means no continuity transaction is + // minted at all. A transaction that exists and waits on the brake + // owns the target, and a target owned by a parked restart accepts + // no deployment -- which is the one thing a lifecycle brake must + // never gate. The Sealing-phase check remains as the guard for a + // transaction minted just before the brake was engaged. + if !self.lifecycle_allows_generation(current, now)? { + continue; + } let command = DeploymentCommand { schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(), command_id: format!("continuity-{}", Uuid::new_v4()), @@ -3282,7 +3442,11 @@ impl Engine { snapshot: &ControlSnapshot, current: &Stored, ) -> Result { - let Some(received) = self.topology().receive(¤t.value.target)? else { + let Some(received) = self.topology().receive( + ¤t.value.target, + ¤t.value.expected.canonical_sha256()?, + )? + else { return Ok(false); }; ensure!( @@ -3429,22 +3593,9 @@ impl Engine { }); } - if current.value.expected_publication_sha256.is_none() { - let now = now_millis()?; - let plan = required(¤t.value.plan, "transaction plan")?; - let expected = required(¤t.value.expected, "Expected projection")?; - let provider_anchor = self.provider_anchor_for_plan(plan)?; - let digest = self - .topology() - .publish_expected(expected, &provider_anchor)?; - return self.persist_same_phase(current, |next| { - next.expected_publication_sha256 = Some(digest); - next.updated_at_unix_millis = now; - next.last_error = None; - Ok(()) - }); - } - + // The candidate's Expected is published in Starting, after the brake + // has admitted this exact transaction. Publishing it is the first + // Verse-visible change a deployment makes, and the brake gates changes. let now = now_millis()?; let mut next = current.value.clone(); match current.value.command_kind { @@ -3482,6 +3633,22 @@ impl Engine { let plan = required(¤t.value.plan, "transaction plan")?; let release = required(¤t.value.sealed_release, "sealed release")?; let installed = required(¤t.value.installed_release, "installed release")?; + if current.value.expected_publication_sha256.is_none() { + // Published under the candidate's own incarnation key, beside + // whatever the target's admitted incarnation currently projects. + // Nothing the incumbent reads about itself changes here. + let now = now_millis()?; + let provider_anchor = self.provider_anchor_for_plan(plan)?; + let digest = self + .topology() + .publish_expected(expected, &provider_anchor)?; + return self.persist_same_phase(current, |next| { + next.expected_publication_sha256 = Some(digest); + next.updated_at_unix_millis = now; + next.last_error = None; + Ok(()) + }); + } if current.value.activation.is_none() { validate_live_providers_for_deploy(current.value.command_kind, || { self.validate_selected_providers_current(plan, now_millis()?) @@ -3661,7 +3828,13 @@ impl Engine { // identity turns ordinary UID reuse into a permanent refusal to // fence. There is nothing left to be isolated from. let incumbent_workload = match incumbent { - Some(value) if self.workload.is_permanently_stopped(&value.value.workload)? => None, + Some(value) + if self + .workload + .is_permanently_stopped(&value.value.workload)? => + { + None + } other => other.map(|value| &value.value.workload), }; let isolation = prove_isolation(workload, incumbent_workload)?; @@ -4574,6 +4747,35 @@ impl Engine { } } + /// The lifecycle brake as it applies to restarting one admitted + /// generation: the same evaluation `lifecycle_allows` makes for a + /// continuity transaction, asked before one is minted. + fn lifecycle_allows_generation( + &self, + current: &Stored, + now: u64, + ) -> Result { + let binding = current.value.plan.parsed_inputs()?.1; + let expected = ¤t.value.expected; + match read_lifecycle_brake(&binding.brakes.lifecycle_store) { + Ok(Some(record)) => Ok(evaluate_idunn_continuity_restart( + IdunnLifecycleBrakeObservation::Present(&record), + &expected.runtime_id, + ¤t.value.target, + now, + ) + .is_ok()), + Ok(None) => Ok(evaluate_idunn_continuity_restart( + IdunnLifecycleBrakeObservation::Missing, + &expected.runtime_id, + ¤t.value.target, + now, + ) + .is_ok()), + Err(_) => Ok(false), + } + } + fn record_gate_wait( &self, current: &Stored, @@ -4856,7 +5058,16 @@ impl Engine { AuthenticatedOdinRuntimeTopologyCorrelation, )>, > { - let Some(received) = self.topology().receive(¤t.value.target)? else { + // Odin keeps one correlation per incarnation, so the transaction's own + // Expected selects exactly the evidence about it. A correlation about + // the incarnation this one is replacing lives under another key and is + // never handed back here. + let expected_sha256 = + required(¤t.value.expected, "Expected projection")?.canonical_sha256()?; + let Some(received) = self + .topology() + .receive(¤t.value.target, &expected_sha256)? + else { return Ok(None); }; ensure!( @@ -4874,29 +5085,6 @@ impl Engine { live.envelope == current.envelope, "transaction changed before topology admission" ); - // A correlation that describes another incarnation is not evidence - // about this one, and it is not a fault either. Odin's stored - // correlation outlives the incarnation that wrote it, so a target being - // replaced -- Odin above all, which writes its own -- routinely finds - // one describing the Expected it is superseding. Treating that as an - // error failed the deployment that would have replaced it, and only the - // superseded process could have refreshed it. - // - // Read as absence instead. This says nothing about trust: the record is - // unauthenticated here and is used only to decide it is about something - // else, which is the same conclusion an attacker would get for free by - // publishing nothing. Everything that follows is fully authenticated. - let describes_this_incarnation = (|| -> Result { - let expected = required(&live.value.expected, "Expected projection")?; - let (record, _) = OdinRuntimeTopologyCorrelationRecord::decode_canonical_signed_payload( - &received.canonical_bytes, - )?; - Ok(record.expected_projection_sha256 == expected.canonical_sha256()?) - })() - .unwrap_or(false); - if !describes_this_incarnation { - return Ok(None); - } let authenticated = self.authenticate_topology_bytes( &snapshot, &live.value, @@ -5088,7 +5276,10 @@ impl Engine { &odin_authority.signer_public_key, self.trusted_topology_context(now), )?; - validate_authenticated_evidence(&generation.latest_odin_observation, &authenticated_latest)?; + validate_authenticated_evidence( + &generation.latest_odin_observation, + &authenticated_latest, + )?; ensure!( is_semantic_ready(&authenticated_latest), "admitted provider's latest Odin observation is no longer Ready" @@ -5258,7 +5449,10 @@ impl Engine { /// /// A transaction that has never started one has no candidate to be dead, so /// it is not permanently stopped -- it is simply not there yet. - fn candidate_is_permanently_stopped(&self, transaction: &DeploymentTransaction) -> Result { + fn candidate_is_permanently_stopped( + &self, + transaction: &DeploymentTransaction, + ) -> Result { let Some(workload) = &transaction.workload else { return Ok(false); }; @@ -5394,22 +5588,27 @@ impl Engine { if abort.topology_reconciliation == CleanupEvidence::Pending { let expected = required(¤t.value.expected, "failed Expected projection")?; let snapshot = ControlSnapshot::read(&self.options.state_store)?; - if let Some(incumbent) = self.exact_incumbent(&snapshot, ¤t.value)? { - // Restoring the incumbent's admitted Expected is what lets - // continuity bring it back. Fencing stopped it and revoked its - // lease, and continuity is the organ that restarts an admitted - // incarnation; deployment does not restart it from here. - let topology = self.topology(); - let failed_provider_anchor = self.provider_anchor_for_plan(required( - ¤t.value.plan, - "failed transaction plan", - )?)?; - let admitted_provider_anchor = - self.provider_anchor_for_plan(&incumbent.value.plan)?; - let expected_sha256 = topology.restore_admitted_expected_only( + // The failed candidate's records go, under its own key. Fencing + // stopped the incumbent and revoked its lease, so the incumbent is + // demoted to Expected-only under its key, which is what lets + // continuity bring it back; deployment does not restart it here. + let topology = self.topology(); + let failed_provider_anchor = self.provider_anchor_for_plan(required( + ¤t.value.plan, + "failed transaction plan", + )?)?; + topology + .withdraw_incarnation( expected, &failed_provider_anchor, current.value.activation.as_ref(), + None, + ) + .context("withdrawing the failed candidate projection")?; + if let Some(incumbent) = self.exact_incumbent(&snapshot, ¤t.value)? { + let admitted_provider_anchor = + self.provider_anchor_for_plan(&incumbent.value.plan)?; + let expected_sha256 = topology.demote_to_expected_only( &incumbent.value.expected, &admitted_provider_anchor, &incumbent.value.activation, @@ -5419,17 +5618,6 @@ impl Engine { expected_sha256 == incumbent.value.expected.canonical_sha256()?, "restored incumbent Expected receipt differs" ); - } else { - let plan = required(¤t.value.plan, "failed transaction plan")?; - let provider_anchor = self.provider_anchor_for_plan(plan)?; - self.topology() - .withdraw_expected( - expected, - &provider_anchor, - current.value.activation.as_ref(), - None, - ) - .context("withdrawing the abandoned Expected projection")?; } return self.persist_same_phase(current, |next| { next.post_fencing_abort @@ -5447,7 +5635,8 @@ impl Engine { ) .context("cleaning the abandoned source")?; return self.persist_same_phase(current, |next| { - next.post_fencing_abort.as_mut().unwrap().source_cleanup = CleanupEvidence::Complete; + next.post_fencing_abort.as_mut().unwrap().source_cleanup = + CleanupEvidence::Complete; Ok(()) }); } @@ -5492,40 +5681,21 @@ impl Engine { } if abort.topology_reconciliation == CleanupEvidence::Pending { let expected = required(¤t.value.expected, "failed Expected projection")?; - let snapshot = ControlSnapshot::read(&self.options.state_store)?; - if let Some(incumbent) = self.exact_incumbent(&snapshot, ¤t.value)? { - let topology = self.topology(); - let failed_provider_anchor = self.provider_anchor_for_plan(required( - ¤t.value.plan, - "failed transaction plan", - )?)?; - let admitted_provider_anchor = - self.provider_anchor_for_plan(&incumbent.value.plan)?; - let expected_sha256 = topology.restore_admitted_expected_only( + // Before fencing the incumbent was never touched: it is still + // running under its own key with its activation and lease. Only the + // failed candidate's records are withdrawn. The incumbent's + // projection is not demoted -- doing so would tell a live process + // it has no activation. + let plan = required(¤t.value.plan, "failed transaction plan")?; + let provider_anchor = self.provider_anchor_for_plan(plan)?; + self.topology() + .withdraw_incarnation( expected, - &failed_provider_anchor, + &provider_anchor, current.value.activation.as_ref(), - &incumbent.value.expected, - &admitted_provider_anchor, - &incumbent.value.activation, - incumbent.value.leasing.lease(), - )?; - ensure!( - expected_sha256 == incumbent.value.expected.canonical_sha256()?, - "restored incumbent Expected receipt differs" - ); - } else { - let plan = required(¤t.value.plan, "failed transaction plan")?; - let provider_anchor = self.provider_anchor_for_plan(plan)?; - self.topology() - .withdraw_expected( - expected, - &provider_anchor, - current.value.activation.as_ref(), - None, - ) - .context("withdrawing exact failed Expected projection")?; - } + None, + ) + .context("withdrawing exact failed Expected projection")?; return self.persist_same_phase(current, |next| { next.pre_fencing_abort .as_mut() @@ -5989,6 +6159,7 @@ fn usage() -> &'static str { idunn serve [runtime options]\n\ idunn up [--state-store PATH] [--no-wait]\n\ idunn status [--state-store PATH] [--command ID]\n\ + idunn cancel [--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\ @@ -5999,7 +6170,8 @@ fn usage() -> &'static str { mod tests { use cultnet_rs::{ GameCultProviderHealthIdentity, OdinRuntimeTopologyCorrelationPurpose, - enroll_service_identity_at, export_service_identity_trust_anchor, + OdinRuntimeTopologyCorrelationRecord, enroll_service_identity_at, + export_service_identity_trust_anchor, }; use tempfile::TempDir; @@ -6067,18 +6239,27 @@ mod tests { }; println!( "LIVE {} target={} phase={:?} ready_seq={} latest_seq={}", - value.transaction_id, value.target, value.phase, ready.publisher_sequence, + value.transaction_id, + value.target, + value.phase, + ready.publisher_sequence, latest.publisher_sequence ); - println!(" old gate (receipt == cursor): {}", if ready == latest { "PASS" } else { "REFUSE" }); + println!( + " old gate (receipt == cursor): {}", + if ready == latest { "PASS" } else { "REFUSE" } + ); let (latest_record, _) = OdinRuntimeTopologyCorrelationRecord::decode_canonical_signed_payload( &latest.canonical_bytes, )?; println!( " cursor says: projection={} instance={:?} ready={} present={} disagreements={}", - latest_record.expected_projection_sha256, latest_record.runtime_instance_id, - latest_record.ready, latest_record.present, latest_record.disagreements.len() + latest_record.expected_projection_sha256, + latest_record.runtime_instance_id, + latest_record.ready, + latest_record.present, + latest_record.disagreements.len() ); let (Some(expected), Some(activation)) = (&value.expected, &value.activation) else { println!(" transaction has not declared an incarnation yet"); @@ -6086,14 +6267,13 @@ mod tests { }; println!( " declared: projection={} instance={}", - expected.canonical_sha256()?, activation.runtime_instance_id + expected.canonical_sha256()?, + activation.runtime_instance_id ); let binds = latest_record.expected_projection_sha256 == expected.canonical_sha256()? && latest_record.runtime_instance_id.as_deref() == Some(activation.runtime_instance_id.as_str()); - println!( - " binds declared incarnation (what authentication enforces): {binds}" - ); + println!(" binds declared incarnation (what authentication enforces): {binds}"); } Ok(()) } @@ -6185,7 +6365,10 @@ mod tests { key: command.command_id.clone(), current: None, }], - &[command_envelope(&command, command.requested_at_unix_millis)?], + &[command_envelope( + &command, + command.requested_at_unix_millis + )?], )? ); let mut opening = live.clone(); @@ -6213,7 +6396,9 @@ mod tests { // Gone from the set that still gates decisions... assert!( - ControlSnapshot::read(&world.state_store)?.transactions.is_empty(), + ControlSnapshot::read(&world.state_store)? + .transactions + .is_empty(), "a finished transaction stayed resident" ); // ...and still answerable, which is what R11 refused to trade away. @@ -6221,6 +6406,19 @@ mod tests { assert_eq!(archived.len(), 1); assert_eq!(archived[0].transaction_id, live.transaction_id); assert!(history.exists()); + // The command went with it. Left resident it would read as queued and + // be frozen again next tick; consumed, it is history beside its + // transaction and `status` still answers for it. + assert!( + ControlSnapshot::read(&world.state_store)? + .commands + .is_empty(), + "a consumed command stayed resident" + ); + let archived_commands = read_history_commands(&world.state_store); + assert_eq!(archived_commands.len(), 1); + assert_eq!(archived_commands[0].command_id, command.command_id); + assert!(!world.engine.freeze_one_queued_command()?); Ok(()) } @@ -6280,7 +6478,11 @@ mod tests { // describing one target's consent. let world = EngineFixture::new()?; assert!( - !world.engine.options.deployment_brake_operator_anchor.exists(), + !world + .engine + .options + .deployment_brake_operator_anchor + .exists(), "fixture must not have written a brake anchor" ); let snapshot = ControlSnapshot::read(&world.state_store)?; @@ -6416,7 +6618,11 @@ mod tests { ) } - fn correlation(&self, sequence: u64, ready: bool) -> Result { + fn correlation( + &self, + sequence: u64, + ready: bool, + ) -> Result { Ok(OdinRuntimeTopologyCorrelationRecord { schema_version: cultnet_rs::ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into(), target: self.expected.target.clone(), @@ -6445,7 +6651,9 @@ mod tests { fn sign(&self, record: &mut OdinRuntimeTopologyCorrelationRecord) -> Result> { record.signature = self .odin_signer - .sign::(&record.unsigned_signature_payload()?) + .sign::( + &record.unsigned_signature_payload()?, + ) .signature; record.canonical_bytes() } @@ -6671,10 +6879,7 @@ mod tests { // Likewise refused at encoding: a Present record must carry its // instance id ("runtime instance identity and observation evidence are // partial"). - assert!( - format!("{error:#}").contains("are partial"), - "{error:#}" - ); + assert!(format!("{error:#}").contains("are partial"), "{error:#}"); Ok(()) } @@ -6688,7 +6893,9 @@ mod tests { let mut stranger = world.correlation(501_713, true)?; stranger.target = "ghostlight".into(); let canonical = world.sign(&mut stranger)?; - let error = world.authenticate(&canonical).expect_err("target must bind"); + let error = world + .authenticate(&canonical) + .expect_err("target must bind"); assert!( format!("{error:#}").contains("substitutes or omits Expected authority"), "{error:#}" @@ -6791,7 +6998,9 @@ mod tests { let before_observation = NOW - DEFAULT_TOPOLOGY_MAXIMUM_FUTURE_SKEW_MILLIS - 1; assert!( - world.authenticate_at(&canonical, before_observation).is_err(), + world + .authenticate_at(&canonical, before_observation) + .is_err(), "an admitted_at earlier than the observation must not authenticate" ); Ok(()) @@ -6823,7 +7032,10 @@ mod tests { value.command_kind, value.phase, value.completion.is_some(), - value.ready.as_ref().map(|evidence| evidence.publisher_sequence), + value + .ready + .as_ref() + .map(|evidence| evidence.publisher_sequence), value .latest_odin_observation .as_ref() @@ -6949,10 +7161,7 @@ mod tests { .map(str::to_owned), ) .unwrap(); - assert!(matches!( - parsed, - Command::Validate { binding: None, .. } - )); + assert!(matches!(parsed, Command::Validate { binding: None, .. })); assert!( parse( @@ -6964,7 +7173,13 @@ mod tests { ); assert!(parse(["validate"].into_iter().map(str::to_owned)).is_err()); for rejected in [ - vec!["validate", "--recipe", "/r.toml", "--deploy-command", "sh -c bad"], + vec![ + "validate", + "--recipe", + "/r.toml", + "--deploy-command", + "sh -c bad", + ], vec!["validate", "--recipe", "/r.toml", "--state-store", "/s.cc"], ] { assert!(parse(rejected.into_iter().map(str::to_owned)).is_err()); diff --git a/src/drivers.rs b/src/drivers.rs index f264d4c..7c45403 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -314,7 +314,9 @@ pub trait TopologyPort { expected: &IdunnExpectedIncarnationRecord, provider_anchor: &ServiceIdentityTrustAnchor, ) -> Result; - fn withdraw_expected( + /// Remove every record of one incarnation. Other incarnations of the same + /// target, and the target's anchor while any remain, are untouched. + fn withdraw_incarnation( &self, expected: &IdunnExpectedIncarnationRecord, provider_anchor: &ServiceIdentityTrustAnchor, @@ -339,12 +341,20 @@ pub trait TopologyPort { activation: &IdunnRuntimeActivationRecord, lease: Option<&IdunnProcessWriteLeaseRecord>, ) -> Result<()>; - fn receive(&self, target: &str) -> Result>; + /// Odin's correlation for one exact incarnation. A correlation about another + /// incarnation of the same target is not this one's evidence and is never + /// returned for it. + fn receive( + &self, + target: &str, + expected_sha256: &str, + ) -> Result>; } #[derive(Clone, Debug, PartialEq, Eq)] pub struct ReceivedOdinTopologyCorrelation { pub target: String, + pub expected_sha256: String, pub canonical_bytes: Vec, } @@ -2487,7 +2497,10 @@ impl SystemdTransientWorkloadDriver { observation.load_credential == expected_load_credential, "load credential", ), - (observation.working_directory == installed, "working directory"), + ( + observation.working_directory == installed, + "working directory", + ), (observation.runtime_bundle == bundle, "runtime bundle"), ( observation.credentials_directory == expected_credentials_directory, @@ -3219,203 +3232,288 @@ impl WriteLeasePort for CultCacheWriteLeaseDriver { /// Idunn projects desired identity plus its own activation and current-lease /// facts here. Service presence is absent by construction; only Odin may /// correlate these records with signed runtime observation into Present/Ready. +/// +/// Every record is keyed by the incarnation it describes, never by the target +/// alone. A target being replaced has two incarnations at once -- the admitted +/// incumbent and the sealed candidate -- and both are projected side by side. +/// Publishing the candidate therefore cannot touch what the incumbent is; the +/// one target-keyed record is the runtime presence trust anchor, which is +/// provider lookup material and the same for every incarnation. pub struct CultCacheTopologyDriver { pub projection_store: PathBuf, pub correlation_store: PathBuf, } -impl CultCacheTopologyDriver { - /// Atomically demotes a failed pre-fence candidate back to the admitted - /// generation's Expected projection. Runtime activation and write-lease - /// records are deliberately absent until admitted-body supervision proves - /// the physical process and lease again. - pub fn restore_admitted_expected_only( - &self, - failed_expected: &IdunnExpectedIncarnationRecord, - failed_provider_anchor: &ServiceIdentityTrustAnchor, - failed_activation: Option<&IdunnRuntimeActivationRecord>, - admitted_expected: &IdunnExpectedIncarnationRecord, - admitted_provider_anchor: &ServiceIdentityTrustAnchor, - admitted_activation: &IdunnRuntimeActivationRecord, - admitted_lease: Option<&IdunnProcessWriteLeaseRecord>, - ) -> Result { - failed_expected.validate()?; - admitted_expected.validate()?; - ensure!( - failed_expected.target == admitted_expected.target, - "projection restoration crosses deployment targets" - ); - let failed_anchor = runtime_presence_trust_anchor(failed_expected, failed_provider_anchor)?; - let admitted_anchor = - runtime_presence_trust_anchor(admitted_expected, admitted_provider_anchor)?; - if let Some(activation) = failed_activation { - activation.validate()?; +/// The key one incarnation's projection records live under. +/// +/// This is a contract with Odin's reader (`IncarnationRef::key` in +/// `odin-daemon`): `{target}@{expected projection sha256}`. +pub fn incarnation_key(expected: &IdunnExpectedIncarnationRecord) -> Result { + Ok(incarnation_key_of( + &expected.target, + &expected.canonical_sha256()?, + )) +} + +pub fn incarnation_key_of(target: &str, expected_sha256: &str) -> String { + format!("{target}@{expected_sha256}") +} + +/// The three record types that belong to one incarnation. +fn is_incarnation_record_type(record_type: &str) -> bool { + matches!( + record_type, + IdunnExpectedIncarnationRecord::TYPE + | IdunnRuntimeActivationRecord::TYPE + | IdunnProcessWriteLeaseRecord::TYPE + ) +} + +/// One incarnation's records as currently projected, each checked to be the +/// exact document the caller names. A record present under the key but +/// differing from the caller's is a substitution and is refused. +struct ProjectedIncarnation { + expected: Option, + activation: Option, + lease: Option, +} + +impl ProjectedIncarnation { + fn read( + entries: &[CultCacheEnvelope], + expected: &IdunnExpectedIncarnationRecord, + activation: Option<&IdunnRuntimeActivationRecord>, + lease: Option<&IdunnProcessWriteLeaseRecord>, + ) -> Result { + let key = incarnation_key(expected)?; + let projected_expected = + projection_entry(entries, IdunnExpectedIncarnationRecord::TYPE, &key)?; + if let Some(envelope) = projected_expected { ensure!( - activation.expected_projection_sha256 == failed_expected.canonical_sha256()?, - "failed activation does not bind the failed Expected projection" + envelope.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) + && IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)? + == *expected, + "projected Expected under this incarnation key is substituted" ); } - admitted_activation.validate()?; - ensure!( - admitted_activation.expected_projection_sha256 - == admitted_expected.canonical_sha256()?, - "admitted activation does not bind the admitted Expected projection" - ); - if let Some(lease) = admitted_lease { - validate_topology_lease(admitted_expected, admitted_activation, lease)?; + let projected_activation = + projection_entry(entries, IdunnRuntimeActivationRecord::TYPE, &key)?; + if let Some(envelope) = projected_activation { + let current = IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload)?; + ensure!( + envelope.schema_id.as_deref() == Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA) + && activation == Some(¤t), + "projected activation under this incarnation key is substituted" + ); } + let projected_lease = projection_entry(entries, IdunnProcessWriteLeaseRecord::TYPE, &key)?; + if let Some(envelope) = projected_lease { + let current = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; + ensure!( + envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA) + && lease == Some(¤t), + "projected write lease under this incarnation key is substituted" + ); + } + Ok(Self { + expected: projected_expected.cloned(), + activation: projected_activation.cloned(), + lease: projected_lease.cloned(), + }) + } +} + +fn expected_envelope(expected: &IdunnExpectedIncarnationRecord) -> Result { + Ok(CultCacheEnvelope { + key: incarnation_key(expected)?, + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: expected.canonical_bytes()?, + stored_at: chrono::Utc::now().to_rfc3339(), + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }) +} + +fn anchor_envelope(anchor: &GameCultServiceTrustAnchorRecord) -> Result { + Ok(CultCacheEnvelope { + key: anchor.trust_anchor_id.clone(), + r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), + payload: rmp_serde::to_vec(anchor)?, + stored_at: rfc3339_millis(anchor.bound_at_unix_millis)?, + schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), + }) +} + +fn activation_envelope( + key: &str, + activation: &IdunnRuntimeActivationRecord, +) -> Result { + Ok(CultCacheEnvelope { + key: key.to_owned(), + r#type: IdunnRuntimeActivationRecord::TYPE.into(), + payload: activation.canonical_bytes()?, + stored_at: rfc3339_millis(activation.issued_at_unix_millis)?, + schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), + }) +} + +/// Same-content envelopes compare equal regardless of `stored_at`, which is +/// a publication timestamp and not part of the record's identity. +fn same_record(current: &CultCacheEnvelope, replacement: &CultCacheEnvelope) -> bool { + current.key == replacement.key + && current.r#type == replacement.r#type + && current.schema_id == replacement.schema_id + && current.payload == replacement.payload +} + +impl CultCacheTopologyDriver { + fn snapshot(&self) -> Result> { + if !self.projection_store.exists() { + return Ok(Vec::new()); + } + SingleFileMessagePackBackingStore::new(&self.projection_store).pull_all_read_only_snapshot() + } + + /// Apply one whole-snapshot mutation with compare-and-swap. `mutate` + /// returns `None` when the snapshot already has the shape it wants, and + /// the replacement set otherwise; every writer below is one of these. + fn mutate(&self, mutate: F) -> Result<()> + where + F: Fn(&[CultCacheEnvelope]) -> Result>>, + { if let Some(parent) = self.projection_store.parent() { fs::create_dir_all(parent)?; } let store = SingleFileMessagePackBackingStore::new(&self.projection_store); - let target = admitted_expected.target.as_str(); - let anchor_key = runtime_presence_trust_anchor_id(target); for _ in 0..8 { - let entries = store.pull_all_read_only_snapshot()?; - let current_expected = - projection_entry(&entries, IdunnExpectedIncarnationRecord::TYPE, target)?; - if let Some(envelope) = current_expected { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA), - "restored Expected projection schema is foreign" - ); - let current = IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)?; - ensure!( - current == *failed_expected || current == *admitted_expected, - "refusing to replace an unknown Expected projection" - ); + let entries = self.snapshot()?; + let Some(replacement) = mutate(&entries)? else { + return Ok(()); + }; + if store.compare_exchange_snapshot(&entries, &replacement)? { + publish_projection_mode(&self.projection_store)?; + return Ok(()); } + } + bail!("CultCache projection changed repeatedly during publication") + } + + fn other_incarnations_of(entries: &[CultCacheEnvelope], target: &str, key: &str) -> bool { + entries.iter().any(|envelope| { + envelope.r#type == IdunnExpectedIncarnationRecord::TYPE + && envelope.key != key + && envelope + .key + .split_once('@') + .is_some_and(|(owner, _)| owner == target) + }) + } + + /// Records this contract does not own: the three incarnation types keyed by + /// the bare target, as the previous single-slot projection wrote them. A + /// publish for that target retires them; nothing reads them. + fn is_legacy_slot_record(envelope: &CultCacheEnvelope, target: &str) -> bool { + is_incarnation_record_type(&envelope.r#type) && envelope.key == target + } + + /// Demote one admitted incarnation to Expected-only. Its activation and + /// write lease are withdrawn; its Expected and anchor are ensured present. + /// This is for an incarnation whose process is gone -- continuity about + /// to restart it, or an abort that already fenced and stopped it -- and + /// says nothing about any other incarnation of the target. + pub fn demote_to_expected_only( + &self, + expected: &IdunnExpectedIncarnationRecord, + provider_anchor: &ServiceIdentityTrustAnchor, + activation: &IdunnRuntimeActivationRecord, + lease: Option<&IdunnProcessWriteLeaseRecord>, + ) -> Result { + expected.validate()?; + activation.validate()?; + ensure!( + activation.expected_projection_sha256 == expected.canonical_sha256()?, + "admitted activation does not bind the admitted Expected projection" + ); + if let Some(lease) = lease { + validate_topology_lease(expected, activation, lease)?; + } + let anchor = runtime_presence_trust_anchor(expected, provider_anchor)?; + let key = incarnation_key(expected)?; + self.mutate(|entries| { + let projected = ProjectedIncarnation::read(entries, expected, Some(activation), lease)?; let current_anchor = projection_entry( - &entries, + entries, GameCultServiceTrustAnchorRecord::TYPE, - &anchor_key, + &anchor.trust_anchor_id, )?; if let Some(envelope) = current_anchor { - let current = service_trust_anchor_from_envelope(envelope)?; ensure!( - current == failed_anchor || current == admitted_anchor, + service_trust_anchor_from_envelope(envelope)? == anchor, "refusing to replace an unknown runtime presence trust anchor" ); } - let current_activation = - projection_entry(&entries, IdunnRuntimeActivationRecord::TYPE, target)?; - if let Some(envelope) = current_activation { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA), - "restored activation projection schema is foreign" - ); - let current = IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload)?; - ensure!( - failed_activation == Some(¤t) || admitted_activation == ¤t, - "refusing to remove an unknown runtime activation projection" - ); - } - let current_lease = - projection_entry(&entries, IdunnProcessWriteLeaseRecord::TYPE, target)?; - if let Some(envelope) = current_lease { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA), - "restored write-lease projection schema is foreign" - ); - let current = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; - ensure!( - admitted_lease == Some(¤t), - "refusing to remove an unknown process write-lease projection" - ); - } - if current_expected.is_some_and(|envelope| { - IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload) - .is_ok_and(|current| current == *admitted_expected) - }) && current_anchor.is_some_and(|envelope| { - service_trust_anchor_from_envelope(envelope) - .is_ok_and(|current| current == admitted_anchor) - }) && current_activation.is_none() - && current_lease.is_none() + if projected.expected.is_some() + && current_anchor.is_some() + && projected.activation.is_none() + && projected.lease.is_none() { - return admitted_expected.canonical_sha256(); + return Ok(None); } let mut replacement = entries .iter() .filter(|envelope| { - !((envelope.key == target - && matches!( - envelope.r#type.as_str(), - IdunnExpectedIncarnationRecord::TYPE - | IdunnRuntimeActivationRecord::TYPE - | IdunnProcessWriteLeaseRecord::TYPE - )) - || (envelope.key == anchor_key - && envelope.r#type == GameCultServiceTrustAnchorRecord::TYPE)) + !(envelope.key == key + || (envelope.r#type == GameCultServiceTrustAnchorRecord::TYPE + && envelope.key == anchor.trust_anchor_id)) }) .cloned() .collect::>(); - replacement.extend([ - CultCacheEnvelope { - key: admitted_expected.target.clone(), - r#type: IdunnExpectedIncarnationRecord::TYPE.into(), - payload: admitted_expected.canonical_bytes()?, - stored_at: chrono::Utc::now().to_rfc3339(), - schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), - }, - CultCacheEnvelope { - key: admitted_anchor.trust_anchor_id.clone(), - r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), - payload: rmp_serde::to_vec(&admitted_anchor)?, - stored_at: rfc3339_millis(admitted_anchor.bound_at_unix_millis)?, - schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), - }, - ]); - if store.compare_exchange_snapshot(&entries, &replacement)? { - publish_projection_mode(&self.projection_store)?; - return admitted_expected.canonical_sha256(); - } - } - bail!("topology projection changed repeatedly during admitted restoration") + replacement.push(match projected.expected { + Some(envelope) => envelope, + None => expected_envelope(expected)?, + }); + replacement.push(match current_anchor { + Some(envelope) => envelope.clone(), + None => anchor_envelope(&anchor)?, + }); + Ok(Some(replacement)) + })?; + expected.canonical_sha256() } - /// Whether the projection currently names any activation for this target. - /// - /// Continuity asks so it can tell a projection that still describes a dead - /// incarnation from one that has already been demoted to Expected-only. - pub fn projected_activation_is_present(&self, target: &str) -> Result { - if !self.projection_store.exists() { - return Ok(false); - } - let entries = SingleFileMessagePackBackingStore::new(&self.projection_store) - .pull_all_read_only_snapshot()?; - Ok(projection_entry(&entries, IdunnRuntimeActivationRecord::TYPE, target)?.is_some()) + /// Whether the projection currently names an activation for this + /// incarnation. Continuity asks so it can tell a projection that still + /// describes a dead incarnation from one already demoted to Expected-only. + pub fn projected_activation_is_present( + &self, + expected: &IdunnExpectedIncarnationRecord, + ) -> Result { + let key = incarnation_key(expected)?; + Ok( + projection_entry(&self.snapshot()?, IdunnRuntimeActivationRecord::TYPE, &key)? + .is_some(), + ) } - /// Whether the projection already carries this admitted Expected. - /// - /// Deliberately says nothing about the activation. Continuity restores the - /// Expected for a target whose process is gone, and at that moment there is - /// no activation to be exact about -- the next one is issued when the - /// workload starts. + /// Whether the projection already carries this Expected and its anchor. + /// Deliberately says nothing about the activation. pub fn admitted_expected_projection_is_exact( &self, expected: &IdunnExpectedIncarnationRecord, provider_anchor: &ServiceIdentityTrustAnchor, ) -> Result { expected.validate()?; - if !self.projection_store.exists() { - return Ok(false); - } - let entries = SingleFileMessagePackBackingStore::new(&self.projection_store) - .pull_all_read_only_snapshot()?; + let entries = self.snapshot()?; let anchor = runtime_presence_trust_anchor(expected, provider_anchor)?; - let expected_is_exact = match projection_entry( - &entries, - IdunnExpectedIncarnationRecord::TYPE, - &expected.target, - )? { - Some(envelope) => { - envelope.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) - && IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)? - == *expected - } - None => false, - }; + let key = incarnation_key(expected)?; + let expected_is_exact = + match projection_entry(&entries, IdunnExpectedIncarnationRecord::TYPE, &key)? { + Some(envelope) => { + envelope.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) + && IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)? + == *expected + } + None => false, + }; let anchor_is_exact = match projection_entry( &entries, GameCultServiceTrustAnchorRecord::TYPE, @@ -3443,31 +3541,10 @@ impl CultCacheTopologyDriver { if let Some(lease) = lease { validate_topology_lease(expected, activation, lease)?; } - if !self.projection_store.exists() { - return Ok(false); - } - let entries = SingleFileMessagePackBackingStore::new(&self.projection_store) - .pull_all_read_only_snapshot()?; + let entries = self.snapshot()?; let anchor = runtime_presence_trust_anchor(expected, provider_anchor)?; - let expected_is_exact = match projection_entry( - &entries, - IdunnExpectedIncarnationRecord::TYPE, - &expected.target, - )? { - Some(envelope) => { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA), - "admitted Expected projection schema is foreign" - ); - ensure!( - IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)? - == *expected, - "admitted Expected projection was replaced by another incarnation" - ); - true - } - None => false, - }; + let projected = ProjectedIncarnation::read(&entries, expected, Some(activation), lease) + .context("admitted runtime projection was replaced")?; let anchor_is_exact = match projection_entry( &entries, GameCultServiceTrustAnchorRecord::TYPE, @@ -3482,45 +3559,10 @@ impl CultCacheTopologyDriver { } None => false, }; - let activation_is_exact = match projection_entry( - &entries, - IdunnRuntimeActivationRecord::TYPE, - &expected.target, - )? { - Some(envelope) => { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA), - "admitted activation projection schema is foreign" - ); - ensure!( - IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload)? - == *activation, - "admitted activation projection was replaced" - ); - true - } - None => false, - }; - let lease_is_exact = match projection_entry( - &entries, - IdunnProcessWriteLeaseRecord::TYPE, - &expected.target, - )? { - Some(envelope) => { - ensure!( - envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA), - "admitted write-lease projection schema is foreign" - ); - let current = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; - ensure!( - lease == Some(¤t), - "admitted write-lease projection was replaced" - ); - true - } - None => lease.is_none(), - }; - Ok(expected_is_exact && anchor_is_exact && activation_is_exact && lease_is_exact) + Ok(projected.expected.is_some() + && anchor_is_exact + && projected.activation.is_some() + && projected.lease.is_some() == lease.is_some()) } } @@ -3617,47 +3659,6 @@ fn service_trust_anchor_from_envelope( Ok(anchor) } -fn replace_records_atomically(path: &Path, replacements: &[CultCacheEnvelope]) -> Result<()> { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; - } - ensure!( - replacements.iter().enumerate().all(|(index, entry)| { - replacements[index + 1..].iter().all(|other| { - (entry.r#type.as_str(), entry.key.as_str()) - != (other.r#type.as_str(), other.key.as_str()) - }) - }), - "atomic projection set contains a duplicate identity" - ); - let store = SingleFileMessagePackBackingStore::new(path); - for _ in 0..8 { - let entries = store.pull_all_read_only_snapshot()?; - let conditions = replacements - .iter() - .map(|replacement| { - Ok(CultCacheExpectedEnvelope { - r#type: replacement.r#type.clone(), - key: replacement.key.clone(), - current: projection_entry(&entries, &replacement.r#type, &replacement.key)? - .cloned(), - }) - }) - .collect::>>()?; - if conditions - .iter() - .zip(replacements) - .all(|(condition, replacement)| condition.current.as_ref() == Some(replacement)) - { - return Ok(()); - } - if store.compare_exchange(&conditions, replacements)? { - return Ok(()); - } - } - bail!("CultCache projection changed too often to publish atomic set") -} - fn projection_entry<'a>( entries: &'a [CultCacheEnvelope], r#type: &str, @@ -3674,101 +3675,6 @@ fn projection_entry<'a>( Ok(current) } -fn insert_exact_process_lease( - path: &Path, - expected: &IdunnExpectedIncarnationRecord, - activation: &IdunnRuntimeActivationRecord, - replacement: CultCacheEnvelope, -) -> Result<()> { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; - } - let store = SingleFileMessagePackBackingStore::new(path); - for _ in 0..8 { - let entries = store.pull_all_read_only_snapshot()?; - let projected_expected = projection_entry( - &entries, - IdunnExpectedIncarnationRecord::TYPE, - &expected.target, - )? - .context("process write lease has no current Expected projection")?; - ensure!( - projected_expected.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) - && IdunnExpectedIncarnationRecord::decode_canonical(&projected_expected.payload)? - == *expected, - "process write lease current Expected projection is substituted" - ); - let projected_activation = projection_entry( - &entries, - IdunnRuntimeActivationRecord::TYPE, - &expected.target, - )? - .context("process write lease has no observed activation projection")?; - ensure!( - projected_activation.schema_id.as_deref() == Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA) - && IdunnRuntimeActivationRecord::decode_canonical(&projected_activation.payload)? - == *activation, - "process write lease observed activation projection is substituted" - ); - let projected_anchor = service_trust_anchor_from_envelope( - projection_entry( - &entries, - GameCultServiceTrustAnchorRecord::TYPE, - &runtime_presence_trust_anchor_id(&expected.target), - )? - .context("process write lease has no runtime presence trust anchor")?, - )?; - ensure!( - projected_anchor.service_id == expected.target - && projected_anchor.runtime_id == expected.runtime_id - && projected_anchor.signer_identity_id == expected.expected_signer_identity_id - && projected_anchor.signing_purpose - == GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE - && projected_anchor.signed_schema == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, - "process write lease runtime presence trust anchor is substituted" - ); - let current = projection_entry(&entries, &replacement.r#type, &replacement.key)?; - if current == Some(&replacement) { - return Ok(()); - } - ensure!( - current.is_none(), - "refusing to replace a different current process write-lease projection" - ); - let conditions = [ - CultCacheExpectedEnvelope { - r#type: IdunnExpectedIncarnationRecord::TYPE.into(), - key: expected.target.clone(), - current: Some(projected_expected.clone()), - }, - CultCacheExpectedEnvelope { - r#type: IdunnRuntimeActivationRecord::TYPE.into(), - key: expected.target.clone(), - current: Some(projected_activation.clone()), - }, - CultCacheExpectedEnvelope { - r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), - key: runtime_presence_trust_anchor_id(&expected.target), - current: projection_entry( - &entries, - GameCultServiceTrustAnchorRecord::TYPE, - &runtime_presence_trust_anchor_id(&expected.target), - )? - .cloned(), - }, - CultCacheExpectedEnvelope { - r#type: replacement.r#type.clone(), - key: replacement.key.clone(), - current: None, - }, - ]; - if store.compare_exchange(&conditions, std::slice::from_ref(&replacement))? { - return Ok(()); - } - } - bail!("CultCache projection changed too often to publish exact record") -} - impl TopologyPort for CultCacheTopologyDriver { fn publish_expected( &self, @@ -3777,30 +3683,52 @@ impl TopologyPort for CultCacheTopologyDriver { ) -> Result { expected.validate()?; let anchor = runtime_presence_trust_anchor(expected, provider_anchor)?; - replace_records_atomically( - &self.projection_store, - &[ - CultCacheEnvelope { - key: expected.target.clone(), - r#type: IdunnExpectedIncarnationRecord::TYPE.into(), - payload: expected.canonical_bytes()?, - stored_at: chrono::Utc::now().to_rfc3339(), - schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), - }, - CultCacheEnvelope { - key: anchor.trust_anchor_id.clone(), - r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), - payload: rmp_serde::to_vec(&anchor)?, - stored_at: rfc3339_millis(anchor.bound_at_unix_millis)?, - schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), - }, - ], - )?; - publish_projection_mode(&self.projection_store)?; + let key = incarnation_key(expected)?; + let target = expected.target.clone(); + self.mutate(|entries| { + let expected_record = expected_envelope(expected)?; + let anchor_record = anchor_envelope(&anchor)?; + let current_expected = + projection_entry(entries, IdunnExpectedIncarnationRecord::TYPE, &key)?; + let current_anchor = projection_entry( + entries, + GameCultServiceTrustAnchorRecord::TYPE, + &anchor.trust_anchor_id, + )?; + let legacy = entries + .iter() + .any(|envelope| Self::is_legacy_slot_record(envelope, &target)); + if !legacy + && current_expected.is_some_and(|current| same_record(current, &expected_record)) + && current_anchor.is_some_and(|current| same_record(current, &anchor_record)) + { + return Ok(None); + } + // Only the Expected and anchor records are replaced. An activation + // or lease already projected under this key is this incarnation's + // own runtime fact and is left exactly as published. + let mut replacement = entries + .iter() + .filter(|envelope| { + !((envelope.key == key + && envelope.r#type == IdunnExpectedIncarnationRecord::TYPE) + || (envelope.r#type == GameCultServiceTrustAnchorRecord::TYPE + && envelope.key == anchor.trust_anchor_id) + || Self::is_legacy_slot_record(envelope, &target)) + }) + .cloned() + .collect::>(); + replacement.push(match current_expected { + Some(current) if same_record(current, &expected_record) => current.clone(), + _ => expected_record, + }); + replacement.push(anchor_record); + Ok(Some(replacement)) + })?; expected.canonical_sha256() } - fn withdraw_expected( + fn withdraw_incarnation( &self, expected: &IdunnExpectedIncarnationRecord, provider_anchor: &ServiceIdentityTrustAnchor, @@ -3824,82 +3752,41 @@ impl TopologyPort for CultCacheTopologyDriver { lease, )?; } - if !self.projection_store.exists() { - return Ok(()); - } - let store = SingleFileMessagePackBackingStore::new(&self.projection_store); - for _ in 0..8 { - let entries = store.pull_all_read_only_snapshot()?; - let mut found_expected = false; - let mut found_anchor = false; - let mut found_activation = false; - let mut found_lease = false; - let mut retained = Vec::with_capacity(entries.len()); - for envelope in &entries { - if envelope.r#type == GameCultServiceTrustAnchorRecord::TYPE - && envelope.key == exact_anchor.trust_anchor_id - { - ensure!( - !found_anchor - && service_trust_anchor_from_envelope(envelope)? == exact_anchor, - "refusing to withdraw a substituted runtime presence trust anchor" - ); - found_anchor = true; - continue; - } - if envelope.key != expected.target { - retained.push(envelope.clone()); - continue; - } - if envelope.r#type == IdunnExpectedIncarnationRecord::TYPE { - ensure!( - !found_expected - && envelope.schema_id.as_deref() - == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) - && IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)? - == *expected, - "refusing to withdraw a substituted Expected projection" - ); - found_expected = true; - continue; - } - if envelope.r#type == IdunnRuntimeActivationRecord::TYPE { - let current = - IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload)?; - ensure!( - !found_activation - && envelope.schema_id.as_deref() - == Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA) - && activation == Some(¤t), - "refusing to withdraw a substituted runtime activation" - ); - found_activation = true; - continue; - } - if envelope.r#type == IdunnProcessWriteLeaseRecord::TYPE { - let current = - IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; - ensure!( - !found_lease - && envelope.schema_id.as_deref() - == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA) - && lease == Some(¤t), - "refusing to withdraw a substituted process write lease" - ); - found_lease = true; - continue; - } - retained.push(envelope.clone()); + let key = incarnation_key(expected)?; + self.mutate(|entries| { + let projected = ProjectedIncarnation::read(entries, expected, activation, lease)?; + let anchor_stays = Self::other_incarnations_of(entries, &expected.target, &key); + let current_anchor = projection_entry( + entries, + GameCultServiceTrustAnchorRecord::TYPE, + &exact_anchor.trust_anchor_id, + )?; + if let Some(envelope) = current_anchor { + ensure!( + service_trust_anchor_from_envelope(envelope)? == exact_anchor, + "refusing to withdraw a substituted runtime presence trust anchor" + ); } - if !found_expected && !found_anchor && !found_activation && !found_lease { - return Ok(()); + let nothing_to_remove = projected.expected.is_none() + && projected.activation.is_none() + && projected.lease.is_none() + && (anchor_stays || current_anchor.is_none()); + if nothing_to_remove { + return Ok(None); } - if store.compare_exchange_snapshot(&entries, &retained)? { - publish_projection_mode(&self.projection_store)?; - return Ok(()); - } - } - bail!("Expected projection changed repeatedly while withdrawing it") + Ok(Some( + entries + .iter() + .filter(|envelope| { + !(envelope.key == key + || (!anchor_stays + && envelope.r#type == GameCultServiceTrustAnchorRecord::TYPE + && envelope.key == exact_anchor.trust_anchor_id)) + }) + .cloned() + .collect(), + )) + }) } fn publish_observed_activation( @@ -3916,17 +3803,38 @@ impl TopologyPort for CultCacheTopologyDriver { && observation.executable_sha256 == expected.artifact_sha256, "observed activation does not name the Expected native process" ); - upsert_record( - &self.projection_store, - CultCacheEnvelope { - key: expected.target.clone(), - r#type: IdunnRuntimeActivationRecord::TYPE.into(), - payload: activation.canonical_bytes()?, - stored_at: rfc3339_millis(activation.issued_at_unix_millis)?, - schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), - }, - )?; - publish_projection_mode(&self.projection_store)?; + let key = incarnation_key(expected)?; + self.mutate(|entries| { + // The Expected must be this one. A prior activation under the key + // is replaced without comparison: every launch is issued a fresh + // one, and the observation just made is the authority on which is + // current. + let projected_expected = + projection_entry(entries, IdunnExpectedIncarnationRecord::TYPE, &key)? + .context("observed activation has no current Expected projection")?; + ensure!( + projected_expected.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA) + && IdunnExpectedIncarnationRecord::decode_canonical( + &projected_expected.payload + )? == *expected, + "observed activation's Expected projection is substituted" + ); + let record = activation_envelope(&key, activation)?; + if projection_entry(entries, IdunnRuntimeActivationRecord::TYPE, &key)? + .is_some_and(|current| same_record(current, &record)) + { + return Ok(None); + } + let mut replacement = entries + .iter() + .filter(|envelope| { + !(envelope.key == key && envelope.r#type == IdunnRuntimeActivationRecord::TYPE) + }) + .cloned() + .collect::>(); + replacement.push(record); + Ok(Some(replacement)) + })?; activation.canonical_sha256() } @@ -3937,19 +3845,45 @@ impl TopologyPort for CultCacheTopologyDriver { lease: &IdunnProcessWriteLeaseRecord, ) -> Result { validate_topology_lease(expected, activation, lease)?; - insert_exact_process_lease( - &self.projection_store, - expected, - activation, - CultCacheEnvelope { - key: expected.target.clone(), + let key = incarnation_key(expected)?; + let anchor_id = runtime_presence_trust_anchor_id(&expected.target); + self.mutate(|entries| { + let projected = + ProjectedIncarnation::read(entries, expected, Some(activation), Some(lease))?; + ensure!( + projected.expected.is_some(), + "process write lease has no current Expected projection" + ); + ensure!( + projected.activation.is_some(), + "process write lease has no observed activation projection" + ); + let projected_anchor = service_trust_anchor_from_envelope( + projection_entry(entries, GameCultServiceTrustAnchorRecord::TYPE, &anchor_id)? + .context("process write lease has no runtime presence trust anchor")?, + )?; + ensure!( + projected_anchor.service_id == expected.target + && projected_anchor.runtime_id == expected.runtime_id + && projected_anchor.signer_identity_id == expected.expected_signer_identity_id + && projected_anchor.signing_purpose + == GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE + && projected_anchor.signed_schema == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + "process write lease runtime presence trust anchor is substituted" + ); + if projected.lease.is_some() { + return Ok(None); + } + let mut replacement = entries.to_vec(); + replacement.push(CultCacheEnvelope { + key: key.clone(), r#type: IdunnProcessWriteLeaseRecord::TYPE.into(), payload: lease.canonical_bytes()?, stored_at: rfc3339_millis(lease.issued_at_unix_millis)?, schema_id: Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into()), - }, - )?; - publish_projection_mode(&self.projection_store)?; + }); + Ok(Some(replacement)) + })?; lease.canonical_sha256() } @@ -3964,50 +3898,42 @@ impl TopologyPort for CultCacheTopologyDriver { if let Some(lease) = lease { validate_topology_lease(expected, activation, lease)?; } - if !self.projection_store.exists() { - return Ok(()); - } - let store = SingleFileMessagePackBackingStore::new(&self.projection_store); - for _ in 0..8 { - let entries = store.pull_all_read_only_snapshot()?; - let mut found = false; - let mut retained = Vec::with_capacity(entries.len()); - for envelope in &entries { - if envelope.key != expected.target - || envelope.r#type != IdunnProcessWriteLeaseRecord::TYPE - { - retained.push(envelope.clone()); - continue; - } - let current = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; - ensure!( - !found - && envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA) - && lease == Some(¤t), - "refusing to withdraw an unexpected process write-lease projection" - ); - found = true; - } - if !found { - return Ok(()); - } - if store.compare_exchange_snapshot(&entries, &retained)? { - publish_projection_mode(&self.projection_store)?; - return Ok(()); - } - } - bail!("process write-lease projection changed repeatedly while withdrawing it") + let key = incarnation_key(expected)?; + self.mutate(|entries| { + let current = projection_entry(entries, IdunnProcessWriteLeaseRecord::TYPE, &key)?; + let Some(envelope) = current else { + return Ok(None); + }; + let projected = IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload)?; + ensure!( + envelope.schema_id.as_deref() == Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA) + && lease == Some(&projected), + "refusing to withdraw an unexpected process write-lease projection" + ); + Ok(Some( + entries + .iter() + .filter(|candidate| *candidate != envelope) + .cloned() + .collect(), + )) + }) } - fn receive(&self, target: &str) -> Result> { + fn receive( + &self, + target: &str, + expected_sha256: &str, + ) -> Result> { require_driver_id(target, "topology target")?; if !self.correlation_store.exists() { return Ok(None); } + let key = incarnation_key_of(target, expected_sha256); let entries = SingleFileMessagePackBackingStore::new(&self.correlation_store) .pull_all_read_only_snapshot()?; let mut matches = entries.iter().filter(|envelope| { - envelope.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE && envelope.key == target + envelope.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE && envelope.key == key }); let Some(envelope) = matches.next() else { return Ok(None); @@ -4022,6 +3948,7 @@ impl TopologyPort for CultCacheTopologyDriver { ); Ok(Some(ReceivedOdinTopologyCorrelation { target: target.to_owned(), + expected_sha256: expected_sha256.to_owned(), canonical_bytes: envelope.payload.clone(), })) } @@ -6008,7 +5935,10 @@ fn remove_exact_root_owned_file(path: &Path, mode: u32) -> Result<()> { /// `ReadOnlyPaths=` on the bundle does not help: it binds the leaf into the /// unit's namespace but grants no traversal on the path above it. #[cfg(unix)] -fn ensure_bundle_is_reachable_by_workload(bundle: &Path, state_group_id: Option) -> Result<()> { +fn ensure_bundle_is_reachable_by_workload( + bundle: &Path, + state_group_id: Option, +) -> Result<()> { use std::os::unix::fs::{MetadataExt, PermissionsExt}; for ancestor in bundle.ancestors().skip(1) { @@ -6020,8 +5950,7 @@ fn ensure_bundle_is_reachable_by_workload(bundle: &Path, state_group_id: Option< let mode = metadata.permissions().mode(); // The workload's uid is allocated by DynamicUser and owns nothing here, // so traversal can only come from the state group or from other. - let by_group = - state_group_id == Some(metadata.gid()) && mode & 0o010 != 0; + let by_group = state_group_id == Some(metadata.gid()) && mode & 0o010 != 0; ensure!( mode & 0o001 != 0 || by_group, "{} is not traversable by the workload, so the runtime bundle cannot be read; give it o+x (0711 leaks no names) or group-own it by the state group with g+x", @@ -6032,7 +5961,10 @@ fn ensure_bundle_is_reachable_by_workload(bundle: &Path, state_group_id: Option< } #[cfg(not(unix))] -fn ensure_bundle_is_reachable_by_workload(_bundle: &Path, _state_group_id: Option) -> Result<()> { +fn ensure_bundle_is_reachable_by_workload( + _bundle: &Path, + _state_group_id: Option, +) -> Result<()> { Ok(()) } @@ -6081,8 +6013,14 @@ fn build_machine_id_file(workspace: &Path) -> Result { let id = build_machine_id(workspace)?; let path = root.join(&id); if !path.exists() { - fs::write(&path, format!("{id} -")).context("writing the build machine-id")?; + fs::write( + &path, + format!( + "{id} +" + ), + ) + .context("writing the build machine-id")?; fs::set_permissions(&path, std::os::unix::fs::PermissionsExt::from_mode(0o444)) .context("sealing the build machine-id")?; } @@ -6718,7 +6656,9 @@ mod tests { let observation = audit_workload_observation("no"); let show = |load: &str, active: &str, sub: &str| -> String { - format!("LoadState={load}\nActiveState={active}\nSubState={sub}\nInvocationID=inv\nDescription=Idunn test\n") + format!( + "LoadState={load}\nActiveState={active}\nSubState={sub}\nInvocationID=inv\nDescription=Idunn test\n" + ) }; for (active, sub) in [ ("active", "running"), @@ -6737,11 +6677,17 @@ mod tests { } std::fs::write(&state, show("loaded", "failed", "failed"))?; - assert!(driver.is_permanently_stopped(&observation)?, "failed is permanent under Restart=no"); + assert!( + driver.is_permanently_stopped(&observation)?, + "failed is permanent under Restart=no" + ); // systemd has no such unit: `systemctl show` exits 0 with not-found. std::fs::write(&state, show("not-found", "inactive", "dead"))?; - assert!(driver.is_permanently_stopped(&observation)?, "a forgotten unit cannot restart"); + assert!( + driver.is_permanently_stopped(&observation)?, + "a forgotten unit cannot restart" + ); // A systemctl failure with no properties is an error, not "stopped". std::fs::write(&state, "")?; @@ -6843,7 +6789,10 @@ mod tests { std::fs::set_permissions(&lock, std::fs::Permissions::from_mode(0o640)).unwrap(); publish_projection_mode(&path).unwrap(); let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; - assert_eq!(mode, 0o644, "every target must be able to read the projection"); + assert_eq!( + mode, 0o644, + "every target must be able to read the projection" + ); let lock_mode = std::fs::metadata(&lock).unwrap().permissions().mode() & 0o777; assert_eq!(lock_mode, 0o644, "the lock is opened alongside the store"); } @@ -6859,16 +6808,19 @@ mod tests { std::fs::create_dir_all(&bundle).unwrap(); std::fs::set_permissions(&runtime_root, std::fs::Permissions::from_mode(0o700)).unwrap(); let error = ensure_bundle_is_reachable_by_workload(&bundle, None).unwrap_err(); - assert!(error.to_string().contains("not traversable by the workload")); + assert!( + error + .to_string() + .contains("not traversable by the workload") + ); std::fs::set_permissions(&runtime_root, std::fs::Permissions::from_mode(0o711)).unwrap(); ensure_bundle_is_reachable_by_workload(&bundle, None).unwrap(); // A 0750 root group-owned by the state group is the other correct // shape, and is what the write-lease hardening expects: it derives the // record's group from this directory. std::fs::set_permissions(&runtime_root, std::fs::Permissions::from_mode(0o750)).unwrap(); - let owning_group = std::os::unix::fs::MetadataExt::gid( - &std::fs::metadata(&runtime_root).unwrap(), - ); + let owning_group = + std::os::unix::fs::MetadataExt::gid(&std::fs::metadata(&runtime_root).unwrap()); ensure_bundle_is_reachable_by_workload(&bundle, Some(owning_group)).unwrap(); assert!(ensure_bundle_is_reachable_by_workload(&bundle, Some(owning_group + 1)).is_err()); } @@ -6878,7 +6830,11 @@ mod tests { let first = build_machine_id(Path::new("/var/lib/gamecult/idunn/staging/tx-a")).unwrap(); let second = build_machine_id(Path::new("/var/lib/gamecult/idunn/staging/tx-b")).unwrap(); assert_eq!(first.len(), 32); - assert!(first.chars().all(|c| c.is_ascii_hexdigit() && !c.is_uppercase())); + assert!( + first + .chars() + .all(|c| c.is_ascii_hexdigit() && !c.is_uppercase()) + ); assert_ne!(first, second); assert_eq!( first, @@ -7098,7 +7054,7 @@ mod tests { upsert_record( &driver.projection_store, CultCacheEnvelope { - key: expected.target.clone(), + key: incarnation_key(expected)?, r#type: IdunnRuntimeActivationRecord::TYPE.into(), payload: activation.canonical_bytes()?, stored_at: rfc3339_millis(activation.issued_at_unix_millis)?, @@ -7295,7 +7251,7 @@ mod tests { upsert_record( &correlation_store, CultCacheEnvelope { - key: "service".into(), + key: incarnation_key_of("service", &digest('1')), r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), payload: opaque.clone(), stored_at: "2026-09-03T00:00:00Z".into(), @@ -7307,10 +7263,12 @@ mod tests { correlation_store, }; + assert_eq!(driver.receive("service", &digest('2'))?, None); assert_eq!( - driver.receive("service")?, + driver.receive("service", &digest('1'))?, Some(ReceivedOdinTopologyCorrelation { target: "service".into(), + expected_sha256: digest('1'), canonical_bytes: opaque, }) ); @@ -7328,7 +7286,8 @@ mod tests { .pull_all_read_only_snapshot()?; assert_eq!(published.len(), 2); assert!(published.iter().any(|entry| { - entry.key == expected.target && entry.r#type == IdunnExpectedIncarnationRecord::TYPE + entry.key == incarnation_key(&expected).unwrap() + && entry.r#type == IdunnExpectedIncarnationRecord::TYPE })); let published_anchor = published .iter() @@ -7346,7 +7305,7 @@ mod tests { upsert_record( &driver.projection_store, CultCacheEnvelope { - key: expected.target.clone(), + key: incarnation_key(&expected)?, r#type: IdunnExpectedIncarnationRecord::TYPE.into(), payload: substituted_expected.canonical_bytes()?, stored_at: "2026-09-03T00:00:00Z".into(), @@ -7355,7 +7314,7 @@ mod tests { )?; assert!( driver - .withdraw_expected(&expected, &provider_anchor, Some(&activation), Some(&lease)) + .withdraw_incarnation(&expected, &provider_anchor, Some(&activation), Some(&lease)) .is_err() ); driver.publish_expected(&expected, &provider_anchor)?; @@ -7375,7 +7334,7 @@ mod tests { )?; assert!( driver - .withdraw_expected(&expected, &provider_anchor, Some(&activation), Some(&lease)) + .withdraw_incarnation(&expected, &provider_anchor, Some(&activation), Some(&lease)) .is_err() ); driver.publish_expected(&expected, &provider_anchor)?; @@ -7385,7 +7344,7 @@ mod tests { upsert_record( &driver.projection_store, CultCacheEnvelope { - key: expected.target.clone(), + key: incarnation_key(&expected)?, r#type: IdunnRuntimeActivationRecord::TYPE.into(), payload: substituted_activation.canonical_bytes()?, stored_at: "2026-09-03T00:00:00Z".into(), @@ -7394,7 +7353,7 @@ mod tests { )?; assert!( driver - .withdraw_expected(&expected, &provider_anchor, Some(&activation), None) + .withdraw_incarnation(&expected, &provider_anchor, Some(&activation), None) .is_err() ); project_activation(&driver, &expected, &activation)?; @@ -7408,7 +7367,7 @@ mod tests { ); assert!( driver - .withdraw_expected( + .withdraw_incarnation( &expected, &provider_anchor, Some(&activation), @@ -7427,12 +7386,22 @@ mod tests { schema_id: Some("test.unrelated.v1".into()), }, )?; - driver.withdraw_expected(&expected, &provider_anchor, Some(&activation), Some(&lease))?; + driver.withdraw_incarnation( + &expected, + &provider_anchor, + Some(&activation), + Some(&lease), + )?; let remaining = SingleFileMessagePackBackingStore::new(&driver.projection_store) .pull_all_read_only_snapshot()?; assert_eq!(remaining.len(), 1); assert_eq!(remaining[0].key, "unrelated"); - driver.withdraw_expected(&expected, &provider_anchor, Some(&activation), Some(&lease))?; + driver.withdraw_incarnation( + &expected, + &provider_anchor, + Some(&activation), + Some(&lease), + )?; Ok(()) } @@ -7446,6 +7415,7 @@ mod tests { driver.publish_expected(&incumbent, &provider_anchor)?; project_activation(&driver, &incumbent, &activation)?; driver.publish_process_write_lease(&incumbent, &activation, &lease)?; + let incumbent_key = incarnation_key(&incumbent)?; let mut candidate = incumbent.clone(); candidate.plan_id = digest('a'); @@ -7455,8 +7425,35 @@ mod tests { let mut candidate_activation = activation.clone(); candidate_activation.expected_projection_sha256 = candidate.canonical_sha256()?; candidate_activation.runtime_instance_id = digest('d'); + let candidate_key = incarnation_key(&candidate)?; + + // Sealing the candidate beside the incumbent changes nothing the + // incumbent owns: its Expected, activation and lease are still there + // under its own key, byte for byte. + let before_candidate = SingleFileMessagePackBackingStore::new(&driver.projection_store) + .pull_all_read_only_snapshot()?; driver.publish_expected(&candidate, &provider_anchor)?; project_activation(&driver, &candidate, &candidate_activation)?; + let with_candidate = SingleFileMessagePackBackingStore::new(&driver.projection_store) + .pull_all_read_only_snapshot()?; + for envelope in &before_candidate { + if envelope.key == incumbent_key { + assert!(with_candidate.contains(envelope)); + } + } + assert!(driver.admitted_runtime_projection_is_exact( + &incumbent, + &provider_anchor, + &activation, + Some(&lease), + )?); + assert_eq!( + with_candidate + .iter() + .filter(|entry| entry.r#type == IdunnExpectedIncarnationRecord::TYPE) + .count(), + 2 + ); upsert_record( &driver.projection_store, @@ -7469,11 +7466,36 @@ mod tests { }, )?; + // A pre-fencing abort withdraws the candidate and only the candidate. + driver.withdraw_incarnation( + &candidate, + &provider_anchor, + Some(&candidate_activation), + None, + )?; + let after_withdrawal = SingleFileMessagePackBackingStore::new(&driver.projection_store) + .pull_all_read_only_snapshot()?; + assert!( + after_withdrawal + .iter() + .all(|entry| entry.key != candidate_key) + ); + assert!(driver.admitted_runtime_projection_is_exact( + &incumbent, + &provider_anchor, + &activation, + Some(&lease), + )?); + assert!( + after_withdrawal + .iter() + .any(|entry| entry.r#type == GameCultServiceTrustAnchorRecord::TYPE) + ); + + // Demotion is for an incumbent whose process is gone: activation and + // lease go, Expected and anchor stay. assert_eq!( - driver.restore_admitted_expected_only( - &candidate, - &provider_anchor, - Some(&candidate_activation), + driver.demote_to_expected_only( &incumbent, &provider_anchor, &activation, @@ -7481,7 +7503,6 @@ mod tests { )?, incumbent.canonical_sha256()? ); - let restored = SingleFileMessagePackBackingStore::new(&driver.projection_store) .pull_all_read_only_snapshot()?; assert_eq!(restored.len(), 3); @@ -7515,18 +7536,13 @@ mod tests { .iter() .any(|entry| entry.key == "unrelated" && entry.r#type == "test.unrelated") ); + assert!(!driver.projected_activation_is_present(&incumbent)?); let idempotent_bytes = fs::read(&driver.projection_store)?; - driver.restore_admitted_expected_only( - &candidate, - &provider_anchor, - Some(&candidate_activation), - &incumbent, - &provider_anchor, - &activation, - Some(&lease), - )?; + driver.demote_to_expected_only(&incumbent, &provider_anchor, &activation, Some(&lease))?; assert_eq!(fs::read(&driver.projection_store)?, idempotent_bytes); + // A record under the incumbent's key that is not the incumbent's is a + // substitution; demotion refuses it and leaves the store as found. let admitted_anchor = runtime_presence_trust_anchor(&incumbent, &provider_anchor)?; let mut unknown_expected = incumbent.clone(); unknown_expected.incarnation_id = "unknown-incarnation".into(); @@ -7539,7 +7555,7 @@ mod tests { unknown_lease.lease_epoch += 1; let substitutions = [ CultCacheEnvelope { - key: incumbent.target.clone(), + key: incumbent_key.clone(), r#type: IdunnExpectedIncarnationRecord::TYPE.into(), payload: unknown_expected.canonical_bytes()?, stored_at: "2026-09-03T00:00:00Z".into(), @@ -7553,14 +7569,14 @@ mod tests { schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), }, CultCacheEnvelope { - key: incumbent.target.clone(), + key: incumbent_key.clone(), r#type: IdunnRuntimeActivationRecord::TYPE.into(), payload: unknown_activation.canonical_bytes()?, stored_at: "2026-09-03T00:00:00Z".into(), schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), }, CultCacheEnvelope { - key: incumbent.target.clone(), + key: incumbent_key.clone(), r#type: IdunnProcessWriteLeaseRecord::TYPE.into(), payload: unknown_lease.canonical_bytes()?, stored_at: "2026-09-03T00:00:00Z".into(), @@ -7573,10 +7589,7 @@ mod tests { let before = fs::read(&driver.projection_store)?; assert!( driver - .restore_admitted_expected_only( - &candidate, - &provider_anchor, - Some(&candidate_activation), + .demote_to_expected_only( &incumbent, &provider_anchor, &activation, @@ -7589,6 +7602,44 @@ mod tests { Ok(()) } + /// The single-slot projection an older Idunn wrote is retired the first + /// time this one publishes for the target, and never read. + #[test] + fn a_legacy_target_keyed_slot_is_retired_on_publish() -> Result<()> { + let temp = tempfile::tempdir()?; + let (expected, activation, _warming, provider_anchor) = authenticated_warming(temp.path())?; + let driver = topology_driver(temp.path(), "legacy"); + upsert_record( + &driver.projection_store, + CultCacheEnvelope { + key: expected.target.clone(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: expected.canonical_bytes()?, + stored_at: "2026-09-03T00:00:00Z".into(), + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }, + )?; + upsert_record( + &driver.projection_store, + CultCacheEnvelope { + key: expected.target.clone(), + r#type: IdunnRuntimeActivationRecord::TYPE.into(), + payload: activation.canonical_bytes()?, + stored_at: "2026-09-03T00:00:00Z".into(), + schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), + }, + )?; + assert!(!driver.admitted_expected_projection_is_exact(&expected, &provider_anchor)?); + assert!(!driver.projected_activation_is_present(&expected)?); + driver.publish_expected(&expected, &provider_anchor)?; + let published = SingleFileMessagePackBackingStore::new(&driver.projection_store) + .pull_all_read_only_snapshot()?; + assert!(published.iter().all(|entry| entry.key != expected.target)); + assert_eq!(published.len(), 2); + assert!(driver.admitted_expected_projection_is_exact(&expected, &provider_anchor)?); + Ok(()) + } + #[cfg(unix)] #[test] fn service_credential_sources_require_one_root_owned_0400_inode() -> Result<()> { diff --git a/src/provisioning.rs b/src/provisioning.rs index 8ae58a9..aa8d067 100644 --- a/src/provisioning.rs +++ b/src/provisioning.rs @@ -8,8 +8,8 @@ use cultnet_rs::{ IDUNN_AUTHENTICATED_PROVIDER_HEALTH_PROJECTION_SIGNING_PURPOSE, IDUNN_DAEMON_HEALTH_TRUST_BINDING_SCHEMA, IDUNN_DEPLOYMENT_BRAKE_AUTHORITY, IDUNN_DEPLOYMENT_BRAKE_ID, IDUNN_DEPLOYMENT_BRAKE_SCHEMA, IDUNN_DEPLOYMENT_BRAKE_SCOPE, - IDUNN_DEPLOYMENT_RELEASE_PURPOSE, IDUNN_LIFECYCLE_BRAKE_AUTHORITY, IDUNN_LIFECYCLE_BRAKE_SCHEMA, - IDUNN_LIFECYCLE_BRAKE_SCOPE, IdunnDaemonHealthTrustBindingRecord, + IDUNN_DEPLOYMENT_RELEASE_PURPOSE, IDUNN_LIFECYCLE_BRAKE_AUTHORITY, + IDUNN_LIFECYCLE_BRAKE_SCHEMA, IDUNN_LIFECYCLE_BRAKE_SCOPE, IdunnDaemonHealthTrustBindingRecord, IdunnDeploymentBrakeObservation, IdunnDeploymentBrakeOperatorIdentity, IdunnDeploymentBrakeRecord, IdunnDeploymentBrakeReleasePurpose, IdunnLifecycleBrakeObservation, IdunnLifecycleBrakeRecord, IdunnServiceIdentity, OdinTopologyIdentity, ServiceIdentityProfile,