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