fix(audio): align ALSA Direct visualizers

cx

crates/qbz-audio/src/alsa_direct.rs

@@ -869,6 +869,18 @@ impl AlsaDirectStream {
         self.channels
     }
 
+    /// Frames accepted by ALSA but not yet presented by the device.
+    ///
+    /// This is observational only: it neither changes PCM parameters nor
+    /// waits for the device. The direct writer uses it to point the passive
+    /// visualizer tap at the audible playhead instead of the queue head.
+    pub fn playback_delay_frames(&self) -> Result<u64, String> {
+        let pcm = self.pcm.lock().unwrap();
+        pcm.delay()
+            .map(|frames| frames.max(0) as u64)
+            .map_err(|e| format!("Failed to query ALSA playback delay: {}", e))
+    }
+
     /// Get device ID
     pub fn device_id(&self) -> &str {
         &self.device_id
@@ -974,6 +986,10 @@ impl AlsaDirectStream {
         2
     }
 
+    pub fn playback_delay_frames(&self) -> Result<u64, String> {
+        Ok(0)
+    }
+
     /// Check if device is a bit-perfect hardware device (always false on non-Linux)
     pub fn is_hw_device(_device_id: &str) -> bool {
         false

crates/qbz-audio/src/visualizer/mod.rs

@@ -20,7 +20,7 @@ pub use ring_buffer::RingBuffer;
 pub use scope::{GONIOMETER_BIT, OSCILLOSCOPE_BIT};
 pub use tapped_source::TappedSource;
 
-use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
 use std::sync::Arc;
 
 /// Number of frequency bins to send to frontend
@@ -33,6 +33,14 @@ pub const FFT_SIZE: usize = 4096;
 /// Target frames per second for visualization updates
 pub const TARGET_FPS: u64 = 30;
 
+/// History retained for output-playhead alignment.
+///
+/// ALSA Direct currently allows up to 500 ms of queued PCM at high sample
+/// rates. 1,048,576 interleaved samples cover that delay plus the FFT window
+/// for stereo PCM through 768 kHz, while keeping the audio-thread write path
+/// allocation-free.
+const RING_HISTORY_SAMPLES: usize = 1_048_576;
+
 /// Shared state for visualization that can be passed to the audio thread
 #[derive(Clone)]
 pub struct VisualizerTap {
@@ -52,17 +60,22 @@ pub struct VisualizerTap {
     /// Requested real-time scope producers. Frontends update this off the
     /// audio thread; zero keeps the added DSP completely idle.
     pub scope_mask: Arc<AtomicU32>,
+    /// Number of interleaved samples queued ahead of the audible playhead.
+    /// Zero for callback-paced/shared backends; ALSA Direct updates it from
+    /// `snd_pcm_delay` without changing the PCM stream.
+    playback_lag_samples: Arc<AtomicUsize>,
 }
 
 impl VisualizerTap {
     /// Create a new tap
     pub fn new() -> Self {
         Self {
-            ring_buffer: Arc::new(RingBuffer::new(FFT_SIZE * 2)),
+            ring_buffer: Arc::new(RingBuffer::new(RING_HISTORY_SAMPLES)),
             enabled: Arc::new(AtomicBool::new(false)),
             paused: Arc::new(AtomicBool::new(false)),
             sample_rate: Arc::new(AtomicU32::new(44100)),
             scope_mask: Arc::new(AtomicU32::new(0)),
+            playback_lag_samples: Arc::new(AtomicUsize::new(0)),
         }
     }
 
@@ -108,6 +121,29 @@ impl VisualizerTap {
     pub fn set_scope_mask(&self, mask: u32) {
         self.scope_mask.store(mask, Ordering::Relaxed);
     }
+
+    /// Align snapshots with a direct backend's audible playhead.
+    ///
+    /// `queued_frames` comes from the output API, so convert it to the
+    /// interleaved-sample units used by [`RingBuffer`]. Saturation keeps a
+    /// malformed device report from wrapping; the ring clamps to retained
+    /// history when the snapshot is taken.
+    pub fn set_output_delay_frames(&self, queued_frames: u64, channels: u16) {
+        let samples = queued_frames.saturating_mul(u64::from(channels));
+        self.playback_lag_samples.store(
+            usize::try_from(samples).unwrap_or(usize::MAX),
+            Ordering::Relaxed,
+        );
+    }
+
+    /// Return callback-paced/shared backends to the newest captured window.
+    pub fn clear_output_delay(&self) {
+        self.playback_lag_samples.store(0, Ordering::Relaxed);
+    }
+
+    pub(crate) fn output_delay_samples(&self) -> usize {
+        self.playback_lag_samples.load(Ordering::Relaxed)
+    }
 }
 
 impl Default for VisualizerTap {

crates/qbz-audio/src/visualizer/processor.rs

@@ -153,7 +153,8 @@ fn run_fft_loop(tap: VisualizerTap, sink: Arc<dyn VizSink>) {
             let sample_rate = tap.sample_rate.load(Ordering::Relaxed);
 
             // Get samples from ring buffer
-            tap.ring_buffer.snapshot(&mut samples);
+            tap.ring_buffer
+                .snapshot_behind(&mut samples, tap.output_delay_samples());
 
             // The scope pipeline is explicitly requested by the visible
             // frontend destination. Its FIR and pitch FFT remain completely

crates/qbz-audio/src/visualizer/ring_buffer.rs

@@ -63,18 +63,40 @@ impl RingBuffer {
     /// Copies the last `dest.len()` samples into the provided slice.
     /// The samples are ordered from oldest to newest.
     pub fn snapshot(&self, dest: &mut [f32]) {
+        self.snapshot_behind(dest, 0);
+    }
+
+    /// Take a snapshot ending `lag_samples` behind the producer head.
+    ///
+    /// Direct output backends can queue a substantial amount of PCM ahead of
+    /// the hardware playhead. The visualizer still taps the exact source
+    /// samples, but asks for a historical window ending at the audible
+    /// position instead of the newest queued sample. A request older than the
+    /// retained ring is clamped to the oldest complete window; before enough
+    /// samples exist, the missing prefix remains silent.
+    pub fn snapshot_behind(&self, dest: &mut [f32], lag_samples: usize) {
+        dest.fill(0.0);
+        if dest.is_empty() || self.size == 0 {
+            return;
+        }
+
         let write_pos = self.write_pos.load(Ordering::Relaxed);
-        let len = dest.len().min(self.size);
+        let window_len = dest.len().min(self.size);
+        let lag_samples = lag_samples
+            .min(self.size.saturating_sub(window_len))
+            .min(write_pos);
+        let end_pos = write_pos - lag_samples;
+        let available = end_pos.min(window_len);
+        let dest_start = dest.len() - available;
+        let source_start = end_pos - available;
 
         // Safety: We're the only reader, and the writer only increments write_pos
         unsafe {
             let buffer = &*self.buffer.get();
 
-            for i in 0..len {
-                // Calculate the index for the i-th oldest sample
-                // We want samples from (write_pos - len) to (write_pos - 1)
-                let idx = (write_pos + self.size - len + i) % self.size;
-                dest[i] = buffer[idx];
+            for i in 0..available {
+                let idx = (source_start + i) % self.size;
+                dest[dest_start + i] = buffer[idx];
             }
         }
     }
@@ -123,4 +145,42 @@ mod tests {
         // Should get the last 4 samples: 6, 7, 8, 9
         assert_eq!(snapshot, [6.0, 7.0, 8.0, 9.0]);
     }
+
+    #[test]
+    fn snapshot_behind_reads_a_historical_window() {
+        let buffer = RingBuffer::new(16);
+        for i in 0..10 {
+            buffer.push(i as f32);
+        }
+
+        let mut snapshot = [0.0f32; 4];
+        buffer.snapshot_behind(&mut snapshot, 3);
+
+        assert_eq!(snapshot, [3.0, 4.0, 5.0, 6.0]);
+    }
+
+    #[test]
+    fn snapshot_behind_zero_fills_history_not_yet_captured() {
+        let buffer = RingBuffer::new(16);
+        buffer.push(1.0);
+        buffer.push(2.0);
+
+        let mut snapshot = [9.0f32; 4];
+        buffer.snapshot_behind(&mut snapshot, 1);
+
+        assert_eq!(snapshot, [0.0, 0.0, 0.0, 1.0]);
+    }
+
+    #[test]
+    fn snapshot_behind_never_reads_past_retained_history() {
+        let buffer = RingBuffer::new(8);
+        for i in 0..12 {
+            buffer.push(i as f32);
+        }
+
+        let mut snapshot = [0.0f32; 4];
+        buffer.snapshot_behind(&mut snapshot, usize::MAX);
+
+        assert_eq!(snapshot, [4.0, 5.0, 6.0, 7.0]);
+    }
 }

crates/qbz-player/src/player/mod.rs

@@ -1630,6 +1630,11 @@ impl Player {
 
                 // Visualizer tap (outermost)
                 if let Some(ref tap) = thread_viz_tap {
+                    // FFT frequency bins must follow the decoded source rate.
+                    // Gapless sources are admitted only when their format
+                    // matches the current stream, so preparing the next one
+                    // cannot move this clock ahead to a different rate.
+                    tap.set_sample_rate(source.sample_rate().get());
                     Box::new(TappedSource::new(
                         source,
                         tap.ring_buffer.clone(),
@@ -2232,6 +2237,7 @@ impl Player {
                                     PlaybackEngine::new_alsa_direct(
                                         alsa_stream.clone(),
                                         hardware_volume,
+                                        thread_viz_tap.clone(),
                                     )
                                 }
                                 #[cfg(target_os = "linux")]
@@ -2676,6 +2682,7 @@ impl Player {
                                     PlaybackEngine::new_alsa_direct(
                                         alsa_stream.clone(),
                                         hardware_volume,
+                                        thread_viz_tap.clone(),
                                     )
                                 }
                                 #[cfg(target_os = "linux")]
@@ -3374,6 +3381,7 @@ impl Player {
                                         PlaybackEngine::new_alsa_direct(
                                             alsa_stream.clone(),
                                             hardware_volume,
+                                            thread_viz_tap.clone(),
                                         )
                                     }
                                     #[cfg(target_os = "linux")]
@@ -3624,6 +3632,7 @@ impl Player {
                                     PlaybackEngine::new_alsa_direct(
                                         alsa_stream.clone(),
                                         hardware_volume,
+                                        thread_viz_tap.clone(),
                                     )
                                 }
                                 #[cfg(target_os = "linux")]

crates/qbz-player/src/player/playback_engine.rs

@@ -8,7 +8,7 @@
 //! to enable gapless playback. When one source ends, the next is picked up
 //! seamlessly without interrupting the PCM stream.
 
-use qbz_audio::AlsaDirectStream;
+use qbz_audio::{AlsaDirectStream, VisualizerTap};
 #[cfg(target_os = "linux")]
 use qbz_audio::JackStream;
 use rodio::{mixer::Mixer, Player as RodioPlayer, Source};
@@ -129,7 +129,11 @@ impl PlaybackEngine {
 
     /// Create ALSA Direct engine with gapless source queue.
     /// Spawns a single writer thread that lives for the engine's lifetime.
-    pub fn new_alsa_direct(stream: Arc<AlsaDirectStream>, hardware_volume: bool) -> Self {
+    pub fn new_alsa_direct(
+        stream: Arc<AlsaDirectStream>,
+        hardware_volume: bool,
+        visualizer_tap: Option<VisualizerTap>,
+    ) -> Self {
         let is_playing = Arc::new(AtomicBool::new(false));
         let should_stop = Arc::new(AtomicBool::new(false));
         let position_frames = Arc::new(AtomicU64::new(0));
@@ -147,6 +151,7 @@ impl PlaybackEngine {
             let queue_c = source_queue.clone();
             let transition_c = source_transition.clone();
             let channels = stream.channels();
+            let visualizer_c = visualizer_tap.clone();
 
             thread::spawn(move || {
                 alsa_writer_thread(
@@ -158,6 +163,7 @@ impl PlaybackEngine {
                     queue_c,
                     transition_c,
                     channels,
+                    visualizer_c,
                 );
             })
         };
@@ -650,6 +656,24 @@ impl PlaybackEngine {
 /// When a source ends, seamlessly picks up the next one from the queue
 /// (gapless transition). If no next source is available, drains the ALSA
 /// buffer and waits for the next source or a stop signal.
+///
+/// The source is consumed in ~10 ms pieces instead of the historical fixed
+/// 8192-frame blocks. This does not change PCM data or ALSA parameters; it
+/// only keeps the passive visualizer tap from receiving ~186 ms bursts at
+/// 44.1 kHz. `snd_pcm_delay` then points snapshots behind the queued PCM at
+/// the part currently reaching the DAC.
+const ALSA_PCM_FEED_TARGET_MS: u64 = 10;
+const ALSA_PCM_FEED_MIN_FRAMES: u64 = 64;
+const ALSA_PCM_FEED_MAX_FRAMES: u64 = 8192;
+
+fn alsa_pcm_feed_frames(sample_rate: u32) -> usize {
+    let frames = u64::from(sample_rate)
+        .saturating_mul(ALSA_PCM_FEED_TARGET_MS)
+        .div_ceil(1000)
+        .clamp(ALSA_PCM_FEED_MIN_FRAMES, ALSA_PCM_FEED_MAX_FRAMES);
+    frames as usize
+}
+
 fn alsa_writer_thread(
     stream: Arc<AlsaDirectStream>,
     is_playing: Arc<AtomicBool>,
@@ -659,14 +683,25 @@ fn alsa_writer_thread(
     source_queue: Arc<SourceQueue<BoxedSampleIter>>,
     source_transition: Arc<AtomicBool>,
     channels: u16,
+    visualizer_tap: Option<VisualizerTap>,
 ) {
-    const CHUNK_FRAMES: usize = 8192;
-    let chunk_samples = CHUNK_FRAMES * channels as usize;
+    let chunk_frames = alsa_pcm_feed_frames(stream.sample_rate());
+    let chunk_samples = chunk_frames * channels as usize;
     let mut buffer_f32 = Vec::with_capacity(chunk_samples);
     let mut current_source: Option<BoxedSampleIter> = None;
     let mut total_frames: u64 = 0;
+    let mut queued_frames = 0u64;
+    let mut visualizer_was_enabled = false;
+    let mut delay_error_reported = false;
 
-    log::info!("[ALSA Direct Engine] Writer thread started (gapless-capable)");
+    if let Some(tap) = visualizer_tap.as_ref() {
+        tap.clear_output_delay();
+    }
+    log::info!(
+        "[ALSA Direct Engine] Writer thread started (gapless-capable, viz feed {} frames / ~{} ms)",
+        chunk_frames,
+        ALSA_PCM_FEED_TARGET_MS
+    );
 
     'thread: loop {
         // Check global stop
@@ -700,7 +735,44 @@ fn alsa_writer_thread(
             std::thread::sleep(Duration::from_millis(50));
         }
 
-        // Fill buffer from current source
+        // Keep the passive tap at the previous audible position while this
+        // next piece is pulled rapidly from the source. Once ALSA accepts it,
+        // the delay query below advances the tap by exactly what the device
+        // consumed. Shared/callback backends leave this offset at zero.
+        let visualizer_enabled = visualizer_tap
+            .as_ref()
+            .map(VisualizerTap::is_enabled)
+            .unwrap_or(false);
+        if visualizer_enabled {
+            if !visualizer_was_enabled {
+                match stream.playback_delay_frames() {
+                    Ok(delay) => queued_frames = delay,
+                    Err(error) => {
+                        if !delay_error_reported {
+                            log::warn!(
+                                "[ALSA Direct Engine] Visualizer delay query failed; using paced feed without playhead compensation: {}",
+                                error
+                            );
+                            delay_error_reported = true;
+                        }
+                        queued_frames = 0;
+                    }
+                }
+            }
+            if let Some(tap) = visualizer_tap.as_ref() {
+                tap.set_output_delay_frames(
+                    queued_frames.saturating_add(chunk_frames as u64),
+                    channels,
+                );
+            }
+        } else if visualizer_was_enabled {
+            if let Some(tap) = visualizer_tap.as_ref() {
+                tap.clear_output_delay();
+            }
+        }
+        visualizer_was_enabled = visualizer_enabled;
+
+        // Fill buffer from current source.
         buffer_f32.clear();
         let source = current_source.as_mut().unwrap();
         let mut source_ended = false;
@@ -715,6 +787,16 @@ fn alsa_writer_thread(
             }
         }
 
+        let frames_in_chunk = buffer_f32.len() / channels as usize;
+        if visualizer_enabled {
+            if let Some(tap) = visualizer_tap.as_ref() {
+                tap.set_output_delay_frames(
+                    queued_frames.saturating_add(frames_in_chunk as u64),
+                    channels,
+                );
+            }
+        }
+
         // Write whatever we have to ALSA (even partial chunks on source end)
         if !buffer_f32.is_empty() {
             if let Err(e) = stream.write_f32(&buffer_f32) {
@@ -722,10 +804,33 @@ fn alsa_writer_thread(
                 break 'thread;
             }
 
-            let frames_written = buffer_f32.len() / channels as usize;
-            total_frames += frames_written as u64;
+            total_frames += frames_in_chunk as u64;
             position_frames.store(total_frames, Ordering::SeqCst);
             duration_frames.store(total_frames, Ordering::SeqCst);
+
+            if visualizer_enabled {
+                match stream.playback_delay_frames() {
+                    Ok(delay) => {
+                        queued_frames = delay;
+                        if let Some(tap) = visualizer_tap.as_ref() {
+                            tap.set_output_delay_frames(queued_frames, channels);
+                        }
+                    }
+                    Err(error) => {
+                        if !delay_error_reported {
+                            log::warn!(
+                                "[ALSA Direct Engine] Visualizer delay query failed; using paced feed without playhead compensation: {}",
+                                error
+                            );
+                            delay_error_reported = true;
+                        }
+                        queued_frames = 0;
+                        if let Some(tap) = visualizer_tap.as_ref() {
+                            tap.clear_output_delay();
+                        }
+                    }
+                }
+            }
         }
 
         if source_ended {
@@ -760,6 +865,9 @@ fn alsa_writer_thread(
     }
 
     is_playing.store(false, Ordering::SeqCst);
+    if let Some(tap) = visualizer_tap.as_ref() {
+        tap.clear_output_delay();
+    }
     log::info!("[ALSA Direct Engine] Writer thread finished");
 }
 
@@ -979,3 +1087,23 @@ fn dop_writer_thread(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::alsa_pcm_feed_frames;
+
+    #[test]
+    fn alsa_pcm_feed_stays_near_ten_ms_across_pcm_rates() {
+        assert_eq!(alsa_pcm_feed_frames(44_100), 441);
+        assert_eq!(alsa_pcm_feed_frames(48_000), 480);
+        assert_eq!(alsa_pcm_feed_frames(96_000), 960);
+        assert_eq!(alsa_pcm_feed_frames(192_000), 1_920);
+        assert_eq!(alsa_pcm_feed_frames(768_000), 7_680);
+    }
+
+    #[test]
+    fn alsa_pcm_feed_has_defensive_bounds() {
+        assert_eq!(alsa_pcm_feed_frames(0), 64);
+        assert_eq!(alsa_pcm_feed_frames(u32::MAX), 8_192);
+    }
+}