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