409 lines
20 KiB
Rust
409 lines
20 KiB
Rust
// file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/persistence.rs
|
|
// version: 3
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum PortResponse {
|
|
Outcome(ksp_store_lib::RawEntityWriteOutcome, ksp_store_lib::RawObservationWriteOutcome),
|
|
Conflict,
|
|
StoreFailure,
|
|
}
|
|
|
|
struct FakePersistencePort {
|
|
calls: std::sync::atomic::AtomicUsize,
|
|
network: ksp_store_lib::RawNetworkId,
|
|
normal_mode_seen: std::sync::atomic::AtomicBool,
|
|
observation_calls: std::sync::atomic::AtomicUsize,
|
|
observation_response: std::option::Option<ksp_store_lib::RawObservationWriteOutcome>,
|
|
response: PortResponse,
|
|
}
|
|
|
|
impl FakePersistencePort {
|
|
fn new(network: ksp_store_lib::RawNetworkId, response: PortResponse) -> Self {
|
|
return Self {
|
|
calls: std::sync::atomic::AtomicUsize::new(0),
|
|
network,
|
|
normal_mode_seen: std::sync::atomic::AtomicBool::new(false),
|
|
observation_calls: std::sync::atomic::AtomicUsize::new(0),
|
|
observation_response: std::option::Option::None,
|
|
response,
|
|
};
|
|
}
|
|
|
|
fn with_observation_response(mut self, observation_response: ksp_store_lib::RawObservationWriteOutcome) -> Self {
|
|
self.observation_response = std::option::Option::Some(observation_response);
|
|
return self;
|
|
}
|
|
}
|
|
|
|
impl crate::RawTransactionIngestPersistencePort for FakePersistencePort {
|
|
fn network_matches(&self, network: &ksp_store_lib::RawNetworkId) -> bool {
|
|
return &self.network == network;
|
|
}
|
|
|
|
fn persist_acquisition<'a>(
|
|
&'a self,
|
|
_transaction: ksp_store_lib::RawTransaction,
|
|
_observation: ksp_store_lib::RawTransactionObservation,
|
|
mode: ksp_store_lib::RawTransactionAcquisitionMode,
|
|
) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawAcquisitionWriteOutcome>> {
|
|
self.calls.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
self.normal_mode_seen.store(mode == ksp_store_lib::RawTransactionAcquisitionMode::Normal, std::sync::atomic::Ordering::Release);
|
|
let response = self.response;
|
|
return std::boxed::Box::pin(async move {
|
|
return match response {
|
|
PortResponse::Outcome(entity, observation) => std::result::Result::Ok(ksp_store_lib::RawAcquisitionWriteOutcome::new(entity, observation)),
|
|
PortResponse::Conflict => std::result::Result::Err(ksp_core_lib::Error::new(ksp_store_lib::ERROR_CODE_RAW_CONFLICT, "synthetic conflict")),
|
|
PortResponse::StoreFailure => std::result::Result::Err(ksp_core_lib::Error::new(
|
|
ksp_core_lib::ErrorCode::new("synthetic_store", "write_failed"),
|
|
"synthetic remote failure text that must not escape",
|
|
)),
|
|
};
|
|
});
|
|
}
|
|
|
|
fn record_observation<'a>(
|
|
&'a self,
|
|
_observation: ksp_store_lib::RawTransactionObservation,
|
|
) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawObservationWriteOutcome>> {
|
|
self.observation_calls.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
let observation_response = self.observation_response;
|
|
let response = self.response;
|
|
return std::boxed::Box::pin(async move {
|
|
if let std::option::Option::Some(outcome) = observation_response {
|
|
return std::result::Result::Ok(outcome);
|
|
}
|
|
return match response {
|
|
PortResponse::Outcome(_, observation) => std::result::Result::Ok(observation),
|
|
PortResponse::Conflict => std::result::Result::Err(ksp_core_lib::Error::new(ksp_store_lib::ERROR_CODE_RAW_CONFLICT, "synthetic conflict")),
|
|
PortResponse::StoreFailure => std::result::Result::Err(ksp_core_lib::Error::new(
|
|
ksp_core_lib::ErrorCode::new("synthetic_store", "write_failed"),
|
|
"synthetic remote failure text that must not escape",
|
|
)),
|
|
};
|
|
});
|
|
}
|
|
}
|
|
|
|
fn acquisition(network_name: &str, signature_byte: u8) -> std::option::Option<ksp_raw_transaction_lib::RawTransactionAcquisition> {
|
|
return acquisition_with_observation_key(network_name, signature_byte, signature_byte);
|
|
}
|
|
|
|
fn acquisition_with_observation_key(
|
|
network_name: &str,
|
|
signature_byte: u8,
|
|
observation_byte: u8,
|
|
) -> std::option::Option<ksp_raw_transaction_lib::RawTransactionAcquisition> {
|
|
return acquisition_with_shape(network_name, signature_byte, observation_byte, 42, std::option::Option::Some(1_700_000_000), "AQID");
|
|
}
|
|
|
|
fn acquisition_with_shape(
|
|
network_name: &str,
|
|
signature_byte: u8,
|
|
observation_byte: u8,
|
|
slot: u64,
|
|
block_time: std::option::Option<i64>,
|
|
transaction_data: &str,
|
|
) -> std::option::Option<ksp_raw_transaction_lib::RawTransactionAcquisition> {
|
|
let network = match ksp_store_lib::RawNetworkId::new(network_name) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let signature = ksp_store_lib::RawTransactionSignature::new([signature_byte; 64]);
|
|
let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64(
|
|
network,
|
|
signature,
|
|
slot,
|
|
block_time,
|
|
transaction_data,
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
);
|
|
let transaction = match ksp_raw_transaction_lib::canonicalize_raw_transaction(material) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let provider = match ksp_store_lib::RawProvenanceCode::new("deterministic-harness") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let protocol = match ksp_store_lib::RawProvenanceCode::new("internal") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let method = match ksp_store_lib::RawProvenanceCode::new("persistence") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_700_000_000_000) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let provenance = ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at);
|
|
return std::option::Option::Some(ksp_raw_transaction_lib::assemble_raw_transaction_acquisition(
|
|
transaction,
|
|
ksp_store_lib::RawObservationKey::new([observation_byte; 32]),
|
|
provenance,
|
|
));
|
|
}
|
|
|
|
fn network(name: &str) -> std::option::Option<ksp_store_lib::RawNetworkId> {
|
|
return match ksp_store_lib::RawNetworkId::new(name) {
|
|
std::result::Result::Ok(value) => std::option::Option::Some(value),
|
|
std::result::Result::Err(_) => std::option::Option::None,
|
|
};
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_normal_mode_classifies_new_idempotent_and_purged_outcomes() {
|
|
let cases = [
|
|
(
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
crate::RawTransactionIngestEntityPersistence::Inserted,
|
|
crate::RawTransactionIngestObservationPersistence::Inserted,
|
|
),
|
|
(
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
|
|
crate::RawTransactionIngestObservationPersistence::Inserted,
|
|
),
|
|
(
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent, ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent),
|
|
crate::RawTransactionIngestEntityPersistence::AlreadyPresent,
|
|
crate::RawTransactionIngestObservationPersistence::AlreadyPresent,
|
|
),
|
|
(
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::SkippedPurged, ksp_store_lib::RawObservationWriteOutcome::NotRecorded),
|
|
crate::RawTransactionIngestEntityPersistence::SkippedPurged,
|
|
crate::RawTransactionIngestObservationPersistence::NotRecorded,
|
|
),
|
|
];
|
|
for (index, (response, expected_entity, expected_observation)) in cases.into_iter().enumerate() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(network, response);
|
|
let acquisition = match acquisition("mainnet", index as u8 + 1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let result = crate::persist_raw_transaction_ingest_acquisition(&port, acquisition).await;
|
|
assert!(result.is_ok());
|
|
let outcome = match result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(outcome.entity(), expected_entity);
|
|
assert_eq!(outcome.observation(), expected_observation);
|
|
assert!(port.normal_mode_seen.load(std::sync::atomic::Ordering::Acquire));
|
|
assert_eq!(port.calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
}
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_content_conflict_maps_to_terminal_worker_code_without_remote_text() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(network, PortResponse::Conflict);
|
|
let acquisition = match acquisition("mainnet", 9) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let result = crate::persist_raw_transaction_ingest_acquisition(&port, acquisition).await;
|
|
let error = match result {
|
|
std::result::Result::Ok(_) => return,
|
|
std::result::Result::Err(value) => value,
|
|
};
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT);
|
|
assert!(error.context().is_empty());
|
|
assert!(!std::format!("{error:?}").contains("synthetic conflict"));
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_store_failure_keeps_only_safe_lower_error_code() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(network, PortResponse::StoreFailure);
|
|
let acquisition = match acquisition("mainnet", 10) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let result = crate::persist_raw_transaction_ingest_acquisition(&port, acquisition).await;
|
|
let error = match result {
|
|
std::result::Result::Ok(_) => return,
|
|
std::result::Result::Err(value) => value,
|
|
};
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED);
|
|
assert_eq!(error.context().len(), 2);
|
|
assert_eq!(error.context()[0].value(), "synthetic_store");
|
|
assert_eq!(error.context()[1].value(), "write_failed");
|
|
assert!(!std::format!("{error:?}").contains("synthetic remote failure text"));
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_rehydrated_or_impossible_store_outcomes_are_runtime_invalid() {
|
|
let cases = [
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Rehydrated, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent),
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::SkippedPurged, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
];
|
|
for (index, response) in cases.into_iter().enumerate() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(network, response);
|
|
let acquisition = match acquisition("mainnet", index as u8 + 20) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let result = crate::persist_raw_transaction_ingest_acquisition(&port, acquisition).await;
|
|
let error = match result {
|
|
std::result::Result::Ok(_) => continue,
|
|
std::result::Result::Err(value) => value,
|
|
};
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
assert_eq!(error.context().len(), 1);
|
|
assert_eq!(error.context()[0].value(), "persistence.store_outcome_invalid");
|
|
}
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_store_network_mismatch_is_rejected_before_write() {
|
|
let network = match network("devnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(
|
|
network,
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
);
|
|
let acquisition = match acquisition("mainnet", 30) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let result = crate::persist_raw_transaction_ingest_acquisition(&port, acquisition).await;
|
|
let error = match result {
|
|
std::result::Result::Ok(_) => return,
|
|
std::result::Result::Err(value) => value,
|
|
};
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
assert_eq!(error.context()[0].value(), "persistence.store_network_mismatch");
|
|
assert_eq!(port.calls.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn v0_3_13_pre_008_same_canonical_identity_uses_atomic_entity_once_then_additional_observation() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(
|
|
network,
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
);
|
|
let convergence = crate::RawTransactionIngestPersistenceConvergence::new(8);
|
|
let first = match acquisition_with_observation_key("mainnet", 41, 1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let second = match acquisition_with_observation_key("mainnet", 41, 2) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let first_outcome = crate::persist_raw_transaction_ingest_converged_acquisition(&port, first, &convergence).await;
|
|
let second_outcome = crate::persist_raw_transaction_ingest_converged_acquisition(&port, second, &convergence).await;
|
|
assert!(first_outcome.is_ok());
|
|
assert!(second_outcome.is_ok());
|
|
assert_eq!(port.calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert_eq!(port.observation_calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
let second_outcome = match second_outcome {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(second_outcome.entity(), crate::RawTransactionIngestEntityPersistence::AlreadyPresent);
|
|
assert_eq!(second_outcome.observation(), crate::RawTransactionIngestObservationPersistence::Inserted);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn v0_3_13_pre_008_replayed_same_observation_is_idempotent_without_raw_resubmission() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(
|
|
network,
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
)
|
|
.with_observation_response(ksp_store_lib::RawObservationWriteOutcome::AlreadyPresent);
|
|
let convergence = crate::RawTransactionIngestPersistenceConvergence::new(8);
|
|
let first = match acquisition_with_observation_key("mainnet", 42, 3) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let second = match acquisition_with_observation_key("mainnet", 42, 3) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let first_result = crate::persist_raw_transaction_ingest_converged_acquisition(&port, first, &convergence).await;
|
|
assert!(first_result.is_ok());
|
|
let second_result = crate::persist_raw_transaction_ingest_converged_acquisition(&port, second, &convergence).await;
|
|
let second_outcome = match second_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(port.calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert_eq!(port.observation_calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert_eq!(second_outcome.observation(), crate::RawTransactionIngestObservationPersistence::AlreadyPresent);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn v0_3_13_pre_009_source_skew_and_provider_disagreement_fail_before_additional_observation() {
|
|
let cases = [
|
|
(43_u64, std::option::Option::Some(1_700_000_000_i64), "AQID"),
|
|
(42_u64, std::option::Option::Some(1_700_000_001_i64), "AQID"),
|
|
(42_u64, std::option::Option::Some(1_700_000_000_i64), "BAUG"),
|
|
];
|
|
for (index, (slot, block_time, transaction_data)) in cases.into_iter().enumerate() {
|
|
let network = match network("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let port = FakePersistencePort::new(
|
|
network,
|
|
PortResponse::Outcome(ksp_store_lib::RawEntityWriteOutcome::Inserted, ksp_store_lib::RawObservationWriteOutcome::Inserted),
|
|
);
|
|
let convergence = crate::RawTransactionIngestPersistenceConvergence::new(8);
|
|
let first = match acquisition_with_shape("mainnet", 51, 10, 42, std::option::Option::Some(1_700_000_000), "AQID") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let divergent = match acquisition_with_shape("mainnet", 51, index as u8 + 20, slot, block_time, transaction_data) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
assert!(crate::persist_raw_transaction_ingest_converged_acquisition(&port, first, &convergence).await.is_ok());
|
|
let conflict = crate::persist_raw_transaction_ingest_converged_acquisition(&port, divergent, &convergence).await;
|
|
let error = match conflict {
|
|
std::result::Result::Ok(_) => return,
|
|
std::result::Result::Err(value) => value,
|
|
};
|
|
assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT);
|
|
assert_eq!(port.calls.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert_eq!(port.observation_calls.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
}
|
|
return;
|
|
}
|