Rivet Actors Rust SDKrivetkit设计约束与开发指南【免费下载链接】actorsRivet Actors are the primitive for stateful workloads. Built for AI agents, collaborative apps, and durable execution.项目地址: https://gitcode.com/GitHub_Trending/riv/actorsRivet Actors 将 AI Agent、协作文档、实时聊天等有状态工作负载抽象为可休眠、可持久化的 Actor 原语。本文聚焦开源仓库中 rivetkit-rust/packages/rivetkit/CLAUDE.md 所定义的 Rust SDK 设计约束结合 rivetkit-rust/packages/rivetkit 的源码、测试与官方示例系统讲解rivetkit的架构定位、trait 化 Actor 编程模型、状态持久化规则、事件循环兼容层以及职责边界帮助 Rust 开发者正确、高效地使用该 SDK 编写有状态 Actor 应用。一、rivetkit 的架构定位rivetkit-core之上的薄类型封装rivetkit不是一套独立的 Actor 运行时而是对底层rivetkit-core的薄类型封装thin typed wrapper。这一点在包描述与源码中均有明确体现Cargo.toml 将包描述为 Rust SDK for RivetKit actors, actions, events, queues, and test harnesses其核心依赖为rivetkit-core、rivetkit-client序列化采用ciboriumCBOR异步运行时为tokio。src/lib.rs 大量pub use rivetkit_core::...直接转出底层类型说明封装层的核心职责是类型化与边界编解码而非重新实现运行时逻辑。该封装策略对应了 rivetkit-rust/CLAUDE.md 中定义的 RivetKit 运行时边界跨运行时的字节边界统一使用Vecu8形状的数据SQL 边界类型显式共享避免从 NAPI 专属的数据库包装器推导运行时 API 契约。rivetkit与rivetkit-typescript保持尽力而为的 API 对齐best-effort parity使得同一套 Actor 概念在两个语言 SDK 中可以映射到等价的结构。设计要点rivetkit的CtxA方法一律作为ActorContext的薄透传thin pass-through不承载核心业务逻辑新加的封装方法只在边界处做 CBOR 编解码并委托给 core。因此使用rivetkit编写的代码其底层行为由rivetkit-core保证。二、设计约束总览CLAUDE.md 用五条硬性约束划定了 SDK 的功能边界。下面逐条展开并给出源码证据与实战含义。1. 不提供vars临时变量 APITypeScript SDK 中存在ctx.vars临时变量 API其存在原因是 TS 用户状态存放在框架中。而 Rust 中临时状态ephemeral state直接作为 Actor struct 的普通字段例如 chat-room-rust 中ChatRoom的started_at_ms: i64持久化状态persisted state即关联类型Actor::State由框架负责保存。因此不要在rivetkit中添加vars访问器来镜像 TypeScript。Rust 开发者应依赖 Rust 的结构体所有权模型将生命周期内数据放在 struct 字段上将需要跨休眠/恢复的数据放在State中。2. 主 API 为Actor 每个 action 一个HandlesM实现封装层的主入口 API 是traitActor 针对每个 action 的HandlesM实现生命周期与分发逻辑集中在run_actor中run_actor位于 src/start.rs它一次性完成启动阶段输入解码、状态创建、create、on_create、on_start并将失败原样上报给运行时握手随后进入事件循环分发。用户 Actor 应通过register_actor/register_actor_with注册见 src/registry.rs而不是直接编写事件循环。trait 化模型下Action 的完整定义如下src/action.rspub trait Action: serde::Serialize DeserializeOwned Send Sync static { type Output: serde::Serialize DeserializeOwned Send static; const NAME: static str; }HandlesAtraitsrc/action.rs要求为每个 Action 实现一个返回Future的handle方法pub trait HandlesA: Action: Actor Sized { type Future: FutureOutput ResultA::Output Send static; fn handle(self: ArcSelf, ctx: CtxSelf, action: A) - Self::Future; }Actor::Actions关联类型使用元组tuple声明可处理的 action 集合宏为从 1 到 128 个元数TUPLE_ARITY_MAX 128见 src/action.rs自动生成ActionSet实现。分发时按name匹配并用encode_positional/decode_positional在边界完成 CBOR 的位置参数编解码。3. 旧事件循环 API 保留兼容新代码一律使用 trait APIRegistry::register、Registry::register_with、Start、RuntimeEvent构成的旧式事件循环 API仍然可用但已在源码中标记为#[deprecated]src/registry.rs 对register的弃用说明为 use register_actor/register_actor_with and implement Actor Handles instead。src/start.rs 中的StartA携带ctx、input、is_new、snapshot、hibernated、events等字段EventsA::recv()返回RuntimeEventA供手动事件循环消费。新示例与文档应使用 trait API事件循环仅作为迁移期的兼容路径。对应地trybuildUI 测试如 action_set_missing_handle.rs 及其.stderr快照会编译失败并给出明确提示确保漏掉Handles实现的编译期错误可读、可诊断。4. 状态修改自动标记 dirtyrequest_save仅用于显式保存点这是最容易影响持久化正确性的规则。源码 src/context.rs 的实现细节如下Ctx::state_mut()src/context.rs先设置dirty标志为true再返回写锁保护的可变引用StateMut的Dropsrc/context.rs在释放写锁后自动调用inner.request_save(...)无需用户手动操作Ctx::set_state()src/context.rs整体替换状态并立即触发request_save。因此通过state_mut()/set_state()修改的持久化状态会自动标记为 dirty 并触发保存request_save()src/context.rs只应在需要显式保存点例如希望状态尽快落盘、或需要配合request_save_with_opts时调用事件循环中收到SerializeState事件时src/start.rs运行时先判断state_dirty()若 dirty 则先执行on_state_changehook再编码StateDeltaCBOR并清除 dirty 标志。实战建议不要在每次字段修改后额外调用request_save这会造成冗余的保存请求依赖 dirty 标记机制只在需要显式持久化时机时使用。5. 不支持 workflow 事件工作流引擎归属rivetkit-typescriptRust Actor永远不会托管 workflow工作流引擎由rivetkit-typescript拥有。对应地不要在 Rust SDK 中暴露ActorEvent::WorkflowHistoryRequested/WorkflowReplayRequested的Event变体或类型在Event::from_coresrc/event.rs中对这类 core 变体使用unreachable!处理src/start.rs 中若 core 侧仍然投递了这两个事件则返回ActorRuntime::NotConfigured错误workflow history / workflow replay。该约束保证 Rust SDK 的职责边界清晰Rust 处理有状态 Actor、action、队列、定时与连接workflow 编排交给 TypeScript 生态。三、Actortrait 全貌生命周期、HTTP、WebSocket 与连接Actortrait 定义在 src/actor.rs它通过 12 个关联类型和约 15 个可覆写方法定义了 Actor 的完整行为面。关联类型如下关联类型含义说明State持久化状态需Serialize DeserializeOwned由框架保存Input创建输入需DeserializeOwned Default缺失时用默认值Actionsaction 集合实现ActionSetSelf的元组Events可广播事件集合实现EventSet的元组Queue队列消息集合实现QueueSetSelf的元组ConnParams连接参数需DeserializeOwned DefaultConnState连接状态需Serialize DeserializeOwned CloneAction分发时的 Action 类型通常为action::Raw关键常量与方法const HAS_DATABASE: bool声明 Actor 是否使用用户数据库。注册时 src/registry.rs 会将其并入ActorConfig::has_database同时内部存储state、KV、queue始终使用 SQLite因此在未编译sqlite-localfeature 时actor_config会强制remote_sqlite true即通过 engine 路由 SQLite。const CONCURRENT_HTTP_CALLBACKS: bool falseMAX_CONCURRENT_HTTP_CALLBACKS: usize 128MAX_CONCURRENT_LIVE_HTTP_CALLBACK_STARTS: usize 0开启并发 HTTP 回调时run_actor会为 Standard 与 Live 两类回调建立独立的信号量池src/start.rs池满时回复 HTTP 429。classify_http_request(Request) - HttpCallbackClass可将请求分类到 Standard / Live 回调池。admit_http_request(...) - HttpCallbackAdmission允许 Actor 在回调构造/注册前同步执行准入逻辑返回Untracked/Tracked { release }/Reject。Tracked的release被包装在HttpCallbackAdmissionGuard中在回调回复交给 core 或回调被 drop 时自动执行src/start.rs。生命周期 hookcreate_state、create、on_create、on_start、run、on_state_change、on_sleep、on_destroy。连接 hookcreate_conn_state、on_before_connect、on_connect、on_disconnect、on_subscribe。HTTP / WebSocketon_fetch返回Response、on_fetch_response返回ActorHttpResponse默认为 buffered可返回StreamingResponse实现流式响应、on_websocket默认bail!(websockets not supported)。所有 hook 默认实现均为空操作或保守默认值最小化 Actor 只需实现create_state/create否则返回ActorRuntime::NotConfigured。运行时启动流程Registry::start()src/registry.rs是独立二进制入口它会阻塞直到收到 SIGINT/SIGTERM然后取消并排空。运行模式由环境变量RIVETKIT_RUNTIME_MODE决定Envoy默认持有一个长生命周期出站 envoystart_envoy直至信号后排空Serverless启动 HTTP listener在首个请求时惰性启动并缓存 envoy对应into_serverless_runtime。连接设置包括用于派生或复用本地 engine 的RIVET_ENGINE_BINARY_PATH从环境变量读取。由于 Rust 没有隐式运行时保持进程存活Registry::start会阻塞需要编程式生命周期控制测试、嵌入场景时应使用serve/serve_with_config并自行驱动CancellationToken。四、状态、输入与快照Start结构解析run_actor消费的StartAsrc/start.rs由 core 的ActorStart通过wrap_start包装而来包含ctx: CtxA类型化上下文input: InputA启动输入。is_present()判断是否携带字节decode()/decode_or_default()/decode_or(f)解码 CBOR 输入缺失输入时回退到A::Input::default()与 TypeScriptcreateState(undefined)语义一致。若输入缺失且类型不是()则返回ActorRuntime::MissingInput。is_new: bool是否为新建 Actorsnapshot: Snapshot休眠恢复的快照。is_new()判断是否为全新实例decode::S()解码状态快照空快照返回Nonehibernated: VecHibernatedA休眠时保留的连接携带连接状态events: EventsA事件流recv()/try_recv()返回RuntimeEventA。启动阶段src/start.rs作为一个整体可失败单元执行若有快照则解码恢复状态否则调用create_state随后A::create、is_new时的on_create、以及on_start。整个阶段通过startup_readyoneshot channel 与运行时握手失败会以真实原因而非通道关闭的泛化错误上报。RuntimeEventAsrc/event.rs共有 10 个变体Action、Http、QueueSend、WebSocketOpen、ConnOpen、ConnClosed、Subscribe、SerializeState、Sleep、Destroy。值得注意的是Events::recv会内部消化ConnectionOpen、DisconnectConn、RunWake这类运行时握手事件src/start.rs只向用户暴露业务相关事件。五、CtxA能力地图状态、SQL、调度、广播与客户端CtxAsrc/context.rs内部持有ActorContextStateCell值 dirty 标志 惰性Client 可选的当前ConnCtx。核心能力分类如下状态访问state()只读、state_mut()写 自动 dirty Drop 时 request_save、set_state()、state_dirty()、clear_state_dirty()、set_initial_state()。SQLite 访问sql()返回SqliteDbdb_exec/db_query/db_execute/db_run提供 CBOR 边界的便捷方法。SqliteDbExt::transactionsrc/sqlite.rs提供 commit-on-success 事务助手支持命名事务与超时。调度与定时schedule()支持after延迟、at定点、cancel、get、listcron()支持setcron 表达式 timezone max_history、every间隔 Duration、get/list/delete/history。唤醒控制keep_awake(future)对应 TSctx.waitUntilfuture 在途期间 Actor 不会休眠、keep_awake_region()返回 guarddrop 时释放、abort_signal()/aborted()运行时销毁信号、register_task注册运行时拥有的后台任务shutdown 时与优雅期限竞争。事件与广播broadcast(name, event)/emit(E)向连接广播事件、conns()/conns_vec()/disconnect_conn/disconnect_conns。生命周期控制sleep()请求休眠、destroy()、stop_with_error(message)带错误停止engine 记录为 crash 并应用崩溃处理、set_alarm。类型化客户端client()惰性构造rivetkit_client::ClientBare 编码 WebSocket 传输用于跨 Actor 调用TypedClientExtsrc/typed_client.rs提供get_typed/get_or_create_typed返回TypedActorHandleA可在编译期绑定 Actor 类型后安全调用其 action。连接上下文ConnCtxAid()、params()、state()/set_state()、send(name, event)、disconnect(reason)、is_hibernatable()。连接参数与状态同样以 CBOR 编解码。注kv()已被标记为 deprecated建议使用嵌入式 SQLitesql()或 Actor state 替代src/context.rsset_prevent_sleep/prevent_sleep同样已弃用为 no-op改用keep_awake或wait_until。六、队列Queue与QueueSet队列 API 与 action 保持同样的 trait 化结构src/queue.rsQueueMessageSerialize DeserializeOwned含NAME与Reply类型HandlesQueueM每个消息一个handle_queue实现QueueSetA元组组合支持最多 16 个消息按名称分发Ctx::queue()返回Queue_发送助手会 CBOR 编码消息体*_raw变体原样透传字节。事件循环投递RuntimeEvent::QueueSend其中wait/timeout_ms字段支持阻塞等待语义src/start.rs 中队列处理器不存在时返回ActorRuntime::NotFound。七、状态持久化的代码级验证约束 4自动 dirty与约束 2run_actor生命周期可由源码直接验证state_mut()在返回写守卫前dirty.store(true, Ordering::Release)StateMut::drop先drop(guard)释放锁再request_save—— 注释明确指出这镜像了 TypeScript 的 write-through 状态代理不依赖优雅关机过程来保证持久化SerializeState事件处理dirty 时先跑on_state_change再encode_state_deltaCBOR 编码StateDelta::ActorState最后clear_state_dirty。src/persist.rs 进一步提供state_delta/state_deltas/conn_hibernation_delta/conn_hibernation_removed_delta等显式 delta 构造工具供手动保存场景使用。八、测试与示例从 e2e 到 UI 编译测试rivetkit的测试体系覆盖多个层级In-process e2e 测试tests/test_harness_e2e.rs 通过test::setup(registry)src/test.rs派生或复用本地 engine解析顺序显式路径 →RIVET_ENGINE_BINARY_PATH→ 健康 engine → 本地工作区构建 → 缓存二进制 → 可选验证下载为并发测试分配唯一 pool 名避免跨路由随后用类型化 handle 发送 action 并断言往返结果。模块单元测试src/start.rs 内置LifecycleActor测试覆盖完整生命周期create_state→create→on_create→on_start→run→on_sleep/on_destroy以及连接预检、订阅拒绝、并发 HTTP 回调的 429 与 admission 释放计数。trybuild UI 测试tests/trybuild.rs 与 tests/ui/ 中的.rs.stderr快照验证漏写Handles/HandlesQueue实现等场景的编译错误信息质量。官方示例examples/chat-room-rust 是完整的 trait API 演示定义ChatRoomActor含HAS_DATABASE true、on_fetch默认 404 之外的 SQLite 建表、sendMessage/getHistory/getStats三个 action、newMessage事件广播、ctx.state_mut()计数其 main.rs 仅两行example_chat_room_rust::registry().start().await。九、快速上手与最佳实践总结最小可运行结构参考 test_harness_e2e.rsuse rivetkit::{Action, Actor, Ctx, Handles, Registry, action}; use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize)] struct Echo { value: String } impl Action for Echo { type Output String; const NAME: static str echo; } struct MyActor; impl Actor for MyActor { type State (); type Input (); type Actions (Echo,); type Events (); type Queue (); type ConnParams (); type ConnState (); type Action action::Raw; // 需要 SQLite 用户数据库时const HAS_DATABASE: bool true; } impl HandlesEcho for MyActor { type Future std::pin::PinBoxdyn std::future::FutureOutput anyhow::ResultString Send; fn handle(self: std::sync::ArcSelf, _ctx: CtxSelf, action: Echo) - Self::Future { Box::pin(async move { Ok(action.value) }) } } #[tokio::main] async fn main() - anyhow::Result() { let mut registry Registry::new(); registry.register_actor::MyActor(myActor); registry.start().await // 或 test::setup(registry) 进行进程内测试 }关键规则速查新代码一律使用ActorHandlestrait API旧事件循环 API 仅为兼容保留临时数据放 struct 字段持久化数据放State不要期望varsAPI依赖state_mut()/set_state()的自动 dirty 保存request_save()仅在需要显式保存点时使用Ctx方法是薄透传不要在应用层绕过类型化 API 直接操作 coreRust Actor 不托管 workflow相关工作流能力在rivetkit-typescript侧测试优先使用test::setup进程内 harness需要控制生命周期时用serveCancellationToken而非阻塞的start()未编译sqlite-localfeature 时内部存储默认走远程 SQLite经 engine 路由用户数据库需显式声明HAS_DATABASE。遵循以上设计约束即可写出与rivetkit-core运行时语义一致、可休眠、可持久化、可测试的 Rust Actor 应用。【免费下载链接】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），仅供参考