feat(spec): source protection (throttling, quotas, circuit breaking)

SPEC §22 (new), 7 layers evaluated in order per request:
 1. payload-size cap (256 KB on every transport)
 2. per-IP / per-connection concurrency cap
 3. per-source token bucket (Redis)
 4. per-company token bucket (Redis)
 5. schema validate (strict JSON Schema / proto)
 6. circuit breaker on the broker (NATS)
 7. per-source quarantine (100 hits/60s => 5 min hard block)

- Per-source config additive to §4 schema
  (rate_limit_per_sec, max_payload_bytes,
   max_concurrent_connections, quarantine_*)
- Per-company gets the rate_limit column only
- Rejection-code matrix pinned per transport
  (HTTP 429 / WS close 1013 / MQTT 0x97 / gRPC Ack{RATE_LIMITED})
- Prometheus metrics: payload_rejected, connection_rejected,
  rate_limited{scope}, schema_rejected, circuit_breaker_state,
  source_quarantined, source_quarantine_active,
  rejection_latency_seconds
- Milestone rollout: M0 layer 3, M1 layers 1/4/5, M5 layer 2,
  M6 dedupe-aware shaping, M9 layers 6+7
- M10 exit adds runaway-source test (loadgen fault-injection:
  one source at 10x cap must not push p99 for other sources
  above 5s)
- Out of v1: WAF/CRS, per-payload-cost rate, auto-tuning,
  geo-fencing, billing integration

PROMPT logs the decision.

PROMPT.md

@@ -82,3 +82,31 @@
   in-cluster path at 5k/s (M10) and the broker+router path at
   50k/s (M10-bench) with the same source-protocol clients we
   ship. The `loadgen/` tool is the new home for that.
+
+**2026-06-13 — three open questions resolved**
+- (Q1) **Yes, build the fake-FCM / fake-Telegram / fake-SMS
+  servers** (`testfakes/`). M10 must not burn 50k FCM credits.
+- (Q2) **M10-bench delivery stub lives at the service level** —
+  a no-op `deliverd` binary in the bench profile, not a NATS
+  subject drop. More realistic: we exercise the real broker +
+  router + the message shape `deliverd` would consume.
+- (Q3) **`loadgen` ships as a single Docker image** with all
+  four binaries; entrypoint picks one via the image's
+  `command:` field in the compose / k8s spec.
+
+**2026-06-13 — Source protection (throttling) is in**
+- Spec §22 (new). Seven layers, evaluated in order per request:
+  payload-size cap → per-IP conn cap → per-source token bucket
+  → per-company token bucket → schema validate → broker
+  circuit breaker → per-source quarantine.
+- Per-source config additive to §4 schema
+  (`rate_limit_per_sec`, `max_payload_bytes`,
+  `max_concurrent_connections`, `quarantine_*`).
+- Rejection-code matrix is the contract across HTTP/WS/MQTT/gRPC
+  (e.g. per-source rate → 429+Retry-After / Ack{RATE_LIMITED} /
+  MQTT reason 0x97 / close 1013).
+- Rollout tied to milestones: M0 layer 3, M1 layers 1/4/5,
+  M5 layer 2, M6 dedupe-aware shaping, M9 layers 6+7.
+- Runaway-source test is part of **M10** exit (loadgen
+  fault-injection: one source at 10× cap must not push p99
+  for *other* sources above 5s).

SPEC.md

@@ -668,20 +668,216 @@ the running stack with p99 ingest latency ≤ 100ms, then
 `loadgen-grpc --mode stress --cluster-target 50000` against
 broker+router stubbed for **M10-bench**.
 
-## 22. Milestones
+## 22. Source protection (throttling, quotas, circuit breaking)
+
+Multi-tenant ingest dies the moment one runaway source saturates the
+shared tier. Seven layers of protection, evaluated in this exact
+order per request, so the cheap gates fire first:
+
+```
+incoming message
+   │
+   ▼
+1. payload-size cap                  ── reject before parse
+   ▼
+2. per-IP / per-connection cap       ── reject DoS, no auth needed
+   ▼
+3. per-source token bucket           ── reject excess from this source
+   ▼
+4. per-company token bucket          ── reject excess summed across sources
+   ▼
+5. schema validate                   ── reject malformed
+   ▼
+6. circuit breaker (downstream)      ── fail fast if broker is sick
+   ▼
+7. per-source quarantine (cold)      ── hard block repeat offenders
+   ▼
+accept → dedupe → publish to NATS
+```
+
+### Layer 1 — Payload size cap
+
+| transport | cap | reject code |
+|---|---|---|
+| HTTP | 256 KB (`Content-Length` + net/http `MaxBytesReader`) | `413 Payload Too Large` |
+| WebSocket | 256 KB per frame | close `1009 Message Too Big` |
+| MQTT | 256 KB per PUBLISH | disconnect with reason code `0x95 Payload format invalid` |
+| gRPC | 256 KB (`grpc.MaxRecvMsgSize` per server) | `RESOURCE_EXHAUSTED` |
+
+256 KB is ~1000× a normal alert (a 256 B alert with 1 KB of `data{}`
+is the 99th percentile). Anything bigger is almost certainly a bug
+or an attack. Configurable per source via `sources.max_payload_bytes`.
+
+### Layer 2 — Per-IP / per-connection concurrency cap
+
+DoS hardening. Default caps:
+
+| transport | cap | rejection |
+|---|---|---|
+| HTTP | 64 concurrent conns per source IP | `429` + close |
+| WebSocket | 32 concurrent sockets per source IP | close `1013 Try Again Later` |
+| MQTT | 256 in-flight per `client_id` | broker holds the packet, no PUBACK |
+| gRPC | 256 in-flight messages per stream | sender-side backpressure via `Ack{retry_after_ms}` |
+
+Implementation: in-memory `sync.Map[ip]atomic.Int64` on `ingestd`,
+with a small janitor that prunes idle entries. No Redis on the hot
+path for this gate (would self-DoS).
+
+### Layer 3 — Per-source token bucket (Redis)
+
+Every `sources` row carries `rate_limit_per_sec` (default 100/s,
+overridable per source). The bucket is a Redis key:
+
+```
+key   = rl:source:{source_id}
+cap   = sources.rate_limit_per_sec
+fill  = cap tokens/sec
+```
+
+Token consumption = one `INCRBY` + (if first in window) `EXPIRE`.
+Use the `redis-cell` module if available (atomic, ~1 RTT); fall
+back to `INCR`+`EXPIRE` otherwise. On reject:
+
+| transport | response |
+|---|---|
+| HTTP | `429` + `Retry-After: <seconds>` + `X-RateLimit-*` headers |
+| WebSocket | server-sent `{"error":"rate_limited","retry_after_ms":N}`; client may close or pause |
+| MQTT | disconnect reason code `0x97 Quota exceeded` (or hold packet — broker config) |
+| gRPC | `Ack{error: RATE_LIMITED, retry_after_ms}` — does *not* close the stream |
+
+`429`s are **not** counted as faults by `ingestd`; they are the
+documented contract. The `loadgen` `--rate` knob uses these
+response codes to drive fault-injection tests.
+
+### Layer 4 — Per-company token bucket (Redis)
+
+Same shape, looser limit. Key `rl:company:{company_id}`, value =
+`companies.rate_limit_per_sec` (default 10k/s, sum of all sources
+under normal use). Catches the case where a company has 100
+sources, each under their per-source cap, but the *sum* saturates
+the broker.
+
+### Layer 5 — Schema validate
+
+Strict JSON Schema (or `.proto` for gRPC). Reject:
+
+- Unknown top-level fields (forward-compat + catches misbehaving clients)
+- Missing `severity`, `company_id`, `source_id`
+- `severity` not in {`info`, `warning`, `critical`, `inminent_colapse`}
+- `dedupe_key` longer than 128 chars
+- `data` object with > 64 keys or > 8 KB serialized
+
+Cost: ~50µs per message, negligible. Errors return `400` / gRPC
+`INVALID_ARGUMENT`. Logged at `warn` level; counted in
+`ingestd_schema_rejects_total{reason}`.
+
+### Layer 6 — Circuit breaker on the broker (NATS)
+
+If NATS JetStream's publish path is unhealthy (ack timeout > 250ms
+3 times in a row, or stream full), `ingestd` opens a circuit
+breaker and immediately returns `503` to all sources. Sources are
+expected to retry with backoff. We do **not** queue alerts in
+`ingestd` itself — that's what the broker is for.
+
+States: `closed` (normal) → `open` (reject everything) → after
+30s cool-down, `half-open` (let one request through) → `closed` on
+success, `open` on failure. Standard `sony/gobreaker` config.
+
+### Layer 7 — Per-source quarantine (the nuclear option)
+
+If a source crosses **N=100 rate-limit hits in M=60s** *after* we
+already sent the proper `429`s, we **temporarily quarantine** the
+source for 5 min:
+
+- All requests answered `403` from a local in-memory map (no Redis
+  call). Protects Redis from being hammered by the same client.
+- Operator gets a Prometheus alert
+  (`ingestd_source_quarantined_total`) and a webhook into
+  `admind` for incident review.
+- After 5 min, quarantine auto-lifts. Persistent offenders are
+  flagged in the admin UI for manual disable.
+
+Quarantine is a **safety valve**, not a primary mechanism. The
+token buckets should make it rare.
+
+### Per-source config (additive to schema in §4)
+
+```sql
+ALTER TABLE sources
+  ADD COLUMN rate_limit_per_sec          INTEGER NOT NULL DEFAULT 100,
+  ADD COLUMN max_payload_bytes           INTEGER NOT NULL DEFAULT 262144,
+  ADD COLUMN max_concurrent_connections  INTEGER NOT NULL DEFAULT 64,
+  ADD COLUMN quarantine_hits_threshold   INTEGER NOT NULL DEFAULT 100,
+  ADD COLUMN quarantine_window_seconds  INTEGER NOT NULL DEFAULT 60,
+  ADD COLUMN quarantine_duration_seconds INTEGER NOT NULL DEFAULT 300;
+```
+
+Same columns on `companies` for the per-company bucket
+(`rate_limit_per_sec` only — the rest are source-scoped).
+
+### Prometheus metrics (additive)
+
+```
+ingestd_payload_rejected_total{transport,reason}      counter
+ingestd_connection_rejected_total{transport}          counter
+ingestd_rate_limited_total{transport,scope}           counter
+# scope ∈ {source, company, quarantine}
+ingestd_schema_rejected_total{reason}                 counter
+ingestd_circuit_breaker_state{component}              gauge   # 0=closed 1=half 2=open
+ingestd_source_quarantined_total{source_id,company_id} counter
+ingestd_source_quarantine_active{source_id}           gauge
+ingestd_rejection_latency_seconds_bucket{transport,reason} histogram
+```
+
+### Rejection code matrix (the contract)
+
+| reason | HTTP | WS | MQTT | gRPC |
+|---|---|---|---|---|
+| payload too large | 413 | close 1009 | reason 0x95 | RESOURCE_EXHAUSTED |
+| per-IP conn cap | 429 + close | close 1013 | reason 0x97 | n/a (per-stream cap) |
+| per-source rate | 429 + Retry-After | `{"error":"rate_limited"}` | reason 0x97 | `Ack{error:RATE_LIMITED,retry_after_ms}` |
+| per-company rate | 429 + Retry-After | same as above | same | same |
+| schema invalid | 400 | `{"error":"invalid"}` | reason 0x99 | INVALID_ARGUMENT |
+| broker down (cb open) | 503 + Retry-After | `{"error":"unavailable"}` | disconnect | `Ack{error:INTERNAL,retry_after_ms:1000}` |
+| quarantined | 403 | close 1008 | disconnect | `Ack{error:UNAUTHENTICATED}` + close |
+
+### Milestone rollout
+
+| milestone | layers shipped |
+|---|---|
+| M0 | layer 3 (per-source bucket, simplest impl) |
+| M1 | + layers 1, 4, 5 (size cap, per-company, schema) |
+| M5 | + layer 2 (per-IP cap, obvious once WS is in) |
+| M6 | + dedupe-aware rate shaping (don't burn a token on a dupe) |
+| M9 | + layer 6 (circuit breaker), layer 7 (quarantine), all metrics + dashboards |
+| M10 | prove one runaway source can't push p99 above 5s for *other* sources (loadgen fault-injection) |
+
+### What we explicitly do NOT add in v1
+
+- **WAF / OWASP CRS** in front of `ingestd`: alerts don't carry
+  SQL/HTML; nginx with CRS is fine if you want it but out of
+  spec.
+- **Per-payload-cost rate limiting** (fat alerts cost more tokens):
+  too clever for v1, easy to bolt on.
+- **Auto-tuning of `rate_limit_per_sec`**: per-source is a config
+  value, ops sets it.
+- **Geo-fencing**: until v2 multi-region.
+- **Quota reset emails / billing integration**: separate concern.
+
+## 23. Milestones
 
 | # | milestone | exit criterion |
 |---|---|---|
 | M0 | Skeleton + docker-compose up | Postgres + NATS + Redis up, all 4 services start, /health green |
-| M1 | HTTP POST ingest end-to-end | send a signed webhook → FCM test push to a fake device |
+| M1 | HTTP POST ingest end-to-end | send a signed webhook → FCM test push to a fake device; layers 1, 3, 4, 5 in |
 | M2 | Recipient resolution | per-source `allowed_targets` honored, subscriptions applied |
 | M3 | Telegram delivery + bot commands | user can `/subscribe` and receive an alert via Telegram |
 | M4 | MQTT ingest | EMQX up, QoS 1, per-company topic ACLs |
-| M5 | WebSocket ingest + live tail | admin UI (or wscat) sees alerts as they arrive |
-| M6 | Dedupe + dedupe_count | burst of 100 identical alerts → recipient sees "×100" |
+| M5 | WebSocket ingest + live tail | admin UI (or wscat) sees alerts as they arrive; layer 2 in |
+| M6 | Dedupe + dedupe_count | burst of 100 identical alerts → recipient sees "×100"; dedupe-aware rate shaping in |
 | M7 | Timescale + ClickHouse | 7d retention + archive job |
 | M8 | DLQ + replay UI | operator can replay a failed delivery |
-| M9 | Observability (Prom/Grafana) | 1 dashboard per tier + per-company drilldown |
-| M10 | Load test 5k/s on docker-compose | soak 10 min, p99 ≤ 5s, zero DLQ, run via `loadgen` |
+| M9 | Observability (Prom/Grafana) | 1 dashboard per tier + per-company drilldown; layers 6, 7 in |
+| M10 | Load test 5k/s on docker-compose | soak 10 min, p99 ≤ 5s, zero DLQ, run via `loadgen`; runaway-source test passes |
 | M10-bench | Broker + router ceiling bench | 50k/s via `loadgen` against broker+router (delivery stubbed); p99 router latency ≤ 50ms; no broker backpressure |
 | M11 | gRPC bidi-streaming ingest | internal Go service pushes ≥ 10k alerts/sec on one stream, p99 server-side `Ack` ≤ 50ms |