From 3aa1922d2189714442650d4829580dd4dd0612a0 Mon Sep 17 00:00:00 2001 From: Alan George Date: Fri, 28 Aug 2026 08:33:35 -0600 Subject: [PATCH 1/3] Initial triage --- docs/testing.md | 44 +++ src/tests/CMakeLists.txt | 43 +++ src/tests/common/tcp_fault_proxy.h | 265 ++++++++++++++++++ .../connection/disconnect_offline_tester.cpp | 265 ++++++++++++++++++ 4 files changed, 617 insertions(+) create mode 100644 src/tests/common/tcp_fault_proxy.h create mode 100644 src/tests/connection/disconnect_offline_tester.cpp diff --git a/docs/testing.md b/docs/testing.md index 33919c0e..5486c806 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -45,6 +45,50 @@ __Note:__ The tests require tokens and a running LiveKit server. See the section | `livekit_integration_tests` | Quick tests (~1-2 minutes) for SDK functionality | | `livekit_stress_tests` | Long-running tests (configurable, default 1 hour) | +## Offline room-operation reproducer + +`livekit_disconnect_offline_tester` is a standalone POSIX-only tester for an +offline `Room::disconnect()` or `LocalParticipant::unpublishTrack()` call. It +publishes a local video track, places a loopback TCP fault proxy in front of a +`ws://` LiveKit server, resets the active signal connection, waits for the room +to enter `Reconnecting`, and invokes the selected operation while reconnect +traffic remains frozen. Forwarding resumes after the chosen observation +period. + +The tester keeps its `RoomDelegate` alive throughout disconnect. This follows +the corrected shutdown ordering from issue #222 and isolates the blocking +behavior from the original report's separate dangling-delegate risk. + +Build it with the normal test build: + +```bash +./build.sh debug-tests +``` + +Supply a non-TLS (`ws://`) server URL and token, either directly or through the +normal test environment: + +```bash +export LIVEKIT_URL=ws://localhost:7880 +export LIVEKIT_TOKEN_A='' +./build-debug/bin/livekit_disconnect_offline_tester \ + --operation disconnect --offline-duration-ms 10000 + +# Exercise the related unpublishTrack wait reported in the follow-up. +./build-debug/bin/livekit_disconnect_offline_tester \ + --operation unpublish-track --offline-duration-ms 10000 +``` + +The proxy cannot be used with `wss://`: it tunnels raw TCP through +`127.0.0.1`, which does not preserve the server hostname required for TLS +certificate validation. The program prints the elapsed time spent inside +the selected operation; an elapsed time near or above the offline duration +reproduces the reported blocking behavior. A healthy implementation should +return promptly without waiting for proxy forwarding to resume. With the +current Rust FFI, the unpublish variant can remain blocked even after forwarding +resumes and must be terminated manually; this reproduces the missing +`UnpublishTrackCallback`, not a fault in the tester. + ## Running a local LiveKit server for tests The integration and stress suites need a running LiveKit server. The easiest diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 9583af73..9edb410f 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -248,6 +248,49 @@ if(INTEGRATION_TEST_SOURCES) ) endif() +# ============================================================================ +# Connection Fault Testers +# ============================================================================ + +# Standalone reproducer for offline disconnect and track-unpublish waits. +# TcpFaultProxy uses POSIX sockets, so this target is unavailable on Windows. +if(UNIX) + add_executable(livekit_disconnect_offline_tester + "${CMAKE_CURRENT_SOURCE_DIR}/connection/disconnect_offline_tester.cpp" + "${CMAKE_CURRENT_SOURCE_DIR}/common/tcp_fault_proxy.h" + ) + target_link_libraries(livekit_disconnect_offline_tester PRIVATE livekit) + target_include_directories(livekit_disconnect_offline_tester PRIVATE + ${LIVEKIT_ROOT_DIR}/include + ${CMAKE_CURRENT_SOURCE_DIR}/common + ) + target_compile_definitions(livekit_disconnect_offline_tester PRIVATE + $<$:_USE_MATH_DEFINES> + ) + + if(APPLE) + add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.dylib" + $ + COMMENT "Copying tester shared libraries" + ) + else() + add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.so" + $ + COMMENT "Copying tester shared libraries" + ) + endif() +endif() + # ============================================================================ # Stress Tests # ============================================================================ diff --git a/src/tests/common/tcp_fault_proxy.h b/src/tests/common/tcp_fault_proxy.h new file mode 100644 index 00000000..ea0ddc1d --- /dev/null +++ b/src/tests/common/tcp_fault_proxy.h @@ -0,0 +1,265 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace livekit::test { + +/// Test-only TCP proxy whose established connections can be frozen or reset. +class TcpFaultProxy { +public: + TcpFaultProxy(std::string upstream_host, std::uint16_t upstream_port) + : upstream_host_(std::move(upstream_host)), upstream_port_(upstream_port) {} + + TcpFaultProxy(const TcpFaultProxy&) = delete; + TcpFaultProxy& operator=(const TcpFaultProxy&) = delete; + + ~TcpFaultProxy() { stop(); } + + /// Bind an ephemeral loopback port and begin accepting connections. + void start() { + if (running_.exchange(true)) return; + listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_fd_ < 0) { + running_.store(false); + throw std::runtime_error(socketError("failed to create proxy listener")); + } + int reuse_address = 1; + (void)::setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &reuse_address, sizeof(reuse_address)); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = 0; + if (::bind(listen_fd_, reinterpret_cast(&address), sizeof(address)) != 0) { + const auto error = socketError("failed to bind proxy listener"); + closeSocket(listen_fd_); + running_.store(false); + throw std::runtime_error(error); + } + if (::listen(listen_fd_, 16) != 0) { + const auto error = socketError("failed to listen on proxy socket"); + closeSocket(listen_fd_); + running_.store(false); + throw std::runtime_error(error); + } + socklen_t address_length = sizeof(address); + if (::getsockname(listen_fd_, reinterpret_cast(&address), &address_length) != 0) { + const auto error = socketError("failed to resolve proxy listener port"); + closeSocket(listen_fd_); + running_.store(false); + throw std::runtime_error(error); + } + listen_port_ = ntohs(address.sin_port); + const int listener = listen_fd_; + accept_thread_ = std::thread([this, listener]() { acceptLoop(listener); }); + } + + /// Stop the listener and all active forwarding workers. + void stop() { + if (!running_.exchange(false)) return; + paused_.store(false); + pause_cv_.notify_all(); + closeSocket(listen_fd_); + if (accept_thread_.joinable()) accept_thread_.join(); + const auto connections = connectionSnapshot(); + for (const auto& connection : connections) connection->close(); + for (const auto& connection : connections) connection->join(); + const std::lock_guard lock(connections_mutex_); + connections_.clear(); + } + + /// Freeze traffic in both directions without closing sockets. + void pause() { paused_.store(true); } + + /// Resume traffic on existing and newly accepted connections. + void resume() { + paused_.store(false); + pause_cv_.notify_all(); + } + + /// Abruptly close every currently accepted connection while retaining the listener. + void resetConnections() { + for (const auto& connection : connectionSnapshot()) connection->close(); + } + + /// Return the loopback port selected by start(). + std::uint16_t listenPort() const { return listen_port_; } + /// Return the total number of client connections accepted by this proxy. + std::uint64_t acceptedConnectionCount() const { return accepted_connection_count_.load(); } + +private: + class Connection { + public: + Connection(TcpFaultProxy& owner, int client_fd, int upstream_fd) + : owner_(owner), client_fd_(client_fd), upstream_fd_(upstream_fd) {} + ~Connection() { + close(); + join(); + } + void start() { + client_to_upstream_ = std::thread([this]() { pump(client_fd_, upstream_fd_); }); + upstream_to_client_ = std::thread([this]() { pump(upstream_fd_, client_fd_); }); + } + void close() { + if (!open_.exchange(false)) return; + owner_.pause_cv_.notify_all(); + closeSocket(client_fd_); + closeSocket(upstream_fd_); + } + void join() { + if (client_to_upstream_.joinable()) client_to_upstream_.join(); + if (upstream_to_client_.joinable()) upstream_to_client_.join(); + } + + private: + void pump(int source_fd, int destination_fd) { + std::array buffer{}; + while (open_.load() && owner_.running_.load()) { + if (!owner_.waitUntilResumed(open_)) break; + const auto bytes_read = ::recv(source_fd, buffer.data(), buffer.size(), 0); + if (bytes_read <= 0) break; + if (!owner_.waitUntilResumed(open_)) break; + std::size_t bytes_sent = 0; + while (bytes_sent < static_cast(bytes_read) && open_.load() && owner_.running_.load()) { + const auto result = sendNoSignal(destination_fd, buffer.data() + bytes_sent, + static_cast(bytes_read) - bytes_sent); + if (result <= 0) { + close(); + return; + } + bytes_sent += static_cast(result); + } + } + close(); + } + TcpFaultProxy& owner_; + int client_fd_{-1}; + int upstream_fd_{-1}; + std::atomic_bool open_{true}; + std::thread client_to_upstream_; + std::thread upstream_to_client_; + }; + + static std::string socketError(const std::string& message) { return message + ": " + std::strerror(errno); } + static void closeSocket(int& socket_fd) { + if (socket_fd < 0) return; + (void)::shutdown(socket_fd, SHUT_RDWR); + (void)::close(socket_fd); + socket_fd = -1; + } + static ssize_t sendNoSignal(int socket_fd, const void* data, std::size_t size) { +#ifdef MSG_NOSIGNAL + return ::send(socket_fd, data, size, MSG_NOSIGNAL); +#else + return ::send(socket_fd, data, size, 0); +#endif + } + static void configureSocket(int socket_fd) { +#ifdef SO_NOSIGPIPE + int suppress_sigpipe = 1; + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_NOSIGPIPE, &suppress_sigpipe, sizeof(suppress_sigpipe)); +#else + (void)socket_fd; +#endif + } + bool waitUntilResumed(const std::atomic_bool& connection_open) { + std::unique_lock lock(pause_mutex_); + pause_cv_.wait( + lock, [this, &connection_open]() { return !paused_.load() || !running_.load() || !connection_open.load(); }); + return running_.load() && connection_open.load(); + } + int connectUpstream() const { + addrinfo hints{}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + addrinfo* addresses = nullptr; + const auto port = std::to_string(upstream_port_); + if (::getaddrinfo(upstream_host_.c_str(), port.c_str(), &hints, &addresses) != 0) return -1; + int upstream_fd = -1; + for (auto* address = addresses; address != nullptr; address = address->ai_next) { + upstream_fd = ::socket(address->ai_family, address->ai_socktype, address->ai_protocol); + if (upstream_fd < 0) continue; + configureSocket(upstream_fd); + if (::connect(upstream_fd, address->ai_addr, address->ai_addrlen) == 0) break; + closeSocket(upstream_fd); + } + ::freeaddrinfo(addresses); + return upstream_fd; + } + void acceptLoop(int listener) { + while (running_.load()) { + sockaddr_storage client_address{}; + socklen_t client_address_length = sizeof(client_address); + const int client_fd = ::accept(listener, reinterpret_cast(&client_address), &client_address_length); + if (client_fd < 0) { + if (running_.load() && errno == EINTR) continue; + break; + } + configureSocket(client_fd); + const int upstream_fd = connectUpstream(); + if (upstream_fd < 0) { + int fd = client_fd; + closeSocket(fd); + continue; + } + auto connection = std::make_shared(*this, client_fd, upstream_fd); + { + const std::lock_guard lock(connections_mutex_); + connections_.push_back(connection); + } + ++accepted_connection_count_; + connection->start(); + } + } + std::vector> connectionSnapshot() const { + const std::lock_guard lock(connections_mutex_); + return connections_; + } + std::string upstream_host_; + std::uint16_t upstream_port_; + int listen_fd_{-1}; + std::uint16_t listen_port_{0}; + std::atomic_bool running_{false}; + std::atomic_bool paused_{false}; + std::atomic accepted_connection_count_{0}; + std::thread accept_thread_; + mutable std::mutex connections_mutex_; + std::vector> connections_; + std::mutex pause_mutex_; + std::condition_variable pause_cv_; +}; + +} // namespace livekit::test diff --git a/src/tests/connection/disconnect_offline_tester.cpp b/src/tests/connection/disconnect_offline_tester.cpp new file mode 100644 index 00000000..414d6ee0 --- /dev/null +++ b/src/tests/connection/disconnect_offline_tester.cpp @@ -0,0 +1,265 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "tcp_fault_proxy.h" + +namespace { + +using namespace std::chrono_literals; + +struct ServerAddress { + std::string host; + std::uint16_t port; + std::string path; +}; + +struct Options { + enum class Operation { + Disconnect, + UnpublishTrack, + }; + + std::string url; + std::string token; + std::chrono::milliseconds offline_duration{30s}; + Operation operation{Operation::Disconnect}; +}; + +/// Mirrors the application-owned delegate in issue #222. +class ShutdownDelegate final : public livekit::RoomDelegate { +public: + void onReconnecting(livekit::Room&, const livekit::ReconnectingEvent&) override { + reconnecting_.store(true); + std::cout << "ShutdownDelegate::onReconnecting invoked.\n"; + } + + void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent&) override { + std::cout << "ShutdownDelegate::onDisconnected invoked.\n"; + } + + bool reconnecting() const { return reconnecting_.load(); } + +private: + std::atomic_bool reconnecting_{false}; +}; + +[[noreturn]] void usage(const char* executable, const std::string& error = {}) { + if (!error.empty()) std::cerr << "error: " << error << '\n'; + std::cerr << "Usage: " << executable + << " [--url ws://host:port[/path]] [--token JWT] [--offline-duration-ms milliseconds]" + " [--operation disconnect|unpublish-track]\n" + "Defaults: LIVEKIT_URL and LIVEKIT_TOKEN_A. This TCP proxy supports ws:// only.\n"; + std::exit(error.empty() ? EXIT_SUCCESS : EXIT_FAILURE); +} + +std::string valueFromEnv(const char* name) { + const char* value = std::getenv(name); + return value == nullptr ? "" : value; +} + +Options parseOptions(int argc, char* argv[]) { + Options options{valueFromEnv("LIVEKIT_URL"), valueFromEnv("LIVEKIT_TOKEN_A")}; + for (int index = 1; index < argc; ++index) { + const std::string argument = argv[index]; + if (argument == "--help" || argument == "-h") usage(argv[0]); + if (index + 1 == argc) usage(argv[0], "missing value for " + argument); + const std::string value = argv[++index]; + if (argument == "--url") + options.url = value; + else if (argument == "--token") + options.token = value; + else if (argument == "--operation") { + if (value == "disconnect") + options.operation = Options::Operation::Disconnect; + else if (value == "unpublish-track") + options.operation = Options::Operation::UnpublishTrack; + else + usage(argv[0], "invalid --operation value"); + } else if (argument == "--offline-duration-ms") { + try { + options.offline_duration = std::chrono::milliseconds(std::stoll(value)); + } catch (const std::exception&) { + usage(argv[0], "invalid --offline-duration-ms value"); + } + } else { + usage(argv[0], "unknown argument " + argument); + } + } + if (options.url.empty()) usage(argv[0], "LIVEKIT_URL or --url is required"); + if (options.token.empty()) usage(argv[0], "LIVEKIT_TOKEN_A or --token is required"); + if (options.offline_duration.count() < 0) usage(argv[0], "offline duration must be non-negative"); + return options; +} + +ServerAddress parseWsUrl(const std::string& url) { + constexpr char kScheme[] = "ws://"; + if (url.compare(0, sizeof(kScheme) - 1, kScheme) != 0) { + throw std::invalid_argument("the fault proxy requires a ws:// URL (wss:// is not supported)"); + } + const std::string authority_and_path = url.substr(sizeof(kScheme) - 1); + const auto path_start = authority_and_path.find('/'); + const std::string authority = authority_and_path.substr(0, path_start); + if (authority.empty() || authority.find('@') != std::string::npos || authority.front() == '[') { + throw std::invalid_argument("URL must use a hostname or IPv4 address with an explicit port"); + } + const auto colon = authority.rfind(':'); + if (colon == std::string::npos || colon == 0 || colon + 1 == authority.size()) { + throw std::invalid_argument("URL must include an explicit port, for example ws://localhost:7880"); + } + unsigned long port = 0; + try { + port = std::stoul(authority.substr(colon + 1)); + } catch (const std::exception&) { + throw std::invalid_argument("URL contains an invalid port"); + } + if (port == 0 || port > 65535) throw std::invalid_argument("URL port is outside 1..65535"); + return {authority.substr(0, colon), static_cast(port), + path_start == std::string::npos ? "" : authority_and_path.substr(path_start)}; +} + +bool waitForConnection(const livekit::test::TcpFaultProxy& proxy, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (proxy.acceptedConnectionCount() != 0) return true; + std::this_thread::sleep_for(10ms); + } + return false; +} + +bool waitForReconnecting(const ShutdownDelegate& delegate, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + std::cout << "### Waiting for reconnecting: " << delegate.reconnecting() << "\n"; + if (delegate.reconnecting()) return true; + std::this_thread::sleep_for(10ms); + } + return false; +} + +} // namespace + +int main(int argc, char* argv[]) { + try { + std::cout << std::unitbuf; + const Options options = parseOptions(argc, argv); + const ServerAddress upstream = parseWsUrl(options.url); + livekit::test::TcpFaultProxy proxy(upstream.host, upstream.port); + proxy.start(); + const std::string proxy_url = "ws://127.0.0.1:" + std::to_string(proxy.listenPort()) + upstream.path; + + std::cout << "Connecting through " << proxy_url << " to " << options.url << '\n'; + livekit::initialize(livekit::LogLevel::Debug); + std::unique_ptr room = std::make_unique(); + auto delegate = std::make_unique(); + room->setDelegate(delegate.get()); + if (!room->connect(proxy_url, options.token, {})) { + livekit::shutdown(); + throw std::runtime_error("Room::connect failed"); + } + if (!waitForConnection(proxy, 2s)) { + (void)room->disconnect(); + livekit::shutdown(); + throw std::runtime_error("proxy did not observe the initial signal connection"); + } + + auto participant = room->localParticipant().lock(); + if (!participant) { + (void)room->disconnect(); + livekit::shutdown(); + throw std::runtime_error("connected room has no local participant"); + } + auto video_source = std::make_shared(16, 16); + auto video_track = + participant->publishVideoTrack("offline-operation-track", video_source, livekit::TrackSource::SOURCE_CAMERA); + if (!video_track || !video_track->publication()) { + (void)room->disconnect(); + livekit::shutdown(); + throw std::runtime_error("failed to publish the local video track"); + } + const std::string track_sid = video_track->publication()->sid(); + std::cout << "Published local video track " << track_sid << ".\n"; + + std::cout << "Connected. Pausing proxy and resetting " << proxy.acceptedConnectionCount() + << " established connection(s).\n"; + + // Kick off thread in parallel to pause the proxy and reset the connections. + std::thread pause_proxy([&proxy, duration = options.offline_duration]() { + std::cout << "### Pausing connection\n"; + proxy.pause(); + proxy.resetConnections(); + + // std::this_thread::sleep_for(duration); + // std::cout << "### Restoring connection after " << duration.count() << " ms.\n"; + // proxy.resume(); + }); + + if (!waitForReconnecting(*delegate, 5s)) { + std::cout << "### Disconnecting room\n"; + (void)room->disconnect(); + livekit::shutdown(); + throw std::runtime_error("room did not enter Reconnecting after the signal connection was reset"); + } + + const auto started = std::chrono::steady_clock::now(); + std::optional disconnect_result; + std::exception_ptr operation_error; + try { + if (options.operation == Options::Operation::Disconnect) { + std::cout << "Calling Room::disconnect(ClientInitiated) with a live delegate; it should not wait for the " + "network.\n"; + disconnect_result = room->disconnect(livekit::DisconnectReason::ClientInitiated); + } else { + std::cout << "Calling LocalParticipant::unpublishTrack; it should not wait for the network.\n"; + participant->unpublishTrack(track_sid); + } + } catch (...) { + operation_error = std::current_exception(); + } + const auto elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - started); + pause_proxy.join(); + if (operation_error) std::rethrow_exception(operation_error); + + if (disconnect_result.has_value()) { + std::cout << "Room::disconnect returned " << std::boolalpha << *disconnect_result; + } else { + std::cout << "LocalParticipant::unpublishTrack returned"; + (void)room->disconnect(); + } + std::cout << " after " << elapsed.count() << " ms.\n"; + + room.reset(); + delegate.reset(); + livekit::shutdown(); + return EXIT_SUCCESS; + } catch (const std::exception& error) { + std::cerr << "disconnect_offline_tester failed: " << error.what() << '\n'; + return EXIT_FAILURE; + } +} From 400db4a2a73a95b3e89c71da59ac04825f8b0599 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 31 Aug 2026 18:46:34 -0600 Subject: [PATCH 2/3] Now can reproduce it --- .../connection/disconnect_offline_tester.cpp | 198 ++++++++++-------- 1 file changed, 112 insertions(+), 86 deletions(-) diff --git a/src/tests/connection/disconnect_offline_tester.cpp b/src/tests/connection/disconnect_offline_tester.cpp index 414d6ee0..f2553505 100644 --- a/src/tests/connection/disconnect_offline_tester.cpp +++ b/src/tests/connection/disconnect_offline_tester.cpp @@ -16,16 +16,21 @@ #include +#include #include #include +#include +#include #include #include #include #include +#include #include #include #include #include +#include #include "tcp_fault_proxy.h" @@ -51,22 +56,46 @@ struct Options { Operation operation{Operation::Disconnect}; }; -/// Mirrors the application-owned delegate in issue #222. -class ShutdownDelegate final : public livekit::RoomDelegate { +/// Tracks the reconnect transition without blocking the FFI callback thread. +class ReconnectTrackingDelegate final : public livekit::RoomDelegate { public: + void onConnectionStateChanged(livekit::Room&, const livekit::ConnectionStateChangedEvent& event) override { + { + const std::scoped_lock lock(mutex_); + connected_ = event.state == livekit::ConnectionState::Connected; + } + connected_cv_.notify_all(); + } + void onReconnecting(livekit::Room&, const livekit::ReconnectingEvent&) override { - reconnecting_.store(true); - std::cout << "ShutdownDelegate::onReconnecting invoked.\n"; + { + const std::scoped_lock lock(mutex_); + reconnecting_ = true; + } + reconnecting_cv_.notify_all(); + std::cout << "ReconnectTrackingDelegate::onReconnecting invoked.\n"; } void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent&) override { - std::cout << "ShutdownDelegate::onDisconnected invoked.\n"; + std::cout << "ReconnectTrackingDelegate::onDisconnected invoked.\n"; } - bool reconnecting() const { return reconnecting_.load(); } + bool waitForConnected(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return connected_cv_.wait_for(lock, timeout, [this]() { return connected_; }); + } + + bool waitForReconnecting(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return reconnecting_cv_.wait_for(lock, timeout, [this]() { return reconnecting_; }); + } private: - std::atomic_bool reconnecting_{false}; + std::mutex mutex_; + std::condition_variable reconnecting_cv_; + std::condition_variable connected_cv_; + bool reconnecting_{false}; + bool connected_{false}; }; [[noreturn]] void usage(const char* executable, const std::string& error = {}) { @@ -143,28 +172,10 @@ ServerAddress parseWsUrl(const std::string& url) { path_start == std::string::npos ? "" : authority_and_path.substr(path_start)}; } -bool waitForConnection(const livekit::test::TcpFaultProxy& proxy, std::chrono::milliseconds timeout) { - const auto deadline = std::chrono::steady_clock::now() + timeout; - while (std::chrono::steady_clock::now() < deadline) { - if (proxy.acceptedConnectionCount() != 0) return true; - std::this_thread::sleep_for(10ms); - } - return false; -} - -bool waitForReconnecting(const ShutdownDelegate& delegate, std::chrono::milliseconds timeout) { - const auto deadline = std::chrono::steady_clock::now() + timeout; - while (std::chrono::steady_clock::now() < deadline) { - std::cout << "### Waiting for reconnecting: " << delegate.reconnecting() << "\n"; - if (delegate.reconnecting()) return true; - std::this_thread::sleep_for(10ms); - } - return false; -} - } // namespace int main(int argc, char* argv[]) { + std::atomic_bool capture_running{true}; try { std::cout << std::unitbuf; const Options options = parseOptions(argc, argv); @@ -173,90 +184,105 @@ int main(int argc, char* argv[]) { proxy.start(); const std::string proxy_url = "ws://127.0.0.1:" + std::to_string(proxy.listenPort()) + upstream.path; + setenv("RUST_LOG", "info", 1); + std::cout << "Connecting through " << proxy_url << " to " << options.url << '\n'; livekit::initialize(livekit::LogLevel::Debug); std::unique_ptr room = std::make_unique(); - auto delegate = std::make_unique(); + auto delegate = std::make_unique(); room->setDelegate(delegate.get()); if (!room->connect(proxy_url, options.token, {})) { livekit::shutdown(); throw std::runtime_error("Room::connect failed"); } - if (!waitForConnection(proxy, 2s)) { - (void)room->disconnect(); - livekit::shutdown(); - throw std::runtime_error("proxy did not observe the initial signal connection"); + + // Give some time for the connection to be established + if (!delegate->waitForConnected(10s)) { + throw std::runtime_error("Connection timed out"); } - auto participant = room->localParticipant().lock(); - if (!participant) { - (void)room->disconnect(); - livekit::shutdown(); - throw std::runtime_error("connected room has no local participant"); + auto local_participant = room->localParticipant().lock(); + if (!local_participant) { + throw std::runtime_error("Local participant invalid"); } - auto video_source = std::make_shared(16, 16); + constexpr int kAudioSampleRate = 48000; + constexpr int kAudioChannels = 1; + constexpr int kAudioSamplesPerFrame = kAudioSampleRate / 100; + constexpr int kVideoWidth = 320; + constexpr int kVideoHeight = 180; + constexpr auto kPublishDuration = 10s; + + auto audio_source = std::make_shared(kAudioSampleRate, kAudioChannels); + auto audio_track = local_participant->publishAudioTrack("offline-test-audio", audio_source, + livekit::TrackSource::SOURCE_MICROPHONE); + if (!audio_track || !audio_track->publication()) { + throw std::runtime_error("Publish audio track failed"); + } + + auto video_source = std::make_shared(kVideoWidth, kVideoHeight); auto video_track = - participant->publishVideoTrack("offline-operation-track", video_source, livekit::TrackSource::SOURCE_CAMERA); + local_participant->publishVideoTrack("offline-test-video", video_source, livekit::TrackSource::SOURCE_CAMERA); if (!video_track || !video_track->publication()) { - (void)room->disconnect(); - livekit::shutdown(); - throw std::runtime_error("failed to publish the local video track"); + throw std::runtime_error("Publish video track failed"); } - const std::string track_sid = video_track->publication()->sid(); - std::cout << "Published local video track " << track_sid << ".\n"; - std::cout << "Connected. Pausing proxy and resetting " << proxy.acceptedConnectionCount() - << " established connection(s).\n"; - - // Kick off thread in parallel to pause the proxy and reset the connections. - std::thread pause_proxy([&proxy, duration = options.offline_duration]() { - std::cout << "### Pausing connection\n"; - proxy.pause(); - proxy.resetConnections(); + std::cout << "Publishing audio and video for " << kPublishDuration.count() + << " seconds before pausing the proxy.\n"; + std::thread audio_thread([audio_source, &capture_running]() { + const livekit::AudioFrame frame = + livekit::AudioFrame::create(kAudioSampleRate, kAudioChannels, kAudioSamplesPerFrame); + auto next_frame = std::chrono::steady_clock::now(); + while (capture_running.load()) { + try { + audio_source->captureFrame(frame); + } catch (const std::exception& error) { + std::cerr << "Audio capture failed during disconnect: " << error.what() << '\n'; + } + next_frame += 10ms; + std::this_thread::sleep_until(next_frame); + } + }); - // std::this_thread::sleep_for(duration); - // std::cout << "### Restoring connection after " << duration.count() << " ms.\n"; - // proxy.resume(); + std::thread video_thread([video_source, &capture_running]() { + livekit::VideoFrame frame = + livekit::VideoFrame::create(kVideoWidth, kVideoHeight, livekit::VideoBufferType::RGBA); + std::fill_n(frame.data(), frame.dataSize(), static_cast(0x80)); + auto next_frame = std::chrono::steady_clock::now(); + while (capture_running.load()) { + try { + video_source->captureFrame(frame); + } catch (const std::exception& error) { + std::cerr << "Video capture failed during disconnect: " << error.what() << '\n'; + } + next_frame += 33ms; + std::this_thread::sleep_until(next_frame); + } }); - if (!waitForReconnecting(*delegate, 5s)) { - std::cout << "### Disconnecting room\n"; - (void)room->disconnect(); - livekit::shutdown(); - throw std::runtime_error("room did not enter Reconnecting after the signal connection was reset"); - } + std::this_thread::sleep_for(kPublishDuration); + std::cout << "Finished the 10-second connected media period; capture will continue through disconnect.\n"; - const auto started = std::chrono::steady_clock::now(); - std::optional disconnect_result; - std::exception_ptr operation_error; - try { - if (options.operation == Options::Operation::Disconnect) { - std::cout << "Calling Room::disconnect(ClientInitiated) with a live delegate; it should not wait for the " - "network.\n"; - disconnect_result = room->disconnect(livekit::DisconnectReason::ClientInitiated); - } else { - std::cout << "Calling LocalParticipant::unpublishTrack; it should not wait for the network.\n"; - participant->unpublishTrack(track_sid); - } - } catch (...) { - operation_error = std::current_exception(); - } - const auto elapsed = - std::chrono::duration_cast(std::chrono::steady_clock::now() - started); - pause_proxy.join(); - if (operation_error) std::rethrow_exception(operation_error); + std::thread proxy_thread([&proxy]() { + std::cout << "### Pausing proxy\n"; + proxy.pause(); + std::this_thread::sleep_for(120s); + std::cout << "### Resuming proxy\n"; + proxy.resume(); + }); - if (disconnect_result.has_value()) { - std::cout << "Room::disconnect returned " << std::boolalpha << *disconnect_result; - } else { - std::cout << "LocalParticipant::unpublishTrack returned"; - (void)room->disconnect(); - } - std::cout << " after " << elapsed.count() << " ms.\n"; + std::cout << "### Waiting for reconnect signal...\n"; + delegate->waitForReconnecting(60s); + std::cout << "### Disconnecting room\n"; room.reset(); + capture_running.store(false); + audio_thread.join(); + video_thread.join(); + std::cout << "### Resetting delegate\n"; delegate.reset(); + std::cout << "### Shutting down LiveKit\n"; livekit::shutdown(); + return EXIT_SUCCESS; } catch (const std::exception& error) { std::cerr << "disconnect_offline_tester failed: " << error.what() << '\n'; From dee78edea91b61ef8a7f370e447595d389adf1e7 Mon Sep 17 00:00:00 2001 From: Alan George Date: Mon, 31 Aug 2026 19:51:59 -0600 Subject: [PATCH 3/3] Try tester onlien --- .github/workflows/tests.yml | 25 ++- docs/testing.md | 38 ++--- src/tests/CMakeLists.txt | 76 +++++---- src/tests/common/tcp_fault_proxy.h | 147 +++++++++++++----- .../connection/disconnect_offline_tester.cpp | 78 ++++++++-- 5 files changed, 258 insertions(+), 106 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 92f04bf5..c39ff707 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -348,7 +348,9 @@ jobs: --gtest_output="xml:${{ env.BUILD_DIR }}\unit-test-results.xml" - name: Start livekit-server - if: matrix.e2e-testing && (inputs.integration_repeat > 0 || inputs.run_stress_tests) + # The offline-disconnect tester below always needs a local server, even + # when a manual run disables integration and stress tests. + if: matrix.e2e-testing id: livekit_server uses: livekit/dev-server-action@5d4d5337a875e2d1afd37bed03c601d159dab002 # v1.1.1 with: @@ -359,7 +361,7 @@ jobs: # Needed by token helper script - name: Install livekit-cli - if: matrix.e2e-testing && (inputs.integration_repeat > 0 || inputs.run_stress_tests) + if: matrix.e2e-testing shell: bash env: # Windows installs lk via `gh api` / `gh release download`, which need this env var @@ -431,8 +433,25 @@ jobs: --gtest_recreate_environments_when_repeating=1 \ --gtest_output=xml:${{ env.BUILD_DIR }}/stress-test-results.xml + # Keep this last: it is a standalone regression reproducer rather than + # part of the unit, integration, or stress suites. Git Bash is present + # on GitHub's Windows image and lets all matrix entries share the token + # helper and invocation. + - name: Run offline disconnect tester + if: matrix.e2e-testing + timeout-minutes: 3 + shell: bash + run: | + set -euo pipefail + source scripts/set-test-tokens.sh + tester="${{ env.BUILD_DIR }}/bin/livekit_disconnect_offline_tester" + if [[ "$RUNNER_OS" == "Windows" ]]; then + tester+=".exe" + fi + bash scripts/run-with-backtrace.sh "$tester" --offline-duration-ms 5000 + - name: Dump livekit-server log on failure - if: failure() && matrix.e2e-testing && (inputs.integration_repeat > 0 || inputs.run_stress_tests) + if: failure() && matrix.e2e-testing shell: bash run: tail -n 500 "${{ steps.livekit_server.outputs.log-path }}" || true diff --git a/docs/testing.md b/docs/testing.md index 5486c806..acc9b544 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -47,17 +47,17 @@ __Note:__ The tests require tokens and a running LiveKit server. See the section ## Offline room-operation reproducer -`livekit_disconnect_offline_tester` is a standalone POSIX-only tester for an -offline `Room::disconnect()` or `LocalParticipant::unpublishTrack()` call. It -publishes a local video track, places a loopback TCP fault proxy in front of a -`ws://` LiveKit server, resets the active signal connection, waits for the room -to enter `Reconnecting`, and invokes the selected operation while reconnect -traffic remains frozen. Forwarding resumes after the chosen observation -period. - -The tester keeps its `RoomDelegate` alive throughout disconnect. This follows -the corrected shutdown ordering from issue #222 and isolates the blocking -behavior from the original report's separate dangling-delegate risk. +`livekit_disconnect_offline_tester` is a standalone cross-platform tester for +an offline `Room::disconnect()` or `LocalParticipant::unpublishTrack()` call. It +publishes local audio and video tracks, captures media for ten seconds, stops +capture, then pauses traffic through a loopback TCP fault proxy in front of a +`ws://` LiveKit server. By default it calls the selected operation immediately, +before the room observes the failure; this matches the timing in issue #222. +Forwarding resumes after the chosen observation period. + +The tester releases application-held audio/video sources before explicit +disconnect and keeps its `RoomDelegate` alive until after disconnect returns. +This follows the corrected shutdown ordering from issue #222. Build it with the normal test build: @@ -77,17 +77,19 @@ export LIVEKIT_TOKEN_A='' # Exercise the related unpublishTrack wait reported in the follow-up. ./build-debug/bin/livekit_disconnect_offline_tester \ --operation unpublish-track --offline-duration-ms 10000 + +# Compare the separate case where LiveKit has already reported Reconnecting. +./build-debug/bin/livekit_disconnect_offline_tester \ + --operation disconnect --disconnect-timing after-reconnecting --offline-duration-ms 30000 ``` The proxy cannot be used with `wss://`: it tunnels raw TCP through `127.0.0.1`, which does not preserve the server hostname required for TLS -certificate validation. The program prints the elapsed time spent inside -the selected operation; an elapsed time near or above the offline duration -reproduces the reported blocking behavior. A healthy implementation should -return promptly without waiting for proxy forwarding to resume. With the -current Rust FFI, the unpublish variant can remain blocked even after forwarding -resumes and must be terminated manually; this reproduces the missing -`UnpublishTrackCallback`, not a fault in the tester. +certificate validation. The proxy interrupts signaling traffic only; it does +not disable direct UDP media transport. A healthy implementation should return +promptly without waiting for proxy forwarding to resume. With the current Rust +FFI, the unpublish variant can remain blocked even after forwarding resumes and +may need to be terminated manually. ## Running a local LiveKit server for tests diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 9edb410f..18c60a45 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -253,42 +253,52 @@ endif() # ============================================================================ # Standalone reproducer for offline disconnect and track-unpublish waits. -# TcpFaultProxy uses POSIX sockets, so this target is unavailable on Windows. -if(UNIX) - add_executable(livekit_disconnect_offline_tester - "${CMAKE_CURRENT_SOURCE_DIR}/connection/disconnect_offline_tester.cpp" - "${CMAKE_CURRENT_SOURCE_DIR}/common/tcp_fault_proxy.h" +add_executable(livekit_disconnect_offline_tester + "${CMAKE_CURRENT_SOURCE_DIR}/connection/disconnect_offline_tester.cpp" + "${CMAKE_CURRENT_SOURCE_DIR}/common/tcp_fault_proxy.h" +) +target_link_libraries(livekit_disconnect_offline_tester PRIVATE + livekit + $<$:ws2_32> +) +target_include_directories(livekit_disconnect_offline_tester PRIVATE + ${LIVEKIT_ROOT_DIR}/include + ${CMAKE_CURRENT_SOURCE_DIR}/common +) +target_compile_definitions(livekit_disconnect_offline_tester PRIVATE + $<$:_USE_MATH_DEFINES> +) + +if(WIN32) + add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/livekit_ffi.dll" + $ + COMMENT "Copying tester DLLs" ) - target_link_libraries(livekit_disconnect_offline_tester PRIVATE livekit) - target_include_directories(livekit_disconnect_offline_tester PRIVATE - ${LIVEKIT_ROOT_DIR}/include - ${CMAKE_CURRENT_SOURCE_DIR}/common +elseif(APPLE) + add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.dylib" + $ + COMMENT "Copying tester shared libraries" ) - target_compile_definitions(livekit_disconnect_offline_tester PRIVATE - $<$:_USE_MATH_DEFINES> +else() + add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.so" + $ + COMMENT "Copying tester shared libraries" ) - - if(APPLE) - add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD - COMMAND ${CMAKE_COMMAND} -E copy_if_different - $ - $ - COMMAND ${CMAKE_COMMAND} -E copy_if_different - "$/liblivekit_ffi.dylib" - $ - COMMENT "Copying tester shared libraries" - ) - else() - add_custom_command(TARGET livekit_disconnect_offline_tester POST_BUILD - COMMAND ${CMAKE_COMMAND} -E copy_if_different - $ - $ - COMMAND ${CMAKE_COMMAND} -E copy_if_different - "$/liblivekit_ffi.so" - $ - COMMENT "Copying tester shared libraries" - ) - endif() endif() # ============================================================================ diff --git a/src/tests/common/tcp_fault_proxy.h b/src/tests/common/tcp_fault_proxy.h index ea0ddc1d..ce18b2fc 100644 --- a/src/tests/common/tcp_fault_proxy.h +++ b/src/tests/common/tcp_fault_proxy.h @@ -16,11 +16,19 @@ #pragma once +#if defined(_WIN32) +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif +#include +#include +#else #include #include #include #include #include +#endif #include #include @@ -52,13 +60,13 @@ class TcpFaultProxy { /// Bind an ephemeral loopback port and begin accepting connections. void start() { if (running_.exchange(true)) return; + initializeSocketLibrary(); listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); - if (listen_fd_ < 0) { + if (isInvalidSocket(listen_fd_)) { running_.store(false); throw std::runtime_error(socketError("failed to create proxy listener")); } - int reuse_address = 1; - (void)::setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &reuse_address, sizeof(reuse_address)); + enableAddressReuse(listen_fd_); sockaddr_in address{}; address.sin_family = AF_INET; address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); @@ -75,7 +83,7 @@ class TcpFaultProxy { running_.store(false); throw std::runtime_error(error); } - socklen_t address_length = sizeof(address); + SocketLength address_length = sizeof(address); if (::getsockname(listen_fd_, reinterpret_cast(&address), &address_length) != 0) { const auto error = socketError("failed to resolve proxy listener port"); closeSocket(listen_fd_); @@ -83,22 +91,24 @@ class TcpFaultProxy { throw std::runtime_error(error); } listen_port_ = ntohs(address.sin_port); - const int listener = listen_fd_; + const SocketHandle listener = listen_fd_; accept_thread_ = std::thread([this, listener]() { acceptLoop(listener); }); } /// Stop the listener and all active forwarding workers. - void stop() { + void stop() noexcept { if (!running_.exchange(false)) return; paused_.store(false); pause_cv_.notify_all(); closeSocket(listen_fd_); if (accept_thread_.joinable()) accept_thread_.join(); - const auto connections = connectionSnapshot(); + std::vector> connections; + { + const std::scoped_lock lock(connections_mutex_); + connections.swap(connections_); + } for (const auto& connection : connections) connection->close(); for (const auto& connection : connections) connection->join(); - const std::lock_guard lock(connections_mutex_); - connections_.clear(); } /// Freeze traffic in both directions without closing sockets. @@ -121,9 +131,19 @@ class TcpFaultProxy { std::uint64_t acceptedConnectionCount() const { return accepted_connection_count_.load(); } private: +#if defined(_WIN32) + using SocketHandle = SOCKET; + using SocketLength = int; + static constexpr SocketHandle kInvalidSocket = INVALID_SOCKET; +#else + using SocketHandle = int; + using SocketLength = socklen_t; + static constexpr SocketHandle kInvalidSocket = -1; +#endif + class Connection { public: - Connection(TcpFaultProxy& owner, int client_fd, int upstream_fd) + Connection(TcpFaultProxy& owner, SocketHandle client_fd, SocketHandle upstream_fd) : owner_(owner), client_fd_(client_fd), upstream_fd_(upstream_fd) {} ~Connection() { close(); @@ -145,11 +165,12 @@ class TcpFaultProxy { } private: - void pump(int source_fd, int destination_fd) { - std::array buffer{}; + void pump(SocketHandle source_fd, SocketHandle destination_fd) { + constexpr std::size_t kBufferSize = static_cast(16) * 1024U; + std::array buffer{}; while (open_.load() && owner_.running_.load()) { if (!owner_.waitUntilResumed(open_)) break; - const auto bytes_read = ::recv(source_fd, buffer.data(), buffer.size(), 0); + const auto bytes_read = receive(source_fd, buffer.data(), buffer.size()); if (bytes_read <= 0) break; if (!owner_.waitUntilResumed(open_)) break; std::size_t bytes_sent = 0; @@ -166,28 +187,77 @@ class TcpFaultProxy { close(); } TcpFaultProxy& owner_; - int client_fd_{-1}; - int upstream_fd_{-1}; + SocketHandle client_fd_{kInvalidSocket}; + SocketHandle upstream_fd_{kInvalidSocket}; std::atomic_bool open_{true}; std::thread client_to_upstream_; std::thread upstream_to_client_; }; - static std::string socketError(const std::string& message) { return message + ": " + std::strerror(errno); } - static void closeSocket(int& socket_fd) { - if (socket_fd < 0) return; + static void initializeSocketLibrary() { +#if defined(_WIN32) + static std::once_flag initialized; + std::call_once(initialized, []() { + WSADATA data{}; + const int error = ::WSAStartup(MAKEWORD(2, 2), &data); + if (error != 0) { + throw std::runtime_error("failed to initialize Winsock: " + std::to_string(error)); + } + }); +#endif + } + + static bool isInvalidSocket(SocketHandle socket_fd) { return socket_fd == kInvalidSocket; } + + static std::string socketError(const std::string& message) { +#if defined(_WIN32) + return message + ": WSA error " + std::to_string(::WSAGetLastError()); +#else + return message + ": " + std::strerror(errno); +#endif + } + + static void enableAddressReuse(SocketHandle socket_fd) { + int reuse_address = 1; +#if defined(_WIN32) + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&reuse_address), + sizeof(reuse_address)); +#else + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &reuse_address, sizeof(reuse_address)); +#endif + } + + static void closeSocket(SocketHandle& socket_fd) { + if (isInvalidSocket(socket_fd)) return; +#if defined(_WIN32) + (void)::shutdown(socket_fd, SD_BOTH); + (void)::closesocket(socket_fd); +#else (void)::shutdown(socket_fd, SHUT_RDWR); (void)::close(socket_fd); - socket_fd = -1; +#endif + socket_fd = kInvalidSocket; } - static ssize_t sendNoSignal(int socket_fd, const void* data, std::size_t size) { + + static int receive(SocketHandle socket_fd, std::uint8_t* buffer, std::size_t size) { +#if defined(_WIN32) + return ::recv(socket_fd, reinterpret_cast(buffer), static_cast(size), 0); +#else + return static_cast(::recv(socket_fd, buffer, size, 0)); +#endif + } + + static int sendNoSignal(SocketHandle socket_fd, const void* data, std::size_t size) { #ifdef MSG_NOSIGNAL - return ::send(socket_fd, data, size, MSG_NOSIGNAL); + return static_cast(::send(socket_fd, data, size, MSG_NOSIGNAL)); +#elif defined(_WIN32) + return ::send(socket_fd, reinterpret_cast(data), static_cast(size), 0); #else - return ::send(socket_fd, data, size, 0); + return static_cast(::send(socket_fd, data, size, 0)); #endif } - static void configureSocket(int socket_fd) { + + static void configureSocket(SocketHandle socket_fd) { #ifdef SO_NOSIGPIPE int suppress_sigpipe = 1; (void)::setsockopt(socket_fd, SOL_SOCKET, SO_NOSIGPIPE, &suppress_sigpipe, sizeof(suppress_sigpipe)); @@ -201,17 +271,17 @@ class TcpFaultProxy { lock, [this, &connection_open]() { return !paused_.load() || !running_.load() || !connection_open.load(); }); return running_.load() && connection_open.load(); } - int connectUpstream() const { + SocketHandle connectUpstream() const { addrinfo hints{}; hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM; addrinfo* addresses = nullptr; const auto port = std::to_string(upstream_port_); - if (::getaddrinfo(upstream_host_.c_str(), port.c_str(), &hints, &addresses) != 0) return -1; - int upstream_fd = -1; + if (::getaddrinfo(upstream_host_.c_str(), port.c_str(), &hints, &addresses) != 0) return kInvalidSocket; + SocketHandle upstream_fd = kInvalidSocket; for (auto* address = addresses; address != nullptr; address = address->ai_next) { upstream_fd = ::socket(address->ai_family, address->ai_socktype, address->ai_protocol); - if (upstream_fd < 0) continue; + if (isInvalidSocket(upstream_fd)) continue; configureSocket(upstream_fd); if (::connect(upstream_fd, address->ai_addr, address->ai_addrlen) == 0) break; closeSocket(upstream_fd); @@ -219,25 +289,28 @@ class TcpFaultProxy { ::freeaddrinfo(addresses); return upstream_fd; } - void acceptLoop(int listener) { + void acceptLoop(SocketHandle listener) { while (running_.load()) { sockaddr_storage client_address{}; - socklen_t client_address_length = sizeof(client_address); - const int client_fd = ::accept(listener, reinterpret_cast(&client_address), &client_address_length); - if (client_fd < 0) { + SocketLength client_address_length = sizeof(client_address); + const SocketHandle client_fd = + ::accept(listener, reinterpret_cast(&client_address), &client_address_length); + if (isInvalidSocket(client_fd)) { +#if !defined(_WIN32) if (running_.load() && errno == EINTR) continue; +#endif break; } configureSocket(client_fd); - const int upstream_fd = connectUpstream(); - if (upstream_fd < 0) { - int fd = client_fd; + const SocketHandle upstream_fd = connectUpstream(); + if (isInvalidSocket(upstream_fd)) { + SocketHandle fd = client_fd; closeSocket(fd); continue; } auto connection = std::make_shared(*this, client_fd, upstream_fd); { - const std::lock_guard lock(connections_mutex_); + const std::scoped_lock lock(connections_mutex_); connections_.push_back(connection); } ++accepted_connection_count_; @@ -245,12 +318,12 @@ class TcpFaultProxy { } } std::vector> connectionSnapshot() const { - const std::lock_guard lock(connections_mutex_); + const std::scoped_lock lock(connections_mutex_); return connections_; } std::string upstream_host_; std::uint16_t upstream_port_; - int listen_fd_{-1}; + SocketHandle listen_fd_{kInvalidSocket}; std::uint16_t listen_port_{0}; std::atomic_bool running_{false}; std::atomic_bool paused_{false}; diff --git a/src/tests/connection/disconnect_offline_tester.cpp b/src/tests/connection/disconnect_offline_tester.cpp index f2553505..9256e7a4 100644 --- a/src/tests/connection/disconnect_offline_tester.cpp +++ b/src/tests/connection/disconnect_offline_tester.cpp @@ -45,6 +45,11 @@ struct ServerAddress { }; struct Options { + enum class DisconnectTiming { + Immediate, + AfterReconnecting, + }; + enum class Operation { Disconnect, UnpublishTrack, @@ -52,8 +57,9 @@ struct Options { std::string url; std::string token; - std::chrono::milliseconds offline_duration{30s}; + std::chrono::milliseconds offline_duration{120s}; Operation operation{Operation::Disconnect}; + DisconnectTiming disconnect_timing{DisconnectTiming::Immediate}; }; /// Tracks the reconnect transition without blocking the FFI callback thread. @@ -102,7 +108,8 @@ class ReconnectTrackingDelegate final : public livekit::RoomDelegate { if (!error.empty()) std::cerr << "error: " << error << '\n'; std::cerr << "Usage: " << executable << " [--url ws://host:port[/path]] [--token JWT] [--offline-duration-ms milliseconds]" - " [--operation disconnect|unpublish-track]\n" + " [--operation disconnect|unpublish-track]" + " [--disconnect-timing immediate|after-reconnecting]\n" "Defaults: LIVEKIT_URL and LIVEKIT_TOKEN_A. This TCP proxy supports ws:// only.\n"; std::exit(error.empty() ? EXIT_SUCCESS : EXIT_FAILURE); } @@ -112,6 +119,14 @@ std::string valueFromEnv(const char* name) { return value == nullptr ? "" : value; } +void setRustLogLevel() { +#if defined(_WIN32) + if (_putenv_s("RUST_LOG", "info") != 0) throw std::runtime_error("Unable to set RUST_LOG"); +#else + if (setenv("RUST_LOG", "info", 1) != 0) throw std::runtime_error("Unable to set RUST_LOG"); +#endif +} + Options parseOptions(int argc, char* argv[]) { Options options{valueFromEnv("LIVEKIT_URL"), valueFromEnv("LIVEKIT_TOKEN_A")}; for (int index = 1; index < argc; ++index) { @@ -130,6 +145,13 @@ Options parseOptions(int argc, char* argv[]) { options.operation = Options::Operation::UnpublishTrack; else usage(argv[0], "invalid --operation value"); + } else if (argument == "--disconnect-timing") { + if (value == "immediate") + options.disconnect_timing = Options::DisconnectTiming::Immediate; + else if (value == "after-reconnecting") + options.disconnect_timing = Options::DisconnectTiming::AfterReconnecting; + else + usage(argv[0], "invalid --disconnect-timing value"); } else if (argument == "--offline-duration-ms") { try { options.offline_duration = std::chrono::milliseconds(std::stoll(value)); @@ -184,7 +206,7 @@ int main(int argc, char* argv[]) { proxy.start(); const std::string proxy_url = "ws://127.0.0.1:" + std::to_string(proxy.listenPort()) + upstream.path; - setenv("RUST_LOG", "info", 1); + setRustLogLevel(); std::cout << "Connecting through " << proxy_url << " to " << options.url << '\n'; livekit::initialize(livekit::LogLevel::Debug); @@ -225,9 +247,10 @@ int main(int argc, char* argv[]) { if (!video_track || !video_track->publication()) { throw std::runtime_error("Publish video track failed"); } + const std::string video_track_sid = video_track->publication()->sid(); std::cout << "Publishing audio and video for " << kPublishDuration.count() - << " seconds before pausing the proxy.\n"; + << " seconds before simulating the network loss.\n"; std::thread audio_thread([audio_source, &capture_running]() { const livekit::AudioFrame frame = livekit::AudioFrame::create(kAudioSampleRate, kAudioChannels, kAudioSamplesPerFrame); @@ -260,24 +283,49 @@ int main(int argc, char* argv[]) { }); std::this_thread::sleep_for(kPublishDuration); - std::cout << "Finished the 10-second connected media period; capture will continue through disconnect.\n"; + capture_running.store(false); + audio_thread.join(); + video_thread.join(); + std::cout << "Finished the 10-second connected media period.\n"; - std::thread proxy_thread([&proxy]() { - std::cout << "### Pausing proxy\n"; - proxy.pause(); - std::this_thread::sleep_for(120s); + std::cout << "### Pausing proxy\n"; + proxy.pause(); + std::thread proxy_thread([&proxy, duration = options.offline_duration]() { + std::this_thread::sleep_for(duration); std::cout << "### Resuming proxy\n"; proxy.resume(); }); - std::cout << "### Waiting for reconnect signal...\n"; - delegate->waitForReconnecting(60s); + if (options.disconnect_timing == Options::DisconnectTiming::AfterReconnecting) { + std::cout << "### Waiting for reconnect signal...\n"; + if (!delegate->waitForReconnecting(60s)) { + proxy.resume(); + proxy_thread.join(); + capture_running.store(false); + audio_thread.join(); + video_thread.join(); + throw std::runtime_error("Room did not enter Reconnecting within 60 seconds"); + } + } else { + std::cout << "### Disconnecting immediately, before LiveKit reports Reconnecting\n"; + } + + if (options.operation == Options::Operation::Disconnect) { + // Match the corrected reporter sequence: application media sources are + // released before an explicit client-initiated room disconnect. + audio_source.reset(); + video_source.reset(); + std::cout << "### Calling Room::disconnect(ClientInitiated)\n"; + (void)room->disconnect(livekit::DisconnectReason::ClientInitiated); + } else { + std::cout << "### Calling LocalParticipant::unpublishTrack\n"; + local_participant->unpublishTrack(video_track_sid); + audio_source.reset(); + video_source.reset(); + } - std::cout << "### Disconnecting room\n"; + proxy_thread.join(); room.reset(); - capture_running.store(false); - audio_thread.join(); - video_thread.join(); std::cout << "### Resetting delegate\n"; delegate.reset(); std::cout << "### Shutting down LiveKit\n";