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,
+		}
+	}
+}

--- a/crates/fleet-base/src/keys.rs
+++ b/crates/fleet-base/src/keys.rs
@@ -1,17 +1,17 @@
 use std::str::FromStr as _;
 
 use age::Recipient;
-use anyhow::{Result, anyhow};
+use anyhow::{Result, anyhow, bail};
 use futures::{StreamExt as _, TryStreamExt as _};
 use itertools::Itertools as _;
 use tracing::warn;
 
-use crate::host::Config;
+use crate::{fleetdata::SecretOwner, host::Config};
 
 impl Config {
-	pub fn cached_key(&self, host: &str) -> Option<String> {
-		let data = self.data();
-		let key = data.hosts.get(host).map(|h| &h.encryption_key);
+	fn cached_host_key(&self, host: &str) -> Option<String> {
+		let hosts = self.data.hosts.read().expect("no poisoning");
+		let key = hosts.get(host).map(|h| &h.encryption_key);
 		if let Some(key) = key
 			&& key.is_empty()
 		{
@@ -20,13 +20,13 @@
 		key.cloned()
 	}
 	pub fn update_key(&self, host: &str, key: String) {
-		let mut data = self.data_mut();
-		let host = data.hosts.entry(host.to_string()).or_default();
+		let mut hosts = self.data.hosts.write().expect("no poisoning");
+		let host = hosts.entry(host.to_string()).or_default();
 		host.encryption_key = key.trim().to_string();
 	}
 
-	pub async fn key(&self, host: &str) -> anyhow::Result<String> {
-		if let Some(key) = self.cached_key(host) {
+	pub async fn host_key(&self, host: &str) -> anyhow::Result<String> {
+		if let Some(key) = self.cached_host_key(host) {
 			Ok(key)
 		} else {
 			warn!("Loading key for {}", host);
@@ -38,18 +38,24 @@
 			Ok(key)
 		}
 	}
+	pub async fn key(&self, owner: &SecretOwner) -> anyhow::Result<String> {
+		if let Some(host) = owner.as_host() {
+			self.host_key(host).await
+		} else {
+			bail!("only host keys supported for now")
+		}
+	}
 	/// Insecure, requires root
-	pub async fn recipient(&self, host: &str) -> anyhow::Result<Box<dyn Recipient>> {
+	pub async fn recipient(&self, host: &SecretOwner) -> anyhow::Result<Box<dyn Recipient>> {
 		let key = self.key(host).await?;
 		age::ssh::Recipient::from_str(&key)
 			.map_err(|e| anyhow!("parse recipient error: {e:?}"))
 			.map(|v| Box::new(v) as Box<dyn Recipient>)
 	}
 
-	pub async fn recipients(&self, hosts: Vec<String>) -> Result<Vec<Box<dyn Recipient>>> {
-		let hosts = self.expand_owner_set(hosts)?;
+	pub async fn recipients(&self, hosts: Vec<SecretOwner>) -> Result<Vec<Box<dyn Recipient>>> {
 		futures::stream::iter(hosts.iter())
-			.then(|m| self.recipient(m.as_ref()))
+			.then(|m| self.recipient(m))
 			.try_collect::<Vec<_>>()
 			.await
 	}
@@ -58,9 +64,8 @@
 	pub async fn orphaned_data(&self) -> Result<Vec<String>> {
 		let mut out = Vec::new();
 		let host_names = self.list_hosts()?.into_iter().map(|h| h.name).collect_vec();
-		for hostname in self
-			.data()
-			.hosts
+		let hosts = self.data.hosts.read().expect("no poisoning");
+		for hostname in hosts
 			.iter()
 			.filter(|(_, host)| !host.encryption_key.is_empty())
 			.map(|(n, _)| n)

--- a/crates/fleet-base/src/lib.rs
+++ b/crates/fleet-base/src/lib.rs
@@ -5,5 +5,4 @@
 mod keys;
 pub mod opts;
 pub mod primops;
-pub mod secret;
 pub mod secret_storage;

--- a/crates/fleet-base/src/opts.rs
+++ b/crates/fleet-base/src/opts.rs
@@ -1,5 +1,5 @@
 use std::{
-	collections::BTreeMap,
+	collections::{BTreeMap, BTreeSet},
 	env::current_dir,
 	ffi::OsString,
 	str::FromStr,
@@ -7,6 +7,7 @@
 };
 
 use anyhow::{Context, Result, bail};
+use chrono::Utc;
 use nix_eval::{
 	FetchSettings, FlakeLockFlags, FlakeReference, FlakeReferenceParseFlags, FlakeSettings, Value,
 	gc_now, nix_go, util::assert_warn,
@@ -212,7 +213,7 @@
 		}
 		let bytes =
 			std::fs::read_to_string(&fleet_data_path).context("reading fleet state (fleet.nix)")?;
-		let data = Arc::new(Mutex::new(FleetData::from_str(&bytes)?));
+		let data = Arc::new(FleetData::from_str(&bytes)?);
 
 		init_primops();
 
@@ -265,6 +266,10 @@
 			gc_now();
 		}
 		let config = Config(Arc::new(FleetConfigInternals {
+			// TODO: Load from somewhere
+			prefer_identities: BTreeSet::new(),
+			now: Utc::now(),
+
 			directory,
 			data,
 			flake_outputs: flake,

--- a/crates/fleet-base/src/primops.rs
+++ b/crates/fleet-base/src/primops.rs
@@ -4,19 +4,16 @@
 use anyhow::{Context, bail, ensure};
 use fleet_shared::SecretData;
 use itertools::Itertools;
-use nix_eval::{NativeFn, Value, nix_go, nix_go_json};
+use nix_eval::{NativeFn, Value, await_in_nix, nix_go, nix_go_json};
 use serde::Deserialize;
 use tracing::{info, warn};
 
 use crate::fleetdata::{
 	Expectations, FleetSecretData, FleetSecretDistribution, FleetSecretPart, GeneratorPart,
+	RegenerationConstraints, SecretOwner,
 };
 use crate::host::{Config, ConfigHost};
-use crate::secret::{RegenerationReason, secret_needs_regeneration};
 use anyhow::{Result, anyhow};
-
-#[derive(thiserror::Error, Debug)]
-enum Error {}
 
 pub static PRIMOPS_DATA: OnceLock<Config> = OnceLock::new();
 
@@ -28,7 +25,6 @@
 }
 
 pub fn get_pkgs_and_generators(host_on: &ConfigHost, recipients: Vec<String>) -> Result<Value> {
-	info!("get pkgs");
 	let pkgs = host_on.pkgs()?;
 	let default_mk_secret_generators = nix_go!(pkgs.mkSecretGenerators);
 	let generators = nix_go!(default_mk_secret_generators(Obj { recipients }));
@@ -57,6 +53,31 @@
 	Ok(default_generator_drv)
 }
 
+fn secret_to_parts(
+	secret_name: &str,
+	secret: &BTreeMap<String, FleetSecretPart>,
+	expected: &BTreeMap<String, GeneratorPart>,
+) -> Value {
+	let mut out = HashMap::new();
+	for (part_name, part) in secret {
+		if !expected.contains_key(part_name) {
+			warn!(
+				"secret {secret_name} part {part_name} is stored, but not defined in nixos config, it will not be passed to nix"
+			);
+			continue;
+		};
+		out.insert(
+			part_name.as_str(),
+			Value::new_attrs(HashMap::from_iter([(
+				"raw",
+				Value::new_str(&part.raw.to_string()),
+			)])),
+		);
+	}
+
+	Value::new_attrs(out)
+}
+
 pub async fn generate(
 	config: &Config,
 	expectations: Expectations,
@@ -76,9 +97,12 @@
 			} else {
 				config.local_host()
 			};
-			let pkgs_and_generators =
-				get_pkgs_and_generators(&host_on, expectations.owners.iter().cloned().collect())
-					.context("failed to get pkgs for target host")?;
+			let mut recipients = Vec::new();
+			for owner in &expectations.owners {
+				recipients.push(config.key(owner).await?);
+			}
+			let pkgs_and_generators = get_pkgs_and_generators(&host_on, recipients)
+				.context("failed to get pkgs for target host")?;
 			let generator = call_package(config, &pkgs_and_generators, generator)
 				.context("failed to evaluate generator for target host")?;
 
@@ -147,15 +171,8 @@
 				generation_data: expectations.generation_data.clone(),
 			};
 
-			let new_data = FleetSecretDistribution {
-				secret: new_data,
-				owners: expectations.owners.clone(),
-				_deprecated_managed: true,
-			};
-
-			if let Some(reason) = secret_needs_regeneration(&new_data, &expectations) {
-				bail!("newly generated secret needs to be regenerated: {reason}")
-			}
+			let new_data =
+				FleetSecretDistribution::new(expectations.owners.clone(), new_data, config.now);
 
 			Ok(new_data)
 		}
@@ -166,18 +183,41 @@
 }
 
 pub fn init_primops() {
-	info!("initializing primops");
 	NativeFn::new(
+		c"__fleetEnsureHostSecrets",
+		c"Ensure no extra secrets are stored for the host, pruning unknown",
+		[c"host", c"expectedNonshared", c"expectedShared", c"rest"],
+		|_es, [host, expected_nonshared, expected_shared, rest]| {
+			let host = SecretOwner::host(host.to_string()?);
+			let expected_nonshared: BTreeSet<String> = expected_nonshared.as_json()?;
+			let expected_shared: BTreeSet<String> = expected_shared.as_json()?;
+
+			let mut expected = expected_nonshared;
+			expected.extend(expected_shared);
+
+			let config = PRIMOPS_DATA
+				.get()
+				.expect("primops data should be set on init");
+
+			config
+				.data
+				.secrets
+				.write()
+				.expect("no poisoning")
+				.prune_host(&host, expected);
+
+			Ok(rest.clone())
+		},
+	)
+	.register();
+	NativeFn::new(
 		c"__fleetEnsureHostSecret",
 		c"Ensure secret existence for a host, regenerating it in case of some mismatch",
 		[c"host", c"secret", c"generator"],
 		|es, [host, secret, generator]| {
-			info!("get host");
-			let host = host.to_string()?;
-			info!("get secret");
+			let host = SecretOwner::host(&host.to_string()?);
 			let secret = secret.to_string()?;
 
-			info!("get config");
 			let config = PRIMOPS_DATA
 				.get()
 				.expect("primops data should be set on init");
@@ -193,50 +233,101 @@
 
 				ensure!(expected_owners.contains(&host), "secret {secret} does not define {host} as expected owner");
 
-				(true, shared_def.generator()?, expected_owners)
+				(Some(shared_def.clone()), shared_def.generator()?, expected_owners)
 			} else {
 				if shared_def.is_some() {
 					bail!("hosts can only have their own generators for non-shared secrets, either set host secret generator to \"shared\", or remove shared secret generator at fleetConfiguration.secrets.{secret}.generator")
 				}
 
-				(false, generator.clone(), BTreeSet::from_iter([host.clone()]))
+				(None, generator.clone(), BTreeSet::from_iter([host.clone()]))
 			};
 
-			let default_generator_drv = get_default_generator_drv(config, &generator).context("failed to evaluate default generator")?;
-			let expectations = Expectations {
+			let default_generator_drv = get_default_generator_drv(config, &generator)?;
+			let mut expectations = Expectations {
 				parts: nix_go_json!(default_generator_drv.parts),
 				generation_data: nix_go_json!(default_generator_drv.generationData),
-				owners: expected_owners,
+				owners: expected_owners.clone(),
+			};
+			let constraints = if let Some(shared) = &shared{
+				RegenerationConstraints {
+					allow_different: nix_go_json!(default_generator_drv.allowDifferent) && shared.allow_different()?,
+					regenerate_on_owner_added: shared.regenerate_on_owner_added()?,
+					regenerate_on_owner_removed: shared.regenerate_on_owner_added()?,
+				}
+			} else {
+				RegenerationConstraints::host_personal()
 			};
 
-			let reason: RegenerationReason = 'regenerate: {
-				let Some(existing) = config
-					.host_secret(&host, &secret) else {
-					break 'regenerate RegenerationReason::Missing;
+			let mut secrets = config.data.secrets.write().expect("no poisoning");
+			let dists = secrets.get_or_create(&secret);
+
+				if shared.is_some() {
+					dists.prune_shared(&expected_owners, !constraints.allow_different, &expectations.parts, &expectations.generation_data, constraints.regenerate_on_owner_removed, constraints.regenerate_on_owner_added, &config.prefer_identities, config.now);
+				} else {
+					dists.prune_host(host.clone(), &expectations.parts, &expectations.generation_data, config.now);
+				};
+
+				if let Some(dist) = dists.get(&host) {
+					return Ok(secret_to_parts(&secret, &dist.secret.parts, &expectations.parts));
 				};
-				if let Some(reason) = secret_needs_regeneration(&existing, &expectations) {
-					break 'regenerate reason;
+
+				let mut reencrypt_targets = expectations.owners.clone();
+				for dist in dists.distributions() {
+					for own in dist.owners() {
+						reencrypt_targets.remove(own);
+					}
 				}
+				if !constraints.regenerate_on_owner_added {
+					if let Some(unpruned) = dists.try_unprune(host.clone()) {
+						return Ok(secret_to_parts(&secret, &unpruned.secret.parts, &expectations.parts));
+					} else if let Some(best) = dists.best_distribution_for_reencryption(&config.prefer_identities) {
+						let new_owners = reencrypt_targets.clone();
+						let mut reencrypt_targets = reencrypt_targets;
+						reencrypt_targets.extend(best.owners().cloned());
 
-				let mut parts = expectations.parts.clone();
+						let mut preferred = best.owners().collect_vec();
+						preferred.sort_by_key(|v| !config.prefer_identities.contains(*v));
+
+						warn!("reencrypting secret {secret} as it is missing for host {host}");
 
-				let mut out = HashMap::new();
-				for (part_name, part) in &existing.secret.parts {
-					let Some(definition) = parts.remove(part_name) else {
-						warn!("secret {secret} part {part_name} is stored, but not defined in nixos config, it will not be passed to nix");
-						continue;
+						for owner in preferred {
+							if let Some(hostname) = owner.as_host() && let Ok(host) = config.host(hostname) {
+								let best = best.clone();
+								let reencrypt_targets = reencrypt_targets.clone();
+								let reencrypted = match await_in_nix(async move {
+										host.reencrypt_distribution(&best, reencrypt_targets.clone(), config.now).await
+								}) {
+									Ok(r) => r,
+									Err(e) => {
+										warn!("reencryption failed on {hostname}: {e:?}");
+										continue;
+									}
+								};
+								dists.extend(reencrypted.clone(), format!("secret was reencrypted to extend with new owners: {new_owners:?}"));
+								return Ok(secret_to_parts(&secret, &reencrypted.secret.parts, &expectations.parts));
+							};
+						}
+						warn!("failed to reencrypt using any host")
 					};
-					assert!(definition.encrypted != part.raw.encrypted, "encryption status is checked by secret_needs_regeneration");
-					out.insert(part_name.as_str(), Value::new_attrs(HashMap::from_iter([("raw", Value::new_str(&part.raw.to_string()))])));
-				}
-				assert!(parts.is_empty(), "secret part is missing, secret_needs_regeneration should check that");
+				};
 
-				return Ok(Value::new_attrs(out))
-			};
+			if constraints.allow_different {
+				for dist in dists.distributions() {
+					for own in dist.owners() {
+						expectations.owners.remove(own);
+					}
+				}
+			}
+			info!("secret {secret} is being generated for {:?}", expectations.owners);
 
-			todo!()
+			let expectations_ = expectations.clone();
+			let generated = await_in_nix(async move {
+				generate(config, expectations_, &generator, &default_generator_drv).await
+			})?;
 
+			dists.extend(generated.clone(), format!("secret was generated"));
 
+			return Ok(secret_to_parts(&secret, &generated.secret.parts, &expectations.parts));
 		},
 	)
 	.register();

--- a/crates/fleet-shared/Cargo.toml
+++ b/crates/fleet-shared/Cargo.toml
@@ -8,4 +8,3 @@
 base64 = "0.22.1"
 serde = "1.0.219"
 unicode_categories = "0.1.1"
-z85 = "3.0.6"

--- a/crates/fleet-shared/src/encoding.rs
+++ b/crates/fleet-shared/src/encoding.rs
@@ -1,5 +1,4 @@
 use std::{
-	collections::BTreeMap,
 	fmt::{self, Display},
 	str::FromStr,
 };
@@ -15,7 +14,6 @@
 }
 
 const BASE64_ENCODED_PREFIX: &str = "<BASE64-ENCODED>\n";
-const Z85_ENCODED_PREFIX: &str = "<Z85-ENCODED>\n";
 // Multiline text in Nix can only end with \n, which is not cool for actual single-line strings.
 const PLAINTEXT_NEWLINE_PREFIX: &str = "<PLAINTEXT-NL>\n";
 const PLAINTEXT_PREFIX: &str = "<PLAINTEXT>";
@@ -54,18 +52,12 @@
 			STANDARD_NO_PAD
 				.decode(unprefixed.replace(['\n', '\t', ' '], ""))
 				.map_err(|e| format!("base64-encoded failed: {e}"))?
-		} else if let Some(unprefixed) = string.strip_prefix(Z85_ENCODED_PREFIX) {
-			z85::decode(unprefixed.replace(['\n', '\t', ' '], ""))
-				.map_err(|e| format!("z85-encoded failed: {e}"))?
 		} else if let Some(unprefixed) = string.strip_prefix(PLAINTEXT_NEWLINE_PREFIX) {
 			unprefixed.as_bytes().to_owned()
 		} else if let Some(unprefixed) = string.strip_prefix(PLAINTEXT_PREFIX) {
 			unprefixed.as_bytes().to_owned()
 		} else {
-			let secret_prefix = format!("{SECRET_PREFIX}{Z85_ENCODED_PREFIX}");
-			return Err(format!(
-				"unknown secret encoding. If you're migrating from old version of fleet, prefix public secret fields with {PLAINTEXT_PREFIX:?}, and encrypted data with {secret_prefix:?}: {string}"
-			));
+			return Err(format!("unknown secret encoding"));
 		};
 		Ok(Self { data, encrypted })
 	}

--- a/crates/nix-eval/Cargo.toml
+++ b/crates/nix-eval/Cargo.toml
@@ -18,6 +18,7 @@
 test-log = { version = "0.2.18", features = ["trace"] }
 tracing-indicatif = { version = "0.3.13", optional = true }
 vte = { version = "0.15.0", features = ["ansi"] }
+tokio.workspace = true
 
 [build-dependencies]
 bindgen = "0.72.0"

--- a/crates/nix-eval/src/lib.rs
+++ b/crates/nix-eval/src/lib.rs
@@ -2,7 +2,7 @@
 use std::cell::RefCell;
 use std::ffi::{CStr, CString, c_char, c_int, c_uint, c_void};
 use std::ptr::{null, null_mut};
-use std::sync::LazyLock;
+use std::sync::{Arc, LazyLock, OnceLock};
 use std::{array, fmt, slice};
 use std::{collections::HashMap, path::PathBuf};
 
@@ -10,9 +10,10 @@
 use itertools::Itertools;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
+use std::mem::transmute;
 
 pub use anyhow::Result;
-use tracing::{info, instrument, warn};
+use tracing::{Instrument, info, instrument, warn};
 
 use self::logging::{ErrorInfoBuilder, nix_logging_cxx};
 use self::nix_cxx::set_fetcher_setting;
@@ -172,9 +173,16 @@
 #[repr(transparent)]
 pub struct NixContext(*mut c_context);
 impl NixContext {
-	pub fn set_err(&mut self, err: NixErrorKind, msg: &CStr) {
+	pub fn set_err_raw(&mut self, err: NixErrorKind, msg: &CStr) {
 		unsafe { set_err_msg(self.0, err as c_int, msg.as_ptr()) };
 	}
+	pub fn set_err(&mut self, err: anyhow::Error) {
+		let mut fmt = format!("{err:?}").replace("\0", "\\0");
+		self.set_err_raw(
+			NixErrorKind::Generic,
+			&CString::new(fmt).expect("NUL bytes were just replaced"),
+		);
+	}
 	pub fn new() -> Self {
 		let ctx = unsafe { c_context_create() };
 		Self(ctx)
@@ -931,6 +939,8 @@
 	}
 }
 
+static TOKIO_FOR_NIX: OnceLock<Arc<tokio::runtime::Runtime>> = OnceLock::new();
+
 pub fn init_libraries() {
 	unsafe { GC_allow_register_threads() };
 
@@ -945,37 +955,33 @@
 	nix_logging_cxx::apply_tracing_logger();
 }
 
+pub fn init_tokio_for_nix(tokio: Arc<tokio::runtime::Runtime>) {
+	TOKIO_FOR_NIX
+		.set(tokio)
+		.expect("tokio for nix should only be initialized once");
+}
+
+pub fn await_in_nix<F: Send + 'static>(f: impl Future<Output = F> + Send + 'static) -> F {
+	// It should be possible to do Handle::current(), but some of the planned features don't work well with that
+	let runtime = TOKIO_FOR_NIX
+		.get()
+		.expect("init_tokio_for_nix was not called");
+	std::thread::spawn(move || runtime.block_on(f)).join().expect("await_in_nix inner thread panicked")
+}
+
 unsafe extern "C" fn nix_primop_closure_adapter<const N: usize>(
 	user_data: *mut c_void,
-	context: *mut c_context,
+	mut context: *mut c_context,
 	state: *mut nix_raw::EvalState,
 	args: *mut *mut value,
 	ret: *mut value,
 ) {
 	let user_closure: &UserClosure<N> = unsafe { &*user_data.cast_const().cast() };
-	let mut e = None;
 	let args: [&Value; N] = array::from_fn(|i| {
 		let v: &mut Value = unsafe { &mut *args.add(i).cast() };
-
-		info!("forcing arg");
-		if matches!(v.type_of(), NixType::Thunk)
-			&& let Err(err) = v.force(state)
-		{
-			e = Some(err);
-		};
 		v as &Value
 	});
-	info!("args forced");
-	let ctx: &mut NixContext = unsafe { &mut *context.cast() };
-
-	if let Some(e) = e {
-		warn!("set err = {e}");
-		unsafe { init_int(context, ret, 0) };
-		return ctx.set_err(
-			NixErrorKind::Unknown,
-			&CString::new(e.to_string()).expect("forcing argument value failed"),
-		);
-	}
+	let ctx: &mut NixContext = unsafe { transmute(&mut context) };
 
 	let state: &EvalState = unsafe { std::mem::transmute(&state) };
 
@@ -984,12 +990,7 @@
 			unsafe { copy_value(context, ret, v.0) };
 		}
 		Err(e) => {
-			unsafe { init_int(context, ret, 0) };
-			warn!("set err = {e:#?}");
-			ctx.set_err(
-				NixErrorKind::Unknown,
-				&CString::new(e.to_string()).expect("error should not contain internal nuls"),
-			);
+			ctx.set_err(e);
 		}
 	}
 }

--- a/crates/nix-eval/src/macros.rs
+++ b/crates/nix-eval/src/macros.rs
@@ -68,13 +68,13 @@
 #[macro_export]
 macro_rules! nix_go {
 	(@o($o:expr, $path:expr) . $var:ident $($tt:tt)*) => {{
-		nix_go!(@o($o.get_field(stringify!($var)).context(concat!("getting nested ", $path))?, $path) $($tt)*)
+		nix_go!(@o(tokio::task::block_in_place(|| $o.get_field(stringify!($var))).context(concat!("getting nested ", $path))?, $path) $($tt)*)
 	}};
 	(@o($o:expr, $path:expr) [ $v:expr ] $($tt:tt)*) => {{
-		nix_go!(@o($o.get_field($v).context(concat!("getting nested ", $path))?, $path) $($tt)*)
+		nix_go!(@o(tokio::task::block_in_place(|| $o.get_field($v)).context(concat!("getting nested ", $path))?, $path) $($tt)*)
 	}};
 	(@o($o:expr, $path:expr) ($($var:tt)*) $($tt:tt)*) => {
-		nix_go!(@o($o.call($crate::nix_expr_inner!($($var)+)).context(concat!("getting nested ", $path))?, $path) $($tt)*)
+		nix_go!(@o(tokio::task::block_in_place(|| $o.call($crate::nix_expr_inner!($($var)+))).context(concat!("getting nested ", $path))?, $path) $($tt)*)
 	};
 	(@o($o:expr, $path:expr)) => {$o};
 	($field:ident $($tt:tt)+) => {{
@@ -87,6 +87,6 @@
 #[macro_export]
 macro_rules! nix_go_json {
 	($($tt:tt)*) => {{
-		$crate::nix_go!($($tt)*).as_json()?
+		tokio::task::block_in_place(|| $crate::nix_go!($($tt)*).as_json())?
 	}};
 }

--- a/flake.lock
+++ b/flake.lock
@@ -2,10 +2,10 @@
   "nodes": {
     "crane": {
       "locked": {
-        "lastModified": 1767461147,
+        "lastModified": 1768700043,
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "7d59256814085fd9666a2ae3e774dc5ee216b630",
+        "rev": "935de8bd6838d940988bb065be2a2034259327b9",
         "type": "github"
       },
       "original": {
@@ -37,10 +37,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1767609335,
+        "lastModified": 1768135262,
         "owner": "hercules-ci",
         "repo": "flake-parts",
-        "rev": "250481aafeb741edfe23d29195671c19b36b6dca",
+        "rev": "80daad04eddbbf5a4d883996a73f3f542fa437ac",
         "type": "github"
       },
       "original": {
@@ -111,10 +111,10 @@
         "nixpkgs-regression": "nixpkgs-regression"
       },
       "locked": {
-        "lastModified": 1767670640,
+        "lastModified": 1768702010,
         "owner": "deltarocks",
         "repo": "nix",
-        "rev": "2181cd07134c9049bd77b7f48c3b1ea8647267de",
+        "rev": "b05b52670b9c7affff5b9be3edb539a1603c39e6",
         "type": "github"
       },
       "original": {
@@ -126,10 +126,10 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1767657734,
+        "lastModified": 1768697925,
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "d4ccebf51ee4dbeb9df364dce1fe9848635c1258",
+        "rev": "665062f7df2c7db8fdbbec4f1b730091143828a3",
         "type": "github"
       },
       "original": {
@@ -190,10 +190,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1767667566,
+        "lastModified": 1768617670,
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "056ce5b125ab32ffe78c7d3e394d9da44733c95e",
+        "rev": "56d0fbdd732f3686e8414b857cf885038fc17d57",
         "type": "github"
       },
       "original": {
@@ -223,10 +223,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1767468822,
+        "lastModified": 1768158989,
         "owner": "numtide",
         "repo": "treefmt-nix",
-        "rev": "d56486eb9493ad9c4777c65932618e9c2d0468fc",
+        "rev": "e96d59dff5c0d7fddb9d113ba108f03c3ef99eca",
         "type": "github"
       },
       "original": {

--- a/modules/nixos.nix
+++ b/modules/nixos.nix
@@ -10,7 +10,7 @@
 let
   inherit (lib.attrsets) mapAttrs;
   inherit (lib.options) mkOption;
-  inherit (lib.types) deferredModule unspecified;
+  inherit (lib.types) deferredModule unspecified uniq str;
   inherit (lib.strings) escapeNixIdentifier;
   inherit (fleetLib.options) mkHostsOption;
 
@@ -24,9 +24,18 @@
       '';
       type = deferredModule;
     };
-    hosts = mkHostsOption (hostArgs: {
+    hosts = mkHostsOption (hostArgs: let
+      hostName = hostArgs.config._module.args.name;
+    in {
       inherit _file;
       options = {
+        name = mkOption {
+          description = ''
+            Host name (alias)
+          '';
+          type = uniq str;
+          default = hostName;
+        };
         nixos = mkOption {
           description = ''
             Nixos configuration for the current host.
@@ -42,7 +51,7 @@
               prefix = [
                 "fleetConfiguration"
                 "hosts"
-                hostArgs.config._module.args.name
+                hostName
                 "nixos"
               ];
               modules = (import "${modulesPath}/module-list.nix") ++ [

--- a/modules/nixos/secrets.nix
+++ b/modules/nixos/secrets.nix
@@ -15,7 +15,12 @@
   inherit (lib.stringsWithDeps) stringAfter;
   inherit (lib.options) mkOption literalExpression;
   inherit (lib.lists) optional elem;
-  inherit (lib.attrsets) mapAttrs mapAttrsToList;
+  inherit (lib.attrsets)
+    mapAttrs
+    mapAttrsToList
+    filterAttrs
+    attrNames
+    ;
   inherit (lib.modules) mkIf;
   inherit (lib.types)
     submodule
@@ -25,9 +30,9 @@
     uniq
     functionTo
     package
-    bool
     enum
     either
+    listOf
     ;
   inherit (fleetLib.strings) decodeRawSecret;
 
@@ -132,10 +137,33 @@
 in
 {
   options = {
+    _providedSharedSecrets = mkOption {
+      description = ''
+        List of shared secrets, for which the current host was specified as `expectedOwners`
+      '';
+      type = listOf str;
+      default = [];
+      internal = true;
+    };
     secrets = mkOption {
       type = attrsOf secretType;
       default = { };
-      apply = mapAttrs (_: secret: secret.parts // { definition = secret; });
+      apply =
+        secrets:
+        mapAttrs (_: secret: secret.parts // { definition = secret; })
+
+          (
+            let
+              hostName = host.name;
+              expectedNonshared = attrNames (filterAttrs (_: def: def.generator != "shared") secrets);
+              expectedShared = config._providedSharedSecrets;
+            in
+            builtins.deepSeq [
+              hostName
+              expectedNonshared
+              expectedShared
+            ] (builtins.fleetEnsureHostSecrets hostName expectedNonshared expectedShared secrets)
+          );
       description = "Host-local secrets";
     };
     system.secretsData = mkOption {
@@ -163,7 +191,7 @@
           (secret.definition.generator == "shared") == hasSharedDefinition
           && (
             hasSharedDefinition
-            -> (elem host._module.args.name fleetConfiguration.secrets.${name}.expectedOwners)
+            -> (elem host.name fleetConfiguration.secrets.${name}.expectedOwners)
           );
         message =
           if hasSharedDefinition then