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
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 return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", tx))
82}
83
84// AddTxTest inserts a tx into the L2DB. This is useful for test purposes,
85// but in production txs will only be inserted through the API
86func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error {
87 // transform tx from *common.PoolL2Tx to PoolL2TxWrite
88 insertTx := &PoolL2TxWrite{
89 TxID: tx.TxID,
90 FromIdx: tx.FromIdx,
91 TokenID: tx.TokenID,
92 Amount: tx.Amount,
93 Fee: tx.Fee,
94 Nonce: tx.Nonce,
95 State: common.PoolL2TxStatePending,
96 Signature: tx.Signature,
97 RqAmount: tx.RqAmount,
98 Type: tx.Type,
99 }
100 if tx.ToIdx != 0 {
101 insertTx.ToIdx = &tx.ToIdx
102 }
103 nilAddr := ethCommon.BigToAddress(big.NewInt(0))
104 if tx.ToEthAddr != nilAddr {
105 insertTx.ToEthAddr = &tx.ToEthAddr
106 }
107 if tx.RqFromIdx != 0 {
108 insertTx.RqFromIdx = &tx.RqFromIdx
109 }
110 if tx.RqToIdx != 0 { // if true, all Rq... fields must be different to nil
111 insertTx.RqToIdx = &tx.RqToIdx
112 insertTx.RqTokenID = &tx.RqTokenID
113 insertTx.RqFee = &tx.RqFee
114 insertTx.RqNonce = &tx.RqNonce
115 }
116 if tx.RqToEthAddr != nilAddr {
117 insertTx.RqToEthAddr = &tx.RqToEthAddr
118 }
119 if tx.ToBJJ != common.EmptyBJJComp {
120 insertTx.ToBJJ = &tx.ToBJJ
121 }
122 if tx.RqToBJJ != common.EmptyBJJComp {
123 insertTx.RqToBJJ = &tx.RqToBJJ
124 }
125 f := new(big.Float).SetInt(tx.Amount)
126 amountF, _ := f.Float64()
127 insertTx.AmountFloat = amountF
128 // insert tx
129 return tracerr.Wrap(meddler.Insert(l2db.db, "tx_pool", insertTx))
130}
131
132// selectPoolTxAPI select part of queries to get PoolL2TxRead
133const selectPoolTxAPI = `SELECT tx_pool.tx_id, hez_idx(tx_pool.from_idx, token.symbol) AS from_idx, tx_pool.from_eth_addr,
134tx_pool.from_bjj, hez_idx(tx_pool.to_idx, token.symbol) AS to_idx, tx_pool.to_eth_addr,
135tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
136tx_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,
137hez_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,
138tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
139token.item_id AS token_item_id, token.eth_block_num, token.eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update
140FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
141
142// selectPoolTxCommon select part of queries to get common.PoolL2Tx
143const selectPoolTxCommon = `SELECT tx_pool.tx_id, from_idx, to_idx, tx_pool.to_eth_addr,
144tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
145tx_pool.state, tx_pool.signature, tx_pool.timestamp, rq_from_idx,
146rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount,
147tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
148fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update
149FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
150
151// GetTx return the specified Tx in common.PoolL2Tx format
152func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
153 tx := new(common.PoolL2Tx)
154 return tx, tracerr.Wrap(meddler.QueryRow(
155 l2db.db, tx,
156 selectPoolTxCommon+"WHERE tx_id = $1;",
157 txID,
158 ))
159}
160
161// GetTxAPI return the specified Tx in PoolTxAPI format
162func (l2db *L2DB) GetTxAPI(txID common.TxID) (*PoolTxAPI, error) {
163 tx := new(PoolTxAPI)
164 return tx, tracerr.Wrap(meddler.QueryRow(
165 l2db.db, tx,
166 selectPoolTxAPI+"WHERE tx_id = $1;",
167 txID,
168 ))
169}
170
171// GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee
172func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) {
173 var txs []*common.PoolL2Tx
174 err := meddler.QueryAll(
175 l2db.db, &txs,
176 selectPoolTxCommon+"WHERE state = $1",
177 common.PoolL2TxStatePending,
178 )
179 return db.SlicePtrsToSlice(txs).([]common.PoolL2Tx), tracerr.Wrap(err)
180}
181
182// StartForging updates the state of the transactions that will begin the forging process.
183// The state of the txs referenced by txIDs will be changed from Pending -> Forging
184func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
185 if len(txIDs) == 0 {
186 return nil
187 }
188 query, args, err := sqlx.In(
189 `UPDATE tx_pool
190 SET state = ?, batch_num = ?
191 WHERE state = ? AND tx_id IN (?);`,
192 common.PoolL2TxStateForging,
193 batchNum,
194 common.PoolL2TxStatePending,
195 txIDs,
196 )
197 if err != nil {
198 return tracerr.Wrap(err)
199 }
200 query = l2db.db.Rebind(query)
201 _, err = l2db.db.Exec(query, args...)
202 return tracerr.Wrap(err)
203}
204
205// DoneForging updates the state of the transactions that have been forged
206// so the state of the txs referenced by txIDs will be changed from Forging -> Forged
207func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
208 if len(txIDs) == 0 {
209 return nil
210 }
211 query, args, err := sqlx.In(
212 `UPDATE tx_pool
213 SET state = ?, batch_num = ?
214 WHERE state = ? AND tx_id IN (?);`,
215 common.PoolL2TxStateForged,
216 batchNum,
217 common.PoolL2TxStateForging,
218 txIDs,
219 )
220 if err != nil {
221 return tracerr.Wrap(err)
222 }
223 query = l2db.db.Rebind(query)
224 _, err = l2db.db.Exec(query, args...)
225 return tracerr.Wrap(err)
226}
227
228// InvalidateTxs updates the state of the transactions that are invalid.
229// The state of the txs referenced by txIDs will be changed from * -> Invalid
230func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
231 if len(txIDs) == 0 {
232 return nil
233 }
234 query, args, err := sqlx.In(
235 `UPDATE tx_pool
236 SET state = ?, batch_num = ?
237 WHERE tx_id IN (?);`,
238 common.PoolL2TxStateInvalid,
239 batchNum,
240 txIDs,
241 )
242 if err != nil {
243 return tracerr.Wrap(err)
244 }
245 query = l2db.db.Rebind(query)
246 _, err = l2db.db.Exec(query, args...)
247 return tracerr.Wrap(err)
248}
249
250// GetPendingUniqueFromIdxs returns from all the pending transactions, the set
251// of unique FromIdx
252func (l2db *L2DB) GetPendingUniqueFromIdxs() ([]common.Idx, error) {
253 var idxs []common.Idx
254 rows, err := l2db.db.Query(`SELECT DISTINCT from_idx FROM tx_pool
255 WHERE state = $1;`, common.PoolL2TxStatePending)
256 if err != nil {
257 return nil, tracerr.Wrap(err)
258 }
259 defer db.RowsClose(rows)
260 var idx common.Idx
261 for rows.Next() {
262 err = rows.Scan(&idx)
263 if err != nil {
264 return nil, tracerr.Wrap(err)
265 }
266 idxs = append(idxs, idx)
267 }
268 return idxs, nil
269}
270
271var invalidateOldNoncesQuery = fmt.Sprintf(`
272 UPDATE tx_pool SET
273 state = '%s',
274 batch_num = %%d
275 FROM (VALUES
276 (NULL::::BIGINT, NULL::::BIGINT),
277 (:idx, :nonce)
278 ) as updated_acc (idx, nonce)
279 WHERE tx_pool.state = '%s' AND
280 tx_pool.from_idx = updated_acc.idx AND
281 tx_pool.nonce < updated_acc.nonce;
282 `, common.PoolL2TxStateInvalid, common.PoolL2TxStatePending)
283
284// InvalidateOldNonces invalidate txs with nonces that are smaller or equal than their
285// respective accounts nonces. The state of the affected txs will be changed
286// from Pending to Invalid
287func (l2db *L2DB) InvalidateOldNonces(updatedAccounts []common.IdxNonce, batchNum common.BatchNum) (err error) {
288 if len(updatedAccounts) == 0 {
289 return nil
290 }
291 // Fill the batch_num in the query with Sprintf because we are using a
292 // named query which works with slices, and doens't handle an extra
293 // individual argument.
294 query := fmt.Sprintf(invalidateOldNoncesQuery, batchNum)
295 if _, err := sqlx.NamedExec(l2db.db, query, updatedAccounts); err != nil {
296 return tracerr.Wrap(err)
297 }
298 return nil
299}
300
301// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
302// The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
303func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
304 _, err := l2db.db.Exec(
305 `UPDATE tx_pool SET batch_num = NULL, state = $1
306 WHERE (state = $2 OR state = $3) AND batch_num > $4`,
307 common.PoolL2TxStatePending,
308 common.PoolL2TxStateForged,
309 common.PoolL2TxStateInvalid,
310 lastValidBatch,
311 )
312 return tracerr.Wrap(err)
313}
314
315// Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
316// it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
317func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) {
318 txn, err := l2db.db.Beginx()
319 if err != nil {
320 return tracerr.Wrap(err)
321 }
322 defer func() {
323 // Rollback the transaction if there was an error.
324 if err != nil {
325 db.Rollback(txn)
326 }
327 }()
328 // Delete pending txs that have been in the pool after the TTL if maxTxs is reached
329 now := time.Now().UTC().Unix()
330 _, err = txn.Exec(
331 `DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
332 l2db.maxTxs,
333 time.Unix(now-int64(l2db.ttl.Seconds()), 0),
334 )
335 if err != nil {
336 return tracerr.Wrap(err)
337 }
338 // Delete txs that have been marked as forged / invalid after the safety period
339 _, err = txn.Exec(
340 `DELETE FROM tx_pool
341 WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
342 currentBatchNum-l2db.safetyPeriod,
343 common.PoolL2TxStateForged,
344 common.PoolL2TxStateInvalid,
345 )
346 if err != nil {
347 return tracerr.Wrap(err)
348 }
349 return tracerr.Wrap(txn.Commit())
350}