F2: publish-success counter + smoke assertion + NATS resource alerts

M11 NATS investigation medium-term (prevent recurrence):

  1. New counter: ba_ingestd_nats_publish_total{result=ok|error}
     - IngestdMetrics.NATSPublishTotal in internal/observability
     - Incremented in pipeline.go at 3 sites:
         * PublishAsync submission error (CB path)
         * PublishAsync submission error (no-CB path)
         * observeAsyncAck: ok on broker ack
         * observeAsyncAck: error on broker rejection / timeout
     - This is the metric the M11 10-min soak was missing.
       Receive rate alone is not enough — gRPC receive and NATS
       publish are decoupled, so a broken publish path can hide
       behind a healthy receive metric. See M11_NATS_INVESTIGATION.md.

  2. Smoke assertion: scripts/m11_lib.py::assert_nats_publish_rate_near
     - Now queried in the soak monitor loop (step3)
     - Receives a 'publish_ok' column in the per-minute log
     - Fails fast if NATS publish OK rate drops below target * (1-tol)
     - Catches the next class of this bug at smoke time

  3. PromQL alert rules: deploy/prometheus/rules/nats.yml
     - NatsJetStreamStorageHigh: >80% of max_storage for 5m (warning)
     - NatsJetStreamStorageCritical: >95% for 1m (critical)
     - IngestdNatsPublishErrorsHigh: >5% publish errors for 2m
     - IngestdReceivePublishMismatch: receive rate > publish OK + 100/s
     Mounted into the prometheus container at /etc/prometheus/rules.
     prometheus.yml gains a rule_files directive.

  4. docker-compose: mount the rules dir read-only into prometheus.

This is the F2 layer from M11_NATS_INVESTIGATION.md. After this, the
F1 fix becomes self-defending: any future regression on the publish
path will be caught at smoke time AND alert time, not silently.

deploy/prometheus/prometheus.yml

@@ -2,6 +2,11 @@ global:
   scrape_interval: 5s
   evaluation_interval: 10s
 
+# F2: alerting rules for NATS resource limits + publish-path health.
+# See deploy/prometheus/rules/nats.yml and M11_NATS_INVESTIGATION.md.
+rule_files:
+  - /etc/prometheus/rules/*.yml
+
 scrape_configs:
   # ── M0–M9 ingest tier ────────────────────────────────────
   - job_name: ingestd

deploy/prometheus/rules/nats.yml

@@ -0,0 +1,86 @@
+groups:
+  - name: nats_resource_limits
+    interval: 30s
+    rules:
+      # F2 (M11 NATS investigation): alert when NATS JetStream storage
+      # gets close to the server-level max_storage cap. The M11 2026-06-16
+      # finding was that storage silently filled to the cap (5.46 GiB
+      # default) and the broker started rejecting publishes, but the
+      # receive metric kept showing green. This alert would have fired
+      # well before the system went red.
+      #
+      # Threshold: 80% of max_storage for 5 min. F1 raised max_storage
+      # to 10 GiB (deploy/nats/nats.conf), so 80% = 8 GiB used. Real
+      # usage after F1 should be ~1.1 GiB (ALERTS 1 GiB + DELIVERIES
+      # 100 MiB + DLQ 10 MiB).
+      - alert: NatsJetStreamStorageHigh
+        expr: |
+          (jetstream_stats_storage / on() jetstream_config_max_storage) > 0.80
+        for: 5m
+        labels:
+          severity: warning
+        annotations:
+          summary: "NATS JetStream storage > 80% of max_storage"
+          description: |
+            NATS JetStream storage is at {{ $value | humanizePercentage }} of
+            max_storage. With stream-level MaxBytes caps from F1, the
+            streams self-trim, so anything above 80% indicates either
+            unexpected growth or a misconfigured stream. Check the
+            ALERTS / DELIVERIES / DLQ stream state via the NATS
+            monitoring endpoint (http://nats:8222/jsz?streams=true).
+
+      - alert: NatsJetStreamStorageCritical
+        expr: |
+          (jetstream_stats_storage / on() jetstream_config_max_storage) > 0.95
+        for: 1m
+        labels:
+          severity: critical
+        annotations:
+          summary: "NATS JetStream storage > 95% of max_storage"
+          description: |
+            NATS is about to enter the 'resource limits exceeded' state and
+            start rejecting publishes. The M11 NATS investigation documents
+            this failure mode in detail (M11_NATS_INVESTIGATION.md).
+
+  - name: nats_publish_path
+    interval: 30s
+    rules:
+      # F2: alert when ingestd's NATS publish success rate is materially
+      # below the receive rate. This catches the "system looks healthy
+      # but publishes are silently failing" class of bug that the M11
+      # 10-min soak missed.
+      - alert: IngestdNatsPublishErrorsHigh
+        expr: |
+          (
+            sum(rate(ba_ingestd_nats_publish_total{result="error"}[5m]))
+            /
+            sum(rate(ba_ingestd_nats_publish_total[5m]))
+          ) > 0.05
+        for: 2m
+        labels:
+          severity: warning
+        annotations:
+          summary: "ingestd NATS publish error rate > 5%"
+          description: |
+            More than 5% of ingestd NATS publish attempts are failing.
+            Check the ingestd logs and the NATS server logs. The M11
+            NATS investigation is the playbook for diagnosing this.
+
+      - alert: IngestdReceivePublishMismatch
+        expr: |
+          (
+            sum(rate(ba_ingestd_alerts_received_total{transport="grpc",result="accepted"}[5m]))
+            -
+            sum(rate(ba_ingestd_nats_publish_total{result="ok"}[5m]))
+          ) > 100
+        for: 2m
+        labels:
+          severity: warning
+        annotations:
+          summary: "ingestd receive rate > publish OK rate"
+          description: |
+            The gRPC receive path is accepting alerts at a rate more than
+            100/s higher than the NATS publish path is acknowledging. This
+            indicates the publish path is broken even if individual publish
+            failures are within tolerance. The M11 10-min soak was a false
+            positive because it only checked the receive rate.

docker-compose.yml

@@ -297,6 +297,8 @@ services:
       - --config.file=/etc/prometheus/prometheus.yml
     volumes:
       - ./deploy/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
+      # F2: alert rules (NATS resource limits, ingestd publish-path health).
+      - ./deploy/prometheus/rules:/etc/prometheus/rules:ro
     ports: ["9090:9090"]
     depends_on: [ingestd, routerd, deliverd-fcm, deliverd-telegram, telegramd, admind]
 

internal/observability/metrics.go

@@ -66,6 +66,17 @@ type IngestdMetrics struct {
 	GRPCInflight   *prometheus.HistogramVec // ba_ingestd_grpc_inflight_per_stream{source_id}
 	GRPCRateLimited *prometheus.CounterVec  // ba_ingestd_grpc_rate_limited_total{source_id}
 	GRPCAckLatency *prometheus.HistogramVec // ba_ingestd_grpc_ack_latency_seconds{source_id}
+
+	// --- F2: NATS publish outcome (M11 NATS investigation) ---
+	// Receivers of ba_ingestd_alerts_received_total cannot tell whether
+	// a received alert was successfully published to NATS. The receive
+	// path (gRPC server) and publish path (pipeline) are decoupled, and
+	// the M11 2026-06-16 finding showed a broken publish path that
+	// looked healthy on the receive metric. This counter tags every
+	// publish attempt as ok / error so the smoke can assert that the
+	// publish path is healthy too.
+	// See M11_NATS_INVESTIGATION.md, "Medium-term (prevent recurrence)".
+	NATSPublishTotal *prometheus.CounterVec // ba_ingestd_nats_publish_total{result}
 }
 
 // NewIngestdMetrics registers and returns the ingestd metrics.
@@ -115,6 +126,13 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 			Buckets:      prometheus.DefBuckets,
 			ConstLabels:  prometheus.Labels{"service": serviceName},
 		}, []string{"source_id"}),
+		NATSPublishTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "ingestd",
+			Name:      "nats_publish_total",
+			Help:      "NATS JetStream publish attempts by result (ok/error). F2: catches publish-path failures that the receive metric misses.",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"result"}),
 		MQTTMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Namespace: "ba",
 			Subsystem: "ingestd",

internal/pipeline/pipeline.go

@@ -254,6 +254,10 @@ func (d *Deps) Process(ctx context.Context, body []byte, sig string) Result {
 		publishErr = d.CircuitBreaker.Do(ctx, func() error {
 			f, err := d.JetStream.PublishAsync(subject, payload)
 			if err != nil {
+				// F2: submission-level failure (queue full, JS stopped, etc.)
+				if d.Metrics != nil {
+					d.Metrics.NATSPublishTotal.WithLabelValues("error").Inc()
+				}
 				return err
 			}
 			fut = f
@@ -265,6 +269,10 @@ func (d *Deps) Process(ctx context.Context, body []byte, sig string) Result {
 	} else {
 		fut, err := d.JetStream.PublishAsync(subject, payload)
 		if err != nil {
+			// F2: submission-level failure (no CB path).
+			if d.Metrics != nil {
+				d.Metrics.NATSPublishTotal.WithLabelValues("error").Inc()
+			}
 			publishErr = err
 		} else if fut != nil {
 			go observeAsyncAck(fut, d, a.SourceID, subject, start)
@@ -358,10 +366,18 @@ func observeAsyncAck(fut nats.PubAckFuture, d *Deps, sourceID, subject string, s
 	}
 	select {
 	case <-fut.Ok():
+		// F2: broker accepted and persisted the message.
 		if d != nil && d.Metrics != nil {
 			d.Metrics.PublishLatency.WithLabelValues(sourceID).Observe(time.Since(sentAt).Seconds())
+			d.Metrics.NATSPublishTotal.WithLabelValues("ok").Inc()
 		}
 	case err := <-fut.Err():
+		// F2: broker rejected / timed out. This is the failure mode that
+		// the M11 NATS investigation missed (system looked healthy on
+		// the receive metric while publishes were silently failing).
+		if d != nil && d.Metrics != nil {
+			d.Metrics.NATSPublishTotal.WithLabelValues("error").Inc()
+		}
 		if d != nil && d.Logger != nil {
 			d.Logger.Warn("async publish failed", "subject", subject, "source_id", sourceID, "err", err)
 		}

scripts/m11_lib.py

@@ -88,6 +88,46 @@ def assert_grpc_p99_under(threshold_ms: float, window_seconds: int = 60) -> floa
     return p99_ms
 
 
+def assert_nats_publish_rate_near(target: float, tolerance: float = 0.10,
+                                  window_seconds: int = 30) -> float:
+    """
+    Assert the NATS publish OK rate is within tolerance of target.
+
+    F2 (M11 NATS investigation): receive rate alone is not enough — a broken
+    publish path can hide behind a healthy receive metric. The M11 10-min
+    soak that "shipped" M11 was a false positive: ba_ingestd_alerts_received_total
+    counts gRPC receive, but PublishAsync to NATS was failing because the
+    server hit its max_storage cap. The smoke stayed green while the publish
+    path was broken end-to-end.
+
+    This assertion catches the next class of this bug by querying
+    ba_ingestd_nats_publish_total{result="ok"} and checking it tracks the
+    receive rate.
+
+    Query: sum(rate(ba_ingestd_nats_publish_total{result="ok"}[window]))
+    """
+    query = (
+        f'sum(rate(ba_ingestd_nats_publish_total{{result="ok"}}[{window_seconds}s]))'
+    )
+    results = scrape(query)
+    if not results:
+        raise AssertionError(
+            f"NATS publish OK rate query returned no data. "
+            f"Is the F2 metric (ba_ingestd_nats_publish_total) exposed by ingestd? "
+            f"(query: {query})"
+        )
+    rate = float(results[0]["value"][1])
+    min_rate = target * (1 - tolerance)
+    if rate < min_rate:
+        raise AssertionError(
+            f"NATS publish OK rate {rate:.0f}/s is below target {target:.0f}/s "
+            f"(tolerance ±{tolerance*100:.0f}%, min allowed: {min_rate:.0f}/s, "
+            f"window={window_seconds}s). This means the publish path is broken "
+            f"even though gRPC receive may be green. See M11_NATS_INVESTIGATION.md."
+        )
+    return rate
+
+
 def assert_dlq_count_equals(expected: int, window_seconds: int) -> int:
     """
     Assert the DLQ row count (delta over window) equals expected.
@@ -107,6 +147,25 @@ def assert_dlq_count_equals(expected: int, window_seconds: int) -> int:
     return int(count)
 
 
+
+    """
+    Assert the DLQ row count (delta over window) equals expected.
+    Returns the actual count.
+    """
+    query = (
+        f'increase(ba_deliverd_dlq_total[{window_seconds}s])'
+    )
+    results = scrape(query)
+    if not results:
+        return 0  # No data → assume 0
+    count = float(results[0]["value"][1])
+    if abs(count - expected) > 0.5:
+        raise AssertionError(
+            f"DLQ count {count:.0f} over last {window_seconds}s does not equal expected {expected}"
+        )
+    return int(count)
+
+
 def get_prometheus_targets() -> dict[str, str]:
     """
     Return a dict mapping service name → health status from Prometheus target health API.

scripts/m11_smoke.py

@@ -156,14 +156,21 @@ def step3_monitor_soak() -> list[dict]:
         try:
             rate = lib.assert_grpc_rate_near(
                 CLUSTER_TARGET, tolerance=RATE_TOLERANCE, window_seconds=30)
+            # F2: assert the NATS publish path is healthy too. Without
+            # this, a broken publish path can hide behind a green receive
+            # metric (the M11 10-min soak was a false positive for exactly
+            # this reason). See M11_NATS_INVESTIGATION.md.
+            publish_ok = lib.assert_nats_publish_rate_near(
+                CLUSTER_TARGET, tolerance=RATE_TOLERANCE, window_seconds=30)
             p99_ms = lib.assert_grpc_p99_under(P99_THRESHOLD_MS, window_seconds=60)
             dlq = lib.assert_dlq_count_equals(DLQ_EXPECTED, window_seconds=60)
             streams = lib.get_grpc_streams_active()
-            print(f"  [{elapsed_min}m] rate={rate:.0f}/s p99={p99_ms:.1f}ms "
-                  f"dlq={dlq} streams={streams}")
+            print(f"  [{elapsed_min}m] rate={rate:.0f}/s publish_ok={publish_ok:.0f}/s "
+                  f"p99={p99_ms:.1f}ms dlq={dlq} streams={streams}")
             samples.append({
                 "elapsed_min": elapsed_min,
                 "rate": rate,
+                "publish_ok": publish_ok,
                 "p99_ms": p99_ms,
                 "dlq": dlq,
                 "streams": streams,