|
| 1 | +"""Smart Capture guardrails — rate limiting and flapping protection.""" |
| 2 | + |
| 3 | +import logging |
| 4 | +import threading |
| 5 | +import time |
| 6 | + |
| 7 | +log = logging.getLogger("docsis.smart_capture.guardrails") |
| 8 | + |
| 9 | + |
| 10 | +class GuardrailChain: |
| 11 | + """Applies guardrails in order: flapping, global cooldown, per-trigger |
| 12 | + cooldown, max actions per hour. |
| 13 | +
|
| 14 | + Thread-safe — all state access is protected by a lock. |
| 15 | +
|
| 16 | + Cooldown semantics: 0 = disabled (no cooldown, always allow). |
| 17 | + This differs from NotificationDispatcher where 0 = never send. |
| 18 | +
|
| 19 | + Global cooldown is batch-aware: it is checked once per source event |
| 20 | + (not per trigger), and updated only after all triggers for that event |
| 21 | + have been evaluated. This prevents a second trigger for the same event |
| 22 | + from being suppressed by the first trigger's fire. |
| 23 | + """ |
| 24 | + |
| 25 | + def __init__(self, config_mgr): |
| 26 | + self._config = config_mgr |
| 27 | + self._lock = threading.Lock() |
| 28 | + self._last_global_fire: float = 0.0 |
| 29 | + self._last_trigger_fire: dict[str, float] = {} |
| 30 | + self._hourly_fires: list[float] = [] |
| 31 | + self._trigger_match_history: dict[str, list[float]] = {} |
| 32 | + |
| 33 | + def check_batch(self, trigger_results): |
| 34 | + """Evaluate guardrails for a batch of (trigger, event) pairs from one source event. |
| 35 | +
|
| 36 | + Args: |
| 37 | + trigger_results: list of (trigger, event) tuples that matched. |
| 38 | +
|
| 39 | + Returns: |
| 40 | + list of (trigger, event, allowed, reason) tuples. |
| 41 | +
|
| 42 | + Global cooldown is checked once against the batch. Per-trigger |
| 43 | + cooldown and hourly limit are checked per trigger. Flapping counts |
| 44 | + all matches (not just allowed ones) to detect chattering input. |
| 45 | + """ |
| 46 | + if not trigger_results: |
| 47 | + return [] |
| 48 | + |
| 49 | + results = [] |
| 50 | + with self._lock: |
| 51 | + now = time.monotonic() |
| 52 | + |
| 53 | + # 1. Global cooldown — checked once for the whole batch |
| 54 | + global_cd = int(self._config.get("sc_global_cooldown", 300)) |
| 55 | + global_blocked = ( |
| 56 | + global_cd > 0 |
| 57 | + and (now - self._last_global_fire) < global_cd |
| 58 | + ) |
| 59 | + global_reason = None |
| 60 | + if global_blocked: |
| 61 | + remaining = int(global_cd - (now - self._last_global_fire)) |
| 62 | + global_reason = f"global_cooldown: {remaining}s remaining" |
| 63 | + |
| 64 | + any_allowed = False |
| 65 | + for trigger, event in trigger_results: |
| 66 | + trigger_key = f"{trigger.event_type}:{trigger.action_type}" |
| 67 | + |
| 68 | + # Record match for flapping (counts ALL matches, not just fires) |
| 69 | + history = self._trigger_match_history.get(trigger_key, []) |
| 70 | + window = int(self._config.get("sc_flapping_window", 3600)) |
| 71 | + history = [t for t in history if (now - t) < window] |
| 72 | + history.append(now) |
| 73 | + self._trigger_match_history[trigger_key] = history |
| 74 | + |
| 75 | + # 2. Flapping — checked before cooldowns |
| 76 | + threshold = int(self._config.get("sc_flapping_threshold", 3)) |
| 77 | + if threshold > 0 and len(history) > threshold: |
| 78 | + results.append((trigger, event, False, |
| 79 | + f"flapping: {len(history)} matches in {window}s for {trigger_key}")) |
| 80 | + continue |
| 81 | + |
| 82 | + # 3. Global cooldown |
| 83 | + if global_blocked: |
| 84 | + results.append((trigger, event, False, global_reason)) |
| 85 | + continue |
| 86 | + |
| 87 | + # 4. Per-trigger cooldown |
| 88 | + trigger_cd = int(self._config.get("sc_trigger_cooldown", 900)) |
| 89 | + last_fire = self._last_trigger_fire.get(trigger_key, 0.0) |
| 90 | + if trigger_cd > 0 and (now - last_fire) < trigger_cd: |
| 91 | + remaining = int(trigger_cd - (now - last_fire)) |
| 92 | + results.append((trigger, event, False, |
| 93 | + f"trigger_cooldown: {remaining}s remaining for {trigger_key}")) |
| 94 | + continue |
| 95 | + |
| 96 | + # 5. Max actions per hour |
| 97 | + max_per_hour = int(self._config.get("sc_max_actions_per_hour", 4)) |
| 98 | + cutoff = now - 3600 |
| 99 | + self._hourly_fires = [t for t in self._hourly_fires if t > cutoff] |
| 100 | + if max_per_hour > 0 and len(self._hourly_fires) >= max_per_hour: |
| 101 | + results.append((trigger, event, False, |
| 102 | + f"max_actions_per_hour: {len(self._hourly_fires)}/{max_per_hour} used")) |
| 103 | + continue |
| 104 | + |
| 105 | + # All passed |
| 106 | + self._last_trigger_fire[trigger_key] = now |
| 107 | + self._hourly_fires.append(now) |
| 108 | + any_allowed = True |
| 109 | + results.append((trigger, event, True, None)) |
| 110 | + |
| 111 | + # Update global cooldown only if at least one execution was allowed |
| 112 | + if any_allowed: |
| 113 | + self._last_global_fire = now |
| 114 | + |
| 115 | + return results |
0 commit comments