use super::*;
use crate::datatypes::Value;
use crate::graph::cdc::{self, CdcEnrichment};
use crate::graph::session::execute::{execute_mut, ExecuteOptions};
use crate::graph::session::Session;
use crate::graph::storage::mode::{new_dir_graph_in_mode, StorageMode};
use crate::graph::storage::GraphRead;
use crate::graph::wal::{DurabilityLevel, WAL_FORMAT_VERSION};
use std::collections::HashMap;
use std::sync::Arc;
fn run(graph: &mut DirGraph, query: &str) {
let params = HashMap::new();
let opts = ExecuteOptions::eager(¶ms);
execute_mut(graph, query, &opts).unwrap_or_else(|e| panic!("query failed: {query}: {e}"));
}
fn expect_failure(graph: &mut DirGraph, query: &str) {
let params = HashMap::new();
let opts = ExecuteOptions::eager(¶ms);
if execute_mut(graph, query, &opts).is_ok() {
panic!("expected {query} to fail mid-statement");
}
}
const FAILS_AFTER_WRITING: &str = "CREATE (x:Item {id: 200, name: 'new'}) \
WITH x MATCH (a:Item {id: 1}) SET a.name = 'clobbered' \
WITH a MATCH (m:Item {id: 1}) SET m.qty = duration({months: 2147483648})";
fn commit(graph: &mut DirGraph) {
cdc::drain_at_commit(graph);
}
fn events(graph: &DirGraph) -> Vec<CdcEvent> {
cdc::read(graph, 0, None, &[]).expect("capture must be enabled")
}
fn node_id(event: &CdcEvent) -> Value {
match &event.change {
CdcChange::Node { id, .. } => id.clone(),
other => panic!("expected a node event, got {other:?}"),
}
}
fn node_property(event: &CdcEvent, key: &str) -> Option<Value> {
match &event.change {
CdcChange::Node { after, .. } => after.as_ref().and_then(|state| {
state
.properties
.iter()
.find(|(name, _)| name == key)
.map(|(_, value)| value.clone())
}),
other => panic!("expected a node event, got {other:?}"),
}
}
fn node_before_property(event: &CdcEvent, key: &str) -> Option<Value> {
match &event.change {
CdcChange::Node { before, .. } => before.as_ref().and_then(|state| {
state
.properties
.iter()
.find(|(name, _)| name == key)
.map(|(_, value)| value.clone())
}),
other => panic!("expected a node event, got {other:?}"),
}
}
fn node_after(event: &CdcEvent) -> Option<crate::graph::cdc::NodeState> {
match &event.change {
CdcChange::Node { after, .. } => after.clone(),
other => panic!("expected a node event, got {other:?}"),
}
}
fn node_before(event: &CdcEvent) -> Option<crate::graph::cdc::NodeState> {
match &event.change {
CdcChange::Node { before, .. } => before.clone(),
other => panic!("expected a node event, got {other:?}"),
}
}
fn seeded() -> DirGraph {
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable on a plain in-memory graph");
run(
&mut graph,
"CREATE (:Item {id: 1, name: 'one', qty: 10}), (:Item {id: 2, name: 'two', qty: 20})",
);
run(
&mut graph,
"MATCH (a:Item {id: 1}), (b:Item {id: 2}) CREATE (a)-[:LINKS {weight: 1}]->(b)",
);
commit(&mut graph);
let handle = graph.cdc_log().expect("enabled").clone();
handle
.lock()
.unwrap_or_else(|p| p.into_inner())
.reconfigure(super::DEFAULT_CAPACITY, CdcEnrichment::Off);
let drained = events(&graph).len();
assert!(drained > 0, "the fixture itself must publish");
graph
}
fn cursor(graph: &DirGraph) -> u64 {
cdc::status(graph).expect("enabled").current
}
fn since(graph: &DirGraph, from: u64) -> Vec<CdcEvent> {
cdc::read(graph, from, None, &[]).expect("enabled")
}
#[test]
fn create_update_and_delete_publish_one_event_each() {
let mut graph = seeded();
let from = cursor(&graph);
run(&mut graph, "CREATE (:Item {id: 3, name: 'three'})");
commit(&mut graph);
run(&mut graph, "MATCH (i:Item {id: 3}) SET i.name = 'renamed'");
commit(&mut graph);
run(&mut graph, "MATCH (i:Item {id: 3}) DELETE i");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
published
.iter()
.map(|event| (event.kind, event.change.element()))
.collect::<Vec<_>>(),
vec![
(CdcEventKind::Create, "node"),
(CdcEventKind::Update, "node"),
(CdcEventKind::Delete, "node"),
],
"each committed statement publishes exactly one event for the node it touched"
);
assert_eq!(node_id(&published[0]), Value::Int64(3));
assert_eq!(
node_property(&published[0], "name"),
Some(Value::String("three".into())),
"a create carries the after-state"
);
assert_eq!(
node_property(&published[1], "name"),
Some(Value::String("renamed".into())),
"an update carries the state after the write, not before it"
);
assert!(
matches!(
&published[2].change,
CdcChange::Node { after: None, id, .. } if *id == Value::Int64(3)
),
"a delete carries identity only — v1 keeps no before-image"
);
}
#[test]
fn edge_create_update_and_delete_publish_edge_events() {
let mut graph = seeded();
let from = cursor(&graph);
run(
&mut graph,
"MATCH (a:Item {id: 2}), (b:Item {id: 1}) CREATE (a)-[:LINKS {weight: 5}]->(b)",
);
commit(&mut graph);
run(
&mut graph,
"MATCH (:Item {id: 2})-[r:LINKS]->(:Item {id: 1}) SET r.weight = 9",
);
commit(&mut graph);
run(
&mut graph,
"MATCH (:Item {id: 2})-[r:LINKS]->(:Item {id: 1}) DELETE r",
);
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
published
.iter()
.map(|event| (event.kind, event.change.element()))
.collect::<Vec<_>>(),
vec![
(CdcEventKind::Create, "edge"),
(CdcEventKind::Update, "edge"),
(CdcEventKind::Delete, "edge"),
]
);
match &published[0].change {
CdcChange::Edge {
conn_type,
src_id,
tgt_id,
after,
..
} => {
assert_eq!(conn_type, "LINKS");
assert_eq!(*src_id, Value::Int64(2));
assert_eq!(*tgt_id, Value::Int64(1));
assert_eq!(
after.as_ref().map(|state| state.properties.clone()),
Some(vec![("weight".to_string(), Value::Int64(5))]),
"an edge event names both endpoints logically and carries its properties"
);
}
other => panic!("expected an edge event, got {other:?}"),
}
assert!(matches!(
&published[2].change,
CdcChange::Edge { after: None, .. }
));
}
#[test]
fn create_then_set_in_one_statement_collapses_to_one_create() {
let mut graph = seeded();
let from = cursor(&graph);
run(
&mut graph,
"CREATE (i:Item {id: 7, name: 'seven'}) SET i.name = 'final', i.qty = 3",
);
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
published.len(),
1,
"one entity changed in one commit, so one event: {published:?}"
);
assert_eq!(published[0].kind, CdcEventKind::Create);
assert_eq!(
node_property(&published[0], "name"),
Some(Value::String("final".into())),
"the collapsed event carries the state the commit left behind"
);
}
#[test]
fn repeated_updates_in_one_commit_collapse_to_one_update() {
let mut graph = seeded();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'a'");
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'b'");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(published.len(), 1, "{published:?}");
assert_eq!(published[0].kind, CdcEventKind::Update);
assert_eq!(
node_property(&published[0], "name"),
Some(Value::String("b".into()))
);
}
#[test]
fn secondary_labels_publish_an_update_carrying_the_label_set() {
let mut graph = seeded();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Featured");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(published.len(), 1, "{published:?}");
assert_eq!(published[0].kind, CdcEventKind::Update);
match &published[0].change {
CdcChange::Node { after, .. } => assert_eq!(
after.as_ref().map(|state| state.labels.clone()),
Some(vec!["Featured".to_string()]),
"a label change is an update whose after-state names the labels"
),
other => panic!("expected a node event, got {other:?}"),
}
}
#[test]
fn create_then_delete_in_one_commit_publishes_only_the_delete() {
let mut graph = seeded();
let from = cursor(&graph);
run(
&mut graph,
"CREATE (i:Item {id: 42, name: 'ephemeral'}) WITH i MATCH (d:Item {id: 42}) DELETE d",
);
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
published.iter().map(|event| event.kind).collect::<Vec<_>>(),
vec![CdcEventKind::Delete],
"{published:?}"
);
}
#[test]
fn a_rolled_back_statement_publishes_nothing() {
let mut graph = seeded();
let from = cursor(&graph);
expect_failure(&mut graph, FAILS_AFTER_WRITING);
commit(&mut graph);
assert_eq!(
since(&graph, from),
Vec::new(),
"a rolled-back statement's writes are gone from the graph; publishing them \
would tell a consumer about data that never existed"
);
run(&mut graph, "CREATE (x:Item {id: 201, name: 'new'})");
commit(&mut graph);
assert_eq!(since(&graph, from).len(), 1);
}
#[test]
fn a_failed_statement_does_not_poison_the_commit_around_it() {
let mut graph = seeded();
let from = cursor(&graph);
run(&mut graph, "CREATE (:Item {id: 8, name: 'eight'})");
expect_failure(&mut graph, FAILS_AFTER_WRITING);
run(&mut graph, "CREATE (:Item {id: 9, name: 'nine'})");
commit(&mut graph);
let published = since(&graph, from);
let ids: Vec<Value> = published.iter().map(node_id).collect();
assert_eq!(
ids,
vec![Value::Int64(8), Value::Int64(9)],
"only the two committed creates may appear: {published:?}"
);
}
#[test]
fn a_rolled_back_transaction_publishes_nothing() {
let session = Session::new(seeded());
let from = cursor(&session.snapshot());
let mut tx = session.begin();
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 99, name: 'doomed'})",
);
session.rollback(tx);
let graph = session.snapshot();
assert_eq!(
since(&graph, from),
Vec::new(),
"a transaction that was never committed must not appear in the stream"
);
assert_eq!(
graph.graph.node_count(),
2,
"non-vacuity: the rollback really did discard the write"
);
let mut tx = session.begin();
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 99, name: 'kept'})",
);
session.commit(tx, false);
assert_eq!(
since(&session.snapshot(), from).len(),
1,
"non-vacuity: the committed twin of that transaction does publish"
);
}
#[test]
fn a_commit_taken_over_a_held_reader_lands_once_in_the_shared_log() {
let session = Session::new(seeded());
let reader = session.snapshot();
let from = cursor(&reader);
let mut tx = session.begin();
{
let working = tx.working_mut().expect("writable tx");
run(
working,
"MATCH (i:Item {id: 1}) SET i.name = 'written under a reader'",
);
assert!(
working
.graph
.recording()
.is_some_and(|recording| recording.inner().is_forked()),
"precondition: the write must be taken on a Recording(Forked) backend — \
the composition this arm exists to cover"
);
}
session.commit(tx, false);
let writer = session.snapshot();
assert!(
!Arc::ptr_eq(&reader, &writer),
"precondition: the commit must have published a different graph, or the \
copy-on-write shape under test never happened"
);
assert_eq!(
since(&writer, from).len(),
1,
"exactly one event, not one per holder"
);
assert_eq!(
since(&reader, from).len(),
1,
"and the reader's handle addresses the same log — a fork shares it"
);
assert_eq!(
cdc::status(&reader).map(|status| status.epoch),
cdc::status(&writer).map(|status| status.epoch),
"one epoch across the fork"
);
}
#[test]
fn a_rolled_back_statement_under_a_held_reader_publishes_nothing() {
let session = Session::new(seeded());
let reader = session.snapshot();
let from = cursor(&reader);
let mut tx = session.begin();
{
let working = tx.working_mut().expect("writable tx");
run(working, "CREATE (:Item {id: 50, name: 'kept'})");
assert!(
working
.graph
.recording()
.is_some_and(|recording| recording.inner().is_forked()),
"precondition: Recording(Forked)"
);
expect_failure(working, FAILS_AFTER_WRITING);
}
session.commit(tx, false);
let published = since(&session.snapshot(), from);
assert_eq!(
published.iter().map(node_id).collect::<Vec<_>>(),
vec![Value::Int64(50)],
"only the statement that survived may publish: {published:?}"
);
assert_eq!(
since(&reader, from).len(),
1,
"and the reader addresses the same log"
);
}
#[test]
fn eviction_bounds_the_ring_and_advances_the_earliest_watermark() {
let mut graph = DirGraph::new();
cdc::enable(&mut graph, Some(4), CdcEnrichment::Off).expect("enable");
for id in 0..40 {
run(&mut graph, &format!("CREATE (:Item {{id: {id}}})"));
commit(&mut graph);
let status = cdc::status(&graph).expect("enabled");
assert!(
status.buffered <= 4,
"the ring must stay within its capacity at every step, saw {status:?}"
);
}
let status = cdc::status(&graph).expect("enabled");
assert_eq!(status.buffered, 4);
assert_eq!(status.current, 40, "every commit still consumed a sequence");
assert_eq!(
status.earliest, 37,
"the earliest readable position advances as events are evicted"
);
let retained = events(&graph);
assert_eq!(retained.first().map(|event| event.seq), Some(37));
assert_eq!(
cdc::read(&graph, 0, None, &[]).map(|read| read.len()),
Some(4),
"a cursor older than the watermark reads what survives; B2 turns that \
into a typed 'cursor too old' refusal"
);
}
#[test]
fn a_single_commit_larger_than_the_ring_is_still_bounded() {
let mut graph = DirGraph::new();
cdc::enable(&mut graph, Some(8), CdcEnrichment::Off).expect("enable");
let creates = (0..500)
.map(|id| format!("(:Item {{id: {id}}})"))
.collect::<Vec<_>>()
.join(", ");
run(&mut graph, &format!("CREATE {creates}"));
commit(&mut graph);
let status = cdc::status(&graph).expect("enabled");
assert_eq!(status.buffered, 8, "{status:?}");
assert_eq!(status.current, 500);
assert_eq!(status.earliest, 493);
}
#[test]
fn re_enabling_resizes_in_place_and_keeps_the_epoch() {
let mut graph = DirGraph::new();
let first = cdc::enable(&mut graph, Some(100), CdcEnrichment::Off).expect("enable");
for id in 0..10 {
run(&mut graph, &format!("CREATE (:Item {{id: {id}}})"));
}
commit(&mut graph);
let resized = cdc::enable(&mut graph, Some(3), CdcEnrichment::Off).expect("re-enable");
assert_eq!(
resized.epoch, first.epoch,
"a live consumer's cursors must survive a capacity change"
);
assert_eq!(resized.capacity, 3);
assert_eq!(resized.buffered, 3, "the shrink evicts from the front");
assert_eq!(resized.earliest, 8);
}
#[test]
fn enable_wraps_a_plain_graph_without_claiming_the_write_ahead_log() {
let mut graph = DirGraph::new();
assert!(!graph.graph.is_recording());
let status =
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable on a plain graph");
assert_eq!(status.capacity, super::DEFAULT_CAPACITY);
assert_eq!(status.current, 0);
assert!(
graph.graph.is_recording(),
"enable installs the capture seam the events are derived from"
);
assert!(
!graph.graph.is_wal_owner(),
"…without presenting itself as a durable owner, or a later durable open \
would be refused and the duplicate-id rule would silently tighten"
);
}
#[test]
fn enable_does_not_impose_the_durable_duplicate_id_refusal() {
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable");
run(&mut graph, "CREATE (:Item {id: 1, name: 'first'})");
run(&mut graph, "CREATE (:Item {id: 1, name: 'second'})");
commit(&mut graph);
assert_eq!(graph.graph.node_count(), 2);
}
#[test]
fn disable_drops_the_log_and_stops_publishing() {
let mut graph = seeded();
assert!(cdc::disable(&mut graph), "was enabled");
assert!(!cdc::disable(&mut graph), "already off");
assert!(cdc::status(&graph).is_none());
assert!(cdc::read(&graph, 0, None, &[]).is_none());
assert!(
!graph.graph.is_recording(),
"disable takes the capture layer off a graph no write-ahead log owns — \
leaving it would keep charging the write path for a stream nobody reads"
);
run(&mut graph, "CREATE (:Item {id: 5})");
commit(&mut graph);
assert!(cdc::status(&graph).is_none());
let restarted = cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("re-enable");
assert_eq!(
restarted.current, 0,
"a restart is a new epoch and a new sequence: nothing from before it is \
addressable"
);
}
#[test]
fn disable_leaves_a_durable_graphs_capture_layer_alone() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = tmp.path().join("graph.kgl");
let session = Session::open_durable(
Arc::new(DirGraph::new()),
&path.to_string_lossy(),
DurabilityLevel::Full,
)
.expect("durable open");
let mut tx = session.begin();
{
let working = tx.working_mut().expect("writable tx");
cdc::enable(working, None, CdcEnrichment::Off).expect("enable");
assert!(cdc::disable(working), "was enabled");
assert!(
working.graph.is_wal_owner(),
"disabling the stream must not disarm the log"
);
}
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 1})",
);
session.commit(tx, false);
let graph = session.snapshot();
assert!(graph.graph.is_wal_owner());
assert!(cdc::status(&graph).is_none());
let wal = crate::graph::wal::wal_path(&path);
assert!(
std::fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0) > 0,
"and the commit after it must still reach the log"
);
}
#[test]
fn disk_mode_refuses_enable() {
let dir = tempfile::tempdir().expect("tempdir");
let mut graph = new_dir_graph_in_mode(StorageMode::Disk, Some(dir.path())).expect("disk graph");
let error = cdc::enable(&mut graph, None, CdcEnrichment::Off).expect_err("disk must refuse");
let message = error.to_string();
assert!(
message.contains("storage='disk'") && message.contains("generation"),
"the refusal must name the mode and why its change boundary is different: {message}"
);
assert!(!graph.cdc_enabled());
}
#[test]
fn mapped_mode_serves_the_stream() {
let mut graph = new_dir_graph_in_mode(StorageMode::Mapped, None).expect("mapped graph");
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("mapped must serve");
run(&mut graph, "CREATE (:Item {id: 1, name: 'one'})");
commit(&mut graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'renamed'");
commit(&mut graph);
run(&mut graph, "MATCH (i:Item {id: 1}) DELETE i");
commit(&mut graph);
let published = events(&graph);
assert_eq!(
published.iter().map(|event| event.kind).collect::<Vec<_>>(),
vec![
CdcEventKind::Create,
CdcEventKind::Update,
CdcEventKind::Delete
],
"{published:?}"
);
assert_eq!(
node_property(&published[1], "name"),
Some(Value::String("renamed".into()))
);
}
#[test]
fn a_bulk_load_publishes_one_create_per_row() {
use crate::datatypes::DataFrame;
use crate::graph::mutation::maintain::add_nodes;
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable");
let columns = vec!["id".to_string(), "name".to_string()];
let rows: Vec<Vec<Value>> = (0..25)
.map(|i| vec![Value::Int64(i), Value::String(format!("row-{i}"))])
.collect();
let frame = DataFrame::from_cypher_rows(columns, rows).expect("dataframe");
add_nodes(
&mut graph,
frame,
"Person".to_string(),
"id".to_string(),
Some("name".to_string()),
None,
)
.expect("bulk load");
commit(&mut graph);
let published = events(&graph);
assert_eq!(published.len(), 25, "one event per row, not per write");
assert!(
published
.iter()
.all(|event| event.kind == CdcEventKind::Create),
"a loaded row is a created node: {published:?}"
);
assert_eq!(node_id(&published[0]), Value::Int64(0));
}
#[test]
fn capacity_zero_and_oversize_are_refused() {
let mut graph = DirGraph::new();
assert!(cdc::enable(&mut graph, Some(0), CdcEnrichment::Off).is_err());
assert!(cdc::enable(
&mut graph,
Some(super::MAX_CAPACITY + 1),
CdcEnrichment::Off
)
.is_err());
assert!(!graph.cdc_enabled(), "a refused enable installs nothing");
assert!(cdc::enable(&mut graph, Some(super::MAX_CAPACITY), CdcEnrichment::Off).is_ok());
}
#[test]
fn independent_copy_re_mints_the_epoch_and_starts_empty() {
let graph = seeded();
let copy = graph.independent_copy();
let original = cdc::status(&graph).expect("enabled");
let copied = cdc::status(©).expect("an independent copy keeps capture on");
assert_ne!(
copied.epoch, original.epoch,
"an independent lineage needs an identity of its own, like graph_id — a \
cursor from the original must be refusable, not resolvable here"
);
assert_eq!(copied.buffered, 0, "the copy has published nothing");
assert_eq!(copied.current, 0);
assert_eq!(copied.capacity, original.capacity);
}
#[test]
fn a_clone_shares_the_log_but_an_independent_copy_does_not() {
let mut graph = seeded();
let from = cursor(&graph);
let mut copy = graph.independent_copy();
run(&mut copy, "CREATE (:Item {id: 300})");
commit(&mut copy);
assert_eq!(
since(&graph, from),
Vec::new(),
"a write to the copy must not appear in the original's stream"
);
assert_eq!(since(©, 0).len(), 1);
let clone = graph.clone();
run(&mut graph, "CREATE (:Item {id: 301})");
commit(&mut graph);
assert_eq!(
since(&clone, from).len(),
1,
"a plain clone shares the log — that is what makes a copy-on-write commit \
visible exactly once"
);
}
#[test]
fn a_cdc_enabled_graph_can_still_be_opened_durably() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = tmp.path().join("graph.kgl");
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable");
let session = Session::open_durable(
Arc::new(graph),
&path.to_string_lossy(),
DurabilityLevel::Full,
)
.expect("a CDC-enabled graph is not a second durable owner");
assert_eq!(session.durability(), Some(DurabilityLevel::Full));
assert!(session.snapshot().graph.is_wal_owner());
}
#[test]
fn a_durable_commit_publishes_to_the_log_and_the_stream() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = tmp.path().join("graph.kgl");
let session = Session::open_durable(
Arc::new(DirGraph::new()),
&path.to_string_lossy(),
DurabilityLevel::Full,
)
.expect("durable open");
let mut tx = session.begin();
cdc::enable(
tx.working_mut().expect("writable tx"),
Some(16),
CdcEnrichment::Off,
)
.expect("enable");
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 1, name: 'durable'})",
);
session.commit(tx, false);
let graph = session.snapshot();
assert_eq!(
events(&graph)
.iter()
.map(|event| event.kind)
.collect::<Vec<_>>(),
vec![CdcEventKind::Create],
"the durable drain must publish, not consume, the capture buffer"
);
let wal = crate::graph::wal::wal_path(&path);
let size = std::fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0);
assert!(
size > 0,
"and the write-ahead log must still hold the frame for that commit"
);
}
#[test]
fn a_durable_commit_that_could_not_be_logged_publishes_nothing() {
use crate::graph::session::CommitOutcome;
let tmp = tempfile::tempdir().expect("tempdir");
let path = tmp.path().join("graph.kgl");
let session = Session::open_durable(
Arc::new(DirGraph::new()),
&path.to_string_lossy(),
DurabilityLevel::Full,
)
.expect("durable open");
let mut tx = session.begin();
cdc::enable(
tx.working_mut().expect("writable tx"),
None,
CdcEnrichment::Off,
)
.expect("enable");
session.commit(tx, false);
let from = cursor(&session.snapshot());
session.set_fail_append(true);
let mut tx = session.begin();
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 1, name: 'unlogged'})",
);
assert!(matches!(
session.commit(tx, false),
CommitOutcome::DurabilityFailed { .. }
));
assert_eq!(
since(&session.snapshot(), from),
Vec::new(),
"a commit the caller was told failed must not appear in the stream"
);
session.set_fail_append(false);
let mut tx = session.begin();
run(
tx.working_mut().expect("writable tx"),
"CREATE (:Item {id: 1, name: 'logged'})",
);
assert!(matches!(
session.commit(tx, false),
CommitOutcome::Committed { .. }
));
assert_eq!(since(&session.snapshot(), from).len(), 1);
}
#[test]
fn cdc_does_not_move_the_wal_format_version() {
assert_eq!(
WAL_FORMAT_VERSION, 3,
"CDC derives create-vs-update from an in-memory capture marker; if this \
constant moved, a `MutationOp` gained a field it did not need"
);
}
#[test]
fn a_columnar_set_captures_the_value_it_overwrote() {
let mut graph = columnar_mapped_graph();
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 99");
commit(&mut graph);
let published = events(&graph);
let update = published.last().expect("the SET must publish");
assert_eq!(update.kind, CdcEventKind::Update);
assert_eq!(
node_before_property(update, "qty"),
Some(Value::Int64(10)),
"before must be the value the write destroyed, not the one it wrote"
);
assert_eq!(
node_property(update, "qty"),
Some(Value::Int64(99)),
"and after must still be the new value"
);
assert_eq!(
node_before(update).map(|state| state.title),
Some(Value::String("one".into())),
"the image is the whole entity, not just the touched cell"
);
}
#[test]
fn columnar_write_is_the_path_under_test() {
use crate::graph::storage::GraphRead;
let graph = columnar_mapped_graph();
let idx = graph
.graph
.node_indices()
.find(|idx| {
graph
.graph
.node_view(*idx)
.is_some_and(|view| view.id().as_ref() == &Value::Int64(1))
})
.expect("the fixture's node");
assert!(
graph
.graph
.node_weight(idx)
.and_then(|node| node.properties.columnar_row_id())
.is_some(),
"the fixture must leave the node columnar, or `set_via_column_master` \
bails and this file's columnar arm tests nothing"
);
}
fn columnar_mapped_graph() -> DirGraph {
use crate::datatypes::DataFrame;
use crate::graph::mutation::maintain::add_nodes;
let mut graph = new_dir_graph_in_mode(StorageMode::Mapped, None).expect("mapped graph");
let columns = vec!["id".to_string(), "name".to_string(), "qty".to_string()];
let rows: Vec<Vec<Value>> = vec![
vec![
Value::Int64(1),
Value::String("one".to_string()),
Value::Int64(10),
],
vec![
Value::Int64(2),
Value::String("two".to_string()),
Value::Int64(20),
],
];
let frame = DataFrame::from_cypher_rows(columns, rows).expect("dataframe");
add_nodes(
&mut graph,
frame,
"Item".to_string(),
"id".to_string(),
Some("name".to_string()),
None,
)
.expect("bulk load");
cdc::enable(&mut graph, None, CdcEnrichment::Full).expect("mapped must serve full capture");
commit(&mut graph);
graph
}
fn seeded_full() -> DirGraph {
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Full).expect("enable full");
run(
&mut graph,
"CREATE (:Item {id: 1, name: 'one', qty: 10}), (:Item {id: 2, name: 'two', qty: 20})",
);
run(
&mut graph,
"MATCH (a:Item {id: 1}), (b:Item {id: 2}) CREATE (a)-[:LINKS {weight: 1}]->(b)",
);
commit(&mut graph);
graph
}
#[test]
fn enrichment_off_captures_no_before_image() {
let mut graph = seeded();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'renamed'");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(published.len(), 1);
assert_eq!(published[0].kind, CdcEventKind::Update);
assert_eq!(
node_before(&published[0]),
None,
"an off log must not pay for, or report, a before-image"
);
assert!(
node_property(&published[0], "name").is_some(),
"and the after half is unaffected"
);
}
#[test]
fn a_create_has_no_before_image_even_under_full() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "CREATE (:Item {id: 3, name: 'three'})");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(published.len(), 1);
assert_eq!(published[0].kind, CdcEventKind::Create);
assert_eq!(node_before(&published[0]), None);
}
#[test]
fn an_update_reports_the_state_on_both_sides_of_the_commit() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 99");
commit(&mut graph);
let update = &since(&graph, from)[0];
assert_eq!(node_before_property(update, "qty"), Some(Value::Int64(10)));
assert_eq!(node_property(update, "qty"), Some(Value::Int64(99)));
assert_eq!(
node_before_property(update, "name"),
Some(Value::String("one".into())),
"an untouched property is still part of the image"
);
}
#[test]
fn a_delete_reports_the_state_it_destroyed() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 2}) DETACH DELETE i");
commit(&mut graph);
let published = since(&graph, from);
let node_delete = published
.iter()
.find(|event| {
event.kind == CdcEventKind::Delete && matches!(event.change, CdcChange::Node { .. })
})
.expect("the node delete must publish");
assert_eq!(
node_before_property(node_delete, "name"),
Some(Value::String("two".into()))
);
assert_eq!(
node_before_property(node_delete, "qty"),
Some(Value::Int64(20))
);
let edge_delete = published
.iter()
.find(|event| {
event.kind == CdcEventKind::Delete && matches!(event.change, CdcChange::Edge { .. })
})
.expect("the detached edge must publish too");
let CdcChange::Edge { before, .. } = &edge_delete.change else {
unreachable!("filtered above");
};
assert_eq!(
before.as_ref().and_then(|state| state
.properties
.iter()
.find(|(k, _)| k == "weight")
.cloned()),
Some(("weight".to_string(), Value::Int64(1))),
"a relationship's before-image carries its properties: {before:?}"
);
}
#[test]
fn a_write_then_delete_in_one_commit_still_reports_the_before_image() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 55");
run(&mut graph, "MATCH (i:Item {id: 1}) DETACH DELETE i");
commit(&mut graph);
let published = since(&graph, from);
let node_delete = published
.iter()
.find(|event| {
event.kind == CdcEventKind::Delete && matches!(event.change, CdcChange::Node { .. })
})
.expect("the delete must publish");
assert_eq!(
node_before_property(node_delete, "qty"),
Some(Value::Int64(10)),
"the image is the state at the start of the commit, not after the SET"
);
}
#[test]
fn before_is_the_commit_start_state_across_several_statements() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 11");
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 12");
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 13");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(published.len(), 1, "three writes, one entity, one event");
assert_eq!(
node_before_property(&published[0], "qty"),
Some(Value::Int64(10)),
"the first touch wins: 10 is what the commit found, 11 and 12 are interior"
);
assert_eq!(node_property(&published[0], "qty"), Some(Value::Int64(13)));
}
#[test]
fn a_transaction_reports_the_pre_transaction_state() {
let session = Session::new(seeded_full());
let from = cursor(&session.snapshot());
let mut tx = session.begin();
{
let working = tx.working_mut().expect("writable tx");
run(working, "MATCH (i:Item {id: 1}) SET i.qty = 41");
run(working, "MATCH (i:Item {id: 1}) SET i.qty = 42");
}
session.commit(tx, false);
let published = since(&session.snapshot(), from);
assert_eq!(published.len(), 1, "{published:?}");
assert_eq!(
node_before_property(&published[0], "qty"),
Some(Value::Int64(10)),
"the transaction's before is the state it opened on"
);
assert_eq!(node_property(&published[0], "qty"), Some(Value::Int64(42)));
}
#[test]
fn a_label_write_reports_the_label_set_it_replaced() {
let mut graph = seeded_full();
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Featured");
commit(&mut graph);
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Archived");
commit(&mut graph);
let update = &since(&graph, from)[0];
assert_eq!(
node_before(update).map(|state| state.labels),
Some(vec!["Featured".to_string()]),
"before is the set the commit found, not the one it left"
);
let mut after_labels = node_after(update).expect("after").labels;
after_labels.sort();
assert_eq!(after_labels, vec!["Archived", "Featured"]);
}
#[test]
fn a_property_write_before_a_label_write_still_reports_the_old_labels() {
let mut graph = seeded_full();
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Featured");
commit(&mut graph);
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.qty = 77");
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Archived");
commit(&mut graph);
let update = &since(&graph, from)[0];
let before = node_before(update).expect("before");
assert_eq!(
before.labels,
vec!["Featured".to_string()],
"the property write opened the image; the label write filled its labels in"
);
assert_eq!(
before
.properties
.iter()
.find(|(k, _)| k == "qty")
.map(|(_, v)| v.clone()),
Some(Value::Int64(10)),
"and the property half is still the commit-start one"
);
}
#[test]
fn a_deleted_nodes_before_image_keeps_its_labels() {
let mut graph = seeded_full();
run(&mut graph, "MATCH (i:Item {id: 1}) SET i:Featured");
commit(&mut graph);
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) DETACH DELETE i");
commit(&mut graph);
let node_delete = since(&graph, from)
.into_iter()
.find(|event| {
event.kind == CdcEventKind::Delete && matches!(event.change, CdcChange::Node { .. })
})
.expect("the delete must publish");
assert_eq!(
node_before(&node_delete).map(|state| state.labels),
Some(vec!["Featured".to_string()]),
"labels live above storage, so a delete's image has to be filled in \
before the label index is swept"
);
}
#[test]
fn a_no_op_label_remove_publishes_nothing_under_full() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) REMOVE i:NeverHad");
commit(&mut graph);
assert_eq!(
since(&graph, from),
Vec::new(),
"offering a before-image must not invent a change"
);
}
#[test]
fn a_rolled_back_statement_discards_its_before_image_too() {
let mut graph = seeded_full();
let from = cursor(&graph);
expect_failure(&mut graph, FAILS_AFTER_WRITING);
commit(&mut graph);
assert_eq!(since(&graph, from), Vec::new());
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'legit'");
commit(&mut graph);
let update = &since(&graph, from)[0];
assert_eq!(
node_before_property(update, "name"),
Some(Value::String("one".into())),
"a truncated image must not survive into the next commit"
);
}
fn round_trip(graph: &DirGraph, dir: &std::path::Path, name: &str) -> Arc<DirGraph> {
let path = dir.join(name);
let path = path.to_string_lossy().to_string();
crate::graph::io::file::write_kgl(graph, &path).expect("save");
crate::graph::io::file::load_file(&path).expect("load")
}
#[test]
fn a_save_stamps_the_running_epochs_position_and_a_load_restores_it() {
let tmp = tempfile::tempdir().expect("tempdir");
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable");
run(&mut graph, "CREATE (:Item {id: 1})");
commit(&mut graph);
run(&mut graph, "CREATE (:Item {id: 2})");
commit(&mut graph);
let status = cdc::status(&graph).expect("enabled");
let loaded = round_trip(&graph, tmp.path(), "stamped.kgl");
assert_eq!(
loaded.cdc_handoff,
Some(crate::graph::cdc::CdcHandoff {
epoch: status.epoch,
last_seq: status.current,
}),
"the stamp must name the epoch and where it ended"
);
assert!(
!loaded.cdc_enabled(),
"the stamp is a diagnostic about a log that is gone, not a resumed one"
);
}
#[test]
fn a_graph_that_never_captured_stamps_nothing() {
let tmp = tempfile::tempdir().expect("tempdir");
let mut graph = DirGraph::new();
run(&mut graph, "CREATE (:Item {id: 1})");
let loaded = round_trip(&graph, tmp.path(), "unstamped.kgl");
assert_eq!(loaded.cdc_handoff, None);
}
#[test]
fn a_save_after_disable_carries_the_earlier_stamp_forward() {
let tmp = tempfile::tempdir().expect("tempdir");
let mut graph = DirGraph::new();
cdc::enable(&mut graph, None, CdcEnrichment::Off).expect("enable");
run(&mut graph, "CREATE (:Item {id: 1})");
commit(&mut graph);
let status = cdc::status(&graph).expect("enabled");
cdc::disable(&mut graph);
let loaded = round_trip(&graph, tmp.path(), "after-disable.kgl");
assert_eq!(
loaded.cdc_handoff.map(|handoff| handoff.epoch),
Some(status.epoch)
);
let again = round_trip(&loaded, tmp.path(), "second-hop.kgl");
assert_eq!(again.cdc_handoff, loaded.cdc_handoff);
}
const DELETE_THEN_FAILS: &str = "MATCH (i:Item {id: 1}) DETACH DELETE i \
WITH 1 AS x MATCH (m:Item {id: 2}) SET m.qty = duration({months: 2147483648})";
#[test]
fn a_rolled_back_delete_leaves_the_earlier_ops_before_image_intact() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'changed'");
expect_failure(&mut graph, DELETE_THEN_FAILS);
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
published.len(),
1,
"only the surviving update publishes: {published:?}"
);
assert_eq!(published[0].kind, CdcEventKind::Update);
assert_eq!(
node_before_property(&published[0], "name"),
Some(Value::String("one".into())),
"the image the rolled-back delete borrowed must still be here"
);
assert_eq!(
node_property(&published[0], "name"),
Some(Value::String("changed".into())),
"and the after half is the write that survived"
);
}
#[test]
fn the_same_commit_without_a_delete_reports_its_before_image() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'changed'");
commit(&mut graph);
let published = since(&graph, from);
assert_eq!(
node_before_property(&published[0], "name"),
Some(Value::String("one".into()))
);
}
#[test]
fn a_delete_after_a_rolled_back_delete_still_reports_the_commit_start_image() {
let mut graph = seeded_full();
let from = cursor(&graph);
run(&mut graph, "MATCH (i:Item {id: 1}) SET i.name = 'changed'");
expect_failure(&mut graph, DELETE_THEN_FAILS);
run(&mut graph, "MATCH (i:Item {id: 1}) DETACH DELETE i");
commit(&mut graph);
let node_delete = since(&graph, from)
.into_iter()
.find(|event| {
event.kind == CdcEventKind::Delete && matches!(event.change, CdcChange::Node { .. })
})
.expect("the second delete must publish");
assert_eq!(
node_before_property(&node_delete, "name"),
Some(Value::String("one".into())),
"before is the state the commit found, not what the surviving SET left"
);
}