fix(graphql): resume collectors from durable checkpoints after restart

.github/workflows/test.yml

@@ -36,6 +36,26 @@ on:
     - cron: '0 0 * * *' # run ci every night
 
 jobs:
+  graphql-resume:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        database: [mysql, postgres]
+    steps:
+    - uses: actions/checkout@v7
+    - name: GraphQL real database crash and resume
+      env:
+        DEVLAKE_CRASH_DRIVER: ${{ matrix.database }}
+        TMPDIR: ${{ runner.temp }}
+      run: bash backend/test/graphql-resume-crash.sh
+    - name: Preserve failure and resume logs
+      if: always()
+      uses: actions/upload-artifact@v7
+      with:
+        name: graphql-resume-${{ matrix.database }}
+        path: ${{ runner.temp }}/devlake-graphql-crash.*/*.log
   test:
     strategy:
       matrix:

backend/core/models/graphql_collector_state.go

@@ -0,0 +1,38 @@
+/*
+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 models
+
+import "time"
+
+type GraphqlCollectorState struct {
+	ScopeHash     string `gorm:"primaryKey;type:char(64)"`
+	InputHash     string `gorm:"primaryKey;type:varchar(64)"`
+	TaskID        uint64
+	RawDataTable  string `gorm:"type:varchar(255)"`
+	RawDataParams string `gorm:"type:text"`
+	Completed     bool
+	SkipCursor    string `gorm:"type:text"`
+	UpdatedAt     time.Time
+	Contract      string `gorm:"type:varchar(64)"`
+	InputDigest   string `gorm:"type:varchar(64)"`
+	RawCount      *int64
+}
+
+func (GraphqlCollectorState) TableName() string {
+	return "_devlake_graphql_collector_states"
+}

backend/core/models/migrationscripts/20260916_add_graphql_collector_states.go

@@ -0,0 +1,52 @@
+/*
+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 migrationscripts
+
+import (
+	"time"
+
+	"github.com/apache/devlake/core/context"
+	"github.com/apache/devlake/core/errors"
+	"github.com/apache/devlake/core/plugin"
+)
+
+type graphqlCollectorState20260916 struct {
+	ScopeHash     string `gorm:"primaryKey;type:char(64)"`
+	InputHash     string `gorm:"primaryKey;type:varchar(64)"`
+	TaskID        uint64
+	RawDataTable  string `gorm:"type:varchar(255)"`
+	RawDataParams string `gorm:"type:text"`
+	Completed     bool
+	SkipCursor    string `gorm:"type:text"`
+	UpdatedAt     time.Time
+	Contract      string `gorm:"type:varchar(64)"`
+	InputDigest   string `gorm:"type:varchar(64)"`
+	RawCount      *int64
+}
+
+func (graphqlCollectorState20260916) TableName() string { return "_devlake_graphql_collector_states" }
+
+type addGraphqlCollectorStates struct{}
+
+var _ plugin.MigrationScript = (*addGraphqlCollectorStates)(nil)
+
+func (*addGraphqlCollectorStates) Up(res context.BasicRes) errors.Error {
+	return res.GetDal().AutoMigrate(&graphqlCollectorState20260916{})
+}
+func (*addGraphqlCollectorStates) Version() uint64 { return 20260916000001 }
+func (*addGraphqlCollectorStates) Name() string    { return "add graphql collector checkpoints" }

backend/core/models/migrationscripts/graphql_checkpoint_test.go

@@ -0,0 +1,38 @@
+/*
+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 migrationscripts
+
+import (
+	"github.com/apache/devlake/core/models"
+	"github.com/stretchr/testify/require"
+	"reflect"
+	"testing"
+)
+
+func TestGraphqlCheckpointMigrationSnapshot(t *testing.T) {
+	snapshot := reflect.TypeOf(graphqlCollectorState20260916{})
+	runtime := reflect.TypeOf(models.GraphqlCollectorState{})
+	require.Equal(t, snapshot.NumField(), runtime.NumField())
+	for i := 0; i < snapshot.NumField(); i++ {
+		a, b := snapshot.Field(i), runtime.Field(i)
+		require.Equal(t, a.Name, b.Name)
+		require.Equal(t, a.Type, b.Type)
+		require.Equal(t, a.Tag, b.Tag)
+	}
+	require.Equal(t, (graphqlCollectorState20260916{}).TableName(), (models.GraphqlCollectorState{}).TableName())
+}

backend/core/models/migrationscripts/register.go

@@ -152,5 +152,6 @@ func All() []plugin.MigrationScript {
 		new(addSprintVelocityFields),
 		new(addBlueprintIdIndexToPipelines),
 		new(expandDomainTextColumns),
+		new(addGraphqlCollectorStates),
 	}
 }

backend/core/plugin/plugin_task.go

@@ -19,6 +19,8 @@ package plugin
 
 import (
 	"context"
+	"sync"
+	"time"
 
 	corecontext "github.com/apache/devlake/core/context"
 	"github.com/apache/devlake/core/errors"
@@ -27,6 +29,49 @@ import (
 
 type ProgressType int
 
+type taskIDContextKey struct{}
+type taskStartContextKey struct{}
+type taskCodeContextKey struct{}
+
+var loadedPluginCode sync.Map
+
+// RegisterPluginCode records the loaded shared object's digest, not its path.
+func RegisterPluginCode(name, digest string) { loadedPluginCode.Store(name, digest) }
+
+func WithTaskCode(ctx context.Context, name string) context.Context {
+	digest, _ := loadedPluginCode.Load(name)
+	if digest == nil {
+		digest = ""
+	}
+	return context.WithValue(ctx, taskCodeContextKey{}, digest)
+}
+
+func TaskCode(ctx context.Context) string {
+	digest, _ := ctx.Value(taskCodeContextKey{}).(string)
+	return digest
+}
+
+func WithTaskStartedAt(ctx context.Context, started time.Time) context.Context {
+	return context.WithValue(ctx, taskStartContextKey{}, started)
+}
+
+func TaskStartedAt(ctx context.Context) (time.Time, bool) {
+	started, ok := ctx.Value(taskStartContextKey{}).(time.Time)
+	return started, ok
+}
+
+// WithTaskID identifies the persisted task across process restarts, without
+// changing the TaskContext interface implemented by plugins and test clients.
+func WithTaskID(ctx context.Context, id uint64) context.Context {
+	return context.WithValue(ctx, taskIDContextKey{}, id)
+}
+
+// TaskID returns zero for callers outside the persisted task runner.
+func TaskID(ctx context.Context) uint64 {
+	id, _ := ctx.Value(taskIDContextKey{}).(uint64)
+	return id
+}
+
 const (
 	TaskSetProgress ProgressType = iota
 	TaskIncProgress

backend/core/runner/loader.go

@@ -18,8 +18,12 @@ limitations under the License.
 package runner
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
+	"io"
 	"io/fs"
+	"os"
 	"path/filepath"
 	goplugin "plugin"
 	"strings"
@@ -70,6 +74,20 @@ func LoadGoPlugins(basicRes context.BasicRes) errors.Error {
 			if !ok {
 				return errors.Default.New(fmt.Sprintf("%s PluginEntry must implement PluginMeta interface", pluginName))
 			}
+			file, readErr := os.Open(path)
+			if readErr != nil {
+				return readErr
+			}
+			hash := sha256.New()
+			_, readErr = io.Copy(hash, file)
+			closeErr := file.Close()
+			if readErr != nil {
+				return readErr
+			}
+			if closeErr != nil {
+				return closeErr
+			}
+			plugin.RegisterPluginCode(pluginName, hex.EncodeToString(hash.Sum(nil)))
 			wg.Add(1)
 			go func(pluginName string, pluginMeta plugin.PluginMeta) {
 				defer func() {

backend/core/runner/run_task.go

@@ -134,6 +134,7 @@ func RunTask(
 	}
 
 	// start execution
+	task.BeganAt = &beganAt
 	logger.Info("start executing task: %d", task.ID)
 	dbe := db.UpdateColumns(task, []dal.DalSet{
 		{ColumnName: "status", Value: models.TASK_RUNNING},
@@ -248,6 +249,11 @@ func RunPluginSubTasks(
 		}
 	}
 
+	ctx = plugin.WithTaskID(ctx, task.ID)
+	ctx = plugin.WithTaskCode(ctx, task.Plugin)
+	if task.BeganAt != nil {
+		ctx = plugin.WithTaskStartedAt(ctx, *task.BeganAt)
+	}
 	taskCtx := contextimpl.NewDefaultTaskContext(ctx, basicRes, task.Plugin, subtasksFlag, progress)
 	if closeablePlugin, ok := pluginTask.(plugin.CloseablePluginTask); ok {
 		defer closeablePlugin.Close(taskCtx)
@@ -294,8 +300,16 @@ func RunPluginSubTasks(
 		}
 		subtask = append(subtask, s)
 	}
-	if err := basicRes.GetDal().CreateOrUpdate(subtask); err != nil {
-		basicRes.GetLogger().Error(err, "error writing subtask list to DB")
+	for i := range subtask {
+		count, err := basicRes.GetDal().Count(dal.From(&models.Subtask{}), dal.Where("task_id = ? AND name = ?", task.ID, subtask[i].Name))
+		if err != nil {
+			return err
+		}
+		if count == 0 {
+			if err := basicRes.GetDal().Create(&subtask[i]); err != nil {
+				return err
+			}
+		}
 	}
 
 	// execute subtasks in order
@@ -324,7 +338,7 @@ func RunPluginSubTasks(
 		if !subtaskMeta.ForceRunOnResume {
 			if task.ID > 0 {
 				sfc := errors.Must1(basicRes.GetDal().Count(
-					dal.From(&models.Subtask{}), dal.Where("task_id = ? AND name = ? AND finished_at IS NOT NULL", task.ID, subtaskMeta.Name),
+					dal.From(&models.Subtask{}), dal.Where("task_id = ? AND name = ? AND finished_at IS NOT NULL AND is_failed = ?", task.ID, subtaskMeta.Name, false),
 				),
 				)
 				subtaskFinished = sfc > 0
@@ -421,29 +435,40 @@ func runSubtask(
 		Number:  subtaskNumber,
 		BeganAt: &beginAt,
 	}
-	recordSubtask(basicRes, subtask)
-	// defer to record subtask status
-	defer func() {
-		finishedAt := time.Now()
+	if err := recordSubtask(basicRes, subtask); err != nil {
+		return err
+	}
+	// A panic leaves this subtask unfinished. Only successful execution may
+	// create the marker used to skip work after a process restart.
+	err := entryPoint(ctx)
+	finishedAt := time.Now()
+	subtask.IsFailed = err != nil
+	if err == nil {
 		subtask.FinishedAt = &finishedAt
-		subtask.SpentSeconds = finishedAt.Unix() - beginAt.Unix()
-
-		recordSubtask(basicRes, subtask)
-	}()
-	return entryPoint(ctx)
+	} else {
+		subtask.Message = err.Error()
+	}
+	subtask.SpentSeconds = finishedAt.Unix() - beginAt.Unix()
+	if recordErr := recordSubtask(basicRes, subtask); recordErr != nil {
+		if err == nil {
+			return recordErr
+		}
+		return errors.Default.Combine([]error{err, recordErr})
+	}
+	return err
 }
 
-func recordSubtask(basicRes context.BasicRes, subtask *models.Subtask) {
+func recordSubtask(basicRes context.BasicRes, subtask *models.Subtask) errors.Error {
 	where := dal.Where("task_id = ? and name = ?", subtask.TaskID, subtask.Name)
-	if err := basicRes.GetDal().UpdateColumns(subtask, []dal.DalSet{
+	return basicRes.GetDal().UpdateColumns(subtask, []dal.DalSet{
 		{ColumnName: "began_at", Value: subtask.BeganAt},
 		{ColumnName: "finished_at", Value: subtask.FinishedAt},
 		{ColumnName: "spent_seconds", Value: subtask.SpentSeconds},
 		//{ColumnName: "finished_records", Value: subtask.FinishedRecords}, // FinishedRecords is zero always.
 		{ColumnName: "number", Value: subtask.Number},
-	}, where); err != nil {
-		basicRes.GetLogger().Error(err, "error writing subtask %d status to DB: %v", subtask.ID)
-	}
+		{ColumnName: "is_failed", Value: subtask.IsFailed},
+		{ColumnName: "message", Value: subtask.Message},
+	}, where)
 }
 
 func getTaskLogger(parentLogger log.Logger, task *models.Task) (log.Logger, errors.Error) {