Adds lag tracking and compensation

apis/src/lag_tracking/decaying_stats.rs

@@ -0,0 +1,22 @@
+#[derive(Clone, Debug)]
+pub struct DecayingStats {
+    pub mean: f64,
+    pub deviation: f64,
+    pub decay: f64,
+}
+
+impl DecayingStats {
+    pub fn record(&mut self, value: f64) {
+        let delta = self.mean - value;
+        self.mean = value + self.decay * delta;
+        self.deviation = self.decay * self.deviation + (1.0 - self.decay) * delta.abs();
+    }
+
+    pub fn empty() -> Self {
+        DecayingStats {
+            mean: 0.0,
+            deviation: 4.0,
+            decay: 0.85,
+        }
+    }
+}

apis/src/lag_tracking/lag_tracker.rs

@@ -0,0 +1,103 @@
+use crate::lag_tracking::{decaying_stats::DecayingStats, stats::Stats};
+
+#[derive(Clone, Debug)]
+pub struct LagTracker {
+    pub quota_gain: f64,
+    pub quota: f64,
+    pub quota_max: f64,
+    pub lag_estimator: DecayingStats,
+    pub uncomp_stats: Stats,
+    pub lag_stats: Stats,
+    pub comp_est_sq_err: f64,
+    pub comp_est_overs: f64,
+    pub comp_estimate: Option<f64>,
+}
+
+impl LagTracker {
+    pub fn new(base: usize, inc: usize) -> Self {
+        let quota_gain = Self::quota_base_inc(base, inc);
+
+        Self {
+            quota_gain,
+            quota: quota_gain * 3.0,
+            quota_max: quota_gain * 7.0,
+            lag_estimator: DecayingStats::empty(),
+            uncomp_stats: Stats::default(),
+            lag_stats: Stats::default(),
+            comp_est_sq_err: 0.0,
+            comp_est_overs: 0.0,
+            comp_estimate: None,
+        }
+    }
+
+    fn quota_base_inc(base: usize, inc: usize) -> f64 {
+        let game_time = base as f64 + inc as f64 * 40.0;
+        ((game_time / 2.5 + 15.0) / 1000.0).min(100.0)
+    }
+
+    pub fn on_move(&mut self, lag: f64) -> f64 {
+        let comp = lag.min(self.quota);
+        let uncomped = lag - comp;
+        let ce_diff = self.comp_estimate.unwrap_or(1.0) - comp;
+        let new_quota = (self.quota + self.quota_gain - comp).min(self.quota_max);
+
+        if uncomped != 0.0 || self.uncomp_stats.samples != 0 {
+            self.uncomp_stats.record(uncomped);
+        }
+
+        self.lag_stats.record(lag.min(2000.0));
+        self.comp_est_sq_err += ce_diff * ce_diff;
+        self.comp_est_overs += ce_diff.min(0.0);
+        self.quota = new_quota;
+        comp
+    }
+
+    pub fn record_lag(&mut self, lag: f64) {
+        self.lag_estimator.record(lag);
+        self.comp_estimate = Some(
+            0_f64
+                .min(self.lag_estimator.mean - 0.8 * self.lag_estimator.deviation)
+                .min(self.quota_max),
+        );
+    }
+
+    pub fn moves(&self) -> usize {
+        self.lag_stats.samples
+    }
+
+    pub fn lag_mean(&self) -> Option<f64> {
+        if self.moves() > 0 {
+            Some(self.lag_stats.total / self.moves() as f64)
+        } else {
+            None
+        }
+    }
+
+    pub fn comp_est_std_err(&self) -> Option<f64> {
+        if self.moves() > 2 {
+            Some(self.comp_est_sq_err.sqrt() / ((self.moves() - 2) as f64))
+        } else {
+            None
+        }
+    }
+
+    pub fn comp_avg(&self) -> Option<f64> {
+        if self.moves() > 0 {
+            Some(self.total_comp() / self.moves() as f64)
+        } else {
+            None
+        }
+    }
+
+    pub fn total_comp(&self) -> f64 {
+        self.total_lag() - self.total_uncomped()
+    }
+
+    pub fn total_lag(&self) -> f64 {
+        self.lag_stats.total
+    }
+
+    pub fn total_uncomped(&self) -> f64 {
+        self.uncomp_stats.total
+    }
+}

apis/src/lag_tracking/lags.rs

@@ -0,0 +1,48 @@
+use crate::lag_tracking::lag_tracker::LagTracker;
+use shared_types::GameId;
+use std::{collections::HashMap, sync::RwLock};
+use uuid::Uuid;
+
+#[derive(Debug)]
+pub struct Lags {
+    trackers: RwLock<HashMap<(Uuid, GameId), LagTracker>>,
+}
+
+impl Lags {
+    pub fn new() -> Self {
+        Self {
+            trackers: RwLock::new(HashMap::new()),
+        }
+    }
+
+    pub fn track_lag(
+        &self,
+        uuid: Uuid,
+        game: GameId,
+        lag: f64,
+        base: usize,
+        inc: usize,
+    ) -> Option<f64> {
+        if let Ok(mut uuids_lags) = self.trackers.write() {
+            let user_lags = uuids_lags
+                .entry((uuid, game))
+                .or_insert(LagTracker::new(base, inc));
+            user_lags.record_lag(lag / 1000.0);
+            let comp = Some(user_lags.on_move(lag / 1000.0));
+            return comp;
+        }
+        None
+    }
+
+    pub fn remove(&self, uuid: Uuid, game: GameId) {
+        if let Ok(mut trackers) = self.trackers.write() {
+            trackers.remove(&(uuid, game));
+        }
+    }
+}
+
+impl Default for Lags {
+    fn default() -> Self {
+        Self::new()
+    }
+}

apis/src/lag_tracking/mod.rs

@@ -0,0 +1,4 @@
+pub mod decaying_stats;
+pub mod lag_tracker;
+pub mod lags;
+pub mod stats;

apis/src/lag_tracking/stats.rs

@@ -0,0 +1,25 @@
+#[derive(Clone, Debug)]
+pub struct Stats {
+    pub samples: usize,
+    pub total: f64,
+}
+
+impl Stats {
+    pub fn new() -> Self {
+        Stats {
+            samples: 0,
+            total: 0.0,
+        }
+    }
+
+    pub fn record(&mut self, value: f64) {
+        self.samples += 1;
+        self.total += value;
+    }
+}
+
+impl Default for Stats {
+    fn default() -> Self {
+        Self::new()
+    }
+}

apis/src/lib.rs

@@ -2,6 +2,7 @@ pub mod app;
 pub mod common;
 pub mod components;
 pub mod functions;
+pub mod lag_tracking;
 pub mod pages;
 pub mod ping;
 pub mod providers;

apis/src/main.rs

@@ -1,14 +1,15 @@
 pub mod common;
 pub mod functions;
 pub mod jobs;
+pub mod lag_tracking;
 pub mod ping;
 pub mod responses;
 pub mod websockets;
 use actix_session::config::PersistentSession;
 use actix_web::cookie::time::Duration;
 use actix_web::middleware::Compress;
+use lag_tracking::lags::Lags;
 use ping::pings::Pings;
-use std::sync::{Arc, RwLock};
 use websockets::tournament_game_start::TournamentGameStart;
 
 cfg_if::cfg_if! { if #[cfg(feature = "ssr")] {
@@ -53,7 +54,8 @@ async fn main() -> std::io::Result<()> {
         .await
         .expect("Failed to get pool");
     let chat_history = Data::new(Chats::new());
-    let pings = Data::new(Arc::new(RwLock::new(Pings::new())));
+    let pings = Data::new(Pings::new());
+    let lags = Data::new(Lags::new());
     let websocket_server = Data::new(WsServer::new(pings.clone(), pool.clone()).start());
     let tournament_game_start = Data::new(TournamentGameStart::new());
 
@@ -73,6 +75,7 @@ async fn main() -> std::io::Result<()> {
             .app_data(Data::clone(&websocket_server))
             .app_data(Data::clone(&tournament_game_start))
             .app_data(Data::clone(&pings))
+            .app_data(Data::clone(&lags))
             // serve JS/WASM/CSS from `pkg`
             .service(Files::new("/pkg", format!("{site_root}/pkg")))
             // serve other assets from the `assets` directory

apis/src/ping/pings.rs

@@ -1,10 +1,10 @@
 use super::stats::PingStats;
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::RwLock};
 use uuid::Uuid;
 
 #[derive(Debug)]
 pub struct Pings {
-    pub pings: HashMap<Uuid, PingStats>,
+    pub pings: RwLock<HashMap<Uuid, PingStats>>,
 }
 
 impl Default for Pings {
@@ -16,22 +16,25 @@ impl Default for Pings {
 impl Pings {
     pub fn new() -> Self {
         Self {
-            pings: HashMap::new(),
+            pings: HashMap::new().into(),
         }
     }
 
-    pub fn set_nonce(&mut self, user_id: Uuid, nonce: u64) {
-        let ping_stats = self.pings.entry(user_id).or_default();
+    pub fn set_nonce(&self, user_id: Uuid, nonce: u64) {
+        let mut binding = self.pings.write().unwrap();
+        let ping_stats = binding.entry(user_id).or_default();
         ping_stats.set_nonce(nonce);
     }
 
-    pub fn update(&mut self, user_id: Uuid, nonce: u64) -> f64 {
-        let ping_stats = self.pings.entry(user_id).or_default();
+    pub fn update(&self, user_id: Uuid, nonce: u64) -> f64 {
+        let mut binding = self.pings.write().unwrap();
+        let ping_stats = binding.entry(user_id).or_default();
         ping_stats.update(nonce)
     }
 
     pub fn value(&self, user_id: Uuid) -> f64 {
-        if let Some(ping_stats) = self.pings.get(&user_id) {
+        let binding = self.pings.read().unwrap();
+        if let Some(ping_stats) = binding.get(&user_id) {
             return ping_stats.value();
         }
         0.0

apis/src/websockets/api/game/handler.rs

@@ -3,6 +3,8 @@ use super::{
     control_handler::GameControlHandler, join_handler::JoinHandler,
     timeout_handler::TimeoutHandler, turn_handler::TurnHandler,
 };
+use crate::lag_tracking::lags::Lags;
+use crate::ping::pings::Pings;
 use crate::websockets::internal_server_message::InternalServerMessage;
 use crate::websockets::messages::WsMessage;
 use crate::websockets::tournament_game_start::TournamentGameStart;
@@ -25,6 +27,8 @@ pub struct GameActionHandler {
     received_from: actix::Recipient<WsMessage>,
     chat_storage: actix_web::web::Data<Chats>,
     game_start: actix_web::web::Data<TournamentGameStart>,
+    pings: actix_web::web::Data<Pings>,
+    lags: actix_web::web::Data<Lags>,
     username: String,
 }
 
@@ -36,6 +40,8 @@ impl GameActionHandler {
         received_from: actix::Recipient<WsMessage>,
         chat_storage: actix_web::web::Data<Chats>,
         game_start: actix_web::web::Data<TournamentGameStart>,
+        pings: actix_web::web::Data<Pings>,
+        lags: actix_web::web::Data<Lags>,
         pool: &DbPool,
     ) -> Result<Self> {
         let (username, user_id) = user_details;
@@ -57,6 +63,8 @@ impl GameActionHandler {
             chat_storage,
             game_start,
             user_id,
+            pings,
+            lags,
         })
     }
 
@@ -70,9 +78,17 @@ impl GameActionHandler {
             GameAction::Turn(turn) => {
                 self.ensure_not_finished()?;
                 self.ensure_user_is_player()?;
-                TurnHandler::new(turn, &self.game, &self.username, self.user_id, &self.pool)
-                    .handle()
-                    .await?
+                TurnHandler::new(
+                    turn,
+                    &self.game,
+                    &self.username,
+                    self.user_id,
+                    self.pings.clone(),
+                    self.lags.clone(),
+                    &self.pool,
+                )
+                .handle()
+                .await?
             }
             GameAction::Control(control) => {
                 self.ensure_not_finished()?;