broad-announce/M5_VERIFICATION.md

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)

# 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.