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 5ac573b8d fix(blueprint): retain scheduled planning failures (#9234)
5ac573b8d is described below
commit 5ac573b8d462d24b8b50b11d78a4fc861a902172
Author: Jose Camargo <[email protected]>
AuthorDate: Wed Oct 21 22:27:55 2026 -0300
fix(blueprint): retain scheduled planning failures (#9234)
* fix(blueprint): retain scheduled planning failures
Signed-off-by: Jose Camargo <[email protected]>
* fix(ui): prevent reruns for pipelines without tasks
Signed-off-by: Jose Camargo <[email protected]>
---------
Signed-off-by: Jose Camargo <[email protected]>
---
backend/DevelopmentManual.md | 3 +
backend/server/services/blueprint.go | 18 +-
.../blueprint_plan_failure_integration_test.go | 110 ++++++++
.../server/services/blueprint_plan_failure_test.go | 298 +++++++++++++++++++++
backend/server/services/pipeline_helper.go | 48 ++++
backend/test/Readme.md | 18 ++
.../src/routes/pipeline/components/info.test.tsx | 92 +++++++
config-ui/src/routes/pipeline/components/info.tsx | 15 +-
8 files changed, 593 insertions(+), 9 deletions(-)
diff --git a/backend/DevelopmentManual.md b/backend/DevelopmentManual.md
index 1043c7d4a..26208d55f 100644
--- a/backend/DevelopmentManual.md
+++ b/backend/DevelopmentManual.md
@@ -76,6 +76,9 @@ Note the documentation structure of the functions in these
example: that's the s
This is where we also define the logic used by the server to run the plugin
based on the constraints set by a blueprint (the user's configuration); that is,
a "pipeline plan", which defines what input data to feed into the plugin,
which subtasks (see `tasks` package) to execute, and which other plugins to run
after this plugin's execution and the input to each of them (typically "helper"
plugins).
See
[this](https://github.com/apache/incubator-DevLake/blob/main/backend/plugins/gitlab/api/blueprint_v200.go)
for reference.
+
+ If plan generation returns an error during a scheduled blueprint run,
DevLake records a failed pipeline with the planning error, start/end
timestamps, and blueprint labels, then invokes the configured pipeline-status
notification service after the record is committed. This pipeline has no tasks
because planning did not complete. Fix the blueprint or connection and trigger
a new run; rerunning tasks from this failed record cannot regenerate its plan.
Manual triggers continue to return [...]
+
* `e2e`: This package contains "pseudo-integration-tests" for plugins; tests
that use mocked data and test the plugin's extraction and conversion logic of
that data against an actual database. The test data is formatted in CSV files
structured based on the expected database table structures.
* `impl`: This package is where the plugin is actually defined, and is the
"entrypoint": it is the implementation of multiple interfaces. See
[this](https://github.com/apache/incubator-DevLake/tree/main/backend/plugins/gitlab/impl)
for example.
* At a minimum, a plugin must implement
[PluginMeta](https://github.com/apache/incubator-DevLake/blob/main/backend/core/plugin/plugin_meta.go).
diff --git a/backend/server/services/blueprint.go
b/backend/server/services/blueprint.go
index 9f78f5760..6acc95849 100644
--- a/backend/server/services/blueprint.go
+++ b/backend/server/services/blueprint.go
@@ -22,6 +22,7 @@ import (
"fmt"
"strings"
"sync"
+ "time"
"github.com/apache/devlake/helpers/pluginhelper/services"
@@ -54,7 +55,7 @@ type BlueprintJob struct {
func (bj BlueprintJob) Run() {
blueprint := bj.Blueprint
- pipeline, err := createPipelineByBlueprint(blueprint,
&blueprint.SyncPolicy)
+ pipeline, err := createPipelineByBlueprint(blueprint,
&blueprint.SyncPolicy, true)
if err == ErrEmptyPlan {
blueprintLog.Info("Empty plan, blueprint id:[%d] blueprint
name:[%s]", blueprint.ID, blueprint.Name)
return
@@ -309,13 +310,24 @@ func reloadBlueprint(blueprint *models.Blueprint)
errors.Error {
return nil
}
-func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy
*models.SyncPolicy) (*models.Pipeline, errors.Error) {
+func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy
*models.SyncPolicy, recordPlanFailure bool) (*models.Pipeline, errors.Error) {
var plan models.PipelinePlan
var err errors.Error
+ beganAt := time.Now()
if blueprint.Mode == models.BLUEPRINT_MODE_NORMAL {
plan, err = MakePlanForBlueprint(blueprint, syncPolicy)
if err != nil {
blueprintLog.Error(err, fmt.Sprintf("failed to
MakePlanForBlueprint on blueprint:[%d][%s]", blueprint.ID, blueprint.Name))
+ // Manual triggers return the error to their caller.
Scheduled attempts
+ // need a terminal record because no pipeline has been
created yet.
+ if recordPlanFailure && err != ErrEmptyPlan {
+ failed, saveErr :=
createFailedBlueprintPipeline(blueprint, beganAt, err)
+ if saveErr != nil {
+ blueprintLog.Error(saveErr, "failed to
record planning failure for blueprint #%d", blueprint.ID)
+ } else if notifyErr :=
NotifyExternal(failed.ID); notifyErr != nil {
+ blueprintLog.Error(notifyErr, "failed
to send planning failure notification for pipeline #%d", failed.ID)
+ }
+ }
return nil, err
}
} else {
@@ -427,7 +439,7 @@ func TriggerBlueprint(id uint64, triggerSyncPolicy
*models.TriggerSyncPolicy, sh
SkipOnFail: false,
TimeAfter: nil,
TriggerSyncPolicy: *triggerSyncPolicy,
- })
+ }, false)
if err != nil {
return nil, err
}
diff --git a/backend/server/services/blueprint_plan_failure_integration_test.go
b/backend/server/services/blueprint_plan_failure_integration_test.go
new file mode 100644
index 000000000..241112bae
--- /dev/null
+++ b/backend/server/services/blueprint_plan_failure_integration_test.go
@@ -0,0 +1,110 @@
+/*
+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 services
+
+import (
+ "os"
+ "testing"
+ "time"
+
+ "github.com/apache/devlake/core/errors"
+ "github.com/apache/devlake/core/models"
+ contextimpl "github.com/apache/devlake/impls/context"
+ "github.com/apache/devlake/impls/dalgorm"
+ "github.com/apache/devlake/impls/logruslog"
+ "github.com/stretchr/testify/require"
+ "gorm.io/driver/mysql"
+ "gorm.io/driver/postgres"
+ "gorm.io/gorm"
+)
+
+// Use dedicated, disposable databases: this test creates the pipeline tables.
+// Unset DSNs keep ordinary unit-test runs independent of database servers.
+func TestBlueprintPlanFailureDatabase(t *testing.T) {
+ for _, driver := range []string{"mysql", "postgres"} {
+ t.Run(driver, func(t *testing.T) {
+ env := "DEVLAKE_TEST_MYSQL_DSN"
+ if driver == "postgres" {
+ env = "DEVLAKE_TEST_POSTGRES_DSN"
+ }
+ dsn := os.Getenv(env)
+ if dsn == "" {
+ t.Skip("set " + env + " to a disposable
database")
+ }
+ var dialect gorm.Dialector = mysql.Open(dsn)
+ if driver == "postgres" {
+ dialect = postgres.Open(dsn)
+ }
+ dalgorm.Init("01234567890123456789012345678901")
+ database, err := gorm.Open(dialect, &gorm.Config{})
+ require.NoError(t, err)
+ sqlDB, err := database.DB()
+ require.NoError(t, err)
+ t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+ require.NoError(t,
database.AutoMigrate(&models.Pipeline{}, &models.DbPipelineLabel{},
&models.Task{}))
+ testDB := dalgorm.NewDalgorm(database)
+ withProjectTestDatabase(t, testDB)
+ previousRes := basicRes
+ basicRes = contextimpl.NewDefaultBasicRes(nil,
logruslog.Global, testDB)
+ t.Cleanup(func() { basicRes = previousRes })
+
+ bp := &models.Blueprint{Name:
"planning-failure-integration", Labels: []string{"integration"}}
+ bp.ID = uint64(time.Now().UnixNano())
+ t.Cleanup(func() {
+ var ids []uint64
+ require.NoError(t,
database.Model(&models.Pipeline{}).Where("blueprint_id = ?", bp.ID).Pluck("id",
&ids).Error)
+ if len(ids) > 0 {
+ require.NoError(t,
database.Where("pipeline_id IN ?", ids).Delete(&models.DbPipelineLabel{}).Error)
+ require.NoError(t,
database.Where("blueprint_id = ?", bp.ID).Delete(&models.Pipeline{}).Error)
+ }
+ })
+ planErr := errors.Default.New("source planning failed")
+ stored, saveErr := createFailedBlueprintPipeline(bp,
time.Now(), planErr)
+ require.NoError(t, saveErr)
+ require.NotZero(t, stored.ID)
+ rows, count, readErr :=
GetDbPipelines(&PipelineQuery{BlueprintId: bp.ID, Status: models.TASK_FAILED,
Label: "integration"})
+ require.NoError(t, readErr)
+ require.EqualValues(t, 1, count)
+ require.Len(t, rows, 1)
+ require.Equal(t, stored.ID, rows[0].ID)
+ require.Equal(t, bp.Labels, rows[0].Labels)
+ require.Empty(t, rows[0].Plan)
+ require.Contains(t, rows[0].Message, planErr.Error())
+ require.NotNil(t, rows[0].BeganAt)
+ require.NotNil(t, rows[0].FinishedAt)
+ _, pending, readErr :=
GetDbPipelines(&PipelineQuery{BlueprintId: bp.ID, Pending: 1})
+ require.NoError(t, readErr)
+ require.Zero(t, pending)
+ var tasks int64
+ require.NoError(t,
database.Model(&models.Task{}).Where("pipeline_id = ?",
stored.ID).Count(&tasks).Error)
+ require.Zero(t, tasks)
+
+ // The duplicate primary key makes the label INSERT
fail after the
+ // pipeline INSERT. Neither the second pipeline nor its
labels may survive.
+ bp.Labels = []string{"duplicate", "duplicate"}
+ failed, saveErr := createFailedBlueprintPipeline(bp,
time.Now(), planErr)
+ require.Error(t, saveErr)
+ require.Nil(t, failed)
+ var pipelines, labels int64
+ require.NoError(t,
database.Model(&models.Pipeline{}).Where("blueprint_id = ?",
bp.ID).Count(&pipelines).Error)
+ require.EqualValues(t, 1, pipelines)
+ require.NoError(t,
database.Model(&models.DbPipelineLabel{}).Where("name = ?",
"duplicate").Count(&labels).Error)
+ require.Zero(t, labels)
+ })
+ }
+}
diff --git a/backend/server/services/blueprint_plan_failure_test.go
b/backend/server/services/blueprint_plan_failure_test.go
new file mode 100644
index 000000000..561ee8f36
--- /dev/null
+++ b/backend/server/services/blueprint_plan_failure_test.go
@@ -0,0 +1,298 @@
+/*
+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 services
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/devlake/core/errors"
+ "github.com/apache/devlake/core/models"
+ "github.com/apache/devlake/core/plugin"
+ contextimpl "github.com/apache/devlake/impls/context"
+ "github.com/apache/devlake/impls/logruslog"
+ mockdal "github.com/apache/devlake/mocks/core/dal"
+ mockplugin "github.com/apache/devlake/mocks/core/plugin"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/mock"
+ "github.com/stretchr/testify/require"
+)
+
+type blueprintFailureNotification func(PipelineNotificationParam) errors.Error
+
+func (f blueprintFailureNotification) PipelineStatusChanged(p
PipelineNotificationParam) errors.Error {
+ return f(p)
+}
+
+// Service globals follow the existing service tests; these tests must not run
+// in parallel, and normal-pipeline notification goroutines are joined below.
+func blueprintFailureTestDB(t *testing.T) *mockdal.Dal {
+ t.Helper()
+ testDB := mockdal.NewDal(t)
+ withProjectTestDatabase(t, testDB)
+ oldRes, oldLogger := basicRes, logger
+ oldDefault, oldCustom := defaultNotificationService,
customPipelineNotificationService
+ basicRes = contextimpl.NewDefaultBasicRes(nil, logruslog.Global, testDB)
+ logger = logruslog.Global
+ defaultNotificationService, customPipelineNotificationService = nil, nil
+ t.Cleanup(func() {
+ basicRes, logger = oldRes, oldLogger
+ defaultNotificationService, customPipelineNotificationService =
oldDefault, oldCustom
+ })
+ return testDB
+}
+
+func failingPlanBlueprint(t *testing.T, planErr errors.Error)
*models.Blueprint {
+ t.Helper()
+ name := "test-blueprint-plan-failure"
+ source := new(mockplugin.CompositeDataSourcePluginBlueprintV200)
+ source.On("MakeDataSourcePipelinePlanV200", uint64(7), mock.Anything).
+ Return(models.PipelinePlan(nil), []plugin.Scope(nil),
planErr).Once()
+ require.NoError(t, plugin.RegisterPlugin(name, source))
+ t.Cleanup(func() {
+ delete(plugin.AllPlugins(), name)
+ source.AssertExpectations(t)
+ })
+ bp := &models.Blueprint{
+ Name: "scheduled-source", Mode: models.BLUEPRINT_MODE_NORMAL,
+ Priority: 3, Labels: []string{"engineering"},
+ Connections: []*models.BlueprintConnection{{
+ PluginName: name, ConnectionId: 7,
+ Scopes: []*models.BlueprintScope{{ScopeId: "repo"}},
+ }},
+ }
+ bp.ID = 12
+ bp.SkipOnFail = true
+ return bp
+}
+
+func expectPipelineNotificationRead(t *testing.T, testDB *mockdal.Dal,
pipeline **models.Pipeline) {
+ t.Helper()
+ testDB.On("First", mock.Anything, mock.Anything).Run(func(args
mock.Arguments) {
+ switch target := args.Get(0).(type) {
+ case *models.Pipeline:
+ *target = **pipeline
+ case *models.Blueprint:
+ target.ProjectName = "engineering"
+ default:
+ t.Errorf("unexpected notification lookup: %T", target)
+ }
+ }).Return(nil).Twice()
+ testDB.On("Pluck", "name", mock.Anything,
mock.Anything).Return(nil).Twice()
+}
+
+func TestBlueprintJobRecordsPlanFailure(t *testing.T) {
+ for _, notificationFails := range []bool{false, true} {
+ name := "notifies after commit"
+ if notificationFails {
+ name = "notification failure keeps history"
+ }
+ t.Run(name, func(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ planErr := errors.Unauthorized.New("installation
credentials rejected")
+ bp := failingPlanBlueprint(t, planErr)
+ started := time.Now()
+ tx := mockdal.NewTransaction(t)
+ testDB.On("Begin").Return(tx).Once()
+ var stored *models.Pipeline
+ tx.On("Create", mock.Anything).Run(func(args
mock.Arguments) {
+ switch row := args.Get(0).(type) {
+ case *models.Pipeline:
+ row.ID = 123
+ stored = row
+ assert.Equal(t, models.TASK_FAILED,
row.Status)
+ assert.Equal(t, bp.ID, row.BlueprintId)
+ assert.Equal(t, bp.Name, row.Name)
+ assert.Equal(t, bp.Priority,
row.Priority)
+ assert.Equal(t, bp.SyncPolicy,
row.SyncPolicy)
+ assert.Equal(t, models.PipelinePlan{},
row.Plan)
+ assert.Zero(t, row.TotalTasks)
+ assert.Zero(t, row.FinishedTasks)
+ require.NotNil(t, row.BeganAt)
+ require.NotNil(t, row.FinishedAt)
+ assert.False(t,
row.BeganAt.Before(started))
+ assert.False(t,
row.FinishedAt.Before(*row.BeganAt))
+ assert.Equal(t,
int(row.FinishedAt.Sub(*row.BeganAt).Seconds()), row.SpentSeconds)
+ assert.Contains(t, row.Message,
planErr.Error())
+ assert.Contains(t, row.ErrorName,
"Failed to generate pipeline plan")
+ case *[]models.DbPipelineLabel:
+ require.NotNil(t, stored)
+ assert.Equal(t,
[]models.DbPipelineLabel{{PipelineId: 123, Name: "engineering"}}, *row)
+ default:
+ t.Errorf("planning failure must not
create tasks: %T", row)
+ }
+ }).Return(nil).Twice()
+ committed := false
+ tx.On("Commit").Run(func(mock.Arguments) { committed =
true }).Return(nil).Once()
+ expectPipelineNotificationRead(t, testDB, &stored)
+ notifications := 0
+ customPipelineNotificationService =
blueprintFailureNotification(func(p PipelineNotificationParam) errors.Error {
+ assert.True(t, committed)
+ assert.Equal(t, uint64(123), p.PipelineID)
+ assert.Equal(t, "engineering", p.ProjectName)
+ assert.Equal(t, models.TASK_FAILED, p.Status)
+ assert.Equal(t, stored.FinishedAt, p.FinishedAt)
+ notifications++
+ if notificationFails {
+ return errors.Default.New("notification
unavailable")
+ }
+ return nil
+ })
+ BlueprintJob{Blueprint: bp}.Run()
+ assert.Equal(t, 1, notifications)
+ tx.AssertNotCalled(t, "Rollback")
+ })
+ }
+}
+
+func TestBlueprintPlanFailureRollsBackAndPreservesCause(t *testing.T) {
+ for _, stage := range []string{"pipeline", "labels", "commit",
"rollback"} {
+ t.Run(stage, func(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ planErr := errors.Default.New("source planning failed")
+ bp := failingPlanBlueprint(t, planErr)
+ tx := mockdal.NewTransaction(t)
+ testDB.On("Begin").Return(tx).Once()
+ writeErr := errors.Default.New("database write failed")
+ if stage == "pipeline" || stage == "rollback" {
+ tx.On("Create",
mock.Anything).Return(writeErr).Once()
+ } else {
+ tx.On("Create",
mock.Anything).Return(nil).Once()
+ if stage == "labels" {
+ tx.On("Create",
mock.Anything).Return(writeErr).Once()
+ } else {
+ tx.On("Create",
mock.Anything).Return(nil).Once()
+ tx.On("Commit").Return(writeErr).Once()
+ }
+ }
+ var rollbackErr errors.Error
+ if stage == "rollback" {
+ rollbackErr = errors.Default.New("rollback
failed")
+ }
+ tx.On("Rollback").Return(rollbackErr).Once()
+ customPipelineNotificationService =
blueprintFailureNotification(func(PipelineNotificationParam) errors.Error {
+ t.Error("uncommitted failure must not be
notified")
+ return nil
+ })
+ pipeline, err := createPipelineByBlueprint(bp,
&bp.SyncPolicy, true)
+ assert.Nil(t, pipeline)
+ assert.Same(t, planErr, err)
+ })
+ }
+}
+
+func TestBlueprintManualPlanningFailureDoesNotCreateHistory(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ planErr := errors.Default.New("source planning failed")
+ bp := failingPlanBlueprint(t, planErr)
+ bp.Enable = true
+ testDB.On("First", mock.Anything, mock.Anything).Run(func(args
mock.Arguments) {
+ *args.Get(0).(*models.Blueprint) = *bp
+ }).Return(nil).Once()
+ testDB.On("Pluck", "name", mock.Anything,
mock.Anything).Return(nil).Once()
+ testDB.On("All", mock.Anything, mock.Anything).Return(nil).Twice()
+ pipeline, err := TriggerBlueprint(bp.ID, &models.TriggerSyncPolicy{},
false)
+ assert.Nil(t, pipeline)
+ assert.Same(t, planErr, err)
+ testDB.AssertNotCalled(t, "Begin")
+}
+
+func TestBlueprintPlanFailureWithoutLabels(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ planErr := errors.Default.New("source planning failed")
+ bp := failingPlanBlueprint(t, planErr)
+ bp.Labels = nil
+ tx := mockdal.NewTransaction(t)
+ testDB.On("Begin").Return(tx).Once()
+ tx.On("Create", mock.MatchedBy(func(row *models.Pipeline) bool {
+ return row.Status == models.TASK_FAILED && len(row.Labels) == 0
+ })).Return(nil).Once()
+ tx.On("Commit").Return(nil).Once()
+ pipeline, err := createPipelineByBlueprint(bp, &bp.SyncPolicy, true)
+ assert.Nil(t, pipeline)
+ assert.Same(t, planErr, err)
+ tx.AssertNotCalled(t, "Rollback")
+}
+
+func TestBlueprintJobEmptyPlanSentinelDoesNotCreateHistory(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ bp := failingPlanBlueprint(t, ErrEmptyPlan)
+ BlueprintJob{Blueprint: bp}.Run()
+ testDB.AssertNotCalled(t, "Begin")
+}
+
+func TestBlueprintJobExistingCreationPaths(t *testing.T) {
+ for _, scenario := range []string{"success", "empty", "pending",
"insert failure"} {
+ t.Run(scenario, func(t *testing.T) {
+ testDB := blueprintFailureTestDB(t)
+ bp := &models.Blueprint{Name: "advanced", Mode:
models.BLUEPRINT_MODE_ADVANCED}
+ bp.ID = 20
+ if scenario != "empty" {
+ bp.Plan = models.PipelinePlan{{{Plugin:
"test"}}}
+ }
+ tx := mockdal.NewTransaction(t)
+ testDB.On("Begin").Return(tx).Once()
+ tx.On("LockTables", mock.Anything).Return(nil).Once()
+ tx.On("UnlockTables").Return(nil).Once()
+ var pending int64
+ if scenario == "pending" {
+ pending = 1
+ }
+ tx.On("Count", mock.Anything).Return(pending,
nil).Once()
+ var stored *models.Pipeline
+ if scenario != "pending" {
+ var insertErr errors.Error
+ if scenario == "insert failure" {
+ insertErr = errors.Default.New("insert
failed")
+ }
+ tx.On("Create", mock.Anything).Run(func(args
mock.Arguments) {
+ stored = args.Get(0).(*models.Pipeline)
+ stored.ID = 456
+ assert.Equal(t, models.TASK_CREATED,
stored.Status)
+ }).Return(insertErr).Once()
+ }
+ notified := make(chan struct{})
+ if scenario == "success" {
+ tx.On("Create", mock.Anything).Run(func(args
mock.Arguments) {
+ task := args.Get(0).(*models.Task)
+ assert.Equal(t, "test", task.Plugin)
+ assert.Equal(t, uint64(456),
task.PipelineId)
+ }).Return(nil).Once()
+ tx.On("Update",
mock.Anything).Return(nil).Once()
+ tx.On("Commit").Return(nil).Once()
+ expectPipelineNotificationRead(t, testDB,
&stored)
+ customPipelineNotificationService =
blueprintFailureNotification(func(p PipelineNotificationParam) errors.Error {
+ assert.Equal(t, models.TASK_CREATED,
p.Status)
+ close(notified)
+ return nil
+ })
+ } else {
+ tx.On("Rollback").Return(nil).Once()
+ }
+ BlueprintJob{Blueprint: bp}.Run()
+ if scenario == "success" {
+ select {
+ case <-notified:
+ case <-time.After(5 * time.Second):
+ t.Fatal("created notification did not
complete")
+ }
+ assert.Equal(t, 1, stored.TotalTasks)
+ }
+ })
+ }
+}
diff --git a/backend/server/services/pipeline_helper.go
b/backend/server/services/pipeline_helper.go
index f3ac38671..cb9758e8a 100644
--- a/backend/server/services/pipeline_helper.go
+++ b/backend/server/services/pipeline_helper.go
@@ -122,6 +122,54 @@ func CreateDbPipeline(newPipeline *models.NewPipeline)
(pipeline *models.Pipelin
return dbPipeline, nil
}
+// createFailedBlueprintPipeline records a scheduled attempt that could not
make
+// a plan. Insert the terminal status and labels atomically: it must never
enter
+// the execution queue, and no synthetic task should be dispatched.
+func createFailedBlueprintPipeline(blueprint *models.Blueprint, beganAt
time.Time, planErr errors.Error) (*models.Pipeline, errors.Error) {
+ finishedAt := time.Now()
+ failure := errors.Default.Wrap(planErr, "Failed to generate pipeline
plan; fix the blueprint and trigger a new run")
+ pipeline := &models.Pipeline{
+ Name: blueprint.Name,
+ BlueprintId: blueprint.ID,
+ Plan: models.PipelinePlan{},
+ Status: models.TASK_FAILED,
+ BeganAt: &beganAt,
+ FinishedAt: &finishedAt,
+ SpentSeconds: int(finishedAt.Sub(beganAt).Seconds()),
+ Message: failure.Error(),
+ ErrorName: failure.Messages().Format(),
+ Priority: blueprint.Priority,
+ SyncPolicy: blueprint.SyncPolicy,
+ Labels: append([]string(nil), blueprint.Labels...),
+ }
+ tx := db.Begin()
+ committed := false
+ defer func() {
+ if !committed {
+ if err := tx.Rollback(); err != nil {
+ globalPipelineLog.Error(err, "failed to roll
back planning failure for blueprint #%d", blueprint.ID)
+ }
+ }
+ }()
+ if err := tx.Create(pipeline); err != nil {
+ return nil, err
+ }
+ if len(pipeline.Labels) > 0 {
+ labels := make([]models.DbPipelineLabel, 0,
len(pipeline.Labels))
+ for _, label := range pipeline.Labels {
+ labels = append(labels,
models.DbPipelineLabel{PipelineId: pipeline.ID, Name: label})
+ }
+ if err := tx.Create(&labels); err != nil {
+ return nil, err
+ }
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ committed = true
+ return pipeline, nil
+}
+
// GetDbPipelines by query
func GetDbPipelines(query *PipelineQuery) ([]*models.Pipeline, int64,
errors.Error) {
// process query parameters
diff --git a/backend/test/Readme.md b/backend/test/Readme.md
index 046bce5c9..87bfcfa30 100644
--- a/backend/test/Readme.md
+++ b/backend/test/Readme.md
@@ -41,3 +41,21 @@ export PKG_CONFIG_PATH=/path/to/libgit2/build
export CGO_LDFLAGS='-L/path/to/libgit2/build -Wl,-rpath,/path/to/libgit2/build'
make e2e-test
```
+
+### Planning-failure database regression test
+
+`TestBlueprintPlanFailureDatabase` verifies failed-pipeline history, exclusion
from
+the pending queue, and atomic rollback when label insertion fails. Use
dedicated,
+disposable databases: the test creates pipeline, label, and task tables, and
cleans
+up its own records. It does not require external data sources or a running
server.
+
+After generating the core and helper mocks, run from `backend/`:
+
+```bash
+export
DEVLAKE_TEST_MYSQL_DSN='merico:merico@tcp(127.0.0.1:3306)/blueprint_test?charset=utf8mb4&parseTime=True&loc=UTC'
+export DEVLAKE_TEST_POSTGRES_DSN='host=127.0.0.1 user=postgres
dbname=blueprint_test sslmode=disable'
+go test -count=1 -v -run TestBlueprintPlanFailureDatabase ./server/services
+```
+
+Each database subtest is skipped when its DSN is unset; ordinary unit tests do
not
+require database servers.
diff --git a/config-ui/src/routes/pipeline/components/info.test.tsx
b/config-ui/src/routes/pipeline/components/info.test.tsx
new file mode 100644
index 000000000..d87c22b70
--- /dev/null
+++ b/config-ui/src/routes/pipeline/components/info.test.tsx
@@ -0,0 +1,92 @@
+/*
+ * 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.
+ *
+ */
+
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { fireEvent, render, screen, waitFor } from '@testing-library/react';
+import { ThemeProvider } from 'styled-components';
+
+import { getTheme } from '@/theme/tokens';
+import { IPipeline, IPipelineStatus } from '@/types';
+
+import { PipelineInfo } from './info';
+
+const state = vi.hoisted(() => ({ data: undefined as IPipeline | undefined,
rerun: vi.fn() }));
+vi.mock('@/hooks', () => ({ useAutoRefresh: () => ({ data: state.data }) }));
+vi.mock('@/api', () => ({ default: { pipeline: { rerun: state.rerun } } }));
+vi.mock('@/components', () => ({ Loading: () => <span>Loading</span> }));
+vi.mock('@/utils', async () => ({
+ ...(await import('@/utils/operator')),
+ ...(await import('@/utils/time')),
+}));
+
+const terminalStatuses = [
+ IPipelineStatus.COMPLETED,
+ IPipelineStatus.PARTIAL,
+ IPipelineStatus.FAILED,
+ IPipelineStatus.CANCELLED,
+];
+
+function renderPipeline(status: IPipelineStatus, totalTasks: number) {
+ state.data = {
+ id: 42,
+ status,
+ totalTasks,
+ finishedTasks: 0,
+ stage: 0,
+ beganAt: null,
+ finishedAt: null,
+ message: 'Failed to generate pipeline plan; fix the blueprint and trigger
a new run',
+ };
+ return render(
+ <ThemeProvider theme={getTheme('light')}>
+ <PipelineInfo id={42} />
+ </ThemeProvider>,
+ );
+}
+
+describe('pipeline task reruns', () => {
+ beforeEach(() => {
+ state.rerun.mockReset().mockResolvedValue(undefined);
+ });
+
+ it.each(terminalStatuses)('does not offer a task rerun for %s with no
tasks', (status) => {
+ renderPipeline(status, 0);
+ expect(screen.queryByRole('button')).toBeNull();
+ expect(state.rerun).not.toHaveBeenCalled();
+ });
+
+ it('keeps the planning error and recovery instructions visible', () => {
+ renderPipeline(IPipelineStatus.FAILED, 0);
+ expect(screen.getByText('Failed to generate pipeline plan; fix the
blueprint and trigger a new run')).toBeTruthy();
+ expect(screen.getByText('0/0')).toBeTruthy();
+ });
+
+ it.each(terminalStatuses)('allows existing tasks to be rerun for %s', async
(status) => {
+ renderPipeline(status, 1);
+ fireEvent.click(screen.getByRole('button', { name: 'Rerun pipeline' }));
+ await waitFor(() =>
expect(state.rerun).toHaveBeenCalledExactlyOnceWith(42));
+ await waitFor(() =>
+ expect(screen.getByRole('button', { name: 'Rerun pipeline'
}).getAttribute('disabled')).toBeNull(),
+ );
+ });
+
+ it('does not offer reruns while the pipeline is running', () => {
+ renderPipeline(IPipelineStatus.RUNNING, 1);
+ expect(screen.queryByRole('button', { name: 'Rerun pipeline'
})).toBeNull();
+ });
+});
diff --git a/config-ui/src/routes/pipeline/components/info.tsx
b/config-ui/src/routes/pipeline/components/info.tsx
index 426f3a8c0..7ef98dfbe 100644
--- a/config-ui/src/routes/pipeline/components/info.tsx
+++ b/config-ui/src/routes/pipeline/components/info.tsx
@@ -111,12 +111,15 @@ export const PipelineInfo = ({ id }: Props) => {
{[IPipelineStatus.ACTIVE, IPipelineStatus.RUNNING,
IPipelineStatus.RERUN].includes(status) && (
<Button loading={operating} icon={<StopOutlined />}
onClick={handleCancel} />
)}
- {[
- IPipelineStatus.COMPLETED,
- IPipelineStatus.PARTIAL,
- IPipelineStatus.FAILED,
- IPipelineStatus.CANCELLED,
- ].includes(status) && <Button loading={operating}
icon={<RedoOutlined />} onClick={handleRerun} />}
+ {totalTasks > 0 &&
+ [
+ IPipelineStatus.COMPLETED,
+ IPipelineStatus.PARTIAL,
+ IPipelineStatus.FAILED,
+ IPipelineStatus.CANCELLED,
+ ].includes(status) && (
+ <Button aria-label="Rerun pipeline" loading={operating}
icon={<RedoOutlined />} onClick={handleRerun} />
+ )}
</li>
</ul>
{IPipelineStatus.FAILED === status && <p
className="'message'">{message}</p>}
