pamoja_loopback/
faulty.rs

1//! A transport decorator that injects send failures for degraded-link testing.
2
3use pamoja_core::{Error, Message, Receive, Result, Transport};
4
5/// Wraps a [`Transport`] and fails a configurable number of upcoming sends.
6///
7/// This simulates an intermittent link so offline-first behavior can be proven
8/// rather than assumed: pair it with a store-and-forward drain and assert that
9/// every record still arrives, in order, once the link recovers. Connecting,
10/// subscribing, and receiving pass straight through to the wrapped transport;
11/// only sends are failed.
12///
13/// # Examples
14///
15/// ```
16/// use pamoja_core::Transport;
17/// use pamoja_loopback::{Faulty, LoopbackBroker, LoopbackTransport};
18///
19/// # async fn run() -> pamoja_core::Result<()> {
20/// let broker = LoopbackBroker::new();
21/// let mut node = Faulty::new(LoopbackTransport::new(broker), 1);
22/// node.connect().await?;
23///
24/// // The first send fails, simulating a dropped link; the next succeeds.
25/// assert!(node.send("t", b"x").await.is_err());
26/// node.send("t", b"x").await?;
27/// # Ok(())
28/// # }
29/// ```
30pub struct Faulty<T> {
31    inner: T,
32    upcoming_failures: usize,
33}
34
35impl<T> Faulty<T> {
36    /// Wraps `inner`, failing its next `failures` sends before passing through.
37    ///
38    /// # Arguments
39    ///
40    /// * `inner` - the transport to decorate.
41    /// * `failures` - how many of the next [`send`](Transport::send) calls fail.
42    ///
43    /// # Returns
44    ///
45    /// A decorator that injects the requested failures, then delegates to `inner`.
46    pub fn new(inner: T, failures: usize) -> Self {
47        Self {
48            inner,
49            upcoming_failures: failures,
50        }
51    }
52
53    /// Arms the decorator to fail the next `count` sends.
54    ///
55    /// # Arguments
56    ///
57    /// * `count` - the number of upcoming sends to fail, simulating another link
58    ///   outage.
59    pub fn fail_next(&mut self, count: usize) {
60        self.upcoming_failures = count;
61    }
62}
63
64impl<T: Transport + Send> Transport for Faulty<T> {
65    async fn connect(&mut self) -> Result<()> {
66        self.inner.connect().await
67    }
68
69    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
70        if self.upcoming_failures > 0 {
71            self.upcoming_failures -= 1;
72            return Err(Error::Transport("simulated link failure".to_owned()));
73        }
74        self.inner.send(topic, payload).await
75    }
76
77    async fn subscribe(&mut self, topic: &str) -> Result<()> {
78        self.inner.subscribe(topic).await
79    }
80}
81
82impl<T: Receive + Send> Receive for Faulty<T> {
83    async fn recv(&mut self) -> Result<Option<Message>> {
84        self.inner.recv().await
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use crate::{LoopbackBroker, LoopbackTransport};
92
93    #[tokio::test]
94    async fn fails_the_configured_sends_then_passes_through() {
95        let broker = LoopbackBroker::new();
96        let mut gateway = LoopbackTransport::new(broker.clone());
97        gateway.connect().await.expect("connect");
98        gateway.subscribe("#").await.expect("subscribe");
99
100        let mut node = Faulty::new(LoopbackTransport::new(broker), 2);
101        node.connect().await.expect("connect");
102
103        assert!(node.send("t", b"1").await.is_err());
104        assert!(node.send("t", b"2").await.is_err());
105        node.send("t", b"3")
106            .await
107            .expect("third send passes through");
108
109        // Only the delivered payload reaches the gateway.
110        let message = gateway.recv().await.expect("recv").expect("a message");
111        assert_eq!(message.payload, b"3");
112    }
113}
pamoja_loopback/faulty.rs

pamoja_loopback/
faulty.rs
