sim links push back

1 files changed, 33 insertions, 18 deletions

diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 09e7952..d218d5c 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -184,16 +184,19 @@ impl SimNet {
         self.lock().counts
     }
 
-    /// Hands a message to the net: decides now whether and when it arrives.
-    fn send(&self, from: &str, run: u64, message: &EncodedMessage) {
-        let delays = {
+    /// Hands a message to the net: decides now whether and when it arrives. Returns how
+    /// long until the link starts sending it (on a link with a rate, after those before
+    /// it): the sender waits that long, as a stream's writer waits while the stream is
+    /// full (Tor's), so what a node sends piles up in it, not on the link.
+    fn send(&self, from: &str, run: u64, message: &EncodedMessage) -> Duration {
+        let (delays, wait) = {
             let mut guard = self.lock();
             let inner = &mut *guard;
             inner.counts.sent += 1;
             let gone = inner.runs.get(from).is_some_and(|current| *current != run);
             if gone || inner.down.contains(from) || inner.down.contains(&message.peer) {
                 inner.counts.lost += 1;
-                return;
+                return Duration::ZERO;
             }
             let params = inner
                 .links
@@ -202,7 +205,7 @@ impl SimNet {
                 .clone();
             if params.drop.contains(&message.kind) || inner.rng.chance(params.loss) {
                 inner.counts.lost += 1;
-                return;
+                return Duration::ZERO;
             }
             let copies = if inner.rng.chance(params.duplicate) {
                 inner.counts.duplicated += 1;
@@ -211,10 +214,9 @@ impl SimNet {
                 1
             };
             // On a link with a rate: after those before it, as long as its size takes.
-            let queued = params
-                .rate
-                .filter(|rate| *rate > 0)
-                .map_or(Duration::ZERO, |rate| {
+            let (queued, wait) = params.rate.filter(|rate| *rate > 0).map_or(
+                (Duration::ZERO, Duration::ZERO),
+                |rate| {
                     let now = tokio::time::Instant::now();
                     let link = (from.to_owned(), message.peer.clone());
                     let start = inner
@@ -226,11 +228,13 @@ impl SimNet {
                     let free =
                         start + Duration::from_secs_f64(message.body.len() as f64 / rate as f64);
                     inner.busy_until.insert(link, free);
-                    free - now
-                });
-            (0..copies)
+                    (free - now, start - now)
+                },
+            );
+            let delays = (0..copies)
                 .map(|_| queued + inner.rng.duration_in(params.delay))
-                .collect::<Vec<_>>()
+                .collect::<Vec<_>>();
+            (delays, wait)
         };
         for delay in delays {
             let net = self.clone();
@@ -246,6 +250,7 @@ impl SimNet {
                 net.arrive(&to, delivery);
             });
         }
+        wait
     }
 
     /// The message reaches `to`, unless either end went down meanwhile.
@@ -364,7 +369,11 @@ impl Transport for SimTransport {
     /// Always accepted: the net loses messages silently, like Mycelium once it has
     /// taken them.
     async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
-        self.net.send(&self.id, self.run, message);
+        // On a link with a rate, once it has sent those before (`SimNet::send`).
+        let wait = self.net.send(&self.id, self.run, message);
+        if !wait.is_zero() {
+            tokio::time::sleep(wait).await;
+        }
         Ok(())
     }
 
@@ -503,8 +512,10 @@ mod tests {
         assert_eq!(*at_a.lock().unwrap(), vec!["b"]);
     }
 
-    /// On a link with a rate, messages go one after another, each its size at the rate:
-    /// at a message every 100 ms, one of three is in after 150 ms, all after 400.
+    /// On a link with a rate, messages go one after another, each its size at the rate,
+    /// and a sender waits while those before its message are sent: at a message every
+    /// 100 ms, sending three takes 200 ms (the third waits for two), and all are in 100
+    /// ms after.
     #[tokio::test]
     async fn a_rate_queues_messages() {
         let net = SimNet::new(1);
@@ -516,12 +527,16 @@ mod tests {
             rate: Some(size * 10),
             ..LinkParams::default()
         });
+        let started = tokio::time::Instant::now();
         for _ in 0..3 {
             a.send_message("b", &status()).await.unwrap();
         }
+        let took = started.elapsed();
+        assert!(
+            took >= Duration::from_millis(190) && took < Duration::from_millis(290),
+            "{took:?}"
+        );
         tokio::time::sleep(Duration::from_millis(150)).await;
-        assert_eq!(at_b.lock().unwrap().len(), 1);
-        tokio::time::sleep(Duration::from_millis(250)).await;
         assert_eq!(at_b.lock().unwrap().len(), 3);
     }