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.

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