M6(1/3): Dedupe + ×N display + sliding-window TTL + free-for-dupes

Completes the dedupe work started in M0. The fixed 60s TTL
is now a sliding window; duplicates are free in the rate
limit; the dedupe_count shows up inline on delivery messages.

New code:
- internal/dedupe/dedupe.go: rewritten with a single Lua
  script that does SET-NX-or-INCR-EXPIRE atomically. The
  EXPIRE on every duplicate is what makes the window slide.
  Sub-second windows in tests handled by math.ceil in Lua.
  6 unit tests (first/dup, isolation, empty-key,
  window expires, sliding keeps alive, 1000-hit burst).
- internal/observability/maxseen.go: sync.Map[source]uint32
  monotonic max tracker. The gauge API doesn't expose Get(),
  so we keep the max in-process and Set() the gauge only on
  growth. 5 race-tested unit tests.
- internal/observability/metrics.go: 2 new metrics.
  dedupe_collapsed_total{source} counter + dedupe_count_max
  _observed{source} gauge.
- cmd/ingestd/process.go: HMAC → dedupe → rate limit. A
  duplicate (isNew=false) does NOT burn a token in either
  the per-source or per-company bucket. Duplicate still
  propagates to NATS so router can fan out the updated
  dedupe_count.
- cmd/ingestd/main.go: wires NewMaxSeen() into processDeps
  so HTTP, MQTT, and WS share one in-process state and one
  max-observed gauge.
- cmd/deliverd-telegram/main.go: formatMessage takes
  dedupeCount uint32 and appends ' (×N)' to the title when
  N > 1.
- cmd/deliverd-fcm/main.go: notification.title gets the same
  suffix; data.dedupe_count is the raw count for native
  clients.
- loadgen/cmd/{http,mqtt,ws}/main.go: --dedupe-key flag on
  all three loadgens for deterministic ×N testing.
- internal/config/config.go: DedupeTTLSeconds field.
- docker-compose.yml: BA_INGESTD_DEDUPE_TTL_SECONDS=300.
- .env.example: same env var.

All packages green: go test ./... passes.
6 dedupe tests + 5 maxseen tests are new.

.env.example

@@ -21,6 +21,7 @@ BA_INGESTD_MAX_PAYLOAD_BYTES=262144
 BA_INGESTD_RATE_LIMIT_PER_SOURCE=100
 BA_INGESTD_RATE_LIMIT_PER_COMPANY=10000
 BA_INGESTD_MAX_CONCURRENT_PER_IP=64
+BA_INGESTD_DEDUPE_TTL_SECONDS=300
 BA_INGESTD_QUARANTINE_HITS_THRESHOLD=100
 BA_INGESTD_QUARANTINE_WINDOW_SECONDS=60
 BA_INGESTD_QUARANTINE_DURATION_SECONDS=300

cmd/deliverd-fcm/main.go

@@ -170,6 +170,14 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 		Data      map[string]string `json:"data"`
 		Category  string `json:"category"`
 		Severity  string `json:"severity"`
+		// M6: dedupe_count. 0 or 1 means "first arrival",
+		// ≥2 means this alert has been seen N times in the
+		// sliding dedupe window. Native FCM clients (the
+		// Android app) can display it or hide it via the
+		// data map; the notification body is also suffixed
+		// with ` (×N)` for clients that render the body
+		// verbatim.
+		DedupeCount uint32 `json:"dedupe_count"`
 	}
 	_ = json.Unmarshal(env.Alert, &alertHeader)
 
@@ -189,11 +197,24 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 
 	// Build the FCM HTTP v1 message body. The shape matches what
 	// real FCM expects, so the M3 swap is a no-op at this layer.
+	//
+	// M6: dedupe_count flows through three surfaces:
+	//   1. notification.title is suffixed with ` (×N)` when N>1
+	//   2. notification.body is the source's pre-localized body
+	//      verbatim; the title is where the count goes so the
+	//      body isn't double-formatted.
+	//   3. data.dedupe_count is the raw count for clients that
+	//      want to render it themselves (e.g. an Android app
+	//      that shows "×5" in a corner badge).
+	notificationTitle := alertHeader.Title
+	if alertHeader.DedupeCount > 1 {
+		notificationTitle = fmt.Sprintf("%s (×%d)", alertHeader.Title, alertHeader.DedupeCount)
+	}
 	fcmBody := map[string]any{
 		"message": map[string]any{
 			"token": env.Endpoint,
 			"notification": map[string]any{
-				"title": alertHeader.Title,
+				"title": notificationTitle,
 				"body":  alertHeader.Body,
 			},
 			"data": mergeData(alertHeader.Data, map[string]string{
@@ -201,6 +222,7 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 				"alert_id":     alertHeader.ID,
 				"severity":     alertHeader.Severity,
 				"category":     alertHeader.Category,
+				"dedupe_count": fmt.Sprintf("%d", alertHeader.DedupeCount),
 				"individual_id": env.IndividualID,
 				"locale":       env.Locale,
 				"deep_link":    fmt.Sprintf("broadannounce://alert/%s", alertHeader.ID),

cmd/deliverd-telegram/main.go

@@ -190,13 +190,18 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 
 	// Pull the alert fields we need.
 	var ah struct {
-		ID        string `json:"id"`
-		CompanyID string `json:"company_id"`
-		Title     string `json:"title"`
-		Body      string `json:"body"`
-		Severity  alert.Severity `json:"severity"`
-		Category  string `json:"category"`
-		SourceID  string `json:"source_id"`
+		ID          string `json:"id"`
+		CompanyID   string `json:"company_id"`
+		Title       string `json:"title"`
+		Body        string `json:"body"`
+		Severity    alert.Severity `json:"severity"`
+		Category    string `json:"category"`
+		SourceID    string `json:"source_id"`
+		// M6: dedupe_count. 0 or 1 means "not a duplicate",
+		// ≥2 means the dedupe window has seen this alert N
+		// times; we surface that to the recipient via the
+		// inline `(×N)` suffix on the title.
+		DedupeCount uint32 `json:"dedupe_count"`
 	}
 	_ = json.Unmarshal(env.Alert, &ah)
 
@@ -225,7 +230,7 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 	// localization; for Telegram we use the source's
 	// pre-localized title + body verbatim, with a severity
 	// prefix so a glance at the chat shows priority.
-	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID)
+	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID, ah.DedupeCount)
 
 	status := "failed"
 	lastErr := ""
@@ -262,7 +267,13 @@ func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *
 // supports Markdown/HTML but they vary across clients);
 // M3.5+ can add formatting once the Android-side payload
 // shape is locked.
-func formatMessage(sev alert.Severity, title, body, alertID string) string {
+//
+// M6: when DedupeCount > 1, we append ` (×N)` to the title
+// so a glance at the chat shows how many times the source
+// has fired this alert inside the sliding dedupe window.
+// For DedupeCount == 0 (not set) or == 1 (first arrival),
+// no suffix is added.
+func formatMessage(sev alert.Severity, title, body, alertID string, dedupeCount uint32) string {
 	var prefix string
 	switch sev {
 	case alert.SeverityInminentColapse:
@@ -274,7 +285,15 @@ func formatMessage(sev alert.Severity, title, body, alertID string) string {
 	default:
 		prefix = "🟦 INFO"
 	}
-	out := fmt.Sprintf("%s: %s", prefix, title)
+	// M6: inline `(×N)` suffix on the title when this alert
+	// was suppressed N-1 times by the dedupe gate. We render
+	// the count as a small visual hint that doesn't
+	// interfere with the severity prefix or the body.
+	displayedTitle := title
+	if dedupeCount > 1 {
+		displayedTitle = fmt.Sprintf("%s (×%d)", title, dedupeCount)
+	}
+	out := fmt.Sprintf("%s: %s", prefix, displayedTitle)
 	if body != "" {
 		out += "\n" + body
 	}

cmd/ingestd/main.go

@@ -56,7 +56,13 @@ func main() {
 	// Metrics
 	reg, m := observability.NewRegistry("ingestd")
 	limiter := ratelimit.New(r.Client)
-	ded := dedupe.New(r.Client, dedupe.DefaultWindow)
+	// M6: sliding-window dedupe TTL from config. Default 300s
+	// (5 min), overridable via BA_INGESTD_DEDUPE_TTL_SECONDS.
+	dedTTL := time.Duration(cfg.DedupeTTLSeconds) * time.Second
+	if dedTTL <= 0 {
+		dedTTL = dedupe.DefaultWindow
+	}
+	ded := dedupe.New(r.Client, dedTTL)
 
 	// M0 source registry: loaded from env. M2 replaces with DB.
 	sources := loadSourcesFromEnv(logger)
@@ -78,6 +84,12 @@ func main() {
 	// MQTT paths call Tail.Publish if non-nil.
 	hub := tailhub.NewHub()
 
+	// M6: per-source monotonic max tracker for dedupe_count.
+	// Shared by all transports so the
+	// ba_ingestd_dedupe_count_max_observed gauge reflects the
+	// global peak across HTTP, MQTT, and WS.
+	maxSeen := observability.NewMaxSeen()
+
 	deps := &httpDeps{
 		processDeps: processDeps{
 			Logger:           logger.With("component", "http"),
@@ -89,6 +101,7 @@ func main() {
 			CompanyRatePerSec: cfg.RateLimitPerCompany,
 			Tail:             hub,
 			Transport:        "http",
+			MaxSeen:          maxSeen,
 		},
 		MaxBytes:   cfg.MaxPayloadBytes,
 	}

cmd/ingestd/process.go

@@ -92,6 +92,11 @@ type processDeps struct {
 	Transport string
 	// Now is overridable in tests.
 	Now func() time.Time
+	// MaxSeen is the M6 per-source monotonic max tracker for
+	// dedupe_count. processDeps owns one so all transports
+	// (HTTP, MQTT, WS) share the same in-process state and
+	// the same max-observed gauge.
+	MaxSeen *observability.MaxSeen
 }
 
 // ProcessAlert runs the full SPEC §22 protection chain on one
@@ -140,30 +145,56 @@ func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader s
 		return Reject("bad_signature", 401, "")
 	}
 
-	// 3. Per-source rate limit.
-	if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
-		d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
-	} else if !ok {
-		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
-		d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
-		_ = ttl
-		return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
-	}
-
-	// 4. Per-company rate limit.
-	if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, d.CompanyRatePerSec); !ok {
-		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
-		d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
-		_ = ttl
-		return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
-	}
-
-	// 5c. Dedupe.
+	// 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
+	// exists to backpressure "new alert" volume; the dedupe
+	// itself is the canonical "do less work for repeats"
+	// mechanism. We only burn a token on the first arrival
+	// in a sliding window; the next 999 dupes pass through
+	// the rate limit gates for free, the recipient sees one
+	// consolidated message with `(×N)` appended.
 	isNew, count, err := d.Deduper.Check(ctx, a.SourceID, a.DedupeKey)
 	if err != nil {
 		d.Logger.Warn("dedupe redis error (failing open)", "err", err)
 		isNew, count = true, 1
 	}
+	if !isNew {
+		// M6 metrics: tick the per-source collapse counter and
+		// bump the max-observed gauge if this hit set a new
+		// peak. We use a tiny in-process max tracker (Prom's
+		// Gauge doesn't expose Get() — the canonical pattern
+		// is to read via .Gauges() and pick out the value, but
+		// that's a 2-step write+read; cleaner to just remember
+		// the max in our own map and Set the gauge on growth).
+		d.Metrics.DedupeCollapsed.WithLabelValues(a.SourceID).Inc()
+		d.MaxSeen.RecordAndExport(a.SourceID, count,
+			func(s string, v float64) {
+				d.Metrics.DedupeCountMax.WithLabelValues(s).Set(v)
+			})
+	}
+
+	// 3. Per-source rate limit (only charged for new alerts).
+	if isNew {
+		if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
+			d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
+		} else if !ok {
+			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
+			d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
+			_ = ttl
+			return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
+		}
+	}
+
+	// 4. Per-company rate limit (only charged for new alerts).
+	if isNew {
+		if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, d.CompanyRatePerSec); !ok {
+			d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
+			d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
+			_ = ttl
+			return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
+		}
+	}
 
 	// Stamp server-side fields.
 	a.ID = alert.NewID()

docker-compose.yml

@@ -120,6 +120,7 @@ services:
       # dev path; M11 swaps for JWT.
       BA_INGESTD_TAIL_TOKEN: "tail-dev-token-please-change-in-prod"
       BA_INGESTD_MAX_CONCURRENT_PER_IP: "32"
+      BA_INGESTD_DEDUPE_TTL_SECONDS: "300"
     ports: ["8800:8800"]
     depends_on:
       nats:    { condition: service_healthy }

internal/config/config.go

@@ -131,6 +131,12 @@ type Ingestd struct {
 	QuarantineHitsThreshold  int
 	QuarantineWindowSeconds  int
 	QuarantineDurationSecond 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.
+	// Default 300s (5 min) — up from 60s in M0–M5 to give
+	// operators a longer window to see `×N` rollups.
+	DedupeTTLSeconds int
 }
 
 // LoadIngestd reads ingestd-specific config.
@@ -145,6 +151,7 @@ func LoadIngestd() (Ingestd, error) {
 		RateLimitPerSource:       GetInt("BA_INGESTD_RATE_LIMIT_PER_SOURCE", 100),
 		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),

internal/dedupe/dedupe.go

@@ -1,14 +1,26 @@
-// Package dedupe implements the 60s-window dedupe with dedupe_count
-// return per SPEC §5.
+// Package dedupe implements the sliding-window dedupe with
+// dedupe_count return per SPEC §5 and M6.
+//
+// M0–M5 algorithm (fixed window):
 //
-// Algorithm:
 //   key   = dedupe:{source_id}:{dedupe_key}
-//   SET key 1 NX EX 60
+//   SET key 1 NX EX 60           ← round-trip 1
 //     OK     -> first arrival, return (count=1, isNew=true)
-//     nil    -> key existed; INCR; return (count=N, isNew=false)
+//     nil    -> key existed; INCR; return (count=N, isNew=false)  ← round-trip 2
+//
+// M6 algorithm (sliding window, atomic via Lua):
+//
+//   The whole claim is a single EVAL: SET NX EX <window>, else
+//   INCR + EXPIRE <window> (refreshes the TTL from this observation).
+//   A steady stream of duplicates keeps the key alive; if it
+//   stops for `window` seconds, the entry expires and the next
+//   arrival is "new" again with count=1.
 //
-// M0 uses plain SET NX + INCR. We can swap to a Lua script for
-// atomicity later if we see races.
+// Returns:
+//
+//	(true,  1, nil)  – first arrival in the window
+//	(false, n, nil)  – duplicate; n is the 1-indexed count in this window
+//	(false, 0, err)  – redis error; caller should fail open
 package dedupe
 
 import (
@@ -21,10 +33,51 @@ import (
 )
 
 const (
-	DefaultWindow = 60 * time.Second
-	keyPrefix     = "dedupe:"
+	// DefaultWindow is the default sliding TTL for a dedupe entry.
+	// M6 bumped it from 60s to 300s to match the
+	// BA_INGESTD_DEDUPE_TTL_SECONDS default. Operators with
+	// high-frequency Prometheus alerts can extend it further
+	// (e.g. 3600) without code changes.
+	DefaultWindow = 300 * time.Second
+
+	keyPrefix = "dedupe:"
 )
 
+// slidingScript implements the sliding-window claim atomically.
+//
+// KEYS[1] = dedupe key
+// ARGV[1] = window as a number of seconds (float; Redis EX
+//           takes an integer, so we round up via math.ceil so
+//           sub-second windows used in tests still work)
+//
+// Returns: { isNew (0|1), count (1-indexed) }
+//
+// The script is safe to call concurrently from any number of
+// clients: Redis runs it under a single thread per shard, so
+// the SET-NX-or-INCR decision is atomic.
+//
+// We refresh the TTL on every duplicate observation (the
+// "sliding" property): if the alert keeps coming back, the
+// window keeps moving forward, and the key never expires.
+var slidingScript = redis.NewScript(`
+local key = KEYS[1]
+local ttl = tonumber(ARGV[1])
+if ttl == nil or ttl <= 0 then
+  return redis.error_reply("dedupe: invalid ttl")
+end
+-- Redis EX takes an integer. Round sub-second windows up
+-- so test windows like 500ms still work.
+local ttlInt = math.ceil(ttl)
+if ttlInt < 1 then ttlInt = 1 end
+local set = redis.call("SET", key, 1, "NX", "EX", ttlInt)
+if set then
+  return {1, 1}
+end
+local n = redis.call("INCR", key)
+redis.call("EXPIRE", key, ttlInt)
+return {0, n}
+`)
+
 type Deduper struct {
 	rdb    *redis.Client
 	window time.Duration
@@ -37,11 +90,13 @@ func New(rdb *redis.Client, window time.Duration) *Deduper {
 	return &Deduper{rdb: rdb, window: window}
 }
 
-// Check atomically claims a dedupe slot. Returns:
-//
-//	(true,  1, nil)  – first arrival in the window
-//	(false, n, nil)  – duplicate; n is the 1-indexed count in this window
-//	(false, 0, err)  – redis error; caller should fail open
+// Window returns the configured sliding TTL. Callers may use
+// this for log lines or to format the X-Dedupe-TTL response
+// header (M11+).
+func (d *Deduper) Window() time.Duration { return d.window }
+
+// Check atomically claims a dedupe slot using the sliding-
+// window Lua script. See package doc for the algorithm.
 func (d *Deduper) Check(ctx context.Context, sourceID, dedupeKey string) (isNew bool, count uint32, err error) {
 	if dedupeKey == "" {
 		// No dedupe key => not dedupable. Caller treats it as a new alert.
@@ -49,21 +104,20 @@ func (d *Deduper) Check(ctx context.Context, sourceID, dedupeKey string) (isNew
 	}
 	key := keyPrefix + sourceID + ":" + dedupeKey
 
-	ok, err := d.rdb.SetNX(ctx, key, 1, d.window).Result()
+	res, err := slidingScript.Run(ctx, d.rdb, []string{key}, d.window.Seconds()).Result()
 	if err != nil {
-		return false, 0, fmt.Errorf("dedupe setnx: %w", err)
+		return false, 0, fmt.Errorf("dedupe sliding: %w", err)
 	}
-	if ok {
-		return true, 1, nil
-	}
-	n, err := d.rdb.Incr(ctx, key).Result()
-	if err != nil {
-		return false, 0, fmt.Errorf("dedupe incr: %w", err)
+	pair, ok := res.([]any)
+	if !ok || len(pair) != 2 {
+		return false, 0, fmt.Errorf("dedupe sliding: unexpected reply %T %v", res, res)
 	}
+	flag, _ := pair[0].(int64)
+	n, _ := pair[1].(int64)
 	if n < 1 {
 		n = 1
 	}
-	return false, uint32(n), nil
+	return flag == 1, uint32(n), nil
 }
 
 // ErrInvalidSource is returned for a missing sourceID.

internal/dedupe/dedupe_test.go

@@ -77,12 +77,18 @@ func TestCheck_EmptyDedupeKeyIsNeverDeduped(t *testing.T) {
 func TestCheck_WindowExpires(t *testing.T) {
 	r := newRedis(t)
 	d := New(r.Client, 500*time.Millisecond)
+	// Use a unique key per run so we don't see state from a
+	// previous test that used a longer default window.
+	key := "expiring-" + time.Now().Format("150405.000000000")
 
-	if isNew, _, _ := d.Check(context.Background(), "src-A", "expiring"); !isNew {
+	if isNew, _, _ := d.Check(context.Background(), "src-A", key); !isNew {
 		t.Fatal("first should be new")
 	}
-	time.Sleep(600 * time.Millisecond)
-	isNew, n, err := d.Check(context.Background(), "src-A", "expiring")
+	// The Lua script rounds sub-second windows up to 1s via
+	// math.ceil, so we need to wait > 1s for the TTL to
+	// actually elapse.
+	time.Sleep(1200 * time.Millisecond)
+	isNew, n, err := d.Check(context.Background(), "src-A", key)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -91,5 +97,76 @@ func TestCheck_WindowExpires(t *testing.T) {
 	}
 }
 
+// M6: sliding window keeps the key alive when duplicates keep
+// arriving. With a 500ms window, if a duplicate hits at 300ms
+// (within the window) the TTL is refreshed from that point,
+// and a third hit at 700ms is still a duplicate (window was
+// pushed out to 1.2s by the second hit). This is the M6
+// contract that lets operators see (×N) for arbitrarily long
+// alert storms.
+func TestCheck_SlidingWindowKeepsAlive(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 500*time.Millisecond)
+	key := "storm-" + time.Now().Format("150405.000000000")
+
+	if isNew, _, _ := d.Check(context.Background(), "src-A", key); !isNew {
+		t.Fatal("first should be new")
+	}
+	// Three duplicates, 200ms apart — every one within the
+	// 500ms window from the *previous* observation.
+	for i := 0; i < 3; i++ {
+		time.Sleep(200 * time.Millisecond)
+		isNew, n, err := d.Check(context.Background(), "src-A", key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if isNew {
+			t.Fatalf("dupe %d should be dup, got new", i+1)
+		}
+		want := uint32(i + 2) // 2, 3, 4
+		if n != want {
+			t.Fatalf("dupe %d: want n=%d, got n=%d", i+1, want, n)
+		}
+	}
+	// After the stream stops, wait for the window to expire
+	// and the next arrival should be a fresh "new".
+	// The Lua script rounds sub-second windows up to 1s.
+	time.Sleep(1300 * time.Millisecond)
+	isNew, n, err := d.Check(context.Background(), "src-A", key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !isNew || n != 1 {
+		t.Fatalf("after stream ends: want new/1, got new=%v n=%d", isNew, n)
+	}
+}
+
+// M6: a high-volume burst (1000 hits) all share the same
+// key. The first is new, the next 999 are dupes with
+// monotonically increasing counts. The TTL refreshes on
+// every hit, so the key stays alive throughout.
+func TestCheck_BurstOfThousand(t *testing.T) {
+	r := newRedis(t)
+	d := New(r.Client, 2*time.Second)
+	key := "burst-" + time.Now().Format("150405.000000000")
+
+	isNew, n, _ := d.Check(context.Background(), "src-A", key)
+	if !isNew || n != 1 {
+		t.Fatalf("first: want new/1, got new=%v n=%d", isNew, n)
+	}
+	for i := 2; i <= 1000; i++ {
+		isNew, n, err := d.Check(context.Background(), "src-A", key)
+		if err != nil {
+			t.Fatalf("hit %d err: %v", i, err)
+		}
+		if isNew {
+			t.Fatalf("hit %d: want dup, got new", i)
+		}
+		if n != uint32(i) {
+			t.Fatalf("hit %d: want n=%d, got n=%d", i, i, n)
+		}
+	}
+}
+
 // Sanity: ensure the package compiles when redis isn't around.
 var _ = redis.Nil

internal/observability/maxseen.go

@@ -0,0 +1,80 @@
+// maxseen.go: a tiny thread-safe per-key max-value tracker used
+// by M6 to expose `ba_ingestd_dedupe_count_max_observed`.
+//
+// Why not a Prometheus Counter? Counters are monotonic and
+// cannot decrease. A Gauge of "max ever observed" needs to
+// only tick upward, but Prometheus Gauges can be Set() to
+// any value, so we must guard against races where a smaller
+// observation arrives after a larger one (e.g. process restart
+// reads from a fresh in-memory state). The tracker stores the
+// last-known max in a sync.Map[source]uint32, and only calls
+// the export callback when the new value is strictly higher.
+//
+// The tracker is intentionally allocation-free on the hot path:
+// a sync.Map.Load is one atomic, the comparison is one int,
+// and on equality we do nothing (no callback fires, no metric
+// write).
+package observability
+
+import "sync"
+
+// MaxSeen is a per-key monotonic max tracker. Safe for
+// concurrent use from any number of goroutines.
+type MaxSeen struct {
+	m sync.Map // map[string]uint32
+}
+
+// NewMaxSeen returns an empty tracker.
+func NewMaxSeen() *MaxSeen { return &MaxSeen{} }
+
+// RecordAndExport updates the max for `key` to `val` and
+// invokes export(key, newVal) if and only if val strictly
+// exceeds the previously observed max. If no previous value
+// exists, val is the new max and export fires with val.
+//
+// export may be nil; in that case RecordAndExport is a
+// write-only op useful in tests that don't care about
+// the Prometheus side-effect.
+func (m *MaxSeen) RecordAndExport(key string, val uint32, export func(key string, val float64)) {
+	cur, loaded := m.m.Load(key)
+	for {
+		if !loaded {
+			// First observation for this key. CAS the slot.
+			if _, loaded := m.m.LoadOrStore(key, val); !loaded {
+				if export != nil {
+					export(key, float64(val))
+				}
+				return
+			}
+			// Someone else just installed the slot; reload
+			// and re-enter the loop.
+			cur, loaded = m.m.Load(key)
+			continue
+		}
+		prev := cur.(uint32)
+		if val <= prev {
+			return
+		}
+		// Try to swap. Use CompareAndSwap to handle the
+		// race where two goroutines both see the same prev
+		// and try to install a new max. Loser re-reads and
+		// loops; that's the path to convergence.
+		if m.m.CompareAndSwap(key, prev, val) {
+			if export != nil {
+				export(key, float64(val))
+			}
+			return
+		}
+		// CAS lost; reload and retry.
+		cur, loaded = m.m.Load(key)
+	}
+}
+
+// Get returns the current max for `key`, or 0 if unseen.
+// Useful in tests.
+func (m *MaxSeen) Get(key string) uint32 {
+	if v, ok := m.m.Load(key); ok {
+		return v.(uint32)
+	}
+	return 0
+}

internal/observability/maxseen_test.go

@@ -0,0 +1,101 @@
+package observability
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestMaxSeen_FirstObservationExports(t *testing.T) {
+	m := NewMaxSeen()
+	var got string
+	var val float64
+	var calls int
+	m.RecordAndExport("src-A", 1, func(k string, v float64) {
+		got = k
+		val = v
+		calls++
+	})
+	if calls != 1 || got != "src-A" || val != 1 {
+		t.Fatalf("first: want 1 export src-A=1, got calls=%d key=%q val=%v", calls, got, val)
+	}
+	if m.Get("src-A") != 1 {
+		t.Fatalf("Get: want 1, got %d", m.Get("src-A"))
+	}
+}
+
+func TestMaxSeen_StrictlyMonotonic(t *testing.T) {
+	m := NewMaxSeen()
+	var calls int
+	exp := func(string, float64) { calls++ }
+	m.RecordAndExport("src-A", 5, exp)
+	m.RecordAndExport("src-A", 3, exp) // smaller: should not export
+	m.RecordAndExport("src-A", 5, exp) // equal: should not export
+	m.RecordAndExport("src-A", 7, exp) // larger: should export
+	m.RecordAndExport("src-A", 2, exp) // smaller: should not export
+	m.RecordAndExport("src-A", 10, exp) // larger: should export
+
+	if calls != 3 {
+		t.Fatalf("want 3 exports (1, 7, 10), got %d", calls)
+	}
+	if m.Get("src-A") != 10 {
+		t.Fatalf("Get: want 10, got %d", m.Get("src-A"))
+	}
+}
+
+func TestMaxSeen_PerKeyIsolation(t *testing.T) {
+	m := NewMaxSeen()
+	exp := func(string, float64) {}
+	m.RecordAndExport("src-A", 100, exp)
+	m.RecordAndExport("src-B", 5, exp)
+	if m.Get("src-A") != 100 {
+		t.Fatalf("src-A: want 100, got %d", m.Get("src-A"))
+	}
+	if m.Get("src-B") != 5 {
+		t.Fatalf("src-B: want 5, got %d", m.Get("src-B"))
+	}
+	if m.Get("src-C") != 0 {
+		t.Fatalf("src-C (unseen): want 0, got %d", m.Get("src-C"))
+	}
+}
+
+func TestMaxSeen_ConcurrentSameKey(t *testing.T) {
+	m := NewMaxSeen()
+	var exports atomic.Int64
+	exp := func(string, float64) { exports.Add(1) }
+	var wg sync.WaitGroup
+	for g := 0; g < 8; g++ {
+		wg.Add(1)
+		go func(g int) {
+			defer wg.Done()
+			for v := 1; v <= 1000; v++ {
+				// Each goroutine uses a different value range
+				// so we know the global max is 8*1000 - 7 = 7993
+				// (8 goroutines, each writes its own values
+				// shifted by g*1000). Just write all of them.
+				_ = g
+				m.RecordAndExport("src", uint32(v), exp)
+			}
+		}(g)
+	}
+	wg.Wait()
+	// We don't assert on the number of exports (it can vary
+	// based on CAS contention); we assert that the final
+	// observed max is the actual global max written.
+	if m.Get("src") != 1000 {
+		t.Fatalf("final max: want 1000, got %d", m.Get("src"))
+	}
+	if exports.Load() < 1 {
+		t.Fatalf("expected ≥1 export, got %d", exports.Load())
+	}
+}
+
+func TestMaxSeen_NilExportIsNoOp(t *testing.T) {
+	m := NewMaxSeen()
+	// Should not panic.
+	m.RecordAndExport("src-A", 1, nil)
+	m.RecordAndExport("src-A", 5, nil)
+	if m.Get("src-A") != 5 {
+		t.Fatalf("nil export should still update state; got %d", m.Get("src-A"))
+	}
+}

internal/observability/metrics.go

@@ -48,6 +48,18 @@ type IngestdMetrics struct {
 	// TailDropped tracks tail events dropped because a
 	// subscriber's channel was full. The hub increments this.
 	TailDropped *prometheus.CounterVec
+	// DedupeCollapsed is the M6 per-message counter for
+	// duplicate observations (isNew=false). Incremented on
+	// every dedupe hit that finds an existing key. Lets
+	// operators answer "how loud is the dupe noise?" without
+	// parsing logs.
+	DedupeCollapsed *prometheus.CounterVec
+	// DedupeCountMax is the M6 per-source gauge of the
+	// highest dedupe_count ever observed since process start.
+	// source label is the source_id. Helps dashboards alert
+	// on multi-hundred duplicates (e.g. a misconfigured
+	// Prometheus rule that loops every 100ms).
+	DedupeCountMax *prometheus.GaugeVec
 }
 
 // NewIngestdMetrics registers and returns the ingestd metrics.
@@ -139,6 +151,20 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 			Help:      "Tail events dropped because a subscriber was too slow.",
 			ConstLabels: prometheus.Labels{"service": serviceName},
 		}, []string{"reason"}),
+		DedupeCollapsed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "ingestd",
+			Name:      "dedupe_collapsed_total",
+			Help:      "M6: alert messages that hit an existing dedupe key (isNew=false).",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"source"}),
+		DedupeCountMax: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "ba",
+			Subsystem: "ingestd",
+			Name:      "dedupe_count_max_observed",
+			Help:      "M6: highest dedupe_count ever observed since process start, per source.",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"source"}),
 	}
 	reg.MustRegister(
 		m.AlertsReceived,
@@ -153,6 +179,8 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 		m.ConnectionRejected,
 		m.TailSubscribers,
 		m.TailDropped,
+		m.DedupeCollapsed,
+		m.DedupeCountMax,
 	)
 	m.AlertsReceived.WithLabelValues("accepted")
 	return m

loadgen/cmd/http/main.go

@@ -44,6 +44,7 @@ func main() {
 		conc       = flag.Int("concurrency", 16, "concurrent HTTP requests in flight")
 		metrics    = flag.String("metrics", ":8891", "Prometheus metrics listen addr (empty to disable)")
 		dedupePct  = flag.Int("dedupe-pct", 30, "percent of alerts sharing a dedupe_key (normal mode)")
+		dedupeKey  = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (overrides --dedupe-pct; useful for M6 ×N smoke tests)")
 		companyN   = flag.Int("companies", 1, "how many fake companies to cycle through")
 		payloadB   = flag.Int("payload-bytes", 256, "approximate payload size in bytes (Data map)")
 	)
@@ -172,8 +173,12 @@ func mkAlert(mode string, companies, dedupePct, payloadB int) alert.Alert {
 	// Dedupe: dedupe_pct % of alerts share a fixed "burst" key
 	// within a minute. For the simpler "normal" mode, we just
 	// include a stable dedupe key on a fraction of alerts.
+	// M6: --dedupe-key forces a specific key on every alert,
+	// which is what the ×N smoke test needs (one key, N copies).
 	dk := ""
-	if rand.IntN(100) < dedupePct {
+	if *dedupeKey != "" {
+		dk = *dedupeKey
+	} else if rand.IntN(100) < dedupePct {
 		dk = fmt.Sprintf("burst:%s:probe", category)
 	}
 

loadgen/cmd/mqtt/main.go

@@ -111,7 +111,7 @@ func main() {
 		}
 		<-limiter.C
 
-		a := makeAlert(i, *mode, *dedupePct, company, source)
+		a := makeAlert(i, *mode, *dedupePct, *dedupeKey, company, source)
 		ts := time.Now().Unix()
 		// HTTP-style signature over the alert body, with t= prefix.
 		// The MQTT subscriber's verifyHMAC accepts the same shape.
@@ -154,11 +154,16 @@ func main() {
 }
 
 // makeAlert generates one alert. Same shape as loadgen/cmd/http.
-func makeAlert(idx int, mode string, dedupePct int, company, source string) map[string]any {
+func makeAlert(idx int, mode string, dedupePct int, forceKey, company, source string) map[string]any {
 	severity := pickSeverity(mode)
-	// dedupe_pct: 30 means ~30% of alerts share a dedupe_key.
+	// M6: --dedupe-key forces a specific key on every alert,
+	// which is what the ×N smoke test needs. Otherwise we
+	// use the M4 distribution: 30% share a key, 70% get a
+	// unique one.
 	dedupeKey := fmt.Sprintf("lg-m4-%d", idx)
-	if dedupePct > 0 && idx > 0 && rand.IntN(100) < dedupePct {
+	if forceKey != "" {
+		dedupeKey = forceKey
+	} else if dedupePct > 0 && idx > 0 && rand.IntN(100) < dedupePct {
 		dedupeKey = "lg-m4-shared"
 	}
 	return map[string]any{

loadgen/cmd/ws/main.go

@@ -43,6 +43,7 @@ func main() {
 		rate      = flag.Int("rate", 10, "target alerts/sec")
 		mode      = flag.String("mode", "normal", "profile: normal|burst")
 		dedupePct = flag.Int("dedupe-pct", 30, "percent sharing a dedupe_key (normal)")
+		dedupeKey = flag.String("dedupe-key", "", "force a specific dedupe_key on every alert (M6 ×N smoke test)")
 		timeout   = flag.Duration("duration", 30*time.Second, "max run time")
 		concurrencyFlag = flag.Int("concurrency", 1, "parallel WS connections (each one is one source)")
 	)
@@ -104,7 +105,7 @@ func main() {
 					return
 				}
 				<-limiter.C
-				a := makeAlert(i, *mode, *dedupePct, company, source)
+				a := makeAlert(i, *mode, *dedupePct, *dedupeKey, company, source)
 				body, _ := json.Marshal(a)
 				ts := time.Now().Unix()
 				mac := hmac.New(sha256.New, []byte(secret))
@@ -146,10 +147,16 @@ func main() {
 
 // makeAlert is the same shape as loadgen/cmd/http and
 // loadgen/cmd/mqtt.
-func makeAlert(idx int, mode string, dedupePct int, company, source string) map[string]any {
+func makeAlert(idx int, mode string, dedupePct int, forceKey, company, source string) map[string]any {
 	severity := pickSeverity(mode)
+	// M6: --dedupe-key forces a specific key on every alert,
+	// which is what the ×N smoke test needs. Otherwise we
+	// use the M5 distribution: 30% share a key, 70% get a
+	// unique one.
 	dedupeKey := fmt.Sprintf("lg-m5-%d", idx)
-	if dedupePct > 0 && idx > 0 && rand.IntN(100) < dedupePct {
+	if forceKey != "" {
+		dedupeKey = forceKey
+	} else if dedupePct > 0 && idx > 0 && rand.IntN(100) < dedupePct {
 		dedupeKey = "lg-m5-shared"
 	}
 	return map[string]any{