fix: add cache fixes

pkg/querier/bucket_cache.go

@@ -55,6 +55,9 @@ func (bc *bucketCache) GetMissRanges(
 	// Get query window
 	startMs, endMs := q.Window()
 
+	stepMs := uint64(step.Milliseconds())
+	startOffsetMs := calculateStartOffset(q, startMs, stepMs)
+
 	bc.logger.DebugContext(ctx, "getting miss ranges", slog.String("fingerprint", q.Fingerprint()), slog.Uint64("start", startMs), slog.Uint64("end", endMs))
 
 	// Generate cache key
@@ -74,11 +77,8 @@ func (bc *bucketCache) GetMissRanges(
 		return nil, missing
 	}
 
-	// Extract step interval if this is a builder query
-	stepMs := uint64(step.Milliseconds())
-
 	// Find missing ranges with step alignment
-	missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs)
+	missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs, startOffsetMs)
 	bc.logger.DebugContext(ctx, "missing ranges", slog.Any("missing", missing), slog.Uint64("step", stepMs))
 
 	// If no cached data overlaps with requested range, return empty result
@@ -95,8 +95,8 @@ func (bc *bucketCache) GetMissRanges(
 	// Merge buckets into a single result
 	mergedResult := bc.mergeBuckets(ctx, relevantBuckets, data.Warnings)
 
-	// Filter the merged result to only include values within the requested time range
-	mergedResult = bc.filterResultToTimeRange(mergedResult, startMs, endMs)
+	_, isPromQL := q.(*promqlQuery)
+	mergedResult = bc.filterResultToTimeRange(mergedResult, startMs, endMs, stepMs, isPromQL)
 
 	return mergedResult, missing
 }
@@ -106,6 +106,9 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 	// Get query window
 	startMs, endMs := q.Window()
 
+	stepMs := uint64(step.Milliseconds())
+	startOffsetMs := calculateStartOffset(q, startMs, stepMs)
+
 	// Calculate the flux boundary - data after this point should not be cached
 	currentMs := uint64(time.Now().UnixMilli())
 	fluxBoundary := currentMs - uint64(bc.fluxInterval.Milliseconds())
@@ -146,19 +149,14 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 
 	// Adjust start and end times to only cache complete intervals
 	cachableStartMs := startMs
-	stepMs := uint64(step.Milliseconds())
 
 	// If we have a step interval, adjust boundaries to only cache complete intervals
 	if stepMs > 0 {
 		// If start is not aligned, round up to next step boundary (first complete interval)
-		if startMs%stepMs != 0 {
-			cachableStartMs = ((startMs / stepMs) + 1) * stepMs
-		}
+		cachableStartMs = alignUpToStep(startMs, stepMs, startOffsetMs)
 
 		// If end is not aligned, round down to previous step boundary (last complete interval)
-		if cachableEndMs%stepMs != 0 {
-			cachableEndMs = (cachableEndMs / stepMs) * stepMs
-		}
+		cachableEndMs = alignDownToStep(cachableEndMs, stepMs, startOffsetMs)
 
 		// If after adjustment we have no complete intervals, don't cache
 		if cachableStartMs >= cachableEndMs {
@@ -206,8 +204,9 @@ func (bc *bucketCache) generateCacheKey(q qbtypes.Query) string {
 	return fmt.Sprintf("v5:query:%s", fingerprint)
 }
 
-// findMissingRangesWithStep identifies time ranges not covered by cached buckets with step alignment.
-func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64) []*qbtypes.TimeRange {
+// findMissingRangesWithStep identifies time ranges not covered by cached buckets
+// with step alignment. Boundaries are whole steps from startOffsetMs.
+func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64, startOffsetMs uint64) []*qbtypes.TimeRange {
 	// When step is 0 or window is too small to be cached, use simple algorithm
 	if stepMs == 0 || (startMs+stepMs) > endMs {
 		return bc.findMissingRangesBasic(buckets, startMs, endMs)
@@ -220,8 +219,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		currentMs := startMs
 
 		// Check if start is not aligned - add partial window
-		if startMs%stepMs != 0 {
-			nextAggStart := startMs - (startMs % stepMs) + stepMs
+		if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
 			missing = append(missing, &qbtypes.TimeRange{
 				From: startMs,
 				To:   min(nextAggStart, endMs),
@@ -267,8 +265,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 	currentMs := startMs
 
 	// Check if start is not aligned - add partial window
-	if startMs%stepMs != 0 {
-		nextAggStart := startMs - (startMs % stepMs) + stepMs
+	if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
 		missing = append(missing, &qbtypes.TimeRange{
 			From: startMs,
 			To:   min(nextAggStart, endMs),
@@ -287,11 +284,7 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		}
 
 		// Align bucket boundaries to step intervals
-		alignedBucketStart := bucket.StartMs
-		if bucket.StartMs%stepMs != 0 {
-			// Round up to next step boundary
-			alignedBucketStart = bucket.StartMs - (bucket.StartMs % stepMs) + stepMs
-		}
+		alignedBucketStart := alignUpToStep(bucket.StartMs, stepMs, startOffsetMs)
 
 		// Add gap before this bucket if needed
 		if currentMs < alignedBucketStart && currentMs < endMs {
@@ -304,9 +297,12 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 		// Update current position to the end of this bucket
 		// But ensure it's aligned to step boundary
 		bucketEnd := min(bucket.EndMs, endMs)
-		if bucketEnd%stepMs != 0 && bucketEnd < endMs {
+		// The step the window ends inside reaches past it, so that stretch is
+		// missing however far the bucket runs.
+		bucketEnd = min(bucketEnd, alignDownToStep(endMs, stepMs, startOffsetMs))
+		if bucketEnd < endMs {
 			// Round down to step boundary
-			bucketEnd = bucketEnd - (bucketEnd % stepMs)
+			bucketEnd = alignDownToStep(bucketEnd, stepMs, startOffsetMs)
 		}
 		currentMs = max(currentMs, bucketEnd)
 	}
@@ -323,6 +319,42 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
 	return missing
 }
 
+// calculateStartOffset returns how far into a step a query's values sit. Only
+// promql reports at the window start and every step after it; the rest report
+// on absolute step boundaries.
+func calculateStartOffset(q qbtypes.Query, startMs, stepMs uint64) uint64 {
+	if _, isPromQL := q.(*promqlQuery); !isPromQL || stepMs == 0 {
+		return 0
+	}
+	return startMs % stepMs
+}
+
+// With a 5m step and no offset the times seen by a query are 10:00, 10:05, 10:10. So 10:07
+// is at an offset of 2m, and 10:05 is at 0.
+//
+// With a 1m step and a 30s offset the times seen are 10:00:30, 10:01:30, 10:02:30. So 10:01:00
+// is at an offset of 30s.
+func calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	if stepMs == 0 {
+		return 0
+	}
+	return ((timestampMs % stepMs) + stepMs - startOffsetMs%stepMs) % stepMs
+}
+
+// alignUpToStep returns the first time seen by a query at or after timestampMs.
+func alignUpToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	offset := calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
+	if offset == 0 {
+		return timestampMs
+	}
+	return timestampMs - offset + stepMs
+}
+
+// alignDownToStep returns the last time seen by a query at or before timestampMs.
+func alignDownToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
+	return timestampMs - calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
+}
+
 // findMissingRangesBasic is the simple algorithm without step alignment.
 func (bc *bucketCache) findMissingRangesBasic(buckets []*qbtypes.CachedBucket, startMs, endMs uint64) []*qbtypes.TimeRange {
 	// Check if already sorted before sorting
@@ -760,11 +792,23 @@ func max(a, b uint64) uint64 {
 }
 
 // filterResultToTimeRange filters the result to only include values within the requested time range.
-func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs, endMs uint64) *qbtypes.Result {
+func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs, endMs, stepMs uint64, isPromQL bool) *qbtypes.Result {
 	if result == nil || result.Value == nil {
 		return result
 	}
 
+	maxTimestampMs := endMs
+	// A promql value at T is the query evaluated at T, so T == endMs is inside the
+	// requested range. For every other query type the value at T aggregates
+	// [T, T+stepMs), which the requested range contains only when T <= endMs-stepMs.
+	if !isPromQL {
+		if stepMs > 0 {
+			maxTimestampMs = endMs - stepMs
+		} else {
+			maxTimestampMs = endMs - 1
+		}
+	}
+
 	switch result.Type {
 	case qbtypes.RequestTypeTimeSeries:
 		if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
@@ -789,7 +833,7 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs,
 					// Filter values to only include those within the requested time range
 					for _, value := range series.Values {
 						timestampMs := uint64(value.Timestamp)
-						if timestampMs >= startMs && timestampMs < endMs {
+						if timestampMs >= startMs && timestampMs <= maxTimestampMs {
 							filteredSeries.Values = append(filteredSeries.Values, value)
 						}
 					}

pkg/querier/bucket_cache_bench_test.go

@@ -201,7 +201,7 @@ func BenchmarkBucketCache_FindMissingRangesWithStep(b *testing.B) {
 			b.ReportAllocs()
 
 			for i := 0; i < b.N; i++ {
-				missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs)
+				missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs, 0)
 				_ = missing
 			}
 		})
@@ -327,7 +327,7 @@ func BenchmarkBucketCache_FilterResultToTimeRange(b *testing.B) {
 			b.ReportAllocs()
 
 			for i := 0; i < b.N; i++ {
-				filtered := bc.filterResultToTimeRange(result, startMs, endMs)
+				filtered := bc.filterResultToTimeRange(result, startMs, endMs, 0, true)
 				_ = filtered
 			}
 		})

pkg/querier/bucket_cache_test.go

@@ -3,6 +3,7 @@ package querier
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"testing"
 	"time"
 
@@ -529,7 +530,7 @@ func TestBucketCache_FindMissingRanges_EdgeCases(t *testing.T) {
 	}
 
 	// Query range that spans all buckets
-	missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500)
+	missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500, 0)
 
 	// Expected missing ranges: 500-1000, 2000-2500, 4000-5000, 6000-6500
 	assert.Len(t, missing, 4)
@@ -1069,8 +1070,11 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
 	// Get cached data - should be filtered to requested range
 	cached, missing := bc.GetMissRanges(ctx, orgID, query2, qbtypes.Step{Duration: 1000 * time.Millisecond})
 
-	// Should have no missing ranges
-	assert.Len(t, missing, 0)
+	// The value at 3000 stands for the whole step to 4000, which reaches past the
+	// window, so it is left to be recomputed as a partial rather than served.
+	require.Len(t, missing, 1)
+	assert.Equal(t, uint64(3000), missing[0].From)
+	assert.Equal(t, uint64(3500), missing[0].To)
 	assert.NotNil(t, cached)
 
 	// Verify the cached result only contains values within the requested range
@@ -1080,29 +1084,77 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
 	require.Len(t, tsData.Aggregations[0].Series, 1)
 
 	series := tsData.Aggregations[0].Series[0]
-	assert.Len(t, series.Values, 2) // Only values at 2000 and 3000 should be included
+	require.Len(t, series.Values, 1)
 
 	// Verify the exact values
 	assert.Equal(t, int64(2000), series.Values[0].Timestamp)
 	assert.Equal(t, float64(20), series.Values[0].Value)
-	assert.Equal(t, int64(3000), series.Values[1].Timestamp)
-	assert.Equal(t, float64(30), series.Values[1].Value)
 
 	// Value at 1000 should not be included (before requested range)
 	// Value at 4000 should not be included (after requested range)
 }
 
+// A promql value is the query evaluated at a single moment rather than over a
+// span, so the one at the window's end belongs to it and has to survive caching.
+func TestBucketCache_PromQLKeepsTheValueAtTheWindowEnd(t *testing.T) {
+	bc := createTestBucketCache(t)
+	ctx := context.Background()
+	orgID := valuer.UUID{}
+	step := qbtypes.Step{Duration: time.Minute}
+
+	query := &promqlQuery{
+		logger:      slog.Default(),
+		query:       qbtypes.PromQuery{Query: "up", Step: step},
+		tr:          qbtypes.TimeRange{From: 600_000, To: 780_000},
+		requestType: qbtypes.RequestTypeTimeSeries,
+	}
+
+	bc.Put(ctx, orgID, query, step, &qbtypes.Result{
+		Type: qbtypes.RequestTypeTimeSeries,
+		Value: &qbtypes.TimeSeriesData{
+			QueryName: "A",
+			Aggregations: []*qbtypes.AggregationBucket{{
+				Series: []*qbtypes.TimeSeries{{
+					Values: []*qbtypes.TimeSeriesValue{
+						{Timestamp: 600_000, Value: 1},
+						{Timestamp: 660_000, Value: 2},
+						{Timestamp: 720_000, Value: 3},
+						{Timestamp: 780_000, Value: 4},
+					},
+				}},
+			}},
+		},
+	})
+	time.Sleep(10 * time.Millisecond)
+
+	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
+	assert.Empty(t, missing)
+	require.NotNil(t, cached)
+
+	tsData, ok := cached.Value.(*qbtypes.TimeSeriesData)
+	require.True(t, ok)
+	require.Len(t, tsData.Aggregations, 1)
+	require.Len(t, tsData.Aggregations[0].Series, 1)
+
+	timestamps := []int64{}
+	for _, value := range tsData.Aggregations[0].Series[0].Values {
+		timestamps = append(timestamps, value.Timestamp)
+	}
+	assert.Equal(t, []int64{600_000, 660_000, 720_000, 780_000}, timestamps)
+}
+
 func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 	bc := createTestBucketCache(t)
 
 	tests := []struct {
-		name         string
-		buckets      []*qbtypes.CachedBucket
-		startMs      uint64
-		endMs        uint64
-		stepMs       uint64
-		expectedMiss []*qbtypes.TimeRange
-		description  string
+		name          string
+		buckets       []*qbtypes.CachedBucket
+		startMs       uint64
+		endMs         uint64
+		stepMs        uint64
+		startOffsetMs uint64
+		expectedMiss  []*qbtypes.TimeRange
+		description   string
 	}{
 		{
 			name:    "start_not_aligned_to_step",
@@ -1152,6 +1204,32 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 			},
 			description: "Window smaller than step should use basic algorithm",
 		},
+		{
+			name:          "start_aligned_to_its_own_offset",
+			buckets:       []*qbtypes.CachedBucket{},
+			startMs:       1500,
+			endMs:         5000,
+			stepMs:        1000,
+			startOffsetMs: 500,
+			expectedMiss: []*qbtypes.TimeRange{
+				{From: 1500, To: 5000},
+			},
+			description: "A query reporting every 1000ms from 1500 needs no partial window at its own start",
+		},
+		{
+			name: "gap_lands_on_the_offset",
+			buckets: []*qbtypes.CachedBucket{
+				{StartMs: 1500, EndMs: 3500},
+			},
+			startMs:       1500,
+			endMs:         5500,
+			stepMs:        1000,
+			startOffsetMs: 500,
+			expectedMiss: []*qbtypes.TimeRange{
+				{From: 3500, To: 5500},
+			},
+			description: "The refetched range starts where the cached one ends, on an instant the query reports at",
+		},
 		{
 			name:    "zero_step_uses_basic_algorithm",
 			buckets: []*qbtypes.CachedBucket{},
@@ -1168,7 +1246,7 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// Mock current time for flux boundary tests
-			result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs)
+			result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs, tt.startOffsetMs)
 
 			// Compare lengths first
 			assert.Len(t, result, len(tt.expectedMiss), tt.description)

pkg/querier/promql_query.go

@@ -170,6 +170,12 @@ func (q *promqlQuery) Fingerprint() string {
 		q.query.Step.String(),
 	}
 
+	// Two windows a fraction of a step apart describe different instants, so
+	// they must not share an entry.
+	if stepMs := uint64(q.query.Step.Milliseconds()); stepMs > 0 && q.tr.From%stepMs != 0 {
+		parts = append(parts, fmt.Sprintf("offset=%d", q.tr.From%stepMs))
+	}
+
 	return strings.Join(parts, "&")
 }
 

pkg/querier/promql_query_test.go

@@ -461,6 +461,37 @@ func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
 	assert.Empty(t, q.Fingerprint())
 }
 
+// promql reports at the window start and every step after it, so a window
+// starting later inside the step describes instants the earlier one never does.
+func TestFingerprintSeparatesWindowsInsideAStep(t *testing.T) {
+	minuteStep := qbv5.Step{Duration: time.Minute}
+
+	onTheMinute := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 600_000, To: 1_200_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	halfAStepLater := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 630_000, To: 1_230_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	aWholeMinuteLater := (&promqlQuery{
+		logger:      slog.Default(),
+		query:       qbv5.PromQuery{Query: "up", Step: minuteStep},
+		tr:          qbv5.TimeRange{From: 900_000, To: 1_500_000},
+		requestType: qbv5.RequestTypeTimeSeries,
+	}).Fingerprint()
+
+	require.NotEmpty(t, onTheMinute)
+	assert.NotEqual(t, onTheMinute, halfAStepLater, "windows half a step apart share no instants")
+	assert.Equal(t, onTheMinute, aWholeMinuteLater, "windows whole steps apart report at the same instants")
+}
+
 func TestToResultDropsNonFiniteValues(t *testing.T) {
 	tests := []struct {
 		description        string

tests/integration/tests/promqlconformance/03_nonfinite_cache.py

@@ -58,8 +58,7 @@ def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
     assert set(first["active_job"].values()) == {25.0}, sorted(set(first["active_job"].values()))
     assert len(first["active_job"]) == expected_points, f"expected {expected_points} points, got {len(first['active_job'])}"
 
-    # The cached read excludes end_ms, the one legitimate difference.
+    # Both reads must agree exactly, including the point promql reports at end_ms.
     assert set(second) == set(first), sorted(second)
     for job_name, points in first.items():
-        expected = {ts: value for ts, value in points.items() if ts < end_ms}
-        assert second[job_name] == expected, f"{job_name}: got {len(second[job_name])} of {len(expected)} points"
+        assert second[job_name] == points, f"{job_name}: got {len(second[job_name])} of {len(points)} points"

tests/integration/tests/queriermetrics/15_cache.py

@@ -0,0 +1,325 @@
+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 (
+    assert_results_equal,
+    build_builder_query,
+    get_series_values,
+    make_query_request,
+)
+
+MINUTE_MS = 60_000
+
+
+def test_builder_shortening_the_time_range_at_the_end(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    # the cache outlives the run, so a fixed name would serve the previous run's
+    # points back to this one
+    metric_name = f"cache_end_shortened_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a multiple of the 5m step makes the base query span
+    # two whole steps, so both its points are complete
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms_base_query = start_time_ms + 10 * MINUTE_MS
+    end_time_ms_shortened_query = start_time_ms + 7 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
+
+    # the 5m step splits the ten minutes into two points, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. The second changes partway through,
+    # 256 until minute 7 and then 4096, so ending the range at minute 7 has to
+    # reach a different value than ending it at minute 10
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(16, 16, 16, 16, 16, 256, 256, 4096, 4096, 4096)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+    points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["value"], point.get("partial", False)) for point in points]
+    assert returned_points == [(16, False), (4096, False)]
+
+    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened end")
+
+    # the shortened end reaches only minutes 5-6 of the second point, so it comes
+    # back as 256 and partial, where the cached one spans all five minutes
+    for label, response in (("from cache", from_cache), ("uncached", uncached)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["value"], point.get("partial", False)) for point in points]
+        assert returned_points == [(16, False), (256, True)], label
+
+
+def test_builder_shortening_the_time_range_at_the_start(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    metric_name = f"cache_start_shortened_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a multiple of the 5m step makes the base query span
+    # two whole steps, so both its points are complete
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
+    start_time_ms_base_query = int(start_time.timestamp() * 1000)
+    start_time_ms_shortened_query = start_time_ms_base_query + 3 * MINUTE_MS
+    end_time_ms = start_time_ms_base_query + 10 * MINUTE_MS
+
+    query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
+
+    # the 5m step splits the ten minutes into two points, each the max over its
+    # own step: minutes 0-4 and minutes 5-9. Only minute 0 holds 65536, so a first
+    # point reaching it says the whole step was read even though the shortened
+    # range opens at minute 3
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(65536, 16, 16, 16, 16, 4096, 4096, 4096, 4096, 4096)[minute],
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(10)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False)
+    assert base_query.status_code == HTTPStatus.OK, base_query.text
+    points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["value"], point.get("partial", False)) for point in points]
+    assert returned_points == [(65536, False), (4096, False)]
+
+    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False)
+    assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened start")
+
+    # starting inside the first point's step flags that point partial without
+    # clipping its value, which still covers the whole step and so reaches the
+    # 65536 at minute 0
+    for label, response in (("from cache", from_cache), ("uncached", uncached)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["value"], point.get("partial", False)) for point in points]
+        assert returned_points == [(65536, True), (4096, False)], label
+
+
+def test_promql_running_the_same_query_twice(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    metric_name = f"cache_repeat_total_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+    end_time_ms = start_time_ms + 2 * MINUTE_MS
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"sum(increase({metric_name}[2m]))", "step": 60}}]
+
+    # the counter opens a minute before the query so its first point has something
+    # to increase over, and starts far above its own rise across the range, below
+    # which increase clips its back-extrapolation at the counter's zero point. It
+    # rises by a different amount each minute, so every point is its own number
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=(1000, 1010, 1030, 1060, 1100)[minute + 1],
+                temporality="Cumulative",
+                type_="Sum",
+                is_monotonic=True,
+            )
+            for minute in range(-1, 4)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    assert first.status_code == HTTPStatus.OK, first.text
+
+    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
+    assert second.status_code == HTTPStatus.OK, second.text
+
+    assert_results_equal(first.json(), second.json(), "A", "the same query twice")
+
+    # promql reports a point at the instant the range closes, and the second run,
+    # answered out of what the first one cached, has to keep it
+    for run, response in (("first", first), ("second", second)):
+        points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"]) for point in points]
+        ## at each timestamp t, promql looks at points in (t-2minutes, t].
+        assert returned_points == [
+            (start_time_ms, 20),  # t = 0, points taken 1000, 1010. hence diff over 1m is 10, extrapolated to 20.
+            (start_time_ms + MINUTE_MS, 40),  # t = 1m, points taken 1010, 1030. hence diff over 1m is 20, extrapolated to 40.
+            (end_time_ms, 60),  # t = 2m, points taken 1030, 1060. hence diff over 1m is 30, extrapolated to 60.
+        ], f"{run} run"
+
+
+def test_promql_shifting_the_time_range(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    metric_name = f"cache_shift_gauge_{uuid4().hex[:8]}"
+
+    # 40 minutes back clears the flux interval, which holds recent data out of
+    # the cache. Flooring to a whole minute is what makes the first query aligned
+    # to its 1m step, and the unaligned one half a step off it
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
+    aligned_start_time_ms = int(start_time.timestamp() * 1000)
+    aligned_end_time_ms = aligned_start_time_ms + 3 * MINUTE_MS
+    unaligned_start_time_ms = aligned_start_time_ms + MINUTE_MS // 2
+    unaligned_end_time_ms = aligned_end_time_ms + MINUTE_MS // 2
+
+    query = [{"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric_name}[2m])", "step": 60}}]
+
+    # a sample every 30s, rising by 100 each time. The two queries report 30s
+    # apart, so they land on different samples and share no value between them
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(seconds=30 * half_minute),
+                value=100 * (half_minute + 4),
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for half_minute in range(-3, 8)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False)
+    assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
+
+    # what the cache now holds, and what the unaligned query must not be served
+    points = sorted(get_series_values(aligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
+    returned_points = [(point["timestamp"], point["value"]) for point in points]
+    ## at each timestamp t, promql takes the highest sample in (t-2minutes, t],
+    ## which is the one at t itself since the gauge only rises.
+    assert returned_points == [
+        (aligned_start_time_ms, 400),  # t = 0
+        (aligned_start_time_ms + MINUTE_MS, 600),  # t = 1m
+        (aligned_start_time_ms + 2 * MINUTE_MS, 800),  # t = 2m
+        (aligned_end_time_ms, 1000),  # t = 3m
+    ]
+
+    unaligned_and_uncached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=True)
+    assert unaligned_and_uncached.status_code == HTTPStatus.OK, unaligned_and_uncached.text
+
+    # promql reports at the range start plus whole steps, so these points sit 30s
+    # off the cached ones. The first run stores them, the second reads them back
+    for run in ("first", "second"):
+        unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False)
+        assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
+        assert_results_equal(unaligned_and_cached.json(), unaligned_and_uncached.json(), "A", f"unaligned query, {run} run")
+
+        points = sorted(get_series_values(unaligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"]) for point in points]
+        ## every point falls on a sample the aligned run never reported, so being
+        ## served the cached run's answer shows up in the values and not only the
+        ## timestamps.
+        assert returned_points == [
+            (unaligned_start_time_ms, 500),  # t = 30s
+            (unaligned_start_time_ms + MINUTE_MS, 700),  # t = 1m30s
+            (unaligned_start_time_ms + 2 * MINUTE_MS, 900),  # t = 2m30s
+            (unaligned_end_time_ms, 1100),  # t = 3m30s
+        ], f"unaligned query, {run} run"
+
+
+def test_builder_refreshing_a_sliding_time_range(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    metric_name = f"cache_sliding_{uuid4().hex[:8]}"
+
+    # 90 minutes back so even the twentieth refresh closes clear of the flux
+    # interval, which holds recent data out of the cache
+    start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=90)).timestamp()) // 60 * 60, tz=UTC)
+    start_time_ms = int(start_time.timestamp() * 1000)
+
+    query = [build_builder_query("A", metric_name, "max", "max")]
+
+    # the 1m step gives one point per seeded minute, and a value no other minute
+    # carries, so a point stitched in from the wrong range reads as the wrong minute
+    insert_metrics(
+        [
+            Metrics(
+                metric_name=metric_name,
+                labels={"service": "api"},
+                timestamp=start_time + timedelta(minutes=minute),
+                value=1000 + minute,
+                type_="Gauge",
+                is_monotonic=False,
+            )
+            for minute in range(80)
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # a dashboard left open on a one hour range, re-running a minute later each time
+    for refresh in range(20):
+        refresh_start_ms = start_time_ms + refresh * MINUTE_MS
+        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False)
+        assert from_cache.status_code == HTTPStatus.OK, from_cache.text
+
+        # each refresh is stitched out of overlapping cached ranges, so this catches
+        # a point served twice, dropped, or carried over from an earlier refresh
+        points = sorted(get_series_values(from_cache.json(), "A"), key=lambda point: point["timestamp"])
+        returned_points = [(point["timestamp"], point["value"], point.get("partial", False)) for point in points]
+        expected_points = [(start_time_ms + minute * MINUTE_MS, 1000 + minute, False) for minute in range(refresh, refresh + 60)]
+        assert returned_points == expected_points, f"refresh {refresh} did not return the minutes it covers"
+
+    last_refresh_start_ms = start_time_ms + 19 * MINUTE_MS
+    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True)
+    assert uncached.status_code == HTTPStatus.OK, uncached.text
+
+    assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")