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