use std::collections::BTreeMap;
use std::error::Error;
use std::fmt;
use std::fs::File;
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use cap_std::fs::{Dir, OpenOptions};
use vsh_types::{
ApprovalBinding, ApprovalId, BlobId, PrincipalId, SnapshotId, TransactionId, TransactionState,
TransitionError,
};
mod directory;
mod persistent;
pub use directory::{DataDirectory, DataDirectoryError};
pub use persistent::{FileStoreConfig, FileTransactionStore};
static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BlobStore {
blobs_dir: PathBuf,
directory: DataDirectory,
}
impl BlobStore {
pub fn open(data_dir: impl AsRef<Path>) -> Result<Self, BlobStoreError> {
let data_dir = data_dir.as_ref();
let directory = DataDirectory::open_trusted(data_dir).map_err(|source| {
BlobStoreError::io("open data directory", data_dir, io::Error::other(source))
})?;
Self::open_in(&directory)
}
pub fn open_in(data_directory: &DataDirectory) -> Result<Self, BlobStoreError> {
let directory = data_directory.open_real_child("blobs").map_err(|source| {
BlobStoreError::io(
"open capability-rooted blob store",
&data_directory.path().join("blobs"),
source,
)
})?;
directory::sync_directory(data_directory.directory()).map_err(|source| {
BlobStoreError::io("sync blob-store parent", data_directory.path(), source)
})?;
Ok(Self {
blobs_dir: directory.path().to_path_buf(),
directory,
})
}
#[must_use]
pub fn blobs_dir(&self) -> &Path {
&self.blobs_dir
}
pub fn put(&self, bytes: &[u8]) -> Result<BlobId, BlobStoreError> {
let id = BlobId::digest(bytes);
let (hex, shard_path, target_path) = self.location_for(id);
let shard_name = &hex[..2];
let target_name = &hex[2..];
let shard = self
.directory
.open_real_child(shard_name)
.map_err(|source| BlobStoreError::io("open blob shard", &shard_path, source))?;
if entry_exists(shard.directory(), target_name)
.map_err(|source| BlobStoreError::io("inspect immutable blob", &target_path, source))?
{
Self::verify_existing(id, shard.directory(), target_name, &target_path)?;
return Ok(id);
}
let (mut file, temporary_name, temporary_path) =
Self::create_temporary(shard.directory(), &shard_path)?;
if let Err(source) = file.write_all(bytes) {
drop(file);
let _ = shard.directory().remove_file(&temporary_name);
return Err(BlobStoreError::io(
"write temporary blob",
&temporary_path,
source,
));
}
if let Err(source) = file.sync_all() {
drop(file);
let _ = shard.directory().remove_file(&temporary_name);
return Err(BlobStoreError::io(
"sync temporary blob",
&temporary_path,
source,
));
}
drop(file);
match shard
.directory()
.rename(&temporary_name, shard.directory(), target_name)
{
Ok(()) => {}
Err(_source)
if entry_exists(shard.directory(), target_name).map_err(|source| {
BlobStoreError::io("inspect raced immutable blob", &target_path, source)
})? =>
{
let _ = shard.directory().remove_file(&temporary_name);
Self::verify_existing(id, shard.directory(), target_name, &target_path)?;
return Ok(id);
}
Err(source) => {
let _ = shard.directory().remove_file(&temporary_name);
return Err(BlobStoreError::io(
"install immutable blob",
&target_path,
source,
));
}
}
directory::sync_directory(shard.directory())
.map_err(|source| BlobStoreError::io("sync blob shard", &shard_path, source))?;
Self::verify_existing(id, shard.directory(), target_name, &target_path)?;
Ok(id)
}
pub fn get(&self, id: BlobId) -> Result<Vec<u8>, BlobStoreError> {
self.get_bounded(id, usize::MAX)
}
pub fn get_bounded(&self, id: BlobId, maximum: usize) -> Result<Vec<u8>, BlobStoreError> {
let (hex, _shard_path, path) = self.location_for(id);
let shard_name = &hex[..2];
let target_name = &hex[2..];
let shard = self
.directory
.directory()
.open_dir(shard_name)
.map_err(|source| BlobStoreError::io("open blob shard", &path, source))?;
let mut options = OpenOptions::new();
options.read(true);
let mut file = directory::open_real_file(&shard, target_name, &options)
.map_err(|source| BlobStoreError::io("open immutable blob", &path, source))?;
let declared = usize::try_from(
file.metadata()
.map_err(|source| BlobStoreError::io("inspect immutable blob", &path, source))?
.len(),
)
.unwrap_or(usize::MAX);
if declared > maximum {
return Err(BlobStoreError::SizeLimit {
path,
observed: declared,
maximum,
});
}
let mut bytes = Vec::with_capacity(declared);
Read::by_ref(&mut file)
.take(u64::try_from(maximum).unwrap_or(u64::MAX).saturating_add(1))
.read_to_end(&mut bytes)
.map_err(|source| BlobStoreError::io("read immutable blob", &path, source))?;
if bytes.len() > maximum {
return Err(BlobStoreError::SizeLimit {
path,
observed: bytes.len(),
maximum,
});
}
let actual = BlobId::digest(&bytes);
if actual != id {
return Err(BlobStoreError::Corrupt {
path,
expected: id,
actual,
});
}
Ok(bytes)
}
pub fn contains(&self, id: BlobId) -> Result<bool, BlobStoreError> {
let (hex, _shard_path, path) = self.location_for(id);
let shard_name = &hex[..2];
let target_name = &hex[2..];
let shard = match self.directory.directory().open_dir(shard_name) {
Ok(shard) => shard,
Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(false),
Err(source) => return Err(BlobStoreError::io("open blob shard", &path, source)),
};
if !entry_exists(&shard, target_name)
.map_err(|source| BlobStoreError::io("inspect immutable blob", &path, source))?
{
return Ok(false);
}
Self::verify_existing(id, &shard, target_name, &path)?;
Ok(true)
}
fn create_temporary(
shard: &Dir,
shard_path: &Path,
) -> Result<(File, String, PathBuf), BlobStoreError> {
for _ in 0..1024 {
let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let name = format!(".vsh-blob-{}-{sequence}.tmp", std::process::id());
let path = shard_path.join(&name);
let mut options = OpenOptions::new();
options.write(true).create_new(true);
match directory::open_real_file(shard, &name, &options) {
Ok(file) => return Ok((file, name, path)),
Err(source) if source.kind() == io::ErrorKind::AlreadyExists => {}
Err(source) => {
return Err(BlobStoreError::io("create temporary blob", &path, source));
}
}
}
Err(BlobStoreError::TemporaryNameExhausted {
directory: shard_path.to_owned(),
})
}
fn verify_existing(
id: BlobId,
shard: &Dir,
name: &str,
path: &Path,
) -> Result<(), BlobStoreError> {
let mut options = OpenOptions::new();
options.read(true);
let mut file = directory::open_real_file(shard, name, &options)
.map_err(|source| BlobStoreError::io("verify immutable blob", path, source))?;
let actual = hash_file(&mut file)
.map_err(|source| BlobStoreError::io("verify immutable blob", path, source))?;
if actual != id {
return Err(BlobStoreError::Corrupt {
path: path.to_owned(),
expected: id,
actual,
});
}
Ok(())
}
#[cfg(test)]
fn path_for(&self, id: BlobId) -> PathBuf {
self.location_for(id).2
}
fn location_for(&self, id: BlobId) -> (String, PathBuf, PathBuf) {
let hex = id.to_string();
let shard = self.blobs_dir.join(&hex[..2]);
let target = shard.join(&hex[2..]);
(hex, shard, target)
}
}
fn entry_exists(directory: &Dir, name: &str) -> io::Result<bool> {
match directory.symlink_metadata(name) {
Ok(metadata) if metadata.is_file() && !metadata.is_symlink() => Ok(true),
Ok(_) => Err(io::Error::new(
io::ErrorKind::InvalidData,
"immutable blob path is not a real file",
)),
Err(source) if source.kind() == io::ErrorKind::NotFound => Ok(false),
Err(source) => Err(source),
}
}
fn hash_file(file: &mut File) -> io::Result<BlobId> {
let mut hasher = blake3::Hasher::new();
let mut buffer = [0_u8; 16 * 1024];
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
Ok(BlobId::from_bytes(*hasher.finalize().as_bytes()))
}
#[derive(Debug)]
#[non_exhaustive]
pub enum BlobStoreError {
Io {
operation: &'static str,
path: PathBuf,
source: io::Error,
},
Corrupt {
path: PathBuf,
expected: BlobId,
actual: BlobId,
},
SizeLimit {
path: PathBuf,
observed: usize,
maximum: usize,
},
TemporaryNameExhausted {
directory: PathBuf,
},
}
impl BlobStoreError {
fn io(operation: &'static str, path: &Path, source: io::Error) -> Self {
Self::Io {
operation,
path: path.to_owned(),
source,
}
}
}
impl fmt::Display for BlobStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Io {
operation,
path,
source,
} => write!(formatter, "{operation} at {}: {source}", path.display()),
Self::Corrupt {
path,
expected,
actual,
} => write!(
formatter,
"blob corruption at {}: expected {expected}, got {actual}",
path.display()
),
Self::SizeLimit {
path,
observed,
maximum,
} => write!(
formatter,
"blob at {} is {observed} bytes; maximum is {maximum}",
path.display()
),
Self::TemporaryNameExhausted { directory } => write!(
formatter,
"could not allocate a unique temporary blob in {}",
directory.display()
),
}
}
}
impl Error for BlobStoreError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Io { source, .. } => Some(source),
Self::Corrupt { .. } | Self::SizeLimit { .. } | Self::TemporaryNameExhausted { .. } => {
None
}
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TransactionRecord {
id: TransactionId,
base_snapshot: SnapshotId,
state: TransactionState,
artifact: Option<BlobId>,
approval: Option<ApprovalGrant>,
}
impl TransactionRecord {
#[must_use]
pub const fn new(id: TransactionId, base_snapshot: SnapshotId) -> Self {
Self {
id,
base_snapshot,
state: TransactionState::Created,
artifact: None,
approval: None,
}
}
#[must_use]
pub const fn with_artifact(mut self, artifact: BlobId) -> Self {
self.artifact = Some(artifact);
self
}
#[must_use]
pub const fn id(&self) -> TransactionId {
self.id
}
#[must_use]
pub const fn base_snapshot(&self) -> SnapshotId {
self.base_snapshot
}
#[must_use]
pub const fn state(&self) -> TransactionState {
self.state
}
#[must_use]
pub const fn artifact(&self) -> Option<BlobId> {
self.artifact
}
#[must_use]
pub const fn approval(&self) -> Option<ApprovalGrant> {
self.approval
}
pub fn transition(&mut self, next: TransactionState) -> Result<(), TransitionError> {
if !self.state.can_transition_to(next) {
return Err(TransitionError {
from: self.state,
to: next,
});
}
self.state = next;
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ApprovalGrant {
binding: ApprovalBinding,
id: ApprovalId,
}
impl ApprovalGrant {
pub fn new(
transaction: TransactionId,
principal: PrincipalId,
issued_at_unix_ms: u64,
expires_at_unix_ms: u64,
) -> Result<Self, ApprovalGrantError> {
if expires_at_unix_ms <= issued_at_unix_ms {
return Err(ApprovalGrantError::InvalidWindow {
issued_at_unix_ms,
expires_at_unix_ms,
});
}
let binding = ApprovalBinding {
transaction,
principal,
issued_at_unix_ms,
expires_at_unix_ms,
};
Ok(Self {
binding,
id: binding.approval_id(),
})
}
#[must_use]
pub const fn id(self) -> ApprovalId {
self.id
}
#[must_use]
pub const fn transaction(self) -> TransactionId {
self.binding.transaction
}
#[must_use]
pub const fn principal(self) -> PrincipalId {
self.binding.principal
}
#[must_use]
pub const fn issued_at_unix_ms(self) -> u64 {
self.binding.issued_at_unix_ms
}
#[must_use]
pub const fn expires_at_unix_ms(self) -> u64 {
self.binding.expires_at_unix_ms
}
#[must_use]
pub const fn is_expired_at(self, now_unix_ms: u64) -> bool {
now_unix_ms >= self.binding.expires_at_unix_ms
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ApprovalGrantError {
InvalidWindow {
issued_at_unix_ms: u64,
expires_at_unix_ms: u64,
},
}
impl fmt::Display for ApprovalGrantError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidWindow {
issued_at_unix_ms,
expires_at_unix_ms,
} => write!(
formatter,
"approval expiry {expires_at_unix_ms} must follow issuance {issued_at_unix_ms}"
),
}
}
}
impl Error for ApprovalGrantError {}
#[derive(Debug, Eq, PartialEq)]
pub struct CommitReservation {
transaction: TransactionId,
base_snapshot: SnapshotId,
}
impl CommitReservation {
#[must_use]
pub const fn transaction(&self) -> TransactionId {
self.transaction
}
#[must_use]
pub const fn base_snapshot(&self) -> SnapshotId {
self.base_snapshot
}
}
pub trait TransactionStore: Send + Sync {
fn create(&self, record: TransactionRecord) -> Result<(), TransactionStoreError>;
fn get(&self, id: TransactionId) -> Result<TransactionRecord, TransactionStoreError>;
fn compare_and_transition(
&self,
id: TransactionId,
expected: TransactionState,
next: TransactionState,
) -> Result<TransactionRecord, TransactionStoreError>;
fn approve(
&self,
id: TransactionId,
grant: ApprovalGrant,
) -> Result<TransactionRecord, TransactionStoreError>;
fn reserve(
&self,
id: TransactionId,
now_unix_ms: u64,
) -> Result<CommitReservation, TransactionStoreError>;
}
#[derive(Clone, Debug, Default)]
pub struct MemoryTransactionStore {
records: Arc<Mutex<BTreeMap<TransactionId, TransactionRecord>>>,
}
impl MemoryTransactionStore {
fn lock(
&self,
) -> Result<MutexGuard<'_, BTreeMap<TransactionId, TransactionRecord>>, TransactionStoreError>
{
self.records
.lock()
.map_err(|_| TransactionStoreError::Poisoned)
}
}
impl TransactionStore for MemoryTransactionStore {
fn create(&self, record: TransactionRecord) -> Result<(), TransactionStoreError> {
let mut records = self.lock()?;
if records.contains_key(&record.id()) {
return Err(TransactionStoreError::Duplicate { id: record.id() });
}
records.insert(record.id(), record);
Ok(())
}
fn get(&self, id: TransactionId) -> Result<TransactionRecord, TransactionStoreError> {
self.lock()?
.get(&id)
.cloned()
.ok_or(TransactionStoreError::NotFound { id })
}
fn compare_and_transition(
&self,
id: TransactionId,
expected: TransactionState,
next: TransactionState,
) -> Result<TransactionRecord, TransactionStoreError> {
let mut records = self.lock()?;
let record = records
.get_mut(&id)
.ok_or(TransactionStoreError::NotFound { id })?;
if record.state() != expected {
return Err(TransactionStoreError::StateConflict {
id,
expected,
actual: record.state(),
});
}
record
.transition(next)
.map_err(TransactionStoreError::Transition)?;
Ok(record.clone())
}
fn approve(
&self,
id: TransactionId,
grant: ApprovalGrant,
) -> Result<TransactionRecord, TransactionStoreError> {
if grant.transaction() != id {
return Err(TransactionStoreError::ApprovalBindingMismatch {
requested: id,
bound: grant.transaction(),
});
}
let mut records = self.lock()?;
let record = records
.get_mut(&id)
.ok_or(TransactionStoreError::NotFound { id })?;
if record.state() != TransactionState::PendingApproval {
return Err(TransactionStoreError::StateConflict {
id,
expected: TransactionState::PendingApproval,
actual: record.state(),
});
}
record
.transition(TransactionState::Approved)
.map_err(TransactionStoreError::Transition)?;
record.approval = Some(grant);
Ok(record.clone())
}
fn reserve(
&self,
id: TransactionId,
now_unix_ms: u64,
) -> Result<CommitReservation, TransactionStoreError> {
let mut records = self.lock()?;
let record = records
.get_mut(&id)
.ok_or(TransactionStoreError::NotFound { id })?;
match record.state() {
TransactionState::AutoApproved => {}
TransactionState::Approved => {
let grant = record
.approval()
.ok_or(TransactionStoreError::MissingApproval { id })?;
if grant.is_expired_at(now_unix_ms) {
record
.transition(TransactionState::Expired)
.map_err(TransactionStoreError::Transition)?;
return Err(TransactionStoreError::ApprovalExpired {
id,
expired_at_unix_ms: grant.expires_at_unix_ms(),
observed_at_unix_ms: now_unix_ms,
});
}
}
actual => {
return Err(TransactionStoreError::NotReservable { id, actual });
}
}
record
.transition(TransactionState::Reserved)
.map_err(TransactionStoreError::Transition)?;
Ok(CommitReservation {
transaction: id,
base_snapshot: record.base_snapshot(),
})
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum TransactionStoreError {
Duplicate {
id: TransactionId,
},
NotFound {
id: TransactionId,
},
StateConflict {
id: TransactionId,
expected: TransactionState,
actual: TransactionState,
},
Transition(TransitionError),
ApprovalBindingMismatch {
requested: TransactionId,
bound: TransactionId,
},
MissingApproval {
id: TransactionId,
},
ApprovalExpired {
id: TransactionId,
expired_at_unix_ms: u64,
observed_at_unix_ms: u64,
},
NotReservable {
id: TransactionId,
actual: TransactionState,
},
PersistentIo {
operation: &'static str,
kind: io::ErrorKind,
},
PersistentCorrupt {
offset: u64,
reason: &'static str,
},
PersistentLogLimit {
observed: u64,
maximum: u64,
},
PersistentRecordLimit {
observed: usize,
maximum: usize,
},
Poisoned,
}
impl fmt::Display for TransactionStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Duplicate { id } => write!(formatter, "duplicate transaction: {id}"),
Self::NotFound { id } => write!(formatter, "unknown transaction: {id}"),
Self::StateConflict {
id,
expected,
actual,
} => write!(
formatter,
"transaction {id} state conflict: expected {expected:?}, got {actual:?}"
),
Self::Transition(source) => fmt::Display::fmt(source, formatter),
Self::ApprovalBindingMismatch { requested, bound } => write!(
formatter,
"approval binding mismatch: requested {requested}, grant covers {bound}"
),
Self::MissingApproval { id } => {
write!(formatter, "transaction {id} has no bound approval grant")
}
Self::ApprovalExpired {
id,
expired_at_unix_ms,
observed_at_unix_ms,
} => write!(
formatter,
"transaction {id} approval expired at {expired_at_unix_ms}; observed {observed_at_unix_ms}"
),
Self::NotReservable { id, actual } => {
write!(
formatter,
"transaction {id} is not reservable from {actual:?}"
)
}
Self::PersistentIo { operation, kind } => {
write!(
formatter,
"persistent transaction store {operation} failed: {kind}"
)
}
Self::PersistentCorrupt { offset, reason } => write!(
formatter,
"persistent transaction store is corrupt at byte {offset}: {reason}"
),
Self::PersistentLogLimit { observed, maximum } => write!(
formatter,
"persistent transaction log would use {observed} bytes; maximum is {maximum}"
),
Self::PersistentRecordLimit { observed, maximum } => write!(
formatter,
"persistent transaction store would retain {observed} records; maximum is {maximum}"
),
Self::Poisoned => formatter.write_str("transaction store lock is poisoned"),
}
}
}
impl Error for TransactionStoreError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Transition(source) => Some(source),
Self::Duplicate { .. }
| Self::NotFound { .. }
| Self::StateConflict { .. }
| Self::ApprovalBindingMismatch { .. }
| Self::MissingApproval { .. }
| Self::ApprovalExpired { .. }
| Self::NotReservable { .. }
| Self::PersistentIo { .. }
| Self::PersistentCorrupt { .. }
| Self::PersistentLogLimit { .. }
| Self::PersistentRecordLimit { .. }
| Self::Poisoned => None,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use std::io;
use std::sync::{Arc, Barrier};
use std::thread;
struct TestDirectory(PathBuf);
impl Drop for TestDirectory {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.0);
}
}
fn test_store(name: &str) -> (TestDirectory, BlobStore) {
let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let root = std::env::temp_dir().join(format!(
"vsh-store-test-{}-{sequence}-{name}",
std::process::id()
));
let guard = TestDirectory(root.clone());
let store = BlobStore::open(root).unwrap();
(guard, store)
}
#[test]
fn blobs_are_content_addressed_deduplicated_and_verified() {
let (_guard, store) = test_store("round-trip");
let bytes = b"immutable vsh blob";
let first = store.put(bytes).unwrap();
let second = store.put(bytes).unwrap();
assert_eq!(first, second);
assert_eq!(store.get(first).unwrap(), bytes);
assert!(store.contains(first).unwrap());
assert!(!store.contains(BlobId::from_bytes([0xff; 32])).unwrap());
}
#[test]
fn corrupt_blob_is_never_returned() {
let (_guard, store) = test_store("corrupt");
let id = store.put(b"expected").unwrap();
let path = store.path_for(id);
fs::write(&path, b"tampered").unwrap();
let error = store.get(id).unwrap_err();
assert!(matches!(
error,
BlobStoreError::Corrupt {
expected,
actual: _,
path: _
} if expected == id
));
assert!(store.put(b"expected").is_err());
}
#[test]
fn bounded_blob_read_rejects_oversize_before_returning_bytes() {
let (_guard, store) = test_store("bounded");
let id = store.put(b"12345678").unwrap();
let error = store.get_bounded(id, 7).unwrap_err();
assert!(matches!(
error,
BlobStoreError::SizeLimit {
observed: 8,
maximum: 7,
path: _
}
));
assert_eq!(store.get_bounded(id, 8).unwrap(), b"12345678");
}
#[test]
fn record_preserves_identity_and_validates_transitions() {
let id = TransactionId::from_bytes([1; 32]);
let snapshot = SnapshotId::from_bytes([2; 32]);
let mut record = TransactionRecord::new(id, snapshot);
assert_eq!(record.id(), id);
assert_eq!(record.base_snapshot(), snapshot);
assert_eq!(record.state(), TransactionState::Created);
record.transition(TransactionState::Running).unwrap();
assert_eq!(record.state(), TransactionState::Running);
let error = record.transition(TransactionState::Committed).unwrap_err();
assert_eq!(
error,
TransitionError {
from: TransactionState::Running,
to: TransactionState::Committed,
}
);
assert_eq!(record.state(), TransactionState::Running);
}
fn policy_complete_store(id: TransactionId, state: TransactionState) -> MemoryTransactionStore {
let store = MemoryTransactionStore::default();
store
.create(TransactionRecord::new(id, SnapshotId::from_bytes([2; 32])))
.unwrap();
store
.compare_and_transition(id, TransactionState::Created, TransactionState::Running)
.unwrap();
store
.compare_and_transition(
id,
TransactionState::Running,
TransactionState::VirtualComplete,
)
.unwrap();
store
.compare_and_transition(id, TransactionState::VirtualComplete, state)
.unwrap();
store
}
#[test]
fn approval_is_bound_to_exact_transaction_and_expires_closed() {
let id = TransactionId::from_bytes([3; 32]);
let other = TransactionId::from_bytes([4; 32]);
let store = policy_complete_store(id, TransactionState::PendingApproval);
let principal = PrincipalId::digest_label("fresh-judge");
let wrong_grant = ApprovalGrant::new(other, principal, 100, 200).unwrap();
assert!(matches!(
store.approve(id, wrong_grant),
Err(TransactionStoreError::ApprovalBindingMismatch {
requested,
bound
}) if requested == id && bound == other
));
let grant = ApprovalGrant::new(id, principal, 100, 200).unwrap();
let approval_id = grant.id();
let approved = store.approve(id, grant).unwrap();
assert_eq!(approved.approval().unwrap().id(), approval_id);
assert!(matches!(
store.reserve(id, 200),
Err(TransactionStoreError::ApprovalExpired { .. })
));
assert_eq!(store.get(id).unwrap().state(), TransactionState::Expired);
}
#[test]
fn approval_window_must_be_forward_and_digest_is_deterministic() {
let id = TransactionId::from_bytes([5; 32]);
let principal = PrincipalId::digest_label("judge");
assert!(ApprovalGrant::new(id, principal, 10, 10).is_err());
let first = ApprovalGrant::new(id, principal, 10, 11).unwrap();
let second = ApprovalGrant::new(id, principal, 10, 11).unwrap();
assert_eq!(first.id(), second.id());
}
#[test]
fn atomic_reservation_is_single_use_under_concurrency() {
const CONTENDERS: usize = 8;
let id = TransactionId::from_bytes([6; 32]);
let store = policy_complete_store(id, TransactionState::AutoApproved);
let barrier = Arc::new(Barrier::new(CONTENDERS));
let mut threads = Vec::new();
for _ in 0..CONTENDERS {
let store = store.clone();
let barrier = Arc::clone(&barrier);
threads.push(thread::spawn(move || {
barrier.wait();
store.reserve(id, 0).is_ok()
}));
}
let winners = threads
.into_iter()
.map(|thread| thread.join().unwrap())
.filter(|won| *won)
.count();
assert_eq!(winners, 1);
assert_eq!(store.get(id).unwrap().state(), TransactionState::Reserved);
assert!(matches!(
store.reserve(id, 0),
Err(TransactionStoreError::NotReservable {
actual: TransactionState::Reserved,
..
})
));
}
#[cfg(unix)]
#[test]
fn blob_shard_symlink_cannot_redirect_a_write() {
use std::os::unix::fs::symlink;
let (_guard, store) = test_store("shard-symlink");
let outside = std::env::temp_dir().join(format!(
"vsh-store-outside-{}-{}",
std::process::id(),
TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir(&outside).unwrap();
let id = BlobId::digest(b"cannot escape");
let shard = &id.to_string()[..2];
symlink(&outside, store.blobs_dir().join(shard)).unwrap();
assert!(store.put(b"cannot escape").is_err());
assert_eq!(fs::read_dir(&outside).unwrap().count(), 0);
fs::remove_dir(&outside).unwrap();
}
#[test]
fn public_store_errors_have_stable_messages_and_sources() {
let first = TransactionId::from_bytes([1; 32]);
let second = TransactionId::from_bytes([2; 32]);
let transition = TransitionError {
from: TransactionState::Created,
to: TransactionState::Committed,
};
let errors = [
TransactionStoreError::Duplicate { id: first },
TransactionStoreError::NotFound { id: first },
TransactionStoreError::StateConflict {
id: first,
expected: TransactionState::Running,
actual: TransactionState::Created,
},
TransactionStoreError::Transition(transition),
TransactionStoreError::ApprovalBindingMismatch {
requested: first,
bound: second,
},
TransactionStoreError::MissingApproval { id: first },
TransactionStoreError::ApprovalExpired {
id: first,
expired_at_unix_ms: 10,
observed_at_unix_ms: 11,
},
TransactionStoreError::NotReservable {
id: first,
actual: TransactionState::Denied,
},
TransactionStoreError::PersistentIo {
operation: "read",
kind: io::ErrorKind::PermissionDenied,
},
TransactionStoreError::PersistentCorrupt {
offset: 7,
reason: "test",
},
TransactionStoreError::PersistentLogLimit {
observed: 2,
maximum: 1,
},
TransactionStoreError::PersistentRecordLimit {
observed: 2,
maximum: 1,
},
TransactionStoreError::Poisoned,
];
for error in errors {
assert!(!error.to_string().is_empty());
assert_eq!(
Error::source(&error).is_some(),
matches!(error, TransactionStoreError::Transition(_))
);
}
let expected = BlobId::from_bytes([3; 32]);
let actual = BlobId::from_bytes([4; 32]);
let blob_errors = [
BlobStoreError::Io {
operation: "read",
path: PathBuf::from("blob"),
source: io::Error::other("test"),
},
BlobStoreError::Corrupt {
path: PathBuf::from("blob"),
expected,
actual,
},
BlobStoreError::SizeLimit {
path: PathBuf::from("blob"),
observed: 2,
maximum: 1,
},
BlobStoreError::TemporaryNameExhausted {
directory: PathBuf::from("blobs"),
},
];
for error in blob_errors {
assert!(!error.to_string().is_empty());
assert_eq!(
Error::source(&error).is_some(),
matches!(error, BlobStoreError::Io { .. })
);
}
let approval =
ApprovalGrant::new(first, PrincipalId::digest_label("test"), 2, 1).unwrap_err();
assert_eq!(
approval.to_string(),
"approval expiry 1 must follow issuance 2"
);
}
}