Rivet Gasoline 持久执行引擎深度解析Workflow、Signal、Message 与 Activity 的设计与实现【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actorsGasoline位于 engine/packages/gasoline是 Rivet Engine 中承载绝大多数持久化工作负载的 durable execution 引擎。本文基于仓库内部文档 Gasoline 概览 及其配套源码系统讲解 Gasoline 的核心抽象Workflows、Signals、Messages、Activities、Operations、工作流历史回放机制、步骤类型与组合方式以及 Activity 的正确拆分原则帮助你理解 Rivet 如何在从数据库重新唤醒工作流的语义下保证确定性执行与失败重试。一、Gasoline 是什么Rivet Engine 的持久执行内核Rivet 的核心原语是 Actors——面向有状态工作负载AI agents、协作应用、durable execution的运行时抽象。而在引擎侧真正让这些有状态语义落地的是 Gasoline一个运行在 Rivet Engine 之上的持久执行durable execution引擎。从内部文档的定义看Gasoline 由五类构件组成构件定位持久性Workflows类似 actor 的概念空闲时可以睡眠从内存中移除持久Signals工作流之间、以及外部服务如 api与工作流之间的通信持久Messages工作流向其他服务发送的fire-and-forget通信短暂ephemeralActivities原生 Rust 函数的薄封装每个 Activity 可独立重试执行结果持久Operations原生 Rust 函数的薄封装用于与 Gasoline 生态的干净互操作—从源码结构看这五类构件在 src 目录 中一一对应独立模块workflow.rs、signal.rs、message.rs、activity.rs、operation.rs并配合history/历史与位置计算、worker.rs执行调度、ctx/各种上下文等模块组织。模块划分与文档中的概念划分完全一致可以印证文档描述的准确性。二、Workflows基于历史回放的持久执行2.1 核心模型durable steps 与回放内部文档给出的工作流执行模型可以概括为三句话Workflow 由一系列 durable steps 组成。每一步完成后其结果作为 workflow history 存入数据库遇到等待时休眠当工作流执行到需要等待的步骤例如 signal 到达、sleep 超时前它会被移出内存直到事件发生或超时唤醒时回放历史当工作流从数据库被重新唤醒回内存时所有已完成的步骤会被replay——即不再真正执行代码、不再写入数据库而是用数据库中已有的数据来模拟当时的响应。这个记录历史 重放的模式recording/replay正是 durable execution 引擎的典型设计执行本身不保证不中断但通过把每个步骤的确定结果持久化任何时刻的崩溃或内存驱逐都可以从历史恢复出完全一致的语义状态。2.2 源码中的 Workflow 契约在 workflow.rs 中Workflowtrait 定义了工作流的核心契约#[async_trait] pub trait Workflow { type Input: WorkflowInput; type Output: Serialize DeserializeOwned Debug Send; const NAME: static str; const PRUNE_VARIANT: PruneVariant; async fn run(ctx: mut WorkflowCtx, input: Self::Input) - ResultSelf::Output; }几个值得注意的实现事实Input/Output必须可序列化工作流输入输出要能存入数据库并在回放时还原因此Serialize DeserializeOwned是硬性要求NAME是静态常量工作流按名称寻址这也是后文用 tags 匹配工作流能成立的前提之一PRUNE_VARIANT决定工作流完成后的数据命运对应源码中的PruneVariant枚举/// Determines how the data related to a workflow is handled after the workflow completes. pub enum PruneVariant { /// Delete all of the workflows data after it completes and is pruned (state history). #[default] All 0, /// Delete just the workflows history after it completes and is pruned (state persists). History 1, /// Dont delete any workflow data after it completes. None 2, }即工作流完成后可以选择删除全部数据state history默认值、仅删除 history 保留 state、或全部保留。工作流还通过StateGuard管理有锁的 JSON state以反序列化形式持有数据的同时在 Drop 时自动写回见 workflow.rs#L27-L65。一个最小的真实工作流例子来自 sleep_test.rs#[workflow(SleepTestWorkflow)] pub async fn sleep_test_workflow(ctx: mut WorkflowCtx, input: SleepTestInput) - Result() { ctx.sleep(input.duration_ms).await?; Ok(()) }ctx.sleep(...)就是一个会让工作流睡着的 durable step——执行到这一步后工作流会被移出内存直到时长到达。2.3 工作流历史的 Location 编码回放正确性的基础回放机制之所以成立依赖于历史事件的精确寻址。配套文档 Workflow History 所引用的 WORKFLOW_HISTORY.md 描述了底层机制每个事件step都有一个由坐标组成的location坐标由正整数ordinate构成。例如{1}— 第一个事件{1, 4}— 第一个事件的第四个子事件{0.1}— 插入在第一个事件之前的第一个事件{4, 0.3.1, 0.6}— 更复杂的插入位置。关键规则包括坐标从1开始而非0以便在{1}之前插入事件时不出现负数分支loop、closure 等内部使用会追加一个新坐标{1}处的分支子事件从{1, 1}开始向已有历史中插入新事件必须通过版本化步骤versioned steps且插入事件的版本必须高于其后继事件的版本否则回放时会报HistoryDiverged。这套编码支撑了两个实际工程需求在不知道工作流完整步骤列表的情况下动态分配位置以及修改一个已有部分历史的工作流。例如在{2}之前插入一个 v2 事件它会被分配为{1.1}在{1.1}与{1.2}均为 v2之间再插入则需要 v3位置为{1.1.1}在{1}之前插入则从{0.1}开始继续前置会累加{0.0.1}、{0.0.0.1}。对应实现位于 history/location.rs。三、Workflow Steps哪些操作是 durable 的3.1 步骤类型总览内部文档明确所有 workflow step 都是 durable 的——即在工作流回放时它们是 NO-OP直接跳过或返回之前已计算好的结果。具体包括步骤说明Signal发送给工作流的持久数据包。可配置 timeout 与 ready batch sizeMessage从工作流发送到订阅者的短暂数据包Activity对 bare function 的薄封装带自动指数退避重试Sub workflow发布另一个工作流和/或等待另一个工作流完成后再继续Loop允许智能处理 workflow history 的前提下反复执行同一闭包Sleep睡眠等待一段时间Removed events / version checks高级的 workflow history 步骤详见 WORKFLOW_HISTORY.md关于删除事件源码在 history/removed.rs 中实现了文档描述的ctx.removed::_()语义对已经执行过该步骤的工作流回放时跳过、不动数据库对尚未执行的工作流则插入一个removed事件。两种情况下 location 保持一致。关于 Loop实现细节也值得记录一个位于{2}的循环每次迭代占据一个独立分支{2, 1}、{2, 2}…迭代内的事件是该分支的子事件{2, 2, 1}、{2, 2, 2}。由于基于循环的状态机理论上可以无限运行Gasoline 会把已完成迭代的完整历史迁移到数据库中的forgotten event history区域回放时不再拉取——因为每次迭代彼此独立历史迭代不应影响当前迭代。相关逻辑对应 builder/workflow/lupe.rs。3.2 组合方式Join 与 Closure文档列出了两种组合手段Join— 并行运行多个 activity 或闭包Closure— 在 workflow history 中创建一个分支。3.3 一条硬约束workflow 体内不能直接跑 operation文档特别强调你无法在 workflow 主体中直接执行 operations必须先放进 activity。这符合 durable execution 的通用原则——workflow 体必须是可回放的确定性序列而外部 I/O网络请求、数据库调用等属于副作用只能发生在 activity 这个可独立重试的执行单元里。Operations 的定位则是与 Gasoline 生态的干净互操作对应实现为 operation.rs。四、Activitiesworkflow 的血肉与拆分原则4.1 Activity 契约Activity 是执行用户代码的地方是 workflow 的核心工作量所在。在 activity.rs 中Activitytrait 要求#[async_trait] pub trait Activity { type Input: ActivityInput; type Output: Serialize DeserializeOwned Debug Send; const NAME: static str; const MAX_RETRIES: usize; const TIMEOUT: std::time::Duration; async fn run(ctx: ActivityCtx, input: Self::Input) - ResultSelf::Output; }从常量定义可以看到每个 Activity 声明了自己的名称用于历史寻址与重试标识、最大重试次数MAX_RETRIES与超时时间TIMEOUT与文档中每个 Activity 失败时可独立重试的描述相互印证。重试由引擎按 backoff 策略自动执行。4.2 关键设计原则一个 Activity 只做一个动作内部文档对 Activity 的组成方式给出了明确指引应当把 Activity 组织成失败及后续重试不会造成副作用的形态即每个 Activity 应限制为 1 个action重试时不会被同一 Activity 的前一次执行所破坏。文档中的反例bad composition- Activity 1 - Transaction 1: 向数据库插入 user 行 - Transaction 2: 将 user id 插入 group 表如果该 Activity 在 Transaction 2 失败它会被带 backoff 重试。但此时 user 行已存在重试会在 Transaction 1 上直接报数据库错误——这个 Activity 永远无法通过重试成功。文档给出的两种修正方式将两个查询合并进一个事务或把两个事务拆成两个 Activity每个 Activity 一个事务。4.3 从测试用例看重试语义仓库的测试 activity_test.rs 覆盖了 Activity 的失败重试场景验证了Activity 失败 → 退避重试 → 结果写入历史的链路ctx/目录下的 standalone.rs 等文件则表明 Activity 也支持脱离 workflow 独立调用standalone 场景这与 Operations/Activities 作为原生 Rust 函数薄封装的定位一致。五、Signals外部与工作流之间唯一的持久通信通道5.1 定位与寻址Signals 是服务工作流之外的任何事物→ 工作流、以及工作流 → 工作流之间唯一的通信形式。发送一个 signal 需要满足以下条件之一目标工作流的workflow ID或与某个未完成的现存工作流匹配的tags。如果找到目标工作流signal 会被加入该工作流的队列工作流通过listen步骤消费 signal。若一个工作流完成时队列中仍有 pending signals这些 signal 会被标记为acknowledged并实质上被遗忘。从源码看Signaltrait 本身极简signal.rspub trait Signal { const NAME: static str; }真正的工作量在Listentraitlisten.rsWorkflowCtx::listen和WorkflowCtx::query_signal都通过它从工作流数据库中拉取 signal。同时仓库提供了join_signal!宏可以把多个 signal 类型合并成一个枚举来统一监听示例来自源码注释join_signal!(MyJoinSignal { MySignal, MySignal2, }); // 监听 match ctx.listen::MyJoinSignal() { MySignal(sig) println!(received MySignal {sig:?}), MySignal2(sig) println!(received MySignal2 {sig:?}), }5.2 相关测试signal_test.rs 与 listen_timeout.rs 分别验证了 signal 的收发与signal 监听带超时的行为对应文档中signal 可以有 timeout的参数说明。六、Messages短暂的状态推送Messages 是从工作流发出的短暂数据包定位是实时通信的状态更新而不是持久通信——文档明确消息没有被任何接收方消费也是可以的It is ok for messages to not be consumed by any receiver。接收方需要订阅其name及其tags的一个子集。源码 message.rs 印证了这一 pubsub 语义pub trait Message: Debug Send Sync Serialize DeserializeOwned static { const NAME: static str; const TAIL_TTL: std::time::Duration; fn subject() - String { format!(gasoline.msg.{}, Self::NAME) } }每个 Message 声明NAME并映射到gasoline.msg.{NAME}这一 pubsub subjectTAIL_TTL表示消息尾部保留时间说明消息确实只在有限时间窗口内可被订阅消费接收侧的PubsubMessageM携带ray_id、req_id、msg_ts与反序列化后的 body提供了消息何时被创建的时间戳等元数据。Signal 与 Message 的分工总结要确定送达工作流并影响其执行用 Signal要广播一个即时状态、丢了也无所谓用 Message。七、TagsID 不可用时的寻址约定Workflows、signals、messages 三者都使用 tags 作为唯一 ID 不可用时的便利寻址手段Signal 只要知道工作流的name 和 tags 的一个子集就可以发送给该工作流内部效率技巧signal 的 tags 应按从最独特到最不独特排序。文档给出的例子给定一个工作流tags 为 - namespace foo - type normal 则 signal 应先写 namespace foo再写 type normal即把区分度高的 tagnamespace放在前面低区分度的type放在后面便于内部做前缀式的高效匹配。tags 的工具实现位于 utils/tags.rs相关行为测试见 tags_test.rs。八、小结Gasoline 的完整心智模型把内部文档与源码证据串起来Rivet 引擎侧 Gasoline 的运转方式可以总结为写workflow 体声明一串 durable stepsactivity / signal listen / message / sub-workflow / loop / sleep每个 step 完成即把结果写入 workflow history带 location 编码睡遇到等待类 step 时工作流被移出内存等待事件signal、sleep 到期醒从数据库唤醒后按历史回放——已完成 step 不再执行直接复用记录的结果直到当前分叉点执行副作用所有外部 I/O 收敛在 activity 中按一个 activity 一个动作的原则拆分失败由MAX_RETRIES backoff 自动重试通信入向持久通信用 signal按 ID 或 name tags 寻址出向实时广播用 messagepubsub 订阅可丢失演进借助 versioned steps、check_version与removedstep可以安全地修改已有部分历史的工作流而不破坏回放一致性。延伸阅读路径均为仓库内文件docs-internal/engine/GASOLINE/OVERVIEW.md — 本文主体文档docs-internal/engine/GASOLINE/WORKFLOW_HISTORY.md — location 编码、版本化插入与删除事件细节docs-internal/engine/GASOLINE/OPERATIONS.md、WORKFLOW_PRUNING.md、GOTCHAS.md — 配套内部文档engine/packages/gasoline — 引擎实现重点模块workflow.rs、activity.rs、signal.rs、message.rs、history/、tests/workflows/。注意docs-internal目录属于引擎团队内部设计文档其中部分文件如 DESIGN_GUIDE.md目前仍是 TODO 占位状态引用其内容时应以本文所核实的源码实现为准。【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actors创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考