From 7d6ba6de9331274c4b27706fa942674d16b0e368 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 13:42:24 +0800 Subject: [PATCH] feat(gateway): add retry_on_refused for pymsg MsgClient (v0.6.1) - Add MsgConnectionRefusedException to distinguish ECONNREFUSED from other connection errors; openConnection() now throws it when connect() returns ECONNREFUSED. - MsgClient::call() adds retry_on_refused/retry_interval parameters (defaults false/1.0s), matching Python pymsg.MsgClient.call(retry_on_refused=True). - Retry loop: only retries on ECONNREFUSED; other exceptions (timeout, connection reset, business errors) propagate immediately. Throttled logging: first failure prints immediately, subsequent logs every 5 seconds. - DbC::call_msg() enables retry_on_refused=true (db is a hard dependency; gateway blocks waiting for db on startup instead of crashing). - LogC::_send() explicitly passes retry_on_refused=false (fail-safe; never block the background log thread when db is unavailable). - captcha/smtp auxiliary services keep default retry_on_refused=false (fail-fast; blocking in request threads would exhaust the thread pool). - Update README with pymsg protocol documentation, downstream retry strategy table, and v0.6.1 changelog entry. --- gateway/README.md | 31 ++++++++++++++- gateway/inc/DbC.hpp | 26 +++++++------ gateway/inc/LogC.hpp | 3 +- gateway/inc/Msg.hpp | 91 ++++++++++++++++++++++++++++++++++++-------- gateway/src/main.cpp | 2 +- 5 files changed, 121 insertions(+), 32 deletions(-) diff --git a/gateway/README.md b/gateway/README.md index e3236db..df0ecef 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -28,6 +28,29 @@ UniServe Framework 的应用层网关，C++20 实现。 | CAPTCHA | 一次性、5 分钟过期，登录与注册均可选启用 | | 速率限制 | 内存计数器，自动清理超过 1 万条的老条目 | | TLS 可选 | `Server.tls = off` 时纯 HTTP（配合 WAF），`on` 时用 `full_chain.crt` / `server.key` | +| 下游服务容错 | pymsg 客户端 `retry_on_refused` 机制：db 等强依赖服务未启动时自动阻塞重试，避免启动顺序问题；日志客户端 fail-safe 静默丢弃 | + +## pymsg 内部通信协议 + +gateway 通过项目内的 [pymsg](../pymsg/)（长度前缀帧 TCP 协议）与内部微服务通信，C++ 实现在 `inc/Msg.hpp`，与 Python 端字节级兼容。 + +**线协议**：`[4字节大端uint32长度][UTF-8载荷]`，最大 64 MiB。 + +**JSON RPC 约定**：请求形如 `{"_route": "<路由名>", ...}`，响应形如 `{"success": true, "data": ...}` 或 `{"success": false, "error": "..."}`。 + +**下游服务调用方式**： + +| 下游 | 端口 | `_route` | 调用位置 | 重试策略 | +|------|------|----------|---------|---------| +| db | 8005 | `execute` | `DbC::call_msg()`（所有数据库 CRUD） | `retry_on_refused=true`，db 未就绪时阻塞等待 | +| db | 8005 | `execute` | `LogC::_send()`（日志异步批量推送） | `retry_on_refused=false`，fail-safe 静默丢弃 | +| captcha | 8009 | `verify` / `image` | `verify_captcha_remote()`、`/captcha/image` 代理 | 默认不重试，连接失败 fail-closed 返回 502/false | +| smtp | 8008 | `verify_email` / `send_email` | `verify_email_code_remote()`、`/register/send_code` | 默认不重试，连接失败 fail-closed 返回 502/false | + +**retry_on_refused 机制**（`MsgClient::call()`，对应 Python 端 `MsgClient.call(retry_on_refused=True)`）： +- 仅对 `ECONNREFUSED`（目标端口无服务监听）重试，超时、连接重置、业务错误立即抛出 +- 固定 1 秒间隔重试；首次失败立即打印 stderr 提示，之后每 5 秒打印一次等待日志（节制日志，避免刷屏） +- 用于解决微服务启动顺序不确定问题：gateway 启动时 db 未就绪，首次 db 调用自动阻塞等待直到 db 上线 ## 接口 @@ -265,6 +288,9 @@ gateway/ ├── src/ │ └── main.cpp # 全部实现：PBKDF2 / PasswordValidator / RateLimiter / 路由 / handler ├── inc/ +│ ├── Msg.hpp # pymsg C++ 实现：长度前缀帧协议、MsgClient（含 retry_on_refused 重试）、MsgException 族 +│ ├── DbC.hpp # db 服务客户端：通过 pymsg 调 dbv2 execute 路由（retry_on_refused=true） +│ ├── LogC.hpp # 统一日志客户端：异步批量队列 + pymsg INSERT（fail-safe，不重试） │ ├── Conf.hpp # ConfIni：基于 boost::property_tree 的 ini 解析 │ ├── ExpBase.hpp # 异常基类（带 source_location） │ └── LocalTime.hpp # 本地时间工具（getCurrent / parse） @@ -276,14 +302,15 @@ gateway/ ## 版本 -`0.6.0`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印： +`0.6.1`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印： ``` -Gateway 0.6.0 started +Gateway 0.6.1 started ``` 近期变更： +- `0.6.1`：pymsg 客户端增加 `retry_on_refused` 重试机制，与 Python 端 `pymsg.MsgClient.call(retry_on_refused=True)` 行为对齐——新增 `MsgConnectionRefusedException` 异常类（区分 ECONNREFUSED 与其他连接错误）；`MsgClient::call()` 新增 `retry_on_refused`/`retry_interval` 参数，仅对连接拒绝重试，固定间隔 1s，节制日志（首次立即打印 + 每 5s 一次），其他异常（超时/连接重置/业务错误）立即抛出；`DbC::call_msg()` 启用 `retry_on_refused=true`（db 是硬依赖，启动顺序不确定时自动阻塞等待）；`LogC::_send()` 显式传 `retry_on_refused=false`（日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞后台线程）；captcha/smtp 辅助服务保持默认不重试（fail-fast，请求线程中永久阻塞会耗尽线程池）。 - `0.6.0`：WebSocket 反向代理（仅管理员）——新增 `[WsProxy]` 配置段，支持 `routes = /path=ws://host:port/target;...` 精确路径映射；握手阶段校验 Token 中 `is_admin=true`（基于 `Admin.users` 白名单），通过后双向透明桥接文本/二进制帧到后端，支持 `ws://` 和 `wss://`；后端连接错误/鉴权失败均优雅关闭。依赖 libhttp v3.1+ 的 `ws_proxy()` 接口。顺带修复 `LocalTime.hpp` 在 Apple clang 下 C++20 tzdb 不可用的编译问题（改用 `strftime`/`get_time`/`mktime`）。 - `0.5.4`：修复邮件轰炸漏洞（审计报告 P0-1）——`/register/send_code` 强制 CAPTCHA 校验（经 captcha:8009，Token 一次性消费），新增邮箱级双窗口限流（3 次/小时 + 10 次/天）；smtp 层同步增加邮箱级长期限流兜底 + 全局熔断（`SMTP_HOURLY_LIMIT` 环境变量，默认 100 封/小时）；前端 `register.html`/`register.js` 同步接入图形验证码（复用 `/captcha/image` 代理路由，方案 A：仅在「获取验证码」时要求 CAPTCHA，注册提交不加）。 - `0.5.3`：配合 smtp 同邮箱 60 秒发送冷却——`/register/send_code` 在 smtp 返回 `retry_after` 时按 429 透传其提示文案（其余发送失败仍用通用 500 文案，SMTP 内部错误细节不向外暴露）。 diff --git a/gateway/inc/DbC.hpp b/gateway/inc/DbC.hpp index 00ded14..dbbb559 100644 --- a/gateway/inc/DbC.hpp +++ b/gateway/inc/DbC.hpp @@ -25,18 +25,20 @@ namespace qing { } // 新版 pymsg 接口：通过 MsgClient（TCP 长度前缀帧协议）调用 dbv2。 - // 自动添加 _route = "execute"，返回 dbv2 pymsg 完整 JSON 响应 - // {"success":..., "data":[...]}。 - // 响应格式与 call_http 兼容，调用方可直接替换。 - static json::value call_msg(const json::value& params, - double timeout = DEFAULT_TIMEOUT) { - json::value jv_route = params; - jv_route.as_object()["_route"] = "execute"; - pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); - std::string resp = client.call( - json::serialize(jv_route), timeout); - return json::parse(resp); - } + // 自动添加 _route = "execute"，返回 dbv2 pymsg 完整 JSON 响应 + // {"success":..., "data":[...]}。 + // 响应格式与 call_http 兼容，调用方可直接替换。 + // db 是强依赖，启用 retry_on_refused：db 未启动时阻塞等待，避免启动顺序问题。 + static json::value call_msg(const json::value& params, + double timeout = DEFAULT_TIMEOUT) { + json::value jv_route = params; + jv_route.as_object()["_route"] = "execute"; + pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); + std::string resp = client.call( + json::serialize(jv_route), timeout, + /*retry_on_refused=*/true); + return json::parse(resp); + } private: static constexpr const char* _DBURL = "http://127.0.0.1:8005/execute"; diff --git a/gateway/inc/LogC.hpp b/gateway/inc/LogC.hpp index a685d9e..66eaa88 100644 --- a/gateway/inc/LogC.hpp +++ b/gateway/inc/LogC.hpp @@ -196,7 +196,8 @@ private: payload["values"] = std::move(values); const std::string resp = client.call( - json::serialize(payload), _TIMEOUT); + json::serialize(payload), _TIMEOUT, + /*retry_on_refused=*/false); // 日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞重试 auto obj = json::parse(resp).as_object(); if (!obj.at("success").as_bool()) { std::string err = "unknown"; diff --git a/gateway/inc/Msg.hpp b/gateway/inc/Msg.hpp index fd7a956..90c1151 100644 --- a/gateway/inc/Msg.hpp +++ b/gateway/inc/Msg.hpp @@ -17,11 +17,14 @@ */ #include +#include #include +#include #include #include #include #include +#include #include #include @@ -59,6 +62,13 @@ public: explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {} }; +/// 目标服务未启动（connect 返回 ECONNREFUSED），用于 retry_on_refused 重试机制 +class MsgConnectionRefusedException : public MsgConnectionException { +public: + explicit MsgConnectionRefusedException(const std::string &msg) + : MsgConnectionException(msg) {} +}; + // ============================================================================ // Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload // ============================================================================ @@ -155,10 +165,14 @@ inline int openConnection(const std::string& addr, int port, double timeout) { if (::connect(fd, reinterpret_cast(&serv_addr), sizeof(serv_addr)) < 0) { + int saved_errno = errno; ::close(fd); - throw MsgConnectionException( - "could not connect to " + addr + ":" + std::to_string(port) + - " - " + ::strerror(errno)); + std::string err_msg = "could not connect to " + addr + ":" + + std::to_string(port) + " - " + ::strerror(saved_errno); + if (saved_errno == ECONNREFUSED) { + throw MsgConnectionRefusedException(err_msg); + } + throw MsgConnectionException(err_msg); } return fd; @@ -242,19 +256,64 @@ public: /// /// 不会给消息添加 ``name:`` 前缀（适用于 RPC/API 调用）。 /// - /// @param msg 要发送的原始字符串。 - /// @param timeout 超时秒数，默认 10.0。 - /// @returns 服务器返回的字符串。 - std::string call(const std::string &msg, double timeout = 10.0) { - int fd = detail::openConnection(_addr, _port, timeout); - try { - pmsg_send(fd, msg); - std::string resp = pmsg_recv(fd); - ::close(fd); - return resp; - } catch (...) { - ::close(fd); - throw; + /// @param msg 要发送的原始字符串。 + /// @param timeout 超时秒数，默认 10.0。 + /// @param retry_on_refused 目标服务未启动（ECONNREFUSED）时是否阻塞重试， + /// 直到连接成功。适用于启动顺序不确定的服务发现场景。 + /// 默认 false（连接拒绝立即抛异常）。 + /// @param retry_interval 重试间隔秒数，默认 1.0。 + /// @returns 服务器返回的字符串。 + std::string call(const std::string &msg, double timeout = 10.0, + bool retry_on_refused = false, + double retry_interval = 1.0) { + if (!retry_on_refused) { + // 快速路径：不重试，直接执行一次，保持原有行为 + int fd = detail::openConnection(_addr, _port, timeout); + try { + pmsg_send(fd, msg); + std::string resp = pmsg_recv(fd); + ::close(fd); + return resp; + } catch (...) { + ::close(fd); + throw; + } + } + + // 重试路径：仅对 ECONNREFUSED 重试，其他错误立即抛出 + constexpr double kLogEvery = 5.0; // 重试期间打印等待提示的间隔（秒） + int attempt = 0; + auto last_log = std::chrono::steady_clock::now() - + std::chrono::duration_cast( + std::chrono::duration(kLogEvery)); + + while (true) { + try { + int fd = detail::openConnection(_addr, _port, timeout); + try { + pmsg_send(fd, msg); + std::string resp = pmsg_recv(fd); + ::close(fd); + return resp; + } catch (...) { + ::close(fd); + throw; + } + } catch (const MsgConnectionRefusedException &) { + ++attempt; + auto now = std::chrono::steady_clock::now(); + double elapsed = std::chrono::duration(now - last_log).count(); + if (elapsed >= kLogEvery || attempt == 1) { + std::fprintf(stderr, + "[pymsg] 等待服务 %s:%d 就绪... (已重试 %d 次)\n", + _addr.c_str(), _port, attempt); + std::fflush(stderr); + last_log = now; + } + std::this_thread::sleep_for( + std::chrono::duration(retry_interval)); + } + // 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出 } } diff --git a/gateway/src/main.cpp b/gateway/src/main.cpp index 2437b17..dec0a30 100644 --- a/gateway/src/main.cpp +++ b/gateway/src/main.cpp @@ -28,7 +28,7 @@ using namespace qing; namespace json = boost::json; namespace po = boost::program_options; -constexpr const char *_VER = "0.6.0"; +constexpr const char *_VER = "0.6.1"; ////////////////////////////////// #include