diff --git a/.golangci.yaml b/.golangci.yaml index be360a1f287..fc10b954906 100644 --- a/.golangci.yaml +++ b/.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 diff --git a/server/api/badge.go b/server/api/badge.go index b6400095b56..379b339842b 100644 --- a/server/api/badge.go +++ b/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 diff --git a/server/pipeline/cancel.go b/server/pipeline/cancel.go index 7e34b8d632e..8f81d9f0625 100644 --- a/server/pipeline/cancel.go +++ b/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 } diff --git a/server/pipeline/pipeline_status.go b/server/pipeline/pipeline_status.go index 2ab801b0dbb..58a7affc757 100644 --- a/server/pipeline/pipeline_status.go +++ b/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) -} diff --git a/server/pipeline/pipeline_status_test.go b/server/pipeline/pipeline_status_test.go index e0a1b0f3a06..9568832fb08 100644 --- a/server/pipeline/pipeline_status_test.go +++ b/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) -} diff --git a/server/pipeline/status_test.go b/server/pipeline/status_test.go deleted file mode 100644 index 1268da7c257..00000000000 --- a/server/pipeline/status_test.go +++ /dev/null @@ -1,91 +0,0 @@ -// Copyright 2026 Woodpecker Authors -// -// Licensed 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 pipeline - -import ( - "testing" - - "github.com/stretchr/testify/assert" - - "go.woodpecker-ci.org/woodpecker/v3/server/model" -) - -func TestStatusValueMerge(t *testing.T) { - tests := []struct { - s model.StatusValue - t model.StatusValue - e model.StatusValue - }{ - { - s: model.StatusSuccess, - t: model.StatusSkipped, - e: model.StatusSuccess, - }, - { - s: model.StatusSuccess, - t: model.StatusSuccess, - e: model.StatusSuccess, - }, - { - s: model.StatusFailure, - t: model.StatusSuccess, - e: model.StatusFailure, - }, - { - s: model.StatusRunning, - t: model.StatusSuccess, - e: model.StatusRunning, - }, - { - s: model.StatusRunning, - t: model.StatusFailure, - e: model.StatusRunning, - }, - { - s: model.StatusFailure, - t: model.StatusKilled, - e: model.StatusKilled, - }, - { - s: model.StatusSkipped, - t: model.StatusKilled, - e: model.StatusKilled, - }, - { - s: model.StatusSkipped, - t: model.StatusSkipped, - e: model.StatusSkipped, - }, - { - s: model.StatusSkipped, - t: model.StatusCanceled, - e: model.StatusKilled, - }, - { - s: model.StatusSuccess, - t: model.StatusCanceled, - e: model.StatusKilled, - }, - { - s: model.StatusFailure, - t: model.StatusCanceled, - e: model.StatusKilled, - }, - } - for _, tt := range tests { - assert.Equal(t, tt.e, MergeStatusValues(tt.s, tt.t)) - assert.Equal(t, tt.e, MergeStatusValues(tt.t, tt.s)) - } -} diff --git a/server/pipeline/step_status.go b/server/pipeline/step_status.go index 100a8a00acb..926488f4f18 100644 --- a/server/pipeline/step_status.go +++ b/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) -} diff --git a/server/pipeline/step_status_test.go b/server/pipeline/step_status_test.go index ae50a0c3974..1cf657d4b9b 100644 --- a/server/pipeline/step_status_test.go +++ b/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) - }) -} diff --git a/server/pipeline/workflow_status.go b/server/pipeline/workflow_status.go index 143408a103f..b33d232d098 100644 --- a/server/pipeline/workflow_status.go +++ b/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) -} diff --git a/server/pipeline/workflow_status_test.go b/server/pipeline/workflow_status_test.go index 89b921bbcb7..f697044db37 100644 --- a/server/pipeline/workflow_status_test.go +++ b/server/pipeline/workflow_status_test.go @@ -25,160 +25,6 @@ import ( store_mocks "go.woodpecker-ci.org/woodpecker/v3/server/store/mocks" ) -func TestWorkflowStatus(t *testing.T) { - tests := []struct { - s []*model.Step - e model.StatusValue - }{ - { - s: []*model.Step{ - { - State: model.StatusFailure, - Failure: model.FailureIgnore, - }, - { - State: model.StatusSuccess, - Failure: model.FailureFail, - }, - }, - e: model.StatusSuccess, - }, - { - s: []*model.Step{ - { - State: model.StatusSuccess, - Failure: model.FailureFail, - }, - { - State: model.StatusSuccess, - Failure: model.FailureIgnore, - }, - }, - e: model.StatusSuccess, - }, - { - s: []*model.Step{ - { - State: model.StatusFailure, - Failure: model.FailureFail, - }, - { - State: model.StatusSuccess, - Failure: model.FailureFail, - }, - }, - e: model.StatusFailure, - }, - { - s: []*model.Step{ - { - State: model.StatusSuccess, - Failure: model.FailureFail, - }, - { - State: model.StatusPending, - Failure: model.FailureFail, - }, - }, - e: model.StatusPending, - }, - { - s: []*model.Step{ - { - State: model.StatusSuccess, - Failure: model.FailureFail, - }, - { - State: model.StatusPending, - Failure: model.FailureIgnore, - }, - }, - e: model.StatusPending, - }, - { - s: []*model.Step{ - { - State: model.StatusSuccess, - Failure: model.FailureIgnore, - }, - { - State: model.StatusPending, - Failure: model.FailureFail, - }, - }, - e: model.StatusPending, - }, - { - s: []*model.Step{ - { - State: model.StatusSuccess, - Failure: model.FailureIgnore, - }, - { - State: model.StatusPending, - Failure: model.FailureIgnore, - }, - }, - e: model.StatusPending, - }, - { - s: []*model.Step{ - { - State: model.StatusRunning, - Failure: model.FailureFail, - }, - { - State: model.StatusPending, - Failure: model.FailureFail, - }, - }, - e: model.StatusRunning, - }, - { - s: []*model.Step{ - { - State: model.StatusRunning, - Failure: model.FailureIgnore, - }, - { - State: model.StatusPending, - Failure: model.FailureIgnore, - }, - }, - e: model.StatusRunning, - }, - { - s: []*model.Step{ - { - State: model.StatusRunning, - Failure: model.FailureIgnore, - }, - { - State: model.StatusPending, - Failure: model.FailureFail, - }, - }, - e: model.StatusRunning, - }, - { - s: []*model.Step{ - { - State: model.StatusRunning, - Failure: model.FailureFail, - }, - { - State: model.StatusPending, - Failure: model.FailureIgnore, - }, - }, - e: model.StatusRunning, - }, - } - for _, tt := range tests { - assert.Equal(t, tt.e, WorkflowStatus(tt.s)) - } -} - func TestUpdateWorkflowStatusToRunning(t *testing.T) { t.Run("should update workflow to running status", func(t *testing.T) { workflow := model.Workflow{ @@ -202,99 +48,3 @@ func TestUpdateWorkflowStatusToRunning(t *testing.T) { mockStore.AssertCalled(t, "WorkflowUpdate", mock.Anything) }) } - -func TestUpdateWorkflowToStatusSkipped(t *testing.T) { - t.Run("should update workflow to skipped status", func(t *testing.T) { - workflow := model.Workflow{ - ID: 2, - State: model.StatusPending, - } - - mockStore := store_mocks.NewMockStore(t) - mockStore.On("WorkflowUpdate", mock.MatchedBy(func(w *model.Workflow) bool { - return w.ID == 2 && w.State == model.StatusSkipped - })).Return(nil) - - result, err := UpdateWorkflowToStatusSkipped(mockStore, workflow) - - assert.NoError(t, err) - assert.Equal(t, model.StatusSkipped, result.State) - mockStore.AssertCalled(t, "WorkflowUpdate", mock.Anything) - }) -} - -func TestUpdateWorkflowStatusToDone(t *testing.T) { - t.Run("should mark as skipped when not started", func(t *testing.T) { - workflow := model.Workflow{ - ID: 3, - State: model.StatusRunning, - Children: []*model.Step{}, - } - state := rpc.WorkflowState{ - Started: 0, // Not started - Finished: 1234567900, - Error: "", - } - - mockStore := store_mocks.NewMockStore(t) - mockStore.On("WorkflowUpdate", mock.MatchedBy(func(w *model.Workflow) bool { - return w.State == model.StatusSkipped && w.Finished == 1234567900 - })).Return(nil) - - result, err := UpdateWorkflowStatusToDone(mockStore, workflow, state) - - assert.NoError(t, err) - assert.Equal(t, model.StatusSkipped, result.State) - assert.Equal(t, int64(1234567900), result.Finished) - }) - - t.Run("should mark as failure when error exists", func(t *testing.T) { - workflow := model.Workflow{ - ID: 5, - State: model.StatusRunning, - Children: []*model.Step{}, - } - state := rpc.WorkflowState{ - Started: 1234567800, - Finished: 1234567900, - Error: "some error occurred", - } - - mockStore := store_mocks.NewMockStore(t) - mockStore.On("WorkflowUpdate", mock.MatchedBy(func(w *model.Workflow) bool { - return w.State == model.StatusFailure - })).Return(nil) - - result, err := UpdateWorkflowStatusToDone(mockStore, workflow, state) - - assert.NoError(t, err) - assert.Equal(t, model.StatusFailure, result.State) - assert.Equal(t, "some error occurred", result.Error) - }) - - t.Run("should mark as success when all children are successful", func(t *testing.T) { - successStep := &model.Step{ - ID: 1, - State: model.StatusSuccess, - } - workflow := model.Workflow{ - ID: 6, - State: model.StatusRunning, - Children: []*model.Step{successStep}, - } - state := rpc.WorkflowState{ - Started: 1234567800, - Finished: 1234567900, - Error: "", - } - - mockStore := store_mocks.NewMockStore(t) - mockStore.On("WorkflowUpdate", mock.Anything).Return(nil) - - result, err := UpdateWorkflowStatusToDone(mockStore, workflow, state) - - assert.NoError(t, err) - assert.Equal(t, model.StatusSuccess, result.State) - assert.Equal(t, int64(1234567900), result.Finished) - }) -} diff --git a/server/rpc/rpc.go b/server/rpc/rpc.go index bea83c565f8..78b7894466d 100644 --- a/server/rpc/rpc.go +++ b/server/rpc/rpc.go @@ -83,19 +83,19 @@ func (s *RPC) Next(c context.Context, agentFilter rpc.Filter) (*rpc.Workflow, er log.Trace().Msgf("Agent %s[%d] tries to pull task with labels: %v", agent.Name, agent.ID, agentFilter.Labels) - rpcWorkflow, err := s.scheduler.Poll(c, agent.ID, agentFilter, func(taskID string) error { - // a skipped workflow is finalized through the regular Done flow; it - // was never initialized by an agent, so lock it to this agent first - // to satisfy the workflow ownership check. - if err := s.lockAgentToWorkflow(c, agent, taskID); err != nil { - return err - } - return s.Done(c, taskID, rpc.WorkflowState{}) + rpcWorkflow, err := s.scheduler.Poll(c, agent.ID, agentFilter, func(repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow) { + // the scheduler finalized a skipped workflow; sync its status to the + // forge and record metrics, the same caller-side follow-up Done does. + s.updateForgeStatus(c, repo, pipeline, workflow) + s.recordPipelineMetrics(repo, pipeline, workflow) }) if err != nil || rpcWorkflow == nil { return nil, err } + // Lock the polled workflow to this agent so subsequent agent RPCs (Init, + // Update, Wait, ...) pass the ownership check in sanitize.go. The scheduler + // hands out the workflow but does not record the owning agent itself. if err := s.lockAgentToWorkflow(c, agent, rpcWorkflow.ID); err != nil { return nil, err } @@ -355,43 +355,14 @@ func (s *RPC) Done(c context.Context, strWorkflowID string, state rpc.WorkflowSt logger.Debug().Msgf("workflow state in store: %#v", workflow) logger.Debug().Msgf("gRPC Done with state: %#v", state) - // Complete any still-running children (e.g. service containers) before - // computing the workflow status, so their final state is reflected. - s.completeChildrenIfParentCompleted(workflow, state.Finished) - - if workflow, err = pipeline.UpdateWorkflowStatusToDone(s.store, *workflow, state); err != nil { - logger.Error().Err(err).Msgf("pipeline.UpdateWorkflowStatusToDone: cannot update workflow state: %s", err) - } - - var queueErr error - if !state.Canceled { - if workflow.Failing() { - queueErr = s.scheduler.Error(c, strWorkflowID, fmt.Errorf("workflow finished with error %s", state.Error)) - } else { - queueErr = s.scheduler.Done(c, strWorkflowID, workflow.State) - } - } else { - if workflow.Started > 0 { - queueErr = s.scheduler.Done(c, strWorkflowID, model.StatusKilled) - } else { - queueErr = s.scheduler.Done(c, strWorkflowID, model.StatusCanceled) - } - } - if queueErr != nil { - logger.Error().Err(queueErr).Msg("queue.Done: cannot ack workflow") - } - - currentPipeline.Workflows, err = s.store.WorkflowGetTree(currentPipeline) + // The scheduler owns the workflow completion: it finalizes the workflow and + // its children, signals the queue, rolls the pipeline up and notifies + // subscribers. We are left with the forge sync, log streams and metrics. + currentPipeline, workflow, err = s.scheduler.FinishWorkflow(c, repo, currentPipeline, workflow, state) if err != nil { return err } - if !model.IsThereRunningStage(currentPipeline.Workflows) { - if currentPipeline, err = pipeline.UpdateStatusToDone(s.store, *currentPipeline, pipeline.PipelineStatus(currentPipeline.Workflows), workflow.Finished); err != nil { - logger.Error().Err(err).Msgf("pipeline.UpdateStatusToDone: cannot update workflows final state") - } - } - s.updateForgeStatus(c, repo, currentPipeline, workflow) // make sure writes to pubsub are non blocking (https://github.com/woodpecker-ci/woodpecker/blob/c919f32e0b6432a95e1a6d3d0ad662f591adf73f/server/logging/log.go#L9) @@ -405,19 +376,22 @@ func (s *RPC) Done(c context.Context, strWorkflowID string, state rpc.WorkflowSt } }() - if err := s.scheduler.PublishPipelineEvent(c, repo, currentPipeline); err != nil { - return err - } + s.recordPipelineMetrics(repo, currentPipeline, workflow) + + return s.updateAgentLastWork(agent) +} - if currentPipeline.Status == model.StatusSuccess || currentPipeline.Status == model.StatusFailure { - s.pipelineCount.WithLabelValues(repo.FullName, currentPipeline.Branch, string(currentPipeline.Status), "total").Inc() - s.pipelineTime.WithLabelValues(repo.FullName, currentPipeline.Branch, string(currentPipeline.Status), "total").Set(float64(currentPipeline.Finished - currentPipeline.Started)) +// recordPipelineMetrics records the prometheus metrics for a finished workflow +// and its pipeline. It is shared by the regular Done path and the skipped +// workflow follow-up so both report consistently. +func (s *RPC) recordPipelineMetrics(repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow) { + if pipeline.Status == model.StatusSuccess || pipeline.Status == model.StatusFailure { + s.pipelineCount.WithLabelValues(repo.FullName, pipeline.Branch, string(pipeline.Status), "total").Inc() + s.pipelineTime.WithLabelValues(repo.FullName, pipeline.Branch, string(pipeline.Status), "total").Set(float64(pipeline.Finished - pipeline.Started)) } - if currentPipeline.IsMultiPipeline() { - s.pipelineTime.WithLabelValues(repo.FullName, currentPipeline.Branch, string(workflow.State), workflow.Name).Set(float64(workflow.Finished - workflow.Started)) + if pipeline.IsMultiPipeline() { + s.pipelineTime.WithLabelValues(repo.FullName, pipeline.Branch, string(workflow.State), workflow.Name).Set(float64(workflow.Finished - workflow.Started)) } - - return s.updateAgentLastWork(agent) } // Log writes a log entry to the database and publishes it to the pubsub. @@ -549,20 +523,6 @@ func (s *RPC) ReportHealth(ctx context.Context, status string) error { return s.store.AgentUpdate(agent) } -func (s *RPC) completeChildrenIfParentCompleted(completedWorkflow *model.Workflow, finished int64) { - for _, c := range completedWorkflow.Children { - if c.Running() { - if updated, err := pipeline.UpdateStepToStatusSkipped(s.store, *c, finished, model.StatusKilled); err != nil { - log.Error().Err(err).Msgf("done: cannot update step_id %d child state", c.ID) - } else { - // Update in-memory state so WorkflowStatus sees the final state - c.State = updated.State - c.Finished = updated.Finished - } - } - } -} - func (s *RPC) updateForgeStatus(ctx context.Context, repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow) { user, err := s.store.GetUser(repo.UserID) if err != nil { diff --git a/server/rpc/rpc_test.go b/server/rpc/rpc_test.go index a5323fe3252..22ecc8f8e67 100644 --- a/server/rpc/rpc_test.go +++ b/server/rpc/rpc_test.go @@ -26,7 +26,6 @@ import ( "go.woodpecker-ci.org/woodpecker/v3/rpc" "go.woodpecker-ci.org/woodpecker/v3/server/model" - "go.woodpecker-ci.org/woodpecker/v3/server/pipeline" store_mocks "go.woodpecker-ci.org/woodpecker/v3/server/store/mocks" ) @@ -113,43 +112,6 @@ func TestRegisterAgent(t *testing.T) { }) } -func TestCompleteChildrenIfParentCompleted(t *testing.T) { - t.Run("When a service step is still running it should update state so workflow finishes as success", func(t *testing.T) { - successStep := &model.Step{ - ID: 1, - State: model.StatusSuccess, - Started: 1234567800, - } - runningService := &model.Step{ - ID: 2, - State: model.StatusRunning, - Started: 1234567800, - } - workflow := model.Workflow{ - ID: 7, - State: model.StatusRunning, - Children: []*model.Step{successStep, runningService}, - } - - mockStore := store_mocks.NewMockStore(t) - mockStore.On("StepUpdate", mock.Anything).Return(nil) - mockStore.On("WorkflowUpdate", mock.Anything).Return(nil) - - s := RPC{store: mockStore} - s.completeChildrenIfParentCompleted(&workflow, 1234567900) - - assert.Equal(t, model.StatusSuccess, runningService.State) - assert.Equal(t, int64(1234567900), runningService.Finished) - - result, err := pipeline.UpdateWorkflowStatusToDone(mockStore, workflow, rpc.WorkflowState{ - Started: 1234567800, - Finished: 1234567900, - }) - require.NoError(t, err) - assert.Equal(t, model.StatusSuccess, result.State) - }) -} - func TestUpdateAgentLastWork(t *testing.T) { t.Run("When last work was never updated it should update last work timestamp", func(t *testing.T) { agent := model.Agent{ diff --git a/server/scheduler/impl.go b/server/scheduler/impl.go index d41ba2aefa5..e9dcee091e8 100644 --- a/server/scheduler/impl.go +++ b/server/scheduler/impl.go @@ -18,6 +18,9 @@ import ( "context" "encoding/json" "fmt" + "strconv" + "sync" + "time" "github.com/oklog/ulid/v2" "github.com/rs/zerolog/log" @@ -26,6 +29,7 @@ import ( "go.woodpecker-ci.org/woodpecker/v3/server/model" "go.woodpecker-ci.org/woodpecker/v3/server/pubsub" "go.woodpecker-ci.org/woodpecker/v3/server/queue" + "go.woodpecker-ci.org/woodpecker/v3/server/status" "go.woodpecker-ci.org/woodpecker/v3/server/store" ) @@ -35,6 +39,9 @@ type impl struct { store store.Store q queue.Queue ps pubsub.PubSub + + paused bool + lock sync.RWMutex } // @@ -61,13 +68,7 @@ func (p *impl) KickAgentWorkers(agentID int64) { p.q.KickAgentWorkers(agentID) } -func (p *impl) Pause() { - p.q.Pause() -} - -// TODO: markSkipped is a callback helper that is only needed as we use the rpc.Done to mark skipped workflows as done -// this is a hack for another refactor later. -func (p *impl) Poll(c context.Context, agentID int64, agentFilter rpc.Filter, markSkipped func(taskID string) error) (*rpc.Workflow, error) { +func (p *impl) Poll(c context.Context, agentID int64, agentFilter rpc.Filter, onSkipped SkippedWorkflowFunc) (*rpc.Workflow, error) { filter := createFilterFunc(agentFilter) for { @@ -83,15 +84,79 @@ func (p *impl) Poll(c context.Context, agentID int64, agentFilter rpc.Filter, ma return workflow, err } - // task should not run, so let the caller mark it as done - if err := markSkipped(task.ID); err != nil { - log.Error().Err(err).Msgf("marking workflow task '%s' as done failed", task.ID) + // the task's dependencies preclude it from running, so finalize its + // workflow as skipped before polling the next task. + if err := p.finalizeSkippedWorkflow(c, task.ID, onSkipped); err != nil { + log.Error().Err(err).Msgf("marking workflow task '%s' as skipped failed", task.ID) } } } +// finalizeSkippedWorkflow marks the workflow with the given ID as skipped, +// reusing the regular workflow-completion path. An empty WorkflowState +// (Started == 0) makes FinishWorkflow resolve the workflow to its skipped +// state. Once the scheduling state is persisted and published, onSkipped is +// invoked so the caller can sync the workflow's status to the forge, the one +// follow-up the scheduler cannot perform itself. +func (p *impl) finalizeSkippedWorkflow(c context.Context, taskID string, onSkipped SkippedWorkflowFunc) error { + workflowID, err := strconv.ParseInt(taskID, 10, 64) + if err != nil { + return err + } + + workflow, err := p.store.WorkflowLoad(workflowID) + if err != nil { + return err + } + + // only finalize a workflow that has not reached a terminal or blocked state + // yet, mirroring the guard the previous rpc.Done path applied. + switch workflow.State { + case model.StatusCreated, model.StatusPending, model.StatusRunning: + default: + return nil + } + + if workflow.Children, err = p.store.StepListFromWorkflowFind(workflow); err != nil { + return err + } + + pipeline, err := p.store.GetPipeline(workflow.PipelineID) + if err != nil { + return err + } + + repo, err := p.store.GetRepo(pipeline.RepoID) + if err != nil { + return err + } + + pipeline, workflow, err = p.FinishWorkflow(c, repo, pipeline, workflow, rpc.WorkflowState{}) + if err != nil { + return err + } + + if onSkipped != nil { + onSkipped(repo, pipeline, workflow) + } + + return nil +} + +func (p *impl) Pause() { + p.lock.Lock() + defer p.lock.Unlock() + if !p.paused { + p.q.Pause() + } +} + func (p *impl) Resume() { - p.q.Resume() + p.lock.Lock() + defer p.lock.Unlock() + if !p.paused { + p.q.Resume() + } } func (p *impl) Wait(c context.Context, id string) error { @@ -149,13 +214,166 @@ func (p *impl) StartPipeline(c context.Context, repo *model.Repo, pipeline *mode return p.q.PushAtOnce(c, tasks) } -// CancelWorkflows evicts the given workflows from the queue, signaling a -// cancellation (queue.ErrCancel) to any agents currently waiting on them. -// An empty list is a no-op. -func (p *impl) CancelWorkflows(c context.Context, workflowIDs []string) error { - if len(workflowIDs) == 0 { - return nil +// CancelWorkflows owns the full cancellation of a pipeline's workflows. It +// evicts the running/pending workflows from the queue first (so any waiting +// agents receive the cancellation signal as early as possible), then marks the +// still-pending workflows and steps as skipped, transitions the pipeline to its +// killed state, and publishes the resulting state change. The returned pipeline +// carries its refreshed workflow tree so the caller can sync the forge status. +func (p *impl) CancelWorkflows(c context.Context, repo *model.Repo, pipeline *model.Pipeline, workflows []*model.Workflow, cancelInfo *model.CancelInfo) (*model.Pipeline, error) { + // First evict the running and pending workflows from the queue, signaling + // the cancellation (queue.ErrCancel) to any agents currently waiting on + // them. + var workflowIDs []string + for _, w := range workflows { + if w.State == model.StatusRunning || w.State == model.StatusPending { + workflowIDs = append(workflowIDs, fmt.Sprint(w.ID)) + } + } + if len(workflowIDs) > 0 { + if err := p.q.ErrorAtOnce(c, workflowIDs, queue.ErrCancel); err != nil { + log.Error().Err(err).Msgf("cancel workflows: %v", workflowIDs) + } + } + + // Mark the still-pending workflows and steps as skipped. Running ones are + // finalized by their agents once they observe the cancellation signal. + hasPendingOnly := true + for _, workflow := range workflows { + if workflow.State == model.StatusPending { + workflow.State = model.StatusSkipped + if err := p.store.WorkflowUpdate(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 { + step.State = model.StatusCanceled + if err := p.store.StepUpdate(step); err != nil { + log.Error().Err(err).Msgf("cannot update step with id %d state", step.ID) + } + } + } } - return p.q.ErrorAtOnce(c, workflowIDs, queue.ErrCancel) + plState := model.StatusKilled + if hasPendingOnly { + plState = model.StatusCanceled + } + pipeline.Status = plState + pipeline.Finished = time.Now().Unix() + pipeline.CancelInfo = cancelInfo + if err := p.store.UpdatePipeline(pipeline); err != nil { + log.Error().Err(err).Msgf("UpdateToStatusKilled: %v", pipeline) + return nil, err + } + + var err error + if pipeline.Workflows, err = p.store.WorkflowGetTree(pipeline); err != nil { + return nil, err + } + + if err := p.PublishPipelineEvent(c, repo, pipeline); err != nil { + log.Error().Err(err).Msg("could not push pipeline status change to pubsub provider") + } + + return pipeline, nil +} + +// FinishWorkflow owns the completion of a single workflow. It finalizes any +// still-running children, computes and persists the workflow's final state, +// acknowledges the workflow on the queue, rolls the pipeline up to its done +// state once no stage is left running, and publishes the resulting change. +// The updated pipeline (with its refreshed tree) and workflow are returned so +// the caller can sync the forge status, close log streams and record metrics, +// the concerns that do not belong to the scheduler. +func (p *impl) FinishWorkflow(c context.Context, repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow, state rpc.WorkflowState) (*model.Pipeline, *model.Workflow, error) { + // Complete any still-running children (e.g. service containers) before + // computing the workflow status, so their final state is reflected. + p.completeRunningChildren(workflow, state.Finished) + + updateWorkflowStateToDone(workflow, state) + if err := p.store.WorkflowUpdate(workflow); err != nil { + log.Error().Err(err).Msgf("cannot update workflow %d state", workflow.ID) + } + + if err := p.ackWorkflow(c, workflow, state); err != nil { + log.Error().Err(err).Msg("queue.Done: cannot ack workflow") + } + + var err error + if pipeline.Workflows, err = p.store.WorkflowGetTree(pipeline); err != nil { + return nil, nil, err + } + + if !model.IsThereRunningStage(pipeline.Workflows) { + pipeline.Status = status.PipelineStatus(pipeline.Workflows) + pipeline.Finished = workflow.Finished + if err := p.store.UpdatePipeline(pipeline); err != nil { + log.Error().Err(err).Msg("cannot update pipeline final state") + } + } + + if err := p.PublishPipelineEvent(c, repo, pipeline); err != nil { + log.Error().Err(err).Msg("could not push pipeline status change to pubsub provider") + } + + return pipeline, workflow, nil +} + +// completeRunningChildren finalizes the still-running steps of a completed +// workflow so the workflow status reflects their final state. A step that had +// already started (e.g. a service/daemon) is considered successful and gets its +// finish time set; one that never started is marked killed. +func (p *impl) completeRunningChildren(workflow *model.Workflow, finished int64) { + for _, child := range workflow.Children { + if !child.Running() { + continue + } + child.State = model.StatusKilled + if child.Started != 0 { + child.State = model.StatusSuccess // for daemons that are killed + child.Finished = finished + } + if err := p.store.StepUpdate(child); err != nil { + log.Error().Err(err).Msgf("done: cannot update step_id %d child state", child.ID) + } + } +} + +// ackWorkflow acknowledges the workflow on the queue, signaling either an error +// or a successful completion depending on the reported state. +func (p *impl) ackWorkflow(c context.Context, workflow *model.Workflow, state rpc.WorkflowState) error { + id := fmt.Sprint(workflow.ID) + + switch { + case state.Canceled && workflow.Started > 0: + return p.q.Done(c, id, model.StatusKilled) + case state.Canceled: + return p.q.Done(c, id, model.StatusCanceled) + case workflow.Failing(): + return p.q.Error(c, id, fmt.Errorf("workflow finished with error %s", state.Error)) + default: + return p.q.Done(c, id, workflow.State) + } +} + +// updateWorkflowStateToDone computes the final state of a finished workflow +// from its reported state and its children. +func updateWorkflowStateToDone(workflow *model.Workflow, state rpc.WorkflowState) { + workflow.Finished = state.Finished + workflow.Error = state.Error + if state.Started == 0 { + workflow.State = model.StatusSkipped + } else { + workflow.State = status.WorkflowStatus(workflow.Children) + } + if workflow.Error != "" { + workflow.State = model.StatusFailure + } + if state.Canceled { + workflow.State = model.StatusKilled + } } diff --git a/server/scheduler/impl_test.go b/server/scheduler/impl_test.go new file mode 100644 index 00000000000..3eaf1321b02 --- /dev/null +++ b/server/scheduler/impl_test.go @@ -0,0 +1,61 @@ +// Copyright 2026 Woodpecker Authors +// +// Licensed 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 scheduler + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + + "go.woodpecker-ci.org/woodpecker/v3/rpc" + "go.woodpecker-ci.org/woodpecker/v3/server/model" + store_mocks "go.woodpecker-ci.org/woodpecker/v3/server/store/mocks" +) + +func TestCompleteRunningChildren(t *testing.T) { + t.Run("a still-running service step is finished so the workflow succeeds", func(t *testing.T) { + successStep := &model.Step{ + ID: 1, + State: model.StatusSuccess, + Started: 1234567800, + } + runningService := &model.Step{ + ID: 2, + State: model.StatusRunning, + Started: 1234567800, + } + workflow := model.Workflow{ + ID: 7, + State: model.StatusRunning, + Children: []*model.Step{successStep, runningService}, + } + + mockStore := store_mocks.NewMockStore(t) + mockStore.On("StepUpdate", mock.Anything).Return(nil) + + p := &impl{store: mockStore} + p.completeRunningChildren(&workflow, 1234567900) + + assert.Equal(t, model.StatusSuccess, runningService.State) + assert.Equal(t, int64(1234567900), runningService.Finished) + + updateWorkflowStateToDone(&workflow, rpc.WorkflowState{ + Started: 1234567800, + Finished: 1234567900, + }) + assert.Equal(t, model.StatusSuccess, workflow.State) + }) +} diff --git a/server/scheduler/scheduler.go b/server/scheduler/scheduler.go index 7a2d80160ee..df712059f31 100644 --- a/server/scheduler/scheduler.go +++ b/server/scheduler/scheduler.go @@ -28,6 +28,12 @@ import ( // is skipped; the int is a match score (higher is better). type FilterFn func(*model.Task) (bool, int) +// SkippedWorkflowFunc is invoked after the scheduler has finalized a workflow +// as skipped. It lets the caller run the follow-up that does not belong to the +// scheduler, namely syncing the workflow's status to the forge. The scheduler +// has already persisted the skipped state and notified subscribers. +type SkippedWorkflowFunc func(repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow) + // Scheduler coordinates the queue and pubsub providers behind a single // surface. The low-level enqueue (queue.PushAtOnce) and publish // (pubsub.Publish) calls are intentionally not exposed: callers use the @@ -37,12 +43,12 @@ type Scheduler interface { // Queue operations. // // Poll blocks until the next runnable workflow for the given agent is - // available. It applies the agent's label filter, transparently skips - // tasks whose dependencies preclude running (invoking markSkipped so the - // caller can finalize them), and returns the runnable workflow. - // TODO: markSkipped is a callback helper that is only needed as we use the rpc.Done to mark skipped workflows as done - // this is a hack for another refactor later. - Poll(c context.Context, agentID int64, agentFilter rpc.Filter, markSkipped func(taskID string) error) (*rpc.Workflow, error) + // available. It applies the agent's label filter and transparently + // finalizes any task whose dependencies preclude it from running as + // skipped, invoking onSkipped afterwards so the caller can perform the + // non-scheduling follow-up (e.g. forge status) that the scheduler cannot + // do itself. It then returns the next runnable workflow. + Poll(c context.Context, agentID int64, agentFilter rpc.Filter, onSkipped SkippedWorkflowFunc) (*rpc.Workflow, error) Extend(c context.Context, agentID int64, workflowID string) error Done(c context.Context, id string, exitStatus model.StatusValue) error Error(c context.Context, id string, err error) error @@ -59,10 +65,22 @@ type Scheduler interface { PublishPipelineEvent(c context.Context, repo *model.Repo, pipeline *model.Pipeline) error StartPipeline(c context.Context, repo *model.Repo, pipeline *model.Pipeline, tasks []*model.Task) error - // CancelWorkflows cancels the given workflows: it evicts them from the - // queue and signals the cancellation to any agents waiting on them. This is - // the entry point for the scheduler to later own the full cancel cleanup. - CancelWorkflows(c context.Context, workflowIDs []string) error + // FinishWorkflow owns the completion of a single workflow: it finalizes + // the still-running children, persists the workflow's final state, + // acknowledges it on the queue, rolls the pipeline up to its done state + // once nothing is left running and publishes the change. It returns the + // updated pipeline (with its refreshed tree) and workflow so the caller + // can sync the forge status, close log streams and record metrics. + FinishWorkflow(c context.Context, repo *model.Repo, pipeline *model.Pipeline, workflow *model.Workflow, state rpc.WorkflowState) (*model.Pipeline, *model.Workflow, error) + + // CancelWorkflows owns the full cancellation of a pipeline's workflows: + // it evicts the running/pending workflows from the queue (signaling the + // cancellation to any agents waiting on them), marks the still-pending + // workflows and steps as skipped, transitions the pipeline to its killed + // state, and publishes the resulting state change to subscribers. It + // returns the updated (killed) pipeline so the caller can sync the forge + // status, which is the only cancellation concern left to the caller. + CancelWorkflows(c context.Context, repo *model.Repo, pipeline *model.Pipeline, workflows []*model.Workflow, cancelInfo *model.CancelInfo) (*model.Pipeline, error) } func NewScheduler(ctx context.Context, store store.Store, q queue.Queue, ps pubsub.PubSub) Scheduler { diff --git a/server/pipeline/status.go b/server/status/status.go similarity index 64% rename from server/pipeline/status.go rename to server/status/status.go index 8b06f9bd38e..da5e84225ed 100644 --- a/server/pipeline/status.go +++ b/server/status/status.go @@ -12,7 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pipeline +// Package status holds the pure status-aggregation logic shared between the +// pipeline layer, the scheduler and the API. It only depends on the model and +// must not import any package that orchestrates state, so the scheduler can use +// it without creating an import cycle. +package status import "go.woodpecker-ci.org/woodpecker/v3/server/model" @@ -43,7 +47,7 @@ var statusPriorityOrder = []model.StatusValue{ model.StatusSkipped, } -var priorityMap map[model.StatusValue]int = buildPriorityMap() +var priorityMap = buildPriorityMap() func buildPriorityMap() map[model.StatusValue]int { m := map[model.StatusValue]int{} @@ -53,6 +57,8 @@ func buildPriorityMap() map[model.StatusValue]int { return m } +// MergeStatusValues merges two status values, returning the more significant of +// the two according to statusPriorityOrder. func MergeStatusValues(s, t model.StatusValue) model.StatusValue { // both are skipped due to cancellation -> canceled if s == model.StatusCanceled && t == model.StatusCanceled { @@ -67,3 +73,27 @@ func MergeStatusValues(s, t model.StatusValue) model.StatusValue { } return statusPriorityOrder[min(priorityMap[s], priorityMap[t])] } + +// WorkflowStatus determines a workflow status based on its 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 +} + +// PipelineStatus determines a pipeline status based on its corresponding workflow list. +func PipelineStatus(workflows []*model.Workflow) model.StatusValue { + status := model.StatusSuccess + + for _, p := range workflows { + status = MergeStatusValues(status, p.State) + } + + return status +} diff --git a/server/status/status_test.go b/server/status/status_test.go new file mode 100644 index 00000000000..16fc69ac1f0 --- /dev/null +++ b/server/status/status_test.go @@ -0,0 +1,245 @@ +// Copyright 2026 Woodpecker Authors +// +// Licensed 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 status + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "go.woodpecker-ci.org/woodpecker/v3/server/model" +) + +func TestStatusValueMerge(t *testing.T) { + tests := []struct { + s model.StatusValue + t model.StatusValue + e model.StatusValue + }{ + { + s: model.StatusSuccess, + t: model.StatusSkipped, + e: model.StatusSuccess, + }, + { + s: model.StatusSuccess, + t: model.StatusSuccess, + e: model.StatusSuccess, + }, + { + s: model.StatusFailure, + t: model.StatusSuccess, + e: model.StatusFailure, + }, + { + s: model.StatusRunning, + t: model.StatusSuccess, + e: model.StatusRunning, + }, + { + s: model.StatusRunning, + t: model.StatusFailure, + e: model.StatusRunning, + }, + { + s: model.StatusFailure, + t: model.StatusKilled, + e: model.StatusKilled, + }, + { + s: model.StatusSkipped, + t: model.StatusKilled, + e: model.StatusKilled, + }, + { + s: model.StatusSkipped, + t: model.StatusSkipped, + e: model.StatusSkipped, + }, + { + s: model.StatusSkipped, + t: model.StatusCanceled, + e: model.StatusKilled, + }, + { + s: model.StatusSuccess, + t: model.StatusCanceled, + e: model.StatusKilled, + }, + { + s: model.StatusFailure, + t: model.StatusCanceled, + e: model.StatusKilled, + }, + } + for _, tt := range tests { + assert.Equal(t, tt.e, MergeStatusValues(tt.s, tt.t)) + assert.Equal(t, tt.e, MergeStatusValues(tt.t, tt.s)) + } +} + +func TestWorkflowStatus(t *testing.T) { + tests := []struct { + s []*model.Step + e model.StatusValue + }{ + { + s: []*model.Step{ + { + State: model.StatusFailure, + Failure: model.FailureIgnore, + }, + { + State: model.StatusSuccess, + Failure: model.FailureFail, + }, + }, + e: model.StatusSuccess, + }, + { + s: []*model.Step{ + { + State: model.StatusSuccess, + Failure: model.FailureFail, + }, + { + State: model.StatusSuccess, + Failure: model.FailureIgnore, + }, + }, + e: model.StatusSuccess, + }, + { + s: []*model.Step{ + { + State: model.StatusFailure, + Failure: model.FailureFail, + }, + { + State: model.StatusSuccess, + Failure: model.FailureFail, + }, + }, + e: model.StatusFailure, + }, + { + s: []*model.Step{ + { + State: model.StatusSuccess, + Failure: model.FailureFail, + }, + { + State: model.StatusPending, + Failure: model.FailureFail, + }, + }, + e: model.StatusPending, + }, + { + s: []*model.Step{ + { + State: model.StatusSuccess, + Failure: model.FailureFail, + }, + { + State: model.StatusPending, + Failure: model.FailureIgnore, + }, + }, + e: model.StatusPending, + }, + { + s: []*model.Step{ + { + State: model.StatusSuccess, + Failure: model.FailureIgnore, + }, + { + State: model.StatusPending, + Failure: model.FailureFail, + }, + }, + e: model.StatusPending, + }, + { + s: []*model.Step{ + { + State: model.StatusSuccess, + Failure: model.FailureIgnore, + }, + { + State: model.StatusPending, + Failure: model.FailureIgnore, + }, + }, + e: model.StatusPending, + }, + { + s: []*model.Step{ + { + State: model.StatusRunning, + Failure: model.FailureFail, + }, + { + State: model.StatusPending, + Failure: model.FailureFail, + }, + }, + e: model.StatusRunning, + }, + { + s: []*model.Step{ + { + State: model.StatusRunning, + Failure: model.FailureIgnore, + }, + { + State: model.StatusPending, + Failure: model.FailureIgnore, + }, + }, + e: model.StatusRunning, + }, + { + s: []*model.Step{ + { + State: model.StatusRunning, + Failure: model.FailureIgnore, + }, + { + State: model.StatusPending, + Failure: model.FailureFail, + }, + }, + e: model.StatusRunning, + }, + { + s: []*model.Step{ + { + State: model.StatusRunning, + Failure: model.FailureFail, + }, + { + State: model.StatusPending, + Failure: model.FailureIgnore, + }, + }, + e: model.StatusRunning, + }, + } + for _, tt := range tests { + assert.Equal(t, tt.e, WorkflowStatus(tt.s)) + } +}