From b7317254bbbc3aa98f877678193ee2fcd6e0cb86 Mon Sep 17 00:00:00 2001 From: Daniel Prudky Date: Mon, 6 Jul 2026 15:57:07 +0200 Subject: [PATCH 1/4] feat: add QUIC operator transport support (BAF-1744) - Introduce OperatorChannel/QuicOperatorServer and wire them into the module context, letting forward_status()/wait_for_command() route through QUIC when quic_port is configured, falling back to the existing Fleet HTTP API otherwise - Add msquic as a new opt-in dependency, pulled in only when FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER is set - Parse new quic_* config keys (quic_port, quic_cert_path, quic_key_path, quic_ca_path, quic_alpn, quic_listen_address, quic_disable_client_auth) and validate required TLS material before starting the server - Add mutex locking around con->devices in device_connected()/device_disconnected(), now required because wait_for_command()'s QUIC branch reads it from a different thread --- CMakeLists.txt | 9 + cmake/Dependencies.cmake | 3 + .../external_server_api_structures.hpp | 9 + .../operator_stream/OperatorChannel.hpp | 60 +++ .../operator_stream/QuicOperatorServer.hpp | 132 ++++++ source/external_server_api.cpp | 156 ++++++ .../OperatorChannel.cpp | 43 ++ .../QuicOperatorServer.cpp | 443 ++++++++++++++++++ 8 files changed, 855 insertions(+) create mode 100644 include/bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp create mode 100644 include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp create mode 100644 source/transparent_operator_stream/OperatorChannel.cpp create mode 100644 source/transparent_operator_stream/QuicOperatorServer.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index f4ed7c5..cffaae9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -70,6 +70,12 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) FIND_PACKAGE(Boost CONFIG REQUIRED) FIND_PACKAGE(ZLIB REQUIRED) FIND_PACKAGE(cpprestsdk REQUIRED) + + # Operator-facing QUIC transport (BAF-1744), opt-in alongside the Fleet HTTP API above — + # ported from teleop-module's equivalent (BAF-1670). Wire format is plain JSON + # (nlohmann::json, already a dependency above), not protobuf — see QuicOperatorServer's + # class doc comment for why. + FIND_PACKAGE(msquic CONFIG REQUIRED) ENDIF () CMDEF_ADD_LIBRARY( @@ -117,6 +123,8 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) LIBRARY_GROUP transparent-external-server TYPE SHARED SOURCES "source/external_server_api.cpp" "source/memory_management.cpp" "source/device_management.cpp" + "source/transparent_operator_stream/OperatorChannel.cpp" + "source/transparent_operator_stream/QuicOperatorServer.cpp" VERSION ${TRANSPARENT_MODULE_VERSION} ) # Explicitly add Boost dependencies due to obsolete cpprestsdk cmake export package @@ -130,6 +138,7 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) PRIVATE transparent_module_sources fleet-http-client-shared::fleet-http-client-shared + msquic PUBLIC fleet-protocol-interface::module-maintainer-external-server-interface ) diff --git a/cmake/Dependencies.cmake b/cmake/Dependencies.cmake index c8487bf..298cea0 100644 --- a/cmake/Dependencies.cmake +++ b/cmake/Dependencies.cmake @@ -11,5 +11,8 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) BA_PACKAGE_LIBRARY(boost v1.86.0) BA_PACKAGE_LIBRARY(cpprestsdk v2.10.20) BA_PACKAGE_LIBRARY(zlib v1.3.2) + # Operator-facing QUIC transport (BAF-1744, ported from teleop-module's BAF-1670). Wire + # format is plain JSON (nlohmann-json above), not protobuf. + BA_PACKAGE_LIBRARY(msquic v2.5.6) ENDIF () diff --git a/include/bringauto/transparent_module_utils/external_server_api_structures.hpp b/include/bringauto/transparent_module_utils/external_server_api_structures.hpp index cca1250..7dac4f1 100644 --- a/include/bringauto/transparent_module_utils/external_server_api_structures.hpp +++ b/include/bringauto/transparent_module_utils/external_server_api_structures.hpp @@ -2,8 +2,11 @@ #include #include +#include +#include #include +#include #include #include #include @@ -19,6 +22,12 @@ namespace bringauto::transparent_module_utils std::mutex mutex; std::condition_variable con_variable; long last_command_timestamp; + + /// Operator-facing QUIC transport (BAF-1744) — used instead of the Fleet HTTP API above + /// when the config supplies quic_port; otherwise both stay unused/idle. Declared in this + /// order so operator_channel outlives quic_server (the server holds a reference to it). + operator_stream::OperatorChannel operator_channel; + std::unique_ptr quic_server; }; } // namespace bringauto::transparent_module_utils diff --git a/include/bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp b/include/bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp new file mode 100644 index 0000000..103fb44 --- /dev/null +++ b/include/bringauto/transparent_module_utils/operator_stream/OperatorChannel.hpp @@ -0,0 +1,60 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace bringauto::transparent_module_utils::operator_stream { + +/** + * One command received from an operator over QUIC, handed across the thread boundary to the + * synchronous ES API. `payload` is the opaque command buffer (today: the BAF-1651 JSON envelope) + * passed through verbatim; device_* are the fleet-protocol device coordinates kept for routing. + * + * Ported from teleop-module's OperatorChannel (BAF-1670) for the Transparent module's QUIC + * operator transport (BAF-1744) — see that repo's equivalent file for the original design notes. + */ +struct OperatorCommand { + std::vector payload; + std::int64_t timestamp_ms{0}; + std::uint32_t module_id{0}; + std::uint32_t device_type{0}; + std::string device_role; + std::string device_name; +}; + +/** + * Thread-safe seam between the QUIC server's ASYNCHRONOUS msquic receive callback (producer) and + * the module's SYNCHRONOUS, blocking external-server API (consumer) — see + * QuicOperatorServer for the producer side and external_server_api.cpp's wait_for_command() for + * the consumer side. + */ +class OperatorChannel { +public: + /// Producer (QUIC recv-callback thread): hand over a command and wake a waiter. Coalesces to + /// the newest — a pending unconsumed command is dropped (freshest wins). No-op after shutdown(). + void push(OperatorCommand command); + + /// Consumer (ES `wait_for_command` thread): block up to `timeout` for the latest command. + /// Returns std::nullopt on timeout or after shutdown(). + [[nodiscard]] std::optional waitForCommand(std::chrono::milliseconds timeout); + + /// Drop queued commands — e.g. on operator disconnect. + void clear(); + + /// Unblock all waiters and refuse further waits/pushes (module teardown). + void shutdown(); + +private: + std::mutex mutex_; + std::condition_variable cv_; + /// Single coalescing slot: push() always supersedes any unconsumed command, so at most one is held. + std::optional pending_; + bool shutdown_{false}; +}; + +} // namespace bringauto::transparent_module_utils::operator_stream diff --git a/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp b/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp new file mode 100644 index 0000000..9f38cef --- /dev/null +++ b/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp @@ -0,0 +1,132 @@ +#pragma once + +#include + +#include + +#include +#include +#include +#include + +namespace bringauto::transparent_module_utils::operator_stream { + +/** + * Configuration for the operator-facing QUIC server. mTLS optional (disable client auth + + * self-signed server cert for loopback bring-up). + */ +struct QuicOperatorServerConfig { + std::uint16_t port{0}; + std::string alpn{"transparent-operator"}; + std::string certPath; // server certificate (PEM) + std::string keyPath; // server private key (PEM) + std::string caCertsPath; // CA for verifying operator client certs (empty = no CA file) + bool disableClientAuth{false}; // true for loopback/dev; false requires operator client certs + std::string listenAddress; // empty = any +}; + +/** + * Operator-facing QUIC server living inside the Transparent module's external-server plugin — + * the operator leg, used alongside (not instead of) the existing Fleet HTTP API path (BAF-1744; + * ported from teleop-module's QuicOperatorServer, BAF-1670, single-operator Phase 1). + * + * Wire format is plain JSON (nlohmann::json), not protobuf: QUIC is just a transport, and this + * module's whole design already treats every payload as opaque JSON text (see its README), so a + * protobuf envelope around that JSON would be pure overhead — plus teleop-module's equivalent + * class, which this was ported from, needs its own .proto file distinct from ours anyway (same + * external-server-cpp process, process-global protobuf descriptor pool), a problem plain JSON + * doesn't have at all. Envelope shape: {"kind": "hello"|"status"|"command", "device": {"module_id", + * "type", "role", "name"}, "timestamp_ms": , "payload": }. + * + * cloud -> operator : sendStatus() opens a short unidirectional stream per envelope{kind:status} + * (START|FIN), freeing the send buffer on SEND_COMPLETE. + * operator -> cloud : each operator command arrives as a unidirectional stream; the receive + * callback accumulates the bytes, parses envelope{kind:command}, and hands + * it to the OperatorChannel (which unblocks the module's wait_for_command()). + * + * Threading: msquic invokes the callbacks on its own worker thread(s). Commands cross into the + * synchronous module API through the OperatorChannel (the thread-safe seam). sendStatus() is called + * on the ES thread (from forward_status) and msquic StreamSend is thread-safe. + */ +class QuicOperatorServer { +public: + QuicOperatorServer(QuicOperatorServerConfig config, OperatorChannel &channel); + ~QuicOperatorServer(); + + QuicOperatorServer(const QuicOperatorServer &) = delete; + QuicOperatorServer &operator=(const QuicOperatorServer &) = delete; + + /// Open msquic, registration, configuration, credential (mTLS), and the listener. False on any failure. + [[nodiscard]] bool initialize(); + + /// Start the listener. False on failure. + [[nodiscard]] bool start(); + + /// Stop the listener + drop the operator connection. + void stop(); + + /// Outcome of sendStatus(). NoOperator is an expected best-effort drop (nobody to send to), + /// distinct from SendFailed (serialize / StreamOpen / StreamSend error) so the caller can react. + enum class SendResult { Sent, NoOperator, SendFailed }; + + /// Send one status to the connected operator (best-effort: NoOperator if none). Called from the + /// ES thread. payload is the opaque status buffer (today: JSON); device_* are the fleet-protocol coords. + SendResult sendStatus(std::uint32_t module_id, std::uint32_t device_type, const std::string &device_role, + const std::string &device_name, const std::uint8_t *payload, std::size_t payload_size, + std::int64_t timestamp_ms); + + [[nodiscard]] bool hasOperator() const { return operatorConnection_.load() != nullptr; } + +private: + /// Heap buffer kept alive across an async StreamSend; freed in the send callback on SEND_COMPLETE. + struct SendBuffer { + explicit SendBuffer(std::string &&data) : payload(std::move(data)) { + buffer.Length = static_cast(payload.size()); + buffer.Buffer = reinterpret_cast(payload.data()); + } + std::string payload; + QUIC_BUFFER buffer{}; + }; + + /// Per-inbound-stream accumulation context (a command may arrive in multiple RECEIVE events). + struct InboundStreamContext { + QuicOperatorServer *server{nullptr}; + HQUIC connection{nullptr}; // owning connection; commands are accepted only while it is the operator + std::string bytes; + }; + + bool loadQuic(); + bool initRegistration(); + bool initConfiguration(); + bool initCredential(); + bool initListener(); + + /// Parse an accumulated command frame and hand it to the channel. + void handleCommandBytes(const std::string &bytes); + + static QUIC_STATUS QUIC_API listenerCallback(HQUIC listener, void *context, QUIC_LISTENER_EVENT *event); + static QUIC_STATUS QUIC_API connectionCallback(HQUIC connection, void *context, QUIC_CONNECTION_EVENT *event); + static QUIC_STATUS QUIC_API commandStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event); + static QUIC_STATUS QUIC_API sendStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event); + + QuicOperatorServerConfig config_; + OperatorChannel &channel_; + + const QUIC_API_TABLE *quic_{nullptr}; + HQUIC registration_{nullptr}; + HQUIC quicConfig_{nullptr}; + HQUIC listener_{nullptr}; + QUIC_ADDR quicAddr_{}; + QUIC_BUFFER alpnBuffer_{}; + + /// Single active operator connection (Phase 1: single operator). + std::atomic operatorConnection_{nullptr}; + std::atomic running_{false}; + + /// Serializes use of the connection handle (sendStatus' StreamOpen/StreamSend) against its + /// teardown (ConnectionClose in the SHUTDOWN_COMPLETE callback). The atomic above guards the + /// pointer value only, not the lifetime of the handle it points to. + std::mutex connectionMutex_; +}; + +} // namespace bringauto::transparent_module_utils::operator_stream diff --git a/source/external_server_api.cpp b/source/external_server_api.cpp index 10f0dc6..34dc4fe 100644 --- a/source/external_server_api.cpp +++ b/source/external_server_api.cpp @@ -5,13 +5,97 @@ #include #include +#include +#include #include +#include #include #include #include using namespace std::chrono_literals; +namespace +{ + namespace op = bringauto::transparent_module_utils::operator_stream; + + /// Parse the quic_* config keys into a QuicOperatorServerConfig. Returns std::nullopt if + /// quic_port is absent/zero (BAF-1744: QUIC is opt-in alongside the existing Fleet HTTP API — + /// see the header comment on context::quic_server). + std::optional parseQuicConfig( + const bringauto::fleet_protocol::cxx::KeyValueConfig &config) + { + op::QuicOperatorServerConfig quicConfig{}; + int port = 0; + + for (auto i = config.cbegin(); i != config.cend(); ++i) + { + if (i->first == "quic_port") + { + try + { + port = std::stoi(i->second); + } + catch (const std::exception &) + { + std::cerr << "[transparent-module] init: invalid quic_port" << std::endl; + return std::nullopt; + } + } + else if (i->first == "quic_cert_path") + { + quicConfig.certPath = i->second; + } + else if (i->first == "quic_key_path") + { + quicConfig.keyPath = i->second; + } + else if (i->first == "quic_ca_path") + { + quicConfig.caCertsPath = i->second; + } + else if (i->first == "quic_alpn") + { + quicConfig.alpn = i->second; + } + else if (i->first == "quic_listen_address") + { + quicConfig.listenAddress = i->second; + } + else if (i->first == "quic_disable_client_auth") + { + quicConfig.disableClientAuth = (i->second == "true" || i->second == "1"); + } + } + + if (port <= 0) + { + return std::nullopt; // QUIC not configured — caller falls back to Fleet HTTP API + } + if (port > 65535) + { + std::cerr << "[transparent-module] init: quic_port must be in range 1-65535" << std::endl; + return std::nullopt; + } + quicConfig.port = static_cast(port); + + if (quicConfig.certPath.empty() || quicConfig.keyPath.empty()) + { + std::cerr << "[transparent-module] init: quic_port set but missing quic_cert_path/quic_key_path" + << std::endl; + return std::nullopt; + } + if (!quicConfig.disableClientAuth && quicConfig.caCertsPath.empty()) + { + std::cerr << "[transparent-module] init: quic_ca_path is required when client authentication is " + "enabled (set quic_ca_path, or quic_disable_client_auth=true for dev/loopback)" + << std::endl; + return std::nullopt; + } + return quicConfig; + } +} // namespace + void *init(const config config_data) { auto *context = new struct bringauto::transparent_module_utils::context; @@ -145,6 +229,22 @@ void *init(const config config_data) fleet_api_config, request_frequency_guard_config); context->last_command_timestamp = 0; + + // BAF-1744: QUIC operator transport is opt-in (present only if the config supplies quic_port) — + // when absent, forward_status()/wait_for_command() fall back to the Fleet HTTP API above unchanged. + if (auto quicConfig = parseQuicConfig(config)) + { + context->quic_server = + std::make_unique(std::move(*quicConfig), context->operator_channel); + if (!context->quic_server->initialize() || !context->quic_server->start()) + { + std::cerr << "[transparent-module] init: QUIC operator server failed to initialize/start" << std::endl; + delete context; + return nullptr; + } + std::cerr << "[transparent-module] init: QUIC operator server enabled" << std::endl; + } + return context; } @@ -174,6 +274,20 @@ int forward_status(const buffer device_status, const device_identification devic bringauto::fleet_protocol::cxx::BufferAsString device_name(&device.device_name); bringauto::fleet_protocol::cxx::BufferAsString device_status_str(&device_status); + // BAF-1744: QUIC operator transport, when configured, replaces the Fleet HTTP API send below + // for this call — see context::quic_server's doc comment. + if (con->quic_server) + { + using SendResult = op::QuicOperatorServer::SendResult; + const auto sendResult = con->quic_server->sendStatus( + device.module, device.device_type, std::string(device_role.getStringView()), + std::string(device_name.getStringView()), static_cast(device_status.data), + device_status.size_in_bytes, 0); + // NoOperator is an expected best-effort drop (nobody connected) — report OK so the ES does + // not treat a missing operator as a failure. A genuine send error is surfaced as NOT_OK. + return sendResult == SendResult::SendFailed ? NOT_OK : OK; + } + con->fleet_api_client->setDeviceIdentification(bringauto::fleet_protocol::cxx::DeviceID( device.module, device.device_type, 0, // priority @@ -241,6 +355,10 @@ int device_disconnected(const int disconnect_type, const device_identification d const std::string_view device_device_name(static_cast(device.device_name.data), device.device_name.size_in_bytes); + // BAF-1744: con->mutex now also guards con->devices for wait_for_command()'s QUIC branch, + // which reads it from a different thread — this lock did not need to exist before that read + // was added. + std::lock_guard lock(con->mutex); for (auto it = con->devices.begin(); it != con->devices.end(); it++) { const std::string_view it_device_role(static_cast(it->device_role.data), it->device_role.size_in_bytes); @@ -289,6 +407,8 @@ int device_connected(const device_identification device, void *context) } std::memcpy(new_device.device_name.data, device.device_name.data, new_device.device_name.size_in_bytes); + // BAF-1744: see the matching lock in device_disconnected() for why this is needed now. + std::lock_guard lock(con->mutex); con->devices.emplace_back(new_device); return OK; } @@ -302,6 +422,42 @@ int wait_for_command(int timeout_time_in_ms, void *context) auto con = static_cast(context); + // BAF-1744: QUIC operator transport, when configured, replaces the Fleet HTTP polling below — + // see context::quic_server's doc comment. + if (con->quic_server) + { + constexpr int kDefaultWaitTimeoutMs = 1000; + const int effectiveTimeoutMs = timeout_time_in_ms > 0 ? timeout_time_in_ms : kDefaultWaitTimeoutMs; + auto command = con->operator_channel.waitForCommand(std::chrono::milliseconds(effectiveTimeoutMs)); + if (!command.has_value()) + { + return TIMEOUT_OCCURRED; + } + + std::lock_guard lock(con->mutex); + // Transparent module doesn't discriminate by device_type (see its README: "All device + // numbers are valid but do not affect the module's function") — route to the first + // connected device on the addressed module. + auto target = std::find_if(con->devices.begin(), con->devices.end(), [&](const device_identification &dev) { + return static_cast(dev.module) == command->module_id; + }); + if (target == con->devices.end()) + { + std::cerr << "[transparent-module] wait_for_command: operator command for module=" + << command->module_id << " but no matching device connected, dropping" << std::endl; + return TIMEOUT_OCCURRED; + } + + bringauto::fleet_protocol::cxx::BufferAsString targetRole(&target->device_role); + bringauto::fleet_protocol::cxx::BufferAsString targetName(&target->device_name); + std::string payload(command->payload.begin(), command->payload.end()); + con->command_vector.emplace_back( + std::move(payload), bringauto::fleet_protocol::cxx::DeviceID( + target->module, target->device_type, target->priority, + std::string(targetRole.getStringView()), std::string(targetName.getStringView()))); + return OK; + } + std::unique_lock lock(con->mutex); std::pair>, bringauto::fleet_protocol::http_client::FleetApiClient::ReturnCode> diff --git a/source/transparent_operator_stream/OperatorChannel.cpp b/source/transparent_operator_stream/OperatorChannel.cpp new file mode 100644 index 0000000..d4ea212 --- /dev/null +++ b/source/transparent_operator_stream/OperatorChannel.cpp @@ -0,0 +1,43 @@ +#include + +namespace bringauto::transparent_module_utils::operator_stream { + +void OperatorChannel::push(OperatorCommand command) { + { + std::lock_guard lock(mutex_); + if (shutdown_) { + return; + } + // Only the freshest command matters — coalesce by dropping any not-yet-consumed command so + // wait_for_command never forwards a stale command a newer one already superseded. + pending_ = std::move(command); + } + cv_.notify_one(); +} + +std::optional OperatorChannel::waitForCommand(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + const bool ready = cv_.wait_for(lock, timeout, [this] { return shutdown_ || pending_.has_value(); }); + if (!ready || shutdown_ || !pending_) { + return std::nullopt; + } + OperatorCommand command = std::move(*pending_); + pending_.reset(); + return command; +} + +void OperatorChannel::clear() { + std::lock_guard lock(mutex_); + pending_.reset(); +} + +void OperatorChannel::shutdown() { + { + std::lock_guard lock(mutex_); + shutdown_ = true; + pending_.reset(); + } + cv_.notify_all(); +} + +} // namespace bringauto::transparent_module_utils::operator_stream diff --git a/source/transparent_operator_stream/QuicOperatorServer.cpp b/source/transparent_operator_stream/QuicOperatorServer.cpp new file mode 100644 index 0000000..c9626f0 --- /dev/null +++ b/source/transparent_operator_stream/QuicOperatorServer.cpp @@ -0,0 +1,443 @@ +#include + +#include // QuicAddrSetFamily / QuicAddrSetPort helpers + +#include + +#include +#include +#include + +namespace bringauto::transparent_module_utils::operator_stream { + +namespace { +constexpr std::string_view APP_NAME{"transparent-operator-server"}; + +// Upper bound on a single inbound command frame — small JSON; this caps the per-stream +// accumulation buffer so a peer cannot exhaust memory by streaming bytes without ever sending FIN +// (one such buffer exists per concurrent unidirectional stream). +constexpr std::size_t MAX_COMMAND_FRAME_BYTES{64U * 1024U}; + +// This module has no logging infrastructure of its own (unlike teleop-module's EsLogger, which +// this class is ported from) — plain stderr lines are enough for this bring-up (BAF-1744). +void logInfo(const std::string &msg) { std::cerr << "[transparent-quic] INFO: " << msg << std::endl; } +void logWarning(const std::string &msg) { std::cerr << "[transparent-quic] WARN: " << msg << std::endl; } +void logError(const std::string &msg) { std::cerr << "[transparent-quic] ERROR: " << msg << std::endl; } + +std::string toHex(unsigned value) { + std::ostringstream oss; + oss << std::hex << value; + return oss.str(); +} + +// Append every QUIC_BUFFER segment of a RECEIVE event onto the per-stream accumulator. +void appendBuffers(std::string &out, const QUIC_BUFFER *buffers, std::uint32_t bufferCount, std::uint64_t totalLength) { + out.reserve(out.size() + totalLength); + for (std::uint32_t i = 0; i < bufferCount; ++i) { + out.append(reinterpret_cast(buffers[i].Buffer), buffers[i].Length); + } +} +} // namespace + +QuicOperatorServer::QuicOperatorServer(QuicOperatorServerConfig config, OperatorChannel &channel) + : config_(std::move(config)), channel_(channel) { + alpnBuffer_ = {static_cast(config_.alpn.size()), + reinterpret_cast(config_.alpn.data())}; +} + +QuicOperatorServer::~QuicOperatorServer() { + stop(); + if (quicConfig_ != nullptr) { + quic_->ConfigurationClose(quicConfig_); + } + if (registration_ != nullptr) { + quic_->RegistrationClose(registration_); + } + if (quic_ != nullptr) { + MsQuicClose(quic_); + } +} + +bool QuicOperatorServer::initialize() { + return loadQuic() && initRegistration() && initConfiguration() && initCredential() && initListener(); +} + +bool QuicOperatorServer::loadQuic() { + const QUIC_STATUS status = MsQuicOpen2(&quic_); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: MsQuicOpen2 failed, status=" + std::to_string(static_cast(status))); + return false; + } + return true; +} + +bool QuicOperatorServer::initRegistration() { + QUIC_REGISTRATION_CONFIG config{}; + config.AppName = APP_NAME.data(); + config.ExecutionProfile = QUIC_EXECUTION_PROFILE_LOW_LATENCY; + const QUIC_STATUS status = quic_->RegistrationOpen(&config, ®istration_); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: RegistrationOpen failed, status=" + + std::to_string(static_cast(status))); + return false; + } + return true; +} + +bool QuicOperatorServer::initConfiguration() { + QUIC_SETTINGS settings{}; + settings.IsSet.IdleTimeoutMs = TRUE; + settings.IdleTimeoutMs = 30000; + settings.IsSet.KeepAliveIntervalMs = TRUE; + settings.KeepAliveIntervalMs = 5000; + // The operator opens unidirectional streams to send commands (one per command); allow plenty. + settings.IsSet.PeerUnidiStreamCount = TRUE; + settings.PeerUnidiStreamCount = 1024; + settings.IsSet.SendBufferingEnabled = TRUE; + settings.SendBufferingEnabled = TRUE; + settings.IsSet.ServerResumptionLevel = TRUE; + settings.ServerResumptionLevel = QUIC_SERVER_RESUME_AND_ZERORTT; + + const QUIC_STATUS status = + quic_->ConfigurationOpen(registration_, &alpnBuffer_, 1, &settings, sizeof(settings), nullptr, &quicConfig_); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: ConfigurationOpen failed, status=" + + std::to_string(static_cast(status))); + return false; + } + return true; +} + +bool QuicOperatorServer::initCredential() { + QUIC_CERTIFICATE_FILE certificateFile{}; + certificateFile.CertificateFile = config_.certPath.c_str(); + certificateFile.PrivateKeyFile = config_.keyPath.c_str(); + + QUIC_CREDENTIAL_CONFIG credential{}; + credential.Type = QUIC_CREDENTIAL_TYPE_CERTIFICATE_FILE; + credential.CertificateFile = &certificateFile; + if (!config_.caCertsPath.empty()) { + credential.Flags |= QUIC_CREDENTIAL_FLAG_SET_CA_CERTIFICATE_FILE; + credential.CaCertificateFile = config_.caCertsPath.c_str(); + } + if (config_.disableClientAuth) { + logWarning("QUIC operator server: client authentication DISABLED (dev/loopback only)"); + } else { + credential.Flags |= QUIC_CREDENTIAL_FLAG_REQUIRE_CLIENT_AUTHENTICATION; + } + + const QUIC_STATUS status = quic_->ConfigurationLoadCredential(quicConfig_, &credential); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: ConfigurationLoadCredential failed, status=" + + std::to_string(static_cast(status))); + return false; + } + return true; +} + +bool QuicOperatorServer::initListener() { + QUIC_STATUS status = quic_->ListenerOpen(registration_, listenerCallback, this, &listener_); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: ListenerOpen failed, status=" + + std::to_string(static_cast(status))); + return false; + } + if (!config_.listenAddress.empty()) { + // Bind the configured address (IPv4 or IPv6); QuicAddrFromString also sets the port. + if (!QuicAddrFromString(config_.listenAddress.c_str(), config_.port, &quicAddr_)) { + logError("QUIC operator server: invalid quic_listen_address '" + config_.listenAddress + "'"); + quic_->ListenerClose(listener_); + listener_ = nullptr; + return false; + } + } else { + // No address configured: bind any interface, dual-stack (IPv4 + IPv6). + QuicAddrSetFamily(&quicAddr_, QUIC_ADDRESS_FAMILY_UNSPEC); + QuicAddrSetPort(&quicAddr_, config_.port); + } + return true; +} + +bool QuicOperatorServer::start() { + const QUIC_STATUS status = quic_->ListenerStart(listener_, &alpnBuffer_, 1, &quicAddr_); + if (QUIC_FAILED(status)) { + logError("QUIC operator server: ListenerStart failed, status=" + + std::to_string(static_cast(status))); + // The listener was opened but never started, so running_ stays false and stop() will not tear it + // down — close it here. Otherwise the handle leaks and RegistrationClose in the destructor can + // block on the still-open listener (e.g. when the port is already in use). + quic_->ListenerClose(listener_); + listener_ = nullptr; + return false; + } + running_.store(true); + logInfo("QUIC operator server: listening on port " + std::to_string(config_.port) + " (alpn=" + config_.alpn + ")"); + return true; +} + +void QuicOperatorServer::stop() { + if (!running_.exchange(false)) { + return; + } + if (listener_ != nullptr) { + quic_->ListenerStop(listener_); + } + HQUIC conn = operatorConnection_.exchange(nullptr); + if (conn != nullptr) { + quic_->ConnectionShutdown(conn, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0); + } + channel_.shutdown(); +} + +QuicOperatorServer::SendResult QuicOperatorServer::sendStatus(std::uint32_t module_id, std::uint32_t device_type, + const std::string &device_role, const std::string &device_name, + const std::uint8_t *payload, std::size_t payload_size, + std::int64_t timestamp_ms) { + if (operatorConnection_.load() == nullptr) { + return SendResult::NoOperator; // no operator connected — drop (best-effort) + } + + // payload is opaque bytes that happen to already be JSON text almost always (see the class doc + // comment) — embed it as a nested JSON value so the wire form has no extra string-escaping + // layer, falling back to a plain string if it genuinely isn't valid JSON. + nlohmann::json envelope; + envelope["kind"] = "status"; + envelope["device"] = {{"module_id", module_id}, {"type", device_type}, {"role", device_role}, {"name", device_name}}; + envelope["timestamp_ms"] = timestamp_ms; + try { + envelope["payload"] = nlohmann::json::parse(payload, payload + payload_size); + } catch (const nlohmann::json::exception &) { + envelope["payload"] = std::string(reinterpret_cast(payload), payload_size); + } + + std::string serialized; + try { + // replace (not the default strict): the non-JSON fallback above can embed arbitrary bytes as + // a JSON string, and strict dump() throws on invalid UTF-8 in that string, dropping the whole + // status. replace swaps invalid sequences for U+FFFD instead of failing the send. + serialized = envelope.dump(-1, ' ', false, nlohmann::json::error_handler_t::replace); + } catch (const nlohmann::json::exception &e) { + logError(std::string("QUIC operator server: failed to serialize status, dropping: ") + e.what()); + return SendResult::SendFailed; + } + + // Hold connectionMutex_ across the whole handle use so the SHUTDOWN_COMPLETE callback cannot + // ConnectionClose the handle while we are opening/sending on it. Re-load under the lock: the + // connection may have been torn down between the early-out check above and here. + std::lock_guard lock(connectionMutex_); + HQUIC connection = operatorConnection_.load(); + if (connection == nullptr) { + return SendResult::NoOperator; // operator disconnected while we were serializing — drop + } + + HQUIC stream{nullptr}; + QUIC_STATUS status_rc = + quic_->StreamOpen(connection, QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL, sendStreamCallback, this, &stream); + if (QUIC_FAILED(status_rc)) { + logError("QUIC operator server: StreamOpen (status) failed, status=" + + std::to_string(static_cast(status_rc))); + return SendResult::SendFailed; + } + auto sendBuffer = std::make_unique(std::move(serialized)); + status_rc = quic_->StreamSend(stream, &sendBuffer->buffer, 1, QUIC_SEND_FLAG_START | QUIC_SEND_FLAG_FIN, + sendBuffer.get()); + if (QUIC_FAILED(status_rc)) { + logError("QUIC operator server: StreamSend (status) failed, status=" + + std::to_string(static_cast(status_rc))); + // sendBuffer is freed automatically on return. The stream was opened but never started (START + // rode on the failed send), so no sendStreamCallback will fire to close it — close it here to + // avoid leaking the handle. + quic_->StreamClose(stream); + return SendResult::SendFailed; + } + // Ownership transfers to msquic; sendStreamCallback reclaims it on SEND_COMPLETE. + sendBuffer.release(); + return SendResult::Sent; +} + +void QuicOperatorServer::handleCommandBytes(const std::string &bytes) { + nlohmann::json envelope; + try { + envelope = nlohmann::json::parse(bytes); + } catch (const nlohmann::json::exception &e) { + logWarning("QUIC operator server: failed to parse command JSON (" + std::to_string(bytes.size()) + + " bytes): " + e.what()); + return; + } + if (!envelope.is_object() || envelope.value("kind", std::string{}) != "command") { + // Malformed, or "hello" — ignored for now (single operator, always active). + return; + } + + const auto deviceIt = envelope.find("device"); + const nlohmann::json &device = deviceIt != envelope.end() ? *deviceIt : nlohmann::json::object(); + + OperatorCommand out; + out.timestamp_ms = envelope.value("timestamp_ms", static_cast(0)); + out.module_id = device.value("module_id", 0u); + out.device_type = device.value("type", 0u); + out.device_role = device.value("role", std::string{}); + out.device_name = device.value("name", std::string{}); + // The BAF-1651 envelope was embedded as a nested JSON value by the sender — re-serialize it + // back to bytes here, since OperatorCommand::payload/downstream consumers (pop_command(), + // rtsp-server's onFleetCommand()) all expect a flat JSON string, not a parsed value. + const std::string payloadStr = envelope.contains("payload") ? envelope["payload"].dump() : std::string("{}"); + out.payload.assign(payloadStr.begin(), payloadStr.end()); + channel_.push(std::move(out)); +} + +QUIC_STATUS QUIC_API QuicOperatorServer::listenerCallback(HQUIC listener, void *context, QUIC_LISTENER_EVENT *event) { + auto *self = static_cast(context); + switch (event->Type) { + case QUIC_LISTENER_EVENT_NEW_CONNECTION: + self->quic_->SetCallbackHandler(event->NEW_CONNECTION.Connection, + reinterpret_cast(connectionCallback), self); + return self->quic_->ConnectionSetConfiguration(event->NEW_CONNECTION.Connection, self->quicConfig_); + case QUIC_LISTENER_EVENT_STOP_COMPLETE: + if (!event->STOP_COMPLETE.AppCloseInProgress) { + self->quic_->ListenerClose(listener); + } + self->listener_ = nullptr; + return QUIC_STATUS_SUCCESS; + default: + return QUIC_STATUS_NOT_SUPPORTED; + } +} + +QUIC_STATUS QUIC_API QuicOperatorServer::connectionCallback(HQUIC connection, void *context, + QUIC_CONNECTION_EVENT *event) { + auto *self = static_cast(context); + switch (event->Type) { + case QUIC_CONNECTION_EVENT_CONNECTED: { + HQUIC expected = nullptr; + if (!self->operatorConnection_.compare_exchange_strong(expected, connection)) { + logWarning("QUIC operator server: an operator is already connected, rejecting new one " + "(single-operator)"); + self->quic_->ConnectionShutdown(connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0); + return QUIC_STATUS_SUCCESS; + } + logInfo("QUIC operator server: operator connected"); + return QUIC_STATUS_SUCCESS; + } + case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT: { + logInfo("QUIC operator server: SHUTDOWN_INITIATED_BY_TRANSPORT, status=0x" + + toHex(static_cast(event->SHUTDOWN_INITIATED_BY_TRANSPORT.Status))); + HQUIC expected = connection; + if (self->operatorConnection_.compare_exchange_strong(expected, nullptr)) { + self->channel_.clear(); + logInfo("QUIC operator server: operator disconnected"); + } else { + logWarning("QUIC operator server: SHUTDOWN_INITIATED_BY_TRANSPORT for a connection that was not " + "the tracked operator (already replaced/cleared?)"); + } + return QUIC_STATUS_SUCCESS; + } + case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_PEER: { + logInfo("QUIC operator server: SHUTDOWN_INITIATED_BY_PEER, error_code=" + + std::to_string(event->SHUTDOWN_INITIATED_BY_PEER.ErrorCode)); + HQUIC expected = connection; + if (self->operatorConnection_.compare_exchange_strong(expected, nullptr)) { + self->channel_.clear(); + logInfo("QUIC operator server: operator disconnected"); + } else { + logWarning("QUIC operator server: SHUTDOWN_INITIATED_BY_PEER for a connection that was not " + "the tracked operator (already replaced/cleared?)"); + } + return QUIC_STATUS_SUCCESS; + } + case QUIC_CONNECTION_EVENT_PEER_STREAM_STARTED: { + // Only the accepted operator may feed the command channel. A stream opened by any other + // connection (a second operator being rejected, or one mid-shutdown) is rejected so its + // bytes never reach the device. Phase 1 is single-operator. + // msquic requires every stream surfaced in PEER_STREAM_STARTED to be either accepted + // (SetCallbackHandler) or closed (StreamClose); without a registered handler there is no + // SHUTDOWN_COMPLETE to reclaim on, so StreamShutdown alone would leak the handle. StreamClose + // rejects it cleanly. + if (connection != self->operatorConnection_.load()) { + self->quic_->StreamClose(event->PEER_STREAM_STARTED.Stream); + return QUIC_STATUS_SUCCESS; + } + auto streamCtx = std::make_unique(InboundStreamContext{self, connection, {}}); + self->quic_->SetCallbackHandler(event->PEER_STREAM_STARTED.Stream, + reinterpret_cast(commandStreamCallback), streamCtx.get()); + // Ownership transfers to commandStreamCallback; reclaimed on SHUTDOWN_COMPLETE. + streamCtx.release(); + return QUIC_STATUS_SUCCESS; + } + case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE: + logInfo("QUIC operator server: SHUTDOWN_COMPLETE, AppCloseInProgress=" + + std::to_string(static_cast(event->SHUTDOWN_COMPLETE.AppCloseInProgress))); + if (!event->SHUTDOWN_COMPLETE.AppCloseInProgress) { + // Serialize against sendStatus: it may still be inside StreamOpen/StreamSend on this + // handle. Closing it concurrently would be a use-after-free. + std::lock_guard lock(self->connectionMutex_); + self->quic_->ConnectionClose(connection); + } + return QUIC_STATUS_SUCCESS; + default: + return QUIC_STATUS_SUCCESS; + } +} + +QUIC_STATUS QUIC_API QuicOperatorServer::commandStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event) { + auto *ctx = static_cast(context); + switch (event->Type) { + case QUIC_STREAM_EVENT_RECEIVE: { + if (event->RECEIVE.BufferCount > 0) { + appendBuffers(ctx->bytes, event->RECEIVE.Buffers, event->RECEIVE.BufferCount, + event->RECEIVE.TotalBufferLength); + } + if (ctx->bytes.size() > MAX_COMMAND_FRAME_BYTES) { + logWarning("QUIC operator server: inbound command frame exceeded " + + std::to_string(MAX_COMMAND_FRAME_BYTES) + " bytes, aborting stream"); + ctx->bytes.clear(); + ctx->server->quic_->StreamShutdown(stream, QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 0); + return QUIC_STATUS_SUCCESS; + } + if (event->RECEIVE.Flags & QUIC_RECEIVE_FLAG_FIN) { + if (ctx->connection == ctx->server->operatorConnection_.load()) { + ctx->server->handleCommandBytes(ctx->bytes); + } + ctx->bytes.clear(); + } + return QUIC_STATUS_SUCCESS; + } + case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN: + if (!ctx->bytes.empty()) { + if (ctx->connection == ctx->server->operatorConnection_.load()) { + ctx->server->handleCommandBytes(ctx->bytes); + } + ctx->bytes.clear(); + } + return QUIC_STATUS_SUCCESS; + case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: { + // Reclaim the context whose ownership PEER_STREAM_STARTED handed to msquic; freed at scope end. + std::unique_ptr owned{ctx}; + if (!event->SHUTDOWN_COMPLETE.AppCloseInProgress) { + ctx->server->quic_->StreamClose(stream); + } + return QUIC_STATUS_SUCCESS; + } + default: + return QUIC_STATUS_SUCCESS; + } +} + +QUIC_STATUS QUIC_API QuicOperatorServer::sendStreamCallback(HQUIC stream, void *context, QUIC_STREAM_EVENT *event) { + auto *self = static_cast(context); + switch (event->Type) { + case QUIC_STREAM_EVENT_SEND_COMPLETE: + // Reclaim the buffer whose ownership sendStatus handed to msquic; freed at end of statement. + std::unique_ptr{static_cast(event->SEND_COMPLETE.ClientContext)}; + return QUIC_STATUS_SUCCESS; + case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: + if (!event->SHUTDOWN_COMPLETE.AppCloseInProgress) { + self->quic_->StreamClose(stream); + } + return QUIC_STATUS_SUCCESS; + default: + return QUIC_STATUS_SUCCESS; + } +} + +} // namespace bringauto::transparent_module_utils::operator_stream From dff087c4b5077a62fc3a1737bc7042d580a845fd Mon Sep 17 00:00:00 2001 From: Daniel Prudky Date: Tue, 7 Jul 2026 12:00:40 +0200 Subject: [PATCH 2/4] fix(BAF-1744): harden QUIC operator status/command handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pass a real epoch timestamp to sendStatus() instead of hardcoded 0, so operator-side status messages carry accurate timing - only take con->mutex in device_connected()/device_disconnected() when con->quic_server is configured, since the lock is only needed for wait_for_command()'s QUIC branch and unconditional locking would add the Fleet-HTTP blocking-call stall to every (dis)connect event in non-QUIC deployments - wrap handleCommandBytes() envelope parsing in try/catch for nlohmann::json::exception, since it runs inside a QUIC_API callback invoked by msquic's C library — an uncaught type_error from a malformed field would otherwise terminate the whole external-server process --- source/external_server_api.cpp | 27 ++++++++--- .../QuicOperatorServer.cpp | 47 +++++++++++-------- 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/source/external_server_api.cpp b/source/external_server_api.cpp index 34dc4fe..54293cb 100644 --- a/source/external_server_api.cpp +++ b/source/external_server_api.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -279,10 +280,13 @@ int forward_status(const buffer device_status, const device_identification devic if (con->quic_server) { using SendResult = op::QuicOperatorServer::SendResult; + const auto nowMs = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); const auto sendResult = con->quic_server->sendStatus( device.module, device.device_type, std::string(device_role.getStringView()), std::string(device_name.getStringView()), static_cast(device_status.data), - device_status.size_in_bytes, 0); + device_status.size_in_bytes, static_cast(nowMs)); // NoOperator is an expected best-effort drop (nobody connected) — report OK so the ES does // not treat a missing operator as a failure. A genuine send error is surfaced as NOT_OK. return sendResult == SendResult::SendFailed ? NOT_OK : OK; @@ -356,9 +360,16 @@ int device_disconnected(const int disconnect_type, const device_identification d device.device_name.size_in_bytes); // BAF-1744: con->mutex now also guards con->devices for wait_for_command()'s QUIC branch, - // which reads it from a different thread — this lock did not need to exist before that read - // was added. - std::lock_guard lock(con->mutex); + // which reads it from a different thread — but only take it when QUIC is actually configured + // for this context. wait_for_command()'s non-QUIC/Fleet-HTTP branch holds this same mutex + // across a blocking con->fleet_api_client->getCommands() call; locking unconditionally here + // would add that stall to every (dis)connect event even in deployments that never use QUIC, + // where con->devices has no concurrent reader at all. + std::unique_lock lock(con->mutex, std::defer_lock); + if (con->quic_server) + { + lock.lock(); + } for (auto it = con->devices.begin(); it != con->devices.end(); it++) { const std::string_view it_device_role(static_cast(it->device_role.data), it->device_role.size_in_bytes); @@ -407,8 +418,12 @@ int device_connected(const device_identification device, void *context) } std::memcpy(new_device.device_name.data, device.device_name.data, new_device.device_name.size_in_bytes); - // BAF-1744: see the matching lock in device_disconnected() for why this is needed now. - std::lock_guard lock(con->mutex); + // BAF-1744: see the matching lock in device_disconnected() for why this is conditional. + std::unique_lock lock(con->mutex, std::defer_lock); + if (con->quic_server) + { + lock.lock(); + } con->devices.emplace_back(new_device); return OK; } diff --git a/source/transparent_operator_stream/QuicOperatorServer.cpp b/source/transparent_operator_stream/QuicOperatorServer.cpp index c9626f0..3a4bb07 100644 --- a/source/transparent_operator_stream/QuicOperatorServer.cpp +++ b/source/transparent_operator_stream/QuicOperatorServer.cpp @@ -264,26 +264,35 @@ void QuicOperatorServer::handleCommandBytes(const std::string &bytes) { " bytes): " + e.what()); return; } - if (!envelope.is_object() || envelope.value("kind", std::string{}) != "command") { - // Malformed, or "hello" — ignored for now (single operator, always active). - return; - } + // Every field access below can throw nlohmann::json::type_error if a field is present but has + // the wrong JSON type (e.g. "device" is a string, not an object) — this function is invoked + // from commandStreamCallback, a QUIC_API callback handed directly to msquic's C library, which + // is not exception-safe: an uncaught exception here would unwind into msquic and terminate the + // whole external-server process for every device on the module, not just this connection. + try { + if (!envelope.is_object() || envelope.value("kind", std::string{}) != "command") { + // Malformed, or "hello" — ignored for now (single operator, always active). + return; + } - const auto deviceIt = envelope.find("device"); - const nlohmann::json &device = deviceIt != envelope.end() ? *deviceIt : nlohmann::json::object(); - - OperatorCommand out; - out.timestamp_ms = envelope.value("timestamp_ms", static_cast(0)); - out.module_id = device.value("module_id", 0u); - out.device_type = device.value("type", 0u); - out.device_role = device.value("role", std::string{}); - out.device_name = device.value("name", std::string{}); - // The BAF-1651 envelope was embedded as a nested JSON value by the sender — re-serialize it - // back to bytes here, since OperatorCommand::payload/downstream consumers (pop_command(), - // rtsp-server's onFleetCommand()) all expect a flat JSON string, not a parsed value. - const std::string payloadStr = envelope.contains("payload") ? envelope["payload"].dump() : std::string("{}"); - out.payload.assign(payloadStr.begin(), payloadStr.end()); - channel_.push(std::move(out)); + const auto deviceIt = envelope.find("device"); + const nlohmann::json &device = deviceIt != envelope.end() ? *deviceIt : nlohmann::json::object(); + + OperatorCommand out; + out.timestamp_ms = envelope.value("timestamp_ms", static_cast(0)); + out.module_id = device.value("module_id", 0u); + out.device_type = device.value("type", 0u); + out.device_role = device.value("role", std::string{}); + out.device_name = device.value("name", std::string{}); + // The BAF-1651 envelope was embedded as a nested JSON value by the sender — re-serialize it + // back to bytes here, since OperatorCommand::payload and its downstream consumers (e.g. + // pop_command()) expect a flat JSON string, not a parsed value. + const std::string payloadStr = envelope.contains("payload") ? envelope["payload"].dump() : std::string("{}"); + out.payload.assign(payloadStr.begin(), payloadStr.end()); + channel_.push(std::move(out)); + } catch (const nlohmann::json::exception &e) { + logWarning(std::string("QUIC operator server: malformed command envelope field(s), dropping: ") + e.what()); + } } QUIC_STATUS QUIC_API QuicOperatorServer::listenerCallback(HQUIC listener, void *context, QUIC_LISTENER_EVENT *event) { From f4b40e6a4021ac0e9c66ee6cfa401f72b6ab21cf Mon Sep 17 00:00:00 2001 From: Daniel Prudky Date: Tue, 7 Jul 2026 13:54:27 +0200 Subject: [PATCH 3/4] feat(BAF-1744): switch QUIC operator wire format to protobuf - replace the plain-JSON envelope with OperatorMessage (proto/transparent_operator_stream.proto) in sendStatus() and handleCommandBytes(), matching teleop-module's QuicOperatorServer this was ported from - give it its own .proto/package instead of sharing teleop-module's, since external-server-cpp dlopens both shared libraries into the same process and protobuf's generated-descriptor registry is process-global - keep the BAF-1651 JSON envelope riding as opaque bytes inside StatusUpdate/CommandRequest, so this class stays decoupled from that schema - add Protobuf as a build dependency (cmake/Dependencies.cmake, CMakeLists.txt) and wire PROTOBUF_GENERATE_CPP for the new .proto --- CMakeLists.txt | 13 ++- cmake/Dependencies.cmake | 6 +- .../operator_stream/QuicOperatorServer.hpp | 18 ++-- proto/transparent_operator_stream.proto | 58 +++++++++++++ .../QuicOperatorServer.cpp | 85 +++++++------------ 5 files changed, 111 insertions(+), 69 deletions(-) create mode 100644 proto/transparent_operator_stream.proto diff --git a/CMakeLists.txt b/CMakeLists.txt index cffaae9..79f60d6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -72,10 +72,13 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) FIND_PACKAGE(cpprestsdk REQUIRED) # Operator-facing QUIC transport (BAF-1744), opt-in alongside the Fleet HTTP API above — - # ported from teleop-module's equivalent (BAF-1670). Wire format is plain JSON - # (nlohmann::json, already a dependency above), not protobuf — see QuicOperatorServer's - # class doc comment for why. + # ported from teleop-module's equivalent (BAF-1670): msquic + protobuf for the operator + # stream wire protocol (own proto/transparent_operator_stream.proto — see its file header + # for why it can't share teleop-module's operator_stream.proto). FIND_PACKAGE(msquic CONFIG REQUIRED) + FIND_PACKAGE(Protobuf REQUIRED) + PROTOBUF_GENERATE_CPP(TRANSPARENT_OPERATOR_PROTO_SRCS TRANSPARENT_OPERATOR_PROTO_HDRS + proto/transparent_operator_stream.proto) ENDIF () CMDEF_ADD_LIBRARY( @@ -125,8 +128,11 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) SOURCES "source/external_server_api.cpp" "source/memory_management.cpp" "source/device_management.cpp" "source/transparent_operator_stream/OperatorChannel.cpp" "source/transparent_operator_stream/QuicOperatorServer.cpp" + ${TRANSPARENT_OPERATOR_PROTO_SRCS} VERSION ${TRANSPARENT_MODULE_VERSION} ) + # Generated transparent_operator_stream.pb.h lands in the binary dir. + TARGET_INCLUDE_DIRECTORIES(transparent-external-server-shared PRIVATE ${CMAKE_CURRENT_BINARY_DIR}) # Explicitly add Boost dependencies due to obsolete cpprestsdk cmake export package TARGET_LINK_LIBRARIES(fleet-http-client-shared::fleet-http-client-shared INTERFACE @@ -139,6 +145,7 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) transparent_module_sources fleet-http-client-shared::fleet-http-client-shared msquic + protobuf::libprotobuf PUBLIC fleet-protocol-interface::module-maintainer-external-server-interface ) diff --git a/cmake/Dependencies.cmake b/cmake/Dependencies.cmake index 298cea0..b1cbbf8 100644 --- a/cmake/Dependencies.cmake +++ b/cmake/Dependencies.cmake @@ -11,8 +11,10 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) BA_PACKAGE_LIBRARY(boost v1.86.0) BA_PACKAGE_LIBRARY(cpprestsdk v2.10.20) BA_PACKAGE_LIBRARY(zlib v1.3.2) - # Operator-facing QUIC transport (BAF-1744, ported from teleop-module's BAF-1670). Wire - # format is plain JSON (nlohmann-json above), not protobuf. + # Operator-facing QUIC transport (BAF-1744, ported from teleop-module's BAF-1670): msquic + + # protobuf for the operator stream protocol (own .proto — see its file header for why it can't + # share teleop-module's). BA_PACKAGE_LIBRARY(msquic v2.5.6) + BA_PACKAGE_LIBRARY(protobuf v4.21.12) ENDIF () diff --git a/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp b/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp index 9f38cef..035c092 100644 --- a/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp +++ b/include/bringauto/transparent_module_utils/operator_stream/QuicOperatorServer.hpp @@ -30,18 +30,18 @@ struct QuicOperatorServerConfig { * the operator leg, used alongside (not instead of) the existing Fleet HTTP API path (BAF-1744; * ported from teleop-module's QuicOperatorServer, BAF-1670, single-operator Phase 1). * - * Wire format is plain JSON (nlohmann::json), not protobuf: QUIC is just a transport, and this - * module's whole design already treats every payload as opaque JSON text (see its README), so a - * protobuf envelope around that JSON would be pure overhead — plus teleop-module's equivalent - * class, which this was ported from, needs its own .proto file distinct from ours anyway (same - * external-server-cpp process, process-global protobuf descriptor pool), a problem plain JSON - * doesn't have at all. Envelope shape: {"kind": "hello"|"status"|"command", "device": {"module_id", - * "type", "role", "name"}, "timestamp_ms": , "payload": }. + * Wire format is protobuf (proto/transparent_operator_stream.proto's OperatorMessage), matching + * teleop-module's QuicOperatorServer this was ported from — but its own .proto/package, since + * external-server-cpp dlopens both teleop-external-server-shared.so and + * transparent-external-server-shared.so into the same process, and protobuf's generated-descriptor + * registry is process-global: two .so's registering the same .proto file/package would crash. The + * BAF-1651 JSON envelope itself still rides as opaque `bytes` inside OperatorMessage's + * StatusUpdate/CommandRequest — this class stays decoupled from that schema. * - * cloud -> operator : sendStatus() opens a short unidirectional stream per envelope{kind:status} + * cloud -> operator : sendStatus() opens a short unidirectional stream per OperatorMessage{status} * (START|FIN), freeing the send buffer on SEND_COMPLETE. * operator -> cloud : each operator command arrives as a unidirectional stream; the receive - * callback accumulates the bytes, parses envelope{kind:command}, and hands + * callback accumulates the bytes, parses OperatorMessage{command}, and hands * it to the OperatorChannel (which unblocks the module's wait_for_command()). * * Threading: msquic invokes the callbacks on its own worker thread(s). Commands cross into the diff --git a/proto/transparent_operator_stream.proto b/proto/transparent_operator_stream.proto new file mode 100644 index 0000000..13248d0 --- /dev/null +++ b/proto/transparent_operator_stream.proto @@ -0,0 +1,58 @@ +syntax = "proto3"; + +// GUI <-> transparent-module QUIC stream protocol (BAF-1744), modeled on teleop-module's +// operator_stream.proto (BAF-1670) — same message shapes, but its own package/file: external-server-cpp +// dlopens both teleop-external-server-shared.so and transparent-external-server-shared.so into the +// same process, and protobuf's generated-descriptor registry is process-global, so two .so's +// registering the same .proto file/package would crash. A distinct package avoids that collision. +// +// Transport: each message travels on its own UNIDIRECTIONAL QUIC stream between the GUI (client) +// and transparent-module's external-server plugin (server). The sender opens a stream, writes the +// serialized OperatorMessage, and closes it with FIN — the stream boundary frames the message, so +// there is NO length prefix and one stream carries exactly one OperatorMessage. +// +// Direction: +// rtsp-server -> GUI : StatusUpdate (device-reported status / BAF-1651 response envelope) +// GUI -> rtsp-server : CommandRequest (a BAF-1651 command envelope) +// +// The BAF-1651 JSON envelope ({type, version, action, id, timestamp, payload}) is passed through +// VERBATIM as opaque bytes — this protocol only carries the Fleet Protocol device_id + timestamp +// needed to route the message, and stays decoupled from the BAF-1651 schema. +// NB: `operator_stream`, not `operator` — `operator` is a C++ keyword (invalid generated namespace). +package bringauto.transparent.operator_stream; + +// Fleet-protocol device coordinates of the addressed device (module 3 / type 1 / role rtsp_server). +message DeviceId { + uint32 module_id = 1; + uint32 type = 2; + string role = 3; + string name = 4; +} + +// rtsp-server -> GUI: one status/response push. +message StatusUpdate { + DeviceId device = 1; + bytes payload = 2; // opaque BAF-1651 JSON envelope + int64 timestamp_ms = 3; +} + +// GUI -> rtsp-server: one streaming-control command. +message CommandRequest { + DeviceId device = 1; + bytes payload = 2; // opaque BAF-1651 JSON envelope + int64 timestamp_ms = 3; +} + +// GUI -> rtsp-server: sent once on connect. Single-operator (Phase 1 of the pattern this was +// ported from) — contents ignored for now. +message OperatorHello { + string operator_id = 1; +} + +message OperatorMessage { + oneof kind { + OperatorHello hello = 1; // GUI -> rtsp-server + StatusUpdate status = 2; // rtsp-server -> GUI + CommandRequest command = 3; // GUI -> rtsp-server + } +} diff --git a/source/transparent_operator_stream/QuicOperatorServer.cpp b/source/transparent_operator_stream/QuicOperatorServer.cpp index 3a4bb07..938ffdc 100644 --- a/source/transparent_operator_stream/QuicOperatorServer.cpp +++ b/source/transparent_operator_stream/QuicOperatorServer.cpp @@ -1,8 +1,8 @@ #include -#include // QuicAddrSetFamily / QuicAddrSetPort helpers +#include "transparent_operator_stream.pb.h" -#include +#include // QuicAddrSetFamily / QuicAddrSetPort helpers #include #include @@ -197,27 +197,19 @@ QuicOperatorServer::SendResult QuicOperatorServer::sendStatus(std::uint32_t modu return SendResult::NoOperator; // no operator connected — drop (best-effort) } - // payload is opaque bytes that happen to already be JSON text almost always (see the class doc - // comment) — embed it as a nested JSON value so the wire form has no extra string-escaping - // layer, falling back to a plain string if it genuinely isn't valid JSON. - nlohmann::json envelope; - envelope["kind"] = "status"; - envelope["device"] = {{"module_id", module_id}, {"type", device_type}, {"role", device_role}, {"name", device_name}}; - envelope["timestamp_ms"] = timestamp_ms; - try { - envelope["payload"] = nlohmann::json::parse(payload, payload + payload_size); - } catch (const nlohmann::json::exception &) { - envelope["payload"] = std::string(reinterpret_cast(payload), payload_size); - } + bringauto::transparent::operator_stream::OperatorMessage message; + auto *status = message.mutable_status(); + auto *device = status->mutable_device(); + device->set_module_id(module_id); + device->set_type(device_type); + device->set_role(device_role); + device->set_name(device_name); + status->set_payload(payload, payload_size); + status->set_timestamp_ms(timestamp_ms); std::string serialized; - try { - // replace (not the default strict): the non-JSON fallback above can embed arbitrary bytes as - // a JSON string, and strict dump() throws on invalid UTF-8 in that string, dropping the whole - // status. replace swaps invalid sequences for U+FFFD instead of failing the send. - serialized = envelope.dump(-1, ' ', false, nlohmann::json::error_handler_t::replace); - } catch (const nlohmann::json::exception &e) { - logError(std::string("QUIC operator server: failed to serialize status, dropping: ") + e.what()); + if (!message.SerializeToString(&serialized)) { + logError("QUIC operator server: failed to serialize status, dropping"); return SendResult::SendFailed; } @@ -256,43 +248,26 @@ QuicOperatorServer::SendResult QuicOperatorServer::sendStatus(std::uint32_t modu } void QuicOperatorServer::handleCommandBytes(const std::string &bytes) { - nlohmann::json envelope; - try { - envelope = nlohmann::json::parse(bytes); - } catch (const nlohmann::json::exception &e) { - logWarning("QUIC operator server: failed to parse command JSON (" + std::to_string(bytes.size()) + - " bytes): " + e.what()); + bringauto::transparent::operator_stream::OperatorMessage message; + if (!message.ParseFromString(bytes)) { + logWarning("QUIC operator server: failed to parse OperatorMessage (" + std::to_string(bytes.size()) + + " bytes), dropping"); return; } - // Every field access below can throw nlohmann::json::type_error if a field is present but has - // the wrong JSON type (e.g. "device" is a string, not an object) — this function is invoked - // from commandStreamCallback, a QUIC_API callback handed directly to msquic's C library, which - // is not exception-safe: an uncaught exception here would unwind into msquic and terminate the - // whole external-server process for every device on the module, not just this connection. - try { - if (!envelope.is_object() || envelope.value("kind", std::string{}) != "command") { - // Malformed, or "hello" — ignored for now (single operator, always active). - return; - } - - const auto deviceIt = envelope.find("device"); - const nlohmann::json &device = deviceIt != envelope.end() ? *deviceIt : nlohmann::json::object(); - - OperatorCommand out; - out.timestamp_ms = envelope.value("timestamp_ms", static_cast(0)); - out.module_id = device.value("module_id", 0u); - out.device_type = device.value("type", 0u); - out.device_role = device.value("role", std::string{}); - out.device_name = device.value("name", std::string{}); - // The BAF-1651 envelope was embedded as a nested JSON value by the sender — re-serialize it - // back to bytes here, since OperatorCommand::payload and its downstream consumers (e.g. - // pop_command()) expect a flat JSON string, not a parsed value. - const std::string payloadStr = envelope.contains("payload") ? envelope["payload"].dump() : std::string("{}"); - out.payload.assign(payloadStr.begin(), payloadStr.end()); - channel_.push(std::move(out)); - } catch (const nlohmann::json::exception &e) { - logWarning(std::string("QUIC operator server: malformed command envelope field(s), dropping: ") + e.what()); + if (!message.has_command()) { + // Malformed, or Hello — ignored for now (single operator, always active). + return; } + const auto &command = message.command(); + OperatorCommand out; + const auto &payload = command.payload(); + out.payload.assign(payload.begin(), payload.end()); + out.timestamp_ms = command.timestamp_ms(); + out.module_id = command.device().module_id(); + out.device_type = command.device().type(); + out.device_role = command.device().role(); + out.device_name = command.device().name(); + channel_.push(std::move(out)); } QUIC_STATUS QUIC_API QuicOperatorServer::listenerCallback(HQUIC listener, void *context, QUIC_LISTENER_EVENT *event) { From 507fa9f0c56c5cd6a4f824cfd937a823a042df58 Mon Sep 17 00:00:00 2001 From: Daniel Prudky Date: Tue, 7 Jul 2026 13:55:06 +0200 Subject: [PATCH 4/4] fix(BAF-1744): opt module manager into MG push-only forwarding - implement forward_command_on_receive() to return OK, mirroring teleop-module's module_manager.cpp - without this, MG's default fallback re-invokes generate_command() on every DeviceStatus and testing_device_generate_command() echoes current_command back, so the same command gets redelivered to the device on every status tick forever - returning OK switches MG to push-only mode: new commands forward immediately and a status with nothing new gets an EMPTY DeviceCommand instead of the stale one --- source/module_manager.cpp | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/source/module_manager.cpp b/source/module_manager.cpp index 74ce00b..4a14f5e 100644 --- a/source/module_manager.cpp +++ b/source/module_manager.cpp @@ -45,3 +45,17 @@ int command_data_valid(const buffer command, unsigned int device_type) { return td::testing_device_command_data_valid(command); } + +// BAF-1744: opt into MG's push-only forwarding (mirrors teleop-module/source/module_manager.cpp). +// Without this, MG's default fallback (fleet_protocol/module_maintainer/module_gateway/ +// module_manager.h's forward_command_on_receive doc comment) re-invokes generate_command() with +// the same current_command on every single DeviceStatus when there's nothing new queued from the +// External Server — and testing_device_generate_command() is a pure passthrough that echoes +// current_command right back, so the device sees the same command redelivered on every status +// tick forever. Returning OK here switches MG to push-only mode: a new command from ES is +// forwarded immediately (ModuleHandler::handleCommandForward), and a status with nothing new gets +// an EMPTY DeviceCommand back (ModuleHandler::handleStatus) instead of the stale one. +int forward_command_on_receive(unsigned int device_type) +{ + return OK; +}