diff --git a/consensus/XDPoS/engines/engine_v2/engine.go b/consensus/XDPoS/engines/engine_v2/engine.go index c64f493a801d..11ff45565611 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{}] @@ -119,6 +122,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 9e27523d5da4..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,7 +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 { - 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 fb002018b49f..78d26e95ed35 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -2758,6 +2758,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 2484bda6905e..cb403832de0c 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/downloader/downloader.go b/eth/downloader/downloader.go index 538637624156..2ab56ea21788 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