fix(unified-storage): only fetch from history table if rv changed (#104740)

pkg/storage/unified/sql/notifier_sql.go

@@ -8,6 +8,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/grafana/grafana-app-sdk/logging"
+
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 	"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
 )
@@ -138,7 +139,7 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str
 				continue
 			}
 			for group, items := range grv {
-				for resource := range items {
+				for resource, latestRV := range items {
 					// If we haven't seen this resource before, we start from 0.
 					if _, ok := since[group]; !ok {
 						since[group] = make(map[string]int64)
@@ -147,7 +148,17 @@ func (p *pollingNotifier) poller(ctx context.Context, since groupResourceRV, str
 						since[group][resource] = 0
 					}
 
-					// Poll for new events.
+					// We don't need to poll if the RV hasn't changed.
+					if since[group][resource] >= latestRV {
+						p.log.Debug("polling for resource skipped",
+							"group", group,
+							"resource", resource,
+							"latestKnownRV", since[group][resource],
+							"latestFetchedRV", latestRV)
+						continue
+					}
+
+					// Poll for new events since the last known RV.
 					next, err := p.poll(ctx, group, resource, since[group][resource], stream)
 					if err != nil {
 						p.log.Error("polling for resource", "err", err)

pkg/storage/unified/sql/notifier_sql_test.go

@@ -221,23 +221,28 @@ func TestPollingNotifier(t *testing.T) {
 			Action:          1,
 		}
 
-		var latestRVsCalled bool
+		var listLatestRVsCalledCounter int
 		listLatestRVs := func(ctx context.Context) (groupResourceRV, error) {
-			latestRVsCalled = true
+			// On the first call return 0, then the highest known RV.
+			var value int64 = 0
+			if listLatestRVsCalledCounter > 0 {
+				value = testEvent.ResourceVersion
+			}
+			listLatestRVsCalledCounter++
 			return groupResourceRV{
 				"test-group": map[string]int64{
-					"test-resource": 0,
+					"test-resource": value,
 				},
 			}, nil
 		}
 
-		var historyPollCalled bool
+		var historyPollCalledCounter int
 		once := sync.Once{}
 		historyPoll := func(ctx context.Context, grp string, res string, since int64) ([]*historyPollResponse, error) {
 			// only assert the first time - this may be called multiple times
 			// depending on the host hardware etc, due to timing issues...
+			historyPollCalledCounter++
 			once.Do(func() {
-				historyPollCalled = true
 				require.Equal(t, "test-group", grp)
 				require.Equal(t, "test-resource", res)
 				require.Equal(t, int64(0), since)
@@ -274,8 +279,8 @@ func TestPollingNotifier(t *testing.T) {
 			require.Equal(t, "test-name", event.Key.Name)
 			require.Equal(t, int64(2), event.ResourceVersion)
 			require.Equal(t, "test-folder", event.Folder)
-			require.True(t, latestRVsCalled, "listLatestRVs should be called")
-			require.True(t, historyPollCalled, "historyPoll should be called")
+			require.True(t, listLatestRVsCalledCounter > 0, "listLatestRVs should be called at least once")
+			require.True(t, historyPollCalledCounter == 1, "historyPoll should be called exactly once")
 		case <-time.After(100 * time.Millisecond):
 			t.Fatal("timeout waiting for event")
 		}