Core Concepts

Understanding the computation graph model that powers tflo.

The Computation Graph

At the heart of tflo is a declarative computation graph. Instead of imperatively looping over data and manually tracking window state, you describe what to compute, and tflo handles how.

The graph is built at runtime and compiled into an optimized execution plan. It manages:

• Window state — ring buffers, running sums, etc.
• Timestamp handling — duration-based windows vs. count-based.
• Lazy evaluation — only compute what’s needed for output.

Properties & Timestamps

Every computation graph starts with two fundamental declarations:

Timestamp

t.timestamp(|tick| tick.ts)

Timestamps are i64 and must be monotonically non-decreasing for correct window behavior. They’re used to determine which values fall within a duration window.

Properties

let price = t.prop(|tick| tick.price)

Properties extract f64 values from your input type. You can define as many properties as you need, and each property has its own independent window state.

Window Kinds

tflo supports two windowing strategies:

Duration windows are ideal for time-series data where intervals may be irregular. Count windows are simpler and faster when you just want "last N elements."

Keyed Execution

When events arrive from many emitters — sensors, hosts, instruments, or any other partition — tflo can split execution by key. Each key gets its own independent window state:

use tflo_core::prelude::*;
use tflo_ops::prelude::*;

events.into_iter()
    .tflo_keyed(
        |e| e.source.clone(),              // partition by event source
        OutOfOrderPolicy::Error,            // out-of-order handling policy
        |t| {                               // computation per key
            t.timestamp(|x| x.ts);
            let value = t.prop(|x| x.value);
            value.ema(20_u64.secs())
        },
    )
// See tflo-core/src/keyed.rs for the full keyed execution API.

Full runnable example: tflo-examples/examples/docs-quick-start

Internally, tflo maintains a HashMap<Key, KeyedGraphState> and routes each input record to the correct partition before computing.

Iterator vs Stream

tflo works with both Iterator and Stream (via tflo-core's async feature):

Iterator (synchronous)

let results = my_data.into_iter()
    .tflo(|t| { ... })
    .collect::<Vec<_>>();

Full runnable example: tflo-examples/examples/docs-quick-start

Stream (async)

// Requires tflo-core with "async" feature: cargo add tflo-core --features async
use tflo_core::prelude::*;
use tflo_ops::prelude::*;

let results: Vec<f64> = my_stream
    .tflo(|t| {
        t.timestamp(|x| x.ts);
        let price = t.prop(|x| x.price);
        price.sma(5_u64.secs())
    })
    .collect()
    .await;
// See tflo_core::r#async for the TFloStreamExt trait.

Full runnable example (async): tflo-examples/examples/docs-advanced

The async adapter integrates with Tokio and supports backpressure. It’s the backbone for integrations like Kafka.

Typed Absent

tflo replaces NaN-as-sentinel with an explicit Result<f64, Absent>. Every reason a value can be missing is named: the window hasn't warmed up, a division would have produced infinity, a policy filter excluded the record, an operator rejected the domain. Downstream code branches on the reason, not on a single poisoned float.

// Computed is the engine-side output type for every node.
pub type Computed = Result<f64, Absent>;

pub enum Absent {
    WarmingUp,      // window has not yet collected enough samples
    DivideByZero,   // a division would have produced an infinity
    FilteredOut,    // a CEL / Rhai / Rego policy excluded this record
    DomainError,    // operator rejected the input (e.g. log of negative)
}

// Downstream code branches on the reason, not on a single poisoned float.
match value {
    Ok(v) => publish(v),
    Err(Absent::WarmingUp)   => skip_until_warm(),
    Err(Absent::DivideByZero) => alert(),
    Err(reason) => log_with_context(reason),
}

This is the typed-absence model the engine is built around — it's why processing-time mode is intentionally not supported (see non-goals): the typed semantics depend on deterministic, event-time-driven evaluation.

Timers

Event-time timers with per-key heap ordering. Schedule a fire at an arbitrary future timestamp; the engine drains due timers in (fire_ts, seq) order so ties fire deterministically. State is serialized via postcard alongside the rest of the snapshot, so timers survive checkpoint and restore.

// Per-key, event-time, snapshot-safe. Heap ordering is (fire_ts, seq)
// for deterministic firing on ties.
use tflo_core::timer::TimerService;

// Inside a per-key operator: schedule a fire 10 s into the future.
timers.schedule(fire_ts = ts + 10_000, payload = MyTimer::DwellOver);

// Engine drains due timers in order and feeds them back through the graph
// alongside the regular event stream.

Timers are the substrate the existing detectors use for "within T" semantics — pulse-width registers a TooLong timer on the rising edge and cancels it on the falling edge. They're the cleanest way to express dwell timers, watchdog timeouts, and absence detection in a snapshot-safe way, and the natural substrate for any sequence-pattern detector you compose from primitives.

Session-tumbling windows

A third window kind alongside duration and count: gap-based session boundaries with a max-length tumble. The session closes either when the inactivity gap is exceeded or when the session has run for max_session — useful for sensor bursts, user sessions, or any "things that arrive together" partitioning where you need a fallback bound.

// Session-tumbling window: gap-based session boundaries, with a
// max-length tumble so very long sessions still emit.
use tflo_ops::prelude::*;

events.into_iter()
    .tflo(|t| {
        t.timestamp(|e| e.ts);
        let value = t.prop(|e| e.value);
        value.session_tumbling(
            inactivity_gap = 30_u64.secs(),
            max_session    = 5_u64.mins(),
        )
    })
    .collect::<Vec<_>>();

Keyed execution and ShardRouter

Keyed execution stays the same surface you saw above — .tflo_keyed(key_fn, policy, builder). What's new is that the engine now knows which keys it owns via a pluggable ShardRouter<K> trait. The default is LocalShard (owns every key); in a sharded cluster KafkaShardRouter bridges this to consumer-group rebalance.

Snapshots are stamped with a topology fingerprint via Builder::fingerprint(). On restore, a fingerprint mismatch is rejected — you can't accidentally hydrate state produced by a different topology.

// Topology hash stamped into every snapshot.
let fp = builder.fingerprint();   // [u8; 32]

// On restore, fingerprint mismatch is rejected — no silently loading state
// that was produced by a different topology.
graph.restore(&loaded)?;          // returns Err if fingerprints differ

See Contracts for the trait sketches, and the reference deployment for end-to-end use.

Pattern matching

Single-condition detectors (cross, hysteresis, glitch, runt, pulse-width) compose readily for "this event meets this condition" shapes. For multi-event sequences — "A then B within T", "A then no B within T", "three identical events within T" — the tflo-cep crate provides a closure-based builder over the same engine substrate (timers, snapshot-safe state, typed signals). The tflo-cep-wasm sibling exposes the same engine to JavaScript through WebAssembly — one state machine, two idiomatic surfaces.

See event patterns for the API and worked examples in both Rust and TypeScript, and browser analytics for the end-to-end browser workflow.