fix(serving): low-severity audit batch findings (paroche/syndesis/archon/akouo)

crates/aitesis/src/repo.rs

@@ -585,6 +585,34 @@ mod tests {
         assert_eq!(approved.len(), 1);
     }
 
+    #[tokio::test]
+    async fn list_paginates_and_counts() {
+        let pool = setup().await;
+        let user = UserId::new();
+
+        for _ in 0..5 {
+            insert_request(&pool, &make_request(user, RequestStatus::Submitted))
+                .await
+                .unwrap();
+        }
+
+        let first_page = list_by_user(&pool, &user, Page::new(2, 0)).await.unwrap();
+        assert_eq!(first_page.len(), 2);
+        let last_page = list_by_user(&pool, &user, Page::new(2, 4)).await.unwrap();
+        assert_eq!(last_page.len(), 1);
+
+        let total = count_requests(&pool, Some(&user), None).await.unwrap();
+        assert_eq!(total, 5);
+        let submitted = count_requests(&pool, None, Some(RequestStatus::Submitted))
+            .await
+            .unwrap();
+        assert_eq!(submitted, 5);
+        let denied = count_requests(&pool, None, Some(RequestStatus::Denied))
+            .await
+            .unwrap();
+        assert_eq!(denied, 0);
+    }
+
     #[tokio::test]
     async fn count_pending_by_user_counts_active_statuses() {
         let pool = setup().await;

crates/akouo-android/src/lib.rs

@@ -90,11 +90,17 @@ struct SeekCommand {
 
 type EventListeners = Arc<Mutex<Vec<(u64, Arc<dyn EventListener>)>>>;
 
+// WHY: the callback is stored as Arc so invokers clone the handle out and
+// drop the Mutex guard BEFORE calling into foreign code — holding a
+// std::sync::Mutex across an FFI on_frame call stalls every other task
+// (drop, callback replace) for as long as the Android side blocks.
+type SharedAudioCallback = Arc<Mutex<Option<Arc<dyn AudioCallback>>>>;
+
 #[derive(uniffi::Object)]
 pub struct AndroidEngine {
     runtime: RuntimeThread,
     state: Arc<AtomicU8>,
-    audio_callback: Arc<Mutex<Option<Box<dyn AudioCallback>>>>,
+    audio_callback: SharedAudioCallback,
     event_listeners: EventListeners,
     next_listener_id: AtomicU64,
     playback_task: Mutex<Option<JoinHandle<()>>>,
@@ -149,7 +155,7 @@ impl AndroidEngine {
             .audio_callback
             .lock()
             .unwrap_or_else(|e| e.into_inner());
-        *guard = Some(callback);
+        *guard = Some(Arc::from(callback));
     }
 
     /// Registers an event listener and returns a subscription id for
@@ -333,12 +339,12 @@ impl AndroidEngine {
 
     #[cfg(test)]
     fn emit_test_frame(&self, samples: Vec<f64>) {
-        if let Some(callback) = self
+        let callback = self
             .audio_callback
             .lock()
             .unwrap_or_else(|e| e.into_inner())
-            .as_ref()
-        {
+            .clone();
+        if let Some(callback) = callback {
             callback.on_frame(samples);
         }
     }
@@ -360,7 +366,7 @@ impl Drop for AndroidEngine {
 
 struct PlaybackTaskContext {
     state: Arc<AtomicU8>,
-    callback: Arc<Mutex<Option<Box<dyn AudioCallback>>>>,
+    callback: SharedAudioCallback,
     listeners: EventListeners,
     seek_rx: mpsc::Receiver<SeekCommand>,
     ring_capacity: usize,
@@ -372,7 +378,7 @@ struct DrainTaskContext {
     state: Arc<AtomicU8>,
     producer_done: Arc<AtomicBool>,
     underruns: Arc<AtomicU64>,
-    callback: Arc<Mutex<Option<Box<dyn AudioCallback>>>>,
+    callback: SharedAudioCallback,
     listeners: EventListeners,
     callback_samples: usize,
 }
@@ -544,12 +550,14 @@ async fn drain_callback_task(context: DrainTaskContext) {
         }
 
         if context.ring.pop_frame(&mut out) {
-            if let Some(callback) = context
+            // WHY: clone the Arc and drop the guard before the foreign call —
+            // a blocking on_frame must never pin the callback Mutex.
+            let callback = context
                 .callback
                 .lock()
                 .unwrap_or_else(|e| e.into_inner())
-                .as_ref()
-            {
+                .clone();
+            if let Some(callback) = callback {
                 callback.on_frame(out.clone());
             }
             continue;
@@ -562,12 +570,12 @@ async fn drain_callback_task(context: DrainTaskContext) {
             }
             let mut tail = vec![0.0; remaining];
             if context.ring.pop_frame(&mut tail) {
-                if let Some(callback) = context
+                let callback = context
                     .callback
                     .lock()
                     .unwrap_or_else(|e| e.into_inner())
-                    .as_ref()
-                {
+                    .clone();
+                if let Some(callback) = callback {
                     callback.on_frame(tail);
                 }
                 continue;
@@ -930,6 +938,103 @@ mod tests {
         );
     }
 
+    /// Blocks inside on_frame until released, and flags entry — proves the
+    /// callback Mutex is NOT held across the foreign call.
+    struct BlockingCallback {
+        entered: Arc<AtomicBool>,
+        release: Arc<AtomicBool>,
+    }
+
+    impl AudioCallback for BlockingCallback {
+        fn on_frame(&self, _samples: Vec<f64>) {
+            self.entered.store(true, Ordering::SeqCst);
+            while !self.release.load(Ordering::SeqCst) {
+                std::thread::sleep(Duration::from_millis(1));
+            }
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn drain_task_does_not_hold_callback_lock_across_on_frame() {
+        let entered = Arc::new(AtomicBool::new(false));
+        let release = Arc::new(AtomicBool::new(false));
+        let callback: SharedAudioCallback = Arc::new(Mutex::new(Some(Arc::new(BlockingCallback {
+            entered: Arc::clone(&entered),
+            release: Arc::clone(&release),
+        })
+            as Arc<dyn AudioCallback>)));
+
+        let ring = Arc::new(RingBuffer::new(64));
+        assert!(ring.push_frame(&[0.0; 8]));
+        let state = Arc::new(AtomicU8::new(STATE_PLAYING));
+        let drain = tokio::spawn(drain_callback_task(DrainTaskContext {
+            ring,
+            state: Arc::clone(&state),
+            producer_done: Arc::new(AtomicBool::new(true)),
+            underruns: Arc::new(AtomicU64::new(0)),
+            callback: Arc::clone(&callback),
+            listeners: Arc::new(Mutex::new(Vec::new())),
+            callback_samples: 8,
+        }));
+
+        // Wait until the drain task is inside the (blocking) on_frame call.
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        while !entered.load(Ordering::SeqCst) {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "drain task never invoked the callback"
+            );
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+
+        // While on_frame is still blocked, replacing the callback must not
+        // deadlock — the guard was dropped before the foreign call.
+        let replaced = tokio::task::spawn_blocking(move || {
+            let mut guard = callback.lock().unwrap_or_else(|e| e.into_inner());
+            *guard = None;
+            true
+        });
+        let replaced = tokio::time::timeout(Duration::from_secs(5), replaced)
+            .await
+            .expect("callback lock held across on_frame — replace deadlocked")
+            .expect("replace thread panicked");
+        assert!(replaced);
+
+        // Unblock and shut down.
+        release.store(true, Ordering::SeqCst);
+        state.store(STATE_STOPPED, Ordering::SeqCst);
+        tokio::time::timeout(Duration::from_secs(5), drain)
+            .await
+            .expect("drain task must exit")
+            .expect("drain task panicked");
+    }
+
+    #[tokio::test]
+    async fn playback_task_reports_open_decoder_failure() {
+        let (_seek_tx, seek_rx) = mpsc::channel::<SeekCommand>(4);
+        let context = PlaybackTaskContext {
+            state: Arc::new(AtomicU8::new(STATE_PLAYING)),
+            callback: Arc::new(Mutex::new(None)),
+            listeners: Arc::new(Mutex::new(Vec::new())),
+            seek_rx,
+            ring_capacity: DEFAULT_RING_CAPACITY,
+            callback_samples: DEFAULT_CALLBACK_SAMPLES,
+        };
+
+        let result = playback_task(
+            PathBuf::from("/nonexistent/akouo-android-test.mp3"),
+            "/nonexistent/akouo-android-test.mp3".to_string(),
+            context,
+        )
+        .await;
+
+        let message = result.expect_err("open_decoder failure must surface as Err");
+        assert!(
+            !message.is_empty(),
+            "error message must describe the failure"
+        );
+    }
+
     // --- #398: seek must reposition playback ---
 
     #[tokio::test]

crates/akouo-core/src/decode/metadata.rs

@@ -25,11 +25,27 @@ pub struct TrackMetadata {
     pub replaygain_album_peak: Option<f32>,
 
     /// EBU R128 track loudness OFFSET in 1/256 LU units (i16).
+    ///
+    /// NOTE: `crate::config::ReplayGainConfig::r128_track_gain` is f64 dB —
+    /// convert with [`r128_offset_to_db`] before handing the value to DSP.
     pub r128_track_gain: Option<i16>,
     /// EBU R128 album loudness OFFSET in 1/256 LU units (i16).
+    ///
+    /// NOTE: `crate::config::ReplayGainConfig::r128_album_gain` is f64 dB —
+    /// convert with [`r128_offset_to_db`] before handing the value to DSP.
     pub r128_album_gain: Option<i16>,
 }
 
+/// Converts a raw R128 tag offset (1/256 LU, as stored in `R128_*_GAIN`
+/// tags) to dB for `crate::config::ReplayGainConfig`.
+///
+/// WHY: R128 offsets are already dB-equivalent LU — the only conversion is
+/// the fixed-point 1/256 scale; a raw i16 must never reach a dB consumer.
+#[must_use]
+pub fn r128_offset_to_db(raw: i16) -> f64 {
+    f64::from(raw) / 256.0
+}
+
 impl TrackMetadata {
     #[must_use]
     pub fn is_empty(&self) -> bool {
@@ -233,6 +249,13 @@ mod tests {
         assert!(info.is_none());
     }
 
+    #[test]
+    fn r128_offset_to_db_scales_by_256() {
+        assert!((r128_offset_to_db(256) - 1.0).abs() < 1e-12);
+        assert!((r128_offset_to_db(-512) - (-2.0)).abs() < 1e-12);
+        assert!((r128_offset_to_db(0)).abs() < 1e-12);
+    }
+
     #[test]
     fn parse_gain_db_with_suffix() {
         let v = parse_gain_db("-3.50 dB").unwrap_or_default();

crates/akouo-core/src/decode/probe.rs

@@ -171,6 +171,37 @@ mod tests {
         f
     }
 
+    /// Builds a minimal MP3 file: MPEG-1 Layer III, 128 kbps, 44.1 kHz frames
+    /// (header FF FB 90 00, 417-byte frames) with silent payloads.
+    fn mp3_tempfile(frames: usize) -> NamedTempFile {
+        const FRAME_LEN: usize = 417;
+        let mut v = Vec::with_capacity(frames * FRAME_LEN);
+        for _ in 0..frames {
+            v.extend_from_slice(&[0xFF, 0xFB, 0x90, 0x00]);
+            v.extend(std::iter::repeat_n(0u8, FRAME_LEN - 4));
+        }
+        let mut f = tempfile::Builder::new().suffix(".mp3").tempfile().unwrap();
+        f.write_all(&v).unwrap();
+        f
+    }
+
+    #[tokio::test]
+    async fn probe_mp3_returns_mp3_codec() {
+        let f = mp3_tempfile(8);
+        let codec = probe_codec(f.path()).await.unwrap();
+        assert!(matches!(codec, Codec::Mp3), "expected Mp3, got {codec:?}");
+    }
+
+    #[test]
+    fn map_codec_routes_opus_mp3_and_flac() {
+        use symphonia::core::codecs::audio::well_known::{
+            CODEC_ID_FLAC, CODEC_ID_MP3, CODEC_ID_OPUS,
+        };
+        assert!(matches!(map_codec(CODEC_ID_OPUS), Codec::Opus));
+        assert!(matches!(map_codec(CODEC_ID_MP3), Codec::Mp3));
+        assert!(matches!(map_codec(CODEC_ID_FLAC), Codec::Flac));
+    }
+
     #[tokio::test]
     async fn probe_wav_returns_wav_codec() {
         let f = wav_tempfile(2, 44100, &[0i16; 4]);

crates/akouo-core/src/decode/symphonia.rs

@@ -67,8 +67,21 @@ impl SymphoniaDecoder {
         let codec = map_codec(p.codec);
         let gapless_info = extract_gapless(track, &codec);
 
-        let sample_rate = p.sample_rate.unwrap_or(44100);
-        let channels = p.channels.as_ref().map(|c| c.count() as u16).unwrap_or(2);
+        // WHY: some containers legitimately omit these params, but the
+        // substitution must be observable — a silent 44100/2 default masks
+        // a wrong-rate/wrong-layout playback bug at its source.
+        let sample_rate = p.sample_rate.unwrap_or_else(|| {
+            warn!("codec params omit sample rate; defaulting to 44100 Hz");
+            44100
+        });
+        let channels = p
+            .channels
+            .as_ref()
+            .map(|c| c.count() as u16)
+            .unwrap_or_else(|| {
+                warn!("codec params omit channel layout; defaulting to 2 channels");
+                2
+            });
         let duration = track
             .num_frames
             .map(|n| Duration::from_secs_f64(n as f64 / sample_rate as f64));

crates/akouo-core/src/dsp/volume.rs

@@ -246,6 +246,16 @@ mod tests {
         assert_eq!(quantize_i32(-1.0), -2_147_483_647);
     }
 
+    #[test]
+    fn quantize_family_shares_symmetric_scale_convention() {
+        // WHY: -1.0 must map to -MAX (proportional to the 32767 / 8388607 /
+        // 2147483647 scale family) in every width — the output stage once
+        // carried a divergent i32 quantizer mapping -1.0 to i32::MIN.
+        assert_eq!(i32::from(quantize_i16(-1.0)), -32_767);
+        assert_eq!(quantize_i24(-1.0), -8_388_607);
+        assert_eq!(quantize_i32(-1.0), -2_147_483_647);
+    }
+
     #[test]
     fn quantize_f32_round_trips() {
         let x = 0.12345_f64;