From f0e8dd0747158a64abcf215f95b36b3ec9ffb80b Mon Sep 17 00:00:00 2001 From: Volodymyr Zahorniak Date: Mon, 2 Nov 2026 16:59:27 +0200 Subject: [PATCH] fix(circleci): limit unfinished details recollection to recent workflows (#9304) The unfinished-details collectors of collectJobs and collectWorkflows selected every workflow in a non-terminal status. A workflow whose approval job is never approved stays on_hold forever, so it was collected again on every run and the raw tables kept growing. Select only workflows created within the last 30 days. --- .../circleci/e2e/job_collector_test.go | 97 ++++++++++++++----- .../plugins/circleci/tasks/job_collector.go | 17 ++-- backend/plugins/circleci/tasks/shared.go | 11 +++ .../circleci/tasks/workflow_collector.go | 22 +++-- 4 files changed, 108 insertions(+), 39 deletions(-) diff --git a/backend/plugins/circleci/e2e/job_collector_test.go b/backend/plugins/circleci/e2e/job_collector_test.go index f3988b48a9f..f27963e0c43 100644 --- a/backend/plugins/circleci/e2e/job_collector_test.go +++ b/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,36 +33,33 @@ 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() @@ -67,16 +67,61 @@ func TestCircleciUnfinishedJobsInputIterator(t *testing.T) { 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) +} diff --git a/backend/plugins/circleci/tasks/job_collector.go b/backend/plugins/circleci/tasks/job_collector.go index 2e950107a3b..730edaa9c8e 100644 --- a/backend/plugins/circleci/tasks/job_collector.go +++ b/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 } diff --git a/backend/plugins/circleci/tasks/shared.go b/backend/plugins/circleci/tasks/shared.go index f3be144dd14..69a92db0da4 100644 --- a/backend/plugins/circleci/tasks/shared.go +++ b/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 diff --git a/backend/plugins/circleci/tasks/workflow_collector.go b/backend/plugins/circleci/tasks/workflow_collector.go index 7847cfe2281..024917a85c1 100644 --- a/backend/plugins/circleci/tasks/workflow_collector.go +++ b/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 {