企业官网建设流程全解析

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/dora

DORA（Dataflow-Oriented Robotic Architecture）为机器人及 AI 数据流内置了一套覆盖节点、输入、数据流与多机部署多层次的容错机制：节点可按策略自动重启、可检测失联上游与挂起节点、输入不可用时优雅降级，协调器（coordinator）还能把状态持久化到磁盘以承受自身崩溃与重启。本文以官方容错文档 docs/fault-tolerance.md 为主体，结合 daemon、运行期 API 与 coordinator-store 的源码实现，系统讲解 DORA 容错体系的配置参数、底层工作原理、Rust 节点侧 API 以及完整实战场景，帮助你为生产级数据流构建「防御纵深」。

1. 功能总览：从节点重启到协调器状态持久化

DORA 的容错能力分布在数据流生命周期中的不同层次，每一层解决一类故障：

从实现层面看，容错逻辑主要位于 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 Policies）

2.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的实现）：

1. 策略检查：重启策略是否允许？
   • Never→ 不重启；
   • OnFailure→ 仅在退出码 != 0 时重启；
   • Always→ 重启。
2. 禁用检查：disable_restart是否被置位？（在所有输入关闭或通过stop_all手动停止时置位）
3. 窗口检查：若设置了restart_window且自首次重启以来窗口已过期，将计数器重置为 0。
4. 上限检查：若max_restarts > 0且窗口内计数超过它，则永久放弃。
5. 退避：若设置了restart_delay，按计算出的延迟睡眠（醒来后重新检查disable_restart）。
6. 重新拉起：使用相同配置重新生成节点进程。

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是一个Arc<AtomicBool>，在 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-failure

3.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: Arc<AtomicU64>字段，存储最近一次通信的时间戳（epoch 毫秒）。它由节点的通信处理器（node_communication/mod.rs）在节点每次向 daemon 发送任何请求时原子更新（事件订阅、输出发送等）。

健康检查函数遍历所有运行中的节点：

1. 跳过未设置health_check_timeout的节点；
2. 跳过尚未连接的节点（不在connected_nodes中）；
3. 计算elapsed_ms = now - last_activity；
4. 若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_timeout）

health_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.rs）：

struct InputDeadline { timeout: Duration, // 配置的超时 last_received: Option<Instant>, // 最近一次数据到达时间 (None = 未武装) }

这些条目存储在RunningDataflow.input_deadlines中，以(NodeId, DataId)为键。

截止时间武装（Deadline arming）：last_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函数：

1. 扫描所有input_deadlines条目；
2. 仅对已武装的条目（last_received = Some(_)），若last_received.elapsed() > timeout，该输入视为「broken」；
3. (node_id, input_id)对从input_deadlines移到broken_inputs；
4. daemon 调用break_input()，向下游节点发送InputClosed { id }；
5. 如果节点的所有输入现在都已关闭（且没有 broken/recoverable 的），发送AllInputsClosed并禁用该节点的重启。

截止时间的武装/重置：每次数据到达输入时，其last_received被设置为Some(Instant::now())——既武装了先前未武装的截止时间，也重置了已武装的截止时间。

4.3 熔断器：自动恢复

熔断器在RunningDataflow.broken_inputs中跟踪 broken 输入。当新的数据到达一个 broken 输入时：

1. 数据正常投递给节点；
2. 移除broken_inputs条目；
3. 输入重新加入open_inputs；
4. 创建新的InputDeadline（重新开始计时）；
5. 向节点发送InputRecovered { id }事件；
6. 递增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 API（Rust）

InputTracker辅助类型跟踪输入健康状态并为每个输入缓存最近收到的值，让优雅降级变得简单。

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.rs）：

• states: HashMap<DataId, InputState>——每个输入的当前状态（Healthy 或 Closed）；
• cache: HashMap<DataId, DoraArray>——每个输入最近收到的值。

在Event::Input时，两张 map 都更新（state = Healthy，cache = data 克隆）。在Event::InputClosed时，仅 state 变化（cache 保留）。在Event::InputRecovered时，state 恢复为 Healthy。cache 从不被清除，因此last_value()即使在输入关闭后也始终返回最近的数据。

另外，该结构还支持可选的「输入 → 源节点」映射：提供后，NodeRestarted事件会把源自被重启节点的任何Closed输入转移恢复，与第 2.7 节的 daemon 事件形成闭环。

注意：DoraArray包装了一个Arc支撑的 Arrow 数组，因此 cache 克隆是引用计数的（廉价）。

5.2 API 参考

6. 可观测性（Observability）

daemon 使用原子计数器（FaultToleranceStats，定义在 fault_tolerance.rs）跟踪容错事件，并在健康检查间隔内每 5 秒输出一次汇总日志。

6.1 计数器

所有计数器使用Ordering::Relaxed，因为它们是信息性的，不需要严格的有序保证。FaultToleranceStats::any_nonzero()提供一个快捷判断，供日志路径决定是否输出汇总。

6.2 日志输出

当任一计数器非零时，daemon 输出一条结构化日志：

INFO fault tolerance stats restarts=3 health_kills=0 input_timeouts=1 cb_recoveries=1

这些计数器对 daemon 进程的整个生命周期是累积的，不会在数据流之间重置。

7. 分布式健康（Distributed Health）

在多 daemon 部署中，协调器监控 daemon 心跳。

7.1 协议

• 心跳间隔：3 秒（协调器向每个 daemon 发送心跳）；
• 断连阈值：30 秒无响应；
• 检测：每次心跳扫描时，协调器移除超过阈值未响应的 daemon；
• 通知：协调器向所有剩余 daemon 广播PeerDaemonDisconnected { daemon_id }。

7.2 DaemonInfo

ConnectedMachinesCLI 查询返回Vec<DaemonInfo>：

pub struct DaemonInfo { pub daemon_id: DaemonId, pub last_heartbeat_ago_ms: u64, // 距上次心跳的毫秒数 }

这让监控工具能够检测「活着但响应慢」的 daemon。

7.3 Daemon 侧处理

daemon 收到PeerDaemonDisconnected时记录一条结构化警告：

WARN peer daemon disconnected daemon_id=machine-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 memory

redb后端需要redb-backendCargo feature，默认 CLI 构建已启用。

8.3 持久化的内容

存储跟踪三类记录：

记录使用 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 持久化点

协调器在事件循环中的以下时刻写入存储：

1. 数据流启动（ControlRequest::Start）——以Pending状态创建记录；
2. 数据流 spawn 完成（所有 daemon 返回DataflowSpawnResult成功）——更新为Running；
3. spawn 失败（DataflowSpawnResult错误）——更新为Failed并携带实际错误消息；
4. 请求停止（ControlRequest::Stop或StopByName）——更新为Stopping；
5. 所有节点完成（DataflowFinishedOnDaemon）——更新为Succeeded或Failed并携带每个节点的错误详情；
6. 优雅关闭（Ctrl-C 或Destroy命令）——在发送 stop 消息之前，把所有运行中的数据流标记为Stopping。

如果存储写入失败，协调器记录警告并继续使用内存状态运行。这防止存储故障阻塞数据流生命周期。

8.6 启动恢复

协调器带着包含上次运行数据的 redb 存储启动时，执行恢复流程：

1. 通过store.list_dataflows()读取所有持久化的数据流记录；
2. 对任何非终态（Pending、Running、Stopping）的记录：
   • 标记为Failed { error: "coordinator restarted" }；
   • 递增 generation 计数器；
   • 将更新后的记录写回存储；
3. 终态记录（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.rs：

pub trait CoordinatorStore: Send + Sync { fn put_dataflow(&self, record: &DataflowRecord) -> Result<()>; fn get_dataflow(&self, uuid: &Uuid) -> Result<Option<DataflowRecord>>; fn list_dataflows(&self) -> Result<Vec<DataflowRecord>>; 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: - result

10. 实战场景（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

相机崩溃时发生什么：

1. camera-driver以非零退出码退出；
2. daemon 评估on-failure策略 → 2s 退避后重启；
3. 中断期间，object-detector在 5s 后收到InputClosed { id: "frames" }；
4. planner在 10s 后收到InputClosed { id: "detections" }；
5. 相机重启并开始生产帧；
6. object-detector收到新帧数据 +InputRecovered { id: "frames" }（熔断器恢复）；
7. 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, _ => {} } }

场景 2：OOM 崩溃的 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

重启时发生什么：

1. 协调器打开~/.dora/coordinator.redb并读取持久化的数据流记录；
2. 找到状态为Running的my-pipeline；
3. 将其标记为Failed { error: "coordinator restarted" }，递增 generation；
4. 记录日志：INFO recovering stale dataflow <uuid> ("my-pipeline") -> marking as Failed；
5. dora list现在显示my-pipeline及其最终状态和时间戳；
6. daemon 独立检测到协调器断连并停止其节点；
7. 用户可以启动新数据流——协调器完全可用。

关键收益：协调器在重启后保留完整的数据流生命周期事件历史。没有--store redb，所有状态都会丢失，操作员无法得知崩溃前在运行什么。

场景 7：always-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-3s
• OOM/瞬时 bug：0.5-1.0s

慷慨地设置health_check_timeout。至少应为节点最长预期处理时间的 2-3 倍。ML 推理节点可能需要 60s+。太短的话，健康节点会在正常处理中被杀死。

按输入设置input_timeout。并非所有输入都需要相同超时。高频输入（IMU、相机）用较短超时，慢速/突发源（GPS、批处理结果）用较长超时。一个好的起点是预期发布间隔的 3-5 倍。

关键路径使用InputTracker。当节点必须在输入降级时继续运行时，用InputTracker回退到缓存数据。这对传感器融合、规划和控制节点至关重要。

生产部署使用--store redb。redb 后端确保协调器在崩溃和重启后保留数据流历史。内存默认对开发足够，但退出时丢失所有状态。redb 文件很小（与数据流记录数成正比），开销可忽略。

组合特性实现防御纵深：

• restart_policy+restart_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