Make worker thread-safe

Author: AlexPykavy

tests/worker/start_task_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 
 	"github.com/google/uuid"
@@ -61,3 +62,46 @@ func TestStartTask__ShouldReturnAnErrorIfCreatingTheSameTaskTwice(t *testing.T)
 	assert.Equal(t, 1, len(api.Worker.Tasks), "Tasks map should contain 1 task")
 	assert.NotNil(t, api.Worker.Tasks[testTask.ID], "Persisted task ID should match the one from request")
 }
+
+func TestStartTask__AllButOneRequestsShouldFailIfCreatingTheSameTaskSimultaneously(t *testing.T) {
+	api := &worker.Api{
+		Worker: &worker.Worker{
+			Tasks: make(map[uuid.UUID]*task.Task),
+		},
+	}
+	testTask := helper.PrintFileTask("print-task", helper.HostsFilePath)
+	numOfRequests := 10
+	requests := make([]*http.Request, numOfRequests)
+	responseRecorders := make([]*httptest.ResponseRecorder, numOfRequests)
+
+	var wg sync.WaitGroup
+	for i := range numOfRequests {
+		wg.Add(1)
+
+		requests[i] = helper.NewTaskPostRequest(testTask)
+		responseRecorders[i] = httptest.NewRecorder()
+
+		go func() {
+			defer wg.Done()
+
+			api.HandleCreateTask(responseRecorders[i], requests[i])
+		}()
+	}
+
+	wg.Wait()
+
+	succeededRequests, conflictedRequests := 0, 0
+	for i := range numOfRequests {
+		switch responseRecorders[i].Code {
+		case http.StatusCreated:
+			succeededRequests++
+		case http.StatusConflict:
+			conflictedRequests++
+		}
+	}
+
+	assert.Equal(t, 1, succeededRequests, "There should be only 1 succeeded request")
+	assert.Equal(t, numOfRequests-1, conflictedRequests, "There should be only N-1 conflicted requests")
+	assert.Equal(t, 1, len(api.Worker.Tasks), "Tasks map should contain 1 task")
+	assert.NotNil(t, api.Worker.Tasks[testTask.ID], "Persisted task ID should match the one from request")
+}

worker/worker.go

@@ -6,30 +6,44 @@ import (
 	"maps"
 	"os"
 	"os/exec"
+	"sync"
 
 	"github.com/google/uuid"
 )
 
 type Worker struct {
+	sync.RWMutex
+
 	Tasks map[uuid.UUID]*task.Task
 }
 
 func (w *Worker) ListTasks() iter.Seq[*task.Task] {
+	w.RLock()
+	defer w.RUnlock()
+
 	return maps.Values(w.Tasks)
 }
 
 func (w *Worker) GetTask(id uuid.UUID) *task.Task {
+	w.RLock()
+	defer w.RUnlock()
+
 	return w.Tasks[id]
 }
 
 func (w *Worker) StartTask(t task.Task) error {
+	w.Lock()
+
 	if _, ok := w.Tasks[t.ID]; ok {
+		w.Unlock()
 		return task.ErrAlreadyExists
 	}
 
 	t.Cmd = exec.Command(t.Executable, t.Args...)
 	w.Tasks[t.ID] = &t
 
+	w.Unlock()
+
 	stdout, err := t.Cmd.CombinedOutput()
 	os.Stdout.Write(stdout)
 	if err != nil {
@@ -40,7 +54,10 @@ func (w *Worker) StartTask(t task.Task) error {
 }
 
 func (w *Worker) StopTask(id uuid.UUID) error {
-	t := w.GetTask(id)
+	w.Lock()
+	defer w.Unlock()
+
+	t := w.Tasks[id]
 	if t == nil {
 		return task.ErrNotExists
 	}