gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 00:17:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(targets,notify,audit): share a builder MockTarget testkit (#6717)

Browse Source 
* test(targets): ship a builder MockTarget testkit and retire the in-crate Target mocks

Adds crates/targets/src/testkit.rs with a builder-style MockTarget implementing Target<E> for every E: PluginEvent, with orthogonal off-by-default knobs: disabled/active override, health delay plus health-started signal plus a drop-guard counter proving a cancelled probe future was dropped, an init failure budget (usize::MAX = always fail) plus blocking init plus an init counter, a close counter/signal/semaphore gate with a runtime block toggle, a save counter plus save failure budget, caller-supplied store and failed-store handles, and a shared final-failure counter. Clones and clone_dyn share all counters, so an observer clone keeps watching a target after it is boxed into a runtime.

The module is gated as #[cfg(any(test, feature = "test-support"))]: in-crate unit tests get it via cfg(test), and downstream test suites opt in through the new off-by-default test-support cargo feature (test-support = [], activating no dependencies).

Migrates the five in-crate duplicate mocks onto it: the plugin.rs registry-factory TestTarget, the runtime/adapter.rs lifecycle TestTarget (init/close/store knobs), the runtime/mod.rs TestTarget plus HealthDropGuard (close gating and health-probe tests), the target/mod.rs MoveTestTarget (folded into the test_support helper constructors used by the NATS JetStream failed-store tests), and the target/mod.rs StoreBackedTarget (the default send_from_store purge test; the mock deliberately does not override send_from_store or handle_terminal_failure). The forced init failure now uses TargetError::Initialization instead of the old adapter mock's Configuration; the adapter derives its redacted failure summary from the target id alone, so the migrated assertions are unchanged.

Leak guard: testkit unit tests assert the crate manifest still declares default = [] and that test-support = [] stays a pure cfg gate, complementing the compile-level cfg gate that keeps the mock out of production builds.

Part of rustfs/backlog#1846 (cluster 3, step 1).

* test(notify,audit): migrate the Target mocks onto the shared testkit

Retires the hand-written Target mocks in crates/notify and crates/audit in favor of rustfs_targets::testkit::MockTarget: notify's notifier.rs TestTarget/DeferredTestTarget/ClosableTestTarget, lifecycle.rs BlockingInitTarget/RetryInitTarget (rebuilt as observed MockTarget templates cloned by their plugin-descriptor factories, sharing the init signal, close counter, and single-failure init budget across generations), runtime_view.rs TestTarget, runtime_facade.rs TestTarget, and audit's pipeline.rs MockTarget, system.rs TestTarget, registry.rs CloseTestTarget, plus TestTarget and FailingTarget in audit/tests/pipeline_layer_test.rs. Removing the two integration-test mocks also removes their respelled PluginEvent bound (E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned), which the plugin-contract rules require to be spelled only via PluginEvent.

lifecycle.rs ReplayTarget stays bespoke on purpose: its generation tags, mpsc observation channels, gated send_raw delivery, and ObservedQueueStore model the replay pipeline itself and would contort a general-purpose mock. The other bespoke mocks named out of scope in PR-3a (ProgrammedTarget, ClassifyingTarget, the ReloadableTargetTls fakes) are likewise untouched.

New testkit knobs, each defaulted off and unit-tested: with_id (rename a clone while keeping the shared counters, for factory templates), with_first_save_gate (the first save notifies entered and waits on release; several mocks may share one pair), with_health_gate (is_active waits on a release handle after notifying health_started), with_delivery_snapshot (fixed snapshot overriding the store-derived default), with_close_failures (close-failure budget, default TargetError::Storage) with with_close_failure_error to shape the variant (audit's registry test pins TargetError::Unknown), and an always-on is_enabled call counter exposed as enabled_call_count (notifier's generation tests count dispatcher selections through it).

Both crates enable the testkit through a dev-dependency on rustfs-targets with the test-support feature; the feature stays out of default and activates no dependencies, so production builds are unchanged.

Part of rustfs/backlog#1846 (cluster 3, step 2).
This commit is contained in:
Zhengchao An
2026-08-27 11:28:16 +08:00
committed by GitHub
parent 0e27f57c40
commit 8ddbf05924
18 changed files with 902 additions and 1471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/audit/Cargo.toml
+1
View File
@@ -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 }
crates/audit/src/pipeline.rs
+15 -81
View File
@@ -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<Notify>, Arc<Notify>)>,
}
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<Notify>, release: Arc<Notify>) -> Self {
self.health_gate = Some((started, release));
self
}
}
#[async_trait]
impl<E> Target<E> for MockTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
if let Some((started, release)) = &self.health_gate {
started.notify_one();
release.notified().await;
}
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> 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<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<MockTarget>) -> 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
crates/audit/src/registry.rs
+14 -76
View File
@@ -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<AtomicUsize>,
fail_on_close: bool,
}
impl CloseTestTarget {
fn new(id: TargetID, close_calls: Arc<AtomicUsize>, fail_on_close: bool) -> Self {
Self {
id,
close_calls,
fail_on_close,
}
}
}
#[async_trait::async_trait]
impl Target<AuditEntry> for CloseTestTarget {
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<AuditEntry>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<AuditEntry> + 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());
}
}
crates/audit/src/system.rs
+11 -70
View File
@@ -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<AtomicUsize>,
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<E> Target<E> for TestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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);
crates/audit/tests/pipeline_layer_test.rs
+18 -139
View File
@@ -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<AtomicUsize>,
id: TargetID,
init_calls: Arc<AtomicUsize>,
}
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<E> Target<E> for TestTarget
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<AtomicUsize>,
}
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<E> Target<E> for FailingTarget
where
E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> 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<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<AuditEntry>)>) -> AuditPipeline {
@@ -154,8 +34,8 @@ fn pipeline_with_targets(targets: Vec<(&str, SharedTarget<AuditEntry>)>) -> 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<rustfs_audit::AuditEntry>],
crates/notify/Cargo.toml
+1
View File
@@ -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 }
crates/notify/src/lifecycle.rs
+23 -127
View File
@@ -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<BlockingInitState>,
}
#[async_trait]
impl Target<Event> for BlockingInitTarget {
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<Event>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<Event> + 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<AtomicBool>,
id: TargetID,
should_fail: bool,
}
#[async_trait]
impl Target<Event> for RetryInitTarget {
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<Event>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<Event> + 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<BlockingInitState>,
Arc<Notify>,
MockTarget,
NotifyLifecycleCoordinator,
NotifyRuntimeView,
Arc<AtomicUsize>,
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::<Event>::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<usize>, 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::<Event>::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)
}))
},
));
crates/notify/src/notifier.rs
+67 -304
View File
@@ -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<Notify>, Arc<Notify>)>,
close_calls: Arc<AtomicUsize>,
close_entered: Option<Arc<Notify>>,
id: TargetID,
enabled: bool,
save_calls: Arc<AtomicUsize>,
selected_calls: Arc<AtomicUsize>,
store: Option<QueueStore<QueuedPayload>>,
}
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<Notify>, release: Arc<Notify>) -> Self {
self.block_first_save = Some((entered, release));
self
}
fn with_store(mut self, store: QueueStore<QueuedPayload>) -> Self {
self.store = Some(store);
self
}
fn with_close_observer(mut self, close_entered: Arc<Notify>) -> Self {
self.close_entered = Some(close_entered);
self
}
}
#[async_trait]
impl<E> Target<E> for TestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(self.enabled)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> 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<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
self.store
.as_ref()
.map(|store| store as &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync))
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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::<usize>()
};
let total_calls = || targets.iter().map(|target| target.save_call_count()).sum::<usize>();
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<AtomicUsize>,
store: QueueStore<QueuedPayload>,
}
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<E> Target<E> for DeferredTestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
self.save_calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
Some(&self.store)
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<Event>])
.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<Event>])
.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<AtomicUsize>,
}
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<E> Target<E> for ClosableTestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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");
}
}
crates/notify/src/runtime_facade.rs
+18 -111
View File
@@ -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<Arc<Notify>>,
close_error: bool,
close_release: Option<Arc<Notify>>,
close_calls: Arc<AtomicUsize>,
id: TargetID,
store: Option<QueueStore<QueuedPayload>>,
}
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<Notify>, release: Arc<Notify>) -> 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<QueuedPayload>) -> Self {
self.store = Some(store);
self
}
}
#[async_trait]
impl<E> Target<E> for TestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
self.store
.as_ref()
.map(|store| store as &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync))
}
fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
Box::new(self.clone())
}
async fn init(&self) -> Result<(), TargetError> {
Ok(())
}
fn is_enabled(&self) -> bool {
true
}
}
fn build_facade() -> (NotifyRuntimeFacade, Arc<EventNotifier>, Arc<RwLock<ReplayWorkerManager>>) {
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::<QueuedPayload>::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<Event>],
@@ -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
crates/notify/src/runtime_view.rs
+22 -127
View File
@@ -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<AtomicU64>,
failed_store_length: u64,
id: TargetID,
health_started: Option<Arc<Notify>>,
health_release: Option<Arc<Notify>>,
total_messages: Arc<AtomicU64>,
}
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<Notify>, release: Arc<Notify>) -> 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<E> Target<E> for TestTarget
where
E: rustfs_targets::PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
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<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<dyn Target<Event> + Send + Sync>)
.add(Arc::new(target) as Arc<dyn Target<Event> + Send + Sync>)
.expect("test target should be added");
let runtime_view = NotifyRuntimeView::new(target_list.clone(), Arc::new(RwLock::new(ReplayWorkerManager::new())));
crates/targets/Cargo.toml
+3
View File
@@ -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",
crates/targets/src/lib.rs
+2
View File
@@ -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,
crates/targets/src/plugin.rs
+4 -58
View File
@@ -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<E> Target<E> for TestTarget
where
E: PluginEvent,
{
fn id(&self) -> crate::arn::TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
Box::new(self.clone())
}
fn is_enabled(&self) -> bool {
true
}
}
fn builtin_adapter() -> BuiltinPluginRuntimeAdapter<String> {
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))),
));
}
crates/targets/src/runtime/adapter.rs
+48 -127
View File
@@ -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<AtomicUsize>,
id: TargetID,
init_calls: Arc<AtomicUsize>,
init_entered: Option<Arc<Notify>>,
init_fails: bool,
store: Option<Arc<TestStore>>,
}
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<Notify>) -> 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::<QueuedPayload>::new(dir.path(), 16, ".queue");
store.open().expect("queue store should open");
self.store = Some(Arc::new(store));
self
}
}
#[async_trait]
impl<E> Target<E> for TestTarget
where
E: PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
self.store.as_deref()
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<TestStore> {
let dir = tempdir().expect("tempdir should be created for queue store tests");
let store = QueueStore::<QueuedPayload>::new(dir.path(), 16, ".queue");
store.open().expect("queue store should open");
Arc::new(store)
}
fn builtin_adapter() -> BuiltinPluginRuntimeAdapter<String> {
@@ -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::<QueuedPayload>::new(&queue_path, 16, ".queue")));
let target = MockTarget::new("primary", "webhook").with_store(Arc::new(QueueStore::<QueuedPayload>::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::<QueuedPayload>::new(&invalid_base, 16, ".queue")));
let target = MockTarget::new("primary", "webhook").with_store(Arc::new(QueueStore::<QueuedPayload>::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<Box<dyn Target<String> + 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);
}
}
crates/targets/src/runtime/mod.rs
+28 -132
View File
@@ -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<AtomicBool>,
close_gate: Arc<Semaphore>,
close_calls: Arc<AtomicUsize>,
enabled: bool,
health_delay: Duration,
health_drops: Arc<AtomicUsize>,
health_started: Arc<Notify>,
close_started: Arc<Notify>,
}
struct HealthDropGuard(Arc<AtomicUsize>);
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<E> Target<E> for TestTarget
where
E: PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
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<EntityTarget<E>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
None
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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::<String>::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::<String>::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::<String>::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::<String>::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::<String>::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::<String>::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<SharedTarget<String>> = (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<String>
})
.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<SharedTarget<String>> = 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]
crates/targets/src/target/mod.rs
+18 -106
View File
@@ -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<Arc<QueueStore<QueuedPayload>>>,
pub(crate) failed: Arc<AtomicU64>,
}
#[async_trait]
impl Target<String> for MoveTestTarget {
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<String>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + 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<dyn Target<String> + 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<dyn Target<String> + 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<QueueStore<QueuedPayload>>) -> Arc<dyn Target<String> + 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<QueuedPayload>,
}
#[async_trait]
impl Target<String> for StoreBackedTarget {
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
Ok(true)
}
async fn save(&self, _event: Arc<EntityTarget<String>>) -> Result<(), TargetError> {
Ok(())
}
async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
Ok(())
}
async fn close(&self) -> Result<(), TargetError> {
Ok(())
}
fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
Some(&self.store)
}
fn clone_dyn(&self) -> Box<dyn Target<String> + 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<dyn Target<String> + 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.
crates/targets/src/target/nats/jetstream.rs
+9 -13
View File
@@ -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::<QueuedPayload>::new_with_compression(&dir, 8, ".test", false));
store.open().unwrap();
let failed = Arc::new(AtomicU64::new(0));
let target: Arc<dyn Target<String> + 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<dyn Target<String> + 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();
}
crates/targets/src/testkit.rs
+600
View File
@@ -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<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + 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<dyn Fn() -> 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<AtomicUsize>);
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<bool>,
health_delay: Duration,
health_started: Arc<Notify>,
/// 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<Arc<Notify>>,
health_drops: Arc<AtomicUsize>,
enabled_calls: Arc<AtomicUsize>,
init_calls: Arc<AtomicUsize>,
init_failures_remaining: Arc<AtomicUsize>,
/// When set, `init` notifies the handle on entry and then never returns.
blocking_init: Option<Arc<Notify>>,
close_calls: Arc<AtomicUsize>,
close_started: Arc<Notify>,
block_on_close: Arc<AtomicBool>,
close_gate: Arc<Semaphore>,
close_failures_remaining: Arc<AtomicUsize>,
close_failure_error: Option<ErrorFactory>,
save_calls: Arc<AtomicUsize>,
save_failures_remaining: Arc<AtomicUsize>,
/// 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<Notify>, Arc<Notify>)>,
final_failures: Arc<AtomicU64>,
delivery_snapshot: Option<TargetDeliverySnapshot>,
store: Option<SharedQueuedStore>,
failed_store: Option<Arc<dyn FailedEventStore>>,
}
impl MockTarget {
/// Creates an enabled, reachable, storeless mock identified as `<id>:<name>`.
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<Notify>) -> 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<Notify>) -> 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<Notify>, release: Arc<Notify>) -> 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<dyn FailedEventStore>) -> 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<Semaphore> {
Arc::clone(&self.close_gate)
}
/// Notified once every time a `close` call starts.
pub fn close_started(&self) -> Arc<Notify> {
Arc::clone(&self.close_started)
}
/// Notified once every time an `is_active` probe starts.
pub fn health_started(&self) -> Arc<Notify> {
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<E> Target<E> for MockTarget
where
E: PluginEvent,
{
fn id(&self) -> TargetID {
self.id.clone()
}
async fn is_active(&self) -> Result<bool, TargetError> {
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<EntityTarget<E>>) -> 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<u8>, _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<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
self.store.as_deref()
}
fn failed_store(&self) -> Option<&dyn FailedEventStore> {
self.failed_store.as_deref()
}
fn clone_dyn(&self) -> Box<dyn Target<E> + 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<EntityTarget<String>> {
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<String> = &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<dyn Target<String> + 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<String> = &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<String> = &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<String> = &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<String> = &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<dyn Target<String> + 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<dyn Target<String> + 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<String> = &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<String> = &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<String> = &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<String> = &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"
);
}
}