pamoja_loopback/
faulty.rs

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

pamoja_loopback/
faulty.rs
