music: keep preparation out of the guild lock and requeue failed connects

1 files changed, 119 insertions, 99 deletions

diff --git a/src/music.rs b/src/music.rs
index 0b907d3..f1c0ad9 100644
--- a/src/music.rs
+++ b/src/music.rs
@@ -104,13 +104,17 @@ struct GuildMusic {
     voice: serenity::ChannelId,
     /// Text channel where auto-advance announcements are posted.
     announce: Option<serenity::ChannelId>,
+    /// Bumped by interrupts (`now`, skip, stop). In-flight preparation for an
+    /// older generation is discarded instead of starting over the new track.
+    generation: u64,
 }
 
 /// Shared, per-guild music state. Stored in [`crate::Data`].
 pub struct MusicState {
     guilds: Mutex<HashMap<u64, GuildMusic>>,
-    /// One transition lock per guild. Held across the whole stop/prepare/play
-    /// sequence so concurrent `/music` commands cannot interleave.
+    /// One state lock per guild. Held only for state mutations; downloads,
+    /// compression, and voice connects happen outside it so one slow track
+    /// cannot stall every other `/music` command in the guild.
     locks: Mutex<HashMap<u64, Arc<Mutex<()>>>>,
     voice: VoiceManager,
     http: Arc<serenity::Http>,
@@ -150,55 +154,54 @@ impl MusicState {
         position: QueuePosition,
     ) -> Result<EnqueueOutcome> {
         let lock = self.guild_lock(guild).await;
-        let _transition = lock.lock().await;
-
-        let was_playing;
-        {
+        let (was_playing, queued_position, interrupted) = {
+            let _transition = lock.lock().await;
             let mut map = self.guilds.lock().await;
             let gm = map.entry(guild).or_insert_with(|| GuildMusic {
                 current: None,
                 queue: VecDeque::new(),
                 voice: voice_channel,
                 announce,
+                generation: 0,
             });
             gm.voice = voice_channel;
             gm.announce = announce;
-            was_playing = gm.current.is_some();
+            let was_playing = gm.current.is_some();
 
             match position {
-                QueuePosition::Now | QueuePosition::Next => gm.queue.push_front(request),
+                QueuePosition::Now => gm.queue.push_front(request),
+                QueuePosition::Next => gm.queue.push_front(request),
                 QueuePosition::Last => gm.queue.push_back(request),
             }
-        }
+            let queued_position = match position {
+                QueuePosition::Next => 1,
+                QueuePosition::Last => gm.queue.len(),
+                QueuePosition::Now => 1,
+            };
 
-        if was_playing && position != QueuePosition::Now {
-            let position = {
-                let map = self.guilds.lock().await;
-                let len = map.get(&guild).map(|gm| gm.queue.len()).unwrap_or(1);
-                if position == QueuePosition::Next {
-                    1
-                } else {
-                    len
-                }
+            let interrupted = if was_playing && position == QueuePosition::Now {
+                gm.generation += 1;
+                gm.current.take()
+            } else {
+                None
             };
-            return Ok(EnqueueOutcome::Queued { position });
+            (was_playing, queued_position, interrupted)
+        };
+
+        if let Some(current) = interrupted {
+            let _ = current.handle.stop();
         }
 
-        if was_playing {
-            // `Now`: drop the current track. The new request is at the front of
-            // the queue; play_next_locked will pick it up.
-            let current = {
-                let mut map = self.guilds.lock().await;
-                map.get_mut(&guild).and_then(|gm| gm.current.take())
-            };
-            if let Some(current) = current {
-                let _ = current.handle.stop();
-            }
+        if was_playing && position != QueuePosition::Now {
+            return Ok(EnqueueOutcome::Queued {
+                position: queued_position,
+            });
         }
 
         // The command layer posts the "now playing" message for direct
-        // enqueues, so don't announce here as well.
-        let played = self.play_next_locked(guild, false).await?;
+        // enqueues, so don't announce here as well. Preparation happens
+        // outside the guild lock.
+        let played = self.play_next(guild, false).await?;
         Ok(if played {
             EnqueueOutcome::Started
         } else {
@@ -210,35 +213,30 @@ impl MusicState {
     /// Returns `true` when a track started, `false` when the queue was empty.
     pub async fn skip(self: &Arc<Self>, guild: u64) -> Result<bool> {
         let lock = self.guild_lock(guild).await;
-        let _transition = lock.lock().await;
-
-        let current = {
+        let (current, has_queue) = {
+            let _transition = lock.lock().await;
             let mut map = self.guilds.lock().await;
-            map.get_mut(&guild).and_then(|gm| gm.current.take())
+            let Some(gm) = map.get_mut(&guild) else {
+                return Ok(false);
+            };
+            gm.generation += 1;
+            (gm.current.take(), !gm.queue.is_empty())
         };
         if let Some(current) = current {
             let _ = current.handle.stop();
         }
-
-        let has_queue = {
-            let map = self.guilds.lock().await;
-            map.get(&guild)
-                .map(|gm| !gm.queue.is_empty())
-                .unwrap_or(false)
-        };
         if !has_queue {
             return Ok(false);
         }
 
-        self.play_next_locked(guild, true).await
+        self.play_next(guild, true).await
     }
 
     /// Stop playback and drop the queue (does not disconnect from voice).
     pub async fn stop(self: &Arc<Self>, guild: u64) {
         let lock = self.guild_lock(guild).await;
-        let _transition = lock.lock().await;
-
         let current = {
+            let _transition = lock.lock().await;
             let mut map = self.guilds.lock().await;
             map.remove(&guild).and_then(|gm| gm.current)
         };
@@ -258,24 +256,17 @@ impl MusicState {
         request: TrackRequest,
     ) -> Result<()> {
         let lock = self.guild_lock(guild).await;
-        let _transition = lock.lock().await;
-
         let current = {
-            let mut map = self.guilds.lock().await;
-            map.get_mut(&guild).and_then(|gm| gm.current.take())
-        };
-        if let Some(current) = current {
-            let _ = current.handle.stop();
-        }
-
-        {
+            let _transition = lock.lock().await;
             let mut map = self.guilds.lock().await;
             let gm = map.entry(guild).or_insert_with(|| GuildMusic {
                 current: None,
                 queue: VecDeque::new(),
                 voice: voice_channel,
                 announce,
+                generation: 0,
             });
+            gm.generation += 1;
             gm.voice = voice_channel;
             // Keep the last text channel when the caller has nowhere to
             // announce (cron's channel is a voice channel).
@@ -283,9 +274,13 @@ impl MusicState {
                 gm.announce = announce;
             }
             gm.queue.push_front(request);
+            gm.current.take()
+        };
+        if let Some(current) = current {
+            let _ = current.handle.stop();
         }
 
-        if self.play_next_locked(guild, true).await? {
+        if self.play_next(guild, true).await? {
             Ok(())
         } else {
             Err(Error::Custom(lang::L_MUSIC_QUEUE_EMPTY.into()))
@@ -314,22 +309,28 @@ impl MusicState {
         }
     }
 
-    /// Pop and play the next playable track. Caller must hold the guild
-    /// transition lock.
-    async fn play_next_locked(self: &Arc<Self>, guild: u64, announce: bool) -> Result<bool> {
+    /// Pop and play the next playable track.
+    ///
+    /// Input preparation and the voice connect happen outside the guild state
+    /// lock, so a slow download cannot block other commands. A generation
+    /// token makes an in-flight track abort if it was interrupted meanwhile.
+    async fn play_next(self: &Arc<Self>, guild: u64, announce: bool) -> Result<bool> {
         let mut last_error: Option<Error> = None;
         loop {
-            let request = {
+            let (request, voice_channel, announce_channel, generation) = {
+                let lock = self.guild_lock(guild).await;
+                let _transition = lock.lock().await;
                 let mut map = self.guilds.lock().await;
-                match map.get_mut(&guild).and_then(|gm| gm.queue.pop_front()) {
-                    Some(request) => request,
-                    None => {
-                        return match last_error {
-                            Some(e) => Err(e),
-                            None => Ok(false),
-                        };
-                    }
-                }
+                let Some(gm) = map.get_mut(&guild) else {
+                    return Ok(false);
+                };
+                let Some(request) = gm.queue.pop_front() else {
+                    return match last_error {
+                        Some(e) => Err(e),
+                        None => Ok(false),
+                    };
+                };
+                (request, gm.voice, gm.announce, gm.generation)
             };
 
             let input = match build_input(&request, &self.reqwest).await {
@@ -345,38 +346,58 @@ impl MusicState {
                 }
             };
 
-            let (voice_channel, announce_channel) = {
-                let map = self.guilds.lock().await;
-                match map.get(&guild) {
-                    Some(gm) => (gm.voice, gm.announce),
-                    None => return Ok(false),
-                }
-            };
-
             let call = match self.voice.get_call(serenity::GuildId::new(guild)) {
                 Some(call) => call,
-                None => {
-                    self.voice
-                        .connect(serenity::GuildId::new(guild), voice_channel)
-                        .await?
-                }
+                None => match self
+                    .voice
+                    .connect(serenity::GuildId::new(guild), voice_channel)
+                    .await
+                {
+                    Ok(call) => call,
+                    Err(e) => {
+                        // Put the track back unless an interrupt superseded it.
+                        let lock = self.guild_lock(guild).await;
+                        let _transition = lock.lock().await;
+                        let mut map = self.guilds.lock().await;
+                        if let Some(gm) = map.get_mut(&guild)
+                            && gm.generation == generation
+                        {
+                            gm.queue.push_front(request);
+                        }
+                        return Err(e);
+                    }
+                },
             };
 
-            let handle = {
-                let mut call = call.lock().await;
-                call.play_input(input)
-            };
-            let token = Uuid::new_v4();
-            {
+            let (handle, token) = {
+                let lock = self.guild_lock(guild).await;
+                let _transition = lock.lock().await;
                 let mut map = self.guilds.lock().await;
-                if let Some(gm) = map.get_mut(&guild) {
-                    gm.current = Some(CurrentTrack {
-                        token,
-                        handle: handle.clone(),
-                        request: request.clone(),
-                    });
+                let Some(gm) = map.get_mut(&guild) else {
+                    return Ok(false);
+                };
+                if gm.generation != generation {
+                    // Interrupted while preparing; the replacement request is
+                    // already at the front of the queue.
+                    continue;
                 }
-            }
+                if gm.current.is_some() {
+                    // Another preparation won the race; let it handle this.
+                    gm.queue.push_front(request);
+                    return Ok(false);
+                }
+                let handle = {
+                    let mut call = call.lock().await;
+                    call.play_input(input)
+                };
+                let token = Uuid::new_v4();
+                gm.current = Some(CurrentTrack {
+                    token,
+                    handle: handle.clone(),
+                    request: request.clone(),
+                });
+                (handle, token)
+            };
 
             let _ = handle.add_event(
                 songbird::Event::Track(songbird::TrackEvent::End),
@@ -405,10 +426,9 @@ impl MusicState {
     /// Called from the `End` event of a track. Only advances if the track that
     /// ended is still the current one.
     async fn on_track_end(self: Arc<Self>, guild: u64, token: Uuid) {
-        let lock = self.guild_lock(guild).await;
-        let _transition = lock.lock().await;
-
         let is_current = {
+            let lock = self.guild_lock(guild).await;
+            let _transition = lock.lock().await;
             let mut map = self.guilds.lock().await;
             match map.get_mut(&guild) {
                 Some(gm) => match &gm.current {
@@ -423,7 +443,7 @@ impl MusicState {
         };
 
         if is_current {
-            if let Err(e) = self.play_next_locked(guild, true).await {
+            if let Err(e) = self.play_next(guild, true).await {
                 error!(guild, error = %e, "failed to advance music queue");
             }
         }