Files

14 KiB
Raw Permalink Blame History

Rust 插件编写说明

面向「用 Rust 编写 chongming 进程插件」的开发指引,基于 chongming-worker-rs SDK。 配套:线协议规范(唯一契约)、多语言演进规划、 Python 进程插件开发指南、SDK 说明。

先决条件:Rust 工具链(cargo ≥ 1.70)。Rust 插件以独立进程运行 (load_mode = "external"),与网关零导入关系,仅经 Zenoh 线协议通讯。


1. 脚手架:gen_plugin --rust

推荐直接用脚手架生成自包含项目模板(等价 workers/rsdemo 的最小结构):

uv run gen_plugin myplugin --rust              # → workers/myplugin/
uv run gen_plugin myplugin analyze count --rust # 指定初始操作

生成内容(config 自包含在插件根,与 Python 进程插件一致;工程化分层,对位 workers/rsdemo):

workers/myplugin/
├── config.toml                 # 最小根配置(仅 [zenoh].json5_path,worker 启动必需)
├── config/
│   ├── zenoh_config.json5      # Zenoh 会话配置(相对 config.toml 解析为绝对路径)
│   └── workers/myplugin/
│       ├── config.toml         # [workers] 注册 + [worker] 分组 + defaults + cache/database
│       └── <op>.toml           # 每个操作的端点定义(唯一事实来源)
├── Cargo.toml                  # path 依赖 chongming-worker-rs;features: redis-cache / database
├── src/
│   ├── lib.rs                  # 库层:声明各模块 + 文档
│   ├── main.rs                 # 入口(薄):CLI 解析 + 日志 + runtime + App::init + serve + dispose
│   ├── args.rs                 # clap 命令行参数(--type/--count/--config/--config-dir/--instance)
│   ├── locator.rs              # 插件根自定位 plugin_root() + 配置路径解析(Locator)
│   ├── util.rs                 # 通用小工具 now_ts / deser(含单测)
│   ├── app.rs                  # 装配 App:数据库生命周期 init/dispose + 端点注册 + 模型指纹校验
│   ├── generated_models.rs     # gen_models 生成(勿手动修改)
│   └── handlers/<op>.rs        # 每个端点的业务实现(每端点一文件)
├── build.sh                    # 一键打包(构建 release + 打包插件根含配置)
├── README.md / .gitignore

也可以手动复用现有参考插件 workers/rsdemo/ 作为模板(更完整,含流式、数据库示例)。


2. 定义端点(TOML 是唯一事实来源)

编辑 config/workers/<type>/<op>.toml,格式与 Python 插件完全一致 (见 进程插件开发指南 [GET] / [POST] / [GET.response_model] 约定)。

# config/workers/myplugin/hello.toml
[GET]
summary = "问候"
params = ["name: str"]
auth_required = false

[GET.response_model]
greeting = { type = "str", required = true }
worker_id = { type = "str", required = false, default = "" }
timestamp = { type = "float", required = true }

端点名 <op>.toml 决定线协议 subject:myplugin.hello 与 HTTP 路径 /myplugin/hello。


3. 生成模型

模型由端点 TOML 自动生成,命名/类型映射对齐 Python(hello+GET → HelloGetRequest):

cargo run --manifest-path utils/worker-rs/Cargo.toml --bin gen_models -- \
    --types myplugin --config-dir workers/myplugin/config/workers \
    --out workers/myplugin/src/generated_models.rs

类型映射:str→String / int→i64 / float→f64 / bool→bool / any→Value / list[X]→Vec<X> / dict[K,V]→HashMap<K,V>;required=false → Option<T>(空值省略)。

一致性保证(强于 Python):启动时 check_models_fingerprint 比对 TOML 指纹, 改 TOML 未重跑 gen_models 即告警并提示重跑,杜绝模型漂移。


4. 实现 handler(src/main.rs)

use chongming_worker_rs::prelude::*;
use chongming_worker_rs::{HandlerRegistry, ServeOptions};
mod generated_models; use generated_models as models;

// 普通(非流式)端点
async fn hello(request: Request) -> Result<Reply, String> {
    let req: models::HelloGetRequest =
        serde_json::from_value(request.data).map_err(|e| format!("参数解析失败: {e}"))?;
    Ok(Reply::Json(serde_json::to_value(&models::HelloGetResponse {
        greeting: format!("Hello, {}!", req.name),
        worker_id: Some(request.worker_id), // 框架注入副本 ID
        timestamp: now_ts(),
    }).map_err(|e| e.to_string())?))
}

// 流式端点(SSE 逐块)
async fn stream(request: Request) -> Result<Reply, String> {
    let stream = async_stream::stream! {
        for i in 0..5 { yield Ok(json!({"index": i})); }
    };
    Ok(Reply::Stream(Box::pin(stream)))
}

在 main() 中注册并 serve(脚手架已搭好,main.rs 内含插件根自定位 plugin_root(), 从任意 CWD/部署路径运行都能命中自身 config.toml):

let opts = ServeOptions { config_path, config_dir, count, instance };
tokio::runtime::Builder::new_multi_thread().enable_all().build()?.block_on(async {
    let mut registry = HandlerRegistry::new();
    registry.register("myplugin.hello", "GET", hello);
    registry.register("myplugin.stream", "POST", stream);
    chongming_worker_rs::serve::serve(&type_, Arc::new(registry), opts).await
})
CLI 参数 说明
--type 插件类型(= config/workers/<type>/)
--count 副本数,可覆盖 [workers].count
--config 全局 config.toml 路径(读取 [zenoh] 节,默认 config.toml)
--instance 实例 id(多实例分摊;缺省 {hostname}:{pid})

生命周期(serve 内部自动完成):发布路由描述符 → 声明 liveliness token → 声明业务 + __admin__ Queryable → 每 10s 重发布(自愈)→ SIGTERM/SIGINT 优雅摘除。 在此之外需要资源初始化/清理时,把 Arc<Database> 之类的资源捕获进 handler 闭包, 在 serve 返回后做 dispose(见 §7)。


5. 配置三件套

5.1 运行配置 [workers.config]

config/workers/<type>/config.toml 的 [workers.config] 段会被展平传给你。 端点 handler 一般不需要自己读它——需要时用 chongming_worker_rs::config::load_plugin_config 取 runtime 获取。

5.2 多实例共享缓存 [workers.cache]

Rust 进程插件支持两种共享缓存后端(schema 与 Python chongming-cache 完全一致):

  1. 本机共享 sqlite(feature local-cache):同一份 SQLite 文件由 Python 与 Rust 读写, 实现单机多实例/跨语言收敛,无需 Redis。需 cargo build --features local-cache:
    [workers.cache]
    backend = "sqlite"          # schema 与 Python SqliteBackend 一致(旧名 diskcache 兼容)
    path = "./cache_data/cache.db"   # 与网关该实例 `[cache].path` 指向同一文件
    launch_worker_count = "cache"
    
  2. 跨主机 redis(feature redis-cache):多机部署必须用 Redis 共享缓存:
    [workers.cache]
    backend = "redis"
    url = "redis://127.0.0.1:6379/0"
    prefix = "chongming:"         # 与网关 config.toml [cache].redis_prefix 对齐(缺省空)
    launch_worker_count = "cache"
    
    未指定 backend(缺省)时 Rust 走 Cache::None,退化为本地 toml 语义。

prefix 必须与网关 [cache].redis_prefix(如 "chongming:")一致,Rust 插件才会读写 与网关同一批共享键。若网关未设前缀可省略。缓存值统一以 JSON 存储(Python 侧用 orjson、Rust 侧用 serde_json),任意键均可跨语言互通;与 Rust 共享的键请使用 JSON 兼容类型(dict/list/int/str/bool/null)。

编译启用:bash build.sh dist "redis-cache"(或 Cargo 依赖加 chongming-worker-rs/redis-cache)。 未配置时单实例直接读写本地 config.toml(默认无缓存),行为一致。

5.2.1 配置优先来源模式(launch_worker_count)

[workers.cache].launch_worker_count 与网关根 config.toml 的同名键语义一致,决定启动时 [workers.config] 与 count 从哪一侧取值(SDK 的 config::resolve_runtime_config / cache::resolve_worker_count 按同一规则实现,与 Python SDK 对位):

环节 "toml"(默认) "cache"
启动时 以 config/workers/<type>/config.toml 为准,并把该段覆盖写入缓存(网关读缓存 → 显示 toml 值) 以缓存为准,插件初始化即读缓存值;缓存缺失才用 toml 引导并写回
插件实际使用 toml 值 缓存值(跨重启保持)
config_set 改动的持久性 仅本次运行有效,下次启动被文件覆盖 跨重启保持(写入缓存)

插件初始化读配置(如 [workers.config].database)请用 config::effective_config_section 或 database::init_database(内部已按模式解析),不要直接读 load_plugin_config(...).runtime["config"]——后者恒为 toml 种子,cache 模式下会与网关 get_config 显示的缓存值不一致(表现为"前端显示缓存值、实际跑 toml")。

5.3 数据库 [workers.config.database](可选)

见 §7。


6. 管理指令(admin)

每副本注册 worker/<type>/<id>/__admin__ Queryable,收到信封 {"meta":{}, "data":{"cmd":...}} 分派执行:

指令 作用
status 副本状态(epoch/index/status/worker_id/instance_id)
restart / stop / drain 重启 / 停止 / 优雅排空(隔离新请求 + 等在途归零,30s 超时)
get_config / config_set 读 / 改 [workers.config](保留键校验 + 原子落盘 + 热更重启)
rescale 调整副本数(多实例按缓存全局分摊)

可经网关 /admin/workers/<type>/... 端点或直接对 worker/<type>/<id>/__admin__ 发 Zenoh Query 触发。开发者无需改任何代码,SDK 已内置。


7. 数据库接入(对齐 Python chongming_database)

Rust 用 chongming-worker-rs 的 database 模块(sqlx)获得与 Python chongming_database 等价的能力:SQLite / PostgreSQL 异步连接池 + 生命周期。

  • 声明后端(在 [workers.config].database,缺省/未配置则插件不建库,对齐 db_name = None):

    [workers.config.database]
    type = "sqlite"                # "sqlite"(默认) | "postgresql"
    database_path = "./data/myplugin.db"
    # postgresql → type = "postgresql"; url = "postgres://user:pwd@host:5432/db"
    
  • 初始化并把 Arc<Database> 捕获进 handler(main.rs):

    #[cfg(feature = "database")]
    let db = chongming_worker_rs::database::init_database(
        &type_, &config_dir,
        Some("CREATE TABLE IF NOT EXISTS kv (k TEXT PRIMARY KEY, v TEXT NOT NULL)"),
    ).await?.map(Arc::new); // Option<Arc<Database>>
    
    #[cfg(feature = "database")]
    if let Some(db) = db.clone() {
        registry.register("myplugin.db_ping", "GET", move |req| { let db = db.clone(); Box::pin(async move {
            let pool = db.pool().await?;
            sqlx::query("UPDATE ...").execute(&pool).await.map_err(|e| e.to_string())?;
            Ok(Reply::Json(json!({"ok": true})))
        })});
    }
    
  • 依赖:Cargo.toml 开启 database feature(脚手架已内置:database = ["chongming-worker-rs/database", "sqlx"]), 打包 bash build.sh dist "database"。

  • 差异点:SQLite 连接串 sqlite:<path>?mode=rwc(rwc 自动建库),PostgreSQL postgres://...; SQL 占位符遵循各自方言(SQLite ?、PostgreSQL $1);默认不启用 TLS,PostgreSQL 走 SSL 需追加 tls-rustls-ring 等 feature。参考实现见 workers/rsdemo 的 db_ping。


8. 打包与部署

在插件根目录执行(配置已自包含,打包脚本会把 release 二进制 + 全部配置带上):

cd workers/myplugin
bash build.sh                       # → dist/myplugin.zip
bash build.sh dist "redis-cache"    # 需要多实例共享缓存时
bash build.sh dist "database"       # 需要数据库(sqlx)时

产物 zip 根即容器工作目录:含 release 二进制 + config.toml + config/ + start.sh, 解压即运行(./start.sh,无需 uv / Python 运行时)。容器部署:

unzip myplugin.zip -d /app && cd /app && ./start.sh
  • load_mode = "external" 必须保留:网关跳过进程内加载,路由完全由描述符驱动。
  • 与 Python 插件可同 type 混布:描述符 hash 一致 → 网关幂等去重零冲突,负载均衡透明分发。

9. 本地联调

# 起一个本地 Zenoh Router(替代 zenohd)
.venv/bin/python utils/worker-rs/scripts/zenoh_router.py &

# 协议级端到端自检(描述符 / RPC / 流式 / 错误 / 摘除 / 自愈)
.venv/bin/python utils/worker-rs/scripts/verify_e2e.py \
    --type myplugin --bin workers/myplugin/target/debug/myplugin

10. Rust 侧行为要点

要点 说明
契约校验 发布前自检:无 handler 的端点告警跳过,不发布残缺路由
模型生成 gen_models 生成强类型结构体 + 启动时指纹校验(TOML 变更未重生成即告警)
handler 注入 worker_id 经 Request.worker_id、user_info 经 Request.user_info 注入(非函数参数)
配置缓存后端 local-cache(sqlite,与 Python 共享同一 .db)/ redis-cache(跨主机);缺省无缓存回退
数据库 走 sqlx(见 §7),配置键与 chongming_database 对齐

11. 参考材料

  • 线协议规范:./plugin-wire-protocol.md(唯一契约,附录 A 事实来源索引)
  • 多语言演进规划:./multi-language-plugin-plan.md(含数据库/缓存对齐表)
  • 完整示例插件:../workers/rsdemo/(流式、错误、数据库、管理指令全覆盖)
  • 脚手架:uv run gen_plugin --help(--rust)