use std::path::Path;
use std::sync::{Arc, Mutex};
use super::transaction::Session;
use crate::graph::dir_graph::DirGraph;
use crate::graph::durability;
use crate::graph::storage::recording::resolve_ops;
use crate::graph::storage::GraphRead;
use crate::graph::wal::{DurabilityLevel, Wal, WalFrame};
#[derive(Debug)]
pub(super) struct DurableState {
wal: Wal,
next_lsn: u64,
level: DurabilityLevel,
diverged: bool,
#[cfg(test)]
fail_append: bool,
}
const DIVERGED_MSG: &str = "this durable session was mutated through Session::write / \
Session::transact, which the write-ahead log cannot describe: those mutations are \
captured but never drained into a frame, so the log no longer describes the graph. \
Take a checkpoint (Session::save) to fold them in and start a fresh log, or run \
mutations through Session::begin / Session::commit, which are logged.";
pub(super) const DIRECT_WRITE_REFUSAL: &str =
"a durable session does not support direct writes through Session::write / \
Session::transact: their mutations are captured by the recording backend but \
nothing drains that buffer into a WAL frame, and the next copy-on-write fork \
resets it — the write would apply and then vanish on a crash, with no error. \
Run mutations through Session::begin / Session::commit, which append a frame \
before publishing.";
impl Session {
pub fn open_durable(
graph: Arc<DirGraph>,
checkpoint_path: &str,
level: DurabilityLevel,
) -> Result<Session, String> {
if level.logs() && graph.graph.is_disk() {
return Err(format!(
"durable={} is not supported for storage='disk' (only 'off' is). A disk \
graph commits by publishing an immutable generation, so its durability \
boundary is the generation publish, not a logical write-ahead log: a \
replayed WAL frame and a published generation can each describe the same \
commit, and reconciling them needs a generation-aware log this release \
does not have. Use Session::save checkpoints for disk graphs, or a \
mapped / in-memory graph if you need per-commit crash safety.",
level.name(),
));
}
let mut graph = graph;
let opened = durability::open_log(&mut graph, Path::new(checkpoint_path), level)
.map_err(|e| e.to_string())?;
let Some((wal, next_lsn)) = opened else {
return Ok(Session::from_arc(graph));
};
Ok(Session::with_durable(
graph,
DurableState {
wal,
next_lsn,
level,
diverged: false,
#[cfg(test)]
fail_append: false,
},
))
}
pub fn durability(&self) -> Option<DurabilityLevel> {
self.durable
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_ref()
.map(|ds| ds.level)
}
pub fn sync(&self) -> Result<(), String> {
let _graph = self.graph.lock().unwrap_or_else(|p| p.into_inner());
let mut slot = self.durable.lock().unwrap_or_else(|p| p.into_inner());
let Some(ds) = slot.as_mut() else {
return Err(
"sync() needs a session opened with a write-ahead log (Session::open_durable \
at level 'full' or 'normal'). This session has none, so there is nothing to \
flush and no power-safe point to take — call Session::save to write a \
checkpoint instead."
.to_string(),
);
};
if ds.diverged {
return Err(DIVERGED_MSG.to_string());
}
if ds.level == DurabilityLevel::Full {
return Ok(());
}
ds.wal.sync().map_err(|e| e.to_string())
}
pub(super) fn log_working_commit(&self, working: &mut DirGraph) -> Result<(), String> {
let mut slot = self.durable.lock().unwrap_or_else(|p| p.into_inner());
let Some(ds) = slot.as_mut() else {
return Ok(());
};
if ds.diverged {
return Err(DIVERGED_MSG.to_string());
}
let raw = match working.graph.recording_mut() {
Some(rg) => rg.take_ops(),
None => {
return Err(
"durable commit found no recording backend: this session's write-capture \
layer was replaced, so the commit cannot be logged and has not been \
published."
.to_string(),
)
}
};
if raw.is_empty() {
return Ok(());
}
let ops = resolve_ops(&raw, &working.graph, &working.interner, |idx| {
working.secondary_label_names(idx)
});
#[cfg(test)]
if ds.fail_append {
return Err("injected WAL append failure".to_string());
}
let lsn = ds.next_lsn;
ds.wal
.append(&WalFrame { lsn, ops })
.map_err(|e| e.to_string())?;
ds.next_lsn = lsn + 1;
Ok(())
}
pub(super) fn checkpoint_prologue(&self, graph: &mut Arc<DirGraph>) -> Result<bool, String> {
let mut slot = self.durable.lock().unwrap_or_else(|p| p.into_inner());
let Some(ds) = slot.as_mut() else {
return Ok(false);
};
durability::checkpoint_prologue(&mut ds.wal, ds.next_lsn, Arc::make_mut(graph))
.map_err(|e| e.to_string())?;
Ok(true)
}
pub(super) fn checkpoint_epilogue(&self, graph: &mut Arc<DirGraph>) -> Result<(), String> {
let mut slot = self.durable.lock().unwrap_or_else(|p| p.into_inner());
let Some(ds) = slot.as_mut() else {
return Ok(());
};
durability::checkpoint_epilogue(&mut ds.wal, Arc::make_mut(graph))
.map_err(|e| e.to_string())?;
ds.diverged = false;
Ok(())
}
pub(super) fn mark_diverged(&self) {
if let Some(ds) = self
.durable
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_mut()
{
ds.diverged = true;
}
}
pub fn check_direct_write_allowed(&self) -> Result<(), String> {
if self
.durable
.lock()
.unwrap_or_else(|p| p.into_inner())
.is_some()
{
return Err(DIRECT_WRITE_REFUSAL.to_string());
}
Ok(())
}
#[cfg(test)]
pub(super) fn set_fail_append(&self, fail: bool) {
if let Some(ds) = self
.durable
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_mut()
{
ds.fail_append = fail;
}
}
#[cfg(test)]
pub(super) fn next_lsn(&self) -> Option<u64> {
self.durable
.lock()
.unwrap_or_else(|p| p.into_inner())
.as_ref()
.map(|ds| ds.next_lsn)
}
fn with_durable(graph: Arc<DirGraph>, state: DurableState) -> Session {
Session {
graph: Mutex::new(graph),
durable: Mutex::new(Some(state)),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::graph::session::execute::{execute_mut, execute_read, ExecuteOptions};
use crate::graph::session::CommitOutcome;
use crate::graph::storage::mode::{new_dir_graph_in_mode, StorageMode};
use crate::graph::wal::{recover, wal_path};
use std::collections::HashMap;
fn params() -> HashMap<String, crate::datatypes::Value> {
HashMap::new()
}
fn commit_query(session: &Session, query: &str) -> CommitOutcome {
let params = params();
let opts = ExecuteOptions::eager(¶ms);
let mut tx = session.begin();
execute_mut(tx.working_mut().unwrap(), query, &opts).unwrap();
session.commit(tx, true)
}
fn refusal(result: Result<Session, String>) -> String {
match result {
Err(message) => message,
Ok(_) => panic!("expected a refusal, got an open session"),
}
}
fn count_nodes(graph: &DirGraph) -> usize {
let params = params();
let opts = ExecuteOptions::eager(¶ms);
execute_read(graph, "MATCH (n:N) RETURN n.id AS id", &opts)
.unwrap()
.result
.rows
.len()
}
fn fresh(path: &std::path::Path) -> Session {
Session::open_durable(
Arc::new(DirGraph::new()),
&path.to_string_lossy(),
DurabilityLevel::Full,
)
.unwrap()
}
fn reopen(path: &std::path::Path, level: DurabilityLevel) -> Result<Session, String> {
let p = path.to_string_lossy().into_owned();
let graph = if path.exists() {
crate::graph::io::file::load_file(&p).unwrap()
} else {
Arc::new(DirGraph::new())
};
Session::open_durable(graph, &p, level)
}
#[test]
fn committed_writes_replay_after_a_crash_shaped_reopen() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = fresh(&path);
assert!(matches!(
commit_query(&session, "CREATE (:N {id: 1})"),
CommitOutcome::Committed { .. }
));
assert!(matches!(
commit_query(&session, "CREATE (:N {id: 2})"),
CommitOutcome::Committed { .. }
));
assert_eq!(count_nodes(&session.snapshot()), 2);
drop(session);
assert!(!path.exists(), "the crash-shaped run wrote no checkpoint");
let recovered = reopen(&path, DurabilityLevel::Full).unwrap();
assert_eq!(
count_nodes(&recovered.snapshot()),
2,
"both committed writes must come back out of the log"
);
assert_eq!(recovered.next_lsn(), Some(3));
match recovered.snapshot().graph.recording() {
Some(rg) => assert_eq!(
rg.ops_len(),
0,
"replay ran before the capture wrap, so it captured nothing"
),
None => panic!("a durable session must be wrapped for capture"),
}
}
#[test]
fn a_failed_append_blocks_the_publish() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
let version_before = session.version();
let snapshot_before = session.snapshot();
session.set_fail_append(true);
let outcome = commit_query(&session, "CREATE (:N {id: 2})");
match outcome {
CommitOutcome::DurabilityFailed { ref error } => {
assert!(error.contains("injected"), "unexpected error: {error}")
}
other => panic!("expected DurabilityFailed, got {other:?}"),
}
assert_eq!(
session.version(),
version_before,
"a commit that could not be logged must not bump the version"
);
assert!(
Arc::ptr_eq(&snapshot_before, &session.snapshot()),
"a commit that could not be logged must not swap the live Arc"
);
assert_eq!(
count_nodes(&session.snapshot()),
1,
"the unlogged write must not be visible"
);
session.set_fail_append(false);
assert!(matches!(
commit_query(&session, "CREATE (:N {id: 2})"),
CommitOutcome::Committed { .. }
));
assert_eq!(count_nodes(&session.snapshot()), 2);
}
#[test]
fn save_truncates_the_log_and_stamps_the_replay_gate() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let wpath = wal_path(&path);
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
commit_query(&session, "CREATE (:N {id: 2})");
let logged_len = std::fs::metadata(&wpath).unwrap().len();
assert!(logged_len > 0);
session.save(&path.to_string_lossy(), true).unwrap();
let after_len = std::fs::metadata(&wpath).unwrap().len();
assert!(
after_len < logged_len,
"checkpoint must truncate the log: {logged_len} -> {after_len}"
);
assert!(recover(&wpath).unwrap().is_empty());
let reloaded = crate::graph::io::file::load_file(&path.to_string_lossy()).unwrap();
assert_eq!(reloaded.checkpoint_lsn, 2, "next_lsn(3) - 1");
assert_eq!(count_nodes(&reloaded), 2);
assert_eq!(session.next_lsn(), Some(3));
commit_query(&session, "CREATE (:N {id: 3})");
drop(session);
let frames = recover(&wpath).unwrap();
assert_eq!(frames.len(), 1, "only the post-checkpoint commit is logged");
assert_eq!(frames[0].lsn, 3);
let recovered = reopen(&path, DurabilityLevel::Full).unwrap();
assert_eq!(
count_nodes(&recovered.snapshot()),
3,
"checkpointed 2 + replayed 1"
);
}
#[test]
fn a_stale_prefix_below_the_checkpoint_is_not_replayed() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let wpath = wal_path(&path);
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
let logged = std::fs::read(&wpath).unwrap();
session.save(&path.to_string_lossy(), true).unwrap();
drop(session);
std::fs::write(&wpath, &logged).unwrap();
let recovered = reopen(&path, DurabilityLevel::Full).unwrap();
assert_eq!(count_nodes(&recovered.snapshot()), 1);
assert_eq!(
recovered.next_lsn(),
Some(2),
"the stale frame must not advance the log-sequence counter"
);
}
#[test]
fn level_off_refuses_an_unreplayed_sidecar_but_accepts_a_stale_one() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let wpath = wal_path(&path);
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
drop(session);
let err = refusal(reopen(&path, DurabilityLevel::Off));
assert!(err.contains("'full' or 'normal'"), "message was: {err}");
assert!(err.contains("move the sidecar aside"), "message was: {err}");
let session = reopen(&path, DurabilityLevel::Full).unwrap();
let logged = std::fs::read(&wpath).unwrap();
session.save(&path.to_string_lossy(), true).unwrap();
drop(session);
std::fs::write(&wpath, &logged).unwrap();
let plain = reopen(&path, DurabilityLevel::Off).unwrap();
assert!(plain.durability().is_none());
assert_eq!(count_nodes(&plain.snapshot()), 1);
}
#[test]
fn an_empty_sidecar_is_not_a_refusal_at_level_off() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = reopen(&path, DurabilityLevel::Off).unwrap();
assert!(session.durability().is_none());
}
#[test]
fn disk_graphs_are_refused_at_every_logging_level() {
let dir = tempfile::tempdir().unwrap();
let graph_dir = dir.path().join("disk-graph");
for level in [DurabilityLevel::Full, DurabilityLevel::Normal] {
let g = new_dir_graph_in_mode(StorageMode::Disk, Some(&graph_dir)).unwrap();
let err = refusal(Session::open_durable(
Arc::new(g),
&dir.path().join("g.kgl").to_string_lossy(),
level,
));
assert!(err.contains("storage='disk'"), "message was: {err}");
assert!(err.contains(level.name()), "message was: {err}");
}
let g = new_dir_graph_in_mode(StorageMode::Disk, Some(&graph_dir)).unwrap();
let session = Session::open_durable(
Arc::new(g),
&dir.path().join("g.kgl").to_string_lossy(),
DurabilityLevel::Off,
)
.unwrap();
assert!(session.durability().is_none());
}
#[test]
fn a_second_durable_owner_over_one_graph_is_refused() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
let err = refusal(Session::open_durable(
session.snapshot(),
&path.to_string_lossy(),
DurabilityLevel::Full,
));
assert!(err.contains("already wrapped"), "message was: {err}");
}
#[test]
fn mapped_graphs_are_durable_too() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let g = new_dir_graph_in_mode(StorageMode::Mapped, None).unwrap();
let session = Session::open_durable(
Arc::new(g),
&path.to_string_lossy(),
DurabilityLevel::Normal,
)
.unwrap();
assert_eq!(session.durability(), Some(DurabilityLevel::Normal));
commit_query(&session, "CREATE (:N {id: 1})");
session.sync().unwrap();
drop(session);
let recovered = reopen(&path, DurabilityLevel::Normal).unwrap();
assert_eq!(count_nodes(&recovered.snapshot()), 1);
}
#[test]
fn open_durable_reports_an_unopenable_log() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
std::fs::create_dir(wal_path(&path)).unwrap();
let err = refusal(Session::open_durable(
Arc::new(DirGraph::new()),
&path.to_string_lossy(),
DurabilityLevel::Full,
));
assert!(
err.contains("write-ahead log"),
"IO failure must name the log: {err}"
);
}
#[test]
fn direct_writes_are_refused_and_latch_the_session() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = fresh(&path);
commit_query(&session, "CREATE (:N {id: 1})");
let err = session
.check_direct_write_allowed()
.expect_err("expected a refusal");
assert!(err.contains("does not support direct writes"));
{
let mut graph = session.write();
graph.bump_version();
}
match commit_query(&session, "CREATE (:N {id: 2})") {
CommitOutcome::DurabilityFailed { ref error } => {
assert!(error.contains("Session::write"), "message was: {error}")
}
other => panic!("expected DurabilityFailed, got {other:?}"),
}
assert!(session
.sync()
.expect_err("expected a refusal")
.contains("Session::write"));
session.save(&path.to_string_lossy(), true).unwrap();
assert!(session.sync().is_ok());
assert!(matches!(
commit_query(&session, "CREATE (:N {id: 2})"),
CommitOutcome::Committed { .. }
));
}
#[test]
fn transact_latches_a_durable_session_too() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let session = fresh(&path);
session
.transact(|working| {
working.bump_version();
Ok::<_, &'static str>(())
})
.unwrap();
assert!(matches!(
commit_query(&session, "CREATE (:N {id: 1})"),
CommitOutcome::DurabilityFailed { .. }
));
}
#[test]
fn a_commit_with_no_writes_appends_no_frame() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("g.kgl");
let wpath = wal_path(&path);
let session = fresh(&path);
let header_len = std::fs::metadata(&wpath).unwrap().len();
assert!(matches!(
session.commit(session.begin(), true),
CommitOutcome::NoWritesNoOp
));
let mut tx = session.begin();
tx.working_mut().unwrap();
assert!(matches!(
session.commit(tx, true),
CommitOutcome::Committed { .. }
));
assert_eq!(
std::fs::metadata(&wpath).unwrap().len(),
header_len,
"an empty commit must not append a frame"
);
assert_eq!(
session.next_lsn(),
Some(1),
"an empty commit must not consume an LSN"
);
commit_query(&session, "CREATE (:N {id: 1})");
assert!(std::fs::metadata(&wpath).unwrap().len() > header_len);
assert_eq!(session.next_lsn(), Some(2));
}
#[test]
fn sync_is_an_error_on_a_non_durable_session() {
let session = Session::new(DirGraph::new());
assert!(session.durability().is_none());
let err = session.sync().expect_err("expected a refusal");
assert!(err.contains("write-ahead log"), "message was: {err}");
}
}