Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

support --sync.loop.block.limit in newHeads notification #9684

Merged
merged 4 commits into from
Mar 13, 2024
Merged

support --sync.loop.block.limit in newHeads notification #9684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3e7e8c8
save
AskAlexSharov Mar 12, 2024
007cb47
save
AskAlexSharov Mar 12, 2024
c4d2b58
save
AskAlexSharov Mar 12, 2024
e414179
save
AskAlexSharov Mar 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 12 additions & 11 deletions eth/stagedsync/stage_finish.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@ import (
"context"
"encoding/binary"
"fmt"
"github.com/ledgerwatch/erigon-lib/kv/dbutils"
"time"

"github.com/ledgerwatch/erigon-lib/kv/dbutils"

libcommon "github.com/ledgerwatch/erigon-lib/common"
"github.com/ledgerwatch/erigon-lib/common/hexutility"
"github.com/ledgerwatch/erigon-lib/gointerfaces"
Expand Down Expand Up @@ -148,18 +149,18 @@ func NotifyNewHeaders(ctx context.Context, finishStageBeforeSync uint64, finishS
var notifyTo = notifyFrom
var notifyToHash libcommon.Hash
var headersRlp [][]byte
if err := tx.ForEach(kv.Headers, hexutility.EncodeTs(notifyFrom), func(k, headerRLP []byte) error {
if len(headerRLP) == 0 {
if err := tx.ForEach(kv.HeaderCanonical, hexutility.EncodeTs(notifyFrom), func(k, hash []byte) (err error) {
if len(hash) == 0 {
return nil
}
notifyTo = binary.BigEndian.Uint64(k)
var err error
if notifyToHash, err = blockReader.CanonicalHash(ctx, tx, notifyTo); err != nil {
logger.Warn("[Finish] failed checking if header is cannonical")
blockNum := binary.BigEndian.Uint64(k)
if blockNum > finishStageAfterSync { //[from,to)
return nil
}

headerHash := libcommon.BytesToHash(k[8:])
if notifyToHash == headerHash {
notifyTo = blockNum
notifyToHash = libcommon.BytesToHash(hash)
headerRLP := rawdb.ReadHeaderRLP(tx, notifyToHash, notifyTo)
if headerRLP != nil {
headersRlp = append(headersRlp, libcommon.CopyBytes(headerRLP))
}

Expand All @@ -182,7 +183,7 @@ func NotifyNewHeaders(ctx context.Context, finishStageBeforeSync uint64, finishS
notifier.OnLogs(logs)
}
logTiming := time.Since(t)
logger.Info("RPC Daemon notified of new headers", "from", notifyFrom-1, "to", notifyTo, "hash", notifyToHash, "header sending", headerTiming, "log sending", logTiming)
logger.Info("RPC Daemon notified of new headers", "from", notifyFrom-1, "to", notifyTo, "amount", len(headersRlp), "hash", notifyToHash, "header sending", headerTiming, "log sending", logTiming)
}
return nil
}
Expand Down
6 changes: 5 additions & 1 deletion turbo/stages/stageloop.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,11 @@ func (h *Hook) afterRun(tx kv.Tx, finishProgressBefore uint64) error {
}

if notifications != nil && notifications.Events != nil {
if err = stagedsync.NotifyNewHeaders(h.ctx, finishProgressBefore, head, h.sync.PrevUnwindPoint(), notifications.Events, tx, h.logger, blockReader); err != nil {
finishStageAfterSync, err := stages.GetStageProgress(tx, stages.Finish)
if err != nil {
return err
}
if err = stagedsync.NotifyNewHeaders(h.ctx, finishProgressBefore, finishStageAfterSync, h.sync.PrevUnwindPoint(), notifications.Events, tx, h.logger, blockReader); err != nil {
return nil
}
}
Expand Down
Loading