Fix data race on finality checkpoint

pkg/beacon/beacon.go

@@ -190,6 +190,7 @@ type node struct {
 	nodeVersionMu   sync.RWMutex
 	peers           types.Peers
 	finality        *v1.Finality
+	finalityMu      sync.RWMutex
 	spec            *state.Spec
 	specMu          sync.RWMutex
 	wallclock       *ethwallclock.EthereumBeaconChain
@@ -387,6 +388,9 @@ func (n *node) Status() *Status {
 }
 
 func (n *node) Finality() (*v1.Finality, error) {
+	n.finalityMu.RLock()
+	defer n.finalityMu.RUnlock()
+
 	if n.finality == nil {
 		return nil, errors.New("finality not available")
 	}

pkg/beacon/beacon_test.go

@@ -6,6 +6,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/chuckpreslar/emission"
+	eapi "github.com/ethpandaops/go-eth2-client/api"
+	v1 "github.com/ethpandaops/go-eth2-client/api/v1"
+	"github.com/ethpandaops/go-eth2-client/spec/phase0"
 	"github.com/sirupsen/logrus"
 )
 
@@ -123,3 +127,58 @@ func TestLifecycleStartStopSequence(t *testing.T) {
 		t.Error("context was not cancelled after Stop")
 	}
 }
+
+// finalityClient is a minimal eth2client.FinalityProvider used to drive
+// FetchFinality without a real beacon node.
+type finalityClient struct{}
+
+func (f *finalityClient) Name() string    { return "fake" }
+func (f *finalityClient) Address() string { return "fake://" }
+func (f *finalityClient) IsActive() bool  { return true }
+func (f *finalityClient) IsSynced() bool  { return true }
+
+func (f *finalityClient) Finality(
+	_ context.Context, _ *eapi.FinalityOpts,
+) (*eapi.Response[*v1.Finality], error) {
+	return &eapi.Response[*v1.Finality]{
+		Data: &v1.Finality{
+			Finalized:         &phase0.Checkpoint{Epoch: 1, Root: phase0.Root{0x01}},
+			Justified:         &phase0.Checkpoint{Epoch: 2, Root: phase0.Root{0x02}},
+			PreviousJustified: &phase0.Checkpoint{Epoch: 3, Root: phase0.Root{0x03}},
+		},
+	}, nil
+}
+
+// TestFinalityMutex exercises FetchFinality and Finality() concurrently,
+// mirroring the shape of the epoch cron, the finalized_checkpoint event
+// handler and a direct consumer call all hitting finality at once. It
+// should pass cleanly under -race.
+func TestFinalityMutex(t *testing.T) {
+	n := &node{
+		log:    logrus.New(),
+		broker: emission.NewEmitter(),
+		client: &finalityClient{},
+	}
+
+	var wg sync.WaitGroup
+
+	for range 8 {
+		wg.Go(func() {
+			for range 50 {
+				if _, err := n.FetchFinality(context.Background(), "head"); err != nil {
+					t.Error(err)
+				}
+			}
+		})
+	}
+
+	for range 8 {
+		wg.Go(func() {
+			for range 50 {
+				_, _ = n.Finality()
+			}
+		})
+	}
+
+	wg.Wait()
+}

pkg/beacon/fetch.go

@@ -113,6 +113,8 @@ func (n *node) FetchFinality(ctx context.Context, stateID string) (*v1.Finality,
 	finality := rsp.Data
 
 	if stateID == "head" {
+		n.finalityMu.Lock()
+
 		changed := false
 		if n.finality == nil ||
 			finality.Finalized.Root != n.finality.Finalized.Root ||
@@ -126,6 +128,8 @@ func (n *node) FetchFinality(ctx context.Context, stateID string) (*v1.Finality,
 
 		n.finality = finality
 
+		n.finalityMu.Unlock()
+
 		if changed {
 			n.publishFinalityCheckpointUpdated(ctx, finality)
 		}