fix(deployments): stop full scans and coalesce cache misses

go.mod

@@ -84,7 +84,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/exp v0.0.0-20260112195511-716be5621a96 // indirect
 	golang.org/x/mod v0.37.0 // indirect
-	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sync v0.22.0
 	golang.org/x/telemetry v0.0.0-20260625142307-59b4966ccb57 // indirect
 	golang.org/x/tools v0.47.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect

pkg/github/deployments.go

@@ -67,23 +67,7 @@ func GetAllDeployments(ctx context.Context, client models.Client, opts models.Li
 
 	deployments := []*googlegithub.Deployment{}
 
-	// Build the list options with filters
-	listOpts := &googlegithub.DeploymentsListOptions{
-		ListOptions: googlegithub.ListOptions{PerPage: 100},
-	}
-
-	if opts.SHA != "" {
-		listOpts.SHA = opts.SHA
-	}
-	if opts.GitRef != "" {
-		listOpts.Ref = opts.GitRef
-	}
-	if opts.Task != "" {
-		listOpts.Task = opts.Task
-	}
-	if opts.Environment != "" {
-		listOpts.Environment = opts.Environment
-	}
+	listOpts := toDeploymentsListOptions(opts)
 
 	page := 1
 	for page != 0 {
@@ -104,20 +88,53 @@ func GetAllDeployments(ctx context.Context, client models.Client, opts models.Li
 	return DeploymentsWrapper(deployments), nil
 }
 
-// GetDeploymentsInRange retrieves every deployment from the repository and then returns the ones that fall within the given time range.
+// GetDeploymentsInRange retrieves deployments in the given time range.
 func GetDeploymentsInRange(ctx context.Context, client models.Client, opts models.ListDeploymentsOptions, from time.Time, to time.Time) (DeploymentsWrapper, error) {
-	deployments, err := GetAllDeployments(ctx, client, opts)
-	if err != nil {
-		return nil, err
+	if opts.Owner == "" || opts.Repository == "" {
+		return nil, nil
 	}
 
 	filtered := []*googlegithub.Deployment{}
+	listOpts := toDeploymentsListOptions(opts)
 
-	for _, deployment := range deployments {
-		if createdAt := deployment.CreatedAt.GetTime(); createdAt != nil && !createdAt.Before(from) && !createdAt.After(to) {
-			filtered = append(filtered, deployment)
+	page := 1
+	for page != 0 {
+		listOpts.Page = page
+		deployments, resp, err := client.ListDeployments(ctx, opts.Owner, opts.Repository, listOpts)
+		if err != nil {
+			return nil, fmt.Errorf("listing deployments: opts=%+v: %v", opts, err)
+		}
+
+		olderDeployment := false
+		for _, deployment := range deployments {
+			createdAt := deployment.CreatedAt.GetTime()
+			if createdAt == nil {
+				continue
+			}
+			if !createdAt.Before(from) && !createdAt.After(to) {
+				filtered = append(filtered, deployment)
+			}
+			if createdAt.Before(from) {
+				olderDeployment = true
+			}
+		}
+
+		// GitHub lists deployments newest-first, so later pages cannot be in range.
+		if olderDeployment || resp == nil {
+			break
 		}
+		page = resp.NextPage
 	}
 
 	return DeploymentsWrapper(filtered), nil
 }
+
+func toDeploymentsListOptions(opts models.ListDeploymentsOptions) *googlegithub.DeploymentsListOptions {
+	return &googlegithub.DeploymentsListOptions{
+		SHA:         opts.SHA,
+		Ref:         opts.GitRef,
+		Task:        opts.Task,
+		Environment: opts.Environment,
+		ListOptions: googlegithub.ListOptions{PerPage: 100},
+	}
+}

pkg/github/deployments_test.go

@@ -13,6 +13,8 @@ import (
 type mockDeploymentsClient struct {
 	mockDeployments []*googlegithub.Deployment
 	mockResponse    *googlegithub.Response
+	pages           map[int]deploymentPage
+	requests        []*googlegithub.DeploymentsListOptions
 	expectedOwner   string
 	expectedRepo    string
 	t               *testing.T
@@ -42,11 +44,22 @@ func (m *mockDeploymentsClient) ListAlertsForOrg(ctx context.Context, owner stri
 	return nil, nil, nil
 }
 
+type deploymentPage struct {
+	deployments []*googlegithub.Deployment
+	nextPage    int
+}
+
 func (m *mockDeploymentsClient) ListDeployments(ctx context.Context, owner, repo string, opts *googlegithub.DeploymentsListOptions) ([]*googlegithub.Deployment, *googlegithub.Response, error) {
 	if owner != m.expectedOwner || repo != m.expectedRepo {
 		m.t.Errorf("Expected owner/repo to be %s/%s, got %s/%s", m.expectedOwner, m.expectedRepo, owner, repo)
 	}
+	request := *opts
+	m.requests = append(m.requests, &request)
 
+	if m.pages != nil {
+		page := m.pages[opts.Page]
+		return page.deployments, &googlegithub.Response{NextPage: page.nextPage}, nil
+	}
 	return m.mockDeployments, m.mockResponse, nil
 }
 
@@ -239,6 +252,48 @@ func TestGetDeploymentsInRange(t *testing.T) {
 	}
 }
 
+func TestGetDeploymentsInRangePaginatesUntilOlderDeployment(t *testing.T) {
+	from := time.Date(2025, 1, 10, 0, 0, 0, 0, time.UTC)
+	to := time.Date(2025, 1, 20, 0, 0, 0, 0, time.UTC)
+	createdAt := func(id int64, at *time.Time) *googlegithub.Deployment {
+		deployment := &googlegithub.Deployment{ID: googlegithub.Ptr(id)}
+		if at != nil {
+			deployment.CreatedAt = &googlegithub.Timestamp{Time: *at}
+		}
+		return deployment
+	}
+	afterTo := to.Add(time.Hour)
+	inRange := from.Add(time.Hour)
+	beforeFrom := from.Add(-time.Hour)
+	opts := models.ListDeploymentsOptions{
+		Repository: "repo", Owner: "owner", SHA: "sha", GitRef: "main", Task: "deploy", Environment: "production",
+	}
+	client := &mockDeploymentsClient{
+		expectedOwner: "owner", expectedRepo: "repo", t: t,
+		pages: map[int]deploymentPage{
+			1: {deployments: []*googlegithub.Deployment{createdAt(1, &afterTo), createdAt(2, nil), createdAt(3, &inRange)}, nextPage: 2},
+			2: {deployments: []*googlegithub.Deployment{createdAt(4, &inRange), createdAt(5, &beforeFrom)}, nextPage: 3},
+			3: {deployments: []*googlegithub.Deployment{createdAt(6, &inRange)}},
+		},
+	}
+
+	deployments, err := GetDeploymentsInRange(context.Background(), client, opts, from, to)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(deployments) != 2 || deployments[0].GetID() != 3 || deployments[1].GetID() != 4 {
+		t.Fatalf("expected in-range deployments 3 and 4, got %+v", deployments)
+	}
+	if len(client.requests) != 2 || client.requests[0].Page != 1 || client.requests[1].Page != 2 {
+		t.Fatalf("expected requests for pages 1 and 2 only, got %+v", client.requests)
+	}
+	for _, request := range client.requests {
+		if request.PerPage != 100 || request.SHA != opts.SHA || request.Ref != opts.GitRef || request.Task != opts.Task || request.Environment != opts.Environment {
+			t.Errorf("deployment filters were not preserved: %+v", request)
+		}
+	}
+}
+
 func TestDeploymentsWrapperFrames(t *testing.T) {
 	// Create test data
 	createdAt := &googlegithub.Timestamp{Time: time.Now().Add(-48 * time.Hour)}

pkg/plugin/datasource_caching.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/grafana/github-datasource/pkg/dfutil"
 	"github.com/grafana/github-datasource/pkg/models"
@@ -51,6 +52,8 @@ type CachedDatasource struct {
 
 	mu    sync.RWMutex // protects the cache map against concurrent access
 	cache map[string]CachedResult
+
+	deploymentRequests singleflight.Group
 }
 
 func (c *CachedDatasource) getCache(req backend.DataQuery) (dfutil.Framer, error) {
@@ -268,8 +271,21 @@ func (c *CachedDatasource) HandleDeploymentsQuery(ctx context.Context, q *models
 		return value, err
 	}
 
-	f, err := c.datasource.HandleDeploymentsQuery(ctx, q, req)
-	return c.saveCache(req, f, err)
+	key, err := getCacheKey(req)
+	if err != nil {
+		return nil, err
+	}
+	value, err, _ := c.deploymentRequests.Do(key, func() (interface{}, error) {
+		if value, err := c.getCache(req); err == nil {
+			return value, nil
+		}
+		f, err := c.datasource.HandleDeploymentsQuery(context.WithoutCancel(ctx), q, req)
+		return c.saveCache(req, f, err)
+	})
+	if value == nil {
+		return nil, err
+	}
+	return value.(dfutil.Framer), err
 }
 
 // HandleOrganizationsQuery is the cache wrapper for the organizations query handler

pkg/plugin/datasource_caching_test.go

@@ -1,13 +1,19 @@
 package plugin
 
 import (
+	"context"
 	"encoding/json"
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/grafana/github-datasource/pkg/dfutil"
+	"github.com/grafana/github-datasource/pkg/models"
 )
 
 // mockFramer is a struct implementing the Framer interface that returns predefined frames for testing purposes
@@ -25,6 +31,85 @@ var framesA = data.Frames{data.NewFrame("A", nil)}
 var dataQueryB = backend.DataQuery{JSON: json.RawMessage(`{"query": "B"}`)}
 var framesB = data.Frames{data.NewFrame("B", nil)}
 
+type blockingDeploymentsDatasource struct {
+	Datasource
+	calls        atomic.Int32
+	started      chan struct{}
+	release      <-chan struct{}
+	checkContext bool
+	once         sync.Once
+}
+
+func (d *blockingDeploymentsDatasource) HandleDeploymentsQuery(ctx context.Context, _ *models.DeploymentsQuery, _ backend.DataQuery) (dfutil.Framer, error) {
+	d.calls.Add(1)
+	d.once.Do(func() { close(d.started) })
+	<-d.release
+	if d.checkContext && ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+	return mockFramer{frames: framesA}, nil
+}
+
+func TestCachedDeploymentsQueryCoalescesCacheMisses(t *testing.T) {
+	const queries = 10
+	release := make(chan struct{})
+	datasource := &blockingDeploymentsDatasource{started: make(chan struct{}), release: release}
+	cached := WithCaching(datasource)
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	errs := make(chan error, queries)
+
+	for range queries {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			_, err := cached.HandleDeploymentsQuery(context.Background(), &models.DeploymentsQuery{}, dataQueryA)
+			errs <- err
+		}()
+	}
+	close(start)
+	<-datasource.started
+	time.Sleep(10 * time.Millisecond)
+	if calls := datasource.calls.Load(); calls != 1 {
+		t.Fatalf("expected one underlying deployment query, got %d", calls)
+	}
+	close(release)
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestCachedDeploymentsQueryIgnoresLeaderCancellation(t *testing.T) {
+	release := make(chan struct{})
+	datasource := &blockingDeploymentsDatasource{started: make(chan struct{}), release: release, checkContext: true}
+	cached := WithCaching(datasource)
+	leaderCtx, cancel := context.WithCancel(context.Background())
+	errs := make(chan error, 2)
+
+	go func() {
+		_, err := cached.HandleDeploymentsQuery(leaderCtx, &models.DeploymentsQuery{}, dataQueryA)
+		errs <- err
+	}()
+	<-datasource.started
+	cancel()
+	go func() {
+		_, err := cached.HandleDeploymentsQuery(context.Background(), &models.DeploymentsQuery{}, dataQueryA)
+		errs <- err
+	}()
+	close(release)
+
+	for range 2 {
+		if err := <-errs; err != nil {
+			t.Fatalf("expected shared deployment query to survive leader cancellation: %v", err)
+		}
+	}
+}
+
 func TestWithCaching(t *testing.T) {
 	cachedDS := WithCaching(nil)