chore(deps): security bumps in Cargo.lock (RUSTSEC-2026 fixes) (#16143)

Co-authored-by: changshenhan <217217832+changshenhan@users.noreply.github.com>

Cargo.lock

@@ -985,9 +985,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
 name = "bytes"
-version = "1.10.1"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
+checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
 dependencies = [
  "serde 1.0.228",
 ]
@@ -1791,9 +1791,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
 dependencies = [
  "crossbeam-utils",
 ]
@@ -6496,9 +6496,9 @@ dependencies = [
 
 [[package]]
 name = "quinn-proto"
-version = "0.11.13"
+version = "0.11.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31"
+checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e"
 dependencies = [
  "bytes",
  "getrandom 0.3.2",
@@ -7139,6 +7139,7 @@ dependencies = [
  "lazy_static",
  "libpulse-binding",
  "libpulse-simple-binding",
+ "libsamplerate-sys",
  "libxdo-sys",
  "mac_address",
  "magnum-opus",

Cargo.toml

@@ -22,7 +22,7 @@ path = "src/service.rs"
 
 [features]
 inline = []
-use_samplerate = ["samplerate"]
+use_samplerate = ["samplerate", "libsamplerate-sys"]
 use_rubato = ["rubato"]
 use_dasp = ["dasp"]
 flutter = ["flutter_rust_bridge"]
@@ -67,6 +67,7 @@ magnum-opus = { git = "https://github.com/rustdesk-org/magnum-opus" }
 dasp = { version = "0.11", features = ["signal", "interpolate-linear", "interpolate"], optional = true }
 rubato = { version = "0.12", optional = true }
 samplerate = { version = "0.2", optional = true }
+libsamplerate-sys = { version = "0.1.12", optional = true }
 uuid = { version = "1.3", features = ["v4"] }
 num_cpus = "1.15"
 bytes = { version = "1.4", features = ["serde"] }

src/audio_resampler.rs

@@ -0,0 +1,242 @@
+use hbb_common::thiserror;
+
+#[cfg(test)]
+pub(crate) mod allocation_tests;
+
+#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+mod sinc;
+
+const INTERPOLATION_MARGIN_FRAMES: usize = 2;
+const PENDING_PACKET_CAPACITY: usize = 2;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct AudioResamplerConfig {
+    pub input_rate: u32,
+    pub output_rate: u32,
+    pub channels: u16,
+}
+
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+pub(crate) enum AudioResamplerError {
+    #[error(
+        "invalid audio resampler configuration: input_rate={}, output_rate={}, channels={}",
+        .0.input_rate, .0.output_rate, .0.channels
+    )]
+    InvalidConfig(AudioResamplerConfig),
+    #[error("invalid resampler output frame size: {output_frames}")]
+    InvalidOutputFrameSize { output_frames: usize },
+    #[error("audio resampler input length {samples} is not divisible by channel count {channels}")]
+    IncompleteFrame { samples: usize, channels: usize },
+    #[error("audio resampler output capacity overflow")]
+    CapacityOverflow,
+    #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+    #[error("audio resampler backend failed: {0}")]
+    Backend(String),
+}
+
+pub(crate) struct FixedFrameAudioResampler {
+    resampler: AudioResampler,
+    output_samples: usize,
+    pending_samples: Vec<f32>,
+}
+
+#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+// SAFETY: libsamplerate's src_new state owns heap data and has no thread affinity.
+// This wrapper never exposes or shares that state; processing requires &mut self.
+unsafe impl Send for FixedFrameAudioResampler {}
+
+impl FixedFrameAudioResampler {
+    pub(crate) fn new(
+        config: AudioResamplerConfig,
+        output_frames: usize,
+    ) -> Result<Self, AudioResamplerError> {
+        if output_frames == 0 {
+            return Err(AudioResamplerError::InvalidOutputFrameSize { output_frames });
+        }
+        let channels = validate_config(config)?;
+        let output_samples = output_frames
+            .checked_mul(channels)
+            .ok_or(AudioResamplerError::CapacityOverflow)?;
+        let input_frames = output_frames
+            .checked_mul(config.input_rate as usize)
+            .ok_or(AudioResamplerError::CapacityOverflow)?
+            .div_ceil(config.output_rate as usize);
+        let capacity = output_samples
+            .checked_mul(PENDING_PACKET_CAPACITY)
+            .and_then(|samples| samples.checked_add(channels * INTERPOLATION_MARGIN_FRAMES))
+            .ok_or(AudioResamplerError::CapacityOverflow)?;
+        let mut resampler = AudioResampler::new(config)?;
+        resampler.reserve_input(input_frames)?;
+        Ok(Self {
+            resampler,
+            output_samples,
+            pending_samples: Vec::with_capacity(capacity),
+        })
+    }
+
+    pub(crate) fn process_with(
+        &mut self,
+        input: &[f32],
+        mut on_packet: impl FnMut(&[f32]),
+    ) -> Result<(), AudioResamplerError> {
+        self.resampler
+            .process_into(input, &mut self.pending_samples)?;
+        let complete_samples =
+            self.pending_samples.len() / self.output_samples * self.output_samples;
+        for packet in self.pending_samples[..complete_samples].chunks_exact(self.output_samples) {
+            on_packet(packet);
+        }
+        self.pending_samples.drain(..complete_samples);
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn process(&mut self, input: &[f32]) -> Result<Vec<Vec<f32>>, AudioResamplerError> {
+        let mut packets = Vec::new();
+        self.process_with(input, |packet| packets.push(packet.to_owned()))?;
+        Ok(packets)
+    }
+}
+
+pub(crate) struct AudioResampler {
+    #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
+    backend: StreamingLinearAudioResampler,
+    #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+    backend: sinc::SincAudioResampler,
+}
+
+impl AudioResampler {
+    pub(crate) fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
+        Ok(Self {
+            #[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+            backend: sinc::SincAudioResampler::new(config)?,
+            #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
+            backend: StreamingLinearAudioResampler::new(config)?,
+        })
+    }
+
+    pub(crate) fn process(&mut self, input: &[f32]) -> Result<Vec<f32>, AudioResamplerError> {
+        let mut output = Vec::new();
+        self.process_into(input, &mut output)?;
+        Ok(output)
+    }
+
+    // Append samples so capture can retain an incomplete output packet in the same buffer.
+    fn process_into(
+        &mut self,
+        input: &[f32],
+        output: &mut Vec<f32>,
+    ) -> Result<(), AudioResamplerError> {
+        self.backend.process_into(input, output)
+    }
+
+    fn reserve_input(&mut self, _frames: usize) -> Result<(), AudioResamplerError> {
+        #[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
+        {
+            let capacity = _frames
+                .checked_add(INTERPOLATION_MARGIN_FRAMES)
+                .and_then(|frames| frames.checked_mul(self.backend.channels))
+                .ok_or(AudioResamplerError::CapacityOverflow)?;
+            self.backend.buffered_samples.reserve(capacity);
+        }
+        Ok(())
+    }
+}
+
+#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
+struct StreamingLinearAudioResampler {
+    config: AudioResamplerConfig,
+    channels: usize,
+    buffered_samples: Vec<f32>,
+    next_position: u64,
+}
+
+#[cfg(not(all(feature = "use_samplerate", not(feature = "use_dasp"))))]
+impl StreamingLinearAudioResampler {
+    fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
+        Ok(Self {
+            config,
+            channels: validate_config(config)?,
+            buffered_samples: Vec::new(),
+            next_position: 0,
+        })
+    }
+
+    fn process_into(
+        &mut self,
+        input: &[f32],
+        output: &mut Vec<f32>,
+    ) -> Result<(), AudioResamplerError> {
+        validate_input(input, self.channels)?;
+        let capacity = self.output_capacity(input.len())?;
+        output.reserve(capacity);
+        self.buffered_samples.extend_from_slice(input);
+        while self.write_next_frame(output) {
+            self.next_position += self.config.input_rate as u64;
+        }
+        self.discard_consumed_frames();
+        Ok(())
+    }
+
+    fn output_capacity(&self, input_samples: usize) -> Result<usize, AudioResamplerError> {
+        let input_frames = input_samples / self.channels;
+        let scaled_frames = input_frames
+            .checked_mul(self.config.output_rate as usize)
+            .ok_or(AudioResamplerError::CapacityOverflow)?
+            / self.config.input_rate as usize;
+        scaled_frames
+            .checked_add(INTERPOLATION_MARGIN_FRAMES)
+            .and_then(|frames| frames.checked_mul(self.channels))
+            .ok_or(AudioResamplerError::CapacityOverflow)
+    }
+
+    fn write_next_frame(&self, output: &mut Vec<f32>) -> bool {
+        let output_rate = self.config.output_rate as u64;
+        let frame_count = self.buffered_samples.len() / self.channels;
+        let frame = (self.next_position / output_rate) as usize;
+        let fraction = self.next_position % output_rate;
+        if frame >= frame_count || (fraction != 0 && frame + 1 >= frame_count) {
+            return false;
+        }
+        let weight = fraction as f32 / output_rate as f32;
+        for channel in 0..self.channels {
+            let current = self.buffered_samples[frame * self.channels + channel];
+            let next_frame = frame + usize::from(fraction != 0);
+            let next = self.buffered_samples[next_frame * self.channels + channel];
+            output.push(current + (next - current) * weight);
+        }
+        true
+    }
+
+    fn discard_consumed_frames(&mut self) {
+        let output_rate = self.config.output_rate as u64;
+        let available_frames = self.buffered_samples.len() / self.channels;
+        let consumed_frames = ((self.next_position / output_rate) as usize).min(available_frames);
+        self.buffered_samples
+            .drain(0..consumed_frames * self.channels);
+        self.next_position -= consumed_frames as u64 * output_rate;
+    }
+}
+
+fn validate_config(config: AudioResamplerConfig) -> Result<usize, AudioResamplerError> {
+    if config.input_rate == 0 || config.output_rate == 0 || config.channels == 0 {
+        return Err(AudioResamplerError::InvalidConfig(config));
+    }
+    Ok(config.channels as usize)
+}
+
+fn validate_input(input: &[f32], channels: usize) -> Result<(), AudioResamplerError> {
+    if input.len() % channels != 0 {
+        return Err(AudioResamplerError::IncompleteFrame {
+            samples: input.len(),
+            channels,
+        });
+    }
+    Ok(())
+}
+
+#[cfg(all(test, not(all(feature = "use_samplerate", not(feature = "use_dasp")))))]
+mod tests;
+
+#[cfg(all(test, feature = "use_samplerate", not(feature = "use_dasp")))]
+mod samplerate_tests;

src/audio_resampler/allocation_tests.rs

@@ -0,0 +1,142 @@
+use super::{AudioResamplerConfig, FixedFrameAudioResampler};
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::cell::Cell;
+
+struct CountingAllocator;
+
+thread_local! {
+    static ALLOCATIONS: Cell<Option<usize>> = const { Cell::new(None) };
+}
+
+fn record_allocation() {
+    let _ = ALLOCATIONS.try_with(|count| {
+        if let Some(value) = count.get() {
+            count.set(Some(value + 1));
+        }
+    });
+}
+
+unsafe impl GlobalAlloc for CountingAllocator {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        record_allocation();
+        unsafe { System.alloc(layout) }
+    }
+
+    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
+        record_allocation();
+        unsafe { System.alloc_zeroed(layout) }
+    }
+
+    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 {
+        record_allocation();
+        unsafe { System.realloc(ptr, layout, size) }
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+
+#[global_allocator]
+static ALLOCATOR: CountingAllocator = CountingAllocator;
+
+pub(crate) fn assert_no_allocations(process: impl FnOnce()) {
+    struct ResetCounter;
+    impl Drop for ResetCounter {
+        fn drop(&mut self) {
+            ALLOCATIONS.with(|count| count.set(None));
+        }
+    }
+
+    ALLOCATIONS.with(|count| assert!(count.replace(Some(0)).is_none()));
+    let reset = ResetCounter;
+    process();
+    let allocations = ALLOCATIONS.with(|count| count.get().unwrap());
+    drop(reset);
+    assert_eq!(
+        allocations, 0,
+        "PCM processing allocated on the capture thread"
+    );
+}
+
+#[test]
+fn capture_resampling_reuses_buffers() {
+    const PACKETS_PER_SECOND: usize = 100;
+    const PACKET_COUNT: usize = 100;
+    const MAX_STARTUP_DELAY_PACKETS: usize = 1;
+    const SIGNAL_LEVEL: f32 = 0.25;
+    const RATE_PAIRS: [(u32, u32); 6] = [
+        (32_000, 24_000),
+        (44_100, 24_000),
+        (44_100, 48_000),
+        (48_000, 24_000),
+        (96_000, 48_000),
+        (192_000, 48_000),
+    ];
+
+    for (input_rate, output_rate) in RATE_PAIRS {
+        for channels in [1, 2, 4, 6, 8] {
+            let config = AudioResamplerConfig {
+                input_rate,
+                output_rate,
+                channels,
+            };
+            let input =
+                vec![SIGNAL_LEVEL; input_rate as usize / PACKETS_PER_SECOND * channels as usize];
+            let frames = output_rate as usize / PACKETS_PER_SECOND;
+            let mut resampler = FixedFrameAudioResampler::new(config, frames).unwrap();
+            let mut packets = 0;
+            let mut energy = 0.0;
+            assert_no_allocations(|| {
+                for _ in 0..PACKET_COUNT {
+                    resampler
+                        .process_with(&input, |packet| {
+                            assert_eq!(packet.len(), frames * channels as usize);
+                            energy += packet.iter().map(|sample| sample * sample).sum::<f32>();
+                            packets += 1;
+                        })
+                        .unwrap();
+                }
+            });
+            assert!((PACKET_COUNT - MAX_STARTUP_DELAY_PACKETS..=PACKET_COUNT).contains(&packets));
+            assert!(energy > SIGNAL_LEVEL);
+        }
+    }
+}
+
+#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
+#[test]
+fn sinc_output_matches_the_existing_backend() {
+    use super::AudioResampler;
+
+    const INPUT_FRAMES: usize = 2_048;
+    const CHUNK_FRAMES: usize = 73;
+    const SIGNAL_STEP: f32 = 0.07;
+    for (input_rate, output_rate) in [(44_100, 24_000), (44_100, 48_000), (96_000, 48_000)] {
+        for channels in [1, 2, 4, 6, 8] {
+            let config = AudioResamplerConfig {
+                input_rate,
+                output_rate,
+                channels,
+            };
+            let input: Vec<_> = (0..INPUT_FRAMES * channels as usize)
+                .map(|sample| (sample as f32 * SIGNAL_STEP).sin())
+                .collect();
+            let mut actual = AudioResampler::new(config).unwrap();
+            let expected = samplerate::Samplerate::new(
+                samplerate::ConverterType::SincBestQuality,
+                input_rate,
+                output_rate,
+                channels as usize,
+            )
+            .unwrap();
+            for chunk in input.chunks(CHUNK_FRAMES * channels as usize) {
+                assert_eq!(
+                    actual.process(chunk).unwrap(),
+                    expected.process(chunk).unwrap()
+                );
+                assert_eq!(actual.process(&[]).unwrap(), expected.process(&[]).unwrap());
+            }
+        }
+    }
+}

src/audio_resampler/samplerate_tests.rs

@@ -0,0 +1,157 @@
+use super::{AudioResampler, AudioResamplerConfig, AudioResamplerError, FixedFrameAudioResampler};
+
+const INPUT_RATE: u32 = 44_100;
+const OUTPUT_RATE: u32 = 48_000;
+const CHANNELS: u16 = 2;
+const INPUT_PACKET_FRAMES: usize = INPUT_RATE as usize / PACKETS_PER_SECOND;
+const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND;
+const PACKET_COUNT: usize = 20;
+const PACKETS_PER_SECOND: usize = 100;
+const MIN_CONTINUITY_PACKETS: usize = 2;
+const TONE_FREQUENCY_HZ: f32 = 997.0;
+const TONE_AMPLITUDE: f32 = 0.5;
+const MAX_BOUNDARY_RESIDUAL: f32 = 0.02;
+const INCOMPLETE_SAMPLE_COUNT: usize = 1;
+const DOWNSAMPLE_RATE: u32 = 24_000;
+const REJECTED_TONE_HZ: f64 = 18_000.0;
+const MAX_ALIAS_RMS: f64 = 0.01;
+const MIN_PASSBAND_RMS: f64 = 0.3;
+
+fn stereo_tone(frames: usize) -> Vec<f32> {
+    (0..frames)
+        .flat_map(|frame| {
+            let phase =
+                std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / INPUT_RATE as f32;
+            let sample = TONE_AMPLITUDE * phase.sin();
+            [sample, sample]
+        })
+        .collect()
+}
+
+fn maximum_boundary_residual(packets: &[Vec<f32>]) -> f32 {
+    packets.windows(2).fold(0.0, |maximum, pair| {
+        let previous = &pair[0];
+        let current = &pair[1];
+        let last = previous.len() - CHANNELS as usize;
+        let penultimate = last - CHANNELS as usize;
+        (0..CHANNELS as usize).fold(maximum, |maximum, channel| {
+            let predicted = previous[last + channel]
+                + (previous[last + channel] - previous[penultimate + channel]);
+            maximum.max((current[channel] - predicted).abs())
+        })
+    })
+}
+
+fn stereo_config() -> AudioResamplerConfig {
+    AudioResamplerConfig {
+        input_rate: INPUT_RATE,
+        output_rate: OUTPUT_RATE,
+        channels: CHANNELS,
+    }
+}
+
+#[test]
+fn moving_capture_resampler_preserves_pending_audio() {
+    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
+    let packet_samples = INPUT_PACKET_FRAMES * CHANNELS as usize;
+    let mut expected_resampler =
+        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
+    let expected: Vec<_> = input
+        .chunks(packet_samples)
+        .flat_map(|packet| expected_resampler.process(packet).unwrap())
+        .collect();
+    let mut moved_resampler =
+        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
+    let mut output = moved_resampler.process(&input[..packet_samples]).unwrap();
+    let remaining = std::thread::spawn(move || {
+        input[packet_samples..]
+            .chunks(packet_samples)
+            .flat_map(|packet| moved_resampler.process(packet).unwrap())
+            .collect::<Vec<_>>()
+    })
+    .join()
+    .unwrap();
+    output.extend(remaining);
+
+    assert!(output.len() >= MIN_CONTINUITY_PACKETS);
+    assert!(output
+        .iter()
+        .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize));
+    assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL);
+    assert_eq!(output, expected);
+}
+
+fn downsampled_rms(input: &[f32]) -> f64 {
+    let config = AudioResamplerConfig {
+        output_rate: DOWNSAMPLE_RATE,
+        ..stereo_config()
+    };
+    let output_frames = DOWNSAMPLE_RATE as usize / PACKETS_PER_SECOND;
+    let mut resampler = FixedFrameAudioResampler::new(config, output_frames).unwrap();
+    let output: Vec<f32> = input
+        .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize)
+        .flat_map(|packet| resampler.process(packet).unwrap().into_iter().flatten())
+        .collect();
+
+    assert!(output.len() >= output_frames * CHANNELS as usize * MIN_CONTINUITY_PACKETS);
+    let mean_square = output
+        .iter()
+        .map(|sample| f64::from(*sample).powi(2))
+        .sum::<f64>()
+        / output.len() as f64;
+    mean_square.sqrt()
+}
+
+#[test]
+fn capture_downsampling_filters_out_of_band_audio() {
+    let input: Vec<_> = (0..INPUT_PACKET_FRAMES * PACKET_COUNT)
+        .flat_map(|frame| {
+            let phase =
+                std::f64::consts::TAU * REJECTED_TONE_HZ * frame as f64 / f64::from(INPUT_RATE);
+            let sample = (f64::from(TONE_AMPLITUDE) * phase.sin()) as f32;
+            [sample, sample]
+        })
+        .collect();
+    let rms = downsampled_rms(&input);
+    assert!(
+        rms < MAX_ALIAS_RMS,
+        "out-of-band output RMS {rms} exceeded {MAX_ALIAS_RMS}"
+    );
+
+    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
+    let rms = downsampled_rms(&input);
+    assert!(
+        rms > MIN_PASSBAND_RMS,
+        "in-band output RMS {rms} fell below {MIN_PASSBAND_RMS}"
+    );
+}
+
+#[test]
+fn samplerate_backend_preserves_streaming_continuity() {
+    let input = stereo_tone(INPUT_PACKET_FRAMES * PACKET_COUNT);
+    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
+    let packets: Vec<_> = input
+        .chunks(INPUT_PACKET_FRAMES * CHANNELS as usize)
+        .map(|packet| resampler.process(packet).unwrap())
+        .filter(|packet| !packet.is_empty())
+        .collect();
+
+    assert!(packets.len() >= MIN_CONTINUITY_PACKETS);
+    assert!(packets
+        .iter()
+        .all(|packet| packet.len() % CHANNELS as usize == 0));
+    assert!(maximum_boundary_residual(&packets) <= MAX_BOUNDARY_RESIDUAL);
+}
+
+#[test]
+fn samplerate_backend_reports_incomplete_frame_context() {
+    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
+
+    assert_eq!(
+        resampler.process(&[0.0]).unwrap_err(),
+        AudioResamplerError::IncompleteFrame {
+            samples: INCOMPLETE_SAMPLE_COUNT,
+            channels: CHANNELS as usize,
+        }
+    );
+}

src/audio_resampler/sinc.rs

@@ -0,0 +1,112 @@
+use super::{AudioResamplerConfig, AudioResamplerError};
+use libsamplerate_sys as sys;
+use std::ptr::NonNull;
+
+const OUTPUT_MARGIN_FRAMES: usize = 1;
+
+pub(super) struct SincAudioResampler {
+    state: NonNull<sys::SRC_STATE>,
+    config: AudioResamplerConfig,
+}
+
+impl SincAudioResampler {
+    pub(super) fn new(config: AudioResamplerConfig) -> Result<Self, AudioResamplerError> {
+        super::validate_config(config)?;
+        let ratio = f64::from(config.output_rate) / f64::from(config.input_rate);
+        if unsafe { sys::src_is_valid_ratio(ratio) } == 0 {
+            return Err(backend_error(
+                config,
+                samplerate::ErrorCode::BadSrcRatio as _,
+            ));
+        }
+        let mut error = 0;
+        // SAFETY: src_new allocates independent state; this owner releases it in Drop.
+        let state = unsafe {
+            sys::src_new(
+                sys::SRC_SINC_BEST_QUALITY as _,
+                config.channels.into(),
+                &mut error,
+            )
+        };
+        let state = NonNull::new(state).ok_or_else(|| backend_error(config, error))?;
+        Ok(Self { state, config })
+    }
+
+    pub(super) fn process_into(
+        &mut self,
+        input: &[f32],
+        output: &mut Vec<f32>,
+    ) -> Result<(), AudioResamplerError> {
+        super::validate_input(input, self.config.channels as usize)?;
+        let mut consumed = 0;
+        loop {
+            let (used, generated) = self.process_block(&input[consumed..], output)?;
+            consumed += used;
+            if consumed == input.len() {
+                return Ok(());
+            }
+            if used == 0 && generated == 0 {
+                return Err(AudioResamplerError::Backend(
+                    "libsamplerate made no progress while input remained".to_owned(),
+                ));
+            }
+        }
+    }
+
+    fn process_block(
+        &mut self,
+        input: &[f32],
+        output: &mut Vec<f32>,
+    ) -> Result<(usize, usize), AudioResamplerError> {
+        let channels = self.config.channels as usize;
+        let input_frames = input.len() / channels;
+        let output_frames = input_frames
+            .checked_mul(self.config.output_rate as usize)
+            .map(|frames| frames / self.config.input_rate as usize)
+            .and_then(|frames| frames.checked_add(OUTPUT_MARGIN_FRAMES))
+            .ok_or(AudioResamplerError::CapacityOverflow)?;
+        let start = output.len();
+        let end = output_frames
+            .checked_mul(channels)
+            .and_then(|samples| start.checked_add(samples))
+            .ok_or(AudioResamplerError::CapacityOverflow)?;
+        let mut data = sys::SRC_DATA {
+            data_in: input.as_ptr(),
+            input_frames: input_frames
+                .try_into()
+                .map_err(|_| AudioResamplerError::CapacityOverflow)?,
+            output_frames: output_frames
+                .try_into()
+                .map_err(|_| AudioResamplerError::CapacityOverflow)?,
+            src_ratio: f64::from(self.config.output_rate) / f64::from(self.config.input_rate),
+            ..Default::default()
+        };
+        output.resize(end, 0.0);
+        data.data_out = output[start..].as_mut_ptr();
+        // SAFETY: state is exclusively owned; disjoint slices cover the declared frame counts.
+        let error = unsafe { sys::src_process(self.state.as_ptr(), &mut data) };
+        let generated = data.output_frames_gen as usize * channels;
+        output.truncate(start + generated);
+        if error != 0 {
+            return Err(backend_error(self.config, error));
+        }
+        Ok((data.input_frames_used as usize * channels, generated))
+    }
+}
+
+impl Drop for SincAudioResampler {
+    fn drop(&mut self) {
+        // SAFETY: this owner holds the only handle returned by src_new.
+        unsafe { sys::src_delete(self.state.as_ptr()) };
+    }
+}
+
+fn backend_error(config: AudioResamplerConfig, code: i32) -> AudioResamplerError {
+    AudioResamplerError::Backend(format!(
+        "input_rate={}, output_rate={}, channels={}: {:?}",
+        config.input_rate,
+        config.output_rate,
+        config.channels,
+        samplerate::Error::from_int(code)
+    ))
+}

src/audio_resampler/tests.rs

@@ -0,0 +1,178 @@
+use super::{AudioResampler, AudioResamplerConfig, FixedFrameAudioResampler};
+
+const INPUT_RATE: u32 = 24_000;
+const OUTPUT_RATE: u32 = 48_000;
+const CHANNELS: u16 = 2;
+const CHUNK_FRAMES: usize = 240;
+const CHUNK_COUNT: usize = 4;
+const TONE_FREQUENCY_HZ: f32 = 997.0;
+const TONE_AMPLITUDE: f32 = 0.5;
+const MAX_BOUNDARY_RESIDUAL: f32 = 0.02;
+const LOOK_AHEAD_OUTPUT_FRAMES: usize = 1;
+const UNEVEN_CHUNK_FRAMES: usize = 73;
+const MONO_CHANNELS: u16 = 1;
+const UNIT_RATE: u32 = 1;
+const DOUBLE_RATE: u32 = 2;
+const FIRST_DOWNSAMPLE_PACKET: [f32; 3] = [0.0, 1.0, 2.0];
+const SECOND_DOWNSAMPLE_PACKET: [f32; 4] = [3.0, 4.0, 5.0, 6.0];
+const EXPECTED_DOWNSAMPLED_OUTPUT: [f32; 4] = [0.0, 2.0, 4.0, 6.0];
+const PACKETS_PER_SECOND: usize = 100;
+const OUTPUT_PACKET_FRAMES: usize = OUTPUT_RATE as usize / PACKETS_PER_SECOND;
+const RATE_44_1_KHZ: u32 = 44_100;
+const FLOAT_TOLERANCE: f32 = 0.000_001;
+const MIN_CONTINUITY_PACKETS: usize = 2;
+
+fn stereo_tone_at_rate(frames: usize, sample_rate: u32) -> Vec<f32> {
+    (0..frames)
+        .flat_map(|frame| {
+            let phase =
+                std::f32::consts::TAU * TONE_FREQUENCY_HZ * frame as f32 / sample_rate as f32;
+            let sample = TONE_AMPLITUDE * phase.sin();
+            [sample, sample]
+        })
+        .collect()
+}
+
+fn stereo_tone(frames: usize) -> Vec<f32> {
+    stereo_tone_at_rate(frames, INPUT_RATE)
+}
+
+fn maximum_tone_prediction_residual(sample_rate: u32) -> f32 {
+    let half_step = std::f32::consts::PI * TONE_FREQUENCY_HZ / sample_rate as f32;
+    4.0 * TONE_AMPLITUDE * half_step.sin().powi(2)
+}
+
+fn maximum_boundary_residual(chunks: &[Vec<f32>]) -> f32 {
+    chunks.windows(2).fold(0.0, |maximum, pair| {
+        let previous = &pair[0];
+        let current = &pair[1];
+        let last = previous.len() - CHANNELS as usize;
+        let penultimate = last - CHANNELS as usize;
+        (0..CHANNELS as usize).fold(maximum, |maximum, channel| {
+            let predicted = previous[last + channel]
+                + (previous[last + channel] - previous[penultimate + channel]);
+            maximum.max((current[channel] - predicted).abs())
+        })
+    })
+}
+
+fn stereo_config() -> AudioResamplerConfig {
+    AudioResamplerConfig {
+        input_rate: INPUT_RATE,
+        output_rate: OUTPUT_RATE,
+        channels: CHANNELS,
+    }
+}
+
+#[test]
+fn preserves_decoded_packet_continuity_and_output_ratio() {
+    let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT);
+    let mut whole_resampler = AudioResampler::new(stereo_config()).unwrap();
+    let whole_output = whole_resampler.process(&input).unwrap();
+    let expected_frames = CHUNK_FRAMES * CHUNK_COUNT * OUTPUT_RATE as usize / INPUT_RATE as usize
+        - LOOK_AHEAD_OUTPUT_FRAMES;
+
+    for chunk_frames in [CHUNK_FRAMES, UNEVEN_CHUNK_FRAMES] {
+        let mut resampler = AudioResampler::new(stereo_config()).unwrap();
+        let output: Vec<_> = input
+            .chunks(chunk_frames * CHANNELS as usize)
+            .map(|chunk| resampler.process(chunk).unwrap())
+            .collect();
+        let residual = maximum_boundary_residual(&output);
+        assert!(
+            residual <= MAX_BOUNDARY_RESIDUAL,
+            "packet boundary residual {residual} exceeded {MAX_BOUNDARY_RESIDUAL}, chunk_frames={chunk_frames}"
+        );
+        let output_frames = output.iter().map(Vec::len).sum::<usize>() / CHANNELS as usize;
+        assert_eq!(output_frames, expected_frames);
+        assert_eq!(output.concat(), whole_output, "chunk_frames={chunk_frames}");
+    }
+}
+
+#[test]
+fn rejects_incomplete_interleaved_frames() {
+    let mut resampler = AudioResampler::new(stereo_config()).unwrap();
+
+    assert!(resampler.process(&[TONE_AMPLITUDE]).is_err());
+}
+
+#[test]
+fn interpolates_mono_samples() {
+    let config = AudioResamplerConfig {
+        input_rate: UNIT_RATE,
+        output_rate: DOUBLE_RATE,
+        channels: MONO_CHANNELS,
+    };
+    let mut resampler = AudioResampler::new(config).unwrap();
+
+    assert_eq!(
+        resampler.process(&[0.0, 1.0, 2.0]).unwrap(),
+        [0.0, 0.5, 1.0, 1.5, 2.0]
+    );
+}
+
+#[test]
+fn rejects_zero_rate_configuration() {
+    let config = AudioResamplerConfig {
+        input_rate: 0,
+        output_rate: OUTPUT_RATE,
+        channels: CHANNELS,
+    };
+
+    assert!(AudioResampler::new(config).is_err());
+}
+
+#[test]
+fn downsamples_across_packet_boundaries() {
+    let config = AudioResamplerConfig {
+        input_rate: DOUBLE_RATE,
+        output_rate: UNIT_RATE,
+        channels: MONO_CHANNELS,
+    };
+    let mut resampler = AudioResampler::new(config).unwrap();
+    let mut output = resampler.process(&FIRST_DOWNSAMPLE_PACKET).unwrap();
+    output.extend(resampler.process(&SECOND_DOWNSAMPLE_PACKET).unwrap());
+
+    assert_eq!(output, EXPECTED_DOWNSAMPLED_OUTPUT);
+}
+
+#[test]
+fn sender_resampler_emits_only_complete_continuous_frames() {
+    let input = stereo_tone(CHUNK_FRAMES * CHUNK_COUNT);
+    let mut resampler =
+        FixedFrameAudioResampler::new(stereo_config(), OUTPUT_PACKET_FRAMES).unwrap();
+    let output: Vec<_> = input
+        .chunks(CHUNK_FRAMES * CHANNELS as usize)
+        .flat_map(|chunk| resampler.process(chunk).unwrap())
+        .collect();
+
+    assert!(output.len() >= MIN_CONTINUITY_PACKETS);
+    assert!(output
+        .iter()
+        .all(|packet| packet.len() == OUTPUT_PACKET_FRAMES * CHANNELS as usize));
+    assert!(maximum_boundary_residual(&output) <= MAX_BOUNDARY_RESIDUAL);
+}
+
+#[test]
+fn sender_downsampling_preserves_packet_continuity() {
+    let input_packet_frames = RATE_44_1_KHZ as usize / PACKETS_PER_SECOND;
+    let input = stereo_tone_at_rate(input_packet_frames * CHUNK_COUNT, RATE_44_1_KHZ);
+    let config = AudioResamplerConfig {
+        input_rate: RATE_44_1_KHZ,
+        output_rate: INPUT_RATE,
+        channels: CHANNELS,
+    };
+    let mut resampler =
+        FixedFrameAudioResampler::new(config, INPUT_RATE as usize / PACKETS_PER_SECOND).unwrap();
+    let packets: Vec<_> = input
+        .chunks(input_packet_frames * CHANNELS as usize)
+        .flat_map(|packet| resampler.process(packet).unwrap())
+        .collect();
+    let residual = maximum_boundary_residual(&packets);
+
+    assert_eq!(packets.len(), CHUNK_COUNT);
+    assert!(
+        residual <= maximum_tone_prediction_residual(INPUT_RATE) + FLOAT_TOLERANCE,
+        "sender packet boundary residual {residual} exceeded the tone curvature"
+    );
+}

src/client.rs

@@ -93,6 +93,10 @@ use crate::ui_session_interface::SessionPermissionConfig;
 
 pub use super::lang::*;
 
+#[cfg(not(target_os = "linux"))]
+mod audio_playback;
+#[cfg(all(test, not(target_os = "linux")))]
+mod audio_state_tests;
 pub mod file_trait;
 pub mod helper;
 pub mod io_loop;
@@ -2053,6 +2057,8 @@ pub struct AudioHandler {
     simple: Option<psimple::Simple>,
     #[cfg(not(target_os = "linux"))]
     audio_buffer: AudioBuffer,
+    #[cfg(not(target_os = "linux"))]
+    audio_resampler: Option<crate::audio_resampler::AudioResampler>,
     sample_rate: (u32, u32),
     #[cfg(not(target_os = "linux"))]
     audio_stream: Option<Box<dyn StreamTrait>>,
@@ -2060,7 +2066,55 @@ pub struct AudioHandler {
     #[cfg(not(target_os = "linux"))]
     device_channel: u16,
     #[cfg(not(target_os = "linux"))]
-    ready: Arc<std::sync::Mutex<bool>>,
+    playback_status: Arc<audio_playback::AudioPlaybackStatus>,
+}
+
+#[cfg(not(target_os = "linux"))]
+#[derive(Clone, Copy)]
+struct DecodedAudioConfig {
+    sample_rate: u32,
+    input_channels: u16,
+    output_channels: u16,
+}
+
+#[cfg(not(target_os = "linux"))]
+fn create_audio_resampler(
+    input_rate: u32,
+    output_rate: u32,
+    channels: u16,
+) -> ResultType<Option<crate::audio_resampler::AudioResampler>> {
+    if input_rate == output_rate {
+        return Ok(None);
+    }
+    Ok(Some(crate::audio_resampler::AudioResampler::new(
+        crate::audio_resampler::AudioResamplerConfig {
+            input_rate,
+            output_rate,
+            channels,
+        },
+    )?))
+}
+
+#[cfg(not(target_os = "linux"))]
+fn prepare_decoded_audio(
+    input: &[f32],
+    resampler: Option<&mut crate::audio_resampler::AudioResampler>,
+    config: DecodedAudioConfig,
+) -> Result<Vec<f32>, crate::audio_resampler::AudioResamplerError> {
+    let mut output = match resampler {
+        Some(resampler) => resampler.process(input)?,
+        None => input.to_owned(),
+    };
+    if config.input_channels != config.output_channels {
+        output = crate::audio_rechannel(
+            output,
+            config.sample_rate,
+            config.sample_rate,
+            config.input_channels,
+            config.output_channels,
+        );
+    }
+    Ok(output)
 }
 
 #[cfg(not(target_os = "linux"))]
@@ -2068,6 +2122,7 @@ struct AudioBuffer(
     pub Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
     usize,
     [usize; 30],
+    Arc<std::sync::atomic::AtomicUsize>,
 );
 
 #[cfg(not(target_os = "linux"))]
@@ -2079,6 +2134,7 @@ impl Default for AudioBuffer {
             )),
             48000 * 2,
             [0; 30],
+            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
         )
     }
 }
@@ -2153,27 +2209,36 @@ impl AudioBuffer {
         let skip = (cap * max / (30 * N) + 1) & (!1);
         if (having > skip * 3) && (skip > 0) {
             lock.skip(skip);
-            log::info!("skip {skip}, based {max} {zero}");
+            let generation = self.signal_discontinuity();
+            drop(lock);
+            log::info!("skip {skip}, based {max} {zero}, generation={generation}");
         }
     }
 
+    /// The caller must hold the PCM buffer lock while signaling the discard.
+    fn signal_discontinuity(&self) -> usize {
+        self.3
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+            .wrapping_add(1)
+    }
+
     /// append pcm to audio buffer, if buffered data
     /// exceeds AUDIO_BUFFER_MS,  only AUDIO_BUFFER_MS
     /// will be kept.
     fn append_pcm2(&self, buffer: &[f32]) -> usize {
         let mut lock = self.0.lock().unwrap();
         let cap = lock.capacity();
-        if buffer.len() > cap {
-            lock.push_slice_overwrite(buffer);
-            return cap;
-        }
-
         let having = lock.occupied_len() + buffer.len();
-        if having > cap {
-            lock.skip(having - cap);
-        }
         lock.push_slice_overwrite(buffer);
-        lock.occupied_len()
+        let discard = (having > cap).then(|| (having - cap, self.signal_discontinuity()));
+        let occupied = lock.occupied_len();
+        drop(lock);
+        if let Some((discarded, generation)) = discard {
+            log::debug!(
+                "Audio buffer capacity discard: samples={discarded}, generation={generation}"
+            );
+        }
+        occupied
     }
 
     /// append pcm to audio buffer, trying to drop data
@@ -2185,6 +2250,41 @@ impl AudioBuffer {
     }
 }
 
+#[cfg(all(test, not(target_os = "linux")))]
+mod audio_buffer_discontinuity_tests {
+    use super::AudioBuffer;
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc, Mutex,
+    };
+
+    const BUFFER_CAPACITY: usize = 4;
+    const BUFFER_LEVELS: usize = 30;
+    const FIRST_INPUT: [f32; 2] = [0.1, 0.2];
+    const OVERFLOWING_INPUT: [f32; 3] = [0.3, 0.4, 0.5];
+    const OVERSIZED_INPUT: [f32; 5] = [0.6, 0.7, 0.8, 0.9, 1.0];
+
+    #[test]
+    fn capacity_discards_signal_discontinuities() {
+        let audio_buffer = AudioBuffer(
+            Arc::new(Mutex::new(ringbuf::HeapRb::new(BUFFER_CAPACITY))),
+            BUFFER_CAPACITY,
+            [0; BUFFER_LEVELS],
+            Arc::new(AtomicUsize::new(0)),
+        );
+
+        assert_eq!(audio_buffer.append_pcm2(&FIRST_INPUT), FIRST_INPUT.len());
+        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 0);
+        assert_eq!(
+            audio_buffer.append_pcm2(&OVERFLOWING_INPUT),
+            BUFFER_CAPACITY
+        );
+        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 1);
+        assert_eq!(audio_buffer.append_pcm2(&OVERSIZED_INPUT), BUFFER_CAPACITY);
+        assert_eq!(audio_buffer.3.load(Ordering::Relaxed), 2);
+    }
+}
+
 impl AudioHandler {
     #[cfg(target_os = "linux")]
     fn start_audio(&mut self, format0: AudioFormat) -> ResultType<()> {
@@ -2238,6 +2338,9 @@ impl AudioHandler {
         }
 
         self.sample_rate = (format0.sample_rate, config.sample_rate.0);
+        let audio_resampler = create_audio_resampler(
+            format0.sample_rate, config.sample_rate.0, format0.channels as _,
+        )?;
         let mut build_output_stream = |config: StreamConfig| match sample_format {
             cpal::SampleFormat::I8 => self.build_output_stream::<i8>(&config, &device),
             cpal::SampleFormat::I16 => self.build_output_stream::<i16>(&config, &device),
@@ -2262,6 +2365,7 @@ impl AudioHandler {
         } else {
             build_output_stream(config)?;
         }
+        self.audio_resampler = audio_resampler;
 
         Ok(())
     }
@@ -2274,10 +2378,17 @@ impl AudioHandler {
         }
         match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) {
             Ok(d) => {
+                #[cfg(target_os = "linux")]
+                let keep_existing_stream = self.simple.is_some()
+                    && self.sample_rate.0 == f.sample_rate
+                    && u32::from(self.channels) == f.channels;
+                #[cfg(not(target_os = "linux"))]
+                let keep_existing_stream = false;
                 let buffer = vec![0.; f.sample_rate as usize * f.channels as usize];
                 self.audio_decoder = Some((d, buffer));
                 self.channels = f.channels as _;
-                allow_err!(self.start_audio(f));
+                let result = self.start_audio(f);
+                self.handle_audio_start_result(result, keep_existing_stream);
             }
             Err(err) => {
                 log::error!("Failed to create audio decoder: {}", err);
@@ -2285,11 +2396,31 @@ impl AudioHandler {
         }
     }
 
+    fn handle_audio_start_result(&mut self, result: ResultType<()>, keep_existing_stream: bool) {
+        if let Err(error) = result {
+            if keep_existing_stream {
+                log::error!(
+                    "Failed to replace audio playback stream; keeping the existing compatible stream: {error:#}"
+                );
+            } else {
+                *self = Self::default();
+                log::error!("Failed to start audio playback: {error:#}");
+            }
+        }
+    }
+
     /// Handle audio frame and play it.
     #[inline]
     pub fn handle_frame(&mut self, frame: AudioFrame) {
         #[cfg(not(target_os = "linux"))]
-        if self.audio_stream.is_none() || !self.ready.lock().unwrap().clone() {
+        self.playback_status.report_errors();
+        #[cfg(not(target_os = "linux"))]
+        if self.audio_stream.is_none()
+            || !self
+                .playback_status
+                .ready
+                .load(std::sync::atomic::Ordering::Acquire)
+        {
             return;
         }
         #[cfg(target_os = "linux")]
@@ -2298,39 +2429,40 @@ impl AudioHandler {
             return;
         }
         self.audio_decoder.as_mut().map(|(d, buffer)| {
-            if let Ok(n) = d.decode_float(&frame.data, buffer, false) {
-                let channels = self.channels;
-                let n = n * (channels as usize);
-                #[cfg(not(target_os = "linux"))]
-                {
-                    let sample_rate0 = self.sample_rate.0;
-                    let sample_rate = self.sample_rate.1;
-                    let mut buffer = buffer[0..n].to_owned();
-                    if sample_rate != sample_rate0 {
-                        buffer = crate::audio_resample(
-                            &buffer[0..n],
-                            sample_rate0,
-                            sample_rate,
-                            channels,
-                        );
-                    }
-                    if self.channels != self.device_channel {
-                        buffer = crate::audio_rechannel(
-                            buffer,
-                            sample_rate,
-                            sample_rate,
-                            self.channels,
-                            self.device_channel,
-                        );
-                    }
-                    self.audio_buffer.append_pcm(&buffer);
-                }
-                #[cfg(target_os = "linux")]
-                {
-                    let data_u8 =
-                        unsafe { std::slice::from_raw_parts::<u8>(buffer.as_ptr() as _, n * 4) };
-                    self.simple.as_mut().map(|x| x.write(data_u8));
+            let decoded_frames = match d.decode_float(&frame.data, buffer, false) {
+                Ok(decoded_frames) => decoded_frames,
+                Err(error) => {
+                    log::warn!("Failed to decode audio frame: {error:?}");
+                    return;
                 }
+            };
+            let channels = self.channels;
+            let n = decoded_frames * channels as usize;
+            #[cfg(not(target_os = "linux"))]
+            {
+                let config = DecodedAudioConfig {
+                    sample_rate: self.sample_rate.1,
+                    input_channels: self.channels,
+                    output_channels: self.device_channel,
+                };
+                let buffer = match prepare_decoded_audio(
+                    &buffer[0..n],
+                    self.audio_resampler.as_mut(),
+                    config,
+                ) {
+                    Ok(output) => output,
+                    Err(error) => {
+                        log::error!("Failed to resample decoded audio: {error:#}");
+                        return;
+                    }
+                };
+                self.audio_buffer.append_pcm(&buffer);
+            }
+            #[cfg(target_os = "linux")]
+            {
+                let data_u8 =
+                    unsafe { std::slice::from_raw_parts::<u8>(buffer.as_ptr() as _, n * 4) };
+                self.simple.as_mut().map(|x| x.write(data_u8));
             }
         });
     }
@@ -2350,63 +2482,28 @@ impl AudioHandler {
         self.audio_buffer
             .resize(config.sample_rate.0 as _, config.channels as _);
         let audio_buffer = self.audio_buffer.0.clone();
-        let ready = self.ready.clone();
+        let discontinuity_generation = self.audio_buffer.3.clone();
+        let mut playback_writer = audio_playback::AudioPlaybackWriter::new(
+            audio_playback::AudioPlaybackConfig {
+                sample_rate: config.sample_rate.0,
+                channels: config.channels as usize,
+            },
+            audio_buffer,
+            discontinuity_generation,
+        )?;
+        let playback_status = playback_writer.status.clone();
         let timeout = None;
         let stream = device.build_output_stream(
             config,
-            move |data: &mut [T], info: &cpal::OutputCallbackInfo| {
-                if !*ready.lock().unwrap() {
-                    *ready.lock().unwrap() = true;
-                }
-
-                let mut n = data.len();
-                let mut lock = audio_buffer.lock().unwrap();
-                let mut having = lock.occupied_len();
-                // android two timestamps, one from zero, another not
-                #[cfg(not(target_os = "android"))]
-                if having < n {
-                    let tms = info.timestamp();
-                    let how_long = tms
-                        .playback
-                        .duration_since(&tms.callback)
-                        .unwrap_or(Duration::from_millis(0));
-
-                    // must long enough to fight back scheuler delay
-                    if how_long > Duration::from_millis(6) && how_long < Duration::from_millis(3000)
-                    {
-                        drop(lock);
-                        std::thread::sleep(how_long.div_f32(1.2));
-                        lock = audio_buffer.lock().unwrap();
-                        having = lock.occupied_len();
-                    }
-
-                    if having < n {
-                        n = having;
-                    }
-                }
-                #[cfg(target_os = "android")]
-                if having < n {
-                    n = having;
-                }
-                let mut elems = vec![0.0f32; n];
-                if n > 0 {
-                    lock.pop_slice(&mut elems);
-                }
-                drop(lock);
-
-                let mut input = elems.into_iter();
-                for sample in data.iter_mut() {
-                    *sample = match input.next() {
-                        Some(x) => T::from_sample(x),
-                        _ => T::from_sample(0.),
-                    };
-                }
+            move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
+                playback_writer.write_output(data);
             },
             err_fn,
             timeout,
         )?;
         stream.play()?;
         self.audio_stream = Some(Box::new(stream));
+        self.playback_status = playback_status;
         Ok(())
     }
 }
@@ -2426,6 +2523,27 @@ mod audio_format_tests {
         assert!(!is_supported_audio_channel_count(0));
         assert!(!is_supported_audio_channel_count(u32::MAX));
     }
+
+    #[test]
+    fn failed_audio_start_discards_format_state() {
+        use super::{anyhow, AudioDecoder, AudioHandler, Stereo};
+
+        const SAMPLE_RATE: u32 = 48_000;
+        const CHANNELS: u16 = 2;
+        let decoder = AudioDecoder::new(SAMPLE_RATE, Stereo).unwrap();
+        let mut handler = AudioHandler {
+            audio_decoder: Some((decoder, Vec::new())),
+            sample_rate: (SAMPLE_RATE, SAMPLE_RATE),
+            channels: CHANNELS,
+            ..Default::default()
+        };
+
+        handler.handle_audio_start_result(Err(anyhow!("Injected playback startup failure")), false);
+
+        assert!(handler.audio_decoder.is_none());
+        assert_eq!(handler.channels, 0);
+        assert_eq!(handler.sample_rate, (0, 0));
+    }
 }
 
 /// Video handler for the [`Client`].

src/client/audio_playback.rs

@@ -0,0 +1,218 @@
+use hbb_common::{log, thiserror};
+use ringbuf::{ring_buffer::RbBase, Rb};
+use std::sync::{
+    atomic::{AtomicBool, AtomicUsize, Ordering},
+    TryLockError,
+};
+
+pub(super) const UNDERRUN_DECLICK_MS: usize = 5;
+const MILLISECONDS_PER_SECOND: usize = 1_000;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) struct AudioPlaybackConfig {
+    pub sample_rate: u32,
+    pub channels: usize,
+}
+
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+pub(super) enum AudioPlaybackError {
+    #[error(
+        "invalid audio playback configuration: sample_rate={}, channels={}",
+        .0.sample_rate, .0.channels
+    )]
+    InvalidConfig(AudioPlaybackConfig),
+    #[error("audio playback frame has {samples} samples for {channels} channels")]
+    IncompleteFrame { samples: usize, channels: usize },
+    #[error("audio playback transition frame count overflow")]
+    FrameCountOverflow,
+}
+
+pub(super) struct AudioPlaybackRecovery {
+    channels: usize,
+    transition_frames: usize,
+    transition_frame: usize,
+    had_input: bool,
+    transition_start: Vec<f32>,
+    output_frame: Vec<f32>,
+}
+
+#[derive(Default)]
+pub(super) struct AudioPlaybackStatus {
+    pub(super) ready: AtomicBool,
+    contentions: AtomicUsize,
+    buffer_poisoned: AtomicBool,
+}
+
+impl AudioPlaybackStatus {
+    pub(super) fn report_errors(&self) {
+        let contentions = self.contentions.swap(0, Ordering::Relaxed);
+        if contentions != 0 {
+            log::debug!("Audio playback PCM buffer contention: callbacks={contentions}");
+        }
+        if self.buffer_poisoned.swap(false, Ordering::Relaxed) {
+            log::error!("Audio playback stopped reading a poisoned PCM buffer");
+        }
+    }
+}
+
+pub(super) struct AudioPlaybackWriter {
+    audio_buffer: std::sync::Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
+    discontinuity_generation: std::sync::Arc<AtomicUsize>,
+    observed_discontinuity_generation: usize,
+    buffered_input: Vec<f32>,
+    recovery: AudioPlaybackRecovery,
+    pub(super) status: std::sync::Arc<AudioPlaybackStatus>,
+    buffer_failed: bool,
+}
+
+impl AudioPlaybackWriter {
+    pub(super) fn new(
+        config: AudioPlaybackConfig,
+        audio_buffer: std::sync::Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
+        discontinuity_generation: std::sync::Arc<AtomicUsize>,
+    ) -> Result<Self, AudioPlaybackError> {
+        let recovery = AudioPlaybackRecovery::new(config)?;
+        let buffer_capacity = audio_buffer.lock().unwrap().capacity();
+        let observed_discontinuity_generation = discontinuity_generation.load(Ordering::Relaxed);
+        Ok(Self {
+            audio_buffer,
+            discontinuity_generation,
+            observed_discontinuity_generation,
+            buffered_input: vec![0.0; buffer_capacity],
+            recovery,
+            status: Default::default(),
+            buffer_failed: false,
+        })
+    }
+
+    fn read_buffer(&mut self, requested_samples: usize) -> usize {
+        if self.buffer_failed {
+            return 0;
+        }
+        let mut buffer = match self.audio_buffer.try_lock() {
+            Ok(buffer) => buffer,
+            Err(TryLockError::WouldBlock) => {
+                // Keep queued PCM and its generation for the next successful read.
+                self.status.contentions.fetch_add(1, Ordering::Relaxed);
+                return 0;
+            }
+            Err(TryLockError::Poisoned(_)) => {
+                self.buffer_failed = true;
+                self.status.ready.store(false, Ordering::Release);
+                self.status.buffer_poisoned.store(true, Ordering::Relaxed);
+                return 0;
+            }
+        };
+        let generation = self.discontinuity_generation.load(Ordering::Relaxed);
+        let channels = self.recovery.channels;
+        let samples = buffer.occupied_len().min(requested_samples) / channels * channels;
+        buffer.pop_slice(&mut self.buffered_input[..samples]);
+        drop(buffer);
+        if generation != self.observed_discontinuity_generation {
+            self.recovery.begin_discontinuity();
+            self.observed_discontinuity_generation = generation;
+        }
+        samples
+    }
+
+    pub(super) fn write_output<T>(&mut self, output: &mut [T])
+    where
+        T: cpal::Sample + cpal::FromSample<f32>,
+    {
+        self.status
+            .ready
+            .store(!self.buffer_failed, Ordering::Release);
+        let requested_samples = output.len().min(self.buffered_input.len());
+        let channel_count = self.recovery.channels;
+        let available_samples = self.read_buffer(requested_samples);
+        let available_frames = available_samples / channel_count;
+        for (frame_index, output_frame) in output.chunks_mut(channel_count).enumerate() {
+            let input = if frame_index < available_frames {
+                let start = frame_index * channel_count;
+                Some(&self.buffered_input[start..start + channel_count])
+            } else {
+                None
+            };
+            match self.recovery.process_frame(input) {
+                Ok(recovered) => {
+                    for (output, sample) in output_frame.iter_mut().zip(recovered) {
+                        *output = T::from_sample(*sample);
+                    }
+                }
+                Err(error) => {
+                    log::error!("Failed to recover audio underflow: {error}");
+                    output_frame.fill(T::from_sample(0.0));
+                }
+            }
+        }
+    }
+}
+
+impl AudioPlaybackRecovery {
+    pub(super) fn new(config: AudioPlaybackConfig) -> Result<Self, AudioPlaybackError> {
+        if config.sample_rate == 0 || config.channels == 0 {
+            return Err(AudioPlaybackError::InvalidConfig(config));
+        }
+        let transition_frames = (config.sample_rate as usize)
+            .checked_mul(UNDERRUN_DECLICK_MS)
+            .ok_or(AudioPlaybackError::FrameCountOverflow)?
+            / MILLISECONDS_PER_SECOND;
+        if transition_frames == 0 {
+            return Err(AudioPlaybackError::InvalidConfig(config));
+        }
+        Ok(Self {
+            channels: config.channels,
+            transition_frames,
+            transition_frame: transition_frames,
+            had_input: false,
+            transition_start: vec![0.0; config.channels],
+            output_frame: vec![0.0; config.channels],
+        })
+    }
+
+    pub(super) fn process_frame(
+        &mut self,
+        input: Option<&[f32]>,
+    ) -> Result<&[f32], AudioPlaybackError> {
+        if input.is_some_and(|frame| frame.len() != self.channels) {
+            return Err(AudioPlaybackError::IncompleteFrame {
+                samples: input.map_or(0, <[f32]>::len),
+                channels: self.channels,
+            });
+        }
+        self.begin_transition(input.is_some());
+        let target_weight = self.advance_transition();
+        for channel in 0..self.channels {
+            let target = input.map_or(0.0, |frame| frame[channel]);
+            self.output_frame[channel] =
+                self.transition_start[channel] * (1.0 - target_weight) + target * target_weight;
+        }
+        Ok(&self.output_frame)
+    }
+
+    pub(super) fn begin_discontinuity(&mut self) {
+        self.transition_start.copy_from_slice(&self.output_frame);
+        self.transition_frame = 0;
+    }
+
+    fn begin_transition(&mut self, has_input: bool) {
+        if has_input == self.had_input {
+            return;
+        }
+        self.transition_start.copy_from_slice(&self.output_frame);
+        self.transition_frame = 0;
+        self.had_input = has_input;
+    }
+
+    fn advance_transition(&mut self) -> f32 {
+        if self.transition_frame >= self.transition_frames {
+            return 1.0;
+        }
+        self.transition_frame += 1;
+        self.transition_frame as f32 / self.transition_frames as f32
+    }
+}
+
+#[cfg(test)]
+#[path = "audio_playback_tests.rs"]
+mod tests;

src/client/audio_playback_tests.rs

@@ -0,0 +1,218 @@
+use super::{AudioPlaybackConfig, AudioPlaybackError, AudioPlaybackRecovery, AudioPlaybackWriter};
+use ringbuf::{ring_buffer::RbBase, Rb};
+use std::{
+    sync::{atomic::Ordering, mpsc, Arc, Mutex},
+    time::Duration,
+};
+
+const SAMPLE_RATE: u32 = 48_000;
+const CHANNELS: usize = 2;
+const ACTIVE_FRAME: [f32; CHANNELS] = [0.8, -0.8];
+const OPPOSITE_ACTIVE_FRAME: [f32; CHANNELS] = [-0.8, 0.8];
+const ACTIVE_FRAMES: usize = 300;
+const SILENT_FRAMES: usize = 300;
+const TRANSITION_FRAMES: usize =
+    SAMPLE_RATE as usize * super::UNDERRUN_DECLICK_MS / super::MILLISECONDS_PER_SECOND;
+const MAX_SAMPLE_STEP: f32 = 0.01;
+
+#[test]
+fn writing_audio_observes_discard_and_releases_buffer_lock() {
+    const INPUT: [f32; 4] = [0.1, 0.2, 0.3, 0.4];
+    const GENERATION: usize = 7;
+    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(INPUT.len())));
+    let generation = Arc::new(super::AtomicUsize::new(0));
+    let config = AudioPlaybackConfig {
+        sample_rate: SAMPLE_RATE,
+        channels: CHANNELS,
+    };
+    let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap();
+    {
+        let mut buffer = buffer.lock().unwrap();
+        buffer.push_slice(&INPUT);
+        generation.store(GENERATION, Ordering::Relaxed);
+    }
+    let mut output = [0.0_f32; INPUT.len()];
+
+    writer.write_output(&mut output);
+
+    assert_eq!(writer.buffered_input, INPUT);
+    assert_eq!(writer.observed_discontinuity_generation, GENERATION);
+    assert_eq!(buffer.try_lock().unwrap().occupied_len(), 0);
+}
+
+fn maximum_sample_step(samples: &[f32]) -> f32 {
+    samples
+        .windows(CHANNELS + 1)
+        .map(|window| (window[CHANNELS] - window[0]).abs())
+        .fold(0.0, f32::max)
+}
+
+#[test]
+fn smooths_underflow_and_explicit_audio_discontinuities() {
+    for explicit_discontinuity in [false, true] {
+        let config = AudioPlaybackConfig {
+            sample_rate: SAMPLE_RATE,
+            channels: CHANNELS,
+        };
+        let mut recovery = AudioPlaybackRecovery::new(config).unwrap();
+        let mut output = Vec::new();
+        for _ in 0..ACTIVE_FRAMES {
+            output.extend_from_slice(recovery.process_frame(Some(&ACTIVE_FRAME)).unwrap());
+        }
+        let transition_end = TRANSITION_FRAMES * CHANNELS;
+        assert_eq!(
+            &output[transition_end - CHANNELS..transition_end],
+            ACTIVE_FRAME.as_slice(),
+            "explicit_discontinuity={explicit_discontinuity}"
+        );
+        let resumed_frame = if explicit_discontinuity {
+            recovery.begin_discontinuity();
+            &OPPOSITE_ACTIVE_FRAME
+        } else {
+            for _ in 0..SILENT_FRAMES {
+                output.extend_from_slice(recovery.process_frame(None).unwrap());
+            }
+            &ACTIVE_FRAME
+        };
+        for _ in 0..ACTIVE_FRAMES {
+            output.extend_from_slice(recovery.process_frame(Some(resumed_frame)).unwrap());
+        }
+        let maximum = maximum_sample_step(&output);
+        assert!(
+            maximum <= MAX_SAMPLE_STEP,
+            "step {maximum} exceeded {MAX_SAMPLE_STEP}, explicit={explicit_discontinuity}"
+        );
+        assert_eq!(
+            &output[output.len() - CHANNELS..],
+            resumed_frame,
+            "explicit_discontinuity={explicit_discontinuity}"
+        );
+    }
+}
+
+#[test]
+fn validates_configuration_and_frame_size() {
+    let invalid_config = AudioPlaybackConfig {
+        sample_rate: 0,
+        channels: CHANNELS,
+    };
+    assert_eq!(
+        AudioPlaybackRecovery::new(invalid_config).err(),
+        Some(AudioPlaybackError::InvalidConfig(invalid_config))
+    );
+
+    let config = AudioPlaybackConfig {
+        sample_rate: SAMPLE_RATE,
+        channels: CHANNELS,
+    };
+    let mut recovery = AudioPlaybackRecovery::new(config).unwrap();
+    assert_eq!(
+        recovery.process_frame(Some(&[0.5])).err(),
+        Some(AudioPlaybackError::IncompleteFrame {
+            samples: 1,
+            channels: CHANNELS,
+        })
+    );
+}
+
+const CALLBACK_SAMPLES: usize = 64;
+const CALLBACK_TIMEOUT: Duration = Duration::from_secs(2);
+const DISCARD_GENERATION: usize = 1;
+
+fn write_while_buffer_is_locked(
+    mut writer: AudioPlaybackWriter,
+    buffer: &Arc<Mutex<ringbuf::HeapRb<f32>>>,
+    generation: &Arc<super::AtomicUsize>,
+) -> (AudioPlaybackWriter, [f32; CALLBACK_SAMPLES]) {
+    let mut guard = buffer.lock().unwrap();
+    let queued = OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES);
+    guard.push_slice(&queued);
+    generation.store(DISCARD_GENERATION, Ordering::Relaxed);
+    let (completed_tx, completed_rx) = mpsc::channel();
+    let callback = std::thread::spawn(move || {
+        let mut output = [0.0; CALLBACK_SAMPLES];
+        crate::audio_resampler::allocation_tests::assert_no_allocations(|| {
+            writer.write_output(&mut output);
+        });
+        completed_tx.send((writer, output)).unwrap();
+    });
+    let completed = completed_rx.recv_timeout(CALLBACK_TIMEOUT);
+    let retained = guard.occupied_len();
+    drop(guard);
+    callback.join().unwrap();
+    let result = completed.expect("playback callback waited for the buffer owner");
+    assert_eq!(retained, queued.len());
+    result
+}
+
+#[test]
+fn playback_contention_preserves_queued_audio_and_recovers_after_release() {
+    let samples = ACTIVE_FRAMES * CHANNELS;
+    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(samples)));
+    let generation = Arc::new(super::AtomicUsize::new(0));
+    let config = AudioPlaybackConfig {
+        sample_rate: SAMPLE_RATE,
+        channels: CHANNELS,
+    };
+    let mut writer = AudioPlaybackWriter::new(config, buffer.clone(), generation.clone()).unwrap();
+    buffer
+        .lock()
+        .unwrap()
+        .push_slice(&ACTIVE_FRAME.repeat(ACTIVE_FRAMES));
+    let mut output = vec![0.0; samples];
+    writer.write_output(&mut output);
+    assert_eq!(&output[samples - CHANNELS..], &ACTIVE_FRAME);
+
+    let (mut writer, gap) = write_while_buffer_is_locked(writer, &buffer, &generation);
+
+    assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 1);
+    assert!(writer.status.ready.load(Ordering::Acquire));
+    assert_eq!(writer.observed_discontinuity_generation, 0);
+    assert!(maximum_sample_step(&gap) <= MAX_SAMPLE_STEP);
+    assert!(gap[0] > 0.0 && gap[0] < ACTIVE_FRAME[0]);
+    assert_eq!(gap[1], -gap[0]);
+    assert!(gap[CALLBACK_SAMPLES - CHANNELS] > 0.0);
+    writer.write_output(&mut output);
+    let mut transition = gap[gap.len() - CHANNELS..].to_vec();
+    transition.extend_from_slice(&output[..TRANSITION_FRAMES * CHANNELS]);
+    assert!(maximum_sample_step(&transition) <= MAX_SAMPLE_STEP);
+    assert_eq!(
+        writer.buffered_input,
+        OPPOSITE_ACTIVE_FRAME.repeat(ACTIVE_FRAMES)
+    );
+    assert_eq!(writer.observed_discontinuity_generation, DISCARD_GENERATION);
+    assert_eq!(&output[samples - CHANNELS..], &OPPOSITE_ACTIVE_FRAME);
+    assert_eq!(buffer.lock().unwrap().occupied_len(), 0);
+}
+
+#[test]
+fn poisoned_playback_buffer_reports_once_without_panicking_in_the_callback() {
+    let buffer = Arc::new(Mutex::new(ringbuf::HeapRb::new(CALLBACK_SAMPLES)));
+    let config = AudioPlaybackConfig {
+        sample_rate: SAMPLE_RATE,
+        channels: CHANNELS,
+    };
+    let mut writer =
+        AudioPlaybackWriter::new(config, buffer.clone(), Arc::new(super::AtomicUsize::new(0)))
+            .unwrap();
+    assert!(std::thread::spawn(move || {
+        let _guard = buffer.lock().unwrap();
+        panic!("Injected PCM buffer failure");
+    })
+    .join()
+    .is_err());
+    let mut output = [ACTIVE_FRAME[0]; CALLBACK_SAMPLES];
+
+    crate::audio_resampler::allocation_tests::assert_no_allocations(|| {
+        writer.write_output(&mut output);
+    });
+
+    assert_eq!(output, [0.0; CALLBACK_SAMPLES]);
+    assert!(!writer.status.ready.load(Ordering::Acquire));
+    assert!(writer.status.buffer_poisoned.load(Ordering::Relaxed));
+    writer.status.report_errors();
+    writer.write_output(&mut output);
+    assert!(!writer.status.buffer_poisoned.load(Ordering::Relaxed));
+    assert_eq!(writer.status.contentions.load(Ordering::Relaxed), 0);
+    assert!(!writer.status.ready.load(Ordering::Acquire));
+}

src/client/audio_state_tests.rs

@@ -0,0 +1,113 @@
+use super::{create_audio_resampler, AudioDecoder, AudioFrame, AudioHandler, Stereo};
+use cpal::traits::StreamTrait;
+use hbb_common::anyhow::anyhow;
+use magnum_opus::{Application::LowDelay, Encoder};
+use ringbuf::{ring_buffer::RbBase, Rb};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+
+const INPUT_RATE: u32 = 24_000;
+const OUTPUT_RATE: u32 = 48_000;
+const CHANNELS: u16 = 2;
+const PACKETS_PER_SECOND: usize = 100;
+const MAX_PACKET_BYTES: usize = 4_096;
+const SAMPLE_VALUE: f32 = 0.25;
+
+struct TrackedAudioStream(Arc<AtomicBool>);
+
+impl StreamTrait for TrackedAudioStream {
+    fn play(&self) -> Result<(), cpal::PlayStreamError> {
+        Ok(())
+    }
+
+    fn pause(&self) -> Result<(), cpal::PauseStreamError> {
+        Ok(())
+    }
+}
+
+impl Drop for TrackedAudioStream {
+    fn drop(&mut self) {
+        self.0.store(true, Ordering::SeqCst);
+    }
+}
+
+fn decoder(sample_rate: u32) -> (AudioDecoder, Vec<f32>) {
+    (
+        AudioDecoder::new(sample_rate, Stereo).unwrap(),
+        vec![0.0; sample_rate as usize * CHANNELS as usize],
+    )
+}
+
+fn active_handler(input_rate: u32) -> (AudioHandler, Arc<AtomicBool>) {
+    let dropped = Arc::new(AtomicBool::new(false));
+    let handler = AudioHandler {
+        audio_decoder: Some(decoder(input_rate)),
+        audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(),
+        sample_rate: (input_rate, OUTPUT_RATE),
+        audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))),
+        channels: CHANNELS,
+        device_channel: CHANNELS,
+        ..Default::default()
+    };
+    handler.playback_status.ready.store(true, Ordering::Release);
+    (handler, dropped)
+}
+
+fn audio_frame() -> AudioFrame {
+    let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize;
+    let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap();
+    AudioFrame {
+        data: encoder
+            .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES)
+            .unwrap()
+            .into(),
+        ..Default::default()
+    }
+}
+
+#[test]
+fn failed_format_change_discards_old_playback_state() {
+    let (mut handler, dropped) = active_handler(INPUT_RATE);
+    handler
+        .audio_buffer
+        .0
+        .lock()
+        .unwrap()
+        .push_slice(&[SAMPLE_VALUE; CHANNELS as usize]);
+    handler.audio_decoder = Some(decoder(OUTPUT_RATE));
+    handler.sample_rate = (OUTPUT_RATE, OUTPUT_RATE);
+
+    handler.handle_audio_start_result(
+        Err(anyhow!("Injected output stream startup failure")),
+        false,
+    );
+
+    assert!(dropped.load(Ordering::SeqCst));
+    assert!(handler.audio_stream.is_none());
+    assert!(handler.audio_resampler.is_none());
+    assert!(handler.audio_decoder.is_none());
+    assert!(!handler.playback_status.ready.load(Ordering::Acquire));
+    handler.handle_frame(audio_frame());
+    assert_eq!(handler.audio_buffer.0.lock().unwrap().occupied_len(), 0);
+}
+
+#[test]
+fn successful_start_or_compatible_failure_preserves_audio_packet_duration() {
+    for result in [
+        Ok(()),
+        Err(anyhow!("Injected compatible stream replacement failure")),
+    ] {
+        let (mut handler, dropped) = active_handler(OUTPUT_RATE);
+
+        handler.handle_audio_start_result(result, true);
+        handler.handle_frame(audio_frame());
+
+        assert!(!dropped.load(Ordering::SeqCst));
+        assert_eq!(
+            handler.audio_buffer.0.lock().unwrap().occupied_len(),
+            OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize
+        );
+    }
+}

src/common.rs

@@ -408,6 +408,11 @@ pub fn resample_channels(
     }
 }
 
+#[cfg(all(feature = "use_dasp", feature = "use_samplerate"))]
+compile_error!(
+    "features `use_dasp` and `use_samplerate` are mutually exclusive; disable default features before selecting `use_samplerate`"
+);
+
 #[cfg(feature = "use_dasp")]
 pub fn audio_resample(
     data: &[f32],
@@ -444,7 +449,7 @@ pub fn audio_resample(
     }
 }
 
-#[cfg(feature = "use_samplerate")]
+#[cfg(all(feature = "use_samplerate", not(feature = "use_dasp")))]
 pub fn audio_resample(
     data: &[f32],
     sample_rate0: u32,

src/lib.rs

@@ -1,3 +1,5 @@
+#[cfg(any(test, not(target_os = "linux")))]
+mod audio_resampler;
 mod keyboard;
 /// cbindgen:ignore
 pub mod platform;

src/server/audio_service.rs

@@ -15,7 +15,9 @@
 use super::*;
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
 use hbb_common::anyhow::anyhow;
-use magnum_opus::{Application::*, Channels::*, Encoder};
+#[cfg(any(target_os = "linux", target_os = "android"))]
+use magnum_opus::Application::LowDelay;
+use magnum_opus::{Channels::*, Encoder};
 use std::sync::atomic::{AtomicBool, Ordering};
 
 pub const NAME: &'static str = "audio";
@@ -97,10 +99,11 @@ mod pa_impl {
         RESTARTING.store(false, Ordering::SeqCst);
         #[cfg(target_os = "linux")]
         let mut stream = crate::ipc::connect(1000, "_pa").await?;
-        unsafe {
-            AUDIO_ZERO_COUNT = 0;
-        }
-        let mut encoder = Encoder::new(crate::platform::PA_SAMPLE_RATE, Stereo, LowDelay)?;
+        let mut encoder = AudioEncoder::new(Encoder::new(
+            crate::platform::PA_SAMPLE_RATE,
+            Stereo,
+            LowDelay,
+        )?);
         #[cfg(target_os = "linux")]
         allow_err!(
             stream
@@ -172,8 +175,11 @@ pub fn is_screen_capture_kit_available() -> bool {
 }
 
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
-#[path = "audio_capture_error.rs"]
+mod audio_capture;
+#[cfg(not(any(target_os = "linux", target_os = "android")))]
 mod audio_capture_error;
+#[cfg(not(any(target_os = "linux", target_os = "android")))]
+mod audio_capture_queue;
 
 #[cfg(not(any(target_os = "linux", target_os = "android")))]
 mod cpal_impl {
@@ -182,14 +188,15 @@ mod cpal_impl {
     use super::*;
     use cpal::{
         traits::{DeviceTrait, HostTrait, StreamTrait},
-        BufferSize, Device, Host, InputCallbackInfo, StreamConfig, SupportedStreamConfig,
+        Device, Host, InputCallbackInfo, SupportedStreamConfig,
     };
 
     lazy_static::lazy_static! {
         static ref HOST: Host = cpal::default_host();
-        static ref INPUT_BUFFER: Arc<Mutex<std::collections::VecDeque<f32>>> = Default::default();
     }
 
+    const AUDIO_PACKETS_PER_SECOND: usize = 100;
+
     #[cfg(feature = "screencapturekit")]
     lazy_static::lazy_static! {
         static ref HOST_SCREEN_CAPTURE_KIT: Result<Host, cpal::HostUnavailable> = cpal::host_from_id(cpal::HostId::ScreenCaptureKit);
@@ -197,7 +204,20 @@ mod cpal_impl {
 
     #[derive(Default)]
     pub struct State {
-        stream: Option<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)>,
+        stream: Option<ActiveCaptureStream>,
+    }
+
+    struct ActiveCaptureStream {
+        stream: Option<Box<dyn StreamTrait>>,
+        format: Arc<Message>,
+        _encoder_worker: audio_capture_queue::CaptureEncoderWorker,
+        errors: CaptureErrorHandler,
+    }
+
+    impl Drop for ActiveCaptureStream {
+        fn drop(&mut self) {
+            self.stream.take();
+        }
     }
 
     impl super::service::Reset for State {
@@ -215,8 +235,8 @@ mod cpal_impl {
             }
             _ => {}
         }
-        if let Some((_, format, _)) = &state.stream {
-            sp.send_shared(format.clone());
+        if let Some(stream) = &state.stream {
+            sp.send_shared(stream.format.clone());
             #[cfg(target_os = "macos")]
             log::info!("Audio capture stream recreated; replacement format sent");
         }
@@ -232,8 +252,8 @@ mod cpal_impl {
                 }
                 _ => {}
             }
-            if let Some((_, format, _)) = &state.stream {
-                sps.send_shared(format.clone());
+            if let Some(stream) = &state.stream {
+                sps.send_shared(stream.format.clone());
             }
             Ok(())
         })?;
@@ -241,10 +261,10 @@ mod cpal_impl {
     }
 
     pub fn run(sp: EmptyExtraFieldService, state: &mut State) -> ResultType<()> {
-        if let Some((_, _, errors)) = &state.stream {
-            if errors.needs_restart() {
-                // Recreate on the service thread, outside the backend's error callback.
-                log::warn!("Recreating interrupted audio capture stream");
+        if let Some(stream) = &state.stream {
+            if stream.errors.needs_restart() {
+                // Recreate on the service thread, outside the capture callbacks.
+                log::warn!("Recreating audio capture stream after an error");
                 super::restart();
             }
         }
@@ -255,29 +275,89 @@ mod cpal_impl {
         }
     }
 
-    fn send(
-        data: Vec<f32>,
-        sample_rate0: u32,
-        sample_rate: u32,
+    #[derive(Clone, Copy)]
+    struct CaptureFrameProcessorConfig {
+        input_rate: u32,
+        output_rate: u32,
         device_channel: u16,
         encode_channel: u16,
-        encoder: &mut Encoder,
-        sp: &GenericService,
-    ) {
-        let mut data = data;
-        if sample_rate0 != sample_rate {
-            data = crate::common::audio_resample(&data, sample_rate0, sample_rate, device_channel);
+    }
+
+    struct CaptureFrameProcessor {
+        config: CaptureFrameProcessorConfig,
+        resampler: Option<crate::audio_resampler::FixedFrameAudioResampler>,
+        sender: audio_capture_queue::CapturePcmSender,
+        rechannel_buffer: Vec<f32>,
+    }
+
+    struct CaptureStreamOutput {
+        sender: audio_capture_queue::CapturePcmSender,
+        sample_rate: u32,
+        encode_channel: magnum_opus::Channels,
+    }
+
+    impl CaptureFrameProcessor {
+        fn new(
+            config: CaptureFrameProcessorConfig,
+            sender: audio_capture_queue::CapturePcmSender,
+        ) -> ResultType<Self> {
+            let resampler = if config.input_rate == config.output_rate {
+                None
+            } else {
+                let output_frames = config.output_rate as usize / AUDIO_PACKETS_PER_SECOND;
+                Some(crate::audio_resampler::FixedFrameAudioResampler::new(
+                    crate::audio_resampler::AudioResamplerConfig {
+                        input_rate: config.input_rate,
+                        output_rate: config.output_rate,
+                        channels: config.device_channel,
+                    },
+                    output_frames,
+                )?)
+            };
+            Ok(Self {
+                config,
+                resampler,
+                sender,
+                rechannel_buffer: Vec::with_capacity(
+                    capture_packet_layout(config.output_rate, config.encode_channel)?.1,
+                ),
+            })
         }
-        if device_channel != encode_channel {
-            data = crate::common::audio_rechannel(
-                data,
-                sample_rate,
-                sample_rate,
-                device_channel,
-                encode_channel,
+
+        fn process(&mut self, data: &[f32]) -> ResultType<()> {
+            let config = self.config;
+            let sender = &mut self.sender;
+            let rechannel_buffer = &mut self.rechannel_buffer;
+            let mut send_packet = |packet: &[f32]| {
+                let packet =
+                    audio_capture::rechannel(packet, config.device_channel, rechannel_buffer);
+                sender.submit(packet);
+            };
+            if let Some(resampler) = self.resampler.as_mut() {
+                resampler.process_with(data, send_packet).with_context(|| {
+                    format!(
+                        "Failed to resample captured audio from {} Hz to {} Hz",
+                        config.input_rate, config.output_rate
+                    )
+                })?;
+            } else {
+                send_packet(data);
+            }
+            Ok(())
+        }
+    }
+
+    fn capture_packet_layout(sample_rate: u32, channels: u16) -> ResultType<(usize, usize)> {
+        if sample_rate < AUDIO_PACKETS_PER_SECOND as u32 || channels == 0 {
+            bail!("Invalid audio capture layout: sample_rate={sample_rate}, channels={channels}");
+        }
+        let frames = sample_rate as usize / AUDIO_PACKETS_PER_SECOND;
+        let samples = frames.checked_mul(channels as usize).with_context(|| {
+            format!(
+                "Audio capture frame size overflow: sample_rate={sample_rate}, channels={channels}"
             )
-        }
-        send_f32(&data, encoder, sp);
+        })?;
+        Ok((frames, samples))
     }
 
     #[cfg(feature = "screencapturekit")]
@@ -367,9 +447,7 @@ mod cpal_impl {
         Ok((device, format))
     }
 
-    fn play(
-        sp: &GenericService,
-    ) -> ResultType<(Box<dyn StreamTrait>, Arc<Message>, CaptureErrorHandler)> {
+    fn play(sp: &GenericService) -> ResultType<ActiveCaptureStream> {
         use cpal::SampleFormat::*;
         let (device, config) = get_device()?;
         let sp = sp.clone();
@@ -387,109 +465,274 @@ mod cpal_impl {
             48000
         };
         let ch = if config.channels() > 1 { Stereo } else { Mono };
+        let max_channels = config.channels().max(ch as u16);
+        let (_, max_packet_samples) = capture_packet_layout(sample_rate, max_channels)?;
+        let encoder_config = audio_capture_queue::CaptureEncoderConfig {
+            sample_rate,
+            encode_channel: ch,
+            max_packet_samples,
+        };
+        let (sender, encoder_worker) =
+            audio_capture_queue::start_capture_encoder(encoder_config, sp)?;
+        let output = CaptureStreamOutput {
+            sender,
+            sample_rate,
+            encode_channel: ch,
+        };
         let (stream, errors) = match config.sample_format() {
-            I8 => build_input_stream::<i8>(device, &config, sp, sample_rate, ch)?,
-            I16 => build_input_stream::<i16>(device, &config, sp, sample_rate, ch)?,
-            I32 => build_input_stream::<i32>(device, &config, sp, sample_rate, ch)?,
-            I64 => build_input_stream::<i64>(device, &config, sp, sample_rate, ch)?,
-            U8 => build_input_stream::<u8>(device, &config, sp, sample_rate, ch)?,
-            U16 => build_input_stream::<u16>(device, &config, sp, sample_rate, ch)?,
-            U32 => build_input_stream::<u32>(device, &config, sp, sample_rate, ch)?,
-            U64 => build_input_stream::<u64>(device, &config, sp, sample_rate, ch)?,
-            F32 => build_input_stream::<f32>(device, &config, sp, sample_rate, ch)?,
-            F64 => build_input_stream::<f64>(device, &config, sp, sample_rate, ch)?,
+            I8 => build_input_stream::<i8>(device, &config, output)?,
+            I16 => build_input_stream::<i16>(device, &config, output)?,
+            I32 => build_input_stream::<i32>(device, &config, output)?,
+            I64 => build_input_stream::<i64>(device, &config, output)?,
+            U8 => build_input_stream::<u8>(device, &config, output)?,
+            U16 => build_input_stream::<u16>(device, &config, output)?,
+            U32 => build_input_stream::<u32>(device, &config, output)?,
+            U64 => build_input_stream::<u64>(device, &config, output)?,
+            F32 => build_input_stream::<f32>(device, &config, output)?,
+            F64 => build_input_stream::<f64>(device, &config, output)?,
             f => bail!("unsupported audio format: {:?}", f),
         };
         stream.play()?;
         #[cfg(target_os = "macos")]
         log::info!("Audio capture start call succeeded");
-        Ok((
-            Box::new(stream),
-            Arc::new(create_format_msg(sample_rate, ch as _)),
+        Ok(ActiveCaptureStream {
+            stream: Some(Box::new(stream)),
+            format: Arc::new(create_format_msg(sample_rate, ch as _)),
+            _encoder_worker: encoder_worker,
             errors,
-        ))
+        })
+    }
+
+    fn convert_input_samples<T>(data: &[T]) -> impl Iterator<Item = f32> + '_
+    where
+        T: cpal::SizedSample,
+        f32: cpal::FromSample<T>,
+    {
+        data.iter()
+            .map(|sample| <f32 as cpal::FromSample<T>>::from_sample_(*sample))
+    }
+
+    #[cfg(target_os = "macos")]
+    fn log_capture_startup<T>(
+        data: &[T],
+        received_samples: bool,
+        received_signal: bool,
+    ) -> (bool, bool)
+    where
+        T: cpal::SizedSample,
+        f32: cpal::FromSample<T>,
+    {
+        // Starting capture does not guarantee sample delivery or audible data.
+        if !received_samples && !data.is_empty() {
+            log::info!(
+                "Audio capture received first PCM block: {} samples",
+                data.len()
+            );
+        }
+        let has_signal = received_signal
+            || convert_input_samples(data).any(|sample| sample.is_finite() && sample != 0.0);
+        if !received_signal && has_signal {
+            log::info!("Audio capture received first nonzero PCM");
+        }
+        (received_samples || !data.is_empty(), has_signal)
     }
 
     fn build_input_stream<T>(
         device: cpal::Device,
         config: &cpal::SupportedStreamConfig,
-        sp: GenericService,
-        sample_rate: u32,
-        encode_channel: magnum_opus::Channels,
+        output: CaptureStreamOutput,
     ) -> ResultType<(cpal::Stream, CaptureErrorHandler)>
     where
-        T: cpal::SizedSample + dasp::sample::ToSample<f32>,
+        T: cpal::SizedSample,
+        f32: cpal::FromSample<T>,
     {
         let errors = CaptureErrorHandler::default();
         let callback_errors = errors.clone();
         let err_fn = move |err| callback_errors.handle(err);
+        let processor_errors = errors.clone();
         #[cfg(target_os = "macos")]
         let (mut received_samples, mut received_signal) = (false, false);
         let sample_rate_0 = config.sample_rate().0;
-        log::debug!("Audio sample rate : {}", sample_rate);
-        unsafe {
-            AUDIO_ZERO_COUNT = 0;
-        }
+        log::debug!("Audio sample rate : {}", output.sample_rate);
         let device_channel = config.channels();
-        let mut encoder = Encoder::new(sample_rate, encode_channel, LowDelay)?;
-        // https://www.opus-codec.org/docs/html_api/group__opusencoder.html#gace941e4ef26ed844879fde342ffbe546
-        // https://chromium.googlesource.com/chromium/deps/opus/+/1.1.1/include/opus.h
-        // Do not set `frame_size = sample_rate as usize / 100;`
-        // Because we find `sample_rate as usize / 100` will cause encoder error in `encoder.encode_vec_float()` sometimes.
-        // https://github.com/xiph/opus/blob/2554a89e02c7fc30a980b4f7e635ceae1ecba5d6/src/opus_encoder.c#L725
-        let frame_size = sample_rate_0 as usize / 100; // 10 ms
-        let encode_len = frame_size * encode_channel as usize;
-        let rechannel_len = encode_len * device_channel as usize / encode_channel as usize;
-        INPUT_BUFFER.lock().unwrap().clear();
-        let timeout = None;
-        let stream_config = StreamConfig {
-            channels: device_channel,
-            sample_rate: config.sample_rate(),
-            buffer_size: BufferSize::Default,
+        let (_, capture_frame_samples) = capture_packet_layout(sample_rate_0, device_channel)?;
+        let mut frame = audio_capture::CaptureFrameBuffer::new(capture_frame_samples)?;
+        let processor_config = CaptureFrameProcessorConfig {
+            input_rate: sample_rate_0,
+            output_rate: output.sample_rate,
+            device_channel,
+            encode_channel: output.encode_channel as _,
         };
+        let mut processor = CaptureFrameProcessor::new(processor_config, output.sender)?;
+        let timeout = None;
         let stream = device.build_input_stream(
-            &stream_config,
+            &config.config(),
             move |data: &[T], _: &InputCallbackInfo| {
-                let buffer: Vec<f32> = data.iter().map(|s| T::to_sample(*s)).collect();
+                if processor_errors.needs_restart() {
+                    return;
+                }
                 #[cfg(target_os = "macos")]
                 {
-                    // Starting capture does not guarantee sample delivery or audible data.
-                    if !received_samples && !buffer.is_empty() {
-                        received_samples = true;
-                        log::info!(
-                            "Audio capture received first PCM block: {} samples",
-                            buffer.len()
-                        );
-                    }
-                    if !received_signal
-                        && buffer
-                            .iter()
-                            .any(|sample| sample.is_finite() && *sample != 0.0)
-                    {
-                        received_signal = true;
-                        log::info!("Audio capture received first nonzero PCM");
-                    }
-                }
-                let mut lock = INPUT_BUFFER.lock().unwrap();
-                lock.extend(buffer);
-                while lock.len() >= rechannel_len {
-                    let frame: Vec<f32> = lock.drain(0..rechannel_len).collect();
-                    send(
-                        frame,
-                        sample_rate_0,
-                        sample_rate,
-                        device_channel,
-                        encode_channel as _,
-                        &mut encoder,
-                        &sp,
-                    );
+                    (received_samples, received_signal) =
+                        log_capture_startup(data, received_samples, received_signal);
                 }
+                frame.process(convert_input_samples(data), |frame| {
+                    processor_errors.process_frame(|| processor.process(frame));
+                });
             },
             err_fn,
             timeout,
         )?;
         Ok((stream, errors))
     }
+
+    #[cfg(test)]
+    mod tests {
+        use super::super::audio_capture_queue::{
+            new_pcm_handoff, start_capture_encoder, CaptureEncoderConfig,
+        };
+        use super::{
+            capture_packet_layout, convert_input_samples, CaptureFrameProcessor,
+            CaptureFrameProcessorConfig,
+        };
+        use crate::audio_resampler::allocation_tests::assert_no_allocations;
+        use crate::server::EmptyExtraFieldService;
+        use magnum_opus::Channels::{Mono, Stereo};
+
+        const INVALID_CAPTURE_RATE: u32 = 99;
+        const RATE_24_KHZ: u32 = 24_000;
+        const RATE_44_1_KHZ: u32 = 44_100;
+        const RATE_48_KHZ: u32 = 48_000;
+        const MONO_CHANNELS: u16 = 1;
+        const NEGATIVE_FULL_SCALE_LIMIT: f32 = -0.99;
+        const POSITIVE_FULL_SCALE_LIMIT: f32 = 0.99;
+        const STEREO_CHANNELS: u16 = 2;
+        const SURROUND_CHANNELS: u16 = 6;
+        const ZERO_CHANNELS: u16 = 0;
+
+        #[test]
+        fn capture_sample_conversion_uses_cpal_traits() {
+            let input = [i16::MIN, 0, i16::MAX];
+            let output: Vec<_> = convert_input_samples(&input).collect();
+
+            assert_eq!(output.len(), input.len());
+            assert!(output[0] <= NEGATIVE_FULL_SCALE_LIMIT);
+            assert_eq!(output[1], 0.0);
+            assert!(output[2] >= POSITIVE_FULL_SCALE_LIMIT);
+        }
+
+        #[test]
+        fn capture_packet_layout_validates_rate_and_channels() {
+            let expected_frames = RATE_48_KHZ as usize / super::AUDIO_PACKETS_PER_SECOND;
+            assert_eq!(
+                capture_packet_layout(RATE_48_KHZ, STEREO_CHANNELS).unwrap(),
+                (expected_frames, expected_frames * STEREO_CHANNELS as usize)
+            );
+            assert!(capture_packet_layout(INVALID_CAPTURE_RATE, MONO_CHANNELS).is_err());
+            assert!(capture_packet_layout(RATE_48_KHZ, ZERO_CHANNELS).is_err());
+        }
+
+        #[test]
+        fn capture_callback_pipeline_does_not_allocate_after_warmup() {
+            for (input_rate, output_rate, device_channel, encode_channel) in [
+                (RATE_48_KHZ, RATE_48_KHZ, MONO_CHANNELS, MONO_CHANNELS),
+                (RATE_48_KHZ, RATE_48_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
+                (RATE_44_1_KHZ, RATE_24_KHZ, STEREO_CHANNELS, STEREO_CHANNELS),
+                (RATE_48_KHZ, RATE_48_KHZ, SURROUND_CHANNELS, STEREO_CHANNELS),
+            ] {
+                assert_capture_processor_does_not_allocate(CaptureFrameProcessorConfig {
+                    input_rate,
+                    output_rate,
+                    device_channel,
+                    encode_channel,
+                });
+            }
+        }
+
+        #[test]
+        fn capture_pcm_handoff_reuses_buffers_and_accounts_for_loss() {
+            const QUEUE_CAPACITY: usize = 2;
+            const PACKET_SAMPLES: usize = 4;
+            const FIRST: [f32; PACKET_SAMPLES] = [1.0; PACKET_SAMPLES];
+            const SECOND: [f32; PACKET_SAMPLES] = [2.0; PACKET_SAMPLES];
+            const THIRD: [f32; PACKET_SAMPLES] = [3.0; PACKET_SAMPLES];
+            const OVERSIZED_SAMPLES: usize = PACKET_SAMPLES + 1;
+            const OVERSIZED: [f32; OVERSIZED_SAMPLES] = [1.0; OVERSIZED_SAMPLES];
+
+            let (mut sender, receiver) = new_pcm_handoff(QUEUE_CAPACITY, PACKET_SAMPLES).unwrap();
+            sender.set_wake_thread(std::thread::current()).unwrap();
+            assert_no_allocations(|| {
+                sender.submit(&FIRST);
+                sender.submit(&SECOND);
+                sender.submit(&THIRD);
+            });
+
+            let loss = receiver.take_loss();
+            assert_eq!(loss.dropped, 1);
+            assert_eq!(loss.oversized, 0);
+            assert_eq!(loss.recycle_failures, 0);
+            let second = receiver.pop().unwrap();
+            let third = receiver.pop().unwrap();
+            assert_eq!(second, SECOND);
+            assert_eq!(third, THIRD);
+            receiver.recycle(second);
+            receiver.recycle(third);
+            assert!(receiver.is_empty());
+
+            assert_no_allocations(|| sender.submit(&OVERSIZED));
+            let loss = receiver.take_loss();
+            assert_eq!(loss.dropped, 0);
+            assert_eq!(loss.oversized, 1);
+            assert_eq!(loss.recycle_failures, 0);
+            assert!(receiver.is_empty());
+        }
+
+        #[test]
+        fn capture_pcm_handoff_rejects_invalid_layouts() {
+            assert!(new_pcm_handoff(0, 1).is_err());
+            assert!(new_pcm_handoff(1, 0).is_err());
+        }
+
+        fn assert_capture_processor_does_not_allocate(config: CaptureFrameProcessorConfig) {
+            const INPUT_LEVEL: f32 = 0.25;
+            const TEST_SERVICE_NAME: &str = "audio-allocation-test";
+
+            let service = EmptyExtraFieldService::new(TEST_SERVICE_NAME.to_owned(), true).sp;
+            let encode_channel = if config.encode_channel == MONO_CHANNELS {
+                Mono
+            } else {
+                Stereo
+            };
+            let encoder_config = CaptureEncoderConfig {
+                sample_rate: config.output_rate,
+                encode_channel,
+                max_packet_samples: config.output_rate as usize / super::AUDIO_PACKETS_PER_SECOND
+                    * config.device_channel.max(config.encode_channel) as usize,
+            };
+            let (sender, worker) = start_capture_encoder(encoder_config, service).unwrap();
+            let mut processor = CaptureFrameProcessor::new(config, sender).unwrap();
+            let errors = super::CaptureErrorHandler::default();
+            let input = vec![
+                INPUT_LEVEL;
+                config.input_rate as usize / super::AUDIO_PACKETS_PER_SECOND
+                    * config.device_channel as usize
+            ];
+            let mut frame_buffer =
+                super::audio_capture::CaptureFrameBuffer::new(input.len()).unwrap();
+
+            frame_buffer.process(convert_input_samples(&input), |frame| {
+                errors.process_frame(|| processor.process(frame));
+            });
+            assert_no_allocations(|| {
+                frame_buffer.process(convert_input_samples(&input), |frame| {
+                    errors.process_frame(|| processor.process(frame));
+                });
+            });
+            assert!(!errors.needs_restart());
+            drop(processor);
+            drop(worker);
+        }
+    }
 }
 
 fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
@@ -505,28 +748,43 @@ fn create_format_msg(sample_rate: u32, channels: u16) -> Message {
     msg
 }
 
-// use AUDIO_ZERO_COUNT for the Noise(Zero) Gate Attack Time
+// Use a per-encoder counter for the Noise(Zero) Gate Attack Time.
 // every audio data length is set to 480
 // MAX_AUDIO_ZERO_COUNT=800 is similar as Gate Attack Time 3~5s(Linux) || 6~8s(Windows)
 const MAX_AUDIO_ZERO_COUNT: u16 = 800;
-static mut AUDIO_ZERO_COUNT: u16 = 0;
 
-fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
-    if data.iter().filter(|x| **x != 0.).next().is_some() {
-        unsafe {
-            AUDIO_ZERO_COUNT = 0;
+struct AudioEncoder {
+    encoder: Encoder,
+    zero_count: u16,
+}
+
+impl AudioEncoder {
+    fn new(encoder: Encoder) -> Self {
+        Self {
+            encoder,
+            zero_count: 0,
         }
-    } else {
-        unsafe {
-            if AUDIO_ZERO_COUNT > MAX_AUDIO_ZERO_COUNT {
-                if AUDIO_ZERO_COUNT == MAX_AUDIO_ZERO_COUNT + 1 {
-                    log::debug!("Audio Zero Gate Attack");
-                    AUDIO_ZERO_COUNT += 1;
-                }
-                return;
+    }
+
+    fn should_encode(&mut self, data: &[f32]) -> bool {
+        if data.iter().filter(|x| **x != 0.).next().is_some() {
+            self.zero_count = 0;
+        } else if self.zero_count > MAX_AUDIO_ZERO_COUNT {
+            if self.zero_count == MAX_AUDIO_ZERO_COUNT + 1 {
+                log::debug!("Audio Zero Gate Attack");
+                self.zero_count += 1;
             }
-            AUDIO_ZERO_COUNT += 1;
+            return false;
+        } else {
+            self.zero_count += 1;
         }
+        true
+    }
+}
+
+fn send_f32(data: &[f32], encoder: &mut AudioEncoder, sp: &GenericService) {
+    if !encoder.should_encode(data) {
+        return;
     }
     #[cfg(target_os = "android")]
     {
@@ -539,6 +797,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
             let n = input_size / BATCH_SIZE;
             for i in 0..n {
                 match encoder
+                    .encoder
                     .encode_vec_float(&data[i * BATCH_SIZE..(i + 1) * BATCH_SIZE], BATCH_SIZE)
                 {
                     Ok(data) => {
@@ -549,7 +808,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
                         });
                         sp.send(msg_out);
                     }
-                    Err(_) => {}
+                    Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
                 }
             }
         } else {
@@ -559,7 +818,7 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
     }
 
     #[cfg(not(target_os = "android"))]
-    match encoder.encode_vec_float(data, data.len() * 6) {
+    match encoder.encoder.encode_vec_float(data, data.len() * 6) {
         Ok(data) => {
             let mut msg_out = Message::new();
             msg_out.set_audio_frame(AudioFrame {
@@ -568,6 +827,6 @@ fn send_f32(data: &[f32], encoder: &mut Encoder, sp: &GenericService) {
             });
             sp.send(msg_out);
         }
-        Err(_) => {}
+        Err(error) => log::warn!("Failed to encode audio frame: {error:?}"),
     }
 }

src/server/audio_service/audio_capture.rs

@@ -0,0 +1,144 @@
+use hbb_common::anyhow::{bail, Result};
+
+const STEREO_CHANNELS: usize = 2;
+
+pub(super) struct CaptureFrameBuffer {
+    samples: Vec<f32>,
+    filled: usize,
+}
+
+impl CaptureFrameBuffer {
+    pub(super) fn new(samples: usize) -> Result<Self> {
+        if samples == 0 {
+            bail!("Audio capture frame must contain at least one sample");
+        }
+        Ok(Self {
+            samples: vec![0.0; samples],
+            filled: 0,
+        })
+    }
+
+    pub(super) fn process(
+        &mut self,
+        input: impl Iterator<Item = f32>,
+        mut on_frame: impl FnMut(&[f32]),
+    ) {
+        for sample in input {
+            self.samples[self.filled] = sample;
+            self.filled += 1;
+            if self.filled == self.samples.len() {
+                self.filled = 0;
+                on_frame(&self.samples);
+            }
+        }
+    }
+}
+
+pub(super) fn rechannel<'a>(
+    input: &'a [f32],
+    channels: u16,
+    output: &'a mut Vec<f32>,
+) -> &'a [f32] {
+    let input = if channels > STEREO_CHANNELS as u16 {
+        &input[..input.len() / channels as usize * channels as usize]
+    } else {
+        input
+    };
+    output.clear();
+    match channels {
+        3 => rechannel_frame::<3>(input, output),
+        4 => rechannel_frame::<4>(input, output),
+        5 => rechannel_frame::<5>(input, output),
+        6 => rechannel_frame::<6>(input, output),
+        7 => rechannel_frame::<7>(input, output),
+        8 => rechannel_frame::<8>(input, output),
+        // Preserve the existing passthrough for mono/stereo and unsupported layouts.
+        _ => return input,
+    }
+    output
+}
+
+fn rechannel_frame<const CHANNELS: usize>(input: &[f32], output: &mut Vec<f32>) {
+    use fon::{
+        chan::{Ch32, Channel},
+        Frame,
+    };
+
+    for samples in input.chunks_exact(CHANNELS) {
+        let mut frame = Frame::<Ch32, CHANNELS>::default();
+        for (channel, sample) in frame.channels_mut().iter_mut().zip(samples) {
+            *channel = (*sample).into();
+        }
+        // Match the same-rate Stream::pipe conversion before fon's SinkTo conversion.
+        let stereo = frame.to::<Ch32, CHANNELS>().to::<Ch32, STEREO_CHANNELS>();
+        output.extend(stereo.channels().iter().map(|channel| channel.to_f32()));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{rechannel, CaptureFrameBuffer};
+    use crate::audio_resampler::allocation_tests::assert_no_allocations;
+
+    #[test]
+    fn capture_channel_conversion_reuses_storage_and_preserves_mapping() {
+        const SAMPLE_RATE: u32 = 48_000;
+        const FRAMES: usize = 7;
+        const STEREO_CHANNELS: u16 = 2;
+        const MIN_SAMPLE: f32 = -1.25;
+        const SAMPLE_STEP: f32 = 0.2;
+
+        for channels in [1, 2, 3, 4, 5, 6, 7, 8, 16] {
+            let input: Vec<_> = (0..FRAMES * channels as usize + 1)
+                .map(|sample| MIN_SAMPLE + sample as f32 * SAMPLE_STEP)
+                .collect();
+            let expected = crate::common::audio_rechannel(
+                input.clone(),
+                SAMPLE_RATE,
+                SAMPLE_RATE,
+                channels,
+                channels.min(STEREO_CHANNELS),
+            );
+            let mut output = Vec::with_capacity(FRAMES * STEREO_CHANNELS as usize);
+            assert_no_allocations(|| {
+                let actual = rechannel(&input, channels, &mut output);
+                assert_eq!(
+                    actual, expected,
+                    "channel mapping changed for {channels} channels"
+                );
+            });
+        }
+    }
+
+    #[test]
+    fn capture_framing_retains_partial_input_without_allocating() {
+        const FRAME_SAMPLES: usize = 6;
+        const CALLBACK_SIZES: [usize; 7] = [0, 1, 17, 2, 257, 492, 5];
+        let input: Vec<_> = (0..CALLBACK_SIZES.iter().sum::<usize>())
+            .map(|sample| sample as f32)
+            .collect();
+        let mut buffer = CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
+        let mut input_position = 0;
+        let mut output_position = 0;
+
+        assert!(CaptureFrameBuffer::new(0).is_err());
+        assert_no_allocations(|| {
+            for samples in CALLBACK_SIZES {
+                let end = input_position + samples;
+                buffer.process(input[input_position..end].iter().copied(), |frame| {
+                    assert_eq!(
+                        frame,
+                        &input[output_position..output_position + FRAME_SAMPLES]
+                    );
+                    output_position += FRAME_SAMPLES;
+                });
+                input_position = end;
+                assert_eq!(
+                    output_position,
+                    input_position / FRAME_SAMPLES * FRAME_SAMPLES
+                );
+            }
+        });
+        assert_eq!(output_position, input.len());
+    }
+}

src/server/audio_service/audio_capture_encoder.rs

@@ -0,0 +1,128 @@
+use super::super::AudioEncoder;
+use super::{
+    send_f32, CaptureEncoderConfig, CaptureEncoderContext, CapturePcmReceiver, CapturePcmStats,
+    CAPTURE_PCM_QUEUE_PACKETS,
+};
+use hbb_common::log;
+use magnum_opus::Channels;
+use std::{
+    sync::atomic::Ordering,
+    time::{Duration, Instant},
+};
+
+const CAPTURE_DECLICK_MS: usize = 5;
+const CAPTURE_PACKET_MS: usize = 10;
+const MILLISECONDS_PER_SECOND: usize = 1_000;
+const MAX_ENCODE_CHANNELS: usize = Channels::Stereo as usize;
+const CAPTURE_STATS_LOG_INTERVAL: Duration = Duration::from_secs(5);
+
+struct CaptureEncoderState {
+    channels: usize,
+    expected_sequence: usize,
+    fade_frames: usize,
+    last_frame: [f32; MAX_ENCODE_CHANNELS],
+    reporter: CaptureStatsReporter,
+}
+
+impl CaptureEncoderState {
+    fn new(sample_rate: u32, channels: Channels) -> Self {
+        // The encoder has already validated the supported Opus rate and channel count.
+        let fade_frames = sample_rate as usize * CAPTURE_DECLICK_MS / MILLISECONDS_PER_SECOND;
+        Self {
+            channels: channels as usize,
+            expected_sequence: 0,
+            fade_frames,
+            last_frame: [0.0; MAX_ENCODE_CHANNELS],
+            reporter: CaptureStatsReporter::new(),
+        }
+    }
+
+    fn next_packet(&mut self, receiver: &CapturePcmReceiver) -> Option<Vec<f32>> {
+        self.reporter.pending.add(receiver.take_stats());
+        self.reporter.report(false);
+        let (sequence, mut packet) = receiver.pop_packet()?;
+        self.smooth_packet(sequence, &mut packet);
+        Some(packet)
+    }
+
+    fn smooth_packet(&mut self, sequence: usize, packet: &mut [f32]) {
+        if sequence != self.expected_sequence {
+            // Capture packets contain 10 ms of PCM, so the transition fits in this packet.
+            for (index, frame) in packet
+                .chunks_exact_mut(self.channels)
+                .take(self.fade_frames)
+                .enumerate()
+            {
+                let weight = (index + 1) as f32 / self.fade_frames as f32;
+                for (channel, sample) in frame.iter_mut().enumerate() {
+                    *sample = self.last_frame[channel] * (1.0 - weight) + *sample * weight;
+                }
+            }
+        }
+        self.expected_sequence = sequence.wrapping_add(1);
+        if let Some(frame) = packet.chunks_exact(self.channels).next_back() {
+            self.last_frame[..self.channels].copy_from_slice(frame);
+        }
+    }
+}
+
+struct CaptureStatsReporter {
+    pending: CapturePcmStats,
+    last_report: Instant,
+}
+
+impl CaptureStatsReporter {
+    fn new() -> Self {
+        Self {
+            pending: Default::default(),
+            last_report: Instant::now(),
+        }
+    }
+
+    fn report(&mut self, force: bool) {
+        if self.pending.is_empty() {
+            return;
+        }
+        if !force && self.last_report.elapsed() < CAPTURE_STATS_LOG_INTERVAL {
+            return;
+        }
+        let stats = std::mem::take(&mut self.pending);
+        log::debug!(
+            "Audio capture PCM handoff stats: observed_max_queued_packets={}, approx_queued_audio_ms={}, capacity_packets={}",
+            stats.max_queued_packets,
+            stats.max_queued_packets.saturating_mul(CAPTURE_PACKET_MS),
+            CAPTURE_PCM_QUEUE_PACKETS
+        );
+        if !stats.loss.is_empty() {
+            log::warn!(
+                "Audio capture PCM handoff loss: dropped={}, contention_dropped={}, oversized={}, recycle_failures={}",
+                stats.loss.dropped,
+                stats.loss.contention_dropped,
+                stats.loss.oversized,
+                stats.loss.recycle_failures
+            );
+        }
+        self.last_report = Instant::now();
+    }
+}
+
+pub(super) fn run_capture_encoder(context: CaptureEncoderContext, config: CaptureEncoderConfig) {
+    let mut encoder = AudioEncoder::new(context.encoder);
+    let mut state = CaptureEncoderState::new(config.sample_rate, config.encode_channel);
+    loop {
+        while let Some(packet) = state.next_packet(&context.receiver) {
+            send_f32(&packet, &mut encoder, &context.service);
+            context.receiver.recycle(packet);
+        }
+        if context.stop.load(Ordering::Acquire) && context.receiver.is_empty() {
+            state.reporter.pending.add(context.receiver.take_stats());
+            state.reporter.report(true);
+            return;
+        }
+        std::thread::park_timeout(CAPTURE_STATS_LOG_INTERVAL);
+    }
+}
+
+#[cfg(test)]
+#[path = "audio_capture_encoder_tests.rs"]
+mod tests;

src/server/audio_service/audio_capture_encoder_tests.rs

@@ -0,0 +1,186 @@
+use super::super::{new_pcm_handoff, CAPTURE_PCM_QUEUE_PACKETS};
+use super::*;
+use crate::audio_resampler::allocation_tests::assert_no_allocations;
+use magnum_opus::{Application::LowDelay, Decoder, Encoder};
+
+const SAMPLE_RATE: u32 = 48_000;
+const PACKETS_PER_SECOND: usize = 100;
+const PACKET_COUNT: usize = 120;
+const PAUSE_PACKET: usize = 50;
+const DROPPED_PACKETS: usize = 6;
+const SIGNAL_FREQUENCY: f64 = 97.0;
+const SIGNAL_AMPLITUDE: f64 = 0.5;
+const ACTIVE_LEVEL: f32 = 0.8;
+const SAMPLE_TOLERANCE: f32 = 0.000001;
+const MAX_ENCODE_BYTES_PER_SAMPLE: usize = 6;
+
+fn signal_packet(index: usize, channels: usize) -> Vec<f32> {
+    let frames = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
+    (0..frames)
+        .flat_map(|frame| {
+            let phase = std::f64::consts::TAU * SIGNAL_FREQUENCY * (index * frames + frame) as f64
+                / f64::from(SAMPLE_RATE);
+            (0..channels)
+                .map(move |channel| (SIGNAL_AMPLITUDE * (phase + channel as f64).sin()) as f32)
+        })
+        .collect()
+}
+
+fn encoded_audio(dropped: usize, channels: Channels) -> Vec<f32> {
+    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels as usize;
+    let (mut sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, samples).unwrap();
+    let mut state = CaptureEncoderState::new(SAMPLE_RATE, channels);
+    let mut encoder = Encoder::new(SAMPLE_RATE, channels, LowDelay).unwrap();
+    let mut decoder = Decoder::new(SAMPLE_RATE, channels).unwrap();
+    let mut decoded = vec![0.0; samples];
+    let mut output = Vec::new();
+    let resume_packet = PAUSE_PACKET + CAPTURE_PCM_QUEUE_PACKETS + dropped - 1;
+    for index in 0..PACKET_COUNT {
+        sender.submit(&signal_packet(index, channels as usize));
+        if (PAUSE_PACKET..resume_packet).contains(&index) {
+            continue;
+        }
+        if index == resume_packet {
+            let stats = receiver.take_stats();
+            assert_eq!(stats.max_queued_packets, CAPTURE_PCM_QUEUE_PACKETS);
+            assert_eq!(stats.loss.dropped, dropped);
+            assert_eq!(stats.loss.contention_dropped, 0);
+            assert_eq!(stats.loss.oversized, 0);
+            assert_eq!(stats.loss.recycle_failures, 0);
+        }
+        while let Some(packet) = state.next_packet(&receiver) {
+            let encoded = encoder
+                .encode_vec_float(&packet, samples * MAX_ENCODE_BYTES_PER_SAMPLE)
+                .unwrap();
+            let frames = decoder.decode_float(&encoded, &mut decoded, false).unwrap();
+            assert_eq!(frames * channels as usize, samples);
+            output.extend_from_slice(&decoded);
+            receiver.recycle(packet);
+        }
+    }
+    assert_eq!(output.len(), (PACKET_COUNT - dropped) * samples);
+    output
+}
+
+fn maximum_join_step(pcm: &[f32], channels: usize) -> f32 {
+    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND * channels;
+    pcm[(PAUSE_PACKET - 1) * samples..(PAUSE_PACKET + 2) * samples]
+        .windows(channels + 1)
+        .map(|window| (window[channels] - window[0]).abs())
+        .fold(0.0, f32::max)
+}
+
+#[test]
+fn capture_overflow_is_smoothed_before_encoding() {
+    const MAX_STEP_RATIO: f32 = 2.0;
+    for channels in [Channels::Mono, Channels::Stereo] {
+        let clean = encoded_audio(0, channels);
+        let overflow = encoded_audio(DROPPED_PACKETS, channels);
+        let clean_step = maximum_join_step(&clean, channels as usize);
+        let overflow_step = maximum_join_step(&overflow, channels as usize);
+        assert!(clean_step > 0.0);
+        assert!(
+            overflow_step < clean_step * MAX_STEP_RATIO,
+            "capture discard introduced a sharp join: clean={clean_step}, overflow={overflow_step}"
+        );
+    }
+}
+
+#[test]
+fn rejected_packets_mark_the_gap_after_already_queued_audio() {
+    const CAPACITY: usize = 3;
+    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
+    let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
+    let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
+    let active = vec![ACTIVE_LEVEL; samples];
+    let opposite = vec![-ACTIVE_LEVEL; samples];
+    sender.submit(&active);
+    let packet = state.next_packet(&receiver).unwrap();
+    assert_eq!(packet, active);
+    receiver.recycle(packet);
+    sender.submit(&active);
+    sender.submit(&active);
+    sender.submit(&vec![ACTIVE_LEVEL; samples + 1]);
+    sender.submit(&opposite);
+    assert_eq!(receiver.take_loss().oversized, 1);
+    assert_no_allocations(|| {
+        for _ in 0..CAPACITY - 1 {
+            let packet = state.next_packet(&receiver).unwrap();
+            assert_eq!(packet, active);
+            receiver.recycle(packet);
+        }
+        let packet = state.next_packet(&receiver).unwrap();
+        let expected = ACTIVE_LEVEL * (1.0 - 2.0 / state.fade_frames as f32);
+        assert!((packet[0] - expected).abs() < SAMPLE_TOLERANCE);
+        assert_eq!(&packet[samples / 2..], &opposite[samples / 2..]);
+        receiver.recycle(packet);
+    });
+}
+
+#[test]
+fn gaps_and_sequence_wrap_preserve_channel_history() {
+    const RATE_8_KHZ: u32 = 8_000;
+    for (rate, channels) in [
+        (RATE_8_KHZ, Channels::Mono),
+        (RATE_8_KHZ, Channels::Stereo),
+        (SAMPLE_RATE, Channels::Mono),
+        (SAMPLE_RATE, Channels::Stereo),
+    ] {
+        let channels_count = channels as usize;
+        let mut state = CaptureEncoderState::new(rate, channels);
+        let samples = rate as usize / PACKETS_PER_SECOND * channels_count;
+        let active: Vec<_> = [ACTIVE_LEVEL, -ACTIVE_LEVEL][..channels_count]
+            .iter()
+            .copied()
+            .cycle()
+            .take(samples)
+            .collect();
+        let mut first = active.clone();
+        state.smooth_packet(0, &mut first);
+        assert_eq!(first, active);
+        let mut opposite: Vec<_> = active.iter().map(|sample| -sample).collect();
+        let mut resumed = active.clone();
+        assert_no_allocations(|| {
+            state.smooth_packet(2, &mut opposite);
+            state.smooth_packet(4, &mut resumed);
+        });
+        for channel in 0..channels_count {
+            let expected = active[channel] * (1.0 - 2.0 / state.fade_frames as f32);
+            assert!((opposite[channel] - expected).abs() < SAMPLE_TOLERANCE);
+            let previous = opposite[samples - channels_count + channel];
+            let expected = previous + (active[channel] - previous) / state.fade_frames as f32;
+            assert!((resumed[channel] - expected).abs() < SAMPLE_TOLERANCE);
+        }
+        assert_eq!(&resumed[samples / 2..], &active[samples / 2..]);
+        state.expected_sequence = usize::MAX;
+        let mut last = active.clone();
+        let mut wrapped: Vec<_> = active.iter().map(|sample| -sample).collect();
+        let expected = wrapped.clone();
+        assert_no_allocations(|| {
+            state.smooth_packet(usize::MAX, &mut last);
+            state.smooth_packet(0, &mut wrapped);
+        });
+        assert_eq!(last, active);
+        assert_eq!(wrapped, expected);
+    }
+}
+
+#[test]
+fn capture_loss_is_reported_while_packets_remain_queued() {
+    const CAPACITY: usize = 2;
+    let samples = SAMPLE_RATE as usize / PACKETS_PER_SECOND;
+    let (mut sender, receiver) = new_pcm_handoff(CAPACITY, samples).unwrap();
+    let mut state = CaptureEncoderState::new(SAMPLE_RATE, Channels::Mono);
+    let before_report = Instant::now() - CAPTURE_STATS_LOG_INTERVAL;
+    state.reporter.last_report = before_report;
+    let packet = vec![ACTIVE_LEVEL; samples];
+    for _ in 0..=CAPACITY {
+        sender.submit(&packet);
+    }
+    let packet = state.next_packet(&receiver).unwrap();
+    assert!(!receiver.is_empty());
+    assert!(state.reporter.last_report > before_report);
+    assert!(state.reporter.pending.is_empty());
+    assert!(receiver.take_loss().is_empty());
+    receiver.recycle(packet);
+}

src/server/audio_service/audio_capture_error.rs

@@ -11,7 +11,10 @@ pub(super) struct CaptureErrorHandler {
 
 impl CaptureErrorHandler {
     pub(super) fn handle(&self, error: cpal::StreamError) {
-        if matches!(error, cpal::StreamError::StreamInterrupted { .. }) {
+        if matches!(
+            error,
+            cpal::StreamError::StreamInterrupted { .. } | cpal::StreamError::DeviceNotAvailable
+        ) {
             // ScreenCaptureKit can stop capture while the remote session stays open.
             // The observed -3821 error does not identify its underlying trigger.
             // https://developer.apple.com/documentation/screencapturekit/scstreamdelegate/stream(_:didstopwitherror:)
@@ -23,6 +26,20 @@ impl CaptureErrorHandler {
         }
     }
 
+    pub(super) fn process_frame(&self, process: impl FnOnce() -> hbb_common::ResultType<()>) {
+        // Defensive recovery: persistent processing failures with valid capture input
+        // have not been reproduced. Stop using a failed processor until stream replacement.
+        if self.needs_restart() {
+            return;
+        }
+        if let Err(error) = process() {
+            self.interrupted.store(true, Ordering::Relaxed);
+            hbb_common::log::error!(
+                "Failed to process captured audio frame; requesting stream restart: {error:#}"
+            );
+        }
+    }
+
     pub(super) fn needs_restart(&self) -> bool {
         self.interrupted.load(Ordering::Relaxed)
     }
@@ -69,4 +86,44 @@ mod tests {
         });
         assert!(!errors.needs_restart());
     }
+
+    #[test]
+    fn processing_failure_skips_remaining_frames_until_stream_replacement() {
+        const FRAME_SAMPLES: usize = 2;
+        const FRAMES_PER_CALLBACK: usize = 3;
+        const CALLBACK_COUNT: usize = 2;
+        const FAILURE_CALL: usize = 2;
+        let errors = CaptureErrorHandler::default();
+        let callback_errors = errors.clone();
+        let mut framer =
+            super::super::audio_capture::CaptureFrameBuffer::new(FRAME_SAMPLES).unwrap();
+        let input = [0.0; FRAME_SAMPLES * FRAMES_PER_CALLBACK];
+        let mut processed = 0;
+        let mut failures = 0;
+
+        // Inject an error to test recovery; this is not a valid-input backend failure reproduction.
+        for _ in 0..CALLBACK_COUNT {
+            framer.process(input.iter().copied(), |_| {
+                callback_errors.process_frame(|| {
+                    processed += 1;
+                    if processed >= FAILURE_CALL {
+                        failures += 1;
+                        hbb_common::anyhow::bail!("Injected capture processing failure");
+                    }
+                    Ok(())
+                });
+            });
+        }
+
+        assert_eq!(processed, FAILURE_CALL);
+        assert_eq!(failures, 1);
+        assert!(errors.needs_restart());
+        let replacement = CaptureErrorHandler::default();
+        replacement.process_frame(|| {
+            processed += 1;
+            Ok(())
+        });
+        assert_eq!(processed, FAILURE_CALL + 1);
+        assert!(!replacement.needs_restart());
+    }
 }

src/server/audio_service/audio_capture_queue.rs

@@ -0,0 +1,289 @@
+use super::{send_f32, GenericService};
+use hbb_common::{
+    anyhow::{bail, Context, Result},
+    log,
+};
+use magnum_opus::{Application::LowDelay, Channels, Encoder};
+use std::{
+    collections::VecDeque,
+    sync::{
+        atomic::{AtomicBool, AtomicUsize, Ordering},
+        Arc, Mutex, OnceLock, TryLockError,
+    },
+    thread::{JoinHandle, Thread},
+};
+
+#[path = "audio_capture_encoder.rs"]
+mod encoder;
+
+const CAPTURE_PCM_QUEUE_PACKETS: usize = 10;
+const CAPTURE_ENCODER_THREAD_NAME: &str = "audio-encoder";
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(super) struct CapturePcmLoss {
+    pub(super) dropped: usize,
+    pub(super) contention_dropped: usize,
+    pub(super) oversized: usize,
+    pub(super) recycle_failures: usize,
+}
+
+impl CapturePcmLoss {
+    pub(super) fn is_empty(&self) -> bool {
+        self.dropped == 0 && self.oversized == 0 && self.recycle_failures == 0
+    }
+
+    pub(super) fn add(&mut self, other: Self) {
+        self.dropped += other.dropped;
+        self.contention_dropped += other.contention_dropped;
+        self.oversized += other.oversized;
+        self.recycle_failures += other.recycle_failures;
+    }
+}
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(super) struct CapturePcmStats {
+    pub(super) loss: CapturePcmLoss,
+    pub(super) max_queued_packets: usize,
+}
+
+impl CapturePcmStats {
+    pub(super) fn is_empty(&self) -> bool {
+        self.loss.is_empty() && self.max_queued_packets == 0
+    }
+
+    pub(super) fn add(&mut self, other: Self) {
+        self.loss.add(other.loss);
+        self.max_queued_packets = self.max_queued_packets.max(other.max_queued_packets);
+    }
+}
+
+struct CapturePcmHandoff {
+    buffers: Mutex<CapturePcmBuffers>,
+    wake_thread: OnceLock<Thread>,
+    other_dropped: AtomicUsize,
+    contention_dropped: AtomicUsize,
+    oversized: AtomicUsize,
+    recycle_failures: AtomicUsize,
+    max_queued_packets: AtomicUsize,
+    max_samples: usize,
+}
+
+struct CapturePcmBuffers {
+    available: Vec<Vec<f32>>,
+    ready: VecDeque<(usize, Vec<f32>)>,
+}
+
+pub(super) struct CapturePcmSender {
+    handoff: Arc<CapturePcmHandoff>,
+    sequence: usize,
+}
+
+pub(super) struct CapturePcmReceiver {
+    handoff: Arc<CapturePcmHandoff>,
+}
+
+pub(super) struct CaptureEncoderConfig {
+    pub(super) sample_rate: u32,
+    pub(super) encode_channel: Channels,
+    pub(super) max_packet_samples: usize,
+}
+
+pub(super) struct CaptureEncoderWorker {
+    stop: Arc<AtomicBool>,
+    handle: Option<JoinHandle<()>>,
+}
+
+impl Drop for CaptureEncoderWorker {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::Release);
+        if let Some(handle) = self.handle.take() {
+            handle.thread().unpark();
+            if let Err(error) = handle.join() {
+                log::error!("Failed to join audio encoder thread: {error:?}");
+            }
+        }
+    }
+}
+
+struct CaptureEncoderContext {
+    receiver: CapturePcmReceiver,
+    encoder: Encoder,
+    service: GenericService,
+    stop: Arc<AtomicBool>,
+}
+
+pub(super) fn new_pcm_handoff(
+    capacity: usize,
+    max_samples: usize,
+) -> Result<(CapturePcmSender, CapturePcmReceiver)> {
+    if capacity == 0 || max_samples == 0 {
+        bail!("Audio capture PCM handoff requires nonzero capacity and packet size");
+    }
+    let handoff = Arc::new(CapturePcmHandoff {
+        buffers: Mutex::new(CapturePcmBuffers {
+            available: Vec::with_capacity(capacity),
+            ready: VecDeque::with_capacity(capacity),
+        }),
+        wake_thread: OnceLock::new(),
+        other_dropped: AtomicUsize::new(0),
+        contention_dropped: AtomicUsize::new(0),
+        oversized: AtomicUsize::new(0),
+        recycle_failures: AtomicUsize::new(0),
+        max_queued_packets: AtomicUsize::new(0),
+        max_samples,
+    });
+    {
+        // Initialize the mutex on this thread, including on platforms with lazy allocation.
+        let mut buffers = handoff.buffers.lock().unwrap();
+        for _ in 0..capacity {
+            buffers.available.push(Vec::with_capacity(max_samples));
+        }
+    }
+    Ok((
+        CapturePcmSender {
+            handoff: handoff.clone(),
+            sequence: 0,
+        },
+        CapturePcmReceiver { handoff },
+    ))
+}
+
+impl CapturePcmSender {
+    pub(super) fn set_wake_thread(&self, thread: Thread) -> Result<()> {
+        if self.handoff.wake_thread.set(thread).is_err() {
+            bail!("Audio capture PCM wake thread is already configured");
+        }
+        Ok(())
+    }
+
+    pub(super) fn submit(&mut self, input: &[f32]) {
+        let sequence = self.sequence;
+        self.sequence = self.sequence.wrapping_add(1);
+        if input.len() > self.handoff.max_samples {
+            self.handoff.oversized.fetch_add(1, Ordering::Relaxed);
+            self.wake();
+            return;
+        }
+        // Do not wait for a descheduled worker. Contention rejects the current
+        // packet even if buffers are available; this is separate from drop-oldest
+        // when the buffer pool is exhausted.
+        let mut buffers = match self.handoff.buffers.try_lock() {
+            Ok(buffers) => buffers,
+            Err(TryLockError::WouldBlock) => {
+                self.handoff
+                    .contention_dropped
+                    .fetch_add(1, Ordering::Relaxed);
+                self.wake();
+                return;
+            }
+            Err(TryLockError::Poisoned(error)) => {
+                self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
+                log::error!("Audio capture PCM handoff is poisoned: {error}");
+                self.wake();
+                return;
+            }
+        };
+        if let Some(mut buffer) = self.take_buffer(&mut buffers) {
+            buffer.clear();
+            buffer.extend_from_slice(input);
+            buffers.ready.push_back((sequence, buffer));
+            self.handoff
+                .max_queued_packets
+                .fetch_max(buffers.ready.len(), Ordering::Relaxed);
+        } else {
+            self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
+        }
+        drop(buffers);
+        self.wake();
+    }
+
+    fn take_buffer(&self, buffers: &mut CapturePcmBuffers) -> Option<Vec<f32>> {
+        buffers.available.pop().or_else(|| {
+            let buffer = buffers.ready.pop_front().map(|(_, buffer)| buffer);
+            if buffer.is_some() {
+                self.handoff.other_dropped.fetch_add(1, Ordering::Relaxed);
+            }
+            buffer
+        })
+    }
+
+    fn wake(&self) {
+        if let Some(thread) = self.handoff.wake_thread.get() {
+            thread.unpark();
+        }
+    }
+}
+
+impl CapturePcmReceiver {
+    #[cfg(test)]
+    pub(super) fn pop(&self) -> Option<Vec<f32>> {
+        self.pop_packet().map(|(_, buffer)| buffer)
+    }
+
+    fn pop_packet(&self) -> Option<(usize, Vec<f32>)> {
+        self.handoff.buffers.lock().unwrap().ready.pop_front()
+    }
+
+    pub(super) fn recycle(&self, mut buffer: Vec<f32>) {
+        buffer.clear();
+        let mut buffers = self.handoff.buffers.lock().unwrap();
+        if buffers.available.len() == buffers.available.capacity() {
+            self.handoff
+                .recycle_failures
+                .fetch_add(1, Ordering::Relaxed);
+        } else {
+            buffers.available.push(buffer);
+        }
+    }
+
+    pub(super) fn is_empty(&self) -> bool {
+        self.handoff.buffers.lock().unwrap().ready.is_empty()
+    }
+
+    pub(super) fn take_loss(&self) -> CapturePcmLoss {
+        let contention_dropped = self.handoff.contention_dropped.swap(0, Ordering::Relaxed);
+        CapturePcmLoss {
+            dropped: self.handoff.other_dropped.swap(0, Ordering::Relaxed) + contention_dropped,
+            contention_dropped,
+            oversized: self.handoff.oversized.swap(0, Ordering::Relaxed),
+            recycle_failures: self.handoff.recycle_failures.swap(0, Ordering::Relaxed),
+        }
+    }
+
+    pub(super) fn take_stats(&self) -> CapturePcmStats {
+        CapturePcmStats {
+            loss: self.take_loss(),
+            max_queued_packets: self.handoff.max_queued_packets.swap(0, Ordering::Relaxed),
+        }
+    }
+}
+
+pub(super) fn start_capture_encoder(
+    config: CaptureEncoderConfig,
+    service: GenericService,
+) -> Result<(CapturePcmSender, CaptureEncoderWorker)> {
+    let (sender, receiver) = new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, config.max_packet_samples)?;
+    let encoder = Encoder::new(config.sample_rate, config.encode_channel, LowDelay)?;
+    let stop = Arc::new(AtomicBool::new(false));
+    let context = CaptureEncoderContext {
+        receiver,
+        encoder,
+        service,
+        stop: stop.clone(),
+    };
+    let handle = std::thread::Builder::new()
+        .name(CAPTURE_ENCODER_THREAD_NAME.to_owned())
+        .spawn(move || encoder::run_capture_encoder(context, config))
+        .with_context(|| "Failed to start audio encoder thread")?;
+    let wake_thread = handle.thread().clone();
+    let worker = CaptureEncoderWorker {
+        stop,
+        handle: Some(handle),
+    };
+    sender.set_wake_thread(wake_thread)?;
+    Ok((sender, worker))
+}
+
+#[cfg(test)]
+#[path = "audio_capture_queue_tests.rs"]
+mod tests;

src/server/audio_service/audio_capture_queue_tests.rs

@@ -0,0 +1,169 @@
+use super::*;
+use crate::audio_resampler::allocation_tests::assert_no_allocations;
+use std::{sync::mpsc, time::Duration};
+
+const PACKET_SAMPLES: usize = 4;
+const TEST_TIMEOUT: Duration = Duration::from_secs(2);
+const CONCURRENT_PACKETS: usize = 10_000;
+
+#[derive(Clone, Copy)]
+enum PausePoint {
+    Recycle,
+    Consume,
+}
+
+struct WorkerPause {
+    entered: mpsc::Sender<()>,
+    resume: mpsc::Receiver<()>,
+}
+
+fn paused_worker(
+    receiver: CapturePcmReceiver,
+    point: PausePoint,
+    pause: WorkerPause,
+) -> CapturePcmReceiver {
+    let recycled = match point {
+        PausePoint::Recycle => Some(receiver.pop_packet().unwrap().1),
+        PausePoint::Consume => None,
+    };
+    let consumed = {
+        let mut buffers = receiver.handoff.buffers.lock().unwrap();
+        let consumed = match recycled {
+            Some(mut packet) => {
+                packet.clear();
+                buffers.available.push(packet);
+                None
+            }
+            None => Some(buffers.ready.pop_front().unwrap().1),
+        };
+        pause.entered.send(()).unwrap();
+        pause.resume.recv().unwrap();
+        consumed
+    };
+    if let Some(packet) = consumed {
+        receiver.recycle(packet);
+    }
+    receiver
+}
+
+fn assert_callback_progress(point: PausePoint) {
+    let (mut sender, receiver) =
+        new_pcm_handoff(CAPTURE_PCM_QUEUE_PACKETS, PACKET_SAMPLES).unwrap();
+    for sequence in 0..CAPTURE_PCM_QUEUE_PACKETS {
+        sender.submit(&[sequence as f32; PACKET_SAMPLES]);
+    }
+    let (entered_tx, entered_rx) = mpsc::channel();
+    let (resume_tx, resume_rx) = mpsc::channel();
+    let pause = WorkerPause {
+        entered: entered_tx,
+        resume: resume_rx,
+    };
+    let worker = std::thread::spawn(move || paused_worker(receiver, point, pause));
+    sender.set_wake_thread(worker.thread().clone()).unwrap();
+    entered_rx.recv_timeout(TEST_TIMEOUT).unwrap();
+    let (completed_tx, completed_rx) = mpsc::channel();
+    let callback = std::thread::spawn(move || {
+        let packet = [CAPTURE_PCM_QUEUE_PACKETS as f32; PACKET_SAMPLES];
+        assert_no_allocations(|| sender.submit(&packet));
+        completed_tx.send(()).unwrap();
+        sender
+    });
+    let completed = completed_rx.recv_timeout(TEST_TIMEOUT);
+    // Release and join both threads before asserting progress, including on failure.
+    resume_tx.send(()).unwrap();
+    let receiver = worker.join().unwrap();
+    let mut sender = callback.join().unwrap();
+    assert!(completed.is_ok(), "callback waited for the encoder worker");
+    assert_eq!(
+        receiver.take_loss(),
+        CapturePcmLoss {
+            dropped: 1,
+            contention_dropped: 1,
+            ..Default::default()
+        }
+    );
+    assert_packets_after_contention(&mut sender, &receiver);
+}
+
+fn assert_packets_after_contention(sender: &mut CapturePcmSender, receiver: &CapturePcmReceiver) {
+    let next_sequence = CAPTURE_PCM_QUEUE_PACKETS + 1;
+    sender.submit(&[next_sequence as f32; PACKET_SAMPLES]);
+    for expected in (1..CAPTURE_PCM_QUEUE_PACKETS).chain(std::iter::once(next_sequence)) {
+        let (sequence, packet) = receiver.pop_packet().unwrap();
+        assert_eq!(sequence, expected);
+        assert_eq!(packet, [expected as f32; PACKET_SAMPLES]);
+        receiver.recycle(packet);
+    }
+    assert!(receiver.is_empty());
+    assert!(receiver.take_loss().is_empty());
+    assert_pool_restored(receiver, CAPTURE_PCM_QUEUE_PACKETS);
+}
+
+fn assert_pool_restored(receiver: &CapturePcmReceiver, capacity: usize) {
+    let buffers = receiver.handoff.buffers.lock().unwrap();
+    assert_eq!(buffers.available.len(), capacity);
+    assert!(buffers
+        .available
+        .iter()
+        .all(|buffer| buffer.capacity() >= PACKET_SAMPLES));
+}
+
+#[test]
+fn callback_finishes_while_worker_recycles_a_buffer() {
+    assert_callback_progress(PausePoint::Recycle);
+}
+
+#[test]
+fn callback_finishes_while_worker_releases_a_ready_packet() {
+    assert_callback_progress(PausePoint::Consume);
+}
+
+fn consume_concurrently(
+    receiver: CapturePcmReceiver,
+    finished: Arc<AtomicBool>,
+    first_sequence: usize,
+) -> (CapturePcmReceiver, usize) {
+    let mut received = 0;
+    let mut next_ordinal = 0;
+    loop {
+        if let Some((sequence, packet)) = receiver.pop_packet() {
+            let ordinal = sequence.wrapping_sub(first_sequence);
+            assert!(ordinal >= next_ordinal && ordinal < CONCURRENT_PACKETS);
+            assert_eq!(packet, [ordinal as f32; PACKET_SAMPLES]);
+            next_ordinal = ordinal + 1;
+            received += 1;
+            receiver.recycle(packet);
+        } else if finished.load(Ordering::Acquire) && receiver.is_empty() {
+            return (receiver, received);
+        } else {
+            std::thread::yield_now();
+        }
+    }
+}
+
+#[test]
+fn concurrent_handoff_preserves_order_buffers_and_loss_counts_across_sequence_wrap() {
+    const FIRST_SEQUENCE: usize = usize::MAX - CONCURRENT_PACKETS / 2;
+    for capacity in [1, 2, CAPTURE_PCM_QUEUE_PACKETS] {
+        let (mut sender, receiver) = new_pcm_handoff(capacity, PACKET_SAMPLES).unwrap();
+        sender.sequence = FIRST_SEQUENCE;
+        let finished = Arc::new(AtomicBool::new(false));
+        let worker_finished = finished.clone();
+        let worker = std::thread::spawn(move || {
+            consume_concurrently(receiver, worker_finished, FIRST_SEQUENCE)
+        });
+        assert_no_allocations(|| {
+            for ordinal in 0..CONCURRENT_PACKETS {
+                sender.submit(&[ordinal as f32; PACKET_SAMPLES]);
+            }
+        });
+        finished.store(true, Ordering::Release);
+        let (receiver, received) = worker.join().unwrap();
+        let stats = receiver.take_stats();
+        assert_eq!(received + stats.loss.dropped, CONCURRENT_PACKETS);
+        assert_eq!(stats.loss.oversized, 0);
+        assert_eq!(stats.loss.recycle_failures, 0);
+        assert!(stats.max_queued_packets <= capacity);
+        assert_pool_restored(&receiver, capacity);
+    }
+}