#[cfg_attr(coverage_nightly, coverage(off))]
mod windows;
use std::env;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use crate::SessionId;
use crate::pal::error::{PalError, PalErrorKind};
use crate::pal::session_store::SessionStore;
use crate::pal::session_store::fs_store::windows::{
RecordFile, move_file_no_replace, move_file_replace,
};
use crate::pal::session_store::stored::StoredSession;
use crate::session_record::{ProcessIdentity, SessionRecord};
#[derive(Clone, Debug)]
pub(crate) struct FsSessionStore {
root: PathBuf,
}
fn parse_stored(bytes: &[u8]) -> Result<StoredSession, PalError> {
serde_json::from_slice(bytes).map_err(|error| {
PalError::with_source(
PalErrorKind::Other,
io::Error::new(io::ErrorKind::InvalidData, error),
)
})
}
fn parse_record(bytes: &[u8], expected: SessionId) -> Result<Option<SessionRecord>, PalError> {
let StoredSession::Published(record) = parse_stored(bytes)? else {
return Ok(None);
};
if record.id != expected {
return Ok(None);
}
Ok(Some(record))
}
fn replace_file(tmp: &Path, dest: &Path) -> io::Result<()> {
move_file_replace(tmp, dest)
}
#[cfg_attr(coverage_nightly, coverage(off))]
fn serialization_error(error: serde_json::Error) -> PalError {
PalError::with_source(
PalErrorKind::Other,
io::Error::new(io::ErrorKind::InvalidData, error),
)
}
fn next_session_id(id: SessionId) -> Result<SessionId, PalError> {
id.get()
.checked_add(1)
.and_then(SessionId::from_u32)
.ok_or_else(|| PalError::new(PalErrorKind::Other))
}
fn remove_failed_staging(staging: &Path, installed: &Result<SessionId, PalError>) {
if installed.is_err() {
_ = fs::remove_file(staging);
}
}
impl FsSessionStore {
pub(crate) fn new(root: PathBuf) -> Self {
Self { root }
}
fn record_path(&self, id: SessionId) -> PathBuf {
self.root.join(format!("{}.json", id.get()))
}
fn staging_path(&self, owner: &ProcessIdentity) -> PathBuf {
static NEXT: AtomicU64 = AtomicU64::new(0);
let attempt = NEXT.fetch_add(1, Ordering::Relaxed);
self.root.join(format!(
"{}-{}-{attempt}.claim",
owner.pid, owner.creation_time
))
}
#[cfg_attr(test, mutants::skip)]
fn install_claim(&self, staging: &Path) -> Result<SessionId, PalError> {
let mut id = SessionId::MIN;
loop {
match move_file_no_replace(staging, &self.record_path(id)) {
Ok(()) => return Ok(id),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
id = next_session_id(id)?;
}
Err(error) => return Err(PalError::from_io(error)),
}
}
}
fn stored(&self) -> Result<Vec<(SessionId, StoredSession)>, PalError> {
let entries = match fs::read_dir(&self.root) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(PalError::from_io(error)),
};
let mut stored = Vec::new();
for entry in entries {
let entry = entry.map_err(PalError::from_io)?;
let name = entry.file_name();
let Some(name) = name.to_str() else {
continue;
};
let Some(stem) = name.strip_suffix(".json") else {
continue;
};
let Ok(raw) = stem.parse::<u32>() else {
continue;
};
let Some(id) = SessionId::from_u32(raw) else {
continue;
};
let bytes = match read_if_present(&entry.path()) {
Ok(Some(bytes)) => bytes,
Ok(None) => continue,
Err(error) => return Err(PalError::from_io(error)),
};
let Ok(parsed) = parse_stored(&bytes) else {
continue;
};
stored.push((id, parsed));
}
stored.sort_by_key(|(id, _stored)| id.get());
Ok(stored)
}
}
impl SessionStore for FsSessionStore {
#[cfg(test)]
fn root(&self) -> PathBuf {
self.root.clone()
}
fn allocate_id(&self, owner: &ProcessIdentity) -> Result<SessionId, PalError> {
fs::create_dir_all(&self.root).map_err(PalError::from_io)?;
let claim = serde_json::to_vec(&StoredSession::Reserved { owner: *owner })
.map_err(serialization_error)?;
let staging = self.staging_path(owner);
write_new_file(&staging, &claim).map_err(PalError::from_io)?;
let installed = self.install_claim(&staging);
remove_failed_staging(&staging, &installed);
installed
}
fn publish(&self, record: &SessionRecord) -> Result<(), PalError> {
fs::create_dir_all(&self.root).map_err(PalError::from_io)?;
let id = record.id;
let path = self.record_path(id);
let tmp = self.root.join(format!("{}.json.tmp", id.get()));
let json = serde_json::to_vec_pretty(&StoredSession::Published(record.clone()))
.map_err(serialization_error)?;
let mut file = File::create(&tmp).map_err(PalError::from_io)?;
file.write_all(&json).map_err(PalError::from_io)?;
file.sync_all().map_err(PalError::from_io)?;
drop(file);
replace_file(&tmp, &path).map_err(PalError::from_io)
}
fn read(&self, id: SessionId) -> Result<Option<SessionRecord>, PalError> {
let path = self.record_path(id);
match fs::read(&path) {
Ok(bytes) if bytes.is_empty() => Ok(None),
Ok(bytes) => parse_record(&bytes, id),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(PalError::from_io(error)),
}
}
fn list(&self) -> Result<Vec<SessionRecord>, PalError> {
Ok(self
.stored()?
.into_iter()
.filter_map(|(id, stored)| match stored {
StoredSession::Published(record) if record.id == id => Some(record),
_ => None,
})
.collect())
}
fn list_reservations(&self) -> Result<Vec<(SessionId, ProcessIdentity)>, PalError> {
Ok(self
.stored()?
.into_iter()
.filter_map(|(id, stored)| match stored {
StoredSession::Reserved { owner } => Some((id, owner)),
StoredSession::Published(_record) => None,
})
.collect())
}
fn delete_owned_by(&self, id: SessionId, owner: &ProcessIdentity) -> Result<(), PalError> {
let file = match RecordFile::open(&self.record_path(id)) {
Ok(file) => file,
Err(error) if is_absent(&error) => return Ok(()),
Err(error) => return Err(PalError::from_io(error)),
};
let bytes = file.read().map_err(PalError::from_io)?;
let owned = match parse_stored(&bytes) {
Ok(StoredSession::Reserved { owner: current }) => current == *owner,
Ok(StoredSession::Published(record)) => record.supervisor == *owner,
Err(_error) => false,
};
if !owned {
return Ok(());
}
file.delete().map_err(PalError::from_io)
}
fn canonicalize(&self, path: &Path) -> Result<PathBuf, PalError> {
fs::canonicalize(path).map_err(PalError::from_io)
}
fn current_dir(&self) -> Result<PathBuf, PalError> {
env::current_dir().map_err(PalError::from_io)
}
}
fn is_absent(error: &io::Error) -> bool {
matches!(error.kind(), io::ErrorKind::NotFound)
}
fn read_if_present(path: &Path) -> io::Result<Option<Vec<u8>>> {
match fs::read(path) {
Ok(bytes) => Ok(Some(bytes)),
Err(error) if is_absent(&error) => Ok(None),
Err(error) => Err(error),
}
}
fn write_new_file(path: &Path, content: &[u8]) -> io::Result<()> {
let mut file = OpenOptions::new().write(true).create_new(true).open(path)?;
file.write_all(content)?;
file.sync_all()
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
use std::collections::HashSet;
use std::path::Path;
use std::sync::{Arc, Barrier};
use std::{iter, thread};
use tempfile::TempDir;
use testing::with_watchdog;
use super::*;
use crate::AppCommand;
use crate::protocol::PROTOCOL_VERSION;
fn store() -> (TempDir, FsSessionStore) {
let dir = TempDir::new().unwrap();
let store = FsSessionStore::new(dir.path().to_path_buf());
(dir, store)
}
fn record(id: SessionId, dir: &Path) -> SessionRecord {
SessionRecord {
id,
supervisor: ProcessIdentity {
pid: 1,
creation_time: 1,
},
pipe_name: "pipe".to_string(),
launch_directory: dir.to_path_buf(),
command: AppCommand::for_test(&["app.exe"]),
started_at_unix_ms: 1,
attached: false,
protocol_version: PROTOCOL_VERSION,
}
}
#[test]
fn advancing_the_last_session_id_reports_exhaustion() {
let last = SessionId::from_u32(u32::MAX).unwrap();
next_session_id(last).unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn an_install_failure_removes_its_staging_file() {
let dir = TempDir::new().unwrap();
let staging = dir.path().join("failed.claim");
fs::write(&staging, b"claim").unwrap();
let installed = Err(PalError::new(PalErrorKind::Other));
remove_failed_staging(&staging, &installed);
assert!(!staging.exists());
}
#[test]
#[cfg_attr(miri, ignore)]
fn installing_a_missing_staging_file_reports_the_filesystem_error() {
let (dir, store) = store();
store
.install_claim(&dir.path().join("missing.claim"))
.unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_claim_names_its_owner_until_it_is_published() {
let (dir, store) = store();
let owner = ProcessIdentity::for_test(7);
let id = store.allocate_id(&owner).unwrap();
assert_eq!(store.list_reservations().unwrap(), vec![(id, owner)]);
assert!(store.list().unwrap().is_empty());
assert!(store.read(id).unwrap().is_none());
store.publish(&record(id, dir.path())).unwrap();
assert!(store.list_reservations().unwrap().is_empty());
assert_eq!(store.list().unwrap().len(), 1);
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_claim_is_deleted_only_by_its_owner() {
let (_dir, store) = store();
let owner = ProcessIdentity::for_test(7);
let id = store.allocate_id(&owner).unwrap();
store
.delete_owned_by(id, &ProcessIdentity::for_test(8))
.unwrap();
assert_eq!(store.list_reservations().unwrap().len(), 1);
store.delete_owned_by(id, &owner).unwrap();
assert!(store.list_reservations().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_is_deleted_only_by_its_own_supervisor() {
let (dir, store) = store();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let mut published = record(id, dir.path());
published.supervisor = ProcessIdentity {
pid: 20,
creation_time: 200,
};
store.publish(&published).unwrap();
let stale = ProcessIdentity {
pid: 20,
creation_time: 199,
};
store.delete_owned_by(id, &stale).unwrap();
assert_eq!(store.read(id).unwrap().unwrap(), published);
store.delete_owned_by(id, &published.supervisor).unwrap();
assert!(store.read(id).unwrap().is_none());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_file_that_is_still_empty_reads_as_absent() {
let (dir, store) = store();
let id = SessionId::from_u32(3).unwrap();
fs::write(dir.path().join(format!("{}.json", id.get())), b"").unwrap();
assert!(store.read(id).unwrap().is_none());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_that_cannot_be_inspected_is_not_deleted() {
let (dir, store) = store();
let id = SessionId::from_u32(3).unwrap();
fs::create_dir_all(dir.path().join(format!("{}.json", id.get()))).unwrap();
store
.delete_owned_by(id, &ProcessIdentity::for_test(7))
.unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn deleting_what_is_not_there_or_not_readable_is_not_an_error() {
let (dir, store) = store();
let owner = ProcessIdentity::for_test(7);
let id = store.allocate_id(&owner).unwrap();
fs::write(dir.path().join(format!("{}.json", id.get())), b"not json").unwrap();
store.delete_owned_by(id, &owner).unwrap();
assert!(dir.path().join(format!("{}.json", id.get())).exists());
store
.delete_owned_by(SessionId::from_u32(99).unwrap(), &owner)
.unwrap();
}
#[test]
#[cfg_attr(miri, ignore)]
fn allocates_smallest_unused_and_reuses_after_delete() {
let (dir, store) = store();
assert_eq!(store.root(), dir.path());
let first = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let second = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
assert_eq!(first.get(), 1);
assert_eq!(second.get(), 2);
store
.delete_owned_by(first, &ProcessIdentity::for_test(1))
.unwrap();
let reused = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
assert_eq!(reused.get(), 1);
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_reserved_but_unpublished_id_reads_as_absent() {
let (_dir, store) = store();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
assert!(store.read(id).unwrap().is_none());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_claimed_id_is_readable_as_a_claim_the_moment_it_exists() {
let (dir, store) = store();
let owner = ProcessIdentity::for_test(1);
let id = store.allocate_id(&owner).unwrap();
let raw = fs::read(dir.path().join(format!("{}.json", id.get()))).unwrap();
assert_eq!(
parse_stored(&raw).unwrap(),
StoredSession::Reserved { owner }
);
assert_eq!(store.list_reservations().unwrap(), vec![(id, owner)]);
}
#[test]
#[cfg_attr(miri, ignore)]
fn allocating_leaves_no_staging_file_behind() {
let (dir, store) = store();
_ = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let leftovers: Vec<_> = fs::read_dir(dir.path())
.unwrap()
.map(|entry| entry.unwrap().file_name())
.filter(|name| !name.to_string_lossy().ends_with(".json"))
.collect();
assert!(leftovers.is_empty(), "left behind {leftovers:?}");
}
#[test]
fn a_staging_name_carries_the_complete_process_identity() {
let store = FsSessionStore::new(PathBuf::from("store"));
let owner = ProcessIdentity {
pid: 73,
creation_time: 987_654_321,
};
let name = store
.staging_path(&owner)
.file_name()
.unwrap()
.to_string_lossy()
.into_owned();
assert!(name.starts_with(&format!("{}-{}-", owner.pid, owner.creation_time)));
}
#[test]
#[cfg_attr(miri, ignore)]
fn concurrent_allocations_are_unique() {
with_watchdog(|| {
const WORKERS: usize = 8;
let dir = TempDir::new().unwrap();
let root = dir.path().to_path_buf();
let start = Arc::new(Barrier::new(WORKERS));
let threads: Vec<_> = iter::repeat_with(|| {
let root = root.clone();
let start = Arc::clone(&start);
thread::spawn(move || {
let store = FsSessionStore::new(root);
start.wait();
store.allocate_id(&ProcessIdentity::for_test(1)).unwrap()
})
})
.take(WORKERS)
.collect();
let mut ids = HashSet::new();
for handle in threads {
assert!(ids.insert(handle.join().unwrap().get()));
}
assert_eq!(ids.len(), WORKERS);
});
}
#[test]
#[cfg_attr(miri, ignore)]
fn publish_read_list_delete() {
let (dir, store) = store();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let rec = record(id, dir.path());
store.publish(&rec).unwrap();
assert_eq!(store.read(id).unwrap().unwrap(), rec);
assert_eq!(store.list().unwrap(), vec![rec.clone()]);
store.delete_owned_by(id, &rec.supervisor).unwrap();
assert!(store.read(id).unwrap().is_none());
assert!(store.list().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_that_is_already_gone_reads_as_nothing_to_read() {
let dir = TempDir::new().unwrap();
assert!(
read_if_present(&dir.path().join("1.json"))
.unwrap()
.is_none()
);
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_store_that_does_not_exist_yet_holds_no_sessions() {
let dir = TempDir::new().unwrap();
let store = FsSessionStore::new(dir.path().join("never-created"));
assert!(store.list().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn list_skips_files_that_are_not_usable_records() {
let (dir, store) = store();
fs::write(dir.path().join("0.json"), b"{\"id\":0}").unwrap();
fs::write(dir.path().join("not-json.json"), b"nope").unwrap();
fs::write(dir.path().join("readme.txt"), b"not a record").unwrap();
fs::write(dir.path().join("5.json"), b"nope").unwrap();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let rec = record(id, dir.path());
store.publish(&rec).unwrap();
assert_eq!(store.list().unwrap(), vec![rec]);
assert!(store.list_reservations().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_the_store_cannot_read_is_reported_rather_than_omitted() {
let (dir, store) = store();
fs::create_dir_all(dir.path().join("6.json")).unwrap();
store.list().unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn read_reports_a_corrupt_record() {
let (dir, store) = store();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
fs::write(dir.path().join(format!("{}.json", id.get())), b"nope").unwrap();
assert_eq!(store.read(id).unwrap_err().kind(), PalErrorKind::Other);
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_path_that_is_not_a_file_is_an_error() {
let (dir, store) = store();
let id = SessionId::MIN;
fs::create_dir_all(dir.path().join(format!("{}.json", id.get()))).unwrap();
store.read(id).unwrap_err();
store
.delete_owned_by(id, &ProcessIdentity::for_test(1))
.unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_root_that_is_not_a_directory_is_an_error() {
let dir = TempDir::new().unwrap();
let root = dir.path().join("store");
fs::write(&root, b"not a directory").unwrap();
let store = FsSessionStore::new(root);
store.list().unwrap_err();
store.list_reservations().unwrap_err();
store
.allocate_id(&ProcessIdentity::for_test(1))
.unwrap_err();
store
.publish(&record(SessionId::MIN, dir.path()))
.unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_staging_path_that_is_not_a_file_fails_the_publish() {
let (dir, store) = store();
let id = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
fs::create_dir_all(dir.path().join(format!("{}.json.tmp", id.get()))).unwrap();
store.publish(&record(id, dir.path())).unwrap_err();
assert_eq!(store.list_reservations().unwrap().len(), 1);
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_too_large_to_read_in_one_go_is_still_its_owner_s_to_delete() {
let (dir, store) = store();
let owner = ProcessIdentity::for_test(1);
let id = store.allocate_id(&owner).unwrap();
let mut record = record(id, &dir.path().join("d".repeat(9_000)));
record.command = AppCommand::from_argv(vec!["app.exe".to_string(), "a".repeat(20_000)])
.expect("test argv names an executable");
store.publish(&record).unwrap();
store.delete_owned_by(id, &record.supervisor).unwrap();
assert!(store.list().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn deleting_an_absent_record_succeeds() {
let (_dir, store) = store();
store
.delete_owned_by(SessionId::MIN, &ProcessIdentity::for_test(1))
.unwrap();
}
#[test]
#[cfg_attr(miri, ignore)]
fn a_record_that_replaced_the_one_inspected_survives_the_delete() {
let (dir, store) = store();
let owner = ProcessIdentity::for_test(1);
let id = store.allocate_id(&owner).unwrap();
let path = dir.path().join(format!("{}.json", id.get()));
let inspected = RecordFile::open(&path).unwrap();
fs::remove_file(&path).unwrap();
let successor = record(id, dir.path());
store.publish(&successor).unwrap();
inspected.delete().unwrap();
drop(inspected);
assert_eq!(store.list().unwrap().len(), 1);
store.delete_owned_by(id, &successor.supervisor).unwrap();
assert!(store.list().unwrap().is_empty());
}
#[test]
#[cfg_attr(miri, ignore)]
fn canonicalize_resolves_an_existing_path_and_rejects_a_missing_one() {
let (dir, store) = store();
let resolved = store.canonicalize(dir.path()).unwrap();
assert!(resolved.is_absolute());
assert_eq!(resolved, fs::canonicalize(dir.path()).unwrap());
store.canonicalize(&dir.path().join("absent")).unwrap_err();
}
#[test]
#[cfg_attr(miri, ignore)]
fn read_rejects_filename_id_mismatch() {
let (dir, store) = store();
let first = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
let rec = record(first, dir.path());
store.publish(&rec).unwrap();
let second = store.allocate_id(&ProcessIdentity::for_test(1)).unwrap();
fs::write(
dir.path().join(format!("{}.json", second.get())),
serde_json::to_vec(&StoredSession::Published(rec.clone())).unwrap(),
)
.unwrap();
assert!(store.read(second).unwrap().is_none());
assert_eq!(store.list().unwrap(), vec![rec.clone()]);
store.delete_owned_by(second, &rec.supervisor).unwrap();
assert_eq!(store.read(first).unwrap().unwrap(), rec);
}
#[test]
#[cfg_attr(miri, ignore)]
fn missing_root_lists_and_reads_empty() {
let dir = TempDir::new().unwrap();
let missing = dir.path().join("no-such-store");
let store = FsSessionStore::new(missing);
assert!(store.list().unwrap().is_empty());
assert!(store.read(SessionId::MIN).unwrap().is_none());
}
}