Fixes repeated call audio drops

rust/telepathy-audio/src/devices/mock_host.rs

@@ -100,16 +100,20 @@ impl<I: Default, O: Default> Default for MockAudioHost<I, O> {
     }
 }
 
-/// In-process audio input that emits silence at real-time pace.
+/// In-process audio input that emits a deterministic signal at real-time pace.
 #[derive(Debug, Clone)]
 pub struct MockAudioInput {
     sample_rate: u32,
+    sample_index: u64,
 }
 
 impl MockAudioInput {
-    /// Creates a new mock input that emits silence at the given sample rate.
+    /// Creates a new mock input that emits changing non-silent samples at the given sample rate.
     pub fn new(sample_rate: u32) -> Self {
-        Self { sample_rate }
+        Self {
+            sample_rate,
+            sample_index: 0,
+        }
     }
 }
 
@@ -125,7 +129,11 @@ impl AudioInput for MockAudioInput {
         if frame_seconds.is_normal() || frame_seconds > 0.0 {
             thread::sleep(Duration::from_secs_f64(frame_seconds));
         }
-        dst.fill(0.0);
+        for sample in dst.iter_mut() {
+            let ramp_position = (self.sample_index % 96) as f32 / 95.0;
+            *sample = (ramp_position * 2.0 - 1.0) * 0.25;
+            self.sample_index = self.sample_index.wrapping_add(1);
+        }
         Ok(dst.len())
     }
 }
@@ -143,3 +151,24 @@ impl AudioOutput for MockAudioOutput {
         Ok(0)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn mock_audio_input_emits_changing_non_silent_samples() {
+        let mut input = MockAudioInput::new(1_000_000);
+        let mut first = [0.0; 4];
+        let mut second = [0.0; 4];
+
+        let first_read = input.read_into(&mut first).unwrap();
+        let second_read = input.read_into(&mut second).unwrap();
+
+        assert_eq!(first_read, first.len());
+        assert_eq!(second_read, second.len());
+        assert!(first.iter().any(|sample| *sample != 0.0));
+        assert!(second.iter().any(|sample| *sample != 0.0));
+        assert_ne!(first, second);
+    }
+}

rust/telepathy-core/src/internal/connections.rs

@@ -4,8 +4,8 @@ use bytes::Bytes;
 use iroh::endpoint::Connection;
 use kanal::{AsyncReceiver, Sender};
 use std::collections::{BTreeMap, HashSet};
-use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::Relaxed;
+use std::sync::atomic::{AtomicU32, AtomicUsize};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use telepathy_audio::FRAME_SIZE;
@@ -89,15 +89,15 @@ pub(crate) struct ConstConnection {
     /// (if not cloned), leveraging `Bytes::try_into_mut()` for recovery.
     packet_buffers: Arc<BufferPool>,
 
-    sequence_number: u32,
+    sequence_number: Arc<AtomicU32>,
 }
 
 impl ConstConnection {
-    pub(crate) fn new(connection: Connection) -> Self {
+    pub(crate) fn new(connection: Connection, sequence_number: Arc<AtomicU32>) -> Self {
         Self {
             connection,
             packet_buffers: Arc::new(BufferPool::new(PACKET_POOL_SIZE, PACKET_BUFFER_CAPACITY)),
-            sequence_number: 0,
+            sequence_number,
         }
     }
 }
@@ -106,7 +106,11 @@ impl TelepathyConnection for ConstConnection {
     fn send(&mut self, packet: &Bytes) -> usize {
         let pooled_buffer = BufferPool::acquire(&self.packet_buffers);
         let is_keep_alive = is_keep_alive_payload(packet.as_ref());
-        let sequence_number = self.sequence_number;
+        let sequence_number = if is_keep_alive {
+            self.sequence_number.load(Relaxed)
+        } else {
+            self.sequence_number.fetch_add(1, Relaxed)
+        };
         let pooled_bytes = prepare_packet(
             pooled_buffer,
             packet.as_ref(),
@@ -120,10 +124,6 @@ impl TelepathyConnection for ConstConnection {
             .send_datagram((*pooled_bytes).clone())
             .is_ok();
 
-        if ok && !is_keep_alive {
-            self.sequence_number = self.sequence_number.wrapping_add(1);
-        }
-
         ok as usize
     }
 }
@@ -146,7 +146,7 @@ pub(crate) struct DynamicConnection {
     /// `insert_audio` to anchor playout: the first packet it sees becomes the
     /// anchor, jump-starting playback at that point rather than waiting for
     /// the original sequence origin.
-    sequence_number: u32,
+    sequence_number: Arc<AtomicU32>,
 }
 
 impl TelepathyConnection for DynamicConnection {
@@ -164,7 +164,7 @@ impl TelepathyConnection for DynamicConnection {
         let mut i = 0;
         let mut successful_sends = 0;
         let is_keep_alive = is_keep_alive_payload(packet.as_ref());
-        let sequence_number = self.sequence_number;
+        let sequence_number = self.sequence_number.load(Relaxed);
 
         while i < self.connections.len() {
             let pooled_buffer = BufferPool::acquire(&self.packet_buffers);
@@ -197,19 +197,19 @@ impl TelepathyConnection for DynamicConnection {
         }
 
         if successful_sends > 0 && !is_keep_alive {
-            self.sequence_number = self.sequence_number.wrapping_add(1);
+            self.sequence_number.fetch_add(1, Relaxed);
         }
         successful_sends
     }
 }
 
 impl DynamicConnection {
-    pub(crate) fn new(new_connections: SharedConnections) -> Self {
+    pub(crate) fn new(new_connections: SharedConnections, sequence_number: Arc<AtomicU32>) -> Self {
         Self {
             new_connections,
             connections: Vec::new(),
             packet_buffers: Arc::new(BufferPool::new(PACKET_POOL_SIZE, PACKET_BUFFER_CAPACITY)),
-            sequence_number: 0,
+            sequence_number,
         }
     }
 }
@@ -516,3 +516,37 @@ fn is_keep_alive_payload(payload: &[u8]) -> bool {
 fn seq_before(a: u32, b: u32) -> bool {
     a != b && a.wrapping_sub(b) > 0x8000_0000
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stale_keepalive_rejects_restarted_sequence_number() {
+        let mut jitter = AudioJitterBuffer::new(48_000);
+
+        jitter.reset_after_keepalive(80);
+
+        assert!(!jitter.insert_audio(0, Bytes::from_static(b"fresh call"), Instant::now()));
+    }
+
+    #[test]
+    fn keepalive_accepts_equal_floor_sequence_number() {
+        // The keepalive carries the *next* audio sequence number, so the first
+        // real frame after silence is allowed to arrive at exactly that floor.
+        let mut jitter = AudioJitterBuffer::new(48_000);
+
+        jitter.reset_after_keepalive(80);
+
+        assert!(jitter.insert_audio(80, Bytes::from_static(b"fresh call"), Instant::now()));
+    }
+
+    #[test]
+    fn stale_keepalive_accepts_continued_sequence_number() {
+        let mut jitter = AudioJitterBuffer::new(48_000);
+
+        jitter.reset_after_keepalive(80);
+
+        assert!(jitter.insert_audio(81, Bytes::from_static(b"fresh call"), Instant::now()));
+    }
+}

rust/telepathy-core/src/internal/core.rs

@@ -1436,7 +1436,7 @@ where
         if let Some(o) = optional {
             let input_handle = spawn_task(audio_input(
                 input_helper.receiver(),
-                ConstConnection::new(o.connection.clone()),
+                ConstConnection::new(o.connection.clone(), self.core_state.audio_sequence.clone()),
                 stop_io.clone(),
             ));
 
@@ -1721,7 +1721,10 @@ where
 
         let input_handle = spawn_task(audio_input(
             input_helper.receiver(),
-            DynamicConnection::new(connection_sender.clone()),
+            DynamicConnection::new(
+                connection_sender.clone(),
+                self.core_state.audio_sequence.clone(),
+            ),
             stop_io.clone(),
         ));
 

rust/telepathy-core/src/internal/state.rs

@@ -9,7 +9,7 @@ use iroh::{PublicKey, SecretKey, TransportAddr};
 use std::collections::HashMap;
 use std::hash::{DefaultHasher, Hash, Hasher};
 use std::sync::atomic::Ordering::Relaxed;
-use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize};
 use std::sync::{Arc, Mutex as StdMutex};
 use std::time::Duration;
 use telepathy_audio::RnnModel;
@@ -416,6 +416,9 @@ pub struct CoreState {
 
     /// serializes access to shared volume state
     output_lock: Arc<StdMutex<()>>,
+
+    /// global audio sequence number
+    pub(crate) audio_sequence: Arc<AtomicU32>,
 }
 
 impl CoreState {

system-tests/scenarios/call_repeated_without_restart.yaml

@@ -0,0 +1,60 @@
+name: call_repeated_without_restart
+steps:
+  - actor: alice
+    expect_event:
+      type: ready
+      timeout: 10.0
+      match:
+        version: "*"
+  - actor: bob
+    expect_event:
+      type: ready
+      timeout: 10.0
+      match:
+        version: "*"
+  - concurrent:
+      - actor: alice
+        send:
+          cmd: add_contact
+          args:
+            contact_id: bob
+            peer_id: ${bob.peer_id}
+        expect_ack:
+          ok: true
+      - actor: bob
+        send:
+          cmd: add_contact
+          args:
+            contact_id: alice
+            peer_id: ${alice.peer_id}
+        expect_ack:
+          ok: true
+  - concurrent:
+      - actor: alice
+        send:
+          cmd: start_session
+          args:
+            contact_id: bob
+        expect_ack:
+          ok: true
+      - actor: bob
+        send:
+          cmd: start_session
+          args:
+            contact_id: alice
+        expect_ack:
+          ok: true
+  - actor: alice
+    expect_event:
+      type: session_status
+      timeout: 45.0
+      match:
+        status: Connected
+        peer: ${bob.peer_id}
+  - actor: bob
+    expect_event:
+      type: session_status
+      timeout: 45.0
+      match:
+        status: Connected
+        peer: ${alice.peer_id}