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