fix(blueprint): retain scheduled planning failures

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).

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
 	}

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)
+		})
+	}
+}