From 943908076335ee3277b15f169d71d5d303e6ef21 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Thu, 11 Jun 2026 10:02:11 +0200 Subject: [PATCH] fix(workflow): target the -main run in Wait and AbortRun, fix error handling Initiate starts the real Run workflow as a detached child with id "-main" (ParentClosePolicy ABANDON) and returns as soon as that child has started. The Initiate workflow (id == instanceID) is thus already completed by the time the API calls Wait/AbortRun: - AbortRun cancelled the completed Initiate execution, so cancellation never reached the running stages (wait_event/delay were unabortable). - Wait returned immediately on the completed Initiate execution, so ?wait=true returned a non-terminated instance. Both now target "-main" (matching ReadInstanceHistory). Wait also mishandled errors: errors.Is(err, &serviceerror.NotFound{}) can never match (no Is/Unwrap on that type) and errors.Unwrap(err) returned nil for non-wrapped errors, turning a failure into a success. Use errors.As for NotFound and return the original error otherwise. Adds TestWait covering both the terminate-wait and not-found paths. --- internal/workflow/manager.go | 16 ++++++--- internal/workflow/manager_test.go | 55 +++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 4 deletions(-) diff --git a/internal/workflow/manager.go b/internal/workflow/manager.go index 8460c61..6a24757 100644 --- a/internal/workflow/manager.go +++ b/internal/workflow/manager.go @@ -117,13 +117,18 @@ func (m *WorkflowManager) RunWorkflow(ctx context.Context, id string, variables } func (m *WorkflowManager) Wait(ctx context.Context, instanceID string) error { + // The actual work runs in the detached child workflow "-main"; + // the Initiate workflow (id == instanceID) completes as soon as that child + // has started. Waiting on instanceID would therefore return immediately, + // before the run is terminated, so we wait on the running child. if err := m.temporalClient. - GetWorkflow(ctx, instanceID, ""). + GetWorkflow(ctx, instanceID+"-main", ""). Get(ctx, nil); err != nil { - if errors.Is(err, &serviceerror.NotFound{}) { + var notFound *serviceerror.NotFound + if errors.As(err, ¬Found) { return ErrInstanceNotFound } - return errors.Unwrap(err) + return err } return nil } @@ -176,7 +181,10 @@ func (m *WorkflowManager) AbortRun(ctx context.Context, instanceID string) error return errors.Wrap(err, "retrieving workflow execution") } - return m.temporalClient.CancelWorkflow(ctx, instanceID, "") + // Cancel the detached child workflow that carries the actual run; the + // Initiate workflow (id == instanceID) has already completed, so cancelling + // it would be a no-op and never reach the running stages. + return m.temporalClient.CancelWorkflow(ctx, instanceID+"-main", "") } func (m *WorkflowManager) ListInstances(ctx context.Context, pagination ListInstancesQuery) (*bunpaginate.Cursor[Instance], error) { diff --git a/internal/workflow/manager_test.go b/internal/workflow/manager_test.go index 1ce2646..d208b79 100644 --- a/internal/workflow/manager_test.go +++ b/internal/workflow/manager_test.go @@ -76,3 +76,58 @@ func TestConfig(t *testing.T) { return len(updatedInstance.Statuses) == 1 }, 2*time.Second, 100*time.Millisecond) } + +func TestWait(t *testing.T) { + t.Parallel() + + database := srv.NewDatabase(t) + db, err := bunconnect.OpenSQLDB(logging.TestingContext(), bunconnect.ConnectionOptions{ + DatabaseSourceName: database.ConnString(), + }) + require.NoError(t, err) + t.Cleanup(func() { + _ = db.Close() + }) + require.NoError(t, storage.Migrate(logging.TestingContext(), db)) + + taskQueue := uuid.NewString() + w := temporalworker.New(logging.Testing(), devServer.Client(), taskQueue, + []temporalworker.DefinitionSet{ + NewWorkflows("test", false).DefinitionSet(), + temporalworker.NewDefinitionSet().Append(temporalworker.Definition{ + Name: "NoOp", + Func: (&stages.NoOp{}).GetWorkflow(), + }), + }, + []temporalworker.DefinitionSet{ + NewActivities(publish.NoOpPublisher, db).DefinitionSet(), + }, + worker.Options{}, + ) + require.NoError(t, w.Start()) + t.Cleanup(w.Stop) + + manager := NewManager(db, devServer.Client(), "test", taskQueue, false) + + t.Run("waits for the -main run to terminate", func(t *testing.T) { + config := Config{Stages: []RawStage{{"noop": map[string]any{}}}} + wf, err := manager.Create(logging.TestingContext(), config) + require.NoError(t, err) + i, err := manager.RunWorkflow(logging.TestingContext(), wf.ID, map[string]string{}) + require.NoError(t, err) + + // Wait must block on the detached "-main" child, not the Initiate + // workflow (which returns immediately). Once it returns, the instance + // must already be terminated. + require.NoError(t, manager.Wait(logging.TestingContext(), i.ID)) + + updated, err := manager.GetInstance(logging.TestingContext(), i.ID) + require.NoError(t, err) + require.True(t, updated.Terminated) + }) + + t.Run("unknown instance returns ErrInstanceNotFound", func(t *testing.T) { + err := manager.Wait(logging.TestingContext(), "does-not-exist") + require.ErrorIs(t, err, ErrInstanceNotFound) + }) +}