From 59a975fd6aa56c2b4e1c417ddd284568d362b87c Mon Sep 17 00:00:00 2001 From: Jose Camargo Date: Sun, 18 Oct 2026 10:36:01 -0300 Subject: [PATCH 1/2] fix(blueprint): retain scheduled planning failures Signed-off-by: Jose Camargo --- backend/DevelopmentManual.md | 3 + backend/server/services/blueprint.go | 18 +- .../services/blueprint_plan_failure_test.go | 298 ++++++++++++++++++ backend/server/services/pipeline_helper.go | 48 +++ 4 files changed, 364 insertions(+), 3 deletions(-) create mode 100644 backend/server/services/blueprint_plan_failure_test.go diff --git a/backend/DevelopmentManual.md b/backend/DevelopmentManual.md index 1043c7d4a64..26208d55ff5 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 planning errors directly to the caller without creating a pipeline. If saving the failed record or sending its notification fails, check the lake logs for the additional error. + * `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 9f78f576037..6acc95849d3 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_test.go b/backend/server/services/blueprint_plan_failure_test.go new file mode 100644 index 00000000000..561ee8f36bb --- /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 f3ac38671b2..cb9758e8a7d 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 From 540636fd68d8d49b07e39b0d20ddef1dcdada5ab Mon Sep 17 00:00:00 2001 From: Jose Camargo Date: Sun, 18 Oct 2026 15:00:10 -0300 Subject: [PATCH 2/2] fix(ui): prevent reruns for pipelines without tasks Signed-off-by: Jose Camargo --- ...blueprint_plan_failure_integration_test.go | 110 ++++++++++++++++++ backend/test/Readme.md | 18 +++ .../routes/pipeline/components/info.test.tsx | 92 +++++++++++++++ .../src/routes/pipeline/components/info.tsx | 15 ++- 4 files changed, 229 insertions(+), 6 deletions(-) create mode 100644 backend/server/services/blueprint_plan_failure_integration_test.go create mode 100644 config-ui/src/routes/pipeline/components/info.test.tsx 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 00000000000..241112bae1e --- /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/test/Readme.md b/backend/test/Readme.md index 046bce5c99a..87bfcfa30b2 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 00000000000..d87c22b7059 --- /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: () => Loading })); +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( + + + , + ); +} + +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 426f3a8c0c6..7ef98dfbefa 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) && (