From 5fabfe279508e1e60fea701f2a743ab6c69b6210 Mon Sep 17 00:00:00 2001 From: Shivam Sharma <192242476+ShivamSharma43@users.noreply.github.com> Date: Tue, 10 Nov 2026 23:10:16 +0530 Subject: [PATCH] fix(circleci): collect pipelines when the sync policy has no timeAfter CollectPipelines dereferenced SyncPolicy().TimeAfter for every pipeline in the response. TimeAfter is nil for advanced mode blueprints, when the start date is cleared, and for directrun without timeAfter, so the response parser panicked and the subtask failed with a nil pointer dereference. Move the response parsing into parsePipelines, keep every pipeline when timeAfter is nil, and check that SyncPolicy() isn't nil before reading it. Closes #9367 Generated-by: Claude Code (Claude Opus 5.5) Co-Authored-By: Claude Opus 5.5 Signed-off-by: Shivam Sharma <192242476+ShivamSharma43@users.noreply.github.com> --- .../circleci/tasks/pipeline_collector.go | 53 ++++++++++------ .../circleci/tasks/pipeline_collector_test.go | 63 +++++++++++++++++++ 2 files changed, 96 insertions(+), 20 deletions(-) create mode 100644 backend/plugins/circleci/tasks/pipeline_collector_test.go diff --git a/backend/plugins/circleci/tasks/pipeline_collector.go b/backend/plugins/circleci/tasks/pipeline_collector.go index fc981f99979..96c56c2f838 100644 --- a/backend/plugins/circleci/tasks/pipeline_collector.go +++ b/backend/plugins/circleci/tasks/pipeline_collector.go @@ -20,6 +20,7 @@ package tasks import ( "encoding/json" "net/http" + "time" "github.com/apache/devlake/core/errors" "github.com/apache/devlake/core/plugin" @@ -41,7 +42,10 @@ var CollectPipelinesMeta = plugin.SubTaskMeta{ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_TABLE) logger := taskCtx.GetLogger() - timeAfter := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy().TimeAfter + var timeAfter *time.Time + if syncPolicy := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy(); syncPolicy != nil { + timeAfter = syncPolicy.TimeAfter + } logger.Info("collect pipelines") collector, err := api.NewStatefulApiCollectorForFinalizableEntity(api.FinalizableApiCollectorArgs{ RawDataSubTaskArgs: *rawDataSubTaskArgs, @@ -53,25 +57,7 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { UrlTemplate: "/v2/project/{{ .Params.ProjectSlug }}/pipeline", Query: BuildQueryParamsWithPageToken, ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) { - data := CircleciPageTokenResp[[]json.RawMessage]{} - err := api.UnmarshalResponse(res, &data) - - if err != nil { - return nil, err - } - filteredItems := []json.RawMessage{} - for _, item := range data.Items { - pipelineCreatedAt, err := extractCreatedAt(item) - - if err != nil { - return nil, err - } - if pipelineCreatedAt.Before(*timeAfter) { - return filteredItems, api.ErrFinishCollect - } - filteredItems = append(filteredItems, item) - } - return filteredItems, nil + return parsePipelines(res, timeAfter) }, }, GetCreated: extractCreatedAt, @@ -83,3 +69,30 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error { } return collector.Execute() } + +// parsePipelines returns the pipelines of a page. CircleCI lists pipelines +// newest first, so the collection ends at the first pipeline created before +// timeAfter. timeAfter is nil when the sync policy has no time range (e.g. an +// advanced mode blueprint), and then every pipeline is kept. +func parsePipelines(res *http.Response, timeAfter *time.Time) ([]json.RawMessage, errors.Error) { + data := CircleciPageTokenResp[[]json.RawMessage]{} + err := api.UnmarshalResponse(res, &data) + if err != nil { + return nil, err + } + if timeAfter == nil { + return data.Items, nil + } + filteredItems := []json.RawMessage{} + for _, item := range data.Items { + pipelineCreatedAt, err := extractCreatedAt(item) + if err != nil { + return nil, err + } + if pipelineCreatedAt.Before(*timeAfter) { + return filteredItems, api.ErrFinishCollect + } + filteredItems = append(filteredItems, item) + } + return filteredItems, nil +} diff --git a/backend/plugins/circleci/tasks/pipeline_collector_test.go b/backend/plugins/circleci/tasks/pipeline_collector_test.go new file mode 100644 index 00000000000..b35c6435dce --- /dev/null +++ b/backend/plugins/circleci/tasks/pipeline_collector_test.go @@ -0,0 +1,63 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to You under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package tasks + +import ( + "bytes" + "io" + "net/http" + "testing" + "time" + + "github.com/apache/devlake/helpers/pluginhelper/api" + "github.com/stretchr/testify/assert" +) + +const pipelinesPage = `{ + "items": [ + {"id": "p3", "created_at": "2026-03-01T00:00:00Z"}, + {"id": "p2", "created_at": "2026-02-01T00:00:00Z"}, + {"id": "p1", "created_at": "2026-01-01T00:00:00Z"} + ], + "next_page_token": "token" +}` + +func makePipelinesResponse() *http.Response { + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(bytes.NewBufferString(pipelinesPage)), + Request: &http.Request{}, + } +} + +func TestParsePipelinesWithoutTimeAfter(t *testing.T) { + // An advanced mode blueprint has no timeAfter: keep every pipeline + // instead of dereferencing a nil time. + items, err := parsePipelines(makePipelinesResponse(), nil) + assert.Nil(t, err) + assert.Len(t, items, 3) +} + +func TestParsePipelinesStopsBeforeTimeAfter(t *testing.T) { + timeAfter := time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC) + items, err := parsePipelines(makePipelinesResponse(), &timeAfter) + assert.ErrorIs(t, err, api.ErrFinishCollect) + assert.Len(t, items, 2) + assert.Contains(t, string(items[0]), `"p3"`) + assert.Contains(t, string(items[1]), `"p2"`) +}