fix(dispatch): reduce locking contention

Reduce the amount of time spent holding locks in the dispatcher:

- doMaintenance() holds RW lock only during deletion
- Groups() holds R lock, copies, releases
- processAlert() holds R(W) locks only when necessary

This results in -70% maintenance overhead or +19518.99% alert processing rate improvement:
```
                                                 │ bench-dispatch-main.txt │        bench-dispatch-fix-locking.txt         │
                                                 │   baseline_alerts/sec   │ baseline_alerts/sec  vs base                  │
Dispatch_100k_AggregationGroups_10k_Empty-12                 1.899M ±   1%          1.919M ±  2%          ~ (p=0.063 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12                 1.934M ±   2%          1.967M ±  1%     +1.69% (p=0.015 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12                 1.926M ±   3%          1.931M ±  4%          ~ (p=0.436 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12                 2.087M ±  10%          2.030M ±  7%          ~ (p=0.912 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12                 1.922M ±   3%          2.118M ± 10%    +10.16% (p=0.000 n=10)
Dispatch_20k_AggregationGroups_Groups_Impact-12              180.7k ±  42%         2128.6k ±  4%  +1077.93% (p=0.000 n=10)
Dispatch_50k_AggregationGroups_Groups_Impact-12              57.02k ± 196%        2089.55k ±  2%  +3564.69% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_Groups_Impact-12             19.61k ±  35%        1899.27k ±  2%  +9582.74% (p=0.000 n=10)
geomean                                                      524.6k                 2.008M         +282.84%

                                             │ bench-dispatch-main.txt │         bench-dispatch-fix-locking.txt         │
                                             │ maintenance_overhead_%  │ maintenance_overhead_%  vs base                │
Dispatch_100k_AggregationGroups_10k_Empty-12              17.185 ±  7%              5.672 ± 45%  -66.99% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12               36.50 ± 11%              12.56 ± 14%  -65.59% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12               55.44 ± 13%              23.11 ± 30%  -58.32% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12              125.65 ± 27%              28.98 ± 53%  -76.94% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12              172.40 ± 36%              37.76 ± 82%  -78.10% (p=0.000 n=10)
geomean                                                    59.62                    17.83        -70.10%
```

Signed-off-by: Siavash Safi <siavash@cloudflare.com>

Author: siavashs

dispatch/dispatch.go

@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"sort"
 	"sync"
 	"time"
@@ -185,19 +186,32 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 }
 
 func (d *Dispatcher) doMaintenance() {
+	d.mtx.RLock()
+	empty := make(map[*Route]map[model.Fingerprint]*aggrGroup)
+	for route, groups := range d.aggrGroupsPerRoute {
+		for fp, ag := range groups {
+			if ag.empty() {
+				if empty[route] == nil {
+					empty[route] = make(map[model.Fingerprint]*aggrGroup)
+				}
+				empty[route][fp] = ag
+			}
+		}
+	}
+	d.mtx.RUnlock()
 	d.mtx.Lock()
-	defer d.mtx.Unlock()
-	for _, groups := range d.aggrGroupsPerRoute {
-		for _, ag := range groups {
+	for route, groups := range empty {
+		for fp, ag := range groups {
 			if ag.empty() {
 				ag.stop()
 				d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey())
-				delete(groups, ag.fingerprint())
+				delete(d.aggrGroupsPerRoute[route], fp)
 				d.aggrGroupsNum--
 				d.metrics.aggrGroups.Dec()
 			}
 		}
 	}
+	d.mtx.Unlock()
 }
 
 // AlertGroup represents how alerts exist within an aggrGroup.
@@ -224,16 +238,19 @@ func (ag AlertGroups) Len() int { return len(ag) }
 func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*types.Alert, time.Time) bool) (AlertGroups, map[model.Fingerprint][]string) {
 	groups := AlertGroups{}
 
+	// Take a snapshot copy of the map to avoid blocking alert processing.
+	snapshot := make(map[*Route]map[model.Fingerprint]*aggrGroup)
 	d.mtx.RLock()
-	defer d.mtx.RUnlock()
+	maps.Copy(d.aggrGroupsPerRoute, snapshot)
+	d.mtx.RUnlock()
 
 	// Keep a list of receivers for an alert to prevent checking each alert
 	// again against all routes. The alert has already matched against this
 	// route on ingestion.
 	receivers := map[model.Fingerprint][]string{}
 
 	now := time.Now()
-	for route, ags := range d.aggrGroupsPerRoute {
+	for route, ags := range snapshot {
 		if !routeFilter(route) {
 			continue
 		}
@@ -316,30 +333,34 @@ func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
 	fp := groupLabels.Fingerprint()
 
 	d.mtx.Lock()
-	defer d.mtx.Unlock()
-
 	routeGroups, ok := d.aggrGroupsPerRoute[route]
 	if !ok {
 		routeGroups = map[model.Fingerprint]*aggrGroup{}
 		d.aggrGroupsPerRoute[route] = routeGroups
 	}
 
 	ag, ok := routeGroups[fp]
+	d.mtx.Unlock()
 	if ok {
 		ag.insert(alert)
 		return
 	}
 
 	// If the group does not exist, create it. But check the limit first.
-	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && d.aggrGroupsNum >= limit {
+	d.mtx.RLock()
+	aggrGroupsNum := d.aggrGroupsNum
+	d.mtx.RUnlock()
+	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && aggrGroupsNum >= limit {
 		d.metrics.aggrGroupLimitReached.Inc()
-		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", d.aggrGroupsNum, "limit", limit, "alert", alert.Name())
+		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", aggrGroupsNum, "limit", limit, "alert", alert.Name())
 		return
 	}
 
 	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger)
+	d.mtx.Lock()
 	routeGroups[fp] = ag
 	d.aggrGroupsNum++
+	d.mtx.Unlock()
 	d.metrics.aggrGroups.Inc()
 
 	// Insert the 1st alert in the group before starting the group's run()