// Copyright 2015 The go-ethereum Authors // This file is part of the go-ethereum library. // // The go-ethereum library is free software: you can redistribute it and/or modify // it under the terms of the GNU Lesser General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // The go-ethereum library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . package trie import ( "errors" "fmt" "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" ) // ErrNotRequested is returned by the trie sync when it's requested to process a // node it did not request. var ErrNotRequested = errors.New("not requested") // ErrAlreadyProcessed is returned by the trie sync when it's requested to process a // node it already processed previously. var ErrAlreadyProcessed = errors.New("already processed") // maxFetchesPerDepth is the maximum number of pending trie nodes per depth. The // role of this value is to limit the number of trie nodes that get expanded in // memory if the node was configured with a significant number of peers. const maxFetchesPerDepth = 16384 var ( // deletionGauge is the metric to track how many trie node deletions // are performed in total during the sync process. deletionGauge = metrics.NewRegisteredGauge("trie/sync/delete", nil) // lookupGauge is the metric to track how many trie node lookups are // performed to determine if node needs to be deleted. lookupGauge = metrics.NewRegisteredGauge("trie/sync/lookup", nil) // accountNodeSyncedGauge is the metric to track how many account trie // node are written during the sync. accountNodeSyncedGauge = metrics.NewRegisteredGauge("trie/sync/nodes/account", nil) // storageNodeSyncedGauge is the metric to track how many account trie // node are written during the sync. storageNodeSyncedGauge = metrics.NewRegisteredGauge("trie/sync/nodes/storage", nil) // codeSyncedGauge is the metric to track how many contract codes are // written during the sync. codeSyncedGauge = metrics.NewRegisteredGauge("trie/sync/codes", nil) ) // SyncPath is a path tuple identifying a particular trie node either in a single // trie (account) or a layered trie (account -> storage). // // Content wise the tuple either has 1 element if it addresses a node in a single // trie or 2 elements if it addresses a node in a stacked trie. // // To support aiming arbitrary trie nodes, the path needs to support odd nibble // lengths. To avoid transferring expanded hex form over the network, the last // part of the tuple (which needs to index into the middle of a trie) is compact // encoded. In case of a 2-tuple, the first item is always 32 bytes so that is // simple binary encoded. // // Examples: // - Path 0x9 -> {0x19} // - Path 0x99 -> {0x0099} // - Path 0x01234567890123456789012345678901012345678901234567890123456789019 -> {0x0123456789012345678901234567890101234567890123456789012345678901, 0x19} // - Path 0x012345678901234567890123456789010123456789012345678901234567890199 -> {0x0123456789012345678901234567890101234567890123456789012345678901, 0x0099} type SyncPath [][]byte // NewSyncPath converts an expanded trie path from nibble form into a compact // version that can be sent over the network. func NewSyncPath(path []byte) SyncPath { // If the hash is from the account trie, append a single item, if it // is from a storage trie, append a tuple. Note, the length 64 is // clashing between account leaf and storage root. It's fine though // because having a trie node at 64 depth means a hash collision was // found and we're long dead. if len(path) < 64 { return SyncPath{hexToCompact(path)} } return SyncPath{hexToKeybytes(path[:64]), hexToCompact(path[64:])} } // LeafCallback is a callback type invoked when a trie operation reaches a leaf // node. // // The keys is a path tuple identifying a particular trie node either in a single // trie (account) or a layered trie (account -> storage). Each key in the tuple // is in the raw format(32 bytes). // // The path is a composite hexary path identifying the trie node. All the key // bytes are converted to the hexary nibbles and composited with the parent path // if the trie node is in a layered trie. // // It's used by state sync and commit to allow handling external references // between account and storage tries. And also it's used in the state healing // for extracting the raw states(leaf nodes) with corresponding paths. type LeafCallback func(keys [][]byte, path []byte, leaf []byte, parent common.Hash, parentPath []byte) error // nodeRequest represents a scheduled or already in-flight trie node retrieval request. type nodeRequest struct { hash common.Hash // Hash of the trie node to retrieve path []byte // Merkle path leading to this node for prioritization data []byte // Data content of the node, cached until all subtrees complete parent *nodeRequest // Parent state node referencing this entry deps int // Number of dependencies before allowed to commit this node callback LeafCallback // Callback to invoke if a leaf node it reached on this branch } // codeRequest represents a scheduled or already in-flight bytecode retrieval request. type codeRequest struct { hash common.Hash // Hash of the contract bytecode to retrieve path []byte // Merkle path leading to this node for prioritization data []byte // Data content of the node, cached until all subtrees complete parents []*nodeRequest // Parent state nodes referencing this entry (notify all upon completion) } // NodeSyncResult is a response with requested trie node along with its node path. type NodeSyncResult struct { Path string // Path of the originally unknown trie node Data []byte // Data content of the retrieved trie node } // CodeSyncResult is a response with requested bytecode along with its hash. type CodeSyncResult struct { Hash common.Hash // Hash the originally unknown bytecode Data []byte // Data content of the retrieved bytecode } // nodeOp represents an operation upon the trie node. It can either represent a // deletion to the specific node or a node write for persisting retrieved node. type nodeOp struct { owner common.Hash // identifier of the trie (empty for account trie) path []byte // path from the root to the specified node. blob []byte // the content of the node (nil for deletion) hash common.Hash // hash of the node content (empty for node deletion) } // isDelete indicates if the operation is a database deletion. func (op *nodeOp) isDelete() bool { return len(op.blob) == 0 } // syncMemBatch is an in-memory buffer of successfully downloaded but not yet // persisted data items. type syncMemBatch struct { scheme string // State scheme identifier codes map[common.Hash][]byte // In-memory batch of recently completed codes nodes []nodeOp // In-memory batch of recently completed/deleted nodes size uint64 // Estimated batch-size of in-memory data. } // newSyncMemBatch allocates a new memory-buffer for not-yet persisted trie nodes. func newSyncMemBatch(scheme string) *syncMemBatch { return &syncMemBatch{ scheme: scheme, codes: make(map[common.Hash][]byte), } } // hasCode reports the contract code with specific hash is already cached. func (batch *syncMemBatch) hasCode(hash common.Hash) bool { _, ok := batch.codes[hash] return ok } // addCode caches a contract code database write operation. func (batch *syncMemBatch) addCode(hash common.Hash, code []byte) { batch.codes[hash] = code batch.size += common.HashLength + uint64(len(code)) } // addNode caches a node database write operation. func (batch *syncMemBatch) addNode(owner common.Hash, path []byte, blob []byte, hash common.Hash) { if batch.scheme == rawdb.PathScheme { if owner == (common.Hash{}) { batch.size += uint64(len(path) + len(blob)) } else { batch.size += common.HashLength + uint64(len(path)+len(blob)) } } else { batch.size += common.HashLength + uint64(len(blob)) } batch.nodes = append(batch.nodes, nodeOp{ owner: owner, path: path, blob: blob, hash: hash, }) } // delNode caches a node database delete operation. func (batch *syncMemBatch) delNode(owner common.Hash, path []byte) { if batch.scheme != rawdb.PathScheme { log.Error("Unexpected node deletion", "owner", owner, "path", path, "scheme", batch.scheme) return // deletion is not supported in hash mode. } if owner == (common.Hash{}) { batch.size += uint64(len(path)) } else { batch.size += common.HashLength + uint64(len(path)) } batch.nodes = append(batch.nodes, nodeOp{ owner: owner, path: path, }) } // Sync is the main state trie synchronisation scheduler, which provides yet // unknown trie hashes to retrieve, accepts node data associated with said hashes // and reconstructs the trie step by step until all is done. type Sync struct { scheme string // Node scheme descriptor used in database. database ethdb.KeyValueReader // Persistent database to check for existing entries membatch *syncMemBatch // Memory buffer to avoid frequent database writes nodeReqs map[string]*nodeRequest // Pending requests pertaining to a trie node path codeReqs map[common.Hash]*codeRequest // Pending requests pertaining to a code hash queue *prque.Prque[int64, any] // Priority queue with the pending requests fetches map[int]int // Number of active fetches per trie node depth } // NewSync creates a new trie data download scheduler. func NewSync(root common.Hash, database ethdb.KeyValueReader, callback LeafCallback, scheme string) *Sync { ts := &Sync{ scheme: scheme, database: database, membatch: newSyncMemBatch(scheme), nodeReqs: make(map[string]*nodeRequest), codeReqs: make(map[common.Hash]*codeRequest), queue: prque.New[int64, any](nil), // Ugh, can contain both string and hash, whyyy fetches: make(map[int]int), } ts.AddSubTrie(root, nil, common.Hash{}, nil, callback) return ts } // AddSubTrie registers a new trie to the sync code, rooted at the designated // parent for completion tracking. The given path is a unique node path in // hex format and contain all the parent path if it's layered trie node. func (s *Sync) AddSubTrie(root common.Hash, path []byte, parent common.Hash, parentPath []byte, callback LeafCallback) { if root == types.EmptyRootHash { return } owner, inner := ResolvePath(path) exist, inconsistent := s.hasNode(owner, inner, root) if exist { // The entire subtrie is already present in the database. return } else if inconsistent { // There is a pre-existing node with the wrong hash in DB, remove it. s.membatch.delNode(owner, inner) } // Assemble the new sub-trie sync request req := &nodeRequest{ hash: root, path: path, callback: callback, } // If this sub-trie has a designated parent, link them together if parent != (common.Hash{}) { ancestor := s.nodeReqs[string(parentPath)] if ancestor == nil { panic(fmt.Sprintf("sub-trie ancestor not found: %x", parent)) } ancestor.deps++ req.parent = ancestor } s.scheduleNodeRequest(req) } // AddCodeEntry schedules the direct retrieval of a contract code that should not // be interpreted as a trie node, but rather accepted and stored into the database // as is. func (s *Sync) AddCodeEntry(hash common.Hash, path []byte, parent common.Hash, parentPath []byte) { // Short circuit if the entry is empty or already known if hash == types.EmptyCodeHash { return } if s.membatch.hasCode(hash) { return } // If database says duplicate, the blob is present for sure. // Note we only check the existence with new code scheme, snap // sync is expected to run with a fresh new node. Even there // exists the code with legacy format, fetch and store with // new scheme anyway. if rawdb.HasCodeWithPrefix(s.database, hash) { return } // Assemble the new sub-trie sync request req := &codeRequest{ path: path, hash: hash, } // If this sub-trie has a designated parent, link them together if parent != (common.Hash{}) { ancestor := s.nodeReqs[string(parentPath)] // the parent of codereq can ONLY be nodereq if ancestor == nil { panic(fmt.Sprintf("raw-entry ancestor not found: %x", parent)) } ancestor.deps++ req.parents = append(req.parents, ancestor) } s.scheduleCodeRequest(req) } // Missing retrieves the known missing nodes from the trie for retrieval. To aid // both eth/6x style fast sync and snap/1x style state sync, the paths of trie // nodes are returned too, as well as separate hash list for codes. func (s *Sync) Missing(max int) ([]string, []common.Hash, []common.Hash) { var ( nodePaths []string nodeHashes []common.Hash codeHashes []common.Hash ) for !s.queue.Empty() && (max == 0 || len(nodeHashes)+len(codeHashes) < max) { // Retrieve the next item in line item, prio := s.queue.Peek() // If we have too many already-pending tasks for this depth, throttle depth := int(prio >> 56) if s.fetches[depth] > maxFetchesPerDepth { break } // Item is allowed to be scheduled, add it to the task list s.queue.Pop() s.fetches[depth]++ switch item := item.(type) { case common.Hash: codeHashes = append(codeHashes, item) case string: req, ok := s.nodeReqs[item] if !ok { log.Error("Missing node request", "path", item) continue // System very wrong, shouldn't happen } nodePaths = append(nodePaths, item) nodeHashes = append(nodeHashes, req.hash) } } return nodePaths, nodeHashes, codeHashes } // ProcessCode injects the received data for requested item. Note it can // happen that the single response commits two pending requests(e.g. // there are two requests one for code and one for node but the hash // is same). In this case the second response for the same hash will // be treated as "non-requested" item or "already-processed" item but // there is no downside. func (s *Sync) ProcessCode(result CodeSyncResult) error { // If the code was not requested or it's already processed, bail out req := s.codeReqs[result.Hash] if req == nil { return ErrNotRequested } if req.data != nil { return ErrAlreadyProcessed } req.data = result.Data return s.commitCodeRequest(req) } // ProcessNode injects the received data for requested item. Note it can // happen that the single response commits two pending requests(e.g. // there are two requests one for code and one for node but the hash // is same). In this case the second response for the same hash will // be treated as "non-requested" item or "already-processed" item but // there is no downside. func (s *Sync) ProcessNode(result NodeSyncResult) error { // If the trie node was not requested or it's already processed, bail out req := s.nodeReqs[result.Path] if req == nil { return ErrNotRequested } if req.data != nil { return ErrAlreadyProcessed } // Decode the node data content and update the request node, err := decodeNode(req.hash.Bytes(), result.Data) if err != nil { return err } req.data = result.Data // Create and schedule a request for all the children nodes requests, err := s.children(req, node) if err != nil { return err } if len(requests) == 0 && req.deps == 0 { s.commitNodeRequest(req) } else { req.deps += len(requests) for _, child := range requests { s.scheduleNodeRequest(child) } } return nil } // Commit flushes the data stored in the internal membatch out to persistent // storage, returning any occurred error. The whole data set will be flushed // in an atomic database batch. func (s *Sync) Commit(dbw ethdb.Batch) error { // Flush the pending node writes into database batch. var ( account int storage int ) for _, op := range s.membatch.nodes { if op.isDelete() { // node deletion is only supported in path mode. if op.owner == (common.Hash{}) { rawdb.DeleteAccountTrieNode(dbw, op.path) } else { rawdb.DeleteStorageTrieNode(dbw, op.owner, op.path) } deletionGauge.Inc(1) } else { if op.owner == (common.Hash{}) { account += 1 } else { storage += 1 } rawdb.WriteTrieNode(dbw, op.owner, op.path, op.hash, op.blob, s.scheme) } } accountNodeSyncedGauge.Inc(int64(account)) storageNodeSyncedGauge.Inc(int64(storage)) // Flush the pending code writes into database batch. for hash, value := range s.membatch.codes { rawdb.WriteCode(dbw, hash, value) } codeSyncedGauge.Inc(int64(len(s.membatch.codes))) s.membatch = newSyncMemBatch(s.scheme) // reset the batch return nil } // MemSize returns an estimated size (in bytes) of the data held in the membatch. func (s *Sync) MemSize() uint64 { return s.membatch.size } // Pending returns the number of state entries currently pending for download. func (s *Sync) Pending() int { return len(s.nodeReqs) + len(s.codeReqs) } // scheduleNodeRequest inserts a new state retrieval request into the fetch queue. If there // is already a pending request for this node, the new request will be discarded // and only a parent reference added to the old one. func (s *Sync) scheduleNodeRequest(req *nodeRequest) { s.nodeReqs[string(req.path)] = req // Schedule the request for future retrieval. This queue is shared // by both node requests and code requests. prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents for i := 0; i < 14 && i < len(req.path); i++ { prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order } s.queue.Push(string(req.path), prio) } // scheduleCodeRequest inserts a new state retrieval request into the fetch queue. If there // is already a pending request for this node, the new request will be discarded // and only a parent reference added to the old one. func (s *Sync) scheduleCodeRequest(req *codeRequest) { // If we're already requesting this node, add a new reference and stop if old, ok := s.codeReqs[req.hash]; ok { old.parents = append(old.parents, req.parents...) return } s.codeReqs[req.hash] = req // Schedule the request for future retrieval. This queue is shared // by both node requests and code requests. prio := int64(len(req.path)) << 56 // depth >= 128 will never happen, storage leaves will be included in their parents for i := 0; i < 14 && i < len(req.path); i++ { prio |= int64(15-req.path[i]) << (52 - i*4) // 15-nibble => lexicographic order } s.queue.Push(req.hash, prio) } // children retrieves all the missing children of a state trie entry for future // retrieval scheduling. func (s *Sync) children(req *nodeRequest, object node) ([]*nodeRequest, error) { // Gather all the children of the node, irrelevant whether known or not type childNode struct { path []byte node node } var children []childNode switch node := (object).(type) { case *shortNode: key := node.Key if hasTerm(key) { key = key[:len(key)-1] } children = []childNode{{ node: node.Val, path: append(append([]byte(nil), req.path...), key...), }} // Mark all internal nodes between shortNode and its **in disk** // child as invalid. This is essential in the case of path mode // scheme; otherwise, state healing might overwrite existing child // nodes silently while leaving a dangling parent node within the // range of this internal path on disk and the persistent state // ends up with a very weird situation that nodes on the same path // are not inconsistent while they all present in disk. This property // would break the guarantee for state healing. // // While it's possible for this shortNode to overwrite a previously // existing full node, the other branches of the fullNode can be // retained as they are not accessible with the new shortNode, and // also the whole sub-trie is still untouched and complete. // // This step is only necessary for path mode, as there is no deletion // in hash mode at all. if _, ok := node.Val.(hashNode); ok && s.scheme == rawdb.PathScheme { owner, inner := ResolvePath(req.path) for i := 1; i < len(key); i++ { // While checking for a non-existent item in Pebble can be less efficient // without a bloom filter, the relatively low frequency of lookups makes // the performance impact negligible. var exists bool if owner == (common.Hash{}) { exists = rawdb.HasAccountTrieNode(s.database, append(inner, key[:i]...)) } else { exists = rawdb.HasStorageTrieNode(s.database, owner, append(inner, key[:i]...)) } if exists { s.membatch.delNode(owner, append(inner, key[:i]...)) log.Debug("Detected dangling node", "owner", owner, "path", append(inner, key[:i]...)) } } lookupGauge.Inc(int64(len(key) - 1)) } case *fullNode: for i := 0; i < 17; i++ { if node.Children[i] != nil { children = append(children, childNode{ node: node.Children[i], path: append(append([]byte(nil), req.path...), byte(i)), }) } } default: panic(fmt.Sprintf("unknown node: %+v", node)) } // Iterate over the children, and request all unknown ones var ( missing = make(chan *nodeRequest, len(children)) pending sync.WaitGroup batchMu sync.Mutex ) for _, child := range children { // Notify any external watcher of a new key/value node if req.callback != nil { if node, ok := (child.node).(valueNode); ok { var paths [][]byte if len(child.path) == 2*common.HashLength { paths = append(paths, hexToKeybytes(child.path)) } else if len(child.path) == 4*common.HashLength { paths = append(paths, hexToKeybytes(child.path[:2*common.HashLength])) paths = append(paths, hexToKeybytes(child.path[2*common.HashLength:])) } if err := req.callback(paths, child.path, node, req.hash, req.path); err != nil { return nil, err } } } // If the child references another node, resolve or schedule. // We check all children concurrently. if node, ok := (child.node).(hashNode); ok { path := child.path hash := common.BytesToHash(node) pending.Add(1) go func() { defer pending.Done() owner, inner := ResolvePath(path) exist, inconsistent := s.hasNode(owner, inner, hash) if exist { return } else if inconsistent { // There is a pre-existing node with the wrong hash in DB, remove it. batchMu.Lock() s.membatch.delNode(owner, inner) batchMu.Unlock() } // Locally unknown node, schedule for retrieval missing <- &nodeRequest{ path: path, hash: hash, parent: req, callback: req.callback, } }() } } pending.Wait() requests := make([]*nodeRequest, 0, len(children)) for done := false; !done; { select { case miss := <-missing: requests = append(requests, miss) default: done = true } } return requests, nil } // commitNodeRequest finalizes a retrieval request and stores it into the membatch. If any // of the referencing parent requests complete due to this commit, they are also // committed themselves. func (s *Sync) commitNodeRequest(req *nodeRequest) error { // Write the node content to the membatch owner, path := ResolvePath(req.path) s.membatch.addNode(owner, path, req.data, req.hash) // Removed the completed node request delete(s.nodeReqs, string(req.path)) s.fetches[len(req.path)]-- // Check parent for completion if req.parent != nil { req.parent.deps-- if req.parent.deps == 0 { if err := s.commitNodeRequest(req.parent); err != nil { return err } } } return nil } // commitCodeRequest finalizes a retrieval request and stores it into the membatch. If any // of the referencing parent requests complete due to this commit, they are also // committed themselves. func (s *Sync) commitCodeRequest(req *codeRequest) error { // Write the node content to the membatch s.membatch.addCode(req.hash, req.data) // Removed the completed code request delete(s.codeReqs, req.hash) s.fetches[len(req.path)]-- // Check all parents for completion for _, parent := range req.parents { parent.deps-- if parent.deps == 0 { if err := s.commitNodeRequest(parent); err != nil { return err } } } return nil } // hasNode reports whether the specified trie node is present in the database. // 'exists' is true when the node exists in the database and matches the given root // hash. The 'inconsistent' return value is true when the node exists but does not // match the expected hash. func (s *Sync) hasNode(owner common.Hash, path []byte, hash common.Hash) (exists bool, inconsistent bool) { // If node is running with hash scheme, check the presence with node hash. if s.scheme == rawdb.HashScheme { return rawdb.HasLegacyTrieNode(s.database, hash), false } // If node is running with path scheme, check the presence with node path. var blob []byte if owner == (common.Hash{}) { blob = rawdb.ReadAccountTrieNode(s.database, path) } else { blob = rawdb.ReadStorageTrieNode(s.database, owner, path) } h := newBlobHasher() defer h.release() exists = hash == h.hash(blob) inconsistent = !exists && len(blob) != 0 return exists, inconsistent } // ResolvePath resolves the provided composite node path by separating the // path in account trie if it's existent. func ResolvePath(path []byte) (common.Hash, []byte) { var owner common.Hash if len(path) >= 2*common.HashLength { owner = common.BytesToHash(hexToKeybytes(path[:2*common.HashLength])) path = path[2*common.HashLength:] } return owner, path } // blobHasher is used to compute the sha256 hash of the provided data. type blobHasher struct{ state crypto.KeccakState } // blobHasherPool is the pool for reusing pre-allocated hash state. var blobHasherPool = sync.Pool{ New: func() interface{} { return &blobHasher{state: crypto.NewKeccakState()} }, } func newBlobHasher() *blobHasher { return blobHasherPool.Get().(*blobHasher) } func (h *blobHasher) hash(data []byte) common.Hash { return crypto.HashData(h.state, data) } func (h *blobHasher) release() { blobHasherPool.Put(h) }