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:
"""Rate limits for one tenant.

client_limit applies independently to each client, while tenant_limit applies
to all admitted requests from the tenant. Both limits use exact rolling
windows. active_timeout_ms bounds how long an absent fairness waiter can
retain its turn. eviction_ms controls client-state retention.
"""

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("limits must be positive")
    if self.client_window_ms <= 0 or self.tenant_window_ms <= 0:
        raise ValueError("window lengths must be positive")
    if self.active_timeout_ms <= 0:
        raise ValueError("active_timeout_ms must be positive")
    if self.eviction_ms < max(self.client_window_ms, self.tenant_window_ms):
        raise ValueError("eviction_ms must be at least as large as both windows")
    if self.eviction_ms < self.active_timeout_ms:
        raise ValueError("eviction_ms must be at least 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:
"""Thread-safe, single-process, exact sliding-window rate limiter.

Sliding-window algorithm:
  Every admitted request timestamp is stored in a deque. Before evaluating a
  request, timestamps t satisfying t <= now - window are removed. Thus a
  request exactly one full window old no longer consumes capacity. This is a
  true log-based sliding window, so it has no calendar-bucket boundary burst.

  The result is exact, with O(1) amortized deque operations and memory
  proportional to admitted requests still inside configured windows. This is
  more accurate than weighted buckets but uses more memory. Limits provide a
  hard bound on live timestamp entries, and stale client objects are evicted.

Fair sharing:
  Uncontended tenant capacity is work-conserving and can be used by any
  client. Once the tenant ceiling is reached, denied clients enter a FIFO
  queue with at most one entry per client. As capacity expires, the head gets
  the next admission, then leaves the queue. A continuously busy client must
  rejoin behind other contenders, implementing round-robin redistribution
  without permanently reserving unused per-client shares. Inactive queue
  entries expire after active_timeout_ms so an abandoned turn cannot block
  the tenant forever.

Clock behavior:
  now_ms should come from a monotonic clock. If a caller supplies a duplicate
  or decreasing timestamp, time is clamped to the tenant's greatest observed
  timestamp. This prevents expired history from becoming live again.

Concurrency and deployment assumptions:
  A single re-entrant lock makes allow, configuration, and sweeping atomic
  across threads and async tasks sharing this object. For distributed use,
  the same state transition can be implemented atomically in Redis with a
  Lua script and sorted sets, or in a transactional strongly consistent
  store. FIFO waiter metadata must be updated in the same transaction as the
  request logs; local locks alone do not coordinate multiple processes.
"""

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:
    """Create or replace a tenant's configuration atomically."""
    config.validate()
    with self._lock:
        self._configs[tenant_id] = config

def allow(self, tenant_id: Hashable, client_id: Hashable, now_ms: int) -> AllowResult:
    """Atomically decide whether to admit one request.

    retry_after_ms is exact for a direct client/window or tenant/window
    denial. For a FIFO fairness denial it is the earliest temporal retry;
    actual admission also depends on earlier active waiters taking or
    abandoning their turns.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    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))

        # Capacity exists. During contention, only the FIFO head may take it.
        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:
    """Evict stale clients and empty tenant runtime states.

    Applications can call this periodically when traffic is low. Normal
    allow calls also clean the tenant being accessed. Returns the number of
    client and tenant runtime objects removed. Configurations are retained.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    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]:
    """Return aggregate state sizes, useful for monitoring and tests."""
    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"tenant {tenant_id!r} has no configuration")
    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)
        # A client blocked by its own quota does not reserve a tenant turn.
        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)

    # Both old slots expire, but a cannot monopolize them: b and c queued.
    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)
    # At 1001 the waiter is inactive, so another client may use capacity.
    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)

    # The active request is not dropped before its own window expires.
    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

Helper data class for tenant configuration

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:
"""
A thread-safe, multi-tenant rate limiter using a sliding window log algorithm.

This implementation provides per-client rate limiting under a global per-tenant cap.
It uses a fair-sharing policy where no single client can exhaust the tenant's quota
due to its own individual limit.

Algorithm: Sliding Window Log
- For each client and tenant, we maintain a deque (a list-like container with fast appends
  and pops from both ends) of timestamps of requests received within the window.
- When a new request arrives, we first discard all timestamps older than the current
  time minus the window size.
- Then, we check if the number of remaining timestamps is less than the configured rate limit.
- If the request is allowed, the current timestamp is added to the log.

Tradeoffs:
- Accuracy: This algorithm is perfectly accurate. It correctly tracks the request count
  within the precise sliding window, avoiding issues with fixed-window counters where
  bursts at the boundary of windows can exceed the rate.
- Memory Usage: The memory footprint is proportional to the rate limit, as it stores one
  timestamp per allowed request within the window. For a client with a rate of 1000, it
  will store up to 1000 timestamps (approx. 8KB per client). This can be significant for
  systems with a very large number of active clients or very high rate limits.

Assumptions:
- Single-Process Deployment: State is stored in memory within the class instance. It is
  thread-safe for use within a single application process.
- Monotonically Increasing Clock: The `now_ms` parameter is expected to be a monotonically
  increasing timestamp in milliseconds. The limiter protects itself against time going
  backwards by ignoring requests with timestamps older than the most recently processed one.

Distributed Extension:
To support a distributed environment (e.g., multiple API servers), the state must be
externalized to a central, low-latency store like Redis.
- State Storage: A Redis Sorted Set could be used for each client/tenant log, with both the
  score and value being the request timestamp. This allows for efficient range queries.
- Atomicity: Redis transactions (MULTI/EXEC) or Lua scripts would be essential to perform
  the check-and-update operation (prune old timestamps, check count, add new timestamp)
  atomically, preventing race conditions between different servers.
- Eviction: Redis's built-in key expiration (`EXPIRE`) would naturally handle the eviction
  of stale client data.
"""

def __init__(self):
    self._tenants: Dict[str, Dict] = {}
    self._lock = threading.Lock()  # Global lock for adding/removing tenants

def add_tenant(self, tenant_id: str, config: TenantConfig):
    """Adds or updates a tenant's configuration."""
    with self._lock:
        if tenant_id not in self._tenants:
            self._tenants[tenant_id] = {
                "config": config,
                "clients": {},
                "tenant_log": deque(),
                "lock": threading.Lock() # Per-tenant lock for thread safety
            }
        else:
            # Update config on the fly
            self._tenants[tenant_id]['config'] = config

def allow(self, tenant_id: str, client_id: str, now_ms: Optional[int] = None) -> Tuple[bool, int]:
    """
    Determines if a request is allowed for a given tenant and client.

    Args:
        tenant_id: The identifier for the tenant.
        client_id: The identifier for the client.
        now_ms: The current timestamp in milliseconds. If None, the system time is used.

    Returns:
        A tuple (is_allowed, retry_after_ms).
        - is_allowed (bool): True if the request is permitted, False otherwise.
        - retry_after_ms (int): If denied, the number of milliseconds to wait before the
          next request might be allowed. 0 if the request is allowed.
    """
    if now_ms is None:
        now_ms = int(time.time() * 1000)

    with self._lock:
        if tenant_id not in self._tenants:
            raise ValueError(f"Tenant '{tenant_id}' not configured.")
        tenant_state = self._tenants[tenant_id]

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

        # 1. Get or create client state
        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]

        # Defend against clock going backwards
        if now_ms < client_state['last_seen_ms']:
            return (False, 1) # Deny and suggest a small retry delay

        # 2. Prune and check per-client limit
        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. Prune and check per-tenant limit
        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. If both are allowed, record the request and permit it
        client_state['log'].append(now_ms)
        client_state['last_seen_ms'] = now_ms
        tenant_state['tenant_log'].append(now_ms)

        # 5. Opportunistic cleanup of stale clients for this tenant
        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]:
    """Helper to prune a timestamp log and check it against a rate limit."""
    # Prune old timestamps
    window_start_ms = now_ms - window_ms
    while log and log[0] <= window_start_ms:
        log.popleft()

    # Check if limit is exceeded
    if len(log) >= rate:
        oldest_timestamp = log[0]
        retry_after_ms = (oldest_timestamp + window_ms) - now_ms
        return (False, max(1, retry_after_ms)) # Ensure retry is at least 1ms

    return (True, 0)

def _cleanup_stale_clients(self, tenant_state: Dict, now_ms: int):
    """
    Removes clients that have been inactive for a long time.
    An inactive client is one whose last request is older than twice the window size.
    This is an opportunistic cleanup, run inline with a request.
    """
    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]

--- Test Suite ---

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):
    """A single client should be throttled after exceeding its rate."""
    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) # Should be around 1000ms - 5ms

def test_window_sliding(self):
    """Requests should be allowed again after the window has passed."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)

    # This request should be denied
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 6)
    self.assertFalse(allowed)

    # After the window for the first request passes, a new one is allowed
    # First request was at self.now, window is 1000ms.
    # At self.now + 1001, the first request expires.
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 1001)
    self.assertTrue(allowed, "Request should be allowed after window slides")

def test_idle_client_returns(self):
    """An idle client's quota should be fresh when it returns."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)
    
    # Client goes idle for more than a window
    new_time = self.now + 2000
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + i)
        self.assertTrue(allowed, "Requests should be allowed for a returning client")
    
    allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + 5)
    self.assertFalse(allowed, "Client should be throttled again after using new quota")

def test_tenant_limit_and_fairness(self):
    """Multiple clients should collectively hit the tenant limit, demonstrating fairness."""
    # 3 clients, each makes 5 requests. Total 15 requests, which is the tenant limit.
    for i in range(5):
        # Client 1
        self.assertTrue(self.limiter.allow('test_tenant', 'c1', self.now + i)[0])
        # Client 2
        self.assertTrue(self.limiter.allow('test_tenant', 'c2', self.now + i)[0])
        # Client 3
        self.assertTrue(self.limiter.allow('test_tenant', 'c3', self.now + i)[0])

    # At this point, tenant log has 15 requests. The next request from any client should fail.
    self.assertFalse(self.limiter.allow('test_tenant', 'c4', self.now + 6)[0])
    self.assertFalse(self.limiter.allow('test_tenant', 'c1', self.now + 6)[0])

    # Test fairness: one client hits its limit, but another can proceed.
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('fair_tenant', TenantConfig(client_rate=2, client_window_ms=1000, tenant_rate=10, tenant_window_ms=1000))
    # Hog client uses up its quota
    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]) # Hog is blocked

    # Another client should still be able to make requests
    self.assertTrue(limiter.allow('fair_tenant', 'fair_client', self.now + 3)[0])

def test_stale_client_eviction(self):
    """Stale clients should be removed from memory."""
    self.limiter.allow('test_tenant', 'stale_client', self.now)
    self.assertIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

    # Make a new request far in the future, triggering cleanup
    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):
    """Should handle concurrent requests to the same client correctly."""
    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) # small sleep to increase chance of contention

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

    # The number of allowed requests should be exactly the client rate limit.
    self.assertEqual(allowed_count[0], 100)

def test_clock_regression(self):
    """Should deny requests if the clock moves backwards."""
    self.assertTrue(self.limiter.allow('test_tenant', 'client1', self.now + 100)[0])
    # A request with an earlier timestamp should be denied
    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()