use std::path::{Path, PathBuf};
use std::time::SystemTime;
use linkrs_core::types::{CommitLsn, IndexGeneration};
use serde::{Deserialize, Serialize};
use crate::sqlite_outbox::OutboxSnapshot;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CheckpointManifest {
pub format_version: u32,
pub checkpoint_id: u64,
pub timestamp: u64,
pub storage_lsn: CommitLsn,
pub outbox_lsn: CommitLsn,
pub outbox_enabled: bool,
pub safe_lsn: CommitLsn,
pub storage_snapshot: StorageSnapshotRef,
pub outbox_snapshot: Option<OutboxSnapshotRef>,
pub index_manifests: Vec<IndexManifestRef>,
#[serde(default)]
pub max_commit_timestamp: u64,
#[serde(default)]
pub snapshot_timestamp: Option<u64>,
#[serde(default)]
pub schema_catalog_version: u64,
#[serde(default)]
pub sequence_version: u64,
pub manifest_checksum: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct StorageSnapshotRef {
pub path: PathBuf,
pub size_bytes: u64,
pub checksum: u32,
pub files: Vec<StorageFileRef>,
pub checkpoint_seq: u64,
pub vertex_count: u64,
pub edge_count: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct StorageFileRef {
pub path: PathBuf,
pub size_bytes: u64,
pub checksum: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct OutboxSnapshotRef {
pub path: PathBuf,
pub size_bytes: u64,
pub checksum: u32,
pub materialized_lsn: CommitLsn,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct IndexManifestRef {
pub space_id: u64,
pub index_id: u64,
pub generation: IndexGeneration,
pub path: PathBuf,
pub size_bytes: u64,
pub checksum: u32,
}
impl CheckpointManifest {
pub const CURRENT_FORMAT_VERSION: u32 = 5;
pub fn storage_snapshot_from_directory(
path: impl AsRef<Path>,
checkpoint_seq: u64,
vertex_count: u64,
edge_count: u64,
) -> Result<StorageSnapshotRef, String> {
let path = path.as_ref();
let files = collect_storage_files(path)?;
let size_bytes = files.iter().map(|file| file.size_bytes).sum();
let checksum = storage_files_checksum(&files)?;
Ok(StorageSnapshotRef {
path: path.to_path_buf(),
size_bytes,
checksum,
files,
checkpoint_seq,
vertex_count,
edge_count,
})
}
pub fn new(
checkpoint_id: u64,
storage_lsn: CommitLsn,
storage_snapshot: StorageSnapshotRef,
outbox_snapshot: Option<OutboxSnapshotRef>,
index_manifests: Vec<IndexManifestRef>,
) -> Result<Self, String> {
Self::new_with_outbox_state(
checkpoint_id,
storage_lsn,
storage_snapshot,
outbox_snapshot,
index_manifests,
false,
)
}
pub fn new_with_outbox(
checkpoint_id: u64,
storage_lsn: CommitLsn,
storage_snapshot: StorageSnapshotRef,
outbox_snapshot: Option<OutboxSnapshotRef>,
index_manifests: Vec<IndexManifestRef>,
) -> Result<Self, String> {
Self::new_with_outbox_state(
checkpoint_id,
storage_lsn,
storage_snapshot,
outbox_snapshot,
index_manifests,
true,
)
}
fn new_with_outbox_state(
checkpoint_id: u64,
storage_lsn: CommitLsn,
storage_snapshot: StorageSnapshotRef,
outbox_snapshot: Option<OutboxSnapshotRef>,
index_manifests: Vec<IndexManifestRef>,
outbox_enabled: bool,
) -> Result<Self, String> {
let outbox_lsn = outbox_snapshot
.as_ref()
.map(|s| s.materialized_lsn)
.unwrap_or(CommitLsn::ZERO);
let safe_lsn = if outbox_enabled && outbox_snapshot.is_none() {
CommitLsn::ZERO
} else if outbox_snapshot.is_some() && outbox_lsn < storage_lsn {
outbox_lsn
} else {
storage_lsn
};
let mut manifest = Self {
format_version: Self::CURRENT_FORMAT_VERSION,
checkpoint_id,
timestamp: SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
storage_lsn,
outbox_lsn,
outbox_enabled,
safe_lsn,
storage_snapshot,
outbox_snapshot,
index_manifests,
max_commit_timestamp: storage_lsn.get(),
snapshot_timestamp: None,
schema_catalog_version: 0,
sequence_version: 0,
manifest_checksum: 0,
};
manifest.manifest_checksum = manifest.compute_checksum()?;
Ok(manifest)
}
pub fn with_snapshot_timestamp(mut self, snapshot_ts: u64) -> Result<Self, String> {
self.max_commit_timestamp = snapshot_ts;
self.snapshot_timestamp = Some(snapshot_ts);
self.manifest_checksum = self.compute_checksum()?;
Ok(self)
}
fn compute_checksum(&self) -> Result<u32, String> {
let mut clone = self.clone();
clone.manifest_checksum = 0;
let bytes = postcard::to_allocvec(&clone)
.map_err(|error| format!("Failed to serialize checkpoint manifest: {error}"))?;
Ok(crc32fast::hash(&bytes))
}
pub fn verify_checksum(&self) -> Result<bool, String> {
let expected = self.manifest_checksum;
let mut clone = self.clone();
clone.manifest_checksum = 0;
let bytes = postcard::to_allocvec(&clone)
.map_err(|error| format!("Failed to serialize checkpoint manifest: {error}"))?;
let computed = crc32fast::hash(&bytes);
Ok(computed == expected)
}
pub fn validate(&self) -> Result<(), String> {
if self.format_version != Self::CURRENT_FORMAT_VERSION
&& self.format_version != Self::CURRENT_FORMAT_VERSION - 1
{
return Err(format!(
"Unsupported manifest format version: {}",
self.format_version
));
}
if !self.verify_checksum()? {
return Err("Manifest checksum verification failed".to_string());
}
let outbox_lsn = self
.outbox_snapshot
.as_ref()
.map(|s| s.materialized_lsn)
.unwrap_or(CommitLsn::ZERO);
let expected_safe = if self.outbox_enabled && self.outbox_snapshot.is_none() {
CommitLsn::ZERO
} else if self.outbox_snapshot.is_some() && outbox_lsn < self.storage_lsn {
outbox_lsn
} else {
self.storage_lsn
};
if self.safe_lsn != expected_safe {
return Err(format!(
"Invalid safe_lsn: expected {}, got {}",
expected_safe, self.safe_lsn
));
}
if self.storage_snapshot.checkpoint_seq != self.checkpoint_id {
return Err(format!(
"Storage snapshot checkpoint sequence {} does not match manifest {}",
self.storage_snapshot.checkpoint_seq, self.checkpoint_id
));
}
if self.outbox_snapshot.is_some() && self.outbox_lsn > self.storage_lsn {
return Err(format!(
"Outbox snapshot LSN {} is ahead of storage LSN {}",
self.outbox_lsn, self.storage_lsn
));
}
if self.outbox_snapshot.is_none() && self.outbox_lsn != CommitLsn::ZERO {
return Err(
"Outbox LSN is non-zero but outbox snapshot reference is missing".to_string(),
);
}
verify_storage_reference(&self.storage_snapshot)?;
if let Some(snapshot) = &self.outbox_snapshot {
verify_file_reference(
&snapshot.path,
snapshot.size_bytes,
snapshot.checksum,
"outbox snapshot",
)?;
}
for manifest in &self.index_manifests {
verify_file_reference(
&manifest.path,
manifest.size_bytes,
manifest.checksum,
"index manifest",
)?;
}
Ok(())
}
pub fn load(path: impl AsRef<Path>) -> Result<Self, String> {
let bytes = std::fs::read(path.as_ref()).map_err(|error| error.to_string())?;
let manifest: Self = postcard::from_bytes(&bytes).map_err(|error| error.to_string())?;
manifest.validate()?;
Ok(manifest)
}
pub fn save_atomic(&self, destination: impl AsRef<Path>) -> Result<(), String> {
let destination = destination.as_ref();
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
}
let temporary = destination.with_extension("tmp");
let bytes = postcard::to_allocvec(self).map_err(|error| error.to_string())?;
std::fs::write(&temporary, &bytes).map_err(|error| error.to_string())?;
let file = std::fs::OpenOptions::new()
.read(true)
.open(&temporary)
.map_err(|error| error.to_string())?;
file.sync_all().map_err(|error| error.to_string())?;
drop(file);
if destination.exists() {
std::fs::remove_file(destination).map_err(|error| error.to_string())?;
}
std::fs::rename(&temporary, destination).map_err(|error| error.to_string())?;
if let Some(parent) = destination.parent() {
std::fs::File::open(parent)
.and_then(|dir| dir.sync_all())
.map_err(|error| error.to_string())?;
}
Ok(())
}
pub fn outbox_snapshot_from(outbox: &OutboxSnapshot) -> OutboxSnapshotRef {
OutboxSnapshotRef {
path: outbox.path.clone(),
size_bytes: outbox.size_bytes,
checksum: outbox.checksum,
materialized_lsn: outbox.materialized_lsn,
}
}
}
pub struct CheckpointManifestManager {
manifest_dir: PathBuf,
}
impl CheckpointManifestManager {
pub fn new(manifest_dir: impl AsRef<Path>) -> Self {
Self {
manifest_dir: manifest_dir.as_ref().to_path_buf(),
}
}
pub fn init(&self) -> Result<(), String> {
std::fs::create_dir_all(&self.manifest_dir).map_err(|error| error.to_string())?;
std::fs::File::open(&self.manifest_dir)
.and_then(|dir| dir.sync_all())
.map_err(|error| error.to_string())?;
Ok(())
}
pub fn manifest_path(&self, checkpoint_id: u64) -> PathBuf {
self.manifest_dir
.join(format!("manifest_{:020}.postcard", checkpoint_id))
}
pub fn publish(&self, manifest: &CheckpointManifest) -> Result<(), String> {
let path = self.manifest_path(manifest.checkpoint_id);
manifest.save_atomic(&path)?;
Ok(())
}
pub fn load_latest(&self) -> Result<Option<CheckpointManifest>, String> {
if !self.manifest_dir.exists() {
return Ok(None);
}
let mut manifests: Vec<(u64, PathBuf)> = std::fs::read_dir(&self.manifest_dir)
.map_err(|error| error.to_string())?
.filter_map(|entry| entry.ok())
.filter_map(|entry| {
let path = entry.path();
if !path.is_file() {
return None;
}
let name = path.file_name()?.to_str()?;
let id = name
.strip_prefix("manifest_")?
.strip_suffix(".postcard")?
.parse::<u64>()
.ok()?;
Some((id, path))
})
.collect();
manifests.sort_by_key(|(id, _)| std::cmp::Reverse(*id));
for (_, path) in manifests {
match CheckpointManifest::load(&path) {
Ok(manifest) => return Ok(Some(manifest)),
Err(error) => {
log::warn!(
"Skipping invalid checkpoint manifest {}: {}",
path.display(),
error
);
}
}
}
Ok(None)
}
pub fn load(&self, checkpoint_id: u64) -> Result<Option<CheckpointManifest>, String> {
let path = self.manifest_path(checkpoint_id);
if !path.exists() {
return Ok(None);
}
CheckpointManifest::load(path).map(Some)
}
pub fn list_manifests(&self) -> Result<Vec<u64>, String> {
if !self.manifest_dir.exists() {
return Ok(Vec::new());
}
let mut ids: Vec<u64> = std::fs::read_dir(&self.manifest_dir)
.map_err(|error| error.to_string())?
.filter_map(|entry| entry.ok())
.filter_map(|entry| {
let path = entry.path();
let name = path.file_name()?.to_str()?;
name.strip_prefix("manifest_")?
.strip_suffix(".postcard")?
.parse::<u64>()
.ok()
})
.collect();
ids.sort();
Ok(ids)
}
pub fn latest_safe_lsn(&self) -> Result<CommitLsn, String> {
self.load_latest()?
.map(|m| m.safe_lsn)
.ok_or_else(|| "No published manifest found".to_string())
}
pub fn cleanup_old(&self, keep_newer_than: u64) -> Result<usize, String> {
let mut removed = 0;
for id in self.list_manifests()? {
if id < keep_newer_than {
let path = self.manifest_path(id);
std::fs::remove_file(path).map_err(|error| error.to_string())?;
removed += 1;
}
}
if removed > 0 {
std::fs::File::open(&self.manifest_dir)
.and_then(|dir| dir.sync_all())
.map_err(|error| error.to_string())?;
}
Ok(removed)
}
}
fn verify_file_reference(
path: &Path,
expected_size: u64,
expected_checksum: u32,
kind: &str,
) -> Result<(), String> {
let bytes = std::fs::read(path)
.map_err(|error| format!("Failed to read {kind} {}: {error}", path.display()))?;
if bytes.len() as u64 != expected_size {
return Err(format!("{kind} size mismatch: {}", path.display()));
}
if crc32fast::hash(&bytes) != expected_checksum {
return Err(format!("{kind} checksum mismatch: {}", path.display()));
}
Ok(())
}
fn collect_storage_files(root: &Path) -> Result<Vec<StorageFileRef>, String> {
fn visit(root: &Path, directory: &Path, files: &mut Vec<StorageFileRef>) -> Result<(), String> {
for entry in std::fs::read_dir(directory).map_err(|error| error.to_string())? {
let path = entry.map_err(|error| error.to_string())?.path();
if path.is_dir() {
visit(root, &path, files)?;
continue;
}
if !path.is_file() {
continue;
}
if path
.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| name == "edge_wal.bin")
{
continue;
}
let bytes = std::fs::read(&path).map_err(|error| error.to_string())?;
let relative = path
.strip_prefix(root)
.map_err(|error| error.to_string())?
.to_path_buf();
files.push(StorageFileRef {
path: relative,
size_bytes: bytes.len() as u64,
checksum: crc32fast::hash(&bytes),
});
}
Ok(())
}
if !root.is_dir() {
return Err(format!(
"Storage snapshot directory does not exist: {}",
root.display()
));
}
let mut files = Vec::new();
visit(root, root, &mut files)?;
files.sort_by(|left, right| left.path.cmp(&right.path));
Ok(files)
}
fn storage_files_checksum(files: &[StorageFileRef]) -> Result<u32, String> {
let bytes = postcard::to_allocvec(files).map_err(|error| error.to_string())?;
Ok(crc32fast::hash(&bytes))
}
fn verify_storage_reference(reference: &StorageSnapshotRef) -> Result<(), String> {
let actual = collect_storage_files(&reference.path)?;
if actual != reference.files {
return Err(format!(
"Storage snapshot file list mismatch: {}",
reference.path.display()
));
}
let size_bytes: u64 = actual.iter().map(|file| file.size_bytes).sum();
if size_bytes != reference.size_bytes {
return Err(format!(
"Storage snapshot size mismatch: {}",
reference.path.display()
));
}
let checksum = storage_files_checksum(&actual)?;
if checksum != reference.checksum {
return Err(format!(
"Storage snapshot checksum mismatch: {}",
reference.path.display()
));
}
for file in &actual {
if file.path.is_absolute()
|| file
.path
.components()
.any(|component| component == std::path::Component::ParentDir)
{
return Err(format!(
"Invalid storage snapshot file path: {}",
file.path.display()
));
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn create_test_storage_snapshot_ref(temp_dir: &Path) -> StorageSnapshotRef {
let snapshot_path = temp_dir.join("storage_snapshot");
std::fs::create_dir_all(&snapshot_path).unwrap();
std::fs::write(snapshot_path.join("checkpoint.meta"), b"checkpoint").unwrap();
CheckpointManifest::storage_snapshot_from_directory(&snapshot_path, 1, 100, 50).unwrap()
}
fn create_test_outbox_snapshot_ref(temp_dir: &Path) -> OutboxSnapshotRef {
let snapshot_path = temp_dir.join("outbox_snapshot.sqlite");
std::fs::write(&snapshot_path, b"test").unwrap();
OutboxSnapshotRef {
path: snapshot_path,
size_bytes: 4,
checksum: crc32fast::hash(b"test"),
materialized_lsn: CommitLsn::new(100),
}
}
#[test]
fn test_manifest_creation_and_validation() {
let temp_dir = TempDir::new().unwrap();
let storage_ref = create_test_storage_snapshot_ref(temp_dir.path());
let outbox_ref = create_test_outbox_snapshot_ref(temp_dir.path());
let manifest = CheckpointManifest::new(
1,
CommitLsn::new(100),
storage_ref,
Some(outbox_ref),
Vec::new(),
)
.expect("checkpoint manifest should be created");
assert!(manifest.validate().is_ok());
assert_eq!(manifest.safe_lsn, CommitLsn::new(100));
assert_eq!(manifest.storage_lsn, CommitLsn::new(100));
assert_eq!(manifest.outbox_lsn, CommitLsn::new(100));
}
#[test]
fn test_safe_lsn_is_minimum() {
let temp_dir = TempDir::new().unwrap();
let storage_ref = create_test_storage_snapshot_ref(temp_dir.path());
let manifest1 = CheckpointManifest::new(
1,
CommitLsn::new(50),
storage_ref.clone(),
Some(OutboxSnapshotRef {
path: temp_dir.path().join("outbox.sqlite"),
size_bytes: 4,
checksum: 0,
materialized_lsn: CommitLsn::new(50),
}),
Vec::new(),
)
.expect("checkpoint manifest should be created");
assert_eq!(manifest1.safe_lsn, CommitLsn::new(50));
let manifest2 = CheckpointManifest::new(
2,
CommitLsn::new(100),
storage_ref,
Some(OutboxSnapshotRef {
path: temp_dir.path().join("outbox2.sqlite"),
size_bytes: 4,
checksum: 0,
materialized_lsn: CommitLsn::new(20),
}),
Vec::new(),
)
.expect("checkpoint manifest should be created");
assert_eq!(manifest2.safe_lsn, CommitLsn::new(20));
}
#[test]
fn enabled_outbox_without_snapshot_keeps_wal_reclaim_boundary_at_zero() {
let temp_dir = TempDir::new().unwrap();
let manifest = CheckpointManifest::new_with_outbox(
1,
CommitLsn::new(100),
create_test_storage_snapshot_ref(temp_dir.path()),
None,
Vec::new(),
)
.expect("checkpoint manifest should be created");
assert_eq!(manifest.safe_lsn, CommitLsn::ZERO);
assert!(manifest.validate().is_ok());
}
#[test]
fn test_manifest_save_and_load() {
let temp_dir = TempDir::new().unwrap();
let storage_ref = create_test_storage_snapshot_ref(temp_dir.path());
let outbox_ref = create_test_outbox_snapshot_ref(temp_dir.path());
let manifest = CheckpointManifest::new(
1,
CommitLsn::new(100),
storage_ref,
Some(outbox_ref),
Vec::new(),
)
.expect("checkpoint manifest should be created");
let manifest_path = temp_dir.path().join("test_manifest.postcard");
manifest.save_atomic(&manifest_path).unwrap();
let loaded = CheckpointManifest::load(&manifest_path).unwrap();
assert_eq!(loaded.checkpoint_id, manifest.checkpoint_id);
assert_eq!(loaded.safe_lsn, manifest.safe_lsn);
}
fn create_storage_snapshot_ref(temp_dir: &Path, checkpoint_seq: u64) -> StorageSnapshotRef {
let snapshot_path = temp_dir.join(format!("storage_snapshot_{}", checkpoint_seq));
std::fs::create_dir_all(&snapshot_path).unwrap();
std::fs::write(snapshot_path.join("checkpoint.meta"), b"checkpoint").unwrap();
CheckpointManifest::storage_snapshot_from_directory(&snapshot_path, checkpoint_seq, 100, 50)
.unwrap()
}
#[test]
fn test_manifest_manager() {
let temp_dir = TempDir::new().unwrap();
let manager = CheckpointManifestManager::new(temp_dir.path());
manager.init().unwrap();
let storage_ref1 = create_storage_snapshot_ref(temp_dir.path(), 1);
let storage_ref2 = create_storage_snapshot_ref(temp_dir.path(), 2);
let outbox_ref = create_test_outbox_snapshot_ref(temp_dir.path());
let manifest1 = CheckpointManifest::new(
1,
CommitLsn::new(100),
storage_ref1,
Some(outbox_ref.clone()),
Vec::new(),
)
.expect("checkpoint manifest should be created");
let manifest2 = CheckpointManifest::new(
2,
CommitLsn::new(100),
storage_ref2,
Some(outbox_ref),
Vec::new(),
)
.expect("checkpoint manifest should be created");
manager.publish(&manifest1).unwrap();
manager.publish(&manifest2).unwrap();
let latest = manager.load_latest().unwrap().unwrap();
assert_eq!(latest.checkpoint_id, 2);
assert_eq!(latest.safe_lsn, CommitLsn::new(100));
let safe_lsn = manager.latest_safe_lsn().unwrap();
assert_eq!(safe_lsn, CommitLsn::new(100));
let manifests = manager.list_manifests().unwrap();
assert_eq!(manifests, vec![1, 2]);
}
#[test]
fn test_checksum_verification() {
let temp_dir = TempDir::new().unwrap();
let storage_ref = create_test_storage_snapshot_ref(temp_dir.path());
let mut manifest =
CheckpointManifest::new(1, CommitLsn::new(100), storage_ref, None, Vec::new())
.expect("checkpoint manifest should be created");
assert!(manifest.verify_checksum().expect("checksum should compute"));
manifest.checkpoint_id = 999;
assert!(!manifest.verify_checksum().expect("checksum should compute"));
}
#[test]
fn test_snapshot_timestamp_fix_and_roundtrip() {
let temp_dir = TempDir::new().unwrap();
let storage_ref = create_test_storage_snapshot_ref(temp_dir.path());
let legacy = CheckpointManifest::new(
1,
CommitLsn::new(100),
storage_ref.clone(),
None,
Vec::new(),
)
.expect("checkpoint manifest should be created");
assert_eq!(legacy.snapshot_timestamp, None);
let stamped =
CheckpointManifest::new(1, CommitLsn::new(100), storage_ref, None, Vec::new())
.expect("checkpoint manifest should be created")
.with_snapshot_timestamp(40)
.expect("snapshot stamp should apply");
assert_eq!(stamped.max_commit_timestamp, 40);
assert_eq!(stamped.snapshot_timestamp, Some(40));
assert!(stamped.validate().is_ok());
let path = temp_dir.path().join("stamped.postcard");
stamped.save_atomic(&path).unwrap();
let loaded = CheckpointManifest::load(&path).unwrap();
assert_eq!(loaded.snapshot_timestamp, Some(40));
assert_eq!(loaded.max_commit_timestamp, 40);
}
}