Fix proposer to use advanced state for sync committee position calculation (#15905)

* Sync committee use correct state to calculate position

* Unit test

Author: terencechain

beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go

@@ -229,7 +229,7 @@ func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.Signed
 		sBlk.SetVoluntaryExits(vs.getExits(head, sBlk.Block().Slot()))
 
 		// Set sync aggregate. New in Altair.
-		vs.setSyncAggregate(ctx, sBlk)
+		vs.setSyncAggregate(ctx, sBlk, head)
 
 		// Set bls to execution change. New in Capella.
 		vs.setBlsToExecData(sBlk, head)

beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair.go

@@ -5,6 +5,7 @@ import (
 	"context"
 
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/core/helpers"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/interfaces"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
@@ -20,12 +21,12 @@ import (
 	"github.com/prysmaticlabs/go-bitfield"
 )
 
-func (vs *Server) setSyncAggregate(ctx context.Context, blk interfaces.SignedBeaconBlock) {
+func (vs *Server) setSyncAggregate(ctx context.Context, blk interfaces.SignedBeaconBlock, headState state.BeaconState) {
 	if blk.Version() < version.Altair {
 		return
 	}
 
-	syncAggregate, err := vs.getSyncAggregate(ctx, slots.PrevSlot(blk.Block().Slot()), blk.Block().ParentRoot())
+	syncAggregate, err := vs.getSyncAggregate(ctx, slots.PrevSlot(blk.Block().Slot()), blk.Block().ParentRoot(), headState)
 	if err != nil {
 		log.WithError(err).Error("Could not get sync aggregate")
 		emptySig := [96]byte{0xC0}
@@ -47,7 +48,7 @@ func (vs *Server) setSyncAggregate(ctx context.Context, blk interfaces.SignedBea
 
 // getSyncAggregate retrieves the sync contributions from the pool to construct the sync aggregate object.
 // The contributions are filtered based on matching of the input root and slot then profitability.
-func (vs *Server) getSyncAggregate(ctx context.Context, slot primitives.Slot, root [32]byte) (*ethpb.SyncAggregate, error) {
+func (vs *Server) getSyncAggregate(ctx context.Context, slot primitives.Slot, root [32]byte, headState state.BeaconState) (*ethpb.SyncAggregate, error) {
 	_, span := trace.StartSpan(ctx, "ProposerServer.getSyncAggregate")
 	defer span.End()
 
@@ -62,7 +63,7 @@ func (vs *Server) getSyncAggregate(ctx context.Context, slot primitives.Slot, ro
 	// Contributions have to match the input root
 	proposerContributions := proposerSyncContributions(poolContributions).filterByBlockRoot(root)
 
-	aggregatedContributions, err := vs.aggregatedSyncCommitteeMessages(ctx, slot, root, poolContributions)
+	aggregatedContributions, err := vs.aggregatedSyncCommitteeMessages(ctx, slot, root, poolContributions, headState)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get aggregated sync committee messages")
 	}
@@ -123,6 +124,7 @@ func (vs *Server) aggregatedSyncCommitteeMessages(
 	slot primitives.Slot,
 	root [32]byte,
 	poolContributions []*ethpb.SyncCommitteeContribution,
+	st state.BeaconState,
 ) ([]*ethpb.SyncCommitteeContribution, error) {
 	subcommitteeCount := params.BeaconConfig().SyncCommitteeSubnetCount
 	subcommitteeSize := params.BeaconConfig().SyncCommitteeSize / subcommitteeCount
@@ -146,10 +148,7 @@ func (vs *Server) aggregatedSyncCommitteeMessages(
 			messageSigs = append(messageSigs, msg.Signature)
 		}
 	}
-	st, err := vs.HeadFetcher.HeadState(ctx)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not get head state")
-	}
+
 	positions, err := helpers.CurrentPeriodPositions(st, messageIndices)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not get sync committee positions")

beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair_test.go

@@ -9,6 +9,7 @@ import (
 	mockSync "github.com/OffchainLabs/prysm/v6/beacon-chain/sync/initial-sync/testing"
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/blocks"
+	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v6/crypto/bls"
 	"github.com/OffchainLabs/prysm/v6/encoding/bytesutil"
 	ethpb "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
@@ -51,15 +52,15 @@ func TestProposer_GetSyncAggregate_OK(t *testing.T) {
 		require.NoError(t, proposerServer.SyncCommitteePool.SaveSyncCommitteeContribution(cont))
 	}
 
-	aggregate, err := proposerServer.getSyncAggregate(t.Context(), 1, bytesutil.ToBytes32(conts[0].BlockRoot))
+	aggregate, err := proposerServer.getSyncAggregate(t.Context(), 1, bytesutil.ToBytes32(conts[0].BlockRoot), st)
 	require.NoError(t, err)
 	require.DeepEqual(t, bitfield.Bitvector32{0xf, 0xf, 0xf, 0xf}, aggregate.SyncCommitteeBits)
 
-	aggregate, err = proposerServer.getSyncAggregate(t.Context(), 2, bytesutil.ToBytes32(conts[0].BlockRoot))
+	aggregate, err = proposerServer.getSyncAggregate(t.Context(), 2, bytesutil.ToBytes32(conts[0].BlockRoot), st)
 	require.NoError(t, err)
 	require.DeepEqual(t, bitfield.Bitvector32{0xaa, 0xaa, 0xaa, 0xaa}, aggregate.SyncCommitteeBits)
 
-	aggregate, err = proposerServer.getSyncAggregate(t.Context(), 3, bytesutil.ToBytes32(conts[0].BlockRoot))
+	aggregate, err = proposerServer.getSyncAggregate(t.Context(), 3, bytesutil.ToBytes32(conts[0].BlockRoot), st)
 	require.NoError(t, err)
 	require.DeepEqual(t, bitfield.NewBitvector32(), aggregate.SyncCommitteeBits)
 }
@@ -68,7 +69,7 @@ func TestServer_SetSyncAggregate_EmptyCase(t *testing.T) {
 	b, err := blocks.NewSignedBeaconBlock(util.NewBeaconBlockAltair())
 	require.NoError(t, err)
 	s := &Server{} // Sever is not initialized with sync committee pool.
-	s.setSyncAggregate(t.Context(), b)
+	s.setSyncAggregate(t.Context(), b, nil)
 	agg, err := b.Block().Body().SyncAggregate()
 	require.NoError(t, err)
 
@@ -138,7 +139,7 @@ func TestProposer_GetSyncAggregate_IncludesSyncCommitteeMessages(t *testing.T) {
 	}
 
 	// The final sync aggregates must have indexes [0,1,2,3] set for both subcommittees
-	sa, err := proposerServer.getSyncAggregate(t.Context(), 1, r)
+	sa, err := proposerServer.getSyncAggregate(t.Context(), 1, r, st)
 	require.NoError(t, err)
 	assert.Equal(t, true, sa.SyncCommitteeBits.BitAt(0))
 	assert.Equal(t, true, sa.SyncCommitteeBits.BitAt(1))
@@ -194,8 +195,99 @@ func Test_aggregatedSyncCommitteeMessages_NoIntersectionWithPoolContributions(t
 		BlockRoot:         r[:],
 	}
 
-	aggregated, err := proposerServer.aggregatedSyncCommitteeMessages(t.Context(), 1, r, []*ethpb.SyncCommitteeContribution{cont})
+	aggregated, err := proposerServer.aggregatedSyncCommitteeMessages(t.Context(), 1, r, []*ethpb.SyncCommitteeContribution{cont}, st)
 	require.NoError(t, err)
 	require.Equal(t, 1, len(aggregated))
 	assert.Equal(t, false, aggregated[0].AggregationBits.BitAt(3))
 }
+
+func TestGetSyncAggregate_CorrectStateAtSyncCommitteePeriodBoundary(t *testing.T) {
+	helpers.ClearCache()
+	syncPeriodBoundaryEpoch := primitives.Epoch(274176) // Real epoch from the bug report
+	slotsPerEpoch := params.BeaconConfig().SlotsPerEpoch
+
+	preEpochState, keys := util.DeterministicGenesisStateAltair(t, 100)
+	require.NoError(t, preEpochState.SetSlot(primitives.Slot(syncPeriodBoundaryEpoch)*slotsPerEpoch-1)) // Last slot of previous epoch
+
+	postEpochState := preEpochState.Copy()
+	require.NoError(t, postEpochState.SetSlot(primitives.Slot(syncPeriodBoundaryEpoch)*slotsPerEpoch+2)) // After 2 missed slots
+
+	oldCommittee := &ethpb.SyncCommittee{
+		Pubkeys: make([][]byte, params.BeaconConfig().SyncCommitteeSize),
+	}
+	newCommittee := &ethpb.SyncCommittee{
+		Pubkeys: make([][]byte, params.BeaconConfig().SyncCommitteeSize),
+	}
+
+	for i := 0; i < int(params.BeaconConfig().SyncCommitteeSize); i++ {
+		if i < len(keys) {
+			oldCommittee.Pubkeys[i] = keys[i%len(keys)].PublicKey().Marshal()
+			// Use different keys for new committee to simulate rotation
+			newCommittee.Pubkeys[i] = keys[(i+10)%len(keys)].PublicKey().Marshal()
+		}
+	}
+
+	require.NoError(t, preEpochState.SetCurrentSyncCommittee(oldCommittee))
+	require.NoError(t, postEpochState.SetCurrentSyncCommittee(newCommittee))
+
+	mockChainService := &chainmock.ChainService{
+		State: postEpochState,
+	}
+
+	proposerServer := &Server{
+		HeadFetcher:       mockChainService,
+		SyncChecker:       &mockSync.Sync{IsSyncing: false},
+		SyncCommitteePool: synccommittee.NewStore(),
+	}
+
+	slot := primitives.Slot(syncPeriodBoundaryEpoch)*slotsPerEpoch + 1 // First slot of new epoch
+	blockRoot := [32]byte{0x01, 0x02, 0x03}
+
+	msg1 := &ethpb.SyncCommitteeMessage{
+		Slot:           slot,
+		BlockRoot:      blockRoot[:],
+		ValidatorIndex: 0, // This validator is in position 0 of OLD committee
+		Signature:      bls.NewAggregateSignature().Marshal(),
+	}
+	msg2 := &ethpb.SyncCommitteeMessage{
+		Slot:           slot,
+		BlockRoot:      blockRoot[:],
+		ValidatorIndex: 1, // This validator is in position 1 of OLD committee
+		Signature:      bls.NewAggregateSignature().Marshal(),
+	}
+
+	require.NoError(t, proposerServer.SyncCommitteePool.SaveSyncCommitteeMessage(msg1))
+	require.NoError(t, proposerServer.SyncCommitteePool.SaveSyncCommitteeMessage(msg2))
+
+	aggregateWrongState, err := proposerServer.getSyncAggregate(t.Context(), slot, blockRoot, postEpochState)
+	require.NoError(t, err)
+
+	aggregateCorrectState, err := proposerServer.getSyncAggregate(t.Context(), slot, blockRoot, preEpochState)
+	require.NoError(t, err)
+
+	wrongStateBits := bitfield.Bitlist(aggregateWrongState.SyncCommitteeBits)
+	correctStateBits := bitfield.Bitlist(aggregateCorrectState.SyncCommitteeBits)
+
+	wrongStateHasValidators := false
+	correctStateHasValidators := false
+
+	for i := 0; i < len(wrongStateBits); i++ {
+		if wrongStateBits[i] != 0 {
+			wrongStateHasValidators = true
+			break
+		}
+	}
+
+	for i := 0; i < len(correctStateBits); i++ {
+		if correctStateBits[i] != 0 {
+			correctStateHasValidators = true
+			break
+		}
+	}
+
+	assert.Equal(t, true, correctStateHasValidators, "Correct state should include validators that sent messages")
+	assert.Equal(t, false, wrongStateHasValidators, "Wrong state should not find validators in incorrect sync committee")
+
+	t.Logf("Wrong state aggregate bits: %x (has validators: %v)", wrongStateBits, wrongStateHasValidators)
+	t.Logf("Correct state aggregate bits: %x (has validators: %v)", correctStateBits, correctStateHasValidators)
+}

changelog/ttsao_fix-sync-aggregate-state.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Sync committee uses correct state to calculate position