repositories / smith
smith
There are many coding harnesses - but this one is fast
owned by admin
smith-core/src/store.rs
Raw//! Durable session storage: append-only writer, incremental bounded loader,
//! and atomic repair publication (`SMH-SPEC-SPEC0001`, Sessions).
use crate::frame::{
FILE_PROLOGUE, Frame, Recovery, encode_entry, index_bytes, visit_frames,
visit_frames_reporting, visit_metadata_reporting, write_frame,
};
use crate::session::{Entry, EntryFrame, Session};
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::{EntryId, SessionId};
use smith::tool::CancelHandle;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
/// Exclusive access to one session, held on a sidecar lock file.
///
/// The lock lives beside the session file rather than on it, because repair
/// publishes a replacement inode: a lock taken on the data file would leave
/// a waiter holding a descriptor on the replaced inode, and its appends
/// would vanish. The sidecar is never replaced, so every participant
/// serializes on the same object and opens the data file only after winning.
#[derive(Debug)]
struct SessionLock {
file: File,
path: PathBuf,
}
impl SessionLock {
/// Acquire the lock for `path`, creating parent directories and the
/// sidecar as needed. Blocks while another holder owns the session.
fn acquire(path: &Path) -> Result<Self> {
Self::lock(path, false)
}
fn try_acquire(path: &Path) -> Result<Self> {
Self::lock(path, true)
}
fn lock(path: &Path, nonblocking: bool) -> Result<Self> {
if let Some(parent) = path.parent()
&& !parent.as_os_str().is_empty()
{
std::fs::create_dir_all(parent)
.map_err(io_error(smith::error::StorageCode::StoreMkdir, parent))?;
}
let path = canonical_session_path(path)?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path(&path))
.map_err(io_error(
smith::error::StorageCode::StoreLockOpen,
&lock_path(&path),
))?;
if nonblocking {
fs2::FileExt::try_lock_exclusive(&file).map_err(io_error(
smith::error::StorageCode::StoreBusy,
&lock_path(&path),
))?;
} else {
fs2::FileExt::lock_exclusive(&file).map_err(io_error(
smith::error::StorageCode::StoreLock,
&lock_path(&path),
))?;
}
Ok(Self { file, path })
}
fn path(&self) -> &Path {
&self.path
}
}
impl Drop for SessionLock {
fn drop(&mut self) {
// Best-effort unlock; the OS also releases on close.
let _unlock_result: std::io::Result<()> = fs2::FileExt::unlock(&self.file);
}
}
fn scratch_file(operation: &str) -> Result<(PathBuf, File)> {
let (path, file) = new_temp_file(
&std::env::temp_dir().join("smith-scratch"),
operation,
SessionId::new(),
)?;
std::fs::remove_file(&path).map_err(io_error(
smith::error::StorageCode::StoreScratchRemove,
&path,
))?;
Ok((path, file))
}
/// Sidecar lock file path for a session file.
fn readers_path(path: &Path) -> PathBuf {
let mut name = path.as_os_str().to_os_string();
name.push(".readers.lock");
PathBuf::from(name)
}
fn lock_path(path: &Path) -> PathBuf {
let mut name = path.as_os_str().to_os_string();
name.push(".lock");
PathBuf::from(name)
}
/// Resolve existing data symlinks, or a new file's parent symlinks, before
/// deriving sidecar paths.
///
/// Unix rejects data files with multiple hard links.
/// Bind mounts and non-Unix hard-link aliases have no portable safe identity
/// query, so callers must not expose one session through them.
fn canonical_session_path(path: &Path) -> Result<PathBuf> {
match std::fs::canonicalize(path) {
Ok(path) => {
reject_ambiguous_data_alias(&path)?;
Ok(path)
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
if std::fs::symlink_metadata(path).is_ok() {
return Err(io_error(smith::error::StorageCode::StoreCanonicalize, path)(error));
}
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
let parent = std::fs::canonicalize(parent)
.map_err(io_error(smith::error::StorageCode::StoreCanonicalize, path))?;
let name = path.file_name().ok_or_else(|| {
storage_fault(
smith::error::StorageCode::StoreCanonicalize,
path,
format!("{} has no file name", path.display()),
)
})?;
Ok(parent.join(name))
}
Err(error) => Err(io_error(smith::error::StorageCode::StoreCanonicalize, path)(error)),
}
}
#[cfg(unix)]
fn reject_ambiguous_data_alias(path: &Path) -> Result<()> {
use std::os::unix::fs::MetadataExt;
let metadata =
std::fs::metadata(path).map_err(io_error(smith::error::StorageCode::StoreStat, path))?;
if metadata.is_file() && metadata.nlink() > 1 {
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionAmbiguousAlias,
message: format!("{} has multiple hard links", path.display()),
});
}
Ok(())
}
#[cfg(not(unix))]
fn reject_ambiguous_data_alias(_: &Path) -> Result<()> {
Ok(())
}
fn validate_named_identity(path: &Path, session_id: SessionId) -> Result<()> {
if let Some(named_id) = path
.file_stem()
.and_then(|stem| stem.to_str())
.and_then(|stem| stem.parse::<SessionId>().ok())
&& named_id != session_id
{
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionIdMismatch,
message: format!(
"{} names session {named_id}, not {session_id}",
path.display()
),
});
}
Ok(())
}
fn reject_existing_target(path: &Path) -> Result<()> {
match std::fs::symlink_metadata(path) {
Ok(_) => Err(SmithError::Session {
code: smith::error::SessionCode::SessionExists,
message: format!("{} already exists", path.display()),
}),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(io_error(smith::error::StorageCode::CreateStat, path)(error)),
}
}
#[cfg(test)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum PublicationCheckpoint {
TempCreated,
PartialWrite,
CompletedWrite,
TempSynced,
Published,
ParentSynced,
Cleanup,
}
#[cfg(test)]
type PublicationHook = Box<dyn FnMut(PublicationCheckpoint)>;
#[cfg(test)]
thread_local! {
static PUBLICATION_HOOK: std::cell::RefCell<Option<PublicationHook>> = const { std::cell::RefCell::new(None) };
static PARENT_SYNC_FAILURE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
#[cfg(test)]
fn run_publication_hook(point: PublicationCheckpoint) {
PUBLICATION_HOOK.with(|hook| {
if let Some(hook) = hook.borrow_mut().as_mut() {
hook(point);
}
});
}
macro_rules! publication_checkpoint {
($point:ident) => {{
#[cfg(test)]
run_publication_hook(PublicationCheckpoint::$point);
}};
}
#[cfg(test)]
fn set_publication_hook(hook: impl FnMut(PublicationCheckpoint) + 'static) {
PUBLICATION_HOOK.with(|slot| *slot.borrow_mut() = Some(Box::new(hook)));
}
#[cfg(test)]
fn clear_publication_hook() {
PUBLICATION_HOOK.with(|slot| *slot.borrow_mut() = None);
}
#[cfg(test)]
fn set_parent_sync_failure(fail: bool) {
PARENT_SYNC_FAILURE.with(|slot| slot.set(fail));
}
fn new_temp_file(path: &Path, operation: &str, id: SessionId) -> Result<(PathBuf, File)> {
let name = path.file_name().ok_or_else(|| {
storage_fault(
smith::error::StorageCode::CreateTmp,
path,
format!("{} has no file name", path.display()),
)
})?;
let mut last = path.to_path_buf();
for attempt in 0..128 {
let mut temporary = name.to_os_string();
temporary.push(format!(".smith-{operation}-{id}-{attempt}.tmp"));
let temporary = path.with_file_name(temporary);
last.clone_from(&temporary);
let mut options = OpenOptions::new();
options.read(true).write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
match options.open(&temporary) {
Ok(file) => return Ok((temporary, file)),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => {
return Err(io_error(smith::error::StorageCode::CreateTmp, &temporary)(
error,
));
}
}
}
Err(storage_fault(
smith::error::StorageCode::CreateTmp,
&last,
format!(
"could not allocate a unique temporary beside {}",
path.display()
),
))
}
/// Append-only session writer holding exclusive access to its session.
///
/// Opening repairs a torn tail (truncating to the last complete frame
/// boundary) before appending, so acknowledged appends are always
/// reachable by the loader. The lock serializes writers against each other
/// and against repair.
#[derive(Debug)]
pub struct SessionWriter {
file: File,
config: Config,
path: PathBuf,
/// A failed write or sync may have appended an unreachable frame prefix.
poisoned: bool,
/// Last known entry boundary, used only for reconstructible hints.
last: Option<(EntryId, u64)>,
/// Hints since the last reconstructible selected-tip checkpoint.
since_checkpoint: u16,
/// Held for the writer's lifetime; released on drop.
_lock: SessionLock,
}
impl SessionWriter {
/// Open (creating parent directories) and lock the session file.
///
/// # Errors
///
/// Returns a configuration, locking, or input/output error.
pub fn open<P: AsRef<Path>>(path: P, config: Config) -> Result<Self> {
Self::open_with_recovery(path, config).map(|(writer, _)| writer)
}
/// Open and lock the session file, returning bounded damage/tail metadata.
/// No frame payloads are retained by the durable open path.
fn open_with_recovery<P: AsRef<Path>>(path: P, config: Config) -> Result<(Self, Recovery)> {
// Win the lock before opening the data file, so a repair that is
// publishing a replacement cannot hand out a stale descriptor.
let lock = SessionLock::try_acquire(path.as_ref())?;
Self::attach(lock.path().to_path_buf(), config, lock)
}
/// Take over an already-locked session file.
fn attach(path: PathBuf, config: Config, lock: SessionLock) -> Result<(Self, Recovery)> {
// Validation already names the offending field path.
config.validate()?;
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)
.map_err(io_error(smith::error::StorageCode::StoreOpen, &path))?;
// Drop a torn tail before appending so new frames stay reachable.
// Complete but damaged frames are left in place and stay reported.
if file
.metadata()
.map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
.len()
== 0
{
file.write_all(&FILE_PROLOGUE)
.map_err(io_error(smith::error::StorageCode::StoreWrite, &path))?;
file.sync_data()
.map_err(io_error(smith::error::StorageCode::StoreSync, &path))?;
}
let recovery = walk_file_bounded(&file, &path, config.frame_bytes_max)?;
let (first, _) = SessionReader::open(&path)?.first(config.frame_bytes_max)?;
if let Some(named_id) = path
.file_stem()
.and_then(|stem| stem.to_str())
.and_then(|stem| stem.parse::<smith::id::SessionId>().ok())
&& let Some(Frame::Header { header }) = first
&& header.session_id != named_id
{
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionIdMismatch,
message: format!(
"{} records session {}, not {named_id}",
path.display(),
header.session_id
),
});
}
let len = file
.metadata()
.map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
.len();
if recovery.good_end < len {
let coordination = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(readers_path(&path))
.map_err(io_error(
smith::error::StorageCode::StoreReaderLockOpen,
&readers_path(&path),
))?;
fs2::FileExt::try_lock_exclusive(&coordination).map_err(io_error(
smith::error::StorageCode::StoreReadersActive,
&readers_path(&path),
))?;
file.set_len(recovery.good_end)
.map_err(io_error(smith::error::StorageCode::StoreTruncate, &path))?;
file.sync_data()
.map_err(io_error(smith::error::StorageCode::StoreSync, &path))?;
fs2::FileExt::unlock(&coordination).map_err(io_error(
smith::error::StorageCode::StoreReaderUnlock,
&readers_path(&path),
))?;
}
let since_checkpoint = recovery_checkpoint_count(&file, &path, config.frame_bytes_max)?;
file.seek(SeekFrom::Start(recovery.good_end))
.map_err(io_error(smith::error::StorageCode::StoreSeek, &path))?;
Ok((
Self {
file,
config,
path,
poisoned: false,
last: None,
since_checkpoint,
_lock: lock,
},
recovery,
))
}
/// Append one frame. The encoded size is checked against the configured
/// limit before any bytes are written; a rejected frame leaves the file
/// unchanged.
///
/// # Errors
///
/// Returns a framing error for oversized frames, or an input/output error.
/// A write or sync error poisons this writer; later appends fail.
pub fn append_frame(&mut self, frame: &Frame) -> Result<()> {
let bytes = frame.encode(self.config.frame_bytes_max)?;
self.write(&bytes)
}
/// Append one session record as an entry frame, framing its arena
/// `content` bytes directly. Size limits apply as in
/// [`SessionWriter::append_frame`].
///
/// # Errors
///
/// Returns a framing error for oversized frames, or an input/output error.
/// A write or sync error poisons this writer; later appends fail.
pub fn append_entry(&mut self, entry: &Entry, content: &[u8]) -> Result<()> {
let bytes = encode_entry(entry, content, self.config.frame_bytes_max)?;
self.write_entry_bytes(&bytes, entry.id, entry.parent)
}
#[expect(
clippy::large_types_passed_by_value,
reason = "Copy parent ID is consumed once to resolve an index hint"
)]
fn write_entry_bytes(
&mut self,
bytes: &[u8],
id: EntryId,
parent: Option<EntryId>,
) -> Result<()> {
let parent_offset = parent.and_then(|id| {
self.last
.filter(|(last, _)| *last == id)
.map(|(_, offset)| offset)
});
let offset = self
.file
.stream_position()
.map_err(io_error(smith::error::StorageCode::StoreSeek, &self.path))?;
self.write(bytes)?;
self.last = Some((id, offset));
if (parent.is_none() || parent_offset.is_some()) && self.config.frame_bytes_max >= 48 {
self.write(&index_bytes(id, offset, parent, parent_offset))?;
}
self.since_checkpoint += 1;
if self.since_checkpoint == 64 && self.config.frame_bytes_max >= 24 {
self.append_frame(&Frame::Checkpoint { id, offset })?;
self.since_checkpoint = 0;
}
Ok(())
}
/// Persist a branch-tip selection, including when no new entry was appended.
///
/// # Errors
/// Returns `ENTRY_NOT_FOUND` for an absent entry or a storage fault.
pub fn select_entry(&mut self, id: EntryId) -> Result<()> {
let (_, offset) = SessionReader::open(&self.path)?
.find_entry(id, self.config.frame_bytes_max)?
.ok_or_else(|| missing_entry(id))?;
self.append_frame(&Frame::Selection { id })?;
if self.config.frame_bytes_max >= 24 {
self.append_frame(&Frame::Checkpoint { id, offset })?;
self.since_checkpoint = 0;
}
self.last = Some((id, offset));
Ok(())
}
fn write(&mut self, bytes: &[u8]) -> Result<()> {
self.io_step(smith::error::StorageCode::StoreWrite, |file| {
file.write_all(bytes)
})?;
self.io_step(smith::error::StorageCode::StoreSync, |file| {
file.sync_data()
})
}
fn io_step(
&mut self,
code: smith::error::StorageCode,
step: impl FnOnce(&mut File) -> std::io::Result<()>,
) -> Result<()> {
if self.poisoned {
return Err(storage_fault(
smith::error::StorageCode::StorePoisoned,
&self.path,
format!("{} may contain a partial append", self.path.display()),
));
}
match step(&mut self.file) {
Ok(()) => Ok(()),
Err(error) => {
self.poisoned = true;
Err(io_error(code, &self.path)(error))
}
}
}
#[cfg(test)]
fn write_step(&mut self, step: impl FnOnce(&mut File) -> Result<()>) -> Result<()> {
if self.poisoned {
return Err(storage_fault(
smith::error::StorageCode::StorePoisoned,
&self.path,
format!("{} may contain a partial append", self.path.display()),
));
}
match step(&mut self.file) {
Ok(()) => Ok(()),
Err(error) => {
self.poisoned = true;
Err(error)
}
}
}
/// Path of the underlying session file.
#[must_use]
pub fn path(&self) -> &Path {
&self.path
}
}
/// A session file opened for use.
#[derive(Debug)]
pub struct OpenSession {
/// Restored, or newly created, session state.
pub session: Session,
/// Locked writer positioned after the last complete frame.
pub writer: SessionWriter,
/// Damage and torn-tail metadata found while loading; `frames` is empty
/// on bounded durable open, not a second copy of session history.
pub recovery: Recovery,
}
/// Open a session file, writing a header frame when the file is new.
///
/// Reopening restores the persisted session and branch IDs, so state
/// reconstructed from the same file is identical on every load. A file that
/// holds frames but no header cannot be attributed to a session and fails
/// explicitly instead of silently becoming a new session.
///
/// # Errors
///
/// Returns [`SmithError::Session`] when the file holds frames but no header,
/// or a locking, framing, or input/output error.
pub fn open_session(path: &Path, config: Config) -> Result<OpenSession> {
let (mut writer, recovery) = SessionWriter::open_with_recovery(path, config)?;
let reader = SessionReader::open(&writer.path)?;
let session = if recovery.good_end > FILE_PROLOGUE.len() as u64 {
reader.selected_session(writer.config.frame_bytes_max)?
} else if recovery.damage.is_empty() {
let session = path
.file_stem()
.and_then(|stem| stem.to_str())
.and_then(|stem| stem.parse().ok())
.map_or_else(Session::new, Session::new_with_id);
writer.append_frame(&Frame::Header {
header: session.header(),
})?;
session
} else {
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionHeaderMissing,
message: format!("{} holds frames but no session header", path.display()),
});
};
if let Some(id) = session.selected_id() {
writer.last = reader
.find_entry(id, writer.config.frame_bytes_max)?
.map(|(_, offset)| (id, offset));
}
Ok(OpenSession {
session,
writer,
recovery,
})
}
/// Publish `session` as a new session file and open it for appending.
///
/// The file is written completely and then published by a no-replace hard
/// link, so an interrupted fork leaves no half-written session behind and an
/// existing path is never overwritten.
///
/// # Errors
///
/// Returns [`SmithError::Session`] when the target already holds a session,
/// or a locking, framing, or input/output error.
pub fn create_session(path: &Path, session: &Session, config: Config) -> Result<OpenSession> {
config.validate()?;
if session
.active_branch()
.entries()
.first()
.is_some_and(|root| root.parent.is_some())
{
return Err(session_fault(
smith::error::SessionCode::SessionIncompleteContext,
"effective context omits earlier ancestors; fork through SessionReader",
));
}
let lock = SessionLock::acquire(path)?;
let path = lock.path().to_path_buf();
validate_named_identity(&path, session.id())?;
reject_existing_target(&path)?;
let (temporary, mut out) = new_temp_file(&path, "create", session.id())?;
publication_checkpoint!(TempCreated);
let written = (|| {
out.write_all(&FILE_PROLOGUE)
.map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
publication_checkpoint!(PartialWrite);
write_frame(
&mut out,
&Frame::Header {
header: session.header(),
},
config.frame_bytes_max,
)?;
for entry in session.active_branch().entries() {
let bytes = encode_entry(entry, session.bytes(entry), config.frame_bytes_max)?;
out.write_all(&bytes)
.map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
}
if let Some(id) = session.selected_id()
&& session
.active_branch()
.entries()
.last()
.map(|entry| entry.id)
!= Some(id)
{
write_frame(&mut out, &Frame::Selection { id }, config.frame_bytes_max)?;
}
publication_checkpoint!(CompletedWrite);
out.sync_all()
.map_err(io_error(smith::error::StorageCode::CreateSync, &temporary))?;
publication_checkpoint!(TempSynced);
Ok(())
})();
drop(out);
if let Err(error) = written {
// Preserve the write failure; temporary-file removal is best-effort cleanup.
let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
return Err(error);
}
publish_no_clobber(&temporary, &path)?;
publication_checkpoint!(Cleanup);
let (writer, recovery) = SessionWriter::attach(path, config, lock)?;
let session =
SessionReader::open(writer.path())?.selected_session(writer.config.frame_bytes_max)?;
Ok(OpenSession {
session,
writer,
recovery,
})
}
/// Visit frames within the file size observed at open, without retaining history.
///
/// A live incomplete tail is reported but never repaired by this read-only path.
/// Callers must inspect damage and torn tails before treating a scan as complete.
///
/// # Errors
/// Returns an input/output error or the visitor's error.
pub fn scan_session(
path: &Path,
frame_bytes_max: usize,
visit: impl FnMut(Frame) -> Result<()>,
) -> Result<Recovery> {
SessionReader::open(path)?.scan(frame_bytes_max, visit)
}
/// Whether the per-session exclusive writer lock is currently held.
///
/// # Errors
/// Returns an I/O error if the lock file cannot be probed.
pub fn writer_active(path: &Path) -> Result<bool> {
let path = canonical_session_path(path)?;
let file = match OpenOptions::new()
.read(true)
.write(true)
.open(lock_path(&path))
{
Ok(file) => file,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(e) => {
return Err(io_error(
smith::error::StorageCode::StoreLockOpen,
&lock_path(&path),
)(e));
}
};
match fs2::FileExt::try_lock_exclusive(&file) {
Ok(()) => {
fs2::FileExt::unlock(&file).map_err(io_error(
smith::error::StorageCode::StoreUnlock,
&lock_path(&path),
))?;
Ok(false)
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(true),
Err(e) => Err(io_error(
smith::error::StorageCode::StoreLock,
&lock_path(&path),
)(e)),
}
}
/// A pinned file generation with a shared maintenance lock, independent of writer ownership.
/// The snapshot length is captured at scan start; subsequent appends require a new scan.
#[derive(Debug)]
pub struct SessionReader {
file: File,
readers: File,
snapshot_len: u64,
writer_active: bool,
}
impl Drop for SessionReader {
fn drop(&mut self) {
// A fork shares this lock's open-file description until exec. Unlock it
// explicitly so a dropped snapshot cannot delay tail repair in that window.
let _unlock_result: std::io::Result<()> = fs2::FileExt::unlock(&self.readers);
}
}
struct SnapshotRead<'a> {
file: &'a File,
at: u64,
end: u64,
}
impl Read for SnapshotRead<'_> {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
let available = usize::try_from(self.end.saturating_sub(self.at))
.unwrap_or(usize::MAX)
.min(buffer.len());
if available == 0 {
return Ok(0);
}
#[cfg(unix)]
let n = std::os::unix::fs::FileExt::read_at(self.file, &mut buffer[..available], self.at)?;
#[cfg(windows)]
let n =
std::os::windows::fs::FileExt::seek_read(self.file, &mut buffer[..available], self.at)?;
#[cfg(not(any(unix, windows)))]
let n = {
let mut clone = self.file.try_clone()?;
clone.seek(SeekFrom::Start(self.at))?;
clone.read(&mut buffer[..available])?
};
debug_assert!(
n <= available,
"snapshot reader must not advance beyond its validated byte window"
);
self.at += n as u64;
Ok(n)
}
}
#[cfg(test)]
thread_local! {
static METADATA_PASSES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
static PROOF_VISITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
static CHECKPOINT_HITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
// Fixed-width, offset-sorted structural proof. Scratch is unlinked before
// scanning; file size grows, resident memory does not. Only the sequential
// prologue walk can populate these records, never a magic-shaped payload.
struct Proof {
file: File,
path: PathBuf,
len: u64,
selected_checkpoint: Option<ProofEntry>,
cancel: Option<CancelHandle>,
}
#[derive(Clone, Copy)]
struct ProofEntry {
id: EntryId,
parent: Option<EntryId>,
offset: u64,
parent_offset: Option<u64>,
compaction: bool,
}
impl Proof {
const WIDTH: u64 = 49;
fn check_cancelled(&self) -> Result<()> {
if self.cancel.as_ref().is_some_and(CancelHandle::is_cancelled) {
return Err(SmithError::Cancelled);
}
Ok(())
}
fn get(&mut self, index: u64) -> Result<ProofEntry> {
self.check_cancelled()?;
#[cfg(test)]
PROOF_VISITS.with(|visits| visits.set(visits.get() + 1));
self.file
.seek(SeekFrom::Start(index * Self::WIDTH))
.map_err(io_error(
smith::error::StorageCode::StoreProofSeek,
&self.path,
))?;
let mut bytes = [0; 49];
self.file.read_exact(&mut bytes).map_err(io_error(
smith::error::StorageCode::StoreProofRead,
&self.path,
))?;
let id =
EntryId::from_u128(u128::from_le_bytes(bytes[..16].try_into().map_err(
|_| session_fault(smith::error::SessionCode::StoreProof, "short id"),
)?));
let parent =
u128::from_le_bytes(bytes[16..32].try_into().map_err(|_| {
session_fault(smith::error::SessionCode::StoreProof, "short parent")
})?);
let offset =
u64::from_le_bytes(bytes[32..40].try_into().map_err(|_| {
session_fault(smith::error::SessionCode::StoreProof, "short offset")
})?);
let parent_offset = u64::from_le_bytes(
bytes[40..48]
.try_into()
.map_err(|_| session_fault(smith::error::SessionCode::StoreProof, "short hint"))?,
);
Ok(ProofEntry {
id,
parent: (parent != 0).then(|| EntryId::from_u128(parent)),
offset,
parent_offset: (parent_offset != 0).then_some(parent_offset),
compaction: bytes[48] != 0,
})
}
fn push(&mut self, entry: &ProofEntry) -> Result<()> {
self.check_cancelled()?;
self.file
.seek(SeekFrom::Start(self.len * Self::WIDTH))
.map_err(io_error(
smith::error::StorageCode::StoreProofSeek,
&self.path,
))?;
self.file
.write_all(&entry.id.as_u128().to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))?;
self.file
.write_all(&entry.parent.map_or(0, |id| id.as_u128()).to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))?;
self.file
.write_all(&entry.offset.to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))?;
self.file
.write_all(&entry.parent_offset.unwrap_or(0).to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))?;
self.file
.write_all(&[u8::from(entry.compaction)])
.map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))?;
self.len += 1;
Ok(())
}
fn set_hint(&mut self, at: u64) -> Result<()> {
self.check_cancelled()?;
self.file
.seek(SeekFrom::Start((self.len - 1) * Self::WIDTH + 40))
.map_err(io_error(
smith::error::StorageCode::StoreProofSeek,
&self.path,
))?;
self.file.write_all(&at.to_le_bytes()).map_err(io_error(
smith::error::StorageCode::StoreProofWrite,
&self.path,
))
}
fn find(&mut self, id: EntryId) -> Result<Option<ProofEntry>> {
let mut found = None;
for i in 0..self.len {
self.check_cancelled()?;
let record = self.get(i)?;
if record.id == id {
if found.is_some() {
return Err(session_fault(
smith::error::SessionCode::SessionDuplicateId,
format!("duplicate entry {id}"),
));
}
found = Some(record);
}
}
Ok(found)
}
fn at_offset(&mut self, offset: u64) -> Result<Option<ProofEntry>> {
let (mut start, mut end) = (0, self.len);
while start < end {
self.check_cancelled()?;
let middle = start + (end - start) / 2;
let record = self.get(middle)?;
match record.offset.cmp(&offset) {
std::cmp::Ordering::Less => start = middle + 1,
std::cmp::Ordering::Greater => end = middle,
std::cmp::Ordering::Equal => return Ok(Some(record)),
}
}
Ok(None)
}
fn parent(&mut self, child: &ProofEntry) -> Result<Option<ProofEntry>> {
self.check_cancelled()?;
let Some(id) = child.parent else {
return Ok(None);
};
// The hint is usable only if it names a structurally proven entry,
// matches the parent identity, and precedes the child. Identity lookup also
// rejects duplicate IDs, including outside the selected branch.
let hinted = child
.parent_offset
.and_then(|at| (at < child.offset).then_some(at));
let target = hinted.map(|at| self.at_offset(at)).transpose()?.flatten();
let found = self.find(id)?.ok_or_else(|| missing_entry(id))?;
let parent = target
.filter(|candidate| candidate.id == id && candidate.offset == found.offset)
.unwrap_or(found);
if parent.offset >= child.offset {
return Err(session_fault(
smith::error::SessionCode::SessionParent,
"parent does not precede child",
));
}
Ok(Some(parent))
}
}
impl SessionReader {
const fn snapshot_read(&self, at: u64) -> SnapshotRead<'_> {
SnapshotRead {
file: &self.file,
at,
end: self.snapshot_len,
}
}
fn proof(
&self,
max: usize,
cancel: Option<CancelHandle>,
) -> Result<(Proof, crate::session::SessionHeader, Option<EntryId>)> {
let (path, file) = new_temp_file(
&std::env::temp_dir().join("smith-proof"),
"scan",
SessionId::new(),
)?;
std::fs::remove_file(&path)
.map_err(io_error(smith::error::StorageCode::StoreProofRemove, &path))?;
let mut proof = Proof {
file,
path,
len: 0,
selected_checkpoint: None,
cancel,
};
let mut header = None;
let mut selected = None;
let mut first = true;
self.scan_metadata_positions(max, |offset, frame| {
proof.check_cancelled()?;
match frame {
Frame::Header { header: h } if first => header = Some(h),
Frame::Header { .. } => {
return Err(session_fault(
smith::error::SessionCode::SessionHeaderDuplicate,
"duplicate header",
));
}
Frame::Known { entry } if header.is_some() => {
selected = Some(entry.id);
proof.selected_checkpoint = None;
proof.push(&ProofEntry {
id: entry.id,
parent: entry.parent,
offset,
parent_offset: None,
compaction: entry.kind
== (crate::session::EntryKind::Meta { compaction: true }),
})?;
}
Frame::Index {
id,
offset: target,
parent,
parent_offset,
} if header.is_some() => {
if let Some(previous) = (proof.len > 0)
.then(|| proof.get(proof.len - 1))
.transpose()?
&& previous.id == id
&& previous.offset == target
&& previous.parent == parent
&& let Some(parent_at) = parent_offset.filter(|at| *at < target)
{
proof.set_hint(parent_at)?;
}
}
Frame::Selection { id } if header.is_some() => {
selected = Some(id);
proof.selected_checkpoint = None;
}
Frame::Checkpoint { id, offset: target } if header.is_some() => {
if selected == Some(id) && target < offset {
proof.selected_checkpoint =
proof.at_offset(target)?.filter(|entry| entry.id == id);
}
}
Frame::Unknown { .. } if header.is_some() => {}
_ => {
return Err(session_fault(
smith::error::SessionCode::SessionHeaderMissing,
"missing first header",
));
}
}
first = false;
Ok(())
})?;
Ok((
proof,
header.ok_or_else(|| {
session_fault(
smith::error::SessionCode::SessionHeaderMissing,
"missing first header",
)
})?,
selected,
))
}
fn scan_metadata_positions(
&self,
max: usize,
mut visit: impl FnMut(u64, Frame) -> Result<()>,
) -> Result<Recovery> {
#[cfg(test)]
METADATA_PASSES.with(|passes| passes.set(passes.get() + 1));
let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
read_prologue(&mut reader)?;
visit_metadata_reporting(
&mut reader,
max,
|offset, frame| visit(offset + FILE_PROLOGUE.len() as u64, frame),
|_| Ok(()),
)
.map(file_offsets)
}
fn entry_at(&self, offset: u64, max: usize) -> Result<EntryFrame> {
let mut file = self.snapshot_read(offset);
let mut frame = None;
let scan = visit_frames(&mut file, max, |found| {
frame = Some(found);
false
})?;
if let Some(fault) = scan.damage.into_iter().next() {
return Err(fault);
}
match frame {
Some(Frame::Known { entry }) => Ok(entry),
_ => Err(session_fault(
smith::error::SessionCode::SessionParent,
"entry boundary no longer contains entry",
)),
}
}
/// Scan structural frame boundaries from the fixed prologue, reporting
/// absolute offsets without retaining a per-file index.
fn scan_positions(
&self,
max: usize,
mut visit: impl FnMut(u64, Frame) -> Result<()>,
) -> Result<Recovery> {
struct Counted<'a, R> {
inner: R,
bytes: &'a std::cell::Cell<u64>,
}
impl<R: Read> Read for Counted<'_, R> {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buffer)?;
self.bytes.set(self.bytes.get() + n as u64);
Ok(n)
}
}
let bytes = std::cell::Cell::new(0);
let mut reader = Counted {
inner: BufReader::with_capacity(64 * 1024, self.snapshot_read(0)),
bytes: &bytes,
};
read_prologue(&mut reader)?;
let offset = std::cell::Cell::new(FILE_PROLOGUE.len() as u64);
let mut failure = None;
let mut damage = None;
let mut scan = visit_frames_reporting(
&mut reader,
max,
|frame| {
let current = offset.replace(bytes.get());
match visit(current, frame) {
Ok(()) => true,
Err(error) => {
failure = Some(error);
false
}
}
},
|fault| {
offset.set(bytes.get());
if damage.is_none() {
damage = Some(fault);
}
true
},
)?;
if let Some(error) = failure {
return Err(error);
}
if let Some(error) = damage {
scan.damage.push(error);
}
Ok(file_offsets(scan))
}
/// Find an entry identity by walking authoritative entry frames at structural
/// boundaries; embedded indexes are hints, never the source of identity.
///
/// # Errors
/// Returns storage faults or a duplicate identity error.
pub fn find_entry(&self, id: EntryId, max: usize) -> Result<Option<(EntryFrame, u64)>> {
let mut found = None;
self.scan_metadata_positions(max, |offset, frame| {
if let Frame::Known { entry } = frame
&& entry.id == id
{
if found.is_some() {
return Err(session_fault(
smith::error::SessionCode::SessionDuplicateId,
format!("duplicate entry {id}"),
));
}
found = Some(offset);
}
Ok(())
})?;
found
.map(|offset| self.entry_at(offset, max).map(|entry| (entry, offset)))
.transpose()
}
/// Durable session and root-branch identity without materializing content.
///
/// # Errors
/// Returns a structural fault for an invalid or missing header.
pub fn header(&self, max: usize) -> Result<crate::session::SessionHeader> {
self.proof(max, None).map(|(_, header, _)| header)
}
/// Selected entry identity without materializing any content.
///
/// # Errors
/// Returns a structural fault for invalid selection or ancestry.
pub fn selected_tip(&self, max: usize) -> Result<Option<EntryId>> {
self.selected_tip_with_cancel(max, None)
}
/// Selected entry identity, checking cancellation at each bounded traversal boundary.
///
/// # Errors
/// Returns [`SmithError::Cancelled`] or a structural fault for invalid selection or ancestry.
pub fn selected_tip_cancelled(
&self,
max: usize,
cancel: &CancelHandle,
) -> Result<Option<EntryId>> {
self.selected_tip_with_cancel(max, Some(cancel.clone()))
}
fn selected_tip_with_cancel(
&self,
max: usize,
cancel: Option<CancelHandle>,
) -> Result<Option<EntryId>> {
let (mut proof, _, selected) = self.proof(max, cancel)?;
if let Some(id) = selected {
proof.find(id)?.ok_or_else(|| missing_entry(id))?;
}
Ok(selected)
}
/// Selected effective context from latest compaction boundary to tip.
/// Rootward ancestry is validated using a bounded, file-backed proof;
/// only effective payloads become resident.
///
/// # Errors
/// Returns a fault for missing ancestry, invalid header or selection.
pub fn selected_session(&self, max: usize) -> Result<Session> {
self.selected_session_with_cancel(max, None)
}
fn selected_session_with_cancel(
&self,
max: usize,
cancel: Option<CancelHandle>,
) -> Result<Session> {
let (mut proof, header, selected) = self.proof(max, cancel)?;
let (scratch_path, mut offsets) = scratch_file("context")?;
let mut length = 0u64;
if let Some(id) = selected {
let mut cursor = proof.find(id)?.ok_or_else(|| missing_entry(id))?;
if let Some(hint) = proof
.selected_checkpoint
.filter(|hint| hint.id == id && hint.offset == cursor.offset)
{
cursor = hint;
#[cfg(test)]
CHECKPOINT_HITS.with(|hits| hits.set(hits.get() + 1));
}
let mut keep = true;
loop {
proof.check_cancelled()?;
if keep {
offsets
.write_all(&cursor.offset.to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreScratchWrite,
&scratch_path,
))?;
length += 1;
if cursor.compaction {
keep = false;
}
}
match proof.parent(&cursor)? {
Some(parent) => cursor = parent,
None => break,
}
}
}
let mut session = Session::from_chain(header, Vec::new())?;
for i in (0..length).rev() {
proof.check_cancelled()?;
offsets.seek(SeekFrom::Start(i * 8)).map_err(io_error(
smith::error::StorageCode::StoreScratchSeek,
&scratch_path,
))?;
let mut bytes = [0; 8];
offsets.read_exact(&mut bytes).map_err(io_error(
smith::error::StorageCode::StoreScratchRead,
&scratch_path,
))?;
session.push_restored(self.entry_at(u64::from_le_bytes(bytes), max)?)?;
}
Ok(session)
}
/// Replay a rooted chain in chronological order with bounded retained
/// payload. One metadata pass validates structural offsets into a
/// file-backed proof; parent lookups scan fixed-width proof records.
///
/// # Errors
/// Returns a missing-entry, ancestry, storage, or callback fault.
pub fn replay_at(
&self,
id: EntryId,
max: usize,
visit: impl FnMut(EntryFrame) -> Result<()>,
) -> Result<()> {
self.replay_at_with_cancel(id, max, None, visit)
}
/// Replay a rooted chain, observing cancellation between bounded reads.
///
/// # Errors
/// Returns [`SmithError::Cancelled`], a storage fault, or a callback fault.
pub fn replay_at_cancelled(
&self,
id: EntryId,
max: usize,
cancel: &CancelHandle,
visit: impl FnMut(EntryFrame) -> Result<()>,
) -> Result<()> {
self.replay_at_with_cancel(id, max, Some(cancel.clone()), visit)
}
fn replay_at_with_cancel(
&self,
id: EntryId,
max: usize,
cancel: Option<CancelHandle>,
mut visit: impl FnMut(EntryFrame) -> Result<()>,
) -> Result<()> {
// Spill only physical addresses, never payloads. Unlink immediately:
// the open descriptor owns the scratch lifetime even on early errors.
let (mut proof, _, _) = self.proof(max, cancel)?;
let (scratch_path, mut scratch) = scratch_file("replay")?;
let mut cursor = proof.find(id)?.ok_or_else(|| missing_entry(id))?;
let mut depth = 0u64;
loop {
proof.check_cancelled()?;
scratch
.write_all(&cursor.offset.to_le_bytes())
.map_err(io_error(
smith::error::StorageCode::StoreScratchWrite,
&scratch_path,
))?;
depth += 1;
match proof.parent(&cursor)? {
Some(parent) => cursor = parent,
None => break,
}
}
for position in (0..depth).rev() {
proof.check_cancelled()?;
scratch
.seek(SeekFrom::Start(position * 8))
.map_err(io_error(
smith::error::StorageCode::StoreScratchSeek,
&scratch_path,
))?;
let mut address = [0; 8];
scratch.read_exact(&mut address).map_err(io_error(
smith::error::StorageCode::StoreScratchRead,
&scratch_path,
))?;
visit(self.entry_at(u64::from_le_bytes(address), max)?)?;
}
Ok(())
}
/// Fork one structurally reachable chain to a new independently named file.
///
/// # Errors
/// Returns a missing-entry, ancestry, publication or storage fault.
pub fn fork_at(&self, id: EntryId, path: &Path, config: Config) -> Result<OpenSession> {
self.fork_at_with_cancel(id, path, config, None)
}
/// Fork one rooted chain, observing cancellation between bounded traversal work.
///
/// # Errors
/// Returns [`SmithError::Cancelled`], a storage fault, or a publication fault.
pub fn fork_at_cancelled(
&self,
id: EntryId,
path: &Path,
config: Config,
cancel: &CancelHandle,
) -> Result<OpenSession> {
self.fork_at_with_cancel(id, path, config, Some(cancel.clone()))
}
fn fork_at_with_cancel(
&self,
id: EntryId,
path: &Path,
config: Config,
cancel: Option<CancelHandle>,
) -> Result<OpenSession> {
if cancel.as_ref().is_some_and(CancelHandle::is_cancelled) {
return Err(SmithError::Cancelled);
}
config.validate()?;
let source = path
.file_stem()
.and_then(|stem| stem.to_str())
.and_then(|stem| stem.parse::<SessionId>().ok())
.map_or_else(Session::new, Session::new_with_id);
let lock = SessionLock::acquire(path)?;
let target = lock.path().to_path_buf();
validate_named_identity(&target, source.id())?;
reject_existing_target(&target)?;
let (temporary, mut out) = new_temp_file(&target, "fork", source.id())?;
publication_checkpoint!(TempCreated);
let written = (|| {
out.write_all(&FILE_PROLOGUE)
.map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
publication_checkpoint!(PartialWrite);
write_frame(
&mut out,
&Frame::Header {
header: source.header(),
},
config.frame_bytes_max,
)?;
self.replay_at_with_cancel(id, config.frame_bytes_max, cancel.clone(), |entry| {
write_frame(&mut out, &Frame::Known { entry }, config.frame_bytes_max)
})?;
publication_checkpoint!(CompletedWrite);
out.sync_all()
.map_err(io_error(smith::error::StorageCode::CreateSync, &temporary))?;
publication_checkpoint!(TempSynced);
Ok(())
})();
drop(out);
if let Err(error) = written {
// Preserve the write failure; temporary-file removal is best-effort cleanup.
let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
return Err(error);
}
publish_no_clobber(&temporary, &target)?;
publication_checkpoint!(Cleanup);
let (writer, recovery) = SessionWriter::attach(target, config, lock)?;
let session = Self::open(writer.path())?
.selected_session_with_cancel(writer.config.frame_bytes_max, cancel)?;
Ok(OpenSession {
session,
writer,
recovery,
})
}
#[expect(
clippy::large_types_passed_by_value,
reason = "Copy IDs and offsets are compared once while validating an index record"
)]
fn verify_index(
&self,
last: Option<(EntryId, u64, Option<EntryId>)>,
id: EntryId,
target: u64,
parent: Option<EntryId>,
parent_offset: Option<u64>,
max: usize,
) -> Result<()> {
if last != Some((id, target, parent)) {
return Err(session_fault(
smith::error::SessionCode::SessionIndex,
"index target is not preceding entry boundary",
));
}
if let Some(parent) = parent {
let (_, actual) = self
.find_entry(parent, max)?
.ok_or_else(|| missing_entry(parent))?;
if parent_offset != Some(actual) || actual >= target {
return Err(session_fault(
smith::error::SessionCode::SessionIndex,
"index parent offset or identity mismatch",
));
}
} else if parent_offset.is_some() {
return Err(session_fault(
smith::error::SessionCode::SessionIndex,
"root index has parent offset",
));
}
Ok(())
}
/// Check every frame and content body in the pinned snapshot, including
/// unrelated branches and all embedded location hints.
///
/// # Errors
/// Reports damaged frames, invalid ancestry/offsets, or a torn tail.
pub fn verify_full(&self, max: usize) -> Result<()> {
let mut first = true;
let mut last_entry = None;
let scan = self.scan_positions(max, |offset, frame| {
match frame {
Frame::Header { .. } if first => {}
Frame::Header { .. } => {
return Err(session_fault(
smith::error::SessionCode::SessionHeaderDuplicate,
"duplicate header",
));
}
Frame::Known { entry } if !first => {
// Full verification rejects duplicate identities even when the
// selected branch never references either copy.
self.find_entry(entry.id, max)?
.ok_or_else(|| missing_entry(entry.id))?;
let content = crate::content_codec::EntryContentView::decode(&entry.content)?;
if content.kind() != entry.kind {
return Err(session_fault(
smith::error::SessionCode::EntryKindMismatch,
format!("entry {} kind mismatch", entry.id),
));
}
if let Some(parent) = entry.parent {
let (_, parent_offset) = self
.find_entry(parent, max)?
.ok_or_else(|| missing_entry(parent))?;
if parent_offset >= offset {
return Err(session_fault(
smith::error::SessionCode::SessionParent,
"parent does not precede child",
));
}
}
last_entry = Some((entry.id, offset, entry.parent));
}
Frame::Index {
id,
offset: target,
parent,
parent_offset,
} if !first => {
self.verify_index(last_entry, id, target, parent, parent_offset, max)?;
last_entry = None;
}
Frame::Selection { id } if !first => {
self.find_entry(id, max)?.ok_or_else(|| missing_entry(id))?;
last_entry = None;
}
Frame::Checkpoint { id, offset: target } if !first => {
if self
.find_entry(id, max)?
.is_none_or(|(_, actual)| actual != target)
|| target >= offset
{
return Err(session_fault(
smith::error::SessionCode::SessionCheckpoint,
"checkpoint target is not an earlier entry boundary",
));
}
last_entry = None;
}
Frame::Unknown { .. } if !first => last_entry = None,
_ => {
return Err(session_fault(
smith::error::SessionCode::SessionHeaderMissing,
"missing first header",
));
}
}
first = false;
Ok(())
})?;
if first {
return Err(session_fault(
smith::error::SessionCode::SessionHeaderMissing,
"missing first header",
));
}
if let Some(fault) = scan.damage.into_iter().next() {
return Err(fault);
}
if scan.torn_tail_at.is_some() {
return Err(session_fault(
smith::error::SessionCode::SessionTorn,
"incomplete tail",
));
}
Ok(())
}
/// Open without taking the writer lock or modifying session content.
///
/// # Errors
/// Returns an input/output error for missing data or failed coordination.
pub fn open(path: &Path) -> Result<Self> {
Self::open_with_cancel(path, None)
}
/// Open a reader snapshot, observing cancellation while maintenance holds the reader lock.
///
/// # Errors
/// Returns [`SmithError::Cancelled`] or an input/output error for failed coordination.
pub fn open_cancelled(path: &Path, cancel: &CancelHandle) -> Result<Self> {
Self::open_with_cancel(path, Some(cancel))
}
fn open_with_cancel(path: &Path, cancel: Option<&CancelHandle>) -> Result<Self> {
let path = canonical_session_path(path)?;
let readers = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(readers_path(&path))
.map_err(io_error(
smith::error::StorageCode::StoreReaderLockOpen,
&readers_path(&path),
))?;
if let Some(cancel) = cancel {
loop {
if cancel.is_cancelled() {
return Err(SmithError::Cancelled);
}
match fs2::FileExt::try_lock_shared(&readers) {
Ok(()) => break,
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(std::time::Duration::from_millis(10));
}
Err(error) => {
return Err(io_error(
smith::error::StorageCode::StoreReaderLock,
&readers_path(&path),
)(error));
}
}
}
} else {
fs2::FileExt::lock_shared(&readers).map_err(io_error(
smith::error::StorageCode::StoreReaderLock,
&readers_path(&path),
))?;
}
let file =
File::open(&path).map_err(io_error(smith::error::StorageCode::StoreOpen, &path))?;
let snapshot_len = file
.metadata()
.map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
.len();
let writer_active = writer_active(&path)?;
Ok(Self {
file,
readers,
snapshot_len,
writer_active,
})
}
/// Visit a consistent file-size snapshot with per-frame bounded allocation.
///
/// # Errors
/// Returns an I/O error or the visitor's error.
pub fn scan(
&self,
frame_bytes_max: usize,
mut visit: impl FnMut(Frame) -> Result<()>,
) -> Result<Recovery> {
let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
read_prologue(&mut reader)?;
let mut failure = None;
let scan = visit_frames(&mut reader, frame_bytes_max, |frame| match visit(frame) {
Ok(()) => true,
Err(error) => {
failure = Some(error);
false
}
});
let scan = file_offsets(scan?);
if let Some(error) = failure {
return Err(error);
}
Ok(scan)
}
/// Decode the first frame only, without reading the history.
///
/// # Errors
/// Returns an I/O error when the file cannot be read.
pub fn first(&self, frame_bytes_max: usize) -> Result<(Option<Frame>, Recovery)> {
let first_limit = u64::try_from(frame_bytes_max)
.unwrap_or(u64::MAX)
.saturating_add(
u64::try_from(
FILE_PROLOGUE.len() + crate::frame::HEADER_LEN + crate::frame::BODY_PREFIX_LEN,
)
.unwrap_or(u64::MAX),
)
.min(self.snapshot_len);
let mut reader = BufReader::new(self.snapshot_read(0).take(first_limit));
read_prologue(&mut reader)?;
let mut first = None;
let scan = visit_frames(&mut reader, frame_bytes_max, |frame| {
first = Some(frame);
false
})?;
Ok((first, file_offsets(scan)))
}
/// Whether a writer owned this session when this reader acquired its snapshot.
#[must_use]
pub const fn writer_active(&self) -> bool {
self.writer_active
}
/// File size of the pinned generation at reader acquisition.
#[must_use]
pub const fn file_size(&self) -> u64 {
self.snapshot_len
}
/// Scan a snapshot while reporting damage through `report_damage`.
///
/// Unlike [`SessionReader::scan`], this method retains no damage, keeping
/// callers that report or reject faults incrementally bounded.
///
/// # Errors
/// Returns an I/O error, visitor error, or damage-reporting error.
pub fn scan_reporting(
&self,
frame_bytes_max: usize,
mut visit: impl FnMut(Frame) -> Result<bool>,
mut report_damage: impl FnMut(SmithError) -> Result<()>,
) -> Result<Recovery> {
let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
read_prologue(&mut reader)?;
let failure = std::cell::RefCell::new(None);
let scan = visit_frames_reporting(
&mut reader,
frame_bytes_max,
|frame| match visit(frame) {
Ok(keep_going) => keep_going,
Err(error) => {
*failure.borrow_mut() = Some(error);
false
}
},
|damage| match report_damage(shift_damage(damage)) {
Ok(()) => true,
Err(error) => {
*failure.borrow_mut() = Some(error);
false
}
},
)?;
if let Some(error) = failure.into_inner() {
return Err(error);
}
Ok(file_offsets(scan))
}
}
/// Repair a session file by republishing only its complete frames.
///
/// Damaged frames are dropped; [`Recovery::damage_count`] counts all faults
/// while [`Recovery::damage`] retains only the first report.
/// An explicit selection whose entry is dropped fails before publication.
/// The original content stays intact if any step fails before publication.
/// Repair takes the session lock, so it waits for a live writer to finish
/// instead of replacing the file underneath it.
///
/// # Errors
///
/// Returns a locking, framing, or input/output error.
pub fn repair_session(path: &Path, frame_bytes_max: usize) -> Result<Recovery> {
let lock = SessionLock::acquire(path)?;
let path = lock.path();
let _coordination = maintenance_lock(path)?;
let (temporary, mut out) = new_temp_file(path, "repair", SessionId::new())?;
publication_checkpoint!(TempCreated);
let written = (|| {
out.write_all(&FILE_PROLOGUE)
.map_err(io_error(smith::error::StorageCode::RepairWrite, &temporary))?;
publication_checkpoint!(PartialWrite);
let mut source = BufReader::with_capacity(
64 * 1024,
File::open(path).map_err(io_error(smith::error::StorageCode::StoreOpen, path))?,
);
read_prologue(&mut source)?;
let mut last = None;
let mut count = 0u16;
let mut selection = None;
let mut has_header = false;
let mut first_damage = None;
let mut failure = None;
let mut recovery = visit_frames_reporting(
&mut source,
frame_bytes_max,
|frame| {
let result = (|| {
match frame {
Frame::Index { .. } | Frame::Checkpoint { .. } => {}
Frame::Selection { id } => selection = Some(id),
Frame::Header { header } => {
has_header = true;
write_frame(&mut out, &Frame::Header { header }, frame_bytes_max)?;
}
Frame::Known { entry } => {
repair_entry(
&mut out,
path,
entry,
frame_bytes_max,
has_header,
&mut last,
&mut count,
)?;
}
other @ Frame::Unknown { .. } => {
write_frame(&mut out, &other, frame_bytes_max)?;
}
}
Ok(())
})();
if let Err(error) = result {
failure = Some(error);
false
} else {
true
}
},
|damage| {
if first_damage.is_none() {
first_damage = Some(shift_damage(damage));
}
true
},
)?;
if let Some(error) = failure {
return Err(error);
}
finish_repair_selection(&temporary, &mut out, selection.as_ref(), frame_bytes_max)?;
if let Some(damage) = first_damage {
recovery.damage.push(damage);
}
publication_checkpoint!(CompletedWrite);
out.sync_all()
.map_err(io_error(smith::error::StorageCode::RepairSync, &temporary))?;
publication_checkpoint!(TempSynced);
Ok(file_offsets(recovery))
})();
drop(out);
let recovery = match written {
Ok(recovery) => recovery,
Err(error) => {
// Preserve the recovery failure; temporary-file removal is best-effort cleanup.
let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
return Err(error);
}
};
if let Err(error) = std::fs::rename(&temporary, path) {
// Preserve the publish failure; temporary-file removal is best-effort cleanup.
let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
return Err(io_error(smith::error::StorageCode::RepairPublish, path)(
error,
));
}
publication_checkpoint!(Published);
sync_parent(path)?;
publication_checkpoint!(ParentSynced);
// `rename` removed the temporary name as it published the replacement.
publication_checkpoint!(Cleanup);
drop(lock);
Ok(recovery)
}
fn repair_entry(
out: &mut File,
path: &Path,
entry: EntryFrame,
max: usize,
has_header: bool,
last: &mut Option<(EntryId, u64)>,
count: &mut u16,
) -> Result<()> {
let offset = out
.stream_position()
.map_err(io_error(smith::error::StorageCode::RepairSeek, path))?;
let id = entry.id;
let parent = entry.parent;
write_frame(out, &Frame::Known { entry }, max)?;
let parent_offset =
parent.and_then(|parent| last.filter(|(id, _)| *id == parent).map(|(_, at)| at));
*last = Some((id, offset));
if has_header && (parent.is_none() || parent_offset.is_some()) && max >= 48 {
out.write_all(&index_bytes(id, offset, parent, parent_offset))
.map_err(io_error(smith::error::StorageCode::RepairWrite, path))?;
}
*count += 1;
if has_header && *count == 64 && max >= 24 {
write_frame(out, &Frame::Checkpoint { id, offset }, max)?;
*count = 0;
}
Ok(())
}
fn maintenance_lock(path: &Path) -> Result<File> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(readers_path(path))
.map_err(io_error(
smith::error::StorageCode::StoreReaderLockOpen,
&readers_path(path),
))?;
fs2::FileExt::try_lock_exclusive(&file).map_err(io_error(
smith::error::StorageCode::StoreReadersActive,
&readers_path(path),
))?;
Ok(file)
}
fn finish_repair_selection(
path: &Path,
out: &mut File,
selection: Option<&EntryId>,
max: usize,
) -> Result<()> {
let Some(&id) = selection else {
return Ok(());
};
let mut source = BufReader::new(
File::open(path).map_err(io_error(smith::error::StorageCode::RepairOpen, path))?,
);
read_prologue(&mut source)?;
let mut found = None;
visit_metadata_reporting(
&mut source,
max,
|at, frame| {
if let Frame::Known { entry } = frame
&& entry.id == id
{
found = Some(at + FILE_PROLOGUE.len() as u64);
}
Ok(())
},
|_| Ok(()),
)?;
let offset = found.ok_or_else(|| missing_entry(id))?;
write_frame(out, &Frame::Selection { id }, max)?;
if max >= 24 {
write_frame(out, &Frame::Checkpoint { id, offset }, max)?;
}
Ok(())
}
fn recovery_checkpoint_count(file: &File, path: &Path, max: usize) -> Result<u16> {
let mut reader = BufReader::new(
file.try_clone()
.map_err(io_error(smith::error::StorageCode::StoreOpen, path))?,
);
reader
.seek(SeekFrom::Start(0))
.map_err(io_error(smith::error::StorageCode::StoreSeek, path))?;
read_prologue(&mut reader)?;
let (path, file) = scratch_file("cadence")?;
let mut proof = Proof {
file,
path,
len: 0,
selected_checkpoint: None,
cancel: None,
};
let mut count = 0;
visit_metadata_reporting(
&mut reader,
max,
|offset, frame| {
match frame {
Frame::Known { entry } => {
proof.push(&ProofEntry {
id: entry.id,
parent: entry.parent,
offset: offset + FILE_PROLOGUE.len() as u64,
parent_offset: None,
compaction: false,
})?;
count = (count + 1) % 64;
}
Frame::Checkpoint { id, offset: at }
if proof.at_offset(at)?.is_some_and(|entry| entry.id == id) =>
{
count = 0;
}
_ => {}
}
Ok(())
},
|_| Ok(()),
)?;
Ok(count)
}
/// Walk an already-open file from its start using the shared frame reader.
///
/// # Errors
///
/// Returns an input/output error when the file cannot be re-read.
fn walk_file_bounded(file: &File, path: &Path, frame_bytes_max: usize) -> Result<Recovery> {
let mut handle = file
.try_clone()
.map_err(io_error(smith::error::StorageCode::StoreOpen, path))?;
handle
.seek(SeekFrom::Start(0))
.map_err(io_error(smith::error::StorageCode::StoreSeek, path))?;
let mut reader = BufReader::with_capacity(64 * 1024, handle);
read_prologue(&mut reader)?;
let mut damage = None;
let mut recovery = visit_frames_reporting(
&mut reader,
frame_bytes_max,
|_| true,
|fault| {
if damage.is_none() {
damage = Some(fault);
}
true
},
)
.map(file_offsets)?;
if let Some(fault) = damage {
recovery.damage.push(shift_damage(fault));
}
Ok(recovery)
}
fn read_prologue(reader: &mut impl Read) -> Result<()> {
let mut magic = [0; FILE_PROLOGUE.len()];
reader.read_exact(&mut magic).map_err(io_error(
smith::error::StorageCode::SessionPrologue,
Path::new("<session>"),
))?;
if magic != FILE_PROLOGUE {
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionPrologue,
message: "invalid session format or version; expected SMH3 prologue".into(),
});
}
Ok(())
}
fn file_offsets(mut recovery: Recovery) -> Recovery {
recovery.good_end += FILE_PROLOGUE.len() as u64;
recovery.torn_tail_at = recovery
.torn_tail_at
.map(|n| n + FILE_PROLOGUE.len() as u64);
for damage in &mut recovery.damage {
if let SmithError::Frame {
offset: Some(n), ..
} = damage
{
*n += FILE_PROLOGUE.len() as u64;
}
}
recovery
}
const fn shift_damage(mut damage: SmithError) -> SmithError {
if let SmithError::Frame {
offset: Some(n), ..
} = &mut damage
{
*n += FILE_PROLOGUE.len() as u64;
}
damage
}
/// Rename a synced temporary file into an absent target without any fallback.
///
/// `renamore::rename_exclusive` is `renameat2(RENAME_NOREPLACE)` on Linux,
/// `renamex_np(RENAME_EXCL)` on macOS, and `MoveFileExW` without replace flags
/// on Windows. Unsupported filesystems fail rather than weakening no-clobber.
fn publish_no_clobber(temporary: &Path, target: &Path) -> Result<()> {
renamore::rename_exclusive(temporary, target)
.map_err(io_error(smith::error::StorageCode::CreatePublish, target))?;
publication_checkpoint!(Published);
sync_parent(target)?;
publication_checkpoint!(ParentSynced);
Ok(())
}
fn sync_parent(path: &Path) -> Result<()> {
let parent = path
.parent()
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
#[cfg(test)]
if PARENT_SYNC_FAILURE.with(std::cell::Cell::get) {
return Err(published_parent_sync_error(
parent,
"injected parent directory sync failure",
));
}
sync_directory(parent).map_err(|error| published_parent_sync_error(parent, error))
}
fn published_parent_sync_error(path: &Path, error: impl std::fmt::Display) -> SmithError {
SmithError::Storage {
code: smith::error::StorageCode::StoreDirSync,
operation: smith::error::StorageOperation::SyncParentDirectoryAfterPublication,
path: path.to_path_buf(),
message: error.to_string(),
publication: Some(smith::error::PublicationState::PublishedDurabilityUncertain),
}
}
#[cfg(not(windows))]
fn sync_directory(path: &Path) -> std::io::Result<()> {
File::open(path)?.sync_all()
}
#[cfg(windows)]
fn sync_directory(path: &Path) -> std::io::Result<()> {
use std::os::windows::fs::OpenOptionsExt;
// FILE_FLAG_BACKUP_SEMANTICS permits opening a directory handle safely.
OpenOptions::new()
.read(true)
.custom_flags(0x0200_0000)
.open(path)?
.sync_all()
}
fn session_fault(code: smith::error::SessionCode, message: impl Into<String>) -> SmithError {
SmithError::Session {
code,
message: message.into(),
}
}
fn missing_entry(id: EntryId) -> SmithError {
session_fault(
smith::error::SessionCode::EntryNotFound,
format!("entry {id} not found"),
)
}
fn io_error(code: smith::error::StorageCode, path: &Path) -> impl Fn(std::io::Error) -> SmithError {
storage_error(code, storage_operation(code), path)
}
const fn storage_operation(code: smith::error::StorageCode) -> smith::error::StorageOperation {
use smith::error::{StorageCode, StorageOperation};
match code {
StorageCode::StoreCanonicalize => StorageOperation::Canonicalize,
StorageCode::StoreStat | StorageCode::CreateStat => StorageOperation::Stat,
StorageCode::StoreOpen
| StorageCode::StoreLockOpen
| StorageCode::StoreReaderLockOpen
| StorageCode::RepairOpen => StorageOperation::Open,
StorageCode::StoreRead
| StorageCode::StoreProofRead
| StorageCode::StoreScratchRead
| StorageCode::FrameRead
| StorageCode::SessionPrologue
| StorageCode::FrameBoundary
| StorageCode::AuthRead => StorageOperation::Read,
StorageCode::StoreWrite
| StorageCode::StoreProofWrite
| StorageCode::StoreScratchWrite
| StorageCode::CreateWrite
| StorageCode::RepairWrite
| StorageCode::StorePoisoned => StorageOperation::Write,
StorageCode::StoreSeek
| StorageCode::StoreProofSeek
| StorageCode::StoreScratchSeek
| StorageCode::RepairSeek => StorageOperation::Seek,
StorageCode::StoreSync
| StorageCode::CreateSync
| StorageCode::RepairSync
| StorageCode::StoreDirSync => StorageOperation::Sync,
StorageCode::StoreTruncate => StorageOperation::Truncate,
StorageCode::StoreMkdir | StorageCode::RuntimeDir => StorageOperation::CreateDirectory,
StorageCode::AuthStore => StorageOperation::AuthStoreOperation,
StorageCode::StoreLock
| StorageCode::StoreBusy
| StorageCode::StoreReaderLock
| StorageCode::StoreReadersActive => StorageOperation::Lock,
StorageCode::StoreUnlock | StorageCode::StoreReaderUnlock => StorageOperation::Unlock,
StorageCode::StoreProofRemove
| StorageCode::StoreScratchRemove
| StorageCode::CreateTempRemove => StorageOperation::Remove,
StorageCode::CreateTmp => StorageOperation::CreateTemporaryFile,
StorageCode::CreatePublish | StorageCode::RepairPublish => StorageOperation::Publish,
}
}
fn storage_fault(
code: smith::error::StorageCode,
path: &Path,
message: impl Into<String>,
) -> SmithError {
SmithError::Storage {
code,
operation: storage_operation(code),
path: path.to_path_buf(),
message: message.into(),
publication: None,
}
}
fn storage_error(
code: smith::error::StorageCode,
operation: smith::error::StorageOperation,
path: &Path,
) -> impl Fn(std::io::Error) -> SmithError + '_ {
move |e| SmithError::Storage {
code,
operation,
path: path.to_path_buf(),
message: e.to_string(),
publication: None,
}
}
#[cfg(test)]
#[expect(
clippy::unwrap_used,
clippy::panic,
reason = "tests may panic on invariant violations"
)]
mod tests {
use super::*;
use crate::session::COMPACTION_KIND;
use crate::session::{EntryContent, EntryFrame, EntryKind, Session};
use smith::message::{ContentBlock, Message, Role};
use smith::structured::{StructuredData, ToolArguments};
use smith_alloc::{Scope, measure};
use std::sync::Mutex;
static MEASURING: Mutex<()> = Mutex::new(());
fn tmp_path(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("smith_store_tests_{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
dir.join(name)
}
fn message(text: &str) -> EntryContent {
EntryContent::Message(Message::with_text(Role::User, text))
}
fn entry(text: &str) -> EntryFrame {
EntryFrame::new(None, &message(text)).unwrap()
}
fn text(entry: &EntryFrame) -> Option<String> {
match EntryContent::decode(&entry.content).unwrap() {
EntryContent::Message(m) => match &m.blocks[0] {
ContentBlock::Text(t) => Some(t.clone()),
_ => None,
},
_ => None,
}
}
fn frame(text: &str) -> Frame {
Frame::Known { entry: entry(text) }
}
fn file_bytes(frames: impl IntoIterator<Item = u8>) -> Vec<u8> {
FILE_PROLOGUE.into_iter().chain(frames).collect()
}
/// Eager frame collection belongs only in recovery assertions; durable
/// product paths use `SessionReader`'s bounded readers directly.
fn scanned_frames(path: &Path, frame_bytes_max: usize) -> Recovery {
let reader = SessionReader::open(path).unwrap();
let mut frames = Vec::new();
let mut recovery = reader
.scan(frame_bytes_max, |frame| {
frames.push(frame);
Ok(())
})
.unwrap();
recovery.frames = frames;
recovery
}
#[test]
fn durable_selection_replays_and_forks_a_sibling_without_loading_unrelated_content() {
let path = tmp_path("durable-selection.smh");
let fork_path = tmp_path("durable-selection-fork.smh");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&fork_path);
let cfg = Config::default_valid();
let root = entry("root");
let left = EntryFrame::new(Some(root.id), &message("left")).unwrap();
let right = EntryFrame::new(Some(root.id), &message("right")).unwrap();
{
let mut opened = open_session(&path, cfg.clone()).unwrap();
for entry in [&root, &left, &right] {
opened
.writer
.append_frame(&Frame::Known {
entry: entry.clone(),
})
.unwrap();
}
opened.writer.select_entry(left.id).unwrap();
}
let reopened = open_session(&path, cfg.clone()).unwrap();
assert_eq!(reopened.session.selected_id(), Some(left.id));
assert_eq!(reopened.session.active_branch().entries().len(), 2);
assert_eq!(
reopened
.session
.content(&reopened.session.active_branch().entries()[1])
.unwrap(),
EntryContent::decode(&left.content).unwrap()
);
drop(reopened);
let reader = SessionReader::open(&path).unwrap();
let fork = reader.fork_at(right.id, &fork_path, cfg).unwrap();
assert_eq!(fork.session.selected_id(), Some(right.id));
assert_ne!(
fork.session.id(),
SessionReader::open(&path)
.unwrap()
.selected_session(Config::default_valid().frame_bytes_max)
.unwrap()
.id()
);
assert_eq!(
fork.session
.content(&fork.session.active_branch().entries()[1])
.unwrap(),
EntryContent::decode(&right.content).unwrap()
);
drop(fork);
drop(reader);
let _ = std::fs::remove_file(path);
let _ = std::fs::remove_file(fork_path);
}
#[test]
fn full_verification_detects_duplicate_ids_outside_selected_branch() {
let path = tmp_path("duplicate-unselected.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let unselected = entry("one");
let mut duplicate = entry("two");
duplicate.id = unselected.id;
for entry in [unselected, duplicate] {
opened.writer.append_frame(&Frame::Known { entry }).unwrap();
}
drop(opened);
let reader = SessionReader::open(&path).unwrap();
assert_eq!(
reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
"SESSION_DUPLICATE_ID"
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn missing_session_file_reports_storage_operation_path_and_source() {
let path = tmp_path("missing-storage-error.smh");
let _ = std::fs::remove_file(&path);
let error = SessionReader::open(&path).unwrap_err();
match error {
SmithError::Storage {
code,
operation,
path: actual,
message,
..
} => {
assert_eq!(code, smith::error::StorageCode::StoreOpen);
assert_eq!(operation.as_str(), "open");
assert_eq!(actual, path);
assert!(!message.is_empty());
}
other => panic!("expected storage error, got {other:?}"),
}
let _ = std::fs::remove_file(readers_path(&path));
}
#[test]
fn missing_selected_entry_fails_open_without_retargeting_a_branch() {
let path = tmp_path("missing-selected.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: entry("valid"),
})
.unwrap();
opened
.writer
.append_frame(&Frame::Selection { id: EntryId::new() })
.unwrap();
drop(opened);
assert_eq!(
open_session(&path, cfg).unwrap_err().code(),
"ENTRY_NOT_FOUND"
);
let _ = std::fs::remove_file(path);
}
#[test]
fn damaged_index_does_not_hide_following_entries() {
let path = tmp_path("damaged-index.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let session = Session::new();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
let mut bytes = FILE_PROLOGUE.to_vec();
bytes.extend(
Frame::Header {
header: session.header(),
}
.encode(cfg.frame_bytes_max)
.unwrap(),
);
bytes.extend(
Frame::Known {
entry: root.clone(),
}
.encode(cfg.frame_bytes_max)
.unwrap(),
);
let mut invalid = Frame::Index {
id: root.id,
offset: 8,
parent: None,
parent_offset: None,
}
.encode(cfg.frame_bytes_max)
.unwrap();
invalid[crate::frame::HEADER_LEN + 4] ^= 0x80;
bytes.extend(invalid);
bytes.extend(
Frame::Known {
entry: child.clone(),
}
.encode(cfg.frame_bytes_max)
.unwrap(),
);
std::fs::write(&path, bytes).unwrap();
let reader = SessionReader::open(&path).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
Some(child.id)
);
assert_eq!(
reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
"FRAME_CHECKSUM"
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn repair_discards_stale_offsets_without_losing_authoritative_entries() {
let path = tmp_path("repair-index-offset.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let record = *opened.session.append(&message("kept")).unwrap();
opened
.writer
.append_entry(&record, opened.session.bytes(&record))
.unwrap();
let selected = record.id;
drop(opened);
let original = std::fs::read(&path).unwrap();
let header_end = FILE_PROLOGUE.len()
+ Frame::Header {
header: SessionReader::open(&path)
.unwrap()
.selected_session(cfg.frame_bytes_max)
.unwrap()
.header(),
}
.encode(cfg.frame_bytes_max)
.unwrap()
.len();
let mut bytes = original[..header_end].to_vec();
let mut damaged = Frame::Unknown {
kind: 99,
version: 1,
payload: b"bad".to_vec(),
}
.encode(cfg.frame_bytes_max)
.unwrap();
damaged[crate::frame::HEADER_LEN + 4] ^= 1;
bytes.extend(damaged);
bytes.extend(&original[header_end..]);
std::fs::write(&path, bytes).unwrap();
repair_session(&path, cfg.frame_bytes_max).unwrap();
let reader = SessionReader::open(&path).unwrap();
reader.verify_full(cfg.frame_bytes_max).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
Some(selected)
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn reopen_keeps_only_effective_payload_after_compaction() {
let path = tmp_path("effective-compaction.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let before = *opened
.session
.append(&message(&"x".repeat(400_000)))
.unwrap();
opened
.writer
.append_entry(&before, opened.session.bytes(&before))
.unwrap();
let compact = EntryContent::Meta {
kind: COMPACTION_KIND.into(),
detail: StructuredData::from_serializable(&serde_json::json!({"summary": ""})).unwrap(),
};
let boundary = *opened.session.append(&compact).unwrap();
opened
.writer
.append_entry(&boundary, opened.session.bytes(&boundary))
.unwrap();
let after = *opened.session.append(&message("after")).unwrap();
opened
.writer
.append_entry(&after, opened.session.bytes(&after))
.unwrap();
drop(opened);
let reopened = open_session(&path, cfg).unwrap();
assert_eq!(reopened.session.selected_id(), Some(after.id));
assert_eq!(reopened.session.compaction_boundary(), Some(boundary.id));
assert_eq!(reopened.session.active_branch().entries().len(), 2);
assert!(reopened.session.bytes(&before).is_empty());
drop(reopened);
let _ = std::fs::remove_file(path);
}
#[test]
fn replay_visits_file_metadata_once_for_linear_history() {
let path = tmp_path("replay-pass-count.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let mut parent = None;
for _ in 0..96 {
let e = EntryFrame::new(parent, &message("x")).unwrap();
parent = Some(e.id);
opened
.writer
.append_frame(&Frame::Known { entry: e })
.unwrap();
}
drop(opened);
let reader = SessionReader::open(&path).unwrap();
METADATA_PASSES.with(|passes| passes.set(0));
let mut visited = 0;
reader
.replay_at(parent.unwrap(), cfg.frame_bytes_max, |_| {
visited += 1;
Ok(())
})
.unwrap();
assert_eq!(visited, 96);
METADATA_PASSES
.with(|passes| assert!(passes.get() <= 2, "{} metadata passes", passes.get()));
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn effective_context_cannot_publish_a_broken_rooted_fork() {
let path = tmp_path("partial-source.smh");
let target = tmp_path("partial-target.smh");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&target);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let first = *opened.session.append(&message("before")).unwrap();
opened
.writer
.append_entry(&first, opened.session.bytes(&first))
.unwrap();
let compact = EntryContent::Meta {
kind: COMPACTION_KIND.into(),
detail: StructuredData::from_serializable(&serde_json::json!({"summary": ""})).unwrap(),
};
let boundary = *opened.session.append(&compact).unwrap();
opened
.writer
.append_entry(&boundary, opened.session.bytes(&boundary))
.unwrap();
drop(opened);
let resumed = open_session(&path, cfg.clone()).unwrap();
assert_eq!(
create_session(&target, &resumed.session, cfg)
.unwrap_err()
.code(),
"SESSION_INCOMPLETE_CONTEXT"
);
assert!(!target.exists());
drop(resumed);
let _ = std::fs::remove_file(path);
}
#[test]
fn snapshot_reads_pinned_inode_after_uncoordinated_replacement() {
let path = tmp_path("pinned-generation.smh");
let replacement = tmp_path("pinned-generation-replacement.smh");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&replacement);
let cfg = Config::default_valid();
let mut first = open_session(&path, cfg.clone()).unwrap();
let root = entry("old");
first
.writer
.append_frame(&Frame::Known {
entry: root.clone(),
})
.unwrap();
let old_id = first.session.id();
drop(first);
let snapshot = SessionReader::open(&path).unwrap();
let newer = open_session(&replacement, cfg.clone()).unwrap();
let new_id = newer.session.id();
drop(newer);
std::fs::rename(&replacement, &path).unwrap();
assert_eq!(
snapshot.selected_session(cfg.frame_bytes_max).unwrap().id(),
old_id
);
assert_eq!(
snapshot
.find_entry(root.id, cfg.frame_bytes_max)
.unwrap()
.unwrap()
.0
.id,
root.id
);
assert_eq!(
SessionReader::open(&path)
.unwrap()
.selected_session(cfg.frame_bytes_max)
.unwrap()
.id(),
new_id
);
drop(snapshot);
let _ = std::fs::remove_file(path);
}
#[test]
fn replay_and_fork_visit_growth_is_quadratic_not_cubic() {
fn visits(depth: usize) -> (usize, usize) {
let path = tmp_path(&format!("growth-{depth}.smh"));
let target = tmp_path(&format!("growth-fork-{depth}.smh"));
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&target);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let mut parent = None;
for _ in 0..depth {
let entry = EntryFrame::new(parent, &message("x")).unwrap();
parent = Some(entry.id);
opened.writer.append_frame(&Frame::Known { entry }).unwrap();
}
drop(opened);
let reader = SessionReader::open(&path).unwrap();
PROOF_VISITS.with(|visits| visits.set(0));
reader
.replay_at(parent.unwrap(), cfg.frame_bytes_max, |_| Ok(()))
.unwrap();
let replay = PROOF_VISITS.with(std::cell::Cell::get);
PROOF_VISITS.with(|visits| visits.set(0));
let fork = reader.fork_at(parent.unwrap(), &target, cfg).unwrap();
assert_eq!(fork.session.selected_id(), parent);
let fork_visits = PROOF_VISITS.with(std::cell::Cell::get);
drop(fork);
drop(reader);
let _ = std::fs::remove_file(path);
let _ = std::fs::remove_file(target);
(replay, fork_visits)
}
let small = visits(64);
let large = visits(128);
assert!(
large.0 <= 5 * small.0,
"replay proof visits: {small:?} -> {large:?}"
);
assert!(
large.1 <= 5 * small.1,
"fork proof visits: {small:?} -> {large:?}"
);
}
#[test]
fn create_keeps_selection_of_existing_ancestor() {
let path = tmp_path("create-ancestor-selection.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
let source = Session::from_frames(&[
Frame::Header {
header: Session::new().header(),
},
Frame::Known {
entry: root.clone(),
},
Frame::Known { entry: child },
Frame::Selection { id: root.id },
])
.unwrap();
let opened = create_session(&path, &source, cfg.clone()).unwrap();
assert_eq!(opened.session.selected_id(), Some(root.id));
drop(opened);
let reopened = open_session(&path, cfg).unwrap();
assert_eq!(reopened.session.selected_id(), Some(root.id));
drop(reopened);
let _ = std::fs::remove_file(path);
}
#[test]
fn valid_checkpoint_locates_selected_ancestor() {
let path = tmp_path("selected-checkpoint-hint.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
let mut opened = open_session(&path, cfg.clone()).unwrap();
for record in [&root, &child] {
opened
.writer
.append_frame(&Frame::Known {
entry: record.clone(),
})
.unwrap();
}
opened.writer.select_entry(root.id).unwrap();
drop(opened);
CHECKPOINT_HITS.with(|hits| hits.set(0));
let reader = SessionReader::open(&path).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
Some(root.id)
);
CHECKPOINT_HITS.with(|hits| assert_eq!(hits.get(), 1));
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn invalid_checkpoint_does_not_reset_writer_cadence() {
let path = tmp_path("invalid-cadence.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
for _ in 0..63 {
let record = *opened.session.append(&message("turn")).unwrap();
opened
.writer
.append_entry(&record, opened.session.bytes(&record))
.unwrap();
}
opened
.writer
.append_frame(&Frame::Checkpoint {
id: EntryId::new(),
offset: 123,
})
.unwrap();
drop(opened);
let mut reopened = open_session(&path, cfg.clone()).unwrap();
let record = *reopened.session.append(&message("last")).unwrap();
reopened
.writer
.append_entry(&record, reopened.session.bytes(&record))
.unwrap();
drop(reopened);
let mut count = 0;
SessionReader::open(&path)
.unwrap()
.scan(cfg.frame_bytes_max, |frame| {
if matches!(frame, Frame::Checkpoint { .. }) {
count += 1;
}
Ok(())
})
.unwrap();
assert_eq!(count, 2);
let _ = std::fs::remove_file(path);
}
#[test]
fn writer_publishes_reconstructible_checkpoints_at_the_cadence() {
let path = tmp_path("checkpoint-cadence.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
for _ in 0..63 {
let record = *opened.session.append(&message("turn")).unwrap();
opened
.writer
.append_entry(&record, opened.session.bytes(&record))
.unwrap();
}
drop(opened);
let mut reopened = open_session(&path, cfg.clone()).unwrap();
let record = *reopened.session.append(&message("turn")).unwrap();
reopened
.writer
.append_entry(&record, reopened.session.bytes(&record))
.unwrap();
let selected = reopened.session.selected_id();
drop(reopened);
let reader = SessionReader::open(&path).unwrap();
let mut checkpoints = 0;
reader
.scan(cfg.frame_bytes_max, |frame| {
if matches!(frame, Frame::Checkpoint { .. }) {
checkpoints += 1;
}
Ok(())
})
.unwrap();
assert_eq!(checkpoints, 1);
reader.verify_full(cfg.frame_bytes_max).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
selected
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn forged_checkpoint_is_ignored_for_resume_but_fails_full_verification() {
let path = tmp_path("forged-checkpoint.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: root.clone(),
})
.unwrap();
opened
.writer
.append_frame(&Frame::Checkpoint {
id: root.id,
offset: FILE_PROLOGUE.len() as u64 + 1,
})
.unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: child.clone(),
})
.unwrap();
drop(opened);
let reader = SessionReader::open(&path).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
Some(child.id)
);
assert_eq!(
reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
"SESSION_CHECKPOINT"
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn forged_index_never_hides_later_entries_and_full_verification_rejects_it() {
let path = tmp_path("forged-index.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut opened = open_session(&path, cfg.clone()).unwrap();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: root.clone(),
})
.unwrap();
opened
.writer
.append_frame(&Frame::Index {
id: root.id,
offset: FILE_PROLOGUE.len() as u64 + 1,
parent: None,
parent_offset: None,
})
.unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: child.clone(),
})
.unwrap();
drop(opened);
let reader = SessionReader::open(&path).unwrap();
assert_eq!(
reader
.selected_session(cfg.frame_bytes_max)
.unwrap()
.selected_id(),
Some(child.id)
);
assert_eq!(
reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
"SESSION_INDEX"
);
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn create_validates_config_before_creating_session_files() {
let path = tmp_path("config-before-effects/nested/session.smh");
let _ = std::fs::remove_dir_all(path.parent().unwrap().parent().unwrap());
let mut config = Config::default_valid();
config.frame_bytes_max = 0;
assert!(create_session(&path, &Session::new(), config).is_err());
assert!(!path.parent().unwrap().exists());
}
#[test]
fn create_rejects_named_identity_mismatch_before_publication() {
let session = Session::new();
let path = tmp_path(&format!("{}.smh", smith::id::SessionId::new()));
let _ = std::fs::remove_file(&path);
assert_eq!(
create_session(&path, &session, Config::default_valid())
.unwrap_err()
.code(),
"SESSION_ID_MISMATCH"
);
assert!(!path.exists());
let _ = std::fs::remove_file(lock_path(&path));
}
#[cfg(unix)]
#[test]
fn create_temp_symlink_cannot_truncate_its_target() {
let path = tmp_path("create-temp-link.smh");
let temporary = path.with_extension("smith-create.tmp");
let victim = tmp_path("create-temp-victim");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&temporary);
std::fs::write(&victim, b"do not truncate").unwrap();
std::os::unix::fs::symlink(&victim, &temporary).unwrap();
create_session(&path, &Session::new(), Config::default_valid()).unwrap();
assert_eq!(std::fs::read(&victim).unwrap(), b"do not truncate");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&temporary);
let _ = std::fs::remove_file(&victim);
}
#[cfg(unix)]
#[test]
fn repair_temp_symlink_cannot_truncate_its_target() {
let path = tmp_path("repair-temp-link.smh");
let temporary = path.with_extension("smith-repair.tmp");
let victim = tmp_path("repair-temp-victim");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&temporary);
std::fs::write(&path, file_bytes(frame("intact").encode(1 << 20).unwrap())).unwrap();
std::fs::write(&victim, b"do not truncate").unwrap();
std::os::unix::fs::symlink(&victim, &temporary).unwrap();
repair_session(&path, Config::default_valid().frame_bytes_max).unwrap();
assert_eq!(std::fs::read(&victim).unwrap(), b"do not truncate");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&temporary);
let _ = std::fs::remove_file(&victim);
}
fn text_of(frame: &Frame) -> String {
match frame {
Frame::Known { entry } => text(entry).unwrap(),
other => panic!("unexpected frame {other:?}"),
}
}
#[cfg(target_arch = "x86_64")]
#[test]
fn observer_handle_has_constant_size_independent_of_history() {
assert_eq!(std::mem::size_of::<SessionReader>(), 24);
}
#[test]
fn reporting_scan_does_not_retain_streamed_damage() {
let path = tmp_path("reporting-damage.smh");
let _ = std::fs::remove_file(&path);
let mut malformed = frame("bad").encode(1 << 20).unwrap();
malformed[crate::frame::HEADER_LEN + 4] ^= 0xff;
std::fs::write(&path, file_bytes(malformed.repeat(128))).unwrap();
let reader = SessionReader::open(&path).unwrap();
let mut reported = 0;
let recovery = reader
.scan_reporting(
Config::default_valid().frame_bytes_max,
|_| Ok(true),
|_| {
reported += 1;
Ok(())
},
)
.unwrap();
assert_eq!(reported, 128);
assert!(recovery.damage.is_empty());
let _ = std::fs::remove_file(path);
}
#[test]
fn reader_snapshot_keeps_initial_writer_ownership_and_length() {
let path = tmp_path("reader-snapshot.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
writer.append_frame(&frame("before")).unwrap();
let reader = SessionReader::open(&path).unwrap();
assert!(reader.writer_active());
let snapshot_len = reader.file_size();
writer.append_frame(&frame("after")).unwrap();
drop(writer);
assert_eq!(reader.file_size(), snapshot_len);
let mut frames = 0;
reader
.scan(cfg.frame_bytes_max, |_| {
frames += 1;
Ok(())
})
.unwrap();
assert_eq!(frames, 1);
let _ = std::fs::remove_file(path);
}
#[test]
fn streaming_scan_visits_entries_without_retaining_history() {
// SMH-SPEC-SPEC0001: Sessions, Working sets; a snapshot never follows live appends.
let path = tmp_path("stream_scan.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
for _ in 0..128 {
writer.append_frame(&frame("entry")).unwrap();
}
let mut count = 0;
let scan = scan_session(&path, cfg.frame_bytes_max, |frame| {
assert!(matches!(frame, Frame::Known { .. }));
count += 1;
Ok(())
})
.unwrap();
assert_eq!(count, 128);
assert!(scan.damage.is_empty());
assert_eq!(scan.torn_tail_at, None);
let mut corrupt = File::options().append(true).open(&path).unwrap();
corrupt.write_all(&[0, 1]).unwrap();
let scan = scan_session(&path, cfg.frame_bytes_max, |_| Ok(())).unwrap();
assert!(scan.torn_tail_at.is_some());
drop(writer);
let _ = std::fs::remove_file(path);
}
#[test]
fn append_load_round_trip_preserves_entries() {
let path = tmp_path("round_trip.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let first = entry("one");
let second = EntryFrame::new(Some(first.id), &message("two")).unwrap();
{
let mut opened = open_session(&path, cfg.clone()).unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: first.clone(),
})
.unwrap();
opened
.writer
.append_frame(&Frame::Known {
entry: second.clone(),
})
.unwrap();
}
let session = SessionReader::open(&path)
.unwrap()
.selected_session(cfg.frame_bytes_max)
.unwrap();
assert_eq!(
session
.active_branch()
.entries()
.iter()
.map(|e| e.id)
.collect::<Vec<_>>(),
vec![first.id, second.id]
);
assert_eq!(session.selected_id(), Some(second.id));
let _ = std::fs::remove_file(&path);
}
#[test]
fn successful_io_error_context_setup_does_not_allocate() {
let path = Path::new("/tmp/smith-store-allocation-check");
let _serial = MEASURING
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let (_, cost) = measure(Scope::IoRequest, || {
let error = io_error(smith::error::StorageCode::StoreWrite, path);
drop(std::hint::black_box(error));
});
assert_eq!(cost.allocations, 0);
assert_eq!(cost.bytes, 0);
}
#[test]
fn successful_write_does_not_allocate_error_context() {
let path = tmp_path("write-allocation-check.smh");
let _ = std::fs::remove_file(&path);
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
let bytes = frame("allocation-check")
.encode(Config::default_valid().frame_bytes_max)
.unwrap();
let _serial = MEASURING
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let (result, cost) = measure(Scope::IoRequest, || writer.write(&bytes));
result.unwrap();
assert_eq!(cost.allocations, 0);
assert_eq!(cost.bytes, 0);
drop(writer);
let _ = std::fs::remove_file(path);
}
#[test]
fn torn_tail_is_truncated_on_open_and_appends_stay_visible() {
let path = tmp_path("torn_tail.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
{
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&frame("kept")).unwrap();
}
// Simulate a crash mid-append: partial header + payload bytes.
let good = Frame::Known {
entry: entry("kept"),
}
.encode(cfg.frame_bytes_max)
.unwrap();
let mut torn = good;
torn.extend_from_slice(&[0x10, 0x00, 0x00, 0x00, 0xAA, 0xBB]);
std::fs::write(&path, file_bytes(torn)).unwrap();
{
// Open repairs the tail, then a new append must be loadable.
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&frame("after")).unwrap();
}
let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
let texts: Vec<String> = frames
.iter()
.filter_map(|f| match f {
Frame::Known { entry } => text(entry),
_ => None,
})
.collect();
assert_eq!(texts, vec!["kept".to_string(), "after".to_string()]);
let _ = std::fs::remove_file(&path);
}
#[test]
fn loader_stops_at_torn_tail_without_header_fabrication() {
let path = tmp_path("loader_torn.smh");
let cfg = Config::default_valid();
let good = frame("a").encode(cfg.frame_bytes_max).unwrap();
let mut bytes = good;
bytes.extend_from_slice(&[0xFF, 0xFF]); // partial header
std::fs::write(&path, file_bytes(bytes)).unwrap();
let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
assert_eq!(frames.len(), 1);
let _ = std::fs::remove_file(&path);
}
#[test]
fn reopening_a_multi_frame_session_keeps_every_frame() {
let path = tmp_path("reopen_multi.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
{
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&frame("one")).unwrap();
w.append_frame(&frame("two")).unwrap();
}
{
// Reopening must not mistake a later frame boundary for a torn tail.
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&frame("three")).unwrap();
}
let recovery = scanned_frames(&path, cfg.frame_bytes_max);
assert_eq!(recovery.frames.len(), 3);
assert!(recovery.damage.is_empty());
assert_eq!(recovery.torn_tail_at, None);
let _ = std::fs::remove_file(&path);
}
#[test]
fn oversized_complete_frame_is_skipped_and_following_frames_reachable() {
let path = tmp_path("oversized_skip.smh");
// declared = 5000 > limit 1024, payload fully present, then a valid frame.
let mut bytes = Frame::Unknown {
kind: 9,
version: 1,
payload: vec![0xAB; 5000],
}
.encode(6000)
.unwrap();
bytes.extend_from_slice(&frame("tail").encode(1 << 20).unwrap());
std::fs::write(&path, file_bytes(bytes)).unwrap();
let recovery = scanned_frames(&path, 1024);
assert_eq!(recovery.frames.len(), 1);
assert!(matches!(recovery.frames[0], Frame::Known { .. }));
assert_eq!(recovery.damage[0].code(), "FRAME_TOO_LARGE");
let _ = std::fs::remove_file(&path);
}
#[test]
fn repair_reports_all_faults_without_retaining_their_frames() {
let path = tmp_path("repair-multiple-faults.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut bytes = FILE_PROLOGUE.to_vec();
for _ in 0..2 {
let mut broken = frame("damaged").encode(cfg.frame_bytes_max).unwrap();
broken[crate::frame::HEADER_LEN + 4] ^= 1;
bytes.extend(broken);
}
std::fs::write(&path, bytes).unwrap();
let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
assert_eq!(recovery.damage_count, 2);
assert_eq!(recovery.damage.len(), 1);
assert!(recovery.frames.is_empty());
let _ = std::fs::remove_file(path);
}
#[test]
fn repair_streams_unknown_frames_and_keeps_final_selection() {
let path = tmp_path("repair-stream-selection.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
let unknown = Frame::Unknown {
kind: 0xfabc,
version: 17,
payload: vec![42; 1024],
};
let mut opened = open_session(&path, cfg.clone()).unwrap();
for entry in [&root, &child] {
opened
.writer
.append_frame(&Frame::Known {
entry: entry.clone(),
})
.unwrap();
}
opened.writer.append_frame(&unknown).unwrap();
opened.writer.select_entry(root.id).unwrap();
drop(opened);
let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
assert!(
recovery.frames.is_empty(),
"repair must not retain historical frames"
);
let mut seen_unknown = false;
SessionReader::open(&path)
.unwrap()
.scan(cfg.frame_bytes_max, |frame| {
if frame == unknown {
seen_unknown = true;
}
Ok(())
})
.unwrap();
assert!(seen_unknown);
let reopened = open_session(&path, cfg).unwrap();
assert_eq!(reopened.session.selected_id(), Some(root.id));
drop(reopened);
let _ = std::fs::remove_file(path);
}
#[test]
fn repair_rejects_damaged_explicit_selection_without_publication() {
let dir = tmp_path("repair-selected-damaged");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("session.smh");
let cfg = Config::default_valid();
let session = Session::new();
let root = entry("root");
let selected = EntryFrame::new(Some(root.id), &message("selected")).unwrap();
let mut bytes = FILE_PROLOGUE.to_vec();
for frame in [
Frame::Header {
header: session.header(),
},
Frame::Known {
entry: root.clone(),
},
Frame::Known {
entry: selected.clone(),
},
Frame::Selection { id: selected.id },
] {
bytes.extend(frame.encode(cfg.frame_bytes_max).unwrap());
}
let selected_start = FILE_PROLOGUE.len()
+ Frame::Header {
header: session.header(),
}
.encode(cfg.frame_bytes_max)
.unwrap()
.len()
+ Frame::Known { entry: root }
.encode(cfg.frame_bytes_max)
.unwrap()
.len();
let selected_checksum = selected_start + crate::frame::HEADER_LEN + 4;
bytes[selected_checksum] ^= 1;
std::fs::write(&path, &bytes).unwrap();
let before_temps = std::fs::read_dir(&dir)
.unwrap()
.filter_map(std::result::Result::ok)
.filter(|entry| {
entry
.file_name()
.to_string_lossy()
.starts_with("session.smh.smith-repair-")
})
.count();
let error = repair_session(&path, cfg.frame_bytes_max).unwrap_err();
assert_eq!(error.code(), "ENTRY_NOT_FOUND");
assert_eq!(std::fs::read(&path).unwrap(), bytes);
let after_temps = std::fs::read_dir(&dir)
.unwrap()
.filter_map(std::result::Result::ok)
.filter(|entry| {
entry
.file_name()
.to_string_lossy()
.starts_with("session.smh.smith-repair-")
})
.count();
assert_eq!(after_temps, before_temps);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn repair_republishes_file_without_torn_tail() {
let path = tmp_path("repair.smh");
let cfg = Config::default_valid();
let good = frame("a").encode(cfg.frame_bytes_max).unwrap();
let mut bytes = good;
bytes.extend_from_slice(&frame("b").encode(cfg.frame_bytes_max).unwrap()[..6]);
std::fs::write(&path, file_bytes(bytes)).unwrap();
let before = std::fs::metadata(&path).unwrap().len();
let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
assert!(recovery.frames.is_empty());
assert_eq!(recovery.torn_tail_at, Some(before - 6));
let after = std::fs::metadata(&path).unwrap().len();
assert!(after < before);
assert_eq!(scanned_frames(&path, cfg.frame_bytes_max).frames.len(), 1);
let _ = std::fs::remove_file(&path);
}
#[test]
fn writer_rejects_oversized_frame_without_touching_file() {
let path = tmp_path("oversize_reject.smh");
let _ = std::fs::remove_file(&path);
let mut cfg = Config::default_valid();
cfg.frame_bytes_max = 4096;
{
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&frame("small")).unwrap();
let big = EntryFrame::new(
None,
&EntryContent::Message({
let mut m = Message::with_text(Role::User, "x");
m.add_block(ContentBlock::text("y".repeat(8192)));
m
}),
)
.unwrap();
let error = w.append_frame(&Frame::Known { entry: big }).unwrap_err();
assert_eq!(error.code(), "FRAME_TOO_LARGE");
}
assert_eq!(scanned_frames(&path, cfg.frame_bytes_max).frames.len(), 1);
let _ = std::fs::remove_file(&path);
}
#[test]
fn session_identity_mismatch_does_not_mutate_source() {
// SMH-SPEC-SPEC0001: Sessions, identity independent of offsets and filenames.
let original = Session::new();
let path = tmp_path(&format!("{}.smh", original.id()));
let renamed = tmp_path(&format!("{}.smh", smith::id::SessionId::new()));
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&renamed);
drop(open_session(&path, Config::default_valid()).unwrap());
std::fs::rename(&path, &renamed).unwrap();
let before = std::fs::read(&renamed).unwrap();
assert_eq!(
open_session(&renamed, Config::default_valid())
.unwrap_err()
.code(),
"SESSION_ID_MISMATCH"
);
assert_eq!(std::fs::read(&renamed).unwrap(), before);
let _ = std::fs::remove_file(renamed);
}
#[test]
fn observer_prevents_implicit_tail_repair() {
// SMH-SPEC-SPEC0001: Sessions, maintenance cannot invalidate a pinned reader.
let path = tmp_path("observer_torn.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
writer.append_frame(&frame("intact")).unwrap();
drop(writer);
let mut tail = File::options().append(true).open(&path).unwrap();
tail.write_all(&[0, 1]).unwrap();
let before = std::fs::read(&path).unwrap();
let observer = SessionReader::open(&path).unwrap();
assert_eq!(
SessionWriter::open(&path, cfg).unwrap_err().code(),
"STORE_READERS_ACTIVE"
);
assert_eq!(std::fs::read(&path).unwrap(), before);
drop(observer);
let _ = std::fs::remove_file(path);
}
#[test]
fn observer_coexists_with_writer_but_blocks_replacement() {
// SMH-SPEC-SPEC0001: Sessions, Access ownership and repair coordination.
let path = tmp_path("observer_lock.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
writer.append_frame(&frame("one")).unwrap();
let observer = SessionReader::open(&path).unwrap();
assert_eq!(
observer
.scan(cfg.frame_bytes_max, |_| Ok(()))
.unwrap()
.good_end,
std::fs::metadata(&path).unwrap().len()
);
assert_eq!(
SessionWriter::open(&path, cfg.clone()).unwrap_err().code(),
"STORE_BUSY"
);
drop(writer);
assert_eq!(
repair_session(&path, cfg.frame_bytes_max)
.unwrap_err()
.code(),
"STORE_READERS_ACTIVE"
);
drop(observer);
repair_session(&path, cfg.frame_bytes_max).unwrap();
let _ = std::fs::remove_file(path);
}
#[cfg(unix)]
#[test]
fn symlinked_data_and_parent_paths_share_one_writer_lock() {
let dir = tmp_path("symlinked-writer-lock");
let real_dir = dir.join("real");
let parent_alias = dir.join("parent-alias");
let data_alias = dir.join("data-alias.smh");
let real_path = real_dir.join("session.smh");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&real_dir).unwrap();
std::os::unix::fs::symlink(&real_dir, &parent_alias).unwrap();
let writer = SessionWriter::open(&real_path, Config::default_valid()).unwrap();
std::os::unix::fs::symlink(&real_path, &data_alias).unwrap();
for alias in [&parent_alias.join("session.smh"), &data_alias] {
assert_eq!(
SessionWriter::open(alias, Config::default_valid())
.unwrap_err()
.code(),
"STORE_BUSY"
);
}
drop(writer);
let _ = std::fs::remove_dir_all(dir);
}
#[cfg(unix)]
#[test]
fn multiply_linked_session_data_is_rejected() {
let dir = tmp_path("multiply-linked-session");
let path = dir.join("session.smh");
let alias = dir.join("alias.smh");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
drop(SessionWriter::open(&path, Config::default_valid()).unwrap());
std::fs::hard_link(&path, &alias).unwrap();
assert_eq!(
SessionWriter::open(&path, Config::default_valid())
.unwrap_err()
.code(),
"SESSION_AMBIGUOUS_ALIAS"
);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn failed_write_poisoning_rejects_later_appends() {
let path = tmp_path("poisoned-writer.smh");
let _ = std::fs::remove_file(&path);
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
let healthy = writer.file.try_clone().unwrap();
writer.file = File::open(path.parent().unwrap()).unwrap();
assert_eq!(
writer.append_frame(&frame("partial")).unwrap_err().code(),
"STORE_WRITE"
);
writer.file = healthy;
assert_eq!(
writer.append_frame(&frame("later")).unwrap_err().code(),
"STORE_POISONED"
);
let _ = std::fs::remove_file(path);
}
#[test]
fn fault_injected_partial_write_and_sync_failure_poison_writers() {
let partial_path = tmp_path("poisoned-partial-writer.smh");
let sync_path = tmp_path("poisoned-sync-writer.smh");
let _ = std::fs::remove_file(&partial_path);
let _ = std::fs::remove_file(&sync_path);
let bytes = frame("pending")
.encode(Config::default_valid().frame_bytes_max)
.unwrap();
let mut partial = SessionWriter::open(&partial_path, Config::default_valid()).unwrap();
assert_eq!(
partial
.write_step(|file| {
file.write_all(&bytes[..bytes.len() / 2]).map_err(io_error(
smith::error::StorageCode::StoreWrite,
&partial_path,
))?;
Err(storage_fault(
smith::error::StorageCode::StoreWrite,
&partial_path,
"injected partial write failure",
))
})
.unwrap_err()
.code(),
"STORE_WRITE"
);
assert_eq!(
partial.append_frame(&frame("later")).unwrap_err().code(),
"STORE_POISONED"
);
drop(partial);
let mut sync = SessionWriter::open(&sync_path, Config::default_valid()).unwrap();
assert_eq!(
sync.write_step(|file| {
file.write_all(&bytes)
.map_err(io_error(smith::error::StorageCode::StoreWrite, &sync_path))?;
Err(storage_fault(
smith::error::StorageCode::StoreSync,
&sync_path,
"injected sync failure",
))
})
.unwrap_err()
.code(),
"STORE_SYNC"
);
assert_eq!(
sync.append_frame(&frame("later")).unwrap_err().code(),
"STORE_POISONED"
);
drop(sync);
let _ = std::fs::remove_file(partial_path);
let _ = std::fs::remove_file(sync_path);
}
#[test]
fn lock_serializes_second_writer_handle() {
let path = tmp_path("locked.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let writer = SessionWriter::open(&path, cfg).unwrap();
// While the first writer holds the session, an independent handle
// cannot take it: writers cannot interleave frames.
let probe = OpenOptions::new()
.read(true)
.write(true)
.open(lock_path(&path))
.unwrap();
assert!(fs2::FileExt::try_lock_exclusive(&probe).is_err());
drop(writer);
assert!(fs2::FileExt::try_lock_exclusive(&probe).is_ok());
let _ = std::fs::remove_file(&path);
}
#[test]
fn repair_waits_for_a_live_writer_and_keeps_its_entries() {
let path = tmp_path("repair_race.smh");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(lock_path(&path));
let cfg = Config::default_valid();
let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
writer.append_frame(&frame("one")).unwrap();
let repair_path = path.clone();
let limit = cfg.frame_bytes_max;
let (entered, repair_started) = std::sync::mpsc::sync_channel(1);
let repair = std::thread::spawn(move || {
entered.send(()).unwrap();
repair_session(&repair_path, limit).unwrap()
});
// synchronize: the repair has started; the writer's lock holds it
// until the drop, so an append now is still committed to the file
// the repair will republish.
repair_started
.recv_timeout(std::time::Duration::from_secs(10))
.unwrap();
writer.append_frame(&frame("two")).unwrap();
drop(writer);
let recovery = repair.join().unwrap();
assert!(recovery.frames.is_empty());
let loaded = scanned_frames(&path, cfg.frame_bytes_max);
let texts: Vec<String> = loaded.frames.iter().map(text_of).collect();
assert_eq!(texts, vec!["one".to_string(), "two".to_string()]);
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(lock_path(&path));
}
#[test]
fn unknown_frames_survive_write_and_load() {
let path = tmp_path("unknown_frames.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let unknown = Frame::Unknown {
kind: 0x4242,
version: 3,
payload: vec![1, 2, 3, 4, 5, 6],
};
{
let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
w.append_frame(&unknown).unwrap();
w.append_frame(&frame("known")).unwrap();
}
let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
assert_eq!(frames.len(), 2);
match &frames[0] {
Frame::Unknown {
kind,
version,
payload,
} => {
assert_eq!(payload, &vec![1, 2, 3, 4, 5, 6]);
assert_eq!((*kind, *version), (0x4242, 3));
}
other => panic!("expected unknown frame, got {other:?}"),
}
match &frames[1] {
Frame::Known { entry } => assert_eq!(text(entry).as_deref(), Some("known")),
other => panic!("expected known frame, got {other:?}"),
}
let _ = std::fs::remove_file(&path);
}
#[test]
fn end_to_end_tool_activity_persists_and_reconstructs() {
let dir = tmp_path("e2e");
std::fs::create_dir_all(&dir).unwrap();
let workdir = dir.join("work");
std::fs::create_dir_all(&workdir).unwrap();
let session_path = dir.join("e2e.smh");
let _ = std::fs::remove_file(&session_path);
let cfg = Config::default_valid();
let opened_id;
{
let opened = open_session(&session_path, cfg.clone()).unwrap();
opened_id = opened.session.id();
let mut ts = crate::tools::ToolSession::with_session(
workdir.clone(),
opened.session,
opened.writer,
);
ts.invoke(
"write",
&ToolArguments::from_serializable(
&serde_json::json!({"path": "out/hello.txt", "content": "hi"}),
)
.unwrap(),
)
.unwrap();
ts.invoke(
"read",
&ToolArguments::from_serializable(&serde_json::json!({"path": "out/hello.txt"}))
.unwrap(),
)
.unwrap();
}
let restored = SessionReader::open(&session_path)
.unwrap()
.selected_session(cfg.frame_bytes_max)
.unwrap();
assert_eq!(restored.id(), opened_id);
assert_eq!(restored.active_branch().entries().len(), 6);
// Tool call/result/completion triples stay adjacent and ordered.
let kinds: Vec<&str> = restored
.active_branch()
.entries()
.iter()
.map(|e| match e.kind {
EntryKind::ToolCall { .. } => "call",
EntryKind::ToolResult { .. } => "result",
_ => "other",
})
.collect();
assert_eq!(
kinds,
vec!["call", "result", "other", "call", "result", "other"]
);
// The file effect actually happened under the invocation workdir.
assert_eq!(
std::fs::read_to_string(workdir.join("out/hello.txt")).unwrap(),
"hi"
);
let _ = std::fs::remove_file(&session_path);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn reopened_session_keeps_its_identity_and_replays_identically() {
let path = tmp_path("identity.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let created = {
let mut opened = open_session(&path, cfg.clone()).unwrap();
opened.writer.append_frame(&frame("one")).unwrap();
opened.session
};
let first = SessionReader::open(&path).unwrap();
let second = SessionReader::open(&path).unwrap();
let first_header = first.header(cfg.frame_bytes_max).unwrap();
let second_header = second.header(cfg.frame_bytes_max).unwrap();
assert_eq!(first_header.session_id, created.id());
assert_eq!(first_header.branch_id, created.active_branch().id());
assert_eq!(second_header.session_id, first_header.session_id);
// Replay evidence is identical across independent durable readers.
let records = |reader: &SessionReader| {
let mut records = Vec::new();
crate::trace::replay(reader, cfg.frame_bytes_max, |record| {
records.push(record);
Ok(())
})
.unwrap();
records
};
assert_eq!(records(&first), records(&second));
let _ = std::fs::remove_file(&path);
}
#[test]
fn forking_publishes_a_new_session_file_and_leaves_the_source_intact() {
let dir = tmp_path("fork_dir");
std::fs::create_dir_all(&dir).unwrap();
let source_path = dir.join("source.smh");
let fork_path = dir.join("fork.smh");
let cfg = Config::default_valid();
let (source_session, boundary_id, leaf_id) = {
let mut opened = open_session(&source_path, cfg.clone()).unwrap();
let root = *opened.session.append(&message("root")).unwrap();
opened
.writer
.append_entry(&root, opened.session.bytes(&root))
.unwrap();
let boundary = *opened
.session
.append(&EntryContent::Meta {
kind: COMPACTION_KIND.to_string(),
detail: StructuredData::from_serializable(
&serde_json::json!({"summary": "", "kept": 1}),
)
.unwrap(),
})
.unwrap();
opened
.writer
.append_entry(&boundary, opened.session.bytes(&boundary))
.unwrap();
let tip = *opened.session.append(&message("after")).unwrap();
opened
.writer
.append_entry(&tip, opened.session.bytes(&tip))
.unwrap();
assert_ne!(root.id, tip.id);
(opened.session, boundary.id, tip.id)
};
let source_reader = SessionReader::open(&source_path).unwrap();
let fork_id = {
let forked = source_reader
.fork_at(boundary_id, &fork_path, cfg.clone())
.unwrap();
assert_ne!(forked.session.id(), source_session.id());
assert_eq!(forked.session.selected_id(), Some(boundary_id));
assert_eq!(forked.session.compaction_boundary(), Some(boundary_id));
assert!(forked.recovery.damage.is_empty());
forked.session.id()
};
// The fork is durable: reopening restores the same identity and leaf.
{
let reopened = open_session(&fork_path, cfg.clone()).unwrap();
assert_eq!(reopened.session.id(), fork_id);
assert_eq!(reopened.session.selected_id(), Some(boundary_id));
assert_eq!(reopened.session.active_branch().entries().len(), 1);
assert_eq!(reopened.session.compaction_boundary(), Some(boundary_id));
// Full historical chain remains on disk, including pre-compaction content.
let mut history = Vec::new();
SessionReader::open(&fork_path)
.unwrap()
.replay_at(boundary_id, cfg.frame_bytes_max, |entry| {
history.push(entry);
Ok(())
})
.unwrap();
assert_eq!(history.len(), 2);
assert_eq!(
history[0].content,
source_session.bytes(&source_session.active_branch().entries()[0])
);
}
// The source keeps its own identity, tip, and entry count.
{
let source_again = open_session(&source_path, cfg.clone()).unwrap();
assert_eq!(source_again.session.id(), source_session.id());
assert_eq!(source_again.session.selected_id(), Some(leaf_id));
assert_eq!(source_again.session.active_branch().entries().len(), 2);
let mut count = 0;
SessionReader::open(&source_path)
.unwrap()
.replay_at(leaf_id, cfg.frame_bytes_max, |_| {
count += 1;
Ok(())
})
.unwrap();
assert_eq!(count, 3);
}
// A populated target is refused instead of overwritten.
let error = source_reader
.fork_at(boundary_id, &fork_path, cfg)
.unwrap_err();
assert_eq!(error.code(), "SESSION_EXISTS");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn legacy_session_without_prologue_is_rejected_without_repair() {
let path = tmp_path("legacy-no-prologue.smh");
let bytes = frame("legacy").encode(1 << 20).unwrap();
std::fs::write(&path, &bytes).unwrap();
assert_eq!(
open_session(&path, Config::default_valid())
.unwrap_err()
.code(),
"SESSION_PROLOGUE"
);
assert_eq!(std::fs::read(&path).unwrap(), bytes);
let _ = std::fs::remove_file(&path);
}
#[test]
fn session_file_without_header_is_rejected() {
let path = tmp_path("headerless.smh");
let cfg = Config::default_valid();
std::fs::write(
&path,
file_bytes(frame("orphan").encode(cfg.frame_bytes_max).unwrap()),
)
.unwrap();
let error = open_session(&path, cfg).unwrap_err();
assert_eq!(error.code(), "SESSION_HEADER_MISSING");
let _ = std::fs::remove_file(&path);
}
#[test]
fn cancellation_stops_replay_at_entry_boundary_and_leaves_no_fork() {
let path = tmp_path("cancel-replay.smh");
let fork = tmp_path("cancel-replay-fork.smh");
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&fork);
let config = Config::default_valid();
let mut opened = open_session(&path, config.clone()).unwrap();
let root = entry("root");
let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
for entry in [&root, &child] {
opened
.writer
.append_frame(&Frame::Known {
entry: entry.clone(),
})
.unwrap();
}
drop(opened);
let reader = SessionReader::open(&path).unwrap();
let cancel = CancelHandle::new();
let mut visited = 0;
let error = reader
.replay_at_cancelled(child.id, config.frame_bytes_max, &cancel, |_| {
visited += 1;
cancel.cancel();
Ok(())
})
.unwrap_err();
assert_eq!(error.code(), "CANCELLED");
assert_eq!(visited, 1);
assert_eq!(
reader
.fork_at_cancelled(child.id, &fork, config, &cancel)
.unwrap_err()
.code(),
"CANCELLED"
);
assert!(!fork.exists());
drop(reader);
let _ = std::fs::remove_file(path);
}
#[test]
fn reader_drop_releases_an_inherited_snapshot_lock() {
let path = tmp_path("reader-drop-unlocks-inherited-lock.smh");
let _ = std::fs::remove_file(&path);
let opened = open_session(&path, Config::default_valid()).unwrap();
drop(opened);
let reader = SessionReader::open(&path).unwrap();
// A fork inherits this open-file description. Closing only this process's
// descriptor must still release the maintenance lock before the child execs.
let inherited_readers = reader.readers.try_clone().unwrap();
drop(reader);
let coordination = maintenance_lock(&path).unwrap();
drop(coordination);
drop(inherited_readers);
let _ = std::fs::remove_file(path);
}
#[test]
fn checkpoint_tail_snapshots_reopen_to_the_last_authoritative_entry() {
let path = tmp_path("checkpoint-tail-snapshots.smh");
let _ = std::fs::remove_file(&path);
let cfg = Config::default_valid();
let (before_checkpoint, selected, boundary, ancestry) = {
let mut opened = open_session(&path, cfg.clone()).unwrap();
let root = *opened.session.append(&message("root")).unwrap();
opened
.writer
.append_entry(&root, opened.session.bytes(&root))
.unwrap();
let boundary = *opened
.session
.append(&EntryContent::Meta {
kind: COMPACTION_KIND.to_string(),
detail: StructuredData::from_serializable(
&serde_json::json!({"summary": "", "kept": 1}),
)
.unwrap(),
})
.unwrap();
opened
.writer
.append_entry(&boundary, opened.session.bytes(&boundary))
.unwrap();
let mut ancestry = vec![root.id, boundary.id];
for _ in 0..61 {
let record = *opened.session.append(&message("turn")).unwrap();
opened
.writer
.append_entry(&record, opened.session.bytes(&record))
.unwrap();
ancestry.push(record.id);
}
let record = *opened.session.append(&message("selected")).unwrap();
opened
.writer
.append_entry(&record, opened.session.bytes(&record))
.unwrap();
ancestry.push(record.id);
(
std::fs::read(&path).unwrap(),
record.id,
boundary.id,
ancestry,
)
};
let checkpoint = Frame::Checkpoint {
id: selected,
offset: 0,
}
.encode(cfg.frame_bytes_max)
.unwrap();
let checkpoint_at = before_checkpoint.len() - checkpoint.len();
for end in checkpoint_at..before_checkpoint.len() {
std::fs::write(&path, &before_checkpoint[..end]).unwrap();
if end > checkpoint_at {
assert_eq!(
SessionReader::open(&path)
.unwrap()
.scan(cfg.frame_bytes_max, |_| Ok(()))
.unwrap()
.torn_tail_at,
Some(u64::try_from(checkpoint_at).unwrap())
);
}
let reopened = open_session(&path, cfg.clone()).unwrap();
assert_eq!(reopened.session.selected_id(), Some(selected));
assert_eq!(reopened.session.compaction_boundary(), Some(boundary));
drop(reopened);
let mut restored_ancestry = Vec::new();
SessionReader::open(&path)
.unwrap()
.replay_at(selected, cfg.frame_bytes_max, |entry| {
restored_ancestry.push(entry.id);
Ok(())
})
.unwrap();
assert_eq!(restored_ancestry, ancestry);
assert_eq!(
SessionReader::open(&path)
.unwrap()
.scan(cfg.frame_bytes_max, |_| Ok(()))
.unwrap()
.torn_tail_at,
None
);
}
let _ = std::fs::remove_file(path);
}
fn publication_fixture(path: &Path, cfg: &Config) -> (Vec<u8>, [EntryId; 3]) {
let mut opened = open_session(path, cfg.clone()).unwrap();
let root = *opened.session.append(&message("root")).unwrap();
opened
.writer
.append_entry(&root, opened.session.bytes(&root))
.unwrap();
let boundary = *opened
.session
.append(&EntryContent::Meta {
kind: COMPACTION_KIND.to_string(),
detail: StructuredData::from_serializable(
&serde_json::json!({"summary": "", "kept": 1}),
)
.unwrap(),
})
.unwrap();
opened
.writer
.append_entry(&boundary, opened.session.bytes(&boundary))
.unwrap();
let leaf = *opened.session.append(&message("leaf")).unwrap();
opened
.writer
.append_entry(&leaf, opened.session.bytes(&leaf))
.unwrap();
opened
.writer
.append_frame(&Frame::Unknown {
kind: 99,
version: 1,
payload: b"preserve me".to_vec(),
})
.unwrap();
drop(opened);
(
std::fs::read(path).unwrap(),
[root.id, boundary.id, leaf.id],
)
}
fn assert_complete_publication(path: &Path, cfg: &Config, ids: [EntryId; 3], unknown: bool) {
let reader = SessionReader::open(path).unwrap();
reader.verify_full(cfg.frame_bytes_max).unwrap();
let session = reader.selected_session(cfg.frame_bytes_max).unwrap();
assert_eq!(session.selected_id(), Some(ids[2]));
assert_eq!(session.compaction_boundary(), Some(ids[1]));
let mut ancestry = Vec::new();
reader
.replay_at(ids[2], cfg.frame_bytes_max, |entry| {
ancestry.push(entry.id);
Ok(())
})
.unwrap();
assert_eq!(ancestry, ids);
assert_eq!(
scanned_frames(path, cfg.frame_bytes_max)
.frames
.iter()
.any(|frame| matches!(frame, Frame::Unknown { kind: 99, .. })),
unknown
);
}
fn publication_temps(dir: &Path, operation: &str) -> Vec<PathBuf> {
std::fs::read_dir(dir)
.unwrap()
.map(|entry| entry.unwrap().path())
.filter(|path| {
path.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| name.contains(&format!(".smith-{operation}-")))
})
.collect()
}
#[test]
fn fork_published_checkpoint_reopens_complete_history() {
let dir = tmp_path("fork-hard-link-interval");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let source = dir.join("source.smh");
let target = dir.join("fork.smh");
let cfg = Config::default_valid();
let (_, ids) = publication_fixture(&source, &cfg);
let observed = std::rc::Rc::new(std::cell::Cell::new(false));
let observed_at_hook = std::rc::Rc::clone(&observed);
let target_at_hook = target.clone();
set_publication_hook(move |point| {
if point == PublicationCheckpoint::Published {
observed_at_hook.set(true);
assert!(SessionReader::open(&target_at_hook).is_ok());
}
});
SessionReader::open(&source)
.unwrap()
.fork_at(ids[2], &target, cfg)
.unwrap();
clear_publication_hook();
assert!(observed.get());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn fork_and_repair_publication_interruptions_never_expose_partial_sessions() {
use std::panic::{AssertUnwindSafe, catch_unwind, panic_any};
let cfg = Config::default_valid();
let points = [
PublicationCheckpoint::TempCreated,
PublicationCheckpoint::PartialWrite,
PublicationCheckpoint::CompletedWrite,
PublicationCheckpoint::TempSynced,
PublicationCheckpoint::Published,
PublicationCheckpoint::ParentSynced,
PublicationCheckpoint::Cleanup,
];
for point in points {
let dir = tmp_path(&format!("publication-interrupt-{point:?}"));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let source = dir.join("source.smh");
let target = dir.join("fork.smh");
let (before, ids) = publication_fixture(&source, &cfg);
let reader = SessionReader::open(&source).unwrap();
set_publication_hook(move |actual| {
if actual == point {
// Panic simulates abrupt control-flow loss: fork's error cleanup is not run.
panic_any(actual);
}
});
assert!(
catch_unwind(AssertUnwindSafe(|| reader.fork_at(
ids[2],
&target,
cfg.clone()
)))
.is_err()
);
clear_publication_hook();
assert_eq!(std::fs::read(&source).unwrap(), before);
assert_complete_publication(&source, &cfg, ids, true);
if matches!(
point,
PublicationCheckpoint::TempCreated
| PublicationCheckpoint::PartialWrite
| PublicationCheckpoint::CompletedWrite
| PublicationCheckpoint::TempSynced
) {
assert!(!target.exists());
assert!(SessionReader::open(&target).is_err());
assert_eq!(publication_temps(&dir, "fork").len(), 1);
} else {
assert_complete_publication(&target, &cfg, ids, false);
}
drop(reader);
let _ = std::fs::remove_dir_all(dir);
}
for point in points {
let dir = tmp_path(&format!("repair-publication-interrupt-{point:?}"));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let source = dir.join("source.smh");
let (before, ids) = publication_fixture(&source, &cfg);
set_publication_hook(move |actual| {
if actual == point {
// Panic simulates abrupt control-flow loss: repair's error cleanup is not run.
panic_any(actual);
}
});
assert!(
catch_unwind(AssertUnwindSafe(|| repair_session(
&source,
cfg.frame_bytes_max
)))
.is_err()
);
clear_publication_hook();
if matches!(
point,
PublicationCheckpoint::TempCreated
| PublicationCheckpoint::PartialWrite
| PublicationCheckpoint::CompletedWrite
| PublicationCheckpoint::TempSynced
) {
assert_eq!(std::fs::read(&source).unwrap(), before);
assert_complete_publication(&source, &cfg, ids, true);
assert_eq!(publication_temps(&dir, "repair").len(), 1);
} else {
assert_complete_publication(&source, &cfg, ids, true);
assert!(publication_temps(&dir, "repair").is_empty());
}
let _ = std::fs::remove_dir_all(dir);
}
}
#[test]
fn fork_publish_collision_preserves_target_and_temporary_source() {
let dir = tmp_path("fork-publish-failure");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let source = dir.join("source.smh");
let target = dir.join("fork.smh");
let cfg = Config::default_valid();
let (before, ids) = publication_fixture(&source, &cfg);
let occupied = target.clone();
set_publication_hook(move |point| {
if point == PublicationCheckpoint::TempSynced {
std::fs::write(&occupied, b"occupied during publish").unwrap();
}
});
let error = SessionReader::open(&source)
.unwrap()
.fork_at(ids[2], &target, cfg)
.unwrap_err();
clear_publication_hook();
match error {
SmithError::Storage {
code,
operation,
path,
message,
publication,
} => {
assert_eq!(code, smith::error::StorageCode::CreatePublish);
assert_eq!(publication, None);
assert_eq!(operation.as_str(), "publish");
assert_eq!(path, target);
assert!(!message.is_empty());
}
other => panic!("expected publication storage error, got {other:?}"),
}
assert_eq!(std::fs::read(&source).unwrap(), before);
assert_eq!(std::fs::read(&target).unwrap(), b"occupied during publish");
assert_eq!(publication_temps(&dir, "fork").len(), 1);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn concurrent_publishers_choose_exactly_one_winner_without_overwrite() {
use std::sync::{Arc, Barrier};
let dir = tmp_path("concurrent-no-clobber-publication");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let target = dir.join("target.smh");
let barrier = Arc::new(Barrier::new(4));
let mut workers = Vec::new();
for index in 0..4 {
let temp = dir.join(format!("temp-{index}.smh"));
let bytes = format!("publisher-{index}").into_bytes();
let mut file = File::create(&temp).unwrap();
file.write_all(&bytes).unwrap();
file.sync_all().unwrap();
let target = target.clone();
let barrier = Arc::clone(&barrier);
workers.push(std::thread::spawn(move || {
barrier.wait();
(bytes, temp.clone(), publish_no_clobber(&temp, &target))
}));
}
let results: Vec<_> = workers
.into_iter()
.map(|worker| worker.join().unwrap())
.collect();
let winners: Vec<_> = results
.iter()
.filter(|(_, _, result)| result.is_ok())
.collect();
assert_eq!(winners.len(), 1);
assert_eq!(std::fs::read(&target).unwrap(), winners[0].0);
for (_, temp, result) in results {
assert_eq!(temp.exists(), result.is_err());
}
let _ = std::fs::remove_dir_all(dir);
}
#[cfg(unix)]
#[test]
fn no_clobber_publication_never_leaves_an_alias() {
use std::os::unix::fs::MetadataExt;
let dir = tmp_path("no-clobber-publication-nlink");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let temp = dir.join("temp.smh");
let target = dir.join("target.smh");
std::fs::write(&temp, b"new").unwrap();
publish_no_clobber(&temp, &target).unwrap();
assert!(!temp.exists());
assert_eq!(std::fs::metadata(&target).unwrap().nlink(), 1);
std::fs::write(&temp, b"retry").unwrap();
let before = std::fs::read(&target).unwrap();
assert!(publish_no_clobber(&temp, &target).is_err());
assert_eq!(std::fs::read(&target).unwrap(), before);
assert_eq!(std::fs::metadata(&temp).unwrap().nlink(), 1);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn parent_sync_failure_reports_published_durability_uncertain_on_wire() {
let dir = tmp_path("fork-parent-sync-failure");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let source = dir.join("source.smh");
let target = dir.join("fork.smh");
let cfg = Config::default_valid();
let (before, ids) = publication_fixture(&source, &cfg);
set_parent_sync_failure(true);
let error = SessionReader::open(&source)
.unwrap()
.fork_at(ids[2], &target, cfg.clone())
.unwrap_err();
set_parent_sync_failure(false);
assert_eq!(std::fs::read(&source).unwrap(), before);
assert_complete_publication(&target, &cfg, ids, false);
assert_eq!(
serde_json::to_string(&error).unwrap(),
format!(
r#"{{"Storage":{{"code":"STORE_DIR_SYNC","operation":"sync parent directory after publication","path":"{}","message":"injected parent directory sync failure","publication":"published-durability-uncertain"}}}}"#,
dir.display()
)
);
match error {
SmithError::Storage { publication, .. } => assert_eq!(
publication,
Some(smith::error::PublicationState::PublishedDurabilityUncertain)
),
other => panic!("expected published storage error, got {other:?}"),
}
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn temporary_name_exhaustion_reports_final_attempt_without_mutation_or_leak() {
let dir = tmp_path("temporary-name-exhaustion");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("session.smh");
let id = SessionId::from_u128(7);
for attempt in 0..128 {
std::fs::create_dir(
path.with_file_name(format!("session.smh.smith-create-{id}-{attempt}.tmp")),
)
.unwrap();
}
let before = std::fs::read_dir(&dir).unwrap().count();
let error = new_temp_file(&path, "create", id).unwrap_err();
match error {
SmithError::Storage {
code,
operation,
path: attempted,
message,
..
} => {
assert_eq!(code, smith::error::StorageCode::CreateTmp);
assert_eq!(operation.as_str(), "create temporary file");
assert_eq!(
attempted,
path.with_file_name(format!("session.smh.smith-create-{id}-127.tmp"))
);
assert!(!message.is_empty());
}
other => panic!("expected storage error, got {other:?}"),
}
assert!(!path.exists());
assert_eq!(std::fs::read_dir(&dir).unwrap().count(), before);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn header_len_matches_protocol() {
assert_eq!(crate::frame::HEADER_LEN, 12);
assert_eq!(crate::frame::BODY_PREFIX_LEN, 8);
}
}