diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index b0ab36e36e1..6a22a96590f 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -1273,7 +1273,7 @@ func (b *bleveIndex) updateIndexWithLatestModifications(ctx context.Context, req startTime := time.Now() listRV, docs, err := b.updaterFn(ctx, b, sinceRV) - if err == nil && listRV > 0 { + if err == nil && listRV > 0 && listRV != sinceRV { err = b.updateResourceVersion(listRV) // updates b.resourceVersion } diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index be6789844e6..6a567dbb2c5 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -11,7 +11,6 @@ import ( "time" "github.com/go-sql-driver/mysql" - "github.com/grafana/grafana/pkg/util/sqlite" "github.com/jackc/pgx/v5/pgconn" "github.com/lib/pq" "github.com/prometheus/client_golang/prometheus" @@ -20,6 +19,8 @@ import ( "google.golang.org/protobuf/proto" apierrors "k8s.io/apimachinery/pkg/api/errors" + "github.com/grafana/grafana/pkg/util/sqlite" + "github.com/grafana/grafana-app-sdk/logging" "github.com/grafana/grafana/pkg/storage/unified/resource" @@ -642,21 +643,34 @@ func (b *backend) ListModifiedSince(ctx context.Context, key resource.Namespaced } } + rollbackOnDefer := true + defer func() { + if rollbackOnDefer { + if terr := tx.Rollback(); terr != nil { + b.log.Warn("Error rolling back transaction in ListModifiedSince", "error", terr) + } + } + }() + // Fetch latest RV within the transaction latestRv, err := b.fetchLatestRV(ctx, tx, b.dialect, key.Group, key.Resource) if err != nil { - terr := tx.Rollback() - if terr != nil { - b.log.Warn("Error rolling back transaction in ListModifiedSince", "error", terr) - } return 0, func(yield func(*resource.ModifiedResource, error) bool) { yield(nil, err) } } + // If latest RV is the same as request RV, there's nothing to report, and we can avoid running another query. + if latestRv == sinceRv { + return 0, func(yield func(*resource.ModifiedResource, error) bool) { /* nothing to return */ } + } + // since results are sorted by name ASC and rv DESC, we can get away with tracking the last seen lastSeen := "" + // We will rollback after iteration has finished. + rollbackOnDefer = false + // rollback transaction if iterator not called within 30 seconds rollbackTimer := time.AfterFunc(30*time.Second, func() { if err := tx.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) { diff --git a/pkg/storage/unified/testing/storage_backend.go b/pkg/storage/unified/testing/storage_backend.go index 4b2b263bdb0..61b4ba260fd 100644 --- a/pkg/storage/unified/testing/storage_backend.go +++ b/pkg/storage/unified/testing/storage_backend.go @@ -3,6 +3,7 @@ package test import ( "context" "fmt" + "iter" "net/http" "slices" "strings" @@ -523,11 +524,22 @@ func runTestIntegrationBackendListModifiedSince(t *testing.T, backend resource.S latestRv, seq := backend.ListModifiedSince(ctx, key, rvDeleted) require.GreaterOrEqual(t, latestRv, rvDeleted) - counter := 0 - for range seq { - counter++ + isEmpty(t, seq) + }) + + t.Run("no events for subsequent listModifiedSince calls", func(t *testing.T) { + key := resource.NamespacedResource{ + Namespace: ns, + Group: "group", + Resource: "resource", } - require.Equal(t, 0, counter) // no events should be returned + latestRv1, seq := backend.ListModifiedSince(ctx, key, rvDeleted) + require.GreaterOrEqual(t, latestRv1, rvDeleted) + isEmpty(t, seq) + + latestRv2, seq := backend.ListModifiedSince(ctx, key, latestRv1) + require.GreaterOrEqual(t, latestRv1, latestRv2) + isEmpty(t, seq) }) t.Run("will only return modified events for the given key", func(t *testing.T) { @@ -582,6 +594,14 @@ func runTestIntegrationBackendListModifiedSince(t *testing.T, backend resource.S }) } +func isEmpty(t *testing.T, seq iter.Seq2[*resource.ModifiedResource, error]) { + counter := 0 + for range seq { + counter++ + } + require.Equal(t, 0, counter) +} + func runTestIntegrationBackendListHistory(t *testing.T, backend resource.StorageBackend, nsPrefix string) { ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second)) server := newServer(t, backend)