From ea01f2bbbfb7d191465edf4f8725a9ad05cc5bd4 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Mon, 20 Jul 2026 20:06:04 +0800 Subject: [PATCH] fix(consensus/XDPoS,core,eth/downloader): rebuild missing V2 gap snapshots A node can lose its persisted V2 gap snapshot when the process exits between writeHeadBlock and StoreSnapshot in writeBlockWithState: the head markers are already on disk, the snapshot is not, and nothing recreates it. getSnapshot then fails with a leveldb "not found" error for the whole affected epoch and the node drops out of consensus participation. Making that write ordering atomic prevents new holes but cannot repair a database that already has one. Add a guarded self-healing path in engine_v2.getSnapshot: when no snapshot is stored for the gap block, rebuild it from the committed state trie at gapHeader.Root, then persist and cache it. The derivation itself moves into engine_v2.BuildSnapshotFromState, which eth/downloader.generateSnapshot now also uses. The downloader already carried its own copy of this derivation, and the unstable xdc_sort ordering it shares with core.BlockChain.UpdateM1 must not drift: a different equal-stake order yields a different masternode set. The shared builder also refuses to produce an empty snapshot, which the downloader previously persisted and which would then permanently mask the missing masternode list. A gap pivot state sync that finds no candidates therefore now fails loudly instead of storing that empty snapshot. The rebuild is only attempted when it is both safe and useful: - skip when a snapshot is already stored: only a genuinely missing key is healed, so a decode or I/O error never overwrites a stored masternode set; - skip at or before V2 SwitchBlock, where the initial snapshot still comes from Initial() rather than the state trie; - reject numbers that are not real gap blocks, since they can originate from unauthenticated vote/timeout messages; - skip gap blocks more than two epochs behind the chain head, so a peer cannot force trie reads and database writes for arbitrary historic gap numbers; - skip when the chain reader does not implement the new GapStateReader interface, which core.BlockChain satisfies via a compile-time assertion; - skip while the gap block itself is not imported yet, where the missing state is transient and retrying later succeeds. Past that point the gap block is imported, so a rebuild that failed on its state cannot start working later: the trie is pruned and can never come back, so the gap block is recorded in an LRU and never retried, instead of re-reading the trie on every vote verification. Concurrent callers for the same gap block are collapsed with singleflight. Pruned state is logged at warn rather than error, and everything degrades to debug while syncing, where a missing gap snapshot is expected and not actionable. The rebuild needs the gap block's state root to still be readable, which the targeted case implies: a node only keeps the gap block as its head across a restart when that root was committed. Otherwise loadLastState finds no head state, repair() rewinds to an ancestor that has one, and re-importing the gap block runs UpdateM1 and writes the snapshot again. What is left are roots that are gone for good, such as offline pruning, gap blocks below a fast sync pivot, or a side chain whose trie was collected. No later attempt can recover those, which is what the permanent give-up above encodes. A rebuilt snapshot is derived from the gap block state, while UpdateM1 takes candidates from the head state and stakes from the validator contract at "latest". Those agree on the canonical import path, where the head is the gap block itself, but not necessarily on the reorg path, so a rebuilt masternode set can differ from what a peer persisted through a reorg. That is accepted here: the alternative is no snapshot at all. Add coverage for the state-derived builder and for every getSnapshot guard, and seed the downloader test genesis with the minimum masternode voting contract storage the builder reads. --- consensus/XDPoS/engines/engine_v2/engine.go | 8 + consensus/XDPoS/engines/engine_v2/snapshot.go | 202 ++++++- .../XDPoS/engines/engine_v2/snapshot_test.go | 565 ++++++++++++++++++ core/blockchain.go | 1 + core/blockchain_reader.go | 5 + core/rawdb/accessors_xdc.go | 5 + eth/backend.go | 1 + eth/downloader/downloader.go | 35 +- eth/downloader/testchain_test.go | 20 +- 9 files changed, 811 insertions(+), 31 deletions(-) diff --git a/consensus/XDPoS/engines/engine_v2/engine.go b/consensus/XDPoS/engines/engine_v2/engine.go index ea0fa4d93b61..15afb4ceb560 100644 --- a/consensus/XDPoS/engines/engine_v2/engine.go +++ b/consensus/XDPoS/engines/engine_v2/engine.go @@ -30,6 +30,7 @@ import ( "github.com/XinFinOrg/XDPoSChain/params" "github.com/XinFinOrg/XDPoSChain/trie" "golang.org/x/sync/errgroup" + "golang.org/x/sync/singleflight" ) type XDPoS_v2 struct { @@ -41,6 +42,8 @@ type XDPoS_v2 struct { whosTurn common.Address // Record waiting for who to mine snapshots *lru.Cache[common.Hash, *SnapshotV2] // Snapshots for gap block + snapshotFlight singleflight.Group // Prevents concurrent snapshot reconstruction for the same gap block + failedRebuilds *lru.Cache[common.Hash, struct{}] // Gap blocks whose snapshot cannot be rebuilt, never retried signatures *utils.SigLRU // Signatures of recent blocks to speed up mining epochSwitches *lru.Cache[common.Hash, *types.EpochSwitchInfo] // infos of epoch: master nodes, epoch switch block info, parent of that info verifiedHeaders *lru.Cache[common.Hash, struct{}] @@ -78,6 +81,10 @@ type XDPoS_v2 struct { HookReward func(chain consensus.ChainReader, state vm.StateDB, parentState *state.StateDB, header *types.Header) (map[string]interface{}, error) HookPenalty func(chain consensus.ChainReader, number *big.Int, parentHash common.Hash, candidates []common.Address) ([]common.Address, error) + // HookSyncing reports whether the node is currently syncing from the network. + // It is optional and only used to keep expected-while-syncing logs quiet. + HookSyncing func() bool + ForensicsProcessor *Forensics votePoolCollectionTime time.Time @@ -114,6 +121,7 @@ func New(chainConfig *params.ChainConfig, db ethdb.Database, minePeriodCh chan i verifiedHeaders: lru.NewCache[common.Hash, struct{}](utils.InMemorySnapshots), snapshots: lru.NewCache[common.Hash, *SnapshotV2](utils.InMemorySnapshots), + failedRebuilds: lru.NewCache[common.Hash, struct{}](utils.InMemorySnapshots), epochSwitches: lru.NewCache[common.Hash, *types.EpochSwitchInfo](int(utils.InMemoryEpochs)), timeoutWorker: timeoutTimer, BroadcastCh: make(chan interface{}), diff --git a/consensus/XDPoS/engines/engine_v2/snapshot.go b/consensus/XDPoS/engines/engine_v2/snapshot.go index f0439bfaa8cf..7bac56888c68 100644 --- a/consensus/XDPoS/engines/engine_v2/snapshot.go +++ b/consensus/XDPoS/engines/engine_v2/snapshot.go @@ -2,15 +2,36 @@ package engine_v2 import ( "encoding/json" + "errors" "fmt" "github.com/XinFinOrg/XDPoSChain/common" + xdc_sort "github.com/XinFinOrg/XDPoSChain/common/sort" "github.com/XinFinOrg/XDPoSChain/consensus" + "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils" "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/state" + "github.com/XinFinOrg/XDPoSChain/core/types" "github.com/XinFinOrg/XDPoSChain/ethdb" "github.com/XinFinOrg/XDPoSChain/log" ) +// snapshotRebuildMaxEpochsBehind is how far behind the chain head a gap block +// may be for its snapshot to still be rebuilt on demand. +const snapshotRebuildMaxEpochsBehind = 2 + +// errGapStateUnavailable marks a rebuild that failed because the gap block's +// state trie is gone, which is the normal outcome on a pruning node. +var errGapStateUnavailable = errors.New("gap block state unavailable") + +// GapStateReader is an optional extension of consensus.ChainReader that allows +// engine_v2 to open committed state tries by root hash. +// *core.BlockChain satisfies this interface; lightweight mock readers used in +// tests typically do not, in which case self-healing is skipped. +type GapStateReader interface { + StateAt(root common.Hash) (*state.StateDB, error) +} + // Snapshot is the state of the smart contract validator list // The validator list is used on next epoch candidates nodes // If we don't have the snapshot, then we have to trace back the gap block smart contract state which is very costly @@ -56,6 +77,69 @@ func StoreSnapshot(s *SnapshotV2, db ethdb.Database) error { return rawdb.WriteXdposV2Snapshot(db, s.Hash, blob) } +// BuildSnapshotFromState derives a gap block snapshot from the candidate list +// and stakes held in a committed state trie, so no EVM client is required. It +// is the single source of truth for that derivation: eth/downloader uses it +// after a gap pivot state sync, engine_v2 uses it to rebuild snapshots that +// were never persisted. +// +// NOTE: keep code sync with core.BlockChain.UpdateM1 +func BuildSnapshotFromState(statedb *state.StateDB, number uint64, hash common.Hash) (*SnapshotV2, error) { + candidates := statedb.GetCandidates() + pairs := make([]utils.Masternode, 0, len(candidates)) + for _, candidate := range candidates { + // Redundant here (GetCandidates already skips zero addresses), kept to + // mirror UpdateM1, whose contract fallback can return them. + if !candidate.IsZero() { + pairs = append(pairs, utils.Masternode{Address: candidate, Stake: statedb.GetCandidateCap(candidate)}) + } + } + // An empty snapshot would load back fine and permanently mask the missing + // masternode list, so refuse to build one. + if len(pairs) == 0 { + return nil, fmt.Errorf("no candidates found in state of block %d (%s)", number, hash.Hex()) + } + // Must stay xdc_sort.Slice with this exact comparator: the comparator is not + // a strict weak ordering and the sort is unstable, so any other sort would + // order equal stakes differently from UpdateM1 and fork the masternode set. + xdc_sort.Slice(pairs, func(i, j int) bool { + return pairs[i].Stake.Cmp(pairs[j].Stake) >= 0 + }) + masterNodes := make([]common.Address, len(pairs)) + for i, p := range pairs { + masterNodes[i] = p.Address + // Lets an operator cross-check a suspect snapshot against the rest of the network. + log.Debug("Snapshot candidate", "number", number, "index", i, "address", p.Address, "stake", p.Stake) + } + return NewSnapshot(number, hash, masterNodes), nil +} + +// rebuildSnapshot reconstructs a missing gap block snapshot from the committed +// state trie at gapHeader.Root, then persists it and adds it to the in-memory +// LRU cache. +// +// core.BlockChain.UpdateM1 takes the candidate list from the head state but +// always reads stakes from the validator contract at "latest". Those are the +// gap block's values on the canonical import path, where the head is the gap +// block itself, but not necessarily on the reorg path, where UpdateM1 runs +// after the head has moved. A rebuilt snapshot therefore reflects the gap block +// state, which may differ from what a peer persisted through the reorg path. +func (x *XDPoS_v2) rebuildSnapshot(sr GapStateReader, gapHeader *types.Header) (*SnapshotV2, error) { + statedb, err := sr.StateAt(gapHeader.Root) + if err != nil { + return nil, fmt.Errorf("%w: %w", errGapStateUnavailable, err) + } + snap, err := BuildSnapshotFromState(statedb, gapHeader.Number.Uint64(), gapHeader.Hash()) + if err != nil { + return nil, err + } + if err := StoreSnapshot(snap, x.db); err != nil { + return nil, fmt.Errorf("cannot store rebuilt snapshot: %w", err) + } + x.snapshots.Add(snap.Hash, snap) + return snap, nil +} + // retrieves candidates nodes list in map type func (s *SnapshotV2) GetMappedCandidates() map[common.Address]struct{} { ms := make(map[common.Address]struct{}) @@ -74,6 +158,114 @@ func (s *SnapshotV2) IsCandidates(address common.Address) bool { return false } +// selfHealSnapshot rebuilds a gap block snapshot that was never persisted, e.g. +// because the process exited between writeHeadBlock and StoreSnapshot in +// writeBlockWithState, leaving the head markers on disk without the snapshot. +// +// It returns (nil, nil) when a rebuild is not attempted, in which case the +// caller must report the original load failure. +func (x *XDPoS_v2) selfHealSnapshot(chain consensus.ChainReader, gapHeader *types.Header) (*SnapshotV2, error) { + gapBlockNum := gapHeader.Number.Uint64() + gapBlockHash := gapHeader.Hash() + + // The initial V2 gap block snapshot comes from Initial(), not from the state trie. + if gapBlockNum <= x.config.V2.SwitchBlock.Uint64() { + return nil, nil + } + // The number can come straight from an unauthenticated vote/timeout message + // (isGapNumber), so reject non-gap numbers before doing any further work. + // Mirrors the assertion in UpdateMasternodes. + if gapBlockNum%x.config.Epoch != x.config.Epoch-x.config.Gap { + return nil, nil + } + // Entries are only recorded for an imported gap block whose trie is pruned, so + // a rebuild that failed once can never start working again. Checked before any + // database read, as this runs once per vote/timeout message. + if _, failed := x.failedRebuilds.Get(gapBlockHash); failed { + return nil, nil + } + // Without this an attacker could cycle through every historic gap number and + // force one trie read plus one database write each. + currentHeader := chain.CurrentHeader() + if currentHeader == nil { + return nil, nil + } + if head := currentHeader.Number.Uint64(); head > gapBlockNum+snapshotRebuildMaxEpochsBehind*x.config.Epoch { + log.Debug("Skip snapshot rebuild, gap block too far behind head", "number", gapBlockNum, "hash", gapBlockHash.Hex(), "head", head) + return nil, nil + } + sr, ok := chain.(GapStateReader) + if !ok { + log.Debug("Skip snapshot rebuild, chain reader cannot open state", "number", gapBlockNum, "hash", gapBlockHash.Hex()) + return nil, nil + } + // Canonical headers can run ahead of block import while syncing, so the state + // may simply not exist yet. This case must not be recorded as a permanent + // failure below, retrying once the block lands works. + if chain.GetBlock(gapBlockHash, gapBlockNum) == nil { + log.Debug("Skip snapshot rebuild, gap block not imported yet", "number", gapBlockNum, "hash", gapBlockHash.Hex()) + return nil, nil + } + + // A missing snapshot is expected while syncing: gap blocks below a fast sync + // pivot never run UpdateM1. That is not actionable, so keep it quiet. + syncing := x.HookSyncing != nil && x.HookSyncing() + logMissing := log.Warn + if syncing { + logMissing = log.Debug + } + + // Collapse concurrent callers (e.g. votes from many peers for the same gap + // block) into one rebuild. Logging happens inside the closure so that one + // rebuild logs one line. + // NOTE: some callers hold x.lock, so the trie read below blocks the consensus + // loop; it is bounded to one read per gap block. + result, err, _ := x.snapshotFlight.Do(string(gapBlockHash[:]), func() (interface{}, error) { + // A concurrent caller may have stored the snapshot while this one queued. + if snap, err := loadSnapshot(x.db, gapBlockHash); err == nil { + x.snapshots.Add(snap.Hash, snap) + return snap, nil + } + // Only a genuinely missing snapshot is healed. A decode error means the key + // is there, and overwriting it with a state-derived rebuild could replace a + // correct masternode set with a different one. + has, err := rawdb.HasXdposV2Snapshot(x.db, gapBlockHash) + if err != nil { + log.Debug("Skip snapshot rebuild, cannot probe stored snapshot", "err", err, "number", gapBlockNum, "hash", gapBlockHash.Hex()) + return nil, nil + } + if has { + return nil, nil + } + logMissing("Cannot find snapshot from last gap block, rebuilding", "number", gapBlockNum, "hash", gapBlockHash.Hex()) + rebuilt, err := x.rebuildSnapshot(sr, gapHeader) + if err != nil { + // rawdb.WriteXdposV2Snapshot log.Crit's on a failed Put, so a store error + // never reaches here and every failure below is final. + x.failedRebuilds.Add(gapBlockHash, struct{}{}) + logFailed := log.Error + switch { + case syncing: + logFailed = log.Debug + case errors.Is(err, errGapStateUnavailable): + logFailed = log.Warn + } + logFailed("Failed to rebuild snapshot", "err", err, "number", gapBlockNum, "hash", gapBlockHash.Hex(), "root", gapHeader.Root.Hex()) + return nil, err + } + log.Info("Rebuild snapshot OK", "number", gapBlockNum, "hash", gapBlockHash.Hex(), "root", gapHeader.Root.Hex(), "candidates", len(rebuilt.NextEpochCandidates)) + return rebuilt, nil + }) + if err != nil { + return nil, err + } + snap, ok := result.(*SnapshotV2) + if !ok || snap == nil { + return nil, nil + } + return snap, nil +} + // snapshot retrieves the authorization snapshot at a given point in time. func (x *XDPoS_v2) getSnapshot(chain consensus.ChainReader, number uint64, isGapNumber bool) (*SnapshotV2, error) { var gapBlockNum uint64 @@ -104,8 +296,14 @@ func (x *XDPoS_v2) getSnapshot(chain consensus.ChainReader, number uint64, isGap // If an on-disk checkpoint snapshot can be found, use that snap, err := loadSnapshot(x.db, gapBlockHash) if err != nil { - log.Error("Cannot find snapshot from last gap block", "err", err, "number", gapBlockNum, "hash", gapBlockHash) - return nil, err + rebuilt, rebuildErr := x.selfHealSnapshot(chain, gapHeader) + if rebuildErr != nil { + return nil, errors.Join(err, rebuildErr) + } + if rebuilt == nil { + return nil, err + } + return rebuilt, nil } log.Trace("Loaded snapshot from disk", "number", gapBlockNum, "hash", gapBlockHash) diff --git a/consensus/XDPoS/engines/engine_v2/snapshot_test.go b/consensus/XDPoS/engines/engine_v2/snapshot_test.go index 66289eb9f660..ac640792cdd4 100644 --- a/consensus/XDPoS/engines/engine_v2/snapshot_test.go +++ b/consensus/XDPoS/engines/engine_v2/snapshot_test.go @@ -1,12 +1,23 @@ package engine_v2 import ( + "errors" "fmt" + "math/big" + "sync" + "sync/atomic" "testing" "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/common/lru" + "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils" "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/state" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/ethdb" "github.com/XinFinOrg/XDPoSChain/ethdb/leveldb" + "github.com/XinFinOrg/XDPoSChain/params" + "github.com/stretchr/testify/assert" ) func TestGetMasterNodes(t *testing.T) { @@ -21,6 +32,10 @@ func TestGetMasterNodes(t *testing.T) { } } +// ============================================================================ +// Snapshot persistence and recovery tests +// ============================================================================ + func TestStoreLoadSnapshot(t *testing.T) { snap := NewSnapshot(1, common.Hash{0x1}, nil) dir := t.TempDir() @@ -40,3 +55,553 @@ func TestStoreLoadSnapshot(t *testing.T) { t.Error("load snapshot failed", err) } } + +type snapshotChainReader struct { + headersByNumber map[uint64]*types.Header + headersByHash map[common.Hash]*types.Header + blocksByHash map[common.Hash]*types.Block + statesByRoot map[common.Hash]*state.StateDB + currentHeader *types.Header +} + +func newSnapshotChainReader() *snapshotChainReader { + return &snapshotChainReader{ + headersByNumber: make(map[uint64]*types.Header), + headersByHash: make(map[common.Hash]*types.Header), + blocksByHash: make(map[common.Hash]*types.Block), + statesByRoot: make(map[common.Hash]*state.StateDB), + } +} + +// addHeader registers a fully imported block: both the canonical header and the +// block body are visible, as they are once insertChain is done with it. +func (m *snapshotChainReader) addHeader(header *types.Header) { + m.addHeaderOnly(header) + m.blocksByHash[header.Hash()] = types.NewBlockWithHeader(header) +} + +// addHeaderOnly registers a canonical header whose block is not imported yet, +// the state a node is in while headers run ahead of block import. +func (m *snapshotChainReader) addHeaderOnly(header *types.Header) { + m.headersByNumber[header.Number.Uint64()] = header + m.headersByHash[header.Hash()] = header + if m.currentHeader == nil || m.currentHeader.Number.Cmp(header.Number) < 0 { + m.currentHeader = header + } +} + +func (m *snapshotChainReader) setHead(number uint64) { + m.currentHeader = &types.Header{Number: new(big.Int).SetUint64(number)} +} + +func (m *snapshotChainReader) Config() *params.ChainConfig { + return nil +} + +func (m *snapshotChainReader) CurrentHeader() *types.Header { + return m.currentHeader +} + +func (m *snapshotChainReader) GetHeader(hash common.Hash, number uint64) *types.Header { + header := m.headersByHash[hash] + if header == nil || header.Number.Uint64() != number { + return nil + } + return header +} + +func (m *snapshotChainReader) GetHeaderByNumber(number uint64) *types.Header { + return m.headersByNumber[number] +} + +func (m *snapshotChainReader) GetHeaderByHash(hash common.Hash) *types.Header { + return m.headersByHash[hash] +} + +func (m *snapshotChainReader) GetBlock(hash common.Hash, number uint64) *types.Block { + block := m.blocksByHash[hash] + if block == nil || block.NumberU64() != number { + return nil + } + return block +} + +func (m *snapshotChainReader) StateAt(root common.Hash) (*state.StateDB, error) { + st, ok := m.statesByRoot[root] + if !ok { + return nil, errors.New("state not found") + } + return st, nil +} + +// headerOnlyChainReader is a consensus.ChainReader that deliberately does not +// implement GapStateReader, so snapshot self-healing cannot run. +type headerOnlyChainReader struct { + inner *snapshotChainReader +} + +func (m *headerOnlyChainReader) Config() *params.ChainConfig { return m.inner.Config() } +func (m *headerOnlyChainReader) CurrentHeader() *types.Header { return m.inner.CurrentHeader() } +func (m *headerOnlyChainReader) GetHeader(hash common.Hash, number uint64) *types.Header { + return m.inner.GetHeader(hash, number) +} +func (m *headerOnlyChainReader) GetHeaderByNumber(number uint64) *types.Header { + return m.inner.GetHeaderByNumber(number) +} +func (m *headerOnlyChainReader) GetHeaderByHash(hash common.Hash) *types.Header { + return m.inner.GetHeaderByHash(hash) +} +func (m *headerOnlyChainReader) GetBlock(hash common.Hash, number uint64) *types.Block { + return m.inner.GetBlock(hash, number) +} + +// testGapNumber is a V2-era gap block number for the config in newTestEngineV2: +// testGapNumber%Epoch == Epoch-Gap and testGapNumber > V2.SwitchBlock. +const testGapNumber = 1350 + +func newTestEngineV2(db ethdb.Database) *XDPoS_v2 { + return &XDPoS_v2{ + config: ¶ms.XDPoSConfig{ + Epoch: 900, + Gap: 450, + V2: ¶ms.V2{ + SwitchBlock: big.NewInt(450), + }, + }, + db: db, + snapshots: lru.NewCache[common.Hash, *SnapshotV2](utils.InMemorySnapshots), + failedRebuilds: lru.NewCache[common.Hash, struct{}](utils.InMemorySnapshots), + } +} + +type countingSnapshotChainReader struct { + *snapshotChainReader + hold chan struct{} // when non-nil, StateAt blocks until it is closed + stateAtCalls int32 +} + +func (m *countingSnapshotChainReader) StateAt(root common.Hash) (*state.StateDB, error) { + atomic.AddInt32(&m.stateAtCalls, 1) + if m.hold != nil { + <-m.hold + } + return m.snapshotChainReader.StateAt(root) +} + +// gatedSnapshotDB signals every database read, which lets a test wait until all +// concurrent getSnapshot callers are past the on-disk snapshot lookup. +type gatedSnapshotDB struct { + ethdb.Database + reads chan struct{} +} + +func (db *gatedSnapshotDB) Get(key []byte) ([]byte, error) { + select { + case db.reads <- struct{}{}: + default: + } + return db.Database.Get(key) +} + +func mustBuildStateWithCandidates(t *testing.T, candidates []common.Address, stakes map[common.Address]*big.Int) (common.Hash, *state.StateDB) { + t.Helper() + + stateDisk := rawdb.NewMemoryDatabase() + stateDB, err := state.New(types.EmptyRootHash, state.NewDatabase(stateDisk)) + assert.NoError(t, err) + + slotCandidates := common.BigToHash(new(big.Int).SetUint64(8)) + stateDB.SetState(common.MasternodeVotingSMCBinary, slotCandidates, common.BigToHash(new(big.Int).SetUint64(uint64(len(candidates))))) + + for i, addr := range candidates { + stateDB.SetState(common.MasternodeVotingSMCBinary, state.GetLocDynamicArrAtElement(slotCandidates, uint64(i), 1), common.BytesToHash(addr.Bytes())) + + locValidator := state.GetLocMappingAtKey(addr.Hash(), 1) + capSlot := common.BigToHash(new(big.Int).Add(locValidator, big.NewInt(1))) + stateDB.SetState(common.MasternodeVotingSMCBinary, capSlot, common.BigToHash(stakes[addr])) + } + + root, err := stateDB.Commit(1, false) + assert.NoError(t, err) + assert.NoError(t, stateDB.Database().TrieDB().Commit(root, false)) + + committedState, err := state.New(root, state.NewDatabase(stateDisk)) + assert.NoError(t, err) + return root, committedState +} + +func TestRebuildSnapshot_ReconstructsAndPersists(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + lowStake := common.Address{0x01} + highStake := common.Address{0x02} + candidates := []common.Address{lowStake, highStake} + stakes := map[common.Address]*big.Int{ + lowStake: big.NewInt(100), + highStake: big.NewInt(200), + } + root, committedState := mustBuildStateWithCandidates(t, candidates, stakes) + + reader := newSnapshotChainReader() + reader.statesByRoot[root] = committedState + + gapHeader := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + snap, err := engine.rebuildSnapshot(reader, gapHeader) + assert.NoError(t, err) + assert.NotNil(t, snap) + assert.Equal(t, uint64(testGapNumber), snap.Number) + assert.Equal(t, gapHeader.Hash(), snap.Hash) + assert.Equal(t, []common.Address{highStake, lowStake}, snap.NextEpochCandidates) + + persisted, err := loadSnapshot(engine.db, gapHeader.Hash()) + assert.NoError(t, err) + assert.Equal(t, snap.NextEpochCandidates, persisted.NextEpochCandidates) + + cached, ok := engine.snapshots.Get(gapHeader.Hash()) + assert.True(t, ok) + assert.Equal(t, snap.NextEpochCandidates, cached.NextEpochCandidates) +} + +func TestGetSnapshot_RebuildsWhenSnapshotMissing(t *testing.T) { + dir := t.TempDir() + disk, err := leveldb.New(dir, 256, 0, "", false) + assert.NoError(t, err) + + engine := newTestEngineV2(rawdb.NewDatabase(disk)) + + addrA := common.Address{0x0a} + addrB := common.Address{0x0b} + candidates := []common.Address{addrA, addrB} + stakes := map[common.Address]*big.Int{ + addrA: big.NewInt(15), + addrB: big.NewInt(30), + } + root, committedState := mustBuildStateWithCandidates(t, candidates, stakes) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + reader := newSnapshotChainReader() + reader.addHeader(header) + reader.statesByRoot[root] = committedState + + _, err = loadSnapshot(engine.db, header.Hash()) + assert.Error(t, err) + + snap, err := engine.getSnapshot(reader, testGapNumber, true) + assert.NoError(t, err) + assert.NotNil(t, snap) + assert.Equal(t, header.Hash(), snap.Hash) + assert.Equal(t, uint64(testGapNumber), snap.Number) + assert.Equal(t, []common.Address{addrB, addrA}, snap.NextEpochCandidates) + + persisted, err := loadSnapshot(engine.db, header.Hash()) + assert.NoError(t, err) + assert.Equal(t, snap.NextEpochCandidates, persisted.NextEpochCandidates) +} + +func TestGetSnapshot_SingleflightDeduplicatesRebuild(t *testing.T) { + const workers = 12 + + gated := &gatedSnapshotDB{Database: rawdb.NewMemoryDatabase(), reads: make(chan struct{}, workers)} + engine := newTestEngineV2(gated) + + addrA := common.Address{0x11} + addrB := common.Address{0x22} + candidates := []common.Address{addrA, addrB} + stakes := map[common.Address]*big.Int{ + addrA: big.NewInt(40), + addrB: big.NewInt(80), + } + root, committedState := mustBuildStateWithCandidates(t, candidates, stakes) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + + reader := &countingSnapshotChainReader{ + snapshotChainReader: baseReader, + hold: make(chan struct{}), + } + + results := make(chan *SnapshotV2, workers) + errorsCh := make(chan error, workers) + + var wg sync.WaitGroup + wg.Add(workers) + for range workers { + go func() { + defer wg.Done() + snap, err := engine.getSnapshot(reader, testGapNumber, true) + if err != nil { + errorsCh <- err + return + } + results <- snap + }() + } + + // Wait until every worker has missed the on-disk snapshot, so none of them + // can short-circuit on the cache once the rebuild completes. + for range workers { + <-gated.reads + } + close(reader.hold) + + wg.Wait() + close(results) + close(errorsCh) + + for err := range errorsCh { + assert.NoError(t, err) + } + + for snap := range results { + assert.NotNil(t, snap) + assert.Equal(t, header.Hash(), snap.Hash) + assert.Equal(t, uint64(testGapNumber), snap.Number) + assert.Equal(t, []common.Address{addrB, addrA}, snap.NextEpochCandidates) + } + + assert.Equal(t, int32(1), atomic.LoadInt32(&reader.stateAtCalls), "singleflight should deduplicate concurrent rebuilds") +} + +func TestGetSnapshot_SkipsRebuildAtOrBeforeSwitchBlock(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + // SwitchBlock is 450 in the test config, and 450 is a gap number: its + // snapshot comes from Initial(), never from the state trie. + header := &types.Header{Number: new(big.Int).SetUint64(450), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + snap, err := engine.getSnapshot(reader, 450, true) + assert.Error(t, err) + assert.Nil(t, snap) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) +} + +// TestRebuildSnapshot_EqualStakeOrdering locks in the order that equal-stake +// candidates get: the comparator is not a strict weak ordering, so a different +// sort implementation would produce a different masternode set and fork. +func TestRebuildSnapshot_EqualStakeOrdering(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + addrA := common.Address{0x01} + addrB := common.Address{0x02} + addrC := common.Address{0x03} + candidates := []common.Address{addrA, addrB, addrC} + stakes := map[common.Address]*big.Int{ + addrA: big.NewInt(100), + addrB: big.NewInt(100), + addrC: big.NewInt(100), + } + root, committedState := mustBuildStateWithCandidates(t, candidates, stakes) + + reader := newSnapshotChainReader() + reader.statesByRoot[root] = committedState + reader.setHead(testGapNumber) + + snap, err := engine.rebuildSnapshot(reader, &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root}) + assert.NoError(t, err) + assert.Equal(t, []common.Address{addrC, addrB, addrA}, snap.NextEpochCandidates) +} + +func TestGetSnapshot_SkipsRebuildFarBehindHead(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + // More than snapshotRebuildMaxEpochsBehind epochs past the gap block: an + // unauthenticated message must not be able to walk old state tries. + baseReader.setHead(testGapNumber + snapshotRebuildMaxEpochsBehind*engine.config.Epoch + 1) + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + snap, err := engine.getSnapshot(reader, testGapNumber, true) + assert.Error(t, err) + assert.Nil(t, snap) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) +} + +func TestGetSnapshot_SkipsRebuildForNonGapNumber(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + // A vote/timeout message can carry any gap number, so a number that is not a + // gap block must never reach the state trie or persist a snapshot. + const nonGapNumber = testGapNumber + 1 + header := &types.Header{Number: new(big.Int).SetUint64(nonGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + snap, err := engine.getSnapshot(reader, nonGapNumber, true) + assert.Error(t, err) + assert.Nil(t, snap) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) + + _, err = loadSnapshot(engine.db, header.Hash()) + assert.Error(t, err) +} + +func TestGetSnapshot_SkipsRebuildWhenChainCannotReadState(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + + snap, err := engine.getSnapshot(&headerOnlyChainReader{inner: baseReader}, testGapNumber, true) + assert.Error(t, err) + assert.Nil(t, snap) + + _, err = loadSnapshot(engine.db, header.Hash()) + assert.Error(t, err) +} + +func TestGetSnapshot_SkipsRebuildWhenBlockNotImported(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + // Canonical header only: the block body has not been imported yet, so the + // missing state is transient and must not be recorded as a failed rebuild. + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeaderOnly(header) + baseReader.statesByRoot[root] = committedState + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + snap, err := engine.getSnapshot(reader, testGapNumber, true) + assert.Error(t, err) + assert.Nil(t, snap) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) + + _, failed := engine.failedRebuilds.Get(header.Hash()) + assert.False(t, failed) + + // Once the block lands, the rebuild runs. + baseReader.addHeader(header) + snap, err = engine.getSnapshot(reader, testGapNumber, true) + assert.NoError(t, err) + assert.Equal(t, []common.Address{{0x01}}, snap.NextEpochCandidates) +} + +func TestGetSnapshot_GivesUpAfterFailedRebuild(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + // No state registered for the header root: the gap block is imported but its + // trie is pruned, so the rebuild can never succeed. + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: common.Hash{0xaa}} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + _, err := engine.getSnapshot(reader, testGapNumber, true) + assert.Error(t, err) + assert.Equal(t, int32(1), atomic.LoadInt32(&reader.stateAtCalls)) + + _, err = engine.getSnapshot(reader, testGapNumber, true) + assert.Error(t, err) + assert.Equal(t, int32(1), atomic.LoadInt32(&reader.stateAtCalls), "failed rebuild should not be retried") + + _, err = loadSnapshot(engine.db, header.Hash()) + assert.Error(t, err) +} + +func TestRebuildSnapshot_NoCandidatesDoesNotPersist(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, nil, nil) + + reader := newSnapshotChainReader() + reader.statesByRoot[root] = committedState + + gapHeader := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + snap, err := engine.rebuildSnapshot(reader, gapHeader) + assert.Error(t, err) + assert.Nil(t, snap) + + _, err = loadSnapshot(engine.db, gapHeader.Hash()) + assert.Error(t, err) +} + +func TestGetSnapshot_SkipsRebuildWhenSnapshotExists(t *testing.T) { + engine := newTestEngineV2(rawdb.NewMemoryDatabase()) + + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + // A snapshot that exists but cannot be decoded must not be replaced by a + // state-derived rebuild: the stored masternode set may be the correct one. + stored := []byte("not a snapshot") + assert.NoError(t, rawdb.WriteXdposV2Snapshot(engine.db, header.Hash(), stored)) + + snap, err := engine.getSnapshot(reader, testGapNumber, true) + assert.Error(t, err) + assert.Nil(t, snap) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) + + blob, err := rawdb.ReadXdposV2Snapshot(engine.db, header.Hash()) + assert.NoError(t, err) + assert.Equal(t, stored, blob) +} + +// raceSnapshotDB persists a snapshot right after the first failed read, which +// reproduces another caller storing it in the window between the getSnapshot +// lookup and the self-heal path. +type raceSnapshotDB struct { + ethdb.Database + once sync.Once + snap *SnapshotV2 +} + +func (db *raceSnapshotDB) Get(key []byte) ([]byte, error) { + data, err := db.Database.Get(key) + if err != nil { + db.once.Do(func() { + if storeErr := StoreSnapshot(db.snap, db.Database); storeErr != nil { + panic(storeErr) + } + }) + } + return data, err +} + +func TestGetSnapshot_ReturnsSnapshotStoredWhileHealing(t *testing.T) { + root, committedState := mustBuildStateWithCandidates(t, []common.Address{{0x01}}, map[common.Address]*big.Int{{0x01}: big.NewInt(10)}) + + header := &types.Header{Number: new(big.Int).SetUint64(testGapNumber), Root: root} + // Deliberately different from what the state would yield, so the assertions + // tell a reload apart from a rebuild. + stored := NewSnapshot(testGapNumber, header.Hash(), []common.Address{{0xfe}}) + + engine := newTestEngineV2(&raceSnapshotDB{Database: rawdb.NewMemoryDatabase(), snap: stored}) + + baseReader := newSnapshotChainReader() + baseReader.addHeader(header) + baseReader.statesByRoot[root] = committedState + reader := &countingSnapshotChainReader{snapshotChainReader: baseReader} + + snap, err := engine.getSnapshot(reader, testGapNumber, true) + assert.NoError(t, err) + assert.Equal(t, stored.NextEpochCandidates, snap.NextEpochCandidates) + assert.Zero(t, atomic.LoadInt32(&reader.stateAtCalls)) +} diff --git a/core/blockchain.go b/core/blockchain.go index daffceb42984..860be78f772a 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -2719,6 +2719,7 @@ func (bc *BlockChain) GetClient() (bind.ContractBackend, error) { return bc.Client, nil } +// NOTE: keep code sync with engine_v2.BuildSnapshotFromState func (bc *BlockChain) UpdateM1() error { engine, ok := bc.Engine().(*XDPoS.XDPoS) if bc.Config().XDPoS == nil || !ok { diff --git a/core/blockchain_reader.go b/core/blockchain_reader.go index 7bb7a4825803..f9dd4f098ba8 100644 --- a/core/blockchain_reader.go +++ b/core/blockchain_reader.go @@ -21,6 +21,7 @@ import ( "github.com/XinFinOrg/XDPoSChain/common" "github.com/XinFinOrg/XDPoSChain/consensus" + "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/engines/engine_v2" "github.com/XinFinOrg/XDPoSChain/core/rawdb" "github.com/XinFinOrg/XDPoSChain/core/state" "github.com/XinFinOrg/XDPoSChain/core/types" @@ -287,6 +288,10 @@ func (bc *BlockChain) State() (*state.StateDB, error) { return bc.StateAt(bc.CurrentBlock().Root) } +// engine_v2 type-asserts a chain reader against this to rebuild missing gap +// snapshots, so StateAt must keep the signature. +var _ engine_v2.GapStateReader = (*BlockChain)(nil) + // StateAt returns a new mutable state based on a particular point in time. func (bc *BlockChain) StateAt(root common.Hash) (*state.StateDB, error) { statedb, err := state.NewWithChainConfig(root, bc.stateCache, bc.chainConfig) diff --git a/core/rawdb/accessors_xdc.go b/core/rawdb/accessors_xdc.go index 85718d666d5b..98f192f4a380 100644 --- a/core/rawdb/accessors_xdc.go +++ b/core/rawdb/accessors_xdc.go @@ -50,6 +50,11 @@ func ReadXdposV2Snapshot(db ethdb.KeyValueReader, hash common.Hash) ([]byte, err return data, nil } +// HasXdposV2Snapshot reports whether a snapshot is stored for the given hash. +func HasXdposV2Snapshot(db ethdb.KeyValueReader, hash common.Hash) (bool, error) { + return db.Has(xdposV2Key(hash)) +} + // WriteXdposV2Snapshot writes the SnapshotV2 into the database. func WriteXdposV2Snapshot(db ethdb.KeyValueWriter, hash common.Hash, blob []byte) error { if err := db.Put(xdposV2Key(hash), blob); err != nil { diff --git a/eth/backend.go b/eth/backend.go index bc4b0e95dab3..5151c508c3b5 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -383,6 +383,7 @@ func New(stack *node.Node, config *ethconfig.Config, XDCXServ *XDCx.XDCX, lendin */ hooks.AttachConsensusV1Hooks(c, eth.blockchain, chainConfig) hooks.AttachConsensusV2Hooks(c, eth.blockchain, chainConfig) + c.EngineV2.HookSyncing = eth.protocolManager.downloader.Synchronising isSigner := func(address common.Address) bool { return c.IsAuthorisedAddress(eth.blockchain, eth.blockchain.CurrentHeader(), address) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index e901d4439632..56df1f87d288 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -27,9 +27,7 @@ import ( "github.com/XinFinOrg/XDPoSChain" "github.com/XinFinOrg/XDPoSChain/common" - xdc_sort "github.com/XinFinOrg/XDPoSChain/common/sort" "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/engines/engine_v2" - "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils" "github.com/XinFinOrg/XDPoSChain/core/rawdb" "github.com/XinFinOrg/XDPoSChain/core/state" "github.com/XinFinOrg/XDPoSChain/core/types" @@ -1998,36 +1996,17 @@ func (d *Downloader) requestTTL() time.Duration { return ttl } -// generateSnapshot creates and stores a snapshot from the given state and block hash. -// It retrieves candidates from state, sorts them by stake in descending order, -// and stores the snapshot for future use. +// generateSnapshot creates and stores a gap block snapshot derived from the +// given committed state. Callers log the failure. func (d *Downloader) generateSnapshot(statedb *state.StateDB, number uint64, hash common.Hash) error { - candidates := statedb.GetCandidates() - var ms []utils.Masternode - for _, candidate := range candidates { - v := statedb.GetCandidateCap(candidate) - // Skip zero address candidates - if !candidate.IsZero() { - ms = append(ms, utils.Masternode{Address: candidate, Stake: v}) - } - } - xdc_sort.Slice(ms, func(i, j int) bool { - return ms[i].Stake.Cmp(ms[j].Stake) >= 0 - }) - - masterNodes := []common.Address{} - for _, m := range ms { - masterNodes = append(masterNodes, m.Address) - } - - snap := engine_v2.NewSnapshot(number, hash, masterNodes) - log.Info("[generateSnapshot] created snapshot", "number", number, "hash", hash.Hex(), "candidates", len(masterNodes)) - - err := engine_v2.StoreSnapshot(snap, d.stateDB) + snap, err := engine_v2.BuildSnapshotFromState(statedb, number, hash) if err != nil { - log.Error("[generateSnapshot] error while storing snapshot", "hash", hash, "error", err) return err } + if err := engine_v2.StoreSnapshot(snap, d.stateDB); err != nil { + return err + } + log.Info("[generateSnapshot] created snapshot", "number", number, "hash", hash.Hex(), "candidates", len(snap.NextEpochCandidates)) return nil } diff --git a/eth/downloader/testchain_test.go b/eth/downloader/testchain_test.go index e3c3dee4fe88..dc89a1130b57 100644 --- a/eth/downloader/testchain_test.go +++ b/eth/downloader/testchain_test.go @@ -26,6 +26,7 @@ import ( "github.com/XinFinOrg/XDPoSChain/consensus/ethash" "github.com/XinFinOrg/XDPoSChain/core" "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/state" "github.com/XinFinOrg/XDPoSChain/core/types" "github.com/XinFinOrg/XDPoSChain/crypto" "github.com/XinFinOrg/XDPoSChain/params" @@ -45,13 +46,30 @@ var ( }() testGspec = &core.Genesis{ - Alloc: types.GenesisAlloc{testAddress: {Balance: big.NewInt(1000000000000000000)}}, + Alloc: types.GenesisAlloc{ + testAddress: {Balance: big.NewInt(1000000000000000000)}, + common.MasternodeVotingSMCBinary: {Balance: new(big.Int), Storage: testMasternodeVotingStorage()}, + }, BaseFee: big.NewInt(params.InitialBaseFee), Config: testChainConfig, } testGenesis = testGspec.MustCommit(testDB) ) +// testMasternodeVotingStorage lays out the minimum of the masternode voting +// contract that StateDB.GetCandidates/GetCandidateCap read, so gap block +// snapshots derived from state are not empty. +func testMasternodeVotingStorage() map[common.Hash]common.Hash { + candidate := common.HexToAddress("0x1234000000000000000000000000000000000000") + candidatesSlot := common.BigToHash(big.NewInt(8)) + capSlot := common.BigToHash(new(big.Int).Add(state.GetLocMappingAtKey(candidate.Hash(), 1), big.NewInt(1))) + return map[common.Hash]common.Hash{ + candidatesSlot: common.BigToHash(big.NewInt(1)), + state.GetLocDynamicArrAtElement(candidatesSlot, 0, 1): candidate.Hash(), + capSlot: common.BigToHash(big.NewInt(1000)), + } +} + // The common prefix of all test chains: var testChainBase *testChain