Add session recovery with automatic playback resume

When the TCP connection to Spotify access points drops due to network
instability, the keep-alive timeout fires after ~80s and invalidates
the session. The old spirc shuts down and a new Session + Spirc is
created. However, three things went wrong:

1. Session ID mismatch: The new Session generated a fresh random UUID,
   so Spotify could not match it to the previous playback session. The
   automatic transfer in handle_connection_id_update failed because
   the cluster session_id did not match the new session_id.

2. Volume reset: ConnectConfig.initial_volume was set once at startup
   and reused on every reconnection. The new SpircTask initialized the
   mixer to this stale value instead of the users current volume,
   causing a jarring volume jump on reconnect.

3. Playback not resuming: The transfer state from Spotify always has
   is_paused=true after a disconnect, since the device went offline.
   handle_transfer used this to set start_playing=false, so the track
   loaded paused even though the user was actively listening.

Fixes:
- Preserve session_id across reconnections for automatic transfer
- Preserve mixer volume on reconnect, only override after first connect
- Track user play intent via shared Arc<AtomicBool> was_playing flag
- Add force_play parameter to handle_transfer for reconnection resume
- Add shutdown reason strings to session.shutdown for debugging
- Fast-fail audio key requests when session is invalid
- Use RwLock<Session> in player for session hot-swap

Author: Arneball

connect/src/spirc.rs

@@ -38,7 +38,7 @@ use protobuf::MessageField;
 use std::{
     future::Future,
     sync::Arc,
-    sync::atomic::{AtomicUsize, Ordering},
+    sync::atomic::{AtomicBool, AtomicUsize, Ordering},
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 use thiserror::Error;
@@ -112,6 +112,9 @@ struct SpircTask {
     update_state: bool,
 
     spirc_id: usize,
+
+    /// shared flag to preserve play state across reconnections
+    was_playing: Arc<AtomicBool>,
 }
 
 static SPIRC_COUNTER: AtomicUsize = AtomicUsize::new(0);
@@ -163,6 +166,7 @@ impl Spirc {
         credentials: Credentials,
         player: Arc<Player>,
         mixer: Arc<dyn Mixer>,
+        was_playing: Arc<AtomicBool>,
     ) -> Result<(Spirc, impl Future<Output = ()>), Error> {
         fn extract_connection_id(msg: Message) -> Result<String, Error> {
             let connection_id = msg
@@ -262,6 +266,7 @@ impl Spirc {
             update_state: false,
 
             spirc_id,
+            was_playing,
         };
 
         let spirc = Spirc { commands: cmd_tx };
@@ -930,7 +935,9 @@ impl SpircTask {
         use protobuf::Message;
 
         match TransferState::parse_from_bytes(&cluster.transfer_data) {
-            Ok(transfer_state) => self.handle_transfer(transfer_state)?,
+            Ok(transfer_state) => {
+                self.handle_transfer(transfer_state, self.was_playing.load(Ordering::Relaxed))?
+            }
             Err(why) => error!("failed to take over control: {why}"),
         }
 
@@ -1067,7 +1074,7 @@ impl SpircTask {
             }
             // modification and update of the connect_state
             Transfer(transfer) => {
-                self.handle_transfer(transfer.data.expect("by condition checked"))?;
+                self.handle_transfer(transfer.data.expect("by condition checked"), false)?;
                 return self.notify().await;
             }
             Play(mut play) => {
@@ -1164,7 +1171,11 @@ impl SpircTask {
         Ok(())
     }
 
-    fn handle_transfer(&mut self, mut transfer: TransferState) -> Result<(), Error> {
+    fn handle_transfer(
+        &mut self,
+        mut transfer: TransferState,
+        force_play: bool,
+    ) -> Result<(), Error> {
         let mut ctx_uri = match transfer.current_session.context.uri {
             None => Err(SpircError::NoUri("transfer context"))?,
             // can apparently happen when a state is transferred and was started with "uris" via the api
@@ -1250,7 +1261,7 @@ impl SpircTask {
             _ => 0,
         };
 
-        let is_playing = !transfer.playback.is_paused();
+        let is_playing = force_play || !transfer.playback.is_paused();
 
         if self.connect_state.current_track(|t| t.is_autoplay()) || autoplay {
             if let Some(ctx_uri) = ctx_uri {
@@ -1533,6 +1544,8 @@ impl SpircTask {
             _ => return,
         }
 
+        self.was_playing.store(true, Ordering::Relaxed);
+
         // Synchronize the volume from the mixer. This is useful on
         // systems that can switch sources from and back to librespot.
         let current_volume = self.mixer.volume();
@@ -1572,6 +1585,7 @@ impl SpircTask {
             }
             _ => (),
         }
+        self.was_playing.store(false, Ordering::Relaxed);
     }
 
     fn handle_seek(&mut self, position_ms: u32) {
@@ -1888,6 +1902,7 @@ impl SpircTask {
         } else {
             self.play_status = SpircPlayStatus::LoadingPause { position_ms };
         }
+        self.was_playing.store(start_playing, Ordering::Relaxed);
         self.connect_state.set_status(&self.play_status);
 
         Ok(())

core/src/audio_key.rs

@@ -20,6 +20,8 @@ pub enum AudioKeyError {
     Packet(u8),
     #[error("sequence {0} not pending")]
     Sequence(u32),
+    #[error("session is invalid")]
+    SessionInvalid,
     #[error("audio key response timeout")]
     Timeout,
 }
@@ -31,6 +33,7 @@ impl From<AudioKeyError> for Error {
             AudioKeyError::Channel => Error::aborted(err),
             AudioKeyError::Sequence(_) => Error::aborted(err),
             AudioKeyError::Packet(_) => Error::unimplemented(err),
+            AudioKeyError::SessionInvalid => Error::aborted(err),
             AudioKeyError::Timeout => Error::aborted(err),
         }
     }
@@ -79,6 +82,12 @@ impl AudioKeyManager {
     }
 
     pub async fn request(&self, track: SpotifyId, file: FileId) -> Result<AudioKey, Error> {
+        // Fast-fail if session is already invalid to avoid waiting for timeout
+        if self.session().is_invalid() {
+            error!("Audio key request rejected: session is invalid");
+            return Err(AudioKeyError::SessionInvalid.into());
+        }
+
         let (tx, rx) = oneshot::channel();
 
         let seq = self.lock(move |inner| {

core/src/session.rs

@@ -276,7 +276,7 @@ impl Session {
                 error!("{e}");
                 if let Some(session) = session_weak.try_upgrade() {
                     if !session.is_invalid() {
-                        session.shutdown();
+                        session.shutdown("packet sender task failed");
                     }
                 }
             }
@@ -635,8 +635,8 @@ impl Session {
         SessionWeak(Arc::downgrade(&self.0))
     }
 
-    pub fn shutdown(&self) {
-        debug!("Shutdown: Invalidating session");
+    pub fn shutdown(&self, reason: &str) {
+        debug!("Shutdown: Invalidating session: {reason}");
         self.0.data.write().expect(SESSION_DATA_POISON_MSG).invalid = true;
         self.mercury().shutdown();
         self.channel().shutdown();
@@ -893,12 +893,12 @@ where
                 }
                 Poll::Ready(None) => {
                     warn!("Connection to server closed.");
-                    session.shutdown();
+                    session.shutdown("connection closed by server");
                     return Poll::Ready(Ok(()));
                 }
                 Poll::Ready(Some(Err(e))) => {
                     error!("Connection to server closed.");
-                    session.shutdown();
+                    session.shutdown("connection error");
                     return Poll::Ready(Err(e));
                 }
                 Poll::Pending => break,
@@ -927,7 +927,7 @@ where
             match this.keep_alive_state {
                 ExpectingPing | ExpectingPongAck => {
                     if !session.is_invalid() {
-                        session.shutdown();
+                        session.shutdown("keep-alive timeout");
                     }
                     // TODO: Optionally reconnect (with cached/last credentials?)
                     return Poll::Ready(Err(io::Error::new(

examples/play_connect.rs

@@ -1,3 +1,5 @@
+use std::sync::{Arc, atomic::AtomicBool};
+
 use librespot::{
     connect::{ConnectConfig, LoadRequest, LoadRequestOptions, Spirc},
     core::{
@@ -59,8 +61,16 @@ async fn main() -> Result<(), Error> {
         move || sink_builder(None, audio_format),
     );
 
-    let (spirc, spirc_task) =
-        Spirc::new(connect_config, session.clone(), credentials, player, mixer).await?;
+    let was_playing = Arc::new(AtomicBool::new(false));
+    let (spirc, spirc_task) = Spirc::new(
+        connect_config,
+        session.clone(),
+        credentials,
+        player,
+        mixer,
+        was_playing,
+    )
+    .await?;
 
     // these calls can be seen as "queued"
     spirc.activate()?;

playback/src/player.rs

@@ -7,11 +7,11 @@ use std::{
     mem,
     pin::Pin,
     process::exit,
-    sync::Mutex,
     sync::{
         Arc,
         atomic::{AtomicUsize, Ordering},
     },
+    sync::{Mutex, RwLock},
     task::{Context, Poll},
     thread,
     time::{Duration, Instant},
@@ -69,7 +69,7 @@ pub enum SinkStatus {
 pub type SinkEventCallback = Box<dyn Fn(SinkStatus) + Send>;
 
 struct PlayerInternal {
-    session: Session,
+    session: Arc<RwLock<Session>>,
     config: PlayerConfig,
     commands: mpsc::UnboundedReceiver<PlayerCommand>,
     load_handles: Arc<Mutex<HashMap<thread::ThreadId, thread::JoinHandle<()>>>>,
@@ -504,7 +504,7 @@ impl Player {
                 create_local_file_lookup(config.local_file_directories.as_slice());
 
             let internal = PlayerInternal {
-                session,
+                session: Arc::new(RwLock::new(session)),
                 config,
                 commands: cmd_rx,
                 load_handles: Arc::new(Mutex::new(HashMap::new())),
@@ -932,7 +932,7 @@ impl PlayerState {
 }
 
 struct PlayerTrackLoader {
-    session: Session,
+    session: Arc<RwLock<Session>>,
     config: PlayerConfig,
     local_file_lookup: Arc<LocalFileLookup>,
 }
@@ -947,9 +947,10 @@ impl PlayerTrackLoader {
         } else if let Some(alternatives) = audio_item.alternatives {
             let Tracks(alternatives_vec) = alternatives; // required to make `into_iter` able to move
 
+            let session = self.session.read().unwrap().clone();
             let alternatives: FuturesUnordered<_> = alternatives_vec
                 .into_iter()
-                .map(|alt_id| AudioItem::get_file(&self.session, alt_id))
+                .map(|alt_id| AudioItem::get_file(&session, alt_id))
                 .collect();
 
             alternatives
@@ -1019,7 +1020,8 @@ impl PlayerTrackLoader {
             }
         };
 
-        let audio_item = match AudioItem::get_file(&self.session, track_uri).await {
+        let session = self.session.read().unwrap().clone();
+        let audio_item = match AudioItem::get_file(&session, track_uri).await {
             Ok(audio) => match self.find_available_alternative(audio).await {
                 Some(audio) => audio,
                 None => {
@@ -1091,7 +1093,8 @@ impl PlayerTrackLoader {
         // This is only a loop to be able to reload the file if an error occurred
         // while opening a cached file.
         loop {
-            let encrypted_file = AudioFile::open(&self.session, file_id, bytes_per_second);
+            let session = self.session.read().unwrap().clone();
+            let encrypted_file = AudioFile::open(&session, file_id, bytes_per_second);
 
             let encrypted_file = match encrypted_file.await {
                 Ok(encrypted_file) => encrypted_file,
@@ -1108,7 +1111,8 @@ impl PlayerTrackLoader {
             // Not all audio files are encrypted. If we can't get a key, try loading the track
             // without decryption. If the file was encrypted after all, the decoder will fail
             // parsing and bail out, so we should be safe from outputting ear-piercing noise.
-            let key = match self.session.audio_key().request(track_id, file_id).await {
+            let session = self.session.read().unwrap().clone();
+            let key = match session.audio_key().request(track_id, file_id).await {
                 Ok(key) => Some(key),
                 Err(e) => {
                     warn!("Unable to load key, continuing without decryption: {e}");
@@ -1176,7 +1180,7 @@ impl PlayerTrackLoader {
                 Err(e) if is_cached => {
                     warn!("Unable to read cached audio file: {e}. Trying to download it.");
 
-                    match self.session.cache() {
+                    match self.session.read().unwrap().clone().cache() {
                         Some(cache) => {
                             if cache.remove_file(file_id).is_err() {
                                 error!("Error removing file from cache");
@@ -2316,7 +2320,7 @@ impl PlayerInternal {
 
             PlayerCommand::Stop => self.handle_player_stop(),
 
-            PlayerCommand::SetSession(session) => self.session = session,
+            PlayerCommand::SetSession(session) => *self.session.write().unwrap() = session,
 
             PlayerCommand::AddEventSender(sender) => self.event_senders.push(sender),
 

src/main.rs

@@ -31,6 +31,10 @@ use std::{
     pin::Pin,
     process::exit,
     str::FromStr,
+    sync::{
+        Arc,
+        atomic::AtomicBool,
+    },
     time::{Duration, Instant},
 };
 use sysinfo::{ProcessesToUpdate, System};
@@ -1895,9 +1899,11 @@ async fn main() {
     let mut last_credentials = None;
     let mut spirc: Option<Spirc> = None;
     let mut spirc_task: Option<Pin<_>> = None;
+    let was_playing = Arc::new(AtomicBool::new(false));
     let mut auto_connect_times: Vec<Instant> = vec![];
     let mut discovery = None;
     let mut connecting = false;
+    let mut has_been_connected = false;
     let mut _event_handler: Option<EventHandler> = None;
 
     let mut session = Session::new(setup.session_config.clone(), setup.cache.clone());
@@ -2030,7 +2036,7 @@ async fn main() {
                             tokio::spawn(spirc_task);
                         }
                         if !session.is_invalid() {
-                            session.shutdown();
+                            session.shutdown("new credentials received");
                         }
 
                         connecting = true;
@@ -2043,17 +2049,26 @@ async fn main() {
             },
             _ = async {}, if connecting && last_credentials.is_some() => {
                 if session.is_invalid() {
+                    let old_session_id = session.session_id();
                     session = Session::new(setup.session_config.clone(), setup.cache.clone());
+                    if !old_session_id.is_empty() {
+                        session.set_session_id(&old_session_id);
+                        info!("Preserved session_id across reconnection: {old_session_id}");
+                    }
                     player.set_session(session.clone());
                 }
 
-                let connect_config = setup.connect_config.clone();
+                let mut connect_config = setup.connect_config.clone();
+                if has_been_connected {
+                    connect_config.initial_volume = mixer.volume();
+                }
 
                 let (spirc_, spirc_task_) = match Spirc::new(connect_config,
                                                                 session.clone(),
                                                                 last_credentials.clone().unwrap_or_default(),
                                                                 player.clone(),
-                                                                mixer.clone()).await {
+                                                                mixer.clone(),
+                                                                was_playing.clone()).await {
                     Ok((spirc_, spirc_task_)) => (spirc_, spirc_task_),
                     Err(e) => {
                         error!("could not initialize spirc: {e}");
@@ -2064,6 +2079,7 @@ async fn main() {
                 spirc_task = Some(Box::pin(spirc_task_));
 
                 connecting = false;
+                has_been_connected = true;
             },
             _ = async {
                 if let Some(task) = spirc_task.as_mut() {
@@ -2082,7 +2098,7 @@ async fn main() {
                 if last_credentials.is_some() && !reconnect_exceeds_rate_limit() {
                     auto_connect_times.push(Instant::now());
                     if !session.is_invalid() {
-                        session.shutdown();
+                        session.shutdown("spirc shut down unexpectedly");
                     }
                     connecting = true;
                 } else {