challenger: Fix thread safety in op-challenger tests.

op-challenger/game/monitor_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"sync"
 	"testing"
 	"time"
 
@@ -60,15 +61,16 @@ func TestMonitorGames(t *testing.T) {
 		go func() {
 			headerNotSent := true
 			for {
-				if len(sched.scheduled) >= 1 {
+				if len(sched.Scheduled()) >= 1 {
 					break
 				}
-				if mockHeadSource.sub == nil {
+				sub := mockHeadSource.Sub()
+				if sub == nil {
 					continue
 				}
 				if headerNotSent {
 					select {
-					case mockHeadSource.sub.headers <- &ethtypes.Header{
+					case sub.headers <- &ethtypes.Header{
 						Number: big.NewInt(1),
 					}:
 						headerNotSent = false
@@ -80,15 +82,15 @@ func TestMonitorGames(t *testing.T) {
 				// Just to avoid a tight loop
 				time.Sleep(100 * time.Millisecond)
 			}
-			mockHeadSource.err = fmt.Errorf("eth subscribe test error")
+			mockHeadSource.SetErr(fmt.Errorf("eth subscribe test error"))
 			cancel()
 		}()
 
 		monitor.StartMonitoring()
 		<-ctx.Done()
 		monitor.StopMonitoring()
-		require.Len(t, sched.scheduled, 1)
-		require.Equal(t, []common.Address{addr1, addr2}, sched.scheduled[0])
+		require.Len(t, sched.Scheduled(), 1)
+		require.Equal(t, []common.Address{addr1, addr2}, sched.Scheduled()[0])
 	})
 
 	t.Run("Resubscribes on error", func(t *testing.T) {
@@ -103,19 +105,20 @@ func TestMonitorGames(t *testing.T) {
 		go func() {
 			// Wait for the subscription to be created
 			waitErr := wait.For(context.Background(), 5*time.Second, func() (bool, error) {
-				return mockHeadSource.sub != nil, nil
+				return mockHeadSource.Sub() != nil, nil
 			})
 			require.NoError(t, waitErr)
-			mockHeadSource.sub.errChan <- fmt.Errorf("test error")
+			mockHeadSource.Sub().errChan <- fmt.Errorf("test error")
 			for {
-				if len(sched.scheduled) >= 1 {
+				if len(sched.Scheduled()) >= 1 {
 					break
 				}
-				if mockHeadSource.sub == nil {
+				sub := mockHeadSource.Sub()
+				if sub == nil {
 					continue
 				}
 				select {
-				case mockHeadSource.sub.headers <- &ethtypes.Header{
+				case sub.headers <- &ethtypes.Header{
 					Number: big.NewInt(1),
 				}:
 				case <-ctx.Done():
@@ -126,15 +129,15 @@ func TestMonitorGames(t *testing.T) {
 				time.Sleep(100 * time.Millisecond)
 			}
 			require.NoError(t, waitErr)
-			mockHeadSource.err = fmt.Errorf("eth subscribe test error")
+			mockHeadSource.SetErr(fmt.Errorf("eth subscribe test error"))
 			cancel()
 		}()
 
 		monitor.StartMonitoring()
 		<-ctx.Done()
 		monitor.StopMonitoring()
-		require.NotEmpty(t, sched.scheduled) // We might get more than one update scheduled.
-		require.Equal(t, []common.Address{addr1, addr2}, sched.scheduled[0])
+		require.NotEmpty(t, sched.Scheduled()) // We might get more than one update scheduled.
+		require.Equal(t, []common.Address{addr1, addr2}, sched.Scheduled()[0])
 	})
 }
 
@@ -147,8 +150,8 @@ func TestMonitorCreateAndProgressGameAgents(t *testing.T) {
 
 	require.NoError(t, monitor.progressGames(context.Background(), common.Hash{0x01}))
 
-	require.Len(t, sched.scheduled, 1)
-	require.Equal(t, []common.Address{addr1, addr2}, sched.scheduled[0])
+	require.Len(t, sched.Scheduled(), 1)
+	require.Equal(t, []common.Address{addr1, addr2}, sched.Scheduled()[0])
 }
 
 func TestMonitorOnlyScheduleSpecifiedGame(t *testing.T) {
@@ -159,8 +162,8 @@ func TestMonitorOnlyScheduleSpecifiedGame(t *testing.T) {
 
 	require.NoError(t, monitor.progressGames(context.Background(), common.Hash{0x01}))
 
-	require.Len(t, sched.scheduled, 1)
-	require.Equal(t, []common.Address{addr2}, sched.scheduled[0])
+	require.Len(t, sched.Scheduled(), 1)
+	require.Equal(t, []common.Address{addr2}, sched.Scheduled()[0])
 }
 
 func newFDG(proxy common.Address, timestamp uint64) types.GameMetadata {
@@ -197,15 +200,36 @@ func setupMonitorTest(
 }
 
 type mockNewHeadSource struct {
+	sync.Mutex
 	sub *mockSubscription
 	err error
 }
 
+func (m *mockNewHeadSource) Sub() *mockSubscription {
+	m.Lock()
+	defer m.Unlock()
+	return m.sub
+}
+
+func (m *mockNewHeadSource) SetSub(sub *mockSubscription) {
+	m.Lock()
+	defer m.Unlock()
+	m.sub = sub
+}
+
+func (m *mockNewHeadSource) SetErr(err error) {
+	m.Lock()
+	defer m.Unlock()
+	m.err = err
+}
+
 func (m *mockNewHeadSource) EthSubscribe(
-	ctx context.Context,
+	_ context.Context,
 	ch any,
-	args ...any,
+	_ ...any,
 ) (ethereum.Subscription, error) {
+	m.Lock()
+	defer m.Unlock()
 	errChan := make(chan error)
 	m.sub = &mockSubscription{errChan, (ch).(chan<- *ethtypes.Header)}
 	if m.err != nil {
@@ -230,18 +254,26 @@ type stubGameSource struct {
 }
 
 func (s *stubGameSource) FetchAllGamesAtBlock(
-	ctx context.Context,
-	earliest uint64,
-	blockHash common.Hash,
+	_ context.Context,
+	_ uint64,
+	_ common.Hash,
 ) ([]types.GameMetadata, error) {
 	return s.games, nil
 }
 
 type stubScheduler struct {
+	sync.Mutex
 	scheduled [][]common.Address
 }
 
+func (s *stubScheduler) Scheduled() [][]common.Address {
+	s.Lock()
+	defer s.Unlock()
+	return s.scheduled
+}
 func (s *stubScheduler) Schedule(games []types.GameMetadata) error {
+	s.Lock()
+	defer s.Unlock()
 	var addrs []common.Address
 	for _, game := range games {
 		addrs = append(addrs, game.Proxy)

op-challenger/game/scheduler/worker_test.go

@@ -3,6 +3,7 @@ package scheduler
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -28,21 +29,21 @@ func TestWorkerShouldProcessJobsUntilContextDone(t *testing.T) {
 		player: &test.StubGamePlayer{StatusValue: types.GameStatusInProgress},
 	}
 	waitErr := wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
-		return ms.activeCalls >= 1, nil
+		return ms.activeCalls.Load() >= 1, nil
 	})
 	require.NoError(t, waitErr)
-	require.Equal(t, ms.activeCalls, 1)
-	require.Equal(t, ms.idleCalls, 1)
+	require.EqualValues(t, ms.activeCalls.Load(), 1)
+	require.EqualValues(t, ms.idleCalls.Load(), 1)
 
 	in <- job{
 		player: &test.StubGamePlayer{StatusValue: types.GameStatusDefenderWon},
 	}
 	waitErr = wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
-		return ms.activeCalls >= 2, nil
+		return ms.activeCalls.Load() >= 2, nil
 	})
 	require.NoError(t, waitErr)
-	require.Equal(t, ms.activeCalls, 2)
-	require.Equal(t, ms.idleCalls, 2)
+	require.EqualValues(t, ms.activeCalls.Load(), 2)
+	require.EqualValues(t, ms.idleCalls.Load(), 2)
 
 	result1 := readWithTimeout(t, out)
 	result2 := readWithTimeout(t, out)
@@ -56,16 +57,16 @@ func TestWorkerShouldProcessJobsUntilContextDone(t *testing.T) {
 }
 
 type metricSink struct {
-	activeCalls int
-	idleCalls   int
+	activeCalls atomic.Int32
+	idleCalls   atomic.Int32
 }
 
 func (m *metricSink) ThreadActive() {
-	m.activeCalls++
+	m.activeCalls.Add(1)
 }
 
 func (m *metricSink) ThreadIdle() {
-	m.idleCalls++
+	m.idleCalls.Add(1)
 }
 
 func readWithTimeout[T any](t *testing.T, ch <-chan T) T {