v0.3.13-pre.010

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs
-// version: 21
+// version: 22
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -984,6 +984,10 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati
     assert_eq!(first_aggregate.processing_frontier_slot(), std::option::Option::None);
     assert_eq!(first_aggregate.oldest_pending_slot(), std::option::Option::Some(45));
     assert_eq!(first_aggregate.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Active));
+    assert_eq!(first_aggregate.source_total(), 2);
+    assert_eq!(first_aggregate.source_active(), 1);
+    assert_eq!(first_aggregate.source_reconnecting(), 0);
+    assert_eq!(first_aggregate.source_failed(), 0);
     second.publish(
         crate::RawTransactionIngestProcessingFrontierProjection::new(2, std::option::Option::Some(42), std::option::Option::Some(40)).with_source_continuity(
             std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting),
@@ -997,6 +1001,10 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati
     assert_eq!(aggregate.processing_frontier_slot(), std::option::Option::Some(42));
     assert_eq!(aggregate.oldest_pending_slot(), std::option::Option::Some(40));
     assert_eq!(aggregate.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting));
+    assert_eq!(aggregate.source_total(), 2);
+    assert_eq!(aggregate.source_active(), 1);
+    assert_eq!(aggregate.source_reconnecting(), 1);
+    assert_eq!(aggregate.source_failed(), 0);
     assert_eq!(aggregate.source_reconnect_total(), 4);
     assert_eq!(aggregate.source_replay_attempt_total(), 6);
     assert_eq!(aggregate.source_continuity_gap_total(), 1);

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

@@ -1,7 +1,9 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs
-// version: 4
+// version: 5
 
-fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> {
+fn snapshot_foundation_with_source_total(
+    source_total: usize,
+) -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> {
     let network = match ksp_store_lib::RawNetworkId::new("mainnet") {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return std::option::Option::None,
@@ -19,7 +21,11 @@ fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnap
     if lifecycle.start().is_err() {
         return std::option::Option::None;
     }
-    return std::option::Option::Some(crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle));
+    return std::option::Option::Some(crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle, source_total));
+}
+
+fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> {
+    return snapshot_foundation_with_source_total(0);
 }
 
 #[test]
@@ -54,6 +60,10 @@ fn pre_008_initial_snapshot_and_common_projection_are_exact() {
     assert_eq!(concrete.processing_frontier_slot(), std::option::Option::None);
     assert_eq!(concrete.oldest_pending_slot(), std::option::Option::None);
     assert_eq!(concrete.source_state(), std::option::Option::None);
+    assert_eq!(concrete.source_total(), 0);
+    assert_eq!(concrete.source_active(), 0);
+    assert_eq!(concrete.source_reconnecting(), 0);
+    assert_eq!(concrete.source_failed(), 0);
     assert_eq!(concrete.source_reconnect_total(), 0);
     assert_eq!(concrete.source_replay_attempt_total(), 0);
     assert_eq!(concrete.source_continuity_gap_total(), 0);
@@ -199,3 +209,43 @@ fn pre_008_source_continuity_projection_preserves_processing_frontier_and_distin
     assert_eq!(resumed.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle);
     return;
 }
+
+#[test]
+fn v0_3_13_pre_010_multi_source_counts_and_health_are_conservative_and_source_neutral() {
+    let (mut publisher, source) = match snapshot_foundation_with_source_total(3) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let initial = source.current();
+    assert_eq!(initial.source_total(), 3);
+    assert_eq!(initial.source_active(), 0);
+    assert_eq!(initial.source_reconnecting(), 0);
+    assert_eq!(initial.source_failed(), 0);
+    assert_eq!(initial.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unknown);
+    let reconnecting = crate::RawTransactionIngestProcessingFrontierProjection::new(1, std::option::Option::Some(50), std::option::Option::Some(50))
+        .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting), 0, 1, 0)
+        .with_source_counts(3, 2, 1, 0);
+    assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, reconnecting).is_ok());
+    let degraded = source.current();
+    assert_eq!(degraded.source_total(), 3);
+    assert_eq!(degraded.source_active(), 2);
+    assert_eq!(degraded.source_reconnecting(), 1);
+    assert_eq!(degraded.source_failed(), 0);
+    assert_eq!(degraded.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Degraded);
+    assert_eq!(degraded.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Active);
+    let active = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(51), std::option::Option::None)
+        .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), 1, 1, 0)
+        .with_source_counts(3, 3, 0, 0);
+    assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, active).is_ok());
+    let healthy = source.current();
+    assert_eq!(healthy.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Healthy);
+    assert_eq!(healthy.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle);
+    let failed = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(51), std::option::Option::None)
+        .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), 1, 1, 0)
+        .with_source_counts(3, 2, 0, 1);
+    assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, failed).is_ok());
+    let unhealthy = source.current();
+    assert_eq!(unhealthy.source_failed(), 1);
+    assert_eq!(unhealthy.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy);
+    return;
+}