wip

api/client/event/event_stream.go

@@ -34,6 +34,18 @@ type Event struct {
 	Data      []byte
 }
 
+// PublishEvent enqueues an event without blocking the producer. If the channel is full,
+// the event is dropped since only the most recent heads are relevant.
+func PublishEvent(eventsChannel chan<- *Event, event *Event) {
+	if eventsChannel == nil || event == nil {
+		return
+	}
+	select {
+	case eventsChannel <- event:
+	default:
+	}
+}
+
 // EventStream is responsible for subscribing to the Beacon API events endpoint
 // and dispatching received events to subscribers.
 type EventStream struct {
@@ -67,19 +79,20 @@ func (h *EventStream) Subscribe(eventsChannel chan<- *Event) {
 	fullUrl := h.host + "/eth/v1/events?topics=" + allTopics
 	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, fullUrl, nil)
 	if err != nil {
-		eventsChannel <- &Event{
+		PublishEvent(eventsChannel, &Event{
 			EventType: EventConnectionError,
 			Data:      []byte(errors.Wrap(err, "failed to create HTTP request").Error()),
-		}
+		})
+		return
 	}
 	req.Header.Set("Accept", api.EventStreamMediaType)
 	req.Header.Set("Connection", api.KeepAlive)
 	resp, err := h.httpClient.Do(req)
 	if err != nil {
-		eventsChannel <- &Event{
+		PublishEvent(eventsChannel, &Event{
 			EventType: EventConnectionError,
 			Data:      []byte(errors.Wrap(err, client.ErrConnectionIssue.Error()).Error()),
-		}
+		})
 		return
 	}
 
@@ -100,7 +113,6 @@ func (h *EventStream) Subscribe(eventsChannel chan<- *Event) {
 		select {
 		case <-h.ctx.Done():
 			log.Info("Context canceled, stopping event stream")
-			close(eventsChannel)
 			return
 		default:
 			line := scanner.Text()
@@ -109,7 +121,7 @@ func (h *EventStream) Subscribe(eventsChannel chan<- *Event) {
 				// Empty line indicates the end of an event
 				if eventType != "" && data != "" {
 					// Process the event when both eventType and data are set
-					eventsChannel <- &Event{EventType: eventType, Data: []byte(data)}
+					PublishEvent(eventsChannel, &Event{EventType: eventType, Data: []byte(data)})
 				}
 
 				// Reset eventType and data for the next event
@@ -130,9 +142,9 @@ func (h *EventStream) Subscribe(eventsChannel chan<- *Event) {
 	}
 
 	if err := scanner.Err(); err != nil {
-		eventsChannel <- &Event{
+		PublishEvent(eventsChannel, &Event{
 			EventType: EventConnectionError,
 			Data:      []byte(errors.Wrap(err, errors.Wrap(client.ErrConnectionIssue, "scanner failed").Error()).Error()),
-		}
+		})
 	}
 }

api/client/event/event_stream_test.go

@@ -53,7 +53,7 @@ func TestEventStream(t *testing.T) {
 	defer server.Close()
 
 	topics := []string{"head"}
-	eventsChannel := make(chan *Event, 1)
+	eventsChannel := make(chan *Event, 4)
 	stream, err := NewEventStream(t.Context(), http.DefaultClient, server.URL, topics)
 	require.NoError(t, err)
 	go stream.Subscribe(eventsChannel)
@@ -80,7 +80,7 @@ func TestEventStream(t *testing.T) {
 
 func TestEventStreamRequestError(t *testing.T) {
 	topics := []string{"head"}
-	eventsChannel := make(chan *Event, 1)
+	eventsChannel := make(chan *Event, 4)
 	ctx := t.Context()
 
 	// use valid url that will result in failed request with nil body

beacon-chain/operations/payloadattestation/BUILD.bazel

@@ -1,32 +0,0 @@
-load("@prysm//tools/go:def.bzl", "go_library", "go_test")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["pool.go"],
-    importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/operations/payloadattestation",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//consensus-types/primitives:go_default_library",
-        "//crypto/bls:go_default_library",
-        "//crypto/hash:go_default_library",
-        "//encoding/bytesutil:go_default_library",
-        "//proto/prysm/v1alpha1:go_default_library",
-        "@com_github_pkg_errors//:go_default_library",
-        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
-        "@org_golang_google_protobuf//proto:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["pool_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "//consensus-types/primitives:go_default_library",
-        "//crypto/bls:go_default_library",
-        "//encoding/bytesutil:go_default_library",
-        "//proto/prysm/v1alpha1:go_default_library",
-        "//testing/assert:go_default_library",
-        "//testing/require:go_default_library",
-    ],
-)

beacon-chain/operations/payloadattestation/pool.go

@@ -1,179 +0,0 @@
-package payloadattestation
-
-import (
-	"sync"
-
-	"github.com/OffchainLabs/go-bitfield"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
-	"github.com/OffchainLabs/prysm/v7/crypto/bls"
-	"github.com/OffchainLabs/prysm/v7/crypto/hash"
-	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
-	"github.com/pkg/errors"
-	"google.golang.org/protobuf/proto"
-)
-
-var errNilPayloadAttestationMessage = errors.New("nil payload attestation message")
-
-// PoolManager maintains pending payload attestations.
-// This pool is used by proposers to insert payload attestations into new blocks.
-type PoolManager interface {
-	// PendingPayloadAttestations returns all pending aggregated payload attestations.
-	// If a slot is provided, only attestations for that slot are returned.
-	PendingPayloadAttestations(slot ...primitives.Slot) []*ethpb.PayloadAttestation
-	// InsertPayloadAttestation inserts or aggregates a payload attestation
-	// message into the pool. The idx parameter is the PTC committee index
-	// of the validator (position in the bitvector).
-	InsertPayloadAttestation(msg *ethpb.PayloadAttestationMessage, idx uint64) error
-	// Seen returns true if the PTC committee index has already been seen
-	// for the given PayloadAttestationData.
-	Seen(data *ethpb.PayloadAttestationData, idx uint64) bool
-	// MarkIncluded removes the attestation matching the given data from the pool.
-	MarkIncluded(att *ethpb.PayloadAttestation)
-}
-
-// Pool is a concrete implementation of PoolManager.
-// Keyed by hash of PayloadAttestationData; stores aggregated PayloadAttestation.
-type Pool struct {
-	lock    sync.RWMutex
-	pending map[[32]byte]*ethpb.PayloadAttestation
-}
-
-// NewPool returns an initialized pool.
-func NewPool() *Pool {
-	return &Pool{
-		pending: make(map[[32]byte]*ethpb.PayloadAttestation),
-	}
-}
-
-// PendingPayloadAttestations returns all pending payload attestations.
-// If a slot filter is provided, only attestations for that slot are returned.
-func (p *Pool) PendingPayloadAttestations(slot ...primitives.Slot) []*ethpb.PayloadAttestation {
-	p.lock.RLock()
-	defer p.lock.RUnlock()
-
-	result := make([]*ethpb.PayloadAttestation, 0, len(p.pending))
-	for _, att := range p.pending {
-		if len(slot) > 0 && att.Data.Slot != slot[0] {
-			continue
-		}
-		result = append(result, att)
-	}
-	return result
-}
-
-// InsertPayloadAttestation inserts a payload attestation message into the pool,
-// aggregating it with any existing attestation that shares the same PayloadAttestationData.
-// The idx parameter is the PTC committee index used to set the aggregation bit.
-func (p *Pool) InsertPayloadAttestation(msg *ethpb.PayloadAttestationMessage, idx uint64) error {
-	if msg == nil || msg.Data == nil {
-		return errNilPayloadAttestationMessage
-	}
-
-	key, err := dataKey(msg.Data)
-	if err != nil {
-		return errors.Wrap(err, "could not compute data key")
-	}
-
-	p.lock.Lock()
-	defer p.lock.Unlock()
-
-	existing, ok := p.pending[key]
-	if !ok {
-		p.pending[key] = messageToPayloadAttestation(msg, idx)
-		return nil
-	}
-
-	if existing.AggregationBits.BitAt(idx) {
-		return nil
-	}
-
-	sig, err := aggregateSigFromMessage(existing, msg)
-	if err != nil {
-		return errors.Wrap(err, "could not aggregate signatures")
-	}
-	existing.Signature = sig
-	existing.AggregationBits.SetBitAt(idx, true)
-	return nil
-}
-
-// Seen returns true if the PTC committee index has already been seen
-// for the given PayloadAttestationData.
-func (p *Pool) Seen(data *ethpb.PayloadAttestationData, idx uint64) bool {
-	if data == nil {
-		return false
-	}
-
-	key, err := dataKey(data)
-	if err != nil {
-		return false
-	}
-
-	p.lock.RLock()
-	defer p.lock.RUnlock()
-
-	existing, ok := p.pending[key]
-	if !ok {
-		return false
-	}
-	return existing.AggregationBits.BitAt(idx)
-}
-
-// MarkIncluded removes the attestation with matching data from the pool.
-func (p *Pool) MarkIncluded(att *ethpb.PayloadAttestation) {
-	if att == nil || att.Data == nil {
-		return
-	}
-
-	key, err := dataKey(att.Data)
-	if err != nil {
-		return
-	}
-
-	p.lock.Lock()
-	defer p.lock.Unlock()
-
-	delete(p.pending, key)
-}
-
-// messageToPayloadAttestation creates a PayloadAttestation with a single
-// aggregated bit from the passed PayloadAttestationMessage.
-func messageToPayloadAttestation(msg *ethpb.PayloadAttestationMessage, idx uint64) *ethpb.PayloadAttestation {
-	bits := bitfield.NewBitvector512()
-	bits.SetBitAt(idx, true)
-	data := &ethpb.PayloadAttestationData{
-		BeaconBlockRoot:   bytesutil.SafeCopyBytes(msg.Data.BeaconBlockRoot),
-		Slot:              msg.Data.Slot,
-		PayloadPresent:    msg.Data.PayloadPresent,
-		BlobDataAvailable: msg.Data.BlobDataAvailable,
-	}
-	return &ethpb.PayloadAttestation{
-		AggregationBits: bits,
-		Data:            data,
-		Signature:       bytesutil.SafeCopyBytes(msg.Signature),
-	}
-}
-
-// aggregateSigFromMessage returns the aggregated signature by combining the
-// existing aggregated signature with the message's signature.
-func aggregateSigFromMessage(aggregated *ethpb.PayloadAttestation, message *ethpb.PayloadAttestationMessage) ([]byte, error) {
-	aggSig, err := bls.SignatureFromBytesNoValidation(aggregated.Signature)
-	if err != nil {
-		return nil, err
-	}
-	sig, err := bls.SignatureFromBytesNoValidation(message.Signature)
-	if err != nil {
-		return nil, err
-	}
-	return bls.AggregateSignatures([]bls.Signature{aggSig, sig}).Marshal(), nil
-}
-
-// dataKey computes a deterministic key for PayloadAttestationData
-// by hashing its serialized form.
-func dataKey(data *ethpb.PayloadAttestationData) ([32]byte, error) {
-	enc, err := proto.Marshal(data)
-	if err != nil {
-		return [32]byte{}, err
-	}
-	return hash.Hash(enc), nil
-}

beacon-chain/operations/payloadattestation/pool_test.go

@@ -1,291 +0,0 @@
-package payloadattestation
-
-import (
-	"testing"
-
-	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
-	"github.com/OffchainLabs/prysm/v7/crypto/bls"
-	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
-	"github.com/OffchainLabs/prysm/v7/testing/assert"
-	"github.com/OffchainLabs/prysm/v7/testing/require"
-)
-
-func TestPool_PendingPayloadAttestations(t *testing.T) {
-	t.Run("empty pool", func(t *testing.T) {
-		pool := NewPool()
-		atts := pool.PendingPayloadAttestations()
-		assert.Equal(t, 0, len(atts))
-	})
-
-	t.Run("non-empty pool", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		msg1 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		msg2 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 1,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              2,
-				PayloadPresent:    false,
-				BlobDataAvailable: true,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg1, 0))
-		require.NoError(t, pool.InsertPayloadAttestation(msg2, 1))
-		atts := pool.PendingPayloadAttestations()
-		assert.Equal(t, 2, len(atts))
-	})
-
-	t.Run("filter by slot", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		msg1 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		msg2 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 1,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              2,
-				PayloadPresent:    false,
-				BlobDataAvailable: true,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg1, 0))
-		require.NoError(t, pool.InsertPayloadAttestation(msg2, 1))
-
-		atts := pool.PendingPayloadAttestations(primitives.Slot(1))
-		assert.Equal(t, 1, len(atts))
-		assert.Equal(t, primitives.Slot(1), atts[0].Data.Slot)
-
-		atts = pool.PendingPayloadAttestations(primitives.Slot(2))
-		assert.Equal(t, 1, len(atts))
-		assert.Equal(t, primitives.Slot(2), atts[0].Data.Slot)
-
-		atts = pool.PendingPayloadAttestations(primitives.Slot(99))
-		assert.Equal(t, 0, len(atts))
-	})
-}
-
-func TestPool_InsertPayloadAttestation(t *testing.T) {
-	t.Run("nil message", func(t *testing.T) {
-		pool := NewPool()
-		err := pool.InsertPayloadAttestation(nil, 0)
-		require.ErrorContains(t, "nil payload attestation message", err)
-	})
-
-	t.Run("nil data", func(t *testing.T) {
-		pool := NewPool()
-		err := pool.InsertPayloadAttestation(&ethpb.PayloadAttestationMessage{}, 0)
-		require.ErrorContains(t, "nil payload attestation message", err)
-	})
-
-	t.Run("insert creates new entry with correct aggregation bit", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		idx := uint64(5)
-		msg := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg, idx))
-		atts := pool.PendingPayloadAttestations()
-		require.Equal(t, 1, len(atts))
-		assert.Equal(t, true, atts[0].AggregationBits.BitAt(idx))
-		assert.Equal(t, false, atts[0].AggregationBits.BitAt(idx+1))
-	})
-
-	t.Run("duplicate index is no-op", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		idx := uint64(3)
-		msg := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg, idx))
-		firstSig := bytesutil.SafeCopyBytes(pool.PendingPayloadAttestations()[0].Signature)
-
-		require.NoError(t, pool.InsertPayloadAttestation(msg, idx))
-		atts := pool.PendingPayloadAttestations()
-		require.Equal(t, 1, len(atts))
-		assert.DeepEqual(t, firstSig, atts[0].Signature)
-	})
-
-	t.Run("aggregates different indices", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		root := make([]byte, 32)
-		root[0] = 'r'
-		data := &ethpb.PayloadAttestationData{
-			BeaconBlockRoot:   root,
-			Slot:              1,
-			PayloadPresent:    true,
-			BlobDataAvailable: false,
-		}
-		msg1 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data:           data,
-			Signature:      sig,
-		}
-		msg2 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 1,
-			Data:           data,
-			Signature:      sig,
-		}
-
-		require.NoError(t, pool.InsertPayloadAttestation(msg1, 5))
-		require.NoError(t, pool.InsertPayloadAttestation(msg2, 7))
-
-		atts := pool.PendingPayloadAttestations()
-		require.Equal(t, 1, len(atts))
-		assert.Equal(t, true, atts[0].AggregationBits.BitAt(5))
-		assert.Equal(t, true, atts[0].AggregationBits.BitAt(7))
-		assert.Equal(t, false, atts[0].AggregationBits.BitAt(6))
-	})
-
-	t.Run("different data creates separate entries", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		msg1 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		msg2 := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 1,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    false, // different
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg1, 0))
-		require.NoError(t, pool.InsertPayloadAttestation(msg2, 1))
-		atts := pool.PendingPayloadAttestations()
-		assert.Equal(t, 2, len(atts))
-	})
-}
-
-func TestPool_Seen(t *testing.T) {
-	pool := NewPool()
-	sig := bls.NewAggregateSignature().Marshal()
-	data := &ethpb.PayloadAttestationData{
-		BeaconBlockRoot:   make([]byte, 32),
-		Slot:              1,
-		PayloadPresent:    true,
-		BlobDataAvailable: false,
-	}
-
-	assert.Equal(t, false, pool.Seen(data, 5))
-
-	msg := &ethpb.PayloadAttestationMessage{
-		ValidatorIndex: 0,
-		Data:           data,
-		Signature:      sig,
-	}
-	require.NoError(t, pool.InsertPayloadAttestation(msg, 5))
-
-	assert.Equal(t, true, pool.Seen(data, 5))
-	assert.Equal(t, false, pool.Seen(data, 6))
-	assert.Equal(t, false, pool.Seen(nil, 5))
-}
-
-func TestPool_MarkIncluded(t *testing.T) {
-	t.Run("mark included removes from pool", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		msg := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg, 0))
-		assert.Equal(t, 1, len(pool.PendingPayloadAttestations()))
-
-		pool.MarkIncluded(&ethpb.PayloadAttestation{
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-		})
-		assert.Equal(t, 0, len(pool.PendingPayloadAttestations()))
-	})
-
-	t.Run("mark included with non-matching data does nothing", func(t *testing.T) {
-		pool := NewPool()
-		sig := bls.NewAggregateSignature().Marshal()
-		msg := &ethpb.PayloadAttestationMessage{
-			ValidatorIndex: 0,
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              1,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-			Signature: sig,
-		}
-		require.NoError(t, pool.InsertPayloadAttestation(msg, 0))
-
-		pool.MarkIncluded(&ethpb.PayloadAttestation{
-			Data: &ethpb.PayloadAttestationData{
-				BeaconBlockRoot:   make([]byte, 32),
-				Slot:              999,
-				PayloadPresent:    true,
-				BlobDataAvailable: false,
-			},
-		})
-		assert.Equal(t, 1, len(pool.PendingPayloadAttestations()))
-	})
-
-	t.Run("mark included with nil is safe", func(t *testing.T) {
-		pool := NewPool()
-		pool.MarkIncluded(nil)
-		pool.MarkIncluded(&ethpb.PayloadAttestation{})
-	})
-}

validator/client/grpc-api/grpc_validator_client.go

@@ -285,10 +285,10 @@ func (c *grpcValidatorClient) StartEventStream(ctx context.Context, topics []str
 	ctx, span := trace.StartSpan(ctx, "validator.gRPCClient.StartEventStream")
 	defer span.End()
 	if len(topics) == 0 {
-		eventsChannel <- &eventClient.Event{
+		eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 			EventType: eventClient.EventError,
 			Data:      []byte(errors.New("no topics were added").Error()),
-		}
+		})
 		return
 	}
 	// TODO(13563): ONLY WORKS WITH HEAD TOPIC.
@@ -299,10 +299,10 @@ func (c *grpcValidatorClient) StartEventStream(ctx context.Context, topics []str
 		}
 	}
 	if !containsHead {
-		eventsChannel <- &eventClient.Event{
+		eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 			EventType: eventClient.EventConnectionError,
 			Data:      []byte(errors.Wrap(client.ErrConnectionIssue, "gRPC only supports the head topic, and head topic was not passed").Error()),
-		}
+		})
 	}
 	if containsHead && len(topics) > 1 {
 		log.Warn("gRPC only supports the head topic, other topics will be ignored")
@@ -310,10 +310,10 @@ func (c *grpcValidatorClient) StartEventStream(ctx context.Context, topics []str
 
 	stream, err := c.beaconNodeValidatorClient.StreamSlots(ctx, &ethpb.StreamSlotsRequest{VerifiedOnly: true})
 	if err != nil {
-		eventsChannel <- &eventClient.Event{
+		eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 			EventType: eventClient.EventConnectionError,
 			Data:      []byte(errors.Wrap(client.ErrConnectionIssue, err.Error()).Error()),
-		}
+		})
 		return
 	}
 	c.isEventStreamRunning = true
@@ -327,25 +327,25 @@ func (c *grpcValidatorClient) StartEventStream(ctx context.Context, topics []str
 			if ctx.Err() != nil {
 				c.isEventStreamRunning = false
 				if errors.Is(ctx.Err(), context.Canceled) {
-					eventsChannel <- &eventClient.Event{
+					eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 						EventType: eventClient.EventConnectionError,
 						Data:      []byte(errors.Wrap(client.ErrConnectionIssue, ctx.Err().Error()).Error()),
-					}
+					})
 					return
 				}
-				eventsChannel <- &eventClient.Event{
+				eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 					EventType: eventClient.EventError,
 					Data:      []byte(ctx.Err().Error()),
-				}
+				})
 				return
 			}
 			res, err := stream.Recv()
 			if err != nil {
 				c.isEventStreamRunning = false
-				eventsChannel <- &eventClient.Event{
+				eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 					EventType: eventClient.EventConnectionError,
 					Data:      []byte(errors.Wrap(client.ErrConnectionIssue, err.Error()).Error()),
-				}
+				})
 				return
 			}
 			if res == nil {
@@ -357,15 +357,15 @@ func (c *grpcValidatorClient) StartEventStream(ctx context.Context, topics []str
 				CurrentDutyDependentRoot:  hexutil.Encode(res.CurrentDutyDependentRoot),
 			})
 			if err != nil {
-				eventsChannel <- &eventClient.Event{
+				eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 					EventType: eventClient.EventError,
 					Data:      []byte(errors.Wrap(err, "failed to marshal Head Event").Error()),
-				}
+				})
 			}
-			eventsChannel <- &eventClient.Event{
+			eventClient.PublishEvent(eventsChannel, &eventClient.Event{
 				EventType: eventClient.EventHead,
 				Data:      b,
-			}
+			})
 		}
 	}
 }

validator/client/grpc-api/grpc_validator_client_test.go

@@ -223,7 +223,7 @@ func TestStartEventStream(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			eventsChannel := make(chan *eventClient.Event, 1) // Buffer to prevent blocking
+			eventsChannel := make(chan *eventClient.Event, 4) // Buffer to prevent blocking
 			tc.prepare()                                      // Setup mock expectations
 
 			go grpcClient.StartEventStream(ctx, tc.topics, eventsChannel)

validator/client/runner_test.go

@@ -441,7 +441,6 @@ func TestRunnerPushesProposerSettings_ValidContext(t *testing.T) {
 		defer assertValidContext(t, timedCtx, ctx)
 		delay(t)
 	})
-	vcm.EXPECT().EventStreamIsRunning().Return(true).AnyTimes().Do(func() { delay(t) })
 	vcm.EXPECT().SubmitValidatorRegistrations(liveCtx, gomock.Any()).Do(func(ctx context.Context, _ any) {
 		defer assertValidContext(t, timedCtx, ctx) // This is the specific regression test assertion for PR 15369.
 		delay(t)

validator/client/service.go

@@ -66,6 +66,8 @@ type ValidatorService struct {
 	closeClientFunc         func() // validator client stop function is used here
 }
 
+const eventChannelBufferSize = 32
+
 // Config for the validator service.
 type Config struct {
 	Validator               iface.Validator
@@ -234,7 +236,7 @@ func (v *ValidatorService) Start() {
 		distributed:                    v.distributed,
 		disableDutiesPolling:           v.disableDutiesPolling,
 		accountsChangedChannel:         make(chan [][fieldparams.BLSPubkeyLength]byte, 1),
-		eventsChannel:                  make(chan *eventClient.Event, 1),
+		eventsChannel:                  make(chan *eventClient.Event, eventChannelBufferSize),
 	}
 
 	hm := newHealthMonitor(v.ctx, v.cancel, v.maxHealthChecks, v.validator)

validator/client/validator.go

@@ -64,6 +64,11 @@ var (
 	msgNoKeysFetched     = "No validating keys fetched. Waiting for keys..."
 )
 
+const (
+	eventStreamStopped uint32 = iota
+	eventStreamRunning
+)
+
 type validator struct {
 	logValidatorPerformance            bool
 	distributed                        bool
@@ -82,6 +87,7 @@ type validator struct {
 	cachedAttestationData              *ethpb.AttestationData
 	accountsChangedChannel             chan [][fieldparams.BLSPubkeyLength]byte
 	eventsChannel                      chan *eventClient.Event
+	eventStreamState                   atomic.Uint32
 	highestValidSlot                   primitives.Slot
 	submittedAggregates                map[submittedAttKey]*submittedAtt
 	graffitiStruct                     *graffiti.Graffiti
@@ -1211,12 +1217,40 @@ func (v *validator) PushProposerSettings(ctx context.Context, slot primitives.Sl
 }
 
 func (v *validator) StartEventStream(ctx context.Context, topics []string) {
-	if v.EventStreamIsRunning() {
+	if !v.eventStreamState.CompareAndSwap(eventStreamStopped, eventStreamRunning) {
 		log.Debug("EventStream is already running")
 		return
 	}
 	log.WithField("topics", topics).Info("Starting event stream")
-	v.validatorClient.StartEventStream(ctx, topics, v.eventsChannel)
+	go v.runEventStream(ctx, topics)
+}
+
+func (v *validator) runEventStream(ctx context.Context, topics []string) {
+	defer v.eventStreamState.Store(eventStreamStopped)
+	backoff := time.Second
+	const maxBackoff = 30 * time.Second
+
+	for {
+		v.validatorClient.StartEventStream(ctx, topics, v.eventsChannel)
+		if ctx.Err() != nil {
+			return
+		}
+
+		log.WithField("retryIn", backoff).Warn("Event stream ended unexpectedly, attempting to resubscribe")
+		timer := time.NewTimer(backoff)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+		if backoff < maxBackoff {
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
+	}
 }
 
 func (v *validator) checkDependentRoots(ctx context.Context, head *structs.HeadEvent) error {
@@ -1303,7 +1337,7 @@ func (v *validator) ProcessEvent(ctx context.Context, event *eventClient.Event)
 }
 
 func (v *validator) EventStreamIsRunning() bool {
-	return v.validatorClient.EventStreamIsRunning()
+	return v.eventStreamState.Load() == eventStreamRunning
 }
 
 func (v *validator) Host() string {