回答A: OpenAI GPT-5.6
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_wind...
全文を表示 ▼
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()
判定
勝利票
3 / 3
平均スコア
総合点
総評
回答Aは、レートリミッターの包括的かつ堅牢な実装を提供しており、特にテナント飽和のための複雑なフェアシェアリングポリシーを含む、すべての機能要件に完全に対応しています。コードはよく構造化されており、適切なデータ構造を使用し、徹底したドキュメントと指定されたすべてのエッジケースをカバーする強力なテストスイートが含まれています。明示的なスイープと設定可能なエビクションによるメモリ管理はよく設計されています。
採点詳細を表示 ▼
正確さ
重み 35%回答Aは、特にテナント飽和のための複雑なフェアシェアリングポリシーや堅牢なクロック処理を含む、レートリミッターのすべての側面を正しく実装しています。すべての制限とウィンドウロジックは正確です。
完全性
重み 20%回答Aは、フェアシェアリングと包括的なメモリ管理のエビクションを含む、すべての機能要件に対応しており、非常に包括的です。要求されたとおり、明確な説明、仮定、および分散設計に関する注記を提供しています。
コード品質
重み 20%回答Aのコードは、明確な状態管理のための`dataclasses`を使用し、よく構造化されており、優れたインラインドキュメントを備えています。`threading.RLock`の使用は、再入可能な操作に適しています。メソッドは論理的に分離されており、クリーンです。
実用性
重み 15%回答Aは、特にフェアシェアリングメカニズムと`sweep`メソッドによる制御されたメモリ管理のエビクションなど、完全な機能セットにより、高い実用価値を提供します。`debug_state`メソッドも、監視のための便利な追加機能です。
指示遵守
重み 10%回答Aは、複雑なフェアシェアリングポリシー、真のスライディングウィンドウ、並行性安全性、メモリバウンディング、およびコードとテストの両方で指定されたすべてのエッジケースへの対応を含む、すべての指示を綿密にフォローしています。すべての成果物が提供されています。
総合点
総評
回答Aは、ログベースの正確なスライディングウィンドウ、テナントごとの設定、テナント全体のキャップ、同時実行保護、古い状態のクリーンアップ、リトライ計算、明示的なクロック回帰処理、および広範な単体テストスイートを備えた、実質的に完全で実行可能なPython実装を提供します。そのFIFOウェイターメカニズムは、テナントの飽和状態下での信頼できる公平共有ポリシーですが、公平性キュー拒否のretry_after_msは不正確になる可能性があり、多数の拒否されたユニーククライアントのウェイター/クライアント状態は、タイムアウトまたはスイープまで成長する可能性があります。全体として、要求されたライブラリの動作とエッジケースに密接に一致しています。
採点詳細を表示 ▼
正確さ
重み 35%正確なタイムスタンプログと正しい境界プルーニングを使用し、クライアントごとおよびテナントごとの両方のキャップを強制し、過去の時間をクランプし、ミューテーションを安全にシリアル化します。FIFO公平性ロジックはほとんど正しいですが、ヘッド以外の公平性拒否に対するretry_after_msは0になる可能性があり、多数の待機中の拒否されたクライアントのメモリは、即時にはバウンドされず、時間とともにバウンドされます。
完全性
重み 20%要求されたAPI、テナントごとの調整可能な設定、正確なスライディングウィンドウ、テナントキャップ、公平性キュー、同時実行安全性、エビクション/スイープ、仮定、分散拡張に関する注記、および指定されたエッジケースのほぼすべてに対するテストをカバーしています。マイナーなギャップには、不完全な公平性リトライレポートと、やや扱いにくいエビクションテストコードが含まれます。
コード品質
重み 20%構造化されたデータクラス、明確な公開メソッド、検証、インラインドキュメント、型ヒント、およびカプセル化されたヘルパーメソッド。単一のグローバルロックはシンプルで安全ですが、スケーラビリティを制限します。また、いくつかの内部/テストはやや扱いにくいです。
実用性
重み 15%監視/デバッグ状態、定期的なスイープ、設定可能なテナント、およびRedis/Luaへの分散使用のための信頼できるパスを備えた、再利用可能な単一プロセスのライブラリとして実用的です。グローバルロックとタイムスタンプログのメモリコストは、高スケールのデプロイメントを制限する可能性がありますが、トレードオフは述べられています。
指示遵守
重み 10%要求された成果物（完全なコード、文書化されたアルゴリズムとトレードオフ、明示的な仮定、分散拡張、および指定されたエッジケースのテスト）に密接に従っています。APIはタプルの代わりにAllowResultを返しますが、これはプロンプトの下で許容されます。
総合点
総評
回答Aは、ほぼ完全で本番品質のソリューションです。正確なログベースのスライディングウィンドウを境界セマンティクスで実装し、FIFO待機者キューによる真の公平再分配メカニズム（アクティビティタイムアウト付き）、テナントで制限された過去および重複クロックの処理、アトミック性のための単一RLock、インライン削除と明示的なスイープAPIによるバウンドメモリを備えています。指定された6つのエッジケースすべてにターゲットテストがあり、ドキュメンテーションには精度/メモリのトレードオフ、仮定、具体的な分散拡張が説明されています。弱点は軽微です。1つのテストアサーションは、混乱した三項演算子により空虚であり、公平性拒否時のretry_after_msが0であることは曖昧であり、呼び出しごとの削除スキャンは、多くのクライアントを持つテナントのオーバーヘッドを追加します。
採点詳細を表示 ▼
正確さ
重み 35%ログベースのスライディングウィンドウは正確です。削除はt <= now - windowを使用するため、1ウィンドウ古いリクエストが期限切れになり、境界テストはt=999でretry_after_ms=1、t=1000で許可を検証します。過去/重複クロックは、テナントの最大観測タイムスタンプに制限され、期限切れの履歴が復活するのを防ぎます。FIFO待機者キューは、テナント飽和下でのラウンドロビン再分配を証明的に強制します（飽和テストのトレースによって検証されます：クライアントaはキューに入れられたクライアントbとcがそれぞれ解放されたスロットを取得する間にブロックされます）。単一のRLockによりallowはアトミックになり、40スレッドテストは正確に10を許可します。軽微な欠陥：公平性拒否に対するretry_after_ms=0は曖昧であり、1つの削除テストには、allow呼び出しを実際に実行しない空虚な条件付きアサーションが含まれています。
完全性
重み 20%すべての機能要件が満たされています。テナントごとの設定、正確なスライディングウィンドウ、真のFIFO公平再分配によるテナントグローバルキャップ、スレッドセーフティ、インラインクリーンアップと明示的なスイープAPIによる削除、およびデバッグ状態ヘルパーが含まれています。指定された6つのエッジケースすべてに専用テストがあります（正確な境界、アイドル返却、同時レース、制限された過去/重複クロック、公平再分配による飽和、アクティブクライアントを維持する古い削除）。さらに、放棄された待機者ターンとメモリバウンドもカバーしています。仮定とRedis/Lua分散拡張はドキュメンテーションに記載されています。
コード品質
重み 20%設定と結果のためのフローズンデータクラス、検証済みの設定、型付きシグネチャ、削除、待機者パージ、削除ヘルパーの明確な分離、アルゴリズム、公平性セマンティクス、クロック動作、分散拡張を網羅する異常に徹底的なクラスドキュメンテーションにより、よく構造化されています。弱点：削除テストには、混乱した空虚な三項演算子アサーションが含まれています。呼び出しごとのフルクライアント削除は、すべての許可にO(クライアント)のオーバーヘッドを追加します。_remove_waiterはdequeをO(n)で再構築します。
実用性
重み 15%単一プロセスとしてそのままデプロイ可能です。決定論的な注入タイムスタンプによりテスト可能であり、debug_stateは監視をサポートし、sweepはオペレーターに明示的なメモリ制御を提供し、待機者タイムアウトは放棄されたクライアントがテナントを枯渇させるのを防ぎます。分散移行パス（RedisソートセットとLua、同じトランザクション内の待機者メタデータ付き）は具体的に説明されています。呼び出しごとの削除スキャンは、多くのクライアントを持つテナントにとって主なパフォーマンス上の注意点です。
指示遵守
重み 10%仕様に厳密に従っています。allow(tenant_id, client_id, now_ms)シグネチャ、結果のretry_after_ms、テナントごとの調整可能な設定、明示的な精度/メモリトレードオフの議論を伴う真のログベースのスライディングウィンドウ、飽和下での公平再分配、同時実行安全性、バウンドメモリ、テスト内のすべての6つのエッジケース、および明示的な仮定と分散拡張の注記が含まれています。プロンプトに記載されているほぼすべての成果物とエッジケースが満たされています。