[Bug][Collector] Stateful collector silently discards the final page when ResponseParser returns items with ErrFinishCollect

[michaelgriskified]

Description

api_collector_stateful.go's ResponseParser wrapper discards the records returned by the
plugin's own parser whenever that parser signals ErrFinishCollect. The result is a silent,
page-aligned data loss at the timeAfter boundary: the last page of a time-bounded collection
is dropped in full, including the records on it that are inside the requested window.

Nothing errors. The task reports TASK_COMPLETED with zero failed subtasks, and every
internal consistency check passes, because the tool tables and the domain tables agree with
each other — they are both derived from the same truncated raw data. The loss is only visible
by comparing against the upstream API directly.

Root cause

backend/helpers/pluginhelper/api/api_collector_stateful.go:

ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
    items, err := args.CollectNewRecordsByList.ResponseParser(res)
    if err != nil {
        return nil, err          // <-- discards `items` even when err is ErrFinishCollect
    }

Plugin parsers are written to return the valid prefix of a page together with the stop signal.
For example backend/plugins/circleci/tasks/pipeline_collector.go:

for _, item := range data.Items {
    pipelineCreatedAt, err := extractCreatedAt(item)
    if err != nil { return nil, err }
    if pipelineCreatedAt.Before(*timeAfter) {
        return filteredItems, api.ErrFinishCollect   // in-window items + stop
    }
    filteredItems = append(filteredItems, item)
}

The wrapper replaces filteredItems with nil. ApiCollector.fetchAsync then takes the
count == 0 early return and never reaches db.Create, so the page is never persisted:

items, err := collector.args.ResponseParser(res)
if err != nil {
    if errors.Is(err, ErrFinishCollect) {
        logger.Info("a fetch stop by parser, reqInput: #%s", reqData.Params)
        handler = nil
    } ...
}
count := len(items)
if count == 0 {
    collector.args.Ctx.IncProgress(1)
    return nil          // <-- page silently dropped
}

Note fetchAsync itself handles this correctly — it persists whatever it is given before
stopping pagination. The loss is introduced purely by the wrapper nulling the slice.

This affects any plugin using NewStatefulApiCollectorForFinalizableEntity whose
ResponseParser returns a partial page with ErrFinishCollect, not only circleci.

Regression lineage

The circleci parser gained its return filteredItems, api.ErrFinishCollect line in #7820
(merged 2024-08-15), which fixed #7797 "CircleCI pipelines collected from before time range".

That plugin-side change is correct on its own terms — it stops collection at the window
boundary instead of running past it. But because the stateful wrapper nulls the slice whenever
the parser returns an error, the fix effectively traded collecting too much for silently
collecting too little. #7797's symptom was visible in the data; this one is not.

Impact

Loss is bounded by one page, so it is proportionally worst for low-volume scopes — the
opposite of where it is likely to be noticed. Measured on three repositories with
timeAfter = 90 days, PageSize = 20 (the CircleCI default):

repository	upstream API	collected	lost	loss
A (0.38 pipelines/day)	34	20	14	41%
B (3.1 pipelines/day)	276	260	16	5.8%
C (6.3 pipelines/day)	568	560*	8	1.4%

* predicted from the same model; C was collected after the workaround was applied and
returned the full 568.

Reproduction

1. Configure a CircleCI connection and a project scope for a repository with more pipelines in
   the window than one page (>20), where the window boundary falls mid-page.
2. Set a blueprint timeAfter such that the boundary page contains both in-window and
   out-of-window pipelines.
3. Run the blueprint. It completes TASK_COMPLETED, zero failed subtasks.
4. SELECT count(*) FROM _tool_circleci_pipelines WHERE project_slug = '...' returns an exact
   multiple of the page size.
5. Compare with GET /v2/project/{slug}/pipeline counted directly over the same window — the
   difference is the in-window records on the discarded page.

Confirming signals, all consistent:

• _raw_circleci_api_pipelines holds the same truncated count, so the loss is at collection,
  not extraction.
• Collected pipeline number values are perfectly contiguous — the tail is amputated, records
  are not dropped at random.
• A fullSync: true re-run reproduces the identical count.
• In-band detector: collectPipelines.finishedRecords counts pages fetched while
  extractPipelines.finishedRecords counts records kept; pages - kept/PageSize is the number
  of discarded pages. This agreed with the a fetch stop by parser log line on exactly the runs
  that lost data.

Suggested fix

Preserve the partial page when the stop signal is ErrFinishCollect:

items, err := args.CollectNewRecordsByList.ResponseParser(res)
if err != nil {
    if errors.Is(err, ErrFinishCollect) {
        return items, err
    }
    return nil, err
}

fetchAsync already persists the returned items and then stops paginating, so no other change
is needed.

Workaround for operators

Set timeAfter earlier than the window you actually need, so the discarded page falls in the
buffer rather than in the data you rely on. The buffer must exceed one page expressed in
time, which varies with each scope's rate: at PageSize 20 that is ~3 days for a repository
at 6.3 pipelines/day but ~53 days for one at 0.38/day. A fixed buffer chosen for busy scopes
will not protect quiet ones.

Environment

• DevLake v1.0.3-beta17 (8fe26f4), Helm chart 1.0.3-beta9
• PostgreSQL backend
• Plugins: circleci, github_graphql
• PIPELINE_MAX_PARALLEL=1, API_REQUESTS_PER_HOUR=3000