From 217a41d465be172703be48154c5599d945fec714 Mon Sep 17 00:00:00 2001 From: Eduard S Date: Mon, 22 Feb 2021 16:37:18 +0100 Subject: [PATCH] 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/api_test.go | 4 +- api/txspool.go | 6 + api/txspool_test.go | 27 ++-- cli/node/cfg.buidler.toml | 1 + cli/node/main.go | 1 + common/utils.go | 14 ++ config/config.go | 4 + coordinator/coordinator_test.go | 2 +- coordinator/purger_test.go | 2 +- db/historydb/historydb.go | 3 +- db/l2db/apiqueries.go | 62 ++++++-- db/l2db/l2db.go | 18 ++- db/l2db/l2db_test.go | 249 ++++++++++++++++++++------------ db/migrations/0001.sql | 2 +- node/node.go | 1 + test/zkproof/flows_test.go | 2 +- txselector/txselector_test.go | 2 +- 17 files changed, 272 insertions(+), 128 deletions(-) diff --git a/api/api_test.go b/api/api_test.go index 9699a30..2a5941d 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -221,7 +221,7 @@ func TestMain(m *testing.M) { panic(err) } // L2DB - l2DB := l2db.NewL2DB(database, 10, 1000, 24*time.Hour, apiConnCon) + l2DB := l2db.NewL2DB(database, 10, 1000, 0.0, 24*time.Hour, apiConnCon) test.WipeDB(l2DB.DB()) // this will clean HistoryDB and L2DB // Config (smart contract constants) chainID := uint16(0) @@ -585,7 +585,7 @@ func TestTimeout(t *testing.T) { hdbTO := historydb.NewHistoryDB(databaseTO, apiConnConTO) require.NoError(t, err) // L2DB - l2DBTO := l2db.NewL2DB(databaseTO, 10, 1000, 24*time.Hour, apiConnConTO) + l2DBTO := l2db.NewL2DB(databaseTO, 10, 1000, 0.0, 24*time.Hour, apiConnConTO) // API apiGinTO := gin.Default() diff --git a/api/txspool.go b/api/txspool.go index b990818..f09ac0f 100644 --- a/api/txspool.go +++ b/api/txspool.go @@ -2,6 +2,7 @@ package api import ( "errors" + "fmt" "math/big" "net/http" @@ -180,6 +181,11 @@ func (a *API) verifyPoolL2TxWrite(txw *l2db.PoolL2TxWrite) error { if err != nil { return tracerr.Wrap(err) } + // Validate TokenID + if poolTx.TokenID != account.TokenID { + return tracerr.Wrap(fmt.Errorf("tx.TokenID (%v) != account.TokenID (%v)", + poolTx.TokenID, account.TokenID)) + } // Check signature if !poolTx.VerifySignature(a.chainID, account.BJJ) { return tracerr.Wrap(errors.New("wrong signature")) diff --git a/api/txspool_test.go b/api/txspool_test.go index 755c485..379b4d3 100644 --- a/api/txspool_test.go +++ b/api/txspool_test.go @@ -10,6 +10,7 @@ import ( "github.com/hermeznetwork/hermez-node/db/historydb" "github.com/iden3/go-iden3-crypto/babyjub" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // testPoolTxReceive is a struct to be used to assert the response @@ -170,9 +171,9 @@ func TestPoolTxs(t *testing.T) { fetchedTxID := common.TxID{} for _, tx := range tc.poolTxsToSend { jsonTxBytes, err := json.Marshal(tx) - assert.NoError(t, err) + require.NoError(t, err) jsonTxReader := bytes.NewReader(jsonTxBytes) - assert.NoError( + require.NoError( t, doGoodReq( "POST", endpoint, @@ -187,42 +188,42 @@ func TestPoolTxs(t *testing.T) { badTx.Amount = "99950000000000000" badTx.Fee = 255 jsonTxBytes, err := json.Marshal(badTx) - assert.NoError(t, err) + require.NoError(t, err) jsonTxReader := bytes.NewReader(jsonTxBytes) err = doBadReq("POST", endpoint, jsonTxReader, 400) - assert.NoError(t, err) + require.NoError(t, err) // Wrong signature badTx = tc.poolTxsToSend[0] badTx.FromIdx = "hez:foo:1000" jsonTxBytes, err = json.Marshal(badTx) - assert.NoError(t, err) + require.NoError(t, err) jsonTxReader = bytes.NewReader(jsonTxBytes) err = doBadReq("POST", endpoint, jsonTxReader, 400) - assert.NoError(t, err) + require.NoError(t, err) // Wrong to badTx = tc.poolTxsToSend[0] ethAddr := "hez:0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF" badTx.ToEthAddr = ðAddr badTx.ToIdx = nil jsonTxBytes, err = json.Marshal(badTx) - assert.NoError(t, err) + require.NoError(t, err) jsonTxReader = bytes.NewReader(jsonTxBytes) err = doBadReq("POST", endpoint, jsonTxReader, 400) - assert.NoError(t, err) + require.NoError(t, err) // Wrong rq badTx = tc.poolTxsToSend[0] rqFromIdx := "hez:foo:30" badTx.RqFromIdx = &rqFromIdx jsonTxBytes, err = json.Marshal(badTx) - assert.NoError(t, err) + require.NoError(t, err) jsonTxReader = bytes.NewReader(jsonTxBytes) err = doBadReq("POST", endpoint, jsonTxReader, 400) - assert.NoError(t, err) + require.NoError(t, err) // GET endpoint += "/" for _, tx := range tc.poolTxsToReceive { fetchedTx := testPoolTxReceive{} - assert.NoError( + require.NoError( t, doGoodReq( "GET", endpoint+tx.TxID.String(), @@ -233,10 +234,10 @@ func TestPoolTxs(t *testing.T) { } // 400, due invalid TxID err = doBadReq("GET", endpoint+"0xG2241b6f2b1dd772dba391f4a1a3407c7c21f598d86e2585a14e616fb4a255f823", nil, 400) - assert.NoError(t, err) + require.NoError(t, err) // 404, due inexistent TxID in DB err = doBadReq("GET", endpoint+"0x02241b6f2b1dd772dba391f4a1a3407c7c21f598d86e2585a14e616fb4a255f823", nil, 404) - assert.NoError(t, err) + require.NoError(t, err) } func assertPoolTx(t *testing.T, expected, actual testPoolTxReceive) { diff --git a/cli/node/cfg.buidler.toml b/cli/node/cfg.buidler.toml index 941b6ed..4311710 100644 --- a/cli/node/cfg.buidler.toml +++ b/cli/node/cfg.buidler.toml @@ -67,6 +67,7 @@ BJJ = "0x1b176232f78ba0d388ecc5f4896eca2d3b3d4f272092469f559247297f5c0c13" [Coordinator.L2DB] SafetyPeriod = 10 MaxTxs = 512 +MinFeeUSD = 0.0 TTL = "24h" PurgeBatchDelay = 10 InvalidateBatchDelay = 20 diff --git a/cli/node/main.go b/cli/node/main.go index 132b5ab..eb270f3 100644 --- a/cli/node/main.go +++ b/cli/node/main.go @@ -173,6 +173,7 @@ func cmdDiscard(c *cli.Context) error { db, cfg.Coordinator.L2DB.SafetyPeriod, cfg.Coordinator.L2DB.MaxTxs, + cfg.Coordinator.L2DB.MinFeeUSD, cfg.Coordinator.L2DB.TTL.Duration, nil, ) diff --git a/common/utils.go b/common/utils.go index bcffa5d..b086f5e 100644 --- a/common/utils.go +++ b/common/utils.go @@ -62,3 +62,17 @@ func RmEndingZeroes(siblings []*merkletree.Hash) []*merkletree.Hash { } return siblings[:pos] } + +// TokensToUSD is a helper function to calculate the USD value of a certain +// amount of tokens considering the normalized token price (which is the price +// commonly reported by exhanges) +func TokensToUSD(amount *big.Int, decimals uint64, valueUSD float64) float64 { + amountF := new(big.Float).SetInt(amount) + // Divide by 10^decimals to normalize the amount + baseF := new(big.Float).SetInt(new(big.Int).Exp( + big.NewInt(10), big.NewInt(int64(decimals)), nil)) //nolint:gomnd + amountF.Mul(amountF, big.NewFloat(valueUSD)) + amountF.Quo(amountF, baseF) + amountUSD, _ := amountF.Float64() + return amountUSD +} diff --git a/config/config.go b/config/config.go index 07921f6..e01c102 100644 --- a/config/config.go +++ b/config/config.go @@ -105,6 +105,10 @@ type Coordinator struct { // reached, inserts to the pool will be denied until some of // the pending txs are forged. MaxTxs uint32 `validate:"required"` + // MinFeeUSD is the minimum fee in USD that a tx must pay in + // order to be accepted into the pool. Txs with lower than + // minimum fee will be rejected at the API level. + MinFeeUSD float64 // TTL is the Time To Live for L2Txs in the pool. Once MaxTxs // L2Txs is reached, L2Txs older than TTL will be deleted. TTL Duration `validate:"required"` diff --git a/coordinator/coordinator_test.go b/coordinator/coordinator_test.go index 3bb6902..548ac7c 100644 --- a/coordinator/coordinator_test.go +++ b/coordinator/coordinator_test.go @@ -105,7 +105,7 @@ func newTestModules(t *testing.T) modules { db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) test.WipeDB(db) - l2DB := l2db.NewL2DB(db, 10, 100, 24*time.Hour, nil) + l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) historyDB := historydb.NewHistoryDB(db, nil) txSelDBPath, err = ioutil.TempDir("", "tmpTxSelDB") diff --git a/coordinator/purger_test.go b/coordinator/purger_test.go index d26f80f..821ff7e 100644 --- a/coordinator/purger_test.go +++ b/coordinator/purger_test.go @@ -21,7 +21,7 @@ func newL2DB(t *testing.T) *l2db.L2DB { db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) test.WipeDB(db) - return l2db.NewL2DB(db, 10, 100, 24*time.Hour, nil) + return l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) } func newStateDB(t *testing.T) *statedb.LocalStateDB { diff --git a/db/historydb/historydb.go b/db/historydb/historydb.go index 2f67c57..78faf29 100644 --- a/db/historydb/historydb.go +++ b/db/historydb/historydb.go @@ -447,7 +447,8 @@ func (hdb *HistoryDB) addTokens(d meddler.DB, tokens []common.Token) error { )) } -// UpdateTokenValue updates the USD value of a token +// UpdateTokenValue updates the USD value of a token. Value is the price in +// 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, " ") diff --git a/db/l2db/apiqueries.go b/db/l2db/apiqueries.go index 114066d..914300b 100644 --- a/db/l2db/apiqueries.go +++ b/db/l2db/apiqueries.go @@ -1,12 +1,18 @@ package l2db import ( + "fmt" + ethCommon "github.com/ethereum/go-ethereum/common" "github.com/hermeznetwork/hermez-node/common" "github.com/hermeznetwork/tracerr" "github.com/russross/meddler" ) +var ( + errPoolFull = fmt.Errorf("the pool is at full capacity. More transactions are not accepted currently") +) + // AddAccountCreationAuthAPI inserts an account creation authorization into the DB func (l2db *L2DB) AddAccountCreationAuthAPI(auth *common.AccountCreationAuth) error { cancel, err := l2db.apiConnCon.Acquire() @@ -42,20 +48,54 @@ func (l2db *L2DB) AddTxAPI(tx *PoolL2TxWrite) error { return tracerr.Wrap(err) } defer l2db.apiConnCon.Release() - row := l2db.db.QueryRow( - "SELECT COUNT(*) FROM tx_pool WHERE state = $1;", - common.PoolL2TxStatePending, - ) - var totalTxs uint32 - if err := row.Scan(&totalTxs); err != nil { + + row := l2db.db.QueryRow(`SELECT + ($1::NUMERIC * token.usd * fee_percentage($2::NUMERIC)) / + (10.0 ^ token.decimals::NUMERIC) + FROM token WHERE token.token_id = $3;`, + tx.AmountFloat, tx.Fee, tx.TokenID) + var feeUSD float64 + if err := row.Scan(&feeUSD); err != nil { + return tracerr.Wrap(err) + } + if feeUSD < l2db.minFeeUSD { + return tracerr.Wrap(fmt.Errorf("tx.feeUSD (%v) < minFeeUSD (%v)", + feeUSD, l2db.minFeeUSD)) + } + + // Prepare insert SQL query argument parameters + namesPart, err := meddler.Default.ColumnsQuoted(tx, false) + if err != nil { + return err + } + valuesPart, err := meddler.Default.PlaceholdersString(tx, false) + if err != nil { + return err + } + values, err := meddler.Default.Values(tx, false) + if err != nil { + return err + } + + q := fmt.Sprintf( + `INSERT INTO tx_pool (%s) + SELECT %s + WHERE (SELECT COUNT(*) FROM tx_pool WHERE state = $%v) < $%v;`, + namesPart, valuesPart, + len(values)+1, len(values)+2) //nolint:gomnd + values = append(values, common.PoolL2TxStatePending, l2db.maxTxs) + res, err := l2db.db.Exec(q, values...) + if err != nil { + return tracerr.Wrap(err) + } + rowsAffected, err := res.RowsAffected() + if err != nil { return tracerr.Wrap(err) } - if totalTxs >= l2db.maxTxs { - return tracerr.New( - "The pool is at full capacity. More transactions are not accepted currently", - ) + if rowsAffected == 0 { + return tracerr.Wrap(errPoolFull) } - return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", tx)) + return nil } // selectPoolTxAPI select part of queries to get PoolL2TxRead diff --git a/db/l2db/l2db.go b/db/l2db/l2db.go index 5af7dfa..518fc03 100644 --- a/db/l2db/l2db.go +++ b/db/l2db/l2db.go @@ -25,6 +25,7 @@ type L2DB struct { safetyPeriod common.BatchNum ttl time.Duration maxTxs uint32 // limit of txs that are accepted in the pool + minFeeUSD float64 apiConnCon *db.APIConnectionController } @@ -35,6 +36,7 @@ func NewL2DB( db *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, + minFeeUSD float64, TTL time.Duration, apiConnCon *db.APIConnectionController, ) *L2DB { @@ -43,6 +45,7 @@ func NewL2DB( safetyPeriod: safetyPeriod, ttl: TTL, maxTxs: maxTxs, + minFeeUSD: minFeeUSD, apiConnCon: apiConnCon, } } @@ -104,9 +107,8 @@ func (l2db *L2DB) UpdateTxsInfo(txs []common.PoolL2Tx) error { return nil } -// AddTxTest inserts a tx into the L2DB. This is useful for test purposes, -// but in production txs will only be inserted through the API -func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error { +// NewPoolL2TxWriteFromPoolL2Tx creates a new PoolL2TxWrite from a PoolL2Tx +func NewPoolL2TxWriteFromPoolL2Tx(tx *common.PoolL2Tx) *PoolL2TxWrite { // transform tx from *common.PoolL2Tx to PoolL2TxWrite insertTx := &PoolL2TxWrite{ TxID: tx.TxID, @@ -148,6 +150,13 @@ func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error { f := new(big.Float).SetInt(tx.Amount) amountF, _ := f.Float64() insertTx.AmountFloat = amountF + return insertTx +} + +// AddTxTest inserts a tx into the L2DB. This is useful for test purposes, +// but in production txs will only be inserted through the API +func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error { + insertTx := NewPoolL2TxWriteFromPoolL2Tx(tx) // insert tx return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", insertTx)) } @@ -158,7 +167,8 @@ tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce, tx_pool.state, tx_pool.info, tx_pool.signature, tx_pool.timestamp, rq_from_idx, rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount, tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type, -fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update +(fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f) / + (10.0 ^ token.decimals::NUMERIC) AS fee_usd, token.usd_update FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id ` // GetTx return the specified Tx in common.PoolL2Tx format diff --git a/db/l2db/l2db_test.go b/db/l2db/l2db_test.go index d1b0257..dad6a28 100644 --- a/db/l2db/l2db_test.go +++ b/db/l2db/l2db_test.go @@ -2,8 +2,7 @@ package l2db import ( "database/sql" - "math" - "math/big" + "fmt" "os" "testing" "time" @@ -21,12 +20,14 @@ import ( "github.com/stretchr/testify/require" ) +var decimals = uint64(3) +var tokenValue = 1.0 // The price update gives a value of 1.0 USD to the token var l2DB *L2DB var l2DBWithACC *L2DB var historyDB *historydb.HistoryDB var tc *til.Context var tokens map[common.TokenID]historydb.TokenWithUSD -var tokensValue map[common.TokenID]float64 + var accs map[common.Idx]common.Account func TestMain(m *testing.M) { @@ -36,9 +37,9 @@ func TestMain(m *testing.M) { if err != nil { panic(err) } - l2DB = NewL2DB(db, 10, 1000, 24*time.Hour, nil) + l2DB = NewL2DB(db, 10, 1000, 0.0, 24*time.Hour, nil) apiConnCon := dbUtils.NewAPICnnectionController(1, time.Second) - l2DBWithACC = NewL2DB(db, 10, 1000, 24*time.Hour, apiConnCon) + l2DBWithACC = NewL2DB(db, 10, 1000, 0.0, 24*time.Hour, apiConnCon) test.WipeDB(l2DB.DB()) historyDB = historydb.NewHistoryDB(db, nil) // Run tests @@ -59,10 +60,10 @@ func prepareHistoryDB(historyDB *historydb.HistoryDB) error { AddToken(1) AddToken(2) - CreateAccountDeposit(1) A: 2000 - CreateAccountDeposit(2) A: 2000 - CreateAccountDeposit(1) B: 1000 - CreateAccountDeposit(2) B: 1000 + CreateAccountDeposit(1) A: 20000 + CreateAccountDeposit(2) A: 20000 + CreateAccountDeposit(1) B: 10000 + CreateAccountDeposit(2) B: 10000 > batchL1 > batchL1 > block @@ -83,15 +84,23 @@ func prepareHistoryDB(historyDB *historydb.HistoryDB) error { if err != nil { return tracerr.Wrap(err) } + for i := range blocks { + block := &blocks[i] + for j := range block.Rollup.AddedTokens { + token := &block.Rollup.AddedTokens[j] + token.Name = fmt.Sprintf("Token %d", token.TokenID) + token.Symbol = fmt.Sprintf("TK%d", token.TokenID) + token.Decimals = decimals + } + } + tokens = make(map[common.TokenID]historydb.TokenWithUSD) - tokensValue = make(map[common.TokenID]float64) + // tokensValue = make(map[common.TokenID]float64) accs = make(map[common.Idx]common.Account) - value := 5 * 5.389329 now := time.Now().UTC() // Add all blocks except for the last one for i := range blocks[:len(blocks)-1] { - err = historyDB.AddBlockSCData(&blocks[i]) - if err != nil { + if err := historyDB.AddBlockSCData(&blocks[i]); err != nil { return tracerr.Wrap(err) } for _, batch := range blocks[i].Rollup.Batches { @@ -107,39 +116,38 @@ func prepareHistoryDB(historyDB *historydb.HistoryDB) error { Name: token.Name, Symbol: token.Symbol, Decimals: token.Decimals, + USD: &tokenValue, + USDUpdate: &now, } - tokensValue[token.TokenID] = value / math.Pow(10, float64(token.Decimals)) - readToken.USDUpdate = &now - readToken.USD = &value tokens[token.TokenID] = readToken - } - // Set value to the tokens (tokens have no symbol) - tokenSymbol := "" - err := historyDB.UpdateTokenValue(tokenSymbol, value) - if err != nil { - return tracerr.Wrap(err) + // Set value to the tokens + err := historyDB.UpdateTokenValue(readToken.Symbol, *readToken.USD) + if err != nil { + return tracerr.Wrap(err) + } } } return nil } func generatePoolL2Txs() ([]common.PoolL2Tx, error) { + // Fee = 126 corresponds to ~10% setPool := ` Type: PoolL2 - PoolTransfer(1) A-B: 6 (4) - PoolTransfer(2) A-B: 3 (1) - PoolTransfer(1) B-A: 5 (2) - PoolTransfer(2) B-A: 10 (3) - PoolTransfer(1) A-B: 7 (2) - PoolTransfer(2) A-B: 2 (1) - PoolTransfer(1) B-A: 8 (2) - PoolTransfer(2) B-A: 1 (1) - PoolTransfer(1) A-B: 3 (1) - PoolTransferToEthAddr(2) B-A: 5 (2) - PoolTransferToBJJ(2) B-A: 5 (2) - - PoolExit(1) A: 5 (2) - PoolExit(2) B: 3 (1) + PoolTransfer(1) A-B: 6000 (126) + PoolTransfer(2) A-B: 3000 (126) + PoolTransfer(1) B-A: 5000 (126) + PoolTransfer(2) B-A: 10000 (126) + PoolTransfer(1) A-B: 7000 (126) + PoolTransfer(2) A-B: 2000 (126) + PoolTransfer(1) B-A: 8000 (126) + PoolTransfer(2) B-A: 1000 (126) + PoolTransfer(1) A-B: 3000 (126) + PoolTransferToEthAddr(2) B-A: 5000 (126) + PoolTransferToBJJ(2) B-A: 5000 (126) + + PoolExit(1) A: 5000 (126) + PoolExit(2) B: 3000 (126) ` poolL2Txs, err := tc.GeneratePoolL2Txs(setPool) if err != nil { @@ -154,25 +162,74 @@ func TestAddTxTest(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) fetchedTx, err := l2DB.GetTx(poolL2Txs[i].TxID) - assert.NoError(t, err) + require.NoError(t, err) assertTx(t, &poolL2Txs[i], fetchedTx) nameZone, offset := fetchedTx.Timestamp.Zone() assert.Equal(t, "UTC", nameZone) assert.Equal(t, 0, offset) } } + +func TestAddTxAPI(t *testing.T) { + err := prepareHistoryDB(historyDB) + if err != nil { + log.Error("Error prepare historyDB", err) + } + + oldMaxTxs := l2DBWithACC.maxTxs + // set max number of pending txs that can be kept in the pool to 5 + l2DBWithACC.maxTxs = 5 + + poolL2Txs, err := generatePoolL2Txs() + txs := make([]*PoolL2TxWrite, len(poolL2Txs)) + for i := range poolL2Txs { + txs[i] = NewPoolL2TxWriteFromPoolL2Tx(&poolL2Txs[i]) + } + require.NoError(t, err) + require.GreaterOrEqual(t, len(poolL2Txs), 8) + for i := range txs[:5] { + err := l2DBWithACC.AddTxAPI(txs[i]) + require.NoError(t, err) + fetchedTx, err := l2DB.GetTx(poolL2Txs[i].TxID) + require.NoError(t, err) + assertTx(t, &poolL2Txs[i], fetchedTx) + nameZone, offset := fetchedTx.Timestamp.Zone() + assert.Equal(t, "UTC", nameZone) + assert.Equal(t, 0, offset) + } + err = l2DBWithACC.AddTxAPI(txs[5]) + assert.Equal(t, errPoolFull, tracerr.Unwrap(err)) + // reset maxTxs to original value + l2DBWithACC.maxTxs = oldMaxTxs + + // set minFeeUSD to a high value than the tx feeUSD to test the error + // of inserting a tx with lower than min fee + oldMinFeeUSD := l2DBWithACC.minFeeUSD + tx := txs[5] + feeAmount, err := common.CalcFeeAmount(tx.Amount, tx.Fee) + require.NoError(t, err) + feeAmountUSD := common.TokensToUSD(feeAmount, decimals, tokenValue) + // set minFeeUSD higher than the tx fee to trigger the error + l2DBWithACC.minFeeUSD = feeAmountUSD + 1 + err = l2DBWithACC.AddTxAPI(tx) + require.Error(t, err) + assert.Regexp(t, "tx.feeUSD (.*) < minFeeUSD (.*)", err.Error()) + // reset minFeeUSD to original value + l2DBWithACC.minFeeUSD = oldMinFeeUSD +} + func TestUpdateTxsInfo(t *testing.T) { err := prepareHistoryDB(historyDB) if err != nil { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) require.NoError(t, err) @@ -186,7 +243,7 @@ func TestUpdateTxsInfo(t *testing.T) { for i := range poolL2Txs { fetchedTx, err := l2DB.GetTx(poolL2Txs[i].TxID) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, "test", fetchedTx.Info) } } @@ -204,9 +261,8 @@ func assertTx(t *testing.T, expected, actual *common.PoolL2Tx) { assert.Less(t, token.USDUpdate.Unix()-3, actual.AbsoluteFeeUpdate.Unix()) expected.AbsoluteFeeUpdate = actual.AbsoluteFeeUpdate // Set expected fee - f := new(big.Float).SetInt(expected.Amount) - amountF, _ := f.Float64() - expected.AbsoluteFee = *token.USD * amountF * expected.Fee.Percentage() + amountUSD := common.TokensToUSD(expected.Amount, token.Decimals, *token.USD) + expected.AbsoluteFee = amountUSD * expected.Fee.Percentage() test.AssertUSD(t, &expected.AbsoluteFee, &actual.AbsoluteFee) } assert.Equal(t, expected, actual) @@ -231,19 +287,28 @@ func TestGetPending(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) var pendingTxs []*common.PoolL2Tx for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) pendingTxs = append(pendingTxs, &poolL2Txs[i]) } fetchedTxs, err := l2DB.GetPendingTxs() - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, len(pendingTxs), len(fetchedTxs)) for i := range fetchedTxs { assertTx(t, pendingTxs[i], &fetchedTxs[i]) } + // Check AbsoluteFee amount + for i := range fetchedTxs { + tx := &fetchedTxs[i] + feeAmount, err := common.CalcFeeAmount(tx.Amount, tx.Fee) + require.NoError(t, err) + feeAmountUSD := common.TokensToUSD(feeAmount, + tokens[tx.TokenID].Decimals, *tokens[tx.TokenID].USD) + assert.InEpsilon(t, feeAmountUSD, tx.AbsoluteFee, 0.01) + } } func TestStartForging(t *testing.T) { @@ -254,13 +319,13 @@ func TestStartForging(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) var startForgingTxIDs []common.TxID randomizer := 0 // Add txs to DB for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) if poolL2Txs[i].State == common.PoolL2TxStatePending && randomizer%2 == 0 { startForgingTxIDs = append(startForgingTxIDs, poolL2Txs[i].TxID) } @@ -268,11 +333,11 @@ func TestStartForging(t *testing.T) { } // Start forging txs err = l2DB.StartForging(startForgingTxIDs, fakeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Fetch txs and check that they've been updated correctly for _, id := range startForgingTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, common.PoolL2TxStateForging, fetchedTx.State) assert.Equal(t, &fakeBatchNum, fetchedTx.BatchNum) } @@ -286,13 +351,13 @@ func TestDoneForging(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) var startForgingTxIDs []common.TxID randomizer := 0 // Add txs to DB for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) if poolL2Txs[i].State == common.PoolL2TxStatePending && randomizer%2 == 0 { startForgingTxIDs = append(startForgingTxIDs, poolL2Txs[i].TxID) } @@ -300,7 +365,7 @@ func TestDoneForging(t *testing.T) { } // Start forging txs err = l2DB.StartForging(startForgingTxIDs, fakeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) var doneForgingTxIDs []common.TxID randomizer = 0 @@ -312,12 +377,12 @@ func TestDoneForging(t *testing.T) { } // Done forging txs err = l2DB.DoneForging(doneForgingTxIDs, fakeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Fetch txs and check that they've been updated correctly for _, id := range doneForgingTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, common.PoolL2TxStateForged, fetchedTx.State) assert.Equal(t, &fakeBatchNum, fetchedTx.BatchNum) } @@ -331,13 +396,13 @@ func TestInvalidate(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) var invalidTxIDs []common.TxID randomizer := 0 // Add txs to DB for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) if poolL2Txs[i].State != common.PoolL2TxStateInvalid && randomizer%2 == 0 { randomizer++ invalidTxIDs = append(invalidTxIDs, poolL2Txs[i].TxID) @@ -345,11 +410,11 @@ func TestInvalidate(t *testing.T) { } // Invalidate txs err = l2DB.InvalidateTxs(invalidTxIDs, fakeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Fetch txs and check that they've been updated correctly for _, id := range invalidTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, common.PoolL2TxStateInvalid, fetchedTx.State) assert.Equal(t, &fakeBatchNum, fetchedTx.BatchNum) } @@ -363,7 +428,7 @@ func TestInvalidateOldNonces(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) // Update Accounts currentNonce var updateAccounts []common.IdxNonce var currentNonce = common.Nonce(1) @@ -380,13 +445,13 @@ func TestInvalidateOldNonces(t *testing.T) { invalidTxIDs = append(invalidTxIDs, poolL2Txs[i].TxID) } err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) } // sanity check require.Greater(t, len(invalidTxIDs), 0) err = l2DB.InvalidateOldNonces(updateAccounts, fakeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Fetch txs and check that they've been updated correctly for _, id := range invalidTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) @@ -408,7 +473,7 @@ func TestReorg(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) reorgedTxIDs := []common.TxID{} nonReorgedTxIDs := []common.TxID{} @@ -419,7 +484,7 @@ func TestReorg(t *testing.T) { // Add txs to DB for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) if poolL2Txs[i].State == common.PoolL2TxStatePending && randomizer%2 == 0 { startForgingTxIDs = append(startForgingTxIDs, poolL2Txs[i].TxID) allTxRandomize = append(allTxRandomize, poolL2Txs[i].TxID) @@ -431,7 +496,7 @@ func TestReorg(t *testing.T) { } // Start forging txs err = l2DB.StartForging(startForgingTxIDs, lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) var doneForgingTxIDs []common.TxID randomizer = 0 @@ -456,22 +521,22 @@ func TestReorg(t *testing.T) { // Invalidate txs BEFORE reorgBatch --> nonReorg err = l2DB.InvalidateTxs(invalidTxIDs, lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) // Done forging txs in reorgBatch --> Reorg err = l2DB.DoneForging(doneForgingTxIDs, reorgBatch) - assert.NoError(t, err) + require.NoError(t, err) err = l2DB.Reorg(lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) for _, id := range reorgedTxIDs { tx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Nil(t, tx.BatchNum) assert.Equal(t, common.PoolL2TxStatePending, tx.State) } for _, id := range nonReorgedTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, lastValidBatch, *fetchedTx.BatchNum) } } @@ -488,7 +553,7 @@ func TestReorg2(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) reorgedTxIDs := []common.TxID{} nonReorgedTxIDs := []common.TxID{} @@ -499,7 +564,7 @@ func TestReorg2(t *testing.T) { // Add txs to DB for i := range poolL2Txs { err := l2DB.AddTxTest(&poolL2Txs[i]) - assert.NoError(t, err) + require.NoError(t, err) if poolL2Txs[i].State == common.PoolL2TxStatePending && randomizer%2 == 0 { startForgingTxIDs = append(startForgingTxIDs, poolL2Txs[i].TxID) allTxRandomize = append(allTxRandomize, poolL2Txs[i].TxID) @@ -511,7 +576,7 @@ func TestReorg2(t *testing.T) { } // Start forging txs err = l2DB.StartForging(startForgingTxIDs, lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) var doneForgingTxIDs []common.TxID randomizer = 0 @@ -533,22 +598,22 @@ func TestReorg2(t *testing.T) { } // Done forging txs BEFORE reorgBatch --> nonReorg err = l2DB.DoneForging(doneForgingTxIDs, lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) // Invalidate txs in reorgBatch --> Reorg err = l2DB.InvalidateTxs(invalidTxIDs, reorgBatch) - assert.NoError(t, err) + require.NoError(t, err) err = l2DB.Reorg(lastValidBatch) - assert.NoError(t, err) + require.NoError(t, err) for _, id := range reorgedTxIDs { tx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Nil(t, tx.BatchNum) assert.Equal(t, common.PoolL2TxStatePending, tx.State) } for _, id := range nonReorgedTxIDs { fetchedTx, err := l2DBWithACC.GetTxAPI(id) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, lastValidBatch, *fetchedTx.BatchNum) } } @@ -564,7 +629,7 @@ func TestPurge(t *testing.T) { var poolL2Tx []common.PoolL2Tx for i := 0; i < generateTx; i++ { poolL2TxAux, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) poolL2Tx = append(poolL2Tx, poolL2TxAux...) } @@ -591,7 +656,7 @@ func TestPurge(t *testing.T) { deletedIDs = append(deletedIDs, poolL2Tx[i].TxID) } err := l2DB.AddTxTest(&tx) - assert.NoError(t, err) + require.NoError(t, err) } // Set batchNum keeped txs for i := range keepedIDs { @@ -599,17 +664,17 @@ func TestPurge(t *testing.T) { "UPDATE tx_pool SET batch_num = $1 WHERE tx_id = $2;", safeBatchNum, keepedIDs[i], ) - assert.NoError(t, err) + require.NoError(t, err) } // Start forging txs and set batchNum err = l2DB.StartForging(doneForgingTxIDs, toDeleteBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Done forging txs and set batchNum err = l2DB.DoneForging(doneForgingTxIDs, toDeleteBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Invalidate txs and set batchNum err = l2DB.InvalidateTxs(invalidTxIDs, toDeleteBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Update timestamp of afterTTL txs deleteTimestamp := time.Unix(time.Now().UTC().Unix()-int64(l2DB.ttl.Seconds()+float64(4*time.Second)), 0) for _, id := range afterTTLIDs { @@ -618,12 +683,12 @@ func TestPurge(t *testing.T) { "UPDATE tx_pool SET timestamp = $1, state = $2 WHERE tx_id = $3;", deleteTimestamp, common.PoolL2TxStatePending, id, ) - assert.NoError(t, err) + require.NoError(t, err) } // Purge txs err = l2DB.Purge(safeBatchNum) - assert.NoError(t, err) + require.NoError(t, err) // Check results for _, id := range deletedIDs { _, err := l2DB.GetTx(id) @@ -631,7 +696,7 @@ func TestPurge(t *testing.T) { } for _, id := range keepedIDs { _, err := l2DB.GetTx(id) - assert.NoError(t, err) + require.NoError(t, err) } } @@ -645,10 +710,10 @@ func TestAuth(t *testing.T) { for i := 0; i < len(auths); i++ { // Add to the DB err := l2DB.AddAccountCreationAuth(auths[i]) - assert.NoError(t, err) + require.NoError(t, err) // Fetch from DB auth, err := l2DB.GetAccountCreationAuth(auths[i].EthAddr) - assert.NoError(t, err) + require.NoError(t, err) // Check fetched vs generated assert.Equal(t, auths[i].EthAddr, auth.EthAddr) assert.Equal(t, auths[i].BJJ, auth.BJJ) @@ -666,7 +731,7 @@ func TestAddGet(t *testing.T) { log.Error("Error prepare historyDB", err) } poolL2Txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) // We will work with only 3 txs require.GreaterOrEqual(t, len(poolL2Txs), 3) @@ -709,7 +774,7 @@ func TestPurgeByExternalDelete(t *testing.T) { log.Error("Error prepare historyDB", err) } txs, err := generatePoolL2Txs() - assert.NoError(t, err) + require.NoError(t, err) // We will work with 8 txs require.GreaterOrEqual(t, len(txs), 8) diff --git a/db/migrations/0001.sql b/db/migrations/0001.sql index 6e73402..47fcc8e 100644 --- a/db/migrations/0001.sql +++ b/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, + usd NUMERIC, -- value of a normalized token (divided by 10^decimals) usd_update TIMESTAMP WITHOUT TIME ZONE ); diff --git a/node/node.go b/node/node.go index c5e0899..50713cd 100644 --- a/node/node.go +++ b/node/node.go @@ -205,6 +205,7 @@ func NewNode(mode Mode, cfg *config.Node) (*Node, error) { db, cfg.Coordinator.L2DB.SafetyPeriod, cfg.Coordinator.L2DB.MaxTxs, + cfg.Coordinator.L2DB.MinFeeUSD, cfg.Coordinator.L2DB.TTL.Duration, apiConnCon, ) diff --git a/test/zkproof/flows_test.go b/test/zkproof/flows_test.go index ebaa74c..1f714a0 100644 --- a/test/zkproof/flows_test.go +++ b/test/zkproof/flows_test.go @@ -75,7 +75,7 @@ func initTxSelector(t *testing.T, chainID uint16, hermezContractAddr ethCommon.A pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) - l2DB := l2db.NewL2DB(db, 10, 100, 24*time.Hour, nil) + l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) dir, err := ioutil.TempDir("", "tmpSyncDB") require.NoError(t, err) diff --git a/txselector/txselector_test.go b/txselector/txselector_test.go index 13f3303..67a9c40 100644 --- a/txselector/txselector_test.go +++ b/txselector/txselector_test.go @@ -29,7 +29,7 @@ func initTest(t *testing.T, chainID uint16, hermezContractAddr ethCommon.Address pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) - l2DB := l2db.NewL2DB(db, 10, 100, 24*time.Hour, nil) + l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) dir, err := ioutil.TempDir("", "tmpdb") require.NoError(t, err)