mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-26 05:17:22 +00:00
f5c88e1bcb
* clean up test * adding test setup * add chainstart check to sync querier * goimports * goimports * backend cleanup * adding getters * lint * moving to db * gazelle * adding more services setup * adding more changes * adding different test setups * other merge issues * imports * fixing keys * get intial sync able to be set up * new changes and gazelle * use simulated p2p * everything finally works * unexport fields * revert changes from merge * remove mock server * add documentation * gazelle * add another node * review comments * fix tests
419 lines
9.1 KiB
Go
419 lines
9.1 KiB
Go
package initialsync
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/internal"
|
|
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
|
|
"github.com/prysmaticlabs/prysm/shared/bytesutil"
|
|
"github.com/prysmaticlabs/prysm/shared/event"
|
|
"github.com/prysmaticlabs/prysm/shared/p2p"
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
|
"github.com/prysmaticlabs/prysm/shared/ssz"
|
|
"github.com/prysmaticlabs/prysm/shared/testutil"
|
|
logTest "github.com/sirupsen/logrus/hooks/test"
|
|
)
|
|
|
|
type mockP2P struct {
|
|
}
|
|
|
|
func (mp *mockP2P) Subscribe(msg proto.Message, channel chan p2p.Message) event.Subscription {
|
|
return new(event.Feed).Subscribe(channel)
|
|
}
|
|
|
|
func (mp *mockP2P) Broadcast(msg proto.Message) {}
|
|
|
|
func (mp *mockP2P) Send(msg proto.Message, peer p2p.Peer) {
|
|
}
|
|
|
|
type mockSyncService struct {
|
|
hasStarted bool
|
|
isSynced bool
|
|
}
|
|
|
|
func (ms *mockSyncService) Start() {
|
|
ms.hasStarted = true
|
|
}
|
|
|
|
func (ms *mockSyncService) IsSyncedWithNetwork() bool {
|
|
return ms.isSynced
|
|
}
|
|
|
|
func (ms *mockSyncService) ResumeSync() {
|
|
|
|
}
|
|
|
|
type mockChainService struct{}
|
|
|
|
func (ms *mockChainService) IncomingBlockFeed() *event.Feed {
|
|
return &event.Feed{}
|
|
}
|
|
|
|
func TestSetBlock_InitialSync(t *testing.T) {
|
|
hook := logTest.NewGlobal()
|
|
db := internal.SetupDB(t)
|
|
defer internal.TeardownDB(t, db)
|
|
|
|
cfg := &Config{
|
|
P2P: &mockP2P{},
|
|
SyncService: &mockSyncService{},
|
|
BeaconDB: db,
|
|
ChainService: &mockChainService{},
|
|
}
|
|
|
|
ss := NewInitialSyncService(context.Background(), cfg)
|
|
|
|
exitRoutine := make(chan bool)
|
|
delayChan := make(chan time.Time)
|
|
defer func() {
|
|
close(exitRoutine)
|
|
close(delayChan)
|
|
}()
|
|
|
|
go func() {
|
|
ss.run(delayChan)
|
|
exitRoutine <- true
|
|
}()
|
|
|
|
genericHash := make([]byte, 32)
|
|
genericHash[0] = 'a'
|
|
|
|
block := &pb.BeaconBlock{
|
|
Eth1Data: &pb.Eth1Data{
|
|
DepositRootHash32: []byte{1, 2, 3},
|
|
BlockHash32: []byte{4, 5, 6},
|
|
},
|
|
ParentRootHash32: genericHash,
|
|
Slot: params.BeaconConfig().GenesisSlot + 1,
|
|
StateRootHash32: genericHash,
|
|
}
|
|
|
|
blockResponse := &pb.BeaconBlockResponse{Block: block}
|
|
msg1 := p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: blockResponse,
|
|
}
|
|
|
|
ss.blockBuf <- msg1
|
|
|
|
ss.cancel()
|
|
<-exitRoutine
|
|
|
|
stateHash := bytesutil.ToBytes32(blockResponse.Block.StateRootHash32)
|
|
|
|
if stateHash != ss.genesisStateRootHash32 {
|
|
t.Fatalf("Beacon state hash not updated: %#x", blockResponse.Block.StateRootHash32)
|
|
}
|
|
|
|
hook.Reset()
|
|
}
|
|
|
|
func TestSavingBlock_InSync(t *testing.T) {
|
|
hook := logTest.NewGlobal()
|
|
db := internal.SetupDB(t)
|
|
defer internal.TeardownDB(t, db)
|
|
|
|
cfg := &Config{
|
|
P2P: &mockP2P{},
|
|
SyncService: &mockSyncService{},
|
|
BeaconDB: db,
|
|
ChainService: &mockChainService{},
|
|
}
|
|
ss := NewInitialSyncService(context.Background(), cfg)
|
|
|
|
exitRoutine := make(chan bool)
|
|
delayChan := make(chan time.Time)
|
|
|
|
defer func() {
|
|
close(exitRoutine)
|
|
close(delayChan)
|
|
}()
|
|
|
|
go func() {
|
|
ss.run(delayChan)
|
|
exitRoutine <- true
|
|
}()
|
|
|
|
genericHash := make([]byte, 32)
|
|
genericHash[0] = 'a'
|
|
|
|
beaconState := &pb.BeaconState{
|
|
FinalizedEpoch: params.BeaconConfig().GenesisSlot + 1,
|
|
}
|
|
|
|
stateResponse := &pb.BeaconStateResponse{
|
|
BeaconState: beaconState,
|
|
}
|
|
|
|
incorrectState := &pb.BeaconState{
|
|
FinalizedEpoch: params.BeaconConfig().GenesisSlot,
|
|
JustifiedEpoch: params.BeaconConfig().GenesisSlot + 1,
|
|
}
|
|
|
|
incorrectStateResponse := &pb.BeaconStateResponse{
|
|
BeaconState: incorrectState,
|
|
}
|
|
|
|
stateRoot, err := ssz.TreeHash(beaconState)
|
|
if err != nil {
|
|
t.Fatalf("unable to tree hash state: %v", err)
|
|
}
|
|
beaconStateRootHash32 := stateRoot
|
|
|
|
getBlockResponseMsg := func(Slot uint64) p2p.Message {
|
|
block := &pb.BeaconBlock{
|
|
Eth1Data: &pb.Eth1Data{
|
|
DepositRootHash32: []byte{1, 2, 3},
|
|
BlockHash32: []byte{4, 5, 6},
|
|
},
|
|
ParentRootHash32: genericHash,
|
|
Slot: Slot,
|
|
StateRootHash32: beaconStateRootHash32[:],
|
|
}
|
|
|
|
blockResponse := &pb.BeaconBlockResponse{
|
|
Block: block,
|
|
}
|
|
|
|
return p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: blockResponse,
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
t.Fatalf("Unable to hash block %v", err)
|
|
}
|
|
|
|
msg1 := getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 1)
|
|
|
|
// saving genesis block
|
|
ss.blockBuf <- msg1
|
|
|
|
msg2 := p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: incorrectStateResponse,
|
|
}
|
|
|
|
ss.stateBuf <- msg2
|
|
|
|
if ss.currentSlot == incorrectStateResponse.BeaconState.FinalizedEpoch*params.BeaconConfig().SlotsPerEpoch {
|
|
t.Fatalf("Beacon state updated incorrectly: %d", ss.currentSlot)
|
|
}
|
|
|
|
msg2.Data = stateResponse
|
|
|
|
ss.stateBuf <- msg2
|
|
|
|
if beaconStateRootHash32 != ss.genesisStateRootHash32 {
|
|
br := msg1.Data.(*pb.BeaconBlockResponse)
|
|
t.Fatalf("state hash not updated to: %#x instead it is %#x", br.Block.StateRootHash32,
|
|
ss.genesisStateRootHash32)
|
|
}
|
|
|
|
msg1 = getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 1)
|
|
ss.blockBuf <- msg1
|
|
if params.BeaconConfig().GenesisSlot+1 != ss.currentSlot {
|
|
t.Fatalf("Slot saved when it was not supposed too: %v", stateResponse.BeaconState.FinalizedEpoch*params.BeaconConfig().SlotsPerEpoch)
|
|
}
|
|
|
|
msg1 = getBlockResponseMsg(params.BeaconConfig().GenesisSlot + 2)
|
|
ss.blockBuf <- msg1
|
|
|
|
ss.cancel()
|
|
<-exitRoutine
|
|
|
|
br := msg1.Data.(*pb.BeaconBlockResponse)
|
|
|
|
if br.Block.Slot != ss.currentSlot {
|
|
t.Fatalf("Slot not updated despite receiving a valid block: %v", ss.currentSlot)
|
|
}
|
|
|
|
hook.Reset()
|
|
}
|
|
|
|
func TestDelayChan_OK(t *testing.T) {
|
|
hook := logTest.NewGlobal()
|
|
db := internal.SetupDB(t)
|
|
defer internal.TeardownDB(t, db)
|
|
cfg := &Config{
|
|
P2P: &mockP2P{},
|
|
SyncService: &mockSyncService{},
|
|
BeaconDB: db,
|
|
ChainService: &mockChainService{},
|
|
}
|
|
ss := NewInitialSyncService(context.Background(), cfg)
|
|
|
|
exitRoutine := make(chan bool)
|
|
delayChan := make(chan time.Time)
|
|
|
|
defer func() {
|
|
close(exitRoutine)
|
|
close(delayChan)
|
|
}()
|
|
|
|
go func() {
|
|
ss.run(delayChan)
|
|
exitRoutine <- true
|
|
}()
|
|
|
|
genericHash := make([]byte, 32)
|
|
genericHash[0] = 'a'
|
|
|
|
beaconState := &pb.BeaconState{
|
|
FinalizedEpoch: params.BeaconConfig().GenesisSlot + 1,
|
|
}
|
|
|
|
stateResponse := &pb.BeaconStateResponse{
|
|
BeaconState: beaconState,
|
|
}
|
|
|
|
stateRoot, err := ssz.TreeHash(beaconState)
|
|
if err != nil {
|
|
t.Fatalf("unable to tree hash state: %v", err)
|
|
}
|
|
beaconStateRootHash32 := stateRoot
|
|
|
|
block := &pb.BeaconBlock{
|
|
Eth1Data: &pb.Eth1Data{
|
|
DepositRootHash32: []byte{1, 2, 3},
|
|
BlockHash32: []byte{4, 5, 6},
|
|
},
|
|
ParentRootHash32: genericHash,
|
|
Slot: params.BeaconConfig().GenesisSlot + 1,
|
|
StateRootHash32: beaconStateRootHash32[:],
|
|
}
|
|
|
|
blockResponse := &pb.BeaconBlockResponse{
|
|
Block: block,
|
|
}
|
|
|
|
msg1 := p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: blockResponse,
|
|
}
|
|
|
|
msg2 := p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: stateResponse,
|
|
}
|
|
|
|
ss.blockBuf <- msg1
|
|
|
|
ss.stateBuf <- msg2
|
|
|
|
blockResponse.Block.Slot = params.BeaconConfig().GenesisSlot + 1
|
|
msg1.Data = blockResponse
|
|
|
|
ss.blockBuf <- msg1
|
|
|
|
delayChan <- time.Time{}
|
|
|
|
ss.cancel()
|
|
<-exitRoutine
|
|
|
|
testutil.AssertLogsContain(t, hook, "Exiting initial sync and starting normal sync")
|
|
|
|
hook.Reset()
|
|
}
|
|
|
|
func TestRequestBlocksBySlot_OK(t *testing.T) {
|
|
hook := logTest.NewGlobal()
|
|
db := internal.SetupDB(t)
|
|
defer internal.TeardownDB(t, db)
|
|
cfg := &Config{
|
|
P2P: &mockP2P{},
|
|
SyncService: &mockSyncService{},
|
|
ChainService: &mockChainService{},
|
|
BeaconDB: db,
|
|
BlockBufferSize: 100,
|
|
}
|
|
ss := NewInitialSyncService(context.Background(), cfg)
|
|
newState, err := state.GenesisBeaconState(nil, 0, nil)
|
|
if err != nil {
|
|
t.Fatalf("could not create new state %v", err)
|
|
}
|
|
|
|
err = ss.db.SaveState(newState)
|
|
if err != nil {
|
|
t.Fatalf("could not save beacon state %v", err)
|
|
}
|
|
|
|
exitRoutine := make(chan bool)
|
|
delayChan := make(chan time.Time)
|
|
|
|
defer func() {
|
|
close(exitRoutine)
|
|
close(delayChan)
|
|
}()
|
|
|
|
go func() {
|
|
ss.run(delayChan)
|
|
exitRoutine <- true
|
|
}()
|
|
genericHash := make([]byte, 32)
|
|
genericHash[0] = 'a'
|
|
|
|
getBlockResponseMsg := func(Slot uint64) (p2p.Message, [32]byte) {
|
|
|
|
block := &pb.BeaconBlock{
|
|
Eth1Data: &pb.Eth1Data{
|
|
DepositRootHash32: []byte{1, 2, 3},
|
|
BlockHash32: []byte{4, 5, 6},
|
|
},
|
|
ParentRootHash32: genericHash,
|
|
Slot: Slot,
|
|
StateRootHash32: nil,
|
|
}
|
|
|
|
blockResponse := &pb.BeaconBlockResponse{
|
|
Block: block,
|
|
}
|
|
|
|
root, err := ssz.TreeHash(block)
|
|
if err != nil {
|
|
t.Fatalf("unable to tree hash block %v", err)
|
|
}
|
|
|
|
return p2p.Message{
|
|
Peer: p2p.Peer{},
|
|
Data: blockResponse,
|
|
}, root
|
|
}
|
|
|
|
// sending all blocks except for the genesis block
|
|
startSlot := 1 + params.BeaconConfig().GenesisSlot
|
|
for i := startSlot; i < startSlot+10; i++ {
|
|
response, _ := getBlockResponseMsg(i)
|
|
ss.blockBuf <- response
|
|
}
|
|
|
|
initialResponse, _ := getBlockResponseMsg(1 + params.BeaconConfig().GenesisSlot)
|
|
|
|
//sending genesis block
|
|
ss.blockBuf <- initialResponse
|
|
|
|
_, hash := getBlockResponseMsg(9 + params.BeaconConfig().GenesisSlot)
|
|
|
|
expString := fmt.Sprintf("Saved block with root %#x and slot %d for initial sync",
|
|
hash, 9+params.BeaconConfig().GenesisSlot)
|
|
|
|
// waiting for the current slot to come up to the
|
|
// expected one.
|
|
testutil.WaitForLog(t, hook, expString)
|
|
|
|
delayChan <- time.Time{}
|
|
|
|
ss.cancel()
|
|
<-exitRoutine
|
|
|
|
testutil.AssertLogsContain(t, hook, "Exiting initial sync and starting normal sync")
|
|
|
|
hook.Reset()
|
|
}
|