diff --git a/.golangci.yml b/.golangci.yml index a702da524b..5ed06537bd 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -1,31 +1,25 @@ # This file configures github.com/golangci/golangci-lint. - +version: '2' run: - timeout: 20m tests: true - linters: - disable-all: true + default: none enable: - - goimports - - gosimple - - govet - - ineffassign - - misspell - - unconvert - - typecheck - - unused - - staticcheck - bidichk - - durationcheck - copyloopvar - - whitespace - - revive # only certain checks enabled - durationcheck - gocheckcompilerdirectives - - reassign + - govet + - ineffassign - mirror + - misspell + - reassign + - revive # only certain checks enabled + - staticcheck + - unconvert + - unused - usetesting + - whitespace ### linters we tried and will not be using: ### # - structcheck # lots of false positives @@ -36,44 +30,67 @@ linters: # - exhaustive # silly check # - makezero # false positives # - nilerr # several intentional - -linters-settings: - gofmt: - simplify: true - revive: - enable-all-rules: false - # here we enable specific useful rules - # see https://golangci-lint.run/usage/linters/#revive for supported rules + settings: + staticcheck: + checks: + # disable Quickfixes + - -QF1* + revive: + enable-all-rules: false + # here we enable specific useful rules + # see https://golangci-lint.run/usage/linters/#revive for supported rules + rules: + - name: receiver-naming + severity: warning + disabled: false + exclude: + - '' + exclusions: + generated: lax + presets: + - comments + - common-false-positives + - legacy + - std-error-handling rules: - - name: receiver-naming - severity: warning - disabled: false - exclude: [""] - -issues: - # default is true. Enables skipping of directories: - # vendor$, third_party$, testdata$, examples$, Godeps$, builtin$ - exclude-dirs-use-default: true - exclude-files: - - core/genesis_alloc.go - exclude-rules: - - path: crypto/bn256/cloudflare/optate.go - linters: - - deadcode - - staticcheck - - path: crypto/bn256/ - linters: - - revive - - path: cmd/utils/flags.go - text: "SA1019: cfg.TxLookupLimit is deprecated: use 'TransactionHistory' instead." - - path: cmd/utils/flags.go - text: "SA1019: ethconfig.Defaults.TxLookupLimit is deprecated: use 'TransactionHistory' instead." - - path: internal/build/pgp.go - text: 'SA1019: "golang.org/x/crypto/openpgp" is deprecated: this package is unmaintained except for security fixes.' - - path: core/vm/contracts.go - text: 'SA1019: "golang.org/x/crypto/ripemd160" is deprecated: RIPEMD-160 is a legacy hash and should not be used for new applications.' - exclude: - - 'SA1019: event.TypeMux is deprecated: use Feed' - - 'SA1019: strings.Title is deprecated' - - 'SA1019: strings.Title has been deprecated since Go 1.18 and an alternative has been available since Go 1.0: The rule Title uses for word boundaries does not handle Unicode punctuation properly. Use golang.org/x/text/cases instead.' - - 'SA1029: should not use built-in type string as key for value' + - linters: + - deadcode + - staticcheck + path: crypto/bn256/cloudflare/optate.go + - linters: + - revive + path: crypto/bn256/ + - path: cmd/utils/flags.go + text: "SA1019: cfg.TxLookupLimit is deprecated: use 'TransactionHistory' instead." + - path: cmd/utils/flags.go + text: "SA1019: ethconfig.Defaults.TxLookupLimit is deprecated: use 'TransactionHistory' instead." + - path: internal/build/pgp.go + text: 'SA1019: "golang.org/x/crypto/openpgp" is deprecated: this package is unmaintained except for security fixes.' + - path: core/vm/contracts.go + text: 'SA1019: "golang.org/x/crypto/ripemd160" is deprecated: RIPEMD-160 is a legacy hash and should not be used for new applications.' + - path: (.+)\.go$ + text: 'SA1019: event.TypeMux is deprecated: use Feed' + - path: (.+)\.go$ + text: 'SA1019: strings.Title is deprecated' + - path: (.+)\.go$ + text: 'SA1019: strings.Title has been deprecated since Go 1.18 and an alternative has been available since Go 1.0: The rule Title uses for word boundaries does not handle Unicode punctuation properly. Use golang.org/x/text/cases instead.' + - path: (.+)\.go$ + text: 'SA1029: should not use built-in type string as key for value' + paths: + - core/genesis_alloc.go + - third_party$ + - builtin$ + - examples$ +formatters: + enable: + - goimports + settings: + gofmt: + simplify: true + exclusions: + generated: lax + paths: + - core/genesis_alloc.go + - third_party$ + - builtin$ + - examples$ diff --git a/beacon/blsync/client.go b/beacon/blsync/client.go index 3c93754d3d..744f469124 100644 --- a/beacon/blsync/client.go +++ b/beacon/blsync/client.go @@ -23,9 +23,11 @@ import ( "github.com/ethereum/go-ethereum/beacon/light/sync" "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/mclock" "github.com/ethereum/go-ethereum/ethdb/memorydb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rpc" ) @@ -46,7 +48,13 @@ func NewClient(config params.ClientConfig) *Client { var ( db = memorydb.New() committeeChain = light.NewCommitteeChain(db, &config.ChainConfig, config.Threshold, !config.NoFilter) - headTracker = light.NewHeadTracker(committeeChain, config.Threshold) + headTracker = light.NewHeadTracker(committeeChain, config.Threshold, func(checkpoint common.Hash) { + if saved, err := config.SaveCheckpointToFile(checkpoint); saved { + log.Debug("Saved beacon checkpoint", "file", config.CheckpointFile, "checkpoint", checkpoint) + } else if err != nil { + log.Error("Failed to save beacon checkpoint", "file", config.CheckpointFile, "checkpoint", checkpoint, "error", err) + } + }) ) headSync := sync.NewHeadSync(headTracker, committeeChain) diff --git a/beacon/light/head_tracker.go b/beacon/light/head_tracker.go index 010e548ddb..62faf1dbc1 100644 --- a/beacon/light/head_tracker.go +++ b/beacon/light/head_tracker.go @@ -21,7 +21,9 @@ import ( "sync" "time" + "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/log" ) @@ -38,13 +40,15 @@ type HeadTracker struct { hasFinalityUpdate bool prefetchHead types.HeadInfo changeCounter uint64 + saveCheckpoint func(common.Hash) } // NewHeadTracker creates a new HeadTracker. -func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int) *HeadTracker { +func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int, saveCheckpoint func(common.Hash)) *HeadTracker { return &HeadTracker{ committeeChain: committeeChain, minSignerCount: minSignerCount, + saveCheckpoint: saveCheckpoint, } } @@ -100,6 +104,9 @@ func (h *HeadTracker) ValidateFinality(update types.FinalityUpdate) (bool, error if replace { h.finalityUpdate, h.hasFinalityUpdate = update, true h.changeCounter++ + if h.saveCheckpoint != nil && update.Finalized.Slot%params.EpochLength == 0 { + h.saveCheckpoint(update.Finalized.Hash()) + } } return replace, err } diff --git a/beacon/params/config.go b/beacon/params/config.go index be2a40f171..2f6ba082c5 100644 --- a/beacon/params/config.go +++ b/beacon/params/config.go @@ -54,6 +54,7 @@ type ChainConfig struct { GenesisValidatorsRoot common.Hash // Root hash of the genesis validator set, used for signature domain calculation Forks Forks Checkpoint common.Hash + CheckpointFile string } // ForkAtEpoch returns the latest active fork at the given epoch. @@ -211,3 +212,36 @@ func (f Forks) Less(i, j int) bool { } return f[i].knownIndex < f[j].knownIndex } + +// SetCheckpointFile sets the checkpoint import/export file name and attempts to +// read the checkpoint from the file if it already exists. It returns true if +// a checkpoint has been loaded. +func (c *ChainConfig) SetCheckpointFile(checkpointFile string) (bool, error) { + c.CheckpointFile = checkpointFile + file, err := os.ReadFile(checkpointFile) + if os.IsNotExist(err) { + return false, nil // did not load checkpoint + } + if err != nil { + return false, fmt.Errorf("failed to read beacon checkpoint file: %v", err) + } + cp, err := hexutil.Decode(string(file)) + if err != nil { + return false, fmt.Errorf("failed to decode hex string in beacon checkpoint file: %v", err) + } + if len(cp) != 32 { + return false, fmt.Errorf("invalid hex string length in beacon checkpoint file: %d", len(cp)) + } + copy(c.Checkpoint[:len(cp)], cp) + return true, nil +} + +// SaveCheckpointToFile saves the given checkpoint to file if a checkpoint +// import/export file has been specified. +func (c *ChainConfig) SaveCheckpointToFile(checkpoint common.Hash) (bool, error) { + if c.CheckpointFile == "" { + return false, nil + } + err := os.WriteFile(c.CheckpointFile, []byte(checkpoint.Hex()), 0600) + return err == nil, err +} diff --git a/build/checksums.txt b/build/checksums.txt index 5ccb85bba6..6d3b718c3c 100644 --- a/build/checksums.txt +++ b/build/checksums.txt @@ -54,60 +54,60 @@ db128981033ac82a64688a123f631e61297b6b8f52ca913145e57caa8ce94cc3 go1.24.1.windo e28c4e6d0b913955765b46157ab88ae59bb636acaa12d7bec959aa6900f1cebd go1.24.1.windows-arm64.zip 6d352c1f154a102a5b90c480cc64bab205ccf2681e34e78a3a4d3f1ddfbc81e4 go1.24.1.windows-arm64.msi -# version:golangci 1.64.6 +# version:golangci 2.0.2 # https://github.com/golangci/golangci-lint/releases/ -# https://github.com/golangci/golangci-lint/releases/download/v1.64.6/ -08f9459e7125fed2612abd71596e04d172695921aff82120d1c1e5e9b6fff2a3 golangci-lint-1.64.6-darwin-amd64.tar.gz -8c10d0c7c3935b8c2269d628b6a06a8f48d8fb4cc31af02fe4ce0cd18b0704c1 golangci-lint-1.64.6-darwin-arm64.tar.gz -c07fcabb55deb8d2c4d390025568e76162d7f91b1d14bd2311be45d8d440a624 golangci-lint-1.64.6-freebsd-386.tar.gz -8ed1ef1102e1a42983ffcaae8e06de6a540334c3a69e205c610b8a7c92d469cd golangci-lint-1.64.6-freebsd-amd64.tar.gz -8f8dda66d1b4a85cc8a1daf1b69af6559c3eb0a41dd8033148a9ad85cfc0e1d9 golangci-lint-1.64.6-freebsd-armv6.tar.gz -59e8ca1fa254661b2a55e96515b14a10cd02221d443054deac5b28c3c3e12d6b golangci-lint-1.64.6-freebsd-armv7.tar.gz -e3d323d5f132e9b7593141bfe1d19c8460e65a55cea1008ec96945fed563f981 golangci-lint-1.64.6-illumos-amd64.tar.gz -5ce910f2a864c5fbeb32a30cbd506e1b2ef215f7a0f422cd5be6370b13db6f03 golangci-lint-1.64.6-linux-386.deb -2caab682a26b9a5fb94ba24e3a7e1404fa9eab2c12e36ae1c5548d80a1be190c golangci-lint-1.64.6-linux-386.rpm -2d82d0a4716e6d9b70c95103054855cb4b5f20f7bbdee42297f0189955bd14b6 golangci-lint-1.64.6-linux-386.tar.gz -9cd82503e9841abcaa57663fc899587fe90363c26d86a793a98d3080fd25e907 golangci-lint-1.64.6-linux-amd64.deb -981aaca5e5202d4fbb162ec7080490eb67ef5d32add5fb62fb02210597acc9da golangci-lint-1.64.6-linux-amd64.rpm -71e290acbacb7b3ba4f68f0540fb74dc180c4336ac8a6f3bdcd7fcc48b15148d golangci-lint-1.64.6-linux-amd64.tar.gz -718016bb06c1f598a8d23c7964e2643de621dbe5808688cb38857eb0bb773c84 golangci-lint-1.64.6-linux-arm64.deb -ddc0e7b4b10b47cf894aea157e89e3674bbb60f8f5c480110c21c49cc2c1634d golangci-lint-1.64.6-linux-arm64.rpm -99a7ff13dec7a8781a68408b6ecb8a1c5e82786cba3189eaa91d5cdcc24362ce golangci-lint-1.64.6-linux-arm64.tar.gz -90e360f89c244394912b8709fb83a900b6d56cf19141df2afaf9e902ef3057b1 golangci-lint-1.64.6-linux-armv6.deb -46546ff7c98d873ffcdbee06b39dc1024fc08db4fbf1f6309360a44cf976b5c2 golangci-lint-1.64.6-linux-armv6.rpm -e45c1a42868aca0b0ee54d14fb89da216f3b4409367cd7bce3b5f36788b4fc92 golangci-lint-1.64.6-linux-armv6.tar.gz -3cf6ddbbbf358db3de4b64a24f9683bbe2da3f003cfdee86cb610124b57abafb golangci-lint-1.64.6-linux-armv7.deb -508b6712710da10f11aab9ea5e63af415c932dfe7fff718e649d3100b838f797 golangci-lint-1.64.6-linux-armv7.rpm -da9a8bbee86b4dfee73fbc17e0856ec84c5b04919eb09bf3dd5904c39ce41753 golangci-lint-1.64.6-linux-armv7.tar.gz -ad496a58284e1e9c8ac6f993fec429dcd96c85a8c4715dbb6530a5af0dae7fbd golangci-lint-1.64.6-linux-loong64.deb -3bd70fa737b224750254dce616d9a188570e49b894b0cdb2fd04155e2c061350 golangci-lint-1.64.6-linux-loong64.rpm -a535af973499729f2215a84303eb0de61399057aad6901ddea1b4f73f68d5f2c golangci-lint-1.64.6-linux-loong64.tar.gz -6ad2a1cd37ca30303a488abfca2de52aff57519901c6d8d2608656fe9fb05292 golangci-lint-1.64.6-linux-mips64.deb -5f18369f0ca010a02c267352ebe6e3e0514c6debff49899c9e5520906c0da287 golangci-lint-1.64.6-linux-mips64.rpm -3449d6c13307b91b0e8817f8911c07c3412cdb6931b8d101e42db3e9762e91ad golangci-lint-1.64.6-linux-mips64.tar.gz -d4712a348f65dcaf9f6c58f1cfd6d0984d54a902873dca5e76f0d686f5c59499 golangci-lint-1.64.6-linux-mips64le.deb -57cea4538894558cb8c956d6b69c5509e4304546abe4a467478fc9ada0c736d6 golangci-lint-1.64.6-linux-mips64le.rpm -bc030977d44535b2152fddb2732f056d193c043992fe638ddecea21a40ef09fe golangci-lint-1.64.6-linux-mips64le.tar.gz -1ceb4e492efa527d246c61798c581f49113756fab8c39bb3eefdb1fa97041f92 golangci-lint-1.64.6-linux-ppc64le.deb -454e1c2b3583a8644e0c33a970fb4ce35b8f11848e1a770d9095d99d159ad0ab golangci-lint-1.64.6-linux-ppc64le.rpm -fddf30d35923eb6c7bb57520d8645768f802bf86c4cbf5a3a13049efb9e71b82 golangci-lint-1.64.6-linux-ppc64le.tar.gz -bd75f0dd6f65bee5821c433803b28f3c427ef3582764c3d0e7f7fae1c9d468b6 golangci-lint-1.64.6-linux-riscv64.deb -58457207c225cbd5340c8dcb75d9a44aa890d604e28464115a3a9762febaed04 golangci-lint-1.64.6-linux-riscv64.rpm -82639518a613a6705b7e2de5b28c278e875d782a5c97e9c1b2ac10b4ecd7852f golangci-lint-1.64.6-linux-riscv64.tar.gz -131d69204f8ced200b1437731e987cc886edac30dc87e6e1dcbd4f833d351713 golangci-lint-1.64.6-linux-s390x.deb -ca6caf28ca7a1df7cff720f8fd6ac4b8f2eac10c8cbfe7d2eade70876aded570 golangci-lint-1.64.6-linux-s390x.rpm -ed966d28a6512bc2b1120029a9f78ed77f2015e330b589a731d67d59be30b0b1 golangci-lint-1.64.6-linux-s390x.tar.gz -b181132b41943fc77ace7f9f5523085d12d3613f27774a0e35e17dd5e65ba589 golangci-lint-1.64.6-netbsd-386.tar.gz -f7b81c426006e3c699dc8665797a462aecba8c2cd23ac4971472e55184d95bc9 golangci-lint-1.64.6-netbsd-amd64.tar.gz -663d6fb4c3bef57b8aacdb1b1a4634075e55d294ed170724b443374860897ca6 golangci-lint-1.64.6-netbsd-arm64.tar.gz -8c7a76ee822568cc19352bbb9b2b0863dc5e185eb07782adbbaf648afd02ebcd golangci-lint-1.64.6-netbsd-armv6.tar.gz -0ce26d56ce78e516529e087118ba3f1bcd71d311b4c5b2bde6ec24a6e8966d85 golangci-lint-1.64.6-netbsd-armv7.tar.gz -135d5d998168683f46e4fab308cef5aa3c282025b7de6b258f976aadfb820db7 golangci-lint-1.64.6-source.tar.gz -ccdb0cc249531a205304a76308cfa7cda830083d083d557884416a2461148263 golangci-lint-1.64.6-windows-386.zip -2d88f282e08e1853c70bc7c914b5f58beaa6509903d56098aeb9bc3df30ea9d5 golangci-lint-1.64.6-windows-amd64.zip -6a3c6e7afc6916392679d7d6b95ac239827644e3e50ec4e8ca6ab1e4e6db65fe golangci-lint-1.64.6-windows-arm64.zip -9c9c1ef9687651566987f93e76252f7526c386901d7d8aeee044ca88115da4b1 golangci-lint-1.64.6-windows-armv6.zip -4f0df114155791799dfde8cd8cb6307fecfb17fed70b44205486ec925e2f39cf golangci-lint-1.64.6-windows-armv7.zip +# https://github.com/golangci/golangci-lint/releases/download/v2.0.2/ +a88cbdc86b483fe44e90bf2dcc3fec2af8c754116e6edf0aa6592cac5baa7a0e golangci-lint-2.0.2-darwin-amd64.tar.gz +664550e7954f5f4451aae99b4f7382c1a47039c66f39ca605f5d9af1a0d32b49 golangci-lint-2.0.2-darwin-arm64.tar.gz +bda0f0f27d300502faceda8428834a76ca25986f6d9fc2bd41d201c3ed73f08e golangci-lint-2.0.2-freebsd-386.tar.gz +1cbd0c7ade3fb027d61d38a646ec1b51be5846952b4b04a5330e7f4687f2270c golangci-lint-2.0.2-freebsd-amd64.tar.gz +1e828a597726198b2e35acdbcc5f3aad85244d79846d2d2bdb05241c5a535f9e golangci-lint-2.0.2-freebsd-armv6.tar.gz +848b49315dc5cddd0c9ce35e96ab33d584db0ea8fb57bcbf9784f1622bec0269 golangci-lint-2.0.2-freebsd-armv7.tar.gz +cabf9a6beab574c7f98581eb237919e580024759e3cdc05c4d516b044dce6770 golangci-lint-2.0.2-illumos-amd64.tar.gz +2fde80d15ed6527791f106d606120620e913c3a663c90a8596861d0a4461169e golangci-lint-2.0.2-linux-386.deb +804bc6e350a8c613aaa0a33d8d45414a80157b0ba1b2c2335ac859f85ad98ebd golangci-lint-2.0.2-linux-386.rpm +e64beb72fecf581e57d88ae3adb1c9d4bf022694b6bd92e3c8e460910bbdc37d golangci-lint-2.0.2-linux-386.tar.gz +9c55aed174d7a52bb1d4006b36e7edee9023631f6b814a80cb39c9860d6f75c3 golangci-lint-2.0.2-linux-amd64.deb +c55a2ef741a687b4c679696931f7fd4a467babd64c9457cf17bb9632fd1cecd1 golangci-lint-2.0.2-linux-amd64.rpm +89cc8a7810dc63b9a37900da03e37c3601caf46d42265d774e0f1a5d883d53e2 golangci-lint-2.0.2-linux-amd64.tar.gz +a3e78583c4e7ea1b63e82559f126bb3a5b12788676f158526752d53e67824b99 golangci-lint-2.0.2-linux-arm64.deb +bd5dd52b5c9f18aa7a2904eda9a9f91c628e98623fe70b7afcbb847e2de84422 golangci-lint-2.0.2-linux-arm64.rpm +789d5b91219ac68c2336f77d41cd7e33a910420594780f455893f8453d09595b golangci-lint-2.0.2-linux-arm64.tar.gz +534cd4c464a66178714ed68152c1ed7aa73e5700bf409e4ed1a8363adf96afca golangci-lint-2.0.2-linux-armv6.deb +cf7d02905a5fc80b96c9a64621693b4cc7337b1ce29986c19fd72608dafe66c5 golangci-lint-2.0.2-linux-armv6.rpm +a0d81cb527d8fe878377f2356b5773e219b0b91832a6b59e7b9bcf9a90fe0b0e golangci-lint-2.0.2-linux-armv6.tar.gz +dedd5be7fff8cba8fe15b658a59347ea90d7d02a9fff87f09c7687e6da05a8b6 golangci-lint-2.0.2-linux-armv7.deb +85521b6f3ad2f5a2bc9bfe14b9b08623f764964048f75ed6dfcfaf8eb7d57cc1 golangci-lint-2.0.2-linux-armv7.rpm +96471046c7780dda4ea680f65e92c2ef56ff58d40bcffaf6cfe9d6d48e3c27aa golangci-lint-2.0.2-linux-armv7.tar.gz +815d914a7738e4362466b2d11004e8618b696b49e8ace13df2c2b25f28fb1e17 golangci-lint-2.0.2-linux-loong64.deb +f16381e3d8a0f011b95e086d83d620248432b915d01f4beab4d29cfe4dc388b0 golangci-lint-2.0.2-linux-loong64.rpm +1bd8d7714f9c92db6a0f23bae89f39c85ba047bec8eeb42b8748d30ae3228d18 golangci-lint-2.0.2-linux-loong64.tar.gz +ea6e9d4aabb526aa298e47e8b026d8893d918c5eb919ba0ab403e315def74cc5 golangci-lint-2.0.2-linux-mips64.deb +519d8d53af83fdc9c25cc3fba8b663331ac22ef68131d4b0084cb6f425b6f79a golangci-lint-2.0.2-linux-mips64.rpm +80d655a0a1ac1b19dcef4b58fa2a7dadb646cc50ad08d460b5c53cdb421165e4 golangci-lint-2.0.2-linux-mips64.tar.gz +aa0e75384bb482c865d4dfc95d23ceb25666bf20461b67a832f0eea6670312ec golangci-lint-2.0.2-linux-mips64le.deb +f2a8b500fb69bdea1b01df6267aaa5218fa4a58aeb781c1a20d0d802fe465a52 golangci-lint-2.0.2-linux-mips64le.rpm +e66a0c0c9a275f02d27a7caa9576112622306f001d73dfc082cf1ae446fc1242 golangci-lint-2.0.2-linux-mips64le.tar.gz +e85ad51aac6428be2d8a37000d053697371a538a5bcbc1644caa7c5e77f6d0af golangci-lint-2.0.2-linux-ppc64le.deb +906798365eac1944af2a9b9a303e6fd49ec9043307bc681b7a96277f7f8beea5 golangci-lint-2.0.2-linux-ppc64le.rpm +f7f1a271b0af274d6c9ce000f5dc6e1fb194350c67bcc62494f96f791882ba92 golangci-lint-2.0.2-linux-ppc64le.tar.gz +eea8bf643a42bf05de9780530db22923e5ab0d588f0e173594dc6518f2a25d2a golangci-lint-2.0.2-linux-riscv64.deb +4ff40f9fe2954400836e2a011ba4744d00ffab5068a51368552dfce6aba3b81b golangci-lint-2.0.2-linux-riscv64.rpm +531d8f225866674977d630afbf0533eb02f9bec607fb13895f7a2cd7b2e0a648 golangci-lint-2.0.2-linux-riscv64.tar.gz +6f827647046c603f40d97ea5aadc6f48cd0bb5d19f7a3d56500c3b833d2a0342 golangci-lint-2.0.2-linux-s390x.deb +387a090e9576d19ca86aac738172e58e07c19f2784a13bb387f4f0d75fb9c8d3 golangci-lint-2.0.2-linux-s390x.rpm +57de1fb7722a9feb2d11ed0a007a93959d05b9db5929a392abc222e30012467e golangci-lint-2.0.2-linux-s390x.tar.gz +ed95e0492ea86bf79eb661f0334474b2a4255093685ff587eccd797c5a54db7e golangci-lint-2.0.2-netbsd-386.tar.gz +eab81d729778166415d349a80e568b2f2b3a781745a9be3212a92abb1e732daf golangci-lint-2.0.2-netbsd-amd64.tar.gz +d20add73f7c2de2c3b01ed4fd7b63ffcf0a6597d5ea228d1699e92339a3cd047 golangci-lint-2.0.2-netbsd-arm64.tar.gz +4e4f44e6057879cd62424ff1800a767d25a595c0e91d6d48809eea9186b4c739 golangci-lint-2.0.2-netbsd-armv6.tar.gz +51ec17b16d8743ae4098a0171f04f0ed4d64561e3051b982778b0e6c306a1b03 golangci-lint-2.0.2-netbsd-armv7.tar.gz +5482cf27b93fae1765c70ee2a95d4074d038e9dee61bdd61d017ce8893d3a4a8 golangci-lint-2.0.2-source.tar.gz +a35d8fdf3e14079a10880dbbb7586b46faec89be96f086b244b3e565aac80313 golangci-lint-2.0.2-windows-386.zip +fe4b946cc01366b989001215687003a9c4a7098589921f75e6228d6d8cffc15c golangci-lint-2.0.2-windows-amd64.zip +646bd9250ef8c771d85cd22fe8e6f2397ae39599179755e3bbfa9ef97ad44090 golangci-lint-2.0.2-windows-arm64.zip +ce1dc0bad6f8a61d64e6b3779eeb773479c175125d6f686b0e67ef9c8432d16e golangci-lint-2.0.2-windows-armv6.zip +92684a48faabe792b11ac27ca8b25551eff940b0a1e84ad7244e98b4994962db golangci-lint-2.0.2-windows-armv7.zip # This is the builder on PPA that will build Go itself (inception-y), don't modify! # diff --git a/cmd/blsync/main.go b/cmd/blsync/main.go index 60caa4aa2a..39a9407304 100644 --- a/cmd/blsync/main.go +++ b/cmd/blsync/main.go @@ -43,6 +43,7 @@ func main() { utils.BeaconGenesisRootFlag, utils.BeaconGenesisTimeFlag, utils.BeaconCheckpointFlag, + utils.BeaconCheckpointFileFlag, //TODO datadir for optional permanent database utils.MainnetFlag, utils.SepoliaFlag, diff --git a/cmd/devp2p/internal/v5test/discv5tests.go b/cmd/devp2p/internal/v5test/discv5tests.go index 7dbd3c3be5..2139cd8ca6 100644 --- a/cmd/devp2p/internal/v5test/discv5tests.go +++ b/cmd/devp2p/internal/v5test/discv5tests.go @@ -59,8 +59,9 @@ func (s *Suite) AllTests() []utesting.Test { } } -// TestPing sends PING and expects a PONG response. func (s *Suite) TestPing(t *utesting.T) { + t.Log(`This test is just a sanity check. It sends PING and expects a PONG response.`) + conn, l1 := s.listen1(t) defer conn.close() @@ -85,9 +86,10 @@ func checkPong(t *utesting.T, pong *v5wire.Pong, ping *v5wire.Ping, c net.Packet } } -// TestPingLargeRequestID sends PING with a 9-byte request ID, which isn't allowed by the spec. -// The remote node should not respond. func (s *Suite) TestPingLargeRequestID(t *utesting.T) { + t.Log(`This test sends PING with a 9-byte request ID, which isn't allowed by the spec. +The remote node should not respond.`) + conn, l1 := s.listen1(t) defer conn.close() @@ -104,10 +106,11 @@ func (s *Suite) TestPingLargeRequestID(t *utesting.T) { } } -// TestPingMultiIP establishes a session from one IP as usual. The session is then reused -// on another IP, which shouldn't work. The remote node should respond with WHOAREYOU for -// the attempt from a different IP. func (s *Suite) TestPingMultiIP(t *utesting.T) { + t.Log(`This test establishes a session from one IP as usual. The session is then reused +on another IP, which shouldn't work. The remote node should respond with WHOAREYOU for +the attempt from a different IP.`) + conn, l1, l2 := s.listen2(t) defer conn.close() @@ -120,6 +123,7 @@ func (s *Suite) TestPingMultiIP(t *utesting.T) { checkPong(t, resp.(*v5wire.Pong), ping, l1) // Send on l2. This reuses the session because there is only one codec. + t.Log("sending ping from alternate IP", l2.LocalAddr()) ping2 := &v5wire.Ping{ReqID: conn.nextReqID()} conn.write(l2, ping2, nil) switch resp := conn.read(l2).(type) { @@ -158,6 +162,10 @@ func (s *Suite) TestPingMultiIP(t *utesting.T) { // packet instead of a handshake message packet. The remote node should respond with // another WHOAREYOU challenge for the second packet. func (s *Suite) TestPingHandshakeInterrupted(t *utesting.T) { + t.Log(`TestPingHandshakeInterrupted starts a handshake, but doesn't finish it and sends a second ordinary message +packet instead of a handshake message packet. The remote node should respond with +another WHOAREYOU challenge for the second packet.`) + conn, l1 := s.listen1(t) defer conn.close() @@ -181,8 +189,10 @@ func (s *Suite) TestPingHandshakeInterrupted(t *utesting.T) { } } -// TestTalkRequest sends TALKREQ and expects an empty TALKRESP response. func (s *Suite) TestTalkRequest(t *utesting.T) { + t.Log(`This test sends some examples of TALKREQ with a protocol-id of "test-protocol" +and expects an empty TALKRESP response.`) + conn, l1 := s.listen1(t) defer conn.close() @@ -202,6 +212,7 @@ func (s *Suite) TestTalkRequest(t *utesting.T) { } // Empty request ID. + t.Log("sending TALKREQ with empty request-id") resp = conn.reqresp(l1, &v5wire.TalkRequest{Protocol: "test-protocol"}) switch resp := resp.(type) { case *v5wire.TalkResponse: @@ -216,8 +227,9 @@ func (s *Suite) TestTalkRequest(t *utesting.T) { } } -// TestFindnodeZeroDistance checks that the remote node returns itself for FINDNODE with distance zero. func (s *Suite) TestFindnodeZeroDistance(t *utesting.T) { + t.Log(`This test checks that the remote node returns itself for FINDNODE with distance zero.`) + conn, l1 := s.listen1(t) defer conn.close() @@ -233,9 +245,11 @@ func (s *Suite) TestFindnodeZeroDistance(t *utesting.T) { } } -// TestFindnodeResults pings the node under test from multiple nodes. After waiting for them to be -// accepted into the remote table, the test checks that they are returned by FINDNODE. func (s *Suite) TestFindnodeResults(t *utesting.T) { + t.Log(`This test pings the node under test from multiple other endpoints and node identities +(the 'bystanders'). After waiting for them to be accepted into the remote table, the test checks +that they are returned by FINDNODE.`) + // Create bystanders. nodes := make([]*bystander, 5) added := make(chan enode.ID, len(nodes)) @@ -273,6 +287,7 @@ func (s *Suite) TestFindnodeResults(t *utesting.T) { } // Send FINDNODE for all distances. + t.Log("requesting nodes") conn, l1 := s.listen1(t) defer conn.close() foundNodes, err := conn.findnode(l1, dists) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 708dd8fb7a..2279509542 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -79,11 +79,15 @@ if one is set. Otherwise it prints the genesis from the datadir.`, Usage: "Import a blockchain file", ArgsUsage: " ( ... ) ", Flags: slices.Concat([]cli.Flag{ - utils.CacheFlag, utils.GCModeFlag, utils.SnapshotFlag, + utils.CacheFlag, utils.CacheDatabaseFlag, + utils.CacheTrieFlag, utils.CacheGCFlag, + utils.CacheSnapshotFlag, + utils.CacheNoPrefetchFlag, + utils.CachePreimagesFlag, utils.NoCompactionFlag, utils.MetricsEnabledFlag, utils.MetricsEnabledExpensiveFlag, @@ -457,7 +461,7 @@ func importHistory(ctx *cli.Context) error { network = networks[0] } - if err := utils.ImportHistory(chain, db, dir, network); err != nil { + if err := utils.ImportHistory(chain, dir, network); err != nil { return err } fmt.Printf("Import done in %v\n", time.Since(start)) diff --git a/cmd/geth/main.go b/cmd/geth/main.go index cd74fb7b6a..ab46e059f3 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -157,6 +157,7 @@ var ( utils.BeaconGenesisRootFlag, utils.BeaconGenesisTimeFlag, utils.BeaconCheckpointFlag, + utils.BeaconCheckpointFileFlag, }, utils.NetworkFlags, utils.DatabaseFlags) rpcFlags = []cli.Flag{ diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index d91d6ca5a8..d34e15ebc0 100644 --- a/cmd/utils/cmd.go +++ b/cmd/utils/cmd.go @@ -246,8 +246,9 @@ func readList(filename string) ([]string, error) { } // ImportHistory imports Era1 files containing historical block information, -// starting from genesis. -func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, network string) error { +// starting from genesis. The assumption is held that the provided chain +// segment in Era1 file should all be canonical and verified. +func ImportHistory(chain *core.BlockChain, dir string, network string) error { if chain.CurrentSnapBlock().Number.BitLen() != 0 { return errors.New("history import only supported when starting from genesis") } @@ -308,11 +309,6 @@ func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, networ if err != nil { return fmt.Errorf("error reading receipts %d: %w", it.Number(), err) } - if status, err := chain.HeaderChain().InsertHeaderChain([]*types.Header{block.Header()}, start); err != nil { - return fmt.Errorf("error inserting header %d: %w", it.Number(), err) - } else if status != core.CanonStatTy { - return fmt.Errorf("error inserting header %d, not canon: %v", it.Number(), status) - } if _, err := chain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{receipts}, 2^64-1); err != nil { return fmt.Errorf("error inserting body %d: %w", it.Number(), err) } diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index fb2892d2c1..f5fc94cebc 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -342,6 +342,11 @@ var ( Usage: "Beacon chain weak subjectivity checkpoint block hash", Category: flags.BeaconCategory, } + BeaconCheckpointFileFlag = &cli.StringFlag{ + Name: "beacon.checkpoint.file", + Usage: "Beacon chain weak subjectivity checkpoint import/export file", + Category: flags.BeaconCategory, + } BlsyncApiFlag = &cli.StringFlag{ Name: "blsync.engine.api", Usage: "Target EL engine API URL", @@ -1890,7 +1895,7 @@ func MakeBeaconLightConfig(ctx *cli.Context) bparams.ClientConfig { if !ctx.IsSet(BeaconGenesisTimeFlag.Name) { Fatalf("Custom beacon chain config is specified but genesis time is missing") } - if !ctx.IsSet(BeaconCheckpointFlag.Name) { + if !ctx.IsSet(BeaconCheckpointFlag.Name) && !ctx.IsSet(BeaconCheckpointFileFlag.Name) { Fatalf("Custom beacon chain config is specified but checkpoint is missing") } config.ChainConfig = bparams.ChainConfig{ @@ -1915,13 +1920,28 @@ func MakeBeaconLightConfig(ctx *cli.Context) bparams.ClientConfig { } } // Checkpoint is required with custom chain config and is optional with pre-defined config - if ctx.IsSet(BeaconCheckpointFlag.Name) { - if c, err := hexutil.Decode(ctx.String(BeaconCheckpointFlag.Name)); err == nil && len(c) <= 32 { - copy(config.Checkpoint[:len(c)], c) - } else { - Fatalf("Invalid hex string", "beacon.checkpoint", ctx.String(BeaconCheckpointFlag.Name), "error", err) + // If both checkpoint block hash and checkpoint file are specified then the + // client is initialized with the specified block hash and new checkpoints + // are saved to the specified file. + if ctx.IsSet(BeaconCheckpointFileFlag.Name) { + if _, err := config.SetCheckpointFile(ctx.String(BeaconCheckpointFileFlag.Name)); err != nil { + Fatalf("Could not load beacon checkpoint file", "beacon.checkpoint.file", ctx.String(BeaconCheckpointFileFlag.Name), "error", err) } } + if ctx.IsSet(BeaconCheckpointFlag.Name) { + hex := ctx.String(BeaconCheckpointFlag.Name) + c, err := hexutil.Decode(hex) + if err != nil { + Fatalf("Invalid hex string", "beacon.checkpoint", hex, "error", err) + } + if len(c) != 32 { + Fatalf("Invalid hex string length", "beacon.checkpoint", hex, "length", len(c)) + } + copy(config.Checkpoint[:len(c)], c) + } + if config.Checkpoint == (common.Hash{}) { + Fatalf("Beacon checkpoint not specified") + } config.Apis = ctx.StringSlice(BeaconApiFlag.Name) if config.Apis == nil { Fatalf("Beacon node light client API URL not specified") diff --git a/cmd/utils/history_test.go b/cmd/utils/history_test.go index 8654c454f9..d3c6bda1c5 100644 --- a/cmd/utils/history_test.go +++ b/cmd/utils/history_test.go @@ -171,7 +171,7 @@ func TestHistoryImportAndExport(t *testing.T) { if err != nil { t.Fatalf("unable to initialize chain: %v", err) } - if err := ImportHistory(imported, db2, dir, "mainnet"); err != nil { + if err := ImportHistory(imported, dir, "mainnet"); err != nil { t.Fatalf("failed to import chain: %v", err) } if have, want := imported.CurrentHeader(), chain.CurrentHeader(); have.Hash() != want.Hash() { diff --git a/core/bench_test.go b/core/bench_test.go index 155fa6c3b5..00f924076a 100644 --- a/core/bench_test.go +++ b/core/bench_test.go @@ -183,7 +183,7 @@ func benchInsertChain(b *testing.B, disk bool, gen func(int, *BlockGen)) { if !disk { db = rawdb.NewMemoryDatabase() } else { - pdb, err := pebble.New(b.TempDir(), 128, 128, "", false) + pdb, err := pebble.New(b.TempDir(), 128, 128, "", false, true) if err != nil { b.Fatalf("cannot create temporary database: %v", err) } @@ -303,7 +303,7 @@ func makeChainForBench(db ethdb.Database, genesis *Genesis, full bool, count uin func benchWriteChain(b *testing.B, full bool, count uint64) { genesis := &Genesis{Config: params.AllEthashProtocolChanges} for i := 0; i < b.N; i++ { - pdb, err := pebble.New(b.TempDir(), 1024, 128, "", false) + pdb, err := pebble.New(b.TempDir(), 1024, 128, "", false, true) if err != nil { b.Fatalf("error opening database: %v", err) } @@ -316,7 +316,7 @@ func benchWriteChain(b *testing.B, full bool, count uint64) { func benchReadChain(b *testing.B, full bool, count uint64) { dir := b.TempDir() - pdb, err := pebble.New(dir, 1024, 128, "", false) + pdb, err := pebble.New(dir, 1024, 128, "", false, true) if err != nil { b.Fatalf("error opening database: %v", err) } @@ -332,7 +332,7 @@ func benchReadChain(b *testing.B, full bool, count uint64) { b.ResetTimer() for i := 0; i < b.N; i++ { - pdb, err = pebble.New(dir, 1024, 128, "", false) + pdb, err = pebble.New(dir, 1024, 128, "", false, true) if err != nil { b.Fatalf("error opening database: %v", err) } diff --git a/core/blockchain.go b/core/blockchain.go index d80236c902..d56996dadb 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -24,6 +24,7 @@ import ( "math/big" "runtime" "slices" + "sort" "strings" "sync" "sync/atomic" @@ -157,7 +158,8 @@ type CacheConfig struct { // This defines the cutoff block for history expiry. // Blocks before this number may be unavailable in the chain database. - HistoryPruningCutoff uint64 + HistoryPruningCutoffNumber uint64 + HistoryPruningCutoffHash common.Hash } // triedbConfig derives the configures for trie database. @@ -262,7 +264,6 @@ type BlockChain struct { txLookupLock sync.RWMutex txLookupCache *lru.Cache[common.Hash, txLookup] - wg sync.WaitGroup quit chan struct{} // shutdown signal, closed in Stop. stopping atomic.Bool // false if chain is running, true when stopped procInterrupt atomic.Bool // interrupt signaler for block processing @@ -333,10 +334,10 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis bc.processor = NewStateProcessor(chainConfig, bc.hc) genesisHeader := bc.GetHeaderByNumber(0) - bc.genesisBlock = types.NewBlockWithHeader(genesisHeader) - if bc.genesisBlock == nil { + if genesisHeader == nil { return nil, ErrNoGenesis } + bc.genesisBlock = types.NewBlockWithHeader(genesisHeader) bc.currentBlock.Store(nil) bc.currentSnapBlock.Store(nil) @@ -1110,7 +1111,6 @@ func (bc *BlockChain) stopWithoutSaving() { // the mutex should become available quickly. It cannot be taken again after Close has // returned. bc.chainmu.Close() - bc.wg.Wait() } // Stop stops the blockchain service. If any imports are currently in progress @@ -1197,79 +1197,64 @@ const ( SideStatTy ) -// InsertReceiptChain attempts to complete an already existing header chain with -// transaction and receipt data. +// InsertReceiptChain inserts a batch of blocks along with their receipts into +// the database. Unlike InsertChain, this function does not verify the state root +// in the blocks. It is used exclusively for snap sync. All the inserted blocks +// will be regarded as canonical, chain reorg is not supported. +// +// The optional ancientLimit can also be specified and chain segment before that +// will be directly stored in the ancient, getting rid of the chain migration. func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain []types.Receipts, ancientLimit uint64) (int, error) { - // We don't require the chainMu here since we want to maximize the - // concurrency of header insertion and receipt insertion. - bc.wg.Add(1) - defer bc.wg.Done() + // Verify the supplied headers before insertion without lock + var headers []*types.Header + for _, block := range blockChain { + headers = append(headers, block.Header()) - var ( - ancientBlocks, liveBlocks types.Blocks - ancientReceipts, liveReceipts []types.Receipts - ) - // Do a sanity check that the provided chain is actually ordered and linked - for i, block := range blockChain { - if i != 0 { - prev := blockChain[i-1] - if block.NumberU64() != prev.NumberU64()+1 || block.ParentHash() != prev.Hash() { - log.Error("Non contiguous receipt insert", - "number", block.Number(), "hash", block.Hash(), "parent", block.ParentHash(), - "prevnumber", prev.Number(), "prevhash", prev.Hash()) - return 0, fmt.Errorf("non contiguous insert: item %d is #%d [%x..], item %d is #%d [%x..] (parent [%x..])", - i-1, prev.NumberU64(), prev.Hash().Bytes()[:4], - i, block.NumberU64(), block.Hash().Bytes()[:4], block.ParentHash().Bytes()[:4]) - } - } - if block.NumberU64() <= ancientLimit { - ancientBlocks, ancientReceipts = append(ancientBlocks, block), append(ancientReceipts, receiptChain[i]) - } else { - liveBlocks, liveReceipts = append(liveBlocks, block), append(liveReceipts, receiptChain[i]) - } - - // Here we also validate that blob transactions in the block do not contain a sidecar. - // While the sidecar does not affect the block hash / tx hash, sending blobs within a block is not allowed. + // Here we also validate that blob transactions in the block do not + // contain a sidecar. While the sidecar does not affect the block hash + // or tx hash, sending blobs within a block is not allowed. for txIndex, tx := range block.Transactions() { if tx.Type() == types.BlobTxType && tx.BlobTxSidecar() != nil { return 0, fmt.Errorf("block #%d contains unexpected blob sidecar in tx at index %d", block.NumberU64(), txIndex) } } } + if n, err := bc.hc.ValidateHeaderChain(headers); err != nil { + return n, err + } + // Hold the mutation lock + if !bc.chainmu.TryLock() { + return 0, errChainStopped + } + defer bc.chainmu.Unlock() var ( stats = struct{ processed, ignored int32 }{} start = time.Now() size = int64(0) ) - - // updateHead updates the head snap sync block if the inserted blocks are better - // and returns an indicator whether the inserted blocks are canonical. - updateHead := func(head *types.Block) bool { - if !bc.chainmu.TryLock() { - return false + // updateHead updates the head header and head snap block flags. + updateHead := func(header *types.Header) error { + batch := bc.db.NewBatch() + hash := header.Hash() + rawdb.WriteHeadHeaderHash(batch, hash) + rawdb.WriteHeadFastBlockHash(batch, hash) + if err := batch.Write(); err != nil { + return err } - defer bc.chainmu.Unlock() - - // Rewind may have occurred, skip in that case. - if bc.CurrentHeader().Number.Cmp(head.Number()) >= 0 { - rawdb.WriteHeadFastBlockHash(bc.db, head.Hash()) - bc.currentSnapBlock.Store(head.Header()) - headFastBlockGauge.Update(int64(head.NumberU64())) - return true - } - return false + bc.hc.currentHeader.Store(header) + bc.currentSnapBlock.Store(header) + headHeaderGauge.Update(header.Number.Int64()) + headFastBlockGauge.Update(header.Number.Int64()) + return nil } // writeAncient writes blockchain and corresponding receipt chain into ancient store. // // this function only accepts canonical chain data. All side chain will be reverted // eventually. writeAncient := func(blockChain types.Blocks, receiptChain []types.Receipts) (int, error) { - first := blockChain[0] - last := blockChain[len(blockChain)-1] - - // Ensure genesis is in ancients. - if first.NumberU64() == 1 { + // Ensure genesis is in the ancient store + if blockChain[0].NumberU64() == 1 { if frozen, _ := bc.db.Ancients(); frozen == 0 { writeSize, err := rawdb.WriteAncientBlocks(bc.db, []*types.Block{bc.genesisBlock}, []types.Receipts{nil}) if err != nil { @@ -1280,12 +1265,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ log.Info("Wrote genesis to ancients") } } - // Before writing the blocks to the ancients, we need to ensure that - // they correspond to the what the headerchain 'expects'. - // We only check the last block/header, since it's a contiguous chain. - if !bc.HasHeader(last.Hash(), last.NumberU64()) { - return 0, fmt.Errorf("containing header #%d [%x..] unknown", last.Number(), last.Hash().Bytes()[:4]) - } // Write all chain data to ancients. writeSize, err := rawdb.WriteAncientBlocks(bc.db, blockChain, receiptChain) if err != nil { @@ -1298,44 +1277,28 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ if err := bc.db.Sync(); err != nil { return 0, err } - // Update the current snap block because all block data is now present in DB. - previousSnapBlock := bc.CurrentSnapBlock().Number.Uint64() - if !updateHead(blockChain[len(blockChain)-1]) { - // We end up here if the header chain has reorg'ed, and the blocks/receipts - // don't match the canonical chain. - if _, err := bc.db.TruncateHead(previousSnapBlock + 1); err != nil { - log.Error("Can't truncate ancient store after failed insert", "err", err) - } - return 0, errSideChainReceipts - } - - // Delete block data from the main database. - var ( - batch = bc.db.NewBatch() - canonHashes = make(map[common.Hash]struct{}, len(blockChain)) - ) + // Write hash to number mappings + batch := bc.db.NewBatch() for _, block := range blockChain { - canonHashes[block.Hash()] = struct{}{} - if block.NumberU64() == 0 { - continue - } - rawdb.DeleteCanonicalHash(batch, block.NumberU64()) - rawdb.DeleteBlockWithoutNumber(batch, block.Hash(), block.NumberU64()) - } - // Delete side chain hash-to-number mappings. - for _, nh := range rawdb.ReadAllHashesInRange(bc.db, first.NumberU64(), last.NumberU64()) { - if _, canon := canonHashes[nh.Hash]; !canon { - rawdb.DeleteHeader(batch, nh.Hash, nh.Number) - } + rawdb.WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) } if err := batch.Write(); err != nil { return 0, err } + // Update the current snap block because all block data is now present in DB. + if err := updateHead(blockChain[len(blockChain)-1].Header()); err != nil { + return 0, err + } stats.processed += int32(len(blockChain)) return 0, nil } - // writeLive writes blockchain and corresponding receipt chain into active store. + // writeLive writes the blockchain and corresponding receipt chain to the active store. + // + // Notably, in different snap sync cycles, the supplied chain may partially reorganize + // existing local chain segments (reorg around the chain tip). The reorganized part + // will be included in the provided chain segment, and stale canonical markers will be + // silently rewritten. Therefore, no explicit reorg logic is needed. writeLive := func(blockChain types.Blocks, receiptChain []types.Receipts) (int, error) { var ( skipPresenceCheck = false @@ -1346,10 +1309,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ if bc.insertStopped() { return 0, errInsertionInterrupted } - // Short circuit if the owner header is unknown - if !bc.HasHeader(block.Hash(), block.NumberU64()) { - return i, fmt.Errorf("containing header #%d [%x..] unknown", block.Number(), block.Hash().Bytes()[:4]) - } if !skipPresenceCheck { // Ignore if the entire data is already known if bc.HasBlock(block.Hash(), block.NumberU64()) { @@ -1363,7 +1322,8 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ } } // Write all the data out into the database - rawdb.WriteBody(batch, block.Hash(), block.NumberU64(), block.Body()) + rawdb.WriteCanonicalHash(batch, block.Hash(), block.NumberU64()) + rawdb.WriteBlock(batch, block) rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i]) // Write everything belongs to the blocks into the database. So that @@ -1387,21 +1347,27 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ return 0, err } } - updateHead(blockChain[len(blockChain)-1]) + if err := updateHead(blockChain[len(blockChain)-1].Header()); err != nil { + return 0, err + } return 0, nil } - // Write downloaded chain data and corresponding receipt chain data - if len(ancientBlocks) > 0 { - if n, err := writeAncient(ancientBlocks, ancientReceipts); err != nil { + // Split the supplied blocks into two groups, according to the + // given ancient limit. + index := sort.Search(len(blockChain), func(i int) bool { + return blockChain[i].NumberU64() >= ancientLimit + }) + if index > 0 { + if n, err := writeAncient(blockChain[:index], receiptChain[:index]); err != nil { if err == errInsertionInterrupted { return 0, nil } return n, err } } - if len(liveBlocks) > 0 { - if n, err := writeLive(liveBlocks, liveReceipts); err != nil { + if index != len(blockChain) { + if n, err := writeLive(blockChain[index:], receiptChain[index:]); err != nil { if err == errInsertionInterrupted { return 0, nil } @@ -1420,7 +1386,6 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [ context = append(context, []interface{}{"ignored", stats.ignored}...) } log.Debug("Imported new block receipts", context...) - return 0, nil } @@ -2505,15 +2470,83 @@ func (bc *BlockChain) InsertHeaderChain(chain []*types.Header) (int, error) { if i, err := bc.hc.ValidateHeaderChain(chain); err != nil { return i, err } - if !bc.chainmu.TryLock() { return 0, errChainStopped } defer bc.chainmu.Unlock() + _, err := bc.hc.InsertHeaderChain(chain, start) return 0, err } +// InsertHeadersBeforeCutoff inserts the given headers into the ancient store +// as they are claimed older than the configured chain cutoff point. All the +// inserted headers are regarded as canonical and chain reorg is not supported. +func (bc *BlockChain) InsertHeadersBeforeCutoff(headers []*types.Header) (int, error) { + if len(headers) == 0 { + return 0, nil + } + // TODO(rjl493456442): Headers before the configured cutoff have already + // been verified by the hash of cutoff header. Theoretically, header validation + // could be skipped here. + if n, err := bc.hc.ValidateHeaderChain(headers); err != nil { + return n, err + } + if !bc.chainmu.TryLock() { + return 0, errChainStopped + } + defer bc.chainmu.Unlock() + + // Initialize the ancient store with genesis block if it's empty. + var ( + frozen, _ = bc.db.Ancients() + first = headers[0].Number.Uint64() + ) + if first == 1 && frozen == 0 { + _, err := rawdb.WriteAncientBlocks(bc.db, []*types.Block{bc.genesisBlock}, []types.Receipts{nil}) + if err != nil { + log.Error("Error writing genesis to ancients", "err", err) + return 0, err + } + log.Info("Wrote genesis to ancient store") + } else if frozen != first { + return 0, fmt.Errorf("headers are gapped with the ancient store, first: %d, ancient: %d", first, frozen) + } + + // Write headers to the ancient store, with block bodies and receipts set to nil + // to ensure consistency across tables in the freezer. + _, err := rawdb.WriteAncientHeaderChain(bc.db, headers) + if err != nil { + return 0, err + } + if err := bc.db.Sync(); err != nil { + return 0, err + } + // Write hash to number mappings + batch := bc.db.NewBatch() + for _, header := range headers { + rawdb.WriteHeaderNumber(batch, header.Hash(), header.Number.Uint64()) + } + // Write head header and head snap block flags + last := headers[len(headers)-1] + rawdb.WriteHeadHeaderHash(batch, last.Hash()) + rawdb.WriteHeadFastBlockHash(batch, last.Hash()) + if err := batch.Write(); err != nil { + return 0, err + } + // Truncate the useless chain segment (zero bodies and receipts) in the + // ancient store. + if _, err := bc.db.TruncateTail(last.Number.Uint64() + 1); err != nil { + return 0, err + } + // Last step update all in-memory markers + bc.hc.currentHeader.Store(last) + bc.currentSnapBlock.Store(last) + headHeaderGauge.Update(last.Number.Int64()) + headFastBlockGauge.Update(last.Number.Int64()) + return 0, nil +} + // SetBlockValidatorAndProcessorForTesting sets the current validator and processor. // This method can be used to force an invalid blockchain to be verified for tests. // This method is unsafe and should only be used before block import starts. @@ -2533,9 +2566,3 @@ func (bc *BlockChain) SetTrieFlushInterval(interval time.Duration) { func (bc *BlockChain) GetTrieFlushInterval() time.Duration { return time.Duration(bc.flushInterval.Load()) } - -// HistoryPruningCutoff returns the configured history pruning point. -// Blocks before this might not be available in the database. -func (bc *BlockChain) HistoryPruningCutoff() uint64 { - return bc.cacheConfig.HistoryPruningCutoff -} diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 415a0f5442..4114723469 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -407,6 +407,12 @@ func (bc *BlockChain) TxIndexProgress() (TxIndexProgress, error) { return bc.txIndexer.txIndexProgress() } +// HistoryPruningCutoff returns the configured history pruning point. +// Blocks before this might not be available in the database. +func (bc *BlockChain) HistoryPruningCutoff() (uint64, common.Hash) { + return bc.cacheConfig.HistoryPruningCutoffNumber, bc.cacheConfig.HistoryPruningCutoffHash +} + // TrieDB retrieves the low level trie database used for data storage. func (bc *BlockChain) TrieDB() *triedb.Database { return bc.triedb diff --git a/core/blockchain_repair_test.go b/core/blockchain_repair_test.go index 6c52d057ad..3ff1d77fc8 100644 --- a/core/blockchain_repair_test.go +++ b/core/blockchain_repair_test.go @@ -1765,7 +1765,7 @@ func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme s datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false) + pdb, err := pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -1850,7 +1850,7 @@ func testRepairWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme s chain.stopWithoutSaving() // Start a new blockchain back up and see where the repair leads us - pdb, err = pebble.New(datadir, 0, 0, "", false) + pdb, err = pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to reopen persistent key-value database: %v", err) } @@ -1915,7 +1915,7 @@ func testIssue23496(t *testing.T, scheme string) { datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false) + pdb, err := pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -1973,7 +1973,7 @@ func testIssue23496(t *testing.T, scheme string) { chain.stopWithoutSaving() // Start a new blockchain back up and see where the repair leads us - pdb, err = pebble.New(datadir, 0, 0, "", false) + pdb, err = pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to reopen persistent key-value database: %v", err) } diff --git a/core/blockchain_sethead_test.go b/core/blockchain_sethead_test.go index 424854b2bf..e998b510df 100644 --- a/core/blockchain_sethead_test.go +++ b/core/blockchain_sethead_test.go @@ -1969,7 +1969,7 @@ func testSetHeadWithScheme(t *testing.T, tt *rewindTest, snapshots bool, scheme datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false) + pdb, err := pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } diff --git a/core/blockchain_snapshot_test.go b/core/blockchain_snapshot_test.go index 1a6fe38af6..23effea15e 100644 --- a/core/blockchain_snapshot_test.go +++ b/core/blockchain_snapshot_test.go @@ -66,7 +66,7 @@ func (basic *snapshotTestBasic) prepare(t *testing.T) (*BlockChain, []*types.Blo datadir := t.TempDir() ancient := filepath.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false) + pdb, err := pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -257,7 +257,7 @@ func (snaptest *crashSnapshotTest) test(t *testing.T) { chain.triedb.Close() // Start a new blockchain back up and see where the repair leads us - pdb, err := pebble.New(snaptest.datadir, 0, 0, "", false) + pdb, err := pebble.New(snaptest.datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } diff --git a/core/blockchain_test.go b/core/blockchain_test.go index 04dfa87b8b..3f7c03b93c 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -733,13 +733,6 @@ func testFastVsFullChains(t *testing.T, scheme string) { fast, _ := NewBlockChain(fastDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) defer fast.Stop() - headers := make([]*types.Header, len(blocks)) - for i, block := range blocks { - headers[i] = block.Header() - } - if n, err := fast.InsertHeaderChain(headers); err != nil { - t.Fatalf("failed to insert header %d: %v", n, err) - } if n, err := fast.InsertReceiptChain(blocks, receipts, 0); err != nil { t.Fatalf("failed to insert receipt %d: %v", n, err) } @@ -753,9 +746,6 @@ func testFastVsFullChains(t *testing.T, scheme string) { ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) defer ancient.Stop() - if n, err := ancient.InsertHeaderChain(headers); err != nil { - t.Fatalf("failed to insert header %d: %v", n, err) - } if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(len(blocks)/2)); err != nil { t.Fatalf("failed to insert receipt %d: %v", n, err) } @@ -880,13 +870,6 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) { fast, _ := NewBlockChain(fastDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) defer fast.Stop() - headers := make([]*types.Header, len(blocks)) - for i, block := range blocks { - headers[i] = block.Header() - } - if n, err := fast.InsertHeaderChain(headers); err != nil { - t.Fatalf("failed to insert header %d: %v", n, err) - } if n, err := fast.InsertReceiptChain(blocks, receipts, 0); err != nil { t.Fatalf("failed to insert receipt %d: %v", n, err) } @@ -900,9 +883,6 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) { ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) defer ancient.Stop() - if n, err := ancient.InsertHeaderChain(headers); err != nil { - t.Fatalf("failed to insert header %d: %v", n, err) - } if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(3*len(blocks)/4)); err != nil { t.Fatalf("failed to insert receipt %d: %v", n, err) } @@ -916,6 +896,11 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) { // Import the chain as a light node and ensure all pointers are updated lightDb := makeDb() defer lightDb.Close() + + headers := make([]*types.Header, len(blocks)) + for i, block := range blocks { + headers[i] = block.Header() + } light, _ := NewBlockChain(lightDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) if n, err := light.InsertHeaderChain(headers); err != nil { t.Fatalf("failed to insert header %d: %v", n, err) @@ -1710,13 +1695,6 @@ func testBlockchainRecovery(t *testing.T, scheme string) { defer ancientDb.Close() ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) - headers := make([]*types.Header, len(blocks)) - for i, block := range blocks { - headers[i] = block.Header() - } - if n, err := ancient.InsertHeaderChain(headers); err != nil { - t.Fatalf("failed to insert header %d: %v", n, err) - } if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(3*len(blocks)/4)); err != nil { t.Fatalf("failed to insert receipt %d: %v", n, err) } @@ -1741,82 +1719,6 @@ func testBlockchainRecovery(t *testing.T, scheme string) { } } -// This test checks that InsertReceiptChain will roll back correctly when attempting to insert a side chain. -func TestInsertReceiptChainRollback(t *testing.T) { - testInsertReceiptChainRollback(t, rawdb.HashScheme) - testInsertReceiptChainRollback(t, rawdb.PathScheme) -} - -func testInsertReceiptChainRollback(t *testing.T, scheme string) { - // Generate forked chain. The returned BlockChain object is used to process the side chain blocks. - tmpChain, sideblocks, canonblocks, gspec, err := getLongAndShortChains(scheme) - if err != nil { - t.Fatal(err) - } - defer tmpChain.Stop() - // Get the side chain receipts. - if _, err := tmpChain.InsertChain(sideblocks); err != nil { - t.Fatal("processing side chain failed:", err) - } - t.Log("sidechain head:", tmpChain.CurrentBlock().Number, tmpChain.CurrentBlock().Hash()) - sidechainReceipts := make([]types.Receipts, len(sideblocks)) - for i, block := range sideblocks { - sidechainReceipts[i] = tmpChain.GetReceiptsByHash(block.Hash()) - } - // Get the canon chain receipts. - if _, err := tmpChain.InsertChain(canonblocks); err != nil { - t.Fatal("processing canon chain failed:", err) - } - t.Log("canon head:", tmpChain.CurrentBlock().Number, tmpChain.CurrentBlock().Hash()) - canonReceipts := make([]types.Receipts, len(canonblocks)) - for i, block := range canonblocks { - canonReceipts[i] = tmpChain.GetReceiptsByHash(block.Hash()) - } - - // Set up a BlockChain that uses the ancient store. - ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false) - if err != nil { - t.Fatalf("failed to create temp freezer db: %v", err) - } - defer ancientDb.Close() - - ancientChain, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil) - defer ancientChain.Stop() - - // Import the canonical header chain. - canonHeaders := make([]*types.Header, len(canonblocks)) - for i, block := range canonblocks { - canonHeaders[i] = block.Header() - } - if _, err = ancientChain.InsertHeaderChain(canonHeaders); err != nil { - t.Fatal("can't import canon headers:", err) - } - - // Try to insert blocks/receipts of the side chain. - _, err = ancientChain.InsertReceiptChain(sideblocks, sidechainReceipts, uint64(len(sideblocks))) - if err == nil { - t.Fatal("expected error from InsertReceiptChain.") - } - if ancientChain.CurrentSnapBlock().Number.Uint64() != 0 { - t.Fatalf("failed to rollback ancient data, want %d, have %d", 0, ancientChain.CurrentSnapBlock().Number) - } - if frozen, err := ancientChain.db.Ancients(); err != nil || frozen != 1 { - t.Fatalf("failed to truncate ancient data, frozen index is %d", frozen) - } - - // Insert blocks/receipts of the canonical chain. - _, err = ancientChain.InsertReceiptChain(canonblocks, canonReceipts, uint64(len(canonblocks))) - if err != nil { - t.Fatalf("can't import canon chain receipts: %v", err) - } - if ancientChain.CurrentSnapBlock().Number.Uint64() != canonblocks[len(canonblocks)-1].NumberU64() { - t.Fatalf("failed to insert ancient recept chain after rollback") - } - if frozen, _ := ancientChain.db.Ancients(); frozen != uint64(len(canonblocks))+1 { - t.Fatalf("wrong ancients count %d", frozen) - } -} - // Tests that importing a very large side fork, which is larger than the canon chain, // but where the difficulty per block is kept low: this means that it will not // overtake the 'canon' chain until after it's passed canon by about 200 blocks. @@ -2088,14 +1990,6 @@ func testInsertKnownChainData(t *testing.T, typ string, scheme string) { } } else if typ == "receipts" { inserter = func(blocks []*types.Block, receipts []types.Receipts) error { - headers := make([]*types.Header, 0, len(blocks)) - for _, block := range blocks { - headers = append(headers, block.Header()) - } - _, err := chain.InsertHeaderChain(headers) - if err != nil { - return err - } _, err = chain.InsertReceiptChain(blocks, receipts, 0) return err } @@ -2262,14 +2156,6 @@ func testInsertKnownChainDataWithMerging(t *testing.T, typ string, mergeHeight i } } else if typ == "receipts" { inserter = func(blocks []*types.Block, receipts []types.Receipts) error { - headers := make([]*types.Header, 0, len(blocks)) - for _, block := range blocks { - headers = append(headers, block.Header()) - } - i, err := chain.InsertHeaderChain(headers) - if err != nil { - return fmt.Errorf("index %d: %w", i, err) - } _, err = chain.InsertReceiptChain(blocks, receipts, 0) return err } @@ -2605,7 +2491,7 @@ func testSideImportPrunedBlocks(t *testing.T, scheme string) { datadir := t.TempDir() ancient := path.Join(datadir, "ancient") - pdb, err := pebble.New(datadir, 0, 0, "", false) + pdb, err := pebble.New(datadir, 0, 0, "", false, true) if err != nil { t.Fatalf("Failed to create persistent key-value database: %v", err) } @@ -4265,3 +4151,220 @@ func TestEIP7702(t *testing.T) { t.Fatalf("addr2 storage wrong: expected %d, got %d", fortyTwo, actual) } } + +// Tests the scenario that the synchronization target in snap sync has been changed +// with a chain reorg at the tip. In this case the reorg'd segment should be unmarked +// with canonical flags. +func TestChainReorgSnapSync(t *testing.T) { + testChainReorgSnapSync(t, 0) + testChainReorgSnapSync(t, 32) + testChainReorgSnapSync(t, gomath.MaxUint64) +} + +func testChainReorgSnapSync(t *testing.T, ancientLimit uint64) { + // log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true))) + + // Configure and generate a sample block chain + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Config: params.TestChainConfig, + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + } + signer = types.LatestSigner(gspec.Config) + engine = beacon.New(ethash.NewFaker()) + ) + genDb, blocks, receipts := GenerateChainWithGenesis(gspec, engine, 32, func(i int, block *BlockGen) { + block.SetCoinbase(common.Address{0x00}) + + // If the block number is multiple of 3, send a few bonus transactions to the miner + if i%3 == 2 { + for j := 0; j < i%4+1; j++ { + tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, block.header.BaseFee, nil), signer, key) + if err != nil { + panic(err) + } + block.AddTx(tx) + } + } + }) + chainA, receiptsA := GenerateChain(gspec.Config, blocks[len(blocks)-1], engine, genDb, 16, func(i int, gen *BlockGen) { + gen.SetCoinbase(common.Address{0: byte(0xa), 19: byte(i)}) + }) + chainB, receiptsB := GenerateChain(gspec.Config, blocks[len(blocks)-1], engine, genDb, 20, func(i int, gen *BlockGen) { + gen.SetCoinbase(common.Address{0: byte(0xb), 19: byte(i)}) + }) + + db, _ := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false) + defer db.Close() + + chain, _ := NewBlockChain(db, DefaultCacheConfigWithScheme(rawdb.PathScheme), gspec, nil, beacon.New(ethash.NewFaker()), vm.Config{}, nil) + defer chain.Stop() + + if n, err := chain.InsertReceiptChain(blocks, receipts, ancientLimit); err != nil { + t.Fatalf("failed to insert receipt %d: %v", n, err) + } + if n, err := chain.InsertReceiptChain(chainA, receiptsA, ancientLimit); err != nil { + t.Fatalf("failed to insert receipt %d: %v", n, err) + } + // If the common ancestor is below the ancient limit, rewind the chain head. + // It's aligned with the behavior in the snap sync + ancestor := blocks[len(blocks)-1].NumberU64() + if ancestor < ancientLimit { + rawdb.WriteLastPivotNumber(db, ancestor) + chain.SetHead(ancestor) + } + if n, err := chain.InsertReceiptChain(chainB, receiptsB, ancientLimit); err != nil { + t.Fatalf("failed to insert receipt %d: %v", n, err) + } + head := chain.CurrentSnapBlock() + if head.Hash() != chainB[len(chainB)-1].Hash() { + t.Errorf("head snap block #%d: header mismatch: want: %v, got: %v", head.Number, chainB[len(chainB)-1].Hash(), head.Hash()) + } + + // Iterate over all chain data components, and cross reference + for i := 0; i < len(blocks); i++ { + num, hash := blocks[i].NumberU64(), blocks[i].Hash() + header := chain.GetHeaderByNumber(num) + if header.Hash() != hash { + t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash()) + } + } + for i := 0; i < len(chainA); i++ { + num, hash := chainA[i].NumberU64(), chainA[i].Hash() + header := chain.GetHeaderByNumber(num) + if header == nil { + continue + } + if header.Hash() == hash { + t.Errorf("block #%d: unexpected canonical header: %v", num, hash) + } + } + for i := 0; i < len(chainB); i++ { + num, hash := chainB[i].NumberU64(), chainB[i].Hash() + header := chain.GetHeaderByNumber(num) + if header.Hash() != hash { + t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash()) + } + } +} + +// Tests the scenario that all the inserted chain segment are with the configured +// chain cutoff point. In this case the chain segment before the cutoff should +// be persisted without the receipts and bodies; chain after should be persisted +// normally. +func TestInsertChainWithCutoff(t *testing.T) { + testInsertChainWithCutoff(t, 32, 32) // cutoff = 32, ancientLimit = 32 + testInsertChainWithCutoff(t, 32, 64) // cutoff = 32, ancientLimit = 64 (entire chain in ancient) + testInsertChainWithCutoff(t, 32, 65) // cutoff = 32, ancientLimit = 65 (64 blocks in ancient, 1 block in live) +} + +func testInsertChainWithCutoff(t *testing.T, cutoff uint64, ancientLimit uint64) { + // log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true))) + + // Configure and generate a sample block chain + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Config: params.TestChainConfig, + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + } + signer = types.LatestSigner(gspec.Config) + engine = beacon.New(ethash.NewFaker()) + ) + _, blocks, receipts := GenerateChainWithGenesis(gspec, engine, int(2*cutoff), func(i int, block *BlockGen) { + block.SetCoinbase(common.Address{0x00}) + + tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, block.header.BaseFee, nil), signer, key) + if err != nil { + panic(err) + } + block.AddTx(tx) + }) + db, _ := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false) + defer db.Close() + + cutoffBlock := blocks[cutoff-1] + config := DefaultCacheConfigWithScheme(rawdb.PathScheme) + config.HistoryPruningCutoffNumber = cutoffBlock.NumberU64() + config.HistoryPruningCutoffHash = cutoffBlock.Hash() + + chain, _ := NewBlockChain(db, DefaultCacheConfigWithScheme(rawdb.PathScheme), gspec, nil, beacon.New(ethash.NewFaker()), vm.Config{}, nil) + defer chain.Stop() + + var ( + headersBefore []*types.Header + blocksAfter []*types.Block + receiptsAfter []types.Receipts + ) + for i, b := range blocks { + if b.NumberU64() < cutoffBlock.NumberU64() { + headersBefore = append(headersBefore, b.Header()) + } else { + blocksAfter = append(blocksAfter, b) + receiptsAfter = append(receiptsAfter, receipts[i]) + } + } + if n, err := chain.InsertHeadersBeforeCutoff(headersBefore); err != nil { + t.Fatalf("failed to insert headers before cutoff %d: %v", n, err) + } + if n, err := chain.InsertReceiptChain(blocksAfter, receiptsAfter, ancientLimit); err != nil { + t.Fatalf("failed to insert receipt %d: %v", n, err) + } + headSnap := chain.CurrentSnapBlock() + if headSnap.Hash() != blocks[len(blocks)-1].Hash() { + t.Errorf("head snap block #%d: header mismatch: want: %v, got: %v", headSnap.Number, blocks[len(blocks)-1].Hash(), headSnap.Hash()) + } + headHeader := chain.CurrentHeader() + if headHeader.Hash() != blocks[len(blocks)-1].Hash() { + t.Errorf("head header #%d: header mismatch: want: %v, got: %v", headHeader.Number, blocks[len(blocks)-1].Hash(), headHeader.Hash()) + } + headBlock := chain.CurrentBlock() + if headBlock.Hash() != gspec.ToBlock().Hash() { + t.Errorf("head block #%d: header mismatch: want: %v, got: %v", headBlock.Number, gspec.ToBlock().Hash(), headBlock.Hash()) + } + + // Iterate over all chain data components, and cross reference + for i := 0; i < len(blocks); i++ { + num, hash := blocks[i].NumberU64(), blocks[i].Hash() + + // Canonical headers should be visible regardless of cutoff + header := chain.GetHeaderByNumber(num) + if header.Hash() != hash { + t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash()) + } + tail, err := db.Tail() + if err != nil { + t.Fatalf("Failed to get chain tail, %v", err) + } + if tail != cutoffBlock.NumberU64() { + t.Fatalf("Unexpected chain tail, want: %d, got: %d", cutoffBlock.NumberU64(), tail) + } + // Block bodies and receipts before the cutoff should be non-existent + if num < cutoffBlock.NumberU64() { + body := chain.GetBody(hash) + if body != nil { + t.Fatalf("Unexpected block body: %d, cutoff: %d", num, cutoffBlock.NumberU64()) + } + receipts := chain.GetReceiptsByHash(hash) + if receipts != nil { + t.Fatalf("Unexpected block receipts: %d, cutoff: %d", num, cutoffBlock.NumberU64()) + } + } else { + body := chain.GetBody(hash) + if body == nil || len(body.Transactions) != 1 { + t.Fatalf("Missed block body: %d, cutoff: %d", num, cutoffBlock.NumberU64()) + } + receipts := chain.GetReceiptsByHash(hash) + if receipts == nil || len(receipts) != 1 { + t.Fatalf("Missed block receipts: %d, cutoff: %d", num, cutoffBlock.NumberU64()) + } + } + } +} diff --git a/core/error.go b/core/error.go index ce3bbb7888..de95e64636 100644 --- a/core/error.go +++ b/core/error.go @@ -28,8 +28,6 @@ var ( // ErrNoGenesis is returned when there is no Genesis Block. ErrNoGenesis = errors.New("genesis not found in chain") - - errSideChainReceipts = errors.New("side blocks can't be accepted as ancient chain data") ) // List of evm-call-message pre-checking errors. All state transition messages will diff --git a/core/headerchain.go b/core/headerchain.go index cb707a152f..f7acc49bef 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -237,7 +237,8 @@ func (hc *HeaderChain) WriteHeaders(headers []*types.Header) (int, error) { } // writeHeadersAndSetHead writes a batch of block headers and applies the last -// header as the chain head if the fork choicer says it's ok to update the chain. +// header as the chain head. +// // Note: This method is not concurrent-safe with inserting blocks simultaneously // into the chain, as side effects caused by reorganisations cannot be emulated // without the real blocks. Hence, writing headers directly should only be done @@ -272,12 +273,14 @@ func (hc *HeaderChain) writeHeadersAndSetHead(headers []*types.Header) (*headerW return result, nil } +// ValidateHeaderChain verifies that the supplied header chain is contiguous +// and conforms to consensus rules. func (hc *HeaderChain) ValidateHeaderChain(chain []*types.Header) (int, error) { // Do a sanity check that the provided chain is actually ordered and linked for i := 1; i < len(chain); i++ { if chain[i].Number.Uint64() != chain[i-1].Number.Uint64()+1 { - hash := chain[i].Hash() - parentHash := chain[i-1].Hash() + hash, parentHash := chain[i].Hash(), chain[i-1].Hash() + // Chain broke ancestry, log a message (programming error) and skip insertion log.Error("Non contiguous header insert", "number", chain[i].Number, "hash", hash, "parent", chain[i].ParentHash, "prevnumber", chain[i-1].Number, "prevhash", parentHash) @@ -302,7 +305,6 @@ func (hc *HeaderChain) ValidateHeaderChain(chain []*types.Header) (int, error) { return i, err } } - return 0, nil } diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go index 020d35619e..2f62d86e4b 100644 --- a/core/rawdb/accessors_chain.go +++ b/core/rawdb/accessors_chain.go @@ -737,6 +737,30 @@ func writeAncientBlock(op ethdb.AncientWriteOp, block *types.Block, header *type return nil } +// WriteAncientHeaderChain writes the supplied headers along with nil block +// bodies and receipts into the ancient store. It's supposed to be used for +// storing chain segment before the chain cutoff. +func WriteAncientHeaderChain(db ethdb.AncientWriter, headers []*types.Header) (int64, error) { + return db.ModifyAncients(func(op ethdb.AncientWriteOp) error { + for _, header := range headers { + num := header.Number.Uint64() + if err := op.AppendRaw(ChainFreezerHashTable, num, header.Hash().Bytes()); err != nil { + return fmt.Errorf("can't add block %d hash: %v", num, err) + } + if err := op.Append(ChainFreezerHeaderTable, num, header); err != nil { + return fmt.Errorf("can't append block header %d: %v", num, err) + } + if err := op.AppendRaw(ChainFreezerBodiesTable, num, nil); err != nil { + return fmt.Errorf("can't append block body %d: %v", num, err) + } + if err := op.AppendRaw(ChainFreezerReceiptTable, num, nil); err != nil { + return fmt.Errorf("can't append block %d receipts: %v", num, err) + } + } + return nil + }) +} + // DeleteBlock removes all block data associated with a hash. func DeleteBlock(db ethdb.KeyValueWriter, hash common.Hash, number uint64) { DeleteReceipts(db, hash, number) diff --git a/core/rawdb/accessors_chain_test.go b/core/rawdb/accessors_chain_test.go index efd16d5fa7..247e277582 100644 --- a/core/rawdb/accessors_chain_test.go +++ b/core/rawdb/accessors_chain_test.go @@ -464,6 +464,48 @@ func TestAncientStorage(t *testing.T) { } } +func TestWriteAncientHeaderChain(t *testing.T) { + db, err := NewDatabaseWithFreezer(NewMemoryDatabase(), t.TempDir(), "", false) + if err != nil { + t.Fatalf("failed to create database with ancient backend") + } + defer db.Close() + + // Create a test block + var headers []*types.Header + headers = append(headers, &types.Header{ + Number: big.NewInt(0), + Extra: []byte("test block"), + UncleHash: types.EmptyUncleHash, + TxHash: types.EmptyTxsHash, + ReceiptHash: types.EmptyReceiptsHash, + }) + headers = append(headers, &types.Header{ + Number: big.NewInt(1), + Extra: []byte("test block"), + UncleHash: types.EmptyUncleHash, + TxHash: types.EmptyTxsHash, + ReceiptHash: types.EmptyReceiptsHash, + }) + // Write and verify the header in the database + WriteAncientHeaderChain(db, headers) + + for _, header := range headers { + if blob := ReadHeaderRLP(db, header.Hash(), header.Number.Uint64()); len(blob) == 0 { + t.Fatalf("no header returned") + } + if h := ReadCanonicalHash(db, header.Number.Uint64()); h != header.Hash() { + t.Fatalf("no canonical hash returned") + } + if blob := ReadBodyRLP(db, header.Hash(), header.Number.Uint64()); len(blob) != 0 { + t.Fatalf("unexpected body returned") + } + if blob := ReadReceiptsRLP(db, header.Hash(), header.Number.Uint64()); len(blob) != 0 { + t.Fatalf("unexpected body returned") + } + } +} + func TestCanonicalHashIteration(t *testing.T) { var cases = []struct { from, to uint64 diff --git a/core/txindexer.go b/core/txindexer.go index d0fce302f3..31f069995b 100644 --- a/core/txindexer.go +++ b/core/txindexer.go @@ -58,9 +58,10 @@ type txIndexer struct { // newTxIndexer initializes the transaction indexer. func newTxIndexer(limit uint64, chain *BlockChain) *txIndexer { + cutoff, _ := chain.HistoryPruningCutoff() indexer := &txIndexer{ limit: limit, - cutoff: chain.HistoryPruningCutoff(), + cutoff: cutoff, db: chain.db, progress: make(chan chan TxIndexProgress), term: make(chan chan struct{}), diff --git a/eth/api_backend.go b/eth/api_backend.go index 944c357e78..b39dd4cbdb 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -94,7 +94,7 @@ func (b *EthAPIBackend) HeaderByNumber(ctx context.Context, number rpc.BlockNumb } var bn uint64 if number == rpc.EarliestBlockNumber { - bn = b.eth.blockchain.HistoryPruningCutoff() + bn = b.HistoryPruningCutoff() } else { bn = uint64(number) } @@ -152,10 +152,10 @@ func (b *EthAPIBackend) BlockByNumber(ctx context.Context, number rpc.BlockNumbe } bn := uint64(number) // the resolved number if number == rpc.EarliestBlockNumber { - bn = b.eth.blockchain.HistoryPruningCutoff() + bn = b.HistoryPruningCutoff() } block := b.eth.blockchain.GetBlockByNumber(bn) - if block == nil && bn < b.eth.blockchain.HistoryPruningCutoff() { + if block == nil && bn < b.HistoryPruningCutoff() { return nil, ðconfig.PrunedHistoryError{} } return block, nil @@ -167,7 +167,7 @@ func (b *EthAPIBackend) BlockByHash(ctx context.Context, hash common.Hash) (*typ return nil, nil } block := b.eth.blockchain.GetBlock(hash, *number) - if block == nil && *number < b.eth.blockchain.HistoryPruningCutoff() { + if block == nil && *number < b.HistoryPruningCutoff() { return nil, ðconfig.PrunedHistoryError{} } return block, nil @@ -180,7 +180,7 @@ func (b *EthAPIBackend) GetBody(ctx context.Context, hash common.Hash, number rp } body := b.eth.blockchain.GetBody(hash) if body == nil { - if uint64(number) < b.eth.blockchain.HistoryPruningCutoff() { + if uint64(number) < b.HistoryPruningCutoff() { return nil, ðconfig.PrunedHistoryError{} } return nil, errors.New("block body not found") @@ -202,7 +202,7 @@ func (b *EthAPIBackend) BlockByNumberOrHash(ctx context.Context, blockNrOrHash r } block := b.eth.blockchain.GetBlock(hash, header.Number.Uint64()) if block == nil { - if header.Number.Uint64() < b.eth.blockchain.HistoryPruningCutoff() { + if header.Number.Uint64() < b.HistoryPruningCutoff() { return nil, ðconfig.PrunedHistoryError{} } return nil, errors.New("header found, but block body is missing") @@ -265,7 +265,8 @@ func (b *EthAPIBackend) StateAndHeaderByNumberOrHash(ctx context.Context, blockN } func (b *EthAPIBackend) HistoryPruningCutoff() uint64 { - return b.eth.blockchain.HistoryPruningCutoff() + bn, _ := b.eth.blockchain.HistoryPruningCutoff() + return bn } func (b *EthAPIBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) { diff --git a/eth/backend.go b/eth/backend.go index ab612b1de7..79759710b6 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -154,13 +154,18 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { } // Validate history pruning configuration. - var historyPruningCutoff uint64 + var ( + cutoffNumber uint64 + cutoffHash common.Hash + ) if config.HistoryMode == ethconfig.PostMergeHistory { prunecfg, ok := ethconfig.HistoryPrunePoints[genesisHash] if !ok { return nil, fmt.Errorf("no history pruning point is defined for genesis %x", genesisHash) } - historyPruningCutoff = prunecfg.BlockNumber + cutoffNumber = prunecfg.BlockNumber + cutoffHash = prunecfg.BlockHash + log.Info("Chain cutoff configured", "number", cutoffNumber, "hash", cutoffHash) } // Set networkID to chainID by default. @@ -204,16 +209,17 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { EnablePreimageRecording: config.EnablePreimageRecording, } cacheConfig = &core.CacheConfig{ - TrieCleanLimit: config.TrieCleanCache, - TrieCleanNoPrefetch: config.NoPrefetch, - TrieDirtyLimit: config.TrieDirtyCache, - TrieDirtyDisabled: config.NoPruning, - TrieTimeLimit: config.TrieTimeout, - SnapshotLimit: config.SnapshotCache, - Preimages: config.Preimages, - StateHistory: config.StateHistory, - StateScheme: scheme, - HistoryPruningCutoff: historyPruningCutoff, + TrieCleanLimit: config.TrieCleanCache, + TrieCleanNoPrefetch: config.NoPrefetch, + TrieDirtyLimit: config.TrieDirtyCache, + TrieDirtyDisabled: config.NoPruning, + TrieTimeLimit: config.TrieTimeout, + SnapshotLimit: config.SnapshotCache, + Preimages: config.Preimages, + StateHistory: config.StateHistory, + StateScheme: scheme, + HistoryPruningCutoffNumber: cutoffNumber, + HistoryPruningCutoffHash: cutoffHash, } ) if config.VMTrace != "" { @@ -246,7 +252,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { HashScheme: scheme == rawdb.HashScheme, } chainView := eth.newChainView(eth.blockchain.CurrentBlock()) - historyCutoff := eth.blockchain.HistoryPruningCutoff() + historyCutoff, _ := eth.blockchain.HistoryPruningCutoff() var finalBlock uint64 if fb := eth.blockchain.CurrentFinalBlock(); fb != nil { finalBlock = fb.Number.Uint64() @@ -443,7 +449,7 @@ func (s *Ethereum) updateFilterMapsHeads() { if head == nil || newHead.Hash() != head.Hash() { head = newHead chainView := s.newChainView(head) - historyCutoff := s.blockchain.HistoryPruningCutoff() + historyCutoff, _ := s.blockchain.HistoryPruningCutoff() var finalBlock uint64 if fb := s.blockchain.CurrentFinalBlock(); fb != nil { finalBlock = fb.Number.Uint64() diff --git a/eth/downloader/beaconsync.go b/eth/downloader/beaconsync.go index c142ea7435..33ad0f8971 100644 --- a/eth/downloader/beaconsync.go +++ b/eth/downloader/beaconsync.go @@ -273,8 +273,7 @@ func (d *Downloader) findBeaconAncestor() (uint64, error) { // fetchHeaders feeds skeleton headers to the downloader queue for scheduling // until sync errors or is finished. func (d *Downloader) fetchHeaders(from uint64) error { - var head *types.Header - _, tail, _, err := d.skeleton.Bounds() + head, tail, _, err := d.skeleton.Bounds() if err != nil { return err } @@ -294,6 +293,27 @@ func (d *Downloader) fetchHeaders(from uint64) error { fsHeaderContCheckTimer := time.NewTimer(fsHeaderContCheck) defer fsHeaderContCheckTimer.Stop() + // Verify the header at configured chain cutoff, ensuring it's matched with + // the configured hash. Skip the check if the configured cutoff is even higher + // than the sync target, which is definitely not a common case. + if d.chainCutoffNumber != 0 && d.chainCutoffNumber >= from && d.chainCutoffNumber <= head.Number.Uint64() { + h := d.skeleton.Header(d.chainCutoffNumber) + if h == nil { + if d.chainCutoffNumber < tail.Number.Uint64() { + dist := tail.Number.Uint64() - d.chainCutoffNumber + if len(localHeaders) >= int(dist) { + h = localHeaders[dist-1] + } + } + } + if h == nil { + return fmt.Errorf("header at chain cutoff is not available, cutoff: %d", d.chainCutoffNumber) + } + if h.Hash() != d.chainCutoffHash { + return fmt.Errorf("header at chain cutoff mismatched, want: %v, got: %v", d.chainCutoffHash, h.Hash()) + } + } + for { // Some beacon headers might have appeared since the last cycle, make // sure we're always syncing to all available ones diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 3f3f9b7f0c..4d13ae304c 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -20,6 +20,7 @@ package downloader import ( "errors" "fmt" + "sort" "sync" "sync/atomic" "time" @@ -120,6 +121,12 @@ type Downloader struct { committed atomic.Bool ancientLimit uint64 // The maximum block number which can be regarded as ancient data. + // The cutoff block number and hash before which chain segments (bodies + // and receipts) are skipped during synchronization. 0 means the entire + // chain segment is aimed for synchronization. + chainCutoffNumber uint64 + chainCutoffHash common.Hash + // Channels headerProcCh chan *headerTask // Channel to feed the header processor new tasks @@ -163,9 +170,6 @@ type BlockChain interface { // CurrentHeader retrieves the head header from the local chain. CurrentHeader() *types.Header - // InsertHeaderChain inserts a batch of headers into the local chain. - InsertHeaderChain([]*types.Header) (int, error) - // SetHead rewinds the local chain to a new head. SetHead(uint64) error @@ -187,10 +191,17 @@ type BlockChain interface { // SnapSyncCommitHead directly commits the head block to a certain entity. SnapSyncCommitHead(common.Hash) error + // InsertHeadersBeforeCutoff inserts a batch of headers before the configured + // chain cutoff into the ancient store. + InsertHeadersBeforeCutoff([]*types.Header) (int, error) + // InsertChain inserts a batch of blocks into the local chain. InsertChain(types.Blocks) (int, error) - // InsertReceiptChain inserts a batch of receipts into the local chain. + // InsertReceiptChain inserts a batch of blocks along with their receipts + // into the local chain. Blocks older than the specified `ancientLimit` + // are stored directly in the ancient store, while newer blocks are stored + // in the live key-value store. InsertReceiptChain(types.Blocks, []types.Receipts, uint64) (int, error) // Snapshots returns the blockchain snapshot tree to paused it during sync. @@ -199,22 +210,29 @@ type BlockChain interface { // TrieDB retrieves the low level trie database used for interacting // with trie nodes. TrieDB() *triedb.Database + + // HistoryPruningCutoff returns the configured history pruning point. + // Block bodies along with the receipts will be skipped for synchronization. + HistoryPruningCutoff() (uint64, common.Hash) } // New creates a new downloader to fetch hashes and blocks from remote peers. func New(stateDb ethdb.Database, mux *event.TypeMux, chain BlockChain, dropPeer peerDropFn, success func()) *Downloader { + cutoffNumber, cutoffHash := chain.HistoryPruningCutoff() dl := &Downloader{ - stateDB: stateDb, - mux: mux, - queue: newQueue(blockCacheMaxItems, blockCacheInitialItems), - peers: newPeerSet(), - blockchain: chain, - dropPeer: dropPeer, - headerProcCh: make(chan *headerTask, 1), - quitCh: make(chan struct{}), - SnapSyncer: snap.NewSyncer(stateDb, chain.TrieDB().Scheme()), - stateSyncStart: make(chan *stateSync), - syncStartBlock: chain.CurrentSnapBlock().Number.Uint64(), + stateDB: stateDb, + mux: mux, + queue: newQueue(blockCacheMaxItems, blockCacheInitialItems), + peers: newPeerSet(), + blockchain: chain, + chainCutoffNumber: cutoffNumber, + chainCutoffHash: cutoffHash, + dropPeer: dropPeer, + headerProcCh: make(chan *headerTask, 1), + quitCh: make(chan struct{}), + SnapSyncer: snap.NewSyncer(stateDb, chain.TrieDB().Scheme()), + stateSyncStart: make(chan *stateSync), + syncStartBlock: chain.CurrentSnapBlock().Number.Uint64(), } // Create the post-merge skeleton syncer and start the process dl.skeleton = newSkeleton(stateDb, dl.peers, dropPeer, newBeaconBackfiller(dl, success)) @@ -503,6 +521,12 @@ func (d *Downloader) syncToHead() (err error) { } else { d.ancientLimit = 0 } + // Extend the ancient chain segment range if the ancient limit is even + // below the pre-configured chain cutoff. + if d.chainCutoffNumber != 0 && d.chainCutoffNumber > d.ancientLimit { + d.ancientLimit = d.chainCutoffNumber + log.Info("Extend the ancient range with configured cutoff", "cutoff", d.chainCutoffNumber) + } frozen, _ := d.stateDB.Ancients() // Ignore the error here since light client can also hit here. // If a part of blockchain data has already been written into active store, @@ -521,14 +545,23 @@ func (d *Downloader) syncToHead() (err error) { log.Info("Truncated excess ancient chain segment", "oldhead", frozen-1, "newhead", origin) } } + // Skip ancient chain segments if Geth is running with a configured chain cutoff. + // These segments are not guaranteed to be available in the network. + chainOffset := origin + 1 + if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 { + if chainOffset < d.chainCutoffNumber { + chainOffset = d.chainCutoffNumber + log.Info("Skip chain segment before cutoff", "origin", origin, "cutoff", d.chainCutoffNumber) + } + } // Initiate the sync using a concurrent header and content retrieval algorithm - d.queue.Prepare(origin+1, mode) + d.queue.Prepare(chainOffset, mode) // In beacon mode, headers are served by the skeleton syncer fetchers := []func() error{ - func() error { return d.fetchHeaders(origin + 1) }, // Headers are always retrieved - func() error { return d.fetchBodies(origin + 1) }, // Bodies are retrieved during normal and snap sync - func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during snap sync + func() error { return d.fetchHeaders(origin + 1) }, // Headers are always retrieved + func() error { return d.fetchBodies(chainOffset) }, // Bodies are retrieved during normal and snap sync + func() error { return d.fetchReceipts(chainOffset) }, // Receipts are retrieved during snap sync func() error { return d.processHeaders(origin + 1) }, } if mode == ethconfig.SnapSync { @@ -666,7 +699,7 @@ func (d *Downloader) processHeaders(origin uint64) error { return nil } // Otherwise split the chunk of headers into batches and process them - headers, hashes := task.headers, task.hashes + headers, hashes, scheduled := task.headers, task.hashes, false for len(headers) > 0 { // Terminate if something failed in between processing chunks @@ -683,17 +716,21 @@ func (d *Downloader) processHeaders(origin uint64) error { chunkHeaders := headers[:limit] chunkHashes := hashes[:limit] - // In case of header only syncing, validate the chunk immediately - if mode == ethconfig.SnapSync { - // Although the received headers might be all valid, a legacy - // PoW/PoA sync must not accept post-merge headers. Make sure - // that any transition is rejected at this point. - if len(chunkHeaders) > 0 { - if n, err := d.blockchain.InsertHeaderChain(chunkHeaders); err != nil { - log.Warn("Invalid header encountered", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err) - return fmt.Errorf("%w: %v", errInvalidChain, err) - } + // Split the headers around the chain cutoff + var cutoff int + if mode == ethconfig.SnapSync && d.chainCutoffNumber != 0 { + cutoff = sort.Search(len(chunkHeaders), func(i int) bool { + return chunkHeaders[i].Number.Uint64() >= d.chainCutoffNumber + }) + } + // Insert the header chain into the ancient store (with block bodies and + // receipts set to nil) if they fall before the cutoff. + if mode == ethconfig.SnapSync && cutoff != 0 { + if n, err := d.blockchain.InsertHeadersBeforeCutoff(chunkHeaders[:cutoff]); err != nil { + log.Warn("Failed to insert ancient header chain", "number", chunkHeaders[n].Number, "hash", chunkHashes[n], "parent", chunkHeaders[n].ParentHash, "err", err) + return fmt.Errorf("%w: %v", errInvalidChain, err) } + log.Debug("Inserted headers before cutoff", "number", chunkHeaders[cutoff-1].Number, "hash", chunkHashes[cutoff-1]) } // If we've reached the allowed number of pending headers, stall a bit for d.queue.PendingBodies() >= maxQueuedHeaders || d.queue.PendingReceipts() >= maxQueuedHeaders { @@ -704,12 +741,21 @@ func (d *Downloader) processHeaders(origin uint64) error { case <-timer.C: } } - // Otherwise insert the headers for content retrieval - inserts := d.queue.Schedule(chunkHeaders, chunkHashes, origin) - if len(inserts) != len(chunkHeaders) { - return fmt.Errorf("%w: stale headers", errBadPeer) + // Otherwise, schedule the headers for content retrieval (block bodies and + // potentially receipts in snap sync). + // + // Skip the bodies/receipts retrieval scheduling before the cutoff in snap + // sync if chain pruning is configured. + if mode == ethconfig.SnapSync && cutoff != 0 { + chunkHeaders = chunkHeaders[cutoff:] + chunkHashes = chunkHashes[cutoff:] + } + if len(chunkHeaders) > 0 { + scheduled = true + if d.queue.Schedule(chunkHeaders, chunkHashes, origin+uint64(cutoff)) != len(chunkHeaders) { + return fmt.Errorf("%w: stale headers", errBadPeer) + } } - headers = headers[limit:] hashes = hashes[limit:] origin += uint64(limit) @@ -721,11 +767,13 @@ func (d *Downloader) processHeaders(origin uint64) error { } d.syncStatsLock.Unlock() - // Signal the content downloaders of the availability of new tasks - for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} { - select { - case ch <- true: - default: + // Signal the downloader of the availability of new tasks + if scheduled { + for _, ch := range []chan bool{d.queue.blockWakeCh, d.queue.receiptWakeCh} { + select { + case ch <- true: + default: + } } } } @@ -1085,10 +1133,20 @@ func (d *Downloader) reportSnapSyncProgress(force bool) { header = d.blockchain.CurrentHeader() block = d.blockchain.CurrentSnapBlock() ) - syncedBlocks := block.Number.Uint64() - d.syncStartBlock - if syncedBlocks == 0 { + // Prevent reporting if nothing has been synchronized yet + if block.Number.Uint64() <= d.syncStartBlock { return } + // Prevent reporting noise if the actual chain synchronization (headers + // and bodies) hasn't started yet. Inserting the ancient header chain is + // fast enough and would introduce significant bias if included in the count. + if d.chainCutoffNumber != 0 && block.Number.Uint64() <= d.chainCutoffNumber { + return + } + fetchedBlocks := block.Number.Uint64() - d.syncStartBlock + if d.chainCutoffNumber != 0 && d.chainCutoffNumber > d.syncStartBlock { + fetchedBlocks = block.Number.Uint64() - d.chainCutoffNumber + } // Retrieve the current chain head and calculate the ETA latest, _, _, err := d.skeleton.Bounds() if err != nil { @@ -1103,7 +1161,7 @@ func (d *Downloader) reportSnapSyncProgress(force bool) { } var ( left = latest.Number.Uint64() - block.Number.Uint64() - eta = time.Since(d.syncStartTime) / time.Duration(syncedBlocks) * time.Duration(left) + eta = time.Since(d.syncStartTime) / time.Duration(fetchedBlocks) * time.Duration(left) progress = fmt.Sprintf("%.2f%%", float64(block.Number.Uint64())*100/float64(latest.Number.Uint64())) headers = fmt.Sprintf("%v@%v", log.FormatLogfmtUint64(header.Number.Uint64()), common.StorageSize(headerBytes).TerminalString()) diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go index 23c033a8ad..bfe80ddbf1 100644 --- a/eth/downloader/metrics.go +++ b/eth/downloader/metrics.go @@ -25,7 +25,6 @@ import ( var ( headerInMeter = metrics.NewRegisteredMeter("eth/downloader/headers/in", nil) headerReqTimer = metrics.NewRegisteredTimer("eth/downloader/headers/req", nil) - headerDropMeter = metrics.NewRegisteredMeter("eth/downloader/headers/drop", nil) headerTimeoutMeter = metrics.NewRegisteredMeter("eth/downloader/headers/timeout", nil) bodyInMeter = metrics.NewRegisteredMeter("eth/downloader/bodies/in", nil) diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index a1c114f057..000ad97ca9 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -73,7 +73,7 @@ type fetchResult struct { Withdrawals types.Withdrawals } -func newFetchResult(header *types.Header, fastSync bool) *fetchResult { +func newFetchResult(header *types.Header, snapSync bool) *fetchResult { item := &fetchResult{ Header: header, } @@ -82,7 +82,7 @@ func newFetchResult(header *types.Header, fastSync bool) *fetchResult { } else if header.WithdrawalsHash != nil { item.Withdrawals = make(types.Withdrawals, 0) } - if fastSync && !header.EmptyReceipts() { + if snapSync && !header.EmptyReceipts() { item.pending.Store(item.pending.Load() | (1 << receiptType)) } return item @@ -124,19 +124,8 @@ func (f *fetchResult) Done(kind uint) bool { // queue represents hashes that are either need fetching or are being fetched type queue struct { - mode SyncMode // Synchronisation mode to decide on the block parts to schedule for fetching - - // Headers are "special", they download in batches, supported by a skeleton chain - headerHead common.Hash // Hash of the last queued header to verify order - headerTaskPool map[uint64]*types.Header // Pending header retrieval tasks, mapping starting indexes to skeleton headers - headerTaskQueue *prque.Prque[int64, uint64] // Priority queue of the skeleton indexes to fetch the filling headers for - headerPeerMiss map[string]map[uint64]struct{} // Set of per-peer header batches known to be unavailable - headerPendPool map[string]*fetchRequest // Currently pending header retrieval operations - headerResults []*types.Header // Result cache accumulating the completed headers - headerHashes []common.Hash // Result cache accumulating the completed header hashes - headerProced int // Number of headers already processed from the results - headerOffset uint64 // Number of the first header in the result cache - headerContCh chan bool // Channel to notify when header download finishes + mode SyncMode // Synchronisation mode to decide on the block parts to schedule for fetching + headerHead common.Hash // Hash of the last queued header to verify order // All data retrievals below are based on an already assembles header chain blockTaskPool map[common.Hash]*types.Header // Pending block (body) retrieval tasks, mapping hashes to headers @@ -163,7 +152,6 @@ type queue struct { func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue { lock := new(sync.RWMutex) q := &queue{ - headerContCh: make(chan bool, 1), blockTaskQueue: prque.New[int64, *types.Header](nil), blockWakeCh: make(chan bool, 1), receiptTaskQueue: prque.New[int64, *types.Header](nil), @@ -182,9 +170,7 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) { q.closed = false q.mode = ethconfig.FullSync - q.headerHead = common.Hash{} - q.headerPendPool = make(map[string]*fetchRequest) q.blockTaskPool = make(map[common.Hash]*types.Header) q.blockTaskQueue.Reset() @@ -207,14 +193,6 @@ func (q *queue) Close() { q.lock.Unlock() } -// PendingHeaders retrieves the number of header requests pending for retrieval. -func (q *queue) PendingHeaders() int { - q.lock.Lock() - defer q.lock.Unlock() - - return q.headerTaskQueue.Size() -} - // PendingBodies retrieves the number of block body requests pending for retrieval. func (q *queue) PendingBodies() int { q.lock.Lock() @@ -260,54 +238,14 @@ func (q *queue) Idle() bool { return (queued + pending) == 0 } -// ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill -// up an already retrieved header skeleton. -func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) { - q.lock.Lock() - defer q.lock.Unlock() - - // No skeleton retrieval can be in progress, fail hard if so (huge implementation bug) - if q.headerResults != nil { - panic("skeleton assembly already in progress") - } - // Schedule all the header retrieval tasks for the skeleton assembly - q.headerTaskPool = make(map[uint64]*types.Header) - q.headerTaskQueue = prque.New[int64, uint64](nil) - q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains - q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch) - q.headerHashes = make([]common.Hash, len(skeleton)*MaxHeaderFetch) - q.headerProced = 0 - q.headerOffset = from - q.headerContCh = make(chan bool, 1) - - for i, header := range skeleton { - index := from + uint64(i*MaxHeaderFetch) - - q.headerTaskPool[index] = header - q.headerTaskQueue.Push(index, -int64(index)) - } -} - -// RetrieveHeaders retrieves the header chain assemble based on the scheduled -// skeleton. -func (q *queue) RetrieveHeaders() ([]*types.Header, []common.Hash, int) { - q.lock.Lock() - defer q.lock.Unlock() - - headers, hashes, proced := q.headerResults, q.headerHashes, q.headerProced - q.headerResults, q.headerHashes, q.headerProced = nil, nil, 0 - - return headers, hashes, proced -} - // Schedule adds a set of headers for the download queue for scheduling, returning // the new headers encountered. -func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uint64) []*types.Header { +func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uint64) int { q.lock.Lock() defer q.lock.Unlock() // Insert all the headers prioritised by the contained block number - inserts := make([]*types.Header, 0, len(headers)) + var inserts int for i, header := range headers { // Make sure chain order is honoured and preserved throughout hash := hashes[i] @@ -337,7 +275,7 @@ func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uin q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64())) } } - inserts = append(inserts, header) + inserts++ q.headerHead = hash from++ } @@ -390,7 +328,7 @@ func (q *queue) Results(block bool) []*fetchResult { q.resultSize = common.StorageSize(blockCacheSizeWeight)*size + (1-common.StorageSize(blockCacheSizeWeight))*q.resultSize } - // Using the newly calibrated resultsize, figure out the new throttle limit + // Using the newly calibrated result size, figure out the new throttle limit // on the result cache throttleThreshold := uint64((common.StorageSize(blockCacheMemory) + q.resultSize - 1) / q.resultSize) throttleThreshold = q.resultCache.SetThrottleThreshold(throttleThreshold) @@ -428,46 +366,6 @@ func (q *queue) stats() []interface{} { } } -// ReserveHeaders reserves a set of headers for the given peer, skipping any -// previously failed batches. -func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest { - q.lock.Lock() - defer q.lock.Unlock() - - // Short circuit if the peer's already downloading something (sanity check to - // not corrupt state) - if _, ok := q.headerPendPool[p.id]; ok { - return nil - } - // Retrieve a batch of hashes, skipping previously failed ones - send, skip := uint64(0), []uint64{} - for send == 0 && !q.headerTaskQueue.Empty() { - from, _ := q.headerTaskQueue.Pop() - if q.headerPeerMiss[p.id] != nil { - if _, ok := q.headerPeerMiss[p.id][from]; ok { - skip = append(skip, from) - continue - } - } - send = from - } - // Merge all the skipped batches back - for _, from := range skip { - q.headerTaskQueue.Push(from, -int64(from)) - } - // Assemble and return the block download request - if send == 0 { - return nil - } - request := &fetchRequest{ - Peer: p, - From: send, - Time: time.Now(), - } - q.headerPendPool[p.id] = request - return request -} - // ReserveBodies reserves a set of body fetches for the given peer, skipping any // previously failed downloads. Beside the next batch of needed fetches, it also // returns a flag whether empty blocks were queued requiring processing. @@ -594,10 +492,6 @@ func (q *queue) Revoke(peerID string) { q.lock.Lock() defer q.lock.Unlock() - if request, ok := q.headerPendPool[peerID]; ok { - q.headerTaskQueue.Push(request.From, -int64(request.From)) - delete(q.headerPendPool, peerID) - } if request, ok := q.blockPendPool[peerID]; ok { for _, header := range request.Headers { q.blockTaskQueue.Push(header, -int64(header.Number.Uint64())) @@ -612,16 +506,6 @@ func (q *queue) Revoke(peerID string) { } } -// ExpireHeaders cancels a request that timed out and moves the pending fetch -// task back into the queue for rescheduling. -func (q *queue) ExpireHeaders(peer string) int { - q.lock.Lock() - defer q.lock.Unlock() - - headerTimeoutMeter.Mark(1) - return q.expire(peer, q.headerPendPool, q.headerTaskQueue) -} - // ExpireBodies checks for in flight block body requests that exceeded a timeout // allowance, canceling them and returning the responsible peers for penalisation. func (q *queue) ExpireBodies(peer string) int { @@ -670,116 +554,6 @@ func (q *queue) expire(peer string, pendPool map[string]*fetchRequest, taskQueue return len(req.Headers) } -// DeliverHeaders injects a header retrieval response into the header results -// cache. This method either accepts all headers it received, or none of them -// if they do not map correctly to the skeleton. -// -// If the headers are accepted, the method makes an attempt to deliver the set -// of ready headers to the processor to keep the pipeline full. However, it will -// not block to prevent stalling other pending deliveries. -func (q *queue) DeliverHeaders(id string, headers []*types.Header, hashes []common.Hash, headerProcCh chan *headerTask) (int, error) { - q.lock.Lock() - defer q.lock.Unlock() - - var logger log.Logger - if len(id) < 16 { - // Tests use short IDs, don't choke on them - logger = log.New("peer", id) - } else { - logger = log.New("peer", id[:16]) - } - // Short circuit if the data was never requested - request := q.headerPendPool[id] - if request == nil { - headerDropMeter.Mark(int64(len(headers))) - return 0, errNoFetchesPending - } - delete(q.headerPendPool, id) - - headerReqTimer.UpdateSince(request.Time) - headerInMeter.Mark(int64(len(headers))) - - // Ensure headers can be mapped onto the skeleton chain - target := q.headerTaskPool[request.From].Hash() - - accepted := len(headers) == MaxHeaderFetch - if accepted { - if headers[0].Number.Uint64() != request.From { - logger.Trace("First header broke chain ordering", "number", headers[0].Number, "hash", hashes[0], "expected", request.From) - accepted = false - } else if hashes[len(headers)-1] != target { - logger.Trace("Last header broke skeleton structure ", "number", headers[len(headers)-1].Number, "hash", hashes[len(headers)-1], "expected", target) - accepted = false - } - } - if accepted { - parentHash := hashes[0] - for i, header := range headers[1:] { - hash := hashes[i+1] - if want := request.From + 1 + uint64(i); header.Number.Uint64() != want { - logger.Warn("Header broke chain ordering", "number", header.Number, "hash", hash, "expected", want) - accepted = false - break - } - if parentHash != header.ParentHash { - logger.Warn("Header broke chain ancestry", "number", header.Number, "hash", hash) - accepted = false - break - } - // Set-up parent hash for next round - parentHash = hash - } - } - // If the batch of headers wasn't accepted, mark as unavailable - if !accepted { - logger.Trace("Skeleton filling not accepted", "from", request.From) - headerDropMeter.Mark(int64(len(headers))) - - miss := q.headerPeerMiss[id] - if miss == nil { - q.headerPeerMiss[id] = make(map[uint64]struct{}) - miss = q.headerPeerMiss[id] - } - miss[request.From] = struct{}{} - - q.headerTaskQueue.Push(request.From, -int64(request.From)) - return 0, errors.New("delivery not accepted") - } - // Clean up a successful fetch and try to deliver any sub-results - copy(q.headerResults[request.From-q.headerOffset:], headers) - copy(q.headerHashes[request.From-q.headerOffset:], hashes) - - delete(q.headerTaskPool, request.From) - - ready := 0 - for q.headerProced+ready < len(q.headerResults) && q.headerResults[q.headerProced+ready] != nil { - ready += MaxHeaderFetch - } - if ready > 0 { - // Headers are ready for delivery, gather them and push forward (non blocking) - processHeaders := make([]*types.Header, ready) - copy(processHeaders, q.headerResults[q.headerProced:q.headerProced+ready]) - - processHashes := make([]common.Hash, ready) - copy(processHashes, q.headerHashes[q.headerProced:q.headerProced+ready]) - - select { - case headerProcCh <- &headerTask{ - headers: processHeaders, - hashes: processHashes, - }: - logger.Trace("Pre-scheduled new headers", "count", len(processHeaders), "from", processHeaders[0].Number) - q.headerProced += len(processHeaders) - default: - } - } - // Check for termination and return - if len(q.headerTaskPool) == 0 { - q.headerContCh <- false - } - return len(headers), nil -} - // DeliverBodies injects a block body retrieval response into the results queue. // The method returns the number of blocks bodies accepted from the delivery and // also wakes any threads waiting for data delivery. diff --git a/eth/downloader/resultstore.go b/eth/downloader/resultstore.go index e4323c04eb..36c382fefc 100644 --- a/eth/downloader/resultstore.go +++ b/eth/downloader/resultstore.go @@ -76,7 +76,7 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 { // throttled - if true, the store is at capacity, this particular header is not prio now // item - the result to store data into // err - any error that occurred -func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) { +func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, throttled bool, item *fetchResult, err error) { r.lock.Lock() defer r.lock.Unlock() @@ -86,7 +86,7 @@ func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, thro return stale, throttled, item, err } if item == nil { - item = newFetchResult(header, fastSync) + item = newFetchResult(header, snapSync) r.items[index] = item } return stale, throttled, item, err diff --git a/ethdb/pebble/pebble.go b/ethdb/pebble/pebble.go index b87ecb2595..969e67af5a 100644 --- a/ethdb/pebble/pebble.go +++ b/ethdb/pebble/pebble.go @@ -144,7 +144,7 @@ func (l panicLogger) Fatalf(format string, args ...interface{}) { // New returns a wrapped pebble DB object. The namespace is the prefix that the // metrics reporting should use for surfacing internal stats. -func New(file string, cache int, handles int, namespace string, readonly bool) (*Database, error) { +func New(file string, cache int, handles int, namespace string, readonly bool, ephemeral bool) (*Database, error) { // Ensure we have some minimal caching and file guarantees if cache < minCache { cache = minCache @@ -185,7 +185,7 @@ func New(file string, cache int, handles int, namespace string, readonly bool) ( fn: file, log: logger, quitChan: make(chan chan error), - writeOptions: &pebble.WriteOptions{Sync: false}, + writeOptions: &pebble.WriteOptions{Sync: !ephemeral}, } opt := &pebble.Options{ // Pebble has a single combined cache area and the write diff --git a/internal/era/e2store/e2store.go b/internal/era/e2store/e2store.go index 9832b72d48..b0d43bf55a 100644 --- a/internal/era/e2store/e2store.go +++ b/internal/era/e2store/e2store.go @@ -219,3 +219,15 @@ func (r *Reader) FindAll(want uint16) ([]*Entry, error) { off += int64(headerSize + length) } } + +// SkipN skips `n` entries starting from `offset` and returns the new offset. +func (r *Reader) SkipN(offset int64, n uint64) (int64, error) { + for i := uint64(0); i < n; i++ { + length, err := r.LengthAt(offset) + if err != nil { + return 0, err + } + offset += length + } + return offset, nil +} diff --git a/internal/era/era.go b/internal/era/era.go index daf337963d..5129186fe7 100644 --- a/internal/era/era.go +++ b/internal/era/era.go @@ -70,7 +70,7 @@ func ReadDir(dir, network string) ([]string, error) { } parts := strings.Split(entry.Name(), "-") if len(parts) != 3 || parts[0] != network { - // invalid era1 filename, skip + // Invalid era1 filename, skip. continue } if epoch, err := strconv.ParseUint(parts[1], 10, 64); err != nil { @@ -126,6 +126,29 @@ func (e *Era) Close() error { return e.f.Close() } +// GetHeaderByNumber returns the header for the given block number. +func (e *Era) GetHeaderByNumber(num uint64) (*types.Header, error) { + if e.m.start > num || e.m.start+e.m.count <= num { + return nil, errors.New("out-of-bounds") + } + off, err := e.readOffset(num) + if err != nil { + return nil, err + } + + // Read and decompress header. + r, _, err := newSnappyReader(e.s, TypeCompressedHeader, off) + if err != nil { + return nil, err + } + var header types.Header + if err := rlp.Decode(r, &header); err != nil { + return nil, err + } + return &header, nil +} + +// GetBlockByNumber returns the block for the given block number. func (e *Era) GetBlockByNumber(num uint64) (*types.Block, error) { if e.m.start > num || e.m.start+e.m.count <= num { return nil, errors.New("out-of-bounds") @@ -154,6 +177,34 @@ func (e *Era) GetBlockByNumber(num uint64) (*types.Block, error) { return types.NewBlockWithHeader(&header).WithBody(body), nil } +// GetReceiptsByNumber returns the receipts for the given block number. +func (e *Era) GetReceiptsByNumber(num uint64) (types.Receipts, error) { + if e.m.start > num || e.m.start+e.m.count <= num { + return nil, errors.New("out-of-bounds") + } + off, err := e.readOffset(num) + if err != nil { + return nil, err + } + + // Skip over header and body. + off, err = e.s.SkipN(off, 2) + if err != nil { + return nil, err + } + + // Read and decompress receipts. + r, _, err := newSnappyReader(e.s, TypeCompressedReceipts, off) + if err != nil { + return nil, err + } + var receipts types.Receipts + if err := rlp.Decode(r, &receipts); err != nil { + return nil, err + } + return receipts, nil +} + // Accumulator reads the accumulator entry in the Era1 file. func (e *Era) Accumulator() (common.Hash, error) { entry, err := e.s.Find(TypeAccumulator) @@ -187,13 +238,10 @@ func (e *Era) InitialTD() (*big.Int, error) { } off += n - // Skip over next two records. - for i := 0; i < 2; i++ { - length, err := e.s.LengthAt(off) - if err != nil { - return nil, err - } - off += length + // Skip over header and body. + off, err = e.s.SkipN(off, 2) + if err != nil { + return nil, err } // Read total difficulty after first block. diff --git a/internal/era/era_test.go b/internal/era/era_test.go index 72c3b385dd..46fc2e91f3 100644 --- a/internal/era/era_test.go +++ b/internal/era/era_test.go @@ -18,12 +18,15 @@ package era import ( "bytes" + "fmt" "io" "math/big" "os" "testing" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/rlp" ) type testchain struct { @@ -48,9 +51,9 @@ func TestEra1Builder(t *testing.T) { chain = testchain{} ) for i := 0; i < 128; i++ { - chain.headers = append(chain.headers, []byte{byte('h'), byte(i)}) - chain.bodies = append(chain.bodies, []byte{byte('b'), byte(i)}) - chain.receipts = append(chain.receipts, []byte{byte('r'), byte(i)}) + chain.headers = append(chain.headers, mustEncode(&types.Header{Number: big.NewInt(int64(i))})) + chain.bodies = append(chain.bodies, mustEncode(&types.Body{Transactions: []*types.Transaction{types.NewTransaction(0, common.Address{byte(i)}, nil, 0, nil, nil)}})) + chain.receipts = append(chain.receipts, mustEncode(&types.Receipts{{CumulativeGasUsed: uint64(i)}})) chain.tds = append(chain.tds, big.NewInt(int64(i))) } @@ -91,13 +94,25 @@ func TestEra1Builder(t *testing.T) { t.Fatalf("unexpected error %v", it.Error()) } // Check headers. - header, err := io.ReadAll(it.Header) + rawHeader, err := io.ReadAll(it.Header) + if err != nil { + t.Fatalf("error reading header from iterator: %v", err) + } + if !bytes.Equal(rawHeader, chain.headers[i]) { + t.Fatalf("mismatched header: want %s, got %s", chain.headers[i], rawHeader) + } + header, err := e.GetHeaderByNumber(i) if err != nil { t.Fatalf("error reading header: %v", err) } - if !bytes.Equal(header, chain.headers[i]) { - t.Fatalf("mismatched header: want %s, got %s", chain.headers[i], header) + encHeader, err := rlp.EncodeToBytes(header) + if err != nil { + t.Fatalf("error encoding header: %v", err) } + if !bytes.Equal(encHeader, chain.headers[i]) { + t.Fatalf("mismatched header: want %s, got %s", chain.headers[i], encHeader) + } + // Check bodies. body, err := io.ReadAll(it.Body) if err != nil { @@ -106,13 +121,25 @@ func TestEra1Builder(t *testing.T) { if !bytes.Equal(body, chain.bodies[i]) { t.Fatalf("mismatched body: want %s, got %s", chain.bodies[i], body) } + // Check receipts. - receipts, err := io.ReadAll(it.Receipts) + rawReceipts, err := io.ReadAll(it.Receipts) + if err != nil { + t.Fatalf("error reading receipts from iterator: %v", err) + } + if !bytes.Equal(rawReceipts, chain.receipts[i]) { + t.Fatalf("mismatched receipts: want %s, got %s", chain.receipts[i], rawReceipts) + } + receipts, err := e.GetReceiptsByNumber(i) if err != nil { t.Fatalf("error reading receipts: %v", err) } - if !bytes.Equal(receipts, chain.receipts[i]) { - t.Fatalf("mismatched receipts: want %s, got %s", chain.receipts[i], receipts) + encReceipts, err := rlp.EncodeToBytes(receipts) + if err != nil { + t.Fatalf("error encoding receipts: %v", err) + } + if !bytes.Equal(encReceipts, chain.receipts[i]) { + t.Fatalf("mismatched receipts: want %s, got %s", chain.receipts[i], encReceipts) } // Check total difficulty. @@ -144,3 +171,11 @@ func TestEraFilename(t *testing.T) { } } } + +func mustEncode(obj any) []byte { + b, err := rlp.EncodeToBytes(obj) + if err != nil { + panic(fmt.Sprintf("failed in encode obj: %v", err)) + } + return b +} diff --git a/internal/ethapi/api_test.go b/internal/ethapi/api_test.go index 37210aa78b..3fbf32e22e 100644 --- a/internal/ethapi/api_test.go +++ b/internal/ethapi/api_test.go @@ -623,7 +623,10 @@ func (b testBackend) NewMatcherBackend() filtermaps.MatcherBackend { panic("implement me") } -func (b testBackend) HistoryPruningCutoff() uint64 { return b.chain.HistoryPruningCutoff() } +func (b testBackend) HistoryPruningCutoff() uint64 { + bn, _ := b.chain.HistoryPruningCutoff() + return bn +} func TestEstimateGas(t *testing.T) { t.Parallel() diff --git a/node/database.go b/node/database.go index e3ccb91066..b7d0d856cb 100644 --- a/node/database.go +++ b/node/database.go @@ -36,6 +36,11 @@ type openOptions struct { Cache int // the capacity(in megabytes) of the data caching Handles int // number of files to be open simultaneously ReadOnly bool + + // Ephemeral means that filesystem sync operations should be avoided: + // data integrity in the face of a crash is not important. This option + // should typically be used in tests. + Ephemeral bool } // openDatabase opens both a disk-based key-value database such as leveldb or pebble, but also @@ -78,7 +83,7 @@ func openKeyValueDatabase(o openOptions) (ethdb.Database, error) { } if o.Type == rawdb.DBPebble || existingDb == rawdb.DBPebble { log.Info("Using pebble as the backing database") - return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly) + return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly, o.Ephemeral) } if o.Type == rawdb.DBLeveldb || existingDb == rawdb.DBLeveldb { log.Info("Using leveldb as the backing database") @@ -86,7 +91,7 @@ func openKeyValueDatabase(o openOptions) (ethdb.Database, error) { } // No pre-existing database, no user-requested one either. Default to Pebble. log.Info("Defaulting to pebble as the backing database") - return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly) + return newPebbleDBDatabase(o.Directory, o.Cache, o.Handles, o.Namespace, o.ReadOnly, o.Ephemeral) } // newLevelDBDatabase creates a persistent key-value database without a freezer @@ -102,8 +107,8 @@ func newLevelDBDatabase(file string, cache int, handles int, namespace string, r // newPebbleDBDatabase creates a persistent key-value database without a freezer // moving immutable chain segments into cold storage. -func newPebbleDBDatabase(file string, cache int, handles int, namespace string, readonly bool) (ethdb.Database, error) { - db, err := pebble.New(file, cache, handles, namespace, readonly) +func newPebbleDBDatabase(file string, cache int, handles int, namespace string, readonly bool, ephemeral bool) (ethdb.Database, error) { + db, err := pebble.New(file, cache, handles, namespace, readonly, ephemeral) if err != nil { return nil, err } diff --git a/p2p/nat/natpmp.go b/p2p/nat/natpmp.go index 4a9644ac1a..ee07eb4ff6 100644 --- a/p2p/nat/natpmp.go +++ b/p2p/nat/natpmp.go @@ -49,6 +49,9 @@ func (n *pmp) AddMapping(protocol string, extport, intport int, name string, lif if lifetime <= 0 { return 0, errors.New("lifetime must not be <= 0") } + if extport == 0 { + extport = intport + } // Note order of port arguments is switched between our // AddMapping and the client's AddPortMapping. res, err := n.c.AddPortMapping(strings.ToLower(protocol), intport, extport, int(lifetime/time.Second)) diff --git a/p2p/nat/natupnp.go b/p2p/nat/natupnp.go index 1014dc69d2..ed00b8eeb6 100644 --- a/p2p/nat/natupnp.go +++ b/p2p/nat/natupnp.go @@ -86,15 +86,15 @@ func (n *upnp) AddMapping(protocol string, extport, intport int, desc string, li } protocol = strings.ToUpper(protocol) lifetimeS := uint32(lifetime / time.Second) - n.DeleteMapping(protocol, extport, intport) - err = n.withRateLimit(func() error { - return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) - }) - if err == nil { - return uint16(extport), nil + if extport == 0 { + extport = intport + } else { + // Only delete port mapping if the external port was already used by geth. + n.DeleteMapping(protocol, extport, intport) } - // Try addAnyPortMapping if mapping specified port didn't work. + + // Try to add port mapping, preferring the specified external port. err = n.withRateLimit(func() error { p, err := n.addAnyPortMapping(protocol, extport, intport, ip, desc, lifetimeS) if err == nil { @@ -105,18 +105,28 @@ func (n *upnp) AddMapping(protocol string, extport, intport int, desc string, li return uint16(extport), err } +// addAnyPortMapping tries to add a port mapping with the specified external port. +// If the external port is already in use, it will try to assign another port. func (n *upnp) addAnyPortMapping(protocol string, extport, intport int, ip net.IP, desc string, lifetimeS uint32) (uint16, error) { if client, ok := n.client.(*internetgateway2.WANIPConnection2); ok { return client.AddAnyPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) } - // It will retry with a random port number if the client does - // not support AddAnyPortMapping. - extport = n.randomPort() + // For IGDv1 and v1 services we should first try to add with extport. err := n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) - if err != nil { - return 0, err + if err == nil { + return uint16(extport), nil } - return uint16(extport), nil + + // If above fails, we retry with a random port. + // We retry several times because of possible port conflicts. + for i := 0; i < 3; i++ { + extport = n.randomPort() + err := n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + if err == nil { + return uint16(extport), nil + } + } + return 0, err } func (n *upnp) randomPort() int { diff --git a/p2p/server_nat.go b/p2p/server_nat.go index 933993bc1f..5830f950e1 100644 --- a/p2p/server_nat.go +++ b/p2p/server_nat.go @@ -150,14 +150,9 @@ func (srv *Server) portMappingLoop() { continue } - external := m.port - if m.extPort != 0 { - external = m.extPort - } - log := newLogger(m.protocol, external, m.port) - + log := newLogger(m.protocol, m.extPort, m.port) log.Trace("Attempting port mapping") - p, err := srv.NAT.AddMapping(m.protocol, external, m.port, m.name, portMapDuration) + p, err := srv.NAT.AddMapping(m.protocol, m.extPort, m.port, m.name, portMapDuration) if err != nil { log.Debug("Couldn't add port mapping", "err", err) m.extPort = 0 @@ -167,8 +162,8 @@ func (srv *Server) portMappingLoop() { // It was mapped! m.extPort = int(p) m.nextTime = srv.clock.Now().Add(portMapRefreshInterval) - if external != m.extPort { - log = newLogger(m.protocol, m.extPort, m.port) + log = newLogger(m.protocol, m.extPort, m.port) + if m.port != m.extPort { log.Info("NAT mapped alternative port") } else { log.Info("NAT mapped port")