From 1d980136e8c87bcda0262b50c196a92349a38956 Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 00:59:49 +0530 Subject: [PATCH 1/7] Log WebSocket failures and preserve resumed sessions Send heartbeats while flushing backlogs and log dropped updates when the outbound queue is full. --- src/rest/ws/mod.rs | 7 ++++++- src/rest/ws/sender.rs | 38 ++++++++++++++++++++++++++++++-------- src/session/context.rs | 5 +++++ 3 files changed, 41 insertions(+), 9 deletions(-) diff --git a/src/rest/ws/mod.rs b/src/rest/ws/mod.rs index 7d7b629..21b56e3 100644 --- a/src/rest/ws/mod.rs +++ b/src/rest/ws/mod.rs @@ -92,7 +92,12 @@ async fn handle_socket( // Couple the halves: whichever ends first aborts the other, so a dead write task tears the // session down instead of leaving the read loop spinning while events pile up unread. - let write_task = tokio::spawn(sender::write_loop(sink, receiver, backlog)); + let write_task = tokio::spawn(sender::write_loop( + sink, + receiver, + backlog, + context.session_id.clone(), + )); let read_task = tokio::spawn(receiver::read_loop(stream, context.session_id.clone())); let reason = supervise(read_task, write_task).await; diff --git a/src/rest/ws/sender.rs b/src/rest/ws/sender.rs index ba772b8..02856da 100644 --- a/src/rest/ws/sender.rs +++ b/src/rest/ws/sender.rs @@ -6,6 +6,7 @@ use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures_util::stream::SplitSink; use futures_util::SinkExt; use tokio::sync::mpsc; +use tokio::time::Instant; use crate::protocol::message::Message; @@ -21,35 +22,56 @@ pub(super) async fn write_loop( mut sink: Sink, mut receiver: mpsc::Receiver, backlog: Vec, + session_id: String, ) { + let mut next_ping = Instant::now() + PING_INTERVAL; for message in backlog { if !send_text(&mut sink, &message).await { + tracing::warn!(session = %session_id, "websocket event write failed or timed out"); return; } + // A resumed session may have a large event backlog. Keep emitting heartbeats while it is + // flushed; otherwise the read half's idle timeout can kill the socket before replay ends. + if Instant::now() >= next_ping { + if !send_ping(&mut sink, &session_id).await { + return; + } + next_ping = Instant::now() + PING_INTERVAL; + } } - let mut ping = tokio::time::interval(PING_INTERVAL); - ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - ping.tick().await; // skip the immediate first tick - loop { tokio::select! { message = receiver.recv() => { - let Some(message) = message else { return }; + let Some(message) = message else { + tracing::debug!(session = %session_id, "websocket outbound channel closed"); + return; + }; if !send_text(&mut sink, &message).await { + tracing::warn!(session = %session_id, "websocket event write failed or timed out"); return; } } - _ = ping.tick() => { - let ping = sink.send(WsMessage::Ping(Vec::new().into())); - if !matches!(tokio::time::timeout(SEND_TIMEOUT, ping).await, Ok(Ok(()))) { + _ = tokio::time::sleep_until(next_ping) => { + if !send_ping(&mut sink, &session_id).await { return; } + next_ping = Instant::now() + PING_INTERVAL; } } } } +async fn send_ping(sink: &mut Sink, session_id: &str) -> bool { + let ping = sink.send(WsMessage::Ping(Vec::new().into())); + if matches!(tokio::time::timeout(SEND_TIMEOUT, ping).await, Ok(Ok(()))) { + true + } else { + tracing::warn!(session = %session_id, "websocket ping write failed or timed out"); + false + } +} + async fn send_text(sink: &mut Sink, message: &Message) -> bool { let Ok(text) = serde_json::to_string(message) else { return true; // skip an unserializable message, keep the socket diff --git a/src/session/context.rs b/src/session/context.rs index 0078905..fc29dee 100644 --- a/src/session/context.rs +++ b/src/session/context.rs @@ -129,6 +129,11 @@ impl SocketContext { self.dropped_updates.fetch_add(1, Ordering::Relaxed); return; } + tracing::warn!( + session = %self.session_id, + dropped_updates = self.dropped_updates(), + "websocket outbound queue full; parking session for resume" + ); let mut queue = VecDeque::new(); queue.push_back(message); *outbound = Outbound::Queued { queue }; From f7bf5680e43df23cb6bcffd8014a2bd585a7a447 Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 16:11:33 +0530 Subject: [PATCH 2/7] Add PORT override for server binding Update dependencies and bump Kairo to 4.8.5. Lower routine resource metrics logging to debug and exclude local Cargo config from Docker builds. --- .dockerignore | 3 +++ Cargo.lock | 22 +++++++++------------- Cargo.toml | 10 +++++----- README.md | 3 ++- docker-compose.yml | 1 + src/config/mod.rs | 40 ++++++++++++++++++++++++++++++++++++++-- src/node/state.rs | 2 +- 7 files changed, 59 insertions(+), 22 deletions(-) diff --git a/.dockerignore b/.dockerignore index f3dcd51..224df4d 100644 --- a/.dockerignore +++ b/.dockerignore @@ -5,6 +5,9 @@ # Local clones of the git dependencies. Cargo fetches its own copies. /crates +# Local dev-only cargo config (patch/paths overrides pointing at /crates). +.cargo/ + # Local runtime state. /logs config.toml diff --git a/Cargo.lock b/Cargo.lock index bc0049c..037e25f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -701,7 +701,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1410,7 +1410,7 @@ dependencies = [ [[package]] name = "kairo" -version = "4.8.4" +version = "4.8.5" dependencies = [ "axum", "bytes", @@ -1729,8 +1729,7 @@ checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" [[package]] name = "lyrics" -version = "0.2.8" -source = "git+https://github.com/bongo-devs/lyrics?tag=v0.2.8#fb57296eed2224a28ab422a066067da5165c639a" +version = "0.2.9" dependencies = [ "async-trait", "base64 0.22.1", @@ -1833,7 +1832,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2110,8 +2109,7 @@ dependencies = [ [[package]] name = "player" -version = "0.1.9" -source = "git+https://github.com/bongo-devs/player?tag=v0.1.9#9a7d6fbb0d69ae1422d1759e4b8311c843e2c0bf" +version = "0.1.10" dependencies = [ "aes", "base64 0.22.1", @@ -2277,7 +2275,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2845,8 +2843,7 @@ dependencies = [ [[package]] name = "sources" -version = "0.1.26" -source = "git+https://github.com/bongo-devs/sources?tag=v0.1.26#690c1cb736991eecd4bcef82367d33a296e72fcf" +version = "0.1.27" dependencies = [ "aes", "base64 0.22.1", @@ -3626,8 +3623,7 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" -version = "0.1.4" -source = "git+https://github.com/bongo-devs/voice?tag=v0.1.4#aea253ef133db5bd0d9172d6070dde184a7a64f0" +version = "0.1.6" dependencies = [ "aes-gcm", "bytes", @@ -3805,7 +3801,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index eecfa39..160ba87 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "kairo" -version = "4.8.4" +version = "4.8.5" edition = "2021" license = "MIT" description = "Standalone Discord audio sending node, written in Rust. Drop-in Lavalink v4 compatible." @@ -21,19 +21,19 @@ tracing-appender = "0.2" [dependencies.lyrics] git = "https://github.com/bongo-devs/lyrics" -tag = "v0.2.8" +tag = "v0.2.9" [dependencies.player] git = "https://github.com/bongo-devs/player" -tag = "v0.1.9" +tag = "v0.1.10" [dependencies.sources] git = "https://github.com/bongo-devs/sources" -tag = "v0.1.26" +tag = "v0.1.27" [dependencies.voice] git = "https://github.com/bongo-devs/voice" -tag = "v0.1.4" +tag = "v0.1.6" [dependencies.reqwest] version = "0.12" diff --git a/README.md b/README.md index 1601e62..b784948 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,8 @@ docker compose up -d ``` The node serves port 2333 and reads `/app/config.toml`. Set `KAIRO_CONFIG` to read it from -somewhere else. +somewhere else, and `PORT` to override `server.port` — handy when several instances share one +config file. ## Configuration diff --git a/docker-compose.yml b/docker-compose.yml index 4004714..b953ea1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -19,5 +19,6 @@ services: environment: - RUST_LOG=info - KAIRO_CONFIG=/app/config.toml + # - PORT=2333 # Commit arena pages on demand rather than up front, worth about 12 MB at idle. - MIMALLOC_ARENA_EAGER_COMMIT=0 diff --git a/src/config/mod.rs b/src/config/mod.rs index 3dc334d..ff49b0a 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -71,6 +71,7 @@ impl Config { /// Read the file named by the first argument, then `KAIRO_CONFIG`, then `config.toml`. /// /// Panics if the file cannot be read or parsed, since there is nothing to serve without it. + /// The `PORT` environment variable, when set, overrides the file's `server.port`. pub fn new() -> Self { let path = std::env::args() .nth(1) @@ -78,8 +79,10 @@ impl Config { .unwrap_or_else(|| DEFAULT_CONFIG.to_string()); let content = std::fs::read_to_string(&path) .unwrap_or_else(|e| panic!("Failed to read config '{}': {}", path, e)); - Self::from_toml(&content) - .unwrap_or_else(|e| panic!("Failed to parse config '{}': {}", path, e)) + let mut config = Self::from_toml(&content) + .unwrap_or_else(|e| panic!("Failed to parse config '{}': {}", path, e)); + config.set_port(std::env::var("PORT").ok().as_deref()); + config } fn from_toml(content: &str) -> Result { @@ -89,6 +92,17 @@ impl Config { toml::from_str(content) } + // `PORT` is the container convention: it lets several instances share one config file and + // differ only in the port they bind. Parsed separately from the env read so it stays testable. + fn set_port(&mut self, port: Option<&str>) { + let Some(port) = port else { + return; + }; + self.server.port = port.parse().unwrap_or_else(|_| { + panic!("Invalid PORT '{}': expected a number from 0 to 65535", port) + }); + } + /// Build the engine-wide [`AudioConfiguration`] from these settings. pub fn audio_configuration(&self) -> AudioConfiguration { let server = &self.lavalink.server; @@ -122,4 +136,26 @@ mod tests { assert_eq!(cfg.server.port, 2333); assert_eq!(cfg.lavalink.server.player_update_interval, 5); } + + #[test] + fn port_override_wins_over_the_file_value() { + let mut cfg = Config::from_toml("[server]\nport = 2333\n").expect("valid server block"); + cfg.set_port(Some("2334")); + assert_eq!(cfg.server.port, 2334); + } + + #[test] + fn absent_port_override_leaves_the_file_value() { + let mut cfg = Config::from_toml("[server]\nport = 2334\n").expect("valid server block"); + cfg.set_port(None); + assert_eq!(cfg.server.port, 2334); + } + + #[test] + #[should_panic(expected = "Invalid PORT 'nope'")] + fn port_override_must_be_a_number() { + Config::from_toml("") + .expect("an empty config is all defaults") + .set_port(Some("nope")); + } } diff --git a/src/node/state.rs b/src/node/state.rs index 7bd46bd..5399f7d 100644 --- a/src/node/state.rs +++ b/src/node/state.rs @@ -226,7 +226,7 @@ impl AppState { .ok() .flatten(); let gate = crate::session::voice_gate::metrics(); - tracing::info!( + tracing::debug!( fd_count = ?fd_count, fd_limit = ?crate::node::stats::fd_limit(), players = state.sockets().total_players(), From ba2d2d30c8105d45b44049c6aee58a7ffed82469 Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 17:30:28 +0530 Subject: [PATCH 3/7] Lock Bongo dependencies to their tagged Git sources --- Cargo.lock | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 037e25f..e4fb83d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1730,6 +1730,7 @@ checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" [[package]] name = "lyrics" version = "0.2.9" +source = "git+https://github.com/bongo-devs/lyrics?tag=v0.2.9#6d0ff1688ab48a6872d8fe225deb14ca891c540a" dependencies = [ "async-trait", "base64 0.22.1", @@ -2110,6 +2111,7 @@ dependencies = [ [[package]] name = "player" version = "0.1.10" +source = "git+https://github.com/bongo-devs/player?tag=v0.1.10#f64ec671a30edf895bf1fcd17a56f344715940f6" dependencies = [ "aes", "base64 0.22.1", @@ -2844,6 +2846,7 @@ dependencies = [ [[package]] name = "sources" version = "0.1.27" +source = "git+https://github.com/bongo-devs/sources?tag=v0.1.27#3a64132943d80b1b34785bb787d2814a34a4c4aa" dependencies = [ "aes", "base64 0.22.1", @@ -3624,6 +3627,7 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" version = "0.1.6" +source = "git+https://github.com/bongo-devs/voice?tag=v0.1.6#8fff1f6750f15ede1a3684f67d906bf935c2e271" dependencies = [ "aes-gcm", "bytes", From 17e74df73f2dbaf94d40bb99afa8454ff01bacfe Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 18:14:42 +0530 Subject: [PATCH 4/7] Update voice dependency to v0.1.7 --- Cargo.lock | 4 ++-- Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e4fb83d..76b3dbd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3626,8 +3626,8 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" -version = "0.1.6" -source = "git+https://github.com/bongo-devs/voice?tag=v0.1.6#8fff1f6750f15ede1a3684f67d906bf935c2e271" +version = "0.1.7" +source = "git+https://github.com/bongo-devs/voice?tag=v0.1.7#7a6ad8fcc4b71f97e8ffa21b8606022e545e6594" dependencies = [ "aes-gcm", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 160ba87..0e442a4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ tag = "v0.1.27" [dependencies.voice] git = "https://github.com/bongo-devs/voice" -tag = "v0.1.6" +tag = "v0.1.7" [dependencies.reqwest] version = "0.12" From fd0d54a46d18532251da2736acdd6a2f9793d976 Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 20:02:58 +0530 Subject: [PATCH 5/7] fix(ws): keep the WebSocket behavior to the v4 contract Drop the send deadline, the outbound queue cap with park-on-full, the close-frame flush, the server ping and the idle read deadline. Messages queue unbounded while a slow client drains, player updates are skipped at the source while the session is parked, and stats ride a fixed-rate tick. --- Cargo.toml | 3 + src/node/state.rs | 7 +- src/node/tasks.rs | 32 +++++- src/rest/players.rs | 4 +- src/rest/ws/mod.rs | 17 +-- src/rest/ws/receiver.rs | 21 +--- src/rest/ws/sender.rs | 235 +++++++++++++++++++++++++++++--------- src/session/context.rs | 175 ++++++++++------------------ src/session/player/mod.rs | 30 ++++- 9 files changed, 308 insertions(+), 216 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 0e442a4..589ed03 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,6 +79,9 @@ features = ["derive"] version = "0.3" features = ["env-filter", "json", "chrono"] +[dev-dependencies] +tokio = { version = "1.53", features = ["test-util"] } + [profile.release] lto = "thin" codegen-units = 1 diff --git a/src/node/state.rs b/src/node/state.rs index 5399f7d..d6f353b 100644 --- a/src/node/state.rs +++ b/src/node/state.rs @@ -162,15 +162,12 @@ impl AppState { } } - /// Start this session's `stats` timer, firing immediately and then once a minute. - /// - /// Per-session rather than one node-wide loop because `frameStats` are per-session, and this way - /// the timer is cancelled along with the session it belongs to. + /// Start this session's `stats` timer: immediately, then once a minute on a fixed grid. pub fn arm_session_stats(&self, context: &Arc) { let state = self.clone(); // Weak, so a session dropped without `shutdown()` running cannot be kept alive by its timer. let weak = Arc::downgrade(context); - TASKS.add( + TASKS.add_fixed_rate( tasks::session_stats(&context.session_id), Duration::from_secs(STATS_INTERVAL_SECS), move || { diff --git a/src/node/tasks.rs b/src/node/tasks.rs index d47f6ff..2848daa 100644 --- a/src/node/tasks.rs +++ b/src/node/tasks.rs @@ -6,6 +6,7 @@ use std::sync::{LazyLock, Mutex}; use std::time::Duration; use tokio::task::JoinHandle; +use tokio::time::MissedTickBehavior; /// The process-wide task registry. pub static TASKS: LazyLock = LazyLock::new(Tasks::default); @@ -22,16 +23,39 @@ pub struct Tasks { } impl Tasks { - /// Run `handler` every `period`, aborting whatever was registered under `key` before. - pub fn add(&self, key: impl Into, period: Duration, mut handler: F) + /// Run `handler` every `period` (a slow run pushes the next tick out), replacing any task + /// registered under `key`. + pub fn add(&self, key: impl Into, period: Duration, handler: F) where F: FnMut() -> Fut + Send + 'static, Fut: Future + Send + 'static, { - let key = key.into(); + self.spawn(key.into(), period, MissedTickBehavior::Delay, handler); + } + + /// Run `handler` on a fixed-rate grid: missed ticks fire immediately instead of pushing the + /// schedule out. + pub fn add_fixed_rate(&self, key: impl Into, period: Duration, handler: F) + where + F: FnMut() -> Fut + Send + 'static, + Fut: Future + Send + 'static, + { + self.spawn(key.into(), period, MissedTickBehavior::Burst, handler); + } + + fn spawn( + &self, + key: String, + period: Duration, + behavior: MissedTickBehavior, + mut handler: F, + ) where + F: FnMut() -> Fut + Send + 'static, + Fut: Future + Send + 'static, + { let runner = tokio::spawn(async move { let mut interval = tokio::time::interval(period); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + interval.set_missed_tick_behavior(behavior); loop { interval.tick().await; handler().await; diff --git a/src/rest/players.rs b/src/rest/players.rs index e5e8617..6fe13aa 100644 --- a/src/rest/players.rs +++ b/src/rest/players.rs @@ -407,7 +407,7 @@ mod tests { async fn patched_state( background_connect: bool, - ) -> (AppState, tokio::sync::mpsc::Receiver) { + ) -> (AppState, tokio::sync::mpsc::UnboundedReceiver) { let mut config = Config::default(); config.lavalink.server.voice.background_connect = background_connect; // Keep the process-global update interval deterministic: the unit test @@ -415,7 +415,7 @@ mod tests { // clamped 1, and test order within one binary is undefined. config.lavalink.server.player_update_interval = 0; let state = AppState::new(config); - let (sender, rx) = tokio::sync::mpsc::channel(256); + let (sender, rx) = tokio::sync::mpsc::unbounded_channel(); let context = SocketContext::new( SESSION.to_string(), 1, diff --git a/src/rest/ws/mod.rs b/src/rest/ws/mod.rs index 21b56e3..650d15e 100644 --- a/src/rest/ws/mod.rs +++ b/src/rest/ws/mod.rs @@ -22,10 +22,6 @@ use crate::protocol::message::Message; use crate::rest::error::RestError; use crate::session::SocketContext; -/// Outbound channel capacity. Past this the client is not draining, so player updates and stats are -/// dropped and events parked for resume (see `SocketContext::send_message`), bounding memory. -const OUTBOUND_CAPACITY: usize = 1024; - /// `GET /v4/websocket`, upgrade the connection once the handshake headers check out. pub async fn websocket_handler( State(state): State, @@ -81,7 +77,7 @@ async fn handle_socket( user_agent: Option, ) { let (sink, stream) = socket.split(); - let (sender, receiver) = mpsc::channel::(OUTBOUND_CAPACITY); + let (sender, receiver) = mpsc::unbounded_channel::(); let (context, resumed, epoch, backlog) = attach(&state, requested_session.as_deref(), user_id, sender); @@ -90,8 +86,7 @@ async fn handle_socket( log_new_connection(client_name, user_agent); } - // Couple the halves: whichever ends first aborts the other, so a dead write task tears the - // session down instead of leaving the read loop spinning while events pile up unread. + // Both halves are one socket: whichever ends first stops the other. let write_task = tokio::spawn(sender::write_loop( sink, receiver, @@ -104,7 +99,6 @@ async fn handle_socket( tracing::debug!( session = %context.session_id, ?reason, - dropped_updates = context.dropped_updates(), "websocket closing" ); on_disconnect(&state, context, epoch); @@ -134,7 +128,7 @@ enum Ended { Write, } -/// Run both halves until either ends, abort the other, and return which ended so teardown runs once. +/// Run both halves until either ends, then stop the other. async fn supervise(mut read_task: JoinHandle<()>, mut write_task: JoinHandle<()>) -> Ended { tokio::select! { _ = &mut read_task => { @@ -152,11 +146,10 @@ fn attach( state: &AppState, requested_session: Option<&str>, user_id: u64, - sender: mpsc::Sender, + sender: mpsc::UnboundedSender, ) -> (Arc, bool, u64, Vec) { if let Some(id) = requested_session { - // `ready` leads the backlog so the write task flushes it ahead of the resume queue and the - // player updates, the order a client expects on resume. + // `ready` leads the backlog, ahead of the resume queue and the player updates. if let Some(context) = state.sockets().take_resumable(id) { let (epoch, queued) = context.resume_with(sender); let mut backlog = Vec::with_capacity(queued.len() + 1); diff --git a/src/rest/ws/receiver.rs b/src/rest/ws/receiver.rs index 0fbfbaf..127e205 100644 --- a/src/rest/ws/receiver.rs +++ b/src/rest/ws/receiver.rs @@ -1,28 +1,19 @@ -//! The socket's read half: watches for the client closing or going silent. - -use std::time::Duration; +//! The socket's read half: watches for the client closing or erroring. use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures_util::stream::SplitStream; use futures_util::StreamExt; -const READ_IDLE_TIMEOUT: Duration = Duration::from_secs(70); - -/// Read until the client closes, errors, or goes silent past the keepalive window. The v4 protocol -/// carries no inbound commands, so a healthy client's only traffic is its pong replies. +/// Read until the client closes or errors. No read deadline: a quiet client is healthy. pub(super) async fn read_loop(mut stream: SplitStream, session_id: String) { loop { - match tokio::time::timeout(READ_IDLE_TIMEOUT, stream.next()).await { - Ok(Some(Ok(WsMessage::Close(_)))) | Ok(Some(Err(_))) | Ok(None) => break, - Ok(Some(Ok(WsMessage::Text(_)))) => tracing::warn!( + match stream.next().await { + Some(Ok(WsMessage::Close(_))) | Some(Err(_)) | None => break, + Some(Ok(WsMessage::Text(_))) => tracing::warn!( session = %session_id, "Kairo does not support websocket messages. Please use the REST api." ), - Ok(Some(Ok(_))) => {} // ping, pong, binary: liveness only - Err(_) => { - tracing::warn!(session = %session_id, "websocket idle past keepalive; assuming half-open"); - break; - } + Some(Ok(_)) => {} // ping, pong, binary: liveness only } } } diff --git a/src/rest/ws/sender.rs b/src/rest/ws/sender.rs index 02856da..7355494 100644 --- a/src/rest/ws/sender.rs +++ b/src/rest/ws/sender.rs @@ -1,81 +1,202 @@ //! The socket's write half: drains the outbound channel to the client. -use std::time::Duration; - -use axum::extract::ws::{Message as WsMessage, WebSocket}; -use futures_util::stream::SplitSink; +use axum::extract::ws::Message as WsMessage; +use futures_util::Sink; use futures_util::SinkExt; use tokio::sync::mpsc; -use tokio::time::Instant; use crate::protocol::message::Message; -const SEND_TIMEOUT: Duration = Duration::from_secs(10); -const PING_INTERVAL: Duration = Duration::from_secs(20); - -type Sink = SplitSink; - -/// Flush the resume `backlog` in order, then stream live messages, pinging an idle client to -/// surface a half-open socket. Returns when the channel closes or a send stalls past the timeout, -/// which tears the session down. -pub(super) async fn write_loop( - mut sink: Sink, - mut receiver: mpsc::Receiver, +// Flush the resume `backlog` in order, then stream live messages until the channel closes; a slow +// socket waits in the channel, it is never timed out. +pub(super) async fn write_loop( + mut sink: S, + mut receiver: mpsc::UnboundedReceiver, backlog: Vec, session_id: String, -) { - let mut next_ping = Instant::now() + PING_INTERVAL; +) where + S: Sink + Unpin, +{ for message in backlog { if !send_text(&mut sink, &message).await { - tracing::warn!(session = %session_id, "websocket event write failed or timed out"); + tracing::warn!(session = %session_id, "websocket write failed, closing the socket"); return; } - // A resumed session may have a large event backlog. Keep emitting heartbeats while it is - // flushed; otherwise the read half's idle timeout can kill the socket before replay ends. - if Instant::now() >= next_ping { - if !send_ping(&mut sink, &session_id).await { - return; - } - next_ping = Instant::now() + PING_INTERVAL; + } + + while let Some(message) = receiver.recv().await { + if !send_text(&mut sink, &message).await { + tracing::warn!(session = %session_id, "websocket write failed, closing the socket"); + return; } } + tracing::debug!(session = %session_id, "websocket outbound channel closed"); +} - loop { - tokio::select! { - message = receiver.recv() => { - let Some(message) = message else { - tracing::debug!(session = %session_id, "websocket outbound channel closed"); - return; - }; - if !send_text(&mut sink, &message).await { - tracing::warn!(session = %session_id, "websocket event write failed or timed out"); - return; - } +async fn send_text(sink: &mut S, message: &Message) -> bool +where + S: Sink + Unpin, +{ + let Ok(text) = serde_json::to_string(message) else { + return true; // skip an unserializable message, keep the socket + }; + sink.send(WsMessage::Text(text.into())).await.is_ok() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::convert::Infallible; + use std::pin::Pin; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::task::{Context, Poll, Waker}; + use std::time::Duration; + + /// A sink that records everything written, standing in for the socket. + #[derive(Clone, Default)] + struct RecordingSink { + sent: Arc>>, + } + + impl Sink for RecordingSink { + type Error = Infallible; + + fn poll_ready( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + + fn start_send(self: Pin<&mut Self>, item: WsMessage) -> Result<(), Self::Error> { + self.sent.lock().unwrap().push(item); + Ok(()) + } + + fn poll_flush( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + } + + /// A sink that refuses writes until its gate opens, standing in for a stalled TCP window. + #[derive(Clone, Default)] + struct StallingSink { + sent: Arc>>, + open: Arc, + waker: Arc>>, + } + + impl StallingSink { + fn open(&self) { + self.open.store(true, Ordering::SeqCst); + if let Some(waker) = self.waker.lock().unwrap().take() { + waker.wake(); } - _ = tokio::time::sleep_until(next_ping) => { - if !send_ping(&mut sink, &session_id).await { - return; - } - next_ping = Instant::now() + PING_INTERVAL; + } + } + + impl Sink for StallingSink { + type Error = Infallible; + + fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if self.open.load(Ordering::SeqCst) { + return Poll::Ready(Ok(())); } + *self.waker.lock().unwrap() = Some(cx.waker().clone()); + Poll::Pending + } + + fn start_send(self: Pin<&mut Self>, item: WsMessage) -> Result<(), Self::Error> { + self.sent.lock().unwrap().push(item); + Ok(()) + } + + fn poll_flush( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) } } -} -async fn send_ping(sink: &mut Sink, session_id: &str) -> bool { - let ping = sink.send(WsMessage::Ping(Vec::new().into())); - if matches!(tokio::time::timeout(SEND_TIMEOUT, ping).await, Ok(Ok(()))) { - true - } else { - tracing::warn!(session = %session_id, "websocket ping write failed or timed out"); - false + fn ready(session_id: &str) -> Message { + Message::Ready { + resumed: false, + session_id: session_id.to_string(), + } } -} -async fn send_text(sink: &mut Sink, message: &Message) -> bool { - let Ok(text) = serde_json::to_string(message) else { - return true; // skip an unserializable message, keep the socket - }; - let send = sink.send(WsMessage::Text(text.into())); - matches!(tokio::time::timeout(SEND_TIMEOUT, send).await, Ok(Ok(()))) + fn sent_ids(sent: &Arc>>) -> Vec { + sent.lock() + .unwrap() + .iter() + .map(|message| match message { + WsMessage::Text(text) => { + let value: serde_json::Value = serde_json::from_str(text).unwrap(); + value["sessionId"].as_str().unwrap().to_string() + } + other => panic!("unexpected frame: {other:?}"), + }) + .collect() + } + + #[tokio::test] + async fn backlog_flushes_before_live_messages() { + let (tx, rx) = mpsc::unbounded_channel(); + let sink = RecordingSink::default(); + let sent = sink.sent.clone(); + + tx.send(ready("live")).unwrap(); + drop(tx); + + write_loop(sink, rx, vec![ready("one"), ready("two")], "s".to_string()).await; + + assert_eq!(sent_ids(&sent), ["one", "two", "live"]); + } + + // A stalled socket must be waited on, not timed out: the message still arrives on release. + #[tokio::test(start_paused = true)] + async fn a_stalled_socket_is_waited_on_not_timed_out() { + let (tx, rx) = mpsc::unbounded_channel(); + let sink = StallingSink::default(); + let sent = sink.sent.clone(); + let gate = sink.clone(); + + let write = tokio::spawn(write_loop(sink, rx, Vec::new(), "s".to_string())); + tx.send(ready("queued")).unwrap(); + drop(tx); + + // Let the writer reach the stalled sink, then idle far past any send deadline. + tokio::time::sleep(Duration::from_millis(50)).await; + tokio::time::sleep(Duration::from_secs(31)).await; + assert!( + !write.is_finished(), + "a slow socket must be waited on, not cut off" + ); + + gate.open(); + tokio::time::timeout(Duration::from_secs(1), write) + .await + .expect("the writer must deliver once the socket drains") + .unwrap(); + assert_eq!(sent_ids(&sent), ["queued"]); + } } diff --git a/src/session/context.rs b/src/session/context.rs index fc29dee..ac3a6e1 100644 --- a/src/session/context.rs +++ b/src/session/context.rs @@ -8,8 +8,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; -use tokio::sync::mpsc::error::TrySendError; -use tokio::sync::mpsc::Sender; +use tokio::sync::mpsc::UnboundedSender; use tokio::task::AbortHandle; use lyrics::LyricsService; @@ -27,12 +26,9 @@ fn locked(mutex: &Mutex) -> MutexGuard<'_, T> { // Where outbound messages currently go. enum Outbound { - // A live socket: try_send to the write task. Bounded, so a slow client cannot grow this without - // limit; player updates are dropped and events parked when it is full (see `send_message`). - Live(Sender), - // Resumable: queue until the client reconnects. Unbounded like upstream: playback continues - // while parked, so a cap would drop real track events on a long gap, exactly when the client - // needs them to catch up on resume. + // A live socket: the write task drains this channel in order, and a slow client waits here. + Live(UnboundedSender), + // Resumable: queue until the client reconnects, oldest first. Queued { // The pending replay, oldest first. queue: VecDeque, @@ -41,15 +37,6 @@ enum Outbound { Closed, } -// Player updates and stats are periodic snapshots, so a fresh one supersedes a dropped one. Every -// other message (track events, WebSocket-closed, ready) is dropped by neither path. -fn is_droppable(message: &Message) -> bool { - matches!( - message, - Message::PlayerUpdate { .. } | Message::Stats { .. } - ) -} - /// Per-connection state for one session. pub struct SocketContext { /// The session id, as used in REST paths and the `ready` message. @@ -73,8 +60,6 @@ pub struct SocketContext { // session the live socket now owns. connection_epoch: AtomicU64, sponsorblock: Mutex>>, - // Count of player updates / stats dropped under outbound backpressure, for logging and metrics. - dropped_updates: AtomicU64, } impl SocketContext { @@ -84,7 +69,7 @@ impl SocketContext { manager: AudioPlayerManager, crossfade_defaults: Option, lyrics_service: Option>, - sender: Sender, + sender: UnboundedSender, ) -> Arc { Arc::new(Self { session_id, @@ -100,7 +85,6 @@ impl SocketContext { resume_timeout_task: Mutex::new(None), connection_epoch: AtomicU64::new(0), sponsorblock: Mutex::new(HashMap::new()), - dropped_updates: AtomicU64::new(0), }) } @@ -109,46 +93,19 @@ impl SocketContext { self.connection_epoch.load(Ordering::Acquire) } - /// Send a message to the client, or queue it while the session is paused. - /// - /// On a live socket this is a non-blocking `try_send`. When the channel is full (a slow or - /// half-open client) or already closed (the write task ended), a periodic player update or - /// stats is dropped, but a track event is parked into the resume queue instead. Parking also - /// drops the live sender, which ends the write task and tears the socket down so the client - /// reconnects and replays the queue — never a silent loss. + /// Send while live, queue while paused, drop once closed. pub fn send_message(&self, message: Message) { let mut outbound = locked(&self.outbound); match &mut *outbound { - Outbound::Live(sender) => match sender.try_send(message) { - Ok(()) => {} - Err(err) => { - let message = match err { - TrySendError::Full(message) | TrySendError::Closed(message) => message, - }; - if is_droppable(&message) { - self.dropped_updates.fetch_add(1, Ordering::Relaxed); - return; - } - tracing::warn!( - session = %self.session_id, - dropped_updates = self.dropped_updates(), - "websocket outbound queue full; parking session for resume" - ); - let mut queue = VecDeque::new(); - queue.push_back(message); - *outbound = Outbound::Queued { queue }; - } - }, + Outbound::Live(sender) => { + // A gone write task means the socket is finished; dropping mirrors a closed socket. + let _ = sender.send(message); + } Outbound::Queued { queue } => queue.push_back(message), Outbound::Closed => {} } } - /// Player updates and stats dropped so far under outbound backpressure. - pub fn dropped_updates(&self) -> u64 { - self.dropped_updates.load(Ordering::Relaxed) - } - /// Get the player for `guild_id`, creating it on the first request for that guild. pub fn get_or_create_player(self: &Arc, guild_id: u64) -> Arc { let mut players = locked(&self.players); @@ -225,12 +182,11 @@ impl SocketContext { self.session_paused.load(Ordering::Acquire) } - /// Pause the session: queue outbound messages until a resume (or timeout). + /// Pause the session: queue outbound messages until a resume or the resume timeout. pub fn pause(&self) { self.session_paused.store(true, Ordering::Release); let mut outbound = locked(&self.outbound); - // Keep any messages already parked: a track event that overflowed the live channel put us - // here with the event sitting in the queue, and a fresh empty queue would drop it. + // Keep an existing queue: pausing twice must not discard a pending replay. if !matches!(&*outbound, Outbound::Queued { .. }) { *outbound = Outbound::Queued { queue: VecDeque::new(), @@ -252,12 +208,8 @@ impl SocketContext { } } - /// Resume the session onto a fresh outbound channel. - /// - /// Returns the new connection epoch and the parked messages, oldest first. The caller flushes - /// them to the socket ahead of any live message (via the write task's backlog) rather than - /// pushing them back through the bounded channel, where a long backlog would overflow and drop. - pub fn resume_with(&self, sender: Sender) -> (u64, Vec) { + /// Resume onto a fresh outbound channel, returning the new epoch and the parked replay. + pub fn resume_with(&self, sender: UnboundedSender) -> (u64, Vec) { self.stop_resume_timeout(); let mut outbound = locked(&self.outbound); let backlog = match std::mem::replace(&mut *outbound, Outbound::Live(sender)) { @@ -272,10 +224,8 @@ impl SocketContext { ) } - /// Replace the outbound channel for a still-live session (a reconnect without resume state). - /// - /// Returns the new connection epoch, which invalidates the previous socket's teardown. - pub fn attach_sender(&self, sender: Sender) -> u64 { + /// Replace the outbound channel for a live session, returning the new connection epoch. + pub fn attach_sender(&self, sender: UnboundedSender) -> u64 { self.stop_resume_timeout(); *locked(&self.outbound) = Outbound::Live(sender); self.session_paused.store(false, Ordering::Release); @@ -315,7 +265,7 @@ mod tests { use super::*; use tokio::sync::mpsc; - fn context(sender: Sender) -> Arc { + fn context(sender: UnboundedSender) -> Arc { SocketContext::new( "session".to_string(), 1, @@ -339,8 +289,6 @@ mod tests { } fn track_event(guild: &str) -> Message { - // A non-droppable event; its exact kind does not matter here, only that it is not a - // player update or stats. Message::event(crate::protocol::message::EmittedEvent::WebSocketClosed { guild_id: guild.to_string(), code: 4006, @@ -349,22 +297,21 @@ mod tests { }) } - // A parked session keeps every event like upstream: no cap, no dropping, so a - // long gap replays in full order instead of losing the oldest track events. + // A parked session keeps every message, in order, so a long gap replays in full. #[test] fn resume_queue_keeps_everything_in_order() { - let (dead, _dead_rx) = mpsc::channel(1); + let (dead, _dead_rx) = mpsc::unbounded_channel(); let context = context(dead); context.pause(); for n in 0..5_000 { - // `session_id` is just a cheap sequence marker here. + // `session_id` is a cheap sequence marker here. context.send_message(Message::Ready { resumed: false, session_id: n.to_string(), }); } - let (fresh, _fresh_rx) = mpsc::channel(1); + let (fresh, _fresh_rx) = mpsc::unbounded_channel(); let (_epoch, backlog) = context.resume_with(fresh); let replayed: Vec<_> = backlog @@ -379,56 +326,54 @@ mod tests { assert_eq!(replayed[4_999], "4999"); } - // A paused session queues player updates like upstream instead of dropping - // them: the replay backlog carries the whole thing, oldest first. + // Everything queues while parked and replays oldest first, events and snapshots alike. #[test] - fn paused_session_queues_player_updates_with_events() { - let (dead, _dead_rx) = mpsc::channel(1); + fn paused_session_queues_and_replays_in_order() { + let (dead, _dead_rx) = mpsc::unbounded_channel(); let context = context(dead); context.pause(); context.send_message(player_update("1")); - context.send_message(Message::Ready { - resumed: false, - session_id: "kept".to_string(), - }); - - let (fresh, _fresh_rx) = mpsc::channel(1); - let (_epoch, mut backlog) = context.resume_with(fresh); - backlog.reverse(); - - match backlog.pop() { - Some(Message::PlayerUpdate { guild_id, .. }) => assert_eq!(guild_id, "1"), - other => panic!("expected the queued update first, got {other:?}"), - } - match backlog.pop() { - Some(Message::Ready { session_id, .. }) => assert_eq!(session_id, "kept"), - other => panic!("expected the queued event second, got {other:?}"), - } - assert!(backlog.is_empty(), "nothing else was queued"); - } + context.send_message(track_event("2")); + context.send_message(player_update("3")); - // A full live channel drops periodic player updates but never a track event: - // the event parks into the resume queue so it survives to the next resume. - #[tokio::test] - async fn important_events_survive_a_full_queue() { - // Capacity 1, never drained, so the channel is full after the first send. - let (live, _live_rx) = mpsc::channel(1); - let context = context(live); - context.send_message(player_update("fill")); // fills the single slot + let (fresh, _fresh_rx) = mpsc::unbounded_channel(); + let (_epoch, backlog) = context.resume_with(fresh); - context.send_message(player_update("dropped")); // full -> dropped - assert_eq!(context.dropped_updates(), 1); + let kinds: Vec<_> = backlog + .iter() + .map(|message| match message { + Message::PlayerUpdate { guild_id, .. } => format!("update:{guild_id}"), + Message::Event(_) => "event".to_string(), + other => panic!("unexpected replay: {other:?}"), + }) + .collect(); + assert_eq!(kinds, ["update:1", "event", "update:3"]); + } - context.send_message(track_event("kept")); // full -> parked, not dropped - assert_eq!(context.dropped_updates(), 1); + // A backed-up socket must not park the session or lose messages; delivery just lags. + #[test] + fn a_backed_up_socket_does_not_park_or_drop() { + let (live, mut live_rx) = mpsc::unbounded_channel(); + let context = context(live); - // The parked event is now in the resume queue, and the live sender was dropped so the write - // task would see the channel close and tear the socket down. - let (fresh, _fresh_rx) = mpsc::channel(1); - let (_epoch, backlog) = context.resume_with(fresh); + for n in 0..5_000 { + context.send_message(player_update(&n.to_string())); + context.send_message(track_event(&n.to_string())); + } assert!( - matches!(backlog.as_slice(), [Message::Event(_)]), - "the track event must survive a full queue, got {backlog:?}" + !context.is_paused(), + "a backed-up socket must not park the session" ); + + let mut updates = 0; + let mut events = 0; + while let Ok(message) = live_rx.try_recv() { + match message { + Message::PlayerUpdate { .. } => updates += 1, + Message::Event(_) => events += 1, + other => panic!("unexpected message: {other:?}"), + } + } + assert_eq!((updates, events), (5_000, 5_000), "nothing may be dropped"); } } diff --git a/src/session/player/mod.rs b/src/session/player/mod.rs index be06e2b..268eeb7 100644 --- a/src/session/player/mod.rs +++ b/src/session/player/mod.rs @@ -773,11 +773,16 @@ impl LavalinkPlayer { } } - /// Send a player update to the owning context, if it is still alive. + /// Send a player update to the owning context, unless the session is parked for resume. pub fn send_player_update(&self) { - if let Some(context) = self.shared.context.upgrade() { - context.send_message(self.player_update_message()); + let Some(context) = self.shared.context.upgrade() else { + return; + }; + // Snapshots taken while parked are stale; the resume handshake sends fresh ones. + if context.is_paused() { + return; } + context.send_message(self.player_update_message()); } // Arm this player's periodic update timer, phased to now. @@ -791,7 +796,6 @@ impl LavalinkPlayer { let weak = Arc::downgrade(self); let task = self.shared.runtime.spawn(async move { let mut ticker = tokio::time::interval(interval); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); loop { ticker.tick().await; match weak.upgrade() { @@ -956,13 +960,13 @@ mod tests { async fn test_player_with_channel() -> ( Arc, - tokio::sync::mpsc::Receiver, + tokio::sync::mpsc::UnboundedReceiver, // Kept alive: the player holds only a `Weak` context, and dropping this // would close the channel the failure event is asserted on. Arc, ) { let manager = player::AudioPlayerManager::new(); - let (sender, rx) = tokio::sync::mpsc::channel(256); + let (sender, rx) = tokio::sync::mpsc::unbounded_channel(); let context = SocketContext::new( "test-session".to_string(), 1, @@ -1010,6 +1014,20 @@ mod tests { player.clear_voice_claim(&new); } + #[tokio::test] + async fn player_update_is_skipped_while_parked() { + let (player, _rx, context) = test_player_with_channel().await; + context.pause(); + player.send_player_update(); + + let (sender, _sender_rx) = tokio::sync::mpsc::unbounded_channel(); + let (_epoch, backlog) = context.resume_with(sender); + assert!( + backlog.is_empty(), + "a parked session must not queue stale updates, got {backlog:?}" + ); + } + #[tokio::test] async fn commit_allowed_matrix() { let player = test_player().await; From 871bdee18a8e69e9b528c0efd95be9c5e523120f Mon Sep 17 00:00:00 2001 From: appujet Date: Sun, 18 Oct 2026 20:08:44 +0530 Subject: [PATCH 6/7] Upgrade voice dependency to v0.1.8 --- Cargo.lock | 4 ++-- Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 76b3dbd..140ec37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3626,8 +3626,8 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" -version = "0.1.7" -source = "git+https://github.com/bongo-devs/voice?tag=v0.1.7#7a6ad8fcc4b71f97e8ffa21b8606022e545e6594" +version = "0.1.8" +source = "git+https://github.com/bongo-devs/voice?tag=v0.1.8#d2d60013116d10f4def26a2315a9ad1035680b02" dependencies = [ "aes-gcm", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 589ed03..52eb413 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ tag = "v0.1.27" [dependencies.voice] git = "https://github.com/bongo-devs/voice" -tag = "v0.1.7" +tag = "v0.1.8" [dependencies.reqwest] version = "0.12" From 06f38668811dc927eca78d901d722d84696464eb Mon Sep 17 00:00:00 2001 From: appujet Date: Mon, 19 Oct 2026 00:02:33 +0530 Subject: [PATCH 7/7] bump voice to v0.1.9 --- Cargo.lock | 4 ++-- Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 140ec37..1b26819 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3626,8 +3626,8 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" -version = "0.1.8" -source = "git+https://github.com/bongo-devs/voice?tag=v0.1.8#d2d60013116d10f4def26a2315a9ad1035680b02" +version = "0.1.9" +source = "git+https://github.com/bongo-devs/voice?tag=v0.1.9#342a137339382aedd8de81a75ba4bad3bd0a985f" dependencies = [ "aes-gcm", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 52eb413..f3361a1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ tag = "v0.1.27" [dependencies.voice] git = "https://github.com/bongo-devs/voice" -tag = "v0.1.8" +tag = "v0.1.9" [dependencies.reqwest] version = "0.12"