# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-token-bucket-and-atomic-quota-gates-must-an-instance-enforce-multi-tenant-rate-limiting-to-prevent-noisy-neighbor-starvation-and-boundary-burst-amplification.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-token-bucket-and-atomic-quota-gates-must-an-instance-enforce-multi-tenant-rate-limiting-to-prevent-noisy-neighbor-starvation-and-boundary-burst-amplification.warrant.json
# Markdown SHA-256:   301e999b9f1cee33fa515d32d7d38c53703105406d977ad93a2f557768dd0d6b
# Code AST SHA-256:   e4f0f9c039e97ac2ea29099db1514655e0bdf9cc0450ca18d6d69cc7e437d8e3
# Verified Classes:   MultiTenantQuotaExceededError, _TenantBucket, MultiTenantTokenBucketGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Final


class MultiTenantQuotaExceededError(RuntimeError):
    """Raised when a tenant exhausts its token bucket or breaches global concurrency headroom."""
    pass


@dataclass
class _TenantBucket:
    tokens: float
    last_refill_monotonic: float


class MultiTenantTokenBucketGate:
    """
    Enforces per-tenant continuous token-bucket rate limiting (preventing fixed-window boundary bursts):
    1. Refills tokens continuously based on monotonic elapsed time, capped at `burst_capacity`.
    2. Deducts cost atomically; rejects requests before execution when `tokens < cost`.
    """
    def __init__(self, refill_rate_per_sec: float = 10.0, burst_capacity: float = 30.0) -> None:
        if refill_rate_per_sec <= 0 or burst_capacity <= 0:
            raise ValueError("refill_rate_per_sec and burst_capacity must be positive.")
        self.refill_rate_per_sec: Final[float] = float(refill_rate_per_sec)
        self.burst_capacity: Final[float] = float(burst_capacity)
        self._buckets: dict[str, _TenantBucket] = {}

    def acquire(self, tenant_id: str, cost: float, now_monotonic: float) -> float:
        if not tenant_id or not tenant_id.strip():
            raise ValueError("tenant_id must be non-empty.")
        if cost <= 0 or cost > self.burst_capacity:
            raise MultiTenantQuotaExceededError(
                f"Requested cost ({cost}) is invalid or exceeds burst capacity ({self.burst_capacity})."
            )

        bucket = self._buckets.get(tenant_id)
        if bucket is None:
            bucket = _TenantBucket(tokens=self.burst_capacity, last_refill_monotonic=now_monotonic)
            self._buckets[tenant_id] = bucket
        else:
            elapsed = max(0.0, now_monotonic - bucket.last_refill_monotonic)
            bucket.tokens = min(
                self.burst_capacity,
                bucket.tokens + elapsed * self.refill_rate_per_sec,
            )
            bucket.last_refill_monotonic = now_monotonic

        if bucket.tokens < cost:
            deficit = cost - bucket.tokens
            retry_after = deficit / self.refill_rate_per_sec
            raise MultiTenantQuotaExceededError(
                f"Tenant {tenant_id!r} rate-limited. Retry after {retry_after:.2f}s."
            )

        bucket.tokens -= cost
        return bucket.tokens