From 020d80be8e018de51335fb64b353df834a65ed24 Mon Sep 17 00:00:00 2001 From: nick Date: Tue, 27 Oct 2026 09:59:14 +0800 Subject: [PATCH] feat(net)!: bootstrap static Bus service connections Queue each configured same-channel, same-zone target through the existing SS client service during Net startup. Retain registration, heartbeat and retry behavior, and reject ambiguous duplicate configuration or unusable selected target endpoints. Add actual two-process SS roundtrip/restart acceptance and synchronized Bus/Net guides. BREAKING CHANGE: Bus now rejects duplicate application names/types, directed relations and process IDs. Net automatically connects relation-selected processes in the local channel and zone and rejects selected targets without a dialable TCP/UDP intranet endpoint. Remove duplicates and unwanted outgoing relations, and configure reachable target endpoints before upgrading. --- docs/reference/bus.md | 26 +- docs/reference/bus_CN.md | 22 +- docs/reference/network.md | 5 +- docs/reference/network_CN.md | 4 +- src/plugins/bus/bus_configuration.hpp | 24 +- .../net/net_service_manager_module_impl.cpp | 42 +++ tests/app/bus_plugin.cpp | 5 +- tests/app/static_services.cpp | 352 ++++++++++++++++++ tests/core/CMakeLists.txt | 10 + tests/core/bus_configuration.cpp | 58 ++- 10 files changed, 490 insertions(+), 58 deletions(-) create mode 100644 tests/app/static_services.cpp diff --git a/docs/reference/bus.md b/docs/reference/bus.md index a5cf5e163..50868e1b0 100644 --- a/docs/reference/bus.md +++ b/docs/reference/bus.md @@ -38,9 +38,9 @@ config = "bus.plugin.toml" ``` The application must also load Net and Utility as required by the existing module -dependencies. This snippet does not establish cross-plugin initialization order -or start a listener. The TOML schema, required fields, defaults of public value -types, validation diagnostics and message serialization/routing are unchanged. +dependencies. Net queues configured outgoing SS connections during `post_init()`; +the application still starts its listeners explicitly. The public value types and +message serialization interfaces remain unchanged. `check-bus-plugin`, included in application-enabled `check-core`, loads actual Bus/Net/Utility DSOs and verifies both public interface keys, TOML queries, @@ -113,7 +113,11 @@ check_interval = "10s" check_timeout = "3s" ``` -This is a Bus configuration example, not a complete two-process launch configuration. Applications must also load Bus, Net and their dependencies, set each process's Bus ID, start listeners and initiate connections. `relations` supplies target types; reading a relationship does not automatically create a connection. +Applications load Bus, Net and Utility, set a deployment-assigned Bus ID, and start listeners with `create_server()`. During Net `post_init()`, each directed relation selects configured target processes of that type in the local channel and zone. Self is excluded, and each target Bus ID is queued once through the existing SS client service. A relation with no eligible process records queues nothing. Bus queries retain their existing semantics; `get_other_proc(type, list)` itself does not filter channel or zone. + +The selected target's `intranet_endpoint` must contain a dialable TCP or UDP host, a nonzero port and no path. Missing endpoints, unsupported transports and wildcard dial addresses (`0.0.0.0`, `::`) fail Net startup with the target Bus ID and field in the diagnostic. Hostnames remain supported. Wildcard addresses remain valid for listening: where a wildcard bind is needed, use `server_endpoint` for that listener and a reachable `intranet_endpoint` for SS traffic. `intranet_endpoint` currently serves both as the SS listener address and the advertised dial target; there is no separate bind/advertise field. + +Connection requests run on subsequent manager updates; successful startup does not mean the target is connected. Targets can start later or restart at the same configured identity and endpoint. The existing client state machine, SS registration and heartbeat handle the connection lifecycle. Configuration is a startup snapshot: it does not support hot reload, dynamic membership, automatic ID assignment or global uniqueness enforcement. Deployments, including Kubernetes, must assign stable Bus IDs and reachable endpoints; no Kubernetes API integration is implied. Configuring `bus.registry` does not enable dynamic registration or discovery. The Consul registration and health-based discovery paths in the current network service manager remain commented out. This configuration alone cannot discover target processes. @@ -135,7 +139,9 @@ All four sections and their fields are required. Arrays may be empty (`applicati Bus IDs require exactly four decimal components in `0..255`. Nonempty endpoints require a supported protocol and `protocol://host:port` (optional path), with port `0..65535`. Hostnames are retained without DNS lookup. IPv6 literals use brackets, for example `tcp://[::1]:9011`; their numeric syntax is validated locally, and the stored host is `::1` without brackets for arknet. The public endpoint formatter is unchanged. An empty endpoint remains allowed and disables that listener. Both endpoints are validated independently. Registry interval and timeout strings are passed through without interpreting their duration grammar. -The public `app_config`, `proc_config` and `reg_center` fields and query methods remain unchanged. Application maps and other-process maps keep their first entry for a duplicate key; the last matching self-process entry wins, and relations retain input order and duplicates. A missing self-process entry remains unconfigured. These settings do not enable discovery or create network connections. The private schema validator uses the shared compiled `ark::config` file/TOML reader; toml++ stays private to configuration and plugin implementation targets. +The public `app_config`, `proc_config` and `reg_center` fields and query methods remain unchanged. Duplicate application names, application types, directed relations and process Bus IDs now fail Bus initialization with the file, array index and field. A missing self-process entry remains unconfigured. The private schema validator uses the shared compiled `ark::config` file/TOML reader; toml++ stays private to configuration and plugin implementation targets. + +**Upgrade:** remove duplicate records and provide a reachable `intranet_endpoint` for every same-channel, same-zone process selected by an outgoing relation before starting Net. Configurations that previously loaded duplicate or empty selected-target records now fail startup. Relations now initiate outgoing connections automatically; remove relations that should not initiate traffic. The shipped general-purpose configuration includes historical entries and is not a complete deployment inventory; review its relations and endpoints for the chosen process identity. ## Query local and target processes @@ -185,9 +191,9 @@ std::vector configured_targets(ark::plugin_manager& manager) { | `get_other_proc(id, proc)` | Copies the configuration and returns `true` when found; otherwise returns `false` | | `combine_bus_id(type, instance)` | Combines the local channel and zone with the supplied type and instance | -## Establish a TCP connection +## Explicit TCP connections -The target process first starts its listeners with `net_service_manager_module::create_server()`. The following function reads the target's configured intranet endpoint and queues a TCP connection through the network service manager. +The target process starts its listeners with `net_service_manager_module::create_server()`. Normal relation-selected targets are queued automatically. For application-managed targets outside that bootstrap policy, the following function queues an explicit TCP connection through the existing interface. ```cpp #include "ark/bus/bus_module.hpp" @@ -221,7 +227,7 @@ bool queue_tcp_connection(ark::plugin_manager& manager, ark::bus_id_t target_id) `true` means the connection request was queued in the client service and will be processed by subsequent `update()` calls. It does not mean the connection has succeeded. Observe connection events with the client service's `reg_net_event_callback()` and register message handlers with `reg_msg_callback()`. The receiver registers handlers on the corresponding server service. -The current client service handles reconnection through its connection state machine, with a 30-second retry interval in the implementation. This is not a configurable backoff policy in the XML file. Reconnection does not replay application messages whose earlier send failed. +The current client service handles reconnection through its connection state machine, with a 30-second retry interval in the implementation. This is not a configurable backoff policy in the TOML file. Reconnection does not replay application messages whose earlier send failed. ## Send and receive messages @@ -250,9 +256,9 @@ The receive flow is: the network service extracts a `net_msg` → invokes the ha ## Coverage and planned work -These examples cover configuration queries, an explicit TCP connection request and sending through an existing connection. They are not an accepted multi-process deployment example, and do not imply that every protocol is integrated with the service manager. See the [Net plugin](network.md) for transport integration. +`check-static-services` (Unix, application-enabled `check-core`) launches two separately owned processes with the actual Bus/Net/Utility dynamic libraries on a private loopback TCP endpoint. It verifies type/channel/zone eligibility and self exclusion, selected-target endpoint diagnostics, SS registration/session mapping, Protobuf request/response, failed sends while disconnected, and a target restart followed by another round trip through the existing 30-second service retry policy. Both processes stop normally; the controller retains raw child logs locally. Run it with `cmake --build build --target check-static-services --parallel 1` after configuring application and core tests. -TODO: provide a standalone two-process example and tests covering connection setup, message round trips, process restarts, reconnection and cross-host routing. Implement and validate dynamic registration / discovery separately. +This is bounded local functional acceptance, not cross-host routing, UDP integration, Windows process-controller acceptance, capacity or a deployment guarantee. See the [Net plugin](network.md) for transport integration. TODO: provide a standalone deployment example and validate cross-host routing and dynamic registration/discovery separately. ## API sources diff --git a/docs/reference/bus_CN.md b/docs/reference/bus_CN.md index 3158b4976..b0e9c6a2b 100644 --- a/docs/reference/bus_CN.md +++ b/docs/reference/bus_CN.md @@ -27,8 +27,8 @@ name = "ark_bus_plugin" config = "bus.plugin.toml" ``` -应用还需按原有模块依赖加载 Net 与 Utility。该片段不规定跨插件初始化顺序，也不会启动监听。 -TOML 结构、必需字段、公开值类型的默认值、校验诊断、消息序列化与路由均保持不变。 +应用还需按原有模块依赖加载 Net 与 Utility。Net 在 `post_init()` 中加入配置指定的 SS 出站连接， +应用仍须显式启动监听。公开值类型和消息序列化接口保持不变。 启用应用构建时，`check-core` 包含的 `check-bus-plugin` 加载真实 Bus／Net／Utility 动态库， 验证两个公开接口键、TOML 查询、通过记录型 Net 连接传递的 Protobuf 消息头与正文、发送失败及正常重复停服。 @@ -98,7 +98,11 @@ check_interval = "10s" check_timeout = "3s" ``` -这是 Bus 配置示例，不是完整的双进程启动配置。应用还需加载 Bus、Net 及其依赖插件，设置各自的 Bus ID，启动监听并发起连接。`relations` 只提供目标类型列表；读取关系本身不会自动建立连接。 +应用加载 Bus、Net 和 Utility，设置部署分配的 Bus ID，并通过 `create_server()` 启动监听。Net 在 `post_init()` 中按有向关系选择目标类型，并仅连接与本进程渠道和区域相同的静态进程。排除自身后，每个目标 Bus ID 只加入一次既有 SS 客户端服务。关系没有匹配的进程记录时，不加入连接。Bus 查询语义不变；`get_other_proc(type, list)` 本身不筛选渠道或区域。 + +选中目标的 `intranet_endpoint` 必须包含可拨号的 TCP 或 UDP 主机地址、非零端口，且不能带路径。端点缺失、传输协议不受支持或拨号地址为通配地址（`0.0.0.0`、`::`）时，Net 启动失败，诊断包含目标 Bus ID 和字段。支持主机名。监听仍可使用通配地址：需要通配绑定时，用 `server_endpoint` 监听，并为 SS 流量配置可达的 `intranet_endpoint`。目前 `intranet_endpoint` 同时用于 SS 监听和对外提供的拨号地址，没有独立的绑定地址与通告地址字段。 + +连接请求在后续管理器更新中处理；启动成功不代表目标已经连接。目标可以稍后启动，也可以沿用相同身份和端点重启。连接生命周期继续使用既有客户端状态机、SS 注册和心跳。配置是启动时的快照，不支持热重载、动态成员管理、自动分配 ID 或全局唯一性校验。包括 Kubernetes 在内的部署环境须分配稳定的 Bus ID 和可达端点；这不表示已接入 Kubernetes API。 `bus.registry` 有配置项不等于已启用动态注册或发现。当前网络服务管理模块中的 Consul 注册 / 健康检查发现路径仍为注释，不能依靠这份配置自动获得目标进程。 @@ -120,7 +124,9 @@ Bus 运行设置现在要求 TOML 内容，不能只修改旧 XML 文件的扩 Bus ID 必须恰好包含四个 `0..255` 的十进制分量。非空端点必须使用受支持的协议和 `protocol://host:port` 格式（可附加路径），端口为 `0..65535`。主机名原样保留，不进行 DNS 查询。IPv6 字面量必须加方括号，例如 `tcp://[::1]:9011`；解析器仅在本地校验其数值语法，保存供 arknet 使用的无方括号主机名 `::1`。公共端点格式化方法保持不变。端点仍可为空，表示禁用对应监听；两个端点分别校验。注册中心的检查间隔和超时字符串原样传递，不解释时长语法。 -公共 `app_config`、`proc_config`、`reg_center` 字段及查询方法不变。应用映射和其他进程映射中的重复键保留首条，本进程配置取最后一条匹配项；关系保留输入顺序及重复项。本进程未配置时仍保持未配置状态。这些设置不会启用服务发现或创建网络连接。私有结构校验使用 `ark::config` 编译目标提供的共享文件／TOML 读取器；toml++ 仅由配置和插件实现目标内部使用，不进入公开接口。 +公共 `app_config`、`proc_config`、`reg_center` 字段及查询方法不变。应用名称、应用类型、有向关系或进程 Bus ID 重复时，Bus 初始化失败，诊断包含文件、数组下标和字段。本进程未配置时仍保持未配置状态。私有结构校验使用 `ark::config` 编译目标提供的共享文件／TOML 读取器；toml++ 仅由配置和插件实现目标内部使用，不进入公开接口。 + +**升级：**启动 Net 前，删除重复记录，并为出站关系选中的同渠道、同区域进程提供可达的 `intranet_endpoint`。此前可加载的重复记录或选中目标的空端点现在会导致启动失败。关系现在会自动发起出站连接；不应发起流量的关系须删除。随仓库提供的通用配置含历史记录，不是完整的部署清单；应按所选进程身份核对关系及端点。 ## 查询本进程与目标进程 @@ -170,9 +176,9 @@ std::vector configured_targets(ark::plugin_manager& manager) { | `get_other_proc(id, proc)` | 找到时复制配置并返回 `true`；未找到返回 `false` | | `combine_bus_id(type, instance)` | 沿用本进程的渠道和区域，组合指定类型与实例的 ID | -## 建立 TCP 连接 +## 显式 TCP 连接 -目标进程应先通过 `net_service_manager_module::create_server()` 启动监听。下面从静态配置取目标的内网端点，并将 TCP 连接请求交给网络服务管理模块。 +目标进程通过 `net_service_manager_module::create_server()` 启动监听。关系选中的常规目标会自动加入连接；对于应用在该启动策略之外自行管理的目标，可通过以下既有接口显式加入 TCP 连接。 ```cpp #include "ark/bus/bus_module.hpp" @@ -235,9 +241,9 @@ bool send_to_process(ark::msg_module& messages, ark::bus_id_t target_id, ark::ms ## 覆盖范围与后续工作 -本页示例覆盖配置查询、显式 TCP 连接请求和已有连接上的消息发送。它不构成经过验收的多进程部署示例，也不保证所有协议在服务管理层均已接入。底层传输能力见 [Net 插件](network_CN.md)。 +`check-static-services`（Unix，启用应用构建时纳入 `check-core`）在私有回环 TCP 端点上启动两个独立持有的进程，加载真实 Bus／Net／Utility 动态库。测试覆盖类型／渠道／区域筛选与自身排除、选中目标的端点诊断、SS 注册与 Session 映射、Protobuf 请求响应、断连期间发送失败，以及目标重启后按既有 30 秒服务重试策略完成第二次往返。两个进程正常停服，控制进程将原始子进程日志保留在本地。启用应用和核心测试后，运行 `cmake --build build --target check-static-services --parallel 1`。 -TODO：提供独立的双进程示例与测试，覆盖连接建立、消息往返、进程重启、重连和跨机器路由；动态注册 / 发现另行实现和验证。 +这是有界的本地功能验收，不构成跨机器路由、UDP 集成、Windows 进程控制器、容量或部署保证。底层传输能力见 [Net 插件](network_CN.md)。TODO：提供独立部署示例；跨机器路由、动态注册／发现另行实现和验证。 ## API 来源 diff --git a/docs/reference/network.md b/docs/reference/network.md index 6f4324a7a..51c7219ec 100644 --- a/docs/reference/network.md +++ b/docs/reference/network.md @@ -75,7 +75,10 @@ bool connect_tcp(ark::plugin_manager& manager, const ark::bus_addr& target, Connection requests retain their existing asynchronous behavior. Continue calling `manager.update(timestamp)` on the owner thread; keep callback receivers alive -until delivery stops. See [Bus](bus.md) for identity and endpoint configuration. +until delivery stops. Net also queues outgoing SS connections from Bus relations +at startup, restricted to configured target types in the same channel and zone. +Applications still start listeners explicitly. See [Bus](bus.md) for endpoint +validation, migration requirements and the bounded two-process restart test. `check-net-plugin`, included in application-enabled `check-core`, loads actual Bus/Utility/Net DSOs, checks both interface keys, TCP message and HTTP callbacks diff --git a/docs/reference/network_CN.md b/docs/reference/network_CN.md index 696049bfe..8479600da 100644 --- a/docs/reference/network_CN.md +++ b/docs/reference/network_CN.md @@ -63,7 +63,9 @@ bool connect_tcp(ark::plugin_manager& manager, const ark::bus_addr& target, ``` 连接请求保持既有异步语义。继续在所属线程调用 `manager.update(timestamp)`， -并在回调停止前保持接收者有效。进程身份与端点配置见 [Bus](bus_CN.md)。 +并在回调停止前保持接收者有效。Net 也会在启动时根据 Bus 关系加入 SS 出站连接， +仅选择同渠道、同区域的指定目标类型。应用仍须显式启动监听。 +端点校验、迁移要求和有界双进程重启测试见 [Bus](bus_CN.md)。 启用应用构建时，`check-core` 包含的 `check-net-plugin` 加载真实 Bus／Utility／Net 动态库， 验证两个接口键、管理器线程上的 TCP 消息与 HTTP 回调、无需管理器更新即可完成的 future、 diff --git a/src/plugins/bus/bus_configuration.hpp b/src/plugins/bus/bus_configuration.hpp index 095c0000e..5293aef5e 100644 --- a/src/plugins/bus/bus_configuration.hpp +++ b/src/plugins/bus/bus_configuration.hpp @@ -1,10 +1,12 @@ // Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. #pragma once +#include #include #include #include #include +#include #include @@ -115,8 +117,14 @@ inline app_config load(const std::string& path, bus_id_t self_bus_id) { check_keys(app, location, "application", {"name", "type"}); const auto name = string(app, location, "name"); const auto type = static_cast(integer(app, location, "type", 1, 254)); - config.name2types.insert({name, type}); - config.type2names.insert({type, name}); + if (!config.name2types.insert({name, type}).second) + { + fail(location, "name", "duplicate application name: " + name); + } + if (!config.type2names.insert({type, name}).second) + { + fail(location, "type", "duplicate application type"); + } } index = 0; for (const auto& node : array(table, path, "relations")) @@ -130,7 +138,12 @@ inline app_config load(const std::string& path, bus_id_t self_bus_id) { { fail(location, "source/target", "expected configured application names"); } - config.connection_relations[source->second].push_back(target->second); + auto& targets = config.connection_relations[source->second]; + if (std::find(targets.begin(), targets.end(), target->second) != targets.end()) + { + fail(location, "source/target", "duplicate directed relation"); + } + targets.push_back(target->second); } const auto* registry = table["registry"].as_table(); if (!registry) @@ -146,6 +159,7 @@ inline app_config load(const std::string& path, bus_id_t self_bus_id) { config.reg_center.check_interval = string(*registry, registry_path, "check_interval"); config.reg_center.check_timeout = string(*registry, registry_path, "check_timeout"); index = 0; + std::set process_ids; for (const auto& node : array(table, path, "processes")) { const auto location = path + ": bus.processes[" + std::to_string(index++) + "]"; @@ -154,6 +168,10 @@ inline app_config load(const std::string& path, bus_id_t self_bus_id) { {"bus_id", "server_endpoint", "intranet_endpoint", "max_connections", "threads"}); proc_config proc; proc.bus_id = bus_id(string(process, location, "bus_id"), location); + if (!process_ids.insert(proc.bus_id).second) + { + fail(location, "bus_id", "duplicate process Bus ID"); + } proc.max_connection = integer(process, location, "max_connections", 0, std::numeric_limits::max()); proc.thread_num = static_cast(integer(process, location, "threads", 0, 255)); diff --git a/src/plugins/net/net_service_manager_module_impl.cpp b/src/plugins/net/net_service_manager_module_impl.cpp index bef484c6e..a46aa1007 100644 --- a/src/plugins/net/net_service_manager_module_impl.cpp +++ b/src/plugins/net/net_service_manager_module_impl.cpp @@ -18,6 +18,9 @@ * */ +#include +#include + #include "ark/bus/bus_module.hpp" #include "ark/container/map.hpp" #include "ark/module_base.hpp" @@ -69,6 +72,7 @@ class net_service_manager_module_impl final : public module_base get_target_client_connection(const bus_addr& bus_addr); @@ -105,12 +109,50 @@ bool net_service_manager_module_impl::init() { } bool net_service_manager_module_impl::post_init() { + start_configured_clients(); // start health check timer timer_module_->add_forever_timer(0, std::chrono::seconds(20), this, &net_service_manager_module_impl::health_check); return true; } +void net_service_manager_module_impl::start_configured_clients() { + const bus_addr self(bus_module_->get_self_bus_id()); + std::vector types; + bus_module_->get_target_bus_relations(types); + std::set queued; + for (const auto type : types) + { + std::vector processes; + bus_module_->get_other_proc(type, processes); + for (const auto* process : processes) + { + bus_addr target(process->bus_id); + if (target.bus_id == self.bus_id || target.channel_id != self.channel_id || + target.zone_id != self.zone_id || !queued.insert(target.bus_id).second) + { + continue; + } + const auto& ep = process->intranet_ep; + arknet::error_code address_error; + const auto address = asio::ip::make_address(ep.get_ip(), address_error); + if ((ep.proto() != proto_type::tcp && ep.proto() != proto_type::udp) || ep.get_ip().empty() || + ep.get_ip() == "*" || (!address_error && address.is_unspecified()) || ep.get_port() == 0 || + !ep.get_path().empty()) + { + throw std::runtime_error( + "static Bus target " + target.to_string() + + ": intranet_endpoint requires a dialable TCP/UDP host and nonzero port without a path; " + "wildcard bind addresses cannot be advertised"); + } + if (!create_client_service(target, ep.proto(), ep.get_ip(), ep.get_port())) + { + throw std::runtime_error("could not queue static Bus target " + target.to_string()); + } + } + } +} + void net_service_manager_module_impl::health_check(uint64_t timer_id, const guid_t& entity_id) { // TODO: health checking, now using etcd just need to watch. diff --git a/tests/app/bus_plugin.cpp b/tests/app/bus_plugin.cpp index 85718d35b..9f1a3fc28 100644 --- a/tests/app/bus_plugin.cpp +++ b/tests/app/bus_plugin.cpp @@ -77,7 +77,7 @@ intranet_endpoint = '' max_connections = 10 threads = 1 [[bus.processes]] -bus_id = '1.1.30.2' +bus_id = '1.2.30.2' server_endpoint = '' intranet_endpoint = '' max_connections = 10 @@ -95,7 +95,8 @@ check_timeout = '3s' log.set_log_func([](spdlog::level::level_enum, const std::string& text) { std::cout << text << std::endl; }); ark::plugin_manager manager; const ark::bus_addr self(1, 1, 31, 1); - const ark::bus_addr target(1, 1, 30, 2); + // Keep the serialization seam independent of automatic same-zone dialing. + const ark::bus_addr target(1, 2, 30, 2); manager.set_bus_id(self.bus_id); const bool started = manager.start(0, std::string(argv[1]) + "/", "", {{"ark_bus_plugin", "bus.toml"}, {"ark_net_plugin", ""}, {"ark_utility_plugin", ""}}); diff --git a/tests/app/static_services.cpp b/tests/app/static_services.cpp new file mode 100644 index 000000000..11a81eaa4 --- /dev/null +++ b/tests/app/static_services.cpp @@ -0,0 +1,352 @@ +// Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include + +namespace { +using namespace std::chrono_literals; +const ark::bus_addr client_id(1, 1, 31, 1); +const ark::bus_addr server_id(1, 1, 30, 1); +volatile std::sig_atomic_t stopping = 0; + +void on_signal(int) { + stopping = 1; +} + +void require(bool condition, const std::string& message) { + if (!condition) + { + throw std::runtime_error(message); + } +} + +void mark(const std::string& name) { + std::ofstream output(name); + output.exceptions(std::ios::failbit | std::ios::badbit); + output << "ready\n"; +} + +void write_configuration(const std::string& endpoint) { + std::ofstream output("bus.toml"); + output.exceptions(std::ios::failbit | std::ios::badbit); + output << "[bus]\n" + "[[bus.applications]]\nname = 'game'\ntype = 31\n" + "[[bus.applications]]\nname = 'world'\ntype = 30\n" + "[[bus.applications]]\nname = 'login'\ntype = 32\n" + "[[bus.relations]]\nsource = 'game'\ntarget = 'world'\n" + "[[bus.relations]]\nsource = 'game'\ntarget = 'game'\n"; + for (const auto& id : {"1.1.31.1", "1.1.30.1", "1.2.30.1", "2.1.30.1", "1.1.32.1"}) + { + output << "[[bus.processes]]\nbus_id = '" << id << "'\nserver_endpoint = ''\nintranet_endpoint = '" + << (std::string(id) == "1.1.30.1" ? endpoint : "") << "'\nthreads = 1\nmax_connections = 10\n"; + } + output << "[bus.registry]\nhost = '127.0.0.1'\nport = 0\nservice_name = 'static-test'\n" + "check_interval = '10s'\ncheck_timeout = '3s'\n"; +} + +struct receiver +{ + ark::plugin_manager& manager; + ark::net_service_manager_module& network; + ark::msg_module& messages; + int generation; + const std::thread::id owner = std::this_thread::get_id(); + int replies{}; + int connects{}; + int disconnects{}; + + void request(const ark::net_msg* message, ark::conv_id_t session) { + require(std::this_thread::get_id() == owner, "SS request callback left manager owner thread"); + require(message->get_src_bus() == client_id.bus_id && message->get_dst_bus() == server_id.bus_id && + message->get_actor_id() == ark::guid_t{42}, + "SS request addressing changed"); + require(session != 0 && network.get_session_id(client_id.bus_id) == session && + network.get_net_connection_bus(server_id.bus_id, client_id.bus_id) != nullptr, + "SS registration did not establish server route and session before request"); + ark::protobuf::point_3d body; + require(body.ParseFromArray(message->get_msg_data(), static_cast(message->get_msg_length())) && + body.y() == 12.5F && body.z() == -7.0F, + "SS request Protobuf payload changed"); + body.set_x(static_cast(generation)); + require(messages.send_to_intra(session, 45011, body, ark::guid_t{42}), "SS response rejected"); + ++replies; + } + + void response(const ark::net_msg* message, ark::conv_id_t) { + require(std::this_thread::get_id() == owner, "SS response callback left manager owner thread"); + require(message->get_src_bus() == server_id.bus_id && message->get_actor_id() == ark::guid_t{42}, + "SS response addressing changed"); + ark::protobuf::point_3d body; + require(body.ParseFromArray(message->get_msg_data(), static_cast(message->get_msg_length())) && + body.y() == 12.5F && body.z() == -7.0F, + "SS response Protobuf payload changed"); + if (body.x() == 1.0F) + { + if (replies == 0) + { + mark("first-roundtrip"); + } + replies = 1; + } + else + { + require(body.x() == 2.0F && replies == 1 && disconnects == 1 && connects == 2, + "Restart response arrived without exactly one disconnect and reconnect"); + replies = 2; + mark("second-roundtrip"); + } + } + + void event(const ark::net_event* event) { + require(std::this_thread::get_id() == owner, "SS connection event left manager owner thread"); + require(event->get_bus_id() == server_id.bus_id, "Unexpected static target connection event"); + if (event->get_type() == ark::net_event_type::connect) + { + ++connects; + } + else if (event->get_type() == ark::net_event_type::disconnect) + { + ++disconnects; + ark::protobuf::point_3d body; + require(replies == 1 && !network.get_net_connection_bus(client_id.bus_id, server_id.bus_id) && + !messages.send_msg_by_bus_id(server_id.bus_id, 45010, body), + "Disconnected target retained a usable route"); + mark("disconnected"); + } + } +}; + +int peer(const std::string& role, const std::string& plugins, int generation) { + std::signal(SIGTERM, on_signal); + ark::logger log; + ark::logger::init(&log); + log.set_log_func([](spdlog::level::level_enum, const std::string& text) { std::cout << text << std::endl; }); + ark::plugin_manager manager; + manager.set_bus_id(role == "server" ? server_id.bus_id : client_id.bus_id); + bool started = false; + std::string diagnostic; + try + { + started = manager.start(ark::date_time::get_now_time(), plugins + "/", "", + {{"ark_bus_plugin", "bus.toml"}, {"ark_net_plugin", ""}, {"ark_utility_plugin", ""}}); + } + catch (const std::exception& error) + { + if (role != "invalid") + { + throw; + } + diagnostic = error.what(); + } + if (role == "invalid") + { + require(!started && manager.get_startup_failure().find("post_init:") != std::string::npos && + diagnostic.find("static Bus target 1.1.30.1") != std::string::npos && + diagnostic.find("intranet_endpoint") != std::string::npos, + "Invalid static target lacked a specific startup diagnostic: " + diagnostic); + require(manager.stop(ark::date_time::get_now_time()), "Invalid static target cleanup failed"); + return 0; + } + require(started, "Actual plugin startup failed: " + manager.get_startup_failure()); + auto* network = manager.find_module(); + auto* messages = manager.find_module(); + require(network && messages, "Actual Bus/Net interfaces missing"); + receiver received{manager, *network, *messages, generation}; + auto clients = network->get_client_service(ark::app_type::ark_app_world); + if (role == "server") + { + require(!clients, "Undeclared outgoing relation created a server client"); + require(network->create_server(), "Actual SS listener failed to start"); + auto server = network->get_intra_net_server(); + require(server && !network->get_inter_net_server(), "Static target listener selection changed"); + require(server->reg_msg_callback(45010, &received, &receiver::request), "SS request callback rejected"); + mark("server-ready-" + std::to_string(generation)); + } + else + { + require(clients && !network->get_client_service(ark::app_type::ark_app_game) && + !network->get_client_service(ark::app_type::ark_app_login), + "Static bootstrap did not exclude self or undeclared type"); + require(clients->reg_msg_callback(45011, &received, &receiver::response) && + clients->reg_net_event_callback(&received, &receiver::event), + "SS client callbacks rejected"); + } + const auto started_at = std::chrono::steady_clock::now(); + auto send_at = started_at; + // Existing reconnect interval is 30s. Allow 5s for the initial and final + // round trips; do not change or accelerate the production retry clock. + const auto deadline = started_at + 40s; + while (!stopping && std::chrono::steady_clock::now() < deadline && (role == "server" || received.replies != 2)) + { + require(manager.update(ark::date_time::get_now_time()), "Actual plugin update failed"); + if (role == "client") + { + require(!clients->get_connection_info(ark::bus_addr(1, 2, 30, 1).bus_id) && + !clients->get_connection_info(ark::bus_addr(2, 1, 30, 1).bus_id), + "Bootstrap crossed configured channel/zone boundary"); + const auto now = std::chrono::steady_clock::now(); + if (now >= send_at && network->get_net_connection_bus(client_id.bus_id, server_id.bus_id)) + { + ark::protobuf::point_3d body; + body.set_y(12.5F); + body.set_z(-7.0F); + require(messages->send_msg_by_bus_id(server_id.bus_id, 45010, body, ark::guid_t{42}), + "Connected static Bus request rejected"); + send_at = now + 100ms; + } + } + std::this_thread::sleep_for(1ms); + } + require(role == "server" ? stopping != 0 : received.replies == 2, "Bounded SS peer acceptance timed out"); + // Drop public shared services while their DSO implementations remain loaded. + clients.reset(); + require(manager.stop(ark::date_time::get_now_time()) && manager.stop(ark::date_time::get_now_time()), + "Actual plugin normal repeated stop failed"); + return 0; +} + +class child +{ +public: + child(const std::string& executable, const std::string& role, const std::string& plugins, int generation, + const std::string& log_name) { + const auto generation_text = std::to_string(generation); + pid_ = ::fork(); + require(pid_ >= 0, "fork failed"); + if (pid_ == 0) + { + const auto log = ::open(log_name.c_str(), O_CREAT | O_WRONLY | O_TRUNC, 0600); + if (log < 0 || ::dup2(log, STDOUT_FILENO) < 0 || ::dup2(log, STDERR_FILENO) < 0) + { + ::_exit(126); + } + ::close(log); + ::execl(executable.c_str(), executable.c_str(), role.c_str(), plugins.c_str(), generation_text.c_str(), + nullptr); + ::_exit(127); + } + } + + child(const child&) = delete; + child& operator=(const child&) = delete; + + ~child() { + if (pid_ > 0) + { + ::kill(pid_, SIGKILL); + ::waitpid(pid_, &status_, 0); + } + } + + bool exited() { + if (pid_ > 0) + { + const auto result = ::waitpid(pid_, &status_, WNOHANG); + require(result >= 0, "waitpid failed"); + if (result == pid_) + { + pid_ = -1; + } + } + return pid_ < 0; + } + + void stop() { + require(pid_ > 0 && ::kill(pid_, SIGTERM) == 0, "Could not stop owned server child"); + finish(); + } + + void finish() { + const auto deadline = std::chrono::steady_clock::now() + 5s; + while (!exited() && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(5ms); + } + require(exited() && WIFEXITED(status_) && WEXITSTATUS(status_) == 0, "Owned child did not exit normally"); + } + +private: + pid_t pid_{-1}; + int status_{}; +}; + +void await_marker(const std::string& name, child& process, std::chrono::seconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (!std::filesystem::exists(name) && std::chrono::steady_clock::now() < deadline) + { + require(!process.exited(), "Owned child exited before " + name); + std::this_thread::sleep_for(5ms); + } + require(std::filesystem::exists(name), "Timed out waiting for " + name); +} + +int controller(const std::string& executable, const std::string& plugins, const std::string& runtime) { + const auto directory = std::filesystem::path(runtime) / + (std::to_string(::getpid()) + "-" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + std::filesystem::create_directories(directory); + std::filesystem::current_path(directory); + std::cout << "Static Bus raw child logs: " << directory << std::endl; + int invalid_case = 0; + for (const auto& endpoint : {"", "tcp://0.0.0.0:1234", "tcp://[::]:1234", "tcp://[0:0:0:0:0:0:0:0]:1234", + "tcp://127.0.0.1:0", "ws://127.0.0.1:1234", "tcp://127.0.0.1:1234/path"}) + { + write_configuration(endpoint); + child invalid(executable, "invalid", plugins, 0, "invalid-" + std::to_string(++invalid_case) + ".log"); + invalid.finish(); + } + asio::io_context context; + asio::ip::tcp::acceptor reservation(context, {asio::ip::make_address("127.0.0.1"), 0}); + const auto port = reservation.local_endpoint().port(); + reservation.close(); + write_configuration("tcp://127.0.0.1:" + std::to_string(port)); + child first(executable, "server", plugins, 1, "server-1.log"); + await_marker("server-ready-1", first, 5s); + child client(executable, "client", plugins, 0, "client.log"); + await_marker("first-roundtrip", client, 5s); + first.stop(); + await_marker("disconnected", client, 5s); + child second(executable, "server", plugins, 2, "server-2.log"); + await_marker("server-ready-2", second, 5s); + await_marker("second-roundtrip", client, 35s); + client.finish(); + second.stop(); + std::cout << "Static Bus actual two-process DSOs: eligibility, invalid endpoint diagnostics, SS registration, " + "Protobuf request/response, lost-route rejection, target restart, 30s policy reconnect and normal " + "stop passed\n"; + return 0; +} +} // namespace + +int main(int argc, char* argv[]) { + try + { + if (argc == 3) + { + return controller( + std::filesystem::absolute(argv[0]).string(), std::filesystem::absolute(argv[1]).string(), argv[2]); + } + require(argc == 4, "Expected plugin directory and runtime, or peer arguments"); + return peer(argv[1], argv[2], std::stoi(argv[3])); + } + catch (const std::exception& error) + { + std::cerr << error.what() << std::endl; + return 1; + } +} diff --git a/tests/core/CMakeLists.txt b/tests/core/CMakeLists.txt index 34a18fb5d..c98dd81df 100644 --- a/tests/core/CMakeLists.txt +++ b/tests/core/CMakeLists.txt @@ -324,6 +324,16 @@ if(ARK_BUILD_PROTO AND TARGET ark_bus_plugin AND TARGET ark_net_plugin AND TARGE DEPENDS ark_bus_plugin_tests ark_bus_plugin ark_net_plugin ark_utility_plugin USES_TERMINAL VERBATIM) add_dependencies(check-core check-bus-plugin) + if(UNIX) + add_executable(ark_static_services_tests ../app/static_services.cpp) + set_property(TARGET ark_static_services_tests PROPERTY LINK_LIBRARIES ark::bus) + add_custom_target(check-static-services + COMMAND $ "$" + "${CMAKE_CURRENT_BINARY_DIR}/static services runtime" + DEPENDS ark_static_services_tests ark_bus_plugin ark_net_plugin ark_utility_plugin + USES_TERMINAL VERBATIM) + add_dependencies(check-core check-static-services) + endif() endif() if(TARGET ark_http_plugin) diff --git a/tests/core/bus_configuration.cpp b/tests/core/bus_configuration.cpp index 35d72eb74..63bb93c80 100644 --- a/tests/core/bus_configuration.cpp +++ b/tests/core/bus_configuration.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include "../network/doctest.h" @@ -230,40 +231,31 @@ TEST_CASE("Shipped Bus TOML retains the original configuration queries") { CHECK(config.reg_center.ip == "127.0.0.1"); } -TEST_CASE("Bus duplicate keys retain existing insertion and self selection order") { +TEST_CASE("Bus rejects duplicate application names types relations and process IDs with field diagnostics") { configuration_file file; - auto text = replace(configuration, "[[bus.relations]]", R"([[bus.applications]] -name = 'world' -type = 32 -[[bus.applications]] -name = 'alias' -type = 30 -[[bus.relations]])"); - text = replace(text, "[bus.registry]", R"([[bus.relations]] -source = 'game' -target = 'world' -[[bus.processes]] -bus_id = '1.1.30.1' -server_endpoint = '' -intranet_endpoint = '' -max_connections = 7 -threads = 7 -[[bus.processes]] -bus_id = '1.1.31.1' -server_endpoint = '' -intranet_endpoint = '' -max_connections = 8 -threads = 8 -[bus.registry])"); - const auto config = ark::bus_config_detail::load(file.write(text), ark::bus_addr(1, 1, 31, 1).bus_id); - CHECK(config.name2types.at("world") == ark::app_type::ark_app_world); - CHECK(config.type2names.at(ark::app_type::ark_app_world) == "world"); - REQUIRE(config.connection_relations.at(ark::app_type::ark_app_game).size() == 2); - CHECK(config.connection_relations.at(ark::app_type::ark_app_game)[0] == ark::app_type::ark_app_world); - CHECK(config.connection_relations.at(ark::app_type::ark_app_game)[1] == ark::app_type::ark_app_world); - CHECK(config.other_proc_list.at(ark::bus_addr(1, 1, 30, 1).bus_id).max_connection == 5000); - CHECK(config.self_proc.max_connection == 8); - CHECK(config.self_proc.thread_num == 8); + for (const auto& [from, to, field] : + std::vector>{{"name = 'game'", "name = 'world'", "name"}, + {"type = 31", "type = 30", "type"}, + {"[bus.registry]", "[[bus.relations]]\nsource = 'game'\ntarget = 'world'\n[bus.registry]", "source/target"}, + {"'1.1.31.1'", "'1.1.30.1'", "bus_id"}}) + { + for (const auto self : {ark::bus_addr(1, 1, 30, 1).bus_id, ark::bus_addr(1, 1, 31, 1).bus_id}) + { + const auto path = file.write(replace(configuration, from, to)); + try + { + static_cast(ark::bus_config_detail::load(path, self)); + FAIL("duplicate configuration accepted"); + } + catch (const std::runtime_error& error) + { + const std::string message = error.what(); + CHECK(message.find(path) != std::string::npos); + CHECK(message.find("duplicate") != std::string::npos); + CHECK(message.find(field) != std::string::npos); + } + } + } } TEST_CASE("Bus module retains bracketed IPv6 endpoint hosts without brackets for arknet") {