refactor(pymsg): 新增AsyncRpcQueue通用异步队列，log_client基于此重构

- pymsg新增AsyncRpcQueue：提供内存队列+后台线程+攒批处理的通用异步基础设施，支持两种模式：简单RPC模式(client+method+serialize)和自定义handler模式 - confs/dnfs/static的log_client.py重构为基于AsyncRpcQueue的薄封装，保持API完全向后兼容(push/info/error/warning/debug/critical + 全局logs单例) - 消除了三份log_client副本中重复的队列/线程/攒批逻辑(约150行/份)，代码量净减少约50行 - LogClient的行为与旧版一致：逐条INSERT到db、fail-fast网络错误、队列满丢最旧
feat(pymsg): RpcClient 新增 retry_on_refused 阻塞等待模式，解决微服务启动顺序依赖
2026-08-01 12:59:33 +08:00 · 2026-08-01 12:47:08 +08:00 · 2026-08-01 12:38:51 +08:00 · 2026-08-01 12:37:19 +08:00 · 2026-08-01 12:34:53 +08:00 · 2026-08-01 12:33:51 +08:00
21 changed files with 923 additions and 599 deletions
@@ -227,7 +227,7 @@ except RpcError as e:

 ## 集成关系

- 由 `core`（`PMC`）通过 `init.txt` 中的 `bash /opt/micro_servers/confs/_run.sh` 拉起并守护（在 db 之后、ai_review 之前启动）
+- 由 `core`（`PMC`）通过 `init.txt` 中的 `bash /opt/micro_servers/confs/_run.sh` 拉起并守护（confs 最先启动即可；各业务服务通过 pymsg `retry_on_refused=True` 在拉取配置时自动阻塞等待 confs 就绪，不严格要求启动顺序；confs 日志推送 db 为异步 fail-safe，db 不可用时静默丢弃，不阻塞 confs 启动）
 - 各内部服务通过 pymsg `RpcClient` 调 `get` 路由按需获取配置，替代本地散落的 ini/json/yaml
 - 自身日志通过 `log_client.py` 异步推送至 `db:8005` 的 Logs 表（与其他服务一致）

@@ -235,21 +235,35 @@ except RpcError as e:

 | 服务 | 配置文件 | 接入方式 |
 |------|---------|---------|
-| [static](../static/) | [static.ini](configs/static.ini) | 三层 fail-safe：内置默认值 → confs（优先）→ 本地 `config.ini` 回退 |
-| [dnfs](../dnfs/) | [dnfs.ini](configs/dnfs.ini) | 三层 fail-safe：内置默认值 → confs（优先）→ 环境变量覆盖 |
+| [static](../static/) | [static.ini](configs/static.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） |
+| [dnfs](../dnfs/) | [dnfs.ini](configs/dnfs.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） |
+| [db](../db/) | [db.ini](configs/db.ini) | confs 强依赖（pymsg 自动阻塞等待 confs 就绪）→ 环境变量覆盖（最高优先级，Docker 兼容） |

 static 的接入实现见 [static/fsys.py](../static/fsys.py)：
- `_fetch_from_confs()` 通过 `RpcClient` 调 `get` 路由取 `static.ini`，3 秒超时
- `CONFS_HOST` / `CONFS_PORT` 环境变量定位 confs（默认 `127.0.0.1:8010`）
- `_merge_ini()` 仅覆盖已知的 section/key（白名单合并，不引入未预期字段）
- 配置来源会记录到日志（`logs.info/warning`），便于排查实际用了哪一层
+- 两层加载顺序：confs（static.ini，强依赖）→ 环境变量覆盖（优先级最高）
+- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪（每秒重试，每 5 秒打印等待提示），不会因启动顺序导致进程退出
+- static.ini 缺失必需配置项时仍然报错退出（配置错误，非启动时序问题）
+- 配置项涵盖：`[server]` host/port、`[directory]` root、`[ssl]` enabled/cert_file/key_file（ssl 有默认值）、`[cors]` allowed_origins
+- confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010`
+- 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`STATIC_ROOT`/`CONFS_HOST`/`CONFS_PORT`
+- 已删除 `--config` 命令行参数和本地 `config.ini` 回退逻辑

 dnfs 的接入实现见 [dnfs/dnfs.py](../dnfs/dnfs.py)：
- 三层加载顺序：内置默认值 → confs（dnfs.ini）→ 环境变量覆盖（优先级最高）
+- 两层加载顺序：confs（dnfs.ini，强依赖）→ 环境变量覆盖（优先级最高）
+- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，不严格要求启动顺序
+- 配置项涵盖：`[server]` host/port、`[database]` host/port/timeout、`[ai_review]` url/timeout
+- confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010`
 - 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`DB_HOST`/`DB_PORT`/`REVIEW_URL`/`CONFS_HOST`/`CONFS_PORT`
- Docker 环境通过环境变量传入服务名（db/ai_review），覆盖 confs 中的 127.0.0.1 默认值

-其他服务（gateway / waf / ai_review / smtp / db）尚未接入，仍读各自本地配置。
+db 的接入实现见 [db/dbv2.py](../db/dbv2.py)：
+- 两层加载顺序：confs（db.ini，强依赖）→ 环境变量覆盖（优先级最高）
+- pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，不严格要求启动顺序
+- 配置项涵盖：`[server]` host/port/workers、`[database]` url、`[logs]` retention_days/max_records/cleanup_interval
+- confs 连接地址为引导配置：环境变量 `CONFS_HOST`/`CONFS_PORT` 优先，默认 `127.0.0.1:8010`
+- 支持的环境变量：`LISTEN_ADDR`/`LISTEN_PORT`/`WORKERS`/`DATABASE_URL`/`CONFS_HOST`/`CONFS_PORT`
+- 独立运维工具 `db_tools.py`（backup/export/import 等）仍直接读 `DATABASE_URL` 环境变量，不依赖 confs，避免运维脚本依赖服务就绪
+
+其他服务（gateway / waf / ai_review / smtp）尚未接入，仍读各自本地配置。

 ## 项目结构

@@ -260,7 +274,8 @@ confs/
 ├── configs/             # 配置文件存放目录（被分发对象）
 │   ├── example.ini      # 示例配置
 │   ├── static.ini       # static 服务配置（已接入分发）
-│   └── dnfs.ini         # dnfs 服务配置（已接入分发）
+│   ├── dnfs.ini         # dnfs 服务配置（已接入分发）
+│   └── db.ini           # db 服务配置（已接入分发）
 ├── docs/
 │   └── SECURITY_AUDIT_20260801.md  # 安全审计报告
 ├── _set.sh              # 初始化脚本（建 venv + 安装依赖）
@@ -0,0 +1,16 @@
+[server]
+host = 0.0.0.0
+port = 8005
+workers = 4
+
+[database]
+url = sqlite:///database.db
+
+[logs]
+retention_days = 30
+max_records = 1000000
+cleanup_interval = 3600
+
+[confs]
+host = 127.0.0.1
+port = 8010
@@ -10,7 +10,3 @@ timeout = 10
 [ai_review]
 url = http://127.0.0.1:8007/review
 timeout = 10
-
-[confs]
-host = 127.0.0.1
-port = 8010
@@ -1,13 +1,15 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
+"""
+统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。

 设计要点：
  - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
-  - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
+  - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
  - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
-  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
+  - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
+  - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施

-用法：
+用法（与旧版完全兼容）：
    from log_client import logs

    logs.info("dnfs", "节点新增成功 id=42")
@@ -19,28 +21,25 @@
 import json
 import os as _os
 import sys as _sys
-import threading
-import queue
-import time
 import atexit
 from datetime import datetime, timezone

 _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
-from pymsg import RpcClient, RpcError  # noqa: E402
+from pymsg import RpcClient, RpcError, AsyncRpcQueue  # noqa: E402

 # ── 配置 ──────────────────────────────────────────────────────────────

 _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
 _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))

-_BATCH_SIZE = 50         # 队列满 50 条时 flush
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
-_TIMEOUT = 5             # socket 超时（秒）
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_BATCH_SIZE = 50
+_FLUSH_INTERVAL = 2.0
+_TIMEOUT = 5
+_QUEUE_MAXSIZE = 10000


 def _safe_json(obj):
-    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
    try:
        return json.dumps(obj, ensure_ascii=False, default=str)
    except Exception:
@@ -48,163 +47,80 @@ def _safe_json(obj):


 def _now_iso():
-    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
    return datetime.now(timezone.utc).isoformat()


 class LogClient:
-    """异步批量日志推送客户端。"""
+    """异步批量日志推送客户端（API 与旧版完全兼容）。"""

-    def __init__(self, service_name="log",
-                 batch_size=_BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+    def __init__(self, service_name: str = "log",
+                 batch_size: int = _BATCH_SIZE,
+                 flush_interval: float = _FLUSH_INTERVAL):
        self._service_name = service_name
-        self._batch_size = batch_size
-        self._flush_interval = flush_interval
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
-        self._stop = threading.Event()
-        self._flush_signal = threading.Event()
-        self._flush_done = threading.Event()
-        self._thread = threading.Thread(target=self._flush_loop,
-                                        daemon=True, name='log-client')
-        self._thread.start()
-        atexit.register(self._shutdown)
+        self._client = RpcClient(
+            _DBV2_ADDR, _DBV2_PORT,
+            service_name + "-log",
+            timeout=_TIMEOUT,
+        )
+        self._queue = AsyncRpcQueue(
+            handler=self._send_batch,
+            batch_size=batch_size,
+            flush_interval=flush_interval,
+            max_queue_size=_QUEUE_MAXSIZE,
+            on_error=self._on_error,
+        )
+        self._queue.start()
+        atexit.register(self.close)

    # ── 公开接口 ────────────────────────────────────────────────────────

-    def push(self, service, level, message, trace_id=None, extra=None):
+    def push(self, service: str, level: str, message: str,
+             trace_id: str = None, extra=None):
        """推送一条日志（非阻塞，立即返回）。

        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
        """
        entry = {
-            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
+            'ts': _now_iso(),
            'service': service,
            'level': level,
            'message': message,
            'trace_id': trace_id,
            'extra': extra,
        }
-        try:
-            self._queue.put_nowait(entry)
-        except queue.Full:
-            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
-            try:
-                self._queue.get_nowait()
-            except queue.Empty:
-                pass
-            try:
-                self._queue.put_nowait(entry)
-            except queue.Full:
-                pass  # 极端竞争下仍失败则静默丢弃
+        self._queue.put(entry)

-    def debug(self, service, message, **kw):
+    def debug(self, service: str, message: str, **kw):
        self.push(service, 'DEBUG', message, **kw)

-    def info(self, service, message, **kw):
+    def info(self, service: str, message: str, **kw):
        self.push(service, 'INFO', message, **kw)

-    def warning(self, service, message, **kw):
+    def warning(self, service: str, message: str, **kw):
        self.push(service, 'WARNING', message, **kw)

-    def error(self, service, message, **kw):
+    def warn(self, service: str, message: str, **kw):
+        self.push(service, 'WARNING', message, **kw)
+
+    def error(self, service: str, message: str, **kw):
        self.push(service, 'ERROR', message, **kw)

-    def critical(self, service, message, **kw):
+    def critical(self, service: str, message: str, **kw):
        self.push(service, 'CRITICAL', message, **kw)

    def flush(self):
-        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
-        self._flush_done.clear()
-        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
-        self._flush_done.wait(timeout=self._flush_interval + 2.0)
-        # 主线程排空队列中剩余条目
-        self._drain_queue_to_db()
+        """手动 flush 队列中剩余的日志。"""
+        self._queue.flush()

-    def _drain_queue_to_db(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
-        batch = []
-        while True:
-            try:
-                batch.append(self._queue.get_nowait())
-                if len(batch) >= self._batch_size:
-                    self._send(batch)
-                    batch = []
-            except queue.Empty:
-                break
-        if batch:
-            self._send(batch)
+    def close(self):
+        """关闭客户端，flush 剩余日志。"""
+        self._queue.close()

    # ── 内部实现 ────────────────────────────────────────────────────────

-    def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 OOM。
-        """
-        batch = []
-        last_flush = time.monotonic()
-
-        while not self._stop.is_set():
-            try:
-                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                chk_timeout = max(0.1, min(timeout, 0.5))
-                try:
-                    item = self._queue.get(timeout=chk_timeout)
-                    batch.append(item)
-                    while len(batch) < self._batch_size:
-                        try:
-                            batch.append(self._queue.get_nowait())
-                        except queue.Empty:
-                            break
-                except queue.Empty:
-                    pass
-
-                should_flush = (
-                    len(batch) >= self._batch_size or
-                    self._flush_signal.is_set() or
-                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-                )
-                if should_flush and batch:
-                    try:
-                        self._send(batch)
-                    except Exception as e:
-                        # _send 内部已有细粒度异常处理，此处是最终安全网
-                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
-                              file=_sys.stderr)
-                    batch = []
-                    last_flush = time.monotonic()
-                    if self._flush_signal.is_set():
-                        self._flush_signal.clear()
-                        self._flush_done.set()
-            except Exception as e:
-                # 最外层安全网：任何意外异常都不应杀死日志线程
-                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
-                      file=_sys.stderr)
-                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
-
-        # 退出前 flush 剩余
-        if batch:
-            try:
-                self._send(batch)
-            except Exception as e:
-                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
-        self._flush_done.set()
-
-    def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-
-        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
-        不抛异常给调用方。
-        """
-        try:
-            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
-        except Exception as e:
-            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
-            return
-
+    def _send_batch(self, batch: list):
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
        for entry in batch:
            extra_val = entry.get("extra")
            try:
@@ -224,21 +140,21 @@ class LogClient:
                },
            }
            try:
-                client.call("execute", payload, timeout=_TIMEOUT)
+                self._client.call("execute", payload, timeout=_TIMEOUT)
            except RpcError as e:
-                # 业务错误（如表结构不符）：打印后继续下一条
                print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
            except Exception as e:
-                # 网络错误：db 不可用，放弃本批剩余（fail-safe）
-                print(f"[log_client] 发送失败: {e}", file=_sys.stderr)
+                print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
                break

-    def _shutdown(self):
-        """进程退出时 flush 剩余日志。"""
-        self._stop.set()
-        self.flush()
+    def _on_error(self, e: Exception, batch: list):
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
+              file=_sys.stderr)


 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。

-logs = LogClient()
+logs = LogClient(service_name="confs")
@@ -25,10 +25,64 @@
 ./_run.sh     # venv 中执行 python3 dbv2.py
 ```

-服务监听 `tcp://127.0.0.1:8005`。生产环境由 core（Nidus）按 init.txt 统一拉起，守护靠 systemd。
+服务监听 `tcp://0.0.0.0:8005`（默认）。生产环境由 core（Nidus）按 init.txt 统一拉起，守护靠 systemd。

 **注意**：`execute` 路由依赖表已存在（通过反射获取结构）。建表见 `create_sqlite3.sql`（Users / Notes；Logs 表由服务启动时自建）。

+## 配置管理
+
+db 已接入 [confs](../confs/) 统一配置服务，confs 是 db 启动的**强依赖**：
+
+1. **confs 统一配置服务** — 必须成功从 confs 拉取 `db.ini`，否则 db 启动失败直接退出（exit code 1）
+2. **环境变量覆盖**（最高优先级）— Docker/部署环境通过环境变量覆盖 confs 中的值
+
+### confs 配置文件（db.ini）
+
+配置文件托管于 [confs/configs/db.ini](../confs/configs/db.ini)：
+
+```ini
+[server]
+host = 0.0.0.0
+port = 8005
+workers = 4
+
+[database]
+url = sqlite:///database.db
+
+[logs]
+retention_days = 30
+max_records = 1000000
+cleanup_interval = 3600
+```
+
+### 引导配置（confs 连接地址）
+
+confs 自身的连接地址不通过 confs 获取（避免循环依赖），通过以下方式确定（优先级从高到低）：
+- 环境变量 `CONFS_HOST` / `CONFS_PORT`
+- 默认值 `127.0.0.1:8010`
+
+### 环境变量（最高优先级，覆盖 confs 配置值）
+
+| 环境变量 | 说明 | 对应 confs 配置项 |
+|---------|------|------------------|
+| `LISTEN_ADDR` | 监听地址 | `[server] host` |
+| `LISTEN_PORT` | 监听端口 | `[server] port` |
+| `WORKERS` | 回调并发线程数 | `[server] workers` |
+| `DATABASE_URL` | 数据库连接 URL | `[database] url` |
+| `CONFS_HOST` | confs 服务地址（引导配置） | — |
+| `CONFS_PORT` | confs 服务端口（引导配置） | — |
+
+### 启动顺序
+
+db 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**不严格要求启动顺序**：
+- confs 先启动 → db 立即拉取配置，正常启动
+- db 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功
+- 配置完整性校验：若 db.ini 中缺少必需的 section/key 且无对应环境变量覆盖，仍然报错退出
+- 启动成功时控制台打印：`配置来源: confs (db.ini @ host:port)`
+
+> 独立运维工具 `db_tools.py`（backup/export/import/vacuum 等）仍直接读 `DATABASE_URL` 环境变量，不依赖 confs 服务就绪。
+> Docker 环境通过 `docker-entrypoint.sh` 传入环境变量（见 [Dockerfile](Dockerfile)），需确保 `CONFS_HOST`/`CONFS_PORT` 指向 confs 服务地址。
+
 ## 协议与 API

 ### 传输协议（pymsg）
@@ -232,10 +286,10 @@ python3 _view_db.sqlite3.py -s "SELECT service, COUNT(*) FROM Logs GROUP BY serv

 ## 切换数据库

-只需修改环境变量 `DATABASE_URL`，并安装对应驱动：
+推荐方式：修改 confs 的 [db.ini](../confs/configs/db.ini) 中 `[database] url`，然后通过 confs `reload` 路由重载（或重启 db）。也可通过环境变量 `DATABASE_URL` 覆盖（Docker 部署时常用）：

- SQLite（默认）：`sqlite:///database.db`
- PostgreSQL：`postgresql://user:pass@localhost/dbname`（需 `pip install psycopg2`）
+- SQLite（默认）：`sqlite:///database.db`（相对路径）或 `sqlite:////app/data/database.db`（绝对路径，Docker 中使用）
+- PostgreSQL：`postgresql://user:pass@localhost/dbname`（需 `pip install psycopg2-binary`）
 - MySQL：`mysql+pymysql://user:pass@localhost/dbname`（需 `pip install pymysql`）

 SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。
@@ -27,6 +27,7 @@ dbv2 — 基于 pymsg 协议（MsgServer）的数据库服务。
    响应中的日期时间一律序列化为 ISO 8601（T 分隔）。
 """

+import configparser
 import json
 import os
 import threading
@@ -47,9 +48,91 @@ import sys as _sys
 _sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 from pymsg import RpcServer  # noqa: E402

-# ── 数据库配置 ──────────────────────────────────────────────────────────
+# ── 配置加载（两层：confs 强依赖 → 环境变量覆盖）──────────────────────
+# confs 是 db 启动的强依赖：必须成功从 confs 拉取 db.ini，否则直接退出。
+# 环境变量优先级最高（Docker 兼容），可覆盖 confs 中的配置值。
+# confs 自身的连接地址（host/port）通过引导配置确定：环境变量 CONFS_HOST/CONFS_PORT
+# 优先，缺省使用 127.0.0.1:8010（与其他服务一致的约定）。

-DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.db")
+# 引导配置：confs 连接地址（不依赖 confs 本身，可通过环境变量覆盖）
+_confs_host = os.environ.get('CONFS_HOST', '127.0.0.1')
+_confs_port = int(os.environ.get('CONFS_PORT', '8010'))
+
+# 白名单结构：_merge_ini 仅覆盖此处列出的 section/key，防止意外字段
+_db_config = {
+    'server': {'host': '', 'port': '', 'workers': ''},
+    'database': {'url': ''},
+    'logs': {
+        'retention_days': '',
+        'max_records': '',
+        'cleanup_interval': '',
+    },
+}
+
+
+def _merge_ini(text):
+    """把 ini 文本合并进 _db_config（仅覆盖已知的 section/key）。"""
+    ini = configparser.ConfigParser(interpolation=None)
+    ini.read_string(text)
+    for section in _db_config:
+        if ini.has_section(section):
+            for key, value in ini[section].items():
+                if key in _db_config[section]:
+                    _db_config[section][key] = value
+
+
+def _fetch_from_confs():
+    """从 confs 统一配置服务拉取 db.ini 全文。
+
+    使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次，
+    每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。
+    RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。
+    """
+    from pymsg import RpcClient, RpcError  # noqa: E402
+    client = RpcClient(_confs_host, _confs_port, 'db', timeout=5,
+                       retry_on_refused=True)
+    try:
+        data = client.call('get', {'name': 'db.ini'})
+    except RpcError as e:
+        print(f'[dbv2] 致命错误: 从 confs 获取 db.ini 失败: {e}',
+              file=_sys.stderr)
+        _sys.exit(1)
+    return data.get('content')
+
+
+_confs_text = _fetch_from_confs()
+_merge_ini(_confs_text)
+_config_source = 'confs (db.ini)'
+print(f'[dbv2] 配置来源: confs 统一配置服务 (db.ini @ {_confs_host}:{_confs_port})')
+
+# 环境变量覆盖（最高优先级，保持 Docker 兼容）
+if os.environ.get('LISTEN_ADDR'):
+    _db_config['server']['host'] = os.environ['LISTEN_ADDR']
+if os.environ.get('LISTEN_PORT'):
+    _db_config['server']['port'] = os.environ['LISTEN_PORT']
+if os.environ.get('WORKERS'):
+    _db_config['server']['workers'] = os.environ['WORKERS']
+if os.environ.get('DATABASE_URL'):
+    _db_config['database']['url'] = os.environ['DATABASE_URL']
+
+# 配置完整性校验
+_required = {
+    'server': ['host', 'port', 'workers'],
+    'database': ['url'],
+    'logs': ['retention_days', 'max_records', 'cleanup_interval'],
+}
+for section, keys in _required.items():
+    for key in keys:
+        if not _db_config[section][key]:
+            print(f'[dbv2] 致命错误: 配置项 [{section}] {key} 缺失（db.ini 中未配置且无环境变量覆盖）',
+                  file=_sys.stderr)
+            _sys.exit(1)
+
+# 解析为最终常量
+DATABASE_URL = _db_config['database']['url']
+DEFAULT_RETENTION_DAYS = int(_db_config['logs']['retention_days'])
+DEFAULT_MAX_RECORDS = int(_db_config['logs']['max_records'])
+DEFAULT_CLEANUP_INTERVAL = int(_db_config['logs']['cleanup_interval'])

 # ── SQLite 连接配置（针对 Docker/多线程环境优化）──────────────────────

@@ -76,10 +159,6 @@ def _configure_sqlite(dbapi_connection, connection_record):
 MAX_MESSAGE_LEN = 8192   # message 上限 8KB
 MAX_EXTRA_LEN = 4096     # extra 上限 4KB

-DEFAULT_RETENTION_DAYS = 30
-DEFAULT_MAX_RECORDS = 1_000_000
-DEFAULT_CLEANUP_INTERVAL = 3600  # 秒
-
 #: where/values 中的哨兵值：服务端当前 UTC 时间
 NOW_SENTINEL = "$now"

@@ -456,17 +535,19 @@ class DBServer:
 if __name__ == "__main__":
    import argparse

-    default_addr = os.environ.get("LISTEN_ADDR", "0.0.0.0")
-    default_port = int(os.environ.get("LISTEN_PORT", "8005"))
+    default_addr = _db_config['server']['host']
+    default_port = int(_db_config['server']['port'])
+    default_workers = int(_db_config['server']['workers'])

    parser = argparse.ArgumentParser(description="dbv2 — pymsg 数据库服务")
    parser.add_argument("--addr", default=default_addr, help="监听地址")
    parser.add_argument("--port", type=int, default=default_port, help="监听端口")
-    parser.add_argument("--workers", type=int, default=4,
+    parser.add_argument("--workers", type=int, default=default_workers,
                        help="回调并发线程数（0=同步串行模式）")
    args = parser.parse_args()

    server = DBServer(args.addr, args.port, workers=args.workers)
+    print(f'[dbv2] 配置来源: {_config_source}')
    try:
        server.start()
    except KeyboardInterrupt:
@@ -215,27 +215,32 @@ curl -X POST http://localhost:8006/update_note \

 ## 配置

-dnfs 已接入 confs 统一配置服务，采用**三层 fail-safe** 加载模式（与 static 服务一致）：
+dnfs 已接入 confs 统一配置服务，confs 是 dnfs 启动的**强依赖**：

-**加载顺序**：内置默认值 → confs 统一配置服务（dnfs.ini，优先）→ 环境变量覆盖（最高优先级）
+1. **confs 统一配置服务** — 必须成功从 confs 拉取 `dnfs.ini`，否则 dnfs 启动失败直接退出（exit code 1）
+2. **环境变量覆盖**（最高优先级）— Docker/部署环境通过环境变量覆盖 confs 中的值

 ### 配置文件

 集中托管在 [confs/configs/dnfs.ini](../confs/configs/dnfs.ini)，由 confs 服务分发：

-| Section | Key | 默认值 | 说明 |
-|---------|-----|--------|------|
-| `[server]` | `host` | `127.0.0.1` | 监听地址 |
-| `[server]` | `port` | `8006` | 监听端口 |
-| `[database]` | `host` | `127.0.0.1` | db 服务地址 |
-| `[database]` | `port` | `8005` | db 服务端口 |
-| `[database]` | `timeout` | `10` | RPC 调用超时（秒） |
-| `[ai_review]` | `url` | `http://127.0.0.1:8007/review` | AI 审核接口地址 |
-| `[ai_review]` | `timeout` | `10` | AI 审核请求超时（秒） |
-| `[confs]` | `host` | `127.0.0.1` | confs 服务地址 |
-| `[confs]` | `port` | `8010` | confs 服务端口 |
+| Section | Key | 说明 |
+|---------|-----|------|
+| `[server]` | `host` | 监听地址 |
+| `[server]` | `port` | 监听端口 |
+| `[database]` | `host` | db 服务地址 |
+| `[database]` | `port` | db 服务端口 |
+| `[database]` | `timeout` | RPC 调用超时（秒） |
+| `[ai_review]` | `url` | AI 审核接口地址 |
+| `[ai_review]` | `timeout` | AI 审核请求超时（秒） |

-### 环境变量（Docker 环境必需，优先级最高）
+### 引导配置（confs 连接地址）
+
+confs 自身的连接地址不通过 confs 获取（避免循环依赖），通过以下方式确定（优先级从高到低）：
+- 环境变量 `CONFS_HOST` / `CONFS_PORT`
+- 默认值 `127.0.0.1:8010`
+
+### 环境变量（最高优先级，覆盖 confs 配置值）

 环境变量覆盖配置文件值，容器化部署时通过 docker-compose.yml 传入服务名：

@@ -246,10 +251,18 @@ dnfs 已接入 confs 统一配置服务，采用**三层 fail-safe** 加载模
 | `DB_HOST` | `db` | db 服务主机名（docker-compose 服务名） |
 | `DB_PORT` | `8005` | db 服务端口 |
 | `REVIEW_URL` | `http://ai_review:8007/review` | ai_review 服务地址 |
-| `CONFS_HOST` | - | 可选，confs 容器化后使用 |
-| `CONFS_PORT` | - | 可选，confs 容器化后使用 |
+| `CONFS_HOST` | `confs` | confs 服务主机名（引导配置） |
+| `CONFS_PORT` | `8010` | confs 服务端口（引导配置） |

-配置来源会记录到日志（`logs.info/warning`），便于排查实际用了哪一层。本地裸机运行时 confs 不可用会自动回退到内置默认值 + 环境变量。
+### 启动顺序
+
+dnfs 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**拉取配置阶段不严格要求启动顺序**：
+- confs 先启动 → dnfs 立即拉取配置
+- dnfs 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功
+- 配置完整性校验：若 dnfs.ini 中缺少必需的 section/key 且无对应环境变量覆盖，仍然报错退出
+- 启动成功时控制台打印：`配置来源: confs (dnfs.ini @ host:port)`
+
+> 注意：dnfs 对 db/ai_review 的 RPC 调用是运行时按需连接（首次请求时才建立 TCP），db/ai_review 未就绪时相关 API 返回 502 但进程不退出；待依赖服务启动后自动恢复。

 ## 前端

@@ -33,15 +33,21 @@ from log_client import logs
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 from pymsg import RpcClient, RpcError  # noqa: E402

-# ==================== 配置加载（三层 fail-safe） ====================
-# 加载顺序：内置默认值 → confs 统一配置服务（dnfs.ini） → 环境变量覆盖
-# 环境变量优先级最高（Docker 环境使用服务名 db/ai_review 而非 127.0.0.1）
+# ==================== 配置加载（两层：confs 强依赖 → 环境变量覆盖） ====================
+# confs 是 dnfs 启动的强依赖：必须成功从 confs 拉取 dnfs.ini，否则直接退出。
+# 环境变量优先级最高（Docker 环境使用服务名 db/ai_review 而非 127.0.0.1）。
+# confs 自身的连接地址通过引导配置确定：环境变量 CONFS_HOST/CONFS_PORT 优先，
+# 缺省使用 127.0.0.1:8010（与其他服务一致的约定）。

+# 引导配置：confs 连接地址（不依赖 confs 本身）
+_confs_host = os.environ.get('CONFS_HOST', '127.0.0.1')
+_confs_port = int(os.environ.get('CONFS_PORT', '8010'))
+
+# 白名单结构：_merge_ini 仅覆盖此处列出的 section/key，防止意外字段
 _dnfs_config = {
-    'server': {'host': '127.0.0.1', 'port': '8006'},
-    'database': {'host': '127.0.0.1', 'port': '8005', 'timeout': '10'},
-    'ai_review': {'url': 'http://127.0.0.1:8007/review', 'timeout': '10'},
-    'confs': {'host': '127.0.0.1', 'port': '8010'},
+    'server': {'host': '', 'port': ''},
+    'database': {'host': '', 'port': '', 'timeout': ''},
+    'ai_review': {'url': '', 'timeout': ''},
 }


@@ -57,23 +63,24 @@ def _merge_ini(text):


 def _fetch_from_confs():
-    """从 confs 统一配置服务拉取 dnfs.ini 全文，失败返回 None（fail-safe）。"""
+    """从 confs 统一配置服务拉取 dnfs.ini 全文。
+
+    使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次，
+    每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。
+    RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。
+    """
    try:
-        host = os.environ.get('CONFS_HOST', _dnfs_config['confs']['host'])
-        port = int(os.environ.get('CONFS_PORT', _dnfs_config['confs']['port']))
-        data = RpcClient(host, port, 'dnfs', timeout=3).call(
-            'get', {'name': 'dnfs.ini'})
+        data = RpcClient(_confs_host, _confs_port, 'dnfs', timeout=5,
+                         retry_on_refused=True).call('get', {'name': 'dnfs.ini'})
+    except RpcError as e:
+        print(f'[dnfs] 致命错误: 从 confs 获取 dnfs.ini 失败: {e}',
+              file=sys.stderr)
+        sys.exit(1)
    return data.get('content')
-    except Exception:
-        return None


 _confs_text = _fetch_from_confs()
-if _confs_text is not None:
 _merge_ini(_confs_text)
-    logs.info("dnfs", "配置来源: confs 统一配置服务 (dnfs.ini)")
-else:
-    logs.warning("dnfs", "confs 不可用，使用内置默认值 + 环境变量")

 # 环境变量覆盖（优先级最高，Docker 环境需要）
 if os.environ.get('LISTEN_ADDR'):
@@ -86,20 +93,32 @@ if os.environ.get('DB_PORT'):
    _dnfs_config['database']['port'] = os.environ['DB_PORT']
 if os.environ.get('REVIEW_URL'):
    _dnfs_config['ai_review']['url'] = os.environ['REVIEW_URL']
-if os.environ.get('CONFS_HOST'):
-    _dnfs_config['confs']['host'] = os.environ['CONFS_HOST']
-if os.environ.get('CONFS_PORT'):
-    _dnfs_config['confs']['port'] = os.environ['CONFS_PORT']
+
+# 配置完整性校验
+_required = {
+    'server': ['host', 'port'],
+    'database': ['host', 'port', 'timeout'],
+    'ai_review': ['url', 'timeout'],
+}
+for section, keys in _required.items():
+    for key in keys:
+        if not _dnfs_config[section][key]:
+            print(f'[dnfs] 致命错误: 配置项 [{section}] {key} 缺失（dnfs.ini 中未配置且无环境变量覆盖）',
+                  file=sys.stderr)
+            sys.exit(1)

 # 提取配置常量
+_DB_TIMEOUT = int(_dnfs_config['database']['timeout'])
 _DBV2_ADDR = _dnfs_config['database']['host']
 _DBV2_PORT = int(_dnfs_config['database']['port'])
-_DB_TIMEOUT = int(_dnfs_config['database']['timeout'])
 _REVIEW_SERVER_URL = _dnfs_config['ai_review']['url']
 _REVIEW_TIMEOUT = int(_dnfs_config['ai_review']['timeout'])
 _ADDR = _dnfs_config['server']['host']
 _PORT = int(_dnfs_config['server']['port'])

+_config_source = f'confs (dnfs.ini @ {_confs_host}:{_confs_port})'
+print(f'[dnfs] 配置来源: {_config_source}')
+
 # 管理员判定已上移至 gateway：登录时命中 [Admin] users 白名单即在 Token payload
 # 写入 is_admin=true，本服务通过 info.get('is_admin') 读取，不再本地维护名单。
 # 旧 Token 无此字段，按非管理员处理（安全方向）。
@@ -1,13 +1,15 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
+"""
+统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。

 设计要点：
  - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
-  - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
+  - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
  - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
-  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
+  - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
+  - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施

-用法：
+用法（与旧版完全兼容）：
    from log_client import logs

    logs.info("dnfs", "节点新增成功 id=42")
@@ -19,28 +21,25 @@
 import json
 import os as _os
 import sys as _sys
-import threading
-import queue
-import time
 import atexit
 from datetime import datetime, timezone

 _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
-from pymsg import RpcClient, RpcError  # noqa: E402
+from pymsg import RpcClient, RpcError, AsyncRpcQueue  # noqa: E402

 # ── 配置 ──────────────────────────────────────────────────────────────

 _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
 _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))

-_BATCH_SIZE = 50         # 队列满 50 条时 flush
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
-_TIMEOUT = 5             # socket 超时（秒）
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_BATCH_SIZE = 50
+_FLUSH_INTERVAL = 2.0
+_TIMEOUT = 5
+_QUEUE_MAXSIZE = 10000


 def _safe_json(obj):
-    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
    try:
        return json.dumps(obj, ensure_ascii=False, default=str)
    except Exception:
@@ -48,163 +47,80 @@ def _safe_json(obj):


 def _now_iso():
-    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
    return datetime.now(timezone.utc).isoformat()


 class LogClient:
-    """异步批量日志推送客户端。"""
+    """异步批量日志推送客户端（API 与旧版完全兼容）。"""

-    def __init__(self, service_name="log",
-                 batch_size=_BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+    def __init__(self, service_name: str = "log",
+                 batch_size: int = _BATCH_SIZE,
+                 flush_interval: float = _FLUSH_INTERVAL):
        self._service_name = service_name
-        self._batch_size = batch_size
-        self._flush_interval = flush_interval
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
-        self._stop = threading.Event()
-        self._flush_signal = threading.Event()
-        self._flush_done = threading.Event()
-        self._thread = threading.Thread(target=self._flush_loop,
-                                        daemon=True, name='log-client')
-        self._thread.start()
-        atexit.register(self._shutdown)
+        self._client = RpcClient(
+            _DBV2_ADDR, _DBV2_PORT,
+            service_name + "-log",
+            timeout=_TIMEOUT,
+        )
+        self._queue = AsyncRpcQueue(
+            handler=self._send_batch,
+            batch_size=batch_size,
+            flush_interval=flush_interval,
+            max_queue_size=_QUEUE_MAXSIZE,
+            on_error=self._on_error,
+        )
+        self._queue.start()
+        atexit.register(self.close)

    # ── 公开接口 ────────────────────────────────────────────────────────

-    def push(self, service, level, message, trace_id=None, extra=None):
+    def push(self, service: str, level: str, message: str,
+             trace_id: str = None, extra=None):
        """推送一条日志（非阻塞，立即返回）。

        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
        """
        entry = {
-            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
+            'ts': _now_iso(),
            'service': service,
            'level': level,
            'message': message,
            'trace_id': trace_id,
            'extra': extra,
        }
-        try:
-            self._queue.put_nowait(entry)
-        except queue.Full:
-            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
-            try:
-                self._queue.get_nowait()
-            except queue.Empty:
-                pass
-            try:
-                self._queue.put_nowait(entry)
-            except queue.Full:
-                pass  # 极端竞争下仍失败则静默丢弃
+        self._queue.put(entry)

-    def debug(self, service, message, **kw):
+    def debug(self, service: str, message: str, **kw):
        self.push(service, 'DEBUG', message, **kw)

-    def info(self, service, message, **kw):
+    def info(self, service: str, message: str, **kw):
        self.push(service, 'INFO', message, **kw)

-    def warning(self, service, message, **kw):
+    def warning(self, service: str, message: str, **kw):
        self.push(service, 'WARNING', message, **kw)

-    def error(self, service, message, **kw):
+    def warn(self, service: str, message: str, **kw):
+        self.push(service, 'WARNING', message, **kw)
+
+    def error(self, service: str, message: str, **kw):
        self.push(service, 'ERROR', message, **kw)

-    def critical(self, service, message, **kw):
+    def critical(self, service: str, message: str, **kw):
        self.push(service, 'CRITICAL', message, **kw)

    def flush(self):
-        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
-        self._flush_done.clear()
-        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
-        self._flush_done.wait(timeout=self._flush_interval + 2.0)
-        # 主线程排空队列中剩余条目
-        self._drain_queue_to_db()
+        """手动 flush 队列中剩余的日志。"""
+        self._queue.flush()

-    def _drain_queue_to_db(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
-        batch = []
-        while True:
-            try:
-                batch.append(self._queue.get_nowait())
-                if len(batch) >= self._batch_size:
-                    self._send(batch)
-                    batch = []
-            except queue.Empty:
-                break
-        if batch:
-            self._send(batch)
+    def close(self):
+        """关闭客户端，flush 剩余日志。"""
+        self._queue.close()

    # ── 内部实现 ────────────────────────────────────────────────────────

-    def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 OOM。
-        """
-        batch = []
-        last_flush = time.monotonic()
-
-        while not self._stop.is_set():
-            try:
-                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                chk_timeout = max(0.1, min(timeout, 0.5))
-                try:
-                    item = self._queue.get(timeout=chk_timeout)
-                    batch.append(item)
-                    while len(batch) < self._batch_size:
-                        try:
-                            batch.append(self._queue.get_nowait())
-                        except queue.Empty:
-                            break
-                except queue.Empty:
-                    pass
-
-                should_flush = (
-                    len(batch) >= self._batch_size or
-                    self._flush_signal.is_set() or
-                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-                )
-                if should_flush and batch:
-                    try:
-                        self._send(batch)
-                    except Exception as e:
-                        # _send 内部已有细粒度异常处理，此处是最终安全网
-                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
-                              file=_sys.stderr)
-                    batch = []
-                    last_flush = time.monotonic()
-                    if self._flush_signal.is_set():
-                        self._flush_signal.clear()
-                        self._flush_done.set()
-            except Exception as e:
-                # 最外层安全网：任何意外异常都不应杀死日志线程
-                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
-                      file=_sys.stderr)
-                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
-
-        # 退出前 flush 剩余
-        if batch:
-            try:
-                self._send(batch)
-            except Exception as e:
-                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
-        self._flush_done.set()
-
-    def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-
-        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
-        不抛异常给调用方。
-        """
-        try:
-            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
-        except Exception as e:
-            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
-            return
-
+    def _send_batch(self, batch: list):
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
        for entry in batch:
            extra_val = entry.get("extra")
            try:
@@ -224,21 +140,21 @@ class LogClient:
                },
            }
            try:
-                client.call("execute", payload, timeout=_TIMEOUT)
+                self._client.call("execute", payload, timeout=_TIMEOUT)
            except RpcError as e:
-                # 业务错误（如表结构不符）：打印后继续下一条
                print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
            except Exception as e:
-                # 网络错误：db 不可用，放弃本批剩余（fail-safe）
-                print(f"[log_client] 发送失败: {e}", file=_sys.stderr)
+                print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
                break

-    def _shutdown(self):
-        """进程退出时 flush 剩余日志。"""
-        self._stop.set()
-        self.flush()
+    def _on_error(self, e: Exception, batch: list):
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
+              file=_sys.stderr)


 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。

-logs = LogClient()
+logs = LogClient(service_name="dnfs")
@@ -110,6 +110,20 @@ resp = client.call('{"action": "select", "table": "Users"}', timeout=10.0)

 目标不可达时抛出 `ConnectionRefusedError`，超时无回复时抛出 `TimeoutError`。

+**启动等待模式（retry_on_refused）**：
+
+```python
+# 服务启动时阻塞等待下游服务就绪，连接被拒绝时自动重试
+resp = client.call('{"_route": "get", "name": "db.ini"}',
+                   timeout=5.0,
+                   retry_on_refused=True,   # 连接拒绝时阻塞重试
+                   retry_interval=1.0)      # 重试间隔 1 秒（默认）
+```
+
+`retry_on_refused=True` 时，`ConnectionRefusedError` 不会抛出，而是按 `retry_interval`（默认 1.0 秒）间隔不断重试，直到连接成功。重试期间每 5 秒向 stderr 打印一次等待提示（如 `[pymsg] 等待服务 127.0.0.1:8010 就绪... (已重试 N 次)`），避免日志刷屏。仅对 `ConnectionRefusedError` 重试；`TimeoutError`、`ConnectionError`、业务错误等仍立即抛出。
+
+适用场景：微服务启动顺序不确定时，服务进程可在不依赖外部进程管理器的情况下自动等待下游依赖就绪，无需手动 sys.exit + 外部重启。`log_client` 等 fail-safe 场景应保持默认 `retry_on_refused=False`，避免后台线程阻塞。
+
 **3. 长连接监听（回调模式）**

 ```python
@@ -226,7 +240,23 @@ except RpcError as e:
    ...  # 服务端业务错误（success=false）
 ```

-`call(route, payload=None, *, timeout=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。
+| 构造参数 | 含义 |
+|----------|------|
+| `addr` / `port` | 服务端地址与端口 |
+| `name` | 客户端标识（透传 `MsgClient`） |
+| `timeout` | 默认调用超时秒数（默认 `10.0`） |
+| `retry_on_refused` | 连接被拒绝时是否阻塞重试（关键字参数，默认 `False`） |
+| `retry_interval` | 重试间隔秒数（关键字参数，默认 `1.0`） |
+
+`call(route, payload=None, *, timeout=None, retry_on_refused=None, retry_interval=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。`retry_on_refused` 和 `retry_interval` 参数可在单次调用时覆盖构造时的设置。
+
+**启动等待示例**（解决微服务启动顺序问题）：
+
+```python
+# confs 客户端：启动时阻塞等待 confs 就绪，不依赖启动顺序
+rpc = RpcClient("127.0.0.1", 8010, "db", timeout=5, retry_on_refused=True)
+data = rpc.call("get", {"name": "db.ini"})  # confs 未启动时自动阻塞等待
+```

 ## 实际用例

@@ -22,6 +22,7 @@ import select
 import socket as _socket
 import struct
 import threading
+import time
 import traceback
 from concurrent.futures import ThreadPoolExecutor

@@ -200,7 +201,10 @@ class MsgClient:
            )

    # ---- 请求-应答（短连接）------------------------------------------------
-    def call(self, msg: str, *, timeout: "float | None" = None) -> str:
+    def call(self, msg: str, *, timeout: "float | None" = None,
+             retry_on_refused: bool = False,
+             retry_interval: float = 1.0,
+             _log_every: float = 5.0) -> str:
        """发送一条消息并阻塞等待单条回复（短连接：connect → send → recv → close）。

        适合请求-应答模式，如 RPC 调用。不会给消息添加 name 前缀。
@@ -208,16 +212,24 @@ class MsgClient:
        Args:
            msg:             要发送的原始字符串。
            timeout:         超时秒数，默认 10 秒。
+            retry_on_refused: 目标服务未启动（ConnectionRefusedError）时是否阻塞重试，
+                              直到连接成功。适用于启动顺序不确定的服务发现场景。
+                              默认 False（连接拒绝立即抛异常）。
+            retry_interval:  重试间隔秒数，默认 1.0。
+            _log_every:      重试期间打印等待提示的间隔（秒），默认每 5 秒打印一次，避免刷屏。

        Returns:
            str: 服务器返回的字符串。

        Raises:
-            ConnectionRefusedError: 目标服务不可达。
+            ConnectionRefusedError: 目标服务不可达且 ``retry_on_refused=False``。
            TimeoutError:           等待响应超时。
            ConnectionError:        连接在收全数据前断开。
        """
        _tout = timeout if timeout is not None else 10.0
+        _last_log = 0.0
+        _attempt = 0
+        while True:
            try:
                with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as s:
                    s.settimeout(_tout)
@@ -225,9 +237,20 @@ class MsgClient:
                    send(s, msg)
                    return recv(s, timeout=_tout)
            except ConnectionRefusedError:
+                if not retry_on_refused:
                    raise ConnectionRefusedError(
                        f"Could not connect to {self.addr}:{self.port}"
                    )
+                _attempt += 1
+                now = time.monotonic()
+                if now - _last_log >= _log_every or _attempt == 1:
+                    print(
+                        f"[pymsg] 等待服务 {self.addr}:{self.port} 就绪... "
+                        f"(已重试 {_attempt} 次)",
+                        file=__import__('sys').stderr,
+                    )
+                    _last_log = now
+                time.sleep(retry_interval)

    # ---- 持续监听（阻塞当前线程，回调模式）-------------------------------
    def listen(self, on_message: "callable") -> None:
@@ -502,5 +525,6 @@ class MsgServer:

 # ── JSON RPC 约定层（依赖上面的 MsgClient / MsgServer，故放在末尾导入）──
 from .rpc import RpcClient, RpcError, RpcServer  # noqa: E402
+from .async_queue import AsyncRpcQueue  # noqa: E402

-__all__ += ["RpcClient", "RpcServer", "RpcError"]
+__all__ += ["RpcClient", "RpcServer", "RpcError", "AsyncRpcQueue"]
@@ -0,0 +1,202 @@
+"""
+AsyncRpcQueue - 通用异步批量发送队列
+
+提供内存队列 + 后台线程的异步批量处理机制，适用于日志上报、指标采集等
+非关键路径场景。业务线程put操作非阻塞，后台线程攒批处理，失败时通过错误回调
+通知，不影响主业务流程。
+
+两种使用模式：
+
+1. 简单RPC模式（指定client + method + serialize）：
+    queue = AsyncRpcQueue(
+        client=rpc_client,
+        method='log_api.push',
+        batch_size=50,
+        flush_interval=2.0,
+        serialize=lambda items: {'logs': json.dumps(items)},
+    )
+
+2. 自定义处理模式（指定handler回调）：
+    def send_batch(batch):
+        for item in batch:
+            rpc_client.call('execute', item)
+    queue = AsyncRpcQueue(
+        handler=send_batch,
+        batch_size=50,
+        flush_interval=2.0,
+    )
+"""
+
+import threading
+import time
+from collections import deque
+from typing import Any, Callable, Optional
+
+
+class AsyncRpcQueue:
+    """通用异步批量发送队列"""
+
+    def __init__(
+        self,
+        client: Any = None,
+        method: str = None,
+        *,
+        handler: Optional[Callable[[list], None]] = None,
+        batch_size: int = 100,
+        flush_interval: float = 2.0,
+        max_queue_size: int = 10000,
+        serialize: Optional[Callable[[list], dict]] = None,
+        on_error: Optional[Callable[[Exception, list], None]] = None,
+        retry_on_refused: bool = False,
+        retry_interval: float = 1.0,
+    ):
+        """
+        初始化异步队列
+
+        Args:
+            client: RpcClient实例（RPC模式必填）
+            method: RPC方法名（RPC模式必填）
+            handler: 自定义批量处理回调，接收一批items列表。提供此参数时
+                     忽略client/method/serialize/retry_on_refused。
+            batch_size: 批量发送的阈值，达到此数量立即发送
+            flush_interval: 定时flush间隔（秒）
+            max_queue_size: 队列最大长度，超过则丢弃最旧数据
+            serialize: 序列化函数，接收items列表返回RPC参数字典
+                       默认: lambda items: {'items': items}
+            on_error: 错误回调，接收(exception, batch_items)
+            retry_on_refused: 是否在连接被拒绝时重试（RPC模式，透传给client.call）
+            retry_interval: 重试间隔（秒）（RPC模式，透传给client.call）
+        """
+        if handler is None:
+            if client is None or method is None:
+                raise ValueError("必须提供 handler 或 (client + method)")
+            self._handler = self._make_rpc_handler(
+                client, method,
+                serialize or (lambda items: {'items': items}),
+                retry_on_refused, retry_interval,
+            )
+        else:
+            self._handler = handler
+
+        self._batch_size = batch_size
+        self._flush_interval = flush_interval
+        self._max_queue_size = max_queue_size
+        self._on_error = on_error
+
+        self._queue: deque = deque()
+        self._lock = threading.Lock()
+        self._wakeup = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+        self._closed = False
+        self._dropped_count = 0
+
+    @staticmethod
+    def _make_rpc_handler(client, method, serialize, retry_on_refused, retry_interval):
+        """创建RPC模式的默认handler"""
+        def handler(batch: list):
+            params = serialize(batch)
+            client.call(
+                method, params,
+                retry_on_refused=retry_on_refused,
+                retry_interval=retry_interval,
+            )
+        return handler
+
+    def start(self) -> None:
+        """启动后台发送线程"""
+        if self._thread is not None and self._thread.is_alive():
+            return
+        self._closed = False
+        self._thread = threading.Thread(
+            target=self._run,
+            name=f'AsyncRpcQueue',
+            daemon=True,
+        )
+        self._thread.start()
+
+    def put(self, item: Any) -> bool:
+        """
+        放入一条数据到队列（非阻塞）
+
+        Args:
+            item: 要发送的数据项
+
+        Returns:
+            bool: True表示放入成功，False表示队列已关闭
+        """
+        if self._closed:
+            return False
+        with self._lock:
+            if len(self._queue) >= self._max_queue_size:
+                self._queue.popleft()
+                self._dropped_count += 1
+            self._queue.append(item)
+        self._wakeup.set()
+        return True
+
+    def flush(self) -> None:
+        """立即触发一次发送（会阻塞直到处理完成，主要用于关闭时）"""
+        self._send_batch()
+
+    def close(self) -> None:
+        """关闭队列，flush剩余数据后停止后台线程"""
+        if self._closed:
+            return
+        self._closed = True
+        self._wakeup.set()
+        if self._thread is not None:
+            self._thread.join(timeout=5.0)
+        self._send_batch()
+
+    @property
+    def pending_count(self) -> int:
+        """当前队列中待发送的数据量"""
+        with self._lock:
+            return len(self._queue)
+
+    @property
+    def dropped_count(self) -> int:
+        """因队列满而丢弃的数据量"""
+        return self._dropped_count
+
+    def _run(self) -> None:
+        """后台线程主循环"""
+        last_flush = time.time()
+        while not self._closed:
+            self._wakeup.wait(timeout=0.1)
+            self._wakeup.clear()
+
+            now = time.time()
+            should_flush = False
+            with self._lock:
+                if len(self._queue) >= self._batch_size:
+                    should_flush = True
+                elif now - last_flush >= self._flush_interval and self._queue:
+                    should_flush = True
+            if should_flush:
+                self._send_batch()
+                last_flush = time.time()
+
+    def _send_batch(self) -> None:
+        """取一批数据处理"""
+        with self._lock:
+            if not self._queue:
+                return
+            batch = list(self._queue)
+            self._queue.clear()
+        try:
+            self._handler(batch)
+        except Exception as e:
+            if self._on_error:
+                try:
+                    self._on_error(e, batch)
+                except Exception:
+                    pass
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+        return False
@@ -36,21 +36,33 @@ class RpcClient:

        rpc = RpcClient("127.0.0.1", 8005, "dnfs")
        data = rpc.call("execute", {"action": "select", "table": "Users"})
+
+        # 启动时阻塞等待服务就绪（连接被拒绝时自动重试）
+        rpc = RpcClient("127.0.0.1", 8010, "static", retry_on_refused=True)
    """

    def __init__(self, addr: str, port: int, name: str = "rpc",
-                 timeout: float = 10.0) -> None:
+                 timeout: float = 10.0,
+                 *, retry_on_refused: bool = False,
+                 retry_interval: float = 1.0) -> None:
        self._client = MsgClient(addr, port, name)
        self._timeout = timeout
+        self._retry_on_refused = retry_on_refused
+        self._retry_interval = retry_interval

    def call(self, route: str, payload: "dict | None" = None, *,
-             timeout: "float | None" = None):
+             timeout: "float | None" = None,
+             retry_on_refused: "bool | None" = None,
+             retry_interval: "float | None" = None):
        """调用指定路由，返回响应中的 ``data`` 字段。

        Args:
            route:   路由名（注入为 ``_route`` 字段）。
            payload: 业务字段 dict（``_route`` 键会被覆盖）。
            timeout: 本次调用的超时秒数，缺省用构造时的值。
+            retry_on_refused: 本次调用是否在连接被拒绝时阻塞重试，
+                              None 表示使用构造时的值（默认 False）。
+            retry_interval:  本次调用的重试间隔，None 表示使用构造时的值。

        Returns:
            服务端返回的 ``data``（可能为 None）。
@@ -58,13 +70,17 @@ class RpcClient:
        Raises:
            RpcError:              服务端返回 success=false，或响应非法。
            pymsg.TimeoutError:    等待响应超时。
-            ConnectionRefusedError: 目标服务不可达。
+            ConnectionRefusedError: 目标服务不可达且未开启重试。
        """
        req = dict(payload or {})
        req["_route"] = route
+        _retry = self._retry_on_refused if retry_on_refused is None else retry_on_refused
+        _interval = self._retry_interval if retry_interval is None else retry_interval
        resp = self._client.call(
            json.dumps(req, ensure_ascii=False),
            timeout=self._timeout if timeout is None else timeout,
+            retry_on_refused=_retry,
+            retry_interval=_interval,
        )
        try:
            obj = json.loads(resp)
@@ -90,26 +90,52 @@ static :9200 (Flask + CORS)

 ## 配置

-`config.ini`：
+static 已接入 confs 统一配置服务，confs 是 static 启动的**强依赖**：

-```ini
-[server]
-host = 127.0.0.1
-port = 9200
+1. **confs 统一配置服务** — 必须成功从 confs 拉取 `static.ini`，否则 static 启动失败直接退出（exit code 1）
+2. **环境变量覆盖**（最高优先级）— Docker/部署环境通过环境变量覆盖 confs 中的值

-[directory]
-root = ./public
+### 配置文件

-[ssl]
-enabled = false
-cert_file = cert.pem
-key_file = key.pem
+集中托管在 [confs/configs/static.ini](../confs/configs/static.ini)，由 confs 服务分发：

-[cors]
-# 允许跨域的 Origin 白名单（逗号分隔）
-# 注意：生产环境请勿包含 localhost / 127.0.0.1 / 通配符 *
-allowed_origins = https://qsont.me,https://yhdet.top
-```
+| Section | Key | 说明 |
+|---------|-----|------|
+| `[server]` | `host` | 监听地址 |
+| `[server]` | `port` | 监听端口 |
+| `[directory]` | `root` | 静态文件根目录（相对 BASE_DIR） |
+| `[ssl]` | `enabled` | 是否启用 TLS（`true`/`false`，默认 `false`） |
+| `[ssl]` | `cert_file` | SSL 证书路径（默认 `cert.pem`） |
+| `[ssl]` | `key_file` | SSL 私钥路径（默认 `key.pem`） |
+| `[cors]` | `allowed_origins` | CORS 白名单，逗号分隔 |
+
+### 引导配置（confs 连接地址）
+
+confs 自身的连接地址不通过 confs 获取（避免循环依赖），通过以下方式确定（优先级从高到低）：
+- 环境变量 `CONFS_HOST` / `CONFS_PORT`
+- 默认值 `127.0.0.1:8010`
+
+### 环境变量（最高优先级，覆盖 confs 配置值）
+
+| 变量 | 说明 |
+|------|------|
+| `LISTEN_ADDR` | 监听地址（容器内需绑定 `0.0.0.0`） |
+| `LISTEN_PORT` | 监听端口 |
+| `STATIC_ROOT` | 静态文件根目录 |
+| `CONFS_HOST` | confs 服务主机名（引导配置，Docker 环境使用服务名） |
+| `CONFS_PORT` | confs 服务端口（引导配置） |
+
+### 启动顺序
+
+static 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**拉取配置阶段不严格要求启动顺序**：
+- confs 先启动 → static 立即拉取配置
+- static 先启动 → pymsg 自动重试（每秒一次，每 5 秒打印等待提示），confs 启动后自动连接成功
+- 配置完整性校验：若 static.ini 中缺少必需的 section/key（server.host/port、directory.root、cors.allowed_origins）且无对应环境变量覆盖，仍然报错退出
+- ssl 段为可选项，未配置时默认 `enabled=false`、cert/key 使用默认文件名
+
+### 历史遗留：本地 config.ini 已删除
+
+旧版本的 `static/config.ini` 本地回退逻辑已移除，配置统一由 confs 服务分发。

 ## 依赖

@@ -1,5 +1,5 @@
 #!/bin/bash
 cd "$(dirname "$0")" || exit
 source ./venv/bin/activate
-#exec ./fsys.bin --config config.ini 1>/dev/null
-exec python3 ./fsys.py --config config.ini 1>/dev/null
+#exec ./fsys.bin 1>/dev/null
+exec python3 ./fsys.py 1>/dev/null
@@ -1,17 +0,0 @@
-[server]
-host = 127.0.0.1
-port = 9200
-
-[directory]
-root = ./public
-
-[ssl]
-enabled = false
-cert_file = cert.pem
-key_file = key.pem
-
-[cors]
-# 允许跨域的 Origin 白名单（逗号分隔）
-# 注意：生产环境请勿包含 localhost / 127.0.0.1 / 通配符 *
-#       本地开发联调时可临时追加 http://localhost:9200，调试完务必移除
-allowed_origins = https://qsont.me,https://yhdet.top
@@ -36,33 +36,35 @@ file_handler.setLevel(logging.INFO)
 logger.addHandler(file_handler)

 parser = argparse.ArgumentParser(description='File Server')
-parser.add_argument('--config', type=str, default='config.ini', help='Configuration file path')
 args = parser.parse_args()

 app = Flask(__name__)
 apply_security_headers(app)

-# ==================== 配置文件加载 ====================
-# 加载顺序：内置默认值 → confs 统一配置服务（static.ini）；
-# confs 不可用时 fail-safe 回退本地 config.ini
-CONFIG_FILE = args.config
-if not os.path.isabs(CONFIG_FILE):
-    CONFIG_FILE = os.path.join(BASE_DIR, CONFIG_FILE)
+# ==================== 配置加载（两层：confs 强依赖 → 环境变量覆盖） ====================
+# confs 是 static 启动的强依赖：必须成功从 confs 拉取 static.ini，否则直接退出。
+# 环境变量优先级最高（Docker 兼容），可覆盖 confs 中的配置值。
+# confs 自身的连接地址通过引导配置确定：环境变量 CONFS_HOST/CONFS_PORT 优先，
+# 缺省使用 127.0.0.1:8010（与其他服务一致的约定）。

+# 引导配置：confs 连接地址（不依赖 confs 本身）
+_confs_host = os.environ.get('CONFS_HOST', '127.0.0.1')
+_confs_port = int(os.environ.get('CONFS_PORT', '8010'))
+
+# 白名单结构：_merge_ini 仅覆盖此处列出的 section/key，防止意外字段
 file_config = {
-    'server': {'host': '0.0.0.0', 'port': '5000'},
-    'directory': {'root': './public'},
-    'ssl': {'enabled': 'false', 'cert_file': 'cert.pem', 'key_file': 'key.pem'},
-    # CORS 白名单（逗号分隔），生产环境默认值；localhost 仅限本地开发时显式写入 config.ini
-    'cors': {'allowed_origins': 'https://qsont.me,https://yhdet.top'},
+    'server': {'host': '', 'port': ''},
+    'directory': {'root': ''},
+    'ssl': {'enabled': '', 'cert_file': '', 'key_file': ''},
+    'cors': {'allowed_origins': ''},
 }


 def _merge_ini(text):
    """把 ini 文本合并进 file_config（仅覆盖已知的 section/key）。"""
-    ini = configparser.ConfigParser()
+    ini = configparser.ConfigParser(interpolation=None)
    ini.read_string(text)
-    for section in ['server', 'directory', 'ssl', 'cors']:
+    for section in file_config:
        if ini.has_section(section):
            for key, value in ini[section].items():
                if key in file_config[section]:
@@ -70,28 +72,60 @@ def _merge_ini(text):


 def _fetch_from_confs():
-    """从 confs 统一配置服务拉取 static.ini 全文，失败返回 None（fail-safe）。"""
+    """从 confs 统一配置服务拉取 static.ini 全文。
+
+    使用 retry_on_refused=True：confs 未就绪时 pymsg 自动阻塞重试（每秒一次，
+    每 5 秒打印一次等待提示），直到 confs 启动后自动连接成功。
+    RpcError（如配置文件不存在）仍直接报错退出——这是配置错误而非启动时序问题。
+    """
+    from pymsg import RpcClient, RpcError  # noqa: E402
+    client = RpcClient(_confs_host, _confs_port, 'static', timeout=5,
+                       retry_on_refused=True)
    try:
-        from pymsg import RpcClient
-        host = os.environ.get('CONFS_HOST', '127.0.0.1')
-        port = int(os.environ.get('CONFS_PORT', '8010'))
-        data = RpcClient(host, port, 'static', timeout=3).call(
-            'get', {'name': 'static.ini'})
+        data = client.call('get', {'name': 'static.ini'})
+    except RpcError as e:
+        print(f'[static] 致命错误: 从 confs 获取 static.ini 失败: {e}',
+              file=sys.stderr)
+        sys.exit(1)
    return data.get('content')
-    except Exception:
-        return None


 _confs_text = _fetch_from_confs()
-if _confs_text is not None:
 _merge_ini(_confs_text)
-    logger.info("配置来源: confs 统一配置服务 (static.ini)")
-    logs.info("static", "配置来源: confs 统一配置服务 (static.ini)")
-elif os.path.exists(CONFIG_FILE):
-    with open(CONFIG_FILE, encoding='utf-8') as f:
-        _merge_ini(f.read())
-    logger.warning(f"confs 不可用，回退本地配置文件 {CONFIG_FILE}")
-    logs.warning("static", f"confs 不可用，回退本地配置文件 {CONFIG_FILE}")
+
+# 环境变量覆盖（优先级最高，Docker 兼容）
+if os.environ.get('LISTEN_ADDR'):
+    file_config['server']['host'] = os.environ['LISTEN_ADDR']
+if os.environ.get('LISTEN_PORT'):
+    file_config['server']['port'] = os.environ['LISTEN_PORT']
+if os.environ.get('STATIC_ROOT'):
+    file_config['directory']['root'] = os.environ['STATIC_ROOT']
+
+# 配置完整性校验
+_required = {
+    'server': ['host', 'port'],
+    'directory': ['root'],
+    'cors': ['allowed_origins'],
+}
+for section, keys in _required.items():
+    for key in keys:
+        if not file_config[section][key]:
+            print(f'[static] 致命错误: 配置项 [{section}] {key} 缺失（static.ini 中未配置且无环境变量覆盖）',
+                  file=sys.stderr)
+            sys.exit(1)
+
+# ssl 段有合理默认（enabled=false，cert/key 有默认文件名），不强制要求配置
+if not file_config['ssl']['enabled']:
+    file_config['ssl']['enabled'] = 'false'
+if not file_config['ssl']['cert_file']:
+    file_config['ssl']['cert_file'] = 'cert.pem'
+if not file_config['ssl']['key_file']:
+    file_config['ssl']['key_file'] = 'key.pem'
+
+_config_source = f'confs (static.ini @ {_confs_host}:{_confs_port})'
+print(f'[static] 配置来源: {_config_source}')
+logger.info(f"配置来源: {_config_source}")
+logs.info("static", f"配置来源: {_config_source}")

 # ==================== CORS（配置加载完成后初始化） ====================
 _cors_origins = [o.strip() for o in file_config['cors']['allowed_origins'].split(',') if o.strip()]
@@ -1,13 +1,15 @@
-"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
+"""
+统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。

 设计要点：
  - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
-  - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
-  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
+  - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务
+  - fail-safe：db 服务不可用时静默丢弃，绝不影响业务
  - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
-  - 时间戳：日志入队时即记录 ts，反映事件真实发生时间
+  - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间
+  - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施

-用法：
+用法（与旧版完全兼容）：
    from log_client import logs

    logs.info("dnfs", "节点新增成功 id=42")
@@ -19,28 +21,25 @@
 import json
 import os as _os
 import sys as _sys
-import threading
-import queue
-import time
 import atexit
 from datetime import datetime, timezone

 _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
-from pymsg import RpcClient, RpcError  # noqa: E402
+from pymsg import RpcClient, RpcError, AsyncRpcQueue  # noqa: E402

 # ── 配置 ──────────────────────────────────────────────────────────────

 _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
 _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))

-_BATCH_SIZE = 50         # 队列满 50 条时 flush
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
-_TIMEOUT = 5             # socket 超时（秒）
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_BATCH_SIZE = 50
+_FLUSH_INTERVAL = 2.0
+_TIMEOUT = 5
+_QUEUE_MAXSIZE = 10000


 def _safe_json(obj):
-    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
+    """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。"""
    try:
        return json.dumps(obj, ensure_ascii=False, default=str)
    except Exception:
@@ -48,163 +47,80 @@ def _safe_json(obj):


 def _now_iso():
-    """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
+    """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。"""
    return datetime.now(timezone.utc).isoformat()


 class LogClient:
-    """异步批量日志推送客户端。"""
+    """异步批量日志推送客户端（API 与旧版完全兼容）。"""

-    def __init__(self, service_name="log",
-                 batch_size=_BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+    def __init__(self, service_name: str = "log",
+                 batch_size: int = _BATCH_SIZE,
+                 flush_interval: float = _FLUSH_INTERVAL):
        self._service_name = service_name
-        self._batch_size = batch_size
-        self._flush_interval = flush_interval
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
-        self._stop = threading.Event()
-        self._flush_signal = threading.Event()
-        self._flush_done = threading.Event()
-        self._thread = threading.Thread(target=self._flush_loop,
-                                        daemon=True, name='log-client')
-        self._thread.start()
-        atexit.register(self._shutdown)
+        self._client = RpcClient(
+            _DBV2_ADDR, _DBV2_PORT,
+            service_name + "-log",
+            timeout=_TIMEOUT,
+        )
+        self._queue = AsyncRpcQueue(
+            handler=self._send_batch,
+            batch_size=batch_size,
+            flush_interval=flush_interval,
+            max_queue_size=_QUEUE_MAXSIZE,
+            on_error=self._on_error,
+        )
+        self._queue.start()
+        atexit.register(self.close)

    # ── 公开接口 ────────────────────────────────────────────────────────

-    def push(self, service, level, message, trace_id=None, extra=None):
+    def push(self, service: str, level: str, message: str,
+             trace_id: str = None, extra=None):
        """推送一条日志（非阻塞，立即返回）。

        队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
        """
        entry = {
-            'ts': _now_iso(),  # 入队时即生成时间戳，反映真实事件时间
+            'ts': _now_iso(),
            'service': service,
            'level': level,
            'message': message,
            'trace_id': trace_id,
            'extra': extra,
        }
-        try:
-            self._queue.put_nowait(entry)
-        except queue.Full:
-            # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
-            try:
-                self._queue.get_nowait()
-            except queue.Empty:
-                pass
-            try:
-                self._queue.put_nowait(entry)
-            except queue.Full:
-                pass  # 极端竞争下仍失败则静默丢弃
+        self._queue.put(entry)

-    def debug(self, service, message, **kw):
+    def debug(self, service: str, message: str, **kw):
        self.push(service, 'DEBUG', message, **kw)

-    def info(self, service, message, **kw):
+    def info(self, service: str, message: str, **kw):
        self.push(service, 'INFO', message, **kw)

-    def warning(self, service, message, **kw):
+    def warning(self, service: str, message: str, **kw):
        self.push(service, 'WARNING', message, **kw)

-    def error(self, service, message, **kw):
+    def warn(self, service: str, message: str, **kw):
+        self.push(service, 'WARNING', message, **kw)
+
+    def error(self, service: str, message: str, **kw):
        self.push(service, 'ERROR', message, **kw)

-    def critical(self, service, message, **kw):
+    def critical(self, service: str, message: str, **kw):
        self.push(service, 'CRITICAL', message, **kw)

    def flush(self):
-        """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。"""
-        self._flush_done.clear()
-        self._flush_signal.set()  # 通知后台线程立即 flush 其局部 batch
-        # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
-        self._flush_done.wait(timeout=self._flush_interval + 2.0)
-        # 主线程排空队列中剩余条目
-        self._drain_queue_to_db()
+        """手动 flush 队列中剩余的日志。"""
+        self._queue.flush()

-    def _drain_queue_to_db(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
-        batch = []
-        while True:
-            try:
-                batch.append(self._queue.get_nowait())
-                if len(batch) >= self._batch_size:
-                    self._send(batch)
-                    batch = []
-            except queue.Empty:
-                break
-        if batch:
-            self._send(batch)
+    def close(self):
+        """关闭客户端，flush 剩余日志。"""
+        self._queue.close()

    # ── 内部实现 ────────────────────────────────────────────────────────

-    def _flush_loop(self):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 OOM。
-        """
-        batch = []
-        last_flush = time.monotonic()
-
-        while not self._stop.is_set():
-            try:
-                timeout = self._flush_interval - (time.monotonic() - last_flush)
-                chk_timeout = max(0.1, min(timeout, 0.5))
-                try:
-                    item = self._queue.get(timeout=chk_timeout)
-                    batch.append(item)
-                    while len(batch) < self._batch_size:
-                        try:
-                            batch.append(self._queue.get_nowait())
-                        except queue.Empty:
-                            break
-                except queue.Empty:
-                    pass
-
-                should_flush = (
-                    len(batch) >= self._batch_size or
-                    self._flush_signal.is_set() or
-                    (batch and time.monotonic() - last_flush >= self._flush_interval)
-                )
-                if should_flush and batch:
-                    try:
-                        self._send(batch)
-                    except Exception as e:
-                        # _send 内部已有细粒度异常处理，此处是最终安全网
-                        print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
-                              file=_sys.stderr)
-                    batch = []
-                    last_flush = time.monotonic()
-                    if self._flush_signal.is_set():
-                        self._flush_signal.clear()
-                        self._flush_done.set()
-            except Exception as e:
-                # 最外层安全网：任何意外异常都不应杀死日志线程
-                print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
-                      file=_sys.stderr)
-                time.sleep(0.5)  # 防止异常风暴导致 CPU 满载
-
-        # 退出前 flush 剩余
-        if batch:
-            try:
-                self._send(batch)
-            except Exception as e:
-                print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
-        self._flush_done.set()
-
-    def _send(self, batch):
-        """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
-
-        整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
-        不抛异常给调用方。
-        """
-        try:
-            client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
-        except Exception as e:
-            print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
-            return
-
+    def _send_batch(self, batch: list):
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
        for entry in batch:
            extra_val = entry.get("extra")
            try:
@@ -224,21 +140,21 @@ class LogClient:
                },
            }
            try:
-                client.call("execute", payload, timeout=_TIMEOUT)
+                self._client.call("execute", payload, timeout=_TIMEOUT)
            except RpcError as e:
-                # 业务错误（如表结构不符）：打印后继续下一条
                print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
            except Exception as e:
-                # 网络错误：db 不可用，放弃本批剩余（fail-safe）
-                print(f"[log_client] 发送失败: {e}", file=_sys.stderr)
+                print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr)
                break

-    def _shutdown(self):
-        """进程退出时 flush 剩余日志。"""
-        self._stop.set()
-        self.flush()
+    def _on_error(self, e: Exception, batch: list):
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
+              file=_sys.stderr)


 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。

-logs = LogClient()
+logs = LogClient(service_name="static")
@@ -0,0 +1,38 @@
+/* index.css — 社区首页专属样式（通用主题见 theme.css） */
+
+body {
+    align-items: normal;
+}
+
+/* ===== 头部 ===== */
+.header {
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    width: 100%;
+}
+
+.right-group {
+    display: flex;
+    align-items: center;
+}
+
+.right-group > img,
+.right-group > b,
+.right-group > button {
+    margin-left: 5px;
+    margin-right: 5px;
+}
+
+#avatar {
+    border-radius: 50%;
+}
+
+/* ===== 网站简介 ===== */
+.site-intro {
+    max-width: 640px;
+    margin: 0 auto;
+    font-size: 0.95rem;
+    line-height: 1.9;
+    color: var(--fg-secondary);
+}
@@ -1,48 +1,35 @@
 <!DOCTYPE html>
-<html lang="en">
+<html lang="zh-CN">
 <head>
    <meta http-equiv="Content-Type" content="text/html;charset=UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <link rel="stylesheet" href="css/theme.css">
-    <style>
-        body {
-            align-items: normal;
-        }
-        .header {
-            display: flex;
-            justify-content: space-between;
-            align-items: center;
-            width: 100%;
-        }
-        .right-group {
-            display: flex;
-            align-items: center;
-        }
-        .right-group > img,
-        .right-group > b,
-        .right-group > button {
-            margin-left: 5px;
-            margin-right: 5px;
-        }
-    </style>
+    <link rel="stylesheet" href="css/index.css">
    <title>DNFS 欢迎来到树状笔记</title>
 </head>
 <body>
    <header class="header">
        <h1 onclick="window.location.href='index.html'">DNFS 树状笔记</h1>
        <div class="right-group">
-            <img src="favicon.ico" alt="头像加载失败" id="avatar" width="70px" height="70px" style="border-radius: 100px;">
+            <img src="favicon.ico" alt="头像加载失败" id="avatar" width="70px" height="70px">
            <b id="name">DNFS</b>
-            <button onclick="window.location.href='login.html'">登录</button>
-            <button onclick="window.location.href='register.html'">注册</button>
+            <button id="enterBtn" onclick="window.location.href='dnfs.html'" style="display:none">进入笔记</button>
+            <button id="logoutBtn" style="display:none">退出</button>
+            <button id="loginBtn" onclick="window.location.href='login.html'">登录</button>
+            <button id="registerBtn" onclick="window.location.href='register.html'">注册</button>
        </div>
    </header>
    <div id="main-page">
-        <h2>推荐节点</h2>
-        <div id="recommended-node"></div>
+        <h2>关于本站</h2>
+        <p class="site-intro">
+            DNFS 树状笔记是一个以树状结构组织内容的笔记社区：每个节点都可以生长出子节点，
+            像树枝一样层层延展，让知识自然成体系。支持 Markdown 文本、链接卡片、图片与文件分享，
+            注册登录后即可创建属于你自己的知识树。
+        </p>
    </div>
    <footer class="footer">
 		<a href="https://beian.miit.gov.cn/" target="_blank" rel="noopener">蜀ICP备2024107064号-2</a>
 	</footer>
+	<script src="js/index.js"></script>
 </body>
 </html>
@@ -0,0 +1,42 @@
+/* index.js — 社区首页逻辑
+ *
+ * 登录态检测：POST /whoami（HttpOnly q4_token Cookie 自动携带，JS 不经手 Token）
+ * 已登录：显示用户名与"进入笔记/退出"按钮；未登录：显示"登录/注册"
+ */
+(function () {
+	'use strict';
+
+	const nameEl = document.getElementById('name');
+	const enterBtn = document.getElementById('enterBtn');
+	const logoutBtn = document.getElementById('logoutBtn');
+	const loginBtn = document.getElementById('loginBtn');
+	const registerBtn = document.getElementById('registerBtn');
+
+	async function checkLogin() {
+		try {
+			const res = await fetch('/whoami', { method: 'POST' });
+			if (!res.ok) return;
+			const data = await res.json();
+			if (data && data.login && data.user) {
+				nameEl.textContent = String(data.user);
+				enterBtn.style.display = '';
+				logoutBtn.style.display = '';
+				loginBtn.style.display = 'none';
+				registerBtn.style.display = 'none';
+			}
+		} catch (e) {
+			console.warn('index: 登录态检测失败', e.message);
+		}
+	}
+
+	logoutBtn.addEventListener('click', async function () {
+		try {
+			await fetch('/logout', { method: 'POST' });
+		} catch (e) {
+			console.warn('index: 退出登录请求异常', e.message);
+		}
+		location.reload();
+	});
+
+	checkLogin();
+})();
Author	SHA1	Message	Date
qing	d883dc620e	refactor(pymsg): 新增AsyncRpcQueue通用异步队列，log_client基于此重构 - pymsg新增AsyncRpcQueue：提供内存队列+后台线程+攒批处理的通用异步基础设施，支持两种模式：简单RPC模式(client+method+serialize)和自定义handler模式 - confs/dnfs/static的log_client.py重构为基于AsyncRpcQueue的薄封装，保持API完全向后兼容(push/info/error/warning/debug/critical + 全局logs单例) - 消除了三份log_client副本中重复的队列/线程/攒批逻辑(约150行/份)，代码量净减少约50行 - LogClient的行为与旧版一致：逐条INSERT到db、fail-fast网络错误、队列满丢最旧	2026-08-01 12:59:33 +08:00
qing	101aed48be	feat(pymsg): RpcClient 新增 retry_on_refused 阻塞等待模式，解决微服务启动顺序依赖 pymsg 客户端: - MsgClient.call() 新增 retry_on_refused/retry_interval 参数 - retry_on_refused=True 时，ConnectionRefusedError 不抛出，按 retry_interval(默认1s)阻塞重试 - 重试期间每5秒向 stderr 打印等待提示，避免日志刷屏 - 仅对 ConnectionRefusedError 重试；TimeoutError/ConnectionError/RpcError 仍立即抛出 - 默认 retry_on_refused=False，完全向后兼容 - RpcClient 构造函数新增同名关键字参数，call() 支持单次调用覆盖 - 新增 time 模块导入 - 更新 README 文档（新增"启动等待模式"章节和参数表）各服务（db/dnfs/static）: - _fetch_from_confs() 改用 retry_on_refused=True，移除手动 ConnectionRefusedError 捕获和 sys.exit - 保留 RpcError（如配置文件不存在）的友好错误处理和 sys.exit(1) - 启动时不再因 confs 未就绪而退出：pymsg 自动阻塞等待 confs 启动后自动连接成功 - 启动顺序从"刚性要求 confs 先启动"变为"不严格要求启动顺序" - log_client 保持默认 retry_on_refused=False，后台线程不阻塞（fail-safe 丢弃日志） - 更新各服务 README 和 confs README 的启动顺序描述测试验证: - 基本 call() 正常工作 - retry_on_refused=False 连接拒绝立即抛 ConnectionRefusedError - retry_on_refused=True 阻塞等待（延迟1.5s启动服务后成功连接，重试日志正常输出）	2026-08-01 12:47:08 +08:00
qing	0ae107f953	refactor(static): 删除默认配置与本地回退逻辑，confs 改为强依赖 - 配置加载从三层（内置默认值→confs→本地config.ini回退）改为两层（confs 强依赖→环境变量覆盖） - confs 连接失败直接 sys.exit(1)，不再静默使用内置默认值或回退本地 config.ini - file_config 初始值改为空字符串白名单模板，移除所有内置默认值 - 新增配置完整性校验：server.host/port、directory.root、cors.allowed_origins 必须非空 - ssl 段保留合理默认（enabled=false、cert/key 默认文件名），不强制配置 - 区分错误类型：连接拒绝/RpcError/通用异常分别打印明确错误信息到 stderr - confs 连接地址改为引导配置（环境变量 CONFS_HOST/CONFS_PORT 优先，默认 127.0.0.1:8010） - 移除 --config 命令行参数（本地 config.ini 回退已删除） - 删除 static/config.ini 本地回退文件 - 更新 _run.sh 移除 --config 参数 - 更新 static/README.md 与 confs/README.md 文档 - confs 集成关系描述更新为"confs 最先启动"	2026-08-01 12:38:51 +08:00
qing	de0e4168b2	refactor(static): 首页内联样式解耦至 css/index.css - 新增页面级样式文件 css/index.css（与 view.css/aibox.css 惯例一致） - index.html 移除内联 <style> 块与头像内联样式，改为外链 - 头像圆角 100px 优化为 border-radius: 50%	2026-08-01 12:37:19 +08:00
qing	1a5e9f7064	feat(static): 首页添加登录态检测与进入笔记入口 - index.html: lang 改为 zh-CN，新增"关于本站"介绍段落，按钮改为有 id 的登录/注册/进入/退出四按钮（默认隐藏进入/退出） - 新增 index.js: POST /whoami 检测登录态（HttpOnly Cookie 自动携带），已登录则显示用户名+进入笔记/退出按钮，未登录显示登录/注册 - 退出按钮调用 POST /logout 后刷新页面	2026-08-01 12:34:53 +08:00
qing	84adec4722	refactor(dnfs): 删除默认配置逻辑，confs 改为强依赖 - 配置加载从三层（默认值→confs→环境变量）改为两层（confs 强依赖→环境变量覆盖） - confs 连接失败直接 sys.exit(1)，不再静默使用内置默认值 - _dnfs_config 初始值改为空字符串白名单模板，移除内置默认值 - 新增配置完整性校验：confs 返回后检查所有必需 section/key 非空 - 区分错误类型：连接拒绝/RpcError/通用异常分别打印明确错误信息到 stderr - confs 连接地址改为引导配置（环境变量优先，默认 127.0.0.1:8010），移除 dnfs.ini 中 [confs] 段 - 启动顺序调整为 confs → db → dnfs → ai_review/...（dnfs 同时依赖 confs 和 db） - 更新 dnfs/README.md 与 confs/README.md 文档	2026-08-01 12:33:51 +08:00
qing	2cbb93b3b7	refactor(db): 删除 confs 连接失败时使用默认配置的 fail-safe 逻辑 - confs 改为 db 启动的强依赖：连接失败直接 sys.exit(1)，不再静默使用默认值 - 配置加载从三层（默认值→confs→环境变量）改为两层（confs→环境变量覆盖） - _db_config 初始值改为空字符串白名单模板，移除内置默认值 - 新增配置完整性校验：confs 返回后检查所有必需 section/key 非空，缺失则报错退出 - 区分错误类型：连接拒绝/RpcError/通用异常分别打印明确的错误信息到 stderr - confs 连接地址改为引导配置（环境变量优先，默认 127.0.0.1:8010），不再从 db.ini 读取 - 更新 db/README.md 与 confs/README.md，说明启动顺序调整为 confs → db → ...	2026-08-01 12:29:55 +08:00
qing	6ebd544619	feat(db): 迁移 db 配置到 confs 统一配置服务 - 新增 confs/configs/db.ini 集中托管 db 服务配置（server/database/logs/confs） - dbv2.py 采用三层 fail-safe 加载模式：内置默认值 → confs（db.ini，优先）→ 环境变量覆盖（最高优先级，Docker 兼容） - _merge_ini() 白名单合并，仅覆盖已知 section/key，ConfigParser(interpolation=None) 禁用插值 - db 先于 confs 启动时连接失败静默 fail-safe，启动时打印配置来源 - db_tools.py 运维工具保持独立，直接读 DATABASE_URL 环境变量，不依赖 confs - 更新 confs/README.md 与 db/README.md 文档	2026-08-01 12:23:21 +08:00