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 33919c0e..acc9b544 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -45,6 +45,52 @@ __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 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: + +```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 + +# 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 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 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..18c60a45 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -248,6 +248,59 @@ if(INTEGRATION_TEST_SOURCES) ) endif() +# ============================================================================ +# Connection Fault Testers +# ============================================================================ + +# Standalone reproducer for offline disconnect and track-unpublish waits. +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" + ) +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" + ) +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() + # ============================================================================ # 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..ce18b2fc --- /dev/null +++ b/src/tests/common/tcp_fault_proxy.h @@ -0,0 +1,338 @@ +/* + * 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 + +#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 +#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; + initializeSocketLibrary(); + listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); + if (isInvalidSocket(listen_fd_)) { + running_.store(false); + throw std::runtime_error(socketError("failed to create proxy listener")); + } + enableAddressReuse(listen_fd_); + 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); + } + 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_); + running_.store(false); + throw std::runtime_error(error); + } + listen_port_ = ntohs(address.sin_port); + const SocketHandle listener = listen_fd_; + accept_thread_ = std::thread([this, listener]() { acceptLoop(listener); }); + } + + /// Stop the listener and all active forwarding workers. + 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(); + 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(); + } + + /// 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: +#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, SocketHandle client_fd, SocketHandle 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(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 = receive(source_fd, buffer.data(), buffer.size()); + 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_; + SocketHandle client_fd_{kInvalidSocket}; + SocketHandle upstream_fd_{kInvalidSocket}; + std::atomic_bool open_{true}; + std::thread client_to_upstream_; + std::thread upstream_to_client_; + }; + + 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); +#endif + socket_fd = kInvalidSocket; + } + + 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 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 static_cast(::send(socket_fd, data, size, 0)); +#endif + } + + 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)); +#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(); + } + 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 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 (isInvalidSocket(upstream_fd)) 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(SocketHandle listener) { + while (running_.load()) { + sockaddr_storage client_address{}; + 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 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::scoped_lock lock(connections_mutex_); + connections_.push_back(connection); + } + ++accepted_connection_count_; + connection->start(); + } + } + std::vector> connectionSnapshot() const { + const std::scoped_lock lock(connections_mutex_); + return connections_; + } + std::string upstream_host_; + std::uint16_t upstream_port_; + SocketHandle listen_fd_{kInvalidSocket}; + 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..9256e7a4 --- /dev/null +++ b/src/tests/connection/disconnect_offline_tester.cpp @@ -0,0 +1,339 @@ +/* + * 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 +#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 DisconnectTiming { + Immediate, + AfterReconnecting, + }; + + enum class Operation { + Disconnect, + UnpublishTrack, + }; + + std::string url; + std::string token; + 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. +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 { + { + 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 << "ReconnectTrackingDelegate::onDisconnected invoked.\n"; + } + + 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::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 = {}) { + 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]" + " [--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); +} + +std::string valueFromEnv(const char* name) { + const char* value = std::getenv(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) { + 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 == "--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)); + } 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)}; +} + +} // namespace + +int main(int argc, char* argv[]) { + std::atomic_bool capture_running{true}; + 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; + + setRustLogLevel(); + + 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"); + } + + // Give some time for the connection to be established + if (!delegate->waitForConnected(10s)) { + throw std::runtime_error("Connection timed out"); + } + + auto local_participant = room->localParticipant().lock(); + if (!local_participant) { + throw std::runtime_error("Local participant invalid"); + } + 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 = + local_participant->publishVideoTrack("offline-test-video", video_source, livekit::TrackSource::SOURCE_CAMERA); + 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 simulating the network loss.\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::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); + } + }); + + std::this_thread::sleep_for(kPublishDuration); + capture_running.store(false); + audio_thread.join(); + video_thread.join(); + std::cout << "Finished the 10-second connected media period.\n"; + + 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(); + }); + + 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(); + } + + proxy_thread.join(); + room.reset(); + 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'; + return EXIT_FAILURE; + } +}