test(querier): reproduce the lost first runningDiff on a full cache hit

The metrics statement reads one step before the window for runningDiff,
and the cache does not keep that step.

Assisted-by: Claude Fable 5.1

pkg/alertmanager/alertmanagerserver/distpatcher_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/promslog"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -110,42 +111,44 @@ func TestAggrGroup(t *testing.T) {
 		}
 	)
 
-	var (
-		last       = time.Now()
-		current    = time.Now()
-		lastCurMtx = &sync.Mutex{}
-		alertsCh   = make(chan alertmanagertypes.AlertSlice)
-	)
+	type notification struct {
+		alerts     alertmanagertypes.AlertSlice
+		notifiedAt time.Time
+	}
+	alertsCh := make(chan notification)
 
 	ntfy := func(ctx context.Context, alerts ...*alertmanagertypes.Alert) bool {
 		// Validate that the context is properly populated.
-		if _, ok := notify.Now(ctx); !ok {
-			t.Errorf("now missing")
+		notifiedAt, ok := notify.Now(ctx)
+		assert.True(t, ok, "now missing")
+		_, ok = notify.GroupKey(ctx)
+		assert.True(t, ok, "group key missing")
+		lbls, ok := notify.GroupLabels(ctx)
+		if assert.True(t, ok, "group labels missing") {
+			assert.Equal(t, lset, lbls, "wrong group labels")
 		}
-		if _, ok := notify.GroupKey(ctx); !ok {
-			t.Errorf("group key missing")
+		rcv, ok := notify.ReceiverName(ctx)
+		if assert.True(t, ok, "receiver missing") {
+			assert.Equal(t, opts.Receiver, rcv, "wrong receiver")
 		}
-		if lbls, ok := notify.GroupLabels(ctx); !ok || !reflect.DeepEqual(lbls, lset) {
-			t.Errorf("wrong group labels: %q", lbls)
-		}
-		if rcv, ok := notify.ReceiverName(ctx); !ok || rcv != opts.Receiver {
-			t.Errorf("wrong receiver: %q", rcv)
-		}
-		if ri, ok := notify.RepeatInterval(ctx); !ok || ri != notificationConfig.Renotify.RenotifyInterval {
-			t.Errorf("wrong repeat interval: %q", ri)
+		ri, ok := notify.RepeatInterval(ctx)
+		if assert.True(t, ok, "repeat interval missing") {
+			assert.Equal(t, notificationConfig.Renotify.RenotifyInterval, ri, "wrong repeat interval")
 		}
 
-		lastCurMtx.Lock()
-		last = current
-		// Subtract a millisecond to allow for races.
-		current = time.Now().Add(-time.Millisecond)
-		lastCurMtx.Unlock()
-
-		alertsCh <- alertmanagertypes.AlertSlice(alerts)
+		alertsCh <- notification{
+			alerts:     alertmanagertypes.AlertSlice(alerts),
+			notifiedAt: notifiedAt,
+		}
 
 		return true
 	}
 
+	assertNotifiedAfter := func(previous, current time.Time, interval time.Duration) {
+		t.Helper()
+		require.GreaterOrEqual(t, current.Sub(previous), interval, "received batch too early")
+	}
+
 	removeEndsAt := func(as alertmanagertypes.AlertSlice) alertmanagertypes.AlertSlice {
 		for i, a := range as {
 			ac := *a
@@ -156,29 +159,26 @@ func TestAggrGroup(t *testing.T) {
 	}
 
 	// Test regular situation where we wait for group_wait to send out alerts.
+	groupStartedAt := time.Now()
 	ag := newAggrGroup(context.Background(), lset, route, nil, promslog.NewNopLogger(), notificationConfig.Renotify.RenotifyInterval)
 
 	go ag.run(ntfy)
 
 	ag.insert(a1)
+	var lastNotificationAt time.Time
 
 	select {
 	case <-time.After(2 * opts.GroupWait):
-		t.Fatalf("expected initial batch after group_wait")
+		require.FailNow(t, "expected initial batch after group_wait")
 
-	case batch := <-alertsCh:
-		lastCurMtx.Lock()
-		s := time.Since(last)
-		lastCurMtx.Unlock()
-		if s < opts.GroupWait {
-			t.Fatalf("received batch too early after %v", s)
-		}
+	case notification := <-alertsCh:
+		assertNotifiedAfter(groupStartedAt, notification.notifiedAt, opts.GroupWait)
+		lastNotificationAt = notification.notifiedAt
+		batch := notification.alerts
 		exp := removeEndsAt(alertmanagertypes.AlertSlice{a1})
 		sort.Sort(batch)
 
-		if !reflect.DeepEqual(batch, exp) {
-			t.Fatalf("expected alerts %v but got %v", exp, batch)
-		}
+		require.Equal(t, exp, batch)
 	}
 
 	for i := 0; i < 3; i++ {
@@ -187,21 +187,16 @@ func TestAggrGroup(t *testing.T) {
 
 		select {
 		case <-time.After(2 * opts.GroupInterval):
-			t.Fatalf("expected new batch after group interval but received none")
+			require.FailNow(t, "expected new batch after group interval but received none")
 
-		case batch := <-alertsCh:
-			lastCurMtx.Lock()
-			s := time.Since(last)
-			lastCurMtx.Unlock()
-			if s < opts.GroupInterval {
-				t.Fatalf("received batch too early after %v", s)
-			}
+		case notification := <-alertsCh:
+			assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
+			lastNotificationAt = notification.notifiedAt
+			batch := notification.alerts
 			exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a3})
 			sort.Sort(batch)
 
-			if !reflect.DeepEqual(batch, exp) {
-				t.Fatalf("expected alerts %v but got %v", exp, batch)
-			}
+			require.Equal(t, exp, batch)
 		}
 	}
 
@@ -220,15 +215,15 @@ func TestAggrGroup(t *testing.T) {
 	// a2 lies way in the past so the initial group_wait should be skipped.
 	select {
 	case <-time.After(opts.GroupWait / 2):
-		t.Fatalf("expected immediate alert but received none")
+		require.FailNow(t, "expected immediate alert but received none")
 
-	case batch := <-alertsCh:
+	case notification := <-alertsCh:
+		lastNotificationAt = notification.notifiedAt
+		batch := notification.alerts
 		exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a2})
 		sort.Sort(batch)
 
-		if !reflect.DeepEqual(batch, exp) {
-			t.Fatalf("expected alerts %v but got %v", exp, batch)
-		}
+		require.Equal(t, exp, batch)
 	}
 
 	for i := 0; i < 3; i++ {
@@ -237,21 +232,16 @@ func TestAggrGroup(t *testing.T) {
 
 		select {
 		case <-time.After(2 * opts.GroupInterval):
-			t.Fatalf("expected new batch after group interval but received none")
+			require.FailNow(t, "expected new batch after group interval but received none")
 
-		case batch := <-alertsCh:
-			lastCurMtx.Lock()
-			s := time.Since(last)
-			lastCurMtx.Unlock()
-			if s < opts.GroupInterval {
-				t.Fatalf("received batch too early after %v", s)
-			}
+		case notification := <-alertsCh:
+			assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
+			lastNotificationAt = notification.notifiedAt
+			batch := notification.alerts
 			exp := removeEndsAt(alertmanagertypes.AlertSlice{a1, a2, a3})
 			sort.Sort(batch)
 
-			if !reflect.DeepEqual(batch, exp) {
-				t.Fatalf("expected alerts %v but got %v", exp, batch)
-			}
+			require.Equal(t, exp, batch)
 		}
 	}
 
@@ -263,19 +253,14 @@ func TestAggrGroup(t *testing.T) {
 
 	select {
 	case <-time.After(2 * opts.GroupInterval):
-		t.Fatalf("expected new batch after group interval but received none")
-	case batch := <-alertsCh:
-		lastCurMtx.Lock()
-		s := time.Since(last)
-		lastCurMtx.Unlock()
-		if s < opts.GroupInterval {
-			t.Fatalf("received batch too early after %v", s)
-		}
+		require.FailNow(t, "expected new batch after group interval but received none")
+	case notification := <-alertsCh:
+		assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
+		lastNotificationAt = notification.notifiedAt
+		batch := notification.alerts
 		sort.Sort(batch)
 
-		if !reflect.DeepEqual(batch, exp) {
-			t.Fatalf("expected alerts %v but got %v", exp, batch)
-		}
+		require.Equal(t, exp, batch)
 	}
 
 	// Resolve all remaining alerts, they should be removed after the next batch was sent.
@@ -289,24 +274,16 @@ func TestAggrGroup(t *testing.T) {
 
 	select {
 	case <-time.After(2 * opts.GroupInterval):
-		t.Fatalf("expected new batch after group interval but received none")
+		require.FailNow(t, "expected new batch after group interval but received none")
 
-	case batch := <-alertsCh:
-		lastCurMtx.Lock()
-		s := time.Since(last)
-		lastCurMtx.Unlock()
-		if s < opts.GroupInterval {
-			t.Fatalf("received batch too early after %v", s)
-		}
+	case notification := <-alertsCh:
+		assertNotifiedAfter(lastNotificationAt, notification.notifiedAt, opts.GroupInterval)
+		batch := notification.alerts
 		sort.Sort(batch)
 
-		if !reflect.DeepEqual(batch, resolved) {
-			t.Fatalf("expected alerts %v but got %v", resolved, batch)
-		}
+		require.Equal(t, resolved, batch)
 
-		if !ag.empty() {
-			t.Fatalf("Expected aggregation group to be empty after resolving alerts: %v", ag)
-		}
+		require.Eventually(t, ag.empty, 2*opts.GroupInterval, 10*time.Millisecond, "expected aggregation group to be empty after resolving alerts: %v", ag)
 	}
 
 	ag.stop()

pkg/errors/suggestions.go

@@ -9,13 +9,12 @@ import (
 // NewValidReferences / NewSuggestionsOnLevenshteinDistance to phrase the
 // "valid <noun> are ..." list consistently across the codebase.
 const (
-	NounFields       = "fields"
-	NounKeys         = "keys"
-	NounServices     = "services"
-	NounQueryTypes   = "query types"
-	NounRequestTypes = "request types"
-	NounSignals      = "signals"
-	NounReferences   = "references"
+	NounFields     = "fields"
+	NounKeys       = "keys"
+	NounServices   = "services"
+	NounQueryTypes = "query types"
+	NounSignals    = "signals"
+	NounReferences = "references"
 )
 
 const (

pkg/types/querybuildertypes/querybuildertypesv5/formula.go

@@ -118,10 +118,6 @@ func (f QueryBuilderFormula) Validate() error {
 		)
 	}
 
-	if f.Limit < 0 {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be non-negative, got %d", f.Limit)
-	}
-
 	// Validate functions if present
 	for i, fn := range f.Functions {
 		if err := fn.Validate(); err != nil {

pkg/types/querybuildertypes/querybuildertypesv5/request_type.go

@@ -23,8 +23,7 @@ func (r *RequestType) UnmarshalJSON(data []byte) error {
 		*r = v
 		return nil
 	default:
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown request type %q", s).
-			WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown request type %q; allowed values: %s", s, "`scalar`, `time_series`, `raw`, `raw_stream`, `trace`, `distribution`, `heatmap`")
 	}
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -499,18 +499,6 @@ func (q *QueryBuilderQuery[T]) validateOrderBy(cfg validationConfig) error {
 		return q.validateOrderByForAggregation()
 	}
 
-	// a raw order key resolves like a select field, so only an empty one is
-	// invalid before resolution
-	for i, order := range q.Order {
-		if order.Key.Name == "" {
-			orderID := fmt.Sprintf("order by clause #%d", i+1)
-			if q.Name != "" {
-				orderID = fmt.Sprintf("order by clause #%d in query '%s'", i+1, q.Name)
-			}
-			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid empty key name for %s", orderID)
-		}
-	}
-
 	return nil
 }
 
@@ -598,10 +586,19 @@ func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
 		return err
 	}
 
-	if err := r.validateRequestType(); err != nil {
-		return err
+	// Validate request type
+	switch r.RequestType {
+	case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace, RequestTypeTimeSeries, RequestTypeScalar, RequestTypeHeatmap, RequestTypeDistribution:
+		opts = append(opts, GetValidationOptions(r.RequestType)...)
+	default:
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"invalid request type: %s",
+			r.RequestType,
+		).WithAdditional(
+			"Valid request types are: raw, timeseries, scalar, heatmap, distribution",
+		)
 	}
-	opts = append(opts, GetValidationOptions(r.RequestType)...)
 
 	if err := r.validateBucketCounting(); err != nil {
 		return err
@@ -642,10 +639,14 @@ func (r *QueryRangeRequest) ValidateRequestScope() ([]ValidationOption, error) {
 		return nil, err
 	}
 
-	if err := r.validateRequestType(); err != nil {
-		return nil, err
+	var opts []ValidationOption
+	switch r.RequestType {
+	case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace, RequestTypeTimeSeries, RequestTypeScalar, RequestTypeHeatmap, RequestTypeDistribution:
+		opts = GetValidationOptions(r.RequestType)
+	default:
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid request type: %s", r.RequestType).
+			WithAdditional("Valid request types are: raw, timeseries, scalar, heatmap, distribution")
 	}
-	opts := GetValidationOptions(r.RequestType)
 
 	if err := r.validateBucketCounting(); err != nil {
 		return nil, err
@@ -702,20 +703,6 @@ func (e QueryEnvelope) Validate(opts ...ValidationOption) error {
 	return validateQueryEnvelope(e, opts...)
 }
 
-// Enum is the one list of request types a query can use, so the spec and the
-// suggestions agree with what is accepted.
-func (r *QueryRangeRequest) validateRequestType() error {
-	if slices.Contains(RequestType{}.Enum(), any(r.RequestType)) {
-		return nil
-	}
-	if r.RequestType == RequestTypeUnknown {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "requestType is required").
-			WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
-	}
-	return errors.NewInvalidInputf(errors.CodeInvalidInput, "request type %q is not supported", r.RequestType.StringValue()).
-		WithSuggestions(errors.NewValidReferences(errors.NounRequestTypes, RequestType{}.Enum()...))
-}
-
 func (r *QueryRangeRequest) validateTimeRange() error {
 	if r.RequestType == RequestTypeRawStream {
 		return nil

pkg/types/querybuildertypes/querybuildertypesv5/validation_test.go

@@ -1,15 +1,9 @@
 package querybuildertypesv5
 
 import (
-	"encoding/json"
-	"fmt"
 	"strings"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -527,28 +521,6 @@ func TestQueryRangeRequest_ValidateCompositeQuery(t *testing.T) {
 			wantErr: true,
 			errMsg:  "expression cannot be blank",
 		},
-		{
-			name: "Formula_NegativeLimit_Rejected",
-			request: QueryRangeRequest{
-				Start:       1640995200000,
-				End:         1640998800000,
-				RequestType: RequestTypeTimeSeries,
-				CompositeQuery: CompositeQuery{
-					Queries: []QueryEnvelope{
-						{
-							Type: QueryTypeFormula,
-							Spec: QueryBuilderFormula{
-								Name:       "F1",
-								Expression: "A * 2",
-								Limit:      -5,
-							},
-						},
-					},
-				},
-			},
-			wantErr: true,
-			errMsg:  "invalid formula 'F1': limit must be non-negative, got -5",
-		},
 		{
 			name: "promql with empty query should return error",
 			request: QueryRangeRequest{
@@ -724,31 +696,6 @@ func TestQueryRangeRequest_ValidateCompositeQuery(t *testing.T) {
 			wantErr: true,
 			errMsg:  "raw request type is not supported for metric queries",
 		},
-		{
-			name: "Raw_EmptyOrderKey_Rejected",
-			request: QueryRangeRequest{
-				Start:       1640995200000,
-				End:         1640998800000,
-				RequestType: RequestTypeRaw,
-				CompositeQuery: CompositeQuery{
-					Queries: []QueryEnvelope{
-						{
-							Type: QueryTypeBuilder,
-							Spec: QueryBuilderQuery[LogAggregation]{
-								Name:   "B",
-								Signal: telemetrytypes.SignalLogs,
-								Order: []OrderBy{
-									{Key: OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"}}, Direction: OrderDirectionDesc},
-									{Key: OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: ""}}, Direction: OrderDirectionAsc},
-								},
-							},
-						},
-					},
-				},
-			},
-			wantErr: true,
-			errMsg:  "invalid query 'B': invalid empty key name for order by clause #2 in query 'B'",
-		},
 		{
 			name: "timeseries request with group by timestamp should return error",
 			request: QueryRangeRequest{
@@ -1460,44 +1407,6 @@ func TestRequestType_IsAggregation(t *testing.T) {
 	}
 }
 
-func TestQueryRangeRequest_ValidateRequestType(t *testing.T) {
-	validRequestTypes := []string{"valid request types are `scalar`, `time_series`, `raw`, `raw_stream`, `trace`, `heatmap`, `distribution`"}
-
-	testCases := []struct {
-		name        string
-		requestType string
-		message     string
-	}{
-		{
-			name:        "Missing_Required",
-			requestType: "",
-			message:     "requestType is required",
-		},
-		{
-			name:        "Timeseries_Unknown",
-			requestType: `"requestType": "timeseries",`,
-			message:     `unknown request type "timeseries"`,
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			body := fmt.Sprintf(`{"schemaVersion": "v1", "start": 1640995200000, "end": 1640998800000, %s "compositeQuery": {"queries": [{"type": "builder_query", "spec": {"name": "A", "signal": "logs"}}]}}`, testCase.requestType)
-
-			var request QueryRangeRequest
-			err := json.Unmarshal([]byte(body), &request)
-			if err == nil {
-				err = request.Validate()
-			}
-			require.Error(t, err)
-
-			got := errors.AsJSON(err)
-			assert.Equal(t, testCase.message, got.Message)
-			assert.Equal(t, validRequestTypes, got.Suggestions)
-		})
-	}
-}
-
 func TestNonAggregationFieldsSkipped(t *testing.T) {
 	// Fields that only apply to aggregation queries (groupBy, having, aggregations)
 	// should be silently skipped for non-aggregation request types.

tests/integration/tests/queriercommon/13_cache_running_diff.py

@@ -0,0 +1,69 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+from uuid import uuid4
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.metrics import Metrics
+from fixtures.querier import (
+    build_builder_query,
+    build_function,
+    get_all_series,
+    index_series_by_label,
+    make_query_request,
+)
+
+MINUTE = timedelta(minutes=1)
+
+
+def test_running_diff_full_hit_matches_no_cache(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    """
+    Setup:
+    A gauge that grows by 10 every minute over four minutes.
+
+    Tests:
+    runningDiff over the last three of those minutes, requested twice. The
+    metrics statement reads one step before the window so the first
+    interval has a difference. The second request, a full cache hit, must
+    return the same three differences as the request with noCache.
+    """
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # the cache outlives the run, so a fixed name would serve the previous
+    # run's points back to this one
+    metric_name = f"cache_running_diff_{uuid4().hex[:8]}"
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "svc-a"},
+                timestamp=now - (14 - i) * MINUTE,
+                value=100.0 + 10 * i,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for i in range(4)
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_builder_query("A", metric_name, "avg", "avg", group_by=["service"], functions=[build_function("runningDiff")])
+    start_ms = int((now - 13 * MINUTE).timestamp() * 1000)
+    end_ms = int((now - 10 * MINUTE).timestamp() * 1000)
+
+    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+    assert warm.status_code == HTTPStatus.OK, warm.text
+    cached = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+    assert cached.status_code == HTTPStatus.OK, cached.text
+    fresh = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=True)
+    assert fresh.status_code == HTTPStatus.OK, fresh.text
+
+    fresh_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(fresh.json(), "A"), "service")["svc-a"]["values"]}
+    assert sorted(fresh_points.values()) == [10, 10, 10], fresh_points
+
+    cached_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(cached.json(), "A"), "service")["svc-a"]["values"]}
+    assert cached_points == fresh_points