fix(net): broadcast LEFT on disconnect, add keepalive/reaper, cap PLC hiss
Three reported bugs traced to one root cause plus two missing designed features:
1. Stale users + eternal PLC hiss (root cause): ConnSession::close() silently
erased dropped users without broadcasting UserEvent::LEFT, so peers never
learned the user left and their audio engines never called remove_stream —
Opus PLC synthesized comfort noise forever. Fix: broadcast_left() helper
+ close() broadcasts LEFT before erasing.
2. PLC cap (defense-in-depth): on_playback now caps pure PLC at ~2s, then
emits digital silence so a stale stream can never hiss forever even if
remove_stream is skipped. Resets automatically on fresh packets.
3. No timeout / no ping: client never sent Ping, server had no last_seen /
reaper, so half-open connections (NAT timeout, wifi loss, sleep) left
ghost users forever. Fix: client Ping every 15s with RTT measurement,
ConnSession::last_seen bumped on every inbound TCP/UDP frame, steady_timer
reaper sweeps every 15s and drops sessions older than 45s (configurable
via server::Config).
4. UDP KEEPALIVE: client sends plaintext kFrameKeepalive every 5s; 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() sends Disconnect{code=0} via
a flag-based io-thread exit (no double-close race); server handles
client-sent Disconnect with immediate close() + LEFT broadcast.
3 new tests: disconnect_left, plc_cap, reaper_timeout. 21/21 ctest green.
Docs: protocol.md §6/§7, voice.md §6, architecture.md §5, PROGRESS.md.
This commit is contained in:
@@ -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<int>(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<size_t>(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
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<uint8_t> 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::milliseconds>(
|
||||
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::milliseconds>(
|
||||
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<uint8_t> 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<size_t>(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::milliseconds>(
|
||||
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::milliseconds>(
|
||||
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::milliseconds>(
|
||||
std::chrono::steady_clock::now().time_since_epoch()).count(),
|
||||
std::memory_order_release);
|
||||
|
||||
std::vector<uint8_t> buf(2048);
|
||||
while (!udp_stop_.load(std::memory_order_acquire)) {
|
||||
int n = static_cast<int>(::recv(static_cast<sock_t>(fd), reinterpret_cast<char*>(buf.data()),
|
||||
static_cast<int>(buf.size()), 0));
|
||||
if (n < static_cast<int>(voicecat::net::kVoiceHeaderSize)) continue;
|
||||
if (n < static_cast<int>(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<size_t>(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::milliseconds>(
|
||||
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<sock_t>(fd), reinterpret_cast<const char*>(pkt),
|
||||
static_cast<int>(sizeof(pkt)), 0, reinterpret_cast<sockaddr*>(&dest), sizeof(dest));
|
||||
}
|
||||
|
||||
namespace {
|
||||
int64_t client_now_ms() {
|
||||
return std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
|
||||
@@ -140,6 +140,32 @@ struct vc_client {
|
||||
uint64_t server_session_id_{0};
|
||||
std::atomic<uint64_t> 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<int64_t> last_ping_ms_{0};
|
||||
std::atomic<uint64_t> ping_nonce_{1};
|
||||
std::mutex ping_mutex_;
|
||||
std::unordered_map<uint64_t, int64_t> pending_pings_; // nonce → sent_ms
|
||||
std::atomic<int64_t> 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<bool> 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<int64_t> 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();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user