检查调用链重试/退避/超时以防重试风暴，计算延迟表与放大，标记危险配置。含 Python 计算器。
---
name: retry-backoff-policy-calculator
description: 检查调用链上的重试、退避与超时设置以防重试风暴。计算带 full/equal/decorrelated jitter 的指数延迟表、每层最坏延迟、期望尝试次数与请求放大。标记内层超时长于调用方、对非幂等调用重试、对 4xx 重试、忽略 Retry-After、以及缺失上限或截止时间。含 stdlib Python 计算器与 --self-test。
---
# 重试与退避策略计算器
你帮助工程师设置重试、退避与超时策略，使服务更可靠，而不是把小故障放大成大故障。你收集调用链每一跳（浏览器、网关、服务、数据库、第三方 API）的设置，计算它们在失败下真正会做什么，并提出带数字的更安全策略。
## 本技能中的文件
- `scripts/retry_policy.py` - 读取调用链 JSON 并打印延迟表、最坏延迟、期望尝试、放大与标志（仅 Python 3 标准库）
- `references/backoff-math.md` - 退避、抖动、最坏时间、期望尝试、放大与超时嵌套公式
## 工作方式
1. 收集调用链每层的超时、重试次数、退避基数/倍数、抖动类型、是否幂等、可重试状态码。
2. 计算指数退避时间表（含 jitter）。
3. 计算最坏情况端到端延迟与请求放大。
4. 标记：内层超时 > 外层、非幂等重试、4xx 重试、忽略 Retry-After、无最大重试/截止时间、放大过高。
5. 提出更安全的数值策略。
6. 需要时运行 `scripts/retry_policy.py --self-test`。
保留所有代码、JSON 键与脚本原样。以下为原技能完整正文：
# Retry and Backoff Policy Calculator
You help engineers set retry, backoff and timeout policies that make services more reliable instead of turning a small outage into a big one. You collect the settings of every hop in a call chain (browser, gateway, services, databases, third-party APIs), compute what they really do under failure, and propose a safer policy with numbers.
## Files in this skill
- `scripts/retry_policy.py` - reads a call-chain JSON and prints delay schedules, worst-case latency, expected tries, amplification and flags (Python 3 standard library only)
- `references/backoff-math.md` - formulas for backoff, jitter, worst-case time, expected tries, amplification and timeout nesting
- `references/retry-safety-guide.md` - what is safe to retry, idempotency keys, circuit breakers, retry budgets, common library settings
- `templates/retry-policy-review.md` - review report with current chain, findings and proposed policy
- `examples/checkout-chain.json` - a three-layer checkout chain with typical mistakes
## Workflow
### 1. Map the chain
Ask for each hop, outermost first: attempts (total tries), base delay, multiplier, max delay, jitter type, per-try timeout, overall deadline, what it retries on (status codes, "timeout", "network"), whether the call is idempotent, whether an idempotency key is sent, and whether Retry-After is honored. Read them from code or config (SDK clients, HTTP client retry adapters, Envoy/Istio routes, gRPC service config, job queues). Unknown values: use the library default and mark them as assumed.
### 2. Run the calculator
```
python3 scripts/retry_policy.py examples/checkout-chain.json
python3 scripts/retry_policy.py chain.json --failure-rate 0.5 --max-amplification 8
python3 scripts/retry_policy.py chain.json --json > retry-report.json
python3 scripts/retry_policy.py --self-test
```
Exit code 1 means high-severity findings, 2 means bad input.
### 3. Explain the findings
Order: RETRY_UNSAFE and AMPLIFICATION first (they cause duplicate side effects and outages), then INNER_OUTLIVES_CALLER, HOT_LOOP, NO_PER_TRY_TIMEOUT, then medium and low. Explain each with the numbers from the report, for example "one checkout can become 36 calls to the payments provider".
### 4. Propose a fixed policy
Use `references/backoff-math.md` and `references/retry-safety-guide.md`: retry at one layer, full jitter with a cap, inner worst case below the caller's per-try timeout, deadlines passed down, idempotency keys for writes, no retries on 4xx except 408/429, Retry-After honored, a retry budget or circuit breaker. Re-run the script on the proposed JSON and show before and after.
### 5. Write it up
Fill `templates/retry-policy-review.md` and give config snippets for the user's library.
## Example result
For `examples/checkout-chain.json` at 30% failure rate: worst-case amplification 36 calls per checkout (expected 2.74); the gateway layer's 9.3 s worst case outlives the web app's 8 s per-try timeout; the payments layer retries a non-idempotent charge 4 times without an idempotency key and retries 409; no jitter at the gateway; Retry-After ignored on 429/503.
## Rules
- Never recommend retrying a non-idempotent call without an idempotency key.
- Keep numbers honest: label library defaults and guesses as assumptions.
- Prefer fewer retries with good timeouts over many retries.
- Recommend load tests or fault injection to confirm the policy, not just the math.
FILE:references/backoff-math.md
# Backoff and Timeout Math
## Terms
- **attempts**: total tries including the first. attempts 3 = 1 try + 2 retries.
- **base**: first retry delay. **multiplier**: growth factor (usually 2). **cap**: max single delay.
- **per-try timeout**: how long one try may take before it is abandoned.
- **deadline**: total time budget for the call including all tries and delays.
## Delay before retry k (k = 1 .. attempts-1)
d(k) = min(cap, base x multiplier^(k-1))
| Jitter | Delay | Range | Mean |
|---|---|---|---|
| none | d | d | d |
| full | random(0, d) | 0 .. d | d/2 |
| equal | d/2 + random(0, d/2) | d/2 .. d | 0.75 d |
| decorrelated | min(cap, random(base, 3 x previous)) | base .. min(cap, base x 3^k) | simulated |
Full jitter spreads load best and is the default recommendation. Equal jitter keeps a minimum wait. Decorrelated jitter adapts to the previous delay and works well for long-running clients.
## Worst-case duration of one call
worst = attempts x per_try_timeout + sum of max delays, then cut at the deadline if one is set.
Example: 3 tries x 1 s + 100 ms + 200 ms = 3.3 s.
## Expected tries
With an independent failure probability p per try and n attempts:
E[tries] = 1 + p + p^2 + ... + p^(n-1)
p = 0.3, n = 3 -> 1.39. During a full outage p = 1 and E = n.
## Amplification across layers
If layer A retries 3 times and calls B, which retries 3 times and calls C, which retries 4 times, one user request can become 3 x 3 x 4 = 36 calls to C during an outage. This is a retry storm: the retries keep the failing service overloaded, so it cannot recover.
Rules of thumb:
- Retry at one layer only, usually the one closest to the user or the one that owns the idempotency key.
- Inner layers: attempts 1 or 2 with a short deadline.
- Keep worst-case amplification at or below about 10.
- Add a **retry budget**: retries may be at most 10-20% of requests over a sliding window; past that, fail fast.
## Timeout nesting
Each layer's worst case must fit inside its caller's per-try timeout, minus a margin for network and work:
inner_worst <= outer_per_try_timeout - margin
Otherwise the caller abandons the try and retries while the inner layer is still working, which doubles load. Pass a deadline down the chain (for example a `grpc-timeout` or `X-Request-Deadline` header) and stop when it is used up.
## Retry-After
For 429 and 503, servers may send `Retry-After: <seconds>` or an HTTP date. Wait at least that long (use max(Retry-After, computed backoff)), and give up if it is longer than the remaining deadline.
FILE:references/retry-safety-guide.md
# Retry Safety Guide
## What to retry
| Outcome | Retry? | Notes |
|---|---|---|
| Connection refused / reset before the request was sent | Yes | Safe for any method |
| Timeout after the request was sent | Only if idempotent or with an idempotency key | The server may have done the work |
| 408 Request Timeout | Yes, with backoff | |
| 429 Too Many Requests | Yes, after Retry-After | Never faster than the server asks |
| 500 Internal Server Error | Sometimes | Only if the error is known to be transient |
| 502 / 503 / 504 | Yes, with backoff and jitter | Honor Retry-After on 503 |
| 400 / 401 / 403 / 404 / 405 / 410 / 422 | No | Same request fails the same way; fix the request, refresh auth once at most |
| 409 Conflict | Usually no | Exception: an idempotency key that is still in progress, retry after a short wait |
## Idempotency
- GET, HEAD, OPTIONS, PUT and DELETE are idempotent by HTTP semantics, if the server implements them correctly.
- POST and PATCH are not. To retry them, send an **idempotency key** (a UUID per logical operation, e.g. `Idempotency-Key` header) that the server stores with the result for 24 hours or more. Generate the key once, before the first try, and reuse it for every retry.
- Payments, emails, SMS, order creation and stock reservation must never be retried without a key.
## Client checklist
1. Per-try timeout set and shorter than the caller's.
2. Overall deadline set and passed down the chain.
3. Exponential backoff with a cap and full jitter.
4. attempts <= 3 at the edge, 1-2 inside.
5. Retry only on the table's "Yes" rows.
6. Idempotency key on every retried non-idempotent call.
7. Retry budget or circuit breaker so retries stop during an outage.
8. Metrics: tries per request, retry rate, give-up count, by dependency.
9. Log the final error with the attempt count, not every attempt at error level.
## Circuit breaker basics
- Closed: calls flow; count failures in a rolling window.
- Open: after for example 50% failures over at least 20 calls, fail fast for 30 s.
- Half-open: let a few trial calls through; close if they succeed.
A breaker and a retry budget together stop retry storms from deepening an outage.
## Common library settings
- AWS SDKs: standard mode = 3 attempts, full jitter, base 1 s for throttling; adaptive mode adds client-side rate limiting.
- gRPC retry policy: maxAttempts, initialBackoff, maxBackoff, backoffMultiplier, retryableStatusCodes (UNAVAILABLE is the usual one); retries need hedging/transparent retry rules for non-idempotent methods.
- Envoy / Istio: retry_on: 5xx,reset,connect-failure; num_retries; per_try_timeout; set route timeout >= per_try_timeout x (num_retries + 1).
- Python `urllib3.Retry(total=3, backoff_factor=0.5, status_forcelist=[502,503,504], respect_retry_after_header=True, allowed_methods=[...])`.
Check the library docs for the version you use; defaults change.
FILE:templates/retry-policy-review.md
# Retry Policy Review - <service / call chain> - <date>
**Reviewer:** <name> **Input:** <chain JSON / config files> **Failure rate assumed:** <30%>
## Chain
| # | Caller -> callee | Attempts | Backoff (base / x / cap) | Jitter | Per-try timeout | Deadline | Worst case |
|---|---|---|---|---|---|---|---|
| 1 | | | | | | | |
**Worst-case amplification:** <36> calls per request (target <= 10)
**Expected amplification at <p>:** <2.7>
## Findings
| Severity | Layer | Code | Problem | Fix |
|---|---|---|---|---|
| high | | | | |
## Proposed policy
| # | Caller -> callee | Attempts | Backoff | Jitter | Per-try timeout | Deadline | Retry on |
|---|---|---|---|---|---|---|---|
| 1 | | | | full | | | |
**New worst-case amplification:** <n> **New worst-case latency:** <s>
## Follow-ups
- [ ] Idempotency keys for: <calls>
- [ ] Retry budget / circuit breaker for: <dependency>
- [ ] Dashboards: tries per request, retry rate, give-ups
FILE:examples/checkout-chain.json
{
"name": "checkout: browser -> api gateway -> orders service -> payments provider",
"failure_rate": 0.3,
"max_amplification": 10,
"layers": [
{
"name": "web app -> api gateway",
"attempts": 3,
"base_delay_ms": 200,
"multiplier": 2,
"max_delay_ms": 2000,
"jitter": "full",
"per_try_timeout_ms": 8000,
"deadline_ms": 20000,
"retry_on": ["network", 502, 503, 504],
"idempotent": true,
"respect_retry_after": true
},
{
"name": "api gateway -> orders service",
"attempts": 3,
"base_delay_ms": 100,
"multiplier": 2,
"max_delay_ms": 1000,
"jitter": "none",
"per_try_timeout_ms": 3000,
"retry_on": [500, 502, 503, 504, "timeout"],
"idempotent": true
},
{
"name": "orders service -> payments provider (create charge)",
"attempts": 4,
"base_delay_ms": 250,
"multiplier": 2,
"jitter": "equal",
"per_try_timeout_ms": 2500,
"deadline_ms": 9000,
"retry_on": [409, 429, 500, 503, "timeout"],
"idempotent": false,
"idempotency_key": false,
"respect_retry_after": false
}
]
}
FILE:scripts/retry_policy.py
#!/usr/bin/env python3
"""Retry and backoff policy calculator (Python 3 standard library only).
Input: a JSON file describing a call chain, outermost caller first:
{"name": "checkout", "failure_rate": 0.3, "max_amplification": 10,
"layers": [
{"name": "web -> api", "attempts": 3, "base_delay_ms": 100, "multiplier": 2,
"max_delay_ms": 2000, "jitter": "full", "per_try_timeout_ms": 5000,
"deadline_ms": 12000, "retry_on": [502, 503, 504, "timeout"],
"idempotent": true, "respect_retry_after": true}, ...]}
attempts = total tries including the first one (attempts 1 = no retries).
jitter: none | full | equal | decorrelated. Delay before retry k (k = 1..attempts-1):
none d = min(cap, base * multiplier^(k-1))
full U(0, d) equal d/2 + U(0, d/2)
decorrelated min(cap, U(base, 3 * previous)) (AWS style; expected value simulated, seed 7)
Reports per layer: delay schedule (min / expected / max), worst-case time for one call
(attempts x per-try timeout + max delays, cut at deadline), expected tries at the given
failure rate, and for the chain: worst-case and expected request amplification at the
innermost dependency. Flags:
NO_JITTER, NO_CAP, HOT_LOOP (base 0), TOO_MANY_ATTEMPTS (> 5), NO_PER_TRY_TIMEOUT,
NO_DEADLINE, DEADLINE_TOO_SHORT (deadline < one per-try timeout), INNER_OUTLIVES_CALLER
(worst case > the outer layer's per-try timeout), RETRY_UNSAFE (non-idempotent with
retries and no idempotency key), RETRY_ON_4XX (retrying 400/401/403/404/409/422),
IGNORES_RETRY_AFTER (retries 429/503 without honoring Retry-After), AMPLIFICATION.
Usage:
python3 retry_policy.py chain.json
python3 retry_policy.py chain.json --failure-rate 0.5 --json
python3 retry_policy.py --self-test
Exit code: 0 OK, 1 high-severity flags found, 2 bad input.
"""
import argparse
import json
import random
import sys
JITTERS = ("none", "full", "equal", "decorrelated")
BAD_4XX = {400, 401, 403, 404, 405, 409, 410, 422}
SEVERITY = {"INNER_OUTLIVES_CALLER": "high", "RETRY_UNSAFE": "high", "AMPLIFICATION": "high",
"HOT_LOOP": "high", "NO_PER_TRY_TIMEOUT": "high", "RETRY_ON_4XX": "medium",
"NO_JITTER": "medium", "NO_CAP": "medium", "IGNORES_RETRY_AFTER": "medium",
"NO_DEADLINE": "medium", "DEADLINE_TOO_SHORT": "medium", "TOO_MANY_ATTEMPTS": "low"}
class PolicyError(ValueError):
pass
def num(layer, key, default=None, minimum=0):
v = layer.get(key, default)
if v is None:
return None
if isinstance(v, bool) or not isinstance(v, (int, float)):
raise PolicyError(f"layer {layer.get('name', '?')!r}: {key} must be a number")
if v < minimum:
raise PolicyError(f"layer {layer.get('name', '?')!r}: {key} must be >= {minimum}")
return v
def delays(layer):
"""Return [(min, expected, max)] in ms for each retry."""
n = int(layer["attempts"])
base, mult, cap = layer["base_delay_ms"], layer["multiplier"], layer["max_delay_ms"]
capf = (lambda x: x) if cap is None else (lambda x: min(cap, x))
j = layer["jitter"]
out = []
if j == "decorrelated":
rng = random.Random(7)
runs = 20000
sums = [0.0] * (n - 1)
for _ in range(runs):
prev = base
for k in range(n - 1):
prev = capf(rng.uniform(base, prev * 3)) if base > 0 else 0.0
sums[k] += prev
hi = base
for k in range(n - 1):
hi = capf(hi * 3)
out.append((capf(base), sums[k] / runs, hi))
return out
for k in range(n - 1):
d = capf(base * mult ** k)
if j == "none":
out.append((d, d, d))
elif j == "full":
out.append((0.0, d / 2, d))
else:
out.append((d / 2, d * 0.75, d))
return out
def normalize(raw):
layers = raw.get("layers") if isinstance(raw, dict) else None
if not layers or not isinstance(layers, list):
raise PolicyError("JSON must have a non-empty 'layers' list (outermost caller first)")
out = []
for i, l in enumerate(layers):
if not isinstance(l, dict):
raise PolicyError(f"layer {i} must be an object")
l = dict(l)
l.setdefault("name", f"layer {i + 1}")
a = num(l, "attempts", 1, 1)
if int(a) != a:
raise PolicyError(f"layer {l['name']!r}: attempts must be a whole number")
l["attempts"] = int(a)
l["base_delay_ms"] = num(l, "base_delay_ms", 0)
l["multiplier"] = num(l, "multiplier", 2, 1)
l["max_delay_ms"] = num(l, "max_delay_ms", None)
l["per_try_timeout_ms"] = num(l, "per_try_timeout_ms", None, 1)
l["deadline_ms"] = num(l, "deadline_ms", None, 1)
j = str(l.get("jitter", "none")).lower()
if j not in JITTERS:
raise PolicyError(f"layer {l['name']!r}: jitter must be one of {', '.join(JITTERS)}")
l["jitter"] = j
l["retry_on"] = list(l.get("retry_on", []))
l["idempotent"] = bool(l.get("idempotent", True))
l["idempotency_key"] = bool(l.get("idempotency_key", False))
l["respect_retry_after"] = bool(l.get("respect_retry_after", False))
out.append(l)
return out
def expected_tries(n, p):
if p >= 1:
return float(n)
return sum(p ** k for k in range(n))
def analyze(raw, failure_rate=None, max_amp=None):
layers = normalize(raw)
p = failure_rate if failure_rate is not None else raw.get("failure_rate", 0.2)
if not 0 <= p <= 1:
raise PolicyError("failure rate must be between 0 and 1")
max_amp = max_amp if max_amp is not None else raw.get("max_amplification", 10)
res = {"name": raw.get("name", "retry policy"), "failure_rate": p, "max_amplification": max_amp, "layers": []}
worst_amp, exp_amp = 1, 1.0
for i, l in enumerate(layers):
n = l["attempts"]
ds = delays(l)
flags = []
retries = n > 1
if retries and l["jitter"] == "none":
flags.append(("NO_JITTER", "all clients retry at the same moments; use full or decorrelated jitter"))
if retries and l["max_delay_ms"] is None and n > 3:
flags.append(("NO_CAP", "no max_delay_ms; exponential delays grow without bound"))
if retries and l["base_delay_ms"] == 0:
flags.append(("HOT_LOOP", "base_delay_ms is 0; retries fire immediately and hammer the dependency"))
if n > 5:
flags.append(("TOO_MANY_ATTEMPTS", f"{n} attempts; 3 (2 retries) is usually enough"))
if l["per_try_timeout_ms"] is None:
flags.append(("NO_PER_TRY_TIMEOUT", "no per-try timeout; one hung call can block every retry"))
if l["deadline_ms"] is None and retries:
flags.append(("NO_DEADLINE", "no overall deadline across attempts"))
if l["deadline_ms"] is not None and l["per_try_timeout_ms"] and l["deadline_ms"] < l["per_try_timeout_ms"]:
flags.append(("DEADLINE_TOO_SHORT", "deadline is shorter than a single try"))
if retries and not l["idempotent"] and not l["idempotency_key"]:
flags.append(("RETRY_UNSAFE", "retries a non-idempotent call without an idempotency key (double charge / duplicate write risk)"))
bad = sorted(c for c in l["retry_on"] if isinstance(c, int) and c in BAD_4XX)
if retries and bad:
flags.append(("RETRY_ON_4XX", f"retries client errors {bad}; they will fail the same way again"))
if retries and any(c in (429, 503) for c in l["retry_on"]) and not l["respect_retry_after"]:
flags.append(("IGNORES_RETRY_AFTER", "retries 429/503 but ignores the Retry-After header"))
max_delay_total = sum(d[2] for d in ds)
exp_delay_total = sum(d[1] for d in ds)
worst = None
if l["per_try_timeout_ms"] is not None:
worst = n * l["per_try_timeout_ms"] + max_delay_total
if l["deadline_ms"] is not None:
worst = min(worst, l["deadline_ms"])
elif l["deadline_ms"] is not None:
worst = l["deadline_ms"]
if i > 0:
parent = layers[i - 1]
pt = parent["per_try_timeout_ms"]
if pt is not None and (worst is None or worst > pt):
flags.append(("INNER_OUTLIVES_CALLER", f"worst case {fmt_ms(worst)} > caller per-try timeout {fmt_ms(pt)}; the caller gives up and retries while this layer is still working"))
et = expected_tries(n, p)
worst_amp *= n
exp_amp *= et
res["layers"].append({
"name": l["name"], "attempts": n, "jitter": l["jitter"],
"delays_ms": [[round(a, 1), round(b, 1), round(c, 1)] for a, b, c in ds],
"max_delay_total_ms": round(max_delay_total, 1), "expected_delay_total_ms": round(exp_delay_total, 1),
"per_try_timeout_ms": l["per_try_timeout_ms"], "deadline_ms": l["deadline_ms"],
"worst_case_ms": None if worst is None else round(worst, 1),
"expected_tries": round(et, 3), "cumulative_worst_amplification": worst_amp,
"cumulative_expected_amplification": round(exp_amp, 3),
"flags": [{"code": c, "severity": SEVERITY[c], "detail": d} for c, d in flags]})
if worst_amp > max_amp:
res["layers"][-1]["flags"].append({"code": "AMPLIFICATION", "severity": "high",
"detail": f"one top-level request can become {worst_amp} calls to the innermost dependency during an outage (limit {max_amp}); retry at one layer only or add a retry budget"})
res["worst_amplification"] = worst_amp
res["expected_amplification"] = round(exp_amp, 3)
res["high_flags"] = sum(1 for l in res["layers"] for f in l["flags"] if f["severity"] == "high")
return res
def fmt_ms(v):
if v is None:
return "unbounded"
if v >= 1000:
s = f"{v / 1000:.2f}".rstrip("0").rstrip(".")
return s + " s"
return f"{v:g} ms"
def to_markdown(r):
out = [f"# Retry policy report: {r['name']}", "",
f"Failure rate used for expected values: {r['failure_rate']:.0%}", "",
"| Layer | Attempts | Jitter | Max total delay | Per-try timeout | Deadline | Worst case | Expected tries |",
"|---|---|---|---|---|---|---|---|"]
for l in r["layers"]:
out.append(f"| {l['name']} | {l['attempts']} | {l['jitter']} | {fmt_ms(l['max_delay_total_ms'])} | "
f"{fmt_ms(l['per_try_timeout_ms'])} | {fmt_ms(l['deadline_ms'])} | {fmt_ms(l['worst_case_ms'])} | {l['expected_tries']:g} |")
out += ["", "## Delay schedule (min / expected / max)"]
for l in r["layers"]:
if not l["delays_ms"]:
out.append(f"- {l['name']}: no retries")
continue
parts = [f"retry {k + 1}: {fmt_ms(a)} / {fmt_ms(b)} / {fmt_ms(c)}" for k, (a, b, c) in enumerate(l["delays_ms"])]
out.append(f"- {l['name']}: " + "; ".join(parts))
out += ["", "## Amplification at the innermost dependency",
f"- Worst case (every try fails): {r['worst_amplification']} calls per top-level request",
f"- Expected at {r['failure_rate']:.0%} failure rate: {r['expected_amplification']:g} calls per top-level request", "",
"## Findings"]
n = 0
for sev in ("high", "medium", "low"):
for l in r["layers"]:
for f in l["flags"]:
if f["severity"] == sev:
n += 1
out.append(f"- **{sev.upper()} {f['code']}** ({l['name']}): {f['detail']}")
if not n:
out.append("- No findings.")
return "\n".join(out)
def self_test():
cases = []
def ok(name, cond):
cases.append((name, bool(cond)))
def one(**kw):
base = {"name": "x", "attempts": 3, "base_delay_ms": 100, "multiplier": 2, "max_delay_ms": 1000,
"jitter": "full", "per_try_timeout_ms": 1000, "deadline_ms": 5000, "retry_on": [503],
"respect_retry_after": True}
base.update(kw)
return base
def codes(r, i=0):
return {f["code"] for f in r["layers"][i]["flags"]}
d = delays(normalize({"layers": [one(jitter="none", attempts=5)]})[0])
ok("exponential no jitter 100,200,400,800", [x[2] for x in d] == [100, 200, 400, 800])
d = delays(normalize({"layers": [one(jitter="none", attempts=6)]})[0])
ok("cap applied at 1000", d[-1][2] == 1000 and d[-2][2] == 800)
d = delays(normalize({"layers": [one(jitter="full")]})[0])
ok("full jitter mean half", d[0] == (0.0, 50.0, 100))
d = delays(normalize({"layers": [one(jitter="equal")]})[0])
ok("equal jitter 50..100 mean 75", d[0] == (50.0, 75.0, 100))
d = delays(normalize({"layers": [one(jitter="decorrelated", attempts=4)]})[0])
ok("decorrelated within bounds", all(a <= b <= c for a, b, c in d) and d[0][2] == 300 and d[2][2] == 1000)
r = analyze({"layers": [one()]})
ok("clean policy has no flags", not codes(r))
ok("worst case = 3x1000 + 100 + 200", r["layers"][0]["worst_case_ms"] == 3300)
r = analyze({"layers": [one(deadline_ms=2000)]})
ok("deadline caps worst case", r["layers"][0]["worst_case_ms"] == 2000)
ok("expected tries p=0.5 n=3 is 1.75", expected_tries(3, 0.5) == 1.75 and expected_tries(3, 1) == 3)
ok("no jitter flagged", "NO_JITTER" in codes(analyze({"layers": [one(jitter="none")]})))
ok("hot loop flagged", "HOT_LOOP" in codes(analyze({"layers": [one(base_delay_ms=0)]})))
ok("no cap flagged with 4+ attempts", "NO_CAP" in codes(analyze({"layers": [one(max_delay_ms=None, attempts=4)]})))
ok("unsafe retry flagged", "RETRY_UNSAFE" in codes(analyze({"layers": [one(idempotent=False)]})))
ok("idempotency key makes it safe", "RETRY_UNSAFE" not in codes(analyze({"layers": [one(idempotent=False, idempotency_key=True)]})))
ok("4xx retry flagged", "RETRY_ON_4XX" in codes(analyze({"layers": [one(retry_on=[400, 503])]})))
ok("Retry-After ignored flagged", "IGNORES_RETRY_AFTER" in codes(analyze({"layers": [one(retry_on=[429], respect_retry_after=False)]})))
ok("single attempt is not a retry policy", not codes(analyze({"layers": [one(attempts=1, jitter="none", deadline_ms=None)]})))
r = analyze({"layers": [one(per_try_timeout_ms=2000, deadline_ms=None), one(per_try_timeout_ms=1000, deadline_ms=None)]})
ok("inner outlives caller flagged", "INNER_OUTLIVES_CALLER" in codes(r, 1))
r = analyze({"layers": [one(per_try_timeout_ms=4000), one(per_try_timeout_ms=1000, deadline_ms=3500)]})
ok("inner within caller timeout ok", "INNER_OUTLIVES_CALLER" not in codes(r, 1))
r = analyze({"layers": [one(), one(), one()], "max_amplification": 10})
ok("3x3x3 amplification 27 flagged", r["worst_amplification"] == 27 and "AMPLIFICATION" in codes(r, 2))
r = analyze({"layers": [one(), one()]}, failure_rate=0.5)
ok("expected amplification 1.75^2", r["expected_amplification"] == 3.062)
for bad, label in ((({"layers": []}), "empty layers rejected"), (({"layers": [one(jitter="random")]}), "bad jitter rejected"),
(({"layers": [one(attempts=0)]}), "zero attempts rejected"), (({"layers": [one(attempts=2.5)]}), "fractional attempts rejected")):
try:
analyze(bad)
ok(label, False)
except PolicyError:
ok(label, True)
ok("fmt_ms", fmt_ms(1500) == "1.5 s" and fmt_ms(250) == "250 ms" and fmt_ms(None) == "unbounded")
passed = sum(c for _, c in cases)
for name, c in cases:
print(("PASS " if c else "FAIL ") + name)
print(f"{passed}/{len(cases)} passed")
return 0 if passed == len(cases) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Check retry/backoff/timeout policies across a call chain.")
ap.add_argument("file", nargs="?", help="chain JSON ('-' for stdin)")
ap.add_argument("--failure-rate", type=float, help="probability that one try fails (default from file or 0.2)")
ap.add_argument("--max-amplification", type=float, help="flag chains above this worst-case amplification (default 10)")
ap.add_argument("--json", action="store_true")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
if not a.file:
ap.error("chain JSON required (or --self-test)")
try:
raw = json.load(sys.stdin if a.file == "-" else open(a.file, encoding="utf-8"))
r = analyze(raw, a.failure_rate, a.max_amplification)
except (OSError, PolicyError, ValueError) as e:
print(f"error: {e}", file=sys.stderr)
return 2
print(json.dumps(r, indent=2) if a.json else to_markdown(r))
return 1 if r["high_flags"] else 0
if __name__ == "__main__":
sys.exit(main())
上线前模拟 API 限流：令牌桶、固定窗口、滑动日志与滑动窗口等策略对比。
---
name: api-rate-limit-policy-simulator
description: 上线前模拟 API 限流：令牌桶、固定窗口、滑动日志与滑动窗口等策略对比。
---
# API 限流策略模拟器
你是「API 限流策略模拟器」技能助手。上方 description 为中文任务说明。下方保留原技能完整正文（含脚本与代码）；执行时遵循中文说明与原文技术细节。
# API Rate Limit Policy Simulator
You help backend and platform engineers choose rate limits with evidence instead of round numbers. A limit that looks safe on paper ("300 per minute per key") can still let 1.8x that through at a window boundary, throttle a mobile app whose thousands of users share one API key, release a herd of retries at the reset second, and do nothing against a scraper that stays just under it. You describe the policies and the traffic in a small JSON file, replay it, and read the result client by client and policy by policy.
## Files in this skill
- `scripts/rate_sim.py` - simulates token bucket, fixed window, sliding log and sliding counter limiters keyed by client, endpoint, client+endpoint or global, scoped to endpoints and client tiers; generates steady, Poisson or burst traffic (or replays a CSV trace), lets clients retry immediately, with backoff or by honouring Retry-After, and reports 429s, delays, failures, worst-case and observed window counts, peaks vs capacity, findings RL01-RL10 and a sample 429 response (Python 3.8+ standard library only)
- `references/algorithms.md` - how each algorithm behaves, its worst case, memory cost and when to pick it
- `references/policy-design.md` - choosing keys, tiers, numbers, headers and client retry rules; what the finding codes mean
- `templates/rate-limit-review.md` - the review to hand back
- `examples/orders-api.json` - a public orders API with a per-key fixed window, a login limit and a global fixed window; five clients including a shared mobile key, a bulk partner, a scraper and a credential stuffer
- `examples/orders-api-report.md` - the real simulator output on that file
- `examples/orders-api-tuned.json` - the same API after the fixes: per-tier token buckets, a global token bucket sized to capacity
## Workflow
### 1. Collect the inputs
Ask for: the limits as configured today or as proposed (algorithm, key, numbers, which endpoints and which clients), where they are enforced (gateway, CDN, app, Redis), the backend capacity in requests per second it can sustain, and the traffic: the biggest legitimate clients and their shape (steady, bursty batch jobs, many users behind one key or one NAT IP), how each one retries, and any abuse seen in logs. Real access logs beat guesses: export `t,client,endpoint` for a busy 5-15 minutes and pass it with `--trace`.
### 2. Run the simulator
```
python3 scripts/rate_sim.py examples/orders-api.json
python3 scripts/rate_sim.py examples/orders-api-tuned.json
python3 scripts/rate_sim.py policies.json --trace access-sample.csv
python3 scripts/rate_sim.py policies.json --json
python3 scripts/rate_sim.py --self-test
```
Exit 0 = no high findings, 1 = high findings (legitimate clients throttled or admitted traffic over capacity), 2 = bad input. Without a shell, reason with `references/algorithms.md`: per key, compare the client's peak per window with the limit, the worst case (2x limit for fixed windows, burst + rate for buckets) with capacity, and think through what each client does after a 429.
### 3. Read the result
- Legit clients first (RL01): any failed or delayed requests, and how long they waited. One key shared by an app's users needs its own tier or a per-user key.
- Capacity (RL03): the peak admitted second vs what the backend sustains. A global limit is the only thing that protects capacity; per-key limits multiply by the number of keys.
- Shape (RL04, RL10): fixed windows let bursts through at boundaries and hand out a whole window at once; retries without jitter all come back in the same second.
- Abuse (RL02): a scraper under the per-key limit is not stopped by limits; it needs detection, auth or a per-endpoint limit.
- Retries (RL05): attempts per logical request above 1.5 means the clients amplify load when throttled.
### 4. Tune and re-run
Change one thing at a time and re-run with the same traffic: switch fixed windows to token buckets or sliding counters, add tiers with `clients` / `exclude_clients`, size the global bucket to capacity (rate = sustainable rps, burst = what the backend absorbs for a second), add endpoint limits for login and expensive calls, and agree pacing with batch partners. Use `references/policy-design.md` for the numbers.
### 5. Deliver
Fill `templates/rate-limit-review.md`: verdict, a before/after table per client, the policy changes as config, the 429 response and headers, client guidance (Retry-After, backoff with jitter, max retries) and what to monitor after rollout.
## Example result
On `examples/orders-api.json` the shared `mobile-app` key hits the 300/minute per-key window: 1,042 of its 1,839 requests succeed only after waiting up to 60 s (RL01, high). Honouring Retry-After without jitter releases 289 retries in second 240, so admitted traffic peaks at 313 req/s against a capacity of 80 (RL03, RL10). The per-key fixed window lets 544 requests through in one rolling minute (1.8x, RL04), and the scraper still gets 45% of its calls through while its immediate retries push it to 2.67 attempts per request. The login limit stops the credential stuffer (40 of 480). In `examples/orders-api-tuned.json` (tier token buckets, global bucket 50 req/s + 30 burst, partner pacing at 2 req/s) legit clients see no 429s, the peak is 27 req/s and only the scraper findings remain.
## Rules
- Simulate with the traffic you actually have (or a trace), not an average; bursts and retries are where limits fail.
- Never call a limit safe for capacity unless a global limit bounds the total.
- Treat delayed requests of legitimate clients as harm, not as success.
- Always return 429 with Retry-After, add jitter, and tell clients how to back off.
- Rate limits are not abuse protection on their own; say when authentication, bot detection or WAF rules are needed.
- Results are a model: confirm with gateway metrics after rollout and keep the limits in config, not code.
FILE:references/algorithms.md
# Rate limit algorithms
| Algorithm | Admits | Worst case in one window | Memory per key | Good for |
|---|---|---|---|---|
| Token bucket (GCRA) | while tokens remain; refills at `rate`/s up to `burst` | burst + rate x window | 2 numbers | most APIs: smooth average, bounded bursts |
| Fixed window | `limit` per calendar window, counter resets | 2 x limit (end of one window + start of next) | 1 counter | quotas and billing ("10,000 per day"), not capacity |
| Sliding log | at most `limit` in any rolling window | limit (exact) | one timestamp per request | small limits: login, OTP, password reset |
| Sliding counter | previous window weighted by overlap + current | about limit (assumes even spread) | 2 counters | large limits where exactness is not needed |
## Token bucket
- `rate` is the long-run average; `burst` is how many calls can arrive at once after a quiet period.
- `limit` + `window_s` without a burst means rate = limit/window and burst = limit: a client can spend the whole minute in one second. Set the burst explicitly.
- Retry-After = (cost - tokens) / rate: usually well under a second, so retries spread out by themselves.
- GCRA (used by many gateways) stores one timestamp and gives the same admit/reject decisions.
## Fixed window
- Cheap (`INCR` + `EXPIRE` in Redis) and easy to explain, so it is everywhere.
- A client can send `limit` in the last second of a window and `limit` again in the first second of the next.
- Every throttled client gets Retry-After = time to the reset, so they all come back in the same second (retry herd).
- Use it for quotas that match business periods; put a token bucket in front of it for capacity.
## Sliding log
- Exact, but stores every admitted timestamp for the window: 10,000/hour per key = 10,000 entries per key.
- Retry-After = oldest relevant entry + window - now: can be long; tell the client.
## Sliding counter
- Estimate = previous count x (1 - elapsed/window) + current count.
- Error comes from uneven traffic: a burst at the end of the previous window is under-counted later (in the simulator's self-test, 10 requests at the very end of one window plus an even trickle later admit 16 in one rolling minute for a limit of 10).
- Good default for large per-key limits behind a token bucket.
## Concurrency limits (not simulated)
Limits on in-flight requests protect slow endpoints better than rate limits; combine with a rate limit for fairness.
FILE:references/policy-design.md
# Designing rate limit policies
## Pick the key
- **API key / token** for authenticated traffic. Watch for keys shared by many users (mobile apps, SDKs embedded in customer sites): give them a tier or move to per-user keys.
- **User or account id** behind the key when the API knows it; fairer than the key.
- **IP** only for anonymous endpoints; offices, schools, carriers and VPNs put thousands of users behind one IP (CGNAT). Use generous IP limits and stricter account limits.
- **Endpoint** for expensive calls (search, export, bulk) and sensitive ones (login, OTP, password reset).
- **Global** to protect capacity. Without it, per-key limits x number of keys is the real ceiling.
## Pick the numbers
1. Sustained rate per key: p99 of a normal busy client's requests per second x 2-3 headroom.
2. Burst: the largest legitimate batch (page load fan-out, sync job) that must not be throttled.
3. Global: rate = what the backend sustains at acceptable latency (load test), burst = what it absorbs for about a second.
4. Login/OTP: 5-10 per minute per account and per IP, sliding log, plus lockout or CAPTCHA after repeated failures.
5. Quotas (per day/month) are a product decision; keep them separate from capacity limits.
## Tiers
Scope policies with `clients` and `exclude_clients` in the simulator; in production map tiers to plans or key metadata (free, paid, partner, internal). Internal services should usually bypass public limits but stay under the global one.
## Responses and headers
- 429 Too Many Requests, `Retry-After` in seconds, a problem+json body naming the limit.
- Advertise the policy: the IETF httpapi ratelimit-headers draft defines `RateLimit-Policy` and `RateLimit` fields (field syntax changed between drafts); many gateways still send `X-RateLimit-Limit`, `-Remaining`, `-Reset`. Pick one and document it.
- Add 0-20% random jitter to Retry-After for fixed windows so throttled clients do not return together.
- Use 503 + Retry-After for overload (global limit), 429 for per-client limits, so clients can tell them apart.
## Client rules to publish
Honour Retry-After; otherwise exponential backoff with full jitter (1, 2, 4 s ...), at most 3-5 retries, never retry non-idempotent calls without an idempotency key; pace batch jobs instead of bursting.
## Finding codes (rate_sim.py)
| Code | Meaning |
|---|---|
| RL01 | legit client failed or delayed (high above 1% of its requests) |
| RL02 | abusive client still gets more than 25% through |
| RL03 | admitted peak over capacity, or no global limit to bound it |
| RL04 | fixed window let more than the limit through in a rolling window |
| RL05 | client retries amplify attempts 1.5x or more |
| RL06 | noisy neighbour holds most of a global limit while legit clients are rejected |
| RL07 | sliding log with a very large limit (memory) |
| RL08 | reminder: 429 + Retry-After and documented limits |
| RL09 | policy never triggers and is 10x above the busiest key |
| RL10 | retry herd: most of the peak second are retries released together |
FILE:templates/rate-limit-review.md
# Rate limit review: <service>
**Scope:** <endpoints / gateway> **Capacity:** <80 req/s sustained> **Traffic used:** <generated from ... / trace of 10 min on ...>
**Verdict:** <ship / ship with changes / do not ship>
## What happens today
| Client | Kind | Requests | 429s | OK after retry | Failed | Max wait | Main policy |
|---|---|---|---|---|---|---|---|
| <mobile-app> | <legit> | <1,839> | <1,081> | <1,042> | <0> | <60 s> | <per-key minute> |
Peak admitted: <313 req/s in second 240> vs capacity <80>.
## Findings
1. <HIGH RL01 - shared mobile key throttled; 57% of requests delayed up to 60 s>
2. <HIGH RL03 / MEDIUM RL10 - retry herd at window reset, 4x capacity>
3. <MEDIUM RL04 - fixed window 1.8x at boundary>
4. <MEDIUM RL02 - scraper under the per-key limit>
## Proposed policies
```json
<policies block from the tuned config>
```
## After the change (same traffic)
| Client | 429s before | 429s after | Failed before | Failed after |
|---|---|---|---|---|
| <...> | | | | |
Peak admitted after: <27 req/s>.
## Response contract
- 429 + Retry-After (+ jitter) for per-client limits; 503 + Retry-After for global overload
- Headers: <RateLimit-Policy / RateLimit or X-RateLimit-*>
- Client guidance: <honour Retry-After, backoff with jitter, max 3 retries, pace batch jobs>
## Outside rate limits
<bot detection / auth / WAF rule for the scraper; lockout or CAPTCHA for login>
## Monitor after rollout
<429 rate per client and policy, top throttled keys, p99 latency, admitted rps vs capacity; review in 2 weeks>
FILE:examples/orders-api.json
{
"service": "Orders API (public v2)",
"duration_s": 300,
"capacity_rps": 80,
"seed": 7,
"policies": [
{"name": "per-key minute", "algorithm": "fixed_window", "key": "client", "limit": 300, "window_s": 60},
{"name": "login", "algorithm": "sliding_log", "key": "client", "limit": 10, "window_s": 60,
"endpoints": ["/v2/login"]},
{"name": "global", "algorithm": "fixed_window", "key": "global", "limit": 3600, "window_s": 60}
],
"clients": [
{"name": "mobile-app", "kind": "legit", "pattern": "poisson", "rps": 6, "endpoint": "/v2/orders",
"retry": "honor"},
{"name": "partner-erp", "kind": "legit", "pattern": "burst", "burst": 120, "every_s": 60, "start": 58,
"endpoint": "/v2/orders/bulk", "retry": "backoff"},
{"name": "checkout-web", "kind": "legit", "pattern": "poisson", "rps": 3, "endpoint": "/v2/checkout",
"retry": "honor"},
{"name": "price-scraper", "kind": "abusive", "pattern": "steady", "rps": 12, "start": 20,
"endpoint": "/v2/products", "retry": "immediate"},
{"name": "cred-stuffer", "kind": "abusive", "pattern": "steady", "rps": 2, "start": 60,
"endpoint": "/v2/login"}
]
}
FILE:examples/orders-api-report.md
# Rate limit simulation: Orders API (public v2)
Simulated 300s, 7227 logical requests, 4930 admitted, peak 313 req/s admitted (second 240) vs capacity 80.
**Verdict: 2 high finding(s) - fix before shipping these limits**
## Clients
| Client | Kind | Requests | Attempts | Allowed | 429s | OK after retry | Failed | Failed % | Max Retry-After | Top policy |
|---|---|---|---|---|---|---|---|---|---|---|
| price-scraper | abusive | 3360 | 8961 | 1503 | 7458 | 15 | 1857 | 55.3% | 35.2s | per-key minute |
| mobile-app | legit | 1839 | 2920 | 1839 | 1081 | 1042 | 0 | 0.0% | 59.9s | per-key minute |
| checkout-web | legit | 948 | 948 | 948 | 0 | 0 | 0 | 0.0% | 0s | - |
| partner-erp | legit | 600 | 600 | 600 | 0 | 0 | 0 | 0.0% | 0s | - |
| cred-stuffer | abusive | 480 | 480 | 40 | 440 | 0 | 440 | 91.7% | 55s | login |
## Policies
| Policy | Algorithm | Key | Scope | Sustained | Worst case / window | Busiest key / window | Checks | Rejects |
|---|---|---|---|---|---|---|---|---|
| per-key minute | fixed_window | client | all | 5 req/s | 600 per 60s | 544 | 13909 | 8539 |
| login | sliding_log | client | /v2/login | 0.167 req/s | 10 per 60s | 10 | 480 | 440 |
| global | fixed_window | global | all | 60 req/s | 7200 per 60s | 1235 | 13909 | 0 |
## Findings
- **HIGH RL01** Legit client mobile-app is throttled on 56.7% of its requests (0 failed, 1042 only after retrying, waits up to 59.9s), mostly by policy 'per-key minute'; raise that limit or give this client its own tier.
- **HIGH RL03** Admitted traffic peaks at 313 req/s (second 240) over the backend capacity of 80 req/s; add or lower a global limit.
- **MEDIUM RL02** Abusive client price-scraper still gets 45% of its requests through (1503); tighten the per-client limit or add a stricter policy for its pattern.
- **MEDIUM RL04** Fixed window 'per-key minute' let 544 requests through in one rolling 60s window (limit 300, 1.8x) across a window boundary; use sliding_counter or a token bucket if the limit protects capacity.
- **MEDIUM RL05** Client price-scraper sends 2.67x attempts per logical request because it retries throttled calls; retries count against the limit and add load - require Retry-After or backoff with jitter.
- **MEDIUM RL05** Client mobile-app sends 1.59x attempts per logical request because it retries throttled calls; retries count against the limit and add load - require Retry-After or backoff with jitter.
- **MEDIUM RL10** Retry herd: 289 of the 313 requests admitted in second 240 are retries released by the same Retry-After / window reset; add jitter to Retry-After (or client backoff) and prefer a token bucket or sliding window so capacity is not handed out all at once.
- **INFO RL08** Return 429 with Retry-After on every rejection (see the sample headers) and document the limits so clients can pace themselves.
## Sample 429 (first rejection, t=45.0s, price-scraper /v2/products)
```
HTTP/1.1 429 Too Many Requests
Retry-After: 15
RateLimit-Policy: "per-key minute";q=300;w=60
RateLimit: "per-key minute";r=0;t=15
Content-Type: application/problem+json
```
Header names follow the IETF httpapi ratelimit-headers draft; older gateways use X-RateLimit-Limit / -Remaining / -Reset instead.
FILE:examples/orders-api-tuned.json
{
"service": "Orders API (public v2, tuned)",
"notes": "Same traffic as orders-api.json except partner-erp, which agreed to pace its bulk sync at 2 req/s instead of 120-request bursts. Token buckets per tier, a global token bucket sized to capacity.",
"duration_s": 300,
"capacity_rps": 80,
"seed": 7,
"policies": [
{"name": "default key", "algorithm": "token_bucket", "key": "client", "rate": 4, "burst": 40,
"exclude_clients": ["mobile-app", "partner-erp"]},
{"name": "mobile tier", "algorithm": "token_bucket", "key": "client", "rate": 15, "burst": 60,
"clients": ["mobile-app"]},
{"name": "partner tier", "algorithm": "token_bucket", "key": "client", "rate": 3, "burst": 20,
"clients": ["partner-erp"]},
{"name": "login", "algorithm": "sliding_log", "key": "client", "limit": 10, "window_s": 60,
"endpoints": ["/v2/login"]},
{"name": "global", "algorithm": "token_bucket", "key": "global", "rate": 50, "burst": 30}
],
"clients": [
{"name": "mobile-app", "kind": "legit", "pattern": "poisson", "rps": 6, "endpoint": "/v2/orders",
"retry": "honor"},
{"name": "partner-erp", "kind": "legit", "pattern": "steady", "rps": 2, "start": 58,
"endpoint": "/v2/orders/bulk", "retry": "backoff"},
{"name": "checkout-web", "kind": "legit", "pattern": "poisson", "rps": 3, "endpoint": "/v2/checkout",
"retry": "honor"},
{"name": "price-scraper", "kind": "abusive", "pattern": "steady", "rps": 12, "start": 20,
"endpoint": "/v2/products", "retry": "immediate"},
{"name": "cred-stuffer", "kind": "abusive", "pattern": "steady", "rps": 2, "start": 60,
"endpoint": "/v2/login"}
]
}
FILE:scripts/rate_sim.py
#!/usr/bin/env python3
"""API rate limit policy simulator (Python 3.8+ standard library only).
Replays generated or recorded traffic against one or more rate limit policies and shows who
gets throttled, why, for how long, and whether the policies protect the backend without
hurting legitimate clients.
Algorithms (per policy):
token_bucket rate (tokens/s) + burst (bucket size); or limit + window_s (rate = limit/window_s,
burst defaults to limit). GCRA behaves the same for admit/reject.
fixed_window limit requests per window_s, counter resets on window boundaries
(allows up to 2 x limit across a boundary)
sliding_log exact: at most limit requests in any window_s (stores one timestamp per request)
sliding_counter approximation: previous window weighted by overlap + current window (Cloudflare /
Redis style, about limit per rolling window, a few % error)
Config JSON:
{
"service": "Orders API", "duration_s": 300, "capacity_rps": 60, "seed": 7,
"policies": [
{"name": "per-client burst", "algorithm": "token_bucket", "key": "client", "rate": 5, "burst": 20},
{"name": "per-client minute", "algorithm": "fixed_window", "key": "client", "limit": 200, "window_s": 60},
{"name": "login", "algorithm": "sliding_log", "key": "client", "limit": 5, "window_s": 60,
"endpoints": ["/login"]},
{"name": "global", "algorithm": "sliding_counter", "key": "global", "limit": 3000, "window_s": 60}
],
"clients": [
{"name": "mobile-app", "kind": "legit", "pattern": "steady", "rps": 2, "endpoint": "/orders"},
{"name": "nightly-sync", "kind": "legit", "pattern": "burst", "burst": 40, "every_s": 60, "start": 5},
{"name": "scraper", "kind": "abusive", "pattern": "poisson", "rps": 25, "retry": "immediate"}
]
}
key client | endpoint | client+endpoint | global
endpoints optional list: the policy applies only to these endpoints (prefix match)
clients / exclude_clients optional lists: apply only to / skip these clients (tiers, allow lists)
pattern steady (rps) | poisson (rps, seeded) | burst (burst requests every every_s seconds)
start, duration seconds (default 0 and duration_s); cost = tokens per request (default 1)
retry none (default) | immediate (0.1 s later) | backoff (1, 2, 4 s) | honor (Retry-After);
max_retries default 3
kind legit | abusive | unknown - used to judge false positives and leakage
Recorded traffic instead of generated: --trace trace.csv with columns t,client[,endpoint][,cost]
(t in seconds from start). Clients in the config then only add kind/retry information.
Usage:
python3 rate_sim.py config.json [--trace trace.csv] [--json]
python3 rate_sim.py --self-test
Exit codes: 0 = no high findings, 1 = high findings (legit clients throttled or backend over
capacity), 2 = bad input.
"""
import argparse
import csv
import heapq
import json
import math
import random
import sys
from collections import defaultdict, deque
ALGOS = ("token_bucket", "fixed_window", "sliding_log", "sliding_counter")
KEYS = ("client", "endpoint", "client+endpoint", "global")
class InputError(ValueError):
pass
# ----------------------------------------------------------------------------- limiters
class TokenBucket:
def __init__(self, rate, burst):
self.rate, self.burst = float(rate), float(burst)
self.state = {}
def check(self, key, t, cost):
tokens, last = self.state.get(key, (self.burst, t))
tokens = min(self.burst, tokens + (t - last) * self.rate)
self.state[key] = (tokens, t)
if tokens + 1e-9 >= cost:
return True, 0.0, tokens
return False, (cost - tokens) / self.rate, tokens
def consume(self, key, t, cost):
tokens, last = self.state[key]
self.state[key] = (tokens - cost, t)
class FixedWindow:
def __init__(self, limit, window):
self.limit, self.window = float(limit), float(window)
self.state = {}
def check(self, key, t, cost):
w = math.floor(t / self.window)
start, count = self.state.get(key, (w, 0.0))
if start != w:
start, count = w, 0.0
self.state[key] = (start, count)
if count + cost <= self.limit + 1e-9:
return True, 0.0, self.limit - count
return False, (w + 1) * self.window - t, self.limit - count
def consume(self, key, t, cost):
start, count = self.state[key]
self.state[key] = (start, count + cost)
class SlidingLog:
def __init__(self, limit, window):
self.limit, self.window = float(limit), float(window)
self.state = defaultdict(deque)
def check(self, key, t, cost):
log = self.state[key]
while log and log[0] <= t - self.window:
log.popleft()
used = len(log)
if used + cost <= self.limit + 1e-9:
return True, 0.0, self.limit - used
need = int(used + cost - self.limit) # entries that must expire first
idx = min(max(need - 1, 0), used - 1) if used else 0
wait = (log[idx] + self.window - t) if used else self.window
return False, max(wait, 0.001), self.limit - used
def consume(self, key, t, cost):
for _ in range(int(round(cost))):
self.state[key].append(t)
class SlidingCounter:
def __init__(self, limit, window):
self.limit, self.window = float(limit), float(window)
self.state = {}
def _roll(self, key, t):
w = math.floor(t / self.window)
start, prev, curr = self.state.get(key, (w, 0.0, 0.0))
if w == start + 1:
start, prev, curr = w, curr, 0.0
elif w > start + 1:
start, prev, curr = w, 0.0, 0.0
self.state[key] = (start, prev, curr)
return start, prev, curr
def check(self, key, t, cost):
start, prev, curr = self._roll(key, t)
e = t - start * self.window
est = prev * (1 - e / self.window) + curr
if est + cost <= self.limit + 1e-9:
return True, 0.0, self.limit - est
lim, w = self.limit, self.window
if curr + cost <= lim and prev > 0:
wait = w * (1 - (lim - curr - cost) / prev) - e
else:
x = 0.0
if curr > 0 and lim - cost < curr:
x = w * (1 - (lim - cost) / curr)
wait = (w - e) + max(0.0, x)
return False, max(wait, 0.001), self.limit - est
def consume(self, key, t, cost):
start, prev, curr = self.state[key]
self.state[key] = (start, prev, curr + cost)
def make_limiter(p):
algo = p["algorithm"]
if algo == "token_bucket":
return TokenBucket(p["rate"], p["burst"])
if algo == "fixed_window":
return FixedWindow(p["limit"], p["window_s"])
if algo == "sliding_log":
return SlidingLog(p["limit"], p["window_s"])
return SlidingCounter(p["limit"], p["window_s"])
# ----------------------------------------------------------------------------- config
def num(v, what, positive=True):
try:
x = float(v)
except (TypeError, ValueError):
raise InputError("%s must be a number, got %r" % (what, v))
if positive and x <= 0:
raise InputError("%s must be > 0, got %r" % (what, v))
return x
def normalize(cfg):
if not isinstance(cfg, dict):
raise InputError("config must be a JSON object")
pols = cfg.get("policies")
if not pols:
raise InputError("config needs a non-empty 'policies' list")
out = []
for i, p in enumerate(pols, 1):
q = dict(p)
q.setdefault("name", "policy-%d" % i)
algo = q.get("algorithm")
if algo not in ALGOS:
raise InputError("policy %r: algorithm must be one of %s" % (q["name"], ", ".join(ALGOS)))
q.setdefault("key", "client")
if q["key"] not in KEYS:
raise InputError("policy %r: key must be one of %s" % (q["name"], ", ".join(KEYS)))
if algo == "token_bucket":
if "rate" in q:
q["rate"] = num(q["rate"], q["name"] + ".rate")
q["burst"] = num(q.get("burst", max(1.0, q["rate"])), q["name"] + ".burst")
elif "limit" in q:
lim = num(q["limit"], q["name"] + ".limit")
win = num(q.get("window_s", 1), q["name"] + ".window_s")
q["rate"] = lim / win
q["burst"] = num(q.get("burst", lim), q["name"] + ".burst")
else:
raise InputError("policy %r: token_bucket needs rate (+burst) or limit + window_s" % q["name"])
q.setdefault("window_s", 1.0)
q["window_s"] = num(q["window_s"], q["name"] + ".window_s")
else:
q["limit"] = num(q.get("limit"), q["name"] + ".limit")
q["window_s"] = num(q.get("window_s"), q["name"] + ".window_s")
q["endpoints"] = list(q.get("endpoints") or [])
q["clients"] = list(q.get("clients") or [])
q["exclude_clients"] = list(q.get("exclude_clients") or [])
out.append(q)
cfg = dict(cfg)
cfg["policies"] = out
cfg["duration_s"] = num(cfg.get("duration_s", 300), "duration_s")
if cfg.get("capacity_rps") is not None:
cfg["capacity_rps"] = num(cfg["capacity_rps"], "capacity_rps")
clients = []
for i, c in enumerate(cfg.get("clients") or [], 1):
c = dict(c)
c.setdefault("name", "client-%d" % i)
c.setdefault("kind", "unknown")
c.setdefault("pattern", "steady")
c.setdefault("endpoint", "/")
c.setdefault("retry", "none")
if c["retry"] not in ("none", "immediate", "backoff", "honor"):
raise InputError("client %r: retry must be none, immediate, backoff or honor" % c["name"])
c["max_retries"] = int(c.get("max_retries", 3))
c["cost"] = num(c.get("cost", 1), c["name"] + ".cost")
c["start"] = num(c.get("start", 0), c["name"] + ".start", positive=False)
c["duration"] = num(c.get("duration", cfg["duration_s"] - c["start"]), c["name"] + ".duration")
if c["pattern"] in ("steady", "poisson"):
c["rps"] = num(c.get("rps"), c["name"] + ".rps")
elif c["pattern"] == "burst":
c["burst"] = int(num(c.get("burst"), c["name"] + ".burst"))
c["every_s"] = num(c.get("every_s", 60), c["name"] + ".every_s")
else:
raise InputError("client %r: pattern must be steady, poisson or burst" % c["name"])
clients.append(c)
cfg["clients"] = clients
return cfg
def generate(cfg):
rng = random.Random(cfg.get("seed", 1))
reqs = []
for c in cfg["clients"]:
t0, t1 = c["start"], c["start"] + c["duration"]
if c["pattern"] == "steady":
n = int(round(c["duration"] * c["rps"]))
reqs += [(t0 + i / c["rps"], c["name"], c["endpoint"], c["cost"]) for i in range(n)]
elif c["pattern"] == "poisson":
t = t0 + rng.expovariate(c["rps"])
while t < t1:
reqs.append((t, c["name"], c["endpoint"], c["cost"]))
t += rng.expovariate(c["rps"])
else:
t = t0
while t < t1:
reqs += [(t + k * 0.001, c["name"], c["endpoint"], c["cost"]) for k in range(c["burst"])]
t += c["every_s"]
return reqs
def read_trace(path):
reqs = []
with open(path, newline="", encoding="utf-8") as f:
for n, row in enumerate(csv.DictReader(f), 2):
try:
reqs.append((float(row["t"]), row["client"].strip(), (row.get("endpoint") or "/").strip(),
float(row.get("cost") or 1)))
except (KeyError, ValueError, AttributeError):
raise InputError("%s line %d: need t,client[,endpoint][,cost]" % (path, n))
return reqs
# ----------------------------------------------------------------------------- simulation
def policy_key(p, client, endpoint):
if p["endpoints"] and not any(endpoint.startswith(e) for e in p["endpoints"]):
return None
if (p["clients"] and client not in p["clients"]) or client in p["exclude_clients"]:
return None
return {"client": client, "endpoint": endpoint, "client+endpoint": client + " " + endpoint,
"global": "*"}[p["key"]]
def simulate(cfg, reqs):
pols = cfg["policies"]
lims = [make_limiter(p) for p in pols]
cinfo = {c["name"]: c for c in cfg["clients"]}
heap = []
for i, (t, cl, ep, cost) in enumerate(sorted(reqs)):
heapq.heappush(heap, (t, i, cl, ep, cost, 0, i))
seq = len(reqs)
stats = defaultdict(lambda: {"logical": 0, "attempts": 0, "allowed": 0, "rejected": 0, "failed": 0, "delayed": 0,
"max_retry_after": 0.0, "blocked_by": defaultdict(int), "first_reject": None})
pstats = [{"checks": 0, "rejects": 0, "allowed_times": defaultdict(list), "by_client": defaultdict(int)}
for _ in pols]
admitted = []
sample = None
while heap:
t, _, cl, ep, cost, attempt, lid = heapq.heappop(heap)
s = stats[cl]
if attempt == 0:
s["logical"] += 1
s["attempts"] += 1
keys = [policy_key(p, cl, ep) for p in pols]
rejecting, wait, remain = [], 0.0, {}
for j, (lim, k) in enumerate(zip(lims, keys)):
if k is None:
continue
pstats[j]["checks"] += 1
ok, w, rem = lim.check(k, t, cost)
remain[j] = rem
if not ok:
rejecting.append(j)
wait = max(wait, w)
if not rejecting:
for j, (lim, k) in enumerate(zip(lims, keys)):
if k is not None:
lim.consume(k, t, cost)
pstats[j]["allowed_times"][k].append(t)
s["allowed"] += 1
if attempt:
s["delayed"] += 1
admitted.append((t, cl, attempt))
continue
s["rejected"] += 1
s["max_retry_after"] = max(s["max_retry_after"], wait)
if s["first_reject"] is None:
s["first_reject"] = t
s["blocked_by"][pols[rejecting[0]]["name"]] += 1
for j in rejecting:
pstats[j]["rejects"] += 1
pstats[j]["by_client"][cl] += 1
if sample is None:
j = rejecting[0]
sample = {"t": round(t, 3), "client": cl, "endpoint": ep, "policy": pols[j]["name"],
"retry_after": max(1, math.ceil(wait)), "quota": pols[j].get("limit", pols[j].get("burst")),
"window_s": pols[j]["window_s"]}
c = cinfo.get(cl, {})
mode, maxr = c.get("retry", "none"), c.get("max_retries", 3)
if mode != "none" and attempt < maxr:
delay = {"immediate": 0.1, "backoff": 2.0 ** attempt, "honor": wait + 0.05}[mode]
seq += 1
heapq.heappush(heap, (t + delay, seq, cl, ep, cost, attempt + 1, lid))
else:
s["failed"] += 1
return stats, pstats, admitted, sample
def max_in_window(times, w):
best, dq = 0, deque()
for t in sorted(times):
dq.append(t)
while dq[0] <= t - w:
dq.popleft()
best = max(best, len(dq))
return best
def theoretical_max(p):
w = p["window_s"]
if p["algorithm"] == "token_bucket":
return p["burst"] + p["rate"] * w
if p["algorithm"] == "fixed_window":
return 2 * p["limit"]
return p["limit"]
def sustained_rps(p):
return p["rate"] if p["algorithm"] == "token_bucket" else p["limit"] / p["window_s"]
# ----------------------------------------------------------------------------- analysis
def analyse(cfg, reqs):
stats, pstats, admitted, sample = simulate(cfg, reqs)
pols = cfg["policies"]
kinds = {c["name"]: c["kind"] for c in cfg["clients"]}
findings = []
def add(sev, code, msg):
findings.append({"severity": sev, "code": code, "message": msg})
per_sec = defaultdict(int)
retry_sec = defaultdict(int)
for t, _, att in admitted:
per_sec[int(t)] += 1
if att:
retry_sec[int(t)] += 1
peak_rps = max(per_sec.values()) if per_sec else 0
peak_sec = max(per_sec, key=per_sec.get) if per_sec else 0
clients = []
for name in sorted(stats, key=lambda n: (-stats[n]["logical"], n)):
s = stats[name]
fail_pct = 100.0 * s["failed"] / s["logical"] if s["logical"] else 0.0
hurt_pct = 100.0 * (s["failed"] + s["delayed"]) / s["logical"] if s["logical"] else 0.0
top = max(s["blocked_by"], key=s["blocked_by"].get) if s["blocked_by"] else ""
clients.append({"client": name, "kind": kinds.get(name, "unknown"), "logical": s["logical"],
"attempts": s["attempts"], "allowed": s["allowed"], "rejected": s["rejected"],
"failed": s["failed"], "failed_pct": round(fail_pct, 1),
"delayed": s["delayed"],
"amplification": round(s["attempts"] / s["logical"], 2) if s["logical"] else 0,
"max_retry_after_s": round(s["max_retry_after"], 1), "top_policy": top,
"first_reject_s": None if s["first_reject"] is None else round(s["first_reject"], 1)})
k = kinds.get(name, "unknown")
if k == "legit" and hurt_pct > 1.0:
add("high", "RL01", "Legit client %s is throttled on %.1f%% of its requests (%d failed, %d only after "
"retrying, waits up to %.1fs), mostly by policy '%s'; raise that limit or give this client its own tier."
% (name, hurt_pct, s["failed"], s["delayed"], s["max_retry_after"], top))
elif k == "legit" and s["rejected"]:
add("low", "RL01", "Legit client %s sees %d 429s but its retries recover; check it handles Retry-After."
% (name, s["rejected"]))
if k == "abusive" and s["logical"]:
share = 100.0 * s["allowed"] / s["logical"]
if share > 25:
add("medium", "RL02", "Abusive client %s still gets %.0f%% of its requests through (%d); "
"tighten the per-client limit or add a stricter policy for its pattern." % (name, share, s["allowed"]))
amp = s["attempts"] / s["logical"] if s["logical"] else 0
if amp >= 1.5:
add("medium", "RL05", "Client %s sends %.2fx attempts per logical request because it retries throttled "
"calls; retries count against the limit and add load - require Retry-After or backoff with jitter."
% (name, amp))
if per_sec and retry_sec.get(peak_sec, 0) > 0.5 * peak_rps and peak_rps >= 20:
add("medium", "RL10", "Retry herd: %d of the %d requests admitted in second %d are retries released by the same "
"Retry-After / window reset; add jitter to Retry-After (or client backoff) and prefer a token bucket or "
"sliding window so capacity is not handed out all at once." % (retry_sec[peak_sec], peak_rps, peak_sec))
cap = cfg.get("capacity_rps")
if cap:
if peak_rps > cap:
add("high", "RL03", "Admitted traffic peaks at %d req/s (second %d) over the backend capacity of %g req/s; "
"add or lower a global limit." % (peak_rps, peak_sec, cap))
if not any(p["key"] == "global" for p in pols):
per_client = [sustained_rps(p) for p in pols if p["key"] in ("client", "client+endpoint")
and not p["endpoints"] and not p["clients"]]
if per_client:
n = len(stats)
add("medium", "RL03", "No global policy: %d clients x %g req/s per client = %g req/s possible vs capacity %g."
% (n, min(per_client), n * min(per_client), cap))
policies = []
for p, ps in zip(pols, pstats):
obs = max((max_in_window(ts, p["window_s"]) for ts in ps["allowed_times"].values()), default=0)
theo = theoretical_max(p)
policies.append({"name": p["name"], "algorithm": p["algorithm"], "key": p["key"],
"endpoints": p["endpoints"], "clients": p["clients"], "exclude_clients": p["exclude_clients"], "checks": ps["checks"], "rejects": ps["rejects"],
"sustained_rps": round(sustained_rps(p), 3), "window_s": p["window_s"],
"worst_case_per_window": round(theo, 1), "observed_max_per_window": obs})
if p["algorithm"] == "fixed_window" and obs > p["limit"]:
add("medium", "RL04", "Fixed window '%s' let %d requests through in one rolling %gs window (limit %g, "
"%.1fx) across a window boundary; use sliding_counter or a token bucket if the limit protects capacity."
% (p["name"], obs, p["window_s"], p["limit"], obs / p["limit"]))
if p["key"] == "global" and ps["rejects"]:
allowed_by = defaultdict(int)
for t, cl, _ in admitted:
allowed_by[cl] += 1
total = sum(allowed_by.values()) or 1
hog = max(allowed_by, key=allowed_by.get)
victims = [c for c in ps["by_client"] if kinds.get(c) == "legit" and c != hog]
if allowed_by[hog] / total > 0.5 and victims:
add("high" if kinds.get(hog) != "legit" else "medium", "RL06",
"Noisy neighbour: %s holds %.0f%% of admitted traffic while global policy '%s' rejects %s; "
"add a per-client limit below the global one." % (hog, 100.0 * allowed_by[hog] / total,
p["name"], ", ".join(sorted(victims))))
if p["algorithm"] == "sliding_log" and p["limit"] > 1000:
add("low", "RL07", "Sliding log '%s' stores up to %d timestamps per key; prefer sliding_counter for large limits."
% (p["name"], p["limit"]))
if ps["checks"] and not ps["rejects"] and obs and theo > 10 * obs:
add("low", "RL09", "Policy '%s' never triggered; its worst case (%g per %gs) is >10x the busiest key (%d). "
"Fine as a safety net, but it will not stop a surge." % (p["name"], theo, p["window_s"], obs))
if sample:
add("info", "RL08", "Return 429 with Retry-After on every rejection (see the sample headers) and document the "
"limits so clients can pace themselves.")
order = {"high": 0, "medium": 1, "low": 2, "info": 3}
findings.sort(key=lambda f: (order[f["severity"]], f["code"]))
return {"service": cfg.get("service", "API"), "duration_s": cfg["duration_s"], "requests": len(reqs),
"admitted": len(admitted), "peak_admitted_rps": peak_rps, "peak_second": peak_sec,
"capacity_rps": cap, "clients": clients, "policies": policies, "findings": findings,
"sample_429": sample}
# ----------------------------------------------------------------------------- report
def fmt(x):
return ("%g" % x) if isinstance(x, float) else str(x)
def scope(p):
parts = []
if p["endpoints"]:
parts.append(", ".join(p["endpoints"]))
if p["clients"]:
parts.append("only " + ", ".join(p["clients"]))
if p["exclude_clients"]:
parts.append("not " + ", ".join(p["exclude_clients"]))
return "; ".join(parts) or "all"
def report(r):
L = ["# Rate limit simulation: %s" % r["service"], ""]
cap = (" vs capacity %g" % r["capacity_rps"]) if r["capacity_rps"] else ""
L.append("Simulated %gs, %d logical requests, %d admitted, peak %d req/s admitted (second %d)%s."
% (r["duration_s"], r["requests"], r["admitted"], r["peak_admitted_rps"], r["peak_second"], cap))
highs = sum(f["severity"] == "high" for f in r["findings"])
L += ["", "**Verdict: %s**" % ("%d high finding(s) - fix before shipping these limits" % highs if highs
else "policies hold - no high findings"), ""]
L += ["## Clients", "", "| Client | Kind | Requests | Attempts | Allowed | 429s | OK after retry | Failed | Failed % | Max Retry-After | Top policy |",
"|---|---|---|---|---|---|---|---|---|---|---|"]
for c in r["clients"]:
L.append("| %s | %s | %d | %d | %d | %d | %d | %d | %.1f%% | %ss | %s |" % (
c["client"], c["kind"], c["logical"], c["attempts"], c["allowed"], c["rejected"], c["delayed"], c["failed"],
c["failed_pct"], fmt(c["max_retry_after_s"]), c["top_policy"] or "-"))
L += ["", "## Policies", "", "| Policy | Algorithm | Key | Scope | Sustained | Worst case / window | Busiest key / window | Checks | Rejects |",
"|---|---|---|---|---|---|---|---|---|"]
for p in r["policies"]:
L.append("| %s | %s | %s | %s | %s req/s | %s per %ss | %d | %d | %d |" % (
p["name"], p["algorithm"], p["key"], scope(p), fmt(p["sustained_rps"]),
fmt(p["worst_case_per_window"]), fmt(p["window_s"]), p["observed_max_per_window"], p["checks"], p["rejects"]))
L += ["", "## Findings", ""]
if not r["findings"]:
L.append("- none")
for f in r["findings"]:
L.append("- **%s %s** %s" % (f["severity"].upper(), f["code"], f["message"]))
s = r["sample_429"]
if s:
L += ["", "## Sample 429 (first rejection, t=%ss, %s %s)" % (s["t"], s["client"], s["endpoint"]), "", "```",
"HTTP/1.1 429 Too Many Requests",
"Retry-After: %d" % s["retry_after"],
'RateLimit-Policy: "%s";q=%s;w=%s' % (s["policy"], fmt(float(s["quota"])), fmt(float(s["window_s"]))),
'RateLimit: "%s";r=0;t=%d' % (s["policy"], s["retry_after"]),
"Content-Type: application/problem+json", "```",
"Header names follow the IETF httpapi ratelimit-headers draft; older gateways use "
"X-RateLimit-Limit / -Remaining / -Reset instead."]
return "\n".join(L) + "\n"
# ----------------------------------------------------------------------------- self-test
def self_test():
cases = []
def check(name, cond, detail=""):
cases.append((name, bool(cond), detail))
tb = TokenBucket(rate=1, burst=3)
ok = 0
for i in range(5):
a, w, _ = tb.check("k", 0.0, 1)
if a:
tb.consume("k", 0.0, 1)
ok += 1
check("token bucket burst of 3 at t=0", ok == 3, ok)
a, w, _ = tb.check("k", 0.0, 1)
check("token bucket retry-after 1s", (not a) and abs(w - 1.0) < 1e-6, w)
a, w, _ = tb.check("k", 1.0, 1)
check("token bucket refills 1 token per second", a)
fw = FixedWindow(limit=10, window=60)
times = [59.0 + i * 0.01 for i in range(10)] + [60.0 + i * 0.01 for i in range(10)]
allowed = []
for t in times:
a, _, _ = fw.check("k", t, 1)
if a:
fw.consume("k", t, 1)
allowed.append(t)
check("fixed window boundary burst = 2x limit", max_in_window(allowed, 60) == 20, len(allowed))
a, w, _ = fw.check("k", 61.0, 1)
check("fixed window retry-after until reset", (not a) and abs(w - 59.0) < 1e-6, w)
sl = SlidingLog(limit=10, window=60)
allowed = []
for t in times:
a, _, _ = sl.check("k", t, 1)
if a:
sl.consume("k", t, 1)
allowed.append(t)
check("sliding log never exceeds limit", max_in_window(allowed, 60) == 10, len(allowed))
a, w, _ = sl.check("k", 60.5, 1)
check("sliding log retry-after = oldest + window", (not a) and abs(w - (59.0 + 60 - 60.5)) < 1e-6, w)
sc = SlidingCounter(limit=10, window=60)
allowed = []
for t in times + [90.0 + i for i in range(10)]:
a, _, _ = sc.check("k", t, 1)
if a:
sc.consume("k", t, 1)
allowed.append(t)
first = [t for t in allowed if t < 61]
later = [t for t in allowed if t >= 90]
check("sliding counter blocks the boundary burst", len(first) == 10, len(first))
check("sliding counter weights previous window (about half at mid-window)", 4 <= len(later) <= 7, len(later))
cfg = normalize({"service": "t", "duration_s": 60, "capacity_rps": 1000, "policies": [
{"name": "pc", "algorithm": "token_bucket", "key": "client", "rate": 5, "burst": 5}],
"clients": [{"name": "good", "kind": "legit", "rps": 2},
{"name": "bad", "kind": "abusive", "rps": 20, "retry": "immediate"}]})
r = analyse(cfg, generate(cfg))
by = {c["client"]: c for c in r["clients"]}
check("legit client under limit not throttled", by["good"]["rejected"] == 0, by["good"])
check("abusive client mostly throttled", by["bad"]["allowed"] <= 5 * 60 + 10, by["bad"]["allowed"])
check("immediate retries amplify attempts", by["bad"]["amplification"] > 1.5, by["bad"]["amplification"])
codes = {f["code"] for f in r["findings"]}
check("RL05 retry amplification finding", "RL05" in codes, codes)
cfg2 = normalize({"duration_s": 30, "capacity_rps": 10, "policies": [
{"name": "g", "algorithm": "fixed_window", "key": "global", "limit": 600, "window_s": 60}],
"clients": [{"name": "hog", "kind": "abusive", "rps": 30},
{"name": "app", "kind": "legit", "rps": 1, "start": 25, "duration": 5}]})
r2 = analyse(cfg2, generate(cfg2))
codes2 = {f["code"]: f["severity"] for f in r2["findings"]}
check("RL03 over capacity is high", codes2.get("RL03") == "high", codes2)
check("RL06 noisy neighbour on global", "RL06" in codes2, codes2)
check("RL01 legit client starved", codes2.get("RL01") == "high", codes2)
cfg3 = normalize({"duration_s": 10, "policies": [
{"name": "login", "algorithm": "sliding_log", "key": "client", "limit": 2, "window_s": 60, "endpoints": ["/login"]}],
"clients": [{"name": "u", "rps": 1, "endpoint": "/orders"}, {"name": "v", "rps": 1, "endpoint": "/login"}]})
r3 = analyse(cfg3, generate(cfg3))
by3 = {c["client"]: c for c in r3["clients"]}
check("endpoint-scoped policy ignores other endpoints", by3["u"]["rejected"] == 0 and by3["v"]["allowed"] == 2,
(by3["u"], by3["v"]))
check("burst pattern generates bursts", len(generate(normalize({"duration_s": 120, "policies": cfg3["policies"],
"clients": [{"name": "b", "pattern": "burst", "burst": 5, "every_s": 60}]}))) == 10)
cfg4 = normalize({"duration_s": 10, "policies": [
{"name": "default", "algorithm": "token_bucket", "key": "client", "rate": 1, "burst": 1, "exclude_clients": ["vip"]},
{"name": "vip", "algorithm": "token_bucket", "key": "client", "rate": 10, "burst": 10, "clients": ["vip"]}],
"clients": [{"name": "vip", "rps": 5}, {"name": "std", "rps": 5}]})
by4 = {c["client"]: c for c in analyse(cfg4, generate(cfg4))["clients"]}
check("client tiers: vip unthrottled, std limited", by4["vip"]["rejected"] == 0 and by4["std"]["allowed"] <= 11,
(by4["vip"], by4["std"]))
try:
normalize({"policies": [{"algorithm": "leaky"}]})
check("bad algorithm rejected", False)
except InputError:
check("bad algorithm rejected", True)
rep = report(r2)
check("report has verdict and sample 429", "Verdict" in rep and "429 Too Many Requests" in rep)
for name, ok, detail in cases:
print("%s %s%s" % ("PASS" if ok else "FAIL", name, "" if ok else " -> %r" % (detail,)))
passed = sum(1 for _, ok, _ in cases if ok)
print("%d/%d passed" % (passed, len(cases)))
return 0 if passed == len(cases) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Simulate API rate limit policies against traffic.")
ap.add_argument("config", nargs="?")
ap.add_argument("--trace", help="CSV with t,client[,endpoint][,cost] instead of generated traffic")
ap.add_argument("--json", action="store_true")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
if not a.config:
ap.print_usage(sys.stderr)
return 2
try:
with open(a.config, encoding="utf-8") as f:
cfg = normalize(json.load(f))
reqs = read_trace(a.trace) if a.trace else generate(cfg)
if not reqs:
raise InputError("no traffic: add clients to the config or pass --trace")
r = analyse(cfg, reqs)
except (InputError, OSError, json.JSONDecodeError) as e:
print("error: %s" % e, file=sys.stderr)
return 2
print(json.dumps(r, indent=2) if a.json else report(r), end="" if not a.json else "\n")
return 1 if any(f["severity"] == "high" for f in r["findings"]) else 0
if __name__ == "__main__":
sys.exit(main())