diff --git a/CMakeLists.txt b/CMakeLists.txt index f4ed7c5..79f60d6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -70,6 +70,15 @@ 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): 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( @@ -117,8 +126,13 @@ 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" + ${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 @@ -130,6 +144,8 @@ IF (FLEET_PROTOCOL_BUILD_EXTERNAL_SERVER) PRIVATE 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 c8487bf..b1cbbf8 100644 --- a/cmake/Dependencies.cmake +++ b/cmake/Dependencies.cmake @@ -11,5 +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): 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/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..035c092 --- /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 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 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 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 + * 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/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/external_server_api.cpp b/source/external_server_api.cpp index 10f0dc6..54293cb 100644 --- a/source/external_server_api.cpp +++ b/source/external_server_api.cpp @@ -5,13 +5,98 @@ #include #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 +230,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 +275,23 @@ 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 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, 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; + } + con->fleet_api_client->setDeviceIdentification(bringauto::fleet_protocol::cxx::DeviceID( device.module, device.device_type, 0, // priority @@ -241,6 +359,17 @@ 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 — 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); @@ -289,6 +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 conditional. + std::unique_lock lock(con->mutex, std::defer_lock); + if (con->quic_server) + { + lock.lock(); + } con->devices.emplace_back(new_device); return OK; } @@ -302,6 +437,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/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; +} 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..938ffdc --- /dev/null +++ b/source/transparent_operator_stream/QuicOperatorServer.cpp @@ -0,0 +1,427 @@ +#include + +#include "transparent_operator_stream.pb.h" + +#include // QuicAddrSetFamily / QuicAddrSetPort helpers + +#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) + } + + 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; + if (!message.SerializeToString(&serialized)) { + logError("QUIC operator server: failed to serialize status, dropping"); + 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) { + 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; + } + 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) { + 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