You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

344 lines
11 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. package l2db
  2. import (
  3. "fmt"
  4. "math/big"
  5. "time"
  6. ethCommon "github.com/ethereum/go-ethereum/common"
  7. "github.com/hermeznetwork/hermez-node/common"
  8. "github.com/hermeznetwork/hermez-node/db"
  9. "github.com/hermeznetwork/tracerr"
  10. "github.com/jmoiron/sqlx"
  11. //nolint:errcheck // driver for postgres DB
  12. _ "github.com/lib/pq"
  13. "github.com/russross/meddler"
  14. )
  15. // TODO(Edu): Check DB consistency while there's concurrent use from Coordinator/TxSelector & API
  16. // L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant
  17. // due to them being forged or invalid after a safety period
  18. type L2DB struct {
  19. db *sqlx.DB
  20. safetyPeriod common.BatchNum
  21. ttl time.Duration
  22. maxTxs uint32
  23. }
  24. // NewL2DB creates a L2DB.
  25. // To create it, it's needed db connection, safety period expressed in batches,
  26. // maxTxs that the DB should have and TTL (time to live) for pending txs.
  27. func NewL2DB(db *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, TTL time.Duration) *L2DB {
  28. return &L2DB{
  29. db: db,
  30. safetyPeriod: safetyPeriod,
  31. ttl: TTL,
  32. maxTxs: maxTxs,
  33. }
  34. }
  35. // DB returns a pointer to the L2DB.db. This method should be used only for
  36. // internal testing purposes.
  37. func (l2db *L2DB) DB() *sqlx.DB {
  38. return l2db.db
  39. }
  40. // AddAccountCreationAuth inserts an account creation authorization into the DB
  41. func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error {
  42. // return meddler.Insert(l2db.db, "account_creation_auth", auth)
  43. _, err := l2db.db.Exec(
  44. `INSERT INTO account_creation_auth (eth_addr, bjj, signature)
  45. VALUES ($1, $2, $3);`,
  46. auth.EthAddr, auth.BJJ, auth.Signature,
  47. )
  48. return tracerr.Wrap(err)
  49. }
  50. // GetAccountCreationAuth returns an account creation authorization from the DB
  51. func (l2db *L2DB) GetAccountCreationAuth(addr ethCommon.Address) (*common.AccountCreationAuth, error) {
  52. auth := new(common.AccountCreationAuth)
  53. return auth, tracerr.Wrap(meddler.QueryRow(
  54. l2db.db, auth,
  55. "SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
  56. addr,
  57. ))
  58. }
  59. // GetAccountCreationAuthAPI returns an account creation authorization from the DB
  60. func (l2db *L2DB) GetAccountCreationAuthAPI(addr ethCommon.Address) (*AccountCreationAuthAPI, error) {
  61. auth := new(AccountCreationAuthAPI)
  62. return auth, tracerr.Wrap(meddler.QueryRow(
  63. l2db.db, auth,
  64. "SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
  65. addr,
  66. ))
  67. }
  68. // AddTx inserts a tx to the pool
  69. func (l2db *L2DB) AddTx(tx *PoolL2TxWrite) error {
  70. return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", tx))
  71. }
  72. // AddTxTest inserts a tx into the L2DB. This is useful for test purposes,
  73. // but in production txs will only be inserted through the API
  74. func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error {
  75. // transform tx from *common.PoolL2Tx to PoolL2TxWrite
  76. insertTx := &PoolL2TxWrite{
  77. TxID: tx.TxID,
  78. FromIdx: tx.FromIdx,
  79. ToBJJ: tx.ToBJJ,
  80. TokenID: tx.TokenID,
  81. Amount: tx.Amount,
  82. Fee: tx.Fee,
  83. Nonce: tx.Nonce,
  84. State: common.PoolL2TxStatePending,
  85. Signature: tx.Signature,
  86. RqToBJJ: tx.RqToBJJ,
  87. RqAmount: tx.RqAmount,
  88. Type: tx.Type,
  89. }
  90. if tx.ToIdx != 0 {
  91. insertTx.ToIdx = &tx.ToIdx
  92. }
  93. nilAddr := ethCommon.BigToAddress(big.NewInt(0))
  94. if tx.ToEthAddr != nilAddr {
  95. insertTx.ToEthAddr = &tx.ToEthAddr
  96. }
  97. if tx.RqFromIdx != 0 {
  98. insertTx.RqFromIdx = &tx.RqFromIdx
  99. }
  100. if tx.RqToIdx != 0 { // if true, all Rq... fields must be different to nil
  101. insertTx.RqToIdx = &tx.RqToIdx
  102. insertTx.RqTokenID = &tx.RqTokenID
  103. insertTx.RqFee = &tx.RqFee
  104. insertTx.RqNonce = &tx.RqNonce
  105. }
  106. if tx.RqToEthAddr != nilAddr {
  107. insertTx.RqToEthAddr = &tx.RqToEthAddr
  108. }
  109. f := new(big.Float).SetInt(tx.Amount)
  110. amountF, _ := f.Float64()
  111. insertTx.AmountFloat = amountF
  112. // insert tx
  113. return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", insertTx))
  114. }
  115. // selectPoolTxAPI select part of queries to get PoolL2TxRead
  116. const selectPoolTxAPI = `SELECT tx_pool.tx_id, hez_idx(tx_pool.from_idx, token.symbol) AS from_idx, tx_pool.from_eth_addr,
  117. tx_pool.from_bjj, hez_idx(tx_pool.to_idx, token.symbol) AS to_idx, tx_pool.to_eth_addr,
  118. tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
  119. tx_pool.state, tx_pool.signature, tx_pool.timestamp, tx_pool.batch_num, hez_idx(tx_pool.rq_from_idx, token.symbol) AS rq_from_idx,
  120. hez_idx(tx_pool.rq_to_idx, token.symbol) AS rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount,
  121. tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
  122. token.item_id AS token_item_id, token.eth_block_num, token.eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update
  123. FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
  124. // selectPoolTxCommon select part of queries to get common.PoolL2Tx
  125. const selectPoolTxCommon = `SELECT tx_pool.tx_id, from_idx, to_idx, tx_pool.to_eth_addr,
  126. tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
  127. tx_pool.state, tx_pool.signature, tx_pool.timestamp, rq_from_idx,
  128. rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount,
  129. tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
  130. fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update
  131. FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
  132. // GetTx return the specified Tx in common.PoolL2Tx format
  133. func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
  134. tx := new(common.PoolL2Tx)
  135. return tx, tracerr.Wrap(meddler.QueryRow(
  136. l2db.db, tx,
  137. selectPoolTxCommon+"WHERE tx_id = $1;",
  138. txID,
  139. ))
  140. }
  141. // GetTxAPI return the specified Tx in PoolTxAPI format
  142. func (l2db *L2DB) GetTxAPI(txID common.TxID) (*PoolTxAPI, error) {
  143. tx := new(PoolTxAPI)
  144. return tx, tracerr.Wrap(meddler.QueryRow(
  145. l2db.db, tx,
  146. selectPoolTxAPI+"WHERE tx_id = $1;",
  147. txID,
  148. ))
  149. }
  150. // GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee
  151. func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) {
  152. var txs []*common.PoolL2Tx
  153. err := meddler.QueryAll(
  154. l2db.db, &txs,
  155. selectPoolTxCommon+"WHERE state = $1",
  156. common.PoolL2TxStatePending,
  157. )
  158. return db.SlicePtrsToSlice(txs).([]common.PoolL2Tx), tracerr.Wrap(err)
  159. }
  160. // StartForging updates the state of the transactions that will begin the forging process.
  161. // The state of the txs referenced by txIDs will be changed from Pending -> Forging
  162. func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
  163. if len(txIDs) == 0 {
  164. return nil
  165. }
  166. query, args, err := sqlx.In(
  167. `UPDATE tx_pool
  168. SET state = ?, batch_num = ?
  169. WHERE state = ? AND tx_id IN (?);`,
  170. common.PoolL2TxStateForging,
  171. batchNum,
  172. common.PoolL2TxStatePending,
  173. txIDs,
  174. )
  175. if err != nil {
  176. return tracerr.Wrap(err)
  177. }
  178. query = l2db.db.Rebind(query)
  179. _, err = l2db.db.Exec(query, args...)
  180. return tracerr.Wrap(err)
  181. }
  182. // DoneForging updates the state of the transactions that have been forged
  183. // so the state of the txs referenced by txIDs will be changed from Forging -> Forged
  184. func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
  185. if len(txIDs) == 0 {
  186. return nil
  187. }
  188. query, args, err := sqlx.In(
  189. `UPDATE tx_pool
  190. SET state = ?, batch_num = ?
  191. WHERE state = ? AND tx_id IN (?);`,
  192. common.PoolL2TxStateForged,
  193. batchNum,
  194. common.PoolL2TxStateForging,
  195. txIDs,
  196. )
  197. if err != nil {
  198. return tracerr.Wrap(err)
  199. }
  200. query = l2db.db.Rebind(query)
  201. _, err = l2db.db.Exec(query, args...)
  202. return tracerr.Wrap(err)
  203. }
  204. // InvalidateTxs updates the state of the transactions that are invalid.
  205. // The state of the txs referenced by txIDs will be changed from * -> Invalid
  206. func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
  207. if len(txIDs) == 0 {
  208. return nil
  209. }
  210. query, args, err := sqlx.In(
  211. `UPDATE tx_pool
  212. SET state = ?, batch_num = ?
  213. WHERE tx_id IN (?);`,
  214. common.PoolL2TxStateInvalid,
  215. batchNum,
  216. txIDs,
  217. )
  218. if err != nil {
  219. return tracerr.Wrap(err)
  220. }
  221. query = l2db.db.Rebind(query)
  222. _, err = l2db.db.Exec(query, args...)
  223. return tracerr.Wrap(err)
  224. }
  225. // GetPendingUniqueFromIdxs returns from all the pending transactions, the set
  226. // of unique FromIdx
  227. func (l2db *L2DB) GetPendingUniqueFromIdxs() ([]common.Idx, error) {
  228. var idxs []common.Idx
  229. rows, err := l2db.db.Query(`SELECT DISTINCT from_idx FROM tx_pool
  230. WHERE state = $1;`, common.PoolL2TxStatePending)
  231. if err != nil {
  232. return nil, tracerr.Wrap(err)
  233. }
  234. defer db.RowsClose(rows)
  235. var idx common.Idx
  236. for rows.Next() {
  237. err = rows.Scan(&idx)
  238. if err != nil {
  239. return nil, tracerr.Wrap(err)
  240. }
  241. idxs = append(idxs, idx)
  242. }
  243. return idxs, nil
  244. }
  245. var checkNoncesQuery = fmt.Sprintf(`
  246. UPDATE tx_pool SET
  247. state = '%s',
  248. batch_num = %%d
  249. FROM (VALUES
  250. (NULL::::BIGINT, NULL::::BIGINT),
  251. (:idx, :nonce)
  252. ) as updated_acc (idx, nonce)
  253. WHERE tx_pool.from_idx = updated_acc.idx AND tx_pool.nonce <= updated_acc.nonce;
  254. `, common.PoolL2TxStateInvalid)
  255. // CheckNonces invalidate txs with nonces that are smaller or equal than their
  256. // respective accounts nonces. The state of the affected txs will be changed
  257. // from Pending to Invalid
  258. func (l2db *L2DB) CheckNonces(updatedAccounts []common.IdxNonce, batchNum common.BatchNum) (err error) {
  259. if len(updatedAccounts) == 0 {
  260. return nil
  261. }
  262. // Fill the batch_num in the query with Sprintf because we are using a
  263. // named query which works with slices, and doens't handle an extra
  264. // individual argument.
  265. query := fmt.Sprintf(checkNoncesQuery, batchNum)
  266. if _, err := sqlx.NamedExec(l2db.db, query, updatedAccounts); err != nil {
  267. return tracerr.Wrap(err)
  268. }
  269. return nil
  270. }
  271. // Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
  272. // The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
  273. func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
  274. _, err := l2db.db.Exec(
  275. `UPDATE tx_pool SET batch_num = NULL, state = $1
  276. WHERE (state = $2 OR state = $3) AND batch_num > $4`,
  277. common.PoolL2TxStatePending,
  278. common.PoolL2TxStateForged,
  279. common.PoolL2TxStateInvalid,
  280. lastValidBatch,
  281. )
  282. return tracerr.Wrap(err)
  283. }
  284. // Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
  285. // it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
  286. func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) {
  287. txn, err := l2db.db.Beginx()
  288. if err != nil {
  289. return tracerr.Wrap(err)
  290. }
  291. defer func() {
  292. // Rollback the transaction if there was an error.
  293. if err != nil {
  294. db.Rollback(txn)
  295. }
  296. }()
  297. // Delete pending txs that have been in the pool after the TTL if maxTxs is reached
  298. now := time.Now().UTC().Unix()
  299. _, err = txn.Exec(
  300. `DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
  301. l2db.maxTxs,
  302. time.Unix(now-int64(l2db.ttl.Seconds()), 0),
  303. )
  304. if err != nil {
  305. return tracerr.Wrap(err)
  306. }
  307. // Delete txs that have been marked as forged / invalid after the safety period
  308. _, err = txn.Exec(
  309. `DELETE FROM tx_pool
  310. WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
  311. currentBatchNum-l2db.safetyPeriod,
  312. common.PoolL2TxStateForged,
  313. common.PoolL2TxStateInvalid,
  314. )
  315. if err != nil {
  316. return tracerr.Wrap(err)
  317. }
  318. return tracerr.Wrap(txn.Commit())
  319. }