from future import annotations

import threading
import unittest
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, Hashable, Optional

@dataclass(frozen=True)
class TenantConfig:
"""Límites de tasa para un inquilino.

client_limit se aplica de forma independiente a cada cliente, mientras que tenant_limit se aplica
a todas las solicitudes admitidas del inquilino. Ambos límites utilizan ventanas rodantes exactas.
active_timeout_ms limita cuánto tiempo un observador de justicia ausente puede retener su turno.
eviction_ms controla la retención del estado del cliente.
"""

client_limit: int
client_window_ms: int
tenant_limit: int
tenant_window_ms: int
active_timeout_ms: int = 60_000
eviction_ms: int = 120_000

def validate(self) -> None:
    if self.client_limit <= 0 or self.tenant_limit <= 0:
        raise ValueError("los límites deben ser positivos")
    if self.client_window_ms <= 0 or self.tenant_window_ms <= 0:
        raise ValueError("las longitudes de ventana deben ser positivas")
    if self.active_timeout_ms <= 0:
        raise ValueError("active_timeout_ms debe ser positivo")
    if self.eviction_ms < max(self.client_window_ms, self.tenant_window_ms):
        raise ValueError("eviction_ms debe ser al menos tan grande como ambas ventanas")
    if self.eviction_ms < self.active_timeout_ms:
        raise ValueError("eviction_ms debe ser al menos active_timeout_ms")

@dataclass(frozen=True)
class AllowResult:
allowed: bool
retry_after_ms: int = 0

@dataclass
class _ClientState:
admitted: Deque[int] = field(default_factory=deque)
last_seen_ms: int = 0

@dataclass
class _TenantState:
admitted: Deque[int] = field(default_factory=deque)
clients: Dict[Hashable, _ClientState] = field(default_factory=dict)
waiters: Deque[Hashable] = field(default_factory=deque)
waiter_set: set[Hashable] = field(default_factory=set)
last_now_ms: Optional[int] = None
last_seen_ms: int = 0

class SlidingWindowRateLimiter:
"""Limitador de tasa de ventana deslizante seguro para hilos, de un solo proceso y exacto.

Algoritmo de ventana deslizante:
  Cada marca de tiempo de solicitud admitida se almacena en una cola (deque). Antes de evaluar una
  solicitud, se eliminan las marcas de tiempo t que satisfacen t <= ahora - ventana. Por lo tanto, una
  solicitud que tenga exactamente una ventana completa de antigüedad ya no consume capacidad. Esta es una
  verdadera ventana deslizante basada en registros, por lo que no tiene ráfagas en el límite del cubo del calendario.

  El resultado es exacto, con operaciones de cola (deque) amortizadas O(1) y memoria
  proporcional a las solicitudes admitidas que aún se encuentran dentro de las ventanas configuradas. Esto es
  más preciso que los cubos ponderados pero usa más memoria. Los límites proporcionan un límite estricto
  en las entradas de marcas de tiempo en vivo, y los objetos de cliente obsoletos se eliminan.

Reparto equitativo:
  La capacidad del inquilino no disputada conserva el trabajo y puede ser utilizada por cualquier
  cliente. Una vez que se alcanza el techo del inquilino, los clientes denegados ingresan a una cola FIFO
  con como máximo una entrada por cliente. A medida que la capacidad expira, la cabeza obtiene la siguiente admisión,
  luego sale de la cola. Un cliente continuamente ocupado debe volver a unirse detrás de otros contendientes,
  implementando la redistribución de ronda sin reservar permanentemente acciones no utilizadas por cliente.
  Las entradas de cola inactivas expiran después de active_timeout_ms para que un turno abandonado no pueda bloquear
  al inquilino para siempre.

Comportamiento del reloj:
  now_ms debe provenir de un reloj monótono. Si un llamador proporciona una marca de tiempo duplicada
  o decreciente, el tiempo se limita al mayor tiempo observado por el inquilino.
  Esto evita que el historial caducado vuelva a estar activo.

Suposiciones de concurrencia y despliegue:
  Un único bloqueo reentrante hace que la configuración, la admisión y la limpieza sean atómicas
  entre hilos y tareas asíncronas que comparten este objeto. Para uso distribuido,
  la misma transición de estado se puede implementar atómicamente en Redis con un
  script Lua y conjuntos ordenados, o en un almacén fuertemente consistente transaccional.
  Los metadatos del observador FIFO deben actualizarse en la misma transacción que los registros de solicitudes;
  los bloqueos locales solos no coordinan múltiples procesos.
"""

def __init__(self, default_config: Optional[TenantConfig] = None) -> None:
    if default_config is not None:
        default_config.validate()
    self._default_config = default_config
    self._configs: Dict[Hashable, TenantConfig] = {}
    self._tenants: Dict[Hashable, _TenantState] = {}
    self._lock = threading.RLock()

def configure_tenant(self, tenant_id: Hashable, config: TenantConfig) -> None:
    """Crea o reemplaza la configuración de un inquilino atómicamente."""
    config.validate()
    with self._lock:
        self._configs[tenant_id] = config

def allow(self, tenant_id: Hashable, client_id: Hashable, now_ms: int) -> AllowResult:
    """Decide atómicamente si admitir una solicitud.

    retry_after_ms es exacto para una denegación directa de cliente/ventana o inquilino/ventana.
    Para una denegación de justicia FIFO, es el tiempo de reintento temporal más temprano;
    la admisión real también depende de que los observadores activos anteriores tomen o
    abandonen sus turnos.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms debe ser un entero")

    with self._lock:
        cfg = self._config_for(tenant_id)
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            tenant = _TenantState(last_now_ms=now_ms, last_seen_ms=now_ms)
            self._tenants[tenant_id] = tenant

        now = self._normalize_time(tenant, now_ms)
        tenant.last_seen_ms = now
        self._prune(tenant.admitted, now, cfg.tenant_window_ms)

        client = tenant.clients.get(client_id)
        if client is None:
            client = _ClientState(last_seen_ms=now)
            tenant.clients[client_id] = client
        client.last_seen_ms = now
        self._prune(client.admitted, now, cfg.client_window_ms)

        self._purge_waiters(tenant, cfg, now)
        self._evict_clients(tenant, cfg, now, preserve=client_id)

        if len(client.admitted) >= cfg.client_limit:
            self._remove_waiter(tenant, client_id)
            retry = client.admitted[0] + cfg.client_window_ms - now
            return AllowResult(False, max(0, retry))

        if len(tenant.admitted) >= cfg.tenant_limit:
            self._enqueue_waiter(tenant, client_id)
            retry = tenant.admitted[0] + cfg.tenant_window_ms - now
            return AllowResult(False, max(0, retry))

        # Existe capacidad. Durante la contención, solo la cabeza FIFO puede tomarla.
        if tenant.waiters and tenant.waiters[0] != client_id:
            self._enqueue_waiter(tenant, client_id)
            return AllowResult(False, 0)

        if tenant.waiters and tenant.waiters[0] == client_id:
            tenant.waiters.popleft()
            tenant.waiter_set.remove(client_id)

        client.admitted.append(now)
        tenant.admitted.append(now)
        return AllowResult(True, 0)

def sweep(self, now_ms: int) -> int:
    """Elimina clientes obsoletos y estados de ejecución de inquilinos vacíos.

    Las aplicaciones pueden llamar a esto periódicamente cuando el tráfico es bajo.
    Las llamadas normales de admisión también limpian el inquilino que se está accediendo.
    Devuelve el número de objetos de cliente e inquilino en tiempo de ejecución eliminados.
    Las configuraciones se conservan.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms debe ser un entero")

    removed = 0
    with self._lock:
        for tenant_id in list(self._tenants):
            cfg = self._config_for(tenant_id)
            tenant = self._tenants[tenant_id]
            now = self._normalize_time(tenant, now_ms)
            self._prune(tenant.admitted, now, cfg.tenant_window_ms)
            self._purge_waiters(tenant, cfg, now)
            before = len(tenant.clients)
            self._evict_clients(tenant, cfg, now, preserve=None)
            removed += before - len(tenant.clients)

            if (not tenant.clients and not tenant.admitted and not tenant.waiters
                    and now - tenant.last_seen_ms >= cfg.eviction_ms):
                del self._tenants[tenant_id]
                removed += 1
    return removed

def debug_state(self, tenant_id: Hashable) -> dict[str, int]:
    """Devuelve tamaños de estado agregados, útiles para monitoreo y pruebas."""
    with self._lock:
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            return {"clients": 0, "tenant_events": 0, "waiters": 0,
                    "client_events": 0}
        return {
            "clients": len(tenant.clients),
            "tenant_events": len(tenant.admitted),
            "waiters": len(tenant.waiters),
            "client_events": sum(len(c.admitted) for c in tenant.clients.values()),
        }

def _config_for(self, tenant_id: Hashable) -> TenantConfig:
    cfg = self._configs.get(tenant_id, self._default_config)
    if cfg is None:
        raise KeyError(f"el inquilino {tenant_id!r} no tiene configuración")
    return cfg

@staticmethod
def _normalize_time(tenant: _TenantState, supplied: int) -> int:
    if tenant.last_now_ms is None or supplied >= tenant.last_now_ms:
        tenant.last_now_ms = supplied
        return supplied
    return tenant.last_now_ms

@staticmethod
def _prune(events: Deque[int], now: int, window_ms: int) -> None:
    boundary = now - window_ms
    while events and events[0] <= boundary:
        events.popleft()

@staticmethod
def _enqueue_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        tenant.waiters.append(client_id)
        tenant.waiter_set.add(client_id)

@staticmethod
def _remove_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        return
    tenant.waiters = deque(x for x in tenant.waiters if x != client_id)
    tenant.waiter_set.remove(client_id)

def _purge_waiters(self, tenant: _TenantState, cfg: TenantConfig, now: int) -> None:
    kept: Deque[Hashable] = deque()
    kept_set: set[Hashable] = set()
    for client_id in tenant.waiters:
        client = tenant.clients.get(client_id)
        if client is None or now - client.last_seen_ms >= cfg.active_timeout_ms:
            continue
        self._prune(client.admitted, now, cfg.client_window_ms)
        # Un cliente bloqueado por su propia cuota no reserva un turno de inquilino.
        if len(client.admitted) >= cfg.client_limit:
            continue
        kept.append(client_id)
        kept_set.add(client_id)
    tenant.waiters = kept
    tenant.waiter_set = kept_set

@staticmethod
def _evict_clients(tenant: _TenantState, cfg: TenantConfig, now: int,
                   preserve: Optional[Hashable]) -> None:
    for client_id in list(tenant.clients):
        if preserve is not None and client_id == preserve:
            continue
        client = tenant.clients[client_id]
        boundary = now - cfg.client_window_ms
        while client.admitted and client.admitted[0] <= boundary:
            client.admitted.popleft()
        if (not client.admitted and client_id not in tenant.waiter_set
                and now - client.last_seen_ms >= cfg.eviction_ms):
            del tenant.clients[client_id]

class SlidingWindowRateLimiterTests(unittest.TestCase):
def make_limiter(self, **overrides: int) -> SlidingWindowRateLimiter:
values = dict(client_limit=3, client_window_ms=1000,
tenant_limit=20, tenant_window_ms=1000,
active_timeout_ms=1000, eviction_ms=2000)
values.update(overrides)
return SlidingWindowRateLimiter(TenantConfig(**values))

def test_exact_window_boundary_is_expired(self) -> None:
    limiter = self.make_limiter(client_limit=1)
    self.assertTrue(limiter.allow("t", "c", 0).allowed)
    denied = limiter.allow("t", "c", 999)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1)
    self.assertTrue(limiter.allow("t", "c", 1000).allowed)

def test_idle_client_returns_after_full_window(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 10).allowed)
    self.assertTrue(limiter.allow("t", "c", 11).allowed)
    self.assertFalse(limiter.allow("t", "c", 12).allowed)
    self.assertTrue(limiter.allow("t", "c", 1011).allowed)

def test_duplicate_and_backwards_clock_are_clamped(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    denied = limiter.allow("t", "c", 50)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1000)
    self.assertTrue(limiter.allow("t", "c", 1100).allowed)

def test_concurrent_requests_are_atomic(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=100)
    barrier = threading.Barrier(40)
    results: list[bool] = []
    results_lock = threading.Lock()

    def worker() -> None:
        barrier.wait()
        value = limiter.allow("t", "same-client", 500).allowed
        with results_lock:
            results.append(value)

    threads = [threading.Thread(target=worker) for _ in range(40)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    self.assertEqual(sum(results), 10)
    self.assertEqual(len(results), 40)

def test_tenant_cap_combines_all_clients(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "b", 1).allowed)
    denied = limiter.allow("t", "c", 2)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 998)

def test_saturated_tenant_uses_fifo_fair_redistribution(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "a", 1).allowed)

    self.assertFalse(limiter.allow("t", "b", 2).allowed)
    self.assertFalse(limiter.allow("t", "c", 3).allowed)

    # Ambas ranuras antiguas expiran, pero a no puede monopolizarlas: b y c en cola.
    self.assertFalse(limiter.allow("t", "a", 1001).allowed)
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)
    self.assertFalse(limiter.allow("t", "b", 1001).allowed)
    self.assertTrue(limiter.allow("t", "c", 1001).allowed)

def test_abandoned_fairness_turn_eventually_expires(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=1,
                                active_timeout_ms=1000, eviction_ms=2000)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertFalse(limiter.allow("t", "gone", 1).allowed)
    # A las 1001 el observador está inactivo, por lo que otro cliente puede usar la capacidad.
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)

def test_eviction_removes_stale_but_not_active_clients(self) -> None:
    limiter = self.make_limiter(client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=200)
    self.assertTrue(limiter.allow("t", "stale", 0).allowed)
    self.assertTrue(limiter.allow("t", "active", 150).allowed)

    limiter.sweep(201)
    state = limiter.debug_state("t")
    self.assertEqual(state["clients"], 1)
    self.assertEqual(state["client_events"], 1)

    # La solicitud activa no se descarta antes de que expire su propia ventana.
    self.assertFalse(limiter.allow("t", "active", 201).allowed
                     if limiter._default_config.client_limit == 1 else False)
    self.assertEqual(limiter.debug_state("t")["clients"], 1)
    limiter.sweep(351)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

def test_configuration_is_per_tenant(self) -> None:
    limiter = SlidingWindowRateLimiter()
    limiter.configure_tenant("small", TenantConfig(1, 100, 1, 100, 100, 100))
    limiter.configure_tenant("large", TenantConfig(2, 100, 3, 100, 100, 100))
    self.assertTrue(limiter.allow("small", "c", 0).allowed)
    self.assertFalse(limiter.allow("small", "c", 1).allowed)
    self.assertTrue(limiter.allow("large", "c", 0).allowed)
    self.assertTrue(limiter.allow("large", "c", 1).allowed)

def test_memory_is_bounded_by_live_limits_and_sweep(self) -> None:
    limiter = self.make_limiter(client_limit=5, tenant_limit=7,
                                client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=100)
    for index in range(100):
        limiter.allow("t", f"c{index}", 0)
    state = limiter.debug_state("t")
    self.assertLessEqual(state["tenant_events"], 7)
    self.assertLessEqual(state["client_events"], 7)
    limiter.sweep(100)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

if name == "main":
unittest.main()

import time
import threading
from collections import deque
import unittest
from typing import Dict, Tuple, Optional

Clase de datos auxiliar para la configuración del inquilino

class TenantConfig:
def init(self, client_rate: int, client_window_ms: int, tenant_rate: int, tenant_window_ms: int):
self.client_rate = client_rate
self.client_window_ms = client_window_ms
self.tenant_rate = tenant_rate
self.tenant_window_ms = tenant_window_ms

class SlidingWindowRateLimiter:
"""
Un limitador de tasa multiinquilino, seguro para subprocesos, que utiliza un algoritmo de registro de ventana deslizante.

Esta implementación proporciona limitación de tasa por cliente bajo un límite global por inquilino.
Utiliza una política de uso compartido justo en la que ningún cliente individual puede agotar la cuota del inquilino
debido a su propio límite individual.

Algoritmo: Registro de ventana deslizante
- Para cada cliente e inquilino, mantenemos un deque (un contenedor similar a una lista con adiciones
  y extracciones rápidas desde ambos extremos) de marcas de tiempo de las solicitudes recibidas dentro de la ventana.
- Cuando llega una nueva solicitud, primero descartamos todas las marcas de tiempo anteriores al tiempo actual
  menos el tamaño de la ventana.
- Luego, verificamos si el número de marcas de tiempo restantes es menor que el límite de tasa configurado.
- Si se permite la solicitud, la marca de tiempo actual se añade al registro.

Compensaciones:
- Precisión: Este algoritmo es perfectamente preciso. Rastrea correctamente el conteo de solicitudes
  dentro de la ventana deslizante precisa, evitando problemas con contadores de ventana fija donde
  las ráfagas en el límite de las ventanas pueden exceder la tasa.
- Uso de memoria: La huella de memoria es proporcional al límite de tasa, ya que almacena una
  marca de tiempo por solicitud permitida dentro de la ventana. Para un cliente con una tasa de 1000, almacenará hasta 1000 marcas de tiempo (aprox. 8 KB por cliente). Esto puede ser significativo para
  sistemas con una cantidad muy grande de clientes activos o límites de tasa muy altos.

Supuestos:
- Despliegue en un solo proceso: El estado se almacena en memoria dentro de la instancia de la clase. Es
  seguro para subprocesos para su uso dentro de un único proceso de aplicación.
- Reloj monótonamente creciente: Se espera que el parámetro `now_ms` sea una marca de tiempo monótonamente
  creciente en milisegundos. El limitador se protege contra retrocesos en el tiempo
  ignorando solicitudes con marcas de tiempo anteriores a la más recientemente procesada.

Extensión distribuida:
Para admitir un entorno distribuido (por ejemplo, múltiples servidores API), el estado debe
externalizarse a un almacén central de baja latencia como Redis.
- Almacenamiento de estado: Podría utilizarse un Sorted Set de Redis para cada registro de cliente/inquilino, con tanto la
  puntuación como el valor siendo la marca de tiempo de la solicitud. Esto permite consultas de rango eficientes.
- Atomicidad: Las transacciones de Redis (MULTI/EXEC) o los scripts Lua serían esenciales para realizar
  la operación de comprobar y actualizar (purgar marcas de tiempo antiguas, comprobar conteo, añadir nueva marca de tiempo)
  de forma atómica, evitando condiciones de carrera entre distintos servidores.
- Evicción: La expiración de claves integrada de Redis (`EXPIRE`) gestionaría de forma natural la evicción
  de datos obsoletos de clientes.
"""

def __init__(self):
    self._tenants: Dict[str, Dict] = {}
    self._lock = threading.Lock()  # Bloqueo global para añadir/eliminar inquilinos

def add_tenant(self, tenant_id: str, config: TenantConfig):
    """Añade o actualiza la configuración de un inquilino."""
    with self._lock:
        if tenant_id not in self._tenants:
            self._tenants[tenant_id] = {
                "config": config,
                "clients": {},
                "tenant_log": deque(),
                "lock": threading.Lock() # Bloqueo por inquilino para seguridad de subprocesos
            }
        else:
            # Actualizar la configuración sobre la marcha
            self._tenants[tenant_id]['config'] = config

def allow(self, tenant_id: str, client_id: str, now_ms: Optional[int] = None) -> Tuple[bool, int]:
    """
    Determina si una solicitud está permitida para un inquilino y cliente dados.

    Args:
        tenant_id: El identificador del inquilino.
        client_id: El identificador del cliente.
        now_ms: La marca de tiempo actual en milisegundos. Si es None, se utiliza la hora del sistema.

    Returns:
        Una tupla (is_allowed, retry_after_ms).
        - is_allowed (bool): True si la solicitud está permitida, False en caso contrario.
        - retry_after_ms (int): Si se deniega, el número de milisegundos que esperar antes de que la
          siguiente solicitud pueda estar permitida. 0 si la solicitud está permitida.
    """
    if now_ms is None:
        now_ms = int(time.time() * 1000)

    with self._lock:
        if tenant_id not in self._tenants:
            raise ValueError(f"El inquilino '{tenant_id}' no está configurado.")
        tenant_state = self._tenants[tenant_id]

    with tenant_state['lock']:
        config = tenant_state['config']

        # 1. Obtener o crear el estado del cliente
        if client_id not in tenant_state['clients']:
            tenant_state['clients'][client_id] = {"log": deque(), "last_seen_ms": 0}
        client_state = tenant_state['clients'][client_id]

        # Defenderse contra retrocesos del reloj
        if now_ms < client_state['last_seen_ms']:
            return (False, 1) # Denegar y sugerir un pequeño retraso para reintentar

        # 2. Purgar y comprobar el límite por cliente
        client_allowed, client_retry_ms = self._check_limit(
            client_state['log'], config.client_rate, config.client_window_ms, now_ms
        )
        if not client_allowed:
            return (False, client_retry_ms)

        # 3. Purgar y comprobar el límite por inquilino
        tenant_allowed, tenant_retry_ms = self._check_limit(
            tenant_state['tenant_log'], config.tenant_rate, config.tenant_window_ms, now_ms
        )
        if not tenant_allowed:
            return (False, tenant_retry_ms)

        # 4. Si ambos están permitidos, registrar la solicitud y permitirla
        client_state['log'].append(now_ms)
        client_state['last_seen_ms'] = now_ms
        tenant_state['tenant_log'].append(now_ms)

        # 5. Limpieza oportunista de clientes obsoletos para este inquilino
        self._cleanup_stale_clients(tenant_state, now_ms)

        return (True, 0)

def _check_limit(self, log: deque, rate: int, window_ms: int, now_ms: int) -> Tuple[bool, int]:
    """Auxiliar para purgar un registro de marcas de tiempo y comprobarlo contra un límite de tasa."""
    # Purgar marcas de tiempo antiguas
    window_start_ms = now_ms - window_ms
    while log and log[0] <= window_start_ms:
        log.popleft()

    # Comprobar si se excede el límite
    if len(log) >= rate:
        oldest_timestamp = log[0]
        retry_after_ms = (oldest_timestamp + window_ms) - now_ms
        return (False, max(1, retry_after_ms)) # Garantizar que el reintento sea de al menos 1 ms

    return (True, 0)

def _cleanup_stale_clients(self, tenant_state: Dict, now_ms: int):
    """
    Elimina los clientes que han estado inactivos durante mucho tiempo.
    Un cliente inactivo es aquel cuya última solicitud es más antigua que el doble del tamaño de la ventana.
    Esta es una limpieza oportunista, ejecutada en línea con una solicitud.
    """
    config = tenant_state['config']
    stale_threshold_ms = config.client_window_ms * 2
    stale_clients = [
        cid for cid, cstate in tenant_state['clients'].items()
        if now_ms - cstate['last_seen_ms'] > stale_threshold_ms
    ]
    for client_id in stale_clients:
        del tenant_state['clients'][client_id]

--- Conjunto de pruebas ---

class TestSlidingWindowRateLimiter(unittest.TestCase):

def setUp(self):
    self.limiter = SlidingWindowRateLimiter()
    self.tenant_config = TenantConfig(
        client_rate=5,
        client_window_ms=1000,
        tenant_rate=15,
        tenant_window_ms=1000
    )
    self.limiter.add_tenant('test_tenant', self.tenant_config)
    self.now = int(time.time() * 1000)

def test_basic_client_limit(self):
    """Un solo cliente debería ser limitado después de exceder su tasa."""
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + i)
        self.assertTrue(allowed)

    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 5)
    self.assertFalse(allowed)
    self.assertGreater(retry_after, 990) # Debería estar alrededor de 1000 ms - 5 ms

def test_window_sliding(self):
    """Las solicitudes deberían permitirse de nuevo después de que haya pasado la ventana."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)

    # Esta solicitud debería ser denegada
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 6)
    self.assertFalse(allowed)

    # Después de que pase la ventana de la primera solicitud, se permite una nueva
    # La primera solicitud fue en self.now, la ventana es de 1000 ms.
    # En self.now + 1001, la primera solicitud expira.
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 1001)
    self.assertTrue(allowed, "La solicitud debería permitirse después de que la ventana se deslice")

def test_idle_client_returns(self):
    """La cuota de un cliente inactivo debería estar renovada cuando vuelva."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)
    
    # El cliente queda inactivo durante más de una ventana
    new_time = self.now + 2000
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + i)
        self.assertTrue(allowed, "Las solicitudes deberían permitirse para un cliente que regresa")
    
    allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + 5)
    self.assertFalse(allowed, "El cliente debería ser limitado de nuevo después de usar la nueva cuota")

def test_tenant_limit_and_fairness(self):
    """Múltiples clientes deberían alcanzar colectivamente el límite del inquilino, demostrando equidad."""
    # 3 clientes, cada uno hace 5 solicitudes. Total 15 solicitudes, que es el límite del inquilino.
    for i in range(5):
        # Cliente 1
        self.assertTrue(self.limiter.allow('test_tenant', 'c1', self.now + i)[0])
        # Cliente 2
        self.assertTrue(self.limiter.allow('test_tenant', 'c2', self.now + i)[0])
        # Cliente 3
        self.assertTrue(self.limiter.allow('test_tenant', 'c3', self.now + i)[0])

    # En este punto, el registro del inquilino tiene 15 solicitudes. La siguiente solicitud de cualquier cliente debería fallar.
    self.assertFalse(self.limiter.allow('test_tenant', 'c4', self.now + 6)[0])
    self.assertFalse(self.limiter.allow('test_tenant', 'c1', self.now + 6)[0])

    # Probar equidad: un cliente alcanza su límite, pero otro puede continuar.
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('fair_tenant', TenantConfig(client_rate=2, client_window_ms=1000, tenant_rate=10, tenant_window_ms=1000))
    # El cliente acaparador usa toda su cuota
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now)[0])
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now + 1)[0])
    self.assertFalse(limiter.allow('fair_tenant', 'hog', self.now + 2)[0]) # El acaparador está bloqueado

    # Otro cliente aún debería poder hacer solicitudes
    self.assertTrue(limiter.allow('fair_tenant', 'fair_client', self.now + 3)[0])

def test_stale_client_eviction(self):
    """Los clientes obsoletos deberían eliminarse de la memoria."""
    self.limiter.allow('test_tenant', 'stale_client', self.now)
    self.assertIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

    # Hacer una nueva solicitud en un futuro lejano, desencadenando la limpieza
    future_time = self.now + self.tenant_config.client_window_ms * 3
    self.limiter.allow('test_tenant', 'active_client', future_time)

    self.assertNotIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

def test_concurrency(self):
    """Debería gestionar correctamente solicitudes concurrentes al mismo cliente."""
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('concurrent_tenant', TenantConfig(client_rate=100, client_window_ms=1000, tenant_rate=1000, tenant_window_ms=1000))
    start_time = int(time.time() * 1000)
    num_threads = 10
    reqs_per_thread = 15
    allowed_count = [0]
    lock = threading.Lock()

    def worker():
        for _ in range(reqs_per_thread):
            now = int(time.time() * 1000)
            allowed, _ = limiter.allow('concurrent_tenant', 'shared_client', now)
            if allowed:
                with lock:
                    allowed_count[0] += 1
            time.sleep(0.001) # pequeña pausa para aumentar la probabilidad de contención

    threads = [threading.Thread(target=worker) for _ in range(num_threads)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # El número de solicitudes permitidas debería ser exactamente el límite de tasa del cliente.
    self.assertEqual(allowed_count[0], 100)

def test_clock_regression(self):
    """Debería denegar solicitudes si el reloj retrocede."""
    self.assertTrue(self.limiter.allow('test_tenant', 'client1', self.now + 100)[0])
    # Una solicitud con una marca de tiempo anterior debería ser denegada
    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 99)
    self.assertFalse(allowed)
    self.assertEqual(retry_after, 1)

if name == 'main':
unittest.main()