EnerOS v0.14.0

Release Date: January 26, 2025 Codename: Pipeline Git Tag: v0.14.0 Support Status: Stable Total Crates: 34 (4 new) Test Cases: 4680+ (560 new)

Version Overview

EnerOS v0.14.0 “Pipeline” is the core release marking EnerOS’s entry into the “Event-Driven” phase. The primary goal of this version is to introduce a Realtime Data Pipeline and Event-Driven Architecture, enabling massive volumes of telemetry data, equipment events, topology changes, and Agent decisions in the power grid to flow efficiently in Pub/Sub mode, with real-time aggregation and windowed analysis via stream processing.

In traditional power systems, data flow typically follows a “collect → write to database → query” batch processing model, with latencies in the seconds-to-minutes range—incapable of supporting sub-second response scenarios such as fast fault localization or real-time load balancing. Additionally, collaboration between Agents relies on point-to-point calls, leading to tight coupling and poor scalability. v0.14.0 leverages kernel-builtin message buses and stream processing engines to reduce data flow latency to millisecond level, decoupling Agent collaboration into an event-driven model, with a single cluster supporting throughputs of 1,000,000+ events per second.

This version introduces four new crates: eneros-pipeline (pipeline core), eneros-pipeline-mq (message queue integration), eneros-pipeline-stream (stream processing), and eneros-pipeline-window (data windows). The design philosophy is “grid as data flow, events as collaboration language”—power system operation is essentially continuous data flows and discrete event flows; modeling them as a pipeline allows Agents to perceive and respond to grid changes in a unified manner.

Key Metrics

New Features

1. Realtime Data Pipeline (Pub/Sub)

Introduced the eneros-pipeline crate, providing a kernel-level Pub/Sub message bus. All EnerOS components (topology engine, digital twin, Agents, time series storage, API gateway) can act as publishers or subscribers to the pipeline. The pipeline uses a lock-free multicast architecture, with single-message fan-out to 1000 subscribers at latency below 5μs.

Pipeline Basics

use eneros_pipeline::{Pipeline, Topic, Event};

let pipeline = Pipeline::new(PipelineConfig {
    max_topics: 4096,
    max_subscribers_per_topic: 1024,
    retention: Duration::minutes(5),
    backlog_size: 1_000_000,
})?;

// Subscribe to telemetry topic
let mut sub = pipeline.subscribe(Topic::of("telemetry.bus.*"))?;

// Publish telemetry event
pipeline.publish(Topic::of("telemetry.bus.1"), Event::json({
    "voltage": 1.024,
    "current": 512.3,
    "timestamp": now(),
}))?;

// Consume events
while let Some(event) = sub.next().await {
    println!("Received: {:?}", event.payload);
}

Built-in Topic Categories

2. Event-Driven Architecture

All EnerOS components support event-driven collaboration. Agents no longer need to actively poll grid state; instead, they subscribe to topics of interest and are passively triggered to make decisions when events arrive. This pattern significantly reduces Agent CPU usage and naturally decouples collaboration relationships.

Event-Driven Agent

use eneros_pipeline::EventHandler;

// An event-driven voltage monitoring Agent
Agent::new("voltage-monitor", tenant)
    .subscribe("telemetry.bus.*")
    .handler(EventHandler::new(|event, ctx| async move {
        let voltage: f64 = event.payload["voltage"]?;
        
        if voltage < 0.95 {
            // Publish alarm event
            ctx.pipeline().publish(
                Topic::of("alarm.voltage.low"),
                Event::json({
                    "bus": event.topic.suffix(),
                    "voltage": voltage,
                    "severity": "warning",
                }),
            )?;
        }
        Ok(())
    }))
    .spawn()?;

Event Sourcing

Critical operations support Event Sourcing, where all state changes are stored as immutable event sequences, allowing replay to reconstruct state at any point in time:

use eneros_pipeline::event_sourcing;

// Reconstruct topology state from event sequence
let mut reconstructor = event_sourcing::TopologyReconstructor::new();
let events = pipeline.replay("topology.*")
    .from("2025-01-26T00:00:00Z")
    .to("2025-01-26T12:00:00Z")
    .fetch()?;

for event in events {
    reconstructor.apply(event)?;
}
let topology = reconstructor.finalize();

3. Message Queue Integration

Introduced the eneros-pipeline-mq crate, supporting bidirectional bridging between EnerOS pipelines and external message queue systems for easy integration with existing IT infrastructure. All integrations use exactly-once semantics to prevent data loss or duplication.

Supported Message Queues

Kafka Integration Example

use eneros_pipeline_mq::kafka::{KafkaBridge, KafkaConfig};

// Create Kafka bridge
let bridge = KafkaBridge::new(KafkaConfig {
    brokers: vec!["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
    consumer_group: "eneros-consumer",
    exactly_once: true,
    schema_registry: "http://schema-registry:8081",
})?;

// Bridge EnerOS pipeline to Kafka
bridge.forward(
    Topic::of("telemetry.*"),
    KafkaTopic::of("eneros-telemetry"),
).await?;

// Bridge Kafka back to EnerOS pipeline
bridge.reverse(
    KafkaTopic::of("external-events"),
    Topic::of("external.*"),
).await?;

4. Stream Processing

Introduced the eneros-pipeline-stream crate, providing stream processing capabilities on top of the pipeline, supporting operations such as filter, map, aggregate, and join. All computations execute continuously in memory with latency below 10ms.

Stream Processing Example

use eneros_pipeline_stream::{Stream, StreamExt};

// Create a telemetry data stream
let stream = Stream::from_topic("telemetry.bus.*", &pipeline);

// Calculate 1-minute average voltage for each bus
let avg_voltage = stream
    .filter(|e| e.payload["voltage"].is_number())
    .window(Window::tumbling(Duration::minutes(1)))
    .group_by(|e| e.topic.suffix().to_string())
    .aggregate(AggFn::avg(|e| e.payload["voltage"].as_f64()))
    .for_each(|(bus, avg)| async move {
        println!("Bus {} 1-minute average voltage: {:.3} pu", bus, avg);
    })
    .run()
    .await?;

Stream JOIN

// JOIN telemetry stream with topology change stream
let telemetry = Stream::from_topic("telemetry.*", &pipeline);
let topology = Stream::from_topic("topology.change", &pipeline);

telemetry.join(topology)
    .on(|t, topo| t.topic.matches(&topo.payload["affected_topic"]))
    .window(Window::sliding(Duration::seconds(30), Duration::seconds(5)))
    .for_each(|(telem, topo)| async move {
        log::info!("Telemetry after topology change: {:?} <- {:?}", telem, topo);
    })
    .run().await?;

Stream Processing Operators

5. Data Windows

Introduced the eneros-pipeline-window crate, providing rich window semantics to support time-dimensional aggregation in stream processing. Supports three modes: tumbling windows, sliding windows, and session windows, and handles out-of-order data and late data.

Window Types

use eneros_pipeline_window::{Window, WindowType};

// Tumbling window: 1-minute fixed window, no overlap
let tumbling = Window::tumbling(Duration::minutes(1));

// Sliding window: 5-minute window, slides every 1 minute
let sliding = Window::sliding(
    Duration::minutes(5),
    Duration::minutes(1),
);

// Session window: 30-second activity gap
let session = Window::session(Duration::seconds(30));

// Global window: triggered by count
let count = Window::count(1000);

Out-of-Order Data Handling

use eneros_pipeline_window::{Watermark, LateDataPolicy};

let stream = Stream::from_topic("telemetry.*", &pipeline)
    .watermark(Watermark::bounded_out_of_order(Duration::seconds(5)))
    .late_data_policy(LateDataPolicy::AllowAndRecompute)
    .window(Window::tumbling(Duration::minutes(1)))
    .aggregate(AggFn::avg(|e| e.payload["voltage"].as_f64()));

Window Trigger Strategies

Improvements

• Agent Runtime: Agents default to event-driven mode, reducing CPU usage by 65%
• Time Series Storage: Write path supports pipeline subscriptions, allowing consumption upon write
• Observability: New pipeline metrics exposed, including throughput, latency, and backlog
• API Gateway: Supports SSE and WebSocket for real-time pipeline event push

Bug Fixes

• Fixed eneros-pipeline fan-out latency spike when subscriber count exceeds 256 (#1409)
• Fixed eneros-pipeline-stream window calculation error across time zones (#1415)
• Fixed eneros-pipeline-mq Kafka integration data loss during rebalance (#1421)
• Fixed eneros-pipeline-window session window memory leak during prolonged inactivity (#1428)

Breaking Changes

• Agent::run defaults to event-driven: Original polling mode migrated to Agent::run_polling
• Pipeline::subscribe return type: Changed from Iterator to Stream, requires .await
• Event::payload: Changed from String to serde_json::Value, requires deserialization adaptation

Performance Improvements

• Event throughput increased from 50K/s to 1.2M/s (24x improvement)
• End-to-end latency reduced from 200ms to 2.3ms (87x improvement)
• Stream processing latency stable within 10ms
• Average Agent CPU usage reduced by 65%

Contributors

This version was jointly completed by 26 contributors with 478 commits. Special thanks to:

• @pipeline-core: Pipeline core and lock-free multicast
• @mq-integrator: Message queue adaptation and exactly-once
• @stream-engine: Stream processing engine
• @window-master: Window semantics and watermarks

Upgrade Guide

Upgrading from v0.13.0

1. Update Dependencies

# Cargo.toml
[dependencies]
eneros-pipeline = { version = "0.14" }
eneros-pipeline-stream = { version = "0.14" }
eneros-pipeline-window = { version = "0.14" }
eneros-pipeline-mq = { version = "0.14", features = ["kafka"] }

2. Migrate Agents to Event-Driven

// v0.13.0 old approach: polling
Agent::new("monitor", tenant)
    .poll_interval(Duration::seconds(1))
    .handler(|ctx| async move {
        let state = ctx.twin().current_state()?;
        // ...
    });

// v0.14.0 new approach: event-driven
Agent::new("monitor", tenant)
    .subscribe("telemetry.*")
    .handler(|event, ctx| async move {
        let state = event.payload?;
        // ...
    });

3. Configure Message Queue Bridge

# eneros.toml
[pipeline.mq.kafka]
brokers = ["kafka-1:9092", "kafka-2:9092"]
exactly_once = true

[[pipeline.mq.bridge]]
eneros_topic = "telemetry.*"
kafka_topic = "eneros-telemetry"
direction = "forward"

4. Enable Stream Processing

For real-time aggregation scenarios, enable stream processing instead of periodic queries:

// Old: periodic time series database queries
loop {
    let avg = timeseries.query("bus.1.voltage")
        .range(now() - Duration::minutes(1), now())
        .avg()?;
    // ...
    sleep(Duration::minutes(1)).await;
}

// New: stream processing
Stream::from_topic("telemetry.bus.1", &pipeline)
    .window(Window::tumbling(Duration::minutes(1)))
    .aggregate(AggFn::avg(|e| e.payload["voltage"].as_f64()))
    .for_each(|avg| async move { /* ... */ })
    .run().await?;