///|
/// Minimal byte-oriented storage boundary for immutable index files.
pub(open) trait Directory {
fn write(Self, String, Bytes) -> Unit raise PersistenceError
fn read(Self, String) -> Bytes raise PersistenceError
fn exists(Self, String) -> Bool
}
///|
/// Storage guarantees exposed by a Directory v2 implementation.
pub(all) struct DirectoryCapabilities {
atomic_replace : Bool
durable_sync : Bool
exclusive_lock : Bool
read_only_mapping : Bool
} derive(Eq, @debug.Debug)
///|
/// Durable index lifecycle operations layered on the byte-oriented Directory.
pub(open) trait DirectoryV2: Directory {
fn write_atomic(Self, String, Bytes) -> Unit raise PersistenceError
fn sync(Self, String) -> Unit raise PersistenceError
fn remove(Self, String) -> Unit raise PersistenceError
fn list(Self) -> ReadOnlyArray[String] raise PersistenceError
fn try_acquire_lock(Self, String) -> Int64? raise PersistenceError
fn release_lock(Self, String, Int64) -> Unit raise PersistenceError
fn capabilities(Self) -> DirectoryCapabilities
}
///|
fn validate_directory_name(name : String) -> Unit raise PersistenceError {
guard name != "" &&
name != "." &&
name != ".." &&
!name.contains("/") &&
!name.contains("\\") else {
raise PersistenceError::InvalidName(name)
}
}
///|
/// In-memory Directory implementation. Reads and writes copy bytes so a
/// caller cannot mutate an already committed snapshot through an alias.
pub struct MemoryDirectory {
names : Array[String]
contents : Array[Bytes]
locks : Array[String]
}
///|
pub fn MemoryDirectory::new() -> MemoryDirectory {
{ names: [], contents: [], locks: [] }
}
///|
pub impl Directory for MemoryDirectory with fn write(self, name, bytes) {
validate_directory_name(name)
let snapshot = bytes[:].to_owned()
match self.names.search_by(existing => existing == name) {
Some(index) => self.contents[index] = snapshot
None => {
self.names.push(name)
self.contents.push(snapshot)
}
}
}
///|
pub impl Directory for MemoryDirectory with fn read(self, name) {
validate_directory_name(name)
match self.names.search_by(existing => existing == name) {
Some(index) => self.contents[index][:].to_owned()
None => raise PersistenceError::NotFound(name)
}
}
///|
pub impl Directory for MemoryDirectory with fn exists(self, name) {
try validate_directory_name(name) catch {
_ => false
} noraise {
_ => self.names.search_by(existing => existing == name) is Some(_)
}
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn write_atomic(self, name, bytes) {
Directory::write(self, name, bytes)
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn sync(_self, _name) {
()
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn remove(self, name) {
validate_directory_name(name)
match self.names.search_by(existing => existing == name) {
Some(index) => {
ignore(self.names.remove(index))
ignore(self.contents.remove(index))
}
None => ()
}
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn list(self) {
let names = self.names.copy()
names.sort()
ReadOnlyArray::from_array(names)
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn try_acquire_lock(self, name) {
validate_directory_name(name)
if self.locks.search_by(lock => lock == name) is Some(_) {
None
} else {
self.locks.push(name)
Some(1L)
}
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn release_lock(
self,
name,
_token,
) {
validate_directory_name(name)
match self.locks.search_by(lock => lock == name) {
Some(index) => ignore(self.locks.remove(index))
None => ()
}
}
///|
pub impl DirectoryV2 for MemoryDirectory with fn capabilities(_self) {
{
atomic_replace: true,
durable_sync: true,
exclusive_lock: true,
read_only_mapping: true,
}
}
///|
/// Filesystem-backed Directory rooted at one directory.
pub struct FsDirectory {
root : String
}
///|
fn[T] wrap_fs_error(
operation : () -> T raise @fs.IOError,
) -> T raise PersistenceError {
operation() catch {
@fs.IOError::IOError(message) => raise PersistenceError::Io(message)
}
}
///|
pub fn FsDirectory::new(root : String) -> FsDirectory raise PersistenceError {
guard root != "" else { raise PersistenceError::InvalidName(root) }
if @fs.path_exists(root) {
guard wrap_fs_error(() => @fs.is_dir(root)) else {
raise PersistenceError::Io("directory root is not a directory: \{root}")
}
} else {
wrap_fs_error(() => @fs.create_dir(root))
}
{ root, }
}
///|
fn FsDirectory::path(
self : FsDirectory,
name : String,
) -> String raise PersistenceError {
validate_directory_name(name)
"\{self.root}/\{name}"
}
///|
pub impl Directory for FsDirectory with fn write(self, name, bytes) {
let path = self.path(name)
wrap_fs_error(() => @fs.write_bytes_to_file(path, bytes))
}
///|
pub impl Directory for FsDirectory with fn read(self, name) {
let path = self.path(name)
if !@fs.path_exists(path) {
raise PersistenceError::NotFound(name)
}
wrap_fs_error(() => @fs.read_file_to_bytes(path))
}
///|
pub impl Directory for FsDirectory with fn exists(self, name) {
try self.path(name) catch {
_ => false
} noraise {
path => @fs.path_exists(path)
}
}
///|
pub impl DirectoryV2 for FsDirectory with fn write_atomic(self, name, bytes) {
let target = self.path(name)
let temporary_name = ".\{name}.tmp"
let temporary = self.path(temporary_name)
wrap_fs_error(() => @fs.write_bytes_to_file(temporary, bytes))
platform_sync_file(temporary)
platform_atomic_replace(temporary, target)
platform_sync_file(target)
}
///|
pub impl DirectoryV2 for FsDirectory with fn sync(self, name) {
platform_sync_file(self.path(name))
}
///|
pub impl DirectoryV2 for FsDirectory with fn remove(self, name) {
let path = self.path(name)
if @fs.path_exists(path) {
wrap_fs_error(() => @fs.remove_file(path))
}
}
///|
pub impl DirectoryV2 for FsDirectory with fn list(self) {
let names = wrap_fs_error(() => @fs.read_dir(self.root))
names.sort()
ReadOnlyArray::from_array(names)
}
///|
pub impl DirectoryV2 for FsDirectory with fn try_acquire_lock(self, name) {
platform_try_acquire_lock(self.path(name))
}
///|
pub impl DirectoryV2 for FsDirectory with fn release_lock(self, name, token) {
platform_release_lock(self.path(name), token)
}
///|
pub impl DirectoryV2 for FsDirectory with fn capabilities(_self) {
platform_directory_capabilities()
}
///|
/// Read-only mapping facade. Native builds use durable FsDirectory primitives;
/// portable targets retain the same immutable snapshot semantics.
pub struct MmapDirectory {
inner : FsDirectory
}
///|
pub fn MmapDirectory::new(
root : String,
) -> MmapDirectory raise PersistenceError {
{ inner: FsDirectory::new(root) }
}
///|
pub impl Directory for MmapDirectory with fn write(self, name, bytes) {
Directory::write(self.inner, name, bytes)
}
///|
pub impl Directory for MmapDirectory with fn read(self, name) {
Directory::read(self.inner, name)
}
///|
pub impl Directory for MmapDirectory with fn exists(self, name) {
Directory::exists(self.inner, name)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn write_atomic(self, name, bytes) {
DirectoryV2::write_atomic(self.inner, name, bytes)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn sync(self, name) {
DirectoryV2::sync(self.inner, name)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn remove(self, name) {
DirectoryV2::remove(self.inner, name)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn list(self) {
DirectoryV2::list(self.inner)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn try_acquire_lock(self, name) {
DirectoryV2::try_acquire_lock(self.inner, name)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn release_lock(self, name, token) {
DirectoryV2::release_lock(self.inner, name, token)
}
///|
pub impl DirectoryV2 for MmapDirectory with fn capabilities(self) {
let capabilities = DirectoryV2::capabilities(self.inner)
{
atomic_replace: capabilities.atomic_replace,
durable_sync: capabilities.durable_sync,
exclusive_lock: capabilities.exclusive_lock,
read_only_mapping: true,
}
}