plaza_session 0.8.1

//! Impairment on the link, applied where the link is.
//!
//! Delay, jitter and loss are properties of a connection, so they belong to
//! the transport that owns the socket rather than to a queue an arena
//! maintains beside its game state. Every frame crossing an impaired
//! connection is affected: ops, handshakes and probes alike. That is why a
//! round trip measured through here includes the impairment.
//!
//! # What a loss costs
//!
//! [`DirectionProfile::loss`] is the probability a frame is lost in transit.
//! [`Delivery`] says what that means. Each of its two variants models a
//! different kind of link.
//!
//! On a reliable stream, which is what both transports here are, a lost
//! segment never reaches the application as a missing message: TCP
//! retransmits, so it costs [`RETRANSMIT_PENALTY`] and everything queued
//! behind it waits, and what arrives is a latency spike followed by a burst.
//! Deleting a frame would model a link plaza does not have and an application
//! written against that would carry reconciliation for a case that cannot
//! occur. So [`Delivery::Reliable`] is the default and nothing is dropped.
//!
//! On a datagram link the frame is simply gone and the two ends reconcile.
//! [`Delivery::Datagram`] over a stream is therefore a *simulation*. It lets an
//! application's recovery be exercised before the channel it was written for
//! exists.
//!
//! **Over a link that really is a datagram link, do not honour it.** The
//! network loses the frame itself, so applying this too counts the loss twice.
//! `examples/foreign_soil`'s UDP body adds only delay and jitter for that
//! reason.
//!
//! No frame kind is exempt under either model. Under `Reliable` nothing is
//! lost at all. Under `Datagram` a lost probe costs one sample and a lost
//! `Hello` reads as a peer that declared nothing, which is the case that
//! handshake was built to survive.
//!
//! # Order is preserved
//!
//! Release times are made monotone as frames are queued, so a delayed frame
//! holds up everything behind it and a jitter spike arrives as a stall
//! followed by a burst. This is head-of-line blocking, which is why the
//! retransmission penalty above delays more than the one frame that was lost.

use std::collections::VecDeque;
use std::time::Duration;

use crate::manager::Frame;
use plaza_wire::frame;
use tokio::time::Instant;

/// What one retransmission costs. TCP's minimum RTO, which is the floor a real
/// stack waits before deciding a segment is gone.
pub const RETRANSMIT_PENALTY: Duration = Duration::from_millis(200);

/// What a lost frame costs, which depends on the link rather than on the
/// frame.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum Delivery {
  /// A reliable stream: the segment is retransmitted, so the frame arrives
  /// [`RETRANSMIT_PENALTY`] late and everything queued behind it waits too.
  /// What both of plaza's transports actually do.
  #[default]
  Reliable,
  /// A datagram link: the frame is gone and the two ends reconcile. Plaza has
  /// no such transport yet, so choosing this over a WebSocket simulates one.
  /// That lets an application's recovery be exercised before the channel it is
  /// for exists.
  Datagram,
}

/// Impairment applied to one direction of a link.
#[derive(Debug, Clone, Copy, Default, PartialEq)]
pub struct DirectionProfile {
  pub delay: Duration,
  /// Extra delay drawn uniformly from `[0, jitter]` per frame, then clamped by
  /// the release order of what is already queued.
  pub jitter: Duration,
  /// Probability in `[0, 1]` that a frame is lost in transit. What that costs
  /// is [`delivery`](Self::delivery)'s to say.
  pub loss: f32,
  /// What a loss does. Defaults to [`Delivery::Reliable`], which is what the
  /// transports underneath do.
  pub delivery: Delivery,
}

impl DirectionProfile {
  /// Whether this profile does nothing, in which case a frame can skip the
  /// queue entirely.
  pub fn is_passthrough(&self) -> bool {
    self.delay.is_zero() && self.jitter.is_zero() && self.loss <= 0.0
  }

  pub fn delayed(delay: Duration) -> Self {
    Self {
      delay,
      ..Self::default()
    }
  }
}

/// Impairment for one connection, in both directions.
///
/// `up` is what the client sends, `down` what the server sends. A symmetric
/// 100ms round trip is 50ms on each.
#[derive(Debug, Clone, Copy, Default, PartialEq)]
pub struct LinkProfile {
  pub up: DirectionProfile,
  pub down: DirectionProfile,
}

impl LinkProfile {
  /// The same profile in both directions.
  pub fn symmetric(profile: DirectionProfile) -> Self {
    Self {
      up: profile,
      down: profile,
    }
  }

  pub fn is_passthrough(&self) -> bool {
    self.up.is_passthrough() && self.down.is_passthrough()
  }
}

/// Where an application publishes a [`LinkProfile`] to the transport that owns
/// the links: usually `move |profile| session.set_all_link_profiles(profile)`.
///
/// A closure rather than a session handle, so the logic that decides what the
/// link should be stays testable without a socket.
pub type LinkSink = std::sync::Arc<dyn Fn(LinkProfile) + Send + Sync>;

/// A [`LinkSink`] with a latch that skips unchanged settings: the profile
/// reaches the sink only when it differs from the last one published.
pub struct LinkPublisher {
  sink: LinkSink,
  published: parking_lot::Mutex<Option<LinkProfile>>,
}

impl LinkPublisher {
  pub fn new(sink: LinkSink) -> Self {
    Self {
      sink,
      published: parking_lot::Mutex::new(None),
    }
  }

  /// Publishes `profile` if it differs from the last one published, returning
  /// whether it went out.
  pub fn publish(&self, profile: LinkProfile) -> bool {
    let mut published = self.published.lock();
    if *published == Some(profile) {
      return false;
    }
    *published = Some(profile);
    drop(published);
    (self.sink)(profile);
    true
  }
}

/// xorshift64, seeded through splitmix64 so neighbouring connection ids do not
/// produce correlated streams.
///
/// Written here rather than taken as a dependency and seeded from connection
/// ids rather than the clock, so an impaired playground session re-runs the
/// same way.
struct XorShift64(u64);

impl XorShift64 {
  fn new(seed: u64) -> Self {
    let mut z = seed.wrapping_add(0x9E37_79B9_7F4A_7C15);
    z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
    Self((z ^ (z >> 31)) | 1)
  }

  fn next_u64(&mut self) -> u64 {
    let mut x = self.0;
    x ^= x << 13;
    x ^= x >> 7;
    x ^= x << 17;
    self.0 = x;
    x
  }

  /// Uniform in `[0, 1)`, from the 24 bits an `f32` can hold exactly.
  fn next_f32(&mut self) -> f32 {
    (self.next_u64() >> 40) as f32 / (1u32 << 24) as f32
  }
}

/// One direction's delay queue.
pub struct Conditioner {
  queue: VecDeque<(Instant, Frame)>,
  last_release: Option<Instant>,
  rng: XorShift64,
  capacity: usize,
}

impl Conditioner {
  pub fn new(seed: u64, capacity: usize) -> Self {
    Self {
      queue: VecDeque::new(),
      last_release: None,
      rng: XorShift64::new(seed),
      capacity,
    }
  }

  pub fn is_empty(&self) -> bool {
    self.queue.is_empty()
  }

  /// Queues a frame. Returns whether it was queued: false when the buffer is
  /// full, or when a datagram link lost it.
  pub fn push(&mut self, frame_bytes: Frame, profile: &DirectionProfile, now: Instant) -> bool {
    // A local buffer running out rather than a network loss. Control frames
    // are still admitted: refusing a handshake here would wedge a
    // connection for a reason that has nothing to do with the link.
    if self.queue.len() >= self.capacity && frame_bytes.first() == Some(&frame::Kind::Ops.as_byte()) {
      return false;
    }
    let lost = profile.loss > 0.0 && self.rng.next_f32() < profile.loss;
    let retransmit = match (lost, profile.delivery) {
      (false, _) => Duration::ZERO,
      (true, Delivery::Reliable) => RETRANSMIT_PENALTY,
      // Gone. Both control kinds already tolerate it: a lost probe costs a
      // sample by design, and a lost `Hello` reads as a peer that declared
      // nothing, which is the case the handshake was built to survive.
      (true, Delivery::Datagram) => return false,
    };
    let jitter = profile.jitter.mul_f32(self.rng.next_f32());
    let earliest = now + profile.delay + jitter + retransmit;
    let release = match self.last_release {
      Some(previous) if previous > earliest => previous,
      _ => earliest,
    };
    self.last_release = Some(release);
    self.queue.push_back((release, frame_bytes));
    true
  }

  /// When the frame at the head comes due, if there is one.
  pub fn next_release(&self) -> Option<Instant> {
    self.queue.front().map(|(at, _)| *at)
  }

  /// The next frame that has come due, or `None`.
  pub fn pop_ready(&mut self, now: Instant) -> Option<Frame> {
    match self.queue.front() {
      Some((at, _)) if *at <= now => self.queue.pop_front().map(|(_, frame_bytes)| frame_bytes),
      _ => None,
    }
  }

  /// Everything still held, in order, release times ignored. For a close that
  /// flushes what was queued rather than waiting out a simulated delay.
  pub fn drain(&mut self) -> Vec<Frame> {
    self.last_release = None;
    self.queue.drain(..).map(|(_, frame_bytes)| frame_bytes).collect()
  }
}

#[cfg(test)]
mod tests {
  use super::*;
  use crate::manager::DEFAULT_CONDITIONER_CAPACITY as CAP;

  fn ops_frame(n: u8) -> Frame {
    Frame::from(vec![frame::Kind::Ops.as_byte(), n])
  }

  fn ping_frame() -> Frame {
    Frame::from(vec![frame::Kind::Ping.as_byte()])
  }

  #[test]
  fn a_passthrough_profile_is_the_default() {
    assert!(DirectionProfile::default().is_passthrough());
    assert!(LinkProfile::default().is_passthrough());
  }

  #[tokio::test]
  async fn jitter_stalls_the_queue_instead_of_reordering_it() {
    // The property a reliable stream has and a datagram link does not: however
    // the per-frame jitter falls, what comes out is what went in, in order.
    let mut c = Conditioner::new(1, CAP);
    let profile = DirectionProfile {
      delay: Duration::from_millis(10),
      jitter: Duration::from_millis(200),
      ..DirectionProfile::default()
    };
    let now = Instant::now();
    for n in 0..32 {
      assert!(c.push(ops_frame(n), &profile, now));
    }

    let mut releases = Vec::new();
    let mut seen = Vec::new();
    while let Some(at) = c.next_release() {
      releases.push(at);
      seen.push(c.pop_ready(at).expect("due at its own release time")[1]);
    }

    assert_eq!(seen, (0..32).collect::<Vec<_>>(), "arrival order survives jitter");
    assert!(
      releases.windows(2).all(|w| w[0] <= w[1]),
      "a delayed frame holds up the ones behind it"
    );
  }

  #[tokio::test]
  async fn a_frame_is_not_released_before_its_time() {
    let mut c = Conditioner::new(2, CAP);
    let now = Instant::now();
    c.push(ops_frame(1), &DirectionProfile::delayed(Duration::from_millis(50)), now);

    assert!(c.pop_ready(now).is_none(), "not due yet");
    assert!(c.pop_ready(now + Duration::from_millis(49)).is_none());
    assert!(c.pop_ready(now + Duration::from_millis(50)).is_some());
  }

  #[tokio::test]
  async fn a_reliable_link_pays_for_a_loss_in_time_rather_than_in_frames() {
    // What a WebSocket actually does: nothing goes missing whatever the slider
    // reads, because TCP retransmits and the application only sees the delay.
    let mut c = Conditioner::new(3, CAP);
    let certain_loss = DirectionProfile {
      loss: 1.0,
      ..DirectionProfile::default()
    };
    let now = Instant::now();

    for n in 0..16 {
      assert!(c.push(ops_frame(n), &certain_loss, now), "no frame is deleted");
    }
    assert!(c.push(ping_frame(), &certain_loss, now));

    assert_eq!(
      c.next_release(),
      Some(now + RETRANSMIT_PENALTY),
      "the first frame waits out one retransmission"
    );
    let mut seen = 0;
    while c.pop_ready(now + RETRANSMIT_PENALTY * 2).is_some() {
      seen += 1;
    }
    assert_eq!(seen, 17, "and every one of them still arrives");
  }

  #[tokio::test]
  async fn a_datagram_link_loses_frames_outright() {
    // The simulation, for exercising an application's recovery before the
    // channel it is written for exists.
    let mut c = Conditioner::new(3, CAP);
    let certain_loss = DirectionProfile {
      loss: 1.0,
      delivery: Delivery::Datagram,
      ..DirectionProfile::default()
    };
    let now = Instant::now();

    for n in 0..16 {
      assert!(!c.push(ops_frame(n), &certain_loss, now), "gone, not delayed");
    }
    assert!(!c.push(ping_frame(), &certain_loss, now), "a probe is not exempt either");
    assert!(c.is_empty());
  }

  #[tokio::test]
  async fn a_full_queue_still_admits_control_frames() {
    let mut c = Conditioner::new(4, CAP);
    let profile = DirectionProfile::delayed(Duration::from_secs(1));
    let now = Instant::now();

    for _ in 0..crate::manager::DEFAULT_CONDITIONER_CAPACITY {
      assert!(c.push(ops_frame(0), &profile, now));
    }
    assert!(!c.push(ops_frame(0), &profile, now), "the buffer is finite");
    assert!(c.push(ping_frame(), &profile, now), "a probe is not what overflows it");
  }

  #[tokio::test]
  async fn a_configured_capacity_is_what_fills() {
    let mut c = Conditioner::new(4, 3);
    let profile = DirectionProfile::delayed(Duration::from_secs(1));
    let now = Instant::now();

    for _ in 0..3 {
      assert!(c.push(ops_frame(0), &profile, now));
    }
    assert!(!c.push(ops_frame(0), &profile, now));
  }

  #[tokio::test]
  async fn an_empty_queue_has_nothing_to_wait_for() {
    let mut c = Conditioner::new(5, CAP);
    assert!(c.next_release().is_none());
    assert!(c.pop_ready(Instant::now()).is_none());
  }
}