SmartBench/cmvr-es
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
05ab64cf76
cmvr-es/test/quic_gateway/src/quic_test_gateway.cpp
xtkuang 428ee328de feat: complete QUIC edge integration 
Vendor MsQuic with build and install support, add DeviceManager status to configurable heartbeats, and report only enabled devices.

Add the local QUIC gateway, protocol coverage, real MsQuic E2E tests, process smoke tests, and updated integration documentation.
2026-07-24 12:35:04 +08:00

2132 lines
81 KiB
C++
Raw Blame History

#include "quic_test_gateway.h"
#include "media_reassembler.h"
#include <msquic.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <deque>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <limits>
#include <memory>
#include <mutex>
#include <new>
#include <sstream>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>
#include "cmvr/quic_edge/v1/quic_edge.pb.h"
#include "service/quic_edge/include/control_framing.h"
#include "service/quic_edge/include/datagram_packetizer.h"
namespace cmvr::test::quic_gateway {
namespace {
using Envelope = cmvr::quic_edge::v1::EdgeControlEnvelope;
using Clock = std::chrono::steady_clock;
constexpr QUIC_UINT62 kApplicationShutdownCode = 0x100U;
constexpr QUIC_UINT62 kProtocolViolationCode = 0x101U;
constexpr std::uint64_t kProtocolErrorCode = 1U;
constexpr std::uint64_t kDefaultIdleTimeoutMs = 120000U;
constexpr std::uint64_t kFnv1a64OffsetBasis = 14695981039346656037ULL;
constexpr std::uint64_t kFnv1a64Prime = 1099511628211ULL;
std::uint64_t fnv1a64(const std::uint8_t* data, const std::size_t size)
{
std::uint64_t hash = kFnv1a64OffsetBasis;
for (std::size_t index = 0; index < size; ++index) {
hash ^= data[index];
hash *= kFnv1a64Prime;
}
return hash;
}
std::uint64_t fnv1a64(const std::string& data)
{
return fnv1a64(
reinterpret_cast<const std::uint8_t*>(data.data()), data.size());
}
struct TrackDescriptorObservation {
std::uint32_t track_id{0U};
std::string device_id;
std::string source_track_id;
std::string codec;
std::string payload_format;
std::uint64_t codec_generation{0U};
std::uint32_t codec_generation_token{0U};
std::uint32_t width{0U};
std::uint32_t height{0U};
std::uint32_t frames_per_second{0U};
std::uint32_t sample_rate{0U};
std::uint32_t channels{0U};
std::size_t codec_config_bytes{0U};
std::uint64_t codec_config_hash{kFnv1a64OffsetBasis};
};
struct CompletedFrameObservation {
bool present{false};
std::uint32_t track_id{0U};
std::uint64_t frame_sequence{0U};
std::size_t frame_bytes{0U};
std::uint64_t frame_hash{kFnv1a64OffsetBasis};
std::uint16_t flags{0U};
std::uint64_t capture_timestamp_us{0U};
};
std::string jsonEscape(const std::string& input)
{
std::ostringstream output;
for (const unsigned char character : input) {
switch (character) {
case '"': output << "\\\""; break;
case '\\': output << "\\\\"; break;
case '\b': output << "\\b"; break;
case '\f': output << "\\f"; break;
case '\n': output << "\\n"; break;
case '\r': output << "\\r"; break;
case '\t': output << "\\t"; break;
default:
if (character < 0x20U) {
output << "\\u"
<< std::hex << std::setw(4) << std::setfill('0')
<< static_cast<unsigned>(character)
<< std::dec << std::setfill(' ');
} else {
output << static_cast<char>(character);
}
break;
}
}
return output.str();
}
std::string statusText(const char* operation, const QUIC_STATUS status)
{
std::ostringstream output;
output << operation << " failed with QUIC_STATUS 0x"
<< std::hex << static_cast<std::uint64_t>(status);
return output.str();
}
std::string addressWithPort(const QUIC_ADDR& address)
{
QUIC_ADDR_STR text{};
if (QuicAddrToString(&address, &text) == FALSE) return {};
return text.Address;
}
std::string addressWithoutPort(const QUIC_ADDR& address)
{
char text[INET6_ADDRSTRLEN] = {};
const void* bytes = nullptr;
int family = AF_UNSPEC;
if (QuicAddrGetFamily(&address) == QUIC_ADDRESS_FAMILY_INET) {
family = AF_INET;
bytes = &address.Ipv4.sin_addr;
} else if (QuicAddrGetFamily(&address) == QUIC_ADDRESS_FAMILY_INET6) {
family = AF_INET6;
bytes = &address.Ipv6.sin6_addr;
}
if (!bytes || !inet_ntop(family, bytes, text, sizeof(text))) return {};
return text;
}
bool writeTextFile(const std::string& path,
const std::string& contents,
std::string* error)
{
if (path.empty()) return true;
std::ofstream output(path, std::ios::out | std::ios::trunc);
if (!output) {
if (error) *error = "failed to open output file: " + path;
return false;
}
output << contents << '\n';
output.flush();
if (!output) {
if (error) *error = "failed to write output file: " + path;
return false;
}
return true;
}
} // namespace
const char* scenarioName(const Scenario scenario)
{
switch (scenario) {
case Scenario::NORMAL: return "normal";
case Scenario::REJECT_REGISTRATION: return "reject-registration";
case Scenario::DROP_HEARTBEAT_ACK: return "drop-heartbeat-ack";
case Scenario::WRONG_ACK_SESSION: return "wrong-ack-session";
case Scenario::FATAL_PROTOCOL_ERROR: return "fatal-protocol-error";
case Scenario::NONFATAL_PROTOCOL_ERROR: return "nonfatal-protocol-error";
case Scenario::DATAGRAM_DISABLED: return "datagram-disabled";
}
return "unknown";
}
bool parseScenario(const std::string& text, Scenario* scenario)
{
if (!scenario) return false;
static const std::pair<const char*, Scenario> values[] = {
{"normal", Scenario::NORMAL},
{"reject-registration", Scenario::REJECT_REGISTRATION},
{"drop-heartbeat-ack", Scenario::DROP_HEARTBEAT_ACK},
{"wrong-ack-session", Scenario::WRONG_ACK_SESSION},
{"fatal-protocol-error", Scenario::FATAL_PROTOCOL_ERROR},
{"nonfatal-protocol-error", Scenario::NONFATAL_PROTOCOL_ERROR},
{"datagram-disabled", Scenario::DATAGRAM_DISABLED},
};
for (const auto& value : values) {
if (text == value.first) {
*scenario = value.second;
return true;
}
}
return false;
}
class QuicTestGateway::Impl {
public:
explicit Impl(GatewayOptions options)
: options_(std::move(options))
{
}
~Impl()
{
stop();
}
bool start(std::string* error)
{
std::lock_guard lifecycle_lock(lifecycle_mutex_);
if (started_.load()) return true;
if (start_attempted_ || stopped_) {
if (error) {
*error =
"QUIC test gateway instances are single-use";
}
return false;
}
start_attempted_ = true;
if (!validateOptions(error)) return false;
QUIC_STATUS status = MsQuicOpen2(&api_);
if (QUIC_FAILED(status)) {
return failStart(statusText("MsQuicOpen2", status), error);
}
const QUIC_REGISTRATION_CONFIG registration_config = {
"cmvr-quic-test-gateway", QUIC_EXECUTION_PROFILE_LOW_LATENCY};
status = api_->RegistrationOpen(&registration_config, &registration_);
if (QUIC_FAILED(status)) {
return failStart(statusText("RegistrationOpen", status), error);
}
QUIC_BUFFER alpn{};
alpn.Buffer = reinterpret_cast<std::uint8_t*>(
const_cast<char*>(options_.alpn.data()));
alpn.Length = static_cast<std::uint32_t>(options_.alpn.size());
QUIC_SETTINGS settings{};
settings.IsSet.PeerBidiStreamCount = TRUE;
settings.PeerBidiStreamCount = 1U;
settings.IsSet.PeerUnidiStreamCount = TRUE;
settings.PeerUnidiStreamCount = 0U;
settings.IsSet.IdleTimeoutMs = TRUE;
settings.IdleTimeoutMs = kDefaultIdleTimeoutMs;
settings.IsSet.DatagramReceiveEnabled = TRUE;
settings.DatagramReceiveEnabled =
options_.scenario == Scenario::DATAGRAM_DISABLED ? FALSE : TRUE;
status = api_->ConfigurationOpen(
registration_, &alpn, 1U, &settings, sizeof(settings),
nullptr, &configuration_);
if (QUIC_FAILED(status)) {
return failStart(statusText("ConfigurationOpen", status), error);
}
QUIC_CERTIFICATE_FILE certificate{};
certificate.PrivateKeyFile = options_.private_key_file.c_str();
certificate.CertificateFile = options_.certificate_file.c_str();
QUIC_CREDENTIAL_CONFIG credential{};
credential.Type = QUIC_CREDENTIAL_TYPE_CERTIFICATE_FILE;
credential.Flags = QUIC_CREDENTIAL_FLAG_NONE;
credential.CertificateFile = &certificate;
status = api_->ConfigurationLoadCredential(
configuration_, &credential);
if (QUIC_FAILED(status)) {
return failStart(
statusText("ConfigurationLoadCredential", status), error);
}
status = api_->ListenerOpen(
registration_, &Impl::listenerCallback, this, &listener_);
if (QUIC_FAILED(status)) {
return failStart(statusText("ListenerOpen", status), error);
}
QUIC_ADDR address{};
if (QuicAddrFromString(
options_.bind_address.c_str(), options_.port, &address) == FALSE) {
return failStart(
"bind address must be a numeric IPv4 or IPv6 address", error);
}
startWorker();
status = api_->ListenerStart(listener_, &alpn, 1U, &address);
if (QUIC_FAILED(status)) {
return failStart(statusText("ListenerStart", status), error);
}
listener_started_ = true;
QUIC_ADDR bound_address{};
std::uint32_t bound_address_size = sizeof(bound_address);
status = api_->GetParam(
listener_, QUIC_PARAM_LISTENER_LOCAL_ADDRESS,
&bound_address_size, &bound_address);
if (QUIC_FAILED(status) || bound_address_size != sizeof(bound_address)) {
return failStart(statusText(
"GetParam(QUIC_PARAM_LISTENER_LOCAL_ADDRESS)", status), error);
}
bound_port_.store(QuicAddrGetPort(&bound_address));
bound_address_text_ = addressWithPort(bound_address);
started_at_ = Clock::now();
started_.store(true);
const std::string ready = readyJson();
logJson(ready);
std::string file_error;
if (!writeTextFile(options_.ready_file, ready, &file_error)) {
return failStart(file_error, error);
}
return true;
}
void stop()
{
std::unique_lock lifecycle_lock(lifecycle_mutex_);
if (stopped_) return;
stopped_ = true;
started_.store(false);
if (listener_ && listener_started_ && api_) {
api_->ListenerStop(listener_);
std::unique_lock listener_lock(listener_mutex_);
listener_condition_.wait_for(
listener_lock, std::chrono::seconds(5),
[this]() { return listener_stop_complete_; });
listener_started_ = false;
}
if (registration_ && api_) {
api_->RegistrationShutdown(
registration_, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE,
kApplicationShutdownCode);
}
{
std::unique_lock connection_lock(connection_mutex_);
connection_condition_.wait(connection_lock, [this]() {
return connections_.empty() && live_connection_contexts_ == 0U;
});
}
stopWorker();
if (listener_ && api_) {
api_->ListenerClose(listener_);
listener_ = nullptr;
}
if (configuration_ && api_) {
api_->ConfigurationClose(configuration_);
configuration_ = nullptr;
}
if (registration_ && api_) {
api_->RegistrationClose(registration_);
registration_ = nullptr;
}
if (api_) {
MsQuicClose(api_);
api_ = nullptr;
}
const std::string summary = summaryJson();
logJson(summary);
std::string file_error;
if (!writeTextFile(options_.summary_file, summary, &file_error)) {
setRuntimeFailure(file_error);
std::cerr << file_error << '\n';
}
}
std::uint16_t boundPort() const
{
return bound_port_.load();
}
bool completionReached() const
{
if (options_.exit_after_registrations != 0U &&
stats_.registrations_received.load() >=
options_.exit_after_registrations) {
return true;
}
if (options_.exit_after_heartbeats != 0U &&
stats_.heartbeats_received.load() >=
options_.exit_after_heartbeats) {
return true;
}
if (options_.run_for_ms != 0U && started_.load() &&
Clock::now() - started_at_ >=
std::chrono::milliseconds(options_.run_for_ms)) {
return true;
}
return false;
}
bool hasRuntimeFailure() const
{
return runtime_failed_.load();
}
std::string lastError() const
{
std::lock_guard lock(error_mutex_);
return last_error_;
}
std::string summaryJson() const
{
std::string last_node_id;
std::string last_boot_id;
std::string last_session_id;
std::string last_grpc_endpoint_host;
std::string last_observed_source_ip;
std::uint32_t last_grpc_endpoint_port = 0U;
std::uint64_t registration_interface_count = 0U;
std::uint64_t heartbeat_interface_count = 0U;
bool heartbeat_has_device_manager = false;
std::string heartbeat_device_manager_name;
std::string heartbeat_device_manager_version;
std::string heartbeat_device_manager_description;
std::uint64_t heartbeat_device_count = 0U;
std::uint64_t heartbeat_enabled_device_count = 0U;
std::uint64_t heartbeat_disabled_device_count = 0U;
std::uint64_t heartbeat_error_device_count = 0U;
std::uint64_t heartbeat_unknown_health_device_count = 0U;
std::vector<cmvr::quic_edge::v1::ManagedDeviceStatus>
heartbeat_devices;
TrackDescriptorObservation video_descriptor;
TrackDescriptorObservation audio_descriptor;
CompletedFrameObservation video_frame;
CompletedFrameObservation audio_frame;
{
std::lock_guard lock(observation_mutex_);
last_node_id = last_node_id_;
last_boot_id = last_boot_id_;
last_session_id = last_session_id_;
last_grpc_endpoint_host = last_grpc_endpoint_host_;
last_grpc_endpoint_port = last_grpc_endpoint_port_;
last_observed_source_ip = last_observed_source_ip_;
registration_interface_count = registration_interface_count_;
heartbeat_interface_count = heartbeat_interface_count_;
heartbeat_has_device_manager = heartbeat_has_device_manager_;
heartbeat_device_manager_name = heartbeat_device_manager_name_;
heartbeat_device_manager_version =
heartbeat_device_manager_version_;
heartbeat_device_manager_description =
heartbeat_device_manager_description_;
heartbeat_device_count = heartbeat_device_count_;
heartbeat_enabled_device_count =
heartbeat_enabled_device_count_;
heartbeat_disabled_device_count =
heartbeat_disabled_device_count_;
heartbeat_error_device_count = heartbeat_error_device_count_;
heartbeat_unknown_health_device_count =
heartbeat_unknown_health_device_count_;
heartbeat_devices = heartbeat_devices_;
video_descriptor = video_descriptor_;
audio_descriptor = audio_descriptor_;
video_frame = maximum_video_frame_;
audio_frame = maximum_audio_frame_;
}
std::ostringstream output;
output << "{\"event\":\"summary\""
<< ",\"scenario\":\"" << scenarioName(options_.scenario) << '"'
<< ",\"bind\":\"" << jsonEscape(bound_address_text_) << '"'
<< ",\"port\":" << bound_port_.load()
<< ",\"runtime_failed\":"
<< (runtime_failed_.load() ? "true" : "false")
<< ",\"last_error\":\"" << jsonEscape(lastError()) << '"'
<< ",\"last_node_id\":\"" << jsonEscape(last_node_id) << '"'
<< ",\"last_boot_id\":\"" << jsonEscape(last_boot_id) << '"'
<< ",\"last_session_id\":\""
<< jsonEscape(last_session_id) << '"'
<< ",\"last_grpc_endpoint_host\":\""
<< jsonEscape(last_grpc_endpoint_host) << '"'
<< ",\"last_grpc_endpoint_port\":"
<< last_grpc_endpoint_port
<< ",\"last_observed_source_ip\":\""
<< jsonEscape(last_observed_source_ip) << '"'
<< ",\"registration_interface_count\":"
<< registration_interface_count
<< ",\"heartbeat_interface_count\":"
<< heartbeat_interface_count
<< ",\"heartbeat_has_device_manager\":"
<< (heartbeat_has_device_manager ? "true" : "false")
<< ",\"heartbeat_device_manager_name\":\""
<< jsonEscape(heartbeat_device_manager_name) << '"'
<< ",\"heartbeat_device_manager_version\":\""
<< jsonEscape(heartbeat_device_manager_version) << '"'
<< ",\"heartbeat_device_manager_description\":\""
<< jsonEscape(heartbeat_device_manager_description) << '"'
<< ",\"heartbeat_device_count\":"
<< heartbeat_device_count
<< ",\"heartbeat_enabled_device_count\":"
<< heartbeat_enabled_device_count
<< ",\"heartbeat_disabled_device_count\":"
<< heartbeat_disabled_device_count
<< ",\"heartbeat_error_device_count\":"
<< heartbeat_error_device_count
<< ",\"heartbeat_unknown_health_device_count\":"
<< heartbeat_unknown_health_device_count;
output << ",\"heartbeat_devices\":[";
for (std::size_t index = 0U;
index < heartbeat_devices.size(); ++index) {
if (index != 0U) output << ',';
const auto& device = heartbeat_devices[index];
output << "{\"device_id\":\""
<< jsonEscape(device.device_id()) << '"'
<< ",\"kind\":" << static_cast<int>(device.kind())
<< ",\"type_name\":\""
<< jsonEscape(device.type_name()) << '"'
<< ",\"enabled\":"
<< (device.enabled() ? "true" : "false")
<< ",\"manager_state\":"
<< static_cast<int>(device.manager_state())
<< ",\"health\":"
<< static_cast<int>(device.health())
<< ",\"has_error\":"
<< (device.has_error() ? "true" : "false")
<< ",\"error_message\":\""
<< jsonEscape(device.error_message()) << '"'
<< ",\"status_updated_at_unix_ms\":"
<< device.status_updated_at_unix_ms()
<< '}';
}
output << ']'
<< ",\"video_track_id\":" << video_descriptor.track_id
<< ",\"video_device_id\":\""
<< jsonEscape(video_descriptor.device_id) << '"'
<< ",\"video_source_track_id\":\""
<< jsonEscape(video_descriptor.source_track_id) << '"'
<< ",\"video_codec\":\""
<< jsonEscape(video_descriptor.codec) << '"'
<< ",\"video_payload_format\":\""
<< jsonEscape(video_descriptor.payload_format) << '"'
<< ",\"video_codec_generation\":"
<< video_descriptor.codec_generation
<< ",\"video_codec_generation_token\":"
<< video_descriptor.codec_generation_token
<< ",\"video_width\":" << video_descriptor.width
<< ",\"video_height\":" << video_descriptor.height
<< ",\"video_frames_per_second\":"
<< video_descriptor.frames_per_second
<< ",\"video_codec_config_bytes\":"
<< video_descriptor.codec_config_bytes
<< ",\"video_codec_config_hash\":"
<< video_descriptor.codec_config_hash
<< ",\"audio_track_id\":" << audio_descriptor.track_id
<< ",\"audio_device_id\":\""
<< jsonEscape(audio_descriptor.device_id) << '"'
<< ",\"audio_source_track_id\":\""
<< jsonEscape(audio_descriptor.source_track_id) << '"'
<< ",\"audio_codec\":\""
<< jsonEscape(audio_descriptor.codec) << '"'
<< ",\"audio_payload_format\":\""
<< jsonEscape(audio_descriptor.payload_format) << '"'
<< ",\"audio_codec_generation\":"
<< audio_descriptor.codec_generation
<< ",\"audio_codec_generation_token\":"
<< audio_descriptor.codec_generation_token
<< ",\"audio_sample_rate\":"
<< audio_descriptor.sample_rate
<< ",\"audio_channels\":" << audio_descriptor.channels
<< ",\"audio_codec_config_bytes\":"
<< audio_descriptor.codec_config_bytes
<< ",\"audio_codec_config_hash\":"
<< audio_descriptor.codec_config_hash
<< ",\"maximum_video_frame_sequence\":"
<< video_frame.frame_sequence
<< ",\"maximum_video_frame_track_id\":"
<< video_frame.track_id
<< ",\"maximum_video_frame_bytes\":"
<< video_frame.frame_bytes
<< ",\"maximum_video_frame_hash\":"
<< video_frame.frame_hash
<< ",\"maximum_video_frame_flags\":"
<< video_frame.flags
<< ",\"maximum_video_capture_timestamp_us\":"
<< video_frame.capture_timestamp_us
<< ",\"maximum_audio_frame_sequence\":"
<< audio_frame.frame_sequence
<< ",\"maximum_audio_frame_track_id\":"
<< audio_frame.track_id
<< ",\"maximum_audio_frame_bytes\":"
<< audio_frame.frame_bytes
<< ",\"maximum_audio_frame_hash\":"
<< audio_frame.frame_hash
<< ",\"maximum_audio_frame_flags\":"
<< audio_frame.flags
<< ",\"maximum_audio_capture_timestamp_us\":"
<< audio_frame.capture_timestamp_us
<< ",\"connections_accepted\":"
<< stats_.connections_accepted.load()
<< ",\"connections_closed\":"
<< stats_.connections_closed.load()
<< ",\"registrations_received\":"
<< stats_.registrations_received.load()
<< ",\"registrations_accepted\":"
<< stats_.registrations_accepted.load()
<< ",\"registrations_rejected\":"
<< stats_.registrations_rejected.load()
<< ",\"heartbeats_received\":"
<< stats_.heartbeats_received.load()
<< ",\"heartbeat_acks_sent\":"
<< stats_.heartbeat_acks_sent.load()
<< ",\"heartbeat_acks_dropped\":"
<< stats_.heartbeat_acks_dropped.load()
<< ",\"protocol_errors_sent\":"
<< stats_.protocol_errors_sent.load()
<< ",\"protocol_violations\":"
<< stats_.protocol_violations.load()
<< ",\"media_sessions_opened\":"
<< stats_.media_sessions_opened.load()
<< ",\"track_descriptors_received\":"
<< stats_.track_descriptors_received.load()
<< ",\"datagrams_received\":"
<< stats_.datagrams_received.load()
<< ",\"datagrams_invalid\":"
<< stats_.datagrams_invalid.load()
<< ",\"datagrams_queue_dropped\":"
<< stats_.datagrams_queue_dropped.load()
<< ",\"frames_completed\":"
<< stats_.frames_completed.load()
<< ",\"video_frames_completed\":"
<< stats_.video_frames_completed.load()
<< ",\"audio_frames_completed\":"
<< stats_.audio_frames_completed.load()
<< ",\"frame_bytes_completed\":"
<< stats_.frame_bytes_completed.load()
<< ",\"incomplete_frames_expired\":"
<< stats_.incomplete_frames_expired.load()
<< ",\"incomplete_frames_evicted\":"
<< stats_.incomplete_frames_evicted.load()
<< ",\"incomplete_frames_cleared\":"
<< stats_.incomplete_frames_cleared.load()
<< '}';
return output.str();
}
private:
class Connection;
struct AtomicStats {
std::atomic<std::uint64_t> connections_accepted{0};
std::atomic<std::uint64_t> connections_closed{0};
std::atomic<std::uint64_t> registrations_received{0};
std::atomic<std::uint64_t> registrations_accepted{0};
std::atomic<std::uint64_t> registrations_rejected{0};
std::atomic<std::uint64_t> heartbeats_received{0};
std::atomic<std::uint64_t> heartbeat_acks_sent{0};
std::atomic<std::uint64_t> heartbeat_acks_dropped{0};
std::atomic<std::uint64_t> protocol_errors_sent{0};
std::atomic<std::uint64_t> protocol_violations{0};
std::atomic<std::uint64_t> media_sessions_opened{0};
std::atomic<std::uint64_t> track_descriptors_received{0};
std::atomic<std::uint64_t> datagrams_received{0};
std::atomic<std::uint64_t> datagrams_invalid{0};
std::atomic<std::uint64_t> datagrams_queue_dropped{0};
std::atomic<std::uint64_t> frames_completed{0};
std::atomic<std::uint64_t> video_frames_completed{0};
std::atomic<std::uint64_t> audio_frames_completed{0};
std::atomic<std::uint64_t> frame_bytes_completed{0};
std::atomic<std::uint64_t> incomplete_frames_expired{0};
std::atomic<std::uint64_t> incomplete_frames_evicted{0};
std::atomic<std::uint64_t> incomplete_frames_cleared{0};
};
struct WorkItem {
Connection* connection{nullptr};
std::vector<std::uint8_t> bytes;
};
struct SendContext {
explicit SendContext(std::vector<std::uint8_t> value)
: bytes(std::move(value))
{
buffer.Buffer = bytes.data();
buffer.Length = static_cast<std::uint32_t>(bytes.size());
}
std::vector<std::uint8_t> bytes;
QUIC_BUFFER buffer{};
};
struct TrackInfo {
cmvr::quic_edge::v1::MediaKind kind{
cmvr::quic_edge::v1::MEDIA_KIND_UNSPECIFIED};
std::uint32_t generation_token{0};
std::uint64_t generation{0};
std::string codec;
std::string source_track_id;
};
void observeTrackDescriptor(
const cmvr::quic_edge::v1::MediaTrackDescriptor& descriptor)
{
TrackDescriptorObservation observation;
observation.track_id = descriptor.track_id();
observation.device_id = descriptor.device_id();
observation.source_track_id = descriptor.source_track_id();
observation.codec = descriptor.codec();
observation.payload_format = descriptor.payload_format();
observation.codec_generation = descriptor.codec_generation();
observation.codec_generation_token =
descriptor.codec_generation_token();
observation.width = descriptor.width();
observation.height = descriptor.height();
observation.frames_per_second = descriptor.frames_per_second();
observation.sample_rate = descriptor.sample_rate();
observation.channels = descriptor.channels();
observation.codec_config_bytes = descriptor.codec_config().size();
observation.codec_config_hash = fnv1a64(descriptor.codec_config());
std::lock_guard lock(observation_mutex_);
if (descriptor.kind() ==
cmvr::quic_edge::v1::MEDIA_KIND_VIDEO) {
video_descriptor_ = std::move(observation);
} else if (
descriptor.kind() ==
cmvr::quic_edge::v1::MEDIA_KIND_AUDIO) {
audio_descriptor_ = std::move(observation);
}
}
void observeCompletedFrame(
const cmvr::quic_edge::v1::MediaKind kind,
const cmvr::quic_edge::DatagramHeader& header,
const std::vector<std::uint8_t>& payload)
{
std::lock_guard lock(observation_mutex_);
CompletedFrameObservation* observation = nullptr;
if (kind == cmvr::quic_edge::v1::MEDIA_KIND_VIDEO) {
observation = &maximum_video_frame_;
} else if (kind == cmvr::quic_edge::v1::MEDIA_KIND_AUDIO) {
observation = &maximum_audio_frame_;
}
if (!observation ||
(observation->present &&
header.frame_sequence <= observation->frame_sequence)) {
return;
}
observation->present = true;
observation->track_id = header.track_id;
observation->frame_sequence = header.frame_sequence;
observation->frame_bytes = payload.size();
observation->frame_hash = fnv1a64(payload.data(), payload.size());
observation->flags = header.flags;
observation->capture_timestamp_us = header.capture_timestamp_us;
}
class Connection {
public:
Connection(Impl* gateway,
HQUIC handle,
const QUIC_ADDR* remote_address)
: gateway_(gateway),
api_(gateway->api_),
connection_(handle),
decoder_(gateway->options_.maximum_control_frame_bytes),
reassembler_(
gateway->options_.maximum_reassembly_bytes,
gateway->options_.maximum_reassembly_frames,
gateway->options_.maximum_frame_bytes,
std::chrono::milliseconds(
gateway->options_.reassembly_timeout_ms))
{
if (remote_address) {
remote_address_ = *remote_address;
remote_address_text_ = addressWithPort(remote_address_);
remote_ip_ = addressWithoutPort(remote_address_);
}
gateway_->connectionContextCreated();
}
void addRef()
{
references_.fetch_add(1U, std::memory_order_relaxed);
}
void release()
{
if (references_.fetch_sub(1U, std::memory_order_acq_rel) == 1U) {
delete this;
}
}
void abandonRejectedConnection()
{
std::lock_guard operation_lock(operation_mutex_);
connection_ = nullptr;
}
QUIC_STATUS onConnectionEvent(
HQUIC connection, QUIC_CONNECTION_EVENT* event)
{
switch (event->Type) {
case QUIC_CONNECTION_EVENT_CONNECTED:
connected_.store(true);
gateway_->logEvent(
"connection-established",
"\"peer\":\"" + jsonEscape(remote_address_text_) + "\"");
break;
case QUIC_CONNECTION_EVENT_PEER_STREAM_STARTED:
acceptPeerStream(
event->PEER_STREAM_STARTED.Stream,
event->PEER_STREAM_STARTED.Flags);
break;
case QUIC_CONNECTION_EVENT_DATAGRAM_RECEIVED:
receiveDatagram(event->DATAGRAM_RECEIVED.Buffer);
break;
case QUIC_CONNECTION_EVENT_PEER_ADDRESS_CHANGED:
if (event->PEER_ADDRESS_CHANGED.Address) {
std::lock_guard state_lock(state_mutex_);
remote_address_ =
*event->PEER_ADDRESS_CHANGED.Address;
remote_address_text_ =
addressWithPort(remote_address_);
remote_ip_ = addressWithoutPort(remote_address_);
}
break;
case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT:
connected_.store(false);
gateway_->logEvent(
"connection-transport-shutdown",
"\"status\":\"" +
jsonEscape(statusText(
"transport",
event->SHUTDOWN_INITIATED_BY_TRANSPORT.Status)) +
"\"");
break;
case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_PEER:
connected_.store(false);
gateway_->logEvent(
"connection-peer-shutdown",
"\"error_code\":" +
std::to_string(
event->SHUTDOWN_INITIATED_BY_PEER.ErrorCode));
break;
case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE:
completeConnectionShutdown(connection);
break;
default:
break;
}
return QUIC_STATUS_SUCCESS;
}
QUIC_STATUS onStreamEvent(HQUIC stream, QUIC_STREAM_EVENT* event)
{
switch (event->Type) {
case QUIC_STREAM_EVENT_RECEIVE:
return receiveControl(event);
case QUIC_STREAM_EVENT_SEND_COMPLETE:
delete static_cast<SendContext*>(
event->SEND_COMPLETE.ClientContext);
break;
case QUIC_STREAM_EVENT_PEER_SEND_ABORTED:
case QUIC_STREAM_EVENT_PEER_RECEIVE_ABORTED:
connected_.store(false);
break;
case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE:
completeStreamShutdown(stream);
break;
default:
break;
}
return QUIC_STATUS_SUCCESS;
}
void processControlChunk(const std::vector<std::uint8_t>& bytes)
{
std::vector<std::vector<std::uint8_t>> frames;
std::string error;
if (!decoder_.push(bytes, &frames, &error)) {
protocolViolation(error, 0U);
return;
}
for (const auto& frame : frames) {
if (!processControlFrame(frame)) return;
}
}
void processDatagram(const std::vector<std::uint8_t>& bytes)
{
cmvr::quic_edge::DatagramHeader header;
std::string error;
if (!cmvr::quic_edge::DatagramPacketizer::decodeHeader(
bytes, &header, &error)) {
gateway_->stats_.datagrams_invalid.fetch_add(1U);
gateway_->logEvent(
"datagram-invalid",
"\"reason\":\"" + jsonEscape(error) + "\"");
return;
}
constexpr std::uint16_t allowed_flags =
cmvr::quic_edge::DATAGRAM_FLAG_KEY_FRAME |
cmvr::quic_edge::DATAGRAM_FLAG_DISCONTINUITY;
if ((header.flags &
static_cast<std::uint16_t>(~allowed_flags)) != 0U) {
gateway_->stats_.datagrams_invalid.fetch_add(1U);
gateway_->logEvent(
"datagram-invalid",
"\"reason\":\"unsupported flags\"");
return;
}
TrackInfo track;
{
std::lock_guard state_lock(state_mutex_);
const auto found = tracks_.find(header.track_id);
if (!registered_ || media_session_epoch_ == 0U ||
header.session_epoch != media_session_epoch_ ||
found == tracks_.end()) {
gateway_->stats_.datagrams_invalid.fetch_add(1U);
return;
}
track = found->second;
}
const auto expected_kind =
static_cast<cmvr::quic_edge::v1::MediaKind>(
static_cast<std::uint8_t>(header.kind));
if (track.kind != expected_kind ||
track.generation_token != header.codec_generation) {
gateway_->stats_.datagrams_invalid.fetch_add(1U);
return;
}
const auto result = reassembler_.accept(
header,
bytes.data() + cmvr::quic_edge::kDatagramHeaderBytes,
bytes.size() - cmvr::quic_edge::kDatagramHeaderBytes,
Clock::now());
gateway_->stats_.incomplete_frames_expired.fetch_add(
result.expired_frames);
gateway_->stats_.incomplete_frames_evicted.fetch_add(
result.evicted_frames);
switch (result.status) {
case ReassemblyResult::Status::COMPLETED: {
const std::uint64_t frame_count =
gateway_->stats_.frames_completed.fetch_add(1U) + 1U;
if (track.kind ==
cmvr::quic_edge::v1::MEDIA_KIND_VIDEO) {
gateway_->stats_.video_frames_completed.fetch_add(1U);
} else if (
track.kind ==
cmvr::quic_edge::v1::MEDIA_KIND_AUDIO) {
gateway_->stats_.audio_frames_completed.fetch_add(1U);
}
const std::size_t frame_size =
result.completed_frame
? result.completed_frame->payload.size() : 0U;
gateway_->stats_.frame_bytes_completed.fetch_add(frame_size);
if (result.completed_frame) {
gateway_->observeCompletedFrame(
track.kind,
result.completed_frame->header,
result.completed_frame->payload);
}
if (frame_count <= 5U || frame_count % 100U == 0U) {
gateway_->logEvent(
"media-frame-complete",
"\"track_id\":" +
std::to_string(header.track_id) +
",\"frame_sequence\":" +
std::to_string(header.frame_sequence) +
",\"bytes\":" + std::to_string(frame_size));
}
break;
}
case ReassemblyResult::Status::INVALID:
gateway_->stats_.datagrams_invalid.fetch_add(1U);
break;
case ReassemblyResult::Status::CAPACITY_DROPPED:
gateway_->stats_.incomplete_frames_evicted.fetch_add(1U);
break;
case ReassemblyResult::Status::ACCEPTED:
case ReassemblyResult::Status::DUPLICATE:
break;
}
}
void requestShutdown(const QUIC_UINT62 code)
{
std::lock_guard operation_lock(operation_mutex_);
if (connection_ && api_) {
api_->ConnectionShutdown(
connection_, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, code);
}
}
private:
~Connection()
{
gateway_->stats_.incomplete_frames_cleared.fetch_add(
reassembler_.clear());
gateway_->connectionContextDestroyed();
}
void acceptPeerStream(
HQUIC stream, const QUIC_STREAM_OPEN_FLAGS flags)
{
const bool is_unidirectional =
(flags & QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL) != 0;
bool accepted = false;
{
std::lock_guard operation_lock(operation_mutex_);
if (!is_unidirectional && !stream_ &&
!stream_was_opened_) {
stream_ = stream;
stream_was_opened_ = true;
stream_reference_active_ = true;
addRef();
api_->SetCallbackHandler(
stream,
reinterpret_cast<void*>(&Impl::streamCallback),
this);
accepted = true;
}
}
if (accepted) {
gateway_->logEvent(
"control-stream-open",
"\"peer\":\"" + jsonEscape(remote_address_text_) + "\"");
return;
}
api_->SetCallbackHandler(
stream,
reinterpret_cast<void*>(&Impl::rejectedStreamCallback),
gateway_);
api_->StreamShutdown(
stream,
static_cast<QUIC_STREAM_SHUTDOWN_FLAGS>(
QUIC_STREAM_SHUTDOWN_FLAG_ABORT |
QUIC_STREAM_SHUTDOWN_FLAG_IMMEDIATE),
kProtocolViolationCode);
}
QUIC_STATUS receiveControl(QUIC_STREAM_EVENT* event)
{
if (event->RECEIVE.BufferCount != 0U &&
!event->RECEIVE.Buffers) {
requestShutdown(kProtocolViolationCode);
return QUIC_STATUS_INVALID_PARAMETER;
}
std::size_t total_size = 0U;
for (std::uint32_t index = 0U;
index < event->RECEIVE.BufferCount; ++index) {
const QUIC_BUFFER& buffer = event->RECEIVE.Buffers[index];
if (buffer.Length != 0U && !buffer.Buffer) {
requestShutdown(kProtocolViolationCode);
return QUIC_STATUS_INVALID_PARAMETER;
}
if (buffer.Length >
std::numeric_limits<std::size_t>::max() - total_size) {
requestShutdown(kProtocolViolationCode);
return QUIC_STATUS_BUFFER_TOO_SMALL;
}
total_size += buffer.Length;
}
if (total_size == 0U) return QUIC_STATUS_SUCCESS;
// A single MsQuic RECEIVE callback may coalesce several valid
// length-prefixed frames. Bound the callback allocation by the
// work queue, while ControlFrameDecoder enforces the per-frame
// maximum after the bytes reach the worker.
if (total_size > gateway_->options_.maximum_work_queue_bytes) {
protocolViolation(
"control receive chunk exceeds work queue bound", 0U);
return QUIC_STATUS_BUFFER_TOO_SMALL;
}
std::vector<std::uint8_t> bytes;
try {
bytes.reserve(total_size);
for (std::uint32_t index = 0U;
index < event->RECEIVE.BufferCount; ++index) {
const QUIC_BUFFER& buffer = event->RECEIVE.Buffers[index];
bytes.insert(
bytes.end(), buffer.Buffer,
buffer.Buffer + buffer.Length);
}
} catch (...) {
requestShutdown(kProtocolViolationCode);
return QUIC_STATUS_OUT_OF_MEMORY;
}
addRef();
if (!gateway_->enqueueControl(this, std::move(bytes))) {
release();
requestShutdown(kProtocolViolationCode);
return QUIC_STATUS_BUFFER_TOO_SMALL;
}
return QUIC_STATUS_SUCCESS;
}
void receiveDatagram(const QUIC_BUFFER* buffer)
{
gateway_->stats_.datagrams_received.fetch_add(1U);
if (!buffer || !buffer->Buffer ||
buffer->Length < cmvr::quic_edge::kDatagramHeaderBytes) {
gateway_->stats_.datagrams_invalid.fetch_add(1U);
return;
}
std::vector<std::uint8_t> bytes;
try {
bytes.assign(
buffer->Buffer, buffer->Buffer + buffer->Length);
} catch (...) {
gateway_->stats_.datagrams_queue_dropped.fetch_add(1U);
return;
}
addRef();
if (!gateway_->enqueueDatagram(this, std::move(bytes))) {
release();
gateway_->stats_.datagrams_queue_dropped.fetch_add(1U);
}
}
bool processControlFrame(const std::vector<std::uint8_t>& frame)
{
Envelope envelope;
if (!envelope.ParseFromArray(
frame.data(), static_cast<int>(frame.size()))) {
protocolViolation("failed to parse EdgeControlEnvelope", 0U);
return false;
}
if (envelope.protocol_version() !=
cmvr::quic_edge::kProtocolVersion) {
protocolViolation(
"unsupported QUIC edge protocol version",
envelope.message_sequence());
return false;
}
if (has_inbound_sequence_ &&
envelope.message_sequence() <= inbound_sequence_) {
protocolViolation(
"edge control message sequence did not increase",
envelope.message_sequence());
return false;
}
has_inbound_sequence_ = true;
inbound_sequence_ = envelope.message_sequence();
if (!received_first_message_ &&
!envelope.has_node_register_request()) {
protocolViolation(
"NodeRegisterRequest must be the first control message",
envelope.message_sequence());
return false;
}
received_first_message_ = true;
if (envelope.has_node_register_request()) {
return handleRegistration(
envelope.node_register_request(),
envelope.message_sequence());
}
if (envelope.has_node_heartbeat()) {
return handleHeartbeat(
envelope.node_heartbeat(),
envelope.message_sequence());
}
if (envelope.has_media_session_open()) {
return handleMediaSessionOpen(
envelope.media_session_open(),
envelope.message_sequence());
}
if (envelope.has_media_track_descriptor()) {
return handleTrackDescriptor(
envelope.media_track_descriptor(),
envelope.message_sequence());
}
protocolViolation(
"edge sent an unexpected control message",
envelope.message_sequence());
return false;
}
bool handleRegistration(
const cmvr::quic_edge::v1::NodeRegisterRequest& request,
const std::uint64_t related_sequence)
{
gateway_->stats_.registrations_received.fetch_add(1U);
{
std::lock_guard state_lock(state_mutex_);
if (registration_attempted_ || registered_ ||
request.node().node_id().empty() ||
request.node().boot_id().empty()) {
protocolViolation(
"invalid or duplicate node registration",
related_sequence);
return false;
}
registration_attempted_ = true;
}
gateway_->recordRegistration(
request.node().node_id(),
request.node().boot_id(),
request.node().grpc_endpoint().host(),
request.node().grpc_endpoint().port(),
static_cast<std::uint64_t>(
request.node().local_interfaces_size()),
remoteIp());
gateway_->logEvent(
"node-register",
"\"node_id\":\"" +
jsonEscape(request.node().node_id()) +
"\",\"boot_id\":\"" +
jsonEscape(request.node().boot_id()) +
"\",\"interfaces\":" +
std::to_string(request.node().local_interfaces_size()) +
",\"grpc_host\":\"" +
jsonEscape(request.node().grpc_endpoint().host()) +
"\",\"grpc_port\":" +
std::to_string(request.node().grpc_endpoint().port()));
Envelope response;
response.set_protocol_version(cmvr::quic_edge::kProtocolVersion);
response.set_message_sequence(nextOutboundSequence());
auto* registration = response.mutable_node_register_response();
registration->set_heartbeat_interval_ms(
gateway_->options_.heartbeat_interval_ms);
registration->set_observed_source_ip(remoteIp());
if (gateway_->options_.scenario ==
Scenario::REJECT_REGISTRATION) {
registration->set_accepted(false);
registration->set_message("rejected by test scenario");
gateway_->stats_.registrations_rejected.fetch_add(1U);
return sendEnvelope(response);
}
const std::string session_id = gateway_->nextSessionId();
registration->set_accepted(true);
registration->set_session_id(session_id);
registration->set_message("accepted by cmvr QUIC test gateway");
{
std::lock_guard state_lock(state_mutex_);
registered_ = true;
node_id_ = request.node().node_id();
boot_id_ = request.node().boot_id();
session_id_ = session_id;
}
gateway_->stats_.registrations_accepted.fetch_add(1U);
gateway_->recordNode(
request.node().node_id(), request.node().boot_id(), session_id);
if (!sendEnvelope(response)) return false;
if (gateway_->options_.scenario ==
Scenario::FATAL_PROTOCOL_ERROR ||
gateway_->options_.scenario ==
Scenario::NONFATAL_PROTOCOL_ERROR) {
const bool fatal =
gateway_->options_.scenario ==
Scenario::FATAL_PROTOCOL_ERROR;
return sendProtocolError(
fatal ? "fatal test protocol error"
: "nonfatal test protocol error",
related_sequence, fatal);
}
return true;
}
bool handleHeartbeat(
const cmvr::quic_edge::v1::NodeHeartbeat& heartbeat,
const std::uint64_t related_sequence)
{
{
std::lock_guard state_lock(state_mutex_);
if (!registered_ ||
heartbeat.session_id() != session_id_ ||
heartbeat.node_id() != node_id_ ||
heartbeat.boot_id() != boot_id_ ||
(has_heartbeat_sequence_ &&
heartbeat.sequence() <= last_heartbeat_sequence_)) {
protocolViolation(
"heartbeat does not match the registered session",
related_sequence);
return false;
}
has_heartbeat_sequence_ = true;
last_heartbeat_sequence_ = heartbeat.sequence();
}
gateway_->stats_.heartbeats_received.fetch_add(1U);
gateway_->recordHeartbeat(heartbeat, remoteIp());
gateway_->logEvent(
"node-heartbeat",
"\"sequence\":" +
std::to_string(heartbeat.sequence()) +
",\"interfaces\":" +
std::to_string(heartbeat.local_interfaces_size()) +
",\"devices\":" +
std::to_string(
heartbeat.has_device_manager()
? heartbeat.device_manager().devices_size() : 0));
if (gateway_->options_.scenario ==
Scenario::DROP_HEARTBEAT_ACK) {
gateway_->stats_.heartbeat_acks_dropped.fetch_add(1U);
return true;
}
Envelope response;
response.set_protocol_version(cmvr::quic_edge::kProtocolVersion);
response.set_message_sequence(nextOutboundSequence());
auto* ack = response.mutable_node_heartbeat_ack();
ack->set_accepted(true);
ack->set_acknowledged_sequence(heartbeat.sequence());
ack->set_server_time_unix_ms(unixTimeMs());
ack->set_observed_source_ip(remoteIp());
ack->set_session_id(
gateway_->options_.scenario == Scenario::WRONG_ACK_SESSION
? "wrong-test-session" : currentSessionId());
if (!sendEnvelope(response)) return false;
gateway_->stats_.heartbeat_acks_sent.fetch_add(1U);
return true;
}
bool handleMediaSessionOpen(
const cmvr::quic_edge::v1::MediaSessionOpen& open,
const std::uint64_t related_sequence)
{
{
std::lock_guard state_lock(state_mutex_);
if (!registered_ || open.session_id() != session_id_ ||
open.node_id() != node_id_ ||
open.session_epoch() == 0U ||
media_session_epoch_ != 0U) {
protocolViolation(
"invalid media session open", related_sequence);
return false;
}
media_session_epoch_ = open.session_epoch();
tracks_.clear();
}
gateway_->stats_.incomplete_frames_cleared.fetch_add(
reassembler_.reset(open.session_epoch()));
gateway_->stats_.media_sessions_opened.fetch_add(1U);
gateway_->logEvent(
"media-session-open",
"\"session_epoch\":" +
std::to_string(open.session_epoch()));
return true;
}
bool handleTrackDescriptor(
const cmvr::quic_edge::v1::MediaTrackDescriptor& descriptor,
const std::uint64_t related_sequence)
{
if (descriptor.track_id() == 0U ||
descriptor.codec_generation() == 0U ||
descriptor.codec_generation_token() == 0U ||
descriptor.kind() ==
cmvr::quic_edge::v1::MEDIA_KIND_UNSPECIFIED ||
descriptor.codec().empty() ||
descriptor.payload_format().empty() ||
descriptor.source_track_id().empty()) {
protocolViolation(
"invalid media track descriptor", related_sequence);
return false;
}
{
std::lock_guard state_lock(state_mutex_);
if (!registered_ || media_session_epoch_ == 0U) {
protocolViolation(
"track descriptor arrived before media session",
related_sequence);
return false;
}
TrackInfo info;
info.kind = descriptor.kind();
info.generation_token =
descriptor.codec_generation_token();
info.generation = descriptor.codec_generation();
info.codec = descriptor.codec();
info.source_track_id = descriptor.source_track_id();
tracks_[descriptor.track_id()] = std::move(info);
}
gateway_->observeTrackDescriptor(descriptor);
gateway_->stats_.track_descriptors_received.fetch_add(1U);
gateway_->logEvent(
"media-track-descriptor",
"\"track_id\":" +
std::to_string(descriptor.track_id()) +
",\"kind\":" +
std::to_string(static_cast<int>(descriptor.kind())) +
",\"codec\":\"" + jsonEscape(descriptor.codec()) +
"\",\"source_track_id\":\"" +
jsonEscape(descriptor.source_track_id()) + "\"");
return true;
}
bool sendProtocolError(const std::string& message,
const std::uint64_t related_sequence,
const bool fatal)
{
Envelope response;
response.set_protocol_version(cmvr::quic_edge::kProtocolVersion);
response.set_message_sequence(nextOutboundSequence());
auto* protocol_error = response.mutable_protocol_error();
protocol_error->set_code(kProtocolErrorCode);
protocol_error->set_message(message);
protocol_error->set_related_message_sequence(related_sequence);
protocol_error->set_fatal(fatal);
if (!sendEnvelope(response)) return false;
gateway_->stats_.protocol_errors_sent.fetch_add(1U);
return true;
}
void protocolViolation(const std::string& message,
const std::uint64_t related_sequence)
{
gateway_->stats_.protocol_violations.fetch_add(1U);
gateway_->logEvent(
"protocol-violation",
"\"reason\":\"" + jsonEscape(message) +
"\",\"related_sequence\":" +
std::to_string(related_sequence));
requestShutdown(kProtocolViolationCode);
}
bool sendEnvelope(const Envelope& envelope)
{
std::string serialized;
if (!envelope.SerializeToString(&serialized)) {
gateway_->setRuntimeFailure(
"failed to serialize gateway control envelope");
return false;
}
std::vector<std::uint8_t> framed;
std::string error;
if (!cmvr::quic_edge::ControlFrameEncoder::encode(
reinterpret_cast<const std::uint8_t*>(serialized.data()),
serialized.size(),
gateway_->options_.maximum_control_frame_bytes,
&framed, &error)) {
gateway_->setRuntimeFailure(error);
return false;
}
if (framed.size() > std::numeric_limits<std::uint32_t>::max()) {
gateway_->setRuntimeFailure(
"framed control response is too large for MsQuic");
return false;
}
auto* send = new (std::nothrow) SendContext(std::move(framed));
if (!send) {
gateway_->setRuntimeFailure(
"failed to allocate control send context");
return false;
}
QUIC_STATUS status = QUIC_STATUS_INVALID_STATE;
{
std::lock_guard operation_lock(operation_mutex_);
if (stream_ && connected_.load()) {
status = api_->StreamSend(
stream_, &send->buffer, 1U,
QUIC_SEND_FLAG_NONE, send);
}
}
if (QUIC_FAILED(status)) {
delete send;
gateway_->logEvent(
"control-send-failed",
"\"status\":\"" +
jsonEscape(statusText("StreamSend", status)) + "\"");
requestShutdown(kProtocolViolationCode);
return false;
}
return true;
}
void completeStreamShutdown(HQUIC stream)
{
bool release_stream_reference = false;
{
std::lock_guard operation_lock(operation_mutex_);
if (stream_ == stream) {
stream_ = nullptr;
release_stream_reference = stream_reference_active_;
stream_reference_active_ = false;
}
api_->StreamClose(stream);
}
if (release_stream_reference) release();
}
void completeConnectionShutdown(HQUIC callback_handle)
{
HQUIC handle = nullptr;
{
std::lock_guard operation_lock(operation_mutex_);
if (connection_ == callback_handle) {
handle = connection_;
connection_ = nullptr;
}
connected_.store(false);
}
if (handle) api_->ConnectionClose(handle);
gateway_->connectionClosed(callback_handle);
release();
}
std::uint64_t nextOutboundSequence()
{
return outbound_sequence_++;
}
std::string currentSessionId() const
{
std::lock_guard state_lock(state_mutex_);
return session_id_;
}
std::string remoteIp() const
{
std::lock_guard state_lock(state_mutex_);
return remote_ip_;
}
static std::uint64_t unixTimeMs()
{
return static_cast<std::uint64_t>(
std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count());
}
Impl* gateway_;
const QUIC_API_TABLE* api_;
std::atomic<std::uint32_t> references_{1U};
std::atomic<bool> connected_{false};
mutable std::mutex operation_mutex_;
HQUIC connection_{nullptr};
HQUIC stream_{nullptr};
bool stream_reference_active_{false};
bool stream_was_opened_{false};
cmvr::quic_edge::ControlFrameDecoder decoder_;
bool received_first_message_{false};
bool has_inbound_sequence_{false};
std::uint64_t inbound_sequence_{0};
std::uint64_t outbound_sequence_{0};
mutable std::mutex state_mutex_;
bool registration_attempted_{false};
bool registered_{false};
bool has_heartbeat_sequence_{false};
std::uint64_t last_heartbeat_sequence_{0};
std::string node_id_;
std::string boot_id_;
std::string session_id_;
QUIC_ADDR remote_address_{};
std::string remote_address_text_;
std::string remote_ip_;
std::uint64_t media_session_epoch_{0};
std::unordered_map<std::uint32_t, TrackInfo> tracks_;
MediaReassembler reassembler_;
};
bool validateOptions(std::string* error) const
{
auto reject = [error](const std::string& message) {
if (error) *error = message;
return false;
};
if (options_.bind_address.empty()) {
return reject("bind address must not be empty");
}
if (options_.alpn.empty() ||
options_.alpn.size() > QUIC_MAX_ALPN_LENGTH) {
return reject("ALPN must contain 1 to 255 bytes");
}
if (options_.certificate_file.empty() ||
options_.private_key_file.empty()) {
return reject("server certificate and private key are required");
}
if (options_.maximum_control_frame_bytes == 0U ||
options_.maximum_control_frame_bytes >
std::numeric_limits<std::uint32_t>::max() - 4U) {
return reject("maximum control frame size is invalid");
}
if (options_.maximum_reassembly_bytes == 0U ||
options_.maximum_reassembly_frames == 0U ||
options_.maximum_frame_bytes == 0U ||
options_.maximum_work_queue_bytes == 0U ||
options_.reassembly_timeout_ms == 0U) {
return reject("gateway memory and timeout limits must be nonzero");
}
if (options_.maximum_work_queue_bytes <
options_.maximum_control_frame_bytes + 4U) {
return reject(
"work queue limit must fit one maximum control frame");
}
return true;
}
bool failStart(const std::string& message, std::string* error)
{
setRuntimeFailure(message);
if (error) *error = message;
return false;
}
void startWorker()
{
std::lock_guard lock(work_mutex_);
if (worker_.joinable()) return;
worker_stop_ = false;
worker_ = std::thread([this]() { workerLoop(); });
}
void stopWorker()
{
{
std::lock_guard lock(work_mutex_);
worker_stop_ = true;
}
work_condition_.notify_all();
if (worker_.joinable()) worker_.join();
}
void workerLoop()
{
while (true) {
WorkItem item;
bool control = false;
{
std::unique_lock lock(work_mutex_);
work_condition_.wait(lock, [this]() {
return worker_stop_ || !control_work_.empty() ||
!datagram_work_.empty();
});
if (control_work_.empty() && datagram_work_.empty() &&
worker_stop_) {
break;
}
if (!control_work_.empty()) {
item = std::move(control_work_.front());
control_work_.pop_front();
control_work_bytes_ -= item.bytes.size();
control = true;
} else {
item = std::move(datagram_work_.front());
datagram_work_.pop_front();
datagram_work_bytes_ -= item.bytes.size();
}
}
if (item.connection) {
try {
if (control) {
item.connection->processControlChunk(item.bytes);
} else {
item.connection->processDatagram(item.bytes);
}
} catch (const std::bad_alloc&) {
handleWorkerException(
item.connection,
"gateway worker exhausted memory");
} catch (...) {
handleWorkerException(
item.connection,
"gateway worker raised an unexpected exception");
}
item.connection->release();
}
}
}
void handleWorkerException(
Connection* connection, const char* message) noexcept
{
runtime_failed_.store(true);
try {
std::lock_guard lock(error_mutex_);
if (last_error_.empty()) last_error_ = message;
} catch (...) {
// The runtime_failed flag remains observable even when an
// allocation failure prevents retaining the diagnostic string.
}
if (connection) {
try {
connection->requestShutdown(kApplicationShutdownCode);
} catch (...) {
// Never allow a secondary lock failure to escape the worker.
}
}
}
bool enqueueControl(
Connection* connection, std::vector<std::uint8_t> bytes) noexcept
{
const std::size_t byte_count = bytes.size();
try {
std::lock_guard lock(work_mutex_);
if (worker_stop_ ||
byte_count > options_.maximum_work_queue_bytes ||
control_work_bytes_ >
options_.maximum_work_queue_bytes - byte_count) {
return false;
}
control_work_.push_back({connection, std::move(bytes)});
control_work_bytes_ += byte_count;
} catch (...) {
return false;
}
work_condition_.notify_one();
return true;
}
bool enqueueDatagram(
Connection* connection, std::vector<std::uint8_t> bytes) noexcept
{
const std::size_t byte_count = bytes.size();
try {
std::lock_guard lock(work_mutex_);
if (worker_stop_ ||
byte_count > options_.maximum_work_queue_bytes ||
datagram_work_bytes_ >
options_.maximum_work_queue_bytes - byte_count) {
return false;
}
datagram_work_.push_back({connection, std::move(bytes)});
datagram_work_bytes_ += byte_count;
} catch (...) {
return false;
}
work_condition_.notify_one();
return true;
}
void connectionContextCreated()
{
std::lock_guard lock(connection_mutex_);
++live_connection_contexts_;
}
void connectionContextDestroyed()
{
std::lock_guard lock(connection_mutex_);
if (live_connection_contexts_ != 0U) --live_connection_contexts_;
connection_condition_.notify_all();
}
bool trackConnection(HQUIC handle, Connection* connection) noexcept
{
try {
std::lock_guard lock(connection_mutex_);
const auto [unused, inserted] =
connections_.emplace(handle, connection);
static_cast<void>(unused);
if (!inserted) return false;
stats_.connections_accepted.fetch_add(1U);
return true;
} catch (...) {
return false;
}
}
void discardTrackedConnection(HQUIC handle)
{
std::lock_guard lock(connection_mutex_);
connections_.erase(handle);
connection_condition_.notify_all();
}
void connectionClosed(HQUIC handle)
{
{
std::lock_guard lock(connection_mutex_);
connections_.erase(handle);
}
stats_.connections_closed.fetch_add(1U);
connection_condition_.notify_all();
}
std::string nextSessionId()
{
const std::uint64_t number =
next_session_number_.fetch_add(1U);
return "cmvr-test-session-" + std::to_string(number);
}
void recordRegistration(
const std::string& node_id,
const std::string& boot_id,
const std::string& grpc_endpoint_host,
const std::uint32_t grpc_endpoint_port,
const std::uint64_t interface_count,
const std::string& observed_source_ip)
{
std::lock_guard lock(observation_mutex_);
last_node_id_ = node_id;
last_boot_id_ = boot_id;
last_grpc_endpoint_host_ = grpc_endpoint_host;
last_grpc_endpoint_port_ = grpc_endpoint_port;
last_observed_source_ip_ = observed_source_ip;
registration_interface_count_ = interface_count;
}
void recordNode(const std::string& node_id,
const std::string& boot_id,
const std::string& session_id)
{
std::lock_guard lock(observation_mutex_);
last_node_id_ = node_id;
last_boot_id_ = boot_id;
if (!session_id.empty()) last_session_id_ = session_id;
}
void recordHeartbeat(
const cmvr::quic_edge::v1::NodeHeartbeat& heartbeat,
const std::string& observed_source_ip)
{
std::uint64_t enabled_count = 0U;
std::uint64_t disabled_count = 0U;
std::uint64_t error_count = 0U;
std::uint64_t unknown_health_count = 0U;
if (heartbeat.has_device_manager()) {
for (const auto& device :
heartbeat.device_manager().devices()) {
if (device.enabled()) {
++enabled_count;
} else {
++disabled_count;
}
if (device.has_error()) ++error_count;
if (device.health() ==
cmvr::quic_edge::v1::
DEVICE_HEALTH_STATUS_UNSPECIFIED) {
++unknown_health_count;
}
}
}
std::lock_guard lock(observation_mutex_);
heartbeat_interface_count_ =
static_cast<std::uint64_t>(
heartbeat.local_interfaces_size());
last_observed_source_ip_ = observed_source_ip;
heartbeat_has_device_manager_ = heartbeat.has_device_manager();
if (!heartbeat.has_device_manager()) {
heartbeat_device_manager_name_.clear();
heartbeat_device_manager_version_.clear();
heartbeat_device_manager_description_.clear();
heartbeat_device_count_ = 0U;
heartbeat_enabled_device_count_ = 0U;
heartbeat_disabled_device_count_ = 0U;
heartbeat_error_device_count_ = 0U;
heartbeat_unknown_health_device_count_ = 0U;
heartbeat_devices_.clear();
return;
}
const auto& manager = heartbeat.device_manager();
heartbeat_device_manager_name_ = manager.manager_name();
heartbeat_device_manager_version_ = manager.manager_version();
heartbeat_device_manager_description_ =
manager.manager_description();
heartbeat_device_count_ =
static_cast<std::uint64_t>(manager.devices_size());
heartbeat_enabled_device_count_ = enabled_count;
heartbeat_disabled_device_count_ = disabled_count;
heartbeat_error_device_count_ = error_count;
heartbeat_unknown_health_device_count_ = unknown_health_count;
heartbeat_devices_.assign(
manager.devices().begin(), manager.devices().end());
}
void setRuntimeFailure(const std::string& message)
{
runtime_failed_.store(true);
std::lock_guard lock(error_mutex_);
if (last_error_.empty()) last_error_ = message;
}
void logEvent(const std::string& event, const std::string& fields)
{
std::ostringstream output;
output << "{\"event\":\"" << jsonEscape(event) << '"';
if (!fields.empty()) output << ',' << fields;
output << '}';
logJson(output.str());
}
void logJson(const std::string& json) const
{
std::lock_guard lock(log_mutex_);
std::cout << json << std::endl;
}
std::string readyJson() const
{
std::ostringstream output;
output << "{\"event\":\"ready\""
<< ",\"bind\":\"" << jsonEscape(bound_address_text_) << '"'
<< ",\"port\":" << bound_port_.load()
<< ",\"alpn\":\"" << jsonEscape(options_.alpn) << '"'
<< ",\"scenario\":\"" << scenarioName(options_.scenario) << '"'
<< ",\"datagram_enabled\":"
<< (options_.scenario == Scenario::DATAGRAM_DISABLED
? "false" : "true")
<< '}';
return output.str();
}
static QUIC_STATUS QUIC_API listenerCallback(
HQUIC, void* context, QUIC_LISTENER_EVENT* event) noexcept
{
try {
auto* gateway = static_cast<Impl*>(context);
if (!gateway || !event) return QUIC_STATUS_INVALID_PARAMETER;
if (event->Type == QUIC_LISTENER_EVENT_STOP_COMPLETE) {
{
std::lock_guard lock(gateway->listener_mutex_);
gateway->listener_stop_complete_ = true;
}
gateway->listener_condition_.notify_all();
return QUIC_STATUS_SUCCESS;
}
if (event->Type != QUIC_LISTENER_EVENT_NEW_CONNECTION ||
!event->NEW_CONNECTION.Connection) {
return QUIC_STATUS_SUCCESS;
}
Connection* connection = nullptr;
try {
connection = new (std::nothrow) Connection(
gateway, event->NEW_CONNECTION.Connection,
event->NEW_CONNECTION.Info
? event->NEW_CONNECTION.Info->RemoteAddress : nullptr);
} catch (...) {
return QUIC_STATUS_OUT_OF_MEMORY;
}
if (!connection) return QUIC_STATUS_OUT_OF_MEMORY;
if (!gateway->trackConnection(
event->NEW_CONNECTION.Connection, connection)) {
connection->abandonRejectedConnection();
connection->release();
return QUIC_STATUS_OUT_OF_MEMORY;
}
gateway->api_->SetCallbackHandler(
event->NEW_CONNECTION.Connection,
reinterpret_cast<void*>(&Impl::connectionCallback),
connection);
const QUIC_STATUS status =
gateway->api_->ConnectionSetConfiguration(
event->NEW_CONNECTION.Connection,
gateway->configuration_);
if (QUIC_FAILED(status)) {
gateway->discardTrackedConnection(
event->NEW_CONNECTION.Connection);
connection->abandonRejectedConnection();
connection->release();
return status;
}
return QUIC_STATUS_SUCCESS;
} catch (const std::bad_alloc&) {
return QUIC_STATUS_OUT_OF_MEMORY;
} catch (...) {
return QUIC_STATUS_INTERNAL_ERROR;
}
}
static QUIC_STATUS QUIC_API connectionCallback(
HQUIC connection, void* context,
QUIC_CONNECTION_EVENT* event) noexcept
{
try {
auto* state = static_cast<Connection*>(context);
if (!state || !event) return QUIC_STATUS_INVALID_PARAMETER;
return state->onConnectionEvent(connection, event);
} catch (const std::bad_alloc&) {
return QUIC_STATUS_OUT_OF_MEMORY;
} catch (...) {
return QUIC_STATUS_INTERNAL_ERROR;
}
}
static QUIC_STATUS QUIC_API streamCallback(
HQUIC stream, void* context, QUIC_STREAM_EVENT* event) noexcept
{
try {
auto* state = static_cast<Connection*>(context);
if (!state || !event) return QUIC_STATUS_INVALID_PARAMETER;
return state->onStreamEvent(stream, event);
} catch (const std::bad_alloc&) {
return QUIC_STATUS_OUT_OF_MEMORY;
} catch (...) {
return QUIC_STATUS_INTERNAL_ERROR;
}
}
static QUIC_STATUS QUIC_API rejectedStreamCallback(
HQUIC stream, void* context, QUIC_STREAM_EVENT* event) noexcept
{
try {
auto* gateway = static_cast<Impl*>(context);
if (!gateway || !event) return QUIC_STATUS_INVALID_PARAMETER;
if (event->Type == QUIC_STREAM_EVENT_SEND_COMPLETE) {
delete static_cast<SendContext*>(
event->SEND_COMPLETE.ClientContext);
} else if (event->Type == QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE) {
gateway->api_->StreamClose(stream);
}
return QUIC_STATUS_SUCCESS;
} catch (const std::bad_alloc&) {
return QUIC_STATUS_OUT_OF_MEMORY;
} catch (...) {
return QUIC_STATUS_INTERNAL_ERROR;
}
}
GatewayOptions options_;
const QUIC_API_TABLE* api_{nullptr};
HQUIC registration_{nullptr};
HQUIC configuration_{nullptr};
HQUIC listener_{nullptr};
bool listener_started_{false};
mutable std::mutex lifecycle_mutex_;
bool start_attempted_{false};
bool stopped_{false};
std::atomic<bool> started_{false};
Clock::time_point started_at_{};
std::atomic<std::uint16_t> bound_port_{0U};
std::string bound_address_text_;
mutable std::mutex listener_mutex_;
std::condition_variable listener_condition_;
bool listener_stop_complete_{false};
mutable std::mutex connection_mutex_;
std::condition_variable connection_condition_;
std::unordered_map<HQUIC, Connection*> connections_;
std::size_t live_connection_contexts_{0U};
std::atomic<std::uint64_t> next_session_number_{1U};
mutable std::mutex work_mutex_;
std::condition_variable work_condition_;
std::deque<WorkItem> control_work_;
std::deque<WorkItem> datagram_work_;
std::size_t control_work_bytes_{0U};
std::size_t datagram_work_bytes_{0U};
bool worker_stop_{false};
std::thread worker_;
AtomicStats stats_;
std::atomic<bool> runtime_failed_{false};
mutable std::mutex error_mutex_;
std::string last_error_;
mutable std::mutex log_mutex_;
mutable std::mutex observation_mutex_;
std::string last_node_id_;
std::string last_boot_id_;
std::string last_session_id_;
std::string last_grpc_endpoint_host_;
std::uint32_t last_grpc_endpoint_port_{0U};
std::string last_observed_source_ip_;
std::uint64_t registration_interface_count_{0U};
std::uint64_t heartbeat_interface_count_{0U};
bool heartbeat_has_device_manager_{false};
std::string heartbeat_device_manager_name_;
std::string heartbeat_device_manager_version_;
std::string heartbeat_device_manager_description_;
std::uint64_t heartbeat_device_count_{0U};
std::uint64_t heartbeat_enabled_device_count_{0U};
std::uint64_t heartbeat_disabled_device_count_{0U};
std::uint64_t heartbeat_error_device_count_{0U};
std::uint64_t heartbeat_unknown_health_device_count_{0U};
std::vector<cmvr::quic_edge::v1::ManagedDeviceStatus>
heartbeat_devices_;
TrackDescriptorObservation video_descriptor_;
TrackDescriptorObservation audio_descriptor_;
CompletedFrameObservation maximum_video_frame_;
CompletedFrameObservation maximum_audio_frame_;
};
QuicTestGateway::QuicTestGateway(GatewayOptions options)
: impl_(std::make_unique<Impl>(std::move(options)))
{
}
QuicTestGateway::~QuicTestGateway() = default;
bool QuicTestGateway::start(std::string* error)
{
return impl_->start(error);
}
void QuicTestGateway::stop()
{
impl_->stop();
}
std::uint16_t QuicTestGateway::boundPort() const
{
return impl_->boundPort();
}
bool QuicTestGateway::completionReached() const
{
return impl_->completionReached();
}
bool QuicTestGateway::hasRuntimeFailure() const
{
return impl_->hasRuntimeFailure();
}
std::string QuicTestGateway::lastError() const
{
return impl_->lastError();
}
std::string QuicTestGateway::summaryJson() const
{
return impl_->summaryJson();
}
} // namespace cmvr::test::quic_gateway
Reference in New Issue View Git Blame Copy Permalink