Let network faults affect in-flight messages

When injecting network faults (partition, hold) into a link, also update
the state of messages already in the `sent` queue, instead of having the
fault only affect messages that are enqueued in the future.

This changes enables writing tests in which one observes the contents of
a link to find the point in time in which a given message is sent, to
then prevent that message from getting delivered.

Author: teskje

src/top.rs

@@ -449,6 +449,8 @@ impl Link {
                 if do_rand {
                     self.state_a_b = State::RandPartition;
                     self.state_b_a = State::RandPartition;
+
+                    self.sent.clear();
                 }
             }
             (State::RandPartition, _) | (_, State::RandPartition) => {
@@ -463,6 +465,10 @@ impl Link {
     fn hold(&mut self) {
         self.state_a_b = State::Hold;
         self.state_b_a = State::Hold;
+
+        for sent in &mut self.sent {
+            sent.status = DeliveryStatus::Hold;
+        }
     }
 
     // This link becomes healthy, and any held messages are scheduled for delivery.
@@ -479,6 +485,8 @@ impl Link {
     fn explicit_partition(&mut self) {
         self.state_a_b = State::ExplicitPartition;
         self.state_b_a = State::ExplicitPartition;
+
+        self.sent.clear();
     }
 
     fn partition_oneway(&mut self, from: IpAddr, to: IpAddr) {
@@ -487,6 +495,8 @@ impl Link {
         } else {
             self.state_b_a = State::ExplicitPartition;
         }
+
+        self.sent.retain(|sent| sent.src.ip() != from);
     }
 
     fn repair_oneway(&mut self, from: IpAddr, to: IpAddr) {

tests/tcp.rs

@@ -1393,3 +1393,80 @@ fn try_write() -> Result {
 
     sim.run()
 }
+
+#[test]
+fn network_partition_drops_inflight_messages() {
+    let mut sim = Builder::new()
+        .min_message_latency(Duration::from_millis(10))
+        .build();
+
+    sim.host("server", || async {
+        let listener = bind().await?;
+        let (mut sock, _) = listener.accept().await?;
+        sock.write_u8(1).await?;
+        turmoil::partition("server", "client");
+        Ok(())
+    });
+
+    sim.client("client", async {
+        let mut sock = TcpStream::connect(("server", PORT)).await?;
+        sock.read_u8().await?;
+        Ok(())
+    });
+
+    let result = sim.run();
+    assert!(result.is_err());
+}
+
+#[test]
+fn network_partition_oneway_drops_inflight_messages() {
+    let mut sim = Builder::new()
+        .min_message_latency(Duration::from_millis(10))
+        .build();
+
+    sim.host("server", || async {
+        let listener = bind().await?;
+        let (mut sock, _) = listener.accept().await?;
+        sock.write_u8(1).await?;
+        turmoil::partition_oneway("server", "client");
+        Ok(())
+    });
+
+    sim.client("client", async {
+        let mut sock = TcpStream::connect(("server", PORT)).await?;
+        sock.read_u8().await?;
+        Ok(())
+    });
+
+    let result = sim.run();
+    assert!(result.is_err());
+}
+
+#[test]
+fn network_hold_holds_inflight_messages() {
+    let mut sim = Builder::new()
+        .min_message_latency(Duration::from_millis(10))
+        .build();
+
+    sim.host("server", || async {
+        let listener = bind().await?;
+        let (mut sock, _) = listener.accept().await?;
+        sock.write_u8(1).await?;
+
+        turmoil::hold("server", "client");
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        turmoil::release("server", "client");
+
+        Ok(())
+    });
+
+    sim.client("client", async {
+        let mut sock = TcpStream::connect(("server", PORT)).await?;
+        let result = timeout(Duration::from_millis(500), sock.read_u8()).await;
+        assert!(result.is_err());
+        sock.read_u8().await?;
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}