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