broad-announce/M4_VERIFICATION.md

M4 Verification — MQTT ingest

This is the live smoke test for the M4 milestone (MQTT ingest, SPEC §16 + §18 + §22). It exercises the new path end-to-end:

1. EMQX is up with per-source ACLs (default deny on #). The bootstrap file deploy/emqx/auth-built-in-db-bootstrap.csv declares the sources; deploy/emqx/acl.conf restricts each user to their own ba/<co>/<src>/incoming topic.
2. internal/mqttclient — shared paho.MQTT wrapper used by both the ingestd subscriber and the new loadgen publisher.
3. cmd/ingestd/mqtt.go — MQTT subscriber that runs every message through the same processDeps.ProcessAlert pipeline as the HTTP POST handler (parse → validate → HMAC verify → rate-limit → dedupe → publish to NATS).
4. loadgen/cmd/mqtt/ — loadgen-mqtt publisher. Same data shape as loadgen-http (severity mix, dedupe ratio, burst mode).
5. Topic namespace — ba/<company_id>/<source_id>/incoming. Sources publish here; ingestd subscribes to the wildcard ba/+/+/incoming. The + is the EMQX single-level wildcard.
6. Auth model — twofold:
   • MQTT username/password for the broker-level gate. Username = <source_id>-<company_id>, password = HMAC secret (so the same secret serves both as MQTT auth and per-message HMAC). The acl.conf then gates which topic a given user can publish to.
   • Per-message X-BA-Signature in the JSON envelope's auth field. Same verifyHMAC() is reused unchanged.

Prerequisites

• Stack is up: docker compose up -d
• All M1–M3 migrations applied: docker compose up seed
• BA_INGESTD_SOURCES includes acme-001:prom-prod:s3cret-acme (and optionally globex-002:grafana:s3cret-globex for the ACL-violation test in step 5)

• The M4 ingestd env vars are set:

  BA_INGESTD_MQTT_BROKER=tcp://emqx:1883
  BA_INGESTD_MQTT_USERNAME=ingestd
  BA_INGESTD_MQTT_PASSWORD=ingestd-broker-only
  BA_INGESTD_MQTT_SUBSCRIBE=ba/+/+/incoming
  

• The EMQX bootstrap files are mounted (in compose by default): deploy/emqx/{acl.conf,auth-built-in-db-bootstrap.csv}.

• loadgen-mqtt is built:

  cd loadgen && go build -o /tmp/loadgen-mqtt ./cmd/mqtt
  

Automated smoke

bash scripts/m4_smoke.sh

The script:

1. Builds loadgen-mqtt and the 3 failure-path test binaries (m4_badsig, m4_acl_violation, m4_badjson) if not already present.
2. Walks the 5 step-by-step scenarios below, asserting the expected delivery counts and metric counter deltas.
3. Exits 0 on success, $fail_count on failure.

Step-by-step (also documented in M4_SMOKE_LOG.md)

Step 1 — confirm M4 wiring is up

# ingestd's MQTT subscriber is connected
docker logs broad-announce-ingestd-1 2>&1 | tail -3
# should show: "mqtt connected" + "mqtt subscribed" + "ba/+/+/incoming"

# EMQX dashboard is reachable
curl -sS http://localhost:18083/
# EMQX 5.x landing page

# emqx is healthy
docker inspect broad-announce-emqx-1 --format '{{.State.Health.Status}}'
# "healthy"

Step 2 — happy path: 1 alert via MQTT

/tmp/loadgen-mqtt \
  --broker tcp://localhost:1883 \
  --api-key acme-001:prom-prod:s3cret-acme \
  --count 1 --rate 5

Wait 4s, then check deliveries:

docker exec -i broad-announce-postgres-1 psql -U ba -d ba -c "
SELECT individual_id, channel, status FROM deliveries ORDER BY id;"

Expected: 2 rows — Alice fcm (sent) + Alice telegram (sent). Bob has no telegram_chat_id linked at this point (still in M2 default state for the smoke).

Ingestd's MQTT counters tick:

curl -sS http://localhost:8800/metrics | grep ba_ingestd_mqtt
# ba_ingestd_mqtt_messages_total{result="received",service="ingestd"} <N+1>
# ba_ingestd_mqtt_messages_total{result="accepted",service="ingestd"} <M+1>

Step 3 — 5 alerts, mixed severity, 30% dedupe

/tmp/loadgen-mqtt \
  --broker tcp://localhost:1883 \
  --api-key acme-001:prom-prod:s3cret-acme \
  --count 5 --rate 5 --mode normal

Expected: 5 distinct alert_ids, ≥ 10 deliveries (5 alerts × 2 channels for Alice; +4 per alert with severity ≥ critical for Bob). At least 1 dedupe (30% of alerts share a dedupe_key).

Ingestd counter mqtt_messages{result="deduped"} advances by ≥ 1.

Step 4 — bad signature

Build the test binary (script does it automatically):

cat > /tmp/m4_smoke/badsig.go <<'GO'
package main
import (
  "encoding/json"
  "fmt"
  mqtt "github.com/eclipse/paho.mqtt.golang"
  "os"
  "time"
)
func main() {
  opts := mqtt.NewClientOptions().
    AddBroker("tcp://localhost:1883").
    SetClientID("m4-badsig").
    SetUsername("prom-prod-acme-001").
    SetPassword("s3cret-acme").
    SetCleanSession(true)
  c := mqtt.NewClient(opts)
  if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
    fmt.Println("connect err:", tok.Error()); os.Exit(0)
  }
  body, _ := json.Marshal(map[string]any{
    "alert": json.RawMessage(`{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"bad sig"}`),
    "auth":  "t=1700000000,v1=deadbeef",
  })
  tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, body)
  if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
  fmt.Println("bad-sig published; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badsig /tmp/m4_smoke/badsig.go)
/tmp/m4_smoke/m4_badsig

Expected:

• 0 deliveries (HMAC failed, alert dropped)
• ingestd log: WARN msg="mqtt alert rejected" reason=bad_signature
• ba_ingestd_mqtt_messages_total{result="bad_signature"} ticks

Step 5 — ACL violation

cat > /tmp/m4_smoke/acl_violation.go <<'GO'
package main
import (
  "fmt"
  mqtt "github.com/eclipse/paho.mqtt.golang"
  "os"
  "time"
)
func main() {
  opts := mqtt.NewClientOptions().
    AddBroker("tcp://localhost:1883").
    SetClientID("m4-acl-violation").
    SetUsername("prom-prod-acme-001").  // <-- authed as acme source
    SetPassword("s3cret-acme").
    SetCleanSession(true)
  c := mqtt.NewClient(opts)
  if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
    fmt.Println("connect err:", tok.Error()); os.Exit(0)
  }
  // Try to publish to GLOBEX's topic — should be denied by ACL
  tok := c.Publish("ba/globex-002/grafana/incoming", 1, false, []byte("hack"))
  if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
  fmt.Println("published to wrong topic; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_acl_violation /tmp/m4_smoke/acl_violation.go)
/tmp/m4_smoke/m4_acl_violation

Expected:

• 0 deliveries (broker dropped the publish before the subscriber saw it)
• EMQX log: tag: AUTHZ, msg: cannot_publish_to_topic_due_to_not_authorized, ... topic: ba/globex-002/grafana/incoming
• ba_ingestd_mqtt_messages_total{result="received"} is unchanged (broker dropped before subscriber)

Step 6 — non-JSON body

cat > /tmp/m4_smoke/badjson.go <<'GO'
package main
import (
  "fmt"
  mqtt "github.com/eclipse/paho.mqtt.golang"
  "os"
  "time"
)
func main() {
  opts := mqtt.NewClientOptions().
    AddBroker("tcp://localhost:1883").
    SetClientID("m4-badjson").
    SetUsername("prom-prod-acme-001").
    SetPassword("s3cret-acme").
    SetCleanSession(true)
  c := mqtt.NewClient(opts)
  if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
    fmt.Println("connect err:", tok.Error()); os.Exit(0)
  }
  tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, []byte("not json"))
  if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
  fmt.Println("bad-json published; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badjson /tmp/m4_smoke/badjson.go)
/tmp/m4_smoke/m4_badjson

Expected:

• 0 deliveries (alert.Alert.Unmarshal failed)
• ingestd log: WARN msg="mqtt alert rejected" reason=invalid_json
• ba_ingestd_mqtt_messages_total{result="invalid_json"} ticks

What this proves about M4

1. Per-source ACLs work. The acl.conf file uses Erlang-term rules keyed on the username. A user can publish only to its own ba/<co>/<src>/incoming. Step 5 proves a different source can't publish to another source's topic — the broker drops the publish before any subscriber sees it.
2. The same pipeline serves HTTP and MQTT. Steps 2 + 3 show the same ProcessAlert path that ran M0–M3's HTTP-only smoke now runs the MQTT path with no code changes to the rest of the stack (rate limits, dedupe, NATS publish, alert.ID, alert.DedupeCount all behave identically).
3. Per-message HMAC is enforced. Step 4 proves that even an authenticated source can't bypass per-message auth by sending an arbitrary body.
4. Topic parsing is strict. parseIncomingTopic only accepts ba/<co>/<src>/incoming (4 segments, last = incoming). Anything else lands in mqtt_messages{result="bad_topic"} and is dropped. The broker's ACL is the first line of defense; this is the second.
5. Dedupe window is per-source, not per-transport. The same dedupe.Deduper (Redis-backed) is reused for HTTP and MQTT. Step 3's 30% dedupe ratio produces the expected dedupe count in the deliveries table.
6. Metrics are per-message. ba_ingestd_mqtt_messages_total{result="..."} exposes received/accepted/deduped/bad_signature/bad_topic/ invalid_json/... as separate labels. M9 promotes these to the alerts-received total.

Known quirks (and why the smoke script does what it does)

1. faketgmd's getUpdates is a fake (carried over from M3). The smoke script still calls docker compose restart telegramd between steps so the bot's in-process long-poll offset aligns with the reset state. Same quirk; same workaround.
2. EMQX's deny_action = disconnect is set, but in practice the paho client may reconnect fast enough that the test code sees "still connected" after a denied publish. The end-state is the same: the publish was dropped, no delivery happened, and received counter didn't tick.
3. The 3 failure-path test binaries (m4_badsig, m4_acl_violation, m4_badjson) are tiny, statically linked, and live in /tmp/m4_smoke/. The smoke script auto-builds them on first run. They are not committed to the repo.
4. EMQX env-var config beats emqx.conf. Earlier versions of the compose tried to volume-mount emqx.conf; EMQX 5.x rejects partial overrides there with node.cookie required_field (because emqx.conf is a full config, not a partial). The current compose uses env vars (EMQX_AUTHENTICATION__1__BACKEND, EMQX_AUTHORIZATION__NO_MATCH, etc.) and the volume mount is just acl.conf + the bootstrap CSV.

What stays out of M4 (and not supposed to be)

• Per-IP concurrency cap (M5 with WS, SPEC §22 layer 2)
• Circuit breaker + quarantine (M9)
• TLS to EMQX (M11, security milestone)
• Persistent sessions (M11)
• Per-company bot token resolution for sources that share a single EMQX user across companies (M3+ generalization)
• HTTP-style signature in a header (MQTT has no headers; the auth field in the JSON envelope is the equivalent)