Repository: serenity-rs/serenity (language: rust)

Issue:
Sequence number does not get updated for invalid events
Currently on the next branch, although this may occur on the current branch, if an unknown event is received a warn is emitted in the style of 
```
[serenity::gateway::shard]: [ShardInfo { id: ShardId(..), total: .. }] Sequence off; them: 266560, us: 266558
```

This is because the sequence number is part of the GatewayEvent payload and therefore when deserialisation fails it cannot update the Shard's sequence to fix this. Possible fixes would be to make `GatewayEvent` lazy, deserialising to `JsonMap` before deserialising again to `Event` after fetching the sequence number, or maybe some other hack is possible.


Discussion / hints:

Fix patch:
diff --git a/src/gateway/bridge/shard_runner.rs b/src/gateway/bridge/shard_runner.rs
index f64816365ea..83ca87faf3c 100644
--- a/src/gateway/bridge/shard_runner.rs
+++ b/src/gateway/bridge/shard_runner.rs
@@ -421,43 +421,37 @@ impl ShardRunner {
             Err(why) => Err(why),
         };
 
-        let action = match self.shard.handle_event(&gateway_event) {
-            Ok(Some(action)) => Some(action),
-            Ok(None) => None,
+        let is_ack = matches!(gateway_event, Ok(GatewayEvent::HeartbeatAck));
+        let (action, event) = match self.shard.handle_event(gateway_event) {
+            Ok((action, event)) => (action, event),
+            Err(Error::Gateway(
+                why @ (GatewayError::InvalidAuthentication
+                | GatewayError::InvalidGatewayIntents
+                | GatewayError::DisallowedGatewayIntents),
+            )) => {
+                error!("Shard handler received fatal err: {why:?}");
+
+                self.manager.return_with_value(Err(why.clone())).await;
+                return Err(Error::Gateway(why));
+            },
+            Err(Error::Json(_)) => return Ok((None, None, true)),
             Err(why) => {
-                error!("Shard handler received err: {:?}", why);
-
-                match &why {
-                    Error::Gateway(
-                        error @ (GatewayError::InvalidAuthentication
-                        | GatewayError::InvalidGatewayIntents
-                        | GatewayError::DisallowedGatewayIntents),
-                    ) => {
-                        self.manager.return_with_value(Err(error.clone())).await;
-
-                        return Err(why);
-                    },
-                    _ => return Ok((None, None, true)),
-                }
+                error!("Shard handler recieved err: {why:?}");
+                return Ok((None, None, true));
             },
         };
 
-        if let Ok(GatewayEvent::HeartbeatAck) = gateway_event {
+        if is_ack {
             self.update_manager().await;
         }
 
         #[cfg(feature = "voice")]
         {
-            if let Ok(GatewayEvent::Dispatch(_, ref event)) = gateway_event {
+            if let Some(event) = &event {
                 self.handle_voice_event(event).await;
             }
         }
 
-        let event = match gateway_event {
-            Ok(GatewayEvent::Dispatch(_, event)) => Some(event),
-            _ => None,
-        };
-
         Ok((event, action, true))
     }
 
diff --git a/src/gateway/shard.rs b/src/gateway/shard.rs
index 513db7ca68b..e4c39339f13 100644
--- a/src/gateway/shard.rs
+++ b/src/gateway/shard.rs
@@ -266,11 +266,19 @@ impl Shard {
     }
 
     #[cfg_attr(feature = "tracing_instrument", instrument(skip(self)))]
-    fn handle_gateway_dispatch(&mut self, seq: u64, event: &Event) -> Option<ShardAction> {
+    fn handle_gateway_dispatch(
+        &mut self,
+        seq: u64,
+        event: JsonMap,
+        original_str: &str,
+    ) -> Result<(Option<ShardAction>, Option<Event>)> {
         if seq > self.seq + 1 {
             warn!("[{:?}] Sequence off; them: {}, us: {}", self.shard_info, seq, self.seq);
         }
 
+        self.seq = seq;
+        let event = Event::deserialize_and_log(event, original_str)?;
+
         match &event {
             Event::Ready(ready) => {
                 debug!("[{:?}] Received Ready", self.shard_info);
@@ -293,9 +301,7 @@ impl Shard {
             _ => {},
         }
 
-        self.seq = seq;
-
-        None
+        Ok((None, Some(event)))
     }
 
     #[cfg_attr(feature = "tracing_instrument", instrument(skip(self)))]
@@ -428,10 +434,19 @@ impl Shard {
     /// Returns a [`GatewayError::OverloadedShard`] if the shard would have too many guilds
     /// assigned to it.
     #[cfg_attr(feature = "tracing_instrument", instrument(skip(self)))]
-    pub fn handle_event(&mut self, event: &Result<GatewayEvent>) -> Result<Option<ShardAction>> {
-        match event {
-            Ok(GatewayEvent::Dispatch(seq, event)) => Ok(self.handle_gateway_dispatch(*seq, event)),
-            Ok(GatewayEvent::Heartbeat(s)) => Ok(Some(self.handle_heartbeat_event(*s))),
+    pub fn handle_event(
+        &mut self,
+        event: Result<GatewayEvent>,
+    ) -> Result<(Option<ShardAction>, Option<Event>)> {
+        let action = match event {
+            Ok(GatewayEvent::Dispatch {
+                seq,
+                data,
+                original_str,
+            }) => {
+                return self.handle_gateway_dispatch(seq, data, &original_str);
+            },
+            Ok(GatewayEvent::Heartbeat(s)) => Ok(Some(self.handle_heartbeat_event(s))),
             Ok(GatewayEvent::HeartbeatAck) => {
                 self.last_heartbeat_ack = Some(Instant::now());
                 self.last_heartbeat_acknowledged = true;
@@ -440,11 +455,11 @@ impl Shard {
 
                 Ok(None)
             },
-            &Ok(GatewayEvent::Hello(interval)) => {
+            Ok(GatewayEvent::Hello(interval)) => {
                 debug!("[{:?}] Received a Hello; interval: {}", self.shard_info, interval);
 
                 if self.stage == ConnectionStage::Resuming {
-                    return Ok(None);
+                    return Ok((None, None));
                 }
 
                 self.heartbeat_interval = Some(std::time::Duration::from_millis(interval));
@@ -457,7 +472,7 @@ impl Shard {
                     ShardAction::Reconnect(self.reconnection_type())
                 }))
             },
-            &Ok(GatewayEvent::InvalidateSession(resumable)) => {
+            Ok(GatewayEvent::InvalidateSession(resumable)) => {
                 info!("[{:?}] Received session invalidation", self.shard_info);
 
                 Ok(Some(if resumable {
@@ -477,15 +492,12 @@ impl Shard {
                 Ok(Some(ShardAction::Reconnect(self.reconnection_type())))
             },
             Err(why) => {
-                if let Error::Json(_) = why {
-                    // Deserialization errors already get logged when the event is first received
-                } else {
-                    warn!("[{:?}] Unhandled error: {:?}", self.shard_info, why);
-                }
-
+                warn!("[{:?}] Unhandled error: {:?}", self.shard_info, why);
                 Ok(None)
             },
-        }
+        };
+
+        action.map(|a| (a, None))
     }
 
     /// Does a heartbeat if needed. Returns false if something went wrong and the shard should be
diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs
index 0fbf6707e79..d479d8b6962 100644
--- a/src/gateway/ws.rs
+++ b/src/gateway/ws.rs
@@ -8,6 +8,8 @@ use flate2::read::ZlibDecoder;
 use futures::SinkExt;
 #[cfg(feature = "client")]
 use futures::StreamExt;
+#[cfg(feature = "client")]
+use small_fixed_array::FixedString;
 use tokio::net::TcpStream;
 #[cfg(feature = "client")]
 use tokio::time::{timeout, Duration};
@@ -138,9 +140,22 @@ impl WsClient {
             _ => return Ok(None),
         };
 
-        serde_json::from_str(&json_str)
-            .map(Some)
-            .map_err(|err| log_deserialisation_err(&json_str, err))
+        match serde_json::from_str(&json_str) {
+            Ok(mut event) => {
+                if let GatewayEvent::Dispatch {
+                    original_str, ..
+                } = &mut event
+                {
+                    *original_str = FixedString::from_string_trunc(json_str);
+                }
+
+                Ok(Some(event))
+            },
+            Err(err) => {
+                debug!("Failing text: {json_str}");
+                Err(Error::Json(err))
+            },
+        }
     }
 
     pub(crate) async fn send_json(&mut self, value: &impl serde::Serialize) -> Result<()> {
@@ -307,24 +322,3 @@ impl WsClient {
         .await
     }
 }
-
-fn filter_unknown_variant(json_err_dbg: &str) -> bool {
-    if let Some(msg) = json_err_dbg.strip_prefix("Error(\"unknown variant `") {
-        if let Some((variant_name, _)) = msg.split_once('`') {
-            tracing::debug!("Unknown event: {variant_name}");
-            return true;
-        }
-    }
-
-    false
-}
-
-fn log_deserialisation_err(json_str: &str, err: serde_json::Error) -> Error {
-    let json_err_dbg = format!("{err:?}");
-    if !filter_unknown_variant(&json_err_dbg) {
-        warn!("Err deserializing text: {json_err_dbg}");
-    }
-
-    debug!("Failing text: {json_str}");
-    Error::Json(err)
-}
diff --git a/src/model/event.rs b/src/model/event.rs
index 7b1d7506835..6b9a88bcd1f 100644
--- a/src/model/event.rs
+++ b/src/model/event.rs
@@ -11,6 +11,7 @@ use std::collections::HashMap;
 use nonmax::NonMaxU64;
 use serde::de::Error as DeError;
 use serde::Serialize;
+use tracing::{debug, warn};
 
 use super::application::ActionRow;
 use super::prelude::*;
@@ -1057,12 +1058,17 @@ pub struct EntitlementDeleteEvent {
 
 /// [Discord docs](https://discord.com/developers/docs/topics/gateway-events#payload-structure).
 #[cfg_attr(feature = "typesize", derive(typesize::derive::TypeSize))]
-#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, Serialize)]
 #[non_exhaustive]
 #[serde(untagged)]
 pub enum GatewayEvent {
-    Dispatch(u64, Event),
+    Dispatch {
+        seq: u64,
+        // Avoid deserialising straight away to handle errors and get access to `seq`.
+        data: JsonMap,
+        // Used for debugging, if the data cannot be deserialised.
+        original_str: FixedString,
+    },
     Heartbeat(u64),
     Reconnect,
     /// Whether the session can be resumed.
@@ -1078,16 +1084,16 @@ impl<'de> Deserialize<'de> for GatewayEvent {
         let seq = remove_from_map_opt(&mut map, "s")?.flatten();
 
         Ok(match remove_from_map(&mut map, "op")? {
-            Opcode::Dispatch => Self::Dispatch(
-                seq.ok_or_else(|| DeError::missing_field("s"))?,
-                deserialize_val(Value::from(map))?,
-            ),
-            Opcode::Heartbeat => {
-                GatewayEvent::Heartbeat(seq.ok_or_else(|| DeError::missing_field("s"))?)
-            },
-            Opcode::InvalidSession => {
-                GatewayEvent::InvalidateSession(remove_from_map(&mut map, "d")?)
+            Opcode::Dispatch => {
+                Self::Dispatch {
+                    seq: seq.ok_or_else(|| DeError::missing_field("s"))?,
+                    // Filled in in recv_event
+                    original_str: FixedString::new(),
+                    data: map,
+                }
             },
+            Opcode::Heartbeat => Self::Heartbeat(seq.ok_or_else(|| DeError::missing_field("s"))?),
+            Opcode::InvalidSession => Self::InvalidateSession(remove_from_map(&mut map, "d")?),
             Opcode::Hello => {
                 #[derive(Deserialize)]
                 struct HelloPayload {
@@ -1095,10 +1101,10 @@ impl<'de> Deserialize<'de> for GatewayEvent {
                 }
 
                 let inner: HelloPayload = remove_from_map(&mut map, "d")?;
-                GatewayEvent::Hello(inner.heartbeat_interval)
+                Self::Hello(inner.heartbeat_interval)
             },
-            Opcode::Reconnect => GatewayEvent::Reconnect,
-            Opcode::HeartbeatAck => GatewayEvent::HeartbeatAck,
+            Opcode::Reconnect => Self::Reconnect,
+            Opcode::HeartbeatAck => Self::HeartbeatAck,
             _ => return Err(DeError::custom("invalid opcode")),
         })
     }
@@ -1310,4 +1316,31 @@ impl Event {
         let map = serde_json::to_value(self).ok()?;
         Some(map.get("t")?.as_str()?.to_string())
     }
+
+    pub(crate) fn deserialize_and_log(map: JsonMap, original_str: &str) -> Result<Self> {
+        deserialize_val(Value::Object(map))
+            .map_err(|err| log_deserialisation_err(original_str, err))
+    }
+}
+
+fn filter_unknown_variant(json_err_dbg: &str) -> bool {
+    if let Some(msg) = json_err_dbg.strip_prefix("Error(\"unknown variant `") {
+        if let Some((variant_name, _)) = msg.split_once('`') {
+            debug!("Unknown event: {variant_name}");
+            return true;
+        }
+    }
+
+    false
+}
+
+#[cold]
+fn log_deserialisation_err(json_str: &str, err: serde_json::Error) -> Error {
+    let json_err_dbg = format!("{err:?}");
+    if !filter_unknown_variant(&json_err_dbg) {
+        warn!("Err deserializing text: {json_err_dbg}");
+    }
+
+    debug!("Failing text: {json_str}");
+    Error::Json(err)
 }


Test patch: