diff --git a/backend/domain/workflow/entity/vo/node.go b/backend/domain/workflow/entity/vo/node.go index af21716d48..1c56ee2f4a 100644 --- a/backend/domain/workflow/entity/vo/node.go +++ b/backend/domain/workflow/entity/vo/node.go @@ -96,6 +96,12 @@ type wfErr struct { spaceID int64 workflowID int64 cause error + // level holds the error level for this specific error instance. It is kept + // separate from the shared StatusError.Extra() map so that mutating the + // level (see ChangeErrLevel) never writes to a map that may be shared + // across goroutines, e.g. the package-level singletons CancelErr, + // NodeTimeoutErr and WorkflowTimeoutErr. + level ErrorLevel } func (w *wfErr) DebugURL() string { @@ -112,6 +118,10 @@ func (w *wfErr) DebugURL() string { } func (w *wfErr) Level() ErrorLevel { + if w.level != "" { + return w.level + } + if w.StatusError.Extra() == nil { return LevelError } @@ -147,9 +157,16 @@ func (w *wfErr) Unwrap() error { return w.cause } +// ChangeErrLevel returns a copy of the error with its level set to newLevel. +// It deliberately does not mutate the receiver: package-level singletons such +// as CancelErr, NodeTimeoutErr and WorkflowTimeoutErr are shared across +// goroutines, and mutating their underlying StatusError.Extra() map in place +// previously triggered "fatal error: concurrent map writes" when multiple +// nodes handled errors (e.g. timeouts) concurrently. func (w *wfErr) ChangeErrLevel(newLevel ErrorLevel) WorkflowError { - w.StatusError.Extra()["level"] = string(newLevel) - return w + newErr := *w + newErr.level = newLevel + return &newErr } func NewError(code int, opts ...errorx.Option) WorkflowError { @@ -159,6 +176,7 @@ func NewError(code int, opts ...errorx.Option) WorkflowError { _ = errors.As(e, &sErr) wfe := &wfErr{ StatusError: sErr, + level: LevelError, } return wfe @@ -172,6 +190,7 @@ func WrapError(code int, err error, opts ...errorx.Option) WorkflowError { wfe := &wfErr{ StatusError: sErr, cause: err, + level: LevelError, } return wfe } @@ -189,6 +208,7 @@ func NewWarn(code int, opts ...errorx.Option) WorkflowError { _ = errors.As(e, &sErr) wfe := &wfErr{ StatusError: sErr, + level: LevelWarn, } return wfe @@ -202,6 +222,7 @@ func WrapWarn(code int, err error, opts ...errorx.Option) WorkflowError { wfe := &wfErr{ StatusError: sErr, cause: err, + level: LevelWarn, } return wfe } @@ -222,6 +243,7 @@ func newCancel() WorkflowError { _ = errors.As(e, &sErr) wfe := &wfErr{ StatusError: sErr, + level: LevelCancel, } return wfe } @@ -234,6 +256,7 @@ func newNodeTimeout() WorkflowError { _ = errors.As(e, &sErr) wfe := &wfErr{ StatusError: sErr, + level: LevelError, } return wfe } @@ -246,6 +269,7 @@ func newWorkflowTimeout() WorkflowError { _ = errors.As(e, &sErr) wfe := &wfErr{ StatusError: sErr, + level: LevelError, } return wfe } diff --git a/backend/domain/workflow/entity/vo/node_concurrency_test.go b/backend/domain/workflow/entity/vo/node_concurrency_test.go new file mode 100644 index 0000000000..5a94521448 --- /dev/null +++ b/backend/domain/workflow/entity/vo/node_concurrency_test.go @@ -0,0 +1,72 @@ +/* + * Copyright 2025 coze-dev 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 vo + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestChangeErrLevelIsSafeForConcurrency verifies that ChangeErrLevel does +// not mutate shared state on package-level error singletons. The previous +// implementation wrote to StatusError.Extra()["level"] in place, which +// caused "fatal error: concurrent map writes" when multiple goroutines +// concurrently handled timeouts/cancellations. +func TestChangeErrLevelIsSafeForConcurrency(t *testing.T) { + const goroutines = 100 + + var wg sync.WaitGroup + wg.Add(goroutines) + + for i := 0; i < goroutines; i++ { + go func() { + defer wg.Done() + // Call ChangeErrLevel on the global singletons concurrently. + // The fix ensures this no longer writes to a shared map. + e1 := NodeTimeoutErr.ChangeErrLevel(LevelWarn) + e2 := CancelErr.ChangeErrLevel(LevelWarn) + e3 := WorkflowTimeoutErr.ChangeErrLevel(LevelWarn) + + assert.Equal(t, LevelWarn, e1.Level()) + assert.Equal(t, LevelWarn, e2.Level()) + assert.Equal(t, LevelWarn, e3.Level()) + }() + } + + wg.Wait() +} + +// TestChangeErrLevelDoesNotMutateSingleton verifies that calling +// ChangeErrLevel on a global singleton does not change the singleton's +// level — a copy is returned instead. +func TestChangeErrLevelDoesNotMutateSingleton(t *testing.T) { + assert.Equal(t, LevelCancel, CancelErr.Level()) + assert.Equal(t, LevelError, NodeTimeoutErr.Level()) + assert.Equal(t, LevelError, WorkflowTimeoutErr.Level()) + + // Change the level on copies. + _ = CancelErr.ChangeErrLevel(LevelWarn) + _ = NodeTimeoutErr.ChangeErrLevel(LevelWarn) + _ = WorkflowTimeoutErr.ChangeErrLevel(LevelWarn) + + // The singletons must remain unchanged. + assert.Equal(t, LevelCancel, CancelErr.Level()) + assert.Equal(t, LevelError, NodeTimeoutErr.Level()) + assert.Equal(t, LevelError, WorkflowTimeoutErr.Level()) +}