feat(gateway)!: swallow WebSocket errors

Reading received messages on shutdown requires verbosely matching on WebSocket errors to prevent the shard from automatically reconnecting. While potentially useful for debugging, my experience is that WebSocket errors are generally just noise, and as such they are replaced with close messages with a [status code of 1006](https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1) indicating abnormal closure. This has the secondary benefit of removing confusion about transport errors being somehow actionable (any internet connection may naturally error at any time for reasons outside of your control).
twilight-rs · Jul 6, 2024 · 470e221 · 470e221
1 parent 6a44028
commit 470e221
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 48 deletions.
diff --git a/twilight-gateway/README.md b/twilight-gateway/README.md
@@ -87,12 +87,6 @@ async fn runner(mut shard: Shard) {
         let event = match item {
             Ok(Event::GatewayClose(_)) if SHUTDOWN.load(Ordering::Relaxed) => break,
             Ok(event) => event,
-            Err(source)
-                if SHUTDOWN.load(Ordering::Relaxed)
-                    && matches!(source.kind(), ReceiveMessageErrorType::WebSocket) =>
-            {
-                break
-            }
             Err(source) => {
                 tracing::warn!(?source, "error receiving event");
 

diff --git a/twilight-gateway/src/error.rs b/twilight-gateway/src/error.rs
@@ -175,14 +175,6 @@ impl ReceiveMessageError {
             source: Some(Box::new(source)),
         }
     }
-
-    /// Shortcut to create a new error for a websocket error.
-    pub(crate) fn from_websocket(source: tokio_websockets::Error) -> Self {
-        Self {
-            kind: ReceiveMessageErrorType::WebSocket,
-            source: Some(Box::new(source)),
-        }
-    }
 }
 
 impl Display for ReceiveMessageError {
@@ -197,7 +189,6 @@ impl Display for ReceiveMessageError {
                 f.write_str(event)
             }
             ReceiveMessageErrorType::Reconnect => f.write_str("failed to reconnect to the gateway"),
-            ReceiveMessageErrorType::WebSocket => f.write_str("websocket connection error"),
         }
     }
 }
@@ -228,8 +219,6 @@ pub enum ReceiveMessageErrorType {
     },
     /// Shard failed to reconnect to the gateway.
     Reconnect,
-    /// WebSocket connection error.
-    WebSocket,
 }
 
 #[cfg(test)]
@@ -243,7 +232,7 @@ mod tests {
 
     #[test]
     fn receive_message_error_display() {
-        let messages: [(ReceiveMessageErrorType, &str); 4] = [
+        let messages: [(ReceiveMessageErrorType, &str); 3] = [
             (
                 ReceiveMessageErrorType::Compression,
                 "binary message could not be decompressed",
@@ -258,7 +247,6 @@ mod tests {
                 ReceiveMessageErrorType::Reconnect,
                 "failed to reconnect to the gateway",
             ),
-            (ReceiveMessageErrorType::WebSocket, "websocket connection error"),
         ];
 
         for (kind, message) in messages {

diff --git a/twilight-gateway/src/message.rs b/twilight-gateway/src/message.rs
@@ -24,6 +24,9 @@ pub enum Message {
 }
 
 impl Message {
+    /// Close message indicating the connection was closed abnormally.
+    pub(crate) const ABNORMAL_CLOSE: Self = Self::Close(Some(CloseFrame::new(1006, "")));
+
     /// Whether the message is a close message.
     pub const fn is_close(&self) -> bool {
         matches!(self, Self::Close(_))

diff --git a/twilight-gateway/src/shard.rs b/twilight-gateway/src/shard.rs
@@ -437,8 +437,7 @@ impl<Q> Shard<Q> {
     /// continue showing the bot as online until its presence times out.
     ///
     /// To read all remaining messages, continue calling [`poll_next`] until it
-    /// returns [`Message::Close`] or a [`ReceiveMessageErrorType::WebSocket`]
-    /// error type.
+    /// returns [`Message::Close`].
     ///
     /// # Example
     ///
@@ -457,7 +456,6 @@ impl<Q> Shard<Q> {
     ///     match item {
     ///         Ok(Message::Close(_)) => break,
     ///         Ok(Message::Text(_)) => unimplemented!(),
-    ///         Err(source) if matches!(source.kind(), ReceiveMessageErrorType::WebSocket) => break,
     ///         Err(source) => unimplemented!(),
     ///     }
     /// }
@@ -550,19 +548,16 @@ impl<Q> Shard<Q> {
     }
 
     /// Send and flush the pending message.
-    fn poll_flush_pending(
-        &mut self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<(), ReceiveMessageError>> {
+    fn poll_flush_pending(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), WebsocketError>> {
         if self.pending.is_none() {
             return Poll::Ready(Ok(()));
         }
 
-        ready!(Pin::new(self.connection.as_mut().unwrap()).poll_ready(cx)).map_err(|source| {
+        if let Err(e) = ready!(Pin::new(self.connection.as_mut().unwrap()).poll_ready(cx)) {
             self.disconnect(CloseInitiator::Transport);
             self.connection = None;
-            ReceiveMessageError::from_websocket(source)
-        })?;
+            return Poll::Ready(Err(e));
+        }
 
         let pending = self.pending.as_mut().unwrap();
 
@@ -574,18 +569,17 @@ impl<Q> Shard<Q> {
             }
 
             let ws_message = pending.gateway_event.take().unwrap().into_websocket_msg();
-            if let Err(source) = Pin::new(self.connection.as_mut().unwrap()).start_send(ws_message)
-            {
+            if let Err(e) = Pin::new(self.connection.as_mut().unwrap()).start_send(ws_message) {
                 self.disconnect(CloseInitiator::Transport);
                 self.connection = None;
-                return Poll::Ready(Err(ReceiveMessageError::from_websocket(source)));
+                return Poll::Ready(Err(e));
             }
         }
 
-        if let Err(source) = ready!(Pin::new(self.connection.as_mut().unwrap()).poll_flush(cx)) {
+        if let Err(e) = ready!(Pin::new(self.connection.as_mut().unwrap()).poll_flush(cx)) {
             self.disconnect(CloseInitiator::Transport);
             self.connection = None;
-            return Poll::Ready(Err(ReceiveMessageError::from_websocket(source)));
+            return Poll::Ready(Err(e));
         }
 
         if pending.is_heartbeat {
@@ -798,7 +792,9 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                 _ => {}
             }
 
-            ready!(self.poll_flush_pending(cx))?;
+            if ready!(self.poll_flush_pending(cx)).is_err() {
+                return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
+            }
 
             if !self.state.is_disconnected() {
                 if let Poll::Ready(frame) = self.user_channel.close_rx.poll_recv(cx) {
@@ -807,7 +803,9 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                     tracing::debug!("sending close frame from user channel");
                     self.disconnect(CloseInitiator::Shard(frame));
 
-                    ready!(self.poll_flush_pending(cx))?;
+                    if ready!(self.poll_flush_pending(cx)).is_err() {
+                        return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
+                    }
                 }
             }
 
@@ -834,7 +832,9 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                     self.heartbeat_interval_event = false;
                 }
 
-                ready!(self.poll_flush_pending(cx))?;
+                if ready!(self.poll_flush_pending(cx)).is_err() {
+                    return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
+                }
             }
 
             let not_ratelimited = self
@@ -873,7 +873,9 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                     );
                     self.identify_rx = None;
 
-                    ready!(self.poll_flush_pending(cx))?;
+                    if ready!(self.poll_flush_pending(cx)).is_err() {
+                        return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
+                    }
                 }
             }
 
@@ -884,7 +886,9 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                     tracing::debug!("sending command from user channel");
                     self.pending = Pending::text(command, false);
 
-                    ready!(self.poll_flush_pending(cx))?;
+                    if ready!(self.poll_flush_pending(cx)).is_err() {
+                        return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
+                    }
                 }
             }
 
@@ -921,24 +925,18 @@ impl<Q: Queue + Unpin> Stream for Shard<Q> {
                 {
                     continue
                 }
-                Some(Err(source)) => {
+                Some(Err(_)) => {
                     self.disconnect(CloseInitiator::Transport);
-
-                    return Poll::Ready(Some(Err(ReceiveMessageError {
-                        kind: ReceiveMessageErrorType::WebSocket,
-                        source: Some(Box::new(source)),
-                    })));
+                    return Poll::Ready(Some(Ok(Message::ABNORMAL_CLOSE)));
                 }
                 None => {
-                    let res = ready!(Pin::new(self.connection.as_mut().unwrap()).poll_close(cx));
+                    _ = ready!(Pin::new(self.connection.as_mut().unwrap()).poll_close(cx));
                     tracing::debug!("gateway WebSocket connection closed");
                     // Unclean closure.
                     if !self.state.is_disconnected() {
                         self.disconnect(CloseInitiator::Transport);
                     }
                     self.connection = None;
-
-                    res.map_err(ReceiveMessageError::from_websocket)?;
                 }
             }
         };