rustfs_audit/
pipeline.rs

1//  Copyright 2024 RustFS Team
2//
3//  Licensed under the Apache License, Version 2.0 (the "License");
4//  you may not use this file except in compliance with the License.
5//  You may obtain a copy of the License at
6//
7//      http://www.apache.org/licenses/LICENSE-2.0
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14
15use crate::{AuditEntry, AuditResult, observability, system::AuditTargetMetricSnapshot};
16use rustfs_targets::{
17    BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter, ReplayEvent, ReplayWorkerManager, RuntimeActivation, SharedTarget, Target,
18    target::EntityTarget,
19};
20use std::sync::Arc;
21use std::time::Duration;
22use tokio::sync::{Mutex, RwLock};
23use tracing::{debug, error, info, warn};
24
25const LOG_COMPONENT_AUDIT: &str = "audit";
26const LOG_SUBSYSTEM_PIPELINE: &str = "pipeline";
27const EVENT_AUDIT_DISPATCH_SKIPPED: &str = "audit_dispatch_skipped";
28const EVENT_AUDIT_DISPATCH_FAILED: &str = "audit_dispatch_failed";
29const EVENT_AUDIT_BATCH_DISPATCH_SKIPPED: &str = "audit_batch_dispatch_skipped";
30const EVENT_AUDIT_BATCH_DISPATCH_FAILED: &str = "audit_batch_dispatch_failed";
31const EVENT_AUDIT_BATCH_DISPATCH_COMPLETED: &str = "audit_batch_dispatch_completed";
32const EVENT_AUDIT_TARGET_STATE_CHANGED: &str = "audit_target_state_changed";
33const EVENT_AUDIT_REPLAY_DELIVERED: &str = "audit_replay_delivered";
34const EVENT_AUDIT_REPLAY_RETRY_SCHEDULED: &str = "audit_replay_retry_scheduled";
35const EVENT_AUDIT_REPLAY_RETRY_EXHAUSTED: &str = "audit_replay_retry_exhausted";
36const EVENT_AUDIT_REPLAY_DROPPED: &str = "audit_replay_dropped";
37const EVENT_AUDIT_REPLAY_STREAM_STATUS: &str = "audit_replay_stream_status";
38
39#[derive(Clone)]
40pub struct AuditPipeline {
41    registry: Arc<Mutex<crate::AuditRegistry>>,
42}
43
44impl AuditPipeline {
45    pub fn new(registry: Arc<Mutex<crate::AuditRegistry>>) -> Self {
46        Self { registry }
47    }
48
49    /// Fans an audit entry out to every configured target concurrently.
50    ///
51    /// Delivery across targets is unordered: the per-target `save()` calls run
52    /// via `join_all` and may complete in any order. Ordering of entries within
53    /// a single target is preserved by that target's own store/queue, not by
54    /// this fan-out.
55    pub async fn dispatch(&self, entry: Arc<AuditEntry>) -> AuditResult<()> {
56        let start_time = std::time::Instant::now();
57
58        let targets: Vec<SharedTarget<AuditEntry>> = {
59            let registry = self.registry.lock().await;
60            let targets = registry.list_target_values();
61
62            if targets.is_empty() {
63                debug!(
64                    event = EVENT_AUDIT_DISPATCH_SKIPPED,
65                    component = LOG_COMPONENT_AUDIT,
66                    subsystem = LOG_SUBSYSTEM_PIPELINE,
67                    reason = "no_targets_configured",
68                    "Skipped audit dispatch"
69                );
70                return Ok(());
71            }
72
73            targets
74        };
75
76        let mut tasks = Vec::new();
77
78        for target in targets {
79            let entity_target = EntityTarget {
80                object_name: entry.api.name.clone().unwrap_or_default(),
81                bucket_name: entry.api.bucket.clone().unwrap_or_default(),
82                event_name: entry.event,
83                data: (*entry).clone(),
84            };
85
86            let task = async move {
87                let result = target.save(Arc::new(entity_target)).await;
88                (target.id().to_string(), result)
89            };
90
91            tasks.push(task);
92        }
93
94        let results = futures::future::join_all(tasks).await;
95
96        let mut errors = Vec::new();
97        let mut success_count = 0;
98
99        for (target_key, result) in results {
100            match result {
101                Ok(_) => {
102                    success_count += 1;
103                    observability::record_target_success();
104                }
105                Err(e) => {
106                    error!(
107                        event = EVENT_AUDIT_DISPATCH_FAILED,
108                        component = LOG_COMPONENT_AUDIT,
109                        subsystem = LOG_SUBSYSTEM_PIPELINE,
110                        target_id = %target_key,
111                        error = %e,
112                        "Failed to dispatch audit event"
113                    );
114                    errors.push(e);
115                    observability::record_target_failure();
116                }
117            }
118        }
119
120        let dispatch_time = start_time.elapsed();
121
122        if errors.is_empty() {
123            observability::record_audit_success(dispatch_time);
124            return Ok(());
125        }
126
127        observability::record_audit_failure(dispatch_time);
128        let error_count = errors.len();
129
130        if success_count == 0 {
131            // Every configured target rejected the event. For store-backed targets a
132            // failed save() means the entry was neither delivered nor persisted for
133            // replay, so it is lost outright. Propagate the failure instead of
134            // returning Ok so the caller can react (alert, degrade, or reject the
135            // request) rather than assume the audit trail is intact.
136            error!(
137                event = EVENT_AUDIT_DISPATCH_FAILED,
138                component = LOG_COMPONENT_AUDIT,
139                subsystem = LOG_SUBSYSTEM_PIPELINE,
140                error_count = error_count,
141                duration_ms = dispatch_time.as_millis() as u64,
142                "All audit targets failed to receive audit event"
143            );
144            // `errors` is non-empty here, so `remove(0)` cannot panic.
145            return Err(crate::AuditError::Target(errors.remove(0)));
146        }
147
148        // Partial failure: at least one target accepted the event, so the entry is
149        // not lost. Surface the degradation but let the dispatch succeed.
150        warn!(
151            event = EVENT_AUDIT_DISPATCH_FAILED,
152            component = LOG_COMPONENT_AUDIT,
153            subsystem = LOG_SUBSYSTEM_PIPELINE,
154            error_count = error_count,
155            success_count = success_count,
156            duration_ms = dispatch_time.as_millis() as u64,
157            "Some audit targets failed to receive audit event"
158        );
159
160        Ok(())
161    }
162
163    pub async fn dispatch_batch(&self, entries: Vec<Arc<AuditEntry>>) -> AuditResult<()> {
164        let start_time = std::time::Instant::now();
165
166        let targets: Vec<SharedTarget<AuditEntry>> = {
167            let registry = self.registry.lock().await;
168            let targets = registry.list_target_values();
169
170            if targets.is_empty() {
171                debug!(
172                    event = EVENT_AUDIT_BATCH_DISPATCH_SKIPPED,
173                    component = LOG_COMPONENT_AUDIT,
174                    subsystem = LOG_SUBSYSTEM_PIPELINE,
175                    entry_count = entries.len(),
176                    reason = "no_targets_configured",
177                    "Skipped audit batch dispatch"
178                );
179                return Ok(());
180            }
181
182            targets
183        };
184
185        let mut tasks = Vec::new();
186        for target in targets {
187            let entries_clone: Vec<_> = entries.iter().map(Arc::clone).collect();
188
189            let task = async move {
190                let mut success_count = 0;
191                let mut errors = Vec::new();
192                for entry in entries_clone {
193                    let entity_target = EntityTarget {
194                        object_name: entry.api.name.clone().unwrap_or_default(),
195                        bucket_name: entry.api.bucket.clone().unwrap_or_default(),
196                        event_name: entry.event,
197                        data: (*entry).clone(),
198                    };
199                    match target.save(Arc::new(entity_target)).await {
200                        Ok(_) => {
201                            success_count += 1;
202                            observability::record_target_success();
203                        }
204                        Err(e) => {
205                            observability::record_target_failure();
206                            errors.push(e);
207                        }
208                    }
209                }
210                (target.id().to_string(), success_count, errors)
211            };
212            tasks.push(task);
213        }
214
215        let results = futures::future::join_all(tasks).await;
216        let mut total_success = 0;
217        let mut total_errors = 0;
218        let mut first_error: Option<rustfs_targets::TargetError> = None;
219        for (target_id, success_count, errors) in results {
220            total_success += success_count;
221            total_errors += errors.len();
222            for e in errors {
223                error!(
224                    event = EVENT_AUDIT_BATCH_DISPATCH_FAILED,
225                    component = LOG_COMPONENT_AUDIT,
226                    subsystem = LOG_SUBSYSTEM_PIPELINE,
227                    target_id = %target_id,
228                    error = ?e,
229                    "Audit batch dispatch failed"
230                );
231                if first_error.is_none() {
232                    first_error = Some(e);
233                }
234            }
235        }
236
237        let dispatch_time = start_time.elapsed();
238        debug!(
239            event = EVENT_AUDIT_BATCH_DISPATCH_COMPLETED,
240            component = LOG_COMPONENT_AUDIT,
241            subsystem = LOG_SUBSYSTEM_PIPELINE,
242            entry_count = entries.len(),
243            success_count = total_success,
244            error_count = total_errors,
245            duration_ms = dispatch_time.as_millis() as u64,
246            "Completed audit batch dispatch"
247        );
248
249        // No save() across any target/entry succeeded while errors were recorded:
250        // the batch was lost entirely. Propagate rather than silently returning Ok.
251        if total_errors > 0 && total_success == 0 {
252            observability::record_audit_failure(dispatch_time);
253            error!(
254                event = EVENT_AUDIT_BATCH_DISPATCH_FAILED,
255                component = LOG_COMPONENT_AUDIT,
256                subsystem = LOG_SUBSYSTEM_PIPELINE,
257                entry_count = entries.len(),
258                error_count = total_errors,
259                duration_ms = dispatch_time.as_millis() as u64,
260                "All audit targets failed to receive audit batch"
261            );
262            return Err(crate::AuditError::Target(
263                first_error.expect("total_errors > 0 guarantees a captured target error"),
264            ));
265        }
266
267        // Record the aggregate event outcome so batch dispatch reports the same
268        // observability signal as single dispatch (backlog#984): full success or
269        // partial failure both count as a delivered audit event here, since at
270        // least one target accepted every entry that reached this point.
271        observability::record_audit_success(dispatch_time);
272
273        Ok(())
274    }
275
276    pub async fn snapshot_target_metrics(&self) -> Vec<AuditTargetMetricSnapshot> {
277        let registry = self.registry.lock().await;
278        registry
279            .list_target_values()
280            .into_iter()
281            .map(|target| {
282                let delivery = target.delivery_snapshot();
283                AuditTargetMetricSnapshot {
284                    failed_messages: delivery.failed_messages,
285                    failed_store_length: delivery.failed_store_length,
286                    queue_length: delivery.queue_length,
287                    target_id: target.id().to_string(),
288                    total_messages: delivery.total_messages,
289                }
290            })
291            .collect()
292    }
293
294    pub async fn snapshot_target_health(&self) -> Vec<rustfs_targets::RuntimeTargetHealthSnapshot> {
295        let targets = self.registry.lock().await.list_target_values();
296        rustfs_targets::health_snapshots_for_targets(targets).await
297    }
298}
299
300#[derive(Clone)]
301pub struct AuditRuntimeView {
302    registry: Arc<Mutex<crate::AuditRegistry>>,
303}
304
305impl AuditRuntimeView {
306    pub fn new(registry: Arc<Mutex<crate::AuditRegistry>>) -> Self {
307        Self { registry }
308    }
309
310    pub async fn list_targets(&self) -> Vec<String> {
311        let registry = self.registry.lock().await;
312        registry.list_targets()
313    }
314
315    pub async fn get_target_values(&self) -> Vec<SharedTarget<AuditEntry>> {
316        let registry = self.registry.lock().await;
317        registry.list_target_values()
318    }
319
320    pub async fn get_target(&self, target_id: &str) -> Option<String> {
321        let registry = self.registry.lock().await;
322        registry.get_target(target_id).map(|target| target.id().to_string())
323    }
324
325    pub async fn enable_target(&self, target_id: &str) -> AuditResult<()> {
326        let registry = self.registry.lock().await;
327        if registry.get_target(target_id).is_some() {
328            info!(
329                event = EVENT_AUDIT_TARGET_STATE_CHANGED,
330                component = LOG_COMPONENT_AUDIT,
331                subsystem = LOG_SUBSYSTEM_PIPELINE,
332                target_id = %target_id,
333                state = "enabled",
334                "audit target state"
335            );
336            Ok(())
337        } else {
338            Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None))
339        }
340    }
341
342    pub async fn disable_target(&self, target_id: &str) -> AuditResult<()> {
343        let registry = self.registry.lock().await;
344        if registry.get_target(target_id).is_some() {
345            info!(
346                event = EVENT_AUDIT_TARGET_STATE_CHANGED,
347                component = LOG_COMPONENT_AUDIT,
348                subsystem = LOG_SUBSYSTEM_PIPELINE,
349                target_id = %target_id,
350                state = "disabled",
351                "audit target state"
352            );
353            Ok(())
354        } else {
355            Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None))
356        }
357    }
358
359    pub async fn remove_target(&self, target_id: &str) -> AuditResult<()> {
360        let mut registry = self.registry.lock().await;
361        if registry.remove_target(target_id).await.is_some() {
362            info!(
363                event = EVENT_AUDIT_TARGET_STATE_CHANGED,
364                component = LOG_COMPONENT_AUDIT,
365                subsystem = LOG_SUBSYSTEM_PIPELINE,
366                target_id = %target_id,
367                state = "removed",
368                "audit target state"
369            );
370            Ok(())
371        } else {
372            Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None))
373        }
374    }
375
376    pub async fn upsert_target(&self, target_id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) -> AuditResult<()> {
377        if let Err(err) = target.init().await {
378            return Err(crate::AuditError::Target(err));
379        }
380
381        let shared_target: SharedTarget<AuditEntry> = Arc::from(target);
382        let mut registry = self.registry.lock().await;
383        let _ = registry.remove_target(&target_id).await;
384        registry.add_shared_target(target_id.clone(), shared_target);
385        info!(
386            event = EVENT_AUDIT_TARGET_STATE_CHANGED,
387            component = LOG_COMPONENT_AUDIT,
388            subsystem = LOG_SUBSYSTEM_PIPELINE,
389            target_id = %target_id,
390            state = "upserted",
391            "audit target state"
392        );
393        Ok(())
394    }
395}
396
397#[derive(Clone)]
398pub struct AuditRuntimeFacade {
399    registry: Arc<Mutex<crate::AuditRegistry>>,
400    replay_workers: Arc<RwLock<ReplayWorkerManager>>,
401    runtime_adapter: Arc<dyn PluginRuntimeAdapter<AuditEntry>>,
402}
403
404impl AuditRuntimeFacade {
405    pub fn new(registry: Arc<Mutex<crate::AuditRegistry>>, replay_workers: Arc<RwLock<ReplayWorkerManager>>) -> Self {
406        let runtime_adapter = BuiltinPluginRuntimeAdapter::new(
407            Arc::new(move |event: ReplayEvent<AuditEntry>| {
408                Box::pin(async move {
409                    match event {
410                        ReplayEvent::Delivered { key, target } => {
411                            debug!(
412                                event = EVENT_AUDIT_REPLAY_DELIVERED,
413                                component = LOG_COMPONENT_AUDIT,
414                                subsystem = LOG_SUBSYSTEM_PIPELINE,
415                                target_id = %target.id(),
416                                replay_key = %key,
417                                "audit replay delivery"
418                            );
419                            observability::record_target_success();
420                        }
421                        ReplayEvent::RetryableError { error, target, .. } => match error {
422                            rustfs_targets::TargetError::NotConnected => {
423                                debug!(
424                                    event = EVENT_AUDIT_REPLAY_RETRY_SCHEDULED,
425                                    component = LOG_COMPONENT_AUDIT,
426                                    subsystem = LOG_SUBSYSTEM_PIPELINE,
427                                    target_id = %target.id(),
428                                    reason = "not_connected",
429                                    "audit replay delivery"
430                                );
431                            }
432                            rustfs_targets::TargetError::Timeout(_) => {
433                                debug!(
434                                    event = EVENT_AUDIT_REPLAY_RETRY_SCHEDULED,
435                                    component = LOG_COMPONENT_AUDIT,
436                                    subsystem = LOG_SUBSYSTEM_PIPELINE,
437                                    target_id = %target.id(),
438                                    reason = "timeout",
439                                    "audit replay delivery"
440                                );
441                            }
442                            _ => {}
443                        },
444                        ReplayEvent::Dropped { reason, target, .. } => {
445                            warn!(
446                                event = EVENT_AUDIT_REPLAY_DROPPED,
447                                component = LOG_COMPONENT_AUDIT,
448                                subsystem = LOG_SUBSYSTEM_PIPELINE,
449                                target_id = %target.id(),
450                                reason = %reason,
451                                "audit replay delivery"
452                            );
453                            observability::record_target_failure();
454                        }
455                        ReplayEvent::PermanentFailure { error, target, .. } => {
456                            error!(
457                                event = EVENT_AUDIT_REPLAY_DROPPED,
458                                component = LOG_COMPONENT_AUDIT,
459                                subsystem = LOG_SUBSYSTEM_PIPELINE,
460                                target_id = %target.id(),
461                                error = %error,
462                                reason = "permanent_failure",
463                                "audit replay delivery"
464                            );
465                            target.record_final_failure();
466                            observability::record_target_failure();
467                        }
468                        ReplayEvent::RetryExhausted { detail, key, target } => {
469                            warn!(
470                                event = EVENT_AUDIT_REPLAY_RETRY_EXHAUSTED,
471                                component = LOG_COMPONENT_AUDIT,
472                                subsystem = LOG_SUBSYSTEM_PIPELINE,
473                                target_id = %target.id(),
474                                replay_key = %key,
475                                error = %detail,
476                                "audit replay retry budget exhausted, entry stays queued and retries"
477                            );
478                        }
479                        ReplayEvent::UnreadableEntry { key, error, target } => {
480                            warn!(
481                                event = EVENT_AUDIT_REPLAY_DROPPED,
482                                component = LOG_COMPONENT_AUDIT,
483                                subsystem = LOG_SUBSYSTEM_PIPELINE,
484                                target_id = %target.id(),
485                                replay_key = %key,
486                                error = %error,
487                                reason = "unreadable_entry",
488                                "audit replay delivery"
489                            );
490                        }
491                    }
492                })
493            }),
494            Arc::new(|target_id, has_replay| {
495                if has_replay {
496                    info!(
497                        event = EVENT_AUDIT_REPLAY_STREAM_STATUS,
498                        component = LOG_COMPONENT_AUDIT,
499                        subsystem = LOG_SUBSYSTEM_PIPELINE,
500                        target_id = %target_id,
501                        replay_enabled = true,
502                        "audit replay stream"
503                    );
504                } else {
505                    debug!(
506                        event = EVENT_AUDIT_REPLAY_STREAM_STATUS,
507                        component = LOG_COMPONENT_AUDIT,
508                        subsystem = LOG_SUBSYSTEM_PIPELINE,
509                        target_id = %target_id,
510                        replay_enabled = false,
511                        reason = "no_store_configured",
512                        "audit replay stream"
513                    );
514                }
515            }),
516            None,
517            Duration::from_millis(500),
518            Duration::from_millis(500),
519            "Stopping audit stream",
520        );
521
522        Self {
523            registry,
524            replay_workers,
525            runtime_adapter: Arc::new(runtime_adapter),
526        }
527    }
528
529    pub async fn replace_targets(&self, activation: RuntimeActivation<AuditEntry>) -> AuditResult<()> {
530        let mut registry = self.registry.lock().await;
531        let mut replay_workers = self.replay_workers.write().await;
532        self.runtime_adapter
533            .replace_runtime_targets(registry.runtime_manager_mut(), &mut replay_workers, activation)
534            .await
535            .map_err(crate::AuditError::Target)?;
536        Ok(())
537    }
538
539    pub async fn shutdown_runtime(
540        &self,
541        registry: &mut crate::AuditRegistry,
542        replay_workers: &mut ReplayWorkerManager,
543    ) -> AuditResult<()> {
544        self.runtime_adapter
545            .shutdown(registry.runtime_manager_mut(), replay_workers)
546            .await
547            .map_err(crate::AuditError::Target)
548    }
549
550    pub async fn activate_targets_with_replay(
551        &self,
552        targets: Vec<Box<dyn Target<AuditEntry> + Send + Sync>>,
553    ) -> RuntimeActivation<AuditEntry> {
554        self.runtime_adapter.activate_with_replay(targets).await
555    }
556
557    pub async fn stop_replay_workers(&self) {
558        let mut replay_workers = self.replay_workers.write().await;
559        self.runtime_adapter.stop_replay_workers(&mut replay_workers).await;
560    }
561}
562
563#[cfg(test)]
564mod tests {
565    use super::AuditPipeline;
566    use crate::{AuditEntry, AuditError, AuditRegistry};
567    use rustfs_targets::testkit::MockTarget;
568    use std::sync::Arc;
569    use tokio::sync::{Mutex, Notify};
570
571    /// Builds a mock target whose `save()` outcome is fixed at construction so tests can force
572    /// full-success / full-failure / partial-failure fan-outs.
573    fn mock_target(id: &str, fail: bool) -> MockTarget {
574        let target = MockTarget::new(id, "webhook");
575        if fail { target.with_save_failures(usize::MAX) } else { target }
576    }
577
578    fn pipeline_with(targets: Vec<MockTarget>) -> AuditPipeline {
579        let mut registry = AuditRegistry::new();
580        for target in targets {
581            registry.add_target(target.target_id().to_string(), Box::new(target));
582        }
583        AuditPipeline::new(Arc::new(Mutex::new(registry)))
584    }
585
586    fn entry() -> Arc<AuditEntry> {
587        Arc::new(AuditEntry::default())
588    }
589
590    // backlog#962: when every target rejects the event it is lost outright, so
591    // dispatch must return Err rather than swallowing the failures as Ok.
592    #[tokio::test]
593    async fn dispatch_returns_err_when_all_targets_fail() {
594        let pipeline = pipeline_with(vec![mock_target("a:webhook", true), mock_target("b:webhook", true)]);
595        let result = pipeline.dispatch(entry()).await;
596        assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}");
597    }
598
599    // A partially-successful fan-out means the entry reached at least one sink,
600    // so dispatch reports success (degradation is logged, not propagated).
601    #[tokio::test]
602    async fn dispatch_returns_ok_on_partial_failure() {
603        let pipeline = pipeline_with(vec![mock_target("ok:webhook", false), mock_target("bad:webhook", true)]);
604        pipeline.dispatch(entry()).await.expect("partial success should return Ok");
605    }
606
607    #[tokio::test]
608    async fn dispatch_returns_ok_when_all_targets_succeed() {
609        let pipeline = pipeline_with(vec![mock_target("a:webhook", false), mock_target("b:webhook", false)]);
610        pipeline.dispatch(entry()).await.expect("all-success should return Ok");
611    }
612
613    // No configured targets is a benign no-op, not a failure.
614    #[tokio::test]
615    async fn dispatch_returns_ok_with_no_targets() {
616        let pipeline = pipeline_with(vec![]);
617        pipeline.dispatch(entry()).await.expect("no targets should return Ok");
618    }
619
620    #[tokio::test]
621    async fn health_probe_does_not_hold_the_registry_lock() {
622        let release = Arc::new(Notify::new());
623        let target = mock_target("blocked", false).with_health_gate(release.clone());
624        let started = target.health_started();
625        let pipeline = pipeline_with(vec![target]);
626        let registry = Arc::clone(&pipeline.registry);
627        let snapshot_task = tokio::spawn(async move { pipeline.snapshot_target_health().await });
628        started.notified().await;
629
630        let guard = tokio::time::timeout(std::time::Duration::from_secs(1), registry.lock())
631            .await
632            .expect("network health probe must not retain the audit registry lock");
633        drop(guard);
634        release.notify_one();
635
636        assert_eq!(snapshot_task.await.expect("snapshot task should finish").len(), 1);
637    }
638
639    // backlog#962: dispatch_batch must mirror dispatch and propagate a
640    // whole-batch loss instead of returning Ok.
641    #[tokio::test]
642    async fn dispatch_batch_returns_err_when_all_targets_fail() {
643        let pipeline = pipeline_with(vec![mock_target("a:webhook", true)]);
644        let result = pipeline.dispatch_batch(vec![entry(), entry()]).await;
645        assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}");
646    }
647
648    #[tokio::test]
649    async fn dispatch_batch_returns_ok_when_all_targets_succeed() {
650        let pipeline = pipeline_with(vec![mock_target("a:webhook", false), mock_target("b:webhook", false)]);
651        pipeline
652            .dispatch_batch(vec![entry(), entry()])
653            .await
654            .expect("all-success batch should return Ok");
655    }
656}
rustfs_audit/pipeline.rs

rustfs_audit/
pipeline.rs
