This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/devlake.git
The following commit(s) were added to refs/heads/main by this push:
new f3d8859fa fix(framework): keep the partial page when a parser signals
ErrFinishCollect (#9172)
f3d8859fa is described below
commit f3d8859fa4cc69cc60c5777e077f2e4cba5686a5
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Sat Sep 26 22:31:07 2026 -0400
fix(framework): keep the partial page when a parser signals
ErrFinishCollect (#9172)
Closes #9167.
The stateful collector's ResponseParser wrapper discarded the records a
plugin
parser returned whenever that parser signalled ErrFinishCollect:
items, err := args.CollectNewRecordsByList.ResponseParser(res)
if err != nil {
return nil, err
}
Plugin parsers return the valid prefix of the page they stopped on together
with
that signal, exactly as this file's own time filter does a few lines below
(`return validItems, ErrFinishCollect`). fetchAsync persists whatever comes
back
and sets handler to nil to stop paginating, so returning nil throws away
records
that are inside the requested window.
The result is a silent, page-aligned loss at the timeAfter boundary: the
last
page of a time-bounded collection is dropped in full. Nothing errors, the
task
reports TASK_COMPLETED, and the tool and domain tables agree with each other
because both derive from the same truncated raw data, so the loss is only
visible against the upstream API.
The sibling collectors already get this right. api_collector.go's fetchAsync
falls through to the save after logging "a fetch stop by parser", and
graphql_collector.go saves its rows before checking for the signal. The
stateful
wrapper, sitting between the plugin parser and fetchAsync, was the only
place
that dropped the page.
The wrapper is extracted to newStatefulResponseParser so the behaviour can
be
tested without a collector, an API client or a database. The move is
mechanical;
the only behaviour change is preserving the prefix. Passing the prefix
through
the time filter rather than returning it immediately keeps the incremental
dedup guarantee for the records that are kept.
Six tests: two fail on main with the records discarded, and four pin
behaviour
that must not change - real errors still discard the page, an empty finished
page still stops, a page with no signal passes through, and the time filter
still trims a page whose tail predates the last successful collection.
---
.../pluginhelper/api/api_collector_stateful.go | 123 ++++++++-------
.../api/api_collector_stateful_test.go | 165 +++++++++++++++++++++
2 files changed, 238 insertions(+), 50 deletions(-)
diff --git a/backend/helpers/pluginhelper/api/api_collector_stateful.go
b/backend/helpers/pluginhelper/api/api_collector_stateful.go
index d2b9e8245..2f0379709 100644
--- a/backend/helpers/pluginhelper/api/api_collector_stateful.go
+++ b/backend/helpers/pluginhelper/api/api_collector_stateful.go
@@ -167,56 +167,11 @@ func NewStatefulApiCollectorForFinalizableEntity(args
FinalizableApiCollectorArg
return nil, nil
},
MinTickInterval: args.CollectNewRecordsByList.MinTickInterval,
- ResponseParser: func(res *http.Response) ([]json.RawMessage,
errors.Error) {
- items, err :=
args.CollectNewRecordsByList.ResponseParser(res)
- if err != nil {
- return nil, err
- }
- if len(items) == 0 {
- return nil, nil
- }
-
- // time filter or diff sync
- if createdAfter != nil &&
args.CollectNewRecordsByList.GetCreated != nil {
- // if the first record of the page was created
before createdAfter and not a zero value, return empty set and stop
- firstCreated, err :=
args.CollectNewRecordsByList.GetCreated(items[0])
- if err != nil {
- return nil, err
- }
- if firstCreated.Before(*createdAfter) &&
!firstCreated.IsZero() {
- return nil, ErrFinishCollect
- }
-
- // If last record was created before
CreatedAfter, including a zero value, check each record individually
- lastCreated, err :=
args.CollectNewRecordsByList.GetCreated(items[len(items)-1])
- if err != nil {
- return nil, err
- }
- if lastCreated.Before(*createdAfter) {
- var validItems []json.RawMessage
- // Only collect items that were created
after the last successful collection to prevent duplicates
- for _, item := range items {
- itemCreatedAt, err :=
args.CollectNewRecordsByList.GetCreated(item)
- if err != nil {
- return nil, err
- }
-
- if itemCreatedAt.IsZero() {
- // If zero then
timestamp is null on the response - accept as valid for downstream processing
- validItems =
append(validItems, item)
- continue
- }
-
- if
itemCreatedAt.Before(*createdAfter) {
- // Once we reach an
item that was created before the last successful collection, stop & return
- return validItems,
ErrFinishCollect
- }
- validItems = append(validItems,
item)
- }
- }
- }
- return items, err
- },
+ ResponseParser: newStatefulResponseParser(
+ args.CollectNewRecordsByList.ResponseParser,
+ args.CollectNewRecordsByList.GetCreated,
+ createdAfter,
+ ),
AfterResponse: args.CollectNewRecordsByList.AfterResponse,
RequestBody: args.CollectNewRecordsByList.RequestBody,
Method: args.CollectNewRecordsByList.Method,
@@ -268,6 +223,74 @@ func NewStatefulApiCollectorForFinalizableEntity(args
FinalizableApiCollectorArg
return manager, err
}
+// newStatefulResponseParser wraps a plugin's ResponseParser with the
incremental time filter, so a
+// page is cut down to the records created after the last successful
collection.
+//
+// Both this filter and the plugin parsers it wraps stop collecting by
returning the valid prefix of
+// a page together with ErrFinishCollect. fetchAsync persists whatever is
returned and then stops
+// paginating, so the prefix must be passed through rather than dropped:
discarding it silently loses
+// the in-window records on the last page of a time-bounded collection.
+func newStatefulResponseParser(
+ parseResponse func(res *http.Response) ([]json.RawMessage,
errors.Error),
+ getCreated func(item json.RawMessage) (time.Time, errors.Error),
+ createdAfter *time.Time,
+) func(res *http.Response) ([]json.RawMessage, errors.Error) {
+ return func(res *http.Response) ([]json.RawMessage, errors.Error) {
+ items, err := parseResponse(res)
+ parserFinished := errors.Is(err, ErrFinishCollect)
+ if err != nil && !parserFinished {
+ return nil, err
+ }
+ if len(items) == 0 {
+ if parserFinished {
+ return nil, ErrFinishCollect
+ }
+ return nil, nil
+ }
+
+ // time filter or diff sync
+ if createdAfter != nil && getCreated != nil {
+ // if the first record of the page was created before
createdAfter and not a zero value, return empty set and stop
+ firstCreated, err := getCreated(items[0])
+ if err != nil {
+ return nil, err
+ }
+ if firstCreated.Before(*createdAfter) &&
!firstCreated.IsZero() {
+ return nil, ErrFinishCollect
+ }
+
+ // If last record was created before CreatedAfter,
including a zero value, check each record individually
+ lastCreated, err := getCreated(items[len(items)-1])
+ if err != nil {
+ return nil, err
+ }
+ if lastCreated.Before(*createdAfter) {
+ var validItems []json.RawMessage
+ // Only collect items that were created after
the last successful collection to prevent duplicates
+ for _, item := range items {
+ itemCreatedAt, err := getCreated(item)
+ if err != nil {
+ return nil, err
+ }
+
+ if itemCreatedAt.IsZero() {
+ // If zero then timestamp is
null on the response - accept as valid for downstream processing
+ validItems = append(validItems,
item)
+ continue
+ }
+
+ if itemCreatedAt.Before(*createdAfter) {
+ // Once we reach an item that
was created before the last successful collection, stop & return
+ return validItems,
ErrFinishCollect
+ }
+ validItems = append(validItems, item)
+ }
+ }
+ }
+ return items, err
+ }
+}
+
type FinalizableApiCollectorArgs struct {
RawDataSubTaskArgs
ApiClient RateLimitedApiClient
diff --git a/backend/helpers/pluginhelper/api/api_collector_stateful_test.go
b/backend/helpers/pluginhelper/api/api_collector_stateful_test.go
new file mode 100644
index 000000000..cbdd37acc
--- /dev/null
+++ b/backend/helpers/pluginhelper/api/api_collector_stateful_test.go
@@ -0,0 +1,165 @@
+/*
+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 api
+
+import (
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/apache/devlake/core/errors"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func rawItems(values ...string) []json.RawMessage {
+ items := make([]json.RawMessage, 0, len(values))
+ for _, value := range values {
+ items = append(items, json.RawMessage(value))
+ }
+ return items
+}
+
+// A plugin parser stops collecting by returning the valid prefix of the page
it stopped on together
+// with ErrFinishCollect. Those records are inside the requested window, so
they must survive: dropping
+// them loses the last page of a time-bounded collection in full, and nothing
reports an error because
+// the task still completes.
+func TestStatefulResponseParserKeepsPartialPageOnFinish(t *testing.T) {
+ kept := rawItems(`{"id":1}`, `{"id":2}`)
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return kept, ErrFinishCollect
+ },
+ nil,
+ nil,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Equal(t, kept, items)
+ assert.True(t, errors.Is(err, ErrFinishCollect), "the stop signal must
still reach fetchAsync")
+}
+
+// The same, with the incremental filter active: the prefix is in the window,
so the filter keeps it.
+func TestStatefulResponseParserKeepsPartialPageWhileFiltering(t *testing.T) {
+ createdAfter := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ kept := rawItems(`{"created":"2026-03-01"}`, `{"created":"2026-02-01"}`)
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return kept, ErrFinishCollect
+ },
+ func(item json.RawMessage) (time.Time, errors.Error) {
+ var record struct {
+ Created string `json:"created"`
+ }
+ if e := json.Unmarshal(item, &record); e != nil {
+ return time.Time{}, errors.Convert(e)
+ }
+ parsed, e := time.Parse("2006-01-02", record.Created)
+ return parsed, errors.Convert(e)
+ },
+ &createdAfter,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Equal(t, kept, items)
+ assert.True(t, errors.Is(err, ErrFinishCollect))
+}
+
+// An empty page plus the stop signal has nothing to keep, but must still stop.
+func TestStatefulResponseParserStopsOnEmptyFinishedPage(t *testing.T) {
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return nil, ErrFinishCollect
+ },
+ nil,
+ nil,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Empty(t, items)
+ assert.True(t, errors.Is(err, ErrFinishCollect))
+}
+
+// A genuine failure is still a failure, and must not be mistaken for a stop
signal.
+func TestStatefulResponseParserPropagatesRealErrors(t *testing.T) {
+ boom := errors.Default.New("boom")
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return rawItems(`{"id":1}`), boom
+ },
+ nil,
+ nil,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Nil(t, items, "records must be discarded when the page could not
be parsed")
+ assert.Equal(t, boom, err)
+ assert.False(t, errors.Is(err, ErrFinishCollect))
+}
+
+func TestStatefulResponseParserPassesThroughWhenNotFinished(t *testing.T) {
+ all := rawItems(`{"id":1}`, `{"id":2}`)
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return all, nil
+ },
+ nil,
+ nil,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Equal(t, all, items)
+ assert.Nil(t, err)
+}
+
+// Guards the extraction: the incremental filter still trims a page whose tail
predates the last
+// successful collection, and still stops once it reaches such a record.
+func TestStatefulResponseParserTrimsRecordsBeforeCreatedAfter(t *testing.T) {
+ createdAfter := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC)
+ parser := newStatefulResponseParser(
+ func(*http.Response) ([]json.RawMessage, errors.Error) {
+ return rawItems(
+ `{"created":"2026-03-01"}`,
+ `{"created":"2026-02-15"}`,
+ `{"created":"2026-01-20"}`,
+ ), nil
+ },
+ func(item json.RawMessage) (time.Time, errors.Error) {
+ var record struct {
+ Created string `json:"created"`
+ }
+ if e := json.Unmarshal(item, &record); e != nil {
+ return time.Time{}, errors.Convert(e)
+ }
+ parsed, e := time.Parse("2006-01-02", record.Created)
+ return parsed, errors.Convert(e)
+ },
+ &createdAfter,
+ )
+
+ items, err := parser(&http.Response{})
+
+ assert.Equal(t, rawItems(`{"created":"2026-03-01"}`,
`{"created":"2026-02-15"}`), items)
+ assert.True(t, errors.Is(err, ErrFinishCollect))
+}
