Windowing — advanced¶

The everyday window usage is on Windowing (shared mechanics) and Window types (the kinds). This page covers what you reach for less often: the failover guarantees of persistent windows, wiring one window's result into another with Ref / join_with, deferring window construction with the recovery hook, and the one caveat sessions carry at very high key cardinality.

Inspecting calculation selection¶

At INFO level, each window reports window calculation registered when it starts processing on a worker. The log includes its name, worker and calculation_host: rust for supported built-in calculations, python for other accumulators or customized methods. Both use Rust window mechanics. For tumbling and sliding windows, a Multi uses Rust calculation only when every member supports it; Variance and StdDev support both population (ddof=0) and sample (ddof=1) calculation. Use RUST_LOG=turbine=info to include these messages (the default) and TURBINE_LOG_FORMAT=json for structured log output.

Failover-resilient persistent windows¶

PersistentTumbling, PersistentSliding and PersistentSession survive node failure. Accumulator state and the fire-time timer are both persisted and checkpointed to durable storage. When a partition migrates — because a node crashed and Raft redistributed its partitions, or during a rolling upgrade — the new owner restores from the latest snapshot, provided the checkpoint store is one that node can read: with a node-local checkpoint_url there is no snapshot to restore and the window state is simply lost, which is why a stateful cluster refuses to start on one (see Checkpoint storage). Any in-flight window whose deadline has already passed becomes eligible to close on recovery; future windows remain scheduled. Publishing many due groups takes time, so persistence does not guarantee sub-second close latency.

This is a real strength: long windows (hours, days) don't restart when a node dies. There is one correctness boundary — allowed_lateness_ms. If failover completes before window_end + allowed_lateness_ms, the window closes with all the events it would have seen in a steady-state cluster. If failover runs longer than that, the window still fires (the timer is overdue and fires as soon as the new worker boots), but events produced during the outage whose timestamps fall inside the closing window may arrive after the timer and be treated as late.

Pick allowed_lateness_ms wide enough to absorb your expected failover budget. In a cluster, a node that goes down keeps its partitions for raft_rebalance_delay_ms (60 s by default) before they move, so a minute or two of lateness tolerance is a reasonable floor for event-time windows you need to be complete.

Recovery scope¶

For persistent event-time windows, the protection against reopening an already-closed window covers local restart and full-checkpoint restoration of the same partition, including restoration on another node. Turbine restores the committed progress of that partition's data and window closures along with its state. This guarantee applies to the subscription's canonical event-time axis: the clock used to drive its watermark, normally selected by the input topic's event_time=. Windows that inherit that clock use the same restored progress.

Key-group rescaling remains outside this guarantee: it transfers window state and timers but does not carry the partition's temporal progress when splitting or combining workers. Persisting window state alone does not provide the recovery guarantee across this operation.

Windows whose explicit time_column= differs from the subscription's canonical clock are refused at construction. Independent clocks do not yet have separately restored progress; using the subscription's frontier could reject valid rows or reopen closed windows. Inherit the input topic's event_time= clock, or use time_column="_kafka_ts_ms" when the watermark comes from Kafka timestamps.

Window-to-window references (`Ref` / `join_with`)¶

Sometimes a window needs to compare its metric against another window over the same key: "the average latency this minute, falling back to the rolling 15-minute average when this minute is sparse", or "this 5-minute error rate against a 24-hour baseline". Rather than re-deriving the baseline inside every reader, one window can reference another window's last-closed result by name, straight from inside its aggregate expression.

Two pieces wire it up:

The referenced window must give its output a name — that's what Ref(...) looks up. Use .alias("name") for a single output, or Multi(name=...) for several.
The reader declares the dependency with join_with=[upstream] and pulls a named value into its own expression with Ref("upstream_name").output_name.

from turbine import KafkaBroker, RecordBatch, TurbineState
from turbine import aggregates as agg, functions as fn
from turbine import windowing as win

@app.subscribe(kafka.topic("events", event_time="ts", event_time_unit="ms"), output=kafka.topic("metrics"))
class tenant_metrics:
    def __init__(self) -> None:
        # Coarse, longer-horizon baseline — keyed on a SUBSET of the reader's keys.
        self._rollup = win.PersistentTumbling(
            self, name="rollup_15m", size_ms=15 * 60_000, key=["tenant"],
            value=agg.Mean("latency_ms").alias("avg_latency"),   # named → referenceable
            # no close callback: this window exists only to feed the reader
        )
        # Reader: when this 5-min window is sparse, fall back to the rolling baseline.
        self._win = win.PersistentTumbling(
            self, name="metrics_5m", size_ms=5 * 60_000, key=["tenant", "region"],
            value=agg.Mean(fn.coalesce(fn.col("latency_ms"),
                                       win.Ref("rollup_15m").avg_latency)),
            join_with=[self._rollup],
            on_close_each=self._emit,
        )

    def process(self, batch: RecordBatch, state: TurbineState) -> RecordBatch | None:
        self._rollup.update(batch, state)   # both windows see every event —
        self._win.update(batch, state)      # update each one you declared
        return None
    ...

Each window you build still gets its own update() in process() — join_with= wires the reference, not the ingestion. The order of the two update() calls doesn't matter (the reader reads the upstream's last closed value, never its in-flight state — see below).

Rules¶

Name the output. A bare Ref("w") (no field) is rejected at construction — always pick a named output (.alias(name) or Multi(name=...)). Forcing the field keeps the call site self-documenting and refactor-safe: adding a sibling metric later can't silently change which value resolves.
Last-closed value, per key. A Ref resolves to the upstream's most recently closed value for that key — never its currently-open partial. That makes it deterministic on replay and free of ordering races: a given close always reads the same upstream value, regardless of arrival order during accumulation. Until the upstream has closed at least once for a key, the reference is null, so wrap it in fn.coalesce(...) (or another null-tolerant expression). The currently-open upstream state is never exposed; if you genuinely need it, model a second window of equal cadence and accumulate explicitly.
Keys must be a subset. The referenced window's key= must be a subset of the reader's key= — rollup_15m keyed by ["tenant"] is readable from a reader keyed by ["tenant", "region"], and the reader projects the extra dimensions out when it looks up. Referencing a finer-keyed window is rejected (one lookup would map to many rows). Both windows must also share the same subscription partition_key.
Persistent references. Both ends use PersistentTumbling / PersistentSliding / PersistentSession, in any combination. Ref values live in the durable state store, persisted and recovered with the rest of the window state, so references survive restart and rescale.
Reference several, no cycles. join_with= takes a list, so one reader can blend baselines: join_with=[w15, w24h] with fn.coalesce(fn.col("x"), win.Ref("w15").avg, win.Ref("w24h").avg). Each Ref resolves independently (N references = N lookups, never combinatorial). Within a tick the runtime closes referenced windows before their readers, so a reader sees the value produced this tick; cycles (A references B references A) are rejected at construction.

Tumbling vs sliding as the reference. A sliding upstream makes a good rolling baseline: size_ms=15min, slide_ms=1min refreshes its ref value every minute, so the reader compares against a near-current value rather than one that only updates every 15 minutes. Tumbling-as-reference fits harder horizon resets — a daily or hourly baseline.

Cost. A referenced window writes one ref-value row per key on each close — but only if at least one window actually references it (statically known at construction; references nobody consumes cost nothing). A high-cardinality upstream that closes often (a short slide) writes proportionally more, so weigh that against the cadence you pick for the reference.

Dynamic windows¶

Most pipelines build their windows eagerly in __init__. When the set of windows isn't known at construction time, you can defer building them until the first batch arrives. The canonical example is a multi-tenant alerting app that creates one PersistentTumbling per (tenant, rule) pair, with tenants and rules added or removed at runtime — you can't enumerate every window in __init__.

The trade-off: until the first batch reaches process() and triggers construction, the SDK has no record of that window. If a previously-running instance had open windows due to close before any fresh traffic reaches the new instance, they need to be rebuilt before they can fire. The on_recover_persistent_windows(timer_id, state) hook closes that gap — the SDK calls it on a recovered window the processor hasn't built yet, giving you the chance to construct it via your usual lazy code path so it can fire correctly.

@app.subscribe(kafka.topic("events"), output=kafka.topic("counts"))
class event_counters:
    def __init__(self) -> None:
        self._wins: dict[str, win.PersistentTumbling] = {}

    def _build(self, field: str) -> win.PersistentTumbling:
        if field not in self._wins:
            self._wins[field] = win.PersistentTumbling(
                self,
                name=f"events_per_{field}",
                size_ms=60_000,
                key=[field],
                value=agg.Count(),
                on_close_each=self._emit,
            )
        return self._wins[field]

    def process(self, batch: RecordBatch, state: TurbineState) -> RecordBatch | None:
        self._build("region").update(batch, state)
        self._build("application").update(batch, state)
        return None

    def on_recover_persistent_windows(self, timer_id: str, state: TurbineState) -> None:
        # SDK calls this for any recovered window the processor hasn't built yet.
        # Route by name so the right factory rebuilds.
        name = win.PersistentTumbling.name_from_timer_id(timer_id)
        if name == "events_per_region":
            self._build("region")
        elif name == "events_per_application":
            self._build("application")

    def _emit(self, state: TurbineState, window: win.Window, value: float) -> RecordBatch:
        return pa.RecordBatch.from_pylist([{
            **window.group,                    # "region" or "application" depending on the window
            "count": float(value),
            "window_start_ms": window.start_ms,
            "window_end_ms": window.end_ms,
        }])

Notes:

name_from_timer_id is the routing primitive: it parses the window name back out of a recovered timer id, so the hook can dispatch to the matching factory call.
This is the clearest reason a window's name cannot double as its title. Recovery routes on the name because the name is inside the timer id — so renaming a window orphans its recovered timers along with its state. Windows built from configuration (one per rule, per tenant) should be named after whatever is stable about that configuration, typically an id, and given a readable label= for the console. The label is never in a key, so it can follow a rename that the name cannot.
The hook covers all persistent kinds — PersistentTumbling, PersistentSliding and PersistentSession. Use the matching class's name_from_timer_id(timer_id) when the processor manages a mix.

Sessions at scale¶

Session windows carry one performance characteristic worth knowing before you key one by a very high-cardinality column. Because a session close inherently merges, sessions are computed on a per-group path rather than the vectorised state-table layout that tumbling and sliding windows use — so the work scales with the number of distinct keys a batch touches, not just its row count. Keying by a raw user_id with millions of distinct values, where a single batch touches on the order of 100 k distinct sessions, is the regime where this shows up.

PersistentSession batches state loading and writes, but its per-group work still costs more than fixed-boundary window updates at high cardinality. Size the deployment using a representative session workload and the expected number of concurrently open sessions.

Practical guidance:

Low-to-moderate distinct-key counts — thousands of active keys per batch, the typical "sessions per tenant / per device class / per account" shape — run at full session throughput. Use exactly-once when outputs must not be duplicated.
Very high-cardinality keys (hundreds of thousands of distinct sessions per batch) are supported but slower. Spread the load across more shards with partition_key + parallelism (see Partitioning) to recover aggregate throughput, and size the deployment for the reduced per-partition rate. If that still isn't enough, a tumbling approximation keyed on a coarser bucket sidesteps the per-group path entirely.

Closing priority and health¶

Closing automatically gains priority after the worker has continuously observed eligible timers for one second. Observation starts when the worker checks for eligible work, after its current input batch; it does not start at the window's event-time end. A separate safety drain handles very old timers immediately. No additional window or app setting is needed.

This is a scheduling target, not a bound on output publication. An expensive callback, slow output or insufficient processing capacity can still delay publication; priority then takes time away from ingestion and consumer lag can increase. Callbacks and commits always finish before shutdown between chunks.

TURBINE_CLOSE_BACKLOG_THRESHOLD_MS controls health reporting independently (default 60 s). Continuously observed due work must persist past that threshold. Historical watermark jumps can therefore drain in priority while health stays healthy. See monitoring for the due-age and priority metrics.