Add connection keepalives; Disconnect reasions

9 files changed, 63 insertions, 53 deletions

diff --git a/client/game.gd b/client/game.gd
index 62a25f18..2c65138d 100644
--- a/client/game.gd
+++ b/client/game.gd
@@ -387,6 +387,8 @@ func handle_packet(p):
 			$Environment.update(p.effects)
 		"redirect":
 			get_parent().replace_menu("res://gui/menus/game.tscn", p.uri[0])
+		"disconnect":
+			get_parent().replace_menu("res://gui/menus/error.tscn", [tr(p.reason.translation.id), menu.data])
 		"replay_start":
 			is_replay = true
 		"replay_stop":
diff --git a/client/multiplayer.gd b/client/multiplayer.gd
index 3ee303ee..aa9c2c75 100644
--- a/client/multiplayer.gd
+++ b/client/multiplayer.gd
@@ -49,10 +49,9 @@ func _process(_delta):
 	if connected:
 		socket.poll()
 		var state = socket.get_ready_state()
-		if state == WebSocketPeer.STATE_OPEN:
-			while socket.get_available_packet_count():
-				handle_packet(socket.get_packet())
-		elif state == WebSocketPeer.STATE_CLOSED:
+		while socket.get_available_packet_count():
+			handle_packet(socket.get_packet())
+		if state == WebSocketPeer.STATE_CLOSED:
 			var code = socket.get_close_code()
 			var reason = socket.get_close_reason() if code == socket.STATE_CLOSED else tr("c.error.websocket.unavailable")
 			connection_closed.emit(tr("c.error.websocket").format([code, reason, code != -1]))
diff --git a/locale/en.ini b/locale/en.ini
index 1fdc7672..4ae90ff7 100644
--- a/locale/en.ini
+++ b/locale/en.ini
@@ -315,7 +315,9 @@ s.error.too_many_players=Too many players joined
 s.error.tutorial_already_running=Tutorial already running
 s.error.tutorial_no_running=No tutorial running
 s.error.username_length_limit=Username length limit of {0} characters or {1} bytes exceeded.
-s.disconnect_reason.inactivity_kick=You were disconnected because of inactivity. Please reconnect.
+s.disconnect_reason.keepalive_timer=Your connection was interrupted. Please reconnect.
+s.disconnect_reason.channel_overflow=Your connection was too weak. Please reconnect.
+s.disconnect_reason.invalid_packet=Your client sent an invalid packet: {0}
 s.replay.cannot_join=Replays cannot be joined.
 s.state.abort_no_players=Game was aborted due to a lack of players.
 s.state.game_aborted=Game was aborted by {0}.
diff --git a/server/protocol/src/lib.rs b/server/protocol/src/lib.rs
index 15911dd3..6b4e97f1 100644
--- a/server/protocol/src/lib.rs
+++ b/server/protocol/src/lib.rs
@@ -43,6 +43,8 @@ fn test_version_parse() {
     let _ = *VERSION;
 }
 
+pub const KEEPALIVE_INTERVAL: f32 = 1.;
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
 #[serde(transparent)]
 pub struct PlayerID(#[serde(deserialize_with = "deser_i64")] pub i64);
@@ -142,6 +144,7 @@ pub enum PacketS {
     Idle {
         paused: bool,
     },
+    Keepalive,
     Ready,
 
     /// For use in replay sessions only
diff --git a/server/src/main.rs b/server/src/main.rs
index 3fbefaf8..3bf39f5a 100644
--- a/server/src/main.rs
+++ b/server/src/main.rs
@@ -33,7 +33,7 @@ use tokio::{
     net::{TcpListener, TcpStream},
     spawn,
     sync::{RwLock, broadcast},
-    time::interval,
+    time::{interval, sleep},
 };
 use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
 
@@ -51,9 +51,9 @@ pub(crate) struct Args {
     /// Map name to use as lobby
     #[arg(long, default_value = "lobby")]
     lobby: String,
-    /// Inactivity kick timeout in seconds
+    /// Inactivity timeout in seconds
     #[arg(long, default_value_t = 60.)]
-    inactivity_kick_timeout: f32,
+    inactivity_timeout: f32,
     /// Registers this server to the public server registry
     #[arg(long)]
     #[cfg(feature = "register")]
@@ -125,7 +125,7 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
     let (tx, _) = broadcast::channel::<PacketC>(128 * 1024);
 
     let config = ServerConfig {
-        inactivity_kick_timeout: args.inactivity_kick_timeout,
+        inactivity_timeout: args.inactivity_timeout,
         lobby: args.lobby,
     };
 
@@ -214,14 +214,9 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
                         Ok(packet) => {
                             if send_packet(id, &mut sock, packet).await { break; };
                         },
-                        Err(e) => {
-                            broadcast_rx = broadcast_rx.resubscribe();
-                            warn!("{id} Client was lagging; resubscribed: {e}");
-                            let packet =  PacketC::ServerMessage {
-                                message: trm!("s.state.overflow_resubscribe"),
-                                error: true,
-                            };
-                            if send_packet(id, &mut sock, packet).await { break; };
+                        Err(_) => {
+                            warn!("{id} Broadcast packet channel overflowed");
+                            state.write().await.disconnect(id, Some(trm!("s.disconnect_reason.channel_overflow")));
                         }
                     },
                     Some(Ok(message)) = sock.next() => {
@@ -229,7 +224,8 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
                             Message::Text(line) if line.len() < 8196 => match serde_json::from_str(&line) {
                                 Ok(p) => p,
                                 Err(e) => {
-                                    warn!("{id} Invalid json packet: {e}");
+                                    warn!("{id} Invalid packet: {e}");
+                                    state.write().await.disconnect(id, Some(trm!("s.disconnect_reason.invalid_packet", s = e.to_string())));
                                     break;
                                 }
                             },
@@ -243,7 +239,7 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
 
                         if matches!(
                             packet,
-                            PacketS::Movement { .. } | PacketS::ReplayTick { .. }
+                            PacketS::Movement { .. } | PacketS::ReplayTick { .. } | PacketS::Keepalive
                         ) {
                             trace!("{id} <- {packet:?}");
                         } else {
@@ -252,7 +248,7 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
                         let packet_out = match state.write().await.packet_in_outer(id, packet) {
                             Ok(packets) => packets,
                             Err(e) => {
-                                warn!("Client error: {e}");
+                                warn!("{id} Packet error: {e}");
                                 vec![PacketC::ServerMessage {
                                     message: e.into(),
                                     error: true,
@@ -265,10 +261,8 @@ async fn run(data_path: PathBuf, args: Args) -> anyhow::Result<()> {
                    }
                 };
             }
-            state
-                .write()
-                .await
-                .disconnect(id, hurrycurry_protocol::Message::Text(String::default()));
+            state.write().await.disconnect(id, None);
+            sleep(Duration::from_millis(100)).await; // avoids potential godot bug where disconnect packets are lost
         });
     }
     Ok(())
diff --git a/server/src/server.rs b/server/src/server.rs
index 47f556e2..fc2df029 100644
--- a/server/src/server.rs
+++ b/server/src/server.rs
@@ -49,7 +49,7 @@ pub struct ConnectionData {
     pub players: HashSet<PlayerID>,
     pub idle: bool,
     pub ready: bool,
-    pub last_player_input: f32,
+    pub keepalive_timer: f32,
     pub replies: mpsc::Sender<PacketC>,
 }
 
@@ -60,7 +60,7 @@ pub enum AnnounceState {
 }
 
 pub struct ServerConfig {
-    pub inactivity_kick_timeout: f32,
+    pub inactivity_timeout: f32,
     pub lobby: String,
 }
 
@@ -114,7 +114,7 @@ impl Server {
 impl Default for ServerConfig {
     fn default() -> Self {
         Self {
-            inactivity_kick_timeout: 60.,
+            inactivity_timeout: 60.,
             lobby: "lobby".to_string(),
         }
     }
@@ -642,7 +642,7 @@ impl Server {
                 self.score_changed = true;
             }
             PacketS::ReplayTick { .. } => return Err(tre!("s.error.packet_not_supported")),
-            PacketS::Idle { .. } | PacketS::Ready => (),
+            PacketS::Idle { .. } | PacketS::Ready | PacketS::Keepalive => (),
             PacketS::Debug(d) => self.packet_out.push_back(PacketC::Debug(d)),
         }
         Ok(())
diff --git a/server/src/state.rs b/server/src/state.rs
index 1b42243f..23339543 100644
--- a/server/src/state.rs
+++ b/server/src/state.rs
@@ -21,8 +21,8 @@ use crate::{
 };
 use anyhow::Result;
 use hurrycurry_locale::{TrError, tre, trm};
-use hurrycurry_protocol::{Menu, Message, PacketC, PacketS, PlayerID, VERSION};
-use log::{debug, info, trace};
+use hurrycurry_protocol::{KEEPALIVE_INTERVAL, Menu, Message, PacketC, PacketS, PlayerID, VERSION};
+use log::{debug, info, trace, warn};
 use std::{
     collections::HashSet,
     time::{Duration, Instant},
@@ -31,15 +31,16 @@ use tokio::sync::{broadcast, mpsc};
 
 impl Server {
     pub fn tick_outer(&mut self, dt: f32) -> anyhow::Result<()> {
-        let mut idle_kick = Vec::new();
+        let mut keepalive_kick = Vec::new();
         for (cid, conn) in &mut self.connections {
-            conn.last_player_input += dt;
-            if conn.last_player_input > self.config.inactivity_kick_timeout {
-                idle_kick.push(*cid);
+            conn.keepalive_timer += dt;
+            if conn.keepalive_timer > KEEPALIVE_INTERVAL + 10. {
+                keepalive_kick.push(*cid);
             }
         }
-        for cid in idle_kick {
-            self.disconnect(cid, trm!("s.disconnect_reason.inactivity_kick"));
+        for cid in keepalive_kick {
+            warn!("{cid} Client did not send keepalive in time");
+            self.disconnect(cid, Some(trm!("s.disconnect_reason.keepalive_timer")));
         }
 
         if !self.paused {
@@ -111,7 +112,7 @@ impl Server {
                 idle: false,
                 ready: false,
                 players: HashSet::new(),
-                last_player_input: 0.,
+                keepalive_timer: 0.,
                 replies: replies_tx,
             },
         );
@@ -119,9 +120,11 @@ impl Server {
         (init, broadcast_rx, replies_rx)
     }
 
-    pub fn disconnect(&mut self, conn: ConnectionID, reason: Message) {
+    pub fn disconnect(&mut self, conn: ConnectionID, reason: Option<Message>) {
         if let Some(cd) = self.connections.get(&conn) {
-            let _ = cd.replies.try_send(PacketC::Disconnect { reason });
+            if let Some(reason) = reason {
+                let _ = cd.replies.try_send(PacketC::Disconnect { reason });
+            }
             for player in cd.players.clone() {
                 let _ = self.packet_in_outer(conn, PacketS::Leave { player });
             }
@@ -135,8 +138,11 @@ impl Server {
         conn: ConnectionID,
         packet: PacketS,
     ) -> Result<Vec<PacketC>, TrError> {
+        let Some(conn_data) = self.connections.get_mut(&conn) else {
+            return Ok(vec![]);
+        };
         if let Some(p) = get_packet_player(&packet)
-            && !self.connections.get(&conn).unwrap().players.contains(&p)
+            && !conn_data.players.contains(&p)
         {
             return Err(tre!("s.error.packet_sender_invalid"));
         }
@@ -159,11 +165,11 @@ impl Server {
                 }
             }
             PacketS::Ready => {
-                self.connections.get_mut(&conn).unwrap().ready = true;
+                conn_data.ready = true;
                 self.update_paused();
             }
             PacketS::Idle { paused } => {
-                self.connections.get_mut(&conn).unwrap().idle = *paused;
+                conn_data.idle = *paused;
                 self.update_paused();
             }
             PacketS::Leave { player } => {
@@ -174,17 +180,12 @@ impl Server {
                     .remove(player);
             }
             PacketS::Join { .. } => {
-                if self.connections.get_mut(&conn).unwrap().players.len() > 8 {
+                if conn_data.players.len() > 8 {
                     return Err(tre!("s.error.conn_too_many_players"));
                 }
             }
-            PacketS::Interact { .. }
-            | PacketS::Communicate { .. }
-            | PacketS::Movement { boost: true, .. } => {
-                self.connections.get_mut(&conn).unwrap().last_player_input = 0.;
-            }
-            PacketS::Movement { dir, .. } if dir.length() > 0.5 => {
-                self.connections.get_mut(&conn).unwrap().last_player_input = 0.;
+            PacketS::Keepalive => {
+                conn_data.keepalive_timer = 0.;
             }
             _ => (),
         }
@@ -280,6 +281,7 @@ fn get_packet_player(packet: &PacketS) -> Option<PlayerID> {
         | PacketS::Ready
         | PacketS::ApplyScore(_)
         | PacketS::ReplayTick { .. }
-        | PacketS::Debug(_) => None,
+        | PacketS::Debug(_)
+        | PacketS::Keepalive => None,
     }
 }
diff --git a/test-client/main.ts b/test-client/main.ts
index 865e1fb3..f66a46a4 100644
--- a/test-client/main.ts
+++ b/test-client/main.ts
@@ -49,10 +49,17 @@ document.addEventListener("DOMContentLoaded", async () => {
     ws.onmessage = m => {
         packet(JSON.parse(m.data) as PacketC);
     }
-    ws.onclose = () => console.log("close")
+    let keepalive_interval: number
+    ws.onclose = () => {
+        console.log("close")
+        clearInterval(keepalive_interval)
+    }
     ws.onopen = () => {
         console.log("open")
         send({ type: "join", name: "test", character: { color: Math.floor(Math.random() * 100), hairstyle: 0, headwear: 0 }, class: "chef" })
+        keepalive_interval = setInterval(() => {
+            send({ type: "keepalive" })
+        }, 1000)
     }
 
     canvas = document.createElement("canvas");
@@ -156,7 +163,7 @@ function get_item_location_tracking(loc: ItemLocation): V2 {
 }
 
 function send(p: PacketS) {
-    if (p.type != "movement") console.log("send", p);
+    if (p.type != "movement" && p.type != "keepalive") console.log("send", p);
     ws.send(JSON.stringify(p))
 }
 function packet(p: PacketC) {
diff --git a/test-client/protocol.ts b/test-client/protocol.ts
index 16c1fb98..293a085d 100644
--- a/test-client/protocol.ts
+++ b/test-client/protocol.ts
@@ -49,6 +49,7 @@ export type PacketS =
     | { type: "effect", player: PlayerID, name: string } // Sends an effect
     | { type: "replay_tick", dt: number } // Steps forward in replay.
     | { type: "ready" }
+    | { type: "keepalive" }
 
 export type PacketC =
     { type: "version", minor: number, major: number, supports_bincode?: boolean } // Sent once after connecting to ensure you client is compatible