edison/chanora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(voice): harden Android audio and channel joins

Browse Source
This commit is contained in:
Edison Jwa
2026-05-19 01:58:07 +09:00
parent 29a553d4e1
commit 8c253f1d4d
23 changed files with 2948 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/chanora_core/src/lib.rs
+338 -28
View File
@@ -45,6 +45,11 @@ use tokio::sync::{broadcast, oneshot, watch, Mutex};
use tokio::task::JoinHandle;
use tracing::{info, warn};
use chanora_state::channel_join::{
self, AuthoritativeSource, ChannelId as JoinChannelId, ChannelJoinEvent, ChannelJoinState,
ConnectionEpoch, JoinFailureKind,
};
pub mod ptt;
pub use chanora_audio::{
@@ -171,6 +176,20 @@ pub enum SessionEvent {
mute: bool,
/// Current release-tail in milliseconds (0..=500).
release_tail_ms: u32,
/// Last confirmed authoritative channel id from the
/// `channel_join` reducer projection.
current_channel_id: Option<u64>,
/// Non-authoritative pending target channel id from the
/// reducer projection.
pending_target_channel_id: Option<u64>,
/// Whether the reducer currently allows a new join intent.
can_join: bool,
/// Whether the reducer currently allows leave intent.
can_leave: bool,
/// Join projection synchronization state.
join_sync_state: VoiceJoinSyncState,
/// Last stable sanitized join error code, if any.
join_error_code: Option<VoiceJoinErrorCode>,
},
/// iOS audio-session interruption state (SDD-101). Emitted when
/// interruption begins and when it ends (with the platform hint
@@ -183,6 +202,44 @@ pub enum SessionEvent {
},
}
/// Bridge-safe mirror of channel-join projection sync state.
#[derive(Debug, Clone, Copy)]
pub enum VoiceJoinSyncState {
/// Reducer is ready to accept channel actions.
Ready,
/// Reducer is synchronizing against an initial snapshot.
SynchronizingInitialSnapshot,
/// Reducer is synchronizing after reconnect.
SynchronizingReconnect,
}
/// Bridge-safe mirror of stable channel-join error codes.
#[derive(Debug, Clone, Copy)]
pub enum VoiceJoinErrorCode {
/// Duplicate same-target join intent was coalesced.
DuplicateSameTargetCoalesced,
/// A different target was requested while one is already pending.
JoinAlreadyPendingDifferentTarget,
/// Join denied by server policy/permission.
JoinDenied,
/// Join failed due to protocol-level error.
JoinProtocolFailure,
/// Join failed due to transport/network error.
JoinNetworkFailure,
/// Join timed out awaiting confirmation.
JoinTimeout,
/// Pending join was superseded by user leave.
JoinSupersededByLeave,
/// Stale join outcome was ignored.
JoinStaleOutcomeIgnored,
/// Authoritative membership reconciled to different channel.
JoinReconciledDifferentChannel,
/// Join command was rejected before send acceptance.
JoinCommandRejectedBeforeSend,
/// Join intent rejected while reducer synchronizing.
JoinCannotStartWhileSynchronizing,
}
/// Coarse OS-reported network state. Populated by the Flutter side
/// via `connectivity_plus`; on platforms where no signal is wired
/// we stay at `Unknown` forever and the supervisor falls back to
@@ -235,6 +292,7 @@ struct ConnectedState {
/// Audio supervision state. Wrapped in Arc<Mutex<_>> so the
/// supervisor and the public API both see updates.
sup_inner: Arc<Mutex<SupervisorInner>>,
join_state: ChannelJoinState,
}
/// The top-level Chanora session. Owns at most one active server
@@ -281,6 +339,7 @@ pub struct ChanoraSession {
/// from this value. Also persisted to the identity store so
/// the binding survives app restarts.
pending_binding: Arc<Mutex<Option<PttBinding>>>,
next_connection_epoch: Arc<Mutex<u64>>,
}
impl ChanoraSession {
@@ -308,9 +367,17 @@ impl ChanoraSession {
release_tail,
pending_binding: Arc::new(Mutex::new(None)),
ptt_watchdog: Arc::new(Mutex::new(None)),
next_connection_epoch: Arc::new(Mutex::new(1)),
}
}
async fn allocate_connection_epoch(&self) -> ConnectionEpoch {
let mut guard = self.next_connection_epoch.lock().await;
let epoch = *guard;
*guard = guard.saturating_add(1);
ConnectionEpoch(epoch)
}
/// Wire a directory-backed identity store. Called by the bridge
/// during `bridge_init` once Flutter has resolved the platform
/// app-private storage directory. Subsequent [`Self::connect`]
@@ -474,6 +541,17 @@ impl ChanoraSession {
let client = chanora_protocol::ProtocolClient::connect(cfg.clone()).await?;
let snap = client.snapshot().await?;
let epoch = self.allocate_connection_epoch().await;
let mut join_state = ChannelJoinState::new(epoch);
let initial_channel = self.find_own_in(&snap).await.map(|(_, channel)| JoinChannelId(channel));
let _ = channel_join::reduce(
&mut join_state,
ChannelJoinEvent::SnapshotReady {
current_channel: initial_channel,
epoch,
now: std::time::Instant::now(),
},
);
// Set up the supervisor.
let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
@@ -499,6 +577,7 @@ impl ChanoraSession {
self.voice_selector.clone(),
self.pending_binding.clone(),
self.release_tail.clone(),
self.next_connection_epoch.clone(),
));
let _ = self.events_tx.send(SessionEvent::Connected {
@@ -513,6 +592,7 @@ impl ChanoraSession {
supervisor: Some(supervisor),
cfg,
sup_inner,
join_state,
});
// TS3 servers auto-place a newly-connected client into the
@@ -536,7 +616,15 @@ impl ChanoraSession {
drop(guard);
if let Some((_my_id, _channel_id)) = self.find_own_in(&snap).await {
self.voice_selector.set_in_channel(true);
self.emit_voice_state(true).await;
let projection = {
let guard = self.inner.lock().await;
guard
.as_ref()
.map(|state| channel_join::project(&state.join_state))
};
if let Some(projection) = projection {
self.emit_voice_state(projection).await;
}
// Bring the audio engine up so the user can immediately
// hear other speakers + transmit on PTT. Tolerates
// failure the same way voice_join does: server-side
@@ -560,9 +648,23 @@ impl ChanoraSession {
/// Return a fresh snapshot of the current server state.
pub async fn snapshot(&self) -> Result<ServerSnapshot, CoreError> {
let guard = self.inner.lock().await;
let state = guard.as_ref().ok_or(CoreError::NotConnected)?;
Ok(state.protocol.snapshot().await?)
let mut guard = self.inner.lock().await;
let state = guard.as_mut().ok_or(CoreError::NotConnected)?;
let snap = state.protocol.snapshot().await?;
let current_channel = self
.find_own_in(&snap)
.await
.map(|(_, channel)| JoinChannelId(channel));
let epoch = state.join_state.authoritative.confirmed_epoch;
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::SnapshotReady {
current_channel,
epoch,
now: std::time::Instant::now(),
},
);
Ok(snap)
}
/// True if a connection is currently active.
@@ -1002,13 +1104,59 @@ impl ChanoraSession {
channel_id: u64,
password: Option<String>,
) -> Result<(), CoreError> {
let mut guard = self.inner.lock().await;
let state = guard.as_mut().ok_or(CoreError::NotConnected)?;
let join_start = std::time::Instant::now();
let requested = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::UserJoinRequested {
target_channel: JoinChannelId(channel_id),
now: join_start,
},
);
match requested.status {
channel_join::JoinReduceStatus::Rejected(
channel_join::JoinIntentRejected::JoinAlreadyPendingDifferentTarget,
) => {
return Err(CoreError::Protocol(chanora_protocol::ProtocolError::ServerRejected {
code: 0x7001,
message: "join already pending for a different target".to_string(),
}));
}
channel_join::JoinReduceStatus::Rejected(
channel_join::JoinIntentRejected::CannotJoinWhileSynchronizing,
) => {
return Err(CoreError::Protocol(chanora_protocol::ProtocolError::ServerRejected {
code: 0x7002,
message: "join cannot start while synchronizing".to_string(),
}));
}
channel_join::JoinReduceStatus::CoalescedSameTarget => return Ok(()),
_ => {}
}
let generation = requested
.projection
.pending_generation
.ok_or(CoreError::Invariant("missing pending generation after join request"))?;
let pending_key = state
.join_state
.pending
.filter(|pending| pending.generation == generation)
.map(|pending| channel_join::JoinOutcomeKey {
connection_epoch: pending.connection_epoch,
generation: pending.generation,
request_id: pending.request_id,
})
.ok_or(CoreError::Invariant("missing pending key after join request"))?;
// 1. Send the move command. With send_with_result the
// adapter now correlates against the server's typed
// error reply, so on rejection (no permission, wrong
// password, channel full) we get a concrete
// `ProtocolError::ServerRejected` and don't need the
// snapshot polling at all.
if let Err(e) = self.move_to_channel(channel_id, password).await {
if let Err(e) = state.protocol.move_to_channel(channel_id, password).await {
// TS3 error 0x0302 = `channel_already_in`: we're already
// in the target channel, so this is a no-op success.
// Rolling `in_channel` back to false would break PTT
@@ -1017,10 +1165,10 @@ impl ChanoraSession {
// (SDD-094, SAD-081, SRS-204)
if matches!(
&e,
CoreError::Protocol(chanora_protocol::ProtocolError::ServerRejected {
chanora_protocol::ProtocolError::ServerRejected {
code: 0x0302,
..
})
}
) {
info!(
target: "chanora_core",
@@ -1029,14 +1177,27 @@ impl ChanoraSession {
);
// Fall through to audio-start + snapshot confirmation below.
} else {
// Genuine move failures (wrong password, no
// permission, channel full, etc.) must roll back
// local in-channel state.
self.voice_selector.set_in_channel(false);
self.emit_voice_state(false).await;
return Err(e);
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::JoinCommandRejectedBeforeSend {
generation,
error: JoinFailureKind::Denied,
},
);
let projection = channel_join::project(&state.join_state);
let still_in_channel = projection.current_channel.is_some();
self.voice_selector.set_in_channel(still_in_channel);
self.emit_voice_state(projection).await;
return Err(CoreError::Protocol(e));
}
}
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::ProtocolJoinSucceeded {
key: pending_key,
},
);
drop(guard);
// 2. Bring the audio engine up. Tolerate failure: the
// server-side channel move has ALREADY succeeded (step
// 1), so the user is in the channel from every other
@@ -1093,8 +1254,20 @@ impl ChanoraSession {
tokio::time::sleep(std::time::Duration::from_millis(80)).await;
};
if !confirmed {
self.voice_selector.set_in_channel(false);
self.emit_voice_state(false).await;
let mut guard = self.inner.lock().await;
if let Some(state) = guard.as_mut() {
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::JoinTimeout {
key: pending_key,
now: std::time::Instant::now(),
},
);
let projection = channel_join::project(&state.join_state);
let still_in_channel = projection.current_channel.is_some();
self.voice_selector.set_in_channel(still_in_channel);
self.emit_voice_state(projection).await;
}
return Err(CoreError::Protocol(
chanora_protocol::ProtocolError::ServerRejected {
// Use a sentinel "unknown" code (the canonical
@@ -1105,8 +1278,28 @@ impl ChanoraSession {
},
));
}
let mut guard = self.inner.lock().await;
if let Some(state) = guard.as_mut() {
let epoch = state.join_state.authoritative.confirmed_epoch;
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::AuthoritativeSelfMove {
channel: Some(JoinChannelId(channel_id)),
epoch,
source: AuthoritativeSource::Snapshot,
},
);
}
self.voice_selector.set_in_channel(true);
self.emit_voice_state(true).await;
let projection = {
let guard = self.inner.lock().await;
guard
.as_ref()
.map(|state| channel_join::project(&state.join_state))
};
if let Some(projection) = projection {
self.emit_voice_state(projection).await;
}
Ok(())
}
@@ -1124,9 +1317,21 @@ impl ChanoraSession {
/// to false), tears down the audio engine, and emits
/// [`SessionEvent::VoiceState`].
pub async fn voice_leave(&self) -> Result<(), CoreError> {
self.voice_selector.set_in_channel(false);
let mut guard = self.inner.lock().await;
if let Some(state) = guard.as_mut() {
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::UserLeaveRequested {
now: std::time::Instant::now(),
},
);
let projection = channel_join::project(&state.join_state);
self.voice_selector
.set_in_channel(projection.current_channel.is_some());
self.emit_voice_state(projection).await;
}
drop(guard);
self.shutdown_audio_if_idle().await;
self.emit_voice_state(false).await;
Ok(())
}
@@ -1145,8 +1350,15 @@ impl ChanoraSession {
);
}
}
let in_channel = self.voice_selector.in_channel();
self.emit_voice_state(in_channel).await;
let projection = {
let guard = self.inner.lock().await;
guard
.as_ref()
.map(|state| channel_join::project(&state.join_state))
};
if let Some(projection) = projection {
self.emit_voice_state(projection).await;
}
Ok(())
}
@@ -1161,8 +1373,15 @@ impl ChanoraSession {
/// channel or PTT state.
pub async fn set_hard_mute(&self, muted: bool) -> Result<(), CoreError> {
self.voice_selector.set_hard_mute(muted);
let in_channel = self.voice_selector.in_channel();
self.emit_voice_state(in_channel).await;
let projection = {
let guard = self.inner.lock().await;
guard
.as_ref()
.map(|state| channel_join::project(&state.join_state))
};
if let Some(projection) = projection {
self.emit_voice_state(projection).await;
}
Ok(())
}
@@ -1185,8 +1404,15 @@ impl ChanoraSession {
);
}
}
let in_channel = self.voice_selector.in_channel();
self.emit_voice_state(in_channel).await;
let projection = {
let guard = self.inner.lock().await;
guard
.as_ref()
.map(|state| channel_join::project(&state.join_state))
};
if let Some(projection) = projection {
self.emit_voice_state(projection).await;
}
Ok(())
}
@@ -1209,12 +1435,19 @@ impl ChanoraSession {
self.release_tail.clone()
}
async fn emit_voice_state(&self, in_channel: bool) {
async fn emit_voice_state(&self, projection: channel_join::ChannelJoinProjection) {
let in_channel = projection.current_channel.is_some();
let _ = self.events_tx.send(SessionEvent::VoiceState {
in_channel,
transmit_mode: self.voice_selector.mode().as_u8(),
mute: self.voice_selector.hard_mute(),
release_tail_ms: self.release_tail.tail_ms(),
current_channel_id: projection.current_channel.map(|id| id.0),
pending_target_channel_id: projection.pending_target.map(|id| id.0),
can_join: projection.can_join,
can_leave: projection.can_leave,
join_sync_state: map_join_sync_state(projection.sync_state),
join_error_code: projection.last_join_error.map(map_join_error_code),
});
}
@@ -1289,6 +1522,7 @@ async fn supervisor_loop(
voice_selector: Arc<TransmitModeSelector>,
pending_binding: Arc<Mutex<Option<PttBinding>>>,
release_tail: Arc<ReleaseTailTimer>,
next_connection_epoch: Arc<Mutex<u64>>,
) {
let mut lost_rx = initial_lost_rx;
let mut probe = initial_probe;
@@ -1521,10 +1755,23 @@ async fn supervisor_loop(
info!(target: "chanora_core", attempt, "reconnect: dialling");
match chanora_protocol::ProtocolClient::connect(cfg.clone()).await {
Ok(new_client) => {
let new_epoch = {
let mut guard = next_connection_epoch.lock().await;
let epoch = *guard;
*guard = guard.saturating_add(1);
ConnectionEpoch(epoch)
};
// Successful reconnect. Snapshot for the event.
let snap_name = match new_client.snapshot().await {
Ok(s) => s.server_name,
Err(_) => String::new(),
let (snap_name, snap_current_channel) = match new_client.snapshot().await {
Ok(s) => {
let current_channel = s
.clients
.iter()
.find(|c| c.id.0 == s.own_client_id)
.map(|c| JoinChannelId(c.channel.0));
(s.server_name, current_channel)
}
Err(_) => (String::new(), None),
};
let new_lost_rx = match new_client.take_loss_notifier() {
Some(rx) => rx,
@@ -1555,6 +1802,25 @@ async fn supervisor_loop(
let old = std::mem::replace(&mut state.protocol, new_client);
drop(old);
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::ReconnectStarted {
new_epoch,
now: std::time::Instant::now(),
},
);
let _ = channel_join::reduce(
&mut state.join_state,
ChannelJoinEvent::SnapshotReady {
current_channel: snap_current_channel,
epoch: new_epoch,
now: std::time::Instant::now(),
},
);
voice_selector.set_in_channel(
state.join_state.authoritative.current_channel.is_some(),
);
let sup = sup_inner.lock().await;
sup.audio_running && sup.audio_cfg.is_some()
};
@@ -1693,6 +1959,50 @@ fn snapshot_signature(snap: &ServerSnapshot) -> u64 {
h.finish()
}
fn map_join_sync_state(sync: channel_join::ChannelJoinSyncState) -> VoiceJoinSyncState {
match sync {
channel_join::ChannelJoinSyncState::Ready => VoiceJoinSyncState::Ready,
channel_join::ChannelJoinSyncState::Synchronizing {
reason: channel_join::SyncReason::InitialSnapshot,
..
} => VoiceJoinSyncState::SynchronizingInitialSnapshot,
channel_join::ChannelJoinSyncState::Synchronizing {
reason: channel_join::SyncReason::Reconnect,
..
} => VoiceJoinSyncState::SynchronizingReconnect,
}
}
fn map_join_error_code(code: channel_join::JoinErrorCode) -> VoiceJoinErrorCode {
match code {
channel_join::JoinErrorCode::DuplicateSameTargetCoalesced => {
VoiceJoinErrorCode::DuplicateSameTargetCoalesced
}
channel_join::JoinErrorCode::JoinAlreadyPendingDifferentTarget => {
VoiceJoinErrorCode::JoinAlreadyPendingDifferentTarget
}
channel_join::JoinErrorCode::JoinDenied => VoiceJoinErrorCode::JoinDenied,
channel_join::JoinErrorCode::JoinProtocolFailure => VoiceJoinErrorCode::JoinProtocolFailure,
channel_join::JoinErrorCode::JoinNetworkFailure => VoiceJoinErrorCode::JoinNetworkFailure,
channel_join::JoinErrorCode::JoinTimeout => VoiceJoinErrorCode::JoinTimeout,
channel_join::JoinErrorCode::JoinSupersededByLeave => {
VoiceJoinErrorCode::JoinSupersededByLeave
}
channel_join::JoinErrorCode::JoinStaleOutcomeIgnored => {
VoiceJoinErrorCode::JoinStaleOutcomeIgnored
}
channel_join::JoinErrorCode::JoinReconciledDifferentChannel => {
VoiceJoinErrorCode::JoinReconciledDifferentChannel
}
channel_join::JoinErrorCode::JoinCommandRejectedBeforeSend => {
VoiceJoinErrorCode::JoinCommandRejectedBeforeSend
}
channel_join::JoinErrorCode::JoinCannotStartWhileSynchronizing => {
VoiceJoinErrorCode::JoinCannotStartWhileSynchronizing
}
}
}
#[cfg(test)]
mod tests {
use super::*;