Merge pull request #521 from hermeznetwork/feature/robuststatedb

Improve stateDB reliability

cli/node/cfg.buidler.toml

@@ -39,9 +39,9 @@ TokenHEZ = "0x5D94e3e7aeC542aB0F9129B9a7BAdeb5B3Ca0f77"
 TokenHEZName = "Hermez Network Token"
 
 [Coordinator]
-ForgerAddress = "0x05c23b938a85ab26A36E6314a0D02080E9ca6BeD" # Non-Boot Coordinator
+# ForgerAddress = "0x05c23b938a85ab26A36E6314a0D02080E9ca6BeD" # Non-Boot Coordinator
 # ForgerAddressPrivateKey = "0x30f5fddb34cd4166adb2c6003fa6b18f380fd2341376be42cf1c7937004ac7a3"
-# ForgerAddress = "0xb4124ceb3451635dacedd11767f004d8a28c6ee7" # Boot Coordinator
+ForgerAddress = "0xb4124ceb3451635dacedd11767f004d8a28c6ee7" # Boot Coordinator
 # ForgerAddressPrivateKey = "0xa8a54b2d8197bc0b19bb8a084031be71835580a01e70a45a13babd16c9bc1563"
 ConfirmBlocks = 10
 L1BatchTimeoutPerc = 0.999

coordinator/coordinator.go

@@ -172,6 +172,16 @@ func NewCoordinator(cfg Config,
 	return &c, nil
 }
 
+// TxSelector returns the inner TxSelector
+func (c *Coordinator) TxSelector() *txselector.TxSelector {
+	return c.txSelector
+}
+
+// BatchBuilder returns the inner BatchBuilder
+func (c *Coordinator) BatchBuilder() *batchbuilder.BatchBuilder {
+	return c.batchBuilder
+}
+
 func (c *Coordinator) newPipeline(ctx context.Context) (*Pipeline, error) {
 	return NewPipeline(ctx, c.cfg, c.historyDB, c.l2DB, c.txSelector,
 		c.batchBuilder, c.purger, c.txManager, c.provers, &c.consts)

db/kvdb/kvdb.go

@@ -8,6 +8,7 @@ import (
 	"path"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/hermeznetwork/hermez-node/common"
 	"github.com/hermeznetwork/hermez-node/log"
@@ -40,6 +41,7 @@ type KVDB struct {
 	CurrentIdx   common.Idx
 	CurrentBatch common.BatchNum
 	keep         int
+	m            sync.Mutex
 }
 
 // NewKVDB creates a new KVDB, allowing to use an in-memory or in-disk storage.
@@ -141,10 +143,8 @@ func (kvdb *KVDB) reset(batchNum common.BatchNum, closeCurrent bool) error {
 		return nil
 	}
 
-	checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
 	// copy 'BatchNumX' to 'current'
-	err = pebbleMakeCheckpoint(checkpointPath, currentPath)
-	if err != nil {
+	if err := kvdb.MakeCheckpointFromTo(batchNum, currentPath); err != nil {
 		return tracerr.Wrap(err)
 	}
 
@@ -212,22 +212,13 @@ func (kvdb *KVDB) ResetFromSynchronizer(batchNum common.BatchNum, synchronizerKV
 
 	checkpointPath := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, batchNum))
 
-	// use checkpoint from synchronizerKVDB
-	synchronizerCheckpointPath := path.Join(synchronizerKVDB.path,
-		fmt.Sprintf("%s%d", PathBatchNum, batchNum))
-	if _, err := os.Stat(synchronizerCheckpointPath); os.IsNotExist(err) {
-		// if synchronizerKVDB does not have checkpoint at batchNum, return err
-		return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" not exist in Synchronizer",
-			synchronizerCheckpointPath))
-	}
 	// copy synchronizer'BatchNumX' to 'BatchNumX'
-	err = pebbleMakeCheckpoint(synchronizerCheckpointPath, checkpointPath)
-	if err != nil {
+	if err := synchronizerKVDB.MakeCheckpointFromTo(batchNum, checkpointPath); err != nil {
 		return tracerr.Wrap(err)
 	}
 
 	// copy 'BatchNumX' to 'current'
-	err = pebbleMakeCheckpoint(checkpointPath, currentPath)
+	err = kvdb.MakeCheckpointFromTo(batchNum, currentPath)
 	if err != nil {
 		return tracerr.Wrap(err)
 	}
@@ -412,6 +403,25 @@ func (kvdb *KVDB) deleteOldCheckpoints() error {
 	return nil
 }
 
+// MakeCheckpointFromTo makes a checkpoint from the current db at fromBatchNum
+// to the dest folder.  This method is locking, so it can be called from
+// multiple places at the same time.
+func (kvdb *KVDB) MakeCheckpointFromTo(fromBatchNum common.BatchNum, dest string) error {
+	source := path.Join(kvdb.path, fmt.Sprintf("%s%d", PathBatchNum, fromBatchNum))
+	if _, err := os.Stat(source); os.IsNotExist(err) {
+		// if kvdb does not have checkpoint at batchNum, return err
+		return tracerr.Wrap(fmt.Errorf("Checkpoint \"%v\" does not exist", source))
+	}
+	// By locking we allow calling MakeCheckpointFromTo from multiple
+	// places at the same time for the same stateDB.  This allows the
+	// synchronizer to do a reset to a batchNum at the same time as the
+	// pipeline is doing a txSelector.Reset and batchBuilder.Reset from
+	// synchronizer to the same batchNum
+	kvdb.m.Lock()
+	defer kvdb.m.Unlock()
+	return pebbleMakeCheckpoint(source, dest)
+}
+
 func pebbleMakeCheckpoint(source, dest string) error {
 	// Remove dest folder (if it exists) before doing the checkpoint
 	if _, err := os.Stat(dest); !os.IsNotExist(err) {

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/gobuffalo/packr/v2 v2.8.1
 	github.com/hermeznetwork/tracerr v0.3.1-0.20210120162744-5da60b576169
 	github.com/iden3/go-iden3-crypto v0.0.6-0.20201221160344-58e589b6eb4c
-	github.com/iden3/go-merkletree v0.0.0-20210105105053-5a669a1b3958
+	github.com/iden3/go-merkletree v0.0.0-20210119155851-bb53e6ad1a12
 	github.com/jinzhu/copier v0.0.0-20190924061706-b57f9002281a
 	github.com/jmoiron/sqlx v1.2.1-0.20200615141059-0794cb1f47ee
 	github.com/joho/godotenv v1.3.0

go.sum

@@ -332,8 +332,8 @@ github.com/hydrogen18/memlistener v0.0.0-20141126152155-54553eb933fb/go.mod h1:q
 github.com/iden3/go-iden3-crypto v0.0.6-0.20201218111145-a2015adb2f1b/go.mod h1:oBgthFLboAWi9feaBUFy7OxEcyn9vA1khHSL/WwWFyg=
 github.com/iden3/go-iden3-crypto v0.0.6-0.20201221160344-58e589b6eb4c h1:D2u8FFYey6iFXLsqqJZ8R7ch8gZum+/b98whvoSDbyg=
 github.com/iden3/go-iden3-crypto v0.0.6-0.20201221160344-58e589b6eb4c/go.mod h1:oBgthFLboAWi9feaBUFy7OxEcyn9vA1khHSL/WwWFyg=
-github.com/iden3/go-merkletree v0.0.0-20210105105053-5a669a1b3958 h1:nt8NbmIOWBPPa2IRrAgWAijQ25Fhk+1CGJ6x2TxD6FY=
-github.com/iden3/go-merkletree v0.0.0-20210105105053-5a669a1b3958/go.mod h1:C2Ox09W/exsvI8yJM7jkZ/FQiytWOCnpWMuVr4S7jPk=
+github.com/iden3/go-merkletree v0.0.0-20210119155851-bb53e6ad1a12 h1:DXWT0BLCSm7cJmTMQy7+iOlxkA1/5ADglufhLK52e10=
+github.com/iden3/go-merkletree v0.0.0-20210119155851-bb53e6ad1a12/go.mod h1:FdUFTW2qJiwHyy5R70uErwq7Kaq1uskyFdTfodcUJqA=
 github.com/imkira/go-interpol v1.1.0/go.mod h1:z0h2/2T3XF8kyEPpRgJ3kmNv+C43p+I/CoI+jC3w2iA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/influxdb v1.2.3-0.20180221223340-01288bdb0883/go.mod h1:qZna6X/4elxqT3yI9iZYdZrWWdeFOOprn86kgg4+IzY=
@@ -594,9 +594,8 @@ github.com/shirou/gopsutil v2.20.5+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu
 github.com/shurcooL/sanitized_anchor_name v1.0.0 h1:PdmoCO6wvbs+7yrJyMORt4/BmY5IYyJwS/kOiWx8mHo=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/sirupsen/logrus v1.5.0 h1:1N5EYkVAPEywqZRJd7cwnRtCb6xJx7NH3T3WUTF980Q=
-github.com/sirupsen/logrus v1.5.0/go.mod h1:+F7Ogzej0PZc/94MaYx/nvG9jOFMD2osvC3s+Squfpo=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=

node/node.go

@@ -655,4 +655,10 @@ func (n *Node) Stop() {
 		log.Info("Stopping Coordinator...")
 		n.coord.Stop()
 	}
+	// Close kv DBs
+	n.sync.StateDB().Close()
+	if n.mode == ModeCoordinator {
+		n.coord.TxSelector().LocalAccountsDB().Close()
+		n.coord.BatchBuilder().LocalStateDB().Close()
+	}
 }

synchronizer/synchronizer.go

@@ -262,6 +262,11 @@ func NewSynchronizer(ethClient eth.ClientInterface, historyDB *historydb.History
 	return s, s.init()
 }
 
+// StateDB returns the inner StateDB
+func (s *Synchronizer) StateDB() *statedb.StateDB {
+	return s.stateDB
+}
+
 // Stats returns a copy of the Synchronizer Stats.  It is safe to call Stats()
 // during a Sync call
 func (s *Synchronizer) Stats() *Stats {