Sliding-Window Correlation with Redis Sorted Sets

A single-process sliding window dies with the process and cannot be shared across a worker fleet. This guide implements a distributed sliding window on Redis sorted sets — the standard pattern for correlation state that survives restarts and scales horizontally — as one precise technique within temporal correlation windows and the broader alert correlation rule engines pipeline.

Root-Cause Context

Correlation windows held in Python dictionaries have two hard limits: they are bound to one process, and they vanish on restart. A SOC running a consumer group of eight parser workers cannot hold a shared “failed logons per account in the last 60 seconds” count in any one worker’s memory, because each worker sees only its slice of the partitioned stream. Moving window state into Redis solves both problems — but only if the operations are atomic, because a read-modify-write across the network is a race that under-counts a burst exactly when it matters.

A Redis sorted set (ZSET) is the right structure: members are event ids, scores are event timestamps. Sliding the window is then two operations — evict everything with a score below now − window, and count what remains — and both are server-side. The remaining risk is doing eviction, insertion, and counting as separate round-trips; under concurrency that interleaves and mis-counts. Pipelining them into one atomic unit is what makes the distributed window correct.

Prerequisites

The implementation targets Python 3.11+ with the async Redis client. It assumes a reachable Redis 7 instance and events normalized upstream by the parent temporal correlation windows guide.

pip install "redis>=5.0,<6.0" "pydantic>=2.6,<3.0"

Production-Ready Implementation

Each event runs one atomic pipeline: add the member, evict the expired range, count survivors, and set a TTL so idle keys self-clean. The count is compared to the rule threshold.

from __future__ import annotations

import logging

import redis.asyncio as redis
from pydantic import BaseModel, Field

logger = logging.getLogger("soc.correlation.rediswin")


class WindowEvent(BaseModel):
    event_id: str = Field(min_length=1)
    key: str = Field(min_length=1)         # correlation key
    event_time: float = Field(gt=0)        # epoch seconds


class RedisSlidingWindow:
    def __init__(self, client: redis.Redis, window_seconds: float = 60.0,
                 max_members: int = 10_000) -> None:
        self._r = client
        self._w = window_seconds
        self._max = max_members

    def _zkey(self, key: str) -> str:
        return f"win:{key}"

    async def add_and_count(self, ev: WindowEvent) -> int:
        """Atomically insert, evict expired, and return the in-window count."""
        zkey = self._zkey(ev.key)
        cutoff = ev.event_time - self._w
        async with self._r.pipeline(transaction=True) as pipe:
            pipe.zadd(zkey, {ev.event_id: ev.event_time})
            pipe.zremrangebyscore(zkey, "-inf", f"({cutoff}")   # exclusive lower
            pipe.zcard(zkey)
            pipe.expire(zkey, int(self._w) + 5)                 # idle self-clean
            results = await pipe.execute()
        count = int(results[2])
        if count > self._max:                                   # ERR_WINDOW_031
            # Trim to the newest max_members to bound memory.
            await self._r.zremrangebyrank(zkey, 0, count - self._max - 1)
            logger.warning("ERR_WINDOW_031 key=%s trimmed to %d", ev.key, self._max)
            return self._max
        return count

    async def fires(self, ev: WindowEvent, threshold: int) -> bool:
        count = await self.add_and_count(ev)
        if count >= threshold:
            logger.info('{"msg":"window_fire","key":"%s","count":%d}', ev.key, count)
            return True
        return False


async def _demo() -> None:
    client = redis.Redis(host="localhost", port=6379, decode_responses=True)
    win = RedisSlidingWindow(client, window_seconds=60.0)
    base = 1_000_000.0
    fired = False
    for i in range(12):
        fired = await win.fires(
            WindowEvent(event_id=f"e{i}", key="user:jdoe", event_time=base + i),
            threshold=10,
        )
    assert fired is True         # 10th+ event within 60s trips the threshold
    await client.aclose()


if __name__ == "__main__":
    import asyncio
    asyncio.run(_demo())

Wrapping the four commands in a transaction=True pipeline (a Redis MULTI/EXEC) makes insert-evict-count atomic, so two workers processing the same key concurrently cannot interleave into an under-count. The exclusive lower bound ((cutoff) ensures an event exactly at the window edge is treated consistently, and the EXPIRE guarantees idle keys evaporate without a sweeper job.

Error-Code Reference

Operational Notes

• Atomicity is the whole point. Never split add, evict, and count into separate awaited calls; a MULTI/EXEC pipeline (or a Lua script) keeps them race-free under a consumer group.
• Let keys expire themselves. Setting a TTL just past the window means idle correlation keys self-clean, eliminating a separate reaper and bounding total key count to active entities.
• Bound per-key cardinality. A scanning host can push thousands of members into one ZSET; the rank-trim keeps memory flat, mirroring the cap discipline in cross-source event linking.
• Shard by key hash across Redis nodes beyond a single instance’s throughput; the correlation key is the natural shard key.

Verification Checklist

• The 10th event within the window trips a threshold-10 rule; the 9th does not.
• Events older than the window are evicted and not counted.
• Two concurrent workers on the same key produce a correct, non-double-counted total.
• An idle key expires via its TTL without a separate sweeper.
• A per-key flood is trimmed to the cap and emits ERR_WINDOW_031.
• A Redis outage trips the circuit breaker rather than losing events silently.

FAQ

Related

• Temporal Correlation Windows
• Cross-Source Event Linking
• Threshold Tuning Strategies
• Async Log Batching
• Alert Correlation & Rule Engines