From 0d2f471e9a868518899be4d51cb3029f01079f2e Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 4 Nov 2026 01:34:42 +0100 Subject: music: keep preparation out of the guild lock and requeue failed connects --- src/music.rs | 218 ++++++++++++++++++++++++++++++++--------------------------- 1 file changed, 119 insertions(+), 99 deletions(-) (limited to 'src') 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, + /// 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>, - /// 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>>>, voice: VoiceManager, http: Arc, @@ -150,55 +154,54 @@ impl MusicState { position: QueuePosition, ) -> Result { 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, guild: u64) -> Result { 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, 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, guild: u64, announce: bool) -> Result { + /// 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, guild: u64, announce: bool) -> Result { let mut last_error: Option = 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, 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"); } } -- cgit v1.2.3