arnaucube
arnaucube/hermez-nodePublic Self-hosted repository
Merge pull request #571 from hermeznetwork/feature/sql-read-write

Diferentiate SQL connections by read write
Eduard S eduard@iden3.io
Wed, 24 Feb 2021 15:26:43 +0100
commit

155e06484b5ad814c00a126bed281f22497f325a

parent

5cdcae47b80c8f5df9fccfe0e08830c7bb1b1a54

M api/api_test.go → api/api_test.go

@@ -201,7 +201,7 @@ if err != nil {

panic(err) } apiConnCon := db.NewAPICnnectionController(1, time.Second) - hdb := historydb.NewHistoryDB(database, apiConnCon) + hdb := historydb.NewHistoryDB(database, database, apiConnCon) if err != nil { panic(err) }

@@ -216,7 +216,7 @@ panic(err)

} }() // L2DB - l2DB := l2db.NewL2DB(database, 10, 1000, 0.0, 24*time.Hour, apiConnCon) + l2DB := l2db.NewL2DB(database, 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)

@@ -591,10 +591,10 @@ pass := os.Getenv("POSTGRES_PASS")

databaseTO, err := db.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) apiConnConTO := db.NewAPICnnectionController(1, 100*time.Millisecond) - hdbTO := historydb.NewHistoryDB(databaseTO, apiConnConTO) + hdbTO := historydb.NewHistoryDB(databaseTO, databaseTO, apiConnConTO) require.NoError(t, err) // L2DB - l2DBTO := l2db.NewL2DB(databaseTO, 10, 1000, 1.0, 24*time.Hour, apiConnConTO) + l2DBTO := l2db.NewL2DB(databaseTO, databaseTO, 10, 1000, 1.0, 24*time.Hour, apiConnConTO) // API apiGinTO := gin.Default()
M cli/node/cfg.buidler.toml → cli/node/cfg.buidler.toml

@@ -20,11 +20,16 @@ Path = "/tmp/iden3-test/hermez/statedb"

Keep = 256 [PostgreSQL] -Port = 5432 -Host = "localhost" -User = "hermez" -Password = "yourpasswordhere" -Name = "hermez" +PortWrite = 5432 +HostWrite = "localhost" +UserWrite = "hermez" +PasswordWrite = "yourpasswordhere" +NameWrite = "hermez" +# PortRead = 5432 +# HostRead = "localhost" +# UserRead = "hermez" +# PasswordRead = "yourpasswordhere" +# NameRead = "hermez" [Web3] URL = "http://localhost:8545"
M cli/node/main.go → cli/node/main.go

@@ -17,6 +17,7 @@ "github.com/hermeznetwork/hermez-node/log"

"github.com/hermeznetwork/hermez-node/node" "github.com/hermeznetwork/tracerr" "github.com/iden3/go-iden3-crypto/babyjub" + "github.com/jmoiron/sqlx" "github.com/urfave/cli/v2" )

@@ -90,11 +91,11 @@ return nil

} } db, err := dbUtils.ConnectSQLDB( - cfg.PostgreSQL.Port, - cfg.PostgreSQL.Host, - cfg.PostgreSQL.User, - cfg.PostgreSQL.Password, - cfg.PostgreSQL.Name, + cfg.PostgreSQL.PortWrite, + cfg.PostgreSQL.HostWrite, + cfg.PostgreSQL.UserWrite, + cfg.PostgreSQL.PasswordWrite, + cfg.PostgreSQL.NameWrite, ) if err != nil { return tracerr.Wrap(err)

@@ -151,17 +152,36 @@ cfg := _cfg.node

blockNum := c.Int64(flagBlock) log.Infof("Discarding all blocks up to block %v...", blockNum) - db, err := dbUtils.InitSQLDB( - cfg.PostgreSQL.Port, - cfg.PostgreSQL.Host, - cfg.PostgreSQL.User, - cfg.PostgreSQL.Password, - cfg.PostgreSQL.Name, + dbWrite, err := dbUtils.InitSQLDB( + cfg.PostgreSQL.PortWrite, + cfg.PostgreSQL.HostWrite, + cfg.PostgreSQL.UserWrite, + cfg.PostgreSQL.PasswordWrite, + cfg.PostgreSQL.NameWrite, ) if err != nil { return tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err)) } - historyDB := historydb.NewHistoryDB(db, nil) + var dbRead *sqlx.DB + if cfg.PostgreSQL.HostRead == "" { + dbRead = dbWrite + } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite { + return tracerr.Wrap(fmt.Errorf( + "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different", + )) + } else { + dbRead, err = dbUtils.InitSQLDB( + cfg.PostgreSQL.PortRead, + cfg.PostgreSQL.HostRead, + cfg.PostgreSQL.UserRead, + cfg.PostgreSQL.PasswordRead, + cfg.PostgreSQL.NameRead, + ) + if err != nil { + return tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err)) + } + } + historyDB := historydb.NewHistoryDB(dbRead, dbWrite, nil) if err := historyDB.Reorg(blockNum); err != nil { return tracerr.Wrap(fmt.Errorf("historyDB.Reorg: %w", err)) }

@@ -170,7 +190,7 @@ if err != nil {

return tracerr.Wrap(fmt.Errorf("historyDB.GetLastBatchNum: %w", err)) } l2DB := l2db.NewL2DB( - db, + dbRead, dbWrite, cfg.Coordinator.L2DB.SafetyPeriod, cfg.Coordinator.L2DB.MaxTxs, cfg.Coordinator.L2DB.MinFeeUSD,
M config/config.go → config/config.go

@@ -215,17 +215,30 @@ Path string `validate:"required"`

// Keep is the number of checkpoints to keep Keep int `validate:"required"` } `validate:"required"` + // It's possible to use diferentiated SQL connections for read/write. + // If the read configuration is not provided, the write one it's going to be used + // for both reads and writes PostgreSQL struct { - // Port of the PostgreSQL server - Port int `validate:"required"` - // Host of the PostgreSQL server - Host string `validate:"required"` - // User of the PostgreSQL server - User string `validate:"required"` - // Password of the PostgreSQL server - Password string `validate:"required"` - // Name of the PostgreSQL server database - Name string `validate:"required"` + // Port of the PostgreSQL write server + PortWrite int `validate:"required"` + // Host of the PostgreSQL write server + HostWrite string `validate:"required"` + // User of the PostgreSQL write server + UserWrite string `validate:"required"` + // Password of the PostgreSQL write server + PasswordWrite string `validate:"required"` + // Name of the PostgreSQL write server database + NameWrite string `validate:"required"` + // Port of the PostgreSQL read server + PortRead int + // Host of the PostgreSQL read server + HostRead string + // User of the PostgreSQL read server + UserRead string + // Password of the PostgreSQL read server + PasswordRead string + // Name of the PostgreSQL read server database + NameRead string } `validate:"required"` Web3 struct { // URL is the URL of the web3 ethereum-node RPC server
M coordinator/coordinator_test.go → coordinator/coordinator_test.go

@@ -105,8 +105,8 @@ pass := os.Getenv("POSTGRES_PASS")

db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) test.WipeDB(db) - l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) - historyDB := historydb.NewHistoryDB(db, nil) + l2DB := l2db.NewL2DB(db, db, 10, 100, 0.0, 24*time.Hour, nil) + historyDB := historydb.NewHistoryDB(db, db, nil) txSelDBPath, err = ioutil.TempDir("", "tmpTxSelDB") require.NoError(t, err)
M coordinator/purger_test.go → coordinator/purger_test.go

@@ -21,7 +21,7 @@ pass := os.Getenv("POSTGRES_PASS")

db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) test.WipeDB(db) - return l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) + return l2db.NewL2DB(db, db, 10, 100, 0.0, 24*time.Hour, nil) } func newStateDB(t *testing.T) *statedb.LocalStateDB {
M db/historydb/apiqueries.go → db/historydb/apiqueries.go

@@ -34,7 +34,7 @@ }

defer hdb.apiConnCon.Release() batch := &BatchAPI{} return batch, tracerr.Wrap(meddler.QueryRow( - hdb.db, batch, + hdb.dbRead, batch, `SELECT batch.item_id, batch.batch_num, batch.eth_block_num, batch.forger_addr, batch.fees_collected, batch.total_fees_usd, batch.state_root, batch.num_accounts, batch.exit_root, batch.forge_l1_txs_num, batch.slot_num,

@@ -133,10 +133,10 @@ } else {

queryStr += " DESC " } queryStr += fmt.Sprintf("LIMIT %d;", *limit) - query = hdb.db.Rebind(queryStr) + query = hdb.dbRead.Rebind(queryStr) // log.Debug(query) batchPtrs := []*BatchAPI{} - if err := meddler.QueryAll(hdb.db, &batchPtrs, query, args...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &batchPtrs, query, args...); err != nil { return nil, 0, tracerr.Wrap(err) } batches := db.SlicePtrsToSlice(batchPtrs).([]BatchAPI)

@@ -156,7 +156,7 @@ return *bid, tracerr.Wrap(err)

} defer hdb.apiConnCon.Release() err = meddler.QueryRow( - hdb.db, bid, `SELECT bid.*, block.timestamp, coordinator.forger_addr, coordinator.url + hdb.dbRead, bid, `SELECT bid.*, block.timestamp, coordinator.forger_addr, coordinator.url FROM bid INNER JOIN block ON bid.eth_block_num = block.eth_block_num INNER JOIN ( SELECT bidder_addr, MAX(item_id) AS item_id FROM coordinator

@@ -212,9 +212,9 @@ }

if limit != nil { queryStr += fmt.Sprintf("LIMIT %d;", *limit) } - query = hdb.db.Rebind(queryStr) + query = hdb.dbRead.Rebind(queryStr) bidPtrs := []*BidAPI{} - if err := meddler.QueryAll(hdb.db, &bidPtrs, query, args...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &bidPtrs, query, args...); err != nil { return nil, 0, tracerr.Wrap(err) } // log.Debug(query)

@@ -296,9 +296,9 @@ query, argsQ, err := sqlx.In(queryStr, args...)

if err != nil { return nil, 0, tracerr.Wrap(err) } - query = hdb.db.Rebind(query) + query = hdb.dbRead.Rebind(query) bids := []*BidAPI{} - if err := meddler.QueryAll(hdb.db, &bids, query, argsQ...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &bids, query, argsQ...); err != nil { return nil, 0, tracerr.Wrap(err) } if len(bids) == 0 {

@@ -384,9 +384,9 @@ query, argsQ, err := sqlx.In(queryStr, args...)

if err != nil { return nil, 0, tracerr.Wrap(err) } - query = hdb.db.Rebind(query) + query = hdb.dbRead.Rebind(query) tokens := []*TokenWithUSD{} - if err := meddler.QueryAll(hdb.db, &tokens, query, argsQ...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &tokens, query, argsQ...); err != nil { return nil, 0, tracerr.Wrap(err) } if len(tokens) == 0 {

@@ -408,7 +408,7 @@ }

defer hdb.apiConnCon.Release() tx := &TxAPI{} err = meddler.QueryRow( - hdb.db, tx, `SELECT tx.item_id, tx.is_l1, tx.id, tx.type, tx.position, + hdb.dbRead, tx, `SELECT tx.item_id, tx.is_l1, tx.id, tx.type, tx.position, hez_idx(tx.effective_from_idx, token.symbol) AS from_idx, tx.from_eth_addr, tx.from_bjj, hez_idx(tx.to_idx, token.symbol) AS to_idx, tx.to_eth_addr, tx.to_bjj, tx.amount, tx.amount_success, tx.token_id, tx.amount_usd,

@@ -541,10 +541,10 @@ } else {

queryStr += " DESC " } queryStr += fmt.Sprintf("LIMIT %d;", *limit) - query = hdb.db.Rebind(queryStr) + query = hdb.dbRead.Rebind(queryStr) // log.Debug(query) txsPtrs := []*TxAPI{} - if err := meddler.QueryAll(hdb.db, &txsPtrs, query, args...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &txsPtrs, query, args...); err != nil { return nil, 0, tracerr.Wrap(err) } txs := db.SlicePtrsToSlice(txsPtrs).([]TxAPI)

@@ -564,7 +564,7 @@ }

defer hdb.apiConnCon.Release() exit := &ExitAPI{} err = meddler.QueryRow( - hdb.db, exit, `SELECT exit_tree.item_id, exit_tree.batch_num, + hdb.dbRead, exit, `SELECT exit_tree.item_id, exit_tree.batch_num, hez_idx(exit_tree.account_idx, token.symbol) AS account_idx, account.bjj, account.eth_addr, exit_tree.merkle_proof, exit_tree.balance, exit_tree.instant_withdrawn,

@@ -685,10 +685,10 @@ } else {

queryStr += " DESC " } queryStr += fmt.Sprintf("LIMIT %d;", *limit) - query = hdb.db.Rebind(queryStr) + query = hdb.dbRead.Rebind(queryStr) // log.Debug(query) exits := []*ExitAPI{} - if err := meddler.QueryAll(hdb.db, &exits, query, args...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &exits, query, args...); err != nil { return nil, 0, tracerr.Wrap(err) } if len(exits) == 0 {

@@ -707,7 +707,7 @@ }

defer hdb.apiConnCon.Release() var bucketUpdates []*BucketUpdateAPI err = meddler.QueryAll( - hdb.db, &bucketUpdates, + hdb.dbRead, &bucketUpdates, `SELECT num_bucket, withdrawals FROM bucket_update WHERE item_id in(SELECT max(item_id) FROM bucket_update group by num_bucket)

@@ -772,10 +772,10 @@ } else {

queryStr += " DESC " } queryStr += fmt.Sprintf("LIMIT %d;", *limit) - query = hdb.db.Rebind(queryStr) + query = hdb.dbRead.Rebind(queryStr) coordinators := []*CoordinatorAPI{} - if err := meddler.QueryAll(hdb.db, &coordinators, query, args...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &coordinators, query, args...); err != nil { return nil, 0, tracerr.Wrap(err) } if len(coordinators) == 0 {

@@ -795,7 +795,7 @@ }

defer hdb.apiConnCon.Release() auctionVars := &common.AuctionVariables{} err = meddler.QueryRow( - hdb.db, auctionVars, `SELECT * FROM auction_vars;`, + hdb.dbRead, auctionVars, `SELECT * FROM auction_vars;`, ) return auctionVars, tracerr.Wrap(err) }

@@ -816,7 +816,7 @@ WHERE default_slot_set_bid_slot_num < $1

ORDER BY default_slot_set_bid_slot_num DESC LIMIT $2; ` - err = meddler.QueryAll(hdb.db, &auctionVars, query, slotNum, maxItems) + err = meddler.QueryAll(hdb.dbRead, &auctionVars, query, slotNum, maxItems) if err != nil { return nil, tracerr.Wrap(err) }

@@ -832,7 +832,7 @@ return nil, tracerr.Wrap(err)

} defer hdb.apiConnCon.Release() account := &AccountAPI{} - err = meddler.QueryRow(hdb.db, account, `SELECT account.item_id, hez_idx(account.idx, + err = meddler.QueryRow(hdb.dbRead, 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,

@@ -927,10 +927,10 @@ query, argsQ, err := sqlx.In(queryStr, args...)

if err != nil { return nil, 0, tracerr.Wrap(err) } - query = hdb.db.Rebind(query) + query = hdb.dbRead.Rebind(query) accounts := []*AccountAPI{} - if err := meddler.QueryAll(hdb.db, &accounts, query, argsQ...); err != nil { + if err := meddler.QueryAll(hdb.dbRead, &accounts, query, argsQ...); err != nil { return nil, 0, tracerr.Wrap(err) } if len(accounts) == 0 {

@@ -952,7 +952,7 @@ defer hdb.apiConnCon.Release()

metricsTotals := &MetricsTotals{} metrics := &Metrics{} err = meddler.QueryRow( - hdb.db, metricsTotals, `SELECT COUNT(tx.*) as total_txs, + hdb.dbRead, metricsTotals, `SELECT COUNT(tx.*) as total_txs, COALESCE (MIN(tx.batch_num), 0) as batch_num, COALESCE (MIN(block.timestamp), NOW()) AS min_timestamp, COALESCE (MAX(block.timestamp), NOW()) AS max_timestamp FROM tx INNER JOIN block ON tx.eth_block_num = block.eth_block_num

@@ -977,7 +977,7 @@ metrics.TransactionsPerBatch = float64(0)

} err = meddler.QueryRow( - hdb.db, metricsTotals, `SELECT COUNT(*) AS total_batches, + hdb.dbRead, metricsTotals, `SELECT COUNT(*) AS total_batches, COALESCE (SUM(total_fees_usd), 0) AS total_fees FROM batch WHERE batch_num > $1;`, metricsTotals.FirstBatchNum) if err != nil {

@@ -994,7 +994,7 @@ } else {

metrics.AvgTransactionFee = 0 } err = meddler.QueryRow( - hdb.db, metrics, + hdb.dbRead, metrics, `SELECT COUNT(*) AS total_bjjs, COUNT(DISTINCT(bjj)) AS total_accounts FROM account;`) if err != nil { return nil, tracerr.Wrap(err)

@@ -1013,7 +1013,7 @@ }

defer hdb.apiConnCon.Release() metricsTotals := &MetricsTotals{} err = meddler.QueryRow( - hdb.db, metricsTotals, `SELECT COUNT(tx.*) as total_txs, + hdb.dbRead, metricsTotals, `SELECT COUNT(tx.*) as total_txs, COALESCE (MIN(tx.batch_num), 0) as batch_num FROM tx INNER JOIN block ON tx.eth_block_num = block.eth_block_num WHERE block.timestamp >= NOW() - INTERVAL '1 HOURS';`)

@@ -1021,7 +1021,7 @@ if err != nil {

return 0, tracerr.Wrap(err) } err = meddler.QueryRow( - hdb.db, metricsTotals, `SELECT COUNT(*) AS total_batches, + hdb.dbRead, metricsTotals, `SELECT COUNT(*) AS total_batches, COALESCE (SUM(total_fees_usd), 0) AS total_fees FROM batch WHERE batch_num > $1;`, metricsTotals.FirstBatchNum) if err != nil {

@@ -1048,7 +1048,7 @@ }

defer hdb.apiConnCon.Release() account := &common.Account{} err = meddler.QueryRow( - hdb.db, account, `SELECT * FROM account WHERE idx = $1;`, idx, + hdb.dbRead, account, `SELECT * FROM account WHERE idx = $1;`, idx, ) return account, tracerr.Wrap(err) }
M db/historydb/historydb.go → db/historydb/historydb.go

@@ -27,30 +27,35 @@ // TODO(Edu): Document here how HistoryDB is kept consistent

// HistoryDB persist the historic of the rollup type HistoryDB struct { - db *sqlx.DB + dbRead *sqlx.DB + dbWrite *sqlx.DB apiConnCon *db.APIConnectionController } // NewHistoryDB initialize the DB -func NewHistoryDB(db *sqlx.DB, apiConnCon *db.APIConnectionController) *HistoryDB { - return &HistoryDB{db: db, apiConnCon: apiConnCon} +func NewHistoryDB(dbRead, dbWrite *sqlx.DB, apiConnCon *db.APIConnectionController) *HistoryDB { + return &HistoryDB{ + dbRead: dbRead, + dbWrite: dbWrite, + apiConnCon: apiConnCon, + } } // DB returns a pointer to the L2DB.db. This method should be used only for // internal testing purposes. func (hdb *HistoryDB) DB() *sqlx.DB { - return hdb.db + return hdb.dbWrite } // AddBlock insert a block into the DB -func (hdb *HistoryDB) AddBlock(block *common.Block) error { return hdb.addBlock(hdb.db, block) } +func (hdb *HistoryDB) AddBlock(block *common.Block) error { return hdb.addBlock(hdb.dbWrite, block) } func (hdb *HistoryDB) addBlock(d meddler.DB, block *common.Block) error { return tracerr.Wrap(meddler.Insert(d, "block", block)) } // AddBlocks inserts blocks into the DB func (hdb *HistoryDB) AddBlocks(blocks []common.Block) error { - return tracerr.Wrap(hdb.addBlocks(hdb.db, blocks)) + return tracerr.Wrap(hdb.addBlocks(hdb.dbWrite, blocks)) } func (hdb *HistoryDB) addBlocks(d meddler.DB, blocks []common.Block) error {

@@ -69,7 +74,7 @@ // GetBlock retrieve a block from the DB, given a block number

func (hdb *HistoryDB) GetBlock(blockNum int64) (*common.Block, error) { block := &common.Block{} err := meddler.QueryRow( - hdb.db, block, + hdb.dbRead, block, "SELECT * FROM block WHERE eth_block_num = $1;", blockNum, ) return block, tracerr.Wrap(err)

@@ -79,7 +84,7 @@ // GetAllBlocks retrieve all blocks from the DB

func (hdb *HistoryDB) GetAllBlocks() ([]common.Block, error) { var blocks []*common.Block err := meddler.QueryAll( - hdb.db, &blocks, + hdb.dbRead, &blocks, "SELECT * FROM block ORDER BY eth_block_num;", ) return db.SlicePtrsToSlice(blocks).([]common.Block), tracerr.Wrap(err)

@@ -89,7 +94,7 @@ // getBlocks retrieve blocks from the DB, given a range of block numbers defined by from and to

func (hdb *HistoryDB) getBlocks(from, to int64) ([]common.Block, error) { var blocks []*common.Block err := meddler.QueryAll( - hdb.db, &blocks, + hdb.dbRead, &blocks, "SELECT * FROM block WHERE $1 <= eth_block_num AND eth_block_num < $2 ORDER BY eth_block_num;", from, to, )

@@ -100,13 +105,13 @@ // GetLastBlock retrieve the block with the highest block number from the DB

func (hdb *HistoryDB) GetLastBlock() (*common.Block, error) { block := &common.Block{} err := meddler.QueryRow( - hdb.db, block, "SELECT * FROM block ORDER BY eth_block_num DESC LIMIT 1;", + hdb.dbRead, block, "SELECT * FROM block ORDER BY eth_block_num DESC LIMIT 1;", ) return block, tracerr.Wrap(err) } // AddBatch insert a Batch into the DB -func (hdb *HistoryDB) AddBatch(batch *common.Batch) error { return hdb.addBatch(hdb.db, batch) } +func (hdb *HistoryDB) AddBatch(batch *common.Batch) error { return hdb.addBatch(hdb.dbWrite, batch) } func (hdb *HistoryDB) addBatch(d meddler.DB, batch *common.Batch) error { // Calculate total collected fees in USD // Get IDs of collected tokens for fees

@@ -129,9 +134,9 @@ )

if err != nil { return tracerr.Wrap(err) } - query = hdb.db.Rebind(query) + query = hdb.dbWrite.Rebind(query) if err := meddler.QueryAll( - hdb.db, &tokenPrices, query, args..., + hdb.dbWrite, &tokenPrices, query, args..., ); err != nil { return tracerr.Wrap(err) }

@@ -153,7 +158,7 @@ }

// AddBatches insert Bids into the DB func (hdb *HistoryDB) AddBatches(batches []common.Batch) error { - return tracerr.Wrap(hdb.addBatches(hdb.db, batches)) + return tracerr.Wrap(hdb.addBatches(hdb.dbWrite, batches)) } func (hdb *HistoryDB) addBatches(d meddler.DB, batches []common.Batch) error { for i := 0; i < len(batches); i++ {

@@ -168,7 +173,7 @@ // GetBatch returns the batch with the given batchNum

func (hdb *HistoryDB) GetBatch(batchNum common.BatchNum) (*common.Batch, error) { var batch common.Batch err := meddler.QueryRow( - hdb.db, &batch, `SELECT batch.batch_num, batch.eth_block_num, batch.forger_addr, + hdb.dbRead, &batch, `SELECT batch.batch_num, batch.eth_block_num, batch.forger_addr, batch.fees_collected, batch.fee_idxs_coordinator, batch.state_root, batch.num_accounts, batch.last_idx, batch.exit_root, batch.forge_l1_txs_num, batch.slot_num, batch.total_fees_usd FROM batch WHERE batch_num = $1;`,

@@ -181,7 +186,7 @@ // GetAllBatches retrieve all batches from the DB

func (hdb *HistoryDB) GetAllBatches() ([]common.Batch, error) { var batches []*common.Batch err := meddler.QueryAll( - hdb.db, &batches, + hdb.dbRead, &batches, `SELECT batch.batch_num, batch.eth_block_num, batch.forger_addr, batch.fees_collected, batch.fee_idxs_coordinator, batch.state_root, batch.num_accounts, batch.last_idx, batch.exit_root, batch.forge_l1_txs_num, batch.slot_num, batch.total_fees_usd FROM batch

@@ -194,7 +199,7 @@ // GetBatches retrieve batches from the DB, given a range of batch numbers defined by from and to

func (hdb *HistoryDB) GetBatches(from, to common.BatchNum) ([]common.Batch, error) { var batches []*common.Batch err := meddler.QueryAll( - hdb.db, &batches, + hdb.dbRead, &batches, `SELECT batch_num, eth_block_num, forger_addr, fees_collected, fee_idxs_coordinator, state_root, num_accounts, last_idx, exit_root, forge_l1_txs_num, slot_num, total_fees_usd FROM batch WHERE $1 <= batch_num AND batch_num < $2 ORDER BY batch_num;`,

@@ -206,7 +211,7 @@

// GetFirstBatchBlockNumBySlot returns the ethereum block number of the first // batch within a slot func (hdb *HistoryDB) GetFirstBatchBlockNumBySlot(slotNum int64) (int64, error) { - row := hdb.db.QueryRow( + row := hdb.dbRead.QueryRow( `SELECT eth_block_num FROM batch WHERE slot_num = $1 ORDER BY batch_num ASC LIMIT 1;`, slotNum, )

@@ -216,7 +221,7 @@ }

// GetLastBatchNum returns the BatchNum of the latest forged batch func (hdb *HistoryDB) GetLastBatchNum() (common.BatchNum, error) { - row := hdb.db.QueryRow("SELECT batch_num FROM batch ORDER BY batch_num DESC LIMIT 1;") + row := hdb.dbRead.QueryRow("SELECT batch_num FROM batch ORDER BY batch_num DESC LIMIT 1;") var batchNum common.BatchNum return batchNum, tracerr.Wrap(row.Scan(&batchNum)) }

@@ -225,7 +230,7 @@ // GetLastBatch returns the last forged batch

func (hdb *HistoryDB) GetLastBatch() (*common.Batch, error) { var batch common.Batch err := meddler.QueryRow( - hdb.db, &batch, `SELECT batch.batch_num, batch.eth_block_num, batch.forger_addr, + hdb.dbRead, &batch, `SELECT batch.batch_num, batch.eth_block_num, batch.forger_addr, batch.fees_collected, batch.fee_idxs_coordinator, batch.state_root, batch.num_accounts, batch.last_idx, batch.exit_root, batch.forge_l1_txs_num, batch.slot_num, batch.total_fees_usd FROM batch ORDER BY batch_num DESC LIMIT 1;`,

@@ -235,7 +240,7 @@ }

// GetLastL1BatchBlockNum returns the blockNum of the latest forged l1Batch func (hdb *HistoryDB) GetLastL1BatchBlockNum() (int64, error) { - row := hdb.db.QueryRow(`SELECT eth_block_num FROM batch + row := hdb.dbRead.QueryRow(`SELECT eth_block_num FROM batch WHERE forge_l1_txs_num IS NOT NULL ORDER BY batch_num DESC LIMIT 1;`) var blockNum int64

@@ -245,7 +250,7 @@

// GetLastL1TxsNum returns the greatest ForgeL1TxsNum in the DB from forged // batches. If there's no batch in the DB (nil, nil) is returned. func (hdb *HistoryDB) GetLastL1TxsNum() (*int64, error) { - row := hdb.db.QueryRow("SELECT MAX(forge_l1_txs_num) FROM batch;") + row := hdb.dbRead.QueryRow("SELECT MAX(forge_l1_txs_num) FROM batch;") lastL1TxsNum := new(int64) return lastL1TxsNum, tracerr.Wrap(row.Scan(&lastL1TxsNum)) }

@@ -256,15 +261,15 @@ // deleted.

func (hdb *HistoryDB) Reorg(lastValidBlock int64) error { var err error if lastValidBlock < 0 { - _, err = hdb.db.Exec("DELETE FROM block;") + _, err = hdb.dbWrite.Exec("DELETE FROM block;") } else { - _, err = hdb.db.Exec("DELETE FROM block WHERE eth_block_num > $1;", lastValidBlock) + _, err = hdb.dbWrite.Exec("DELETE FROM block WHERE eth_block_num > $1;", lastValidBlock) } return tracerr.Wrap(err) } // AddBids insert Bids into the DB -func (hdb *HistoryDB) AddBids(bids []common.Bid) error { return hdb.addBids(hdb.db, bids) } +func (hdb *HistoryDB) AddBids(bids []common.Bid) error { return hdb.addBids(hdb.dbWrite, bids) } func (hdb *HistoryDB) addBids(d meddler.DB, bids []common.Bid) error { if len(bids) == 0 { return nil

@@ -281,7 +286,7 @@ // GetAllBids retrieve all bids from the DB

func (hdb *HistoryDB) GetAllBids() ([]common.Bid, error) { var bids []*common.Bid err := meddler.QueryAll( - hdb.db, &bids, + hdb.dbRead, &bids, `SELECT bid.slot_num, bid.bid_value, bid.eth_block_num, bid.bidder_addr FROM bid ORDER BY item_id;`, )

@@ -292,7 +297,7 @@ // GetBestBidCoordinator returns the forger address of the highest bidder in a slot by slotNum

func (hdb *HistoryDB) GetBestBidCoordinator(slotNum int64) (*common.BidCoordinator, error) { bidCoord := &common.BidCoordinator{} err := meddler.QueryRow( - hdb.db, bidCoord, + hdb.dbRead, bidCoord, `SELECT ( SELECT default_slot_set_bid FROM auction_vars

@@ -315,7 +320,7 @@ }

// AddCoordinators insert Coordinators into the DB func (hdb *HistoryDB) AddCoordinators(coordinators []common.Coordinator) error { - return tracerr.Wrap(hdb.addCoordinators(hdb.db, coordinators)) + return tracerr.Wrap(hdb.addCoordinators(hdb.dbWrite, coordinators)) } func (hdb *HistoryDB) addCoordinators(d meddler.DB, coordinators []common.Coordinator) error { if len(coordinators) == 0 {

@@ -330,7 +335,7 @@ }

// AddExitTree insert Exit tree into the DB func (hdb *HistoryDB) AddExitTree(exitTree []common.ExitInfo) error { - return tracerr.Wrap(hdb.addExitTree(hdb.db, exitTree)) + return tracerr.Wrap(hdb.addExitTree(hdb.dbWrite, exitTree)) } func (hdb *HistoryDB) addExitTree(d meddler.DB, exitTree []common.ExitInfo) error { if len(exitTree) == 0 {

@@ -418,11 +423,13 @@ }

// AddToken insert a token into the DB func (hdb *HistoryDB) AddToken(token *common.Token) error { - return tracerr.Wrap(meddler.Insert(hdb.db, "token", token)) + return tracerr.Wrap(meddler.Insert(hdb.dbWrite, "token", token)) } // AddTokens insert tokens into the DB -func (hdb *HistoryDB) AddTokens(tokens []common.Token) error { return hdb.addTokens(hdb.db, tokens) } +func (hdb *HistoryDB) AddTokens(tokens []common.Token) error { + return hdb.addTokens(hdb.dbWrite, tokens) +} func (hdb *HistoryDB) addTokens(d meddler.DB, tokens []common.Token) error { if len(tokens) == 0 { return nil

@@ -453,7 +460,7 @@ func (hdb *HistoryDB) UpdateTokenValue(tokenSymbol string, value float64) error {

// Sanitize symbol tokenSymbol = strings.ToValidUTF8(tokenSymbol, " ") - _, err := hdb.db.Exec( + _, err := hdb.dbWrite.Exec( "UPDATE token SET usd = $1 WHERE symbol = $2;", value, tokenSymbol, )

@@ -464,7 +471,7 @@ // GetToken returns a token from the DB given a TokenID

func (hdb *HistoryDB) GetToken(tokenID common.TokenID) (*TokenWithUSD, error) { token := &TokenWithUSD{} err := meddler.QueryRow( - hdb.db, token, `SELECT * FROM token WHERE token_id = $1;`, tokenID, + hdb.dbRead, token, `SELECT * FROM token WHERE token_id = $1;`, tokenID, ) return token, tracerr.Wrap(err) }

@@ -473,7 +480,7 @@ // GetAllTokens returns all tokens from the DB

func (hdb *HistoryDB) GetAllTokens() ([]TokenWithUSD, error) { var tokens []*TokenWithUSD err := meddler.QueryAll( - hdb.db, &tokens, + hdb.dbRead, &tokens, "SELECT * FROM token ORDER BY token_id;", ) return db.SlicePtrsToSlice(tokens).([]TokenWithUSD), tracerr.Wrap(err)

@@ -482,7 +489,7 @@

// GetTokenSymbols returns all the token symbols from the DB func (hdb *HistoryDB) GetTokenSymbols() ([]string, error) { var tokenSymbols []string - rows, err := hdb.db.Query("SELECT symbol FROM token;") + rows, err := hdb.dbRead.Query("SELECT symbol FROM token;") if err != nil { return nil, tracerr.Wrap(err) }

@@ -500,7 +507,7 @@ }

// AddAccounts insert accounts into the DB func (hdb *HistoryDB) AddAccounts(accounts []common.Account) error { - return tracerr.Wrap(hdb.addAccounts(hdb.db, accounts)) + return tracerr.Wrap(hdb.addAccounts(hdb.dbWrite, accounts)) } func (hdb *HistoryDB) addAccounts(d meddler.DB, accounts []common.Account) error { if len(accounts) == 0 {

@@ -523,7 +530,7 @@ // GetAllAccounts returns a list of accounts from the DB

func (hdb *HistoryDB) GetAllAccounts() ([]common.Account, error) { var accs []*common.Account err := meddler.QueryAll( - hdb.db, &accs, + hdb.dbRead, &accs, "SELECT idx, token_id, batch_num, bjj, eth_addr FROM account ORDER BY idx;", ) return db.SlicePtrsToSlice(accs).([]common.Account), tracerr.Wrap(err)

@@ -531,7 +538,7 @@ }

// AddAccountUpdates inserts accUpdates into the DB func (hdb *HistoryDB) AddAccountUpdates(accUpdates []common.AccountUpdate) error { - return tracerr.Wrap(hdb.addAccountUpdates(hdb.db, accUpdates)) + return tracerr.Wrap(hdb.addAccountUpdates(hdb.dbWrite, accUpdates)) } func (hdb *HistoryDB) addAccountUpdates(d meddler.DB, accUpdates []common.AccountUpdate) error { if len(accUpdates) == 0 {

@@ -554,7 +561,7 @@ // GetAllAccountUpdates returns all the AccountUpdate from the DB

func (hdb *HistoryDB) GetAllAccountUpdates() ([]common.AccountUpdate, error) { var accUpdates []*common.AccountUpdate err := meddler.QueryAll( - hdb.db, &accUpdates, + hdb.dbRead, &accUpdates, "SELECT eth_block_num, batch_num, idx, nonce, balance FROM account_update ORDER BY idx;", ) return db.SlicePtrsToSlice(accUpdates).([]common.AccountUpdate), tracerr.Wrap(err)

@@ -565,7 +572,7 @@ // If the tx is originated by a coordinator, BatchNum must be provided. If it's originated by a user,

// BatchNum should be null, and the value will be setted by a trigger when a batch forges the tx. // EffectiveAmount and EffectiveDepositAmount are seted with default values by the DB. func (hdb *HistoryDB) AddL1Txs(l1txs []common.L1Tx) error { - return tracerr.Wrap(hdb.addL1Txs(hdb.db, l1txs)) + return tracerr.Wrap(hdb.addL1Txs(hdb.dbWrite, l1txs)) } // addL1Txs inserts L1 txs to the DB. USD and DepositAmountUSD will be set automatically before storing the tx.

@@ -619,7 +626,7 @@ }

// AddL2Txs inserts L2 txs to the DB. TokenID, USD and FeeUSD will be set automatically before storing the tx. func (hdb *HistoryDB) AddL2Txs(l2txs []common.L2Tx) error { - return tracerr.Wrap(hdb.addL2Txs(hdb.db, l2txs)) + return tracerr.Wrap(hdb.addL2Txs(hdb.dbWrite, l2txs)) } // addL2Txs inserts L2 txs to the DB. TokenID, USD and FeeUSD will be set automatically before storing the tx.

@@ -686,7 +693,7 @@ // GetAllExits returns all exit from the DB

func (hdb *HistoryDB) GetAllExits() ([]common.ExitInfo, error) { var exits []*common.ExitInfo err := meddler.QueryAll( - hdb.db, &exits, + hdb.dbRead, &exits, `SELECT exit_tree.batch_num, exit_tree.account_idx, exit_tree.merkle_proof, exit_tree.balance, exit_tree.instant_withdrawn, exit_tree.delayed_withdraw_request, exit_tree.delayed_withdrawn FROM exit_tree ORDER BY item_id;`,

@@ -698,7 +705,7 @@ // GetAllL1UserTxs returns all L1UserTxs from the DB

func (hdb *HistoryDB) GetAllL1UserTxs() ([]common.L1Tx, error) { var txs []*common.L1Tx err := meddler.QueryAll( - hdb.db, &txs, // Note that '\x' gets parsed as a big.Int with value = 0 + hdb.dbRead, &txs, // Note that '\x' gets parsed as a big.Int with value = 0 `SELECT tx.id, tx.to_forge_l1_txs_num, tx.position, tx.user_origin, tx.from_idx, tx.effective_from_idx, tx.from_eth_addr, tx.from_bjj, tx.to_idx, tx.token_id, tx.amount, (CASE WHEN tx.batch_num IS NULL THEN NULL WHEN tx.amount_success THEN tx.amount ELSE '\x' END) AS effective_amount,

@@ -715,7 +722,7 @@ var txs []*common.L1Tx

// Since the query specifies that only coordinator txs are returned, it's safe to assume // that returned txs will always have effective amounts err := meddler.QueryAll( - hdb.db, &txs, + hdb.dbRead, &txs, `SELECT tx.id, tx.to_forge_l1_txs_num, tx.position, tx.user_origin, tx.from_idx, tx.effective_from_idx, tx.from_eth_addr, tx.from_bjj, tx.to_idx, tx.token_id, tx.amount, tx.amount AS effective_amount,

@@ -730,7 +737,7 @@ // GetAllL2Txs returns all L2Txs from the DB

func (hdb *HistoryDB) GetAllL2Txs() ([]common.L2Tx, error) { var txs []*common.L2Tx err := meddler.QueryAll( - hdb.db, &txs, + hdb.dbRead, &txs, `SELECT tx.id, tx.batch_num, tx.position, tx.from_idx, tx.to_idx, tx.amount, tx.token_id, tx.fee, tx.nonce, tx.type, tx.eth_block_num

@@ -743,7 +750,7 @@ // GetUnforgedL1UserTxs gets L1 User Txs to be forged in the L1Batch with toForgeL1TxsNum.

func (hdb *HistoryDB) GetUnforgedL1UserTxs(toForgeL1TxsNum int64) ([]common.L1Tx, error) { var txs []*common.L1Tx err := meddler.QueryAll( - hdb.db, &txs, // only L1 user txs can have batch_num set to null + hdb.dbRead, &txs, // only L1 user txs can have batch_num set to null `SELECT tx.id, tx.to_forge_l1_txs_num, tx.position, tx.user_origin, tx.from_idx, tx.from_eth_addr, tx.from_bjj, tx.to_idx, tx.token_id, tx.amount, NULL AS effective_amount,

@@ -760,7 +767,7 @@ // TODO: Think about chaning all the queries that return a last value, to queries that return the next valid value.

// GetLastTxsPosition for a given to_forge_l1_txs_num func (hdb *HistoryDB) GetLastTxsPosition(toForgeL1TxsNum int64) (int, error) { - row := hdb.db.QueryRow( + row := hdb.dbRead.QueryRow( "SELECT position FROM tx WHERE to_forge_l1_txs_num = $1 ORDER BY position DESC;", toForgeL1TxsNum, )

@@ -774,15 +781,15 @@ *common.WDelayerVariables, error) {

var rollup common.RollupVariables var auction common.AuctionVariables var wDelayer common.WDelayerVariables - if err := meddler.QueryRow(hdb.db, &rollup, + if err := meddler.QueryRow(hdb.dbRead, &rollup, "SELECT * FROM rollup_vars ORDER BY eth_block_num DESC LIMIT 1;"); err != nil { return nil, nil, nil, tracerr.Wrap(err) } - if err := meddler.QueryRow(hdb.db, &auction, + if err := meddler.QueryRow(hdb.dbRead, &auction, "SELECT * FROM auction_vars ORDER BY eth_block_num DESC LIMIT 1;"); err != nil { return nil, nil, nil, tracerr.Wrap(err) } - if err := meddler.QueryRow(hdb.db, &wDelayer, + if err := meddler.QueryRow(hdb.dbRead, &wDelayer, "SELECT * FROM wdelayer_vars ORDER BY eth_block_num DESC LIMIT 1;"); err != nil { return nil, nil, nil, tracerr.Wrap(err) }

@@ -827,7 +834,7 @@ // GetAllBucketUpdates retrieves all the bucket updates

func (hdb *HistoryDB) GetAllBucketUpdates() ([]common.BucketUpdate, error) { var bucketUpdates []*common.BucketUpdate err := meddler.QueryAll( - hdb.db, &bucketUpdates, + hdb.dbRead, &bucketUpdates, `SELECT eth_block_num, num_bucket, block_stamp, withdrawals FROM bucket_update ORDER BY item_id;`, )

@@ -853,7 +860,7 @@ // GetAllTokenExchanges retrieves all the token exchanges

func (hdb *HistoryDB) GetAllTokenExchanges() ([]common.TokenExchange, error) { var tokenExchanges []*common.TokenExchange err := meddler.QueryAll( - hdb.db, &tokenExchanges, + hdb.dbRead, &tokenExchanges, "SELECT eth_block_num, eth_addr, value_usd FROM token_exchange ORDER BY item_id;", ) return db.SlicePtrsToSlice(tokenExchanges).([]common.TokenExchange), tracerr.Wrap(err)

@@ -881,7 +888,7 @@ // GetAllEscapeHatchWithdrawals retrieves all the escape hatch withdrawals

func (hdb *HistoryDB) GetAllEscapeHatchWithdrawals() ([]common.WDelayerEscapeHatchWithdrawal, error) { var escapeHatchWithdrawals []*common.WDelayerEscapeHatchWithdrawal err := meddler.QueryAll( - hdb.db, &escapeHatchWithdrawals, + hdb.dbRead, &escapeHatchWithdrawals, "SELECT eth_block_num, who_addr, to_addr, token_addr, amount FROM escape_hatch_withdrawal ORDER BY item_id;", ) return db.SlicePtrsToSlice(escapeHatchWithdrawals).([]common.WDelayerEscapeHatchWithdrawal),

@@ -894,7 +901,7 @@ // always exist in the DB and is used to store initialization data that always

// exist in the smart contracts. func (hdb *HistoryDB) SetInitialSCVars(rollup *common.RollupVariables, auction *common.AuctionVariables, wDelayer *common.WDelayerVariables) error { - txn, err := hdb.db.Beginx() + txn, err := hdb.dbWrite.Beginx() if err != nil { return tracerr.Wrap(err) }

@@ -978,7 +985,7 @@ // Synchronizer. Blocks should be inserted in order, leaving no gaps because

// the pagination system of the API/DB depends on this. Within blocks, all // items should also be in the correct order (Accounts, Tokens, Txs, etc.) func (hdb *HistoryDB) AddBlockSCData(blockData *common.BlockData) (err error) { - txn, err := hdb.db.Beginx() + txn, err := hdb.dbWrite.Beginx() if err != nil { return tracerr.Wrap(err) }

@@ -1136,7 +1143,7 @@ // GetCoordinatorAPI returns a coordinator by its bidderAddr

func (hdb *HistoryDB) GetCoordinatorAPI(bidderAddr ethCommon.Address) (*CoordinatorAPI, error) { coordinator := &CoordinatorAPI{} err := meddler.QueryRow( - hdb.db, coordinator, + hdb.dbRead, coordinator, "SELECT * FROM coordinator WHERE bidder_addr = $1 ORDER BY item_id DESC LIMIT 1;", bidderAddr, )

@@ -1145,14 +1152,14 @@ }

// AddAuctionVars insert auction vars into the DB func (hdb *HistoryDB) AddAuctionVars(auctionVars *common.AuctionVariables) error { - return tracerr.Wrap(meddler.Insert(hdb.db, "auction_vars", auctionVars)) + return tracerr.Wrap(meddler.Insert(hdb.dbWrite, "auction_vars", auctionVars)) } // GetTokensTest used to get tokens in a testing context func (hdb *HistoryDB) GetTokensTest() ([]TokenWithUSD, error) { tokens := []*TokenWithUSD{} if err := meddler.QueryAll( - hdb.db, &tokens, + hdb.dbRead, &tokens, "SELECT * FROM TOKEN", ); err != nil { return nil, tracerr.Wrap(err)
M db/historydb/historydb_test.go → db/historydb/historydb_test.go

@@ -39,12 +39,12 @@ db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez")

if err != nil { panic(err) } - historyDB = NewHistoryDB(db, nil) + historyDB = NewHistoryDB(db, db, nil) if err != nil { panic(err) } apiConnCon := dbUtils.NewAPICnnectionController(1, time.Second) - historyDBWithACC = NewHistoryDB(db, apiConnCon) + historyDBWithACC = NewHistoryDB(db, db, apiConnCon) // Run tests result := m.Run() // Close DB

@@ -817,11 +817,11 @@ require.NoError(t, err)

} // Add second batch to trigger the update of the batch_num, // while avoiding the implicit call of setExtraInfoForgedL1UserTxs - err = historyDB.addBlock(historyDB.db, &blocks[1].Block) + err = historyDB.addBlock(historyDB.dbWrite, &blocks[1].Block) require.NoError(t, err) - err = historyDB.addBatch(historyDB.db, &blocks[1].Rollup.Batches[0].Batch) + err = historyDB.addBatch(historyDB.dbWrite, &blocks[1].Rollup.Batches[0].Batch) require.NoError(t, err) - err = historyDB.addAccounts(historyDB.db, blocks[1].Rollup.Batches[0].CreatedAccounts) + err = historyDB.addAccounts(historyDB.dbWrite, blocks[1].Rollup.Batches[0].CreatedAccounts) require.NoError(t, err) // Set the Effective{Amount,DepositAmount} of the L1UserTxs that are forged in the second block

@@ -831,7 +831,7 @@ // Change some values to test all cases

l1Txs[1].EffectiveAmount = big.NewInt(0) l1Txs[2].EffectiveDepositAmount = big.NewInt(0) l1Txs[2].EffectiveAmount = big.NewInt(0) - err = historyDB.setExtraInfoForgedL1UserTxs(historyDB.db, l1Txs) + err = historyDB.setExtraInfoForgedL1UserTxs(historyDB.dbWrite, l1Txs) require.NoError(t, err) dbL1Txs, err := historyDB.GetAllL1UserTxs()

@@ -918,10 +918,10 @@ common.WithdrawInfo{Idx: 258, NumExitRoot: 3, InstantWithdraw: true},

common.WithdrawInfo{Idx: 259, NumExitRoot: 3, InstantWithdraw: false, Owner: tc.UsersByIdx[259].Addr, Token: tokenAddr}, ) - err = historyDB.addBlock(historyDB.db, &block.Block) + err = historyDB.addBlock(historyDB.dbWrite, &block.Block) require.NoError(t, err) - err = historyDB.updateExitTree(historyDB.db, block.Block.Num, + err = historyDB.updateExitTree(historyDB.dbWrite, block.Block.Num, block.Rollup.Withdrawals, block.WDelayer.Withdrawals) require.NoError(t, err)

@@ -951,10 +951,10 @@ Owner: tc.UsersByIdx[257].Addr,

Token: tokenAddr, Amount: big.NewInt(80), }) - err = historyDB.addBlock(historyDB.db, &block.Block) + err = historyDB.addBlock(historyDB.dbWrite, &block.Block) require.NoError(t, err) - err = historyDB.updateExitTree(historyDB.db, block.Block.Num, + err = historyDB.updateExitTree(historyDB.dbWrite, block.Block.Num, block.Rollup.Withdrawals, block.WDelayer.Withdrawals) require.NoError(t, err)

@@ -997,7 +997,7 @@ EthBlockNum: 2,

URL: "bar", }, } - err = historyDB.addCoordinators(historyDB.db, coords) + err = historyDB.addCoordinators(historyDB.dbWrite, coords) require.NoError(t, err) bids := []common.Bid{

@@ -1015,7 +1015,7 @@ Bidder: coords[1].Bidder,

}, } - err = historyDB.addBids(historyDB.db, bids) + err = historyDB.addBids(historyDB.dbWrite, bids) require.NoError(t, err) forger10, err := historyDB.GetBestBidCoordinator(10)

@@ -1053,7 +1053,7 @@ BlockStamp: 5,

Withdrawals: big.NewInt(42), }, } - err := historyDB.addBucketUpdates(historyDB.db, bucketUpdates) + err := historyDB.addBucketUpdates(historyDB.dbWrite, bucketUpdates) require.NoError(t, err) dbBucketUpdates, err := historyDB.GetAllBucketUpdates() require.NoError(t, err)

@@ -1078,7 +1078,7 @@ Address: ethCommon.BigToAddress(big.NewInt(222)),

ValueUSD: 67890, }, } - err := historyDB.addTokenExchanges(historyDB.db, tokenExchanges) + err := historyDB.addTokenExchanges(historyDB.dbWrite, tokenExchanges) require.NoError(t, err) dbTokenExchanges, err := historyDB.GetAllTokenExchanges() require.NoError(t, err)

@@ -1107,7 +1107,7 @@ TokenAddr: ethCommon.BigToAddress(big.NewInt(666)),

Amount: big.NewInt(20003), }, } - err := historyDB.addEscapeHatchWithdrawals(historyDB.db, escapeHatchWithdrawals) + err := historyDB.addEscapeHatchWithdrawals(historyDB.dbWrite, escapeHatchWithdrawals) require.NoError(t, err) dbEscapeHatchWithdrawals, err := historyDB.GetAllEscapeHatchWithdrawals() require.NoError(t, err)
M db/l2db/apiqueries.go → db/l2db/apiqueries.go

@@ -34,7 +34,7 @@ }

defer l2db.apiConnCon.Release() auth := new(AccountCreationAuthAPI) return auth, tracerr.Wrap(meddler.QueryRow( - l2db.db, auth, + l2db.dbRead, auth, "SELECT * FROM account_creation_auth WHERE eth_addr = $1;", addr, ))

@@ -49,7 +49,7 @@ return tracerr.Wrap(err)

} defer l2db.apiConnCon.Release() - row := l2db.db.QueryRow(`SELECT + row := l2db.dbRead.QueryRow(`SELECT ($1::NUMERIC * token.usd * fee_percentage($2::NUMERIC)) / (10.0 ^ token.decimals::NUMERIC) FROM token WHERE token.token_id = $3;`,

@@ -84,7 +84,7 @@ 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...) + res, err := l2db.dbWrite.Exec(q, values...) if err != nil { return tracerr.Wrap(err) }

@@ -118,7 +118,7 @@ }

defer l2db.apiConnCon.Release() tx := new(PoolTxAPI) return tx, tracerr.Wrap(meddler.QueryRow( - l2db.db, tx, + l2db.dbRead, tx, selectPoolTxAPI+"WHERE tx_id = $1;", txID, ))
M db/l2db/l2db.go → db/l2db/l2db.go

@@ -21,7 +21,8 @@

// L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant // due to them being forged or invalid after a safety period type L2DB struct { - db *sqlx.DB + dbRead *sqlx.DB + dbWrite *sqlx.DB safetyPeriod common.BatchNum ttl time.Duration maxTxs uint32 // limit of txs that are accepted in the pool

@@ -33,7 +34,7 @@ // NewL2DB creates a L2DB.

// To create it, it's needed db connection, safety period expressed in batches, // maxTxs that the DB should have and TTL (time to live) for pending txs. func NewL2DB( - db *sqlx.DB, + dbRead, dbWrite *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, minFeeUSD float64,

@@ -41,7 +42,8 @@ TTL time.Duration,

apiConnCon *db.APIConnectionController, ) *L2DB { return &L2DB{ - db: db, + dbRead: dbRead, + dbWrite: dbWrite, safetyPeriod: safetyPeriod, ttl: TTL, maxTxs: maxTxs,

@@ -53,7 +55,7 @@

// DB returns a pointer to the L2DB.db. This method should be used only for // internal testing purposes. func (l2db *L2DB) DB() *sqlx.DB { - return l2db.db + return l2db.dbWrite } // MinFeeUSD returns the minimum fee in USD that is required to accept txs into

@@ -64,7 +66,7 @@ }

// AddAccountCreationAuth inserts an account creation authorization into the DB func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error { - _, err := l2db.db.Exec( + _, err := l2db.dbWrite.Exec( `INSERT INTO account_creation_auth (eth_addr, bjj, signature) VALUES ($1, $2, $3);`, auth.EthAddr, auth.BJJ, auth.Signature,

@@ -76,7 +78,7 @@ // GetAccountCreationAuth returns an account creation authorization from the DB

func (l2db *L2DB) GetAccountCreationAuth(addr ethCommon.Address) (*common.AccountCreationAuth, error) { auth := new(common.AccountCreationAuth) return auth, tracerr.Wrap(meddler.QueryRow( - l2db.db, auth, + l2db.dbRead, auth, "SELECT * FROM account_creation_auth WHERE eth_addr = $1;", addr, ))

@@ -105,7 +107,7 @@ ) as tx_update (id, info)

WHERE tx_pool.tx_id = tx_update.id; ` if len(txUpdates) > 0 { - if _, err := sqlx.NamedExec(l2db.db, query, txUpdates); err != nil { + if _, err := sqlx.NamedExec(l2db.dbWrite, query, txUpdates); err != nil { return tracerr.Wrap(err) } }

@@ -164,7 +166,7 @@ // 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)) + return tracerr.Wrap(meddler.Insert(l2db.dbWrite, "tx_pool", insertTx)) } // selectPoolTxCommon select part of queries to get common.PoolL2Tx

@@ -181,7 +183,7 @@ // GetTx return the specified Tx in common.PoolL2Tx format

func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) { tx := new(common.PoolL2Tx) return tx, tracerr.Wrap(meddler.QueryRow( - l2db.db, tx, + l2db.dbRead, tx, selectPoolTxCommon+"WHERE tx_id = $1;", txID, ))

@@ -191,7 +193,7 @@ // GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee

func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) { var txs []*common.PoolL2Tx err := meddler.QueryAll( - l2db.db, &txs, + l2db.dbRead, &txs, selectPoolTxCommon+"WHERE state = $1", common.PoolL2TxStatePending, )

@@ -216,8 +218,8 @@ )

if err != nil { return tracerr.Wrap(err) } - query = l2db.db.Rebind(query) - _, err = l2db.db.Exec(query, args...) + query = l2db.dbWrite.Rebind(query) + _, err = l2db.dbWrite.Exec(query, args...) return tracerr.Wrap(err) }

@@ -239,8 +241,8 @@ )

if err != nil { return tracerr.Wrap(err) } - query = l2db.db.Rebind(query) - _, err = l2db.db.Exec(query, args...) + query = l2db.dbWrite.Rebind(query) + _, err = l2db.dbWrite.Exec(query, args...) return tracerr.Wrap(err) }

@@ -261,8 +263,8 @@ )

if err != nil { return tracerr.Wrap(err) } - query = l2db.db.Rebind(query) - _, err = l2db.db.Exec(query, args...) + query = l2db.dbWrite.Rebind(query) + _, err = l2db.dbWrite.Exec(query, args...) return tracerr.Wrap(err) }

@@ -270,7 +272,7 @@ // GetPendingUniqueFromIdxs returns from all the pending transactions, the set

// of unique FromIdx func (l2db *L2DB) GetPendingUniqueFromIdxs() ([]common.Idx, error) { var idxs []common.Idx - rows, err := l2db.db.Query(`SELECT DISTINCT from_idx FROM tx_pool + rows, err := l2db.dbRead.Query(`SELECT DISTINCT from_idx FROM tx_pool WHERE state = $1;`, common.PoolL2TxStatePending) if err != nil { return nil, tracerr.Wrap(err)

@@ -311,7 +313,7 @@ // Fill the batch_num in the query with Sprintf because we are using a

// named query which works with slices, and doens't handle an extra // individual argument. query := fmt.Sprintf(invalidateOldNoncesQuery, batchNum) - if _, err := sqlx.NamedExec(l2db.db, query, updatedAccounts); err != nil { + if _, err := sqlx.NamedExec(l2db.dbWrite, query, updatedAccounts); err != nil { return tracerr.Wrap(err) } return nil

@@ -320,7 +322,7 @@

// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg. // The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error { - _, err := l2db.db.Exec( + _, err := l2db.dbWrite.Exec( `UPDATE tx_pool SET batch_num = NULL, state = $1 WHERE (state = $2 OR state = $3 OR state = $4) AND batch_num > $5`, common.PoolL2TxStatePending,

@@ -336,7 +338,7 @@ // Purge deletes transactions that have been forged or marked as invalid for longer than the safety period

// it also deletes pending txs that have been in the L2DB for longer than the ttl if maxTxs has been exceeded func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) { now := time.Now().UTC().Unix() - _, err = l2db.db.Exec( + _, err = l2db.dbWrite.Exec( `DELETE FROM tx_pool WHERE ( batch_num < $1 AND (state = $2 OR state = $3) ) OR (

@@ -357,7 +359,7 @@ // PurgeByExternalDelete deletes all pending transactions marked with true in

// the `external_delete` column. An external process can set this column to // true to instruct the coordinator to delete the tx when possible. func (l2db *L2DB) PurgeByExternalDelete() error { - _, err := l2db.db.Exec( + _, err := l2db.dbWrite.Exec( `DELETE from tx_pool WHERE (external_delete = true AND state = $1);`, common.PoolL2TxStatePending, )
M db/l2db/l2db_test.go → db/l2db/l2db_test.go

@@ -37,11 +37,11 @@ db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez")

if err != nil { panic(err) } - l2DB = NewL2DB(db, 10, 1000, 0.0, 24*time.Hour, nil) + l2DB = NewL2DB(db, db, 10, 1000, 0.0, 24*time.Hour, nil) apiConnCon := dbUtils.NewAPICnnectionController(1, time.Second) - l2DBWithACC = NewL2DB(db, 10, 1000, 0.0, 24*time.Hour, apiConnCon) + l2DBWithACC = NewL2DB(db, db, 10, 1000, 0.0, 24*time.Hour, apiConnCon) test.WipeDB(l2DB.DB()) - historyDB = historydb.NewHistoryDB(db, nil) + historyDB = historydb.NewHistoryDB(db, db, nil) // Run tests result := m.Run() // Close DB

@@ -660,7 +660,7 @@ require.NoError(t, err)

} // Set batchNum keeped txs for i := range keepedIDs { - _, err = l2DB.db.Exec( + _, err = l2DB.dbWrite.Exec( "UPDATE tx_pool SET batch_num = $1 WHERE tx_id = $2;", safeBatchNum, keepedIDs[i], )

@@ -679,7 +679,7 @@ // 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 { // Set timestamp - _, err = l2DB.db.Exec( + _, err = l2DB.dbWrite.Exec( "UPDATE tx_pool SET timestamp = $1, state = $2 WHERE tx_id = $3;", deleteTimestamp, common.PoolL2TxStatePending, id, )

@@ -797,7 +797,7 @@

require.NoError(t, l2DB.StartForging( []common.TxID{txs[4].TxID, txs[5].TxID, txs[6].TxID, txs[7].TxID}, 1)) - _, err = l2DB.db.Exec( + _, err = l2DB.dbWrite.Exec( `UPDATE tx_pool SET external_delete = true WHERE tx_id IN ($1, $2, $3, $4) ;`,
M node/node.go → node/node.go

@@ -62,27 +62,47 @@ // Synchronizer

sync *synchronizer.Synchronizer // General - cfg *config.Node - mode Mode - sqlConn *sqlx.DB - ctx context.Context - wg sync.WaitGroup - cancel context.CancelFunc + cfg *config.Node + mode Mode + sqlConnRead *sqlx.DB + sqlConnWrite *sqlx.DB + ctx context.Context + wg sync.WaitGroup + cancel context.CancelFunc } // NewNode creates a Node func NewNode(mode Mode, cfg *config.Node) (*Node, error) { meddler.Debug = cfg.Debug.MeddlerLogs // Stablish DB connection - db, err := dbUtils.InitSQLDB( - cfg.PostgreSQL.Port, - cfg.PostgreSQL.Host, - cfg.PostgreSQL.User, - cfg.PostgreSQL.Password, - cfg.PostgreSQL.Name, + dbWrite, err := dbUtils.InitSQLDB( + cfg.PostgreSQL.PortWrite, + cfg.PostgreSQL.HostWrite, + cfg.PostgreSQL.UserWrite, + cfg.PostgreSQL.PasswordWrite, + cfg.PostgreSQL.NameWrite, ) if err != nil { - return nil, tracerr.Wrap(err) + return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err)) + } + var dbRead *sqlx.DB + if cfg.PostgreSQL.HostRead == "" { + dbRead = dbWrite + } else if cfg.PostgreSQL.HostRead == cfg.PostgreSQL.HostWrite { + return nil, tracerr.Wrap(fmt.Errorf( + "PostgreSQL.HostRead and PostgreSQL.HostWrite must be different", + )) + } else { + dbRead, err = dbUtils.InitSQLDB( + cfg.PostgreSQL.PortRead, + cfg.PostgreSQL.HostRead, + cfg.PostgreSQL.UserRead, + cfg.PostgreSQL.PasswordRead, + cfg.PostgreSQL.NameRead, + ) + if err != nil { + return nil, tracerr.Wrap(fmt.Errorf("dbUtils.InitSQLDB: %w", err)) + } } var apiConnCon *dbUtils.APIConnectionController if cfg.API.Explorer || mode == ModeCoordinator {

@@ -92,7 +112,7 @@ cfg.API.SQLConnectionTimeout.Duration,

) } - historyDB := historydb.NewHistoryDB(db, apiConnCon) + historyDB := historydb.NewHistoryDB(dbRead, dbWrite, apiConnCon) ethClient, err := ethclient.Dial(cfg.Web3.URL) if err != nil {

@@ -201,7 +221,7 @@ var coord *coordinator.Coordinator

var l2DB *l2db.L2DB if mode == ModeCoordinator { l2DB = l2db.NewL2DB( - db, + dbRead, dbWrite, cfg.Coordinator.L2DB.SafetyPeriod, cfg.Coordinator.L2DB.MaxTxs, cfg.Coordinator.L2DB.MinFeeUSD,

@@ -391,7 +411,8 @@ coord: coord,

sync: sync, cfg: cfg, mode: mode, - sqlConn: db, + sqlConnRead: dbRead, + sqlConnWrite: dbWrite, ctx: ctx, cancel: cancel, }, nil
M priceupdater/priceupdater_test.go → priceupdater/priceupdater_test.go

@@ -20,7 +20,7 @@ // Init DB

pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") assert.NoError(t, err) - historyDB := historydb.NewHistoryDB(db, nil) + historyDB := historydb.NewHistoryDB(db, db, nil) // Clean DB test.WipeDB(historyDB.DB()) // Populate DB
M synchronizer/synchronizer_test.go → synchronizer/synchronizer_test.go

@@ -315,7 +315,7 @@ // Init History DB

pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) - historyDB := historydb.NewHistoryDB(db, nil) + historyDB := historydb.NewHistoryDB(db, db, nil) // Clear DB test.WipeDB(historyDB.DB())
M test/zkproof/flows_test.go → test/zkproof/flows_test.go

@@ -38,7 +38,7 @@ Decimals: 18,

}) } - hdb := historydb.NewHistoryDB(db, nil) + hdb := historydb.NewHistoryDB(db, db, nil) assert.NoError(t, hdb.AddBlock(&common.Block{ Num: 1, }))

@@ -75,7 +75,7 @@ func initTxSelector(t *testing.T, chainID uint16, hermezContractAddr ethCommon.Address, coordUser *til.User) (*txselector.TxSelector, *l2db.L2DB, *statedb.StateDB) {

pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) - l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) + l2DB := l2db.NewL2DB(db, db, 10, 100, 0.0, 24*time.Hour, nil) dir, err := ioutil.TempDir("", "tmpSyncDB") require.NoError(t, err)
M txselector/txselector_test.go → txselector/txselector_test.go

@@ -29,7 +29,7 @@ func initTest(t *testing.T, chainID uint16, hermezContractAddr ethCommon.Address, coordUser *til.User) *TxSelector {

pass := os.Getenv("POSTGRES_PASS") db, err := dbUtils.InitSQLDB(5432, "localhost", "hermez", pass, "hermez") require.NoError(t, err) - l2DB := l2db.NewL2DB(db, 10, 100, 0.0, 24*time.Hour, nil) + l2DB := l2db.NewL2DB(db, db, 10, 100, 0.0, 24*time.Hour, nil) dir, err := ioutil.TempDir("", "tmpdb") require.NoError(t, err)

@@ -106,7 +106,7 @@ Decimals: 18,

}) } - hdb := historydb.NewHistoryDB(db, nil) + hdb := historydb.NewHistoryDB(db, db, nil) assert.NoError(t, hdb.AddBlock(&common.Block{ Num: 1, }))