Fix shards attempting to re-identify on their own

Fix shards by taking away their responsibility to re-identify, instead
shutting down shard runners and going through the shard queuer to
restart a shard runner and its associated shard.

This fixes the case where a running shard's session invalidates and
re-IDENTIFYs within 5 seconds before queued shard starts, causing a
cascading failure of sessions for new shards.

3 files changed, 51 insertions, 35 deletions

diff --git a/src/client/bridge/gateway/shard_runner.rs b/src/client/bridge/gateway/shard_runner.rs
index a682693..43eebb8 100644
--- a/src/client/bridge/gateway/shard_runner.rs
+++ b/src/client/bridge/gateway/shard_runner.rs
@@ -1,4 +1,4 @@
-use gateway::{Shard, ShardAction};
+use gateway::{ReconnectType, Shard, ShardAction};
 use internal::prelude::*;
 use internal::ws_impl::ReceiverExt;
 use model::event::{Event, GatewayEvent};
@@ -143,8 +143,14 @@ impl<H: EventHandler + Send + Sync + 'static> ShardRunner<H> {
 
             let (event, action, successful) = self.recv_event();
 
-            if let Some(ref action) = action {
-                let _ = self.action(action);
+            match action {
+                Some(ShardAction::Reconnect(ReconnectType::Reidentify)) => {
+                    return self.request_restart()
+                },
+                Some(other) => {
+                    let _ = self.action(&other);
+                },
+                None => {},
             }
 
             if let Some(event) = event {
@@ -173,11 +179,14 @@ impl<H: EventHandler + Send + Sync + 'static> ShardRunner<H> {
     /// Returns
     fn action(&mut self, action: &ShardAction) -> Result<()> {
         match *action {
-            ShardAction::Autoreconnect => self.shard.autoreconnect(),
+            ShardAction::Reconnect(ReconnectType::Reidentify) => {
+                self.request_restart()
+            },
+            ShardAction::Reconnect(ReconnectType::Resume) => {
+                self.shard.resume()
+            },
             ShardAction::Heartbeat => self.shard.heartbeat(),
             ShardAction::Identify => self.shard.identify(),
-            ShardAction::Reconnect => self.shard.reconnect(),
-            ShardAction::Resume => self.shard.resume(),
         }
     }
 
@@ -365,8 +374,15 @@ impl<H: EventHandler + Send + Sync + 'static> ShardRunner<H> {
 
                 debug!("Attempting to auto-reconnect");
 
-                if let Err(why) = self.shard.autoreconnect() {
-                    error!("Failed to auto-reconnect: {:?}", why);
+                match self.shard.reconnection_type() {
+                    ReconnectType::Reidentify => return (None, None, false),
+                    ReconnectType::Resume => {
+                        if let Err(why) = self.shard.resume() {
+                            warn!("Failed to resume: {:?}", why);
+
+                            return (None, None, false);
+                        }
+                    },
                 }
 
                 return (None, None, true);
diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs
index 7e4e2b0..a6de656 100644
--- a/src/gateway/mod.rs
+++ b/src/gateway/mod.rs
@@ -148,9 +148,15 @@ impl ConnectionStage {
 }
 
 pub enum ShardAction {
-    Autoreconnect,
     Heartbeat,
     Identify,
-    Reconnect,
+    Reconnect(ReconnectType),
+}
+
+/// The type of reconnection that should be performed.
+pub enum ReconnectType {
+    /// Indicator that a new connection should be made by sending an IDENTIFY.
+    Reidentify,
+    /// Indicator that a new connection should be made by sending a RESUME.
     Resume,
 }
diff --git a/src/gateway/shard.rs b/src/gateway/shard.rs
index c00affc..75dc147 100644
--- a/src/gateway/shard.rs
+++ b/src/gateway/shard.rs
@@ -12,6 +12,7 @@ use super::{
     CurrentPresence,
     ShardAction,
     GatewayError,
+    ReconnectType,
     WsClient,
     WebSocketGatewayClientExt,
 };
@@ -450,7 +451,7 @@ impl Shard {
                             self.shard_info
                         );
 
-                        return Ok(Some(ShardAction::Autoreconnect));
+                        return Ok(Some(ShardAction::Reconnect(self.reconnection_type())));
                     }
                 }
 
@@ -483,7 +484,7 @@ impl Shard {
                     debug!("[Shard {:?}] Received late Hello; autoreconnecting",
                            self.shard_info);
 
-                    ShardAction::Autoreconnect
+                    ShardAction::Reconnect(self.reconnection_type())
                 }))
             },
             Ok(GatewayEvent::InvalidateSession(resumable)) => {
@@ -493,12 +494,14 @@ impl Shard {
                 );
 
                 Ok(Some(if resumable {
-                    ShardAction::Resume
+                    ShardAction::Reconnect(ReconnectType::Resume)
                 } else {
-                    ShardAction::Reconnect
+                    ShardAction::Reconnect(ReconnectType::Reidentify)
                 }))
             },
-            Ok(GatewayEvent::Reconnect) => Ok(Some(ShardAction::Reconnect)),
+            Ok(GatewayEvent::Reconnect) => {
+                Ok(Some(ShardAction::Reconnect(ReconnectType::Reidentify)))
+            },
             Err(Error::Gateway(GatewayError::Closed(ref data))) => {
                 let num = data.as_ref().map(|d| d.status_code);
                 let clean = num == Some(1000);
@@ -572,9 +575,9 @@ impl Shard {
                 }).unwrap_or(true);
 
                 Ok(Some(if resume {
-                    ShardAction::Resume
+                    ShardAction::Reconnect(ReconnectType::Resume)
                 } else {
-                    ShardAction::Reconnect
+                    ShardAction::Reconnect(ReconnectType::Reidentify)
                 }))
             },
             Err(Error::WebSocket(ref why)) => {
@@ -590,7 +593,7 @@ impl Shard {
                 info!("[Shard {:?}] Will attempt to auto-reconnect",
                       self.shard_info);
 
-                Ok(Some(ShardAction::Autoreconnect))
+                Ok(Some(ShardAction::Reconnect(self.reconnection_type())))
             },
             _ => Ok(None),
         }
@@ -700,29 +703,20 @@ impl Shard {
     ///
     /// [`ConnectionStage::Connecting`]: ../../../gateway/enum.ConnectionStage.html#variant.Connecting
     /// [`session_id`]: ../../../gateway/struct.Shard.html#method.session_id
-    pub fn autoreconnect(&mut self) -> Result<()> {
+    pub fn should_reconnect(&mut self) -> Option<ReconnectType> {
         if self.stage == ConnectionStage::Connecting {
-            return Ok(());
+            return None;
         }
 
-        if self.session_id().is_some() {
-            debug!(
-                "[Shard {:?}] Autoreconnector choosing to resume",
-                self.shard_info,
-            );
+        Some(self.reconnection_type())
+    }
 
-            self.resume()?;
+    pub fn reconnection_type(&self) -> ReconnectType {
+        if self.session_id().is_some() {
+            ReconnectType::Resume
         } else {
-            debug!(
-                "[Shard {:?}] Autoreconnector choosing to reconnect",
-                self.shard_info,
-            );
-
-            self.reconnect()?;
+            ReconnectType::Reidentify
         }
-        self.shutdown = true;
-
-        Ok(())
     }
 
     /// Requests that one or multiple [`Guild`]s be chunked.