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