Gossip DB fault tolerance

cmd/db_fault_tolerance.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"bytes"
+	"errors"
+	"github.com/Fantom-foundation/go-lachesis/hash"
+	"github.com/Fantom-foundation/go-lachesis/kvdb"
+	"github.com/Fantom-foundation/go-lachesis/kvdb/flushable"
+	"time"
+)
+
+type RegisteredDbs struct {
+	wrappers map[string]kvdb.FlushableKeyValueStore
+	bareDbs map[string]kvdb.KeyValueStore
+
+	queuedDrops map[string]struct{}
+
+	prevFlushTime time.Time
+}
+
+func (dbs *RegisteredDbs) Drop(name string) {
+	dbs.queuedDrops[name] = struct{}{}
+}
+
+func (dbs *RegisteredDbs) Register(name string, db kvdb.KeyValueStore) {
+	wrapper := flushable.New(db)
+	wrapper.SetDropper(func() {
+		dbs.Drop(name)
+	})
+
+	dbs.bareDbs[name] = db
+	dbs.wrappers[name] = wrapper
+	delete(dbs.queuedDrops, name)
+}
+
+func (dbs *RegisteredDbs) Flush(id hash.Event) error {
+	key := []byte("mark")
+
+	// dirty flag
+	for _, db := range dbs.bareDbs {
+		marker := bytes.NewBuffer(nil)
+		prev, err := db.Get(key)
+		if err != nil {
+			return err
+		}
+		if prev == nil {
+			return errors.New("not found prev flushed state marker")
+		}
+
+		marker.Write([]byte("dirty"))
+		marker.Write(prev)
+		marker.Write([]byte("->"))
+		marker.Write(id.Bytes())
+		err = db.Put(key, marker.Bytes())
+		if err != nil {
+			return err
+		}
+	}
+	// flush (along with clean flag)
+	for _, db := range dbs.wrappers {
+		db.Flush() // flush data
+		err := db.Put(key, id.Bytes())
+		if err != nil {
+			return err
+		}
+		db.Flush() // flush clean flag
+	}
+
+	// drop old DBs
+	for name := range dbs.queuedDrops {
+		db := dbs.bareDbs[name]
+		if db == nil {
+			// ???? we should register all the DBs at startup
+			continue
+		}
+		db.Drop()
+	}
+
+	dbs.prevFlushTime = time.Now()
+	return nil
+}
+
+func (dbs *RegisteredDbs) FlushIfNeeded(id hash.Event) bool {
+	if time.Since(dbs.prevFlushTime) > 10 * time.Minute {
+		dbs.Flush(id)
+		return true
+	}
+
+	totalNotFlushed := 0
+	for _, db := range dbs.wrappers {
+		totalNotFlushed += db.NotFlushedSizeEst()
+	}
+
+	if totalNotFlushed > 100 * 1024 * 1024 {
+		dbs.Flush(id)
+		return true
+	}
+	return false
+}
+
+// call on startup, after all dbs are registered
+func (dbs *RegisteredDbs) CheckDbsSynced() error {
+	key := []byte("mark")
+	var prevId *hash.Event
+	for _, db := range dbs.bareDbs {
+		mark, err := db.Get(key)
+		if err != nil {
+			return err
+		}
+		if bytes.HasPrefix(mark, []byte("dirty")) {
+			return errors.New("dirty")
+		}
+		eventId := hash.BytesToEvent(mark)
+		if prevId == nil {
+			prevId  = &eventId
+		}
+		if eventId != *prevId {
+			return errors.New("not synced")
+		}
+	}
+	return nil
+}

evm_core/apply_genesis.go

@@ -75,6 +75,10 @@ func ApplyGenesis(db ethdb.Database, net *lachesis.Config) (*EvmBlock, error) {
 	}
 
 	writeBlockIndexes(db, blockNum, block.Hash)
+	err = statedb.Database().TrieDB().Cap(0)
+	if err != nil {
+		return nil, err
+	}
 
 	return block, nil
 }

gossip/apply_block.go

@@ -12,8 +12,15 @@ import (
 	"github.com/Fantom-foundation/go-lachesis/inter/pos"
 )
 
-// ApplyBlock execs ordered txns on state.
-func (s *Service) ApplyBlock(block *inter.Block, stateHash common.Hash, validators pos.Validators) (newStateHash common.Hash, newValidators pos.Validators) {
+// onNewBlock execs ordered txns of new block on state.
+func (s *Service) onNewBlock(
+	block *inter.Block,
+	stateHash common.Hash,
+	validators pos.Validators,
+) (
+	newStateHash common.Hash,
+	newValidators pos.Validators,
+) {
 	evmProcessor := evm_core.NewStateProcessor(params.AllEthashProtocolChanges, s.GetEvmStateReader())
 
 	// Assemble block data

gossip/handler.go

@@ -177,6 +177,9 @@ func (pm *ProtocolManager) makeFetcher() (*fetcher.Fetcher, *ordering.EventBuffe
 			if err != nil {
 				return err
 			}
+
+			pm.store.Commit(e.Epoch)
+
 			// If the event is indeed in our own graph, announce it
 			if atomic.LoadUint32(&pm.synced) != 0 { // announce only fresh events
 				pm.BroadcastEvent(e, false)

gossip/service.go

@@ -108,7 +108,7 @@ func NewService(ctx *node.ServiceContext, config Config, store *Store, engine Co
 		engine:       engine,
 		processEvent: svc.processEvent,
 	}
-	svc.engine.Bootstrap(svc.ApplyBlock)
+	svc.engine.Bootstrap(svc.onNewBlock)
 
 	trustedNodes := []string{}
 	svc.serverPool = newServerPool(store.table.Peers, svc.done, &svc.wg, trustedNodes)

gossip/store.go

@@ -7,7 +7,9 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 
+	"github.com/Fantom-foundation/go-lachesis/inter/idx"
 	"github.com/Fantom-foundation/go-lachesis/kvdb"
+	"github.com/Fantom-foundation/go-lachesis/kvdb/flushable"
 	"github.com/Fantom-foundation/go-lachesis/kvdb/memorydb"
 	"github.com/Fantom-foundation/go-lachesis/kvdb/no_key_is_err"
 	"github.com/Fantom-foundation/go-lachesis/kvdb/table"
@@ -16,9 +18,10 @@ import (
 
 // Store is a node persistent storage working over physical key-value database.
 type Store struct {
-	persistentDB kvdb.KeyValueStore
+	bareDb kvdb.KeyValueStore
 
-	table struct {
+	mainDb *flushable.Flushable
+	table  struct {
 		Peers     kvdb.KeyValueStore `table:"peer_"`
 		Events    kvdb.KeyValueStore `table:"event_"`
 		Blocks    kvdb.KeyValueStore `table:"block_"`
@@ -47,14 +50,19 @@ type Store struct {
 // NewStore creates store over key-value db.
 func NewStore(db kvdb.KeyValueStore, makeDb func(name string) kvdb.KeyValueStore) *Store {
 	s := &Store{
-		persistentDB: db,
-		makeDb:       makeDb,
-		Instance:     logger.MakeInstance(),
+		bareDb:   db,
+		mainDb:   flushable.New(db),
+		makeDb:   makeDb,
+		Instance: logger.MakeInstance(),
 	}
 
-	table.MigrateTables(&s.table, s.persistentDB)
+	table.MigrateTables(&s.table, s.mainDb)
 
-	evmTable := no_key_is_err.Wrap(table.New(s.persistentDB, []byte("evm_"))) // ETH expects that "not found" is an error
+	if s.isDirty() {
+		s.Log.Crit("Service DB is possible inconsistent. Recreate it.")
+	}
+
+	evmTable := no_key_is_err.Wrap(table.New(s.mainDb, []byte("evm_"))) // ETH expects that "not found" is an error
 	s.table.Evm = rawdb.NewDatabase(evmTable)
 	s.table.EvmState = state.NewDatabase(s.table.Evm)
 
@@ -74,7 +82,48 @@ func NewMemStore() *Store {
 // Close leaves underlying database.
 func (s *Store) Close() {
 	table.MigrateTables(&s.table, nil)
-	s.persistentDB.Close()
+	s.mainDb.Close()
+}
+
+// Commit changes.
+func (s *Store) Commit(epoch idx.Epoch) {
+	s.setDirty(true)
+	defer s.setDirty(false)
+
+	err := s.commitEpochStore(epoch)
+	if err != nil {
+		s.Log.Crit("epoch DB commit", "err", err)
+	}
+
+	err = s.mainDb.Flush()
+	if err != nil {
+		s.Log.Crit("main DB commit", "err", err)
+	}
+}
+
+// setDirty sets dirty flag.
+func (s *Store) setDirty(flag bool) {
+	key := []byte("is_dirty")
+	val := make([]byte, 1, 1)
+	if flag {
+		val[0] = 1
+	}
+
+	err := s.bareDb.Put(key, val)
+	if err != nil {
+		s.Log.Crit("Failed to put key-value", "err", err)
+	}
+}
+
+// isDirty gets dirty flag.
+func (s *Store) isDirty() bool {
+	key := []byte("is_dirty")
+	val, err := s.bareDb.Get(key)
+	if err != nil {
+		s.Log.Crit("Failed to get value", "err", err)
+	}
+
+	return len(val) > 1 && val[0] != 0
 }
 
 // StateDB returns state database.

gossip/store_epoch.go

@@ -32,6 +32,11 @@ func (s *Store) getEpochStore(epoch idx.Epoch) *epochStore {
 	return tables.(*epochStore)
 }
 
+// commitEpochStore is not safe for concurrent use.
+func (s *Store) commitEpochStore(epoch idx.Epoch) error {
+	return s.commitTmpDb("epoch", uint64(epoch))
+}
+
 // delEpochStore is not safe for concurrent use.
 func (s *Store) delEpochStore(epoch idx.Epoch) {
 	s.delTmpDb("epoch", uint64(epoch))

gossip/store_tmp.go

@@ -6,11 +6,12 @@ import (
 
 	"github.com/Fantom-foundation/go-lachesis/common/bigendian"
 	"github.com/Fantom-foundation/go-lachesis/kvdb"
+	"github.com/Fantom-foundation/go-lachesis/kvdb/flushable"
 )
 
 type (
 	tmpDb struct {
-		Db     kvdb.KeyValueStore
+		Db     *flushable.Flushable
 		Tables interface{}
 	}
 
@@ -23,6 +24,9 @@ type (
 )
 
 func (s *Store) initTmpDbs() {
+	s.tmpDbs.Lock()
+	defer s.tmpDbs.Unlock()
+
 	s.tmpDbs.min = make(map[string]uint64)
 	s.tmpDbs.seq = make(map[string]map[uint64]tmpDb)
 
@@ -61,7 +65,7 @@ func (s *Store) getTmpDb(name string, ver uint64, makeTables func(kvdb.KeyValueS
 		return tmp.Tables
 	}
 
-	db := s.makeDb(tmpDbName(name, ver))
+	db := flushable.New(s.makeDb(tmpDbName(name, ver)))
 	tables := makeTables(db)
 	s.tmpDbs.seq[name][ver] = tmpDb{
 		Db:     db,
@@ -71,6 +75,19 @@ func (s *Store) getTmpDb(name string, ver uint64, makeTables func(kvdb.KeyValueS
 	return tables
 }
 
+func (s *Store) commitTmpDb(name string, ver uint64) error {
+	s.tmpDbs.Lock()
+	defer s.tmpDbs.Unlock()
+
+	min, ok := s.tmpDbs.min[name]
+	if !ok || ver < min {
+		return nil
+	}
+
+	tmp := s.tmpDbs.seq[name][ver]
+	return tmp.Db.Flush()
+}
+
 func (s *Store) delTmpDb(name string, ver uint64) {
 	s.tmpDbs.Lock()
 	defer s.tmpDbs.Unlock()

kvdb/flushable/flushable.go

@@ -26,6 +26,8 @@ type Flushable struct {
 	sizeEstimation *int
 
 	lock *sync.Mutex // we have no guarantees that rbt.Tree works with concurrent reads, so we can't use MutexRW
+
+	dropper func()
 }
 
 // New Flushable wraps underlying DB. All the writes into the cache won't be written in parent until .Flush() is called.
@@ -38,6 +40,14 @@ func New(parent kvdb.KeyValueStore) *Flushable {
 	}
 }
 
+func (w *Flushable) SetDropper(dropper func()) {
+	w.dropper = dropper
+}
+
+func (w *Flushable) SetUnderlyingDB(parent kvdb.KeyValueStore) {
+	w.underlying = parent
+}
+
 // UnderlyingDB of Flushable.
 func (w *Flushable) UnderlyingDB() kvdb.KeyValueStore {
 	return w.underlying
@@ -132,7 +142,9 @@ func (w *Flushable) Close() error {
 
 // Drop whole database.
 func (w *Flushable) Drop() {
-	if droper, ok := w.underlying.(kvdb.Droper); ok {
+	if w.dropper != nil {
+		w.dropper()
+	} else if droper, ok := w.underlying.(kvdb.Droper); ok {
 		droper.Drop()
 	}
 	w.underlying = nil