Enable Light Client req/resp domain (#15281)

* add rpc toppic mappings and types

* placeholder funcs

* bootstrap write chunk

* add rpc toppic mappings and types

* placeholder funcs

* bootstrap write chunk

* deps

* add messageMapping entries

* add handlers and register RPC

* deps

* tests

* read context in tests

* add tests

* add flag and changelog entry

* fix linter

* deps

* increase topic count in ratelimiter test

* handle flag

beacon-chain/p2p/BUILD.bazel

@@ -55,6 +55,7 @@ go_library(
         "//beacon-chain/startup:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
         "//config/features:go_default_library",
+        "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/interfaces:go_default_library",
         "//consensus-types/primitives:go_default_library",

beacon-chain/p2p/rpc_topic_mappings.go

@@ -4,6 +4,7 @@ import (
 	"reflect"
 
 	p2ptypes "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
+	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	pb "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
@@ -43,6 +44,18 @@ const BlobSidecarsByRangeName = "/blob_sidecars_by_range"
 // BlobSidecarsByRootName is the name for the BlobSidecarsByRoot v1 message topic.
 const BlobSidecarsByRootName = "/blob_sidecars_by_root"
 
+// LightClientBootstrapName is the name for the LightClientBootstrap message topic,
+const LightClientBootstrapName = "/light_client_bootstrap"
+
+// LightClientUpdatesByRangeName is the name for the LightClientUpdatesByRange topic.
+const LightClientUpdatesByRangeName = "/light_client_updates_by_range"
+
+// LightClientFinalityUpdateName is the name for the LightClientFinalityUpdate topic.
+const LightClientFinalityUpdateName = "/light_client_finality_update"
+
+// LightClientOptimisticUpdateName is the name for the LightClientOptimisticUpdate topic.
+const LightClientOptimisticUpdateName = "/light_client_optimistic_update"
+
 const (
 	// V1 RPC Topics
 	// RPCStatusTopicV1 defines the v1 topic for the status rpc method.
@@ -66,6 +79,15 @@ const (
 	// /eth2/beacon_chain/req/blob_sidecars_by_root/1/
 	RPCBlobSidecarsByRootTopicV1 = protocolPrefix + BlobSidecarsByRootName + SchemaVersionV1
 
+	// RPCLightClientBootstrapTopicV1 is a topic for requesting a light client bootstrap.
+	RPCLightClientBootstrapTopicV1 = protocolPrefix + LightClientBootstrapName + SchemaVersionV1
+	// RPCLightClientUpdatesByRangeTopicV1 is a topic for requesting light client updates by range.
+	RPCLightClientUpdatesByRangeTopicV1 = protocolPrefix + LightClientUpdatesByRangeName + SchemaVersionV1
+	// RPCLightClientFinalityUpdateTopicV1 is a topic for requesting a light client finality update.
+	RPCLightClientFinalityUpdateTopicV1 = protocolPrefix + LightClientFinalityUpdateName + SchemaVersionV1
+	// RPCLightClientOptimisticUpdateTopicV1 is a topic for requesting a light client Optimistic update.
+	RPCLightClientOptimisticUpdateTopicV1 = protocolPrefix + LightClientOptimisticUpdateName + SchemaVersionV1
+
 	// V2 RPC Topics
 	// RPCBlocksByRangeTopicV2 defines v2 the topic for the blocks by range rpc method.
 	RPCBlocksByRangeTopicV2 = protocolPrefix + BeaconBlocksByRangeMessageName + SchemaVersionV2
@@ -101,6 +123,12 @@ var RPCTopicMappings = map[string]interface{}{
 	RPCBlobSidecarsByRangeTopicV1: new(pb.BlobSidecarsByRangeRequest),
 	// BlobSidecarsByRoot v1 Message
 	RPCBlobSidecarsByRootTopicV1: new(p2ptypes.BlobSidecarsByRootReq),
+
+	// Light client
+	RPCLightClientBootstrapTopicV1:        new([fieldparams.RootLength]byte),
+	RPCLightClientUpdatesByRangeTopicV1:   new(pb.LightClientUpdatesByRangeRequest),
+	RPCLightClientFinalityUpdateTopicV1:   new(interface{}),
+	RPCLightClientOptimisticUpdateTopicV1: new(interface{}),
 }
 
 // Maps all registered protocol prefixes.
@@ -111,14 +139,18 @@ var protocolMapping = map[string]bool{
 // Maps all the protocol message names for the different rpc
 // topics.
 var messageMapping = map[string]bool{
-	StatusMessageName:              true,
-	GoodbyeMessageName:             true,
-	BeaconBlocksByRangeMessageName: true,
-	BeaconBlocksByRootsMessageName: true,
-	PingMessageName:                true,
-	MetadataMessageName:            true,
-	BlobSidecarsByRangeName:        true,
-	BlobSidecarsByRootName:         true,
+	StatusMessageName:               true,
+	GoodbyeMessageName:              true,
+	BeaconBlocksByRangeMessageName:  true,
+	BeaconBlocksByRootsMessageName:  true,
+	PingMessageName:                 true,
+	MetadataMessageName:             true,
+	BlobSidecarsByRangeName:         true,
+	BlobSidecarsByRootName:          true,
+	LightClientBootstrapName:        true,
+	LightClientUpdatesByRangeName:   true,
+	LightClientFinalityUpdateName:   true,
+	LightClientOptimisticUpdateName: true,
 }
 
 // Maps all the RPC messages which are to updated in altair.

beacon-chain/sync/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "rpc_blob_sidecars_by_root.go",
         "rpc_chunked_response.go",
         "rpc_goodbye.go",
+        "rpc_light_client.go",
         "rpc_metadata.go",
         "rpc_ping.go",
         "rpc_send_request.go",
@@ -114,6 +115,7 @@ go_library(
         "//encoding/bytesutil:go_default_library",
         "//encoding/ssz/equality:go_default_library",
         "//io/file:go_default_library",
+        "//math:go_default_library",
         "//monitoring/tracing:go_default_library",
         "//monitoring/tracing/trace:go_default_library",
         "//network/forks:go_default_library",
@@ -168,6 +170,7 @@ go_test(
         "rpc_blob_sidecars_by_root_test.go",
         "rpc_goodbye_test.go",
         "rpc_handler_test.go",
+        "rpc_light_client_test.go",
         "rpc_metadata_test.go",
         "rpc_ping_test.go",
         "rpc_send_request_test.go",
@@ -231,6 +234,7 @@ go_test(
         "//beacon-chain/verification:go_default_library",
         "//cache/lru:go_default_library",
         "//cmd/beacon-chain/flags:go_default_library",
+        "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",

beacon-chain/sync/rate_limiter.go

@@ -80,6 +80,12 @@ func newRateLimiter(p2pProvider p2p.P2P) *limiter {
 	// BlobSidecarsByRangeV1
 	topicMap[addEncoding(p2p.RPCBlobSidecarsByRangeTopicV1)] = blobCollector
 
+	// Light client requests
+	topicMap[addEncoding(p2p.RPCLightClientBootstrapTopicV1)] = leakybucket.NewCollector(1, defaultBurstLimit, leakyBucketPeriod, false /* deleteEmptyBuckets */)
+	topicMap[addEncoding(p2p.RPCLightClientUpdatesByRangeTopicV1)] = leakybucket.NewCollector(1, defaultBurstLimit, leakyBucketPeriod, false /* deleteEmptyBuckets */)
+	topicMap[addEncoding(p2p.RPCLightClientOptimisticUpdateTopicV1)] = leakybucket.NewCollector(1, defaultBurstLimit, leakyBucketPeriod, false /* deleteEmptyBuckets */)
+	topicMap[addEncoding(p2p.RPCLightClientFinalityUpdateTopicV1)] = leakybucket.NewCollector(1, defaultBurstLimit, leakyBucketPeriod, false /* deleteEmptyBuckets */)
+
 	// General topic for all rpc requests.
 	topicMap[rpcLimiterTopic] = leakybucket.NewCollector(5, defaultBurstLimit*2, leakyBucketPeriod, false /* deleteEmptyBuckets */)
 

beacon-chain/sync/rate_limiter_test.go

@@ -18,7 +18,7 @@ import (
 
 func TestNewRateLimiter(t *testing.T) {
 	rlimiter := newRateLimiter(mockp2p.NewTestP2P(t))
-	assert.Equal(t, len(rlimiter.limiterMap), 12, "correct number of topics not registered")
+	assert.Equal(t, len(rlimiter.limiterMap), 16, "correct number of topics not registered")
 }
 
 func TestNewRateLimiter_FreeCorrectly(t *testing.T) {

beacon-chain/sync/rpc.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
 	p2ptypes "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
+	"github.com/OffchainLabs/prysm/v6/config/features"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v6/monitoring/tracing"
@@ -70,14 +71,23 @@ func (s *Service) rpcHandlerByTopicFromFork(forkIndex int) (map[string]rpcHandle
 	// Bellatrix: https://github.com/ethereum/consensus-specs/blob/dev/specs/bellatrix/p2p-interface.md#messages
 	// Altair: https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/p2p-interface.md#messages
 	if forkIndex >= version.Altair {
-		return map[string]rpcHandler{
+		handler := map[string]rpcHandler{
 			p2p.RPCStatusTopicV1:        s.statusRPCHandler,
 			p2p.RPCGoodByeTopicV1:       s.goodbyeRPCHandler,
 			p2p.RPCBlocksByRangeTopicV2: s.beaconBlocksByRangeRPCHandler, // Updated in Altair and modified in Capella
 			p2p.RPCBlocksByRootTopicV2:  s.beaconBlocksRootRPCHandler,    // Updated in Altair and modified in Capella
 			p2p.RPCPingTopicV1:          s.pingHandler,
 			p2p.RPCMetaDataTopicV2:      s.metaDataHandler, // Updated in Altair
-		}, nil
+		}
+
+		if features.Get().EnableLightClient {
+			handler[p2p.RPCLightClientBootstrapTopicV1] = s.lightClientBootstrapRPCHandler
+			handler[p2p.RPCLightClientUpdatesByRangeTopicV1] = s.lightClientUpdatesByRangeRPCHandler
+			handler[p2p.RPCLightClientFinalityUpdateTopicV1] = s.lightClientFinalityUpdateRPCHandler
+			handler[p2p.RPCLightClientOptimisticUpdateTopicV1] = s.lightClientOptimisticUpdateRPCHandler
+		}
+
+		return handler, nil
 	}
 
 	// PhaseO: https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/p2p-interface.md#messages
@@ -246,9 +256,11 @@ func (s *Service) registerRPC(baseTopic string, handle rpcHandler) {
 		// Increment message received counter.
 		messageReceivedCounter.WithLabelValues(topic).Inc()
 
-		// since metadata requests do not have any data in the payload, we
+		// since some requests do not have any data in the payload, we
 		// do not decode anything.
-		if baseTopic == p2p.RPCMetaDataTopicV1 || baseTopic == p2p.RPCMetaDataTopicV2 {
+		if baseTopic == p2p.RPCMetaDataTopicV1 || baseTopic == p2p.RPCMetaDataTopicV2 ||
+			baseTopic == p2p.RPCLightClientOptimisticUpdateTopicV1 ||
+			baseTopic == p2p.RPCLightClientFinalityUpdateTopicV1 {
 			if err := handle(ctx, base, stream); err != nil {
 				messageFailedProcessingCounter.WithLabelValues(topic).Inc()
 				if !errors.Is(err, p2ptypes.ErrWrongForkDigestVersion) {

beacon-chain/sync/rpc_chunked_response.go

@@ -161,3 +161,80 @@ func WriteBlobSidecarChunk(stream libp2pcore.Stream, tor blockchain.TemporalOrac
 	_, err = encoding.EncodeWithMaxLength(stream, sidecar)
 	return err
 }
+
+func WriteLightClientBootstrapChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, bootstrap interfaces.LightClientBootstrap) error {
+	if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
+		return err
+	}
+
+	valRoot := tor.GenesisValidatorsRoot()
+	digest, err := forks.ForkDigestFromEpoch(slots.ToEpoch(bootstrap.Header().Beacon().Slot), valRoot[:])
+	if err != nil {
+		return err
+	}
+
+	obtainedCtx := digest[:]
+	if err = writeContextToStream(obtainedCtx, stream); err != nil {
+		return err
+	}
+
+	_, err = encoding.EncodeWithMaxLength(stream, bootstrap)
+	return err
+}
+
+func WriteLightClientUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientUpdate) error {
+	if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
+		return err
+	}
+
+	valRoot := tor.GenesisValidatorsRoot()
+	digest, err := forks.ForkDigestFromEpoch(slots.ToEpoch(update.AttestedHeader().Beacon().Slot), valRoot[:])
+	if err != nil {
+		return err
+	}
+	obtainedCtx := digest[:]
+
+	if err = writeContextToStream(obtainedCtx, stream); err != nil {
+		return err
+	}
+	_, err = encoding.EncodeWithMaxLength(stream, update)
+	return err
+}
+
+func WriteLightClientOptimisticUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientOptimisticUpdate) error {
+	if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
+		return err
+	}
+
+	valRoot := tor.GenesisValidatorsRoot()
+	digest, err := forks.ForkDigestFromEpoch(slots.ToEpoch(update.AttestedHeader().Beacon().Slot), valRoot[:])
+	if err != nil {
+		return err
+	}
+	obtainedCtx := digest[:]
+
+	if err = writeContextToStream(obtainedCtx, stream); err != nil {
+		return err
+	}
+	_, err = encoding.EncodeWithMaxLength(stream, update)
+	return err
+}
+
+func WriteLightClientFinalityUpdateChunk(stream libp2pcore.Stream, tor blockchain.TemporalOracle, encoding encoder.NetworkEncoding, update interfaces.LightClientFinalityUpdate) error {
+	if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
+		return err
+	}
+
+	valRoot := tor.GenesisValidatorsRoot()
+	digest, err := forks.ForkDigestFromEpoch(slots.ToEpoch(update.AttestedHeader().Beacon().Slot), valRoot[:])
+	if err != nil {
+		return err
+	}
+	obtainedCtx := digest[:]
+
+	if err = writeContextToStream(obtainedCtx, stream); err != nil {
+		return err
+	}
+	_, err = encoding.EncodeWithMaxLength(stream, update)
+	return err
+}

beacon-chain/sync/rpc_light_client.go

@@ -0,0 +1,222 @@
+package sync
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/types"
+	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
+	"github.com/OffchainLabs/prysm/v6/config/params"
+	"github.com/OffchainLabs/prysm/v6/math"
+	"github.com/OffchainLabs/prysm/v6/monitoring/tracing"
+	"github.com/OffchainLabs/prysm/v6/monitoring/tracing/trace"
+	eth "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
+	libp2pcore "github.com/libp2p/go-libp2p/core"
+)
+
+// lightClientBootstrapRPCHandler handles the /eth2/beacon_chain/req/light_client_bootstrap/1/ RPC request.
+func (s *Service) lightClientBootstrapRPCHandler(ctx context.Context, msg interface{}, stream libp2pcore.Stream) error {
+	ctx, span := trace.StartSpan(ctx, "sync.lightClientBootstrapRPCHandler")
+	defer span.End()
+	ctx, cancel := context.WithTimeout(ctx, ttfbTimeout)
+	defer cancel()
+
+	logger := log.WithField("handler", p2p.LightClientBootstrapName[1:])
+
+	SetRPCStreamDeadlines(stream)
+	if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
+		logger.WithError(err).Error("s.rateLimiter.validateRequest")
+		return err
+	}
+	s.rateLimiter.add(stream, 1)
+
+	rawMsg, ok := msg.(*[fieldparams.RootLength]byte)
+	if !ok {
+		logger.Error("Message is not *types.LightClientBootstrapReq")
+		return fmt.Errorf("message is not type %T", &[fieldparams.RootLength]byte{})
+	}
+	blkRoot := *rawMsg
+
+	bootstrap, err := s.cfg.beaconDB.LightClientBootstrap(ctx, blkRoot[:])
+	if err != nil {
+		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("s.cfg.beaconDB.LightClientBootstrap")
+		return err
+	}
+	if bootstrap == nil {
+		s.writeErrorResponseToStream(responseCodeResourceUnavailable, types.ErrResourceUnavailable.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error(fmt.Sprintf("nil bootstrap for root %#x", blkRoot))
+		return err
+	}
+
+	SetStreamWriteDeadline(stream, defaultWriteDuration)
+	if err = WriteLightClientBootstrapChunk(stream, s.cfg.clock, s.cfg.p2p.Encoding(), bootstrap); err != nil {
+		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("WriteLightClientBootstrapChunk")
+		return err
+	}
+
+	logger.Info("lightClientBootstrapRPCHandler completed")
+
+	closeStream(stream, logger)
+	return nil
+}
+
+// lightClientUpdatesByRangeRPCHandler handles the /eth2/beacon_chain/req/light_client_updates_by_range/1/ RPC request.
+func (s *Service) lightClientUpdatesByRangeRPCHandler(ctx context.Context, msg interface{}, stream libp2pcore.Stream) error {
+	ctx, span := trace.StartSpan(ctx, "sync.lightClientUpdatesByRangeRPCHandler")
+	defer span.End()
+	ctx, cancel := context.WithTimeout(ctx, ttfbTimeout)
+	defer cancel()
+
+	logger := log.WithField("handler", p2p.LightClientUpdatesByRangeName[1:])
+
+	SetRPCStreamDeadlines(stream)
+	if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
+		logger.WithError(err).Error("s.rateLimiter.validateRequest")
+		return err
+	}
+	s.rateLimiter.add(stream, 1)
+
+	r, ok := msg.(*eth.LightClientUpdatesByRangeRequest)
+	if !ok {
+		logger.Error("Message is not *eth.LightClientUpdatesByRangeReq")
+		return fmt.Errorf("message is not type %T", &eth.LightClientUpdatesByRangeRequest{})
+	}
+
+	if r.Count == 0 {
+		s.writeErrorResponseToStream(responseCodeInvalidRequest, "count is 0", stream)
+		s.cfg.p2p.Peers().Scorers().BadResponsesScorer().Increment(stream.Conn().RemotePeer())
+		logger.Error("Count is 0")
+		return nil
+	}
+
+	if r.Count > params.BeaconConfig().MaxRequestLightClientUpdates {
+		r.Count = params.BeaconConfig().MaxRequestLightClientUpdates
+	}
+
+	endPeriod, err := math.Add64(r.StartPeriod, r.Count-1)
+	if err != nil {
+		s.writeErrorResponseToStream(responseCodeInvalidRequest, err.Error(), stream)
+		s.cfg.p2p.Peers().Scorers().BadResponsesScorer().Increment(stream.Conn().RemotePeer())
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("End period overflows")
+		return err
+	}
+
+	logger.Infof("LC: requesting updates by range (StartPeriod: %d, EndPeriod: %d)", r.StartPeriod, r.StartPeriod+r.Count-1)
+
+	updates, err := s.cfg.beaconDB.LightClientUpdates(ctx, r.StartPeriod, endPeriod)
+	if err != nil {
+		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("s.cfg.beaconDB.LightClientUpdates")
+		return err
+	}
+
+	if len(updates) == 0 {
+		s.writeErrorResponseToStream(responseCodeResourceUnavailable, types.ErrResourceUnavailable.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.Debugf("No update available for start period %d", r.StartPeriod)
+		return nil
+	}
+
+	for i := r.StartPeriod; i <= endPeriod; i++ {
+		u, ok := updates[i]
+		if !ok {
+			break
+		}
+
+		SetStreamWriteDeadline(stream, defaultWriteDuration)
+		if err = WriteLightClientUpdateChunk(stream, s.cfg.clock, s.cfg.p2p.Encoding(), u); err != nil {
+			s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+			tracing.AnnotateError(span, err)
+			logger.WithError(err).Error("WriteLightClientUpdateChunk")
+			return err
+		}
+		s.rateLimiter.add(stream, 1)
+	}
+
+	logger.Info("lightClientUpdatesByRangeRPCHandler completed")
+
+	closeStream(stream, logger)
+	return nil
+}
+
+// lightClientFinalityUpdateRPCHandler handles the /eth2/beacon_chain/req/light_client_finality_update/1/ RPC request.
+func (s *Service) lightClientFinalityUpdateRPCHandler(ctx context.Context, _ interface{}, stream libp2pcore.Stream) error {
+	ctx, span := trace.StartSpan(ctx, "sync.lightClientFinalityUpdateRPCHandler")
+	defer span.End()
+	_, cancel := context.WithTimeout(ctx, ttfbTimeout)
+	defer cancel()
+
+	logger := log.WithField("handler", p2p.LightClientFinalityUpdateName[1:])
+
+	SetRPCStreamDeadlines(stream)
+	if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
+		logger.WithError(err).Error("s.rateLimiter.validateRequest")
+		return err
+	}
+	s.rateLimiter.add(stream, 1)
+
+	if s.lcStore.LastFinalityUpdate() == nil {
+		s.writeErrorResponseToStream(responseCodeResourceUnavailable, types.ErrResourceUnavailable.Error(), stream)
+		logger.Error("No finality update available")
+		return nil
+	}
+
+	SetStreamWriteDeadline(stream, defaultWriteDuration)
+	if err := WriteLightClientFinalityUpdateChunk(stream, s.cfg.clock, s.cfg.p2p.Encoding(), s.lcStore.LastFinalityUpdate()); err != nil {
+		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("WriteLightClientFinalityUpdateChunk")
+		return err
+	}
+
+	logger.Info("lightClientFinalityUpdateRPCHandler completed")
+
+	closeStream(stream, logger)
+	return nil
+}
+
+// lightClientOptimisticUpdateRPCHandler handles the /eth2/beacon_chain/req/light_client_optimistic_update/1/ RPC request.
+func (s *Service) lightClientOptimisticUpdateRPCHandler(ctx context.Context, _ interface{}, stream libp2pcore.Stream) error {
+	ctx, span := trace.StartSpan(ctx, "sync.lightClientOptimisticUpdateRPCHandler")
+	defer span.End()
+	_, cancel := context.WithTimeout(ctx, ttfbTimeout)
+	defer cancel()
+
+	logger := log.WithField("handler", p2p.LightClientOptimisticUpdateName[1:])
+
+	logger.Info("lightClientOptimisticUpdateRPCHandler invoked")
+
+	SetRPCStreamDeadlines(stream)
+	if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
+		logger.WithError(err).Error("s.rateLimiter.validateRequest")
+		return err
+	}
+	s.rateLimiter.add(stream, 1)
+
+	if s.lcStore.LastOptimisticUpdate() == nil {
+		s.writeErrorResponseToStream(responseCodeResourceUnavailable, types.ErrResourceUnavailable.Error(), stream)
+		logger.Error("No optimistic update available")
+		return nil
+	}
+
+	SetStreamWriteDeadline(stream, defaultWriteDuration)
+	if err := WriteLightClientOptimisticUpdateChunk(stream, s.cfg.clock, s.cfg.p2p.Encoding(), s.lcStore.LastOptimisticUpdate()); err != nil {
+		s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		logger.WithError(err).Error("WriteLightClientOptimisticUpdateChunk")
+		return err
+	}
+
+	logger.Info("lightClientOptimisticUpdateRPCHandler completed")
+
+	closeStream(stream, logger)
+	return nil
+}

beacon-chain/sync/rpc_light_client_test.go

@@ -0,0 +1,521 @@
+package sync
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/OffchainLabs/prysm/v6/async/abool"
+	mockChain "github.com/OffchainLabs/prysm/v6/beacon-chain/blockchain/testing"
+	lightClient "github.com/OffchainLabs/prysm/v6/beacon-chain/core/light-client"
+	db "github.com/OffchainLabs/prysm/v6/beacon-chain/db/testing"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
+	p2ptest "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/testing"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/startup"
+	mockSync "github.com/OffchainLabs/prysm/v6/beacon-chain/sync/initial-sync/testing"
+	"github.com/OffchainLabs/prysm/v6/config/features"
+	"github.com/OffchainLabs/prysm/v6/config/params"
+	leakybucket "github.com/OffchainLabs/prysm/v6/container/leaky-bucket"
+	"github.com/OffchainLabs/prysm/v6/network/forks"
+	pb "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
+	"github.com/OffchainLabs/prysm/v6/runtime/version"
+	"github.com/OffchainLabs/prysm/v6/testing/require"
+	"github.com/OffchainLabs/prysm/v6/testing/util"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/protocol"
+)
+
+func TestRPC_LightClientBootstrap(t *testing.T) {
+	resetFn := features.InitWithReset(&features.Flags{
+		EnableLightClient: true,
+	})
+	defer resetFn()
+
+	ctx := context.Background()
+	p2pService := p2ptest.NewTestP2P(t)
+	p1 := p2ptest.NewTestP2P(t)
+	p2 := p2ptest.NewTestP2P(t)
+	p1.Connect(p2)
+	require.Equal(t, 1, len(p1.BHost.Network().Peers()), "Expected peers to be connected")
+
+	chainService := &mockChain.ChainService{
+		ValidatorsRoot: [32]byte{'A'},
+		Genesis:        time.Unix(time.Now().Unix(), 0),
+	}
+	d := db.SetupDB(t)
+	r := Service{
+		ctx: ctx,
+		cfg: &config{
+			p2p:           p2pService,
+			initialSync:   &mockSync.Sync{IsSyncing: false},
+			chain:         chainService,
+			beaconDB:      d,
+			clock:         startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot),
+			stateNotifier: &mockChain.MockStateNotifier{},
+		},
+		chainStarted: abool.New(),
+		lcStore:      &lightClient.Store{},
+		subHandler:   newSubTopicHandler(),
+		rateLimiter:  newRateLimiter(p1),
+	}
+	pcl := protocol.ID(p2p.RPCLightClientBootstrapTopicV1)
+	topic := string(pcl)
+	r.rateLimiter.limiterMap[topic] = leakybucket.NewCollector(10000, 10000, time.Second, false)
+
+	altairDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().AltairForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	bellatrixDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().BellatrixForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	capellaDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().CapellaForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	denebDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().DenebForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	electraDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().ElectraForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+
+	for i := 1; i <= 5; i++ {
+		t.Run(version.String(i), func(t *testing.T) {
+			l := util.NewTestLightClient(t, i)
+			bootstrap, err := lightClient.NewLightClientBootstrapFromBeaconState(ctx, l.State.Slot(), l.State, l.Block)
+			require.NoError(t, err)
+			blockRoot, err := l.Block.Block().HashTreeRoot()
+			require.NoError(t, err)
+
+			require.NoError(t, r.cfg.beaconDB.SaveLightClientBootstrap(ctx, blockRoot[:], bootstrap))
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+			p2.BHost.SetStreamHandler(pcl, func(stream network.Stream) {
+				defer wg.Done()
+				expectSuccess(t, stream)
+				rpcCtx, err := readContextFromStream(stream)
+				require.NoError(t, err)
+				require.Equal(t, 4, len(rpcCtx))
+
+				var resSSZ []byte
+
+				switch i {
+				case version.Altair:
+					require.DeepSSZEqual(t, altairDigest[:], rpcCtx)
+					var res pb.LightClientBootstrapAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Bellatrix:
+					require.DeepSSZEqual(t, bellatrixDigest[:], rpcCtx)
+					var res pb.LightClientBootstrapAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Capella:
+					require.DeepSSZEqual(t, capellaDigest[:], rpcCtx)
+					var res pb.LightClientBootstrapCapella
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Deneb:
+					require.DeepSSZEqual(t, denebDigest[:], rpcCtx)
+					var res pb.LightClientBootstrapDeneb
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Electra:
+					require.DeepSSZEqual(t, electraDigest[:], rpcCtx)
+					var res pb.LightClientBootstrapElectra
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				default:
+					t.Fatalf("unsupported version %d", i)
+				}
+
+				bootstrapSSZ, err := bootstrap.MarshalSSZ()
+				require.NoError(t, err)
+				require.DeepSSZEqual(t, resSSZ, bootstrapSSZ)
+			})
+
+			stream1, err := p1.BHost.NewStream(context.Background(), p2.BHost.ID(), pcl)
+			require.NoError(t, err)
+			err = r.lightClientBootstrapRPCHandler(ctx, &blockRoot, stream1)
+			require.NoError(t, err)
+
+			if util.WaitTimeout(&wg, 1*time.Second) {
+				t.Fatal("Did not receive stream within 1 sec")
+			}
+		})
+	}
+
+}
+
+func TestRPC_LightClientOptimisticUpdate(t *testing.T) {
+	resetFn := features.InitWithReset(&features.Flags{
+		EnableLightClient: true,
+	})
+	defer resetFn()
+
+	ctx := context.Background()
+	p2pService := p2ptest.NewTestP2P(t)
+	p1 := p2ptest.NewTestP2P(t)
+	p2 := p2ptest.NewTestP2P(t)
+	p1.Connect(p2)
+	require.Equal(t, 1, len(p1.BHost.Network().Peers()), "Expected peers to be connected")
+
+	chainService := &mockChain.ChainService{
+		ValidatorsRoot: [32]byte{'A'},
+		Genesis:        time.Unix(time.Now().Unix(), 0),
+	}
+	d := db.SetupDB(t)
+	r := Service{
+		ctx: ctx,
+		cfg: &config{
+			p2p:           p2pService,
+			initialSync:   &mockSync.Sync{IsSyncing: false},
+			chain:         chainService,
+			beaconDB:      d,
+			clock:         startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot),
+			stateNotifier: &mockChain.MockStateNotifier{},
+		},
+		chainStarted: abool.New(),
+		lcStore:      &lightClient.Store{},
+		subHandler:   newSubTopicHandler(),
+		rateLimiter:  newRateLimiter(p1),
+	}
+	pcl := protocol.ID(p2p.RPCLightClientOptimisticUpdateTopicV1)
+	topic := string(pcl)
+	r.rateLimiter.limiterMap[topic] = leakybucket.NewCollector(10000, 10000, time.Second, false)
+
+	altairDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().AltairForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	bellatrixDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().BellatrixForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	capellaDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().CapellaForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	denebDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().DenebForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	electraDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().ElectraForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+
+	for i := 1; i <= 5; i++ {
+		t.Run(version.String(i), func(t *testing.T) {
+			l := util.NewTestLightClient(t, i)
+
+			update, err := lightClient.NewLightClientOptimisticUpdateFromBeaconState(ctx, l.State.Slot(), l.State, l.Block, l.AttestedState, l.AttestedBlock)
+			require.NoError(t, err)
+
+			r.lcStore.SetLastOptimisticUpdate(update)
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+			p2.BHost.SetStreamHandler(pcl, func(stream network.Stream) {
+				defer wg.Done()
+				expectSuccess(t, stream)
+				var resSSZ []byte
+
+				rpcCtx, err := readContextFromStream(stream)
+				require.NoError(t, err)
+				require.Equal(t, 4, len(rpcCtx))
+
+				switch i {
+				case version.Altair:
+					require.DeepSSZEqual(t, altairDigest[:], rpcCtx)
+					var res pb.LightClientOptimisticUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Bellatrix:
+					require.DeepSSZEqual(t, bellatrixDigest[:], rpcCtx)
+					var res pb.LightClientOptimisticUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Capella:
+					require.DeepSSZEqual(t, capellaDigest[:], rpcCtx)
+					var res pb.LightClientOptimisticUpdateCapella
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Deneb:
+					require.DeepSSZEqual(t, denebDigest[:], rpcCtx)
+					var res pb.LightClientOptimisticUpdateDeneb
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Electra:
+					require.DeepSSZEqual(t, electraDigest[:], rpcCtx)
+					var res pb.LightClientOptimisticUpdateDeneb
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				default:
+					t.Fatalf("unsupported version %d", i)
+				}
+
+				updateSSZ, err := update.MarshalSSZ()
+				require.NoError(t, err)
+				require.DeepSSZEqual(t, resSSZ, updateSSZ)
+			})
+
+			stream1, err := p1.BHost.NewStream(context.Background(), p2.BHost.ID(), pcl)
+			require.NoError(t, err)
+			err = r.lightClientOptimisticUpdateRPCHandler(ctx, nil, stream1)
+			require.NoError(t, err)
+
+			if util.WaitTimeout(&wg, 1*time.Second) {
+				t.Fatal("Did not receive stream within 1 sec")
+			}
+		})
+	}
+}
+
+func TestRPC_LightClientFinalityUpdate(t *testing.T) {
+	resetFn := features.InitWithReset(&features.Flags{
+		EnableLightClient: true,
+	})
+	defer resetFn()
+
+	ctx := context.Background()
+	p2pService := p2ptest.NewTestP2P(t)
+	p1 := p2ptest.NewTestP2P(t)
+	p2 := p2ptest.NewTestP2P(t)
+	p1.Connect(p2)
+	require.Equal(t, 1, len(p1.BHost.Network().Peers()), "Expected peers to be connected")
+
+	chainService := &mockChain.ChainService{
+		ValidatorsRoot: [32]byte{'A'},
+		Genesis:        time.Unix(time.Now().Unix(), 0),
+	}
+	d := db.SetupDB(t)
+	r := Service{
+		ctx: ctx,
+		cfg: &config{
+			p2p:           p2pService,
+			initialSync:   &mockSync.Sync{IsSyncing: false},
+			chain:         chainService,
+			beaconDB:      d,
+			clock:         startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot),
+			stateNotifier: &mockChain.MockStateNotifier{},
+		},
+		chainStarted: abool.New(),
+		lcStore:      &lightClient.Store{},
+		subHandler:   newSubTopicHandler(),
+		rateLimiter:  newRateLimiter(p1),
+	}
+	pcl := protocol.ID(p2p.RPCLightClientFinalityUpdateTopicV1)
+	topic := string(pcl)
+	r.rateLimiter.limiterMap[topic] = leakybucket.NewCollector(10000, 10000, time.Second, false)
+
+	altairDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().AltairForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	bellatrixDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().BellatrixForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	capellaDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().CapellaForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	denebDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().DenebForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	electraDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().ElectraForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+
+	for i := 1; i <= 5; i++ {
+		t.Run(version.String(i), func(t *testing.T) {
+			l := util.NewTestLightClient(t, i)
+
+			update, err := lightClient.NewLightClientFinalityUpdateFromBeaconState(ctx, l.State.Slot(), l.State, l.Block, l.AttestedState, l.AttestedBlock, l.FinalizedBlock)
+			require.NoError(t, err)
+
+			r.lcStore.SetLastFinalityUpdate(update)
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+			p2.BHost.SetStreamHandler(pcl, func(stream network.Stream) {
+				defer wg.Done()
+				expectSuccess(t, stream)
+				var resSSZ []byte
+
+				rpcCtx, err := readContextFromStream(stream)
+				require.NoError(t, err)
+				require.Equal(t, 4, len(rpcCtx))
+
+				switch i {
+				case version.Altair:
+					require.DeepSSZEqual(t, altairDigest[:], rpcCtx)
+					var res pb.LightClientFinalityUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Bellatrix:
+					require.DeepSSZEqual(t, bellatrixDigest[:], rpcCtx)
+					var res pb.LightClientFinalityUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Capella:
+					require.DeepSSZEqual(t, capellaDigest[:], rpcCtx)
+					var res pb.LightClientFinalityUpdateCapella
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Deneb:
+					require.DeepSSZEqual(t, denebDigest[:], rpcCtx)
+					var res pb.LightClientFinalityUpdateDeneb
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Electra:
+					require.DeepSSZEqual(t, electraDigest[:], rpcCtx)
+					var res pb.LightClientFinalityUpdateElectra
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				default:
+					t.Fatalf("unsupported version %d", i)
+				}
+
+				updateSSZ, err := update.MarshalSSZ()
+				require.NoError(t, err)
+				require.DeepSSZEqual(t, resSSZ, updateSSZ)
+			})
+
+			stream1, err := p1.BHost.NewStream(context.Background(), p2.BHost.ID(), pcl)
+			require.NoError(t, err)
+			err = r.lightClientFinalityUpdateRPCHandler(ctx, nil, stream1)
+			require.NoError(t, err)
+
+			if util.WaitTimeout(&wg, 1*time.Second) {
+				t.Fatal("Did not receive stream within 1 sec")
+			}
+		})
+	}
+}
+
+func TestRPC_LightClientUpdatesByRange(t *testing.T) {
+	resetFn := features.InitWithReset(&features.Flags{
+		EnableLightClient: true,
+	})
+	defer resetFn()
+
+	ctx := context.Background()
+	p2pService := p2ptest.NewTestP2P(t)
+	p1 := p2ptest.NewTestP2P(t)
+	p2 := p2ptest.NewTestP2P(t)
+	p1.Connect(p2)
+	require.Equal(t, 1, len(p1.BHost.Network().Peers()), "Expected peers to be connected")
+
+	chainService := &mockChain.ChainService{
+		ValidatorsRoot: [32]byte{'A'},
+		Genesis:        time.Unix(time.Now().Unix(), 0),
+	}
+	d := db.SetupDB(t)
+	r := Service{
+		ctx: ctx,
+		cfg: &config{
+			p2p:           p2pService,
+			initialSync:   &mockSync.Sync{IsSyncing: false},
+			chain:         chainService,
+			beaconDB:      d,
+			clock:         startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot),
+			stateNotifier: &mockChain.MockStateNotifier{},
+		},
+		chainStarted: abool.New(),
+		lcStore:      &lightClient.Store{},
+		subHandler:   newSubTopicHandler(),
+		rateLimiter:  newRateLimiter(p1),
+	}
+	pcl := protocol.ID(p2p.RPCLightClientUpdatesByRangeTopicV1)
+	topic := string(pcl)
+	r.rateLimiter.limiterMap[topic] = leakybucket.NewCollector(10000, 10000, time.Second, false)
+
+	altairDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().AltairForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	bellatrixDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().BellatrixForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	capellaDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().CapellaForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	denebDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().DenebForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+	electraDigest, err := forks.ForkDigestFromEpoch(params.BeaconConfig().ElectraForkEpoch, chainService.ValidatorsRoot[:])
+	require.NoError(t, err)
+
+	for i := 1; i <= 5; i++ {
+		t.Run(version.String(i), func(t *testing.T) {
+			for j := 0; j < 5; j++ {
+				l := util.NewTestLightClient(t, i, util.WithIncreasedAttestedSlot(uint64(j)))
+				update, err := lightClient.NewLightClientUpdateFromBeaconState(ctx, l.State.Slot(), l.State, l.Block, l.AttestedState, l.AttestedBlock, l.FinalizedBlock)
+				require.NoError(t, err)
+				require.NoError(t, r.cfg.beaconDB.SaveLightClientUpdate(ctx, uint64(j), update))
+			}
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+
+			responseCounter := 0
+
+			p2.BHost.SetStreamHandler(pcl, func(stream network.Stream) {
+				defer wg.Done()
+				expectSuccess(t, stream)
+				rpcCtx, err := readContextFromStream(stream)
+				require.NoError(t, err)
+				require.Equal(t, 4, len(rpcCtx))
+
+				var resSSZ []byte
+
+				switch i {
+				case version.Altair:
+					require.DeepSSZEqual(t, altairDigest[:], rpcCtx)
+					var res pb.LightClientUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Bellatrix:
+					require.DeepSSZEqual(t, bellatrixDigest[:], rpcCtx)
+					var res pb.LightClientUpdateAltair
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Capella:
+					require.DeepSSZEqual(t, capellaDigest[:], rpcCtx)
+					var res pb.LightClientUpdateCapella
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Deneb:
+					require.DeepSSZEqual(t, denebDigest[:], rpcCtx)
+					var res pb.LightClientUpdateDeneb
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				case version.Electra:
+					require.DeepSSZEqual(t, electraDigest[:], rpcCtx)
+					var res pb.LightClientUpdateElectra
+					require.NoError(t, r.cfg.p2p.Encoding().DecodeWithMaxLength(stream, &res))
+					resSSZ, err = res.MarshalSSZ()
+					require.NoError(t, err)
+				default:
+					t.Fatalf("unsupported version %d", i)
+				}
+
+				update, err := r.cfg.beaconDB.LightClientUpdates(ctx, 0, 4)
+				require.NoError(t, err)
+				bootstrapSSZ, err := update[uint64(responseCounter)].MarshalSSZ()
+				require.NoError(t, err)
+				require.DeepSSZEqual(t, resSSZ, bootstrapSSZ)
+				responseCounter++
+			})
+
+			stream1, err := p1.BHost.NewStream(context.Background(), p2.BHost.ID(), pcl)
+			require.NoError(t, err)
+
+			msg := pb.LightClientUpdatesByRangeRequest{
+				StartPeriod: 0,
+				Count:       5,
+			}
+			err = r.lightClientUpdatesByRangeRPCHandler(ctx, &msg, stream1)
+			require.NoError(t, err)
+
+			if util.WaitTimeout(&wg, 1*time.Second) {
+				t.Fatal("Did not receive stream within 1 sec")
+			}
+		})
+	}
+
+}