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..795848e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -275,9 +275,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.8.0" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6651c9ed80effdc7db0ff72512157f901af5e3549e341e24b1dd4887d836d838" +checksum = "877bf67ac8e52449eb2f9e98a344ea163beb70dd1f9eebe3cc5357dcfeaee4df" dependencies = [ "find-msvc-tools", "shlex 2.0.1", @@ -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]] @@ -728,9 +728,9 @@ checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" [[package]] name = "find-msvc-tools" -version = "0.1.14" +version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484" +checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" [[package]] name = "fluvio-wasm-timer" @@ -1188,7 +1188,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots 1.0.9", + "webpki-roots 1.0.10", ] [[package]] @@ -1410,7 +1410,7 @@ dependencies = [ [[package]] name = "kairo" -version = "4.8.4" +version = "4.8.5" dependencies = [ "axum", "bytes", @@ -1686,12 +1686,13 @@ dependencies = [ [[package]] name = "libmimalloc-sys" -version = "0.1.49" +version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a45a52f43e1c16f667ccfe4dd8c85b7f7c204fd5e3bf46c5b0db9a5c3c0b8e9" +checksum = "667f4fec20f29dfc6bc7357c582d91796c169ad7e2fce709468aefeb2c099870" dependencies = [ "cc", "cty", + "libc", ] [[package]] @@ -1729,8 +1730,8 @@ 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" +source = "git+https://github.com/bongo-devs/lyrics?tag=v0.2.9#6d0ff1688ab48a6872d8fe225deb14ca891c540a" dependencies = [ "async-trait", "base64 0.22.1", @@ -1787,9 +1788,9 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "mimalloc" -version = "0.1.52" +version = "0.1.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d4139bb28d14ad1facf21d5eb8825051b326e172d216b39f6d31df53cc97862" +checksum = "e1ee66a4b64c74f4ef288bcbb9192ad9c3feaad75193129ac8509af543894fd8" dependencies = [ "libmimalloc-sys", ] @@ -1833,7 +1834,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]] @@ -1984,9 +1985,9 @@ dependencies = [ [[package]] name = "opus-rs" -version = "0.1.37" +version = "0.1.39" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1880f411a21320dcb546c8f38f05ffb75643362ee0235437487999ed543e26ae" +checksum = "d9d1d772e17346cc5d8cf4b2cc088ec26c9708227db4e6f80e1104cd0040052c" [[package]] name = "p256" @@ -2110,8 +2111,8 @@ dependencies = [ [[package]] name = "player" -version = "0.1.9" -source = "git+https://github.com/bongo-devs/player?tag=v0.1.9#9a7d6fbb0d69ae1422d1759e4b8311c843e2c0bf" +version = "0.1.11" +source = "git+https://github.com/bongo-devs/player?tag=v0.1.11#877ac4f31f2df0c661f298c3ef2fb1a2a6518918" dependencies = [ "aes", "base64 0.22.1", @@ -2277,7 +2278,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2499,7 +2500,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", - "webpki-roots 1.0.9", + "webpki-roots 1.0.10", ] [[package]] @@ -2606,9 +2607,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.15" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "ring", "rustls-pki-types", @@ -2845,8 +2846,8 @@ dependencies = [ [[package]] name = "sources" -version = "0.1.26" -source = "git+https://github.com/bongo-devs/sources?tag=v0.1.26#690c1cb736991eecd4bcef82367d33a296e72fcf" +version = "0.1.30" +source = "git+https://github.com/bongo-devs/sources?tag=v0.1.30#e5b405ec197c627dccc1b4265538d0a23271c12d" dependencies = [ "aes", "base64 0.22.1", @@ -3261,9 +3262,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.53.3" +version = "1.54.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" +checksum = "a0c168a474b95bfecfe908a65d4bdffc523144b649a74c09c286afb35d53e0bc" dependencies = [ "bytes", "libc", @@ -3626,8 +3627,8 @@ 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.10" +source = "git+https://github.com/bongo-devs/voice?tag=v0.1.10#c5ab04e51326eed67f0c08ce34e5ddf282530764" dependencies = [ "aes-gcm", "bytes", @@ -3759,14 +3760,14 @@ version = "0.26.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" dependencies = [ - "webpki-roots 1.0.9", + "webpki-roots 1.0.10", ] [[package]] name = "webpki-roots" -version = "1.0.9" +version = "1.0.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9" dependencies = [ "rustls-pki-types", ] @@ -3805,7 +3806,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..bd69e18 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." @@ -9,7 +9,6 @@ description = "Standalone Discord audio sending node, written in Rust. Drop-in L bytes = "1" futures-util = "0.3" libc = "0.2" -mimalloc = "0.1.52" perf_monitor = "0.2.1" rand = "0.8" semver = "1" @@ -21,25 +20,29 @@ 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.11" [dependencies.sources] git = "https://github.com/bongo-devs/sources" -tag = "v0.1.26" +tag = "v0.1.30" [dependencies.voice] git = "https://github.com/bongo-devs/voice" -tag = "v0.1.4" +tag = "v0.1.10" [dependencies.reqwest] version = "0.12" default-features = false features = ["rustls-tls", "json"] +[dependencies.mimalloc] +version = "0.1.48" +features = ["v3"] + [dependencies.tokio] version = "1.53" features = ["tracing", "macros", "signal", "rt-multi-thread"] @@ -63,7 +66,7 @@ version = "0.1" features = ["server-auto", "tokio"] [dependencies.libmimalloc-sys] -version = "0.1.49" +version = "=0.1.44" features = ["extended"] [dependencies.rustls] @@ -79,6 +82,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/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/config.example.toml b/config.example.toml index afbef2c..811b37e 100644 --- a/config.example.toml +++ b/config.example.toml @@ -27,15 +27,6 @@ playerUpdateInterval = 5 # seconds between playerUpdate messages [lavalink.server.timeouts] connectTimeoutMs = 3000 # TCP+TLS connect timeout; 0 = OS default -# Discord voice handshake throttling. A large bot reconnecting 1000+ guilds -# at once would otherwise open that many simultaneous TLS+UDP handshakes and -# starve search/playback HTTP. PATCH answers immediately; connects drain in -# the background through maxConcurrentHandshakes slots. -# [lavalink.server.voice] -# maxConcurrentHandshakes = 32 # live handshakes; rest park (no worker) until a slot frees -# queueWarnMs = 15000 # warn when a queued connect waits past this; the wait itself is unbounded -# handshakeTimeoutMs = 30000 # outer bound per handshake; backstops the voice crate's own 30s abort -# backgroundConnect = false # default; true answers PATCH immediately and connects async # Outbound IP rotation. Binds each request to an address from these CIDR blocks and # rotates away from rate-limited ones. A bare address counts as a /32 or /128. # [lavalink.server.ratelimit] @@ -182,6 +173,47 @@ consumerSecret = "bd8a07e9f23fbe9d808646b730f89b8e" enabled = false searchLimit = 10 +[sources.instagram] # instagram.com/p/..., /reel/... and /reels/... public pages +# Resolves public page metadata only; private and login-gated posts are unsupported. +enabled = false + +[sources.audius] # audius.co track URLs + ausearch: +# Uses Audius's public read API and streams accessible tracks. +enabled = false +allowSearch = true +appName = "kairo" + +[sources.bilibili] # bilibili.com/video/BV... or /video/av... public pages +# Progressive muxed MP4 only; separate DASH audio/video and multipart streams are unsupported. +enabled = false + +[sources.reddit] # reddit.com comments URLs and redd.it post URLs +# Direct public audio links only; Reddit segmented video is unsupported. +enabled = false + +[sources.telegram] # t.me// public channel posts +# Direct public media files only; private channels and manifests are unsupported. +enabled = false + +[sources.tiktok] # tiktok.com/@user/video/ public video pages +enabled = false + +[sources.twitter] # x.com//status/ public video posts +enabled = false + +[sources.vimeo] # vimeo.com/ public video pages +# Requires a public progressive MP4; private links and HLS-only videos are unsupported. +enabled = false + +[sources.vkmusic] # vk.com/audio... URLs + vksearch: +enabled = false +allowSearch = false +# accessToken = "" # explicitly supplied VK user token; required for track lookup/playback + +[sources.yandexmusic] # music.yandex.ru/album//track/ +enabled = false +# token = "" # explicitly supplied OAuth token; required for full playback + [sources.tidal] # tidal.com URLs + tdsearch: / tdrec: # Metadata only since its streams are DRM-gated, plays through sources.mirror. # The public API token is scraped from the web bundle. 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/lavalink.rs b/src/config/lavalink.rs index 92502d0..802514c 100644 --- a/src/config/lavalink.rs +++ b/src/config/lavalink.rs @@ -46,8 +46,6 @@ pub struct LavalinkServerConfig { pub http_config: HttpConfig, /// Outbound HTTP timeouts, `lavalink.server.timeouts.*`. pub timeouts: TimeoutsConfig, - /// Discord voice handshake throttling, `lavalink.server.voice.*`. - pub voice: VoiceConfig, } impl Default for LavalinkServerConfig { @@ -67,7 +65,6 @@ impl Default for LavalinkServerConfig { ratelimit: RatelimitConfig::default(), http_config: HttpConfig::default(), timeouts: TimeoutsConfig::default(), - voice: VoiceConfig::default(), } } } @@ -223,55 +220,6 @@ impl TimeoutsConfig { } } -/// `lavalink.server.voice.*`, Discord voice handshake throttling. -/// -/// A large bot reconnecting 1000+ guilds at once would otherwise open that many -/// simultaneous TLS + UDP handshakes. Every value here has a documented default; -/// there are no other tuning constants for this path. -#[derive(Debug, Clone, Deserialize)] -#[serde(default, rename_all = "camelCase")] -pub struct VoiceConfig { - /// Max concurrent gateway + UDP handshakes. Past this, background connects - /// park on the gate (no worker, almost no memory) until a slot frees. - pub max_concurrent_handshakes: usize, - /// Queue wait that triggers a warn log on acquire. Background waits are - /// unbounded by design (a parked wait holds no worker); this is the warn - /// threshold in milliseconds, not a failure timeout. - pub queue_warn_ms: u64, - /// Outer bound on one handshake. The voice crate aborts its inner exchange - /// after 30 s, so this backstops at the same point rather than killing a - /// handshake the crate would have completed (DAVE keygen can run long under - /// a reconnect storm). - pub handshake_timeout_ms: u64, - /// Connect in the background and answer PATCH immediately (`true`), or block - /// the PATCH on the handshake as Lavalink historically does (`false`, with - /// the pre-change failure mapping: handshake errors return `500`). - pub background_connect: bool, -} - -impl Default for VoiceConfig { - fn default() -> Self { - Self { - max_concurrent_handshakes: 32, - queue_warn_ms: 15_000, - handshake_timeout_ms: 30_000, - background_connect: false, - } - } -} - -impl VoiceConfig { - /// The handshake timeout as a [`Duration`](std::time::Duration). - pub fn handshake_timeout(&self) -> std::time::Duration { - std::time::Duration::from_millis(self.handshake_timeout_ms.max(1)) - } - - /// The queue-warn interval as a [`Duration`](std::time::Duration). - pub fn queue_warn(&self) -> std::time::Duration { - std::time::Duration::from_millis(self.queue_warn_ms.max(1_000)) - } -} - impl LavalinkServerConfig { /// The names of the filters disabled in config. pub fn disabled_filters(&self) -> Vec { diff --git a/src/config/mod.rs b/src/config/mod.rs index 3dc334d..6c21934 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -13,7 +13,7 @@ pub use crossfade::{CrossfadeConfig, CrossfadeCurve}; pub use filters::FiltersToggleConfig; pub use lavalink::{ HttpConfig, LavalinkConfig, LavalinkServerConfig, RatelimitConfig, RatelimitStrategy, - ResamplingQuality, VoiceConfig, + ResamplingQuality, }; pub use logging::{LogFileConfig, LogFormat, LogRotation, LoggingConfig, RequestLoggingConfig}; pub use lyrics::LyricsServerConfig; @@ -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..1a2686d 100644 --- a/src/node/state.rs +++ b/src/node/state.rs @@ -103,7 +103,6 @@ impl AppState { crate::session::player::set_update_interval( state.inner.config.lavalink.server.player_update_interval, ); - crate::session::voice_gate::configure(&state.inner.config.lavalink.server.voice); state.spawn_player_cleanup_task(); state.spawn_resource_monitor_task(); crate::node::runtime_metrics::spawn_runtime_metrics_task(); @@ -162,15 +161,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 || { @@ -211,7 +207,7 @@ impl AppState { ); } - // Log file descriptors plus voice-gate state every 30 s. + // Log file descriptors plus player counts every 30 s. // // This is the gauge that decides the FD-exhaustion theory: if `fd_count` sits // far below `fd_limit` while HTTP still fails, descriptors were never the cause. @@ -225,19 +221,11 @@ impl AppState { .await .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(), playing = state.sockets().total_playing_players(), - gate_in_flight = gate.in_flight, - gate_waiting = gate.waiting, - gate_available = gate.available, - gate_acquired = gate.total_acquired, - gate_timed_out = gate.total_timed_out, - gate_avg_wait_ms = gate.avg_wait_ms, - gate_avg_handshake_ms = gate.avg_handshake_ms, "resource monitor" ); } 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/protocol/load_result.rs b/src/protocol/load_result.rs index 2ee412b..b402f7f 100644 --- a/src/protocol/load_result.rs +++ b/src/protocol/load_result.rs @@ -88,18 +88,19 @@ pub struct Exception { pub severity: Severity, /// The technical cause. pub cause: String, - /// The cause stack trace, which may be the same text as `cause`. + /// The captured Rust backtrace for the exception. pub cause_stack_trace: String, } impl Exception { pub fn from_friendly(error: &player::FriendlyException) -> Self { - let cause = error.cause.clone().unwrap_or_else(|| error.message.clone()); + let cause = error.cause.clone().unwrap_or_else(|| error.to_string()); + let cause_stack_trace = error.cause_stack_trace.clone(); Self { message: Some(error.message.clone()), severity: Severity::from_player(error.severity), - cause: cause.clone(), - cause_stack_trace: cause, + cause, + cause_stack_trace, } } } diff --git a/src/rest/metrics.rs b/src/rest/metrics.rs index 380d014..d89d49a 100644 --- a/src/rest/metrics.rs +++ b/src/rest/metrics.rs @@ -85,84 +85,10 @@ fn render(stats: &Stats) -> String { &format!("{cpu_help} (LL Load)"), stats.cpu.lavalink_load, ); - voice_gate_metrics(&mut out); runtime_metrics(&mut out); out } -// Voice-handshake gate families (`kairo_voice_*`). Counters are monotonic; -// the wait/handshake distributions are histograms so `histogram_quantile` -// works without the lossy lifetime averages in logs. -fn voice_gate_metrics(out: &mut String) { - let gate = crate::session::voice_gate::metrics(); - counter( - out, - "kairo_voice_handshakes_total", - "Total voice handshakes admitted by the gate.", - gate.total_acquired as f64, - ); - counter( - out, - "kairo_voice_handshake_queue_timeouts_total", - "Foreground gate acquires that hit the queue wait (background storm path waits unbounded instead).", - gate.total_timed_out as f64, - ); - counter( - out, - "kairo_voice_handshake_success_total", - "Background connects that committed a live connection.", - gate.total_success as f64, - ); - counter( - out, - "kairo_voice_handshake_timeout_total", - "Background connects that exhausted retries on handshake timeout.", - gate.total_handshake_timeout as f64, - ); - counter( - out, - "kairo_voice_handshake_error_total", - "Background connects that exhausted retries on handshake error.", - gate.total_handshake_error as f64, - ); - counter( - out, - "kairo_voice_handshake_stale_dropped_total", - "Finished handshakes dropped because a newer PATCH claimed the guild.", - gate.total_stale_dropped as f64, - ); - counter( - out, - "kairo_voice_handshake_retries_total", - "Retry attempts past the first handshake attempt.", - gate.total_retries as f64, - ); - gauge( - out, - "kairo_voice_handshakes_in_flight", - "Voice handshakes currently holding a gate permit.", - gate.in_flight as f64, - ); - gauge( - out, - "kairo_voice_handshake_queue_waiting", - "Voice connects currently parked in the gate queue.", - gate.waiting as f64, - ); - histogram_ms( - out, - "kairo_voice_queue_wait", - "Gate queue wait distribution in milliseconds.", - &crate::session::voice_gate::wait_histogram(), - ); - histogram_ms( - out, - "kairo_voice_handshake_duration", - "Handshake permit-hold distribution in milliseconds.", - &crate::session::voice_gate::handshake_histogram(), - ); -} - // Tokio runtime families (`kairo_runtime_*`), stable metrics only: per-worker // cumulative busy seconds (use `rate()` for busy %), alive tasks, global queue. // Blocking-pool gauges need `--cfg tokio_unstable` (see `runtime_metrics`) and @@ -214,32 +140,3 @@ fn gauge(out: &mut String, name: &str, help: &str, value: f64) { let _ = writeln!(out, "# TYPE {name} gauge"); let _ = writeln!(out, "{name} {value}"); } - -// Append one counter family. -fn counter(out: &mut String, name: &str, help: &str, value: f64) { - let _ = writeln!(out, "# HELP {name} {help}"); - let _ = writeln!(out, "# TYPE {name} counter"); - let _ = writeln!(out, "{name} {value}"); -} - -// Append one histogram family (cumulative buckets, `+Inf`, count, sum) from -// `voice_gate` bucket data. Bucket bounds and observations are milliseconds. -fn histogram_ms( - out: &mut String, - name: &str, - help: &str, - data: &crate::session::voice_gate::HistogramData, -) { - let _ = writeln!(out, "# HELP {name} {help}"); - let _ = writeln!(out, "# TYPE {name} histogram"); - let mut cumulative = 0u64; - for (bound, count) in data.buckets.iter().zip(data.counts.iter()) { - cumulative += count; - let _ = writeln!(out, "{name}_bucket{{le=\"{bound}\"}} {cumulative}"); - } - // Any observation past the last bound plus the explicit trailing slot. - cumulative += data.counts.get(data.buckets.len()).copied().unwrap_or(0); - let _ = writeln!(out, "{name}_bucket{{le=\"+Inf\"}} {cumulative}"); - let _ = writeln!(out, "{name}_count {}", data.total); - let _ = writeln!(out, "{name}_sum {}", data.sum_ms); -} diff --git a/src/rest/players.rs b/src/rest/players.rs index e5e8617..711fe5d 100644 --- a/src/rest/players.rs +++ b/src/rest/players.rs @@ -14,7 +14,6 @@ use crate::protocol::omissible::Omissible; use crate::protocol::player::{Player, PlayerUpdate}; use crate::protocol::track::Track; use crate::rest::error::{RestError, RestResult}; -use crate::session::player::{LavalinkPlayer, VoiceClaimGuard}; use crate::session::SocketContext; fn session(state: &AppState, session_id: &str) -> RestResult> { @@ -176,47 +175,13 @@ pub async fn patch_player( None }; - // Voice connects in the background and the PATCH answers immediately. Awaiting - // 1000 handshakes inline would fail the queue tail with 503s that non-retrying - // clients never recover from; parked background tasks hold no worker and drain - // through the gate instead. Trade-off: a PATCH carrying voice + track may start - // playback before the connection is ready. Frames wait in the buffer meanwhile - // and the client learns the outcome from player updates: `connected: true` on - // success (sent by the commit and the gateway-ready event), a - // `WebSocketClosedEvent` (1006) on final failure after retries. Voice - // validation errors stay synchronous (`400` below). Clients should treat a - // `connected: false` that never flips as a failed connect and re-PATCH. - // `backgroundConnect: false` restores the blocking path (handshake errors - // return `500` from this PATCH as before). + // Connect inline, as Lavalink does: this PATCH waits for the handshake and reports its + // failure as `500`. `voice_reconnect` dedupes the same target, so a repeated + // `VOICE_SERVER_UPDATE` for an already-connected guild costs no handshake. Validation + // errors above stay synchronous (`400`). match update.voice { - Omissible::Present(voice) - if player.voice_reconnect(&voice) && player.try_claim_voice(&voice) => - { - let voice_cfg = &state.config().lavalink.server.voice; - let handshake_timeout = voice_cfg.handshake_timeout(); - if voice_cfg.background_connect { - let task_player = Arc::clone(&player); - let hs_player = Arc::clone(&task_player); - let fut_voice = voice.clone(); - let guard = VoiceClaimGuard::new(task_player, voice); - tokio::spawn(async move { - LavalinkPlayer::background_connect(guard, handshake_timeout, || { - hs_player.handshake_voice(&fut_voice) - }) - .await; - }); - } else { - let fut_voice = voice.clone(); - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice); - let hs_player = Arc::clone(&player); - LavalinkPlayer::connect_voice_sync( - guard, - voice_cfg.queue_warn(), - handshake_timeout, - || hs_player.handshake_voice(&fut_voice), - ) - .await?; - } + Omissible::Present(voice) if player.voice_reconnect(&voice) => { + player.connect_voice(voice).await?; } _ => {} } @@ -375,16 +340,14 @@ async fn resolve_track( #[cfg(test)] mod tests { use super::*; - use std::time::Duration; use axum::extract::{Path, Query}; use crate::config::Config; - use crate::protocol::message::{EmittedEvent, Message}; + use crate::protocol::message::Message; use crate::protocol::player::VoiceState; const SESSION: &str = "storm-paths-session"; - const GUILD_BG: u64 = 424242424242424242; const GUILD_SYNC: u64 = 434343434343434343; // Voice against loopback with nothing listening: TCP refuses instantly, so @@ -405,17 +368,14 @@ mod tests { } } - async fn patched_state( - background_connect: bool, - ) -> (AppState, tokio::sync::mpsc::Receiver) { + async fn patched_state() -> (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 // 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, @@ -447,44 +407,15 @@ mod tests { .unwrap() } - // Background path: PATCH answers Ok immediately; the refused handshake - // retries in the background and the session observes a close event. - #[tokio::test] - async fn background_path_accepts_then_reports_close_event() { - let (state, mut rx) = patched_state(true).await; - let result = patch_voice(&state, GUILD_BG).await; - assert!(result.is_ok(), "background PATCH answers immediately"); - - let event = tokio::time::timeout(Duration::from_secs(30), rx.recv()) - .await - .expect("close event arrives") - .expect("channel open"); - match event { - Message::Event(boxed) => match *boxed { - EmittedEvent::WebSocketClosed { - code, - by_remote, - guild_id, - .. - } => { - assert_eq!(code, 1006); - assert!(by_remote); - assert_eq!(guild_id, GUILD_BG.to_string()); - } - other => panic!("expected WebSocketClosed, got {other:?}"), - }, - other => panic!("expected event message, got {other:?}"), - } - assert!(!player_of(&state, GUILD_BG).claim_is(&refused_voice())); - } - - // Synchronous fallback: the same refused handshake fails the PATCH itself. + // Voice connects inline on the PATCH, so the refused handshake fails the request itself: + // the same mapping Lavalink has (its handler throws INTERNAL_SERVER_ERROR). #[tokio::test] - async fn sync_path_returns_handshake_error() { - let (state, _rx) = patched_state(false).await; + async fn patch_voice_returns_handshake_error() { + let (state, _rx) = patched_state().await; let result = patch_voice(&state, GUILD_SYNC).await; - let err = result.expect_err("sync PATCH surfaces the handshake failure"); + let err = result.expect_err("the PATCH surfaces the handshake failure"); assert_eq!(err.status, StatusCode::INTERNAL_SERVER_ERROR); - assert!(!player_of(&state, GUILD_SYNC).claim_is(&refused_voice())); + // Nothing was stored, so a retrying PATCH still wants this target. + assert!(player_of(&state, GUILD_SYNC).voice_reconnect(&refused_voice())); } } diff --git a/src/rest/ws/mod.rs b/src/rest/ws/mod.rs index 7d7b629..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,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, + 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 ba772b8..7355494 100644 --- a/src/rest/ws/sender.rs +++ b/src/rest/ws/sender.rs @@ -1,59 +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 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, +) where + S: Sink + Unpin, +{ for message in backlog { if !send_text(&mut sink, &message).await { + tracing::warn!(session = %session_id, "websocket write failed, closing the socket"); return; } } - 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 }; - if !send_text(&mut sink, &message).await { - return; - } - } - _ = ping.tick() => { - let ping = sink.send(WsMessage::Ping(Vec::new().into())); - if !matches!(tokio::time::timeout(SEND_TIMEOUT, ping).await, Ok(Ok(()))) { - return; - } - } + 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"); } -async fn send_text(sink: &mut Sink, message: &Message) -> bool { +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 }; - let send = sink.send(WsMessage::Text(text.into())); - matches!(tokio::time::timeout(SEND_TIMEOUT, send).await, Ok(Ok(()))) + 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(); + } + } + } + + 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(())) + } + } + + fn ready(session_id: &str) -> Message { + Message::Ready { + resumed: false, + session_id: session_id.to_string(), + } + } + + 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 0078905..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,41 +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; - } - 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); @@ -220,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(), @@ -247,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)) { @@ -267,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); @@ -310,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, @@ -334,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, @@ -344,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 @@ -374,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/mod.rs b/src/session/mod.rs index 1ae1365..1fd581e 100644 --- a/src/session/mod.rs +++ b/src/session/mod.rs @@ -5,7 +5,6 @@ pub mod loss_counter; pub mod lyrics; pub mod manager; pub mod player; -pub mod voice_gate; pub use context::SocketContext; pub use loss_counter::AudioLossCounter; diff --git a/src/session/player/mod.rs b/src/session/player/mod.rs index be06e2b..91a7ef7 100644 --- a/src/session/player/mod.rs +++ b/src/session/player/mod.rs @@ -168,8 +168,12 @@ impl OpusFrameProvider for CountingProvider { } // Forwards voice gateway lifecycle to the WebSocket: a close becomes a `WebSocketClosedEvent`, and -// both a close and a ready are followed by a player update, so the client learns the new `connected` -// and `ping` right away instead of up to `playerUpdateInterval` seconds later. +// both a close and a ready are followed by a player update so the client sees the drop, or the +// gateway coming up, without waiting up to `playerUpdateInterval` seconds for the next tick. +// +// A ready fires inside the handshake, before `commit_voice` stores the connection, so that update +// still reads the previous slot (`connected: false` for a fresh guild). The connect's own result +// reaches the client through the PATCH response body, as it does in Lavalink. struct VoiceCloseForwarder { guild_id: u64, context: Weak, @@ -204,30 +208,6 @@ impl VoiceEventListener for VoiceCloseForwarder { } } -/// Releases a guild's handshake claim when dropped unless the claim moved on. -/// -/// The guard is built before the background task is spawned and moved into it, -/// so every exit releases the claim: success, handshake error, timeout, panic -/// unwind, even an abort before the first poll (dropping the task future drops -/// the guard). A retry with the same target therefore always runs a new -/// handshake instead of seeing a stale claim. -pub struct VoiceClaimGuard { - player: Arc, - voice: VoiceState, -} - -impl VoiceClaimGuard { - pub fn new(player: Arc, voice: VoiceState) -> Self { - Self { player, voice } - } -} - -impl Drop for VoiceClaimGuard { - fn drop(&mut self) { - self.player.clear_voice_claim(&self.voice); - } -} - /// A guild's player: the engine player, the voice connection, and protocol state. pub struct LavalinkPlayer { guild_id: u64, @@ -237,10 +217,6 @@ pub struct LavalinkPlayer { loss: Arc, voice: Mutex>, voice_state: Mutex>, - // Target of the handshake currently running in the background, if any. Duplicate - // VOICE_SERVER_UPDATE PATCHes for the same target skip the gate instead of each - // running a handshake. - voice_claim: Mutex>, // Serializes a whole `PATCH .../players/{guildId}` body. patch_lock: tokio::sync::Mutex<()>, filters: Mutex, @@ -315,7 +291,6 @@ impl LavalinkPlayer { end_marker: Mutex::new(None), lyrics, update_task: Mutex::new(None), - voice_claim: Mutex::new(None), }) } @@ -502,9 +477,9 @@ impl LavalinkPlayer { /// Whether `voice` differs from the current state or the connection dropped. /// - /// Cheap synchronous check used before paying for a handshake and again when - /// committing one. Both checks dedupe repeated `VOICE_SERVER_UPDATE` PATCHes - /// so an older handshake cannot displace a newer connection. + /// Cheap synchronous check used before paying for a handshake and again when committing one. + /// Both checks dedupe repeated `VOICE_SERVER_UPDATE` PATCHes, so a handshake that finished + /// against a superseded target cannot displace the newer connection. pub fn voice_reconnect(&self, voice: &VoiceState) -> bool { let current = self.voice_state.lock().unwrap(); match current.as_ref() { @@ -519,49 +494,10 @@ impl LavalinkPlayer { } } - /// Claim this guild's background handshake slot for `voice`. - /// - /// Returns `false` when a handshake for the same target is already running: - /// the duplicate PATCH skips the gate instead of running a second handshake. - /// A PATCH for a different target overwrites the claim; the older task sees - /// the mismatch on commit and drops its connection (see `commit_voice`). - pub fn try_claim_voice(&self, voice: &VoiceState) -> bool { - let mut claim = self.voice_claim.lock().unwrap_or_else(|e| e.into_inner()); - if claim.as_ref().is_some_and(|current| current == voice) { - return false; - } - *claim = Some(voice.clone()); - true - } - - /// Whether `voice` still owns this guild's handshake claim. - pub fn claim_is(&self, voice: &VoiceState) -> bool { - self.voice_claim - .lock() - .unwrap_or_else(|e| e.into_inner()) - .as_ref() - == Some(voice) - } - - /// Release the handshake claim when it still belongs to `voice`. - pub fn clear_voice_claim(&self, voice: &VoiceState) { - let mut claim = self.voice_claim.lock().unwrap_or_else(|e| e.into_inner()); - if claim.as_ref() == Some(voice) { - *claim = None; - } - } - - /// Whether a finished handshake for `voice` may be stored: the claim still - /// belongs to it and the guild still needs it. Split out so the decision - /// matrix is unit-testable without a live voice connection. - fn commit_allowed(&self, voice: &VoiceState) -> bool { - self.claim_is(voice) && self.voice_reconnect(voice) - } - /// Run the Discord gateway + UDP handshake. No player locks are held across the await. /// - /// The caller must hold a `voice_gate` permit while calling this. DAVE/MLS setup, - /// TLS and UDP discovery all happen here; the returned connection is not yet stored. + /// DAVE/MLS setup, TLS and UDP discovery all happen here; the returned connection is not yet + /// stored, and nothing gates how many of these run at once. pub async fn handshake_voice(&self, voice: &VoiceState) -> Result { let channel_id = voice .channel_id @@ -614,34 +550,12 @@ impl LavalinkPlayer { } } - /// Tell the client a background connect gave up: a `WebSocketClosedEvent` - /// with abnormal-closure code 1006, so existing clients observe the failure - /// without any protocol change. The claim releases via the task guard. - fn emit_connect_failed(&self, attempts: u32, cause: &str) { - tracing::warn!( - guild_id = self.guild_id, - attempts, - cause, - "voice background connect failed; emitting close event" - ); - if let Some(context) = self.shared.context.upgrade() { - context.send_message(Message::event(EmittedEvent::WebSocketClosed { - guild_id: self.guild_id.to_string(), - code: 1006, - reason: format!("voice handshake failed after {attempts} attempts ({cause})"), - by_remote: true, - })); - } - // The task guard releases the claim on return. - } - /// Store a finished handshake. The caller must hold `lock_patch`. /// - /// Drops `connection` and returns `false` when the claim moved on (a newer PATCH - /// claimed a different target) or the guild no longer needs this target, so a - /// stale handshake never displaces a fresh connection. + /// Drops `connection` and returns `false` when the guild no longer needs this target, so a + /// handshake that ran against a superseded voice state never displaces the live connection. pub fn commit_voice(&self, voice: VoiceState, connection: VoiceConnection) -> bool { - if !self.commit_allowed(&voice) { + if !self.voice_reconnect(&voice) { connection.disconnect(); return false; } @@ -654,116 +568,36 @@ impl LavalinkPlayer { true } - /// Maximum handshake attempts per background connect, then a final failure. - /// Three tries ride out a single blip (refused reset, TLS hiccup) without - /// turning every flap into a reconnect storm of its own. - const MAX_HANDSHAKE_ATTEMPTS: u32 = 3; + /// Outer bound on one voice handshake. + /// + /// TLS, UDP discovery and MLS keygen all happen inside `handshake_voice`, and the voice crate + /// bounds only its own WS connect and WS handshake (30 s each). Without an outer bound a + /// half-open socket would hang the PATCH waiting on it forever. + const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); - /// Drive one background voice connect to completion: gate, up to three - /// handshakes with exponential backoff, commit under the guild lock, then a - /// player update on success. + /// Connect `voice` and store the connection, inline on the calling PATCH. /// - /// Takes ownership of a [`VoiceClaimGuard`] built by the spawner: the claim - /// releases on every exit path, including an abort before the first poll. - /// The caller must have claimed the guild via [`try_claim_voice`](Self::try_claim_voice) - /// first. Handshakes come from the `make_handshake` factory (one fresh future - /// per attempt) so tests can run this without a network connection. - pub async fn background_connect( - guard: VoiceClaimGuard, - handshake_timeout: Duration, - mut make_handshake: impl FnMut() -> Fut + Send, - ) where - Fut: std::future::Future> + Send, - { - let player = Arc::clone(&guard.player); - let voice = guard.voice.clone(); - // Held to the end of the task: every `return` below, a panic unwind and a - // task abort all drop it, releasing the claim exactly when it is still ours. - let _guard = guard; - let mut attempt = 0u32; - let connection = loop { - attempt += 1; - // Scoped so the permit drops before any backoff sleep below: a - // sleeping retry must not hold a gate slot. - let result = { - let _permit = match super::voice_gate::acquire_background().await { - Ok(permit) => permit, - Err(err) => { - tracing::warn!( - guild_id = player.guild_id, - error = %err.message, - "voice background acquire failed" - ); - return; - } - }; - tokio::time::timeout(handshake_timeout, make_handshake()).await - }; - match result { - Ok(Ok(connection)) => break connection, - // `handshake_voice` already warned with the elapsed time. - Ok(Err(_)) if attempt < Self::MAX_HANDSHAKE_ATTEMPTS => { - super::voice_gate::record_retry(); - backoff_sleep(attempt).await; - } - Err(_) if attempt < Self::MAX_HANDSHAKE_ATTEMPTS => { - tracing::warn!( - guild_id = player.guild_id, - timeout_ms = handshake_timeout.as_millis(), - attempt, - "voice handshake timed out; backing off" - ); - super::voice_gate::record_retry(); - backoff_sleep(attempt).await; - } - Ok(Err(_)) => { - super::voice_gate::record_outcome(super::voice_gate::HandshakeOutcome::Error); - player.emit_connect_failed(attempt, "handshake error"); - return; - } + /// One attempt, no retry, no gate: this is Lavalink's shape, where the request that carries + /// the voice update waits for the connect and reports its failure. The guild lock is held only + /// for the commit, so a network handshake never blocks a concurrent PATCH. + /// + /// No player update is emitted on success: as in Lavalink, the PATCH response body is what + /// carries the new `connected`/`ping` state. + pub async fn connect_voice(&self, voice: VoiceState) -> Result { + let connection = + match tokio::time::timeout(Self::HANDSHAKE_TIMEOUT, self.handshake_voice(&voice)).await + { + Ok(Ok(connection)) => connection, + Ok(Err(err)) => return Err(err), Err(_) => { - super::voice_gate::record_outcome(super::voice_gate::HandshakeOutcome::Timeout); - player.emit_connect_failed(attempt, "handshake timeout"); - return; + return Err(RestError::internal(format!( + "voice handshake timed out after {} ms", + Self::HANDSHAKE_TIMEOUT.as_millis() + ))); } - } - }; - let _serialized = player.lock_patch().await; - if player.commit_voice(voice, connection) { - super::voice_gate::record_outcome(super::voice_gate::HandshakeOutcome::Success); - player.send_player_update(); - } else { - super::voice_gate::record_outcome(super::voice_gate::HandshakeOutcome::StaleDropped); - } - } - - /// Synchronous fallback for `backgroundConnect: false`: one gated handshake, - /// committed inline, errors returned to the PATCH caller as before. - pub async fn connect_voice_sync( - guard: VoiceClaimGuard, - queue_wait: Duration, - handshake_timeout: Duration, - make_handshake: impl FnOnce() -> Fut + Send, - ) -> Result - where - Fut: std::future::Future> + Send, - { - let player = Arc::clone(&guard.player); - let voice = guard.voice.clone(); - let _guard = guard; - let _permit = super::voice_gate::acquire(queue_wait).await?; - let connection = match tokio::time::timeout(handshake_timeout, make_handshake()).await { - Ok(Ok(connection)) => connection, - Ok(Err(err)) => return Err(err), - Err(_) => { - return Err(RestError::internal(format!( - "voice handshake timed out after {} ms", - handshake_timeout.as_millis() - ))); - } - }; - let _serialized = player.lock_patch().await; - Ok(player.commit_voice(voice, connection)) + }; + let _serialized = self.lock_patch().await; + Ok(self.commit_voice(voice, connection)) } pub fn player_update_message(&self) -> Message { @@ -773,11 +607,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 +630,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() { @@ -911,15 +749,6 @@ impl LavalinkPlayer { } } -/// Sleep between handshake retries: 500 ms doubling per attempt, capped at 5 s. -/// No gate permit is held across the sleep (see `background_connect`). -async fn backoff_sleep(failed_attempt: u32) { - let ms = 500u64 - .saturating_mul(1u64 << failed_attempt.min(4).saturating_sub(1)) - .min(5_000); - tokio::time::sleep(Duration::from_millis(ms)).await; -} - mod events; mod markers; mod transition; @@ -927,7 +756,6 @@ mod transition; #[cfg(test)] mod tests { use super::*; - use tokio::sync::Barrier; // `tokio::time::interval` panics on a zero period, so a `playerUpdateInterval: 0` in the config // has to be clamped before it reaches the update task. @@ -956,13 +784,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. + // would close the channel the test asserts 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, @@ -979,222 +807,31 @@ mod tests { ) } - fn boom() -> Result { - Err(RestError::internal("boom")) - } - #[tokio::test] - async fn duplicate_claim_for_same_target_returns_false() { - let player = test_player().await; - let voice = test_voice("a"); - assert!(player.try_claim_voice(&voice)); - // Same target while in flight: duplicate, skip the gate. - assert!(!player.try_claim_voice(&voice)); - // After release a retry with the same target claims anew: a new handshake runs. - player.clear_voice_claim(&voice); - assert!(player.try_claim_voice(&voice)); - player.clear_voice_claim(&voice); - } - - #[tokio::test] - async fn different_target_overwrites_and_release_keeps_newer() { - let player = test_player().await; - let old = test_voice("a"); - let new = test_voice("b"); - assert!(player.try_claim_voice(&old)); - assert!(player.try_claim_voice(&new)); - assert!(player.claim_is(&new)); - // Releasing the stale target must not clear the newer claim. - player.clear_voice_claim(&old); - assert!(player.claim_is(&new)); - player.clear_voice_claim(&new); - } - - #[tokio::test] - async fn commit_allowed_matrix() { - let player = test_player().await; - let voice = test_voice("a"); - let other = test_voice("b"); - // Fresh guild, claim mine: allowed. - assert!(player.try_claim_voice(&voice)); - assert!(player.commit_allowed(&voice)); - // Claim moved on: stale, not allowed. - assert!(player.try_claim_voice(&other)); - assert!(!player.commit_allowed(&voice)); - assert!(player.commit_allowed(&other)); - player.clear_voice_claim(&other); - } - - #[tokio::test] - async fn claim_guard_clears_on_drop_and_survives_panic() { - let player = test_player().await; - let voice = test_voice("a"); - assert!(player.try_claim_voice(&voice)); - { - let _guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - } - assert!(!player.claim_is(&voice)); - // A panic unwind drops the guard the same way. - assert!(player.try_claim_voice(&voice)); - let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let _guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - panic!("boom"); - })); - assert!(caught.is_err()); - assert!(!player.claim_is(&voice)); - } - - #[tokio::test] - async fn background_connect_clears_claim_on_handshake_error() { - let player = test_player().await; - let voice = test_voice("a"); - assert!(player.try_claim_voice(&voice)); - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - LavalinkPlayer::background_connect(guard, Duration::from_secs(5), || async { boom() }) - .await; - assert!(!player.claim_is(&voice)); - // Retry with the same target claims anew. - assert!(player.try_claim_voice(&voice)); - player.clear_voice_claim(&voice); - } - - #[tokio::test] - async fn background_connect_clears_claim_on_timeout() { - let player = test_player().await; - let voice = test_voice("a"); - assert!(player.try_claim_voice(&voice)); - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - LavalinkPlayer::background_connect(guard, Duration::from_millis(50), || { - std::future::pending() - }) - .await; - assert!(!player.claim_is(&voice)); + 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:?}" + ); } + // A PATCH that carries the voice update the guild already holds must not pay for a second + // handshake. Without a stored connection `voice_reconnect` has to stay true, or the first + // connect after a voice state arrives would be skipped and the guild would stay silent. #[tokio::test] - async fn background_connect_clears_claim_on_abort() { - let player = test_player().await; - let voice = test_voice("a"); - assert!(player.try_claim_voice(&voice)); - // The guard moves into the task future, so even an abort before the - // first poll drops it and releases the claim. - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - let handle = tokio::spawn(LavalinkPlayer::background_connect( - guard, - Duration::from_secs(30), - || async { - tokio::time::sleep(Duration::from_secs(60)).await; - boom() - }, - )); - handle.abort(); - let _ = handle.await; - assert!(!player.claim_is(&voice)); - } - - // Ten racers, one barrier, gated stub: exactly one handshake may start for - // the same target. The stub blocks on a watch channel (not a one-shot - // notify, which a retry would miss) so the winner is still in flight while - // the losers attempt their claims; it then fails to also prove the single - // winner retries to three attempts. - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn duplicate_patch_racers_run_one_handshake() { - use std::sync::atomic::{AtomicUsize, Ordering}; - use tokio::sync::watch; - static WINNERS: AtomicUsize = AtomicUsize::new(0); - static CALLS: AtomicUsize = AtomicUsize::new(0); - + async fn voice_reconnect() { let player = test_player().await; let voice = test_voice("a"); - let barrier = Arc::new(Barrier::new(10)); - let (release_tx, release_rx) = watch::channel(false); - let tasks: Vec<_> = (0..10) - .map(|_| { - let player = Arc::clone(&player); - let barrier = Arc::clone(&barrier); - let release_rx = release_rx.clone(); - let voice = voice.clone(); - tokio::spawn(async move { - barrier.wait().await; - if player.try_claim_voice(&voice) { - WINNERS.fetch_add(1, Ordering::AcqRel); - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - let stub = || { - let mut release_rx = release_rx.clone(); - async move { - CALLS.fetch_add(1, Ordering::AcqRel); - let _ = release_rx.wait_for(|open| *open).await; - boom() - } - }; - LavalinkPlayer::background_connect(guard, Duration::from_secs(30), stub) - .await; - } - }) - }) - .collect(); - // Wait for the single winner to start, let the losers attempt, then release. - for _ in 0..500 { - if WINNERS.load(Ordering::Relaxed) == 1 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - tokio::time::sleep(Duration::from_millis(200)).await; - let _ = release_tx.send(true); - for task in tasks { - task.await.unwrap(); - } - assert_eq!(WINNERS.load(Ordering::Relaxed), 1); - // One winner, three attempts (initial plus two retries). - assert_eq!(CALLS.load(Ordering::Relaxed), 3); - assert!(!player.claim_is(&voice)); - } - - // Failing stub, three attempts then give up: the factory ran exactly three - // times, a close event reached the session channel, and the claim cleared so - // the same target can retry afterwards. - #[tokio::test] - async fn background_connect_retries_then_reports_failure() { - use std::sync::atomic::{AtomicUsize, Ordering}; - - let (player, mut rx, _context) = test_player_with_channel().await; - let voice = test_voice("a"); - static ATTEMPTS: AtomicUsize = AtomicUsize::new(0); - - assert!(player.try_claim_voice(&voice)); - let guard = VoiceClaimGuard::new(Arc::clone(&player), voice.clone()); - LavalinkPlayer::background_connect(guard, Duration::from_secs(5), || async { - ATTEMPTS.fetch_add(1, Ordering::AcqRel); - boom() - }) - .await; - - assert_eq!(ATTEMPTS.load(Ordering::Relaxed), 3); - assert!(!player.claim_is(&voice)); - // Same target can retry afterwards. - assert!(player.try_claim_voice(&voice)); - player.clear_voice_claim(&voice); - - let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()) - .await - .expect("close event arrives") - .expect("channel open"); - match event { - Message::Event(boxed) => match *boxed { - EmittedEvent::WebSocketClosed { - code, - by_remote, - guild_id, - .. - } => { - assert_eq!(code, 1006); - assert!(by_remote); - assert_eq!(guild_id, "7"); - } - other => panic!("expected WebSocketClosed, got {other:?}"), - }, - other => panic!("expected event message, got {other:?}"), - } + assert!(player.voice_reconnect(&voice)); + *player.voice_state.lock().unwrap() = Some(voice.clone()); + // State matches but nothing is connected, so the connect still has to run. + assert!(player.voice_reconnect(&voice)); + assert!(player.voice_reconnect(&test_voice("b"))); } } diff --git a/src/session/voice_gate.rs b/src/session/voice_gate.rs deleted file mode 100644 index 12000a4..0000000 --- a/src/session/voice_gate.rs +++ /dev/null @@ -1,298 +0,0 @@ -//! Global gate for Discord voice handshakes. -//! -//! A large bot reconnecting 1000+ guilds at once would otherwise open that many -//! simultaneous TLS + UDP handshakes. That exhausts file descriptors and starves the -//! runtime so search and playback HTTP starts failing while voice alone looks -//! connected. Handshakes past the limit queue here. -//! -//! Tuning lives in `lavalink.server.voice.*` (`VoiceConfig`), applied once at -//! startup via [`configure`]. There are no other constants on this path. - -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, OnceLock}; -use std::time::{Duration, Instant}; - -use tokio::sync::{OwnedSemaphorePermit, Semaphore}; - -use crate::config::VoiceConfig; -use crate::rest::error::RestError; - -/// Defaults matching `VoiceConfig::default`, used until [`configure`] runs and by -/// tests that never configure the gate. -const DEFAULT_MAX_CONCURRENT: usize = 32; -const DEFAULT_QUEUE_WARN_MS: u64 = 15_000; - -/// Wait-duration buckets (ms) for the Prometheus histogram. -const WAIT_BUCKETS_MS: [u64; 9] = [10, 50, 100, 250, 500, 1000, 2500, 5000, 15000]; -/// Handshake-duration buckets (ms) for the Prometheus histogram. -const HANDSHAKE_BUCKETS_MS: [u64; 7] = [100, 250, 500, 1000, 2000, 5000, 10000]; - -static GATE: OnceLock> = OnceLock::new(); -static QUEUE_WARN_MS: AtomicU64 = AtomicU64::new(DEFAULT_QUEUE_WARN_MS); - -static IN_FLIGHT: AtomicU64 = AtomicU64::new(0); -static WAITING: AtomicU64 = AtomicU64::new(0); -static TOTAL_ACQUIRED: AtomicU64 = AtomicU64::new(0); -static TOTAL_TIMED_OUT: AtomicU64 = AtomicU64::new(0); -static TOTAL_WAIT_MS: AtomicU64 = AtomicU64::new(0); -static TOTAL_HANDSHAKE_MS: AtomicU64 = AtomicU64::new(0); -static TOTAL_SUCCESS: AtomicU64 = AtomicU64::new(0); -static TOTAL_HANDSHAKE_TIMEOUT: AtomicU64 = AtomicU64::new(0); -static TOTAL_HANDSHAKE_ERROR: AtomicU64 = AtomicU64::new(0); -static TOTAL_STALE_DROPPED: AtomicU64 = AtomicU64::new(0); -static TOTAL_RETRIES: AtomicU64 = AtomicU64::new(0); -static WAIT_COUNTS: OnceLock> = OnceLock::new(); -static HANDSHAKE_COUNTS: OnceLock> = OnceLock::new(); - -fn wait_counts() -> &'static [AtomicU64] { - WAIT_COUNTS.get_or_init(|| { - (0..WAIT_BUCKETS_MS.len() + 1) - .map(|_| AtomicU64::new(0)) - .collect() - }) -} - -fn handshake_counts() -> &'static [AtomicU64] { - HANDSHAKE_COUNTS.get_or_init(|| { - (0..HANDSHAKE_BUCKETS_MS.len() + 1) - .map(|_| AtomicU64::new(0)) - .collect() - }) -} - -fn gate() -> Arc { - Arc::clone(GATE.get_or_init(|| Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT)))) -} - -/// Apply `lavalink.server.voice.*`. Call once at startup before serving; later -/// calls are ignored because permits may already be checked out. -pub fn configure(config: &VoiceConfig) { - QUEUE_WARN_MS.store(config.queue_warn_ms.max(1_000), Ordering::Relaxed); - let _ = GATE.set(Arc::new(Semaphore::new( - config.max_concurrent_handshakes.max(1), - ))); -} - -fn observe(counts: &[AtomicU64], buckets: &[u64], value_ms: u64) { - let slot = buckets - .iter() - .position(|b| value_ms <= *b) - .unwrap_or(counts.len() - 1); - counts[slot].fetch_add(1, Ordering::Relaxed); -} - -/// A held handshake slot. Records handshake duration on drop. -pub struct HandshakePermit { - _permit: OwnedSemaphorePermit, - started: Instant, -} - -impl Drop for HandshakePermit { - fn drop(&mut self) { - let elapsed_ms = self.started.elapsed().as_millis() as u64; - TOTAL_HANDSHAKE_MS.fetch_add(elapsed_ms, Ordering::Relaxed); - observe(handshake_counts(), &HANDSHAKE_BUCKETS_MS, elapsed_ms); - IN_FLIGHT.fetch_sub(1, Ordering::Relaxed); - } -} - -/// Snapshot of gate state for logs and debug endpoints. -#[derive(Debug, Clone, Copy)] -pub struct VoiceGateMetrics { - pub in_flight: u64, - pub waiting: u64, - pub available: usize, - pub total_acquired: u64, - pub total_timed_out: u64, - pub avg_wait_ms: u64, - pub avg_handshake_ms: u64, - pub total_success: u64, - pub total_handshake_timeout: u64, - pub total_handshake_error: u64, - pub total_stale_dropped: u64, - pub total_retries: u64, -} - -/// Terminal outcome of one background connect, for the Prometheus counters. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum HandshakeOutcome { - Success, - Timeout, - Error, - StaleDropped, -} - -/// Record one terminal handshake outcome. -pub fn record_outcome(outcome: HandshakeOutcome) { - match outcome { - HandshakeOutcome::Success => TOTAL_SUCCESS.fetch_add(1, Ordering::Relaxed), - HandshakeOutcome::Timeout => TOTAL_HANDSHAKE_TIMEOUT.fetch_add(1, Ordering::Relaxed), - HandshakeOutcome::Error => TOTAL_HANDSHAKE_ERROR.fetch_add(1, Ordering::Relaxed), - HandshakeOutcome::StaleDropped => TOTAL_STALE_DROPPED.fetch_add(1, Ordering::Relaxed), - }; -} - -/// Record one retry (an attempt past the first). -pub fn record_retry() { - TOTAL_RETRIES.fetch_add(1, Ordering::Relaxed); -} - -/// Acquire a handshake slot, failing fast after `queue_wait`. -/// -/// For callers that must answer inline. The background storm path uses -/// [`acquire_background`] instead so a 1000-guild tail is queued, not failed. -pub async fn acquire(queue_wait: Duration) -> Result { - WAITING.fetch_add(1, Ordering::Relaxed); - let started = Instant::now(); - let permit = tokio::time::timeout(queue_wait, gate().acquire_owned()).await; - WAITING.fetch_sub(1, Ordering::Relaxed); - match permit { - Ok(Ok(permit)) => Ok(held(permit, started)), - Ok(Err(_)) => Err(RestError::internal("voice handshake limiter closed")), - Err(_) => { - TOTAL_TIMED_OUT.fetch_add(1, Ordering::Relaxed); - Err(RestError::service_unavailable( - "voice handshake queue full; retry with backoff", - )) - } - } -} - -/// Acquire a handshake slot with no queue timeout, for background connects. -/// -/// One fair FIFO wait: cancelling and re-queueing would lose queue position and -/// starve the tail under churn. A parked acquire holds no worker and almost no -/// memory, so a 1000-guild storm drains through the live handshakes instead of -/// failing its tail with 503s that non-retrying clients never recover from. -/// Waits past the configured warn threshold log once on acquire. Permits are -/// released by timeout-bounded handshakes, so the queue cannot stick. -pub async fn acquire_background() -> Result { - WAITING.fetch_add(1, Ordering::Relaxed); - let started = Instant::now(); - let permit = gate().acquire_owned().await; - WAITING.fetch_sub(1, Ordering::Relaxed); - match permit { - Ok(permit) => { - let wait_ms = started.elapsed().as_millis() as u64; - if wait_ms >= QUEUE_WARN_MS.load(Ordering::Relaxed) { - tracing::warn!( - wait_ms, - waiting = WAITING.load(Ordering::Relaxed), - "voice handshake waited past the warn threshold in the background queue" - ); - } - Ok(held(permit, started)) - } - Err(_) => Err(RestError::internal("voice handshake limiter closed")), - } -} - -fn held(permit: OwnedSemaphorePermit, queued_at: Instant) -> HandshakePermit { - let wait_ms = queued_at.elapsed().as_millis() as u64; - TOTAL_WAIT_MS.fetch_add(wait_ms, Ordering::Relaxed); - observe(wait_counts(), &WAIT_BUCKETS_MS, wait_ms); - TOTAL_ACQUIRED.fetch_add(1, Ordering::Relaxed); - IN_FLIGHT.fetch_add(1, Ordering::Relaxed); - HandshakePermit { - _permit: permit, - started: Instant::now(), - } -} - -/// Current gate state. -pub fn metrics() -> VoiceGateMetrics { - let acquired = TOTAL_ACQUIRED.load(Ordering::Relaxed); - VoiceGateMetrics { - in_flight: IN_FLIGHT.load(Ordering::Relaxed), - waiting: WAITING.load(Ordering::Relaxed), - available: gate().available_permits(), - total_acquired: acquired, - total_timed_out: TOTAL_TIMED_OUT.load(Ordering::Relaxed), - avg_wait_ms: TOTAL_WAIT_MS - .load(Ordering::Relaxed) - .checked_div(acquired) - .unwrap_or(0), - avg_handshake_ms: TOTAL_HANDSHAKE_MS - .load(Ordering::Relaxed) - .checked_div(acquired) - .unwrap_or(0), - total_success: TOTAL_SUCCESS.load(Ordering::Relaxed), - total_handshake_timeout: TOTAL_HANDSHAKE_TIMEOUT.load(Ordering::Relaxed), - total_handshake_error: TOTAL_HANDSHAKE_ERROR.load(Ordering::Relaxed), - total_stale_dropped: TOTAL_STALE_DROPPED.load(Ordering::Relaxed), - total_retries: TOTAL_RETRIES.load(Ordering::Relaxed), - } -} - -/// Histogram exposition data: bucket upper bounds, per-bucket counts, sum, total. -pub struct HistogramData { - pub buckets: &'static [u64], - pub counts: Vec, - pub sum_ms: u64, - pub total: u64, -} - -/// Queue-wait duration distribution. -pub fn wait_histogram() -> HistogramData { - HistogramData { - buckets: &WAIT_BUCKETS_MS, - counts: wait_counts() - .iter() - .map(|c| c.load(Ordering::Relaxed)) - .collect(), - sum_ms: TOTAL_WAIT_MS.load(Ordering::Relaxed), - total: TOTAL_ACQUIRED.load(Ordering::Relaxed), - } -} - -/// Handshake-duration distribution. -pub fn handshake_histogram() -> HistogramData { - HistogramData { - buckets: &HANDSHAKE_BUCKETS_MS, - counts: handshake_counts() - .iter() - .map(|c| c.load(Ordering::Relaxed)) - .collect(), - sum_ms: TOTAL_HANDSHAKE_MS.load(Ordering::Relaxed), - total: TOTAL_ACQUIRED.load(Ordering::Relaxed), - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::atomic::AtomicUsize; - - // 100 concurrent acquirers must never hold more than the cap at once, even - // though every one eventually gets a permit. Proves the gate bounds - // handshake concurrency under a reconnect storm. - #[tokio::test] - async fn concurrent_acquires_never_exceed_the_cap() { - static ACTIVE: AtomicUsize = AtomicUsize::new(0); - static PEAK: AtomicUsize = AtomicUsize::new(0); - - let cap = gate().available_permits() + IN_FLIGHT.load(Ordering::Relaxed) as usize; - let tasks: Vec<_> = (0..100) - .map(|_| { - tokio::spawn(async { - let _permit = acquire(Duration::from_secs(30)) - .await - .expect("gate stays open"); - let held = ACTIVE.fetch_add(1, Ordering::AcqRel) + 1; - PEAK.fetch_max(held, Ordering::Relaxed); - tokio::time::sleep(Duration::from_millis(5)).await; - ACTIVE.fetch_sub(1, Ordering::AcqRel); - }) - }) - .collect(); - for task in tasks { - task.await.unwrap(); - } - assert!( - PEAK.load(Ordering::Relaxed) <= cap.max(1), - "peak {} exceeded the observed cap {cap}", - PEAK.load(Ordering::Relaxed) - ); - assert_eq!(ACTIVE.load(Ordering::Relaxed), 0); - } -}