Throttling Compliance Alert Storms During Chain Resets

This walkthrough sits under Real-Time Compliance Alerting & Webhooks and solves one precise task: keeping a chain-wide reset from drowning your on-call channel in thousands of simultaneous alerts, without muting the single genuine breach that happens to fire during the same hour. A reset is a legitimate event — a category is being reflowed, so every fixture in scope drops below its compliance threshold at nearly the same capture_timestamp. Naive alerting treats that drop as thousands of independent incidents, pages a district manager once per fixture, and trains everyone to ignore the channel by lunchtime. The opposite failure is just as bad: a blanket mute over the reset window swallows the one endcap that collapsed for a real reason. The fix is a small pipeline that sits between your alert stream and your webhook consumers — it detects the reset window, deduplicates identical alerts, aggregates the storm into one per-store rollup, and rate-limits each consumer with a token bucket that still lets a lone breach through. This page builds that pipeline step by step, and each step is independently verifiable.

Prerequisites & Context Jump to heading#

Before wiring the throttler in, confirm these are already in place. The pipeline is a pure transform over an alert stream — it invents no alerts of its own, it only decides which ones leave.

• Runtime: Python 3.11+ on the alerting host. No GPU and no external dependencies beyond the standard library; the reference code uses only dataclasses, collections, and time.
• A keyed alert stream: each alert arrives as a record carrying at least fixture_id, store_id, a banner (chain sub-brand), the breached metric, and a monotonic capture_timestamp. Alerts originate from the compliance payload described in Planogram Sync & SKU Mapping Strategies — the same compliance_percentage, out_of_stock_flags, and registry_revision fields that scoring emits.
• A reset signal or schedule: either a known reset calendar (store, category, planned start and end) or a live burst-rate probe over the stream. Most chains have both; use the schedule as the authoritative window and the burst probe as the safety net for unscheduled reflows.
• A registry_revision on each alert: a chain reset advances the planogram registry, so a reset alert and a steady-state alert differ by revision. This lets the detector confirm a burst is a genuine reset rather than a camera outage. The revision semantics are owned by the Planogram Sync & SKU Mapping Strategies pillar.
• A consumer registry: the set of webhook endpoints you deliver to (Slack channel, PagerDuty service, district-manager email relay), each with its own tolerance for volume. Rate limiting is per consumer, not global.

A note on terms: a storm is any window where the per-store alert rate exceeds what a human can act on; a rollup is one synthesized alert that stands in for every suppressed member of a storm. The goal is fewer alerts that each carry more meaning — never fewer alerts that hide a real breach.

Step 1 — Detect the Reset Window Jump to heading#

The first decision is whether a reset is happening at all, because everything downstream branches on it. Combine two signals. The schedule is authoritative: if a store and category are inside a planned reset window, mark it active. The burst-rate probe is the safety net: if alerts for one store_id cross a rate ceiling within a short window — say more than 40 fixtures breaching in 60 seconds while registry_revision is advancing — treat it as an unscheduled reset even without a calendar entry. Requiring the revision to move guards against a camera or broker outage masquerading as a reset; a true reset advances the registry, an outage does not.

Keep the detector honest about time. Windows are compared against the alert’s own capture_timestamp, not wall-clock arrival, so replayed or delayed events land in the correct window. This is the same backpressure-aware discipline the Message Broker Patterns for Capture Events cluster applies upstream, where a reset also produces a capture burst.

from __future__ import annotations

from collections import defaultdict, deque
from dataclasses import dataclass, field


@dataclass(frozen=True)
class Alert:
    fixture_id: str
    store_id: str
    banner: str
    metric: str            # e.g. "compliance_percentage"
    value: float
    registry_revision: int
    capture_timestamp: float  # epoch seconds, from the source event


@dataclass
class ResetDetector:
    """Marks a store in a reset window via schedule or burst rate."""
    burst_threshold: int = 40          # fixtures per window to call a storm
    burst_window_s: float = 60.0
    scheduled: dict[str, tuple[float, float]] = field(default_factory=dict)
    _recent: dict[str, deque[tuple[float, int]]] = field(
        default_factory=lambda: defaultdict(deque)
    )

    def in_reset(self, alert: Alert) -> bool:
        start_end = self.scheduled.get(alert.store_id)
        if start_end and start_end[0] <= alert.capture_timestamp <= start_end[1]:
            return True
        return self._burst_active(alert)

    def _burst_active(self, alert: Alert) -> bool:
        buf = self._recent[alert.store_id]
        buf.append((alert.capture_timestamp, alert.registry_revision))
        cutoff = alert.capture_timestamp - self.burst_window_s
        while buf and buf[0][0] < cutoff:
            buf.popleft()
        revisions = {rev for _, rev in buf}
        # A genuine reset advances the registry; an outage does not.
        return len(buf) >= self.burst_threshold and len(revisions) > 1

Step 2 — Deduplicate Identical Alerts Within a Window Jump to heading#

A reset does not just fire once per fixture — retries, re-captures, and multiple metrics per fixture mean the same logical alert can appear several times in seconds. Collapse those with a suppression key: a stable hash of the fields that make an alert “the same incident” — here store_id, fixture_id, and metric. The first alert for a key inside the dedup window passes; identical repeats are dropped until the window expires. This runs whether or not a reset is active, because duplicate suppression is always safe: two identical alerts are never two incidents.

import time


@dataclass
class DedupWindow:
    """Drops repeat alerts sharing a suppression key within window_s."""
    window_s: float = 90.0
    _seen: dict[str, float] = field(default_factory=dict)

    @staticmethod
    def suppression_key(alert: Alert) -> str:
        return f"{alert.store_id}:{alert.fixture_id}:{alert.metric}"

    def is_duplicate(self, alert: Alert, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        key = self.suppression_key(alert)
        last = self._seen.get(key)
        if last is not None and now - last < self.window_s:
            return True
        self._seen[key] = now
        return False

The suppression key is deliberately coarse: it keys on the fixture and metric, not on the exact value, so a fixture that flaps between 61.0% and 62.5% during a reflow does not slip past dedup as a “new” alert each time.

Step 3 — Aggregate a Storm Into One Rollup Jump to heading#

When the detector says a store is in a reset window, every deduplicated alert for that store folds into a single rollup rather than paging individually. The rollup carries the counts a responder actually needs — how many fixtures dropped, which banner, the revision range — and a sample of the worst offenders, so it stays actionable without being noisy. Alerts for a store that is not in a reset window skip aggregation entirely and pass through as individual alerts; that is what keeps the lone genuine breach visible.

@dataclass
class StoreRollup:
    store_id: str
    banner: str
    fixture_count: int
    min_value: float
    revisions: tuple[int, ...]
    sample_fixtures: list[str]


class Aggregator:
    """Folds per-fixture reset alerts into one rollup per store."""

    def __init__(self, sample_size: int = 5) -> None:
        self._buffers: dict[str, list[Alert]] = defaultdict(list)
        self._sample_size = sample_size

    def add(self, alert: Alert) -> None:
        self._buffers[alert.store_id].append(alert)

    def flush(self) -> list[StoreRollup]:
        rollups: list[StoreRollup] = []
        for store_id, alerts in self._buffers.items():
            if not alerts:
                continue
            worst = sorted(alerts, key=lambda a: a.value)[: self._sample_size]
            rollups.append(
                StoreRollup(
                    store_id=store_id,
                    banner=alerts[0].banner,
                    fixture_count=len(alerts),
                    min_value=round(min(a.value for a in alerts), 2),
                    revisions=tuple(sorted({a.registry_revision for a in alerts})),
                    sample_fixtures=[a.fixture_id for a in worst],
                )
            )
        self._buffers.clear()
        return rollups

Step 4 — Rate-Limit Each Consumer With a Token Bucket Jump to heading#

Even after dedup and aggregation, a chain of 1200 stores resetting in the same hour yields hundreds of rollups. The final gate is a token bucket per consumer: each consumer starts with a burst allowance — 20 tokens — that refills at a steady rate, say 1 token every 3 seconds. Every delivery spends one token; when the bucket empties, further alerts are dropped (and counted, so you can surface the drop as a metric) until it refills. The burst allowance is deliberately generous so a genuine cluster of a few real breaches all get through, while a runaway flood is capped. Wrap the whole pipeline so the four stages compose in order.

@dataclass
class TokenBucket:
    """Classic token bucket: burst capacity plus steady refill."""
    capacity: int = 20
    refill_per_s: float = 1 / 3
    _tokens: float = field(init=False)
    _last: float = field(init=False)

    def __post_init__(self) -> None:
        if self.capacity <= 0 or self.refill_per_s <= 0:
            raise ValueError("capacity and refill_per_s must be positive")
        self._tokens = float(self.capacity)
        self._last = time.monotonic()

    def allow(self, now: float | None = None) -> bool:
        now = time.monotonic() if now is None else now
        elapsed = max(0.0, now - self._last)
        self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_per_s)
        self._last = now
        if self._tokens >= 1.0:
            self._tokens -= 1.0
            return True
        return False


class AlertThrottler:
    """Reset detect -> dedup -> aggregate -> per-consumer token bucket."""

    def __init__(self, detector: ResetDetector, dedup: DedupWindow) -> None:
        self.detector = detector
        self.dedup = dedup
        self.aggregator = Aggregator()
        self._buckets: dict[str, TokenBucket] = defaultdict(TokenBucket)
        self.dropped = 0

    def ingest(self, alert: Alert, now: float | None = None) -> None:
        if self.dedup.is_duplicate(alert, now):
            return
        if self.detector.in_reset(alert):
            self.aggregator.add(alert)
        else:
            self._emit(f"breach:{alert.store_id}", alert, now)

    def _emit(self, consumer: str, payload: object, now: float | None) -> None:
        if self._buckets[consumer].allow(now):
            deliver(consumer, payload)   # your webhook call
        else:
            self.dropped += 1

    def flush_storm(self, consumer: str, now: float | None = None) -> None:
        for rollup in self.aggregator.flush():
            self._emit(consumer, rollup, now)


def deliver(consumer: str, payload: object) -> None:
    """Stand-in for the webhook POST; replace with a real client."""
    print(f"-> {consumer}: {payload}")

Verification & Testing Jump to heading#

Assert the behaviour deterministically by driving the pipeline with synthetic timestamps rather than sleeping:

1. A storm collapses to one rollup. Feed 2000 alerts across 50 fixtures for one store_id inside a scheduled window, then flush_storm; assert exactly one StoreRollup reaches the consumer and its fixture_count equals the distinct fixtures seen after dedup.
2. A single real breach still fires. Send one alert for a store with no schedule entry and a stream too sparse to trip the burst probe; assert it is delivered individually and never enters the aggregator.
3. Dedup drops identical repeats. Send the same (store_id, fixture_id, metric) three times within window_s; assert only the first is processed and the other two return early.
4. The bucket refills correctly. Drain a TokenBucket(capacity=20) with 20 allow(now) calls at a fixed now, assert the 21st returns False, then advance now by 30 seconds and assert roughly 10 more calls succeed at a 1/3 per-second refill.
5. Drops are counted, not silent. After overflowing a bucket, assert throttler.dropped is greater than zero so the suppression is observable on a dashboard rather than invisible.

A healthy run during a real reset shows the consumer receiving one rollup per resetting store plus any genuine out-of-window breaches, with dropped staying at zero unless a single store somehow exceeds the burst allowance on its own.

Troubleshooting Jump to heading#

Related Jump to heading#

• Real-Time Compliance Alerting & Webhooks — the parent cluster that owns the webhook delivery contract this throttler sits in front of
• Message Broker Patterns for Capture Events — upstream backpressure during the capture burst a chain reset also produces
• Planogram Sync & SKU Mapping Strategies — the source of the compliance payload and the registry_revision semantics the detector relies on