repositories / smith
smith
There are many coding harnesses - but this one is fast
owned by admin
smith-harness/src/inspection.rs
Raw//! Read-only, bounded per-record inspection (`SMH-SPEC-SPEC0001`, Session inspection).
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::SessionId;
use smith_core::content_codec::EntryContentView;
use smith_core::frame::Frame;
use smith_core::session::EntryFrame;
use smith_core::store::SessionReader;
use std::fs::read_dir;
use std::io::Write;
use std::path::{Path, PathBuf};
/// Session files one incremental inspection may visit.
const SESSION_LIST_ENTRIES_MAX: usize = 1024;
fn invalid(path: &Path, reason: impl std::fmt::Display) -> SmithError {
SmithError::Session {
code: smith::error::SessionCode::SessionCorrupt,
message: format!("{}: {reason}", path.display()),
}
}
/// Summary of one named session without decoding its history.
#[derive(Debug)]
pub struct Summary {
/// Identity from the validated file prologue.
pub id: SessionId,
/// On-disk file size observed during discovery.
pub bytes: u64,
}
/// Compare a filename stem against a session identity without allocating:
/// the stem is canonical proquint text exactly when it decodes to the id.
fn stem_is(stem: Option<&std::ffi::OsStr>, id: SessionId) -> bool {
stem.and_then(std::ffi::OsStr::to_str)
.and_then(|text| smith::id::proquint::decode(text).ok())
.is_some_and(|bits| bits == id.as_u128())
}
/// Visit sessions in ID filename order; only each header is decoded.
///
/// This repeatedly enumerates `dir` to select the next path after the prior
/// result. It uses O(1) retained paths and summaries, at O(n²) directory-entry
/// comparisons for n sessions, rather than materializing an index or result set.
///
/// # Errors
/// Fails explicitly if a session path has an invalid prologue or identity.
pub fn list(dir: &Path, mut visit: impl FnMut(Summary) -> Result<()>) -> Result<()> {
let mut after = None;
let mut visited = 0;
while let Some(path) = next_session_path(dir, after.as_deref())? {
if visited == SESSION_LIST_ENTRIES_MAX {
return Err(SmithError::Session {
code: smith::error::SessionCode::SessionListLimit,
message: format!(
"{}: session listing exceeds {SESSION_LIST_ENTRIES_MAX} files; inspect sessions individually with `smith sessions get <id>`",
dir.display()
),
});
}
// The previous path only bounds the search; release it before the
// scan so one path stays live per visited session.
drop(after.take());
let reader = SessionReader::open(&path)?;
let bytes = reader.file_size();
let mut id = None;
let scan = reader.scan_reporting(
Config::default_valid().frame_bytes_max,
|frame| match frame {
Frame::Header { header } => {
id = Some(header.session_id);
Ok(false)
}
_ => Err(invalid(&path, "missing session header")),
},
|damage| Err(invalid(&path, damage)),
)?;
if scan.torn_tail_at.is_some() {
return Err(invalid(&path, "invalid header frame"));
}
let Some(id) = id else {
return Err(invalid(&path, "missing session header"));
};
if !stem_is(path.file_stem(), id) {
return Err(invalid(&path, "filename does not match session identity"));
}
visit(Summary { id, bytes })?;
visited += 1;
after = Some(path);
}
Ok(())
}
fn next_session_path(dir: &Path, after: Option<&Path>) -> Result<Option<PathBuf>> {
let entries = match read_dir(dir) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(invalid(dir, error)),
};
let mut next = None;
for entry in entries {
let path = entry.map_err(|error| invalid(dir, error))?.path();
if path.extension().is_none_or(|extension| extension != "smh")
|| after.is_some_and(|previous| path <= previous)
{
continue;
}
if next
.as_ref()
.is_none_or(|candidate: &PathBuf| path < *candidate)
{
next = Some(path);
}
}
Ok(next)
}
/// Scan all complete entries in physical order within one file-size snapshot.
/// Callback receives one decoded entry at a time; no history is retained.
///
/// # Errors
/// Reports damaged frames, incomplete tails and invalid content explicitly.
pub fn entries(
path: &Path,
mut visit: impl FnMut(&EntryFrame, &EntryContentView<'_>) -> Result<()>,
) -> Result<()> {
let reader = SessionReader::open(path)?;
let mut header = false;
let mut count = 0;
let scan = reader.scan_reporting(
Config::default_valid().frame_bytes_max,
|frame| {
match frame {
Frame::Header { header: found } if count == 0 => {
header = true;
if !stem_is(path.file_stem(), found.session_id) {
return Err(invalid(path, "filename does not match session identity"));
}
}
Frame::Known { entry } if header => {
let content =
EntryContentView::decode(&entry.content).map_err(|e| invalid(path, e))?;
if content.kind() != entry.kind {
return Err(invalid(path, format!("entry {} kind mismatch", entry.id)));
}
visit(&entry, &content)?;
}
Frame::Unknown { .. }
| Frame::Index { .. }
| Frame::Selection { .. }
| Frame::Checkpoint { .. }
if header => {}
_ => return Err(invalid(path, "invalid or duplicate session header")),
}
count += 1;
Ok(true)
},
|damage| Err(invalid(path, damage)),
)?;
if !header {
return Err(invalid(path, "missing session header"));
}
if scan.torn_tail_at.is_some() && !reader.writer_active() {
return Err(invalid(path, "incomplete tail"));
}
Ok(())
}
/// Stream one inspected record to the output boundary without building an entry JSON tree.
///
/// # Errors
/// Returns an I/O error for a broken pipe or failed serialization.
pub fn output(
out: &mut impl Write,
id: SessionId,
entry: &EntryFrame,
content: &EntryContentView<'_>,
json: bool,
get: bool,
) -> std::io::Result<()> {
if json {
write!(out, "{{\"session\":\"{id}\",\"entry\":\"{}\",", entry.id)?;
if get {
match entry.parent {
Some(parent) => write!(out, "\"parent\":\"{parent}\",")?,
None => out.write_all(b"\"parent\":null,")?,
}
write!(out, "\"timestamp_ms\":{},", entry.timestamp_ms)?;
}
out.write_all(b"\"content\":")?;
} else if get {
write!(out, "{id} {} {} ", entry.id, entry.timestamp_ms)?;
} else {
write!(out, "{id}:{}: ", entry.id)?;
}
serde_json::to_writer(&mut *out, content).map_err(std::io::Error::other)?;
if json {
out.write_all(b"}")?;
}
out.write_all(b"\n")
}
/// Search literal UTF-8 substring among recorded string values, not JSON field names.
#[must_use]
pub fn contains(content: &EntryContentView<'_>, pattern: &str) -> bool {
content.contains(pattern)
}
#[cfg(test)]
mod tests {
use super::*;
use smith_alloc::{Counting, Scope, measure};
use smith_core::session::{EntryContent, EntryFrame, Session};
use smith_core::store::SessionWriter;
use std::io::Write;
#[global_allocator]
static ALLOC: Counting = Counting;
const FILES: usize = 4;
const SMALL_ENTRIES: usize = 4;
const LARGE_ENTRIES: usize = 128;
const PAYLOAD_BYTES: usize = 32 * 1024;
// Retain the original ceilings, expressed with their 164-byte reference
// path capacity. Account only for the required live paths (one for list,
// two for grep), whose capacity is determined outside the measured code.
// Longer TMPDIR paths cannot consume the fixed inspection budget, and
// extra path copies or payload allocations still exceed it (SMH-ISSUE-MUTTMPDR).
const REFERENCE_PATH_BYTES: u64 = 164;
const LIST_PEAK_BYTES: u64 = 65_776;
const GREP_PEAK_BYTES: u64 = 131_676;
const PER_ENTRY_VARIANCE_BYTES: u64 = 16;
struct BoundedSink;
impl Write for BoundedSink {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn fixture(dir: &Path, sessions: usize, entries: usize, tag: &str) {
std::fs::create_dir_all(dir).unwrap();
for file in 0..sessions {
let session = Session::new();
let path = dir.join(format!("{}.smh", session.id()));
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
writer
.append_frame(&Frame::Header {
header: session.header(),
})
.unwrap();
for index in 0..entries {
let content = if index % 2 == 0 {
EntryContent::Message(smith::message::Message::with_text(
smith::message::Role::User,
format!("needle {tag} {file} {index} {}", "x".repeat(PAYLOAD_BYTES)),
))
} else {
EntryContent::Meta {
kind: "event".into(),
detail: smith::StructuredData::from_serializable(&serde_json::json!({
"needle": tag,
"index": index,
"payload": "x".repeat(PAYLOAD_BYTES),
"nested": [{"kind": "structured", "values": [true, 0, null]}],
}))
.unwrap(),
}
};
writer
.append_frame(&Frame::Known {
entry: EntryFrame::new(None, &content).unwrap(),
})
.unwrap();
}
}
}
fn list_cost(dir: &Path) -> smith_alloc::Stats {
let (_, cost) = measure(Scope::IoRequest, || {
let mut seen = 0;
list(dir, |_| {
seen += 1;
Ok(())
})
.unwrap();
assert_eq!(seen, FILES);
});
cost
}
fn grep_cost(dir: &Path, entries_per_file: usize) -> smith_alloc::Stats {
let (_, cost) = measure(Scope::IoRequest, || {
let mut sink = BoundedSink;
let mut matches = 0;
list(dir, |summary| {
entries(
&dir.join(format!("{}.smh", summary.id)),
|entry, content| {
if contains(content, "needle") {
matches += 1;
output(&mut sink, summary.id, entry, content, true, false).unwrap();
}
Ok(())
},
)
})
.unwrap();
assert_eq!(matches, FILES * (entries_per_file / 2));
});
cost
}
#[test]
fn stem_comparison_decodes_without_allocating() {
// SMH-SPEC-SPEC0001: filename identity is checked by value, not text.
let id = SessionId::from_u128(0x5E55);
let matching = PathBuf::from(format!("{id}.smh"));
let mismatched = PathBuf::from(format!("{}.smh", SessionId::from_u128(0x5E56)));
let unparsable = PathBuf::from("not-an-id.smh");
let uppercase = PathBuf::from(format!("{}.smh", id.to_string().to_uppercase()));
let rootless = PathBuf::from("/");
let (verdicts, cost) = measure(Scope::IoRequest, || {
[&matching, &mismatched, &unparsable, &uppercase, &rootless]
.map(|path| stem_is(path.file_stem(), id))
});
assert_eq!(verdicts, [true, false, false, false, false]);
assert_eq!(cost.allocations, 0, "stem comparison allocated");
}
#[test]
fn list_visits_stable_order_incrementally() {
let dir = std::env::temp_dir().join(format!("smith-inspect-list-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let mut expected = Vec::new();
for _ in 0..3 {
let session = Session::new();
let path = dir.join(format!("{}.smh", session.id()));
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
writer
.append_frame(&Frame::Header {
header: session.header(),
})
.unwrap();
expected.push(session.id());
}
expected.sort();
let mut visited = Vec::new();
list(&dir, |summary| {
visited.push(summary.id);
Ok(())
})
.unwrap();
assert_eq!(visited, expected);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn listing_stops_at_the_named_session_file_limit() {
let dir = std::env::temp_dir().join(format!("smith-inspect-limit-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
for _ in 0..=SESSION_LIST_ENTRIES_MAX {
let session = Session::new();
let path = dir.join(format!("{}.smh", session.id()));
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
writer
.append_frame(&Frame::Header {
header: session.header(),
})
.unwrap();
}
let error = list(&dir, |_| Ok(())).unwrap_err();
assert_eq!(error.code(), "SESSION_LIST_LIMIT");
assert!(error.to_string().contains("inspect sessions individually"));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn inspection_large_history_peak_is_independent_of_payload_bytes() {
// SMH-SPEC-SPEC0001: list and grep stream bounded records, including
// JSON-like structured content, rather than retaining history or output.
let root = std::env::temp_dir().join(format!("smith-inspect-{}", std::process::id()));
for prefix_bytes in [0, 64, 128] {
let temporary = root.join("p".repeat(prefix_bytes));
std::fs::create_dir_all(&temporary).unwrap();
let temporary = std::fs::canonicalize(&temporary).unwrap();
let small = temporary.join("small");
let large = temporary.join("large");
let path_bytes = u64::try_from(
small
.join(format!("{}.smh", SessionId::from_u128(0)))
.capacity(),
)
.unwrap();
fixture(&small, FILES, SMALL_ENTRIES, "history");
fixture(&large, FILES, LARGE_ENTRIES, "history");
let listed_small = list_cost(&small);
let listed_large = list_cost(&large);
assert_eq!(
listed_small.peak, listed_large.peak,
"listing peak is per file, not history payload bytes"
);
let listed_peak =
listed_large.peak.checked_sub(path_bytes).unwrap() + REFERENCE_PATH_BYTES;
assert!(
listed_peak <= LIST_PEAK_BYTES,
"path-normalized listing peak {listed_peak} bytes exceeds ceiling {LIST_PEAK_BYTES} bytes"
);
let grepped_small = grep_cost(&small, SMALL_ENTRIES);
let grepped_large = grep_cost(&large, LARGE_ENTRIES);
assert!(
grepped_large.peak <= grepped_small.peak + PER_ENTRY_VARIANCE_BYTES,
"grep peak grew from {} to {} bytes",
grepped_small.peak,
grepped_large.peak
);
let grepped_peak =
grepped_large.peak.checked_sub(2 * path_bytes).unwrap() + 2 * REFERENCE_PATH_BYTES;
assert!(
grepped_peak <= GREP_PEAK_BYTES,
"path-normalized grep peak {grepped_peak} bytes exceeds ceiling {GREP_PEAK_BYTES} bytes"
);
assert!(
grepped_large.peak < 256 * 1024,
"grep peak {} bytes for {} bytes of session payload",
grepped_large.peak,
std::fs::read_dir(&large)
.unwrap()
.filter_map(std::result::Result::ok)
.filter_map(|entry| entry.metadata().ok())
.map(|metadata| metadata.len())
.sum::<u64>()
);
}
std::fs::remove_dir_all(root).unwrap();
}
#[test]
fn live_incomplete_tail_is_pending_until_writer_exits() {
// SMH-SPEC-SPEC0001: Sessions, publication and observer snapshots.
let dir = std::env::temp_dir().join(format!("smith-inspect-tail-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let session = Session::new();
let path = dir.join(format!("{}.smh", session.id()));
let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
writer
.append_frame(&Frame::Header {
header: session.header(),
})
.unwrap();
let mut tail = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
tail.write_all(&[4, 0]).unwrap();
assert!(entries(&path, |_, _| Ok(())).is_ok());
drop(writer);
assert_eq!(
entries(&path, |_, _| Ok(())).unwrap_err().code(),
"SESSION_CORRUPT"
);
let _ = std::fs::remove_dir_all(dir);
}
}