pub mod constants {
pub const VOICE_GATEWAY_VERSION: u8 = 8;
pub const DEFAULT_SAMPLE_RATE: u32 = 48_000;
pub const DAVE_INITIAL_VERSION: u16 = 1;
pub const DEFAULT_VOICE_MODE: &str = "xsalsa20_poly1305";
pub const MAX_RECONNECT_ATTEMPTS: u32 = 15;
pub const BACKOFF_BASE_MS: u64 = 1_000;
pub const RECONNECT_DELAY_FRESH_MS: u64 = 3000;
pub const UDP_KEEPALIVE_GAP_MS: u64 = 5000;
pub const WRITE_TASK_SHUTDOWN_MS: u64 = 500;
pub const RTP_VERSION_BYTE: u8 = 0x80;
pub const RTP_OPUS_PAYLOAD_TYPE: u8 = 0x78;
pub const RTP_TIMESTAMP_STEP: u32 = 960;
pub const FRAME_DURATION_MS: u64 = 20;
pub const PCM_FRAME_SAMPLES: usize = 960;
pub const MAX_OPUS_FRAME_SIZE: usize = 4000;
pub const SILENCE_FRAME: [u8; 3] = [0xf8, 0xff, 0xfe];
pub const MAX_SILENCE_FRAMES: u32 = 5;
pub const UDP_PACKET_BUF_CAPACITY: usize = 1500;
pub const DISCOVERY_PACKET_SIZE: usize = 74;
pub const IP_DISCOVERY_TIMEOUT_SECS: u64 = 2;
pub const IP_DISCOVERY_RETRIES: u32 = 10;
pub const IP_DISCOVERY_RETRY_INTERVAL_MS: u64 = 1000;
pub const OP_HEARTBEAT: u8 = 3;
pub const MAX_PENDING_PROPOSALS: usize = 64;
}
pub mod protocol {
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct GatewayPayload {
pub op: u8,
#[serde(skip_serializing_if = "Option::is_none")]
pub seq: Option<u32>,
pub d: Value,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum OpCode {
Identify = 0,
SelectProtocol = 1,
Ready = 2,
Heartbeat = 3,
SessionDescription = 4,
Speaking = 5,
HeartbeatAck = 6,
Resume = 7,
Hello = 8,
Resumed = 9,
ClientConnect = 11,
Video = 12,
ClientDisconnect = 13,
Codecs = 14,
MediaSinkWants = 15,
VoiceBackendVersion = 16,
UserFlags = 18,
VoicePlatform = 20,
Unknown = 255,
}
impl From<u8> for OpCode {
fn from(op: u8) -> Self {
match op {
0 => Self::Identify,
1 => Self::SelectProtocol,
2 => Self::Ready,
3 => Self::Heartbeat,
4 => Self::SessionDescription,
5 => Self::Speaking,
6 => Self::HeartbeatAck,
7 => Self::Resume,
8 => Self::Hello,
9 => Self::Resumed,
11 => Self::ClientConnect,
12 => Self::Video,
13 => Self::ClientDisconnect,
14 => Self::Codecs,
15 => Self::MediaSinkWants,
16 => Self::VoiceBackendVersion,
18 => Self::UserFlags,
20 => Self::VoicePlatform,
_ => Self::Unknown,
}
}
}
pub mod builders {
use super::*;
use serde_json::json;
pub fn identify(
guild_id: String,
user_id: String,
session_id: String,
token: String,
) -> GatewayPayload {
GatewayPayload {
op: OpCode::Identify as u8,
seq: None,
d: json!({
"server_id": guild_id,
"user_id": user_id,
"session_id": session_id,
"token": token,
"video": true,
}),
}
}
pub fn resume(
guild_id: String,
session_id: String,
token: String,
seq_ack: i64,
) -> GatewayPayload {
let seq_ack = seq_ack.max(0);
GatewayPayload {
op: OpCode::Resume as u8,
seq: None,
d: json!({
"server_id": guild_id,
"session_id": session_id,
"token": token,
"video": true,
"seq_ack": seq_ack,
}),
}
}
}
}
pub mod session {
pub mod types {
use serde::{Deserialize, Serialize};
use thiserror::Error;
#[derive(Error, Debug)]
pub enum GatewayError {
#[error("WebSocket error: {0}")]
WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
#[error("Discovery failed: {0}")]
Discovery(String),
#[error("Encoding error: {0}")]
Encoding(String),
#[error("Encryption error: {0}")]
Encryption(String),
#[error("Internal error: {0}")]
Internal(String),
#[error("Other error: {0}")]
Other(#[from] Box<dyn std::error::Error + Send + Sync>),
}
pub fn map_boxed_err<E: std::fmt::Display>(e: E) -> crate::common::types::AnyError {
Box::new(std::io::Error::other(e.to_string()))
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionOutcome {
Reconnect,
Identify,
Shutdown,
}
#[derive(Default, Debug)]
pub struct PersistentSessionState {
pub ssrc: u32,
pub udp_addr: Option<std::net::SocketAddr>,
pub session_key: Option<[u8; 32]>,
pub rtp_state: Option<crate::gateway::udp_link::RtpState>,
pub selected_mode: Option<String>,
}
}
pub mod protocol {
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct GatewayPayload {
pub op: u8,
#[serde(skip_serializing_if = "Option::is_none")]
pub seq: Option<u32>,
pub d: Value,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum OpCode {
Identify = 0,
SelectProtocol = 1,
Ready = 2,
Heartbeat = 3,
SessionDescription = 4,
Speaking = 5,
HeartbeatAck = 6,
Resume = 7,
Hello = 8,
Resumed = 9,
ClientConnect = 11,
Video = 12,
ClientDisconnect = 13,
Codecs = 14,
MediaSinkWants = 15,
VoiceBackendVersion = 16,
UserFlags = 18,
VoicePlatform = 20,
DavePrepareTransition = 21,
DaveExecuteTransition = 22,
DaveTransitionReady = 23,
DavePrepareEpoch = 24,
MlsExternalSender = 25,
MlsProposals = 27,
MlsAnnounceCommitTransition = 29,
MlsWelcome = 30,
MlsInvalidCommitWelcome = 31,
NoRoute = 32,
Unknown = 255,
}
impl From<u8> for OpCode {
fn from(op: u8) -> Self {
match op {
0 => Self::Identify,
1 => Self::SelectProtocol,
2 => Self::Ready,
3 => Self::Heartbeat,
4 => Self::SessionDescription,
5 => Self::Speaking,
6 => Self::HeartbeatAck,
7 => Self::Resume,
8 => Self::Hello,
9 => Self::Resumed,
11 => Self::ClientConnect,
12 => Self::Video,
13 => Self::ClientDisconnect,
14 => Self::Codecs,
15 => Self::MediaSinkWants,
16 => Self::VoiceBackendVersion,
18 => Self::UserFlags,
20 => Self::VoicePlatform,
21 => Self::DavePrepareTransition,
22 => Self::DaveExecuteTransition,
23 => Self::DaveTransitionReady,
24 => Self::DavePrepareEpoch,
25 => Self::MlsExternalSender,
27 => Self::MlsProposals,
29 => Self::MlsAnnounceCommitTransition,
30 => Self::MlsWelcome,
31 => Self::MlsInvalidCommitWelcome,
32 => Self::NoRoute,
_ => Self::Unknown,
}
}
}
pub mod builders {
use super::*;
use serde_json::json;
pub fn identify(
guild_id: String,
user_id: String,
session_id: String,
token: String,
dave_version: u16,
) -> GatewayPayload {
GatewayPayload {
op: OpCode::Identify as u8,
seq: None,
d: json!({
"server_id": guild_id,
"user_id": user_id,
"session_id": session_id,
"token": token,
"video": true,
"max_dave_protocol_version": dave_version,
}),
}
}
pub fn resume(
guild_id: String,
session_id: String,
token: String,
seq_ack: i64,
) -> GatewayPayload {
let seq_ack = seq_ack.max(0);
GatewayPayload {
op: OpCode::Resume as u8,
seq: None,
d: json!({
"server_id": guild_id,
"session_id": session_id,
"token": token,
"video": true,
"seq_ack": seq_ack,
}),
}
}
}
}
pub mod backoff {
use crate::gateway::constants::{BACKOFF_BASE_MS, MAX_RECONNECT_ATTEMPTS};
use std::time::Duration;
#[derive(Debug, Clone, Default)]
pub struct Backoff {
attempt: u32,
}
impl Backoff {
pub const fn new() -> Self {
Self { attempt: 0 }
}
pub fn next_delay(&mut self) -> Duration {
let exponent = self.attempt.min(3);
let ms = BACKOFF_BASE_MS * 2u64.pow(exponent);
self.attempt += 1;
Duration::from_millis(ms)
}
#[inline]
pub const fn is_exhausted(&self) -> bool {
self.attempt >= MAX_RECONNECT_ATTEMPTS
}
#[inline]
pub fn reset(&mut self) {
self.attempt = 0;
}
#[inline]
pub const fn attempt(&self) -> u32 {
self.attempt
}
}
}
pub mod policy {
use super::types::SessionOutcome;
pub struct FailurePolicy {
max_retries: u32,
}
impl FailurePolicy {
pub const fn new(max_retries: u32) -> Self {
Self { max_retries }
}
pub const fn is_retryable(&self, code: u16, attempt: u32) -> bool {
if attempt >= self.max_retries {
return false;
}
matches!(
code,
1000 | 1001
| 1006
| 4000
| 4001
| 4002
| 4003
| 4005
| 4006
| 4009
| 4012
| 4015
| 4016
| 4020
| 4900
)
}
pub fn classify(&self, code: u16) -> SessionOutcome {
match code {
4004 | 4011 | 4014 | 4021 | 4022 => SessionOutcome::Shutdown,
4006 | 4009 => SessionOutcome::Identify,
_ => SessionOutcome::Reconnect,
}
}
}
}
pub mod heartbeat {
use super::protocol::{GatewayPayload, OpCode};
use crate::common::utils::now_ms;
use std::sync::{
Arc,
atomic::{AtomicI64, AtomicU32, AtomicU64, Ordering},
};
use tokio::sync::mpsc::UnboundedSender;
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_util::sync::CancellationToken;
use tracing::warn;
pub struct HeartbeatTracker {
pub last_nonce: Arc<AtomicU64>,
pub sent_at: Arc<AtomicU64>,
pub missed_acks: Arc<AtomicU32>,
}
impl Default for HeartbeatTracker {
fn default() -> Self {
Self {
last_nonce: Arc::new(AtomicU64::new(0)),
sent_at: Arc::new(AtomicU64::new(0)),
missed_acks: Arc::new(AtomicU32::new(0)),
}
}
}
impl HeartbeatTracker {
pub fn new() -> Self {
Self::default()
}
pub fn validate_ack(&self, acked_nonce: u64) -> Option<u64> {
let expected = self.last_nonce.load(Ordering::Relaxed);
if expected != acked_nonce {
warn!("Heartbeat mismatch: sent={expected} got={acked_nonce}");
return None;
}
Some(now_ms().saturating_sub(self.sent_at.load(Ordering::Relaxed)))
}
pub fn spawn(
&self,
tx: UnboundedSender<Message>,
seq_ack: Arc<AtomicI64>,
conn_token: CancellationToken,
interval_ms: u64,
) -> tokio::task::JoinHandle<()> {
let last_nonce = self.last_nonce.clone();
let sent_at = self.sent_at.clone();
let missed_acks = self.missed_acks.clone();
tokio::spawn(async move {
let mut ticker =
tokio::time::interval(tokio::time::Duration::from_millis(interval_ms));
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let missed = missed_acks.fetch_add(1, Ordering::Relaxed);
if missed >= 2 {
warn!("Heartbeat timeout: {missed} missed ACKs.");
conn_token.cancel();
break;
}
let nonce = now_ms();
last_nonce.store(nonce, Ordering::Relaxed);
sent_at.store(nonce, Ordering::Relaxed);
let hb = GatewayPayload {
op: OpCode::Heartbeat as u8,
seq: None,
d: serde_json::json!({
"t": nonce,
"seq_ack": seq_ack.load(Ordering::Relaxed)
}),
};
if let Ok(json) = serde_json::to_string(&hb)
&& tx.send(Message::Text(json.into())).is_err()
{
break;
}
}
})
}
}
}
pub mod voice {
use super::types::GatewayError;
use crate::{
audio::{Mixer, engine::Encoder, filters::FilterChain},
common::types::Shared,
gateway::{
DaveHandler,
constants::{
DISCOVERY_PACKET_SIZE, FRAME_DURATION_MS, IP_DISCOVERY_RETRIES,
IP_DISCOVERY_RETRY_INTERVAL_MS, IP_DISCOVERY_TIMEOUT_SECS, MAX_OPUS_FRAME_SIZE,
MAX_SILENCE_FRAMES, PCM_FRAME_SAMPLES, SILENCE_FRAME, UDP_KEEPALIVE_GAP_MS,
},
udp_link::UDPVoiceTransport,
},
};
use std::{
net::SocketAddr,
sync::{Arc, atomic::Ordering},
time::{Duration, Instant},
};
use tokio::sync::mpsc::UnboundedSender;
use tokio_util::sync::CancellationToken;
use tracing::error;
pub async fn discover_ip(
socket: &tokio::net::UdpSocket,
addr: SocketAddr,
ssrc: u32,
) -> Result<(String, u16), GatewayError> {
let mut packet = [0u8; DISCOVERY_PACKET_SIZE];
packet[0..2].copy_from_slice(&1u16.to_be_bytes());
packet[2..4].copy_from_slice(&70u16.to_be_bytes());
packet[4..8].copy_from_slice(&ssrc.to_be_bytes());
for attempt in 1..=IP_DISCOVERY_RETRIES {
if attempt > 1 {
tokio::time::sleep(Duration::from_millis(IP_DISCOVERY_RETRY_INTERVAL_MS)).await;
}
if let Err(e) = socket.send_to(&packet, addr).await {
if attempt == IP_DISCOVERY_RETRIES {
return Err(GatewayError::Discovery(e.to_string()));
}
continue;
}
let mut client_buf = [0u8; DISCOVERY_PACKET_SIZE];
match tokio::time::timeout(
Duration::from_secs(IP_DISCOVERY_TIMEOUT_SECS),
socket.recv_from(&mut client_buf),
)
.await
{
Ok(Ok((n, peer))) if n >= DISCOVERY_PACKET_SIZE => {
if peer != addr {
continue;
}
let ip = std::str::from_utf8(&client_buf[8..72])
.map_err(|e| GatewayError::Discovery(e.to_string()))?
.trim_end_matches('\0')
.to_owned();
let port = u16::from_be_bytes([client_buf[72], client_buf[73]]);
return Ok((ip, port));
}
_ => {
if attempt == IP_DISCOVERY_RETRIES {
return Err(GatewayError::Discovery("Timed out".into()));
}
}
}
}
Err(GatewayError::Discovery("Exhausted".into()))
}
pub struct SpeakConfig {
pub mixer: Shared<Mixer>,
pub socket: Arc<tokio::net::UdpSocket>,
pub addr: SocketAddr,
pub ssrc: u32,
pub key: [u8; 32],
pub mode: String,
pub dave: Shared<DaveHandler>,
pub filter_chain: Shared<FilterChain>,
pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
pub cancel_token: CancellationToken,
pub speaking_tx: UnboundedSender<bool>,
pub persistent_state: Arc<tokio::sync::Mutex<super::types::PersistentSessionState>>,
}
pub async fn speak_loop(config: SpeakConfig) -> Result<(), GatewayError> {
let rtp_state = { config.persistent_state.lock().await.rtp_state };
let transport = UDPVoiceTransport::new(
config.socket.clone(),
config.addr,
config.ssrc,
config.key,
&config.mode,
rtp_state,
)?;
let mut encoder = Encoder::new().map_err(|e| GatewayError::Encoding(e.to_string()))?;
let mut session = VoiceSession::new(config, transport);
session.run(&mut encoder).await
}
struct VoiceSession {
config: SpeakConfig,
transport: UDPVoiceTransport,
is_speaking: bool,
speaking_holdoff: bool,
last_tx_time: Instant,
active_silence: u32,
}
impl VoiceSession {
fn new(config: SpeakConfig, transport: UDPVoiceTransport) -> Self {
Self {
config,
transport,
is_speaking: false,
speaking_holdoff: false,
last_tx_time: Instant::now(),
active_silence: 0,
}
}
async fn run(&mut self, encoder: &mut Encoder) -> Result<(), GatewayError> {
let mut interval = tokio::time::interval(Duration::from_millis(FRAME_DURATION_MS));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut pcm = vec![0i16; PCM_FRAME_SAMPLES * 2];
let mut opus = vec![0u8; MAX_OPUS_FRAME_SIZE];
let mut ts_pcm = vec![0i16; PCM_FRAME_SAMPLES * 2];
while !self.config.cancel_token.is_cancelled() {
interval.tick().await;
self.tick(encoder, &mut pcm, &mut opus, &mut ts_pcm).await?;
if self
.config
.frames_sent
.load(Ordering::Relaxed)
.is_multiple_of(100)
{
self.config.persistent_state.lock().await.rtp_state =
Some(self.transport.rtp);
}
}
self.config.persistent_state.lock().await.rtp_state = Some(self.transport.rtp);
Ok(())
}
async fn tick(
&mut self,
encoder: &mut Encoder,
pcm: &mut [i16],
opus: &mut [u8],
ts_pcm: &mut [i16],
) -> Result<(), GatewayError> {
macro_rules! try_lock_yield {
($mutex:expr) => {{
let mut guard = None;
for _ in 0..10 {
if let Ok(g) = $mutex.try_lock() {
guard = Some(g);
break;
}
tokio::task::yield_now().await;
}
guard
}};
}
let mut loop_count = 0;
while loop_count < 10 {
loop_count += 1;
let ready_from_ts = {
if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
filters.has_timescale() && filters.fill_frame(ts_pcm)
} else {
false
}
};
if ready_from_ts {
self.set_speaking(true);
self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
if self.speaking_holdoff {
self.speaking_holdoff = false;
self.send_silence().await?;
}
return self.send_pcm(encoder, ts_pcm, opus).await;
}
let mut has_input = false;
let mut opus_data = None;
if let Some(mut mixer) = try_lock_yield!(self.config.mixer) {
if let Some(data) = mixer.take_opus_frame() {
opus_data = Some(data);
} else {
has_input = mixer.mix(pcm);
}
}
if let Some(data) = opus_data {
self.reset_timers();
self.set_speaking(true);
self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
if self.speaking_holdoff {
self.speaking_holdoff = false;
self.send_silence().await?;
}
return self.send_raw(&data).await;
}
if has_input {
self.reset_timers();
self.set_speaking(true);
} else if self.active_silence > 0 {
self.active_silence -= 1;
pcm.fill(0);
self.set_speaking(true);
} else {
self.set_speaking(false);
if self.last_tx_time.elapsed()
>= Duration::from_millis(UDP_KEEPALIVE_GAP_MS)
{
return self.send_silence().await;
}
return Ok(());
}
let has_ts = {
if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
filters.process(pcm);
filters.has_timescale()
} else {
false
}
};
if !has_ts {
if has_input {
self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
} else {
self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
}
if self.speaking_holdoff {
self.speaking_holdoff = false;
self.send_silence().await?;
}
return self.send_pcm(encoder, pcm, opus).await;
}
let filled_on_silence = {
if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
!has_input && filters.fill_frame(ts_pcm)
} else {
false
}
};
if !has_input && !filled_on_silence {
break;
}
}
Ok(())
}
fn set_speaking(&mut self, speaking: bool) {
if speaking != self.is_speaking {
self.is_speaking = speaking;
let _ = self.config.speaking_tx.send(speaking);
if speaking {
self.speaking_holdoff = true;
}
}
}
async fn send_pcm(
&mut self,
encoder: &mut Encoder,
pcm: &[i16],
opus: &mut [u8],
) -> Result<(), GatewayError> {
let size = encoder.encode(pcm, opus).unwrap_or_else(|e| {
error!("Opus encode failed: {e}");
0
});
if size > 0 {
self.send_raw(&opus[..size]).await?;
} else {
self.send_silence().await?;
}
Ok(())
}
async fn send_silence(&mut self) -> Result<(), GatewayError> {
self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
self.send_raw(&SILENCE_FRAME).await
}
async fn send_raw(&mut self, data: &[u8]) -> Result<(), GatewayError> {
let mut dave = self.config.dave.lock().await;
let encrypted = dave
.encrypt_opus(data)
.map_err(|e| GatewayError::Encryption(e.to_string()))?;
drop(dave);
self.transport.transmit_opus(&encrypted).await?;
self.last_tx_time = Instant::now();
Ok(())
}
fn reset_timers(&mut self) {
self.active_silence = MAX_SILENCE_FRAMES;
}
}
}
pub mod handler {
use super::{
VoiceGateway,
backoff::Backoff,
heartbeat::HeartbeatTracker,
protocol::{GatewayPayload, OpCode},
types::{GatewayError, PersistentSessionState, SessionOutcome},
voice::{SpeakConfig, discover_ip, speak_loop},
};
use crate::{
common::types::{Shared, UserId},
gateway::{
DaveHandler,
constants::{DAVE_INITIAL_VERSION, DEFAULT_VOICE_MODE},
},
};
use serde_json::Value;
use std::{
collections::HashSet,
net::SocketAddr,
sync::{
Arc,
atomic::{AtomicI64, Ordering},
},
};
use tokio::sync::mpsc::UnboundedSender;
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, trace, warn};
use uuid::Uuid;
pub struct SessionState<'a> {
gateway: &'a VoiceGateway,
tx: UnboundedSender<Message>,
seq_ack: Arc<AtomicI64>,
ssrc: u32,
udp_addr: Option<SocketAddr>,
selected_mode: String,
connected_users: HashSet<UserId>,
udp_socket: Arc<tokio::net::UdpSocket>,
dave: Shared<DaveHandler>,
heartbeat: HeartbeatTracker,
heartbeat_handle: Option<tokio::task::JoinHandle<()>>,
conn_token: CancellationToken,
speaking_tx: Option<UnboundedSender<bool>>,
session_key: Option<[u8; 32]>,
speak_task: Option<tokio::task::JoinHandle<()>>,
persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
backoff: &'a mut Backoff,
}
impl<'a> SessionState<'a> {
pub async fn new(
gateway: &'a VoiceGateway,
tx: UnboundedSender<Message>,
seq_ack: Arc<AtomicI64>,
conn_token: CancellationToken,
persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
backoff: &'a mut Backoff,
) -> Result<Self, GatewayError> {
let mut socket_guard = gateway.udp_socket.lock().await;
let udp_socket = if let Some(existing) = &*socket_guard {
existing.clone()
} else {
let udp = std::net::UdpSocket::bind("0.0.0.0:0")?;
udp.set_nonblocking(true)?;
let socket = Arc::new(tokio::net::UdpSocket::from_std(udp)?);
*socket_guard = Some(socket.clone());
socket
};
Ok(Self {
gateway,
tx,
seq_ack,
ssrc: 0,
udp_addr: None,
selected_mode: DEFAULT_VOICE_MODE.to_string(),
connected_users: HashSet::from([gateway.user_id]),
udp_socket,
dave: gateway.dave.clone(),
heartbeat: HeartbeatTracker::new(),
heartbeat_handle: None,
conn_token,
speaking_tx: None,
session_key: None,
speak_task: None,
persistent_state,
backoff,
})
}
pub fn set_speaking_tx(&mut self, tx: UnboundedSender<bool>) {
self.speaking_tx = Some(tx);
}
pub fn ssrc(&self) -> u32 {
self.ssrc
}
pub fn tx(&self) -> &UnboundedSender<Message> {
&self.tx
}
pub fn attempt(&self) -> u32 {
self.backoff.attempt()
}
pub fn has_heartbeat(&self) -> bool {
self.heartbeat_handle.is_some()
}
pub async fn handle_text(&mut self, text: String) -> Option<SessionOutcome> {
let payload: GatewayPayload = match serde_json::from_str(&text) {
Ok(p) => p,
Err(e) => {
warn!("[{}] JSON Parse error: {e}", self.gateway.guild_id);
return None;
}
};
if let Some(seq) = payload.seq {
self.seq_ack.store(seq as i64, Ordering::Relaxed);
}
let op = OpCode::from(payload.op);
trace!(
"[{}] RX OP: {:?} (op={})",
self.gateway.guild_id, op, payload.op
);
match op {
OpCode::Hello => self.on_hello(payload.d),
OpCode::Ready => self.on_ready(payload.d).await,
OpCode::SessionDescription => self.on_session_description(payload.d).await,
OpCode::HeartbeatAck => self.on_heartbeat_ack(payload.d),
OpCode::Resumed => self.on_resumed().await,
OpCode::ClientConnect => self.on_user_connect(payload.d).await,
OpCode::ClientDisconnect => self.on_user_disconnect(payload.d).await,
OpCode::VoiceBackendVersion => {
debug!(
"[{}] Voice Backend Version: {:?}",
self.gateway.guild_id, payload.d
);
None
}
OpCode::MediaSinkWants => {
debug!(
"[{}] Media Sink Wants: {:?}",
self.gateway.guild_id, payload.d
);
None
}
OpCode::DavePrepareTransition => {
self.on_dave_prepare_transition(payload.d).await
}
OpCode::DaveExecuteTransition => {
self.on_dave_execute_transition(payload.d).await
}
OpCode::DavePrepareEpoch => self.on_dave_prepare_epoch(payload.d).await,
OpCode::MlsAnnounceCommitTransition => self.on_mls_transition(payload.d).await,
OpCode::MlsInvalidCommitWelcome => {
warn!(
"[{}] DAVE MLS Invalid Commit Welcome received, resetting session",
self.gateway.guild_id
);
self.reset_dave(0).await;
None
}
OpCode::NoRoute => {
warn!(
"[{}] No Route received: {:?}",
self.gateway.guild_id, payload.d
);
None
}
OpCode::Speaking
| OpCode::Video
| OpCode::Codecs
| OpCode::UserFlags
| OpCode::VoicePlatform => None,
_ => None,
}
}
pub async fn handle_binary(&mut self, bin: Vec<u8>) {
if bin.len() < 3 {
return;
}
let seq = u16::from_be_bytes([bin[0], bin[1]]);
let op = bin[2];
let data = &bin[3..];
self.seq_ack.store(seq as i64, Ordering::Relaxed);
let mut dave = self.dave.lock().await;
match op {
25 => {
if let Ok(res) = dave.process_external_sender(data) {
for r in res {
self.send_binary(28, &r);
}
}
}
27 => match dave.process_proposals(data) {
Ok(Some(cw)) => self.send_binary(28, &cw),
Err(e) => {
warn!("[{}] DAVE proposals failed: {e}", self.gateway.guild_id);
self.reset_dave_locked(&mut dave, 0).await;
}
_ => {}
},
29 | 30 => {
let res = if op == 30 {
dave.process_welcome(data)
} else {
dave.process_commit(data)
};
match res {
Ok(tid) if tid != 0 => {
self.send_json(23, serde_json::json!({ "transition_id": tid }))
}
Err(e) => {
let tid = if data.len() >= 2 {
u16::from_be_bytes([data[0], data[1]])
} else {
0
};
warn!(
"[{}] DAVE {} failed (tid {tid}): {e}",
self.gateway.guild_id,
if op == 30 { "welcome" } else { "commit" }
);
self.reset_dave_locked(&mut dave, tid).await;
}
_ => {}
}
}
_ => {}
}
}
fn on_hello(&mut self, d: Value) -> Option<SessionOutcome> {
let interval = d["heartbeat_interval"].as_u64().unwrap_or(30_000);
if let Some(h) = self.heartbeat_handle.take() {
debug!(
"[{}] Restarting heartbeat on HELLO (was already running)",
self.gateway.guild_id
);
h.abort();
}
trace!(
"[{}] Heartbeat interval: {interval}ms",
self.gateway.guild_id
);
self.heartbeat_handle = Some(self.heartbeat.spawn(
self.tx.clone(),
self.seq_ack.clone(),
self.conn_token.clone(),
interval,
));
None
}
async fn on_ready(&mut self, d: Value) -> Option<SessionOutcome> {
let ssrc = d["ssrc"].as_u64();
let ip = d["ip"].as_str();
let port = d["port"].as_u64();
match (ssrc, ip, port) {
(Some(ssrc), Some(ip), Some(port)) if port <= 65535 => {
self.ssrc = ssrc as u32;
let addr_str = format!("{ip}:{port}");
match addr_str.parse::<SocketAddr>() {
Ok(addr) => self.udp_addr = Some(addr),
Err(_) => {
error!(
"[{}] Invalid READY address: {addr_str}",
self.gateway.guild_id
);
return Some(SessionOutcome::Reconnect);
}
}
}
_ => {
error!("[{}] Malformed READY payload", self.gateway.guild_id);
return Some(SessionOutcome::Reconnect);
}
}
if let Some(modes) = d["modes"].as_array() {
let pref = ["aead_aes256_gcm_rtpsize", "xsalsa20_poly1305"];
if let Some(m) = pref
.iter()
.find(|&&p| modes.iter().any(|m| m.as_str() == Some(p)))
{
self.selected_mode = (*m).to_owned();
}
}
debug!(
"[{}] Ready: ssrc={}, mode={}",
self.gateway.guild_id, self.ssrc, self.selected_mode
);
{
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 {
let ver = d["dave_protocol_version"]
.as_u64()
.unwrap_or(DAVE_INITIAL_VERSION as u64)
as u16;
let mut dave = self.dave.lock().await;
if ver == 0 {
dave.reset();
} else {
dave.set_protocol_version(ver);
if let Ok(kp) = dave.setup_session(ver) {
self.send_binary(26, &kp);
}
}
}
let target_addr = match self.udp_addr {
Some(a) => a,
None => return Some(SessionOutcome::Reconnect),
};
match discover_ip(&self.udp_socket, target_addr, self.ssrc).await {
Ok((my_ip, my_port)) => {
self.send_json(OpCode::SelectProtocol as u8, serde_json::json!({
"protocol": "udp",
"rtc_connection_id": Uuid::new_v4().to_string(),
"codecs": [{"name": "opus", "type": "audio", "priority": 1000, "payload_type": 120}],
"data": { "address": my_ip, "port": my_port, "mode": self.selected_mode },
"address": my_ip,
"port": my_port,
"mode": self.selected_mode
}));
self.send_json(
OpCode::Video as u8,
serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
);
self.send_json(
OpCode::Speaking as u8,
serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
);
}
Err(e) => {
error!("[{}] IP discovery failed: {e}", self.gateway.guild_id);
return Some(SessionOutcome::Reconnect);
}
}
self.backoff.reset();
None
}
async fn on_session_description(&mut self, d: Value) -> Option<SessionOutcome> {
let ka = match d["secret_key"].as_array() {
Some(a) if a.len() == 32 => a,
_ => {
error!(
"[{}] Invalid or missing secret_key in VOICE_READY",
self.gateway.guild_id
);
return Some(SessionOutcome::Reconnect);
}
};
let mut key = [0u8; 32];
for (i, v) in ka.iter().enumerate() {
if let Some(val) = v.as_u64()
&& val <= 255
{
key[i] = val as u8;
continue;
}
error!(
"[{}] Invalid secret_key byte at index {i}",
self.gateway.guild_id
);
return Some(SessionOutcome::Reconnect);
}
self.session_key = Some(key);
let addr = match self.udp_addr {
Some(a) => a,
None => return Some(SessionOutcome::Reconnect),
};
{
let mut state = self.persistent_state.lock().await;
state.udp_addr = Some(addr);
state.session_key = Some(key);
state.ssrc = self.ssrc;
state.selected_mode = Some(self.selected_mode.clone());
}
self.start_voice(addr, key).await;
if self.gateway.channel_id.0 > 0 {
let protocol_version = d["dave_protocol_version"]
.as_u64()
.unwrap_or(DAVE_INITIAL_VERSION as u64)
as u16;
let mls_group_id = d["mls_group_id"].as_u64().unwrap_or(0);
let mut dave = self.dave.lock().await;
if protocol_version > 0 {
dave.set_protocol_version(protocol_version);
if let Ok(kp) = dave.setup_session(protocol_version) {
self.send_binary(26, &kp);
}
} else {
dave.reset();
}
debug!(
"DAVE setup context: protocol_version={}, mls_group_id={}",
protocol_version, mls_group_id
);
}
self.backoff.reset();
None
}
async fn on_resumed(&mut self) -> Option<SessionOutcome> {
debug!("[{}] Resumed", self.gateway.guild_id);
let (addr, key, ssrc, mode) = {
let state = self.persistent_state.lock().await;
(
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;
}
if let Some(task) = &self.speak_task
&& task.is_finished()
{
self.speak_task = None;
}
if self.speak_task.is_some() {
debug!(
"[{}] Keeping existing voice loop alive across resume",
self.gateway.guild_id
);
} else {
debug!(
"[{}] Starting voice loop after resume (task was dead or missing)",
self.gateway.guild_id
);
self.start_voice(addr, key).await;
}
}
_ => {
warn!(
"[{}] Resume failed: missing persistent state",
self.gateway.guild_id
);
return Some(SessionOutcome::Identify);
}
}
None
}
fn on_heartbeat_ack(&self, d: Value) -> Option<SessionOutcome> {
let nonce = d["t"].as_u64().unwrap_or(0);
if let Some(rtt) = self.heartbeat.validate_ack(nonce) {
self.gateway.ping.store(rtt as i64, Ordering::Relaxed);
self.heartbeat.missed_acks.store(0, Ordering::Relaxed);
}
None
}
async fn on_user_connect(&mut self, d: Value) -> Option<SessionOutcome> {
if let Some(ids) = d["user_ids"].as_array() {
let mut uids = Vec::new();
for id in ids {
if let Some(uid) = id.as_str().and_then(|s| s.parse::<u64>().ok()) {
self.connected_users.insert(UserId(uid));
uids.push(uid);
}
}
if !uids.is_empty() {
self.dave.lock().await.add_users(&uids);
}
}
None
}
async fn on_user_disconnect(&mut self, d: Value) -> Option<SessionOutcome> {
if let Some(uid) = d["user_id"].as_str().and_then(|s| s.parse::<u64>().ok()) {
self.connected_users.remove(&UserId(uid));
self.dave.lock().await.remove_user(uid);
}
None
}
async fn on_dave_prepare_transition(&mut self, d: Value) -> Option<SessionOutcome> {
let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
debug!(
"[{}] DAVE Prepare Transition: id={}, version={}",
self.gateway.guild_id, tid, ver
);
if self.dave.lock().await.prepare_transition(tid, ver) {
debug!(
"[{}] DAVE Transition Ready (tid={})",
self.gateway.guild_id, tid
);
self.send_json(23, serde_json::json!({ "transition_id": tid }));
}
None
}
async fn on_dave_execute_transition(&mut self, d: Value) -> Option<SessionOutcome> {
let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
debug!(
"[{}] DAVE Execute Transition: id={}",
self.gateway.guild_id, tid
);
self.dave.lock().await.execute_transition(tid);
None
}
async fn on_dave_prepare_epoch(&mut self, d: Value) -> Option<SessionOutcome> {
let epoch = d["epoch"].as_u64().unwrap_or(0);
let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
debug!(
"[{}] DAVE Prepare Epoch: epoch={}, version={}",
self.gateway.guild_id, epoch, ver
);
if let Some(kp) = self.dave.lock().await.prepare_epoch(epoch, ver) {
self.send_binary(26, &kp);
}
None
}
async fn on_mls_transition(&mut self, d: Value) -> Option<SessionOutcome> {
let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
debug!(
"[{}] DAVE MLS Announce Commit Transition: tid={}",
self.gateway.guild_id, tid
);
let ver = d["protocol_version"].as_u64().map(|v| v as u16);
if let Some(v) = ver {
let mut dave = self.dave.lock().await;
if dave.prepare_transition(tid, v) && tid != 0 {
self.send_json(23, serde_json::json!({ "transition_id": tid }));
}
}
None
}
async fn start_voice(&mut self, addr: SocketAddr, key: [u8; 32]) {
if let Some(t) = self.speak_task.take() {
t.abort();
}
let speaking_tx = if let Some(tx) = &self.speaking_tx {
tx.clone()
} else {
error!(
"[{}] speaking_tx is missing, cannot start voice",
self.gateway.guild_id
);
return;
};
let config = SpeakConfig {
mixer: self.gateway.mixer.clone(),
socket: self.udp_socket.clone(),
addr,
ssrc: self.ssrc,
key,
mode: self.selected_mode.clone(),
dave: self.dave.clone(),
filter_chain: self.gateway.filter_chain.clone(),
frames_sent: self.gateway.frames_sent.clone(),
frames_nulled: self.gateway.frames_nulled.clone(),
cancel_token: self.conn_token.clone(),
speaking_tx,
persistent_state: self.persistent_state.clone(),
};
let guild_id = self.gateway.guild_id.clone();
let conn_token = self.conn_token.clone();
self.speak_task = Some(tokio::spawn(async move {
if let Err(e) = speak_loop(config).await {
error!("[{guild_id}] speak_loop failed: {e}");
conn_token.cancel();
}
}));
self.send_json(
OpCode::Video as u8,
serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
);
self.send_json(
OpCode::Speaking as u8,
serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
);
}
async fn reset_dave(&self, tid: u16) {
let mut dave = self.dave.lock().await;
self.reset_dave_locked(&mut dave, tid).await;
}
async fn reset_dave_locked(&self, dave: &mut DaveHandler, tid: u16) {
dave.reset();
self.send_json(31, serde_json::json!({ "transition_id": tid }));
if let Ok(kp) = dave.setup_session(DAVE_INITIAL_VERSION) {
self.send_binary(26, &kp);
}
}
fn send_json(&self, op: u8, d: Value) {
match serde_json::to_string(&GatewayPayload { op, seq: None, d }) {
Ok(json) => {
let _ = self.tx.send(Message::Text(json.into()));
}
Err(e) => {
warn!("[{}] JSON serialization failed: {e}", self.gateway.guild_id);
}
}
}
fn send_binary(&self, op: u8, payload: &[u8]) {
let mut b = vec![op];
b.extend_from_slice(payload);
let _ = self.tx.send(Message::Binary(b.into()));
}
}
impl<'a> Drop for SessionState<'a> {
fn drop(&mut self) {
if let Some(h) = self.heartbeat_handle.take() {
h.abort();
}
if let Some(t) = self.speak_task.take() {
t.abort();
}
}
}
}
use self::{
backoff::Backoff,
policy::FailurePolicy,
types::{GatewayError, PersistentSessionState, SessionOutcome},
};
use crate::{
audio::{Mixer, filters::FilterChain},
common::types::{ChannelId, GuildId, SessionId, Shared, UserId},
gateway::constants::VOICE_GATEWAY_VERSION,
protocol::RustalinkEvent,
};
use futures::{SinkExt, StreamExt};
use std::sync::{
Arc,
atomic::{AtomicI64, Ordering},
};
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
use tokio_tungstenite::tungstenite::protocol::{Message, WebSocketConfig};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
pub struct VoiceGateway {
pub guild_id: GuildId,
pub user_id: UserId,
pub channel_id: ChannelId,
session_id: SessionId,
token: String,
endpoint: String,
pub mixer: Shared<Mixer>,
pub filter_chain: Shared<FilterChain>,
pub ping: Arc<AtomicI64>,
event_tx: Option<UnboundedSender<RustalinkEvent>>,
pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
pub udp_socket: Shared<Option<Arc<tokio::net::UdpSocket>>>,
pub dave: Shared<crate::gateway::DaveHandler>,
outer_token: CancellationToken,
policy: FailurePolicy,
}
pub struct VoiceGatewayConfig {
pub guild_id: GuildId,
pub user_id: UserId,
pub channel_id: ChannelId,
pub session_id: SessionId,
pub token: String,
pub endpoint: String,
pub mixer: Shared<Mixer>,
pub filter_chain: Shared<FilterChain>,
pub ping: Arc<AtomicI64>,
pub event_tx: Option<UnboundedSender<RustalinkEvent>>,
pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
}
impl VoiceGateway {
pub fn new(config: VoiceGatewayConfig) -> Self {
Self {
guild_id: config.guild_id,
user_id: config.user_id,
channel_id: config.channel_id,
session_id: config.session_id,
token: config.token,
endpoint: config.endpoint,
mixer: config.mixer,
filter_chain: config.filter_chain,
ping: config.ping,
event_tx: config.event_tx,
frames_sent: config.frames_sent,
frames_nulled: config.frames_nulled,
udp_socket: Arc::new(tokio::sync::Mutex::new(None)),
dave: Arc::new(tokio::sync::Mutex::new(crate::gateway::DaveHandler::new(
config.user_id,
config.channel_id,
))),
outer_token: CancellationToken::new(),
policy: FailurePolicy::new(3),
}
}
pub async fn run(self) -> Result<(), GatewayError> {
let mut backoff = Backoff::new();
let mut is_resume = false;
let seq_ack = Arc::new(AtomicI64::new(-1));
let persistent_state =
Arc::new(tokio::sync::Mutex::new(PersistentSessionState::default()));
while !self.outer_token.is_cancelled() {
let attempt = backoff.attempt();
match self
.connect(
is_resume,
seq_ack.clone(),
persistent_state.clone(),
&mut backoff,
)
.await
{
Ok(SessionOutcome::Shutdown) => break,
Ok(outcome) => {
if backoff.is_exhausted() {
warn!("[{}] Max attempts reached ({})", self.guild_id, attempt);
break;
}
let delay = backoff.next_delay();
is_resume = matches!(outcome, SessionOutcome::Reconnect);
if !is_resume {
seq_ack.store(-1, Ordering::Relaxed);
*persistent_state.lock().await = PersistentSessionState::default();
*self.udp_socket.lock().await = None;
}
debug!(
"[{}] Retrying ({:?}) in {:?}",
self.guild_id, outcome, delay
);
tokio::time::sleep(delay).await;
}
Err(e) => {
if backoff.is_exhausted() {
error!("[{}] Fatal connection error: {e}", self.guild_id);
break;
}
let delay = backoff.next_delay();
warn!(
"[{}] Connection error: {e}. Retrying in {:?}",
self.guild_id, delay
);
tokio::time::sleep(delay).await;
is_resume = false;
}
}
}
Ok(())
}
async fn connect(
&self,
is_resume: bool,
seq_ack: Arc<AtomicI64>,
persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
backoff: &mut Backoff,
) -> Result<SessionOutcome, GatewayError> {
let endpoint = if self.endpoint.ends_with(":80") {
&self.endpoint[..self.endpoint.len() - 3]
} else {
&self.endpoint
};
let url = format!("wss://{}/?v={}", endpoint, VOICE_GATEWAY_VERSION);
let mut config = WebSocketConfig::default();
config.max_message_size = None;
config.max_frame_size = None;
let (ws_stream, _) =
tokio_tungstenite::connect_async_with_config(&url, Some(config), true).await?;
let (mut write, mut read) = ws_stream.split();
let conn_token = CancellationToken::new();
let write_token = conn_token.clone();
let (ws_tx, mut ws_rx) = unbounded_channel::<Message>();
let writer_handle = tokio::spawn(async move {
while let Some(msg) = tokio::select! {
biased;
_ = write_token.cancelled() => None,
msg = ws_rx.recv() => msg,
} {
if write.send(msg).await.is_err() {
break;
}
}
});
let mut state = handler::SessionState::new(
self,
ws_tx.clone(),
seq_ack.clone(),
conn_token.clone(),
persistent_state,
backoff,
)
.await
.inspect_err(|_e| {
conn_token.cancel();
})?;
let outcome = match read.next().await {
Some(Ok(m)) => self.handle_message(&mut state, m).await,
_ => Some(SessionOutcome::Reconnect),
};
if let Some(out) = outcome {
conn_token.cancel();
writer_handle.abort();
let _ = writer_handle.await;
return Ok(out);
}
if !state.has_heartbeat() {
conn_token.cancel();
writer_handle.abort();
let _ = writer_handle.await;
return Ok(SessionOutcome::Reconnect);
}
let handshake = if is_resume {
debug!(
"[{}] Sending Resume with seq_ack={}",
self.guild_id,
seq_ack.load(Ordering::Relaxed)
);
protocol::builders::resume(
self.guild_id.to_string(),
self.session_id.to_string(),
self.token.clone(),
seq_ack.load(Ordering::Relaxed),
)
} else {
protocol::builders::identify(
self.guild_id.to_string(),
self.user_id.0.to_string(),
self.session_id.to_string(),
self.token.clone(),
1,
)
};
let _ = ws_tx.send(Message::Text(
serde_json::to_string(&handshake).unwrap().into(),
));
let (speaking_tx, mut speaking_rx) = unbounded_channel::<bool>();
state.set_speaking_tx(speaking_tx);
let outcome = loop {
tokio::select! {
biased;
_ = self.outer_token.cancelled() => break SessionOutcome::Shutdown,
_ = conn_token.cancelled() => break SessionOutcome::Reconnect,
Some(speaking) = speaking_rx.recv() => {
self.notify_speaking(&ws_tx, state.ssrc(), speaking);
}
msg = read.next() => match msg {
Some(Ok(m)) => if let Some(out) = self.handle_message(&mut state, m).await {
break out;
},
Some(Err(_)) => break SessionOutcome::Reconnect,
None => break SessionOutcome::Reconnect,
}
}
};
conn_token.cancel();
writer_handle.abort();
let _ = writer_handle.await;
Ok(outcome)
}
async fn handle_message(
&self,
state: &mut handler::SessionState<'_>,
msg: Message,
) -> Option<SessionOutcome> {
match msg {
Message::Text(text) => state.handle_text(text.to_string()).await,
Message::Binary(bin) => {
state.handle_binary(bin.to_vec()).await;
None
}
Message::Close(frame) => {
let code = frame.as_ref().map(|f| f.code.into()).unwrap_or(1000u16);
let reason = frame.map(|f| f.reason.to_string()).unwrap_or_default();
let attempt = state.attempt();
debug!("[{}] Gateway closed: {} ({})", self.guild_id, code, reason);
if !self.policy.is_retryable(code, attempt) {
self.emit_close(code, reason);
}
Some(self.policy.classify(code))
}
Message::Ping(p) => {
let _ = state.tx().send(Message::Pong(p));
None
}
_ => None,
}
}
fn notify_speaking(&self, tx: &UnboundedSender<Message>, ssrc: u32, speaking: bool) {
let msg = protocol::GatewayPayload {
op: protocol::OpCode::Speaking as u8,
seq: None,
d: serde_json::json!({
"speaking": if speaking { 1 } else { 0 },
"delay": 0,
"ssrc": ssrc
}),
};
if let Ok(json) = serde_json::to_string(&msg) {
let _ = tx.send(Message::Text(json.into()));
}
}
fn emit_close(&self, code: u16, reason: String) {
if let Some(tx) = &self.event_tx {
let _ = tx.send(RustalinkEvent::WebSocketClosed {
guild_id: self.guild_id.clone(),
code,
reason,
by_remote: true,
});
}
}
}
}
pub mod udp_link {
use crate::{
common::types::AnyResult,
gateway::{
constants::{
RTP_OPUS_PAYLOAD_TYPE, RTP_TIMESTAMP_STEP, RTP_VERSION_BYTE,
UDP_PACKET_BUF_CAPACITY,
},
session::types::map_boxed_err,
},
};
use davey::{AeadInPlace, Aes256Gcm, KeyInit};
use serde::{Deserialize, Serialize};
use std::{net::SocketAddr, sync::Arc};
use tokio::net::UdpSocket;
use xsalsa20poly1305::XSalsa20Poly1305;
pub struct UDPVoiceTransport {
socket: Arc<UdpSocket>,
address: SocketAddr,
pub ssrc: u32,
pub crypto: CryptoBackend,
pub rtp: RtpState,
pub buffer: Vec<u8>,
}
pub enum CryptoBackend {
XSalsa20Poly1305(Box<XSalsa20Poly1305>),
Aes256Gcm(Box<Aes256Gcm>),
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct RtpState {
pub sequence: u16,
pub timestamp: u32,
pub nonce: u32,
}
impl UDPVoiceTransport {
pub fn new(
socket: Arc<UdpSocket>,
address: SocketAddr,
ssrc: u32,
secret_key: [u8; 32],
mode: &str,
rtp_state: Option<RtpState>,
) -> AnyResult<Self> {
let crypto = match mode {
"aead_aes256_gcm_rtpsize" => {
CryptoBackend::Aes256Gcm(Box::new(Aes256Gcm::new(&secret_key.into())))
}
_ => CryptoBackend::XSalsa20Poly1305(Box::new(XSalsa20Poly1305::new(
&secret_key.into(),
))),
};
Ok(Self {
socket,
address,
ssrc,
crypto,
rtp: rtp_state.unwrap_or_else(RtpState::randomize),
buffer: Vec::with_capacity(UDP_PACKET_BUF_CAPACITY),
})
}
pub async fn send_keepalive(&self, counter: u32) -> AnyResult<()> {
let payload = counter.to_be_bytes();
self.socket.send_to(&payload, self.address).await?;
Ok(())
}
pub async fn transmit_opus(&mut self, opus_data: &[u8]) -> AnyResult<()> {
let (seq, ts, nonce_val) = self.rtp.next();
let mut header = [0u8; 12];
header[0] = RTP_VERSION_BYTE;
header[1] = RTP_OPUS_PAYLOAD_TYPE;
header[2..4].copy_from_slice(&seq.to_be_bytes());
header[4..8].copy_from_slice(&ts.to_be_bytes());
header[8..12].copy_from_slice(&self.ssrc.to_be_bytes());
self.buffer.clear();
self.buffer.extend_from_slice(&header);
self.buffer.extend_from_slice(opus_data);
match &self.crypto {
CryptoBackend::XSalsa20Poly1305(cipher) => {
let mut nonce = [0u8; 24];
nonce[0..12].copy_from_slice(&header);
let tag = cipher
.encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
.map_err(|e| map_boxed_err(format!("XSalsa20 error: {e:?}")))?;
self.buffer.extend_from_slice(&tag);
}
CryptoBackend::Aes256Gcm(cipher) => {
let mut nonce = [0u8; 12];
nonce[0..4].copy_from_slice(&nonce_val.to_be_bytes());
let tag = cipher
.encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
.map_err(|e| map_boxed_err(format!("AES-GCM error: {e:?}")))?;
self.buffer.extend_from_slice(&tag);
self.buffer.extend_from_slice(&nonce_val.to_be_bytes());
}
}
self.socket.send_to(&self.buffer, self.address).await?;
Ok(())
}
}
impl RtpState {
fn randomize() -> Self {
Self {
sequence: rand::random(),
timestamp: rand::random(),
nonce: rand::random(),
}
}
fn next(&mut self) -> (u16, u32, u32) {
let seq = self.sequence;
let ts = self.timestamp;
let n = self.nonce;
self.sequence = self.sequence.wrapping_add(1);
self.timestamp = self.timestamp.wrapping_add(RTP_TIMESTAMP_STEP);
self.nonce = self.nonce.wrapping_add(1);
(seq, ts, n)
}
}
}
pub mod encryption {
use crate::{
common::types::{AnyError, AnyResult, ChannelId, UserId},
gateway::{
constants::{DAVE_INITIAL_VERSION, MAX_PENDING_PROPOSALS, SILENCE_FRAME},
session::types::map_boxed_err,
},
};
use davey::{DaveSession, ProposalsOperationType};
use std::{
collections::{HashMap, HashSet},
num::NonZeroU16,
};
use tracing::{debug, trace, warn};
const DAVE_MIN_VERSION: NonZeroU16 = match NonZeroU16::new(DAVE_INITIAL_VERSION) {
Some(v) => v,
None => unreachable!(),
};
pub struct DaveHandler {
session: Option<DaveSession>,
user_id: UserId,
channel_id: ChannelId,
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,
recognized_users: HashSet<UserId>,
cached_user_ids: Vec<u64>,
}
impl DaveHandler {
pub fn new(user_id: UserId, channel_id: ChannelId) -> Self {
let mut recognized_users = HashSet::new();
recognized_users.insert(user_id);
Self {
session: None,
user_id,
channel_id,
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,
recognized_users,
cached_user_ids: vec![user_id.0],
}
}
pub fn add_users(&mut self, uids: &[u64]) {
for &uid in uids {
self.recognized_users.insert(UserId(uid));
}
self.update_user_cache();
debug!("DAVE adding users: {:?}", uids);
}
pub fn remove_user(&mut self, uid: u64) {
if self.recognized_users.remove(&UserId(uid)) {
self.update_user_cache();
}
debug!("DAVE removing user: {}", uid);
}
fn update_user_cache(&mut self) {
self.cached_user_ids.clear();
self.cached_user_ids
.extend(self.recognized_users.iter().map(|u| u.0));
self.cached_user_ids.sort_unstable();
}
pub fn protocol_version(&self) -> u16 {
self.protocol_version
}
pub fn set_protocol_version(&mut self, version: u16) {
self.protocol_version = version;
}
pub fn setup_session(&mut self, version: u16) -> AnyResult<Vec<u8>> {
if version == 0 {
self.reset();
return Ok(Vec::new());
}
let nz_version = NonZeroU16::new(version).unwrap_or(DAVE_MIN_VERSION);
if let Some(s) = &mut self.session {
s.reinit(nz_version, self.user_id.0, self.channel_id.0, None)
.map_err(map_boxed_err)?;
} else {
let session = DaveSession::new(nz_version, self.user_id.0, self.channel_id.0, None)
.map_err(map_boxed_err)?;
self.session = Some(session);
}
let session = self
.session
.as_mut()
.ok_or_else(|| map_boxed_err("DAVE session initialization failed"))?;
self.protocol_version = version;
self.external_sender_set = false;
self.pending_proposals.clear();
self.pending_handshake.clear();
self.was_ready = false;
debug!("DAVE session setup (v{})", version);
let key_package = session.create_key_package().map_err(map_boxed_err)?;
if let Some(saved) = self.saved_external_sender.as_deref()
&& 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;
self.session = None;
debug!("DAVE session reset to plaintext");
}
pub fn prepare_transition(&mut self, transition_id: u16, protocol_version: u16) -> bool {
self.pending_transitions
.insert(transition_id, protocol_version);
if transition_id == 0 {
self.execute_transition(0);
return false;
}
true
}
pub fn execute_transition(&mut self, transition_id: u16) {
if let Some(next_version) = self.pending_transitions.remove(&transition_id) {
self.protocol_version = next_version;
trace!(
"DAVE transition {} executed (v{})",
transition_id, next_version
);
}
}
pub fn prepare_epoch(&mut self, epoch: u64, protocol_version: u16) -> Option<Vec<u8>> {
if epoch == 1 {
match self.setup_session(protocol_version) {
Ok(kp) => return Some(kp),
Err(e) => warn!("DAVE prepare_epoch setup failed: {e}"),
}
}
None
}
pub fn process_external_sender(&mut self, data: &[u8]) -> AnyResult<Vec<Vec<u8>>> {
let mut responses = Vec::new();
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!(
"DAVE processing {} buffered proposals",
self.pending_proposals.len()
);
for prop_data in std::mem::take(&mut self.pending_proposals) {
if let Ok(Some(res)) =
Self::do_process_proposals(session, &prop_data, &self.cached_user_ids)
{
responses.push(res);
}
}
}
if !self.pending_handshake.is_empty() {
debug!(
"DAVE processing {} buffered handshake messages",
self.pending_handshake.len()
);
for (handshake_data, is_welcome) in std::mem::take(&mut self.pending_handshake)
{
if let Err(e) = self.do_process_handshake(&handshake_data, is_welcome) {
warn!("DAVE buffered handshake processing failed: {e}");
}
}
}
}
Ok(responses)
}
pub fn process_welcome(&mut self, data: &[u8]) -> AnyResult<u16> {
self.process_handshake_message(data, true)
}
pub fn process_commit(&mut self, data: &[u8]) -> AnyResult<u16> {
self.process_handshake_message(data, false)
}
fn process_handshake_message(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<u16> {
let tag = if is_welcome { "welcome" } else { "commit" };
if data.len() < 2 {
let msg = if is_welcome {
"DAVE welcome"
} else {
"DAVE commit"
};
return Err(short_payload_err(msg));
}
let transition_id = u16::from_be_bytes([data[0], data[1]]);
if !self.external_sender_set {
if self.pending_handshake.len() < MAX_PENDING_PROPOSALS {
debug!("DAVE buffering {tag} — external sender not set");
self.pending_handshake.push((data.to_vec(), is_welcome));
} else {
warn!("DAVE handshake buffer full, dropping {tag}");
}
return Ok(transition_id);
}
self.do_process_handshake(data, is_welcome)?;
Ok(transition_id)
}
fn do_process_handshake(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<()> {
let transition_id = u16::from_be_bytes([data[0], data[1]]);
if let Some(session) = &mut self.session {
if is_welcome {
session.process_welcome(&data[2..]).map_err(map_boxed_err)?;
} else {
session.process_commit(&data[2..]).map_err(map_boxed_err)?;
}
if transition_id != 0 {
self.pending_transitions
.insert(transition_id, self.protocol_version);
}
debug!(
"DAVE {} processed (tid {})",
if is_welcome { "welcome" } else { "commit" },
transition_id
);
}
Ok(())
}
pub fn process_proposals(&mut self, data: &[u8]) -> AnyResult<Option<Vec<u8>>> {
if data.is_empty() {
return Err(short_payload_err("DAVE proposals"));
}
if !self.external_sender_set {
if self.pending_proposals.len() < MAX_PENDING_PROPOSALS {
debug!("DAVE buffering proposal — external sender not set");
self.pending_proposals.push(data.to_vec());
} else {
warn!("DAVE proposal buffer full, dropping proposal");
}
return Ok(None);
}
let session = match &mut self.session {
Some(s) => s,
None => return Ok(None),
};
Self::do_process_proposals(session, data, &self.cached_user_ids)
}
fn do_process_proposals(
session: &mut DaveSession,
data: &[u8],
user_ids: &[u64],
) -> AnyResult<Option<Vec<u8>>> {
let op_type = match data[0] {
0 => ProposalsOperationType::APPEND,
1 => ProposalsOperationType::REVOKE,
raw => return Err(map_boxed_err(format!("Unknown DAVE proposals op: {raw}"))),
};
let result = session
.process_proposals(op_type, &data[1..], Some(user_ids))
.map_err(map_boxed_err)?;
if let Some(cw) = result {
let mut out = cw.commit;
if let Some(w) = cw.welcome {
out.extend_from_slice(&w);
}
return Ok(Some(out));
}
Ok(None)
}
pub fn encrypt_opus(&mut self, packet: &[u8]) -> AnyResult<Vec<u8>> {
if packet == SILENCE_FRAME || self.protocol_version == 0 {
return Ok(packet.to_vec());
}
if let Some(session) = &mut self.session {
let is_ready = session.is_ready();
if is_ready != self.was_ready {
if is_ready {
debug!("DAVE session (v{}) is READY", self.protocol_version);
} else {
warn!("DAVE session (v{}) LOST readiness", self.protocol_version);
}
self.was_ready = is_ready;
}
if is_ready {
return session
.encrypt_opus(packet)
.map(|c| c.into_owned())
.map_err(map_boxed_err);
}
}
Ok(packet.to_vec())
}
pub fn voice_privacy_code(&self) -> Option<String> {
self.session
.as_ref()
.and_then(|s| s.voice_privacy_code().map(|c| c.to_string()))
}
}
#[inline]
fn short_payload_err(context: &str) -> AnyError {
map_boxed_err(format!("Invalid {context} payload: too short"))
}
}
pub mod engine {
use crate::{audio::Mixer, common::types::Shared, gateway::constants::DEFAULT_SAMPLE_RATE};
use tokio::sync::Mutex;
pub struct VoiceEngine {
pub mixer: Shared<Mixer>,
pub dave: Option<Shared<crate::gateway::DaveHandler>>,
}
impl VoiceEngine {
pub fn new() -> Self {
Self {
mixer: Shared::new(Mutex::new(Mixer::new(DEFAULT_SAMPLE_RATE))),
dave: None,
}
}
}
impl Default for VoiceEngine {
fn default() -> Self {
Self::new()
}
}
}
pub use encryption::DaveHandler;
pub use engine::VoiceEngine;
pub use session::{VoiceGateway, VoiceGatewayConfig};