From d5f04745972c22ffd36998c9c5a138fec075be9a Mon Sep 17 00:00:00 2001 From: liaochuntao Date: Thu, 6 Aug 2026 02:25:36 +0800 Subject: [PATCH 1/3] feat: migrate thin sdk to control session --- README.md | 17 ++- context-kg/tasks/todo.md | 13 ++ include/pole/client/sidecar_session.h | 35 +++++ specification | 2 +- src/sidecar_session.cc | 191 ++++++++++++++++++++++++-- tests/sidecar_session_test.cc | 110 +++++++++++++-- 6 files changed, 343 insertions(+), 25 deletions(-) create mode 100644 context-kg/tasks/todo.md diff --git a/README.md b/README.md index 4ab7b3e..ecc9e9f 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,9 @@ `pole-client-cpp` 是框架无关的 Pole C++ Thin SDK。它只负责： -- 通过 Unix Domain Socket 与本 Pod Sidecar 建立内部 gRPC bootstrap 会话； -- 接收 Sidecar 主动下发的一次性 HTTP、gRPC、Dubbo、Thrift listener 快照； +- 通过 Unix Domain Socket 与本 Pod Sidecar 建立内部双向 `OpenControlSession`； +- 首个服务端事件原子安装 HTTP、gRPC、Dubbo、Thrift listener 快照； +- 注册或注销本地服务；SDK 在会话重连后重放仍处于 desired 状态的注册； - 为业务请求生成 `latticehub-target-namespace` 与 `latticehub-target-service` 元信息。 业务请求不会经过 bootstrap gRPC；应用仍按原协议访问 SDK 返回的 @@ -31,6 +32,10 @@ target_link_libraries(your_target PRIVATE PoleClient::pole-client-cpp) `SidecarSessionOptions.socket_path` 或 `POLE_SIDECAR_SOCKET` 覆盖。SDK 不硬编码 任何业务 listener 端口；会话断开后快照立即失效，重连并收到完整首帧后恢复。 +Specification 子模块固定在正式集成提交 +`2642bc29c0a512f4da84ec4eb862b1e1ceee9833`（`develop`，PR #25），该提交已包含 +`OpenControlSession` 契约及官方 C++ 生成方式。 + ## 使用 ```cpp @@ -40,6 +45,14 @@ target_link_libraries(your_target PRIVATE PoleClient::pole-client-cpp) auto session = pole::client::SidecarSession::Connect(); const auto grpc_address = session->ListenerAddress(pole::client::Protocol::kGrpc); +const auto registration_id = session->RegisterLocalService({ + .service_namespace = "default", + .service = "orders", + .protocol = pole::client::Protocol::kGrpc, + .local_port = 50051, +}); +session->UnregisterLocalService(registration_id); + const pole::client::TargetService target("default", "orders"); const auto metadata = target.EncodeMetadata(); ``` diff --git a/context-kg/tasks/todo.md b/context-kg/tasks/todo.md new file mode 100644 index 0000000..f9f691f --- /dev/null +++ b/context-kg/tasks/todo.md @@ -0,0 +1,13 @@ +# Thin SDK OpenControlSession migration + +- [x] Verify the `codex/sidecar-service-session-v2` bootstrap contract. +- [x] Replace legacy `OpenSession` with bidirectional `OpenControlSession`. +- [x] Add local-service registration lifecycle and reconnect replay. +- [x] Update documentation and focused tests. +- [x] Run available validation and review the final diff. + +## Review + +- Contract source: Specification `develop` commit `2642bc29c0a512f4da84ec4eb862b1e1ceee9833`. +- Passed: `git diff --check`; `protoc` descriptor validation for the formal bootstrap proto. +- Blocked: CMake configuration requires Protobuf `5.26.1`; this environment offers only `35.1.0`. diff --git a/include/pole/client/sidecar_session.h b/include/pole/client/sidecar_session.h index ec0ca0a..5fbda47 100644 --- a/include/pole/client/sidecar_session.h +++ b/include/pole/client/sidecar_session.h @@ -4,7 +4,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -26,6 +28,26 @@ enum class Protocol : std::uint8_t { kThrift, }; +struct LocalServiceRegistration { + std::string service_namespace; + std::string service; + Protocol protocol; + std::uint16_t local_port; + std::string registration_id; +}; + +enum class LocalServiceState : std::uint8_t { + kUnspecified, + kRegistered, + kUnregistered, + kRejected, +}; + +struct LocalServiceStatus { + LocalServiceState state; + std::string message; +}; + struct SidecarSessionOptions { std::string socket_path; std::string sdk_version = kSdkVersion; @@ -55,9 +77,14 @@ class SidecarSession final { [[nodiscard]] const std::string& socket_path() const noexcept; [[nodiscard]] bool available() const; [[nodiscard]] std::string ListenerAddress(Protocol protocol) const; + std::string RegisterLocalService(LocalServiceRegistration registration); + bool UnregisterLocalService(const std::string& registration_id); + [[nodiscard]] std::optional GetLocalServiceStatus( + const std::string& registration_id) const; void Close(); private: + struct ActiveStream; struct ListenerSnapshot; explicit SidecarSession(SidecarSessionOptions options); @@ -65,13 +92,21 @@ class SidecarSession final { void InstallSnapshot(std::shared_ptr snapshot); void InvalidateSnapshot(); void SetInitializationError(std::string error); + void SendRegistration(const LocalServiceRegistration& registration); + void SendUnregistration(const std::string& registration_id); + void HandleLocalServiceStatus(LocalServiceStatus status, const std::string& registration_id); SidecarSessionOptions options_; mutable std::mutex mutex_; std::condition_variable condition_; std::shared_ptr listeners_; + std::map desired_registrations_; + std::map local_service_statuses_; std::thread worker_; + std::mutex stream_mutex_; + ActiveStream* active_stream_ = nullptr; grpc::ClientContext* active_context_ = nullptr; + std::uint64_t next_registration_id_ = 1; bool initialized_ = false; bool closed_ = false; std::string initialization_error_; diff --git a/specification b/specification index 1590c0e..2642bc2 160000 --- a/specification +++ b/specification @@ -1 +1 @@ -Subproject commit 1590c0e73c46276e78d3ea3a6e0811c0515eba25 +Subproject commit 2642bc29c0a512f4da84ec4eb862b1e1ceee9833 diff --git a/src/sidecar_session.cc b/src/sidecar_session.cc index 5d407e7..265472e 100644 --- a/src/sidecar_session.cc +++ b/src/sidecar_session.cc @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -34,6 +35,36 @@ std::optional FromWireProtocol(BootstrapProtocol protocol) { } } +BootstrapProtocol ToWireProtocol(Protocol protocol) { + switch (protocol) { + case Protocol::kHttp: + return BootstrapProtocol::PROTOCOL_HTTP; + case Protocol::kGrpc: + return BootstrapProtocol::PROTOCOL_GRPC; + case Protocol::kDubbo: + return BootstrapProtocol::PROTOCOL_DUBBO; + case Protocol::kThrift: + return BootstrapProtocol::PROTOCOL_THRIFT; + } + return BootstrapProtocol::PROTOCOL_UNSPECIFIED; +} + +std::optional FromWireLocalServiceState( + ::pole::sidecar::v1::LocalServiceState state) { + switch (state) { + case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_REGISTERED: + return LocalServiceState::kRegistered; + case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_UNREGISTERED: + return LocalServiceState::kUnregistered; + case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_REJECTED: + return LocalServiceState::kRejected; + case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_UNSPECIFIED: + return LocalServiceState::kUnspecified; + default: + return std::nullopt; + } +} + std::string ResolveSocketPath(const std::string& configured) { if (!configured.empty()) { return configured; @@ -64,6 +95,11 @@ struct SidecarSession::ListenerSnapshot { std::array ports{}; }; +struct SidecarSession::ActiveStream { + ::grpc::ClientReaderWriter<::pole::sidecar::v1::ClientEvent, + ::pole::sidecar::v1::SidecarEvent>* writer; +}; + SidecarUnavailableError::SidecarUnavailableError() : std::runtime_error("Pole Sidecar listener snapshot is unavailable") {} @@ -105,6 +141,52 @@ std::string SidecarSession::ListenerAddress(Protocol protocol) const { return "127.0.0.1:" + std::to_string(listeners_->ports[ProtocolIndex(protocol)]); } +std::string SidecarSession::RegisterLocalService(LocalServiceRegistration registration) { + if (registration.service_namespace.empty() || registration.service.empty() || + registration.local_port == 0) { + throw std::invalid_argument("local service namespace, service, and local_port must be set"); + } + { + std::lock_guard lock(mutex_); + if (closed_) { + throw std::runtime_error("Sidecar session is closed"); + } + if (registration.registration_id.empty()) { + registration.registration_id = "cpp-" + std::to_string(next_registration_id_++); + } + const auto [_, inserted] = desired_registrations_.emplace(registration.registration_id, registration); + if (!inserted) { + throw std::invalid_argument("registration_id is already registered"); + } + } + SendRegistration(registration); + return registration.registration_id; +} + +bool SidecarSession::UnregisterLocalService(const std::string& registration_id) { + if (registration_id.empty()) { + throw std::invalid_argument("registration_id must not be empty"); + } + { + std::lock_guard lock(mutex_); + if (desired_registrations_.erase(registration_id) == 0) { + return false; + } + } + SendUnregistration(registration_id); + return true; +} + +std::optional SidecarSession::GetLocalServiceStatus( + const std::string& registration_id) const { + std::lock_guard lock(mutex_); + const auto status = local_service_statuses_.find(registration_id); + if (status == local_service_statuses_.end()) { + return std::nullopt; + } + return status->second; +} + void SidecarSession::Close() { { std::lock_guard lock(mutex_); @@ -113,6 +195,15 @@ void SidecarSession::Close() { } closed_ = true; listeners_.reset(); + } + { + std::lock_guard stream_lock(stream_mutex_); + if (active_stream_ != nullptr) { + active_stream_->writer->WritesDone(); + } + } + { + std::lock_guard lock(mutex_); if (active_context_ != nullptr) { active_context_->TryCancel(); } @@ -150,6 +241,37 @@ void SidecarSession::SetInitializationError(std::string error) { condition_.notify_all(); } +void SidecarSession::SendRegistration(const LocalServiceRegistration& registration) { + std::lock_guard stream_lock(stream_mutex_); + if (active_stream_ == nullptr) { + return; + } + ::pole::sidecar::v1::ClientEvent event; + auto* wire_registration = event.mutable_register_local_service(); + wire_registration->set_registration_id(registration.registration_id); + wire_registration->set_namespace(registration.service_namespace); + wire_registration->set_service(registration.service); + wire_registration->set_protocol(ToWireProtocol(registration.protocol)); + wire_registration->set_local_port(registration.local_port); + active_stream_->writer->Write(event); +} + +void SidecarSession::SendUnregistration(const std::string& registration_id) { + std::lock_guard stream_lock(stream_mutex_); + if (active_stream_ == nullptr) { + return; + } + ::pole::sidecar::v1::ClientEvent event; + event.mutable_unregister_local_service()->set_registration_id(registration_id); + active_stream_->writer->Write(event); +} + +void SidecarSession::HandleLocalServiceStatus(LocalServiceStatus status, + const std::string& registration_id) { + std::lock_guard lock(mutex_); + local_service_statuses_[registration_id] = std::move(status); +} + void SidecarSession::Run() { auto retry_delay = options_.retry_initial_delay; while (true) { @@ -172,18 +294,51 @@ void SidecarSession::Run() { active_context_ = &context; } - ::pole::sidecar::v1::ClientHello hello; - hello.set_sdk_language(kSdkLanguage); - hello.set_sdk_version(options_.sdk_version); - hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_HTTP); - hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_GRPC); - hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_DUBBO); - hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_THRIFT); + auto stream = stub->OpenControlSession(&context); + ::pole::sidecar::v1::ClientEvent hello_event; + auto* hello = hello_event.mutable_hello(); + hello->set_sdk_language(kSdkLanguage); + hello->set_sdk_version(options_.sdk_version); + hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_HTTP); + hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_GRPC); + hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_DUBBO); + hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_THRIFT); + if (!stream->Write(hello_event)) { + stream->Finish(); + { + std::lock_guard lock(mutex_); + active_context_ = nullptr; + } + SetInitializationError("failed to send ClientHello to Sidecar"); + return; + } + + ActiveStream active_stream{stream.get()}; + { + std::lock_guard stream_lock(stream_mutex_); + active_stream_ = &active_stream; + std::map registrations; + { + std::lock_guard lock(mutex_); + registrations = desired_registrations_; + } + for (const auto& [_, registration] : registrations) { + ::pole::sidecar::v1::ClientEvent event; + auto* wire_registration = event.mutable_register_local_service(); + wire_registration->set_registration_id(registration.registration_id); + wire_registration->set_namespace(registration.service_namespace); + wire_registration->set_service(registration.service); + wire_registration->set_protocol(ToWireProtocol(registration.protocol)); + wire_registration->set_local_port(registration.local_port); + if (!stream->Write(event)) { + break; + } + } + } - auto reader = stub->OpenSession(&context, hello); ::pole::sidecar::v1::SidecarEvent event; bool received_snapshot = false; - if (reader->Read(&event)) { + if (stream->Read(&event)) { if (!event.has_listener_snapshot()) { SetInitializationError("first Sidecar event must be listener_snapshot"); } else { @@ -215,12 +370,26 @@ void SidecarSession::Run() { InstallSnapshot(std::move(snapshot)); received_snapshot = true; retry_delay = options_.retry_initial_delay; - while (reader->Read(&event)) { + while (stream->Read(&event)) { + if (event.has_local_service_status()) { + const auto state = FromWireLocalServiceState(event.local_service_status().state()); + if (state.has_value()) { + HandleLocalServiceStatus( + LocalServiceStatus{*state, event.local_service_status().message()}, + event.local_service_status().registration_id()); + } + } } } } } - reader->Finish(); + { + std::lock_guard stream_lock(stream_mutex_); + if (active_stream_ == &active_stream) { + active_stream_ = nullptr; + } + } + stream->Finish(); { std::lock_guard lock(mutex_); active_context_ = nullptr; diff --git a/tests/sidecar_session_test.cc b/tests/sidecar_session_test.cc index 65f8650..908682b 100644 --- a/tests/sidecar_session_test.cc +++ b/tests/sidecar_session_test.cc @@ -1,9 +1,11 @@ #include "pole/client/sidecar_session.h" +#include #include #include +#include #include -#include +#include #include #include @@ -15,24 +17,75 @@ namespace { class BootstrapService final : public pole::sidecar::v1::SidecarSessionService::Service { public: - grpc::Status OpenSession(grpc::ServerContext* context, - const pole::sidecar::v1::ClientHello* hello, - grpc::ServerWriter* writer) override { - assert(hello->sdk_language() == "cpp"); - assert(hello->supported_protocols_size() == 4); - pole::sidecar::v1::SidecarEvent event; - auto* snapshot = event.mutable_listener_snapshot(); + grpc::Status OpenControlSession( + grpc::ServerContext*, + grpc::ServerReaderWriter* stream) override { + const auto session_number = ++session_count_; + pole::sidecar::v1::ClientEvent event; + assert(stream->Read(&event)); + assert(event.has_hello()); + assert(event.hello().sdk_language() == "cpp"); + assert(event.hello().supported_protocols_size() == 4); + + pole::sidecar::v1::SidecarEvent snapshot_event; + auto* snapshot = snapshot_event.mutable_listener_snapshot(); AddListener(snapshot, pole::sidecar::v1::PROTOCOL_HTTP, 21001); AddListener(snapshot, pole::sidecar::v1::PROTOCOL_GRPC, 21002); AddListener(snapshot, pole::sidecar::v1::PROTOCOL_DUBBO, 21003); AddListener(snapshot, pole::sidecar::v1::PROTOCOL_THRIFT, 21004); - writer->Write(event); - while (!context->IsCancelled()) { - std::this_thread::sleep_for(std::chrono::milliseconds(10)); + assert(stream->Write(snapshot_event)); + + assert(stream->Read(&event)); + assert(event.has_register_local_service()); + const auto& registration = event.register_local_service(); + assert(registration.namespace_() == "default"); + assert(registration.service() == "orders"); + assert(registration.protocol() == pole::sidecar::v1::PROTOCOL_GRPC); + assert(registration.local_port() == 50051); + SendStatus(stream, registration.registration_id(), + pole::sidecar::v1::LOCAL_SERVICE_STATE_REGISTERED); + + if (session_number == 1) { + return grpc::Status::OK; + } + + { + std::lock_guard lock(mutex_); + replayed_registration_id_ = registration.registration_id(); } + replayed_.notify_all(); + + while (stream->Read(&event)) { + if (event.has_unregister_local_service()) { + SendStatus(stream, event.unregister_local_service().registration_id(), + pole::sidecar::v1::LOCAL_SERVICE_STATE_UNREGISTERED); + std::lock_guard lock(mutex_); + unregistered_ = true; + replayed_.notify_all(); + } + } + closed_.store(true); + replayed_.notify_all(); return grpc::Status::OK; } + bool WaitForReplay(const std::string& registration_id) { + std::unique_lock lock(mutex_); + return replayed_.wait_for(lock, std::chrono::seconds(3), [&] { + return replayed_registration_id_ == registration_id; + }); + } + + bool WaitForUnregistration() { + std::unique_lock lock(mutex_); + return replayed_.wait_for(lock, std::chrono::seconds(3), [&] { return unregistered_; }); + } + + bool WaitForClose() { + std::unique_lock lock(mutex_); + return replayed_.wait_for(lock, std::chrono::seconds(3), [&] { return closed_.load(); }); + } + private: static void AddListener(pole::sidecar::v1::ListenerSnapshot* snapshot, pole::sidecar::v1::Protocol protocol, std::uint32_t port) { @@ -40,6 +93,23 @@ class BootstrapService final : public pole::sidecar::v1::SidecarSessionService:: listener->set_protocol(protocol); listener->set_port(port); } + + static void SendStatus( + grpc::ServerReaderWriter* stream, + const std::string& registration_id, pole::sidecar::v1::LocalServiceState state) { + pole::sidecar::v1::SidecarEvent status_event; + auto* status = status_event.mutable_local_service_status(); + status->set_registration_id(registration_id); + status->set_state(state); + assert(stream->Write(status_event)); + } + + std::atomic session_count_{0}; + std::atomic closed_{false}; + std::mutex mutex_; + std::condition_variable replayed_; + std::string replayed_registration_id_; + bool unregistered_ = false; }; } // namespace @@ -64,7 +134,25 @@ int main() { assert(session->ListenerAddress(pole::client::Protocol::kDubbo) == "127.0.0.1:21003"); assert(session->ListenerAddress(pole::client::Protocol::kThrift) == "127.0.0.1:21004"); + const auto registration_id = session->RegisterLocalService( + {.service_namespace = "default", + .service = "orders", + .protocol = pole::client::Protocol::kGrpc, + .local_port = 50051}); + assert(service.WaitForReplay(registration_id)); + const auto status_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + while (!session->GetLocalServiceStatus(registration_id).has_value() && + std::chrono::steady_clock::now() < status_deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + assert(session->GetLocalServiceStatus(registration_id).has_value()); + assert(session->GetLocalServiceStatus(registration_id)->state == + pole::client::LocalServiceState::kRegistered); + + assert(session->UnregisterLocalService(registration_id)); + assert(service.WaitForUnregistration()); session->Close(); + assert(service.WaitForClose()); server->Shutdown(); std::filesystem::remove(socket); } From 3923878e5d7579ac112a8597978ea21b94aa2126 Mon Sep 17 00:00:00 2001 From: liaochuntao Date: Thu, 6 Aug 2026 02:37:14 +0800 Subject: [PATCH 2/3] fix: use generated C++ namespace setter --- context-kg/tasks/lessons.md | 3 +++ src/sidecar_session.cc | 4 ++-- 2 files changed, 5 insertions(+), 2 deletions(-) create mode 100644 context-kg/tasks/lessons.md diff --git a/context-kg/tasks/lessons.md b/context-kg/tasks/lessons.md new file mode 100644 index 0000000..1cf7d40 --- /dev/null +++ b/context-kg/tasks/lessons.md @@ -0,0 +1,3 @@ +# 经验记录 + +- Protobuf 字段名 `namespace` 是 C++ 关键字，官方生成器会把访问器命名为 `namespace_()` / `set_namespace_()`；不能按其他语言习惯写成 `set_namespace()`。 diff --git a/src/sidecar_session.cc b/src/sidecar_session.cc index 265472e..e3ebedd 100644 --- a/src/sidecar_session.cc +++ b/src/sidecar_session.cc @@ -249,7 +249,7 @@ void SidecarSession::SendRegistration(const LocalServiceRegistration& registrati ::pole::sidecar::v1::ClientEvent event; auto* wire_registration = event.mutable_register_local_service(); wire_registration->set_registration_id(registration.registration_id); - wire_registration->set_namespace(registration.service_namespace); + wire_registration->set_namespace_(registration.service_namespace); wire_registration->set_service(registration.service); wire_registration->set_protocol(ToWireProtocol(registration.protocol)); wire_registration->set_local_port(registration.local_port); @@ -326,7 +326,7 @@ void SidecarSession::Run() { ::pole::sidecar::v1::ClientEvent event; auto* wire_registration = event.mutable_register_local_service(); wire_registration->set_registration_id(registration.registration_id); - wire_registration->set_namespace(registration.service_namespace); + wire_registration->set_namespace_(registration.service_namespace); wire_registration->set_service(registration.service); wire_registration->set_protocol(ToWireProtocol(registration.protocol)); wire_registration->set_local_port(registration.local_port); From 8797b9a98ac28ce5b1dd18e42f26732d6c20ff59 Mon Sep 17 00:00:00 2001 From: liaochuntao Date: Thu, 6 Aug 2026 02:39:58 +0800 Subject: [PATCH 3/3] test: keep C++ session assertions active in release --- context-kg/tasks/lessons.md | 1 + tests/sidecar_session_test.cc | 3 +++ 2 files changed, 4 insertions(+) diff --git a/context-kg/tasks/lessons.md b/context-kg/tasks/lessons.md index 1cf7d40..2724276 100644 --- a/context-kg/tasks/lessons.md +++ b/context-kg/tasks/lessons.md @@ -1,3 +1,4 @@ # 经验记录 - Protobuf 字段名 `namespace` 是 C++ 关键字，官方生成器会把访问器命名为 `namespace_()` / `set_namespace_()`；不能按其他语言习惯写成 `set_namespace()`。 +- C++ Release 构建会通过 `NDEBUG` 把 `assert(expression)` 连同 expression 本身删除；集成测试不能把 `Read`、`Write` 等有副作用操作只放在 assert 参数里，至少要确保测试翻译单元启用断言或改用始终求值的检查函数。 diff --git a/tests/sidecar_session_test.cc b/tests/sidecar_session_test.cc index 908682b..47e8de3 100644 --- a/tests/sidecar_session_test.cc +++ b/tests/sidecar_session_test.cc @@ -1,6 +1,9 @@ #include "pole/client/sidecar_session.h" #include +#ifdef NDEBUG +#undef NDEBUG +#endif #include #include #include