use super::{StorageMaintenanceAdmission, StorageReadAdmission, invalid};
use bcode_session_models::SessionId;
use std::fs::File;
use std::io;
use std::os::unix::ffi::OsStrExt as _;
use std::path::Path;
pub struct AcknowledgedStorageMaintenance<'a> {
_admission: StorageMaintenanceAdmission,
_registration: &'a mut crate::storage_daemon_registration::StorageDaemonRegistration,
}
pub struct OwnedStorageMaintenance {
_admission: StorageMaintenanceAdmission,
acknowledgements: Vec<crate::storage_daemon_registration::StorageDaemonAcknowledgement>,
}
impl OwnedStorageMaintenance {
pub fn check(&self) -> io::Result<()> {
self.acknowledgements
.iter()
.try_for_each(crate::storage_daemon_registration::StorageDaemonAcknowledgement::check)
}
}
#[derive(Debug)]
pub struct UnacknowledgedStorageDaemon;
impl std::fmt::Display for UnacknowledgedStorageDaemon {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("unacknowledged storage daemon")
}
}
impl std::error::Error for UnacknowledgedStorageDaemon {}
pub struct StorageAdmissionRegistry {
directory: File,
}
impl StorageAdmissionRegistry {
pub fn open(root: &Path) -> io::Result<Self> {
use std::os::unix::fs::OpenOptionsExt as _;
let root = std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NOFOLLOW | libc::O_DIRECTORY | libc::O_CLOEXEC)
.open(root)?;
if !root.metadata()?.is_dir() {
return Err(invalid());
}
let name = c"storage-admission-v1";
let created = unsafe { libc::mkdirat(raw(&root), name.as_ptr(), 0o700) };
if created != 0 && io::Error::last_os_error().kind() != io::ErrorKind::AlreadyExists {
return Err(io::Error::last_os_error());
}
let directory = open_child(&root, name, libc::O_RDONLY | libc::O_DIRECTORY)?;
bcode_metrics::startup::measure("storage_admission.root_sync", || root.sync_all())?;
Ok(Self { directory })
}
pub fn open_session(root: &Path, session_id: SessionId) -> io::Result<Self> {
use std::os::unix::fs::OpenOptionsExt as _;
let root = std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NOFOLLOW | libc::O_DIRECTORY | libc::O_CLOEXEC)
.open(root)?;
let parent = create_directory(&root, c"storage-admission-sessions-v1")?;
let name = std::ffi::CString::new(session_id.to_string()).map_err(|_| invalid())?;
let directory = create_directory(&parent, &name)?;
Ok(Self { directory })
}
pub fn admit_session_maintenance(
root: &Path,
id: SessionId,
) -> io::Result<StorageMaintenanceAdmission> {
let registry = Self::open_session(root, id)?;
match registry.admit_maintenance(4096) {
Ok(admission) => Ok(admission),
Err(error) if error.kind() == io::ErrorKind::InvalidData => {
let report = Self::session_report_bounded(root, id, true, 4096, false)?;
if report.busy || report.invalid || report.retired == 0 {
return Err(error);
}
registry.admit_maintenance(4096)
}
Err(error) => Err(error),
}
}
pub fn session_report(
root: &Path,
id: SessionId,
apply: bool,
) -> io::Result<bcode_session_models::StorageAdmissionReport> {
Self::session_report_bounded(root, id, apply, 1_000_000, true)
}
fn session_report_bounded(
root: &Path,
id: SessionId,
apply: bool,
entry_budget: usize,
permit_empty: bool,
) -> io::Result<bcode_session_models::StorageAdmissionReport> {
use std::io::{Read as _, Seek as _};
use std::os::unix::fs::MetadataExt as _;
use std::os::unix::fs::OpenOptionsExt as _;
let root_handle = std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NOFOLLOW | libc::O_DIRECTORY | libc::O_CLOEXEC)
.open(root)?;
let parent = open_child(
&root_handle,
c"storage-admission-sessions-v1",
libc::O_RDONLY | libc::O_DIRECTORY,
)?;
let name = std::ffi::CString::new(id.to_string()).map_err(|_| invalid())?;
let directory = open_child(&parent, &name, libc::O_RDONLY | libc::O_DIRECTORY)?;
let mut report = bcode_session_models::StorageAdmissionReport::default();
let gate = open_child(&directory, c"gate", libc::O_RDWR)?;
if gate.try_lock().is_err() {
report.busy = true;
return Ok(report);
}
let Ok(entries) = names(&directory, entry_budget) else {
report.invalid = true;
return Ok(report);
};
let mut participants = Vec::new();
for entry in entries {
if entry == "gate" {
continue;
}
let valid = entry
.to_str()
.and_then(|s| s.strip_suffix(".participant"))
.is_some_and(|s| s.parse::<SessionId>().is_ok());
if !valid {
report.invalid = true;
return Ok(report);
}
let name = std::ffi::CString::new(entry.as_bytes()).map_err(|_| invalid())?;
let mut file = open_child(&directory, &name, libc::O_RDONLY)?;
if file.try_lock().is_err() {
report.busy = true;
return Ok(report);
}
let metadata = file.metadata()?;
if permit_empty && metadata.len() == 0 {
report.abandoned += 1;
participants.push((name, metadata.dev(), metadata.ino()));
continue;
}
if metadata.len() != super::DIRTY.len() as u64 {
report.invalid = true;
return Ok(report);
}
let mut bytes = [0; 17];
file.rewind()?;
file.read_exact(&mut bytes)?;
if &bytes == super::DIRTY {
report.abandoned += 1;
} else if &bytes == super::CLEAN {
report.clean += 1;
} else {
report.invalid = true;
return Ok(report);
}
let metadata = file.metadata()?;
participants.push((name, metadata.dev(), metadata.ino()));
}
if apply && !participants.is_empty() {
let _ownership = crate::lease::acquire_session_maintenance_guard(root, id)
.map_err(io::Error::other)?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(io::Error::other)?
.as_millis();
crate::storage_access::record_session_access(
root,
id,
crate::storage_access::StorageAccessKind::History,
u64::try_from(now).map_err(io::Error::other)?,
)?;
for (name, device, inode) in &participants {
let file = open_child(&directory, name, libc::O_RDONLY)?;
file.try_lock().map_err(io::Error::from)?;
let metadata = file.metadata()?;
if metadata.dev() != *device || metadata.ino() != *inode {
return Err(invalid());
}
if unsafe { libc::unlinkat(raw(&directory), name.as_ptr(), 0) } != 0 {
return Err(io::Error::last_os_error());
}
report.retired += 1;
}
directory.sync_all()?;
}
Ok(report)
}
pub fn compact_legacy_readers(root: &Path, entry_budget: usize) -> io::Result<usize> {
use std::io::Read as _;
use std::os::unix::fs::OpenOptionsExt as _;
if entry_budget == 0 || entry_budget > 65_536 {
return Err(invalid());
}
let root = std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NOFOLLOW | libc::O_DIRECTORY | libc::O_CLOEXEC)
.open(root)?;
let directory = open_child(
&root,
c"storage-admission-v1",
libc::O_RDONLY | libc::O_DIRECTORY,
)?;
let gate = open_child(&directory, c"gate", libc::O_RDWR)?;
gate.try_lock().map_err(io::Error::from)?;
let marker = open_child(
&directory,
c"unregistered-read.blocked",
libc::O_RDWR | libc::O_CREAT,
)?;
marker.sync_all()?;
directory.sync_all()?;
let entries = names_bounded(&directory, entry_budget, false)?;
let mut retired = 0;
for entry in entries {
if entry
.to_str()
.and_then(|s| s.strip_suffix(".participant"))
.is_none_or(|s| s.parse::<SessionId>().is_err())
{
continue;
}
let name = std::ffi::CString::new(entry.as_bytes()).map_err(|_| invalid())?;
let Ok(mut file) = open_child(&directory, &name, libc::O_RDONLY) else {
continue;
};
if file.try_lock().is_err() {
continue;
}
let length = file.metadata()?.len();
if length != 0 {
if length != super::DIRTY.len() as u64 {
continue;
}
let mut bytes = [0; 17];
file.read_exact(&mut bytes)?;
if &bytes != super::DIRTY && &bytes != super::CLEAN {
continue;
}
}
if unsafe { libc::unlinkat(raw(&directory), name.as_ptr(), 0) } != 0 {
return Err(io::Error::last_os_error());
}
retired += 1;
}
directory.sync_all()?;
Ok(retired)
}
pub fn retire_completed_daemons(&self, entry_budget: usize) -> io::Result<usize> {
if entry_budget == 0 || entry_budget > 65_536 {
return Err(invalid());
}
let gate = self.gate()?;
gate.try_lock().map_err(io::Error::from)?;
let entries = names(&self.directory, entry_budget)?;
let mut retired = 0;
for name in entries {
let Some(id) = name.to_str().and_then(|name| name.strip_suffix(".daemon")) else {
continue;
};
if id.parse::<SessionId>().is_err() {
continue;
}
let name = std::ffi::CString::new(name.as_bytes()).map_err(|_| invalid())?;
let file = open_child(&self.directory, &name, libc::O_RDONLY)?;
if !matches!(
crate::storage_daemon_registration::daemon_registration_is_complete(file),
Ok(true)
) {
continue;
}
if unsafe { libc::unlinkat(raw(&self.directory), name.as_ptr(), 0) } != 0 {
return Err(io::Error::last_os_error());
}
self.directory.sync_all()?;
retired += 1;
}
gate.unlock()?;
Ok(retired)
}
pub fn register_daemon(
&self,
identity: SessionId,
) -> io::Result<crate::storage_daemon_registration::StorageDaemonRegistration> {
let gate = self.gate()?;
gate.try_lock_shared().map_err(io::Error::from)?;
let name = std::ffi::CString::new(format!("{identity}.daemon")).map_err(|_| invalid())?;
let file = open_child(
&self.directory,
&name,
libc::O_RDWR | libc::O_CREAT | libc::O_EXCL,
)?;
let registration =
bcode_metrics::startup::measure("storage_admission.registration_begin", || {
crate::storage_daemon_registration::StorageDaemonRegistration::begin(file)
})?;
bcode_metrics::startup::measure("storage_admission.registration_directory_sync", || {
self.directory.sync_all()
})?;
gate.unlock()?;
Ok(registration)
}
pub fn admit_read(&self, participant: SessionId) -> io::Result<StorageReadAdmission> {
let gate = self.gate()?;
gate.try_lock_shared().map_err(io::Error::from)?;
let name =
std::ffi::CString::new(format!("{participant}.participant")).map_err(|_| invalid())?;
let file = open_child(&self.directory, &name, libc::O_RDWR | libc::O_CREAT)?;
self.directory.sync_all()?;
StorageReadAdmission::begin(gate, file)
}
pub fn admit_maintenance(
&self,
entry_budget: usize,
) -> io::Result<StorageMaintenanceAdmission> {
self.admit_checked(entry_budget, None)
}
pub fn admit_acknowledged<'a>(
&self,
entry_budget: usize,
registration: &'a mut crate::storage_daemon_registration::StorageDaemonRegistration,
) -> io::Result<AcknowledgedStorageMaintenance<'a>> {
let admission = self.admit_checked(entry_budget, Some(&*registration))?;
Ok(AcknowledgedStorageMaintenance {
_admission: admission,
_registration: registration,
})
}
pub fn admit_owned(
&self,
entry_budget: usize,
acknowledgement: crate::storage_daemon_registration::StorageDaemonAcknowledgement,
) -> io::Result<OwnedStorageMaintenance> {
self.admit_owned_set(entry_budget, vec![acknowledgement])
}
pub fn admit_owned_set(
&self,
entry_budget: usize,
acknowledgements: Vec<crate::storage_daemon_registration::StorageDaemonAcknowledgement>,
) -> io::Result<OwnedStorageMaintenance> {
if acknowledgements.len() > entry_budget {
return Err(invalid());
}
let admission = self.admit_owned_checked(entry_budget, &acknowledgements)?;
Ok(OwnedStorageMaintenance {
_admission: admission,
acknowledgements,
})
}
fn admit_checked(
&self,
entry_budget: usize,
registration: Option<&crate::storage_daemon_registration::StorageDaemonRegistration>,
) -> io::Result<StorageMaintenanceAdmission> {
let acknowledgement = registration
.map(crate::storage_daemon_registration::StorageDaemonRegistration::acknowledgement)
.transpose()?;
self.admit_owned_checked(entry_budget, acknowledgement.as_slice())
}
fn admit_owned_checked(
&self,
entry_budget: usize,
registrations: &[crate::storage_daemon_registration::StorageDaemonAcknowledgement],
) -> io::Result<StorageMaintenanceAdmission> {
if entry_budget == 0 || entry_budget > 65_536 {
return Err(invalid());
}
let gate = self.gate()?;
gate.try_lock().map_err(io::Error::from)?;
let names = names(&self.directory, entry_budget)?;
let mut files = Vec::new();
let mut acknowledged = std::collections::BTreeSet::new();
for name in names {
if name == "gate" {
continue;
}
let text = name.to_str().ok_or_else(invalid)?;
if let Some(id) = text.strip_suffix(".daemon") {
id.parse::<SessionId>().map_err(|_| invalid())?;
let name = std::ffi::CString::new(name.as_bytes()).map_err(|_| invalid())?;
let file = open_child(&self.directory, &name, libc::O_RDONLY)?;
let mut matched = false;
for (index, registration) in registrations.iter().enumerate() {
if registration.acknowledges(&file)? {
if matched || !acknowledged.insert(index) {
return Err(invalid());
}
matched = true;
}
}
if matched {
continue;
}
let complete =
crate::storage_daemon_registration::daemon_registration_is_complete(file)
.map_err(|error| {
if error.kind() == io::ErrorKind::WouldBlock {
io::Error::new(
io::ErrorKind::PermissionDenied,
UnacknowledgedStorageDaemon,
)
} else {
error
}
})?;
if !complete {
return Err(io::Error::new(
io::ErrorKind::PermissionDenied,
UnacknowledgedStorageDaemon,
));
}
continue;
}
let id = text.strip_suffix(".participant").ok_or_else(invalid)?;
id.parse::<SessionId>().map_err(|_| invalid())?;
let name = std::ffi::CString::new(name.as_bytes()).map_err(|_| invalid())?;
files.push(Ok(open_child(&self.directory, &name, libc::O_RDONLY)?));
}
if acknowledged.len() != registrations.len() {
return Err(invalid());
}
StorageMaintenanceAdmission::begin(gate, files)
}
pub fn complete_read(
&self,
participant: SessionId,
mut admission: StorageReadAdmission,
) -> io::Result<()> {
use std::os::unix::fs::MetadataExt as _;
if !admission.may_clean {
return Err(invalid());
}
let name =
std::ffi::CString::new(format!("{participant}.participant")).map_err(|_| invalid())?;
let current = open_child(&self.directory, &name, libc::O_RDONLY)?;
let current = current.metadata()?;
let held = admission.participant.metadata()?;
if current.dev() != held.dev() || current.ino() != held.ino() {
return Err(invalid());
}
if unsafe { libc::unlinkat(raw(&self.directory), name.as_ptr(), 0) } != 0 {
return Err(io::Error::last_os_error());
}
self.directory.sync_all()?;
admission.may_clean = false;
admission.participant.unlock()?;
admission.gate.unlock()
}
pub fn block_maintenance_for_fallback(&self) -> io::Result<()> {
let gate = self.gate()?;
gate.try_lock_shared().map_err(io::Error::from)?;
let marker = open_child(
&self.directory,
c"unregistered-read.blocked",
libc::O_RDWR | libc::O_CREAT,
)?;
marker.sync_all()?;
self.directory.sync_all()?;
gate.unlock()
}
pub fn mark_degraded(&self, participant: SessionId) -> io::Result<()> {
drop(self.admit_read(participant)?);
Ok(())
}
pub fn check_health(&self, entry_budget: usize) -> io::Result<()> {
drop(self.admit_maintenance(entry_budget)?);
Ok(())
}
pub fn retire(&self, participant: SessionId) -> io::Result<()> {
let gate = self.gate()?;
gate.try_lock().map_err(io::Error::from)?;
let name =
std::ffi::CString::new(format!("{participant}.participant")).map_err(|_| invalid())?;
let mut file = open_child(&self.directory, &name, libc::O_RDONLY)?;
file.try_lock_shared().map_err(io::Error::from)?;
super::read_state(&mut file, false)?;
if unsafe { libc::unlinkat(raw(&self.directory), name.as_ptr(), 0) } != 0 {
return Err(io::Error::last_os_error());
}
self.directory.sync_all()?;
file.unlock()?;
gate.unlock()
}
fn gate(&self) -> io::Result<File> {
let file = open_child(&self.directory, c"gate", libc::O_RDWR | libc::O_CREAT)?;
self.directory.sync_all()?;
Ok(file)
}
}
fn create_directory(parent: &File, name: &std::ffi::CStr) -> io::Result<File> {
let created = unsafe { libc::mkdirat(raw(parent), name.as_ptr(), 0o700) };
if created != 0 && io::Error::last_os_error().kind() != io::ErrorKind::AlreadyExists {
return Err(io::Error::last_os_error());
}
let directory = open_child(parent, name, libc::O_RDONLY | libc::O_DIRECTORY)?;
parent.sync_all()?;
Ok(directory)
}
fn raw(file: &File) -> std::os::fd::RawFd {
use std::os::fd::AsRawFd as _;
file.as_raw_fd()
}
fn open_child(parent: &File, name: &std::ffi::CStr, flags: i32) -> io::Result<File> {
use std::os::fd::FromRawFd as _;
use std::os::unix::fs::MetadataExt as _;
let descriptor = unsafe {
libc::openat(
raw(parent),
name.as_ptr(),
flags | libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK,
0o600,
)
};
if descriptor < 0 {
return Err(io::Error::last_os_error());
}
let file = unsafe { File::from_raw_fd(descriptor) };
let metadata = file.metadata()?;
if flags & libc::O_DIRECTORY != 0 {
if !metadata.is_dir() {
return Err(invalid());
}
} else if !metadata.is_file() || metadata.nlink() != 1 {
return Err(invalid());
}
Ok(file)
}
struct Directory(*mut libc::DIR);
impl Drop for Directory {
fn drop(&mut self) {
unsafe {
libc::closedir(self.0);
}
}
}
fn names(directory: &File, budget: usize) -> io::Result<Vec<std::ffi::OsString>> {
names_bounded(directory, budget, true)
}
fn names_bounded(
directory: &File,
budget: usize,
require_complete: bool,
) -> io::Result<Vec<std::ffi::OsString>> {
use std::os::fd::{FromRawFd as _, IntoRawFd as _};
use std::os::unix::ffi::OsStringExt as _;
let fd = open_child(directory, c".", libc::O_RDONLY | libc::O_DIRECTORY)?.into_raw_fd();
let stream = unsafe { libc::fdopendir(fd) };
if stream.is_null() {
let error = io::Error::last_os_error();
drop(unsafe { File::from_raw_fd(fd) });
return Err(error);
}
let stream = Directory(stream);
let mut names = Vec::new();
loop {
unsafe {
*errno() = 0;
}
let entry = unsafe { libc::readdir(stream.0) };
if entry.is_null() {
let error = io::Error::last_os_error();
if error.raw_os_error() != Some(0) {
return Err(error);
}
break;
}
let bytes = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes();
if bytes == b"." || bytes == b".." {
continue;
}
if names.len() == budget {
if require_complete {
return Err(invalid());
}
break;
}
names.push(std::ffi::OsString::from_vec(bytes.to_vec()));
}
Ok(names)
}
#[cfg(target_os = "macos")]
unsafe fn errno() -> *mut libc::c_int {
unsafe { libc::__error() }
}
#[cfg(target_os = "linux")]
unsafe fn errno() -> *mut libc::c_int {
unsafe { libc::__errno_location() }
}
#[cfg(test)]
mod tests {
#[test]
fn legacy_compaction_keeps_blocker_live_daemons_and_unknown_evidence() {
use super::StorageAdmissionRegistry;
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let daemon = registry
.register_daemon(bcode_session_models::SessionId::new())
.expect("daemon");
for _ in 0..5 {
drop(
registry
.admit_read(bcode_session_models::SessionId::new())
.expect("read"),
);
}
let directory = root.path().join("storage-admission-v1");
let unknown = directory.join(format!(
"{}.participant",
bcode_session_models::SessionId::new()
));
std::fs::write(&unknown, b"future").expect("unknown");
assert_eq!(
StorageAdmissionRegistry::compact_legacy_readers(root.path(), 32).expect("compact"),
5
);
assert_eq!(std::fs::read(unknown).expect("preserved"), b"future");
assert!(directory.join("unregistered-read.blocked").exists());
assert!(daemon.healthy());
let active = registry
.admit_read(bcode_session_models::SessionId::new())
.expect("active");
assert!(StorageAdmissionRegistry::compact_legacy_readers(root.path(), 32).is_err());
drop(active);
assert!(registry.admit_maintenance(32).is_err());
}
#[tokio::test]
async fn empty_registration_requires_explicit_recovery_and_unknown_bytes_are_preserved() {
let root = tempfile::tempdir().expect("root");
let sessions = crate::SessionManager::persistent(root.path()).expect("sessions");
let id = sessions
.create_session(None, root.path().to_path_buf())
.await
.expect("session")
.id;
let registry =
super::StorageAdmissionRegistry::open_session(root.path(), id).expect("registry");
drop(
registry
.admit_read(bcode_session_models::SessionId::new())
.expect("read"),
);
let directory = root
.path()
.join("storage-admission-sessions-v1")
.join(id.to_string());
let empty = directory.join(format!(
"{}.participant",
bcode_session_models::SessionId::new()
));
std::fs::write(&empty, []).expect("empty interrupted registration");
assert!(
super::StorageAdmissionRegistry::admit_session_maintenance(root.path(), id).is_err()
);
assert!(empty.exists());
let unknown = directory.join(format!(
"{}.participant",
bcode_session_models::SessionId::new()
));
std::fs::write(&unknown, b"future").expect("unknown record");
let report =
super::StorageAdmissionRegistry::session_report(root.path(), id, true).expect("report");
assert!(report.invalid);
assert_eq!(report.retired, 0);
assert!(empty.exists());
std::fs::remove_file(unknown).expect("remove test obstruction");
let report = super::StorageAdmissionRegistry::session_report(root.path(), id, true)
.expect("recovery");
assert_eq!(report.abandoned, 2);
assert_eq!(report.retired, 2);
assert!(!empty.exists());
drop(registry.admit_maintenance(1).expect("gate only"));
}
#[tokio::test]
async fn explicit_recovery_handles_inventory_larger_than_automatic_budget() {
let root = tempfile::tempdir().expect("root");
let sessions = crate::SessionManager::persistent(root.path()).expect("sessions");
let id = sessions
.create_session(None, root.path().to_path_buf())
.await
.expect("session")
.id;
let registry =
super::StorageAdmissionRegistry::open_session(root.path(), id).expect("registry");
drop(
registry
.admit_read(bcode_session_models::SessionId::new())
.expect("read"),
);
let directory = root
.path()
.join("storage-admission-sessions-v1")
.join(id.to_string());
for _ in 0..4100 {
std::fs::write(
directory.join(format!(
"{}.participant",
bcode_session_models::SessionId::new()
)),
super::super::DIRTY,
)
.expect("record");
}
assert!(
super::StorageAdmissionRegistry::admit_session_maintenance(root.path(), id).is_err()
);
let report = super::StorageAdmissionRegistry::session_report(root.path(), id, false)
.expect("report");
assert_eq!(report.abandoned, 4101);
assert!(!report.invalid);
let report = super::StorageAdmissionRegistry::session_report(root.path(), id, true)
.expect("recovery");
assert_eq!(report.retired, 4101);
drop(registry.admit_maintenance(1).expect("gate only"));
}
#[tokio::test]
async fn automatic_recovery_resets_age_and_preserves_unknown_evidence() {
let root = tempfile::tempdir().expect("root");
let sessions = crate::SessionManager::persistent(root.path()).expect("sessions");
let id = sessions
.create_session(None, root.path().to_path_buf())
.await
.expect("session")
.id;
sessions
.release_session_ownership(id)
.await
.expect("release");
let registry = StorageAdmissionRegistry::open_session(root.path(), id).expect("registry");
let reader = registry.admit_read(SessionId::new()).expect("reader");
assert!(StorageAdmissionRegistry::admit_session_maintenance(root.path(), id).is_err());
drop(reader);
let before = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("time")
.as_millis();
drop(
StorageAdmissionRegistry::admit_session_maintenance(root.path(), id)
.expect("automatic recovery"),
);
let access =
crate::storage_access::observe_session_access(root.path(), id).expect("access");
let crate::storage_access::StorageAccessObservation::Recorded(record) = access else {
panic!("recorded age");
};
assert!(u128::from(record.observed_at_ms) >= before);
drop(
StorageAdmissionRegistry::admit_session_maintenance(root.path(), id)
.expect("idempotent"),
);
assert_eq!(
crate::storage_access::observe_session_access(root.path(), id).expect("same access"),
access
);
drop(registry.admit_read(SessionId::new()).expect("abandoned"));
let path = root
.path()
.join("storage-admission-sessions-v1")
.join(id.to_string())
.join("unknown");
std::fs::write(&path, b"future").expect("fixture");
assert!(StorageAdmissionRegistry::admit_session_maintenance(root.path(), id).is_err());
assert_eq!(std::fs::read(&path).expect("preserved"), b"future");
assert_eq!(
crate::storage_access::observe_session_access(root.path(), id).expect("same access"),
access
);
}
#[tokio::test]
async fn session_recovery_requires_released_readers_and_resets_age() {
let root = tempfile::tempdir().expect("root");
let sessions = crate::SessionManager::persistent(root.path()).expect("sessions");
let id = sessions
.create_session(None, root.path().to_path_buf())
.await
.expect("session")
.id;
sessions
.release_session_ownership(id)
.await
.expect("release");
let registry = StorageAdmissionRegistry::open_session(root.path(), id).expect("registry");
let reader = registry.admit_read(SessionId::new()).expect("reader");
assert!(
StorageAdmissionRegistry::session_report(root.path(), id, true)
.expect("busy")
.busy
);
drop(reader);
let observation =
crate::storage_access::observe_session_access(root.path(), id).expect("before");
let preview =
StorageAdmissionRegistry::session_report(root.path(), id, false).expect("preview");
assert_eq!(preview.abandoned, 1);
assert_eq!(preview.retired, 0);
assert_eq!(
crate::storage_access::observe_session_access(root.path(), id).expect("unchanged"),
observation
);
assert!(registry.admit_maintenance(4096).is_err());
let recovered =
StorageAdmissionRegistry::session_report(root.path(), id, true).expect("recover");
assert_eq!(recovered.retired, 1);
assert!(matches!(
crate::storage_access::observe_session_access(root.path(), id).expect("reset"),
crate::storage_access::StorageAccessObservation::Recorded(_)
));
drop(registry.admit_maintenance(4096).expect("admitted"));
assert_eq!(
StorageAdmissionRegistry::session_report(root.path(), id, true)
.expect("repeat")
.retired,
0
);
drop(registry.admit_read(SessionId::new()).expect("reader"));
let directory = root
.path()
.join("storage-admission-sessions-v1")
.join(id.to_string());
std::fs::write(directory.join("future-format"), b"unknown").expect("fixture");
let invalid =
StorageAdmissionRegistry::session_report(root.path(), id, true).expect("preserved");
assert!(invalid.invalid);
assert_eq!(invalid.retired, 0);
assert_eq!(
std::fs::read(directory.join("future-format")).expect("preserved"),
b"unknown"
);
}
use super::*;
#[test]
fn symlinked_root_is_rejected_without_initializing_target() {
let root = tempfile::tempdir().expect("root");
let target = root.path().join("target");
std::fs::create_dir(&target).expect("target");
let alias = root.path().join("alias");
std::os::unix::fs::symlink(&target, &alias).expect("alias");
assert!(StorageAdmissionRegistry::open(&alias).is_err());
assert_eq!(std::fs::read_dir(&target).expect("untouched").count(), 0);
let registry = StorageAdmissionRegistry::open(&target).expect("direct root");
drop(registry.admit_maintenance(16).expect("healthy direct root"));
}
#[test]
fn retiring_clean_participants_bounds_registry_growth_without_clearing_damage() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
for _ in 0..20 {
let id = SessionId::new();
registry
.admit_read(id)
.expect("read")
.complete()
.expect("complete");
registry.retire(id).expect("retire");
}
drop(
registry
.admit_maintenance(1)
.expect("only coordinator remains"),
);
let dirty = SessionId::new();
let active = registry.admit_read(dirty).expect("active");
assert!(registry.retire(dirty).is_err());
drop(active);
assert!(registry.retire(dirty).is_err());
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn completed_reads_retire_while_other_readers_remain_active() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let long_id = SessionId::new();
let long = registry.admit_read(long_id).expect("long read");
for _ in 0..100 {
let id = SessionId::new();
let short = registry.admit_read(id).expect("short read");
registry
.complete_read(id, short)
.expect("retire without exclusive upgrade");
}
assert_eq!(
std::fs::read_dir(root.path().join("storage-admission-v1"))
.expect("registry")
.count(),
2
);
registry
.complete_read(long_id, long)
.expect("long complete");
drop(
registry
.admit_maintenance(1)
.expect("no leaked participants"),
);
}
#[test]
fn completion_rejects_wrong_participant_identity() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let a = SessionId::new();
let b = SessionId::new();
let first = registry.admit_read(a).expect("a");
let second = registry.admit_read(b).expect("b");
assert!(registry.complete_read(b, first).is_err());
drop(second);
assert!(registry.admit_maintenance(8).is_err());
assert_eq!(
std::fs::read_dir(root.path().join("storage-admission-v1"))
.expect("registry")
.count(),
3
);
}
#[test]
fn durable_fallback_blocker_survives_restart_and_is_idempotent() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
registry.block_maintenance_for_fallback().expect("block");
registry.block_maintenance_for_fallback().expect("repeat");
assert!(registry.admit_maintenance(4096).is_err());
let id = SessionId::new();
let read = registry.admit_read(id).expect("read remains available");
registry.complete_read(id, read).expect("complete");
drop(registry);
let registry = StorageAdmissionRegistry::open(root.path()).expect("restart");
assert!(registry.admit_maintenance(4096).is_err());
assert!(
root.path()
.join("storage-admission-v1/unregistered-read.blocked")
.is_file()
);
}
#[test]
fn fallback_cannot_authorize_read_while_maintenance_is_active() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let maintenance = registry.admit_maintenance(4096).expect("maintenance");
assert!(registry.block_maintenance_for_fallback().is_err());
drop(maintenance);
registry
.block_maintenance_for_fallback()
.expect("after maintenance");
assert!(registry.admit_maintenance(4096).is_err());
}
#[test]
fn daemon_registration_blocks_scan_until_clean_completion_and_survives_failure() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let daemon = registry.register_daemon(SessionId::new()).expect("daemon");
assert!(registry.admit_maintenance(16).is_err());
let read_id = SessionId::new();
let read = registry.admit_read(read_id).expect("reads coexist");
registry
.complete_read(read_id, read)
.expect("complete read");
daemon.finish().expect("healthy drained daemon");
drop(registry.admit_maintenance(16).expect("clean scan"));
let mut failed = registry
.register_daemon(SessionId::new())
.expect("second daemon");
failed.fail();
assert!(failed.finish().is_err());
drop(registry);
let registry = StorageAdmissionRegistry::open(root.path()).expect("restart");
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn live_token_rejects_corrupted_durable_registration_without_repair() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let id = SessionId::new();
let daemon = registry.register_daemon(id).expect("daemon");
let path = root
.path()
.join("storage-admission-v1")
.join(format!("{id}.daemon"));
for content in [
b"BCSTDAEMON2:LIVE!".as_slice(),
b"BCSTDAEMON1:DONE!",
b"truncated",
] {
std::fs::write(&path, content).expect("corrupt fixture");
let token = daemon
.acknowledgement()
.expect("local health alone is insufficient");
assert!(registry.admit_owned(16, token).is_err());
assert_eq!(std::fs::read(&path).expect("preserved"), content);
}
drop(daemon);
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn owned_acknowledgement_keeps_liveness_and_observes_health_failure() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let mut daemon = registry.register_daemon(SessionId::new()).expect("daemon");
let admission = registry
.admit_owned(16, daemon.acknowledgement().expect("token"))
.expect("owned admission");
admission.check().expect("healthy");
daemon.fail();
assert!(admission.check().is_err());
drop(daemon);
assert!(registry.admit_read(SessionId::new()).is_err());
drop(admission);
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn live_acknowledgement_is_exact_and_never_covers_foreign_or_failed_daemons() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let mut local = registry.register_daemon(SessionId::new()).expect("local");
assert!(registry.admit_maintenance(16).is_err());
{
let admission = registry
.admit_acknowledged(16, &mut local)
.expect("live healthy acknowledgement");
assert!(registry.admit_read(SessionId::new()).is_err());
drop(admission);
}
let foreign = registry.register_daemon(SessionId::new()).expect("foreign");
assert!(registry.admit_acknowledged(16, &mut local).is_err());
foreign.finish().expect("foreign completed");
drop(
registry
.admit_acknowledged(16, &mut local)
.expect("foreign clean"),
);
local.fail();
assert!(registry.admit_acknowledged(16, &mut local).is_err());
}
#[test]
fn live_acknowledgement_from_another_registry_is_rejected() {
let root = tempfile::tempdir().expect("root");
let other = tempfile::tempdir().expect("other");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let other_registry = StorageAdmissionRegistry::open(other.path()).expect("other registry");
let mut other_daemon = other_registry
.register_daemon(SessionId::new())
.expect("other daemon");
assert!(registry.admit_acknowledged(16, &mut other_daemon).is_err());
}
#[test]
fn completed_daemon_cleanup_preserves_active_abandoned_and_unknown_evidence() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
for _ in 0..3 {
registry
.register_daemon(SessionId::new())
.expect("register")
.finish()
.expect("clean");
}
let active_id = SessionId::new();
let active = registry.register_daemon(active_id).expect("active");
let abandoned_id = SessionId::new();
drop(registry.register_daemon(abandoned_id).expect("abandoned"));
let unknown = root.path().join("storage-admission-v1/future-format");
std::fs::write(&unknown, b"preserve").expect("unknown");
assert!(registry.retire_completed_daemons(2).is_err());
assert_eq!(registry.retire_completed_daemons(16).expect("cleanup"), 3);
assert_eq!(registry.retire_completed_daemons(16).expect("repeat"), 0);
assert!(
root.path()
.join("storage-admission-v1")
.join(format!("{active_id}.daemon"))
.exists()
);
assert!(
root.path()
.join("storage-admission-v1")
.join(format!("{abandoned_id}.daemon"))
.exists()
);
assert_eq!(std::fs::read(unknown).expect("preserved"), b"preserve");
drop(active);
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn complete_live_set_admits_maintenance_and_any_failure_invalidates_it() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let first = registry.register_daemon(SessionId::new()).expect("first");
let mut second = registry.register_daemon(SessionId::new()).expect("second");
let a = first.acknowledgement().expect("a");
let b = second.acknowledgement().expect("b");
assert!(registry.admit_owned_set(16, vec![a.clone()]).is_err());
assert!(
registry
.admit_owned_set(16, vec![a.clone(), a.clone(), b.clone()])
.is_err()
);
let admission = registry
.admit_owned_set(16, vec![a, b])
.expect("complete live set");
admission.check().expect("healthy");
second.fail();
assert!(admission.check().is_err());
drop(admission);
drop(first);
drop(second);
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn registry_tracks_multiple_readers_and_dirty_restart() {
let root = tempfile::tempdir().expect("root");
let first = StorageAdmissionRegistry::open(root.path()).expect("first");
let second = StorageAdmissionRegistry::open(root.path()).expect("second");
let a = first.admit_read(SessionId::new()).expect("a");
let b = second.admit_read(SessionId::new()).expect("b");
assert!(first.admit_maintenance(16).is_err());
a.complete().expect("a done");
b.complete().expect("b done");
drop(first.admit_maintenance(16).expect("clean"));
drop(second.admit_read(SessionId::new()).expect("abandoned"));
drop(first);
drop(second);
let reopened = StorageAdmissionRegistry::open(root.path()).expect("reopen");
assert!(reopened.admit_maintenance(16).is_err());
reopened
.admit_read(SessionId::new())
.expect("unrelated read")
.complete()
.expect("complete");
}
#[test]
fn incomplete_or_unknown_registry_never_grants_maintenance() {
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
for _ in 0..3 {
registry
.admit_read(SessionId::new())
.expect("read")
.complete()
.expect("done");
}
assert!(registry.admit_maintenance(2).is_err());
drop(registry.admit_maintenance(4).expect("complete scan"));
std::fs::write(root.path().join("storage-admission-v1/future"), b"unknown")
.expect("unknown");
assert!(registry.admit_maintenance(16).is_err());
}
#[test]
fn registry_rejects_symlink_and_hardlink_participants() {
use std::os::unix::fs::symlink;
let root = tempfile::tempdir().expect("root");
let registry = StorageAdmissionRegistry::open(root.path()).expect("registry");
let outside = tempfile::NamedTempFile::new().expect("outside");
let id = SessionId::new();
let path = root
.path()
.join("storage-admission-v1")
.join(format!("{id}.participant"));
symlink(outside.path(), &path).expect("symlink");
assert!(registry.admit_read(id).is_err());
std::fs::remove_file(&path).expect("fixture cleanup");
std::fs::hard_link(outside.path(), &path).expect("hardlink");
assert!(registry.admit_read(id).is_err());
assert_eq!(outside.as_file().metadata().expect("metadata").len(), 0);
}
}