From 50574018db292b9df22c4f9591f0e66458480909 Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 23 Jul 2026 16:33:07 +0000 Subject: [PATCH] spindle/engine: properly interrupt running workflows on cancel, dont overwrite cancelled status Signed-off-by: dawn --- spindle/engine/engine.go | 126 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------- spindle/engine/engine_test.go | 109 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/xrpc/pipeline_cancel_pipeline.go | 24 ++++++++++++++++-------- 3 file(s) changed, 206 insertion(s)(+), 53 deletion(s)(-) diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -16,10 +16,52 @@ "tangled.org/core/spindle/secrets" ) var ( - ErrTimedOut = errors.New("timed out") - ErrWorkflowFailed = errors.New("workflow failed") + ErrTimedOut = errors.New("timed out") + ErrWorkflowFailed = errors.New("workflow failed") + ErrWorkflowCanceled = errors.New("workflow canceled") ) +var ( + activeMu sync.Mutex + activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc) +) + +func CancelWorkflow(wid models.WorkflowId) { + activeMu.Lock() + cancel, ok := activeCancels[wid] + activeMu.Unlock() + if ok { + cancel(ErrWorkflowCanceled) + } +} + +// user cancel, timeout is DeadlineExceeded +func isCanceled(wfCtx context.Context) bool { + return errors.Is(context.Cause(wfCtx), ErrWorkflowCanceled) +} + +// for when recording early wf cancellations +func writeWfError(db *db.DB, n *notifier.Notifier, l *slog.Logger, wfCtx context.Context, wid models.WorkflowId, phase string, err error) { + l = l.With("wid", wid, "phase", phase) + switch { + case isCanceled(wfCtx): + l.Info("workflow canceled") + if dbErr := db.StatusCancelled(wid, "User canceled the workflow", -1, n); dbErr != nil { + l.Error("failed to set workflow status to cancelled", "err", dbErr) + } + case errors.Is(err, ErrTimedOut) || errors.Is(wfCtx.Err(), context.DeadlineExceeded): + l.Info("workflow timed out") + if dbErr := db.StatusTimeout(wid, n); dbErr != nil { + l.Error("failed to set workflow status to timeout", "err", dbErr) + } + default: + l.Error("workflow failed", "err", err) + if dbErr := db.StatusFailed(wid, err.Error(), -1, n); dbErr != nil { + l.Error("failed to set workflow status to failed", "err", dbErr) + } + } +} + type workflowFinalizer interface { FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error } @@ -82,7 +124,10 @@ continue } wg.Go(func() { - + if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() { + l.Info("skipping finished workflow", "wid", wid, "status", st.Status) + return + } defer func() { if s3 != nil { logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String())) @@ -101,17 +146,28 @@ l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid) defer wfLogger.Close() } + timeoutCtx, timeoutCancel := context.WithTimeout(ctx, workflowTimeout) + defer timeoutCancel() + + wfCtx, userCancel := context.WithCancelCause(timeoutCtx) + defer userCancel(nil) + + // allow wf context to be cancelled properly by manual cancel + activeMu.Lock() + activeCancels[wid] = userCancel + activeMu.Unlock() + defer func() { + activeMu.Lock() + delete(activeCancels, wid) + activeMu.Unlock() + }() + l.Info("waiting for slot", "wid", wid) slot := WorkflowSlot(NoopSlot{}) if s, ok := eng.(WorkflowSlotter); ok { - var err error - slot, err = s.AcquireWorkflowSlot(ctx, wid, &w) + slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w) if err != nil { - l.Error("failed to acquire slot", "wid", wid, "err", err) - dbErr := db.StatusFailed(wid, err.Error(), -1, n) - if dbErr != nil { - l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) - } + writeWfError(db, n, l, wfCtx, wid, "waiting for slot", err) return } } @@ -123,39 +179,27 @@ l.Error("failed to set workflow status to running", "wid", wid, "err", err) return } - err = eng.SetupWorkflow(ctx, wid, &w, wfLogger) + err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger) if err != nil { - // TODO(winter): Should this always set StatusFailed? - // In the original, we only do in a subset of cases. - l.Error("setting up workflow", "wid", wid, "err", err) - - destroyErr := eng.DestroyWorkflow(ctx, wid) - if destroyErr != nil { - l.Error("failed to destroy workflow after setup failure", "error", destroyErr) - } - - dbErr := db.StatusFailed(wid, err.Error(), -1, n) - if dbErr != nil { - l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) + if !isCanceled(wfCtx) { + if destroyErr := eng.DestroyWorkflow(ctx, wid); destroyErr != nil { + l.Error("failed to destroy workflow after setup failure", "error", destroyErr) + } } + writeWfError(db, n, l, wfCtx, wid, "setting up workflow", err) return } defer eng.DestroyWorkflow(ctx, wid) - ctx, cancel := context.WithTimeout(ctx, workflowTimeout) - defer cancel() - for stepIdx, step := range w.Steps { - // log start of step if wfLogger != nil { wfLogger. ControlWriter(stepIdx, step, models.StepStatusStart). Write([]byte{0}) } - err = eng.RunStep(ctx, wid, &w, stepIdx, allSecrets, wfLogger) + err = eng.RunStep(wfCtx, wid, &w, stepIdx, allSecrets, wfLogger) - // log end of step if wfLogger != nil { wfLogger. ControlWriter(stepIdx, step, models.StepStatusEnd). @@ -163,29 +207,21 @@ Write([]byte{0}) } if err != nil { - if errors.Is(err, ErrTimedOut) { - dbErr := db.StatusTimeout(wid, n) - if dbErr != nil { - l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr) - } - } else { - dbErr := db.StatusFailed(wid, err.Error(), -1, n) - if dbErr != nil { - l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) - } - } + writeWfError(db, n, l, wfCtx, wid, "running step", err) return } } if finalizer, ok := eng.(workflowFinalizer); ok { - if err := finalizer.FinalizeWorkflow(ctx, wid, &w, wfLogger); err != nil { - dbErr := db.StatusFailed(wid, err.Error(), -1, n) - if dbErr != nil { - l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) - } + if err := finalizer.FinalizeWorkflow(wfCtx, wid, &w, wfLogger); err != nil { + writeWfError(db, n, l, wfCtx, wid, "finalizing", err) return } + } + + if isCanceled(wfCtx) { + writeWfError(db, n, l, wfCtx, wid, "before success", nil) + return } err = db.StatusSuccess(wid, n) diff --git a/spindle/engine/engine_test.go b/spindle/engine/engine_test.go --- a/spindle/engine/engine_test.go +++ b/spindle/engine/engine_test.go @@ -155,3 +155,112 @@ if err != nil || statusUnique.Status != string(models.StatusKindSuccess) { t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err) } } + +func TestCancelWorkflow_NotOverwritten(t *testing.T) { + t.Parallel() + + testDB := newTestDB(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) + + stepStarted := make(chan struct{}) + eng := &mockEngine{ + runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int) error { + close(stepStarted) + <-ctx.Done() + return ctx.Err() + }, + } + + pipelineId := models.PipelineId{ + Knot: "test-knot", + Rkey: "test-rkey", + } + + wid := models.WorkflowId{ + PipelineId: pipelineId, + Name: "cancel_test_job", + } + + pipeline := &models.Pipeline{ + Workflows: map[models.Engine][]models.Workflow{ + eng: { + { + Name: "cancel_test_job", + Steps: []models.Step{mockStep{name: "step1"}}, + }, + }, + }, + } + + cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} + doneChan := make(chan struct{}) + go func() { + StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId) + close(doneChan) + }() + + select { + case <-stepStarted: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for step to start") + } + + _ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil) + CancelWorkflow(wid) + + select { + case <-doneChan: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for StartWorkflows to complete") + } + + // the runner writes StatusCancelled itself when it sees the canceled ctx + // the handler writes nothing for a live wf, so nothing lands after to overwrite it + st, err := testDB.GetStatus(wid) + if err != nil { + t.Fatalf("GetStatus error = %v", err) + } + if st.Status != string(models.StatusKindCancelled) { + t.Fatalf("expected status to be cancelled, got %s", st.Status) + } +} + +func TestSetupTimeout_ReportsTimeout(t *testing.T) { + t.Parallel() + + testDB := newTestDB(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) + + // setup blocks past the workflow timeout, so it should land as timeout not failed + eng := &mockEngine{ + timeout: 100 * time.Millisecond, + setupFunc: func(ctx context.Context, wid models.WorkflowId) error { + <-ctx.Done() + return ctx.Err() + }, + } + + pipelineId := models.PipelineId{Knot: "test-knot", Rkey: "test-rkey"} + wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"} + + pipeline := &models.Pipeline{ + Workflows: map[models.Engine][]models.Workflow{ + eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}}, + }, + } + + cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} + StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId) + + st, err := testDB.GetStatus(wid) + if err != nil { + t.Fatalf("GetStatus error = %v", err) + } + if st.Status != string(models.StatusKindTimeout) { + t.Fatalf("expected status to be timeout, got %s", st.Status) + } + + if len(eng.runStepCalls) != 0 { + t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls)) + } +} diff --git a/spindle/xrpc/pipeline_cancel_pipeline.go b/spindle/xrpc/pipeline_cancel_pipeline.go --- a/spindle/xrpc/pipeline_cancel_pipeline.go +++ b/spindle/xrpc/pipeline_cancel_pipeline.go @@ -7,6 +7,7 @@ "net/http" "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/tangled" + "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" xrpcerr "tangled.org/core/xrpc/errors" ) @@ -86,16 +87,23 @@ Name: wName, } l.Debug("cancel pipeline", "wid", wid) - for _, engine := range x.Engines { + // dont cancel a workflow that already finished + st, err := x.Db.GetStatus(wid) + if err == nil && models.StatusKind(st.Status).IsFinish() { + continue + } + + if err := x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier); err != nil { + fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status cancelled: %w", err))) + return + } + + engine.CancelWorkflow(wid) + + for _, eng := range x.Engines { l.Debug("destroying workflow", "wid", wid) - err := engine.DestroyWorkflow(r.Context(), wid) - if err != nil { + if err := eng.DestroyWorkflow(r.Context(), wid); err != nil { fail(xrpcerr.GenericError(fmt.Errorf("failed to destroy workflow: %w", err))) - return - } - err = x.Db.StatusCancelled(wid, "User canceled the workflow", -1, x.Notifier) - if err != nil { - fail(xrpcerr.GenericError(fmt.Errorf("failed to emit status failed: %w", err))) return } } -- tangled.sh