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 56ca34739 fix(github): use the tool record's updated_at as the convert
watermark (#9250)
56ca34739 is described below
commit 56ca34739100a064125ce4c6c2ad16c3939c67a2
Author: Sadat <[email protected]>
AuthorDate: Mon Oct 26 17:17:44 2026 +0100
fix(github): use the tool record's updated_at as the convert watermark
(#9250)
`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 <[email protected]>
---
.../github/e2e/pr_convertor_incremental_test.go | 167 +++++++++++++++++++++
backend/plugins/github/tasks/cicd_run_convertor.go | 2 +-
.../github/tasks/issue_comment_convertor.go | 2 +-
backend/plugins/github/tasks/issue_convertor.go | 2 +-
.../plugins/github/tasks/pr_comment_convertor.go | 2 +-
backend/plugins/github/tasks/pr_convertor.go | 6 +-
6 files changed, 176 insertions(+), 5 deletions(-)
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 000000000..38f72be93
--- /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 62f6a355a..48876a585 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 80180b3e0..43c974742 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 8184c1343..7adef6740 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 23c29e123..30e57f84d 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 3b39aca77..b5b1a0107 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...)
