fix(worker): prevent concurrent map access race condition (#1143)

pinglin · web-flow · commit e24a0b676ee7 · 2025-11-18T07:19:10.000Z
### Because

- The pipeline worker was crashing with `fatal error: concurrent map
iteration and map write` when multiple goroutines accessed workflow
memory concurrently
- `WorkflowMemory.Get()` returned references to the underlying map data,
which were then accessed outside the mutex lock by `recipe.Eval()` →
`ToJSONValue()`
- This race condition caused OOM crashes during condition evaluation in
`ProcessBatchConditionsActivity`

### This commit

- Adds `Copy()` method to `data.Map` for creating thread-safe deep
copies of map structures
- Adds `Copy()` method to `data.Array` for creating thread-safe deep
copies of array structures
- Adds `copyValue()` helper function to recursively deep copy nested
data structures (maps, arrays)
- Updates `WorkflowMemory.Get()` to return deep copies of `Map` and
`Array` types, preventing shared mutable state across goroutines
- Adds comprehensive unit tests (`copy_test.go`) covering simple/nested
structures, mixed types, nil/empty cases, and concurrent modification
scenarios
- Ensures primitive types are returned as-is to minimize performance
overhead
diff --git a/pkg/data/array.go b/pkg/data/array.go
@@ -104,3 +104,16 @@ func (a Array) ToJSONValue() (v any, err error) {
 	}
 	return jsonArr, nil
 }
+
+// Copy creates a deep copy of the Array to prevent concurrent access issues.
+// This is essential when passing array data across goroutine boundaries.
+func (a Array) Copy() Array {
+	if a == nil {
+		return nil
+	}
+	copied := make(Array, len(a))
+	for i, v := range a {
+		copied[i] = copyValue(v)
+	}
+	return copied
+}
diff --git a/pkg/data/copy_test.go b/pkg/data/copy_test.go
@@ -0,0 +1,288 @@
+package data
+
+import (
+	"testing"
+
+	"github.com/instill-ai/pipeline-backend/pkg/data/format"
+)
+
+func TestMapCopy(t *testing.T) {
+	t.Run("Copy simple map", func(t *testing.T) {
+		original := Map{
+			"key1": NewString("value1"),
+			"key2": NewNumberFromInteger(42),
+			"key3": NewBoolean(true),
+		}
+
+		copied := original.Copy()
+
+		// Verify copied map has same values
+		if !copied.Equal(original) {
+			t.Error("Copied map should equal original")
+		}
+
+		// Verify it's a deep copy by modifying original
+		original["key1"] = NewString("modified")
+		if copied["key1"].(*stringData).String() != "value1" {
+			t.Error("Modifying original should not affect copy")
+		}
+	})
+
+	t.Run("Copy nested map", func(t *testing.T) {
+		original := Map{
+			"outer": Map{
+				"inner": Map{
+					"value": NewString("nested"),
+				},
+			},
+		}
+
+		copied := original.Copy()
+
+		// Verify it's a deep copy
+		originalInner := original["outer"].(Map)["inner"].(Map)
+		originalInner["value"] = NewString("modified")
+
+		copiedInner := copied["outer"].(Map)["inner"].(Map)
+		if copiedInner["value"].(*stringData).String() != "nested" {
+			t.Error("Modifying nested original should not affect copy")
+		}
+	})
+
+	t.Run("Copy map with array", func(t *testing.T) {
+		original := Map{
+			"array": Array{
+				NewString("item1"),
+				NewString("item2"),
+			},
+		}
+
+		copied := original.Copy()
+
+		// Modify original array
+		originalArray := original["array"].(Array)
+		originalArray[0] = NewString("modified")
+
+		// Verify copy is unaffected
+		copiedArray := copied["array"].(Array)
+		if copiedArray[0].(*stringData).String() != "item1" {
+			t.Error("Modifying array in original should not affect copy")
+		}
+	})
+
+	t.Run("Copy nil map", func(t *testing.T) {
+		var original Map
+		copied := original.Copy()
+
+		if copied != nil {
+			t.Error("Copy of nil map should be nil")
+		}
+	})
+
+	t.Run("Copy empty map", func(t *testing.T) {
+		original := Map{}
+		copied := original.Copy()
+
+		if copied == nil {
+			t.Error("Copy of empty map should not be nil")
+		}
+		if len(copied) != 0 {
+			t.Error("Copy of empty map should be empty")
+		}
+	})
+}
+
+func TestArrayCopy(t *testing.T) {
+	t.Run("Copy simple array", func(t *testing.T) {
+		original := Array{
+			NewString("value1"),
+			NewNumberFromInteger(42),
+			NewBoolean(true),
+		}
+
+		copied := original.Copy()
+
+		// Verify copied array has same values
+		if !copied.Equal(original) {
+			t.Error("Copied array should equal original")
+		}
+
+		// Verify it's a deep copy by modifying original
+		original[0] = NewString("modified")
+		if copied[0].(*stringData).String() != "value1" {
+			t.Error("Modifying original should not affect copy")
+		}
+	})
+
+	t.Run("Copy nested array", func(t *testing.T) {
+		original := Array{
+			Array{
+				Array{
+					NewString("nested"),
+				},
+			},
+		}
+
+		copied := original.Copy()
+
+		// Verify it's a deep copy
+		originalInner := original[0].(Array)[0].(Array)
+		originalInner[0] = NewString("modified")
+
+		copiedInner := copied[0].(Array)[0].(Array)
+		if copiedInner[0].(*stringData).String() != "nested" {
+			t.Error("Modifying nested original should not affect copy")
+		}
+	})
+
+	t.Run("Copy array with map", func(t *testing.T) {
+		original := Array{
+			Map{
+				"key": NewString("value"),
+			},
+		}
+
+		copied := original.Copy()
+
+		// Modify original map
+		originalMap := original[0].(Map)
+		originalMap["key"] = NewString("modified")
+
+		// Verify copy is unaffected
+		copiedMap := copied[0].(Map)
+		if copiedMap["key"].(*stringData).String() != "value" {
+			t.Error("Modifying map in original should not affect copy")
+		}
+	})
+
+	t.Run("Copy nil array", func(t *testing.T) {
+		var original Array
+		copied := original.Copy()
+
+		if copied != nil {
+			t.Error("Copy of nil array should be nil")
+		}
+	})
+
+	t.Run("Copy empty array", func(t *testing.T) {
+		original := Array{}
+		copied := original.Copy()
+
+		if copied == nil {
+			t.Error("Copy of empty array should not be nil")
+		}
+		if len(copied) != 0 {
+			t.Error("Copy of empty array should be empty")
+		}
+	})
+}
+
+func TestCopyValue(t *testing.T) {
+	t.Run("Copy nil value", func(t *testing.T) {
+		var original format.Value
+		copied := copyValue(original)
+
+		if copied != nil {
+			t.Error("Copy of nil value should be nil")
+		}
+	})
+
+	t.Run("Copy primitive types", func(t *testing.T) {
+		// Primitive types should be returned as-is since they're immutable
+		tests := []format.Value{
+			NewString("test"),
+			NewNumberFromInteger(123),
+			NewBoolean(true),
+			NewNull(),
+		}
+
+		for _, original := range tests {
+			copied := copyValue(original)
+			if !copied.Equal(original) {
+				t.Errorf("Copied value should equal original for type %T", original)
+			}
+		}
+	})
+
+	t.Run("Copy complex nested structure", func(t *testing.T) {
+		original := Map{
+			"users": Array{
+				Map{
+					"name": NewString("Alice"),
+					"age":  NewNumberFromInteger(30),
+					"tags": Array{
+						NewString("admin"),
+						NewString("active"),
+					},
+				},
+				Map{
+					"name": NewString("Bob"),
+					"age":  NewNumberFromInteger(25),
+					"tags": Array{
+						NewString("user"),
+					},
+				},
+			},
+			"metadata": Map{
+				"version": NewNumberFromInteger(1),
+				"active":  NewBoolean(true),
+			},
+		}
+
+		copied := copyValue(original).(Map)
+
+		// Verify it's a deep copy
+		originalUsers := original["users"].(Array)
+		originalFirstUser := originalUsers[0].(Map)
+		originalFirstUser["name"] = NewString("Modified")
+
+		copiedUsers := copied["users"].(Array)
+		copiedFirstUser := copiedUsers[0].(Map)
+		if copiedFirstUser["name"].(*stringData).String() != "Alice" {
+			t.Error("Deep nested modification should not affect copy")
+		}
+	})
+}
+
+// TestCopyConcurrency verifies that Copy() prevents race conditions
+func TestCopyConcurrency(t *testing.T) {
+	original := Map{
+		"key1": NewString("value1"),
+		"key2": Map{
+			"nested": NewString("nested_value"),
+		},
+	}
+
+	// Create a copy
+	copied := original.Copy()
+
+	// Simulate concurrent modifications on original
+	done := make(chan bool)
+	go func() {
+		for i := 0; i < 100; i++ {
+			original["key1"] = NewString("modified")
+			original["new_key"] = NewNumberFromInteger(i)
+		}
+		done <- true
+	}()
+
+	// Simultaneously access the copy (should not race)
+	go func() {
+		for i := 0; i < 100; i++ {
+			_ = copied["key1"]
+			_ = copied["key2"]
+		}
+		done <- true
+	}()
+
+	<-done
+	<-done
+
+	// Verify copy is unaffected by modifications to original
+	if copied["key1"].(*stringData).String() != "value1" {
+		t.Error("Copy should be unaffected by concurrent modifications to original")
+	}
+	if _, exists := copied["new_key"]; exists {
+		t.Error("Copy should not have new keys added to original")
+	}
+}
diff --git a/pkg/data/map.go b/pkg/data/map.go
@@ -93,3 +93,34 @@ func (m Map) ToJSONValue() (v any, err error) {
 	}
 	return jsonMap, nil
 }
+
+// Copy creates a deep copy of the Map to prevent concurrent access issues.
+// This is essential when passing map data across goroutine boundaries.
+func (m Map) Copy() Map {
+	if m == nil {
+		return nil
+	}
+	copied := make(Map, len(m))
+	for k, v := range m {
+		copied[k] = copyValue(v)
+	}
+	return copied
+}
+
+// copyValue creates a deep copy of any format.Value type.
+func copyValue(v format.Value) format.Value {
+	if v == nil {
+		return nil
+	}
+
+	switch val := v.(type) {
+	case Map:
+		return val.Copy()
+	case Array:
+		return val.Copy()
+	default:
+		// For primitive types (string, number, boolean, etc.) which are
+		// either immutable or contain no shared mutable state, return as-is.
+		return v
+	}
+}
diff --git a/pkg/memory/workflow.go b/pkg/memory/workflow.go
@@ -340,8 +340,23 @@ func (wfm *WorkflowMemory) Get(ctx context.Context, batchIdx int, p string) (mem
 	if err != nil {
 		return nil, err
 	}
-	return wfm.data[batchIdx].Get(pt)
+	value, err := wfm.data[batchIdx].Get(pt)
+	if err != nil {
+		return nil, err
+	}
 
+	// Return a deep copy to prevent concurrent map access issues.
+	// This is critical because the returned value may be accessed outside
+	// the mutex lock (e.g., in recipe.Eval -> ToJSONValue).
+	switch v := value.(type) {
+	case data.Map:
+		return v.Copy(), nil
+	case data.Array:
+		return v.Copy(), nil
+	default:
+		// Primitive types are safe to return as-is
+		return value, nil
+	}
 }
 
 func (wfm *WorkflowMemory) GetBatchSize() int {
