use std::fs;
use std::fs::OpenOptions;
use std::hash::{Hash, Hasher};
use std::io::{Read, Write};
#[cfg(unix)]
use std::os::fd::AsRawFd;
#[cfg(windows)]
use std::os::windows::io::AsRawHandle;
#[cfg(windows)]
use std::os::windows::prelude::OsStrExt;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
use crate::DaemonWorkspaceConfig;
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
pub struct BuildIdentity {
pub version: String,
pub fingerprint: String,
}
pub fn current_build_identity(version: impl Into<String>) -> anyhow::Result<BuildIdentity> {
static FINGERPRINT: OnceLock<Result<String, String>> = OnceLock::new();
let fingerprint = FINGERPRINT.get_or_init(|| {
std::env::current_exe()
.map_err(|error| error.to_string())
.and_then(|path| binary_fingerprint(path).map_err(|error| error.to_string()))
});
Ok(BuildIdentity {
version: version.into(),
fingerprint: fingerprint
.clone()
.map_err(|error| anyhow::anyhow!("cannot fingerprint current binary: {error}"))?,
})
}
pub fn binary_fingerprint(path: impl AsRef<Path>) -> anyhow::Result<String> {
let path = path.as_ref();
let mut file = fs::File::open(path)
.map_err(|error| anyhow::anyhow!("cannot read binary {}: {error}", path.display()))?;
let mut buffer = [0_u8; 64 * 1024];
let mut hash = 0xcbf29ce484222325_u64;
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
for byte in &buffer[..read] {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x100000001b3);
}
}
Ok(format!("fnv1a64:{hash:016x}"))
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
pub struct DaemonRegistryEntry {
pub workspace_root: String,
pub workspace_roots: Vec<String>,
pub project: Option<String>,
pub cache_dir: Option<String>,
pub live_refresh: Option<String>,
pub endpoint: String,
pub token: String,
pub pid: u32,
#[serde(default)]
pub build: BuildIdentity,
#[serde(default)]
pub heartbeat_unix_ms: u64,
}
pub const DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS: u64 = 15_000;
pub fn registry_heartbeat_unix_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
pub fn daemon_registry_heartbeat_expired(entry: &DaemonRegistryEntry) -> bool {
entry.heartbeat_unix_ms == 0
|| registry_heartbeat_unix_ms().saturating_sub(entry.heartbeat_unix_ms)
> DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS
}
pub fn registry_dir() -> PathBuf {
std::env::var_os("CODE_MONIKER_REGISTRY_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| std::env::temp_dir().join("code-moniker-daemons"))
}
pub fn canonical_workspace_root(root: impl AsRef<Path>) -> anyhow::Result<PathBuf> {
let root = root.as_ref();
root.canonicalize()
.map_err(|err| anyhow::anyhow!("cannot canonicalize {}: {err}", root.display()))
}
pub fn canonical_workspace_roots<I, P>(roots: I) -> anyhow::Result<Vec<PathBuf>>
where
I: IntoIterator<Item = P>,
P: AsRef<Path>,
{
let mut canonical = Vec::new();
for root in roots {
let root = canonical_workspace_root(root)?;
if !canonical.contains(&root) {
canonical.push(root);
}
}
if canonical.is_empty() {
canonical.push(canonical_workspace_root(".")?);
}
Ok(canonical)
}
pub fn daemon_workspace_config<I, P>(
roots: I,
project: Option<String>,
cache_dir: Option<PathBuf>,
live_refresh: Option<String>,
) -> anyhow::Result<DaemonWorkspaceConfig>
where
I: IntoIterator<Item = P>,
P: AsRef<Path>,
{
let roots = canonical_workspace_roots(roots)?;
Ok(DaemonWorkspaceConfig {
roots: roots
.into_iter()
.map(|root| root.display().to_string())
.collect(),
project,
cache_dir: cache_dir
.map(normalize_path)
.transpose()?
.map(|path| path.display().to_string()),
live_refresh,
})
}
pub fn validate_daemon_start_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<()> {
let roots = config_roots(config);
if let Some(root) = roots.iter().find(|root| root.parent().is_none()) {
anyhow::bail!(
"refusing to start a code-moniker daemon at filesystem root `{}`; pass an explicit project directory (MCP configurations should use an absolute project path)",
root.display()
);
}
Ok(())
}
pub fn canonical_workspace_config(
config: DaemonWorkspaceConfig,
) -> anyhow::Result<DaemonWorkspaceConfig> {
daemon_workspace_config(
config.roots.iter().map(PathBuf::from),
config.project,
config.cache_dir.map(PathBuf::from),
config.live_refresh,
)
}
pub fn config_from_roots<I, P>(roots: I) -> anyhow::Result<DaemonWorkspaceConfig>
where
I: IntoIterator<Item = P>,
P: AsRef<Path>,
{
daemon_workspace_config(roots, None, None, Some("on-demand".to_string()))
}
pub fn registry_path_for_root(root: impl AsRef<Path>) -> anyhow::Result<PathBuf> {
registry_path_for_roots([root])
}
pub fn registry_path_for_roots<I, P>(roots: I) -> anyhow::Result<PathBuf>
where
I: IntoIterator<Item = P>,
P: AsRef<Path>,
{
registry_path_for_config(&config_from_roots(roots)?)
}
pub fn registry_path_for_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<PathBuf> {
let config = canonical_workspace_config(config.clone())?;
Ok(registry_dir().join(format!("{}.json", stable_config_hash(&config))))
}
pub fn daemon_log_path_for_config(config: &DaemonWorkspaceConfig) -> anyhow::Result<PathBuf> {
Ok(registry_path_for_config(config)?.with_extension("log"))
}
pub fn read_registry_entry(
config: &DaemonWorkspaceConfig,
) -> anyhow::Result<Option<DaemonRegistryEntry>> {
let path = registry_path_for_config(config)?;
match fs::read_to_string(&path) {
Ok(text) => parse_registry_entry(&path, &text).map(Some),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(err) => Err(anyhow::anyhow!(
"cannot read daemon registry entry {}: {err}",
path.display()
)),
}
}
pub fn write_registry_entry(
config: &DaemonWorkspaceConfig,
entry: &DaemonRegistryEntry,
) -> anyhow::Result<()> {
fs::create_dir_all(registry_dir())?;
atomic_write_registry_entry(®istry_path_for_config(config)?, entry)?;
Ok(())
}
pub fn claim_registry_entry(
config: &DaemonWorkspaceConfig,
entry: &DaemonRegistryEntry,
) -> anyhow::Result<bool> {
fs::create_dir_all(registry_dir())?;
claim_registry_file(®istry_path_for_config(config)?, entry)
}
fn claim_registry_file(path: &Path, entry: &DaemonRegistryEntry) -> anyhow::Result<bool> {
claim_registry_file_before_publish(path, entry, || {})
}
fn claim_registry_file_before_publish(
path: &Path,
entry: &DaemonRegistryEntry,
before_publish: impl FnOnce(),
) -> anyhow::Result<bool> {
let temp = path.with_extension(format!("{}.claim.tmp", entry.token));
let text = serde_json::to_vec_pretty(entry)?;
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)?;
let prepared = (|| {
file.write_all(&text)?;
file.sync_all()?;
Ok::<_, anyhow::Error>(())
})();
drop(file);
if let Err(error) = prepared {
let _ = fs::remove_file(temp);
return Err(error);
}
before_publish();
let result = publish_registry_claim(&temp, path);
let _ = fs::remove_file(temp);
result
}
#[cfg(not(windows))]
fn publish_registry_claim(source: &Path, destination: &Path) -> anyhow::Result<bool> {
match fs::hard_link(source, destination) {
Ok(()) => Ok(true),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false),
Err(error) => Err(error.into()),
}
}
#[cfg(windows)]
fn publish_registry_claim(source: &Path, destination: &Path) -> anyhow::Result<bool> {
use windows_sys::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW};
let source = source
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect::<Vec<_>>();
let destination = destination
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect::<Vec<_>>();
if unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_WRITE_THROUGH,
)
} != 0
{
return Ok(true);
}
let error = std::io::Error::last_os_error();
if error.kind() == std::io::ErrorKind::AlreadyExists {
Ok(false)
} else {
Err(error.into())
}
}
pub fn update_registry_entry_if_own(
config: &DaemonWorkspaceConfig,
entry: &DaemonRegistryEntry,
) -> anyhow::Result<bool> {
let path = registry_path_for_config(config)?;
with_registry_lock(&path, || {
let current = match fs::read_to_string(&path) {
Ok(text) => Some(parse_registry_entry(&path, &text)?),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
Err(error) => {
return Err(anyhow::anyhow!(
"cannot read daemon registry entry {}: {error}",
path.display()
));
}
};
let owned = current
.map(|current| current.token == entry.token && current.pid == entry.pid)
.unwrap_or(false);
if owned {
atomic_write_registry_entry(&path, entry)?;
}
Ok(owned)
})
}
fn parse_registry_entry(path: &Path, text: &str) -> anyhow::Result<DaemonRegistryEntry> {
serde_json::from_str(text).map_err(|error| {
anyhow::anyhow!(
"cannot decode daemon registry entry {}: {error}",
path.display()
)
})
}
fn atomic_write_registry_entry(path: &Path, entry: &DaemonRegistryEntry) -> anyhow::Result<()> {
let temp = path.with_extension(format!("{}.tmp", entry.token));
let text = serde_json::to_vec_pretty(entry)?;
{
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&temp)?;
file.write_all(&text)?;
file.sync_all()?;
}
replace_registry_file(&temp, path)?;
Ok(())
}
#[cfg(windows)]
fn replace_registry_file(source: &Path, destination: &Path) -> anyhow::Result<()> {
use windows_sys::Win32::Storage::FileSystem::{
MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW,
};
let source = source
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect::<Vec<_>>();
let destination = destination
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect::<Vec<_>>();
if unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
)
} == 0
{
return Err(std::io::Error::last_os_error().into());
}
Ok(())
}
#[cfg(not(windows))]
fn replace_registry_file(source: &Path, destination: &Path) -> anyhow::Result<()> {
fs::rename(source, destination)?;
Ok(())
}
pub fn remove_registry_entry_if_own(path: &Path, own: &DaemonRegistryEntry) {
let _ = with_registry_lock(path, || {
let current = fs::read_to_string(path)
.ok()
.and_then(|text| serde_json::from_str::<DaemonRegistryEntry>(&text).ok());
let owned = current
.map(|entry| entry.token == own.token && entry.pid == own.pid)
.unwrap_or(false);
if owned {
let _ = fs::remove_file(path);
}
Ok(())
});
}
#[cfg(unix)]
fn with_registry_lock<T>(
path: &Path,
action: impl FnOnce() -> anyhow::Result<T>,
) -> anyhow::Result<T> {
let lock_path = path.with_extension("lock");
let lock = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)?;
if unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX) } == -1 {
return Err(std::io::Error::last_os_error().into());
}
action()
}
#[cfg(not(unix))]
fn with_registry_lock<T>(
path: &Path,
action: impl FnOnce() -> anyhow::Result<T>,
) -> anyhow::Result<T> {
#[cfg(windows)]
{
use windows_sys::Win32::Storage::FileSystem::{
LOCKFILE_EXCLUSIVE_LOCK, LockFileEx, UnlockFileEx,
};
use windows_sys::Win32::System::IO::OVERLAPPED;
let lock_path = path.with_extension("lock");
let lock = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)?;
let mut overlapped = unsafe { std::mem::zeroed::<OVERLAPPED>() };
let handle = lock.as_raw_handle();
if unsafe {
LockFileEx(
handle,
LOCKFILE_EXCLUSIVE_LOCK,
0,
u32::MAX,
u32::MAX,
&mut overlapped,
)
} == 0
{
return Err(std::io::Error::last_os_error().into());
}
let result = action();
let unlock = unsafe { UnlockFileEx(handle, 0, u32::MAX, u32::MAX, &mut overlapped) };
if unlock == 0 && result.is_ok() {
return Err(std::io::Error::last_os_error().into());
}
result
}
#[cfg(not(windows))]
{
let _ = path;
action()
}
}
pub fn list_registry_files() -> anyhow::Result<Vec<(PathBuf, DaemonRegistryEntry)>> {
let dir = registry_dir();
if !dir.exists() {
return Ok(Vec::new());
}
let mut entries = Vec::new();
for entry in fs::read_dir(&dir)? {
let entry = entry?;
if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
continue;
}
let text = fs::read_to_string(entry.path())?;
if let Ok(registry) = serde_json::from_str::<DaemonRegistryEntry>(&text) {
entries.push((entry.path(), registry));
}
}
entries.sort_by(|(_, a), (_, b)| a.workspace_root.cmp(&b.workspace_root));
Ok(entries)
}
pub fn pid_is_alive(pid: u32) -> bool {
#[cfg(unix)]
{
let result = unsafe { libc::kill(pid as libc::pid_t, 0) };
let errno = (result != 0)
.then(|| std::io::Error::last_os_error().raw_os_error())
.flatten();
kill_result_means_alive(result, errno)
}
#[cfg(not(unix))]
{
#[cfg(windows)]
{
windows_pid_is_alive(pid)
}
#[cfg(not(windows))]
{
let _ = pid;
false
}
}
}
#[cfg(unix)]
fn kill_result_means_alive(result: i32, errno: Option<i32>) -> bool {
result == 0 || errno != Some(libc::ESRCH)
}
#[cfg(windows)]
fn windows_pid_is_alive(pid: u32) -> bool {
use windows_sys::Win32::Foundation::{
CloseHandle, ERROR_ACCESS_DENIED, GetLastError, WAIT_TIMEOUT,
};
use windows_sys::Win32::System::Threading::{
OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, WaitForSingleObject,
};
let handle = unsafe {
OpenProcess(
PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_SYNCHRONIZE,
0,
pid,
)
};
if handle.is_null() {
return unsafe { GetLastError() } == ERROR_ACCESS_DENIED;
}
let state = unsafe { WaitForSingleObject(handle, 0) };
unsafe {
CloseHandle(handle);
}
state == WAIT_TIMEOUT
}
pub fn list_registry_entries() -> anyhow::Result<Vec<DaemonRegistryEntry>> {
let mut entries = Vec::new();
for (path, entry) in list_registry_files()? {
if pid_is_alive(entry.pid) {
entries.push(entry);
} else {
remove_registry_entry_if_own(&path, &entry);
}
}
entries.sort_by(|a, b| a.workspace_root.cmp(&b.workspace_root));
Ok(entries)
}
pub fn config_roots(config: &DaemonWorkspaceConfig) -> Vec<PathBuf> {
config.roots.iter().map(PathBuf::from).collect()
}
pub fn workspace_label(roots: &[PathBuf]) -> String {
if roots.len() == 1 {
roots[0].display().to_string()
} else {
roots
.iter()
.map(|root| root.display().to_string())
.collect::<Vec<_>>()
.join(";")
}
}
fn normalize_path(path: PathBuf) -> anyhow::Result<PathBuf> {
if path.is_absolute() {
Ok(path)
} else {
Ok(std::env::current_dir()?.join(path))
}
}
fn stable_config_hash(config: &DaemonWorkspaceConfig) -> String {
let mut hasher = StableHasher::default();
for root in &config.roots {
root.hash(&mut hasher);
0xff_u8.hash(&mut hasher);
}
config.project.hash(&mut hasher);
0xfe_u8.hash(&mut hasher);
config.cache_dir.hash(&mut hasher);
format!("{:016x}", hasher.finish())
}
#[derive(Default)]
struct StableHasher(u64);
impl Hasher for StableHasher {
fn finish(&self) -> u64 {
self.0
}
fn write(&mut self, bytes: &[u8]) {
let mut hash = if self.0 == 0 {
0xcbf29ce484222325
} else {
self.0
};
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x100000001b3);
}
self.0 = hash;
}
}
#[cfg(test)]
mod tests {
use super::*;
fn entry(token: &str, pid: u32) -> DaemonRegistryEntry {
DaemonRegistryEntry {
workspace_root: "/tmp/ws".to_string(),
workspace_roots: vec!["/tmp/ws".to_string()],
project: None,
cache_dir: None,
live_refresh: None,
endpoint: "127.0.0.1:1".to_string(),
token: token.to_string(),
pid,
build: BuildIdentity::default(),
heartbeat_unix_ms: registry_heartbeat_unix_ms(),
}
}
#[test]
fn registry_identity_ignores_the_refresh_mode() {
let base = DaemonWorkspaceConfig {
roots: vec!["/tmp/ws".to_string()],
project: None,
cache_dir: None,
live_refresh: Some("auto".to_string()),
};
let mut on_demand = base.clone();
on_demand.live_refresh = Some("on-demand".to_string());
assert_eq!(
stable_config_hash(&base),
stable_config_hash(&on_demand),
"one workspace must map to one registry slot, whatever the refresh mode"
);
let mut other_project = base.clone();
other_project.project = Some("api".to_string());
assert_ne!(
stable_config_hash(&base),
stable_config_hash(&other_project),
"what gets indexed still separates registry slots"
);
}
#[test]
fn malformed_registry_entry_is_not_treated_as_missing_or_unowned() {
let workspace = tempfile::tempdir().expect("workspace");
let config = config_from_roots([workspace.path()]).expect("workspace config");
let path = registry_path_for_config(&config).expect("registry path");
fs::create_dir_all(path.parent().expect("registry directory"))
.expect("create registry directory");
fs::write(&path, "{not-json").expect("write malformed registry entry");
let read_error = read_registry_entry(&config)
.expect_err("malformed registry entry must be a typed read failure");
assert!(
read_error
.to_string()
.contains("cannot decode daemon registry entry"),
"{read_error:#}"
);
let update_error = update_registry_entry_if_own(&config, &entry("owner", 111))
.expect_err("malformed registry entry must not look like an unowned claim");
assert!(
update_error
.to_string()
.contains("cannot decode daemon registry entry"),
"{update_error:#}"
);
let _ = fs::remove_file(&path);
let _ = fs::remove_file(path.with_extension("lock"));
}
#[cfg(unix)]
#[test]
fn daemon_workspace_rejects_the_filesystem_root() {
let config =
daemon_workspace_config([Path::new("/")], None, None, Some("auto".to_string()))
.expect("filesystem root identity remains available for status and cleanup");
let error = validate_daemon_start_config(&config)
.expect_err("filesystem root must fail before daemon startup");
let message = error.to_string();
assert!(message.contains("refusing to start"), "{message}");
assert!(message.contains("absolute project path"), "{message}");
}
#[test]
fn shutdown_removal_spares_a_successor_entry() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("ws.json");
let old = entry("old-token", 111);
let new = entry("new-token", 222);
fs::write(&path, serde_json::to_string(&new).expect("json")).expect("write");
remove_registry_entry_if_own(&path, &old);
assert!(path.exists(), "the successor's entry must survive");
remove_registry_entry_if_own(&path, &new);
assert!(!path.exists(), "the owner removes its own entry");
remove_registry_entry_if_own(&path, &new);
}
#[test]
fn legacy_registry_entries_default_missing_build_identity() {
let mut value = serde_json::to_value(entry("legacy", 111)).expect("json");
value.as_object_mut().expect("object").remove("build");
let decoded: DaemonRegistryEntry = serde_json::from_value(value).expect("legacy entry");
assert_eq!(decoded.build, BuildIdentity::default());
}
#[test]
fn binary_fingerprint_changes_with_binary_content() {
let temp = tempfile::tempdir().expect("tempdir");
let binary = temp.path().join("code-moniker");
fs::write(&binary, b"old build").expect("old build");
let old = binary_fingerprint(&binary).expect("old fingerprint");
fs::write(&binary, b"new build").expect("new build");
let new = binary_fingerprint(&binary).expect("new fingerprint");
assert_ne!(old, new);
assert!(old.starts_with("fnv1a64:"));
assert!(new.starts_with("fnv1a64:"));
}
#[test]
fn missing_or_old_heartbeat_expires_but_a_fresh_claim_does_not() {
let mut registry = entry("heartbeat", 111);
registry.heartbeat_unix_ms = 0;
assert!(daemon_registry_heartbeat_expired(®istry));
registry.heartbeat_unix_ms =
registry_heartbeat_unix_ms() - DAEMON_REGISTRY_HEARTBEAT_TIMEOUT_MS - 1;
assert!(daemon_registry_heartbeat_expired(®istry));
registry.heartbeat_unix_ms = registry_heartbeat_unix_ms();
assert!(!daemon_registry_heartbeat_expired(®istry));
}
#[test]
fn atomic_registry_update_replaces_a_complete_entry() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("workspace.json");
let initial = entry("same-daemon", 111);
atomic_write_registry_entry(&path, &initial).expect("write initial entry");
let updated = DaemonRegistryEntry {
heartbeat_unix_ms: initial.heartbeat_unix_ms + 1,
..initial.clone()
};
atomic_write_registry_entry(&path, &updated).expect("write updated entry");
let read: DaemonRegistryEntry =
serde_json::from_str(&fs::read_to_string(path).expect("read entry")).expect("json");
assert_eq!(read, updated);
}
#[test]
fn registry_claim_keeps_the_final_path_hidden_until_publication() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("workspace.json");
let owner = entry("owner-token", 111);
let timeout = std::time::Duration::from_secs(5);
let (prepared_tx, prepared_rx) = std::sync::mpsc::sync_channel(1);
let (publish_tx, publish_rx) = std::sync::mpsc::sync_channel(1);
let claim_path = path.clone();
let claim_owner = owner.clone();
let claimant = std::thread::spawn(move || {
claim_registry_file_before_publish(&claim_path, &claim_owner, || {
prepared_tx.send(()).expect("signal prepared claim");
publish_rx
.recv_timeout(timeout)
.expect("receive publication release");
})
});
prepared_rx
.recv_timeout(timeout)
.expect("claimant prepares the complete temporary entry");
assert!(
!path.exists(),
"the final registry path must stay hidden while its complete temporary is pending"
);
publish_tx.send(()).expect("release registry publication");
assert!(
claimant
.join()
.expect("join claimant")
.expect("claim registry"),
"the first complete entry must win the registry claim"
);
let published: DaemonRegistryEntry = serde_json::from_str(
&fs::read_to_string(&path).expect("read published registry entry"),
)
.expect("published registry entry is complete JSON");
assert_eq!(published, owner);
}
#[test]
fn concurrent_registry_claims_publish_exactly_one_complete_owner() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("workspace.json");
let first = entry("first-token", 111);
let second = entry("second-token", 222);
let timeout = std::time::Duration::from_secs(5);
let (prepared_tx, prepared_rx) = std::sync::mpsc::sync_channel(2);
let spawn_claimant = |claim: DaemonRegistryEntry| {
let claim_path = path.clone();
let claimant_prepared_tx = prepared_tx.clone();
let (publish_tx, publish_rx) = std::sync::mpsc::sync_channel(1);
let claimant = std::thread::spawn(move || {
claim_registry_file_before_publish(&claim_path, &claim, || {
claimant_prepared_tx
.send(())
.expect("signal prepared claim");
publish_rx
.recv_timeout(timeout)
.expect("receive publication release");
})
});
(claimant, publish_tx)
};
let (first_claimant, first_publish_tx) = spawn_claimant(first.clone());
let (second_claimant, second_publish_tx) = spawn_claimant(second.clone());
for _ in 0..2 {
prepared_rx
.recv_timeout(timeout)
.expect("claimant prepares a complete temporary entry");
}
assert!(
!path.exists(),
"neither prepared claimant may expose the final path before publication"
);
first_publish_tx
.send(())
.expect("release first registry publication");
second_publish_tx
.send(())
.expect("release second registry publication");
let first_won = first_claimant
.join()
.expect("join first claimant")
.expect("first claim");
let second_won = second_claimant
.join()
.expect("join second claimant")
.expect("second claim");
assert_ne!(first_won, second_won, "exactly one claimant must win");
let published: DaemonRegistryEntry = serde_json::from_str(
&fs::read_to_string(&path).expect("read published registry entry"),
)
.expect("published registry entry is complete JSON");
assert_eq!(published, if first_won { first } else { second });
assert_eq!(
fs::read_dir(dir.path()).expect("list registry dir").count(),
1,
"claim publication must not leave temporary files"
);
}
#[cfg(unix)]
#[test]
fn permission_denied_pid_is_alive_but_missing_pid_is_dead() {
assert!(kill_result_means_alive(-1, Some(libc::EPERM)));
assert!(!kill_result_means_alive(-1, Some(libc::ESRCH)));
assert!(kill_result_means_alive(0, None));
}
#[cfg(windows)]
#[test]
fn current_windows_process_is_alive() {
assert!(pid_is_alive(std::process::id()));
}
#[test]
fn registry_directory_honors_the_environment_override() {
let expected = tempfile::tempdir()
.expect("registry tempdir")
.path()
.join("custom-registry");
let status = std::process::Command::new(std::env::current_exe().expect("test binary"))
.args([
"--exact",
"discovery::tests::registry_directory_environment_child",
"--ignored",
])
.env("CODE_MONIKER_REGISTRY_DIR", &expected)
.env("CODE_MONIKER_REGISTRY_TEST_EXPECTED", &expected)
.status()
.expect("run registry environment child");
assert!(status.success());
}
#[test]
#[ignore = "subprocess fixture"]
fn registry_directory_environment_child() {
let Some(expected) = std::env::var_os("CODE_MONIKER_REGISTRY_TEST_EXPECTED") else {
return;
};
assert_eq!(registry_dir(), PathBuf::from(expected));
}
#[test]
fn registry_lock_serializes_processes() {
let dir = tempfile::tempdir().expect("tempdir");
let target = dir.path().join("workspace.json");
let started = dir.path().join("child-started");
let acquired = dir.path().join("child-acquired");
let mut child = None;
with_registry_lock(&target, || {
child = Some(
std::process::Command::new(std::env::current_exe()?)
.args([
"--exact",
"discovery::tests::registry_lock_child",
"--nocapture",
])
.env("CODE_MONIKER_LOCK_TEST_TARGET", &target)
.env("CODE_MONIKER_LOCK_TEST_STARTED", &started)
.env("CODE_MONIKER_LOCK_TEST_ACQUIRED", &acquired)
.spawn()?,
);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
while !started.exists() && std::time::Instant::now() < deadline {
std::thread::sleep(std::time::Duration::from_millis(10));
}
assert!(started.exists(), "child process did not reach the lock");
std::thread::sleep(std::time::Duration::from_millis(200));
assert!(
!acquired.exists(),
"child acquired the registry lock before its owner released it"
);
Ok(())
})
.expect("hold parent lock");
let status = child
.expect("child process")
.wait()
.expect("wait for child");
assert!(status.success(), "child lock process failed: {status}");
assert!(acquired.exists(), "child never acquired the released lock");
}
#[test]
fn registry_lock_child() {
let (Some(target), Some(started), Some(acquired)) = (
std::env::var_os("CODE_MONIKER_LOCK_TEST_TARGET"),
std::env::var_os("CODE_MONIKER_LOCK_TEST_STARTED"),
std::env::var_os("CODE_MONIKER_LOCK_TEST_ACQUIRED"),
) else {
return;
};
fs::write(&started, b"started").expect("announce child");
with_registry_lock(Path::new(&target), || {
fs::write(&acquired, b"acquired")?;
Ok(())
})
.expect("acquire child lock");
}
}