broad-announce/internal/observability/maxseen.go

// 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
}