DORA 数据流容错指南重启策略、健康检查、输入熔断器与协调器状态持久化【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/doraDORADataflow-Oriented Robotic Architecture为机器人及 AI 数据流内置了一套覆盖节点、输入、数据流与多机部署多层次的容错机制节点可按策略自动重启、可检测失联上游与挂起节点、输入不可用时优雅降级协调器coordinator还能把状态持久化到磁盘以承受自身崩溃与重启。本文以官方容错文档 docs/fault-tolerance.md 为主体结合 daemon、运行期 API 与 coordinator-store 的源码实现系统讲解 DORA 容错体系的配置参数、底层工作原理、Rust 节点侧 API 以及完整实战场景帮助你为生产级数据流构建「防御纵深」。1. 功能总览从节点重启到协调器状态持久化DORA 的容错能力分布在数据流生命周期中的不同层次每一层解决一类故障特性作用范围配置方式重启策略Restart policies每个节点restart_policy、max_restarts、restart_delay等健康监控Health monitoring每个节点health_check_timeout、startup_timeout、health_check_interval数据流级输入超时Input timeouts每个输入input_timeout熔断器Circuit breaker自动触发由input_timeout触发可自动恢复NodeRestarted 事件下游节点上游重启时自动触发InputTracker APIRust 节点dora_node_api::InputTracker可观测性Observability整个 daemon周期性输出的原子计数器分布式健康Distributed health多 daemon协调器心跳监控协调器状态持久化协调器--store redb需redb-backendfeature从实现层面看容错逻辑主要位于 daemon 的 spawn/prepared.rs重启生命周期循环、running_dataflow.rs健康检查、输入熔断、节点运行状态跟踪与 fault_tolerance.rs原子计数器节点侧 API 位于 apis/rust/node/src/event_stream/input_tracker.rs协调器存储抽象位于 libraries/coordinator-store/src/lib.rs。2. 重启策略Restart Policies2.1 配置nodes: - id: my-node path: ./target/debug/my-node restart_policy: on-failure # never | on-failure | always max_restarts: 5 # 0 unlimited (默认: 0) restart_delay: 1.0 # 初始延迟秒 max_restart_delay: 30.0 # 指数退避的上限秒 restart_window: 300.0 # 经过这么多秒后重置计数器2.2 策略类型never默认——节点不重启故障正常向上传播。on-failure——仅当节点以非零退出码退出时重启正常退出退出码 0不重启。always——任何退出都重启但有两个例外数据流被用户停止dora stop或 Ctrl-C所有输入都已关闭且节点以非零退出码退出。2.3 内部重启判定流程当节点进程退出时daemon 按以下顺序评估是否重启对应 spawn/prepared.rs 中restart_loop的实现策略检查重启策略是否允许Never→ 不重启OnFailure→ 仅在退出码 ! 0 时重启Always→ 重启。禁用检查disable_restart是否被置位在所有输入关闭或通过stop_all手动停止时置位窗口检查若设置了restart_window且自首次重启以来窗口已过期将计数器重置为 0。上限检查若max_restarts 0且窗口内计数超过它则永久放弃。退避若设置了restart_delay按计算出的延迟睡眠醒来后重新检查disable_restart。重新拉起使用相同配置重新生成节点进程。daemon 在spawn/prepared.rs的生命周期循环中按节点实例跟踪重启状态。每个节点运行在独立的 tokio task 中因此一个节点的重启不会阻塞其他节点。源码中还有一个值得注意的细节若force_restart_next标志被置位由dora node restart手动重启触发它会覆盖restart_policy即使默认never策略也会重新拉起——这保证了 CLI「重新拉起节点」的承诺。2.4 指数退避设置restart_delay后daemon 会在重启前等待。每次尝试延迟翻倍指数退避并由max_restart_delay封顶。退避指数在内部被限制为 16 以防溢出2^16 65536x倍乘数。以restart_delay: 1.0与max_restart_delay: 10.0为例Attempt 1: wait 1s (1.0 * 2^0) Attempt 2: wait 2s (1.0 * 2^1) Attempt 3: wait 4s (1.0 * 2^2) Attempt 4: wait 8s (1.0 * 2^3) Attempt 5: wait 10s (封顶于 max_restart_delay) Attempt 6: wait 10s (封顶)退避睡眠期间daemon 持续监控disable_restart标志。如果节点在等待重启期间所有输入都关闭了重启会被取消并记录日志restart cancelled: inputs closed during backoff wait。2.5 重启窗口Restart Window设置restart_window后重启计数器会在窗口从当前窗口内第一次重启算起到期后重置。这实现了「N 次重启 / M 秒」的语义。例如max_restarts: 5、restart_window: 300.0表示「每 5 分钟最多重启 5 次」。如果窗口到期而未触及上限计数器重置节点获得新的 5 次尝试机会。2.6 关闭期间禁用重启当 daemon 停止数据流通过stop_all时它会在发送 Stop 事件之前对每个节点调用disable_restart()。这防止重启机制与关闭流程互相打架。disable_restart是一个ArcAtomicBool在 daemon 事件循环与节点的 spawn 生命周期 task 之间共享。2.7 NodeRestarted 事件节点重启时daemon 会向所有消费其输出的下游节点发送NodeRestarted事件。下游节点可据此重置内部状态或缓存记录上游恢复日志重新初始化连接或会话。事件携带重启节点的NodeId。下游节点通过事件流自动接收match event { Event::NodeRestarted { id } { println!(upstream node {id} restarted, resetting state); // 清除来自旧节点实例的缓存状态 } _ {} }daemon 通过dataflow.mappings找到下游节点——该映射把每个节点的输出关联到所有订阅的(receiver_node, input_id)对。每次重启每个唯一的接收者收到一条NodeRestarted事件。实现提示InputTracker也支持在收到NodeRestarted时自动把来自被重启节点的Closed输入切回可恢复状态见第 5 节两种机制可以配合使用。3. 健康监控Health Monitoring被动监控用于检测与 daemon 停止通信的挂起节点。health_check_interval: 2.0 # 秒 (默认: 5.0, 数据流级) nodes: - id: my-node path: ./target/debug/my-node health_check_timeout: 30.0 # 秒 (每节点) restart_policy: on-failure3.1 可配置的健康检查间隔health_check_interval是一个数据流级设置控制 daemon 检查节点健康的频率默认 5.0 秒。更小的值能更快发现挂起节点但带来更多开销。它应设置在数据流 YAML 的顶层而不是每个节点。在 lib.rs 的 daemon 运行循环中健康检查周期直接取自描述符的health_check_interval并通过tokio::time::interval生成一个产生Event::NodeHealthCheckInterval的间隔流实现还防御性地对零周期做回退退回到默认 5 秒避免tokio::time::interval因零周期而 panic。3.2 内部工作原理daemon 在配置的health_check_interval上执行健康检查扫描。每个RunningNode有一个last_activity: ArcAtomicU64字段存储最近一次通信的时间戳epoch 毫秒。它由节点的通信处理器node_communication/mod.rs在节点每次向 daemon 发送任何请求时原子更新事件订阅、输出发送等。健康检查函数遍历所有运行中的节点跳过未设置health_check_timeout的节点跳过尚未连接的节点不在connected_nodes中计算elapsed_ms now - last_activity若elapsed_ms timeout_ms记录警告并杀死节点进程。杀死后走正常的退出处理流程该流程会评估重启策略。因此health_check_timeout与restart_policy: on-failure组合可以自动恢复挂起节点。杀死事件会累加到health_check_kills计数器。3.3 仅限连接后的活性Post-Connection Liveness Only上述第 2 步意味着health_check_timeout约束的是连接后的活性而不是总启动时间。节点在首次订阅事件时在Node::init/DoraNode::init_from_env内部加入connected_nodes超时时钟只从那一刻开始计算。这是有意设计的last_activity被初始化为 spawn 时间戳如果没有连接门控一个合法冷启动Python import、模型权重加载超过health_check_timeout的节点会在启动中途被 SIGKILL——在restart_policy: always/on-failure下这将成为无法逃脱的重启循环。其代价是在订阅之前就挂起的节点——例如 import 或 init 代码中的死锁永远走不到Node::init——不会被health_check_timeout回收。要约束启动时间并恢复初始化前的死锁请设置startup_timeout。3.4 启动截止时间startup_timeouthealth_check_timeout监控节点连接后的状态而startup_timeout约束的是从进程 spawn 到节点与 daemon 连接订阅事件之间的时间。如果一个未连接的节点进程在 spawn 后startup_timeout秒内未连接daemon 记录警告、SIGKILL 该进程、递增startup_timeout_kills计数器并评估restart_policy。nodes: - id: worker path: ./target/debug/worker startup_timeout: 10.0 # spawn 后 10s 内未订阅则杀死 health_check_timeout: 30.0 # 连接后 30s 静默则杀死 restart_policy: on-failure与health_check_timeout一样startup_timeout在每个health_check_interval滴答时评估。3.5 什么算「活动」任何从节点到 daemon 的消息都算事件订阅请求输出数据发送通过共享内存或 TCP定时器滴答确认。从其他节点收到的正常输入数据不会重置定时器——节点必须主动与 daemon 通信。4. 输入超时与熔断器Input Timeouts and Circuit Breaker按输入per-input的超时用于检测上游节点停止生产数据。4.1 配置nodes: - id: downstream-node path: ./target/debug/downstream inputs: sensor_data: source: camera-node/frames input_timeout: 5.0 # 秒input_timeout按输入设置而不是按节点。不同输入可以有不同的超时。警告——不适用于按需输入on-demand inputs。input_timeout假定上游持续发布数据。不要把它设置在仅在响应出站请求时才被填充的输入上服务响应输入、action 结果输入或任何其他突发式/按需通道。自然的空闲期看起来与死掉的上游完全一样会错误触发熔断器。对于按请求等待请改用EventStream::recv_service_response(request_id, server, timeout)或EventStream::recv_action_result(goal_id, server, timeout)——它们约束每个单独请求而不触发数据流级熔断器。更多相关模式参见 docs/patterns.md 第 6 节「Fault tolerance for correlated patterns」。4.2 内部工作原理daemon 为每个带超时的输入维护一个InputDeadline源码见 running_dataflow.rsstruct InputDeadline { timeout: Duration, // 配置的超时 last_received: OptionInstant, // 最近一次数据到达时间 (None 未武装) }这些条目存储在RunningDataflow.input_deadlines中以(NodeId, DataId)为键。截止时间武装Deadline arminglast_received在数据流启动时为None。熔断器的时钟在第一条消息真正到达之前不会针对该输入计时见InputDeadline::is_timed_out。这防止了启动时合法空闲的输入产生误报——对应 dora-rs/adora#149。running_dataflow.rs中的单元测试unarmed_deadline_is_never_timed_out与armed_deadline_past_timeout_is_timed_out明确验证了「未武装的截止时间永远不会超时」这一行为。超时检测在同一个 5 秒健康检查间隔内运行。check_input_timeouts函数扫描所有input_deadlines条目仅对已武装的条目last_received Some(_)若last_received.elapsed() timeout该输入视为「broken」(node_id, input_id)对从input_deadlines移到broken_inputsdaemon 调用break_input()向下游节点发送InputClosed { id }如果节点的所有输入现在都已关闭且没有 broken/recoverable 的发送AllInputsClosed并禁用该节点的重启。截止时间的武装/重置每次数据到达输入时其last_received被设置为Some(Instant::now())——既武装了先前未武装的截止时间也重置了已武装的截止时间。4.3 熔断器自动恢复熔断器在RunningDataflow.broken_inputs中跟踪 broken 输入。当新的数据到达一个 broken 输入时数据正常投递给节点移除broken_inputs条目输入重新加入open_inputs创建新的InputDeadline重新开始计时向节点发送InputRecovered { id }事件递增circuit_breaker_recoveries计数器。这意味着恢复是完全自动的。如果上游节点通过重启策略重启并开始再次生产数据下游节点无缝恢复接收。另外running_dataflow.rs的forget_node_bookkeeping会在动态增删节点时清理对应的input_deadlines/broken_inputs条目避免陈旧条目被反复扫描或永远无法恢复。4.4 节点侧处理在 Rust 节点中在事件循环里处理这些事件use dora_node_api::{DoraNode, Event}; let (mut node, mut events) DoraNode::init_from_env()?; while let Some(event) events.recv() { match event { Event::Input { id, data, .. } { // 正常处理 } Event::InputClosed { id } { // 上游停止在该输入上生产数据。 // 你可以使用缓存数据、跳过处理、提醒操作员等。 } Event::InputRecovered { id } { // 该输入的上游恢复在线。 // 恢复正常处理。 } Event::Stop(_) break, _ {} } }5. InputTracker APIRustInputTracker辅助类型跟踪输入健康状态并为每个输入缓存最近收到的值让优雅降级变得简单。use dora_node_api::{DoraNode, Event, InputTracker, InputState}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } { // 有新数据可用 } Event::InputClosed { id } { // 输入超时 -- 回退到缓存数据 if let Some(stale_data) tracker.last_value(id) { // 使用 stale_data 作为回退 } } Event::Stop(_) break, _ {} } // 检查整体健康状况 if tracker.any_closed() { let closed: Vec_ tracker.closed_inputs(); // 记录日志或调整行为 } }5.1 内部设计InputTracker维护两个HashMap见 apis/rust/node/src/event_stream/input_tracker.rsstates: HashMapDataId, InputState——每个输入的当前状态Healthy 或 Closedcache: HashMapDataId, DoraArray——每个输入最近收到的值。在Event::Input时两张 map 都更新state Healthycache data 克隆。在Event::InputClosed时仅 state 变化cache 保留。在Event::InputRecovered时state 恢复为 Healthy。cache 从不被清除因此last_value()即使在输入关闭后也始终返回最近的数据。另外该结构还支持可选的「输入 → 源节点」映射提供后NodeRestarted事件会把源自被重启节点的任何Closed输入转移恢复与第 2.7 节的 daemon 事件形成闭环。注意DoraArray包装了一个Arc支撑的 Arrow 数组因此 cache 克隆是引用计数的廉价。5.2 API 参考方法返回类型描述new()InputTracker创建空 trackerprocess_event(Event)bool更新状态。若事件相关则返回 truestate(DataId)OptionInputState当前状态Healthy 或 Closedis_closed(DataId)bool检查输入是否关闭last_value(DataId)OptionDoraArray最近收到的值关闭时也可用closed_inputs()VecDataId所有当前关闭的输入any_closed()bool若任意被跟踪输入关闭则返回 true6. 可观测性Observabilitydaemon 使用原子计数器FaultToleranceStats定义在 fault_tolerance.rs跟踪容错事件并在健康检查间隔内每 5 秒输出一次汇总日志。6.1 计数器计数器类型递增时机restartsAtomicU64发起节点重启时在 spawn 生命周期中health_check_killsAtomicU64节点被健康检查杀死无响应startup_timeout_killsAtomicU64节点被启动看门狗杀死init 前未连接input_timeoutsAtomicU64输入超时触发熔断器跳闸circuit_breaker_recoveriesAtomicU64数据到达 broken 输入自动恢复所有计数器使用Ordering::Relaxed因为它们是信息性的不需要严格的有序保证。FaultToleranceStats::any_nonzero()提供一个快捷判断供日志路径决定是否输出汇总。6.2 日志输出当任一计数器非零时daemon 输出一条结构化日志INFO fault tolerance stats restarts3 health_kills0 input_timeouts1 cb_recoveries1这些计数器对 daemon 进程的整个生命周期是累积的不会在数据流之间重置。7. 分布式健康Distributed Health在多 daemon 部署中协调器监控 daemon 心跳。7.1 协议心跳间隔3 秒协调器向每个 daemon 发送心跳断连阈值30 秒无响应检测每次心跳扫描时协调器移除超过阈值未响应的 daemon通知协调器向所有剩余 daemon 广播PeerDaemonDisconnected { daemon_id }。7.2 DaemonInfoConnectedMachinesCLI 查询返回VecDaemonInfopub struct DaemonInfo { pub daemon_id: DaemonId, pub last_heartbeat_ago_ms: u64, // 距上次心跳的毫秒数 }这让监控工具能够检测「活着但响应慢」的 daemon。7.3 Daemon 侧处理daemon 收到PeerDaemonDisconnected时记录一条结构化警告WARN peer daemon disconnected daemon_idmachine-B目前这只是信息性的。未来的工作可能包括自动迁移断连 daemon 上的节点。8. 协调器状态持久化Coordinator State Persistence默认情况下协调器把所有状态保存在内存中。若协调器进程崩溃或被重启所有正在运行的数据流信息都会丢失——daemon 继续运行但变成孤儿用户必须手动重新运行数据流。redb 存储后端解决了这个问题它使用 redb一个纯 Rust 嵌入式键值存储基于写时复制 B 树天生崩溃安全把协调器状态持久化到磁盘上的单个文件。8.1 设计无状态协调器 有状态后端协调器本身保持无状态K8s 意义上的——它可以在任何时候停止和重启。所有持久状态都位于CoordinatorStoretrait 后面的存储后端中Coordinator (无状态进程) | v CoordinatorStore trait | -- InMemoryStore (默认, 无持久化) -- RedbStore (持久化到 ~/.dora/coordinator.redb)这种分离意味着协调器事件循环在正常运行期间从不读取文件系统仅在启动恢复时所有状态变更在明确定义的持久化点写入存储存储可以在不改变协调器逻辑的情况下替换。8.2 启用持久化# 使用默认路径 (~/.dora/coordinator.redb) dora coordinator --store redb # 使用自定义路径 dora coordinator --store redb:/path/to/coordinator.redb # 默认: 仅内存 (无持久化) dora coordinator --store memoryredb后端需要redb-backendCargo feature默认 CLI 构建已启用。8.3 持久化的内容存储跟踪三类记录记录键持久化字段DataflowRecordUUID (16 字节)uuid、name、descriptor (JSON)、status、daemon IDs、generation 计数器、created/updated 时间戳BuildRecordUUID (16 字节)build ID、status、errors、created/updated 时间戳DaemonInfoDaemonId (postcard)daemon ID、machine ID记录使用 postcard 序列化以获得紧凑、快速的编码。8.4 数据流状态生命周期协调器在每次状态转换时持久化数据流状态Start command -- Pending All daemons ready -- Running Stop command -- Stopping All nodes finish -- Succeeded 或 Failed { error } Spawn failure -- Failed { error: spawn failed: ... }每次持久化调用都会递增记录的generation计数器为冲突检测提供单调版本号。8.5 持久化点协调器在事件循环中的以下时刻写入存储数据流启动ControlRequest::Start——以Pending状态创建记录数据流 spawn 完成所有 daemon 返回DataflowSpawnResult成功——更新为Runningspawn 失败DataflowSpawnResult错误——更新为Failed并携带实际错误消息请求停止ControlRequest::Stop或StopByName——更新为Stopping所有节点完成DataflowFinishedOnDaemon——更新为Succeeded或Failed并携带每个节点的错误详情优雅关闭Ctrl-C 或Destroy命令——在发送 stop 消息之前把所有运行中的数据流标记为Stopping。如果存储写入失败协调器记录警告并继续使用内存状态运行。这防止存储故障阻塞数据流生命周期。8.6 启动恢复协调器带着包含上次运行数据的 redb 存储启动时执行恢复流程通过store.list_dataflows()读取所有持久化的数据流记录对任何非终态Pending、Running、Stopping的记录标记为Failed { error: coordinator restarted }递增 generation 计数器将更新后的记录写回存储终态记录Succeeded、Failed保持不变。这确保了崩溃协调器遗留的陈旧数据流不会被误认为正在运行。运行这些数据流的 daemon 会独立检测到协调器断连。8.7 错误详情保留数据流失败时Failed状态包含实际的逐节点错误消息而不是通用字符串Failed { error: node-1: exited with code 137; node-2: failed to spawn node: binary not found }错误从所有 daemon 的DataflowDaemonResult.node_results收集格式化为node_id: error_message用;连接。8.8 Schema 版本化redb 数据库包含一个带schema_version键的meta表。打开时若无版本新数据库写入当前版本若存储的版本与二进制版本匹配正常打开若不匹配数据库被拒绝并报错。这防止了 Dora 版本之间存储记录序列化格式变化时的静默数据损坏。当前 schema 版本为1。8.9 文件安全在 Unix 系统上数据库文件创建后被设置为0600仅属主读写默认目录~/.dora/被设置为0700仅属主通过redb:/path提供的自定义路径会被校验拒绝..组件。8.10 内部架构CoordinatorStoretrait 定义在 libraries/coordinator-store/src/lib.rspub trait CoordinatorStore: Send Sync { fn put_dataflow(self, record: DataflowRecord) - Result(); fn get_dataflow(self, uuid: Uuid) - ResultOptionDataflowRecord; fn list_dataflows(self) - ResultVecDataflowRecord; fn delete_dataflow(self, uuid: Uuid) - Result(); // ... daemon 与 build 相关方法 }RedbStore实现使用三个 redb 表daemons、dataflows、builds采用基于 UUID 的二进制键和 postcard 序列化的值。所有操作都是同步的redb 是同步库协调器直接从异步事件循环调用它们因为它们是快速进程内操作。此外编解码两侧对称地执行 64 MiB 记录大小限制因此永远不会写入一条读者之后会拒绝的记录。postcard 只从交给它的切片读取不会根据长度前缀预先分配因此损坏的行不会驱动大分配。9. 完整 YAML 参考# 数据流级设置 health_check_interval: 2.0 # 健康检查扫描间隔 (默认: 5.0s) nodes: - id: sensor-node path: ./target/debug/sensor inputs: tick: dora/timer/millis/100 outputs: - frames - id: processor path: ./target/debug/processor # 重启策略 restart_policy: on-failure # never | on-failure | always max_restarts: 5 # 0 无限制 restart_delay: 1.0 # 初始退避延迟 (秒) max_restart_delay: 30.0 # 最大退避上限 (秒) restart_window: 300.0 # 经过 N 秒后重置计数器 # 健康监控 health_check_timeout: 30.0 # 无活动 N 秒则杀死 inputs: frames: source: sensor-node/frames input_timeout: 5.0 # 熔断器超时 (秒) queue_size: 10 # 输入缓冲大小 (默认: 10) outputs: - result10. 实战场景Use Case Scenarios场景 1间歇性硬件故障的相机管线相机驱动节点偶尔因 USB 断连而崩溃。处理管线应能挺过这些中断并在相机重连后恢复。nodes: - id: camera-driver path: ./target/debug/camera-driver restart_policy: on-failure max_restarts: 0 # 无限制 -- 硬件故障是预期内的 restart_delay: 2.0 # 等待 USB 重新枚举 max_restart_delay: 30.0 inputs: tick: dora/timer/millis/33 # ~30 FPS outputs: - frames - id: object-detector path: ./target/debug/detector inputs: frames: source: camera-driver/frames input_timeout: 5.0 # 容忍 5s 相机中断 outputs: - detections - id: planner path: ./target/debug/planner inputs: detections: source: object-detector/detections input_timeout: 10.0 # 更长容忍 -- 可以用旧数据规划 lidar: source: lidar-driver/points input_timeout: 3.0相机崩溃时发生什么camera-driver以非零退出码退出daemon 评估on-failure策略 → 2s 退避后重启中断期间object-detector在 5s 后收到InputClosed { id: frames }planner在 10s 后收到InputClosed { id: detections }相机重启并开始生产帧object-detector收到新帧数据 InputRecovered { id: frames }熔断器恢复planner收到检测结果 InputRecovered { id: detections }。planner 中的节点侧处理use dora_node_api::{DoraNode, Event, InputTracker}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } match id.as_ref() { detections plan_with_detections(data), lidar update_lidar_map(data), _ {} }, Event::InputClosed { id } match id.as_ref() { detections { // 相机管线宕机 -- 仅用 lidar 规划 plan_lidar_only(); } lidar { // LiDAR 宕机 -- 使用最后已知的检测数据 if let Some(stale) tracker.last_value(detections.into()) { plan_with_stale_detections(stale); } } _ {} }, Event::Stop(_) break, _ {} } }场景 2OOM 崩溃的 ML 推理节点ML 推理节点偶尔在大输入上内存耗尽。它应该快速重启但在反复失败后放弃表明是系统性问题。nodes: - id: ml-inference path: ./target/debug/ml-inference restart_policy: on-failure max_restarts: 3 restart_delay: 0.5 restart_window: 60.0 # 每分钟 3 次重启 health_check_timeout: 60.0 # ML 推理可能很慢 inputs: images: source: preprocessor/images outputs: - predictions行为节点 OOM 崩溃 → 0.5s 后重启在另一个大输入上再次崩溃 → 1.0s 后重启第三次崩溃 → 2.0s 后重启60s 内第四次崩溃 → 超过max_restarts节点永久失败如果首次崩溃后节点稳定运行 60s重启窗口重置它获得 3 次新机会。场景 3多传感器融合的优雅降级机器人融合多个传感器的数据。单个传感器可能故障但系统应继续以降低的能力运行。nodes: - id: sensor-fusion path: ./target/debug/sensor-fusion inputs: camera: source: camera-node/frames input_timeout: 3.0 lidar: source: lidar-node/points input_timeout: 3.0 imu: source: imu-node/readings input_timeout: 1.0 # IMU 是关键, 短超时 gps: source: gps-node/fix input_timeout: 10.0 # GPS 可能间歇性故障 outputs: - fused-state配合 InputTracker 的节点侧use dora_node_api::{DoraNode, Event, InputTracker}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } { // 处理来自任意传感器的新数据 update_sensor(id, data); compute_and_send_fusion(mut node, tracker); } Event::InputClosed { id } { // 传感器离线 -- 调整融合权重 eprintln!(sensor {id} offline, degrading); compute_and_send_fusion(mut node, tracker); } Event::InputRecovered { id } { // 传感器恢复在线 eprintln!(sensor {id} recovered); } Event::Stop(_) break, _ {} } } fn compute_and_send_fusion(node: mut DoraNode, tracker: InputTracker) { // 可用处用新数据降级传感器用陈旧缓存 let camera tracker.last_value(camera.into()); let lidar tracker.last_value(lidar.into()); let imu tracker.last_value(imu.into()); if tracker.is_closed(imu.into()) { // IMU 是关键 -- 切换到紧急模式 emergency_stop(node); return; } // 融合可用传感器给活动者更高权重 let closed tracker.closed_inputs(); let active_count 4 - closed.len(); // ... 使用 active_count 做置信度加权的融合逻辑 }场景 4长期运行的数据处理管线批处理管线持续运行。处理节点偶尔因第三方库 bug 挂起。健康监控检测并恢复这些挂起。nodes: - id:># 用持久化存储启动协调器 dora coordinator --store redb # 在另一个终端启动一个数据流 dora start examples/rust-dataflow/dataflow.yml --name my-pipeline --detach # 协调器崩溃或被杀死 (例如 OOM, 硬件故障) # ... 时间流逝 ... # 用相同存储重启协调器 dora coordinator --store redb重启时发生什么协调器打开~/.dora/coordinator.redb并读取持久化的数据流记录找到状态为Running的my-pipeline将其标记为Failed { error: coordinator restarted }递增 generation记录日志INFO recovering stale dataflow uuid (my-pipeline) - marking as Faileddora list现在显示my-pipeline及其最终状态和时间戳daemon 独立检测到协调器断连并停止其节点用户可以启动新数据流——协调器完全可用。关键收益协调器在重启后保留完整的数据流生命周期事件历史。没有--store redb所有状态都会丢失操作员无法得知崩溃前在运行什么。场景 7always-restart 的周期性批处理任务一个处理批次后退出、然后应重启处理下一批的节点。nodes: - id: batch-processor path: ./target/debug/batch-proc restart_policy: always # 即使正常退出也重启 max_restarts: 0 # 无限制 restart_delay: 10.0 # 批次间等待 10s max_restart_delay: 10.0 # 无指数增长 inputs: trigger: dora/timer/millis/1 # 立即首次触发 outputs: - batch-result节点处理一个批次以退出码 0 退出等待 10s然后重启处理下一批。always策略确保成功时也重启。设置restart_delay max_restart_delay可得到恒定延迟。11. 最佳实践Best Practices从on-failure开始。always只用于预期会退出并重启的节点如周期性批处理任务。设置max_restarts。无限制重启可能掩盖 bug。从 3-5 开始需要时增加。仅对崩溃可预期且不可避免的节点硬件驱动、外部 API 客户端使用max_restarts: 0。使用restart_window。防止永久重启循环。60-300 秒的窗口是典型值。没有窗口一个启动即崩溃的节点会立即耗尽重启预算。调优restart_delay。从 0.5-1.0 秒开始。太短导致抖动太长延迟恢复。让延迟匹配节点典型启动时间和故障根因USB/硬件重连2-5s网络服务重连1-3sOOM/瞬时 bug0.5-1.0s慷慨地设置health_check_timeout。至少应为节点最长预期处理时间的 2-3 倍。ML 推理节点可能需要 60s。太短的话健康节点会在正常处理中被杀死。按输入设置input_timeout。并非所有输入都需要相同超时。高频输入IMU、相机用较短超时慢速/突发源GPS、批处理结果用较长超时。一个好的起点是预期发布间隔的 3-5 倍。关键路径使用InputTracker。当节点必须在输入降级时继续运行时用InputTracker回退到缓存数据。这对传感器融合、规划和控制节点至关重要。生产部署使用--store redb。redb 后端确保协调器在崩溃和重启后保留数据流历史。内存默认对开发足够但退出时丢失所有状态。redb 文件很小与数据流记录数成正比开销可忽略。组合特性实现防御纵深restart_policyrestart_delay→ 从节点崩溃中恢复health_check_timeout→ 从挂起节点中恢复input_timeout→ 检测陈旧上游数据InputTracker→ 节点代码中的优雅降级--store redb→ 挺过协调器崩溃。这五层能力可独立使用也天然互补节点层重启处理崩溃与挂起数据层熔断器处理上游失联应用层 InputTracker 处理降级协调器层持久化处理进程级故障——共同构成一套从单节点到多机集群的完整容错防线。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
DORA 数据流容错指南：重启策略、健康检查、输入熔断器与协调器状态持久化
DORA 数据流容错指南重启策略、健康检查、输入熔断器与协调器状态持久化【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/doraDORADataflow-Oriented Robotic Architecture为机器人及 AI 数据流内置了一套覆盖节点、输入、数据流与多机部署多层次的容错机制节点可按策略自动重启、可检测失联上游与挂起节点、输入不可用时优雅降级协调器coordinator还能把状态持久化到磁盘以承受自身崩溃与重启。本文以官方容错文档 docs/fault-tolerance.md 为主体结合 daemon、运行期 API 与 coordinator-store 的源码实现系统讲解 DORA 容错体系的配置参数、底层工作原理、Rust 节点侧 API 以及完整实战场景帮助你为生产级数据流构建「防御纵深」。1. 功能总览从节点重启到协调器状态持久化DORA 的容错能力分布在数据流生命周期中的不同层次每一层解决一类故障特性作用范围配置方式重启策略Restart policies每个节点restart_policy、max_restarts、restart_delay等健康监控Health monitoring每个节点health_check_timeout、startup_timeout、health_check_interval数据流级输入超时Input timeouts每个输入input_timeout熔断器Circuit breaker自动触发由input_timeout触发可自动恢复NodeRestarted 事件下游节点上游重启时自动触发InputTracker APIRust 节点dora_node_api::InputTracker可观测性Observability整个 daemon周期性输出的原子计数器分布式健康Distributed health多 daemon协调器心跳监控协调器状态持久化协调器--store redb需redb-backendfeature从实现层面看容错逻辑主要位于 daemon 的 spawn/prepared.rs重启生命周期循环、running_dataflow.rs健康检查、输入熔断、节点运行状态跟踪与 fault_tolerance.rs原子计数器节点侧 API 位于 apis/rust/node/src/event_stream/input_tracker.rs协调器存储抽象位于 libraries/coordinator-store/src/lib.rs。2. 重启策略Restart Policies2.1 配置nodes: - id: my-node path: ./target/debug/my-node restart_policy: on-failure # never | on-failure | always max_restarts: 5 # 0 unlimited (默认: 0) restart_delay: 1.0 # 初始延迟秒 max_restart_delay: 30.0 # 指数退避的上限秒 restart_window: 300.0 # 经过这么多秒后重置计数器2.2 策略类型never默认——节点不重启故障正常向上传播。on-failure——仅当节点以非零退出码退出时重启正常退出退出码 0不重启。always——任何退出都重启但有两个例外数据流被用户停止dora stop或 Ctrl-C所有输入都已关闭且节点以非零退出码退出。2.3 内部重启判定流程当节点进程退出时daemon 按以下顺序评估是否重启对应 spawn/prepared.rs 中restart_loop的实现策略检查重启策略是否允许Never→ 不重启OnFailure→ 仅在退出码 ! 0 时重启Always→ 重启。禁用检查disable_restart是否被置位在所有输入关闭或通过stop_all手动停止时置位窗口检查若设置了restart_window且自首次重启以来窗口已过期将计数器重置为 0。上限检查若max_restarts 0且窗口内计数超过它则永久放弃。退避若设置了restart_delay按计算出的延迟睡眠醒来后重新检查disable_restart。重新拉起使用相同配置重新生成节点进程。daemon 在spawn/prepared.rs的生命周期循环中按节点实例跟踪重启状态。每个节点运行在独立的 tokio task 中因此一个节点的重启不会阻塞其他节点。源码中还有一个值得注意的细节若force_restart_next标志被置位由dora node restart手动重启触发它会覆盖restart_policy即使默认never策略也会重新拉起——这保证了 CLI「重新拉起节点」的承诺。2.4 指数退避设置restart_delay后daemon 会在重启前等待。每次尝试延迟翻倍指数退避并由max_restart_delay封顶。退避指数在内部被限制为 16 以防溢出2^16 65536x倍乘数。以restart_delay: 1.0与max_restart_delay: 10.0为例Attempt 1: wait 1s (1.0 * 2^0) Attempt 2: wait 2s (1.0 * 2^1) Attempt 3: wait 4s (1.0 * 2^2) Attempt 4: wait 8s (1.0 * 2^3) Attempt 5: wait 10s (封顶于 max_restart_delay) Attempt 6: wait 10s (封顶)退避睡眠期间daemon 持续监控disable_restart标志。如果节点在等待重启期间所有输入都关闭了重启会被取消并记录日志restart cancelled: inputs closed during backoff wait。2.5 重启窗口Restart Window设置restart_window后重启计数器会在窗口从当前窗口内第一次重启算起到期后重置。这实现了「N 次重启 / M 秒」的语义。例如max_restarts: 5、restart_window: 300.0表示「每 5 分钟最多重启 5 次」。如果窗口到期而未触及上限计数器重置节点获得新的 5 次尝试机会。2.6 关闭期间禁用重启当 daemon 停止数据流通过stop_all时它会在发送 Stop 事件之前对每个节点调用disable_restart()。这防止重启机制与关闭流程互相打架。disable_restart是一个ArcAtomicBool在 daemon 事件循环与节点的 spawn 生命周期 task 之间共享。2.7 NodeRestarted 事件节点重启时daemon 会向所有消费其输出的下游节点发送NodeRestarted事件。下游节点可据此重置内部状态或缓存记录上游恢复日志重新初始化连接或会话。事件携带重启节点的NodeId。下游节点通过事件流自动接收match event { Event::NodeRestarted { id } { println!(upstream node {id} restarted, resetting state); // 清除来自旧节点实例的缓存状态 } _ {} }daemon 通过dataflow.mappings找到下游节点——该映射把每个节点的输出关联到所有订阅的(receiver_node, input_id)对。每次重启每个唯一的接收者收到一条NodeRestarted事件。实现提示InputTracker也支持在收到NodeRestarted时自动把来自被重启节点的Closed输入切回可恢复状态见第 5 节两种机制可以配合使用。3. 健康监控Health Monitoring被动监控用于检测与 daemon 停止通信的挂起节点。health_check_interval: 2.0 # 秒 (默认: 5.0, 数据流级) nodes: - id: my-node path: ./target/debug/my-node health_check_timeout: 30.0 # 秒 (每节点) restart_policy: on-failure3.1 可配置的健康检查间隔health_check_interval是一个数据流级设置控制 daemon 检查节点健康的频率默认 5.0 秒。更小的值能更快发现挂起节点但带来更多开销。它应设置在数据流 YAML 的顶层而不是每个节点。在 lib.rs 的 daemon 运行循环中健康检查周期直接取自描述符的health_check_interval并通过tokio::time::interval生成一个产生Event::NodeHealthCheckInterval的间隔流实现还防御性地对零周期做回退退回到默认 5 秒避免tokio::time::interval因零周期而 panic。3.2 内部工作原理daemon 在配置的health_check_interval上执行健康检查扫描。每个RunningNode有一个last_activity: ArcAtomicU64字段存储最近一次通信的时间戳epoch 毫秒。它由节点的通信处理器node_communication/mod.rs在节点每次向 daemon 发送任何请求时原子更新事件订阅、输出发送等。健康检查函数遍历所有运行中的节点跳过未设置health_check_timeout的节点跳过尚未连接的节点不在connected_nodes中计算elapsed_ms now - last_activity若elapsed_ms timeout_ms记录警告并杀死节点进程。杀死后走正常的退出处理流程该流程会评估重启策略。因此health_check_timeout与restart_policy: on-failure组合可以自动恢复挂起节点。杀死事件会累加到health_check_kills计数器。3.3 仅限连接后的活性Post-Connection Liveness Only上述第 2 步意味着health_check_timeout约束的是连接后的活性而不是总启动时间。节点在首次订阅事件时在Node::init/DoraNode::init_from_env内部加入connected_nodes超时时钟只从那一刻开始计算。这是有意设计的last_activity被初始化为 spawn 时间戳如果没有连接门控一个合法冷启动Python import、模型权重加载超过health_check_timeout的节点会在启动中途被 SIGKILL——在restart_policy: always/on-failure下这将成为无法逃脱的重启循环。其代价是在订阅之前就挂起的节点——例如 import 或 init 代码中的死锁永远走不到Node::init——不会被health_check_timeout回收。要约束启动时间并恢复初始化前的死锁请设置startup_timeout。3.4 启动截止时间startup_timeouthealth_check_timeout监控节点连接后的状态而startup_timeout约束的是从进程 spawn 到节点与 daemon 连接订阅事件之间的时间。如果一个未连接的节点进程在 spawn 后startup_timeout秒内未连接daemon 记录警告、SIGKILL 该进程、递增startup_timeout_kills计数器并评估restart_policy。nodes: - id: worker path: ./target/debug/worker startup_timeout: 10.0 # spawn 后 10s 内未订阅则杀死 health_check_timeout: 30.0 # 连接后 30s 静默则杀死 restart_policy: on-failure与health_check_timeout一样startup_timeout在每个health_check_interval滴答时评估。3.5 什么算「活动」任何从节点到 daemon 的消息都算事件订阅请求输出数据发送通过共享内存或 TCP定时器滴答确认。从其他节点收到的正常输入数据不会重置定时器——节点必须主动与 daemon 通信。4. 输入超时与熔断器Input Timeouts and Circuit Breaker按输入per-input的超时用于检测上游节点停止生产数据。4.1 配置nodes: - id: downstream-node path: ./target/debug/downstream inputs: sensor_data: source: camera-node/frames input_timeout: 5.0 # 秒input_timeout按输入设置而不是按节点。不同输入可以有不同的超时。警告——不适用于按需输入on-demand inputs。input_timeout假定上游持续发布数据。不要把它设置在仅在响应出站请求时才被填充的输入上服务响应输入、action 结果输入或任何其他突发式/按需通道。自然的空闲期看起来与死掉的上游完全一样会错误触发熔断器。对于按请求等待请改用EventStream::recv_service_response(request_id, server, timeout)或EventStream::recv_action_result(goal_id, server, timeout)——它们约束每个单独请求而不触发数据流级熔断器。更多相关模式参见 docs/patterns.md 第 6 节「Fault tolerance for correlated patterns」。4.2 内部工作原理daemon 为每个带超时的输入维护一个InputDeadline源码见 running_dataflow.rsstruct InputDeadline { timeout: Duration, // 配置的超时 last_received: OptionInstant, // 最近一次数据到达时间 (None 未武装) }这些条目存储在RunningDataflow.input_deadlines中以(NodeId, DataId)为键。截止时间武装Deadline arminglast_received在数据流启动时为None。熔断器的时钟在第一条消息真正到达之前不会针对该输入计时见InputDeadline::is_timed_out。这防止了启动时合法空闲的输入产生误报——对应 dora-rs/adora#149。running_dataflow.rs中的单元测试unarmed_deadline_is_never_timed_out与armed_deadline_past_timeout_is_timed_out明确验证了「未武装的截止时间永远不会超时」这一行为。超时检测在同一个 5 秒健康检查间隔内运行。check_input_timeouts函数扫描所有input_deadlines条目仅对已武装的条目last_received Some(_)若last_received.elapsed() timeout该输入视为「broken」(node_id, input_id)对从input_deadlines移到broken_inputsdaemon 调用break_input()向下游节点发送InputClosed { id }如果节点的所有输入现在都已关闭且没有 broken/recoverable 的发送AllInputsClosed并禁用该节点的重启。截止时间的武装/重置每次数据到达输入时其last_received被设置为Some(Instant::now())——既武装了先前未武装的截止时间也重置了已武装的截止时间。4.3 熔断器自动恢复熔断器在RunningDataflow.broken_inputs中跟踪 broken 输入。当新的数据到达一个 broken 输入时数据正常投递给节点移除broken_inputs条目输入重新加入open_inputs创建新的InputDeadline重新开始计时向节点发送InputRecovered { id }事件递增circuit_breaker_recoveries计数器。这意味着恢复是完全自动的。如果上游节点通过重启策略重启并开始再次生产数据下游节点无缝恢复接收。另外running_dataflow.rs的forget_node_bookkeeping会在动态增删节点时清理对应的input_deadlines/broken_inputs条目避免陈旧条目被反复扫描或永远无法恢复。4.4 节点侧处理在 Rust 节点中在事件循环里处理这些事件use dora_node_api::{DoraNode, Event}; let (mut node, mut events) DoraNode::init_from_env()?; while let Some(event) events.recv() { match event { Event::Input { id, data, .. } { // 正常处理 } Event::InputClosed { id } { // 上游停止在该输入上生产数据。 // 你可以使用缓存数据、跳过处理、提醒操作员等。 } Event::InputRecovered { id } { // 该输入的上游恢复在线。 // 恢复正常处理。 } Event::Stop(_) break, _ {} } }5. InputTracker APIRustInputTracker辅助类型跟踪输入健康状态并为每个输入缓存最近收到的值让优雅降级变得简单。use dora_node_api::{DoraNode, Event, InputTracker, InputState}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } { // 有新数据可用 } Event::InputClosed { id } { // 输入超时 -- 回退到缓存数据 if let Some(stale_data) tracker.last_value(id) { // 使用 stale_data 作为回退 } } Event::Stop(_) break, _ {} } // 检查整体健康状况 if tracker.any_closed() { let closed: Vec_ tracker.closed_inputs(); // 记录日志或调整行为 } }5.1 内部设计InputTracker维护两个HashMap见 apis/rust/node/src/event_stream/input_tracker.rsstates: HashMapDataId, InputState——每个输入的当前状态Healthy 或 Closedcache: HashMapDataId, DoraArray——每个输入最近收到的值。在Event::Input时两张 map 都更新state Healthycache data 克隆。在Event::InputClosed时仅 state 变化cache 保留。在Event::InputRecovered时state 恢复为 Healthy。cache 从不被清除因此last_value()即使在输入关闭后也始终返回最近的数据。另外该结构还支持可选的「输入 → 源节点」映射提供后NodeRestarted事件会把源自被重启节点的任何Closed输入转移恢复与第 2.7 节的 daemon 事件形成闭环。注意DoraArray包装了一个Arc支撑的 Arrow 数组因此 cache 克隆是引用计数的廉价。5.2 API 参考方法返回类型描述new()InputTracker创建空 trackerprocess_event(Event)bool更新状态。若事件相关则返回 truestate(DataId)OptionInputState当前状态Healthy 或 Closedis_closed(DataId)bool检查输入是否关闭last_value(DataId)OptionDoraArray最近收到的值关闭时也可用closed_inputs()VecDataId所有当前关闭的输入any_closed()bool若任意被跟踪输入关闭则返回 true6. 可观测性Observabilitydaemon 使用原子计数器FaultToleranceStats定义在 fault_tolerance.rs跟踪容错事件并在健康检查间隔内每 5 秒输出一次汇总日志。6.1 计数器计数器类型递增时机restartsAtomicU64发起节点重启时在 spawn 生命周期中health_check_killsAtomicU64节点被健康检查杀死无响应startup_timeout_killsAtomicU64节点被启动看门狗杀死init 前未连接input_timeoutsAtomicU64输入超时触发熔断器跳闸circuit_breaker_recoveriesAtomicU64数据到达 broken 输入自动恢复所有计数器使用Ordering::Relaxed因为它们是信息性的不需要严格的有序保证。FaultToleranceStats::any_nonzero()提供一个快捷判断供日志路径决定是否输出汇总。6.2 日志输出当任一计数器非零时daemon 输出一条结构化日志INFO fault tolerance stats restarts3 health_kills0 input_timeouts1 cb_recoveries1这些计数器对 daemon 进程的整个生命周期是累积的不会在数据流之间重置。7. 分布式健康Distributed Health在多 daemon 部署中协调器监控 daemon 心跳。7.1 协议心跳间隔3 秒协调器向每个 daemon 发送心跳断连阈值30 秒无响应检测每次心跳扫描时协调器移除超过阈值未响应的 daemon通知协调器向所有剩余 daemon 广播PeerDaemonDisconnected { daemon_id }。7.2 DaemonInfoConnectedMachinesCLI 查询返回VecDaemonInfopub struct DaemonInfo { pub daemon_id: DaemonId, pub last_heartbeat_ago_ms: u64, // 距上次心跳的毫秒数 }这让监控工具能够检测「活着但响应慢」的 daemon。7.3 Daemon 侧处理daemon 收到PeerDaemonDisconnected时记录一条结构化警告WARN peer daemon disconnected daemon_idmachine-B目前这只是信息性的。未来的工作可能包括自动迁移断连 daemon 上的节点。8. 协调器状态持久化Coordinator State Persistence默认情况下协调器把所有状态保存在内存中。若协调器进程崩溃或被重启所有正在运行的数据流信息都会丢失——daemon 继续运行但变成孤儿用户必须手动重新运行数据流。redb 存储后端解决了这个问题它使用 redb一个纯 Rust 嵌入式键值存储基于写时复制 B 树天生崩溃安全把协调器状态持久化到磁盘上的单个文件。8.1 设计无状态协调器 有状态后端协调器本身保持无状态K8s 意义上的——它可以在任何时候停止和重启。所有持久状态都位于CoordinatorStoretrait 后面的存储后端中Coordinator (无状态进程) | v CoordinatorStore trait | -- InMemoryStore (默认, 无持久化) -- RedbStore (持久化到 ~/.dora/coordinator.redb)这种分离意味着协调器事件循环在正常运行期间从不读取文件系统仅在启动恢复时所有状态变更在明确定义的持久化点写入存储存储可以在不改变协调器逻辑的情况下替换。8.2 启用持久化# 使用默认路径 (~/.dora/coordinator.redb) dora coordinator --store redb # 使用自定义路径 dora coordinator --store redb:/path/to/coordinator.redb # 默认: 仅内存 (无持久化) dora coordinator --store memoryredb后端需要redb-backendCargo feature默认 CLI 构建已启用。8.3 持久化的内容存储跟踪三类记录记录键持久化字段DataflowRecordUUID (16 字节)uuid、name、descriptor (JSON)、status、daemon IDs、generation 计数器、created/updated 时间戳BuildRecordUUID (16 字节)build ID、status、errors、created/updated 时间戳DaemonInfoDaemonId (postcard)daemon ID、machine ID记录使用 postcard 序列化以获得紧凑、快速的编码。8.4 数据流状态生命周期协调器在每次状态转换时持久化数据流状态Start command -- Pending All daemons ready -- Running Stop command -- Stopping All nodes finish -- Succeeded 或 Failed { error } Spawn failure -- Failed { error: spawn failed: ... }每次持久化调用都会递增记录的generation计数器为冲突检测提供单调版本号。8.5 持久化点协调器在事件循环中的以下时刻写入存储数据流启动ControlRequest::Start——以Pending状态创建记录数据流 spawn 完成所有 daemon 返回DataflowSpawnResult成功——更新为Runningspawn 失败DataflowSpawnResult错误——更新为Failed并携带实际错误消息请求停止ControlRequest::Stop或StopByName——更新为Stopping所有节点完成DataflowFinishedOnDaemon——更新为Succeeded或Failed并携带每个节点的错误详情优雅关闭Ctrl-C 或Destroy命令——在发送 stop 消息之前把所有运行中的数据流标记为Stopping。如果存储写入失败协调器记录警告并继续使用内存状态运行。这防止存储故障阻塞数据流生命周期。8.6 启动恢复协调器带着包含上次运行数据的 redb 存储启动时执行恢复流程通过store.list_dataflows()读取所有持久化的数据流记录对任何非终态Pending、Running、Stopping的记录标记为Failed { error: coordinator restarted }递增 generation 计数器将更新后的记录写回存储终态记录Succeeded、Failed保持不变。这确保了崩溃协调器遗留的陈旧数据流不会被误认为正在运行。运行这些数据流的 daemon 会独立检测到协调器断连。8.7 错误详情保留数据流失败时Failed状态包含实际的逐节点错误消息而不是通用字符串Failed { error: node-1: exited with code 137; node-2: failed to spawn node: binary not found }错误从所有 daemon 的DataflowDaemonResult.node_results收集格式化为node_id: error_message用;连接。8.8 Schema 版本化redb 数据库包含一个带schema_version键的meta表。打开时若无版本新数据库写入当前版本若存储的版本与二进制版本匹配正常打开若不匹配数据库被拒绝并报错。这防止了 Dora 版本之间存储记录序列化格式变化时的静默数据损坏。当前 schema 版本为1。8.9 文件安全在 Unix 系统上数据库文件创建后被设置为0600仅属主读写默认目录~/.dora/被设置为0700仅属主通过redb:/path提供的自定义路径会被校验拒绝..组件。8.10 内部架构CoordinatorStoretrait 定义在 libraries/coordinator-store/src/lib.rspub trait CoordinatorStore: Send Sync { fn put_dataflow(self, record: DataflowRecord) - Result(); fn get_dataflow(self, uuid: Uuid) - ResultOptionDataflowRecord; fn list_dataflows(self) - ResultVecDataflowRecord; fn delete_dataflow(self, uuid: Uuid) - Result(); // ... daemon 与 build 相关方法 }RedbStore实现使用三个 redb 表daemons、dataflows、builds采用基于 UUID 的二进制键和 postcard 序列化的值。所有操作都是同步的redb 是同步库协调器直接从异步事件循环调用它们因为它们是快速进程内操作。此外编解码两侧对称地执行 64 MiB 记录大小限制因此永远不会写入一条读者之后会拒绝的记录。postcard 只从交给它的切片读取不会根据长度前缀预先分配因此损坏的行不会驱动大分配。9. 完整 YAML 参考# 数据流级设置 health_check_interval: 2.0 # 健康检查扫描间隔 (默认: 5.0s) nodes: - id: sensor-node path: ./target/debug/sensor inputs: tick: dora/timer/millis/100 outputs: - frames - id: processor path: ./target/debug/processor # 重启策略 restart_policy: on-failure # never | on-failure | always max_restarts: 5 # 0 无限制 restart_delay: 1.0 # 初始退避延迟 (秒) max_restart_delay: 30.0 # 最大退避上限 (秒) restart_window: 300.0 # 经过 N 秒后重置计数器 # 健康监控 health_check_timeout: 30.0 # 无活动 N 秒则杀死 inputs: frames: source: sensor-node/frames input_timeout: 5.0 # 熔断器超时 (秒) queue_size: 10 # 输入缓冲大小 (默认: 10) outputs: - result10. 实战场景Use Case Scenarios场景 1间歇性硬件故障的相机管线相机驱动节点偶尔因 USB 断连而崩溃。处理管线应能挺过这些中断并在相机重连后恢复。nodes: - id: camera-driver path: ./target/debug/camera-driver restart_policy: on-failure max_restarts: 0 # 无限制 -- 硬件故障是预期内的 restart_delay: 2.0 # 等待 USB 重新枚举 max_restart_delay: 30.0 inputs: tick: dora/timer/millis/33 # ~30 FPS outputs: - frames - id: object-detector path: ./target/debug/detector inputs: frames: source: camera-driver/frames input_timeout: 5.0 # 容忍 5s 相机中断 outputs: - detections - id: planner path: ./target/debug/planner inputs: detections: source: object-detector/detections input_timeout: 10.0 # 更长容忍 -- 可以用旧数据规划 lidar: source: lidar-driver/points input_timeout: 3.0相机崩溃时发生什么camera-driver以非零退出码退出daemon 评估on-failure策略 → 2s 退避后重启中断期间object-detector在 5s 后收到InputClosed { id: frames }planner在 10s 后收到InputClosed { id: detections }相机重启并开始生产帧object-detector收到新帧数据 InputRecovered { id: frames }熔断器恢复planner收到检测结果 InputRecovered { id: detections }。planner 中的节点侧处理use dora_node_api::{DoraNode, Event, InputTracker}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } match id.as_ref() { detections plan_with_detections(data), lidar update_lidar_map(data), _ {} }, Event::InputClosed { id } match id.as_ref() { detections { // 相机管线宕机 -- 仅用 lidar 规划 plan_lidar_only(); } lidar { // LiDAR 宕机 -- 使用最后已知的检测数据 if let Some(stale) tracker.last_value(detections.into()) { plan_with_stale_detections(stale); } } _ {} }, Event::Stop(_) break, _ {} } }场景 2OOM 崩溃的 ML 推理节点ML 推理节点偶尔在大输入上内存耗尽。它应该快速重启但在反复失败后放弃表明是系统性问题。nodes: - id: ml-inference path: ./target/debug/ml-inference restart_policy: on-failure max_restarts: 3 restart_delay: 0.5 restart_window: 60.0 # 每分钟 3 次重启 health_check_timeout: 60.0 # ML 推理可能很慢 inputs: images: source: preprocessor/images outputs: - predictions行为节点 OOM 崩溃 → 0.5s 后重启在另一个大输入上再次崩溃 → 1.0s 后重启第三次崩溃 → 2.0s 后重启60s 内第四次崩溃 → 超过max_restarts节点永久失败如果首次崩溃后节点稳定运行 60s重启窗口重置它获得 3 次新机会。场景 3多传感器融合的优雅降级机器人融合多个传感器的数据。单个传感器可能故障但系统应继续以降低的能力运行。nodes: - id: sensor-fusion path: ./target/debug/sensor-fusion inputs: camera: source: camera-node/frames input_timeout: 3.0 lidar: source: lidar-node/points input_timeout: 3.0 imu: source: imu-node/readings input_timeout: 1.0 # IMU 是关键, 短超时 gps: source: gps-node/fix input_timeout: 10.0 # GPS 可能间歇性故障 outputs: - fused-state配合 InputTracker 的节点侧use dora_node_api::{DoraNode, Event, InputTracker}; let (mut node, mut events) DoraNode::init_from_env()?; let mut tracker InputTracker::new(); while let Some(event) events.recv() { tracker.process_event(event); match event { Event::Input { id, data, .. } { // 处理来自任意传感器的新数据 update_sensor(id, data); compute_and_send_fusion(mut node, tracker); } Event::InputClosed { id } { // 传感器离线 -- 调整融合权重 eprintln!(sensor {id} offline, degrading); compute_and_send_fusion(mut node, tracker); } Event::InputRecovered { id } { // 传感器恢复在线 eprintln!(sensor {id} recovered); } Event::Stop(_) break, _ {} } } fn compute_and_send_fusion(node: mut DoraNode, tracker: InputTracker) { // 可用处用新数据降级传感器用陈旧缓存 let camera tracker.last_value(camera.into()); let lidar tracker.last_value(lidar.into()); let imu tracker.last_value(imu.into()); if tracker.is_closed(imu.into()) { // IMU 是关键 -- 切换到紧急模式 emergency_stop(node); return; } // 融合可用传感器给活动者更高权重 let closed tracker.closed_inputs(); let active_count 4 - closed.len(); // ... 使用 active_count 做置信度加权的融合逻辑 }场景 4长期运行的数据处理管线批处理管线持续运行。处理节点偶尔因第三方库 bug 挂起。健康监控检测并恢复这些挂起。nodes: - id:># 用持久化存储启动协调器 dora coordinator --store redb # 在另一个终端启动一个数据流 dora start examples/rust-dataflow/dataflow.yml --name my-pipeline --detach # 协调器崩溃或被杀死 (例如 OOM, 硬件故障) # ... 时间流逝 ... # 用相同存储重启协调器 dora coordinator --store redb重启时发生什么协调器打开~/.dora/coordinator.redb并读取持久化的数据流记录找到状态为Running的my-pipeline将其标记为Failed { error: coordinator restarted }递增 generation记录日志INFO recovering stale dataflow uuid (my-pipeline) - marking as Faileddora list现在显示my-pipeline及其最终状态和时间戳daemon 独立检测到协调器断连并停止其节点用户可以启动新数据流——协调器完全可用。关键收益协调器在重启后保留完整的数据流生命周期事件历史。没有--store redb所有状态都会丢失操作员无法得知崩溃前在运行什么。场景 7always-restart 的周期性批处理任务一个处理批次后退出、然后应重启处理下一批的节点。nodes: - id: batch-processor path: ./target/debug/batch-proc restart_policy: always # 即使正常退出也重启 max_restarts: 0 # 无限制 restart_delay: 10.0 # 批次间等待 10s max_restart_delay: 10.0 # 无指数增长 inputs: trigger: dora/timer/millis/1 # 立即首次触发 outputs: - batch-result节点处理一个批次以退出码 0 退出等待 10s然后重启处理下一批。always策略确保成功时也重启。设置restart_delay max_restart_delay可得到恒定延迟。11. 最佳实践Best Practices从on-failure开始。always只用于预期会退出并重启的节点如周期性批处理任务。设置max_restarts。无限制重启可能掩盖 bug。从 3-5 开始需要时增加。仅对崩溃可预期且不可避免的节点硬件驱动、外部 API 客户端使用max_restarts: 0。使用restart_window。防止永久重启循环。60-300 秒的窗口是典型值。没有窗口一个启动即崩溃的节点会立即耗尽重启预算。调优restart_delay。从 0.5-1.0 秒开始。太短导致抖动太长延迟恢复。让延迟匹配节点典型启动时间和故障根因USB/硬件重连2-5s网络服务重连1-3sOOM/瞬时 bug0.5-1.0s慷慨地设置health_check_timeout。至少应为节点最长预期处理时间的 2-3 倍。ML 推理节点可能需要 60s。太短的话健康节点会在正常处理中被杀死。按输入设置input_timeout。并非所有输入都需要相同超时。高频输入IMU、相机用较短超时慢速/突发源GPS、批处理结果用较长超时。一个好的起点是预期发布间隔的 3-5 倍。关键路径使用InputTracker。当节点必须在输入降级时继续运行时用InputTracker回退到缓存数据。这对传感器融合、规划和控制节点至关重要。生产部署使用--store redb。redb 后端确保协调器在崩溃和重启后保留数据流历史。内存默认对开发足够但退出时丢失所有状态。redb 文件很小与数据流记录数成正比开销可忽略。组合特性实现防御纵深restart_policyrestart_delay→ 从节点崩溃中恢复health_check_timeout→ 从挂起节点中恢复input_timeout→ 检测陈旧上游数据InputTracker→ 节点代码中的优雅降级--store redb→ 挺过协调器崩溃。这五层能力可独立使用也天然互补节点层重启处理崩溃与挂起数据层熔断器处理上游失联应用层 InputTracker 处理降级协调器层持久化处理进程级故障——共同构成一套从单节点到多机集群的完整容错防线。【免费下载链接】doraDORA (Dataflow-Oriented Robotic Architecture) is middleware designed to streamline and simplify the creation of AI-based robotic applications. It offers low latency, composable, and distributed dataflow capabilities. Applications are modeled as directed graphs, also referred to as pipelines.项目地址: https://gitcode.com/GitHub_Trending/do/dora创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考