// Package dedupe implements the sliding-window dedupe with // dedupe_count return per SPEC §5 and M6. // // M0–M5 algorithm (fixed window): // // key = dedupe:{source_id}:{dedupe_key} // 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) ← round-trip 2 // // M6 algorithm (sliding window, atomic via Lua): // // The whole claim is a single EVAL: SET NX EX , else // INCR + EXPIRE (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. // // 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 ( "context" "errors" "fmt" "time" "github.com/redis/go-redis/v9" ) const ( // 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 } func New(rdb *redis.Client, window time.Duration) *Deduper { if window <= 0 { window = DefaultWindow } return &Deduper{rdb: rdb, window: window} } // 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. return true, 1, nil } key := keyPrefix + sourceID + ":" + dedupeKey res, err := slidingScript.Run(ctx, d.rdb, []string{key}, d.window.Seconds()).Result() if err != nil { return false, 0, fmt.Errorf("dedupe sliding: %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 flag == 1, uint32(n), nil } // ErrInvalidSource is returned for a missing sourceID. var ErrInvalidSource = errors.New("dedupe: empty source id")