core/consensus: fixed instance_io race

core/consensus/utils/instance_io.go

@@ -3,6 +3,7 @@
 package utils
 
 import (
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -17,23 +18,20 @@ const (
 // NewInstanceIO returns a new instanceIO.
 func NewInstanceIO[T any]() *InstanceIO[T] {
 	return &InstanceIO[T]{
-		Participated: make(chan struct{}),
-		Proposed:     make(chan struct{}),
-		Running:      make(chan struct{}),
-		RecvBuffer:   make(chan T, RecvBufferSize),
-		HashCh:       make(chan [32]byte, 1),
-		ValueCh:      make(chan proto.Message, 1),
-		ErrCh:        make(chan error, 1),
-		DecidedAtCh:  make(chan time.Time, 1),
+		RecvBuffer:  make(chan T, RecvBufferSize),
+		HashCh:      make(chan [32]byte, 1),
+		ValueCh:     make(chan proto.Message, 1),
+		ErrCh:       make(chan error, 1),
+		DecidedAtCh: make(chan time.Time, 1),
 	}
 }
 
 // InstanceIO defines the async input and output channels of a
 // single consensus instance in the Component.
 type InstanceIO[T any] struct {
-	Participated chan struct{}      // Closed when Participate was called for this instance.
-	Proposed     chan struct{}      // Closed when Propose was called for this instance.
-	Running      chan struct{}      // Closed when runInstance was already called.
+	Participated int32              // Closed when Participate was called for this instance.
+	Proposed     int32              // Closed when Propose was called for this instance.
+	Running      int32              // Closed when runInstance was already called.
 	RecvBuffer   chan T             // Outer receive buffers.
 	HashCh       chan [32]byte      // Async input hash channel.
 	ValueCh      chan proto.Message // Async input value channel.
@@ -44,11 +42,8 @@ type InstanceIO[T any] struct {
 // MarkParticipated marks the instance as participated.
 // It returns an error if the instance was already marked as participated.
 func (io *InstanceIO[T]) MarkParticipated() error {
-	select {
-	case <-io.Participated:
+	if !atomic.CompareAndSwapInt32(&io.Participated, 0, 1) {
 		return errors.New("already participated")
-	default:
-		close(io.Participated)
 	}
 
 	return nil
@@ -57,11 +52,8 @@ func (io *InstanceIO[T]) MarkParticipated() error {
 // MarkProposed marks the instance as proposed.
 // It returns an error if the instance was already marked as proposed.
 func (io *InstanceIO[T]) MarkProposed() error {
-	select {
-	case <-io.Proposed:
+	if !atomic.CompareAndSwapInt32(&io.Proposed, 0, 1) {
 		return errors.New("already proposed")
-	default:
-		close(io.Proposed)
 	}
 
 	return nil
@@ -70,12 +62,5 @@ func (io *InstanceIO[T]) MarkProposed() error {
 // MaybeStart returns true if the instance wasn't running and has been started by this call,
 // otherwise it returns false if the instance was started in the past and is either running now or has completed.
 func (io *InstanceIO[T]) MaybeStart() bool {
-	select {
-	case <-io.Running:
-		return false
-	default:
-		close(io.Running)
-	}
-
-	return true
+	return atomic.CompareAndSwapInt32(&io.Running, 0, 1)
 }

core/consensus/utils/instance_io_test.go

@@ -42,7 +42,11 @@ func TestMaybeStart(t *testing.T) {
 	ok := io.MaybeStart()
 	require.True(t, ok)
 
-	// Second call fails.
+	// Subsequent calls fail.
+	ok = io.MaybeStart()
+	require.False(t, ok)
+	ok = io.MaybeStart()
+	require.False(t, ok)
 	ok = io.MaybeStart()
 	require.False(t, ok)
 }

testutil/compose/static/vouch/Dockerfile

@@ -1,6 +1,6 @@
-FROM wealdtech/ethdo:1.36.1 as ethdo
+FROM wealdtech/ethdo:1.35.2 as ethdo
 
-FROM attestant/vouch:1.9.2
+FROM attestant/vouch:1.9.0
 
 COPY --from=ethdo /app/ethdo /app/ethdo