fix(circleci): limit unfinished details recollection to recent workflows (#9304)

backend/plugins/circleci/e2e/job_collector_test.go

@@ -21,7 +21,10 @@ import (
 	"reflect"
 	"sort"
 	"testing"
+	"time"
 
+	"github.com/apache/devlake/core/dal"
+	"github.com/apache/devlake/core/models/common"
 	"github.com/apache/devlake/helpers/e2ehelper"
 	"github.com/apache/devlake/helpers/pluginhelper/api"
 	"github.com/apache/devlake/plugins/circleci/impl"
@@ -30,53 +33,95 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-// TestCircleciUnfinishedJobsInputIterator is a regression test for
-// https://github.com/apache/devlake/issues/8907. The "collect unfinished job
-// details" collector builds its URL from "/v2/workflow/{{ .Input.Id }}/job" while
-// scanning rows into a models.CircleciJob. Its input query must therefore expose the
-// workflow id in the row's Id field; a bare "DISTINCT workflow_id" left Id empty and
-// produced "/v2/workflow//job" (HTTP 500). This test runs the production query
-// (tasks.UnfinishedJobsInputClauses) through the real iterator and asserts each
-// yielded row's Id is the workflow id, that results are DISTINCT, and that the
-// status/connection filters hold.
-func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
-	var circleci impl.Circleci
-	dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+const unfinishedTestProjectSlug = "github/test/repo"
 
-	const projectSlug = "github/test/repo"
-	dataflowTester.FlushTabler(&models.CircleciJob{})
+var unfinishedTestNow = time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
 
-	seed := []models.CircleciJob{
-		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-1", ProjectSlug: projectSlug, Status: "on_hold"},
-		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-2", ProjectSlug: projectSlug, Status: "running"}, // same workflow -> DISTINCT
-		{ConnectionId: 1, WorkflowId: "wf-queued", Id: "job-3", ProjectSlug: projectSlug, Status: "queued"},
-		{ConnectionId: 1, WorkflowId: "wf-success", Id: "job-4", ProjectSlug: projectSlug, Status: "success"},   // terminal -> excluded
-		{ConnectionId: 2, WorkflowId: "wf-otherconn", Id: "job-5", ProjectSlug: projectSlug, Status: "on_hold"}, // other connection -> excluded
+func daysAgo(days int) *common.Iso8601Time {
+	return &common.Iso8601Time{Time: unfinishedTestNow.AddDate(0, 0, -days)}
+}
+
+func seedUnfinishedWorkflows(t *testing.T, dataflowTester *e2ehelper.DataFlowTester) {
+	dataflowTester.FlushTabler(&models.CircleciWorkflow{})
+	seed := []models.CircleciWorkflow{
+		{ConnectionId: 1, Id: "wf-onhold", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(1)},
+		{ConnectionId: 1, Id: "wf-queued", ProjectSlug: unfinishedTestProjectSlug, Status: "running", CreatedDate: daysAgo(29)},
+		{ConnectionId: 1, Id: "wf-failing", ProjectSlug: unfinishedTestProjectSlug, Status: "failing", CreatedDate: daysAgo(2)},
+		{ConnectionId: 1, Id: "wf-success", ProjectSlug: unfinishedTestProjectSlug, Status: "success", CreatedDate: daysAgo(1)},
+		{ConnectionId: 1, Id: "wf-old-onhold", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(31)},
+		{ConnectionId: 2, Id: "wf-otherconn", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold", CreatedDate: daysAgo(1)},
 	}
 	for i := range seed {
 		assert.Nil(t, dataflowTester.Dal.Create(&seed[i]))
 	}
+}
 
-	cursor, err := dataflowTester.Dal.Cursor(tasks.UnfinishedJobsInputClauses(1, projectSlug)...)
+func collectInputIds(t *testing.T, dataflowTester *e2ehelper.DataFlowTester, clauses []dal.Clause, rowType reflect.Type) []string {
+	cursor, err := dataflowTester.Dal.Cursor(clauses...)
 	assert.Nil(t, err)
-	iter, err := api.NewDalCursorIterator(dataflowTester.Dal, cursor, reflect.TypeOf(models.CircleciJob{}))
+	iter, err := api.NewDalCursorIterator(dataflowTester.Dal, cursor, rowType)
 	assert.Nil(t, err)
 	defer iter.Close()
 
 	var ids []string
 	for iter.HasNext() {
 		item, err := iter.Fetch()
 		assert.Nil(t, err)
-		job := item.(*models.CircleciJob)
-		ids = append(ids, job.Id)
+		ids = append(ids, reflect.ValueOf(item).Elem().FieldByName("Id").String())
 	}
 	sort.Strings(ids)
+	return ids
+}
+
+// TestCircleciUnfinishedJobsInputIterator is a regression test for
+// https://github.com/apache/devlake/issues/8907 and
+// https://github.com/apache/devlake/issues/9304. The "collect unfinished job
+// details" collector builds its URL from "/v2/workflow/{{ .Input.Id }}/job" while
+// scanning rows into a models.CircleciJob. Its input query must therefore expose the
+// workflow id in the row's Id field; a bare "DISTINCT workflow_id" left Id empty and
+// produced "/v2/workflow//job" (HTTP 500). It must also skip workflows older than the
+// lookback window: an approval job that is never approved stays on_hold forever, and
+// its workflow was collected again on every run (#9304).
+func TestCircleciUnfinishedJobsInputIterator(t *testing.T) {
+	var circleci impl.Circleci
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+
+	seedUnfinishedWorkflows(t, dataflowTester)
+	dataflowTester.FlushTabler(&models.CircleciJob{})
+	seed := []models.CircleciJob{
+		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-1", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"},
+		{ConnectionId: 1, WorkflowId: "wf-onhold", Id: "job-2", ProjectSlug: unfinishedTestProjectSlug, Status: "running"}, // same workflow -> DISTINCT
+		{ConnectionId: 1, WorkflowId: "wf-queued", Id: "job-3", ProjectSlug: unfinishedTestProjectSlug, Status: "queued"},
+		{ConnectionId: 1, WorkflowId: "wf-success", Id: "job-4", ProjectSlug: unfinishedTestProjectSlug, Status: "success"},    // terminal -> excluded
+		{ConnectionId: 1, WorkflowId: "wf-old-onhold", Id: "job-5", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"}, // older than lookback -> excluded
+		{ConnectionId: 2, WorkflowId: "wf-otherconn", Id: "job-6", ProjectSlug: unfinishedTestProjectSlug, Status: "on_hold"},  // other connection -> excluded
+	}
+	for i := range seed {
+		assert.Nil(t, dataflowTester.Dal.Create(&seed[i]))
+	}
+
+	clauses := tasks.UnfinishedJobsInputClauses(1, unfinishedTestProjectSlug, tasks.UnfinishedDetailsCreatedAfter(unfinishedTestNow))
+	ids := collectInputIds(t, dataflowTester, clauses, reflect.TypeOf(models.CircleciJob{}))
 
 	// Distinct workflow ids for connection 1's non-terminal jobs, with Id populated
-	// (the URL template reads .Input.Id). wf-success (terminal) and wf-otherconn
-	// (connection 2) are excluded.
+	// (the URL template reads .Input.Id).
 	assert.Equal(t, []string{"wf-onhold", "wf-queued"}, ids)
 	for _, id := range ids {
 		assert.NotEmpty(t, id, "Input.Id must be the workflow id, not empty (#8907)")
 	}
 }
+
+// TestCircleciUnfinishedWorkflowsInputIterator is a regression test for
+// https://github.com/apache/devlake/issues/9304: the "collect unfinished workflow
+// details" collector must skip workflows older than the lookback window.
+func TestCircleciUnfinishedWorkflowsInputIterator(t *testing.T) {
+	var circleci impl.Circleci
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "circleci", circleci)
+
+	seedUnfinishedWorkflows(t, dataflowTester)
+
+	clauses := tasks.UnfinishedWorkflowsInputClauses(1, unfinishedTestProjectSlug, tasks.UnfinishedDetailsCreatedAfter(unfinishedTestNow))
+	ids := collectInputIds(t, dataflowTester, clauses, reflect.TypeOf(models.CircleciWorkflow{}))
+
+	assert.Equal(t, []string{"wf-failing", "wf-onhold", "wf-queued"}, ids)
+}

backend/plugins/circleci/tasks/job_collector.go

@@ -43,14 +43,16 @@ var CollectJobsMeta = plugin.SubTaskMeta{
 
 // UnfinishedJobsInputClauses returns the DAL clauses that select the workflows whose
 // jobs are still in a non-terminal status and therefore need their job details
-// recollected by the CollectJobs "unfinished details" collector.
-func UnfinishedJobsInputClauses(connectionId uint64, projectSlug string) []dal.Clause {
+// recollected by the CollectJobs "unfinished details" collector. Only workflows created
+// at or after createdAfter are selected (#9304); jobs have no created date of their own.
+func UnfinishedJobsInputClauses(connectionId uint64, projectSlug string, createdAfter time.Time) []dal.Clause {
 	return []dal.Clause{
-		dal.Select("DISTINCT workflow_id AS id"), // #8907: alias to id so {{ .Input.Id }} resolves when scanned into CircleciJob
-		dal.From(&models.CircleciJob{}),
+		dal.Select("DISTINCT j.workflow_id AS id"), // #8907: alias to id so {{ .Input.Id }} resolves when scanned into CircleciJob
+		dal.From("_tool_circleci_jobs j"),
+		dal.Join("JOIN _tool_circleci_workflows w ON w.connection_id = j.connection_id AND w.id = j.workflow_id"),
 		dal.Where(
-			"connection_id = ? AND project_slug = ? AND status IN ('running', 'not_running', 'queued', 'on_hold')",
-			connectionId, projectSlug,
+			"j.connection_id = ? AND j.project_slug = ? AND j.status IN ('running', 'not_running', 'queued', 'on_hold') AND w.created_date >= ?",
+			connectionId, projectSlug, createdAfter,
 		),
 	}
 }
@@ -111,7 +113,8 @@ func CollectJobs(taskCtx plugin.SubTaskContext) errors.Error {
 			},
 			BuildInputIterator: func() (api.Iterator, errors.Error) {
 				db := taskCtx.GetDal()
-				cursor, err := db.Cursor(UnfinishedJobsInputClauses(data.Options.ConnectionId, data.Options.ProjectSlug)...)
+				clauses := UnfinishedJobsInputClauses(data.Options.ConnectionId, data.Options.ProjectSlug, UnfinishedDetailsCreatedAfter(time.Now()))
+				cursor, err := db.Cursor(clauses...)
 				if err != nil {
 					return nil, err
 				}

backend/plugins/circleci/tasks/shared.go

@@ -31,6 +31,17 @@ import (
 	"github.com/apache/devlake/plugins/circleci/models"
 )
 
+// UnfinishedDetailsLookback limits how far back the "unfinished details" collectors look.
+// A workflow with an approval job that is never approved stays on_hold forever; without
+// a limit it would be collected again on every run (#9304).
+const UnfinishedDetailsLookback = 30 * 24 * time.Hour
+
+// UnfinishedDetailsCreatedAfter returns the earliest created date of a workflow that the
+// "unfinished details" collectors still collect again.
+func UnfinishedDetailsCreatedAfter(now time.Time) time.Time {
+	return now.Add(-UnfinishedDetailsLookback)
+}
+
 var accountIdGen *didgen.DomainIdGenerator
 var projectIdGen *didgen.DomainIdGenerator
 var pipelineIdGen *didgen.DomainIdGenerator

backend/plugins/circleci/tasks/workflow_collector.go

@@ -42,6 +42,21 @@ var CollectWorkflowsMeta = plugin.SubTaskMeta{
 	DomainTypes:      []string{plugin.DOMAIN_TYPE_CICD},
 }
 
+// UnfinishedWorkflowsInputClauses returns the DAL clauses that select the workflows that
+// are still in a non-terminal status and therefore need their details recollected by the
+// CollectWorkflows "unfinished details" collector. Only workflows created at or after
+// createdAfter are selected (#9304).
+func UnfinishedWorkflowsInputClauses(connectionId uint64, projectSlug string, createdAfter time.Time) []dal.Clause {
+	return []dal.Clause{
+		dal.Select("id"),
+		dal.From(&models.CircleciWorkflow{}),
+		dal.Where(
+			"connection_id = ? AND project_slug = ? AND status IN ('running', 'on_hold', 'failing') AND created_date >= ?",
+			connectionId, projectSlug, createdAfter,
+		),
+	}
+}
+
 func CollectWorkflows(taskCtx plugin.SubTaskContext) errors.Error {
 	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_WORKFLOW_TABLE)
 	logger := taskCtx.GetLogger()
@@ -93,12 +108,7 @@ func CollectWorkflows(taskCtx plugin.SubTaskContext) errors.Error {
 				AfterResponse: ignoreDeletedOrBrokenBuilds,
 			},
 			BuildInputIterator: func() (api.Iterator, errors.Error) {
-				clauses := []dal.Clause{
-					dal.Select("id"),
-					dal.From(&models.CircleciWorkflow{}),
-					dal.Where("connection_id = ? AND project_slug = ? AND status IN ('running', 'on_hold', 'failing')", data.Options.ConnectionId, data.Options.ProjectSlug),
-				}
-
+				clauses := UnfinishedWorkflowsInputClauses(data.Options.ConnectionId, data.Options.ProjectSlug, UnfinishedDetailsCreatedAfter(time.Now()))
 				db := taskCtx.GetDal()
 				cursor, err := db.Cursor(clauses...)
 				if err != nil {