From 4d3c2c45dc155b4d2a1a2284190a98418158fb9b Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Thu, 23 Mar 2023 02:25:16 +0100 Subject: [PATCH] cmd/blsync, beacon/light: implement execBlockSyncer with request.Module --- beacon/light/api/light_api.go | 108 +----- beacon/light/api/sync_server.go | 123 +++---- .../{head_tracker.go => head_validator.go} | 10 +- beacon/light/light_chain.go | 268 +++++++++++++++ beacon/light/request/environment.go | 69 ++++ beacon/light/request/head_tracker.go | 146 +++++++++ beacon/light/request/lock.go | 76 +++-- beacon/light/request/scheduler.go | 138 ++++---- beacon/light/request/server.go | 100 +++--- beacon/light/sync/head_sync.go | 105 ------ beacon/light/sync/head_updater.go | 92 ++++++ beacon/light/sync/header_sync.go | 182 +++++++++++ beacon/light/sync/state_sync.go | 222 +++++++++++++ beacon/light/sync/update_sync.go | 160 +++++---- beacon/merkle/binary_merkle.go | 10 + cmd/blsync/block_sync.go | 307 ++++++++++++++++++ cmd/blsync/main.go | 136 +++----- 17 files changed, 1697 insertions(+), 555 deletions(-) mode change 100644 => 100755 beacon/light/api/light_api.go mode change 100644 => 100755 beacon/light/api/sync_server.go rename beacon/light/{head_tracker.go => head_validator.go} (88%) create mode 100644 beacon/light/light_chain.go create mode 100644 beacon/light/request/environment.go create mode 100644 beacon/light/request/head_tracker.go delete mode 100644 beacon/light/sync/head_sync.go create mode 100644 beacon/light/sync/head_updater.go create mode 100644 beacon/light/sync/header_sync.go create mode 100644 beacon/light/sync/state_sync.go create mode 100755 cmd/blsync/block_sync.go diff --git a/beacon/light/api/light_api.go b/beacon/light/api/light_api.go old mode 100644 new mode 100755 index 0c4fc14a3c..20873d5225 --- a/beacon/light/api/light_api.go +++ b/beacon/light/api/light_api.go @@ -21,7 +21,7 @@ import ( "errors" "fmt" "io" - "math/big" + "net/http" "time" @@ -32,9 +32,6 @@ import ( "github.com/ethereum/go-ethereum/beacon/params" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" - ctypes "github.com/ethereum/go-ethereum/core/types" - "github.com/ethereum/go-ethereum/trie" - "github.com/holiman/uint256" "github.com/protolambda/zrnt/eth2/beacon/capella" "github.com/protolambda/zrnt/eth2/configs" "github.com/protolambda/ztyp/tree" @@ -211,36 +208,15 @@ func (api *BeaconLightApi) GetHeader(blockRoot common.Hash) (types.Header, error } // does not verify state root -func (api *BeaconLightApi) GetHeadStateProof(format merkle.ProofFormat) (merkle.MultiProof, error) { - encFormat, bitLength := EncodeCompactProofFormat(format) +//TODO ... +/*func (api *BeaconLightApi) GetHeadStateProof(format merkle.ProofFormat) (merkle.MultiProof, error) { + encFormat, bitLength := EncodeCompactProofFormat(format) //TODO cache encoding? return api.getStateProof("head", format, encFormat, bitLength) -} +}*/ -type StateProofSub struct { - api *BeaconLightApi - format merkle.ProofFormat - encFormat []byte - bitLength int -} - -func (api *BeaconLightApi) SubscribeStateProof(format merkle.ProofFormat, first, period int) (*StateProofSub, error) { - encFormat, bitLength := EncodeCompactProofFormat(format) - _, err := api.httpGetf("/eth/v0/beacon/proof/subscribe/states?format=0x%x&first=%d&period=%d", encFormat, first, period) - if err != nil && err != ErrNotFound { - // if subscribe endpoint is missing then we expect proof endpoint to serve recent states without subscription - return nil, err - } - return &StateProofSub{ - api: api, - format: format, - encFormat: encFormat, - bitLength: bitLength, - }, nil -} - -// verifies state root -func (sub *StateProofSub) Get(stateRoot common.Hash) (merkle.MultiProof, error) { - proof, err := sub.api.getStateProof(stateRoot.Hex(), sub.format, sub.encFormat, sub.bitLength) +func (api *BeaconLightApi) GetStateProof(stateRoot common.Hash, format merkle.ProofFormat) (merkle.MultiProof, error) { + encFormat, bitLength := EncodeCompactProofFormat(format) //TODO cache encoding? + proof, err := api.getStateProof(stateRoot.Hex(), format, encFormat, bitLength) if err != nil { return merkle.MultiProof{}, err } @@ -329,77 +305,27 @@ func (api *BeaconLightApi) GetCheckpointData(checkpointHash common.Hash) (*light return checkpoint, nil } -// GetExecutionPayload fetches the execution block belonging to the beacon block -// specified by beaconRoot and validates its block hash against the expected execRoot. -func (api *BeaconLightApi) GetExecutionPayload(header types.Header) (*ctypes.Block, error) { - resp, err := api.httpGetf("/eth/v2/beacon/blocks/0x%x", header.Hash()) +func (api *BeaconLightApi) GetBeaconBlock(blockRoot common.Hash) (*capella.BeaconBlock, error) { + resp, err := api.httpGetf("/eth/v2/beacon/blocks/0x%x", blockRoot) if err != nil { return nil, err } - spec := configs.Mainnet - // note: eth2 api endpoints serve bellatrix.SignedBeaconBlock instead - // also try github.com/protolambda/eth2api for api bindings - //var beaconBlock bellatrix.BeaconBlock - var beaconBlock capella.BeaconBlock - myJSONBlockData := resp var beaconBlockMessage struct { Data struct { Message capella.BeaconBlock `json:"message"` } `json:"data"` } - if err := json.Unmarshal(myJSONBlockData, &beaconBlockMessage); err != nil { + if err := json.Unmarshal(resp, &beaconBlockMessage); err != nil { return nil, fmt.Errorf("invalid block json data: %v", err) } - beaconBlock = beaconBlockMessage.Data.Message - beaconBodyRoot := common.Hash(beaconBlock.Body.HashTreeRoot(spec, tree.GetHashFn())) - if beaconBodyRoot != header.BodyRoot { - return nil, fmt.Errorf("Beacon body root hash mismatch (expected: %x, got: %x)", header.BodyRoot.Bytes(), beaconBodyRoot.Bytes()) + beaconBlock := new(capella.BeaconBlock) + *beaconBlock = beaconBlockMessage.Data.Message + root := common.Hash(beaconBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn())) + if root != blockRoot { + return nil, fmt.Errorf("Beacon block root hash mismatch (expected: %x, got: %x)", blockRoot, root) } - - payload := &beaconBlock.Body.ExecutionPayload - txs := make([]*ctypes.Transaction, len(payload.Transactions)) - for i, opaqueTx := range payload.Transactions { - var tx ctypes.Transaction - if err := tx.UnmarshalBinary(opaqueTx); err != nil { - return nil, fmt.Errorf("failed to parse tx %d: %v", i, err) - } - txs[i] = &tx - } - withdrawals := make([]*ctypes.Withdrawal, len(payload.Withdrawals)) - for i, w := range payload.Withdrawals { - withdrawals[i] = &ctypes.Withdrawal{ - Index: uint64(w.Index), - Validator: uint64(w.ValidatorIndex), - Address: common.Address(w.Address), - Amount: uint64(w.Amount), - } - } - wroot := ctypes.DeriveSha(ctypes.Withdrawals(withdrawals), trie.NewStackTrie(nil)) - execHeader := &ctypes.Header{ - ParentHash: common.Hash(payload.ParentHash), - UncleHash: ctypes.EmptyUncleHash, - Coinbase: common.Address(payload.FeeRecipient), - Root: common.Hash(payload.StateRoot), - TxHash: ctypes.DeriveSha(ctypes.Transactions(txs), trie.NewStackTrie(nil)), - ReceiptHash: common.Hash(payload.ReceiptsRoot), - Bloom: ctypes.Bloom(payload.LogsBloom), - Difficulty: big.NewInt(0), // constant - Number: new(big.Int).SetUint64(uint64(payload.BlockNumber)), - GasLimit: uint64(payload.GasLimit), - GasUsed: uint64(payload.GasUsed), - Time: uint64(payload.Timestamp), - Extra: []byte(payload.ExtraData), - MixDigest: common.Hash(payload.PrevRandao), // reused in merge - Nonce: ctypes.BlockNonce{}, // zero - BaseFee: (*uint256.Int)(&payload.BaseFeePerGas).ToBig(), - WithdrawalsHash: &wroot, - } - execBlock := ctypes.NewBlockWithHeader(execHeader).WithBody(txs, nil).WithWithdrawals(withdrawals) - if execBlock.Hash() != common.Hash(payload.BlockHash) { - return nil, fmt.Errorf("Sanity check failed, payload hash does not match.") - } - return execBlock, nil + return beaconBlock, nil } func decodeHeadEvent(enc []byte) (uint64, common.Hash, error) { diff --git a/beacon/light/api/sync_server.go b/beacon/light/api/sync_server.go old mode 100644 new mode 100755 index d107ab2ae4..a2fda4b6a1 --- a/beacon/light/api/sync_server.go +++ b/beacon/light/api/sync_server.go @@ -22,8 +22,10 @@ import ( "github.com/ethereum/go-ethereum/beacon/light" "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/beacon/merkle" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" + "github.com/protolambda/zrnt/eth2/beacon/capella" ) const ( @@ -32,33 +34,41 @@ const ( type SyncServer struct { api *BeaconLightApi - Stop func() lock sync.RWMutex - triggerCallback func() - latestHeadSlot uint64 - latestHeadHash common.Hash - signedHeads []types.SignedHead - canRequestBootstrap bool - firstUpdate uint64 //TODO ... + unsubscribe func() + canRequestBootstrap bool + firstUpdate, afterLastUpdate uint64 + firstState uint64 //TODO ... } func NewSyncServer(api *BeaconLightApi) *SyncServer { - s := &SyncServer{ + return &SyncServer{ api: api, canRequestBootstrap: true, } - s.Stop = s.api.StartHeadListener(s.newHead, s.newSignedHead, func(err error) { - log.Warn("Head event stream error", "err", err) - }) - return s } -func (s *SyncServer) SetTriggerCallback(cb func()) { +func (s *SyncServer) SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(signedHead types.SignedHead)) { s.lock.Lock() - defer s.lock.Unlock() + s.unsubscribe = s.api.StartHeadListener(newHead, func(signedHead types.SignedHead) { + s.lock.Lock() + s.afterLastUpdate = types.PeriodOfSlot(signedHead.Header.Slot + 256) + s.lock.Unlock() + newSignedHead(signedHead) + }, func(err error) { + log.Warn("Head event stream error", "err", err) + }) + s.lock.Unlock() +} - s.triggerCallback = cb +func (s *SyncServer) UnsubscribeHeads() { + s.lock.Lock() + if s.unsubscribe != nil { + s.unsubscribe() + s.unsubscribe = nil + } + s.lock.Unlock() } func (s *SyncServer) Delay() time.Duration { return 0 } //TODO @@ -67,20 +77,6 @@ func (s *SyncServer) Fail(desc string) { log.Warn("API endpoint failure", "URL", s.api.url, "error", desc) } -func (s *SyncServer) LatestHead() (uint64, common.Hash) { - s.lock.RLock() - defer s.lock.RUnlock() - - return s.latestHeadSlot, s.latestHeadHash -} - -func (s *SyncServer) SignedHeads() []types.SignedHead { - s.lock.RLock() - defer s.lock.RUnlock() - - return s.signedHeads -} - func (s *SyncServer) CanRequestBootstrap() bool { s.lock.RLock() defer s.lock.RUnlock() @@ -105,10 +101,7 @@ func (s *SyncServer) UpdateRange() types.PeriodRange { s.lock.RLock() defer s.lock.RUnlock() - if len(s.signedHeads) == 0 { - return types.PeriodRange{} - } - r := types.PeriodRange{First: s.firstUpdate, AfterLast: types.PeriodOfSlot(s.signedHeads[len(s.signedHeads)-1].Header.Slot + 256)} + r := types.PeriodRange{First: s.firstUpdate, AfterLast: s.afterLastUpdate} if !r.IsEmpty() { return r } @@ -125,32 +118,44 @@ func (s *SyncServer) RequestUpdates(first, count uint64, response func([]*types. }() } -func (s *SyncServer) newHead(slot uint64, blockRoot common.Hash) { - s.lock.Lock() - defer s.lock.Unlock() - - s.latestHeadSlot, s.latestHeadHash = slot, blockRoot +func (s *SyncServer) RequestBeaconBlock(blockRoot common.Hash, response func(*capella.BeaconBlock)) { + go func() { + if block, err := s.api.GetBeaconBlock(blockRoot); err == nil { + response(block) + } else { + response(nil) + } + }() } -func (s *SyncServer) newSignedHead(signedHead types.SignedHead) { - s.lock.Lock() - defer s.lock.Unlock() - - if s.signedHeads == nil { - s.signedHeads = []types.SignedHead{signedHead} - s.triggerCallback() - return - } - if lastHead := s.signedHeads[len(s.signedHeads)-1]; signedHead.Header.Slot < lastHead.Header.Slot || - (signedHead.Header.Slot == lastHead.Header.Slot && signedHead.SignerCount() <= lastHead.SignerCount()) { - return - } - if len(s.signedHeads) < maxHeadLength { - s.signedHeads = append(s.signedHeads, signedHead) - s.triggerCallback() - return - } - copy(s.signedHeads[:len(s.signedHeads)-1], s.signedHeads[1:]) - s.signedHeads[len(s.signedHeads)-1] = signedHead - s.triggerCallback() +func (s *SyncServer) RequestBeaconHeader(blockRoot common.Hash, response func(*types.Header)) { + go func() { + if header, err := s.api.GetHeader(blockRoot); err == nil { + response(&header) + } else { + response(nil) + } + }() +} + +func (s *SyncServer) BeaconStateTail() uint64 { + s.lock.RLock() + defer s.lock.RUnlock() + + return s.firstState +} + +func (s *SyncServer) RequestBeaconState(slot uint64, stateRoot common.Hash, format merkle.ProofFormat, response func(*merkle.MultiProof)) { + go func() { + if proof, err := s.api.GetStateProof(stateRoot, format); err == nil { + response(&proof) + } else { + s.lock.Lock() + if slot >= s.firstState { + s.firstState = slot + 1 + } + s.lock.Unlock() + response(nil) + } + }() } diff --git a/beacon/light/head_tracker.go b/beacon/light/head_validator.go similarity index 88% rename from beacon/light/head_tracker.go rename to beacon/light/head_validator.go index 74392af583..09af2eaa9f 100644 --- a/beacon/light/head_tracker.go +++ b/beacon/light/head_validator.go @@ -25,14 +25,14 @@ import ( "github.com/ethereum/go-ethereum/log" ) -type HeadTracker struct { +type HeadValidator struct { lock sync.Mutex committeeChain *CommitteeChain subs []*headSub } -func NewHeadTracker(committeeChain *CommitteeChain) *HeadTracker { - return &HeadTracker{committeeChain: committeeChain} +func NewHeadValidator(committeeChain *CommitteeChain) *HeadValidator { + return &HeadValidator{committeeChain: committeeChain} } type headSub struct { @@ -41,7 +41,7 @@ type headSub struct { callbacks []func(types.SignedHead) } -func (h *HeadTracker) Subscribe(minSignerCount int, callback func(types.SignedHead)) { +func (h *HeadValidator) Subscribe(minSignerCount int, callback func(types.SignedHead)) { h.lock.Lock() defer h.lock.Unlock() @@ -64,7 +64,7 @@ func (h *HeadTracker) Subscribe(minSignerCount int, callback func(types.SignedHe } } -func (h *HeadTracker) Add(head types.SignedHead) error { +func (h *HeadValidator) Add(head types.SignedHead) error { h.lock.Lock() defer h.lock.Unlock() diff --git a/beacon/light/light_chain.go b/beacon/light/light_chain.go new file mode 100644 index 0000000000..6eee16806c --- /dev/null +++ b/beacon/light/light_chain.go @@ -0,0 +1,268 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package light + +import ( + "errors" + "sync" + + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/beacon/merkle" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/lru" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + //"github.com/ethereum/go-ethereum/rlp" +) + +var ( + ErrNotFound = errors.New("not found") + ErrEmptySlot = errors.New("empty slot") + ErrInvalidProofFormat = errors.New("invalid proof format") + ErrInvalidStateRoot = errors.New("invalid state root") +) + +type LightChain struct { + lock sync.RWMutex + db ethdb.KeyValueStore //TODO implement database + chainHead, chainTail types.Header + chainInit bool + stateHead, stateTail types.Header + stateInit bool + headerCache *lru.Cache[slotAndHash, types.Header] + canonicalCache *lru.Cache[uint64, common.Hash] + slotCache *lru.Cache[common.Hash, uint64] + stateCache *lru.Cache[slotAndHash, merkle.Values] + stateProofFormat merkle.ProofFormat //TODO slot/parentSlot dependent format +} + +func NewLightChain(db ethdb.KeyValueStore, stateProofFormat merkle.ProofFormat) *LightChain { + //TODO init from db + return &LightChain{ + db: db, + stateProofFormat: stateProofFormat, + headerCache: lru.NewCache[slotAndHash, types.Header](10000), //TODO use smaller cache when db is implemented + canonicalCache: lru.NewCache[uint64, common.Hash](10000), + slotCache: lru.NewCache[common.Hash, uint64](10000), + stateCache: lru.NewCache[slotAndHash, merkle.Values](10000), + } +} + +type slotAndHash struct { + slot uint64 + hash common.Hash +} + +func (lc *LightChain) AddHeader(header types.Header) { + lc.lock.Lock() + defer lc.lock.Unlock() + + blockRoot := header.Hash() + lc.headerCache.Add(slotAndHash{header.Slot, blockRoot}, header) + lc.slotCache.Add(blockRoot, header.Slot) + if lc.chainInit && blockRoot == lc.chainTail.ParentRoot { + var err error + for err == nil { + lc.canonicalCache.Add(header.Slot, header.Hash()) + for slot := header.Slot + 1; slot < lc.chainTail.Slot; slot++ { + lc.canonicalCache.Add(slot, common.Hash{}) + } + lc.chainTail = header + header, err = lc.GetHeaderByHash(header.ParentRoot) + } + } +} + +func (lc *LightChain) SetChainHead(head types.Header) { + lc.lock.Lock() + defer lc.lock.Unlock() + + if !lc.chainInit { + lc.chainInit = true + lc.chainHead = head + lc.chainTail = head + } + for slot := head.Slot + 1; slot <= lc.chainHead.Slot; slot++ { + lc.canonicalCache.Remove(slot) + } + lc.chainHead = head + for !lc.IsCanonical(head) { + lc.canonicalCache.Add(head.Slot, head.Hash()) + parent, err := lc.GetParent(head) + if err != nil { + for slot := lc.chainTail.Slot; slot < head.Slot; slot++ { + lc.canonicalCache.Remove(slot) + } + lc.chainTail = head + lc.stateInit = false + lc.reinitStateChain(head) + return + } + for slot := parent.Slot + 1; slot < head.Slot; slot++ { + lc.canonicalCache.Add(slot, common.Hash{}) + } + head = parent + } + if lc.stateInit && lc.stateHead.Slot >= head.Slot { + if head.Slot >= lc.stateTail.Slot { + lc.stateHead = head + } else { + lc.stateInit = false + } + } + if lc.stateInit { + lc.extendStateHead() + } else { + lc.reinitStateChain(head) + } +} + +func (lc *LightChain) extendStateHead() { + for slot := lc.stateHead.Slot + 1; slot <= lc.chainHead.Slot; slot++ { + if header, err := lc.GetHeaderBySlot(slot); err == nil { + if lc.HasStateProof(header) { + lc.stateHead = header + } else { + break + } + } + } +} + +func (lc *LightChain) extendStateTail() { + if lc.stateTail.Slot == 0 { + return + } + for slot := lc.stateTail.Slot - 1; slot >= lc.chainTail.Slot; slot-- { + if header, err := lc.GetHeaderBySlot(slot); err == nil { + if lc.HasStateProof(header) { + lc.stateTail = header + } else { + break + } + } + } +} + +func (lc *LightChain) reinitStateChain(header types.Header) { + for slot := header.Slot; slot <= lc.chainHead.Slot; slot++ { + if header, err := lc.GetHeaderBySlot(slot); err == nil && lc.HasStateProof(header) { + lc.stateInit = true + lc.stateHead = header + lc.stateTail = header + lc.extendStateHead() + return + } + } + +} + +func (lc *LightChain) HeaderRange() (head, tail types.Header, init bool) { + lc.lock.RLock() + defer lc.lock.RUnlock() + + return lc.chainHead, lc.chainTail, lc.chainInit +} + +func (lc *LightChain) HasHeader(blockRoot common.Hash) bool { + _, ok := lc.slotCache.Get(blockRoot) + return ok +} + +func (lc *LightChain) GetHeaderByHash(blockRoot common.Hash) (types.Header, error) { + if slot, ok := lc.slotCache.Get(blockRoot); ok { + if header, ok := lc.headerCache.Get(slotAndHash{slot, blockRoot}); ok { + return header, nil + } + log.Error("LightChain slot -> blockRoot entry found but header is missing", "slot", slot, "blockRoot", blockRoot) + } + return types.Header{}, ErrNotFound +} + +func (lc *LightChain) GetHeaderBySlot(slot uint64) (types.Header, error) { + if blockRoot, ok := lc.canonicalCache.Get(slot); ok { + if blockRoot == (common.Hash{}) { + return types.Header{}, ErrEmptySlot + } + if header, ok := lc.headerCache.Get(slotAndHash{slot, blockRoot}); ok { + return header, nil + } + log.Error("LightChain canonical blockRoot entry found but header is missing", "slot", slot, "blockRoot", blockRoot) + } + return types.Header{}, ErrNotFound +} + +func (lc *LightChain) GetParent(header types.Header) (types.Header, error) { + return lc.GetHeaderByHash(header.ParentRoot) +} + +func (lc *LightChain) IsCanonical(header types.Header) bool { + blockRoot, ok := lc.canonicalCache.Get(header.Slot) + return ok && blockRoot == header.Hash() +} + +func (lc *LightChain) StateProofRange() (head, tail types.Header, init bool) { + lc.lock.RLock() + defer lc.lock.RUnlock() + + return lc.stateHead, lc.stateTail, lc.stateInit +} + +func (lc *LightChain) HasStateProof(header types.Header) bool { + _, ok := lc.stateCache.Get(slotAndHash{header.Slot, header.StateRoot}) + return ok +} + +func (lc *LightChain) GetStateProof(header types.Header) (merkle.MultiProof, error) { + values, ok := lc.stateCache.Get(slotAndHash{header.Slot, header.StateRoot}) + if !ok { + return merkle.MultiProof{}, ErrNotFound + } + return merkle.MultiProof{Format: lc.stateProofFormat, Values: values}, nil +} + +func (lc *LightChain) StateProofFormat(header types.Header) merkle.ProofFormat { + return lc.stateProofFormat +} + +func (lc *LightChain) AddStateProof(header types.Header, proof merkle.MultiProof) error { + lc.lock.Lock() + defer lc.lock.Unlock() + + if !merkle.IsEqual(proof.Format, lc.StateProofFormat(header)) { + return ErrInvalidProofFormat + } + if proof.RootHash() != header.StateRoot { + return ErrInvalidStateRoot + } + lc.stateCache.Add(slotAndHash{header.Slot, header.StateRoot}, proof.Values) + if !lc.IsCanonical(header) { + return nil + } + if !lc.stateInit { + lc.stateInit = true + lc.stateHead = header + lc.stateTail = header + return nil + } + if header.Slot > lc.stateHead.Slot && header.Slot <= lc.chainHead.Slot { + lc.extendStateHead() + } else if header.Slot < lc.stateTail.Slot && header.Slot >= lc.chainTail.Slot { + lc.extendStateTail() + } + return nil +} diff --git a/beacon/light/request/environment.go b/beacon/light/request/environment.go new file mode 100644 index 0000000000..e12584fc86 --- /dev/null +++ b/beacon/light/request/environment.go @@ -0,0 +1,69 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package request + +type request interface { + CanSendTo(server *Server) (canSend bool, priority uint64) + SendTo(server *Server) +} + +// Environment allows Module.Process to send requests to a set of servers. The enabled server set can either be all servers that are not delayed or timed out (in case of a module trigger) or a subset of them that have been triggered by a server trigger. +type Environment struct { + *HeadTracker + scheduler *Scheduler + allServers []*Server + canRequestNow map[*Server]struct{} +} + +func (s *Environment) TryRequest(req request) (sent, tryMore bool) { + var ( + maxServerPriority, maxRequestPriority uint64 + bestServer *Server + ) + for server := range s.canRequestNow { + canRequest, serverPriority := server.CanRequestNow() + if !canRequest { + delete(s.canRequestNow, server) + continue + } + canSend, requestPriority := req.CanSendTo(server) + if !canSend || requestPriority < maxRequestPriority || + (requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) { + continue + } + maxServerPriority, maxRequestPriority = serverPriority, requestPriority + bestServer = server + } + if bestServer != nil { + req.SendTo(bestServer) + return true, true + } + return false, len(s.canRequestNow) > 0 +} + +func (s *Environment) CanRequestNow() bool { + return len(s.canRequestNow) > 0 +} + +func (s *Environment) CanRequestLater(req request) bool { + for _, server := range s.allServers { + if canSend, _ := req.CanSendTo(server); canSend { + return true + } + } + return false +} diff --git a/beacon/light/request/head_tracker.go b/beacon/light/request/head_tracker.go new file mode 100644 index 0000000000..da91e7c900 --- /dev/null +++ b/beacon/light/request/head_tracker.go @@ -0,0 +1,146 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package request + +import ( + "sync" + + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/common" +) + +type HeadTracker struct { + newSignedHead func(server *Server, signedHead types.SignedHead) + + validatedLock sync.RWMutex + validatedHead types.Header + validatedHeadTrigger *ModuleTrigger + + prefetchLock sync.RWMutex + serverHeads map[*Server]common.Hash + headInfo map[common.Hash]serverHeadInfo + headCounter uint64 + prefetchHead common.Hash + prefetchHeadTrigger *ModuleTrigger +} + +type serverHeadInfo struct { + serverCount int + headCounter uint64 +} + +func NewHeadTracker(newSignedHead func(server *Server, signedHead types.SignedHead)) *HeadTracker { + return &HeadTracker{ + serverHeads: make(map[*Server]common.Hash), + headInfo: make(map[common.Hash]serverHeadInfo), + newSignedHead: newSignedHead, + } +} + +func (s *HeadTracker) SetupTriggers(trigger func(id string) *ModuleTrigger) { + s.validatedHeadTrigger = trigger("validatedHead") + s.prefetchHeadTrigger = trigger("prefetchHead") +} + +func (s *HeadTracker) SetValidatedHead(head types.Header) { + s.validatedLock.Lock() + defer s.validatedLock.Unlock() + + s.validatedHead = head + s.validatedHeadTrigger.Trigger() +} + +func (s *HeadTracker) ValidatedHead() types.Header { + s.validatedLock.RLock() + defer s.validatedLock.RUnlock() + + return s.validatedHead +} + +func (s *HeadTracker) registerServer(server *Server) { + s.prefetchLock.Lock() + defer s.prefetchLock.Unlock() + + server.SubscribeHeads(func(slot uint64, blockRoot common.Hash) { + s.prefetchLock.Lock() + defer s.prefetchLock.Unlock() + + if server.unregistered { + return + } + server.setHead(slot, blockRoot) + s.setServerHead(server, blockRoot) + server.trigger() + }, func(signedHead types.SignedHead) { + s.newSignedHead(server, signedHead) + }) +} + +func (s *HeadTracker) unregisterServer(server *Server) { + s.prefetchLock.Lock() + defer s.prefetchLock.Unlock() + + server.UnsubscribeHeads() + server.unregistered = true + s.setServerHead(server, common.Hash{}) +} + +func (s *HeadTracker) setServerHead(server *Server, head common.Hash) { + if oldHead, ok := s.serverHeads[server]; ok { + if head == oldHead { + return + } + h := s.headInfo[oldHead] + if h.serverCount--; h.serverCount > 0 { + s.headInfo[oldHead] = h + } else { + delete(s.headInfo, oldHead) + } + } + if head != (common.Hash{}) { + h, ok := s.headInfo[head] + if !ok { + s.headCounter++ + h.headCounter = s.headCounter + } + h.serverCount++ + s.headInfo[head] = h + } + var ( + bestHead common.Hash + bestHeadInfo serverHeadInfo + ) + for head, headInfo := range s.headInfo { + if headInfo.serverCount > bestHeadInfo.serverCount || + (headInfo.serverCount == bestHeadInfo.serverCount && headInfo.headCounter > bestHeadInfo.headCounter) { + bestHead, bestHeadInfo = head, headInfo + } + } + if bestHead != s.prefetchHead { + s.prefetchHead = bestHead + s.prefetchHeadTrigger.Trigger() + } else if head == s.prefetchHead { + server.trigger() + } +} + +func (s *HeadTracker) PrefetchHead() common.Hash { + s.prefetchLock.RLock() + defer s.prefetchLock.RUnlock() + + return s.prefetchHead +} diff --git a/beacon/light/request/lock.go b/beacon/light/request/lock.go index 758522176c..0e80bfc537 100644 --- a/beacon/light/request/lock.go +++ b/beacon/light/request/lock.go @@ -16,32 +16,70 @@ package request +type sentRequest struct { + sentTo *Server + reqId uint64 +} + type SingleLock struct { - requestLock map[*Server]uint64 // servers where the request has been sent and not timed out yet + sentRequest + Trigger *ModuleTrigger } -func (s *SingleLock) CanSend(server *Server) bool { - reqId, ok := s.requestLock[server] - if ok && server.Timeout(reqId) { - delete(s.requestLock, server) - return false +func (s *SingleLock) CanRequest() bool { + if s.sentTo != nil && s.sentTo.hasTimedOut(s.reqId) { + s.sentTo = nil } - return !ok && server.CanSend() + return s.sentTo == nil } -// assumes that canSend returned true (no request lock) -func (s *SingleLock) TrySend(srv *Server) (uint64, bool) { - if s.requestLock == nil { - s.requestLock = make(map[*Server]uint64) - } - if reqId, ok := srv.TrySend(); ok { - s.requestLock[srv] = reqId - return reqId, true - } - return 0, false +func (s *SingleLock) Send(srv *Server) uint64 { + reqId := srv.sendRequest(s.Trigger) + s.sentTo, s.reqId = srv, reqId + return reqId } func (s *SingleLock) Returned(srv *Server, reqId uint64) { - delete(s.requestLock, srv) - srv.Returned(reqId) + if srv == s.sentTo && reqId == s.reqId { + s.sentTo = nil + } + srv.returned(reqId) + if s.Trigger != nil { + s.Trigger.Trigger() + } +} + +type MultiLock struct { + locks map[interface{}]sentRequest // locks are only present in the map when sentTo != nil + Trigger *ModuleTrigger +} + +func (s *MultiLock) CanRequest(id interface{}) bool { + if s.locks == nil { + s.locks = make(map[interface{}]sentRequest) + } + if sl, ok := s.locks[id]; ok { + if sl.sentTo.hasTimedOut(sl.reqId) { + delete(s.locks, id) + } else { + return false + } + } + return true +} + +func (s *MultiLock) Send(srv *Server, id interface{}) uint64 { + reqId := srv.sendRequest(s.Trigger) + s.locks[id] = sentRequest{sentTo: srv, reqId: reqId} + return reqId +} + +func (s *MultiLock) Returned(srv *Server, reqId uint64, id interface{}) { + if s.locks[id] == (sentRequest{sentTo: srv, reqId: reqId}) { + delete(s.locks, id) + } + srv.returned(reqId) + if s.Trigger != nil { + s.Trigger.Trigger() + } } diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 96400df2e9..b4df00b1fe 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -19,16 +19,21 @@ package request import ( "sync" "time" + + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/common" ) const softRequestTimeout = time.Second type Module interface { - Process(servers []*Server) bool // removed if return value is false + SetupTriggers(trigger func(id string, subscribe bool) *ModuleTrigger) + Process(env *Environment) } type RequestServer interface { - SetTriggerCallback(func()) + SubscribeHeads(newHead func(uint64, common.Hash), newSignedHead func(types.SignedHead)) + UnsubscribeHeads() Delay() time.Duration Fail(string) } @@ -47,14 +52,17 @@ func (t *ModuleTrigger) Trigger() { defer t.s.triggerLock.Unlock() for m := range t.triggers { - t.s.moduleTrigger(m) + t.s.triggerModule(m) } } type Scheduler struct { + headTracker *HeadTracker + lock sync.Mutex modules []Module // first has highest priority servers []*Server + triggers map[string]*ModuleTrigger triggeredBy map[Module][]*ModuleTrigger stopCh chan chan struct{} @@ -65,10 +73,12 @@ type Scheduler struct { trServers map[*Server]struct{} } -func NewScheduler() *Scheduler { +func NewScheduler(headTracker *HeadTracker) *Scheduler { return &Scheduler{ + headTracker: headTracker, stopCh: make(chan chan struct{}), triggerCh: make(chan struct{}, 1), + triggers: make(map[string]*ModuleTrigger), triggeredBy: make(map[Module][]*ModuleTrigger), } } @@ -79,46 +89,51 @@ func (s *Scheduler) RegisterModule(m Module) { defer s.lock.Unlock() s.modules = append(s.modules, m) + m.SetupTriggers(func(id string, subscribe bool) *ModuleTrigger { return s.addTrigger(m, id, subscribe) }) } -func (s *Scheduler) AddTriggers(m Module, triggeredBy []*ModuleTrigger) { - s.triggeredBy[m] = append(s.triggeredBy[m], triggeredBy...) - for _, t := range triggeredBy { - if t.triggers == nil { - t.s = s - t.triggers = make(map[Module]struct{}) - } - t.triggers[m] = struct{}{} +func (s *Scheduler) addTrigger(m Module, id string, subscribe bool) *ModuleTrigger { + t, ok := s.triggers[id] + if !ok { + t = new(ModuleTrigger) + s.triggers[id] = t } + if !subscribe { + return t + } + s.triggeredBy[m] = append(s.triggeredBy[m], t) + if t.triggers == nil { + t.s = s + t.triggers = make(map[Module]struct{}) + } + t.triggers[m] = struct{}{} + return t } -func (s *Scheduler) unregisterModule(m Module, t []*ModuleTrigger) { - for i, module := range s.modules { - if module == m { - copy(s.modules[i:len(s.modules)-1], s.modules[i+1:]) - s.modules = s.modules[:len(s.modules)-1] - break - } - } - triggeredBy := s.triggeredBy[m] - delete(s.triggeredBy, m) - for _, t := range triggeredBy { - delete(t.triggers, m) - } -} - -func (s *Scheduler) RegisterServer(RequestServer RequestServer) { +// GetModuleTrigger returns the ModuleTrigger with the given id or creates a new one. +func (s *Scheduler) GetModuleTrigger(id string) *ModuleTrigger { s.lock.Lock() defer s.lock.Unlock() - server := s.newServer(RequestServer) - s.servers = append(s.servers, server) - RequestServer.SetTriggerCallback(func() { - s.ServerTrigger(server) - }) - s.ServerTrigger(server) + t, ok := s.triggers[id] + if !ok { + t = new(ModuleTrigger) + s.triggers[id] = t + } + return t } +// RegisterServer registers a new server. +func (s *Scheduler) RegisterServer(requestServer RequestServer) { + s.lock.Lock() + defer s.lock.Unlock() + + server := s.newServer(requestServer) + s.servers = append(s.servers, server) + s.headTracker.registerServer(server) +} + +// UnregisterServer removes a registered server. func (s *Scheduler) UnregisterServer(RequestServer RequestServer) { s.lock.Lock() defer s.lock.Unlock() @@ -128,16 +143,18 @@ func (s *Scheduler) UnregisterServer(RequestServer RequestServer) { s.servers[i] = s.servers[len(s.servers)-1] s.servers = s.servers[:len(s.servers)-1] server.stop() + s.headTracker.unregisterServer(server) return } } } -// call before registering servers +// Start starts the scheduler. It should be called after registering all modules and before registering any servers. func (s *Scheduler) Start() { go s.syncLoop() } +// Stop stops the scheduler. func (s *Scheduler) Stop() { s.lock.Lock() for _, server := range s.servers { @@ -150,6 +167,7 @@ func (s *Scheduler) Stop() { <-stop } +// syncLoop calls all processable modules in the order of their registration. A round of processing starts whenever there is at least one processable module. Triggers triggered during a processing round do not affect the current round but ensure that there is going to be a next round. func (s *Scheduler) syncLoop() { s.lock.Lock() s.triggerLock.Lock() @@ -179,36 +197,42 @@ func (s *Scheduler) syncLoop() { } } +// processModules runs an entire processing round, calling processable modules with the appropriate Environment. func (s *Scheduler) processModules(trModules map[Module]struct{}, trServers map[*Server]struct{}) { - trs := make([]*Server, 0, len(s.servers)) + mtEnv := Environment{ // enables all servers for triggered modules + HeadTracker: s.headTracker, + scheduler: s, + allServers: s.servers, + canRequestNow: make(map[*Server]struct{}), + } + stEnv := Environment{ // enables triggered servers only for other modules + HeadTracker: s.headTracker, + scheduler: s, + allServers: s.servers, + canRequestNow: make(map[*Server]struct{}), + } for _, server := range s.servers { + if canRequest, _ := server.CanRequestNow(); !canRequest { + continue + } + mtEnv.canRequestNow[server] = struct{}{} if _, ok := trServers[server]; ok { - trs = append(trs, server) + stEnv.canRequestNow[server] = struct{}{} } } - var i int + for _, module := range s.modules { - keep := true if _, ok := trModules[module]; ok { - keep = module.Process(s.servers) - } else if len(trs) > 0 { - keep = module.Process(trs) - } - if keep { - s.modules[i] = module - i++ + module.Process(&mtEnv) + } else if len(stEnv.canRequestNow) > 0 { + module.Process(&stEnv) } } - s.modules = s.modules[:i] } -func (s *Scheduler) ServerTrigger(server *Server) { +// triggerServer ensures that a next processing round is initiated as soon as possible and every module will be called with the given server enabled in its Environment. Should be called when the given server has become available (again) or when its range of servable requests has been expanded. +func (s *Scheduler) triggerServer(server *Server) { s.triggerLock.Lock() - s.serverTrigger(server) - s.triggerLock.Unlock() -} - -func (s *Scheduler) serverTrigger(server *Server) { if s.trServers == nil { s.trServers = make(map[*Server]struct{}) } @@ -217,15 +241,11 @@ func (s *Scheduler) serverTrigger(server *Server) { s.triggerCh <- struct{}{} s.triggered = true } -} - -func (s *Scheduler) ModuleTrigger(module Module) { - s.triggerLock.Lock() - s.moduleTrigger(module) s.triggerLock.Unlock() } -func (s *Scheduler) moduleTrigger(module Module) { +// triggerModule ensures that a next processing round is initiated as soon as possible and the given module will be called with all servers enabled in its Environment. Called by ModuleTrigger.Trigger when the range of possible requests or processable data might have been expanded. +func (s *Scheduler) triggerModule(module Module) { if s.trModules == nil { s.trModules = make(map[Module]struct{}) } diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 326df2eed2..877b1bc4e9 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -20,36 +20,19 @@ import ( "math/rand" "sync" "time" + + "github.com/ethereum/go-ethereum/common" ) -func SelectServer(servers []*Server, priority func(server *Server) uint64) *Server { - var ( - maxPriority uint64 - mpCount int - bestServer *Server - ) - for _, server := range servers { - pri := priority(server) - if pri == 0 || pri < maxPriority { // 0 means it cannot serve the request at all - continue - } - if pri > maxPriority { - maxPriority = pri - mpCount = 1 - bestServer = server - } else { - mpCount++ - if rand.Intn(mpCount) == 0 { - bestServer = server - } - } - } - return bestServer -} - -type Server struct { //TODO name? +type Server struct { RequestServer - scheduler *Scheduler + scheduler *Scheduler + + headLock sync.RWMutex + latestHeadSlot uint64 + latestHeadHash common.Hash + unregistered bool // accessed under HeadTracker.prefetchLock + lock sync.Mutex sent map[uint64]chan struct{} // closed when returned; nil when timed out timeoutCount int @@ -69,51 +52,68 @@ func (s *Scheduler) newServer(server RequestServer) *Server { } } -// guarantees a server trigger later if the result is false -func (s *Server) CanSend() bool { - s.lock.Lock() - defer s.lock.Unlock() +func (s *Server) setHead(slot uint64, blockRoot common.Hash) { + s.headLock.Lock() + defer s.headLock.Unlock() - if s.isDelayed() || s.timeoutCount != 0 { - s.needTrigger = true - return false - } - return true + s.latestHeadSlot, s.latestHeadHash = slot, blockRoot +} + +func (s *Server) trigger() { + s.scheduler.triggerServer(s) +} + +func (s *Server) LatestHead() (uint64, common.Hash) { + s.headLock.RLock() + defer s.headLock.RUnlock() + + return s.latestHeadSlot, s.latestHeadHash } // guarantees a server trigger later if the result is false -func (s *Server) TrySend() (uint64, bool) { +func (s *Server) CanRequestNow() (bool, uint64) { s.lock.Lock() defer s.lock.Unlock() if s.isDelayed() || s.timeoutCount != 0 { s.needTrigger = true - return 0, false + return false, 0 } + return true, uint64(rand.Uint32() + 1) //TODO use priority based on in-flight requests +} + +func (s *Server) sendRequest(timeoutTrigger *ModuleTrigger) uint64 { + s.lock.Lock() + defer s.lock.Unlock() + s.lastReqId++ + reqId := s.lastReqId returnCh := make(chan struct{}) - s.sent[s.lastReqId] = returnCh + s.sent[reqId] = returnCh s.delayChecked = false go func() { timer := time.NewTimer(softRequestTimeout) select { case <-timer.C: s.lock.Lock() - if _, ok := s.sent[s.lastReqId]; ok { - s.sent[s.lastReqId] = nil + if _, ok := s.sent[reqId]; ok { + s.sent[reqId] = nil s.timeoutCount++ } s.lock.Unlock() + if timeoutTrigger != nil { + timeoutTrigger.Trigger() + } case <-returnCh: timer.Stop() case <-s.stopCh: timer.Stop() } }() - return s.lastReqId, true + return reqId } -func (s *Server) Timeout(reqId uint64) bool { +func (s *Server) hasTimedOut(reqId uint64) bool { s.lock.Lock() defer s.lock.Unlock() @@ -121,7 +121,7 @@ func (s *Server) Timeout(reqId uint64) bool { return ok && ch == nil } -func (s *Server) Returned(reqId uint64) { +func (s *Server) returned(reqId uint64) { s.lock.Lock() defer s.lock.Unlock() @@ -130,6 +130,10 @@ func (s *Server) Returned(reqId uint64) { close(ch) } else { s.timeoutCount-- + if s.needTrigger && s.timeoutCount == 0 && !s.isDelayed() { + s.needTrigger = false + s.scheduler.triggerServer(s) + } } delete(s.sent, reqId) } @@ -152,11 +156,11 @@ func (s *Server) isDelayed() bool { case <-timer.C: s.lock.Lock() s.delayed = false - trigger := s.needTrigger && s.timeoutCount == 0 - s.lock.Unlock() - if trigger { - s.scheduler.serverTrigger(s) + if s.needTrigger && s.timeoutCount == 0 { + s.needTrigger = false + s.scheduler.triggerServer(s) } + s.lock.Unlock() case <-s.stopCh: timer.Stop() } diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go deleted file mode 100644 index 2fd901130c..0000000000 --- a/beacon/light/sync/head_sync.go +++ /dev/null @@ -1,105 +0,0 @@ -// Copyright 2023 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package sync - -import ( - "sync" - - "github.com/ethereum/go-ethereum/beacon/light" - "github.com/ethereum/go-ethereum/beacon/light/request" - "github.com/ethereum/go-ethereum/beacon/light/types" -) - -type signedHeadServer interface { - request.RequestServer - SignedHeads() []types.SignedHead -} - -type latestHeads struct { - heads map[uint64]types.SignedHead - oldestSlot uint64 -} - -type HeadSyncer struct { - lock sync.Mutex - headTracker *light.HeadTracker - chain *light.CommitteeChain - added, queued latestHeads - - SignedHeadTrigger request.ModuleTrigger -} - -func NewHeadSyncer(headTracker *light.HeadTracker, chain *light.CommitteeChain) *HeadSyncer { - return &HeadSyncer{ - headTracker: headTracker, - chain: chain, - } -} - -func (s *HeadSyncer) Process(servers []*request.Server) bool { - s.lock.Lock() - defer s.lock.Unlock() - - nextPeriod, ok := s.chain.NextSyncPeriod() - if !ok { - return true - } - for slot, head := range s.queued.heads { - if head.Header.SyncPeriod() <= nextPeriod { - delete(s.queued.heads, slot) - if s.added.add(head) && s.headTracker.Add(head) == nil { - s.SignedHeadTrigger.Trigger() - } - } - } - for _, server := range servers { - if hserver, ok := server.RequestServer.(signedHeadServer); ok { - heads := hserver.SignedHeads() - for _, head := range heads { - if head.Header.SyncPeriod() > nextPeriod { - s.queued.add(head) - } else if s.added.add(head) { - if s.headTracker.Add(head) == nil { - s.SignedHeadTrigger.Trigger() - } else { - hserver.Fail("received invalid signed head") - break - } - } - } - } - } - return true -} - -func (l *latestHeads) add(head types.SignedHead) bool { - if l.heads == nil { - l.heads = make(map[uint64]types.SignedHead) - l.oldestSlot = head.Header.Slot - } - if oldHead, ok := l.heads[head.Header.Slot]; ok { - if head.SignerCount() <= oldHead.SignerCount() { - return false - } - } - l.heads[head.Header.Slot] = head - for len(l.heads) > 4 { - delete(l.heads, l.oldestSlot) - l.oldestSlot++ - } - return true -} diff --git a/beacon/light/sync/head_updater.go b/beacon/light/sync/head_updater.go new file mode 100644 index 0000000000..e8ba60a169 --- /dev/null +++ b/beacon/light/sync/head_updater.go @@ -0,0 +1,92 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package sync + +import ( + "fmt" + "math" + "sync" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/light/types" +) + +type HeadUpdater struct { + headValidator *light.HeadValidator + chain *light.CommitteeChain + lock sync.Mutex + nextSyncPeriod uint64 + queuedHeads map[*request.Server][]types.SignedHead +} + +func NewHeadUpdater(headValidator *light.HeadValidator, chain *light.CommitteeChain) *HeadUpdater { + s := &HeadUpdater{ + headValidator: headValidator, + chain: chain, + nextSyncPeriod: math.MaxUint64, + queuedHeads: make(map[*request.Server][]types.SignedHead), + } + return s +} + +func (s *HeadUpdater) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + trigger("newUpdate", true) +} + +func (s *HeadUpdater) NewSignedHead(server *request.Server, signedHead types.SignedHead) { + nextPeriod, ok := s.chain.NextSyncPeriod() + if !ok || signedHead.Header.SyncPeriod() > nextPeriod { + s.lock.Lock() + s.queuedHeads[server] = append(s.queuedHeads[server], signedHead) //TODO protect against future period spam + s.lock.Unlock() + return + } + if err := s.headValidator.Add(signedHead); err != nil { + server.Fail(fmt.Sprintf("Invalid signed head: %v", err)) + } +} + +func (s *HeadUpdater) Process(env *request.Environment) { + s.lock.Lock() + defer s.lock.Unlock() + + nextPeriod, ok := s.chain.NextSyncPeriod() + if !ok || nextPeriod == s.nextSyncPeriod { + return + } + s.nextSyncPeriod = nextPeriod + + for server, queued := range s.queuedHeads { + j := len(queued) + for i := len(queued) - 1; i >= 0; i-- { + if signedHead := queued[i]; signedHead.Header.SyncPeriod() <= nextPeriod { + if err := s.headValidator.Add(signedHead); err != nil { + server.Fail(fmt.Sprintf("Invalid queued head: %v", err)) + } + } else { + j-- + if j != i { + queued[j] = queued[i] + } + } + } + if j != 0 { + s.queuedHeads[server] = queued[j:] + } + } +} diff --git a/beacon/light/sync/header_sync.go b/beacon/light/sync/header_sync.go new file mode 100644 index 0000000000..1bd5bf5086 --- /dev/null +++ b/beacon/light/sync/header_sync.go @@ -0,0 +1,182 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package sync + +import ( + "math" + "sync" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/common" +) + +type beaconHeaderServer interface { + request.RequestServer + RequestBeaconHeader(blockRoot common.Hash, response func(*types.Header)) +} + +type HeaderSync struct { + lock sync.Mutex + reqLock request.MultiLock + chain *light.LightChain + prefetch bool + targetHead, syncPtr types.Header + targetTailSlot uint64 + selfTrigger, chainTrigger *request.ModuleTrigger +} + +func NewHeaderSync(chain *light.LightChain, prefetch bool) *HeaderSync { + return &HeaderSync{ + chain: chain, + prefetch: prefetch, + targetTailSlot: math.MaxUint64, + } +} + +func (s *HeaderSync) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + s.selfTrigger = trigger("headerSync", true) + s.reqLock.Trigger = s.selfTrigger + trigger("validatedHead", true) + s.chainTrigger = trigger("headerChain", false) +} + +func (s *HeaderSync) SetTailTarget(targetTailSlot uint64) { + s.lock.Lock() + defer s.lock.Unlock() + + if targetTailSlot < s.targetTailSlot { + s.selfTrigger.Trigger() + } + s.targetTailSlot = targetTailSlot +} + +func (s *HeaderSync) Process(env *request.Environment) { + s.lock.Lock() + defer s.lock.Unlock() + + validatedHead := env.ValidatedHead() + if validatedHead != s.targetHead { + s.targetHead = validatedHead + s.syncPtr = validatedHead + s.chain.AddHeader(validatedHead) + } + if s.targetHead == (types.Header{}) { + return + } + chainHead, chainTail, chainInit := s.chain.HeaderRange() + if !chainInit { + s.chain.AddHeader(s.targetHead) + s.chain.SetChainHead(s.targetHead) + s.selfTrigger.Trigger() + s.chainTrigger.Trigger() + } + if s.prefetch { + if prefetchHead := env.PrefetchHead(); !s.chain.HasHeader(prefetchHead) { + s.tryPrefetchHead(env, prefetchHead) + } + } + if chainHead != s.targetHead && !s.trySyncHead(env, chainTail.Slot) { + // always prioritize syncing to the latest head, do not start tail sync until done + return + } + if s.targetTailSlot < chainTail.Slot { + s.trySyncTail(env, chainTail) + } +} + +// returns true if targetHead has been reached +func (s *HeaderSync) trySyncHead(env *request.Environment, chainTailSlot uint64) bool { + for { + if s.syncPtr.Slot <= chainTailSlot || s.chain.IsCanonical(s.syncPtr) { + s.chain.SetChainHead(s.targetHead) + s.chainTrigger.Trigger() + return true + } + if parent, err := s.chain.GetHeaderByHash(s.syncPtr.ParentRoot); err == nil { + s.syncPtr = parent + } else { + s.tryRequestHeader(env, s.syncPtr.ParentRoot, false) + return false + } + } +} + +func (s *HeaderSync) trySyncTail(env *request.Environment, syncTail types.Header) { + for { + if parent, err := s.chain.GetHeaderByHash(syncTail.ParentRoot); err == nil { + syncTail = parent + } else { + s.tryRequestHeader(env, syncTail.ParentRoot, false) + return + } + } +} + +func (s *HeaderSync) tryPrefetchHead(env *request.Environment, head common.Hash) { + if head != (common.Hash{}) && !s.chain.HasHeader(head) { + s.tryRequestHeader(env, head, true) + } +} + +func (s *HeaderSync) tryRequestHeader(env *request.Environment, blockRoot common.Hash, prefetch bool) { + if s.reqLock.CanRequest(blockRoot) { + env.TryRequest(headerRequest{ + HeaderSync: s, + blockRoot: blockRoot, + prefetch: prefetch, + }) + } +} + +type headerRequest struct { + *HeaderSync + blockRoot common.Hash + prefetch bool +} + +func (r headerRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { + if _, ok := server.RequestServer.(beaconHeaderServer); !ok { + return false, 0 + } + if !r.prefetch { + return true, 0 + } + _, headRoot := server.LatestHead() + return r.blockRoot == headRoot, 0 +} + +func (r headerRequest) SendTo(server *request.Server) { + reqId := r.reqLock.Send(server, r.blockRoot) + server.RequestServer.(beaconHeaderServer).RequestBeaconHeader(r.blockRoot, func(header *types.Header) { + r.lock.Lock() + defer r.lock.Unlock() + + r.reqLock.Returned(server, reqId, r.blockRoot) + if header == nil { + server.Fail("error retrieving beacon header") + return + } + _, oldChainTail, _ := r.chain.HeaderRange() + r.chain.AddHeader(*header) + _, chainTail, _ := r.chain.HeaderRange() + if chainTail != oldChainTail { + r.chainTrigger.Trigger() //TODO do this in a nicer way? + } + }) +} diff --git a/beacon/light/sync/state_sync.go b/beacon/light/sync/state_sync.go new file mode 100644 index 0000000000..7446967154 --- /dev/null +++ b/beacon/light/sync/state_sync.go @@ -0,0 +1,222 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package sync + +import ( + "math" + "sync" + "sync/atomic" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/beacon/merkle" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" +) + +type beaconStateServer interface { + request.RequestServer + BeaconStateTail() uint64 + RequestBeaconState(slot uint64, stateRoot common.Hash, format merkle.ProofFormat, response func(*merkle.MultiProof)) +} + +type StateSync struct { + lock sync.Mutex + reqLock request.MultiLock + chain *light.LightChain + prefetch bool + targetTailSlot uint64 + headSyncPossible uint32 + selfTrigger, headStateTrigger *request.ModuleTrigger +} + +func NewStateSync(chain *light.LightChain, prefetch bool) *StateSync { + return &StateSync{ + chain: chain, + prefetch: prefetch, + targetTailSlot: math.MaxUint64, + } +} + +func (s *StateSync) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + s.selfTrigger = trigger("stateSync", true) + s.reqLock.Trigger = s.selfTrigger + trigger("headerChain", true) + trigger("prefetchHeader", true) + s.headStateTrigger = trigger("headState", false) +} + +func (s *StateSync) SetTailTarget(targetTailSlot uint64) { + s.lock.Lock() + defer s.lock.Unlock() + + if targetTailSlot < s.targetTailSlot { + s.selfTrigger.Trigger() + } + s.targetTailSlot = targetTailSlot +} + +func (s *StateSync) Process(env *request.Environment) { + s.lock.Lock() + defer s.lock.Unlock() + + chainHead, chainTail, chainInit := s.chain.HeaderRange() + if !chainInit { + return + } + if s.prefetch { + if header, err := s.chain.GetHeaderByHash(env.PrefetchHead()); err == nil && !s.chain.HasStateProof(header) { + s.tryPrefetchHead(env, header) + } + } + stateHead, stateTail, stateInit := s.chain.StateProofRange() + if !stateInit { + s.tryRequestState(env, chainHead, false) + stateHead, stateTail = chainHead, chainHead + } else if stateHead != chainHead { + if !s.trySyncHead(env, stateHead) { + return + } + } + targetTailSlot := s.targetTailSlot + if chainTail.Slot > targetTailSlot { + targetTailSlot = chainTail.Slot + } + if targetTailSlot < stateTail.Slot { + s.trySyncTail(env, stateTail, targetTailSlot) + } +} + +func (s *StateSync) trySyncHead(env *request.Environment, stateHead types.Header) bool { + slot, lastBlockRoot := stateHead.Slot, stateHead.Hash() + for { + slot++ + header, err := s.chain.GetHeaderBySlot(slot) + if err == light.ErrEmptySlot { + continue + } + if err == light.ErrNotFound { + // no more canonical headers; head sync success + atomic.StoreUint32(&s.headSyncPossible, 1) + return true + } + if err != nil { + log.Error("Unexpected error during state head sync", "error", err) + return false + } + if header.ParentRoot != lastBlockRoot { + s.selfTrigger.Trigger() // reorg happened, stop and retry + return false + } + lastBlockRoot = header.Hash() + if !s.chain.HasStateProof(header) { + if sentOrLocked, tryLater := s.tryRequestState(env, header, false); !sentOrLocked { + if !tryLater { + atomic.StoreUint32(&s.headSyncPossible, 0) + } + return false + } + } + } +} + +func (s *StateSync) HeadSyncPossible() bool { + return atomic.LoadUint32(&s.headSyncPossible) == 1 +} + +func (s *StateSync) trySyncTail(env *request.Environment, stateTail types.Header, targetTailSlot uint64) { + for stateTail.Slot > targetTailSlot { + var err error + stateTail, err = s.chain.GetParent(stateTail) + if err != nil { + log.Error("Unexpected error during state tail sync", "error", err) + return + } + if !s.chain.HasStateProof(stateTail) { + if sentOrLocked, _ := s.tryRequestState(env, stateTail, false); !sentOrLocked { + return + } + } + } +} + +func (s *StateSync) tryPrefetchHead(env *request.Environment, head types.Header) { + s.tryRequestState(env, head, true) +} + +// tryRequestState starts a request for the partial beacon state belonging to the +// specified header if possible. It returns true if further requests should be +// attempted (either starting this one was successful or unnecessary because it +// is already locked by a recent attempt). +func (s *StateSync) tryRequestState(env *request.Environment, header types.Header, prefetch bool) (sentOrLocked, tryLater bool) { + if !s.reqLock.CanRequest(header.StateRoot) { + return true, false + } + req := stateRequest{ + StateSync: s, + header: header, + prefetch: prefetch, + } + sentOrLocked, _ = env.TryRequest(req) + if !sentOrLocked { + tryLater = env.CanRequestLater(req) + } + return +} + +type stateRequest struct { + *StateSync + header types.Header + prefetch bool +} + +func (r stateRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { + if rs, ok := server.RequestServer.(beaconStateServer); !ok || r.header.Slot < rs.BeaconStateTail() { + return false, 0 + } + if !r.prefetch { + return true, 0 + } + _, headRoot := server.LatestHead() + return r.header.Hash() == headRoot, 0 +} + +func (r stateRequest) SendTo(server *request.Server) { + reqId := r.reqLock.Send(server, r.header.StateRoot) + server.RequestServer.(beaconStateServer).RequestBeaconState(r.header.Slot, r.header.StateRoot, r.chain.StateProofFormat(r.header), func(proof *merkle.MultiProof) { + r.lock.Lock() + defer r.lock.Unlock() + + r.reqLock.Returned(server, reqId, r.header.StateRoot) + if proof == nil { + //server.Fail("error retrieving beacon state proof") + return + } + oldStateHead, _, _ := r.chain.StateProofRange() + if err := r.chain.AddStateProof(r.header, *proof); err != nil { + server.Fail("invalid beacon state proof: " + err.Error()) + return + } + chainHead, _, _ := r.chain.HeaderRange() + stateHead, _, _ := r.chain.StateProofRange() + if stateHead == chainHead && oldStateHead != chainHead { + r.headStateTrigger.Trigger() + } + }) +} diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index 5d00c76829..497d51a096 100644 --- a/beacon/light/sync/update_sync.go +++ b/beacon/light/sync/update_sync.go @@ -35,14 +35,14 @@ type checkpointInitServer interface { } type CheckpointInit struct { - request.SingleLock lock sync.Mutex + reqLock request.SingleLock chain *light.CommitteeChain cs *light.CheckpointStore checkpointHash common.Hash initialized bool - InitTrigger request.ModuleTrigger + initTrigger *request.ModuleTrigger } func NewCheckpointInit(chain *light.CommitteeChain, cs *light.CheckpointStore, checkpointHash common.Hash) *CheckpointInit { @@ -53,129 +53,149 @@ func NewCheckpointInit(chain *light.CommitteeChain, cs *light.CheckpointStore, c } } -func (s *CheckpointInit) Process(servers []*request.Server) bool { +func (s *CheckpointInit) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + s.reqLock.Trigger = trigger("checkpointInit", true) + s.initTrigger = trigger("committeeChainInit", false) +} + +func (s *CheckpointInit) Process(env *request.Environment) { s.lock.Lock() defer s.lock.Unlock() if s.initialized { - return false + return } if checkpoint := s.cs.Get(s.checkpointHash); checkpoint != nil { checkpoint.InitChain(s.chain) s.initialized = true - s.InitTrigger.Trigger() - return false + s.initTrigger.Trigger() + return } - srv := request.SelectServer(servers, func(server *request.Server) uint64 { - if cserver, ok := server.RequestServer.(checkpointInitServer); ok && cserver.CanRequestBootstrap() && s.CanSend(server) { - return 1 - } - return 0 - }) - if srv == nil { - return true + if s.reqLock.CanRequest() { + env.TryRequest(checkpointRequest{ + CheckpointInit: s, + checkpointHash: s.checkpointHash, + }) } - reqId, ok := s.TrySend(srv) - if !ok { - return true - } - server := srv.RequestServer.(checkpointInitServer) - server.RequestBootstrap(s.checkpointHash, func(checkpoint *light.CheckpointData) { - s.lock.Lock() - defer s.lock.Unlock() +} - s.Returned(srv, reqId) +type checkpointRequest struct { + *CheckpointInit + checkpointHash common.Hash +} + +func (r checkpointRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { + if cs, ok := server.RequestServer.(checkpointInitServer); !ok || !cs.CanRequestBootstrap() { + return false, 0 + } + return true, 0 +} + +func (r checkpointRequest) SendTo(server *request.Server) { + reqId := r.reqLock.Send(server) + server.RequestServer.(checkpointInitServer).RequestBootstrap(r.checkpointHash, func(checkpoint *light.CheckpointData) { + r.lock.Lock() + defer r.lock.Unlock() + + r.reqLock.Returned(server, reqId) if checkpoint == nil || !checkpoint.Validate() { server.Fail("error retrieving checkpoint data") return } - checkpoint.InitChain(s.chain) - s.cs.Store(checkpoint) - s.initialized = true - s.InitTrigger.Trigger() + checkpoint.InitChain(r.chain) + r.cs.Store(checkpoint) + r.initialized = true + r.initTrigger.Trigger() }) - return true } -type forwardUpdateServer interface { +type updateServer interface { request.RequestServer UpdateRange() types.PeriodRange RequestUpdates(first, count uint64, response func([]*types.LightClientUpdate, []*types.SerializedCommittee)) } -type ForwardUpdateSyncer struct { - request.SingleLock - lock sync.Mutex - chain *light.CommitteeChain +type ForwardUpdateSync struct { + lock sync.Mutex + reqLock request.SingleLock + chain *light.CommitteeChain - NewUpdateTrigger request.ModuleTrigger + newUpdateTrigger *request.ModuleTrigger } -func NewForwardUpdateSyncer(chain *light.CommitteeChain) *ForwardUpdateSyncer { - return &ForwardUpdateSyncer{chain: chain} +func NewForwardUpdateSync(chain *light.CommitteeChain) *ForwardUpdateSync { + return &ForwardUpdateSync{chain: chain} } -func (s *ForwardUpdateSyncer) Process(servers []*request.Server) bool { +func (s *ForwardUpdateSync) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + s.reqLock.Trigger = trigger("forwardUpdateSync", true) + trigger("committeeChainInit", true) + trigger("validatedHead", true) + s.newUpdateTrigger = trigger("newUpdate", true) +} + +func (s *ForwardUpdateSync) Process(env *request.Environment) { s.lock.Lock() defer s.lock.Unlock() first, ok := s.chain.NextSyncPeriod() if !ok { - return true + return } - srv := request.SelectServer(servers, func(server *request.Server) uint64 { - if fserver, ok := server.RequestServer.(forwardUpdateServer); ok && s.CanSend(server) { - updateRange := fserver.UpdateRange() - if first < updateRange.First { - return 0 - } - return updateRange.AfterLast - } - return 0 + env.TryRequest(updateRequest{ + ForwardUpdateSync: s, + first: first, }) - if srv == nil { - return true +} + +type updateRequest struct { + *ForwardUpdateSync + first uint64 +} + +func (r updateRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { + if us, ok := server.RequestServer.(updateServer); ok { + if updateRange := us.UpdateRange(); updateRange.Includes(r.first) { + return true, updateRange.AfterLast + } } - server := srv.RequestServer.(forwardUpdateServer) - updateRange := server.UpdateRange() - if updateRange.AfterLast <= first { - return true - } - reqId, ok := s.TrySend(srv) - if !ok { - return true - } - count := updateRange.AfterLast - first - if count > maxUpdateRequest { //TODO const + return false, 0 +} + +func (r updateRequest) SendTo(server *request.Server) { + us := server.RequestServer.(updateServer) + updateRange := us.UpdateRange() + count := updateRange.AfterLast - r.first + if count > maxUpdateRequest { count = maxUpdateRequest } - server.RequestUpdates(first, count, func(updates []*types.LightClientUpdate, committees []*types.SerializedCommittee) { - s.lock.Lock() - defer s.lock.Unlock() + reqId := r.reqLock.Send(server) + us.RequestUpdates(r.first, count, func(updates []*types.LightClientUpdate, committees []*types.SerializedCommittee) { + r.lock.Lock() + defer r.lock.Unlock() - s.Returned(srv, reqId) + r.reqLock.Returned(server, reqId) if len(updates) != int(count) || len(committees) != int(count) { server.Fail("wrong number of updates received") return } for i, update := range updates { - if update.Header.SyncPeriod() != first+uint64(i) { + if update.Header.SyncPeriod() != r.first+uint64(i) { server.Fail("update with wrong sync period received") return } - if err := s.chain.InsertUpdate(update, committees[i]); err != nil { + if err := r.chain.InsertUpdate(update, committees[i]); err != nil { if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg { server.Fail("invalid update received") } else { log.Error("Unexpected InsertUpdate error", "error", err) } - if i != 0 { - s.NewUpdateTrigger.Trigger() + if i != 0 { // some updates were added + r.newUpdateTrigger.Trigger() } return } } - s.NewUpdateTrigger.Trigger() + r.newUpdateTrigger.Trigger() }) - return true } diff --git a/beacon/merkle/binary_merkle.go b/beacon/merkle/binary_merkle.go index cf54a3a2e2..54cefaa997 100644 --- a/beacon/merkle/binary_merkle.go +++ b/beacon/merkle/binary_merkle.go @@ -69,6 +69,16 @@ type ProofFormat interface { Children() (left, right ProofFormat) // either both or neither should be nil } +// IsEqual returns true if the two formats are the same +func IsEqual(a, b ProofFormat) bool { + al, ar := a.Children() + bl, br := b.Children() + if al == nil || bl == nil { + return al == nil && bl == nil + } + return IsEqual(al, bl) && IsEqual(ar, br) +} + // ProofReader allows traversing and reading a tree structure or a subset of it. // Note: the hash of each traversed node is always requested. If the internal // hash is not available then subtrees are always traversed (first left, then right). diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go new file mode 100755 index 0000000000..a27094ef2b --- /dev/null +++ b/cmd/blsync/block_sync.go @@ -0,0 +1,307 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package main + +import ( + "fmt" + "math/big" + "sync" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + lsync "github.com/ethereum/go-ethereum/beacon/light/sync" + "github.com/ethereum/go-ethereum/beacon/light/types" + "github.com/ethereum/go-ethereum/beacon/merkle" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/lru" + ctypes "github.com/ethereum/go-ethereum/core/types" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/trie" + "github.com/holiman/uint256" + "github.com/protolambda/zrnt/eth2/beacon/capella" + "github.com/protolambda/zrnt/eth2/configs" + "github.com/protolambda/ztyp/tree" +) + +const reverseSyncHeaders = 128 + +type beaconBlockServer interface { + request.RequestServer + RequestBeaconBlock(blockRoot common.Hash, response func(*capella.BeaconBlock)) +} + +type beaconStateServer interface { + request.RequestServer + BeaconStateTail() uint64 + RequestBeaconState(slot uint64, stateRoot common.Hash, format merkle.ProofFormat, response func(merkle.MultiProof)) +} + +type beaconBlockSync struct { + lock sync.Mutex + reqLock request.MultiLock + recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock] + + headUpdater *lsync.HeadUpdater + lightChain *light.LightChain + validatedHead types.Header + headBlock *capella.BeaconBlock // belongs to validatedHead (or nil) + headBlockTrigger, prefetchHeaderTrigger *request.ModuleTrigger +} + +func newBeaconBlockSyncer(lightChain *light.LightChain) *beaconBlockSync { + return &beaconBlockSync{ + lightChain: lightChain, + recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10), + } +} + +func (s *beaconBlockSync) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + s.reqLock.Trigger = trigger("beaconBlockSync", true) + trigger("validatedHead", true) + s.headBlockTrigger = trigger("headBlock", false) + s.prefetchHeaderTrigger = trigger("prefetchHeader", false) +} + +func (s *beaconBlockSync) Process(env *request.Environment) { + s.lock.Lock() + defer s.lock.Unlock() + + validatedHead := env.ValidatedHead() + if validatedHead == (types.Header{}) { + return + } + if validatedHead != s.validatedHead { + s.validatedHead = validatedHead + s.headBlock = nil + if block, ok := s.recentBlocks.Get(validatedHead.Hash()); ok { + s.headBlock = block + s.headBlockTrigger.Trigger() + } + } + + if s.headBlock == nil && s.validatedHead != (types.Header{}) { + if !s.tryRequestBlock(env, s.validatedHead.Hash(), false) { + return + } + } + + prefetchHead := env.PrefetchHead() + if _, ok := s.recentBlocks.Get(prefetchHead); !ok { + s.tryRequestBlock(env, prefetchHead, true) + } +} + +func (s *beaconBlockSync) getHeadBlock() *capella.BeaconBlock { + s.lock.Lock() + defer s.lock.Unlock() + + return s.headBlock +} + +func (s *beaconBlockSync) tryRequestBlock(env *request.Environment, blockRoot common.Hash, prefetch bool) bool { + if !s.reqLock.CanRequest(blockRoot) { + return true + } + _, tryMore := env.TryRequest(blockRequest{ + beaconBlockSync: s, + blockRoot: blockRoot, + prefetch: prefetch, + }) + return tryMore +} + +type blockRequest struct { + *beaconBlockSync + blockRoot common.Hash + prefetch bool +} + +func (r blockRequest) CanSendTo(server *request.Server) (canSend bool, priority uint64) { + if _, ok := server.RequestServer.(beaconBlockServer); !ok { + return false, 0 + } + if !r.prefetch { + return true, 0 + } + _, headRoot := server.LatestHead() + return r.blockRoot == headRoot, 0 +} + +func (r blockRequest) SendTo(server *request.Server) { + reqId := r.reqLock.Send(server, r.blockRoot) + server.RequestServer.(beaconBlockServer).RequestBeaconBlock(r.blockRoot, func(block *capella.BeaconBlock) { + r.lock.Lock() + defer r.lock.Unlock() + + r.reqLock.Returned(server, reqId, r.blockRoot) + if block == nil { + server.Fail("error retrieving beacon block") + return + } + r.recentBlocks.Add(r.blockRoot, block) + if !r.lightChain.HasHeader(r.blockRoot) { + r.lightChain.AddHeader(types.Header{ + Slot: uint64(block.Slot), + ProposerIndex: uint64(block.ProposerIndex), + ParentRoot: common.Hash(block.ParentRoot), + StateRoot: common.Hash(block.StateRoot), + BodyRoot: common.Hash(block.Body.HashTreeRoot(configs.Mainnet, tree.GetHashFn())), + }) + r.prefetchHeaderTrigger.Trigger() + } + if r.validatedHead.Hash() == r.blockRoot { + r.headBlock = block + r.headBlockTrigger.Trigger() + } + }) +} + +func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) { + payload := &beaconBlock.Body.ExecutionPayload + txs := make([]*ctypes.Transaction, len(payload.Transactions)) + for i, opaqueTx := range payload.Transactions { + var tx ctypes.Transaction + if err := tx.UnmarshalBinary(opaqueTx); err != nil { + return nil, fmt.Errorf("failed to parse tx %d: %v", i, err) + } + txs[i] = &tx + } + withdrawals := make([]*ctypes.Withdrawal, len(payload.Withdrawals)) + for i, w := range payload.Withdrawals { + withdrawals[i] = &ctypes.Withdrawal{ + Index: uint64(w.Index), + Validator: uint64(w.ValidatorIndex), + Address: common.Address(w.Address), + Amount: uint64(w.Amount), + } + } + wroot := ctypes.DeriveSha(ctypes.Withdrawals(withdrawals), trie.NewStackTrie(nil)) + execHeader := &ctypes.Header{ + ParentHash: common.Hash(payload.ParentHash), + UncleHash: ctypes.EmptyUncleHash, + Coinbase: common.Address(payload.FeeRecipient), + Root: common.Hash(payload.StateRoot), + TxHash: ctypes.DeriveSha(ctypes.Transactions(txs), trie.NewStackTrie(nil)), + ReceiptHash: common.Hash(payload.ReceiptsRoot), + Bloom: ctypes.Bloom(payload.LogsBloom), + Difficulty: common.Big0, + Number: new(big.Int).SetUint64(uint64(payload.BlockNumber)), + GasLimit: uint64(payload.GasLimit), + GasUsed: uint64(payload.GasUsed), + Time: uint64(payload.Timestamp), + Extra: []byte(payload.ExtraData), + MixDigest: common.Hash(payload.PrevRandao), // reused in merge + Nonce: ctypes.BlockNonce{}, // zero + BaseFee: (*uint256.Int)(&payload.BaseFeePerGas).ToBig(), + WithdrawalsHash: &wroot, + } + execBlock := ctypes.NewBlockWithHeader(execHeader).WithBody(txs, nil).WithWithdrawals(withdrawals) + if execBlockHash := execBlock.Hash(); execBlockHash != common.Hash(payload.BlockHash) { + return nil, fmt.Errorf("Sanity check failed, payload hash does not match (expected %x, got %x)", common.Hash(payload.BlockHash), execBlockHash) + } + return execBlock, nil +} + +type engineApiUpdater struct { + client *rpc.Client + lock sync.Mutex + lastHead common.Hash + headerSync *lsync.HeaderSync + stateSync *lsync.StateSync + blockSync *beaconBlockSync + chain *light.LightChain + updating bool + selfTrigger *request.ModuleTrigger +} + +func (s *engineApiUpdater) SetupTriggers(trigger func(id string, subscribe bool) *request.ModuleTrigger) { + trigger("headBlock", true) + trigger("headState", true) + s.selfTrigger = trigger("engineApiUpdater", true) +} + +func (s *engineApiUpdater) Process(env *request.Environment) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.updating { + return + } + headBlock := s.blockSync.getHeadBlock() + if headBlock == nil { + return + } + headRoot := common.Hash(headBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn())) + if headRoot == s.lastHead { + return + } + if headBlock.Slot > reverseSyncHeaders { + s.headerSync.SetTailTarget(uint64(headBlock.Slot) - reverseSyncHeaders) + } else { + s.headerSync.SetTailTarget(0) + } + + head, err := s.chain.GetHeaderByHash(headRoot) + if err != nil { + return + } + + var finalizedExecRoot common.Hash + if state, err := s.chain.GetStateProof(head); err == nil { + finalizedRoot := common.Hash(state.Values[finalizedBlockIndex]) + if finalized, err := s.chain.GetHeaderByHash(finalizedRoot); err == nil { + if finalizedState, err := s.chain.GetStateProof(finalized); err == nil { + finalizedExecRoot = common.Hash(finalizedState.Values[execBlockIndex]) + } + } + } else { + if s.stateSync.HeadSyncPossible() { + return + } + } + + s.lastHead = headRoot + execBlock, err := getExecBlock(headBlock) + if err != nil { + log.Error("Error extracting execution block from validated beacon block", "error", err) + return + } + execRoot := execBlock.Hash() + if s.client == nil { // dry run, no engine API specified + log.Info("New execution block retrieved", "block number", execBlock.NumberU64(), "block hash", execRoot, "finalized block hash", finalizedExecRoot) + } else { + s.updating = true + go func() { + if status, err := callNewPayloadV1(s.client, execBlock); err == nil { + log.Info("Successful NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "status", status) + } else { + log.Error("Failed NewPayload", "block number", execBlock.NumberU64(), "block hash", execRoot, "error", err) + } + if status, err := callForkchoiceUpdatedV1(s.client, execRoot, finalizedExecRoot); err == nil { + log.Info("Successful ForkchoiceUpdated", "head", execRoot, "finalized", finalizedExecRoot, "status", status) + } else { + log.Error("Failed ForkchoiceUpdated", "head", execRoot, "finalized", finalizedExecRoot, "error", err) + } + s.lock.Lock() + s.updating = false + s.selfTrigger.Trigger() + s.lock.Unlock() + }() + } +} diff --git a/cmd/blsync/main.go b/cmd/blsync/main.go index 69176c7f25..6b158cd9ac 100644 --- a/cmd/blsync/main.go +++ b/cmd/blsync/main.go @@ -33,7 +33,6 @@ import ( "github.com/ethereum/go-ethereum/beacon/params" "github.com/ethereum/go-ethereum/cmd/utils" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/lru" "github.com/ethereum/go-ethereum/common/mclock" ctypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb/memorydb" @@ -97,34 +96,55 @@ func blsync(ctx *cli.Context) error { customHeader[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1]) } + // create data structures var ( - beaconApi = api.NewBeaconLightApi(ctx.String(utils.BeaconApiFlag.Name), customHeader) db = memorydb.New() threshold = ctx.Int(utils.BeaconThresholdFlag.Name) committeeChain = light.NewCommitteeChain(db, chainConfig.Forks, threshold, !ctx.Bool(utils.BeaconNoFilterFlag.Name), light.BLSVerifier{}, &mclock.System{}, func() int64 { return time.Now().UnixNano() }) checkpointStore = light.NewCheckpointStore(db, committeeChain) - headTracker = light.NewHeadTracker(committeeChain) - scheduler = request.NewScheduler() + headValidator = light.NewHeadValidator(committeeChain) + lightChain = light.NewLightChain(db, stateProofFormat) ) committeeChain.SetGenesisData(chainConfig.GenesisData) + headUpdater := sync.NewHeadUpdater(headValidator, committeeChain) + headTracker := request.NewHeadTracker(headUpdater.NewSignedHead) + headValidator.Subscribe(threshold, func(signedHead types.SignedHead) { + headTracker.SetValidatedHead(signedHead.Header) + }) + // create sync modules checkpointInit := sync.NewCheckpointInit(committeeChain, checkpointStore, chainConfig.Checkpoint) - forwardSync := sync.NewForwardUpdateSyncer(committeeChain) - headSync := sync.NewHeadSyncer(headTracker, committeeChain) + forwardSync := sync.NewForwardUpdateSync(committeeChain) + headerSync := sync.NewHeaderSync(lightChain, false) + stateSync := sync.NewStateSync(lightChain, true) + beaconBlockSync := newBeaconBlockSyncer(lightChain) + engineApiUpdater := &engineApiUpdater{ //TODO constructor + client: makeRPCClient(ctx), + headerSync: headerSync, + stateSync: stateSync, + blockSync: beaconBlockSync, + chain: lightChain, + } + + // set up sync modules and triggers + scheduler := request.NewScheduler(headTracker) + headTracker.SetupTriggers(scheduler.GetModuleTrigger) scheduler.RegisterModule(checkpointInit) scheduler.RegisterModule(forwardSync) - scheduler.RegisterModule(headSync) - scheduler.AddTriggers(forwardSync, []*request.ModuleTrigger{&checkpointInit.InitTrigger, &forwardSync.NewUpdateTrigger, &headSync.SignedHeadTrigger}) - scheduler.AddTriggers(headSync, []*request.ModuleTrigger{&forwardSync.NewUpdateTrigger}) - - syncer := &execSyncer{ - api: beaconApi, - client: makeRPCClient(ctx), - execRootCache: lru.NewCache[common.Hash, common.Hash](1000), - } - headTracker.Subscribe(threshold, syncer.newHead) + scheduler.RegisterModule(headUpdater) + scheduler.RegisterModule(beaconBlockSync) + scheduler.RegisterModule(engineApiUpdater) + scheduler.RegisterModule(stateSync) + scheduler.RegisterModule(headerSync) + // start scheduler.Start() - scheduler.RegisterServer(api.NewSyncServer(beaconApi)) + stateSync.SetTailTarget(0) + // register server(s) + for _, url := range utils.SplitAndTrim(ctx.String(utils.BeaconApiFlag.Name)) { + beaconApi := api.NewBeaconLightApi(url, customHeader) + scheduler.RegisterServer(api.NewSyncServer(beaconApi)) + } + // run until stopped <-ctx.Done() scheduler.Stop() return nil @@ -150,85 +170,3 @@ func callForkchoiceUpdatedV1(client *rpc.Client, headHash, finalizedHash common. cancel() return resp.PayloadStatus.Status, err } - -type execSyncer struct { - api *api.BeaconLightApi - sub *api.StateProofSub - client *rpc.Client - execRootCache *lru.Cache[common.Hash, common.Hash] // beacon block root -> execution block root -} - -// newHead fetches state proofs to determine the execution block root and calls -// the engine API if specified -func (e *execSyncer) newHead(signedHead types.SignedHead) { - head := signedHead.Header - log.Info("Received new beacon head", "slot", head.Slot, "blockRoot", head.Hash()) - block, err := e.api.GetExecutionPayload(head) - if err != nil { - log.Error("Error fetching execution payload from beacon API", "error", err) - return - } - blockRoot := block.Hash() - var finalizedExecRoot common.Hash - if e.sub == nil { - if sub, err := e.api.SubscribeStateProof(stateProofFormat, 0, 1); err == nil { - log.Info("Successfully created beacon state subscription") - e.sub = sub - } else { - log.Error("Failed to create beacon state subscription", "error", err) - return - } - } - proof, err := e.sub.Get(head.StateRoot) - if err == nil { - var ( - execBlockRoot = common.Hash(proof.Values[execBlockIndex]) - finalizedBeaconRoot = common.Hash(proof.Values[finalizedBlockIndex]) - beaconRoot = head.Hash() - ) - e.execRootCache.Add(beaconRoot, execBlockRoot) - if blockRoot != execBlockRoot { - log.Error("Execution payload block hash does not match value in beacon state", "expected", execBlockRoot, "got", block.Hash()) - return - } - if _, ok := e.execRootCache.Get(head.ParentRoot); !ok { - e.fetchExecRoots(head.ParentRoot) - } - finalizedExecRoot, _ = e.execRootCache.Get(finalizedBeaconRoot) - } else if err != api.ErrNotFound { - log.Error("Error fetching state proof from beacon API", "error", err) - } - if e.client == nil { // dry run, no engine API specified - log.Info("New execution block retrieved", "block number", block.NumberU64(), "block hash", blockRoot, "finalized block hash", finalizedExecRoot) - return - } - if status, err := callNewPayloadV1(e.client, block); err == nil { - log.Info("Successful NewPayload", "block number", block.NumberU64(), "block hash", blockRoot, "status", status) - } else { - log.Error("Failed NewPayload", "block number", block.NumberU64(), "block hash", blockRoot, "error", err) - } - if status, err := callForkchoiceUpdatedV1(e.client, blockRoot, finalizedExecRoot); err == nil { - log.Info("Successful ForkchoiceUpdated", "head", blockRoot, "finalized", finalizedExecRoot, "status", status) - } else { - log.Error("Failed ForkchoiceUpdated", "head", blockRoot, "finalized", finalizedExecRoot, "error", err) - } -} - -func (e *execSyncer) fetchExecRoots(blockRoot common.Hash) { - for maxFetch := 256; maxFetch > 0; maxFetch-- { - header, err := e.api.GetHeader(blockRoot) - if err != nil { - break - } - proof, err := e.sub.Get(header.StateRoot) - if err != nil { - // exit silently because we expect running into an error when parent is unknown - break - } - e.execRootCache.Add(header.Hash(), common.Hash(proof.Values[execBlockIndex])) - if _, ok := e.execRootCache.Get(header.ParentRoot); ok { - break - } - blockRoot = header.ParentRoot - } -}