use super::{
AedbConfig, AedbInstance, ColumnDef, ColumnType, CommitFinality, ConsistencyMode, DdlOperation,
DurabilityMode, Expr, Mutation, Query, Row, Value, create_table,
};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;
#[tokio::test]
async fn commit_with_visible_finality_can_return_before_durable_head_in_batch_mode() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
durability_mode: DurabilityMode::Batch,
batch_interval_ms: 60_000,
batch_max_bytes: usize::MAX,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
let result = db
.commit_with_finality(
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"fast-visible".to_vec(),
value: b"v".to_vec(),
},
CommitFinality::Visible,
)
.await
.expect("commit");
assert!(
result.durable_head_seq < result.commit_seq,
"visible finality should not require durable head in batch mode"
);
}
#[tokio::test]
async fn commit_with_durable_finality_waits_until_durable_head_catches_up() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
durability_mode: DurabilityMode::Batch,
batch_interval_ms: 60_000,
batch_max_bytes: usize::MAX,
..AedbConfig::default()
};
let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
db.create_project("p").await.expect("project");
let fsync_db = Arc::clone(&db);
let fsync_task = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(20)).await;
fsync_db.force_fsync().await.expect("force fsync");
});
let started = Instant::now();
let result = db
.commit_with_finality(
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"fast-durable".to_vec(),
value: b"v".to_vec(),
},
CommitFinality::Durable,
)
.await
.expect("commit");
fsync_task.await.expect("join fsync");
assert!(
started.elapsed() >= Duration::from_millis(15),
"durable finality should wait for WAL durability in batch mode"
);
assert!(
result.durable_head_seq >= result.commit_seq,
"durable finality must report durable head at or beyond commit sequence"
);
}
#[tokio::test]
#[ignore = "long-running finality latency profile"]
async fn finality_profile_visible_vs_durable_low_latency_mode() {
async fn run_profile(
config: AedbConfig,
finality: CommitFinality,
ops: usize,
) -> (u64, u64, u64, crate::OperationalMetrics) {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
let started = Instant::now();
let mut lat_sum = 0u128;
let mut lat_max = 0u64;
for i in 0..ops {
let op_started = Instant::now();
db.commit_with_finality(
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: format!("finality:{finality:?}:{i}").into_bytes(),
value: i.to_be_bytes().to_vec(),
},
finality,
)
.await
.expect("commit with finality");
let us = op_started.elapsed().as_micros() as u64;
lat_sum = lat_sum.saturating_add(us as u128);
lat_max = lat_max.max(us);
}
db.force_fsync().await.expect("flush");
let elapsed = started.elapsed().as_secs_f64().max(0.001);
let tps = (ops as f64 / elapsed) as u64;
let avg_us = (lat_sum / ops.max(1) as u128) as u64;
let op = db.operational_metrics().await;
(tps, avg_us, lat_max, op)
}
let ops = std::env::var("AEDB_FINALITY_PROFILE_OPS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(600)
.max(200);
let mut low_latency_no_coalesce = AedbConfig::low_latency([1u8; 32]);
low_latency_no_coalesce.durable_ack_coalescing_enabled = false;
low_latency_no_coalesce.durable_ack_coalesce_window_us = 0;
let low_latency_coalesce = AedbConfig::low_latency([1u8; 32]);
let (visible_tps, visible_avg_us, visible_max_us, visible_op) = run_profile(
low_latency_no_coalesce.clone(),
CommitFinality::Visible,
ops,
)
.await;
let (durable_base_tps, durable_base_avg_us, durable_base_max_us, durable_base_op) =
run_profile(low_latency_no_coalesce, CommitFinality::Durable, ops).await;
let (durable_tps, durable_avg_us, durable_max_us, durable_op) =
run_profile(low_latency_coalesce, CommitFinality::Durable, ops).await;
eprintln!(
"finality_profile: ops={} visible_tps={} durable_base_tps={} durable_coalesced_tps={} visible_avg_us={} durable_base_avg_us={} durable_coalesced_avg_us={} visible_max_us={} durable_base_max_us={} durable_coalesced_max_us={} visible_durable_wait_ops={} durable_base_wait_ops={} durable_coalesced_wait_ops={} visible_avg_durable_wait_us={} durable_base_avg_durable_wait_us={} durable_coalesced_avg_durable_wait_us={} visible_wal_sync_ops={} durable_base_wal_sync_ops={} durable_coalesced_wal_sync_ops={} visible_avg_wal_sync_us={} durable_base_avg_wal_sync_us={} durable_coalesced_avg_wal_sync_us={} visible_avg_wal_append_us={} durable_base_avg_wal_append_us={} durable_coalesced_avg_wal_append_us={}",
ops,
visible_tps,
durable_base_tps,
durable_tps,
visible_avg_us,
durable_base_avg_us,
durable_avg_us,
visible_max_us,
durable_base_max_us,
durable_max_us,
visible_op.durable_wait_ops,
durable_base_op.durable_wait_ops,
durable_op.durable_wait_ops,
visible_op.avg_durable_wait_micros,
durable_base_op.avg_durable_wait_micros,
durable_op.avg_durable_wait_micros,
visible_op.wal_sync_ops,
durable_base_op.wal_sync_ops,
durable_op.wal_sync_ops,
visible_op.avg_wal_sync_micros,
durable_base_op.avg_wal_sync_micros,
durable_op.avg_wal_sync_micros,
visible_op.avg_wal_append_micros,
durable_base_op.avg_wal_append_micros,
durable_op.avg_wal_append_micros
);
assert_eq!(
visible_op.queue_full_rejections, 0,
"visible finality profile should not saturate queue"
);
assert_eq!(
durable_base_op.queue_full_rejections, 0,
"durable baseline profile should not saturate queue"
);
assert_eq!(
durable_op.queue_full_rejections, 0,
"durable coalesced profile should not saturate queue"
);
assert_eq!(
visible_op.timeout_rejections, 0,
"visible finality profile should not timeout"
);
assert_eq!(
durable_base_op.timeout_rejections, 0,
"durable baseline profile should not timeout"
);
assert_eq!(
durable_op.timeout_rejections, 0,
"durable coalesced profile should not timeout"
);
assert_eq!(
visible_op.durable_wait_ops, 0,
"visible finality profile should not accumulate durable wait operations"
);
assert!(
durable_op.durable_wait_ops > 0,
"durable finality profile should accumulate durable wait operations"
);
assert!(
durable_tps >= durable_base_tps.saturating_div(2),
"coalesced durable finality regressed severely: base={durable_base_tps} coalesced={durable_tps}"
);
assert!(
durable_tps <= visible_tps.saturating_mul(2),
"durable finality profile produced implausible TPS vs visible: visible={visible_tps} durable={durable_tps}"
);
}
#[tokio::test]
async fn commit_success_is_observable_at_its_commit_seq() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let result = db
.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"inclusion-proof".to_vec(),
value: b"ok".to_vec(),
})
.await
.expect("commit");
let at_seq = db
.kv_get_no_auth(
"p",
"app",
b"inclusion-proof",
ConsistencyMode::AtSeq(result.commit_seq),
)
.await
.expect("kv_get at seq")
.expect("value present at commit seq");
assert_eq!(at_seq.value, b"ok".to_vec());
}
#[tokio::test]
async fn subscribe_commits_delivers_delta_after_commit() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
// Subscribe AFTER setup so we observe only the deltas under test.
let mut rx = db.subscribe_commits();
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"hello".to_vec(),
value: b"world".to_vec(),
})
.await
.expect("commit");
// Drain until we find our KvSet — internal subsystems may also produce
// bookkeeping deltas, but ours must arrive within the timeout window.
let mut saw_kv_set = false;
let mut last_seq = 0u64;
let deadline = std::time::Instant::now() + Duration::from_secs(2);
while std::time::Instant::now() < deadline {
match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
Ok(Ok(delta)) => {
last_seq = last_seq.max(delta.seq);
if delta
.mutations
.iter()
.any(|m| matches!(m, Mutation::KvSet { key, .. } if key == b"hello"))
{
saw_kv_set = true;
break;
}
}
Ok(Err(_)) => break,
Err(_) => break,
}
}
assert!(last_seq > 0, "broadcast delivered at least one delta");
assert!(
saw_kv_set,
"broadcast delta must include the committed KvSet mutation"
);
db.shutdown().await.expect("shutdown");
}
#[tokio::test]
async fn subscribe_commits_lagged_subscriber_can_resume() {
let config = AedbConfig {
commit_broadcast_capacity: 2,
..AedbConfig::default()
};
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
let mut rx = db.subscribe_commits();
db.create_project("p").await.expect("project");
for i in 0..8u8 {
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: vec![i],
value: vec![i],
})
.await
.expect("commit");
}
let mut saw_lagged = false;
let mut delivered = 0usize;
let deadline = std::time::Instant::now() + Duration::from_secs(2);
while std::time::Instant::now() < deadline {
match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
Ok(Ok(_delta)) => {
delivered += 1;
if saw_lagged && delivered >= 2 {
break;
}
}
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => {
saw_lagged = true;
}
Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => break,
Err(_) => break,
}
}
assert!(
saw_lagged,
"slow subscriber must observe RecvError::Lagged with capacity=2 + burst"
);
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: vec![99],
value: vec![99],
})
.await
.expect("post-burst commit");
let post = tokio::time::timeout(Duration::from_secs(1), rx.recv())
.await
.expect("post-burst delivery within timeout")
.expect("post-burst delta");
let resumed = post.mutations.iter().any(|m| {
matches!(
m,
Mutation::KvSet { key, .. } if key == &[99u8]
)
});
assert!(resumed, "subscriber must resume after Lagged error");
db.shutdown().await.expect("shutdown");
}
/// #3: with `row_change_deltas_enabled`, each broadcast `CommitDelta` carries
/// resolved row-level changes — insert vs update distinguished, deletes resolved
/// from a predicate — so a subscription layer can diff without re-querying.
#[tokio::test]
async fn commit_delta_carries_resolved_row_changes_when_enabled() {
use crate::commit::row_change::RowChangeKind;
let dir = tempdir().expect("temp");
let config = AedbConfig {
row_change_deltas_enabled: true,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("arcana").await.expect("project");
db.create_scope("arcana", "app").await.expect("scope");
create_table(
&db,
"arcana",
"app",
"entities",
vec![
ColumnDef {
name: "instance_id".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "entity_id".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "component_name".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "data".into(),
col_type: ColumnType::Json,
nullable: false,
},
],
vec!["instance_id", "entity_id", "component_name"],
)
.await;
let upsert = |eid: &str, comp: &str, data: &str| Mutation::Upsert {
project_id: "arcana".into(),
scope_id: "app".into(),
table_name: "entities".into(),
primary_key: vec![
Value::Text("i1".into()),
Value::Text(eid.into()),
Value::Text(comp.into()),
],
row: Row::from_values(vec![
Value::Text("i1".into()),
Value::Text(eid.into()),
Value::Text(comp.into()),
Value::Json(data.into()),
]),
};
// Subscribe after DDL so the first delivered delta is our first upsert.
let mut rx = db.subscribe_commits();
// New row → Insert, with the new values attached.
db.commit(upsert("e1", "Pos", r#"{"x":1}"#))
.await
.expect("insert");
let d = rx.recv().await.expect("delta");
assert_eq!(d.row_changes.len(), 1);
assert_eq!(d.row_changes[0].kind, RowChangeKind::Insert);
assert_eq!(d.row_changes[0].table_name, "entities");
assert!(matches!(&d.row_changes[0].new_row, Some(r) if r.values.len() == 4));
// Same PK again → Update.
db.commit(upsert("e1", "Pos", r#"{"x":2}"#))
.await
.expect("update");
let d = rx.recv().await.expect("delta");
assert_eq!(d.row_changes.len(), 1);
assert_eq!(d.row_changes[0].kind, RowChangeKind::Update);
// A second component so the despawn resolves two rows.
db.commit(upsert("e1", "Vel", r#"{"dx":3}"#))
.await
.expect("insert2");
let _ = rx.recv().await.expect("delta");
// Despawn via a PK-prefix predicate → both component rows resolved as Delete.
db.commit(Mutation::DeleteWhere {
project_id: "arcana".into(),
scope_id: "app".into(),
table_name: "entities".into(),
predicate: Expr::Eq("instance_id".into(), Value::Text("i1".into()))
.and(Expr::Eq("entity_id".into(), Value::Text("e1".into()))),
limit: None,
})
.await
.expect("despawn");
let d = rx.recv().await.expect("delta");
assert_eq!(d.row_changes.len(), 2, "both components resolved");
assert!(
d.row_changes
.iter()
.all(|c| c.kind == RowChangeKind::Delete)
);
assert!(d.row_changes.iter().all(|c| c.new_row.is_none()));
}
/// #3: when the feature is off (default), deltas carry no row-level changes —
/// zero overhead, unchanged behavior.
#[tokio::test]
async fn commit_delta_row_changes_empty_by_default() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
db.create_scope("p", "app").await.expect("scope");
create_table(
&db,
"p",
"app",
"t",
vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "v".into(),
col_type: ColumnType::Integer,
nullable: false,
},
],
vec!["id"],
)
.await;
let mut rx = db.subscribe_commits();
db.commit(Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "t".into(),
primary_key: vec![Value::Integer(1)],
row: Row::from_values(vec![Value::Integer(1), Value::Integer(9)]),
})
.await
.expect("upsert");
let d = rx.recv().await.expect("delta");
assert!(d.row_changes.is_empty());
}
// --- read-your-writes contract ---------------------------------------------------------------
//
// `commit()` resolves only after the epoch loop has applied the write to the visible state and
// bumped the snapshot generation (executor loop: apply → bump generation under the state lock →
// ack), so an `AtLatest` read issued after the ack MUST observe the write. Consumers build
// worker loops on this (e.g. Arcana's task queue scans for rows a just-acked commit produced);
// if a cache or pipeline change ever lets an ack overtake visibility, work gets silently
// deferred or dropped. These tests pin the contract empirically, single-writer and under
// concurrency, on both immediate and batch/coalesced durability profiles.
async fn open_ryw_instance(config: AedbConfig, dir: &std::path::Path) -> Arc<AedbInstance> {
let db = Arc::new(AedbInstance::open_anonymous(config, dir).expect("open"));
db.create_project("p").await.expect("project");
db.create_scope("p", "s").await.expect("scope");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "ryw".into(),
owner_id: None,
if_not_exists: true,
columns: vec![
ColumnDef {
name: "k".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "v".into(),
col_type: ColumnType::Integer,
nullable: false,
},
],
primary_key: vec!["k".into()],
}))
.await
.expect("create table");
db
}
async fn assert_write_visible_after_ack(db: &AedbInstance, key: String, i: i64) {
let result = db
.commit(Mutation::Upsert {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "ryw".into(),
primary_key: vec![Value::Text(key.clone().into())],
row: Row::from_values(vec![Value::Text(key.clone().into()), Value::Integer(i)]),
})
.await
.expect("commit");
// The published read view must already include our commit...
let probe = db
.snapshot_probe(ConsistencyMode::AtLatest)
.await
.expect("probe");
assert!(
probe >= result.commit_seq,
"AtLatest view seq {probe} lags acked commit_seq {} (read-your-writes violated)",
result.commit_seq
);
// ...and an AtLatest query must actually see the row value we just wrote.
let rows = db
.query(
"p",
"s",
Query::select(&["v"])
.from("ryw")
.where_(Expr::Eq("k".into(), Value::Text(key.clone().into())))
.limit(1),
)
.await
.expect("query")
.rows;
let got = rows.first().and_then(|r| match r.values.first() {
Some(Value::Integer(n)) => Some(*n),
_ => None,
});
assert_eq!(
got,
Some(i),
"AtLatest query after ack of key {key} iteration {i} did not see the write"
);
}
#[tokio::test]
async fn commit_ack_implies_read_your_writes() {
for config in [AedbConfig::default(), AedbConfig::low_latency([5u8; 32])] {
let dir = tempdir().expect("temp");
let db = open_ryw_instance(config, dir.path()).await;
for i in 0..300i64 {
assert_write_visible_after_ack(&db, "solo".to_string(), i).await;
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_commit_acks_imply_read_your_writes() {
let coalesced = AedbConfig {
durable_ack_coalescing_enabled: true,
..AedbConfig::default()
};
for config in [AedbConfig::default(), coalesced] {
let dir = tempdir().expect("temp");
let db = open_ryw_instance(config, dir.path()).await;
let mut handles = Vec::new();
for w in 0..8 {
let db = Arc::clone(&db);
handles.push(tokio::spawn(async move {
for i in 0..100i64 {
assert_write_visible_after_ack(&db, format!("w{w}"), i).await;
}
}));
}
for h in handles {
h.await.expect("worker");
}
}
}