diff --git a/eth/protocols/snap/bal_apply.go b/eth/protocols/snap/bal_apply.go index d67015361b..15c88188c3 100644 --- a/eth/protocols/snap/bal_apply.go +++ b/eth/protocols/snap/bal_apply.go @@ -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 diff --git a/eth/protocols/snap/syncv2.go b/eth/protocols/snap/syncv2.go index 29607a2c83..9fc59097c8 100644 --- a/eth/protocols/snap/syncv2.go +++ b/eth/protocols/snap/syncv2.go @@ -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 diff --git a/eth/protocols/snap/syncv2_test.go b/eth/protocols/snap/syncv2_test.go index 96f8913d56..faa19efcee 100644 --- a/eth/protocols/snap/syncv2_test.go +++ b/eth/protocols/snap/syncv2_test.go @@ -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) + } + } +}