use std::collections::HashMap;
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};
use super::persist::{CodecError, Reader, push_bytes, push_u64, write_atomic};
use crate::tree::Hash;
pub use super::time::{Timestamp, current_timestamp};
const REGISTRY_MAGIC: &[u8; 4] = b"HSR1";
const LOG_MAGIC: &[u8; 4] = b"HCL1";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SnapshotEntry {
pub name: String,
pub root_hash: Hash,
pub timestamp: Timestamp,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommitLogEntry {
pub root_hash: Hash,
pub timestamp: Timestamp,
}
#[derive(Debug)]
pub enum SnapshotError {
DuplicateName(String),
Corrupt(String),
Io(std::io::Error),
}
impl fmt::Display for SnapshotError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::DuplicateName(name) => write!(f, "snapshot name already exists: {name}"),
Self::Corrupt(reason) => write!(f, "snapshot store corrupted: {reason}"),
Self::Io(error) => write!(f, "snapshot store I/O error: {error}"),
}
}
}
impl std::error::Error for SnapshotError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(error) => Some(error),
Self::DuplicateName(_) | Self::Corrupt(_) => None,
}
}
}
impl From<std::io::Error> for SnapshotError {
fn from(error: std::io::Error) -> Self {
Self::Io(error)
}
}
impl From<CodecError> for SnapshotError {
fn from(error: CodecError) -> Self {
match error {
CodecError::Corrupt(reason) => Self::Corrupt(reason),
CodecError::Io(io_error) => Self::Io(io_error),
}
}
}
#[derive(Debug)]
pub struct SnapshotRegistry {
entries: Vec<SnapshotEntry>,
index: HashMap<String, Hash>,
path: Option<PathBuf>,
}
impl SnapshotRegistry {
pub fn new() -> Self {
Self {
entries: Vec::new(),
index: HashMap::new(),
path: None,
}
}
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SnapshotError> {
let path = path.as_ref().to_path_buf();
let entries = match fs::read(&path) {
Ok(bytes) => decode_registry(&bytes)?,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(error) => return Err(SnapshotError::Io(error)),
};
let mut index = HashMap::with_capacity(entries.len());
for entry in &entries {
index.insert(entry.name.clone(), entry.root_hash);
}
Ok(Self {
entries,
index,
path: Some(path),
})
}
pub fn name(&mut self, name: &str, root_hash: Hash) -> Result<(), SnapshotError> {
self.name_at(name, root_hash, current_timestamp())
}
pub fn name_at(
&mut self,
name: &str,
root_hash: Hash,
timestamp: Timestamp,
) -> Result<(), SnapshotError> {
if self.index.contains_key(name) {
return Err(SnapshotError::DuplicateName(name.to_owned()));
}
self.entries.push(SnapshotEntry {
name: name.to_owned(),
root_hash,
timestamp,
});
self.index.insert(name.to_owned(), root_hash);
if let Err(error) = self.persist() {
self.entries.pop();
self.index.remove(name);
return Err(error);
}
Ok(())
}
pub fn get(&self, name: &str) -> Option<Hash> {
self.index.get(name).copied()
}
pub fn remove(&mut self, name: &str) -> Result<Option<SnapshotEntry>, SnapshotError> {
let Some(position) = self.entries.iter().position(|entry| entry.name == name) else {
return Ok(None);
};
let removed = self.entries.remove(position);
self.index.remove(name);
if let Err(error) = self.persist() {
self.entries.insert(position, removed.clone());
self.index.insert(removed.name.clone(), removed.root_hash);
return Err(error);
}
Ok(Some(removed))
}
pub fn list_snapshots(&self) -> Vec<(String, Hash, Timestamp)> {
self.entries
.iter()
.map(|entry| (entry.name.clone(), entry.root_hash, entry.timestamp))
.collect()
}
pub const fn len(&self) -> usize {
self.entries.len()
}
pub const fn is_empty(&self) -> bool {
self.entries.is_empty()
}
fn persist(&self) -> Result<(), SnapshotError> {
if let Some(path) = &self.path {
write_atomic(path, &encode_registry(&self.entries)).map_err(CodecError::from)?;
}
Ok(())
}
}
impl Default for SnapshotRegistry {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug)]
pub struct CommitLog {
entries: Vec<CommitLogEntry>,
path: Option<PathBuf>,
}
impl CommitLog {
pub const fn new() -> Self {
Self {
entries: Vec::new(),
path: None,
}
}
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SnapshotError> {
let path = path.as_ref().to_path_buf();
let entries = match fs::read(&path) {
Ok(bytes) => decode_log(&bytes)?,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(error) => return Err(SnapshotError::Io(error)),
};
Ok(Self {
entries,
path: Some(path),
})
}
pub fn append(&mut self, root_hash: Hash, timestamp: Timestamp) -> Result<(), SnapshotError> {
self.entries.push(CommitLogEntry {
root_hash,
timestamp,
});
if let Err(error) = self.persist() {
self.entries.pop();
return Err(error);
}
Ok(())
}
pub const fn list(&self) -> &[CommitLogEntry] {
self.entries.as_slice()
}
pub const fn len(&self) -> usize {
self.entries.len()
}
pub const fn is_empty(&self) -> bool {
self.entries.is_empty()
}
fn persist(&self) -> Result<(), SnapshotError> {
if let Some(path) = &self.path {
write_atomic(path, &encode_log(&self.entries)).map_err(CodecError::from)?;
}
Ok(())
}
}
impl Default for CommitLog {
fn default() -> Self {
Self::new()
}
}
fn encode_registry(entries: &[SnapshotEntry]) -> Vec<u8> {
let mut bytes = Vec::new();
bytes.extend_from_slice(REGISTRY_MAGIC);
push_u64(&mut bytes, entries.len() as u64);
for entry in entries {
push_bytes(&mut bytes, entry.name.as_bytes());
bytes.extend_from_slice(entry.root_hash.as_bytes());
push_u64(&mut bytes, entry.timestamp);
}
bytes
}
pub(crate) fn decode_registry(bytes: &[u8]) -> Result<Vec<SnapshotEntry>, SnapshotError> {
let mut reader = Reader::new(bytes);
reader.expect_magic(*REGISTRY_MAGIC)?;
let count = reader.read_usize()?;
let mut entries = Vec::with_capacity(count.min(reader.remaining() / 48));
for _ in 0..count {
let name_bytes = reader.read_bytes()?;
let name = String::from_utf8(name_bytes)
.map_err(|_error| SnapshotError::Corrupt("snapshot name is not valid UTF-8".into()))?;
let root_hash = reader.read_hash()?;
let timestamp = reader.read_u64()?;
entries.push(SnapshotEntry {
name,
root_hash,
timestamp,
});
}
reader.finish()?;
Ok(entries)
}
fn encode_log(entries: &[CommitLogEntry]) -> Vec<u8> {
let mut bytes = Vec::new();
bytes.extend_from_slice(LOG_MAGIC);
push_u64(&mut bytes, entries.len() as u64);
for entry in entries {
bytes.extend_from_slice(entry.root_hash.as_bytes());
push_u64(&mut bytes, entry.timestamp);
}
bytes
}
fn decode_log(bytes: &[u8]) -> Result<Vec<CommitLogEntry>, SnapshotError> {
let mut reader = Reader::new(bytes);
reader.expect_magic(*LOG_MAGIC)?;
let count = reader.read_usize()?;
let mut entries = Vec::with_capacity(count.min(reader.remaining() / 40));
for _ in 0..count {
let root_hash = reader.read_hash()?;
let timestamp = reader.read_u64()?;
entries.push(CommitLogEntry {
root_hash,
timestamp,
});
}
reader.finish()?;
Ok(entries)
}
#[cfg(test)]
#[path = "snapshot_tests.rs"]
mod tests;