跳到主要内容

三原语与线上协议

30 秒导读: iii 引擎是一个"值机台"——各种 worker 进程连上来,用一条 WebSocket 通道、 一套 JSON 消息,声明自己有哪些能力、这些能力什么时候该被叫醒、以及它们如何归类。 本章只讲清楚三件事:这三种能力叫什么(三原语)、它们在代码里是什么样的数据结构、 以及 worker 和引擎之间那套消息的契约(字段含义、序列化约定、什么可空)。

本章是 iii 系列的第一站。它讲路由怎么走、调度怎么排、注册表怎么并发、触发器内部怎么实现—— 那些分别留给 引擎中枢一次调用的一生触发器体系。这里只把"词汇表和语法"立住:读完你应该能看着一段 线上 JSON,说出它是哪种消息、每个字段是什么意思、引擎凭它记住了什么。


1. 这是什么(零基础也能懂)

1.1 一句话定义

iii 把"一个能被远程调用的功能"抽象成三个最小概念,合称三原语:

原语白话回答的问题
Function(函数)一段可执行的逻辑单元"能做什么"
Trigger(触发器)一条绑定:"当某类事件发生,就调这个函数""何时做"
Service(服务)函数的分组 / 命名空间,可嵌套"怎么归类"

worker(承载业务代码的进程,可能是 Node、Python、Rust 等)启动后,连上引擎,把自己的函数、 触发器、服务一条条声明上去;之后引擎在合适的时机反过来这些函数干活。

1.2 三原语怎么组合

它们不是平行的三堆东西,而是有明确从属关系:

Service ← 归类 / 命名空间(可嵌套:parent_service_id)
└── Function ← 可执行单元(handler + 输入/输出 schema)

│ Trigger 用 function_id 绑定到某个函数

Trigger ← "某类事件 → 调这个函数" 的一条绑定
└── 归属某个 TriggerType(定义"这类触发器长什么样、配置字段是什么")
  • Function 是原子:真正"干活"的那段代码。
  • Trigger 是绑定:它本身不干活,只说"http 这类事件命中时,请调 function_id 指向的那个函数"。
  • TriggerType 是模具:定义"http 这类触发器"整体长什么样(它的配置 schema、调用负载 schema), 由专门的 worker 提供(例如 iii-http worker 提供 http 类型)。
  • Service 是文件夹:给函数一个可嵌套的命名空间,便于组织和发现。

1.3 用起来什么样

从线上看,worker 声明一个函数,就是往 WebSocket 里发这么一条 JSON(引擎收到后记进注册表):

{
"type": "registerfunction",
"id": "external.my_lambda",
"description": "External Lambda function",
"invocation": {
"url": "https://example.com/lambda",
"timeout_ms": 30000,
"headers": {"x-custom-header": "value"},
"auth": {"type": "bearer", "token_key": "LAMBDA_TOKEN"}
}
}

这条报文在源码里对应 Message::RegisterFunction 变体,可用 serde_json 直接反序列化—— protocol.rs:250 的测试 deserialize_register_function_with_http_invocation 就是拿这段 JSON 验证的。注意 "type": "registerfunction":类型标签全小写、无分隔符,这是全套协议的统一约定(§4.1)。

1.4 一句话直觉

把引擎想成电话总机:worker 是各个分机,开机时先打电话给总机报备"我这有哪些业务、哪些号码 一响就转给我"(注册函数/触发器);之后有人拨号进来,总机按登记表把电话转到对应分机(调用函数), 分机办完把结果回给总机。本章讲的就是"报备"和"转接"这些电话到底该怎么说——即报文格式。


2. 顶层全景(worker 和引擎怎么对话)

2.1 一张图:一根管子,两种方向

worker 与引擎之间是一条 WebSocket 长连接,双向流动同一套 Message JSON。方向大致是:

Worker (SDK 侧) Engine (Rust 侧)
│ │
│ ── registerfunction / registertrigger ──▶ │ 存入
│ ── registertriggertype / registerservice ──▶ │ FunctionsRegistry
│ │ + TriggerRegistry
│ │
│ ◀── invokefunction ── │ "该你这个函数干活了"
│ ── invocationresult ──▶ │ 按 invocation_id 对回结果
│ │
│ ◀── ping ── ── pong ──▶ │ 保活/健康
│ ◀── workerregistered ── │ 握手确认(下发 worker_id)

怎么读这张图: 左半边是"声明类"消息(worker→引擎,把原语登记进注册表);中段是"调用类" (引擎→worker 发起,worker→引擎回结果);底部是"保活/握手"。所有消息共用同一个 Message 枚举。

2.2 部件一句话职责

部件干什么在哪
Message 枚举worker↔引擎全部报文的唯一联合类型,tag=type 小写protocol.rs:42 Message
Function一个函数的内存表示:handler + 输入/输出 schemafunction.rs:29 Function
FunctionsRegistry按 id 存放所有函数,支持增删查function.rs:59 FunctionsRegistry
Trigger / TriggerType一条触发绑定 / 一类触发器的定义trigger.rs:175 Trigger · trigger.rs:61 TriggerType
TriggerRegistrator"某类触发器"具体怎么挂载/卸载的行为契约trigger.rs:163 TriggerRegistrator
HttpInvocationRef报文里的辅助负载类型protocol.rs:17

注册表本身的并发与分发逻辑属于 第 2 章;这里只认"部件是谁、字段是什么"。


3. 三原语的数据结构

这一节逐个拆开三原语在 Rust 里的字段。记住关注点只有两个:每个字段是什么意思契约上什么可空

3.1 Function —— "能做什么"

一个函数在引擎内存里就是 Function(function.rs:29):

字段类型含义
handlerArc<HandlerFn>真正被调用时执行的闭包
_function_idString函数 id(下划线前缀:内部保存,当前少直接读)
_descriptionOption<String>人类可读描述
request_formatOption<Value>输入的 JSON Schema(可选)
response_formatOption<Value>输出的 JSON Schema(可选)
metadataOption<Value>任意附加元数据

handler 的签名是契约的核心(function.rs:25 HandlerFn):

// function.rs:24-26 —— 真实签名
type HandlerFuture = Pin<Box<dyn Future<Output = FunctionResult<Option<Value>, ErrorBody>> + Send>>;
pub type HandlerFn =
dyn Fn(Option<Uuid>, Value, Option<Arc<Session>>) -> HandlerFuture + Send + Sync;

读法:handler 接收 (① invocation_id:可空的调用编号、② data:输入 JSON、③ 可选的 RBAC Session),异步返回一个 FunctionResult<Option<Value>, ErrorBody>call_handler (function.rs:39)只是把这仨参数转交给闭包并 .await

返回值 FunctionResult 有四个变体(function.rs:18),这是"一次执行到底算什么结局"的完整枚举:

变体含义
Success(T)成功,带一个返回值(这里 T = Option<Value>,即可有可无的 JSON)
Failure(E)失败,带一个 ErrorBody
Deferred结果稍后异步给出,现在当场返回(用于排队/延迟场景)
NoResult执行完成,但没有要返回的结果

直觉:Success(None) 是"办完了、返回空";NoResult 是"办完了、根本不产出结果"; Deferred 是"我先接下,结果之后另行送达"。三者区分对 调用生命周期 如何对回结果很关键,本章只需记住它们的语义。

FunctionsRegistry(function.rs:59)是这些函数的家,契约上提供四个基本动作:

方法作用位置
register_function(id, fn)按 id 存入;同 id 会覆盖并打 warnfunction.rs:99
remove(id)按 id 删除function.rs:120
get(id)按 id 取回一份克隆(Option<Function>)function.rs:134
functions_hash()把当前所有函数 id 排序后取调试串,作为"函数集指纹"function.rs:87

functions_hash() 的契约值得记:它先把 id 收进 HashSet、再排序、再 format!("{:?}", …), 所以结果与插入顺序无关、可重复——测试 registry_functions_hash_deterministic_sorted (function.rs:254274)锁定了这一点。它用来判断"函数集变没变"(发现与热重载会用), 具体消费方在 第 5 章。注册表的并发存储(DashMap)与作用域机制留给 第 2 章

3.2 Trigger 与 TriggerType —— "何时做"

Trigger 是一条绑定(trigger.rs:175):

字段类型含义
idString触发器唯一 id(也是相等性的唯一依据,见下)
trigger_typeString属于哪类触发器(如 "http""cron")
function_idString命中时要调用的函数 id
configValue这条触发器的配置(如 HTTP 路径、cron 表达式)
worker_idOption<Uuid>归属哪个 worker(内置进程内触发器为 None)
metadataOption<Value>附加元数据,None 时序列化省略

关键契约:TriggerEq/Hash 只看 id(trigger.rs:186192):

// trigger.rs:186-196 —— 相等与哈希都只用 id
impl PartialEq for Trigger {
fn eq(&self, other: &Self) -> bool { self.id == other.id }
}
impl std::hash::Hash for Trigger {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) { self.id.hash(state); }
}

含义:两个 Trigger 只要 id 相同就被视为"同一个",哪怕 config/function_id 不同。 放进 HashSet 或按 id 建索引时,以 id 去重;换句话说 id 是触发器的主键,重复 id = 覆盖同一条。

TriggerType 是"某一类触发器"的定义(trigger.rs:61):

字段类型含义
idString类型 id(如 "http")
_descriptionString描述
trigger_request_formatOption<Value>注册一条触发器时 config 的 schema
call_request_formatOption<Value>触发器命中、调用 handler 时负载的 schema
call_response_formatOption<Value>handler 必须返回的 schema(多数类型不约束,为 None)
registratorBox<dyn TriggerRegistrator>这类触发器"怎么挂载/卸载"的行为对象
worker_idOption<Uuid>提供该类型的 worker(进程内内置为 None)

三个 *_format 别混:trigger_request_format 是"怎么登记这类触发器", call_request_format 是"命中时发给函数的负载长什么样",call_response_format 是"函数什么" (仅 http 声明了返回 HttpCallResponse,见 trigger.rs:153 call_response_format_for)。 这些 schema 由 TriggerType::new(trigger.rs:72)按 id 查 trigger_formats 模块自动填入。

TriggerRegistrator 是行为契约(trigger.rs:163),只有两个异步方法:

// trigger.rs:163-172 —— 每类触发器都要实现"如何挂载/卸载一条触发绑定"
pub trait TriggerRegistrator: Send + Sync {
fn register_trigger(&self, trigger: Trigger)
-> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>>;
fn unregister_trigger(&self, trigger: Trigger)
-> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>>;
}

它把"http 触发器具体怎么开一条路由、cron 怎么排一个定时"这些实现抽象掉——本章只需知道 "每类触发器都得提供这对方法"。各内置类型的真实实现第 4 章

内置触发器类型表(trigger.rs:16 BUILTIN_TRIGGER_TYPES)是一张静态"类型 id → 提供它的 worker"映射:

trigger-type id提供它的 worker
httpiii-http
croniii-cron
subscribeiii-pubsub
stateiii-state
durable:subscriberiii-queue
streamiii-stream
stream:joiniii-stream
stream:leaveiii-stream
logiii-observability
traceiii-observability
configurationconfiguration

它的用处是报错时给出可操作提示:当有人想注册某类触发器、而对应 worker 没在项目里, builtin_trigger_type_owner(trigger.rs:37)反查出 worker 名,报 RegisterTriggerError::UnknownBuiltin (trigger.rs:49),提示"Run: iii worker add "。它同时也是发现机制里把触发器类型归拢到 所属 worker 的依据(见 trigger.rs:30 的说明注释)。

3.3 Service —— "怎么归类"

Service 在协议里没有独立的内存结构体,它就是一条声明报文 Message::RegisterService(protocol.rs:112):

字段类型含义
idString服务 id
nameString(serde(default))显示名;缺省时反序列化为空串
descriptionOption<String>描述,None 省略
parent_service_idOption<String>父服务 id ——由此形成可嵌套的服务树

parent_service_id 是 Service 唯一的"结构"信息:靠它把服务串成层级/命名空间。


4. 线上协议:Message 枚举

三原语靠"声明"进入引擎,而"声明"就是往 WebSocket 里发 Message。这一节讲这套报文的语法规则和每个变体。

4.1 序列化约定:tag = type,全小写

整个枚举顶上挂着这行(protocol.rs:40-42):

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum Message {}

两条硬约定,记牢:

  1. tag = "type":是"内部标签"式枚举——每条 JSON 用一个 "type" 字段区分变体,其余字段与变体内容平铺在同一层(不是 {"RegisterFunction": {…}} 的嵌套形式)。
  2. rename_all = "lowercase":标签把变体名整体转小写、不加分隔符。于是 RegisterFunction"registerfunction"InvokeFunction"invokefunction"TriggerRegistrationResult"triggerregistrationresult"UnregisterTrigger"unregistertrigger"。测试 deserialize_unregister_trigger_without_type(protocol.rs:222)等直接用小写字面量验证了这点。

同样的 tag=type + 小写约定也用在辅助枚举 TriggerAction(protocol.rs:34),所以它序列化成 {"type":"enqueue","queue":"…"}{"type":"void"}

另一条贯穿全枚举的约定:大量字段带 #[serde(skip_serializing_if = "Option::is_none")]#[serde(default)]——None 的可选字段在线上直接省略,反序列化时缺省即取默认。测试 serialize_invoke_function_without_action_omits_field(protocol.rs:380)确认了 action/ traceparent/baggageNone 时报文里不出现这些键。

4.2 变体总览

按用途把 15 个变体分四组(全部在 protocol.rs:42 Message 内):

A. 声明原语(worker → 引擎)

变体关键字段干什么
RegisterFunctioniddescription?request_format?response_format?metadata?invocation?登记一个函数71
UnregisterFunctionid注销一个函数82
RegisterTriggeridtrigger_typefunction_idconfigmetadata?登记一条触发绑定51
UnregisterTriggeridtrigger_type?注销一条触发绑定(类型可空,见下)66
RegisterTriggerTypeiddescriptiontrigger_request_format?call_request_format?登记"一类触发器"43
RegisterServiceidnamedescription?parent_service_id?登记一个服务112

B. 调用与回执

变体关键字段干什么
InvokeFunctioninvocation_id?function_iddatatraceparent?baggage?action?发起一次函数调用85
InvocationResultinvocation_idfunction_idresult?error?traceparent?baggage?回一次调用的结果98
TriggerRegistrationResultidtrigger_typefunction_iderror?回一条触发器注册的结果59

C. 保活 / 握手

变体干什么
Ping / Pong心跳保活121 / 122
WorkerRegistered引擎下发 worker_id,确认 worker 握手完成123

4.3 两个必须记住的契约细节

invocation_id 一边可空、一边必填。 发起端 InvokeFunction.invocation_idOption<Uuid>(protocol.rs:86,可为 null),而回执端 InvocationResult.invocation_id必填 Uuid(protocol.rs:99)。直觉:invocation_id 是"这次调用的对账号"—— 需要同步拿结果的调用,引擎会给个 id,worker 回执时必须原样带回以便对上号;不需要结果的 "发了就走"式调用(见下 action)可以不带 id(null)。测试 deserialize_invoke_function_with_enqueue_action (protocol.rs:301)里就是 "invocation_id": null

action 决定这次调用怎么"处置"。 InvokeFunction.actionOption<TriggerAction> (protocol.rs:34),两个取值:

TriggerAction线上形态含义
Enqueue { queue }{"type":"enqueue","queue":"payment"}把这次调用投递到名为 queue 的(持久)队列
Void{"type":"void"}发了就走,不期待结果

action 缺省(字段不出现)则是普通同步调用、期待 InvocationResult。测试 deserialize_invoke_function_with_void_action / _without_action(protocol.rs:330357) 覆盖了这三种情况。分发/排队的具体行为属于 调用生命周期,本章只认字段。

UnregisterTrigger.trigger_type 可空。 注销触发器时可以只给 id、不给 trigger_type (protocol.rs:68,serde(default)),测试 deserialize_unregister_trigger_without_type (protocol.rs:222)证实缺省时它是 None——与 §3.2 "触发器以 id 为主键"一致:光凭 id 就能定位。


5. 报文里的辅助负载类型

除了 Message 变体本身,协议里还有几个被变体引用或独立传递的负载结构体,这里逐个点明字段含义。

5.1 HttpInvocationRef —— 外部 HTTP 函数的引用

当一个"函数"其实是外部 HTTP 端点时,RegisterFunction.invocation 带上它(protocol.rs:17):

字段类型含义
urlString目标 URL
methodHttpMethod(serde(default))HTTP 方法,缺省为 POST(default_http_method,protocol.rs:29)
timeout_msOption<u64>超时毫秒
headersHashMap<String,String>附加请求头
authOption<HttpAuthConfig>鉴权配置(如 bearer + token_key)

5.2 ErrorBody —— 统一错误载体

InvocationResult.errorTriggerRegistrationResult.error 等都用它(protocol.rs:175):

字段类型含义
codeString机器可读错误码
messageString人类可读消息
stacktraceOption<String>可选堆栈,None 省略

它有个便捷构造 ErrorBody::new(code, message)(protocol.rs:183,stacktraceNone) 和 Display 实现(protocol.rs:192,格式 code: message)。

5.3 WorkerMetrics —— worker 健康指标

一个辅助负载(不是 Message 变体),用于健康监控(protocol.rs:142)。字段分四类,除 timestamp_msruntime 外均可空省略:

字段含义
内存(字节)memory_heap_used/totalmemory_rssmemory_external各类内存用量
CPUcpu_user_microscpu_system_microscpu_percent用户/系统微秒、占比
运行时event_loop_lag_msuptime_seconds事件循环延迟、运行时长
元数据timestamp_ms(必填)、runtime(必填,如 "node")采样时间戳、运行时名

源码顶部专门写了 JavaScript 精度注记(protocol.rs:128-140):这些 u64 理论上可超过 JS 的 Number.MAX_SAFE_INTEGER(2⁵³−1),但内存要 >9 PB、CPU 微秒要 285 年才会掉精度,实务无碍; 需要绝对精度时 JS 侧按 BigInt 解析。

5.4 StreamChannelRef —— 流通道引用

流式场景用的辅助负载(protocol.rs:207):

字段类型含义
channel_idString通道 id
access_keyString访问密钥
directionChannelDirection方向:read / write,缺省 read(protocol.rs:200)

ChannelDirection 同样是 rename_all = "lowercase",线上写作 "read" / "write"


6. 巧妙之处(可借鉴)

  • 一个联合枚举打通双向通道。 收发同一个 Message,内部标签 type 区分意图——协议表面积 极小,新增消息就是加一个变体,serde 自动搞定编解码(protocol.rs:42)。
  • "可省略即默认"贯穿全协议。 大量 skip_serializing_if/default 让线上报文只带非默认字段, 既省带宽又让协议向后兼容(旧 worker 不发新字段,反序列化取默认)。
  • invocation_id 双态设计。 发起端可空、回执端必填,用同一个字段既表达"要不要对账",又 强制回执必须可对账(protocol.rs:86 vs 99),无需另设标志位。
  • 触发器以 id 为唯一身份。 Eq/Hash 只看 id(trigger.rs:186),使去重、覆盖、按 id 注销都变成"主键操作",注销报文甚至可以省掉 trigger_type
  • 内置类型表兼顾报错与发现。 一张 &[(&str,&str)] 静态表(trigger.rs:16)既产出可操作的 "run: iii worker add …" 提示,又充当把触发器类型归拢到 owner worker 的真源(trigger.rs:30 注释)。

7. 边界与局限(本章范围)

  • 本章讲字段含义与线上契约。路由、分发、调度、注册表并发、作用域/热重载、各触发器类型的 真实挂载逻辑一律不在此——分别见 第 2 章第 3 章第 4 章
  • WorkerMetricsStreamChannelRef 是辅助负载类型,不是 Message 的变体;它们如何被具体消息 携带、由谁消费,超出本章(涉及发现/可观测,见 第 5 章)。
  • worker 与引擎完成握手的时序(何时发 WorkerRegistered、如何认领 worker_id)属于 SDK 与 worker 握手;本章只标出该报文的存在与字段。
  • Function 的 handler 是内存中的闭包 Arc<HandlerFn>(function.rs:25),它跨 WebSocket 序列化——线上传的是 RegisterFunction 声明(id/schema/可选 HTTP 引用),真正的可执行体留在 worker 侧。

8. 代码地图(导航索引)

主题文件符号
全部报文的联合枚举 / tag=type 小写engine/src/protocol.rsMessage
外部 HTTP 函数引用engine/src/protocol.rsHttpInvocationRefdefault_http_method
调用处置动作engine/src/protocol.rsTriggerAction(Enqueue/Void)
统一错误体engine/src/protocol.rsErrorBodyErrorBody::new
worker 健康指标engine/src/protocol.rsWorkerMetrics
流通道引用 / 方向engine/src/protocol.rsStreamChannelRefChannelDirection
函数内存表示engine/src/function.rsFunctionFunction::call_handler
handler 签名engine/src/function.rsHandlerFnHandlerFuture
执行结局枚举engine/src/function.rsFunctionResult(Success/Failure/Deferred/NoResult)
函数注册表(增删查/指纹)engine/src/function.rsFunctionsRegistryregister_functionremovegetfunctions_hash
触发绑定 / 以 id 为身份engine/src/trigger.rsTriggerimpl PartialEq for Triggerimpl Hash for Trigger
触发器类型定义engine/src/trigger.rsTriggerTypeTriggerType::new
触发器挂载/卸载契约engine/src/trigger.rsTriggerRegistrator
内置类型 → owner workerengine/src/trigger.rsBUILTIN_TRIGGER_TYPESbuiltin_trigger_type_owner
注册报错(可操作提示)engine/src/trigger.rsRegisterTriggerError(UnknownBuiltin/Unknown)