diff --git a/rust/telepathy-audio/src/devices/mock_host.rs b/rust/telepathy-audio/src/devices/mock_host.rs index ae7c378..fe50ec8 100644 --- a/rust/telepathy-audio/src/devices/mock_host.rs +++ b/rust/telepathy-audio/src/devices/mock_host.rs @@ -100,16 +100,20 @@ impl Default for MockAudioHost { } } -/// 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); + } +} diff --git a/rust/telepathy-core/src/internal/connections.rs b/rust/telepathy-core/src/internal/connections.rs index 46be259..47e7ea2 100644 --- a/rust/telepathy-core/src/internal/connections.rs +++ b/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, - sequence_number: u32, + sequence_number: Arc, } impl ConstConnection { - pub(crate) fn new(connection: Connection) -> Self { + pub(crate) fn new(connection: Connection, sequence_number: Arc) -> 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, } 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) -> 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())); + } +} diff --git a/rust/telepathy-core/src/internal/core.rs b/rust/telepathy-core/src/internal/core.rs index 53c6029..d250c0e 100644 --- a/rust/telepathy-core/src/internal/core.rs +++ b/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(), )); diff --git a/rust/telepathy-core/src/internal/state.rs b/rust/telepathy-core/src/internal/state.rs index c985c6b..6a645db 100644 --- a/rust/telepathy-core/src/internal/state.rs +++ b/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>, + + /// global audio sequence number + pub(crate) audio_sequence: Arc, } impl CoreState { diff --git a/system-tests/scenarios/call_repeated_without_restart.yaml b/system-tests/scenarios/call_repeated_without_restart.yaml new file mode 100644 index 0000000..a2ff450 --- /dev/null +++ b/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} diff --git a/system-tests/tests/test_scenarios.py b/system-tests/tests/test_scenarios.py index 36b18bf..bac2860 100644 --- a/system-tests/tests/test_scenarios.py +++ b/system-tests/tests/test_scenarios.py @@ -70,9 +70,20 @@ class RoomCliGroup: ), ] +REPEATED_CALL_PROFILE = NetworkProfile( + name="repeated_call_moderate_delay_zero_loss", + delay_ms=300, + jitter_ms=0, + loss_pct=0.0, + burst_loss=False, + seed=104, +) + ROOM_THREE_ACTORS = ["alice", "bob", "carol"] ROOM_FOUR_ACTORS = ["alice", "bob", "carol", "dave"] ROOM_TWENTY_ACTORS = [f"peer{index:02d}" for index in range(1, 21)] +FULL_LOSS_SAMPLES_PER_STAT = 4_800 +MAX_CONSECUTIVE_FULL_LOSS_STATS = 20 def _generate_identity() -> Identity: @@ -493,6 +504,119 @@ def _has_error_event(messages: list[dict]) -> bool: return False +def _call_state_name(state: object) -> str | None: + if isinstance(state, str): + return state + if isinstance(state, dict) and len(state) == 1: + only_key = next(iter(state.keys())) + if isinstance(only_key, str): + return only_key + return None + + +def _is_call_state(event: dict, state_name: str) -> bool: + if event.get("type") != "call_state": + return False + return _call_state_name(event.get("state")) == state_name + + +def _statistics_since(actor: CliProcess, start_index: int) -> list[dict]: + return [ + message + for message in actor.stdout_lines()[start_index:] + if message.get("kind") == "event" and message.get("type") == "statistics" + ] + + +def _max_stat_value(stats: list[dict], field: str) -> float: + values = [ + value + for stat in stats + if isinstance((value := stat.get(field)), (int, float)) + ] + return float(max(values, default=0.0)) + + +def _longest_full_loss_run(stats: list[dict]) -> int: + longest = 0 + current = 0 + for stat in stats: + loss = stat.get("loss") + if isinstance(loss, int) and loss >= FULL_LOSS_SAMPLES_PER_STAT: + current += 1 + longest = max(longest, current) + else: + current = 0 + return longest + + +def _has_sustained_full_loss_then_recovery(stats: list[dict]) -> bool: + current = 0 + for stat in stats: + loss = stat.get("loss") + if isinstance(loss, int) and loss >= FULL_LOSS_SAMPLES_PER_STAT: + current += 1 + continue + if current >= MAX_CONSECUTIVE_FULL_LOSS_STATS: + return True + current = 0 + return False + + +async def _accept_next_call(callee: CliProcess, contact_id: str) -> None: + prompt = await callee.expect_event( + lambda event: event.get("type") == "accept_call_prompt" + and event.get("contact_id") == contact_id, + timeout=20.0, + ) + request_id = prompt.get("request_id") + assert isinstance(request_id, str), f"accept_call_prompt missing request_id: {prompt}" + response = await callee.send( + {"cmd": "accept_call", "args": {"request_id": request_id, "accept": True}} + ) + assert response.get("ok") is True, f"accept_call failed: {response}" + + +async def _start_call_and_wait_connected( + caller: CliProcess, + callee: CliProcess, + *, + caller_contact_id: str, + callee_contact_id: str, +) -> None: + response = await caller.send( + {"cmd": "start_call", "args": {"contact_id": caller_contact_id}} + ) + assert response.get("ok") is True, f"start_call failed: {response}" + await _accept_next_call(callee, callee_contact_id) + await asyncio.gather( + caller.expect_event(lambda event: _is_call_state(event, "Connected"), timeout=20.0), + callee.expect_event(lambda event: _is_call_state(event, "Connected"), timeout=20.0), + ) + + +async def _collect_statistics_window( + actor: CliProcess, + *, + duration: float, +) -> list[dict]: + start_index = len(actor.stdout_lines()) + await asyncio.sleep(duration) + return _statistics_since(actor, start_index) + + +def _assert_no_sustained_full_loss(actor_name: str, stats: list[dict]) -> None: + longest_run = _longest_full_loss_run(stats) + max_loss = _max_stat_value(stats, "loss") + max_download = _max_stat_value(stats, "download_bandwidth") + + assert not _has_sustained_full_loss_then_recovery(stats), ( + f"{actor_name} recorded issue #44 sustained full-loss then recovery; " + f"longest_run={longest_run}, " + f"max_loss={max_loss}, max_download_bandwidth={max_download}, stats={stats}" + ) + + async def _room_cli_group_fixture( *, profile: NetworkProfile, @@ -929,6 +1053,59 @@ def _call_states(messages: list[dict]) -> list[str]: ), f"alice did not receive call_state CallEnded; observed {alice_call_states}" +@pytest.mark.asyncio +@pytest.mark.parametrize( + "profile", [REPEATED_CALL_PROFILE], ids=lambda profile: profile.name +) +async def test_call_repeated_without_restart_keeps_remote_audio( + topology: TopologyManager, + cli_pair: dict[str, CliProcess], + profile: NetworkProfile, +) -> None: + _ = topology, profile + await _run_scenario("call_repeated_without_restart.yaml", cli_pair) + + alice = cli_pair["alice"] + bob = cli_pair["bob"] + + await _start_call_and_wait_connected( + alice, + bob, + caller_contact_id="bob", + callee_contact_id="alice", + ) + + first_call_bob_stats = await _collect_statistics_window(bob, duration=3.0) + _assert_no_sustained_full_loss("bob during first call", first_call_bob_stats) + + # Let the audio sequence climb so the bounded datagram queue holds enough + # stale frames to keep bob fully dark past the 20-stat / 2 s threshold + # during the second call's handshake. + await asyncio.sleep(5.0) + + end_response = await bob.send({"cmd": "end_call", "args": {}}) + assert end_response.get("ok") is True, f"bob end_call failed: {end_response}" + + # bob must start a new call soon after the previous ends + # sleep serves this purpose okay + await asyncio.sleep(0.5) + + await _start_call_and_wait_connected( + alice, + bob, + caller_contact_id="bob", + callee_contact_id="alice", + ) + + second_call_bob_stats = await _collect_statistics_window(bob, duration=4.0) + _assert_no_sustained_full_loss("bob during second call", second_call_bob_stats) + max_second_call_output = _max_stat_value(second_call_bob_stats, "output_level") + assert max_second_call_output > 0.0, ( + "bob did not observe nonzero output_level during second call; " + f"max_output_level={max_second_call_output}, stats={second_call_bob_stats}" + ) + + @pytest.mark.asyncio @pytest.mark.parametrize( "profile",