forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-12 23:13:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth/protocols/snap: add tests and comments (#35477)

Browse source 
This PR updates some descriptions of snap sync v2, attaching two unit
tests.
This commit is contained in:
rjl493456442 2026-08-11 15:15:31 +08:00 committed by GitHub
parent 255842b750
commit 42c5059b58
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 391 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
eth/protocols/snap/bal_apply.go
View file

@ -94,6 +94,12 @@ func (s *syncerV2) isStorageFetched(accountHash, storageHash common.Hash) bool {
// in the database. For each account, it applies the post-block values (highest
// TxIdx entry) for balance, nonce, code, and storage. The storageRoot field is
// intentionally left stale. It will be recomputed during the trie generation.
//
// Correctness rests on the access list enumerating every storage change as an
// individual slot write. This holds post EIP-6780: pre-existing contracts can
// no longer be destructed, so storage only changes via SSTOREs (all recorded),
// Networks with the legacy SELFDESTRUCT break this premise: wholesale storage
// wipes carry no per-slot writes, leaving already-downloaded slots stale.
func (s *syncerV2) applyAccessList(b *bal.BlockAccessList, batch ethdb.Batch) error {
// Iterate over all accounts in the access list
for _, access := range *b {
@ -166,8 +172,7 @@ func (s *syncerV2) applyAccessList(b *bal.BlockAccessList, batch ethdb.Batch) er
}
// Don't create empty accounts in flat state (EIP-161).
isEmpty := account.Balance.IsZero() && account.Nonce == 0 &&
bytes.Equal(account.CodeHash, types.EmptyCodeHash[:])
isEmpty := account.Balance.IsZero() && account.Nonce == 0 && bytes.Equal(account.CodeHash, types.EmptyCodeHash[:])
switch {
case isEmpty && isNew:
// This covers cases where an account is created and destroyed within the

73
eth/protocols/snap/syncv2.go
View file

@ -368,6 +368,14 @@ type SyncPeerV2 interface {
// - The peer remains connected, but does not deliver a response in time
// - The peer delivers a stale response after a previous timeout
// - The peer delivers a refusal to serve the requested state
//
// The whole design is premised on EIP-6780 (self-destruct disabled): already
// downloaded state is never refetched, it is only rolled forward by applying
// BAL diffs, so every state change must surface in the access lists. A legacy
// SELFDESTRUCT wipes an entire storage trie without per-slot writes, which the
// BALs cannot express; the wiped slots would survive in the flat state and
// break the sync's core assumption. Networks with self-destruct enabled are
// therefore not supported.
type syncerV2 struct {
db ethdb.Database // Database to store the trie nodes into (and dedup)
scheme string // Node scheme used in node database
@ -376,20 +384,22 @@ type syncerV2 struct {
tasks []*accountTaskV2 // Current account task set being synced
update chan struct{} // Notification channel for possible sync progression
peers map[string]SyncPeerV2 // Currently active peers to download from
peerJoin *event.Feed // Event feed to react to peers joining
peerDrop *event.Feed // Event feed to react to peers dropping
rates *msgrate.Trackers // Message throughput rates for peers
peerJoin *event.Feed // Event feed to react to peers joining
peerDrop *event.Feed // Event feed to react to peers dropping
rates *msgrate.Trackers // Message throughput rates for peers
// Peer tracking during syncing phase.
//
// These fields should be protected by lock.
peers map[string]SyncPeerV2 // Currently active peers to download from
statelessPeers map[string]struct{} // Peers that failed to deliver state data
accountIdlers map[string]struct{} // Peers that aren't serving account requests
bytecodeIdlers map[string]struct{} // Peers that aren't serving bytecode requests
storageIdlers map[string]struct{} // Peers that aren't serving storage requests
accessListIdlers map[string]struct{} // Peers that aren't serving BAL requests
// Request tracking during syncing phase.
//
// These fields should be protected by lock.
statelessPeers map[string]struct{} // Peers that failed to deliver state data
accountIdlers map[string]struct{} // Peers that aren't serving account requests
bytecodeIdlers map[string]struct{} // Peers that aren't serving bytecode requests
storageIdlers map[string]struct{} // Peers that aren't serving storage requests
accessListIdlers map[string]struct{} // Peers that aren't serving BAL requests
// These fields should be protected by lock.
accountReqs map[uint64]*accountRequestV2 // Account requests currently running
bytecodeReqs map[uint64]*bytecodeRequestV2 // Bytecode requests currently running
@ -507,7 +517,6 @@ func (s *syncerV2) Unregister(id string) error {
// Remove status markers, even if no sync is running
delete(s.statelessPeers, id)
delete(s.accountIdlers, id)
delete(s.storageIdlers, id)
delete(s.bytecodeIdlers, id)
@ -529,6 +538,8 @@ func (s *syncerV2) Sync(target *types.Header, cancel chan struct{}) error {
s.lock.Lock()
s.statelessPeers = make(map[string]struct{})
s.lock.Unlock()
// Track the time when snap sync actually starts
if s.startTime.IsZero() {
s.startTime = time.Now()
}
@ -798,7 +809,6 @@ func isPivotReorged(db ethdb.Database, prev, curr *types.Header) bool {
if canonical == (common.Hash{}) {
return true
}
// If canonical at the old pivot's height has a different hash, the
// old pivot was reorged out.
return canonical != prev.Hash()
@ -1325,26 +1335,26 @@ func (s *syncerV2) pruneStaleState() error {
deleteKeyRange(batch, accountKey(task.Next), keyRangeLimit(bytes.Clone(rawdb.SnapshotAccountPrefix), task.Last))
protected := make([]common.Hash, 0, len(task.stateCompleted)+len(task.SubTasks))
for hash := range task.stateCompleted {
if bytes.Compare(hash[:], task.Next[:]) < 0 {
for accountHash := range task.stateCompleted {
if bytes.Compare(accountHash[:], task.Next[:]) < 0 {
return errors.New("unexpected storage marker before the range")
}
if bytes.Compare(hash[:], task.Last[:]) > 0 {
if bytes.Compare(accountHash[:], task.Last[:]) > 0 {
return errors.New("unexpected storage marker after the range")
}
protected = append(protected, hash)
protected = append(protected, accountHash)
}
for hash := range task.SubTasks {
if _, ok := task.stateCompleted[hash]; ok {
for accountHash := range task.SubTasks {
if _, ok := task.stateCompleted[accountHash]; ok {
return errors.New("unexpected duplicated storage marker")
}
if bytes.Compare(hash[:], task.Next[:]) < 0 {
if bytes.Compare(accountHash[:], task.Next[:]) < 0 {
return errors.New("unexpected storage marker before the range")
}
if bytes.Compare(hash[:], task.Last[:]) > 0 {
if bytes.Compare(accountHash[:], task.Last[:]) > 0 {
return errors.New("unexpected storage marker after the range")
}
protected = append(protected, hash)
protected = append(protected, accountHash)
}
sort.Slice(protected, func(i, j int) bool {
return bytes.Compare(protected[i][:], protected[j][:]) < 0
@ -1524,8 +1534,7 @@ func (s *syncerV2) cleanStorageTasks() {
delete(task.SubTasks, account)
task.pend--
// Mark the state as complete to prevent resyncing, regardless
// if state healing is necessary.
// Mark the state as complete to prevent resyncing
task.stateCompleted[account] = struct{}{}
// If this was the last pending task, forward the account task
@ -2137,8 +2146,9 @@ func (s *syncerV2) processAccountResponse(res *accountResponseV2) {
// Check if the account is a contract with an unknown storage trie
if account.Root != types.EmptyRootHash {
// If the storage was already retrieved in the last cycle, there's no need
// to resync it again, regardless of whether the storage root is consistent
// or not.
// to resync it again, the state difference has already been fixed in the
// bal catchup stage. This relies on access lists covering every storage
// change, guaranteed post EIP-6780.
if _, exist := res.task.stateCompleted[res.hashes[i]]; exist {
// The leftover storage tasks are not expected, unless system is
// very wrong.
@ -2158,10 +2168,6 @@ func (s *syncerV2) processAccountResponse(res *accountResponseV2) {
resumed[res.hashes[i]] = struct{}{}
largeStorageResumedGauge.Inc(1)
} else {
// It's possible that in the hash scheme, the storage, along
// with the trie nodes of the given root, is already present
// in the database. Schedule the storage task anyway to simplify
// the logic here.
res.task.stateTasks[res.hashes[i]] = account.Root
}
res.task.needState[i] = true
@ -2185,7 +2191,7 @@ func (s *syncerV2) processAccountResponse(res *accountResponseV2) {
continue
}
if _, ok := resumed[hash]; !ok {
log.Warn("Aborting suspended storage retrieval", "account", hash)
log.Error("Aborting suspended storage retrieval", "account", hash)
delete(res.task.SubTasks, hash)
largeStorageDiscardGauge.Inc(1)
}
@ -2270,8 +2276,7 @@ func (s *syncerV2) processStorageResponse(res *storageResponseV2) {
res.mainTask.stateTasks[account] = res.roots[i]
continue
}
// State was delivered, if complete mark as not needed any more, otherwise
// mark the account as needing healing
// State was delivered, if complete mark as not needed any more
for j, hash := range res.mainTask.res.hashes {
if account != hash {
continue
@ -2485,6 +2490,7 @@ func (s *syncerV2) OnAccounts(peer SyncPeerV2, id uint64, hashes []common.Hash,
}
}()
s.lock.Lock()
// Ensure the response is for a valid request
req, ok := s.accountReqs[id]
if !ok {
@ -2531,6 +2537,7 @@ func (s *syncerV2) OnAccounts(peer SyncPeerV2, id uint64, hashes []common.Hash,
cont, err := trie.VerifyRangeProof(root, req.origin[:], keys, accounts, nodes.Set())
if err != nil {
logger.Warn("Account range failed proof", "err", err)
// Signal this request as failed, and ready for rescheduling
s.scheduleRevertAccountRequest(req)
return err

344
eth/protocols/snap/syncv2_test.go
View file

@ -3535,3 +3535,347 @@ func testCatchUpAppliesStorageBALs(t *testing.T, scheme string) {
checkSlot(slotNew, vNew, true)
checkSlot(slotMultiTx, vMultiFinal, true)
}
// suspendChunkedContractSync runs an interrupted first download cycle against
// the given state. Storage responses are byte-capped so the contract switches
// into chunked (large-contract) mode, and the cycle is cancelled on the first
// subtask continuation request, leaving suspended SubTasks in the journal and
// a partially downloaded storage prefix on disk. Both are asserted before
// returning.
func suspendChunkedContractSync(t *testing.T, db ethdb.Database, scheme string, pivot *types.Header, accTrie *trie.Trie, accElems []*kv, stTrie *trie.Trie, stElems []*kv, contractHash common.Hash) {
t.Helper()
var (
once sync.Once
cancel = make(chan struct{})
term = func() { once.Do(func() { close(cancel) }) }
)
syncer := newSyncerV2(db, scheme)
src := newTestPeerV2("suspend-seed", t, term)
src.accountTrie = accTrie.Copy()
src.accountValues = accElems
src.setStorageTries(map[common.Hash]*trie.Trie{contractHash: stTrie})
src.storageValues = map[common.Hash][]*kv{
contractHash: stElems,
}
src.storageRequestV2Handler = func(tp *testPeerV2, id uint64, root common.Hash, accounts []common.Hash, origin, limit []byte, max int) error {
// A continuation request proves the first chunked response was fully
// processed and the subtasks exist; cut the cycle right there.
if len(origin) > 0 {
term()
return nil
}
// Byte-cap the initial response so the contract gets chunked.
return defaultStorageRequestHandlerV2(tp, id, root, accounts, origin, limit, 2000)
}
syncer.Register(src)
src.remote = syncer
syncer.loadSyncStatus()
syncer.pivot = pivot // Sync pins this before downloadState
syncer.downloadState(cancel)
syncer.saveSyncStatus()
// The journal must carry the suspended subtasks and a partial storage
// prefix must be on disk, otherwise the fixture proves nothing.
loaded := newSyncerV2(db, scheme)
loaded.loadSyncStatus()
suspended := false
for _, task := range loaded.tasks {
if len(task.SubTasks[contractHash]) > 0 {
suspended = true
}
}
if !suspended {
t.Fatal("fixture: no suspended storage subtasks journaled")
}
downloaded := 0
for _, entry := range stElems {
if len(rawdb.ReadStorageSnapshot(db, contractHash, common.BytesToHash(entry.k))) > 0 {
downloaded++
}
}
if downloaded == 0 || downloaded == len(stElems) {
t.Fatalf("fixture: want partially downloaded storage, got %d/%d slots", downloaded, len(stElems))
}
}
// TestPivotMoveAbortsEmptiedStorage covers the suspended-storage abort path:
// a chunked contract download is interrupted, then the pivot moves to a block
// whose BAL zeroes every slot of the contract (the only way a storage root can
// become empty post EIP-6780). The follow-up Sync must delete the downloaded
// prefix during catch-up, abort the suspended subtasks when the account
// returns with an empty root, and complete against the exact new state root.
func TestPivotMoveAbortsEmptiedStorage(t *testing.T) {
t.Parallel()
testPivotMoveAbortsEmptiedStorage(t, rawdb.HashScheme)
testPivotMoveAbortsEmptiedStorage(t, rawdb.PathScheme)
}
func testPivotMoveAbortsEmptiedStorage(t *testing.T, scheme string) {
contractAddr := common.HexToAddress("0x00000000000000000000000000000000c0ffee02")
contractHash := crypto.Keccak256Hash(contractAddr[:])
// Enough slots that a byte-capped response leaves most of them undelivered.
slotsA := make(map[common.Hash]common.Hash, 500)
for i := 0; i < 500; i++ {
slotsA[common.BigToHash(big.NewInt(int64(i+1)))] = common.BigToHash(big.NewInt(int64(0x10000 + i)))
}
contractTmpl := types.StateAccount{
Nonce: 7,
Balance: uint256.NewInt(123456),
CodeHash: types.EmptyCodeHash[:],
}
// Storage-less filler accounts, identical at A and A+1.
_, _, plain, _ := makeAccountTrieWithAddresses(20, scheme)
// State at pivot A (contract populated) and at A+1 (contract emptied).
accTrieA, accElemsA, stTrieA, stElemsA, rootA := makeStateWithStorageContract(scheme, plain, contractAddr, contractTmpl, slotsA)
accTrieB, accElemsB, _, _, rootB := makeStateWithStorageContract(scheme, plain, contractAddr, contractTmpl, nil)
// The A+1 BAL zeroes every slot: post EIP-6780 an emptied storage is
// always visible as per-slot writes.
cb := bal.NewConstructionBlockAccessList()
for raw := range slotsA {
cb.StorageWrite(0, contractAddr, raw, common.Hash{})
}
var balBuf bytes.Buffer
if err := cb.EncodeRLP(&balBuf); err != nil {
t.Fatal(err)
}
var decodedBAL bal.BlockAccessList
if err := rlp.DecodeBytes(balBuf.Bytes(), &decodedBAL); err != nil {
t.Fatal(err)
}
balHash := decodedBAL.Hash()
db := rawdb.NewMemoryDatabase()
numA := uint64(128)
emptyH := common.Hash{}
zero := uint64(0)
hdrA := &types.Header{
Number: new(big.Int).SetUint64(numA),
Root: rootA,
Difficulty: common.Big0,
BaseFee: common.Big0,
WithdrawalsHash: &emptyH,
BlobGasUsed: &zero,
ExcessBlobGas: &zero,
ParentBeaconRoot: &emptyH,
RequestsHash: &emptyH,
}
rawdb.WriteHeader(db, hdrA)
rawdb.WriteCanonicalHash(db, hdrA.Hash(), numA)
hdrB := &types.Header{
ParentHash: hdrA.Hash(),
Number: new(big.Int).SetUint64(numA + 1),
Root: rootB,
Difficulty: common.Big0,
BaseFee: common.Big0,
WithdrawalsHash: &emptyH,
BlobGasUsed: &zero,
ExcessBlobGas: &zero,
ParentBeaconRoot: &emptyH,
RequestsHash: &emptyH,
BlockAccessListHash: &balHash,
}
rawdb.WriteHeader(db, hdrB)
rawdb.WriteCanonicalHash(db, hdrB.Hash(), numA+1)
// Cycle 1: interrupted download at pivot A with suspended contract subtasks.
suspendChunkedContractSync(t, db, scheme, hdrA, accTrieA, accElemsA, stTrieA, stElemsA, contractHash)
// Cycle 2: the pivot moves to A+1. Catch-up must delete the downloaded
// slots, the account response (empty root) must abort the suspended
// subtasks, and the sync must complete against rootB.
var (
once sync.Once
cancel = make(chan struct{})
term = func() { once.Do(func() { close(cancel) }) }
)
syncer := newSyncerV2(db, scheme)
src := newTestPeerV2("emptied-serve", t, term)
src.accountTrie = accTrieB.Copy()
src.accountValues = accElemsB
src.accessLists = map[common.Hash]rlp.RawValue{
hdrB.Hash(): balBuf.Bytes(),
}
syncer.Register(src)
src.remote = syncer
done := checkStall(t, term)
if err := syncer.Sync(hdrB, cancel); err != nil {
t.Fatalf("pivot move sync failed: %v", err)
}
close(done)
// The emptied contract must not have triggered any storage retrieval.
if n := src.nStorageRequests.Load(); n != 0 {
t.Errorf("unexpected storage requests for emptied contract: %d", n)
}
// The generation inside Sync already verified rootB; re-walk independently.
verifyTrie(scheme, db, rootB, t)
// Not a single downloaded slot may survive.
for _, entry := range stElemsA {
if v := rawdb.ReadStorageSnapshot(db, contractHash, common.BytesToHash(entry.k)); len(v) != 0 {
t.Errorf("stale slot %x survived the emptied contract: %x", entry.k, v)
}
}
}
// TestShortAccountResponseKeepsSuspendedStorage covers the keep branch of the
// suspended-subtask sweep: a resumed cycle whose first account response ends
// short of the contract must keep the suspended subtasks (they lie beyond the
// response), then resume them — not restart from scratch — once a later wave
// covers the contract.
func TestShortAccountResponseKeepsSuspendedStorage(t *testing.T) {
t.Parallel()
testShortAccountResponseKeepsSuspendedStorage(t, rawdb.HashScheme)
testShortAccountResponseKeepsSuspendedStorage(t, rawdb.PathScheme)
}
func testShortAccountResponseKeepsSuspendedStorage(t *testing.T, scheme string) {
contractAddr := common.HexToAddress("0x00000000000000000000000000000000c0ffee03")
contractHash := crypto.Keccak256Hash(contractAddr[:])
// Craft neighbour accounts right around the contract hash. They must share
// its leading nibble so all of them land in the contract's account task:
// two below give the shortened response some content, one sits above.
mkPlain := func(key common.Hash, nonce uint64) *kv {
val, _ := rlp.EncodeToBytes(&types.StateAccount{
Nonce: nonce,
Balance: uint256.NewInt(1000 + nonce),
Root: types.EmptyRootHash,
CodeHash: types.EmptyCodeHash[:],
})
return &kv{key.Bytes(), val}
}
shift := func(h common.Hash, delta int64) common.Hash {
return common.BigToHash(new(big.Int).Add(h.Big(), big.NewInt(delta)))
}
below1, below2, above := shift(contractHash, -2), shift(contractHash, -1), shift(contractHash, 1)
if below1[0]>>4 != contractHash[0]>>4 || above[0]>>4 != contractHash[0]>>4 {
t.Fatal("fixture: neighbour accounts left the contract's task range")
}
plain := []*kv{
mkPlain(below1, 1),
mkPlain(below2, 2),
mkPlain(above, 3),
}
slots := make(map[common.Hash]common.Hash, 500)
for i := 0; i < 500; i++ {
slots[common.BigToHash(big.NewInt(int64(i+1)))] = common.BigToHash(big.NewInt(int64(0x20000 + i)))
}
contractTmpl := types.StateAccount{
Nonce: 7,
Balance: uint256.NewInt(123456),
CodeHash: types.EmptyCodeHash[:],
}
accTrie, accElems, stTrie, stElems, root := makeStateWithStorageContract(scheme, plain, contractAddr, contractTmpl, slots)
db := rawdb.NewMemoryDatabase()
pivot := mkPivot(0, root)
// Cycle 1: interrupted download with suspended contract subtasks.
suspendChunkedContractSync(t, db, scheme, pivot, accTrie, accElems, stTrie, stElems, contractHash)
// Cycle 2: resume at the same pivot. The first response of the contract's
// task is truncated below the contract; the follow-up wave serves it in
// full. Watch the storage requests to tell resumption from a restart.
var (
once sync.Once
cancel = make(chan struct{})
term = func() { once.Do(func() { close(cancel) }) }
truncated atomic.Bool
freshFetch atomic.Bool
resumed atomic.Bool
)
syncer := newSyncerV2(db, scheme)
src := newTestPeerV2("resume-serve", t, term)
src.accountTrie = accTrie.Copy()
src.accountValues = accElems
src.setStorageTries(map[common.Hash]*trie.Trie{contractHash: stTrie})
src.storageValues = map[common.Hash][]*kv{
contractHash: stElems,
}
src.accountRequestV2Handler = func(tp *testPeerV2, id uint64, reqRoot common.Hash, origin common.Hash, limit common.Hash, cap int) error {
if !truncated.Load() && bytes.Compare(origin[:], contractHash[:]) <= 0 && bytes.Compare(contractHash[:], limit[:]) <= 0 {
truncated.Store(true)
// Serve only the accounts below the contract: a legitimate
// shortened response whose proof marks a continuation.
var (
keys []common.Hash
vals [][]byte
)
for _, entry := range tp.accountValues {
if bytes.Compare(entry.k, origin[:]) < 0 {
continue
}
// Explicitly truncate the response at the contract, leaving the
// subtasks as suspended.
if bytes.Compare(entry.k, contractHash[:]) >= 0 {
break
}
keys = append(keys, common.BytesToHash(entry.k))
vals = append(vals, entry.v)
}
if len(keys) == 0 {
t.Errorf("fixture: shortened response empty, origin %x", origin)
}
proof := trienode.NewProofSet()
if err := tp.accountTrie.Prove(origin[:], proof); err != nil {
t.Errorf("could not prove origin: %v", err)
}
if len(keys) > 0 {
if err := tp.accountTrie.Prove(keys[len(keys)-1].Bytes(), proof); err != nil {
t.Errorf("could not prove last item: %v", err)
}
}
if err := tp.remote.OnAccounts(tp, id, keys, vals, proof.List()); err != nil {
t.Errorf("remote side rejected shortened delivery: %v", err)
tp.term()
}
return nil
}
return defaultAccountRequestHandlerV2(tp, id, reqRoot, origin, limit, cap)
}
src.storageRequestV2Handler = func(tp *testPeerV2, id uint64, reqRoot common.Hash, accounts []common.Hash, origin, limit []byte, max int) error {
for _, account := range accounts {
if account == contractHash {
if len(origin) > 0 {
resumed.Store(true)
} else {
freshFetch.Store(true)
}
}
}
return defaultStorageRequestHandlerV2(tp, id, reqRoot, accounts, origin, limit, max)
}
syncer.Register(src)
src.remote = syncer
done := checkStall(t, term)
if err := syncer.Sync(pivot, cancel); err != nil {
t.Fatalf("resumed sync failed: %v", err)
}
close(done)
if !truncated.Load() {
t.Fatal("shortened account response never served")
}
if freshFetch.Load() {
t.Error("suspended subtasks were aborted: contract storage rescheduled from scratch")
}
if !resumed.Load() {
t.Error("suspended subtasks never resumed")
}
verifyTrie(scheme, db, root, t)
for _, entry := range stElems {
if len(rawdb.ReadStorageSnapshot(db, contractHash, common.BytesToHash(entry.k))) == 0 {
t.Errorf("missing slot %x after resumed download", entry.k)
}
}
}