forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

beacon/light: rollback slot-by-slot instead of in a single batch

Browse source
This commit is contained in:
Zsolt Felfoldi 2023-09-08 13:30:03 +02:00 committed by zsfelfoldi
parent b039c2c1c0
commit 2776aefab0
1 changed files with 34 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
beacon/light/committee_chain.go
View file

@ -63,10 +63,10 @@ var (
// Once synced to the current sync period, CommitteeChain can also validate // Once synced to the current sync period, CommitteeChain can also validate
// signed beacon headers. // signed beacon headers.
type CommitteeChain struct { type CommitteeChain struct {
chainmu sync.RWMutex // chainmu guards against concurrent access to the canonicalStore structures
// Note: chainmu avoids concurrent access to the canonicalStore structures
// (updates, committees, fixedRoots) and ensures that they stay consistent // (updates, committees, fixedRoots) and ensures that they stay consistent
// with each other and with committeeCache. // with each other and with committeeCache.
chainmu sync.RWMutex
db ethdb.KeyValueStore db ethdb.KeyValueStore
updates *canonicalStore[*types.LightClientUpdate] updates *canonicalStore[*types.LightClientUpdate]
committees *canonicalStore[*types.SerializedSyncCommittee] committees *canonicalStore[*types.SerializedSyncCommittee]
@ -140,7 +140,6 @@ func NewCommitteeChain(db ethdb.KeyValueStore, config *types.ChainConfig, signer
s.Reset() s.Reset()
} }
// roll back invalid updates (might be necessary if forks have been changed since last time) // roll back invalid updates (might be necessary if forks have been changed since last time)
batch := s.db.NewBatch()
for !s.updates.periods.IsEmpty() { for !s.updates.periods.IsEmpty() {
if update, ok := s.updates.get(s.updates.periods.AfterLast - 1); !ok || s.verifyUpdate(update) { if update, ok := s.updates.get(s.updates.periods.AfterLast - 1); !ok || s.verifyUpdate(update) {
if update == nil { if update == nil {
@ -148,10 +147,9 @@ func NewCommitteeChain(db ethdb.KeyValueStore, config *types.ChainConfig, signer
} }
break break
} }
s.rollback(batch, s.updates.periods.AfterLast) if err := s.rollback(s.updates.periods.AfterLast); err != nil {
} log.Error("Error writing batch into chain database", "error", err)
if err := batch.Write(); err != nil { }
log.Error("Error writing batch into chain database", "error", err)
} }
if !s.committees.periods.IsEmpty() { if !s.committees.periods.IsEmpty() {
log.Trace("Sync committee chain loaded", "first period", s.committees.periods.First, "last period", s.committees.periods.AfterLast-1) log.Trace("Sync committee chain loaded", "first period", s.committees.periods.First, "last period", s.committees.periods.AfterLast-1)
@ -192,9 +190,7 @@ func (s *CommitteeChain) Reset() {
s.chainmu.Lock() s.chainmu.Lock()
defer s.chainmu.Unlock() defer s.chainmu.Unlock()
batch := s.db.NewBatch() if err := s.rollback(0); err != nil {
s.rollback(batch, 0)
if err := batch.Write(); err != nil {
log.Error("Error writing batch into chain database", "error", err) log.Error("Error writing batch into chain database", "error", err)
} }
} }
@ -232,7 +228,9 @@ func (s *CommitteeChain) AddFixedRoot(period uint64, root common.Hash) error {
} }
if oldRoot != (common.Hash{}) && (oldRoot != root) { if oldRoot != (common.Hash{}) && (oldRoot != root) {
// existing old root was different, we have to reorg the chain // existing old root was different, we have to reorg the chain
s.rollback(batch, period) if err := s.rollback(period); err != nil {
return err
}
} }
if err := s.fixedRoots.add(batch, period, root); err != nil { if err := s.fixedRoots.add(batch, period, root); err != nil {
return err return err
@ -363,10 +361,12 @@ func (s *CommitteeChain) InsertUpdate(update *types.LightClientUpdate, nextCommi
return ErrWrongCommitteeRoot return ErrWrongCommitteeRoot
} }
} }
batch := s.db.NewBatch()
if reorg { if reorg {
s.rollback(batch, period+1) if err := s.rollback(period + 1); err != nil {
return err
}
} }
batch := s.db.NewBatch()
if addCommittee { if addCommittee {
if err := s.committees.add(batch, period+1, nextCommittee); err != nil { if err := s.committees.add(batch, period+1, nextCommittee); err != nil {
return err return err
@ -401,13 +401,28 @@ func (s *CommitteeChain) NextSyncPeriod() (uint64, bool) {
// rollback removes all committees and fixed roots from the given period and updates // rollback removes all committees and fixed roots from the given period and updates
// starting from the previous period. // starting from the previous period.
func (s *CommitteeChain) rollback(batch ethdb.Batch, period uint64) { func (s *CommitteeChain) rollback(period uint64) error {
s.deleteCommitteesFrom(batch, period) max := s.updates.periods.AfterLast + 1
s.fixedRoots.deleteFrom(batch, period) if s.committees.periods.AfterLast > max {
if period > 0 { max = s.committees.periods.AfterLast
period--
} }
s.updates.deleteFrom(batch, period) if s.fixedRoots.periods.AfterLast > max {
max = s.fixedRoots.periods.AfterLast
}
for max > period {
max--
batch := s.db.NewBatch()
s.deleteCommitteesFrom(batch, max)
s.fixedRoots.deleteFrom(batch, max)
if max > 0 {
s.updates.deleteFrom(batch, max-1)
}
if err := batch.Write(); err != nil {
log.Error("Error writing batch into chain database", "error", err)
return err
}
}
return nil
} }
// getCommitteeRoot returns the committee root at the given period, either fixed, // getCommitteeRoot returns the committee root at the given period, either fixed,