hermez-node/db/l2db/l2db.go
db/l2db/l2db.goRaw
1package l2db
2
3import (
4 "math/big"
5 "time"
6
7 ethCommon "github.com/ethereum/go-ethereum/common"
8 "github.com/hermeznetwork/hermez-node/common"
9 "github.com/hermeznetwork/hermez-node/db"
10 "github.com/hermeznetwork/tracerr"
11 "github.com/jmoiron/sqlx"
12
13 //nolint:errcheck // driver for postgres DB
14 _ "github.com/lib/pq"
15 "github.com/russross/meddler"
16)
17
18// TODO(Edu): Check DB consistency while there's concurrent use from Coordinator/TxSelector & API
19
20// L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant
21// due to them being forged or invalid after a safety period
22type L2DB struct {
23 db *sqlx.DB
24 safetyPeriod common.BatchNum
25 ttl time.Duration
26 maxTxs uint32
27}
28
29// NewL2DB creates a L2DB.
30// To create it, it's needed db connection, safety period expressed in batches,
31// maxTxs that the DB should have and TTL (time to live) for pending txs.
32func NewL2DB(db *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, TTL time.Duration) *L2DB {
33 return &L2DB{
34 db: db,
35 safetyPeriod: safetyPeriod,
36 ttl: TTL,
37 maxTxs: maxTxs,
38 }
39}
40
41// DB returns a pointer to the L2DB.db. This method should be used only for
42// internal testing purposes.
43func (l2db *L2DB) DB() *sqlx.DB {
44 return l2db.db
45}
46
47// AddAccountCreationAuth inserts an account creation authorization into the DB
48func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error {
49 // return meddler.Insert(l2db.db, "account_creation_auth", auth)
50 _, err := l2db.db.Exec(
51 `INSERT INTO account_creation_auth (eth_addr, bjj, signature)
52 VALUES ($1, $2, $3);`,
53 auth.EthAddr, auth.BJJ, auth.Signature,
54 )
55 return tracerr.Wrap(err)
56}
57
58// GetAccountCreationAuth returns an account creation authorization from the DB
59func (l2db *L2DB) GetAccountCreationAuth(addr ethCommon.Address) (*common.AccountCreationAuth, error) {
60 auth := new(common.AccountCreationAuth)
61 return auth, meddler.QueryRow(
62 l2db.db, auth,
63 "SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
64 addr,
65 )
66}
67
68// GetAccountCreationAuthAPI returns an account creation authorization from the DB
69func (l2db *L2DB) GetAccountCreationAuthAPI(addr ethCommon.Address) (*AccountCreationAuthAPI, error) {
70 auth := new(AccountCreationAuthAPI)
71 return auth, meddler.QueryRow(
72 l2db.db, auth,
73 "SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
74 addr,
75 )
76}
77
78// AddTx inserts a tx to the pool
79func (l2db *L2DB) AddTx(tx *PoolL2TxWrite) error {
80 return meddler.Insert(l2db.db, "tx_pool", tx)
81}
82
83// AddTxTest inserts a tx into the L2DB. This is useful for test purposes,
84// but in production txs will only be inserted through the API
85func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error {
86 // transform tx from *common.PoolL2Tx to PoolL2TxWrite
87 insertTx := &PoolL2TxWrite{
88 TxID: tx.TxID,
89 FromIdx: tx.FromIdx,
90 ToBJJ: tx.ToBJJ,
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 RqToBJJ: tx.RqToBJJ,
98 RqAmount: tx.RqAmount,
99 Type: tx.Type,
100 }
101 if tx.ToIdx != 0 {
102 insertTx.ToIdx = &tx.ToIdx
103 }
104 nilAddr := ethCommon.BigToAddress(big.NewInt(0))
105 if tx.ToEthAddr != nilAddr {
106 insertTx.ToEthAddr = &tx.ToEthAddr
107 }
108 if tx.RqFromIdx != 0 {
109 insertTx.RqFromIdx = &tx.RqFromIdx
110 }
111 if tx.RqToIdx != 0 { // if true, all Rq... fields must be different to nil
112 insertTx.RqToIdx = &tx.RqToIdx
113 insertTx.RqTokenID = &tx.RqTokenID
114 insertTx.RqFee = &tx.RqFee
115 insertTx.RqNonce = &tx.RqNonce
116 }
117 if tx.RqToEthAddr != nilAddr {
118 insertTx.RqToEthAddr = &tx.RqToEthAddr
119 }
120 f := new(big.Float).SetInt(tx.Amount)
121 amountF, _ := f.Float64()
122 insertTx.AmountFloat = amountF
123 // insert tx
124 return meddler.Insert(l2db.db, "tx_pool", insertTx)
125}
126
127// selectPoolTxAPI select part of queries to get PoolL2TxRead
128const selectPoolTxAPI = `SELECT tx_pool.tx_id, hez_idx(tx_pool.from_idx, token.symbol) AS from_idx, tx_pool.from_eth_addr,
129tx_pool.from_bjj, hez_idx(tx_pool.to_idx, token.symbol) AS to_idx, tx_pool.to_eth_addr,
130tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
131tx_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,
132hez_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,
133tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
134token.item_id AS token_item_id, token.eth_block_num, token.eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update
135FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
136
137// selectPoolTxCommon select part of queries to get common.PoolL2Tx
138const selectPoolTxCommon = `SELECT tx_pool.tx_id, from_idx, to_idx, tx_pool.to_eth_addr,
139tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce,
140tx_pool.state, tx_pool.signature, tx_pool.timestamp, rq_from_idx,
141rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount,
142tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type,
143fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update
144FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
145
146// GetTx return the specified Tx in common.PoolL2Tx format
147func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
148 tx := new(common.PoolL2Tx)
149 return tx, meddler.QueryRow(
150 l2db.db, tx,
151 selectPoolTxCommon+"WHERE tx_id = $1;",
152 txID,
153 )
154}
155
156// GetTxAPI return the specified Tx in PoolTxAPI format
157func (l2db *L2DB) GetTxAPI(txID common.TxID) (*PoolTxAPI, error) {
158 tx := new(PoolTxAPI)
159 return tx, meddler.QueryRow(
160 l2db.db, tx,
161 selectPoolTxAPI+"WHERE tx_id = $1;",
162 txID,
163 )
164}
165
166// GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee
167func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) {
168 var txs []*common.PoolL2Tx
169 err := meddler.QueryAll(
170 l2db.db, &txs,
171 selectPoolTxCommon+"WHERE state = $1",
172 common.PoolL2TxStatePending,
173 )
174 return db.SlicePtrsToSlice(txs).([]common.PoolL2Tx), tracerr.Wrap(err)
175}
176
177// StartForging updates the state of the transactions that will begin the forging process.
178// The state of the txs referenced by txIDs will be changed from Pending -> Forging
179func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
180 if len(txIDs) == 0 {
181 return nil
182 }
183 query, args, err := sqlx.In(
184 `UPDATE tx_pool
185 SET state = ?, batch_num = ?
186 WHERE state = ? AND tx_id IN (?);`,
187 common.PoolL2TxStateForging,
188 batchNum,
189 common.PoolL2TxStatePending,
190 txIDs,
191 )
192 if err != nil {
193 return tracerr.Wrap(err)
194 }
195 query = l2db.db.Rebind(query)
196 _, err = l2db.db.Exec(query, args...)
197 return tracerr.Wrap(err)
198}
199
200// DoneForging updates the state of the transactions that have been forged
201// so the state of the txs referenced by txIDs will be changed from Forging -> Forged
202func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
203 if len(txIDs) == 0 {
204 return nil
205 }
206 query, args, err := sqlx.In(
207 `UPDATE tx_pool
208 SET state = ?, batch_num = ?
209 WHERE state = ? AND tx_id IN (?);`,
210 common.PoolL2TxStateForged,
211 batchNum,
212 common.PoolL2TxStateForging,
213 txIDs,
214 )
215 if err != nil {
216 return tracerr.Wrap(err)
217 }
218 query = l2db.db.Rebind(query)
219 _, err = l2db.db.Exec(query, args...)
220 return tracerr.Wrap(err)
221}
222
223// InvalidateTxs updates the state of the transactions that are invalid.
224// The state of the txs referenced by txIDs will be changed from * -> Invalid
225func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
226 if len(txIDs) == 0 {
227 return nil
228 }
229 query, args, err := sqlx.In(
230 `UPDATE tx_pool
231 SET state = ?, batch_num = ?
232 WHERE tx_id IN (?);`,
233 common.PoolL2TxStateInvalid,
234 batchNum,
235 txIDs,
236 )
237 if err != nil {
238 return tracerr.Wrap(err)
239 }
240 query = l2db.db.Rebind(query)
241 _, err = l2db.db.Exec(query, args...)
242 return tracerr.Wrap(err)
243}
244
245// CheckNonces invalidate txs with nonces that are smaller or equal than their respective accounts nonces.
246// The state of the affected txs will be changed from Pending -> Invalid
247func (l2db *L2DB) CheckNonces(updatedAccounts []common.Account, batchNum common.BatchNum) (err error) {
248 if len(updatedAccounts) == 0 {
249 return nil
250 }
251 txn, err := l2db.db.Beginx()
252 if err != nil {
253 return tracerr.Wrap(err)
254 }
255 defer func() {
256 // Rollback the transaction if there was an error.
257 if err != nil {
258 db.Rollback(txn)
259 }
260 }()
261 for i := 0; i < len(updatedAccounts); i++ {
262 _, err = txn.Exec(
263 `UPDATE tx_pool
264 SET state = $1, batch_num = $2
265 WHERE state = $3 AND from_idx = $4 AND nonce <= $5;`,
266 common.PoolL2TxStateInvalid,
267 batchNum,
268 common.PoolL2TxStatePending,
269 updatedAccounts[i].Idx,
270 updatedAccounts[i].Nonce,
271 )
272 if err != nil {
273 return tracerr.Wrap(err)
274 }
275 }
276 return tracerr.Wrap(txn.Commit())
277}
278
279// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
280// The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
281func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
282 _, err := l2db.db.Exec(
283 `UPDATE tx_pool SET batch_num = NULL, state = $1
284 WHERE (state = $2 OR state = $3) AND batch_num > $4`,
285 common.PoolL2TxStatePending,
286 common.PoolL2TxStateForged,
287 common.PoolL2TxStateInvalid,
288 lastValidBatch,
289 )
290 return tracerr.Wrap(err)
291}
292
293// Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
294// it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
295func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) {
296 txn, err := l2db.db.Beginx()
297 if err != nil {
298 return tracerr.Wrap(err)
299 }
300 defer func() {
301 // Rollback the transaction if there was an error.
302 if err != nil {
303 db.Rollback(txn)
304 }
305 }()
306 // Delete pending txs that have been in the pool after the TTL if maxTxs is reached
307 now := time.Now().UTC().Unix()
308 _, err = txn.Exec(
309 `DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
310 l2db.maxTxs,
311 time.Unix(now-int64(l2db.ttl.Seconds()), 0),
312 )
313 if err != nil {
314 return tracerr.Wrap(err)
315 }
316 // Delete txs that have been marked as forged / invalid after the safety period
317 _, err = txn.Exec(
318 `DELETE FROM tx_pool
319 WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
320 currentBatchNum-l2db.safetyPeriod,
321 common.PoolL2TxStateForged,
322 common.PoolL2TxStateInvalid,
323 )
324 if err != nil {
325 return tracerr.Wrap(err)
326 }
327 return tracerr.Wrap(txn.Commit())
328}