Adora

智能数据流导向机器人架构 (Agentic Dataflow-Oriented Robotic Architecture) – 一个 100% Rust 的实时机器人与 AI 应用框架。

为什么选择 Adora？

Performance

• 10-17x faster than ROS2 Python – 100% Rust internals with zero-copy shared memory IPC for messages >4KB, flat latency from 4KB to 4MB payloads
• Apache Arrow native – columnar memory format end-to-end with zero serialization overhead; shared across all language bindings

Developer Experience

• Single CLI, full lifecycle – adora run for local dev, adora up/start for distributed prod, plus build, logs, monitoring, record/replay all from one tool
• Declarative YAML dataflows – define pipelines as directed graphs, connect nodes through typed inputs/outputs, optional type annotations with static validation
• Multi-language nodes – write nodes in Rust, Python, C, or C++ with native APIs (not wrappers); mix languages freely in one dataflow
• Reusable modules – compose sub-graphs as standalone YAML files with typed inputs/outputs, parameters, and nested composition
• Hot reload – live-reload Python operators without restarting the dataflow

Production Readiness

• Fault tolerance – per-node restart policies (never/on-failure/always), exponential backoff, health monitoring, circuit breakers with configurable input timeouts
• Distributed by default – local shared memory between co-located nodes, automatic Zenoh pub-sub for cross-machine communication, SSH-based cluster management with label scheduling
• Configurable queue policies – drop_oldest (default) or backpressure per input, with metrics on dropped messages
• OpenTelemetry – built-in structured logging with rotation/routing, metrics, distributed tracing

Debugging and Observability

• Record/replay – capture dataflow messages to .adorec files, replay offline at any speed with node substitution
• Topic inspection – topic echo to print live data, topic hz TUI for frequency analysis, topic info for schema and bandwidth
• Resource monitoring – adora top TUI showing per-node CPU, memory, queue depth, network I/O across all machines
• Log aggregation – subscribe to adora/logs to receive structured log messages from all nodes without extra wiring
• Trace inspection – trace list and trace view for viewing coordinator spans without external infrastructure

Ecosystem

• Communication patterns – built-in service (request/reply), action (goal/feedback/result), and streaming (session/segment/chunk) patterns via well-known metadata keys
• ROS2 bridge – bidirectional interop with ROS2 topics, services, and actions
• In-process operators – lightweight functions that run inside a shared runtime, avoiding per-node process overhead

下一步

• 安装 Adora
• 快速入门教程
• 架构概览
• Dataflow YAML reference
• Type annotations
• Communication patterns

安装

From crates.io (recommended)

cargo install adora-cli           # CLI（adora 命令）
pip install adora-rs              # Python 节点/算子 API

从源码安装

git clone https://github.com/dora-rs/adora.git
cd adora
cargo build --release -p adora-cli
PATH=$PATH:$(pwd)/target/release

# Python API (requires maturin >= 1.8: pip install maturin)
# Must run from the package directory for dependency resolution
cd apis/python/node && maturin develop --uv && cd ../../..

平台安装器

macOS / Linux：

curl --proto '=https' --tlsv1.2 -LsSf \
  https://github.com/dora-rs/adora/releases/latest/download/adora-cli-installer.sh | sh

Windows：

powershell -ExecutionPolicy ByPass -c "irm https://github.com/dora-rs/adora/releases/latest/download/adora-cli-installer.ps1 | iex"

构建特性

cargo install adora-cli --features redb-backend

验证

adora --version
adora status

Python 快速入门

本指南将引导你使用 Python 编写 adora 数据流的节点和算子。

前提条件

cargo install adora-cli    # CLI（adora 命令）
pip install adora-rs       # Python 节点/算子 API

adora-rs 包已包含 pyarrow 作为依赖。

从源码构建（替代 pip install adora-rs）：

pip install maturin  # requires >= 1.8
cd apis/python/node && maturin develop --uv && cd ../../..

Hello World：发送者与接收者

创建三个文件：

sender.py – 发送 100 条带编号的消息：

import pyarrow as pa
from adora import Node

node = Node()
for i in range(100):
    node.send_output("message", pa.array([i]))

receiver.py – 接收并打印消息：

from adora import Node

node = Node()
for event in node:
    if event["type"] == "INPUT":
        values = event["value"].to_pylist()
        print(f"Received {event['id']}: {values}")
    elif event["type"] == "STOP":
        break

dataflow.yml – 将发送者连接到接收者：

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

运行：

adora run dataflow.yml

Events

每次调用 node.next() 或通过 for event in node 迭代都会返回一个事件字典：

通过检查 event["type"] 来处理事件：

for event in node:
    match event["type"]:
        case "INPUT":
            process(event["id"], event["value"])
        case "INPUT_CLOSED":
            print(f"Input {event['id']} closed")
        case "STOP":
            break

使用 Arrow 数据

所有数据在 adora 中以 Apache Arrow 数组的形式流动。常见模式：

import pyarrow as pa

# Simple values
node.send_output("count", pa.array([42]))
node.send_output("names", pa.array(["alice", "bob"]))

# Read values back
values = event["value"].to_pylist()  # [42] or ["alice", "bob"]

# Structured data
struct = pa.StructArray.from_arrays(
    [pa.array([1.5]), pa.array(["hello"])],
    names=["x", "y"],
)
node.send_output("point", struct)

# Raw bytes (images, serialized data, etc.)
node.send_output("frame", pa.array(raw_bytes))

Operators

算子是节点的轻量级替代方案。它们在 adora 运行时进程内运行（无需单独的操作系统进程），因此对于简单的转换操作更加高效。

定义一个包含 on_event 方法的 Operator 类：

# doubler_op.py
import pyarrow as pa
from adora import AdoraStatus

class Operator:
    def on_event(self, event, send_output) -> AdoraStatus:
        if event["type"] == "INPUT":
            value = event["value"].to_pylist()[0]
            send_output("doubled", pa.array([value * 2]), event["metadata"])
        return AdoraStatus.CONTINUE

在 YAML 中使用 operator 而非 path 来引用它：

nodes:
  - id: timer
    path: adora/timer/millis/500
    outputs:
      - tick

  - id: doubler
    operator:
      python: doubler_op.py
      inputs:
        tick: timer/tick
      outputs:
        - doubled

何时使用算子 vs 节点：

异步节点

对于需要异步 I/O（HTTP 调用、数据库查询等）的节点，请使用 recv_async()：

import asyncio
from adora import Node

async def main():
    node = Node()
    for _ in range(50):
        event = await node.recv_async()
        if event["type"] == "STOP":
            break
        # Do async work here
        result = await fetch_data(event["value"])
        node.send_output("result", result)

asyncio.run(main())

参见 examples/python-async 获取完整示例。

日志

使用 node.log() 进行结构化日志记录，可与 adora logs 集成：

node.log("info", "Processing item", {"count": str(i)})

或者使用 Python 标准的 logging 模块 – adora 会自动捕获 stdout/stderr：

import logging
logging.info("Processing item %d", i)

参见 examples/python-logging 了解日志模块集成。

Timers

内置的定时器节点无需编写任何代码即可生成周期性的定时信号：

nodes:
  - id: tick-source
    path: adora/timer/millis/100    # tick every 100ms
    outputs:
      - tick

  - id: my-node
    path: my_node.py
    inputs:
      tick: tick-source/tick

也可使用：adora/timer/hz/30 表示 30 Hz。

下一步

• Python API 参考 – Node、Operator、DataflowBuilder、CUDA 的完整 API 文档
• 通信模式 – 服务（请求/应答）和动作（目标/反馈/结果）模式
• 示例 – python-dataflow、python-async、python-drain、python-concurrent-rw、python-multiple-arrays
• 分布式部署 – 使用 adora up 跨多台机器运行

Adora Architecture

Comprehensive architecture reference for Adora (AI-Dora, Agentic Dataflow-Oriented Robotic Architecture) — a 100% Rust framework for real-time robotics and AI applications.

Overview and Design Philosophy

Adora is built on four core principles:

1. Dataflow-oriented: Applications are directed graphs of nodes connected by typed data channels. Nodes declare inputs and outputs; the framework handles routing, scheduling, and lifecycle.
2. Zero-copy performance: Messages above 4 KiB use shared memory with 128-byte aligned buffers and atomic coordination, achieving 10-17x lower latency than ROS2.
3. Multi-language: First-class support for Rust, Python (PyO3), C, and C++ nodes — all sharing the same Apache Arrow data format.
4. Four-layer stack: Message protocol, core libraries, daemon/runtime execution, and CLI/coordinator orchestration.

Architecture Stack

┌─────────────────────────────────────────────────┐
│  CLI (adora)          Coordinator (orchestrator) │  Layer 4: Orchestration
├─────────────────────────────────────────────────┤
│  Daemon (per-machine)    Runtime (operators)     │  Layer 3: Execution
├─────────────────────────────────────────────────┤
│  adora-core    shared-memory-server    Node API  │  Layer 2: Core Libraries
├─────────────────────────────────────────────────┤
│  adora-message (protocol + Arrow types)          │  Layer 1: Protocol
└─────────────────────────────────────────────────┘

Workspace Structure

Rust edition 2024, MSRV 1.85.0, workspace version 0.1.0. All crates share the workspace version.

Binaries (7)

Core Libraries (6)

Extension Libraries (5)

API Crates (9)

Component Architecture

CLI

The adora command provides three command groups:

Lifecycle (run, up, down, build, start, stop, restart):

• adora run executes a dataflow locally without coordinator/daemon (single-machine shortcut)
• adora up / adora down manage coordinator + daemon infrastructure
• adora start / adora stop control dataflows on a running coordinator

Monitoring (list, logs, inspect, topic, node, record, replay, trace):

• Real-time inspection with adora inspect top
• Topic subscription and data inspection
• Recording and replay via .adorec files

Setup (status, new, graph, system, completion, self):

• Project scaffolding, dataflow visualization, self-update

Coordinator

The coordinator is an Axum-based WebSocket server that orchestrates distributed deployments.

                          ┌──────────────────┐
                          │   Coordinator     │
            WS /api/control  │  ┌────────────┐  │  WS /api/daemon
   CLI ◄──────────────────►  │  │   State    │  │ ◄──────────────────► Daemon(s)
                          │  │   Store    │  │
                          │  └────────────┘  │
                          │  /api/artifacts  │
                          │  /health         │
                          └──────────────────┘

WebSocket routes:

• /api/control — CLI control plane (build, start, stop, list, logs, topic subscribe)
• /api/daemon — Daemon registration and event stream
• /api/artifacts/{build_id}/{node_id} — Binary artifact downloads
• /health — Health check endpoint

State management: In-memory by default, optional persistent storage via redb backend.

Daemon

The daemon runs one per machine and manages the lifecycle of all nodes on that machine.

┌──────────────────────────────────────────────────────┐
│                     Daemon                           │
│                                                      │
│  ┌──────────┐  ┌───────────┐  ┌──────────────────┐  │
│  │ Event    │  │ Spawner   │  │ Node Comm        │  │
│  │ Loop     │──│ (nodes)   │  │ ┌──────────────┐ │  │
│  │          │  └───────────┘  │ │ TCP listener │ │  │
│  │ Sources: │  ┌───────────┐  │ │ Shmem server │ │  │
│  │ • Coord  │  │ Fault     │  │ │ Unix socket  │ │  │
│  │ • Nodes  │──│ Tolerance │  │ └──────────────┘ │  │
│  │ • Zenoh  │  └───────────┘  └──────────────────┘  │
│  │ • Timers │                                        │
│  └──────────┘                                        │
│                                                      │
│  ┌──────────────────────────────────────────────┐    │
│  │ Running Dataflows                            │    │
│  │  ├─ Node A (process) ◄──► TCP/Shmem          │    │
│  │  ├─ Node B (process) ◄──► TCP/Shmem          │    │
│  │  └─ Runtime (operators) ◄──► TCP/Shmem       │    │
│  └──────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────┘

Event loop (Daemon::run_inner()): Async Tokio event loop merging:

• Coordinator commands (WebSocket)
• Node events (TCP/shared memory)
• Inter-daemon events (Zenoh)
• Heartbeat (5s interval), metrics collection (2s), health checks (5s default)

Node spawning:

1. Create working directory for the node
2. Set up communication channel (TCP, shmem, or Unix domain socket)
3. Serialize NodeConfig to environment variable
4. Spawn process with sanitized environment (blocks LD_PRELOAD, DYLD_INSERT_LIBRARIES, etc.)
5. Monitor via ProcessHandle

Runtime

The runtime executes in-process operators (Python, shared library, WASM) in a dedicated process.

┌──────────────────────────────┐
│          Runtime             │
│                              │
│  ┌────────────────────────┐  │
│  │ Operator Runner        │  │
│  │ (separate thread)      │  │
│  │                        │  │
│  │ SharedLibrary → dlopen │  │
│  │ Python → PyO3          │  │
│  │ Wasm → (planned)       │  │
│  └──────────┬─────────────┘  │
│             │ flume(2)       │
│  ┌──────────▼─────────────┐  │
│  │ Event Merge Loop       │  │
│  │ ├─ OperatorEvent       │  │
│  │ └─ DaemonEvent         │  │
│  └────────────────────────┘  │
└──────────────────────────────┘

• Single-threaded Tokio runtime
• Operator runs in a separate thread, communicates via flume::bounded(2) channel
• Input queue size per data ID configurable (default: 10)

Nodes

Nodes are standalone processes that communicate with the daemon.

Lifecycle:

1. Node starts, reads NodeConfig from environment
2. Registers with daemon via DaemonRequest::Register
3. Subscribes to events via DaemonRequest::Subscribe
4. Processes events in a loop (NextEvent → handle → SendMessage)
5. Reports drop tokens for shared memory cleanup
6. Signals completion via OutputsDone

Communication Protocols

CLI to Coordinator (WebSocket)

JSON-RPC-like message format:

// Request (client → server)
{"id": "uuid", "method": "control", "params": {...}}

// Response (server → client)
{"id": "uuid", "result": {...}}
// or
{"id": "uuid", "error": "message"}

// Event (fire-and-forget, either direction)
{"event": "log", "payload": {...}}

Key control methods: Build, Start, Stop, List, Logs, TopicSubscribe, TopicUnsubscribe, Reload, Restart, Destroy.

Coordinator to Daemon (WebSocket)

Daemon events (daemon → coordinator): BuildResult, SpawnResult, AllNodesReady, AllNodesFinished, Heartbeat, StatusReport, Log, NodeMetrics, Exit.

Coordinator commands (coordinator → daemon): Build, Spawn, AllNodesReady, StopDataflow, ReloadDataflow, Logs, Destroy, Heartbeat.

Daemon to Node (Local)

Three transport options, configured via LocalCommunicationConfig:

TCP (default):

• Binds 127.0.0.1:0 (ephemeral port), TCP_NODELAY enabled
• Frame format: [8-byte u64 LE length][bincode payload]
• Max message: 64 MiB, read timeout: 30s

Shared Memory (zero-copy):

• Four 4 KiB regions per node: control, events, drop tokens, events-close
• Used for messages >= 4096 bytes (ZERO_COPY_THRESHOLD)
• Atomic synchronization with acquire/release ordering

Unix Domain Socket (Unix only):

• Socket at /tmp/{dataflow_id}/{node_id}.sock
• Permissions: 0o700
• Same bincode frame format as TCP

Node → Daemon requests: Register, Subscribe, SendMessage, CloseOutputs, OutputsDone, NextEvent, ReportDropTokens, SubscribeDrop, NodeConfig.

Daemon → Node replies: Result, PreparedMessage, NextEvents, NextDropEvents, NodeConfig, Empty.

Node events: Stop, Reload, Input, InputClosed, InputRecovered, NodeRestarted, AllInputsClosed.

Daemon to Daemon (Zenoh)

Topic pattern:

adora/{network_id}/{dataflow_id}/output/{node_id}/{output_id}

Default network_id is "default".

InterDaemonEvent:

• Output { dataflow_id, node_id, output_id, metadata, data } — data message
• OutputClosed { dataflow_id, node_id, output_id } — stream end

Message Types and Wire Formats

Timestamped Wrapper

All inter-component messages are wrapped in a timestamp:

#![allow(unused)]
fn main() {
pub struct Timestamped<T> {
    pub inner: T,
    pub timestamp: uhlc::Timestamp,  // 混合逻辑时钟
}
}

DataMessage

Transport abstraction for payloads:

#![allow(unused)]
fn main() {
pub enum DataMessage {
    Vec(AVec<u8, ConstAlign<128>>),    // inline, 128-byte aligned
    SharedMemory {
        shared_memory_id: String,
        len: usize,
        drop_token: DropToken,          // UUIDv7, tracks lifetime
    },
}
}

LogMessage

#![allow(unused)]
fn main() {
pub struct LogMessage {
    pub build_id: Option<BuildId>,
    pub dataflow_id: Option<DataflowId>,
    pub node_id: Option<NodeId>,
    pub daemon_id: Option<DaemonId>,
    pub level: LogLevelOrStdout,       // Stdout | LogLevel(Error/Warn/Info/Debug/Trace)
    pub target: Option<String>,
    pub module_path: Option<String>,
    pub file: Option<String>,
    pub line: Option<u32>,
    pub message: String,
    pub timestamp: DateTime<Utc>,
    pub fields: Option<BTreeMap<String, String>>,
}
}

NodeError

#![allow(unused)]
fn main() {
pub struct NodeError {
    pub timestamp: uhlc::Timestamp,
    pub cause: NodeErrorCause,         // GraceDuration | Cascading | FailedToSpawn | Other
    pub exit_status: NodeExitStatus,   // Success | IoError | ExitCode | Signal | Unknown
}
}

Data Format and Metadata

Apache Arrow

All data payloads use Apache Arrow columnar format with 128-byte alignment. Arrow type information is carried in every message via ArrowTypeInfo:

#![allow(unused)]
fn main() {
pub struct ArrowTypeInfo {
    pub data_type: DataType,           // Arrow DataType
    pub len: usize,
    pub null_count: usize,
    pub validity: Option<Vec<u8>>,     // null bitmap
    pub offset: usize,
    pub buffer_offsets: Vec<BufferOffset>,
    pub child_data: Vec<ArrowTypeInfo>,  // recursive for nested types
}
}

元数据

Every message carries structured metadata:

#![allow(unused)]
fn main() {
pub struct Metadata {
    metadata_version: u16,
    timestamp: uhlc::Timestamp,
    pub type_info: ArrowTypeInfo,
    pub parameters: MetadataParameters,   // BTreeMap<String, Parameter>
}
}

Parameter Types

#![allow(unused)]
fn main() {
pub enum Parameter {
    Bool(bool),
    Integer(i64),
    String(String),
    ListInt(Vec<i64>),
    Float(f64),
    ListFloat(Vec<f64>),
    ListString(Vec<String>),
    Timestamp(DateTime<Utc>),
}
}

Well-Known Metadata Keys

零拷贝共享内存

架构

┌────────────────────────────────────────────────────┐
│              Shared Memory Region                  │
│                                                    │
│  ┌──────────┐ ┌──────────┐ ┌──────┐ ┌────┐ ┌────┐│
│  │ Server   │ │ Client   │ │Discon│ │Len │ │Data││
│  │ Event    │ │ Event    │ │(bool)│ │(u64)│ │    ││
│  └──────────┘ └──────────┘ └──────┘ └────┘ └────┘│
│  (raw_sync_2)  (raw_sync_2) AtomicBool AtomicU64  │
└────────────────────────────────────────────────────┘

ShmemChannel

#![allow(unused)]
fn main() {
pub struct ShmemChannel {
    memory: Shmem,
    server_event: Box<dyn EventImpl>,
    client_event: Box<dyn EventImpl>,
    disconnect_offset: usize,
    len_offset: usize,
    data_offset: usize,
    server: bool,
}
}

Synchronization Protocol

Send (write → release store length → signal event → check disconnect):

1. Copy data to shared memory buffer
2. Store message length with Release ordering (publishes data)
3. Signal event to wake receiver
4. Check disconnect flag with Acquire ordering

Receive (wait event → check disconnect → acquire load length → read data):

1. Wait for event signal
2. Check disconnect flag with Acquire ordering
3. Load message length with Acquire ordering (ensures all writes visible)
4. Read and deserialize data from buffer

Thresholds and Limits

DropToken Lifecycle

1. Sender allocates shared memory, generates DropToken (UUIDv7)
2. Sender transmits DataMessage::SharedMemory { shared_memory_id, len, drop_token }
3. Receiver processes data, returns drop_token via ReportDropTokens
4. Sender receives confirmed token, returns memory to cache for reuse

Dataflow Specification

YAML Format

nodes:
  # Standard node (executable)
  - id: my-node
    build: cargo build --release
    path: target/release/my-node
    inputs:
      tick: adora/timer/millis/100
      data: other-node/output
    outputs:
      - result
    restart_policy: on-failure
    max_restarts: 3
    restart_delay: 1.0
    env:
      DEBUG: true

  # Single operator (Python)
  - id: processor
    operator:
      python: process.py
      inputs:
        image: camera/frame
      outputs:
        - detection

  # Multi-operator runtime
  - id: pipeline
    operators:
      - id: stage1
        python: stage1.py
        inputs:
          data: source/output
        outputs:
          - intermediate
      - id: stage2
        shared-library: target/release/libstage2.so
        inputs:
          data: stage1/intermediate
        outputs:
          - final

  # ROS2 bridge
  - id: ros-input
    ros2:
      topic: /robot/state
      message_type: sensor_msgs/JointState
      direction: subscribe
      qos:
        reliable: true
    outputs:
      - joints

Descriptor Structs

#![allow(unused)]
fn main() {
pub struct Descriptor {
    pub nodes: Vec<Node>,
    pub communication: CommunicationConfig,
    pub deploy: Option<Deploy>,
    pub debug: Debug,
    pub health_check_interval: Option<f64>,  // default 5.0s
}
}

Node types (mutually exclusive fields):

• path — standard executable/script
• operator — single in-process operator
• operators — multiple in-process operators
• custom — legacy configuration
• ros2 — declarative ROS2 bridge

Timer Nodes

Built-in timer nodes generate periodic ticks:

• adora/timer/millis/<N> — every N milliseconds
• adora/timer/secs/<N> — every N seconds

Operator Sources

#![allow(unused)]
fn main() {
pub enum OperatorSource {
    SharedLibrary(String),   // .so/.dll path
    Python(PythonSource),    // Python module
    Wasm(String),            // WebAssembly (planned)
}
}

Deploy Configuration

#![allow(unused)]
fn main() {
pub struct Deploy {
    pub machine: Option<String>,
    pub working_dir: Option<PathBuf>,
    pub labels: BTreeMap<String, String>,
    pub distribute: DistributeStrategy,  // Local | Scp | Http
}
}

容错

重启策略

#![allow(unused)]
fn main() {
pub enum RestartPolicy {
    Never,       // default
    OnFailure,   // restart on non-zero exit
    Always,      // restart unless user-stopped or inputs closed
}
}

Configuration fields per node:

• max_restarts — 0 = unlimited
• restart_delay — initial backoff in seconds (doubles each attempt)
• max_restart_delay — caps exponential backoff
• restart_window — reset counter after N seconds (enables “N restarts per M seconds”)
• health_check_timeout — kill node if no activity within this duration

健康监测

• Heartbeat interval: 5 seconds (daemon → coordinator)
• Health check interval: 5 seconds (configurable per dataflow)
• Metrics collection: 2-second interval (CPU, memory, disk, pending messages)

Circuit Breaker

Per-input timeout detection with automatic recovery:

1. Input configured with input_timeout: <seconds>
2. If no data arrives within timeout → InputClosed event sent to node
3. Node marks input as degraded, can use cached last-known value
4. When upstream recovers → InputRecovered event, circuit breaker re-opens
5. Node status transitions: Running → Degraded → Running

Cascading Error Tracking

#![allow(unused)]
fn main() {
pub struct CascadingErrorCauses {
    pub caused_by: BTreeMap<NodeId, NodeId>,
}
}

Tracks which node failure caused downstream failures, enabling root-cause analysis.

Fault Tolerance Metrics

#![allow(unused)]
fn main() {
pub struct FaultToleranceSnapshot {
    pub restarts: u64,
    pub health_check_kills: u64,
    pub input_timeouts: u64,
    pub circuit_breaker_recoveries: u64,
}
}

Reported per daemon via heartbeat events. Visible via adora inspect top.

分布式部署

Multi-Daemon Architecture

  ┌──────────┐       Zenoh        ┌──────────┐
  │ Daemon A │◄──────────────────►│ Daemon B │
  │ Machine 1│    pub/sub         │ Machine 2│
  │          │                    │          │
  │ Node 1   │                    │ Node 3   │
  │ Node 2   │                    │ Node 4   │
  └────┬─────┘                    └────┬─────┘
       │ WS                            │ WS
       └──────────┐  ┌────────────────┘
                  ▼  ▼
             ┌──────────┐
             │Coordinator│
             │  :6013    │
             └──────────┘

Zenoh Topic Naming

adora/{network_id}/{dataflow_id}/output/{node_id}/{output_id}

• network_id isolates separate Adora clusters (default: "default")
• Zenoh router port: 7447, peer port: 5456
• Routing mode: linkstate

Build Distribution

Three strategies via DistributeStrategy:

• Local — each daemon builds from source (default)
• Scp — CLI pushes built binaries via SSH/SCP
• Http — daemons pull from coordinator’s /api/artifacts endpoint

Machine Labels

Nodes can target specific machines via labels:

_unstable_deploy:
  labels:
    gpu: "true"
    arch: "arm64"

Recording and Replay

.adorec Binary Format

[HEADER]
├─ MAGIC: 8 bytes ("ADORAREC")
├─ version: u16 LE (currently 1)
├─ start_nanos: u64 LE (Unix epoch nanoseconds)
├─ dataflow_id: 16 bytes (UUID)
├─ yaml_len: u32 LE
└─ descriptor_yaml: [u8; yaml_len]

[ENTRIES] (repeated)
├─ record_len: u32 LE
├─ node_id_len: u16 LE
├─ node_id: [u8; node_id_len]
├─ output_id_len: u16 LE
├─ output_id: [u8; output_id_len]
├─ timestamp_offset_nanos: u64 LE
├─ event_bytes_len: u32 LE
└─ event_bytes: [u8; event_bytes_len]    (bincode InterDaemonEvent)

[FOOTER] (optional, written on clean finish)
├─ FOOTER_MAGIC: 8 bytes ("ADORAEND")
├─ total_messages: u64 LE
└─ total_bytes: u64 LE

Writer/Reader API

#![allow(unused)]
fn main() {
pub struct RecordingWriter<W: Write> { /* ... */ }
impl<W: Write> RecordingWriter<W> {
    pub fn new(inner: W, header: &RecordingHeader) -> Result<Self>;
    pub fn write_entry(&mut self, entry: &RecordEntry) -> Result<()>;
    pub fn finish(self) -> Result<RecordingFooter>;
}

pub struct RecordingReader<R: Read> { /* ... */ }
impl<R: Read> RecordingReader<R> {
    pub fn open(inner: R) -> Result<Self>;
    pub fn header(&self) -> &RecordingHeader;
    pub fn next_entry(&mut self) -> Result<Option<RecordEntry>>;
}
}

Extensions

Telemetry

Distributed Tracing (adora-tracing):

• OpenTelemetry with OTLP exporter (compatible with Jaeger, Zipkin, Tempo)
• Context propagation across nodes
• Setup: set_up_tracing(name: &str)

Metrics (adora-metrics):

• System metrics via sysinfo (CPU, memory, disk)
• OpenTelemetry meter with OTLP exporter
• Async process observer: run_metrics_monitor(meter_id)

ROS2 桥接

Declarative YAML-based ROS2 integration supporting:

Topics — subscribe (ROS2 → Adora) or publish (Adora → ROS2):

ros2:
  topic: /camera/image
  message_type: sensor_msgs/Image
  direction: subscribe

Services — client or server role:

ros2:
  service: /add_two_ints
  service_type: example_interfaces/AddTwoInts
  role: 客户端

Actions — goal/feedback/result lifecycle:

ros2:
  action: /fibonacci
  action_type: example_interfaces/Fibonacci
  role: 客户端

QoS configuration:

qos:
  reliable: true
  durability: transient_local
  keep_last: 10

Download

File download utility for fetching operator/node binaries from HTTP URLs. Sanitizes filenames, sets executable permissions on Unix.

Key Constants and Defaults

Identifiers and Data Structures

ID Types

Authentication

#![allow(unused)]
fn main() {
pub struct AuthToken(String);  // 64 hex chars (32 bytes)
}

• Generated via cryptographically random bytes
• Stored at <working_dir>/.adora-token
• Constant-time comparison to prevent timing attacks
• Applied to all WebSocket routes

Node Status

#![allow(unused)]
fn main() {
pub enum NodeStatus {
    Running,     // healthy
    Restarting,  // restart in progress
    Degraded,    // circuit breaker open (input timeout)
    Failed,      // terminal failure
}
}

Serialization Summary

Dataflow YAML Specification

Dataflows are defined in YAML files. Each file describes a graph of nodes, their inputs/outputs, and execution parameters.

A JSON Schema is available at the repo root (adora-schema.json) for editor autocompletion and validation.

快速开始

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

Run with adora run dataflow.yml (local mode) or adora up && adora start dataflow.yml (networked mode).

Editor Setup

Add a schema comment at the top of your YAML file for VS Code autocompletion (requires the YAML extension):

# yaml-language-server: $schema=https://raw.githubusercontent.com/dora-rs/adora/main/adora-schema.json
nodes:
  - id: my-node
    # ... autocompletion works here

Root-Level Fields

Node Configuration

Every node requires an id. All other fields are optional (though most nodes need at least path or operator/operators).

Identity

Source

A node’s executable comes from a local path, a git repository, a module reference, or is implicit (operator/ROS2 nodes).

Example with git source:

- id: rust-node
  git: https://github.com/dora-rs/adora.git
  branch: main
  build: cargo build -p example-node --release
  path: target/release/example-node

Data I/O

Inputs

Inputs subscribe to another node’s output using the format <node-id>/<output-id>:

inputs:
  # Short form
  image: camera/frames
  tick: adora/timer/millis/100

  # Long form with options
  sensor_data:
    source: sensor/frames
    queue_size: 10
    queue_policy: drop_oldest
    input_timeout: 5.0

  # Lossless input (blocks sender when full)
  commands:
    source: controller/cmd
    queue_size: 100
    queue_policy: backpressure

Built-in Timers

定时器是以固定间隔发出 tick 的虚拟节点：

inputs:
  tick: adora/timer/millis/100   # 每 100ms
  slow: adora/timer/millis/1000  # 每 1s
  fast: adora/timer/hz/30        # 30 Hz（约 33ms）

Built-in Log Aggregation

Subscribe to structured log messages from all (or filtered) nodes:

inputs:
  all_logs: adora/logs               # all nodes, all levels
  errors:   adora/logs/error         # error+ from all nodes
  sensor:   adora/logs/info/sensor   # info+ from specific node

Each message arrives as a JSON-encoded LogMessage string. See Logging for details.

Outputs

A list of output identifiers the node produces:

outputs:
  - processed_image
  - metadata

运维

Optional type annotations for inputs and outputs. Types are never required – unannotated ports remain fully dynamic.

- id: camera
  path: camera.py
  outputs:
    - image
    - depth
  output_types:
    image: std/media/v1/Image
    depth: std/media/v1/Image

- id: detector
  path: detect.py
  inputs:
    image: camera/image
  input_types:
    image: std/media/v1/Image
  outputs:
    - bbox
  output_types:
    bbox: std/vision/v1/BoundingBox

Type URNs use the format std/<category>/v<version>/<TypeName> and support parameters (e.g. std/media/v1/AudioFrame[sample_type=f32]). See the Type Annotations Guide for the full standard type library, parameterized types, compatibility rules, and user-defined types.

Run adora validate <file> to check type annotations statically. For runtime checking, set ADORA_RUNTIME_TYPE_CHECK=warn or error:

adora validate dataflow.yml
ADORA_RUNTIME_TYPE_CHECK=warn adora run dataflow.yml

Types also appear on adora graph edge labels when annotated.

Module Parameters

When using module:, pass configuration values via params::

- id: fast_pipeline
  module: modules/transform.module.yml
  inputs:
    data: sender/value
  params:
    speed: "2.0"
    mode: turbo

Inside the module, params are available as $PARAM_<UPPERCASE_KEY> in args: and as environment variables. See the Modules Guide for full documentation.

Environment

env:
  MY_VAR: "value"          # string
  DEBUG: true               # boolean
  PORT: 8080                # integer
  RATE: 1.5                 # float
  FROM_HOST:
    __adora_env: HOST_VAR   # read from host environment at runtime

Environment variables apply to both build commands and node execution. Values support $VAR expansion syntax.

日志

Example:

- id: sensor
  path: ./sensor
  min_log_level: info
  send_stdout_as: raw_output
  send_logs_as: log_entries
  max_log_size: "100MB"
  max_rotated_files: 3
  outputs:
    - data
    - raw_output
    - log_entries

When using send_stdout_as or send_logs_as, include the output name in the outputs list so downstream nodes can subscribe to it.

For a complete guide to all logging features, see Logging.

容错

Restart policies:

• never (default): no automatic restart
• on-failure: restart only on non-zero exit code
• always: restart on any exit, except when stopped by user or all inputs closed with success

Example with exponential backoff:

- id: sensor
  path: ./sensor
  restart_policy: on-failure
  max_restarts: 5
  restart_delay: 1.0         # 1s, 2s, 4s, 8s, 16s
  max_restart_delay: 30.0    # capped at 30s
  restart_window: 300.0      # 5 restarts per 5 minutes
  health_check_timeout: 30.0

Deployment

使用 _unstable_deploy 将节点分配到特定机器：

- id: camera-driver
  _unstable_deploy:
    machine: robot-arm
  path: ./target/debug/camera
  outputs:
    - frames

- id: ml-inference
  _unstable_deploy:
    machine: gpu-server
    labels:
      gpu: "true"
    distribute: scp
  path: ./target/debug/inference
  inputs:
    frames: camera-driver/frames

当节点位于不同机器时，通信自动从共享内存切换到 Zenoh 发布/订阅。

算子节点

Operators run in-process inside a shared runtime (no separate process). Use operator for a single operator or operators for multiple.

Single Operator

The id field is optional for single operators (defaults to the node id):

- id: detector
  operator:
    python: detect.py
    build: pip install -r requirements.txt
    inputs:
      image: camera/frames
    outputs:
      - bbox

Multiple Operators

Each operator in operators requires a unique id:

- id: runtime-node
  operators:
    - id: preprocessor
      shared-library: ../../target/debug/libpreprocess
      inputs:
        raw: sensor/data
      outputs:
        - processed
    - id: analyzer
      shared-library: ../../target/debug/libanalyze
      inputs:
        data: runtime-node/preprocessor/processed
      outputs:
        - result

Operator Source Types

Operators also support inputs, outputs, build, send_stdout_as, send_logs_as, min_log_level, max_log_size, and max_rotated_files with the same semantics as node-level fields.

ROS2 桥接

Declare a node as a ROS2 bridge to automatically convert between ROS2 DDS messages and Adora’s Arrow format. No custom code needed.

Single Topic

- id: camera_bridge
  ros2:
    topic: /camera/image_raw
    message_type: sensor_msgs/Image
    direction: subscribe
  outputs:
    - image

Multiple Topics

- id: robot_bridge
  ros2:
    topics:
      - topic: /camera/image_raw
        message_type: sensor_msgs/Image
        direction: subscribe
        output: image
      - topic: /cmd_vel
        message_type: geometry_msgs/Twist
        direction: publish
        input: velocity
    qos:
      reliable: true
  inputs:
    velocity: planner/cmd_vel
  outputs:
    - image

服务桥接

- id: add_service
  ros2:
    service: /add_two_ints
    service_type: example_interfaces/AddTwoInts
    role: 服务端
  inputs:
    request: client_node/request
  outputs:
    - response

动作桥接

- id: nav_action
  ros2:
    action: /navigate
    action_type: nav2_msgs/NavigateToPose
    role: 客户端
  inputs:
    goal: planner/goal
  outputs:
    - feedback
    - result

QoS Configuration

QoS can be set at the bridge level (applies to all topics) or per-topic:

Other ROS2 Fields

Debug

_unstable_debug:
  publish_all_messages_to_zenoh: true

Required for adora topic echo, adora topic hz, and adora topic info commands.

通信模式

Adora supports four communication patterns built on top of the dataflow:

• Topic (default): pub/sub dataflow
• Service: request/reply via request_id metadata
• Action: goal/feedback/result via goal_id/goal_status metadata, with cancellation support
• Streaming: session/segment/chunk via session_id/segment_id/seq/fin/flush metadata, with queue flush for interruption

See Communication Patterns for details and examples.

Full Example

health_check_interval: 10.0

_unstable_debug:
  publish_all_messages_to_zenoh: true

nodes:
  - id: webcam
    operator:
      python: webcam.py
      inputs:
        tick: adora/timer/millis/100
      outputs:
        - image

  - id: detector
    operator:
      python: detect.py
      build: pip install ultralytics
      inputs:
        image: webcam/image
      outputs:
        - bbox

  - id: plotter
    operator:
      python: plot.py
      inputs:
        image: webcam/image
        bbox: detector/bbox

  - id: logger
    path: ./logger
    inputs:
      bbox: detector/bbox
    send_stdout_as: logs
    min_log_level: info
    restart_policy: on-failure
    max_restarts: 3
    outputs:
      - logs

运维

Optional type annotations on dataflow inputs and outputs. Types are never required – unannotated ports remain fully dynamic. Type checking runs at build time and validate time (no runtime overhead by default).

快速开始

nodes:
  - id: camera
    path: camera.py
    outputs:
      - image
    output_types:
      image: std/media/v1/Image

  - id: detector
    path: detect.py
    inputs:
      image: camera/image
    input_types:
      image: std/media/v1/Image
    outputs:
      - bbox
    output_types:
      bbox: std/vision/v1/BoundingBox

Validate with:

adora validate dataflow.yml

# Fail with non-zero exit code on warnings (for CI)
adora validate --strict-types dataflow.yml

# Type checks also run during build
adora build dataflow.yml --strict-types

You can also set strict_types: true at the top level of the YAML to enable strict mode without the CLI flag:

strict_types: true
nodes:
  # ...

Type URN Format

Type URNs follow the pattern std/<category>/v<version>/<TypeName>:

std/core/v1/Float32
std/media/v1/Image
std/vision/v1/BoundingBox

Parameterized Types

Some struct types accept parameters to distinguish variants:

std/media/v1/AudioFrame[sample_type=f32]
std/media/v1/AudioFrame[sample_type=f32,channels=2]

Matching rules:

• Same base + same params -> compatible
• Same base + one side unparameterized -> compatible (wildcard)
• Same base + different param values -> mismatch

# These are compatible (wildcard):
output_types:
  audio: std/media/v1/AudioFrame[sample_type=f32]
input_types:
  audio: std/media/v1/AudioFrame

# These are a mismatch:
output_types:
  audio: std/media/v1/AudioFrame[sample_type=f32]
input_types:
  audio: std/media/v1/AudioFrame[sample_type=i16]

Standard Type Library

std/core/v1

std/math/v1

std/control/v1

std/media/v1

std/vision/v1

验证规则

adora validate and adora build check:

1. Key existence: output_types keys must appear in outputs, input_types keys must appear in inputs
2. URN resolution: All type URNs must exist in the standard or user-defined type library. Typos get “did you mean?” suggestions.
3. Edge compatibility: Connected edges must have compatible types (exact match, implicit widening, or user-defined rules)
4. Timer auto-typing: Timer inputs (adora/timer/*) are automatically typed as std/core/v1/UInt64
5. Type inference: When only the upstream side annotates a type, it is inferred on the downstream input and reported
6. Parameterized types: Parameter mismatches are detected (see above)
7. Metadata patterns: output_metadata keys and pattern shorthands are validated (see below)
8. Schema compatibility: Struct types are checked at the field level – missing fields or wrong field types are flagged

All checks produce warnings (non-fatal by default). Use --strict-types to treat warnings as errors for CI pipelines.

Type warnings:
  - node "camera": output_types key "framez" not found in outputs list
  - node "detector": unknown type "std/vision/v1/BoundingBx" on output "bbox"
    (did you mean "std/vision/v1/BoundingBox"?)
  - node "detector": type mismatch on input "image": upstream camera/image
    declares "std/core/v1/Bytes", but expected "std/media/v1/Image"

Inferred types:
  inferred std/core/v1/Float64 on processor/reading (from sensor/reading)

Type Compatibility Rules

Beyond exact matching, the type checker supports implicit widening conversions:

Widening is transitive up to depth 3 (e.g. UInt8 -> UInt32 -> UInt64 works, but chains of 4+ do not).

User-Defined Compatibility Rules

Add custom rules in the dataflow YAML:

type_rules:
  - from: myproject/SensorV1
    to: myproject/SensorV2

nodes:
  # ...

Metadata Patterns

Nodes that implement communication patterns (services, actions) can declare required metadata keys on their outputs.

Explicit metadata

- id: 服务端
  path: server.py
  outputs:
    - response
  output_metadata:
    response: [request_id]

Pattern shorthand

Use the pattern field to auto-imply required metadata keys:

- id: 服务端
  path: server.py
  pattern: service-server
  outputs:
    - response

User-Defined Types

Projects can define custom types in a types/ directory next to the dataflow. The directory structure determines the URN prefix:

project/
  dataflow.yml
  types/
    myproject/
      sensors/
        v1.yml    # URN prefix: myproject/sensors/v1

Type YAML files use the same format as the standard library:

types:
  MySensor:
    arrow: Struct
    description: Custom sensor reading
    fields:
      - name: temperature
        type: Float32
      - name: humidity
        type: Float32

This creates the URN myproject/sensors/v1/MySensor.

The std/ prefix is reserved and cannot be used for user types.

User types are loaded automatically by adora validate and adora build when a types/ directory exists.

Runtime Type Checking

In addition to static validation, Adora supports optional runtime type checking on send_output(). When enabled, the actual Arrow data type is compared against the declared output_types at send time.

Enable via environment variable:

# Warn on mismatches (log and continue)
ADORA_RUNTIME_TYPE_CHECK=warn adora run dataflow.yml

# Error on mismatches (node returns error)
ADORA_RUNTIME_TYPE_CHECK=error adora run dataflow.yml

Valid values: 1, warn, true (warn mode), error (error mode). Unset or any other value disables checking (zero overhead).

Scope:

• Validates output_types on the sender side (send_output() calls). input_types are checked statically by adora validate but not enforced at runtime
• Covers all languages that send Arrow arrays (Rust, Python, C++ Arrow path)
• Raw byte sends (send_output_bytes, C nodes) are untyped and skip checking
• Complex types (Struct-based: Image, Vector3, etc.) are skipped – only primitive types, String, Bytes, and Bool are validated at runtime

Graph Visualization

When outputs have type annotations, adora graph shows the type on edge labels:

adora graph dataflow.yml --open

Edges display as output_name [TypeName] (e.g. image [Image]).

Operators

Operators support the same output_types, input_types, output_metadata, and pattern fields:

- id: runtime-node
  operators:
    - id: preprocessor
      python: preprocess.py
      inputs:
        raw: sensor/data
      input_types:
        raw: std/core/v1/Bytes
      outputs:
        - processed
      output_types:
        processed: std/media/v1/Image

Modules (Reusable Sub-Dataflows)

Modules let you define reusable sub-graphs of nodes in separate YAML files and compose them into larger dataflows. Modules are expanded at compile time – the runtime never sees them.

快速开始

Module file (modules/transform_module.yml):

module:
  name: transform_pipeline
  inputs: [raw_data]
  outputs: [filtered]

nodes:
  - id: doubler
    path: doubler.py
    inputs:
      data: _mod/raw_data
    outputs:
      - doubled

  - id: filter
    path: filter_even.py
    inputs:
      data: doubler/doubled
    outputs:
      - filtered

Dataflow file (dataflow.yml):

nodes:
  - id: sender
    path: sender.py
    outputs:
      - value

  - id: pipeline
    module: modules/transform_module.yml
    inputs:
      raw_data: sender/value

  - id: receiver
    path: receiver.py
    inputs:
      filtered: pipeline/filtered

After expansion, pipeline becomes two nodes: pipeline.doubler and pipeline.filter, with all wiring resolved automatically.

Module Definition File

A module file has two sections:

module: header

nodes: list

Standard node definitions, with one special syntax: _mod/port_name references a module input port. When expanded, _mod/port_name is replaced with whatever the parent wired to that port.

module:
  name: my_module
  inputs: [camera_feed]
  outputs: [detections]

nodes:
  - id: detector
    path: detect.py
    inputs:
      image: _mod/camera_feed    # resolved to parent's wiring
    outputs:
      - detections

Module-level build

Modules can have a top-level build: command that runs before any inner node builds:

module:
  name: ml_pipeline
  inputs: [image]
  outputs: [result]

build: pip install -r requirements.txt

nodes:
  - id: model
    path: model.py
    inputs:
      image: _mod/image
    outputs:
      - result

Using Modules

Reference a module in a dataflow node using the module: field instead of path::

- id: nav_stack
  module: modules/navigation.module.yml
  inputs:
    goal_pose: localization/goal

The module node’s inputs: map wires parent outputs to module input ports. External nodes reference module outputs as <module_id>/<output_name> (e.g., nav_stack/cmd_vel).

Parameters

Pass configuration values to modules via params::

- id: fast_pipeline
  module: modules/transform_module.yml
  inputs:
    raw_data: sender/value
  params:
    speed: "2.0"
    mode: turbo

Inside the module, reference params in args: using $PARAM_<UPPERCASE_KEY>:

nodes:
  - id: processor
    path: processor.py
    args: --speed $PARAM_SPEED --mode $PARAM_MODE
    inputs:
      data: _mod/raw_data
    outputs:
      - result

Parameters are also injected as environment variables (PARAM_SPEED, PARAM_MODE) into every node inside the module.

Expansion Rules

1. Load the module YAML file and validate its header
2. Prefix all internal node IDs with {module_id}. (e.g., nav_stack.planner)
3. Replace _mod/port_name references with the actual sources from the parent’s input map
4. Rewrite internal cross-references (e.g., planner/path becomes nav_stack.planner/path)
5. Map module-declared outputs to internal node outputs, so nav_stack/cmd_vel resolves to nav_stack.controller/cmd_vel
6. Replace the module node with the expanded flat nodes
7. Substitute params: values in args: fields and inject as env vars

Use adora expand to see the result:

adora expand dataflow.yml

Nested Modules

Modules can reference other modules. The expansion is recursive with a depth limit of 8 levels:

# outer_module.yml
module:
  name: outer
  inputs: [data]
  outputs: [result]

nodes:
  - id: inner
    module: inner_module.yml
    inputs:
      raw: _mod/data

  - id: postprocess
    path: postprocess.py
    inputs:
      data: inner/processed
    outputs:
      - result

After expansion, node IDs are fully qualified: outer.inner.some_node.

Optional Inputs

Declare inputs as optional when a module should work with or without certain connections:

module:
  name: flexible_processor
  inputs: [data]
  inputs_optional: [config]
  outputs: [result]

nodes:
  - id: processor
    path: processor.py
    inputs:
      data: _mod/data
      config: _mod/config    # silently dropped if not wired
    outputs:
      - result

When the parent doesn’t wire config, the input is simply omitted from the expanded node.

Visualization

adora graph renders module boundaries as Mermaid subgraphs, making it easy to see which nodes came from which module:

adora graph dataflow.yml --open

Validation

Validate a standalone module file without a full dataflow:

adora expand --module modules/transform_module.yml

This checks:

• Valid YAML structure
• Module header is present with name, inputs, outputs
• All _mod/ references correspond to declared inputs or optional inputs
• No duplicate node IDs
• Internal wiring is consistent

安全

• Path confinement: Module file paths must resolve within the dataflow’s base directory. Absolute paths and directory traversal (../) outside the base are rejected.
• File size limit: Module files are capped at 1 MB.
• Depth limit: Recursive nesting is capped at 8 levels.
• Param key validation: Parameter keys must be alphanumeric with underscores only.

示例

See examples/module-dataflow/ for a complete working example with a sender, transform module (doubler + filter), and receiver.

adora run examples/module-dataflow/dataflow.yml

通信模式

Adora is a dataflow framework based on pub/sub message passing. On top of basic topics, the framework supports service (request/reply), action (goal/feedback/result), and streaming (session/segment/chunk) patterns using well-known metadata keys. No changes to the daemon, coordinator, or YAML syntax are required – the patterns are implemented as conventions at the node API level.

1. 主题（发布/订阅）

默认模式。一个节点在输出上发布数据，订阅该输出的任何节点都会接收到它。

nodes:
  - id: 发布者
    outputs:
      - data
  - id: 订阅者
    inputs:
      data: publisher/data

适用场景：流式传感器数据、周期性状态、即发即忘事件。

2. 服务（请求/应答）

客户端发送请求并期望收到恰好一个响应，通过 request_id 元数据键进行关联。

约定的元数据键

YAML

nodes:
  - id: 客户端
    inputs:
      tick: adora/timer/millis/500
      response: server/response
    outputs:
      - request

  - id: 服务端
    inputs:
      request: client/request
    outputs:
      - response

节点 API 辅助函数

#![allow(unused)]
fn main() {
// Client: send request with auto-generated request_id
let rid = node.send_service_request("request".into(), params, data)?;

// Server: pass through metadata.parameters (includes request_id)
node.send_service_response("response".into(), metadata.parameters, result)?;
}

服务端必须将传入请求的元数据参数中的 request_id 透传到响应中。客户端使用该键将响应与请求进行匹配。

示例：examples/service-example/

3. 动作（目标/反馈/结果）

客户端发送一个目标并接收周期性的反馈以及最终结果。Action 支持取消操作。

约定的元数据键

目标状态值：

YAML

nodes:
  - id: 客户端
    inputs:
      tick: adora/timer/millis/2000
      feedback: server/feedback
      result: server/result
    outputs:
      - goal
      - cancel

  - id: 服务端
    inputs:
      goal: client/goal
      cancel: client/cancel
    outputs:
      - feedback
      - result

取消模式

客户端在 cancel 输出上发送带有 goal_id 元数据的消息。服务端在处理步骤之间检查取消请求，并发送 goal_status = "canceled" 的结果。

示例：examples/action-example/

4. Streaming (session/segment/chunk)

For real-time pipelines (voice, video, sensor streams) where a user can interrupt mid-stream and queued data must be discarded.

约定的元数据键

YAML

nodes:
  - id: asr
    inputs:
      mic: mic-source/audio
    outputs:
      - text

  - id: llm
    inputs:
      text: asr/text
    outputs:
      - tokens

  - id: tts
    inputs:
      tokens: llm/tokens
    outputs:
      - audio

节点 API

#![allow(unused)]
fn main() {
use adora_node_api::{StreamSegment, AdoraNode};

let mut seg = StreamSegment::new();

// Send chunks with auto-incrementing seq (e.g. inside an ASR node)
node.send_stream_chunk("text".into(), &mut seg, false, chunk_data)?;
// Mark final chunk of a segment
node.send_stream_chunk("text".into(), &mut seg, true, last_chunk)?;

// On user interruption: flush downstream queues and start a new segment.
// The prior segment ends without a fin=true signal -- old data is discarded.
let flush_params = seg.flush();
node.send_output("text".into(), flush_params, empty_data)?;
}

Queue flush behavior

When a message arrives with flush: true in its metadata, the receiver’s input queue is cleared of all older messages before the flush message is delivered. This enables instant interruption in voice pipelines – when the user speaks over TTS output, the ASR node sends a new segment with flush: true, and the TTS node immediately discards any queued audio chunks from the previous response.

Note: flush discards all queued messages on the input regardless of session_id. Do not multiplex independent sessions on a single input when using flush.

Python

# Streaming metadata is a plain dict
params = {
    "session_id": session_id,
    "segment_id": 1,
    "seq": 0,
    "fin": False,
    "flush": True,  # flush older queued messages
}
node.send_output("text", data, metadata={"parameters": params})

5. Choosing a pattern

6. Important details

• goal_status 匹配区分大小写。 请始终使用精确的小写值："succeeded"、"aborted"、"canceled"。ROS2 桥接对无法识别的值默认使用 Aborted。

7. Python compatibility

Python 节点使用相同的元数据约定。参数是键为字符串的普通字典：

import uuid

# Service client (uuid7 for time-ordered IDs, matching Rust API)
params = {"request_id": str(uuid.uuid7())}
node.send_output("request", data, metadata={"parameters": params})

# Service server -- pass through parameters
node.send_output("response", result, metadata=event["metadata"])

注意：uuid.uuid7() 需要 Python 3.13+。在旧版本中，请使用 uuid_utils 包或 uuid.uuid4()（随机 v4 也可用于关联，但会丢失时间排序特性）。

Rust API 参考

本文档介绍用于构建 Adora 数据流组件的两个主要 Rust crate：

• adora-node-api – 用于独立节点可执行文件
• adora-operator-api – 用于由 Adora 运行时管理的进程内算子

节点 API (adora-node-api)

添加到你的 Cargo.toml：

[dependencies]
adora-node-api = { workspace = true }

AdoraNode

用于发送输出和获取节点信息的主要结构体。通过以下初始化函数之一获取。

初始化

#![allow(unused)]
fn main() {
// 推荐：自动检测环境（守护进程、测试或交互模式）。
pub fn init_from_env() -> NodeResult<(Self, EventStream)>

// 与 init_from_env 相同，但出错时不回退到交互模式。
pub fn init_from_env_force() -> NodeResult<(Self, EventStream)>

// 用于动态节点：通过节点 ID 连接到守护进程。
pub fn init_from_node_id(node_id: NodeId) -> NodeResult<(Self, EventStream)>

// 先尝试 init_from_env；回退到 init_from_node_id。
pub fn init_flexible(node_id: NodeId) -> NodeResult<(Self, EventStream)>

// 独立交互模式（在终端提示输入）。
pub fn init_interactive() -> NodeResult<(Self, EventStream)>

// 使用合成输入/输出的集成测试模式。
pub fn init_testing(
    input: TestingInput,
    output: TestingOutput,
    options: TestingOptions,
) -> NodeResult<(Self, EventStream)>
}

init_from_env 是推荐的入口点。它按顺序检查：

1. 由 setup_integration_testing 设置的线程局部测试状态
2. ADORA_NODE_CONFIG 环境变量（由守护进程设置）
3. ADORA_TEST_WITH_INPUTS 环境变量（基于文件的集成测试）
4. 交互终端回退（仅当 stdin 为 TTY 时）

发送输出

所有发送方法会静默忽略未在数据流 YAML 中声明的输出 ID。

#![allow(unused)]
fn main() {
// 发送 Arrow 数组。在有利时将数据复制到共享内存。
pub fn send_output(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<()>

// 发送原始字节。在有利时复制到共享内存。
pub fn send_output_bytes(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data_len: usize,
    data: &[u8],
) -> NodeResult<()>

// 通过闭包发送原始字节以实现零拷贝写入。
pub fn send_output_raw<F>(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data_len: usize,
    data: F,
) -> NodeResult<()>
where
    F: FnOnce(&mut [u8])

// 发送带有显式 Arrow 类型信息的原始字节。
pub fn send_typed_output<F>(
    &mut self,
    output_id: DataId,
    type_info: ArrowTypeInfo,
    parameters: MetadataParameters,
    data_len: usize,
    data: F,
) -> NodeResult<()>
where
    F: FnOnce(&mut [u8])

// 发送带有类型信息的预分配 DataSample。
pub fn send_output_sample(
    &mut self,
    output_id: DataId,
    type_info: ArrowTypeInfo,
    parameters: MetadataParameters,
    sample: Option<DataSample>,
) -> NodeResult<()>

// 将输出 ID 报告为已关闭。不再允许对这些 ID 发送。
pub fn close_outputs(&mut self, outputs_ids: Vec<DataId>) -> NodeResult<()>
}

Service, Action, and Streaming Helpers

Higher-level methods for the communication patterns. These use well-known metadata keys to correlate requests, goals, responses, and streaming segments.

#![allow(unused)]
fn main() {
// 生成唯一的、时间有序的 ID（UUID v7）用于关联。
pub fn new_request_id() -> String
pub fn new_goal_id() -> String   // new_request_id 的别名

// 发送服务请求。将 `request_id` 注入参数并返回。
pub fn send_service_request(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<String>

// 发送服务响应。send_output 的语义别名。
// 调用者必须传递来自传入请求元数据的 request_id。
pub fn send_service_response(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<()>
}

服务示例（客户端发送请求，服务端回复）：

#![allow(unused)]
fn main() {
// 客户端：自动生成并注入 request_id
let rid = node.send_service_request("request".into(), params, data)?;

// Server: pass through metadata.parameters (includes request_id)
node.send_service_response("response".into(), metadata.parameters, result)?;
}

动作示例（客户端发送目标，服务端流式传输反馈 + 结果）：

#![allow(unused)]
fn main() {
use adora_node_api::{GOAL_ID, GOAL_STATUS, GOAL_STATUS_SUCCEEDED, Parameter};

// 客户端：生成 goal_id，附加到参数
let goal_id = AdoraNode::new_goal_id();
params.insert(GOAL_ID.to_string(), Parameter::String(goal_id));
node.send_output("goal".into(), params, data)?;

// 服务端：提取 goal_id，发送带 goal_status 的反馈/结果
let gid = get_string_param(&metadata.parameters, GOAL_ID);
}

Streaming example (real-time voice/video pipeline with interruption):

#![allow(unused)]
fn main() {
use adora_node_api::StreamSegment;

// Create a streaming segment builder (auto-generates session_id)
let mut seg = StreamSegment::new();

// Send chunks with auto-incrementing seq
node.send_stream_chunk("text".into(), &mut seg, false, chunk_data)?;
// Mark final chunk of a segment
node.send_stream_chunk("text".into(), &mut seg, true, last_chunk)?;

// On user interruption: flush downstream queues and start a new segment
let flush_params = seg.flush();
node.send_output("text".into(), flush_params, empty_data)?;
}

See patterns.md for the full guide and examples/service-example and examples/action-example for working code.

数据分配

#![allow(unused)]
fn main() {
// 分配给定大小的 DataSample。
// 对 >= ZERO_COPY_THRESHOLD (4096 字节) 的数据使用共享内存。
pub fn allocate_data_sample(&mut self, data_len: usize) -> NodeResult<DataSample>
}

节点信息

#![allow(unused)]
fn main() {
// 数据流 YAML 中的节点 ID。
pub fn id(&self) -> &NodeId

// 本次数据流运行的唯一标识符。
pub fn dataflow_id(&self) -> &DataflowId

// 此节点的输入/输出配置。
pub fn node_config(&self) -> &NodeRunConfig

// 如果此节点在之前退出或故障后被重启则为 true。
pub fn is_restart(&self) -> bool

// 此节点被重启的次数（首次运行为 0）。
pub fn restart_count(&self) -> u32

// 解析后的数据流 YAML 描述符。
pub fn dataflow_descriptor(&self) -> NodeResult<&Descriptor>
}

日志

Rust nodes have two ways to emit structured logs. Both produce identical structured log entries in the daemon.

Option 1: Node API (recommended for most cases)

All log methods emit structured JSONL to stdout, which the daemon parses automatically. Works with min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

#![allow(unused)]
fn main() {
// 通用结构化日志。级别："error"、"warn"、"info"、"debug"、"trace"。
pub fn log(&self, level: &str, message: &str, target: Option<&str>)

// 带有附加键值字段的结构化日志。
pub fn log_with_fields(
    &self,
    level: &str,
    message: &str,
    target: Option<&str>,
    fields: Option<&BTreeMap<String, String>>,
)

// 便捷方法（无 target 参数）。
pub fn log_error(&self, message: &str)
pub fn log_warn(&self, message: &str)
pub fn log_info(&self, message: &str)
pub fn log_debug(&self, message: &str)
pub fn log_trace(&self, message: &str)
}

Option 2: Rust tracing crate

When adora’s tracing subscriber is initialized (via init_tracing() or the default feature), tracing::info!() etc. output structured JSON to stdout that the daemon parses identically:

#![allow(unused)]
fn main() {
tracing::info!("Sensor started");
tracing::warn!(sensor_id = "temp-01", "High temperature");
}

Use tracing when you want ecosystem integration (spans, instrumentation, OpenTelemetry). Use node.log_*() when you want explicit control or structured fields as BTreeMap.

EventStream

此节点传入事件的异步迭代器。实现了 futures::Stream trait。

收到 Stop 事件后，事件流会自行关闭。节点应在流结束后退出。

#![allow(unused)]
fn main() {
// 阻塞直到下一个事件到达。流关闭时返回 None。
// 使用内部 EventScheduler，可能为公平性重新排序事件。
pub fn recv(&mut self) -> Option<Event>

// 带超时的阻塞。超时时返回 Event::Error。
pub fn recv_timeout(&mut self, dur: Duration) -> Option<Event>

// 带 EventScheduler 重排序的异步接收。
pub async fn recv_async(&mut self) -> Option<Event>

// 带超时的异步接收。超时时返回 Event::Error。
pub async fn recv_async_timeout(&mut self, dur: Duration) -> Option<Event>

// 非阻塞接收。无可用数据时返回 TryRecvError::Empty。
pub fn try_recv(&mut self) -> Result<Event, TryRecvError>

// 非阻塞地排空所有缓冲事件。
// 无可用数据时返回 Some(Vec::new())；流关闭时返回 None。
pub fn drain(&mut self) -> Option<Vec<Event>>

// 如果调度器或接收器中没有缓冲事件则为 true。
pub fn is_empty(&self) -> bool

// Returns and resets accumulated drop counts per input ID.
// For `drop_oldest` inputs, drops happen at `queue_size`.
// For `backpressure` inputs, drops happen at 10x `queue_size` (hard safety cap).
pub fn drain_drop_counts(&mut self) -> HashMap<DataId, u64>
}

EventStream 还实现了 futures::Stream<Item = Event>，因此可以与 StreamExt::next() 和其他组合器一起使用。与 recv/recv_async 不同，Stream 实现不使用 EventScheduler，保留事件的时间顺序。

Event

表示传入事件。此枚举为 #[non_exhaustive]——忽略未知变体以保持向前兼容。

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum Event {
    // 从另一个节点接收到输入。
    Input {
        id: DataId,           // YAML 中的输入 ID（非发送者的输出 ID）
        metadata: Metadata,   // 时间戳和类型信息
        data: ArrowData,      // Apache Arrow 数据
    },

    // 映射到此输入的发送者已退出；不会再有数据到达。
    InputClosed { id: DataId },

    // 先前关闭的输入已恢复（例如，上游节点在超时后恢复）。
    InputRecovered { id: DataId },

    // 上游节点已重启。适用于重置缓存或状态。
    NodeRestarted { id: NodeId },

    // 事件流即将关闭。原因见 StopCause。
    Stop(StopCause),

    // 指示节点重新加载算子（运行时内部使用）。
    Reload { operator_id: Option<OperatorId> },

    // 意外的内部错误。记录日志用于调试。
    Error(String),
}
}

StopCause

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum StopCause {
    // 通过 `adora stop` 或 Ctrl-C 显式停止。请尽快退出，否则将被终止。
    Manual,

    // 所有输入已关闭（上游节点已退出）。仅在节点有输入时发送。
    AllInputsClosed,
}
}

辅助类型

DataSample

适合作为输出消息发送的数据区域。对 >= ZERO_COPY_THRESHOLD 的数据使用共享内存以实现零拷贝传输。

实现了 Deref<Target = [u8]> 和 DerefMut，用于读写底层字节。

Metadata 和 MetadataParameters

#![allow(unused)]
fn main() {
// 附加到每个输入事件的完整元数据。
pub struct Metadata {
    // 包含时间戳、Arrow 类型信息和用户自定义参数。
}

// 发送输出时附加的用户控制的元数据字段。
// BTreeMap<String, Parameter> 的类型别名。
// 默认为空。传递输入的 metadata.parameters 以转发元数据。
pub type MetadataParameters = BTreeMap<String, Parameter>;

// 单个元数据参数值。
pub enum Parameter {
    Bool(bool), Integer(i64), Float(f64), String(String),
    ListInt(Vec<i64>), ListFloat(Vec<f64>), ListString(Vec<String>),
    Timestamp(DateTime<Utc>),
}

// Extract typed parameters, returning None if missing or wrong type.
pub fn get_string_param<'a>(params: &'a MetadataParameters, key: &str) -> Option<&'a str>
pub fn get_integer_param(params: &MetadataParameters, key: &str) -> Option<i64>
pub fn get_bool_param(params: &MetadataParameters, key: &str) -> Option<bool>
}

Well-known metadata keys (for communication patterns):

所有常量均从 adora_node_api 重新导出。

标识类型

#![allow(unused)]
fn main() {
// 运行中数据流实例的唯一标识符（UUID v4）。
pub struct DataflowId(/* ... */);

// 数据流 YAML 中定义的节点标识符。
pub struct NodeId(/* ... */);

// 数据流 YAML 中定义的输入/输出标识符。
pub struct DataId(/* ... */);
}

错误类型

#![allow(unused)]
fn main() {
#[derive(Debug, Error)]
pub enum NodeError {
    Init(String),        // 配置解析、环境变量、守护进程握手
    Connection(String),  // 守护进程连接丢失
    Output(String),      // 发送或关闭失败
    Data(String),        // 分配或描述符解析
    Internal(eyre::Report),  // 意外错误的兜底
}

pub type NodeResult<T> = Result<T, NodeError>;
}

TryRecvError

#![allow(unused)]
fn main() {
pub enum TryRecvError {
    Empty,   // 当前没有可用事件
    Closed,  // 事件流已关闭
}
}

ZERO_COPY_THRESHOLD

#![allow(unused)]
fn main() {
pub const ZERO_COPY_THRESHOLD: usize = 4096;
}

小于此阈值的消息通过 TCP 发送。等于或超过此大小的消息使用共享内存进行零拷贝传输。

ArrowData

#![allow(unused)]
fn main() {
// arrow::array::ArrayRef 的包装器。实现了到内部 ArrayRef 的 Deref。
pub struct ArrowData(pub arrow::array::ArrayRef);
}

来自 Event::Input 的数据以 ArrowData 形式到达。使用 TryFrom 转换或 Arrow API 提取类型化的值。

InputTracker

用于跟踪输入健康状态和缓存每个输入最后接收值的辅助工具。在上游节点超时时用于优雅降级。

#![allow(unused)]
fn main() {
pub struct InputTracker { /* ... */ }

impl InputTracker {
    pub fn new() -> Self

    // 从事件更新状态。如果事件相关则返回 true。
    pub fn process_event(&mut self, event: &Event) -> bool

    // 输入的当前状态（Healthy 或 Closed），如果被追踪的话。
    pub fn state(&self, id: &DataId) -> Option<InputState>

    // 如果输入当前已关闭则为 true。
    pub fn is_closed(&self, id: &DataId) -> bool

    // 输入最后接收到的值。即使关闭后仍可用。
    pub fn last_value(&self, id: &DataId) -> Option<&ArrowData>

    // 所有当前处于 Closed 状态的输入。
    pub fn closed_inputs(&self) -> Vec<&DataId>

    // 如果任何被追踪的输入已关闭则为 true。
    pub fn any_closed(&self) -> bool
}

pub enum InputState {
    Healthy,  // 正常接收数据
    Closed,   // 上游退出或超时
}
}

集成测试

integration_testing 模块提供了无需运行守护进程即可测试节点的工具。

setup_integration_testing

设置线程局部状态，使同一线程上下一次调用 AdoraNode::init_from_env 时以测试模式初始化。

#![allow(unused)]
fn main() {
pub fn setup_integration_testing(
    input: TestingInput,
    output: TestingOutput,
    options: TestingOptions,
)
}

TestingInput

#![allow(unused)]
fn main() {
pub enum TestingInput {
    // 从 JSON 文件加载事件（必须反序列化为 IntegrationTestInput）。
    FromJsonFile(PathBuf),

    // 直接提供事件。
    Input(IntegrationTestInput),
}
}

TestingOutput

#![allow(unused)]
fn main() {
pub enum TestingOutput {
    // 将输出写入 JSONL 文件（创建或覆盖）。
    ToFile(PathBuf),

    // 将输出作为 JSONL 写入任意 writer。
    ToWriter(Box<dyn std::io::Write + Send>),

    // 将每个输出作为 JSON 对象发送到 flume 通道。
    ToChannel(flume::Sender<serde_json::Map<String, serde_json::Value>>),
}
}

TestingOptions

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Default)]
pub struct TestingOptions {
    // 跳过输出中的时间偏移以进行确定性比较。
    pub skip_output_time_offsets: bool,
}
}

环境变量测试

使用 init_from_env 的节点也支持通过环境变量进行基于文件的测试：

算子 API (adora-operator-api)

算子是由 Adora 运行时管理的进程内组件。它们被编译为共享库（.so/.dylib/.dll）并由运行时加载。

添加到你的 Cargo.toml：

[dependencies]
adora-operator-api = { workspace = true }

[lib]
crate-type = ["cdylib"]

AdoraOperator Trait

#![allow(unused)]
fn main() {
pub trait AdoraOperator: Default {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String>;
}
}

实现此 trait 以定义算子的行为。运行时对每个传入事件调用 on_event。返回 AdoraStatus 以控制执行流程。

Event（算子）

算子的 Event 枚举比节点的 Event 更简单，使用 &str 作为 ID。

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum Event<'a> {
    // 收到一个输入。
    Input { id: &'a str, data: ArrowData },

    // 将输入数据解析为 Arrow 数组失败。
    InputParseError { id: &'a str, error: String },

    // 输入被发送者关闭。
    InputClosed { id: &'a str },

    // 算子应停止。
    Stop,
}
}

AdoraOutputSender

#![allow(unused)]
fn main() {
pub struct AdoraOutputSender<'a>(/* ... */);

impl AdoraOutputSender<'_> {
    // 发送输出。`id` 是数据流 YAML 中的输出 ID。
    pub fn send(&mut self, id: String, data: impl Array) -> Result<(), String>
}
}

AdoraStatus

从 on_event 返回以控制算子生命周期。

#![allow(unused)]
fn main() {
pub enum AdoraStatus {
    Continue,  // 继续运行，等待下一个事件
    Stop,      // 停止此算子
    StopAll,   // 停止整个数据流
}
}

register_operator! 宏

生成 Adora 运行时加载和调用算子所需的 FFI 入口点。

#![allow(unused)]
fn main() {
use adora_operator_api::register_operator;

register_operator!(MyOperator);
}

每个 crate 必须在顶层精确调用一次，传入实现了 AdoraOperator 的类型。

快速开始示例：节点

一个接收 tick 输入并发送随机数作为输出的最小节点。

use adora_node_api::{AdoraNode, Event, IntoArrow, adora_core::config::DataId};

fn main() -> eyre::Result<()> {
    let (mut node, mut events) = AdoraNode::init_from_env()?;

    let output = DataId::from("random".to_owned());

    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, metadata, data } => {
                if id.as_str() == "tick" {
                    let value: u64 = fastrand::u64(..);
                    node.send_output(
                        output.clone(),
                        metadata.parameters,
                        value.into_arrow(),
                    )?;
                }
            }
            Event::Stop(_) => {}
            _ => {}
        }
    }

    Ok(())
}

对应的数据流 YAML：

nodes:
  - id: timer
    path: adora/timer/millis/100
    outputs:
      - tick

  - id: my-node
    path: ./target/debug/my-node
    inputs:
      tick: timer/tick
    outputs:
      - random

  - id: sink
    path: ./target/debug/sink
    inputs:
      data: my-node/random

快速开始示例：算子

一个计数 tick 并转发格式化消息的最小算子。

#![allow(unused)]
#![warn(unsafe_op_in_unsafe_fn)]

fn main() {
use adora_operator_api::{
    AdoraOperator, AdoraOutputSender, AdoraStatus, Event, IntoArrow, register_operator,
};

register_operator!(MyOperator);

#[derive(Debug, Default)]
struct MyOperator {
    ticks: usize,
}

impl AdoraOperator for MyOperator {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String> {
        match event {
            Event::Input { id, data } => match *id {
                "tick" => {
                    self.ticks += 1;
                    let msg = format!("tick count: {}", self.ticks);
                    output_sender.send("status".into(), msg.into_arrow())?;
                }
                other => eprintln!("ignoring unexpected input {other}"),
            },
            Event::InputClosed { id } => {
                if *id == "tick" {
                    return Ok(AdoraStatus::Stop);
                }
            }
            Event::Stop => {}
            other => {
                eprintln!("received unknown event {other:?}");
            }
        }

        Ok(AdoraStatus::Continue)
    }
}
}

对应的数据流 YAML：

nodes:
  - id: timer
    path: adora/timer/millis/500
    outputs:
      - tick

  - id: runtime-node
    operator:
      shared_library: ./target/debug/libmy_operator
      inputs:
        tick: timer/tick
      outputs:
        - status

Python API 参考

本文档涵盖用于构建 adora 节点、算子和数据流的 Python API。安装方式：

pip install adora-rs

目录

• 节点 API
  • Node 类
  • 事件字典
  • AdoraStatus 枚举
• 算子 API
  • Operator 类（用户定义）
  • send_output 回调
• DataflowBuilder
  • DataflowBuilder 类
  • Node 类（构建器）
  • Output 类（构建器）
  • Operator 类（构建器）
• CUDA 模块
• 快速开始示例
• DataflowBuilder 示例

节点 API

from adora import Node

Node 类是自定义节点的主要接口。它连接到运行中的数据流，接收输入事件并发送输出。

Node 类

__init__(node_id=None)

创建新节点并连接到运行中的数据流。

# Standard: node ID is read from environment variables set by the daemon
node = Node()

# Dynamic: connect to a running dataflow by explicit node ID
node = Node(node_id="my-dynamic-node")

Parameters:

• node_id（str，可选）—— 动态节点的显式节点 ID。省略时，节点从 adora 守护进程设置的环境变量中读取身份。

异常： 如果节点无法连接到数据流，则抛出 RuntimeError。

next(timeout=None)

从事件流中获取下一个事件。阻塞直到有事件可用或超时过期。

event = node.next()              # block indefinitely
event = node.next(timeout=2.0)   # block up to 2 seconds

Parameters:

• timeout（float，可选）—— 最大等待时间（秒）。

返回： dict —— 一个事件字典，如果所有发送者已被释放或超时过期则为 None。

drain()

非阻塞地获取所有缓冲事件。

events = node.drain()
for event in events:
    print(event["type"])

返回： list[dict] —— 事件字典列表。如果没有缓冲事件则返回空列表。

try_recv()

非阻塞接收。如果有可用的缓冲事件则返回。

event = node.try_recv()
if event is not None:
    print(event["type"])

返回： dict | None —— 一个事件字典，如果没有缓冲事件则为 None。

recv_async(timeout=None)

异步接收。配合 asyncio 使用。

event = await node.recv_async()
event = await node.recv_async(timeout=5.0)

Parameters:

• timeout（float，可选）—— 最大等待时间（秒）。超时时返回错误。

返回： dict | None —— 一个事件字典，如果所有发送者已被释放则为 None。

注意： 此方法为实验性质。PyO3 异步（Rust-Python FFI）集成仍在开发中。

is_empty()

检查事件流中是否有任何缓冲事件。

if not node.is_empty():
    event = node.try_recv()

返回： bool

send_output(output_id, data, metadata=None)

在输出通道上发送数据。

import pyarrow as pa

# Send raw bytes
node.send_output("status", b"OK")

# Send an Apache Arrow array (zero-copy capable)
node.send_output("values", pa.array([1, 2, 3]))

# Send with metadata
node.send_output("image", pa.array(pixels), {"camera_id": "front"})

Parameters:

• output_id（str）—— 数据流 YAML 中声明的输出名称。
• data（bytes | pyarrow.Array）—— 载荷。对简单数据使用 bytes，对零拷贝共享内存传输使用 pyarrow.Array。
• metadata（dict，可选）—— 附加到消息的键值对。支持的值类型：bool、int、float、str、list[int]、list[float]、list[str]、datetime.datetime。

异常： 如果 data 既不是 bytes 也不是 pyarrow.Array，则抛出 RuntimeError。

Service, action, and streaming patterns

Python nodes use the same metadata key conventions as Rust for communication patterns. Parameters are plain dicts with string keys.

约定的元数据键：

服务客户端示例：

import uuid

# Send a request with a unique request_id
request_id = str(uuid.uuid7())  # Python 3.13+; use uuid_utils or uuid.uuid4() on older versions
node.send_output("request", data, {"request_id": request_id})

服务服务端示例：

# Pass through the metadata (includes request_id) from the incoming request
node.send_output("response", result, event["metadata"])

动作客户端示例：

goal_id = str(uuid.uuid7())
node.send_output("goal", data, {"goal_id": goal_id})

Streaming example (flush downstream queues on user interruption):

params = {
    "session_id": session_id,
    "segment_id": 1,
    "seq": 0,
    "fin": False,
    "flush": True,
}
node.send_output("text", data, metadata={"parameters": params})

See patterns.md for the full guide.

日志

Python nodes can log using either Python’s built-in logging module (recommended) or the explicit node API.

Python logging module (auto-bridged):

When Node() is created, it automatically installs a handler that routes Python’s logging module through the adora daemon. No configuration needed:

import logging
from adora import Node

node = Node()  # Installs the logging bridge

logging.info("Sensor initialized")       # -> structured "info" log entry
logging.warning("High temperature")      # -> structured "warn" log entry
logging.debug("Raw bytes: %s", data)     # -> structured "debug" log entry

These log entries are captured with full metadata (level, message, file path, line number) and work with min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

Note: Do not call logging.basicConfig() before creating Node(). The constructor sets up the bridge; calling basicConfig() first may install a conflicting handler.

Explicit node API:

log(level, message, target=None, fields=None)

Emit a structured log message with optional target and key-value fields.

node.log("info", "Processing frame", target="vision")
node.log("error", "Sensor timeout", fields={"sensor": "lidar", "retry": "3"})

Parameters:

• level（str）—— 日志级别："error"、"warn"、"info"、"debug" 或 "trace"。
• message（str）—— 日志消息。
• target（str，可选）—— 目标模块或子系统名称。
• fields（dict[str, str]，可选）—— 结构化键值上下文字段。

Works with the daemon’s min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

log_error(message), log_warn(message), log_info(message), log_debug(message), log_trace(message)

Convenience methods for common log levels:

node.log_error("Connection failed")
node.log_warn("Temperature elevated")
node.log_info("Sensor initialized")
node.log_debug("Raw bytes received")
node.log_trace("Entering loop iteration")

Each is equivalent to node.log(level, message).

When to use which:

dataflow_descriptor()

以 Python 字典形式返回完整的数据流描述符（解析后的数据流 YAML）。

descriptor = node.dataflow_descriptor()
print(descriptor["nodes"])

返回： dict

node_config()

从数据流描述符返回此节点的配置块。

config = node.node_config()
model_path = config.get("model", "default.pt")

返回： dict

dataflow_id()

返回运行中数据流的唯一标识符。

print(node.dataflow_id())  # e.g. "a1b2c3d4-..."

返回： str

is_restart()

检查此节点是否在之前的退出或故障后被重启。用于决定是恢复保存的状态还是重新开始。

if node.is_restart():
    restore_checkpoint()

返回： bool

restart_count()

返回此节点被重启的次数。首次运行时返回 0，第一次重启后返回 1，以此类推。

print(f"Restart #{node.restart_count()}")

返回： int

merge_external_events(subscription)

将 ROS2 订阅流合并到节点的主事件循环中。调用此方法后，ROS2 消息以 kind 设置为 "external" 的事件形式到达。

from adora import Node, Ros2Context, Ros2Node, Ros2NodeOptions, Ros2Topic

node = Node()
ros2_context = Ros2Context()
ros2_node = ros2_context.new_node("listener", Ros2NodeOptions())
topic = Ros2Topic("/chatter", "std_msgs/String", ros2_node)
subscription = ros2_node.create_subscription(topic)

node.merge_external_events(subscription)

for event in node:
    if event["kind"] == "external":
        print("ROS2:", event["value"])
    elif event["type"] == "INPUT":
        print("Adora:", event["id"])

Parameters:

• subscription（adora.Ros2Subscription）—— 通过 adora ROS2 桥接创建的 ROS2 订阅。

迭代支持

Node 类实现了 __iter__ 和 __next__，因此可以直接迭代：

for event in node:
    match event["type"]:
        case "INPUT":
            process(event["value"])
        case "STOP":
            break

迭代器在每次迭代时无超时调用 next()。当事件流关闭时产生 None，从而终止循环。

事件字典

事件以普通 Python 字典形式返回。结构取决于事件类型。

INPUT

从另一个节点收到输入消息。

{
    "type": "INPUT",
    "id": "camera_image",          # input ID as declared in the dataflow YAML
    "kind": "adora",               # "adora" for dataflow events, "external" for ROS2
    "value": <pyarrow.Array>,      # the payload as an Apache Arrow array
    "metadata": {
        "timestamp": datetime,     # UTC-aware datetime.datetime
        "open_telemetry_context": "...",  # tracing context (if enabled)
        ...                        # any user-supplied metadata
    },
}

访问数据：

values = event["value"].to_pylist()     # convert to Python list
array = event["value"].to_numpy()       # convert to NumPy array

INPUT_CLOSED

输入通道已关闭（上游节点已完成）。

{
    "type": "INPUT_CLOSED",
    "id": "camera_image",
    "kind": "adora",
}

STOP

数据流正在关闭。

{
    "type": "STOP",
    "id": "MANUAL" | "ALL_INPUTS_CLOSED",   # stop cause
    "kind": "adora",
}

ERROR

运行时发生错误。

{
    "type": "ERROR",
    "error": "description of the error",
    "kind": "adora",
}

外部（ROS2）

使用 merge_external_events 时，ROS2 消息以如下形式到达：

{
    "kind": "external",
    "value": <pyarrow.Array>,   # the ROS2 message as an Arrow array
}

AdoraStatus 枚举

用作算子 on_event 方法的返回值以控制事件循环。

from adora import AdoraStatus

算子 API

算子在 adora 运行时进程内运行（无单独的操作系统进程）。它们被定义为名为 Operator 的 Python 类，具有 on_event 方法。

Operator 类（用户定义）

创建一个包含 Operator 类的 Python 文件：

from adora import AdoraStatus

class Operator:
    def __init__(self):
        # Initialize state here
        self.count = 0

    def on_event(self, adora_event, send_output) -> AdoraStatus:
        if adora_event["type"] == "INPUT":
            self.count += 1
            # Process the input and optionally send output
            send_output("result", b"processed", adora_event["metadata"])
        return AdoraStatus.CONTINUE

Methods:

• __init__(self) —— 算子加载时调用一次。在此初始化任何状态或模型。
• on_event(self, adora_event, send_output) -> AdoraStatus —— 对每个传入事件调用。必须返回 AdoraStatus 值。

on_event 的参数：

• adora_event（dict）—— 一个事件字典。
• send_output（callable）—— 发送输出数据的回调（见下文）。

运行时还会在算子实例上设置 self.dataflow_descriptor，值为解析后的数据流 YAML 字典。

send_output 回调

send_output 回调传递给 on_event，用于从算子发送数据。

send_output(output_id, data, metadata=None)

Parameters:

• output_id（str）—— 数据流 YAML 中声明的输出名称。
• data（bytes | pyarrow.Array）—— 载荷。
• metadata（dict，可选）—— 要附加的元数据。传递 adora_event["metadata"] 以传播追踪上下文。

Example:

import pyarrow as pa
from adora import AdoraStatus

class Operator:
    def on_event(self, adora_event, send_output) -> AdoraStatus:
        if adora_event["type"] == "INPUT":
            result = pa.array([42], type=pa.int64())
            send_output("output", result, adora_event["metadata"])
        return AdoraStatus.CONTINUE

DataflowBuilder

from adora.builder import DataflowBuilder, Node, Operator, Output

在 Python 中以编程方式构建数据流 YAML。

DataflowBuilder 类

__init__(name="adora-dataflow")

创建新的数据流构建器。

flow = DataflowBuilder("my-robot")

Parameters:

• name（str，可选）—— 数据流名称。默认为 "adora-dataflow"。

add_node(id, **kwargs) -> Node

向数据流添加节点。返回 Node 对象以供进一步配置。

sender = flow.add_node("sender")

Parameters:

• id（str）—— 唯一节点标识符。
• **kwargs —— 传递到 YAML 的额外节点配置。

返回： Node（构建器）

to_yaml(path=None) -> str | None

生成数据流的 YAML 表示。如果提供了 path，则写入文件并返回 None。否则返回 YAML 字符串。

# Write to file
flow.to_yaml("dataflow.yml")

# Get as string
yaml_str = flow.to_yaml()

Parameters:

• path（str，可选）—— 写入 YAML 的文件路径。

返回： str | None

上下文管理器

DataflowBuilder 支持 with 语句：

with DataflowBuilder("my-flow") as flow:
    flow.add_node("sender").path("sender.py")
    flow.to_yaml("dataflow.yml")

Node 类（构建器）

由 DataflowBuilder.add_node() 返回。所有 setter 方法返回 self 以支持链式调用。

path(path) -> Node

设置节点可执行文件或脚本的路径。

node.path("my_node.py")

args(args) -> Node

设置节点的命令行参数。

node.args("--verbose --port 8080")

env(env) -> Node

设置节点的环境变量。

node.env({"MODEL_PATH": "/models/yolo.pt"})

build(command) -> Node

设置节点的构建命令（启动前运行）。

node.build("pip install -r requirements.txt")

git(url, branch=None, tag=None, rev=None) -> Node

将 Git 仓库设置为节点的源。

node.git("https://github.com/org/repo.git", branch="main")

add_operator(operator) -> Node

将 Operator 附加到此节点。

op = Operator("detector", python="object_detection.py")
node.add_operator(op)

add_output(output_id) -> Output

在此节点上声明一个输出，并返回 Output 引用以用作输入源。

output = sender.add_output("data")

add_input(input_id, source, queue_size=None, queue_policy=None) -> Node

将此节点订阅到另一个节点的输出。

# Using an Output object
output = sender.add_output("data")
receiver.add_input("data", output)

# Using a string reference
receiver.add_input("tick", "adora/timer/millis/100")

# With a custom queue size
receiver.add_input("images", camera_output, queue_size=2)

# Lossless input (blocks sender when full)
receiver.add_input("commands", cmd_output, queue_size=100, queue_policy="backpressure")

Parameters:

• input_id（str）—— 此节点上的输入名称。
• source（str | Output）—— 字符串（"node_id/output_id"）或 Output 对象。
• queue_size（int，可选）—— 此输入的最大缓冲消息数。
• queue_policy (str, optional) – "drop_oldest" (default) or "backpressure" (buffers up to 10x queue_size before dropping).

to_dict() -> dict

返回节点的字典表示，用于 YAML 序列化。

Output 类（构建器）

由 Node.add_output() 返回。表示节点输出的引用，用作 add_input() 中的源。

output = sender.add_output("data")
receiver.add_input("sensor_data", output)
str(output)  # "sender/data"

Operator 类（构建器）

定义用于嵌入节点 YAML 配置中的算子。

__init__(id, name=None, description=None, build=None, python=None, shared_library=None, send_stdout_as=None)

op = Operator(
    id="detector",
    python="object_detection.py",
    send_stdout_as="detection_text",
)

Parameters:

• id（str）—— 唯一算子标识符。
• name（str，可选）—— 显示名称。
• description（str，可选）—— 可读的描述。
• build（str，可选）—— 加载前运行的构建命令。
• python（str，可选）—— Python 算子文件的路径。
• shared_library（str，可选）—— 共享库算子的路径。
• send_stdout_as（str，可选）—— 将算子的 stdout 作为具有此 ID 的输出路由。

to_dict() -> dict

返回用于 YAML 序列化的字典表示。

CUDA 模块

from adora.cuda import torch_to_ipc_buffer, ipc_buffer_to_ipc_handle, open_ipc_handle

通过 CUDA IPC 实现节点间零拷贝 GPU 张量共享的实用工具。需要支持 CUDA 的 PyTorch 和 Numba。

torch_to_ipc_buffer(tensor) -> tuple[pyarrow.Array, dict]

将 PyTorch CUDA 张量转换为包含 CUDA IPC 句柄的 Arrow 数组和元数据字典。通过数据流发送两者以无需复制即可共享 GPU 内存。

import torch
import pyarrow as pa
from adora import Node
from adora.cuda import torch_to_ipc_buffer

node = Node()
tensor = torch.randn(1024, 768, device="cuda")
ipc_buffer, metadata = torch_to_ipc_buffer(tensor)
node.send_output("gpu_data", ipc_buffer, metadata)

Parameters:

• tensor（torch.Tensor）—— CUDA 张量。

返回： tuple[pyarrow.Array, dict] —— IPC 句柄（int8 Arrow 数组），以及包含形状、步长、数据类型、大小、偏移和来源信息的元数据。

ipc_buffer_to_ipc_handle(handle_buffer, metadata) -> IpcHandle

从接收到的 Arrow 缓冲区和元数据重建 CUDA IPC 句柄。

from adora.cuda import ipc_buffer_to_ipc_handle

event = node.next()
ipc_handle = ipc_buffer_to_ipc_handle(event["value"], event["metadata"])

Parameters:

• handle_buffer（pyarrow.Array）—— 来自 event["value"] 的 Arrow 数组。
• metadata（dict）—— 来自 event["metadata"] 的元数据。

返回： numba.cuda.cudadrv.driver.IpcHandle

open_ipc_handle(ipc_handle, metadata) -> ContextManager[torch.Tensor]

打开 CUDA IPC 句柄并产生 PyTorch 张量。作为上下文管理器使用以确保正确清理。

from adora.cuda import ipc_buffer_to_ipc_handle, open_ipc_handle

event = node.next()
ipc_handle = ipc_buffer_to_ipc_handle(event["value"], event["metadata"])

with open_ipc_handle(ipc_handle, event["metadata"]) as tensor:
    result = tensor * 2  # use the GPU tensor directly

Parameters:

• ipc_handle（IpcHandle）—— 来自 ipc_buffer_to_ipc_handle 的句柄。
• metadata（dict）—— 包含形状、步长和数据类型信息的元数据字典。

返回： 产生 CUDA 上 torch.Tensor 的上下文管理器。

快速开始示例

一个接收图像、处理并发送结果的完整节点：

#!/usr/bin/env python3
"""示例节点：接收消息、转换并发送输出。"""

import logging

import pyarrow as pa
from adora import Node


def main():
    node = Node()

    for event in node:
        if event["type"] == "INPUT":
            input_id = event["id"]

            if input_id == "message":
                values = event["value"].to_pylist()
                number = values[0]

                # Create a struct array with multiple fields
                result = pa.StructArray.from_arrays(
                    [
                        pa.array([number * 2]),
                        pa.array([f"Message #{number}"]),
                    ],
                    names=["doubled", "description"],
                )
                node.send_output("transformed", result)
                logging.info("Transformed message %d", number)

        elif event["type"] == "STOP":
            logging.info("Node stopping")
            break


if __name__ == "__main__":
    main()

运行方式：

adora run dataflow.yml

DataflowBuilder 示例

以编程方式构建数据流，而非手动编写 YAML：

#!/usr/bin/env python3
"""构建一个简单的 sender -> receiver 数据流。"""

from adora.builder import DataflowBuilder, Operator

flow = DataflowBuilder("example-flow")

# Add a timer-driven sender node
sender = flow.add_node("sender")
sender.path("sender.py")
tick_output = sender.add_output("message")

# Add a receiver that subscribes to the sender
receiver = flow.add_node("receiver")
receiver.path("receiver.py")
receiver.add_input("message", tick_output)

# Add a node with a timer input
timed_node = flow.add_node("periodic")
timed_node.path("periodic.py")
timed_node.add_input("tick", "adora/timer/millis/100")

# Add a node with an operator
runtime_node = flow.add_node("runtime-node")
op = Operator("detector", python="object_detection.py")
runtime_node.add_operator(op)
runtime_node.add_input("image", "camera/image")

# Write or print the YAML
flow.to_yaml("dataflow.yml")
print(flow.to_yaml())

C API 参考

本文档涵盖 Adora 框架提供的两个 C API：用于独立 C 进程的节点 API 和用于 Adora 运行时加载的共享库算子的算子 API。

目录

• 节点 API (adora-node-api-c)
  • Initialization
  • 事件循环
  • 事件检查
  • Output
  • Logging
  • Enums
• 算子 API (adora-operator-api-c)
  • 生命周期函数
  • 事件处理
  • 输入读取
  • 输出发送
  • 内存管理
  • Structs
  • Enums
• 节点示例
• 算子示例
• 构建和链接

节点 API (adora-node-api-c)

头文件：apis/c/node/node_api.h Crate：adora-node-api-c（构建为 staticlib）

节点 API 供作为外部进程参与 Adora 数据流的独立 C 可执行文件使用。守护进程生成进程并设置节点在初始化期间读取的环境变量。

初始化

init_adora_context_from_env

void *init_adora_context_from_env();

从守护进程设置的环境变量初始化 Adora 节点上下文。成功时返回指向上下文的不透明指针，失败时返回 NULL。

返回的指针必须传递给所有期望上下文参数的后续节点 API 调用。节点完成后，使用 free_adora_context 释放。

free_adora_context

void free_adora_context(void *adora_context);

释放先前由 init_adora_context_from_env 创建的上下文。每个上下文必须恰好释放一次。释放后不得再次使用该指针。

事件循环

adora_next_event

void *adora_next_event(void *adora_context);

阻塞直到此节点有可用的下一个事件。返回指向事件的不透明指针，当所有事件流关闭时返回 NULL（表示节点应退出）。

返回的指针不能直接解引用。使用 read_adora_* 函数提取事件类型和载荷。完成后使用 free_adora_event 释放事件。

free_adora_event

void free_adora_event(void *adora_event);

释放先前由 adora_next_event 返回的事件。每个事件必须恰好释放一次。释放后，事件指针和所有派生指针（来自 read_adora_input_id、read_adora_input_data）都将失效。

事件检查

read_adora_event_type

enum AdoraEventType read_adora_event_type(void *adora_event);

返回给定事件的类型。可能的值请参见 AdoraEventType。

read_adora_input_id

void read_adora_input_id(void *adora_event, char **out_ptr, size_t *out_len);

从 AdoraEventType_Input 事件读取输入 ID。将字符串起始指针写入 *out_ptr，字节长度写入 *out_len。字符串是有效的 UTF-8 但不以 null 结尾；使用 out_len 确定其边界。

如果事件不是输入事件，则设置 *out_ptr = NULL 和 *out_len = 0。

返回的指针借用自事件。调用 free_adora_event 后将失效。

read_adora_input_data

void read_adora_input_data(void *adora_event, char **out_ptr, size_t *out_len);

从 AdoraEventType_Input 事件读取原始数据字节。将数据起始指针写入 *out_ptr，字节长度写入 *out_len。

如果事件不是输入事件或输入不携带数据，则设置 *out_ptr = NULL 和 *out_len = 0。

目前仅支持 UInt8 Arrow 数组。其他 Arrow 数据类型会导致运行时 panic。未来版本将使用 Arrow C Data Interface 以支持所有类型。

返回的指针借用自事件。调用 free_adora_event 后将失效。

read_adora_input_timestamp

unsigned long long read_adora_input_timestamp(void *adora_event);

以 uint64 值形式返回输入事件元数据中的混合逻辑时钟时间戳。如果事件不是输入事件则返回 0。

Output

adora_send_output

int adora_send_output(
    void *adora_context,
    const char *id_ptr,
    size_t id_len,
    const char *data_ptr,
    size_t data_len
);

向所有下游订阅者发送输出数据。输出 ID（id_ptr/id_len）必须是有效的 UTF-8 字符串，与数据流 YAML 中节点声明的某个输出匹配。数据（data_ptr/data_len）作为原始字节（UInt8 Arrow 数组）发送。

成功返回 0，错误返回 -1。错误通过 tracing 记录。

如果任何指针参数为 NULL，则立即返回 -1。

日志

adora_log

int adora_log(
    void *adora_context,
    const char *level_ptr,
    size_t level_len,
    const char *msg_ptr,
    size_t msg_len
);

通过 Adora 日志管道发送结构化日志消息。level 和 msg 都必须是有效的 UTF-8 字符串。

有效的日志级别："error"、"warn"、"info"、"debug"、"trace"。

成功返回 0，错误返回 -1。如果任何指针参数为 NULL，则立即返回 -1。

Enums

AdoraEventType

enum AdoraEventType {
    AdoraEventType_Stop,        // 请求优雅关闭
    AdoraEventType_Input,       // 有新输入数据可用
    AdoraEventType_InputClosed, // 输入流已关闭
    AdoraEventType_Error,       // 发生错误
    AdoraEventType_Unknown,     // 无法识别的事件类型
};

算子 API (adora-operator-api-c)

头文件：apis/c/operator/operator_api.h、apis/c/operator/operator_types.h Crate：adora-operator-api-c

算子 API 供加载到 Adora 运行时进程中的共享库（.so/.dylib/.dll）使用。与节点不同，算子没有自己的 main 函数。它们导出三个函数，由运行时在适当的生命周期点调用。

operator_types.h 头文件由 safer-ffi 自动生成，定义了所有 C 兼容的结构体和枚举类型。

生命周期函数

adora_init_operator

AdoraInitResult_t adora_init_operator(void);

运行时加载算子时调用一次。分配并初始化所有算子状态，然后通过 operator_context 字段返回。运行时在每次后续调用中传回此指针。

成功时返回 .result.error = NULL 的 AdoraInitResult_t。

adora_drop_operator

AdoraResult_t adora_drop_operator(void *operator_context);

算子被卸载时调用一次。释放与 operator_context 关联的所有资源。

成功时返回 .error = NULL 的 AdoraResult_t。

事件处理

adora_on_event

OnEventResult_t adora_on_event(
    RawEvent_t *event,
    const SendOutput_t *send_output,
    void *operator_context
);

每当此算子有事件到达时由运行时调用。检查 event 字段以确定事件类型：

使用 send_output 向下游节点发送数据（见 adora_send_operator_output）。返回带有适当 AdoraStatus_t 的 OnEventResult_t 以控制算子生命周期。

输入读取

adora_read_input_id

char *adora_read_input_id(const Input_t *input);

返回新分配的以 null 结尾的字符串，包含输入 ID。调用者必须使用 adora_free_input_id 释放。

adora_read_data

Vec_uint8_t adora_read_data(Input_t *input);

将输入数据作为字节数组读取。从输入中消费底层 Arrow 数组（每个事件只能读取一次数据）。如果输入没有数据或数据已被消费，则返回 .ptr = NULL 的 Vec_uint8_t。

调用者必须使用 adora_free_data 释放返回的数据。

输出发送

adora_send_operator_output

AdoraResult_t adora_send_operator_output(
    const SendOutput_t *send_output,
    const char *id,
    const uint8_t *data_ptr,
    size_t data_len
);

向下游订阅者发送输出数据。id 必须是以 null 结尾的字符串，与算子声明的某个输出匹配。数据（data_ptr/data_len）在内部被转换为 UInt8 Arrow 数组。

成功时返回 .error = NULL 的 AdoraResult_t。

内存管理

算子 API 分配的内存必须由调用者使用相应的函数释放：

void adora_free_input_id(char *input_id);
void adora_free_data(Vec_uint8_t data);

不调用这些函数会导致内存泄漏。不要对这些分配使用 free()——它们由 Rust 运行时分配，必须通过 API 释放。

Structs

Vec_uint8_t

typedef struct Vec_uint8 {
    uint8_t *ptr;
    size_t len;
    size_t cap;
} Vec_uint8_t;

Rust 分配的字节向量。从 ptr 开始访问 len 个字节。不要修改 cap。使用 adora_free_data 释放。

AdoraResult_t

typedef struct AdoraResult {
    Vec_uint8_t *error;  // 成功时为 NULL，失败时指向错误字符串
} AdoraResult_t;

通用结果类型。NULL 错误指针表示成功。非 NULL 时，错误指针包含 UTF-8 错误消息。

AdoraInitResult_t

typedef struct AdoraInitResult {
    AdoraResult_t result;
    void *operator_context;  // 算子状态的不透明指针
} AdoraInitResult_t;

由 adora_init_operator 返回。成功时 result.error 为 NULL，operator_context 持有算子状态指针。

OnEventResult_t

typedef struct OnEventResult {
    AdoraResult_t result;
    AdoraStatus_t status;
} OnEventResult_t;

由 adora_on_event 返回。包含错误/成功结果和控制算子生命周期的状态码。

RawEvent_t

typedef struct RawEvent {
    Input_t *input;           // 输入事件时为非 NULL
    Vec_uint8_t input_closed; // 输入流关闭时为非空
    bool stop;                // 请求关闭时为 true
    Vec_uint8_t error;        // 错误时为非空
} RawEvent_t;

表示传递给算子的事件。多个字段可能同时被设置；按优先级顺序检查它们。

Input_t

typedef struct Input Input_t;  // 不透明

表示输入事件数据的不透明类型。使用 adora_read_input_id 和 adora_read_data 提取其内容。

Output_t

typedef struct Output Output_t;  // 不透明

由 adora_send_operator_output 内部使用的不透明类型。不由用户代码直接创建。

SendOutput_t

typedef struct SendOutput {
    ArcDynFn1_AdoraResult_Output_t send_output;
} SendOutput_t;

传递给 adora_on_event 的回调句柄。将其传递给 adora_send_operator_output 以发送数据。不要将其存储在当前 adora_on_event 调用的范围之外。

Metadata_t

typedef struct Metadata {
    Vec_uint8_t open_telemetry_context;
} Metadata_t;

包含 OpenTelemetry 追踪上下文字符串的事件元数据。

算子枚举

AdoraStatus_t

enum AdoraStatus {
    ADORA_STATUS_CONTINUE = 0,  // 继续运行
    ADORA_STATUS_STOP     = 1,  // 停止此算子
    ADORA_STATUS_STOP_ALL = 2,  // 停止整个数据流
};
typedef uint8_t AdoraStatus_t;

在 OnEventResult_t 中返回，用于在处理事件后控制算子生命周期。

节点示例

一个接收定时器 tick 并发送输出消息的完整 C 节点：

#include <stdio.h>
#include <string.h>
#include "node_api.h"

int main() {
    void *ctx = init_adora_context_from_env();
    if (ctx == NULL) {
        fprintf(stderr, "failed to init adora context\n");
        return 1;
    }

    for (int i = 0; i < 100; i++) {
        void *event = adora_next_event(ctx);
        if (event == NULL)
            break;  // 所有流已关闭

        enum AdoraEventType ty = read_adora_event_type(event);

        if (ty == AdoraEventType_Input) {
            char *id;
            size_t id_len;
            read_adora_input_id(event, &id, &id_len);

            // 发送响应
            char out_id[] = "message";
            char out_data[64];
            int out_len = snprintf(out_data, sizeof(out_data),
                                   "iteration %d", i);

            adora_send_output(ctx, out_id, strlen(out_id),
                              out_data, out_len);
        } else if (ty == AdoraEventType_Stop) {
            free_adora_event(event);
            break;
        }

        free_adora_event(event);
    }

    free_adora_context(ctx);
    return 0;
}

节点的数据流 YAML：

nodes:
  - id: c_node
    path: build/c_node
    inputs:
      timer: adora/timer/millis/100
    outputs:
      - message

算子示例

一个读取输入、维护状态并发送输出的完整 C 算子：

#include "operator_api.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

AdoraInitResult_t adora_init_operator(void) {
    // 分配算子状态（一个简单的计数器）
    int *counter = (int *)calloc(1, sizeof(int));

    AdoraInitResult_t result = {.operator_context = counter};
    return result;
}

AdoraResult_t adora_drop_operator(void *operator_context) {
    free(operator_context);
    AdoraResult_t result = {.error = NULL};
    return result;
}

OnEventResult_t adora_on_event(
    RawEvent_t *event,
    const SendOutput_t *send_output,
    void *operator_context)
{
    OnEventResult_t result = {.status = ADORA_STATUS_CONTINUE};
    int *counter = (int *)operator_context;

    if (event->input != NULL) {
        char *id = adora_read_input_id(event->input);
        Vec_uint8_t data = adora_read_data(event->input);

        if (data.ptr != NULL) {
            *counter += 1;
            printf("received input '%s', counter: %d\n", id, *counter);

            // 将计数器值作为字符串发送
            char buf[64];
            int len = snprintf(buf, sizeof(buf), "count=%d", *counter);
            result.result = adora_send_operator_output(
                send_output, "counter", (uint8_t *)buf, len);

            adora_free_data(data);
        }

        adora_free_input_id(id);
    }

    if (event->stop) {
        result.status = ADORA_STATUS_STOP;
    }

    return result;
}

算子的数据流 YAML：

nodes:
  - id: runtime-node
    operators:
      - id: c_operator
        shared-library: build/operator
        inputs:
          data: source_node/output
        outputs:
          - counter

构建和链接

节点（静态库）

C 节点链接 adora-node-api-c，它构建为静态库。

步骤 1：构建静态库

cargo build -p adora-node-api-c --release

这会生成 target/release/libadora_node_api_c.a（Windows 上为 .lib）。

步骤 2：编译和链接

clang node.c -ladora_node_api_c -L ../../target/release -o build/c_node <FLAGS>

平台特定的链接器标志：

在 Windows 上，为输出文件添加 .exe 扩展名。

算子（共享库）

C 算子被编译为共享库，由 Adora 运行时在启动时加载。

步骤 1：编译为目标文件

clang -c operator.c -o build/operator.o -fdeclspec -fPIC

Windows 上省略 -fPIC。

步骤 2：链接为共享库

# Linux
clang -shared build/operator.o -o build/liboperator.so

# macOS
clang -shared build/operator.o -o build/liboperator.dylib

# Windows
clang -shared build/operator.o -o build/operator.dll

步骤 3：在数据流 YAML 中引用

operators:
  - id: c_operator
    shared-library: build/operator   # 不含 lib 前缀或扩展名
    inputs:
      data: source/output
    outputs:
      - result

shared-library 路径省略平台特定的前缀（lib）和扩展名（.so/.dylib/.dll）。运行时为当前平台解析正确的文件。

头文件路径

节点 API 头文件位于 apis/c/node/node_api.h。算子 API 头文件位于 apis/c/operator/operator_api.h 和 apis/c/operator/operator_types.h。相应调整你的头文件路径：

# Node
clang -I path/to/adora/apis/c/node node.c ...

# Operator
clang -I path/to/adora/apis/c/operator operator.c ...

C++ 兼容性

两组头文件都包含 extern "C" 保护（算子头文件中）或使用 C 兼容的声明（节点头文件中），因此可以直接从 C++ 源文件包含。

C++ API 参考

Adora 通过 CXX（Rust-C++ 互操作）为独立节点和进程内算子提供 C++ 绑定。CXX 桥接从 Rust 定义生成类型安全的 C++ 头文件——无需原始 FFI 或手动 extern "C" 声明。

两个 crate 提供 C++ 接口：

生成的头文件：adora-node-api.h 和 adora-operator-api.h。

节点 API（adora-node-api-cxx）

初始化

#include "adora-node-api.h"

// 从 Adora 守护进程设置的环境变量初始化节点。
// 返回包含事件流和输出发送器的 AdoraNode 结构体。
// 失败时抛出异常。
AdoraNode init_adora_node();

AdoraNode

由 init_adora_node() 返回。在节点的生命周期内拥有事件流和输出发送器。

struct AdoraNode {
    rust::Box<Events>        events;       // 事件流（阻塞接收器）
    rust::Box<OutputSender>  send_output;  // 输出发送器
};

Events

暴露给 C++ 的不透明 Rust 类型。提供对节点传入事件的阻塞迭代。

// 成员函数——直接在 boxed 对象上调用。
rust::Box<AdoraEvent> Events::next();

// 自由函数形式——等价于 events->next()。
rust::Box<AdoraEvent> next_event(rust::Box<Events>& events);

两种形式都会阻塞直到下一个事件到达，并返回一个拥有的 AdoraEvent。

AdoraEvent

不透明 Rust 类型。使用 event_type() 检查其类型，然后使用 event_as_input() 或 event_as_arrow_input() 进行向下转换。

// 确定事件类型。
AdoraEventType event_type(const rust::Box<AdoraEvent>& event);

// 向下转换为原始字节输入。如果事件不是 Input 则抛出异常。
AdoraInput event_as_input(rust::Box<AdoraEvent> event);

// 向下转换为 Arrow FFI 输入（写入 Arrow C Data Interface 结构体）。
// out_array 和 out_schema 必须指向有效的 ArrowArray / ArrowSchema 结构体。
// 成功时返回 error 为空的 AdoraResult。
AdoraResult event_as_arrow_input(
    rust::Box<AdoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

// 与上面相同，但还返回输入 ID 和元数据。
ArrowInputInfo event_as_arrow_input_with_info(
    rust::Box<AdoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

AdoraEventType

enum class AdoraEventType : uint8_t {
    Stop,             // 请求优雅关闭
    Input,            // 输入上有新数据到达
    InputClosed,      // 单个输入已关闭
    Error,            // 发生错误
    Unknown,          // 无法识别的事件变体
    AllInputsClosed,  // 所有输入已关闭（流已结束）
};

AdoraInput

由 event_as_input() 返回。包含原始字节。

struct AdoraInput {
    rust::String     id;    // 输入标识符（如 "tick"、"image"）
    rust::Vec<uint8_t> data;  // 原始载荷字节
};

ArrowInputInfo

由 event_as_arrow_input_with_info() 返回。包含输入 ID、元数据和错误字符串。

struct ArrowInputInfo {
    rust::String       id;        // 输入标识符
    rust::Box<Metadata> metadata; // 附加的元数据
    rust::String       error;     // 成功时为空
};

AdoraResult

由输出发送函数返回。检查 error 字段——为空表示成功。

struct AdoraResult {
    rust::String error;  // 成功时为空字符串
};

OutputSender

不透明 Rust 类型。所有方法以 rust::Box<OutputSender>& 作为第一个参数（来自 AdoraNode::send_output 的发送器）。

send_output

在命名输出上发送原始字节。

AdoraResult send_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data);

send_output_with_metadata

发送带有附加元数据的原始字节。

AdoraResult send_output_with_metadata(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data,
    rust::Box<Metadata> metadata);

send_arrow_output

通过 C Data Interface 发送 Arrow 数组。指针必须引用有效的 ArrowArray 和 ArrowSchema 结构体。成功时 Arrow 数据的所有权转移到 Rust。

AdoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr);

// 带元数据的重载（通过 cxx_name 属性使用相同的 C++ 名称）。
AdoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr,
    rust::Box<Metadata> metadata);

log_message

通过 Adora 日志系统发送日志消息。

AdoraResult log_message(
    const rust::Box<OutputSender>& sender,
    rust::String level,    // 如 "info"、"warn"、"error"
    rust::String message);

元数据

用于向输出附加类型化键值对的不透明 Rust 类型。

Construction

rust::Box<Metadata> new_metadata();

Reading

uint64_t     Metadata::timestamp() const;

bool         Metadata::get_bool(const rust::Str key) const;        // 缺失或类型错误时抛出异常
int64_t      Metadata::get_int(const rust::Str key) const;
double       Metadata::get_float(const rust::Str key) const;
rust::String Metadata::get_str(const rust::Str key) const;

rust::Vec<int64_t>      Metadata::get_list_int(const rust::Str key) const;
rust::Vec<double>       Metadata::get_list_float(const rust::Str key) const;
rust::Vec<rust::String> Metadata::get_list_string(const rust::Str key) const;

int64_t      Metadata::get_timestamp(const rust::Str key) const;   // 自纪元以来的纳秒数
rust::String Metadata::get_json(const rust::Str key) const;        // 单个值作为 JSON 字符串

Writing

所有 setter 在失败时抛出异常。

void Metadata::set_bool(const rust::Str key, bool value);
void Metadata::set_int(const rust::Str key, int64_t value);
void Metadata::set_float(const rust::Str key, double value);
void Metadata::set_string(const rust::Str key, rust::String value);

void Metadata::set_list_int(const rust::Str key, rust::Vec<int64_t> value);
void Metadata::set_list_float(const rust::Str key, rust::Vec<double> value);
void Metadata::set_list_string(const rust::Str key, rust::Vec<rust::String> value);

void Metadata::set_timestamp(const rust::Str key, int64_t nanos);  // 自纪元以来的纳秒数

Introspection

MetadataValueType Metadata::type(const rust::Str key) const;  // 键缺失时抛出异常
rust::String      Metadata::to_json() const;                   // 完整元数据作为 JSON
rust::Vec<rust::String> Metadata::list_keys() const;

MetadataValueType

enum class MetadataValueType : uint8_t {
    Bool,
    Integer,
    Float,
    String,
    ListInt,
    ListFloat,
    ListString,
    Timestamp,
};

Service, Action, and Streaming Patterns

C++ nodes can implement communication patterns using the metadata API. The well-known metadata keys are:

// 服务服务端：传递输入元数据中的 request_id
auto input_metadata = event_as_arrow_input_with_info(event);
send_output_with_metadata(sender, "response", result, std::move(input_metadata.metadata));

// 动作服务端：在结果上设置 goal_id 和 goal_status
auto meta = new_metadata();
meta->set_string("goal_id", goal_id);
meta->set_string("goal_status", "succeeded");
send_output_with_metadata(sender, "result", result_data, std::move(meta));

CombinedEvents（ROS2 集成）

使用可选的 ros2-bridge 特性时，节点事件和 ROS2 订阅事件可以合并为单个流。

// 将 Adora 事件转换为合并流。
CombinedEvents adora_events_into_combined(rust::Box<Events> events);

// 创建空的合并流（用于仅 ROS2 的节点）。
CombinedEvents empty_combined_events();

CombinedEvents 结构体

struct CombinedEvents {
    rust::Box<MergedEvents> events;

    CombinedEvent next();  // 阻塞——返回下一个合并事件
};

CombinedEvent 结构体

struct CombinedEvent {
    rust::Box<MergedAdoraEvent> event;

    bool is_adora() const;  // 如果这是标准 Adora 事件则为 true
};

// 将合并事件向下转换为 AdoraEvent。如果不是 Adora 事件则抛出异常。
rust::Box<AdoraEvent> downcast_adora(CombinedEvent event);

ROS2 subscriptions add their own events to the merged stream. Use subscription->matches(event) and subscription->downcast(event) to handle ROS2-specific events (see the ROS2 Bridge docs).

算子 API（adora-operator-api-cxx）

算子是由 Adora 运行时加载的共享库。C++ 端实现两个由 CXX 桥接调用的函数。

必需的 C++ 接口

你必须提供头文件 operator.h 和实现文件。头文件声明一个 Operator 类和两个自由函数：

// operator.h
#pragma once
#include <memory>
#include "adora-operator-api.h"

class Operator {
public:
    Operator();
    // 添加算子需要的任何状态。
};

std::unique_ptr<Operator> new_operator();

AdoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender);

• new_operator() —— 启动时调用一次；返回算子实例。
• on_input() —— 对每个输入事件调用；处理数据并可选地发送输出。

OutputSender（算子）

在 on_input() 内可用。在命名输出上发送数据。

AdoraSendOutputResult send_output(
    OutputSender& sender,
    rust::Str id,
    rust::Slice<const uint8_t> data);

结果类型

struct AdoraOnInputResult {
    rust::String error;  // 成功时为空
    bool         stop;   // 为 true 时请求优雅关闭
};

struct AdoraSendOutputResult {
    rust::String error;  // 成功时为空
};

快速开始：节点示例

一个接收定时器 tick 并发送计数器的最小节点。

#include "adora-node-api.h"
#include <iostream>
#include <vector>

int main() {
    auto adora_node = init_adora_node();
    unsigned char counter = 0;

    for (;;) {
        auto event = next_event(adora_node.events);
        auto ty = event_type(event);

        if (ty == AdoraEventType::AllInputsClosed) {
            break;
        }
        if (ty == AdoraEventType::Stop) {
            break;
        }
        if (ty == AdoraEventType::Input) {
            auto input = event_as_input(std::move(event));
            counter += 1;

            std::cout << "Input: " << std::string(input.id)
                      << " counter=" << (int)counter << std::endl;

            std::vector<unsigned char> out{counter};
            rust::Slice<const uint8_t> slice{out.data(), out.size()};
            auto result = send_output(adora_node.send_output, "counter", slice);
            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
                return 1;
            }
        }
    }
    return 0;
}

数据流 YAML：

nodes:
  - id: cxx-node
    path: build/my_node
    inputs:
      tick: adora/timer/millis/300
    outputs:
      - counter

快速开始：Arrow 节点示例

一个通过 C Data Interface 接收和发送 Arrow 数组的节点，带有元数据。

#include "adora-node-api.h"
#include <arrow/api.h>
#include <arrow/c/bridge.h>
#include <iostream>

int main() {
    auto adora_node = init_adora_node();

    for (int i = 0; i < 10; i++) {
        auto event = adora_node.events->next();
        auto ty = event_type(event);

        if (ty == AdoraEventType::AllInputsClosed || ty == AdoraEventType::Stop) {
            break;
        }
        if (ty == AdoraEventType::Input) {
            // 接收带元数据的 Arrow 输入
            struct ArrowArray c_array;
            struct ArrowSchema c_schema;
            auto info = event_as_arrow_input_with_info(
                std::move(event),
                reinterpret_cast<uint8_t*>(&c_array),
                reinterpret_cast<uint8_t*>(&c_schema));

            if (!info.error.empty()) {
                std::cerr << std::string(info.error) << std::endl;
                continue;
            }

            std::cout << "Input: " << std::string(info.id)
                      << " ts=" << info.metadata->timestamp() << std::endl;

            auto imported = arrow::ImportArray(&c_array, &c_schema);
            auto array = imported.ValueOrDie();
            std::cout << "Arrow: " << array->ToString() << std::endl;

            // 构建输出 Arrow 数组
            arrow::Int32Builder builder;
            builder.Append(i * 10);
            std::shared_ptr<arrow::Array> out_array;
            builder.Finish(&out_array);

            // 导出并带元数据发送
            struct ArrowArray out_c_array;
            struct ArrowSchema out_c_schema;
            arrow::ExportArray(*out_array, &out_c_array, &out_c_schema);

            auto meta = new_metadata();
            meta->set_string("source", "cpp-arrow-node");
            meta->set_int("iteration", i);

            auto result = send_arrow_output(
                adora_node.send_output, "counter",
                reinterpret_cast<uint8_t*>(&out_c_array),
                reinterpret_cast<uint8_t*>(&out_c_schema),
                std::move(meta));

            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
            }
        }
    }
    return 0;
}

快速开始：算子示例

一个最小的算子共享库。

// operator.cc
#include "operator.h"
#include <iostream>
#include <vector>

Operator::Operator() {}

std::unique_ptr<Operator> new_operator() {
    return std::make_unique<Operator>();
}

AdoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender)
{
    op.counter += 1;

    std::vector<unsigned char> out{op.counter};
    rust::Slice<const uint8_t> slice{out.data(), out.size()};
    auto send_result = send_output(output_sender, rust::Str("status"), slice);

    return AdoraOnInputResult{send_result.error, false};
}

数据流 YAML：

nodes:
  - id: runtime-node
    operators:
      - id: my-operator
        shared-library: build/my_operator
        inputs:
          data: some-node/output
        outputs:
          - status

构建集成（CMake）

推荐的构建方式是使用 CMake 配合 DoraTargets.cmake 辅助脚本（见 examples/cmake-dataflow/）。

项目结构

my-project/
  CMakeLists.txt
  DoraTargets.cmake       # copied from examples/cmake-dataflow/
  node/main.cc
  operator/operator.h
  operator/operator.cc
  dataflow.yml

CMakeLists.txt

cmake_minimum_required(VERSION 3.21)
project(my-dataflow LANGUAGES C CXX)

set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_FLAGS "-fPIC")

include(DoraTargets.cmake)
link_directories(${adora_link_dirs})

# Standalone node (executable)
add_executable(my_node node/main.cc ${node_bridge})
add_dependencies(my_node Adora_cxx)
target_include_directories(my_node PRIVATE ${adora_cxx_include_dir})
target_link_libraries(my_node adora_node_api_cxx)

# Operator (shared library)
add_library(my_operator SHARED
    operator/operator.cc ${operator_bridge})
add_dependencies(my_operator Adora_cxx)
target_include_directories(my_operator PRIVATE
    ${adora_cxx_include_dir} ${adora_c_include_dir}
    ${CMAKE_CURRENT_SOURCE_DIR}/operator)
target_link_libraries(my_operator adora_operator_api_cxx)

install(TARGETS my_node DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/bin)
install(TARGETS my_operator DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/lib)

DoraTargets.cmake 提供的内容

构建步骤

# Option A: Build against local Adora source
mkdir build && cd build
cmake .. -DDORA_ROOT_DIR=/path/to/adora
cmake --build .

# Option B: Build against Adora from GitHub (cloned automatically)
mkdir build && cd build
cmake ..
cmake --build .

要求

• C++20 编译器
• Rust 工具链（用于通过 Cargo 构建 Adora 静态库）
• CMake 3.21+
• Arrow 集成需要：Apache Arrow C++ 库

CXX 桥接注意事项

• 所有 Rust 不透明类型（Events、OutputSender、AdoraEvent、Metadata、MergedEvents、MergedAdoraEvent）通过 rust::Box<T> 访问。
• rust::String、rust::Vec<T> 和 rust::Slice<const T> 是 CXX 桥接类型，与对应的 C++ 标准库类型互操作。参见 CXX 类型参考。
• 在 Rust 中返回 Result<T> 的函数在错误路径上抛出 C++ 异常。
• Arrow FFI 函数（event_as_arrow_input、send_arrow_output）在 Rust 侧是 unsafe 的。调用者必须传递转换为 uint8_t* 的有效 ArrowArray / ArrowSchema 结构体指针。
• 节点库是静态归档（staticlib）。使用 -ladora_node_api_cxx 将其链接到你的可执行文件。
• 算子库也是静态归档。使用 -ladora_operator_api_cxx 将其链接到你的共享库。

Adora CLI 参考

Adora（AI-Dora，面向数据流的机器人架构）是一个 100% Rust 框架，用于构建实时机器人和 AI 应用。本文档从终端用户和开发者两个角度介绍 adora CLI。

目录

• 快速开始
• Installation
• 核心概念
• 数据流描述符
• 命令参考
  • Lifecycle
  • Monitoring
  • Debugging
  • Setup
  • Utility
  • Self-Management
• 环境变量
• 架构指南
• 编写节点
• 编写算子
• Distributed Deployments – see also Distributed Deployment Guide for cluster management, scheduling, and operations
• Troubleshooting
• 调试与可观测性 – 独立指南，涵盖录制/回放、主题检查、日志分析和资源监控
• API References: Rust | Python | C | C++

快速开始

# Create a new project
adora new my-robot --kind dataflow --lang rust

# Run locally (no coordinator/daemon needed)
adora run dataflow.yml

# Or use coordinator/daemon for production
adora up
adora start dataflow.yml --attach
# Ctrl-C to stop
adora down

安装

From crates.io (recommended)

cargo install adora-cli

从源码安装

cargo install --path binaries/cli --locked

验证

adora --version
adora status

核心概念

数据流

数据流是由类型化数据通道连接的节点有向图。节点产生输出，其他节点将其作为输入消费。框架负责数据路由、序列化（Apache Arrow）和生命周期管理。

执行模式

组件角色

CLI  -->  Coordinator  -->  Daemon(s)  -->  Nodes / Operators
              (control plane)  (per machine)    (user code)

• CLI：用户界面。发送命令、显示日志。
• 协调器：跨机器编排数据流生命周期。
• 守护进程：生成节点进程、管理 IPC、收集指标。
• 节点：产生和消费 Arrow 数据的独立进程。
• 算子：在共享运行时内运行的进程内代码（比节点更低延迟）。

数据格式

所有数据以 Apache Arrow 列式数组的形式在系统中流转。这使得同机节点间零拷贝共享内存传输和零序列化开销成为可能。

数据流描述符

数据流在 YAML 文件中定义。以下是完整的 schema：

最小示例

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

完整模式

# Dataflow-level settings
health_check_interval: 5.0    # health check sweep interval in seconds (default: 5.0)

nodes:
  - id: my-node                 # 唯一标识符（必填）
    name: "My Node"             # 人类可读名称（可选）
    description: "..."          # 描述（可选）

    # --- Source (pick one) ---
    path: ./target/debug/my-node          # 本地可执行文件
    # path: https://example.com/node.zip  # 从 URL 下载
    # git: https://github.com/org/repo.git  # 从 git 构建
    #   branch: main            # git 分支（与 tag/rev 互斥）
    #   tag: v1.0               # git 标签
    #   rev: abc123             # git 提交哈希

    # --- Build ---
    build: cargo build -p my-node   # 构建用的 shell 命令（可选）

    # --- Inputs ---
    inputs:
      # Short form: source_node/output_id
      tick: adora/timer/millis/100
      data: other-node/output

      # Long form with options
      sensor_data:
        source: sensor/frames
        queue_size: 10            # input buffer size (default: 10)
        queue_policy: drop_oldest # or "backpressure" (buffers up to 10x queue_size)
        input_timeout: 5.0        # circuit breaker timeout in seconds

    # --- Outputs ---
    outputs:
      - processed
      - status

    # --- Environment ---
    env:
      MY_VAR: "value"
      FROM_ENV:
        __adora_env: HOST_VAR     # 从宿主环境读取
    args: "--verbose"             # 命令行参数

    # --- Fault tolerance ---
    restart_policy: on-failure    # never（默认）| on-failure | always
    max_restarts: 5               # 0 = unlimited
    restart_delay: 1.0            # initial backoff in seconds
    max_restart_delay: 30.0       # backoff cap in seconds
    restart_window: 300.0         # reset counter after N seconds
    health_check_timeout: 30.0    # kill if no activity for N seconds

    # --- Logging ---
    min_log_level: info           # 源级别过滤（守护进程端）
    send_stdout_as: raw_output    # 将原始标准输出路由为数据输出
    send_logs_as: log_entries     # 将结构化日志路由为数据输出
    max_log_size: "50MB"          # 达到此大小时轮转日志文件
    max_rotated_files: 5          # number of rotated files to keep (1-100)

    # --- Deployment ---
    _unstable_deploy:
      machine: A                  # 目标机器/守护进程 ID

# 调试设置
_unstable_debug:
  publish_all_messages_to_zenoh: true   # required for topic echo/hz/info

内置定时器节点

定时器是以固定间隔发出 tick 的虚拟节点：

inputs:
  tick: adora/timer/millis/100   # 每 100ms
  slow: adora/timer/millis/1000  # 每 1s
  fast: adora/timer/hz/30        # 30 Hz（约 33ms）

算子节点

算子在共享运行时中进程内运行（无独立进程）：

nodes:
  # Single operator (shorthand)
  - id: detector
    operator:
      python: detect.py
      build: pip install -r requirements.txt
      inputs:
        image: camera/frames
      outputs:
        - bbox

  # Multiple operators sharing a runtime
  - id: runtime-node
    operators:
      - id: preprocessor
        shared-library: ../../target/debug/libpreprocess
        inputs:
          raw: sensor/data
        outputs:
          - processed
      - id: analyzer
        shared-library: ../../target/debug/libanalyze
        inputs:
          data: runtime-node/preprocessor/processed
        outputs:
          - result

分布式部署

使用 _unstable_deploy 将节点分配到特定机器：

nodes:
  - id: camera-driver
    _unstable_deploy:
      machine: robot-arm
    path: ./target/debug/camera
    outputs:
      - frames

  - id: ml-inference
    _unstable_deploy:
      machine: gpu-server
    path: ./target/debug/inference
    inputs:
      frames: camera-driver/frames
    outputs:
      - predictions

当节点位于不同机器时，通信自动从共享内存切换到 Zenoh 发布/订阅。

命令参考

生命周期命令

adora run

在本地运行数据流，无需协调器或守护进程。适合开发和测试。

adora run <PATH> [OPTIONS]

Examples:

# Basic run
adora run dataflow.yml

# Stop after 10 seconds, only show warnings
adora run dataflow.yml --stop-after 10s --log-level warn

# Python dataflow with uv
adora run dataflow.yml --uv

# Debug one node, silence others
adora run dataflow.yml --log-level warn --log-filter "sensor=debug"

# JSON output for CI pipelines
adora run dataflow.yml --log-format json --stop-after 30s 2>test.json

adora up

在本地模式下启动协调器和守护进程。

adora up

以后台进程方式启动 adora coordinator 和 adora daemon。等待两者就绪后返回。幂等操作：如果已在运行，则不执行任何操作。

adora down（别名：adora destroy）

拆卸协调器和守护进程。首先停止所有运行中的数据流。

adora down [OPTIONS]

adora build

运行数据流描述符中定义的构建命令。

adora build <PATH> [OPTIONS]

Type checking: After expanding modules, build runs the same type checks as validate. Warnings are printed by default; use --strict-types (or set strict_types: true in the YAML) to fail the build on type mismatches. User-defined types in a types/ directory next to the dataflow are loaded automatically.

**构建策略：**如果节点有 _unstable_deploy 部分且协调器可达，构建将分发到目标机器。否则在本地构建。

**Git 源：**带有 git: 字段的节点在构建前会被克隆/更新。构建命令从 git 仓库根目录运行。

adora start

在运行中的协调器上启动数据流。

adora start <PATH> [OPTIONS]

如果既未指定 --attach 也未指定 --detach：在 TTY 中运行时附加，否则分离。

**附加模式：**流式输出日志，优雅处理 Ctrl-C（第一次 = 停止，第二次 = 强制终止）。

**热重载：**监视 Python 算子源文件。文件变更时，向协调器发送重载请求，协调器将其传播到守护进程。

adora stop

停止运行中的数据流。

adora stop [UUID_OR_NAME] [OPTIONS]

如果未提供标识符且在 TTY 中运行，将显示交互式选择器。

**停止序列：**发送 Event::Stop -> 等待优雅关闭时间 -> SIGTERM -> 强制终止。

adora restart

重启运行中的数据流（停止 + 使用存储的描述符重新启动）。无需 YAML 路径 – 协调器保留了原始描述符。

adora restart [UUID] [OPTIONS]

Examples:

# Restart by name
adora restart --name my-app

# Restart by UUID with forced stop
adora restart a1b2c3d4-... --force

adora record

将数据流消息录制到 .adorec 文件以供离线回放。完整工作流请参阅调试指南。

adora record <DATAFLOW_YAML> [OPTIONS]

默认模式向数据流注入录制节点。--proxy 模式需要正在运行的数据流和 publish_all_messages_to_zenoh: true。

adora replay

通过将源节点替换为回放节点来回放录制的 .adorec 文件。完整工作流请参阅调试指南。

adora replay <FILE> [OPTIONS]

监控命令

adora list（别名：adora ps）

列出运行中的数据流及指标。

adora list [OPTIONS]

输出列： UUID、Name、Status、Nodes、CPU、Memory

adora logs

显示和跟踪数据流与节点的日志。

adora logs [UUID_OR_NAME] [NODE] [OPTIONS]

**过滤流水线：**读取/解析 -> 时间过滤 -> Grep -> Tail -> 显示

Examples:

# Follow all nodes live
adora logs my-dataflow --all-nodes --follow

# Last 50 errors from a specific node
adora logs my-dataflow sensor --level error --tail 50

# Search logs from last 5 minutes
adora logs my-dataflow --all-nodes --since 5m --grep "timeout"

# Read local files (no coordinator needed)
adora logs --local --all-nodes --tail 100

# Post-mortem analysis: errors in time window
adora logs --local sensor --since 1h --until 30m --level error

时间格式： 30（秒）、30s、5m、1h、2d

adora inspect top（别名：adora top）

节点资源使用的实时 TUI 监控器（类似 top）。

adora inspect top [OPTIONS]
adora top [OPTIONS]

需要交互式终端（除非使用 --once）。

列： NODE、STATUS、DATAFLOW、PID、CPU%、MEMORY (MB)、RESTARTS、QUEUE、NET TX、NET RX、I/O READ (MB/s)、I/O WRITE (MB/s)

• STATUS：Running、Restarting、Degraded（输入中断）或 Failed
• RESTARTS：每个节点的当前重启次数
• QUEUE：节点输入队列中的待处理消息
• NET TX/RX：通过 Zenoh 跨守护进程发送/接收的累计网络字节数

CPU 值为单核百分比（多核情况下可超过 100%）。指标来自守护进程，因此适用于分布式部署。

脚本示例：

# JSON snapshot for CI/monitoring pipelines
adora top --once | jq '.[].cpu_usage'

adora topic list

列出运行中数据流的所有主题（输出）。

adora topic list [OPTIONS]

adora topic echo

订阅主题并实时显示消息。

adora topic echo [OPTIONS] [DATA...]

需要在描述符中设置 _unstable_debug.publish_all_messages_to_zenoh: true。

adora topic hz

使用 TUI 仪表盘测量主题发布频率。

adora topic hz [OPTIONS] [DATA...]

**需要交互式终端。**显示：Avg (ms)、Avg (Hz)、Min (ms)、Max (ms)、Std (ms)，以及所选主题的速率火花图和直方图。

adora topic info

显示单个主题的详细元数据。

adora topic info [OPTIONS] DATA

订阅主题指定的持续时间并报告：类型（Arrow schema）、发布者、订阅者、消息计数、带宽。

adora node

管理和检查数据流节点。

adora node list

adora node list [OPTIONS]

列出运行中数据流的节点及其状态、CPU、内存和重启次数。

列： NODE、STATUS、PID、CPU%、MEMORY (MB)、RESTARTS、DATAFLOW

adora node info

Show detailed information about a specific node including status, inputs, outputs, and metrics.

adora node info <NODE> [OPTIONS]

adora node restart

Restart a single node within a running dataflow. The daemon stops the node process and respawns it.

adora node restart <NODE> [OPTIONS]

adora node stop

Stop a single node within a running dataflow without stopping the entire dataflow.

adora node stop <NODE> [OPTIONS]

adora topic pub

Publish JSON data to a topic in a running dataflow. Requires publish_all_messages_to_zenoh: true.

adora topic pub <TOPIC> [DATA] [OPTIONS]

Examples:

# Publish a single value
adora topic pub -d my-app sensor/threshold '[42]'

# Publish from file, 10 times
adora topic pub -d my-app sensor/config --file config.json --count 10

adora param

Manage runtime parameters for nodes. Parameters are persisted in the coordinator store and optionally forwarded to running nodes.

adora param list

List all runtime parameters for a node.

adora param list <NODE> [OPTIONS]

adora param get

Get a single runtime parameter value.

adora param get <NODE> <KEY> [OPTIONS]

adora param set

Set a runtime parameter. The value is JSON. The parameter is stored in the coordinator and forwarded to the node if it is running.

adora param set <NODE> <KEY> <VALUE> [OPTIONS]

Examples:

# Set a numeric parameter
adora param set -d my-app sensor threshold 42

# Set a string parameter
adora param set -d my-app camera resolution '"1080p"'

# Set a complex parameter
adora param set -d my-app detector config '{"confidence": 0.8, "nms": 0.5}'

adora param delete

Delete a runtime parameter.

adora param delete <NODE> <KEY> [OPTIONS]

adora doctor

Diagnose environment, coordinator/daemon connectivity, and optionally validate a dataflow YAML.

adora doctor [OPTIONS]

Checks performed:

1. Coordinator reachability
2. Daemon connectivity
3. Active dataflow status
4. Dataflow YAML validation (if --dataflow provided)

Examples:

# Basic health check
adora doctor

# Check environment + validate a dataflow
adora doctor --dataflow dataflow.yml

adora trace list

列出协调器捕获的最近追踪。协调器在内存中捕获 adora_coordinator 和 adora_core crate 的 span（最多 4096 个 span）。无需外部追踪基础设施。

adora trace list [OPTIONS]

输出列： TRACE ID（前 12 个字符）、ROOT SPAN、SPANS、STARTED、DURATION

Example:

adora trace list

TRACE ID      ROOT SPAN          SPANS  STARTED              DURATION
a1b2c3d4e5f6  spawn_dataflow     12     2026-03-01 10:30:05  1.234s
f8e7d6c5b4a3  build_dataflow     5      2026-03-01 10:29:58  0.500s

adora trace view

以缩进树形式查看特定追踪的 span。支持追踪 ID 的前缀匹配。

adora trace view <TRACE_ID> [OPTIONS]

Example:

adora trace view a1b2c3d4

spawn_dataflow [INFO 1.234s] {build_id="abc", session_id="def"}
  build_dataflow [INFO 0.500s]
    download_node [DEBUG 0.200s] {url="..."}
  start_inner [INFO 0.734s]
    spawn_node [INFO 0.100s] {node_id="camera"}
    spawn_node [INFO 0.080s] {node_id="detector"}

追踪 ID 支持前缀匹配：如果前缀唯一标识一个追踪，则自动解析。如果有歧义，系统会提示您使用更长的前缀。

设置命令

adora status（别名：adora check）

检查系统健康状态和连接性。

adora status [OPTIONS]

报告协调器连接状态、守护进程状态和活跃数据流数量。

adora new

从模板生成新的项目或节点。

adora new <NAME> [OPTIONS]

adora expand

Expand module references in a dataflow and print the resulting flat YAML. Useful for debugging module composition.

adora expand <PATH> [OPTIONS]

Examples:

# Expand a dataflow with modules
adora expand dataflow.yml

# Validate a module file
adora expand --module modules/navigation.module.yml

See the Modules Guide for full documentation on module composition.

adora graph

以图形方式可视化数据流。

adora graph <PATH> [OPTIONS]

Without --mermaid, generates an interactive HTML file using mermaid.js. When outputs have type annotations, edge labels include the type name (e.g. image [Image]).

# Generate HTML
adora graph dataflow.yml --open

# Generate Mermaid for GitHub markdown
adora graph dataflow.yml --mermaid

adora validate

Validate a dataflow YAML file and check type annotations.

adora validate <PATH> [OPTIONS]

Checks:

1. Key existence: output_types/input_types keys exist in the corresponding outputs/inputs lists
2. URN resolution: All type URNs resolve in the standard or user-defined type library
3. Edge compatibility: Connected edges have compatible types (exact match, widening, or user-defined rules)
4. Parameterized types: Parameter mismatches (e.g. AudioFrame[sample_type=f32] vs AudioFrame[sample_type=i16])
5. Timer auto-typing: Timer inputs are automatically typed as std/core/v1/UInt64
6. Type inference: When only upstream annotates a type, it is inferred on the downstream input
7. Metadata patterns: output_metadata keys and pattern shorthands are validated
8. Schema compatibility: Struct types are checked at the field level (missing/wrong fields)

User-defined types in a types/ directory next to the dataflow are loaded automatically.

# Validate with warnings
adora validate dataflow.yml

# Strict mode for CI (exit 1 on warnings)
adora validate --strict-types dataflow.yml

See the Type Annotations Guide for the full type library and usage details.

实用命令

adora completion

生成 shell 补全脚本。

adora completion [SHELL]

省略时自动检测 shell。支持：bash、zsh、fish、elvish、powershell。

# Bash
eval "$(adora completion bash)"
echo 'eval "$(adora completion bash)"' >> ~/.bashrc

# Zsh
eval "$(adora completion zsh)"
echo 'eval "$(adora completion zsh)"' >> ~/.zshrc

# Fish
adora completion fish > ~/.config/fish/completions/adora.fish

adora system

系统管理命令。

adora system status [OPTIONS]

目前提供 status 作为子命令（等同于 adora status）。

自管理命令

adora self update

检查并安装 CLI 更新。

adora self update [--check-only]

从 GitHub releases（dora-rs/adora）下载。

adora self uninstall

从系统中移除 CLI。

adora self uninstall [--force]

不使用 --force 时，会提示确认（需要 TTY）。依次尝试 uv pip uninstall、pip uninstall，然后二进制自删除。

环境变量

所有环境变量作为备选值。CLI 标志始终优先。

# Set defaults for a development session
export ADORA_COORDINATOR_ADDR=192.168.1.10
export ADORA_LOG_LEVEL=info
export ADORA_LOG_FORMAT=compact

架构指南

本节面向希望了解框架内部机制、扩展框架或调试问题的开发者。

通信栈

                    ┌─────────────────────────────────────┐
                    │           CLI (adora)                │
                    │   WebSocket (JSON request/reply)     │
                    └─────────────┬───────────────────────┘
                                  │
                    ┌─────────────▼───────────────────────┐
                    │        Coordinator                   │
                    │   WebSocket control + daemon mgmt    │
                    │   State: InMemoryStore | RedbStore   │
                    └──┬──────────────────────────────┬───┘
                       │                              │
          ┌────────────▼──────────┐     ┌─────────────▼──────────┐
          │     Daemon A          │     │     Daemon B           │
          │  (machine: robot)     │     │  (machine: gpu-server) │
          │                       │     │                        │
          │  ┌─────┐  ┌─────┐    │     │  ┌──────┐  ┌───────┐  │
          │  │Node1│  │Node2│    │     │  │Node3 │  │Node4  │  │
          │  └──┬──┘  └──┬──┘    │     │  └──┬───┘  └───┬───┘  │
          │     │shmem    │shmem  │     │     │shmem      │shmem │
          │     └────┬────┘       │     │     └─────┬─────┘      │
          └──────────┼────────────┘     └───────────┼────────────┘
                     │                              │
                     └──────── Zenoh pub/sub ────────┘
                              (cross-machine)

协议层

协调器内部机制

协调器是一个事件驱动的异步服务器：

Event Sources:
  - CLI WebSocket connections (ControlRequest)
  - Daemon WebSocket connections (DaemonEvent)
  - Heartbeat timer (3s interval)
  - External events (for embedding)

Event Loop:
  merge_all(cli_events, daemon_events, heartbeat, external)
    -> handle_event()
    -> update state
    -> persist to store (if redb)
    -> send replies

关键类型：

#![allow(unused)]
fn main() {
// State
RunningDataflow { uuid, name, descriptor, daemons, node_metrics, ... }
RunningBuild    { build_id, errors, log_subscribers, pending_results, ... }
DaemonConnection { sender, pending_replies, last_heartbeat }

// Store trait
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>>;
    // ... daemon and build methods
}
}

存储后端：

• memory（默认）：内存存储，重启后丢失。
• redb：持久化到磁盘（~/.adora/coordinator.redb）。崩溃后可恢复。需要 redb-backend feature。

adora coordinator --store redb
adora coordinator --store redb:/custom/path.redb

守护进程内部机制

守护进程管理单台机器上的节点进程：

Per Node:
  1. Build (if build command specified)
  2. Spawn process with ADORA_NODE_CONFIG env var
  3. Node registers via TCP/shmem handshake
  4. Route inputs/outputs between nodes
  5. Collect metrics (CPU, memory, I/O)
  6. Handle restart policy on exit
  7. Forward logs to coordinator

Communication:
  - Shared memory for messages > 4KB (zero-copy)
  - TCP for control messages and small data
  - flume channels for internal event routing

指标采集：

#![allow(unused)]
fn main() {
struct NodeMetrics {
    pid: u32,
    cpu_usage: f32,      // 单核百分比
    memory_mb: f64,
    disk_read_mb_s: Option<f64>,
    disk_write_mb_s: Option<f64>,
    status: NodeStatus,  // Running | Restarting | Degraded | Failed
    restart_count: u32,
    pending_messages: u64,
}
}

消息类型

所有组件间消息定义在 libraries/message/ 中：

#![allow(unused)]
fn main() {
// 节点标识
struct NodeId(String);      // [a-zA-Z0-9_.-]
struct DataId(String);      // 相同验证
type DataflowId = uuid::Uuid;

// 数据元数据
struct Metadata {
    timestamp: uhlc::Timestamp,    // 混合逻辑时钟
    type_info: ArrowTypeInfo,      // Arrow schema
    parameters: MetadataParameters, // 自定义键值对
}

// 节点事件（守护进程 -> 节点）
enum NodeEvent {
    Stop,
    Reload { operator_id },
    Input { id, metadata, data },
    InputClosed { id },
    InputRecovered { id },
    NodeRestarted { id },
    AllInputsClosed,
}
}

时间戳

Adora 使用统一混合逻辑时钟（UHLC）实现分布式因果关系。每条消息携带一个 uhlc::Timestamp，无需同步时钟即可在跨机器间保持因果顺序。

零拷贝共享内存

对于大消息（> 4KB），守护进程使用共享内存区域：

1. 发送节点向守护进程请求共享内存槽位
2. 守护进程分配区域并返回 ID
3. 发送方将 Arrow 数据直接写入共享内存
4. 守护进程通知接收节点区域 ID
5. 接收方直接从共享内存读取（零拷贝）
6. 接收方完成后发送释放令牌

对于大数据载荷，这比 ROS2 实现了 10-17 倍的延迟降低。

编写节点

Rust 节点

use adora_node_api::{AdoraNode, Event, IntoArrow};
use adora_core::config::DataId;

fn main() -> eyre::Result<()> {
    let (mut node, mut events) = AdoraNode::init_from_env()?;

    let output = DataId::from("result".to_owned());

    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, metadata, data } => {
                // 处理输入数据（Arrow 数组）
                let result: u64 = 42;
                node.send_output(
                    output.clone(),
                    metadata.parameters,
                    result.into_arrow(),
                )?;
            }
            Event::Stop(_) => break,
            Event::InputClosed { id } => {
                eprintln!("input {id} closed");
            }
            Event::InputRecovered { id } => {
                eprintln!("input {id} recovered");
            }
            _ => {}
        }
    }
    Ok(())
}

Cargo.toml:

[dependencies]
adora-node-api = { workspace = true }
eyre = "0.6"

Python 节点

import pyarrow as pa
from adora import Node

node = Node()

for event in node:
    if event["type"] == "INPUT":
        # event["value"] is a PyArrow array
        values = event["value"].to_pylist()
        result = pa.array([sum(values)])
        node.send_output("result", result)
    elif event["type"] == "STOP":
        break

C 节点

#include "node_api.h"

int main() {
    void *ctx = init_adora_context_from_env();
    // ... 使用 adora_next_event / adora_send_output 的事件循环
    free_adora_context(ctx);
    return 0;
}

节点日志

节点可以发出结构化日志：

Rust:

#![allow(unused)]
fn main() {
// 通过 tracing（推荐）
tracing::info!("processing frame {}", frame_id);

// 通过节点 API
node.log_info("processing complete");
node.log_with_fields("info", "reading", None, Some(&fields));
}

Python:

import logging
logging.info("processing frame %d", frame_id)

# Or via node API
node.log("info", "processing complete")

编写算子

算子在共享运行时内以进程内方式运行，避免了进程启动开销。

Rust 算子

#![allow(unused)]
fn main() {
use adora_operator_api::{register_operator, AdoraOperator, AdoraOutputSender, AdoraStatus, Event};

#[register_operator]
#[derive(Default)]
pub struct MyOperator {
    counter: u32,
}

impl AdoraOperator for MyOperator {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String> {
        match event {
            Event::Input { id, data } => {
                self.counter += 1;
                output_sender.send(
                    "count".to_string(),
                    arrow::array::UInt32Array::from(vec![self.counter]),
                )?;
                Ok(AdoraStatus::Continue)
            }
            Event::Stop => Ok(AdoraStatus::Stop),
            _ => Ok(AdoraStatus::Continue),
        }
    }
}
}

Cargo.toml:

[lib]
crate-type = ["cdylib"]

[dependencies]
adora-operator-api = { workspace = true }
arrow = "53"

Python 算子

nodes:
  - id: my-node
    operator:
      python: my_operator.py
      inputs:
        data: source/output
      outputs:
        - result

# my_operator.py
class Operator:
    def __init__(self):
        self.counter = 0

    def on_event(self, event, send_output):
        if event["type"] == "INPUT":
            self.counter += 1
            send_output("result", pa.array([self.counter]))

分布式部署

设置

# Machine A (coordinator + daemon)
adora up

# Machine B (daemon only, pointing to coordinator on Machine A)
adora daemon --interface 0.0.0.0 --coordinator-addr 192.168.1.10 --machine-id B

# Machine C (same)
adora daemon --interface 0.0.0.0 --coordinator-addr 192.168.1.10 --machine-id C

带机器分配的数据流

nodes:
  - id: camera
    _unstable_deploy:
      machine: robot
    path: ./camera-driver
    outputs:
      - frames

  - id: inference
    _unstable_deploy:
      machine: gpu-server
    path: ./ml-model
    inputs:
      frames: camera/frames
    outputs:
      - predictions

  - id: actuator
    _unstable_deploy:
      machine: robot
    path: ./actuator-driver
    inputs:
      commands: inference/predictions

构建和启动

# From any machine with coordinator access
adora build dataflow.yml       # distributed build on target machines
adora start dataflow.yml --name my-robot --attach

监控

# Resource usage across all machines
adora top

# Logs from any node regardless of machine
adora logs my-robot inference --follow

# List all dataflows
adora list

协调器持久化

在生产环境中，使用 redb 存储后端以使协调器在重启后存续：

adora coordinator --store redb

状态持久化到 ~/.adora/coordinator.redb。重启时，过期的数据流会被标记为失败，协调器恢复正常运行。

For managed cluster deployments (cluster.yml, SSH-based lifecycle, label scheduling, systemd services, rolling upgrades), see the Distributed Deployment Guide.

故障排除

关于涵盖录制/回放工作流、主题检查、资源监控和端到端调试场景的综合调试指南，请参阅调试与可观测性指南。

常见问题

“Could not connect to adora-coordinator”

• 先运行 adora up，或检查 ADORA_COORDINATOR_ADDR/ADORA_COORDINATOR_PORT
• 使用 adora status 验证

“publish_all_messages_to_zenoh not enabled”

• 使用 --debug 标志：adora start dataflow.yml --debug 或 adora run dataflow.yml --debug

• 或在数据流 YAML 中添加：

  _unstable_debug:
    publish_all_messages_to_zenoh: true
  

• topic echo、topic hz、topic info 需要此设置

“adora top requires an interactive terminal”

• 这些 TUI 命令需要真实终端（非管道输出）
• 同样适用于 topic hz

节点未接收到输入

• 检查输出名称是否匹配：source_node/output_id
• 验证源节点在其 outputs: 数组中列出了该输出
• 使用 adora topic list 检查可用主题

日志未显示

• 检查 --log-level 设置（默认 stdout 显示所有内容）
• 检查 YAML 中的 min_log_level（在源端过滤）
• 分布式环境下：验证协调器/守护进程的连接性

使用 git 源构建失败

• 验证 git: URL 是否可访问
• 检查 branch、tag 或 rev 是否存在
• 构建命令从 git 仓库根目录运行，而非数据流目录

调试工作流

# 1. Full environment diagnosis
adora doctor --dataflow dataflow.yml

# 2. Start with verbose logging and debug topics
adora run dataflow.yml --log-level trace --debug

# 3. Inspect a specific node
adora node info -d my-dataflow problem-node

# 4. Monitor specific node logs
adora logs my-dataflow problem-node --follow --level debug

# 5. Check resource usage
adora top

# 6. Inspect topic data
adora topic echo -d my-dataflow problem-node/output

# 7. Publish test data to a topic
adora topic pub -d my-dataflow problem-node/input '[1, 2, 3]'

# 8. Measure frequencies
adora topic hz -d my-dataflow --window 5

# 9. View/modify runtime parameters
adora param list -d my-dataflow problem-node
adora param set -d my-dataflow problem-node threshold 42

# 10. Restart a misbehaving node without stopping the dataflow
adora node restart -d my-dataflow problem-node

# 11. View coordinator traces (no external infra needed)
adora trace list
adora trace view <trace-id-prefix>

# 12. Visualize dataflow graph
adora graph dataflow.yml --open

日志文件位置

out/
  <dataflow-uuid>/
    log_<node-id>.jsonl          # current log
    log_<node-id>.1.jsonl        # rotated (previous)
    log_<node-id>.2.jsonl        # rotated (older)

直接读取：

adora logs --local --all-nodes
adora logs --local <node-name> --tail 50

日志

Adora 为实时机器人和 AI 数据流提供结构化日志系统。日志按节点捕获为结构化 JSONL 文件，转发到协调器进行实时流式传输，并可选择性地通过数据流图作为数据消息路由。

Which Logging Approach Should I Use?

Start here if you’re unsure which approach fits your use case.

Language-Specific Quick Start

Python – the simplest path is Python’s built-in logging module:

import logging
from adora import Node

node = Node()  # Automatically bridges Python logging -> adora

logging.info("Sensor started")       # Captured as structured "info" log
logging.warning("High temp: 42C")    # Captured as structured "warn" log
print("raw debug output")            # Captured as "stdout" level

When Node() is created, it installs a handler that routes all Python logging calls through Rust’s tracing system. The daemon parses these as structured log entries with level, message, file, and line number. No extra configuration needed.

You can also use the explicit API for structured fields:

node.log_info("Reading acquired")
node.log("info", "Reading acquired", fields={"sensor_id": "temp-01"})

Rust – use the node API convenience methods:

#![allow(unused)]
fn main() {
let (node, mut events) = AdoraNode::init_from_env()?;

// Convenience methods (recommended for most cases)
node.log_info("Sensor started");
node.log_warn("High temperature");

// With structured fields
let mut fields = BTreeMap::new();
fields.insert("sensor_id".into(), "temp-01".into());
node.log_with_fields("info", "Reading acquired", None, Some(&fields));
}

Alternatively, Rust nodes can use the tracing crate. When adora’s tracing subscriber is initialized (via init_tracing()), tracing::info!() etc. output structured JSON to stdout, which the daemon parses automatically:

#![allow(unused)]
fn main() {
// Also works -- parsed as structured logs by the daemon
tracing::info!("Sensor started");
tracing::warn!(sensor_id = "temp-01", "High temperature");
}

Use node.log_*() when you want explicit control over the log format. Use tracing::*!() when you want ecosystem integration (spans, instrumentation, OpenTelemetry). Both produce identical structured log entries in the daemon.

C – use the adora_log() function:

adora_log(ctx, "info", 4, "Sensor started", 14);

C++ – use the log_message() function:

log_message(node.send_output, "info", "Sensor started");

功能一览

日志文件格式

每个节点在以下路径生成一个 JSONL 文件（每行一个 JSON 对象）：

<working_dir>/out/<dataflow_uuid>/log_<node_id>.jsonl

每行具有以下结构：

{
  "timestamp": "2024-01-15T10:30:00.123Z",
  "level": "info",
  "node_id": "sensor",
  "message": "Starting sensor...",
  "target": "sensor::module",
  "fields": { "key": "value" }
}

节点输出如何变为日志条目

守护进程捕获节点进程的每行 stdout/stderr 并尝试将其解析为结构化日志消息（包含 level、message、timestamp 和可选 fields 的 JSON）。解析成功时保留结构化字段。解析失败时，原始行变为 "stdout"-level 条目。

这意味着使用 Rust 的 tracing 或 log crate 并输出 JSON 的节点会自动获得完整的结构化日志。仅使用 println! 的节点则产生 "stdout"-level 条目。

查看日志：adora run

使用 adora run 运行数据流时，所有节点的日志会在终端上实时显示。

Flags

adora run dataflow.yml [OPTIONS]

日志级别

从最详细到最简洁：

设置 --log-level info 会隐藏 stdout、trace 和 debug 消息。stdout 级别是一个特殊的全通级别，放行所有内容。

级别过滤逻辑

级别过滤使用 LogLevelOrStdout::passes()：

Message level    Filter level    Displayed?
─────────────    ────────────    ──────────
stdout           stdout          yes
stdout           info            no       (stdout only passes stdout filter)
info             stdout          yes      (any log level passes stdout filter)
debug            info            no       (debug is more verbose than info)
error            info            yes      (error is less verbose than info)

按节点覆盖

--log-filter 标志允许您为不同节点设置不同级别：

adora run dataflow.yml --log-level info --log-filter "sensor=debug,planner=warn"

这会为所有节点显示 info 及以上级别，但 sensor（显示 debug 及以上）和 planner（显示 warn 及以上）除外。

格式："node1=level,node2=level"（逗号分隔的 name=level 对）。

输出格式

Pretty（默认）– 彩色、人类可读：

10:30:00 INFO   sensor: Starting sensor...

10:30:01 INFO   [adora]: spawning node processor

10:30:01 stdout sensor: raw output line

• 本地时区的时间戳（HH:MM:SS）
• 级别着色：ERROR（红色）、WARN（黄色）、INFO（绿色）、DEBUG（蓝色）、TRACE（暗色）、stdout（斜体暗蓝）
• 节点名称加粗并根据名称显示唯一颜色
• 系统消息以 [adora] 为前缀
• 生命周期消息（spawning、node finished、stopping）通过空行进行视觉分隔

Json – 完整的 LogMessage 结构体输出为 JSON，每行一条：

{"build_id":null,"dataflow_id":"abc-123","node_id":"sensor","level":"INFO","message":"Starting...","timestamp":"2024-01-15T10:30:00Z",...}

适用于通过管道传递给 jq 或导入日志聚合系统。

Compact – 精简，无色彩：

10:30:00 INFO sensor: Starting sensor...

适用于 CI/CD 环境和日志文件。

查看日志：adora logs

读取历史日志或从运行中的数据流实时流式传输日志。

基本用法

# Read logs for a specific node (via coordinator)
adora logs <dataflow_uuid> <node_name>

# Read local log files directly
adora logs --local <node_name>
adora logs --local --all-nodes

# Stream live logs
adora logs <dataflow_uuid> <node_name> --follow
adora logs --local <node_name> --follow

Flags

时间过滤

--since 和 --until 接受相对于当前时间的持续时间字符串：

# Logs from the last 5 minutes
adora logs --local sensor --since 5m

# Logs from 1 hour ago to 30 minutes ago
adora logs --local sensor --since 1h --until 30m

# Last 10 errors from the past hour
adora logs --local sensor --since 1h --level error --tail 10

支持的时间格式：30（秒）、30s、5m、1h、2d。

文本搜索

--grep 对以下内容执行不区分大小写的子串匹配：

• 日志消息文本
• 节点 ID
• 模块目标

# Find all timeout-related messages
adora logs --local --all-nodes --grep "timeout"

# Find errors from a specific module
adora logs --local sensor --grep "camera::driver" --level error

过滤流水线

所有过滤器按以下顺序应用：

Read/Parse -> Time Filters -> Grep -> Tail -> Display

在协调器模式下使用 --since、--until 或 --grep 时，CLI 从服务器获取所有日志（忽略服务端的 --tail）并在客户端应用所有过滤器。这确保了组合使用过滤器时结果的正确性。

本地模式 vs 协调器模式

本地模式（--local）直接从当前工作目录的 out/ 目录读取 JSONL 文件。无需运行协调器或守护进程。如果使用 --all-nodes 或未指定节点名称，所有日志文件将合并并按时间戳排序。

协调器模式（默认）通过 WebSocket 连接到运行中的协调器。协调器从守护进程的工作目录读取日志文件并将其回传。适用于本地和分布式部署。

跟踪模式

本地跟踪（--local --follow）：每 200ms 轮询日志文件获取新内容。新行被解析、经 --grep 过滤后打印。时间/尾部过滤器仅应用于初始历史输出。

协调器跟踪（--follow）：向协调器打开 WebSocket 订阅。协调器实时转发守护进程的日志消息。级别过滤在服务端应用以提高效率。--grep 和 --since 在客户端对流应用。

环境变量

所有环境变量作为备选值 – CLI 标志始终优先。

Example:

# Set defaults for a development session
export ADORA_LOG_LEVEL=info
export ADORA_LOG_FORMAT=pretty
export ADORA_LOG_FILTER="sensor=debug"

# These are equivalent:
adora run dataflow.yml
adora run dataflow.yml --log-level info --log-format pretty --log-filter "sensor=debug"

# CLI flag overrides env var:
adora run dataflow.yml --log-level debug   # overrides ADORA_LOG_LEVEL=info

YAML 配置

min_log_level

在日志到达日志文件、协调器或 send_logs_as 路由之前，在源端（守护进程端）过滤日志。

nodes:
  - id: noisy-sensor
    path: ./target/debug/sensor
    min_log_level: info    # 抑制此节点的 debug/trace/stdout

有效值：error、warn、info、debug、trace、stdout。

设置后，守护进程在解析后立即丢弃低于此级别的日志消息。这减少了磁盘 I/O、网络流量和日志文件大小。过滤使用与 CLI 显示过滤器相同的 passes() 逻辑。

send_stdout_as

将原始 stdout/stderr 行路由为数据流输出消息。

nodes:
  - id: legacy-node
    path: ./legacy-script.py
    send_stdout_as: raw_output
    outputs:
      - raw_output
      - data

  - id: log-consumer
    inputs:
      logs: legacy-node/raw_output

每行 stdout/stderr 作为 Arrow 编码的字符串发送。这对于集成在 stdout 上输出数据的旧节点很有用（例如使用 print() 的 Python 脚本）。

send_stdout_as 和正常日志文件写入同时进行 – stdout 路由不会抑制日志文件。

send_logs_as

将解析的结构化日志条目路由为数据流输出消息。

nodes:
  - id: sensor
    path: ./target/debug/sensor
    send_logs_as: log_entries
    outputs:
      - data
      - log_entries

  - id: log-aggregator
    inputs:
      sensor_logs: sensor/log_entries

与 send_stdout_as 不同，这仅发送成功解析为结构化日志的行（非原始 stdout）。每个条目序列化为完整的 JSON LogMessage 字符串。min_log_level 过滤器在路由前应用 – 被抑制的消息不会发送。

使用此功能在数据流内部构建日志聚合、告警或监控节点。

adora/logs – Automatic Log Aggregation

Subscribe to logs from all nodes with a single input line – no manual wiring needed:

nodes:
  - id: sensor
    path: sensor.py
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - reading

  - id: processor
    path: processor.py
    inputs:
      reading: sensor/reading
    outputs:
      - result

  - id: log-viewer
    path: log_viewer.py
    inputs:
      logs: adora/logs              # all nodes, all levels
      errors: adora/logs/error      # only error+ from all nodes
      sensor: adora/logs/info/sensor  # info+ from one node

The adora/logs virtual input works like adora/timer – the daemon handles subscription internally. Each log message arrives as a JSON-encoded LogMessage string in an Arrow array. To prevent infinite loops, a node never receives its own log messages.

Syntax:

Levels: stdout, error, warn, info, debug, trace.

When to use adora/logs vs send_logs_as:

See examples/log-aggregator/ for a complete working example.

max_log_size

启用基于大小的日志文件轮转。

nodes:
  - id: sensor
    path: ./target/debug/sensor
    max_log_size: "50MB"

当活动日志文件超过配置大小时，守护进程会：

1. 刷新并关闭当前文件
2. 重命名现有轮转文件：.4.jsonl -> .5.jsonl、.3.jsonl -> .4.jsonl 等
3. 重命名当前文件：log_sensor.jsonl -> log_sensor.1.jsonl
4. 创建新的 log_sensor.jsonl
5. 删除超出轮转限制的文件（默认 5，可通过 max_rotated_files 配置）

命名约定：

log_sensor.jsonl       # current (active)
log_sensor.1.jsonl     # previous
log_sensor.2.jsonl     # older
log_sensor.3.jsonl
log_sensor.4.jsonl
log_sensor.5.jsonl     # oldest (deleted on next rotation)

每个节点的最大磁盘使用量：max_log_size * (1 + max_rotated_files)（1 个活动文件 + N 个轮转文件）。

没有 max_log_size 时，日志文件会无限增长。对于长时间运行的数据流，请始终设置此项。

adora logs --local 命令自动读取节点的所有轮转文件并按时间顺序合并（最旧的轮转文件在前，当前文件在后）。

max_rotated_files

控制保留多少个轮转日志文件（默认：5，范围：1-100）。

nodes:
  - id: sensor
    path: ./target/debug/sensor
    max_log_size: "50MB"
    max_rotated_files: 10    # keep 10 rotated files instead of 5

当 max_rotated_files: 10 且 max_log_size: "50MB" 时，每个节点最大磁盘使用量为 50MB * 11 = 550MB。较低的值节省磁盘空间；较高的值保留更多历史记录。

运行时节点限制

对于运行时节点（算子），每个运行时只允许每个日志字段有一个：

# OK -- single operator
nodes:
  - id: runtime-node
    operator:
      python: process.py
      send_logs_as: logs
      min_log_level: info
      max_log_size: "100MB"

# ERROR -- 多个算子具有冲突的配置
nodes:
  - id: runtime-node
    operators:
      - id: op1
        python: a.py
        send_logs_as: logs1
      - id: op2
        python: b.py
        send_logs_as: logs2    # Error: multiple send_logs_as

当运行时中的单个算子设置这些字段时，输出名称以算子 ID 为前缀（例如 op1/logs）。

节点日志 API

节点可以使用节点 API 以编程方式发出结构化日志消息。这等同于将 JSON 格式的日志行写入 stdout – 守护进程以相同方式解析。

Rust

#![allow(unused)]
fn main() {
use adora_node_api::AdoraNode;
use std::collections::BTreeMap;

let (node, mut events) = AdoraNode::init_from_env()?;

// 指定级别字符串和可选目标的通用日志
node.log("info", "sensor initialized", Some("sensor::init"));

// 便捷方法（无 target 参数）
node.log_error("connection failed");
node.log_warn("temperature elevated");
node.log_info("reading acquired");
node.log_debug("raw bytes received");
node.log_trace("entering loop iteration");

// 结构化字段（键值上下文通过 send_logs_as 保留）
let mut fields = BTreeMap::new();
fields.insert("sensor_id".to_string(), "temp-01".to_string());
fields.insert("reading".to_string(), "42.5".to_string());
node.log_with_fields("info", "reading acquired", None, Some(&fields));
}

level 参数接受 "error"、"warn"（或 "warning"）、"info"、"debug"、"trace"。未知级别默认为 "info"。字段总量上限为 60 KB，以匹配下游 64 KB 的解析限制。

Python

Python nodes have three ways to log, all producing structured log entries:

from adora import Node
import logging

node = Node()

# Option 1: Python's logging module (recommended -- auto-bridged by Node())
logging.info("sensor initialized")
logging.warning("temperature elevated")
logging.debug("raw bytes: %s", data)

# Option 2: Explicit adora API with level string
node.log("info", "sensor initialized", target="sensor.init")
node.log("info", "reading acquired", fields={"sensor_id": "temp-01", "reading": "42.5"})

# Option 3: Convenience methods
node.log_error("connection failed")
node.log_warn("temperature elevated")
node.log_info("reading acquired")
node.log_debug("raw bytes received")
node.log_trace("entering loop iteration")

# This also works but produces "stdout"-level entries (no structure):
print("raw output")

How the Python logging bridge works: When Node() is created, it installs a custom logging.Handler that routes all Python logging calls through Rust’s tracing system. The daemon parses these as structured log entries with level, message, file path, and line number. This happens automatically – no configuration needed.

Common pitfall: Do not call logging.basicConfig() before creating Node(). The node constructor sets up the logging bridge; calling basicConfig() first may install a conflicting handler. If you need custom formatters, configure them after Node() creation.

C

#include "node_api.h"

void *ctx = init_adora_context_from_env();
const char *level = "info";
const char *msg = "sensor initialized";
adora_log(ctx, level, strlen(level), msg, strlen(msg));

C++

// 通过 cxx bridge
auto node = init_adora_node();
log_message(node.send_output, "info", "sensor initialized");

日志工具库（adora-log-utils）

adora-log-utils crate 提供解析、合并、过滤和格式化工具，用于在自定义 sink 节点中处理 LogMessage 条目。在构建通过 send_logs_as 消费日志数据的节点时使用。

API

#![allow(unused)]
fn main() {
use adora_log_utils;

// 从 JSON 解析 LogMessage（从 send_logs_as 接收）
let log = adora_log_utils::parse_log(json_str)?;

// 直接从 Arrow 输入数据解析（事件处理器的便捷方法）
let log = adora_log_utils::parse_log_from_arrow(&data)?;

// 将多个日志流合并为单一时间线
let merged = adora_log_utils::merge_by_timestamp(vec![stream_a, stream_b]);

// 按最低级别过滤
let errors = adora_log_utils::filter_by_level(&logs, &min_level);

// 格式化为 JSON（单行，无尾部换行）
let json = adora_log_utils::format_json(&log);

// 格式化为紧凑单行："<timestamp> <node> <LEVEL>: <message>"
let compact = adora_log_utils::format_compact(&log);

// 格式化为美观输出："[<timestamp>][<LEVEL>][<node>] <message>"
let pretty = adora_log_utils::format_pretty(&log);
}

Dependency

添加到 sink 节点的 Cargo.toml：

[dependencies]
adora-log-utils = { workspace = true }

日志 Sink 示例

三个示例 sink 节点演示如何消费通过 send_logs_as 路由的日志并转发到外部目的地。

文件 Sink（examples/log-sink-file/）

将多个节点的日志流合并为单个 JSONL 文件。适用于统一日志收集。

nodes:
  - id: sensor
    path: sensor.py
    send_logs_as: log_entries
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - reading
      - log_entries

  - id: processor
    path: processor.py
    send_logs_as: log_entries
    inputs:
      reading: sensor/reading
    outputs:
      - result
      - log_entries

  - id: file_sink
    path: log-sink-file
    inputs:
      sensor_logs: sensor/log_entries
      processor_logs: processor/log_entries
    env:
      LOG_FILE: "./combined.jsonl"

文件 sink 从环境变量读取 LOG_FILE（默认 ./combined.jsonl），使用 adora_log_utils::parse_log_from_arrow() 解析每条传入的 Arrow 消息，格式化为 JSON 并追加到文件。

TCP Sink（examples/log-sink-tcp/）

通过 TCP 套接字将日志条目转发到远程日志收集器。适用于缺少本地文件系统且需要将日志流式传输到设备外的嵌入式系统。

nodes:
  - id: source
    path: source.py
    send_logs_as: log_entries
    inputs:
      tick: adora/timer/millis/500
    outputs:
      - data
      - log_entries

  - id: tcp_sink
    path: log-sink-tcp
    inputs:
      logs: source/log_entries
    env:
      SINK_ADDR: "127.0.0.1:9876"

TCP sink 从环境变量读取 SINK_ADDR（默认 127.0.0.1:9876），启动时连接到服务器，并将每条日志条目作为 JSON 行发送。写入失败时自动重连。

告警路由器（examples/log-sink-alert/）

按严重程度拆分传入的日志条目。所有日志转发到 all_logs 输出；仅 error 和 warn 日志转发到 alerts 输出。这使下游节点能够差异化处理告警（例如触发通知、写入专用文件）。

nodes:
  - id: source
    path: my_node.py
    send_stdout_as: log_entries
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - log_entries

  - id: alert_router
    path: log-sink-alert
    inputs:
      logs: source/log_entries
    outputs:
      - all_logs
      - alerts

源节点使用 send_stdout_as 将其 stdout 行路由为 Arrow 字符串数据。路由器使用 adora_log_utils::parse_log_from_arrow() 解析每条日志条目，检查级别，并使用 node.send_output() 将数据转发到相应输出。使用节点 API 的节点也可以使用 send_logs_as 来路由 node.log() 的结构化日志。

构建自定义 Sink

要构建自己的 sink 节点，请遵循以下模式：

use adora_node_api::{AdoraNode, Event};

fn main() -> eyre::Result<()> {
    let (_node, mut events) = AdoraNode::init_from_env()?;

    while let Some(event) = events.recv() {
        match event {
            Event::Input { data, .. } => {
                let log = adora_log_utils::parse_log_from_arrow(&data)?;
                // 处理日志条目：写入文件、通过网络发送等
                let json = adora_log_utils::format_json(&log);
                println!("{json}");
            }
            Event::Stop(_) => break,
            _ => {}
        }
    }
    Ok(())
}

守护进程如何处理日志

理解内部流水线有助于调试和调优。守护进程为每个节点运行一个专用的异步任务，按顺序处理日志行：

Node Process (stdout/stderr)
    |
    v
[1] Capture: lines buffered in mpsc channel (capacity 100)
    |
    v
[2] send_stdout_as: raw line -> Arrow data -> dataflow output
    |
    v
[3] Parse: try JSON structured log, fall back to Stdout-level
    |
    v
[4] min_log_level filter: drop messages below threshold
    |
    v
[5] send_logs_as: LogMessage -> JSON -> Arrow data -> dataflow output
    |
    v
[6] Write JSONL: compact format to log file, track bytes written
    |
    v
[7] Rotation check: if bytes_written >= max_log_size, rotate files
    |
    v
[8] Forward: send LogMessage to display channel (unless ADORA_QUIET)
    |
    v
[9] Sync: fsync log file to disk

关键细节：

• 步骤 2 在解析之前发生，因此 send_stdout_as 捕获每一行包括非结构化输出
• 步骤 4 在步骤 5-8 之前发生，因此 min_log_level 会抑制所有下游处理的消息
• 步骤 5 仅对成功解析的结构化日志触发（步骤 3 成功路径）
• 步骤 8 发送到 flume 通道（adora run 直接模式）或协调器（分布式模式）
• 步骤 9 每次写入后调用 sync_all()，以一定的 I/O 开销为代价确保持久性

结构化日志解析

当节点发出 JSON 格式的日志输出（例如使用 JSON 格式化的 tracing-subscriber）时，守护进程提取：

• level：日志严重程度
• message：日志文本
• target：模块路径
• timestamp：日志发出时间
• fields：任意键值对
• build_id、dataflow_id、node_id、daemon_id：作为后备从字段中提取

守护进程还会在所有消息上设置 dataflow_id、node_id 和 daemon_id，以确保它们始终存在于日志文件中。

协调器日志流协议

当守护进程在协调器下运行（分布式模式）时，日志转发通过 WebSocket 工作：

1. 守护进程 -> 协调器：每条 LogMessage 被封装在 DaemonEvent::Log(message) 中并通过守护进程的 WebSocket 连接发送
2. 协调器存储：协调器存储/转发日志
3. CLI 订阅：CLI 通过其 WebSocket 连接发送 ControlRequest::LogSubscribe { dataflow_id, level }
4. 服务端过滤：协调器仅转发 msg_level <= subscription_level 的消息。这减少了过滤订阅的网络流量
5. CLI 接收：消息作为序列化的 LogMessage 结构体到达

--level 标志映射到 log::LevelFilter：

• stdout -> LevelFilter::Trace（最宽松，接收所有内容）
• info -> LevelFilter::Info（接收 Error、Warn、Info）
• etc.

完整 YAML 参考

nodes:
  - id: sensor
    path: ./target/debug/sensor
    outputs:
      - data
      - raw_output       # 用于 send_stdout_as
      - log_entries       # 用于 send_logs_as

    # Source-level log filtering (daemon-side)
    min_log_level: info          # 抑制 debug/trace/stdout

    # Route stdout to dataflow
    send_stdout_as: raw_output   # 每行 stdout 变为数据消息

    # Route structured logs to dataflow
    send_logs_as: log_entries    # 解析的日志条目变为数据消息

    # Log file rotation
    max_log_size: "50MB"         # 文件超过 50MB 时轮转
    max_rotated_files: 5         # keep 5 rotated files (default, range 1-100)

    inputs:
      tick: adora/timer/millis/100

完整示例

examples/python-logging/ 目录包含一个可运行的三节点流水线，演练了每个日志功能：

sensor (noisy, high-volume) --> processor (structured logs) --> monitor (log aggregator)

数据流配置要点：

nodes:
  - id: sensor
    path: sensor.py
    min_log_level: info       # 在源端抑制 debug 噪声
    max_log_size: "1KB"       # 演示用小值（快速触发轮转）
    inputs:
      tick: adora/timer/millis/50
    outputs:
      - reading

  - id: processor
    path: processor.py
    send_logs_as: log_entries  # 将结构化日志路由为数据
    inputs:
      reading: sensor/reading
    outputs:
      - result
      - log_entries

  - id: monitor
    path: monitor.py
    inputs:
      logs: processor/log_entries
      reading: sensor/reading

每个节点演示的内容：

• sensor – 混合使用 print()（原始 stdout）、logging.info()、logging.debug() 和 logging.warning()。设置 min_log_level: info 后，debug 消息在到达日志文件前被守护进程丢弃。设置 max_log_size: "1KB" 后，几秒钟后日志轮转开始。
• processor – 使用 send_logs_as: log_entries 将其结构化日志条目路由为数据流数据。原始 print() 输出_不_会被路由（仅路由解析后的结构化条目）。
• monitor – 订阅 processor/log_entries 并统计警告/错误，演示数据流内日志聚合。

直接模式（adora run – 单进程，适合快速测试）：

# Basic run
adora run examples/python-logging/dataflow.yml --stop-after 5s

# Only warnings and above
adora run examples/python-logging/dataflow.yml --log-level warn --stop-after 5s

# Per-node overrides
adora run examples/python-logging/dataflow.yml --log-filter "monitor=debug,sensor=warn" --stop-after 5s

# JSON output for machine parsing
adora run examples/python-logging/dataflow.yml --log-format json --stop-after 3s

# Environment variable control
ADORA_LOG_LEVEL=warn adora run examples/python-logging/dataflow.yml --stop-after 5s

分布式模式（adora up + adora start – 协调器/守护进程架构，多机部署必需）：

# Start infrastructure
adora up

# Start attached (live log stream)
adora start examples/python-logging/dataflow.yml --attach

# Or start detached and query logs separately
adora start examples/python-logging/dataflow.yml
adora logs <dataflow-id> sensor --follow                    # stream one node
adora logs <dataflow-id> sensor --follow --level warn       # only warnings
adora logs <dataflow-id> --all-nodes --tail 20              # last 20 lines
adora logs <dataflow-id> processor --grep "error" --since 5m  # targeted search

在分布式模式下，日志流经 节点 -> 守护进程 -> 协调器 -> CLI（通过 WebSocket）。协调器缓冲日志消息直到订阅者连接，因此即使延迟附加也不会丢失日志。YAML 级别设置（min_log_level、send_logs_as、max_log_size）工作方式相同，因为它们在守护进程端应用。