From 3c7b9d0e99ddcba42e2e8630f67385ab0926253d Mon Sep 17 00:00:00 2001 From: Sadat Nazarli <118483002+dvecy@users.noreply.github.com> Date: Fri, 23 Oct 2026 09:59:44 +0200 Subject: [PATCH] fix(github): use the tool record's updated_at as the convert watermark `stateManager.GetSince()` is `_devlake_subtask_state.prev_started_at`, the time the previous run of the subtask started. The GitHub convert subtasks compared it against `github_updated_at`, which is when GitHub last changed the record. Those are two different clocks, so a tool record that DevLake wrote or corrected after the previous run was skipped whenever GitHub had not touched it since then, on that run and on every run after it. The tool layer stayed correct while the domain record was missing or stuck on its old status. A closed pull request from 2019 whose `merged` flag a later collection filled in is the clearest case: `github_updated_at` is 2019, `since` is the previous run a few hours ago, so the record never reaches `pull_requests`. Filter on the tool record's own `updated_at` instead. Gorm maintains it on every write, so it means "DevLake last wrote this row", which is what the watermark is. This is what the other plugins already do, what the example in `data_convertor_stateful.go` documents, and what the remaining GitHub convert subtasks (repo, release, account, cicd job) already do. Covers the subtasks whose watermark is on their own table: pull requests, issues, pull request comments, issue comments and runs. The seven that gate on a joined parent table are left alone, since which table should carry their watermark is a separate question. Existing domain records that were already skipped are not healed by this; they need one full sync or re-transform. Closes #9150 Signed-off-by: Sadat Nazarli <118483002+dvecy@users.noreply.github.com> --- .../e2e/pr_convertor_incremental_test.go | 167 ++++++++++++++++++ .../github/tasks/cicd_run_convertor.go | 2 +- .../github/tasks/issue_comment_convertor.go | 2 +- .../plugins/github/tasks/issue_convertor.go | 2 +- .../github/tasks/pr_comment_convertor.go | 2 +- backend/plugins/github/tasks/pr_convertor.go | 6 +- 6 files changed, 176 insertions(+), 5 deletions(-) create mode 100644 backend/plugins/github/e2e/pr_convertor_incremental_test.go diff --git a/backend/plugins/github/e2e/pr_convertor_incremental_test.go b/backend/plugins/github/e2e/pr_convertor_incremental_test.go new file mode 100644 index 00000000000..38f72be93cd --- /dev/null +++ b/backend/plugins/github/e2e/pr_convertor_incremental_test.go @@ -0,0 +1,167 @@ +/* +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 e2e + +import ( + "context" + "sort" + "testing" + "time" + + "github.com/apache/devlake/core/dal" + coreModels "github.com/apache/devlake/core/models" + "github.com/apache/devlake/core/models/domainlayer/code" + "github.com/apache/devlake/core/plugin" + "github.com/apache/devlake/core/runner" + "github.com/apache/devlake/core/utils" + "github.com/apache/devlake/helpers/e2ehelper" + contextimpl "github.com/apache/devlake/impls/context" + "github.com/apache/devlake/plugins/github/impl" + "github.com/apache/devlake/plugins/github/models" + "github.com/apache/devlake/plugins/github/tasks" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// runSubtaskIncrementally executes one subtask with the subtask state left in place and a +// sync policy that permits incremental mode, which DataFlowTester.Subtask cannot do: it +// flushes _devlake_subtask_state and forces FullSync on every call +// (helpers/e2ehelper/data_flow_tester.go). Same approach as +// plugins/youtrack/e2e/stateful_scenario_test.go, with the subtask and plugin names passed +// in their own positions so the state key matches the one the subtask itself computes. +func runSubtaskIncrementally( + t *testing.T, + dataflowTester *e2ehelper.DataFlowTester, + subtaskMeta plugin.SubTaskMeta, + taskData interface{}, +) { + t.Helper() + subtaskCtx := contextimpl.NewStandaloneSubTaskContext( + context.Background(), + runner.CreateBasicRes(dataflowTester.Cfg, dataflowTester.Log, dataflowTester.Db), + subtaskMeta.Name, + taskData, + dataflowTester.Name, + &coreModels.SyncPolicy{}, + ) + require.NoError(t, subtaskMeta.EntryPoint(subtaskCtx)) +} + +// TestGithubPrConvertorIncremental covers the incremental filter of "Convert Pull Requests". +// +// `stateManager.GetSince()` is the start time of the previous successful run of the subtask +// (`_devlake_subtask_state.prev_started_at`), so the filter has to select the tool records +// DevLake wrote since then, i.e. the ones whose own `updated_at` is at or after it. +// +// Pull request 29 is the regression guard: GitHub last touched it in 2019, DevLake rewrote +// its row after the previous run (`state` and `merged` were corrected), and the domain record +// still has to be updated. Filtering on `github_updated_at` compared that 2019 timestamp +// against a 2026 run time and skipped the record on this run and on every run after it, which +// left the domain record missing or stuck on its old status (#9150). +func TestGithubPrConvertorIncremental(t *testing.T) { + var githubPlugin impl.Github + dataflowTester := e2ehelper.NewDataFlowTester(t, "github", githubPlugin) + + taskData := &tasks.GithubTaskData{ + Options: &tasks.GithubOptions{ + ConnectionId: 1, + Name: "panjf2000/ants", + GithubId: 134018330, + }, + } + + prevStartedAt := time.Date(2026, 9, 16, 16, 24, 0, 0, time.UTC) + // GitHub's timestamp for pull request 29, older than every DevLake run. + collectedLongAgo := time.Date(2019, 8, 8, 8, 31, 19, 0, time.UTC) + writtenAfterPrevRun := time.Date(2026, 9, 17, 10, 0, 0, 0, time.UTC) + writtenBeforePrevRun := time.Date(2026, 9, 15, 10, 0, 0, 0, time.UTC) + + dataflowTester.FlushTabler(&models.GithubPullRequest{}) + dataflowTester.FlushTabler(&code.PullRequest{}) + dataflowTester.FlushTabler(&coreModels.SubtaskState{}) + + // A pull request GitHub last updated in 2019 whose row DevLake rewrote after the previous + // run, one updated on GitHub after the previous run, and one untouched since before it. + seedPullRequest(t, dataflowTester, 101, 29, "closed", true, collectedLongAgo, writtenAfterPrevRun) + seedPullRequest(t, dataflowTester, 102, 30, "closed", true, writtenAfterPrevRun, writtenAfterPrevRun) + seedPullRequest(t, dataflowTester, 103, 31, "open", false, writtenBeforePrevRun, writtenBeforePrevRun) + + // Previous successful run of this subtask. PrevConfig is left empty so the subtask is not + // treated as reconfigured, which would force full sync. + require.NoError(t, dataflowTester.Dal.CreateOrUpdate(&coreModels.SubtaskState{ + Plugin: "github", + Subtask: tasks.ConvertPullRequestsMeta.Name, + Params: utils.ToJsonString(tasks.GithubApiParams{ConnectionId: 1, Name: "panjf2000/ants"}), + PrevStartedAt: &prevStartedAt, + })) + + runSubtaskIncrementally(t, dataflowTester, tasks.ConvertPullRequestsMeta, taskData) + + var domainPullRequests []code.PullRequest + require.NoError(t, dataflowTester.Dal.All(&domainPullRequests)) + converted := make([]int, len(domainPullRequests)) + statuses := map[int]string{} + for i, domainPullRequest := range domainPullRequests { + converted[i] = domainPullRequest.PullRequestKey + statuses[domainPullRequest.PullRequestKey] = domainPullRequest.Status + } + sort.Ints(converted) + + assert.Equal(t, []int{ + 29, // rewritten by DevLake after the previous run, last updated on GitHub in 2019 + 30, // updated on GitHub after the previous run + }, converted) + assert.Equal(t, code.MERGED, statuses[29], "the corrected status has to reach the domain layer") + assert.NotContains(t, converted, 31, "records untouched since before the previous run must be skipped") +} + +// seedPullRequest inserts a tool-layer pull request and then forces `updated_at`, which gorm +// would otherwise set to the time of the insert. +func seedPullRequest( + t *testing.T, + dataflowTester *e2ehelper.DataFlowTester, + githubId int, + number int, + state string, + merged bool, + githubUpdatedAt time.Time, + updatedAt time.Time, +) { + t.Helper() + pullRequest := &models.GithubPullRequest{ + ConnectionId: 1, + GithubId: githubId, + RepoId: 134018330, + Number: number, + State: state, + Merged: merged, + GithubCreatedAt: githubUpdatedAt, + GithubUpdatedAt: githubUpdatedAt, + } + if merged { + pullRequest.MergedAt = &githubUpdatedAt + pullRequest.ClosedAt = &githubUpdatedAt + } + require.NoError(t, dataflowTester.Dal.CreateOrUpdate(pullRequest)) + require.NoError(t, dataflowTester.Dal.UpdateColumn( + &models.GithubPullRequest{}, + "updated_at", + updatedAt, + dal.Where("connection_id = ? AND github_id = ?", 1, githubId), + )) +} diff --git a/backend/plugins/github/tasks/cicd_run_convertor.go b/backend/plugins/github/tasks/cicd_run_convertor.go index 62f6a355a4f..48876a58562 100644 --- a/backend/plugins/github/tasks/cicd_run_convertor.go +++ b/backend/plugins/github/tasks/cicd_run_convertor.go @@ -83,7 +83,7 @@ func ConvertRuns(taskCtx plugin.SubTaskContext) errors.Error { if stateManager.IsIncremental() { since := stateManager.GetSince() if since != nil { - clauses = append(clauses, dal.Where("github_updated_at >= ?", since)) + clauses = append(clauses, dal.Where("updated_at >= ?", since)) } } return db.Cursor(clauses...) diff --git a/backend/plugins/github/tasks/issue_comment_convertor.go b/backend/plugins/github/tasks/issue_comment_convertor.go index 80180b3e015..43c9747420b 100644 --- a/backend/plugins/github/tasks/issue_comment_convertor.go +++ b/backend/plugins/github/tasks/issue_comment_convertor.go @@ -73,7 +73,7 @@ func ConvertIssueComments(taskCtx plugin.SubTaskContext) errors.Error { if stateManager.IsIncremental() { since := stateManager.GetSince() if since != nil { - clauses = append(clauses, dal.Where("_tool_github_issue_comments.github_updated_at >= ?", since)) + clauses = append(clauses, dal.Where("_tool_github_issue_comments.updated_at >= ?", since)) } } return db.Cursor(clauses...) diff --git a/backend/plugins/github/tasks/issue_convertor.go b/backend/plugins/github/tasks/issue_convertor.go index 8184c1343b1..7adef674089 100644 --- a/backend/plugins/github/tasks/issue_convertor.go +++ b/backend/plugins/github/tasks/issue_convertor.go @@ -88,7 +88,7 @@ func ConvertIssues(taskCtx plugin.SubTaskContext) errors.Error { if stateManager.IsIncremental() { since := stateManager.GetSince() if since != nil { - clauses = append(clauses, dal.Where("github_updated_at >= ?", since)) + clauses = append(clauses, dal.Where("updated_at >= ?", since)) } } return db.Cursor(clauses...) diff --git a/backend/plugins/github/tasks/pr_comment_convertor.go b/backend/plugins/github/tasks/pr_comment_convertor.go index 23c29e123a9..30e57f84df9 100644 --- a/backend/plugins/github/tasks/pr_comment_convertor.go +++ b/backend/plugins/github/tasks/pr_comment_convertor.go @@ -76,7 +76,7 @@ func ConvertPullRequestComments(taskCtx plugin.SubTaskContext) errors.Error { if stateManager.IsIncremental() { since := stateManager.GetSince() if since != nil { - clauses = append(clauses, dal.Where("_tool_github_pull_request_comments.github_updated_at >= ?", since)) + clauses = append(clauses, dal.Where("_tool_github_pull_request_comments.updated_at >= ?", since)) } } return db.Cursor(clauses...) diff --git a/backend/plugins/github/tasks/pr_convertor.go b/backend/plugins/github/tasks/pr_convertor.go index 3b39aca77ae..b5b1a01078e 100644 --- a/backend/plugins/github/tasks/pr_convertor.go +++ b/backend/plugins/github/tasks/pr_convertor.go @@ -72,7 +72,11 @@ func ConvertPullRequests(taskCtx plugin.SubTaskContext) errors.Error { if stateManager.IsIncremental() { since := stateManager.GetSince() if since != nil { - clauses = append(clauses, dal.Where("github_updated_at >= ?", since)) + // `since` is when the previous run of this subtask started, so the filter has to + // select the records DevLake wrote after it, not the ones GitHub changed after it: + // a record whose `github_updated_at` is older than `since` still has to be + // converted when a later collection rewrote it. + clauses = append(clauses, dal.Where("updated_at >= ?", since)) } } return db.Cursor(clauses...)