use aedb::AedbInstance;
use aedb::catalog::DdlOperation;
use aedb::catalog::schema::{ColumnDef, IndexType};
use aedb::catalog::types::{ColumnType, Row, Value};
use aedb::commit::tx::{IdempotencyKey, ReadSet, TransactionEnvelope, WriteClass, WriteIntent};
use aedb::commit::validation::Mutation;
use aedb::config::{AedbConfig, StorageMode};
use aedb::error::AedbError;
use aedb::offline;
use aedb::order_book::{
ExecInstruction, OrderRequest, OrderSide, OrderType, SelfTradePrevention, TimeInForce,
};
use aedb::query::plan::QueryOptions;
use aedb::query::plan::{ConsistencyMode, Query};
use proptest::prelude::*;
use proptest::test_runner::TestCaseError;
use tempfile::tempdir;
fn one_u256() -> [u8; 32] {
let mut out = [0u8; 32];
out[31] = 1;
out
}
proptest! {
#![proptest_config(ProptestConfig {
cases: 8,
max_local_rejects: 0,
.. ProptestConfig::default()
})]
#[test]
fn prop_atomicity_no_partial_apply(
suffix in prop::collection::vec(any::<u8>(), 4..16),
value in prop::collection::vec(any::<u8>(), 1..64),
) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let mut key = b"atomic-prop:".to_vec();
key.extend_from_slice(&suffix);
let err = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: None,
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: key.clone(),
value,
},
Mutation::KvDecU256 {
project_id: "p".into(),
scope_id: "app".into(),
key: b"missing-counter".to_vec(),
amount_be: one_u256(),
},
],
},
base_seq: 0,
})
.await
.expect_err("envelope should fail atomically");
assert!(matches!(err, AedbError::Underflow | AedbError::Validation(_)));
let entry = db
.kv_get_no_auth("p", "app", &key, ConsistencyMode::AtLatest)
.await
.expect("kv read");
prop_assert!(entry.is_none());
Ok(())
});
outcome?;
}
#[test]
fn prop_mixed_table_kv_envelope_is_atomic_when_late_mutation_fails(
suffix in prop::collection::vec(any::<u8>(), 4..16),
table_id in 1i64..10_000,
table_amount in 1i64..1_000_000,
kv_value in prop::collection::vec(any::<u8>(), 1..64),
) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let config = AedbConfig {
storage_mode: StorageMode::DiskBacked,
..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: "ledger".into(),
columns: vec![
ColumnDef {
name: "id".into(),
col_type: ColumnType::Integer,
nullable: false,
},
ColumnDef {
name: "owner".into(),
col_type: ColumnType::Text,
nullable: false,
},
ColumnDef {
name: "amount".into(),
col_type: ColumnType::Integer,
nullable: false,
},
],
primary_key: vec!["id".into()],
}))
.await
.expect("table");
db.commit(Mutation::Ddl(DdlOperation::CreateIndex {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "ledger".into(),
index_name: "by_owner".into(),
if_not_exists: false,
columns: vec!["owner".into()],
index_type: IndexType::BTree,
partial_filter: None,
}))
.await
.expect("index");
let mut kv_key = b"mixed-atomic:".to_vec();
kv_key.extend_from_slice(&suffix);
let err = db
.commit_envelope(TransactionEnvelope {
caller: None,
idempotency_key: None,
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: kv_key.clone(),
value: kv_value,
},
Mutation::Upsert {
project_id: "p".into(),
scope_id: "app".into(),
table_name: "ledger".into(),
primary_key: vec![Value::Integer(table_id)],
row: Row::from_values(vec![
Value::Integer(table_id),
Value::Text("alice".into()),
Value::Integer(table_amount),
]),
},
Mutation::KvDecU256 {
project_id: "p".into(),
scope_id: "app".into(),
key: b"missing-counter".to_vec(),
amount_be: one_u256(),
},
],
},
base_seq: 0,
})
.await
.expect_err("late failure should abort mixed envelope");
prop_assert!(matches!(err, AedbError::Underflow | AedbError::Validation(_)));
let kv = db
.kv_get_no_auth("p", "app", &kv_key, ConsistencyMode::AtLatest)
.await
.expect("kv read");
prop_assert!(kv.is_none());
let by_pk = db
.query_no_auth(
"p",
"app",
Query::select(&["id", "owner", "amount"])
.from("ledger")
.where_(aedb::query::plan::Expr::Eq("id".into(), Value::Integer(table_id)))
.limit(1),
QueryOptions::default())
.await
.expect("query table by pk");
prop_assert!(by_pk.rows.is_empty());
let by_index = db
.query_no_auth(
"p",
"app",
Query::select(&["id", "owner", "amount"])
.from("ledger")
.where_(aedb::query::plan::Expr::Eq(
"owner".into(),
Value::Text("alice".into()),
))
.limit(10),
QueryOptions::default())
.await
.expect("query table by index");
prop_assert!(by_index.rows.is_empty());
Ok(())
});
outcome?;
}
#[test]
fn prop_idempotency_exactly_once_across_retries(
key_seed in any::<u128>(),
retries in 2u8..6,
payload in prop::collection::vec(any::<u8>(), 1..48),
) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let db = AedbInstance::open_anonymous(AedbConfig::production([7u8; 32]), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let idem = IdempotencyKey(key_seed.to_be_bytes());
let envelope = TransactionEnvelope {
caller: None,
idempotency_key: Some(idem),
write_class: WriteClass::Economic,
assertions: Vec::new(),
read_set: ReadSet::default(),
write_intent: WriteIntent {
mutations: vec![Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key: b"idem-prop".to_vec(),
value: payload,
}],
},
base_seq: 0,
};
let first = db
.commit_envelope(envelope.clone())
.await
.expect("first commit");
for _ in 0..retries {
let again = db
.commit_envelope(envelope.clone())
.await
.expect("idempotent retry");
prop_assert_eq!(again.commit_seq, first.commit_seq);
}
Ok(())
});
outcome?;
}
#[test]
fn prop_replay_determinism_snapshot_parity(
seed in any::<u64>(),
writes in 16usize..96,
) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let dump_a_dir = tempdir().expect("dump a");
let dump_b_dir = tempdir().expect("dump b");
let dump_a = dump_a_dir.path().join("a.aedbdump");
let dump_b = dump_b_dir.path().join("b.aedbdump");
let config = AedbConfig::production([8u8; 32]);
let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
for i in 0..writes {
let key = format!("replay-prop:{seed}:{i}").into_bytes();
let value = ((seed as usize) ^ i).to_be_bytes().to_vec();
db.commit(Mutation::KvSet {
project_id: "p".into(),
scope_id: "app".into(),
key,
value,
})
.await
.expect("commit");
}
db.shutdown().await.expect("shutdown");
let report_a =
offline::export_snapshot_dump(dir.path(), &config, &dump_a).expect("export a");
let report_b =
offline::export_snapshot_dump(dir.path(), &config, &dump_b).expect("export b");
prop_assert_eq!(report_a.current_seq, report_b.current_seq);
prop_assert_eq!(report_a.parity_checksum_hex, report_b.parity_checksum_hex);
Ok(())
});
outcome?;
}
#[test]
fn prop_secure_mode_rejects_unauthenticated_commits(project_suffix in 0u32..10_000) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let db = AedbInstance::open_secure(AedbConfig::production([9u8; 32]), dir.path())
.expect("open secure");
let err = db
.commit(Mutation::Ddl(DdlOperation::CreateProject {
owner_id: None,
if_not_exists: true,
project_id: format!("p-{project_suffix}"),
}))
.await
.expect_err("secure mode must require authenticated caller");
prop_assert!(matches!(err, AedbError::PermissionDenied(_)));
Ok(())
});
outcome?;
}
#[test]
fn prop_order_book_new_rejects_malformed_inputs(
instrument in prop_oneof![
Just(String::new()),
Just(" ".to_string()),
prop::collection::vec(any::<u8>(), 1025..1200)
.prop_map(|v| String::from_utf8_lossy(&v).into_owned()),
],
owner in prop_oneof![
Just(String::new()),
Just("\n\t ".to_string()),
prop::collection::vec(any::<u8>(), 1025..1200)
.prop_map(|v| String::from_utf8_lossy(&v).into_owned()),
],
client_order_id in prop_oneof![
Just(String::new()),
Just(" ".to_string()),
prop::collection::vec(any::<u8>(), 1025..1200)
.prop_map(|v| String::from_utf8_lossy(&v).into_owned()),
],
) {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("runtime");
let outcome: Result<(), TestCaseError> = rt.block_on(async move {
let dir = tempdir().expect("temp dir");
let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
db.create_project("p").await.expect("project");
let err = db
.order_book_new(
"p",
"app",
OrderRequest {
instrument,
client_order_id,
side: OrderSide::Bid,
order_type: OrderType::Limit,
time_in_force: TimeInForce::Gtc,
exec_instructions: ExecInstruction(0),
self_trade_prevention: SelfTradePrevention::None,
price_ticks: 100,
qty_be: one_u256(),
owner,
account: None,
nonce: 1,
price_limit_ticks: None,
},
)
.await
.expect_err("malformed orderbook request must fail validation");
prop_assert!(matches!(err, AedbError::Validation(_)));
Ok(())
});
outcome?;
}
}