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