869 lines
39 KiB
Rust
869 lines
39 KiB
Rust
// file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs
|
|
// version: 7
|
|
|
|
struct ActiveTaskGuard {
|
|
active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
|
|
}
|
|
|
|
impl ActiveTaskGuard {
|
|
fn new(active: std::sync::Arc<std::sync::atomic::AtomicUsize>) -> Self {
|
|
active.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
return Self { active };
|
|
}
|
|
}
|
|
|
|
impl std::ops::Drop for ActiveTaskGuard {
|
|
fn drop(&mut self) {
|
|
self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
|
|
return;
|
|
}
|
|
}
|
|
|
|
fn settings(network: &str) -> std::option::Option<crate::RawTransactionIngestSettings> {
|
|
let network = match ksp_store_lib::RawNetworkId::new(network) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let worker_id = match ksp_worker_api::WorkerId::new("raw-ingest-runtime-001") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
return std::option::Option::Some(crate::RawTransactionIngestSettings::with_defaults(network, worker_id));
|
|
}
|
|
|
|
async fn wait_for_active_count(active: &std::sync::Arc<std::sync::atomic::AtomicUsize>, expected: usize) -> bool {
|
|
for _ in 0..64 {
|
|
if active.load(std::sync::atomic::Ordering::Acquire) == expected {
|
|
return true;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
return active.load(std::sync::atomic::Ordering::Acquire) == expected;
|
|
}
|
|
|
|
#[test]
|
|
fn pre_004_current_runtime_is_required_before_spawn() {
|
|
let result = super::current_runtime_handle();
|
|
assert!(result.is_err());
|
|
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().len(), 1);
|
|
assert_eq!(error.context()[0].value(), "start.runtime_unavailable");
|
|
return;
|
|
}
|
|
|
|
#[test]
|
|
fn pre_004_store_network_mismatch_is_rejected_without_echoing_network_values() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let other = match ksp_store_lib::RawNetworkId::new("devnet") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let result = super::validate_store_network(&settings, &other);
|
|
assert!(result.is_err());
|
|
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);
|
|
let rendered = std::format!("{error:?}");
|
|
assert!(!rendered.contains("mainnet"));
|
|
assert!(!rendered.contains("devnet"));
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_004_immediate_stop_is_idempotent_and_reaches_stopped_terminal() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let handle = match super::start_foundation(settings, tokio::runtime::Handle::current(), std::option::Option::None) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(handle.snapshot_source().current().worker_snapshot().state(), ksp_worker_api::WorkerState::Starting);
|
|
assert!(handle.request_stop());
|
|
assert!(!handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
assert!(terminal.is_ok());
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
assert!(terminal.is_terminal());
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_004_running_lifecycle_stops_through_cloned_handle() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let handle = match super::start_foundation(settings, tokio::runtime::Handle::current(), std::option::Option::None) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(handle.snapshot_source().current().worker_snapshot().state(), ksp_worker_api::WorkerState::Running);
|
|
let clone = handle.clone();
|
|
assert!(clone.request_stop());
|
|
assert!(!handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_004_dropping_last_control_handle_causes_private_runtime_exit() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let handle = match super::start_foundation(settings, tokio::runtime::Handle::current(), std::option::Option::None) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let observer = handle.snapshot_source();
|
|
std::mem::drop(handle);
|
|
loop {
|
|
let current = observer.current();
|
|
if current.worker_snapshot().state().is_terminal() {
|
|
assert_eq!(current.worker_snapshot().state(), ksp_worker_api::WorkerState::Stopped);
|
|
return;
|
|
}
|
|
let _changed = observer.wait_for_change(current.worker_snapshot().sequence()).await;
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_005_supervisor_joins_all_cooperative_source_tasks_before_terminal() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let source_active = std::sync::Arc::clone(&active);
|
|
let handle = match super::start_foundation_with_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::None,
|
|
move |children, stop_receiver, _admission_sender| {
|
|
for _ in 0..3 {
|
|
let active = std::sync::Arc::clone(&source_active);
|
|
let mut child_stop = stop_receiver.clone();
|
|
let _abort_handle = children.spawn(async move {
|
|
let _guard = ActiveTaskGuard::new(active);
|
|
loop {
|
|
let changed = child_stop.changed().await;
|
|
if changed.is_err() || *child_stop.borrow() {
|
|
return std::result::Result::Ok(());
|
|
}
|
|
}
|
|
});
|
|
}
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_active_count(&active, 3).await);
|
|
assert!(handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_child_on_stop() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let completed = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let source_active = std::sync::Arc::clone(&active);
|
|
let source_completed = std::sync::Arc::clone(&completed);
|
|
let handle = match super::start_foundation_with_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::None,
|
|
move |children, stop_receiver, _admission_sender| {
|
|
let completed = std::sync::Arc::clone(&source_completed);
|
|
let _completed_abort_handle = children.spawn(async move {
|
|
completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
return std::result::Result::Ok(());
|
|
});
|
|
let active = std::sync::Arc::clone(&source_active);
|
|
let mut child_stop = stop_receiver.clone();
|
|
let _active_abort_handle = children.spawn(async move {
|
|
let _guard = ActiveTaskGuard::new(active);
|
|
loop {
|
|
let changed = child_stop.changed().await;
|
|
if changed.is_err() || *child_stop.borrow() {
|
|
return std::result::Result::Ok(());
|
|
}
|
|
}
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_active_count(&active, 1).await);
|
|
for _ in 0..64 {
|
|
if completed.load(std::sync::atomic::Ordering::Acquire) == 1 {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
assert_eq!(completed.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert!(handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum RuntimePortResponse {
|
|
Conflict,
|
|
StoreFailure,
|
|
StoreFailureBlocked,
|
|
SuccessBlocked,
|
|
}
|
|
|
|
struct RuntimePortActiveGuard<'a> {
|
|
active: &'a std::sync::atomic::AtomicUsize,
|
|
}
|
|
|
|
impl std::ops::Drop for RuntimePortActiveGuard<'_> {
|
|
fn drop(&mut self) {
|
|
self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
|
|
return;
|
|
}
|
|
}
|
|
|
|
struct RuntimePersistencePort {
|
|
active: std::sync::atomic::AtomicUsize,
|
|
completed: std::sync::atomic::AtomicUsize,
|
|
max_active: std::sync::atomic::AtomicUsize,
|
|
network: ksp_store_lib::RawNetworkId,
|
|
normal_mode_seen: std::sync::atomic::AtomicBool,
|
|
notify: tokio::sync::Notify,
|
|
released: std::sync::atomic::AtomicBool,
|
|
response: RuntimePortResponse,
|
|
}
|
|
|
|
impl RuntimePersistencePort {
|
|
fn new(network: ksp_store_lib::RawNetworkId, response: RuntimePortResponse, released: bool) -> Self {
|
|
return Self {
|
|
active: std::sync::atomic::AtomicUsize::new(0),
|
|
completed: std::sync::atomic::AtomicUsize::new(0),
|
|
max_active: std::sync::atomic::AtomicUsize::new(0),
|
|
network,
|
|
normal_mode_seen: std::sync::atomic::AtomicBool::new(false),
|
|
notify: tokio::sync::Notify::new(),
|
|
released: std::sync::atomic::AtomicBool::new(released),
|
|
response,
|
|
};
|
|
}
|
|
|
|
fn release(&self) {
|
|
self.released.store(true, std::sync::atomic::Ordering::Release);
|
|
self.notify.notify_waiters();
|
|
return;
|
|
}
|
|
|
|
fn update_max_active(&self, current: usize) {
|
|
let mut observed = self.max_active.load(std::sync::atomic::Ordering::Acquire);
|
|
while current > observed {
|
|
match self.max_active.compare_exchange_weak(observed, current, std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire) {
|
|
std::result::Result::Ok(_) => return,
|
|
std::result::Result::Err(value) => observed = value,
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
|
|
impl crate::RawTransactionIngestPersistencePort for RuntimePersistencePort {
|
|
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.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 {
|
|
if matches!(response, RuntimePortResponse::StoreFailureBlocked | RuntimePortResponse::SuccessBlocked) {
|
|
let current = self.active.fetch_add(1, std::sync::atomic::Ordering::AcqRel) + 1;
|
|
let _active_guard = RuntimePortActiveGuard { active: &self.active };
|
|
self.update_max_active(current);
|
|
while !self.released.load(std::sync::atomic::Ordering::Acquire) {
|
|
self.notify.notified().await;
|
|
}
|
|
if matches!(response, RuntimePortResponse::SuccessBlocked) {
|
|
self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
return std::result::Result::Ok(ksp_store_lib::RawAcquisitionWriteOutcome::new(
|
|
ksp_store_lib::RawEntityWriteOutcome::Inserted,
|
|
ksp_store_lib::RawObservationWriteOutcome::Inserted,
|
|
));
|
|
}
|
|
}
|
|
if matches!(response, RuntimePortResponse::Conflict) {
|
|
self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(ksp_store_lib::ERROR_CODE_RAW_CONFLICT, "synthetic conflict"));
|
|
}
|
|
self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
return std::result::Result::Err(ksp_core_lib::Error::new(
|
|
ksp_core_lib::ErrorCode::new("synthetic_store", "write_failed"),
|
|
"synthetic store failure",
|
|
));
|
|
});
|
|
}
|
|
}
|
|
|
|
fn runtime_ingress(network: &ksp_store_lib::RawNetworkId, signature_byte: u8) -> std::option::Option<crate::RawTransactionIngress> {
|
|
let signature = ksp_store_lib::RawTransactionSignature::new([signature_byte; 64]);
|
|
let material = ksp_raw_transaction_lib::RawTransactionMaterial::binary_base64(
|
|
network.clone(),
|
|
signature,
|
|
42,
|
|
std::option::Option::Some(1_700_000_000),
|
|
"AQID",
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
ksp_raw_transaction_lib::RawTransactionWireField::Omitted,
|
|
);
|
|
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("runtime") {
|
|
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(crate::RawTransactionIngress { material, network: network.clone(), provenance, source_key: [signature_byte; 32] });
|
|
}
|
|
|
|
fn settings_with_persistence_concurrency(concurrency: usize) -> std::option::Option<crate::RawTransactionIngestSettings> {
|
|
return settings_with_runtime_limits(8, concurrency, std::time::Duration::from_secs(5));
|
|
}
|
|
|
|
fn settings_with_runtime_limits(
|
|
admission_queue_capacity: usize,
|
|
persistence_concurrency: usize,
|
|
shutdown_drain_timeout: std::time::Duration,
|
|
) -> std::option::Option<crate::RawTransactionIngestSettings> {
|
|
let network = match ksp_store_lib::RawNetworkId::new("mainnet") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
let worker_id = match ksp_worker_api::WorkerId::new("raw-ingest-runtime-persistence-001") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::option::Option::None,
|
|
};
|
|
return match crate::RawTransactionIngestSettings::new(network, worker_id, admission_queue_capacity, persistence_concurrency, shutdown_drain_timeout) {
|
|
std::result::Result::Ok(value) => std::option::Option::Some(value),
|
|
std::result::Result::Err(_) => std::option::Option::None,
|
|
};
|
|
}
|
|
|
|
async fn wait_for_completed(port: &std::sync::Arc<RuntimePersistencePort>, expected: usize) -> bool {
|
|
for _ in 0..128 {
|
|
if port.completed.load(std::sync::atomic::Ordering::Acquire) == expected {
|
|
return true;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
return port.completed.load(std::sync::atomic::Ordering::Acquire) == expected;
|
|
}
|
|
|
|
async fn wait_for_max_active(port: &std::sync::Arc<RuntimePersistencePort>, expected: usize) -> bool {
|
|
for _ in 0..128 {
|
|
if port.max_active.load(std::sync::atomic::Ordering::Acquire) == expected {
|
|
return true;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
return port.max_active.load(std::sync::atomic::Ordering::Acquire) == expected;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_runtime_bounds_in_flight_store_persistence_to_configured_concurrency() {
|
|
let settings = match settings_with_persistence_concurrency(2) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, false));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
for signature_byte in 1..=4 {
|
|
let ingress = match runtime_ingress(&network, signature_byte) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_max_active(&port, 2).await);
|
|
assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 2);
|
|
assert_eq!(port.max_active.load(std::sync::atomic::Ordering::Acquire), 2);
|
|
port.release();
|
|
assert!(wait_for_completed(&port, 4).await);
|
|
assert!(port.normal_mode_seen.load(std::sync::atomic::Ordering::Acquire));
|
|
assert_eq!(port.max_active.load(std::sync::atomic::Ordering::Acquire), 2);
|
|
assert!(handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_content_conflict_becomes_terminal_after_private_drain() {
|
|
let settings = match settings_with_persistence_concurrency(1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::Conflict, true));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
let ingress = match runtime_ingress(&network, 11) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let source = handle.snapshot_source();
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT));
|
|
assert_eq!(port.completed.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
let snapshot = source.current();
|
|
assert_eq!(snapshot.content_conflict_total(), 1);
|
|
assert_eq!(snapshot.store_failure_total(), 0);
|
|
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_007_store_failure_becomes_terminal_without_exposing_remote_error_text() {
|
|
let settings = match settings_with_persistence_concurrency(1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::StoreFailure, true));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
let ingress = match runtime_ingress(&network, 12) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED));
|
|
assert_eq!(port.completed.load(std::sync::atomic::Ordering::Acquire), 1);
|
|
assert!(!std::format!("{handle:?}").contains("synthetic store failure"));
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_008_runtime_snapshots_count_successful_pipeline_and_retain_terminal() {
|
|
let settings = match settings_with_persistence_concurrency(2) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, true));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
for signature_byte in 21..=23 {
|
|
let ingress = match runtime_ingress(&network, signature_byte) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let concrete_source = handle.snapshot_source();
|
|
let common_source = handle.worker_snapshot_source();
|
|
assert!(wait_for_completed(&port, 3).await);
|
|
assert!(handle.request_stop());
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
let snapshot = concrete_source.current();
|
|
assert_eq!(snapshot.worker_snapshot().state(), ksp_worker_api::WorkerState::Stopped);
|
|
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Healthy);
|
|
assert_eq!(snapshot.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle);
|
|
assert_eq!(snapshot.admitted_total(), 3);
|
|
assert_eq!(snapshot.canonicalized_total(), 3);
|
|
assert_eq!(snapshot.persisted_total(), 3);
|
|
assert_eq!(snapshot.entity_inserted_total(), 3);
|
|
assert_eq!(snapshot.entity_already_present_total(), 0);
|
|
assert_eq!(snapshot.entity_skipped_purged_total(), 0);
|
|
assert_eq!(snapshot.observation_inserted_total(), 3);
|
|
assert_eq!(snapshot.observation_already_present_total(), 0);
|
|
assert_eq!(snapshot.content_conflict_total(), 0);
|
|
assert_eq!(snapshot.store_failure_total(), 0);
|
|
assert_eq!(snapshot.source_failure_total(), 0);
|
|
assert_eq!(snapshot.backpressure_wait_total(), 0);
|
|
assert_eq!(snapshot.admission_queue_depth(), 0);
|
|
assert_eq!(snapshot.in_flight_persistence(), 0);
|
|
let common = ksp_worker_api::WorkerSnapshotSource::current(&common_source);
|
|
assert_eq!(&common, snapshot.worker_snapshot());
|
|
let retained = concrete_source.wait_for_change(snapshot.worker_snapshot().sequence()).await;
|
|
assert_eq!(retained, snapshot);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_008_fault_snapshots_count_classified_store_failures_and_project_unhealthy() {
|
|
let settings = match settings_with_persistence_concurrency(1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::StoreFailure, true));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
let ingress = match runtime_ingress(&network, 24) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let source = handle.snapshot_source();
|
|
let terminal = handle.wait_terminal().await;
|
|
let terminal = match terminal {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED));
|
|
let snapshot = source.current();
|
|
assert_eq!(snapshot.worker_snapshot().state(), terminal);
|
|
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy);
|
|
assert_eq!(snapshot.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle);
|
|
assert_eq!(snapshot.admitted_total(), 1);
|
|
assert_eq!(snapshot.canonicalized_total(), 1);
|
|
assert_eq!(snapshot.persisted_total(), 0);
|
|
assert_eq!(snapshot.store_failure_total(), 1);
|
|
assert_eq!(snapshot.content_conflict_total(), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_009_source_failure_is_counted_and_late_stop_cannot_replace_terminal_fault() {
|
|
let settings = match settings("mainnet") {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let handle = match super::start_foundation_with_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
move |children: &mut tokio::task::JoinSet<ksp_core_lib::Result<()>>, _stop_receiver, _admission_sender| {
|
|
let _abort_handle = children.spawn(async move { std::result::Result::Err(crate::runtime_error("test.source_failed")) });
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
let source = handle.snapshot_source();
|
|
let terminal = match handle.wait_terminal().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED));
|
|
let snapshot = source.current();
|
|
assert_eq!(snapshot.source_failure_total(), 1);
|
|
assert_eq!(snapshot.backpressure_wait_total(), 0);
|
|
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy);
|
|
assert!(!handle.request_stop());
|
|
let retained = match handle.wait_terminal().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(retained, terminal);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_009_stop_does_not_hide_store_failure_observed_during_bounded_drain() {
|
|
let settings = match settings_with_persistence_concurrency(1) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::StoreFailureBlocked, false));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
let ingress = match runtime_ingress(&network, 41) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_max_active(&port, 1).await);
|
|
assert!(handle.request_stop());
|
|
port.release();
|
|
let terminal = match handle.wait_terminal().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED));
|
|
let snapshot = handle.snapshot_source().current();
|
|
assert_eq!(snapshot.store_failure_total(), 1);
|
|
assert_eq!(snapshot.in_flight_persistence(), 0);
|
|
assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_009_saturation_is_observable_without_drop_or_unbounded_admission() {
|
|
let settings = match settings_with_runtime_limits(1, 1, std::time::Duration::from_secs(5)) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, false));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let source_stage = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let source_stage_for_task = source_stage.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
for signature_byte in 51..=53 {
|
|
let ingress = match runtime_ingress(&network, signature_byte) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
source_stage_for_task.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
|
}
|
|
return std::result::Result::Ok(());
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_max_active(&port, 1).await);
|
|
for _ in 0..128 {
|
|
if source_stage.load(std::sync::atomic::Ordering::Acquire) == 2 {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
assert_eq!(source_stage.load(std::sync::atomic::Ordering::Acquire), 2);
|
|
port.release();
|
|
assert!(wait_for_completed(&port, 3).await);
|
|
for _ in 0..128 {
|
|
if source_stage.load(std::sync::atomic::Ordering::Acquire) == 3 {
|
|
break;
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
assert_eq!(source_stage.load(std::sync::atomic::Ordering::Acquire), 3);
|
|
assert!(handle.request_stop());
|
|
let terminal = match handle.wait_terminal().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
|
|
let snapshot = handle.snapshot_source().current();
|
|
assert_eq!(snapshot.persisted_total(), 3);
|
|
assert_eq!(snapshot.source_failure_total(), 0);
|
|
assert!(snapshot.backpressure_wait_total() >= 1);
|
|
assert_eq!(snapshot.admission_queue_depth(), 0);
|
|
assert_eq!(snapshot.in_flight_persistence(), 0);
|
|
return;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn pre_009_drain_timeout_aborts_and_joins_all_owned_source_and_persistence_tasks() {
|
|
let settings = match settings_with_runtime_limits(1, 1, crate::MIN_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return,
|
|
};
|
|
let network = settings.network().clone();
|
|
let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, false));
|
|
let runtime_port: super::PersistencePort = port.clone();
|
|
let source_active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let source_active_for_task = source_active.clone();
|
|
let handle = match super::start_foundation_with_port_and_source_spawner(
|
|
settings,
|
|
tokio::runtime::Handle::current(),
|
|
std::option::Option::Some(runtime_port),
|
|
move |children, _stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
let ingress = match runtime_ingress(&network, 61) {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")),
|
|
};
|
|
if admission_sender.send(ingress).await.is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("test.source_failed"));
|
|
}
|
|
let _guard = ActiveTaskGuard::new(source_active_for_task);
|
|
return std::future::pending::<ksp_core_lib::Result<()>>().await;
|
|
});
|
|
},
|
|
) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert!(wait_for_max_active(&port, 1).await);
|
|
assert!(wait_for_active_count(&source_active, 1).await);
|
|
assert!(handle.request_stop());
|
|
let terminal = match handle.wait_terminal().await {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return,
|
|
};
|
|
assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT));
|
|
let snapshot = handle.snapshot_source().current();
|
|
assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy);
|
|
assert_eq!(snapshot.admission_queue_depth(), 0);
|
|
assert_eq!(snapshot.in_flight_persistence(), 0);
|
|
assert_eq!(source_active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
assert_eq!(port.completed.load(std::sync::atomic::Ordering::Acquire), 0);
|
|
return;
|
|
}
|