Stateful Streams: Edge vs Level Triggers and OHLC Windows

A price alert that detects a threshold crossing needs the previous price. A test of only the current price can notify on every tick above the threshold. The stored state determines which behavior the user receives.

During the ticker comparison, I first studied transport, consumer groups, recovery, and replication. I expected the alert logic to be easier. It required separate decisions about state and ordering.

Transport versus processing

Stream processing can be stateless or stateful. A stateless handler reacts to the current record. A stateful handler also uses earlier records or stored values. Threshold crossing detection needs that earlier state.

The Kafka Streams library runs on the JVM. For my Node.js application, I implemented an alert evaluator and an aggregator for candle windows. Their state and recovery behavior remain application responsibilities.

The alert that fires once: edge vs level

Take a rule like "notify Alice when BBCA rises to or above 9500." You can read that two ways:

• Level: fire on every tick where the price is >= 9500. This stateless check sends another notification for each tick that satisfies the condition.
• Edge: fire only on the tick where the price crosses from below 9500 to at or above it. One notification per crossing, which is what a user wants from "notify me when."

The level test only needs the current tick:

export function matches(op: Operator, price: number, threshold: number): boolean {
  switch (op) {
    case "<": return price < threshold;
    case "<=": return price <= threshold;
    case "=": return price === threshold;
    case ">=": return price >= threshold;
    case ">": return price > threshold;
  }
}

The edge test needs one more thing: the previous price.

export function crossed(
  op: Operator,
  prevPrice: number,
  price: number,
  threshold: number,
): boolean {
  return !matches(op, prevPrice, threshold) && matches(op, price, threshold);
}

A crossing means the condition was false at the previous tick and is true at the current tick. One tick alone cannot establish that change. In my ticker, AlertEvaluator stores the previous price for each symbol.

export class AlertEvaluator {
  private readonly lastPrice = new Map<StockSymbol, number>();
  private readonly rulesBySymbol = new Map<StockSymbol, AlertRule[]>();

  process(tick: Tick): Notification[] {
    const prev = this.lastPrice.get(tick.symbol);
    this.lastPrice.set(tick.symbol, tick.price); // update state AFTER reading prev

    const rules = this.rulesBySymbol.get(tick.symbol);
    if (!rules) return [];

    const out: Notification[] = [];
    for (const rule of rules) {
      const fired =
        rule.trigger === "edge"
          ? prev !== undefined && crossed(rule.operator, prev, tick.price, rule.threshold)
          : matches(rule.operator, tick.price, rule.threshold);
      if (fired) out.push(this.toNotification(rule, tick));
    }
    return out;
  }
}

The evaluator reads the previous price, then updates lastPrice before it evaluates the rules. The next tick thus compares with the current tick. For a symbol's first tick, prev is undefined. An edge rule cannot fire because there is no previous price to establish a crossing.

I replayed the same 200 synthetic ticks with the same thresholds in edge and level modes. The edge mode produced 9 notifications. The level mode produced 194, approximately 21 times as many. These results apply to that input sequence. The required alert behavior determines which mode is correct.

sequenceDiagram
    participant T as Tick Stream
    participant E as AlertEvaluator "state: lastPrice"
    T->>E: price=9480 (below 9500)
    Note over E: lastPrice=9480, no fire
    T->>E: price=9505 (crosses 9500)
    Note over E: EDGE fires (was below, now above)
    Note over E: LEVEL fires too
    T->>E: price=9510 (still above 9500)
    Note over E: EDGE stays silent (no new crossing)
    Note over E: LEVEL fires again

State ownership across partitions

Each process keeps its lastPrice map inside an AlertEvaluator instance. In a partitioned deployment, each pod has an evaluator for its assigned partitions. The map holds state for the symbols in those partitions.

The market-updates and notifications topics use the same symbol key and six partitions, which provides matching partition numbers. A compatible assigner must also place matching partitions on the same pod. This supports local state access. It does not restore state after startup, a crash, or a rebalance.

I first considered co partitioning as a way to make joins cheaper. It also gives state a clear owner. If another pod needs that state, the application must transfer, copy, or rebuild it.

Grouping ticks into tumbling windows

Alerts evaluate individual ticks. A chart candle summarizes an interval with four prices: open, high, low, and close. To build candles from ticks, the application groups them into time windows.

A tumbling window is a fixed size, non overlapping bucket of time, say every 1000 milliseconds. Every tick that arrives gets folded into whichever bucket its timestamp belongs to. When a tick's timestamp belongs to a bucket later than the one currently open, the open bucket is done. It gets closed, emitted, and a new bucket starts.

OhlcAggregator groups ticks by symbol and calculates windowStart from each timestamp:

export class OhlcAggregator {
  private readonly current = new Map<StockSymbol, Candle>();

  constructor(private readonly windowMs: number) {}

  add(tick: Tick): Candle | null {
    const windowStart = Math.floor(tick.ts / this.windowMs) * this.windowMs;
    const cur = this.current.get(tick.symbol);

    if (!cur || cur.windowStart !== windowStart) {
      const closed = cur && cur.windowStart !== windowStart ? cur : null;
      this.current.set(tick.symbol, {
        symbol: tick.symbol,
        windowStart,
        open: tick.price,
        high: tick.price,
        low: tick.price,
        close: tick.price,
        volume: tick.volume,
        count: 1,
      });
      return closed;
    }

    cur.high = Math.max(cur.high, tick.price);
    cur.low = Math.min(cur.low, tick.price);
    cur.close = tick.price;
    cur.volume += tick.volume;
    cur.count++;
    return null;
  }

  flush(): Candle[] {
    const out = [...this.current.values()];
    this.current.clear();
    return out;
  }
}

The expression Math.floor(tick.ts / windowMs) * windowMs computes the window start. In this implementation, a tick in a later window closes the current window. If no later tick arrives, the window remains open until flush() runs.

This design assumes timestamps arrive in order for each symbol. A production implementation needs an explicit policy for late records and state recovery.

graph LR
    subgraph "window 09:00:00"
        A["tick O=9500"] --> B["tick H=9520"] --> C["tick L=9490"]
    end
    subgraph "window 09:00:01"
        D["tick opens new candle"]
    end
    C -->|"tick.ts crosses boundary"| D
    C -.->|closed candle emitted| E["Candle: O/H/L/C, volume, count"]

What 200 ticks look like as candles

The test uses 200 ticks across two symbols, with synthetic timestamps 10ms apart. A 50ms window produces 80 closed candles. Each candle contains open, high, low, close, and volume for its interval. This reduces the number of values the chart must display.

I implemented tumbling windows, which divide time into consecutive intervals that do not overlap. This fits the candle intervals in my chart.

Hopping windows advance by less than their width, so a tick can belong to several windows. Sliding windows update their bounds as events arrive. Session windows group activity and close after an inactivity gap.

State is yours to manage

Kafka Streams can restore configured state stores from changelog topics. My in memory implementation has no automatic restoration. After a crash, it must rebuild state from retained source records or restore a durable snapshot.

Kafka transactions can coordinate Kafka output and offsets, but they do not restore the local map by themselves. A recovery design must also retain and restore state at a position consistent with those offsets.

Redis can also support these patterns. A Lua script can atomically read and update stored state for an edge check. A sorted set can support a windowed aggregate. These Node examples still require application code to manage windows, retention, and recovery.

State and windows in one view

• An edge trigger compares the current and previous values. In the same 200 tick test, edge mode produced 9 notifications and level mode produced 194.
• AlertEvaluator reads lastPrice before it updates that value for the symbol.
• A later window closes the current window in this implementation. A final flush() emits any open windows.
• OhlcAggregator produced 80 candles from 200 ticks with a 50ms window.
• Matching topic partitions and consumer assignments support local state access. They do not provide state restoration.
• Recovery must restore state and coordinate it with source offsets and output.

I first verified this behavior through logs. I then built a live SSE dashboard to inspect ticks, alerts, and partition ownership in the browser.