refactor(providers): share one Coalescer for single-flight request coalescing

The lyrics, MusicBrainz, platform resolver and availability services each
carried their own copy of the same single-flight machinery: a Flight with
an atomic waiter count, a result slot and condvar, a Waiter drop guard, a
worker check that fails once the last waiter leaves, a 25 ms wait_timeout
poll loop, a MAX_FLIGHTS (64) bound and a finish step that publishes the
result, removes the current flight, decrements the active count and wakes
shutdown.

Move that into crate::coalescer (Flight, Waiter, Coalescer). The table
stays inside each service's own state mutex, so cache lookups, joins and
finish remain atomic with that service's cache and generation. Per-site
rules stay at the call sites: cache keys, what finish caches and when,
error values, thread names, worker post-checks, panic messages,
MusicBrainz's atomic try_join, its spawn under the state lock and its
uncached spawn-failure path. Existing tests only change their access path
to the waiter and active counts.

rust_backend/crates/providers/src/coalescer.rs

@@ -0,0 +1,455 @@
+//! Single-flight request coalescing shared by the lyrics, MusicBrainz, resolver and
+//! availability services.
+//!
+//! The flight table lives inside each service's own state mutex, so cache lookups,
+//! flight joins and [`Coalescer::finish`] stay atomic with that service's cache and
+//! generation. Services keep their own key, caching and worker rules; this module
+//! owns waiter accounting, cancellation once the last waiter leaves, the poll loop,
+//! the `MAX_FLIGHTS` bound and idle notification.
+
+use std::borrow::Borrow;
+use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Condvar, Mutex};
+use std::time::Duration;
+
+/// Caller checks are opaque callbacks, so waiters poll them while a flight runs.
+const POLL: Duration = Duration::from_millis(25);
+/// Upper bound on concurrently running workers per service.
+pub(crate) const MAX_FLIGHTS: usize = 64;
+
+/// One shared in-flight request. `T` carries per-service data fixed at creation.
+pub(crate) struct Flight<V, T = ()> {
+    tag: T,
+    waiters: AtomicUsize,
+    result: Mutex<Option<V>>,
+    ready: Condvar,
+}
+
+impl<V, T> Flight<V, T> {
+    pub(crate) fn new(tag: T) -> Arc<Self> {
+        Arc::new(Self {
+            tag,
+            waiters: AtomicUsize::new(0),
+            result: Mutex::new(None),
+            ready: Condvar::new(),
+        })
+    }
+
+    pub(crate) fn tag(&self) -> &T {
+        &self.tag
+    }
+
+    pub(crate) fn waiters(&self) -> usize {
+        self.waiters.load(Ordering::Acquire)
+    }
+
+    /// Registers one more waiter under the service lock. Create the matching
+    /// [`Waiter`] guard once that lock is released.
+    pub(crate) fn join(&self) {
+        self.waiters.fetch_add(1, Ordering::AcqRel);
+    }
+
+    /// Worker-side liveness: fails once the owner closes or every waiter has left.
+    pub(crate) fn check(
+        &self,
+        closed: &AtomicBool,
+        closed_message: &str,
+        cancelled_message: &str,
+    ) -> Result<(), String> {
+        if closed.load(Ordering::Acquire) {
+            Err(closed_message.into())
+        } else if self.waiters() == 0 {
+            Err(cancelled_message.into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Blocks until the flight finishes, running the caller's `check` before every
+    /// poll and once more after the shared result is cloned.
+    pub(crate) fn wait<E>(&self, check: impl Fn() -> Result<(), E>) -> Result<V, E>
+    where
+        V: Clone,
+    {
+        loop {
+            check()?;
+            let result = self.result.lock().expect("flight lock");
+            if let Some(value) = result.as_ref() {
+                let value = value.clone();
+                drop(result);
+                check()?;
+                return Ok(value);
+            }
+            drop(self.ready.wait_timeout(result, POLL).expect("flight lock"));
+        }
+    }
+}
+
+/// Drop guard for one registered waiter. Dropping the last guard makes
+/// [`Flight::check`] fail, which cancels the detached worker.
+pub(crate) struct Waiter<V, T = ()>(Arc<Flight<V, T>>);
+
+impl<V, T> Waiter<V, T> {
+    pub(crate) fn new(flight: Arc<Flight<V, T>>) -> Self {
+        Self(flight)
+    }
+}
+
+impl<V, T> Drop for Waiter<V, T> {
+    fn drop(&mut self) {
+        self.0.waiters.fetch_sub(1, Ordering::AcqRel);
+    }
+}
+
+/// Flight table and running-worker count, kept inside a service's state mutex.
+pub(crate) struct Coalescer<K, V, T = ()> {
+    flights: BTreeMap<K, Arc<Flight<V, T>>>,
+    active: usize,
+}
+
+impl<K, V, T> Default for Coalescer<K, V, T> {
+    fn default() -> Self {
+        Self {
+            flights: BTreeMap::new(),
+            active: 0,
+        }
+    }
+}
+
+impl<K: Ord, V, T> Coalescer<K, V, T> {
+    /// Workers started and not yet finished; shutdown waits on `idle` until zero.
+    pub(crate) fn active(&self) -> usize {
+        self.active
+    }
+
+    pub(crate) fn is_full(&self) -> bool {
+        self.active >= MAX_FLIGHTS
+    }
+
+    /// A flight for `key` that still has waiters. The caller registers itself
+    /// with [`Flight::join`] while still holding the service lock.
+    pub(crate) fn existing<Q>(&self, key: &Q) -> Option<Arc<Flight<V, T>>>
+    where
+        K: Borrow<Q>,
+        Q: Ord + ?Sized,
+    {
+        self.flights
+            .get(key)
+            .filter(|flight| flight.waiters() > 0)
+            .cloned()
+    }
+
+    /// Joins a flight for `key` only while its waiter count is still nonzero,
+    /// so an abandoned flight is never revived between the check and the join.
+    pub(crate) fn try_join<Q>(&self, key: &Q) -> Option<Arc<Flight<V, T>>>
+    where
+        K: Borrow<Q>,
+        Q: Ord + ?Sized,
+    {
+        self.flights
+            .get(key)
+            .filter(|flight| {
+                flight
+                    .waiters
+                    .try_update(Ordering::AcqRel, Ordering::Acquire, |count| {
+                        (count > 0).then_some(count + 1)
+                    })
+                    .is_ok()
+            })
+            .cloned()
+    }
+
+    /// Records a new flight and counts its worker as active. Check
+    /// [`Coalescer::is_full`] first.
+    pub(crate) fn insert(&mut self, key: K, flight: Arc<Flight<V, T>>) {
+        self.flights.insert(key, flight);
+        self.active += 1;
+    }
+
+    /// Whether `flight` is still the table entry for `key`.
+    pub(crate) fn is_current<Q>(&self, key: &Q, flight: &Arc<Flight<V, T>>) -> bool
+    where
+        K: Borrow<Q>,
+        Q: Ord + ?Sized,
+    {
+        self.flights
+            .get(key)
+            .is_some_and(|current| Arc::ptr_eq(current, flight))
+    }
+
+    /// Publishes `result` to the flight's waiters, removes the flight if it is
+    /// still current, and wakes shutdown. Call under the service lock after any
+    /// service-specific caching.
+    pub(crate) fn finish<Q>(
+        &mut self,
+        key: &Q,
+        flight: &Arc<Flight<V, T>>,
+        result: V,
+        idle: &Condvar,
+    ) where
+        K: Borrow<Q>,
+        Q: Ord + ?Sized,
+    {
+        *flight.result.lock().expect("flight lock") = Some(result);
+        flight.ready.notify_all();
+        if self.is_current(key, flight) {
+            self.flights.remove(key);
+        }
+        self.active -= 1;
+        idle.notify_all();
+    }
+
+    #[cfg(test)]
+    pub(crate) fn get<Q>(&self, key: &Q) -> Option<&Arc<Flight<V, T>>>
+    where
+        K: Borrow<Q>,
+        Q: Ord + ?Sized,
+    {
+        self.flights.get(key)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn values(&self) -> impl Iterator<Item = &Arc<Flight<V, T>>> {
+        self.flights.values()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Instant;
+
+    type Outcome = Result<u32, String>;
+
+    #[derive(Default)]
+    struct State {
+        flights: Coalescer<String, Outcome>,
+        cache: BTreeMap<String, u32>,
+    }
+
+    /// Minimal service following the same flow as the four production owners.
+    #[derive(Default)]
+    struct Service {
+        state: Mutex<State>,
+        idle: Condvar,
+        closed: AtomicBool,
+        calls: AtomicUsize,
+        release: AtomicBool,
+        worker_errors: Mutex<Vec<String>>,
+    }
+
+    impl Service {
+        fn lookup(
+            self: &Arc<Self>,
+            key: &str,
+            check: &(dyn Fn() -> Result<(), String> + Sync),
+        ) -> Outcome {
+            let (flight, start) = {
+                let mut state = self.state.lock().unwrap();
+                if let Some(value) = state.cache.get(key) {
+                    return Ok(*value);
+                }
+                let (flight, start) = if let Some(flight) = state.flights.existing(key) {
+                    (flight, false)
+                } else {
+                    if state.flights.is_full() {
+                        return Err("busy".into());
+                    }
+                    let flight = Flight::new(());
+                    state.flights.insert(key.into(), Arc::clone(&flight));
+                    (flight, true)
+                };
+                flight.join();
+                (flight, start)
+            };
+            let _waiter = Waiter::new(Arc::clone(&flight));
+            if start {
+                let service = Arc::clone(self);
+                let worker_flight = Arc::clone(&flight);
+                let worker_key = key.to_owned();
+                std::thread::spawn(move || {
+                    let check = || worker_flight.check(&service.closed, "closed", "cancelled");
+                    service.calls.fetch_add(1, Ordering::AcqRel);
+                    let result = loop {
+                        if let Err(error) = check() {
+                            service.worker_errors.lock().unwrap().push(error.clone());
+                            break Err(error);
+                        }
+                        if service.release.load(Ordering::Acquire) {
+                            break Ok(7);
+                        }
+                        std::thread::sleep(Duration::from_millis(2));
+                    };
+                    let mut state = service.state.lock().unwrap();
+                    if let Ok(value) = result {
+                        state.cache.insert(worker_key.clone(), value);
+                    }
+                    state.flights.finish(
+                        worker_key.as_str(),
+                        &worker_flight,
+                        result,
+                        &service.idle,
+                    );
+                });
+            }
+            flight.wait(check)?
+        }
+    }
+
+    fn until(condition: impl Fn() -> bool) {
+        let deadline = Instant::now() + Duration::from_secs(3);
+        while !condition() {
+            assert!(Instant::now() < deadline, "coalescer test timed out");
+            std::thread::sleep(Duration::from_millis(2));
+        }
+    }
+
+    fn waiters(service: &Service) -> usize {
+        let state = service.state.lock().unwrap();
+        state.flights.values().map(|flight| flight.waiters()).sum()
+    }
+
+    #[test]
+    fn one_flight_serves_every_waiter_and_finish_caches() {
+        let service = Arc::new(Service::default());
+        let threads: Vec<_> = (0..8)
+            .map(|_| {
+                let service = Arc::clone(&service);
+                std::thread::spawn(move || service.lookup("key", &|| Ok(())))
+            })
+            .collect();
+        until(|| waiters(&service) == 8 && service.calls.load(Ordering::Acquire) == 1);
+        std::thread::sleep(Duration::from_millis(10));
+        assert_eq!(service.calls.load(Ordering::Acquire), 1);
+        service.release.store(true, Ordering::Release);
+        for thread in threads {
+            assert_eq!(thread.join().unwrap(), Ok(7));
+        }
+        let state = service.state.lock().unwrap();
+        assert_eq!(state.flights.active(), 0);
+        assert!(state.flights.get("key").is_none());
+        assert_eq!(state.cache.get("key"), Some(&7));
+        drop(state);
+        assert_eq!(service.lookup("key", &|| Ok(())), Ok(7));
+        assert_eq!(service.calls.load(Ordering::Acquire), 1);
+    }
+
+    #[test]
+    fn last_waiter_leaving_cancels_the_worker() {
+        let service = Arc::new(Service::default());
+        let cancelled = Arc::new(AtomicBool::new(false));
+        let caller = {
+            let service = Arc::clone(&service);
+            let cancelled = Arc::clone(&cancelled);
+            std::thread::spawn(move || {
+                service.lookup("key", &|| {
+                    if cancelled.load(Ordering::Acquire) {
+                        Err("caller cancelled".into())
+                    } else {
+                        Ok(())
+                    }
+                })
+            })
+        };
+        until(|| waiters(&service) == 1);
+        cancelled.store(true, Ordering::Release);
+        assert_eq!(caller.join().unwrap(), Err("caller cancelled".into()));
+        until(|| service.state.lock().unwrap().flights.active() == 0);
+        assert_eq!(*service.worker_errors.lock().unwrap(), ["cancelled"]);
+        let state = service.state.lock().unwrap();
+        assert!(state.cache.is_empty());
+        assert!(state.flights.get("key").is_none());
+    }
+
+    #[test]
+    fn closed_owner_is_reported_before_cancellation() {
+        let flight: Arc<Flight<Outcome>> = Flight::new(());
+        let closed = AtomicBool::new(true);
+        assert_eq!(
+            flight.check(&closed, "closed", "cancelled"),
+            Err("closed".into())
+        );
+        closed.store(false, Ordering::Release);
+        assert_eq!(
+            flight.check(&closed, "closed", "cancelled"),
+            Err("cancelled".into())
+        );
+        flight.join();
+        assert_eq!(flight.check(&closed, "closed", "cancelled"), Ok(()));
+    }
+
+    #[test]
+    fn running_flights_are_bounded() {
+        let service = Arc::new(Service::default());
+        let threads: Vec<_> = (0..MAX_FLIGHTS)
+            .map(|index| {
+                let service = Arc::clone(&service);
+                std::thread::spawn(move || service.lookup(&index.to_string(), &|| Ok(())))
+            })
+            .collect();
+        until(|| service.state.lock().unwrap().flights.active() == MAX_FLIGHTS);
+        assert!(service.state.lock().unwrap().flights.is_full());
+        assert_eq!(service.lookup("over", &|| Ok(())), Err("busy".into()));
+        service.release.store(true, Ordering::Release);
+        for thread in threads {
+            assert_eq!(thread.join().unwrap(), Ok(7));
+        }
+        assert!(!service.state.lock().unwrap().flights.is_full());
+        assert_eq!(service.lookup("over", &|| Ok(())), Ok(7));
+    }
+
+    #[test]
+    fn abandoned_flights_are_not_joined_and_stale_finish_keeps_replacement() {
+        let idle = Condvar::new();
+        let mut flights: Coalescer<String, Outcome> = Coalescer::default();
+        let stale = Flight::new(());
+        flights.insert("key".into(), Arc::clone(&stale));
+        assert!(flights.existing("key").is_none());
+        assert!(flights.try_join("key").is_none());
+        assert_eq!(stale.waiters(), 0);
+        let current = Flight::new(());
+        current.join();
+        flights.insert("key".into(), Arc::clone(&current));
+        assert!(flights.try_join("key").is_some());
+        assert_eq!(current.waiters(), 2);
+        assert!(!flights.is_current("key", &stale));
+        flights.finish("key", &stale, Ok(1), &idle);
+        assert_eq!(stale.wait(|| Ok::<_, String>(())), Ok(Ok(1)));
+        assert!(flights.is_current("key", &current));
+        assert_eq!(flights.active(), 1);
+        flights.finish("key", &current, Ok(2), &idle);
+        assert!(flights.get("key").is_none());
+        assert_eq!(flights.active(), 0);
+    }
+
+    #[test]
+    fn finish_wakes_shutdown_waiting_for_idle() {
+        let service = Arc::new(Service::default());
+        let flight = Flight::new(());
+        service
+            .state
+            .lock()
+            .unwrap()
+            .flights
+            .insert("key".into(), Arc::clone(&flight));
+        let shutdown = {
+            let service = Arc::clone(&service);
+            std::thread::spawn(move || {
+                let mut state = service.state.lock().unwrap();
+                while state.flights.active() != 0 {
+                    state = service.idle.wait(state).unwrap();
+                }
+            })
+        };
+        std::thread::sleep(Duration::from_millis(20));
+        assert!(!shutdown.is_finished());
+        service
+            .state
+            .lock()
+            .unwrap()
+            .flights
+            .finish("key", &flight, Ok(1), &service.idle);
+        shutdown.join().unwrap();
+    }
+}

rust_backend/crates/providers/src/lib.rs

@@ -4,6 +4,7 @@
 //! behavior (wire formats, parsing quirks, file layouts) remains a compatibility
 //! contract.
 
+mod coalescer;
 pub mod deezer;
 pub mod lyrics;
 pub mod musicbrainz;

rust_backend/crates/providers/src/lyrics/service.rs

@@ -2,6 +2,7 @@
 //! in-flight requests, and caches negative results.
 
 use super::{CallGraph, CallNode, Check, LyricsError, cache::LyricsCache};
+use crate::coalescer::{Coalescer, Flight, Waiter};
 use spotiflac_core::lyrics::{LyricsResponse, config, config::FetchOptions, matching};
 use spotiflac_core::matching::lowercase;
 use std::collections::BTreeMap;
@@ -16,7 +17,6 @@ const NEGATIVE_TTL: Duration = Duration::from_secs(5 * 60);
 const UNAVAILABLE_COOLDOWN: Duration = Duration::from_secs(10 * 60);
 const PRIORITY_GRACE: Duration = Duration::from_secs(5);
 const MAX_NEGATIVE: usize = 500;
-const MAX_FLIGHTS: usize = 64;
 
 #[derive(Clone, Debug, Default)]
 pub struct SearchRequest {
@@ -56,20 +56,8 @@ pub trait LyricsFetcher: Send + Sync + 'static {
 }
 
 type FetchResult = Result<LyricsResponse, LyricsError>;
-
-struct Flight {
-    node: CallNode,
-    waiters: AtomicUsize,
-    result: Mutex<Option<FetchResult>>,
-    ready: Condvar,
-}
-
-struct Waiter(Arc<Flight>);
-impl Drop for Waiter {
-    fn drop(&mut self) {
-        self.0.waiters.fetch_sub(1, Ordering::AcqRel);
-    }
-}
+/// Each flight carries the call-graph node its worker runs under.
+type LyricsFlight = Flight<FetchResult, CallNode>;
 
 #[derive(Default)]
 struct State {
@@ -79,8 +67,7 @@ struct State {
     generation: u64,
     negative: BTreeMap<String, Instant>,
     health: BTreeMap<String, Instant>,
-    flights: BTreeMap<(u64, String), Arc<Flight>>,
-    active: usize,
+    flights: Coalescer<(u64, String), FetchResult, CallNode>,
 }
 
 struct Inner {
@@ -209,44 +196,33 @@ impl LyricsService {
             }
             state.negative.remove(&key);
             let flight_key = (state.generation, key);
-            let existing = state
-                .flights
-                .get(&flight_key)
-                .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0)
-                .cloned();
+            let existing = state.flights.existing(&flight_key);
             let (flight, start) = if let Some(flight) = existing {
                 (flight, false)
             } else {
-                if state.active >= MAX_FLIGHTS {
+                if state.flights.is_full() {
                     return Err(LyricsError::Other(
                         "too many concurrent lyrics searches".into(),
                     ));
                 }
-                let flight = Arc::new(Flight {
-                    node: self.inner.calls.node(),
-                    waiters: AtomicUsize::new(0),
-                    result: Mutex::new(None),
-                    ready: Condvar::new(),
-                });
-                (flight, true)
+                (Flight::new(self.inner.calls.node()), true)
             };
             let dependency = request
                 .caller
                 .as_ref()
-                .map(|caller| caller.wait_for(&flight.node))
+                .map(|caller| caller.wait_for(flight.tag()))
                 .transpose()
                 .map_err(LyricsError::Recursive)?;
             if start {
                 state
                     .flights
                     .insert(flight_key.clone(), Arc::clone(&flight));
-                state.active += 1;
             }
-            request.caller = Some(flight.node.clone());
-            flight.waiters.fetch_add(1, Ordering::AcqRel);
+            request.caller = Some(flight.tag().clone());
+            flight.join();
             (flight, start, flight_key, providers, dependency)
         };
-        let _waiter = Waiter(Arc::clone(&flight));
+        let _waiter = Waiter::new(Arc::clone(&flight));
         if start {
             let inner = Arc::clone(&self.inner);
             let worker_flight = Arc::clone(&flight);
@@ -255,13 +231,11 @@ impl LyricsService {
                 .name("lyrics-search".into())
                 .spawn(move || {
                     let check = || {
-                        if inner.closed.load(Ordering::Acquire) {
-                            Err("lyrics service is closed".into())
-                        } else if worker_flight.waiters.load(Ordering::Acquire) == 0 {
-                            Err("lyrics request cancelled".into())
-                        } else {
-                            Ok(())
-                        }
+                        worker_flight.check(
+                            &inner.closed,
+                            "lyrics service is closed",
+                            "lyrics request cancelled",
+                        )
                     };
                     let result = catch_unwind(AssertUnwindSafe(|| {
                         inner.fetch_uncached(
@@ -283,22 +257,7 @@ impl LyricsService {
                 );
             }
         }
-        loop {
-            self.inner.check(check)?;
-            let result = flight.result.lock().expect("lyrics flight lock");
-            if let Some(value) = result.as_ref() {
-                let value = value.clone();
-                drop(result);
-                self.inner.check(check)?;
-                return value;
-            }
-            drop(
-                flight
-                    .ready
-                    .wait_timeout(result, POLL)
-                    .expect("lyrics flight lock"),
-            );
-        }
+        flight.wait(|| self.inner.check(check))?
     }
 
     /// Cancel workers, wait for their HTTP/extension calls to unwind, then flush
@@ -306,7 +265,7 @@ impl LyricsService {
     pub fn shutdown(&self) -> Result<(), String> {
         self.inner.closed.store(true, Ordering::Release);
         let mut state = self.inner.state.lock().expect("lyrics service lock");
-        while state.active > 0 {
+        while state.flights.active() > 0 {
             state = self.inner.idle.wait(state).expect("lyrics service lock");
         }
         drop(state);
@@ -329,7 +288,7 @@ impl Inner {
         self.fetcher.check()
     }
 
-    fn finish(&self, key: &(u64, String), flight: &Arc<Flight>, result: FetchResult) {
+    fn finish(&self, key: &(u64, String), flight: &Arc<LyricsFlight>, result: FetchResult) {
         let mut state = self.state.lock().expect("lyrics service lock");
         if state.generation == key.0 && !self.closed.load(Ordering::Acquire) {
             if result.as_ref().is_err_and(|error| {
@@ -345,17 +304,7 @@ impl Inner {
                 state.negative.remove(&key.1);
             }
         }
-        *flight.result.lock().expect("lyrics flight lock") = Some(result);
-        flight.ready.notify_all();
-        if state
-            .flights
-            .get(key)
-            .is_some_and(|current| Arc::ptr_eq(current, flight))
-        {
-            state.flights.remove(key);
-        }
-        state.active -= 1;
-        self.idle.notify_all();
+        state.flights.finish(key, flight, result, &self.idle);
     }
 
     fn fetch_uncached(

rust_backend/crates/providers/src/lyrics/service/tests.rs

@@ -52,7 +52,7 @@ fn waiters(service: &LyricsService) -> usize {
         .unwrap()
         .flights
         .values()
-        .map(|flight| flight.waiters.load(Ordering::Acquire))
+        .map(|flight| flight.waiters())
         .sum()
 }
 fn wait_gate(gate: &AtomicBool, check: &Check<'_>) -> Result<(), LyricsError> {
@@ -143,7 +143,7 @@ fn abandoning_all_waiters_cancels_work_without_negative_caching() {
         worker.join().unwrap(),
         Err(LyricsError::Cancelled(_))
     ));
-    until(|| service.inner.state.lock().unwrap().active == 0);
+    until(|| service.inner.state.lock().unwrap().flights.active() == 0);
     assert_eq!(
         service.fetch(request(), &|| Ok(())).unwrap().source,
         "LRCLIB"
@@ -389,7 +389,7 @@ fn instrumental_heuristic_skips_network_and_provider_panics_release_workers() {
     assert_eq!(response.source, "Heuristic: Instrumental");
     assert!(service.fetch(request(), &|| Ok(())).is_err());
     service.shutdown().unwrap();
-    assert_eq!(service.inner.state.lock().unwrap().active, 0);
+    assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0);
 }
 
 #[test]

rust_backend/crates/providers/src/musicbrainz.rs

@@ -4,20 +4,20 @@ mod http;
 #[cfg(test)]
 mod tests;
 
+use crate::coalescer::{Coalescer, Flight, Waiter};
 use crate::resolver::{Check, ResolverError};
 use spotiflac_core::matching::uppercase;
 use spotiflac_core::metadata::musicbrainz::Response;
 use spotiflac_network::NetworkService;
 use std::collections::BTreeMap;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Condvar, Mutex};
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
 const POLL: Duration = Duration::from_millis(25);
 const MAX_CACHE: usize = 256;
-const MAX_FLIGHTS: usize = 64;
 type Outcome = Result<Arc<Response>, ResolverError>;
 
 /// Trusted native/test configuration. These options are not exposed to JS/FFI.
@@ -84,24 +84,9 @@ impl Cache {
     }
 }
 
-#[derive(Default)]
-struct Flight {
-    waiters: AtomicUsize,
-    result: Mutex<Option<Outcome>>,
-    ready: Condvar,
-}
-
-struct Waiter(Arc<Flight>);
-impl Drop for Waiter {
-    fn drop(&mut self) {
-        self.0.waiters.fetch_sub(1, Ordering::AcqRel);
-    }
-}
-
 #[derive(Default)]
 struct State {
-    active: usize,
-    flights: BTreeMap<String, Arc<Flight>>,
+    flights: Coalescer<String, Outcome>,
     workers: Vec<JoinHandle<()>>,
     cache: Cache,
 }
@@ -192,30 +177,17 @@ impl MusicBrainzClient {
                 self.inner.check(check)?;
                 return value.map(|value| (key, value));
             }
-            if let Some(flight) = state
-                .flights
-                .get(&key)
-                .filter(|flight| {
-                    flight
-                        .waiters
-                        .try_update(Ordering::AcqRel, Ordering::Acquire, |count| {
-                            (count > 0).then_some(count + 1)
-                        })
-                        .is_ok()
-                })
-                .cloned()
-            {
+            if let Some(flight) = state.flights.try_join(&key) {
                 flight
             } else {
-                if state.active >= MAX_FLIGHTS {
+                if state.flights.is_full() {
                     return Err(ResolverError::Failed(
                         "MusicBrainz lookup limit reached".into(),
                     ));
                 }
-                let flight = Arc::new(Flight::default());
-                flight.waiters.store(1, Ordering::Release);
+                let flight = Flight::new(());
+                flight.join();
                 state.flights.insert(key.clone(), flight.clone());
-                state.active += 1;
                 let inner = self.inner.clone();
                 let worker_key = key.clone();
                 let worker_flight = flight.clone();
@@ -224,13 +196,11 @@ impl MusicBrainzClient {
                     .name("musicbrainz".into())
                     .spawn(move || {
                         let check = || {
-                            if inner.closed.load(Ordering::Acquire) {
-                                Err("MusicBrainz client closed".into())
-                            } else if worker_flight.waiters.load(Ordering::Acquire) == 0 {
-                                Err("MusicBrainz request cancelled".into())
-                            } else {
-                                Ok(())
-                            }
+                            worker_flight.check(
+                                &inner.closed,
+                                "MusicBrainz client closed",
+                                "MusicBrainz request cancelled",
+                            )
                         };
                         let result = catch_unwind(AssertUnwindSafe(|| {
                             if worker_key.is_empty() {
@@ -249,36 +219,28 @@ impl MusicBrainzClient {
                         );
                     }) {
                     Ok(worker) => state.workers.push(worker),
-                    Err(error) => {
-                        state.flights.remove(&key);
-                        state.active -= 1;
-                        *flight.result.lock().unwrap() =
-                            Some(Err(ResolverError::Failed(error.to_string())));
-                        self.inner.idle.notify_all();
-                    }
+                    // Spawn failures are published without caching.
+                    Err(error) => state.flights.finish(
+                        &key,
+                        &flight,
+                        Err(ResolverError::Failed(error.to_string())),
+                        &self.inner.idle,
+                    ),
                 }
                 flight
             }
         };
-        let _waiter = Waiter(flight.clone());
-        loop {
-            self.inner.check(check)?;
-            let result_guard = flight.result.lock().unwrap();
-            if let Some(result) = result_guard.as_ref() {
-                let result = result.clone();
-                drop(result_guard);
-                self.inner.check(check)?;
-                return result.map(|value| (key, value));
-            }
-            drop(flight.ready.wait_timeout(result_guard, POLL).unwrap());
-        }
+        let _waiter = Waiter::new(flight.clone());
+        flight
+            .wait(|| self.inner.check(check))?
+            .map(|value| (key, value))
     }
 
     pub fn shutdown(&self) {
         let _shutdown = self.inner.shutdown.lock().unwrap();
         self.inner.closed.store(true, Ordering::Release);
         let mut state = self.inner.state.lock().unwrap();
-        while state.active != 0 {
+        while state.flights.active() != 0 {
             state = self.inner.idle.wait(state).unwrap();
         }
         state.cache.0.clear();
@@ -305,29 +267,21 @@ impl Inner {
         }
     }
 
-    fn finish(&self, key: &str, flight: &Arc<Flight>, result: Outcome) {
+    fn finish(&self, key: &str, flight: &Arc<Flight<Outcome>>, result: Outcome) {
         let mut state = self.state.lock().unwrap();
-        let current = state
-            .flights
-            .get(key)
-            .is_some_and(|current| Arc::ptr_eq(current, flight));
-        if current {
-            state.flights.remove(key);
-            if !self.closed.load(Ordering::Acquire)
-                && flight.waiters.load(Ordering::Acquire) > 0
-                && !matches!(
-                    result,
-                    Err(ResolverError::Cancelled(_) | ResolverError::Closed | ResolverError::Busy)
-                )
-            {
-                state
-                    .cache
-                    .put(key, result.clone(), Instant::now(), &self.options);
-            }
+        // Only the current flight caches, and only while a caller still wants it.
+        if state.flights.is_current(key, flight)
+            && !self.closed.load(Ordering::Acquire)
+            && flight.waiters() > 0
+            && !matches!(
+                result,
+                Err(ResolverError::Cancelled(_) | ResolverError::Closed | ResolverError::Busy)
+            )
+        {
+            state
+                .cache
+                .put(key, result.clone(), Instant::now(), &self.options);
         }
-        *flight.result.lock().unwrap() = Some(result);
-        flight.ready.notify_all();
-        state.active -= 1;
-        self.idle.notify_all();
+        state.flights.finish(key, flight, result, &self.idle);
     }
 }

rust_backend/crates/providers/src/musicbrainz/tests.rs

@@ -1,5 +1,6 @@
 use super::*;
 use spotiflac_core::metadata::musicbrainz::{ArtistCredit, Recording, Release, Tag};
+use std::sync::atomic::AtomicUsize;
 use std::thread;
 
 fn response() -> Arc<Response> {
@@ -174,9 +175,15 @@ fn genre_and_album_artist_share_a_flight_without_cancelling_other_callers() {
         })
         .collect();
     until(|| {
-        client.inner.state.lock().unwrap().flights["USAA00000101"]
-            .waiters
-            .load(Ordering::Acquire)
+        client
+            .inner
+            .state
+            .lock()
+            .unwrap()
+            .flights
+            .get("USAA00000101")
+            .unwrap()
+            .waiters()
             == 9
     });
     cancel.store(true, Ordering::Release);
@@ -236,7 +243,7 @@ fn abandoned_flight_cannot_publish_over_its_replacement() {
     probe.release.store(true, Ordering::Release);
     assert_eq!(second.join().unwrap().unwrap(), "Rock");
     probe.hold_cancelled.store(false, Ordering::Release);
-    until(|| client.inner.state.lock().unwrap().active == 0);
+    until(|| client.inner.state.lock().unwrap().flights.active() == 0);
     assert_eq!(client.genre("same", &|| Ok(())).unwrap(), "Rock");
     assert_eq!(probe.calls.load(Ordering::Acquire), 2);
 }

rust_backend/crates/providers/src/resolver/availability/cache.rs

@@ -4,12 +4,13 @@ use super::{
     AlbumAvailability, TrackAvailability, album, deezer_id_from_metadata, from_deezer, from_links,
     resolved_links,
 };
+use crate::coalescer::{Coalescer, Flight, Waiter};
 use crate::deezer::MetadataLookup;
 use crate::lyrics::{LyricsError, builtin::TrackResolver};
 use crate::resolver::{Check, PlatformResolverService, Resolver, ResolverError};
 use std::collections::BTreeMap;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::{Duration, Instant};
 
@@ -38,25 +39,12 @@ struct Entry {
     result: Outcome,
     expires: Instant,
 }
-#[derive(Default)]
-struct Flight {
-    waiters: AtomicUsize,
-    result: Mutex<Option<Outcome>>,
-    ready: Condvar,
-}
-struct Waiter(Arc<Flight>);
-impl Drop for Waiter {
-    fn drop(&mut self) {
-        self.0.waiters.fetch_sub(1, Ordering::AcqRel);
-    }
-}
 
 struct State {
     generation: u64,
     region: String,
     cache: BTreeMap<Key, Entry>,
-    flights: BTreeMap<(u64, Key), Arc<Flight>>,
-    active: usize,
+    flights: Coalescer<(u64, Key), Outcome>,
 }
 
 struct Inner {
@@ -82,8 +70,7 @@ impl AvailabilityService {
                     generation: 0,
                     region: "US".into(),
                     cache: BTreeMap::new(),
-                    flights: BTreeMap::new(),
-                    active: 0,
+                    flights: Coalescer::default(),
                 }),
                 closed: AtomicBool::new(false),
                 idle: Condvar::new(),
@@ -225,26 +212,21 @@ impl AvailabilityService {
             state.cache.remove(&key);
             let generation = state.generation;
             let flight_key = (generation, key.clone());
-            let existing = state
-                .flights
-                .get(&flight_key)
-                .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0)
-                .cloned();
+            let existing = state.flights.existing(&flight_key);
             let (flight, start) = if let Some(flight) = existing {
                 (flight, false)
             } else {
-                if state.active >= 64 {
+                if state.flights.is_full() {
                     return Err(ResolverError::Busy);
                 }
-                let flight = Arc::new(Flight::default());
+                let flight = Flight::new(());
                 state.flights.insert(flight_key, flight.clone());
-                state.active += 1;
                 (flight, true)
             };
-            flight.waiters.fetch_add(1, Ordering::AcqRel);
+            flight.join();
             (key, generation, flight, start)
         };
-        let _waiter = Waiter(flight.clone());
+        let _waiter = Waiter::new(flight.clone());
         if start {
             let inner = self.inner.clone();
             let worker_key = key.clone();
@@ -253,13 +235,11 @@ impl AvailabilityService {
                 .name("platform-availability".into())
                 .spawn(move || {
                     let check = || {
-                        if inner.closed.load(Ordering::Acquire) {
-                            Err("availability service closed".into())
-                        } else if worker_flight.waiters.load(Ordering::Acquire) == 0 {
-                            Err("availability request cancelled".into())
-                        } else {
-                            Ok(())
-                        }
+                        worker_flight.check(
+                            &inner.closed,
+                            "availability service closed",
+                            "availability request cancelled",
+                        )
                     };
                     let result =
                         catch_unwind(AssertUnwindSafe(|| inner.fetch(&worker_key, &check)))
@@ -278,28 +258,13 @@ impl AvailabilityService {
                 );
             }
         }
-        loop {
-            self.inner.check(check)?;
-            let result = flight.result.lock().unwrap();
-            if let Some(value) = result.as_ref() {
-                let value = value.clone();
-                drop(result);
-                self.inner.check(check)?;
-                return value;
-            }
-            drop(
-                flight
-                    .ready
-                    .wait_timeout(result, Duration::from_millis(25))
-                    .unwrap(),
-            );
-        }
+        flight.wait(|| self.inner.check(check))?
     }
 
     pub fn shutdown(&self) {
         self.inner.closed.store(true, Ordering::Release);
         let mut state = self.inner.state.lock().unwrap();
-        while state.active != 0 {
+        while state.flights.active() != 0 {
             state = self.inner.idle.wait(state).unwrap();
         }
         state.cache.clear();
@@ -391,7 +356,7 @@ impl Inner {
         }
     }
 
-    fn finish(&self, generation: u64, key: &Key, flight: &Arc<Flight>, result: Outcome) {
+    fn finish(&self, generation: u64, key: &Key, flight: &Arc<Flight<Outcome>>, result: Outcome) {
         let mut state = self.state.lock().unwrap();
         if generation == state.generation
             && !self.closed.load(Ordering::Acquire)
@@ -430,18 +395,9 @@ impl Inner {
                 },
             );
         }
-        *flight.result.lock().unwrap() = Some(result);
-        flight.ready.notify_all();
-        let flight_key = (generation, key.clone());
-        if state
+        state
             .flights
-            .get(&flight_key)
-            .is_some_and(|value| Arc::ptr_eq(value, flight))
-        {
-            state.flights.remove(&flight_key);
-        }
-        state.active -= 1;
-        self.idle.notify_all();
+            .finish(&(generation, key.clone()), flight, result, &self.idle);
     }
 }
 
@@ -450,6 +406,7 @@ mod tests {
     use super::*;
     use crate::resolver::{Metadata, Resolution};
     use spotiflac_core::metadata::TrackMetadata;
+    use std::sync::atomic::AtomicUsize;
 
     #[derive(Default)]
     struct Probe {
@@ -547,8 +504,7 @@ mod tests {
                 .values()
                 .next()
                 .unwrap()
-                .waiters
-                .load(Ordering::Acquire)
+                .waiters()
                 == 2
         });
         cancelled.store(true, Ordering::Release);
@@ -627,7 +583,7 @@ mod tests {
         assert_eq!(service.set_region("ID"), Err(ResolverError::Closed));
         assert_eq!(service.region(), Err(ResolverError::Closed));
         assert_eq!(service.clear_cache(), Err(ResolverError::Closed));
-        assert_eq!(service.inner.state.lock().unwrap().active, 0);
+        assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0);
     }
 
     #[test]
@@ -694,7 +650,7 @@ mod tests {
             ),
             Err(ResolverError::Cancelled(_))
         ));
-        until(|| service.inner.state.lock().unwrap().active == 0);
+        until(|| service.inner.state.lock().unwrap().flights.active() == 0);
         assert!(
             !service
                 .inner
@@ -718,7 +674,7 @@ mod tests {
                 service.check_track(&index.to_string(), "", &|| Ok(()))
             }));
         }
-        until(|| service.inner.state.lock().unwrap().active == 64);
+        until(|| service.inner.state.lock().unwrap().flights.active() == 64);
         assert_eq!(
             service.check_track("over-limit", "", &|| Ok(())),
             Err(ResolverError::Busy)

rust_backend/crates/providers/src/resolver/service.rs

@@ -1,18 +1,17 @@
 //! Platform resolver service that resolves URLs and Spotify tracks to other platforms, with caching.
 
 use super::{Check, Metadata, Resolution, Resolver, ResolverError, check_active};
+use crate::coalescer::{Coalescer, Flight, Waiter};
 use crate::lyrics::{LyricsError, builtin::TrackResolver};
 use std::collections::BTreeMap;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::{Duration, Instant};
 
 const POSITIVE_TTL: Duration = Duration::from_secs(30 * 60);
 const NEGATIVE_TTL: Duration = Duration::from_secs(5 * 60);
 const MAX_CACHE: usize = 500;
-const MAX_FLIGHTS: usize = 64;
-const POLL: Duration = Duration::from_millis(25);
 type Outcome = Result<Resolution, ResolverError>;
 
 #[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
@@ -25,20 +24,6 @@ struct Key {
     cached: bool,
 }
 
-#[derive(Default)]
-struct Flight {
-    waiters: AtomicUsize,
-    result: Mutex<Option<Outcome>>,
-    ready: Condvar,
-}
-
-struct Waiter(Arc<Flight>);
-impl Drop for Waiter {
-    fn drop(&mut self) {
-        self.0.waiters.fetch_sub(1, Ordering::AcqRel);
-    }
-}
-
 struct CacheEntry {
     value: Outcome,
     expires: Instant,
@@ -47,8 +32,7 @@ struct CacheEntry {
 struct State {
     region: String,
     generation: u64,
-    active: usize,
-    flights: BTreeMap<Key, Arc<Flight>>,
+    flights: Coalescer<Key, Outcome>,
     cache: BTreeMap<(String, String), CacheEntry>,
 }
 
@@ -74,8 +58,7 @@ impl PlatformResolverService {
                 state: Mutex::new(State {
                     region: "US".into(),
                     generation: 0,
-                    active: 0,
-                    flights: BTreeMap::new(),
+                    flights: Coalescer::default(),
                     cache: BTreeMap::new(),
                 }),
                 idle: Condvar::new(),
@@ -150,26 +133,21 @@ impl PlatformResolverService {
                 artist: hint.artist.clone(),
                 cached,
             };
-            let existing = state
-                .flights
-                .get(&key)
-                .filter(|flight| flight.waiters.load(Ordering::Acquire) > 0)
-                .cloned();
+            let existing = state.flights.existing(&key);
             let (flight, start) = if let Some(flight) = existing {
                 (flight, false)
             } else {
-                if state.active >= MAX_FLIGHTS {
+                if state.flights.is_full() {
                     return Err(ResolverError::Busy);
                 }
-                let flight = Arc::new(Flight::default());
+                let flight = Flight::new(());
                 state.flights.insert(key.clone(), Arc::clone(&flight));
-                state.active += 1;
                 (flight, true)
             };
-            flight.waiters.fetch_add(1, Ordering::AcqRel);
+            flight.join();
             (key, flight, start)
         };
-        let _waiter = Waiter(Arc::clone(&flight));
+        let _waiter = Waiter::new(Arc::clone(&flight));
         if start {
             let inner = Arc::clone(&self.inner);
             let worker_key = key.clone();
@@ -178,13 +156,11 @@ impl PlatformResolverService {
                 .name("platform-resolver".into())
                 .spawn(move || {
                     let check = || {
-                        if inner.closed.load(Ordering::Acquire) {
-                            Err("platform resolver closed".into())
-                        } else if worker_flight.waiters.load(Ordering::Acquire) == 0 {
-                            Err("platform resolution cancelled".into())
-                        } else {
-                            Ok(())
-                        }
+                        worker_flight.check(
+                            &inner.closed,
+                            "platform resolver closed",
+                            "platform resolution cancelled",
+                        )
                     };
                     let result = catch_unwind(AssertUnwindSafe(|| {
                         inner.resolver.resolve(
@@ -209,21 +185,13 @@ impl PlatformResolverService {
                     .finish(&key, &flight, Err(ResolverError::Failed(error.to_string())));
             }
         }
-        loop {
-            self.inner.check(check)?;
-            let result = flight.result.lock().unwrap();
-            if let Some(result) = result.as_ref() {
-                self.inner.check(check)?;
-                return result.clone();
-            }
-            drop(flight.ready.wait_timeout(result, POLL).unwrap());
-        }
+        flight.wait(|| self.inner.check(check))?
     }
 
     pub fn shutdown(&self) {
         self.inner.closed.store(true, Ordering::Release);
         let mut state = self.inner.state.lock().unwrap();
-        while state.active != 0 {
+        while state.flights.active() != 0 {
             state = self.inner.idle.wait(state).unwrap();
         }
         state.cache.clear();
@@ -269,7 +237,7 @@ impl Inner {
         }
     }
 
-    fn finish(&self, key: &Key, flight: &Arc<Flight>, result: Outcome) {
+    fn finish(&self, key: &Key, flight: &Arc<Flight<Outcome>>, result: Outcome) {
         let mut state = self.state.lock().unwrap();
         if key.cached
             && key.generation == state.generation
@@ -303,23 +271,15 @@ impl Inner {
                 },
             );
         }
-        *flight.result.lock().unwrap() = Some(result);
-        flight.ready.notify_all();
-        if state
-            .flights
-            .get(key)
-            .is_some_and(|current| Arc::ptr_eq(current, flight))
-        {
-            state.flights.remove(key);
-        }
-        state.active -= 1;
-        self.idle.notify_all();
+        state.flights.finish(key, flight, result, &self.idle);
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::coalescer::MAX_FLIGHTS;
+    use std::sync::atomic::AtomicUsize;
 
     #[derive(Default)]
     struct Probe {
@@ -392,8 +352,7 @@ mod tests {
                 .values()
                 .next()
                 .unwrap()
-                .waiters
-                .load(Ordering::Acquire)
+                .waiters()
                 == 2
         });
         cancelled.store(true, Ordering::Release);
@@ -434,7 +393,7 @@ mod tests {
             ),
             Err(ResolverError::Cancelled(_))
         ));
-        until(|| service.inner.state.lock().unwrap().active == 0);
+        until(|| service.inner.state.lock().unwrap().flights.active() == 0);
         assert!(service.inner.state.lock().unwrap().cache.is_empty());
         let waiting = {
             let service = service.clone();
@@ -449,7 +408,7 @@ mod tests {
         );
         assert_eq!(service.set_region("ID"), Err(ResolverError::Closed));
         assert_eq!(service.clear_cache(), Err(ResolverError::Closed));
-        assert_eq!(service.inner.state.lock().unwrap().active, 0);
+        assert_eq!(service.inner.state.lock().unwrap().flights.active(), 0);
     }
 
     #[test]