diff --git a/README.en.md b/README.en.md index 6c4f775..06759eb 100644 --- a/README.en.md +++ b/README.en.md @@ -1,4 +1,4 @@ -# EverywhereYouGo (EGo) v1.2.4 +# EverywhereYouGo (EGo) v1.3.0 [中文](README.md) | English @@ -53,17 +53,34 @@ Optionally set `EGO_SECRET_KEY` to customize Flask session key. ## Configuration Files -Configuration is persisted as JSON files in `config/` directory: +Configuration lives in two places, with distinct roles: + +| Storage | Role | +|---------|------| +| SQLite (`ego.db`) | **Runtime source of truth** — all reads/writes go through it | +| `config/*.json` | **Export / backup medium** — for backup, versioning and migration | | File | Content | -|------|------| +|------|---------| | `config/parsers.json` | Parser metadata | | `config/sources.json` | Data source definitions | | `config/channels.json` | Push channel configurations | | `config/templates.json` | Push templates | | `config/bindings.json` | Channel bindings (with condition expressions) | -Can directly edit JSON and restart to take effect, or manage via WebUI. System settings (DND, log level, etc.) and runtime data (message logs) are stored in SQLite (`ego.db`). +**Load rules at startup:** + +1. Database **not empty** → the database wins; JSON is not read, and the current + config is **written back** to `config/*.json` as a snapshot +2. Database **empty** and JSON present → import from JSON (first run / migration / restore) +3. Database empty and no JSON → export the initial config to JSON + +So **use the WebUI for day-to-day config changes** (they take effect immediately). +Hand-editing `config/*.json` is only read on first import when the database is empty — +it is not the normal path for applying changes. + +System settings (DND, log level, etc.), the message log and the queue are also stored in SQLite. +For backup/restore use **Settings → Backup**, which packages `config/*.json` + `parsers/*.py`. ## Parsers @@ -101,7 +118,10 @@ Supports `and`, `or`, parentheses grouping: Set DND time period, messages enter queue and wait, automatically flush when period ends. Urgent routes are not affected by DND. ### Message Deduplication -Channel bindings can configure `dedup_key_expr` and `dedup_window` (default 3600 seconds). Same dedup key will not be sent repeatedly within the window. +Deduplication granularity is **message × channel**: each channel binding has its own +`dedup_key_expr` and `dedup_window` (default 3600 seconds), independent of the others. +A channel that hits is skipped while the rest still send; the whole message is marked +`DISCARDED` only when **every** channel hits. ### Parallel Push When multiple channels match, thread pool sends in parallel, total latency depends on the slowest single channel. @@ -110,13 +130,56 @@ When multiple channels match, thread pool sends in parallel, total latency depen Each data source automatically saves the last 20 request samples, can select samples in WebUI for test parsing and pushing. ### Message Resend -Failed messages support original resend (using parsed msg_json) or re-parse and resend. +Failed messages support original resend (using the parsed msg_json) or re-parse and resend. +By default **only the channels that failed are retried** — already-succeeded channels are +not pushed a second time. Use `scope=all` to force a full re-push. ### Import & Export - **Backup**: Download ZIP package (`config/*.json` + `parsers/*.py`) - **Restore**: Upload ZIP package, automatically takes effect after overwriting configuration - **JSON Import**: Supports dry_run preview, insert/overwrite two modes, dependency check +### Channel Circuit Breaker +Automatically isolates a channel that keeps failing, so one broken third party cannot +drag down the whole send path: + +- Failure ratio > 50% within a sliding window (default 60s), **or** 5 consecutive + failures → trip +- Cooldown backs off exponentially: 30 → 60 → 120 → … → 600 s (capped at 10 min) +- After cooldown it enters half-open probing; 3 consecutive successes restore it +- **4xx does not count as failure** (a business rejection is not a service outage) — + only 5xx / timeouts / connection errors are counted +- While tripped, messages **stay queued**: no retry budget is consumed and nothing is + dropped. State is persisted and survives restarts. + +### Outbound Rate Limiting +Per-channel rate limit (messages per minute) to avoid getting blocked by the remote side. +Retries cannot fix a 429 — limiting has to happen **before** sending. A message that +cannot get a token waits in the queue instead of being dropped. + +### Resilience UI +| Where | What you can do | +|-------|-----------------| +| Channel list → **Resilience column** | See the rate-limit badge and breaker countdown; reset a tripped channel with one click | +| Channel edit dialog | Set this channel's outbound rate limit (empty/0 = unlimited) | +| Settings → **Resilience (channel circuit breaker)** | Hand-tune sliding window, consecutive-failure threshold, cooldown base/cap, probe count, etc. | + +Parameter precedence: `system_config` (settings page / direct DB edit) > environment +variable > built-in default. Changes take effect immediately, no restart needed. + +### Observability +| Endpoint | Description | +|----------|-------------| +| `GET /api/metrics` | Queue depth, DLQ count, per-channel success rate, end-to-end latency, breaker & rate-limit state; `?hours=N` sets the stats window (default 24h) | +| `GET /api/resilience` | Channels currently tripped / rate-limited | +| `GET /api/queue/stats` | Queue and dead-letter counts | +| `GET /api/health` | Health check (SQLite / disk / config / queue) | + +### Graceful Stop +On `SIGTERM` EGo stops accepting new messages first, then waits for in-flight tasks +(up to 30 seconds); anything unfinished is moved to the dead-letter queue — a container +restart does not lose messages. + ## Internationalization Built-in Chinese and English bilingual support, switch languages anytime via language switch button in top-right corner of navigation bar. @@ -146,6 +209,18 @@ Built-in Chinese and English bilingual support, switch languages anytime via lan | `LOG_LEVEL` | `INFO` | Log level | | `EGO_AUTH_TOKEN` | *(empty)* | Access control Token | | `EGO_SECRET_KEY` | *(auto)* | Flask session key | +| `EGO_INGRESS_WORKERS` | `8` | Ingress worker threads per port source | +| `EGO_INGRESS_MAX_QUEUE` | `200` | Ingress queue cap; beyond it returns 503 (backpressure) | +| `EGO_CLEANUP_INTERVAL` | `600` | Interval for purging old messages / dedup keys (s) | +| `EGO_BREAKER_WINDOW` | `60` | Circuit breaker sliding window (s) | +| `EGO_BREAKER_MIN_SAMPLES` | `5` | Min samples before the failure-ratio rule applies | +| `EGO_BREAKER_FAILURE_RATIO` | `0.5` | Failure ratio that trips the breaker | +| `EGO_BREAKER_CONSECUTIVE` | `5` | Consecutive-failure threshold (low-traffic channels) | +| `EGO_BREAKER_OPEN_BASE` | `30` | Base cooldown (s), doubles on each open | +| `EGO_BREAKER_OPEN_MAX` | `600` | Cooldown cap (s) | +| `EGO_BREAKER_HALF_OPEN_OK` | `3` | Consecutive probe successes needed to recover | +| `EGO_RATE_MAX_WAIT` | `1.0` | Max wait for a rate-limit token (s), then defer | +| `EGO_RATE_MISS_TTL` | `30` | Re-check interval for channels without a rate limit (s) | ## License diff --git a/README.md b/README.md index fdac8c3..c01382b 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# EverywhereYouGo (EGo) v1.2.4 +# EverywhereYouGo (EGo) v1.3.0 [English](README.en.md) | 中文 @@ -51,9 +51,14 @@ EGO_AUTH_TOKEN=your-secret-token python3 main.py 可选设置 `EGO_SECRET_KEY` 自定义 Flask session 密钥。 -## 配置文件 +## 配置存储 -配置持久化为 JSON 文件，位于 `config/` 目录： +配置有两份，角色不同： + +| 存储 | 角色 | +|------|------| +| SQLite（`ego.db`） | **运行时真相源** —— 所有读写以库内数据为准 | +| `config/*.json` | **导出 / 备份介质** —— 便于备份、版本管理与迁移 | | 文件 | 内容 | |------|------| @@ -63,7 +68,17 @@ EGO_AUTH_TOKEN=your-secret-token python3 main.py | `config/templates.json` | 推送模板 | | `config/bindings.json` | 渠道绑定（含条件表达式） | -可直接编辑 JSON 后重启生效，也可通过 WebUI 管理。系统设置（DND、日志级别等）和运行时数据（消息日志）存储在 SQLite（`ego.db`）中。 +**启动时的加载规则：** + +1. 数据库**已有**配置 → 以数据库为准，不读 JSON，并把当前配置**刷写**回 `config/*.json` +2. 数据库**为空**且有 JSON → 从 JSON 导入（首次启动 / 迁移 / 恢复） +3. 数据库为空且无 JSON → 导出初始配置到 JSON + +因此 **日常改配置请用 WebUI**（改完即时生效）。直接编辑 `config/*.json` 只在 +「数据库为空」的首次导入场景才会被读取，不是常规生效路径。 + +系统设置（DND、日志级别等）、消息日志与队列同样存储在 SQLite。 +配置备份 / 恢复请用「系统设置 → 备份」，会打包 `config/*.json` + `parsers/*.py`。 ## 解析器 @@ -101,7 +116,9 @@ def parse(raw_body: bytes, headers: dict, query_params: dict) -> dict: 设置免打扰时段后，消息进入队列等待，结束后自动刷新。紧急路由不受 DND 影响。 ### 消息去重 -渠道绑定可配置 `dedup_key_expr` 和 `dedup_window`（默认 3600 秒）。同一去重键在窗口内不重复发送。 +去重粒度是 **消息 × 通道**：每个渠道绑定各自配置 `dedup_key_expr` 与 `dedup_window` +（默认 3600 秒），互不影响。命中的渠道被跳过，其余渠道照常发送； +只有**全部**渠道命中时，整条消息才标记为 `DISCARDED`。 ### 并行推送 多渠道匹配时线程池并行发送，总延迟取决于最慢的单个渠道。 @@ -111,12 +128,49 @@ def parse(raw_body: bytes, headers: dict, query_params: dict) -> dict: ### 消息重发 失败消息支持原始重发（使用已解析的 msg_json）或重新解析后重发。 +默认**只重发上次失败的渠道**，不会把已经成功的渠道重复推送一遍； +需要整体重推时可用 `scope=all`。 ### 导入导出 - **备份**：下载 ZIP 包（`config/*.json` + `parsers/*.py`） - **恢复**：上传 ZIP 包，覆盖配置后自动生效 - **JSON 导入**：支持 dry_run 预览、insert/overwrite 两种模式、依赖检查 +### 通道熔断 +第三方渠道持续故障时自动隔离，避免拖垮整条发送链路： + +- 滑动窗口（默认 60s）内失败率 > 50%，**或**连续失败 ≥ 5 次 → 熔断 +- 冷却时间指数退避 30 → 60 → 120 → … → 600 秒（封顶 10 分钟） +- 冷却结束后进入半开探测，连续 3 次成功才恢复 +- **4xx 不计失败**（业务侧拒绝不等于服务故障），只对 5xx / 超时 / 连接类错误计数 +- 熔断期间消息**留在队列等待**：不消耗重试次数、不丢弃；状态持久化，重启后仍生效 + +### 出站限流 +按通道独立限流（条/分钟），防止发送过快被对方封禁。重试解决不了 429—— +限流必须发生在发送**之前**。拿不到令牌的消息会排队等待，而不是被丢弃。 + +### 韧性界面 +| 位置 | 能做什么 | +|------|---------| +| 通道列表 → **「韧性」列** | 查看限流徽章与熔断倒计时；熔断时可一键「手工恢复」 | +| 通道编辑弹窗 | 设置该通道的出站限流（留空/0 = 不限流） | +| 系统设置 → **韧性（通道熔断）** | 手工调整滑动窗口、连续失败阈值、冷却基数/上限、探测次数等参数 | + +参数取值优先级：`system_config`（设置页 / 直接改库） > 环境变量 > 内置默认， +改完即时生效、无需重启。 + +### 可观测性 +| 端点 | 说明 | +|------|------| +| `GET /api/metrics` | 队列深度、死信总数、各通道成功率、端到端延迟、熔断与限流状态；`?hours=N` 调整统计窗口（默认 24h） | +| `GET /api/resilience` | 当前处于熔断 / 限流状态的通道 | +| `GET /api/queue/stats` | 队列与死信计数 | +| `GET /api/health` | 健康检查（SQLite / 磁盘 / 配置 / 队列） | + +### 优雅停机 +收到 `SIGTERM` 时先停止接收新消息，再等待在途任务完成（最长 30 秒）， +超时未完成的转入死信队列——容器重启不会丢消息。 + ## 国际化 内置中英文双语支持，通过导航栏右上角语言切换按钮随时切换。 @@ -146,6 +200,18 @@ def parse(raw_body: bytes, headers: dict, query_params: dict) -> dict: | `LOG_LEVEL` | `INFO` | 日志等级 | | `EGO_AUTH_TOKEN` | *(空)* | 访问控制 Token | | `EGO_SECRET_KEY` | *(自动)* | Flask session 密钥 | +| `EGO_INGRESS_WORKERS` | `8` | 每个端口数据源的入口工作线程数 | +| `EGO_INGRESS_MAX_QUEUE` | `200` | 入口等待队列上限，超出返回 503（背压） | +| `EGO_CLEANUP_INTERVAL` | `600` | 旧消息 / 去重键的清理间隔（秒） | +| `EGO_BREAKER_WINDOW` | `60` | 熔断滑动窗口（秒） | +| `EGO_BREAKER_MIN_SAMPLES` | `5` | 窗口内触发失败率判定的最少样本数 | +| `EGO_BREAKER_FAILURE_RATIO` | `0.5` | 窗口失败率阈值（超过则熔断） | +| `EGO_BREAKER_CONSECUTIVE` | `5` | 连续失败阈值（照顾低频通道） | +| `EGO_BREAKER_OPEN_BASE` | `30` | 熔断冷却基数（秒），逐次翻倍 | +| `EGO_BREAKER_OPEN_MAX` | `600` | 熔断冷却上限（秒） | +| `EGO_BREAKER_HALF_OPEN_OK` | `3` | 恢复所需连续探测成功次数 | +| `EGO_RATE_MAX_WAIT` | `1.0` | 限流取令牌的最长等待（秒），超时改为延迟重排 | +| `EGO_RATE_MISS_TTL` | `30` | 未配置限流的通道，回查数据库的间隔（秒） | ## License diff --git a/api/__init__.py b/api/__init__.py index 7543d3e..6d25f4a 100644 --- a/api/__init__.py +++ b/api/__init__.py @@ -136,4 +136,12 @@ def _auth_middleware(): app.source_mgr = source_mgr app.auth_token = AUTH_TOKEN + # ── 统一的入参校验错误处理（improvement #27）── + # 业务代码里直接 raise ValidationError 即可，无需每处 try/except + from api.validation import ValidationError + + @app.errorhandler(ValidationError) + def _handle_validation_error(e): + return jsonify({"status": "error", "error": str(e)}), 400 + return app diff --git a/api/channels.py b/api/channels.py index c858544..e1c9302 100644 --- a/api/channels.py +++ b/api/channels.py @@ -3,10 +3,13 @@ """api/channels.py — 通道 CRUD + 插件管理""" import os +import json +import log import db import channel_loader import i18n from flask import Blueprint, request, jsonify +from api.validation import require_name, optional_flag, ValidationError channels_bp = Blueprint("channels", __name__) @@ -18,10 +21,27 @@ def api_channels(): return jsonify(db.get_channels()) +def _channel_config(data): + """通道 config 必须是 JSON 对象（允许传对象或 JSON 字符串）。""" + cfg = (data or {}).get("config", {}) + if isinstance(cfg, str): + try: + cfg = json.loads(cfg or "{}") + except Exception: + raise ValidationError("config must be a JSON object") + if not isinstance(cfg, dict): + raise ValidationError("config must be a JSON object") + return cfg + + @channels_bp.route("/api/channels", methods=["POST"]) def api_create_channel(): - data = request.json - cid = db.create_channel(data["name"], data["type"], data.get("config", {})) + data = request.json or {} + cid = db.create_channel( + require_name(data), + require_name(data, key="type", max_len=64), + _channel_config(data), + ) import config_manager config_manager.sync_table("channels") return jsonify({"id": cid}) @@ -29,11 +49,70 @@ def api_create_channel(): @channels_bp.route("/api/channels/", methods=["PUT"]) def api_update_channel(cid): - data = request.json - db.update_channel(cid, **data) + data = request.json or {} + patch = {} + if "name" in data: + patch["name"] = require_name(data) + if "type" in data: + patch["type"] = require_name(data, key="type", max_len=64) + if "config" in data: + patch["config"] = _channel_config(data) + if "enabled" in data: + patch["enabled"] = optional_flag(data, "enabled") + if patch: + db.update_channel(cid, **patch) + return jsonify({"status": "ok"}) + + +@channels_bp.route("/api/channels/", methods=["DELETE"]) +def api_delete_channel(cid): + """删除通道（级联清理绑定、限流、熔断、去重键；待发任务移入死信）。""" + if not db.get_channel(cid): + return jsonify({"error": i18n._("err.not_found")}), 404 + db.delete_channel(cid) + import config_manager + config_manager.sync_table("channels") return jsonify({"status": "ok"}) +@channels_bp.route("/api/channels//test", methods=["POST"]) +def api_test_channel(cid): + """测试通道配置是否可用。前端期望 {ok: bool, error?}。""" + ch = db.get_channel(cid) + if not ch: + return jsonify({"ok": False, "error": i18n._("err.not_found")}), 404 + try: + cfg = _channel_config({"config": ch["config"]}) + ok = channel_loader.create_channel(ch["type"], cfg).test() + return jsonify({"ok": bool(ok), "error": None if ok else i18n._("ch.test_fail")}) + except Exception as e: + return jsonify({"ok": False, "error": str(e)[:300]}) + + +@channels_bp.route("/api/channels//duplicate", methods=["POST"]) +def api_duplicate_channel(cid): + """复制通道（默认禁用，避免复制出来就开始推送）。""" + src = db.get_channel(cid) + if not src: + return jsonify({"error": i18n._("err.not_found")}), 404 + + new_id = db.create_channel(src["name"] + " (copy)", src["type"], + src["config"], enabled=0) + + # 出站限流存在独立表里，不显式复制就会静默丢失 + try: + from rate_limiter import get_limiter + rate = get_limiter().get_rate(cid) + if rate > 0: + get_limiter().set_rate(new_id, rate) + except Exception as e: + log.logger.warning(f"Duplicate channel {cid}: rate limit not copied: {e}") + + import config_manager + config_manager.sync_table("channels") + return jsonify({"id": new_id}) + + # ── Channel Plugins ── diff --git a/api/messages.py b/api/messages.py index 342ea9a..8dde362 100644 --- a/api/messages.py +++ b/api/messages.py @@ -6,9 +6,14 @@ import db import i18n from flask import Blueprint, request, jsonify +from api.validation import ValidationError, optional_enum, optional_id_list messages_bp = Blueprint("messages", __name__) +RETRY_MODES = ("original", "rerender") +RETRY_SCOPES = ("failed", "all") +BATCH_ACTIONS = ("retry", "ignore", "delete") + @messages_bp.route("/api/messages", methods=["GET"]) def api_get_messages(): @@ -34,8 +39,12 @@ def api_delete_message(msg_id): @messages_bp.route("/api/messages//retry", methods=["POST"]) def api_retry_message(msg_id): import source_manager as sm - mode = request.args.get("mode", "original") - ok, err = sm.retry_message(msg_id, mode) + # mode: 用原解析结果 / 重新解析；scope: 只重发失败渠道 / 整体重推 + mode = optional_enum({"mode": request.args.get("mode", "original")}, + "mode", RETRY_MODES, default="original") + scope = optional_enum({"scope": request.args.get("scope", "failed")}, + "scope", RETRY_SCOPES, default="failed") + ok, err = sm.retry_message(msg_id, mode, scope) return jsonify({"ok": ok, "error": err} if err else {"ok": ok}) @@ -48,17 +57,20 @@ def api_ignore_message(msg_id): @messages_bp.route("/api/messages/batch", methods=["POST"]) def api_batch_messages(): import source_manager as sm - data = request.json - action = data.get("action") - ids = data.get("ids", []) - mode = data.get("mode", "original") + data = request.json or {} + action = optional_enum(data, "action", BATCH_ACTIONS) + if action is None: + raise ValidationError("action is required (retry/ignore/delete)") + ids = optional_id_list(data) + mode = optional_enum(data, "mode", RETRY_MODES, default="original") + scope = optional_enum(data, "scope", RETRY_SCOPES, default="failed") if not ids: return jsonify({"ok": False, "error": i18n._("err.no_ids")}) results = {"ok": 0, "fail": 0, "errors": []} for mid in ids: if action == "retry": - ok, err = sm.retry_message(mid, mode) + ok, err = sm.retry_message(mid, mode, scope) if ok: results["ok"] += 1 else: diff --git a/api/pages.py b/api/pages.py index c4be9bf..ec6bf24 100644 --- a/api/pages.py +++ b/api/pages.py @@ -10,7 +10,7 @@ pages_bp = Blueprint("pages", __name__) PARSERS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "parsers") -VERSION = "1.2.4" +VERSION = "1.3.0" def _render(page, title, active_page="", **kwargs): @@ -129,6 +129,7 @@ def messages_page(): @pages_bp.route("/settings") def settings_page(): + import circuit_breaker config = { "log_level": db.get_log_level(), "dnd_enabled": db.get_config("dnd_enabled", "0"), @@ -137,7 +138,31 @@ def settings_page(): "cleanup": db.get_cleanup_config(), "path_prefix": db.get_config("path_prefix", "in"), } + # 熔断参数：库里没存则留空，页面上以"默认值"占位（留空即用默认） + breaker = {} + for short, key in circuit_breaker.CONFIG_KEYS.items(): + breaker[short] = db.get_config(key, "") + + def _fmt(v): + """整数就按整数显示，避免占位符出现 60.0 / 30.0 这种别扭写法。""" + try: + return int(v) if float(v) == int(v) else v + except (TypeError, ValueError): + return v + translated_statuses = {s: i18n._(f"status.{s}") for s in db.MESSAGE_STATUSES} return _render("settings.html", i18n._("set.title"), "settings", config=config, + breaker=breaker, + breaker_defaults={ + k: _fmt(v) for k, v in { + "window": circuit_breaker.WINDOW_SECONDS, + "min_samples": circuit_breaker.MIN_SAMPLES, + "failure_ratio": circuit_breaker.FAILURE_RATIO, + "consecutive": circuit_breaker.CONSECUTIVE_THRESHOLD, + "open_base": circuit_breaker.OPEN_BASE_SECONDS, + "open_max": circuit_breaker.OPEN_MAX_SECONDS, + "half_open_ok": circuit_breaker.HALF_OPEN_NEEDED, + }.items() + }, message_statuses=translated_statuses) diff --git a/api/sources.py b/api/sources.py index 0575737..19834e1 100644 --- a/api/sources.py +++ b/api/sources.py @@ -8,6 +8,10 @@ import parser_loader import i18n from flask import Blueprint, request, jsonify, current_app +from api.validation import ( + require_name, optional_str, optional_int, optional_port, + optional_flag, optional_slug, +) sources_bp = Blueprint("sources", __name__) @@ -35,21 +39,22 @@ def api_sources(): @sources_bp.route("/api/sources", methods=["POST"]) def api_create_source(): - data = request.json + data = request.json or {} + port = optional_port(data) sid = db.create_source( - name=data["name"], - port=data.get("port"), - parser_id=data.get("parser_id"), - enabled=data.get("enabled", 1), - slug=data.get("slug"), - parent_id=data.get("parent_id"), - path=data.get("path", ""), + name=require_name(data), + port=port, + parser_id=optional_int(data, "parser_id", 1, default=None), + enabled=optional_flag(data, "enabled", default=1), + slug=optional_slug(data), + parent_id=optional_int(data, "parent_id", 1, default=None), + path=optional_str(data, "path", max_len=200, default="") or "", ) if sid is None: return jsonify({"error": i18n._("err.port_in_use")}), 400 # Only start listener for port-mode sources (no parent, has port) sm = current_app.source_mgr - if sm and data.get("port") and not data.get("parent_id"): + if sm and port and not data.get("parent_id"): sm.start_source(sid) import config_manager config_manager.sync_table("sources") @@ -58,16 +63,33 @@ def api_create_source(): @sources_bp.route("/api/sources/", methods=["PUT"]) def api_update_source(sid): - data = request.json + data = request.json or {} old = db.get_source(sid) if not old: return jsonify({"error": i18n._("err.not_found")}), 404 + + # 只接受白名单字段，且逐字段校验（原实现把任意值直接塞进 DB） + patch = {} + if "name" in data: + patch["name"] = require_name(data) + if "port" in data: + patch["port"] = optional_port(data) + if "parser_id" in data: + patch["parser_id"] = optional_int(data, "parser_id", 1, default=None) + if "enabled" in data: + patch["enabled"] = optional_flag(data, "enabled") + if "slug" in data: + patch["slug"] = optional_slug(data) + if "parent_id" in data: + patch["parent_id"] = optional_int(data, "parent_id", 1, default=None) + if "path" in data: + patch["path"] = optional_str(data, "path", max_len=200, default="") or "" + sm = current_app.source_mgr # Only stop/start listener for port-mode sources if sm and old.get("port") and not old.get("parent_id"): sm.stop_source(sid) - db.update_source(sid, **{k: v for k, v in data.items() - if k in ("name", "port", "parser_id", "enabled", "slug", "parent_id", "path")}) + db.update_source(sid, **patch) if sm and old.get("port") and not old.get("parent_id") and data.get("enabled", old["enabled"]): sm.start_source(sid) import config_manager diff --git a/api/system.py b/api/system.py index 5232ee1..83904e7 100644 --- a/api/system.py +++ b/api/system.py @@ -3,14 +3,21 @@ """api/system.py — 健康检查/设置/版本/队列""" import os +import re import shutil import db import version_checker from queue_backend import get_backend from flask import Blueprint, request, jsonify +from api.validation import ( + ValidationError, optional_enum, optional_flag, optional_hhmm, optional_int, + LOG_LEVELS, +) system_bp = Blueprint("system", __name__) +_PREFIX_RE = re.compile(r"^[A-Za-z0-9_\-/]*$") + CONFIG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config") DB_PATH = os.getenv("DB_PATH", "ego.db") @@ -63,12 +70,59 @@ def api_health(): @system_bp.route("/api/settings", methods=["POST"]) def api_update_settings(): - data = request.json + data = request.json or {} + if not isinstance(data, dict): + raise ValidationError("request body must be a JSON object") + + # 已知键做类型/取值校验（improvement #27），避免把无效值写进 system_config + if "log_level" in data: + data["log_level"] = optional_enum(data, "log_level", LOG_LEVELS, default="INFO") + for k in ("dnd_start", "dnd_end"): + if k in data: + optional_hhmm(data, k) # 值不合法会抛 400 + if "dnd_enabled" in data: + data["dnd_enabled"] = str(optional_flag(data, "dnd_enabled", default=0)) + if "path_prefix" in data: + p = str(data.get("path_prefix") or "").strip().strip("/") + if len(p) > 64 or not _PREFIX_RE.match(p): + raise ValidationError("path_prefix may only contain [A-Za-z0-9_-/] (max 64)") + data["path_prefix"] = p + + # 熔断参数（留空 = 不覆盖，回退到环境变量/默认值） + _BREAKER_INT_KEYS = ( + ("breaker_window", 1, 86400), + ("breaker_min_samples", 1, 100000), + ("breaker_consecutive", 1, 100000), + ("breaker_open_base", 1, 86400), + ("breaker_open_max", 1, 86400), + ("breaker_half_open_ok", 1, 1000), + ) + for key, lo, hi in _BREAKER_INT_KEYS: + if key in data and str(data.get(key) or "").strip() != "": + data[key] = str(optional_int(data, key, lo, hi)) + if "breaker_failure_ratio" in data: + raw = str(data.get("breaker_failure_ratio") or "").strip() + if raw != "": + try: + ratio = float(raw) + except ValueError: + raise ValidationError("breaker_failure_ratio must be a number in (0, 1]") + if not (0 < ratio <= 1): + raise ValidationError("breaker_failure_ratio must be in (0, 1]") + data["breaker_failure_ratio"] = str(ratio) + for k, v in data.items(): if k == "log_level": db.set_log_level(v) else: db.set_config(k, v) + + # 熔断参数带 TTL 缓存，改完立刻失效，避免要等缓存过期才生效 + try: + from circuit_breaker import invalidate_param_cache + invalidate_param_cache() + except Exception: + pass return jsonify({"status": "ok"}) @@ -116,3 +170,72 @@ def api_dlq_delete(dlq_id): """删除一条死信记录。""" get_backend().delete_dlq(dlq_id) return jsonify({"status": "ok"}) + + +# ── 韧性：熔断 + 出站限流 ── + +@system_bp.route("/api/metrics", methods=["GET"]) +def api_metrics(): + """简单可观测性端点（improvement v1.3.0 可选项）。 + + 返回队列深度 / 死信总数 / 各通道成功率 / 端到端延迟 / 熔断与限流状态。 + 定位是「curl 一查就有」，**不引入 Prometheus**。 + 可用 `?hours=N` 调整统计窗口（默认 24h，上限 30 天）。 + """ + from circuit_breaker import get_breaker + from rate_limiter import get_limiter + try: + from api.pages import VERSION as _ver + except Exception: + _ver = None + + hours = request.args.get("hours", None, type=int) + if hours is None: + hours = 24 + hours = max(1, min(int(hours), 24 * 30)) + + return jsonify({ + "version": _ver, + "window_hours": hours, + "queue": get_backend().get_stats(), + "messages": db.get_queue_stats(), + "channels": db.get_channel_stats(hours), + "latency": db.get_latency_stats(hours), + "breaker": get_breaker().snapshot(), + "rate_limits": get_limiter().snapshot(), + }) + + +@system_bp.route("/api/resilience", methods=["GET"]) +def api_resilience(): + """通道熔断状态 + 出站限流配置。""" + from circuit_breaker import get_breaker + from rate_limiter import get_limiter + return jsonify({ + "breaker": get_breaker().snapshot(), + "rate_limits": get_limiter().snapshot(), + }) + + +@system_bp.route("/api/resilience/breaker//reset", methods=["POST"]) +def api_breaker_reset(channel_id): + """手工恢复被熔断的通道。""" + from circuit_breaker import get_breaker + get_breaker().reset(channel_id) + return jsonify({"status": "ok"}) + + +@system_bp.route("/api/resilience/rate_limit/", methods=["POST"]) +def api_rate_limit_set(channel_id): + """设置通道出站限流（条/分钟），0 = 不限流。""" + from rate_limiter import get_limiter + data = request.get_json(silent=True) or {} + raw = data.get("per_minute", 0) + try: + per_minute = int(raw) + except (TypeError, ValueError): + return jsonify({"status": "error", "error": "per_minute must be an integer"}), 400 + if per_minute < 0: + return jsonify({"status": "error", "error": "per_minute must be >= 0"}), 400 + get_limiter().set_rate(channel_id, per_minute) + return jsonify({"status": "ok", "channel_id": channel_id, "per_minute": per_minute}) diff --git a/api/templates.py b/api/templates.py index 19ead31..46a984f 100644 --- a/api/templates.py +++ b/api/templates.py @@ -6,6 +6,9 @@ import renderer import i18n from flask import Blueprint, request, jsonify +from api.validation import ( + require_name, optional_str, optional_enum, TEMPLATE_ENGINES, +) templates_bp = Blueprint("templates", __name__) @@ -17,18 +20,30 @@ def api_templates(): @templates_bp.route("/api/templates", methods=["POST"]) def api_create_template(): - data = request.json + data = request.json or {} tid = db.create_template( - data["name"], data.get("engine", "jinja2"), - data.get("title_tpl", ""), data.get("content_tpl", "")) + require_name(data), + optional_enum(data, "engine", TEMPLATE_ENGINES, default="jinja2"), + optional_str(data, "title_tpl", max_len=10000, default="") or "", + optional_str(data, "content_tpl", max_len=50000, default="") or "", + ) return jsonify({"id": tid}) @templates_bp.route("/api/templates/", methods=["PUT"]) def api_update_template(tid): - data = request.json - db.update_template(tid, **{k: v for k, v in data.items() - if k in ("name", "engine", "title_tpl", "content_tpl")}) + data = request.json or {} + patch = {} + if "name" in data: + patch["name"] = require_name(data) + if "engine" in data: + patch["engine"] = optional_enum(data, "engine", TEMPLATE_ENGINES, default="jinja2") + if "title_tpl" in data: + patch["title_tpl"] = optional_str(data, "title_tpl", max_len=10000, default="") or "" + if "content_tpl" in data: + patch["content_tpl"] = optional_str(data, "content_tpl", max_len=50000, default="") or "" + if patch: + db.update_template(tid, **patch) return jsonify({"status": "ok"}) diff --git a/api/validation.py b/api/validation.py new file mode 100644 index 0000000..4cc5d0b --- /dev/null +++ b/api/validation.py @@ -0,0 +1,140 @@ +#!/usr/bin/python3 +# -*- coding: UTF-8 -*- +"""API 入参校验（improvement #27）。 + +自管理场景下，输入写错的代价主要由 operator 自己承担，所以这里**不引入** +JSON Schema 重型框架，只做一组轻量助手，把「缺字段 / 类型不对 / 取值越界」 +这类会直接变成 500 或写出无效配置的情况，转成明确的 400 + 可读错误。 + +用法： + from api.validation import require_name, optional_port, ValidationError + + name = require_name(data) # 缺了会抛 ValidationError + port = optional_port(data) + +`api/__init__.py` 注册了 ValidationError 的统一错误处理，自动返回 400， +因此业务代码里直接 raise 即可。 +""" + +import re + +_SLUG_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$") + +LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") +TEMPLATE_ENGINES = ("jinja2", "simple") + + +class ValidationError(ValueError): + """入参不合法（API 层统一转 400）。""" + + +def _body(data): + if not isinstance(data, dict): + raise ValidationError("request body must be a JSON object") + return data + + +def require_name(data, key="name", max_len=100): + """必填的非空字符串字段（自动 strip）。""" + v = _body(data).get(key) + if not isinstance(v, str) or not v.strip(): + raise ValidationError(f"{key} is required and must be a non-empty string") + v = v.strip() + if len(v) > max_len: + raise ValidationError(f"{key} too long (max {max_len})") + return v + + +def optional_str(data, key, max_len=500, default=None, allow_empty=True): + v = _body(data).get(key, default) + if v is None: + return default + if not isinstance(v, str): + raise ValidationError(f"{key} must be a string") + if not allow_empty and not v.strip(): + raise ValidationError(f"{key} must not be empty") + if len(v) > max_len: + raise ValidationError(f"{key} too long (max {max_len})") + return v + + +def optional_int(data, key, min_value=None, max_value=None, default=None): + v = _body(data).get(key, default) + if v is None: + return default + if isinstance(v, bool) or not isinstance(v, (int, str)): + raise ValidationError(f"{key} must be an integer") + try: + v = int(v) + except (TypeError, ValueError): + raise ValidationError(f"{key} must be an integer") + if min_value is not None and v < min_value: + raise ValidationError(f"{key} must be >= {min_value}") + if max_value is not None and v > max_value: + raise ValidationError(f"{key} must be <= {max_value}") + return v + + +def optional_port(data, key="port"): + """端口：1-65535，空/None 表示不启用端口模式。""" + return optional_int(data, key, 1, 65535, default=None) + + +def optional_flag(data, key, default=None): + """0/1 标志位。""" + v = _body(data).get(key, default) + if v is None: + return default + if isinstance(v, bool): + return 1 if v else 0 + if v in (0, 1, "0", "1"): + return int(v) + raise ValidationError(f"{key} must be 0 or 1") + + +def optional_slug(data, key="slug"): + """路径路由用的 slug：字母数字下划线连字符。""" + v = _body(data).get(key) + if v in (None, ""): + return None + if not isinstance(v, str) or not _SLUG_RE.match(v): + raise ValidationError(f"{key} must match [A-Za-z0-9_-]{{1,64}}") + return v + + +def optional_enum(data, key, allowed, default=None): + v = _body(data).get(key, default) + if v is None: + return default + if v not in allowed: + raise ValidationError(f"{key} must be one of {sorted(allowed)}") + return v + + +def optional_hhmm(data, key, default=None): + """HH:MM 时刻（用于免打扰时段）。""" + v = _body(data).get(key, default) + if v is None: + return default + if not isinstance(v, str) or not _HHMM_RE.match(v): + raise ValidationError(f"{key} must be HH:MM (00:00-23:59)") + return v + + +def optional_id_list(data, key="ids"): + """整数 ID 列表。""" + v = _body(data).get(key) + if v is None: + return [] + if not isinstance(v, list): + raise ValidationError(f"{key} must be a list") + out = [] + for i in v: + if isinstance(i, bool) or not isinstance(i, (int, str)): + raise ValidationError(f"{key} must contain only integers") + try: + out.append(int(i)) + except (TypeError, ValueError): + raise ValidationError(f"{key} must contain only integers") + return out diff --git a/build.py b/build.py index 4d09d8a..12ea615 100644 --- a/build.py +++ b/build.py @@ -5,7 +5,7 @@ 用法: python3 build.py # 构建镜像 python3 build.py --push # 构建并推送 - python3 build.py --tag v1.2.4 # 指定版本标签 + python3 build.py --tag v1.3.0 # 指定版本标签 """ import os @@ -13,7 +13,7 @@ import subprocess import argparse -VERSION = "1.2.4" +VERSION = "1.3.0" IMAGE_NAME = "codenametest/everywhereyougo" DOCKERFILE = "Dockerfile" diff --git a/bus.py b/bus.py index 1760696..106f1fc 100644 --- a/bus.py +++ b/bus.py @@ -3,6 +3,22 @@ """ EGo 事件总线 — 基于 blinker 的同步信号系统。 所有模块通过总线通信，降低耦合度。 + +消息生命周期（单向、按序触发）： + + message.received source_manager.process_message 发出，进入全链路 + ↓ + message.parsed parser_engine 解析成功 + ↓ + message.routed router_engine 路由匹配完成（携带 matched_channels） + ↓ + message.sending sender_engine 进入发送阶段（已入队 / 开始直发） + ↓ + message.sent 全部通道发送成功（终态） + message.failed 任一环节失败：解析失败（stage="parse"）/ 发送失败（stage="send"） + +注意：**只有事件链一处驱动**，`process_message` 不再重复 emit 下游事件 +（历史上曾因此导致每条消息被投递 3 次，见 tests/test_event_chain.py）。 """ from blinker import Namespace diff --git a/circuit_breaker.py b/circuit_breaker.py new file mode 100644 index 0000000..cb2a102 --- /dev/null +++ b/circuit_breaker.py @@ -0,0 +1,326 @@ +#!/usr/bin/python3 +# -*- coding: UTF-8 -*- +""" +通道熔断器 —— CLOSED / OPEN / HALF_OPEN 三态机。 + +规格来源：doc/improvement.md #19（v1.3.0 韧性增强） + + CLOSED ──连续失败 / 窗口失败率超阈值──▶ OPEN + ▲ │ 冷却到期 + │ 连续 N 次探测成功 ▼ + └──────────── HALF_OPEN ◀─────────────┘ + │ 探测失败 + └──▶ OPEN（冷却时间翻倍） + +规则： + - 滑动窗口（默认 60s）内失败率 > 50% 且样本数 >= MIN_SAMPLES → 熔断 + - 或连续失败 >= CONSECUTIVE_THRESHOLD → 熔断（照顾低频通道） + - OPEN 冷却指数退避 30→60→120→240→480→600s（封顶 600s） + - HALF_OPEN 放行探测，连续 HALF_OPEN_NEEDED(3) 次成功才恢复 CLOSED + - **4xx 不计失败**，只对 5xx / 超时 / 连接类错误计数 + - 状态持久化到 SQLite(channel_breaker)，重启后恢复 + +线程安全：单把锁保护全部状态；仅在状态发生变化时落库（热路径不写库）。 +""" + +import os +import re +import time +import threading +import collections + +import log +from db.connection import _conn + +# ── 可调参数 ── +# 取值优先级：system_config（设置页/可直接改库） > 环境变量 > 这里的默认值。 +# 环境变量在模块加载时读一次作为兜底；system_config 支持运行时调整（带 TTL 缓存）， +# 这样用户不必为了改一个阈值重启容器，也不必依赖尚未完备的 API。 +WINDOW_SECONDS = float(os.getenv("EGO_BREAKER_WINDOW", "60")) +MIN_SAMPLES = int(os.getenv("EGO_BREAKER_MIN_SAMPLES", "5")) +FAILURE_RATIO = float(os.getenv("EGO_BREAKER_FAILURE_RATIO", "0.5")) +CONSECUTIVE_THRESHOLD = int(os.getenv("EGO_BREAKER_CONSECUTIVE", "5")) +OPEN_BASE_SECONDS = float(os.getenv("EGO_BREAKER_OPEN_BASE", "30")) +OPEN_MAX_SECONDS = float(os.getenv("EGO_BREAKER_OPEN_MAX", "600")) +HALF_OPEN_NEEDED = int(os.getenv("EGO_BREAKER_HALF_OPEN_OK", "3")) + +# system_config 里的键名（设置页用同一批键） +CONFIG_KEYS = { + "window": "breaker_window", + "min_samples": "breaker_min_samples", + "failure_ratio": "breaker_failure_ratio", + "consecutive": "breaker_consecutive", + "open_base": "breaker_open_base", + "open_max": "breaker_open_max", + "half_open_ok": "breaker_half_open_ok", +} + +_PARAM_TTL = float(os.getenv("EGO_BREAKER_PARAM_TTL", "30")) +_param_cache = {} +_param_cached_at = 0.0 +_param_lock = threading.Lock() + + +def _read_overrides(): + """从 system_config 读运行时覆盖值（带 TTL 缓存，避免每次发送都查库）。""" + global _param_cached_at + now = time.time() + with _param_lock: + if _param_cache and now - _param_cached_at < _PARAM_TTL: + return _param_cache + out = {} + try: + for short, key in CONFIG_KEYS.items(): + row = _conn().execute( + "SELECT value FROM system_config WHERE key=?", (key,)).fetchone() + if row and str(row[0]).strip() != "": + out[short] = str(row[0]).strip() + except Exception: + out = {} + _param_cache.clear() + _param_cache.update(out) + _param_cached_at = now + return _param_cache + + +def param(short, default): + """取一个熔断参数：system_config > 环境变量默认值。""" + raw = _read_overrides().get(short) + if raw is None: + return default + try: + return type(default)(raw) + except (TypeError, ValueError): + log.logger.warning(f"[Breaker] invalid override {short}={raw!r}, using {default}") + return default + + +def invalidate_param_cache(): + """清掉参数缓存（设置页改完即时生效）。""" + global _param_cached_at + with _param_lock: + _param_cache.clear() + _param_cached_at = 0.0 + +CLOSED, OPEN, HALF_OPEN = "closed", "open", "half_open" + +_HTTP_5XX = re.compile(r"\b5\d{2}\b") +_HTTP_4XX = re.compile(r"\b4\d{2}\b") +# 连接 / 超时类关键字（视为服务侧故障） +_FAILURE_HINTS = ( + "timeout", "timed out", "read timed out", + "connection", "max retries exceeded", "refused", + "unreachable", "reset by peer", "eof occurred", +) + + +def classify(error): + """判断一次失败是否计入熔断。 + + 返回 True = 计入失败；False = 忽略（4xx 等业务侧拒绝）。 + 未知错误保守计入失败——「发送没成功」比「漏计故障」代价更小。 + """ + e = (error or "").lower() + if _HTTP_5XX.search(e) or any(h in e for h in _FAILURE_HINTS): + return True + if _HTTP_4XX.search(e): + return False + return True + + +class _ChannelState: + __slots__ = ("state", "opened_at", "open_count", "half_open_ok", + "window", "consecutive_fail") + + def __init__(self): + self.state = CLOSED + self.opened_at = 0.0 + self.open_count = 0 # 累计熔断次数，用于指数退避 + self.half_open_ok = 0 + self.window = collections.deque() # (ts, is_failure) + self.consecutive_fail = 0 + + +class CircuitBreaker: + def __init__(self): + self._lock = threading.Lock() + self._states = {} + self._loaded = False + + # ── 内部 ── + def _st(self, channel_id): + st = self._states.get(channel_id) + if st is None: + st = _ChannelState() + self._states[channel_id] = st + return st + + def _cooldown(self, st): + """本次 OPEN 的冷却时长（指数退避，封顶 OPEN_MAX_SECONDS）。""" + n = max(0, st.open_count - 1) + return min(param("open_base", OPEN_BASE_SECONDS) * (2 ** n), + param("open_max", OPEN_MAX_SECONDS)) + + def _evict(self, st, now): + cutoff = now - param("window", WINDOW_SECONDS) + w = st.window + while w and w[0][0] < cutoff: + w.popleft() + + def _persist(self, st, channel_id): + try: + conn = _conn() + conn.execute( + """INSERT INTO channel_breaker + (channel_id, state, opened_at, open_count, half_open_ok, updated_at) + VALUES (?,?,?,?,?,datetime('now')) + ON CONFLICT(channel_id) DO UPDATE SET + state=excluded.state, + opened_at=excluded.opened_at, + open_count=excluded.open_count, + half_open_ok=excluded.half_open_ok, + updated_at=excluded.updated_at""", + (channel_id, st.state, st.opened_at, st.open_count, st.half_open_ok) + ) + conn.commit() + except Exception as e: + log.logger.warning(f"[Breaker] persist failed (ch={channel_id}): {e}") + + def _open(self, st, channel_id, now, reason): + st.state = OPEN + st.opened_at = now + st.open_count += 1 + st.half_open_ok = 0 + st.window.clear() + self._persist(st, channel_id) + log.logger.warning( + f"[Breaker] Channel {channel_id} → OPEN ({reason}); " + f"cooldown {int(self._cooldown(st))}s" + ) + + # ── 对外 ── + def load(self): + """从 SQLite 恢复状态（进程启动时调用一次）。""" + if self._loaded: + return + try: + rows = _conn().execute("SELECT * FROM channel_breaker").fetchall() + except Exception as e: + log.logger.warning(f"[Breaker] load failed: {e}") + self._loaded = True + return + with self._lock: + for r in rows: + st = self._st(r["channel_id"]) + st.state = r["state"] or CLOSED + st.opened_at = float(r["opened_at"] or 0) + st.open_count = int(r["open_count"] or 0) + st.half_open_ok = 0 # 重启后重新计数探测 + self._loaded = True + if rows: + log.logger.info(f"[Breaker] Restored {len(rows)} channel state(s) from DB") + + def should_allow(self, channel_id): + """是否放行本次发送。返回 (allowed: bool, reason: str|None)。""" + now = time.time() + with self._lock: + st = self._st(channel_id) + if st.state == CLOSED: + return True, None + if st.state == HALF_OPEN: + return True, None # 放行探测 + # OPEN + cd = self._cooldown(st) + if now - st.opened_at >= cd: + st.state = HALF_OPEN + st.half_open_ok = 0 + self._persist(st, channel_id) + log.logger.info(f"[Breaker] Channel {channel_id} → HALF_OPEN (probing)") + return True, None + left = int(cd - (now - st.opened_at)) + return False, f"open, {left}s left" + + def record(self, channel_id, ok, error=""): + """记录一次发送结果。""" + now = time.time() + with self._lock: + st = self._st(channel_id) + + if ok: + st.consecutive_fail = 0 + else: + if not classify(error): + return # 4xx 等不计失败 + st.consecutive_fail += 1 + + if st.state == HALF_OPEN: + if ok: + st.half_open_ok += 1 + if st.half_open_ok >= param("half_open_ok", HALF_OPEN_NEEDED): + st.state = CLOSED + st.open_count = 0 + st.half_open_ok = 0 + st.window.clear() + self._persist(st, channel_id) + log.logger.info(f"[Breaker] Channel {channel_id} → CLOSED (recovered)") + else: + self._open(st, channel_id, now, "probe failed") + return + + if st.state == OPEN: + return # OPEN 期间的结果不计（正常不会走到） + + # CLOSED：维护滑动窗口 + st.window.append((now, not ok)) + self._evict(st, now) + + if not ok and st.consecutive_fail >= param("consecutive", CONSECUTIVE_THRESHOLD): + self._open(st, channel_id, now, + f"{st.consecutive_fail} consecutive failures") + return + + total = len(st.window) + fails = sum(1 for _, f in st.window if f) + if (total >= param("min_samples", MIN_SAMPLES) + and fails / total > param("failure_ratio", FAILURE_RATIO)): + self._open(st, channel_id, now, f"failure ratio {fails}/{total}") + + def reset(self, channel_id): + """手工恢复某通道（运维用）。""" + with self._lock: + st = self._st(channel_id) + st.state = CLOSED + st.open_count = 0 + st.half_open_ok = 0 + st.consecutive_fail = 0 + st.window.clear() + self._persist(st, channel_id) + log.logger.info(f"[Breaker] Channel {channel_id} manually reset → CLOSED") + + def snapshot(self): + """当前处于非 CLOSED 的通道（供 API / 日志用）。""" + now = time.time() + with self._lock: + out = [] + for cid, st in self._states.items(): + if st.state == CLOSED: + continue + out.append({ + "channel_id": cid, + "state": st.state, + "open_count": st.open_count, + "retry_in": (max(0, int(self._cooldown(st) - (now - st.opened_at))) + if st.state == OPEN else 0), + }) + return out + + +# ── 单例 ── +_breaker = None + + +def get_breaker(): + global _breaker + if _breaker is None: + _breaker = CircuitBreaker() + return _breaker diff --git a/db/connection.py b/db/connection.py index 60d432f..d4dc9c2 100644 --- a/db/connection.py +++ b/db/connection.py @@ -31,5 +31,11 @@ def _conn(): conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout=5000") conn.execute("PRAGMA journal_mode=WAL") + # WAL 下 synchronous=NORMAL 不再每次提交都 fsync（仅在 checkpoint 时同步）， + # 大幅降低写入延迟；代价是掉电可能丢最后若干条事务（库本身仍一致）。 + # 详见 doc/improvement.md v1.3.0「SQLite 调优 pragma」。 + conn.execute("PRAGMA synchronous=NORMAL") + # 页缓存 64MB（负值 = KiB），减少重复读盘 + conn.execute("PRAGMA cache_size=-64000") _local.conn = conn return conn diff --git a/db/log_handler.py b/db/log_handler.py new file mode 100644 index 0000000..a89cf07 --- /dev/null +++ b/db/log_handler.py @@ -0,0 +1,34 @@ +#!/usr/bin/python3 +# -*- coding: UTF-8 -*- +""" +把日志写入 SQLite 的 logging.Handler。 + +为什么放在 db/ 而不是 log.py： + 保持依赖方向单向 —— main → db → log。 + 原先 log.py 的 DBHandler 硬编码 `import db`，造成 log ↔ db 互相依赖； + 现在 log.py 只提供「挂载一个处理器」的接口，具体实现由调用方注入。 + +用法（见 main.py）： + from db.log_handler import make_log_handler + log.setup_db_logging(make_log_handler()) +""" + +import logging + +from .queries import add_log + + +class DBLogHandler(logging.Handler): + """把日志记录写入 logs 表。""" + + def emit(self, record): + try: + add_log(record.levelname, record.getMessage(), record.name) + except Exception: + # 日志落库失败绝不能反过来把业务打挂 + pass + + +def make_log_handler(): + """构造一个写库的日志处理器。""" + return DBLogHandler() diff --git a/db/queries.py b/db/queries.py index 64b38c2..4c8aaef 100644 --- a/db/queries.py +++ b/db/queries.py @@ -7,6 +7,7 @@ import json import sqlite3 +import log from .connection import _conn @@ -205,8 +206,40 @@ def update_channel(channel_id, **kwargs): def delete_channel(channel_id): - _conn().execute("DELETE FROM channels WHERE id=?", (channel_id,)) - _conn().commit() + """删除通道，并级联清理引用它的记录。 + + schema 里 source_channels.channel_id 声明了 ON DELETE CASCADE，但连接未开启 + PRAGMA foreign_keys（SQLite 默认关闭），级联不会生效——不显式清理就会留下 + 指向已删通道的孤儿绑定，导致路由持续匹配到一个不存在的通道而反复失败。 + + 待发队列里属于该通道的任务已无法投递，移入死信队列保留（而不是静默丢弃）， + 以便在 WebUI 里可见、可查。 + """ + conn = _conn() + n = conn.execute( + "SELECT COUNT(*) FROM message_queue WHERE channel_id=?", (channel_id,) + ).fetchone()[0] + if n: + log.logger.warning( + f"Deleting channel #{channel_id}: moving {n} queued item(s) to DLQ") + # 注意：这里必须整批搬一次，不能放进循环——循环里执行会让 SELECT 每次 + # 都取回全部行，造成 N 倍重复插入 + conn.execute( + """INSERT INTO dead_letter_queue + (trace_id, source_id, msg_json, channel_id, template_id, + dedup_key, error, retry_count) + SELECT trace_id, source_id, msg_json, channel_id, template_id, + dedup_key, 'channel deleted', retry_count + FROM message_queue WHERE channel_id=?""", + (channel_id,) + ) + conn.execute("DELETE FROM message_queue WHERE channel_id=?", (channel_id,)) + conn.execute("DELETE FROM source_channels WHERE channel_id=?", (channel_id,)) + conn.execute("DELETE FROM dedup_keys WHERE channel_id=?", (channel_id,)) + conn.execute("DELETE FROM channel_breaker WHERE channel_id=?", (channel_id,)) + conn.execute("DELETE FROM channel_rate_limit WHERE channel_id=?", (channel_id,)) + conn.execute("DELETE FROM channels WHERE id=?", (channel_id,)) + conn.commit() def upsert_channel(cid, name, channel_type, config="{}", enabled=1): @@ -453,7 +486,7 @@ def cleanup_old_messages(overrides=None): if hours <= 0: continue cursor = conn.execute( - "DELETE FROM message_log WHERE status=? AND created_at < datetime('now','localtime',?||' hours')", + "DELETE FROM message_log WHERE status=? AND created_at < datetime('now',?||' hours')", (status, f"-{hours}") ) total += cursor.rowcount @@ -466,14 +499,40 @@ def mark_ignored(msg_id): _conn().commit() -def check_dedup(dedup_key, window_seconds): +def dedup_hit(channel_id, dedup_key, window_seconds): + """该通道在窗口内是否已成功发过同一个去重键。 + + 去重粒度是 **message × channel**：不同通道各用各的键、各算各的窗口， + 互不影响。旧实现是"整条消息一个键"，任一通道命中就把整条消息丢掉， + 且只取第一个配了去重表达式的绑定的键（其余绑定被忽略）。 + """ r = _conn().execute( - "SELECT COUNT(*) FROM message_log WHERE dedup_key=? AND status='SUCCESS' AND sent_at > datetime('now','-'||?||' seconds')", - (dedup_key, str(window_seconds)) + "SELECT COUNT(*) FROM dedup_keys WHERE channel_id=? AND dedup_key=? " + "AND sent_at > datetime('now','-'||?||' seconds')", + (channel_id, dedup_key, str(int(window_seconds))) ).fetchone() return r[0] > 0 +def dedup_record(channel_id, dedup_key): + """记录一次成功发送的去重键（供后续窗口判定）。""" + _conn().execute( + "INSERT INTO dedup_keys (channel_id, dedup_key) VALUES (?,?)", + (channel_id, dedup_key) + ) + _conn().commit() + + +def dedup_purge(keep_seconds=7 * 86400): + """清理过老的去重键（窗口远小于保留期，保留 7 天足够）。""" + cur = _conn().execute( + "DELETE FROM dedup_keys WHERE sent_at < datetime('now','-'||?||' seconds')", + (str(int(keep_seconds)),) + ) + _conn().commit() + return cur.rowcount + + def get_queue_stats(): total = _conn().execute("SELECT COUNT(*) FROM message_log WHERE status IN ('PENDING','FAILED','SUCCESS')").fetchone()[0] pending = _conn().execute("SELECT COUNT(*) FROM message_log WHERE status='PENDING'").fetchone()[0] @@ -564,6 +623,62 @@ def set_log_level(level): # Stats # ═══════════════════════════════════════════════ +def get_channel_stats(hours=24): + """按渠道聚合最近的发送结果（供 /api/metrics）。 + + 从 message_log.channel_results（JSON 数组）里数每个渠道的尝试次数与成功次数。 + 窗口内条数有限（自管理场景），在 Python 侧解析即可，不必上 JSON 扩展查询。 + """ + rows = _conn().execute( + "SELECT channel_results FROM message_log " + "WHERE channel_results IS NOT NULL AND channel_results NOT IN ('','[]') " + "AND created_at > datetime('now','-'||?||' hours')", + (str(int(hours)),) + ).fetchall() + + stats = {} + for r in rows: + try: + items = json.loads(r["channel_results"] or "[]") + except Exception: + continue + for item in items: + cid = item.get("channel_id") + key = cid if cid is not None else (item.get("ch_name") or "?") + s = stats.setdefault(str(key), { + "channel_id": cid, "name": item.get("ch_name", "?"), + "type": item.get("ch_type", ""), "attempts": 0, "ok": 0, + }) + s["attempts"] += 1 + if item.get("ok"): + s["ok"] += 1 + + out = list(stats.values()) + for s in out: + s["success_rate"] = round(s["ok"] / s["attempts"], 4) if s["attempts"] else None + out.sort(key=lambda x: str(x["channel_id"])) + return out + + +def get_latency_stats(hours=24): + """端到端延迟（created_at → sent_at，单位秒）。两者都是 UTC。""" + row = _conn().execute( + "SELECT AVG((julianday(sent_at)-julianday(created_at))*86400.0) avg_s, " + " MAX((julianday(sent_at)-julianday(created_at))*86400.0) max_s, " + " COUNT(*) n " + "FROM message_log WHERE sent_at IS NOT NULL " + "AND created_at > datetime('now','-'||?||' hours')", + (str(int(hours)),) + ).fetchone() + if not row or not row["n"]: + return {"samples": 0, "avg_seconds": None, "max_seconds": None} + return { + "samples": row["n"], + "avg_seconds": round(row["avg_s"], 3) if row["avg_s"] is not None else None, + "max_seconds": round(row["max_s"], 3) if row["max_s"] is not None else None, + } + + def get_stats(): qs = get_queue_stats() return { diff --git a/db/schema.py b/db/schema.py index 88549fa..8f88938 100644 --- a/db/schema.py +++ b/db/schema.py @@ -89,6 +89,31 @@ def init_db(): value TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS channel_breaker ( + channel_id INTEGER PRIMARY KEY, + state TEXT NOT NULL DEFAULT 'closed', + opened_at REAL DEFAULT 0, + open_count INTEGER DEFAULT 0, + half_open_ok INTEGER DEFAULT 0, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS channel_rate_limit ( + channel_id INTEGER PRIMARY KEY, + per_minute INTEGER NOT NULL DEFAULT 0, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS dedup_keys ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel_id INTEGER NOT NULL, + dedup_key TEXT NOT NULL, + sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_dedup_lookup + ON dedup_keys(channel_id, dedup_key, sent_at); + CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, @@ -179,6 +204,12 @@ def init_db(): except sqlite3.OperationalError: pass + # message_queue: 延迟重排计数（熔断/限流期间放回队列，不消耗重试次数） + try: + conn.execute("ALTER TABLE message_queue ADD COLUMN defer_count INTEGER DEFAULT 0") + except sqlite3.OperationalError: + pass + # migrate old message_log that lack new columns for col in ['source_name', 'msg_json', 'dedup_key', 'parser_hash', 'channel_results', 'sent_at', 'updated_at']: try: diff --git a/doc/changelog.md b/doc/changelog.md index e960a0f..46a428c 100644 --- a/doc/changelog.md +++ b/doc/changelog.md @@ -4,6 +4,144 @@ --- +## v1.3.0（2026-07-28） + +> 主题：**工程收口** —— 从「功能实现」转向「异常情况下是否可靠」。 +> 可靠性 + 入口并发 + 一个让重试形同失效的时区 bug。 + +### 可靠性（本轮重点） + +- 🆕 **通道熔断（#19）** `circuit_breaker.py`：CLOSED / OPEN / HALF_OPEN 三态机。 + - 滑动窗口 60s 内失败率 > 50%（样本数 ≥ 5）**或**连续失败 ≥ 5 → 熔断（后者照顾低频通道） + - OPEN 冷却指数退避 30→60→120→240→480→600s（封顶 10min） + - HALF_OPEN 连续 3 次探测成功才恢复 CLOSED，恢复后退避计数归零 + - **4xx 不计失败**，只对 5xx / 超时 / 连接类错误计数 + - 状态持久化 `channel_breaker` 表，重启后自动恢复 + - 参数由 `EGO_BREAKER_*` 环境变量覆盖；`GET /api/resilience` 可查，支持手工 reset +- 🆕 **出站限流（#21）** `rate_limiter.py`：每通道独立令牌桶（条/分钟）。 + - 桶容量 = 1 分钟额度（允许小幅突发） + - `acquire()` 最多等 1s，拿不到令牌就**延迟重排**，不长阻塞 worker 线程 + - 配置存 `channel_rate_limit` 表；缓存带 30s TTL 回查，直接改库也能自愈 +- 🆕 **延迟重排** `queue_backend.defer()`：熔断 / 限流期间把任务放回队列， + 但**不消耗重试次数**——否则故障期内消息会被重试耗尽、直接跌进死信队列。 + 超过 50 次仍发不出去，才交回正常重试 / DLQ 路径。 + +> 为什么限流不能靠 Retry 兜底：发送过快 → 429 → 重试 → 再次 429。 +> 限流必须位于 `Worker → Rate Limiter → Channel`（Nginx 只管入站，管不到出站）。 + +- 🆕 **优雅停机**：`worker.stop_workers(timeout=30)` 改为「停消费 → join 等在途任务 → + 超时兜底把未完成的刷入死信队列」；`main.shutdown()` 明确**先停接收端再停 worker** + （反过来的话接收端还会继续塞消息，worker 已退出，消息会卡在队列里）。 + 原实现只置 `_running=False` 就返回，进程随即退出，在途任务会卡在 processing 状态。 + +### 入口并发 + +- 🆕 **端口数据源入口并发** `source_listener/__init__.py`：单线程 `HTTPServer` + → 固定大小工作线程池（`_IngressPool` + `_ThreadPoolHTTPServer`）。 + 原实现下同一数据源的请求被**串行**处理，一个慢解析器会阻塞该数据源上所有后续请求。 + - 线程复用 → `threading.local()` 的 DB 连接随之复用，不再每请求新建 SQLite 连接 + - 队列满回 503 做背压，不无限堆积 + - 新增 `EGO_INGRESS_WORKERS`（默认 8）、`EGO_INGRESS_MAX_QUEUE`（默认 200） +- `request_queue_size = 128`：默认仅 5，突发时内核会拒掉多余连接，客户端只能等 + TCP SYN 重传（实测表现为 ~1.1s 长尾）。 +- `stop_source()` 补 `server_close()`：原来只 `shutdown()`，监听 socket 与线程池不释放。 + +> **范围说明**：路径路由入口（`/in/...`）本就并发（`run_simple(threaded=True)`），不在此列。 +> **实测结论**：并发 8 时 p50 由 127ms 降至 21ms；**但加线程并不提升吞吐**—— +> 吞吐受单进程串行段限制，线程池的价值是「慢请求不阻塞其它请求」，故默认值取 8 而非更大。 + +### 渠道级语义（P1） + +- 🆕 **渠道级重发（Channel-Level Retry）**：`retry_message(msg_id, mode, scope="failed")` + 默认**只重发上次失败的渠道**，不再把已成功的渠道重复推送。 + 结果条目补 `channel_id` 作为定位依据；旧记录缺该字段时自动回退「整条重发」（向后兼容）。 + `scope="all"` 保留整体重推能力。 +- 🆕 **渠道级去重（Channel-Level Dedup）**：去重粒度从「整条消息」下沉到 + **message × channel**。新增 `dedup_keys(channel_id, dedup_key, sent_at)` 表， + 逐绑定判定——命中的只跳过该渠道，全部命中才把消息标记 `DISCARDED`。 + 修掉了原实现的两个问题：只取第一个绑定的去重表达式（其余被 `break` 忽略）、 + 以及任一命中就丢弃整条消息。 + +### 并发一致性（P1） + +- 🔴 **多 worker 结果回写丢更新**：`update_message_results()` 是「读 JSON → 改 → 写回」， + 并发下会互相覆盖。实测 60 线程并发写同一条 trace：**加锁保留 60/60，不加锁只活下来 2/60**。 + 修复：按 `hash(trace_id)` 取模的**条带锁**（固定 64 把，不增长、无需清理）。 + +### 性能 + +- **SQLite 调优 pragma**：`db/connection.py` 补 `synchronous=NORMAL` + `cache_size=-64000`。 + 实测单次 commit 由 **6.98ms → 0.02ms（约 350×）**，整机吞吐（并发 8）由 **31 → 322 req/s（约 10×）**。 + 代价：掉电可能丢最后若干条已提交事务（库本身仍保持一致）。 + +### Bug 修复 + +- 🔴 **重试被推迟约 8 小时（时区基准错误）** + `queue_backend.nack()` 用 Python `datetime.now()`（本地时间）写 `next_retry_at`， + 而 `dequeue()` 比较的是 SQLite `datetime('now')`（**UTC**）。 + 在 UTC+8 等非 UTC 时区部署下，退避时间比当前 UTC 晚 8 小时 → 重试实际被推迟约 8 小时， + **等于重试机制失效**。改用 `datetime('now', '+N seconds')`，两侧统一为 UTC。 + `defer()` 同样处理。回归测试：`tests/test_queue_defer.py::test_retry_due_time_uses_utc_not_local`。 + +- 🔴 **每条消息重复投递 3 次**（自 v1.1.0 / commit `47ac7f9a` 起存在） + `source_manager.process_message()` 除事件链外，自己又重复 emit 了 `message.parsed` + 与 `message.routed`，导致同一消息被投递 3 次（解析链内各发一次 + 这里再发两次）。 + **定位过程**：线上临时探针显示 `receivers=1`（信号只有一个订阅者）但处理器被调用 3 次 + → 不是重复注册，而是重复 emit。修复：链路改为**只由事件总线单向驱动**， + `process_message` 只 emit 一次 `message.received`；`extra_fields`（如 `sub_path`） + 改由 `parser_engine` 在触发 `message.parsed` **之前**合并 + （原先它只在冗余 emit 里才生效）。回归测试：`tests/test_event_chain.py`。 +- 🟠 **`sent_at` 与 `created_at` 时区基准不一致** + `dt_now_str()` 用 `datetime.now()`（本地时间）写 `sent_at`，而 `created_at` 由 + `CURRENT_TIMESTAMP` 生成（UTC）——两者相差一个时区偏移，消息列表里「创建/发送时间」 + 对不上、按二者算延迟会错好几小时。修复：`dt_now_str()` 改为返回 UTC。 + +### API + +- 🆕 韧性接口（`api/system.py`）： + - `GET /api/resilience` — 熔断状态 + 限流配置 + - `POST /api/resilience/rate_limit/` — 设置限流（条/分钟，0 = 不限） + - `POST /api/resilience/breaker//reset` — 手工恢复熔断通道 +- 🆕 `GET /api/metrics`：队列深度 / 死信总数 / 各通道成功率 / 端到端延迟 / + 熔断与限流状态。定位是「curl 一查就有」，**不引入 Prometheus**；`?hours=N` 调整窗口。 +- 🆕 **API 入参校验（#27）**：新增 `api/validation.py`（轻量助手，不引 JSON Schema）， + `ValidationError` 由统一错误处理器转 400。覆盖数据源 / 通道 / 模板 / 设置 / 消息批量的写接口—— + 原先缺字段会直接 500，端口、slug、引擎、时段等无效值会被原样写进配置。 + +### 界面 + +- 🆕 **熔断 / 限流的 WebUI**（此前只有 API，只能 curl 或改库）： + - 通道编辑弹窗新增「出站限流（条/分钟）」输入框，留空/0 = 不限流；保存通道时一并写入 + - 通道列表新增「韧性」列：限流显示徽章（如 `60/分`），熔断显示红色倒计时徽章 + （如 `熔断中 21s`）并提供一键「手工恢复」按钮 + - 中英文文案齐备 + +### 测试 + +- 新增 96 例（102 → **198 passed**）： + - 入口并发 3、熔断 13、限流 9、队列延迟重排 4（P0） + - 渠道级去重/重发 + 多 worker 一致性 14（P1） + - 事件链与生命周期 9（P2，含「入队恰好 1 次」的重复投递回归测试） + - 系统级可靠性 13（故障注入 / 熔断×队列 / 重试→死信 / 崩溃恢复 / 并发消费 / 守恒） + - 优雅停机、`/api/metrics`、入参校验、模块解耦、时间基准、韧性 UI 31（P2 收口） + +### 工程（工具链） + +- 🆕 `skills/ego_deploy/deploy.py`：测试环境**非破坏同步** + （按 md5 只写变化文件、**永不删除**、排除 `config/`/`certs/`/`ego.db*`），替代原 `rm -rf` + `cp -r`。 +- 🔴 修 `ego_deploy` 的**「重启」空操作**：原用 `ps | grep 目录名` 定位进程，但 busybox 的 `ps` + 命令行**不含工作目录**，永远匹配到空列表 → 不 kill、只新起一个抢不到端口的进程， + **旧进程继续跑旧代码，脚本却报「已重启成功」**。改用扫描 `/proc//cwd` + 核对 cmdline， + 并强制校验 **PID 必须变化**，否则明确报错。 +- **优雅停机改用 SIGTERM**：部署脚本先发 SIGTERM 等进程自然退出（超时再 SIGKILL）， + 而不是直接 `kill -9` —— 后者会跳过整个优雅停机路径。 +- **模块解耦**：`log.py` 不再 `import db`；写库的 `DBLogHandler` 移到 + `db/log_handler.py`，由 `main.py` 注入。依赖方向固定为 `main → db → log`。 +- **清理间隔可配**：`EGO_CLEANUP_INTERVAL`（默认 600s），原先硬编码。 +- **`requirements.txt` 移除 UTF-8 BOM**（首字节原为 `\ufeff`）。 + +--- + ## v1.2.4（2026-07-28） ### 部署配置重构（deploy/） diff --git a/doc/roadmap.md b/doc/roadmap.md index 55a4ea5..35b9729 100644 --- a/doc/roadmap.md +++ b/doc/roadmap.md @@ -5,7 +5,7 @@ --- -## v1.3.0（下一个大版本） +## 后续版本（规划中） ### 开放 API（规划中） @@ -13,12 +13,39 @@ - 需配套 API Key 认证机制 - 与路径路由共享入口端口 -### 韧性增强 +### 韧性增强 —— ✅ 已在 v1.3.0 交付 -| 项目 | 改进方向 | 来源 | -|------|----------|------| -| 通道熔断 | 连续失败 N 次自动熔断，定期探测恢复，防第三方宕机拖垮线程池 | 改进文档 #19 | -| 出站通道限流 | 令牌桶（条/分钟）按通道独立，防多通道并发被封号（Nginx 管不到出站） | 改进文档 #21 | +| 项目 | 状态 | 来源 | +|------|------|------| +| 通道熔断 | ✅ v1.3.0 `circuit_breaker.py`（三态机 + 滑动窗口 + 指数退避 + 4xx 不计失败 + 持久化） | 改进文档 #19 | +| 出站通道限流 | ✅ v1.3.0 `rate_limiter.py`（每通道令牌桶，拿不到令牌延迟重排） | 改进文档 #21 | + +### v1.3.0 追加交付 + +| 项 | 说明 | +|------|------| +| Channel-Level Retry | 重发只针对上次失败的渠道，`scope=all` 可整体重推 | +| Channel-Level Dedup | 去重粒度下沉到 message × channel（`dedup_keys` 表） | +| Multi-Worker 一致性 | 结果回写条带锁；实测 2/60 → 60/60 不丢更新 | +| 系统级测试 | 故障注入 / 熔断×队列 / 重试→死信 / 崩溃恢复 / 并发消费 / 守恒 | +| 优雅停机 | 停接收 → 等在途任务（超时 30s）→ 未完成刷死信 | +| 可观测性 | `GET /api/metrics`（队列/死信/通道成功率/延迟/熔断） | +| 韧性 WebUI | 通道弹窗限流输入框 + 列表「韧性」列（熔断倒计时徽章 + 一键恢复） | +| API 入参校验（#27） | 新增 `api/validation.py`，非法入参统一 400 | +| 去重逻辑收敛（#31） | 两处发送路径共用 `_plan_dedup()` | +| 日志模块解耦 | `log.py` 不再依赖 `db`；`DBLogHandler` 移到 `db/log_handler.py` | +| 清理间隔可配 | `EGO_CLEANUP_INTERVAL`（默认 600s） | +| 时间基准修正 | `sent_at` 与 `created_at` 统一 UTC | + +**注意**：本轮还修了一个长期存在的严重 bug —— 每条消息被**重复投递 3 次** +（自 v1.1.0 起，`process_message` 重复 emit 下游事件）。详见 `changelog.md`。 + +### 待办（v1.4.0 候选） + +- 结构化日志（JSON formatter + 敏感信息脱敏） +- 事件总线超时机制（`bus.emit()` 仍是同步阻塞） +- 类型注解补齐 +- 开放 API（REST 投递 + API Key 认证） --- @@ -33,6 +60,7 @@ ## 已完成版本 +- **v1.3.0**（2026-07-28）— 工程收口：通道熔断(#19) + 出站限流(#21) + 延迟重排 + 端口数据源入口线程池 + SQLite pragma 调优（吞吐 10×）；修复「重试被推迟 8 小时」时区 bug，详见 changelog.md - **v1.2.4**（2026-07-28）— 部署配置重构（T4 改 acme.sh + Cloudflare DNS、证书分离、双语 init.sh、端口检测）+ #22a 上传路径穿越修复，详见 changelog.md - **v1.2.3**（2026-07-27）— deploy/ 五套环境配置统一、EGO_SSL_DIR 参数一致性、nginx 反代跳转修复，详见 changelog.md - **v1.2.2**（2026-07-26）— 可靠性加固（线程安全 DB、静默异常日志、备份路径穿越、ZIP 炸弹、CSRF cookie）+ **路径路由（#17，统一入口 `/in/`）**，详见 changelog.md diff --git a/i18n.py b/i18n.py index b2de956..7588cb5 100644 --- a/i18n.py +++ b/i18n.py @@ -319,6 +319,16 @@ def get_all(): "ch.test_fail": "测试失败：", "ch.save_fail": "保存失败：", "ch.delete_fail": "删除失败：", + "ch.col_resilience": "韧性", + "ch.rate_limit_label": "出站限流（条/分钟）", + "ch.rate_limit_help": "0 或留空 = 不限流。超速的消息不会丢，而是排队等待重发。", + "ch.rate_limit_unit": "/分", + "ch.rate_limit_badge_title": "该通道的出站限流", + "ch.breaker_open": "熔断中", + "ch.breaker_probing": "探测中", + "ch.breaker_open_title": "连续失败已触发熔断；冷却结束后自动探测恢复", + "ch.breaker_reset": "手工恢复（立即解除熔断）", + "ch.breaker_reset_done": "已解除熔断", # ── 推送模板 ── "tpl.title": "推送模板", @@ -446,6 +456,15 @@ def get_all(): "set.cleanup_col_desc": "说明", "set.cleanup_col_hours": "保留时长(小时)", "set.cleanup_now": "立即清理", + "set.breaker_title": "韧性（通道熔断）", + "set.breaker_help": "留空 = 使用默认值。连续失败过多或窗口内失败率过高时，该通道会被暂时熔断，冷却结束后自动探测恢复——故障期间消息留在队列里等待，不会被丢弃。", + "set.breaker_window": "滑动窗口（秒）", + "set.breaker_min_samples": "触发判定的最少样本数", + "set.breaker_failure_ratio": "失败率阈值（0~1）", + "set.breaker_consecutive": "连续失败阈值", + "set.breaker_open_base": "冷却基数（秒，逐次翻倍）", + "set.breaker_open_max": "冷却上限（秒）", + "set.breaker_half_open_ok": "恢复所需连续探测成功次数", "set.save": "保存设置", "set.version": "版本:", "set.backup_title": "备份", @@ -918,6 +937,16 @@ def get_all(): "ch.test_fail": "Test failed: ", "ch.save_fail": "Save failed: ", "ch.delete_fail": "Delete failed: ", + "ch.col_resilience": "Resilience", + "ch.rate_limit_label": "Outbound rate limit (per minute)", + "ch.rate_limit_help": "0 or empty = unlimited. Over-rate messages are queued for retry, not dropped.", + "ch.rate_limit_unit": "/min", + "ch.rate_limit_badge_title": "Outbound rate limit for this channel", + "ch.breaker_open": "Circuit open", + "ch.breaker_probing": "Probing", + "ch.breaker_open_title": "Repeated failures tripped the breaker; it auto-probes after cooldown", + "ch.breaker_reset": "Reset (close circuit now)", + "ch.breaker_reset_done": "Circuit closed", # ── Templates ── "tpl.title": "Push Templates", @@ -1045,6 +1074,15 @@ def get_all(): "set.cleanup_col_desc": "Description", "set.cleanup_col_hours": "Retention (hours)", "set.cleanup_now": "Clean Now", + "set.breaker_title": "Resilience (channel circuit breaker)", + "set.breaker_help": "Empty = use default. A channel is tripped after too many consecutive failures or a high failure ratio within the window; it auto-probes back after cooldown. Messages stay queued during the outage instead of being dropped.", + "set.breaker_window": "Sliding window (seconds)", + "set.breaker_min_samples": "Min samples before judging", + "set.breaker_failure_ratio": "Failure ratio threshold (0-1)", + "set.breaker_consecutive": "Consecutive failure threshold", + "set.breaker_open_base": "Cooldown base (s, doubles each time)", + "set.breaker_open_max": "Cooldown cap (seconds)", + "set.breaker_half_open_ok": "Probe successes needed to recover", "set.save": "Save Settings", "set.version": "Version:", "set.backup_title": "Backup", diff --git a/log.py b/log.py index c6b23f8..80805d7 100644 --- a/log.py +++ b/log.py @@ -1,53 +1,56 @@ -#!/usr/bin/python3 -# -*- coding: UTF-8 -*- -"""日志模块""" - -import logging -import colorlog -import os - -log_colors_config = { - 'DEBUG': 'cyan', - 'INFO': 'green', - 'WARNING': 'yellow', - 'ERROR': 'red', - 'CRITICAL': 'red,bg_white', -} - -logger = logging.getLogger('ego') - -console_handler = logging.StreamHandler() - -log_level = os.getenv('LOG_LEVEL', 'INFO') -level = getattr(logging, log_level, logging.INFO) -logger.setLevel(level) -console_handler.setLevel(level) - -console_formatter = colorlog.ColoredFormatter( - fmt='%(log_color)s[%(levelname)s] %(message)s', - log_colors=log_colors_config -) -console_handler.setFormatter(console_formatter) -logger.addHandler(console_handler) - - -class DBHandler(logging.Handler): - """将日志写入数据库""" - def emit(self, record): - try: - import db - db.add_log(record.levelname, record.getMessage(), record.name) - except Exception: - pass - - -_db_handler = None - - -def setup_db_logging(): - global _db_handler - if _db_handler: - return - _db_handler = DBHandler() - _db_handler.setLevel(logger.level) - logger.addHandler(_db_handler) +#!/usr/bin/python3 +# -*- coding: UTF-8 -*- +"""日志模块 + +只负责控制台输出与「可注入的写库处理器挂载点」。 + +依赖方向：main → db → log（单向）。 +本模块**不 import db**——写库处理器由调用方注入，见 `db/log_handler.py`。 +""" + +import logging +import colorlog +import os + +log_colors_config = { + 'DEBUG': 'cyan', + 'INFO': 'green', + 'WARNING': 'yellow', + 'ERROR': 'red', + 'CRITICAL': 'red,bg_white', +} + +logger = logging.getLogger('ego') + +console_handler = logging.StreamHandler() + +log_level = os.getenv('LOG_LEVEL', 'INFO') +level = getattr(logging, log_level, logging.INFO) +logger.setLevel(level) +console_handler.setLevel(level) + +console_formatter = colorlog.ColoredFormatter( + fmt='%(log_color)s[%(levelname)s] %(message)s', + log_colors=log_colors_config +) +console_handler.setFormatter(console_formatter) +logger.addHandler(console_handler) + + +_db_handler = None + + +def setup_db_logging(handler): + """挂载「写入数据库」的日志处理器（幂等）。 + + handler 由调用方注入，例如： + from db.log_handler import make_log_handler + log.setup_db_logging(make_log_handler()) + """ + global _db_handler + if _db_handler: + return _db_handler + handler.setLevel(logger.level) + logger.addHandler(handler) + _db_handler = handler + return handler diff --git a/main.py b/main.py index fb3969d..aba708e 100644 --- a/main.py +++ b/main.py @@ -31,7 +31,7 @@ import worker # 异步发送 worker from web_ui import run_web_ui, has_ssl, ssl_enabled_by_env, app as web_app -VERSION = "1.2.4" +VERSION = "1.3.0" AUTHOR = "codename-test" DESCRIPTION = "EverywhereYouGo (EGo) — 通用信息转发平台" @@ -91,8 +91,9 @@ def init_ego(): log.logger.info("Initializing database...") db.init_db() - # 1.1 挂载数据库日志处理器 - log.setup_db_logging() + # 1.1 挂载数据库日志处理器（处理器由 db/ 提供，log 模块不反向依赖 db） + from db.log_handler import make_log_handler + log.setup_db_logging(make_log_handler()) # 1.5 加载配置（JSON → SQLite） import config_manager @@ -111,6 +112,11 @@ def init_ego(): log.logger.info(f"Log level: {log_level}") # 3. 启动异步发送 worker（恢复崩溃遗留任务 + 开始消费队列） + # 先恢复熔断状态与限流配置，再放 worker 出笼 + import circuit_breaker + import rate_limiter + circuit_breaker.get_breaker().load() + rate_limiter.get_limiter().load() worker.start_workers() # 4. 启动所有数据源监听 @@ -128,12 +134,16 @@ def init_ego(): dnd_thread.start() log.logger.info("DND queue checker started.") - # 6. 启动消息清理线程（每 10 分钟清理一次旧消息） + # 6. 启动消息清理线程（间隔可配，默认 10 分钟） + cleanup_interval = int(os.getenv("EGO_CLEANUP_INTERVAL", "600")) + def cleanup_loop(): while True: - time.sleep(600) + time.sleep(cleanup_interval) try: db.cleanup_old_messages() + # 顺带清理过老的去重键（窗口远小于保留期） + db.dedup_purge() except Exception as e: log.logger.error(f"Cleanup error: {e}") @@ -163,6 +173,25 @@ def cleanup_loop(): return mgr +def shutdown(mgr): + """优雅停机：**先停接收**（不再进新消息），**再停 worker**（等在途完成）。 + + 顺序很重要：先停 worker 的话，接收端还会继续塞新消息进来， + 这些消息会卡在队列里没人处理。 + + 提到模块级是为了可测试——tests/test_p2_hardening.py 会断言调用顺序。 + """ + log.logger.info("Shutting down...") + try: + mgr.stop_all() # 1) 停接收 + except Exception as e: + log.logger.error(f"stop_all failed: {e}") + try: + worker.stop_workers() # 2) 停 worker：等在途完成，超时刷死信 + except Exception as e: + log.logger.error(f"stop_workers failed: {e}") + + def main(): """启动入口：Flask 开发服务器。""" print(WELCOME) @@ -185,9 +214,7 @@ def main(): # 信号处理 def signal_handler(sig, frame): - log.logger.info("Shutting down...") - worker.stop_workers() - mgr.stop_all() + shutdown(mgr) sys.exit(0) signal.signal(signal.SIGINT, signal_handler) @@ -196,9 +223,7 @@ def signal_handler(sig, frame): try: run_web_ui(web_port, ssl_port) except KeyboardInterrupt: - log.logger.info("Shutting down...") - worker.stop_workers() - mgr.stop_all() + shutdown(mgr) if __name__ == "__main__": diff --git a/parser_engine/__init__.py b/parser_engine/__init__.py index 9a32a3a..b64b9ba 100644 --- a/parser_engine/__init__.py +++ b/parser_engine/__init__.py @@ -28,10 +28,14 @@ def _calc_parser_hash(filename): return "" -def _on_message_received(sender, *, trace_id, source_id, raw_body, headers, query_params): +def _on_message_received(sender, *, trace_id, source_id, raw_body, headers, query_params, + extra_fields=None): """ message.received 事件处理器。 执行解析器，成功时触发 message.parsed，返回 (ok, msg) 元组。 + + extra_fields 在这里合并（而非调用方），因为路由发生在 message.parsed 之后， + 而这些字段（如路径路由的 sub_path）必须赶在路由之前进入 msg。 """ src = db.get_source(source_id) parser = db.get_parser(src["parser_id"]) if src and src.get("parser_id") else None @@ -58,11 +62,17 @@ def _on_message_received(sender, *, trace_id, source_id, raw_body, headers, quer bus.emit(bus.message_failed, trace_id=trace_id, stage="parse", error=str(e)[:500]) return False, None + # 合并额外字段（如 sub_path）；必须在触发 message.parsed 之前 + if extra_fields and isinstance(msg, dict): + msg.update(extra_fields) + if isinstance(msg, dict): + msg["_trace_id"] = trace_id + # 存储解析结果 + 解析器版本哈希 msg_json = json.dumps(msg, ensure_ascii=False) db.update_message(trace_id, status="PARSED", msg_json=msg_json, parser_hash=parser_hash) - # 触发 message.parsed 事件 + # 触发 message.parsed 事件（→ router_engine → message.routed → sender_engine） bus.emit(bus.message_parsed, trace_id=trace_id, source_id=source_id, msg=msg) return True, msg diff --git a/queue_backend.py b/queue_backend.py index a0db26f..55eedce 100644 --- a/queue_backend.py +++ b/queue_backend.py @@ -96,17 +96,18 @@ def nack(self, queue_id, error=""): ) else: # 计算下次重试时间（指数退避） + # 注意：必须用 SQLite 的 datetime('now') 做基准——它返回 UTC， + # 与 dequeue() 的比较条件一致。原先用 Python 的 datetime.now() + # （本地时间）写入，在 UTC+8 等时区下会让重试被推迟约 8 小时。 delay = RETRY_DELAYS[min(retry_count - 1, len(RETRY_DELAYS) - 1)] - next_retry = ( - datetime.datetime.now() + datetime.timedelta(seconds=delay) - ).strftime("%Y-%m-%d %H:%M:%S") conn.execute( """UPDATE message_queue - SET status='pending', retry_count=?, next_retry_at=?, + SET status='pending', retry_count=?, + next_retry_at=datetime('now', ?), last_error=? WHERE id=?""", - (retry_count, next_retry, str(error)[:500], queue_id) + (retry_count, f"+{int(delay)} seconds", str(error)[:500], queue_id) ) log.logger.info( f"[{row['trace_id']}] Retry {retry_count}/{max_retries} " @@ -115,6 +116,60 @@ def nack(self, queue_id, error=""): conn.commit() + def defer(self, queue_id, delay_seconds=5, max_defers=50): + """延迟重排：放回队列但**不消耗重试次数**。 + + 用于熔断（circuit open）与限流（rate limited）期间的等待—— + 这类「没轮到我发」不应算作发送失败，否则消息会在故障期内被耗尽重试次数， + 直接跌进死信队列。 + + 超过 max_defers 次仍未能发出，则交回正常重试/DLQ 路径处理， + 避免故障通道让队列无限堆积。 + """ + with self._lock: + conn = _conn() + row = conn.execute( + "SELECT defer_count FROM message_queue WHERE id=?", (queue_id,) + ).fetchone() + if not row: + return + dc = (row["defer_count"] or 0) + 1 + if dc <= max_defers: + # 基准同样用 SQLite 的 datetime('now')（UTC），与 dequeue() 比较条件一致 + conn.execute( + "UPDATE message_queue SET status='pending', defer_count=?, " + "next_retry_at=datetime('now', ?) WHERE id=?", + (dc, f"+{int(delay_seconds)} seconds", queue_id) + ) + conn.commit() + return + # 超过延迟上限：锁外走正常重试/死信路径 + self.nack(queue_id, f"deferred {dc - 1} times without sending (circuit open / rate limited)") + + def flush_processing_to_dlq(self, reason="shutdown"): + """把仍处于 processing 的任务移入死信队列（优雅停机超时兜底）。 + + 正常情况下 `worker.stop_workers(timeout)` 会等在途任务跑完； + 超时兜底时把它们落到死信队列，避免消息卡在 processing 状态无人处理。 + """ + with self._lock: + conn = _conn() + rows = conn.execute( + "SELECT * FROM message_queue WHERE status='processing'").fetchall() + for row in rows: + conn.execute( + """INSERT INTO dead_letter_queue + (trace_id, source_id, msg_json, channel_id, template_id, + dedup_key, error, retry_count) + VALUES (?,?,?,?,?,?,?,?)""", + (row["trace_id"], row["source_id"], row["msg_json"], + row["channel_id"], row["template_id"], row["dedup_key"], + str(reason)[:1000], row["retry_count"]) + ) + conn.execute("DELETE FROM message_queue WHERE id=?", (row["id"],)) + conn.commit() + return len(rows) + def get_stats(self): """返回队列统计信息。""" conn = _conn() diff --git a/rate_limiter.py b/rate_limiter.py new file mode 100644 index 0000000..a876e66 --- /dev/null +++ b/rate_limiter.py @@ -0,0 +1,155 @@ +#!/usr/bin/python3 +# -*- coding: UTF-8 -*- +""" +出站通道限流 —— 每通道独立令牌桶。 + +规格来源：doc/improvement.md #21 + +背景：Retry 解决不了 API 限额——发送过快 → 429 → 重试 → 再次 429。 +限流必须位于「Worker → Rate Limiter → Channel」，而不是靠重试兜底。 +（Nginx 只能管入站，管不到出站。） + +设计： + - 速率按「条/分钟」配置，存 channel_rate_limit 表；0 或缺省 = 不限流 + - 桶容量 = 1 分钟额度，允许小幅突发 + - acquire() 最多等 max_wait 秒；拿不到令牌就返回 False， + 由调用方「延迟重排」（queue.defer）而不是长阻塞—— + 否则单通道限流会把 worker 线程占住，拖累其它通道。 +""" + +import os +import time +import threading + +import log +from db.connection import _conn + +MAX_WAIT_SECONDS = float(os.getenv("EGO_RATE_MAX_WAIT", "1.0")) +_SLEEP_SLICE = 0.1 +# 未配置限流的通道，多久回查一次数据库。 +# 缓存只在 set_rate() 时更新，若有人直接改库（或另一个进程写入）， +# 这个 TTL 让配置在 30s 内自愈，而不必重启服务。 +_MISS_TTL = float(os.getenv("EGO_RATE_MISS_TTL", "30")) + + +class _Bucket: + __slots__ = ("rate", "capacity", "tokens", "updated") + + def __init__(self, rate): + self.rate = rate + self.capacity = float(rate) # 桶容量 = 1 分钟额度 + self.tokens = float(rate) + self.updated = time.time() + + +class RateLimiter: + def __init__(self): + self._lock = threading.Lock() + self._rates = {} # channel_id -> per_minute + self._buckets = {} + self._checked_at = {} # channel_id -> 上次回查数据库的时间 + self._loaded = False + + # ── 配置 ── + def load(self): + """从 SQLite 载入各通道速率。""" + try: + rows = _conn().execute( + "SELECT channel_id, per_minute FROM channel_rate_limit" + ).fetchall() + except Exception as e: + log.logger.warning(f"[RateLimit] load failed: {e}") + self._loaded = True + return + with self._lock: + self._rates = {r["channel_id"]: int(r["per_minute"] or 0) for r in rows} + self._checked_at = {cid: time.time() for cid in self._rates} + self._buckets.clear() + self._loaded = True + active = {k: v for k, v in self._rates.items() if v > 0} + if active: + log.logger.info(f"[RateLimit] Loaded limits: {active}") + + def get_rate(self, channel_id): + """取通道速率（条/分钟）。未配置的通道按 TTL 回查数据库，避免缓存僵化。""" + now = time.time() + rate = self._rates.get(channel_id) + if rate is None or (rate == 0 and now - self._checked_at.get(channel_id, 0) > _MISS_TTL): + try: + row = _conn().execute( + "SELECT per_minute FROM channel_rate_limit WHERE channel_id=?", + (channel_id,) + ).fetchone() + rate = int(row["per_minute"] or 0) if row else 0 + except Exception: + rate = 0 + self._rates[channel_id] = rate + self._checked_at[channel_id] = now + return rate + + def set_rate(self, channel_id, per_minute): + """设置通道速率（条/分钟），0 = 不限流。""" + per_minute = max(0, int(per_minute)) + conn = _conn() + conn.execute( + """INSERT INTO channel_rate_limit (channel_id, per_minute, updated_at) + VALUES (?,?,datetime('now')) + ON CONFLICT(channel_id) DO UPDATE SET + per_minute=excluded.per_minute, updated_at=excluded.updated_at""", + (channel_id, per_minute) + ) + conn.commit() + with self._lock: + self._rates[channel_id] = per_minute + self._checked_at[channel_id] = time.time() + self._buckets.pop(channel_id, None) + log.logger.info(f"[RateLimit] Channel {channel_id} limit set to {per_minute}/min") + + # ── 取令牌 ── + def _bucket(self, channel_id, rate): + b = self._buckets.get(channel_id) + if b is None or b.rate != rate: + b = _Bucket(rate) + self._buckets[channel_id] = b + return b + + def acquire(self, channel_id, max_wait=None): + """尝试取一个令牌。返回 True/False（False = 调用方应延迟重排）。""" + rate = self.get_rate(channel_id) + if rate <= 0: + return True + if max_wait is None: + max_wait = MAX_WAIT_SECONDS + + per_sec = rate / 60.0 + deadline = time.time() + max_wait + + while True: + with self._lock: + b = self._bucket(channel_id, rate) + now = time.time() + b.tokens = min(b.capacity, b.tokens + (now - b.updated) * per_sec) + b.updated = now + if b.tokens >= 1.0: + b.tokens -= 1.0 + return True + need = (1.0 - b.tokens) / per_sec + + if time.time() + need > deadline: + return False + time.sleep(min(need, _SLEEP_SLICE)) + + def snapshot(self): + with self._lock: + return {k: v for k, v in self._rates.items() if v > 0} + + +# ── 单例 ── +_limiter = None + + +def get_limiter(): + global _limiter + if _limiter is None: + _limiter = RateLimiter() + return _limiter diff --git a/requirements.txt b/requirements.txt index ce7d0be..8685899 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -flask>=3.0,<3.1 +flask>=3.0,<3.1 werkzeug>=3.0,<3.1 requests>=2.28 jinja2>=3.1 diff --git a/sender_engine/__init__.py b/sender_engine/__init__.py index 65fc3db..362f98f 100644 --- a/sender_engine/__init__.py +++ b/sender_engine/__init__.py @@ -16,49 +16,105 @@ import db import bus import renderer +import circuit_breaker +import rate_limiter from channel_loader import create_channel from queue_backend import get_backend +# 熔断/限流命中时的延迟重排间隔（秒）——不消耗重试次数，见 queue_backend.defer() +CIRCUIT_DEFER_SECONDS = 10 +RATE_DEFER_SECONDS = 5 + + +def _binding_dedup_key(sc, msg): + """算某条绑定的去重键（未配置去重则返回空串）。""" + expr = (sc.get("dedup_key_expr") or "").strip() + if not expr: + return "" + try: + return _eval_dedup_key(expr, msg) or "" + except Exception as e: + log.logger.warning(f"Dedup key eval failed ({expr}): {e}") + return "" + + +def _record_dedup(channel_id, key): + """发送成功后记录去重键（供后续窗口判定）。""" + if not key: + return + try: + db.dedup_record(channel_id, key) + except Exception as e: + log.logger.warning(f"Dedup record failed (ch={channel_id}): {e}") + + +def _plan_dedup(matched, msg, trace_id): + """逐绑定做**渠道级去重**判定，返回 (plan, skipped, first_key)。 + + - plan: [(binding, dedup_key)] —— 需要发送的渠道 + - skipped: 被去重拦下的渠道数 + - first_key: 首个去重键（写入 message_log.dedup_key，仅供展示） + + 异步入队路径与直接发送路径**共用**本函数，避免同一套判定逻辑写两遍 + （improvement #31）。 + """ + plan = [] + skipped = 0 + first_key = "" + for sc in matched: + key = _binding_dedup_key(sc, msg) + if key: + first_key = first_key or key + window = sc.get("dedup_window", 3600) or 3600 + if db.dedup_hit(sc["channel_id"], key, window): + skipped += 1 + log.logger.info( + f"[{trace_id}] Dedup hit (ch={sc['channel_id']}): {key} within {window}s" + ) + continue + plan.append((sc, key)) + return plan, skipped, first_key + def _on_message_routed(sender, *, trace_id, source_id, msg, matched_channels): """ message.routed 事件处理器（Webhook 路径）。 - 去重检查后，将每个通道的发送任务入队，HTTP 立即返回。 + + **渠道级去重**：逐绑定计算并判定各自的去重键，命中的只跳过该渠道， + 其余渠道照常发送；只有全部渠道都被命中时，整条消息才标记 DISCARDED。 + + （旧实现取「第一个配了去重表达式的绑定」算出一个键用于所有渠道， + 一旦命中就把整条消息丢掉，且其余绑定的去重表达式被完全忽略。） """ - # 去重检查 - dedup_key = None - for sc in matched_channels: - dedup_expr = (sc.get("dedup_key_expr") or "").strip() - if dedup_expr: - dedup_key = _eval_dedup_key(dedup_expr, msg) - if dedup_key: - window = sc.get("dedup_window", 3600) or 3600 - if db.check_dedup(dedup_key, window): - log.logger.info(f"[{trace_id}] Dedup hit: key={dedup_key}") - db.update_message(trace_id, status="DISCARDED", dedup_key=dedup_key, - error=f"Dedup hit: {dedup_key} within {window}s") - return True, msg - break - - # 更新消息状态 + plan, skipped, first_key = _plan_dedup(matched_channels, msg, trace_id) + msg_json = json.dumps(msg, ensure_ascii=False) - db.update_message(trace_id, status="SENDING", msg_json=msg_json, dedup_key=dedup_key or "") - # 每个通道入队一条任务 + if not plan: + db.update_message(trace_id, status="DISCARDED", dedup_key=first_key, + error=f"Dedup hit on all {skipped} channel(s)") + log.logger.info(f"[{trace_id}] All {skipped} channel(s) deduped, discarded") + return True, msg + + db.update_message(trace_id, status="SENDING", msg_json=msg_json, dedup_key=first_key) + bus.emit(bus.message_sending, trace_id=trace_id, source_id=source_id, msg=msg, + channels=[sc["channel_id"] for sc, _ in plan]) + queue = get_backend() - for sc in matched_channels: + for sc, key in plan: queue.enqueue( trace_id=trace_id, source_id=source_id, msg_json=msg_json, channel_id=sc["channel_id"], template_id=sc["template_id"], - dedup_key=dedup_key or "", + dedup_key=key, max_retries=3, ) log.logger.info( - f"[{trace_id}] Enqueued {len(matched_channels)} channel(s) for async send" + f"[{trace_id}] Enqueued {len(plan)} channel(s) for async send" + + (f", {skipped} deduped" if skipped else "") ) return True, msg @@ -87,6 +143,18 @@ def process_queue_item(item): ch_name = ch["name"] ch_type = ch["type"] + # 熔断闸门：通道处于 OPEN 时直接延迟重排，不消耗重试次数 + # （故障期内让消息留在队列里等恢复，而不是被重试耗尽跌进死信队列） + breaker = circuit_breaker.get_breaker() + allowed, reason = breaker.should_allow(channel_id) + if not allowed: + log.logger.warning(f"[{trace_id}] Circuit open for {ch_name} ({reason}), deferring") + return False, { + "ch_name": ch_name, "ch_type": ch_type, "ok": False, + "error": f"Circuit open: {reason}", + "deferred": True, "defer_seconds": CIRCUIT_DEFER_SECONDS, + } + # 解析消息 try: msg = json.loads(msg_json) if isinstance(msg_json, str) else msg_json @@ -111,12 +179,25 @@ def process_queue_item(item): "ok": False, "error": f"Render: {str(e)[:200]}" } + # 出站限流：拿不到令牌就延迟重排（令牌在"即将真正发送"时才取） + limiter = rate_limiter.get_limiter() + if not limiter.acquire(channel_id): + log.logger.info(f"[{trace_id}] Rate limited on {ch_name}, deferring") + return False, { + "ch_name": ch_name, "ch_type": ch_type, "ok": False, + "error": f"Rate limited (limit {limiter.get_rate(channel_id)}/min)", + "deferred": True, "defer_seconds": RATE_DEFER_SECONDS, + } + # 发送 try: ch_config = json.loads(ch["config"]) if isinstance(ch["config"], str) else ch["config"] channel = create_channel(ch_type, ch_config) ok, err = channel.send(rendered["title"], rendered["content"]) + breaker.record(channel_id, ok, err or "") if ok: + # 渠道级去重：发送成功才记录本渠道自己的去重键 + _record_dedup(channel_id, item.get("dedup_key")) log.logger.info(f"[{trace_id}] Sent via {ch_name}") return True, { "ch_name": ch_name, "ch_type": ch_type, "ok": True, "error": None @@ -128,6 +209,7 @@ def process_queue_item(item): "ok": False, "error": err or "Send returned False" } except Exception as e: + breaker.record(channel_id, False, str(e)) log.logger.error(f"[{trace_id}] Send error ({ch_name}): {e}") return False, { "ch_name": ch_name, "ch_type": ch_type, @@ -135,14 +217,43 @@ def process_queue_item(item): } +# ── 结果回写的并发保护（P1-3）── +# channel_results 是「读 → 改 → 写」的 JSON 列，多 worker 并发回写会丢失更新 +# （lost update）；「查 pending → 置终态」也存在 check-then-act 竞争。 +# 用条带锁（固定 64 把，按 trace_id 取模）把同一 trace 的回写串行化： +# 不增长、无需清理；不同 trace 偶有共享锁只影响少量并行度，不影响正确性。 +_RESULT_LOCKS = [threading.Lock() for _ in range(64)] + + +def _result_lock(trace_id): + return _RESULT_LOCKS[hash(str(trace_id)) % len(_RESULT_LOCKS)] + + def update_message_results(trace_id, channel_result): """ 追加一个通道的发送结果到 message_log，并判断是否所有通道都完成了。 所有通道完成后更新整体状态。 + + 同一 trace 的回写用条带锁串行化，避免多 worker 下丢失更新。 + 终态事件（message.sent / message.failed）在**释放锁之后**再 emit， + 避免订阅者在同一条 trace 上回调本函数造成自锁。 """ + with _result_lock(trace_id): + terminal = _update_message_results_locked(trace_id, channel_result) + + if terminal == "sent": + bus.emit(bus.message_sent, trace_id=trace_id) + elif terminal == "failed": + rec = db.get_message(trace_id) + bus.emit(bus.message_failed, trace_id=trace_id, stage="send", + error=(rec or {}).get("error", "")) + + +def _update_message_results_locked(trace_id, channel_result): + """返回终态字符串 'sent' / 'failed'，未到终态返回 None。""" rec = db.get_message(trace_id) if not rec: - return + return None # 解析已有结果 try: @@ -160,19 +271,22 @@ def update_message_results(trace_id, channel_result): cr_json = json.dumps(results, ensure_ascii=False) if pending_for_trace == 0: - # 所有通道都完成了 + # 所有通道都完成了 —— 消息到达终态，在总线生命周期里收口 all_ok = all(r.get("ok") for r in results) if all_ok: db.update_message(trace_id, status="SUCCESS", channel_results=cr_json, sent_at=dt_now_str()) + return "sent" else: failed_names = _summarize_failures(results) db.update_message(trace_id, status="FAILED", channel_results=cr_json, error=f"Failed: {failed_names}") + return "failed" else: # 还有任务在处理中，只更新 channel_results db.update_message(trace_id, channel_results=cr_json) + return None def _count_pending_for_trace(trace_id): @@ -199,28 +313,27 @@ def send_to_channels(trace_id, source_id, msg, matched): def _do_send_direct(trace_id, source_id, msg, matched): - """直接并行发送所有通道（绕过队列）。""" - db.update_message(trace_id, status="SENDING") + """直接并行发送所有通道（绕过队列）。 - # 去重 - dedup_key = None - for sc in matched: - dedup_expr = (sc.get("dedup_key_expr") or "").strip() - if dedup_expr: - dedup_key = _eval_dedup_key(dedup_expr, msg) - if dedup_key: - window = sc.get("dedup_window", 3600) or 3600 - if db.check_dedup(dedup_key, window): - db.update_message(trace_id, status="DISCARDED", dedup_key=dedup_key, - error=f"Dedup hit: {dedup_key} within {window}s") - return True, msg - break + 渠道级去重：逐绑定判定，命中的只跳过该渠道；全部命中才整条 DISCARDED。 + 发送成功时记录该渠道自己的去重键。 + """ + plan, skipped, first_key = _plan_dedup(matched, msg, trace_id) + + if not plan: + db.update_message(trace_id, status="DISCARDED", dedup_key=first_key, + error=f"Dedup hit on all {skipped} channel(s)") + return True, msg + + db.update_message(trace_id, status="SENDING", dedup_key=first_key) + bus.emit(bus.message_sending, trace_id=trace_id, source_id=source_id, msg=msg, + channels=[sc["channel_id"] for sc, _ in plan]) channel_results = [] all_ok = True result_lock = threading.Lock() - def _send_one(sc): + def _send_one(sc, dedup_key): nonlocal all_ok tmpl = db.get_template(sc["template_id"]) ch = db.get_channel(sc["channel_id"]) @@ -229,7 +342,9 @@ def _send_one(sc): ch_name = ch["name"] ch_type = ch["type"] - result = {"ch_name": ch_name, "ch_type": ch_type, "ok": False, "error": None} + # channel_id 是渠道级重发/去重的定位依据，必须随结果落库 + result = {"ch_name": ch_name, "ch_type": ch_type, "ok": False, "error": None, + "channel_id": sc["channel_id"]} try: rendered = renderer.render_template( @@ -253,6 +368,8 @@ def _send_one(sc): result["error"] = err or "Send returned False" with result_lock: all_ok = False + else: + _record_dedup(sc["channel_id"], dedup_key) except Exception as e: result["error"] = str(e)[:500] with result_lock: @@ -261,7 +378,7 @@ def _send_one(sc): return result with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool: - futures = [pool.submit(_send_one, sc) for sc in matched] + futures = [pool.submit(_send_one, sc, key) for sc, key in plan] for f in concurrent.futures.as_completed(futures): r = f.result() if r is not None: @@ -270,17 +387,25 @@ def _send_one(sc): cr_json = json.dumps(channel_results, ensure_ascii=False) if all_ok: db.update_message(trace_id, status="SUCCESS", channel_results=cr_json, - dedup_key=dedup_key, sent_at=dt_now_str()) + dedup_key=first_key, sent_at=dt_now_str()) + bus.emit(bus.message_sent, trace_id=trace_id, source_id=source_id) else: + failed = _summarize_failures(channel_results) db.update_message(trace_id, status="FAILED", channel_results=cr_json, - dedup_key=dedup_key, - error=f"Some channels failed: {_summarize_failures(channel_results)}") + dedup_key=first_key, error=f"Some channels failed: {failed}") + bus.emit(bus.message_failed, trace_id=trace_id, stage="send", error=failed) return all_ok, msg def dt_now_str(): - return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + """返回 **UTC** 时间字符串，与 SQLite 的 `CURRENT_TIMESTAMP` 保持一致。 + + 原先用 `datetime.now()`（本地时间）写入 sent_at，而 created_at 由 + `CURRENT_TIMESTAMP` 生成（UTC），两者相差一个时区偏移—— + 消息列表里"创建时间/发送时间"会对不上，按二者计算的延迟也会错好几小时。 + """ + return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M:%S") def _summarize_failures(channel_results): diff --git a/source_listener/__init__.py b/source_listener/__init__.py index 89c8f77..bf1c909 100644 --- a/source_listener/__init__.py +++ b/source_listener/__init__.py @@ -6,7 +6,9 @@ 收到请求后通过事件总线触发后续处理链路。 """ +import os import threading +import collections import json import log import db @@ -17,6 +19,18 @@ # ── 安全限制 ────────────────────────────── MAX_BODY_SIZE = 5 * 1024 * 1024 # 5 MB +# ── 入口并发 ────────────────────────────── +# 每个数据源一个固定大小的工作线程池。线程复用使 db/connection.py 的 +# threading.local() 连接随之复用，而不是「每请求新建一个 SQLite 连接」。 +# +# 关于默认值 8（实测依据，2026-07-28，测试机 4 核）： +# 吞吐在并发 1 时最高（≈48 req/s），并发升高不再提升（受单进程串行段限制）； +# 而 p50 延迟随并发线性增长（1→21ms, 8→131ms, 16→218ms）。 +# 线程池的意义是「慢请求不阻塞其它请求」，而非提升吞吐，故取较小值以压低延迟。 +INGRESS_WORKERS = int(os.getenv("EGO_INGRESS_WORKERS", "8")) +# 等待队列上限，超过则直接回 503（背压），防止请求无限堆积 +INGRESS_MAX_QUEUE = int(os.getenv("EGO_INGRESS_MAX_QUEUE", "200")) + # ── 样本数据存储 ────────────────────────────── _sample_store = {} _sample_lock = threading.Lock() @@ -118,6 +132,118 @@ def log_message(self, format, *args): pass +# ── 入口工作线程池 ────────────────────────── + +class _IngressPool: + """固定大小的工作线程池（守护线程 + 有界队列）。 + + 相比 socketserver 的「每请求开一个新线程」： + - 线程复用 → db/connection.py 的 threading.local() 连接被复用； + - 守护线程 → 进程退出不被在途请求阻塞（不会引入 atexit join 延迟）； + - 队列有界 → submit() 在满载时返回 False，由调用方回 503 做背压。 + """ + + def __init__(self, max_workers, max_queue): + self._max_queue = max_queue + self._queue = collections.deque() + self._cond = threading.Condition() + self._stopped = False + for i in range(max_workers): + threading.Thread( + target=self._worker, daemon=True, name=f"ingress-{i}" + ).start() + + def submit(self, fn, *args): + """入队。队列已满返回 False（调用方据此拒绝请求）。""" + with self._cond: + if self._stopped or len(self._queue) >= self._max_queue: + return False + self._queue.append((fn, args)) + self._cond.notify() + return True + + def _worker(self): + while True: + with self._cond: + while not self._queue and not self._stopped: + self._cond.wait() + if self._stopped: + return + fn, args = self._queue.popleft() + try: + fn(*args) + except Exception as e: + log.logger.warning(f"Ingress worker error: {e}") + + def shutdown(self): + with self._cond: + self._stopped = True + self._cond.notify_all() + + +# ── 入口 HTTP 服务（固定线程池） ────────────── + +class _ThreadPoolHTTPServer(HTTPServer): + """固定大小线程池的 HTTP 服务。 + + 原实现用单线程 HTTPServer：同一数据源上的请求被**串行**处理， + 只要有一个慢解析器，该数据源上所有后续请求都要排队等待。 + 改为线程池后，慢请求不再阻塞其它请求。 + + （路径路由走 Flask，`run_simple(threaded=True)` 本就并发，不受此影响。） + """ + + # HTTPServer 默认 allow_reuse_address=1；此处显式声明以免误改 + allow_reuse_address = True + # 监听队列（accept backlog）。默认值仅 5，突发流量下内核会拒掉多余连接， + # 客户端只能等 TCP SYN 重传（约 1s）——实测并发 16/32 时出现 ~1.1s 长尾。 + request_queue_size = 128 + + def __init__(self, server_address, handler_cls, + max_workers=INGRESS_WORKERS, max_queue=INGRESS_MAX_QUEUE): + super().__init__(server_address, handler_cls) + self._pool = _IngressPool(max_workers, max_queue) + + def process_request(self, request, client_address): + if not self._pool.submit(self._process_request_thread, request, client_address): + log.logger.warning( + f"Source [{getattr(self, 'source_id', '?')}] ingress queue full " + f"({INGRESS_MAX_QUEUE}), rejecting connection from {client_address}" + ) + try: + request.sendall( + b"HTTP/1.0 503 Service Unavailable\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n\r\n" + ) + except Exception: + pass + self.shutdown_request(request) + return + + def _process_request_thread(self, request, client_address): + try: + self.finish_request(request, client_address) + except Exception: + self.handle_error(request, client_address) + finally: + self.shutdown_request(request) + + def handle_error(self, request, client_address): + """覆盖默认实现：异常写入应用日志，而非直接打到 stderr。""" + import traceback + log.logger.warning( + f"Source [{getattr(self, 'source_id', '?')}] unhandled error in request " + f"from {client_address}: {traceback.format_exc()}" + ) + + def server_close(self): + try: + super().server_close() + finally: + self._pool.shutdown() + + # ── ListenerManager ────────────────────────── class ListenerManager: @@ -143,7 +269,7 @@ def start_source(self, source_id): self.stop_source(source_id) try: - server = HTTPServer(("0.0.0.0", src["port"]), _HookHandler) + server = _ThreadPoolHTTPServer(("0.0.0.0", src["port"]), _HookHandler) server.source_id = source_id server.timeout = 10 # 读取超时 10 秒 self._servers[source_id] = server @@ -165,8 +291,9 @@ def start_source(self, source_id): def stop_source(self, source_id): """停止单个数据源的 HTTP 监听。""" if source_id in self._servers: - self._servers[source_id].shutdown() - del self._servers[source_id] + server = self._servers.pop(source_id) + server.shutdown() # 停止 serve_forever 循环 + server.server_close() # 释放监听 socket 与工作线程池 if source_id in self._threads: del self._threads[source_id] log.logger.info(f"Source {source_id} stopped") diff --git a/source_manager.py b/source_manager.py index 593f122..9d00525 100644 --- a/source_manager.py +++ b/source_manager.py @@ -66,41 +66,27 @@ def process_message(source_id, raw_body: bytes, headers: dict, query_params: dic # 1. 记录原始消息 db.create_message_log(trace_id, source_id, src_name, raw_str, "RECEIVED") - # 2. 解析（parser_engine 监听 message.received） + # 2. 事件总线驱动全链路：解析 → 路由 → 入队 + # message.received → parser_engine → message.parsed → router_engine + # → message.routed → sender_engine + # + # ⚠️ 修复记录（2026-07-28）：原实现除事件链外，自己又重复 emit 了 + # message.parsed 与 message.routed，导致**同一条消息被投递 3 次** + # （解析链内各发一次 + 这里再发两次）。该 bug 自 v1.1.0（commit 47ac7f9a） + # 起一直存在，回归测试见 tests/test_event_chain.py。 + # extra_fields（如 sub_path）改由 parser_engine 在路由之前合并。 results = bus.emit( bus.message_received, trace_id=trace_id, source_id=source_id, raw_body=raw_body, headers=headers, query_params=query_params, + extra_fields=extra_fields, ) - parse_ok, msg = _extract_result(results) + parse_ok, msg = _extract_result(results) or (False, None) if not parse_ok: return False, None - # 2.5 合并额外字段（如路径路由的 sub_path） - if extra_fields and isinstance(msg, dict): - msg.update(extra_fields) - if isinstance(msg, dict): - msg["_trace_id"] = trace_id - - # 3. 路由（router_engine 监听 message.parsed） - results = bus.emit( - bus.message_parsed, - trace_id=trace_id, source_id=source_id, msg=msg, - ) - route_result = _extract_result(results) - if route_result is None: - return True, msg - - matched, msg = route_result - - # 4. 发送（sender_engine 监听 message.routed） - results = bus.emit( - bus.message_routed, - trace_id=trace_id, source_id=source_id, - msg=msg, matched_channels=matched, - ) - send_result = _extract_result(results) - return (send_result[0] if send_result else True), msg + # 3. 路由与入队已在事件链内完成，此处不再重复触发 + return True, msg def _extract_result(results): @@ -146,10 +132,34 @@ def flush_queue_for_source(source_id): # ── 重发 ────────────────────────────────────── -def retry_message(msg_id, mode="original"): +def _failed_channel_ids(rec): + """从 channel_results 取上次**失败**的 channel_id 集合。 + + 返回 None 表示无法判定（旧记录没有 channel_id、或没有结果）， + 调用方应回退到"整条重发"的旧行为。 + """ + try: + results = json.loads(rec.get("channel_results") or "[]") + except Exception: + return None + if not results: + return None + ids, has_id = set(), False + for r in results: + if "channel_id" in r: + has_id = True + if not r.get("ok"): + ids.add(r["channel_id"]) + return ids if has_id else None + + +def retry_message(msg_id, mode="original", scope="failed"): """ 重发一条失败消息。 - mode: "original" = 用存好的 msg_json 重发; "rerender" = 重新解析 raw_body。 + + mode: "original" = 用存好的 msg_json 重发; "rerender" = 重新解析 raw_body + scope: "failed"（默认）= **只重发上次失败的渠道**，避免把已成功的渠道重复推送； + "all" = 重发全部匹配渠道（旧行为，用于确实想整体重推的场景） """ rec = db.get_message_by_id(msg_id) if not rec: @@ -192,5 +202,26 @@ def retry_message(msg_id, mode="original"): if not matched: return False, "No matching channels" + # ── 渠道级重发：只挑上次失败的渠道 ── + if scope != "all": + failed_ids = _failed_channel_ids(rec) + if failed_ids is None: + log.logger.info( + f"[Retry #{msg_id}] channel_results 缺少 channel_id，回退为整条重发" + ) + elif not failed_ids: + return False, "上次所有渠道均已成功，无需重发（如需整体重推请用 scope=all）" + else: + targets = [sc for sc in matched if sc["channel_id"] in failed_ids] + if not targets: + return False, ("上次失败的渠道已不在当前匹配规则中" + "（绑定可能已改动），无法只重发失败渠道") + log.logger.info( + f"[Retry #{msg_id}] Channel-level retry: " + f"{[sc['channel_id'] for sc in targets]} of " + f"{[sc['channel_id'] for sc in matched]} matched" + ) + matched = targets + ok, _ = sender_engine.send_to_channels(rec["trace_id"], rec["source_id"], msg, matched) return ok, None diff --git a/templates/channels_page.html b/templates/channels_page.html index aafe36f..3572086 100644 --- a/templates/channels_page.html +++ b/templates/channels_page.html @@ -62,6 +62,7 @@

{{ _("ch.title") }}

{{ _("ch.col_name") }} {{ _("ch.col_type") }} {{ _("ch.col_status") }} + {{ _("ch.col_resilience") }} {{ _("ch.col_actions") }} @@ -79,6 +80,7 @@

{{ _("ch.title") }}

{{ _("common.disabled") }} {% endif %} + — @@ -176,6 +178,12 @@