From c8f31322b4fce235f2f4bc56f0e3704775aa6a9f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 6 Aug 2026 11:13:30 +0100 Subject: [PATCH] Enforce singleton resident writers --- .../src/bin/epiphany-heartbeat-store.rs | 4 +- epiphany-core/src/bin/epiphany-swarm.rs | 10 +- epiphany-core/src/resident_readiness.rs | 106 ++++++++++++++++++ 3 files changed, 115 insertions(+), 5 deletions(-) diff --git a/epiphany-core/src/bin/epiphany-heartbeat-store.rs b/epiphany-core/src/bin/epiphany-heartbeat-store.rs index f3fb9ba2..1a2dc02a 100644 --- a/epiphany-core/src/bin/epiphany-heartbeat-store.rs +++ b/epiphany-core/src/bin/epiphany-heartbeat-store.rs @@ -28,7 +28,8 @@ use epiphany_core::tick_heartbeat_store; use epiphany_core::update_heartbeat_heat_store; use epiphany_core::validate_agent_memory_store; use epiphany_core::{ - ResidentProviderReadiness, authenticate_epiphany_packaged_release, capture_process_instance, + ResidentProviderReadiness, acquire_resident_process_singleton, + authenticate_epiphany_packaged_release, capture_process_instance, publish_resident_provider_readiness, }; use std::env; @@ -457,6 +458,7 @@ fn main() -> Result<()> { )); } let store_path = store_path.ok_or_else(|| anyhow!("serve requires --store"))?; + let _singleton = acquire_resident_process_singleton("heartbeat", &store_path)?; let artifact_dir = artifact_dir.ok_or_else(|| anyhow!("serve requires --artifact-dir"))?; let provider_release = if resident_self_store.is_some() { diff --git a/epiphany-core/src/bin/epiphany-swarm.rs b/epiphany-core/src/bin/epiphany-swarm.rs index eff60f58..ab91dc7c 100644 --- a/epiphany-core/src/bin/epiphany-swarm.rs +++ b/epiphany-core/src/bin/epiphany-swarm.rs @@ -4,10 +4,11 @@ use epiphany_core::{ ChildObservation, CoordinatorLaunch, LaunchedCoordinator, ProcessInstanceIdentity, ProcessInstanceObservation, ResidentProviderReadiness, ResidentReadinessRequest, ResidentSelfOutcome, ResidentSelfPolicy, ResidentSelfPorts, ResidentSelfPressure, - ResidentSelfState, acknowledge_resident_self_launch, authenticate_resident_self_policy, - bind_runtime_repository_domain, bridge_admitted_persona_feedback_to_heartbeat, - cancel_resident_self_turn, capture_process_instance, complete_resident_self_turn, - coordinator_run_receipts, derive_resident_cognition_readiness, enqueue_resident_self_pressure, + ResidentSelfState, acknowledge_resident_self_launch, acquire_resident_process_singleton, + authenticate_resident_self_policy, bind_runtime_repository_domain, + bridge_admitted_persona_feedback_to_heartbeat, cancel_resident_self_turn, + capture_process_instance, complete_resident_self_turn, coordinator_run_receipts, + derive_resident_cognition_readiness, enqueue_resident_self_pressure, import_bifrost_persona_feedback_deliveries, ingest_resident_self_domain_pressure, load_epiphany_cultmesh_swarm_brake, load_resident_self_state, observe_process_instance, prepare_resident_self_launch, publish_resident_provider_readiness, resident_self_child_claim, @@ -60,6 +61,7 @@ fn main() -> Result<()> { println!("{}", serde_json::to_string_pretty(&projection)?); return Ok(()); } + let _singleton = acquire_resident_process_singleton("resident-self", &args.state_store)?; if let Some(pressure) = args.pressure.as_ref() { enqueue_resident_self_pressure(&args.state_store, pressure)?; } diff --git a/epiphany-core/src/resident_readiness.rs b/epiphany-core/src/resident_readiness.rs index 276846c4..6819b9fa 100644 --- a/epiphany-core/src/resident_readiness.rs +++ b/epiphany-core/src/resident_readiness.rs @@ -10,13 +10,104 @@ use cultcache_rs::{ SingleFileMessagePackBackingStore, }; use serde::Serialize; +use sha2::{Digest, Sha256}; +#[cfg(windows)] +use std::ffi::OsStr; use std::fs; +#[cfg(unix)] +use std::fs::File; +#[cfg(unix)] +use std::os::fd::AsRawFd; +#[cfg(windows)] +use std::os::windows::ffi::OsStrExt; use std::path::{Path, PathBuf}; +#[cfg(windows)] +use windows_sys::Win32::Foundation::{CloseHandle, ERROR_ALREADY_EXISTS, GetLastError, HANDLE}; +#[cfg(windows)] +use windows_sys::Win32::System::Threading::{CreateMutexW, ReleaseMutex}; + pub const RESIDENT_PROVIDER_READINESS_SCHEMA_VERSION: &str = "epiphany.resident_cognition.provider_readiness.v0"; const PROVIDER_READINESS_KEY: &str = "provider-readiness"; +pub struct ResidentProcessSingletonGuard { + #[cfg(windows)] + handle: HANDLE, + #[cfg(unix)] + file: File, +} + +impl Drop for ResidentProcessSingletonGuard { + fn drop(&mut self) { + #[cfg(windows)] + unsafe { + let _ = ReleaseMutex(self.handle); + CloseHandle(self.handle); + } + #[cfg(unix)] + unsafe { + let _ = libc::flock(self.file.as_raw_fd(), libc::LOCK_UN); + } + } +} + +pub fn acquire_resident_process_singleton( + role: &str, + store: impl AsRef, +) -> Result { + if !matches!(role, "heartbeat" | "resident-self") { + bail!("unsupported resident singleton role: {role}"); + } + let store = canonical_or_absolute(store.as_ref()); + let identity = format!( + "{:x}", + Sha256::digest(format!("{role}|{}", store.to_string_lossy().to_lowercase()).as_bytes()) + ); + #[cfg(windows)] + { + let name = OsStr::new(&format!("Global\\EpiphanyResident-{role}-{identity}")) + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let handle = unsafe { CreateMutexW(std::ptr::null_mut(), 1, name.as_ptr()) }; + if handle.is_null() { + bail!( + "resident {role} singleton mutex creation failed: {}", + unsafe { GetLastError() } + ); + } + if unsafe { GetLastError() } == ERROR_ALREADY_EXISTS { + unsafe { CloseHandle(handle) }; + bail!( + "resident {role} owner already exists for canonical store {}", + store.display() + ); + } + return Ok(ResidentProcessSingletonGuard { handle }); + } + #[cfg(unix)] + { + let path = std::env::temp_dir().join(format!("epiphany-resident-{role}-{identity}.lock")); + let file = File::options() + .create(true) + .read(true) + .write(true) + .open(path)?; + if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 { + bail!( + "resident {role} owner already exists for canonical store {}", + store.display() + ); + } + return Ok(ResidentProcessSingletonGuard { file }); + } + #[allow(unreachable_code)] + Err(anyhow!( + "resident process singleton is unsupported on this platform" + )) +} + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[cultcache( type = "epiphany.resident_cognition.provider_readiness", @@ -627,6 +718,21 @@ fn credential_file_ready(path: &Path) -> bool { mod tests { use super::*; + #[test] + fn resident_process_singleton_excludes_only_the_same_role_and_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = temp.path().join("resident.cc"); + fs::write(&store, [])?; + let owner = acquire_resident_process_singleton("resident-self", &store)?; + assert!(acquire_resident_process_singleton("resident-self", &store).is_err()); + let other_role = acquire_resident_process_singleton("heartbeat", &store)?; + let other_store = + acquire_resident_process_singleton("resident-self", temp.path().join("other.cc"))?; + drop((owner, other_role, other_store)); + assert!(acquire_resident_process_singleton("resident-self", &store).is_ok()); + Ok(()) + } + #[test] fn resident_readiness_reads_the_canonical_brake_store_truth() -> Result<()> { let temp = tempfile::tempdir()?;