Validate Range Availibility

beacon-chain/db/iface/interface.go

@@ -111,6 +111,7 @@ type HeadAccessDatabase interface {
 
 	// Support for checkpoint sync and backfill.
 	SaveOrigin(ctx context.Context, serState, serBlock []byte) error
+	SaveOriginCheckpointBlockRoot(ctx context.Context, blockRoot [32]byte) error
 	SaveBackfillStatus(context.Context, *dbval.BackfillStatus) error
 	BackfillFinalizedIndex(ctx context.Context, blocks []blocks.ROBlock, finalizedChildRoot [32]byte) error
 }

beacon-chain/p2p/types/rpc_errors.go

@@ -14,4 +14,5 @@ var (
 	ErrInvalidRequest         = errors.New("invalid range, step or count")
 	ErrBlobLTMinRequest       = errors.New("blob slot < minimum_request_epoch")
 	ErrMaxBlobReqExceeded     = errors.New("requested more than MAX_REQUEST_BLOB_SIDECARS")
+	ErrResourceUnavailable    = errors.New("resource requested unavailable")
 )

beacon-chain/sync/BUILD.bazel

@@ -233,6 +233,7 @@ go_test(
         "//encoding/bytesutil:go_default_library",
         "//encoding/ssz/equality:go_default_library",
         "//network/forks:go_default_library",
+        "//proto/dbval:go_default_library",
         "//proto/engine/v1:go_default_library",
         "//proto/eth/v2:go_default_library",
         "//proto/prysm/v1alpha1:go_default_library",

beacon-chain/sync/rpc_beacon_blocks_by_range.go

@@ -6,6 +6,7 @@ import (
 
 	libp2pcore "github.com/libp2p/go-libp2p/core"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v4/beacon-chain/db"
 	p2ptypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/types"
 	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
@@ -38,6 +39,19 @@ func (s *Service) beaconBlocksByRangeRPCHandler(ctx context.Context, msg interfa
 		tracing.AnnotateError(span, err)
 		return err
 	}
+	available, err := s.validateRangeAvailability(ctx, rp)
+	if err != nil {
+		log.WithError(err).Debug("error in validating range availability")
+		s.writeErrorResponseToStream(responseCodeServerError, p2ptypes.ErrGeneric.Error(), stream)
+		tracing.AnnotateError(span, err)
+		return err
+	}
+	if !available {
+		log.Debug("error in validating range availability")
+		s.writeErrorResponseToStream(responseCodeResourceUnavailable, p2ptypes.ErrResourceUnavailable.Error(), stream)
+		tracing.AnnotateError(span, err)
+		return nil
+	}
 
 	blockLimiter, err := s.rateLimiter.topicCollector(string(stream.Protocol()))
 	if err != nil {
@@ -126,6 +140,35 @@ func validateRangeRequest(r *pb.BeaconBlocksByRangeRequest, current primitives.S
 	return rp, nil
 }
 
+func (s *Service) validateRangeAvailability(ctx context.Context, rp rangeParams) (bool, error) {
+	startBlock := rp.start
+	bs, err := s.cfg.beaconDB.BackfillStatus(ctx)
+	if err != nil {
+		// If this is a genesis or legacy node, we retrieve the checkpoint
+		// sync block.
+		if !errors.Is(err, db.ErrNotFound) {
+			return false, err
+		}
+		cpr, err := s.cfg.beaconDB.OriginCheckpointBlockRoot(ctx)
+		if errors.Is(err, db.ErrNotFoundOriginBlockRoot) {
+			// This would be a genesis synced node and
+			// would be able to serve all the required blocks.
+			return true, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		cBlk, err := s.cfg.beaconDB.Block(ctx, cpr)
+		if err != nil {
+			return false, err
+		}
+		// Mark range as unavailable if our checkpoint block has
+		// as slot less than or equal to the start of the request.
+		return cBlk.Block().Slot() <= startBlock, nil
+	}
+	return primitives.Slot(bs.LowSlot) <= startBlock, nil
+}
+
 func (s *Service) writeBlockBatchToStream(ctx context.Context, batch blockBatch, stream libp2pcore.Stream) error {
 	ctx, span := trace.StartSpan(ctx, "sync.WriteBlockRangeToStream")
 	defer span.End()

beacon-chain/sync/rpc_beacon_blocks_by_range_test.go

@@ -21,13 +21,15 @@ import (
 	p2ptest "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/testing"
 	p2ptypes "github.com/prysmaticlabs/prysm/v4/beacon-chain/p2p/types"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/startup"
+	state_native "github.com/prysmaticlabs/prysm/v4/beacon-chain/state/state-native"
 	"github.com/prysmaticlabs/prysm/v4/cmd/beacon-chain/flags"
 	fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	leakybucket "github.com/prysmaticlabs/prysm/v4/container/leaky-bucket"
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
+	"github.com/prysmaticlabs/prysm/v4/proto/dbval"
 	enginev1 "github.com/prysmaticlabs/prysm/v4/proto/engine/v1"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/testing/assert"
@@ -1105,3 +1107,210 @@ func TestRPCBeaconBlocksByRange_FilterBlocks_PreviousRoot(t *testing.T) {
 	// pointer should reference a new root.
 	require.NotEqual(t, cf.prevRoot, [32]byte{})
 }
+
+func TestRPCBeaconBlocksByRange_ValidateRangeAvailibility(t *testing.T) {
+	tests := []struct {
+		name          string
+		rp            rangeParams
+		expectedError error
+		available     bool
+		svcCreator    func(t *testing.T) *Service
+	}{
+		{
+			name: "Single Block from genesis",
+			rp: rangeParams{
+				start: 1,
+				end:   1,
+				size:  1,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 3,
+				})
+				require.NoError(t, err)
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     true,
+		},
+		{
+			name: "Normal Range from genesis",
+			rp: rangeParams{
+				start: 100,
+				end:   130,
+				size:  31,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 150,
+				})
+				require.NoError(t, err)
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     true,
+		},
+		{
+			name: "Beyond head from genesis",
+			rp: rangeParams{
+				start: 100,
+				end:   130,
+				size:  31,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 100,
+				})
+				require.NoError(t, err)
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     true,
+		},
+		{
+			name: "Before our checkpoint block",
+			rp: rangeParams{
+				start: 100,
+				end:   130,
+				size:  31,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 200,
+				})
+				require.NoError(t, err)
+				nBlock := util.NewBeaconBlock()
+				nBlock.Block.Slot = 150
+				root, err := nBlock.Block.HashTreeRoot()
+				require.NoError(t, err)
+				roBlock, err := blocks.NewSignedBeaconBlock(nBlock)
+				require.NoError(t, err)
+				require.NoError(t, d.SaveOriginCheckpointBlockRoot(context.Background(), root))
+				require.NoError(t, d.SaveBlock(context.Background(), roBlock))
+
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     false,
+		},
+		{
+			name: "After our checkpoint block",
+			rp: rangeParams{
+				start: 190,
+				end:   200,
+				size:  11,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 200,
+				})
+				require.NoError(t, err)
+				nBlock := util.NewBeaconBlock()
+				nBlock.Block.Slot = 150
+				root, err := nBlock.Block.HashTreeRoot()
+				require.NoError(t, err)
+				roBlock, err := blocks.NewSignedBeaconBlock(nBlock)
+				require.NoError(t, err)
+				require.NoError(t, d.SaveOriginCheckpointBlockRoot(context.Background(), root))
+				require.NoError(t, d.SaveBlock(context.Background(), roBlock))
+
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     true,
+		},
+		{
+			name: "Before lowest in backfill status",
+			rp: rangeParams{
+				start: 90,
+				end:   130,
+				size:  41,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 200,
+				})
+				require.NoError(t, err)
+				nBlock := util.NewBeaconBlock()
+				nBlock.Block.Slot = 150
+				root, err := nBlock.Block.HashTreeRoot()
+				require.NoError(t, err)
+				roBlock, err := blocks.NewSignedBeaconBlock(nBlock)
+				require.NoError(t, err)
+				require.NoError(t, d.SaveOriginCheckpointBlockRoot(context.Background(), root))
+				require.NoError(t, d.SaveBlock(context.Background(), roBlock))
+				require.NoError(t, d.SaveBackfillStatus(context.Background(), &dbval.BackfillStatus{
+					LowSlot:       100,
+					LowRoot:       make([]byte, 32),
+					LowParentRoot: make([]byte, 32),
+					OriginSlot:    150,
+					OriginRoot:    root[:],
+				}))
+
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     false,
+		},
+		{
+			name: "After lowest in backfill status",
+			rp: rangeParams{
+				start: 110,
+				end:   130,
+				size:  21,
+			},
+			svcCreator: func(t *testing.T) *Service {
+				d := db.SetupDB(t)
+				clock := startup.NewClock(time.Now(), [32]byte{})
+				st, err := state_native.InitializeFromProtoPhase0(&ethpb.BeaconState{
+					Slot: 200,
+				})
+				require.NoError(t, err)
+				nBlock := util.NewBeaconBlock()
+				nBlock.Block.Slot = 150
+				root, err := nBlock.Block.HashTreeRoot()
+				require.NoError(t, err)
+				roBlock, err := blocks.NewSignedBeaconBlock(nBlock)
+				require.NoError(t, err)
+				require.NoError(t, d.SaveOriginCheckpointBlockRoot(context.Background(), root))
+				require.NoError(t, d.SaveBlock(context.Background(), roBlock))
+				require.NoError(t, d.SaveBackfillStatus(context.Background(), &dbval.BackfillStatus{
+					LowSlot:       100,
+					LowRoot:       make([]byte, 32),
+					LowParentRoot: make([]byte, 32),
+					OriginSlot:    150,
+					OriginRoot:    root[:],
+				}))
+
+				return &Service{cfg: &config{beaconDB: d, clock: clock, chain: &chainMock.ChainService{State: st}}}
+			},
+			expectedError: nil,
+			available:     true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := tt.svcCreator(t)
+			avail, err := s.validateRangeAvailability(context.Background(), tt.rp)
+			if tt.expectedError != nil {
+				assert.ErrorContains(t, tt.expectedError.Error(), err, "got incorrect error")
+			} else {
+				assert.NoError(t, err, "wanted no error")
+			}
+			assert.Equal(t, tt.available, avail)
+		})
+	}
+}