diff --git a/cli/exec/exec.go b/cli/exec/exec.go index 9da07e62b..052077603 100644 --- a/cli/exec/exec.go +++ b/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", diff --git a/pipeline/backend/dummy/dummy.go b/pipeline/backend/dummy/dummy.go index 601a842d8..27483ddb5 100644 --- a/pipeline/backend/dummy/dummy.go +++ b/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 } diff --git a/pipeline/runtime/helpers_test.go b/pipeline/runtime/helpers_test.go index 14f5631db..83b74e267 100644 --- a/pipeline/runtime/helpers_test.go +++ b/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" { diff --git a/pipeline/runtime/runtime.go b/pipeline/runtime/runtime.go index 8003b26a7..a52bee41e 100644 --- a/pipeline/runtime/runtime.go +++ b/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) diff --git a/pipeline/runtime/runtime_test.go b/pipeline/runtime/runtime_test.go index ffe4ba2e5..01d913bf6 100644 --- a/pipeline/runtime/runtime_test.go +++ b/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") + } +} diff --git a/pipeline/runtime/workflow.go b/pipeline/runtime/workflow.go index 0f3be10bd..9fa82a3c7 100644 --- a/pipeline/runtime/workflow.go +++ b/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() diff --git a/pipeline/runtime/workflow_test.go b/pipeline/runtime/workflow_test.go index 28c2b531e..6a0d0107b 100644 --- a/pipeline/runtime/workflow_test.go +++ b/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()) } diff --git a/pipeline/tracing/tracer.go b/pipeline/tracing/tracer.go index 1f1233af7..1392e0169 100644 --- a/pipeline/tracing/tracer.go +++ b/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 })