Merge pull request #42 from bongodevs/dev

v1.0.8

Author: Appu

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rustalink"
-version = "1.0.7"
+version = "1.0.8"
 edition = "2024"
 license = "Apache-2.0"
 authors = ["appujet", "notdeltaxd", "contributors"]

Cargo.toml

@@ -79,10 +79,8 @@ impl TapeEffect {
         }
     }
 
-    /// Process a PCM frame. Unlike the previous implementation which worked in-place,
-    /// this uses an internal resampling buffer to handle rate changes correctly.
     pub fn process(&mut self, frame: &mut [i16]) {
-        if frame.is_empty() {
+        if frame.is_empty() || !self.is_active() {
             return;
         }
 
@@ -156,7 +154,8 @@ impl TapeEffect {
 
         if self.read_pos > (self.sample_rate as f64 * channels as f64 * 2.0) {
             let integral = (self.read_pos.floor() as usize / channels) * channels;
-            self.input_buffer.drain(0..integral);
+            self.input_buffer.copy_within(integral.., 0);
+            self.input_buffer.truncate(self.input_buffer.len() - integral);
             self.read_pos -= integral as f64;
         }
     }

src/audio/effects/tape.rs

@@ -126,7 +126,6 @@ impl VolumeEffect {
         value.signum() * limited.min(INT16_MAX_F)
     }
 
-    /// Process a stereo interleaved i16 frame **in-place**.
     pub fn process(&mut self, frame: &mut [i16]) {
         let sample_count = frame.len();
         if sample_count == 0 {
@@ -158,6 +157,10 @@ impl VolumeEffect {
             (v, v)
         };
 
+        if !self.fade_active && (gain_start - 1.0).abs() < 0.0001 {
+            return;
+        }
+
         let step = if sample_count > 1 {
             (gain_end - gain_start) / (sample_count - 1) as f32
         } else {

src/audio/effects/volume.rs

@@ -49,6 +49,7 @@ pub struct AudioProcessor {
     source_rate: u32,
     channels: usize,
     config: PlayerConfig,
+    recoverable_errors: u32,
 }
 
 impl AudioProcessor {
@@ -119,6 +120,7 @@ impl AudioProcessor {
             source_rate: sample_rate,
             channels,
             config,
+            recoverable_errors: 0,
         })
     }
 
@@ -152,6 +154,7 @@ impl AudioProcessor {
 
             match self.decoder.decode(&packet) {
                 Ok(decoded) => {
+                    self.recoverable_errors = 0;
                     let spec = *decoded.spec();
                     let mut buf = self.sample_buf.take().unwrap_or_else(|| {
                         SampleBuffer::<i16>::new(decoded.capacity() as u64, spec)
@@ -260,7 +263,14 @@ impl AudioProcessor {
                     self.sample_buf = Some(buf);
                 }
                 Err(Error::IoError(e)) if e.kind() == ErrorKind::UnexpectedEof => break,
-                Err(Error::DecodeError(e)) => warn!("Decode error (recoverable): {e}"),
+                Err(Error::DecodeError(e)) => {
+                    self.recoverable_errors += 1;
+                    if self.recoverable_errors == 1 {
+                        warn!("Decode error (recoverable): {e}");
+                    } else if self.recoverable_errors.is_multiple_of(100) {
+                        warn!("Decode error (recoverable, x{}): {e}", self.recoverable_errors);
+                    }
+                }
                 Err(e) => {
                     self.send_error(format!("Decode error: {e}"));
                     return Err(e);

src/audio/processor.rs

@@ -7,30 +7,35 @@ pub struct SincResampler {
     index: f32,
     channels: usize,
     taps: usize,
-    window: Vec<f32>,
+    table: Vec<f32>,
     buffer: Vec<VecDeque<f32>>,
 }
 
 impl SincResampler {
     pub fn new(source_rate: u32, target_rate: u32, channels: usize) -> Self {
         let taps = 32;
-        let mut window = Vec::with_capacity(taps);
+        let mut table = Vec::with_capacity(taps);
         let m = taps as f32 - 1.0;
+        let half_taps = (taps / 2) as f32;
 
         for i in 0..taps {
+            let offset = i as f32 - half_taps;
+            
             let a0 = 0.42;
             let a1 = 0.5;
             let a2 = 0.08;
             let pi_n_m = 2.0 * std::f32::consts::PI * i as f32 / m;
-            window.push(a0 - a1 * pi_n_m.cos() + a2 * (2.0 * pi_n_m).cos());
+            let window = a0 - a1 * pi_n_m.cos() + a2 * (2.0 * pi_n_m).cos();
+
+            table.push(Self::sinc(offset) * window);
         }
 
         Self {
             ratio: source_rate as f32 / target_rate as f32,
             index: 0.0,
             channels,
             taps,
-            window,
+            table,
             buffer: vec![VecDeque::from(vec![0.0; taps]); channels],
         }
     }
@@ -45,7 +50,6 @@ impl SincResampler {
 
     pub fn process(&mut self, input: &[i16], output: &mut PooledBuffer) {
         let num_frames = input.len() / self.channels;
-        let half_taps = (self.taps / 2) as f32;
 
         for frame in 0..num_frames {
             for ch in 0..self.channels {
@@ -58,8 +62,7 @@ impl SincResampler {
                     let mut sum = 0.0;
 
                     for i in 0..self.taps {
-                        let offset = (i as f32 - half_taps) - self.index;
-                        sum += self.buffer[ch][i] * Self::sinc(offset) * self.window[i];
+                        sum += self.buffer[ch][i] * self.table[i];
                     }
 
                     output.push(sum.clamp(i16::MIN as f32, i16::MAX as f32) as i16);

src/audio/resample/sinc.rs

@@ -252,7 +252,6 @@ pub fn prefetch_loop(
                 state.buffered += n;
                 state.chunks.push_back(chunk);
                 cvar.notify_all();
-                retry_count = 0;
             }
             Ok(None) => {
                 response = None;
@@ -274,7 +273,7 @@ pub fn prefetch_loop(
                 retry_count += 1;
 
                 if retry_count > MAX_FETCH_RETRIES {
-                    warn!("prefetch: read failed fatally: {}", e);
+                    warn!("prefetch: read failed fatally after {} retries: {}", MAX_FETCH_RETRIES, e);
                     let (lock, cvar) = &*shared;
                     let mut state = lock.lock();
                     state.error = Some(e.to_string());

src/audio/source/http/prefetcher.rs

@@ -19,6 +19,7 @@ pub struct CircularFileWriter {
 struct WriterState {
     file: Option<File>,
     lines_since_prune: u32,
+    is_pruning: bool,
 }
 
 impl CircularFileWriter {
@@ -29,6 +30,7 @@ impl CircularFileWriter {
             state: Arc::new(Mutex::new(WriterState {
                 file: None,
                 lines_since_prune: 0,
+                is_pruning: false,
             })),
         }
     }
@@ -45,26 +47,42 @@ impl CircularFileWriter {
         Ok(state.file.as_mut().expect("file was just opened"))
     }
 
-    fn prune(&self, state: &mut WriterState) -> io::Result<()> {
-        // Close file before pruning
-        state.file = None;
+    fn spawn_prune(&self) {
+        let path = self.path.clone();
+        let max_lines = self.max_lines;
+        let state_arc = self.state.clone();
 
-        if !Path::new(&self.path).exists() {
+        std::thread::spawn(move || {
+            if let Err(e) = Self::do_prune(&path, max_lines) {
+                eprintln!("Failed to prune log file '{}': {}", path, e);
+            }
+            let mut state = state_arc.lock();
+            state.is_pruning = false;
+        });
+    }
+
+    fn do_prune(path: &str, max_lines: u32) -> io::Result<()> {
+        if !Path::new(path).exists() {
             return Ok(());
         }
 
         let lines: Vec<String> = {
-            let file = File::open(&self.path)?;
+            let file = File::open(path)?;
             let reader = BufReader::new(file);
             reader.lines().collect::<Result<_, _>>()?
         };
 
-        if lines.len() > self.max_lines as usize {
-            let start = lines.len() - self.max_lines as usize;
-            let mut file = File::create(&self.path)?;
-            for line in &lines[start..] {
-                writeln!(file, "{}", line)?;
+        if lines.len() > max_lines as usize {
+            let start = lines.len() - max_lines as usize;
+            // Atomic-ish replacement: write to .tmp then rename
+            let tmp_path = format!("{}.tmp", path);
+            {
+                let mut file = File::create(&tmp_path)?;
+                for line in &lines[start..] {
+                    writeln!(file, "{}", line)?;
+                }
             }
+            std::fs::rename(tmp_path, path)?;
         }
         Ok(())
     }
@@ -73,19 +91,19 @@ impl CircularFileWriter {
 impl io::Write for CircularFileWriter {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         let mut state = self.state.lock();
+        
         let file = self.ensure_file_open(&mut state)?;
-
         file.write_all(buf)?;
 
         let new_lines = buf.iter().filter(|&&b| b == b'\n').count() as u32;
         state.lines_since_prune += new_lines;
 
         let prune_threshold = (self.max_lines / 10).max(50);
-        if state.lines_since_prune >= prune_threshold {
-            if let Err(e) = self.prune(&mut state) {
-                eprintln!("Failed to prune log file '{}': {}", self.path, e);
-            }
+        if state.lines_since_prune >= prune_threshold && !state.is_pruning {
+            state.is_pruning = true;
             state.lines_since_prune = 0;
+            state.file = None; // Close file so rename can happen on Windows if needed
+            self.spawn_prune();
         }
 
         Ok(buf.len())

src/common/logger/writer.rs

@@ -26,6 +26,7 @@ pub struct DaveHandler {
     protocol_version: u16,
     pending_transitions: HashMap<u16, u16>,
     external_sender_set: bool,
+    saved_external_sender: Option<Vec<u8>>,
     pending_proposals: Vec<Vec<u8>>,
     pending_handshake: Vec<(Vec<u8>, bool)>,
     was_ready: bool,
@@ -43,6 +44,7 @@ impl DaveHandler {
             protocol_version: 0,
             pending_transitions: HashMap::new(),
             external_sender_set: false,
+            saved_external_sender: None,
             pending_proposals: Vec::new(),
             pending_handshake: Vec::new(),
             was_ready: false,
@@ -92,13 +94,31 @@ impl DaveHandler {
         self.was_ready = false;
 
         debug!("DAVE session setup (v{})", version);
-        session.create_key_package().map_err(map_boxed_err)
+        let key_package = session.create_key_package().map_err(map_boxed_err)?;
+
+        if let Some(saved) = self.saved_external_sender.clone() {
+            if let Some(sess) = &mut self.session {
+                match sess.set_external_sender(&saved) {
+                    Ok(()) => {
+                        self.external_sender_set = true;
+                        debug!("DAVE re-applied saved external sender after epoch reset");
+                    }
+                    Err(e) => {
+                        warn!("DAVE failed to re-apply saved external sender: {e}");
+                        self.saved_external_sender = None;
+                    }
+                }
+            }
+        }
+
+        Ok(key_package)
     }
 
     pub fn reset(&mut self) {
         self.protocol_version = 0;
         self.pending_transitions.clear();
         self.external_sender_set = false;
+        self.saved_external_sender = None;
         self.pending_proposals.clear();
         self.pending_handshake.clear();
         self.was_ready = false;
@@ -143,6 +163,7 @@ impl DaveHandler {
         if let Some(session) = &mut self.session {
             session.set_external_sender(data).map_err(map_boxed_err)?;
             self.external_sender_set = true;
+            self.saved_external_sender = Some(data.to_vec());
 
             if !self.pending_proposals.is_empty() {
                 debug!(

src/gateway/encryption.rs

@@ -312,6 +312,7 @@ impl<'a> SessionState<'a> {
         {
             let mut state = self.persistent_state.lock().await;
             state.ssrc = self.ssrc;
+            state.selected_mode = Some(self.selected_mode.clone());
         }
 
         if self.gateway.channel_id.0 > 0 {
@@ -413,6 +414,7 @@ impl<'a> SessionState<'a> {
                 state.udp_addr = Some(addr);
                 state.session_key = Some(key);
                 state.ssrc = self.ssrc;
+                state.selected_mode = Some(self.selected_mode.clone());
             }
 
             self.launch_speak_loop(addr, key).await;
@@ -456,16 +458,19 @@ impl<'a> SessionState<'a> {
 
         self.backoff.reset();
 
-        let (addr, key, ssrc) = {
+        let (addr, key, ssrc, mode) = {
             let state = self.persistent_state.lock().await;
-            (state.udp_addr, state.session_key, state.ssrc)
+            (state.udp_addr, state.session_key, state.ssrc, state.selected_mode.clone())
         };
 
         match (addr, key) {
             (Some(addr), Some(key)) => {
                 self.udp_addr = Some(addr);
                 self.session_key = Some(key);
                 self.ssrc = ssrc;
+                if let Some(m) = mode {
+                    self.selected_mode = m;
+                }
 
                 self.launch_speak_loop(addr, key).await;
                 self.send_json(