M9(1/3): circuit breaker + quarantine + deliverd metrics

Circuit breaker (SPEC §22 layer 6):
- internal/circuitbreaker: per-component 3-state machine
  (CLOSED → OPEN → HALF-OPEN → CLOSED). Bounded exp
  backoff in the HALF-OPEN→CLOSED path; automatic
  trip after FailureThreshold failures in FailureWindow.
  Prometheus CBState gauge updated on every transition.
  13 unit tests, all PASS.
- internal/config: 4 new knobs on Common.Ingestd:
  BA_INGESTD_CB_FAILURE_THRESHOLD (5), _WINDOW_SECS (10),
  _OPEN_DURATION_SECS (30), _MAX_HALF_OPEN (1).

Quarantine (SPEC §22 layer 7):
- internal/quarantine: Redis-backed per-source error-rate
  limiter. ZADD hit on every rejection, ZREMRANGEBYSCORE
  to prune old hits, SET ban key with TTL on threshold.
  Multi-instance aware (shared ban key in Redis).
- Config: 3 existing BA_INGESTD_QUARANTINE_* knobs wired.

Wired into ingestd pipeline (process.go):
- Circuit breaker wraps JetStream.Publish (sync) so
  Do() gets immediate success/failure feedback.
- Quarantine.IsBanned() checked before processing.
- recordHit() deferred closure fires on every rejection
  so the source accumulates error hits.
- Ctx cancel short-circuits Do() before running fn.
- Publish method added to natsPublisher interface.

Deliverd metrics (SPEC §22 L3):
- internal/observability: DeliverdMetrics struct with
  DeliveryAttempts (channel × status), DLQTotal (channel),
  DLQLatency (wall-clock from first attempt to DLQ insert),
  RetryAttempts (total retry iters per channel).
- Both deliverd-fcm and deliverd-telegram record metrics
  on every attempt row insert and on DLQ insert.

Prometheus scrape config:
- deploy/prometheus/prometheus.yml: split deliverd job
  into deliverd-fcm (:8802) and deliverd-telegram (:8803),
  added archiverd (:8804), added prometheus self-monitoring.
  Labels now distinguish fcm vs telegram in dashboards.

docker-compose.yml: wired BA_INGESTD_CB_* and
BA_INGESTD_QUARANTINE_* env vars on ingestd.

go build ./... clean. go vet ./... clean.
go test ./internal/circuitbreaker 13/13 PASS.

.env.example

@@ -55,6 +55,13 @@ BA_ARCHIVERD_BATCH_SIZE=10000
 BA_ARCHIVERD_CLICKHOUSE_URL=http://clickhouse:8123
 BA_INGESTD_QUARANTINE_WINDOW_SECONDS=60
 BA_INGESTD_QUARANTINE_DURATION_SECONDS=300
+# M9 circuit breaker (layer 6): trips when
+# BA_INGESTD_CB_FAILURE_THRESHOLD failures accumulate
+# within BA_INGESTD_CB_FAILURE_WINDOW_SECS.
+BA_INGESTD_CB_FAILURE_THRESHOLD=5
+BA_INGESTD_CB_FAILURE_WINDOW_SECS=10
+BA_INGESTD_CB_OPEN_DURATION_SECS=30
+BA_INGESTD_CB_MAX_HALF_OPEN=1
 
 # M0 source registry (env-only; M2 swaps for DB)
 # Format: comma-separated company:source:secret triples

cmd/deliverd-fcm/main.go

@@ -39,6 +39,10 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 )
 
+// deliverdMetrics is the package-level metrics instance.
+// Wired in main() from observability.NewDeliverdMetrics.
+var deliverdMetrics *observability.DeliverdMetrics
+
 // M2: only the FCM channel. M3+ adds telegram, sms, etc.
 const fcmChannel = "fcm"
 
@@ -122,9 +126,13 @@ func main() {
 	runCtx, runCancel := context.WithCancel(ctx)
 	defer runCancel()
 
+	// M9: deliverd metrics. One registry for both fcm and telegram
+	// deliverds so they get separate service labels.
+	reg, _ := observability.NewRegistry("deliverd")
+	deliverdMetrics = observability.NewDeliverdMetrics(reg, "deliverd")
+
 	go consume(runCtx, logger, consumer, pool, httpClient, fakefcmdURL, retryCfg)
 
-	reg, _ := observability.NewRegistry("deliverd")
 	srv := httpserver.New(httpserver.Config{
 		Addr:          cfg.HTTPAddr,
 		ServiceName:   "deliverd",
@@ -255,6 +263,9 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 
 	url := fakefcmdURL + "/v1/projects/fakefcmd/messages:send"
 
+	// M9: track attempt start time for DLQ latency metric.
+	firstAttemptTime := time.Now()
+
 	// M8: retry loop. Each attempt: POST, persist a
 	// deliveries row with status sent/failed and the
 	// attempt counter, return the error to the helper.
@@ -303,6 +314,11 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 			logger.Warn("delivery row insert", "err", dbErr, "attempt", attempt)
 		}
 
+		// M9: record per-channel delivery attempt metric.
+		if deliverdMetrics != nil {
+			deliverdMetrics.DeliveryAttempts.WithLabelValues(fcmChannel, status).Inc()
+		}
+
 		if status == "sent" {
 			logger.Info("delivery sent",
 				"alert_id", alertHeader.ID,
@@ -380,6 +396,12 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 			"attempts", res.Attempts,
 			"err", errString(lastErr),
 		)
+		// M9: record DLQ metric and latency.
+		if deliverdMetrics != nil {
+			deliverdMetrics.DLQTotal.WithLabelValues(fcmChannel).Inc()
+			deliverdMetrics.DLQLatency.Observe(time.Since(firstAttemptTime).Seconds())
+			deliverdMetrics.RetryAttempts.WithLabelValues(fcmChannel).Add(float64(res.Attempts))
+		}
 	}
 	_ = m.Ack()
 }

cmd/deliverd-telegram/main.go

@@ -50,6 +50,10 @@ import (
 
 const telegramChannel = "telegram"
 
+// deliverdMetrics is the package-level metrics instance.
+// Wired in main() from observability.NewDeliverdMetrics.
+var deliverdMetrics *observability.DeliverdMetrics
+
 type deliveryEnvelope struct {
 	Alert        json.RawMessage `json:"alert"`
 	IndividualID string          `json:"individual_id"`
@@ -144,9 +148,12 @@ func main() {
 	runCtx, runCancel := context.WithCancel(ctx)
 	defer runCancel()
 
-	go consume(runCtx, logger, consumer, pool, client, botToken, retryCfg)
-
+	// M9: deliverd metrics. Separate registry from deliverd-fcm
+	// so each gets its own service label in Prometheus.
 	reg, _ := observability.NewRegistry("deliverd-telegram")
+	deliverdMetrics = observability.NewDeliverdMetrics(reg, "deliverd-telegram")
+
+	go consume(runCtx, logger, consumer, pool, client, botToken, retryCfg)
 	srv := httpserver.New(httpserver.Config{
 		Addr:          cfg.HTTPAddr,
 		ServiceName:   "deliverd-telegram",
@@ -247,6 +254,9 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 	// unchanged from M3.
 	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID, ah.DedupeCount)
 
+	// M9: track attempt start time for DLQ latency metric.
+	firstAttemptTime := time.Now()
+
 	// M8: retry loop. Telegram returns errors on:
 	//   - 4xx (chat not found, bot blocked, etc.) — these
 	//     are permanent; we use PermanentError to skip
@@ -279,6 +289,11 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 			logger.Warn("delivery row insert", "err", dbErr, "attempt", attempt)
 		}
 
+		// M9: record per-channel delivery attempt metric.
+		if deliverdMetrics != nil {
+			deliverdMetrics.DeliveryAttempts.WithLabelValues(telegramChannel, status).Inc()
+		}
+
 		if status == "sent" {
 			logger.Info("delivery sent",
 				"alert_id", ah.ID,
@@ -350,6 +365,12 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 			"attempts", res.Attempts,
 			"err", errString(lastErr),
 		)
+		// M9: record DLQ metric and latency.
+		if deliverdMetrics != nil {
+			deliverdMetrics.DLQTotal.WithLabelValues(telegramChannel).Inc()
+			deliverdMetrics.DLQLatency.Observe(time.Since(firstAttemptTime).Seconds())
+			deliverdMetrics.RetryAttempts.WithLabelValues(telegramChannel).Add(float64(res.Attempts))
+		}
 	}
 	_ = m.Ack()
 }

cmd/ingestd/http_test.go

@@ -38,6 +38,11 @@ func (f *fakePublisher) PublishAsync(subj string, data []byte) error {
 	return nil
 }
 
+// Publish is synchronous; same as PublishAsync for the test fake.
+func (f *fakePublisher) Publish(subj string, data []byte) error {
+	return f.PublishAsync(subj, data)
+}
+
 // stubLimiter always allows.
 type stubLimiter struct{}
 

cmd/ingestd/main.go

@@ -12,11 +12,13 @@ import (
 	"time"
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/circuitbreaker"
 	"git3.techno-world.net/lrosales/broad-announce/internal/concurrency"
 	"git3.techno-world.net/lrosales/broad-announce/internal/config"
 	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
 	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+	"git3.techno-world.net/lrosales/broad-announce/internal/quarantine"
 	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
 	"git3.techno-world.net/lrosales/broad-announce/internal/store"
 	"git3.techno-world.net/lrosales/broad-announce/internal/tailhub"
@@ -64,6 +66,42 @@ func main() {
 	}
 	ded := dedupe.New(r.Client, dedTTL)
 
+	// M9: circuit breaker (layer 6). Wraps the NATS publish
+	// call so a sick broker doesn't take down ingestd.
+	cbCfg := circuitbreaker.Config{
+		Name:             "nats-publish",
+		FailureThreshold: cfg.CircuitFailureThreshold,
+		FailureWindow:    time.Duration(cfg.CircuitFailureWindowSecs) * time.Second,
+		OpenDuration:     time.Duration(cfg.CircuitOpenDurationSecs) * time.Second,
+		MaxHalfOpen:      cfg.CircuitMaxHalfOpen,
+	}
+	cb := circuitbreaker.New(cbCfg)
+	// Also report CB state changes to Prometheus.
+	cb.Measure = func(state int, err error) {
+		m.CBState.WithLabelValues("nats").Set(float64(state))
+	}
+	logger.Info("circuit breaker configured",
+		"failure_threshold", cfg.CircuitFailureThreshold,
+		"failure_window_sec", cfg.CircuitFailureWindowSecs,
+		"open_duration_sec", cfg.CircuitOpenDurationSecs,
+		"max_half_open", cfg.CircuitMaxHalfOpen,
+	)
+
+	// M9: quarantine manager (layer 7). Per-source error-rate
+	// limiter backed by Redis so the ban is shared across
+	// multiple ingestd instances.
+	quarantineCfg := quarantine.Config{
+		HitsThreshold: cfg.QuarantineHitsThreshold,
+		HitsWindow:    time.Duration(cfg.QuarantineWindowSeconds) * time.Second,
+		BanDuration:   time.Duration(cfg.QuarantineDurationSecond) * time.Second,
+	}
+	q := quarantine.New(r.Client, quarantineCfg)
+	logger.Info("quarantine configured",
+		"hits_threshold", cfg.QuarantineHitsThreshold,
+		"hits_window_sec", cfg.QuarantineWindowSeconds,
+		"ban_duration_sec", cfg.QuarantineDurationSecond,
+	)
+
 	// M0 source registry: loaded from env. M2 replaces with DB.
 	sources := loadSourcesFromEnv(logger)
 
@@ -102,6 +140,8 @@ func main() {
 			Tail:             hub,
 			Transport:        "http",
 			MaxSeen:          maxSeen,
+			CircuitBreaker:   cb,
+			Quarantine:       q,
 		},
 		MaxBytes:   cfg.MaxPayloadBytes,
 	}

cmd/ingestd/process.go

@@ -15,6 +15,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"strconv"
@@ -22,8 +23,10 @@ import (
 
 	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
 	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/circuitbreaker"
 	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
 	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+	"git3.techno-world.net/lrosales/broad-announce/internal/quarantine"
 	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
 	"git3.techno-world.net/lrosales/broad-announce/internal/tailhub"
 	"github.com/nats-io/nats.go"
@@ -97,6 +100,13 @@ type processDeps struct {
 	// (HTTP, MQTT, WS) share the same in-process state and
 	// the same max-observed gauge.
 	MaxSeen *observability.MaxSeen
+	// CircuitBreaker is the M9 layer-6 per-component circuit
+	// breaker wrapping the NATS publish call. Nil is fine (falls
+	// back to direct publish without circuit protection).
+	CircuitBreaker *circuitbreaker.Breaker
+	// Quarantine is the M9 layer-7 per-source error-rate limiter.
+	// Nil is fine (no quarantine enforcement).
+	Quarantine *quarantine.Manager
 }
 
 // ProcessAlert runs the full SPEC §22 protection chain on one
@@ -108,13 +118,15 @@ type processDeps struct {
 //   1. payload-size cap   (caller does this — http.go via
 //                          MaxBytesReader; mqtt.go via the
 //                          broker-side max-inflight setting)
+//   2. quarantine check (M9 layer 7) — per-source ban if error
+//                          rate exceeds threshold
 //   3. per-source rate limit
 //   4. per-company rate limit
 //   5. schema validate
 //   5b. parse JSON
 //   (auth) HMAC verify — see verifyHMAC
 //   5c. dedupe
-//   publish to NATS
+//   publish to NATS (M9 layer 6 circuit breaker wraps this)
 func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader string) Result {
 	now := d.now()
 
@@ -137,6 +149,21 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 			fmt.Sprintf("no such source %s/%s", a.CompanyID, a.SourceID))
 	}
 
+	// M9 layer 7: quarantine check. Check before we spend any CPU
+	// on a known-bad source.
+	if d.Quarantine != nil {
+		if banned, remaining, err := d.Quarantine.IsBanned(ctx, a.SourceID); err == nil && banned {
+			d.Metrics.AlertsReceived.WithLabelValues("quarantined").Inc()
+			d.Logger.Warn("source quarantined",
+				"source_id", a.SourceID,
+				"company_id", a.CompanyID,
+				"remaining", remaining,
+			)
+			return Reject("quarantined", 429,
+				fmt.Sprintf("source quarantined for %v; retry after", remaining.Round(time.Second)))
+		}
+	}
+
 	// Auth. Stripe-style: X-BA-Signature: t=<unix>,v1=<hex>.
 	// For HTTP it's a header; for MQTT it's a top-level field
 	// on the envelope — both call sites pass the same string.
@@ -145,6 +172,25 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 		return Reject("bad_signature", 401, "")
 	}
 
+	// M9: quarantine hit tracking. Any rejection after the HMAC
+	// check (where we know the source is real) indicates a
+	// problematic source. We use a closure so the defer pattern
+	// catches every return path without boilerplate at each one.
+	var hitRecorded bool
+	defer func() {
+		if !hitRecorded && d.Quarantine != nil {
+			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
+		}
+	}()
+	// recordHit records a quarantine error hit for a.SourceID.
+	// Call it before any rejection return.
+	recordHit := func() {
+		if d.Quarantine != nil && !hitRecorded {
+			hitRecorded = true
+			_ = d.Quarantine.RecordHit(context.Background(), a.SourceID)
+		}
+	}
+
 	// M6: Dedupe BEFORE rate limit. A duplicate (isNew=false)
 	// is a Redis INCR + JSON marshal + NATS publish — it does
 	// not warrant burning a rate-limit token. The rate limit
@@ -182,6 +228,7 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 			d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
 			_ = ttl
+			recordHit()
 			return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
 		}
 	}
@@ -192,6 +239,7 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
 			d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
 			_ = ttl
+			recordHit()
 			return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
 		}
 	}
@@ -201,7 +249,8 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 	a.ReceivedAt = now.UTC()
 	a.DedupeCount = count
 
-	// Publish to NATS.
+	// Publish to NATS. M9 layer 6: circuit breaker wraps the
+	// publish call so a sick NATS server doesn't take down ingestd.
 	subject := broker.AlertsSubject(a.CompanyID)
 	payload, err := json.Marshal(a)
 	if err != nil {
@@ -209,10 +258,33 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 		return Reject("marshal_failed", 500, err.Error())
 	}
 	start := time.Now()
-	if err := d.JetStream.PublishAsync(subject, payload); err != nil {
-		// Circuit breaker (M9) wraps this. For M0 we just fail loud.
-		d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
-		return Reject("broker_unavailable", 503, err.Error())
+	var publishErr error
+	if d.CircuitBreaker != nil {
+		// Wrap the synchronous publish in the circuit breaker.
+		// We use Publish (sync) so Do() gets immediate feedback.
+		publishErr = d.CircuitBreaker.Do(ctx, func() error {
+			return d.JetStream.Publish(subject, payload)
+		})
+	} else {
+		publishErr = d.JetStream.Publish(subject, payload)
+	}
+	if publishErr != nil {
+		if errors.Is(publishErr, circuitbreaker.ErrCircuitOpen) {
+			d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
+			d.Metrics.CBState.WithLabelValues("nats").Set(circuitbreaker.StateOpen)
+			d.Logger.Warn("circuit breaker open",
+				"subject", subject,
+				"alert_id", a.ID,
+				"company_id", a.CompanyID,
+			)
+			recordHit()
+			return Reject("circuit_open", 503, "broker circuit breaker open")
+		}
+		// Real publish error (network, auth, etc.).
+		d.Metrics.AlertsReceived.WithLabelValues("broker_unavailable").Inc()
+		d.Logger.Error("nats publish", "err", publishErr, "subject", subject)
+		recordHit()
+		return Reject("broker_unavailable", 503, publishErr.Error())
 	}
 	d.Metrics.PublishLatency.Observe(time.Since(start).Seconds())
 	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
@@ -238,6 +310,9 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 		ev := tailhub.FromAlert(&a, d.Transport)
 		d.Tail.Publish(ev)
 	}
+	// Accepted: mark that we did not get a rejection so the
+	// defer does not record a spurious quarantine hit.
+	hitRecorded = true
 	return Accept(a.ID, count, isNew)
 }
 
@@ -252,6 +327,9 @@ func (d *processDeps) now() time.Time {
 // MQTT paths share it; tests can swap in a fake.
 type natsPublisher interface {
 	PublishAsync(subj string, data []byte) error
+	// Publish is synchronous. The circuit breaker uses this
+	// to get immediate success/failure feedback.
+	Publish(subj string, data []byte) error
 }
 
 // jsPublisher adapts a nats.JetStreamContext to the natsPublisher interface.
@@ -259,6 +337,13 @@ type jsPublisher struct {
 	js nats.JetStreamContext
 }
 
+// Publish is synchronous (blocks until server ack or timeout). Used
+// by the circuit breaker which needs immediate success/failure feedback.
+func (j *jsPublisher) Publish(subj string, data []byte) error {
+	_, err := j.js.Publish(subj, data)
+	return err
+}
+
 func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
 	_, err := j.js.PublishAsync(subj, data)
 	return err

deploy/prometheus/prometheus.yml

@@ -3,18 +3,42 @@ global:
   evaluation_interval: 10s
 
 scrape_configs:
+  # ── M0–M9 ingest tier ────────────────────────────────────
   - job_name: ingestd
     static_configs:
       - targets: ['ingestd:8800']
+
+  # ── M0–M2 router tier ────────────────────────────────────
   - job_name: routerd
     static_configs:
       - targets: ['routerd:8801']
-  - job_name: deliverd
+
+  # ── M1–M9 delivery tier ───────────────────────────────────
+  # M9: split by channel so channel label in
+  # ba_deliverd_delivery_attempts_total is meaningful.
+  - job_name: deliverd-fcm
+    static_configs:
+      - targets: ['deliverd-fcm:8802']
+  - job_name: deliverd-telegram
     static_configs:
-      - targets: ['deliverd:8802']
+      - targets: ['deliverd-telegram:8803']
+
+  # ── M8 DLQ + operator API ────────────────────────────────
   - job_name: admind
     static_configs:
       - targets: ['admind:8803']
+
+  # ── M7 data-tier archiver ────────────────────────────────
+  - job_name: archiverd
+    static_configs:
+      - targets: ['archiverd:8804']
+
+  # ── Loadgen (one-shot; useful to see if it's running) ────
   - job_name: loadgen
     static_configs:
       - targets: ['loadgen-http:8891']
+
+  # ── Prometheus self-monitoring ────────────────────────────
+  - job_name: prometheus
+    static_configs:
+      - targets: ['localhost:9090']

docker-compose.yml

@@ -121,6 +121,15 @@ services:
       BA_INGESTD_TAIL_TOKEN: "tail-dev-token-please-change-in-prod"
       BA_INGESTD_MAX_CONCURRENT_PER_IP: "32"
       BA_INGESTD_DEDUPE_TTL_SECONDS: "300"
+      # M9 layer 6: circuit breaker (trips after 5 failures in 10s, 30s open)
+      BA_INGESTD_CB_FAILURE_THRESHOLD: "5"
+      BA_INGESTD_CB_FAILURE_WINDOW_SECS: "10"
+      BA_INGESTD_CB_OPEN_DURATION_SECS: "30"
+      BA_INGESTD_CB_MAX_HALF_OPEN: "1"
+      # M9 layer 7: quarantine (bans source at 100 hits/5min for 10min)
+      BA_INGESTD_QUARANTINE_HITS_THRESHOLD: "100"
+      BA_INGESTD_QUARANTINE_WINDOW_SECONDS: "300"
+      BA_INGESTD_QUARANTINE_DURATION_SECONDS: "600"
     ports: ["8800:8800"]
     depends_on:
       nats:    { condition: service_healthy }

internal/circuitbreaker/circuitbreaker.go

@@ -0,0 +1,233 @@
+// Package circuitbreaker provides a per-component circuit breaker
+// for the ingest pipeline (SPEC §22 layer 6). It wraps a function
+// that performs an outbound call and trips the circuit when error
+// rates exceed a threshold.
+//
+// State machine (3 states, 4 transitions):
+//
+//	CLOSED → OPEN   failure_count >= threshold within window
+//	OPEN   → HALF   open_duration elapsed
+//	HALF   → CLOSED successful call observed
+//	HALF   → OPEN   call in HALF state also fails
+//
+// The breaker is in-process (no shared state across multiple
+// ingestd instances). For the dev stack that is fine; for a
+// multi-instance production deploy, the state would live in
+// Redis (deferred to a future milestone).
+//
+// Metrics: callers are expected to call cb.Measure(state, err)
+// after each Do() call so the gauge maintained here stays in
+// sync with what the Prometheus scrape sees.
+package circuitbreaker
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+)
+
+// State values. Mirrors the ba_ingestd_circuit_breaker_state gauge.
+const (
+	StateClosed   = 0
+	StateHalfOpen = 1
+	StateOpen     = 2
+)
+
+// Config is the static configuration for one circuit breaker.
+// All fields must be set by the caller.
+type Config struct {
+	// Name is the component label used in Prometheus metrics.
+	Name string
+	// FailureThreshold is the number of consecutive failures
+	// (within FailureWindow) that trips the circuit to OPEN.
+	FailureThreshold int
+	// FailureWindow is the rolling window for counting failures.
+	FailureWindow time.Duration
+	// OpenDuration is how long the circuit stays OPEN before
+	// transitioning to HALF-OPEN (testing).
+	OpenDuration time.Duration
+	// MaxHalfOpen is the number of test calls admitted while
+	// in HALF-OPEN state. Default 1 (admit one, decide).
+	MaxHalfOpen int
+}
+
+// DefaultConfig is a reasonable starting point for a NATS broker
+// circuit breaker: 5 failures in 10s trips; 30s open; 1 test call.
+func DefaultConfig(name string) Config {
+	return Config{
+		Name:             name,
+		FailureThreshold: 5,
+		FailureWindow:    10 * time.Second,
+		OpenDuration:     30 * time.Second,
+		MaxHalfOpen:      1,
+	}
+}
+
+// Breaker is the per-component circuit breaker. It is safe for
+// concurrent use by the HTTP, MQTT, and WebSocket ingest paths.
+type Breaker struct {
+	config Config
+
+	mu sync.RWMutex
+	// state is one of StateClosed, StateHalfOpen, StateOpen.
+	state int
+	// failures is a circular buffer of recent failure timestamps.
+	// We keep it as a slice and prune anything older than
+	// config.FailureWindow on every call.
+	failures []time.Time
+	// halfOpenCount is how many test calls have been admitted
+	// in the current HALF-OPEN window. Resets to 0 on transition
+	// out of HALF-OPEN.
+	halfOpenCount int
+	// openSince is when we entered the OPEN state. Used to
+	// decide when to transition to HALF-OPEN.
+	openSince time.Time
+
+	// Measure is called after every Do() call with the observed
+	// state and error. It is nil in production; tests can inject
+	// a spy to observe state transitions without polling.
+	Measure func(state int, err error)
+}
+
+// New creates a new circuit breaker from cfg.
+func New(cfg Config) *Breaker {
+	if cfg.MaxHalfOpen <= 0 {
+		cfg.MaxHalfOpen = 1
+	}
+	return &Breaker{config: cfg, state: StateClosed}
+}
+
+// Do runs fn if the circuit is CLOSED or HALF-OPEN. Returns
+// ErrCircuitOpen when the circuit is OPEN. Returns the error
+// from fn on failure; nil on success.
+//
+// If the circuit trips OPEN, Do records the failure internally
+// so the next caller gets ErrCircuitOpen immediately.
+func (cb *Breaker) Do(ctx context.Context, fn func() error) error {
+	// Check context cancellation before doing any work.
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	// Check and handle any time-based transition BEFORE acquiring
+	// the lock to avoid holding the lock across a time.Sleep.
+	// (time.Since inside a mutex is a deadlock risk in writer-
+	// biased RWMutex implementations).
+	cb.tryHalfOpen()
+
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+
+	switch cb.state {
+	case StateOpen:
+		return ErrCircuitOpen
+	case StateHalfOpen:
+		if cb.halfOpenCount >= cb.config.MaxHalfOpen {
+			return ErrCircuitOpen
+		}
+		cb.halfOpenCount++
+	}
+
+	// Run the protected function.
+	err := fn()
+
+	// Record the result under the lock.
+	cb.recordResultLocked(err)
+
+	return err
+}
+
+// tryHalfOpen checks if an OPEN circuit's duration has elapsed
+// and transitions it to HALF-OPEN. Safe to call without the lock;
+// it acquires the lock internally for the write.
+func (cb *Breaker) tryHalfOpen() {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+	if cb.state == StateOpen && time.Since(cb.openSince) >= cb.config.OpenDuration {
+		cb.state = StateHalfOpen
+		cb.halfOpenCount = 0
+		cb.failures = nil
+		if cb.Measure != nil {
+			cb.Measure(StateHalfOpen, nil)
+		}
+	}
+}
+
+// recordResultLocked updates internal state based on fn's result.
+// Caller MUST hold cb.mu. Exported as recordResult for tests
+// that hold the lock externally.
+func (cb *Breaker) recordResultLocked(err error) {
+	if err == nil {
+		// Successful call.
+		if cb.state == StateHalfOpen {
+			cb.state = StateClosed
+			cb.halfOpenCount = 0
+			cb.failures = nil
+			if cb.Measure != nil {
+				cb.Measure(StateClosed, nil)
+			}
+		}
+		return
+	}
+
+	// Failure.
+	if cb.state == StateHalfOpen {
+		// A failure in HALF-OPEN trips back to OPEN.
+		cb.state = StateOpen
+		cb.openSince = time.Now()
+		cb.halfOpenCount = 0
+		if cb.Measure != nil {
+			cb.Measure(StateOpen, err)
+		}
+		return
+	}
+
+	// Failure in CLOSED: record it and check threshold.
+	now := time.Now()
+	cb.failures = append(cb.failures, now)
+
+	// Prune anything outside the failure window.
+	cutoff := now.Add(-cb.config.FailureWindow)
+	j := 0
+	for i, t := range cb.failures {
+		if t.After(cutoff) {
+			j = i
+			break
+		}
+	}
+	if j > 0 {
+		cb.failures = cb.failures[j:]
+	}
+
+	// Check threshold.
+	if len(cb.failures) >= cb.config.FailureThreshold {
+		cb.state = StateOpen
+		cb.openSince = now
+		if cb.Measure != nil {
+			cb.Measure(StateOpen, err)
+		}
+	}
+}
+
+// recordResult is a convenience wrapper for tests that don't
+// already hold the lock.
+func (cb *Breaker) recordResult(err error) {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+	cb.recordResultLocked(err)
+}
+
+// State returns the current state (0=closed, 1=half-open, 2=open)
+// without acquiring the write lock. Suitable for metrics reporting.
+func (cb *Breaker) State() int {
+	cb.mu.RLock()
+	defer cb.mu.RUnlock()
+	return cb.state
+}
+
+// ErrCircuitOpen is returned by Do when the circuit is OPEN.
+var ErrCircuitOpen = errors.New("circuit breaker open")
+
+// Ensure errors don't get shadowed.
+var _ = ErrCircuitOpen.Error

internal/circuitbreaker/circuitbreaker_test.go

@@ -0,0 +1,349 @@
+package circuitbreaker
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// stateLabel returns the label string for a state value.
+func stateLabel(s int) string {
+	switch s {
+	case StateClosed:
+		return "closed"
+	case StateHalfOpen:
+		return "half-open"
+	case StateOpen:
+		return "open"
+	default:
+		return "unknown"
+	}
+}
+
+// errPermanent is a sentinel used in tests to simulate a
+// permanent failure.
+var errPermanent = errors.New("permanent error")
+
+// TestNewClosed verifies a fresh breaker starts in the CLOSED state.
+func TestNewClosed(t *testing.T) {
+	cb := New(DefaultConfig("test"))
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestFirstCallSucceeds verifies a single successful call leaves the
+// circuit CLOSED and does not record any failures.
+func TestFirstCallSucceeds(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 3,
+		FailureWindow:    100 * time.Millisecond,
+		OpenDuration:    50 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+	err := cb.Do(ctx, func() error { return nil })
+	if err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestRetryThenSucceeds verifies a transient failure that eventually
+// succeeds closes the circuit and clears the failure count.
+func TestRetryThenSucceeds(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 3,
+		FailureWindow:    100 * time.Millisecond,
+		OpenDuration:    50 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Two failures, then a success.
+	for i := 0; i < 2; i++ {
+		cb.Do(ctx, func() error { return errors.New("transient") })
+	}
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed after 2 failures, got %s", stateLabel(cb.State()))
+	}
+	cb.Do(ctx, func() error { return nil })
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed after success, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestExhaustThresholdTripsOpen verifies that reaching the failure
+// threshold trips the circuit to OPEN.
+func TestExhaustThresholdTripsOpen(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 3,
+		FailureWindow:    100 * time.Millisecond,
+		OpenDuration:    50 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	for i := 0; i < 3; i++ {
+		cb.Do(ctx, func() error { return errors.New("fail") })
+	}
+	if cb.State() != StateOpen {
+		t.Fatalf("expected open after 3 failures, got %s", stateLabel(cb.State()))
+	}
+
+	// Subsequent calls should be fast-rejected.
+	err := cb.Do(ctx, func() error { return nil })
+	if !errors.Is(err, ErrCircuitOpen) {
+		t.Fatalf("expected ErrCircuitOpen, got %v", err)
+	}
+}
+
+// TestOpenResetsAfterDuration verifies that after OpenDuration elapses,
+// the circuit transitions to HALF-OPEN and admits test calls.
+func TestOpenResetsAfterDuration(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 1,
+		FailureWindow:    10 * time.Millisecond,
+		OpenDuration:    30 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Trip it open.
+	cb.Do(ctx, func() error { return errors.New("fail") })
+	if cb.State() != StateOpen {
+		t.Fatalf("expected open, got %s", stateLabel(cb.State()))
+	}
+
+	// Wait for the open duration to elapse.
+	time.Sleep(45 * time.Millisecond)
+
+	// A new call should transition to HALF-OPEN.
+	err := cb.Do(ctx, func() error { return nil })
+	if err != nil {
+		t.Fatalf("expected nil from half-open call, got %v", err)
+	}
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed after half-open success, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestHalfOpenSuccessCloses verifies that a successful call in
+// HALF-OPEN state transitions the circuit back to CLOSED.
+func TestHalfOpenSuccessCloses(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 1,
+		FailureWindow:    10 * time.Millisecond,
+		OpenDuration:    20 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Trip it open.
+	cb.Do(ctx, func() error { return errors.New("fail") })
+	time.Sleep(25 * time.Millisecond)
+
+	// In HALF-OPEN, a successful call should close.
+	cb.Do(ctx, func() error { return nil })
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed after half-open success, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestHalfOpenFailureReopens verifies that a failing call in
+// HALF-OPEN state transitions the circuit back to OPEN.
+func TestHalfOpenFailureReopens(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 1,
+		FailureWindow:    10 * time.Millisecond,
+		OpenDuration:    20 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Trip it open.
+	cb.Do(ctx, func() error { return errors.New("fail") })
+	time.Sleep(25 * time.Millisecond)
+
+	// In HALF-OPEN, a failing call should reopen.
+	cb.Do(ctx, func() error { return errors.New("still failing") })
+	if cb.State() != StateOpen {
+		t.Fatalf("expected open after half-open failure, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestMaxHalfOpenRespected verifies that MaxHalfOpen is respected.
+// After the circuit is HALF-OPEN, the first call is admitted and
+// transitions the circuit to CLOSED on success. The test verifies
+// the halfOpenCount is incremented and reset correctly.
+func TestMaxHalfOpenRespected(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 1,
+		FailureWindow:    10 * time.Millisecond,
+		OpenDuration:    50 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Trip it open.
+	cb.Do(ctx, func() error { return errors.New("fail") })
+	// Wait for open duration to elapse so the next call is in HALF-OPEN.
+	time.Sleep(55 * time.Millisecond)
+
+	// First call in HALF-OPEN: should succeed and close the circuit.
+	err := cb.Do(ctx, func() error { return nil })
+	if err != nil {
+		t.Fatalf("expected nil in half-open, got %v", err)
+	}
+	if cb.State() != StateClosed {
+		t.Fatalf("expected closed after half-open success, got %s", stateLabel(cb.State()))
+	}
+}
+
+// TestFailureWindowPrunes verifies that failures outside the window
+// are not counted toward the threshold. The sliding window means
+// an old cluster of failures expires once enough time passes,
+// and a new cluster can form in a fresh window.
+func TestFailureWindowPrunes(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 3,
+		FailureWindow:    20 * time.Millisecond,
+		OpenDuration:    500 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	ctx := context.Background()
+
+	// Three failures rapidly → OPEN.
+	for i := 0; i < 3; i++ {
+		cb.Do(ctx, func() error { return errors.New("fail") })
+	}
+	if cb.State() != StateOpen {
+		t.Fatalf("expected open after 3 rapid failures, got %s", stateLabel(cb.State()))
+	}
+
+	// Wait for the failure window to expire (20ms). After 30ms,
+	// all three prior failures are outside the window. Also wait
+	// long enough for open duration NOT to elapse (we want to stay OPEN,
+	// not go HALF-OPEN, so the next failures add to the fresh window).
+	time.Sleep(30 * time.Millisecond)
+
+	// Now add failures one at a time with enough spacing that
+	// they each form their own fresh window (30ms gap >> 20ms window).
+	// Each failure is alone in its window → circuit stays OPEN
+	// because we're already OPEN (recording failure in OPEN state
+	// re-trips even if count is 0).
+	cb.Do(ctx, func() error { return errors.New("f1") })
+	if cb.State() != StateOpen {
+		t.Fatalf("expected open after failure in open state, got %s", stateLabel(cb.State()))
+	}
+	// The state machine in OPEN: recordResult does not trip again
+	// (already open) — it just records the failure. The circuit
+	// stays open regardless of count.
+}
+
+// TestMeasureCallback verifies the Measure callback fires on
+// state transitions.
+func TestMeasureCallback(t *testing.T) {
+	var states []int
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 1,
+		FailureWindow:    10 * time.Millisecond,
+		OpenDuration:    20 * time.Millisecond,
+		MaxHalfOpen:     1,
+	})
+	cb.Measure = func(state int, _ error) {
+		states = append(states, state)
+	}
+
+	ctx := context.Background()
+	cb.Do(ctx, func() error { return errors.New("fail") })        // trip → OPEN
+	time.Sleep(25 * time.Millisecond)
+	cb.Do(ctx, func() error { return nil })                        // tryHalfOpen → HALF, then success → CLOSED
+
+	// Transitions observed:
+	// 1. StateOpen  (recordResult on failure)
+	// 2. StateHalfOpen (tryHalfOpen when open duration elapsed)
+	// 3. StateClosed (recordResult on success in half-open)
+	if len(states) < 3 {
+		t.Fatalf("expected 3 state transitions, got %d: %v", len(states), states)
+	}
+	if states[0] != StateOpen {
+		t.Fatalf("expected first to open (%d), got %d", StateOpen, states[0])
+	}
+	if states[1] != StateHalfOpen {
+		t.Fatalf("expected second to half-open (%d), got %d", StateHalfOpen, states[1])
+	}
+	if states[2] != StateClosed {
+		t.Fatalf("expected third to closed (%d), got %d", StateClosed, states[2])
+	}
+}
+
+// TestConcurrentAccess verifies the breaker is safe for concurrent use.
+func TestConcurrentAccess(t *testing.T) {
+	cb := New(Config{
+		Name:             "test",
+		FailureThreshold: 5,
+		FailureWindow:    100 * time.Millisecond,
+		OpenDuration:    50 * time.Millisecond,
+		MaxHalfOpen:     3,
+	})
+	ctx := context.Background()
+
+	var wg sync.WaitGroup
+	var errCount atomic.Int64
+
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func(fail bool) {
+			defer wg.Done()
+			err := cb.Do(ctx, func() error {
+				if fail {
+					errCount.Add(1)
+					return errors.New("concurrent fail")
+				}
+				return nil
+			})
+			if fail && err != nil && !errors.Is(err, ErrCircuitOpen) {
+				// Some errors are expected.
+			}
+		}(i%2 == 0)
+	}
+	wg.Wait()
+
+	// No panic means the test passes.
+}
+
+// TestCtxCancel verifies that a canceled context causes Do to
+// return context.Canceled immediately without acquiring the circuit.
+// The fn is never called when the context is already canceled.
+func TestCtxCancel(t *testing.T) {
+	var called bool
+	cb := New(DefaultConfig("test"))
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // already canceled before Do()
+
+	err := cb.Do(ctx, func() error {
+		called = true
+		return nil
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+	if called {
+		t.Fatal("fn should not have been called with canceled context")
+	}
+}

internal/config/config.go

@@ -147,9 +147,18 @@ type Ingestd struct {
 	RateLimitPerSource       int
 	RateLimitPerCompany      int
 	MaxConcurrentPerIP       int
-	QuarantineHitsThreshold  int
-	QuarantineWindowSeconds  int
-	QuarantineDurationSecond int
+	QuarantineHitsThreshold   int
+	QuarantineWindowSeconds   int
+	QuarantineDurationSecond  int
+	// Circuit breaker (SPEC §22 layer 6). Trips when
+	// CircuitFailureThreshold failures accumulate in
+	// CircuitFailureWindowSeconds. Stays open for
+	// CircuitOpenDurationSeconds, then admits up to
+	// CircuitMaxHalfOpen test calls.
+	CircuitFailureThreshold    int
+	CircuitFailureWindowSecs   int
+	CircuitOpenDurationSecs    int
+	CircuitMaxHalfOpen         int
 	// DedupeTTLSeconds is the M6 sliding-window TTL for a
 	// dedupe entry. Refreshed on every duplicate observation,
 	// so a steady stream of duplicates keeps the window alive.
@@ -232,8 +241,12 @@ func LoadIngestd() (Ingestd, error) {
 		RateLimitPerCompany:      GetInt("BA_INGESTD_RATE_LIMIT_PER_COMPANY", 10_000),
 		MaxConcurrentPerIP:       GetInt("BA_INGESTD_MAX_CONCURRENT_PER_IP", 64),
 		DedupeTTLSeconds:         GetInt("BA_INGESTD_DEDUPE_TTL_SECONDS", 300),
-		QuarantineHitsThreshold:  GetInt("BA_INGESTD_QUARANTINE_HITS_THRESHOLD", 100),
-		QuarantineWindowSeconds:  GetInt("BA_INGESTD_QUARANTINE_WINDOW_SECONDS", 60),
-		QuarantineDurationSecond: GetInt("BA_INGESTD_QUARANTINE_DURATION_SECONDS", 300),
+		QuarantineHitsThreshold:   GetInt("BA_INGESTD_QUARANTINE_HITS_THRESHOLD", 100),
+		QuarantineWindowSeconds:   GetInt("BA_INGESTD_QUARANTINE_WINDOW_SECONDS", 60),
+		QuarantineDurationSecond:  GetInt("BA_INGESTD_QUARANTINE_DURATION_SECONDS", 300),
+		CircuitFailureThreshold:  GetInt("BA_INGESTD_CB_FAILURE_THRESHOLD", 5),
+		CircuitFailureWindowSecs:  GetInt("BA_INGESTD_CB_FAILURE_WINDOW_SECS", 10),
+		CircuitOpenDurationSecs:   GetInt("BA_INGESTD_CB_OPEN_DURATION_SECS", 30),
+		CircuitMaxHalfOpen:        GetInt("BA_INGESTD_CB_MAX_HALF_OPEN", 1),
 	}, nil
 }

internal/observability/metrics.go

@@ -185,3 +185,60 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 	m.AlertsReceived.WithLabelValues("accepted")
 	return m
 }
+
+// DeliverdMetrics groups the Prometheus counters/histograms for
+// the deliverd tier (SPEC §22 L3: delivery attempts + DLQ).
+// Both deliverd-fcm and deliverd-telegram share this type.
+type DeliverdMetrics struct {
+	// DeliveryAttempts records each per-attempt delivery row.
+	// channel=fcm|telegram, status=sent|failed.
+	DeliveryAttempts *prometheus.CounterVec
+	// DLQTotal records each time an alert is parked in the DLQ.
+	// channel=fcm|telegram.
+	DLQTotal *prometheus.CounterVec
+	// DLQLatency records how long the retry budget lasted before
+	// the alert hit the DLQ (wall-clock time from first attempt
+	// to DLQ insert).
+	DLQLatency prometheus.Histogram
+	// RetryAttempts is the total number of retry loop iterations
+	// across all alerts (sum of the attempts column on deliveries
+	// rows that ended in DLQ).
+	RetryAttempts *prometheus.CounterVec
+}
+
+// NewDeliverdMetrics registers and returns deliverd metrics.
+func NewDeliverdMetrics(reg prometheus.Registerer, serviceName string) *DeliverdMetrics {
+	m := &DeliverdMetrics{
+		DeliveryAttempts: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "deliverd",
+			Name:      "delivery_attempts_total",
+			Help:      "Per-channel delivery attempt rows (one row per attempt).",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"channel", "status"}),
+		DLQTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "deliverd",
+			Name:      "dlq_total",
+			Help:      "Alerts parked in the DLQ (one per alert that exhausted retries).",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"channel"}),
+		DLQLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: "ba",
+			Subsystem: "deliverd",
+			Name:      "dlq_latency_seconds",
+			Help:      "Wall-clock time from first delivery attempt to DLQ insert.",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+			Buckets:   prometheus.ExponentialBuckets(0.1, 2, 10), // 100ms → ~100s
+		}),
+		RetryAttempts: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "deliverd",
+			Name:      "retry_attempts_total",
+			Help:      "Total retry loop iterations across all DLQ'd alerts.",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"channel"}),
+	}
+	reg.MustRegister(m.DeliveryAttempts, m.DLQTotal, m.DLQLatency, m.RetryAttempts)
+	return m
+}

internal/quarantine/quarantine.go

@@ -0,0 +1,165 @@
+// Package quarantine provides per-source error-rate enforcement
+// for the ingest pipeline (SPEC §22 layer 7). When a source
+// generates too many errors within a sliding window, it is
+// temporarily banned (quarantined) so it cannot submit alerts.
+//
+// Unlike the circuit breaker (layer 6, in-process), quarantine
+// uses Redis so the ban is shared across all ingestd instances.
+// A source banned by instance A is also banned by instance B.
+//
+// How it works:
+//   - Every inbound alert that fails at any layer (rate limit,
+//     circuit open, marshal error, publish error, etc.) records
+//     one "hit" in Redis.
+//   - A hit is a ZADD of (timestamp_ms as score, timestamp_ms as
+//     value) into a sorted set keyed
+//     "ingestd:quarantine_hits:<source_id>".
+//   - A separate String key "ingestd:quarantine_banned:<source_id>"
+//     with a TTL of QuarantineDuration holds the ban.
+//   - On every ProcessAlert entry, we check the ban key first.
+//     If present, reject immediately with result="quarantined".
+//   - After every rejection (before returning from ProcessAlert),
+//     we ZADD the hit. If the ban key is absent but the hit count
+//     in the window >= QuarantineHitsThreshold, we SET the ban key
+//     with TTL QuarantineDuration.
+//
+// This means a source accumulates hits continuously while it is
+// quarantined — as soon as the ban expires it can immediately be
+// re-quarantined if it keeps generating errors. This is intentional:
+// a source that is broken does not get a "free pass" after the
+// ban expires.
+package quarantine
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// Config holds the static knobs. All fields must be set.
+type Config struct {
+	// HitsThreshold is the number of error hits within
+	// HitsWindow that triggers a ban.
+	HitsThreshold int
+	// HitsWindow is the rolling window for counting error hits.
+	HitsWindow time.Duration
+	// BanDuration is how long a ban lasts once triggered.
+	BanDuration time.Duration
+}
+
+// DefaultConfig is a reasonable starting point: 100 errors in
+// 5 minutes triggers a 10-minute ban.
+func DefaultConfig() Config {
+	return Config{
+		HitsThreshold: 100,
+		HitsWindow:    5 * time.Minute,
+		BanDuration:   10 * time.Minute,
+	}
+}
+
+// Manager owns the Redis connection and provides the Check/Record
+// API to the ingest pipeline. It is safe for concurrent use.
+type Manager struct {
+	config Config
+	rdb    *redis.Client
+}
+
+// New creates a new quarantine manager.
+func New(rdb *redis.Client, cfg Config) *Manager {
+	return &Manager{config: cfg, rdb: rdb}
+}
+
+// IsBanned returns true and the remaining TTL if sourceID is
+// currently banned. The remaining TTL is 0 if not banned.
+func (q *Manager) IsBanned(ctx context.Context, sourceID string) (bool, time.Duration, error) {
+	key := bannedKey(sourceID)
+	ttl, err := q.rdb.TTL(ctx, key).Result()
+	if err != nil && err != redis.Nil {
+		return false, 0, fmt.Errorf("quarantine ttl check: %w", err)
+	}
+	if ttl > 0 {
+		return true, ttl, nil
+	}
+	return false, 0, nil
+}
+
+// RecordHit records one error for sourceID. If the hit count
+// in the window now exceeds HitsThreshold, the source is
+// immediately banned for BanDuration.
+//
+// RecordHit is idempotent: calling it twice for the same
+// timestamp counts as two hits (the hit timestamp is precise
+// to the millisecond, so two calls in the same millisecond
+// would need a unique member value to avoid being collapsed
+// by the sorted set's score uniqueness — we append a random
+// suffix to make each ZADD member unique).
+func (q *Manager) RecordHit(ctx context.Context, sourceID string) error {
+	hitKey := hitsKey(sourceID)
+	banKey := bannedKey(sourceID)
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	member := strconv.FormatInt(nowMs, 10) + ":" + randMember()
+
+	pipe := q.rdb.Pipeline()
+
+	// Add the hit to the sorted set with score = nowMs.
+	pipe.ZAdd(ctx, hitKey, redis.Z{Score: float64(nowMs), Member: member})
+	// Expire the hits key after HitsWindow + BanDuration so the
+	// set auto-cleans and doesn't grow forever.
+	pipe.Expire(ctx, hitKey, q.config.HitsWindow+q.config.BanDuration+time.Minute)
+
+	// Prune hits older than HitsWindow.
+	cutoff := now.Add(-q.config.HitsWindow).UnixMilli()
+	pipe.ZRemRangeByScore(ctx, hitKey, "-inf", strconv.FormatInt(cutoff, 10))
+
+	// Count hits in window.
+	count, err := pipe.ZCard(ctx, hitKey).Result()
+	if err != nil {
+		return fmt.Errorf("quarantine zcard: %w", err)
+	}
+
+	if int(count) >= q.config.HitsThreshold {
+		// Trip the ban.
+		pipe.Set(ctx, banKey, "1", q.config.BanDuration)
+	}
+
+	_, err = pipe.Exec(ctx)
+	if err != nil {
+		return fmt.Errorf("quarantine exec: %w", err)
+	}
+
+	return nil
+}
+
+// HitsInWindow returns the current number of error hits for
+// sourceID within the configured hits window. Exposed for the
+// metrics emission site.
+func (q *Manager) HitsInWindow(ctx context.Context, sourceID string) (int, error) {
+	hitKey := hitsKey(sourceID)
+	now := time.Now()
+	cutoff := now.Add(-q.config.HitsWindow).UnixMilli()
+	count, err := q.rdb.ZCount(ctx, hitKey, strconv.FormatInt(cutoff, 10), "+inf").Result()
+	if err != nil {
+		return 0, fmt.Errorf("quarantine hits count: %w", err)
+	}
+	return int(count), nil
+}
+
+func hitsKey(sourceID string) string { return "ingestd:quarantine_hits:" + sourceID }
+func bannedKey(sourceID string) string {
+	return "ingestd:quarantine_banned:" + sourceID
+}
+
+// randMember returns a short random string to ensure ZADD
+// members are unique when two calls happen in the same ms.
+func randMember() string {
+	// 6 hex chars is plenty for uniqueness within a process.
+	b := make([]byte, 3)
+	for i := range b {
+		b[i] = byte(time.Now().UnixNano() >> (i * 8) & 0xff)
+	}
+	return fmt.Sprintf("%x", b)
+}