git.delta.rocks / jrsonnet / refs/commits / 69498f520d8e

--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1149,7 +1149,6 @@
  "base64 0.22.1",
  "serde",
  "unicode_categories",
- "z85",
 ]
 
 [[package]]
@@ -2089,6 +2088,7 @@
  "serde_json",
  "test-log",
  "thiserror 2.0.17",
+ "tokio",
  "tracing",
  "tracing-indicatif",
  "vte 0.15.0",
@@ -4638,12 +4638,6 @@
  "syn",
  "synstructure",
 ]
-
-[[package]]
-name = "z85"
-version = "3.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b3a41ce106832b4da1c065baa4c31cf640cf965fa1483816402b7f6b96f0a64"
 
 [[package]]
 name = "zerocopy"

--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -7,8 +7,9 @@
 	host::{Config, DeployKind, GenerationStorage},
 	opts::FleetOpts,
 };
+use futures::{StreamExt as _, stream::FuturesUnordered};
 use nix_eval::nix_go;
-use tokio::task::{LocalSet, spawn_blocking};
+use tokio::task::spawn_blocking;
 use tracing::{Instrument, error, field, info, info_span, warn};
 
 #[derive(Parser)]
@@ -47,7 +48,7 @@
 				"--profile",
 				format!(
 					"/nix/var/nix/profiles/{}-{hostname}",
-					config.data().gc_root_prefix
+					config.data.gc_root_prefix
 				),
 			)
 			.arg(&out_output);
@@ -60,14 +61,14 @@
 impl BuildSystems {
 	pub async fn run(self, config: &Config, opts: &FleetOpts) -> Result<()> {
 		let hosts = opts.filter_skipped(config.list_hosts()?)?;
-		let set = LocalSet::new();
+		let mut tasks = FuturesUnordered::new();
 		let build_attr = self.build_attr.clone();
 		for host in hosts {
 			let config = config.clone();
 			let span = info_span!("build", host = field::display(&host.name));
 			let hostname = host.name;
 			let build_attr = build_attr.clone();
-			set.spawn_local(
+			tasks.push(
 				(async move {
 					let built = match build_task(config, hostname.clone(), &build_attr).await {
 						Ok(path) => path,
@@ -88,7 +89,7 @@
 				.instrument(span),
 			);
 		}
-		set.await;
+		for _task in tasks.next().await {}
 		Ok(())
 	}
 }
@@ -96,7 +97,7 @@
 impl Deploy {
 	pub async fn run(self, config: &Config, opts: &FleetOpts) -> Result<()> {
 		let hosts = opts.filter_skipped(config.list_hosts()?)?;
-		let set = LocalSet::new();
+		let mut tasks = FuturesUnordered::new();
 		for host in hosts.into_iter() {
 			let config = config.clone();
 			let span = info_span!("deploy", host = field::display(&host.name));
@@ -112,7 +113,7 @@
 				host.set_legacy_ssh_store(legacy);
 			};
 
-			set.spawn_local(
+			tasks.push(
 				(async move {
 					let built = match build_task(config.clone(), hostname.clone(), "toplevel-fleet")
 						.await
@@ -170,7 +171,7 @@
 				.instrument(span),
 			);
 		}
-		set.await;
+		for _task in tasks.next().await {}
 		Ok(())
 	}
 }

--- a/cmds/fleet/src/cmds/rollback.rs
+++ b/cmds/fleet/src/cmds/rollback.rs
@@ -56,7 +56,7 @@
 		.collect::<HashSet<_>>();
 	let mut stored_locally = config
 		.local_host()
-		.list_generations(&format!("{}-{}", config.data().gc_root_prefix, host.name))
+		.list_generations(&format!("{}-{}", config.data.gc_root_prefix, host.name))
 		.await
 		.inspect_err(|e| {
 			warn!("failed to list generations available locally: {e}");

--- a/cmds/fleet/src/cmds/secrets/mod.rs
+++ b/cmds/fleet/src/cmds/secrets/mod.rs
@@ -412,7 +412,7 @@
 					if opts.should_skip(&host)? {
 						continue;
 					}
-					config.key(&host.name).await?;
+					config.host_key(&host.name).await?;
 				}
 			}
 			Secret::Read {
@@ -421,6 +421,7 @@
 				part: part_name,
 				mut prefer_identities,
 			} => {
+				/*
 				let Some(secret) = config.shared_secret(&name) else {
 					bail!("secret doesn't exists");
 				};
@@ -460,6 +461,8 @@
 					part.raw.data.clone()
 				};
 				stdout().write_all(&data)?;
+				*/
+				todo!()
 			}
 			Secret::Regenerate {
 				prefer_identities,
@@ -605,6 +608,7 @@
 				todo!()
 			}
 			Secret::List {} => {
+				/*
 				let _span = info_span!("loading secrets").entered();
 				let configured = config.list_configured_shared()?;
 				#[derive(Tabled)]
@@ -638,6 +642,8 @@
 					*/
 				}
 				// info!("loaded\n{}", Table::new(table).to_string())
+				*/
+				todo!()
 			}
 			Secret::Edit {
 				name,
@@ -645,7 +651,7 @@
 				part,
 				add,
 			} => {
-				let secret = config
+				/*let secret = config
 					.host_secret(&machine, &name)
 					.context("secret not found")?;
 				if let Some(data) = secret.secret.parts.get(&part) {
@@ -656,7 +662,8 @@
 					String::new()
 				} else {
 					bail!("part {part} not found in secret {name}. Did you mean to `--add` it?");
-				};
+				};*/
+				todo!()
 			}
 		}
 		Ok(())

--- a/cmds/fleet/src/cmds/tf.rs
+++ b/cmds/fleet/src/cmds/tf.rs
@@ -72,9 +72,9 @@
 			let tf_data: TfData = serde_json::from_slice(&data.stdout)
 				.context("failed to parse terraform fleet output")?;
 
-			let mut data = config.data();
 			debug!("synchronized done = {tf_data:?}");
-			data.extra.insert(
+			let mut extra = config.data.extra.write().expect("no poisoning");
+			extra.insert(
 				"terraformHosts".to_owned(),
 				serde_json::to_value(tf_data.hosts).expect("should be valid extra"),
 			);

--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -4,7 +4,7 @@
 // pub(crate) mod command;
 pub(crate) mod extra_args;
 
-use std::{env, ffi::OsString, process::ExitCode};
+use std::{env, ffi::OsString, process::ExitCode, sync::Arc};
 
 use anyhow::{Result, bail};
 use clap::{CommandFactory, Parser};
@@ -23,7 +23,9 @@
 use human_repr::HumanCount;
 #[cfg(feature = "indicatif")]
 use indicatif::{ProgressState, ProgressStyle};
-use nix_eval::{gc_register_my_thread, gc_unregister_my_thread, init_libraries};
+use nix_eval::{
+	gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
+};
 use tracing::{Instrument, error, info, info_span};
 #[cfg(feature = "indicatif")]
 use tracing_indicatif::IndicatifLayer;
@@ -39,9 +41,9 @@
 			info!("nothing to prefetch: no prefetch directory");
 			return Ok(());
 		}
-		let tasks = <FuturesUnordered<LocalBoxFuture<Result<()>>>>::new();
+		let tasks = FuturesUnordered::new();
 		for entry in std::fs::read_dir(&prefetch_dir)? {
-			tasks.push(Box::pin(async {
+			tasks.push(async {
 				let entry = entry?;
 				if !entry.metadata()?.is_file() {
 					bail!("only files should exist in prefetch directory");
@@ -59,7 +61,7 @@
 				status.arg("store").arg("prefetch-file").arg(path);
 				status.run_nix_string().instrument(span).await?;
 				Ok(())
-			}));
+			});
 		}
 		tasks.try_collect::<Vec<()>>().await?;
 		Ok(())
@@ -190,7 +192,7 @@
 
 	init_libraries();
 
-	tokio::runtime::Builder::new_multi_thread()
+	let runtime = tokio::runtime::Builder::new_multi_thread()
 		.enable_all()
 		.on_thread_start(|| {
 			gc_register_my_thread();
@@ -199,8 +201,13 @@
 			gc_unregister_my_thread();
 		})
 		.build()
-		.expect("failed to build runtime")
-		.block_on(async {
+		.expect("failed to build runtime");
+	let runtime = Arc::new(runtime);
+
+	init_tokio_for_nix(runtime.clone());
+
+	runtime.block_on(async {
+		tokio::task::spawn(async move {
 			if let Err(e) = main_real(opts).await {
 				error!("{e:#}");
 				ExitCode::FAILURE
@@ -208,6 +215,9 @@
 				ExitCode::SUCCESS
 			}
 		})
+		.await
+		.expect("primary task panicked")
+	})
 	// async_main(opts)
 }
 

--- a/crates/fleet-base/src/command.rs
+++ b/crates/fleet-base/src/command.rs
@@ -334,7 +334,7 @@
 	let mut stderr = child.stderr.take().unwrap();
 	let stdout = child.stdout.take().unwrap();
 	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin>> = Some(Box::new(stdout));
+	let mut out: Option<Box<dyn AsyncRead + Unpin + Send>> = Some(Box::new(stdout));
 	let mut ob = want_stdout
 		.then(|| out.take().unwrap())
 		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
@@ -397,7 +397,7 @@
 	let mut stderr = child.stderr().take().unwrap();
 	let stdout = child.stdout().take().unwrap();
 	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin>> = Some(Box::new(stdout));
+	let mut out: Option<Box<dyn AsyncRead + Unpin + Send>> = Some(Box::new(stdout));
 	let mut ob = want_stdout
 		.then(|| out.take().unwrap())
 		.unwrap_or_else(|| Box::new(EmptyAsyncRead));

--- a/crates/fleet-base/src/deploy.rs
+++ b/crates/fleet-base/src/deploy.rs
@@ -78,7 +78,7 @@
 	// unit name conflict in systemd-run
 	// This code is tied to rollback.nix
 	if !disable_rollback && action.should_create_rollback_marker() {
-		let _span = info_span!("preparing").entered();
+		// let _span = info_span!("preparing").entered();
 		info!("preparing for rollback");
 		let generation = get_current_generation(host).await?;
 		info!(
@@ -179,7 +179,7 @@
 		// FIXME: Connection might be disconnected after activation run
 
 		if action.should_activate() && !failed {
-			let _span = info_span!("activating").entered();
+			// let _span = info_span!("activating").entered();
 			info!("executing activation script");
 			let specialised = if let Some(specialisation) = specialisation {
 				let mut specialised = built.join("specialisation");

--- a/crates/fleet-base/src/fleetdata.rs
+++ b/crates/fleet-base/src/fleetdata.rs
@@ -1,10 +1,12 @@
 use std::{
+	cmp::Ordering,
 	collections::{
 		BTreeMap, BTreeSet,
 		btree_map::{self, Entry},
 	},
+	fmt,
 	io::{self, Cursor},
-	ops::Deref,
+	sync::RwLock,
 };
 
 use age::Recipient;
@@ -77,19 +79,18 @@
 	pub manager_keys: Vec<ManagerKey>,
 
 	#[serde(default)]
-	pub hosts: BTreeMap<String, HostData>,
+	pub hosts: RwLock<BTreeMap<String, HostData>>,
 
 	#[serde(default, alias = "shared_secrets")]
-	pub secrets: FleetSecrets,
+	pub secrets: RwLock<FleetSecrets>,
 
 	// extra_name => anything
 	#[serde(default)]
-	#[serde(skip_serializing_if = "BTreeMap::is_empty")]
-	pub extra: BTreeMap<String, Value>,
+	pub extra: RwLock<BTreeMap<String, Value>>,
 
 	#[serde(default)]
-	#[serde(skip_serializing_if = "BTreeMap::is_empty")]
-	host_secrets: BTreeMap<String, BTreeMap<String, FleetSecretDistribution>>,
+	#[serde(skip_serializing)]
+	host_secrets: BTreeMap<SecretOwner, BTreeMap<String, FleetSecretDistribution>>,
 }
 impl FleetData {
 	pub fn from_str(s: &str) -> anyhow::Result<Self> {
@@ -97,6 +98,8 @@
 		if !data.host_secrets.is_empty() {
 			info!("migrating host secrets into shared secrets structure");
 			data.secrets
+				.write()
+				.expect("no poisoning")
 				.merge_from_hosts(std::mem::take(&mut data.host_secrets));
 		}
 		Ok(data)
@@ -130,128 +133,431 @@
 #[serde(rename_all = "camelCase")]
 #[must_use]
 pub struct FleetSecretData {
-	#[serde(default = "Utc::now")]
 	pub created_at: DateTime<Utc>,
-	#[serde(default)]
-	#[serde(skip_serializing_if = "Option::is_none", alias = "expire_at")]
+	#[serde(default, skip_serializing_if = "Option::is_none", alias = "expire_at")]
 	pub expires_at: Option<DateTime<Utc>>,
 
 	#[serde(flatten)]
 	pub parts: BTreeMap<String, FleetSecretPart>,
 
-	#[serde(default)]
-	#[serde(skip_serializing_if = "Value::is_null")]
+	#[serde(default, skip_serializing_if = "Value::is_null")]
 	pub generation_data: Value,
 }
 
+fn is_false(b: &bool) -> bool {
+	*b == false
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialOrd, Ord, PartialEq, Eq)]
+#[repr(transparent)]
+pub struct SecretOwner(String);
+
+impl fmt::Display for SecretOwner {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		write!(f, "host:{}", self.0)
+	}
+}
+
+impl SecretOwner {
+	pub fn host(s: impl AsRef<str>) -> SecretOwner {
+		SecretOwner(s.as_ref().to_owned())
+	}
+	pub fn as_host(&self) -> Option<&str> {
+		Some(&self.0)
+	}
+}
+
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(rename_all = "camelCase")]
 #[must_use]
 pub struct FleetSecretDistribution {
 	#[serde(default)]
-	pub owners: BTreeSet<String>,
+	owners: BTreeSet<SecretOwner>,
+	#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+	owners_pending_prune: BTreeMap<SecretOwner, String>,
+
 	#[serde(flatten)]
 	pub secret: FleetSecretData,
 
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pending_prune: Option<String>,
 	#[serde(default, skip_serializing, alias = "managed")]
-	pub _deprecated_managed: bool,
+	_deprecated_managed: bool,
+}
+
+const EMPTY_PENDING_PRUNE: &BTreeMap<SecretOwner, String> = &BTreeMap::new();
+impl FleetSecretDistribution {
+	pub fn new(owners: BTreeSet<SecretOwner>, secret: FleetSecretData, now: DateTime<Utc>) -> Self {
+		assert!(
+			!owners.is_empty(),
+			"distribution should have at least one owner"
+		);
+		if let Some(expires_at) = &secret.expires_at {
+			assert!(
+				*expires_at > now,
+				"secret should not be expired on creation"
+			);
+		}
+		Self {
+			owners,
+			secret,
+			owners_pending_prune: BTreeMap::new(),
+			pending_prune: None,
+			_deprecated_managed: true,
+		}
+	}
+
+	fn owners_ex(&self, including_pruned: bool) -> impl Iterator<Item = &SecretOwner> {
+		let pending_prune = if including_pruned {
+			&self.owners_pending_prune
+		} else {
+			EMPTY_PENDING_PRUNE
+		};
+		self.owners.iter().chain(pending_prune.keys())
+	}
+	pub fn owners(&self) -> impl Iterator<Item = &SecretOwner> {
+		self.owners_ex(false)
+	}
+
+	pub fn prune(&mut self, reason: String) {
+		assert!(
+			self.pending_prune.is_none(),
+			"it shouldn't be possible to prune the same distribution twice using public api"
+		);
+		self.pending_prune = Some(reason);
+	}
+	pub fn prune_owners(&mut self, owners: &BTreeSet<SecretOwner>, reason: String) {
+		// if self.owners.iter().all(|o| owners.contains(o)) && self.owners_pending_prune.is_empty() {
+		// 	self.prune(format!("all owners were pruned: {reason}"));
+		// 	return;
+		// }
+		for owner in owners {
+			if self.owners.remove(owner) {
+				self.owners_pending_prune
+					.insert(owner.to_owned(), reason.clone());
+			}
+		}
+		// if self.owners.is_empty() {
+		// 	self.prune("no owners left".to_owned());
+		// }
+	}
+	pub fn unprune_owner(&mut self, owner: SecretOwner) {
+		if self.owners_pending_prune.remove(&owner).is_some() {
+			self.owners.insert(owner);
+		}
+	}
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug, Default)]
 #[must_use]
-pub struct FleetSecretDistributions(Vec<FleetSecretDistribution>);
+pub struct FleetSecretDistributions {
+	stored: Vec<FleetSecretDistribution>,
+}
 
-impl Deref for FleetSecretDistributions {
-	type Target = [FleetSecretDistribution];
-
-	fn deref(&self) -> &Self::Target {
-		self.0.as_slice()
+fn compare_dists(
+	a: &FleetSecretDistribution,
+	b: &FleetSecretDistribution,
+	prefer_identities: &BTreeSet<SecretOwner>,
+	include_pruned_owners: bool,
+) -> Ordering {
+	use Ordering::*;
+	if prefer_identities.is_empty() {
+		let a_has = a
+			.owners_ex(include_pruned_owners)
+			.any(|o| prefer_identities.contains(o));
+		let b_has = b
+			.owners_ex(include_pruned_owners)
+			.any(|o| prefer_identities.contains(o));
+		match (a_has, b_has) {
+			(true, false) => return Greater,
+			(false, true) => return Less,
+			_ => {}
+		}
+	}
+	match (a.secret.expires_at, b.secret.expires_at) {
+		(None, Some(_)) => return Greater,
+		(Some(_), None) => return Less,
+		(Some(a), Some(b)) => {
+			// Later is better
+			return a.cmp(&b);
+		}
+		(None, None) => {}
 	}
+
+	// Which one is easier to access
+	return a.owners.len().cmp(&b.owners.len());
 }
 
 impl FleetSecretDistributions {
-	pub fn owners(&self) -> impl Iterator<Item = &String> {
-		self.0.iter().flat_map(|v| v.owners.iter())
+	/// Drop expired distributions
+	fn prune_expired(&mut self, now: DateTime<Utc>) {
+		for ele in self.distributions_mut() {
+			if let Some(expires_at) = ele.secret.expires_at {
+				if expires_at < now {
+					ele.prune(format!("expired during check at {now}"));
+				}
+			}
+		}
+	}
+	/// Perform all pruning relevant to shared secrets
+	/// Also see expected_owner_removed
+	pub fn prune_shared(
+		&mut self,
+		expected_owners: &BTreeSet<SecretOwner>,
+		unique: bool,
+		expected_parts: &BTreeMap<String, GeneratorPart>,
+		expected_generation_data: &Value,
+		regenerate_on_owner_removed: bool,
+		regenerate_on_owner_added: bool,
+		prefer_identities: &BTreeSet<SecretOwner>,
+		now: DateTime<Utc>,
+	) {
+		self.prune_expired(now);
+		self.prune_generation_data(expected_generation_data, None);
+		self.prune_missing_parts(expected_parts, None);
+
+		let current_owners = self.owners().cloned().collect::<BTreeSet<SecretOwner>>();
+
+		let mut to_add = expected_owners.difference(&current_owners);
+		if to_add.next().is_some() && unique && regenerate_on_owner_added {
+			for dist in self.distributions_mut() {
+				dist.prune(format!(
+					"owners missing, can't add new distribution, regeneration preferred"
+				));
+			}
+			return;
+		}
+
+		for to_remove in current_owners.difference(&expected_owners) {
+			self.entry(to_remove.clone()).remove(
+				regenerate_on_owner_removed,
+				"owner was removed from expected owners list, regenerate_on_owner_removed is set"
+					.to_string(),
+			);
+		}
+		if unique {
+			self.prune_nonunique(prefer_identities);
+		}
+	}
+	pub fn prune_host(
+		&mut self,
+		owner: SecretOwner,
+		expected_parts: &BTreeMap<String, GeneratorPart>,
+		expected_generation_data: &Value,
+		now: DateTime<Utc>,
+	) {
+		self.prune_expired(now);
+		self.prune_generation_data(expected_generation_data, Some(&owner));
+		// TODO: Owner-based pruning is warranted (e.g host no longer has secret defined)
+		self.prune_missing_parts(expected_parts, Some(&owner));
+	}
+	/// Position of best distributions as in iterator returned by distributions()
+	/// None if distributions not found
+	fn best_idx(
+		&self,
+		prefer_identities: &BTreeSet<SecretOwner>,
+		include_pruned_owners: bool,
+	) -> Option<usize> {
+		self.distributions()
+			.enumerate()
+			.max_by(|(_, a), (_, b)| {
+				compare_dists(&a, &b, prefer_identities, include_pruned_owners)
+			})
+			.map(|(p, _)| p)
+	}
+	/// Secret wants to be the same on all hosts, leave only one unpruned version of it
+	fn prune_nonunique(&mut self, prefer_identities: &BTreeSet<SecretOwner>) {
+		if self.distributions().next().is_none() {
+			return;
+		}
+		let best = self.best_idx(prefer_identities, false).expect("not empty");
+		for (i, dist) in self.distributions_mut().enumerate() {
+			if i != best {
+				dist.prune(
+					"secret wants to be the same on all hosts, only the best one was left"
+						.to_owned(),
+				);
+			}
+		}
+	}
+
+	pub fn try_unprune(&mut self, owner: SecretOwner) -> Option<&FleetSecretDistribution> {
+		assert!(self.get(&owner).is_none(), "secret is not pruned for host");
+		if let Some(dist) = self
+			.distributions_mut()
+			.find(|v| v.owners_pending_prune.contains_key(&owner))
+		{
+			dist.unprune_owner(owner);
+			Some(dist)
+		} else {
+			None
+		}
+	}
+
+	pub fn best_distribution_for_reencryption(
+		&mut self,
+		prefer_identities: &BTreeSet<SecretOwner>,
+	) -> Option<&mut FleetSecretDistribution> {
+		let best_idx = self.best_idx(prefer_identities, true)?;
+		self.distributions_mut().nth(best_idx)
+	}
+
+	fn prune_missing_parts(
+		&mut self,
+		expected_parts: &BTreeMap<String, GeneratorPart>,
+		filter_owner: Option<&SecretOwner>,
+	) {
+		'dist: for ele in self.distributions_mut() {
+			if let Some(filter_owner) = filter_owner {
+				if !ele.owners.contains(filter_owner) {
+					continue;
+				}
+				// Note: secret still can have multiple owners even if it is host-owned
+				// in this case we expect that all owners using the same generator, so we can prune distribution for all of them
+			}
+			for (name, part) in expected_parts {
+				let Some(stored_part) = ele.secret.parts.get(name) else {
+					ele.prune(format!("secret definition added new part: {name}"));
+					continue 'dist;
+				};
+				if part.encrypted != stored_part.raw.encrypted {
+					ele.prune(format!(
+						"secret definition now requires part to be {}",
+						if part.encrypted {
+							"encrypted"
+						} else {
+							"non-encrypted"
+						}
+					));
+					continue 'dist;
+				}
+			}
+		}
+	}
+	fn prune_generation_data(
+		&mut self,
+		expected_generation_data: &Value,
+		filter_owner: Option<&SecretOwner>,
+	) {
+		for ele in self.distributions_mut() {
+			if let Some(filter_owner) = filter_owner {
+				if !ele.owners.contains(filter_owner) {
+					continue;
+				}
+				// Note: secret still can have multiple owners even if it is host-owned
+				// in this case we expect that all owners using the same generator, so we can prune distribution for all of them
+			}
+			if ele.secret.generation_data != *expected_generation_data {
+				ele.prune(format!(
+					"expected generation data mismatch: {expected_generation_data:?}"
+				));
+			}
+		}
 	}
+
+	/// Prune all distributions with no unpruned owners.
+	/// For ease of reencryption where possible, it is only called on persistence, when in memory - pruned owners are kept and
+	/// can decrypt their secrets.
+	fn prune_dead(&mut self) {
+		for ele in self.distributions_mut() {
+			if ele.owners.is_empty() {
+				ele.prune("no owners left".to_owned());
+			}
+		}
+	}
+
+	pub fn distributions(&self) -> impl Iterator<Item = &FleetSecretDistribution> {
+		self.stored.iter().filter(|v| v.pending_prune.is_none())
+	}
+	pub fn distributions_mut(&mut self) -> impl Iterator<Item = &mut FleetSecretDistribution> {
+		self.stored.iter_mut().filter(|v| v.pending_prune.is_none())
+	}
+	pub fn owners(&self) -> impl Iterator<Item = &SecretOwner> {
+		self.distributions().flat_map(|v| v.owners.iter())
+	}
 	#[allow(
 		clippy::len_without_is_empty,
 		reason = "should not be empty for a long time"
 	)]
 	pub fn len(&self) -> usize {
-		self.0.len()
+		self.distributions().count()
 	}
 
-	pub fn get(&self, owner: &str) -> Option<&FleetSecretDistribution> {
-		self.0.iter().find(|d| d.owners.contains(owner))
+	pub fn get(&self, owner: &SecretOwner) -> Option<&FleetSecretDistribution> {
+		self.distributions().find(|d| d.owners.contains(owner))
 	}
-	fn entry(&mut self, owner: String) -> DistEntry<'_> {
-		let Some(idx) = self.0.iter().position(|d| d.owners.contains(&owner)) else {
+	fn entry(&mut self, owner: SecretOwner) -> DistEntry<'_> {
+		let Some((idx, dist)) = self
+			.distributions()
+			.enumerate()
+			.find(|(_, d)| d.owners.contains(&owner))
+		else {
 			return DistEntry::Vacant(VacantDistEntry {
 				distributions: self,
-				owner,
+				owners: BTreeSet::from([owner]),
 			});
 		};
 		DistEntry::Occupied(OccupiedDistEntry {
+			owners: dist.owners.clone(),
 			distributions: self,
 			idx,
-			owner,
 		})
 	}
-	fn extend(&mut self, dist: FleetSecretDistribution) {
-		for owner in &dist.owners {
-			self.entry(owner.to_owned()).remove();
+	pub fn extend(&mut self, dist: FleetSecretDistribution, reason: String) {
+		for ele in self.distributions_mut() {
+			ele.prune_owners(&dist.owners, reason.clone());
 		}
-		self.0.push(dist);
+		self.stored.push(dist);
 	}
-	pub fn contains(&self, owner: &str) -> bool {
-		self.0.iter().any(|d| d.owners.contains(owner))
+	pub fn contains(&self, owner: &SecretOwner) -> bool {
+		self.distributions().any(|d| d.owners.contains(owner))
 	}
 }
 
 struct OccupiedDistEntry<'d> {
 	distributions: &'d mut FleetSecretDistributions,
 	idx: usize,
-	owner: String,
+	owners: BTreeSet<SecretOwner>,
 }
 impl<'d> OccupiedDistEntry<'d> {
-	fn remove(self) -> VacantDistEntry<'d> {
-		let dist = &mut self.distributions.0[self.idx];
-		assert!(
-			dist.owners.remove(&self.owner),
-			"entry exists, as we have its reference"
-		);
-		if dist.owners.is_empty() {
-			self.distributions.0.remove(self.idx);
+	fn remove(self, whole_dist: bool, reason: String) -> VacantDistEntry<'d> {
+		let dist = &mut self.distributions.stored[self.idx];
+		if whole_dist {
+			dist.prune(reason);
+		} else {
+			dist.prune_owners(&self.owners, reason);
 		}
 		VacantDistEntry {
 			distributions: self.distributions,
-			owner: self.owner,
+			owners: self.owners,
 		}
 	}
-	fn set(self, secret: FleetSecretData) -> Self {
-		self.remove().set(secret)
+	fn set(self, secret: FleetSecretData, reason: String) -> Self {
+		self.remove(false, reason).set(secret)
 	}
 }
 struct VacantDistEntry<'d> {
 	distributions: &'d mut FleetSecretDistributions,
-	owner: String,
+	owners: BTreeSet<SecretOwner>,
 }
 impl<'d> VacantDistEntry<'d> {
 	fn set(self, secret: FleetSecretData) -> OccupiedDistEntry<'d> {
 		let Self {
 			distributions,
-			owner,
+			owners,
 		} = self;
-		let idx = distributions.0.len();
-		distributions.0.push(FleetSecretDistribution {
-			owners: BTreeSet::from_iter([owner.clone()]),
+		let idx = distributions.stored.len();
+		distributions.stored.push(FleetSecretDistribution {
+			owners: owners.clone(),
 			secret,
 
+			owners_pending_prune: BTreeMap::new(),
+			pending_prune: None,
 			_deprecated_managed: true,
 		});
 		OccupiedDistEntry {
 			distributions,
-			owner,
+			owners,
 			idx,
 		}
 	}
@@ -262,16 +568,16 @@
 	Occupied(OccupiedDistEntry<'d>),
 }
 impl DistEntry<'_> {
-	fn remove(self) -> Self {
+	fn remove(self, whole_dist: bool, reason: String) -> Self {
 		match self {
 			DistEntry::Vacant(_) => self,
-			DistEntry::Occupied(o) => Self::Vacant(o.remove()),
+			DistEntry::Occupied(o) => Self::Vacant(o.remove(whole_dist, reason)),
 		}
 	}
-	fn set(self, secret: FleetSecretData) -> Self {
+	fn set(self, secret: FleetSecretData, reason: String) -> Self {
 		Self::Occupied(match self {
 			DistEntry::Vacant(e) => e.set(secret),
-			DistEntry::Occupied(e) => e.set(secret),
+			DistEntry::Occupied(e) => e.set(secret, reason),
 		})
 	}
 }
@@ -281,8 +587,13 @@
 	where
 		S: serde::Serializer,
 	{
+		let mut v = self.clone();
+		v.prune_dead();
 		let mut found_hosts = BTreeSet::new();
-		for ele in self.0.iter() {
+		for ele in v.distributions() {
+			if ele.pending_prune.is_some() {
+				continue;
+			}
 			if ele.owners.is_empty() {
 				panic!("consistency: secret distribution has no defined owners");
 			}
@@ -294,10 +605,15 @@
 				}
 			}
 		}
-		match self.0.len() {
+		match v.stored.len() {
 			0 => panic!("consistency: empty distributions"),
-			1 => self.0[0].serialize(serializer),
-			_ => self.0.serialize(serializer),
+			1 => v.stored[0].serialize(serializer),
+			_ => {
+				let mut sorted = v.stored.clone();
+				// Store outdated distributions last
+				sorted.sort_by_key(|v| v.pending_prune.is_some() as u32);
+				sorted.serialize(serializer)
+			}
 		}
 	}
 }
@@ -313,15 +629,18 @@
 			Many(Vec<FleetSecretDistribution>),
 		}
 		let d = Distributions::deserialize(deserializer)?;
-		let ds = match d {
+		let stored = match d {
 			Distributions::One(d) => vec![d],
 			Distributions::Many(ds) => ds,
 		};
-		if ds.is_empty() {
+		if stored.is_empty() {
 			return Err(de::Error::custom("consistency: empty distributions"));
 		}
 		let mut found_hosts = BTreeSet::new();
-		for ele in ds.iter() {
+		for ele in stored.iter() {
+			if ele.pending_prune.is_some() {
+				continue;
+			}
 			if ele.owners.is_empty() {
 				return Err(de::Error::custom(
 					"consistency: secret distribution has no defined owners",
@@ -335,73 +654,65 @@
 				}
 			}
 		}
-		Ok(Self(ds))
+		Ok(Self { stored })
 	}
 }
 
-#[derive(Serialize, Deserialize, Default)]
+#[derive(Deserialize, Default)]
 pub struct FleetSecrets(BTreeMap<String, FleetSecretDistributions>);
 
+impl Serialize for FleetSecrets {
+	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: serde::Serializer,
+	{
+		let data: BTreeMap<String, FleetSecretDistributions> = self
+			.0
+			.iter()
+			.filter(|(_, v)| !v.stored.is_empty())
+			.map(|(k, v)| (k.clone(), v.clone()))
+			.collect();
+
+		data.serialize(serializer)
+	}
+}
+
 impl FleetSecrets {
 	pub fn keys(&self) -> btree_map::Keys<String, FleetSecretDistributions> {
 		self.0.keys()
 	}
 
-	pub fn keys_for_owner(&self, owner: &str) -> impl Iterator<Item = &String> {
+	pub fn keys_for_owner(&self, owner: &SecretOwner) -> impl Iterator<Item = &String> {
 		self.0
 			.iter()
 			.filter(|(_, d)| d.contains(owner))
 			.map(|(n, _)| n)
 	}
 
-	pub fn drop_owner_no_reencrypt(&mut self, secret: &str, owner: &str) -> bool {
-		let Entry::Occupied(mut dists) = self.0.entry(secret.to_owned()) else {
-			return false;
-		};
-		let DistEntry::Occupied(dist) = dists.get_mut().entry(owner.to_owned()) else {
-			return false;
-		};
-
-		dist.remove();
-
-		if dists.get().0.is_empty() {
-			dists.remove();
-		};
-
-		true
-	}
-	pub fn set_single_data(&mut self, secret: String, owner: String, data: FleetSecretData) {
-		let e = self
-			.0
-			.entry(secret.to_owned())
-			.or_insert_with(|| FleetSecretDistributions(Default::default()));
-		e.entry(owner.to_owned()).set(data);
-	}
 	pub fn set_data(&mut self, secret: String, data: FleetSecretDistribution) {
 		match self.0.entry(secret) {
 			Entry::Vacant(e) => {
-				e.insert(FleetSecretDistributions(vec![data]));
+				e.insert(FleetSecretDistributions { stored: vec![data] });
 			}
 			Entry::Occupied(mut e) => {
 				let dists = e.get_mut();
-				dists.extend(data)
+				dists.extend(data, "secret data was replaced".to_owned())
 			}
 		}
-	}
-	pub fn get_single(&self, secret: &str, owner: &str) -> Option<&FleetSecretDistribution> {
-		let secret = self.0.get(secret)?;
-		secret.get(owner)
 	}
 	pub fn get(&self, secret: &str) -> Option<&FleetSecretDistributions> {
 		self.0.get(secret)
 	}
+	pub fn get_mut(&mut self, secret: &str) -> Option<&mut FleetSecretDistributions> {
+		self.0.get_mut(secret)
+	}
 
-	pub fn contains_for_owner(&self, secret: &str, owner: &str) -> bool {
-		let Some(secret) = self.0.get(secret) else {
-			return false;
-		};
-		secret.contains(owner)
+	pub fn get_or_create(&mut self, secret: &str) -> &mut FleetSecretDistributions {
+		self.0
+			.entry(secret.to_owned())
+			.or_insert(FleetSecretDistributions::default())
 	}
+
 	pub fn contains(&self, secret: &str) -> bool {
 		self.0.contains_key(secret)
 	}
@@ -411,7 +722,7 @@
 
 	fn merge_from_hosts(
 		&mut self,
-		host_secrets: BTreeMap<String, BTreeMap<String, FleetSecretDistribution>>,
+		host_secrets: BTreeMap<SecretOwner, BTreeMap<String, FleetSecretDistribution>>,
 	) {
 		for (host, host_secrets) in host_secrets {
 			for (secret_name, mut secret_data) in host_secrets {
@@ -420,11 +731,27 @@
 			}
 		}
 	}
+
+	pub fn prune_host(&mut self, host: &SecretOwner, expected_nonshared: BTreeSet<String>) {
+		for (name, dists) in self.0.iter_mut() {
+			if expected_nonshared.contains(name) {
+				continue;
+			}
+			for dist in dists.distributions_mut() {
+				if dist.owners.contains(host) {
+					dist.prune_owners(
+						&BTreeSet::from([host.to_owned()]),
+						"host no longer defines this secret".to_owned(),
+					);
+				}
+			}
+		}
+	}
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Expectations {
-	pub owners: BTreeSet<String>,
+	pub owners: BTreeSet<SecretOwner>,
 	pub generation_data: serde_json::Value,
 	pub parts: BTreeMap<String, GeneratorPart>,
 }
@@ -432,3 +759,26 @@
 pub struct GeneratorPart {
 	pub encrypted: bool,
 }
+
+#[derive(Debug, Clone, Copy)]
+pub struct RegenerationConstraints {
+	pub allow_different: bool,
+	pub regenerate_on_owner_added: bool,
+	pub regenerate_on_owner_removed: bool,
+}
+impl RegenerationConstraints {
+	pub fn host_personal() -> Self {
+		Self {
+			allow_different: false,
+			regenerate_on_owner_added: true,
+			regenerate_on_owner_removed: true,
+		}
+	}
+	pub fn without_preferences(self) -> Self {
+		Self {
+			allow_different: self.allow_different,
+			regenerate_on_owner_added: false,
+			regenerate_on_owner_removed: false,
+		}
+	}
+}