1 files changed, 71 insertions, 10 deletions

diff --git a/src/server/core.rs b/src/server/core.rs
index 3cd6469..0b2efb4 100644
--- a/src/server/core.rs
+++ b/src/server/core.rs
@@ -1,22 +1,26 @@
-use bcrypt::{hash, verify, DEFAULT_COST};
+use bcrypt::{DEFAULT_COST, hash, verify};
 use rusqlite::Connection;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 
 use crate::blackjack::BlackjackState;
 use crate::crash::CrashState;
+use crate::protocol::{REPLAY_CACHE_LEN, Request};
 use crate::server::database;
-use crate::version::{download_url, hello_line, Version, VERSION};
 use crate::server::games::blackjack::{BlackjackGame, BlackjackHand};
 use crate::server::games::crash::{Clock, CrashGame, CrashRound, RealClock};
 use crate::server::games::registry::GameRegistry;
 use crate::server::games::session::SessionManager;
 use crate::server::games::traits::{GameError, GamePhase, GameState};
+use crate::version::{VERSION, Version, download_url, hello_line, pong_line, unix_time_ms};
 
 #[derive(Debug, Default)]
 pub struct ConnectionState {
     user: Option<String>,
     greeted: bool,
+    /// Last replies on this connection, oldest first. Request ids are unique
+    /// while cached, so a retried request replays its first result.
+    replays: VecDeque<(u64, String, String)>,
 }
 
 /// Settle reservations left behind by a previous process.
@@ -39,6 +43,27 @@ impl ConnectionState {
     pub fn greeted(&self) -> bool {
         self.greeted
     }
+
+    fn replay(&self, id: u64, payload: &str) -> Option<Result<String, String>> {
+        self.replays
+            .iter()
+            .find(|(cached_id, _, _)| *cached_id == id)
+            .map(|(_, cached_payload, reply)| {
+                if cached_payload == payload {
+                    Ok(reply.clone())
+                } else {
+                    Err("ERROR Request id reused with a different command".to_string())
+                }
+            })
+    }
+
+    fn remember(&mut self, id: u64, payload: &str, reply: &str) {
+        if self.replays.len() >= REPLAY_CACHE_LEN {
+            self.replays.pop_front();
+        }
+        self.replays
+            .push_back((id, payload.to_string(), reply.to_string()));
+    }
 }
 
 pub struct ServerCore {
@@ -49,6 +74,8 @@ pub struct ServerCore {
     blackjack: BlackjackGame,
     crash: CrashGame,
     clock: Arc<dyn Clock>,
+    /// Last time a crash tick was pushed, in seconds on the game clock.
+    last_tick: HashMap<String, f64>,
 }
 
 impl ServerCore {
@@ -65,6 +92,7 @@ impl ServerCore {
             blackjack: BlackjackGame::new(),
             crash: CrashGame::new(),
             clock: Arc::new(RealClock::new()),
+            last_tick: HashMap::new(),
         }
     }
 
@@ -96,6 +124,21 @@ impl ServerCore {
         Ok(Self::new(conn))
     }
 
+    /// Handle one enveloped request, replaying a cached reply for a repeated
+    /// request id. The payload grammar is unchanged; the envelope only adds
+    /// correlation and idempotency.
+    pub fn handle_request(&mut self, state: &mut ConnectionState, request: &Request) -> String {
+        if let Some(reply) = state.replay(request.id, &request.payload) {
+            return match reply {
+                Ok(reply) => reply,
+                Err(message) => message,
+            };
+        }
+        let reply = self.handle_message(state, &request.payload);
+        state.remember(request.id, &request.payload, &reply);
+        reply
+    }
+
     pub fn handle_message(&mut self, state: &mut ConnectionState, msg: &str) -> String {
         let parts: Vec<&str> = msg.splitn(5, ' ').collect();
 
@@ -201,7 +244,7 @@ impl ServerCore {
                     }
                 }
             }
-            "PING" => "PONG".to_string(),
+            "PING" => pong_line(unix_time_ms()),
             "INFO" => self.handle_info(state),
             _ => msg.to_string(),
         }
@@ -212,7 +255,8 @@ impl ServerCore {
             return "ERROR Invalid HELLO format. Usage: HELLO <version>".to_string();
         }
 
-        let (Some(client), Some(server)) = (Version::parse(parts[1]), Version::parse(VERSION)) else {
+        let (Some(client), Some(server)) = (Version::parse(parts[1]), Version::parse(VERSION))
+        else {
             return "ERROR Invalid version".to_string();
         };
 
@@ -225,7 +269,7 @@ impl ServerCore {
         }
 
         state.greeted = true;
-        hello_line(VERSION, VERSION, &download_url())
+        hello_line(VERSION, VERSION, &download_url(), unix_time_ms())
     }
 
     fn handle_register(&mut self, username: &str, credential: &str) -> String {
@@ -529,8 +573,8 @@ impl ServerCore {
 
         // The session is the audit guard for the deduction, so it exists before
         // the balance is touched.
-        let crash_point = self.crash.next_crash_point();
-        let round = CrashRound::new(bet as u64, crash_point, self.clock.now());
+        let seed = self.crash.next_seed();
+        let round = CrashRound::new(bet as u64, seed, self.clock.now(), unix_time_ms());
         let mut data = HashMap::new();
         data.insert("round".to_string(), round.to_json());
         data.insert("balance".to_string(), balance.to_string());
@@ -643,11 +687,27 @@ impl ServerCore {
             if !round.is_running() {
                 continue;
             }
-            let elapsed = (self.clock.now() - round.started_at).max(0.0);
-            if crate::crash::multiplier_at_secs(elapsed) < round.crash_point {
+            let now = self.clock.now();
+            let elapsed = (now - round.started_at).max(0.0);
+            let multiplier = crate::crash::multiplier_at_secs(elapsed);
+            if multiplier < round.crash_point {
+                let last = self
+                    .last_tick
+                    .get(&session.session_id)
+                    .copied()
+                    .unwrap_or(f64::NEG_INFINITY);
+                if now - last >= 0.1 {
+                    self.last_tick.insert(session.session_id.clone(), now);
+                    out.push(format!(
+                        "crash.tick session={} t={} m={multiplier:.4}",
+                        session.session_id,
+                        unix_time_ms()
+                    ));
+                }
                 continue;
             }
             round.crash();
+            self.last_tick.remove(&session.session_id);
             match self.settle_crash(user, &session.session_id, &round, 0) {
                 Ok(balance) => {
                     let state = round.state(&session.session_id, balance);
@@ -678,6 +738,7 @@ impl ServerCore {
             payout as i64,
             &details,
         );
+        self.last_tick.remove(session_id);
         let _ = self.sessions.remove_session(session_id);
         result.map_err(|error| error.to_string())
     }