# Provider Circuit Breaker / Cooldown Pattern

Provider cooldown tracking behind a Clean Architecture port, with in-memory and Redis backends. Designed for debrid/third-party API providers that return 429 (rate limit) or 5xx (server error).

## Overview

When a provider API returns 429 Too Many Requests or repeated 5xx errors, subsequent resolve attempts for the same user+provider are short-circuited for a cooldown period. This prevents:
- Hitting the provider's rate limit harder (429 → longer cooldown → worse UX)
- Wasting time on a degraded provider (5xx → timeout → user waits 30s for nothing)
- Cascading failures (multiple users hitting the same degraded provider simultaneously)

## Port Design

```python
from typing import Protocol

class ProviderCooldown(Protocol):
    async def is_cooling_down(self, user_id: int, provider: str) -> bool:
        """Return True if the provider is still in cooldown for this user."""
        ...

    async def mark_cooldown(self, user_id: int, provider: str, seconds: int) -> None:
        """Mark the provider as cooling down for the given seconds."""
        ...
```

## Cooldown Key Design

```python
def _cooldown_key(user_id: int, provider: str) -> str:
    return f"cooldown:{user_id}:{provider}"
```

Key is scoped by user + provider. This means:
- User A's Real-Debrid cooldown doesn't affect User B
- User A's Real-Debrid cooldown doesn't affect User A's AllDebrid
- Each user+provider pair has independent cooldown state

## In-Memory Implementation

```python
class InMemoryProviderCooldown:
    def __init__(self):
        self._store: dict[str, float] = {}

    async def is_cooling_down(self, user_id, provider):
        key = _cooldown_key(user_id, provider)
        expires_at = self._store.get(key)
        if expires_at is None:
            return False
        if time.monotonic() >= expires_at:
            del self._store[key]
            return False
        return True

    async def mark_cooldown(self, user_id, provider, seconds):
        if seconds < 1:
            return
        key = _cooldown_key(user_id, provider)
        self._store[key] = time.monotonic() + seconds
```

## Redis Implementation

```python
class RedisProviderCooldown:
    def __init__(self, redis, key_prefix="cooldown:"):
        self._redis = redis
        self._key_prefix = key_prefix

    async def is_cooling_down(self, user_id, provider):
        with suppress(Exception):
            value = await self._redis.get(f"{self._key_prefix}{user_id}:{provider}")
            return value is not None
        return False  # Fail-open: if Redis is down, allow through

    async def mark_cooldown(self, user_id, provider, seconds):
        if seconds < 1:
            return
        with suppress(Exception):
            await self._redis.setex(
                f"{self._key_prefix}{user_id}:{provider}",
                seconds, "1",
            )
```

## Retry-After Parsing

When a provider returns HTTP 429, the response often includes a `Retry-After` header. Parse it to set a precise cooldown:

```python
def _parse_retry_after(header_value: str | None) -> int | None:
    """Parse Retry-After header, returning seconds or None if absent.

    Supports both delta-seconds (integer) and HTTP-date formats.
    Caps at 5 minutes to prevent unreasonably long cooldowns.
    """
    if not header_value:
        return None
    try:
        seconds = int(header_value)
        return max(1, min(seconds, 300))  # Cap at 5 minutes
    except ValueError:
        return None  # HTTP-date format — skip for simplicity
```

## Exception-Based Cooldown Propagation

The `ProviderStreamUnavailableError` carries an optional `cooldown_seconds` hint:

```python
class ProviderStreamUnavailableError(StreamResolveError):
    def __init__(self, message=None, *, cooldown_seconds: int | None = None):
        super().__init__(message)
        self.cooldown_seconds = cooldown_seconds  # Hint for circuit breaker
```

Provider resolver helpers (`_request_json`, `_request_no_content`) set `cooldown_seconds` from Retry-After on 429, or use a default (30s) on 5xx:

```python
if response.status_code == 429:
    retry_after = _parse_retry_after(response.headers.get("Retry-After"))
    raise ProviderStreamUnavailableError(cooldown_seconds=retry_after)
if response.status_code >= 500:
    raise ProviderStreamUnavailableError(cooldown_seconds=30)  # Default
if response.status_code >= 400:
    raise ProviderStreamUnavailableError(cooldown_seconds=30)  # Default
```

## Use Case Integration

```python
class StreamResolveUseCase:
    def __init__(self, ..., provider_cooldown: ProviderCooldown | None = None):
        self._provider_cooldown = provider_cooldown

    async def resolve(self, user_id, request):
        # ... account selection ...

        # Check cooldown before resolving
        if self._provider_cooldown is not None:
            if await self._try_check_cooldown(user_id, provider_name):
                raise ProviderStreamUnavailableError()

        # ... cache check ...

        # Resolve
        try:
            results = await resolver.resolve(source, api_key)
        except StreamResolveError as exc:
            if self._provider_cooldown is not None:
                cooldown = getattr(exc, "cooldown_seconds", 30) or 30
                await self._try_mark_cooldown(user_id, provider_name, cooldown)
            raise
        except Exception:
            if self._provider_cooldown is not None:
                await self._try_mark_cooldown(user_id, provider_name, 30)
            raise ProviderStreamUnavailableError() from None

        # ... cache store ...
        return results
```

**Fail-open methods:**

```python
async def _try_check_cooldown(self, user_id, provider) -> bool:
    if self._provider_cooldown is None:
        return False
    try:
        return await self._provider_cooldown.is_cooling_down(user_id, provider)
    except Exception:
        return False  # Fail-open: allow through if cooldown check fails

async def _try_mark_cooldown(self, user_id, provider, seconds) -> None:
    if self._provider_cooldown is None:
        return
    with suppress(Exception):
        await self._provider_cooldown.mark_cooldown(user_id, provider, seconds)
```

## Testing

```python
class TestInMemoryProviderCooldown:
    @pytest.fixture
    def cooldown(self):
        return InMemoryProviderCooldown()

    async def test_not_cooling_down_by_default(self, cooldown):
        assert await cooldown.is_cooling_down(1, "real_debrid") is False

    async def test_mark_and_check(self, cooldown):
        await cooldown.mark_cooldown(1, "real_debrid", 60)
        assert await cooldown.is_cooling_down(1, "real_debrid") is True

    async def test_different_users_independent(self, cooldown):
        await cooldown.mark_cooldown(1, "real_debrid", 60)
        assert await cooldown.is_cooling_down(1, "real_debrid") is True
        assert await cooldown.is_cooling_down(2, "real_debrid") is False

    async def test_different_providers_independent(self, cooldown):
        await cooldown.mark_cooldown(1, "real_debrid", 60)
        assert await cooldown.is_cooling_down(1, "real_debrid") is True
        assert await cooldown.is_cooling_down(1, "all_debrid") is False

    async def test_zero_seconds_does_nothing(self, cooldown):
        await cooldown.mark_cooldown(1, "real_debrid", 0)
        assert await cooldown.is_cooling_down(1, "real_debrid") is False

    async def test_fail_open_on_error(self):
        class _BrokenRedis:
            async def get(self, key): raise ConnectionError("Down")
            async def setex(self, key, sec, val): raise ConnectionError("Down")
            async def delete(self, *keys): raise ConnectionError("Down")

        cooldown = RedisProviderCooldown(_BrokenRedis())
        assert await cooldown.is_cooling_down(1, "real_debrid") is False
        await cooldown.mark_cooldown(1, "real_debrid", 60)  # Should not raise
```

## Why Not a Full Circuit Breaker State Machine?

Traditional circuit breaker libraries (like `pybreaker`) implement a 3-state machine: CLOSED → OPEN → HALF_OPEN. For debrid/streaming providers:

- **Not needed:** Providers rarely auto-recover within a half-open window. The cooldown is simpler and achieves the same effect.
- **Too complex:** A state machine adds `on_success` / `on_failure` callbacks, half-open probe logic, and failure threshold counters. The cooldown pattern maps directly to the provider's `Retry-After` header.
- **Simpler testing:** No state transitions to verify. Test: add cooldown → check → assert blocked → wait → check → assert allowed.

**When to use a full circuit breaker instead:**
- Provider health fluctuates rapidly (seconds-level recovery)
- You need to probe with health-check requests (not production traffic)
- Multiple downstream services share a single provider connection pool

## Key Invariants

1. **Cooldown is per user+provider** — never global (a different user may have a different rate limit)
2. **Fail-open on Redis errors** — provider resolves when cooldown state is unavailable
3. **Retry-After from 429 takes priority** — exact match to provider's requested backoff, capped at 5 min
4. **Default cooldown for 5xx: 30s** — enough to let transient failures clear, short enough to not harm UX
5. **Zero/negative seconds are no-ops** — prevents caller errors from creating permanent cooldowns
6. **Circuit breaker is optional** — `provider_cooldown=None` by default for backward compatibility