diff --git a/crates/aitesis/src/repo.rs b/crates/aitesis/src/repo.rs index d1e5d6fa..8b3562e1 100644 --- a/crates/aitesis/src/repo.rs +++ b/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; diff --git a/crates/akouo-android/src/lib.rs b/crates/akouo-android/src/lib.rs index 8eeb9a6c..0b1c8018 100644 --- a/crates/akouo-android/src/lib.rs +++ b/crates/akouo-android/src/lib.rs @@ -90,11 +90,17 @@ struct SeekCommand { type EventListeners = Arc)>>>; +// 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>>>; + #[derive(uniffi::Object)] pub struct AndroidEngine { runtime: RuntimeThread, state: Arc, - audio_callback: Arc>>>, + audio_callback: SharedAudioCallback, event_listeners: EventListeners, next_listener_id: AtomicU64, playback_task: Mutex>>, @@ -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) { - 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, - callback: Arc>>>, + callback: SharedAudioCallback, listeners: EventListeners, seek_rx: mpsc::Receiver, ring_capacity: usize, @@ -372,7 +378,7 @@ struct DrainTaskContext { state: Arc, producer_done: Arc, underruns: Arc, - callback: Arc>>>, + 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, + release: Arc, + } + + impl AudioCallback for BlockingCallback { + fn on_frame(&self, _samples: Vec) { + 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))); + + 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::(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] diff --git a/crates/akouo-core/src/decode/metadata.rs b/crates/akouo-core/src/decode/metadata.rs index 7ef52440..a564a054 100644 --- a/crates/akouo-core/src/decode/metadata.rs +++ b/crates/akouo-core/src/decode/metadata.rs @@ -25,11 +25,27 @@ pub struct TrackMetadata { pub replaygain_album_peak: Option, /// 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, /// 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, } +/// 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(); diff --git a/crates/akouo-core/src/decode/probe.rs b/crates/akouo-core/src/decode/probe.rs index d1996598..fa82550e 100644 --- a/crates/akouo-core/src/decode/probe.rs +++ b/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]); diff --git a/crates/akouo-core/src/decode/symphonia.rs b/crates/akouo-core/src/decode/symphonia.rs index 86222a83..3b34b834 100644 --- a/crates/akouo-core/src/decode/symphonia.rs +++ b/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)); diff --git a/crates/akouo-core/src/dsp/volume.rs b/crates/akouo-core/src/dsp/volume.rs index cd0f568a..18e0e5c2 100644 --- a/crates/akouo-core/src/dsp/volume.rs +++ b/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; diff --git a/crates/akouo-core/src/engine.rs b/crates/akouo-core/src/engine.rs index 9224ee1f..d110adad 100644 --- a/crates/akouo-core/src/engine.rs +++ b/crates/akouo-core/src/engine.rs @@ -145,6 +145,11 @@ impl Engine { /// Spawns decode and DSP tasks via `tokio::spawn`; must be called within a Tokio runtime. #[instrument(skip(self))] pub fn play(&self, source: AudioSource) -> Result<(), EngineError> { + // WHY: the session lock is held across the whole spawn sequence — a + // stop() racing between task spawn and session store would otherwise + // find `session` empty and be unable to abort the new tasks. + let mut guard = self.session.lock().unwrap_or_else(|e| e.into_inner()); + // Atomically transition STOPPED → PLAYING. self.state .compare_exchange( @@ -221,8 +226,7 @@ impl Engine { .instrument(tracing::info_span!("dsp_task")), ); - // Store session. - let mut guard = self.session.lock().unwrap_or_else(|e| e.into_inner()); + // Store session (lock held since before the CAS). *guard = Some(PlaybackSession { decode_task, dsp_task, @@ -267,16 +271,32 @@ impl Engine { } /// Stops playback and resets the engine to idle. + /// + /// Awaits both pipeline tasks after aborting them, so when `stop()` + /// returns their resources (decoder thread, output stream) are released + /// — an immediate `play()` never races the old session's teardown. #[instrument(skip(self))] - pub fn stop(&self) -> Result<(), EngineError> { + pub async fn stop(&self) -> Result<(), EngineError> { self.state.store(STATE_STOPPED, Ordering::SeqCst); - let mut guard = self.session.lock().unwrap_or_else(|e| e.into_inner()); - if let Some(session) = guard.take() { + let session = { + let mut guard = self.session.lock().unwrap_or_else(|e| e.into_inner()); + guard.take() + }; + if let Some(session) = session { session.decode_task.abort(); session.dsp_task.abort(); + let (decode_result, dsp_result) = tokio::join!(session.decode_task, session.dsp_task); + // WHY: cancellation is the expected join outcome after abort(); + // a genuine task panic must still be surfaced in the log. + for result in [decode_result, dsp_result] { + if let Err(e) = result + && !e.is_cancelled() + { + warn!(error = %e, "pipeline task panicked during stop"); + } + } } - drop(guard); // WHY: send fails only when no receivers exist; dropping is intentional self.event_tx.send(EngineEvent::PlaybackStopped).ok(); @@ -561,6 +581,8 @@ async fn dsp_task_fn(params: DspTaskParams) { #[cfg(feature = "native-output")] let mut backend: Option = None; + #[cfg(feature = "native-output")] + let mut last_underrun_count: u64 = 0; loop { if state.load(Ordering::Relaxed) == STATE_STOPPED { @@ -765,6 +787,17 @@ async fn dsp_task_fn(params: DspTaskParams) { tokio::task::yield_now().await; } + // Poll the output backend's underrun counter (~ once per frame) and + // surface increases as EngineEvent::Underrun. + #[cfg(feature = "native-output")] + if let Some(b) = backend.as_ref() + && let Some(count) = underrun_increase(last_underrun_count, b.underrun_count()) + { + last_underrun_count = count; + // WHY: send fails only when no receivers exist; dropping is intentional + event_tx.send(EngineEvent::Underrun { count }).ok(); + } + // Throttle signal path updates to avoid watch channel spam (~4 Hz). if last_snapshot_update.elapsed() >= Duration::from_millis(250) { last_snapshot_update = Instant::now(); @@ -791,6 +824,20 @@ async fn dsp_task_fn(params: DspTaskParams) { } } +/// Returns the new cumulative count when the underrun counter increased, +/// `None` otherwise. Pure so the emission policy is unit-testable without +/// audio hardware. +#[cfg_attr( + not(any(feature = "native-output", test)), + expect( + dead_code, + reason = "polled from the native-output DSP loop; kept unconditional so the policy stays unit-tested in every build" + ) +)] +fn underrun_increase(previous: u64, current: u64) -> Option { + (current > previous).then_some(current) +} + /// Surfaces an asynchronous output-stream error: emits `EngineEvent::Error`, stops /// playback state, and emits `PlaybackStopped`. fn report_output_error( @@ -920,6 +967,14 @@ mod tests { config } + #[test] + fn underrun_increase_emits_only_on_growth() { + assert_eq!(underrun_increase(0, 0), None); + assert_eq!(underrun_increase(0, 3), Some(3)); + assert_eq!(underrun_increase(3, 3), None); + assert_eq!(underrun_increase(3, 7), Some(7)); + } + #[test] fn engine_new_succeeds_with_default_config() { let engine = Engine::new(EngineConfig::default()); @@ -954,7 +1009,7 @@ mod tests { "expected PlaybackStarted, got {evt:?}" ); - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -972,7 +1027,7 @@ mod tests { "expected AlreadyPlaying" ); - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -996,7 +1051,7 @@ mod tests { } } - engine.stop().unwrap(); + engine.stop().await.unwrap(); // Collect events, expect PlaybackStopped. let mut saw_stopped = false; @@ -1028,7 +1083,7 @@ mod tests { tokio::time::sleep(Duration::from_millis(10)).await; } - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -1049,7 +1104,7 @@ mod tests { // (With native-output disabled the DSP task still processes frames.) let _ = snap; // no assertion on content - just must not panic - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -1077,7 +1132,7 @@ mod tests { .unwrap(); assert!(matches!(evt, EngineEvent::PlaybackResumed)); - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -1089,6 +1144,25 @@ mod tests { assert!(snap.source.is_none()); } + /// stop() is a synchronization point: after it returns, the old session's + /// tasks are joined, so an immediate play() must always succeed — cycling + /// rapidly would previously race the CAS against un-aborted task teardown. + #[tokio::test] + async fn stop_then_immediate_play_cycles_cleanly() { + let engine = Engine::new(headless_config()).unwrap(); + let wav = make_wav(2, 44100, 2.0); + + for i in 0..5 { + engine + .play(AudioSource::File(wav.path().to_path_buf())) + .unwrap_or_else(|e| panic!("cycle {i}: play after stop must succeed: {e}")); + timeout(Duration::from_secs(5), engine.stop()) + .await + .expect("stop must not hang") + .unwrap(); + } + } + // --- #386: seek must reposition the decoder, not fabricate completion --- #[tokio::test] @@ -1194,7 +1268,7 @@ mod tests { ), } - engine.stop().unwrap(); + engine.stop().await.unwrap(); } #[tokio::test] @@ -1222,7 +1296,7 @@ mod tests { .expect("seek while paused must succeed"); assert!(actual <= Duration::from_secs(5), "landed at {actual:?}"); - engine.stop().unwrap(); + engine.stop().await.unwrap(); } // --- #387: oversized frame must error, not livelock --- @@ -1459,6 +1533,6 @@ mod tests { engine .play(AudioSource::File(wav2.path().to_path_buf())) .expect("engine must accept a new play() after a stream error"); - engine.stop().unwrap(); + engine.stop().await.unwrap(); } } diff --git a/crates/akouo-core/src/gapless/mod.rs b/crates/akouo-core/src/gapless/mod.rs index 4e6ba873..74cc7b3a 100644 --- a/crates/akouo-core/src/gapless/mod.rs +++ b/crates/akouo-core/src/gapless/mod.rs @@ -231,16 +231,18 @@ impl Default for CarryBuffer { pub struct GaplessScheduler { pre_buffer: PreBuffer, carry_buffer: Option, - transition_mode: TransitionMode, + // WHY: Option makes "user explicitly chose a mode" distinguishable from + // "never set" — None means auto-detect via AlbumDetector. + user_transition_mode: Option, prefetch_active: bool, } impl GaplessScheduler { - pub fn new(pre_buffer: PreBuffer, transition_mode: TransitionMode) -> Self { + pub fn new(pre_buffer: PreBuffer, user_transition_mode: Option) -> Self { Self { pre_buffer, carry_buffer: None, - transition_mode, + user_transition_mode, prefetch_active: false, } } @@ -278,25 +280,32 @@ impl GaplessScheduler { /// Selects the transition mode for moving FROM `current` to `next`. /// - /// Uses album detection to pick gapless automatically; falls back to - /// `self.transition_mode` only when overriding user preference takes precedence - /// (i.e., the user has explicitly SET Gap or Crossfade in settings - the caller - /// is responsible for deciding when to honour that over album detection). + /// An explicit user-set mode (via [`Self::set_transition_mode`]) takes + /// precedence; otherwise album detection picks gapless automatically. pub fn select_transition_mode( &self, current: &TrackMetadata, next: &TrackMetadata, ) -> TransitionMode { + if let Some(mode) = &self.user_transition_mode { + return mode.clone(); + } AlbumDetector::should_gapless(current, next) } - /// Overrides the default transition mode (used when the user changes settings). + /// Overrides transition selection (used when the user changes settings). pub fn set_transition_mode(&mut self, mode: TransitionMode) { - self.transition_mode = mode; + self.user_transition_mode = Some(mode); + } + + /// Clears the user override, returning to album-detection auto mode. + pub fn clear_transition_mode(&mut self) { + self.user_transition_mode = None; } - pub fn transition_mode(&self) -> &TransitionMode { - &self.transition_mode + /// The explicit user override, when one is set. + pub fn transition_mode(&self) -> Option<&TransitionMode> { + self.user_transition_mode.as_ref() } pub fn pre_buffer(&self) -> &PreBuffer { @@ -326,10 +335,7 @@ impl GaplessScheduler { impl Default for GaplessScheduler { fn default() -> Self { - Self::new( - PreBuffer::new(10.0, 44100 * 10 * 2), - TransitionMode::default(), - ) + Self::new(PreBuffer::new(10.0, 44100 * 10 * 2), None) } } @@ -621,10 +627,7 @@ mod tests { // ── GaplessScheduler ───────────────────────────────────────────────────── fn scheduler_with_threshold(threshold_secs: f64) -> GaplessScheduler { - GaplessScheduler::new( - PreBuffer::new(threshold_secs, 1000), - TransitionMode::default(), - ) + GaplessScheduler::new(PreBuffer::new(threshold_secs, 1000), None) } #[test] @@ -691,6 +694,26 @@ mod tests { ); } + #[test] + fn transition_mode_selection_respects_explicit_user_override() { + let mut sched = scheduler_with_threshold(10.0); + sched.set_transition_mode(TransitionMode::Crossfade { duration_ms: 1500 }); + // Same album would auto-select Gapless — the explicit override wins. + let current = track(Some("OK Computer"), Some("Radiohead"), Some(1)); + let next = track(Some("OK Computer"), Some("Radiohead"), Some(2)); + assert_eq!( + sched.select_transition_mode(¤t, &next), + TransitionMode::Crossfade { duration_ms: 1500 } + ); + + // Clearing the override returns to auto-detection. + sched.clear_transition_mode(); + assert_eq!( + sched.select_transition_mode(¤t, &next), + TransitionMode::Gapless + ); + } + // ── CarryBuffer ────────────────────────────────────────────────────────── #[test] diff --git a/crates/akouo-core/src/output/cpal.rs b/crates/akouo-core/src/output/cpal.rs index 42826561..8c96d739 100644 --- a/crates/akouo-core/src/output/cpal.rs +++ b/crates/akouo-core/src/output/cpal.rs @@ -4,6 +4,11 @@ use std::sync::{Arc, Mutex}; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; use tracing::warn; +// WHY: quantization lives in dsp::volume — a second local implementation +// drifted to an asymmetric formula (-1.0 mapped to i32::MIN here but +// -2_147_483_647 in the DSP path), so the output stage shares the one +// canonical symmetric-scale quantizer family. +use crate::dsp::volume::{quantize_i16, quantize_i32}; use crate::error::OutputError; #[cfg(target_os = "linux")] use crate::output::pipewire; @@ -15,6 +20,10 @@ use crate::output::{ pub struct CpalOutputBackend { host: cpal::Host, stream: Mutex>, + // WHY: the counter must outlive open() — the engine polls it to emit + // EngineEvent::Underrun; a counter local to open() was dropped + // immediately and the event could never fire. + underruns: Arc, #[cfg(target_os = "linux")] pipewire_rate_forced: Mutex, } @@ -24,10 +33,17 @@ impl CpalOutputBackend { Self { host: cpal::default_host(), stream: Mutex::new(None), + underruns: Arc::new(AtomicU64::new(0)), #[cfg(target_os = "linux")] pipewire_rate_forced: Mutex::new(false), } } + + /// Cumulative underrun count for the currently open stream. + #[must_use] + pub fn underrun_count(&self) -> u64 { + self.underruns.load(Ordering::Relaxed) + } } impl Default for CpalOutputBackend { @@ -170,8 +186,9 @@ impl OutputBackend for CpalOutputBackend { const MAX_SAMPLES: usize = 8192; let mut f64_buf = vec![0.0f64; MAX_SAMPLES]; let mut callback = data_callback; - let underruns = Arc::new(AtomicU64::new(0)); - let underruns_rt = Arc::clone(&underruns); + // Fresh stream, fresh counter. + self.underruns.store(0, Ordering::Relaxed); + let underruns_rt = Arc::clone(&self.underruns); let error_cb = make_stream_error_callback(device_name.clone(), error_tx); @@ -463,16 +480,6 @@ fn write_to_data(data: &mut cpal::Data, f64_src: &[f64], total_samples: usize, _ } } -#[inline(always)] -fn quantize_i32(s: f64) -> i32 { - (s * 2_147_483_648.0).clamp(-2_147_483_648.0, 2_147_483_647.0) as i32 -} - -#[inline(always)] -fn quantize_i16(s: f64) -> i16 { - (s * 32_768.0).clamp(-32_768.0, 32_767.0) as i16 -} - #[cfg(test)] mod tests { use super::*; @@ -527,29 +534,32 @@ mod tests { ); } + // NOTE: quantize_i16/quantize_i32 are the canonical dsp::volume + // symmetric-scale quantizers — -1.0 maps to -MAX (not MIN), matching + // the i16/i24 convention used throughout the DSP pipeline. #[test] fn quantize_i32_full_scale() { assert_eq!(quantize_i32(1.0), i32::MAX); - assert_eq!(quantize_i32(-1.0), i32::MIN); + assert_eq!(quantize_i32(-1.0), -i32::MAX); assert_eq!(quantize_i32(0.0), 0); } #[test] fn quantize_i32_clips() { assert_eq!(quantize_i32(2.0), i32::MAX); - assert_eq!(quantize_i32(-2.0), i32::MIN); + assert_eq!(quantize_i32(-2.0), -i32::MAX); } #[test] fn quantize_i16_full_scale() { assert_eq!(quantize_i16(1.0), i16::MAX); - assert_eq!(quantize_i16(-1.0), i16::MIN); + assert_eq!(quantize_i16(-1.0), -i16::MAX); assert_eq!(quantize_i16(0.0), 0); } #[test] fn quantize_i16_clips() { assert_eq!(quantize_i16(2.0), i16::MAX); - assert_eq!(quantize_i16(-2.0), i16::MIN); + assert_eq!(quantize_i16(-2.0), -i16::MAX); } } diff --git a/crates/akouo-core/src/queue.rs b/crates/akouo-core/src/queue.rs index 5a99b1ee..977be84b 100644 --- a/crates/akouo-core/src/queue.rs +++ b/crates/akouo-core/src/queue.rs @@ -47,11 +47,14 @@ impl PlayQueue { reason = "PlayQueue::next() advances the queue and returns the track; naming matches domain language, not Iterator" )] pub fn next(&mut self) -> Option { - if let Some(current) = self.current().map(PathBuf::from) { - self.history.push(current); - } let next_index = self.current_index + 1; if next_index < self.tracks.len() { + // WHY: history records only completed advances — pushing before + // the bounds check grew history without bound when callers poll + // next() repeatedly at end-of-queue. + if let Some(current) = self.current().map(PathBuf::from) { + self.history.push(current); + } self.current_index = next_index; self.current().map(PathBuf::from) } else { @@ -182,6 +185,18 @@ mod tests { assert_eq!(q.history()[1], PathBuf::from("b.flac")); } + #[test] + fn next_at_end_of_queue_does_not_grow_history_on_repeated_calls() { + let mut q = PlayQueue::from_tracks(paths(&["a.flac"])); + assert!(q.next().is_none()); + assert!(q.next().is_none()); + assert!(q.next().is_none()); + assert!( + q.history().is_empty(), + "failed advances must not accumulate history entries" + ); + } + #[test] fn push_appends_to_queue() { let mut q = PlayQueue::new(); diff --git a/crates/apotheke/src/repo/book.rs b/crates/apotheke/src/repo/book.rs index 6e766ed1..f8296df9 100644 --- a/crates/apotheke/src/repo/book.rs +++ b/crates/apotheke/src/repo/book.rs @@ -92,6 +92,31 @@ pub async fn list_books(pool: &SqlitePool, limit: i64, offset: i64) -> Result Result, DbError> { + sqlx::query_scalar( + "SELECT DISTINCT mr.display_name + FROM book_authors ba + JOIN media_registry mr ON mr.id = ba.person_id + WHERE ba.role = 'author' + ORDER BY mr.display_name LIMIT ? OFFSET ?", + ) + .bind(limit) + .bind(offset) + .fetch_all(pool) + .await + .context(QuerySnafu { + table: "book_authors", + }) +} + pub async fn update_book( pool: &SqlitePool, id: &[u8], diff --git a/crates/apotheke/src/repo/user.rs b/crates/apotheke/src/repo/user.rs index 2db3cb62..a9aee7a9 100644 --- a/crates/apotheke/src/repo/user.rs +++ b/crates/apotheke/src/repo/user.rs @@ -155,6 +155,33 @@ pub async fn list_users(pool: &SqlitePool, limit: i64, offset: i64) -> Result Result, DbError> { + sqlx::query_as::<_, User>( + "SELECT id, username, display_name, password_hash, role, is_active, + created_at, last_login_at + FROM users WHERE is_active = 1 ORDER BY username LIMIT ? OFFSET ?", + ) + .bind(limit) + .bind(offset) + .fetch_all(pool) + .await + .context(QuerySnafu { table: "users" }) +} + +pub async fn count_active_users(pool: &SqlitePool) -> Result { + sqlx::query_scalar("SELECT COUNT(*) FROM users WHERE is_active = 1") + .fetch_one(pool) + .await + .context(QuerySnafu { table: "users" }) +} + pub async fn update_user( pool: &SqlitePool, id: &[u8], diff --git a/crates/archon/src/migrate.rs b/crates/archon/src/migrate.rs index 2e36908e..d72711e9 100644 --- a/crates/archon/src/migrate.rs +++ b/crates/archon/src/migrate.rs @@ -165,11 +165,23 @@ fn migrate_blocking( match std::fs::rename(&file_path, &canonical_abs) { Ok(()) => {} Err(e) if e.raw_os_error() == Some(18) => { - // EXDEV: cross-device move — fall back to copy + delete. - std::fs::copy(&file_path, &canonical_abs).with_context(|_| MigrateIoSnafu { + // EXDEV: cross-device move — staged copy, then rename, then + // delete the source. A crash mid-sequence leaves either a + // `.migrating` temp (cleanable, never scanned as media) or + // a completed destination + intact source (re-running the + // migration is idempotent); never two live library copies. + let staging = canonical_abs.with_extension("migrating"); + std::fs::copy(&file_path, &staging).with_context(|_| MigrateIoSnafu { operation: format!( "copy (cross-device) {} -> {}", file_path.display(), + staging.display() + ), + })?; + std::fs::rename(&staging, &canonical_abs).with_context(|_| MigrateIoSnafu { + operation: format!( + "rename staged copy {} -> {}", + staging.display(), canonical_abs.display() ), })?; @@ -195,12 +207,34 @@ fn migrate_blocking( } // Write sidecar if it doesn't already exist. - if let (Some(sc_path), Some(content)) = (sidecar_abs, sidecar) - && !sc_path.exists() - { - std::fs::write(&sc_path, content).with_context(|_| MigrateIoSnafu { - operation: format!("write sidecar {}", sc_path.display()), - })?; + // + // WHY: create_new makes the OS enforce first-writer-wins — the old + // exists()-then-write pair raced a concurrent writer and clobbered + // whichever sidecar landed first. + if let (Some(sc_path), Some(content)) = (sidecar_abs, sidecar) { + use std::io::Write; + match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&sc_path) + { + Ok(mut file) => { + file.write_all(content.as_bytes()) + .with_context(|_| MigrateIoSnafu { + operation: format!("write sidecar {}", sc_path.display()), + })?; + } + // An existing sidecar is authoritative — matching the old + // intended skip semantics. + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(e) => { + return Err(HostError::MigrateIo { + operation: format!("create sidecar {}", sc_path.display()), + source: e, + location: snafu::location!(), + }); + } + } } report.processed += 1; diff --git a/crates/archon/src/play.rs b/crates/archon/src/play.rs index 2c5cf0c0..05c3ff7d 100644 --- a/crates/archon/src/play.rs +++ b/crates/archon/src/play.rs @@ -31,7 +31,7 @@ pub async fn run_play(args: PlayArgs, out: &mut impl Write) -> Result<(), HostEr } } - engine.stop().context(AudioEngineSnafu)?; + engine.stop().await.context(AudioEngineSnafu)?; Ok(()) } diff --git a/crates/archon/src/render/mod.rs b/crates/archon/src/render/mod.rs index 80c8873a..fa1857cd 100644 --- a/crates/archon/src/render/mod.rs +++ b/crates/archon/src/render/mod.rs @@ -33,12 +33,17 @@ pub struct RenderArgs { pub config_path: Option, } +// WHY: the kernel exposes the hostname as a file — reading it avoids the +// old `hostname` subprocess, whose result depended on $PATH and spawn cost. fn default_renderer_name() -> String { - std::process::Command::new("hostname") - .output() - .ok() - .and_then(|o| String::from_utf8(o.stdout).ok()) - .map(|s| s.trim().to_string()) + ["/proc/sys/kernel/hostname", "/etc/hostname"] + .iter() + .find_map(|path| { + std::fs::read_to_string(path) + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + }) .unwrap_or_else(|| "harmonia-renderer".to_string()) } @@ -149,3 +154,61 @@ fn pin_first_seen_server( info!("pinned server certificate fingerprint on first discovery (trust-on-first-use)"); Ok(new_creds) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_renderer_name_is_non_empty_without_subprocess() { + let name = default_renderer_name(); + assert!(!name.trim().is_empty()); + } + + fn discovered(fingerprint: Option<&str>) -> discovery::DiscoveredServer { + discovery::DiscoveredServer { + instance_name: "Harmonia Test".to_string(), + addr: "127.0.0.1:4433".parse().expect("loopback addr"), + server_id: Some("srv-1".to_string()), + cert_fingerprint: fingerprint.map(str::to_string), + protocol_version: Some("1".to_string()), + } + } + + #[test] + fn pin_first_seen_server_persists_tofu_credentials() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let fp = "ab".repeat(32); + + let creds = + pin_first_seen_server(dir.path(), &discovered(Some(&fp))).expect("pin succeeds"); + assert_eq!(creds.api_key, "", "no key exists before pairing"); + assert_eq!(creds.server_fingerprint, fp); + assert_eq!(creds.server_name, "Harmonia Test"); + assert!( + creds.paired_at.ends_with('Z') && creds.paired_at.contains('T'), + "paired_at must be an ISO-8601 UTC timestamp, got {}", + creds.paired_at + ); + + // Round-trips through the credential store. + let loaded = credentials::load_credentials(dir.path()) + .expect("load succeeds") + .expect("credentials persisted"); + assert_eq!(loaded.server_fingerprint, fp); + assert_eq!(loaded.server_name, "Harmonia Test"); + } + + #[test] + fn pin_first_seen_server_fails_closed_without_fingerprint() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let result = pin_first_seen_server(dir.path(), &discovered(None)); + assert!(result.is_err(), "no fingerprint must not pin anything"); + assert!( + credentials::load_credentials(dir.path()) + .expect("load succeeds") + .is_none(), + "a failed pin must not persist credentials" + ); + } +} diff --git a/crates/archon/src/render/pipeline.rs b/crates/archon/src/render/pipeline.rs index 7c1ca8b2..13805548 100644 --- a/crates/archon/src/render/pipeline.rs +++ b/crates/archon/src/render/pipeline.rs @@ -30,6 +30,9 @@ pub struct RenderPipeline { underrun_count: Arc, device_name: Option, output_config: PipelineOutputConfig, + // Stream parameters cached at open_output so drain() can convert the + // remaining sample count into real time. + stream_rate: Option<(u32, u16)>, } struct PipelineOutputConfig { @@ -63,6 +66,7 @@ impl RenderPipeline { exclusive_mode: config.output.exclusive_mode, bit_depth: config.output.bit_depth, }, + stream_rate: None, }) } @@ -185,6 +189,7 @@ impl RenderPipeline { location: snafu::location!(), })?; + self.stream_rate = Some((sample_rate, channels)); info!( sample_rate, channels, @@ -209,11 +214,20 @@ impl RenderPipeline { } /// Drains remaining audio FROM the ring buffer before shutdown. + /// + /// Sleeps for the buffered audio's real-time duration (capped) instead of + /// a fixed interval — a fixed sleep truncated deep buffers and overslept + /// shallow ones. pub async fn drain(&self) { let remaining = self.ring.available_to_read(); if remaining > 0 { - info!(remaining_samples = remaining, "draining audio buffer"); - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let wait = drain_duration(remaining, self.stream_rate); + info!( + remaining_samples = remaining, + wait_ms = wait.as_millis(), + "draining audio buffer" + ); + tokio::time::sleep(wait).await; } } @@ -227,10 +241,49 @@ impl RenderPipeline { } } +/// Real-time duration of `remaining` interleaved samples at the cached stream +/// rate, capped at 5 seconds. Falls back to 200 ms when no stream has opened +/// (nothing was ever pushed at a known rate). +fn drain_duration(remaining: usize, stream_rate: Option<(u32, u16)>) -> Duration { + const FALLBACK: Duration = Duration::from_millis(200); + const CAP: Duration = Duration::from_secs(5); + match stream_rate { + Some((sample_rate, channels)) if sample_rate > 0 && channels > 0 => { + let per_second = f64::from(sample_rate) * f64::from(channels); + Duration::from_secs_f64(remaining as f64 / per_second).min(CAP) + } + _ => FALLBACK, + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn drain_duration_scales_with_remaining_samples() { + // 44100 Hz stereo: 88200 samples = 1 second. + assert_eq!( + drain_duration(88_200, Some((44_100, 2))), + Duration::from_secs(1) + ); + assert_eq!( + drain_duration(44_100, Some((44_100, 2))), + Duration::from_millis(500) + ); + // Capped for absurd depths. + assert_eq!( + drain_duration(usize::MAX, Some((44_100, 2))), + Duration::from_secs(5) + ); + // Unknown or degenerate rates fall back. + assert_eq!(drain_duration(1024, None), Duration::from_millis(200)); + assert_eq!( + drain_duration(1024, Some((0, 2))), + Duration::from_millis(200) + ); + } + #[test] fn buffer_depth_calculation() { let config = RendererConfig::default(); diff --git a/crates/archon/src/render/playout.rs b/crates/archon/src/render/playout.rs index 879dad94..587ec031 100644 --- a/crates/archon/src/render/playout.rs +++ b/crates/archon/src/render/playout.rs @@ -82,9 +82,13 @@ impl PlayoutPipeline { // WHY: Convert server playout timestamp to local time by subtracting the // server's clock OFFSET. If server clock is ahead (positive OFFSET), - // local playout time is earlier. - let local_playout = (i64::try_from(frame.playout_ts).unwrap_or_default() // WHY: audio playout timestamps fit comfortably in i64 - - self.clock_offset_us) as u64; + // local playout time is earlier. The subtraction stays in i64 space and + // clamps at zero — a negative result cast to u64 wrapped to a huge + // value and turned an already-due frame INTO a near-infinite Hold. + let local_playout_i64 = i64::try_from(frame.playout_ts) + .unwrap_or_default() // WHY: audio playout timestamps fit comfortably in i64 + .saturating_sub(self.clock_offset_us); + let local_playout = u64::try_from(local_playout_i64).unwrap_or(0); if local_now_us >= local_playout { let late_by = local_now_us - local_playout; @@ -272,6 +276,23 @@ mod tests { assert_eq!(pipe.evaluate(1_000_000), Some(PlayoutDecision::Play)); } + #[test] + fn clock_offset_exceeding_playout_ts_does_not_wrap_to_giant_hold() { + let mut pipe = PlayoutPipeline::new(); + // Server clock far ahead: local playout would be negative — the frame + // is already due, never Hold with a near-u64::MAX wait. + pipe.set_clock_offset(2_000_000); + pipe.enqueue(frame(0, 100_000)); + let decision = pipe.evaluate(1_000_000); + assert!( + matches!( + decision, + Some(PlayoutDecision::Play | PlayoutDecision::Late { .. }) + ), + "negative local playout must be treated as due, got {decision:?}" + ); + } + #[test] fn process_returns_ready_frames() { let mut pipe = PlayoutPipeline::new(); diff --git a/crates/archon/src/render/runner.rs b/crates/archon/src/render/runner.rs index bd103de9..6ddab22f 100644 --- a/crates/archon/src/render/runner.rs +++ b/crates/archon/src/render/runner.rs @@ -230,7 +230,7 @@ async fn connect_and_run( let stream_sample_rate = accept.sample_rate; let stream_channels = accept.channels; - let mut audio_task = tokio::spawn( + let audio_task = tokio::spawn( async move { receive_audio( audio_recv, @@ -248,36 +248,76 @@ async fn connect_and_run( let status_report = Arc::clone(&status); let shutdown_status = shutdown.child_token(); - let mut status_task = tokio::spawn( + let status_task = tokio::spawn( async move { send_status_reports(ctrl_send, &status_report, shutdown_status).await } .instrument(tracing::info_span!("status_report")), ); - tokio::select! { + // WHY: handles live in Options so the reap below never re-polls a handle + // a select branch already consumed (polling a completed JoinHandle panics). + let mut audio_task = Some(audio_task); + let mut status_task = Some(status_task); + + let outcome = tokio::select! { biased; _ = shutdown.cancelled() => { info!("shutdown requested, draining"); + Ok(()) } - result = &mut audio_task => { + // INVARIANT: the branch's Option is Some here — it is only taken by + // the branch itself, immediately after completion. + result = async { audio_task.as_mut().expect("audio task present").await } => { + audio_task = None; // WHY: the sibling task would otherwise outlive this connection // attempt as a detached leak; the reconnect loop spawns fresh tasks. - status_task.abort(); - if let Err(e) = join_outcome("audio", result) { - status.set_device_state(DeviceState::Stopped); - return Err(e); + if let Some(t) = &status_task { + t.abort(); } + join_outcome("audio", result) } - result = &mut status_task => { - audio_task.abort(); - if let Err(e) = join_outcome("status", result) { - status.set_device_state(DeviceState::Stopped); - return Err(e); + result = async { status_task.as_mut().expect("status task present").await } => { + status_task = None; + if let Some(t) = &audio_task { + t.abort(); } + join_outcome("status", result) } - } + }; + + // WHY: no task may survive past this connection attempt — a loser (or the + // shutdown path's pair) left running would double-process the next + // connection's device. Reap = bounded graceful wait, then abort + join. + reap_task("audio", audio_task).await; + reap_task("status", status_task).await; status.set_device_state(DeviceState::Stopped); - Ok(()) + outcome +} + +/// Waits briefly for a still-running task to exit on its own (cancellation +/// tokens fire before this is called), then aborts and joins it so the caller +/// is guaranteed no orphan survives. +async fn reap_task( + task: &'static str, + handle: Option>>, +) { + const GRACEFUL_EXIT_WINDOW: Duration = Duration::from_secs(5); + let Some(mut handle) = handle else { + return; + }; + if let Ok(result) = tokio::time::timeout(GRACEFUL_EXIT_WINDOW, &mut handle).await { + // WHY: the select winner already carries the connection outcome; a + // straggler's error is log-only here. + join_outcome(task, result).ok(); + return; + } + warn!(task, "renderer task did not exit in time; aborting"); + handle.abort(); + if let Err(e) = handle.await + && !e.is_cancelled() + { + warn!(task, error = %e, "renderer task panicked during abort"); + } } /// Classifies a joined renderer task result at the handling site. @@ -668,4 +708,29 @@ mod connect_and_run_tests { "a failed audio task must propagate so the reconnect backoff engages" ); } + #[tokio::test(start_paused = true)] + async fn reap_task_aborts_a_wedged_task_and_returns() { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + let started = Arc::new(AtomicBool::new(false)); + let started_task = Arc::clone(&started); + let wedged = tokio::spawn(async move { + started_task.store(true, Ordering::SeqCst); + std::future::pending::<()>().await; + Ok::<(), RenderError>(()) + }); + + // Bounded even for a task that never exits on its own. + tokio::time::timeout(Duration::from_secs(30), reap_task("wedged", Some(wedged))) + .await + .expect("reap_task must be bounded"); + assert!(started.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn reap_task_tolerates_consumed_handle() { + // A select-winner branch leaves None behind; reap must be a no-op. + reap_task("consumed", None).await; + } } diff --git a/crates/archon/src/render/server.rs b/crates/archon/src/render/server.rs index 18a7fcea..6f8969f5 100644 --- a/crates/archon/src/render/server.rs +++ b/crates/archon/src/render/server.rs @@ -209,6 +209,7 @@ async fn handle_renderer_connection( message: e.to_string(), location: snafu::location!(), })?; + validate_session_init(&init)?; info!(name = %init.name, version = init.protocol_version, "session init received"); // INVARIANT: no session state (session_id, registry entry, streams) exists @@ -248,6 +249,14 @@ async fn handle_renderer_connection( }) .await; + // WHY: the guard's Drop removes the entry even when run_renderer_session + // panics and unwinds — the spawn wrapper only logs the JoinError, so + // without the guard a panicking session leaked a stale registry entry. + let _cleanup = RegistryCleanupGuard { + registry: Arc::clone(®istry), + session_id: session_id.clone(), + }; + // INVARIANT: every fallible operation between add and remove lives inside // run_renderer_session, so no `?` can return past the registry cleanup. let result = run_renderer_session( @@ -269,6 +278,29 @@ async fn handle_renderer_connection( result } +/// Removes the session's registry entry on drop, covering panic unwinds. +/// Removal is idempotent, so the normal-path explicit remove and this guard +/// coexist safely. +struct RegistryCleanupGuard { + registry: Arc, + session_id: RendererSessionId, +} + +impl Drop for RegistryCleanupGuard { + fn drop(&mut self) { + let registry = Arc::clone(&self.registry); + let session_id = self.session_id.clone(); + // WHY: RendererRegistry::remove is async (RwLock) and Drop is sync — + // spawn the removal; outside a runtime (process teardown) the entry + // dies with the registry anyway. + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + registry.remove(&session_id).await; + }); + } + } +} + /// Session body between registry add and remove: opens the audio stream and /// consumes status reports until disconnection. async fn run_renderer_session( @@ -306,12 +338,35 @@ async fn read_status_loop( registry.update_status(session_id, status).await; } } + // WHY: a protocol violation (oversized/malformed frame) is a real + // failure the caller's warn! must surface; only transport-level + // termination (peer closed, stream ended) is a clean disconnect. + Err(e @ RenderError::Protocol { .. }) => return Err(e), Err(_) => break, } } Ok(()) } +/// Longest renderer name stored/logged verbatim. Anything longer is a +/// protocol violation, not a display concern. +const MAX_RENDERER_NAME_BYTES: usize = 128; + +// WHY: SessionInit arrives from an unauthenticated peer — every field it +// carries into the registry/logs needs a bound before it is stored. +fn validate_session_init(init: &SessionInit) -> Result<(), RenderError> { + if init.name.len() > MAX_RENDERER_NAME_BYTES { + return Err(RenderError::Protocol { + message: format!( + "renderer name of {} bytes exceeds the {MAX_RENDERER_NAME_BYTES}-byte bound", + init.name.len() + ), + location: snafu::location!(), + }); + } + Ok(()) +} + // INVARIANT: registration requires a configured, non-empty server-side key; // an absent key rejects every peer (fail closed), never accept-all. fn api_key_matches(expected: Option<&str>, presented: &str) -> bool { @@ -422,6 +477,62 @@ mod tests { assert!(id.chars().all(|c| c.is_ascii_hexdigit())); } + #[test] + fn session_init_name_over_bound_is_rejected() { + let long = SessionInit { + name: "x".repeat(MAX_RENDERER_NAME_BYTES + 1), + protocol_version: 1, + api_key: String::new(), + }; + assert!(matches!( + validate_session_init(&long), + Err(RenderError::Protocol { .. }) + )); + + let ok = SessionInit { + name: "x".repeat(MAX_RENDERER_NAME_BYTES), + protocol_version: 1, + api_key: String::new(), + }; + assert!(validate_session_init(&ok).is_ok()); + } + + #[tokio::test] + async fn cleanup_guard_removes_entry_on_panic_unwind() { + let registry = Arc::new(RendererRegistry::new()); + let session_id = RendererSessionId("panicky".into()); + registry + .add(ConnectedRenderer { + name: "p".into(), + session_id: session_id.clone(), + connected_at: Instant::now(), + last_status: None, + }) + .await; + + let registry_task = Arc::clone(®istry); + let sid = session_id.clone(); + let task = tokio::spawn(async move { + let _guard = RegistryCleanupGuard { + registry: registry_task, + session_id: sid, + }; + panic!("simulated session panic"); + }); + assert!(task.await.is_err(), "task must have panicked"); + + // The guard spawns the removal; give it a bounded window to land. + let mut cleaned = false; + for _ in 0..100 { + if registry.list().await.is_empty() { + cleaned = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!(cleaned, "panic unwind must not leak the registry entry"); + } + #[test] fn api_key_matches_only_on_exact_configured_key() { assert!(api_key_matches(Some("key-123"), "key-123")); diff --git a/crates/archon/src/serve.rs b/crates/archon/src/serve.rs index 6f9e957a..0ea7d9d5 100644 --- a/crates/archon/src/serve.rs +++ b/crates/archon/src/serve.rs @@ -227,7 +227,7 @@ fn parse_search_media_type(media_type: &str) -> Result Ok(zetesis::SearchMediaType::Movie), "music" | "album" | "music_album" => Ok(zetesis::SearchMediaType::Music), "book" | "books" | "audiobook" | "comic" => Ok(zetesis::SearchMediaType::Book), - other => Err(ServiceError::Internal(format!( + other => Err(ServiceError::InvalidInput(format!( "unsupported search media_type: {other}" ))), } @@ -1376,9 +1376,11 @@ mod search_adapter_tests { } #[test] - fn parse_search_media_type_rejects_unknown_values() { + fn parse_search_media_type_rejects_unknown_values_as_invalid_input() { + // WHY: user-supplied media_type must map to a 400-class error, never + // fold into Internal (which the HTTP layer reports as a 500). let error = parse_search_media_type("podcast").expect_err("unsupported media type"); - assert!(matches!(error, ServiceError::Internal(_))); + assert!(matches!(error, ServiceError::InvalidInput(_))); } #[tokio::test] @@ -1422,6 +1424,201 @@ mod tests { use super::*; + async fn test_pools() -> Arc { + let pool = SqlitePool::connect("sqlite::memory:") + .await + .expect("in-memory sqlite opens"); + MIGRATOR.run(&pool).await.expect("migrations run"); + Arc::new(apotheke::DbPools { + read: pool.clone(), + write: pool, + }) + } + + #[tokio::test] + async fn role_of_rejects_inactive_user() { + let db = test_pools().await; + let user_id = themelion::UserId::new(); + let user = apotheke::repo::user::User { + id: user_id.as_bytes().to_vec(), + username: "dormant".to_string(), + display_name: "Dormant".to_string(), + password_hash: "x".to_string(), + role: "member".to_string(), + is_active: 0, + created_at: "2026-01-01T00:00:00Z".to_string(), + last_login_at: None, + }; + apotheke::repo::user::insert_user(&db.write, &user) + .await + .expect("insert user"); + + let provider = RequestRoleProvider { db }; + let result = provider.role_of(user_id).await; + assert!( + matches!( + result, + Err(aitesis::AitesisError::InsufficientPermission { .. }) + ), + "inactive user must be rejected, got {result:?}" + ); + } + + fn media_request(media_type: themelion::MediaType) -> aitesis::MediaRequest { + aitesis::MediaRequest { + id: themelion::RequestId::new(), + user_id: themelion::UserId::new(), + media_type, + title: "Kind of Blue".to_string(), + external_id: None, + status: aitesis::RequestStatus::Approved, + decided_by: None, + decided_at: None, + deny_reason: None, + want_id: None, + created_at: jiff::Timestamp::now(), + } + } + + #[tokio::test] + async fn create_want_persists_want_with_first_quality_profile() { + let db = test_pools().await; + // WHY: migrations seed default profiles — the monitor must select the + // first music profile by name, so the expectation derives from the + // same repo query rather than a hand-seeded row. + let profile_id = apotheke::repo::quality::list_profiles_for_type(&db.read, "music") + .await + .expect("profiles query") + .into_iter() + .next() + .expect("a default music quality profile exists") + .id; + + let monitor = RequestMonitor { + db: Arc::clone(&db), + }; + let request = media_request(themelion::MediaType::Music); + let want_id = monitor.create_want(&request).await.expect("want created"); + + #[derive(sqlx::FromRow)] + struct WantRow { + media_type: String, + title: String, + quality_profile_id: i64, + status: String, + source: Option, + } + let row = sqlx::query_as::<_, WantRow>( + "SELECT media_type, title, quality_profile_id, status, source FROM wants WHERE id = ?", + ) + .bind(want_id.as_bytes().to_vec()) + .fetch_one(&db.read) + .await + .expect("persisted want row"); + + assert_eq!(row.media_type, "music_album"); + assert_eq!(row.title, "Kind of Blue"); + assert_eq!(row.quality_profile_id, profile_id); + assert_eq!(row.status, "searching"); + assert_eq!(row.source.as_deref(), Some("request")); + } + + #[tokio::test] + async fn create_want_rejects_news_media_type() { + let db = test_pools().await; + let monitor = RequestMonitor { db }; + let request = media_request(themelion::MediaType::News); + let result = monitor.create_want(&request).await; + assert!( + matches!( + result, + Err(aitesis::AitesisError::MediaIdentityInvalid { .. }) + ), + "news must be rejected, got {result:?}" + ); + } + + #[test] + fn request_media_types_maps_each_variant() { + assert_eq!( + request_media_types(themelion::MediaType::Music), + Some(("music_album", "music")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Audiobook), + Some(("audiobook", "audiobook")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Book), + Some(("book", "book")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Comic), + Some(("comic", "comic")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Podcast), + Some(("podcast", "podcast")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Movie), + Some(("movie", "movie")) + ); + assert_eq!( + request_media_types(themelion::MediaType::Tv), + Some(("tv_series", "tv")) + ); + assert_eq!(request_media_types(themelion::MediaType::News), None); + } + + fn config_with_download_dir(dir: PathBuf) -> horismos::Config { + let mut config = horismos::Config::default(); + config.ergasia.download_dir = dir; + config + } + + #[test] + fn validate_download_dir_rejects_missing_dir() { + let config = config_with_download_dir(PathBuf::from("/nonexistent/harmonia-dl")); + let error = validate_download_dir(&config).expect_err("missing dir must fail"); + assert!(error.to_string().contains("does not exist"), "{error}"); + } + + #[test] + fn validate_download_dir_rejects_unwritable_dir() { + use std::os::unix::fs::PermissionsExt; + + // WHY: root bypasses permission bits — the assertion would be + // meaningless, so the case is skipped for uid 0. + let dir = tempfile::TempDir::new().expect("tempdir"); + let probe = dir.path().join(".probe"); + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o555)) + .expect("chmod"); + if std::fs::write(&probe, b"").is_ok() { + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o755)) + .expect("chmod back"); + return; // running as root (or an ACL grants write) — not testable + } + + let config = config_with_download_dir(dir.path().to_path_buf()); + let error = validate_download_dir(&config).expect_err("unwritable dir must fail"); + assert!(error.to_string().contains("not writable"), "{error}"); + + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o755)) + .expect("chmod back for cleanup"); + } + + #[test] + fn validate_download_dir_accepts_writable_dir_and_cleans_up() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let config = config_with_download_dir(dir.path().to_path_buf()); + validate_download_dir(&config).expect("writable dir passes"); + assert!( + !dir.path().join(".harmonia-write-test").exists(), + "the write-probe file must be cleaned up" + ); + } + #[test] fn resolve_listen_addr_accepts_ipv4() { let addr = resolve_listen_addr("0.0.0.0", 4433).expect("ipv4 wildcard parses"); diff --git a/crates/horismos/src/subsystems.rs b/crates/horismos/src/subsystems.rs index b0dab354..16444280 100644 --- a/crates/horismos/src/subsystems.rs +++ b/crates/horismos/src/subsystems.rs @@ -60,6 +60,15 @@ pub struct ParocheConfig { /// Leaving it unset rejects every renderer registration (fail closed). #[serde(default)] pub renderer_api_key: Option, + /// Whether the KOSync endpoint accepts anonymous self-registration. + /// KOReader clients self-register by protocol; operators exposing the + /// server beyond a trusted network can close the abuse surface here. + #[serde(default = "default_true")] + pub kosync_registration_enabled: bool, +} + +fn default_true() -> bool { + true } impl std::fmt::Debug for ParocheConfig { @@ -74,6 +83,10 @@ impl std::fmt::Debug for ParocheConfig { "renderer_api_key", &self.renderer_api_key.as_ref().map(|_| "[redacted]"), ) + .field( + "kosync_registration_enabled", + &self.kosync_registration_enabled, + ) .finish() } } @@ -87,6 +100,7 @@ impl Default for ParocheConfig { transcode_concurrency: 2, opds_page_size: 50, renderer_api_key: None, + kosync_registration_enabled: true, } } } diff --git a/crates/paroche/src/discovery/advertise.rs b/crates/paroche/src/discovery/advertise.rs index 1e39b8aa..c1048909 100644 --- a/crates/paroche/src/discovery/advertise.rs +++ b/crates/paroche/src/discovery/advertise.rs @@ -67,7 +67,7 @@ impl AdvertisedService { let fullname_check = service_fullname.clone(); let mut tx_opt = Some(tx); - tokio::spawn( + let monitor_task = tokio::spawn( async move { loop { match monitor.recv_async().await { @@ -96,10 +96,17 @@ impl AdvertisedService { .instrument(info_span!("mdns.announce_monitor")), ); - tokio::time::timeout(std::time::Duration::from_secs(5), rx) + // WHY: the monitor loop only exits on Announce/Error/closed-channel; + // when the confirmation wait fails the task must be aborted or it + // outlives start() and leaks, looping on daemon events forever. + let confirmed = tokio::time::timeout(std::time::Duration::from_secs(5), rx) .await - .map_err(|_| "mDNS registration timed out".to_string())? - .map_err(|_| "mDNS registration channel dropped".to_string())?; + .map_err(|_| "mDNS registration timed out".to_string()) + .and_then(|recv| recv.map_err(|_| "mDNS registration channel dropped".to_string())); + if let Err(e) = confirmed { + monitor_task.abort(); + return Err(e); + } info!( instance = %params.instance_name, diff --git a/crates/paroche/src/error.rs b/crates/paroche/src/error.rs index e067d977..d616f9c9 100644 --- a/crates/paroche/src/error.rs +++ b/crates/paroche/src/error.rs @@ -96,6 +96,9 @@ impl From for ParocheError { match error { crate::state::ServiceError::NotFound => ParocheError::NotFound, crate::state::ServiceError::NotAvailable => ParocheError::Unavailable, + crate::state::ServiceError::InvalidInput(message) => { + ParocheError::Validation { message } + } crate::state::ServiceError::Internal(message) => { // WHY: the HTTP body carries only a correlation id; the detail // must land in the log here or it is lost entirely. diff --git a/crates/paroche/src/opds/catalog/mod.rs b/crates/paroche/src/opds/catalog/mod.rs index 1930d7a3..518092a6 100644 --- a/crates/paroche/src/opds/catalog/mod.rs +++ b/crates/paroche/src/opds/catalog/mod.rs @@ -510,6 +510,62 @@ pub async fn shelf_v2( publications, })) } + "authors" => { + let page = pq.page.max(1); + // INVARIANT: opds_page_size is a config usize (default 50); all as-casts here are safe + let page_size_usize = state.config.paroche.opds_page_size; + let page_size = page_size_usize as i64; + let offset = ((page - 1) * page_size_usize as u64) as i64; + + let mut authors = + apotheke::repo::book::list_authors(&state.db.read, page_size + 1, offset).await?; + let has_next = authors.len() > page_size_usize; + authors.truncate(page_size_usize); + + let mut links = vec![ + OpdsLink::new( + "self", + format!("/opds/v2/shelf/authors?page={page}"), + MIME_OPDS_V2, + ), + OpdsLink::new("start", "/opds/v2/catalog", MIME_OPDS_V2), + ]; + if has_next { + links.push(OpdsLink::new( + "next", + format!("/opds/v2/shelf/authors?page={}", page + 1), + MIME_OPDS_V2, + )); + } + + let count = authors.len() as u64; + // WHY: authors are a navigation tier, not publications — each + // entry links to the search feed scoped to that author's name. + let navigation: Vec<_> = authors + .into_iter() + .map(|name| NavigationLink { + href: format!( + "/opds/v2/search?q={}", + crate::opds::search::urlencoded(&name) + ), + title: name, + link_type: MIME_OPDS_V2.to_string(), + rel: "subsection".to_string(), + }) + .collect(); + + Ok(OpdsV2Response(OpdsFeed { + metadata: FeedMetadata { + title: "Authors".to_string(), + number_of_items: Some(count), + items_per_page: Some(page_size_usize as u64), + current_page: Some(page), + }, + links, + navigation, + publications: vec![], + })) + } _ => Err(ParocheError::NotFound), } } diff --git a/crates/paroche/src/opds/catalog/tests.rs b/crates/paroche/src/opds/catalog/tests.rs index 555d0b92..4ad018e1 100644 --- a/crates/paroche/src/opds/catalog/tests.rs +++ b/crates/paroche/src/opds/catalog/tests.rs @@ -442,3 +442,112 @@ async fn books_v1_unauthenticated_returns_401() { .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); } + +async fn insert_authors(state: &AppState, n: usize) { + for i in 0..n { + let person_id = uuid::Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO media_registry (id, entity_type, display_name) + VALUES (?, 'person', ?)", + ) + .bind(&person_id) + .bind(format!("Author {i:04}")) + .execute(&state.db.write) + .await + .unwrap(); + + let book_id = uuid::Uuid::now_v7().as_bytes().to_vec(); + sqlx::query("INSERT INTO books (id, title) VALUES (?, ?)") + .bind(&book_id) + .bind(format!("Authored Book {i:04}")) + .execute(&state.db.write) + .await + .unwrap(); + sqlx::query("INSERT INTO book_authors (book_id, person_id, role) VALUES (?, ?, 'author')") + .bind(&book_id) + .bind(&person_id) + .execute(&state.db.write) + .await + .unwrap(); + } +} + +#[tokio::test] +async fn shelf_authors_lists_author_navigation() { + let (state, auth) = test_state().await; + let token = admin_token(&auth).await; + insert_authors(&state, 3).await; + let app = opds_routes().with_state(state); + let resp = app + .oneshot( + Request::builder() + .uri("/v2/shelf/authors") + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let nav = body["navigation"].as_array().unwrap(); + assert_eq!(nav.len(), 3); + let titles: Vec<_> = nav.iter().map(|n| n["title"].as_str().unwrap()).collect(); + assert!(titles.contains(&"Author 0000")); + let hrefs: Vec<_> = nav.iter().map(|n| n["href"].as_str().unwrap()).collect(); + assert!( + hrefs + .iter() + .all(|h| h.starts_with("/opds/v2/search?q=Author")), + "author entries must link to the scoped search feed: {hrefs:?}" + ); +} + +#[tokio::test] +async fn shelf_authors_paginates_with_next_link() { + let (state, auth) = test_state().await; + let token = admin_token(&auth).await; + // Default page size is 50; insert 51 to trigger next link + insert_authors(&state, 51).await; + let app = opds_routes().with_state(state); + let resp = app + .clone() + .oneshot( + Request::builder() + .uri("/v2/shelf/authors") + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["navigation"].as_array().unwrap().len(), 50); + let links = body["links"].as_array().unwrap(); + assert!( + links.iter().any(|l| l["rel"].as_str() == Some("next")), + "expected next link for 51 authors" + ); + + let resp = app + .oneshot( + Request::builder() + .uri("/v2/shelf/authors?page=2") + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["navigation"].as_array().unwrap().len(), 1); + let links = body["links"].as_array().unwrap(); + assert!( + !links.iter().any(|l| l["rel"].as_str() == Some("next")), + "no next link on the last page" + ); +} diff --git a/crates/paroche/src/opds/search.rs b/crates/paroche/src/opds/search.rs index 252f868d..06a6c4a3 100644 --- a/crates/paroche/src/opds/search.rs +++ b/crates/paroche/src/opds/search.rs @@ -88,7 +88,7 @@ pub async fn search_v1( } } -fn urlencoded(s: &str) -> String { +pub(crate) fn urlencoded(s: &str) -> String { s.chars() .flat_map(|c| { if c.is_alphanumeric() || matches!(c, '-' | '_' | '.' | '~') { diff --git a/crates/paroche/src/routes/indexer.rs b/crates/paroche/src/routes/indexer.rs index fd10ee11..bb5aef73 100644 --- a/crates/paroche/src/routes/indexer.rs +++ b/crates/paroche/src/routes/indexer.rs @@ -317,6 +317,7 @@ fn indexer_service_error(error: ServiceError) -> ParocheError { match error { ServiceError::NotFound => ParocheError::NotFound, ServiceError::NotAvailable => ParocheError::Unavailable, + ServiceError::InvalidInput(message) => ParocheError::Validation { message }, ServiceError::Internal(_) => ParocheError::Internal, } } diff --git a/crates/paroche/src/routes/kosync.rs b/crates/paroche/src/routes/kosync.rs index 4c78bf20..0532cffe 100644 --- a/crates/paroche/src/routes/kosync.rs +++ b/crates/paroche/src/routes/kosync.rs @@ -100,6 +100,13 @@ async fn create_user( State(state): State, Json(body): Json, ) -> Result<(StatusCode, Json), ParocheError> { + // WHY: KOReader self-registration is protocol behavior, but unlimited + // anonymous account creation is an abuse surface — operators can close + // it via config once their readers are enrolled. + if !state.config.paroche.kosync_registration_enabled { + return Err(ParocheError::Forbidden); + } + if body.username.trim().is_empty() || body.password.is_blank() { return Err(ParocheError::Validation { message: "username and password are required".to_string(), @@ -365,6 +372,35 @@ mod tests { assert_eq!(parsed["username"], "reader1"); } + #[tokio::test] + async fn create_user_rejected_when_registration_disabled() { + let (mut state, _) = test_state().await; + let mut config = (*state.config).clone(); + config.paroche.kosync_registration_enabled = false; + state.config = std::sync::Arc::new(config); + let app = super::super::super::build_router(state); + + let create_body = serde_json::json!({ + "username": "reader6", + "password": "mypassword" + }); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/kosync/users/create") + .header("content-type", "application/json") + .body(axum::body::Body::from( + serde_json::to_string(&create_body).unwrap(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + } + #[test] fn constant_time_str_eq_matches_expected_semantics() { assert!(constant_time_str_eq( diff --git a/crates/paroche/src/routes/request.rs b/crates/paroche/src/routes/request.rs index 2f22f91a..4f6611f8 100644 --- a/crates/paroche/src/routes/request.rs +++ b/crates/paroche/src/routes/request.rs @@ -479,6 +479,83 @@ mod tests { Ok(resp.status()) } + #[tokio::test] + async fn list_requests_paginates_at_the_database() -> TestResult<()> { + let (mut state, auth) = test_state().await; + let config = horismos::AitesisConfig { + max_pending_per_user: 10, + max_requests_per_day: 100, + auto_approve_admins: false, + }; + let service = Arc::new(aitesis::AitesisServiceImpl::new( + state.db.read.clone(), + state.db.write.clone(), + config, + TestRoles, + TestIdentity, + TestMonitor, + )); + state.requests = Arc::new(TestRequestAdapter(service)); + + auth.create_user(CreateUserRequest { + username: "pager".to_string(), + display_name: "Pager".to_string(), + password: "password123".to_string(), + role: UserRole::Member, + }) + .await + .context(CreateUserSnafu)?; + let token = auth + .login("pager", "password123") + .await + .context(LoginSnafu)? + .access_token; + let app = crate::build_router(state); + + for i in 0..5 { + let body = json!({ + "media_type": "music", + "title": format!("Album {i}"), + "external_id": null + }); + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/v1/requests") + .header("Content-Type", "application/json") + .header("Authorization", format!("Bearer {token}")) + .body(Body::from( + serde_json::to_vec(&body).context(SerializeRequestBodySnafu)?, + )) + .context(BuildRequestSnafu)?, + ) + .await + .unwrap_or_else(|e| match e {}); + assert_eq!(resp.status(), StatusCode::CREATED); + } + + let resp = app + .oneshot( + Request::builder() + .uri("/api/v1/requests?per_page=2&page=3") + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .context(BuildRequestSnafu)?, + ) + .await + .unwrap_or_else(|e| match e {}); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap_or_default(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap_or_default(); + assert_eq!(body["data"].as_array().map(Vec::len), Some(1)); + assert_eq!(body["meta"]["total"], 5); + Ok(()) + } + #[tokio::test] async fn submit_request_enforces_aitesis_pending_limit() -> TestResult<()> { let (mut state, auth) = test_state().await; diff --git a/crates/paroche/src/routes/subtitle.rs b/crates/paroche/src/routes/subtitle.rs index c59bc572..05add31a 100644 --- a/crates/paroche/src/routes/subtitle.rs +++ b/crates/paroche/src/routes/subtitle.rs @@ -121,6 +121,7 @@ fn subtitle_service_error(error: ServiceError) -> ParocheError { match error { ServiceError::NotAvailable => ParocheError::Unavailable, ServiceError::NotFound => ParocheError::NotFound, + ServiceError::InvalidInput(message) => ParocheError::Validation { message }, ServiceError::Internal(_) => ParocheError::Internal, } } diff --git a/crates/paroche/src/routes/user.rs b/crates/paroche/src/routes/user.rs index d635fb83..bd5aec72 100644 --- a/crates/paroche/src/routes/user.rs +++ b/crates/paroche/src/routes/user.rs @@ -1,5 +1,5 @@ use axum::Json; -use axum::extract::{Path, State}; +use axum::extract::{Path, Query, State}; use axum::http::StatusCode; use exousia::user::{CreateUserRequest, UserRole}; use exousia::{AuthService, RequireAdmin, TokenPair}; @@ -111,19 +111,46 @@ pub async fn logout( Ok(StatusCode::NO_CONTENT) } +#[derive(Deserialize)] +pub struct UserListQuery { + #[serde(default = "default_page")] + pub page: u64, + #[serde(default = "default_per_page")] + pub per_page: u64, +} +fn default_page() -> u64 { + 1 +} +fn default_per_page() -> u64 { + 100 +} + pub async fn list_users( State(state): State, _admin: RequireAdmin, + Query(query): Query, ) -> Result { - let users = apotheke::repo::user::list_users(&state.db.read, 100, 0) + let per_page = query.per_page.clamp(1, 100); + let page = query.page.max(1); + let offset = (page - 1) * per_page; + + // INVARIANT: per_page <= 100 and page comes from a u64 query param; + // the i64 conversions cannot overflow for any page the DB can hold. + let users = apotheke::repo::user::list_active_users( + &state.db.read, + per_page as i64, + i64::try_from(offset).unwrap_or(i64::MAX), + ) + .await + .map_err(ParocheError::from)?; + let total = apotheke::repo::user::count_active_users(&state.db.read) .await .map_err(ParocheError::from)?; // WHY: deactivated users are soft-deleted — they must not reappear in - // the roster (DELETE /users/{id} contract). + // the roster (DELETE /users/{id} contract); the repo query filters them. let data: Vec = users .into_iter() - .filter(|u| u.is_active != 0) .filter_map(|u| { let id_bytes = &u.id; let uuid = uuid::Uuid::from_slice(id_bytes).ok()?; @@ -143,7 +170,12 @@ pub async fn list_users( .map(UserResponse::from) .collect(); - Ok(ApiResponse::ok(data)) + Ok(ApiResponse::paginated( + data, + page, + per_page, + u64::try_from(total).unwrap_or(0), + )) } pub async fn create_user( @@ -499,6 +531,66 @@ mod tests { assert_eq!(resp.status(), StatusCode::NO_CONTENT); } + #[tokio::test] + async fn list_users_pages_beyond_100_users() { + let (state, auth) = test_state().await; + let admin = admin_setup(&auth).await; + + // Seed 104 members directly at the repo layer (bypassing Argon2) + for i in 0..104 { + let user = apotheke::repo::user::User { + id: uuid::Uuid::now_v7().as_bytes().to_vec(), + username: format!("user{i:03}"), + display_name: format!("User {i:03}"), + password_hash: "x".to_string(), + role: "member".to_string(), + is_active: 1, + created_at: "2026-01-01T00:00:00Z".to_string(), + last_login_at: None, + }; + apotheke::repo::user::insert_user(&state.db.write, &user) + .await + .unwrap(); + } + + let app = make_app(state); + + // Default page caps at 100 + let resp = app + .clone() + .oneshot( + Request::builder() + .uri("/users") + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(json["data"].as_array().unwrap().len(), 100); + // 104 seeded members + the admin + assert_eq!(json["meta"]["total"], 105); + + // Second page returns the remainder + let resp = app + .oneshot( + Request::builder() + .uri("/users?page=2") + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(json["data"].as_array().unwrap().len(), 5); + } + #[tokio::test] async fn admin_can_list_users() { let (state, auth) = test_state().await; diff --git a/crates/paroche/src/state.rs b/crates/paroche/src/state.rs index 71f506f5..bfcf672f 100644 --- a/crates/paroche/src/state.rs +++ b/crates/paroche/src/state.rs @@ -77,6 +77,8 @@ pub enum ServiceError { NotAvailable, /// The requested resource was not found by the service. NotFound, + /// The caller's input was rejected by validation (HTTP 400, never 500). + InvalidInput(String), /// An internal service error. Internal(String), } diff --git a/crates/paroche/src/subsonic/auth.rs b/crates/paroche/src/subsonic/auth.rs index 776c3f2a..4d128ab3 100644 --- a/crates/paroche/src/subsonic/auth.rs +++ b/crates/paroche/src/subsonic/auth.rs @@ -8,6 +8,15 @@ use super::types::{ }; use crate::state::AppState; +// WHY: subtle's ConstantTimeEq avoids the timing side-channel of a +// short-circuiting `!=` on secret material; unequal lengths return +// not-equal without comparing content (length is public here — the +// expected side is always a 32-char MD5 hex string). +fn constant_time_str_eq(a: &str, b: &str) -> bool { + use subtle::ConstantTimeEq; + a.as_bytes().ct_eq(b.as_bytes()).into() +} + // WHY: wire DTO — Subsonic API auth user fields deserialized from the request. #[derive(Debug, Clone)] pub struct SubsonicUser { @@ -67,7 +76,7 @@ pub async fn authenticate( .map_err(|_| respond_error(fmt, ERR_WRONG_CREDS, "wrong username or password"))?; let expected = format!("{:x}", md5::compute(format!("{}{}", row.password, salt))); - if expected != token { + if !constant_time_str_eq(&expected, token) { return Err(respond_error( fmt, ERR_WRONG_CREDS, @@ -150,3 +159,34 @@ pub async fn set_subsonic_password( .await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::constant_time_str_eq; + + fn flip_hex_char(c: char) -> char { + if c == '0' { '1' } else { '0' } + } + + fn with_flipped_at(s: &str, index: usize) -> String { + s.chars() + .enumerate() + .map(|(i, c)| if i == index { flip_hex_char(c) } else { c }) + .collect() + } + + #[test] + fn constant_time_str_eq_matches_token_semantics() { + let expected = format!("{:x}", md5::compute("secretsalt")); + assert!(constant_time_str_eq(&expected, &expected.clone())); + + // Near-misses differing in the first and last byte both reject + let early = with_flipped_at(&expected, 0); + assert!(!constant_time_str_eq(&expected, &early)); + let late = with_flipped_at(&expected, expected.len() - 1); + assert!(!constant_time_str_eq(&expected, &late)); + + // Length mismatch + assert!(!constant_time_str_eq(&expected, "")); + } +} diff --git a/crates/paroche/src/subsonic/playlists.rs b/crates/paroche/src/subsonic/playlists.rs index 539aad08..bff5f310 100644 --- a/crates/paroche/src/subsonic/playlists.rs +++ b/crates/paroche/src/subsonic/playlists.rs @@ -356,12 +356,25 @@ pub async fn update_playlist( }; let user_id_bytes = user.user_id.as_bytes().to_vec(); - // Verify ownership + // WHY: single transaction — partial metadata/track updates must not + // survive a mid-flight failure, and failures must surface to the client. + let mut tx = match state.db.write.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::warn!(error = %e, "update_playlist: begin transaction failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); + } + }; + + // WHY: the ownership check runs inside the write transaction and every + // mutation repeats the owner predicate — a concurrent delete/transfer + // between check and write cannot slip a mutation onto a playlist the + // caller no longer owns (the old pre-transaction SELECT was a TOCTOU). let owned: Option = sqlx::query_scalar("SELECT 1 FROM subsonic_playlists WHERE id = ? AND owner_id = ?") .bind(&id_bytes) .bind(&user_id_bytes) - .fetch_optional(&state.db.read) + .fetch_optional(&mut *tx) .await .unwrap_or(None); @@ -369,22 +382,13 @@ pub async fn update_playlist( return respond_error(user.format, ERR_NOT_FOUND, "not found"); } - // WHY: single transaction — partial metadata/track updates must not - // survive a mid-flight failure, and failures must surface to the client. - let mut tx = match state.db.write.begin().await { - Ok(tx) => tx, - Err(e) => { - tracing::warn!(error = %e, "update_playlist: begin transaction failed"); - return respond_error(user.format, ERR_GENERIC, "could not update playlist"); - } - }; - if let Some(name) = &q.name && let Err(e) = sqlx::query( - "UPDATE subsonic_playlists SET name = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + "UPDATE subsonic_playlists SET name = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ? AND owner_id = ?", ) .bind(name) .bind(&id_bytes) + .bind(&user_id_bytes) .execute(&mut *tx) .await { @@ -394,10 +398,11 @@ pub async fn update_playlist( if let Some(comment) = &q.comment && let Err(e) = sqlx::query( - "UPDATE subsonic_playlists SET comment = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + "UPDATE subsonic_playlists SET comment = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ? AND owner_id = ?", ) .bind(comment) .bind(&id_bytes) + .bind(&user_id_bytes) .execute(&mut *tx) .await { @@ -407,10 +412,11 @@ pub async fn update_playlist( if let Some(public) = q.public && let Err(e) = sqlx::query( - "UPDATE subsonic_playlists SET public = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", + "UPDATE subsonic_playlists SET public = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ? AND owner_id = ?", ) .bind(if public { 1i64 } else { 0i64 }) .bind(&id_bytes) + .bind(&user_id_bytes) .execute(&mut *tx) .await { @@ -489,14 +495,23 @@ pub async fn delete_playlist( }; let user_id_bytes = user.user_id.as_bytes().to_vec(); - if let Err(e) = sqlx::query("DELETE FROM subsonic_playlists WHERE id = ? AND owner_id = ?") + let result = match sqlx::query("DELETE FROM subsonic_playlists WHERE id = ? AND owner_id = ?") .bind(&id_bytes) .bind(user_id_bytes) .execute(&state.db.write) .await { - tracing::warn!(error = %e, "delete_playlist: delete failed"); - return respond_error(user.format, ERR_GENERIC, "could not delete playlist"); + Ok(result) => result, + Err(e) => { + tracing::warn!(error = %e, "delete_playlist: delete failed"); + return respond_error(user.format, ERR_GENERIC, "could not delete playlist"); + } + }; + + // WHY: zero rows means the playlist does not exist or belongs to another + // user — a success response would mask the failed ownership check. + if result.rows_affected() == 0 { + return respond_error(user.format, ERR_NOT_FOUND, "not found"); } respond_ok(user.format, "", None) @@ -665,6 +680,107 @@ mod tests { assert!(body.contains("status=\"ok\"")); } + #[tokio::test] + async fn delete_playlist_not_owned_returns_not_found() { + let (app, state, key) = subsonic_app().await; + + // Seed a second user owning a playlist the caller must not delete + let other_id = uuid::Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO users (id, username, display_name, password_hash, role, is_active, created_at) + VALUES (?, 'other', 'Other', 'x', 'member', 1, '2026-01-01T00:00:00Z')", + ) + .bind(&other_id) + .execute(&state.db.write) + .await + .unwrap(); + let other_pl = uuid::Uuid::now_v7(); + sqlx::query("INSERT INTO subsonic_playlists (id, owner_id, name) VALUES (?, ?, 'Theirs')") + .bind(other_pl.as_bytes().to_vec()) + .bind(&other_id) + .execute(&state.db.write) + .await + .unwrap(); + + for id in [other_pl.to_string(), uuid::Uuid::now_v7().to_string()] { + let resp = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/rest/deletePlaylist.view?apiKey={key}&id={id}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert!( + body.contains("status=\"failed\""), + "expected failed status for {id}, got: {body}" + ); + assert!( + body.contains(r#"code="70""#), + "expected not-found code for {id}, got: {body}" + ); + } + + // The other user's playlist survives + let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM subsonic_playlists WHERE id = ?") + .bind(other_pl.as_bytes().to_vec()) + .fetch_one(&state.db.read) + .await + .unwrap(); + assert_eq!(rows, 1); + } + + #[tokio::test] + async fn update_playlist_not_owned_returns_not_found() { + let (app, state, key) = subsonic_app().await; + + let other_id = uuid::Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO users (id, username, display_name, password_hash, role, is_active, created_at) + VALUES (?, 'other2', 'Other2', 'x', 'member', 1, '2026-01-01T00:00:00Z')", + ) + .bind(&other_id) + .execute(&state.db.write) + .await + .unwrap(); + let other_pl = uuid::Uuid::now_v7(); + sqlx::query("INSERT INTO subsonic_playlists (id, owner_id, name) VALUES (?, ?, 'Theirs')") + .bind(other_pl.as_bytes().to_vec()) + .bind(&other_id) + .execute(&state.db.write) + .await + .unwrap(); + + let resp = app + .oneshot( + Request::builder() + .uri(format!( + "/rest/updatePlaylist.view?apiKey={key}&playlistId={other_pl}&name=Stolen" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert!( + body.contains(r#"code="70""#), + "expected not-found code, got: {body}" + ); + + let name: String = sqlx::query_scalar("SELECT name FROM subsonic_playlists WHERE id = ?") + .bind(other_pl.as_bytes().to_vec()) + .fetch_one(&state.db.read) + .await + .unwrap(); + assert_eq!(name, "Theirs", "non-owner update must not apply"); + } + #[tokio::test] async fn create_playlist_insert_failure_returns_error() { let (app, state, key) = subsonic_app().await; diff --git a/crates/syndesis/src/client/buffer.rs b/crates/syndesis/src/client/buffer.rs index ef183b88..2278a84a 100644 --- a/crates/syndesis/src/client/buffer.rs +++ b/crates/syndesis/src/client/buffer.rs @@ -145,7 +145,10 @@ impl JitterBuffer { .next_back() .expect("len >= 2 guard above") // INVARIANT: see comment above .timestamp_us; - ((last_ts.saturating_sub(first_ts)) / 1000) as u16 + // WHY: the status-report wire field is u16 (protocol codec put_u16); + // a >65s span must saturate, not wrap — a wrapped small value would + // invert the server's flow-control decision under the worst backlog. + u16::try_from(last_ts.saturating_sub(first_ts) / 1000).unwrap_or(u16::MAX) } } @@ -216,6 +219,15 @@ mod tests { assert_eq!(buf.depth_ms(), 50); } + #[test] + fn depth_ms_saturates_for_large_span() { + let mut buf = JitterBuffer::new(); + buf.insert(test_frame(0, 0)); + // Span of 70_000ms exceeds u16::MAX ms — must saturate, not wrap. + buf.insert(test_frame(1, 70_000_000)); + assert_eq!(buf.depth_ms(), u16::MAX); + } + #[test] fn jitter_buffer_evicts_oldest_when_full() { let config = ClientConfig { diff --git a/crates/syndesis/src/config.rs b/crates/syndesis/src/config.rs index 30db2d5b..74632501 100644 --- a/crates/syndesis/src/config.rs +++ b/crates/syndesis/src/config.rs @@ -116,6 +116,13 @@ pub struct ServerConfig { /// Consecutive low-buffer status reports before a renderer is marked /// degraded. pub degraded_lag_count: u32, + /// Whether new-renderer pairing is accepted at all. Disable once the + /// household's renderers are enrolled to close the open-enrollment + /// surface. + pub pairing_enabled: bool, + /// Maximum pairing attempts accepted per minute across all peers. + /// Bounds anonymous key-minting while pairing is enabled. + pub pairing_max_attempts_per_min: u32, } impl Default for ServerConfig { @@ -127,6 +134,8 @@ impl Default for ServerConfig { buffer_low_watermark_ms: 80, zone_low_watermark_ms: 50, degraded_lag_count: 10, + pairing_enabled: true, + pairing_max_attempts_per_min: 5, } } } @@ -182,6 +191,8 @@ mod tests { buffer_low_watermark_ms: 60, zone_low_watermark_ms: 40, degraded_lag_count: 5, + pairing_enabled: false, + pairing_max_attempts_per_min: 3, }, }; let toml = toml::to_string(&original).expect("serialize"); diff --git a/crates/syndesis/src/error.rs b/crates/syndesis/src/error.rs index 1b074a08..cad74b63 100644 --- a/crates/syndesis/src/error.rs +++ b/crates/syndesis/src/error.rs @@ -121,6 +121,18 @@ pub enum SyndesisError { location: snafu::Location, }, + #[snafu(display("pairing is disabled by server policy"))] + PairingDisabled { + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("pairing rate limit exceeded"))] + PairingRateLimited { + #[snafu(implicit)] + location: snafu::Location, + }, + #[snafu(display("cert fingerprint mismatch (TOFU violation)"))] FingerprintMismatch { #[snafu(implicit)] diff --git a/crates/syndesis/src/lib.rs b/crates/syndesis/src/lib.rs index 37aca561..ece3191b 100644 --- a/crates/syndesis/src/lib.rs +++ b/crates/syndesis/src/lib.rs @@ -19,7 +19,8 @@ pub use protocol::session_frame::{ SessionInit as SessionInitMsg, SessionRejected, }; pub use server::auth::{ - SessionOutcome, build_pairing_challenge, build_pairing_complete, handle_session_init, + PairingGate, SessionOutcome, build_pairing_challenge, build_pairing_complete, + handle_session_init, }; pub use tls::{ ObservedFingerprint, SelfSignedCert, compute_fingerprint, generate_self_signed_simple, diff --git a/crates/syndesis/src/server/auth.rs b/crates/syndesis/src/server/auth.rs index 6b6188ff..2c1aa73e 100644 --- a/crates/syndesis/src/server/auth.rs +++ b/crates/syndesis/src/server/auth.rs @@ -19,6 +19,63 @@ pub enum SessionOutcome { Paired(PairingOutcome), } +/// Admission policy for the pairing flow: an on/off switch plus a global +/// fixed-window rate limit on attempts. +/// +/// WHY: `is_new` pairing mints an API key for an unauthenticated peer — +/// without a gate any network peer can enroll unlimited renderers. +pub struct PairingGate { + enabled: bool, + max_attempts: u32, + window: std::time::Duration, + state: std::sync::Mutex, +} + +struct PairingWindow { + started: std::time::Instant, + attempts: u32, +} + +impl PairingGate { + /// Build the gate from server config (`pairing_enabled`, + /// `pairing_max_attempts_per_min`). + #[must_use] + pub fn from_config(config: &crate::config::ServerConfig) -> Self { + Self { + enabled: config.pairing_enabled, + max_attempts: config.pairing_max_attempts_per_min, + window: std::time::Duration::from_secs(60), + state: std::sync::Mutex::new(PairingWindow { + started: std::time::Instant::now(), + attempts: 0, + }), + } + } + + /// Admit or reject one pairing attempt. + pub fn admit(&self) -> Result<(), SyndesisError> { + if !self.enabled { + return Err(SyndesisError::PairingDisabled { + location: snafu::location!(), + }); + } + // WHY: poisoning is impossible to act on here — the guarded state is + // two plain integers, safe to reuse after a panicked writer. + let mut window = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if window.started.elapsed() >= self.window { + window.started = std::time::Instant::now(); + window.attempts = 0; + } + if window.attempts >= self.max_attempts { + return Err(SyndesisError::PairingRateLimited { + location: snafu::location!(), + }); + } + window.attempts += 1; + Ok(()) + } +} + /// Process a `SessionInit` frame from a connecting renderer. /// /// - `is_new: true` -> run the pairing flow (generate + store API key). @@ -32,8 +89,10 @@ pub async fn handle_session_init( write_pool: &SqlitePool, init: &SessionInitMsg, peer_cert_fingerprint: &str, + pairing_gate: &PairingGate, ) -> Result { if init.is_new { + pairing_gate.admit()?; let req = PairingRequest { renderer_name: &init.renderer_name, renderer_id: &init.renderer_id.0, @@ -87,6 +146,10 @@ mod tests { pool } + fn open_gate() -> PairingGate { + PairingGate::from_config(&crate::config::ServerConfig::default()) + } + fn renderer_id() -> String { uuid::Uuid::now_v7().to_string() } @@ -103,7 +166,7 @@ mod tests { is_new: true, }; - let outcome = handle_session_init(&pool, &pool, &init, "aabbcc") + let outcome = handle_session_init(&pool, &pool, &init, "aabbcc", &open_gate()) .await .unwrap(); @@ -126,13 +189,14 @@ mod tests { is_new: true, }; - let api_key = match handle_session_init(&pool, &pool, &init, "fingerprint_renderer") - .await - .unwrap() - { - SessionOutcome::Paired(o) => o.api_key, - _ => panic!("expected paired"), - }; + let api_key = + match handle_session_init(&pool, &pool, &init, "fingerprint_renderer", &open_gate()) + .await + .unwrap() + { + SessionOutcome::Paired(o) => o.api_key, + _ => panic!("expected paired"), + }; let auth_init = SessionInitMsg { renderer_name: "Test Renderer".to_string(), @@ -141,7 +205,14 @@ mod tests { is_new: false, }; - let result = handle_session_init(&pool, &pool, &auth_init, "fingerprint_renderer").await; + let result = handle_session_init( + &pool, + &pool, + &auth_init, + "fingerprint_renderer", + &open_gate(), + ) + .await; assert!(result.is_ok()); match result.unwrap() { @@ -162,7 +233,7 @@ mod tests { is_new: true, }; - handle_session_init(&pool, &pool, &init, "fp") + handle_session_init(&pool, &pool, &init, "fp", &open_gate()) .await .unwrap(); @@ -173,7 +244,7 @@ mod tests { is_new: false, }; - let result = handle_session_init(&pool, &pool, &auth_init, "fp").await; + let result = handle_session_init(&pool, &pool, &auth_init, "fp", &open_gate()).await; assert!(matches!(result, Err(SyndesisError::InvalidApiKey { .. }))); } @@ -192,7 +263,7 @@ mod tests { is_new: true, }; - let api_key = match handle_session_init(&pool, &pool, &init, "fp") + let api_key = match handle_session_init(&pool, &pool, &init, "fp", &open_gate()) .await .unwrap() { @@ -209,7 +280,7 @@ mod tests { is_new: false, }; - let result = handle_session_init(&pool, &pool, &auth_init, "fp").await; + let result = handle_session_init(&pool, &pool, &auth_init, "fp", &open_gate()).await; assert!(matches!( result, @@ -229,13 +300,14 @@ mod tests { is_new: true, }; - let api_key = match handle_session_init(&pool, &pool, &init, "original-fingerprint") - .await - .unwrap() - { - SessionOutcome::Paired(o) => o.api_key, - _ => panic!("expected paired"), - }; + let api_key = + match handle_session_init(&pool, &pool, &init, "original-fingerprint", &open_gate()) + .await + .unwrap() + { + SessionOutcome::Paired(o) => o.api_key, + _ => panic!("expected paired"), + }; let auth_init = SessionInitMsg { renderer_name: "Test Renderer".to_string(), @@ -244,11 +316,69 @@ mod tests { is_new: false, }; - let result = handle_session_init(&pool, &pool, &auth_init, "different-fingerprint").await; + let result = handle_session_init( + &pool, + &pool, + &auth_init, + "different-fingerprint", + &open_gate(), + ) + .await; assert!(matches!( result, Err(SyndesisError::FingerprintMismatch { .. }) )); } + + #[tokio::test] + async fn pairing_rejected_when_disabled() { + let pool = setup().await; + let gate = PairingGate::from_config(&crate::config::ServerConfig { + pairing_enabled: false, + ..crate::config::ServerConfig::default() + }); + + let init = SessionInitMsg { + renderer_name: "Test Renderer".to_string(), + renderer_id: RendererSyncId(renderer_id()), + api_key: None, + is_new: true, + }; + + let result = handle_session_init(&pool, &pool, &init, "fp", &gate).await; + assert!(matches!(result, Err(SyndesisError::PairingDisabled { .. }))); + } + + #[tokio::test] + async fn pairing_rate_limited_after_n_attempts() { + let pool = setup().await; + let gate = PairingGate::from_config(&crate::config::ServerConfig { + pairing_max_attempts_per_min: 2, + ..crate::config::ServerConfig::default() + }); + + for i in 0..2 { + let init = SessionInitMsg { + renderer_name: format!("Renderer {i}"), + renderer_id: RendererSyncId(renderer_id()), + api_key: None, + is_new: true, + }; + let result = handle_session_init(&pool, &pool, &init, "fp", &gate).await; + assert!(result.is_ok(), "attempt {i} within the budget must pass"); + } + + let init = SessionInitMsg { + renderer_name: "Renderer over budget".to_string(), + renderer_id: RendererSyncId(renderer_id()), + api_key: None, + is_new: true, + }; + let result = handle_session_init(&pool, &pool, &init, "fp", &gate).await; + assert!(matches!( + result, + Err(SyndesisError::PairingRateLimited { .. }) + )); + } } diff --git a/crates/syndesis/src/server/mod.rs b/crates/syndesis/src/server/mod.rs index 9cead8c3..5667f0f6 100644 --- a/crates/syndesis/src/server/mod.rs +++ b/crates/syndesis/src/server/mod.rs @@ -7,7 +7,8 @@ pub mod zone; use std::net::SocketAddr; pub use auth::{ - SessionOutcome, build_pairing_challenge, build_pairing_complete, handle_session_init, + PairingGate, SessionOutcome, build_pairing_challenge, build_pairing_complete, + handle_session_init, }; pub use session::StreamSession; use snafu::ResultExt; diff --git a/crates/syndesis/src/server/session.rs b/crates/syndesis/src/server/session.rs index 41d739d4..2a6ea7fc 100644 --- a/crates/syndesis/src/server/session.rs +++ b/crates/syndesis/src/server/session.rs @@ -307,6 +307,79 @@ mod tests { use super::*; + fn init(codecs: Vec, rates: Vec, channels: Vec) -> SessionInit { + SessionInit { + protocol_version: PROTOCOL_VERSION, + supported_codecs: codecs, + sample_rates: rates, + channel_configs: channels, + } + } + + #[test] + fn negotiate_rejects_version_mismatch() { + let mut bad = init(vec![AudioCodec::Flac], vec![48000], vec![2]); + bad.protocol_version = PROTOCOL_VERSION + 1; + let result = negotiate_params(&bad); + assert!(matches!(result, Err(SyndesisError::Negotiation { .. }))); + } + + #[test] + fn negotiate_prefers_flac_over_pcm() { + let both = init( + vec![AudioCodec::Pcm, AudioCodec::Flac], + vec![48000], + vec![2], + ); + let (codec, rate, channels) = negotiate_params(&both).expect("negotiation succeeds"); + assert_eq!(codec, AudioCodec::Flac); + assert_eq!(rate, 48000); + assert_eq!(channels, 2); + } + + #[test] + fn negotiate_falls_back_to_pcm() { + let pcm_only = init(vec![AudioCodec::Pcm], vec![44100], vec![2]); + let (codec, rate, _) = negotiate_params(&pcm_only).expect("negotiation succeeds"); + assert_eq!(codec, AudioCodec::Pcm); + assert_eq!(rate, 44100); + } + + #[test] + fn negotiate_rejects_no_common_codec() { + let none = init(vec![], vec![48000], vec![2]); + assert!(matches!( + negotiate_params(&none), + Err(SyndesisError::Negotiation { .. }) + )); + } + + #[test] + fn negotiate_rejects_unsupported_sample_rate() { + let odd_rate = init(vec![AudioCodec::Flac], vec![22050], vec![2]); + assert!(matches!( + negotiate_params(&odd_rate), + Err(SyndesisError::Negotiation { .. }) + )); + } + + #[test] + fn negotiate_channel_policy_prefers_stereo_then_first() { + let multi = init(vec![AudioCodec::Flac], vec![48000], vec![6, 2]); + let (_, _, channels) = negotiate_params(&multi).expect("negotiation succeeds"); + assert_eq!(channels, 2, "stereo preferred when offered"); + + let surround_only = init(vec![AudioCodec::Flac], vec![48000], vec![6]); + let (_, _, channels) = negotiate_params(&surround_only).expect("negotiation succeeds"); + assert_eq!(channels, 6, "first offered config when no stereo"); + + let empty = init(vec![AudioCodec::Flac], vec![48000], vec![]); + assert!(matches!( + negotiate_params(&empty), + Err(SyndesisError::Negotiation { .. }) + )); + } + #[tokio::test(start_paused = true)] async fn delayed_interval_first_tick_waits_full_period() { let period = Duration::from_secs(5); diff --git a/crates/syndesis/src/server/zone.rs b/crates/syndesis/src/server/zone.rs index ec71db33..fc7b20c0 100644 --- a/crates/syndesis/src/server/zone.rs +++ b/crates/syndesis/src/server/zone.rs @@ -227,7 +227,7 @@ impl ZoneStream { } /// Run the zone stream: decode from source, fan-out to all members. - pub async fn run(&mut self, mut source: S, cancel: watch::Receiver) { + pub async fn run(&mut self, mut source: S, mut cancel: watch::Receiver) { self.play_state = ZonePlayState::Playing; loop { @@ -245,14 +245,25 @@ impl ZoneStream { continue; } - match source.next_frame().await { - Some(frame) => { - self.fan_out_frame(frame).await; - } - None => { - debug!("zone audio source exhausted"); - break; + // WHY: biased select with cancel first — a shutdown must preempt + // an in-flight next_frame() await; the loop-top check alone would + // leave shutdown blocked until the source produces a frame. + tokio::select! { + biased; + changed = cancel.changed() => { + if changed.is_err() || *cancel.borrow() { + break; + } } + frame = source.next_frame() => match frame { + Some(frame) => { + self.fan_out_frame(frame).await; + } + None => { + debug!("zone audio source exhausted"); + break; + } + }, } } @@ -385,6 +396,64 @@ mod tests { assert_eq!(zone.member_count(), 0); } + #[tokio::test] + async fn fan_out_marks_degraded_when_channel_full() { + let server_config = ServerConfig { + frame_channel_capacity: 1, + ..ServerConfig::default() + }; + let mut zone = ZoneStream::with_configs(server_config, ClockConfig::default()); + // Hold the receiver without draining so the channel fills + let _rx = zone.add_renderer("r1"); + + zone.fan_out_frame(test_frame(0, 1000)).await; + assert!(zone.degraded_renderers().is_empty(), "first frame fits"); + + zone.fan_out_frame(test_frame(1, 2000)).await; + assert!( + zone.degraded_renderers().contains(&"r1".to_string()), + "overflowing the channel must mark the renderer degraded" + ); + } + + #[tokio::test] + async fn run_cancels_promptly_while_awaiting_source() { + /// Source whose next_frame never resolves — models a stalled decoder. + struct PendingSource; + impl AudioSource for PendingSource { + async fn next_frame(&mut self) -> Option { + std::future::pending().await + } + } + + let (cancel_tx, cancel_rx) = watch::channel(false); + let mut zone = ZoneStream::new(); + + let run = async move { + zone.run(PendingSource, cancel_rx).await; + }; + let fire_cancel = async move { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + // WHY: send fails only when the run future already dropped its + // receiver — either way cancellation is moot at that point. + cancel_tx.send(true).ok(); + }; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(2), + futures_join(run, fire_cancel), + ) + .await; + assert!( + result.is_ok(), + "run() must return promptly once cancel fires, even mid-await" + ); + } + + async fn futures_join(a: A, b: B) { + tokio::join!(a, b); + } + #[test] fn needs_backpressure_when_buffer_low() { let mut zone = ZoneStream::new();