use std::any::Any;
use std::io::Read;
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use super::backup::BackupMode;
use super::files::{DirPath, GraphFile};
use super::maintenance::BackupReport;
use super::snapshot_dir::{DirSnapshotReader, write_graph_snapshot};
use super::wal_files::{WalWriter, open_segment_reader};
use crate::error::GraphError;
use crate::graph::hooks::{CommitHook, HookError, JournalHook, JournalSlot, Mark};
use crate::graph::{
ChangeSet, DefinitionKind, Graph, GraphStorage, PersistentStorage, RollbackReason,
StorageCapabilities, TraversalGraph,
};
use crate::persist::error::{PersistError, StoreFailure};
use crate::persist::journal::Journal;
use crate::persist::recover::{Recovered, Recovery, RecoveryReport, RecoveryTarget, recover_into};
use crate::persist::snapshot::{ReadOptions, WriteOptions, load_snapshot, write_snapshot};
pub(crate) trait StoreGraph: Send + Sync + 'static {
fn into_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
fn read(&self) -> Box<dyn LiveRead + '_>;
fn write(&self) -> Box<dyn LiveWrite + '_>;
}
pub(crate) trait LiveRead {
fn last_commit_seq(&self) -> u64;
fn last_committed_at(&self) -> i64;
fn graph_id(&self) -> u128;
fn definition_kinds(&self) -> Vec<DefinitionKind>;
fn critical_definition(&self) -> Option<String>;
fn write_snapshot_live(&self, tmp: &DirPath, write: &WriteOptions) -> Result<(), PersistError>;
fn backup_from_memory(
&self,
store: &DirPath,
graph_file: &GraphFile,
write: &WriteOptions,
target: &DirPath,
mode: BackupMode,
) -> Result<BackupReport, PersistError>;
}
pub(crate) trait LiveWrite: LiveRead {
fn mark(&mut self, name: &str) -> Result<Mark, GraphError>;
fn reload(&mut self, plan: &LoadPlan) -> Result<RecoveryReport, PersistError>;
fn clear(&mut self);
fn attach_journal(&mut self, journal: JournalSlot) -> Result<(), PersistError>;
fn register_guard_hook(&mut self, hook: Box<dyn CommitHook>) -> Result<(), PersistError>;
}
impl<S: PersistentStorage + Default> StoreGraph for Graph<S> {
fn into_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
self
}
fn read(&self) -> Box<dyn LiveRead + '_> {
Box::new(Live(Graph::read(self)))
}
fn write(&self) -> Box<dyn LiveWrite + '_> {
Box::new(Live(Graph::write(self)))
}
}
struct Live<G>(G);
impl<G> LiveRead for Live<G>
where
G: Deref,
G::Target: PersistentStorage,
{
fn last_commit_seq(&self) -> u64 {
self.0.last_commit_seq()
}
fn last_committed_at(&self) -> i64 {
self.0.last_committed_at()
}
fn graph_id(&self) -> u128 {
self.0.graph_id()
}
fn definition_kinds(&self) -> Vec<DefinitionKind> {
self.0.definitions().map(|d| d.kind()).collect()
}
fn critical_definition(&self) -> Option<String> {
crate::persist::definitions::critical_unknown(self.0.definitions())
}
fn write_snapshot_live(&self, tmp: &DirPath, write: &WriteOptions) -> Result<(), PersistError> {
write_graph_snapshot(tmp, &*self.0, write).map(|_| ())
}
fn backup_from_memory(
&self,
store: &DirPath,
graph_file: &GraphFile,
write: &WriteOptions,
target: &DirPath,
mode: BackupMode,
) -> Result<BackupReport, PersistError> {
super::from_memory::backup_from_memory(store, &*self.0, graph_file, write, target, mode)
}
}
impl<G> LiveWrite for Live<G>
where
G: DerefMut,
G::Target: PersistentStorage,
{
fn mark(&mut self, name: &str) -> Result<Mark, GraphError> {
self.0.mark(name)
}
fn reload(&mut self, plan: &LoadPlan) -> Result<RecoveryReport, PersistError> {
let graph = &mut *self.0;
graph.clear();
plan.run_into(graph)
}
fn clear(&mut self) {
self.0.clear();
}
fn attach_journal(&mut self, journal: JournalSlot) -> Result<(), PersistError> {
require_journal_capabilities(self.0.capabilities())?;
Ok(self.0.attach_journal(journal)?)
}
fn register_guard_hook(&mut self, hook: Box<dyn CommitHook>) -> Result<(), PersistError> {
Ok(self.0.register_commit_hook(hook)?)
}
}
pub(crate) fn require_journal_capabilities(
capabilities: StorageCapabilities,
) -> Result<(), PersistError> {
let missing = match (capabilities.atomic, capabilities.change_capture) {
(true, true) => return Ok(()),
(false, false) => "atomic and change_capture",
(false, true) => "atomic",
(true, false) => "change_capture",
};
Err(PersistError::store(
"attach the store's journal",
StoreFailure::MissingCapability,
format!(
"the graph's storage does not claim the capability {missing} \
(StorageCapabilities): a journal cannot honour a veto without it"
),
))
}
pub(crate) trait GraphMaker {
fn check(&self) -> Result<(), PersistError> {
Ok(())
}
fn empty(&self) -> Arc<dyn StoreGraph>;
fn salvaged(&self, graph: TraversalGraph) -> Result<Arc<dyn StoreGraph>, PersistError>;
}
pub(crate) struct TraversalGraphs;
impl GraphMaker for TraversalGraphs {
fn empty(&self) -> Arc<dyn StoreGraph> {
Arc::new(Graph::new_persistent(TraversalGraph::new()))
}
fn salvaged(&self, graph: TraversalGraph) -> Result<Arc<dyn StoreGraph>, PersistError> {
Ok(Arc::new(Graph::new_persistent(graph)))
}
}
pub(crate) struct Storages<F>(pub(crate) F);
impl<S, F> GraphMaker for Storages<F>
where
S: PersistentStorage + Default,
F: Fn() -> S,
{
fn check(&self) -> Result<(), PersistError> {
require_journal_capabilities((self.0)().capabilities())
}
fn empty(&self) -> Arc<dyn StoreGraph> {
Arc::new(Graph::new_persistent((self.0)()))
}
fn salvaged(&self, graph: TraversalGraph) -> Result<Arc<dyn StoreGraph>, PersistError> {
let mut bytes = Vec::new();
write_snapshot(
&graph,
&mut bytes,
&WriteOptions::new().with_codec(crate::persist::Codec::None),
)?;
drop(graph);
let mut storage = (self.0)();
load_snapshot(
bytes.as_slice(),
&ReadOptions::new().with_max_value_depth(4096),
&mut storage,
)?;
Ok(Arc::new(Graph::new_persistent(storage)))
}
}
pub(crate) struct LoadPlan {
pub(crate) snapshot: DirPath,
pub(crate) journals: Vec<(DirPath, Option<u64>)>,
pub(crate) target: RecoveryTarget,
pub(crate) read: ReadOptions,
pub(crate) chain: Vec<(u128, u64)>,
}
impl LoadPlan {
fn journal_readers(&self) -> Result<Vec<Box<dyn Read + Send>>, PersistError> {
self.journals
.iter()
.map(|(path, limit)| open_segment_reader(path, *limit))
.collect()
}
pub(crate) fn run(&self) -> Result<Recovered, PersistError> {
self.run_in(TraversalGraph::new())
}
pub(crate) fn run_in<S: PersistentStorage>(
&self,
empty: S,
) -> Result<Recovered<S>, PersistError> {
let mut recovery = Recovery::new_in(empty)
.snapshot(DirSnapshotReader::open(&self.snapshot)?)
.target(self.target.clone())
.options(self.read.clone())
.chain(self.chain.clone());
for reader in self.journal_readers()? {
recovery = recovery.journal(reader);
}
recovery.run()
}
pub(crate) fn run_into<S: PersistentStorage + ?Sized>(
&self,
graph: &mut S,
) -> Result<RecoveryReport, PersistError> {
let journals = self
.journal_readers()?
.into_iter()
.map(|reader| reader as Box<dyn Read>)
.collect();
recover_into(
graph,
DirSnapshotReader::open(&self.snapshot)?,
journals,
self.target.clone(),
&self.read,
Some(&self.chain),
)
}
}
#[derive(Clone, Default)]
pub(crate) struct JournalCell(Arc<parking_lot::Mutex<Option<Journal<WalWriter>>>>);
impl JournalCell {
pub(crate) fn slot(&self) -> JournalSlot {
JournalSlot::new(Box::new(CellHook(self.clone())))
}
pub(crate) fn set(&self, journal: Option<Journal<WalWriter>>) {
let old = std::mem::replace(&mut *self.0.lock(), journal);
drop(old);
}
pub(crate) fn sync_now(&self) -> Option<Result<(), HookError>> {
self.0.lock().as_mut().map(|journal| journal.sync_now())
}
pub(crate) fn record_checkpoint(
&self,
commit_seq: u64,
time: i64,
) -> Option<Result<(), HookError>> {
self.0
.lock()
.as_mut()
.map(|journal| journal.record_checkpoint(commit_seq, time))
}
}
struct CellHook(JournalCell);
impl CellHook {
fn detached() -> HookError {
Box::new(PersistError::store(
"journal a commit",
StoreFailure::Closed,
"the store's journal is detached (a rollback or restore failed, or the store went \
read-only): reopen the store",
))
}
}
impl CommitHook for CellHook {
fn name(&self) -> &str {
"journal"
}
fn before_commit(&mut self, changes: &ChangeSet) -> Result<(), HookError> {
match self.0.0.lock().as_mut() {
Some(journal) => journal.before_commit(changes),
None => Err(Self::detached()),
}
}
fn after_commit(&mut self, changes: &ChangeSet) {
if let Some(journal) = self.0.0.lock().as_mut() {
journal.after_commit(changes);
}
}
fn after_rollback(&mut self, reason: &RollbackReason) {
if let Some(journal) = self.0.0.lock().as_mut() {
journal.after_rollback(reason);
}
}
}
impl JournalHook for CellHook {
fn record_mark(&mut self, mark: &Mark) -> Result<(), HookError> {
match self.0.0.lock().as_mut() {
Some(journal) => journal.record_mark(mark),
None => Err(Self::detached()),
}
}
fn record_checkpoint(&mut self, commit_seq: u64, time: i64) -> Result<(), HookError> {
self.0.record_checkpoint(commit_seq, time).unwrap_or(Ok(()))
}
fn sync_now(&mut self) -> Result<(), HookError> {
self.0.sync_now().unwrap_or(Ok(()))
}
}
#[cfg(test)]
mod tests {
use compact_str::CompactString;
use super::*;
use crate::graph::definition::Definition;
use crate::graph::schema::GraphSchema;
use crate::graph::{EdgeHandle, VertexHandle};
use crate::value::{ElementProperty, PropertyMap};
#[derive(Default)]
struct ClaimsLess {
inner: TraversalGraph,
claims: StorageCapabilities,
}
impl GraphStorage for ClaimsLess {
type VertexId = VertexHandle;
type EdgeId = EdgeHandle;
fn capabilities(&self) -> StorageCapabilities {
self.claims
}
fn vertex_count(&self) -> usize {
self.inner.vertex_count()
}
fn edge_count(&self) -> usize {
self.inner.edge_count()
}
fn vertices(&self) -> impl Iterator<Item = VertexHandle> + '_ {
self.inner.vertices()
}
fn edges(&self) -> impl Iterator<Item = EdgeHandle> + '_ {
self.inner.edges()
}
fn get_vertex_by_id(&self, id: &str) -> Option<VertexHandle> {
GraphStorage::get_vertex_by_id(&self.inner, id)
}
fn get_edge_by_id(&self, id: &str) -> Option<EdgeHandle> {
GraphStorage::get_edge_by_id(&self.inner, id)
}
fn out_edges(&self, node: VertexHandle) -> impl Iterator<Item = EdgeHandle> + '_ {
self.inner.out_edges(node)
}
fn in_edges(&self, node: VertexHandle) -> impl Iterator<Item = EdgeHandle> + '_ {
self.inner.in_edges(node)
}
fn vertex_id(&self, node: VertexHandle) -> Option<&str> {
self.inner.vertex_id(node)
}
fn edge_id(&self, edge: EdgeHandle) -> Option<&str> {
self.inner.edge_id(edge)
}
fn vertex_label(&self, node: VertexHandle) -> Option<&str> {
self.inner.vertex_label(node)
}
fn edge_label(&self, edge: EdgeHandle) -> Option<&str> {
self.inner.edge_label(edge)
}
fn vertex_labels(&self, node: VertexHandle) -> &[CompactString] {
self.inner.vertex_labels(node)
}
fn vertex_properties(&self, node: VertexHandle) -> Option<&PropertyMap> {
self.inner.vertex_properties(node)
}
fn edge_properties(&self, edge: EdgeHandle) -> Option<&PropertyMap> {
self.inner.edge_properties(edge)
}
fn edge_endpoints(&self, edge: EdgeHandle) -> Option<(VertexHandle, VertexHandle)> {
self.inner.edge_endpoints(edge)
}
fn add_vertex(
&mut self,
id: Option<CompactString>,
labels: Vec<CompactString>,
properties: PropertyMap,
) -> Result<VertexHandle, GraphError> {
self.inner.add_vertex(id, labels, properties)
}
fn add_vertex_label(
&mut self,
node: VertexHandle,
label: CompactString,
) -> Result<bool, GraphError> {
self.inner.add_vertex_label(node, label)
}
fn remove_vertex_label(
&mut self,
node: VertexHandle,
label: &str,
) -> Result<bool, GraphError> {
self.inner.remove_vertex_label(node, label)
}
fn add_edge(
&mut self,
id: Option<CompactString>,
label: Option<CompactString>,
from: VertexHandle,
to: VertexHandle,
properties: PropertyMap,
) -> Result<EdgeHandle, GraphError> {
self.inner.add_edge(id, label, from, to, properties)
}
fn set_vertex_property(
&mut self,
node: VertexHandle,
key: CompactString,
value: ElementProperty,
) -> Result<(), GraphError> {
self.inner.set_vertex_property(node, key, value)
}
fn set_edge_property(
&mut self,
edge: EdgeHandle,
key: CompactString,
value: ElementProperty,
) -> Result<(), GraphError> {
self.inner.set_edge_property(edge, key, value)
}
fn drop_vertex(&mut self, node: VertexHandle) -> Result<(), GraphError> {
self.inner.drop_vertex(node)
}
fn drop_vertex_property(
&mut self,
prop: &str,
node: VertexHandle,
) -> Result<(), GraphError> {
self.inner.drop_vertex_property(prop, node)
}
fn drop_edge(&mut self, edge: EdgeHandle) -> Result<(), GraphError> {
self.inner.drop_edge(edge)
}
fn drop_edge_property(&mut self, prop: &str, edge: EdgeHandle) -> Result<(), GraphError> {
self.inner.drop_edge_property(prop, edge)
}
}
impl PersistentStorage for ClaimsLess {
fn graph_id(&self) -> u128 {
self.inner.graph_id()
}
fn set_graph_id(&mut self, graph_id: u128) {
self.inner.set_graph_id(graph_id)
}
fn last_committed_at(&self) -> i64 {
PersistentStorage::last_committed_at(&self.inner)
}
fn id_sequences(&self) -> (u64, u64) {
self.inner.id_sequences()
}
fn raise_position(&mut self, commit_seq: u64, committed_at: i64) {
self.inner.raise_position(commit_seq, committed_at)
}
fn load_vertex(
&mut self,
id: CompactString,
labels: Vec<CompactString>,
properties: PropertyMap,
) -> Result<VertexHandle, GraphError> {
self.inner.load_vertex(id, labels, properties)
}
fn load_edge(
&mut self,
id: CompactString,
label: Option<CompactString>,
from_id: &str,
to_id: &str,
properties: PropertyMap,
) -> Result<EdgeHandle, GraphError> {
self.inner.load_edge(id, label, from_id, to_id, properties)
}
fn load_schema(&mut self, schema: GraphSchema) -> Result<(), GraphError> {
self.inner.load_schema(schema)
}
fn load_definition(&mut self, definition: Definition) -> Result<(), GraphError> {
self.inner.load_definition(definition)
}
fn clear(&mut self) {
self.inner.clear()
}
fn attach_journal(&mut self, journal: JournalSlot) -> Result<(), GraphError> {
self.inner.attach_journal(journal)
}
}
fn claims(atomic: bool, change_capture: bool) -> StorageCapabilities {
let mut claims = TraversalGraph::new().capabilities();
claims.atomic = atomic;
claims.change_capture = change_capture;
claims
}
#[test]
fn a_journal_needs_an_atomic_storage_with_change_capture() {
for (atomic, change_capture, missing) in [
(false, false, Some("atomic and change_capture")),
(false, true, Some("atomic")),
(true, false, Some("change_capture")),
(true, true, None),
] {
let graph: Arc<dyn StoreGraph> = Arc::new(Graph::new_persistent(ClaimsLess {
inner: TraversalGraph::new(),
claims: claims(atomic, change_capture),
}));
let result = graph.write().attach_journal(JournalCell::default().slot());
let has_journal = {
let typed = graph
.clone()
.into_any()
.downcast::<Graph<ClaimsLess>>()
.unwrap();
typed.read().inner.undo.hooks.has_journal()
};
match missing {
None => {
result.unwrap();
assert!(has_journal);
}
Some(missing) => {
let err = result.unwrap_err();
assert!(
matches!(
err,
PersistError::Store {
cause: StoreFailure::MissingCapability,
..
}
),
"{err:?}"
);
let text = err.to_string();
assert!(
text.contains(&format!("the capability {missing} ")),
"{text}"
);
assert!(!has_journal);
}
}
}
let typed: Arc<dyn StoreGraph> = Arc::new(Graph::new_persistent(TraversalGraph::new()));
typed
.write()
.attach_journal(JournalCell::default().slot())
.unwrap();
}
#[test]
fn store_create_in_and_open_in_refuse_a_storage_without_the_capabilities() {
use crate::persist::{MemDir, Store, StoreDir, StoreOptions};
fn missing_capability(err: PersistError) {
assert!(
matches!(
err,
PersistError::Store {
cause: StoreFailure::MissingCapability,
..
}
),
"{err:?}"
);
}
let dir = MemDir::named("memory:claims");
let err = Store::create_in(
&dir,
ClaimsLess {
inner: TraversalGraph::tinkerpop_modern(),
claims: claims(true, false),
},
StoreOptions::new(),
)
.unwrap_err();
assert!(err.to_string().contains("change_capture"), "{err}");
missing_capability(err);
assert!(dir.list("").unwrap_or_default().is_empty());
let store = Store::create_with(
&dir,
TraversalGraph::tinkerpop_modern(),
StoreOptions::new(),
)
.unwrap();
store.close().unwrap();
let factory = || ClaimsLess {
inner: TraversalGraph::new(),
claims: claims(false, true),
};
missing_capability(Store::open_in(&dir, StoreOptions::new(), factory).unwrap_err());
let backup = MemDir::named("memory:claims-backup");
Store::open(&dir, StoreOptions::new())
.unwrap()
.backup(&backup)
.unwrap();
missing_capability(
Store::open_backup_in(&backup, StoreOptions::new(), factory).unwrap_err(),
);
}
#[test]
fn an_empty_journal_cell_refuses_commits_and_marks() {
let graph = Graph::new_persistent(TraversalGraph::new());
let cell = JournalCell::default();
graph.write().attach_journal(cell.slot()).unwrap();
let err = graph
.write()
.traversal_mut()
.add_v("person")
.to_list()
.unwrap_err();
assert!(err.to_string().contains("journal is detached"), "{err}");
assert_eq!(graph.read().vertex_count(), 0);
assert!(graph.write().mark("m").is_err());
assert!(cell.sync_now().is_none());
assert!(cell.record_checkpoint(1, 1).is_none());
}
}