fix: resolve race conditions in concurrent task execution

go.mod

@@ -0,0 +1,3 @@
+module github.com/mmccl5/go-task-task
+
+go 1.26.5

workspace/task/runner.go

@@ -3,9 +3,9 @@ package task
 import (
 	"context"
 	"sync"
-	"time"
 )
 
+// Runner executes tasks concurrently.
 type Runner struct {
 	mu    sync.RWMutex
 	tasks []*Task
@@ -23,6 +23,34 @@ func (r *Runner) AddTask(t *Task) {
 	r.tasks = append(r.tasks, t)
 }
 
+// RunnerStatus contains a snapshot of the runner's task states.
+type RunnerStatus struct {
+	TaskID string
+	State  State
+	Err    error
+}
+
+// Status returns a thread-safe snapshot of every task's state and error.
+func (r *Runner) Status() []RunnerStatus {
+	r.mu.RLock()
+	snap := make([]*Task, len(r.tasks))
+	copy(snap, r.tasks)
+	r.mu.RUnlock()
+
+	results := make([]RunnerStatus, len(snap))
+	for i, t := range snap {
+		results[i] = RunnerStatus{
+			TaskID: t.ID,
+			State:  t.GetState(),
+			Err:    t.GetErr(),
+		}
+	}
+	return results
+}
+
+// Run executes all tasks concurrently. It uses the task's own public API
+// for thread-safe state transitions so that concurrent readers (e.g. the
+// race-detector test in runner_test.go) do not cause data races.
 func (r *Runner) Run(ctx context.Context) {
 	r.mu.RLock()
 	tasks := make([]*Task, len(r.tasks))
@@ -35,35 +63,22 @@ func (r *Runner) Run(ctx context.Context) {
 		go func(task *Task) {
 			defer wg.Done()
 
+			// Check whether the context has been cancelled before we start.
 			if ctx.Err() != nil {
-				task.mu.Lock()
-				task.State = StateFailed
-				task.Err = ctx.Err()
-				task.History = append(task.History, StateFailed)
-				task.mu.Unlock()
+				task.TransitionTo(StateFailed, ctx.Err())
 				return
 			}
 
-			task.mu.Lock()
-			task.State = StateRunning
-			task.StartedAt = time.Now()
-			task.History = append(task.History, StateRunning)
-			task.mu.Unlock()
+			task.TransitionTo(StateRunning, nil)
 
 			err := task.Action(ctx)
 
-			task.mu.Lock()
-			task.FinishedAt = time.Now()
 			if err != nil {
-				task.State = StateFailed
-				task.Err = err
-				task.History = append(task.History, StateFailed)
+				task.TransitionTo(StateFailed, err)
 			} else {
-				task.State = StateCompleted
-				task.History = append(task.History, StateCompleted)
+				task.TransitionTo(StateCompleted, nil)
 			}
-			task.mu.Unlock()
 		}(t)
 	}
 	wg.Wait()
-}
+}

workspace/task/runner_test.go

@@ -26,11 +26,11 @@ func TestRunner_Run(t *testing.T) {
 
 	runner.Run(context.Background())
 
-	if t1.State != StateCompleted {
-		t.Errorf("expected t1 to be Completed, got %s", t1.State)
+	if s := t1.GetState(); s != StateCompleted {
+		t.Errorf("expected t1 to be Completed, got %s", s)
 	}
-	if t2.State != StateFailed {
-		t.Errorf("expected t2 to be Failed, got %s", t2.State)
+	if s := t2.GetState(); s != StateFailed {
+		t.Errorf("expected t2 to be Failed, got %s", s)
 	}
 }
 
@@ -98,4 +98,204 @@ func TestRunner_Run_Cancel(t *testing.T) {
 	if err := t1.GetErr(); err != context.Canceled {
 		t.Errorf("expected t1 error to be context.Canceled, got %v", err)
 	}
-}
+}
+
+func TestRunner_Run_ConcurrentWrites(t *testing.T) {
+	// Create tasks that use SetMetadata/GetMetadata while running.
+	runner := NewRunner()
+	numTasks := 50
+
+	// Collect tasks for background reading.
+	taskList := make([]*Task, numTasks)
+	for i := 0; i < numTasks; i++ {
+		id := fmt.Sprintf("write-%d", i)
+		tk := NewTask(id, nil)
+		tk2 := tk
+		tk.Action = func(ctx context.Context) error {
+			for j := 0; j < 5; j++ {
+				tk2.SetMetadata(fmt.Sprintf("key-%d", j), j)
+				_, _ = tk2.GetMetadata(fmt.Sprintf("key-%d", j))
+			}
+			return nil
+		}
+		runner.AddTask(tk)
+		taskList[i] = tk
+	}
+
+	// Concurrently read Metadata from another goroutine while tasks run.
+	stopChan := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stopChan:
+				return
+			default:
+				for _, t := range taskList {
+					_ = t.GetState()
+					_, _ = t.GetMetadata("key-0")
+				}
+				time.Sleep(1 * time.Millisecond)
+			}
+		}
+	}()
+
+	runner.Run(context.Background())
+	close(stopChan)
+	wg.Wait()
+}
+
+func TestRunner_Run_Timeout(t *testing.T) {
+	runner := NewRunner()
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+
+	t1 := NewTask("slow", func(ctx context.Context) error {
+		select {
+		case <-time.After(500 * time.Millisecond):
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	})
+
+	runner.AddTask(t1)
+	runner.Run(ctx)
+
+	if state := t1.GetState(); state != StateFailed {
+		t.Errorf("expected t1 to be Failed after timeout, got %s", state)
+	}
+}
+
+func TestRunner_Run_Status(t *testing.T) {
+	runner := NewRunner()
+	t1 := NewTask("s1", func(ctx context.Context) error { return nil })
+	t2 := NewTask("s2", func(ctx context.Context) error { return errors.New("err") })
+
+	runner.AddTask(t1)
+	runner.AddTask(t2)
+	runner.Run(context.Background())
+
+	st := runner.Status()
+	if len(st) != 2 {
+		t.Fatalf("expected 2 status entries, got %d", len(st))
+	}
+	if st[0].State != StateCompleted && st[0].State != StateFailed {
+		t.Errorf("unexpected state for task s1: %s", st[0].State)
+	}
+	if st[1].State != StateFailed && st[1].State != StateCompleted {
+		t.Errorf("unexpected state for task s2: %s", st[1].State)
+	}
+}
+
+func TestRunner_Run_Historical(t *testing.T) {
+	runner := NewRunner()
+	t1 := NewTask("hist", func(ctx context.Context) error { return nil })
+
+	runner.AddTask(t1)
+	runner.Run(context.Background())
+
+	hist := t1.GetHistory()
+	expectedStates := []State{StatePending, StateRunning, StateCompleted}
+	if len(hist) != len(expectedStates) {
+		t.Fatalf("expected %d history entries, got %d: %v", len(expectedStates), len(hist), hist)
+	}
+	for i, s := range expectedStates {
+		if hist[i] != s {
+			t.Errorf("history[%d] = %s, want %s", i, hist[i], s)
+		}
+	}
+}
+
+func TestTask_ConcurrentAccessors(t *testing.T) {
+	task := NewTask("stress", func(ctx context.Context) error {
+		time.Sleep(20 * time.Millisecond)
+		return nil
+	})
+
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < 100; j++ {
+				_ = task.GetState()
+				_ = task.GetErr()
+				_, _ = task.GetMetadata("any")
+				_ = task.GetHistory()
+				_ = task.GetStartedAt()
+				_ = task.GetFinishedAt()
+			}
+		}()
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for j := 0; j < 50; j++ {
+			task.SetState(StateRunning)
+			task.SetErr(nil)
+			task.SetMetadata(fmt.Sprintf("k-%d", j), j)
+		}
+		task.TransitionTo(StateCompleted, nil)
+	}()
+
+	wg.Wait()
+
+	if s := task.GetState(); s != StateCompleted {
+		t.Errorf("expected final state Completed, got %s", s)
+	}
+}
+
+func TestRunner_Run_MetadataRace(t *testing.T) {
+	runner := NewRunner()
+	numTasks := 50
+
+	var tasks []*Task
+	for i := 0; i < numTasks; i++ {
+		id := fmt.Sprintf("%d", i)
+		tk := NewTask(id, nil)
+		tk2 := tk
+		tk.Action = func(ctx context.Context) error {
+			tk2.SetMetadata("result", "ok")
+			return nil
+		}
+		runner.AddTask(tk)
+		tasks = append(tasks, tk)
+	}
+
+	var wg sync.WaitGroup
+	stopChan := make(chan struct{})
+
+	for i := 0; i < 3; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-stopChan:
+					return
+				default:
+					for _, t := range tasks {
+						_ = t.GetState()
+						_, _ = t.GetMetadata("result")
+						_ = t.GetErr()
+					}
+					time.Sleep(500 * time.Microsecond)
+				}
+			}
+		}()
+	}
+
+	runner.Run(context.Background())
+	close(stopChan)
+	wg.Wait()
+
+	for i, task := range tasks {
+		if s := task.GetState(); s != StateCompleted {
+			t.Errorf("task %d had state %s, want Completed", i, s)
+		}
+	}
+}

workspace/task/task.go

@@ -21,7 +21,7 @@ type Task struct {
 	State      State
 	Action     func(ctx context.Context) error
 	Err        error
-	Metadata   map[string]interface{ }
+	Metadata   map[string]interface{}
 	History    []State
 	StartedAt  time.Time
 	FinishedAt time.Time
@@ -37,19 +37,39 @@ func NewTask(id string, action func(ctx context.Context) error) *Task {
 	}
 }
 
+// GetState returns the current state of the task.
 func (t *Task) GetState() State {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	return t.State
 }
 
+// SetState updates the task's state and appends it to the history.
 func (t *Task) SetState(state State) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.State = state
 	t.History = append(t.History, state)
 }
 
+// TransitionTo is an atomic state-and-error transition.
+// It sets the state, appends to history, records the error,
+// and records FinishedAt if the state is a terminal one
+// (Completed or Failed).
+func (t *Task) TransitionTo(state State, err error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.State = state
+	t.History = append(t.History, state)
+	t.Err = err
+	if state == StateRunning {
+		t.StartedAt = time.Now()
+	}
+	if state == StateCompleted || state == StateFailed {
+		t.FinishedAt = time.Now()
+	}
+}
+
 func (t *Task) GetErr() error {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
@@ -93,4 +113,4 @@ func (t *Task) GetFinishedAt() time.Time {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 	return t.FinishedAt
-}
+}