ronantakizawa/codereview-bench huggingface dataset:
Repository: prometheus/prometheus
File: web/api/v1/api_test.go (Go)

Code before:
				Result: promql.Scalar{
					V: 0.333,
					T: timestamp.FromTime(start.Add(123 * time.Second)),
				},
			},
		},
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{"0.333"},
			},
			response: &QueryData{
				ResultType: parser.ValueTypeScalar,
				Result: promql.Scalar{
					V: 0.333,
					T: timestamp.FromTime(api.now()),
				},
			},
		},
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{
					`label_replace(vector(42), "foo", "bar", "", "") or label_replace(vector(3.1415), "dings", "bums", "", "")`,
				},
				"time": []string{"123.4"},
			},
			responseAsJSON: `{
		"resultType": "vector",
		"result": [
			{
				"metric": {
					"foo": "bar"
				},
				"value": [123.4, "42"]
			},
			{
				"metric": {
					"dings": "bums"
				},
				"value": [123.4, "3.1415"]
			}
		]
	}`,
		},
		{
			endpoint: api.queryRange,
			query: url.Values{
				"query": []string{"time()"},
				"start": []string{"0"},
				"end":   []string{"2"},

Review comment (suggestion):
I would add a `"limit": []string{"2"},` here and then repeat the test with `"limit": []string{"1"},` and `"limit": []string{"0"},`.

Code after:
				Result: promql.Scalar{
					V: 0.333,
					T: timestamp.FromTime(start.Add(123 * time.Second)),
				},
			},
		},
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{"0.333"},
			},
			response: &QueryData{
				ResultType: parser.ValueTypeScalar,
				Result: promql.Scalar{
					V: 0.333,
					T: timestamp.FromTime(api.now()),
				},
			},
		},
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{
					`label_replace(vector(42), "foo", "bar", "", "") or label_replace(vector(3.1415), "dings", "bums", "", "")`,
				},
				"time":  []string{"123.4"},
				"limit": []string{"2"},
			},
			warningsCount: 0,
			responseAsJSON: `{
		"resultType": "vector",
		"result": [
			{
				"metric": {
					"foo": "bar"
				},
				"value": [123.4, "42"]
			},
			{
				"metric": {
					"dings": "bums"
				},
				"value": [123.4, "3.1415"]
			}
		]
	}`,
		},
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{
Repository: prometheus/prometheus
File: web/api/v1/api_test.go (Go)

Code before:
				"end":   []string{"2"},
				"step":  []string{"1"},
			},
			response: &QueryData{
				ResultType: parser.ValueTypeMatrix,
				Result: promql.Matrix{
					promql.Series{
						Floats: []promql.FPoint{
							{F: 0, T: timestamp.FromTime(start)},
							{F: 1, T: timestamp.FromTime(start.Add(1 * time.Second))},
							{F: 2, T: timestamp.FromTime(start.Add(2 * time.Second))},
						},
						// No Metric returned - use zero value for comparison.
					},
				},
			},
		},
		// Test empty vector result
		{
			endpoint: api.query,
			query: url.Values{
				"query": []string{"bottomk(2, notExists)"},
			},
			responseAsJSON: `{"resultType":"vector","result":[]}`,
		},
		// limit disabled
		{
			endpoint: api.queryRange,
			query: url.Values{
				"query": []string{"bottomk(2, notExists)"},
				"start": []string{"0"},
				"end":   []string{"2"},
				"step":  []string{"1"},
				"limit": []string{"0"},
			},
			responseAsJSON: `{"resultType":"matrix","result":[]}`,
			warningsCount:  0,
		},
		{
			endpoint: api.queryRange,
			query: url.Values{
				"query": []string{"bottomk(2, notExists)"},
				"start": []string{"0"},
				"end":   []string{"2"},
				"step":  []string{"1"},
				"limit": []string{"1"},
			},
			responseAsJSON: `{"resultType":"matrix","result":[]}`,
			warningsCount:  0,
		},
		{

Review comment (suggestion):
The remaining tests below (through line 1392) don't make sense as they only have one metric. If you repeat the tests above with the different permutations as recommended, you can remove all the tests below.

Code after:
				},
				"time":  []string{"123.4"},
				"limit": []string{"0"},
			},
			responseAsJSON: `{
		"resultType": "vector",
		"result": [
			{
				"metric": {
					"foo": "bar"
				},
				"value": [123.4, "42"]
			},
			{
				"metric": {
					"dings": "bums"
				},
				"value": [123.4, "3.1415"]
			}
		]
	}`,
			warningsCount: 0,
		},
		// limit=0 means no limit.
		{
			endpoint: api.queryRange,
			query: url.Values{
				"query": []string{
					`label_replace(vector(42), "foo", "bar", "", "") or label_replace(vector(3.1415), "dings", "bums", "", "")`,
				},
				"start": []string{"0"},
				"end":   []string{"2"},
				"step":  []string{"1"},
				"limit": []string{"0"},
			},
			response: &QueryData{
				ResultType: parser.ValueTypeMatrix,
				Result: promql.Matrix{
					promql.Series{
						Metric: labels.FromMap(map[string]string{"dings": "bums"}),
						Floats: []promql.FPoint{
							{F: 3.1415, T: timestamp.FromTime(start)},
							{F: 3.1415, T: timestamp.FromTime(start.Add(1 * time.Second))},
							{F: 3.1415, T: timestamp.FromTime(start.Add(2 * time.Second))},
						},
					},
					promql.Series{
						Metric: labels.FromMap(map[string]string{"foo": "bar"}),
						Floats: []promql.FPoint{
							{F: 42, T: timestamp.FromTime(start)},
							{F: 42, T: timestamp.FromTime(start.Add(1 * time.Second))},
Repository: prometheus/prometheus
File: util/annotations/annotations.go (Go)

Code before:
	if *a == nil {
		*a = Annotations{}
	}
	prevErr, exists := (*a)[err.Error()]
	if exists {
		var anErr AnnoError
		if errors.As(err, &anErr) {
			err = anErr.Merge(prevErr)
		}
	}
	(*a)[err.Error()] = err
	return *a
}

// Merge adds the contents of the second set of Annotations to the first, modifying
// the first in-place, and returns the merged first Annotations for convenience.
func (a *Annotations) Merge(aa Annotations) Annotations {
	if *a == nil {
		if aa == nil {
			return nil
		}
		*a = Annotations{}
	}
	for key, val := range aa {
		prevVal, exists := (*a)[key]
		if exists {
			var anErr AnnoError
			if errors.As(val, &anErr) {
				val = anErr.Merge(prevVal)
			}
		}
		(*a)[key] = val
	}
	return *a
}

// AsErrors is a convenience function to return the annotations map as a slice
// of errors.
func (a Annotations) AsErrors() []error {
	arr := make([]error, 0, len(a))
	for _, err := range a {
		arr = append(arr, err)
	}
	return arr
}

// AsStrings is a convenience function to return the annotations map as 2 slices
// of strings, separated into warnings and infos. The query string is used to get the
// line number and character offset positioning info of the elements which trigger an
// annotation. We limit the number of warnings and infos returned here with maxWarnings
// and maxInfos respectively (0 for no limit).

Review comment (suggestion):
As above, this should work:

```suggestion
		if prevVal, exists := (*a)[key]; exists {
```

Code after:
	if *a == nil {
		*a = Annotations{}
	}
	if prevErr, exists := (*a)[err.Error()]; exists {
		var anErr annoError
		if errors.As(err, &anErr) {
			err = anErr.Merge(prevErr)
		}
	}
	(*a)[err.Error()] = err
	return *a
}

// Merge adds the contents of the second set of Annotations to the first, modifying
// the first in-place, and returns the merged first Annotations for convenience.
func (a *Annotations) Merge(aa Annotations) Annotations {
	if *a == nil {
		if aa == nil {
			return nil
		}
		*a = Annotations{}
	}
	for key, val := range aa {
		if prevVal, exists := (*a)[key]; exists {
			var anErr annoError
			if errors.As(val, &anErr) {
				val = anErr.Merge(prevVal)
			}
		}
		(*a)[key] = val
	}
	return *a
}

// AsErrors is a convenience function to return the annotations map as a slice
// of errors.
func (a Annotations) AsErrors() []error {
	arr := make([]error, 0, len(a))
	for _, err := range a {
		arr = append(arr, err)
	}
	return arr
}

// AsStrings is a convenience function to return the annotations map as 2 slices
// of strings, separated into warnings and infos. The query string is used to get the
// line number and character offset positioning info of the elements which trigger an
// annotation. We limit the number of warnings and infos returned here with maxWarnings
// and maxInfos respectively (0 for no limit).
func (a Annotations) AsStrings(query string, maxWarnings, maxInfos int) (warnings, infos []string) {
	warnings = make([]string, 0, maxWarnings+1)
Repository: prometheus/prometheus
File: util/annotations/annotations.go (Go)

Code before:
}

type annoErr struct {
	PositionRange posrange.PositionRange
	Err           error
	Query         string
}

func (e *annoErr) Error() string {
	if e.Query == "" {
		return e.Err.Error()
	}
	return fmt.Sprintf("%s (%s)", e.Err, e.PositionRange.StartPosInput(e.Query, 0))
}

func (e *annoErr) Unwrap() error {
	return e.Err
}

func (e *annoErr) SetQuery(query string) {
	e.Query = query
}

func (e *annoErr) Merge(_ error) error {
	return e
}

func maybeAddMetricName(anno error, metricName string) error {
	if metricName == "" {
		return anno
	}
	return fmt.Errorf("%w for metric name %q", anno, metricName)
}

// NewInvalidQuantileWarning is used when the user specifies an invalid quantile
// value, i.e. a float that is outside the range [0, 1] or NaN.
func NewInvalidQuantileWarning(q float64, pos posrange.PositionRange) error {
	return &annoErr{
		PositionRange: pos,
		Err:           fmt.Errorf("%w, got %g", InvalidQuantileWarning, q),
	}
}

// NewInvalidRatioWarning is used when the user specifies an invalid ratio
// value, i.e. a float that is outside the range [-1, 1] or NaN.
func NewInvalidRatioWarning(q, to float64, pos posrange.PositionRange) error {
	return &annoErr{
		PositionRange: pos,
		Err:           fmt.Errorf("%w, got %g, capping to %g", InvalidRatioWarning, q, to),
	}
}

Review comment (suggestion):
Let's have a doc comment to explain that this implementation doesn't actually merge but rather ignores the provided error.

Code after:
}

type annoErr struct {
	PositionRange posrange.PositionRange
	Err           error
	Query         string
}

func (e *annoErr) Error() string {
	if e.Query == "" {
		return e.Err.Error()
	}
	return fmt.Sprintf("%s (%s)", e.Err, e.PositionRange.StartPosInput(e.Query, 0))
}

func (e *annoErr) Unwrap() error {
	return e.Err
}

func (e *annoErr) SetQuery(query string) {
	e.Query = query
}

// We do not merge generic annotations, instead we just ignore the provided error
// and return the original.
func (e *annoErr) Merge(_ error) error {
	return e
}

func maybeAddMetricName(anno error, metricName string) error {
	if metricName == "" {
		return anno
	}
	return fmt.Errorf("%w for metric name %q", anno, metricName)
}

// NewInvalidQuantileWarning is used when the user specifies an invalid quantile
// value, i.e. a float that is outside the range [0, 1] or NaN.
func NewInvalidQuantileWarning(q float64, pos posrange.PositionRange) error {
	return &annoErr{
		PositionRange: pos,
		Err:           fmt.Errorf("%w, got %g", InvalidQuantileWarning, q),
	}
}

// NewInvalidRatioWarning is used when the user specifies an invalid ratio
// value, i.e. a float that is outside the range [-1, 1] or NaN.
func NewInvalidRatioWarning(q, to float64, pos posrange.PositionRange) error {
	return &annoErr{
		PositionRange: pos,
		Err:           fmt.Errorf("%w, got %g, capping to %g", InvalidRatioWarning, q, to),
Repository: prometheus/prometheus
File: rules/group.go (Go)

Code before:
				_, err = app.Append(0, lset, timestamp.FromTime(ts.Add(-ruleQueryOffset)), math.Float64frombits(value.StaleNaN))
				unwrappedErr := errors.Unwrap(err)
				if unwrappedErr == nil {
					unwrappedErr = err
				}
				switch {
				case unwrappedErr == nil:
				case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample),
					errors.Is(unwrappedErr, storage.ErrTooOldSample),
					errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):
					// Do not count these in logging, as this is expected if series
					// is exposed from a different rule.
				default:
					logger.Warn("Adding stale sample failed", "sample", lset.String(), "err", err)
				}
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}
	for _, concurrentRules := range ctrl.Sort(ctx, g) {
		for _, ruleIdx := range concurrentRules {
			select {
			case <-g.done:
				return
			default:
			}

			rule := g.rules[ruleIdx]
			if len(concurrentRules) > 1 && ctrl.Allow(ctx, g, rule) {
				wg.Add(1)

				go eval(ruleIdx, rule, func() {
					wg.Done()
					ctrl.Done(ctx)
				})
			} else {
				eval(ruleIdx, rule, nil)
			}
		}
		wg.Wait()
	}

	g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())
	g.cleanupStaleSeries(ctx, ts)
}


Review comment (performance):
```suggestion
	for _, batch := range ctrl.Sort(ctx, g) {
		for _, ruleIndex := range batch {
```

Code after:
				_, err = app.Append(0, lset, timestamp.FromTime(ts.Add(-ruleQueryOffset)), math.Float64frombits(value.StaleNaN))
				unwrappedErr := errors.Unwrap(err)
				if unwrappedErr == nil {
					unwrappedErr = err
				}
				switch {
				case unwrappedErr == nil:
				case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample),
					errors.Is(unwrappedErr, storage.ErrTooOldSample),
					errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):
					// Do not count these in logging, as this is expected if series
					// is exposed from a different rule.
				default:
					logger.Warn("Adding stale sample failed", "sample", lset.String(), "err", err)
				}
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}
	for _, batch := range ctrl.SplitGroupIntoBatches(ctx, g) {
		for _, ruleIndex := range batch {
			select {
			case <-g.done:
				return
			default:
			}

			rule := g.rules[ruleIndex]
			if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
				wg.Add(1)

				go eval(ruleIndex, rule, func() {
					wg.Done()
					ctrl.Done(ctx)
				})
			} else {
				eval(ruleIndex, rule, nil)
			}
		}
		// It is important that we finish processing any rules in this current batch - before we move into the next one.
		wg.Wait()
	}

	g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())
	g.cleanupStaleSeries(ctx, ts)
}
Repository: prometheus/prometheus
File: rules/group.go (Go)

Code before:
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}
	for _, concurrentRules := range ctrl.Sort(ctx, g) {
		for _, ruleIdx := range concurrentRules {
			select {
			case <-g.done:
				return
			default:
			}

			rule := g.rules[ruleIdx]
			if len(concurrentRules) > 1 && ctrl.Allow(ctx, g, rule) {
				wg.Add(1)

				go eval(ruleIdx, rule, func() {
					wg.Done()
					ctrl.Done(ctx)
				})
			} else {
				eval(ruleIdx, rule, nil)
			}
		}
		wg.Wait()
	}

	g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())
	g.cleanupStaleSeries(ctx, ts)
}

func (g *Group) QueryOffset() time.Duration {
	if g.queryOffset != nil {
		return *g.queryOffset
	}

	if g.opts.DefaultRuleQueryOffset != nil {
		return g.opts.DefaultRuleQueryOffset()
	}

	return time.Duration(0)
}

func (g *Group) cleanupStaleSeries(ctx context.Context, ts time.Time) {
	if len(g.staleSeries) == 0 {
		return
	}
	app := g.opts.Appendable.Appender(ctx)
	app.SetOptions(g.appOpts)
	queryOffset := g.QueryOffset()
	for _, s := range g.staleSeries {

Review comment (performance):
```suggestion
                // It is important that we finish processing any rules in this current batch - before we move into the next one.
		wg.Wait()
```

Code after:
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}
	for _, batch := range ctrl.SplitGroupIntoBatches(ctx, g) {
		for _, ruleIndex := range batch {
			select {
			case <-g.done:
				return
			default:
			}

			rule := g.rules[ruleIndex]
			if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
				wg.Add(1)

				go eval(ruleIndex, rule, func() {
					wg.Done()
					ctrl.Done(ctx)
				})
			} else {
				eval(ruleIndex, rule, nil)
			}
		}
		// It is important that we finish processing any rules in this current batch - before we move into the next one.
		wg.Wait()
	}

	g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())
	g.cleanupStaleSeries(ctx, ts)
}

func (g *Group) QueryOffset() time.Duration {
	if g.queryOffset != nil {
		return *g.queryOffset
	}

	if g.opts.DefaultRuleQueryOffset != nil {
		return g.opts.DefaultRuleQueryOffset()
	}

	return time.Duration(0)
}

func (g *Group) cleanupStaleSeries(ctx context.Context, ts time.Time) {
	if len(g.staleSeries) == 0 {
		return
	}
	app := g.opts.Appendable.Appender(ctx)
	app.SetOptions(g.appOpts)
	queryOffset := g.QueryOffset()
Repository: prometheus/prometheus
File: rules/group.go (Go)

Code before:
				unwrappedErr := errors.Unwrap(err)
				if unwrappedErr == nil {
					unwrappedErr = err
				}
				switch {
				case unwrappedErr == nil:
				case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample),
					errors.Is(unwrappedErr, storage.ErrTooOldSample),
					errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):
					// Do not count these in logging, as this is expected if series
					// is exposed from a different rule.
				default:
					logger.Warn("Adding stale sample failed", "sample", lset.String(), "err", err)
				}
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}

	batches := ctrl.SplitGroupIntoBatches(ctx, g)
	if batches == nil || len(batches) == len(g.rules) {
		// Sequential evaluation.
		for i, rule := range g.rules {
			eval(i, rule, nil)
		}
	} else {
		// Concurrent evaluation.
		for _, batch := range ctrl.SplitGroupIntoBatches(ctx, g) {
			for _, ruleIndex := range batch {
				rule := g.rules[ruleIndex]
				if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
					wg.Add(1)

					go eval(ruleIndex, rule, func() {
						wg.Done()
						ctrl.Done(ctx)
					})
				} else {
					eval(ruleIndex, rule, nil)
				}
			}
			// It is important that we finish processing any rules in this current batch - before we move into the next one.
			wg.Wait()
		}
	}


Review comment (performance):
I think we should scope out this look into a separate function so that we can unit test it - I'd love to feed it different outcomes of `SplitGroupIntoBatches` (e.g. no batch, batch != g.rules, both 0 batch and 0 rules, 1 batch and 1 rule, 3 batch and 3 rules, 3 batch and 5 rules and 5 batch and 3 rules) just to ensure this does what we think it does in the correct way. 

Code after:
					errors.Is(unwrappedErr, storage.ErrTooOldSample),
					errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):
					// Do not count these in logging, as this is expected if series
					// is exposed from a different rule.
				default:
					logger.Warn("Adding stale sample failed", "sample", lset.String(), "err", err)
				}
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}

	batches := ctrl.SplitGroupIntoBatches(ctx, g)
	if len(batches) == 0 {
		// Sequential evaluation when batches aren't set.
		// This is the behaviour without a defined RuleConcurrencyController
		for i, rule := range g.rules {
			// Check if the group has been stopped.
			select {
			case <-g.done:
				return
			default:
			}
			eval(i, rule, nil)
		}
	} else {
		// Concurrent evaluation.
		for _, batch := range batches {
			for _, ruleIndex := range batch {
				// Check if the group has been stopped.
				select {
				case <-g.done:
					wg.Wait()
					return
				default:
				}
				rule := g.rules[ruleIndex]
				if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
					wg.Add(1)

					go eval(ruleIndex, rule, func() {
						wg.Done()
						ctrl.Done(ctx)
					})
				} else {
					eval(ruleIndex, rule, nil)
Repository: prometheus/prometheus
File: rules/group.go (Go)

Code before:
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}

	batches := ctrl.SplitGroupIntoBatches(ctx, g)
	if len(batches) == 0 {
		// Sequential evaluation when batches aren't set.
		// This is the behaviour without a defined RuleConcurrencyController
		for i, rule := range g.rules {
			// Check if the group has been stopped.
			select {
			case <-g.done:
				return
			default:
			}
			eval(i, rule, nil)
		}
	} else {
		// Concurrent evaluation.
		for _, batch := range ctrl.SplitGroupIntoBatches(ctx, g) {
			for _, ruleIndex := range batch {
				// Check if the group has been stopped.
				select {
				case <-g.done:
					return
				default:
				}
				rule := g.rules[ruleIndex]
				if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
					wg.Add(1)

					go eval(ruleIndex, rule, func() {
						wg.Done()
						ctrl.Done(ctx)
					})
				} else {
					eval(ruleIndex, rule, nil)
				}
			}
			// It is important that we finish processing any rules in this current batch - before we move into the next one.
			wg.Wait()
		}
	}

	g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())

Review comment (performance):
```suggestion
		for _, batch := range batches {
```

Code after:
			}
		}
	}

	var wg sync.WaitGroup
	ctrl := g.opts.RuleConcurrencyController
	if ctrl == nil {
		ctrl = sequentialRuleEvalController{}
	}

	batches := ctrl.SplitGroupIntoBatches(ctx, g)
	if len(batches) == 0 {
		// Sequential evaluation when batches aren't set.
		// This is the behaviour without a defined RuleConcurrencyController
		for i, rule := range g.rules {
			// Check if the group has been stopped.
			select {
			case <-g.done:
				return
			default:
			}
			eval(i, rule, nil)
		}
	} else {
		// Concurrent evaluation.
		for _, batch := range batches {
			for _, ruleIndex := range batch {
				// Check if the group has been stopped.
				select {
				case <-g.done:
					wg.Wait()
					return
				default:
				}
				rule := g.rules[ruleIndex]
				if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
					wg.Add(1)

					go eval(ruleIndex, rule, func() {
						wg.Done()
						ctrl.Done(ctx)
					})
				} else {
					eval(ruleIndex, rule, nil)
				}
			}
			// It is important that we finish processing any rules in this current batch - before we move into the next one.
			wg.Wait()
		}
	}