broad-announce/internal/dedupe/dedupe.go

// 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 <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.
//
// 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")