M10(2/5): per-source latency histogram label + m10_lib.py + m10_smoke.py

W2: assertion harness + per-source latency breakdown

New:
- internal/observability/metrics.go — PublishLatency changed from
  prometheus.Histogram to *prometheus.HistogramVec with source_id label.
  Enables per-source p99 isolation checks in the smoke.
- cmd/ingestd/process.go — PublishLatency.Observe() now passes a.SourceID
  as label so Prometheus can filter by source.
- scripts/m10_lib.py — shared assertion library: scrape(), assert_ingest_p99_under(),
  per_source_p99(), assert_dlq_count_equals(), assert_rate_near(),
  assert_no_rate_limit_storm(), wait_until(), get_circuit_breaker_state()
- scripts/m10_smoke.py — 5-step M10 soak smoke:
  Step 1 pre-flight (services up, DLQ baseline, CB closed)
  Step 2 start 3-instance loadgen + 10min soak monitoring (30s samples)
  Step 3 runaway-source fault injection (placeholder — W4 enhancement)
  Step 4 DLQ invariant
  Step 5 teardown
  Note: Step 3 (runaway-source) is a placeholder; real implementation
  needs a 4th loadgen instance at 10× cap targeting acme-001.

cmd/ingestd/process.go

@@ -286,7 +286,7 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 		recordHit()
 		return Reject("broker_unavailable", 503, publishErr.Error())
 	}
-	d.Metrics.PublishLatency.Observe(time.Since(start).Seconds())
+	d.Metrics.PublishLatency.WithLabelValues(a.SourceID).Observe(time.Since(start).Seconds())
 	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
 
 	if isNew {

internal/observability/metrics.go

@@ -23,7 +23,7 @@ type IngestdMetrics struct {
 	RateLimitHits  *prometheus.CounterVec   // scope=source|company
 	Quarantines    *prometheus.CounterVec
 	CBState        *prometheus.GaugeVec
-	PublishLatency prometheus.Histogram
+	PublishLatency *prometheus.HistogramVec // labels: source_id
 	// MQTTMessages is the M4 per-message counter; labels mirror
 	// the same result taxonomy as AlertsReceived (accepted,
 	// deduped, bad_topic, bad_signature, unknown_source,
@@ -101,14 +101,14 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 			Help:      "0=closed, 1=half_open, 2=open.",
 			ConstLabels: prometheus.Labels{"service": serviceName},
 		}, []string{"component"}),
-		PublishLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
-			Namespace: "ba",
-			Subsystem: "ingestd",
-			Name:      "publish_latency_seconds",
-			Help:      "Time to publish an accepted alert to NATS.",
-			Buckets:   prometheus.DefBuckets,
-			ConstLabels: prometheus.Labels{"service": serviceName},
-		}),
+		PublishLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace:   "ba",
+			Subsystem:    "ingestd",
+			Name:         "publish_latency_seconds",
+			Help:         "Time to publish an accepted alert to NATS.",
+			Buckets:      prometheus.DefBuckets,
+			ConstLabels:  prometheus.Labels{"service": serviceName},
+		}, []string{"source_id"}),
 		MQTTMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Namespace: "ba",
 			Subsystem: "ingestd",

scripts/m10_lib.py

@@ -0,0 +1,190 @@
+"""
+m10_lib.py — shared assertion library for M10 smoke scripts.
+Used by both m10_smoke.py (5k/s soak) and m10_bench_smoke.py (50k/s broker ceiling).
+
+All Prometheus queries use URL-encoding-safe Python urllib to avoid the
+bashquoting bugs that plagued m9_smoke.sh.
+"""
+import urllib.request
+import urllib.parse
+import time
+import sys
+
+PROM = "http://localhost:9090"
+PROM_TIMEOUT = 10  # seconds
+
+
+def scrape(query: str) -> list[dict]:
+    """
+    Run a Prometheus query and return the result vector.
+    Returns [] on error or no data.
+    """
+    url = f"{PROM}/api/v1/query?query={urllib.parse.quote(query, safe='')}"
+    try:
+        with urllib.request.urlopen(url, timeout=PROM_TIMEOUT) as r:
+            data = r.read()
+        d = __import__("json").loads(data)
+        if d.get("status") != "success":
+            return []
+        return d.get("data", {}).get("result", [])
+    except Exception:
+        return []
+
+
+def assert_ingest_p99_under(threshold: float, window_seconds: int = 60,
+                             label_filter: str = "") -> float:
+    """
+    Assert p99 ingest latency (ba_ingestd_publish_latency_seconds) is under threshold.
+    Uses histogram_quantile over the given sliding window.
+    Returns the actual p99 value, or raises AssertionError.
+    """
+    filter_expr = f',{{{label_filter}}}' if label_filter else ''
+    query = (
+        f'histogram_quantile(0.99, '
+        f'rate(ba_ingestd_publish_latency_seconds_bucket{{{filter_expr}}}[{window_seconds}s]))'
+    )
+    results = scrape(query)
+    if not results:
+        raise AssertionError(
+            f"p99 query returned no data. Is ingestd up and generating load? "
+            f"(query: {query})"
+        )
+    p99 = float(results[0]["value"][1])
+    if p99 > threshold:
+        raise AssertionError(
+            f"p99 ingest latency {p99:.3f}s exceeds threshold {threshold}s "
+            f"(window={window_seconds}s, filter='{label_filter}')"
+        )
+    return p99
+
+
+def per_source_p99(source_id: str, window_seconds: int = 60) -> float:
+    """
+    Return the p99 publish latency for a specific source_id.
+    Returns 0.0 if no data.
+    """
+    query = (
+        f'histogram_quantile(0.99, '
+        f'rate(ba_ingestd_publish_latency_seconds_bucket{{source_id="{source_id}"}}[{window_seconds}s]))'
+    )
+    results = scrape(query)
+    if not results:
+        return 0.0
+    return float(results[0]["value"][1])
+
+
+def assert_dlq_count_equals(expected: int, window_seconds: int) -> int:
+    """
+    Assert the DLQ row count (delta over window) equals expected.
+    Uses Prometheus counter delta (ba_deliverd_dlq_total).
+    Returns the actual count.
+    """
+    query = (
+        f'increase(ba_deliverd_dlq_total[{window_seconds}s])'
+    )
+    results = scrape(query)
+    if not results:
+        # No DLQ entries at all
+        actual = 0
+    else:
+        actual = int(float(results[0]["value"][1]))
+    if actual != expected:
+        raise AssertionError(
+            f"DLQ count {actual} != expected {expected} "
+            f"(window={window_seconds}s). Run 'docker exec ... psql ... "
+            f"SELECT count(*) FROM deliveries_dlq' for precise count."
+        )
+    return actual
+
+
+def assert_rate_near(target: float, tolerance: float = 0.05,
+                      window_seconds: int = 30) -> float:
+    """
+    Assert the cluster-wide accept rate is within tolerance of target.
+    tolerance=0.05 means ±5%.
+    Returns the actual rate.
+    """
+    query = (
+        f'rate(ba_ingestd_alerts_received_total{{result="accepted"}}[{window_seconds}s])'
+    )
+    results = scrape(query)
+    if not results:
+        raise AssertionError(
+            f"rate query returned no data. Is ingestd receiving alerts? (query: {query})"
+        )
+    # Sum all series (multiple source_ids)
+    actual = sum(float(r["value"][1]) for r in results)
+    lower = target * (1 - tolerance)
+    upper = target * (1 + tolerance)
+    if not (lower <= actual <= upper):
+        raise AssertionError(
+            f"rate {actual:.1f}/s is outside ±{tolerance*100:.0f}% "
+            f"tolerance of target {target}/s (range: {lower:.1f}–{upper:.1f})"
+        )
+    return actual
+
+
+def assert_no_rate_limit_storm(window_seconds: int = 30) -> float:
+    """
+    Assert rate-limited hits are < 5% of total accepted.
+    Returns the rate-limited fraction.
+    """
+    accepted_q = f'rate(ba_ingestd_alerts_received_total{{result="accepted"}}[{window_seconds}s])'
+    rl_q = f'rate(ba_ingestd_alerts_received_total{{result="rate_limited"}}[{window_seconds}s])'
+    accepted_results = scrape(accepted_q)
+    rl_results = scrape(rl_q)
+    accepted = sum(float(r["value"][1]) for r in accepted_results) if accepted_results else 0
+    rl = sum(float(r["value"][1]) for r in rl_results) if rl_results else 0
+    if accepted == 0:
+        return 0.0
+    frac = rl / accepted
+    if frac > 0.05:
+        raise AssertionError(
+            f"rate-limited fraction {frac:.1%} exceeds 5% threshold "
+            f"(accepted={accepted:.1f}/s, rate_limited={rl:.1f}/s)"
+        )
+    return frac
+
+
+def wait_until(fn, timeout_seconds: float, interval: float = 1.0, name: str = ""):
+    """
+    Poll fn() every interval until it returns truthy or timeout.
+    Returns fn() result on success, raises on timeout.
+    """
+    deadline = time.time() + timeout_seconds
+    last_err = None
+    while time.time() < deadline:
+        try:
+            result = fn()
+            if result:
+                return result
+        except Exception as e:
+            last_err = e
+        time.sleep(interval)
+    raise AssertionError(
+        f"{name or 'condition'} did not become true within {timeout_seconds}s"
+        + (f": {last_err}" if last_err else "")
+    )
+
+
+def get_prometheus_targets() -> dict[str, str]:
+    """
+    Return {job_name: up_value} for all Prometheus scrape targets.
+    up_value is "1" (up) or "0" (down).
+    """
+    results = scrape('up')
+    return {r["metric"]["job"]: r["value"][1] for r in results}
+
+
+def get_metric_series_count(metric_name: str) -> int:
+    """Return number of time series for a metric."""
+    results = scrape(metric_name)
+    return len(results)
+
+
+def get_circuit_breaker_state(component: str = "nats") -> int:
+    """Return current circuit breaker state: 0=CLOSED, 1=HALF-OPEN, 2=OPEN."""
+    results = scrape(f'ba_ingestd_circuit_breaker_state{{component="{component}"}}')
+    if not results:
+        return -1  # not initialized
+    return int(float(results[0]["value"][1]))

scripts/m10_smoke.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""
+m10_smoke.py — M10 soak test (5k/s, 10 min, zero DLQ, runaway-source isolation)
+
+Usage:
+  # Local docker-compose:
+  docker compose --profile loadgen-m10 up -d
+  python3 scripts/m10_smoke.py
+
+  # Remote parres:
+  ssh root@192.168.44.94 "cd /root/broad-announce && docker compose --profile loadgen-m10 up -d && python3 scripts/m10_smoke.py"
+
+Exit code 0 = all green. Exit code 1 = assertion failed.
+
+Step 1 — pre-flight
+Step 2 — 5k/s soak (10 min, ramp 30s)
+Step 3 — runaway-source fault injection (60s)
+Step 4 — DLQ invariant
+Step 5 — teardown (profile down)
+"""
+import subprocess
+import sys
+import time
+import urllib.request
+import urllib.parse
+import json
+import base64
+
+sys.path.insert(0, __file__.rsplit("/", 1)[0])
+import m10_lib as lib
+
+PROM = "http://localhost:9090"
+GRAFANA = "http://localhost:3001"
+SOAK_DURATION_MIN = 10      # minutes
+SOAK_RAMP_SEC = 30           # ramp-up seconds
+RUNAWAY_DURATION_SEC = 60    # runaway fault injection duration
+CLUSTER_TARGET = 5000       # alerts/sec cluster-wide target
+P99_THRESHOLD = 5.0         # seconds — p99 must be under this
+DLQ_EXPECTED = 0            # zero DLQ is the invariant
+RATE_TOLERANCE = 0.05       # ±5%
+
+
+def curl_json(url: str) -> dict | None:
+    try:
+        with urllib.request.urlopen(url, timeout=10) as r:
+            return json.loads(r.read())
+    except Exception:
+        return None
+
+
+def pass_(msg: str):
+    print(f"  ✅ {msg}")
+
+
+def warn_(msg: str):
+    print(f"  ⚠️  {msg}")
+
+
+def fail_(msg: str):
+    print(f"  ❌ {msg}")
+    sys.exit(1)
+
+
+def step1_preflight() -> dict:
+    """Verify all services are up and DLQ baseline is clean."""
+    print("Step 1 — pre-flight")
+    targets = lib.get_prometheus_targets()
+    required = ["ingestd", "routerd", "deliverd-fcm", "deliverd-telegram",
+                "admind", "archiverd", "prometheus",
+                "loadgen-http-1", "loadgen-http-2", "loadgen-http-3"]
+    all_up = True
+    for svc in required:
+        status = targets.get(svc, "0")
+        if status == "1":
+            pass_(f"{svc} is up")
+        else:
+            warn_(f"{svc} is {'not scraped' if status == '0' else status}")
+            all_up = False
+
+    # Check DLQ baseline
+    dlq_now = lib.assert_dlq_count_equals(0, window_seconds=60)
+    pass_(f"DLQ baseline clean: {dlq_now} rows")
+
+    # Check circuit breaker is closed
+    cb_state = lib.get_circuit_breaker_state("nats")
+    cb_names = {0: "CLOSED", 1: "HALF-OPEN", 2: "OPEN"}
+    if cb_state == 0:
+        pass_(f"circuit breaker CLOSED (nats)")
+    elif cb_state < 0:
+        warn_(f"circuit breaker gauge not initialized yet (expected on cold start)")
+    else:
+        warn_(f"circuit breaker is {cb_names.get(cb_state, cb_state)} — may recover under load")
+
+    if not all_up:
+        fail_("not all services are up — fix before running M10 smoke")
+    return targets
+
+
+def step2_start_loadgen() -> subprocess.Popen:
+    """Start the 3-instance loadgen cluster."""
+    print("\nStep 2 — starting 3-instance loadgen cluster (5k/s)")
+    proc = subprocess.Popen(
+        ["docker", "compose", "--profile", "loadgen-m10", "up", "-d"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    code = proc.wait()
+    if code != 0:
+        fail_(f"docker compose --profile loadgen-m10 up -d failed (exit {code})")
+    pass_("3 loadgen-http instances started")
+    # Wait for ramp-up
+    print(f"  waiting {SOAK_RAMP_SEC}s for ramp-up to complete...", flush=True)
+    time.sleep(SOAK_RAMP_SEC)
+    pass_(f"ramp-up complete — now targeting {CLUSTER_TARGET}/s")
+    return proc
+
+
+def step2_monitor_soak() -> dict:
+    """
+    Monitor the soak: sample p99 + rate + DLQ every 30s.
+    Fails fast on any breach.
+    Returns dict of samples for the log.
+    """
+    print(f"\n  Monitoring soak for {SOAK_DURATION_MIN} minutes...")
+    samples = []
+    start = time.time()
+    deadline = start + SOAK_DURATION_MIN * 60
+
+    sample_interval = 30  # seconds between samples
+    while time.time() < deadline:
+        time.sleep(sample_interval)
+        elapsed = int(time.time() - start) // 60
+
+        try:
+            rate = lib.assert_rate_near(CLUSTER_TARGET, tolerance=RATE_TOLERANCE, window_seconds=30)
+            p99 = lib.assert_ingest_p99_under(P99_THRESHOLD, window_seconds=60)
+            dlq = lib.assert_dlq_count_equals(DLQ_EXPECTED, window_seconds=60)
+            print(f"  [{elapsed}m] rate={rate:.0f}/s p99={p99:.3f}s dlq={dlq}")
+            samples.append({"elapsed_min": elapsed, "rate": rate, "p99": p99, "dlq": dlq})
+        except AssertionError as e:
+            fail_(f"soak breach at {elapsed}m: {e}")
+
+    return samples
+
+
+def step3_runaway_test() -> None:
+    """
+    Runaway-source fault injection.
+    - Start a 4th loadgen instance firing at 10× per-source cap for the same company.
+    - Verify p99 for the OTHER sources (acme-002, acme-003) stays under threshold.
+    - The runaway (acme-001) can be anything.
+    """
+    print(f"\nStep 3 — runaway-source fault injection ({RUNAWAY_DURATION_SEC}s)")
+    print("  (not yet implemented — requires --rate override on loadgen-http-1)")
+    warn_(f"runaway-source test skipped (W4 enhancement pending)")
+    # TODO: spin up a 4th instance at 10× cap targeting acme-001
+    # Expected: per_source_p99("acme-002") < 5s and per_source_p99("acme-003") < 5s
+
+
+def step4_dlq_invariant(samples: list[dict]) -> int:
+    """Assert zero DLQ rows for the entire soak window."""
+    print("\nStep 4 — DLQ invariant check")
+    # The soak samples already checked DLQ delta, but do a final absolute check.
+    # Query the full soak window.
+    soak_seconds = SOAK_DURATION_MIN * 60
+    dlq = lib.assert_dlq_count_equals(DLQ_EXPECTED, window_seconds=soak_seconds)
+    pass_(f"DLQ count over {SOAK_DURATION_MIN}min soak: {dlq} (expected 0)")
+    return dlq
+
+
+def step5_teardown(proc) -> None:
+    """Bring down the loadgen cluster."""
+    print("\nStep 5 — teardown")
+    r = subprocess.run(
+        ["docker", "compose", "--profile", "loadgen-m10", "down", "-v"],
+        capture_output=True,
+    )
+    if r.returncode == 0:
+        pass_("loadgen cluster torn down")
+    else:
+        warn_(f"teardown returned {r.returncode}: {r.stderr.decode().strip()}")
+
+
+def print_summary(samples: list[dict], dlq_final: int) -> None:
+    """Print a summary table of the soak run."""
+    print("\n=== M10 Soak Summary ===")
+    print(f"Duration:  {SOAK_DURATION_MIN} min")
+    print(f"Target:    {CLUSTER_TARGET}/s (±{RATE_TOLERANCE*100:.0f}%)")
+    print(f"p99 threshold: {P99_THRESHOLD}s")
+    print()
+    if samples:
+        print(f"{'Time':>6}  {'Rate/s':>8}  {'p99(s)':>7}  {'DLQ':>4}")
+        print("-" * 35)
+        for s in samples:
+            print(f"{s['elapsed_min']:>5}m  {s['rate']:>8.0f}  {s['p99']:>7.3f}  {s['dlq']:>4}")
+    print()
+    print(f"Final DLQ count: {dlq_final} (expected 0)")
+    print()
+    print("🎉 M10 smoke: all checks complete.")
+
+
+def main():
+    print(f"M10 Soak Test — target {CLUSTER_TARGET}/s for {SOAK_DURATION_MIN} min")
+    print(f"p99 threshold: {P99_THRESHOLD}s | DLQ expected: {DLQ_EXPECTED}")
+    print()
+
+    try:
+        step1_preflight()
+        proc = step2_start_loadgen()
+        samples = step2_monitor_soak()
+        step3_runaway_test()
+        dlq_final = step4_dlq_invariant(samples)
+        step5_teardown(proc)
+        print_summary(samples, dlq_final)
+    except AssertionError as e:
+        print(f"\n💥 M10 smoke FAILED: {e}")
+        sys.exit(1)
+    except KeyboardInterrupt:
+        print("\nInterrupted.")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()