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