use super::*;
use holo_hash::DnaHash;
use std::sync::Arc;
fn dht_id() -> Dht {
Dht::new(Arc::new(DnaHash::from_raw_36(vec![0u8; 36])))
}
fn agent(seed: u8) -> AgentPubKey {
AgentPubKey::from_raw_36(vec![seed; 36])
}
#[tokio::test]
async fn delete_live_ephemeral_scheduled_functions_roundtrip() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let author = agent(1);
store
.db
.upsert_scheduled_function(InsertScheduledFunction {
author: &author,
zome_name: "z",
scheduled_fn: "f",
maybe_schedule: b"",
start_at: Timestamp::from_micros(50),
end_at: Timestamp::from_micros(300),
ephemeral: true,
})
.await
.unwrap();
let deleted = store
.delete_live_ephemeral_scheduled_functions(&author, Timestamp::from_micros(100))
.await
.unwrap();
assert_eq!(deleted, 1);
let deleted2 = store
.delete_live_ephemeral_scheduled_functions(&author, Timestamp::from_micros(100))
.await
.unwrap();
assert_eq!(deleted2, 0);
}
#[tokio::test]
async fn upsert_scheduled_function_none_schedule_writes_ephemeral_row() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let author = agent(2);
store
.db
.upsert_scheduled_function(InsertScheduledFunction {
author: &author,
zome_name: "z",
scheduled_fn: "f",
maybe_schedule: b"",
start_at: Timestamp::from_micros(0),
end_at: Timestamp::from_micros(100),
ephemeral: false,
})
.await
.unwrap();
let rows = store
.upsert_scheduled_function(
&author,
&ScheduledFn::new("z".into(), "f".into()),
&None,
Timestamp::from_micros(50),
)
.await
.unwrap();
assert_eq!(rows, 1, "None upsert should write exactly one row");
let deleted = store
.db
.delete_live_ephemeral_scheduled_functions(&author, Timestamp::from_micros(60))
.await
.unwrap();
assert_eq!(
deleted, 1,
"row should be ephemeral with start_at <= 60 after None upsert"
);
}
#[tokio::test]
async fn mark_chain_op_receipts_complete_no_row() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op_hash = DhtOpHash::from_raw_36(vec![1u8; 36]);
let err = store
.mark_chain_op_receipts_complete(&op_hash)
.await
.unwrap_err();
assert!(matches!(err, DhtStoreError::ChainOpPublishMissing));
}
#[tokio::test]
async fn purge_all_empties_every_table() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let author = AgentPubKey::from_raw_36(vec![1u8; 36]);
store
.db
.upsert_scheduled_function(InsertScheduledFunction {
author: &author,
zome_name: "z",
scheduled_fn: "f",
maybe_schedule: b"",
start_at: Timestamp::from_micros(1),
end_at: Timestamp::from_micros(2),
ephemeral: true,
})
.await
.unwrap();
store.purge_all().await.unwrap();
let pool = store.db.pool();
for (table, sql) in [
("Action", "SELECT COUNT(*) FROM Action"),
("Entry", "SELECT COUNT(*) FROM Entry"),
("PrivateEntry", "SELECT COUNT(*) FROM PrivateEntry"),
("CapGrant", "SELECT COUNT(*) FROM CapGrant"),
("CapClaim", "SELECT COUNT(*) FROM CapClaim"),
("ChainLock", "SELECT COUNT(*) FROM ChainLock"),
("LimboChainOp", "SELECT COUNT(*) FROM LimboChainOp"),
("LimboWarrant", "SELECT COUNT(*) FROM LimboWarrant"),
("ChainOp", "SELECT COUNT(*) FROM ChainOp"),
("ChainOpPublish", "SELECT COUNT(*) FROM ChainOpPublish"),
(
"ValidationReceipt",
"SELECT COUNT(*) FROM ValidationReceipt",
),
("Warrant", "SELECT COUNT(*) FROM Warrant"),
("WarrantPublish", "SELECT COUNT(*) FROM WarrantPublish"),
("Link", "SELECT COUNT(*) FROM Link"),
("DeletedLink", "SELECT COUNT(*) FROM DeletedLink"),
("UpdatedRecord", "SELECT COUNT(*) FROM UpdatedRecord"),
("DeletedRecord", "SELECT COUNT(*) FROM DeletedRecord"),
(
"ScheduledFunction",
"SELECT COUNT(*) FROM ScheduledFunction",
),
] {
let count: i64 = sqlx::query_scalar(sql).fetch_one(pool).await.unwrap();
assert_eq!(count, 0, "{table} not empty after purge_all");
}
}
fn build_test_store_record_op_hashed(seed: u8) -> DhtOpHashed {
use holo_hash::{ActionHash, EntryHash};
use holochain_serialized_bytes::UnsafeBytes;
use holochain_types::dht_op::{ChainOp, DhtOp, DhtOpHashed};
use holochain_types::prelude::{AppEntryBytes, Entry, RecordEntry, Signature};
use holochain_zome_types::action::{Action, Create, EntryType};
use holochain_zome_types::entry_def::EntryVisibility;
use holochain_zome_types::prelude::AppEntryDef;
let author = AgentPubKey::from_raw_36(vec![seed; 36]);
let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
let entry = Entry::App(AppEntryBytes(
holochain_serialized_bytes::SerializedBytes::from(UnsafeBytes::from(vec![seed; 8])),
));
let sig = Signature::from([seed; 64]);
let action = Action::Create(Create {
author: author.clone(),
timestamp: Timestamp::from_micros(seed as i64 * 1000),
action_seq: 1,
prev_action: ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36]),
entry_type: EntryType::App(AppEntryDef::new(
0.into(),
0.into(),
EntryVisibility::Public,
)),
entry_hash: entry_hash.clone(),
weight: Default::default(),
});
let op = DhtOp::ChainOp(Box::new(ChainOp::StoreRecord(
sig,
action,
RecordEntry::Present(entry),
)));
DhtOpHashed::from_content_sync(op)
}
fn build_test_warrant_op_hashed(seed: u8) -> DhtOpHashed {
use holochain_types::dht_op::{DhtOp, DhtOpHashed};
use holochain_types::warrant::WarrantOp;
use holochain_zome_types::op::ChainOpType;
use holochain_zome_types::prelude::{
ChainIntegrityWarrant, Signature, SignedWarrant, Warrant, WarrantProof,
};
let action_author = AgentPubKey::from_raw_36(vec![seed; 36]);
let warrantee = AgentPubKey::from_raw_36(vec![seed.wrapping_add(50); 36]);
let action_hash = holo_hash::ActionHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
let warrant = SignedWarrant::new(
Warrant::new(
WarrantProof::ChainIntegrity(ChainIntegrityWarrant::InvalidChainOp {
action_author: action_author.clone(),
action: (action_hash, Signature::from([seed; 64])),
chain_op_type: ChainOpType::StoreRecord,
}),
AgentPubKey::from_raw_36(vec![seed.wrapping_add(10); 36]),
Timestamp::from_micros(seed as i64 * 1000),
warrantee,
),
Signature::from([seed.wrapping_add(1); 64]),
);
let op = DhtOp::WarrantOp(Box::new(WarrantOp::from(warrant)));
DhtOpHashed::from_content_sync(op)
}
#[tokio::test]
async fn record_incoming_ops_inserts_limbo_chain_op() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(1);
let op_hash = op.as_hash().clone();
let action_hash = {
let action = op.as_content().as_chain_op().unwrap().action();
holo_hash::ActionHash::with_data_sync(&action)
};
store.record_incoming_ops(vec![op]).await.unwrap();
let found = store.db.as_ref().get_action(action_hash).await.unwrap();
assert!(
found.is_some(),
"Action row not found after record_incoming_ops"
);
let row = store
.db
.as_ref()
.get_limbo_chain_op(op_hash)
.await
.unwrap()
.expect("LimboChainOp row not found");
assert_eq!(row.require_receipt, 1, "require_receipt should be 1 (true)");
assert!(row.serialized_size > 0, "serialized_size should be > 0");
}
#[tokio::test]
async fn record_incoming_ops_inserts_limbo_warrant() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let warrant_op = build_test_warrant_op_hashed(1);
let op_hash = warrant_op.as_hash().clone();
store.record_incoming_ops(vec![warrant_op]).await.unwrap();
let row = store.db.as_ref().get_limbo_warrant(op_hash).await.unwrap();
assert!(
row.is_some(),
"LimboWarrant row not found after record_incoming_ops"
);
let row = row.unwrap();
assert!(row.serialized_size > 0, "serialized_size should be > 0");
}
#[tokio::test]
async fn record_sys_validation_outcome_chain_op() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(10);
let op_hash = op.as_hash().clone();
store.record_incoming_ops(vec![op]).await.unwrap();
let row_before = store
.db
.as_ref()
.get_limbo_chain_op(op_hash.clone())
.await
.unwrap()
.expect("LimboChainOp row not found after seed");
assert_eq!(row_before.sys_validation_status, None);
store
.record_chain_op_sys_validation_outcomes(vec![(op_hash.clone(), SysOutcome::Accepted)])
.await
.unwrap();
let row = store
.db
.as_ref()
.get_limbo_chain_op(op_hash)
.await
.unwrap()
.expect("LimboChainOp row not found after update");
assert_eq!(row.sys_validation_status, Some(1));
}
#[tokio::test]
async fn record_sys_validation_outcome_warrant() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_warrant_op_hashed(20);
let op_hash = op.as_hash().clone();
store.record_incoming_ops(vec![op]).await.unwrap();
let row_before = store
.db
.as_ref()
.get_limbo_warrant(op_hash.clone())
.await
.unwrap()
.expect("LimboWarrant row not found after seed");
assert_eq!(row_before.sys_validation_status, None);
store
.record_warrant_sys_validation_outcomes(vec![(op_hash.clone(), SysOutcome::Rejected)])
.await
.unwrap();
let row = store
.db
.as_ref()
.get_limbo_warrant(op_hash)
.await
.unwrap()
.expect("LimboWarrant row not found after update");
assert_eq!(row.sys_validation_status, Some(2));
}
#[tokio::test]
async fn record_app_validation_outcome_accepted() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(11);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.app_validation_status, None);
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.app_validation_status, Some(1));
}
#[tokio::test]
async fn record_app_validation_outcome_rejected() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(12);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Rejected)])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.app_validation_status, Some(2));
}
#[tokio::test]
async fn record_incoming_ops_dedupes_on_conflict() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(2);
let op_hash = op.as_hash().clone();
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store.record_incoming_ops(vec![op]).await.unwrap();
let row = store.db.as_ref().get_limbo_chain_op(op_hash).await.unwrap();
assert!(row.is_some(), "LimboChainOp row should still be present");
}
#[tokio::test]
async fn integrate_ready_ops_promotes_ready_chain_op() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(50);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
let promoted = store
.integrate_ready_ops(Timestamp::from_micros(999))
.await
.unwrap();
assert_eq!(promoted, vec![op.as_hash().clone()]);
assert!(store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.is_none());
let row = store
.db()
.as_ref()
.get_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.when_integrated, 999);
assert_eq!(
row.validation_status,
i64::from(holochain_zome_types::dht_v2::RecordValidity::Accepted)
);
}
#[tokio::test]
async fn integrate_ready_ops_skips_unready() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(51);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
let promoted = store
.integrate_ready_ops(Timestamp::from_micros(999))
.await
.unwrap();
assert!(promoted.is_empty());
assert!(store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.is_some());
assert!(store
.db()
.as_ref()
.get_chain_op(op.as_hash().clone())
.await
.unwrap()
.is_none());
}
#[tokio::test]
async fn integrate_ready_ops_promotes_warrant() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let warrant = build_test_warrant_op_hashed(52);
store
.record_incoming_ops(vec![warrant.clone()])
.await
.unwrap();
store
.record_warrant_sys_validation_outcomes(vec![(
warrant.as_hash().clone(),
SysOutcome::Accepted,
)])
.await
.unwrap();
let promoted = store
.integrate_ready_ops(Timestamp::from_micros(999))
.await
.unwrap();
assert_eq!(promoted, vec![warrant.as_hash().clone()]);
assert!(store
.db()
.as_ref()
.get_limbo_warrant(warrant.as_hash().clone())
.await
.unwrap()
.is_none());
assert!(store
.db()
.as_ref()
.get_warrant(warrant.as_hash().clone())
.await
.unwrap()
.is_some());
}
#[tokio::test]
async fn record_validation_receipt_inserts_and_counts() {
use holochain_types::prelude::Signature;
use holochain_types::prelude::{SignedValidationReceipt, ValidationReceipt, ValidationStatus};
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(60);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
store
.integrate_ready_ops(Timestamp::from_micros(1))
.await
.unwrap();
let receipt = SignedValidationReceipt {
receipt: ValidationReceipt {
dht_op_hash: op.as_hash().clone(),
validation_status: ValidationStatus::Valid,
validators: vec![AgentPubKey::from_raw_36(vec![5u8; 36])],
when_integrated: Timestamp::from_micros(1),
},
validators_signatures: vec![Signature([0u8; 64])],
};
let count = store.record_validation_receipt(&receipt).await.unwrap();
assert_eq!(count, 1);
let count = store.record_validation_receipt(&receipt).await.unwrap();
assert_eq!(count, 1);
}
#[tokio::test]
async fn apply_countersigning_success_clears_withhold() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(80);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
store
.integrate_ready_ops(Timestamp::from_micros(1))
.await
.unwrap();
store
.db()
.insert_chain_op_publish(op.as_hash(), None, None, Some(true))
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_chain_op_publish(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.withhold_publish, Some(1));
store
.clear_op_withhold_publishes(vec![op.as_hash().clone()])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_chain_op_publish(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.withhold_publish, None);
}
#[tokio::test]
async fn apply_countersigning_success_no_op_when_row_absent() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let dummy_hash = DhtOpHash::from_raw_36(vec![0xAA; 36]);
store
.clear_op_withhold_publishes(vec![dummy_hash])
.await
.unwrap();
}
#[tokio::test]
async fn record_published_op_hashes_updates_publish_time() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(90);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
store
.integrate_ready_ops(Timestamp::from_micros(1))
.await
.unwrap();
store
.db()
.insert_chain_op_publish(op.as_hash(), None, None, None)
.await
.unwrap();
store
.record_published_op_hashes(vec![op.as_hash().clone()], Timestamp::from_micros(42))
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_chain_op_publish(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.last_publish_time, Some(42));
}
#[tokio::test]
async fn reject_chain_op_rejects_integrated_op() {
use holochain_zome_types::dht_v2::RecordValidity;
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(100);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
store
.integrate_ready_ops(Timestamp::from_micros(1))
.await
.unwrap();
sqlx::query("UPDATE ChainOp SET locally_validated = 0 WHERE hash = ?")
.bind(op.as_hash().get_raw_36())
.execute(store.db().pool())
.await
.unwrap();
store
.reject_chain_ops(vec![op.as_hash().clone()])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.validation_status, i64::from(RecordValidity::Rejected));
}
#[tokio::test]
async fn reject_chain_op_no_op_for_locally_validated_integrated_op() {
use holochain_zome_types::dht_v2::RecordValidity;
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(102);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.record_chain_op_sys_validation_outcomes(vec![(op.as_hash().clone(), SysOutcome::Accepted)])
.await
.unwrap();
store
.record_app_validation_outcomes(vec![(op.as_hash().clone(), AppOutcome::Accepted)])
.await
.unwrap();
store
.integrate_ready_ops(Timestamp::from_micros(1))
.await
.unwrap();
store
.reject_chain_ops(vec![op.as_hash().clone()])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.validation_status, i64::from(RecordValidity::Accepted));
}
#[tokio::test]
async fn reject_chain_op_rejects_limbo_op() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let op = build_test_store_record_op_hashed(101);
store.record_incoming_ops(vec![op.clone()]).await.unwrap();
store
.reject_chain_ops(vec![op.as_hash().clone()])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_limbo_chain_op(op.as_hash().clone())
.await
.unwrap()
.unwrap();
assert_eq!(row.sys_validation_status, Some(2));
assert_eq!(row.app_validation_status, None);
}
#[tokio::test]
async fn record_locally_validated_warrants_inserts_warrant() {
let store = DhtStore::new_test(dht_id()).await.unwrap();
let warrant_op = build_test_warrant_op_hashed(30);
store
.record_locally_validated_warrants(vec![warrant_op.clone()])
.await
.unwrap();
let row = store
.db()
.as_ref()
.get_warrant(warrant_op.as_hash().clone())
.await
.unwrap()
.expect("warrant row missing");
let expected_warrantee = AgentPubKey::from_raw_36(vec![80u8; 36]);
assert_eq!(row.warrantee, expected_warrantee.get_raw_36().to_vec());
}