use std::{
collections::HashSet,
fs::{File, OpenOptions},
io::{Read, Seek, Write},
path::{Path, PathBuf},
};
use anyhow::{Context as _, ensure};
use sha2::{Digest, Sha256};
const MAGIC: &[u8] = b"KSPENDING01\n";
const FIXED_HEADER_BYTES: usize = 2 + 4 + 4 + 8 + 32;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StoredPendingObject {
pub file_name: String,
pub media_type: String,
pub bytes: Vec<u8>,
}
#[derive(Clone, Debug)]
pub struct PendingObjectStore {
directory: PathBuf,
namespace: String,
format_id: String,
}
impl PendingObjectStore {
pub fn new(
directory: impl Into<PathBuf>,
namespace: impl Into<String>,
format_id: impl Into<String>,
) -> anyhow::Result<Self> {
let namespace = namespace.into();
let format_id = format_id.into();
validate_namespace(&namespace)?;
ensure!(!format_id.is_empty(), "format ID cannot be empty");
u16::try_from(format_id.len()).context("format ID is too long")?;
Ok(Self {
directory: directory.into(),
namespace,
format_id,
})
}
pub fn install(
&self,
position: u64,
file_name: &str,
media_type: &str,
bytes: &[u8],
) -> anyhow::Result<()> {
ensure!(
!file_name.trim().is_empty(),
"object filename cannot be empty"
);
ensure!(
!media_type.trim().is_empty(),
"object media type cannot be empty"
);
let format_id = self.format_id.as_bytes();
let format_id_len = u16::try_from(format_id.len()).context("format ID is too long")?;
let file_name_len =
u32::try_from(file_name.len()).context("object filename exceeds 4 GiB")?;
let media_type_len =
u32::try_from(media_type.len()).context("object media type exceeds 4 GiB")?;
let object_len = u64::try_from(bytes.len()).context("object exceeds addressable size")?;
let final_path = self.final_path(position);
let temp_path = self.temp_path(position);
ensure!(
!final_path.exists(),
"pending object file {} already exists",
final_path.display()
);
if temp_path.exists() {
std::fs::remove_file(&temp_path)
.with_context(|| format!("removing stale temporary {}", temp_path.display()))?;
}
let mut file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&temp_path)
.with_context(|| format!("creating temporary {}", temp_path.display()))?;
file.write_all(MAGIC)?;
file.write_all(&format_id_len.to_le_bytes())?;
file.write_all(&file_name_len.to_le_bytes())?;
file.write_all(&media_type_len.to_le_bytes())?;
file.write_all(&object_len.to_le_bytes())?;
file.write_all(&Sha256::digest(bytes))?;
file.write_all(format_id)?;
file.write_all(file_name.as_bytes())?;
file.write_all(media_type.as_bytes())?;
file.write_all(bytes)?;
file.sync_all()
.with_context(|| format!("synchronizing temporary {}", temp_path.display()))?;
std::fs::rename(&temp_path, &final_path).with_context(|| {
format!(
"renaming {} to {}",
temp_path.display(),
final_path.display()
)
})?;
sync_directory(&self.directory)
}
pub fn read(&self, position: u64) -> anyhow::Result<StoredPendingObject> {
let path = self.final_path(position);
let mut file = File::open(&path)
.with_context(|| format!("opening pending object {}", path.display()))?;
let mut magic = vec![0_u8; MAGIC.len()];
file.read_exact(&mut magic)?;
ensure!(
magic == MAGIC,
"{} is not a pending-object file",
path.display()
);
let mut fixed = [0_u8; FIXED_HEADER_BYTES];
file.read_exact(&mut fixed)?;
let format_id_len = u16::from_le_bytes(fixed[0..2].try_into().unwrap()) as usize;
let file_name_len = u32::from_le_bytes(fixed[2..6].try_into().unwrap()) as usize;
let media_type_len = u32::from_le_bytes(fixed[6..10].try_into().unwrap()) as usize;
let object_len = u64::from_le_bytes(fixed[10..18].try_into().unwrap());
let checksum = &fixed[18..50];
let variable_len = format_id_len
.checked_add(file_name_len)
.and_then(|value| value.checked_add(media_type_len))
.context("pending-object header length overflow")?;
let expected_file_len = u64::try_from(MAGIC.len() + FIXED_HEADER_BYTES)
.context("pending-object fixed header does not fit u64")?
.checked_add(
u64::try_from(variable_len)
.context("pending-object variable header does not fit u64")?,
)
.and_then(|value| value.checked_add(object_len))
.context("pending-object declared length overflow")?;
ensure!(
file.metadata()?.len() == expected_file_len,
"pending-object declared length differs from file length"
);
let mut variable = vec![0_u8; variable_len];
file.read_exact(&mut variable)?;
let format_id = std::str::from_utf8(&variable[..format_id_len])?;
ensure!(
format_id == self.format_id,
"unsupported pending-object format {format_id}"
);
let file_name_end = format_id_len + file_name_len;
let file_name = std::str::from_utf8(&variable[format_id_len..file_name_end])?.to_owned();
let media_type =
std::str::from_utf8(&variable[file_name_end..file_name_end + media_type_len])?
.to_owned();
ensure!(!file_name.trim().is_empty(), "object filename is empty");
ensure!(!media_type.trim().is_empty(), "object media type is empty");
let object_len =
usize::try_from(object_len).context("pending object does not fit memory")?;
let mut bytes = vec![0_u8; object_len];
file.read_exact(&mut bytes)?;
ensure!(
file.stream_position()? == expected_file_len,
"pending-object file has trailing bytes"
);
let actual_checksum = Sha256::digest(&bytes);
ensure!(
&actual_checksum[..] == checksum,
"pending-object checksum mismatch"
);
Ok(StoredPendingObject {
file_name,
media_type,
bytes,
})
}
pub fn verify_all(&self, referenced_positions: &[u64]) -> anyhow::Result<()> {
for &position in referenced_positions {
self.read(position)?;
}
Ok(())
}
pub fn reconcile(&self, referenced_positions: &[u64]) -> anyhow::Result<()> {
let referenced = referenced_positions.iter().copied().collect::<HashSet<_>>();
let mut removed = false;
if self.directory.exists() {
for entry in std::fs::read_dir(&self.directory)
.with_context(|| format!("reading directory {}", self.directory.display()))?
{
let entry = entry?;
let Some(name) = entry.file_name().to_str().map(str::to_owned) else {
continue;
};
let Some((position, temporary)) = self.parse_file_name(&name) else {
continue;
};
if temporary || !referenced.contains(&position) {
std::fs::remove_file(entry.path())
.with_context(|| format!("removing {}", entry.path().display()))?;
removed = true;
}
}
}
if removed {
sync_directory(&self.directory)?;
}
self.verify_all(referenced_positions)
}
pub fn delete_all(&self) -> anyhow::Result<()> {
if self.directory.exists() {
for entry in std::fs::read_dir(&self.directory)
.with_context(|| format!("reading directory {}", self.directory.display()))?
{
let entry = entry?;
let Some(name) = entry.file_name().to_str().map(str::to_owned) else {
continue;
};
if self.parse_file_name(&name).is_some() {
std::fs::remove_file(entry.path())
.with_context(|| format!("removing {}", entry.path().display()))?;
}
}
}
sync_directory(&self.directory)
}
fn final_path(&self, position: u64) -> PathBuf {
self.directory
.join(format!("{}-{position}.pending-object", self.namespace))
}
fn temp_path(&self, position: u64) -> PathBuf {
self.directory
.join(format!("{}-{position}.pending-object.tmp", self.namespace))
}
fn parse_file_name(&self, name: &str) -> Option<(u64, bool)> {
let tail = name.strip_prefix(&format!("{}-", self.namespace))?;
let (number, temporary) = if let Some(number) = tail.strip_suffix(".pending-object.tmp") {
(number, true)
} else {
(tail.strip_suffix(".pending-object")?, false)
};
if number.is_empty() || number.starts_with('0') && number != "0" {
return None;
}
let position = number.parse::<u64>().ok()?;
(position.to_string() == number).then_some((position, temporary))
}
}
fn validate_namespace(namespace: &str) -> anyhow::Result<()> {
ensure!(!namespace.is_empty(), "namespace cannot be empty");
ensure!(namespace.len() <= 255, "namespace exceeds 255 characters");
ensure!(
namespace
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')),
"namespace contains characters that are unsafe in filenames"
);
Ok(())
}
fn sync_directory(directory: &Path) -> anyhow::Result<()> {
File::open(directory)
.with_context(|| format!("opening directory {}", directory.display()))?
.sync_all()
.with_context(|| format!("synchronizing directory {}", directory.display()))
}
#[cfg(test)]
mod tests {
use std::time::{SystemTime, UNIX_EPOCH};
use super::*;
struct TestDirectory(PathBuf);
impl TestDirectory {
fn new(label: &str) -> Self {
Self(std::env::temp_dir().join(format!(
"pending-object-store-{label}-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
)))
}
}
impl Drop for TestDirectory {
fn drop(&mut self) {
if self.0.exists() {
std::fs::remove_dir_all(&self.0).unwrap();
}
}
}
#[test]
fn construction_validates_configuration_without_touching_filesystem() {
let directory = TestDirectory::new("new");
let store = PendingObjectStore::new(&directory.0, "session_1", "0.2.1").unwrap();
assert!(!store.directory.exists());
assert!(PendingObjectStore::new(&directory.0, "", "0.2.1").is_err());
assert!(PendingObjectStore::new(&directory.0, "bad/name", "0.2.1").is_err());
assert!(PendingObjectStore::new(&directory.0, "a".repeat(256), "0.2.1").is_err());
assert!(PendingObjectStore::new(&directory.0, "valid", "").is_err());
assert!(PendingObjectStore::new(&directory.0, "valid", "x".repeat(65_536)).is_err());
assert!(!directory.0.exists());
}
#[test]
fn install_matches_the_session_log_format_and_read_verifies_it() {
let directory = TestDirectory::new("format");
std::fs::create_dir(&directory.0).unwrap();
let store = PendingObjectStore::new(&directory.0, "session-7", "0.2.1").unwrap();
store
.install(3, "notes.txt", "text/plain", b"durable bytes")
.unwrap();
let mut expected = Vec::new();
expected.extend_from_slice(MAGIC);
expected.extend_from_slice(&5_u16.to_le_bytes());
expected.extend_from_slice(&9_u32.to_le_bytes());
expected.extend_from_slice(&10_u32.to_le_bytes());
expected.extend_from_slice(&13_u64.to_le_bytes());
expected.extend_from_slice(&Sha256::digest(b"durable bytes"));
expected.extend_from_slice(b"0.2.1notes.txttext/plaindurable bytes");
assert_eq!(
std::fs::read(directory.0.join("session-7-3.pending-object")).unwrap(),
expected
);
assert!(!directory.0.join("session-7-3.pending-object.tmp").exists());
assert_eq!(
store.read(3).unwrap(),
StoredPendingObject {
file_name: "notes.txt".into(),
media_type: "text/plain".into(),
bytes: b"durable bytes".to_vec(),
}
);
assert!(
store
.install(3, "replacement", "text/plain", b"new")
.is_err()
);
assert_eq!(
std::fs::read(directory.0.join("session-7-3.pending-object")).unwrap(),
expected
);
}
#[test]
fn install_removes_only_its_exact_stale_temporary_file() {
let directory = TestDirectory::new("temporary");
std::fs::create_dir(&directory.0).unwrap();
let store = PendingObjectStore::new(&directory.0, "space", "format").unwrap();
let stale = directory.0.join("space-4.pending-object.tmp");
let other = directory.0.join("space-04.pending-object.tmp");
std::fs::write(&stale, b"stale").unwrap();
std::fs::write(&other, b"keep").unwrap();
store.install(4, "a", "b", b"c").unwrap();
assert!(!stale.exists());
assert!(other.exists());
assert!(
PendingObjectStore::new(directory.0.join("missing"), "space", "format")
.unwrap()
.install(0, "a", "b", b"c")
.is_err()
);
}
#[test]
fn reads_reject_wrong_format_length_checksum_and_empty_metadata() {
let directory = TestDirectory::new("invalid");
std::fs::create_dir(&directory.0).unwrap();
let store = PendingObjectStore::new(&directory.0, "object", "format-a").unwrap();
store.install(0, "file", "type", b"bytes").unwrap();
assert!(
PendingObjectStore::new(&directory.0, "object", "format-b")
.unwrap()
.read(0)
.is_err()
);
let path = directory.0.join("object-0.pending-object");
let valid = std::fs::read(&path).unwrap();
let mut corrupted = valid.clone();
*corrupted.last_mut().unwrap() ^= 0xff;
std::fs::write(&path, &corrupted).unwrap();
assert!(store.read(0).is_err());
let mut trailing = valid.clone();
trailing.push(0);
std::fs::write(&path, trailing).unwrap();
assert!(store.read(0).is_err());
let mut empty_name = valid;
let fixed_start = MAGIC.len();
empty_name[fixed_start + 2..fixed_start + 6].copy_from_slice(&0_u32.to_le_bytes());
std::fs::write(&path, empty_name).unwrap();
assert!(store.read(0).is_err());
}
#[test]
fn verification_does_not_clean_and_reconciliation_is_exact() {
let directory = TestDirectory::new("reconcile");
std::fs::create_dir(&directory.0).unwrap();
let store = PendingObjectStore::new(&directory.0, "abc", "format").unwrap();
store.install(1, "one", "type", b"one").unwrap();
store.install(2, "two", "type", b"two").unwrap();
let temporary = directory.0.join("abc-8.pending-object.tmp");
let leading_zero = directory.0.join("abc-08.pending-object.tmp");
let similar_prefix = directory.0.join("abc-other-9.pending-object");
let malformed = directory.0.join("abc-x.pending-object");
std::fs::write(&temporary, b"remove").unwrap();
std::fs::write(&leading_zero, b"keep").unwrap();
std::fs::write(&similar_prefix, b"keep").unwrap();
std::fs::write(&malformed, b"keep").unwrap();
store.verify_all(&[1]).unwrap();
assert!(temporary.exists());
store.reconcile(&[1]).unwrap();
assert!(directory.0.join("abc-1.pending-object").exists());
assert!(!directory.0.join("abc-2.pending-object").exists());
assert!(!temporary.exists());
assert!(leading_zero.exists());
assert!(similar_prefix.exists());
assert!(malformed.exists());
assert!(store.reconcile(&[99]).is_err());
}
#[test]
fn delete_all_removes_only_recognized_names() {
let directory = TestDirectory::new("delete");
std::fs::create_dir(&directory.0).unwrap();
let store = PendingObjectStore::new(&directory.0, "abc", "format").unwrap();
store.install(0, "zero", "type", b"zero").unwrap();
let temporary = directory.0.join("abc-1.pending-object.tmp");
let leading_zero = directory.0.join("abc-01.pending-object");
let similar_prefix = directory.0.join("abcd-2.pending-object");
std::fs::write(&temporary, b"remove").unwrap();
std::fs::write(&leading_zero, b"keep").unwrap();
std::fs::write(&similar_prefix, b"keep").unwrap();
store.delete_all().unwrap();
assert!(!directory.0.join("abc-0.pending-object").exists());
assert!(!temporary.exists());
assert!(leading_zero.exists());
assert!(similar_prefix.exists());
}
}