v0.3.11-pre.009

crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs
-// version: 2
+// version: 3
 
 fn material(
     network: &str,
@@ -239,3 +239,24 @@ fn pre_006_network_guards_reject_ingress_or_material_mismatch_without_echoing_va
     }
     return;
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_009_full_queue_dequeue_marks_source_neutral_backpressure_observation() {
+    let (network, material) = match material("mainnet", 13, "AQID") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let provenance = match provenance() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let (mut admission, sender) = crate::RawTransactionAdmission::new(1);
+    let ingress = crate::RawTransactionIngress { material, network, provenance, source_key: [13; 32] };
+    assert!(sender.send(ingress).await.is_ok());
+    assert!(admission.receive().await.is_some());
+    assert!(admission.take_backpressure_wait_observed());
+    admission.close();
+    assert!(admission.receive().await.is_none());
+    assert!(!admission.take_backpressure_wait_observed());
+    return;
+}

crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs
-// version: 5
+// version: 6
 
 struct ActiveTaskGuard {
     active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
@@ -169,7 +169,7 @@ async fn pre_005_supervisor_joins_all_cooperative_source_tasks_before_terminal()
                     loop {
                         let changed = child_stop.changed().await;
                         if changed.is_err() || *child_stop.borrow() {
-                            return;
+                            return std::result::Result::Ok(());
                         }
                     }
                 });
@@ -209,7 +209,7 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi
             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;
+                return std::result::Result::Ok(());
             });
             let active = std::sync::Arc::clone(&source_active);
             let mut child_stop = stop_receiver.clone();
@@ -218,7 +218,7 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi
                 loop {
                     let changed = child_stop.changed().await;
                     if changed.is_err() || *child_stop.borrow() {
-                        return;
+                        return std::result::Result::Ok(());
                     }
                 }
             });
@@ -250,9 +250,21 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi
 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,
@@ -310,18 +322,20 @@ impl crate::RawTransactionIngestPersistencePort for RuntimePersistencePort {
         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::SuccessBlocked) {
+            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;
                 }
-                self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
-                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::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);
@@ -369,6 +383,14 @@ fn runtime_ingress(network: &ksp_store_lib::RawNetworkId, signature_byte: u8) ->
 }
 
 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,
@@ -377,7 +399,7 @@ fn settings_with_persistence_concurrency(concurrency: usize) -> std::option::Opt
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return std::option::Option::None,
     };
-    return match crate::RawTransactionIngestSettings::new(network, worker_id, 8, concurrency, std::time::Duration::from_secs(5)) {
+    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,
     };
@@ -421,13 +443,13 @@ async fn pre_007_runtime_bounds_in_flight_store_persistence_to_configured_concur
                 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::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
                     };
                     if admission_sender.send(ingress).await.is_err() {
-                        return;
+                        return std::result::Result::Err(crate::runtime_error("test.source_failed"));
                     }
                 }
-                return;
+                return std::result::Result::Ok(());
             });
         },
     ) {
@@ -469,10 +491,12 @@ async fn pre_007_content_conflict_becomes_terminal_after_private_drain() {
             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::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
                 };
-                let _sent = admission_sender.send(ingress).await;
-                return;
+                if admission_sender.send(ingress).await.is_err() {
+                    return std::result::Result::Err(crate::runtime_error("test.source_failed"));
+                }
+                return std::result::Result::Ok(());
             });
         },
     ) {
@@ -511,10 +535,12 @@ async fn pre_007_store_failure_becomes_terminal_without_exposing_remote_error_te
             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::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
                 };
-                let _sent = admission_sender.send(ingress).await;
-                return;
+                if admission_sender.send(ingress).await.is_err() {
+                    return std::result::Result::Err(crate::runtime_error("test.source_failed"));
+                }
+                return std::result::Result::Ok(());
             });
         },
     ) {
@@ -550,13 +576,13 @@ async fn pre_008_runtime_snapshots_count_successful_pipeline_and_retain_terminal
                 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::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
                     };
                     if admission_sender.send(ingress).await.is_err() {
-                        return;
+                        return std::result::Result::Err(crate::runtime_error("test.source_failed"));
                     }
                 }
-                return;
+                return std::result::Result::Ok(());
             });
         },
     ) {
@@ -615,10 +641,12 @@ async fn pre_008_fault_snapshots_count_classified_store_failures_and_project_unh
             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::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")),
                 };
-                let _sent = admission_sender.send(ingress).await;
-                return;
+                if admission_sender.send(ingress).await.is_err() {
+                    return std::result::Result::Err(crate::runtime_error("test.source_failed"));
+                }
+                return std::result::Result::Ok(());
             });
         },
     ) {
@@ -643,3 +671,195 @@ async fn pre_008_fault_snapshots_count_classified_store_failures_and_project_unh
     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, _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;
+}

crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs
-// version: 1
+// version: 2
 
 fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> {
     let network = match ksp_store_lib::RawNetworkId::new("mainnet") {
@@ -78,7 +78,7 @@ async fn pre_008_slow_concrete_and_common_listeners_coalesce_to_same_latest_sequ
     };
     let observed = source.current().worker_snapshot().sequence();
     assert!(publisher.publish_state(ksp_worker_api::WorkerState::Running, 0, 0).is_ok());
-    assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 1, 1).is_ok());
+    assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 1, 1, false).is_ok());
     let concrete = source.wait_for_change(observed).await;
     let common = ksp_worker_api::WorkerSnapshotSource::wait_for_change(&source, observed).await;
     assert_eq!(concrete.worker_snapshot().sequence(), common.sequence());
@@ -113,3 +113,24 @@ async fn pre_008_terminal_latest_value_is_retained_after_publisher_drop() {
     assert_eq!(common, retained.worker_snapshot().clone());
     return;
 }
+
+#[test]
+fn pre_009_source_failure_and_backpressure_counters_advance_without_changing_projection_contract() {
+    let (mut publisher, source) = match snapshot_foundation() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert!(publisher.publish_state(ksp_worker_api::WorkerState::Running, 0, 0).is_ok());
+    assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 0, 1, true).is_ok());
+    assert!(publisher.record_source_failure(ksp_worker_api::WorkerState::Running, 0, 1).is_ok());
+    let concrete = source.current();
+    assert_eq!(concrete.admitted_total(), 1);
+    assert_eq!(concrete.canonicalized_total(), 1);
+    assert_eq!(concrete.backpressure_wait_total(), 1);
+    assert_eq!(concrete.source_failure_total(), 1);
+    assert_eq!(concrete.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Healthy);
+    assert_eq!(concrete.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Active);
+    let common = ksp_worker_api::WorkerSnapshotSource::current(&source);
+    assert_eq!(&common, concrete.worker_snapshot());
+    return;
+}