0.3.16-pre.007

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: 12
+// version: 13
 
 struct ActiveTaskGuard {
     active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
@@ -249,6 +249,7 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi
 #[derive(Clone, Copy)]
 enum RuntimePortResponse {
     Conflict,
+    QuarantinedConflict,
     StoreFailure,
     StoreFailureBlocked,
     SuccessBlocked,
@@ -337,35 +338,13 @@ impl crate::RawTransactionIngestPersistencePort for RuntimePersistencePort {
                     ));
                 }
             }
-            if matches!(response, RuntimePortResponse::Conflict) {
+            if matches!(response, RuntimePortResponse::QuarantinedConflict) {
                 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 record_observation<'a>(
-        &'a self,
-        _observation: ksp_store_lib::RawTransactionObservation,
-    ) -> ksp_store_lib::StoreApiFuture<'a, ksp_store_lib::Result<ksp_store_lib::RawObservationWriteOutcome>> {
-        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::RawObservationWriteOutcome::Inserted);
-                }
+                return std::result::Result::Ok(ksp_store_lib::RawAcquisitionWriteOutcome::with_transaction_variant(
+                    ksp_store_lib::RawEntityWriteOutcome::AlreadyPresent,
+                    ksp_store_lib::RawObservationWriteOutcome::Inserted,
+                    ksp_store_lib::RawTransactionVariantWriteOutcome::QuarantinedConflict,
+                ));
             }
             if matches!(response, RuntimePortResponse::Conflict) {
                 self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
@@ -505,7 +484,50 @@ async fn pre_007_runtime_bounds_in_flight_store_persistence_to_configured_concur
 }
 
 #[tokio::test(flavor = "current_thread")]
-async fn pre_007_content_conflict_becomes_terminal_after_private_drain() {
+async fn v0_3_16_pre_007_quarantined_conflict_keeps_worker_running_and_degraded() {
+    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::QuarantinedConflict, 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, 10) {
+                    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();
+    assert!(wait_for_completed(&port, 1).await);
+    let snapshot = source.current();
+    assert_eq!(snapshot.worker_snapshot().state(), ksp_worker_api::WorkerState::Running);
+    assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Degraded);
+    assert_eq!(snapshot.persisted_total(), 1);
+    assert_eq!(snapshot.content_conflict_total(), 1);
+    assert_eq!(snapshot.store_failure_total(), 0);
+    assert!(handle.request_stop());
+    let terminal = handle.wait_terminal().await;
+    assert_eq!(terminal, std::result::Result::Ok(ksp_worker_api::WorkerState::Stopped));
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_007_legacy_store_conflict_error_becomes_terminal_after_private_drain() {
     let settings = match settings_with_persistence_concurrency(1) {
         std::option::Option::Some(value) => value,
         std::option::Option::None => return,