use super::{
ActionCommitOutcome, ActionEnvelopeRequest, AedbConfig, AedbError, AedbInstance, CallerContext,
ColumnDef, ColumnType, CommitFinality, ConsistencyMode, DdlOperation, DurabilityMode,
ErrorResourceType, Expr, IdempotencyKey, KvU256MissingPolicy, Mutation, Permission, Query,
ReadAssertion, RecoveryMode, Row, StorageMode, TransactionEnvelope, Value, WriteClass,
WriteIntent, u256_be_test,
};
use crate::commit::tx::ReadSet;
use crate::commit::validation::{CompareOp, KvU256OverflowPolicy};
use crate::preflight::PreflightResult;
use crate::query::plan::QueryOptions;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;
#[tokio::test]
async fn production_profile_requires_hmac() {
let dir = tempdir().expect("temp");
let invalid = AedbConfig {
recovery_mode: RecoveryMode::Strict,
durability_mode: DurabilityMode::Full,
hash_chain_required: true,
manifest_hmac_key: None,
..AedbConfig::default()
};
let err = AedbInstance::open_production(invalid, dir.path())
.err()
.expect("must reject missing hmac");
assert!(matches!(err, AedbError::InvalidConfig { .. }));
let valid = AedbConfig::production([7u8; 32]);
AedbInstance::open_production(valid, dir.path()).expect("open production");
}
#[tokio::test]
async fn production_profile_rejects_batch_durability() {
let dir = tempdir().expect("temp");
let mut weak = AedbConfig::production([7u8; 32]);
weak.durability_mode = DurabilityMode::Batch;
let err = AedbInstance::open_production(weak, dir.path())
.err()
.expect("production profile must reject batch durability");
assert!(matches!(err, AedbError::InvalidConfig { .. }));
}
#[test]
fn arcana_profile_rejects_short_hmac_key() {
let weak = AedbConfig::default().with_hmac_key(vec![9u8; 16]);
let err = crate::lib_helpers::validate_arcana_config(&weak)
.expect_err("short hmac key must be rejected");
assert!(matches!(err, AedbError::InvalidConfig { .. }));
}
#[test]
fn low_latency_profile_uses_batch_durability_with_strict_recovery() {
let cfg = AedbConfig::low_latency([5u8; 32]);
assert_eq!(cfg.durability_mode, DurabilityMode::Batch);
assert_eq!(cfg.recovery_mode, RecoveryMode::Strict);
assert!(cfg.hash_chain_required);
assert!(cfg.batch_interval_ms > 0);
assert!(cfg.batch_max_bytes > 0);
assert!(cfg.manifest_hmac_key.is_some());
assert_eq!(cfg.storage_mode, StorageMode::DiskBacked);
assert_eq!(cfg.persistent_value_inline_threshold_bytes, 0);
}
#[tokio::test]
async fn envelope_read_bytes_budget_passes_for_tiny_envelope() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
max_read_bytes_per_envelope: 1024,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
db.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: None,
write_class: WriteClass::Standard,
assertions: vec![ReadAssertion::KeyExists {
project_id: "p".into(),
scope_id: "app".into(),
key: b"absent".to_vec(),
expected: false,
}],
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"k".to_vec(),
value: b"v".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect("tiny envelope must fit budget");
}
#[tokio::test]
async fn envelope_read_bytes_budget_rejects_large_scan() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
max_scan_rows: 1_000_000,
max_read_bytes_per_envelope: 512,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: None,
if_not_exists: false,
project_id: "p".into(),
scope_id: "app".into(),
table_name: "items".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "blob".into(),
col_type: ColumnType::Blob,
nullable: false,
},
],
primary_key: vec!["id".into()],
}))
.await
.expect("create table");
for i in 0..64i64 {
db.commit(Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "items".into(),
primary_key: vec![Value::Integer(i)],
row: Row::from_values(vec![Value::Integer(i), Value::Blob(vec![0xAB; 128])]),
})
.await
.expect("seed row");
}
let err = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: None,
write_class: WriteClass::Standard,
assertions: vec![ReadAssertion::CountCompare {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "items".into(),
filter: None,
op: CompareOp::Gte,
threshold: 0,
}],
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"_".to_vec(),
value: b"_".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect_err("aggregate read byte budget must reject large scan");
match err {
AedbError::Validation(msg) => {
assert!(
msg.contains("max_read_bytes_per_envelope"),
"informative limit message expected, got: {msg}"
);
}
other => panic!("expected Validation error, got: {other:?}"),
}
}
#[tokio::test]
async fn commit_ddl_batch_is_atomic_and_reports_per_op_results() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("arcana").await.expect("project");
let failed = db
.commit_ddl_batch(vec![
DdlOperation::CreateTable {
owner_id: None,
if_not_exists: false,
project_id: "arcana".into(),
scope_id: "app".into(),
table_name: "users".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "name".into(),
col_type: ColumnType::Text,
nullable: false,
},
],
primary_key: vec!["id".into()],
},
DdlOperation::CreateTable {
owner_id: None,
if_not_exists: false,
project_id: "arcana".into(),
scope_id: "missing".into(),
table_name: "invalid".into(),
columns: vec![ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
}],
primary_key: vec!["id".into()],
},
])
.await
.expect_err("invalid second ddl should fail entire batch");
assert!(matches!(
failed,
AedbError::NotFound {
resource_type: ErrorResourceType::Scope,
..
}
));
assert!(
!db.table_exists("arcana", "app", "users")
.await
.expect("users table exists check")
);
let first = db
.commit_ddl_batch(vec![DdlOperation::CreateTable {
owner_id: None,
if_not_exists: true,
project_id: "arcana".into(),
scope_id: "app".into(),
table_name: "users".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "name".into(),
col_type: ColumnType::Text,
nullable: false,
},
],
primary_key: vec!["id".into()],
}])
.await
.expect("first batch");
assert_eq!(first.results.len(), 1);
assert!(first.results[0].applied);
assert!(
db.table_exists("arcana", "app", "users")
.await
.expect("table exists")
);
let second = db
.commit_ddl_batch(vec![DdlOperation::CreateTable {
owner_id: None,
if_not_exists: true,
project_id: "arcana".into(),
scope_id: "app".into(),
table_name: "users".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "name".into(),
col_type: ColumnType::Text,
nullable: false,
},
],
primary_key: vec!["id".into()],
}])
.await
.expect("second batch");
assert_eq!(second.results.len(), 1);
assert!(!second.results[0].applied);
}
#[tokio::test]
async fn commit_ddl_batch_supports_dependent_ddl_ordering() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
let batch = db
.commit_ddl_batch(vec![
DdlOperation::CreateProject {
owner_id: None,
project_id: "arcana".into(),
if_not_exists: true,
},
DdlOperation::CreateScope {
owner_id: None,
project_id: "arcana".into(),
scope_id: "ops".into(),
if_not_exists: true,
},
DdlOperation::CreateTable {
owner_id: None,
if_not_exists: true,
project_id: "arcana".into(),
scope_id: "ops".into(),
table_name: "users".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "name".into(),
col_type: ColumnType::Text,
nullable: false,
},
],
primary_key: vec!["id".into()],
},
])
.await
.expect("dependent batch");
assert_eq!(batch.results.len(), 3);
assert!(batch.results.iter().all(|r| r.applied));
assert!(
db.project_exists("arcana")
.await
.expect("project exists after batch")
);
assert!(
db.scope_exists("arcana", "ops")
.await
.expect("scope exists after batch")
);
assert!(
db.table_exists("arcana", "ops", "users")
.await
.expect("table exists after batch")
);
}
#[tokio::test]
async fn idempotency_prunes_by_commit_window() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(
AedbConfig {
idempotency_window_commits: 1,
..AedbConfig::default()
},
dir.path(),
)
.expect("open");
db.create_project("p").await.expect("project");
let first = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: Some(IdempotencyKey([1u8; 16])),
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"k".to_vec(),
value: b"v1".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect("first commit");
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"other".to_vec(),
value: b"v2".to_vec(),
})
.await
.expect("advance seq");
let retried = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: Some(IdempotencyKey([1u8; 16])),
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"k".to_vec(),
value: b"v3".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect("retried commit");
assert!(
retried.commit_seq > first.commit_seq,
"idempotency record should expire by sequence window"
);
}
#[tokio::test]
async fn idempotency_prunes_by_time_window() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(
AedbConfig {
idempotency_window_commits: 100,
idempotency_window_seconds: 1,
..AedbConfig::default()
},
dir.path(),
)
.expect("open");
db.create_project("p").await.expect("project");
let first = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: Some(IdempotencyKey([9u8; 16])),
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"time".to_vec(),
value: b"v1".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect("first commit");
tokio::time::sleep(Duration::from_millis(1_100)).await;
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"tick".to_vec(),
value: b"v2".to_vec(),
})
.await
.expect("advance and prune");
let retried = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: Some(IdempotencyKey([9u8; 16])),
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"time".to_vec(),
value: b"v3".to_vec(),
}],
},
base_seq: 0,
})
.await
.expect("retried commit");
assert!(
retried.commit_seq > first.commit_seq,
"idempotency record should expire by time window"
);
}
#[test]
fn open_rejects_invalid_config() {
let dir = tempdir().expect("temp");
let bad = AedbConfig {
durability_mode: DurabilityMode::Batch,
batch_interval_ms: 0,
..AedbConfig::default()
};
let err = AedbInstance::open_anonymous(bad, dir.path())
.err()
.expect("invalid config");
assert!(matches!(err, crate::error::AedbError::InvalidConfig { .. }));
let too_large_txn = AedbConfig {
max_transaction_bytes: crate::wal::frame::MAX_FRAME_BODY_BYTES + 1,
..AedbConfig::default()
};
let err = AedbInstance::open_anonymous(too_large_txn, dir.path())
.err()
.expect("oversized transaction bound");
assert!(matches!(err, crate::error::AedbError::InvalidConfig { .. }));
let deadlock_unsafe = AedbConfig {
recovery_mode: RecoveryMode::Strict,
coordinator_locking_enabled: false,
..AedbConfig::default()
};
let err = AedbInstance::open_anonymous(deadlock_unsafe, dir.path())
.err()
.expect("strict mode must require coordinator locking");
assert!(matches!(err, crate::error::AedbError::InvalidConfig { .. }));
}
#[tokio::test]
async fn metrics_surface_reflects_commits() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let m = db.metrics();
assert!(m.commits_total >= 1);
let op = db.operational_metrics().await;
assert!(op.commits_total >= 1);
assert!(op.read_set_conflicts <= op.conflict_rejections);
assert!(op.queue_depth >= op.inflight_commits);
assert!(op.snapshot_age_micros <= u64::MAX / 2);
}
#[tokio::test]
async fn failed_multi_mutation_envelope_is_atomic_and_has_no_partial_effects() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let before = db.head_state().await.visible_head_seq;
let err = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: None,
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: Default::default(),
write_intent: WriteIntent {
mutations: vec![
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"atomic-a".to_vec(),
value: b"1".to_vec(),
},
Mutation::KvDecU256 {
project_id: "p".into(),
scope_id: "app".into(),
key: b"missing-counter".to_vec(),
amount_be: {
let mut out = [0u8; 32];
out[31] = 1;
out
},
},
],
},
base_seq: before,
})
.await
.expect_err("envelope should fail");
assert!(
matches!(
err,
AedbError::Underflow | AedbError::Validation(_) | AedbError::Conflict(_)
),
"expected semantic failure, got: {err:?}"
);
let after = db.head_state().await.visible_head_seq;
assert_eq!(after, before, "failed envelope must not advance head");
let leaked = db
.kv_get_no_auth("p", "app", b"atomic-a", ConsistencyMode::AtLatest)
.await
.expect("kv_get")
.is_some();
assert!(
!leaked,
"failed envelope must not partially apply mutations"
);
}
#[tokio::test]
async fn action_envelope_applied_once() {
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_action_envelope(ActionEnvelopeRequest {
caller: None,
idempotency_key: IdempotencyKey([1u8; 16]),
write_class: WriteClass::Standard,
base_seq: 0,
assertions: Vec::new(),
mutations: vec![
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"action:flag".to_vec(),
value: b"ok".to_vec(),
},
Mutation::KvAddU256Ex {
project_id: "p".into(),
scope_id: "app".into(),
key: b"action:counter".to_vec(),
amount_be: u256_be_test(3),
on_missing: KvU256MissingPolicy::TreatAsZero,
on_overflow: KvU256OverflowPolicy::Reject,
},
],
})
.await
.expect("action commit");
assert_eq!(result.outcome, ActionCommitOutcome::Applied);
let flag = db
.kv_get_no_auth("p", "app", b"action:flag", ConsistencyMode::AtLatest)
.await
.expect("flag read")
.expect("flag exists");
assert_eq!(flag.value, b"ok".to_vec());
}
#[tokio::test]
async fn action_envelope_duplicate_returns_duplicate_outcome() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let req = ActionEnvelopeRequest {
caller: None,
idempotency_key: IdempotencyKey([2u8; 16]),
write_class: WriteClass::Standard,
base_seq: 0,
assertions: Vec::new(),
mutations: vec![Mutation::KvAddU256Ex {
project_id: "p".into(),
scope_id: "app".into(),
key: b"action:dup-counter".to_vec(),
amount_be: u256_be_test(1),
on_missing: KvU256MissingPolicy::TreatAsZero,
on_overflow: KvU256OverflowPolicy::Reject,
}],
};
let first = db
.commit_action_envelope(req.clone())
.await
.expect("first action");
let second = db
.commit_action_envelope(req)
.await
.expect("duplicate action");
assert_eq!(first.outcome, ActionCommitOutcome::Applied);
assert_eq!(second.outcome, ActionCommitOutcome::Duplicate);
assert_eq!(second.commit_seq, first.commit_seq);
let counter = db
.kv_get_no_auth("p", "app", b"action:dup-counter", ConsistencyMode::AtLatest)
.await
.expect("counter read")
.expect("counter exists");
assert_eq!(
primitive_types::U256::from_big_endian(&counter.value),
primitive_types::U256::one(),
);
}
#[tokio::test]
async fn single_envelope_atomicity() {
let dir = tempdir().expect("temp");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"guard".to_vec(),
value: b"1".to_vec(),
})
.await
.expect("guard seed");
let before = db.head_state().await.visible_head_seq;
let err = db
.commit_action_envelope(ActionEnvelopeRequest {
caller: None,
idempotency_key: IdempotencyKey([3u8; 16]),
write_class: WriteClass::Standard,
base_seq: before,
assertions: vec![ReadAssertion::KeyExists {
project_id: "p".into(),
scope_id: "app".into(),
key: b"guard".to_vec(),
expected: false,
}],
mutations: vec![
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"atomic:x".to_vec(),
value: b"x".to_vec(),
},
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"atomic:y".to_vec(),
value: b"y".to_vec(),
},
],
})
.await
.expect_err("assertion failure");
assert!(matches!(err, AedbError::AssertionFailed { .. }));
assert!(
db.head_state().await.visible_head_seq >= before,
"assertion failure may emit an audit commit, but data mutations must not apply"
);
assert!(
db.kv_get_no_auth("p", "app", b"atomic:x", ConsistencyMode::AtLatest)
.await
.expect("read x")
.is_none()
);
assert!(
db.kv_get_no_auth("p", "app", b"atomic:y", ConsistencyMode::AtLatest)
.await
.expect("read y")
.is_none()
);
}
#[tokio::test]
async fn retry_idempotency_is_exactly_once_under_commit_pressure() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
commit_timeout_ms: 1,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
let key = IdempotencyKey([91u8; 16]);
let envelope = TransactionEnvelope {
caller: None,
idempotency_key: Some(key),
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: Default::default(),
write_intent: WriteIntent {
mutations: vec![
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"__slow_parallel_worker__".to_vec(),
value: b"slow".to_vec(),
},
Mutation::KvIncU256 {
project_id: "p".into(),
scope_id: "app".into(),
key: b"timeout-idem-counter".to_vec(),
amount_be: {
let mut out = [0u8; 32];
out[31] = 1;
out
},
},
],
},
base_seq: 0,
};
let noisy_db = Arc::new(db);
let mut noise_tasks = Vec::new();
for worker in 0..8 {
let db_clone = Arc::clone(&noisy_db);
noise_tasks.push(tokio::spawn(async move {
for i in 0..200usize {
let _ = db_clone
.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: format!("noise:{worker}:{i}").into_bytes(),
value: b"n".to_vec(),
})
.await;
}
}));
}
let mut saw_timeout = false;
let second = loop {
match noisy_db.commit_envelope(envelope.clone()).await {
Ok(result) => break result,
Err(AedbError::Timeout) => {
saw_timeout = true;
tokio::time::sleep(Duration::from_millis(5)).await;
}
Err(other) => panic!("unexpected error during retry loop: {other:?}"),
}
};
for t in noise_tasks {
t.await.expect("join noise worker");
}
noisy_db
.wait_for_durable(second.commit_seq)
.await
.expect("durable ack");
let third = noisy_db
.commit_envelope(envelope)
.await
.expect("repeat idempotent retry");
assert_eq!(
third.commit_seq, second.commit_seq,
"all retries must resolve to one commit sequence"
);
let counter = noisy_db
.kv_get_no_auth(
"p",
"app",
b"timeout-idem-counter",
ConsistencyMode::AtLatest,
)
.await
.expect("kv counter")
.expect("counter exists");
assert_eq!(
primitive_types::U256::from_big_endian(&counter.value),
primitive_types::U256::one(),
"counter must be applied once"
);
let op = noisy_db.operational_metrics().await;
if saw_timeout {
assert!(
op.timeout_rejections >= 1,
"timeout path should be observable in operational metrics"
);
}
}
#[tokio::test]
async fn multi_update_transaction_envelope_updates_table_and_kv() {
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");
db.commit(Mutation::Ddl(DdlOperation::CreateTable {
owner_id: None,
if_not_exists: false,
project_id: "p".into(),
scope_id: "app".into(),
table_name: "accounts".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "balance".into(),
col_type: ColumnType::Integer,
nullable: false,
},
],
primary_key: vec!["id".into()],
}))
.await
.expect("accounts table");
db.commit(Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "accounts".into(),
primary_key: vec![Value::Integer(1)],
row: Row {
values: vec![Value::Integer(1), Value::Integer(100)],
},
})
.await
.expect("seed account 1");
db.commit(Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "accounts".into(),
primary_key: vec![Value::Integer(2)],
row: Row {
values: vec![Value::Integer(2), Value::Integer(80)],
},
})
.await
.expect("seed account 2");
db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
actor_id: None,
delegable: false,
caller_id: "reader".into(),
permission: Permission::KvRead {
project_id: "p".into(),
scope_id: Some("app".into()),
prefix: None,
},
}))
.await
.expect("grant kv read");
let caller = CallerContext::new("reader");
let result = db
.commit_envelope_with_finality(
TransactionEnvelope {
caller: None,
idempotency_key: None,
write_class: WriteClass::Standard,
assertions: Vec::new(),
read_set: Default::default(),
write_intent: WriteIntent {
mutations: vec![
Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "accounts".into(),
primary_key: vec![Value::Integer(1)],
row: Row {
values: vec![Value::Integer(1), Value::Integer(90)],
},
},
Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "accounts".into(),
primary_key: vec![Value::Integer(2)],
row: Row {
values: vec![Value::Integer(2), Value::Integer(90)],
},
},
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"tx:last".to_vec(),
value: b"t1".to_vec(),
},
Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"ledger:1->2".to_vec(),
value: b"10".to_vec(),
},
],
},
base_seq: db.head_state().await.visible_head_seq,
},
CommitFinality::Visible,
)
.await
.expect("multi-update tx");
assert!(
result.durable_head_seq < result.commit_seq,
"visible finality should return before durable head in batch mode"
);
let acct1 = db
.query_no_auth(
"p",
"app",
Query::select(&["balance"])
.from("accounts")
.where_(Expr::Eq("id".into(), Value::Integer(1)))
.limit(1),
QueryOptions::default(),
)
.await
.expect("query account1");
let acct2 = db
.query_no_auth(
"p",
"app",
Query::select(&["balance"])
.from("accounts")
.where_(Expr::Eq("id".into(), Value::Integer(2)))
.limit(1),
QueryOptions::default(),
)
.await
.expect("query account2");
assert_eq!(acct1.rows.len(), 1);
assert_eq!(acct2.rows.len(), 1);
assert_eq!(acct1.rows[0].values[0], Value::Integer(90));
assert_eq!(acct2.rows[0].values[0], Value::Integer(90));
let tx_last = db
.kv_get("p", "app", b"tx:last", ConsistencyMode::AtLatest, &caller)
.await
.expect("tx:last");
let ledger = db
.kv_get(
"p",
"app",
b"ledger:1->2",
ConsistencyMode::AtLatest,
&caller,
)
.await
.expect("ledger entry");
assert_eq!(
tx_last.as_ref().map(|v| v.value.clone()),
Some(b"t1".to_vec())
);
assert_eq!(
ledger.as_ref().map(|v| v.value.clone()),
Some(b"10".to_vec())
);
}
#[tokio::test]
#[ignore = "manual profiling: end-to-end pipeline breakdown (commit/checkpoint/recovery)"]
async fn profile_end_to_end_pipeline_breakdown() {
fn percentile(sorted: &[u128], p: f64) -> u128 {
if sorted.is_empty() {
return 0;
}
let percentile_index = ((sorted.len() as f64 - 1.0) * p).round() as usize;
sorted[percentile_index.min(sorted.len() - 1)]
}
let dir = tempdir().expect("temp");
let mut config = AedbConfig {
durability_mode: DurabilityMode::Batch,
batch_interval_ms: 10,
batch_max_bytes: usize::MAX,
recovery_mode: RecoveryMode::Permissive,
hash_chain_required: false,
..AedbConfig::default()
};
config.manifest_hmac_key = None;
let db = Arc::new(AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open"));
db.create_project("p").await.expect("project");
for i in 0..20_000usize {
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: format!("seed:{i:06}").into_bytes(),
value: vec![b's'; 256],
})
.await
.expect("seed");
}
let workers = 8usize;
let commits_per_worker = 500usize;
let commit_started = Instant::now();
let mut tasks = Vec::with_capacity(workers);
for worker in 0..workers {
let db = Arc::clone(&db);
tasks.push(tokio::spawn(async move {
let mut latencies_us = Vec::with_capacity(commits_per_worker);
for i in 0..commits_per_worker {
let started = Instant::now();
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: format!("profile:commit:{worker:02}:{i:06}").into_bytes(),
value: vec![b'c'; 256],
})
.await
.expect("commit");
latencies_us.push(started.elapsed().as_micros());
}
latencies_us
}));
}
let mut all_commit_latencies = Vec::with_capacity(workers * commits_per_worker);
for task in tasks {
let mut worker_latencies = task.await.expect("worker join");
all_commit_latencies.append(&mut worker_latencies);
}
let commit_elapsed = commit_started.elapsed();
all_commit_latencies.sort_unstable();
let commit_tps = (workers * commits_per_worker) as f64 / commit_elapsed.as_secs_f64();
let commit_p50 = percentile(&all_commit_latencies, 0.50);
let commit_p99 = percentile(&all_commit_latencies, 0.99);
let op_after_commits = db.operational_metrics().await;
let checkpoint_total_started = Instant::now();
let checkpoint_lock_started = Instant::now();
let _checkpoint_guard = db.checkpoint_lock.lock().await;
let checkpoint_lock_wait = checkpoint_lock_started.elapsed();
let snapshot_started = Instant::now();
let checkpoint_seq = db.executor.durable_head_seq_now();
let lease = db
.acquire_snapshot(ConsistencyMode::AtSeq(checkpoint_seq))
.await
.expect("checkpoint snapshot");
let snapshot_elapsed = snapshot_started.elapsed();
let idempotency_started = Instant::now();
let mut idempotency = db.executor.idempotency_snapshot().await;
idempotency.retain(|_, record| record.commit_seq <= checkpoint_seq);
let idempotency_elapsed = idempotency_started.elapsed();
let write_checkpoint_started = Instant::now();
let checkpoint = crate::checkpoint::writer::write_checkpoint_with_key(
lease.view.keyspace.as_ref(),
lease.view.catalog.as_ref(),
checkpoint_seq,
&db.dir,
db._config.checkpoint_key(),
db._config.checkpoint_key_id.clone(),
idempotency,
db._config.checkpoint_compression_level,
)
.expect("write checkpoint");
let write_checkpoint_elapsed = write_checkpoint_started.elapsed();
let segments_started = Instant::now();
let segments = crate::lib_helpers::read_segments_for_checkpoint(&db.dir, checkpoint_seq)
.expect("read segments for checkpoint");
let active_segment_seq = segments
.last()
.map(|segment| segment.segment_seq)
.unwrap_or(checkpoint_seq.saturating_add(1));
let segments_elapsed = segments_started.elapsed();
let manifest_started = Instant::now();
let manifest = crate::manifest::schema::Manifest {
durable_seq: checkpoint_seq,
visible_seq: checkpoint_seq,
active_segment_seq,
checkpoints: vec![checkpoint.clone()],
segments: segments.clone(),
..Default::default()
};
crate::manifest::atomic::write_manifest_atomic_signed(
&manifest,
&db.dir,
db._config.hmac_key(),
)
.expect("write manifest");
let manifest_elapsed = manifest_started.elapsed();
let checkpoint_total_elapsed = checkpoint_total_started.elapsed();
drop(_checkpoint_guard);
let checkpoint_bytes = std::fs::metadata(db.dir.join(&checkpoint.filename))
.expect("checkpoint stat")
.len();
drop(db);
let reopen_started = Instant::now();
let reopened = AedbInstance::open_anonymous(config, dir.path()).expect("reopen");
let reopen_elapsed = reopen_started.elapsed();
let reopen_metrics = reopened.operational_metrics().await;
eprintln!(
"pipeline_profile commit_phase: workers={} commits_per_worker={} commits={} elapsed_ms={} tps={:.2} p50_us={} p99_us={} | prestage_validate_ops={} avg_prestage_validate_us={} epoch_process_ops={} avg_epoch_process_us={} avg_wal_append_us={} avg_wal_sync_us={} avg_coordinator_apply_us={}",
workers,
commits_per_worker,
workers * commits_per_worker,
commit_elapsed.as_millis(),
commit_tps,
commit_p50,
commit_p99,
op_after_commits.prestage_validate_ops,
op_after_commits.avg_prestage_validate_micros,
op_after_commits.epoch_process_ops,
op_after_commits.avg_epoch_process_micros,
op_after_commits.avg_wal_append_micros,
op_after_commits.avg_wal_sync_micros,
op_after_commits.avg_coordinator_apply_micros
);
eprintln!(
"pipeline_profile checkpoint_phase: seq={} total_ms={} lock_wait_ms={} snapshot_ms={} idempotency_ms={} write_checkpoint_ms={} segment_scan_ms={} manifest_ms={} checkpoint_bytes={} retained_segments={}",
checkpoint_seq,
checkpoint_total_elapsed.as_millis(),
checkpoint_lock_wait.as_millis(),
snapshot_elapsed.as_millis(),
idempotency_elapsed.as_millis(),
write_checkpoint_elapsed.as_millis(),
segments_elapsed.as_millis(),
manifest_elapsed.as_millis(),
checkpoint_bytes,
segments.len()
);
eprintln!(
"pipeline_profile recovery_phase: reopen_ms={} startup_recovery_micros={} startup_recovered_seq={} durable_head_seq={} visible_head_seq={}",
reopen_elapsed.as_millis(),
reopen_metrics.startup_recovery_micros,
reopen_metrics.startup_recovered_seq,
reopen_metrics.durable_head_seq,
reopen_metrics.visible_head_seq
);
}
#[tokio::test]
async fn preflight_uses_instance_config_limits() {
let dir = tempdir().expect("temp");
let config = AedbConfig {
max_kv_value_bytes: 4,
..AedbConfig::default()
};
let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
db.create_project("p").await.expect("project");
let result = db
.preflight(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"k".to_vec(),
value: b"too-large".to_vec(),
})
.await;
assert!(
matches!(result, PreflightResult::Err { reason } if reason.contains("value too large"))
);
}
#[test]
fn mutation_write_keys_table_row_variants() {
use crate::commit::tx::WriteKey;
let row = Row {
values: vec![Value::Integer(7), Value::Text("a".into())],
};
let pk = vec![Value::Integer(7)];
let insert = Mutation::Insert {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
primary_key: pk.clone(),
row: row.clone(),
};
let keys = insert.write_keys();
assert_eq!(keys.len(), 1);
assert!(matches!(
&keys[0],
WriteKey::TableRow { project_id, scope_id, table_name, primary_key }
if project_id == "p" && scope_id == "s" && table_name == "t" && primary_key == &pk
));
let delete = Mutation::Delete {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
primary_key: pk.clone(),
};
let keys = delete.write_keys();
assert!(matches!(&keys[0], WriteKey::TableRow { primary_key, .. } if primary_key == &pk));
let batch = Mutation::InsertBatch {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
rows: vec![row.clone()],
};
let keys = batch.write_keys();
assert!(matches!(
&keys[0],
WriteKey::TableRange { table_name, .. } if table_name == "t"
));
}
#[test]
fn read_set_intersects_positive_and_negative_cases() {
use crate::commit::tx::{
ReadBound, ReadKey, ReadRange, ReadRangeEntry, ReadSet, ReadSetEntry, WriteKey,
};
let mut rs = ReadSet::default();
rs.points.push(ReadSetEntry {
key: ReadKey::TableRow {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
primary_key: vec![Value::Integer(1)],
},
version_at_read: 0,
});
rs.ranges.push(ReadRangeEntry {
range: ReadRange::KvRange {
project_id: "p".into(),
scope_id: "s".into(),
start: ReadBound::Included(b"a".to_vec()),
end: ReadBound::Excluded(b"c".to_vec()),
},
max_version_at_read: 0,
structural_version_at_read: 0,
});
assert!(rs.intersects(&[WriteKey::TableRow {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
primary_key: vec![Value::Integer(1)],
}]));
assert!(rs.intersects(&[WriteKey::KvKey {
project_id: "p".into(),
scope_id: "s".into(),
key: b"b".to_vec(),
}]));
assert!(rs.intersects(&[WriteKey::ScopeAll {
project_id: "p".into(),
scope_id: "s".into(),
}]));
assert!(!rs.intersects(&[WriteKey::TableRow {
project_id: "p".into(),
scope_id: "other".into(),
table_name: "t".into(),
primary_key: vec![Value::Integer(1)],
}]));
assert!(!rs.intersects(&[WriteKey::TableRow {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "other".into(),
primary_key: vec![Value::Integer(1)],
}]));
assert!(!rs.intersects(&[WriteKey::KvKey {
project_id: "p".into(),
scope_id: "s".into(),
key: b"c".to_vec(),
}]));
assert!(!rs.intersects(&[]));
let rs_table_only = ReadSet {
points: vec![ReadSetEntry {
key: ReadKey::TableRow {
project_id: "p".into(),
scope_id: "s".into(),
table_name: "t".into(),
primary_key: vec![Value::Integer(1)],
},
version_at_read: 0,
}],
ranges: Vec::new(),
};
assert!(!rs_table_only.intersects(&[WriteKey::KvKey {
project_id: "p".into(),
scope_id: "s".into(),
key: b"k".to_vec(),
}]));
}