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