mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
beacon/light: various small changes
This commit is contained in:
parent
7c613fe752
commit
f044766bf5
15 changed files with 114 additions and 178 deletions
|
|
@ -131,15 +131,15 @@ func (api *BeaconLightApi) GetBestUpdatesAndCommittees(firstPeriod, count uint64
|
||||||
//
|
//
|
||||||
// See data structure definition here:
|
// See data structure definition here:
|
||||||
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientoptimisticupdate
|
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientoptimisticupdate
|
||||||
func (api *BeaconLightApi) GetOptimisticHeadUpdate() (types.SignedHead, error) {
|
func (api *BeaconLightApi) GetOptimisticHeadUpdate() (types.SignedHeader, error) {
|
||||||
resp, err := api.httpGet("/eth/v1/beacon/light_client/optimistic_update")
|
resp, err := api.httpGet("/eth/v1/beacon/light_client/optimistic_update")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return types.SignedHead{}, err
|
return types.SignedHeader{}, err
|
||||||
}
|
}
|
||||||
return decodeOptimisticHeadUpdate(resp)
|
return decodeOptimisticHeadUpdate(resp)
|
||||||
}
|
}
|
||||||
|
|
||||||
func decodeOptimisticHeadUpdate(enc []byte) (types.SignedHead, error) {
|
func decodeOptimisticHeadUpdate(enc []byte) (types.SignedHeader, error) {
|
||||||
var data struct {
|
var data struct {
|
||||||
Data struct {
|
Data struct {
|
||||||
Header types.JsonBeaconHeader `json:"attested_header"`
|
Header types.JsonBeaconHeader `json:"attested_header"`
|
||||||
|
|
@ -148,22 +148,22 @@ func decodeOptimisticHeadUpdate(enc []byte) (types.SignedHead, error) {
|
||||||
} `json:"data"`
|
} `json:"data"`
|
||||||
}
|
}
|
||||||
if err := json.Unmarshal(enc, &data); err != nil {
|
if err := json.Unmarshal(enc, &data); err != nil {
|
||||||
return types.SignedHead{}, err
|
return types.SignedHeader{}, err
|
||||||
}
|
}
|
||||||
if data.Data.Header.Beacon.StateRoot == (common.Hash{}) {
|
if data.Data.Header.Beacon.StateRoot == (common.Hash{}) {
|
||||||
// workaround for different event encoding format in Lodestar
|
// workaround for different event encoding format in Lodestar
|
||||||
if err := json.Unmarshal(enc, &data.Data); err != nil {
|
if err := json.Unmarshal(enc, &data.Data); err != nil {
|
||||||
return types.SignedHead{}, err
|
return types.SignedHeader{}, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(data.Data.Aggregate.BitMask) != params.SyncCommitteeBitmaskSize {
|
if len(data.Data.Aggregate.BitMask) != params.SyncCommitteeBitmaskSize {
|
||||||
return types.SignedHead{}, errors.New("invalid sync_committee_bits length")
|
return types.SignedHeader{}, errors.New("invalid sync_committee_bits length")
|
||||||
}
|
}
|
||||||
if len(data.Data.Aggregate.Signature) != params.BlsSignatureSize {
|
if len(data.Data.Aggregate.Signature) != params.BlsSignatureSize {
|
||||||
return types.SignedHead{}, errors.New("invalid sync_committee_signature length")
|
return types.SignedHeader{}, errors.New("invalid sync_committee_signature length")
|
||||||
}
|
}
|
||||||
return types.SignedHead{
|
return types.SignedHeader{
|
||||||
Header: data.Data.Header.Beacon,
|
Header: data.Data.Header.Beacon,
|
||||||
SyncAggregate: data.Data.Aggregate,
|
SyncAggregate: data.Data.Aggregate,
|
||||||
SignatureSlot: uint64(data.Data.SignatureSlot),
|
SignatureSlot: uint64(data.Data.SignatureSlot),
|
||||||
|
|
@ -309,7 +309,7 @@ func decodeHeadEvent(enc []byte) (uint64, common.Hash, error) {
|
||||||
// head updates and calls the specified callback functions when they are received.
|
// head updates and calls the specified callback functions when they are received.
|
||||||
// The callbacks are also called for the current head and optimistic head at startup.
|
// The callbacks are also called for the current head and optimistic head at startup.
|
||||||
// They are never called concurrently.
|
// They are never called concurrently.
|
||||||
func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot common.Hash), signedFn func(head types.SignedHead), errFn func(err error)) func() {
|
func (api *BeaconLightApi) StartHeadListener(headFn func(slot uint64, blockRoot common.Hash), signedFn func(head types.SignedHeader), errFn func(err error)) func() {
|
||||||
closeCh := make(chan struct{}) // initiate closing the stream
|
closeCh := make(chan struct{}) // initiate closing the stream
|
||||||
closedCh := make(chan struct{}) // stream closed (or failed to create)
|
closedCh := make(chan struct{}) // stream closed (or failed to create)
|
||||||
stoppedCh := make(chan struct{}) // sync loop stopped
|
stoppedCh := make(chan struct{}) // sync loop stopped
|
||||||
|
|
|
||||||
|
|
@ -39,7 +39,7 @@ func NewSyncServer(api *BeaconLightApi) *SyncServer {
|
||||||
return &SyncServer{api: api}
|
return &SyncServer{api: api}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *SyncServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(signedHead types.SignedHead)) {
|
func (s *SyncServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(signedHead types.SignedHeader)) {
|
||||||
s.unsubscribe = s.api.StartHeadListener(newHead, newSignedHead, func(err error) {
|
s.unsubscribe = s.api.StartHeadListener(newHead, newSignedHead, func(err error) {
|
||||||
log.Warn("Head event stream error", "err", err)
|
log.Warn("Head event stream error", "err", err)
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -421,22 +421,22 @@ func (s *CommitteeChain) getSyncCommittee(period uint64) syncCommittee {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// VerifySignedHead returns true if the given signed head has a valid signature
|
// VerifySignedHeader returns true if the given signed head has a valid signature
|
||||||
// according to the local committee chain. The caller should ensure that the
|
// according to the local committee chain. The caller should ensure that the
|
||||||
// committees advertised by the same source where the signed head came from are
|
// committees advertised by the same source where the signed head came from are
|
||||||
// synced before verifying the signature.
|
// synced before verifying the signature.
|
||||||
// The age of the header is also returned (the time elapsed since the beginning
|
// The age of the header is also returned (the time elapsed since the beginning
|
||||||
// of the given slot, according to the local system clock). If enforceTime is
|
// of the given slot, according to the local system clock). If enforceTime is
|
||||||
// true then negative age (future) headers are rejected.
|
// true then negative age (future) headers are rejected.
|
||||||
func (s *CommitteeChain) VerifySignedHead(head types.SignedHead) (bool, time.Duration) {
|
func (s *CommitteeChain) VerifySignedHeader(head types.SignedHeader) (bool, time.Duration) {
|
||||||
s.lock.RLock()
|
s.lock.RLock()
|
||||||
defer s.lock.RUnlock()
|
defer s.lock.RUnlock()
|
||||||
|
|
||||||
return s.verifySignedHead(head)
|
return s.verifySignedHeader(head)
|
||||||
}
|
}
|
||||||
|
|
||||||
// (rlock required)
|
// (rlock required)
|
||||||
func (s *CommitteeChain) verifySignedHead(head types.SignedHead) (bool, time.Duration) {
|
func (s *CommitteeChain) verifySignedHeader(head types.SignedHeader) (bool, time.Duration) {
|
||||||
var (
|
var (
|
||||||
slotTime = int64(time.Second) * int64(s.genesisData.GenesisTime+head.Header.Slot*12)
|
slotTime = int64(time.Second) * int64(s.genesisData.GenesisTime+head.Header.Slot*12)
|
||||||
age = time.Duration(s.unixNano() - slotTime)
|
age = time.Duration(s.unixNano() - slotTime)
|
||||||
|
|
@ -460,7 +460,7 @@ func (s *CommitteeChain) verifyUpdate(update *types.LightClientUpdate) bool {
|
||||||
// verification. Though in reality SignatureSlot is always bigger than update.Header.Slot,
|
// verification. Though in reality SignatureSlot is always bigger than update.Header.Slot,
|
||||||
// setting them as equal here enforces the rule that they have to be in the same sync
|
// setting them as equal here enforces the rule that they have to be in the same sync
|
||||||
// period in order for the light client update proof to be meaningful.
|
// period in order for the light client update proof to be meaningful.
|
||||||
ok, age := s.verifySignedHead(types.SignedHead{Header: update.Header, SyncAggregate: update.SyncAggregate, SignatureSlot: update.Header.Slot})
|
ok, age := s.verifySignedHeader(update.SignedHeader)
|
||||||
if age < 0 {
|
if age < 0 {
|
||||||
log.Warn("Future committee update received", "age", age)
|
log.Warn("Future committee update received", "age", age)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -287,21 +287,21 @@ func (c *committeeChainTest) insertUpdate(tc *testCommitteeChain, period uint64,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *committeeChainTest) verifySignedHead(tc *testCommitteeChain, period float64, expOk bool) {
|
func (c *committeeChainTest) verifySignedHeader(tc *testCommitteeChain, period float64, expOk bool) {
|
||||||
signedHead := tc.makeTestSignedHead(types.Header{Slot: uint64(period * float64(params.SyncPeriodLength))}, 400)
|
signedHead := tc.makeTestSignedHead(types.Header{Slot: uint64(period * float64(params.SyncPeriodLength))}, 400)
|
||||||
if ok, _ := c.chain.VerifySignedHead(signedHead); ok != expOk {
|
if ok, _ := c.chain.VerifySignedHeader(signedHead); ok != expOk {
|
||||||
c.t.Errorf("Incorrect output from VerifySignedHead at period %f (expected %v, got %v)", period, expOk, ok)
|
c.t.Errorf("Incorrect output from VerifySignedHeader at period %f (expected %v, got %v)", period, expOk, ok)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *committeeChainTest) verifyRange(tc *testCommitteeChain, begin, end uint64) {
|
func (c *committeeChainTest) verifyRange(tc *testCommitteeChain, begin, end uint64) {
|
||||||
if begin > 0 {
|
if begin > 0 {
|
||||||
c.verifySignedHead(tc, float64(begin)-0.5, false)
|
c.verifySignedHeader(tc, float64(begin)-0.5, false)
|
||||||
}
|
}
|
||||||
for period := begin; period <= end; period++ {
|
for period := begin; period <= end; period++ {
|
||||||
c.verifySignedHead(tc, float64(period)+0.5, true)
|
c.verifySignedHeader(tc, float64(period)+0.5, true)
|
||||||
}
|
}
|
||||||
c.verifySignedHead(tc, float64(end)+1.5, false)
|
c.verifySignedHeader(tc, float64(end)+1.5, false)
|
||||||
}
|
}
|
||||||
|
|
||||||
func newTestGenesis() GenesisData {
|
func newTestGenesis() GenesisData {
|
||||||
|
|
@ -378,9 +378,9 @@ type testCommitteeChain struct {
|
||||||
genesisData GenesisData
|
genesisData GenesisData
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tc *testCommitteeChain) makeTestSignedHead(header types.Header, signerCount int) types.SignedHead {
|
func (tc *testCommitteeChain) makeTestSignedHead(header types.Header, signerCount int) types.SignedHeader {
|
||||||
bitmask := makeBitmask(signerCount)
|
bitmask := makeBitmask(signerCount)
|
||||||
return types.SignedHead{
|
return types.SignedHeader{
|
||||||
Header: header,
|
Header: header,
|
||||||
SyncAggregate: types.SyncAggregate{
|
SyncAggregate: types.SyncAggregate{
|
||||||
BitMask: bitmask,
|
BitMask: bitmask,
|
||||||
|
|
|
||||||
|
|
@ -38,10 +38,10 @@ func NewHeadValidator(committeeChain *CommitteeChain) *HeadValidator {
|
||||||
type headSub struct {
|
type headSub struct {
|
||||||
minSignerCount int
|
minSignerCount int
|
||||||
nextSlot uint64
|
nextSlot uint64
|
||||||
callbacks []func(types.SignedHead)
|
callbacks []func(types.SignedHeader)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *HeadValidator) Subscribe(minSignerCount int, callback func(types.SignedHead)) {
|
func (h *HeadValidator) Subscribe(minSignerCount int, callback func(types.SignedHeader)) {
|
||||||
h.lock.Lock()
|
h.lock.Lock()
|
||||||
defer h.lock.Unlock()
|
defer h.lock.Unlock()
|
||||||
|
|
||||||
|
|
@ -60,15 +60,15 @@ func (h *HeadValidator) Subscribe(minSignerCount int, callback func(types.Signed
|
||||||
copy(h.subs[insertAt+1:], h.subs[insertAt:len(h.subs)-1])
|
copy(h.subs[insertAt+1:], h.subs[insertAt:len(h.subs)-1])
|
||||||
h.subs[insertAt] = &headSub{
|
h.subs[insertAt] = &headSub{
|
||||||
minSignerCount: minSignerCount,
|
minSignerCount: minSignerCount,
|
||||||
callbacks: []func(types.SignedHead){callback},
|
callbacks: []func(types.SignedHeader){callback},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *HeadValidator) Add(head types.SignedHead) error {
|
func (h *HeadValidator) Add(head types.SignedHeader) error {
|
||||||
h.lock.Lock()
|
h.lock.Lock()
|
||||||
defer h.lock.Unlock()
|
defer h.lock.Unlock()
|
||||||
|
|
||||||
sigOk, age := h.committeeChain.VerifySignedHead(head)
|
sigOk, age := h.committeeChain.VerifySignedHeader(head)
|
||||||
if age < 0 {
|
if age < 0 {
|
||||||
log.Warn("Future signed head received", "age", age)
|
log.Warn("Future signed head received", "age", age)
|
||||||
}
|
}
|
||||||
|
|
@ -79,7 +79,7 @@ func (h *HeadValidator) Add(head types.SignedHead) error {
|
||||||
return errors.New("invalid header signature")
|
return errors.New("invalid header signature")
|
||||||
}
|
}
|
||||||
|
|
||||||
signerCount := head.SignerCount()
|
signerCount := head.SyncAggregate.SignerCount()
|
||||||
for _, sub := range h.subs {
|
for _, sub := range h.subs {
|
||||||
if sub.minSignerCount > signerCount {
|
if sub.minSignerCount > signerCount {
|
||||||
break
|
break
|
||||||
|
|
|
||||||
|
|
@ -33,7 +33,7 @@ import (
|
||||||
// the trusted engine API while head BLS signatures are validated later as they
|
// the trusted engine API while head BLS signatures are validated later as they
|
||||||
// appear, in order to be passed on to other clients.
|
// appear, in order to be passed on to other clients.
|
||||||
type HeadTracker struct {
|
type HeadTracker struct {
|
||||||
newSignedHead func(server *Server, signedHead types.SignedHead)
|
newSignedHead func(server *Server, signedHead types.SignedHeader)
|
||||||
|
|
||||||
validatedLock sync.RWMutex
|
validatedLock sync.RWMutex
|
||||||
validatedHead types.Header
|
validatedHead types.Header
|
||||||
|
|
@ -54,7 +54,7 @@ type serverHeadInfo struct {
|
||||||
|
|
||||||
// NewHeadTracker creates a new HeadTracker. The newSignedHead head callback is
|
// NewHeadTracker creates a new HeadTracker. The newSignedHead head callback is
|
||||||
// called whenever a signed head is received from any of the connected servers.
|
// called whenever a signed head is received from any of the connected servers.
|
||||||
func NewHeadTracker(newSignedHead func(server *Server, signedHead types.SignedHead)) *HeadTracker {
|
func NewHeadTracker(newSignedHead func(server *Server, signedHead types.SignedHeader)) *HeadTracker {
|
||||||
return &HeadTracker{
|
return &HeadTracker{
|
||||||
serverHeads: make(map[*Server]common.Hash),
|
serverHeads: make(map[*Server]common.Hash),
|
||||||
headInfo: make(map[common.Hash]serverHeadInfo),
|
headInfo: make(map[common.Hash]serverHeadInfo),
|
||||||
|
|
@ -101,7 +101,7 @@ func (s *HeadTracker) registerServer(server *Server) {
|
||||||
server.setHead(slot, blockRoot)
|
server.setHead(slot, blockRoot)
|
||||||
s.setServerHead(server, blockRoot)
|
s.setServerHead(server, blockRoot)
|
||||||
server.scheduler.triggerServer(server)
|
server.scheduler.triggerServer(server)
|
||||||
}, func(signedHead types.SignedHead) {
|
}, func(signedHead types.SignedHeader) {
|
||||||
s.newSignedHead(server, signedHead)
|
s.newSignedHead(server, signedHead)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -86,7 +86,7 @@ type Scheduler struct {
|
||||||
|
|
||||||
triggerCh chan struct{} // restarts waiting sync loop
|
triggerCh chan struct{} // restarts waiting sync loop
|
||||||
testWaitCh chan struct{} // accepts sends when sync loop is waiting
|
testWaitCh chan struct{} // accepts sends when sync loop is waiting
|
||||||
testTimerCh chan bool // sends true when simulated timer is processed; false when stopped
|
testTimerResults []bool // true is appended when simulated timer is processed; false when stopped
|
||||||
triggerLock sync.Mutex
|
triggerLock sync.Mutex
|
||||||
waiting, triggered bool
|
waiting, triggered bool
|
||||||
trModules map[Module]struct{}
|
trModules map[Module]struct{}
|
||||||
|
|
|
||||||
|
|
@ -28,13 +28,13 @@ import (
|
||||||
|
|
||||||
type testRequestServer struct {
|
type testRequestServer struct {
|
||||||
newHead func(uint64, common.Hash)
|
newHead func(uint64, common.Hash)
|
||||||
newSignedHead func(types.SignedHead)
|
newSignedHead func(types.SignedHeader)
|
||||||
clock *mclock.Simulated
|
clock *mclock.Simulated
|
||||||
delayUntil mclock.AbsTime
|
delayUntil mclock.AbsTime
|
||||||
failed bool
|
failed bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *testRequestServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(types.SignedHead)) {
|
func (s *testRequestServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(types.SignedHeader)) {
|
||||||
s.newHead, s.newSignedHead = newHead, newSignedHead
|
s.newHead, s.newSignedHead = newHead, newSignedHead
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -89,7 +89,7 @@ func newSchedulerTest(t *testing.T, clock *mclock.Simulated, serverCount, trigge
|
||||||
st := &schedulerTest{
|
st := &schedulerTest{
|
||||||
t: t,
|
t: t,
|
||||||
clock: clock,
|
clock: clock,
|
||||||
scheduler: NewScheduler(NewHeadTracker(func(*Server, types.SignedHead) {}), clock),
|
scheduler: NewScheduler(NewHeadTracker(func(*Server, types.SignedHeader) {}), clock),
|
||||||
modules: make([]*testModule, len(moduleTriggers)),
|
modules: make([]*testModule, len(moduleTriggers)),
|
||||||
triggers: make([]*ModuleTrigger, triggerCount),
|
triggers: make([]*ModuleTrigger, triggerCount),
|
||||||
processCh: make(chan testProcess),
|
processCh: make(chan testProcess),
|
||||||
|
|
@ -219,7 +219,7 @@ func (r *testRequest) returned() {
|
||||||
|
|
||||||
func TestServerTrigger(t *testing.T) {
|
func TestServerTrigger(t *testing.T) {
|
||||||
st := newSchedulerTest(t, &mclock.Simulated{}, 3, 2, [][]int{{0}, {}})
|
st := newSchedulerTest(t, &mclock.Simulated{}, 3, 2, [][]int{{0}, {}})
|
||||||
st.scheduler.testTimerCh = make(chan bool)
|
st.scheduler.testTimerResults = []bool{}
|
||||||
req := &testRequest{}
|
req := &testRequest{}
|
||||||
req.reqLock.Trigger = st.triggers[0]
|
req.reqLock.Trigger = st.triggers[0]
|
||||||
|
|
||||||
|
|
@ -251,9 +251,13 @@ func TestServerTrigger(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
expectTimerFinished := func(exp bool) {
|
expectTimerFinished := func(exp bool) {
|
||||||
if processed := <-st.scheduler.testTimerCh; processed != exp {
|
if len(st.scheduler.testTimerResults) == 0 {
|
||||||
t.Fatalf("Invalid simulated timer result (got processed == %v, expected %v)", processed, exp)
|
t.Fatalf("No timer results found (expected %v)", exp)
|
||||||
}
|
}
|
||||||
|
if st.scheduler.testTimerResults[0] != exp {
|
||||||
|
t.Fatalf("Invalid simulated timer result (got finished == %v, expected %v)", st.scheduler.testTimerResults[0], exp)
|
||||||
|
}
|
||||||
|
st.scheduler.testTimerResults = st.scheduler.testTimerResults[1:]
|
||||||
}
|
}
|
||||||
|
|
||||||
st.servers[2].delayUntil = mclock.AbsTime(time.Second * 5)
|
st.servers[2].delayUntil = mclock.AbsTime(time.Second * 5)
|
||||||
|
|
|
||||||
|
|
@ -29,7 +29,7 @@ import (
|
||||||
// RequestServer is a general server interface that can be extended by modules
|
// RequestServer is a general server interface that can be extended by modules
|
||||||
// with specific request types.
|
// with specific request types.
|
||||||
type RequestServer interface {
|
type RequestServer interface {
|
||||||
SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(types.SignedHead))
|
SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(types.SignedHeader))
|
||||||
UnsubscribeHeads()
|
UnsubscribeHeads()
|
||||||
DelayUntil() mclock.AbsTime // no requests should be sent before this
|
DelayUntil() mclock.AbsTime // no requests should be sent before this
|
||||||
Fail(string) // report server failure
|
Fail(string) // report server failure
|
||||||
|
|
@ -49,10 +49,10 @@ type Server struct {
|
||||||
moduleData map[Module]*interface{}
|
moduleData map[Module]*interface{}
|
||||||
|
|
||||||
lock sync.Mutex
|
lock sync.Mutex
|
||||||
sent map[uint64]chan struct{} // closed when returned; nil when timed out
|
timeouts map[uint64]mclock.Timer // stopped when request has returned; nil when timed out
|
||||||
timeoutCount int
|
timeoutCount int
|
||||||
delayUntil mclock.AbsTime
|
delayUntil mclock.AbsTime
|
||||||
delayTimer mclock.ChanTimer // if non-nil then expires at delayUntil
|
delayTimer mclock.Timer // if non-nil then expires at delayUntil
|
||||||
needTrigger bool
|
needTrigger bool
|
||||||
lastReqId uint64
|
lastReqId uint64
|
||||||
stopCh chan struct{}
|
stopCh chan struct{}
|
||||||
|
|
@ -64,7 +64,7 @@ func (s *Scheduler) newServer(server RequestServer) *Server {
|
||||||
RequestServer: server,
|
RequestServer: server,
|
||||||
scheduler: s,
|
scheduler: s,
|
||||||
moduleData: make(map[Module]*interface{}),
|
moduleData: make(map[Module]*interface{}),
|
||||||
sent: make(map[uint64]chan struct{}),
|
timeouts: make(map[uint64]mclock.Timer),
|
||||||
stopCh: make(chan struct{}),
|
stopCh: make(chan struct{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -113,32 +113,23 @@ func (s *Server) isDelayed() bool {
|
||||||
if delayUntil == s.delayUntil {
|
if delayUntil == s.delayUntil {
|
||||||
return s.delayTimer != nil
|
return s.delayTimer != nil
|
||||||
}
|
}
|
||||||
s.delayUntil = delayUntil
|
|
||||||
if s.delayTimer != nil {
|
if s.delayTimer != nil {
|
||||||
if s.delayTimer.Stop() {
|
// Note: is stopping the timer is unsuccessful then the resulting AfterFunc
|
||||||
if s.scheduler.testTimerCh != nil {
|
// call will just do nothing
|
||||||
s.scheduler.testTimerCh <- false // simulated timer stopped
|
s.stopTimer(s.delayTimer)
|
||||||
}
|
|
||||||
} else {
|
|
||||||
s.delayTimer = nil
|
s.delayTimer = nil
|
||||||
}
|
}
|
||||||
}
|
s.delayUntil = delayUntil
|
||||||
delay := time.Duration(delayUntil - s.scheduler.clock.Now())
|
delay := time.Duration(delayUntil - s.scheduler.clock.Now())
|
||||||
if delay <= 0 {
|
if delay <= 0 {
|
||||||
s.delayTimer = nil
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
if s.delayTimer == nil {
|
s.delayTimer = s.scheduler.clock.AfterFunc(delay, func() {
|
||||||
s.delayTimer = s.scheduler.clock.NewTimer(delay)
|
if s.scheduler.testTimerResults != nil {
|
||||||
} else {
|
s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, true) // simulated timer finished
|
||||||
s.delayTimer.Reset(delay)
|
|
||||||
}
|
}
|
||||||
timer := s.delayTimer
|
|
||||||
go func() {
|
|
||||||
select {
|
|
||||||
case <-timer.C():
|
|
||||||
s.lock.Lock()
|
s.lock.Lock()
|
||||||
if s.delayTimer == timer {
|
if s.delayTimer != nil && s.delayUntil == delayUntil { // do nothing if there is a new timer now
|
||||||
s.delayTimer = nil
|
s.delayTimer = nil
|
||||||
if s.needTrigger && s.timeoutCount == 0 {
|
if s.needTrigger && s.timeoutCount == 0 {
|
||||||
s.needTrigger = false
|
s.needTrigger = false
|
||||||
|
|
@ -146,15 +137,7 @@ func (s *Server) isDelayed() bool {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
s.lock.Unlock()
|
s.lock.Unlock()
|
||||||
if s.scheduler.testTimerCh != nil {
|
})
|
||||||
s.scheduler.testTimerCh <- true // simulated timer processed
|
|
||||||
}
|
|
||||||
case <-s.stopCh:
|
|
||||||
if timer.Stop() && s.scheduler.testTimerCh != nil {
|
|
||||||
s.scheduler.testTimerCh <- false // simulated timer stopped
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -166,44 +149,36 @@ func (s *Server) sendRequest(timeoutTrigger *ModuleTrigger) uint64 {
|
||||||
|
|
||||||
s.lastReqId++
|
s.lastReqId++
|
||||||
reqId := s.lastReqId
|
reqId := s.lastReqId
|
||||||
returnCh := make(chan struct{})
|
s.timeouts[reqId] = s.scheduler.clock.AfterFunc(softRequestTimeout, func() {
|
||||||
s.sent[reqId] = returnCh
|
if s.scheduler.testTimerResults != nil {
|
||||||
timer := s.scheduler.clock.NewTimer(softRequestTimeout)
|
s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, true) // simulated timer finished
|
||||||
go func() {
|
|
||||||
select {
|
|
||||||
case <-timer.C():
|
|
||||||
s.lock.Lock()
|
|
||||||
if _, ok := s.sent[reqId]; ok {
|
|
||||||
s.sent[reqId] = nil
|
|
||||||
s.timeoutCount++
|
|
||||||
}
|
}
|
||||||
s.lock.Unlock()
|
s.lock.Lock()
|
||||||
|
if s.timeouts[reqId] != nil {
|
||||||
|
s.timeouts[reqId] = nil
|
||||||
|
s.timeoutCount++
|
||||||
if timeoutTrigger != nil {
|
if timeoutTrigger != nil {
|
||||||
timeoutTrigger.Trigger()
|
timeoutTrigger.Trigger()
|
||||||
}
|
}
|
||||||
if s.scheduler.testTimerCh != nil {
|
|
||||||
s.scheduler.testTimerCh <- true // simulated timer processed
|
|
||||||
}
|
}
|
||||||
case <-returnCh:
|
s.lock.Unlock()
|
||||||
if timer.Stop() && s.scheduler.testTimerCh != nil {
|
})
|
||||||
s.scheduler.testTimerCh <- false // simulated timer stopped
|
|
||||||
}
|
|
||||||
case <-s.stopCh:
|
|
||||||
if timer.Stop() && s.scheduler.testTimerCh != nil {
|
|
||||||
s.scheduler.testTimerCh <- false // simulated timer stopped
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
return reqId
|
return reqId
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Server) stopTimer(timer mclock.Timer) {
|
||||||
|
if timer.Stop() && s.scheduler.testTimerResults != nil {
|
||||||
|
s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, false) // simulated timer stopped
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// hasTimedOut returns true if the given request has timed out.
|
// hasTimedOut returns true if the given request has timed out.
|
||||||
func (s *Server) hasTimedOut(reqId uint64) bool {
|
func (s *Server) hasTimedOut(reqId uint64) bool {
|
||||||
s.lock.Lock()
|
s.lock.Lock()
|
||||||
defer s.lock.Unlock()
|
defer s.lock.Unlock()
|
||||||
|
|
||||||
ch, ok := s.sent[reqId]
|
timer, ok := s.timeouts[reqId]
|
||||||
return ok && ch == nil
|
return ok && timer == nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// returned stops the timeout timer and removes the entry associated with the
|
// returned stops the timeout timer and removes the entry associated with the
|
||||||
|
|
@ -213,9 +188,9 @@ func (s *Server) returned(reqId uint64) {
|
||||||
s.lock.Lock()
|
s.lock.Lock()
|
||||||
defer s.lock.Unlock()
|
defer s.lock.Unlock()
|
||||||
|
|
||||||
if ch, ok := s.sent[reqId]; ok {
|
if timer, ok := s.timeouts[reqId]; ok {
|
||||||
if ch != nil {
|
if timer != nil {
|
||||||
close(ch)
|
s.stopTimer(timer)
|
||||||
} else {
|
} else {
|
||||||
s.timeoutCount--
|
s.timeoutCount--
|
||||||
if s.needTrigger && s.timeoutCount == 0 && !s.isDelayed() {
|
if s.needTrigger && s.timeoutCount == 0 && !s.isDelayed() {
|
||||||
|
|
@ -223,11 +198,18 @@ func (s *Server) returned(reqId uint64) {
|
||||||
s.scheduler.triggerServer(s)
|
s.scheduler.triggerServer(s)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
delete(s.sent, reqId)
|
delete(s.timeouts, reqId)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// stop stops all goroutines associated with the server.
|
// stop stops all goroutines associated with the server.
|
||||||
func (s *Server) stop() {
|
func (s *Server) stop() {
|
||||||
close(s.stopCh)
|
for _, timer := range s.timeouts {
|
||||||
|
if timer != nil {
|
||||||
|
s.stopTimer(timer)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if s.delayTimer != nil {
|
||||||
|
s.stopTimer(s.delayTimer)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -31,7 +31,7 @@ type HeadUpdater struct {
|
||||||
chain *light.CommitteeChain
|
chain *light.CommitteeChain
|
||||||
lock sync.Mutex
|
lock sync.Mutex
|
||||||
nextSyncPeriod uint64
|
nextSyncPeriod uint64
|
||||||
queuedHeads map[*request.Server][]types.SignedHead
|
queuedHeads map[*request.Server][]types.SignedHeader
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewHeadUpdater(headValidator *light.HeadValidator, chain *light.CommitteeChain) *HeadUpdater {
|
func NewHeadUpdater(headValidator *light.HeadValidator, chain *light.CommitteeChain) *HeadUpdater {
|
||||||
|
|
@ -39,7 +39,7 @@ func NewHeadUpdater(headValidator *light.HeadValidator, chain *light.CommitteeCh
|
||||||
headValidator: headValidator,
|
headValidator: headValidator,
|
||||||
chain: chain,
|
chain: chain,
|
||||||
nextSyncPeriod: math.MaxUint64,
|
nextSyncPeriod: math.MaxUint64,
|
||||||
queuedHeads: make(map[*request.Server][]types.SignedHead),
|
queuedHeads: make(map[*request.Server][]types.SignedHeader),
|
||||||
}
|
}
|
||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
@ -49,7 +49,7 @@ func (s *HeadUpdater) SetupModuleTriggers(trigger func(id string, subscribe bool
|
||||||
trigger("newUpdate", true)
|
trigger("newUpdate", true)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *HeadUpdater) NewSignedHead(server *request.Server, signedHead types.SignedHead) {
|
func (s *HeadUpdater) NewSignedHead(server *request.Server, signedHead types.SignedHeader) {
|
||||||
nextPeriod, ok := s.chain.NextSyncPeriod()
|
nextPeriod, ok := s.chain.NextSyncPeriod()
|
||||||
if !ok || signedHead.Header.SyncPeriod() > nextPeriod {
|
if !ok || signedHead.Header.SyncPeriod() > nextPeriod {
|
||||||
s.lock.Lock()
|
s.lock.Lock()
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,7 @@ import (
|
||||||
|
|
||||||
const SerializedCommitteeSize = (params.SyncCommitteeSize + 1) * params.BlsPubkeySize
|
const SerializedCommitteeSize = (params.SyncCommitteeSize + 1) * params.BlsPubkeySize
|
||||||
|
|
||||||
|
// SerializedCommittee is the serialized version of a sync committee
|
||||||
type SerializedCommittee [SerializedCommitteeSize]byte
|
type SerializedCommittee [SerializedCommitteeSize]byte
|
||||||
|
|
||||||
// jsonSyncCommittee is the JSON representation of a sync committee
|
// jsonSyncCommittee is the JSON representation of a sync committee
|
||||||
|
|
@ -137,6 +138,8 @@ type SyncCommittee struct {
|
||||||
aggregate *bls.Pubkey
|
aggregate *bls.Pubkey
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// VerifySignature returns true if the given sync aggregate is a valid signature
|
||||||
|
// or the given hash
|
||||||
func (sc *SyncCommittee) VerifySignature(signingRoot common.Hash, aggregate *SyncAggregate) bool {
|
func (sc *SyncCommittee) VerifySignature(signingRoot common.Hash, aggregate *SyncAggregate) bool {
|
||||||
var (
|
var (
|
||||||
sig bls.Signature
|
sig bls.Signature
|
||||||
|
|
@ -195,6 +198,7 @@ func (s *SyncAggregate) UnmarshalJSON(input []byte) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SignerCount returns the number of signers in the aggregate signature
|
||||||
func (s *SyncAggregate) SignerCount() int {
|
func (s *SyncAggregate) SignerCount() int {
|
||||||
var count int
|
var count int
|
||||||
for _, v := range s.BitMask {
|
for _, v := range s.BitMask {
|
||||||
|
|
|
||||||
|
|
@ -32,6 +32,8 @@ import (
|
||||||
"github.com/minio/sha256-simd"
|
"github.com/minio/sha256-simd"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const syncCommitteeDomain = 7
|
||||||
|
|
||||||
// Fork describes a single beacon chain fork and also stores the calculated
|
// Fork describes a single beacon chain fork and also stores the calculated
|
||||||
// signature domain used after this fork.
|
// signature domain used after this fork.
|
||||||
type Fork struct {
|
type Fork struct {
|
||||||
|
|
@ -79,7 +81,7 @@ func computeDomain(forkVersion []byte, genesisValidatorsRoot common.Hash) merkle
|
||||||
hasher.Write(forkVersion32[:])
|
hasher.Write(forkVersion32[:])
|
||||||
hasher.Write(genesisValidatorsRoot[:])
|
hasher.Write(genesisValidatorsRoot[:])
|
||||||
hasher.Sum(forkDataRoot[:0])
|
hasher.Sum(forkDataRoot[:0])
|
||||||
domain[0] = 7
|
domain[0] = syncCommitteeDomain
|
||||||
copy(domain[4:], forkDataRoot[:28])
|
copy(domain[4:], forkDataRoot[:28])
|
||||||
return domain
|
return domain
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -108,53 +108,13 @@ func PeriodOfSlot(slot uint64) uint64 {
|
||||||
return slot >> params.Log2SyncPeriodLength
|
return slot >> params.Log2SyncPeriodLength
|
||||||
}
|
}
|
||||||
|
|
||||||
// HeaderWithoutState stores beacon header fields except the state root which can
|
// SignedHeader represents a beacon header signed by a sync committee
|
||||||
// be reconstructed from a partial beacon state proof stored alongside the header
|
|
||||||
type HeaderWithoutState struct {
|
|
||||||
Slot uint64
|
|
||||||
ProposerIndex uint64
|
|
||||||
ParentRoot, BodyRoot common.Hash
|
|
||||||
}
|
|
||||||
|
|
||||||
// Hash calculates the block root of the header
|
|
||||||
func (bh *HeaderWithoutState) Hash(stateRoot common.Hash) common.Hash {
|
|
||||||
return bh.Proof(stateRoot).RootHash()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Proof returns a MultiProof of the header
|
|
||||||
func (bh *HeaderWithoutState) Proof(stateRoot common.Hash) merkle.MultiProof {
|
|
||||||
var values [8]merkle.Value // values corresponding to indices 8 to 15 of the beacon header tree
|
|
||||||
binary.LittleEndian.PutUint64(values[params.BhiSlot-8][:8], bh.Slot)
|
|
||||||
binary.LittleEndian.PutUint64(values[params.BhiProposerIndex-8][:8], bh.ProposerIndex)
|
|
||||||
values[params.BhiParentRoot-8] = merkle.Value(bh.ParentRoot)
|
|
||||||
values[params.BhiStateRoot-8] = merkle.Value(stateRoot)
|
|
||||||
values[params.BhiBodyRoot-8] = merkle.Value(bh.BodyRoot)
|
|
||||||
return merkle.MultiProof{Format: headerFormat, Values: values[:]}
|
|
||||||
}
|
|
||||||
|
|
||||||
// FullHeader reconstructs a full Header from a HeaderWithoutState and a state root
|
|
||||||
func (bh *HeaderWithoutState) FullHeader(stateRoot common.Hash) Header {
|
|
||||||
return Header{
|
|
||||||
Slot: bh.Slot,
|
|
||||||
ProposerIndex: bh.ProposerIndex,
|
|
||||||
ParentRoot: bh.ParentRoot,
|
|
||||||
StateRoot: stateRoot,
|
|
||||||
BodyRoot: bh.BodyRoot,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// SignedHead represents a beacon header signed by a sync committee
|
|
||||||
//
|
//
|
||||||
// Note: this structure is created from either an optimistic update or an instant update:
|
// Note: this structure is created from either an optimistic update or an instant update:
|
||||||
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientoptimisticupdate
|
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientoptimisticupdate
|
||||||
// https://github.com/zsfelfoldi/beacon-APIs/blob/instant_update/apis/beacon/light_client/instant_update.yaml
|
// https://github.com/zsfelfoldi/beacon-APIs/blob/instant_update/apis/beacon/light_client/instant_update.yaml
|
||||||
type SignedHead struct {
|
type SignedHeader struct {
|
||||||
Header Header // signed beacon header
|
Header Header // signed beacon header
|
||||||
SyncAggregate SyncAggregate // sync committee signature aggregate
|
SyncAggregate SyncAggregate // sync committee signature aggregate
|
||||||
SignatureSlot uint64 // slot in which the signature has been created (newer than Header.Slot, determines the signing sync committee)
|
SignatureSlot uint64 // slot in which the signature has been created (newer than Header.Slot, determines the signing sync committee)
|
||||||
}
|
}
|
||||||
|
|
||||||
// SignerCount returns the number of individual signers in the signature aggregate
|
|
||||||
func (s *SignedHead) SignerCount() int {
|
|
||||||
return s.SyncAggregate.SignerCount()
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -26,8 +26,6 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
const MaxUpdateScoresLength = 128 // max number of advertised update scores of most recent periods
|
|
||||||
|
|
||||||
// LightClientUpdate is a proof of the next sync committee root based on a header
|
// LightClientUpdate is a proof of the next sync committee root based on a header
|
||||||
// signed by the sync committee of the given period. Optionally the update can
|
// signed by the sync committee of the given period. Optionally the update can
|
||||||
// prove quasi-finality by the signed header referring to a previous, finalized
|
// prove quasi-finality by the signed header referring to a previous, finalized
|
||||||
|
|
@ -37,9 +35,7 @@ const MaxUpdateScoresLength = 128 // max number of advertised update scores of m
|
||||||
// See data structure definition here:
|
// See data structure definition here:
|
||||||
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientupdate
|
// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientupdate
|
||||||
type LightClientUpdate struct {
|
type LightClientUpdate struct {
|
||||||
Header Header
|
SignedHeader
|
||||||
SyncAggregate SyncAggregate
|
|
||||||
SignatureSlot uint64
|
|
||||||
NextSyncCommitteeRoot common.Hash
|
NextSyncCommitteeRoot common.Hash
|
||||||
NextSyncCommitteeBranch merkle.Values
|
NextSyncCommitteeBranch merkle.Values
|
||||||
FinalizedHeader Header
|
FinalizedHeader Header
|
||||||
|
|
@ -83,7 +79,7 @@ func (u *CommitteeUpdate) MarshalJSON() ([]byte, error) {
|
||||||
Header: JsonBeaconHeader{Beacon: u.Update.Header},
|
Header: JsonBeaconHeader{Beacon: u.Update.Header},
|
||||||
NextSyncCommittee: u.NextSyncCommittee,
|
NextSyncCommittee: u.NextSyncCommittee,
|
||||||
NextSyncCommitteeBranch: u.Update.NextSyncCommitteeBranch,
|
NextSyncCommitteeBranch: u.Update.NextSyncCommitteeBranch,
|
||||||
FinalizedHeader: JsonBeaconHeader{Beacon: u.Update.FinalizedHeader}, //TODO should we encode it when not present?
|
FinalizedHeader: JsonBeaconHeader{Beacon: u.Update.FinalizedHeader},
|
||||||
FinalityBranch: u.Update.FinalityBranch,
|
FinalityBranch: u.Update.FinalityBranch,
|
||||||
SyncAggregate: u.Update.SyncAggregate,
|
SyncAggregate: u.Update.SyncAggregate,
|
||||||
SignatureSlot: common.Decimal(u.Update.SignatureSlot),
|
SignatureSlot: common.Decimal(u.Update.SignatureSlot),
|
||||||
|
|
@ -100,9 +96,11 @@ func (u *CommitteeUpdate) UnmarshalJSON(input []byte) error {
|
||||||
u.Version = dec.Version
|
u.Version = dec.Version
|
||||||
u.NextSyncCommittee = dec.Data.NextSyncCommittee
|
u.NextSyncCommittee = dec.Data.NextSyncCommittee
|
||||||
u.Update = &LightClientUpdate{
|
u.Update = &LightClientUpdate{
|
||||||
|
SignedHeader: SignedHeader{
|
||||||
Header: dec.Data.Header.Beacon,
|
Header: dec.Data.Header.Beacon,
|
||||||
SyncAggregate: dec.Data.SyncAggregate,
|
SyncAggregate: dec.Data.SyncAggregate,
|
||||||
SignatureSlot: uint64(dec.Data.SignatureSlot),
|
SignatureSlot: uint64(dec.Data.SignatureSlot),
|
||||||
|
},
|
||||||
NextSyncCommitteeRoot: u.NextSyncCommittee.Root(),
|
NextSyncCommitteeRoot: u.NextSyncCommittee.Root(),
|
||||||
NextSyncCommitteeBranch: dec.Data.NextSyncCommitteeBranch,
|
NextSyncCommitteeBranch: dec.Data.NextSyncCommitteeBranch,
|
||||||
FinalizedHeader: dec.Data.FinalizedHeader.Beacon,
|
FinalizedHeader: dec.Data.FinalizedHeader.Beacon,
|
||||||
|
|
@ -196,20 +194,6 @@ type PeriodRange struct {
|
||||||
First, AfterLast uint64
|
First, AfterLast uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
/*func (a PeriodRange) Shared(b PeriodRange) PeriodRange {
|
|
||||||
if b.First > a.First {
|
|
||||||
a.First = b.First
|
|
||||||
}
|
|
||||||
if b.AfterLast < a.AfterLast {
|
|
||||||
a.AfterLast = b.AfterLast
|
|
||||||
}
|
|
||||||
return a
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a PeriodRange) IsValid() bool {
|
|
||||||
return a.AfterLast >= a.First
|
|
||||||
}*/
|
|
||||||
|
|
||||||
func (a PeriodRange) IsEmpty() bool {
|
func (a PeriodRange) IsEmpty() bool {
|
||||||
return a.AfterLast == a.First
|
return a.AfterLast == a.First
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -108,7 +108,7 @@ func blsync(ctx *cli.Context) error {
|
||||||
committeeChain.SetGenesisData(chainConfig.GenesisData)
|
committeeChain.SetGenesisData(chainConfig.GenesisData)
|
||||||
headUpdater := sync.NewHeadUpdater(headValidator, committeeChain)
|
headUpdater := sync.NewHeadUpdater(headValidator, committeeChain)
|
||||||
headTracker := request.NewHeadTracker(headUpdater.NewSignedHead)
|
headTracker := request.NewHeadTracker(headUpdater.NewSignedHead)
|
||||||
headValidator.Subscribe(threshold, func(signedHead types.SignedHead) {
|
headValidator.Subscribe(threshold, func(signedHead types.SignedHeader) {
|
||||||
headTracker.SetValidatedHead(signedHead.Header)
|
headTracker.SetValidatedHead(signedHead.Header)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue