[CLD-272]: feat(operations): enable concurrency support (#95)

CCIP team has raised that they want to start using Operations API and
they have a use case where they wan to deploy multiple contracts in
parallel (multiple go routine calling executeSequence), currently
Operations API is not thread safe due to the reporter.

- update MemoryReporter and RecentReporter to be thread safe

JIRA: https://smartcontract-it.atlassian.net/browse/CLD-272

Author: graham-chainlink

.changeset/itchy-dragons-shout.md

@@ -0,0 +1,5 @@
+---
+"chainlink-deployments-framework": minor
+---
+
+feat: Concurrency support for Operations API

.github/workflows/pull-request-main.yml

@@ -41,5 +41,5 @@ jobs:
       - name: Build and test
         uses: smartcontractkit/.github/actions/ci-test-go@eeb76b5870e3c17856d5a60fd064a053c023b5f5 # [email protected]
         with:
-          go-test-cmd: go test -coverprofile=coverage.txt $(go list ./...)
+          go-test-cmd: go test -race -coverprofile=coverage.txt $(go list ./...)
           use-go-cache: true

operations/execute_test.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"errors"
 	"math"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/Masterminds/semver/v3"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
@@ -666,6 +668,138 @@ func Test_loadPreviousSuccessfulReport(t *testing.T) {
 	}
 }
 
+func Test_ExecuteSequence_Concurrent(t *testing.T) {
+	t.Parallel()
+
+	version := semver.MustParse("1.0.0")
+
+	op := NewOperation("increment", version, "increment by 1",
+		func(b Bundle, deps any, input int) (output int, err error) {
+			return input + 1, nil
+		})
+
+	sequence := NewSequence("concurrent-seq", version, "concurrent sequence test",
+		func(b Bundle, deps any, input int) (int, error) {
+			res, err := ExecuteOperation(b, op, nil, input)
+			if err != nil {
+				return 0, err
+			}
+
+			// Introduce a small delay to increase chance of race conditions
+			time.Sleep(time.Millisecond)
+
+			return res.Output, nil
+		})
+
+	reporter := NewMemoryReporter()
+	bundle := NewBundle(context.Background, logger.Test(t), reporter)
+
+	const numGoroutines = 10
+	var wg sync.WaitGroup
+	wg.Add(numGoroutines)
+
+	// Channel to collect results
+	type result struct {
+		report SequenceReport[int, int]
+		err    error
+	}
+	results := make(chan result, numGoroutines)
+
+	for i := range numGoroutines {
+		go func(input int) {
+			defer wg.Done()
+
+			report, err := ExecuteSequence(bundle, sequence, nil, input)
+			results <- result{report, err}
+		}(i) // Each goroutine uses its index as input
+	}
+
+	wg.Wait()
+	close(results)
+
+	// Collect and verify results
+	for res := range results {
+		require.NoError(t, res.err, "ExecuteSequence should not return an error")
+		require.Nil(t, res.report.Err, "Report error should be nil")
+
+		// Output should be input + 1
+		input := res.report.Input
+		expectedOutput := input + 1
+		assert.Equal(t, expectedOutput, res.report.Output,
+			"Output should be input + 1 for input %d", input)
+
+		// Verify execution reports
+		assert.Len(t, res.report.ExecutionReports, 2,
+			"Should have 2 execution reports (sequence + operation)")
+	}
+
+	// Verify reporter has all reports
+	allReports, err := reporter.GetReports()
+	require.NoError(t, err)
+
+	// We expect 2*numGoroutines reports (1 sequence + 1 operation per goroutine)
+	assert.Len(t, allReports, numGoroutines*2,
+		"Reporter should have %d reports", numGoroutines*2)
+}
+
+func Test_ExecuteOperation_Concurrent(t *testing.T) {
+	t.Parallel()
+
+	version := semver.MustParse("1.0.0")
+
+	op := NewOperation("increment", version, "increment by 1",
+		func(b Bundle, deps any, input int) (output int, err error) {
+			// Introduce a small delay to increase chance of race conditions
+			time.Sleep(time.Millisecond)
+			return input + 1, nil
+		})
+
+	reporter := NewMemoryReporter()
+	bundle := NewBundle(context.Background, logger.Test(t), reporter)
+
+	const numGoroutines = 10
+	var wg sync.WaitGroup
+	wg.Add(numGoroutines)
+
+	// Channel to collect results
+	type result struct {
+		report Report[int, int]
+		err    error
+	}
+	results := make(chan result, numGoroutines)
+
+	for i := range numGoroutines {
+		go func(input int) {
+			defer wg.Done()
+
+			report, err := ExecuteOperation(bundle, op, nil, input)
+			results <- result{report, err}
+		}(i) // Each goroutine uses its index as input
+	}
+
+	wg.Wait()
+	close(results)
+
+	for res := range results {
+		require.NoError(t, res.err, "ExecuteOperation should not return an error")
+		require.Nil(t, res.report.Err, "Report error should be nil")
+
+		// Output should be input + 1
+		input := res.report.Input
+		expectedOutput := input + 1
+		assert.Equal(t, expectedOutput, res.report.Output,
+			"Output should be input + 1 for input %d", input)
+	}
+
+	// Verify reporter has all reports
+	allReports, err := reporter.GetReports()
+	require.NoError(t, err)
+
+	// We expect numGoroutines reports (1 per goroutine)
+	assert.Len(t, allReports, numGoroutines,
+		"Reporter should have %d reports", numGoroutines)
+}
+
 type errorReporter struct {
 	Reporter
 	GetReportError           error

operations/report.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -93,8 +94,10 @@ type Reporter interface {
 }
 
 // MemoryReporter stores reports in memory.
+// This is thread-safe and can be used in a multi-threaded environment.
 type MemoryReporter struct {
 	reports []Report[any, any]
+	mu      sync.RWMutex
 }
 
 type MemoryReporterOption func(*MemoryReporter)
@@ -119,19 +122,32 @@ func NewMemoryReporter(options ...MemoryReporterOption) *MemoryReporter {
 
 // AddReport adds a report to the memory reporter.
 func (e *MemoryReporter) AddReport(report Report[any, any]) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
 	e.reports = append(e.reports, report)
 
 	return nil
 }
 
 // GetReports returns all reports.
 func (e *MemoryReporter) GetReports() ([]Report[any, any], error) {
-	return e.reports, nil
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
+	// Create a copy to avoid data races after returning
+	reports := make([]Report[any, any], len(e.reports))
+	copy(reports, e.reports)
+
+	return reports, nil
 }
 
 // GetReport returns a report by ID.
 // Returns ErrReportNotFound if the report is not found.
 func (e *MemoryReporter) GetReport(id string) (Report[any, any], error) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
 	for _, report := range e.reports {
 		if report.ID == id {
 			return report, nil
@@ -145,14 +161,29 @@ func (e *MemoryReporter) GetReport(id string) (Report[any, any], error) {
 // It does this by recursively fetching all the child reports.
 // Useful when returning all the reports in a sequence to the changeset output.
 func (e *MemoryReporter) GetExecutionReports(seqID string) ([]Report[any, any], error) {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
 	var allReports []Report[any, any]
 
 	var getReportsRecursively func(id string) error
 	getReportsRecursively = func(id string) error {
-		report, err := e.GetReport(id)
-		if err != nil {
-			return err
+		var report Report[any, any]
+		found := false
+
+		for _, r := range e.reports {
+			if r.ID == id {
+				report = r
+				found = true
+
+				break
+			}
+		}
+
+		if !found {
+			return fmt.Errorf("report_id %s: %w", id, ErrReportNotFound)
 		}
+
 		for _, childID := range report.ChildOperationReports {
 			if err := getReportsRecursively(childID); err != nil {
 				return err
@@ -172,24 +203,35 @@ func (e *MemoryReporter) GetExecutionReports(seqID string) ([]Report[any, any],
 
 // RecentReporter is a wrapper around a Reporter that keeps track of the most recent reports.
 // Useful when trying to get a list of reports that was recently added in a sequence.
+// It is thread-safe and can be used in a multi-threaded environment.
 type RecentReporter struct {
 	Reporter
 	recentReports []Report[any, any]
+	mu            sync.RWMutex
 }
 
 // AddReport adds a report to the recent reporter.
 func (e *RecentReporter) AddReport(report Report[any, any]) error {
+	// First add to underlying reporter
 	err := e.Reporter.AddReport(report)
 	if err != nil {
 		return err
 	}
+
+	// Then add to recent reports
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
 	e.recentReports = append(e.recentReports, report)
 
 	return nil
 }
 
 // GetRecentReports returns all the reports that was added since the construction of the RecentReporter.
 func (e *RecentReporter) GetRecentReports() []Report[any, any] {
+	e.mu.RLock()
+	defer e.mu.RUnlock()
+
 	return e.recentReports
 }