Fix/voice issue

.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

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.9"
 
 [dependencies.reqwest]
 version = "0.12"
@@ -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

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
 

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

src/config/mod.rs

@@ -71,15 +71,18 @@ 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)
             .or_else(|| std::env::var("KAIRO_CONFIG").ok())
             .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<Self, toml::de::Error> {
@@ -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"));
+    }
 }

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<SocketContext>) {
         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 || {
@@ -226,7 +223,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(),

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<Tasks> = 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<F, Fut>(&self, key: impl Into<String>, 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<F, Fut>(&self, key: impl Into<String>, period: Duration, handler: F)
     where
         F: FnMut() -> Fut + Send + 'static,
         Fut: Future<Output = ()> + 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<F, Fut>(&self, key: impl Into<String>, period: Duration, handler: F)
+    where
+        F: FnMut() -> Fut + Send + 'static,
+        Fut: Future<Output = ()> + Send + 'static,
+    {
+        self.spawn(key.into(), period, MissedTickBehavior::Burst, handler);
+    }
+
+    fn spawn<F, Fut>(
+        &self,
+        key: String,
+        period: Duration,
+        behavior: MissedTickBehavior,
+        mut handler: F,
+    ) where
+        F: FnMut() -> Fut + Send + 'static,
+        Fut: Future<Output = ()> + 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;

src/rest/players.rs

@@ -407,15 +407,15 @@ mod tests {
 
     async fn patched_state(
         background_connect: bool,
-    ) -> (AppState, tokio::sync::mpsc::Receiver<Message>) {
+    ) -> (AppState, tokio::sync::mpsc::UnboundedReceiver<Message>) {
         let mut config = Config::default();
         config.lavalink.server.voice.background_connect = background_connect;
         // Keep the process-global update interval deterministic: the unit test
         // for clamping assumes the first `set_update_interval` call wins with a
         // 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,

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<AppState>,
@@ -81,7 +77,7 @@ async fn handle_socket(
     user_agent: Option<String>,
 ) {
     let (sink, stream) = socket.split();
-    let (sender, receiver) = mpsc::channel::<Message>(OUTBOUND_CAPACITY);
+    let (sender, receiver) = mpsc::unbounded_channel::<Message>();
 
     let (context, resumed, epoch, backlog) =
         attach(&state, requested_session.as_deref(), user_id, sender);
@@ -90,16 +86,19 @@ 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.
-    let write_task = tokio::spawn(sender::write_loop(sink, receiver, backlog));
+    // Both halves are one socket: whichever ends first stops the other.
+    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;
 
     tracing::debug!(
         session = %context.session_id,
         ?reason,
-        dropped_updates = context.dropped_updates(),
         "websocket closing"
     );
     on_disconnect(&state, context, epoch);
@@ -129,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 => {
@@ -147,11 +146,10 @@ fn attach(
     state: &AppState,
     requested_session: Option<&str>,
     user_id: u64,
-    sender: mpsc::Sender<Message>,
+    sender: mpsc::UnboundedSender<Message>,
 ) -> (Arc<SocketContext>, bool, u64, Vec<Message>) {
     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);