From 964e3779825cb1292e43f90865f7c1c201d78288 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Fri, 10 Jul 2026 13:18:41 +0100 Subject: [PATCH] Fix concurrent map access on task index map in flow tasks The concurrent and matrix flow tasks wrote the task index map inside the child spawn loop while each child's watcher goroutine read it. That is a concurrent map read and write, which the Go runtime turns into a fatal error that cannot be recovered, taking down the whole process and every running test. Populate the index map fully before spawning any child goroutine, since the mapping is known up front and never changes after that. Add race regression tests that drive the executors with many children and fail under -race without the fix. --- pkg/tasks/run_task_matrix/task.go | 9 +- pkg/tasks/run_task_matrix/task_test.go | 124 ++++++++++++++++++++ pkg/tasks/run_tasks_concurrent/task.go | 9 +- pkg/tasks/run_tasks_concurrent/task_test.go | 122 +++++++++++++++++++ 4 files changed, 260 insertions(+), 4 deletions(-) create mode 100644 pkg/tasks/run_task_matrix/task_test.go create mode 100644 pkg/tasks/run_tasks_concurrent/task_test.go diff --git a/pkg/tasks/run_task_matrix/task.go b/pkg/tasks/run_task_matrix/task.go index 1be318eb..88a51eeb 100644 --- a/pkg/tasks/run_task_matrix/task.go +++ b/pkg/tasks/run_task_matrix/task.go @@ -123,6 +123,13 @@ func (t *Task) Execute(ctx context.Context) error { var currentTaskWaitChan, previousTaskWaitChan chan bool + // populate the task index map before spawning any child goroutines. + // the watcher goroutines read this map, so writing it here first avoids + // a concurrent map read/write against the spawn loop below. + for i := range t.tasks { + t.taskIdxMap[t.tasks[i]] = i + } + // start child tasks for i := range t.tasks { taskWaitGroup.Add(1) @@ -132,8 +139,6 @@ func (t *Task) Execute(ctx context.Context) error { currentTaskWaitChan = make(chan bool) } - t.taskIdxMap[t.tasks[i]] = i - go func(i int, taskWaitChan, prevTaskWaitChan chan bool) { defer taskWaitGroup.Done() diff --git a/pkg/tasks/run_task_matrix/task_test.go b/pkg/tasks/run_task_matrix/task_test.go new file mode 100644 index 00000000..ce1f442a --- /dev/null +++ b/pkg/tasks/run_task_matrix/task_test.go @@ -0,0 +1,124 @@ +package runtaskmatrix + +import ( + "context" + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/sirupsen/logrus" +) + +// stubTaskState reports a finished, successful child task so the watcher reads +// the task index map once and exits. +type stubTaskState struct { + index types.TaskIndex +} + +func (s *stubTaskState) Index() types.TaskIndex { return s.index } +func (s *stubTaskState) ParentIndex() types.TaskIndex { return 0 } +func (s *stubTaskState) ID() string { return "" } +func (s *stubTaskState) Name() string { return "stub" } +func (s *stubTaskState) Title() string { return "stub" } +func (s *stubTaskState) Description() string { return "" } +func (s *stubTaskState) Config() any { return nil } +func (s *stubTaskState) Timeout() time.Duration { return 0 } +func (s *stubTaskState) GetTaskStatusVars() types.Variables { return nil } +func (s *stubTaskState) GetScopeOwner() types.TaskIndex { return 0 } + +func (s *stubTaskState) GetTaskStatus() *types.TaskStatus { + return &types.TaskStatus{Index: s.index, IsRunning: false, Result: types.TaskResultSuccess} +} + +func (s *stubTaskState) GetTaskResultUpdateChan(oldResult types.TaskResult) <-chan bool { + return nil +} + +// stubScheduler runs the supplied watch function in its own goroutine, which is +// what concurrently reads the task index map. +type stubScheduler struct{} + +func (s *stubScheduler) GetTaskState(taskIndex types.TaskIndex) types.TaskState { + return &stubTaskState{index: taskIndex} +} +func (s *stubScheduler) GetTaskCount() uint64 { return 0 } +func (s *stubScheduler) GetAllTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetAllCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetServices() types.TaskServices { return nil } +func (s *stubScheduler) GetTestRunID() uint64 { return 0 } +func (s *stubScheduler) GetTestRunCtx() context.Context { return context.Background() } +func (s *stubScheduler) TestResultPath() (string, error) { return "", nil } + +func (s *stubScheduler) ParseTaskOptions(_ helper.IRawMessage) (*types.TaskOptions, error) { + return nil, nil +} + +func (s *stubScheduler) ExecuteTask(ctx context.Context, taskIndex types.TaskIndex, taskWatchFn func(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex)) error { + if taskWatchFn != nil { + watchCtx, cancel := context.WithCancel(ctx) + go taskWatchFn(watchCtx, cancel, taskIndex) + } + + return nil +} + +var ( + _ types.TaskSchedulerRunner = (*stubScheduler)(nil) + _ types.TaskState = (*stubTaskState)(nil) +) + +// TestExecuteTaskIndexMapNoRace runs the matrix executor with many children so the +// per-child watcher goroutines read taskIdxMap while the executor spawns the +// remaining children. The index map must be fully written before any goroutine +// starts, otherwise this races (run with -race). +func TestExecuteTaskIndexMapNoRace(t *testing.T) { + const childCount = 64 + + logger := logrus.New() + logger.SetOutput(io.Discard) + logger.SetLevel(logrus.PanicLevel) + + tasks := make([]types.TaskIndex, childCount) + for i := range tasks { + tasks[i] = types.TaskIndex(i + 1) + } + + config := DefaultConfig() + config.RunConcurrent = true + + task := &Task{ + config: config, + logger: logger, + tasks: tasks, + taskIdxMap: map[types.TaskIndex]int{}, + resultNotifyChan: make(chan taskResultUpdate, childCount), + } + task.ctx = &types.TaskContext{ + Scheduler: &stubScheduler{}, + Index: types.TaskIndex(1000), + SetResult: func(types.TaskResult) {}, + ReportProgress: func(float64, string) {}, + } + + done := make(chan error, 1) + go func() { + done <- task.Execute(context.Background()) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Execute returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Execute did not complete in time") + } + + if len(task.taskIdxMap) != childCount { + t.Fatalf("taskIdxMap has %d entries, want %d", len(task.taskIdxMap), childCount) + } +} diff --git a/pkg/tasks/run_tasks_concurrent/task.go b/pkg/tasks/run_tasks_concurrent/task.go index b8049cde..6a1f4499 100644 --- a/pkg/tasks/run_tasks_concurrent/task.go +++ b/pkg/tasks/run_tasks_concurrent/task.go @@ -124,12 +124,17 @@ func (t *Task) Execute(ctx context.Context) error { t.taskCtx = taskCtx + // populate the task index map before spawning any child goroutines. + // the watcher goroutines read this map, so writing it here first avoids + // a concurrent map read/write against the spawn loop below. + for i := range t.tasks { + t.taskIdxMap[t.tasks[i]] = i + } + // start child tasks for i := range t.tasks { taskWaitGroup.Add(1) - t.taskIdxMap[t.tasks[i]] = i - go func(i int) { defer taskWaitGroup.Done() diff --git a/pkg/tasks/run_tasks_concurrent/task_test.go b/pkg/tasks/run_tasks_concurrent/task_test.go new file mode 100644 index 00000000..e5ad89e2 --- /dev/null +++ b/pkg/tasks/run_tasks_concurrent/task_test.go @@ -0,0 +1,122 @@ +package runtasksconcurrent + +import ( + "context" + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/sirupsen/logrus" +) + +// stubTaskState reports a finished, successful child task so the watcher reads +// the task index map once and exits. +type stubTaskState struct { + index types.TaskIndex +} + +func (s *stubTaskState) Index() types.TaskIndex { return s.index } +func (s *stubTaskState) ParentIndex() types.TaskIndex { return 0 } +func (s *stubTaskState) ID() string { return "" } +func (s *stubTaskState) Name() string { return "stub" } +func (s *stubTaskState) Title() string { return "stub" } +func (s *stubTaskState) Description() string { return "" } +func (s *stubTaskState) Config() any { return nil } +func (s *stubTaskState) Timeout() time.Duration { return 0 } +func (s *stubTaskState) GetTaskStatusVars() types.Variables { return nil } +func (s *stubTaskState) GetScopeOwner() types.TaskIndex { return 0 } + +func (s *stubTaskState) GetTaskStatus() *types.TaskStatus { + return &types.TaskStatus{Index: s.index, IsRunning: false, Result: types.TaskResultSuccess} +} + +func (s *stubTaskState) GetTaskResultUpdateChan(oldResult types.TaskResult) <-chan bool { + return nil +} + +// stubScheduler mirrors the real scheduler's contract that matters for this test: +// ExecuteTask runs the supplied watch function in its own goroutine, which is what +// concurrently reads the task index map. +type stubScheduler struct{} + +func (s *stubScheduler) GetTaskState(taskIndex types.TaskIndex) types.TaskState { + return &stubTaskState{index: taskIndex} +} +func (s *stubScheduler) GetTaskCount() uint64 { return 0 } +func (s *stubScheduler) GetAllTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetAllCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetServices() types.TaskServices { return nil } +func (s *stubScheduler) GetTestRunID() uint64 { return 0 } +func (s *stubScheduler) GetTestRunCtx() context.Context { return context.Background() } +func (s *stubScheduler) TestResultPath() (string, error) { return "", nil } + +func (s *stubScheduler) ParseTaskOptions(_ helper.IRawMessage) (*types.TaskOptions, error) { + return nil, nil +} + +func (s *stubScheduler) ExecuteTask(ctx context.Context, taskIndex types.TaskIndex, taskWatchFn func(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex)) error { + if taskWatchFn != nil { + watchCtx, cancel := context.WithCancel(ctx) + go taskWatchFn(watchCtx, cancel, taskIndex) + } + + return nil +} + +var ( + _ types.TaskSchedulerRunner = (*stubScheduler)(nil) + _ types.TaskState = (*stubTaskState)(nil) +) + +// TestExecuteTaskIndexMapNoRace runs the concurrent executor with many children so +// the per-child watcher goroutines read taskIdxMap while the executor spawns the +// remaining children. The index map must be fully written before any goroutine +// starts, otherwise this races (run with -race). +func TestExecuteTaskIndexMapNoRace(t *testing.T) { + const childCount = 64 + + logger := logrus.New() + logger.SetOutput(io.Discard) + logger.SetLevel(logrus.PanicLevel) + + tasks := make([]types.TaskIndex, childCount) + for i := range tasks { + tasks[i] = types.TaskIndex(i + 1) + } + + task := &Task{ + config: DefaultConfig(), + logger: logger, + tasks: tasks, + taskIdxMap: map[types.TaskIndex]int{}, + resultNotifyChan: make(chan taskResultUpdate, childCount), + } + task.ctx = &types.TaskContext{ + Scheduler: &stubScheduler{}, + Index: types.TaskIndex(1000), + SetResult: func(types.TaskResult) {}, + ReportProgress: func(float64, string) {}, + } + + done := make(chan error, 1) + go func() { + done <- task.Execute(context.Background()) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Execute returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Execute did not complete in time") + } + + if len(task.taskIdxMap) != childCount { + t.Fatalf("taskIdxMap has %d entries, want %d", len(task.taskIdxMap), childCount) + } +}