diff --git a/cmvr-es/config/README.md b/cmvr-es/config/README.md index e127e5c4..2375dbdc 100644 --- a/cmvr-es/config/README.md +++ b/cmvr-es/config/README.md @@ -73,9 +73,45 @@ cmvr_es.pb.txt - 新增 loader 对不认识的 enum 和未设置的 oneof 必须明确失败；当前个别历史路径仍有退化默认行为，不应复制； - 设备端口、坐标系、速度和单位写入注释； - `enable` 应由 manager 层控制，后端内部的 enable 字段不能替代 manager 开关； -- QUIC 需要 TaskManager 与 `QuicEdgeConfig.enable` 同时开启； +- QUIC 任务需要在 TaskManager 中显式开启； - QUIC 零媒体轨道是合法配置。 +### QUIC 多平台 + +`QuicEdgeTask` 可以同时连接多个平台。原有顶层 +`server_host`、`server_port`、`tls` 继续表示主平台；每个 `platforms` 条目会与主平台 +并行运行。也可以不配置顶层目标，只使用一个或多个 `platforms` 条目： + +```protobuf +platforms { + id: "operations" + server_host: "192.168.0.222" + server_port: 4433 + enable_media: false + tls { + ca_file: "certs/cmvr-quic-ca.crt" + server_name: "192.168.0.222" + } +} +platforms { + id: "analytics" + server_host: "192.168.0.223" + server_port: 4433 + enable_media: true + tls { + ca_file: "certs/cmvr-quic-ca.crt" + server_name: "192.168.0.223" + } +} +``` + +平台 ID 和 `host:port` 必须分别唯一；配置兼容主平台时，其平台 ID 使用任务 +`QuicEdgeConfig.id`，新增条目也不能与它重名。每个平台拥有独立的 QUIC 连接、 +注册会话、心跳序号、ACK 超时和重连退避，一个平台断线不会阻塞其他平台。 +`enable_media` 默认为 `false`，此时仍发送注册、心跳、网络接口和设备状态，但不会 +复制音视频；设为 `true` 才会把全局 `tracks` 转发到该平台。兼容的顶层主平台保持 +原有媒体行为。 + ### gRPC 相机实时流 [`tasks/grpc_server_task/grpc_server_task.pb.txt`](tasks/grpc_server_task/grpc_server_task.pb.txt) diff --git a/cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt b/cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt index d502fa57..a886cfbf 100644 --- a/cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt +++ b/cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt @@ -29,6 +29,21 @@ quic_edge { allow_insecure: false } + # Additional platforms run concurrently with the primary endpoint above. + # Each one has an independent connection, registration, heartbeat ACK state + # and reconnect loop. Media forwarding is opt-in for additional platforms. + # platforms { + # id: "operations_backup" + # server_host: "192.168.0.223" + # server_port: 4433 + # enable_media: false + # tls { + # ca_file: "certs/cmvr-quic-ca.crt" + # server_name: "192.168.0.223" + # allow_insecure: false + # } + # } + reconnect { initial_delay_ms: 500 maximum_delay_ms: 30000 diff --git a/cmvr-es/task/quic_edge_task/include/quic_edge_task.h b/cmvr-es/task/quic_edge_task/include/quic_edge_task.h index 8e10575d..5e46a54f 100644 --- a/cmvr-es/task/quic_edge_task/include/quic_edge_task.h +++ b/cmvr-es/task/quic_edge_task/include/quic_edge_task.h @@ -1,9 +1,11 @@ #ifndef CMVR_ES_QUIC_EDGE_TASK_H #define CMVR_ES_QUIC_EDGE_TASK_H +#include #include #include #include +#include #include "cmvr/config/quic_edge_config/quic_edge_config.pb.h" #include "service/quic_edge/include/quic_edge_service.h" @@ -13,7 +15,12 @@ namespace cmvr::task { class QuicEdgeTask final : public Task { public: + using ServiceFactory = std::function( + config::QuicEdgeConfig)>; + explicit QuicEdgeTask(const config::QuicEdgeConfig& config); + QuicEdgeTask(const config::QuicEdgeConfig& config, + ServiceFactory service_factory); ~QuicEdgeTask() override; const std::string& id() const override { return id_; } @@ -32,11 +39,19 @@ public: std::string detailStatusString() const override; private: + struct PlatformService { + std::string id; + bool media_enabled{false}; + config::QuicEdgeConfig config; + std::unique_ptr service; + }; + TaskState mappedState() const; config::QuicEdgeConfig config_; std::string id_; - std::unique_ptr service_; + ServiceFactory service_factory_; + std::vector services_; mutable std::mutex mutex_; TaskState state_{TaskState::UNINITIALIZED}; diff --git a/cmvr-es/task/quic_edge_task/src/quic_edge_task.cpp b/cmvr-es/task/quic_edge_task/src/quic_edge_task.cpp index 941615e7..a37bb855 100644 --- a/cmvr-es/task/quic_edge_task/src/quic_edge_task.cpp +++ b/cmvr-es/task/quic_edge_task/src/quic_edge_task.cpp @@ -1,7 +1,10 @@ #include "task/quic_edge_task/include/quic_edge_task.h" +#include #include +#include #include +#include #include "cmvr/config/task_manager_config/task_manager_config.pb.h" #include "common/base/logging/logger.h" @@ -13,6 +16,107 @@ namespace cmvr::task { namespace { +struct PlatformConfig { + std::string id; + bool media_enabled{false}; + config::QuicEdgeConfig config; +}; + +void resolveTlsFiles(config::QuicEdgeTlsConfig* tls) +{ + if (!tls) return; + if (!tls->ca_file().empty()) { + tls->set_ca_file(ConfigHelper::resolveConfigFile(tls->ca_file())); + } + if (!tls->certificate_file().empty()) { + tls->set_certificate_file( + ConfigHelper::resolveConfigFile(tls->certificate_file())); + } + if (!tls->private_key_file().empty()) { + tls->set_private_key_file( + ConfigHelper::resolveConfigFile(tls->private_key_file())); + } +} + +bool appendPlatformConfig( + const config::QuicEdgeConfig& root_config, + const std::string& platform_id, + const std::string& server_host, + const std::uint32_t server_port, + const config::QuicEdgeTlsConfig& tls, + const bool media_enabled, + std::unordered_set* platform_ids, + std::set>* endpoints, + std::vector* platforms, + std::string* error) +{ + if (platform_id.empty()) { + if (error) *error = "QUIC edge platform id is empty"; + return false; + } + if (!platform_ids->insert(platform_id).second) { + if (error) *error = "duplicate QUIC edge platform id: " + platform_id; + return false; + } + if (!endpoints->emplace(server_host, server_port).second) { + if (error) { + *error = "duplicate QUIC edge platform endpoint: " + server_host + + ':' + std::to_string(server_port); + } + return false; + } + + PlatformConfig platform; + platform.id = platform_id; + platform.media_enabled = media_enabled; + platform.config = root_config; + platform.config.set_server_host(server_host); + platform.config.set_server_port(server_port); + *platform.config.mutable_tls() = tls; + platform.config.clear_platforms(); + if (!media_enabled) platform.config.clear_tracks(); + platforms->push_back(std::move(platform)); + return true; +} + +bool expandPlatformConfigs(const config::QuicEdgeConfig& config, + std::vector* platforms, + std::string* error) +{ + if (!platforms) { + if (error) *error = "QUIC edge platform output is null"; + return false; + } + platforms->clear(); + std::unordered_set platform_ids; + std::set> endpoints; + + const bool has_primary_endpoint = !config.server_host().empty() || + config.server_port() != 0U || config.has_tls(); + if (has_primary_endpoint && + !appendPlatformConfig( + config, config.id(), config.server_host(), config.server_port(), + config.tls(), true, &platform_ids, &endpoints, platforms, error)) { + return false; + } + + for (const auto& configured_platform : config.platforms()) { + if (!appendPlatformConfig( + config, configured_platform.id(), + configured_platform.server_host(), + configured_platform.server_port(), configured_platform.tls(), + configured_platform.enable_media(), &platform_ids, &endpoints, + platforms, error)) { + return false; + } + } + if (platforms->empty()) { + if (error) *error = "QUIC edge has no platform endpoint"; + return false; + } + return true; +} + std::shared_ptr createQuicEdgeTask(const config::TaskConfigEntry& entry) { if (entry.id().empty() || entry.config_file().empty()) { @@ -31,17 +135,9 @@ std::shared_ptr createQuicEdgeTask(const config::TaskConfigEntry& entry) << ", config=" << config.id(); return nullptr; } - if (!config.tls().ca_file().empty()) { - config.mutable_tls()->set_ca_file( - ConfigHelper::resolveConfigFile(config.tls().ca_file())); - } - if (!config.tls().certificate_file().empty()) { - config.mutable_tls()->set_certificate_file( - ConfigHelper::resolveConfigFile(config.tls().certificate_file())); - } - if (!config.tls().private_key_file().empty()) { - config.mutable_tls()->set_private_key_file( - ConfigHelper::resolveConfigFile(config.tls().private_key_file())); + if (config.has_tls()) resolveTlsFiles(config.mutable_tls()); + for (auto& platform : *config.mutable_platforms()) { + if (platform.has_tls()) resolveTlsFiles(platform.mutable_tls()); } if (config.software_version().empty()) { config.set_software_version(device::DeviceManager::getInstance().version()); @@ -52,7 +148,19 @@ std::shared_ptr createQuicEdgeTask(const config::TaskConfigEntry& entry) } // namespace QuicEdgeTask::QuicEdgeTask(const config::QuicEdgeConfig& config) - : config_(config), id_(config.id()) + : QuicEdgeTask( + config, + [](config::QuicEdgeConfig platform_config) { + return std::make_unique( + std::move(platform_config)); + }) +{ +} + +QuicEdgeTask::QuicEdgeTask(const config::QuicEdgeConfig& config, + ServiceFactory service_factory) + : config_(config), id_(config.id()), + service_factory_(std::move(service_factory)) { } @@ -69,13 +177,40 @@ bool QuicEdgeTask::init() state_ = TaskState::FAILED; return false; } - service_ = std::make_unique(config_); + + std::vector platform_configs; std::string error; - if (!service_->initialize(&error)) { + if (!service_factory_) { + last_error_ = "QUIC edge service factory is empty"; + state_ = TaskState::FAILED; + return false; + } + if (!expandPlatformConfigs(config_, &platform_configs, &error)) { last_error_ = std::move(error); state_ = TaskState::FAILED; return false; } + + std::vector initialized_services; + initialized_services.reserve(platform_configs.size()); + for (auto& platform_config : platform_configs) { + auto service = service_factory_(platform_config.config); + if (!service) { + last_error_ = "failed to create QUIC edge service for platform " + + platform_config.id; + state_ = TaskState::FAILED; + return false; + } + if (!service->initialize(&error)) { + last_error_ = "platform " + platform_config.id + ": " + error; + state_ = TaskState::FAILED; + return false; + } + initialized_services.push_back(PlatformService{ + std::move(platform_config.id), platform_config.media_enabled, + std::move(platform_config.config), std::move(service)}); + } + services_ = std::move(initialized_services); last_error_.clear(); state_ = TaskState::IDLE; return true; @@ -101,20 +236,30 @@ bool QuicEdgeTask::start() admission.generation() == admission_generation; } if ((state_ != TaskState::IDLE && state_ != TaskState::STOPPED) || - !service_) { + services_.empty()) { last_error_ = "QUIC edge task is not initialized"; state_ = TaskState::FAILED; return false; } - std::string error; - if (!service_->start(&error)) { - last_error_ = std::move(error); - state_ = TaskState::FAILED; - return false; + + std::size_t started_services = 0U; + for (auto& platform : services_) { + std::string error; + if (!platform.service->start(&error)) { + for (std::size_t index = 0U; + index < started_services; ++index) { + services_[index].service->stop(); + } + last_error_ = "platform " + platform.id + ": " + error; + state_ = TaskState::FAILED; + return false; + } + ++started_services; } last_error_.clear(); state_ = TaskState::RUNNING; - CMVR_LOG(INFO) << "[QuicEdgeTask] Started, id=" << id_; + CMVR_LOG(INFO) << "[QuicEdgeTask] Started, id=" << id_ + << ", platforms=" << services_.size(); } bool admission_current = false; @@ -142,26 +287,37 @@ void QuicEdgeTask::stop() std::lock_guard lock(mutex_); // stop() is the task lifecycle terminator used by TaskManager shutdown. // Unlike StopAll's stopActivity(), it intentionally tears down transport. - if (service_) service_->stop(); + for (auto& platform : services_) platform.service->stop(); if (state_ != TaskState::FAILED) state_ = TaskState::STOPPED; } bool QuicEdgeTask::stopActivity() { std::lock_guard lock(mutex_); - if (!service_ || state_ == TaskState::FAILED) { + if (services_.empty() || state_ == TaskState::FAILED) { return false; } // System StopAll must preserve the QUIC presence channel. Only old media // subscriptions are fenced; registration and heartbeat stay online. - return service_->interruptMediaActivities(); + bool all_stopped = true; + for (auto& platform : services_) { + if (!platform.service->interruptMediaActivities()) { + all_stopped = false; + } + } + return all_stopped; } TaskState QuicEdgeTask::mappedState() const { - if (!service_ || state_ != TaskState::RUNNING) return state_; - return service_->state() == quic_edge::QuicEdgeServiceState::FAILED - ? TaskState::FAILED : state_; + if (services_.empty() || state_ != TaskState::RUNNING) return state_; + for (const auto& platform : services_) { + if (platform.service->state() == + quic_edge::QuicEdgeServiceState::FAILED) { + return TaskState::FAILED; + } + } + return state_; } TaskState QuicEdgeTask::state() const @@ -195,31 +351,38 @@ std::string QuicEdgeTask::detailStatusString() const std::lock_guard lock(mutex_); std::ostringstream output; output << taskStateToString(mappedState()); - if (service_) { - const auto stats = service_->stats(); - const auto service_status = service_->status(); - output << " service=" << quic_edge::toString(service_->state()) - << " target=" << config_.server_host() << ':' << config_.server_port() - << " node_id=" << service_status.node_id - << " registered=" << service_status.registered - << " connections=" << stats.successful_connections - << " registrations=" << stats.registrations_accepted - << " heartbeat_sequence=" << service_status.heartbeat_sequence - << " heartbeat_acks=" << stats.heartbeats_acknowledged - << " media_tracks=" << service_status.active_media_tracks - << " frames=" << stats.frames_queued - << " datagrams=" << stats.datagrams_queued; - if (!service_status.session_id.empty()) { - output << " session=" << service_status.session_id; + if (!services_.empty()) { + output << " platforms=" << services_.size(); + for (const auto& platform : services_) { + const auto stats = platform.service->stats(); + const auto service_status = platform.service->status(); + output << " platform[" << platform.id << "]={" + << "service=" << quic_edge::toString(platform.service->state()) + << ",target=" << platform.config.server_host() << ':' + << platform.config.server_port() + << ",media_enabled=" << platform.media_enabled + << ",node_id=" << service_status.node_id + << ",registered=" << service_status.registered + << ",connections=" << stats.successful_connections + << ",registrations=" << stats.registrations_accepted + << ",heartbeat_sequence=" << service_status.heartbeat_sequence + << ",heartbeat_acks=" << stats.heartbeats_acknowledged + << ",media_tracks=" << service_status.active_media_tracks + << ",frames=" << stats.frames_queued + << ",datagrams=" << stats.datagrams_queued; + if (!service_status.session_id.empty()) { + output << ",session=" << service_status.session_id; + } + if (!service_status.observed_source_ip.empty()) { + output << ",observed_ip=" << service_status.observed_source_ip; + } + if (!service_status.last_media_error.empty()) { + output << ",media_error=" << service_status.last_media_error; + } + const std::string service_error = platform.service->lastError(); + if (!service_error.empty()) output << ",error=" << service_error; + output << '}'; } - if (!service_status.observed_source_ip.empty()) { - output << " observed_ip=" << service_status.observed_source_ip; - } - if (!service_status.last_media_error.empty()) { - output << " media_error=" << service_status.last_media_error; - } - const std::string service_error = service_->lastError(); - if (!service_error.empty()) output << " error=" << service_error; } else if (!last_error_.empty()) { output << " error=" << last_error_; } diff --git a/cmvr-es/task/quic_edge_task/tests/quic_edge_task_test.cpp b/cmvr-es/task/quic_edge_task/tests/quic_edge_task_test.cpp index e0c1b0b4..99a05f00 100644 --- a/cmvr-es/task/quic_edge_task/tests/quic_edge_task_test.cpp +++ b/cmvr-es/task/quic_edge_task/tests/quic_edge_task_test.cpp @@ -1,10 +1,179 @@ +#include +#include +#include +#include +#include #include +#include +#include +#include +#include +#include +#include +#include "cmvr/quic_edge/v1/quic_edge.pb.h" +#include "manager/media_source_hub/include/media_source_hub.h" +#include "service/quic_edge/include/control_framing.h" #include "service/stop_all/include/stop_all_admission_gate.h" #include "task/quic_edge_task/include/quic_edge_task.h" namespace { +using Clock = std::chrono::steady_clock; +using QuicEdgeService = cmvr::quic_edge::QuicEdgeService; + +class HeartbeatTransport final : public cmvr::quic_edge::QuicTransport { +public: + explicit HeartbeatTransport(std::string platform_id) + : platform_id_(std::move(platform_id)) + { + } + + bool connect(const cmvr::config::QuicEdgeConfig&, + std::chrono::milliseconds, + std::string*) override + { + connected_.store(true); + return true; + } + + void disconnect() override + { + connected_.store(false); + condition_.notify_all(); + } + + bool isConnected() const override { return connected_.load(); } + std::size_t maximumDatagramBytes() const override { return 1200U; } + std::size_t maximumDatagramBatchPackets() const override { return 32U; } + + cmvr::quic_edge::TransportSendResult sendControl( + std::vector framed_message, + std::string*) override + { + if (!connected_.load()) { + return cmvr::quic_edge::TransportSendResult::DISCONNECTED; + } + + std::vector> frames; + std::string error; + cmvr::quic_edge::ControlFrameDecoder decoder(1024U * 1024U); + if (!decoder.push(framed_message, &frames, &error) || + frames.size() != 1U) { + return cmvr::quic_edge::TransportSendResult::ERROR; + } + cmvr::quic_edge::v1::EdgeControlEnvelope request; + if (!request.ParseFromArray( + frames.front().data(), + static_cast(frames.front().size()))) { + return cmvr::quic_edge::TransportSendResult::ERROR; + } + + cmvr::quic_edge::v1::EdgeControlEnvelope response; + response.set_protocol_version(cmvr::quic_edge::kProtocolVersion); + { + std::lock_guard lock(mutex_); + response.set_message_sequence(server_message_sequence_++); + if (request.has_node_register_request()) { + ++registrations_; + auto* registration = response.mutable_node_register_response(); + registration->set_accepted(true); + registration->set_session_id("session-" + platform_id_); + registration->set_heartbeat_interval_ms(250U); + } else if (request.has_node_heartbeat()) { + const auto& heartbeat = request.node_heartbeat(); + ++heartbeats_; + last_heartbeat_sequence_.store(heartbeat.sequence()); + auto* ack = response.mutable_node_heartbeat_ack(); + ack->set_accepted(true); + ack->set_session_id(heartbeat.session_id()); + ack->set_acknowledged_sequence(heartbeat.sequence()); + } else { + return cmvr::quic_edge::TransportSendResult::ERROR; + } + if (!enqueueResponse(response)) { + return cmvr::quic_edge::TransportSendResult::ERROR; + } + } + condition_.notify_all(); + return cmvr::quic_edge::TransportSendResult::QUEUED; + } + + cmvr::quic_edge::TransportReceiveResult receiveControl( + std::vector* chunk, + const std::chrono::milliseconds timeout, + std::string*) override + { + if (!chunk) return cmvr::quic_edge::TransportReceiveResult::ERROR; + std::unique_lock lock(mutex_); + condition_.wait_for(lock, timeout, [this] { + return !responses_.empty() || !connected_.load(); + }); + if (!responses_.empty()) { + *chunk = std::move(responses_.front()); + responses_.pop_front(); + return cmvr::quic_edge::TransportReceiveResult::DATA; + } + return connected_.load() + ? cmvr::quic_edge::TransportReceiveResult::TIMEOUT + : cmvr::quic_edge::TransportReceiveResult::DISCONNECTED; + } + + cmvr::quic_edge::TransportSendResult sendDatagramBatch( + std::vector, + std::string*) override + { + return connected_.load() + ? cmvr::quic_edge::TransportSendResult::QUEUED + : cmvr::quic_edge::TransportSendResult::DISCONNECTED; + } + + std::uint64_t registrations() const { return registrations_.load(); } + std::uint64_t heartbeats() const { return heartbeats_.load(); } + std::uint64_t lastHeartbeatSequence() const + { + return last_heartbeat_sequence_.load(); + } + +private: + bool enqueueResponse( + const cmvr::quic_edge::v1::EdgeControlEnvelope& response) + { + std::string serialized; + if (!response.SerializeToString(&serialized)) return false; + std::vector framed; + std::string error; + if (!cmvr::quic_edge::ControlFrameEncoder::encode( + reinterpret_cast(serialized.data()), + serialized.size(), 1024U * 1024U, &framed, &error)) { + return false; + } + responses_.push_back(std::move(framed)); + return true; + } + + std::string platform_id_; + std::atomic connected_{false}; + std::atomic registrations_{0U}; + std::atomic heartbeats_{0U}; + std::atomic last_heartbeat_sequence_{0U}; + mutable std::mutex mutex_; + std::condition_variable condition_; + std::deque> responses_; + std::uint64_t server_message_sequence_{0U}; +}; + +template +bool waitUntil(const std::chrono::milliseconds timeout, Predicate predicate) +{ + const auto deadline = Clock::now() + timeout; + while (Clock::now() < deadline) { + if (predicate()) return true; + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + return predicate(); +} + cmvr::config::QuicEdgeConfig validConfig() { cmvr::config::QuicEdgeConfig config; @@ -32,6 +201,36 @@ cmvr::config::QuicEdgeConfig validConfig() return config; } +cmvr::config::QuicEdgeConfig multiPlatformConfig() +{ + auto config = validConfig(); + config.set_id("quic-multi-platform-test"); + config.clear_server_host(); + config.clear_server_port(); + config.clear_tls(); + + auto* platform_a = config.add_platforms(); + platform_a->set_id("platform-a"); + platform_a->set_server_host("192.0.2.10"); + platform_a->set_server_port(4433U); + platform_a->mutable_tls()->set_allow_insecure(true); + + auto* platform_b = config.add_platforms(); + platform_b->set_id("platform-b"); + platform_b->set_server_host("198.51.100.20"); + platform_b->set_server_port(4434U); + platform_b->mutable_tls()->set_allow_insecure(true); + platform_b->set_enable_media(true); + + auto* disabled_track = config.add_tracks(); + disabled_track->set_track_id(1U); + disabled_track->set_source_kind( + cmvr::config::QuicEdgeTrackConfig::SOURCE_KIND_CAMERA); + disabled_track->set_device_id("disabled-test-camera"); + disabled_track->set_enable(false); + return config; +} + } // namespace int main() @@ -97,7 +296,93 @@ int main() std::cerr << "QUIC service did not remain available after StopAll\n"; return 1; } + admission_task.stop(); admission.clearForTesting(); + + cmvr::media::MediaSourceHub media_hub; + std::vector transports; + std::vector services; + std::vector expanded_configs; + cmvr::task::QuicEdgeTask multi_platform_task( + multiPlatformConfig(), + [&](cmvr::config::QuicEdgeConfig platform_config) { + expanded_configs.push_back(platform_config); + auto transport = std::make_unique( + platform_config.server_host()); + transports.push_back(transport.get()); + auto service = std::make_unique( + std::move(platform_config), std::move(transport), media_hub, + [] { return cmvr::device::DeviceManagerSnapshot{}; }); + services.push_back(service.get()); + return service; + }); + if (!multi_platform_task.init() || expanded_configs.size() != 2U || + transports.size() != 2U || services.size() != 2U) { + std::cerr << "multi-platform QUIC task did not create two services: " + << multi_platform_task.detailStatusString() << '\n'; + return 1; + } + if (expanded_configs[0].platforms_size() != 0 || + expanded_configs[1].platforms_size() != 0 || + expanded_configs[0].server_host() != "192.0.2.10" || + expanded_configs[1].server_host() != "198.51.100.20" || + expanded_configs[0].tracks_size() != 0 || + expanded_configs[1].tracks_size() != 1) { + std::cerr << "multi-platform QUIC task expanded invalid service configs\n"; + return 1; + } + if (!multi_platform_task.start()) { + std::cerr << "multi-platform QUIC task did not start: " + << multi_platform_task.detailStatusString() << '\n'; + return 1; + } + const bool both_platforms_online = waitUntil( + std::chrono::seconds(2), [&] { + return transports[0]->registrations() >= 1U && + transports[1]->registrations() >= 1U && + transports[0]->heartbeats() >= 1U && + transports[1]->heartbeats() >= 1U && + services[0]->stats().heartbeats_acknowledged >= 1U && + services[1]->stats().heartbeats_acknowledged >= 1U; + }); + if (!both_platforms_online || + transports[0]->lastHeartbeatSequence() != 1U || + transports[1]->lastHeartbeatSequence() != 1U) { + std::cerr << "both platforms did not independently ACK a heartbeat: " + << multi_platform_task.detailStatusString() << '\n'; + return 1; + } + const std::string multi_status = multi_platform_task.detailStatusString(); + if (multi_status.find("platforms=2") == std::string::npos || + multi_status.find("platform[platform-a]") == std::string::npos || + multi_status.find("platform[platform-b]") == std::string::npos) { + std::cerr << "multi-platform QUIC status is incomplete: " + << multi_status << '\n'; + return 1; + } + multi_platform_task.stop(); + if (multi_platform_task.state() != cmvr::task::TaskState::STOPPED) { + std::cerr << "multi-platform QUIC task did not stop cleanly\n"; + return 1; + } + + auto duplicate_endpoint_config = validConfig(); + auto* duplicate_platform = duplicate_endpoint_config.add_platforms(); + duplicate_platform->set_id("duplicate-primary"); + duplicate_platform->set_server_host( + duplicate_endpoint_config.server_host()); + duplicate_platform->set_server_port( + duplicate_endpoint_config.server_port()); + duplicate_platform->mutable_tls()->set_allow_insecure(true); + cmvr::task::QuicEdgeTask duplicate_endpoint_task( + duplicate_endpoint_config); + if (duplicate_endpoint_task.init() || + duplicate_endpoint_task.detailStatusString().find( + "duplicate QUIC edge platform endpoint") == std::string::npos) { + std::cerr << "duplicate QUIC platform endpoint was not rejected\n"; + return 1; + } + std::cout << "quic_edge_task_test: PASS\n"; return 0; } diff --git a/protos/cmvr/config/quic_edge_config/quic_edge_config.proto b/protos/cmvr/config/quic_edge_config/quic_edge_config.proto index 48486a9c..8e01ef93 100644 --- a/protos/cmvr/config/quic_edge_config/quic_edge_config.proto +++ b/protos/cmvr/config/quic_edge_config/quic_edge_config.proto @@ -43,9 +43,27 @@ message QuicEdgeTrackConfig { string source_track_id = 6; } +message QuicEdgePlatformConfig { + // Stable operator-facing identifier used in logs and task status. It must be + // unique within the task and differ from QuicEdgeConfig.id when the + // backward-compatible primary endpoint is also configured. + string id = 1; + string server_host = 2; + uint32 server_port = 3; + QuicEdgeTlsConfig tls = 4; + + // False keeps registration, heartbeat, IP and device-state reporting while + // preventing accidental duplication of configured media tracks. + bool enable_media = 5; +} + message QuicEdgeConfig { string id = 1; reserved 2; + + // Backward-compatible primary platform endpoint. When configured, it runs + // alongside every endpoint in platforms. New configurations may leave these + // fields empty and declare every destination in platforms instead. string server_host = 3; uint32 server_port = 4; string alpn = 5; @@ -87,6 +105,10 @@ message QuicEdgeConfig { // Immutable robot product identity, for example: // CN-CMVR-MBLRV1-CHAGAN-20260731-001. string robot_id = 22; + + // Every platform owns an independent QUIC connection, registration, + // heartbeat/ACK sequence and reconnect loop. + repeated QuicEdgePlatformConfig platforms = 23; } message QuicEdgeRootConfig { diff --git a/protos/cmvr/quic_edge/v1/README.md b/protos/cmvr/quic_edge/v1/README.md index be3c2178..f34377e8 100644 --- a/protos/cmvr/quic_edge/v1/README.md +++ b/protos/cmvr/quic_edge/v1/README.md @@ -85,8 +85,8 @@ their exact scope and manual gateway options. To enable node presence without media: -1. Configure the gateway address, ALPN `cmvr-quic-edge/1`, TLS trust, node ID, - advertised gRPC endpoint and heartbeat values in +1. Configure one or more gateway addresses, ALPN `cmvr-quic-edge/1`, TLS + trust, node ID, advertised gRPC endpoint and heartbeat values in `cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt`. Certificate paths are relative to the cmvr-es configuration root, for example `certs/quic_gateway_ca.pem`. @@ -99,6 +99,14 @@ media must not suppress node registration, IP reporting or heartbeat. `maximum_frame_bytes` must fit in one atomically admitted DATAGRAM batch; reduce it or increase `datagram_send_queue_depth` when changing the DATAGRAM size. +The backward-compatible top-level `server_host`, `server_port` and `tls` +describe the primary platform. Every `platforms` entry runs concurrently with +that primary endpoint, or the list may be used by itself. Platform IDs and +`host:port` pairs must be unique. Each platform owns its own connection, +registration session, heartbeat/ACK sequence and reconnect loop. Additional +platforms default to presence-only operation; set their `enable_media=true` to +forward the shared `tracks` as a separate media session. + ## Reliable control stream The edge opens one bidirectional reliable stream after the QUIC handshake. Both