refactor: remove parking_lot dependency and refine feature selections (#1543)

- refine dependency features and versions in Cargo.toml files
- switch from parking_lot to std sync primitives
- remove dashmap dependency and use DefaultKeyedStateStore
- update crates

Replace parking_lot with std::sync::{Mutex, RwLock, Condvar} throughout the
codebase. Update dependencies and code to use poisoning-aware locks, adding
explicit panic messages where necessary. Update governor to use DashMapStateStore
for rate limiting.

Author: roderickvd

Cargo.lock

@@ -179,7 +179,6 @@ tokio = { version = "1", features = [
     "macros",
     "signal",
     "sync",
-    "parking_lot",
     "process",
 ] }
 url = "2.2"

Cargo.toml

@@ -23,12 +23,11 @@ librespot-core = { version = "0.7.1", path = "../core", default-features = false
 aes = "0.8"
 bytes = "1"
 ctr = "0.9"
-futures-util = "0.3"
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
 http-body-util = "0.1"
 hyper = { version = "1.6", features = ["http1", "http2"] }
 hyper-util = { version = "0.1", features = ["client", "http2"] }
 log = "0.4"
-parking_lot = { version = "0.12", features = ["deadlock_detection"] }
 tempfile = "3"
 thiserror = "2"
-tokio = { version = "1", features = ["macros", "parking_lot", "sync"] }
+tokio = { version = "1", features = ["macros", "sync"] }

audio/Cargo.toml

@@ -8,13 +8,14 @@ use std::{
         Arc, OnceLock,
         atomic::{AtomicBool, AtomicUsize, Ordering},
     },
+    sync::{Condvar, Mutex},
     time::Duration,
 };
 
 use futures_util::{StreamExt, TryFutureExt, future::IntoStream};
 use hyper::{Response, StatusCode, body::Incoming, header::CONTENT_RANGE};
 use hyper_util::client::legacy::ResponseFuture;
-use parking_lot::{Condvar, Mutex};
+
 use tempfile::NamedTempFile;
 use thiserror::Error;
 use tokio::sync::{Semaphore, mpsc, oneshot};
@@ -27,6 +28,8 @@ use crate::range_set::{Range, RangeSet};
 
 pub type AudioFileResult = Result<(), librespot_core::Error>;
 
+const DOWNLOAD_STATUS_POISON_MSG: &str = "audio download status mutex should not be poisoned";
+
 #[derive(Error, Debug)]
 pub enum AudioFileError {
     #[error("other end of channel disconnected")]
@@ -163,7 +166,10 @@ impl StreamLoaderController {
 
     pub fn range_available(&self, range: Range) -> bool {
         if let Some(ref shared) = self.stream_shared {
-            let download_status = shared.download_status.lock();
+            let download_status = shared
+                .download_status
+                .lock()
+                .expect(DOWNLOAD_STATUS_POISON_MSG);
 
             range.length
                 <= download_status
@@ -214,19 +220,24 @@ impl StreamLoaderController {
         self.fetch(range);
 
         if let Some(ref shared) = self.stream_shared {
-            let mut download_status = shared.download_status.lock();
+            let mut download_status = shared
+                .download_status
+                .lock()
+                .expect(DOWNLOAD_STATUS_POISON_MSG);
             let download_timeout = AudioFetchParams::get().download_timeout;
 
             while range.length
                 > download_status
                     .downloaded
                     .contained_length_from_value(range.start)
             {
-                if shared
+                let (new_download_status, wait_result) = shared
                     .cond
-                    .wait_for(&mut download_status, download_timeout)
-                    .timed_out()
-                {
+                    .wait_timeout(download_status, download_timeout)
+                    .expect(DOWNLOAD_STATUS_POISON_MSG);
+
+                download_status = new_download_status;
+                if wait_result.timed_out() {
                     return Err(AudioFileError::WaitTimeout.into());
                 }
 
@@ -558,7 +569,11 @@ impl Read for AudioFileStreaming {
         let mut ranges_to_request = RangeSet::new();
         ranges_to_request.add_range(&Range::new(offset, length_to_request));
 
-        let mut download_status = self.shared.download_status.lock();
+        let mut download_status = self
+            .shared
+            .download_status
+            .lock()
+            .expect(DOWNLOAD_STATUS_POISON_MSG);
 
         ranges_to_request.subtract_range_set(&download_status.downloaded);
         ranges_to_request.subtract_range_set(&download_status.requested);
@@ -571,12 +586,14 @@ impl Read for AudioFileStreaming {
 
         let download_timeout = AudioFetchParams::get().download_timeout;
         while !download_status.downloaded.contains(offset) {
-            if self
+            let (new_download_status, wait_result) = self
                 .shared
                 .cond
-                .wait_for(&mut download_status, download_timeout)
-                .timed_out()
-            {
+                .wait_timeout(download_status, download_timeout)
+                .expect(DOWNLOAD_STATUS_POISON_MSG);
+
+            download_status = new_download_status;
+            if wait_result.timed_out() {
                 return Err(io::Error::new(
                     io::ErrorKind::TimedOut,
                     Error::deadline_exceeded(AudioFileError::WaitTimeout),
@@ -619,6 +636,7 @@ impl Seek for AudioFileStreaming {
             .shared
             .download_status
             .lock()
+            .expect(DOWNLOAD_STATUS_POISON_MSG)
             .downloaded
             .contains(requested_pos as usize);
 

audio/src/fetch/mod.rs

@@ -33,6 +33,7 @@ enum ReceivedData {
 }
 
 const ONE_SECOND: Duration = Duration::from_secs(1);
+const DOWNLOAD_STATUS_POISON_MSG: &str = "audio download status mutex should not be poisoned";
 
 async fn receive_data(
     shared: Arc<AudioFileShared>,
@@ -124,7 +125,10 @@ async fn receive_data(
     if bytes_remaining > 0 {
         {
             let missing_range = Range::new(offset, bytes_remaining);
-            let mut download_status = shared.download_status.lock();
+            let mut download_status = shared
+                .download_status
+                .lock()
+                .expect(DOWNLOAD_STATUS_POISON_MSG);
             download_status.requested.subtract_range(&missing_range);
             shared.cond.notify_all();
         }
@@ -189,7 +193,11 @@ impl AudioFileFetch {
         // The iteration that follows spawns streamers fast, without awaiting them,
         // so holding the lock for the entire scope of this function should be faster
         // then locking and unlocking multiple times.
-        let mut download_status = self.shared.download_status.lock();
+        let mut download_status = self
+            .shared
+            .download_status
+            .lock()
+            .expect(DOWNLOAD_STATUS_POISON_MSG);
 
         ranges_to_request.subtract_range_set(&download_status.downloaded);
         ranges_to_request.subtract_range_set(&download_status.requested);
@@ -227,7 +235,11 @@ impl AudioFileFetch {
         let mut missing_data = RangeSet::new();
         missing_data.add_range(&Range::new(0, self.shared.file_size));
         {
-            let download_status = self.shared.download_status.lock();
+            let download_status = self
+                .shared
+                .download_status
+                .lock()
+                .expect(DOWNLOAD_STATUS_POISON_MSG);
             missing_data.subtract_range_set(&download_status.downloaded);
             missing_data.subtract_range_set(&download_status.requested);
         }
@@ -349,7 +361,11 @@ impl AudioFileFetch {
                 let received_range = Range::new(data.offset, data.data.len());
 
                 let full = {
-                    let mut download_status = self.shared.download_status.lock();
+                    let mut download_status = self
+                        .shared
+                        .download_status
+                        .lock()
+                        .expect(DOWNLOAD_STATUS_POISON_MSG);
                     download_status.downloaded.add_range(&received_range);
                     self.shared.cond.notify_all();
 
@@ -415,7 +431,10 @@ pub(super) async fn audio_file_fetch(
             initial_request.offset + initial_request.length,
         );
 
-        let mut download_status = shared.download_status.lock();
+        let mut download_status = shared
+            .download_status
+            .lock()
+            .expect(DOWNLOAD_STATUS_POISON_MSG);
         download_status.requested.add_range(&requested_range);
     }
 
@@ -466,7 +485,11 @@ pub(super) async fn audio_file_fetch(
 
         if fetch.shared.is_download_streaming() && fetch.has_download_slots_available() {
             let bytes_pending: usize = {
-                let download_status = fetch.shared.download_status.lock();
+                let download_status = fetch
+                    .shared
+                    .download_status
+                    .lock()
+                    .expect(DOWNLOAD_STATUS_POISON_MSG);
 
                 download_status
                     .requested

audio/src/fetch/receive.rs

@@ -22,12 +22,12 @@ librespot-core = { version = "0.7.1", path = "../core", default-features = false
 librespot-playback = { version = "0.7.1", path = "../playback", default-features = false }
 librespot-protocol = { version = "0.7.1", path = "../protocol", default-features = false }
 
-futures-util = "0.3"
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
 log = "0.4"
 protobuf = "3.7"
 rand = { version = "0.9", default-features = false, features = ["small_rng"] }
 serde_json = "1.0"
 thiserror = "2"
-tokio = { version = "1", features = ["macros", "parking_lot", "sync"] }
-tokio-stream = "0.1"
-uuid = { version = "1.18", features = ["v4"] }
+tokio = { version = "1", features = ["macros", "sync"] }
+tokio-stream = { version = "0.1", default-features = false }
+uuid = { version = "1.18", default-features = false, features = ["v4"] }

connect/Cargo.toml

@@ -51,16 +51,12 @@ data-encoding = "2.9"
 flate2 = "1.1"
 form_urlencoded = "1.2"
 futures-core = "0.3"
-futures-util = { version = "0.3", features = [
+futures-util = { version = "0.3", default-features = false, features = [
     "alloc",
     "bilock",
-    "sink",
     "unstable",
 ] }
-governor = { version = "0.10", default-features = false, features = [
-    "std",
-    "jitter",
-] }
+governor = { version = "0.10", default-features = false, features = ["std"] }
 hmac = "0.12"
 httparse = "1.10"
 http = "1.3"
@@ -84,14 +80,13 @@ num-bigint = "0.4"
 num-derive = "0.4"
 num-integer = "0.1"
 num-traits = "0.2"
-parking_lot = { version = "0.12", features = ["deadlock_detection"] }
 pbkdf2 = { version = "0.12", default-features = false, features = ["hmac"] }
 pin-project-lite = "0.2"
 priority-queue = "2.5"
 protobuf = "3.7"
 protobuf-json-mapping = "3.7"
 quick-xml = { version = "0.38", features = ["serialize"] }
-rand = "0.9"
+rand = { version = "0.9", default-features = false, features = ["thread_rng"] }
 rsa = "0.9"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -104,23 +99,22 @@ tokio = { version = "1", features = [
     "io-util",
     "macros",
     "net",
-    "parking_lot",
     "rt",
     "sync",
     "time",
 ] }
-tokio-stream = "0.1"
+tokio-stream = { version = "0.1", default-features = false }
 tokio-tungstenite = { version = "0.27", default-features = false }
-tokio-util = { version = "0.7", features = ["codec"] }
+tokio-util = { version = "0.7", default-features = false }
 url = "2"
 uuid = { version = "1", default-features = false, features = ["v4"] }
 
 [build-dependencies]
-rand = "0.9"
+rand = { version = "0.9", default-features = false, features = ["thread_rng"] }
 rand_distr = "0.5"
 vergen-gitcl = { version = "1.0", default-features = false, features = [
     "build",
 ] }
 
 [dev-dependencies]
-tokio = { version = "1", features = ["macros", "parking_lot"] }
+tokio = { version = "1", features = ["macros"] }

core/Cargo.toml

@@ -4,16 +4,17 @@ use std::{
     fs::{self, File},
     io::{self, Read, Write},
     path::{Path, PathBuf},
-    sync::Arc,
+    sync::{Arc, Mutex},
     time::SystemTime,
 };
 
-use parking_lot::Mutex;
 use priority_queue::PriorityQueue;
 use thiserror::Error;
 
 use crate::{Error, FileId, authentication::Credentials, error::ErrorKind};
 
+const CACHE_LIMITER_POISON_MSG: &str = "cache limiter mutex should not be poisoned";
+
 #[derive(Debug, Error)]
 pub enum CacheError {
     #[error("audio cache location is not configured")]
@@ -189,15 +190,24 @@ impl FsSizeLimiter {
     }
 
     fn add(&self, file: &Path, size: u64) {
-        self.limiter.lock().add(file, size, SystemTime::now())
+        self.limiter
+            .lock()
+            .expect(CACHE_LIMITER_POISON_MSG)
+            .add(file, size, SystemTime::now())
     }
 
     fn touch(&self, file: &Path) -> bool {
-        self.limiter.lock().update(file, SystemTime::now())
+        self.limiter
+            .lock()
+            .expect(CACHE_LIMITER_POISON_MSG)
+            .update(file, SystemTime::now())
     }
 
     fn remove(&self, file: &Path) -> bool {
-        self.limiter.lock().remove(file)
+        self.limiter
+            .lock()
+            .expect(CACHE_LIMITER_POISON_MSG)
+            .remove(file)
     }
 
     fn prune_internal<F: FnMut() -> Option<PathBuf>>(mut pop: F) -> Result<(), Error> {
@@ -232,7 +242,7 @@ impl FsSizeLimiter {
     }
 
     fn prune(&self) -> Result<(), Error> {
-        Self::prune_internal(|| self.limiter.lock().pop())
+        Self::prune_internal(|| self.limiter.lock().expect(CACHE_LIMITER_POISON_MSG).pop())
     }
 
     fn new(path: &Path, limit: u64) -> Result<Self, Error> {

core/src/cache.rs

@@ -1,20 +1,23 @@
+pub(crate) const COMPONENT_POISON_MSG: &str = "component mutex should not be poisoned";
+
 macro_rules! component {
     ($name:ident : $inner:ident { $($key:ident : $ty:ty = $value:expr,)* }) => {
         #[derive(Clone)]
-        pub struct $name(::std::sync::Arc<($crate::session::SessionWeak, ::parking_lot::Mutex<$inner>)>);
+        pub struct $name(::std::sync::Arc<($crate::session::SessionWeak, ::std::sync::Mutex<$inner>)>);
         impl $name {
             #[allow(dead_code)]
             pub(crate) fn new(session: $crate::session::SessionWeak) -> $name {
                 debug!(target:"librespot::component", "new {}", stringify!($name));
 
-                $name(::std::sync::Arc::new((session, ::parking_lot::Mutex::new($inner {
+                $name(::std::sync::Arc::new((session, ::std::sync::Mutex::new($inner {
                     $($key : $value,)*
                 }))))
             }
 
             #[allow(dead_code)]
             fn lock<F: FnOnce(&mut $inner) -> R, R>(&self, f: F) -> R {
-                let mut inner = (self.0).1.lock();
+                let mut inner = (self.0).1.lock()
+                    .expect($crate::component::COMPONENT_POISON_MSG);
                 f(&mut inner)
             }