fix(youtrack): preserve incremental state and replace stale data

Align changed-issue windows for comments and changelogs, keep only the latest issue and workflow-state snapshots on replay, and remove domain comments that were deleted upstream. Make API retries cancellation-aware with a single retry owner, and extend e2e scenarios and fixtures to cover these transitions.

Skip 404s from the per-issue comment and activity endpoints so an issue deleted upstream no longer fails the subtask and freezes its bookmark. Delete the scope's domain issue labels and assignees before converting, so they are replaced even when a run emits none.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: Evgenii Davydov

backend/plugins/youtrack/e2e/changelog_test.go

@@ -44,6 +44,10 @@ import (
 func TestYoutrackIssueChangelogDataFlow(t *testing.T) {
 	var youtrack impl.Youtrack
 	dataflowTester := e2ehelper.NewDataFlowTester(t, "youtrack", youtrack)
+	// stateful scenario: the rerun at the end must exercise the real
+	// config-change transition against persisted state, not the tester's
+	// forced full sync
+	beginStatefulScenario(t, dataflowTester)
 
 	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issues.csv", "_raw_youtrack_issues")
 	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issue_changelogs.csv", "_raw_youtrack_issue_changelogs")
@@ -53,8 +57,8 @@ func TestYoutrackIssueChangelogDataFlow(t *testing.T) {
 
 	for _, projectId := range []string{"0-1", "0-2"} {
 		taskData := newTaskData(projectId, zeroConfig())
-		dataflowTester.Subtask(tasks.ExtractIssuesMeta, taskData)
-		dataflowTester.Subtask(tasks.ExtractIssueChangelogsMeta, taskData)
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssuesMeta, taskData)
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssueChangelogsMeta, taskData)
 	}
 
 	dataflowTester.VerifyTableWithOptions(models.YoutrackIssueChangelog{}, e2ehelper.TableOptions{
@@ -115,7 +119,7 @@ func TestYoutrackIssueChangelogDataFlow(t *testing.T) {
 	dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_youtrack_workflow_states.csv", &models.YoutrackWorkflowState{})
 	dataflowTester.FlushTabler(&ticket.IssueChangelogs{})
 	for _, projectId := range []string{"0-1", "0-2"} {
-		dataflowTester.Subtask(tasks.ConvertIssueChangelogsMeta, newTaskData(projectId, zeroConfig()))
+		runSubtaskPreservingState(t, dataflowTester, tasks.ConvertIssueChangelogsMeta, newTaskData(projectId, zeroConfig()))
 	}
 
 	dataflowTester.VerifyTableWithOptions(ticket.IssueChangelogs{}, e2ehelper.TableOptions{
@@ -194,14 +198,15 @@ func TestYoutrackIssueChangelogDataFlow(t *testing.T) {
 	assert.Positive(t, domainCountWhere("author_name != ''"), "AuthorName carries the display name")
 
 	// re-walks are idempotent via upsert by activity id: the
-	// config change flips the extractor to full sync, so every
-	// raw row is reprocessed — the counts must not move
+	// config change against the PRESERVED state flips the extractor to full
+	// sync — no collection involved — so every raw row is
+	// reprocessed and the counts must not move
 	reextract := zeroConfig()
 	reextract.StoryPointField = "Days in stage"
 	for _, projectId := range []string{"0-1", "0-2"} {
 		taskData := newTaskData(projectId, reextract)
-		dataflowTester.Subtask(tasks.ExtractIssueChangelogsMeta, taskData)
-		dataflowTester.Subtask(tasks.ConvertIssueChangelogsMeta, taskData)
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssueChangelogsMeta, taskData)
+		runSubtaskPreservingState(t, dataflowTester, tasks.ConvertIssueChangelogsMeta, taskData)
 	}
 	assert.Equal(t, toolCount, countWhere(&models.YoutrackIssueChangelog{}, "1 = 1"),
 		"a full re-extraction upserts by activity id — no duplicate tool rows")

backend/plugins/youtrack/e2e/comment_test.go

@@ -146,3 +146,45 @@ func TestYoutrackCommentDataFlow(t *testing.T) {
 	require.NoError(t, dataflowTester.Db.Model(&crossdomain.Account{}).Count(&domainAccounts).Error)
 	assert.Positive(t, domainAccounts)
 }
+
+// TestYoutrackCommentAllDeletedReplacement covers the replacement edge the
+// lazy divider cleanup cannot reach: a scope whose comments have ALL become
+// deleted emits zero output rows, and the previously converted domain rows
+// must still disappear (deleted comments leave domain metrics).
+func TestYoutrackCommentAllDeletedReplacement(t *testing.T) {
+	var youtrack impl.Youtrack
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "youtrack", youtrack)
+
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issues.csv", "_raw_youtrack_issues")
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issue_comments.csv", "_raw_youtrack_issue_comments")
+	dataflowTester.FlushTabler(&models.YoutrackIssue{})
+	dataflowTester.FlushTabler(&models.YoutrackIssueComment{})
+	dataflowTester.FlushTabler(&ticket.IssueComment{})
+
+	for _, projectId := range []string{"0-1", "0-2"} {
+		taskData := newTaskData(projectId, zeroConfig())
+		dataflowTester.Subtask(tasks.ExtractIssuesMeta, taskData)
+		dataflowTester.Subtask(tasks.ExtractCommentsMeta, taskData)
+	}
+	for _, projectId := range []string{"0-1", "0-2"} {
+		dataflowTester.Subtask(tasks.ConvertCommentsMeta, newTaskData(projectId, zeroConfig()))
+	}
+
+	domainCount := func() int64 {
+		var n int64
+		require.NoError(t, dataflowTester.Db.Model(&ticket.IssueComment{}).Count(&n).Error)
+		return n
+	}
+	require.Positive(t, domainCount(), "populated scope converts comments into the domain layer")
+
+	// every comment becomes deleted upstream (the tool layer keeps the rows,
+	// flag set — exactly what the next incremental collection would extract)
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssueComment{}).
+		Where("connection_id = 1").Update("deleted", true).Error)
+
+	for _, projectId := range []string{"0-1", "0-2"} {
+		dataflowTester.Subtask(tasks.ConvertCommentsMeta, newTaskData(projectId, zeroConfig()))
+	}
+	assert.Zero(t, domainCount(),
+		"conversion emitting zero rows must still replace the scope's previous domain output")
+}

backend/plugins/youtrack/e2e/issue_mapping_test.go

@@ -18,7 +18,9 @@ limitations under the License.
 package e2e
 
 import (
+	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/apache/devlake/core/models/common"
 	"github.com/apache/devlake/core/models/domainlayer/ticket"
@@ -130,6 +132,120 @@ func TestYoutrackIssueDataFlowWithMappings(t *testing.T) {
 	assert.Zero(t, storyAsRequirement, "no row keeps the stale mapping")
 }
 
+// TestYoutrackMappingReplayPreservesState covers the transition the
+// DataFlowTester.Subtask reruns above cannot: Subtask deletes subtask state
+// and forces FullSync, so it can never verify that a config change
+// invalidates PERSISTED incremental state. This scenario keeps the state:
+//
+//	run 1  initial extraction (full sync: no prior state)
+//	run 2  incremental update — one issue re-collected without its tag
+//	run 3  mapping change — full re-extraction replayed from the raw layer
+//
+// asserting (a) the incremental run processes only the new raw version,
+// (b) the replay never re-hits the API and never resurrects the removed
+// tag from the retained old version, and (c) the new mapping applies.
+func TestYoutrackMappingReplayPreservesState(t *testing.T) {
+	var youtrack impl.Youtrack
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "youtrack", youtrack)
+	beginStatefulScenario(t, dataflowTester)
+
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issues.csv", "_raw_youtrack_issues")
+	dataflowTester.FlushTabler(&models.YoutrackIssue{})
+	dataflowTester.FlushTabler(&models.YoutrackIssueLabel{})
+	dataflowTester.FlushTabler(&models.YoutrackAccount{})
+
+	// run 1: initial full extraction
+	for _, projectId := range []string{"0-1", "0-2"} {
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssuesMeta, newTaskData(projectId, mappedConfig()))
+	}
+
+	// the fixture's tagged issue in project 0-1 (PROJ1-1, tag "Долго в работе")
+	const taggedIssue = "2-1"
+	countLabels := func(issueId string) int64 {
+		var n int64
+		require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssueLabel{}).
+			Where("connection_id = 1 AND issue_id = ?", issueId).Count(&n).Error)
+		return n
+	}
+	var totalLabels int64
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssueLabel{}).
+		Where("connection_id = 1").Count(&totalLabels).Error)
+	require.Equal(t, int64(1), countLabels(taggedIssue), "run 1 extracts the fixture's tag")
+	require.Positive(t, totalLabels)
+
+	// the incremental update: the same issue re-collected tagless, updated later
+	var rawCountBefore int64
+	require.NoError(t, dataflowTester.Db.Table("_raw_youtrack_issues").Count(&rawCountBefore).Error)
+	type rawIssueRow struct {
+		Params string
+		Data   []byte
+		Url    string
+	}
+	var rows []rawIssueRow
+	require.NoError(t, dataflowTester.Db.Table("_raw_youtrack_issues").
+		Where("params LIKE ?", `%0-1%`).Find(&rows).Error)
+	var src rawIssueRow
+	var v2data []byte
+	for _, row := range rows {
+		var issue map[string]interface{}
+		require.NoError(t, json.Unmarshal(row.Data, &issue))
+		if issue["id"] != taggedIssue {
+			continue
+		}
+		issue["tags"] = []interface{}{}
+		issue["updated"] = issue["updated"].(float64) + 60000
+		v2, err := json.Marshal(issue)
+		require.NoError(t, err)
+		v2data = v2
+		src = row
+		break
+	}
+	require.NotNil(t, v2data, "the fixture must contain issue %s in project 0-1", taggedIssue)
+	require.NoError(t, dataflowTester.Db.Table("_raw_youtrack_issues").Create(map[string]interface{}{
+		"params":     src.Params,
+		"data":       v2data,
+		"url":        src.Url,
+		"created_at": time.Now(), // inside run 2's incremental window
+	}).Error)
+
+	// run 2: same config, state preserved -> incremental — only the new
+	// version is processed, the tag disappears
+	for _, projectId := range []string{"0-1", "0-2"} {
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssuesMeta, newTaskData(projectId, mappedConfig()))
+	}
+	assert.Zero(t, countLabels(taggedIssue), "incremental re-extraction replaces the issue's labels")
+	assert.Equal(t, totalLabels-1, func() int64 {
+		var n int64
+		require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssueLabel{}).
+			Where("connection_id = 1").Count(&n).Error)
+		return n
+	}(), "every other issue's labels are untouched")
+
+	// run 3: a mapping change must invalidate the persisted incremental
+	// state -> full re-extraction replayed from the raw layer. The retained
+	// v1 still carries the tag; only latest-version expansion keeps it dead.
+	remapped := mappedConfig()
+	remapped.TypeMappings = map[string]string{"Story": ticket.TASK, "Bug": ticket.BUG}
+	for _, projectId := range []string{"0-1", "0-2"} {
+		runSubtaskPreservingState(t, dataflowTester, tasks.ExtractIssuesMeta, newTaskData(projectId, remapped))
+	}
+	assert.Zero(t, countLabels(taggedIssue),
+		"mapping replay must not resurrect the tag from the retained old raw version")
+
+	var storyAsTask, storyAsRequirement int64
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssue{}).
+		Where("connection_id = 1 AND type_name = 'Story' AND std_type = ?", ticket.TASK).Count(&storyAsTask).Error)
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssue{}).
+		Where("connection_id = 1 AND type_name = 'Story' AND std_type = ?", ticket.REQUIREMENT).Count(&storyAsRequirement).Error)
+	assert.Positive(t, storyAsTask, "the changed mapping re-applies from the raw layer")
+	assert.Zero(t, storyAsRequirement, "no row keeps the stale mapping")
+
+	var rawCountAfter int64
+	require.NoError(t, dataflowTester.Db.Table("_raw_youtrack_issues").Count(&rawCountAfter).Error)
+	assert.Equal(t, rawCountBefore+1, rawCountAfter,
+		"the replay read the raw layer only — no collection call (the task data carries no API client)")
+}
+
 // TestYoutrackIssueDataFlowMissingField covers the renamed-field trap: a
 // configured-but-absent field warns per scope and leaves the dedicated
 // columns empty; the run completes (warn, never fail).

backend/plugins/youtrack/e2e/stale_domain_rows_test.go

@@ -0,0 +1,99 @@
+/*
+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 (
+	"testing"
+
+	"github.com/apache/devlake/core/models/domainlayer/ticket"
+	"github.com/apache/devlake/core/utils"
+	"github.com/apache/devlake/helpers/e2ehelper"
+	"github.com/apache/devlake/plugins/youtrack/impl"
+	"github.com/apache/devlake/plugins/youtrack/models"
+	"github.com/apache/devlake/plugins/youtrack/tasks"
+	"github.com/stretchr/testify/require"
+)
+
+// The divider's lazy cleanup fires only when a converter emits an output row,
+// so the domain tables below must be replaced explicitly: once every source
+// row disappears, the scope's previous domain output must disappear too.
+
+func countRows(t *testing.T, dataflowTester *e2ehelper.DataFlowTester, model interface{}) int64 {
+	var n int64
+	require.NoError(t, dataflowTester.Db.Model(model).Count(&n).Error)
+	return n
+}
+
+// TestYoutrackIssueLabelsAllRemovedReplacement: the project's last tag is
+// removed — the extractor wipes the tool labels, the converter emits nothing.
+func TestYoutrackIssueLabelsAllRemovedReplacement(t *testing.T) {
+	var youtrack impl.Youtrack
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "youtrack", youtrack)
+
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_youtrack_issues.csv", "_raw_youtrack_issues")
+	dataflowTester.FlushTabler(&models.YoutrackIssue{})
+	dataflowTester.FlushTabler(&models.YoutrackIssueLabel{})
+	dataflowTester.FlushTabler(&ticket.IssueLabel{})
+
+	projectIds := []string{"0-1", "0-2"}
+	for _, projectId := range projectIds {
+		dataflowTester.Subtask(tasks.ExtractIssuesMeta, newTaskData(projectId, zeroConfig()))
+	}
+	for _, projectId := range projectIds {
+		dataflowTester.Subtask(tasks.ConvertIssueLabelsMeta, newTaskData(projectId, zeroConfig()))
+	}
+	require.Positive(t, countRows(t, dataflowTester, &ticket.IssueLabel{}), "populated scope converts labels into the domain layer")
+
+	require.NoError(t, dataflowTester.Db.Where("connection_id = 1").Delete(&models.YoutrackIssueLabel{}).Error)
+
+	for _, projectId := range projectIds {
+		dataflowTester.Subtask(tasks.ConvertIssueLabelsMeta, newTaskData(projectId, zeroConfig()))
+	}
+	require.Zero(t, countRows(t, dataflowTester, &ticket.IssueLabel{}),
+		"conversion emitting zero labels must still replace the scope's previous domain labels")
+}
+
+// TestYoutrackIssueAssigneesAllRemovedReplacement: every issue of the project
+// becomes unassigned — issues still convert, but no assignee row is emitted.
+func TestYoutrackIssueAssigneesAllRemovedReplacement(t *testing.T) {
+	var youtrack impl.Youtrack
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "youtrack", youtrack)
+
+	dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_youtrack_projects.csv", &models.YoutrackProject{})
+	dataflowTester.ImportCsvIntoTabler("./snapshot_tables/_tool_youtrack_issues_assignee_seed.csv", &models.YoutrackIssue{})
+	dataflowTester.FlushTabler(&ticket.Issue{})
+	dataflowTester.FlushTabler(&ticket.BoardIssue{})
+	dataflowTester.FlushTabler(&ticket.IssueAssignee{})
+	// the seed carries no raw origin; stamp the one the issue extractor would
+	// have set, since the converter copies it onto every domain row
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssue{}).Where("connection_id = 1").Updates(map[string]interface{}{
+		"_raw_data_table":  "_raw_youtrack_issues",
+		"_raw_data_params": utils.ToJsonString(models.YoutrackApiParams{ConnectionId: 1, ProjectId: "0-1"}),
+	}).Error)
+
+	dataflowTester.Subtask(tasks.ConvertIssuesMeta, newTaskData("0-1", zeroConfig()))
+	require.Positive(t, countRows(t, dataflowTester, &ticket.IssueAssignee{}), "an assigned issue converts into issue_assignees")
+
+	require.NoError(t, dataflowTester.Db.Model(&models.YoutrackIssue{}).
+		Where("connection_id = 1").Updates(map[string]interface{}{"assignee_id": "", "assignee_name": ""}).Error)
+
+	dataflowTester.Subtask(tasks.ConvertIssuesMeta, newTaskData("0-1", zeroConfig()))
+	require.Positive(t, countRows(t, dataflowTester, &ticket.Issue{}), "the issue itself still converts")
+	require.Zero(t, countRows(t, dataflowTester, &ticket.IssueAssignee{}),
+		"conversion emitting zero assignees must still replace the scope's previous domain assignees")
+}

backend/plugins/youtrack/e2e/stateful_scenario_test.go

@@ -0,0 +1,65 @@
+/*
+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"
+	"testing"
+
+	"github.com/apache/devlake/core/models"
+	"github.com/apache/devlake/core/plugin"
+	"github.com/apache/devlake/core/runner"
+	"github.com/apache/devlake/helpers/e2ehelper"
+	contextimpl "github.com/apache/devlake/impls/context"
+	"github.com/stretchr/testify/require"
+)
+
+// beginStatefulScenario opens a test that exercises the framework's
+// incremental/full-sync TRANSITIONS — the thing DataFlowTester.Subtask
+// cannot cover, because it deletes _devlake_subtask_state and forces
+// FullSync: true on every call (helpers/e2ehelper/data_flow_tester.go).
+//
+// Within the scenario, subtasks run through runSubtaskPreservingState: the
+// subtask state survives between calls and the sync policy permits
+// incremental mode, so a stateful extractor's second run goes incremental
+// on an unchanged config and full-syncs on a config change — exactly the
+// production semantics. The first run still full-syncs (no prior state).
+//
+// The scenario flushes _devlake_subtask_state at entry because the e2e
+// database is shared across tests and state keys (plugin, subtask, params)
+// do not include the test name.
+func beginStatefulScenario(t *testing.T, dataflowTester *e2ehelper.DataFlowTester) {
+	t.Helper()
+	dataflowTester.FlushTabler(&models.SubtaskState{})
+}
+
+// runSubtaskPreservingState executes one subtask inside a stateful scenario
+// (see beginStatefulScenario): state is neither flushed nor forced into
+// full sync. Panics on error, like DataFlowTester.Subtask.
+func runSubtaskPreservingState(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),
+		dataflowTester.Name,
+		taskData,
+		subtaskMeta.Name,
+		&models.SyncPolicy{},
+	)
+	require.NoError(t, subtaskMeta.EntryPoint(subtaskCtx))
+}