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.
This commit is contained in:
qing committed
1 parent
b30ce1424b
commit
7d6ba6de93
5 files changed
+121
-32
No files matched your search
+29
-2
@@ -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 内部错误细节不向外暴露）。
|
||||
|
||||
+14
-12
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
+75
-16
@@ -17,11 +17,14 @@
|
||||
*/
|
||||
|
||||
#include <cerrno>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
@@ -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<struct sockaddr *>(&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::steady_clock::duration>(
|
||||
std::chrono::duration<double>(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<double>(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<double>(retry_interval));
|
||||
}
|
||||
// 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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 <sstream>
|
||||
|
||||
Reference in new issue
Block a user