Add minPriceUSD in L2DB, check maxTxs atomically

- Add config parameter `Coordinator.L2DB.MinPriceUSD` which allows rejecting
  txs to the pool that have a fee lower than the minimum.
- In pool tx insertion, checking the number of pending txs atomically with the
  insertion to avoid data races leading to more than MaxTxs pending txs in the
  pool.

api/account.go

@@ -4,7 +4,10 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
+	"github.com/hermeznetwork/hermez-node/apitypes"
 	"github.com/hermeznetwork/hermez-node/db/historydb"
+	"github.com/hermeznetwork/hermez-node/db/statedb"
+	"github.com/hermeznetwork/tracerr"
 )
 
 func (a *API) getAccount(c *gin.Context) {
@@ -20,6 +23,16 @@ func (a *API) getAccount(c *gin.Context) {
 		return
 	}
 
+	// Get balance from stateDB
+	account, err := a.s.LastGetAccount(*idx)
+	if err != nil {
+		retSQLErr(err, c)
+		return
+	}
+
+	apiAccount.Balance = apitypes.NewBigIntStr(account.Balance)
+	apiAccount.Nonce = account.Nonce
+
 	c.JSON(http.StatusOK, apiAccount)
 }
 
@@ -44,6 +57,26 @@ func (a *API) getAccounts(c *gin.Context) {
 		return
 	}
 
+	// Get balances from stateDB
+	if err := a.s.LastRead(func(sdb *statedb.Last) error {
+		for x, apiAccount := range apiAccounts {
+			idx, err := stringToIdx(string(apiAccount.Idx), "Account Idx")
+			if err != nil {
+				return tracerr.Wrap(err)
+			}
+			account, err := sdb.GetAccount(*idx)
+			if err != nil {
+				return tracerr.Wrap(err)
+			}
+			apiAccounts[x].Balance = apitypes.NewBigIntStr(account.Balance)
+			apiAccounts[x].Nonce = account.Nonce
+		}
+		return nil
+	}); err != nil {
+		retSQLErr(err, c)
+		return
+	}
+
 	// Build succesfull response
 	type accountResponse struct {
 		Accounts     []historydb.AccountAPI `json:"accounts"`

api/api.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hermeznetwork/hermez-node/common"
 	"github.com/hermeznetwork/hermez-node/db/historydb"
 	"github.com/hermeznetwork/hermez-node/db/l2db"
+	"github.com/hermeznetwork/hermez-node/db/statedb"
 	"github.com/hermeznetwork/tracerr"
 )
 
@@ -33,6 +34,7 @@ type Status struct {
 type API struct {
 	h             *historydb.HistoryDB
 	cg            *configAPI
+	s             *statedb.StateDB
 	l2            *l2db.L2DB
 	status        Status
 	chainID       uint16
@@ -44,6 +46,7 @@ func NewAPI(
 	coordinatorEndpoints, explorerEndpoints bool,
 	server *gin.Engine,
 	hdb *historydb.HistoryDB,
+	sdb *statedb.StateDB,
 	l2db *l2db.L2DB,
 	config *Config,
 ) (*API, error) {
@@ -63,6 +66,7 @@ func NewAPI(
 			AuctionConstants:  config.AuctionConstants,
 			WDelayerConstants: config.WDelayerConstants,
 		},
+		s:             sdb,
 		l2:            l2db,
 		status:        Status{},
 		chainID:       config.ChainID,

api/api_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/hermeznetwork/hermez-node/db"
 	"github.com/hermeznetwork/hermez-node/db/historydb"
 	"github.com/hermeznetwork/hermez-node/db/l2db"
+	"github.com/hermeznetwork/hermez-node/db/statedb"
 	"github.com/hermeznetwork/hermez-node/log"
 	"github.com/hermeznetwork/hermez-node/test"
 	"github.com/hermeznetwork/hermez-node/test/til"
@@ -215,6 +216,10 @@ func TestMain(m *testing.M) {
 			panic(err)
 		}
 	}()
+	sdb, err := statedb.NewStateDB(statedb.Config{Path: dir, Keep: 128, Type: statedb.TypeTxSelector, NLevels: 0})
+	if err != nil {
+		panic(err)
+	}
 	// L2DB
 	l2DB := l2db.NewL2DB(database, 10, 1000, 0.0, 24*time.Hour, apiConnCon)
 	test.WipeDB(l2DB.DB()) // this will clean HistoryDB and L2DB
@@ -234,6 +239,7 @@ func TestMain(m *testing.M) {
 		true,
 		apiGin,
 		hdb,
+		sdb,
 		l2DB,
 		&_config,
 	)
@@ -344,6 +350,19 @@ func TestMain(m *testing.M) {
 		}
 	}
 
+	// lastBlockNum2 := blocksData[len(blocksData)-1].Block.EthBlockNum
+
+	// Add accounts to StateDB
+	for i := 0; i < len(commonAccounts); i++ {
+		if _, err := api.s.CreateAccount(commonAccounts[i].Idx, &commonAccounts[i]); err != nil {
+			panic(err)
+		}
+	}
+	// Make a checkpoint to make the accounts available in Last
+	if err := api.s.MakeCheckpoint(); err != nil {
+		panic(err)
+	}
+
 	// Generate Coordinators and add them to HistoryDB
 	const nCoords = 10
 	commonCoords := test.GenCoordinators(nCoords, commonBlocks)
@@ -510,41 +529,13 @@ func TestMain(m *testing.M) {
 	testTxs := genTestTxs(commonL1Txs, commonL2Txs, commonAccounts, testTokens, commonBlocks)
 	testBatches, testFullBatches := genTestBatches(commonBlocks, commonBatches, testTxs)
 	poolTxsToSend, poolTxsToReceive := genTestPoolTxs(commonPoolTxs, testTokens, commonAccounts)
-	// Add balance and nonce to historyDB
-	accounts := genTestAccounts(commonAccounts, testTokens)
-	accUpdates := []common.AccountUpdate{}
-	for i := 0; i < len(accounts); i++ {
-		balance := new(big.Int)
-		balance.SetString(string(*accounts[i].Balance), 10)
-		idx, err := stringToIdx(string(accounts[i].Idx), "foo")
-		if err != nil {
-			panic(err)
-		}
-		accUpdates = append(accUpdates, common.AccountUpdate{
-			EthBlockNum: 0,
-			BatchNum:    1,
-			Idx:         *idx,
-			Nonce:       0,
-			Balance:     balance,
-		})
-		accUpdates = append(accUpdates, common.AccountUpdate{
-			EthBlockNum: 0,
-			BatchNum:    1,
-			Idx:         *idx,
-			Nonce:       accounts[i].Nonce,
-			Balance:     balance,
-		})
-	}
-	if err := api.h.AddAccountUpdates(accUpdates); err != nil {
-		panic(err)
-	}
 	tc = testCommon{
 		blocks:           commonBlocks,
 		tokens:           testTokens,
 		batches:          testBatches,
 		fullBatches:      testFullBatches,
 		coordinators:     testCoords,
-		accounts:         accounts,
+		accounts:         genTestAccounts(commonAccounts, testTokens),
 		txs:              testTxs,
 		exits:            testExits,
 		poolTxsToSend:    poolTxsToSend,
@@ -621,6 +612,7 @@ func TestTimeout(t *testing.T) {
 		true,
 		apiGinTO,
 		hdbTO,
+		nil,
 		l2DBTO,
 		&_config,
 	)

api/txspool.go

@@ -171,13 +171,13 @@ func (a *API) verifyPoolL2TxWrite(txw *l2db.PoolL2TxWrite) error {
 	if err != nil {
 		return tracerr.Wrap(err)
 	}
-	// Validate feeAmount
+	// Get public key
-	_, err = common.CalcFeeAmount(poolTx.Amount, poolTx.Fee)
+	account, err := a.s.LastGetAccount(poolTx.FromIdx)
 	if err != nil {
 		return tracerr.Wrap(err)
 	}
-	// Get public key
+	// Validate feeAmount
-	account, err := a.h.GetCommonAccountAPI(poolTx.FromIdx)
+	_, err = common.CalcFeeAmount(poolTx.Amount, poolTx.Fee)
 	if err != nil {
 		return tracerr.Wrap(err)
 	}

batchbuilder/batchbuilder.go

@@ -64,10 +64,7 @@ func (bb *BatchBuilder) BuildBatch(coordIdxs []common.Idx, configBatch *ConfigBa
 	tp := txprocessor.NewTxProcessor(bbStateDB, configBatch.TxProcessorConfig)
 
 	ptOut, err := tp.ProcessTxs(coordIdxs, l1usertxs, l1coordinatortxs, pooll2txs)
-	if err != nil {
+	return ptOut.ZKInputs, tracerr.Wrap(err)
-		return nil, tracerr.Wrap(err)
-	}
-	return ptOut.ZKInputs, nil
 }
 
 // LocalStateDB returns the underlying LocalStateDB

cli/node/cfg.buidler.toml

@@ -32,6 +32,7 @@ URL = "http://localhost:8545"
 [Synchronizer]
 SyncLoopInterval = "1s"
 StatsRefreshPeriod = "1s"
+StoreAccountUpdates = true
 
 [SmartContracts]
 Rollup   = "0x8EEaea23686c319133a7cC110b840d1591d9AeE0"

common/l1tx_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestNewL1UserTxID(t *testing.T) {
+func TestNewL1UserTx(t *testing.T) {
 	toForge := int64(123456)
 	l1Tx := &L1Tx{
 		ToForgeL1TxsNum: &toForge,
@@ -30,38 +30,6 @@ func TestNewL1UserTxID(t *testing.T) {
 	l1Tx, err := NewL1Tx(l1Tx)
 	assert.NoError(t, err)
 	assert.Equal(t, "0x00a6cbae3b8661fb75b0919ca6605a02cfb04d9c6dd16870fa0fcdf01befa32768", l1Tx.TxID.String())
-
-	maxInt64 := 0xFFFF_FFFF_FFFF_FFFF >> 1
-
-	toForge = int64(maxInt64)
-	l1Tx = &L1Tx{
-		ToForgeL1TxsNum: &toForge,
-		Position:        maxInt64,
-		UserOrigin:      true,
-	}
-	l1Tx, err = NewL1Tx(l1Tx)
-	assert.NoError(t, err)
-	assert.Equal(t, "0x001ff31eb325f324652bfe6b607a19e04789e082ee3b779eefe4a466062ea331d9", l1Tx.TxID.String())
-
-	toForge = int64(maxInt64 - 1)
-	l1Tx = &L1Tx{
-		ToForgeL1TxsNum: &toForge,
-		Position:        maxInt64 - 1,
-		UserOrigin:      true,
-	}
-	l1Tx, err = NewL1Tx(l1Tx)
-	assert.NoError(t, err)
-	assert.Equal(t, "0x0003434eca58d35fd85795e3a6cce67c8801deb805ea1f7429cc270aa9f35ea403", l1Tx.TxID.String())
-
-	toForge = int64(0)
-	l1Tx = &L1Tx{
-		ToForgeL1TxsNum: &toForge,
-		Position:        0,
-		UserOrigin:      true,
-	}
-	l1Tx, err = NewL1Tx(l1Tx)
-	assert.NoError(t, err)
-	assert.Equal(t, "0x006bd2dd6bd408cbee33429358bf24fdc64612fbf8b1b4db604518f40ffd34b607", l1Tx.TxID.String())
 }
 
 func TestNewL1CoordinatorTx(t *testing.T) {

common/l2tx_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestNewL2TxID(t *testing.T) {
+func TestNewL2Tx(t *testing.T) {
 	l2Tx := &L2Tx{
 		FromIdx: 87654,
 		ToIdx:   300,

config/config.go

@@ -240,6 +240,11 @@ type Node struct {
 		// `Eth.LastBatch`).  This value only affects the reported % of
 		// synchronization of blocks and batches, nothing else.
 		StatsRefreshPeriod Duration `validate:"required"`
+		// StoreAccountUpdates when set to true makes the synchronizer
+		// store every account update in the account_update SQL table.
+		// This allows querying nonces and balances from the HistoryDB
+		// via SQL.
+		StoreAccountUpdates bool
 	} `validate:"required"`
 	SmartContracts struct {
 		// Rollup is the address of the Hermez.sol smart contract

coordinator/txmanager.go

@@ -231,7 +231,7 @@ func (t *TxManager) sendRollupForgeBatch(ctx context.Context, batchInfo *BatchIn
 				"err", err, "gasPrice", auth.GasPrice, "batchNum", batchInfo.BatchNum)
 			auth.GasPrice = addPerc(auth.GasPrice, 10)
 			attempt--
-		} else {
+		} else if err != nil {
 			log.Errorw("TxManager ethClient.RollupForgeBatch",
 				"attempt", attempt, "err", err, "block", t.stats.Eth.LastBlock.Num+1,
 				"batchNum", batchInfo.BatchNum)

db/historydb/apiqueries.go

@@ -835,16 +835,8 @@ func (hdb *HistoryDB) GetAccountAPI(idx common.Idx) (*AccountAPI, error) {
 	err = meddler.QueryRow(hdb.db, account, `SELECT account.item_id, hez_idx(account.idx, 
 	token.symbol) as idx, account.batch_num, account.bjj, account.eth_addr,
 	token.token_id, token.item_id AS token_item_id, token.eth_block_num AS token_block,
-		token.eth_addr as token_eth_addr, token.name, token.symbol, token.decimals, token.usd,
+	token.eth_addr as token_eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update 
-		token.usd_update, account_update.nonce, account_update.balance 
+	FROM account INNER JOIN token ON account.token_id = token.token_id WHERE idx = $1;`, idx)
-		FROM account inner JOIN (
-			SELECT idx, nonce, balance 
-			FROM account_update
-			WHERE idx = $1
-			ORDER BY item_id DESC LIMIT 1
-		) AS account_update ON account_update.idx = account.idx 
-		INNER JOIN token ON account.token_id = token.token_id
-		WHERE account.idx = $1;`, idx)
 
 	if err != nil {
 		return nil, tracerr.Wrap(err)
@@ -872,13 +864,8 @@ func (hdb *HistoryDB) GetAccountsAPI(
 	queryStr := `SELECT account.item_id, hez_idx(account.idx, token.symbol) as idx, account.batch_num, 
 	account.bjj, account.eth_addr, token.token_id, token.item_id AS token_item_id, token.eth_block_num AS token_block,
 	token.eth_addr as token_eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update, 
-	account_update.nonce, account_update.balance, COUNT(*) OVER() AS total_items
+	COUNT(*) OVER() AS total_items
-	FROM account inner JOIN (
+	FROM account INNER JOIN token ON account.token_id = token.token_id `
-		SELECT DISTINCT idx,
-		first_value(nonce) over(partition by idx ORDER BY item_id DESC) as nonce,
-		first_value(balance) over(partition by idx ORDER BY item_id DESC) as balance
-		FROM account_update
-	) AS account_update ON account_update.idx = account.idx INNER JOIN token ON account.token_id = token.token_id `
 	// Apply filters
 	nextIsAnd := false
 	// ethAddr filter
@@ -1037,18 +1024,3 @@ func (hdb *HistoryDB) GetAvgTxFeeAPI() (float64, error) {
 
 	return avgTransactionFee, nil
 }
-
-// GetCommonAccountAPI returns the account associated to an account idx
-func (hdb *HistoryDB) GetCommonAccountAPI(idx common.Idx) (*common.Account, error) {
-	cancel, err := hdb.apiConnCon.Acquire()
-	defer cancel()
-	if err != nil {
-		return nil, tracerr.Wrap(err)
-	}
-	defer hdb.apiConnCon.Release()
-	account := &common.Account{}
-	err = meddler.QueryRow(
-		hdb.db, account, `SELECT * FROM account WHERE idx = $1;`, idx,
-	)
-	return account, tracerr.Wrap(err)
-}

db/historydb/historydb.go

@@ -448,7 +448,7 @@ func (hdb *HistoryDB) addTokens(d meddler.DB, tokens []common.Token) error {
 }
 
 // UpdateTokenValue updates the USD value of a token.  Value is the price in
-// USD of a normalized token (1 token = 10^decimals units)
+// USD of a normalized token (token amount divided by 10^decimals)
 func (hdb *HistoryDB) UpdateTokenValue(tokenSymbol string, value float64) error {
 	// Sanitize symbol
 	tokenSymbol = strings.ToValidUTF8(tokenSymbol, " ")

db/historydb/views.go

@@ -239,8 +239,8 @@ type AccountAPI struct {
 	BatchNum         common.BatchNum     `meddler:"batch_num"`
 	PublicKey        apitypes.HezBJJ     `meddler:"bjj"`
 	EthAddr          apitypes.HezEthAddr `meddler:"eth_addr"`
-	Nonce            common.Nonce        `meddler:"nonce"`   // max of 40 bits used
+	Nonce            common.Nonce        `meddler:"-"` // max of 40 bits used
-	Balance          *apitypes.BigIntStr `meddler:"balance"` // max of 192 bits used
+	Balance          *apitypes.BigIntStr `meddler:"-"` // max of 192 bits used
 	TotalItems       uint64              `meddler:"total_items"`
 	FirstItem        uint64              `meddler:"first_item"`
 	LastItem         uint64              `meddler:"last_item"`

db/migrations/0001.sql

@@ -47,7 +47,7 @@ CREATE TABLE token (
     name VARCHAR(20) NOT NULL,
     symbol VARCHAR(10) NOT NULL,
     decimals INT NOT NULL,
-    usd NUMERIC, -- value of a normalized token (1 token = 10^decimals units)
+    usd NUMERIC, -- value of a normalized token (divided by 10^decimals)
     usd_update TIMESTAMP WITHOUT TIME ZONE
 );
 
@@ -662,35 +662,35 @@ CREATE TABLE account_creation_auth (
 );
 
 -- +migrate Down
--- triggers
+-- drop triggers
-DROP TRIGGER IF EXISTS trigger_token_usd_update ON token;
+DROP TRIGGER trigger_token_usd_update ON token;
-DROP TRIGGER IF EXISTS trigger_set_tx ON tx;
+DROP TRIGGER trigger_set_tx ON tx;
-DROP TRIGGER IF EXISTS trigger_forge_l1_txs ON batch;
+DROP TRIGGER trigger_forge_l1_txs ON batch;
-DROP TRIGGER IF EXISTS trigger_set_pool_tx ON tx_pool;
+DROP TRIGGER trigger_set_pool_tx ON tx_pool;
--- functions
+-- drop functions
-DROP FUNCTION IF EXISTS hez_idx;
+DROP FUNCTION hez_idx;
-DROP FUNCTION IF EXISTS set_token_usd_update;
+DROP FUNCTION set_token_usd_update;
-DROP FUNCTION IF EXISTS fee_percentage;
+DROP FUNCTION fee_percentage;
-DROP FUNCTION IF EXISTS set_tx;
+DROP FUNCTION set_tx;
-DROP FUNCTION IF EXISTS forge_l1_user_txs;
+DROP FUNCTION forge_l1_user_txs;
-DROP FUNCTION IF EXISTS set_pool_tx;
+DROP FUNCTION set_pool_tx;
--- drop tables IF EXISTS
+-- drop tables
-DROP TABLE IF EXISTS account_creation_auth;
+DROP TABLE account_creation_auth;
-DROP TABLE IF EXISTS tx_pool;
+DROP TABLE tx_pool;
-DROP TABLE IF EXISTS auction_vars;
+DROP TABLE auction_vars;
-DROP TABLE IF EXISTS rollup_vars;
+DROP TABLE rollup_vars;
-DROP TABLE IF EXISTS escape_hatch_withdrawal;
+DROP TABLE escape_hatch_withdrawal;
-DROP TABLE IF EXISTS bucket_update;
+DROP TABLE bucket_update;
-DROP TABLE IF EXISTS token_exchange;
+DROP TABLE token_exchange;
-DROP TABLE IF EXISTS wdelayer_vars;
+DROP TABLE wdelayer_vars;
-DROP TABLE IF EXISTS tx;
+DROP TABLE tx;
-DROP TABLE IF EXISTS exit_tree;
+DROP TABLE exit_tree;
-DROP TABLE IF EXISTS account_update;
+DROP TABLE account_update;
-DROP TABLE IF EXISTS account;
+DROP TABLE account;
-DROP TABLE IF EXISTS token;
+DROP TABLE token;
-DROP TABLE IF EXISTS bid;
+DROP TABLE bid;
-DROP TABLE IF EXISTS batch;
+DROP TABLE batch;
-DROP TABLE IF EXISTS coordinator;
+DROP TABLE coordinator;
-DROP TABLE IF EXISTS block;
+DROP TABLE block;
--- sequences
+-- drop sequences
-DROP SEQUENCE IF EXISTS tx_item_id;
+DROP SEQUENCE tx_item_id;

node/node.go

@@ -184,6 +184,7 @@ func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
 
 	sync, err := synchronizer.NewSynchronizer(client, historyDB, stateDB, synchronizer.Config{
 		StatsRefreshPeriod:  cfg.Synchronizer.StatsRefreshPeriod.Duration,
+		StoreAccountUpdates: cfg.Synchronizer.StoreAccountUpdates,
 		ChainID:             chainIDU16,
 	})
 	if err != nil {
@@ -246,6 +247,9 @@ func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
 		if err != nil {
 			return nil, tracerr.Wrap(err)
 		}
+		if err != nil {
+			return nil, tracerr.Wrap(err)
+		}
 		serverProofs := make([]prover.Client, len(cfg.Coordinator.ServerProofs))
 		for i, serverProofCfg := range cfg.Coordinator.ServerProofs {
 			serverProofs[i] = prover.NewProofServerClient(serverProofCfg.URL,
@@ -427,6 +431,7 @@ func NewNodeAPI(
 		coordinatorEndpoints, explorerEndpoints,
 		engine,
 		hdb,
+		sdb,
 		l2db,
 		config,
 	)

synchronizer/synchronizer.go

@@ -207,6 +207,7 @@ type SCConsts struct {
 // Config is the Synchronizer configuration
 type Config struct {
 	StatsRefreshPeriod  time.Duration
+	StoreAccountUpdates bool
 	ChainID             uint16
 }
 
@@ -993,6 +994,7 @@ func (s *Synchronizer) rollupSync(ethBlock *common.Block) (*common.RollupData, e
 		}
 		batchData.CreatedAccounts = processTxsOut.CreatedAccounts
 
+		if s.cfg.StoreAccountUpdates {
 			batchData.UpdatedAccounts = make([]common.AccountUpdate, 0,
 				len(processTxsOut.UpdatedAccounts))
 			for _, acc := range processTxsOut.UpdatedAccounts {
@@ -1005,6 +1007,7 @@ func (s *Synchronizer) rollupSync(ethBlock *common.Block) (*common.RollupData, e
 						Balance:     acc.Balance,
 					})
 			}
+		}
 
 		slotNum := int64(0)
 		if ethBlock.Num >= s.consts.Auction.GenesisBlockNum {

synchronizer/synchronizer_test.go

@@ -347,6 +347,7 @@ func TestSyncGeneral(t *testing.T) {
 	// Create Synchronizer
 	s, err := NewSynchronizer(client, historyDB, stateDB, Config{
 		StatsRefreshPeriod:  0 * time.Second,
+		StoreAccountUpdates: true,
 	})
 	require.NoError(t, err)
 
@@ -738,6 +739,7 @@ func TestSyncForgerCommitment(t *testing.T) {
 	// Create Synchronizer
 	s, err := NewSynchronizer(client, historyDB, stateDB, Config{
 		StatsRefreshPeriod:  0 * time.Second,
+		StoreAccountUpdates: true,
 	})
 	require.NoError(t, err)
 
@@ -838,6 +840,7 @@ func TestSyncForgerCommitment(t *testing.T) {
 
 		s2, err := NewSynchronizer(client, historyDB, stateDB, Config{
 			StatsRefreshPeriod:  0 * time.Second,
+			StoreAccountUpdates: true,
 		})
 		require.NoError(t, err)
 		stats = s2.Stats()