# M5 Verification — WebSocket ingest + live tail + per-IP cap

**Status:** shipped 2026-06-14
**Branch:** master
**Commits:** see `git log --oneline | grep M5`

This milestone adds the third ingest transport (WebSocket) on top
of HTTP (M0) and MQTT (M4), plus the SPEC §22 "layer 2" per-IP
concurrency cap, plus an in-process live-tail endpoint for
operators (`GET /v1/tail/ws`).

## What landed

| Surface | Method | Path | Auth | Notes |
| --- | --- | --- | --- | --- |
| Ingest | `GET` (WS upgrade) | `/v1/ingest/ws` | first frame `{api_key}` | text-only, per-IP cap 32 |
| Live tail | `GET` (WS upgrade) | `/v1/tail/ws` | `?token=` / `Authorization: Bearer` / `X-BA-Tail-Token` | `?company_id=` filter |
| Metrics | `GET` | `/metrics` | n/a | `ba_ingestd_ws_*`, `ba_ingestd_tail_*`, `ba_ingestd_connection_rejected_total` |

## New env vars (ingestd)

| Var | Default | Purpose |
| --- | --- | --- |
| `BA_INGESTD_TAIL_TOKEN` | empty (disables) | static token for `/v1/tail/ws` |
| `BA_INGESTD_MAX_CONCURRENT_PER_IP` | 32 | layer-2 per-IP cap for WS ingest |

## Files

```
migrations/005_ws.up.sql           # adds max_concurrent_connections to sources/companies
migrations/005_ws.down.sql
internal/concurrency/perip.go      # sync.Map[ip]atomic.Int64 + janitor
internal/concurrency/perip_test.go # race-tested
internal/tailhub/hub.go            # in-process pub/sub, drop-on-full
internal/tailhub/hub_test.go       # fan-out + filter + drops
internal/wsclient/client.go        # thin gorilla wrapper
internal/wsclient/client_test.go
loadgen/cmd/ws/main.go             # loadgen-ws binary
loadgen/cmd/m5drivers/tail/main.go # /tmp/m5-tail-test driver
cmd/ingestd/ws.go                  # WS ingest endpoint
cmd/ingestd/wstail.go              # WS tail endpoint
cmd/ingestd/process.go             # tail.Publish fan-out (shared by all transports)
cmd/ingestd/main.go                # env wiring, deps composition
internal/observability/metrics.go  # 5 new metric vectors
docker-compose.yml                 # env vars on ingestd
```

## Scenarios (spec-style)

### Step 2 — happy path: 1 alert via WS → 2 deliveries
Client opens a WS, sends `{api_key}` auth frame, receives
`{result:"ready"}`, sends one alert envelope, receives
`{alert_id, result:"accepted"}`. The downstream router fans out
to Alice (fcm) and Alice (telegram) → 2 deliveries in Postgres.

**Asserted:**
- `ba_ingestd_ws_messages_total{result="received"}` +1
- `ba_ingestd_ws_messages_total{result="accepted"}` +1
- `deliveries` +2

### Step 3 — 5 alerts, mixed severity + 30% dedupe → 10 deliveries
Same source (`acme-001:prom-prod`), 5 alerts, default `normal`
profile (30% dedupe means ~2 of 5 share a `dedupe_key`). Alice
subscribes at `min_severity=info`; Bob at `critical`. So the
~3 new alerts all hit Alice, the critical subset also hits Bob.
With 2 channels each, expected ≥ 8 deliveries (often 10 if all
5 are new or all 2 are deduped on the same key).

**Asserted:**
- `ba_ingestd_ws_messages_total{result="received"}` +5
- `ba_ingestd_ws_messages_total{result="accepted"}` ≥3
- `deliveries` ≥8

### Step 4 — bad signature (correct user, wrong HMAC) → 0 deliveries
Auth passes (valid key), but the envelope `auth` string contains
a wrong v1 HMAC. ingestd's HMAC verify returns `bad_signature`,
the alert is rejected, no delivery, the counter ticks.

**Asserted:**
- `ba_ingestd_ws_messages_total{result="bad_signature"}` +1
- `deliveries` +0

### Step 5 — per-IP concurrency cap (35 conns, cap=32) → 30-32 ok + 2-3 rejected
A test driver opens 35 concurrent WS connections from the same
IP. The `concurrency.PerIP` gate allows the first 32, the next
3 are rejected at the upgrade. Each rejected conn is counted in
`ba_ingestd_connection_rejected_total{transport="ws"}`. A
`closed_per_ip_cap` state is also bumped on the connection
lifecycle counter.

**Asserted:**
- perip_test output: `ok=30..32 rejected=2..3`
- `ba_ingestd_connection_rejected_total{transport="ws"}` +2..3
- `ba_ingestd_ws_connections_total{state="closed_per_ip_cap"}` +2..3

### Step 6 — live tail (no filter): subscribe then send → 1 frame arrives
A test driver connects to `/v1/tail/ws?token=...` (no company
filter). The handler:
1. validates the token
2. subscribes to the tail hub **before** the upgrade completes
   (so events in the dial→subscribe window are not lost — see
   `cmd/ingestd/wstail.go` for the race that motivated this
   ordering)
3. upgrades, then enters the stream loop

After `tail_subscribers` reaches 1, the test sends 1 alert via
WS. The hub fans out, the tail handler writes the JSON frame
back, the test driver receives it.

**Asserted:**
- `ba_ingestd_tail_subscribers` = 1 while connected
- test driver receives ≥ 1 `FRAME:` line
- `ba_ingestd_tail_subscribers` returns to 0 after disconnect

### Step 7 — live tail (company filter): acme filtered, globex passes
Same as Step 6 but with `?company_id=globex-002`. The test sends
one acme alert (should be filtered out) and one globex alert
(should pass).

**Asserted:**
- 0 acme frames in tail output
- ≥ 1 globex frame in tail output

## Run

```
cd /root/.openclaw/workspace/broad-announce
bash scripts/m5_smoke.sh
```

Exit code = number of failed checks (0 on success).

## Curl sanity check (no test harness)

```bash
# Tail
wscat -c "ws://localhost:8800/v1/tail/ws?token=tail-dev-token-please-change-in-prod"

# Ingest (from another shell)
curl -i -X POST http://localhost:8800/v1/ingest \
  -H "X-BA-Key: acme-001:prom-prod:s3cret-acme" \
  -H "X-BA-Signature: t=1700000000,v1=$(echo -n '{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"curl-test"}' | openssl dgst -sha256 -hmac s3cret-acme | awk '{print $2}')" \
  -H "Content-Type: application/json" \
  -d '{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"curl-test"}'
```

## Out of scope for M5 (deferred)

- HTTP `POST /v1/ingest` per-IP cap — applied in M10 (the "M5-bump" item from SPEC §22).
- JWT-based tail auth — applied in M11 (security milestone).
- Cross-node tail fan-out (Redis pub/sub) — applied when the
  cluster is sharded (M15+), not relevant for the single-node
  baseline.