fix(workflow): prevent concurrent map writes on shared error singletons

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
 }

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())
+}