diff --git a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go b/beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go index ad326a8af1c7..0d97b31afa10 100644 --- a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go +++ b/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) diff --git a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair.go b/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair.go index ad6c50ba1946..74de8a403055 100644 --- a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair.go +++ b/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") diff --git a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair_test.go b/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair_test.go index 63123eb70e46..440d29826e24 100644 --- a/beacon-chain/rpc/prysm/v1alpha1/validator/proposer_altair_test.go +++ b/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 := ðpb.SyncCommittee{ + Pubkeys: make([][]byte, params.BeaconConfig().SyncCommitteeSize), + } + newCommittee := ðpb.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 := ðpb.SyncCommitteeMessage{ + Slot: slot, + BlockRoot: blockRoot[:], + ValidatorIndex: 0, // This validator is in position 0 of OLD committee + Signature: bls.NewAggregateSignature().Marshal(), + } + msg2 := ðpb.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) +} diff --git a/changelog/ttsao_fix-sync-aggregate-state.md b/changelog/ttsao_fix-sync-aggregate-state.md new file mode 100644 index 000000000000..d4401c58dbe1 --- /dev/null +++ b/changelog/ttsao_fix-sync-aggregate-state.md @@ -0,0 +1,3 @@ +### Fixed + +- Sync committee uses correct state to calculate position \ No newline at end of file