Snowflake 动态档位限流：Token Bucket、默认策略与可控时钟

题目重点：最新电面给 free/pro/enterprise 客户端、运行时改限额与全局回退；本文实现令牌桶、时钟注入、配置变化语义和分布式追问。

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

训练版每个 clientId 独立持有令牌桶，tier 策略由 capacity 与每秒 refill rate 定义，准入一次消耗 1 令牌；capacity≥1、rate≥0。
运行时 capacity 上调不立即赠送令牌，已有 token 按新 capacity 封顶；新 client 第一次请求时桶满。降档则立刻夹到新上限。策略变更的生效时点为下一次 allow。
无 tier 或配置无效走有效的全局默认 Policy；时钟单调且可注入。并发锁与分布式原子更新属于进一步实现要求，示例代码为单线程。

解题路线

从滑动窗口、固定窗口、token bucket 比较：本题要允许可控 burst 且运行时改限额，因此选 token bucket；若业务要求任意 W 秒严格不超过 N 次，需改滑动窗口。
每次 allow 查最新 client tier 和 Policy，读取桶 (tokens,lastTime)，按 elapsed×refill rate 补充并取 min(capacity,…)；若 token≥1，扣 1 并放行，否则拒绝。
配置变更时不要把 capacity 当 refill 倍率，也不要把旧桶自动重置为满桶；通过注入 clock 精确测试边界，不依赖 sleep。
集群化时把每 client 状态放在按 key 分片的 Redis/事务 KV，通过 Lua 或 CAS 把补充、检查、扣减组成原子操作；若跨区本地额度切片，应明确误放上限。
用与实现分开的数值状态模型随机改变 free 档并发送 12 万次请求对照，覆盖默认配置与时间推进。

手动走一遍

free capacity=5、rate=1/s，t=0 连续 5 次通过、第 6 次拒绝；t=2 可再通过 2 次。若此时 capacity 改为 10，桶仍从当前余额继续按新 rate 补充，不瞬间变 10；若降到 2，余额被夹到 2。无档位 client 使用全局默认。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

"""Per-client token buckets with live tier policy lookup and injectable clock."""

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class Policy:
    capacity: float
    refill_per_second: float


class TierLimiter:
    def __init__(self, policies: dict[str, Policy], client_tiers: dict[str, str],
                 default: Policy, clock: Callable[[], float]):
        self.policies = policies
        self.client_tiers = client_tiers
        self.default = default
        self.clock = clock
        self.buckets: dict[str, tuple[float, float]] = {}

    @staticmethod
    def _valid(policy: Policy | None) -> bool:
        return policy is not None and policy.capacity >= 1 and policy.refill_per_second >= 0

    def allow(self, client_id: str) -> bool:
        policy = self.policies.get(self.client_tiers.get(client_id, ""))
        if not self._valid(policy):
            policy = self.default
        if not self._valid(policy):
            raise ValueError("invalid default policy")
        now = self.clock()
        tokens, previous = self.buckets.get(client_id, (policy.capacity, now))
        if now < previous:
            raise ValueError("clock moved backward")
        tokens = min(policy.capacity, tokens + (now - previous) * policy.refill_per_second)
        accepted = tokens >= 1
        if accepted:
            tokens -= 1
        self.buckets[client_id] = (tokens, now)
        return accepted

复杂度

单线程每请求期望 O(1) 时间与每客户端 O(1) 状态，总 O(C) 空间，C 为活跃客户端数。集群方案额外有一次远程原子操作延迟，热点 client 会串行化；空闲状态需 TTL 与回收规则，否则 C 只增长。

易错点与追问

容易失分的边界

补令牌写成 elapsed×capacity，时间流逝后瞬间超发。
改 capacity 时把已有桶设为满桶，让配额更新赠送突发。
测试依靠真实 sleep，速度慢且边界不稳定。
单实例 map 复制到每台机器却没有共享状态，同 client 跨实例突破限制。
缺失配置和错误配置没有可解释的默认策略。

继续深挖

若 tier 改动时要立刻按新 capacity 填满，怎样定义迁移并避免并发双充？
严格滑动 60 秒限额与 token bucket 在边界处有何差异？
Redis 故障时拒绝还是放行，如何按业务风险定策略？
冷热客户端状态 TTL 与审计复盘怎样兼容？
AI 辅助实现时，如何展示算法取舍、审查生成代码并现场覆盖边界测试？