test: make new test work on reruns

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_mysql/icon.svg

@@ -1 +0,0 @@
-<svg xmlns="http://www.w3.org/2000/svg" width="24px" height="24px" viewBox="0 0 24 24"><defs><style>.cls-1{fill:#aecbfa;}.cls-1,.cls-2,.cls-3{fill-rule:evenodd;}.cls-2{fill:#669df6;}.cls-3{fill:#4285f4;}</style></defs><title>Icon_24px_SQL_Color</title><g data-name="Product Icons"><g ><polygon class="cls-1" points="4.67 10.44 4.67 13.45 12 17.35 12 14.34 4.67 10.44"/><polygon class="cls-1" points="4.67 15.09 4.67 18.1 12 22 12 18.99 4.67 15.09"/><polygon class="cls-2" points="12 17.35 19.33 13.45 19.33 10.44 12 14.34 12 17.35"/><polygon class="cls-2" points="12 22 19.33 18.1 19.33 15.09 12 18.99 12 22"/><polygon class="cls-3" points="19.33 8.91 19.33 5.9 12 2 12 5.01 19.33 8.91"/><polygon class="cls-2" points="12 2 4.67 5.9 4.67 8.91 12 5.01 12 2"/><polygon class="cls-1" points="4.67 5.87 4.67 8.89 12 12.79 12 9.77 4.67 5.87"/><polygon class="cls-2" points="12 12.79 19.33 8.89 19.33 5.87 12 9.77 12 12.79"/></g></g></svg>

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_mysql/integration.json

@@ -1,136 +0,0 @@
-{
-  "id": "cloudsql_mysql",
-  "title": "GCP Cloud SQL for MySQL",
-  "icon": "file://icon.svg",
-  "overview": "file://overview.md",
-  "supportedSignals": {
-    "metrics": true,
-    "logs": true
-  },
-  "dataCollected": {
-    "metrics": [
-      {
-        "name": "cloudsql.googleapis.com/database/up",
-        "unit": "Count",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/instance_state",
-        "unit": "Count",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/replication/replica_lag",
-        "unit": "Seconds",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/cpu/utilization",
-        "unit": "Percent",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/memory/utilization",
-        "unit": "Percent",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/disk/utilization",
-        "unit": "Percent",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/network/connections",
-        "unit": "Count",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/queries",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/dml_operations_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/threads",
-        "unit": "Count",
-        "type": "Gauge",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/innodb/buffer_pool_reads_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/innodb/buffer_pool_read_requests_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/slow_queries_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/aborted_connects_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/innodb/deadlocks_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/mysql/innodb/row_lock_waits_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/disk/read_ops_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      },
-      {
-        "name": "cloudsql.googleapis.com/database/disk/write_ops_count",
-        "unit": "Count",
-        "type": "Sum",
-        "description": ""
-      }
-    ],
-    "logs": []
-  },
-  "telemetryCollectionStrategy": {
-    "gcp": {}
-  },
-  "assets": {
-    "dashboards": [
-      {
-        "id": "overview",
-        "title": "GCP Cloud SQL for MySQL Overview",
-        "description": "Overview of GCP Cloud SQL for MySQL metrics",
-        "definition": "file://assets/dashboards/overview.json"
-      }
-    ]
-  }
-}

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_mysql/overview.md

@@ -1,3 +0,0 @@
-### Monitor GCP Cloud SQL for MySQL with SigNoz
-
-Collect key GCP Cloud SQL for MySQL metrics and view them with an out of the box dashboard.

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_postgres/assets/dashboards/overview.json

@@ -784,57 +784,40 @@
             {
               "kind": "time_series",
               "spec": {
+                "name": "A",
                 "plugin": {
-                  "kind": "signoz/CompositeQuery",
+                  "kind": "signoz/BuilderQuery",
                   "spec": {
-                    "queries": [
+                    "name": "A",
+                    "signal": "metrics",
+                    "source": "",
+                    "aggregations": [
                       {
-                        "type": "builder_query",
-                        "spec": {
-                          "name": "A",
-                          "stepInterval": 0,
-                          "signal": "metrics",
-                          "source": "",
-                          "aggregations": [
-                            {
-                              "metricName": "cloudsql.googleapis.com/database/cpu/utilization",
-                              "temporality": "",
-                              "timeAggregation": "max",
-                              "spaceAggregation": "max",
-                              "reduceTo": ""
-                            }
-                          ],
-                          "disabled": false,
-                          "filter": {
-                            "expression": "project_id = $project_id AND database_id in $database_id AND gcp.resource_type = 'cloudsql_database' "
-                          },
-                          "groupBy": [
-                            {
-                              "name": "database_id",
-                              "signal": "",
-                              "fieldContext": "attribute",
-                              "fieldDataType": "string"
-                            }
-                          ],
-                          "order": null,
-                          "selectFields": null,
-                          "secondaryAggregations": null,
-                          "functions": null,
-                          "legend": "{{database_id}}"
-                        }
-                      },
-                      {
-                        "type": "builder_formula",
-                        "spec": {
-                          "name": "F1",
-                          "expression": "100 * A",
-                          "disabled": false,
-                          "order": null,
-                          "functions": null,
-                          "legend": "{{database_id}}"
-                        }
+                        "metricName": "cloudsql.googleapis.com/database/cpu/utilization",
+                        "temporality": "",
+                        "timeAggregation": "max",
+                        "spaceAggregation": "max",
+                        "reduceTo": "avg"
                       }
-                    ]
+                    ],
+                    "disabled": false,
+                    "filter": {
+                      "expression": "project_id = $project_id AND database_id in $database_id AND gcp.resource_type = 'cloudsql_database' "
+                    },
+                    "groupBy": [
+                      {
+                        "name": "database_id",
+                        "signal": "",
+                        "fieldContext": "attribute",
+                        "fieldDataType": "string"
+                      }
+                    ],
+                    "order": [],
+                    "having": {
+                      "expression": ""
+                    },
+                    "functions": [],
+                    "legend": "{{database_id}}"
                   }
                 }
               }
@@ -1434,4 +1417,4 @@
     "refreshInterval": "",
     "links": []
   }
-}
+}

pkg/modules/dashboard/impldashboard/v2_module.go

@@ -245,13 +245,11 @@ func (module *module) PatchV2(ctx context.Context, orgID valuer.UUID, id valuer.
 }
 
 func (module *module) DeleteV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID) error {
-	// Read the storable, not the decoded v2 dashboard: deleting must work even
-	// when the stored data is corrupt or never migrated off the v1 schema.
-	storable, err := module.store.Get(ctx, orgID, id)
+	existing, err := module.GetV2(ctx, orgID, id)
 	if err != nil {
 		return err
 	}
-	if err := storable.ErrIfNotDeletable(); err != nil {
+	if err := existing.ErrIfNotDeletable(); err != nil {
 		return err
 	}
 

pkg/querier/promql_query.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"math"
 	"regexp"
 	"sort"
 	"strings"
@@ -478,11 +479,19 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
 
 		for idx := range v.Floats {
 			p := v.Floats[idx]
+			// NaN and +/-Inf have no JSON number form and nothing to plot; the
+			// builder path drops them while scanning rows (see consume.go).
+			if math.IsNaN(p.F) || math.IsInf(p.F, 0) {
+				continue
+			}
 			s.Values = append(s.Values, &qbv5.TimeSeriesValue{
 				Timestamp: p.T,
 				Value:     p.F,
 			})
 		}
+		if len(s.Values) == 0 {
+			continue
+		}
 		series = append(series, &s)
 	}
 
@@ -494,13 +503,11 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
 	}
 	statsMu.Unlock()
 
-	tsData := &qbv5.TimeSeriesData{
-		QueryName: q.query.Name,
-		Aggregations: []*qbv5.AggregationBucket{
-			{
-				Series: series,
-			},
-		},
+	tsData := &qbv5.TimeSeriesData{QueryName: q.query.Name}
+	// No bucket at all when nothing survived: a bucket holding no series reads
+	// as "filtered to empty" to the cache, which stores it as a real result.
+	if len(series) > 0 {
+		tsData.Aggregations = []*qbv5.AggregationBucket{{Series: series}}
 	}
 
 	var payload any = tsData

pkg/querier/promql_query_test.go

@@ -2,14 +2,21 @@ package querier
 
 import (
 	"log/slog"
+	"math"
 	"strings"
+	"sync"
 	"testing"
+	"time"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/prometheus/prometheustest"
 	qbv5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestRemoveAllVarMatchers(t *testing.T) {
@@ -453,3 +460,82 @@ func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
 	}
 	assert.Empty(t, q.Fingerprint())
 }
+
+func TestToResultDropsNonFiniteValues(t *testing.T) {
+	tests := []struct {
+		description        string
+		floats             []promql.FPoint
+		expectedTimestamps []int64
+		expectedValues     []float64
+	}{
+		{
+			description:        "finite values pass through untouched",
+			floats:             []promql.FPoint{{T: 1000, F: 1.5}, {T: 2000, F: 2.5}},
+			expectedTimestamps: []int64{1000, 2000},
+			expectedValues:     []float64{1.5, 2.5},
+		},
+		{
+			description:        "a ratio's 0/0 points are dropped, the rest kept",
+			floats:             []promql.FPoint{{T: 1000, F: 1.5}, {T: 2000, F: math.NaN()}, {T: 3000, F: 2.5}},
+			expectedTimestamps: []int64{1000, 3000},
+			expectedValues:     []float64{1.5, 2.5},
+		},
+		{
+			description:        "both infinities are dropped",
+			floats:             []promql.FPoint{{T: 1000, F: math.Inf(1)}, {T: 2000, F: 4.5}, {T: 3000, F: math.Inf(-1)}},
+			expectedTimestamps: []int64{2000},
+			expectedValues:     []float64{4.5},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			q := &promqlQuery{query: qbv5.PromQuery{Name: "A"}, requestType: qbv5.RequestTypeTimeSeries}
+			matrix := promql.Matrix{{Metric: labels.FromStrings("job_name", "dbBloatMonitorJob"), Floats: test.floats}}
+
+			var mu sync.Mutex
+			var rows, bytes uint64
+			result := q.toResult(matrix, nil, time.Now(), &mu, &rows, &bytes)
+
+			tsData, ok := result.Value.(*qbv5.TimeSeriesData)
+			require.True(t, ok)
+			require.Len(t, tsData.Aggregations, 1)
+			require.Len(t, tsData.Aggregations[0].Series, 1)
+
+			timestamps := make([]int64, 0, len(test.expectedTimestamps))
+			values := make([]float64, 0, len(test.expectedValues))
+			for _, v := range tsData.Aggregations[0].Series[0].Values {
+				timestamps = append(timestamps, v.Timestamp)
+				values = append(values, v.Value)
+			}
+			assert.Equal(t, test.expectedTimestamps, timestamps)
+			assert.Equal(t, test.expectedValues, values)
+		})
+	}
+}
+
+// A series left with nothing must not surface as an empty series, and a result
+// left with no series must carry no aggregation bucket at all — the cache reads
+// a bucket holding no series as a real, filtered-to-empty result and stores it.
+func TestToResultDropsSeriesAndBucketLeftEmpty(t *testing.T) {
+	q := &promqlQuery{query: qbv5.PromQuery{Name: "A"}, requestType: qbv5.RequestTypeTimeSeries}
+	matrix := promql.Matrix{
+		{Metric: labels.FromStrings("job_name", "idleJob"), Floats: []promql.FPoint{{T: 1000, F: math.NaN()}}},
+		{Metric: labels.FromStrings("job_name", "activeJob"), Floats: []promql.FPoint{{T: 1000, F: 7.5}}},
+	}
+
+	var mu sync.Mutex
+	var rows, bytes uint64
+	tsData, ok := q.toResult(matrix, nil, time.Now(), &mu, &rows, &bytes).Value.(*qbv5.TimeSeriesData)
+	require.True(t, ok)
+	require.Len(t, tsData.Aggregations, 1)
+	require.Len(t, tsData.Aggregations[0].Series, 1, "the all-NaN series is gone")
+	assert.Equal(t, "activeJob", tsData.Aggregations[0].Series[0].Labels[0].Value)
+
+	allNaN := promql.Matrix{
+		{Metric: labels.FromStrings("job_name", "idleJob"), Floats: []promql.FPoint{{T: 1000, F: math.NaN()}}},
+	}
+	tsData, ok = q.toResult(allNaN, nil, time.Now(), &mu, &rows, &bytes).Value.(*qbv5.TimeSeriesData)
+	require.True(t, ok)
+	assert.Empty(t, tsData.Aggregations)
+}

pkg/types/cloudintegrationtypes/serviceid.go

@@ -46,7 +46,6 @@ var (
 	GCPServiceComputeEngine    = ServiceID{valuer.NewString("computeengine")}
 	GCPServiceGKE              = ServiceID{valuer.NewString("gke")}
 	GCPServiceCloudStorage     = ServiceID{valuer.NewString("cloudstorage")}
-	GCPServiceCloudSQLMySQL    = ServiceID{valuer.NewString("cloudsql_mysql")}
 )
 
 func (ServiceID) Enum() []any {
@@ -83,7 +82,6 @@ func (ServiceID) Enum() []any {
 		GCPServiceComputeEngine,
 		GCPServiceGKE,
 		GCPServiceCloudStorage,
-		GCPServiceCloudSQLMySQL,
 	}
 }
 
@@ -126,7 +124,6 @@ var SupportedServices = map[CloudProviderType][]ServiceID{
 		GCPServiceComputeEngine,
 		GCPServiceGKE,
 		GCPServiceCloudStorage,
-		GCPServiceCloudSQLMySQL,
 	},
 }
 

pkg/types/dashboardtypes/dashboard.go

@@ -201,18 +201,6 @@ func (storableDashboardData *StorableDashboardData) GetWidgetIds() []string {
 	return widgetIds
 }
 
-// ErrIfNotDeletable gates deletion on the columns alone, never on Data, so a
-// dashboard whose data is corrupt or stuck on the v1 schema stays deletable.
-func (storable StorableDashboard) ErrIfNotDeletable() error {
-	if storable.Locked {
-		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "cannot delete a locked dashboard, please unlock the dashboard to delete")
-	}
-	if !storable.Source.isUserDeletable() {
-		return errors.Newf(errors.TypeInvalidInput, ErrCodeDashboardImmutable, "%s dashboards cannot be deleted", storable.Source)
-	}
-	return nil
-}
-
 func (dashboard *Dashboard) ErrIfNotMutable() error {
 	if dashboard.Source == SourceIntegration {
 		return errors.Newf(errors.TypeInvalidInput, ErrCodeDashboardImmutable, "integration dashboards cannot be modified")

pkg/types/dashboardtypes/dashboard_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/SigNoz/signoz/pkg/types"
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/stretchr/testify/assert"
 )
@@ -82,64 +81,3 @@ func TestCanUpdate_MultipleDeletions_ByDiff(t *testing.T) {
 		})
 	}
 }
-
-func TestStorableDashboardErrIfNotDeletable(t *testing.T) {
-	testCases := []struct {
-		subtestName     string
-		locked          bool
-		source          Source
-		data            StorableDashboardData
-		expectDeletable bool
-	}{
-		{
-			subtestName:     "user dashboard on the v2 schema",
-			source:          SourceUser,
-			data:            StorableDashboardData{"metadata": map[string]any{"schemaVersion": SchemaVersion}},
-			expectDeletable: true,
-		},
-		{
-			subtestName:     "user dashboard still on the v1 schema",
-			source:          SourceUser,
-			data:            StorableDashboardData{"widgets": makeTestWidgets("a")},
-			expectDeletable: true,
-		},
-		{
-			subtestName:     "user dashboard with unreadable data",
-			source:          SourceUser,
-			data:            StorableDashboardData{"metadata": "not-an-object"},
-			expectDeletable: true,
-		},
-		{
-			subtestName:     "locked user dashboard",
-			locked:          true,
-			source:          SourceUser,
-			data:            StorableDashboardData{"widgets": makeTestWidgets("a")},
-			expectDeletable: false,
-		},
-		{
-			subtestName:     "system dashboard",
-			source:          SourceSystem,
-			data:            StorableDashboardData{"widgets": makeTestWidgets("a")},
-			expectDeletable: false,
-		},
-		{
-			subtestName:     "integration dashboard",
-			source:          SourceIntegration,
-			data:            StorableDashboardData{"widgets": makeTestWidgets("a")},
-			expectDeletable: false,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.subtestName, func(t *testing.T) {
-			storable := StorableDashboard{
-				Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
-				OrgID:        valuer.GenerateUUID(),
-				Locked:       tc.locked,
-				Source:       tc.source,
-				Data:         tc.data,
-			}
-			assert.Equal(t, tc.expectDeletable, storable.ErrIfNotDeletable() == nil)
-		})
-	}
-}

pkg/types/dashboardtypes/perses_dashboard.go

@@ -129,6 +129,16 @@ func (d *DashboardV2) LockUnlock(lock bool, isAdmin bool, updatedBy string) erro
 	return nil
 }
 
+func (d *DashboardV2) ErrIfNotDeletable() error {
+	if d.Locked {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "cannot delete a locked dashboard, please unlock the dashboard to delete")
+	}
+	if !d.Source.isUserDeletable() {
+		return errors.Newf(errors.TypeInvalidInput, ErrCodeDashboardImmutable, "%s dashboards cannot be deleted", d.Source)
+	}
+	return nil
+}
+
 func (d *DashboardV2) ErrIfNotClonable() error {
 	if !d.Source.isClonable() {
 		return errors.Newf(errors.TypeInvalidInput, ErrCodeDashboardImmutable, "%s dashboards cannot be cloned", d.Source)

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

@@ -0,0 +1,65 @@
+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 get_all_series, make_query_request
+
+HOUR_MS = 3_600_000
+SAMPLE_INTERVAL_MS = 60_000
+
+
+def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_metrics: Callable[[list[Metrics]], None],
+) -> None:
+    # 12h ending on an hour boundary 15m ago — old enough to be cached.
+    end_ms = (int((datetime.now(tz=UTC) - timedelta(minutes=15)).timestamp() * 1000) // HOUR_MS) * HOUR_MS
+    start_ms = end_ms - 12 * HOUR_MS
+
+    sum_metric = f"job_duration_sum_{uuid4().hex[:8]}"
+    count_metric = f"job_duration_count_{uuid4().hex[:8]}"
+
+    # active_job divides finite; idle_job is 0/0 at every step.
+    series = {"active_job": (100.0, 4.0), "idle_job": (0.0, 0.0)}
+    metrics: list[Metrics] = []
+    for job_name, (sum_value, count_value) in series.items():
+        for ts_ms in range(start_ms, end_ms + 1, SAMPLE_INTERVAL_MS):
+            timestamp = datetime.fromtimestamp(ts_ms / 1000, tz=UTC)
+            metrics.append(Metrics(metric_name=sum_metric, labels={"job_name": job_name}, timestamp=timestamp, value=sum_value))
+            metrics.append(Metrics(metric_name=count_metric, labels={"job_name": job_name}, timestamp=timestamp, value=count_value))
+    insert_metrics(metrics)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    promql = f"sum by (job_name) ({sum_metric}) / sum by (job_name) ({count_metric})"
+
+    def run() -> tuple[dict[str, dict[int, object]], int]:
+        query = {"type": "promql", "spec": {"name": "A", "query": promql}}
+        response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
+        assert response.status_code == HTTPStatus.OK, response.text[:300]
+        body = response.json()
+        out: dict[str, dict[int, object]] = {}
+        for entry in get_all_series(body, "A") or []:
+            labels = {l["key"]["name"]: str(l["value"]) for l in entry.get("labels") or []}
+            out[labels["job_name"]] = {v["timestamp"]: v["value"] for v in entry.get("values") or []}
+        return out, int(body["data"]["meta"]["stepIntervals"]["A"])
+
+    # First populates the cache, second must be served from it.
+    first, step_seconds = run()
+    second, _ = run()
+
+    expected_points = (end_ms - start_ms) // (step_seconds * 1000) + 1
+    assert set(first) == {"active_job"}, f"the 0/0 series must not reach the response: {sorted(first)}"
+    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.
+    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"