use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use crate::core::config::base_dir;
pub const REGISTRATION_KIND: &str = "session";
fn default_kind() -> String {
REGISTRATION_KIND.to_string()
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionRegistration {
#[serde(default = "default_kind")]
pub kind: String,
pub session_id: String,
pub name: Option<String>,
pub socket_path: String,
pub pid: u32,
pub started_at: DateTime<Utc>,
}
pub fn registry_dir() -> PathBuf {
let dir = std::env::var_os("SYNAPS_RUNTIME_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| base_dir().join("run"));
if let Err(e) = std::fs::create_dir_all(&dir) {
tracing::warn!("registry: failed to create run dir {:?}: {}", dir, e);
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700));
}
std::fs::canonicalize(&dir).unwrap_or(dir)
}
pub fn sanitize_session_id(raw: &str) -> String {
raw.chars()
.map(|c| {
if c.is_alphanumeric() || c == '-' || c == '_' {
c
} else {
'_'
}
})
.collect::<String>()
}
pub fn socket_path_for_session(session_id: &str) -> String {
socket_path_in_dir(®istry_dir(), session_id)
}
fn socket_path_in_dir(dir: &std::path::Path, session_id: &str) -> String {
let safe_id = sanitize_session_id(session_id);
dir.join(format!("{}.sock", safe_id))
.to_string_lossy()
.into_owned()
}
pub(crate) fn lock_dir(dir: &std::path::Path) -> Result<std::fs::File, String> {
let mut options = std::fs::OpenOptions::new();
options.read(true).write(true).create(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options
.mode(0o600)
.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
}
let file = options
.open(dir.join(".session-registry.lock"))
.map_err(|e| e.to_string())?;
if !file.metadata().map_err(|e| e.to_string())?.is_file() {
return Err("registry lock is not a regular file".into());
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
let meta = file.metadata().map_err(|e| e.to_string())?;
if meta.uid() != unsafe { libc::geteuid() }
|| meta.nlink() != 1
|| meta.mode() & 0o777 != 0o600
{
return Err("unsafe registry lock owner, links or permissions".into());
}
}
let deadline = std::time::Instant::now() + std::time::Duration::from_millis(100);
loop {
if fs4::fs_std::FileExt::try_lock_exclusive(&file).map_err(|e| e.to_string())? {
return Ok(file);
}
if std::time::Instant::now() >= deadline {
return Err("registry lock busy".into());
}
std::thread::sleep(std::time::Duration::from_millis(2));
}
}
fn same_owner(a: &SessionRegistration, b: &SessionRegistration) -> bool {
a.session_id == b.session_id
&& a.pid == b.pid
&& a.started_at == b.started_at
&& a.socket_path == b.socket_path
}
fn generated_endpoint_missing(dir: &std::path::Path, reg: &SessionRegistration) -> bool {
#[cfg(unix)]
{
let path = dir.join(generated_basename(reg));
std::path::Path::new(®.socket_path) == path
&& std::fs::symlink_metadata(path)
.is_err_and(|e| e.kind() == std::io::ErrorKind::NotFound)
}
#[cfg(not(unix))]
{
let _ = (dir, reg);
false
}
}
fn generated_basename(reg: &SessionRegistration) -> String {
use sha2::{Digest, Sha256};
let identity = format!("{}\0{}\0{}", reg.session_id, reg.pid, reg.started_at);
let digest = Sha256::digest(identity.as_bytes());
let hex: String = digest[..16].iter().map(|b| format!("{b:02x}")).collect();
format!("s-{hex}.sock")
}
fn regular_card(path: &std::path::Path) -> bool {
!is_reserved_json(path)
&& std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_file())
}
fn valid_card(path: &std::path::Path, reg: &SessionRegistration) -> bool {
reg.kind == REGISTRATION_KIND
&& !is_reserved_json(path)
&& path.file_name()
== Some(std::ffi::OsStr::new(&format!(
"{}.json",
sanitize_session_id(®.session_id)
)))
&& std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_file())
}
fn remove_socket(dir: &std::path::Path, reg: &SessionRegistration) {
let sock = std::path::Path::new(®.socket_path);
let legacy = format!("{}.sock", sanitize_session_id(®.session_id));
if (sock == dir.join(generated_basename(reg)) || sock == dir.join(legacy))
&& sock.parent() == Some(dir)
{
super::socket::cleanup_socket(®.socket_path);
}
}
pub struct SessionEndpoint {
pub registration: SessionRegistration,
dir: PathBuf,
shutdown: std::sync::Arc<std::sync::atomic::AtomicBool>,
task: tokio::task::JoinHandle<()>,
socket: super::socket::SocketIdentity,
}
impl SessionEndpoint {
pub fn start(
id: &str,
name: Option<String>,
queue: std::sync::Arc<super::queue::EventQueue>,
) -> Result<Self, String> {
let dir = std::fs::canonicalize(registry_dir()).map_err(|e| e.to_string())?;
Self::start_in(&dir, id, name, queue)
}
fn start_in(
dir: &std::path::Path,
id: &str,
name: Option<String>,
queue: std::sync::Arc<super::queue::EventQueue>,
) -> Result<Self, String> {
let _lock = lock_dir(dir)?;
scan_registrations_locked(dir, None);
let mut registration = SessionRegistration {
kind: REGISTRATION_KIND.into(),
session_id: id.into(),
name,
socket_path: String::new(),
pid: std::process::id(),
started_at: Utc::now(),
};
registration.socket_path = dir
.join(generated_basename(®istration))
.to_string_lossy()
.into_owned();
let (listener, socket) = super::socket::bind_session_socket(®istration.socket_path)
.map_err(|e| e.to_string())?;
if let Err(e) = write_registration(®istration, dir) {
socket.remove();
return Err(e);
}
let shutdown = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let task = super::socket::serve_session_socket(
listener,
registration.socket_path.clone(),
queue,
shutdown.clone(),
);
Ok(Self {
registration,
dir: dir.to_path_buf(),
shutdown,
task,
socket,
})
}
pub fn update_name(&self, name: Option<&str>) -> Result<(), String> {
let _lock = lock_dir(&self.dir)?;
let path = self.dir.join(format!(
"{}.json",
sanitize_session_id(&self.registration.session_id)
));
if !regular_card(&path) {
return Err("session card missing".into());
}
let mut disk: SessionRegistration =
serde_json::from_str(&std::fs::read_to_string(&path).map_err(|e| e.to_string())?)
.map_err(|e| e.to_string())?;
if !valid_card(&path, &disk) || !same_owner(&disk, &self.registration) {
return Err("session generation replaced".into());
}
disk.name = name.map(str::to_owned);
write_registration(&disk, &self.dir)
}
pub fn shutdown(&self) {
self.shutdown
.store(true, std::sync::atomic::Ordering::Release);
self.task.abort();
self.socket.remove();
let Ok(_lock) = lock_dir(&self.dir) else {
return;
};
unregister_owned_locked(&self.registration, &self.dir, false);
}
}
impl Drop for SessionEndpoint {
fn drop(&mut self) {
self.shutdown();
}
}
pub fn register_session(reg: &SessionRegistration) -> Result<(), String> {
register_session_in(reg, ®istry_dir())
}
fn register_session_in(reg: &SessionRegistration, dir: &std::path::Path) -> Result<(), String> {
let _lock = lock_dir(dir)?;
write_registration(reg, dir)
}
fn write_registration(reg: &SessionRegistration, dir: &std::path::Path) -> Result<(), String> {
use std::io::Write;
let safe_id = sanitize_session_id(®.session_id);
let path = dir.join(format!("{}.json", safe_id));
if is_reserved_json(&path) {
return Err("reserved session id".into());
}
match std::fs::symlink_metadata(&path) {
Ok(meta) => {
if !meta.file_type().is_file() {
return Err("not a regular session card".into());
}
let old: SessionRegistration =
serde_json::from_str(&std::fs::read_to_string(&path).map_err(|e| e.to_string())?)
.map_err(|e| e.to_string())?;
if !valid_card(&path, &old) || old.session_id != reg.session_id {
return Err("refusing to replace a foreign session card".into());
}
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e.to_string()),
}
let tmp = dir.join(format!(".session-{}.tmp", uuid::Uuid::new_v4().simple()));
let json = serde_json::to_string(reg).map_err(|e| format!("serialize error: {}", e))?;
let mut options = std::fs::OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
let result = (|| {
let mut file = options.open(&tmp)?;
file.write_all(json.as_bytes())?;
drop(file);
std::fs::rename(&tmp, &path)
})();
if result.is_err() {
let _ = std::fs::remove_file(&tmp);
}
result.map_err(|e: std::io::Error| e.to_string())?;
Ok(())
}
pub fn update_session_name(session_id: &str, name: Option<&str>) -> Result<(), String> {
update_session_name_in(session_id, name, ®istry_dir())
}
fn update_session_name_in(
session_id: &str,
name: Option<&str>,
dir: &std::path::Path,
) -> Result<(), String> {
let _lock = lock_dir(dir)?;
let safe_id = sanitize_session_id(session_id);
let path = dir.join(format!("{}.json", safe_id));
if !regular_card(&path) {
return Err("not a regular session card".into());
}
let content =
std::fs::read_to_string(&path).map_err(|e| format!("read {}: {}", path.display(), e))?;
let mut reg: SessionRegistration =
serde_json::from_str(&content).map_err(|e| format!("parse {}: {}", path.display(), e))?;
if !valid_card(&path, ®) {
return Err("not a session card".into());
}
reg.name = name.map(str::to_string);
write_registration(®, dir)
}
pub fn unregister_session(session_id: &str) {
unregister_session_in(session_id, ®istry_dir());
}
fn unregister_session_in(session_id: &str, dir: &std::path::Path) {
let Ok(_lock) = lock_dir(dir) else {
return;
};
let path = dir.join(format!("{}.json", sanitize_session_id(session_id)));
if !regular_card(&path) {
return;
}
if let Ok(content) = std::fs::read_to_string(&path) {
if let Ok(reg) = serde_json::from_str::<SessionRegistration>(&content) {
if valid_card(&path, ®) {
remove_registration_files(dir, &path, ®);
}
}
}
}
pub fn unregister_owned(reg: &SessionRegistration) {
unregister_owned_in(reg, ®istry_dir());
}
fn unregister_owned_in(reg: &SessionRegistration, dir: &std::path::Path) {
let Ok(_lock) = lock_dir(dir) else {
return;
};
unregister_owned_locked(reg, dir, true);
}
fn unregister_owned_locked(
reg: &SessionRegistration,
dir: &std::path::Path,
remove_claimed_socket: bool,
) {
let safe_id = sanitize_session_id(®.session_id);
let path = dir.join(format!("{}.json", safe_id));
if !regular_card(&path) {
return;
}
let Ok(content) = std::fs::read_to_string(&path) else {
return;
};
let Ok(on_disk) = serde_json::from_str::<SessionRegistration>(&content) else {
return;
};
if !valid_card(&path, &on_disk)
|| on_disk.session_id != reg.session_id
|| on_disk.pid != reg.pid
|| on_disk.started_at != reg.started_at
|| on_disk.socket_path != reg.socket_path
{
tracing::debug!(session = %reg.session_id, "registry: registration was replaced; leaving the newer one");
return;
}
if remove_claimed_socket {
remove_socket(dir, &on_disk);
}
let _ = std::fs::remove_file(path);
}
fn remove_registration_files(
dir: &std::path::Path,
path: &std::path::Path,
reg: &SessionRegistration,
) {
if !valid_card(path, reg) {
return;
}
remove_socket(dir, reg);
let _ = std::fs::remove_file(path);
}
pub fn sweep_stale_registrations(own_pid: Option<u32>) -> usize {
sweep_stale_registrations_in(®istry_dir(), own_pid)
}
fn sweep_stale_registrations_in(dir: &std::path::Path, own_pid: Option<u32>) -> usize {
scan_registrations(dir, own_pid).1
}
fn pid_is_alive(pid: u32) -> bool {
if pid == 0 || pid > i32::MAX as u32 {
return false;
}
#[cfg(unix)]
{
let result = unsafe { libc::kill(pid as libc::pid_t, 0) };
result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}
#[cfg(not(unix))]
{
let _ = pid;
true
}
}
fn is_reserved_json(path: &std::path::Path) -> bool {
path.file_stem()
.and_then(|s| s.to_str())
.is_some_and(|stem| stem == "daemon" || stem.starts_with("daemon-"))
}
pub fn list_active_sessions() -> Vec<SessionRegistration> {
list_active_sessions_in(®istry_dir())
}
fn list_active_sessions_in(dir: &std::path::Path) -> Vec<SessionRegistration> {
scan_registrations(dir, None).0
}
fn scan_registrations(
dir: &std::path::Path,
own_pid: Option<u32>,
) -> (Vec<SessionRegistration>, usize) {
let Ok(_lock) = lock_dir(dir) else {
return (Vec::new(), 0);
};
scan_registrations_locked(dir, own_pid)
}
fn scan_registrations_locked(
dir: &std::path::Path,
own_pid: Option<u32>,
) -> (Vec<SessionRegistration>, usize) {
let Ok(entries) = std::fs::read_dir(dir) else {
return (Vec::new(), 0);
};
let mut live = Vec::new();
let mut removed = 0;
for entry in entries.flatten() {
let path = entry.path();
if !path.extension().is_some_and(|e| e == "json") || is_reserved_json(&path) {
continue;
}
if !regular_card(&path) {
continue;
}
let Ok(content) = std::fs::read_to_string(&path) else {
continue;
};
let reg = match serde_json::from_str::<SessionRegistration>(&content) {
Ok(reg) if reg.kind == REGISTRATION_KIND => reg,
Ok(reg) => {
tracing::debug!(path = %path.display(), kind = %reg.kind, "registry: not a session registration — skipped");
continue;
}
Err(e) => {
tracing::debug!(path = %path.display(), error = %e, "registry: unparseable json — skipped, not ours");
continue;
}
};
if !valid_card(&path, ®) {
continue;
}
if own_pid != Some(reg.pid)
&& pid_is_alive(reg.pid)
&& !generated_endpoint_missing(dir, ®)
{
live.push(reg);
} else {
remove_registration_files(dir, &path, ®);
removed += 1;
}
}
(live, removed)
}
pub fn find_session_registration(query: &str) -> Option<SessionRegistration> {
find_session_registration_in(query, ®istry_dir())
}
fn find_session_registration_in(query: &str, dir: &std::path::Path) -> Option<SessionRegistration> {
let sessions = list_active_sessions_in(dir);
if let Some(reg) = sessions.iter().find(|r| r.session_id == query) {
return Some(reg.clone());
}
if let Some(reg) = sessions.iter().find(|r| r.name.as_deref() == Some(query)) {
return Some(reg.clone());
}
let matches: Vec<_> = sessions
.iter()
.filter(|r| r.session_id.starts_with(query))
.collect();
if matches.len() == 1 {
Some(matches[0].clone())
} else {
None
}
}
#[cfg(test)]
mod tests {
use super::*;
use serial_test::serial;
use tempfile::TempDir;
#[cfg(unix)]
#[tokio::test]
async fn locked_drop_removes_endpoint_and_next_scan_repairs_live_pid_card() {
let tmp = tmp_registry();
let endpoint = SessionEndpoint::start_in(
tmp.path(),
"locked",
None,
std::sync::Arc::new(super::super::queue::EventQueue::new(10)),
)
.unwrap();
let path = endpoint.registration.socket_path.clone();
let lock = lock_dir(tmp.path()).unwrap();
drop(endpoint);
assert!(!std::path::Path::new(&path).exists());
assert!(tmp.path().join("locked.json").exists());
drop(lock);
assert_eq!(sweep_stale_registrations_in(tmp.path(), None), 1);
assert!(!tmp.path().join("locked.json").exists());
}
#[tokio::test]
async fn owned_rename_refuses_replacement_and_uses_captured_root() {
let tmp = tmp_registry();
let other = tmp_registry();
let queue = std::sync::Arc::new(super::super::queue::EventQueue::new(10));
let old = SessionEndpoint::start_in(tmp.path(), "same", None, queue.clone()).unwrap();
let new = SessionEndpoint::start_in(tmp.path(), "same", None, queue.clone()).unwrap();
let unrelated = SessionEndpoint::start_in(other.path(), "same", None, queue).unwrap();
assert!(old.update_name(Some("wrong")).is_err());
new.update_name(Some("right")).unwrap();
assert_eq!(
list_active_sessions_in(tmp.path())[0].name.as_deref(),
Some("right")
);
assert!(list_active_sessions_in(other.path())[0].name.is_none());
drop(new);
assert!(other.path().join("same.json").exists());
drop(unrelated);
}
#[tokio::test]
async fn publication_failure_rolls_back_bound_endpoint() {
let tmp = tmp_registry();
assert!(SessionEndpoint::start_in(
tmp.path(),
"daemon",
None,
std::sync::Arc::new(super::super::queue::EventQueue::new(10))
)
.is_err());
let names: Vec<_> = std::fs::read_dir(tmp.path())
.unwrap()
.map(|e| e.unwrap().file_name())
.collect();
assert_eq!(
names,
vec![std::ffi::OsString::from(".session-registry.lock")]
);
}
#[cfg(unix)]
#[tokio::test]
async fn replacement_survives_old_task_ending_and_immediate_drop() {
let tmp = tmp_registry();
let queue = std::sync::Arc::new(super::super::queue::EventQueue::new(10));
let old = SessionEndpoint::start_in(tmp.path(), "same", None, queue.clone()).unwrap();
let old_path = old.registration.socket_path.clone();
let new = SessionEndpoint::start_in(tmp.path(), "same", None, queue).unwrap();
assert_ne!(old_path, new.registration.socket_path);
update_session_name_in("same", Some("renamed"), tmp.path()).unwrap();
drop(old);
assert!(!std::path::Path::new(&old_path).exists());
tokio::task::yield_now().await;
let _client = tokio::net::UnixStream::connect(&new.registration.socket_path)
.await
.unwrap();
let new_path = new.registration.socket_path.clone();
assert_eq!(
list_active_sessions_in(tmp.path())[0].name.as_deref(),
Some("renamed")
);
drop(new);
assert!(!std::path::Path::new(&new_path).exists());
assert!(!tmp.path().join("same.json").exists());
}
#[test]
fn lock_is_permanent_and_contention_is_bounded() {
let tmp = tmp_registry();
let lock = lock_dir(tmp.path()).unwrap();
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let ready = barrier.clone();
let dir = tmp.path().to_path_buf();
let contender = std::thread::spawn(move || {
ready.wait();
assert!(lock_dir(&dir).is_err());
});
barrier.wait();
contender.join().unwrap();
drop(lock);
assert!(tmp.path().join(".session-registry.lock").exists());
assert!(lock_dir(tmp.path()).is_ok());
}
#[cfg(unix)]
#[tokio::test]
async fn stale_sweep_and_registration_are_serialized() {
let tmp = tmp_registry();
reg_with_socket(tmp.path(), "same", 0);
let lock = lock_dir(tmp.path()).unwrap();
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let ready = barrier.clone();
let dir = tmp.path().to_path_buf();
let sweep = std::thread::spawn(move || {
ready.wait();
assert_eq!(
sweep_stale_registrations_in(&dir, Some(std::process::id())),
0
);
});
barrier.wait();
sweep.join().unwrap();
drop(lock);
let endpoint = SessionEndpoint::start_in(
tmp.path(),
"same",
None,
std::sync::Arc::new(super::super::queue::EventQueue::new(10)),
)
.unwrap();
assert_eq!(sweep_stale_registrations_in(tmp.path(), None), 0);
let _client = tokio::net::UnixStream::connect(&endpoint.registration.socket_path)
.await
.unwrap();
}
#[cfg(unix)]
#[test]
fn malicious_socket_claims_leave_user_files_and_other_sockets() {
let tmp = tmp_registry();
let outside = tmp_registry();
let outside_path = outside.path().join("victim.sock");
let _outside = std::os::unix::net::UnixListener::bind(&outside_path).unwrap();
let unrelated = tmp.path().join("other.sock");
let _other = std::os::unix::net::UnixListener::bind(&unrelated).unwrap();
let regular = tmp.path().join("victim.sock");
std::fs::write(®ular, b"user data").unwrap();
std::os::unix::fs::symlink(outside.path(), tmp.path().join("link")).unwrap();
for path in [
outside_path.clone(),
unrelated.clone(),
regular.clone(),
tmp.path().join("link/victim.sock"),
tmp.path().join("../victim.sock"),
] {
let mut reg = make_reg("victim", None, 0);
reg.socket_path = path.to_string_lossy().into_owned();
register_session_in(®, tmp.path()).unwrap();
sweep_stale_registrations_in(tmp.path(), None);
}
assert!(outside_path.exists());
assert!(unrelated.exists());
assert_eq!(std::fs::read(regular).unwrap(), b"user data");
}
#[cfg(windows)]
#[tokio::test]
async fn replacement_pipe_is_bound_before_publication() {
let tmp = tmp_registry();
let queue = std::sync::Arc::new(super::super::queue::EventQueue::new(10));
let old = SessionEndpoint::start_in(tmp.path(), "same", None, queue.clone()).unwrap();
let new = SessionEndpoint::start_in(tmp.path(), "same", None, queue).unwrap();
drop(old);
let pipe = super::super::socket::pipe_name_for(&new.registration.socket_path);
let _client = tokio::net::windows::named_pipe::ClientOptions::new()
.open(pipe)
.unwrap();
}
fn tmp_registry() -> TempDir {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path()).unwrap();
dir
}
fn make_reg(id: &str, name: Option<&str>, pid: u32) -> SessionRegistration {
SessionRegistration {
kind: REGISTRATION_KIND.to_string(),
session_id: id.to_string(),
name: name.map(|s| s.to_string()),
socket_path: format!("/unused/{id}.sock"),
pid,
started_at: Utc::now(),
}
}
fn dir_buf(tmp: &TempDir) -> PathBuf {
tmp.path().to_path_buf()
}
#[test]
fn register_creates_file() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("abc-1234", None, std::process::id());
register_session_in(®, &dir).unwrap();
assert!(dir.join("abc-1234.json").exists());
}
#[test]
fn list_returns_live_sessions() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let pid = std::process::id();
let reg = make_reg("live-0001", Some("my-agent"), pid);
register_session_in(®, &dir).unwrap();
let sessions = list_active_sessions_in(&dir);
assert!(sessions.iter().any(|r| r.session_id == "live-0001"));
}
#[test]
fn find_by_exact_id() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("find-exact-01", None, std::process::id());
register_session_in(®, &dir).unwrap();
let found = find_session_registration_in("find-exact-01", &dir);
assert!(found.is_some());
assert_eq!(found.unwrap().session_id, "find-exact-01");
}
#[test]
fn find_by_name() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("named-session-01", Some("prod-agent"), std::process::id());
register_session_in(®, &dir).unwrap();
let found = find_session_registration_in("prod-agent", &dir);
assert!(found.is_some());
assert_eq!(found.unwrap().session_id, "named-session-01");
}
#[test]
fn find_by_partial_prefix() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("prefix-abcdef-01", None, std::process::id());
register_session_in(®, &dir).unwrap();
let found = find_session_registration_in("prefix-abc", &dir);
assert!(found.is_some());
assert_eq!(found.unwrap().session_id, "prefix-abcdef-01");
}
#[test]
fn ambiguous_prefix_returns_none() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let pid = std::process::id();
register_session_in(&make_reg("dup-aaaa-01", None, pid), &dir).unwrap();
register_session_in(&make_reg("dup-aaaa-02", None, pid), &dir).unwrap();
let found = find_session_registration_in("dup-aaaa", &dir);
assert!(found.is_none(), "ambiguous prefix should return None");
}
#[test]
fn foreign_json_is_never_unlinked() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let daemon = dir.join("daemon.json");
std::fs::write(
&daemon,
r#"{"pid":1,"version":"0.9.0","exe":"/usr/bin/synaps","socket":"/x/daemon.sock","protocol":2}"#,
)
.unwrap();
let garbage = dir.join("garbage.json");
std::fs::write(&garbage, "this is not json {").unwrap();
let other = dir.join("other-kind.json");
std::fs::write(
&other,
r#"{"kind":"widget","session_id":"x","name":null,"socket_path":"/x.sock","pid":1,"started_at":"2024-01-01T00:00:00Z"}"#,
)
.unwrap();
let legacy = dir.join("legacy-0001.json");
std::fs::write(
&legacy,
format!(
r#"{{"session_id":"legacy-0001","name":"old","socket_path":"{}","pid":{},"started_at":"2024-01-01T00:00:00Z"}}"#,
dir.join("legacy-0001.sock").display(),
std::process::id()
),
)
.unwrap();
let reg = make_reg("live-0002", Some("ambient"), std::process::id());
register_session_in(®, &dir).unwrap();
for _ in 0..2 {
let found = find_session_registration_in("ambient", &dir).expect("session resolved");
assert_eq!(found.session_id, "live-0002");
assert!(
find_session_registration_in("old", &dir).is_some(),
"legacy reg counts"
);
let ids: Vec<_> = list_active_sessions_in(&dir)
.into_iter()
.map(|r| r.session_id)
.collect();
assert!(!ids.contains(&"x".to_string()), "foreign kind not listed");
assert!(daemon.exists(), "daemon.json must survive");
assert!(garbage.exists(), "garbage.json must survive");
assert!(other.exists(), "other-kind.json must survive");
assert!(legacy.exists());
}
assert!(std::fs::read_to_string(&daemon).unwrap().contains("0.9.0"));
}
#[test]
fn update_name_rewrites_registration() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("rename-0001", None, std::process::id());
register_session_in(®, &dir).unwrap();
assert!(find_session_registration_in("ambient", &dir).is_none());
update_session_name_in("rename-0001", Some("ambient"), &dir).unwrap();
let found = find_session_registration_in("ambient", &dir).expect("resolves by new name");
assert_eq!(found.session_id, "rename-0001");
assert_eq!(
found.socket_path, reg.socket_path,
"rest of the record kept"
);
update_session_name_in("rename-0001", None, &dir).unwrap();
assert!(
find_session_registration_in("ambient", &dir).is_none(),
"cleared"
);
assert!(
update_session_name_in("ghost", Some("x"), &dir).is_err(),
"no reg → error"
);
}
#[test]
fn unregister_removes_file() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("unreg-0001", None, std::process::id());
register_session_in(®, &dir).unwrap();
let path = dir.join("unreg-0001.json");
assert!(path.exists());
unregister_session_in("unreg-0001", &dir);
assert!(!path.exists());
}
#[test]
fn unregister_is_idempotent() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
unregister_session_in("ghost-session-99", &dir);
}
#[cfg(unix)]
#[test]
fn stale_pid_pruned() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("stale-dead-pid", None, 999999);
register_session_in(®, &dir).unwrap();
let sessions = list_active_sessions_in(&dir);
assert!(
!sessions.iter().any(|r| r.session_id == "stale-dead-pid"),
"stale registration should have been pruned"
);
assert!(!dir.join("stale-dead-pid.json").exists());
}
#[cfg(unix)]
fn reg_with_socket(dir: &std::path::Path, id: &str, pid: u32) -> SessionRegistration {
let reg = SessionRegistration {
kind: REGISTRATION_KIND.to_string(),
session_id: id.to_string(),
name: None,
socket_path: socket_path_in_dir(dir, id),
pid,
started_at: Utc::now(),
};
let _ = std::fs::remove_file(®.socket_path);
let _socket = std::os::unix::net::UnixListener::bind(®.socket_path).unwrap();
register_session_in(®, dir).unwrap();
reg
}
#[cfg(unix)]
#[test]
fn unregister_owned_removes_its_own_registration_and_socket() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = reg_with_socket(&dir, "owned", std::process::id());
unregister_owned_in(®, &dir);
assert!(!dir.join("owned.json").exists());
assert!(!dir.join("owned.sock").exists());
unregister_owned_in(®, &dir);
}
#[cfg(unix)]
#[test]
fn unregister_owned_leaves_a_newer_registration_of_the_same_id() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let old = reg_with_socket(&dir, "same-id", std::process::id());
std::thread::sleep(std::time::Duration::from_millis(2));
let newer = reg_with_socket(&dir, "same-id", std::process::id());
assert_ne!(old.started_at, newer.started_at);
unregister_owned_in(&old, &dir);
assert!(dir.join("same-id.json").exists(), "newer registration kept");
assert!(dir.join("same-id.sock").exists(), "newer socket kept");
let live = list_active_sessions_in(&dir);
assert_eq!(live.len(), 1);
assert_eq!(live[0].started_at, newer.started_at);
}
#[cfg(unix)]
#[test]
fn a_renamed_registration_is_still_owned_and_removed() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = reg_with_socket(&dir, "renamed", std::process::id());
update_session_name_in("renamed", Some("ambient"), &dir).unwrap();
unregister_owned_in(®, &dir);
assert!(
!dir.join("renamed.json").exists(),
"a rename keeps ownership"
);
}
#[cfg(unix)]
#[test]
fn daemon_start_sweep_removes_dead_and_own_pid_and_keeps_other_live_owners() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let own = std::process::id();
let other_live = unsafe { libc::getppid() } as u32;
reg_with_socket(&dir, "dead-owner", 999_999);
reg_with_socket(&dir, "pre-reload", own);
reg_with_socket(&dir, "other-process", other_live);
std::fs::write(dir.join("daemon.json"), r#"{"pid":999999}"#).unwrap();
assert_eq!(sweep_stale_registrations_in(&dir, Some(own)), 2);
for gone in ["dead-owner", "pre-reload"] {
assert!(
!dir.join(format!("{gone}.json")).exists(),
"{gone} card removed"
);
assert!(
!dir.join(format!("{gone}.sock")).exists(),
"{gone} socket removed"
);
}
assert!(dir.join("other-process.json").exists());
assert!(dir.join("other-process.sock").exists());
assert!(
dir.join("daemon.json").exists(),
"daemon.json is never a session"
);
}
#[cfg(unix)]
#[test]
fn sweep_without_own_pid_keeps_registrations_of_this_process() {
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
reg_with_socket(&dir, "mine", std::process::id());
reg_with_socket(&dir, "dead", 999_999);
assert_eq!(sweep_stale_registrations_in(&dir, None), 1);
assert!(dir.join("mine.json").exists());
assert!(!dir.join("dead.json").exists());
}
#[test]
#[serial(synaps_base_dir)]
fn socket_path_format() {
let path = socket_path_for_session("20240101-120000-ab12");
assert!(
path.ends_with("/run/20240101-120000-ab12.sock"),
"got: {}",
path
);
assert!(!path.contains("/tmp/"), "socket should not be in /tmp");
}
#[cfg(unix)]
#[test]
fn short_runtime_dirs_keep_long_session_socket_bindable_and_isolated() {
use std::os::unix::net::UnixListener;
let long_session = "s".repeat(80);
let path_a = socket_path_in_dir(std::path::Path::new("/tmp/a"), &long_session);
let path_b = socket_path_in_dir(std::path::Path::new("/tmp/b"), &long_session);
assert!(path_a.len() < 108, "Unix socket path too long: {path_a}");
assert_ne!(
path_a, path_b,
"per-user runtime roots must isolate same session ids"
);
let pid = std::process::id();
let a = std::path::PathBuf::from(format!("/tmp/sa{pid}"));
let b = std::path::PathBuf::from(format!("/tmp/sb{pid}"));
std::fs::create_dir_all(&a).unwrap();
std::fs::create_dir_all(&b).unwrap();
let _a = UnixListener::bind(socket_path_in_dir(&a, &long_session)).unwrap();
let _b = UnixListener::bind(socket_path_in_dir(&b, &long_session)).unwrap();
let _ = std::fs::remove_dir_all(&a);
let _ = std::fs::remove_dir_all(&b);
}
#[cfg(unix)]
#[test]
fn registration_file_is_0600() {
use std::os::unix::fs::PermissionsExt;
let tmp = tmp_registry();
let dir = dir_buf(&tmp);
let reg = make_reg("perms-check-01", None, std::process::id());
register_session_in(®, &dir).unwrap();
let path = dir.join("perms-check-01.json");
let perms = std::fs::metadata(&path).unwrap().permissions();
assert_eq!(perms.mode() & 0o777, 0o600, "registry file should be 0600");
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DaemonPaths {
pub dir: PathBuf,
pub sock: PathBuf,
pub lock: PathBuf,
pub json: PathBuf,
pub pid: PathBuf,
}
pub fn daemon_paths(profile: Option<&str>) -> DaemonPaths {
daemon_paths_in(®istry_dir(), profile)
}
pub fn daemon_paths_in(dir: &std::path::Path, profile: Option<&str>) -> DaemonPaths {
let stem = match profile.map(sanitize_session_id).filter(|p| !p.is_empty()) {
Some(p) => format!("daemon-{p}"),
None => "daemon".to_string(),
};
DaemonPaths {
dir: dir.to_path_buf(),
sock: dir.join(format!("{stem}.sock")),
lock: dir.join(format!("{stem}.lock")),
json: dir.join(format!("{stem}.json")),
pid: dir.join(format!("{stem}.pid")),
}
}
#[cfg(test)]
mod daemon_paths_tests {
use super::*;
#[test]
fn daemon_paths_are_profile_scoped_and_sanitised() {
let d = std::path::Path::new("/run/x");
let p = daemon_paths_in(d, None);
assert_eq!(p.sock, d.join("daemon.sock"));
assert_eq!(p.lock, d.join("daemon.lock"));
assert_eq!(p.json, d.join("daemon.json"));
assert_eq!(p.pid, d.join("daemon.pid"));
let p = daemon_paths_in(d, Some("work/../evil"));
assert_eq!(p.sock, d.join("daemon-work____evil.sock"));
}
}