diff --git a/pkg/util/debouncer/debouncer.go b/pkg/util/debouncer/debouncer.go
new file mode 100644
index 00000000000..a6c0e7b22e3
--- /dev/null
+++ b/pkg/util/debouncer/debouncer.go
@@ -0,0 +1,310 @@
+package debouncer
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/grafana/dskit/instrument"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	ErrBufferFull = errors.New("debouncer buffer full")
+)
+
+type ProcessFunc[T comparable] func(context.Context, T) error
+type ErrorFunc[T comparable] func(T, error)
+
+type metrics struct {
+	itemsAddedCounter           prometheus.Counter
+	itemsDroppedCounter         prometheus.Counter
+	itemsProcessedCounter       prometheus.Counter
+	processingErrorsCounter     prometheus.Counter
+	processingDurationHistogram prometheus.Histogram
+}
+
+func newMetrics(reg prometheus.Registerer, name string) *metrics {
+	return &metrics{
+		itemsAddedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "debouncer_items_added_total",
+			Help: "Total number of items added to the debouncer",
+			ConstLabels: prometheus.Labels{
+				"name": name,
+			},
+		}),
+		itemsDroppedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "debouncer_items_dropped_total",
+			Help: "Total number of items dropped due to a full buffer",
+			ConstLabels: prometheus.Labels{
+				"name": name,
+			},
+		}),
+		itemsProcessedCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "debouncer_items_processed_total",
+			Help: "Total number of items processed by the debouncer",
+			ConstLabels: prometheus.Labels{
+				"name": name,
+			},
+		}),
+		processingErrorsCounter: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "debouncer_processing_errors_total",
+			Help: "Total number of errors during processing",
+			ConstLabels: prometheus.Labels{
+				"name": name,
+			},
+		}),
+		processingDurationHistogram: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
+			Name:                            "debouncer_processing_duration_seconds",
+			Help:                            "Time taken to process items",
+			Buckets:                         instrument.DefBuckets,
+			NativeHistogramBucketFactor:     1.1,
+			NativeHistogramMaxBucketNumber:  160,
+			NativeHistogramMinResetDuration: time.Hour,
+			ConstLabels: prometheus.Labels{
+				"name": name,
+			},
+		}),
+	}
+}
+
+// DebouncerOpts hold all the options to create a debouncer group.
+type DebouncerOpts[T comparable] struct {
+	// Name should be a unique name for this debouncer group. It is
+	// also used a name label value for the metrics.
+	Name string
+	// BufferSize is the maximum number of pending events to buffer.
+	BufferSize int
+
+	// ErrorHandler is the function that is called when a process for a given
+	// key returns an error while running.
+	ErrorHandler ErrorFunc[T]
+	// ProcessHandler is the function that is called once a process for a given
+	// key should be run.
+	ProcessHandler ProcessFunc[T]
+	// MinWait is the cooldown period after receiving an event. If another event with the
+	// same key arrives during this period, the timer resets and we wait another MinWait duration.
+	MinWait time.Duration
+	// MaxWait is the maximum time any event will wait before processing. Even if new events
+	// for the same key keep arriving, we guarantee processing after MaxWait from the first event.
+	MaxWait time.Duration
+	Reg     prometheus.Registerer
+}
+
+type Group[T comparable] struct {
+	buffer chan T
+
+	// mutex protecting the debouncers map.
+	debouncersMu sync.Mutex
+	debouncers   map[T]*debouncer[T]
+
+	wg             sync.WaitGroup
+	ctx            context.Context
+	cancel         context.CancelFunc
+	errorHandler   ErrorFunc[T]
+	processHandler ProcessFunc[T]
+	minWait        time.Duration
+	maxWait        time.Duration
+	metrics        *metrics
+}
+
+// NewGroup creates a new debouncer group for processing events with unique keys.
+//
+// A debouncer group helps optimize expensive operations by:
+// 1. Grouping identical events that occur in rapid succession
+// 2. Processing each unique key only once after waiting periods expire
+//
+// Example usage:
+//
+//	group := debouncer.NewGroup(DebouncerOpts[string]{
+//		BufferSize:     1000,
+//		ProcessHandler: func(ctx context.Context, key string) error {
+//		  // This is where you perform the expensive operation
+//		  return doSuperExpensiveCommand(key)
+//		}
+//		MinWait:        time.Second * 10,
+//		MaxWait:        time.Minute,
+//	})
+//
+//	// Start the debouncer group.
+//	group.Start(ctx)
+//
+//	// Queue events
+//	if err := group.Add("user-1"); err != nil {
+//	  // Do something with the error.
+//	}
+//	// Adding the same key resets MinWait but not MaxWait
+//	if err := group.Add("user-1"); err != nil {
+//	  // Do something with the error.
+//	}
+//
+// The event will be processed when either MinWait expires (after the most recent add)
+// or MaxWait expires (after the first add), whichever comes first.
+func NewGroup[T comparable](opts DebouncerOpts[T]) (*Group[T], error) {
+	if opts.BufferSize <= 0 {
+		opts.BufferSize = 100
+	}
+
+	if opts.MinWait <= 0 {
+		opts.MinWait = time.Minute
+	}
+	if opts.MaxWait <= 0 {
+		opts.MaxWait = 5 * time.Minute
+	}
+	if opts.MinWait > opts.MaxWait {
+		return nil, errors.New("minWait is bigger than maxWait")
+	}
+
+	if opts.ProcessHandler == nil {
+		return nil, errors.New("processHandler is required")
+	}
+
+	if opts.ErrorHandler == nil {
+		opts.ErrorHandler = func(_ T, _ error) {}
+	}
+
+	return &Group[T]{
+		buffer:         make(chan T, opts.BufferSize),
+		debouncers:     make(map[T]*debouncer[T]),
+		processHandler: opts.ProcessHandler,
+		errorHandler:   opts.ErrorHandler,
+		minWait:        opts.MinWait,
+		maxWait:        opts.MaxWait,
+		metrics:        newMetrics(opts.Reg, opts.Name),
+	}, nil
+}
+
+// Add will create a new debouncer for the given Key if it doesn't exist yet.
+// If a key has already a debouncer it will either reset the MinWait timer for
+// this key, or if they key is already running its process be no-op.
+func (g *Group[T]) Add(value T) error {
+	select {
+	case g.buffer <- value:
+		g.metrics.itemsAddedCounter.Inc()
+		return nil
+	default:
+		g.metrics.itemsDroppedCounter.Inc()
+		return ErrBufferFull
+	}
+}
+
+func (g *Group[T]) Start(ctx context.Context) {
+	g.ctx, g.cancel = context.WithCancel(ctx)
+	g.wg.Add(1)
+	go func() {
+		defer g.wg.Done()
+		for {
+			select {
+			case <-g.ctx.Done():
+				return
+			case value := <-g.buffer:
+				g.processValue(value)
+			}
+		}
+	}()
+}
+
+func (g *Group[T]) Stop() {
+	if g.cancel != nil {
+		g.cancel()
+		g.wg.Wait()
+	}
+}
+
+func (g *Group[T]) processValue(key T) {
+	g.debouncersMu.Lock()
+	deb, ok := g.debouncers[key]
+	if !ok {
+		deb = newDebouncer[T](g.minWait, g.maxWait, key, func(v T) {
+			g.processWithMetrics(g.ctx, v, g.processHandler)
+
+			g.debouncersMu.Lock()
+			defer g.debouncersMu.Unlock()
+			if current, exists := g.debouncers[key]; exists && current == deb {
+				delete(g.debouncers, key)
+			}
+		})
+		g.wg.Add(1)
+		go func() {
+			defer g.wg.Done()
+			deb.run(g.ctx)
+		}()
+		g.debouncers[key] = deb
+	}
+	g.debouncersMu.Unlock()
+
+	deb.reset()
+}
+
+func (g *Group[T]) processWithMetrics(ctx context.Context, value T, processFunc ProcessFunc[T]) {
+	timer := prometheus.NewTimer(g.metrics.processingDurationHistogram)
+	defer timer.ObserveDuration()
+	g.metrics.itemsProcessedCounter.Inc()
+
+	if err := processFunc(ctx, value); err != nil {
+		g.errorHandler(value, err)
+		g.metrics.processingErrorsCounter.Inc()
+	}
+}
+
+// debouncer handles debouncing for a specific key.
+type debouncer[T comparable] struct {
+	key         T
+	resetChan   chan struct{}
+	minWait     time.Duration
+	maxWait     time.Duration
+	processFunc func(T)
+}
+
+// newDebouncer creates a new key debouncer.
+func newDebouncer[T comparable](minWait, maxWait time.Duration, key T, processFunc func(T)) *debouncer[T] {
+	deb := &debouncer[T]{
+		key:         key,
+		resetChan:   make(chan struct{}, 1),
+		minWait:     minWait,
+		maxWait:     maxWait,
+		processFunc: processFunc,
+	}
+	return deb
+}
+
+// reset triggers a timer reset for the minWait.
+func (d *debouncer[T]) reset() {
+	select {
+	case d.resetChan <- struct{}{}:
+		// Value sent successfully.
+	default:
+		// Value was dropped. Is not an issue as
+		// a reset is already about to being processed
+		// or the process is being run.
+	}
+}
+
+// run manages the debouncing process for a specific key.
+func (d *debouncer[T]) run(ctx context.Context) {
+	// Create timers after getting the first updateChan.
+	minTimer := time.NewTimer(d.minWait)
+	maxTimer := time.NewTimer(d.maxWait)
+	defer func() {
+		minTimer.Stop()
+		maxTimer.Stop()
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-d.resetChan:
+			minTimer.Stop()
+			minTimer.Reset(d.minWait)
+		case <-minTimer.C:
+			d.processFunc(d.key)
+			return
+		case <-maxTimer.C:
+			d.processFunc(d.key)
+			return
+		}
+	}
+}
diff --git a/pkg/util/debouncer/debouncer_test.go b/pkg/util/debouncer/debouncer_test.go
new file mode 100644
index 00000000000..711c8761836
--- /dev/null
+++ b/pkg/util/debouncer/debouncer_test.go
@@ -0,0 +1,217 @@
+package debouncer
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDebouncer(t *testing.T) {
+	t.Run("should process values after min wait", func(t *testing.T) {
+		var processedMu sync.Mutex
+		processedValues := make(map[string]int)
+
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize: 10,
+			ProcessHandler: func(ctx context.Context, value string) error {
+				processedMu.Lock()
+				processedValues[value]++
+				processedMu.Unlock()
+				return nil
+			},
+			MinWait: 10 * time.Millisecond,
+			MaxWait: 500 * time.Millisecond,
+		})
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		group.Start(ctx)
+
+		require.NoError(t, group.Add("key1"))
+		require.NoError(t, group.Add("key2"))
+		// Should be deduplicated.
+		require.NoError(t, group.Add("key1"))
+
+		require.Eventually(t, func() bool {
+			// We should have processed key1 and key2 exactly once.
+			processedMu.Lock()
+			if processedValues["key1"] == 1 && processedValues["key2"] == 1 {
+				return true
+			}
+			processedMu.Unlock()
+			return false
+		}, time.Millisecond*200, time.Millisecond*20)
+	})
+
+	t.Run("should process values after max wait", func(t *testing.T) {
+		processed := make(map[string]int, 1)
+
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize: 10,
+			ProcessHandler: func(ctx context.Context, value string) error {
+				processed[value]++
+				return nil
+			},
+			MinWait: 50 * time.Millisecond,
+			MaxWait: 500 * time.Millisecond,
+		})
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		group.Start(ctx)
+
+		ticker := time.NewTicker(time.Millisecond * 40)
+		defer ticker.Stop()
+
+		start := time.Now()
+
+		for counter := 0; counter < 25; counter++ {
+			<-ticker.C
+			_ = group.Add("key1")
+			if processed["key1"] == 1 {
+				break
+			}
+		}
+
+		require.WithinDuration(t, start.Add(time.Millisecond*500), time.Now(), time.Millisecond*100)
+	})
+
+	t.Run("should handle buffer full", func(t *testing.T) {
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize:     1,
+			ProcessHandler: func(ctx context.Context, value string) error { return nil },
+			MinWait:        10 * time.Millisecond,
+			MaxWait:        100 * time.Millisecond,
+		})
+		require.NoError(t, err)
+
+		require.NoError(t, group.Add("key1"))
+		// Buffer should be full by now as we are not reading from it yet.
+		require.ErrorIs(t, group.Add("key2"), ErrBufferFull)
+	})
+
+	t.Run("should track metrics", func(t *testing.T) {
+		var wg sync.WaitGroup
+
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize: 10,
+			ProcessHandler: func(ctx context.Context, value string) error {
+				wg.Done()
+				return nil
+			},
+			MinWait: 10 * time.Millisecond,
+			MaxWait: 100 * time.Millisecond,
+
+			Name: "test",
+		})
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		group.Start(ctx)
+
+		wg.Add(1)
+		require.NoError(t, group.Add("key1"))
+		require.NoError(t, group.Add("key1"))
+
+		wg.Wait()
+
+		require.Equal(t, float64(2), testutil.ToFloat64(group.metrics.itemsAddedCounter))
+		require.Equal(t, float64(1), testutil.ToFloat64(group.metrics.itemsProcessedCounter))
+	})
+
+	t.Run("should handle errors", func(t *testing.T) {
+		var (
+			wg          sync.WaitGroup
+			errs        = make(chan error, 10)
+			expectedErr = errors.New("test error")
+		)
+
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize: 10,
+			ProcessHandler: func(ctx context.Context, value string) error {
+				wg.Done()
+				return expectedErr
+			},
+			MinWait:      10 * time.Millisecond,
+			MaxWait:      100 * time.Millisecond,
+			Reg:          prometheus.NewPedanticRegistry(),
+			Name:         "test_errors",
+			ErrorHandler: func(_ string, err error) { errs <- err },
+		})
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		group.Start(ctx)
+
+		wg.Add(1)
+		require.NoError(t, group.Add("key1"))
+
+		wg.Wait()
+
+		select {
+		case err := <-errs:
+			require.Equal(t, expectedErr, err)
+		default:
+			t.Fatal("expected error")
+		}
+
+		require.Equal(t, float64(1), testutil.ToFloat64(group.metrics.processingErrorsCounter))
+	})
+
+	t.Run("should gracefully handle stops", func(t *testing.T) {
+		// Create a channel to signal when processing is done.
+		done := make(chan struct{})
+
+		group, err := NewGroup(DebouncerOpts[string]{
+			BufferSize: 10,
+			ProcessHandler: func(ctx context.Context, item string) error {
+				// Start a goroutine to wait for context cancellation.
+				go func() {
+					<-ctx.Done()
+					close(done)
+				}()
+				return nil
+			},
+			MinWait: 50 * time.Millisecond,
+			MaxWait: 500 * time.Millisecond,
+		})
+		require.NoError(t, err)
+
+		// Start the group with a context
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		group.Start(ctx)
+
+		// Send an item to trigger processing.
+		require.NoError(t, group.Add("key-1"))
+
+		// Give the group a moment to process the item.
+		time.Sleep(100 * time.Millisecond)
+
+		// Stop the group, which should cancel the context.
+		group.Stop()
+
+		// Wait for the done signal or timeout.
+		select {
+		case <-done:
+			// Success - the group was stopped and the context was canceled
+		case <-time.After(time.Second):
+			t.Fatal("Timed out waiting for group to stop")
+		}
+	})
+}