diff --git a/crates/audit/Cargo.toml b/crates/audit/Cargo.toml index 6d81d97ef..8ec993a27 100644 --- a/crates/audit/Cargo.toml +++ b/crates/audit/Cargo.toml @@ -67,6 +67,7 @@ tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "time", tracing = { workspace = true, features = ["std", "attributes"] } [dev-dependencies] +rustfs-targets = { workspace = true, features = ["test-support"] } async-trait = { workspace = true } temp-env = { workspace = true } url = { workspace = true } diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index 316dea93c..9330b7707 100644 --- a/crates/audit/src/pipeline.rs +++ b/crates/audit/src/pipeline.rs @@ -564,88 +564,21 @@ impl AuditRuntimeFacade { mod tests { use super::AuditPipeline; use crate::{AuditEntry, AuditError, AuditRegistry}; - use async_trait::async_trait; - use rustfs_targets::arn::TargetID; - use rustfs_targets::store::{Key, Store}; - use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use rustfs_targets::{StoreError, Target, TargetError}; + use rustfs_targets::testkit::MockTarget; use std::sync::Arc; use tokio::sync::{Mutex, Notify}; - /// Mock target whose `save()` outcome is fixed at construction so tests can - /// force full-success / full-failure / partial-failure fan-outs. - #[derive(Clone)] - struct MockTarget { - id: TargetID, - fail: bool, - health_gate: Option<(Arc, Arc)>, - } - - impl MockTarget { - fn new(id: &str, fail: bool) -> Self { - Self { - id: TargetID::new(id.to_string(), "webhook".to_string()), - fail, - health_gate: None, - } - } - - fn with_health_gate(mut self, started: Arc, release: Arc) -> Self { - self.health_gate = Some((started, release)); - self - } - } - - #[async_trait] - impl Target for MockTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - if let Some((started, release)) = &self.health_gate { - started.notify_one(); - release.notified().await; - } - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - if self.fail { - Err(TargetError::Configuration("forced save failure".to_string())) - } else { - Ok(()) - } - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - true - } + /// Builds a mock target whose `save()` outcome is fixed at construction so tests can force + /// full-success / full-failure / partial-failure fan-outs. + fn mock_target(id: &str, fail: bool) -> MockTarget { + let target = MockTarget::new(id, "webhook"); + if fail { target.with_save_failures(usize::MAX) } else { target } } fn pipeline_with(targets: Vec) -> AuditPipeline { let mut registry = AuditRegistry::new(); for target in targets { - registry.add_target(target.id.to_string(), Box::new(target)); + registry.add_target(target.target_id().to_string(), Box::new(target)); } AuditPipeline::new(Arc::new(Mutex::new(registry))) } @@ -658,7 +591,7 @@ mod tests { // dispatch must return Err rather than swallowing the failures as Ok. #[tokio::test] async fn dispatch_returns_err_when_all_targets_fail() { - let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", true), MockTarget::new("b:webhook", true)]); + let pipeline = pipeline_with(vec![mock_target("a:webhook", true), mock_target("b:webhook", true)]); let result = pipeline.dispatch(entry()).await; assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}"); } @@ -667,13 +600,13 @@ mod tests { // so dispatch reports success (degradation is logged, not propagated). #[tokio::test] async fn dispatch_returns_ok_on_partial_failure() { - let pipeline = pipeline_with(vec![MockTarget::new("ok:webhook", false), MockTarget::new("bad:webhook", true)]); + let pipeline = pipeline_with(vec![mock_target("ok:webhook", false), mock_target("bad:webhook", true)]); pipeline.dispatch(entry()).await.expect("partial success should return Ok"); } #[tokio::test] async fn dispatch_returns_ok_when_all_targets_succeed() { - let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", false), MockTarget::new("b:webhook", false)]); + let pipeline = pipeline_with(vec![mock_target("a:webhook", false), mock_target("b:webhook", false)]); pipeline.dispatch(entry()).await.expect("all-success should return Ok"); } @@ -686,9 +619,10 @@ mod tests { #[tokio::test] async fn health_probe_does_not_hold_the_registry_lock() { - let started = Arc::new(Notify::new()); let release = Arc::new(Notify::new()); - let pipeline = pipeline_with(vec![MockTarget::new("blocked", false).with_health_gate(started.clone(), release.clone())]); + let target = mock_target("blocked", false).with_health_gate(release.clone()); + let started = target.health_started(); + let pipeline = pipeline_with(vec![target]); let registry = Arc::clone(&pipeline.registry); let snapshot_task = tokio::spawn(async move { pipeline.snapshot_target_health().await }); started.notified().await; @@ -706,14 +640,14 @@ mod tests { // whole-batch loss instead of returning Ok. #[tokio::test] async fn dispatch_batch_returns_err_when_all_targets_fail() { - let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", true)]); + let pipeline = pipeline_with(vec![mock_target("a:webhook", true)]); let result = pipeline.dispatch_batch(vec![entry(), entry()]).await; assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}"); } #[tokio::test] async fn dispatch_batch_returns_ok_when_all_targets_succeed() { - let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", false), MockTarget::new("b:webhook", false)]); + let pipeline = pipeline_with(vec![mock_target("a:webhook", false), mock_target("b:webhook", false)]); pipeline .dispatch_batch(vec![entry(), entry()]) .await diff --git a/crates/audit/src/registry.rs b/crates/audit/src/registry.rs index 0ee82a92d..4e88d0a89 100644 --- a/crates/audit/src/registry.rs +++ b/crates/audit/src/registry.rs @@ -286,70 +286,10 @@ impl AuditRegistry { #[cfg(test)] mod tests { use super::AuditRegistry; - use crate::{AuditEntry, AuditError}; - use rustfs_targets::arn::TargetID; - use rustfs_targets::store::{Key, Store}; - use rustfs_targets::target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use rustfs_targets::{StoreError, Target, TargetError}; - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - - #[derive(Clone)] - struct CloseTestTarget { - id: TargetID, - close_calls: Arc, - fail_on_close: bool, - } - - impl CloseTestTarget { - fn new(id: TargetID, close_calls: Arc, fail_on_close: bool) -> Self { - Self { - id, - close_calls, - fail_on_close, - } - } - } - - #[async_trait::async_trait] - impl Target for CloseTestTarget { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - if self.fail_on_close { - Err(TargetError::Unknown("close failed".to_string())) - } else { - Ok(()) - } - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - true - } - } + use crate::AuditError; + use rustfs_targets::TargetError; + use rustfs_targets::target::ChannelTargetType; + use rustfs_targets::testkit::MockTarget; #[test] fn registry_registers_amqp_factory() { @@ -361,23 +301,21 @@ mod tests { #[tokio::test] async fn close_all_returns_first_error_and_clears_targets() { let mut registry = AuditRegistry::new(); - let ok_calls = Arc::new(AtomicUsize::new(0)); - let fail_calls = Arc::new(AtomicUsize::new(0)); + let ok = MockTarget::new("ok", "webhook"); + let ok_observer = ok.clone(); + let fail = MockTarget::new("fail", "webhook") + .with_close_failures(usize::MAX) + .with_close_failure_error(|| TargetError::Unknown("close failed".to_string())); + let fail_observer = fail.clone(); - let ok_id = TargetID::new("ok".to_string(), "webhook".to_string()); - let fail_id = TargetID::new("fail".to_string(), "webhook".to_string()); - - registry.add_target(ok_id.to_string(), Box::new(CloseTestTarget::new(ok_id, Arc::clone(&ok_calls), false))); - registry.add_target( - fail_id.to_string(), - Box::new(CloseTestTarget::new(fail_id, Arc::clone(&fail_calls), true)), - ); + registry.add_target(ok.target_id().to_string(), Box::new(ok)); + registry.add_target(fail.target_id().to_string(), Box::new(fail)); let result = registry.close_all().await; assert!(matches!(result, Err(AuditError::Target(TargetError::Unknown(_))))); - assert_eq!(ok_calls.load(Ordering::SeqCst), 1); - assert_eq!(fail_calls.load(Ordering::SeqCst), 1); + assert_eq!(ok_observer.close_call_count(), 1); + assert_eq!(fail_observer.close_call_count(), 1); assert!(registry.list_targets().is_empty()); } } diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index 4d79a04e6..0dcddd59e 100644 --- a/crates/audit/src/system.rs +++ b/crates/audit/src/system.rs @@ -577,76 +577,17 @@ fn warn_audit_state(state: &str, reason: Option<&str>) { mod tests { use super::{AuditSystem, AuditSystemState}; use crate::{AuditEntry, AuditError}; - use async_trait::async_trait; use rustfs_targets::ReplayWorkerManager; - use rustfs_targets::arn::TargetID; - use rustfs_targets::store::{Key, Store}; - use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use rustfs_targets::{StoreError, Target, TargetError}; + use rustfs_targets::testkit::MockTarget; use std::collections::HashMap; use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::mpsc; - #[derive(Clone)] - struct TestTarget { - close_calls: Arc, - id: TargetID, - } - - impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - close_calls: Arc::new(AtomicUsize::new(0)), - id: TargetID::new(id.to_string(), name.to_string()), - } - } - } - - #[async_trait] - impl Target for TestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - true - } - } - #[tokio::test] async fn reload_with_empty_config_stops_existing_runtime() { let system = AuditSystem::new(); - let target = TestTarget::new("primary", "webhook"); - let close_calls = Arc::clone(&target.close_calls); + let target = MockTarget::new("primary", "webhook"); + let observer = target.clone(); { let mut registry = system.registry.lock().await; @@ -671,7 +612,7 @@ mod tests { assert_eq!(system.get_state().await, AuditSystemState::Stopped); assert!(system.list_targets().await.is_empty()); assert_eq!(system.runtime_status_snapshot().await, ReplayWorkerManager::new().snapshot(0)); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); assert_eq!(*system.config.read().await, Some(rustfs_config::server_config::Config(HashMap::new()))); } @@ -693,7 +634,7 @@ mod tests { // Seed a target + replay worker so both critical sections touch real state. { let mut registry = system.registry.lock().await; - registry.add_target("primary:webhook".to_string(), Box::new(TestTarget::new("primary", "webhook"))); + registry.add_target("primary:webhook".to_string(), Box::new(MockTarget::new("primary", "webhook"))); } { let mut replay_workers = system.stream_cancellers.write().await; @@ -793,8 +734,8 @@ mod tests { async fn commit_closes_old_targets_before_installing_new() { let system = AuditSystem::new(); - let old = TestTarget::new("old", "webhook"); - let old_close = Arc::clone(&old.close_calls); + let old = MockTarget::new("old", "webhook"); + let old_observer = old.clone(); { let mut registry = system.registry.lock().await; registry.add_target("old:webhook".to_string(), Box::new(old)); @@ -809,17 +750,17 @@ mod tests { *state = AuditSystemState::Running; } - let new = TestTarget::new("new", "webhook"); - let new_close = Arc::clone(&new.close_calls); + let new = MockTarget::new("new", "webhook"); + let new_observer = new.clone(); system .commit_runtime_targets(vec![Box::new(new)], AuditSystemState::Running) .await .expect("commit should succeed"); // Old target closed exactly once during the pre-install shutdown. - assert_eq!(old_close.load(Ordering::SeqCst), 1); + assert_eq!(old_observer.close_call_count(), 1); // New target installed and left open. - assert_eq!(new_close.load(Ordering::SeqCst), 0); + assert_eq!(new_observer.close_call_count(), 0); assert_eq!(system.list_targets().await, vec!["new:webhook".to_string()]); // Old replay worker stopped; the store-less new target adds none. assert_eq!(system.runtime_status_snapshot().await.replay_worker_count, 0); diff --git a/crates/audit/tests/pipeline_layer_test.rs b/crates/audit/tests/pipeline_layer_test.rs index 5d5608db2..903244ee0 100644 --- a/crates/audit/tests/pipeline_layer_test.rs +++ b/crates/audit/tests/pipeline_layer_test.rs @@ -12,136 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. -use async_trait::async_trait; use rustfs_audit::{AuditEntry, AuditError, AuditPipeline, AuditRegistry, AuditRuntimeFacade, AuditRuntimeView}; -use rustfs_targets::arn::TargetID; -use rustfs_targets::store::{Key, Store}; -use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; -use rustfs_targets::{SharedTarget, StoreError, Target, TargetError}; -use serde::{Serialize, de::DeserializeOwned}; +use rustfs_targets::SharedTarget; +use rustfs_targets::testkit::MockTarget; use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{Mutex, RwLock}; -#[derive(Clone)] -struct TestTarget { - close_calls: Arc, - id: TargetID, - init_calls: Arc, -} - -impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - close_calls: Arc::new(AtomicUsize::new(0)), - id: TargetID::new(id.to_string(), name.to_string()), - init_calls: Arc::new(AtomicUsize::new(0)), - } - } -} - -#[async_trait] -impl Target for TestTarget -where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, -{ - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - self.init_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } -} - -/// A target whose `save()` always fails, used to exercise the dispatch +/// Builds a target whose `save()` always fails, used to exercise the dispatch /// failure-propagation paths. -#[derive(Clone)] -struct FailingTarget { - id: TargetID, - save_calls: Arc, -} - -impl FailingTarget { - fn new(id: &str, name: &str) -> Self { - Self { - id: TargetID::new(id.to_string(), name.to_string()), - save_calls: Arc::new(AtomicUsize::new(0)), - } - } -} - -#[async_trait] -impl Target for FailingTarget -where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, -{ - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - self.save_calls.fetch_add(1, Ordering::SeqCst); - Err(TargetError::Storage("disk full".to_string())) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } +fn failing_target(id: &str, name: &str) -> MockTarget { + MockTarget::new(id, name).with_save_failures(usize::MAX) } fn pipeline_with_targets(targets: Vec<(&str, SharedTarget)>) -> AuditPipeline { @@ -154,8 +34,8 @@ fn pipeline_with_targets(targets: Vec<(&str, SharedTarget)>) -> Audi #[tokio::test] async fn audit_pipeline_dispatch_propagates_total_failure() { - let failing = FailingTarget::new("primary", "webhook"); - let save_calls = Arc::clone(&failing.save_calls); + let failing = failing_target("primary", "webhook"); + let observer = failing.clone(); let pipeline = pipeline_with_targets(vec![("primary:webhook", Arc::new(failing))]); let result = pipeline.dispatch(Arc::new(AuditEntry::default())).await; @@ -164,13 +44,13 @@ async fn audit_pipeline_dispatch_propagates_total_failure() { matches!(result, Err(AuditError::Target(_))), "dispatch must surface an error when every target fails, got {result:?}" ); - assert_eq!(save_calls.load(Ordering::SeqCst), 1, "the failing target should have been invoked"); + assert_eq!(observer.save_call_count(), 1, "the failing target should have been invoked"); } #[tokio::test] async fn audit_pipeline_dispatch_tolerates_partial_failure() { - let failing = FailingTarget::new("primary", "webhook"); - let healthy = TestTarget::new("secondary", "webhook"); + let failing = failing_target("primary", "webhook"); + let healthy = MockTarget::new("secondary", "webhook"); let pipeline = pipeline_with_targets(vec![ ("primary:webhook", Arc::new(failing)), ("secondary:webhook", Arc::new(healthy)), @@ -186,7 +66,7 @@ async fn audit_pipeline_dispatch_tolerates_partial_failure() { #[tokio::test] async fn audit_pipeline_dispatch_batch_propagates_total_failure() { - let failing = FailingTarget::new("primary", "webhook"); + let failing = failing_target("primary", "webhook"); let pipeline = pipeline_with_targets(vec![("primary:webhook", Arc::new(failing))]); let entries = vec![Arc::new(AuditEntry::default()), Arc::new(AuditEntry::default())]; @@ -200,8 +80,8 @@ async fn audit_pipeline_dispatch_batch_propagates_total_failure() { #[tokio::test] async fn audit_pipeline_dispatch_batch_tolerates_partial_failure() { - let failing = FailingTarget::new("primary", "webhook"); - let healthy = TestTarget::new("secondary", "webhook"); + let failing = failing_target("primary", "webhook"); + let healthy = MockTarget::new("secondary", "webhook"); let pipeline = pipeline_with_targets(vec![ ("primary:webhook", Arc::new(failing)), ("secondary:webhook", Arc::new(healthy)), @@ -266,9 +146,8 @@ async fn audit_runtime_facade_activates_empty_target_list() { async fn audit_runtime_view_upsert_and_remove_target() { let registry = Arc::new(Mutex::new(AuditRegistry::new())); let runtime_view = AuditRuntimeView::new(registry.clone()); - let target = TestTarget::new("primary", "webhook"); - let init_calls = Arc::clone(&target.init_calls); - let close_calls = Arc::clone(&target.close_calls); + let target = MockTarget::new("primary", "webhook"); + let observer = target.clone(); runtime_view .upsert_target("primary:webhook".to_string(), Box::new(target)) @@ -276,7 +155,7 @@ async fn audit_runtime_view_upsert_and_remove_target() { .expect("upsert should succeed"); assert_eq!(runtime_view.list_targets().await, vec!["primary:webhook".to_string()]); - assert_eq!(init_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.init_call_count(), 1); runtime_view .remove_target("primary:webhook") @@ -284,7 +163,7 @@ async fn audit_runtime_view_upsert_and_remove_target() { .expect("remove should succeed"); assert!(runtime_view.list_targets().await.is_empty()); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } #[tokio::test] @@ -292,7 +171,7 @@ async fn audit_runtime_facade_replace_targets_commits_runtime_state() { let registry = Arc::new(Mutex::new(AuditRegistry::new())); let replay_workers = Arc::new(RwLock::new(rustfs_targets::ReplayWorkerManager::new())); let facade = AuditRuntimeFacade::new(registry.clone(), replay_workers.clone()); - let target = TestTarget::new("primary", "webhook"); + let target = MockTarget::new("primary", "webhook"); let activation = rustfs_targets::RuntimeActivation { replay_workers: rustfs_targets::ReplayWorkerManager::new(), targets: vec![Arc::new(target) as rustfs_targets::SharedTarget], diff --git a/crates/notify/Cargo.toml b/crates/notify/Cargo.toml index 024a4ac64..1bf4fb78a 100644 --- a/crates/notify/Cargo.toml +++ b/crates/notify/Cargo.toml @@ -91,6 +91,7 @@ metrics = { workspace = true } quick-xml = { workspace = true, features = ["serialize", "serde-types", "encoding"] } [dev-dependencies] +rustfs-targets = { workspace = true, features = ["test-support"] } tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] } tracing-subscriber = { workspace = true, features = ["env-filter", "time"] } axum = { workspace = true } diff --git a/crates/notify/src/lifecycle.rs b/crates/notify/src/lifecycle.rs index b0b37b612..86f97ae70 100644 --- a/crates/notify/src/lifecycle.rs +++ b/crates/notify/src/lifecycle.rs @@ -476,6 +476,7 @@ mod tests { use rustfs_targets::arn::TargetID; use rustfs_targets::store::{Key, QueueStore, Store}; use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; + use rustfs_targets::testkit::MockTarget; use rustfs_targets::{ EventName, ReplayWorkerManager, StoreError, Target, TargetError, TargetPluginDescriptor, TargetPluginRegistry, }; @@ -488,108 +489,6 @@ mod tests { const RETRY_INIT_TARGET_TYPE: &str = "lifecycle_retry_init"; const REPLAY_TARGET_TYPE: &str = "lifecycle_replay"; - struct BlockingInitState { - close_calls: AtomicUsize, - init_entered: Notify, - } - - #[derive(Clone)] - struct BlockingInitTarget { - id: TargetID, - state: Arc, - } - - #[async_trait] - impl Target for BlockingInitTarget { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.state.close_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - self.state.init_entered.notify_one(); - std::future::pending::<()>().await; - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } - } - - #[derive(Clone)] - struct RetryInitTarget { - fail_once: Arc, - id: TargetID, - should_fail: bool, - } - - #[async_trait] - impl Target for RetryInitTarget { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - if self.should_fail && self.fail_once.swap(false, Ordering::AcqRel) { - return Err(TargetError::Initialization("forced transient init failure".to_string())); - } - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } - } - struct ReplayState { active_workers: AtomicUsize, completed_deliveries: AtomicUsize, @@ -929,28 +828,24 @@ mod tests { } fn blocking_init_runtime() -> ( - Arc, + Arc, + MockTarget, NotifyLifecycleCoordinator, NotifyRuntimeView, Arc, Config, ) { - let state = Arc::new(BlockingInitState { - close_calls: AtomicUsize::new(0), - init_entered: Notify::new(), - }); + // One observed template feeds every factory-constructed instance, so the returned + // observer sees the shared init signal and close counter across generations. + let init_entered = Arc::new(Notify::new()); + let template = MockTarget::new("primary", BLOCKING_INIT_TARGET_TYPE).with_blocking_init(init_entered.clone()); + let observer = template.clone(); let mut plugins = TargetPluginRegistry::::new(); - let factory_state = state.clone(); plugins.register(TargetPluginDescriptor::new( BLOCKING_INIT_TARGET_TYPE, &[ENABLE_KEY], |_config| Ok(()), - move |id, _config| { - Ok(Box::new(BlockingInitTarget { - id: TargetID::new(id, BLOCKING_INIT_TARGET_TYPE.to_string()), - state: factory_state.clone(), - })) - }, + move |id, _config| Ok(Box::new(template.clone().with_id(&id, BLOCKING_INIT_TARGET_TYPE))), )); let config = config_with_enabled_test_target(BLOCKING_INIT_TARGET_TYPE, "primary"); let handoff_count = Arc::new(AtomicUsize::new(0)); @@ -962,7 +857,7 @@ mod tests { observer_count.fetch_add(1, Ordering::SeqCst); })), ); - (state, coordinator, runtime_view, handoff_count, config) + (init_entered, observer, coordinator, runtime_view, handoff_count, config) } async fn recv_until_generation(receiver: &mut mpsc::UnboundedReceiver, expected: usize, context: &str) { @@ -1001,10 +896,10 @@ mod tests { #[tokio::test] async fn real_runtime_disable_supersedes_target_init_and_leaves_no_runtime_state() { - let (state, coordinator, runtime_view, handoff_count, config) = blocking_init_runtime(); + let (init_entered, observer, coordinator, runtime_view, handoff_count, config) = blocking_init_runtime(); let enable = coordinator.set_mode(true, Some(config)); - state.init_entered.notified().await; + init_entered.notified().await; let disable = coordinator.set_mode(false, None); enable.wait().await.expect("superseded real enable should finish"); @@ -1013,7 +908,7 @@ mod tests { let status = runtime_view.runtime_status_snapshot().await; assert_eq!(status.target_count, 0); assert_eq!(status.replay_worker_count, 0); - assert_eq!(state.close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); assert_eq!(handoff_count.load(Ordering::SeqCst), 0); assert_eq!(coordinator.state(), NotificationRuntimeState::LiveOnly); } @@ -1057,10 +952,10 @@ mod tests { #[tokio::test] async fn real_runtime_terminate_cancels_target_init_and_is_final() { - let (state, coordinator, runtime_view, handoff_count, config) = blocking_init_runtime(); + let (init_entered, observer, coordinator, runtime_view, handoff_count, config) = blocking_init_runtime(); let enable = coordinator.set_mode(true, Some(config)); - state.init_entered.notified().await; + init_entered.notified().await; let terminate = coordinator.terminate(); enable.wait().await.expect("superseded real enable should finish"); @@ -1069,7 +964,7 @@ mod tests { let status = runtime_view.runtime_status_snapshot().await; assert_eq!(status.target_count, 0); assert_eq!(status.replay_worker_count, 0); - assert_eq!(state.close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); assert_eq!(handoff_count.load(Ordering::SeqCst), 0); assert_eq!(coordinator.state(), NotificationRuntimeState::Terminated); let err = coordinator @@ -1085,18 +980,19 @@ mod tests { #[tokio::test] async fn partial_activation_reports_error_and_same_config_can_retry() { - let fail_once = Arc::new(AtomicBool::new(true)); + // The template's single-failure init budget is shared by every clone the factory hands + // out, so the "bad" instance fails once in the first generation and recovers on retry. + let bad_template = MockTarget::new("bad", RETRY_INIT_TARGET_TYPE).with_init_failures(1); let mut plugins = TargetPluginRegistry::::new(); - let factory_fail_once = fail_once.clone(); plugins.register(TargetPluginDescriptor::new( RETRY_INIT_TARGET_TYPE, &[ENABLE_KEY], |_config| Ok(()), move |id, _config| { - Ok(Box::new(RetryInitTarget { - should_fail: id == "bad", - id: TargetID::new(id, RETRY_INIT_TARGET_TYPE.to_string()), - fail_once: factory_fail_once.clone(), + Ok(Box::new(if id == "bad" { + bad_template.clone() + } else { + MockTarget::new(&id, RETRY_INIT_TARGET_TYPE) })) }, )); diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index 7b1aaf237..ad378679a 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -610,18 +610,10 @@ impl TargetList { mod tests { use super::*; use crate::{rule_engine::NotifyRuleEngine, rules::RulesMap}; - use async_trait::async_trait; use rustfs_s3_types::EventName; - use rustfs_targets::StoreError; - use rustfs_targets::{ - ReplayWorkerManager, TargetError, - store::{Key, QueueStore, Store}, - target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}, - }; - use std::sync::{ - Arc, - atomic::{AtomicUsize, Ordering}, - }; + use rustfs_targets::testkit::MockTarget; + use rustfs_targets::{ReplayWorkerManager, store::QueueStore}; + use std::sync::Arc; use tokio::sync::Notify; #[tokio::test] @@ -680,105 +672,6 @@ mod tests { assert!(suffix_targets.is_empty()); } - #[derive(Clone)] - struct TestTarget { - block_first_save: Option<(Arc, Arc)>, - close_calls: Arc, - close_entered: Option>, - id: TargetID, - enabled: bool, - save_calls: Arc, - selected_calls: Arc, - store: Option>, - } - - impl TestTarget { - fn new(id: &str, name: &str, enabled: bool) -> Self { - Self { - block_first_save: None, - close_calls: Arc::new(AtomicUsize::new(0)), - close_entered: None, - id: TargetID::new(id.to_string(), name.to_string()), - enabled, - save_calls: Arc::new(AtomicUsize::new(0)), - selected_calls: Arc::new(AtomicUsize::new(0)), - store: None, - } - } - - fn with_blocked_first_save(mut self, entered: Arc, release: Arc) -> Self { - self.block_first_save = Some((entered, release)); - self - } - - fn with_store(mut self, store: QueueStore) -> Self { - self.store = Some(store); - self - } - - fn with_close_observer(mut self, close_entered: Arc) -> Self { - self.close_entered = Some(close_entered); - self - } - } - - #[async_trait] - impl Target for TestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(self.enabled) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - let call = self.save_calls.fetch_add(1, Ordering::SeqCst); - if call == 0 - && let Some((entered, release)) = &self.block_first_save - { - entered.notify_one(); - release.notified().await; - } - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - if let Some(close_entered) = &self.close_entered { - close_entered.notify_one(); - } - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - self.store - .as_ref() - .map(|store| store as &(dyn Store + Send + Sync)) - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - let cloned = self.clone(); - Box::new(cloned) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - self.selected_calls.fetch_add(1, Ordering::SeqCst); - self.enabled - } - } - #[tokio::test] async fn lifecycle_pause_drains_entered_deferred_dispatch_and_blocks_new_dispatch() { let metrics = Arc::new(NotificationMetrics::new()); @@ -787,12 +680,12 @@ mod tests { let save_entered = Arc::new(Notify::new()); let save_release = Arc::new(Notify::new()); let queue_dir = tempfile::tempdir().expect("queue tempdir should be created"); - let target = TestTarget::new("gated-target", "webhook", true) - .with_blocked_first_save(save_entered.clone(), save_release.clone()) - .with_store(QueueStore::new(queue_dir.path(), 16, ".event")); + let target = MockTarget::new("gated-target", "webhook") + .with_first_save_gate(save_entered.clone(), save_release.clone()) + .with_store(Arc::new(QueueStore::new(queue_dir.path(), 16, ".event"))); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier .target_list() @@ -817,7 +710,7 @@ mod tests { } }); save_entered.notified().await; - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); + assert_eq!(target.save_call_count(), 1); let mut pause = Box::pin(facade.pause_dispatch()); tokio::select! { @@ -830,7 +723,7 @@ mod tests { first_dispatch.await.expect("first dispatch task should finish"); let pause_guard = pause.await; - let replacement = TestTarget::new("gated-target", "webhook", true); + let replacement = MockTarget::new("gated-target", "webhook"); { let target_list = notifier.target_list(); let mut target_list = target_list.write().await; @@ -848,19 +741,19 @@ mod tests { _ = std::future::ready(()) => {} } assert_eq!( - replacement.selected_calls.load(Ordering::SeqCst), + replacement.enabled_call_count(), 0, "a paused dispatch must not select a target from the replacement generation early" ); - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); - assert_eq!(replacement.save_calls.load(Ordering::SeqCst), 0); + assert_eq!(target.save_call_count(), 1); + assert_eq!(replacement.save_call_count(), 0); drop(pause_guard); second_dispatch.await; - assert_eq!(target.selected_calls.load(Ordering::SeqCst), 1); - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); - assert_eq!(replacement.selected_calls.load(Ordering::SeqCst), 1); - assert_eq!(replacement.save_calls.load(Ordering::SeqCst), 1); + assert_eq!(target.enabled_call_count(), 1); + assert_eq!(target.save_call_count(), 1); + assert_eq!(replacement.enabled_call_count(), 1); + assert_eq!(replacement.save_call_count(), 1); } #[tokio::test] @@ -870,11 +763,10 @@ mod tests { let notifier = Arc::new(EventNotifier::new(metrics.clone(), rule_engine.clone())); let save_entered = Arc::new(Notify::new()); let save_release = Arc::new(Notify::new()); - let target = - TestTarget::new("direct-target", "webhook", true).with_blocked_first_save(save_entered.clone(), save_release.clone()); + let target = MockTarget::new("direct-target", "webhook").with_first_save_gate(save_entered.clone(), save_release.clone()); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier .target_list() @@ -921,13 +813,11 @@ mod tests { }); let first_entered = Arc::new(Notify::new()); let first_release = Arc::new(Notify::new()); - let close_entered = Arc::new(Notify::new()); - let target = TestTarget::new("direct-target", "webhook", true) - .with_blocked_first_save(first_entered.clone(), first_release.clone()) - .with_close_observer(close_entered); + let target = + MockTarget::new("direct-target", "webhook").with_first_save_gate(first_entered.clone(), first_release.clone()); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier .target_list() @@ -964,7 +854,7 @@ mod tests { } }); tokio::time::timeout(std::time::Duration::from_secs(1), async { - while target.selected_calls.load(Ordering::SeqCst) != 2 { + while target.enabled_call_count() != 2 { tokio::task::yield_now().await; } }) @@ -978,14 +868,14 @@ mod tests { result = &mut replace => panic!("replacement closed a generation with selected direct sends: {result:?}"), _ = std::future::ready(()) => {} } - assert_eq!(target.close_calls.load(Ordering::SeqCst), 0); + assert_eq!(target.close_call_count(), 0); first_release.notify_one(); first.await.expect("first direct dispatch should finish"); second.await.expect("permit-waiting direct dispatch should be cancelled"); replace.await.expect("replacement should close after direct leases drain"); - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); - assert_eq!(target.close_calls.load(Ordering::SeqCst), 1); + assert_eq!(target.save_call_count(), 1); + assert_eq!(target.close_call_count(), 1); assert_eq!(metrics.processing_count(), 0); assert_eq!(metrics.processed_count(), 1); assert_eq!(metrics.skipped_count(), 1); @@ -999,12 +889,12 @@ mod tests { let save_entered = Arc::new(Notify::new()); let save_release = Arc::new(Notify::new()); let queue_dir = tempfile::tempdir().expect("queue tempdir should be created"); - let target = TestTarget::new("deferred", "webhook", true) - .with_blocked_first_save(save_entered.clone(), save_release.clone()) - .with_store(QueueStore::new(queue_dir.path(), 16, ".event")); + let target = MockTarget::new("deferred", "webhook") + .with_first_save_gate(save_entered.clone(), save_release.clone()) + .with_store(Arc::new(QueueStore::new(queue_dir.path(), 16, ".event"))); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier .target_list() @@ -1060,10 +950,10 @@ mod tests { let mut targets = Vec::new(); let mut rules_map = RulesMap::new(); for index in 0..TARGETS { - let target = TestTarget::new(&format!("deferred-{index}"), "webhook", true) - .with_blocked_first_save(entered.clone(), release.clone()) - .with_store(QueueStore::new(queue_dir.path().join(index.to_string()), 16, ".event")); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + let target = MockTarget::new(&format!("deferred-{index}"), "webhook") + .with_first_save_gate(entered.clone(), release.clone()) + .with_store(Arc::new(QueueStore::new(queue_dir.path().join(index.to_string()), 16, ".event"))); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); notifier .target_list() .write() @@ -1082,12 +972,7 @@ mod tests { .await; } }); - let total_calls = || { - targets - .iter() - .map(|target| target.save_calls.load(Ordering::SeqCst)) - .sum::() - }; + let total_calls = || targets.iter().map(|target| target.save_call_count()).sum::(); tokio::time::timeout(std::time::Duration::from_secs(1), async { while total_calls() != LIMIT { tokio::task::yield_now().await; @@ -1122,20 +1007,19 @@ mod tests { }); let direct_entered = Arc::new(Notify::new()); let direct_release = Arc::new(Notify::new()); - let direct = - TestTarget::new("direct", "webhook", true).with_blocked_first_save(direct_entered.clone(), direct_release.clone()); + let direct = MockTarget::new("direct", "webhook").with_first_save_gate(direct_entered.clone(), direct_release.clone()); let deferred_entered = Arc::new(Notify::new()); let deferred_release = Arc::new(Notify::new()); let queue_dir = tempfile::tempdir().expect("queue tempdir should be created"); - let deferred = TestTarget::new("deferred", "webhook", true) - .with_blocked_first_save(deferred_entered.clone(), deferred_release.clone()) - .with_store(QueueStore::new(queue_dir.path(), 16, ".event")); + let deferred = MockTarget::new("deferred", "webhook") + .with_first_save_gate(deferred_entered.clone(), deferred_release.clone()) + .with_store(Arc::new(QueueStore::new(queue_dir.path(), 16, ".event"))); let mut direct_rules = RulesMap::new(); - direct_rules.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), direct.id.clone()); + direct_rules.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), direct.target_id()); rule_engine.set_bucket_rules("direct-bucket", direct_rules).await; let mut deferred_rules = RulesMap::new(); - deferred_rules.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), deferred.id.clone()); + deferred_rules.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), deferred.target_id()); rule_engine.set_bucket_rules("deferred-bucket", deferred_rules).await; { let target_list = notifier.target_list(); @@ -1198,12 +1082,12 @@ mod tests { let rule_engine = NotifyRuleEngine::new(); let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new()), rule_engine.clone()); - let enabled_target = TestTarget::new("enabled-target", "webhook", true); - let disabled_target = TestTarget::new("disabled-target", "webhook", false); + let enabled_target = MockTarget::new("enabled-target", "webhook"); + let disabled_target = MockTarget::new("disabled-target", "webhook").disabled(); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), enabled_target.id.clone()); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), disabled_target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), enabled_target.target_id()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), disabled_target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier @@ -1222,17 +1106,17 @@ mod tests { let event = Arc::new(Event::new_test_event("bucket", "object", EventName::ObjectCreatedPut)); notifier.send(event).await; - assert_eq!(enabled_target.save_calls.load(Ordering::SeqCst), 1); - assert_eq!(disabled_target.save_calls.load(Ordering::SeqCst), 0); + assert_eq!(enabled_target.save_call_count(), 1); + assert_eq!(disabled_target.save_call_count(), 0); } #[tokio::test] async fn send_event_respects_prefix_suffix_filters() { let rule_engine = NotifyRuleEngine::new(); let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new()), rule_engine.clone()); - let target = TestTarget::new("filtered-target", "webhook", true); + let target = MockTarget::new("filtered-target", "webhook"); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "uploads/*.csv".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "uploads/*.csv".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier.target_list().write().await.add(Arc::new(target.clone())).unwrap(); @@ -1248,7 +1132,7 @@ mod tests { ))) .await; - assert_eq!(target.save_calls.load(Ordering::SeqCst), 0); + assert_eq!(target.save_call_count(), 0); notifier .send(Arc::new(Event::new_test_event( @@ -1258,72 +1142,18 @@ mod tests { ))) .await; - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); + assert_eq!(target.save_call_count(), 1); } - /// A store-backed (deferred) target. `save` only enqueues to the store, so - /// the actual delivery happens later in the replay worker. - #[derive(Clone)] - struct DeferredTestTarget { - id: TargetID, - save_calls: Arc, - store: QueueStore, - } - - impl DeferredTestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - id: TargetID::new(id.to_string(), name.to_string()), - save_calls: Arc::new(AtomicUsize::new(0)), - // The store is never actually written to here: `save` below only bumps a - // counter. It just has to exist so the notifier treats this target as - // store-backed (deferred delivery), exercising the deferred counting path. - store: QueueStore::new(std::env::temp_dir().join("rustfs-notify-979-noop-store"), 0, ""), - } - } - } - - #[async_trait] - impl Target for DeferredTestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - self.save_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - Some(&self.store) - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } + /// Builds a store-backed (deferred) mock target. `save` only bumps the mock's counter, so the + /// attached store is never written to: it just has to exist so the notifier treats the target + /// as store-backed (deferred delivery), exercising the deferred counting path. + fn deferred_test_target(id: &str, name: &str) -> MockTarget { + MockTarget::new(id, name).with_store(Arc::new(QueueStore::new( + std::env::temp_dir().join("rustfs-notify-979-noop-store"), + 0, + "", + ))) } /// Regression test for backlog#979 (a): dispatching to a store-backed @@ -1338,9 +1168,9 @@ mod tests { let rule_engine = NotifyRuleEngine::new(); let notifier = EventNotifier::new(metrics.clone(), rule_engine.clone()); - let target = DeferredTestTarget::new("deferred-target", "webhook"); + let target = deferred_test_target("deferred-target", "webhook"); let mut rules_map = RulesMap::new(); - rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.target_id()); rule_engine.set_bucket_rules("bucket", rules_map).await; notifier.target_list().write().await.add(Arc::new(target.clone())).unwrap(); @@ -1349,7 +1179,7 @@ mod tests { .send(Arc::new(Event::new_test_event("bucket", "object", EventName::ObjectCreatedPut))) .await; - assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); + assert_eq!(target.save_call_count(), 1); assert_eq!( metrics.processing_count(), 1, @@ -1394,87 +1224,20 @@ mod tests { let rule_engine = NotifyRuleEngine::new(); let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new()), rule_engine); - let old_target = ClosableTestTarget::new("old", "webhook"); + let old_target = MockTarget::new("old", "webhook"); notifier .init_bucket_targets_shared(vec![Arc::new(old_target.clone()) as SharedTarget]) .await .expect("initial install should succeed"); - assert_eq!(old_target.close_calls.load(Ordering::SeqCst), 0, "target must not close on first install"); + assert_eq!(old_target.close_call_count(), 0, "target must not close on first install"); - let new_target = ClosableTestTarget::new("new", "webhook"); + let new_target = MockTarget::new("new", "webhook"); notifier .init_bucket_targets_shared(vec![Arc::new(new_target.clone()) as SharedTarget]) .await .expect("replacement install should succeed"); - assert_eq!( - old_target.close_calls.load(Ordering::SeqCst), - 1, - "the replaced target must be closed exactly once" - ); - assert_eq!( - new_target.close_calls.load(Ordering::SeqCst), - 0, - "the freshly installed target must stay open" - ); - } - - /// A target that records `close()` invocations, for lifecycle assertions. - #[derive(Clone)] - struct ClosableTestTarget { - id: TargetID, - close_calls: Arc, - } - - impl ClosableTestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - id: TargetID::new(id.to_string(), name.to_string()), - close_calls: Arc::new(AtomicUsize::new(0)), - } - } - } - - #[async_trait] - impl Target for ClosableTestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } + assert_eq!(old_target.close_call_count(), 1, "the replaced target must be closed exactly once"); + assert_eq!(new_target.close_call_count(), 0, "the freshly installed target must stay open"); } } diff --git a/crates/notify/src/runtime_facade.rs b/crates/notify/src/runtime_facade.rs index f692880c5..282b77c9b 100644 --- a/crates/notify/src/runtime_facade.rs +++ b/crates/notify/src/runtime_facade.rs @@ -409,108 +409,14 @@ mod tests { Event, integration::NotificationMetrics, notifier::EventNotifier, rule_engine::NotifyRuleEngine, runtime_view::NotifyRuntimeView, }; - use async_trait::async_trait; use rustfs_targets::arn::TargetID; - use rustfs_targets::store::{Key, QueueStore, Store}; - use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use rustfs_targets::{ReplayWorkerManager, SharedTarget, StoreError, Target, TargetError}; + use rustfs_targets::store::QueueStore; + use rustfs_targets::target::QueuedPayload; + use rustfs_targets::testkit::MockTarget; + use rustfs_targets::{ReplayWorkerManager, SharedTarget, TargetError}; use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{Notify, RwLock, Semaphore}; - #[derive(Clone)] - struct TestTarget { - close_entered: Option>, - close_error: bool, - close_release: Option>, - close_calls: Arc, - id: TargetID, - store: Option>, - } - - impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - close_entered: None, - close_error: false, - close_release: None, - close_calls: Arc::new(AtomicUsize::new(0)), - id: TargetID::new(id.to_string(), name.to_string()), - store: None, - } - } - - fn with_blocking_close(mut self, entered: Arc, release: Arc) -> Self { - self.close_entered = Some(entered); - self.close_release = Some(release); - self - } - - fn with_close_error(mut self) -> Self { - self.close_error = true; - self - } - - fn with_store(mut self, store: QueueStore) -> Self { - self.store = Some(store); - self - } - } - - #[async_trait] - impl Target for TestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - if let Some(entered) = &self.close_entered { - entered.notify_one(); - } - if let Some(release) = &self.close_release { - release.notified().await; - } - if self.close_error { - return Err(TargetError::Storage("forced close failure".to_string())); - } - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - self.store - .as_ref() - .map(|store| store as &(dyn Store + Send + Sync)) - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } - } - fn build_facade() -> (NotifyRuntimeFacade, Arc, Arc>) { let metrics = Arc::new(NotificationMetrics::new()); let notifier = Arc::new(EventNotifier::new(metrics.clone(), NotifyRuleEngine::new())); @@ -555,8 +461,8 @@ mod tests { async fn compatibility_activation_stays_dormant_until_ordered_replace() { let (facade, _, replay_workers) = build_facade(); let queue_root = tempfile::tempdir().expect("queue root"); - let store = QueueStore::new_with_compression(queue_root.path(), 16, ".event", false); - let target = TestTarget::new("primary", "webhook").with_store(store); + let store = QueueStore::::new_with_compression(queue_root.path(), 16, ".event", false); + let target = MockTarget::new("primary", "webhook").with_store(Arc::new(store)); let activation = facade.activate_targets_with_replay(vec![Box::new(target)]).await; assert_eq!(activation.targets.len(), 1); @@ -574,7 +480,7 @@ mod tests { #[tokio::test] async fn runtime_facade_replace_targets_commits_runtime_state() { let (facade, notifier, replay_workers) = build_facade(); - let target = TestTarget::new("primary", "webhook"); + let target = MockTarget::new("primary", "webhook"); let activation = rustfs_targets::RuntimeActivation { replay_workers: ReplayWorkerManager::new(), targets: vec![Arc::new(target) as SharedTarget], @@ -594,9 +500,9 @@ mod tests { #[tokio::test] async fn runtime_queries_do_not_wait_for_target_close() { let (facade, notifier, replay_workers) = build_facade(); - let close_entered = Arc::new(Notify::new()); - let close_release = Arc::new(Notify::new()); - let target = TestTarget::new("primary", "webhook").with_blocking_close(close_entered.clone(), close_release.clone()); + let target = MockTarget::new("primary", "webhook"); + target.set_block_on_close(true); + let observer = target.clone(); facade .replace_targets(rustfs_targets::RuntimeActivation { replay_workers: ReplayWorkerManager::new(), @@ -609,12 +515,12 @@ mod tests { let facade = facade.clone(); async move { facade.shutdown_checked().await } }); - close_entered.notified().await; + observer.close_started().notified().await; let target_list = notifier.target_list(); assert!(target_list.try_read().is_ok(), "target list lock must not be held during close"); assert!(replay_workers.try_read().is_ok(), "replay manager lock must not be held during close"); - close_release.notify_one(); + observer.close_gate().add_permits(1); shutdown .await .expect("shutdown task should not panic") @@ -664,7 +570,7 @@ mod tests { #[tokio::test] async fn shutdown_returns_close_error_after_detaching_runtime() { let (facade, notifier, replay_workers) = build_facade(); - let target = TestTarget::new("primary", "webhook").with_close_error(); + let target = MockTarget::new("primary", "webhook").with_close_failures(usize::MAX); facade .replace_targets(rustfs_targets::RuntimeActivation { replay_workers: ReplayWorkerManager::new(), @@ -686,9 +592,10 @@ mod tests { #[tokio::test(start_paused = true)] async fn shutdown_bounds_a_target_that_never_closes() { let (facade, notifier, replay_workers) = build_facade(); - let close_entered = Arc::new(Notify::new()); - let never_release = Arc::new(Notify::new()); - let target = TestTarget::new("primary", "webhook").with_blocking_close(close_entered.clone(), never_release); + // The close gate never receives a permit, so this target's close blocks forever. + let target = MockTarget::new("primary", "webhook"); + target.set_block_on_close(true); + let observer = target.clone(); facade .replace_targets(rustfs_targets::RuntimeActivation { replay_workers: ReplayWorkerManager::new(), @@ -701,7 +608,7 @@ mod tests { let facade = facade.clone(); async move { facade.shutdown_checked().await } }); - close_entered.notified().await; + observer.close_started().notified().await; tokio::time::advance(super::TARGET_CLOSE_TIMEOUT).await; let err = shutdown diff --git a/crates/notify/src/runtime_view.rs b/crates/notify/src/runtime_view.rs index b17df852e..3ea587113 100644 --- a/crates/notify/src/runtime_view.rs +++ b/crates/notify/src/runtime_view.rs @@ -76,128 +76,13 @@ impl NotifyRuntimeView { mod tests { use super::NotifyRuntimeView; use crate::{Event, notifier::TargetList}; - use async_trait::async_trait; use rustfs_targets::arn::TargetID; - use rustfs_targets::store::{Key, Store}; - use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliverySnapshot}; - use rustfs_targets::{ReplayWorkerManager, StoreError, Target, TargetError}; - use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }; + use rustfs_targets::target::TargetDeliverySnapshot; + use rustfs_targets::testkit::MockTarget; + use rustfs_targets::{ReplayWorkerManager, Target}; + use std::sync::Arc; use tokio::sync::{Notify, RwLock}; - #[derive(Clone)] - struct TestTarget { - active: bool, - enabled: bool, - failed_messages: Arc, - failed_store_length: u64, - id: TargetID, - health_started: Option>, - health_release: Option>, - total_messages: Arc, - } - - impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - active: true, - enabled: true, - failed_messages: Arc::new(AtomicU64::new(0)), - failed_store_length: 0, - id: TargetID::new(id.to_string(), name.to_string()), - health_started: None, - health_release: None, - total_messages: Arc::new(AtomicU64::new(0)), - } - } - - fn with_active(mut self, active: bool) -> Self { - self.active = active; - self - } - - fn with_failed_store_length(mut self, failed_store_length: u64) -> Self { - self.failed_store_length = failed_store_length; - self - } - - fn with_enabled(mut self, enabled: bool) -> Self { - self.enabled = enabled; - self - } - - fn with_health_gate(mut self, started: Arc, release: Arc) -> Self { - self.health_started = Some(started); - self.health_release = Some(release); - self - } - - fn record_successes(&self, count: u64) { - self.total_messages.store(count, Ordering::Relaxed); - } - - fn record_failures(&self, count: u64) { - self.failed_messages.store(count, Ordering::Relaxed); - } - } - - #[async_trait] - impl Target for TestTarget - where - E: rustfs_targets::PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - if let (Some(started), Some(release)) = (&self.health_started, &self.health_release) { - started.notify_one(); - release.notified().await; - } - Ok(self.active) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn is_enabled(&self) -> bool { - self.enabled - } - - fn delivery_snapshot(&self) -> TargetDeliverySnapshot { - TargetDeliverySnapshot { - failed_messages: self.failed_messages.load(Ordering::Relaxed), - failed_store_length: self.failed_store_length, - queue_length: 0, - total_messages: self.total_messages.load(Ordering::Relaxed), - } - } - } - #[tokio::test] async fn runtime_view_reports_empty_runtime_queries() { let runtime_view = NotifyRuntimeView::new( @@ -230,12 +115,22 @@ mod tests { let target_list = Arc::new(RwLock::new(TargetList::new())); let replay_workers = Arc::new(RwLock::new(ReplayWorkerManager::new())); - let online = Arc::new(TestTarget::new("primary", "webhook").with_failed_store_length(7)); - online.record_successes(3); - online.record_failures(1); + let online = Arc::new(MockTarget::new("primary", "webhook").with_delivery_snapshot(TargetDeliverySnapshot { + failed_messages: 1, + failed_store_length: 7, + queue_length: 0, + total_messages: 3, + })); - let disabled = Arc::new(TestTarget::new("backup", "mqtt").with_enabled(false).with_active(false)); - disabled.record_successes(2); + let disabled = Arc::new( + MockTarget::new("backup", "mqtt") + .disabled() + .with_active(false) + .with_delivery_snapshot(TargetDeliverySnapshot { + total_messages: 2, + ..TargetDeliverySnapshot::default() + }), + ); { let mut targets = target_list.write().await; @@ -287,13 +182,13 @@ mod tests { #[tokio::test] async fn health_probe_does_not_hold_the_target_list_read_lock() { let target_list = Arc::new(RwLock::new(TargetList::new())); - let started = Arc::new(Notify::new()); let release = Arc::new(Notify::new()); - let target = Arc::new(TestTarget::new("blocked", "webhook").with_health_gate(started.clone(), release.clone())); + let target = MockTarget::new("blocked", "webhook").with_health_gate(release.clone()); + let started = target.health_started(); target_list .write() .await - .add(target as Arc + Send + Sync>) + .add(Arc::new(target) as Arc + Send + Sync>) .expect("test target should be added"); let runtime_view = NotifyRuntimeView::new(target_list.clone(), Arc::new(RwLock::new(ReplayWorkerManager::new()))); diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index 29b179e12..81f132b78 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -13,6 +13,9 @@ documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/" [features] default = [] +# Exposes the builder-style `testkit::MockTarget` to downstream test suites. Never enable this +# from a production `[dependencies]` entry; in-crate unit tests get the module via cfg(test). +test-support = [] hotpath = [ "hotpath/hotpath", "hotpath/tokio", diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index 7b9c77ce7..02b3bb945 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -26,6 +26,8 @@ pub mod runtime; pub mod store; pub mod sys; pub mod target; +#[cfg(any(test, feature = "test-support"))] +pub mod testkit; pub use catalog::extension::{ OPS_DIAGNOSTICS_EXTENSION_API_VERSION, OPS_PROFILER_EXTENSION_API_VERSION, S3_HOOK_EXTENSION_API_VERSION, diff --git a/crates/targets/src/plugin.rs b/crates/targets/src/plugin.rs index 43e5ba8ad..8973302be 100644 --- a/crates/targets/src/plugin.rs +++ b/crates/targets/src/plugin.rs @@ -421,61 +421,15 @@ where #[cfg(test)] mod tests { use super::{TargetPluginDescriptor, TargetPluginRegistry}; - use crate::PluginEvent; + use crate::TargetError; use crate::runtime::adapter::BuiltinPluginRuntimeAdapter; - use crate::store::{Key, Store}; - use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use crate::{StoreError, Target, TargetError}; - use async_trait::async_trait; + use crate::testkit::MockTarget; use rustfs_config::ENABLE_KEY; use rustfs_config::server_config::{Config, KVS}; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; - #[derive(Clone)] - struct TestTarget { - id: crate::arn::TargetID, - } - - #[async_trait] - impl Target for TestTarget - where - E: PluginEvent, - { - fn id(&self) -> crate::arn::TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - true - } - } - fn builtin_adapter() -> BuiltinPluginRuntimeAdapter { BuiltinPluginRuntimeAdapter::new( Arc::new(|_event| Box::pin(async {})), @@ -494,11 +448,7 @@ mod tests { "test", &[ENABLE_KEY, "endpoint"], |_config| Ok(()), - |id, _config| { - Ok(Box::new(TestTarget { - id: crate::arn::TargetID::new(id, "test".to_string()), - })) - }, + |id, _config| Ok(Box::new(MockTarget::new(&id, "test"))), )); let mut cfg = Config(HashMap::new()); @@ -532,11 +482,7 @@ mod tests { target_type, &[ENABLE_KEY, "endpoint"], |_config| Ok(()), - move |id, _config| { - Ok(Box::new(TestTarget { - id: crate::arn::TargetID::new(id, target_type.to_string()), - })) - }, + move |id, _config| Ok(Box::new(MockTarget::new(&id, target_type))), )); } diff --git a/crates/targets/src/runtime/adapter.rs b/crates/targets/src/runtime/adapter.rs index 87c2a23ff..6ebe996b2 100644 --- a/crates/targets/src/runtime/adapter.rs +++ b/crates/targets/src/runtime/adapter.rs @@ -441,13 +441,10 @@ where #[cfg(test)] mod tests { use super::{BuiltinPluginRuntimeAdapter, MAX_PARALLEL_STORE_OPENS, PluginRuntimeAdapter}; - use crate::PluginEvent; - use crate::arn::TargetID; use crate::store::{Key, QueueStore, Store}; - use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use crate::{StoreError, Target, TargetError}; - use async_trait::async_trait; - use std::sync::atomic::{AtomicUsize, Ordering}; + use crate::target::QueuedPayload; + use crate::testkit::MockTarget; + use crate::{StoreError, Target}; use std::sync::{Arc, Condvar, Mutex}; use std::time::Duration; use tempfile::tempdir; @@ -569,96 +566,14 @@ mod tests { } } - #[derive(Clone)] - struct TestTarget { - close_calls: Arc, - id: TargetID, - init_calls: Arc, - init_entered: Option>, - init_fails: bool, - store: Option>, - } - - impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - close_calls: Arc::new(AtomicUsize::new(0)), - id: TargetID::new(id.to_string(), name.to_string()), - init_calls: Arc::new(AtomicUsize::new(0)), - init_entered: None, - init_fails: false, - store: None, - } - } - - fn with_failed_init(mut self) -> Self { - self.init_fails = true; - self - } - - fn with_pending_init(mut self, init_entered: Arc) -> Self { - self.init_entered = Some(init_entered); - self - } - - fn with_store(mut self) -> Self { - let dir = tempdir().expect("tempdir should be created for queue store tests"); - let store = QueueStore::::new(dir.path(), 16, ".queue"); - store.open().expect("queue store should open"); - self.store = Some(Arc::new(store)); - self - } - } - - #[async_trait] - impl Target for TestTarget - where - E: PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - self.store.as_deref() - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - async fn init(&self) -> Result<(), TargetError> { - self.init_calls.fetch_add(1, Ordering::SeqCst); - if let Some(init_entered) = &self.init_entered { - init_entered.notify_one(); - return std::future::pending().await; - } - if self.init_fails { - return Err(TargetError::Configuration("forced init failure".to_string())); - } - Ok(()) - } - - fn is_enabled(&self) -> bool { - true - } + /// Builds the tempdir-backed, already-opened queue store the store-backed mock targets use. + /// The tempdir handle is dropped here on purpose, matching the previous in-module mock: these + /// tests never write through the store, they only need an openable handle. + fn opened_queue_store() -> Arc { + let dir = tempdir().expect("tempdir should be created for queue store tests"); + let store = QueueStore::::new(dir.path(), 16, ".queue"); + store.open().expect("queue store should open"); + Arc::new(store) } fn builtin_adapter() -> BuiltinPluginRuntimeAdapter { @@ -684,7 +599,7 @@ mod tests { #[tokio::test] async fn builtin_adapter_skips_non_store_target_when_init_fails() { let adapter = builtin_adapter(); - let target = TestTarget::new("primary", "webhook").with_failed_init(); + let target = MockTarget::new("primary", "webhook").with_init_failures(usize::MAX); let activation = adapter.activate_with_replay(vec![Box::new(target)]).await; @@ -695,7 +610,9 @@ mod tests { #[tokio::test] async fn builtin_adapter_keeps_store_backed_target_when_init_fails() { let adapter = builtin_adapter(); - let target = TestTarget::new("primary", "webhook").with_failed_init().with_store(); + let target = MockTarget::new("primary", "webhook") + .with_init_failures(usize::MAX) + .with_store(opened_queue_store()); let activation = adapter.activate_with_replay(vec![Box::new(target)]).await; @@ -706,7 +623,9 @@ mod tests { #[tokio::test] async fn prepared_store_target_reports_init_failure_without_dropping_queue_runtime() { let adapter = builtin_adapter(); - let target = TestTarget::new("primary", "webhook").with_failed_init().with_store(); + let target = MockTarget::new("primary", "webhook") + .with_init_failures(usize::MAX) + .with_store(opened_queue_store()); let prepared = adapter.prepare_targets(vec![Box::new(target)]).await; assert_eq!(prepared.targets.len(), 1); @@ -729,11 +648,10 @@ mod tests { async fn cancellable_preparation_returns_current_and_remaining_targets_for_shutdown() { let adapter = builtin_adapter(); let init_entered = Arc::new(Notify::new()); - let first = TestTarget::new("first", "webhook").with_pending_init(init_entered.clone()); - let first_close_calls = first.close_calls.clone(); - let second = TestTarget::new("second", "webhook"); - let second_close_calls = second.close_calls.clone(); - let second_init_calls = second.init_calls.clone(); + let first = MockTarget::new("first", "webhook").with_blocking_init(init_entered.clone()); + let first_observer = first.clone(); + let second = MockTarget::new("second", "webhook"); + let second_observer = second.clone(); let cancellation = CancellationToken::new(); let prepare_adapter = adapter.clone(); let prepare_cancellation = cancellation.clone(); @@ -755,9 +673,9 @@ mod tests { .close_prepared(prepared) .await .expect("cancelled targets should close"); - assert_eq!(first_close_calls.load(Ordering::SeqCst), 1); - assert_eq!(second_close_calls.load(Ordering::SeqCst), 1); - assert_eq!(second_init_calls.load(Ordering::SeqCst), 0); + assert_eq!(first_observer.close_call_count(), 1); + assert_eq!(second_observer.close_call_count(), 1); + assert_eq!(second_observer.init_call_count(), 0); } #[tokio::test] @@ -765,8 +683,11 @@ mod tests { let adapter = builtin_adapter(); let dir = tempdir().expect("tempdir should be created"); let queue_path = dir.path().join("queue"); - let mut target = TestTarget::new("primary", "webhook"); - target.store = Some(Arc::new(QueueStore::::new(&queue_path, 16, ".queue"))); + let target = MockTarget::new("primary", "webhook").with_store(Arc::new(QueueStore::::new( + &queue_path, + 16, + ".queue", + ))); let prepared = adapter.prepare_targets(vec![Box::new(target)]).await; assert!(!queue_path.exists(), "dormant preparation must not open the queue store"); @@ -789,15 +710,18 @@ mod tests { let dir = tempdir().expect("tempdir should be created"); let invalid_base = dir.path().join("not-a-directory"); std::fs::write(&invalid_base, b"file").expect("invalid queue base should be created"); - let mut target = TestTarget::new("primary", "webhook"); - let close_calls = target.close_calls.clone(); - target.store = Some(Arc::new(QueueStore::::new(&invalid_base, 16, ".queue"))); + let target = MockTarget::new("primary", "webhook").with_store(Arc::new(QueueStore::::new( + &invalid_base, + 16, + ".queue", + ))); + let observer = target.clone(); let activation = adapter.activate_with_replay(vec![Box::new(target)]).await; assert!(activation.targets.is_empty()); assert!(activation.replay_workers.is_empty()); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } #[tokio::test] @@ -810,14 +734,13 @@ mod tests { let mut targets: Vec + Send + Sync>> = Vec::with_capacity(TARGETS); let mut expected_ids = Vec::with_capacity(TARGETS); for index in 0..TARGETS { - let mut target = TestTarget::new(&format!("target-{index}"), "webhook"); - expected_ids.push(target.id.to_string()); let open_gate = gate.clone(); - target.store = Some(Arc::new(TestOpenStore { + let target = MockTarget::new(&format!("target-{index}"), "webhook").with_store(Arc::new(TestOpenStore { before_clone: Arc::new(|| {}), before_open: Arc::new(move || open_gate.enter()), store: QueueStore::new(dir.path().join(index.to_string()), 16, ".queue"), })); + expected_ids.push(target.target_id().to_string()); targets.push(Box::new(target)); } @@ -848,13 +771,12 @@ mod tests { async fn panicking_store_open_rejects_and_closes_only_that_target() { let adapter = builtin_adapter(); let dir = tempdir().expect("tempdir should be created"); - let mut target = TestTarget::new("panicking", "webhook"); - let close_calls = target.close_calls.clone(); - target.store = Some(Arc::new(TestOpenStore { + let target = MockTarget::new("panicking", "webhook").with_store(Arc::new(TestOpenStore { before_clone: Arc::new(|| {}), before_open: Arc::new(|| panic!("forced store open panic: do-not-expose-payload")), store: QueueStore::new(dir.path(), 16, ".queue"), })); + let observer = target.clone(); let prepared = adapter.prepare_targets(vec![Box::new(target)]).await; let (opened, rejected) = adapter.open_prepared_stores(prepared); @@ -875,20 +797,19 @@ mod tests { .close_prepared(rejected) .await .expect("a target rejected after a store panic should close"); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } #[tokio::test] async fn panicking_store_clone_cannot_publish_target_without_replay_worker() { let adapter = builtin_adapter(); let dir = tempdir().expect("tempdir should be created"); - let mut target = TestTarget::new("panicking-clone", "webhook"); - let close_calls = target.close_calls.clone(); - target.store = Some(Arc::new(TestOpenStore { + let target = MockTarget::new("panicking-clone", "webhook").with_store(Arc::new(TestOpenStore { before_clone: Arc::new(|| panic!("forced store clone panic: do-not-expose-payload")), before_open: Arc::new(|| {}), store: QueueStore::new(dir.path(), 16, ".queue"), })); + let observer = target.clone(); let prepared = adapter.prepare_targets(vec![Box::new(target)]).await; let (opened, open_rejected) = adapter.open_prepared_stores(prepared); @@ -906,14 +827,14 @@ mod tests { .close_prepared(rejected) .await .expect("a target rejected during replay activation should close"); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } #[tokio::test] async fn builtin_adapter_shutdown_clears_runtime_and_replay_workers() { let adapter = builtin_adapter(); - let target = TestTarget::new("primary", "webhook"); - let close_calls = Arc::clone(&target.close_calls); + let target = MockTarget::new("primary", "webhook"); + let observer = target.clone(); let mut runtime = crate::runtime::TargetRuntimeManager::new(); let mut replay_workers = crate::runtime::ReplayWorkerManager::new(); @@ -933,6 +854,6 @@ mod tests { assert!(runtime.is_empty()); assert!(replay_workers.is_empty()); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } } diff --git a/crates/targets/src/runtime/mod.rs b/crates/targets/src/runtime/mod.rs index 7999a33ce..d6451df94 100644 --- a/crates/targets/src/runtime/mod.rs +++ b/crates/targets/src/runtime/mod.rs @@ -965,17 +965,12 @@ where #[cfg(test)] mod tests { use super::{HEALTH_PROBE_CONCURRENCY, TargetRuntimeManager, health_snapshots_for_targets}; - use crate::PluginEvent; - use crate::StoreError; - use crate::arn::TargetID; - use crate::store::{Key, Store}; - use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use crate::{SharedTarget, Target, TargetError}; - use async_trait::async_trait; + use crate::SharedTarget; + use crate::store::Key; + use crate::testkit::MockTarget; use std::sync::Arc; - use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::atomic::Ordering; use std::time::Duration; - use tokio::sync::{Notify, Semaphore}; #[tokio::test(start_paused = true)] async fn seed_interval_start_backdates_by_one_interval() { @@ -1037,108 +1032,11 @@ mod tests { ); } - #[derive(Clone)] - struct TestTarget { - id: TargetID, - block_on_close: Arc, - close_gate: Arc, - close_calls: Arc, - enabled: bool, - health_delay: Duration, - health_drops: Arc, - health_started: Arc, - close_started: Arc, - } - - struct HealthDropGuard(Arc); - - impl Drop for HealthDropGuard { - fn drop(&mut self) { - self.0.fetch_add(1, Ordering::SeqCst); - } - } - - impl TestTarget { - fn new(id: &str, name: &str) -> Self { - Self { - id: TargetID::new(id.to_string(), name.to_string()), - block_on_close: Arc::new(AtomicBool::new(false)), - close_gate: Arc::new(Semaphore::new(0)), - close_calls: Arc::new(AtomicUsize::new(0)), - enabled: true, - health_delay: Duration::ZERO, - health_drops: Arc::new(AtomicUsize::new(0)), - health_started: Arc::new(Notify::new()), - close_started: Arc::new(Notify::new()), - } - } - - fn with_health_delay(id: &str, delay: Duration) -> Self { - Self { - health_delay: delay, - ..Self::new(id, "webhook") - } - } - - fn disabled(id: &str) -> Self { - Self { - enabled: false, - ..Self::new(id, "webhook") - } - } - } - - #[async_trait] - impl Target for TestTarget - where - E: PluginEvent, - { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - self.health_started.notify_one(); - let _drop_guard = HealthDropGuard(Arc::clone(&self.health_drops)); - tokio::time::sleep(self.health_delay).await; - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - self.close_calls.fetch_add(1, Ordering::SeqCst); - self.close_started.notify_one(); - if self.block_on_close.load(Ordering::SeqCst) { - let _permit = self.close_gate.acquire().await.expect("close gate should remain open"); - } - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - None - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - self.enabled - } - } - #[tokio::test] async fn runtime_manager_removes_and_closes_target() { let mut manager = TargetRuntimeManager::::new(); - let target = TestTarget::new("primary", "webhook"); - let close_calls = Arc::clone(&target.close_calls); + let target = MockTarget::new("primary", "webhook"); + let observer = target.clone(); manager.add_boxed(Box::new(target)); assert_eq!(manager.len(), 1); @@ -1146,14 +1044,14 @@ mod tests { let removed = manager.remove_and_close("primary:webhook").await; assert!(removed.is_some()); assert_eq!(manager.len(), 0); - assert_eq!(close_calls.load(Ordering::SeqCst), 1); + assert_eq!(observer.close_call_count(), 1); } #[tokio::test(start_paused = true)] async fn runtime_manager_starts_all_target_closes_before_waiting_for_completion() { let mut manager = TargetRuntimeManager::::new(); - let first = TestTarget::new("first", "webhook"); - let second = TestTarget::new("second", "webhook"); + let first = MockTarget::new("first", "webhook"); + let second = MockTarget::new("second", "webhook"); let first_observer = first.clone(); let second_observer = second.clone(); manager.add_boxed(Box::new(first)); @@ -1164,34 +1062,34 @@ mod tests { .into_iter() .next() .expect("two targets should have a first close key"); - let (blocked, unblocked) = if first_close_key == first_observer.id.to_string() { + let (blocked, unblocked) = if first_close_key == first_observer.target_id().to_string() { (first_observer, second_observer) } else { (second_observer, first_observer) }; - blocked.block_on_close.store(true, Ordering::SeqCst); + blocked.set_block_on_close(true); let close_task = tokio::spawn(async move { manager.clear_and_close().await }); - tokio::time::timeout(std::time::Duration::from_secs(1), blocked.close_started.notified()) + tokio::time::timeout(std::time::Duration::from_secs(1), blocked.close_started().notified()) .await .expect("the first target close should start"); - tokio::time::timeout(std::time::Duration::from_secs(1), unblocked.close_started.notified()) + tokio::time::timeout(std::time::Duration::from_secs(1), unblocked.close_started().notified()) .await .expect("a blocked first close must not prevent the second close from starting"); assert!(!close_task.is_finished(), "clear_and_close must still await the blocked target"); - blocked.close_gate.add_permits(1); + blocked.close_gate().add_permits(1); let errors = close_task.await.expect("clear_and_close task should join"); assert!(errors.is_empty()); - assert_eq!(blocked.close_calls.load(Ordering::SeqCst), 1); - assert_eq!(unblocked.close_calls.load(Ordering::SeqCst), 1); + assert_eq!(blocked.close_call_count(), 1); + assert_eq!(unblocked.close_call_count(), 1); } #[test] fn runtime_manager_snapshots_targets() { let mut manager = TargetRuntimeManager::::new(); - manager.add_boxed(Box::new(TestTarget::new("primary", "webhook"))); + manager.add_boxed(Box::new(MockTarget::new("primary", "webhook"))); let snapshots = manager.snapshots(); assert_eq!(snapshots.len(), 1); @@ -1202,7 +1100,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn health_snapshot_allows_a_four_second_probe() { let mut manager = TargetRuntimeManager::::new(); - manager.add_boxed(Box::new(TestTarget::with_health_delay("slow", Duration::from_secs(4)))); + manager.add_boxed(Box::new(MockTarget::new("slow", "webhook").with_health_delay(Duration::from_secs(4)))); let snapshots = manager.health_snapshots().await; @@ -1214,7 +1112,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn health_snapshot_times_out_after_five_seconds() { let mut manager = TargetRuntimeManager::::new(); - manager.add_boxed(Box::new(TestTarget::with_health_delay("stalled", Duration::from_secs(6)))); + manager.add_boxed(Box::new(MockTarget::new("stalled", "webhook").with_health_delay(Duration::from_secs(6)))); let snapshots = manager.health_snapshots().await; @@ -1227,10 +1125,9 @@ mod tests { async fn health_collection_deadline_does_not_scale_with_target_count() { let mut manager = TargetRuntimeManager::::new(); for index in 0..24 { - manager.add_boxed(Box::new(TestTarget::with_health_delay( - &format!("stalled-{index}"), - Duration::from_secs(30), - ))); + manager.add_boxed(Box::new( + MockTarget::new(&format!("stalled-{index}"), "webhook").with_health_delay(Duration::from_secs(30)), + )); } let started = tokio::time::Instant::now(); @@ -1263,11 +1160,11 @@ mod tests { async fn disabled_target_does_not_wait_for_probe_capacity() { let mut targets: Vec> = (0..HEALTH_PROBE_CONCURRENCY) .map(|index| { - Arc::new(TestTarget::with_health_delay(&format!("stalled-{index}"), Duration::from_secs(30))) + Arc::new(MockTarget::new(&format!("stalled-{index}"), "webhook").with_health_delay(Duration::from_secs(30))) as SharedTarget }) .collect(); - targets.push(Arc::new(TestTarget::disabled("disabled"))); + targets.push(Arc::new(MockTarget::new("disabled", "webhook").disabled())); let snapshots = health_snapshots_for_targets(targets).await; let disabled = snapshots @@ -1281,20 +1178,19 @@ mod tests { #[tokio::test] async fn cancelling_health_collection_drops_in_flight_probe() { - let target = TestTarget::with_health_delay("slow", Duration::from_secs(30)); - let health_drops = Arc::clone(&target.health_drops); - let health_started = Arc::clone(&target.health_started); + let target = MockTarget::new("slow", "webhook").with_health_delay(Duration::from_secs(30)); + let observer = target.clone(); let targets: Vec> = vec![Arc::new(target)]; let collector = tokio::spawn(health_snapshots_for_targets(targets)); - tokio::time::timeout(Duration::from_secs(1), health_started.notified()) + tokio::time::timeout(Duration::from_secs(1), observer.health_started().notified()) .await .expect("health probe should start"); collector.abort(); let join_error = collector.await.expect_err("health collector should be cancelled"); assert!(join_error.is_cancelled()); - assert_eq!(health_drops.load(Ordering::SeqCst), 1); + assert_eq!(observer.health_drop_count(), 1); } #[tokio::test] diff --git a/crates/targets/src/target/mod.rs b/crates/targets/src/target/mod.rs index def914ce6..1d6f48154 100644 --- a/crates/targets/src/target/mod.rs +++ b/crates/targets/src/target/mod.rs @@ -960,76 +960,29 @@ pub(crate) fn ensure_rustls_provider_installed() { #[cfg(test)] pub(crate) mod test_support { - use super::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use crate::arn::TargetID; - use crate::store::{FailedEventStore, Key, QueueStore, Store}; - use crate::{StoreError, Target, TargetError}; - use async_trait::async_trait; + use super::{QueuedPayload, QueuedPayloadMeta}; + use crate::Target; + use crate::store::QueueStore; + use crate::testkit::MockTarget; use rustfs_s3_types::EventName; use std::path::PathBuf; use std::sync::Arc; - use std::sync::atomic::{AtomicU64, Ordering}; use uuid::Uuid; - /// A minimal target for failed-store move tests: every delivery method succeeds, the optional - /// store backs the store and failed-store accessors, and final failures land on a shared counter. - #[derive(Clone)] - pub(crate) struct MoveTestTarget { - pub(crate) id: TargetID, - pub(crate) store: Option>>, - pub(crate) failed: Arc, - } - - #[async_trait] - impl Target for MoveTestTarget { - fn id(&self) -> TargetID { - self.id.clone() - } - async fn is_active(&self) -> Result { - Ok(true) - } - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - self.store - .as_deref() - .map(|store| store as &(dyn Store<_, Error = StoreError, Key = Key> + Send + Sync)) - } - fn failed_store(&self) -> Option<&dyn FailedEventStore> { - self.store.as_deref().map(|store| store as &dyn FailedEventStore) - } - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - fn is_enabled(&self) -> bool { - true - } - fn record_final_failure(&self) { - self.failed.fetch_add(1, Ordering::Relaxed); - } - } - + /// A minimal target for failed-store move tests: every delivery method succeeds, no store is + /// attached, and final failures land on the mock's shared counter. pub(crate) fn move_test_target() -> Arc + Send + Sync> { - Arc::new(MoveTestTarget { - id: TargetID::new("target-a".to_string(), "nats".to_string()), - store: None, - failed: Arc::new(AtomicU64::new(0)), - }) + Arc::new(MockTarget::new("target-a", "nats")) } + /// Like [`move_test_target`], but the given store backs both the store and failed-store + /// accessors, matching a target whose live queue also parks terminal failures. pub(crate) fn move_test_target_with_store(store: Arc>) -> Arc + Send + Sync> { - Arc::new(MoveTestTarget { - id: TargetID::new("target-a".to_string(), "nats".to_string()), - store: Some(store), - failed: Arc::new(AtomicU64::new(0)), - }) + Arc::new( + MockTarget::new("target-a", "nats") + .with_store(store.clone()) + .with_failed_store(store), + ) } pub(crate) fn failed_store_dir(name: &str) -> PathBuf { @@ -1472,47 +1425,6 @@ mod tests { assert!(dir.starts_with("rustfs-redis-tenant_alpha-"), "unexpected subdir: {dir}"); } - #[derive(Clone)] - struct StoreBackedTarget { - id: TargetID, - store: QueueStore, - } - - #[async_trait] - impl Target for StoreBackedTarget { - fn id(&self) -> TargetID { - self.id.clone() - } - - async fn is_active(&self) -> Result { - Ok(true) - } - - async fn save(&self, _event: Arc>) -> Result<(), TargetError> { - Ok(()) - } - - async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { - Ok(()) - } - - async fn close(&self) -> Result<(), TargetError> { - Ok(()) - } - - fn store(&self) -> Option<&(dyn Store + Send + Sync)> { - Some(&self.store) - } - - fn clone_dyn(&self) -> Box + Send + Sync> { - Box::new(self.clone()) - } - - fn is_enabled(&self) -> bool { - true - } - } - #[tokio::test] async fn send_from_store_purges_missing_or_empty_entry() { let dir = std::env::temp_dir().join(format!("rustfs-send-from-store-{}", Uuid::new_v4())); @@ -1541,10 +1453,10 @@ mod tests { .expect("event file should exist"); std::fs::write(&event_file, b"").unwrap(); - let target = StoreBackedTarget { - id: TargetID::new("primary".to_string(), "webhook".to_string()), - store: store.clone(), - }; + // The default send_from_store implementation is under test here, so the mock must not + // override it; it only supplies the backing store. + let target: Box + Send + Sync> = + Box::new(crate::testkit::MockTarget::new("primary", "webhook").with_store(Arc::new(store.clone()))); // A NotFound/empty entry must be purged (index + file) rather than // silently skipped and replayed forever. diff --git a/crates/targets/src/target/nats/jetstream.rs b/crates/targets/src/target/nats/jetstream.rs index 663525e9a..eeb68af82 100644 --- a/crates/targets/src/target/nats/jetstream.rs +++ b/crates/targets/src/target/nats/jetstream.rs @@ -333,18 +333,15 @@ pub(crate) fn retry_lifetime(ack_timeout: Duration) -> Duration { mod tests { use super::*; use crate::Target; - use crate::arn::TargetID; use crate::store::{FailedEventStore, QueueStore}; use crate::target::TargetType; use crate::target::nats::test_support::*; use crate::target::nats::validation::STREAM_VALIDATION_FAILED_DETAIL; - use crate::target::test_support::{ - MoveTestTarget, failed_store_dir, move_test_target, move_test_target_with_store, sample_queued, - }; + use crate::target::test_support::{failed_store_dir, move_test_target, move_test_target_with_store, sample_queued}; use crate::target::{build_target_tls_fingerprint, persist_queued_payload_to_store}; + use crate::testkit::MockTarget; use async_nats::jetstream::context::PublishError; use rustfs_config::NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS; - use std::sync::atomic::AtomicU64; use uuid::Uuid; #[test] @@ -1210,12 +1207,11 @@ mod tests { let store = Arc::new(QueueStore::::new_with_compression(&dir, 8, ".test", false)); store.open().unwrap(); - let failed = Arc::new(AtomicU64::new(0)); - let target: Arc + Send + Sync> = Arc::new(MoveTestTarget { - id: TargetID::new("target-a".to_string(), "nats".to_string()), - store: Some(store.clone()), - failed: failed.clone(), - }); + let mock = MockTarget::new("target-a", "nats") + .with_store(store.clone()) + .with_failed_store(store.clone()); + let observer = mock.clone(); + let target: Arc + Send + Sync> = Arc::new(mock); let key = store.put_raw(&sample_queued("minted-id").encode().unwrap()).unwrap(); let error = TargetError::JetStreamPublish { @@ -1225,11 +1221,11 @@ mod tests { move_entry_to_failed_store(&*store, &*store, &target.id(), &key, &error, 0) .await .unwrap(); - assert_eq!(failed.load(Ordering::Relaxed), 0, "the move itself does not count the failure"); + assert_eq!(observer.final_failure_count(), 0, "the move itself does not count the failure"); // The replay worker follows every move with one hook emit that records the failure. target.record_final_failure(); - assert_eq!(failed.load(Ordering::Relaxed), 1, "one failed delivery counts exactly once"); + assert_eq!(observer.final_failure_count(), 1, "one failed delivery counts exactly once"); let _ = store.delete(); } diff --git a/crates/targets/src/testkit.rs b/crates/targets/src/testkit.rs new file mode 100644 index 000000000..c23eacad2 --- /dev/null +++ b/crates/targets/src/testkit.rs @@ -0,0 +1,600 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Builder-style [`Target`] mock shared by this crate's unit tests and, behind the off-by-default +//! `test-support` cargo feature, by downstream test suites. +//! +//! The module is compiled only under `cfg(test)` or when a dependent explicitly opts in via the +//! `test-support` feature, so the mock can never reach a production binary. Every knob defaults +//! off: a plain [`MockTarget::new`] is an enabled, reachable, storeless target whose delivery +//! methods all succeed immediately. The mock emits no tracing events. + +use crate::arn::TargetID; +use crate::plugin::PluginEvent; +use crate::store::{FailedEventStore, Key, Store}; +use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliverySnapshot}; +use crate::{StoreError, Target, TargetError}; +use async_trait::async_trait; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::time::Duration; +use tokio::sync::{Notify, Semaphore}; + +/// The queued-payload store handle a [`MockTarget`] serves from its [`Target::store`] accessor. +pub type SharedQueuedStore = Arc + Send + Sync>; + +/// Builds the error a failing mock operation returns, so a test that pins an error variant can +/// shape the failure instead of matching the mock's default. +pub type ErrorFactory = Arc TargetError + Send + Sync>; + +/// Increments its counter when dropped, however the owning future ends, so a test that holds the +/// probe open past its caller's deadline can prove the cancelled health future was actually +/// dropped instead of left running. +struct HealthDropGuard(Arc); + +impl Drop for HealthDropGuard { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +/// Consumes one unit of a failure budget, returning true while the budget is not exhausted. +/// A budget of `usize::MAX` behaves as "always fail" for any realistic call count. +fn consume_failure_budget(budget: &AtomicUsize) -> bool { + budget + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .is_ok() +} + +/// A configurable mock implementation of [`Target`]. +/// +/// All observable state (call counters, gates, signals) lives behind shared handles, so a clone +/// kept aside keeps observing the original after it is boxed into a runtime, and +/// [`Target::clone_dyn`] clones observe the same counters. Builder methods consume `self` and +/// each knob defaults off; accessors read the shared state from any clone. +#[derive(Clone)] +pub struct MockTarget { + id: TargetID, + enabled: bool, + /// Overrides the [`Target::is_active`] result; defaults to the enabled flag. + active: Option, + health_delay: Duration, + health_started: Arc, + /// When set, `is_active` waits on this handle (after notifying `health_started`) before + /// answering, so a test can hold a probe in flight and release it on demand. + health_gate: Option>, + health_drops: Arc, + enabled_calls: Arc, + init_calls: Arc, + init_failures_remaining: Arc, + /// When set, `init` notifies the handle on entry and then never returns. + blocking_init: Option>, + close_calls: Arc, + close_started: Arc, + block_on_close: Arc, + close_gate: Arc, + close_failures_remaining: Arc, + close_failure_error: Option, + save_calls: Arc, + save_failures_remaining: Arc, + /// When set, the first `save` (counted across all clones) notifies the first handle and then + /// waits on the second before returning; later saves pass straight through. + first_save_gate: Option<(Arc, Arc)>, + final_failures: Arc, + delivery_snapshot: Option, + store: Option, + failed_store: Option>, +} + +impl MockTarget { + /// Creates an enabled, reachable, storeless mock identified as `:`. + pub fn new(id: &str, name: &str) -> Self { + Self { + id: TargetID::new(id.to_string(), name.to_string()), + enabled: true, + active: None, + health_delay: Duration::ZERO, + health_started: Arc::new(Notify::new()), + health_gate: None, + health_drops: Arc::new(AtomicUsize::new(0)), + enabled_calls: Arc::new(AtomicUsize::new(0)), + init_calls: Arc::new(AtomicUsize::new(0)), + init_failures_remaining: Arc::new(AtomicUsize::new(0)), + blocking_init: None, + close_calls: Arc::new(AtomicUsize::new(0)), + close_started: Arc::new(Notify::new()), + block_on_close: Arc::new(AtomicBool::new(false)), + close_gate: Arc::new(Semaphore::new(0)), + close_failures_remaining: Arc::new(AtomicUsize::new(0)), + close_failure_error: None, + save_calls: Arc::new(AtomicUsize::new(0)), + save_failures_remaining: Arc::new(AtomicUsize::new(0)), + first_save_gate: None, + final_failures: Arc::new(AtomicU64::new(0)), + delivery_snapshot: None, + store: None, + failed_store: None, + } + } + + /// Replaces the mock's identity while keeping every shared counter and gate, so a plugin + /// factory can clone one observed template per constructed instance and bind the instance id + /// the registry hands it. + pub fn with_id(mut self, id: &str, name: &str) -> Self { + self.id = TargetID::new(id.to_string(), name.to_string()); + self + } + + /// Marks the target disabled: `is_enabled` returns false and `health` short-circuits. + pub fn disabled(mut self) -> Self { + self.enabled = false; + self + } + + /// Overrides the `is_active` result independently of the enabled flag. + pub fn with_active(mut self, active: bool) -> Self { + self.active = Some(active); + self + } + + /// Makes `is_active` sleep for `delay` (after notifying [`Self::health_started`]) before + /// answering, so probe timeouts and cancellation can be exercised under a paused clock. + pub fn with_health_delay(mut self, delay: Duration) -> Self { + self.health_delay = delay; + self + } + + /// Makes `is_active` wait on `release` after notifying [`Self::health_started`], so a test + /// can hold a probe in flight and release it on demand. Independent of the health delay. + pub fn with_health_gate(mut self, release: Arc) -> Self { + self.health_gate = Some(release); + self + } + + /// Fails the first `failures` `init` calls with [`TargetError::Initialization`], then + /// succeeds. Pass `usize::MAX` for a target whose init always fails. + pub fn with_init_failures(mut self, failures: usize) -> Self { + self.init_failures_remaining = Arc::new(AtomicUsize::new(failures)); + self + } + + /// Makes `init` notify `entered` and then never return, so cancellation of an in-flight + /// initialization can be exercised. + pub fn with_blocking_init(mut self, entered: Arc) -> Self { + self.blocking_init = Some(entered); + self + } + + /// Fails the first `failures` `save` calls with [`TargetError::Request`], then succeeds. + /// Pass `usize::MAX` for a target whose save always fails. + pub fn with_save_failures(mut self, failures: usize) -> Self { + self.save_failures_remaining = Arc::new(AtomicUsize::new(failures)); + self + } + + /// Gates the first `save` (counted across all clones): it notifies `entered` and then waits on + /// `release` before returning. Later saves pass straight through. Several mocks may share one + /// `entered`/`release` pair to gate their first saves collectively. + pub fn with_first_save_gate(mut self, entered: Arc, release: Arc) -> Self { + self.first_save_gate = Some((entered, release)); + self + } + + /// Fails the first `failures` `close` calls with [`TargetError::Storage`] (or the error shaped + /// by [`Self::with_close_failure_error`]), then succeeds. Pass `usize::MAX` for a target whose + /// close always fails. Counting, [`Self::close_started`], and the close gate still run before + /// the failure fires. + pub fn with_close_failures(mut self, failures: usize) -> Self { + self.close_failures_remaining = Arc::new(AtomicUsize::new(failures)); + self + } + + /// Shapes the error a failing `close` returns, for tests that pin the error variant. + pub fn with_close_failure_error(mut self, factory: impl Fn() -> TargetError + Send + Sync + 'static) -> Self { + self.close_failure_error = Some(Arc::new(factory)); + self + } + + /// Serves a fixed snapshot from `delivery_snapshot()` instead of deriving one from the + /// attached stores. + pub fn with_delivery_snapshot(mut self, snapshot: TargetDeliverySnapshot) -> Self { + self.delivery_snapshot = Some(snapshot); + self + } + + /// Serves `store` from the [`Target::store`] accessor. The caller owns the backing store and + /// its directory lifecycle; the mock only hands out the reference. + pub fn with_store(mut self, store: SharedQueuedStore) -> Self { + self.store = Some(store); + self + } + + /// Serves `failed_store` from the [`Target::failed_store`] accessor. Pass the same underlying + /// store as [`Self::with_store`] to model a target whose live queue also parks terminal + /// failures. + pub fn with_failed_store(mut self, failed_store: Arc) -> Self { + self.failed_store = Some(failed_store); + self + } + + /// Returns the mock's identity without needing an event-type annotation. + pub fn target_id(&self) -> TargetID { + self.id.clone() + } + + /// When `block` is set, `close` waits on [`Self::close_gate`] after counting and signalling, + /// until the gate receives a permit. Takes effect for closes that start after the call. + pub fn set_block_on_close(&self, block: bool) { + self.block_on_close.store(block, Ordering::SeqCst); + } + + /// The semaphore a blocked `close` waits on; add a permit to release it. + pub fn close_gate(&self) -> Arc { + Arc::clone(&self.close_gate) + } + + /// Notified once every time a `close` call starts. + pub fn close_started(&self) -> Arc { + Arc::clone(&self.close_started) + } + + /// Notified once every time an `is_active` probe starts. + pub fn health_started(&self) -> Arc { + Arc::clone(&self.health_started) + } + + /// How many times `init` was called across all clones. + pub fn init_call_count(&self) -> usize { + self.init_calls.load(Ordering::SeqCst) + } + + /// How many times `close` was called across all clones. + pub fn close_call_count(&self) -> usize { + self.close_calls.load(Ordering::SeqCst) + } + + /// How many times `save` was called across all clones. + pub fn save_call_count(&self) -> usize { + self.save_calls.load(Ordering::SeqCst) + } + + /// How many times `is_enabled` was called across all clones, so a test can assert whether a + /// dispatcher consulted (selected) this target at all. + pub fn enabled_call_count(&self) -> usize { + self.enabled_calls.load(Ordering::SeqCst) + } + + /// How many `is_active` futures finished or were dropped mid-flight across all clones. + pub fn health_drop_count(&self) -> usize { + self.health_drops.load(Ordering::SeqCst) + } + + /// How many final delivery failures were recorded across all clones. + pub fn final_failure_count(&self) -> u64 { + self.final_failures.load(Ordering::Relaxed) + } +} + +#[async_trait] +impl Target for MockTarget +where + E: PluginEvent, +{ + fn id(&self) -> TargetID { + self.id.clone() + } + + async fn is_active(&self) -> Result { + self.health_started.notify_one(); + // Held across the gate and delay so an aborted probe future is observable via + // health_drop_count. + let _drop_guard = HealthDropGuard(Arc::clone(&self.health_drops)); + if let Some(release) = &self.health_gate { + release.notified().await; + } + tokio::time::sleep(self.health_delay).await; + Ok(self.active.unwrap_or(self.enabled)) + } + + async fn save(&self, _event: Arc>) -> Result<(), TargetError> { + let call = self.save_calls.fetch_add(1, Ordering::SeqCst); + if call == 0 + && let Some((entered, release)) = &self.first_save_gate + { + entered.notify_one(); + release.notified().await; + } + if consume_failure_budget(&self.save_failures_remaining) { + return Err(TargetError::Request("forced save failure".to_string())); + } + Ok(()) + } + + async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { + Ok(()) + } + + async fn close(&self) -> Result<(), TargetError> { + self.close_calls.fetch_add(1, Ordering::SeqCst); + self.close_started.notify_one(); + if self.block_on_close.load(Ordering::SeqCst) { + let _permit = self.close_gate.acquire().await.expect("close gate should remain open"); + } + if consume_failure_budget(&self.close_failures_remaining) { + return Err(match &self.close_failure_error { + Some(factory) => factory(), + None => TargetError::Storage("forced close failure".to_string()), + }); + } + Ok(()) + } + + fn store(&self) -> Option<&(dyn Store + Send + Sync)> { + self.store.as_deref() + } + + fn failed_store(&self) -> Option<&dyn FailedEventStore> { + self.failed_store.as_deref() + } + + fn clone_dyn(&self) -> Box + Send + Sync> { + Box::new(self.clone()) + } + + async fn init(&self) -> Result<(), TargetError> { + self.init_calls.fetch_add(1, Ordering::SeqCst); + if let Some(entered) = &self.blocking_init { + entered.notify_one(); + return std::future::pending().await; + } + if consume_failure_budget(&self.init_failures_remaining) { + return Err(TargetError::Initialization("forced init failure".to_string())); + } + Ok(()) + } + + fn is_enabled(&self) -> bool { + self.enabled_calls.fetch_add(1, Ordering::SeqCst); + self.enabled + } + + fn delivery_snapshot(&self) -> TargetDeliverySnapshot { + // The fixed snapshot wins when configured; otherwise mirror the trait's default impl, + // deriving the depths from the attached stores. + match &self.delivery_snapshot { + Some(snapshot) => snapshot.clone(), + None => TargetDeliverySnapshot { + failed_store_length: self + .failed_store + .as_deref() + .map_or(0, |failed_store| failed_store.failed_len() as u64), + queue_length: self.store.as_deref().map_or(0, |store| store.len() as u64), + ..TargetDeliverySnapshot::default() + }, + } + } + + fn record_final_failure(&self) { + self.final_failures.fetch_add(1, Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use super::MockTarget; + use crate::Target; + use crate::target::EntityTarget; + use rustfs_s3_types::EventName; + use std::sync::Arc; + + fn sample_event() -> Arc> { + Arc::new(EntityTarget { + object_name: "obj.txt".to_string(), + bucket_name: "bucket-a".to_string(), + event_name: EventName::ObjectCreatedPut, + data: "payload".to_string(), + }) + } + + // Leak-guard contract, part one: the mock is compiled only under cfg(test) or the + // off-by-default `test-support` feature (see lib.rs), so production builds cannot even name + // it. This test documents the default surface a consumer gets when it does opt in. + #[tokio::test] + async fn defaults_are_inert() { + let target = MockTarget::new("primary", "webhook"); + let handle: &dyn Target = ⌖ + + assert_eq!(handle.id().to_string(), "primary:webhook"); + assert!(handle.is_enabled()); + assert!(handle.is_active().await.expect("the default probe succeeds")); + handle.init().await.expect("the default init succeeds"); + handle.save(sample_event()).await.expect("the default save succeeds"); + handle.close().await.expect("the default close succeeds"); + assert!(handle.store().is_none()); + assert!(handle.failed_store().is_none()); + + assert_eq!(target.init_call_count(), 1); + assert_eq!(target.save_call_count(), 1); + assert_eq!(target.close_call_count(), 1); + assert_eq!(target.final_failure_count(), 0); + } + + #[tokio::test] + async fn clones_and_clone_dyn_share_the_same_counters() { + let target = MockTarget::new("primary", "webhook"); + let observer = target.clone(); + let boxed: Box + Send + Sync> = Box::new(target); + let second = boxed.clone_dyn(); + + boxed.close().await.expect("close succeeds"); + second.close().await.expect("close succeeds"); + second.record_final_failure(); + + assert_eq!(observer.close_call_count(), 2); + assert_eq!(observer.final_failure_count(), 1); + } + + #[tokio::test] + async fn init_failure_budget_fails_first_then_succeeds() { + let target = MockTarget::new("primary", "webhook").with_init_failures(2); + let handle: &dyn Target = ⌖ + + assert!(handle.init().await.is_err()); + assert!(handle.init().await.is_err()); + handle.init().await.expect("the failure budget is spent, so init succeeds"); + assert_eq!(target.init_call_count(), 3); + } + + #[tokio::test] + async fn save_failure_budget_fails_first_then_succeeds() { + let target = MockTarget::new("primary", "webhook").with_save_failures(1); + let handle: &dyn Target = ⌖ + + assert!(handle.save(sample_event()).await.is_err()); + handle + .save(sample_event()) + .await + .expect("the failure budget is spent, so save succeeds"); + assert_eq!(target.save_call_count(), 2); + } + + #[tokio::test] + async fn active_override_decouples_the_probe_from_enablement() { + let target = MockTarget::new("primary", "webhook").with_active(false); + let handle: &dyn Target = ⌖ + + assert!(handle.is_enabled()); + assert!(!handle.is_active().await.expect("the probe itself succeeds")); + assert_eq!(target.enabled_call_count(), 1, "every is_enabled call is counted"); + } + + #[tokio::test] + async fn with_id_renames_but_keeps_the_shared_state() { + let template = MockTarget::new("template", "webhook"); + let renamed = template.clone().with_id("instance", "webhook"); + assert_eq!(renamed.target_id().to_string(), "instance:webhook"); + + let handle: &dyn Target = &renamed; + handle.close().await.expect("close succeeds"); + assert_eq!(template.close_call_count(), 1, "a renamed clone still feeds the template's counters"); + } + + #[tokio::test] + async fn first_save_gate_blocks_only_the_first_save() { + let entered = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let target = MockTarget::new("primary", "webhook").with_first_save_gate(entered.clone(), release.clone()); + let observer = target.clone(); + + let gated: Arc + Send + Sync> = Arc::new(target); + let first = tokio::spawn({ + let gated = Arc::clone(&gated); + async move { gated.save(sample_event()).await } + }); + entered.notified().await; + assert_eq!(observer.save_call_count(), 1, "the gated save is counted before it parks"); + + gated + .save(sample_event()) + .await + .expect("a later save passes straight through"); + release.notify_one(); + first + .await + .expect("the gated save task should join") + .expect("the gated save succeeds after release"); + assert_eq!(observer.save_call_count(), 2); + } + + #[tokio::test] + async fn health_gate_holds_the_probe_until_released() { + let release = Arc::new(tokio::sync::Notify::new()); + let target = MockTarget::new("primary", "webhook").with_health_gate(release.clone()); + let started = target.health_started(); + + let probing: Arc + Send + Sync> = Arc::new(target); + let probe = tokio::spawn(async move { probing.is_active().await }); + started.notified().await; + assert!(!probe.is_finished(), "the probe must stay in flight until released"); + + release.notify_one(); + assert!( + probe + .await + .expect("the probe task should join") + .expect("the released probe succeeds"), + "the released probe reports the configured reachability" + ); + } + + #[tokio::test] + async fn close_failure_budget_uses_the_configured_error_shape() { + let target = MockTarget::new("primary", "webhook").with_close_failures(1); + let handle: &dyn Target = ⌖ + assert!( + matches!(handle.close().await, Err(crate::TargetError::Storage(_))), + "the default close failure is storage-flavored" + ); + handle.close().await.expect("the failure budget is spent, so close succeeds"); + assert_eq!(target.close_call_count(), 2); + + let pinned = MockTarget::new("primary", "webhook") + .with_close_failures(usize::MAX) + .with_close_failure_error(|| crate::TargetError::Unknown("close failed".to_string())); + let pinned_handle: &dyn Target = &pinned; + assert!(matches!(pinned_handle.close().await, Err(crate::TargetError::Unknown(_)))); + } + + #[tokio::test] + async fn delivery_snapshot_override_replaces_the_derived_snapshot() { + use crate::target::TargetDeliverySnapshot; + + let plain = MockTarget::new("primary", "webhook"); + let plain_handle: &dyn Target = &plain; + assert_eq!(plain_handle.delivery_snapshot(), TargetDeliverySnapshot::default()); + + let fixed = TargetDeliverySnapshot { + failed_messages: 1, + failed_store_length: 7, + queue_length: 0, + total_messages: 3, + }; + let target = MockTarget::new("primary", "webhook").with_delivery_snapshot(fixed.clone()); + let handle: &dyn Target = ⌖ + assert_eq!(handle.delivery_snapshot(), fixed); + } + + // Leak-guard contract, part two: the `test-support` feature must never ship by default and + // must stay a pure cfg gate, so no production dependency edge can drag the mock in. + #[test] + fn test_support_feature_never_ships_by_default() { + let manifest = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/Cargo.toml")) + .expect("the crate manifest should be readable"); + let default_features = manifest + .lines() + .map(str::trim) + .find(|line| line.starts_with("default = ")) + .expect("the crate manifest should declare a default feature list"); + assert_eq!(default_features, "default = []", "test-support must stay out of the default feature set"); + let feature = manifest + .lines() + .map(str::trim) + .find(|line| line.starts_with("test-support = ")) + .expect("the crate manifest should declare the test-support feature"); + assert_eq!( + feature, "test-support = []", + "test-support must stay a pure cfg gate that activates no dependencies" + ); + } +}