Refactor use store in server scheduler

.golangci.yaml

@@ -197,6 +197,8 @@ linters:
           alias: service_log
         - pkg: go.woodpecker-ci.org/woodpecker/v3/server/rpc
           alias: server_rpc
+        - pkg: go.woodpecker-ci.org/woodpecker/v3/server/status
+          alias: pipeline_status
         - pkg: go.woodpecker-ci.org/woodpecker/v3/agent/rpc
           alias: agent_rpc
         - pkg: go.woodpecker-ci.org/woodpecker/v3/shared/utils

server/api/badge.go

@@ -30,7 +30,7 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v3/server/badges"
 	"go.woodpecker-ci.org/woodpecker/v3/server/ccmenu"
 	"go.woodpecker-ci.org/woodpecker/v3/server/model"
-	"go.woodpecker-ci.org/woodpecker/v3/server/pipeline"
+	pipeline_status "go.woodpecker-ci.org/woodpecker/v3/server/status"
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
 	"go.woodpecker-ci.org/woodpecker/v3/server/store/types"
 )
@@ -122,7 +122,7 @@ func GetBadge(c *gin.Context) {
 					stepName := c.Query("step")
 					if len(stepName) == 0 {
 						if status != nil {
-							merged := pipeline.MergeStatusValues(*status, wf.State)
+							merged := pipeline_status.MergeStatusValues(*status, wf.State)
 							status = &merged
 						} else {
 							status = &wf.State
@@ -134,7 +134,7 @@ func GetBadge(c *gin.Context) {
 					for _, s := range wf.Children {
 						if s.Name == stepName {
 							if status != nil {
-								merged := pipeline.MergeStatusValues(*status, s.State)
+								merged := pipeline_status.MergeStatusValues(*status, s.State)
 								status = &merged
 							} else {
 								status = &s.State

server/pipeline/cancel.go

@@ -16,7 +16,6 @@ package pipeline
 
 import (
 	"context"
-	"fmt"
 	"slices"
 
 	"github.com/rs/zerolog/log"
@@ -38,59 +37,16 @@ func Cancel(ctx context.Context, _forge forge.Forge, store store.Store, repo *mo
 		return &ErrNotFound{Msg: err.Error()}
 	}
 
-	// First cancel/evict the running and pending workflows from the queue
-	var workflowsToCancel []string
-	for _, w := range workflows {
-		if w.State == model.StatusRunning || w.State == model.StatusPending {
-			workflowsToCancel = append(workflowsToCancel, fmt.Sprint(w.ID))
-		}
-	}
-
-	if err := server.Config.Services.Scheduler.CancelWorkflows(ctx, workflowsToCancel); err != nil {
-		log.Error().Err(err).Msgf("cancel workflows: %v", workflowsToCancel)
-	}
-
-	hasPendingOnly := true
-
-	// Then update the DB status for pending pipelines
-	// Running ones will be set when the agents stop on the cancel signal
-	for _, workflow := range workflows {
-		if workflow.State == model.StatusPending {
-			if _, err = UpdateWorkflowToStatusSkipped(store, *workflow); err != nil {
-				log.Error().Err(err).Msgf("cannot update workflow with id %d state", workflow.ID)
-			}
-		} else {
-			hasPendingOnly = false
-		}
-		for _, step := range workflow.Children {
-			if step.State == model.StatusPending {
-				if _, err = UpdateStepToStatusSkipped(store, *step, 0, model.StatusCanceled); err != nil {
-					log.Error().Err(err).Msgf("cannot update workflow with id %d state", workflow.ID)
-				}
-			}
-		}
-	}
-
-	plState := model.StatusKilled
-	if hasPendingOnly {
-		plState = model.StatusCanceled
-	}
-	killedPipeline, err := UpdateToStatusKilled(store, *pipeline, cancelInfo, plState)
+	// The scheduler owns the cancellation: it evicts the workflows from the
+	// queue, persists the skipped/killed state and notifies subscribers. The
+	// only thing left for us is to sync the forge status afterwards.
+	killedPipeline, err := server.Config.Services.Scheduler.CancelWorkflows(ctx, repo, pipeline, workflows, cancelInfo)
 	if err != nil {
-		log.Error().Err(err).Msgf("UpdateToStatusKilled: %v", pipeline)
 		return err
 	}
 
 	updatePipelineStatus(ctx, _forge, killedPipeline, repo, user)
 
-	if killedPipeline.Workflows, err = store.WorkflowGetTree(killedPipeline); err != nil {
-		return err
-	}
-
-	if err := server.Config.Services.Scheduler.PublishPipelineEvent(ctx, repo, killedPipeline); err != nil {
-		log.Error().Err(err).Msg("could not push pipeline status change to pubsub provider")
-	}
-
 	return nil
 }
 

server/pipeline/pipeline_status.go

@@ -23,17 +23,6 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
 )
 
-// PipelineStatus determine pipeline status based on corresponding workflow list.
-func PipelineStatus(workflows []*model.Workflow) model.StatusValue {
-	status := model.StatusSuccess
-
-	for _, p := range workflows {
-		status = MergeStatusValues(status, p.State)
-	}
-
-	return status
-}
-
 func UpdateToStatusRunning(store store.Store, pipeline model.Pipeline, started int64) (*model.Pipeline, error) {
 	pipeline.Status = model.StatusRunning
 	pipeline.Started = started
@@ -69,10 +58,3 @@ func UpdateToStatusError(store store.Store, pipeline model.Pipeline, err error)
 	pipeline.Finished = pipeline.Started
 	return &pipeline, store.UpdatePipeline(&pipeline)
 }
-
-func UpdateToStatusKilled(store store.Store, pipeline model.Pipeline, cancelInfo *model.CancelInfo, state model.StatusValue) (*model.Pipeline, error) {
-	pipeline.Status = state
-	pipeline.Finished = time.Now().Unix()
-	pipeline.CancelInfo = cancelInfo
-	return &pipeline, store.UpdatePipeline(&pipeline)
-}

server/pipeline/pipeline_status_test.go

@@ -89,19 +89,3 @@ func TestUpdateToStatusError(t *testing.T) {
 	assert.LessOrEqual(t, now, pipeline.Started)
 	assert.Equal(t, pipeline.Started, pipeline.Finished)
 }
-
-func TestUpdateToStatusKilled(t *testing.T) {
-	t.Parallel()
-
-	now := time.Now().Unix()
-	cancelInfo := &model.CancelInfo{
-		SupersededBy: 2,
-	}
-
-	pipeline, _ := UpdateToStatusKilled(mockStorePipeline(t), model.Pipeline{}, cancelInfo, model.StatusKilled)
-
-	assert.Equal(t, model.StatusKilled, pipeline.Status)
-	assert.NotNil(t, pipeline.CancelInfo)
-	assert.EqualValues(t, 2, pipeline.CancelInfo.SupersededBy)
-	assert.LessOrEqual(t, now, pipeline.Finished)
-}

server/pipeline/step_status.go

@@ -150,12 +150,3 @@ func cancelPipelineFromStep(ctx context.Context, store store.Store, step *model.
 		CanceledByStep: step.Name,
 	})
 }
-
-func UpdateStepToStatusSkipped(store store.Store, step model.Step, finished int64, status model.StatusValue) (*model.Step, error) {
-	step.State = status
-	if step.Started != 0 {
-		step.State = model.StatusSuccess // for daemons that are killed
-		step.Finished = finished
-	}
-	return &step, store.StepUpdate(&step)
-}

server/pipeline/step_status_test.go

@@ -283,27 +283,3 @@ func TestUpdateStepStatus(t *testing.T) {
 		assert.Equal(t, model.StatusKilled, step.State)
 	})
 }
-
-func TestUpdateStepToStatusSkipped(t *testing.T) {
-	t.Parallel()
-
-	t.Run("NotStarted", func(t *testing.T) {
-		t.Parallel()
-
-		step, err := UpdateStepToStatusSkipped(mockStoreStep(t), model.Step{}, int64(1), model.StatusSkipped)
-
-		assert.NoError(t, err)
-		assert.Equal(t, model.StatusSkipped, step.State)
-		assert.Equal(t, int64(0), step.Finished)
-	})
-
-	t.Run("AlreadyStarted", func(t *testing.T) {
-		t.Parallel()
-
-		step, err := UpdateStepToStatusSkipped(mockStoreStep(t), model.Step{Started: 42}, int64(100), model.StatusSkipped)
-
-		assert.NoError(t, err)
-		assert.Equal(t, model.StatusSuccess, step.State)
-		assert.Equal(t, int64(100), step.Finished)
-	})
-}

server/pipeline/workflow_status.go

@@ -20,43 +20,8 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v3/server/store"
 )
 
-// WorkflowStatus determine workflow status based on corresponding step list.
-func WorkflowStatus(steps []*model.Step) model.StatusValue {
-	status := model.StatusSuccess
-
-	for _, p := range steps {
-		if p.Failure == model.FailureFail || !p.Failing() {
-			status = MergeStatusValues(status, p.State)
-		}
-	}
-
-	return status
-}
-
 func UpdateWorkflowStatusToRunning(store store.Store, workflow model.Workflow, state rpc.WorkflowState) (*model.Workflow, error) {
 	workflow.Started = state.Started
 	workflow.State = model.StatusRunning
 	return &workflow, store.WorkflowUpdate(&workflow)
 }
-
-func UpdateWorkflowToStatusSkipped(store store.Store, workflow model.Workflow) (*model.Workflow, error) {
-	workflow.State = model.StatusSkipped
-	return &workflow, store.WorkflowUpdate(&workflow)
-}
-
-func UpdateWorkflowStatusToDone(store store.Store, workflow model.Workflow, state rpc.WorkflowState) (*model.Workflow, error) {
-	workflow.Finished = state.Finished
-	workflow.Error = state.Error
-	if state.Started == 0 {
-		workflow.State = model.StatusSkipped
-	} else {
-		workflow.State = WorkflowStatus(workflow.Children)
-	}
-	if workflow.Error != "" {
-		workflow.State = model.StatusFailure
-	}
-	if state.Canceled {
-		workflow.State = model.StatusKilled
-	}
-	return &workflow, store.WorkflowUpdate(&workflow)
-}