diff --git a/PROGRESS.md b/PROGRESS.md index 4f87138..256f6a7 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -10,6 +10,44 @@ up instantly. Newest status at the top. ## ▶ Where we left off / next action +- **Done:** **Disconnect, timeout & keepalive system** (2026-06-18). Three reported bugs + traced to one root cause + two missing designed features, all fixed: + 1. **Stale users after disconnect + eternal PLC hiss** (root cause): `ConnSession::close()` + silently erased dropped users from the registry without broadcasting `UserEvent::LEFT`. + Peers never learned the user left → their user lists stayed stale AND their audio + engines never called `remove_stream` → Opus PLC synthesized comfort noise forever + (the "soft hissing that never goes away"). **Fix:** `SessionRegistry::broadcast_left()` + helper (mirrors `kick_user`'s first half); `ConnSession::close()` now broadcasts LEFT + before erasing. Regression test `test_disconnect_left` (`tests/`). + 2. **PLC cap** (defense-in-depth): `AudioEngine::on_playback` now caps pure-PLC at ~2 s + (`kPlcCapSamples`); after that it emits digital silence instead of more comfort noise, + so a stale stream can never hiss forever even if `remove_stream` is never called. Resets + automatically when fresh packets arrive. Test `test_plc_cap` (`tests/`). + 3. **No timeout / no ping** (missing feature): the client never sent `Ping`, the server had + no `last_seen` / reaper, and half-open connections (NAT timeout, wifi loss, sleep) left + ghost users forever. **Fix:** client sends `Ping` every 15 s from the io thread (RTT + measured from `Pong` nonce); `ConnSession::last_seen` bumped on every inbound TCP/UDP + frame; `asio::steady_timer` reaper sweeps every 15 s and drops sessions older than 45 s + (configurable via `server::Config::reaper_timeout_ms`/`reaper_sweep_ms`). Each drop + broadcasts LEFT via fix #1. Test `test_reaper_timeout` (`tests/`, 2 s timeout for fast + turnaround). + 4. **UDP KEEPALIVE** (missing feature): client sends a plaintext `KEEPALIVE` frame every + 5 s (`voice_frame.h::kFrameKeepalive`); server bumps `last_seen` + echoes back. Keeps + NAT bindings alive and lets media activity defer the reaper independently of TCP. + 5. **Graceful client disconnect:** `vc_disconnect()` now sends `Disconnect{code=0}` when + fully authenticated (`VC_STATE_CONNECTED`): it queues the envelope, sets a + `graceful_disconnect_pending_` flag, and joins the io thread — the io thread's + `drain_sends()` sends the Disconnect, sees the flag, sets `io_stop_`, and exits + naturally (its cleanup handles `teardown_voice()` + socket close). No main-thread + socket close → no double-close race. Server handles client-sent `Disconnect` + (`handle_client_disconnect` → `close()` → immediate LEFT broadcast, no reaper/EOF + wait). In non-authenticated states, the original force-close path runs (with TOFU + unblock). + - Docs updated: `docs/protocol.md §6/§7` (graceful disconnect, pinned N=3, reaper, last_seen), + `docs/voice.md §6` (KEEPALIVE plaintext + echo + last_seen), `docs/architecture.md §5` + (reaper timer), `server::Config` (reaper fields). `ctest --preset m2-dev` — **21/21 green** + (3 new tests: disconnect_left, plc_cap, reaper_timeout). + - **Done:** **Stereo screen-audio loopback capture on Windows** (2026-06-17). The WASAPI loopback path (`start_loopback_capture`) used to hardcode `cfg.capture.channels = 1`, downmixing the system's stereo mix to mono before the encoder ever saw it — so even on a @@ -141,8 +179,10 @@ up instantly. Newest status at the top. response-vs-broadcast contract in `docs/protocol.md` §6. Registry-level admin broadcasts (move/mute/kick/channel CRUD) already used `exclude=0` and were correct. `ctest --test-dir build/m1-dev` — **18/18 green** (PowerShell); Windows `VoiceCat.App` builds 0 warnings. - **Latent, not fixed:** the client never sends the `Ping` keepalive that `docs/protocol.md` §7 - describes (only the server answers pings) — unrelated to this bug, noted for later. + ~~**Latent, not fixed:** the client never sends the `Ping` keepalive that `docs/protocol.md` §7 + describes (only the server answers pings) — unrelated to this bug, noted for later.~~ + **Resolved** (2026-06-18): full keepalive/timeout/disconnect system implemented — see + "disconnect, timeout & keepalive" entry below. - **Done:** **Fixed a *second* silent-playback bug — the playout clock free-ran and drifted off the stream** (2026-06-17, reported live: both `vccli` and the Windows client showed `talking=1/0` correctly on VAD/PTT, mic + screen-share were recognized by peers, but nothing was audible). diff --git a/core/src/audio/audio_engine.cpp b/core/src/audio/audio_engine.cpp index c92ed8b..f86279e 100644 --- a/core/src/audio/audio_engine.cpp +++ b/core/src/audio/audio_engine.cpp @@ -26,6 +26,16 @@ int64_t now_ms() { constexpr int32_t kResyncAheadSamples = 48000 * 200 / 1000; // clock 200 ms ahead → re-seed constexpr int32_t kResyncBehindSamples = 48000 * 500 / 1000; // clock 500 ms behind → re-seed +// PLC cap: after this many consecutive samples of pure packet-loss concealment (no real +// packet decoded), stop calling opus_decode(nullptr,0,...) and emit silence instead. Opus +// PLC synthesizes soft comfort noise that never "exhausts" (opus_decode always returns +// frame_samples > 0 for PLC), so without a cap a stale stream left in the mixer after its +// source disconnects would hiss forever. 2 s bounds the hiss to a brief gap while still +// bridging normal network jitter/PTT silences. The primary fix for stale streams is the +// server's UserEvent::LEFT broadcast (which triggers remove_stream); this is +// defense-in-depth against any future regression that skips remove_stream. +constexpr int32_t kPlcCapSamples = 48000 * 2; // 2 s @ 48 kHz + #ifdef VOICECAT_HAS_AUDIO // device_id encoding (DeviceInfo::id / AudioParams::*_device_id): a hex string of the raw // ma_device_id bytes. Opaque on purpose — names aren't guaranteed unique, and this is the only @@ -459,8 +469,18 @@ void AudioEngine::on_playback(int16_t* out, ma_uint32 frames) { maybe_frame->payload.data(), static_cast(maybe_frame->payload.size()), stream.decode_scratch.data(), frame_samples); + stream.plc_samples_since_real = 0; // real packet — reset PLC streak + } else if (stream.plc_samples_since_real >= kPlcCapSamples) { + // PLC cap exhausted: emit silence instead of more comfort noise. Keeps the + // ring fed and the playout clock advancing so timing is correct if the + // source resumes, but bounds the hiss to ~2 s (kPlcCapSamples). + std::memset(stream.decode_scratch.data(), 0, + static_cast(frame_samples) * stream.ring_channels * + sizeof(int16_t)); + n = frame_samples; } else { n = stream.decoder.decode(nullptr, 0, stream.decode_scratch.data(), frame_samples); + if (n > 0) stream.plc_samples_since_real += n; // track PLC streak } if (n <= 0) break; // decoder error/exhausted PLC; rest of this period stays silent diff --git a/core/src/audio/audio_engine.h b/core/src/audio/audio_engine.h index 388f8f4..27dc2b6 100644 --- a/core/src/audio/audio_engine.h +++ b/core/src/audio/audio_engine.h @@ -317,6 +317,14 @@ class AudioEngine { // dropped/never-due (silent playback). false until the first frame seeds it (on_playback). bool playout_started = false; + // PLC cap (defense-in-depth): consecutive samples produced by packet-loss + // concealment since the last real decoded frame. Reset to 0 on every real frame. + // When it exceeds kPlcCapSamples (audio_engine.cpp), on_playback stops calling + // opus_decode(nullptr,0,...) and emits silence instead — bounding the comfort-noise + // hiss to ~2 s so a stale stream can never hiss forever even if remove_stream is + // never called. See on_playback's decode loop. + int64_t plc_samples_since_real = 0; + // M3: listener-chosen, local-only noise reduction (docs/voice.md §10). Lazily // created only when enabled — bounded by how many remote streams this listener // subscribes to, so no separate instance cap is needed. diff --git a/core/src/core/client.cpp b/core/src/core/client.cpp index 7cf5312..7f29772 100644 --- a/core/src/core/client.cpp +++ b/core/src/core/client.cpp @@ -102,6 +102,35 @@ vc_result vc_client::disconnect() { auto cur = state_net_.load(std::memory_order_acquire); if (cur == VC_STATE_DISCONNECTED && !io_thread_.joinable()) return VC_ERR_NOT_CONNECTED; + // Graceful disconnect (Tier 5): queue Disconnect{code=0}, set a flag, and let the io + // thread drain the queue + send it + exit naturally. The io thread's cleanup handles + // teardown_voice() and socket close — the main thread just joins. This avoids the + // double-close race that a send_cv_ wait + main-thread socket close would create + // (the server closes the connection on receipt of Disconnect, so the io thread exits + // while the main thread is still waiting, and both try to close the same socket). + // Only when fully authenticated: the server gates Disconnect handling on Authenticated, + // and in earlier states (TLS handshake, TOFU gate, auth pending) the io thread may be + // blocked outside the read loop (e.g. tofu_cv_) where the flag would never be checked. + if (cur == VC_STATE_CONNECTED && io_thread_.joinable()) { + voicecat::v1::Envelope env; + env.set_request_id(next_req_id_++); + auto* d = env.mutable_disconnect(); + d->set_code(0); // 0 = graceful client-initiated + d->set_reason("client disconnect"); + queue_envelope(env); + graceful_disconnect_pending_.store(true, std::memory_order_release); + + // Wait for the io thread to drain the queue, send the Disconnect, and exit. + // Its cleanup handles teardown_voice() + socket close. No main-thread socket + // close needed. If the io thread is stuck (unlikely), the caller can force-close + // by calling disconnect() again — but the second call hits the non-graceful path + // below since io_thread_ is no longer joinable after the join returns. + if (io_thread_.joinable()) io_thread_.join(); + return VC_OK; + } + + // Non-graceful path: force-close (original logic). Used when the io thread is already + // gone, the connection is in an early state, or as a fallback. io_stop_.store(true, std::memory_order_release); // Unblock a run_io() thread that's currently waiting on vc_confirm_server_identity() — @@ -285,9 +314,37 @@ void vc_client::run_io(std::string host, uint16_t port) { voicecat::protocol::FrameCodec codec; std::vector buf(16384); + // Seed the keepalive clock so the first Ping goes out ~15s after connect, + // not immediately. + last_ping_ms_.store(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(), + std::memory_order_release); + while (!io_stop_.load(std::memory_order_acquire)) { drain_sends(); + // Graceful disconnect: if disconnect() queued a Disconnect{code=0} and set + // the flag, drain_sends() just sent it. Exit the read loop now — the server + // will close the connection on receipt, but we don't need to wait for that. + // Setting io_stop_ prevents the emit_disconnected() call below the loop + // (this is a user-initiated exit, not an error). + if (graceful_disconnect_pending_.load(std::memory_order_acquire)) { + io_stop_.store(true, std::memory_order_release); + break; + } + + // Keepalive: send a Ping every ~15s so the server's reaper doesn't drop us + // (docs/protocol.md §7). The 50ms read timeout means this loop spins ~20×/s, + // plenty of resolution for a 15s interval. + { + auto now_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + if (now_ms - last_ping_ms_.load(std::memory_order_acquire) >= kPingIntervalMs) { + send_ping(); + drain_sends(); // flush the ping immediately + } + } + int n = tls_->read(buf.data(), buf.size()); if (voicecat::crypto::TlsContext::is_timeout_error(n)) continue; if (n <= 0) break; @@ -358,15 +415,22 @@ void vc_client::drain_sends() { std::vector frame; { std::lock_guard lk(send_mutex_); - if (send_queue_.empty()) return; + if (send_queue_.empty()) { + send_cv_.notify_all(); // unblock disconnect()'s drain wait + return; + } frame = std::move(send_queue_.front()); send_queue_.pop_front(); } - if (!tls_) return; + if (!tls_) { send_cv_.notify_all(); return; } size_t off = 0; while (off < frame.size()) { int n = tls_->write(frame.data() + off, frame.size() - off); - if (n <= 0) { io_stop_.store(true); return; } + if (n <= 0) { + io_stop_.store(true); + send_cv_.notify_all(); // unblock disconnect() even on write failure + return; + } off += static_cast(n); } } @@ -379,6 +443,21 @@ void vc_client::queue_envelope(const voicecat::v1::Envelope& env) { send_queue_.push_back(std::move(frame)); } +void vc_client::send_ping() { + auto now_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + last_ping_ms_.store(now_ms, std::memory_order_release); + uint64_t nonce = ping_nonce_.fetch_add(1, std::memory_order_relaxed); + { + std::lock_guard lk(ping_mutex_); + pending_pings_[nonce] = now_ms; + } + voicecat::v1::Envelope env; + env.set_request_id(next_req_id_++); + env.mutable_ping()->set_nonce(nonce); + queue_envelope(env); +} + // ── Protocol dispatch ───────────────────────────────────────────────────────── void vc_client::handle_envelope(const voicecat::v1::Envelope& env) { @@ -413,8 +492,19 @@ void vc_client::handle_envelope(const voicecat::v1::Envelope& env) { case voicecat::v1::Envelope::kStreamAnnounceResult: handle_stream_announce_result(env.request_id(), env.stream_announce_result()); break; - case voicecat::v1::Envelope::kPong: - break; // ignore keepalive responses + case voicecat::v1::Envelope::kPong: { + // Correlate the echoed nonce to measure RTT (docs/protocol.md §7). + uint64_t nonce = env.pong().nonce(); + auto now_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + std::lock_guard lk(ping_mutex_); + auto it = pending_pings_.find(nonce); + if (it != pending_pings_.end()) { + last_rtt_ms_.store(now_ms - it->second, std::memory_order_relaxed); + pending_pings_.erase(it); + } + break; + } case voicecat::v1::Envelope::kGenericResult: { vc_event ev{}; ev.type = VC_EVENT_GENERIC_RESULT; @@ -769,14 +859,29 @@ void vc_client::run_udp_recv() { setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); #endif + // Seed the keepalive clock so the first KEEPALIVE goes out ~5s after binding, not + // immediately (the UdpBinding bootstrap itself is a recent packet). + last_udp_keepalive_ms_.store(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(), + std::memory_order_release); + std::vector buf(2048); while (!udp_stop_.load(std::memory_order_acquire)) { int n = static_cast(::recv(static_cast(fd), reinterpret_cast(buf.data()), static_cast(buf.size()), 0)); - if (n < static_cast(voicecat::net::kVoiceHeaderSize)) continue; + if (n < static_cast(voicecat::net::kVoiceHeaderSize)) { + // Timeout or short packet — send a KEEPALIVE if the interval has elapsed. + send_udp_keepalive(); + continue; + } voicecat::net::VoiceFrame hdr{}; if (!voicecat::net::parse_header(buf.data(), static_cast(n), hdr)) continue; + if (hdr.type == voicecat::net::kFrameKeepalive) { + // Echoed keepalive from the server — media path is alive. (RTT measurement + // could be added here later by correlating a nonce; not needed for NAT/timeout.) + continue; + } if (hdr.type != voicecat::net::kFrameVoice) continue; if (!media_recv_crypto_) continue; @@ -797,6 +902,29 @@ void vc_client::run_udp_recv() { } } +void vc_client::send_udp_keepalive() { + int fd = udp_fd_.load(std::memory_order_acquire); + if (fd == -1) return; + + auto now_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + if (now_ms - last_udp_keepalive_ms_.load(std::memory_order_acquire) < kUdpKeepaliveIntervalMs) + return; + last_udp_keepalive_ms_.store(now_ms, std::memory_order_release); + + // Plaintext KEEPALIVE: 14-byte header, type=2, no payload, no AEAD. The server + // identifies us by the verified UDP endpoint (set during the UdpBinding handshake). + uint8_t pkt[voicecat::net::kVoiceHeaderSize] = {0}; + pkt[0] = voicecat::net::kFrameKeepalive; + + sockaddr_in dest{}; + dest.sin_family = AF_INET; + dest.sin_addr.s_addr = udp_dest_addr_; + dest.sin_port = udp_dest_port_; + ::sendto(static_cast(fd), reinterpret_cast(pkt), + static_cast(sizeof(pkt)), 0, reinterpret_cast(&dest), sizeof(dest)); +} + namespace { int64_t client_now_ms() { return std::chrono::duration_cast( diff --git a/core/src/core/client.h b/core/src/core/client.h index c967f8f..108e1b4 100644 --- a/core/src/core/client.h +++ b/core/src/core/client.h @@ -140,6 +140,32 @@ struct vc_client { uint64_t server_session_id_{0}; std::atomic next_req_id_{1}; + // Keepalive: client sends a Ping every ~15s (docs/protocol.md §7) so the server's + // last_seen stays fresh and the reaper doesn't drop us. Pong echoes the nonce, which + // we correlate to measure RTT. The read loop's 50ms TLS timeout means it spins fast + // enough to check the ping interval with ample resolution. + static constexpr int64_t kPingIntervalMs = 15000; + std::atomic last_ping_ms_{0}; + std::atomic ping_nonce_{1}; + std::mutex ping_mutex_; + std::unordered_map pending_pings_; // nonce → sent_ms + std::atomic last_rtt_ms_{0}; + + // Graceful disconnect: set by disconnect() after queueing Disconnect{code=0}. The io + // thread checks this after drain_sends() — when set, it sets io_stop_ and exits the + // read loop, so the Disconnect is sent before the thread ends. The main thread just + // joins; no socket close from the main thread (the io thread's cleanup closes it), + // avoiding the double-close race that the send_cv_ wait approach exposed. + std::atomic graceful_disconnect_pending_{false}; + + // UDP keepalive: send a lightweight KEEPALIVE frame every ~5s to hold NAT bindings and + // bump the server's last_seen independently of the TCP ping (docs/voice.md §6). Sent + // as plaintext (no AEAD) — the server identifies the sender by its already-verified UDP + // endpoint, and the TCP reaper is the real timeout authority. Avoids racing the + // non-atomic send_counter_ in SodiumMediaCrypto::seal() with the audio callback thread. + static constexpr int64_t kUdpKeepaliveIntervalMs = 5000; + std::atomic last_udp_keepalive_ms_{0}; + // Client-side session model. Mutated only on io_thread_ (handle_server_state/ // handle_user_event/handle_channel_event), but read from any thread via the M4 // list_channels/list_users/list_user_streams getters — session_model_mu_ guards both. @@ -270,6 +296,10 @@ struct vc_client { void finish_udp_binding(); // udp_thread_ entry point: recv loop, AEAD-open, decode, push to audio_engine_. void run_udp_recv(); + // Send a plaintext KEEPALIVE frame to the server media endpoint. Called from run_udp_recv + // every kUdpKeepaliveIntervalMs to hold NAT bindings + bump the server's last_seen + // (docs/voice.md §6). Plaintext — no AEAD — to avoid racing the audio thread's seal(). + void send_udp_keepalive(); // capture_cb passed to audio_engine_.start(): encode + seal + send one frame for the // given local stream `kind` (M3: multiple concurrent local streams are possible). void on_capture_frame(int kind, const int16_t* pcm, int samples, int channels); @@ -292,6 +322,11 @@ struct vc_client { // Queue an encoded envelope to be sent on io_thread_. void queue_envelope(const voicecat::v1::Envelope& env); + // Keepalive: send a Ping envelope with a fresh nonce and record the sent time for + // RTT measurement when the Pong arrives. Called from the read loop when kPingIntervalMs + // has elapsed. (docs/protocol.md §7) + void send_ping(); + // Drain send_queue_ by doing blocking TLS writes (called on io_thread_). void drain_sends(); diff --git a/docs/architecture.md b/docs/architecture.md index faa8e4e..254974d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -197,6 +197,10 @@ Design notes: and presence can be subscribed more broadly. This keeps fan-out bounded on big servers. - **Stateless-ish media.** UDP carries no auth per packet beyond the media-AEAD session; the 5-tuple→session binding is established once via a token (see protocol.md §4). +- **Keepalive reaper.** An `asio::steady_timer` sweeps every 15 s and drops any session + whose `last_seen` (bumped on every inbound TCP or UDP frame) is older than 45 s. Each + drop broadcasts `UserEvent::LEFT` so peers clean up immediately. This catches half-open + connections that never produce a TCP EOF. Configurable via `server::Config`. - **Single process, scalable later.** v1 is one process, one machine. The session registry and router are written behind interfaces so a future build can sit them behind a shared bus for multi-node, but that is explicitly out of scope for now. diff --git a/docs/protocol.md b/docs/protocol.md index bba6f54..7da805a 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -274,6 +274,8 @@ message TextMessage { user, kick, ban, server-mute, set-permission, create/reset/delete account). Error `code`s are an enumerated, stable list. - Fatal conditions send **`Disconnect { code; reason }`** then close the TLS connection. + `code ≥ 1` is server-sent (1 = protocol error, 2 = kicked). `code = 0` is client-sent + graceful disconnect (§7): the server broadcasts `UserEvent::LEFT` and closes immediately. - **The response is for the request; the broadcast is for the state.** A `*Result` only acknowledges the actor's request (correlation via `request_id`, error text, and any actor-private payload — e.g. the channel `AudioConfig` in `JoinChannelResult`). The @@ -285,10 +287,27 @@ message TextMessage { ## 7. Keepalive & timeouts -- **TCP:** `Ping`/`Pong` every ~15 s; missing N consecutive pongs → drop. `Pong` echoes the - `Ping` nonce so RTT is measurable. +- **TCP:** `Ping`/`Pong` every ~15 s; missing 3 consecutive pongs (45 s) → the server's + reaper drops the session. `Pong` echoes the `Ping` nonce so RTT is measurable. The + client sends `Ping` automatically from its io thread; the server answers with `Pong` + in any state. +- **`last_seen` reaper.** Every `ConnSession` tracks `last_seen` — bumped on *any* + inbound TCP frame (not just `Ping`) and on any inbound UDP voice/keepalive frame. A + periodic sweep (`asio::steady_timer`, every 15 s) drops sessions whose `last_seen` is + older than 45 s. Each drop calls `close()`, which broadcasts `UserEvent::LEFT` to + remaining clients — so half-open connections (NAT timeout, wifi loss without RST, + laptop sleep) that never produce a TCP EOF are cleaned up, and peers' audio engines + `remove_stream` and stop PLC. The timeout and sweep interval are configurable via + `server::Config::reaper_timeout_ms` / `reaper_sweep_ms` (set to 0 to disable). - **UDP:** a separate lightweight keepalive on the media channel (voice.md §6) keeps NAT bindings alive and detects media-path failure independently of the control channel. +- **Graceful disconnect.** A client ending its session sends `Disconnect { code = 0; + reason }` before closing the socket. The server calls `close()` on receipt — + broadcasting `UserEvent::LEFT` immediately, without waiting for TCP EOF or the reaper. + The client's `vc_disconnect()` queues this message and waits for the io thread to flush + it before closing the socket. `code = 0` is reserved for client-initiated graceful + disconnect; server-sent fatal `Disconnect` uses `code ≥ 1` (1 = protocol error, + 2 = kicked). ## 8. Extensibility checklist diff --git a/docs/voice.md b/docs/voice.md index 0879fc6..7c9457e 100644 --- a/docs/voice.md +++ b/docs/voice.md @@ -137,7 +137,12 @@ Each receiver keeps an **adaptive jitter buffer per ssrc**. ## 6. UDP keepalive & NAT - A `KEEPALIVE` (type 2) frame flows both directions on the media channel every ~5 s to - hold NAT bindings and measure media-path RTT/loss independent of TCP. + hold NAT bindings and measure media-path RTT/loss independent of TCP. The frame is + plaintext (14-byte header, no payload, no AEAD) — the server identifies the sender by + its already-verified UDP endpoint (established during the `UdpBinding` handshake). On + receipt the server bumps the sender's `last_seen` (so media activity defers the TCP + reaper independently of control-channel traffic) and echoes the frame back so the + client can measure media-path RTT. - If the media path dies but TCP is alive, the client surfaces a "voice disconnected" state and attempts UDP re-binding (re-derive media keys + fresh `UdpBinding`) without dropping the control session. diff --git a/server/src/conn_session.cpp b/server/src/conn_session.cpp index 9fc8ba6..645cb95 100644 --- a/server/src/conn_session.cpp +++ b/server/src/conn_session.cpp @@ -59,10 +59,12 @@ void ConnSession::set_io(SendFn send_fn, CloseFn close_fn) { } void ConnSession::begin() { + touch_last_seen(); // Nothing to do at TCP level — wait for ClientHello } void ConnSession::on_frame(std::vector frame) { + touch_last_seen(); // any inbound TCP activity keeps this session off the reaper's list voicecat::v1::Envelope env; if (!protocol::decode_envelope(frame, env)) return; @@ -87,6 +89,12 @@ void ConnSession::on_frame(std::vector frame) { case voicecat::v1::Envelope::kPing: handle_ping(env.ping()); break; + case voicecat::v1::Envelope::kDisconnect: + // Client-initiated graceful disconnect (code=0). Falls through to close() which + // broadcasts UserEvent::LEFT — same as a TCP drop, but immediate (no reaper/EOF + // wait). Gated on Authenticated so a pre-auth stray Disconnect can't skip cleanup. + if (st == State::Authenticated) handle_client_disconnect(env.disconnect()); + break; case voicecat::v1::Envelope::kLeaveChannel: if (st == State::Authenticated) handle_leave_channel(); @@ -172,7 +180,19 @@ void ConnSession::close() { if (closed_.exchange(true)) return; state_.store(State::Disconnecting, std::memory_order_release); uint32_t uid = user_id_.load(); - if (uid) registry_->remove_user(uid); + if (uid) { + // Broadcast LEFT BEFORE erasing the user, so remaining clients (and the audio + // engine's remove_stream path on each peer) learn about the departure. This + // covers ungraceful disconnects (TCP drop, crash, network loss) that previously + // silently erased the user from the registry without notifying anyone — which + // left stale users in peer client lists and kept Opus PLC hissing forever on + // peers whose remove_stream was never triggered. Mirrors kick_user's first half + // (session_registry.cpp kick_user). broadcast_left releases its shared_lock + // before remove_user acquires the unique_lock, so no deadlock; and send_envelope + // on this session is a no-op now that closed_ is true. + registry_->broadcast_left(uid, ""); + registry_->remove_user(uid); + } if (session_id_) registry_->unregister_session(session_id_); if (close_fn_) close_fn_(); } @@ -475,6 +495,15 @@ void ConnSession::handle_ping(const voicecat::v1::Ping& msg) { send_envelope(env); } +void ConnSession::handle_client_disconnect(const voicecat::v1::Disconnect& /*msg*/) { + // Graceful client-initiated disconnect (code=0). close() broadcasts UserEvent::LEFT and + // cleans up the registry — identical to the TCP-drop path, but triggered by the client's + // explicit goodbye so peers learn about the departure immediately (no reaper timeout, + // no EOF detection delay). Idempotent: the closed_ guard in close() handles the + // inevitable follow-up TCP EOF when the client closes its socket. + close(); +} + void ConnSession::handle_udp_binding(uint64_t req_id, const voicecat::v1::UdpBinding& msg) { if (msg.ack()) return; // server→client direction; ignore if echoed back diff --git a/server/src/conn_session.h b/server/src/conn_session.h index a5e56b0..9f08d49 100644 --- a/server/src/conn_session.h +++ b/server/src/conn_session.h @@ -15,6 +15,7 @@ #include #include +#include #include #include #include @@ -82,12 +83,24 @@ class ConnSession : public std::enable_shared_from_this { uint64_t session_id() const { return session_id_; } uint32_t user_id() const { return user_id_; } + // Keepalive: bump last_seen to "now" on any inbound activity (TCP frame or UDP voice + // frame). The server's reaper sweeps sessions whose last_seen is older than the + // timeout (docs/protocol.md §7). Stored as steady_clock ms since epoch in an atomic so + // the reaper thread can read it lock-free. + void touch_last_seen() { + last_seen_ms_.store(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(), + std::memory_order_release); + } + int64_t last_seen_ms() const { return last_seen_ms_.load(std::memory_order_acquire); } + private: void handle_client_hello(uint64_t req_id, const voicecat::v1::ClientHello& msg); void handle_auth_request(uint64_t req_id, const voicecat::v1::AuthRequest& msg); void handle_join_channel(uint64_t req_id, const voicecat::v1::JoinChannelRequest& msg); void handle_text_message(const voicecat::v1::TextMessage& msg); void handle_ping(const voicecat::v1::Ping& msg); + void handle_client_disconnect(const voicecat::v1::Disconnect& msg); void handle_udp_binding(uint64_t req_id, const voicecat::v1::UdpBinding& msg); void handle_stream_announce(uint64_t req_id, const voicecat::v1::StreamAnnounce& msg); void handle_stream_stop(const voicecat::v1::StreamStop& msg); @@ -136,6 +149,11 @@ class ConnSession : public std::enable_shared_from_this { std::atomic user_id_{0}; std::atomic closed_{false}; + // Keepalive: steady_clock ms since epoch of the last inbound activity. Bumped by + // touch_last_seen() on every on_frame() and (via media_relay) on every UDP voice frame. + // The reaper (server.cpp) drops sessions whose last_seen is older than 45s. + std::atomic last_seen_ms_{0}; + // M2 UDP / media std::array udp_token_{}; mutable std::mutex udp_ep_mu_; diff --git a/server/src/media_relay.cpp b/server/src/media_relay.cpp index d37dd56..704df9f 100644 --- a/server/src/media_relay.cpp +++ b/server/src/media_relay.cpp @@ -130,7 +130,20 @@ void MediaRelay::on_udp_frame(const uint8_t* data, size_t len, return; } - // kFrameKeepalive or unknown: silently discard. + if (frame_type == voicecat::net::kFrameKeepalive) { + // Plaintext KEEPALIVE (docs/voice.md §6): identify the sender by its verified UDP + // endpoint, bump last_seen so the reaper doesn't drop a client whose TCP control + // channel is idle but whose media path is alive, and echo the keepalive back so the + // client can measure media-path RTT/loss independently of the TCP ping. + auto session = registry_->find_by_udp_endpoint(sender); + if (!session) return; + session->touch_last_seen(); + // Echo back to the sender (same plaintext header). + udp_.send_to(data, len, sender); + return; + } + + // Unknown frame type: silently discard. } } // namespace voicecat::server diff --git a/server/src/server.cpp b/server/src/server.cpp index 0af91a1..875e823 100644 --- a/server/src/server.cpp +++ b/server/src/server.cpp @@ -8,7 +8,10 @@ #include #include +#include +#include #include +#include #include "conn_session.h" #include "core/worker_pool.h" @@ -148,6 +151,32 @@ int Server::run() { io.stop(); }); + // ── Keepalive reaper (docs/protocol.md §7) ───────────────────────────────── + // Sweeps every reaper_sweep_ms and drops any session whose last_seen is older than + // reaper_timeout_ms. Each close() broadcasts UserEvent::LEFT via the Tier 1 fix, so + // peers learn about the timeout exactly like a normal disconnect — their audio engines + // call remove_stream and stop PLC. This catches half-open connections (NAT timeout, + // wifi loss without RST, laptop sleep) that never produce a TCP EOF and would otherwise + // leave ghost users forever. Disabled when reaper_timeout_ms <= 0. + asio::steady_timer reaper_timer(io); + std::function arm_reaper; + if (cfg_.reaper_timeout_ms > 0) { + arm_reaper = [&] { + reaper_timer.expires_after(std::chrono::milliseconds(cfg_.reaper_sweep_ms)); + reaper_timer.async_wait([&](std::error_code ec) { + if (ec || io.stopped()) return; + for (auto& sess : registry->find_stale_sessions(cfg_.reaper_timeout_ms)) { + std::printf("[server] reaper: dropping stale session %llu (user %u)\n", + static_cast(sess->session_id()), + sess->user_id()); + sess->close(); + } + arm_reaper(); + }); + }; + arm_reaper(); + } + std::printf("[voicecat-server] %s — TCP :%u UDP :%u\n", cfg_.server_name.c_str(), bound, media_bound); std::printf("[voicecat-server] fingerprint: %s\n", diff --git a/server/src/server.h b/server/src/server.h index 733d15d..21f61f6 100644 --- a/server/src/server.h +++ b/server/src/server.h @@ -24,6 +24,13 @@ struct Config { std::function on_ready; // Called with the actual bound UDP media port once the relay is ready. std::function on_media_ready; + + // Keepalive reaper (docs/protocol.md §7): sweep every reaper_sweep_ms and drop any + // session whose last inbound TCP/UDP activity is older than reaper_timeout_ms. Defaults + // match the doc: 15s sweep, 45s timeout (3 missed 15s pongs). Set reaper_timeout_ms = 0 + // to disable the reaper entirely (useful for tests that don't want it). + int64_t reaper_timeout_ms = 45000; + int64_t reaper_sweep_ms = 15000; }; class Server { diff --git a/server/src/session_registry.cpp b/server/src/session_registry.cpp index df28bfc..93426f2 100644 --- a/server/src/session_registry.cpp +++ b/server/src/session_registry.cpp @@ -3,6 +3,7 @@ #ifdef VOICECAT_HAS_NET #include +#include #include #include @@ -202,6 +203,22 @@ void SessionRegistry::broadcast_unlocked(const voicecat::v1::Envelope& env, } } +std::vector> SessionRegistry::find_stale_sessions( + int64_t max_age_ms) const { + std::shared_lock lk(mu_); + auto now_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + std::vector> stale; + for (auto& [sid, weak] : sessions_) { + auto sess = weak.lock(); + if (!sess) continue; + int64_t seen = sess->last_seen_ms(); + if (seen == 0) continue; // not yet initialized — skip (shouldn't happen after begin()) + if (now_ms - seen > max_age_ms) stale.push_back(sess); + } + return stale; +} + // ── Permissions ─────────────────────────────────────────────────────────────── void SessionRegistry::set_session_permissions(uint64_t session_id, @@ -253,6 +270,11 @@ bool SessionRegistry::kick_user(uint32_t user_id, const std::string& reason) { return true; } +void SessionRegistry::broadcast_left(uint32_t user_id, const std::string& reason) { + std::shared_lock lk(mu_); + broadcast_unlocked(make_left_event(user_id, reason), /*exclude*/ 0); +} + bool SessionRegistry::ban_user(uint32_t user_id, const std::string& reason, int64_t expires_at) { { std::unique_lock lk(mu_); diff --git a/server/src/session_registry.h b/server/src/session_registry.h index 1398d5c..51b1bf9 100644 --- a/server/src/session_registry.h +++ b/server/src/session_registry.h @@ -66,6 +66,12 @@ class SessionRegistry { // Remove a user (called on disconnect after auth). void remove_user(uint32_t user_id); + // Broadcast a UserEvent::LEFT for a user to all other sessions. Called by + // ConnSession::close() before remove_user() so remaining clients learn about an + // ungraceful disconnect (TCP drop, crash, network loss). Mirrors the first half of + // kick_user(). Takes the shared lock internally; safe to call from ConnSession::close. + void broadcast_left(uint32_t user_id, const std::string& reason); + // Move a user to a channel. Returns false if channel doesn't exist. bool set_user_channel(uint32_t user_id, uint32_t channel_id); @@ -82,6 +88,12 @@ class SessionRegistry { // Broadcast an envelope to all sessions except the excluded one. void broadcast(const voicecat::v1::Envelope& env, uint64_t exclude_session_id = 0) const; + // Return all sessions whose last_seen is older than max_age_ms (steady_clock ms), i.e. + // have not had any inbound TCP or UDP activity in that span. The reaper (server.cpp) + // calls close() on each — which broadcasts UserEvent::LEFT via the Tier 1 fix. Locks + // only to collect the list; close() runs outside the lock (mirrors kick_user's pattern). + std::vector> find_stale_sessions(int64_t max_age_ms) const; + private: void broadcast_unlocked(const voicecat::v1::Envelope& env, uint64_t exclude_session_id = 0) const; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 323ea5c..106d22c 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -60,6 +60,15 @@ if(VOICECAT_USE_VCPKG_DEPS) target_include_directories(test_opus_codec PRIVATE ${VC_TEST_INTERNAL_INCLUDES}) add_test(NAME opus_codec COMMAND test_opus_codec) + # PLC cap: after ~2s of pure PLC (no real packets), the mixer emits silence instead of + # comfort noise — bounds the eternal-hiss failure mode (defense-in-depth for the + # disconnect/LEFT fix). White-box AudioEngine test, no server needed. + add_executable(test_plc_cap test_plc_cap.cpp) + target_link_libraries(test_plc_cap PRIVATE voicecat::voicecat) + target_compile_features(test_plc_cap PRIVATE cxx_std_20) + target_include_directories(test_plc_cap PRIVATE ${VC_TEST_INTERNAL_INCLUDES}) + add_test(NAME plc_cap COMMAND test_plc_cap) + # M2 exit criterion: two headless clients relay encrypted Opus frames via the SFU. add_executable(test_m2_voice test_m2_voice.cpp) target_link_libraries(test_m2_voice PRIVATE voicecat::server) @@ -145,4 +154,24 @@ if(VOICECAT_USE_VCPKG_DEPS) target_include_directories(test_m5_channel_crud PRIVATE ${VC_TEST_INTERNAL_INCLUDES}) add_test(NAME m5_channel_crud COMMAND test_m5_channel_crud) set_tests_properties(m5_channel_crud PROPERTIES TIMEOUT 90) + + # Disconnect/timeout: server broadcasts UserEvent::LEFT on TCP drop (no more ghost + # users or eternal PLC hiss on peers). Also covers the keepalive/reaper paths added + # alongside the LEFT-broadcast fix. + add_executable(test_disconnect_left test_disconnect_left.cpp) + target_link_libraries(test_disconnect_left PRIVATE voicecat::server) + target_compile_features(test_disconnect_left PRIVATE cxx_std_20) + target_include_directories(test_disconnect_left PRIVATE ${VC_TEST_INTERNAL_INCLUDES}) + add_test(NAME disconnect_left COMMAND test_disconnect_left) + set_tests_properties(disconnect_left PROPERTIES TIMEOUT 90) + + # Reaper: half-open connections (no TCP EOF) are dropped after the configurable timeout, + # peers get UserEvent::LEFT, the stale client gets disconnected. Uses a 2s timeout for + # fast test turnaround (production default is 45s). + add_executable(test_reaper_timeout test_reaper_timeout.cpp) + target_link_libraries(test_reaper_timeout PRIVATE voicecat::server) + target_compile_features(test_reaper_timeout PRIVATE cxx_std_20) + target_include_directories(test_reaper_timeout PRIVATE ${VC_TEST_INTERNAL_INCLUDES}) + add_test(NAME reaper_timeout COMMAND test_reaper_timeout) + set_tests_properties(reaper_timeout PROPERTIES TIMEOUT 30) endif() diff --git a/tests/test_disconnect_left.cpp b/tests/test_disconnect_left.cpp new file mode 100644 index 0000000..9b7a95a --- /dev/null +++ b/tests/test_disconnect_left.cpp @@ -0,0 +1,221 @@ +/* + * test_disconnect_left — regression test for the ungraceful-disconnect LEFT bug. + * + * Verifies that when a client's TCP connection drops (vc_disconnect / socket close / + * process kill), the server broadcasts UserEvent::LEFT to remaining clients — so peer + * user lists stay fresh and peer audio engines remove the stale stream (no eternal PLC + * hiss). Before the fix, ConnSession::close() silently erased the user from the registry + * without broadcasting, leaving ghost users and never-ending comfort noise on peers. + */ +#include + +#ifdef VOICECAT_HAS_NET + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "voicecat.h" +#include "server.h" +#include "db.h" + +struct EventStore { + std::mutex mu; + std::condition_variable cv; + + bool auth_ok{false}; + bool auth_done{false}; + uint32_t self_user_id{0}; + bool channel_list_received{false}; + bool disconnected{false}; + + std::vector joined_users; + std::vector left_users; + + vc_client* client{nullptr}; + const char* label{nullptr}; +}; + +static void on_event(void* user, const vc_event* ev) { + auto* s = static_cast(user); + std::lock_guard lk(s->mu); + switch (ev->type) { + case VC_EVENT_SERVER_IDENTITY: + vc_confirm_server_identity(s->client, 1); + break; + case VC_EVENT_AUTH_RESULT: + s->auth_ok = (ev->result == VC_OK); + s->auth_done = true; + s->self_user_id = ev->user_id; + break; + case VC_EVENT_CHANNEL_LIST: + s->channel_list_received = true; + break; + case VC_EVENT_USER_JOINED: + s->joined_users.push_back(ev->user_id); + break; + case VC_EVENT_USER_LEFT: + s->left_users.push_back(ev->user_id); + break; + case VC_EVENT_DISCONNECTED: + s->disconnected = true; + break; + default: + break; + } + s->cv.notify_all(); +} + +template +static bool wait_for(EventStore& s, Pred pred, int timeout_ms) { + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + std::unique_lock lk(s.mu); + return s.cv.wait_until(lk, deadline, [&] { return pred(s); }); +} + +static int g_failures = 0; +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + std::printf("FAIL: %s (%s:%d)\n", #cond, __FILE__, __LINE__); \ + ++g_failures; \ + } \ + } while (0) + +static bool user_list_contains(vc_client* c, uint32_t uid) { + vc_user_list ul{}; + if (vc_list_users(c, &ul) != VC_OK) return false; + bool found = false; + for (size_t i = 0; i < ul.count; ++i) { + if (ul.items[i].id == uid) { found = true; break; } + } + vc_free_user_list(&ul); + return found; +} + +int main() { + auto tmp = std::filesystem::temp_directory_path() / + ("vctest_disc_left_" + std::to_string( + std::chrono::steady_clock::now().time_since_epoch().count())); + std::filesystem::create_directories(tmp); + std::string data_dir = tmp.string(); + + std::atomic bound_port{0}; + std::mutex ready_mu; + std::condition_variable ready_cv; + bool ready{false}; + + voicecat::server::Config cfg; + cfg.data_dir = data_dir; + cfg.bind_port = 0; + cfg.server_name = "VoiceCat-DiscLeftTest"; + cfg.allow_guests = true; + cfg.on_ready = [&](uint16_t p) { + bound_port.store(p); + { std::lock_guard lk(ready_mu); ready = true; } + ready_cv.notify_all(); + }; + + voicecat::server::Server server(cfg); + std::thread server_thread([&] { server.run(); }); + + { + std::unique_lock lk(ready_mu); + if (!ready_cv.wait_for(lk, std::chrono::seconds(10), [&] { return ready; })) { + std::printf("FAIL: server did not become ready\n"); + server.stop(); + server_thread.join(); + std::filesystem::remove_all(tmp); + return 1; + } + } + uint16_t port = bound_port.load(); + + auto make_client = [&](const char* label, const char* nick) -> EventStore* { + auto* ev = new EventStore(); + ev->label = label; + vc_callbacks cb{on_event, nullptr, ev}; + vc_config cfgx{label, "0.1", VC_LOG_OFF}; + ev->client = vc_client_create(&cfgx, cb); + if (!ev->client) return nullptr; + if (vc_connect(ev->client, "127.0.0.1", port) != VC_OK) return nullptr; + if (vc_authenticate_guest(ev->client, nick) != VC_OK) return nullptr; + return ev; + }; + + EventStore* evA = make_client("clientA", "Alpha"); + CHECK(evA != nullptr); + CHECK(wait_for(*evA, [](EventStore& s) { return s.auth_ok; }, 8000)); + CHECK(wait_for(*evA, [](EventStore& s) { return s.channel_list_received; }, 3000)); + uint32_t a_uid = evA->self_user_id; + CHECK(a_uid != 0); + + EventStore* evB = make_client("clientB", "Bravo"); + CHECK(evB != nullptr); + CHECK(wait_for(*evB, [](EventStore& s) { return s.auth_ok; }, 8000)); + CHECK(wait_for(*evB, [](EventStore& s) { return s.channel_list_received; }, 3000)); + uint32_t b_uid = evB->self_user_id; + CHECK(b_uid != 0); + + // A should see B join (broadcast_user_joined fires when B authenticates). + CHECK(wait_for(*evA, [b_uid](EventStore& s) { + for (auto u : s.joined_users) if (u == b_uid) return true; + return false; + }, 5000)); + + // Both should see each other in the authoritative user list. + CHECK(user_list_contains(evA->client, b_uid)); + CHECK(user_list_contains(evB->client, a_uid)); + + // ── Drop A's TCP connection abruptly (no LeaveChannelRequest, no Goodbye — + // just close the socket, exactly like vccli Ctrl-C or a network drop). ── + vc_disconnect(evA->client); + vc_client_destroy(evA->client); + // (evA is now a dangling store; only evB is observed below.) + + // B must receive VC_EVENT_USER_LEFT for A — the core fix under test. + CHECK(wait_for(*evB, [a_uid](EventStore& s) { + for (auto u : s.left_users) if (u == a_uid) return true; + return false; + }, 5000)); + + // B's authoritative user list must no longer contain A. + // Give the event a moment to propagate through the SessionModel, then poll briefly. + bool a_gone = false; + for (int i = 0; i < 20; ++i) { + if (!user_list_contains(evB->client, a_uid)) { a_gone = true; break; } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + CHECK(a_gone); + + // ── Cleanup ────────────────────────────────────────────────────────────── + vc_disconnect(evB->client); + vc_client_destroy(evB->client); + delete evA; + delete evB; + + server.stop(); + server_thread.join(); + std::filesystem::remove_all(tmp); + + if (g_failures == 0) { + std::printf("disconnect_left: all checks passed\n"); + return 0; + } + std::printf("disconnect_left: %d failure(s)\n", g_failures); + return 1; +} + +#else // !VOICECAT_HAS_NET + +int main() { + std::printf("disconnect_left: SKIP (VOICECAT_HAS_NET not defined)\n"); + return 0; +} + +#endif // VOICECAT_HAS_NET diff --git a/tests/test_plc_cap.cpp b/tests/test_plc_cap.cpp new file mode 100644 index 0000000..822929c --- /dev/null +++ b/tests/test_plc_cap.cpp @@ -0,0 +1,148 @@ +/* + * test_plc_cap — verifies the PLC cap in AudioEngine::on_playback. + * + * After ~2s of pure packet-loss concealment (no real packets decoded), the mixer stops + * calling opus_decode(nullptr,0,...) and emits digital silence instead. This bounds the + * Opus comfort-noise hiss so a stale stream left in the mixer can never hiss forever — + * defense-in-depth for the server's UserEvent::LEFT broadcast (the primary fix that + * triggers remove_stream on disconnect). Also verifies that a fresh real packet resets + * the PLC streak and audio resumes. + * + * White-box: drives AudioEngine::mix_for_test directly (no audio hardware needed). + */ +#include +#include +#include +#include + +#if defined(VOICECAT_HAS_AUDIO) && defined(VOICECAT_HAS_OPUS) + +#include "audio/audio_engine.h" +#include "codec/opus_codec.h" + +static int g_failures = 0; +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + std::printf("FAIL [%s:%d]: %s\n", __FILE__, __LINE__, #cond); \ + ++g_failures; \ + } \ + } while (0) + +static double rms(const int16_t* pcm, int n) { + double sum = 0.0; + for (int i = 0; i < n; ++i) sum += static_cast(pcm[i]) * pcm[i]; + return std::sqrt(sum / n); +} + +static int64_t abs_energy(const int16_t* pcm, int n) { + int64_t e = 0; + for (int i = 0; i < n; ++i) e += static_cast(std::abs(static_cast(pcm[i]))); + return e; +} + +int main() { + voicecat::audio::AudioEngine engine; + voicecat::audio::AudioParams p; + p.sample_rate = 48000; + p.capture_channels = 1; + p.playback_channels = 2; + p.frame_ms = 20; + CHECK(engine.start(p)); // no capture_cb — headless safe (devices may fail to init; ok) + + voicecat::codec::OpusParams op; + op.sample_rate = 48000; + op.frame_ms = 20; + op.stereo = false; + int frame_samples = voicecat::codec::opus_frame_samples(op); // 960 + + // Encode a loud sine wave to seed the decoder's PLC state. + voicecat::codec::OpusEncoder enc; + CHECK(enc.init(op)); + std::vector sine(static_cast(frame_samples)); + for (int i = 0; i < frame_samples; ++i) { + float t = static_cast(i) / 48000.0f; + sine[i] = static_cast(std::sin(2.0f * 3.14159265f * 440.0f * t) * 20000.0f); + } + uint8_t opus_buf[1500]; + int opus_len = enc.encode(sine.data(), frame_samples, opus_buf, sizeof(opus_buf)); + CHECK(opus_len > 0); + + const uint32_t ssrc = 1; + engine.init_recv_stream(ssrc, op); + + // Push one real frame to seed the decoder. + voicecat::audio::JitterBuffer::Frame f; + f.seq = 0; + f.timestamp = 0; + f.fec_present = false; + f.payload.assign(opus_buf, opus_buf + opus_len); + engine.push_recv_frame(ssrc, std::move(f)); + + // mix_for_test period — 480 frames @ 48kHz = 10ms (typical WASAPI shared period). + const uint32_t pb_frames = 480; + const int out_n = static_cast(pb_frames) * 2; // stereo interleaved + std::vector out(static_cast(out_n), 0); + + // 1) Decode the real frame (first mix call) — seeds PLC state. + engine.mix_for_test(out.data(), pb_frames); + + // 2) Drive ~50ms of pure PLC — should produce comfort noise (non-zero). + double early_rms = 0.0; + for (int i = 0; i < 5; ++i) { + engine.mix_for_test(out.data(), pb_frames); + early_rms = std::max(early_rms, rms(out.data(), out_n)); + } + CHECK(early_rms > 1.0); // PLC of a loud sine is audible, not digital silence + + // 3) Drive well past the 2s PLC cap (250 callbacks = 2.5s of output). + // After the cap, on_playback emits silence (memset 0) instead of PLC noise. + for (int i = 0; i < 250; ++i) { + engine.mix_for_test(out.data(), pb_frames); + } + + // 4) The output must now be digital silence (all zeros), not comfort noise. + // Drain a couple more callbacks to flush any ring residue, then assert. + int64_t energy = 0; + for (int i = 0; i < 3; ++i) { + engine.mix_for_test(out.data(), pb_frames); + energy = std::max(energy, abs_energy(out.data(), out_n)); + } + CHECK(energy == 0); // capped PLC = silence + + // 5) Resumption: push a fresh real frame — PLC streak resets, audio returns. + voicecat::audio::JitterBuffer::Frame f2; + f2.seq = 1; + f2.timestamp = 200000; // far ahead — playout-clock re-sync snaps to it + f2.fec_present = false; + f2.payload.assign(opus_buf, opus_buf + opus_len); + engine.push_recv_frame(ssrc, std::move(f2)); + + double resume_rms = 0.0; + for (int i = 0; i < 5; ++i) { // a few calls to flush silence residue + decode real + engine.mix_for_test(out.data(), pb_frames); + resume_rms = std::max(resume_rms, rms(out.data(), out_n)); + } + CHECK(resume_rms > 1.0); // real audio is back + + engine.remove_stream(ssrc); + engine.stop(); + enc.destroy(); + + if (g_failures == 0) { + std::printf("plc_cap: all checks passed (early_rms=%.1f resume_rms=%.1f)\n", + early_rms, resume_rms); + return 0; + } + std::printf("plc_cap: %d failure(s)\n", g_failures); + return 1; +} + +#else + +int main() { + std::printf("plc_cap: SKIP (VOICECAT_HAS_AUDIO or VOICECAT_HAS_OPUS not defined)\n"); + return 0; +} + +#endif diff --git a/tests/test_reaper_timeout.cpp b/tests/test_reaper_timeout.cpp new file mode 100644 index 0000000..30fe340 --- /dev/null +++ b/tests/test_reaper_timeout.cpp @@ -0,0 +1,206 @@ +/* + * test_reaper_timeout — verifies the server's keepalive reaper (docs/protocol.md §7). + * + * Simulates a half-open connection: client B authenticates then goes completely silent + * (no TCP traffic, no pings — the 15s ping interval far exceeds the test's 2s reaper + * timeout). Client A stays alive by sending channel text every 500ms, which bumps its + * last_seen on the server. After ~2s the reaper drops B: B's TCP connection is closed + * (B sees VC_EVENT_DISCONNECTED) and A receives VC_EVENT_USER_LEFT for B (the Tier 1 + * LEFT-broadcast fires from close()). + * + * This catches the "ghost user forever" failure mode for half-open connections (NAT + * timeout, wifi loss without RST, laptop sleep) that never produce a TCP EOF. + */ +#include + +#ifdef VOICECAT_HAS_NET + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "voicecat.h" +#include "server.h" +#include "db.h" + +struct EventStore { + std::mutex mu; + std::condition_variable cv; + + bool auth_ok{false}; + uint32_t self_user_id{0}; + bool channel_list_received{false}; + bool disconnected{false}; + + std::vector left_users; + + vc_client* client{nullptr}; + const char* label{nullptr}; +}; + +static void on_event(void* user, const vc_event* ev) { + auto* s = static_cast(user); + std::lock_guard lk(s->mu); + switch (ev->type) { + case VC_EVENT_SERVER_IDENTITY: + vc_confirm_server_identity(s->client, 1); + break; + case VC_EVENT_AUTH_RESULT: + s->auth_ok = (ev->result == VC_OK); + s->self_user_id = ev->user_id; + break; + case VC_EVENT_CHANNEL_LIST: + s->channel_list_received = true; + break; + case VC_EVENT_USER_LEFT: + s->left_users.push_back(ev->user_id); + break; + case VC_EVENT_DISCONNECTED: + s->disconnected = true; + break; + default: + break; + } + s->cv.notify_all(); +} + +template +static bool wait_for(EventStore& s, Pred pred, int timeout_ms) { + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + std::unique_lock lk(s.mu); + return s.cv.wait_until(lk, deadline, [&] { return pred(s); }); +} + +static int g_failures = 0; +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + std::printf("FAIL: %s (%s:%d)\n", #cond, __FILE__, __LINE__); \ + ++g_failures; \ + } \ + } while (0) + +int main() { + auto tmp = std::filesystem::temp_directory_path() / + ("vctest_reaper_" + std::to_string( + std::chrono::steady_clock::now().time_since_epoch().count())); + std::filesystem::create_directories(tmp); + std::string data_dir = tmp.string(); + + std::atomic bound_port{0}; + std::mutex ready_mu; + std::condition_variable ready_cv; + bool ready{false}; + + voicecat::server::Config cfg; + cfg.data_dir = data_dir; + cfg.bind_port = 0; + cfg.server_name = "VoiceCat-ReaperTest"; + cfg.allow_guests = true; + cfg.reaper_timeout_ms = 2000; // 2s — drop sessions silent for this long + cfg.reaper_sweep_ms = 500; // check every 500ms + cfg.on_ready = [&](uint16_t p) { + bound_port.store(p); + { std::lock_guard lk(ready_mu); ready = true; } + ready_cv.notify_all(); + }; + + voicecat::server::Server server(cfg); + std::thread server_thread([&] { server.run(); }); + + { + std::unique_lock lk(ready_mu); + if (!ready_cv.wait_for(lk, std::chrono::seconds(10), [&] { return ready; })) { + std::printf("FAIL: server did not become ready\n"); + server.stop(); + server_thread.join(); + std::filesystem::remove_all(tmp); + return 1; + } + } + uint16_t port = bound_port.load(); + + auto make_client = [&](const char* label, const char* nick) -> EventStore* { + auto* ev = new EventStore(); + ev->label = label; + vc_callbacks cb{on_event, nullptr, ev}; + vc_config cfgx{label, "0.1", VC_LOG_OFF}; + ev->client = vc_client_create(&cfgx, cb); + if (!ev->client) return nullptr; + if (vc_connect(ev->client, "127.0.0.1", port) != VC_OK) return nullptr; + if (vc_authenticate_guest(ev->client, nick) != VC_OK) return nullptr; + return ev; + }; + + EventStore* evA = make_client("clientA", "Alpha"); + CHECK(evA != nullptr); + CHECK(wait_for(*evA, [](EventStore& s) { return s.auth_ok; }, 8000)); + CHECK(wait_for(*evA, [](EventStore& s) { return s.channel_list_received; }, 3000)); + uint32_t a_uid = evA->self_user_id; + CHECK(a_uid != 0); + + EventStore* evB = make_client("clientB", "Bravo"); + CHECK(evB != nullptr); + CHECK(wait_for(*evB, [](EventStore& s) { return s.auth_ok; }, 8000)); + CHECK(wait_for(*evB, [](EventStore& s) { return s.channel_list_received; }, 3000)); + uint32_t b_uid = evB->self_user_id; + CHECK(b_uid != 0); + + // A stays alive by sending channel text every 500ms (bumps A's last_seen on the server). + // B goes completely silent — no TCP traffic, no pings (15s ping >> 2s reaper timeout). + // Start the keepalive IMMEDIATELY: the reaper timeout is only 2s, so A must begin + // sending well before its last_seen goes stale. + std::atomic keepalive_stop{false}; + std::thread keepalive([&] { + while (!keepalive_stop.load()) { + vc_send_text(evA->client, VC_TEXT_CHANNEL, 1, "."); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + }); + + // Give B time to go stale and the reaper to fire (2s timeout + 500ms sweep + margin). + // A must receive VC_EVENT_USER_LEFT for B. + CHECK(wait_for(*evA, [b_uid](EventStore& s) { + for (auto u : s.left_users) if (u == b_uid) return true; + return false; + }, 10000)); + + // B's TCP connection is closed by the reaper → B sees VC_EVENT_DISCONNECTED. + CHECK(wait_for(*evB, [](EventStore& s) { return s.disconnected; }, 5000)); + + // ── Cleanup ────────────────────────────────────────────────────────────── + keepalive_stop.store(true); + keepalive.join(); + + vc_disconnect(evA->client); + if (!evB->disconnected) vc_disconnect(evB->client); + vc_client_destroy(evA->client); + vc_client_destroy(evB->client); + delete evA; + delete evB; + + server.stop(); + server_thread.join(); + std::filesystem::remove_all(tmp); + + if (g_failures == 0) { + std::printf("reaper_timeout: all checks passed\n"); + return 0; + } + std::printf("reaper_timeout: %d failure(s)\n", g_failures); + return 1; +} + +#else // !VOICECAT_HAS_NET + +int main() { + std::printf("reaper_timeout: SKIP (VOICECAT_HAS_NET not defined)\n"); + return 0; +} + +#endif // VOICECAT_HAS_NET