diff --git a/beacon/light/api/api_server.go b/beacon/light/api/api_server.go new file mode 100755 index 0000000000..a925bd3536 --- /dev/null +++ b/beacon/light/api/api_server.go @@ -0,0 +1,93 @@ +// 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 api + +import ( + "sync/atomic" + + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/light/sync" + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" +) + +type ApiServer struct { + api *BeaconLightApi + eventCallback func(event request.Event) + unsubscribe func() + lastId uint64 +} + +func NewApiServer(api *BeaconLightApi) *ApiServer { + return &ApiServer{api: api} +} + +func (s *ApiServer) Subscribe(eventCallback func(event request.Event)) { + s.eventCallback = eventCallback + s.unsubscribe = s.api.StartHeadListener(func(slot uint64, blockRoot common.Hash) { + eventCallback(request.Event{Type: sync.EvNewHead, Data: types.HeadInfo{Slot: slot, BlockRoot: blockRoot}}) + }, func(head types.SignedHeader) { + eventCallback(request.Event{Type: sync.EvNewSignedHead, Data: head}) + }, func(err error) { + log.Warn("Head event stream error", "err", err) + }) +} + +func (s *ApiServer) SendRequest(req request.Request) request.ID { + id := request.ID(atomic.AddUint64(&s.lastId, 1)) + go func() { + var resp request.Response + switch data := req.(type) { + case sync.ReqUpdates: + if updates, committees, err := s.api.GetBestUpdatesAndCommittees(data.FirstPeriod, data.Count); err == nil { + resp = sync.RespUpdates{Updates: updates, Committees: committees} + } + /*case sync.ReqOptimisticHead: + if signedHead, err := s.api.GetOptimisticHeadUpdate(); err == nil { + resp = signedHead + }*/ //TODO ??? + case sync.ReqHeader: + if header, err := s.api.GetHeader(common.Hash(data)); err == nil { + resp = header + } + case sync.ReqCheckpointData: + if bootstrap, err := s.api.GetCheckpointData(common.Hash(data)); err == nil { + resp = bootstrap + } + case sync.ReqBeaconBlock: + if block, err := s.api.GetBeaconBlock(common.Hash(data)); err == nil { + resp = block + } + default: + } + if resp != nil { + s.eventCallback(request.Event{Type: request.EvResponse, Data: request.IdAndResponse{ID: id, Response: resp}}) + } else { + s.eventCallback(request.Event{Type: request.EvFail, Data: id}) + } + }() + return id +} + +// Note: UnsubscribeHeads should not be called concurrently with SubscribeHeads +func (s *ApiServer) Unsubscribe() { + if s.unsubscribe != nil { + s.unsubscribe() + s.unsubscribe = nil + } +} diff --git a/beacon/light/api/light_api.go b/beacon/light/api/light_api.go new file mode 100755 index 0000000000..10be896028 --- /dev/null +++ b/beacon/light/api/light_api.go @@ -0,0 +1,420 @@ +// Copyright 2022 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 detaiapi. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package api + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "time" + + "github.com/donovanhide/eventsource" + "github.com/ethereum/go-ethereum/beacon/merkle" + "github.com/ethereum/go-ethereum/beacon/params" + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/protolambda/zrnt/eth2/beacon/capella" + "github.com/protolambda/zrnt/eth2/configs" + "github.com/protolambda/ztyp/tree" +) + +var ( + ErrNotFound = errors.New("404 Not Found") + ErrInternal = errors.New("500 Internal Server Error") +) + +type CommitteeUpdate struct { + Version string + Update types.LightClientUpdate + NextSyncCommittee types.SerializedSyncCommittee +} + +// See data structure definition here: +// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientupdate +type committeeUpdateJson struct { + Version string `json:"version"` + Data committeeUpdateData `json:"data"` +} + +type committeeUpdateData struct { + Header jsonBeaconHeader `json:"attested_header"` + NextSyncCommittee types.SerializedSyncCommittee `json:"next_sync_committee"` + NextSyncCommitteeBranch merkle.Values `json:"next_sync_committee_branch"` + FinalizedHeader *jsonBeaconHeader `json:"finalized_header,omitempty"` + FinalityBranch merkle.Values `json:"finality_branch,omitempty"` + SyncAggregate types.SyncAggregate `json:"sync_aggregate"` + SignatureSlot common.Decimal `json:"signature_slot"` +} + +type jsonBeaconHeader struct { + Beacon types.Header `json:"beacon"` +} + +// UnmarshalJSON unmarshals from JSON. +func (u *CommitteeUpdate) UnmarshalJSON(input []byte) error { + var dec committeeUpdateJson + if err := json.Unmarshal(input, &dec); err != nil { + return err + } + u.Version = dec.Version + u.NextSyncCommittee = dec.Data.NextSyncCommittee + u.Update = types.LightClientUpdate{ + AttestedHeader: types.SignedHeader{ + Header: dec.Data.Header.Beacon, + Signature: dec.Data.SyncAggregate, + SignatureSlot: uint64(dec.Data.SignatureSlot), + }, + NextSyncCommitteeRoot: u.NextSyncCommittee.Root(), + NextSyncCommitteeBranch: dec.Data.NextSyncCommitteeBranch, + FinalityBranch: dec.Data.FinalityBranch, + } + if dec.Data.FinalizedHeader != nil { + u.Update.FinalizedHeader = &dec.Data.FinalizedHeader.Beacon + } + return nil +} + +// fetcher is an interface useful for debug-harnessing the http api. +type fetcher interface { + Do(req *http.Request) (*http.Response, error) +} + +// BeaconLightApi requests light client information from a beacon node REST API. +// Note: all required API endpoints are currently only implemented by Lodestar. +type BeaconLightApi struct { + url string + client fetcher + customHeaders map[string]string +} + +func NewBeaconLightApi(url string, customHeaders map[string]string) *BeaconLightApi { + return &BeaconLightApi{ + url: url, + client: &http.Client{ + Timeout: time.Second * 10, + }, + customHeaders: customHeaders, + } +} + +func (api *BeaconLightApi) httpGet(path string) ([]byte, error) { + req, err := http.NewRequest("GET", api.url+path, nil) + if err != nil { + return nil, err + } + for k, v := range api.customHeaders { + req.Header.Set(k, v) + } + resp, err := api.client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + switch resp.StatusCode { + case 200: + return io.ReadAll(resp.Body) + case 404: + return nil, ErrNotFound + case 500: + return nil, ErrInternal + default: + return nil, fmt.Errorf("Unexpected error from API endpoint \"%s\": status code %d", path, resp.StatusCode) + } +} + +func (api *BeaconLightApi) httpGetf(format string, params ...any) ([]byte, error) { + return api.httpGet(fmt.Sprintf(format, params...)) +} + +// GetBestUpdateAndCommittee fetches and validates LightClientUpdate for given +// period and full serialized committee for the next period (committee root hash +// equals update.NextSyncCommitteeRoot). +// Note that the results are validated but the update signature should be verified +// by the caller as its validity depends on the update chain. +// TODO handle valid partial results +func (api *BeaconLightApi) GetBestUpdatesAndCommittees(firstPeriod, count uint64) ([]*types.LightClientUpdate, []*types.SerializedSyncCommittee, error) { + resp, err := api.httpGetf("/eth/v1/beacon/light_client/updates?start_period=%d&count=%d", firstPeriod, count) + if err != nil { + return nil, nil, err + } + + var data []CommitteeUpdate + if err := json.Unmarshal(resp, &data); err != nil { + return nil, nil, err + } + if len(data) != int(count) { + return nil, nil, errors.New("invalid number of committee updates") + } + updates := make([]*types.LightClientUpdate, int(count)) + committees := make([]*types.SerializedSyncCommittee, int(count)) + for i, d := range data { + if d.Update.AttestedHeader.Header.SyncPeriod() != firstPeriod+uint64(i) { + return nil, nil, errors.New("wrong committee update header period") + } + if err := d.Update.Validate(); err != nil { + return nil, nil, err + } + if d.NextSyncCommittee.Root() != d.Update.NextSyncCommitteeRoot { + return nil, nil, errors.New("wrong sync committee root") + } + updates[i], committees[i] = new(types.LightClientUpdate), new(types.SerializedSyncCommittee) + *updates[i], *committees[i] = d.Update, d.NextSyncCommittee + } + return updates, committees, nil +} + +// GetOptimisticHeadUpdate fetches a signed header based on the latest available +// optimistic update. Note that the signature should be verified by the caller +// as its validity depends on the update chain. +// +// See data structure definition here: +// https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientoptimisticupdate +func (api *BeaconLightApi) GetOptimisticHeadUpdate() (types.SignedHeader, error) { + resp, err := api.httpGet("/eth/v1/beacon/light_client/optimistic_update") + if err != nil { + return types.SignedHeader{}, err + } + return decodeOptimisticHeadUpdate(resp) +} + +func decodeOptimisticHeadUpdate(enc []byte) (types.SignedHeader, error) { + var data struct { + Data struct { + Header jsonBeaconHeader `json:"attested_header"` + Aggregate types.SyncAggregate `json:"sync_aggregate"` + SignatureSlot common.Decimal `json:"signature_slot"` + } `json:"data"` + } + if err := json.Unmarshal(enc, &data); err != nil { + return types.SignedHeader{}, err + } + if data.Data.Header.Beacon.StateRoot == (common.Hash{}) { + // workaround for different event encoding format in Lodestar + if err := json.Unmarshal(enc, &data.Data); err != nil { + return types.SignedHeader{}, err + } + } + + if len(data.Data.Aggregate.Signers) != params.SyncCommitteeBitmaskSize { + return types.SignedHeader{}, errors.New("invalid sync_committee_bits length") + } + if len(data.Data.Aggregate.Signature) != params.BLSSignatureSize { + return types.SignedHeader{}, errors.New("invalid sync_committee_signature length") + } + return types.SignedHeader{ + Header: data.Data.Header.Beacon, + Signature: data.Data.Aggregate, + SignatureSlot: uint64(data.Data.SignatureSlot), + }, nil +} + +// GetHead fetches and validates the beacon header with the given blockRoot. +// If blockRoot is null hash then the latest head header is fetched. +func (api *BeaconLightApi) GetHeader(blockRoot common.Hash) (types.Header, error) { + var blockId string + if blockRoot == (common.Hash{}) { + blockId = "head" + } else { + blockId = blockRoot.Hex() + } + resp, err := api.httpGetf("/eth/v1/beacon/headers/%s", blockId) + if err != nil { + return types.Header{}, err + } + + var data struct { + Data struct { + Root common.Hash `json:"root"` + Canonical bool `json:"canonical"` + Header struct { + Message types.Header `json:"message"` + Signature hexutil.Bytes `json:"signature"` + } `json:"header"` + } `json:"data"` + } + if err := json.Unmarshal(resp, &data); err != nil { + return types.Header{}, err + } + header := data.Data.Header.Message + if blockRoot == (common.Hash{}) { + blockRoot = data.Data.Root + } + if header.Hash() != blockRoot { + return types.Header{}, errors.New("retrieved beacon header root does not match") + } + return header, nil +} + +// GetCheckpointData fetches and validates bootstrap data belonging to the given checkpoint. +func (api *BeaconLightApi) GetCheckpointData(checkpointHash common.Hash) (*types.BootstrapData, error) { + resp, err := api.httpGetf("/eth/v1/beacon/light_client/bootstrap/0x%x", checkpointHash[:]) + if err != nil { + return nil, err + } + + // See data structure definition here: + // https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/light-client/sync-protocol.md#lightclientbootstrap + type bootstrapData struct { + Data struct { + Header jsonBeaconHeader `json:"header"` + Committee *types.SerializedSyncCommittee `json:"current_sync_committee"` + CommitteeBranch merkle.Values `json:"current_sync_committee_branch"` + } `json:"data"` + } + + var data bootstrapData + if err := json.Unmarshal(resp, &data); err != nil { + return nil, err + } + if data.Data.Committee == nil { + return nil, errors.New("sync committee is missing") + } + header := data.Data.Header.Beacon + if header.Hash() != checkpointHash { + return nil, fmt.Errorf("invalid checkpoint block header, have %v want %v", header.Hash(), checkpointHash) + } + checkpoint := &types.BootstrapData{ + Header: header, + CommitteeBranch: data.Data.CommitteeBranch, + CommitteeRoot: data.Data.Committee.Root(), + Committee: data.Data.Committee, + } + if err := checkpoint.Validate(checkpointHash); err != nil { + return nil, fmt.Errorf("invalid sync committee Merkle proof: %w", err) + } + return checkpoint, nil +} + +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 + } + + var beaconBlockMessage struct { + Data struct { + Message capella.BeaconBlock `json:"message"` + } `json:"data"` + } + if err := json.Unmarshal(resp, &beaconBlockMessage); err != nil { + return nil, fmt.Errorf("invalid block json data: %v", err) + } + 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) + } + return beaconBlock, nil +} + +func decodeHeadEvent(enc []byte) (uint64, common.Hash, error) { + var data struct { + Slot common.Decimal `json:"slot"` + Block common.Hash `json:"block"` + } + if err := json.Unmarshal(enc, &data); err != nil { + return 0, common.Hash{}, err + } + return uint64(data.Slot), data.Block, nil +} + +// StartHeadListener creates an event subscription for heads and signed (optimistic) +// 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. +// They are never called concurrently. +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 + closedCh := make(chan struct{}) // stream closed (or failed to create) + stoppedCh := make(chan struct{}) // sync loop stopped + streamCh := make(chan *eventsource.Stream, 1) + go func() { + defer close(closedCh) + // when connected to a Lodestar node the subscription blocks until the + // first actual event arrives; therefore we create the subscription in + // a separate goroutine while letting the main goroutine sync up to the + // current head + req, err := http.NewRequest("GET", api.url+"/eth/v1/events?topics=head&topics=light_client_optimistic_update", nil) + if err != nil { + errFn(fmt.Errorf("Error creating event subscription request: %v", err)) + return + } + for k, v := range api.customHeaders { + req.Header.Set(k, v) + } + stream, err := eventsource.SubscribeWithRequest("", req) + if err != nil { + errFn(fmt.Errorf("Error creating event subscription: %v", err)) + close(streamCh) + return + } + streamCh <- stream + <-closeCh + stream.Close() + }() + go func() { + defer close(stoppedCh) + + if head, err := api.GetHeader(common.Hash{}); err == nil { + headFn(head.Slot, head.Hash()) + } + if signedHead, err := api.GetOptimisticHeadUpdate(); err == nil { + signedFn(signedHead) + } + stream := <-streamCh + if stream == nil { + return + } + for { + select { + case event, ok := <-stream.Events: + if !ok { + break + } + switch event.Event() { + case "head": + if slot, blockRoot, err := decodeHeadEvent([]byte(event.Data())); err == nil { + headFn(slot, blockRoot) + } else { + errFn(fmt.Errorf("Error decoding head event: %v", err)) + } + case "light_client_optimistic_update": + if signedHead, err := decodeOptimisticHeadUpdate([]byte(event.Data())); err == nil { + signedFn(signedHead) + } else { + errFn(fmt.Errorf("Error decoding optimistic update event: %v", err)) + } + default: + errFn(fmt.Errorf("Unexpected event: %s", event.Event())) + } + case err, ok := <-stream.Errors: + if !ok { + break + } + errFn(err) + } + } + }() + return func() { + close(closeCh) + <-closedCh + <-stoppedCh + } +} diff --git a/beacon/light/api/light_api_test.go b/beacon/light/api/light_api_test.go new file mode 100644 index 0000000000..2fc03e61ac --- /dev/null +++ b/beacon/light/api/light_api_test.go @@ -0,0 +1,43 @@ +package api + +import ( + "bytes" + "io" + "net/http" + "testing" + + "github.com/ethereum/go-ethereum/common" +) + +type testFetcher struct { + response string +} + +type nopCloser struct { + io.Reader +} + +func (nopCloser) Close() error { return nil } + +func makeApi(response string) *BeaconLightApi { + return &BeaconLightApi{client: &testFetcher{response}} +} + +func (f *testFetcher) Do(req *http.Request) (*http.Response, error) { + res := new(http.Response) + res.StatusCode = 200 + res.Body = nopCloser{bytes.NewBufferString(f.response)} + return res, nil +} + +func TestGetCheckpointData(t *testing.T) { + resp := + `{ "data": + { + "header": {} + } +}` + _, err := makeApi(resp).GetCheckpointData(common.HexToHash("0xc78009fdf07fc56a11f122370658a353aaa542ed63e44c4bc15ff4cd105ab33c")) + t.Logf("err: %v", err) + //TODO finish this test +} diff --git a/beacon/light/committee_chain.go b/beacon/light/committee_chain.go index d707f8cc34..93fb6b669b 100644 --- a/beacon/light/committee_chain.go +++ b/beacon/light/committee_chain.go @@ -86,6 +86,11 @@ func NewCommitteeChain(db ethdb.KeyValueStore, config *types.ChainConfig, signer return newCommitteeChain(db, config, signerThreshold, enforceTime, blsVerifier{}, &mclock.System{}, func() int64 { return time.Now().UnixNano() }) } +// NewTestCommitteeChain creates a new CommitteeChain for testing. +func NewTestCommitteeChain(db ethdb.KeyValueStore, config *types.ChainConfig, signerThreshold int, enforceTime bool, clock *mclock.Simulated) *CommitteeChain { + return newCommitteeChain(db, config, signerThreshold, enforceTime, dummyVerifier{}, clock, func() int64 { return int64(clock.Now()) }) +} + // newCommitteeChain creates a new CommitteeChain with the option of replacing the // clock source and signature verification for testing purposes. func newCommitteeChain(db ethdb.KeyValueStore, config *types.ChainConfig, signerThreshold int, enforceTime bool, sigVerifier committeeSigVerifier, clock mclock.Clock, unixNano func() int64) *CommitteeChain { @@ -183,18 +188,15 @@ func (s *CommitteeChain) Reset() { } } -// CheckpointInit initializes a CommitteeChain based on the checkpoint. +// CheckpointInit initializes a CommitteeChain based on a previously validated +// checkpoint. // Note: if the chain is already initialized and the committees proven by the // checkpoint do match the existing chain then the chain is retained and the // new checkpoint becomes fixed. -func (s *CommitteeChain) CheckpointInit(bootstrap *types.BootstrapData) error { +func (s *CommitteeChain) CheckpointInit(bootstrap types.BootstrapData) error { s.chainmu.Lock() defer s.chainmu.Unlock() - if err := bootstrap.Validate(); err != nil { - return err - } - period := bootstrap.Header.SyncPeriod() if err := s.deleteFixedCommitteeRootsFrom(period + 2); err != nil { s.Reset() diff --git a/beacon/light/committee_chain_test.go b/beacon/light/committee_chain_test.go index 60ea2a0efd..57b6d7175c 100644 --- a/beacon/light/committee_chain_test.go +++ b/beacon/light/committee_chain_test.go @@ -241,12 +241,12 @@ func newCommitteeChainTest(t *testing.T, config types.ChainConfig, signerThresho signerThreshold: signerThreshold, enforceTime: enforceTime, } - c.chain = newCommitteeChain(c.db, &config, signerThreshold, enforceTime, dummyVerifier{}, c.clock, func() int64 { return int64(c.clock.Now()) }) + c.chain = NewTestCommitteeChain(c.db, &config, signerThreshold, enforceTime, c.clock) return c } func (c *committeeChainTest) reloadChain() { - c.chain = newCommitteeChain(c.db, &c.config, c.signerThreshold, c.enforceTime, dummyVerifier{}, c.clock, func() int64 { return int64(c.clock.Now()) }) + c.chain = NewTestCommitteeChain(c.db, &c.config, c.signerThreshold, c.enforceTime, c.clock) } func (c *committeeChainTest) setClockPeriod(period float64) { diff --git a/beacon/light/head_tracker.go b/beacon/light/head_tracker.go new file mode 100644 index 0000000000..2b2afb1f60 --- /dev/null +++ b/beacon/light/head_tracker.go @@ -0,0 +1,91 @@ +// 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" + "time" + + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/log" +) + +type HeadTracker struct { + lock sync.RWMutex + committeeChain *CommitteeChain + minSignerCount int + signedHead types.SignedHeader + headSignerCount int + prefetchHead types.HeadInfo +} + +func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int) *HeadTracker { + return &HeadTracker{ + committeeChain: committeeChain, + minSignerCount: minSignerCount, + } +} + +func (h *HeadTracker) ValidatedHead() types.SignedHeader { + h.lock.RLock() + defer h.lock.RUnlock() + + return h.signedHead +} + +func (h *HeadTracker) Validate(head types.SignedHeader) (bool, error) { + h.lock.Lock() + defer h.lock.Unlock() + + signerCount := head.Signature.SignerCount() + if signerCount < h.minSignerCount { + return false, errors.New("low signer count") + } + if head.Header.Slot < h.signedHead.Header.Slot || (head.Header.Slot == h.signedHead.Header.Slot && signerCount <= h.headSignerCount) { + return false, nil + } + sigOk, age, err := h.committeeChain.VerifySignedHeader(head) + if err != nil { + return false, err + } + if age < 0 { + log.Warn("Future signed head received", "age", age) + } + if age > time.Minute*2 { + log.Warn("Old signed head received", "age", age) + } + if !sigOk { + return false, errors.New("invalid header signature") + } + h.signedHead, h.headSignerCount = head, signerCount + return true, nil +} + +func (h *HeadTracker) PrefetchHead() types.HeadInfo { + h.lock.RLock() + defer h.lock.RUnlock() + + return h.prefetchHead +} + +func (h *HeadTracker) SetPrefetchHead(head types.HeadInfo) { + h.lock.Lock() + defer h.lock.Unlock() + + h.prefetchHead = head +} diff --git a/beacon/light/request/request.go b/beacon/light/request/request.go new file mode 100644 index 0000000000..15038ff20f --- /dev/null +++ b/beacon/light/request/request.go @@ -0,0 +1,64 @@ +// 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 any + Response any + ID uint64 + ServerAndId struct { + Server Server + Id ID + } +) + +// one per sync process +type RequestTracker struct { + servers serverSet // one per trigger + scheduler *Scheduler + module Module + requestEvents []RequestEvent +} + +func (p *RequestTracker) TryRequest(requestFn func(server Server) (Request, float32)) (ServerAndId, Request) { + var ( + maxServerPriority, maxRequestPriority float32 + bestServer Server + bestRequest Request + ) + maxServerPriority, maxRequestPriority = -1000, -1000 + for server, _ := range p.servers { + canRequest, serverPriority := server.CanRequestNow() + if !canRequest { + delete(p.servers, server) + continue + } + request, requestPriority := requestFn(server) + if request == nil || requestPriority < maxRequestPriority || + (requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) { + continue + } + maxServerPriority, maxRequestPriority = serverPriority, requestPriority + bestServer, bestRequest = server, request + } + if bestServer == nil { + return ServerAndId{}, nil + } + id := ServerAndId{Server: bestServer, Id: bestServer.SendRequest(bestRequest)} + p.scheduler.pending[id] = pendingRequest{request: bestRequest, module: p.module} + return id, bestRequest +} diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go new file mode 100644 index 0000000000..00c5aabaa7 --- /dev/null +++ b/beacon/light/request/scheduler.go @@ -0,0 +1,261 @@ +// 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/common/mclock" + "github.com/ethereum/go-ethereum/log" +) + +// Module represents an update mechanism which is typically responsible for a +// passive data structure or a certain aspect of it. When registered to a Scheduler, +// it can be triggered either by server events, other modules or itself. +type Module interface { + // Process is a non-blocking function that is called whenever the module is + // triggered. It can start network requests through the received Environment + // and/or do other data processing tasks. If triggers are set up correctly, + // Process is eventually called whenever it might have something new to do + // either because the data structures have been changed or because new servers + // became available or new requests became available at existing ones. + // + // Note: Process functions of different modules are never called concurrently; + // they are called by Scheduler in the same order of priority as they were + // registered in. + Process(*RequestTracker, []RequestEvent, []ServerEvent) bool +} + +// Scheduler is a modular network data retrieval framework that coordinates multiple +// servers and retrieval mechanisms (modules). It implements a trigger mechanism +// that calls the Process function of registered modules whenever either the state +// of existing data structures or connected servers could allow new operations. +type Scheduler struct { + lock sync.Mutex + clock mclock.Clock + modules []Module // first has highest priority + trackers map[Module]*RequestTracker + servers map[Server]struct{} + pending map[ServerAndId]pendingRequest + serverEvents []ServerEvent + stopCh chan chan struct{} + + triggerCh chan struct{} // restarts waiting sync loop + // testWaitCh chan struct{} // accepts sends when sync loop is waiting + // testTimerResults []bool // true is appended when simulated timer is processed; false when stopped +} + +type ServerEvent struct { + Server Server + Type int + Data any +} + +type RequestEvent struct { + ServerAndId + Request Request + Response Response + Timeout, Finalized bool +} + +type pendingRequest struct { + request Request + module Module + timeout bool +} + +// NewScheduler creates a new Scheduler. +func NewScheduler(clock mclock.Clock) *Scheduler { + s := &Scheduler{ + clock: clock, + servers: make(map[Server]struct{}), + trackers: make(map[Module]*RequestTracker), + pending: make(map[ServerAndId]pendingRequest), + stopCh: make(chan chan struct{}), + // Note: testWaitCh should not have capacity in order to ensure + // that after a trigger happens testWaitCh will block until the resulting + // processing round has been finished + triggerCh: make(chan struct{}, 1), + //testWaitCh: make(chan struct{}), + } + return s +} + +// RegisterModule registers a module. Should be called before starting the scheduler. +// In each processing round the order of module processing depends on the order of +// registration. +func (s *Scheduler) RegisterModule(m Module) { + s.lock.Lock() + defer s.lock.Unlock() + + s.modules = append(s.modules, m) + s.trackers[m] = &RequestTracker{ + scheduler: s, + module: m, + } +} + +// RegisterServer registers a new server. +func (s *Scheduler) RegisterServer(server Server) { + s.lock.Lock() + defer s.lock.Unlock() + + s.handleEvent(server, Event{Type: EvRegistered}) + server.Subscribe(func(event Event) { + s.lock.Lock() + if _, ok := s.servers[server]; ok { + s.handleEvent(server, event) + } else { + log.Error("Event received from unsubscribed server") + } + s.lock.Unlock() + }) + s.servers[server] = struct{}{} +} + +// UnregisterServer removes a registered server. +func (s *Scheduler) UnregisterServer(server Server) { + s.lock.Lock() + defer s.lock.Unlock() + + server.Unsubscribe() + delete(s.servers, server) + s.handleEvent(server, Event{Type: EvUnregistered}) +} + +// 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 { + server.Unsubscribe() + } + s.servers = nil + s.lock.Unlock() + stop := make(chan struct{}) + s.stopCh <- 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() { + for { + s.processModules() + loop: + for { + select { + case stop := <-s.stopCh: + close(stop) + return + case <-s.triggerCh: + break loop + //case <-s.testWaitCh: + } + } + } +} + +// processModules runs an entire processing round, calling processable modules +// with the appropriate Environment. +func (s *Scheduler) processModules() { + s.lock.Lock() + servers := make(serverSet) + for server, _ := range s.servers { + if ok, _ := server.CanRequestNow(); ok { + servers[server] = struct{}{} + } + } + serverEvents := s.serverEvents + s.serverEvents = nil + s.lock.Unlock() + + for _, module := range s.modules { + s.lock.Lock() + tracker := s.trackers[module] + tracker.servers = servers + requestEvents := tracker.requestEvents + tracker.requestEvents = nil + s.lock.Unlock() + if module.Process(tracker, requestEvents, serverEvents) { + s.Trigger() + } + } +} + +func (s *Scheduler) Trigger() { + select { + case s.triggerCh <- struct{}{}: + default: + } +} + +func (s *Scheduler) addRequestEvent(server Server, id ID, response Response, timeout, finalized bool) { + sid := ServerAndId{Server: server, Id: id} + if pr, ok := s.pending[sid]; ok { + tracker := s.trackers[pr.module] + timeout = timeout || pr.timeout + tracker.requestEvents = append(tracker.requestEvents, RequestEvent{ + ServerAndId: sid, + Request: pr.request, + Response: response, + Timeout: timeout, + Finalized: finalized, + }) + if timeout && !finalized { + pr.timeout = true + s.pending[sid] = pr + } else { + delete(s.pending, sid) + } + } +} + +func (s *Scheduler) addServerEvent(server Server, event Event) { + s.serverEvents = append(s.serverEvents, ServerEvent{Server: server, Type: event.Type, Data: event.Data}) +} + +func (s *Scheduler) handleEvent(server Server, event Event) { + s.Trigger() + switch event.Type { + case EvResponse: + idr := event.Data.(IdAndResponse) + s.addRequestEvent(server, idr.ID, idr.Response, false, true) + case EvFail: + s.addRequestEvent(server, event.Data.(ID), nil, false, true) + server.Fail("failed request") + case EvTimeout: + s.addRequestEvent(server, event.Data.(ID), nil, true, false) + case EvUnregistered: + for id, _ := range s.pending { + if id.Server != server { + continue + } + s.addRequestEvent(server, id.Id, nil, false, true) + } + s.addServerEvent(server, event) + default: + s.addServerEvent(server, event) + } +} diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go new file mode 100644 index 0000000000..359a095554 --- /dev/null +++ b/beacon/light/request/server.go @@ -0,0 +1,347 @@ +// 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 ( + "math" + "math/rand" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/log" +) + +const ( + // request events + EvResponse = iota // data: IdAndResponse; sent by RequestServer + EvFail // data: ID; sent by RequestServer + EvTimeout // data: ID; sent by serverWithTimeout + // server events + EvRegistered // data: nil; sent by Scheduler + EvUnregistered // data: nil; sent by Scheduler + EvCanRequestAgain // data: nil; sent by serverWithLimits + EvAppSpecific // application specific events (sent by RequestServer) start at this index +) + +const ( + softRequestTimeout = time.Second + hardRequestTimeout = time.Second * 10 +) + +const ( + parallelAdjustUp = 0.1 + parallelAdjustDown = 1 + minParallelLimit = 1 + defaultParallelLimit = 3 + minFailureDelay = time.Millisecond * 100 + maxFailureDelay = time.Minute +) + +// RequestServer can send a set of requests pre-defined by the application and +// signal events through the event callback. After each request, it should send +// back either EvResponse or EvFail. Additionally, it may also send application- +// defined events that the Modules can interpret. +type RequestServer interface { + Subscribe(eventCallback func(event Event)) + SendRequest(request Request) ID + Unsubscribe() +} + +type Server interface { + RequestServer + CanRequestNow() (bool, float32) + Fail(desc string) +} + +func NewServer(rs RequestServer, clock mclock.Clock) Server { + s := &serverWithLimits{} + s.serverWithTimeout.RequestServer = rs + s.serverWithTimeout.init(clock) + s.init() + return s +} + +type serverSet map[Server]struct{} + +type Event struct { + Type int + Data any +} + +type IdAndResponse struct { + ID ID + Response Response +} + +// serverWithTimeout wraps a RequestServer and implements timeouts. After +// softRequestTimeout it sends an EvTimeout after which and EvResponse or an +// EvFail will still follow (EvTimeout cannot follow the latter two). +// After hardRequestTimeout it sends an EvFail and blocks any further events +// related to the given request coming from the parent RequestServer. +type serverWithTimeout struct { + RequestServer + lock sync.Mutex + clock mclock.Clock + childEventCb func(event Event) + timeouts map[ID]mclock.Timer +} + +func (s *serverWithTimeout) init(clock mclock.Clock) { + s.clock = clock + s.timeouts = make(map[ID]mclock.Timer) +} + +func (s *serverWithTimeout) Subscribe(eventCallback func(event Event)) { + s.lock.Lock() + defer s.lock.Unlock() + + s.childEventCb = eventCallback + s.RequestServer.Subscribe(s.eventCallback) +} + +func (s *serverWithTimeout) eventCallback(event Event) { + s.lock.Lock() + defer s.lock.Unlock() + + switch event.Type { + case EvResponse, EvFail: + var id ID + if event.Type == EvResponse { + id = event.Data.(IdAndResponse).ID + } else { + id = event.Data.(ID) + } + if timer, ok := s.timeouts[id]; ok { + // Note: if stopping the timer is unsuccessful then the resulting AfterFunc + // call will just do nothing + s.stopTimer(timer) + delete(s.timeouts, id) + s.childEventCb(event) + } + default: + s.childEventCb(event) + } +} + +func (s *serverWithTimeout) SendRequest(request Request) (reqId ID) { + s.lock.Lock() + defer s.lock.Unlock() + + reqId = s.RequestServer.SendRequest(request) + s.timeouts[reqId] = s.clock.AfterFunc(softRequestTimeout, func() { + /*if s.testTimerResults != nil { + s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + defer s.lock.Unlock() + + if _, ok := s.timeouts[reqId]; !ok { + return + } + s.timeouts[reqId] = s.clock.AfterFunc(hardRequestTimeout-softRequestTimeout, func() { + /*if s.testTimerResults != nil { + s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + defer s.lock.Unlock() + + if _, ok := s.timeouts[reqId]; !ok { + return + } + delete(s.timeouts, reqId) + s.childEventCb(Event{Type: EvFail, Data: reqId}) + }) + s.childEventCb(Event{Type: EvTimeout, Data: reqId}) + }) + return reqId +} + +// stop stops all goroutines associated with the server. +func (s *serverWithTimeout) Unsubscribe() { + s.lock.Lock() + defer s.lock.Unlock() + + for _, timer := range s.timeouts { + if timer != nil { + s.stopTimer(timer) + } + } + s.childEventCb = nil + s.RequestServer.Unsubscribe() +} + +func (s *serverWithTimeout) stopTimer(timer mclock.Timer) { + timer.Stop() + /*if timer.Stop() && s.scheduler.testTimerResults != nil { + s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, false) // simulated timer stopped + }*/ +} + +// serverWithLimits wraps serverWithTimeout and implements Server. It limits the +// number of parallel in-flight requests and prevents sending new requests when a +// pending one has already timed out. It also implements a failure delay mechanism +// that adds an exponentially growing delay each time a request fails (wrong answer +// or hard timeout). This makes the syncing mechanism less brittle as temporary +// failures of the server might happen sometimes, but still avoids hammering a +// non-functional server with requests. +type serverWithLimits struct { + serverWithTimeout + lock sync.Mutex + childEventCb func(event Event) + softTimeouts map[ID]struct{} + pendingCount, timeoutCount int + parallelLimit float32 + sendEvent bool + delayTimer mclock.Timer + delayCounter int + failureDelayEnd mclock.AbsTime + failureDelay float64 +} + +func (s *serverWithLimits) init() { + s.softTimeouts = make(map[ID]struct{}) + s.parallelLimit = defaultParallelLimit +} + +func (s *serverWithLimits) Subscribe(eventCallback func(event Event)) { + s.lock.Lock() + defer s.lock.Unlock() + + s.childEventCb = eventCallback + s.serverWithTimeout.Subscribe(s.eventCallback) +} + +func (s *serverWithLimits) eventCallback(event Event) { + s.lock.Lock() + defer s.lock.Unlock() + + switch event.Type { + case EvTimeout: + s.softTimeouts[event.Data.(ID)] = struct{}{} + s.timeoutCount++ + s.parallelLimit -= parallelAdjustDown + if s.parallelLimit < minParallelLimit { + s.parallelLimit = minParallelLimit + } + case EvResponse, EvFail: + var id ID + if event.Type == EvResponse { + id = event.Data.(IdAndResponse).ID + } else { + id = event.Data.(ID) + } + if _, ok := s.softTimeouts[id]; ok { + delete(s.softTimeouts, id) + s.timeoutCount-- + } + if event.Type == EvResponse && s.pendingCount >= int(s.parallelLimit) { + s.parallelLimit -= parallelAdjustUp + } + s.pendingCount-- + s.canRequestNow() // send event if needed + } + s.childEventCb(event) +} + +func (s *serverWithLimits) SendRequest(request Request) (reqId ID) { + s.lock.Lock() + defer s.lock.Unlock() + + s.pendingCount++ + id := s.serverWithTimeout.SendRequest(request) + return id +} + +// stop stops all goroutines associated with the server. +func (s *serverWithLimits) Unsubscribe() { + s.lock.Lock() + defer s.lock.Unlock() + + if s.delayTimer != nil { + s.stopTimer(s.delayTimer) + s.delayTimer = nil + } + s.childEventCb = nil + s.serverWithTimeout.Unsubscribe() +} + +func (s *serverWithLimits) canRequestNow() (bool, float32) { + if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) { + return false, 0 + } + if s.sendEvent { + s.childEventCb(Event{Type: EvCanRequestAgain}) + s.sendEvent = false + } + if s.parallelLimit < minParallelLimit { + s.parallelLimit = minParallelLimit + } + return true, -(float32(s.pendingCount) + rand.Float32()) / s.parallelLimit +} + +// EvCanRequestAgain guaranteed if it returns false +func (s *serverWithLimits) CanRequestNow() (bool, float32) { + s.lock.Lock() + defer s.lock.Unlock() + + canSend, priority := s.canRequestNow() + if !canSend { + s.sendEvent = true + } + return canSend, priority +} + +func (s *serverWithLimits) delay(delay time.Duration) { + if s.delayTimer != nil { + // Note: if stopping the timer is unsuccessful then the resulting AfterFunc + // call will just do nothing + s.stopTimer(s.delayTimer) + s.delayTimer = nil + } + + s.delayCounter++ + delayCounter := s.delayCounter + s.delayTimer = s.clock.AfterFunc(delay, func() { + /*if s.scheduler.testTimerResults != nil { + s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + if s.delayTimer != nil && s.delayCounter == delayCounter { // do nothing if there is a new timer now + s.delayTimer = nil + s.canRequestNow() // send event if necessary + } + s.lock.Unlock() + }) +} + +func (s *serverWithLimits) Fail(desc string) { + s.lock.Lock() + defer s.lock.Unlock() + + log.Debug("Server error", "description", desc) + s.failureDelay *= 2 + now := s.clock.Now() + if now > s.failureDelayEnd { + s.failureDelay *= math.Pow(2, -float64(now-s.failureDelayEnd)/float64(maxFailureDelay)) + } + if s.failureDelay < float64(minFailureDelay) { + s.failureDelay = float64(minFailureDelay) + } + s.failureDelayEnd = now + mclock.AbsTime(s.failureDelay) + s.delay(time.Duration(s.failureDelay)) +} diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go new file mode 100644 index 0000000000..94a5e4e0ff --- /dev/null +++ b/beacon/light/sync/head_sync.go @@ -0,0 +1,150 @@ +// 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" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/types" +) + +type HeadSync struct { + headTracker *light.HeadTracker + chain *light.CommitteeChain + nextSyncPeriod uint64 + chainInit bool + queuedHeads map[request.Server][]types.SignedHeader + serverHeads map[request.Server]types.HeadInfo + headServerCount map[types.HeadInfo]headServerCount + headCounter uint64 + prefetchHead types.HeadInfo +} + +type headServerCount struct { + serverCount int + headCounter uint64 +} + +func NewHeadSync(headTracker *light.HeadTracker, chain *light.CommitteeChain) *HeadSync { + s := &HeadSync{ + headTracker: headTracker, + chain: chain, + nextSyncPeriod: math.MaxUint64, + queuedHeads: make(map[request.Server][]types.SignedHeader), + serverHeads: make(map[request.Server]types.HeadInfo), + headServerCount: make(map[types.HeadInfo]headServerCount), + } + return s +} + +// Process implements request.Module +func (s *HeadSync) Process(tracker *request.RequestTracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) (trigger bool) { + nextPeriod, chainInit := s.chain.NextSyncPeriod() + if nextPeriod != s.nextSyncPeriod || chainInit != s.chainInit { + s.nextSyncPeriod, s.chainInit = nextPeriod, chainInit + s.processQueuedHeads() + } + for _, event := range serverEvents { + switch event.Type { + case EvNewHead: + trigger = trigger || s.setServerHead(event.Server, event.Data.(types.HeadInfo)) + case EvNewSignedHead: + s.newSignedHead(event.Server, event.Data.(types.SignedHeader)) + case request.EvUnregistered: + trigger = trigger || s.setServerHead(event.Server, types.HeadInfo{}) + delete(s.serverHeads, event.Server) + delete(s.queuedHeads, event.Server) + } + } + return +} + +func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedHeader) { + if signedHead.Header.SyncPeriod() > s.nextSyncPeriod { + s.queuedHeads[server] = append(s.queuedHeads[server], signedHead) //TODO protect against future period spam + return + } + if _, err := s.headTracker.Validate(signedHead); err != nil { + server.Fail(fmt.Sprintf("Invalid signed head: %v", err)) + } +} + +func (s *HeadSync) processQueuedHeads() { + for server, queued := range s.queuedHeads { + j := len(queued) + for i := len(queued) - 1; i >= 0; i-- { + if signedHead := queued[i]; signedHead.Header.SyncPeriod() <= s.nextSyncPeriod { + if _, err := s.headTracker.Validate(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:] + } + } +} + +// setServerHead processes non-validated server head announcements and updates +// the prefetch head if necessary. +//TODO report server failure if a server announces many heads that do not become validated soon. +func (s *HeadSync) setServerHead(server request.Server, head types.HeadInfo) bool { + if oldHead, ok := s.serverHeads[server]; ok { + if head == oldHead { + return false + } + h := s.headServerCount[oldHead] + if h.serverCount--; h.serverCount > 0 { + s.headServerCount[oldHead] = h + } else { + delete(s.headServerCount, oldHead) + } + } + if head != (types.HeadInfo{}) { + h, ok := s.headServerCount[head] + if !ok { + s.headCounter++ + h.headCounter = s.headCounter + } + h.serverCount++ + s.headServerCount[head] = h + } + var ( + bestHead types.HeadInfo + bestHeadInfo headServerCount + ) + for head, headServerCount := range s.headServerCount { + if headServerCount.serverCount > bestHeadInfo.serverCount || + (headServerCount.serverCount == bestHeadInfo.serverCount && headServerCount.headCounter > bestHeadInfo.headCounter) { + bestHead, bestHeadInfo = head, headServerCount + } + } + if bestHead == s.prefetchHead { + return false + } + s.prefetchHead = bestHead + s.headTracker.SetPrefetchHead(bestHead) + return true +} diff --git a/beacon/light/sync/types.go b/beacon/light/sync/types.go new file mode 100644 index 0000000000..5b2df1359a --- /dev/null +++ b/beacon/light/sync/types.go @@ -0,0 +1,41 @@ +// 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 ( + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/common" +) + +const ( + EvNewHead = iota + request.EvAppSpecific + EvNewSignedHead +) + +type ( + ReqUpdates struct { + FirstPeriod, Count uint64 + } + RespUpdates struct { + Updates []*types.LightClientUpdate + Committees []*types.SerializedSyncCommittee + } + ReqHeader common.Hash + ReqCheckpointData common.Hash + ReqBeaconBlock common.Hash +) diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go new file mode 100644 index 0000000000..e9a1ef84ee --- /dev/null +++ b/beacon/light/sync/update_sync.go @@ -0,0 +1,248 @@ +// 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 ( + "sort" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" +) + +const maxUpdateRequest = 8 + +type CheckpointInit struct { + chain *light.CommitteeChain + checkpointHash common.Hash + pending bool + initialized bool +} + +func NewCheckpointInit(chain *light.CommitteeChain, checkpointHash common.Hash) *CheckpointInit { + return &CheckpointInit{ + chain: chain, + checkpointHash: checkpointHash, + } +} + +// Process implements request.Module +func (s *CheckpointInit) Process(tracker *request.RequestTracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) bool { + if s.initialized { + return false + } + for _, event := range requestEvents { + if event.Timeout != event.Finalized { + s.pending = false + } + if event.Response != nil { + if checkpoint, ok := event.Response.(*types.BootstrapData); ok && checkpoint.Validate(common.Hash(event.Request.(ReqCheckpointData))) == nil { + s.chain.CheckpointInit(*checkpoint) //TODO + s.initialized = true + return true + } + event.Server.Fail("invalid checkpoint data") + } + } + if !s.pending { + if _, request := tracker.TryRequest(func(server request.Server) (request.Request, float32) { + return ReqCheckpointData(s.checkpointHash), 0 + }); request != nil { + s.pending = true + } + } + return false +} + +type ForwardUpdateSync struct { + chain *light.CommitteeChain + rangeLock rangeLock + processQueue []request.RequestEvent + nextSyncPeriod map[request.Server]uint64 +} + +func NewForwardUpdateSync(chain *light.CommitteeChain) *ForwardUpdateSync { + return &ForwardUpdateSync{ + chain: chain, + rangeLock: make(rangeLock), + nextSyncPeriod: make(map[request.Server]uint64), + } +} + +type rangeLock map[uint64]int + +func (r rangeLock) lock(first, count uint64, add int) { + for i := first; i < first+count; i++ { + if v := r[i] + add; v > 0 { + r[i] = v + } else { + delete(r, i) + } + } +} + +func (r rangeLock) firstUnlocked(start, maxCount uint64) (first, count uint64) { + first = start + for { + if _, ok := r[first]; !ok { + break + } + first++ + } + for { + count++ + if count == maxCount { + break + } + if _, ok := r[first+count]; ok { + break + } + } + return +} + +func (s *ForwardUpdateSync) verifyRange(event request.RequestEvent) bool { + request, ok := event.Request.(ReqUpdates) + if !ok { + return false + } + response, ok := event.Response.(RespUpdates) + if !ok { + return false + } + if uint64(len(response.Updates)) != request.Count || uint64(len(response.Committees)) != request.Count { + return false + } + for i, update := range response.Updates { + if update.AttestedHeader.Header.SyncPeriod() != request.FirstPeriod+uint64(i) { + return false + } + } + return true +} + +// returns true for partial success +func (s *ForwardUpdateSync) processResponse(event request.RequestEvent) (success bool) { + response, ok := event.Response.(RespUpdates) + if !ok { + return false + } + for i, update := range response.Updates { + if err := s.chain.InsertUpdate(update, response.Committees[i]); err != nil { + if err == light.ErrInvalidPeriod { + // there is a gap in the update periods; stop processing without + // failing and try again next time + return + } + if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg { + event.Server.Fail("invalid update received") + } else { + log.Error("Unexpected InsertUpdate error", "error", err) + } + return + } + success = true + } + return +} + +type updateResponseList []request.RequestEvent + +func (u updateResponseList) Len() int { return len(u) } +func (u updateResponseList) Swap(i, j int) { u[i], u[j] = u[j], u[i] } +func (u updateResponseList) Less(i, j int) bool { + return u[i].Request.(ReqUpdates).FirstPeriod < u[j].Request.(ReqUpdates).FirstPeriod +} + +// Process implements request.Module +func (s *ForwardUpdateSync) Process(tracker *request.RequestTracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) (trigger bool) { + // iterate events and add responses to process queue + for _, event := range requestEvents { + if event.Response != nil && !s.verifyRange(event) { + event.Server.Fail("invalid update range") + event.Response = nil + } + req := event.Request.(ReqUpdates) + if event.Response != nil { + // there is a response with a valid format; put it in the process queue + s.processQueue = append(s.processQueue, event) + if event.Timeout { + // it was already timed out and unlocked; lock again until processed + s.rangeLock.lock(req.FirstPeriod, req.Count, 1) + } + } else if event.Timeout != event.Finalized { + // unlock if timed out or returned with an invalid response without + // previously being unlocked by a timeout + s.rangeLock.lock(req.FirstPeriod, req.Count, -1) + } + } + + // try processing ordered list of available responses + sort.Sort(updateResponseList(s.processQueue)) //TODO + for s.processQueue != nil { + event := s.processQueue[0] + if !s.processResponse(event) { + break + } + trigger = true + req := event.Request.(ReqUpdates) + s.rangeLock.lock(req.FirstPeriod, req.Count, -1) + s.processQueue = s.processQueue[1:] + if len(s.processQueue) == 0 { + s.processQueue = nil + } + } + + // update nextSyncPeriod of servers based on server events + for _, event := range serverEvents { + switch event.Type { + case EvNewSignedHead: + signedHead := event.Data.(types.SignedHeader) + s.nextSyncPeriod[event.Server] = types.SyncPeriod(signedHead.Header.Slot + 256) + case request.EvUnregistered: + delete(s.nextSyncPeriod, event.Server) + } + } + + // start new requests if necessary + startPeriod, chainInit := s.chain.NextSyncPeriod() + if !chainInit { + return false + } + for { + firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest) + if _, request := tracker.TryRequest(func(server request.Server) (request.Request, float32) { + nextPeriod := s.nextSyncPeriod[server] + if nextPeriod <= firstPeriod { + return nil, 0 + } + count := maxCount + if nextPeriod < firstPeriod+maxCount { + count = nextPeriod - firstPeriod + } + return ReqUpdates{FirstPeriod: firstPeriod, Count: count}, float32(count) + }); request != nil { + req := request.(ReqUpdates) + s.rangeLock.lock(req.FirstPeriod, req.Count, 1) + } else { + break + } + } + return +} diff --git a/beacon/types/light_sync.go b/beacon/types/light_sync.go index 3284081e4d..1198754058 100644 --- a/beacon/types/light_sync.go +++ b/beacon/types/light_sync.go @@ -25,6 +25,12 @@ import ( "github.com/ethereum/go-ethereum/common" ) +// HeadInfo represents an unvalidated new head announcement. +type HeadInfo struct { + Slot uint64 + BlockRoot common.Hash +} + // BootstrapData contains a sync committee where light sync can be started, // together with a proof through a beacon header and corresponding state. // Note: BootstrapData is fetched from a server based on a known checkpoint hash. @@ -36,7 +42,10 @@ type BootstrapData struct { } // Validate verifies the proof included in BootstrapData. -func (c *BootstrapData) Validate() error { +func (c *BootstrapData) Validate(checkpointHash common.Hash) error { + if c.Header.Hash() != checkpointHash { + return errors.New("wrong checkpoint hash") + } if c.CommitteeRoot != c.Committee.Root() { return errors.New("wrong committee root") } diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go new file mode 100755 index 0000000000..5410de02db --- /dev/null +++ b/cmd/blsync/block_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 main + +import ( + "fmt" + "math/big" + "sync/atomic" + + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/light/sync" + "github.com/ethereum/go-ethereum/beacon/types" + "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 beaconBlockSync struct { + recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock] + validatedHead types.Header + pending map[common.Hash]struct{} + serverHeads map[request.Server]common.Hash + headTracker *light.HeadTracker +} + +func newBeaconBlockSyncer(headTracker *light.HeadTracker) *beaconBlockSync { + return &beaconBlockSync{ + headTracker: headTracker, + recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10), + pending: make(map[common.Hash]struct{}), + serverHeads: make(map[request.Server]common.Hash), + } +} + +// Process implements request.Module +func (s *beaconBlockSync) Process(tracker *request.RequestTracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) (trigger bool) { + s.validatedHead = s.headTracker.ValidatedHead().Header + if s.validatedHead == (types.Header{}) { + return false + } + + // iterate events and add valid responses to recentBlocks + for _, event := range requestEvents { + blockRoot := common.Hash(event.Request.(sync.ReqBeaconBlock)) + if event.Response != nil { + block := event.Response.(*capella.BeaconBlock) + s.recentBlocks.Add(blockRoot, block) + if blockRoot == s.validatedHead.Hash() { + trigger = true + } + } + if event.Timeout != event.Finalized { + // unlock if timed out or returned with an invalid response without + // previously being unlocked by a timeout + delete(s.pending, blockRoot) + } + } + + // update server heads + for _, event := range serverEvents { + switch event.Type { + case sync.EvNewHead: + s.serverHeads[event.Server] = event.Data.(types.HeadInfo).BlockRoot + case request.EvUnregistered: + delete(s.serverHeads, event.Server) + } + } + + // start new requests if necessary + s.tryRequestBlock(tracker, s.validatedHead.Hash(), false) + if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) { + s.tryRequestBlock(tracker, prefetchHead, true) + } + return +} + +// belongs to validatedHead (or nil) +func (s *beaconBlockSync) getHeadBlock() *capella.BeaconBlock { + block, _ := s.recentBlocks.Get(s.validatedHead.Hash()) + return block +} + +func (s *beaconBlockSync) tryRequestBlock(tracker *request.RequestTracker, blockRoot common.Hash, prefetch bool) { + if _, ok := s.recentBlocks.Get(blockRoot); ok { + return + } + if _, ok := s.pending[blockRoot]; ok { + return + } + if _, request := tracker.TryRequest(func(server request.Server) (request.Request, float32) { + if prefetch && s.serverHeads[server] != blockRoot { + // when requesting a not yet validated head, request it from someone + // who has announced it already + return nil, 0 + } + return sync.ReqBeaconBlock(blockRoot), 0 + }); request != nil { + s.pending[blockRoot] = struct{}{} + } +} + +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 + trigger func() + lastHead common.Hash + blockSync *beaconBlockSync + updating uint32 +} + +// Process implements request.Module +func (s *engineApiUpdater) Process(tracker *request.RequestTracker, requestEvents []request.RequestEvent, serverEvents []request.ServerEvent) bool { + if atomic.LoadUint32(&s.updating) == 1 { + return false + } + headBlock := s.blockSync.getHeadBlock() + if headBlock == nil { + return false + } + headRoot := common.Hash(headBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn())) + if headRoot == s.lastHead { + return false + } + + s.lastHead = headRoot + execBlock, err := getExecBlock(headBlock) + if err != nil { + log.Error("Error extracting execution block from validated beacon block", "error", err) + return false + } + 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) + } else { + atomic.StoreUint32(&s.updating, 1) + go func() { + if status, err := callNewPayloadV2(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, common.Hash{}); err == nil { + log.Info("Successful ForkchoiceUpdated", "head", execRoot, "status", status) + } else { + log.Error("Failed ForkchoiceUpdated", "head", execRoot, "error", err) + } + atomic.StoreUint32(&s.updating, 0) + s.trigger() + }() + } + return false +} diff --git a/cmd/blsync/config.go b/cmd/blsync/config.go new file mode 100644 index 0000000000..5bb80dc5aa --- /dev/null +++ b/cmd/blsync/config.go @@ -0,0 +1,142 @@ +// Copyright 2022 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 ( + "context" + + "github.com/ethereum/go-ethereum/beacon/types" + "github.com/ethereum/go-ethereum/cmd/utils" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/rpc" + "github.com/urfave/cli/v2" +) + +// lightClientConfig contains beacon light client configuration +type lightClientConfig struct { + *types.ChainConfig + Checkpoint common.Hash +} + +var ( + MainnetConfig = lightClientConfig{ + ChainConfig: (&types.ChainConfig{ + GenesisValidatorsRoot: common.HexToHash("0x4b363db94e286120d76eb905340fdd4e54bfe9f06bf33ff6cf5ad27f511bfe95"), + GenesisTime: 1606824023, + }). + AddFork("GENESIS", 0, []byte{0, 0, 0, 0}). + AddFork("ALTAIR", 74240, []byte{1, 0, 0, 0}). + AddFork("BELLATRIX", 144896, []byte{2, 0, 0, 0}). + AddFork("CAPELLA", 194048, []byte{3, 0, 0, 0}), + Checkpoint: common.HexToHash("0x388be41594ec7d6a6894f18c73f3469f07e2c19a803de4755d335817ed8e2e5a"), + } + + SepoliaConfig = lightClientConfig{ + ChainConfig: (&types.ChainConfig{ + GenesisValidatorsRoot: common.HexToHash("0xd8ea171f3c94aea21ebc42a1ed61052acf3f9209c00e4efbaaddac09ed9b8078"), + GenesisTime: 1655733600, + }). + AddFork("GENESIS", 0, []byte{144, 0, 0, 105}). + AddFork("ALTAIR", 50, []byte{144, 0, 0, 112}). + AddFork("BELLATRIX", 100, []byte{144, 0, 0, 113}). + AddFork("CAPELLA", 56832, []byte{144, 0, 0, 114}), + Checkpoint: common.HexToHash("0x1005a6d9175e96bfbce4d35b80f468e9bff0b674e1e861d16e09e10005a58e81"), + } + + GoerliConfig = lightClientConfig{ + ChainConfig: (&types.ChainConfig{ + GenesisValidatorsRoot: common.HexToHash("0x043db0d9a83813551ee2f33450d23797757d430911a9320530ad8a0eabc43efb"), + GenesisTime: 1614588812, + }). + AddFork("GENESIS", 0, []byte{0, 0, 16, 32}). + AddFork("ALTAIR", 36660, []byte{1, 0, 16, 32}). + AddFork("BELLATRIX", 112260, []byte{2, 0, 16, 32}). + AddFork("CAPELLA", 162304, []byte{3, 0, 16, 32}), + Checkpoint: common.HexToHash("0x53a0f4f0a378e2c4ae0a9ee97407eb69d0d737d8d8cd0a5fb1093f42f7b81c49"), + } +) + +func makeChainConfig(ctx *cli.Context) lightClientConfig { + utils.CheckExclusive(ctx, utils.MainnetFlag, utils.GoerliFlag, utils.SepoliaFlag) + customConfig := ctx.IsSet(utils.BeaconConfigFlag.Name) || ctx.IsSet(utils.BeaconGenesisRootFlag.Name) || ctx.IsSet(utils.BeaconGenesisTimeFlag.Name) + var config lightClientConfig + switch { + case ctx.Bool(utils.MainnetFlag.Name): + config = MainnetConfig + case ctx.Bool(utils.SepoliaFlag.Name): + config = SepoliaConfig + case ctx.Bool(utils.GoerliFlag.Name): + config = GoerliConfig + default: + if !customConfig { + config = MainnetConfig + } + } + if customConfig && config.Forks != nil { + utils.Fatalf("Cannot use custom beacon chain config flags in combination with pre-defined network config") + } + if ctx.IsSet(utils.BeaconGenesisRootFlag.Name) { + if c, err := hexutil.Decode(ctx.String(utils.BeaconGenesisRootFlag.Name)); err == nil && len(c) <= 32 { + copy(config.GenesisValidatorsRoot[:len(c)], c) + } else { + utils.Fatalf("Invalid hex string", "beacon.genesis.gvroot", ctx.String(utils.BeaconGenesisRootFlag.Name), "error", err) + } + } + if ctx.IsSet(utils.BeaconGenesisTimeFlag.Name) { + config.GenesisTime = ctx.Uint64(utils.BeaconGenesisTimeFlag.Name) + } + if ctx.IsSet(utils.BeaconConfigFlag.Name) { + if err := config.ChainConfig.LoadForks(ctx.String(utils.BeaconConfigFlag.Name)); err != nil { + utils.Fatalf("Could not load beacon chain config file", "file name", ctx.String(utils.BeaconConfigFlag.Name), "error", err) + } + } + if ctx.IsSet(utils.BeaconCheckpointFlag.Name) { + if c, err := hexutil.Decode(ctx.String(utils.BeaconCheckpointFlag.Name)); err == nil && len(c) <= 32 { + copy(config.Checkpoint[:len(c)], c) + } else { + utils.Fatalf("Invalid hex string", "beacon.checkpoint", ctx.String(utils.BeaconCheckpointFlag.Name), "error", err) + } + } + return config +} + +func makeRPCClient(ctx *cli.Context) *rpc.Client { + if !ctx.IsSet(utils.BlsyncApiFlag.Name) { + log.Warn("No engine API target specified, performing a dry run") + return nil + } + if !ctx.IsSet(utils.BlsyncJWTSecretFlag.Name) { + utils.Fatalf("JWT secret parameter missing") //TODO use default if datadir is specified + } + + engineApiUrl, jwtFileName := ctx.String(utils.BlsyncApiFlag.Name), ctx.String(utils.BlsyncJWTSecretFlag.Name) + var jwtSecret [32]byte + if jwt, err := node.ObtainJWTSecret(jwtFileName); err == nil { + copy(jwtSecret[:], jwt) + } else { + utils.Fatalf("Error loading or generating JWT secret: %v", err) + } + auth := node.NewJWTAuth(jwtSecret) + cl, err := rpc.DialOptions(context.Background(), engineApiUrl, rpc.WithHTTPAuth(auth)) + if err != nil { + utils.Fatalf("Could not create RPC client: %v", err) + } + return cl +} diff --git a/cmd/blsync/main.go b/cmd/blsync/main.go new file mode 100644 index 0000000000..555a4317b2 --- /dev/null +++ b/cmd/blsync/main.go @@ -0,0 +1,159 @@ +// Copyright 2022 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 ( + "context" + "fmt" + "os" + "strings" + "time" + + "github.com/ethereum/go-ethereum/beacon/engine" + "github.com/ethereum/go-ethereum/beacon/light" + "github.com/ethereum/go-ethereum/beacon/light/api" + "github.com/ethereum/go-ethereum/beacon/light/request" + "github.com/ethereum/go-ethereum/beacon/light/sync" + "github.com/ethereum/go-ethereum/cmd/utils" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" + ctypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb/memorydb" + "github.com/ethereum/go-ethereum/internal/flags" + "github.com/ethereum/go-ethereum/rpc" + "github.com/urfave/cli/v2" +) + +var ( + verbosityFlag = &cli.IntFlag{ + Name: "verbosity", + Usage: "Logging verbosity: 0=silent, 1=error, 2=warn, 3=info, 4=debug, 5=detail", + Value: 3, + Category: flags.LoggingCategory, + } + vmoduleFlag = &cli.StringFlag{ + Name: "vmodule", + Usage: "Per-module verbosity: comma-separated list of = (e.g. eth/*=5,p2p=4)", + Value: "", + Hidden: true, + Category: flags.LoggingCategory, + } +) + +func main() { + app := flags.NewApp("beacon light syncer tool") + app.Flags = []cli.Flag{ + utils.BeaconApiFlag, + utils.BeaconApiHeaderFlag, + utils.BeaconThresholdFlag, + utils.BeaconNoFilterFlag, + utils.BeaconConfigFlag, + utils.BeaconGenesisRootFlag, + utils.BeaconGenesisTimeFlag, + utils.BeaconCheckpointFlag, + //TODO datadir for optional permanent database + utils.MainnetFlag, + utils.SepoliaFlag, + utils.GoerliFlag, + utils.BlsyncApiFlag, + utils.BlsyncJWTSecretFlag, + verbosityFlag, + vmoduleFlag, + } + app.Action = blsync + + if err := app.Run(os.Args); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func blsync(ctx *cli.Context) error { + if !ctx.IsSet(utils.BeaconApiFlag.Name) { + utils.Fatalf("Beacon node light client API URL not specified") + } + var ( + chainConfig = makeChainConfig(ctx) + customHeader = make(map[string]string) + ) + + for _, s := range ctx.StringSlice(utils.BeaconApiHeaderFlag.Name) { + kv := strings.Split(s, ":") + if len(kv) != 2 { + utils.Fatalf("Invalid custom API header entry: %s", s) + } + customHeader[strings.TrimSpace(kv[0])] = strings.TrimSpace(kv[1]) + } + + // create data structures + var ( + db = memorydb.New() + threshold = ctx.Int(utils.BeaconThresholdFlag.Name) + committeeChain = light.NewCommitteeChain(db, chainConfig.ChainConfig, threshold, !ctx.Bool(utils.BeaconNoFilterFlag.Name)) + headTracker = light.NewHeadTracker(committeeChain, threshold) + ) + headSync := sync.NewHeadSync(headTracker, committeeChain) + + // set up scheduler and sync modules + scheduler := request.NewScheduler(&mclock.System{}) + + checkpointInit := sync.NewCheckpointInit(committeeChain, chainConfig.Checkpoint) + forwardSync := sync.NewForwardUpdateSync(committeeChain) + beaconBlockSync := newBeaconBlockSyncer(headTracker) + engineApiUpdater := &engineApiUpdater{ //TODO constructor + client: makeRPCClient(ctx), + blockSync: beaconBlockSync, + } + + scheduler.RegisterModule(checkpointInit) + scheduler.RegisterModule(forwardSync) + scheduler.RegisterModule(headSync) + scheduler.RegisterModule(beaconBlockSync) + scheduler.RegisterModule(engineApiUpdater) + // start + scheduler.Start() + // register server(s) + for _, url := range ctx.StringSlice(utils.BeaconApiFlag.Name) { + beaconApi := api.NewBeaconLightApi(url, customHeader) + scheduler.RegisterServer(request.NewServer(api.NewApiServer(beaconApi), &mclock.System{})) + } + // run until stopped + <-ctx.Done() + scheduler.Stop() + return nil +} + +func callNewPayloadV2(client *rpc.Client, block *ctypes.Block) (string, error) { + var resp engine.PayloadStatusV1 + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + err := client.CallContext(ctx, &resp, "engine_newPayloadV2", *engine.BlockToExecutableData(block, nil, nil).ExecutionPayload) + cancel() + return resp.Status, err +} + +func callForkchoiceUpdatedV1(client *rpc.Client, headHash, finalizedHash common.Hash) (string, error) { + var resp engine.ForkChoiceResponse + update := engine.ForkchoiceStateV1{ + HeadBlockHash: headHash, + SafeBlockHash: finalizedHash, + FinalizedBlockHash: finalizedHash, + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + err := client.CallContext(ctx, &resp, "engine_forkchoiceUpdatedV1", update, nil) + cancel() + return resp.PayloadStatus.Status, err +} diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index fad567cd55..e002975d53 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -36,6 +36,7 @@ import ( "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/accounts/keystore" + bparams "github.com/ethereum/go-ethereum/beacon/params" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/fdlimit" "github.com/ethereum/go-ethereum/core" @@ -281,6 +282,58 @@ var ( Value: ethconfig.Defaults.TransactionHistory, Category: flags.StateCategory, } + // Beacon client light sync settings + BeaconApiFlag = &cli.StringSliceFlag{ + Name: "beacon.api", + Usage: "Beacon node (CL) light client API URL. This flag can be given multiple times.", + Category: flags.BeaconCategory, + } + BeaconApiHeaderFlag = &cli.StringSliceFlag{ + Name: "beacon.api.header", + Usage: "Pass custom HTTP header fields to the emote beacon node API in \"key:value\" format. This flag can be given multiple times.", + Category: flags.BeaconCategory, + } + BeaconThresholdFlag = &cli.IntFlag{ + Name: "beacon.threshold", + Usage: "Beacon sync committee participation threshold", + Value: bparams.SyncCommitteeSupermajority, + Category: flags.BeaconCategory, + } + BeaconNoFilterFlag = &cli.BoolFlag{ + Name: "beacon.nofilter", + Usage: "Disable future slot signature filter", + Category: flags.BeaconCategory, + } + BeaconConfigFlag = &cli.StringFlag{ + Name: "beacon.config", + Usage: "Beacon chain config YAML file", + Category: flags.BeaconCategory, + } + BeaconGenesisRootFlag = &cli.StringFlag{ + Name: "beacon.genesis.gvroot", + Usage: "Beacon chain genesis validators root", + Category: flags.BeaconCategory, + } + BeaconGenesisTimeFlag = &cli.Uint64Flag{ + Name: "beacon.genesis.time", + Usage: "Beacon chain genesis time", + Category: flags.BeaconCategory, + } + BeaconCheckpointFlag = &cli.StringFlag{ + Name: "beacon.checkpoint", + Usage: "Beacon chain weak subjectivity checkpoint block hash", + Category: flags.BeaconCategory, + } + BlsyncApiFlag = &cli.StringFlag{ + Name: "blsync.engine.api", + Usage: "Target EL engine API URL", + Category: flags.BeaconCategory, + } + BlsyncJWTSecretFlag = &cli.StringFlag{ + Name: "blsync.jwtsecret", + Usage: "Path to a JWT secret to use for target engine API endpoint", + Category: flags.BeaconCategory, + } // Transaction pool settings TxPoolLocalsFlag = &cli.StringFlag{ Name: "txpool.locals", diff --git a/go.mod b/go.mod index 6591bee62f..ca45364b8b 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,8 @@ require ( github.com/crate-crypto/go-kzg-4844 v0.7.0 github.com/davecgh/go-spew v1.1.1 github.com/deckarep/golang-set/v2 v2.1.0 - github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127 + github.com/donovanhide/eventsource v0.0.0-20210830082556-c59027999da0 + github.com/dop251/goja v0.0.0-20230605162241-28ee0ee714f3 github.com/ethereum/c-kzg-4844 v0.4.0 github.com/fatih/color v1.13.0 github.com/ferranbt/fastssz v0.1.2 @@ -54,6 +55,8 @@ require ( github.com/olekukonko/tablewriter v0.0.5 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7 + github.com/protolambda/zrnt v0.30.0 + github.com/protolambda/ztyp v0.2.2 github.com/rs/cors v1.7.0 github.com/shirou/gopsutil v3.21.4-0.20210419000835-c7a38de76ee5+incompatible github.com/status-im/keycard-go v0.2.0 diff --git a/go.sum b/go.sum index cc74e15cb4..640be56289 100644 --- a/go.sum +++ b/go.sum @@ -149,9 +149,11 @@ github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwu github.com/dlclark/regexp2 v1.7.0 h1:7lJfhqlPssTb1WQx4yvTHN0uElPEv52sbaECrAQxjAo= github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI= +github.com/donovanhide/eventsource v0.0.0-20210830082556-c59027999da0 h1:C7t6eeMaEQVy6e8CarIhscYQlNmw5e3G36y7l7Y21Ao= +github.com/donovanhide/eventsource v0.0.0-20210830082556-c59027999da0/go.mod h1:56wL82FO0bfMU5RvfXoIwSOP2ggqqxT+tAfNEIyxuHw= github.com/dop251/goja v0.0.0-20211022113120-dc8c55024d06/go.mod h1:R9ET47fwRVRPZnOGvHxxhuZcbrMCuiqOz3Rlrh4KSnk= -github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127 h1:qwcF+vdFrvPSEUDSX5RVoRccG8a5DhOdWdQ4zN62zzo= -github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127/go.mod h1:QMWlm50DNe14hD7t24KEqZuUdC9sOTy8W6XbCU1mlw4= +github.com/dop251/goja v0.0.0-20230605162241-28ee0ee714f3 h1:+3HCtB74++ClLy8GgjUQYeC8R4ILzVcIe8+5edAJJnE= +github.com/dop251/goja v0.0.0-20230605162241-28ee0ee714f3/go.mod h1:QMWlm50DNe14hD7t24KEqZuUdC9sOTy8W6XbCU1mlw4= github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA7c8pLvoGndExHudxTDKZ84Pyvv+90pbBjbTz0Y= github.com/dop251/goja_nodejs v0.0.0-20211022123610-8dd9abb0616d/go.mod h1:DngW8aVqWbuLRMHItjPUyqdj+HWPvnQe8V8y1nDpIbM= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= @@ -239,6 +241,7 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb h1:PBC98N2aIaM3XXiurYmW7fx4GZkL8feAMVq7nEjURHk= github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= @@ -298,6 +301,7 @@ github.com/holiman/billy v0.0.0-20240216141850-2abb0c79d3c4 h1:X4egAf/gcS1zATw6w github.com/holiman/billy v0.0.0-20240216141850-2abb0c79d3c4/go.mod h1:5GuXa7vkL8u9FkFuWdVvfR5ix8hRB7DbOAaYULamFpc= github.com/holiman/bloomfilter/v2 v2.0.3 h1:73e0e/V0tCydx14a0SCYS/EWCxgwLZ18CZcZKVu0fao= github.com/holiman/bloomfilter/v2 v2.0.3/go.mod h1:zpoh+gs7qcpqrHr3dB55AMiJwo0iURXE7ZOP9L9hSkA= +github.com/holiman/uint256 v1.2.0/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25ApIH5Jw= github.com/holiman/uint256 v1.2.4 h1:jUc4Nk8fm9jZabQuqr2JzednajVmBpC+oiTiXZJEApU= github.com/holiman/uint256 v1.2.4/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= @@ -380,6 +384,10 @@ github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI= github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4= +github.com/mediocregopher/mediocre-go-lib v0.0.0-20181029021733-cb65787f37ed/go.mod h1:dSsfyI2zABAdhcbvkXqgxOxrCsbYeHCPgrZkku60dSg= +github.com/mediocregopher/radix/v3 v3.3.0/go.mod h1:EmfVyvspXz1uZEyPBMyGK+kjWiKQGvsUt6O3Pj+LDCQ= +github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc= +github.com/minio/sha256-simd v0.1.0/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U= github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g= github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM= github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag= @@ -448,8 +456,14 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA= github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0VU= github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA= +github.com/protolambda/bls12-381-util v0.0.0-20210720105258-a772f2aac13e/go.mod h1:MPZvj2Pr0N8/dXyTPS5REeg2sdLG7t8DRzC1rLv925w= github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7 h1:cZC+usqsYgHtlBaGulVnZ1hfKAi8iWtujBnRLQE698c= github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7/go.mod h1:IToEjHuttnUzwZI5KBSM/LOOW3qLbbrHOEfp3SbECGY= +github.com/protolambda/messagediff v1.4.0/go.mod h1:LboJp0EwIbJsePYpzh5Op/9G1/4mIztMRYzzwR0dR2M= +github.com/protolambda/zrnt v0.30.0 h1:pHEn69ZgaDFGpLGGYG1oD7DvYI7RDirbMBPfbC+8p4g= +github.com/protolambda/zrnt v0.30.0/go.mod h1:qcdX9CXFeVNCQK/q0nswpzhd+31RHMk2Ax/2lMsJ4Jw= +github.com/protolambda/ztyp v0.2.2 h1:rVcL3vBu9W/aV646zF6caLS/dyn9BN8NYiuJzicLNyY= +github.com/protolambda/ztyp v0.2.2/go.mod h1:9bYgKGqg3wJqT9ac1gI2hnVb0STQq7p/1lapqrqY1dU= github.com/prysmaticlabs/gohashtree v0.0.1-alpha.0.20220714111606-acbb2962fb48 h1:cSo6/vk8YpvkLbk9v3FO97cakNmUoxwi2KMP8hd5WIw= github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= @@ -842,6 +856,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/internal/flags/categories.go b/internal/flags/categories.go index 3ff0767921..c044e28f38 100644 --- a/internal/flags/categories.go +++ b/internal/flags/categories.go @@ -20,6 +20,7 @@ import "github.com/urfave/cli/v2" const ( EthCategory = "ETHEREUM" + BeaconCategory = "BEACON CHAIN" LightCategory = "LIGHT CLIENT" DevCategory = "DEVELOPER CHAIN" StateCategory = "STATE HISTORY MANAGEMENT" diff --git a/node/node.go b/node/node.go index dfa83d58c7..c5cb552d27 100644 --- a/node/node.go +++ b/node/node.go @@ -339,15 +339,9 @@ func (n *Node) closeDataDir() { } } -// obtainJWTSecret loads the jwt-secret, either from the provided config, -// or from the default location. If neither of those are present, it generates -// a new secret and stores to the default location. -func (n *Node) obtainJWTSecret(cliParam string) ([]byte, error) { - fileName := cliParam - if len(fileName) == 0 { - // no path provided, use default - fileName = n.ResolvePath(datadirJWTKey) - } +// ObtainJWTSecret loads the jwt-secret from the provided config. If the file is not +// present, it generates a new secret and stores to the given location. +func ObtainJWTSecret(fileName string) ([]byte, error) { // try reading from file if data, err := os.ReadFile(fileName); err == nil { jwtSecret := common.FromHex(strings.TrimSpace(string(data))) @@ -373,6 +367,18 @@ func (n *Node) obtainJWTSecret(cliParam string) ([]byte, error) { return jwtSecret, nil } +// obtainJWTSecret loads the jwt-secret, either from the provided config, +// or from the default location. If neither of those are present, it generates +// a new secret and stores to the default location. +func (n *Node) obtainJWTSecret(cliParam string) ([]byte, error) { + fileName := cliParam + if len(fileName) == 0 { + // no path provided, use default + fileName = n.ResolvePath(datadirJWTKey) + } + return ObtainJWTSecret(fileName) +} + // startRPC is a helper method to configure all the various RPC endpoints during node // startup. It's not meant to be called at any time afterwards as it makes certain // assumptions about the state of the node.