Fix workflow hang on services (#6507)

because we now wait for all steps to trace status back before we return, the defere did not tear down services anymore ...

... we now explicit tear down services and steps after all stages have executed.

Also adds tests to check for that and update the dummy backend to fullfill the interface contract of killing all "running" steps with DestroyWorkflow.

cli/exec/exec.go

@@ -48,7 +48,6 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v3/pipeline/frontend/yaml/matrix"
 	"go.woodpecker-ci.org/woodpecker/v3/pipeline/logging"
 	pipeline_runtime "go.woodpecker-ci.org/woodpecker/v3/pipeline/runtime"
-	"go.woodpecker-ci.org/woodpecker/v3/pipeline/tracing"
 	pipeline_utils "go.woodpecker-ci.org/woodpecker/v3/pipeline/utils"
 	"go.woodpecker-ci.org/woodpecker/v3/shared/constant"
 	"go.woodpecker-ci.org/woodpecker/v3/shared/utils"
@@ -324,7 +323,6 @@ func execWithAxis(ctx context.Context, c *cli.Command, file, repoPath string, ax
 
 	return pipeline_runtime.New(compiled, backendEngine,
 		pipeline_runtime.WithContext(pipelineCtx), //nolint:contextcheck
-		pipeline_runtime.WithTracer(tracing.NoOpTracer),
 		pipeline_runtime.WithLogger(defaultLogger),
 		pipeline_runtime.WithDescription(map[string]string{
 			"CLI": "exec",

pipeline/backend/dummy/dummy.go

@@ -61,6 +61,11 @@ func stepKey(taskUUID, stepUUID string) string {
 	return "task_" + taskUUID + "_step_" + stepUUID
 }
 
+// workflowKey returns the kv-store key for a workflow's state.
+func workflowKey(taskUUID string) string {
+	return "task_" + taskUUID
+}
+
 // New returns a dummy backend.
 func New() backend_types.Backend {
 	return &dummy{
@@ -92,7 +97,7 @@ func (e *dummy) SetupWorkflow(_ context.Context, _ *backend_types.Config, taskUU
 		return fmt.Errorf("expected fail to setup workflow")
 	}
 	log.Trace().Str("taskUUID", taskUUID).Msg("create workflow environment")
-	e.kv.Store("task_"+taskUUID, "setup")
+	e.kv.Store(workflowKey(taskUUID), make(chan struct{}))
 	return nil
 }
 
@@ -100,7 +105,7 @@ func (e *dummy) StartStep(_ context.Context, step *backend_types.Step, taskUUID
 	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s", step.Name)
 
 	// internal state checks
-	_, exist := e.kv.Load("task_" + taskUUID)
+	_, exist := e.kv.Load(workflowKey(taskUUID))
 	if !exist {
 		return fmt.Errorf("expect env of workflow %s to exist but found none to destroy", taskUUID)
 	}
@@ -131,7 +136,7 @@ func canceledState() *backend_types.State {
 
 // sleepWithContext blocks for the given duration or until ctx is canceled.
 // Returns true if canceled, false if the sleep completed normally.
-func sleepWithContext(ctx context.Context, d time.Duration) (canceled bool) {
+func sleepWithContext(ctx context.Context, stop <-chan struct{}, d time.Duration) (canceled bool) {
 	if ctx.Err() != nil {
 		return true
 	}
@@ -142,17 +147,23 @@ func sleepWithContext(ctx context.Context, d time.Duration) (canceled bool) {
 		return false
 	case <-ctx.Done():
 		return true
+	case <-stop:
+		return false
 	}
 }
 
 func (e *dummy) WaitStep(ctx context.Context, step *backend_types.Step, taskUUID string) (*backend_types.State, error) {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("wait for step %s", step.Name)
 
-	_, exist := e.kv.Load("task_" + taskUUID)
+	rawWC, exist := e.kv.Load(workflowKey(taskUUID))
 	if !exist {
 		err := fmt.Errorf("expect env of workflow %s to exist but found none to destroy", taskUUID)
 		return &backend_types.State{Error: err}, err
 	}
+	wc, ok := rawWC.(chan struct{})
+	if !ok {
+		return nil, fmt.Errorf("workflow stop chan not found")
+	}
 
 	key := stepKey(taskUUID, step.UUID)
 
@@ -173,12 +184,12 @@ func (e *dummy) WaitStep(ctx context.Context, step *backend_types.Step, taskUUID
 			err = fmt.Errorf("WaitStep fail to parse sleep duration: %w", err)
 			return &backend_types.State{Error: err}, err
 		}
-		if sleepWithContext(ctx, toSleep) {
+		if sleepWithContext(ctx, wc, toSleep) {
 			e.kv.Store(key, stepStateDone)
 			return canceledState(), nil
 		}
 	} else if step.Type == backend_types.StepTypeService {
-		if sleepWithContext(ctx, testServiceTimeout) {
+		if sleepWithContext(ctx, wc, testServiceTimeout) {
 			// context for service closed — we can move forward
 		} else {
 			err := fmt.Errorf("WaitStep fail due to timeout of service after 1 second")
@@ -207,7 +218,7 @@ func (e *dummy) WaitStep(ctx context.Context, step *backend_types.Step, taskUUID
 func (e *dummy) TailStep(_ context.Context, step *backend_types.Step, taskUUID string) (io.ReadCloser, error) {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("tail logs of step %s", step.Name)
 
-	_, exist := e.kv.Load("task_" + taskUUID)
+	_, exist := e.kv.Load(workflowKey(taskUUID))
 	if !exist {
 		return nil, fmt.Errorf("expect env of workflow %s to exist but found none to destroy", taskUUID)
 	}
@@ -233,7 +244,7 @@ func (e *dummy) TailStep(_ context.Context, step *backend_types.Step, taskUUID s
 func (e *dummy) DestroyStep(_ context.Context, step *backend_types.Step, taskUUID string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("stop step %s", step.Name)
 
-	_, exist := e.kv.Load("task_" + taskUUID)
+	_, exist := e.kv.Load(workflowKey(taskUUID))
 	if !exist {
 		return nil
 	}
@@ -258,11 +269,17 @@ func (e *dummy) DestroyStep(_ context.Context, step *backend_types.Step, taskUUI
 func (e *dummy) DestroyWorkflow(_ context.Context, _ *backend_types.Config, taskUUID string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("delete workflow environment")
 
-	_, exist := e.kv.Load("task_" + taskUUID)
+	rawWC, exist := e.kv.Load(workflowKey(taskUUID))
 	if !exist {
 		return fmt.Errorf("expect env of workflow %s to exist but found none to destroy", taskUUID)
 	}
-	e.kv.Delete("task_" + taskUUID)
+	wc, ok := rawWC.(chan struct{})
+	if !ok {
+		return fmt.Errorf("workflow stop chan not found")
+	}
+	close(wc)
+
+	e.kv.Delete(workflowKey(taskUUID))
 	return nil
 }
 

pipeline/runtime/helpers_test.go

@@ -19,6 +19,7 @@ package runtime
 import (
 	"io"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/mock"
 
@@ -48,6 +49,9 @@ func newTestLogger(t *testing.T) logging.Logger {
 // on a mockery-generated MockTracer. Thread-safe because mock.Mock.Calls
 // is append-only and we only read after the workflow completes.
 func getTracerStates(tracer *tracer_mocks.MockTracer) []state.State {
+	// for systems under load we wait for tracer to make it's calls
+	time.Sleep(120 * time.Microsecond)
+
 	var states []state.State
 	for _, call := range tracer.Calls {
 		if call.Method == "Trace" {

pipeline/runtime/runtime.go

@@ -62,6 +62,7 @@ func New(spec *backend_types.Config, backend backend_types.Backend, opts ...Opti
 	r.engine = backend
 	r.ctx = context.Background()
 	r.taskUUID = ulid.Make().String()
+	r.tracer = tracing.NoOpTracer
 	r.tracerLock = sync.Mutex{}
 	for _, opt := range opts {
 		opt(r)

pipeline/runtime/runtime_test.go

@@ -76,6 +76,15 @@ func withDetached() func(*backend_types.Step) {
 	}
 }
 
+// withUnboundedDetached models a detached step that runs until the workflow tears it down.
+func withUnboundedDetached() func(*backend_types.Step) {
+	return func(s *backend_types.Step) {
+		s.Type = backend_types.StepTypeService
+		s.Detached = true
+		s.Environment[dummy.EnvKeyStepSleep] = "3m"
+	}
+}
+
 func withService() func(*backend_types.Step) {
 	return func(s *backend_types.Step) {
 		s.Type = backend_types.StepTypeService
@@ -84,6 +93,15 @@ func withService() func(*backend_types.Step) {
 	}
 }
 
+// withUnboundedService models a real-world service that runs until the workflow tears it down.
+func withUnboundedService() func(*backend_types.Step) {
+	return func(s *backend_types.Step) {
+		s.Type = backend_types.StepTypeService
+		s.Detached = true
+		s.Environment[dummy.EnvKeyStepSleep] = "3m"
+	}
+}
+
 func withPlugin() func(*backend_types.Step) {
 	return func(s *backend_types.Step) {
 		s.Type = backend_types.StepTypePlugin
@@ -104,6 +122,12 @@ func withStartFail() func(*backend_types.Step) {
 	}
 }
 
+func withSleep(d string) func(*backend_types.Step) {
+	return func(s *backend_types.Step) {
+		s.Environment[dummy.EnvKeyStepSleep] = d
+	}
+}
+
 //
 // Trace assertion helpers.
 //
@@ -155,9 +179,8 @@ func TestWorkflowCloneBuildDeploy(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.NoError(t, r.Run(t.Context()))
 
-	assert.NoError(t, err)
 	traces := getTracerStates(tracer)
 	assert.Len(t, traces, 6)
 	for i := 0; i < 6; i += 2 {
@@ -185,7 +208,7 @@ func TestWorkflowWithServiceStep(t *testing.T) {
 					cmdStep("db", withService()),
 					cmdStep("build"),
 				}},
-				{Steps: []*backend_types.Step{cmdStep("test")}},
+				{Steps: []*backend_types.Step{cmdStep("test", withSleep("250ms"))}},
 			},
 		},
 		dummy.New(),
@@ -227,17 +250,18 @@ func TestWorkflowWithServiceStep(t *testing.T) {
 		assert.Less(t, startedIdx, exitedIdx, "%s started must precede %s exited", name, name)
 	}
 
-	// The contract of a service/detached step in stage 1: its exit trace arrives
-	// AFTER stage 2's steps have already been traced. That's the whole point of
-	// detaching — it must not block the next stage.
+	// The contract of a service/detached step: it does not block the next
+	// stage. Verify that stage 2's `test` step started before db (in stage 1)
+	// reported its exit — i.e. test was running in parallel with db, not
+	// queued behind it.
 	dbExitIdx := indexOfTrace(traces, func(s state.State) bool {
-		return s.CurrStep != nil && s.CurrStep.Name == "db" && s.CurrStepState.Exited
+		return s == *findLastTraceByName(traces, "db")
 	})
-	testExitIdx := indexOfTrace(traces, func(s state.State) bool {
-		return s.CurrStep != nil && s.CurrStep.Name == "test" && s.CurrStepState.Exited
+	testStartedIdx := indexOfTrace(traces, func(s state.State) bool {
+		return s == *findFirstTraceByName(traces, "test")
 	})
-	assert.Greater(t, dbExitIdx, testExitIdx,
-		"db (service) must complete after test (next stage) — otherwise it wasn't really detached")
+	assert.Less(t, testStartedIdx, dbExitIdx,
+		"test (next stage) must start before db (service) exits — otherwise db blocked stage 2")
 
 	// Runtime-injected env vars should be present on the test step's exit trace.
 	testExit := findLastTraceByName(traces, "test")
@@ -249,6 +273,7 @@ func TestWorkflowWithServiceStep(t *testing.T) {
 	// Strip runtime-injected env for a structural comparison of the step itself.
 	delete(testExit.CurrStep.Environment, "CI_PIPELINE_STARTED")
 	delete(testExit.CurrStep.Environment, "CI_STEP_STARTED")
+	delete(testExit.CurrStep.Environment, dummy.EnvKeyStepSleep)
 	assert.EqualValues(t, state.State{
 		Workflow: state.Workflow{Started: testExit.Workflow.Started},
 		CurrStep: &backend_types.Step{
@@ -367,8 +392,8 @@ func TestWorkflowOnFailureStepSkippedOnSuccess(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
-	require.NoError(t, err)
+	require.NoError(t, r.Run(t.Context()))
+
 	traces := getTracerStates(tracer)
 
 	firstCleanupTrace := findFirstTraceByName(traces, "cleanup-on-fail")
@@ -394,9 +419,8 @@ func TestWorkflowFailureIgnore(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.NoError(t, r.Run(t.Context()), "pipeline should succeed when failing step has failure=ignore")
 
-	assert.NoError(t, err, "pipeline should succeed when failing step has failure=ignore")
 	assert.NotNil(t, findStartedTrace(getTracerStates(tracer), "build"), "build step should run after ignored failure")
 
 	last := findLastTraceByName(getTracerStates(tracer), "build")
@@ -422,9 +446,8 @@ func TestWorkflowFailureIgnoreDoesNotSetWorkflowError(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.NoError(t, r.Run(t.Context()))
 
-	assert.NoError(t, err)
 	traces := getTracerStates(tracer)
 	firstDeployTrace := findFirstTraceByName(traces, "deploy")
 	lastDeployTrace := findLastTraceByName(traces, "deploy")
@@ -509,9 +532,8 @@ func TestWorkflowParallelStepsInStage(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.NoError(t, r.Run(t.Context()))
 
-	assert.NoError(t, err)
 	assert.Len(t, getTracerStates(tracer), 10)
 }
 
@@ -532,9 +554,8 @@ func TestWorkflowParallelStepOneFailsOthersComplete(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.Error(t, r.Run(t.Context()))
 
-	assert.Error(t, err)
 	assert.Len(t, getTracerStates(tracer), 4, "both parallel steps should complete and be traced")
 
 	lastFast := findLastTraceByName(getTracerStates(tracer), "test-fast")
@@ -563,9 +584,8 @@ func TestWorkflowStepStartFailure(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.Error(t, r.Run(t.Context()))
 
-	assert.Error(t, err)
 	deployTrace := findFirstTraceByName(getTracerStates(tracer), "build")
 	require.NotNil(t, deployTrace)
 	assert.EqualValues(t, backend_types.State{}, deployTrace.CurrStepState)
@@ -648,9 +668,8 @@ func TestWorkflowServiceWithParallelBuildAndOnFailure(t *testing.T) {
 		WithLogger(newTestLogger(t)),
 	)
 
-	err := r.Run(t.Context())
+	assert.Error(t, r.Run(t.Context()))
 
-	assert.Error(t, err)
 	traces := getTracerStates(tracer)
 
 	assert.NotNil(t, findStartedTrace(traces, "notify"), "notify (OnFailure) should have started")
@@ -1289,7 +1308,7 @@ func TestWorkflowFailingServiceDoesNotFailWorkflow(t *testing.T) {
 					cmdStep("db", withService(), withExitCode(1)),
 					cmdStep("build"),
 				}},
-				{Steps: []*backend_types.Step{cmdStep("deploy")}},
+				{Steps: []*backend_types.Step{cmdStep("deploy", withSleep("120ms"))}},
 			},
 		},
 		dummy.New(),
@@ -1341,7 +1360,7 @@ func TestWorkflowFailingDetachedStepDoesNotFailWorkflow(t *testing.T) {
 					cmdStep("background-worker", withDetached(), withExitCode(2)),
 					cmdStep("main-build"),
 				}},
-				{Steps: []*backend_types.Step{cmdStep("deploy")}},
+				{Steps: []*backend_types.Step{cmdStep("deploy", withSleep("120ms"))}},
 			},
 		},
 		dummy.New(),
@@ -1367,3 +1386,76 @@ func TestWorkflowFailingDetachedStepDoesNotFailWorkflow(t *testing.T) {
 	assert.True(t, deployExit.CurrStepState.Exited)
 	assert.Equal(t, 0, deployExit.CurrStepState.ExitCode)
 }
+
+// TestWorkflowUnboundedServiceDoesNotHang asserts that when all normal steps
+// have finished, a long-running service does NOT keep the workflow blocked
+// forever. The runtime must tear the service down on its own (the whole point
+// of declaring a step as a service is that it runs alongside the build, not
+// that the build waits for it).
+//
+// Regression for https://github.com/woodpecker-ci/woodpecker/commit/4dd3be7f96
+// which moved the upload waitgroup from per-upload (logger/tracer) to
+// per-detached-goroutine. The detached goroutine wraps WaitStep, which on
+// services blocks until the workflow context is canceled — so the workflow
+// hangs waiting for its own service to exit.
+func TestWorkflowUnboundedServiceDoesNotHang(t *testing.T) {
+	t.Parallel()
+	r := New(
+		&backend_types.Config{
+			Stages: []*backend_types.Stage{
+				{Steps: []*backend_types.Step{
+					cmdStep("db", withUnboundedService()),
+					cmdStep("build"),
+				}},
+				{Steps: []*backend_types.Step{cmdStep("test")}},
+			},
+		},
+		dummy.New(),
+		WithTracer(newTestTracer(t)),
+		WithLogger(newTestLogger(t)),
+	)
+
+	// Use a deadline well below the dummy backend's testServiceTimeout (1s) so
+	// that if this test "passes" it's because the runtime tore the service down,
+	// not because dummy's safety timeout fired.
+	done := make(chan error, 1)
+	go func() { done <- r.Run(t.Context()) }()
+
+	select {
+	case err := <-done:
+		assert.NoError(t, err)
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("workflow hung: runtime did not tear down the unbounded service after normal steps finished")
+	}
+}
+
+// TestWorkflowUnboundedDetachedDoesNotHang is the same as the service test but
+// for plain detached steps (Detached=true, Type=commands). The bug is the same
+// — a long-running detached step also pins the upload waitgroup.
+func TestWorkflowUnboundedDetachedDoesNotHang(t *testing.T) {
+	t.Parallel()
+	r := New(
+		&backend_types.Config{
+			Stages: []*backend_types.Stage{
+				{Steps: []*backend_types.Step{
+					cmdStep("background-worker", withUnboundedDetached()),
+					cmdStep("build"),
+				}},
+				{Steps: []*backend_types.Step{cmdStep("test")}},
+			},
+		},
+		dummy.New(),
+		WithTracer(newTestTracer(t)),
+		WithLogger(newTestLogger(t)),
+	)
+
+	done := make(chan error, 1)
+	go func() { done <- r.Run(t.Context()) }()
+
+	select {
+	case err := <-done:
+		assert.NoError(t, err)
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("workflow hung: runtime did not tear down the unbounded detached step after normal steps finished")
+	}
+}

pipeline/runtime/workflow.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"golang.org/x/sync/errgroup"
@@ -39,8 +40,7 @@ func (r *Runtime) Run(runnerCtx context.Context) error {
 	logger := r.makeLogger()
 	r.logStages()
 
-	// we make sure cleanup always happens
-	defer func() {
+	destroyWorkflowFunc := sync.OnceFunc(func() {
 		ctx := runnerCtx //nolint:contextcheck
 		if ctx.Err() != nil {
 			// runnerCtx itself is done — fall back to a short-lived shutdown context.
@@ -49,7 +49,10 @@ func (r *Runtime) Run(runnerCtx context.Context) error {
 		if err := r.engine.DestroyWorkflow(ctx, r.spec, r.taskUUID); err != nil {
 			logger.Error().Err(err).Msg("could not destroy workflow")
 		}
-	}()
+	})
+
+	// we make sure cleanup always happens
+	defer destroyWorkflowFunc()
 
 	r.started = time.Now().Unix()
 
@@ -71,6 +74,9 @@ func (r *Runtime) Run(runnerCtx context.Context) error {
 		}
 	}
 
+	// Now we can shutdown the workflow
+	destroyWorkflowFunc()
+
 	// Ensure all logs/traces are uploaded before finishing
 	logger.Debug().Msg("waiting for logs and traces upload")
 	r.uploadWait.Wait()

pipeline/runtime/workflow_test.go

@@ -35,7 +35,7 @@ import (
 
 func TestRunNilTracer(t *testing.T) {
 	t.Parallel()
-	r := New(&backend_types.Config{}, dummy.New(), WithLogger(newTestLogger(t)))
+	r := New(&backend_types.Config{}, dummy.New(), WithLogger(newTestLogger(t)), WithTracer(nil))
 
 	err := r.Run(t.Context())
 
@@ -308,7 +308,7 @@ func TestNewDefaults(t *testing.T) {
 	assert.Equal(t, spec, r.spec)
 	assert.NotEmpty(t, r.taskUUID)
 	assert.NotNil(t, r.ctx)
-	assert.Nil(t, r.tracer)
+	assert.NotNil(t, r.tracer)
 	assert.NotNil(t, r.engine)
 	assert.NoError(t, r.err.Get())
 }

pipeline/tracing/tracer.go

@@ -33,6 +33,4 @@ func (f TraceFunc) Trace(state *state.State) error {
 }
 
 // NoOpTracer provides a tracer that does nothing.
-var NoOpTracer = TraceFunc(func(state *state.State) error {
-	return nil
-})
+var NoOpTracer = TraceFunc(func(*state.State) error { return nil })