hermez-node/db/l2db/l2db.go
db/l2db/l2db.goRaw
1package l2db
2
3import (
4 "errors"
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/hermez-node/log"
12 "github.com/iden3/go-iden3-crypto/babyjub"
13 "github.com/jmoiron/sqlx"
14
15 //nolint:errcheck // driver for postgres DB
16 _ "github.com/lib/pq"
17 "github.com/russross/meddler"
18)
19
20// TODO(Edu): Check DB consistency while there's concurrent use from Coordinator/TxSelector & API
21
22// L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant
23// due to them being forged or invalid after a safety period
24type L2DB struct {
25 db *sqlx.DB
26 safetyPeriod common.BatchNum
27 ttl time.Duration
28 maxTxs uint32
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(db *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, TTL time.Duration) *L2DB {
35 return &L2DB{
36 db: db,
37 safetyPeriod: safetyPeriod,
38 ttl: TTL,
39 maxTxs: maxTxs,
40 }
41}
42
43// DB returns a pointer to the L2DB.db. This method should be used only for
44// internal testing purposes.
45func (l2db *L2DB) DB() *sqlx.DB {
46 return l2db.db
47}
48
49// AddAccountCreationAuth inserts an account creation authorization into the DB
50func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error {
51 return meddler.Insert(l2db.db, "account_creation_auth", auth)
52}
53
54// GetAccountCreationAuth returns an account creation authorization into the DB
55func (l2db *L2DB) GetAccountCreationAuth(addr *ethCommon.Address) (*common.AccountCreationAuth, error) {
56 if addr == nil {
57 return nil, errors.New("addr cannot be nil")
58 }
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// AddTxTest inserts a tx into the L2DB. This is useful for test purposes,
68// but in production txs will only be inserted through the API (method TBD)
69func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error {
70 type withouUSD struct {
71 TxID common.TxID `meddler:"tx_id"`
72 FromIdx common.Idx `meddler:"from_idx"`
73 ToIdx *common.Idx `meddler:"to_idx"`
74 ToEthAddr *ethCommon.Address `meddler:"to_eth_addr"`
75 ToBJJ *babyjub.PublicKey `meddler:"to_bjj"`
76 TokenID common.TokenID `meddler:"token_id"`
77 Amount *big.Int `meddler:"amount,bigint"`
78 AmountFloat float64 `meddler:"amount_f"`
79 Fee common.FeeSelector `meddler:"fee"`
80 Nonce common.Nonce `meddler:"nonce"`
81 State common.PoolL2TxState `meddler:"state"`
82 Signature *babyjub.Signature `meddler:"signature"`
83 Timestamp time.Time `meddler:"timestamp,utctime"`
84 BatchNum *common.BatchNum `meddler:"batch_num"`
85 RqFromIdx *common.Idx `meddler:"rq_from_idx"`
86 RqToIdx *common.Idx `meddler:"rq_to_idx"`
87 RqToEthAddr *ethCommon.Address `meddler:"rq_to_eth_addr"`
88 RqToBJJ *babyjub.PublicKey `meddler:"rq_to_bjj"`
89 RqTokenID *common.TokenID `meddler:"rq_token_id"`
90 RqAmount *big.Int `meddler:"rq_amount,bigintnull"`
91 RqFee *common.FeeSelector `meddler:"rq_fee"`
92 RqNonce *uint64 `meddler:"rq_nonce"`
93 Type common.TxType `meddler:"tx_type"`
94 }
95 return meddler.Insert(l2db.db, "tx_pool", &withouUSD{
96 TxID: tx.TxID,
97 FromIdx: tx.FromIdx,
98 ToIdx: tx.ToIdx,
99 ToEthAddr: tx.ToEthAddr,
100 ToBJJ: tx.ToBJJ,
101 TokenID: tx.TokenID,
102 Amount: tx.Amount,
103 AmountFloat: tx.AmountFloat,
104 Fee: tx.Fee,
105 Nonce: tx.Nonce,
106 State: tx.State,
107 Signature: tx.Signature,
108 Timestamp: tx.Timestamp,
109 BatchNum: tx.BatchNum,
110 RqFromIdx: tx.RqFromIdx,
111 RqToIdx: tx.RqToIdx,
112 RqToEthAddr: tx.RqToEthAddr,
113 RqToBJJ: tx.RqToBJJ,
114 RqTokenID: tx.RqTokenID,
115 RqAmount: tx.RqAmount,
116 RqFee: tx.RqFee,
117 RqNonce: tx.RqNonce,
118 Type: tx.Type,
119 })
120}
121
122// selectPoolTx select part of queries to get common.PoolL2Tx
123const selectPoolTx = `SELECT tx_pool.*, token.usd * tx_pool.amount_f AS value_usd,
124fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update
125FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
126
127// GetTx return the specified Tx
128func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
129 tx := new(common.PoolL2Tx)
130 return tx, meddler.QueryRow(
131 l2db.db, tx,
132 selectPoolTx+"WHERE tx_id = $1;",
133 txID,
134 )
135}
136
137// GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee
138func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) {
139 var txs []*common.PoolL2Tx
140 err := meddler.QueryAll(
141 l2db.db, &txs,
142 selectPoolTx+"WHERE state = $1 AND token.usd IS NOT NULL",
143 common.PoolL2TxStatePending,
144 )
145 return db.SlicePtrsToSlice(txs).([]common.PoolL2Tx), err
146}
147
148// StartForging updates the state of the transactions that will begin the forging process.
149// The state of the txs referenced by txIDs will be changed from Pending -> Forging
150func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
151 query, args, err := sqlx.In(
152 `UPDATE tx_pool
153 SET state = ?, batch_num = ?
154 WHERE state = ? AND tx_id IN (?);`,
155 common.PoolL2TxStateForging,
156 batchNum,
157 common.PoolL2TxStatePending,
158 txIDs,
159 )
160 if err != nil {
161 return err
162 }
163 query = l2db.db.Rebind(query)
164 _, err = l2db.db.Exec(query, args...)
165 return err
166}
167
168// DoneForging updates the state of the transactions that have been forged
169// so the state of the txs referenced by txIDs will be changed from Forging -> Forged
170func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
171 query, args, err := sqlx.In(
172 `UPDATE tx_pool
173 SET state = ?, batch_num = ?
174 WHERE state = ? AND tx_id IN (?);`,
175 common.PoolL2TxStateForged,
176 batchNum,
177 common.PoolL2TxStateForging,
178 txIDs,
179 )
180 if err != nil {
181 return err
182 }
183 query = l2db.db.Rebind(query)
184 _, err = l2db.db.Exec(query, args...)
185 return err
186}
187
188// InvalidateTxs updates the state of the transactions that are invalid.
189// The state of the txs referenced by txIDs will be changed from * -> Invalid
190func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
191 query, args, err := sqlx.In(
192 `UPDATE tx_pool
193 SET state = ?, batch_num = ?
194 WHERE tx_id IN (?);`,
195 common.PoolL2TxStateInvalid,
196 batchNum,
197 txIDs,
198 )
199 if err != nil {
200 return err
201 }
202 query = l2db.db.Rebind(query)
203 _, err = l2db.db.Exec(query, args...)
204 return err
205}
206
207// CheckNonces invalidate txs with nonces that are smaller or equal than their respective accounts nonces.
208// The state of the affected txs will be changed from Pending -> Invalid
209func (l2db *L2DB) CheckNonces(updatedAccounts []common.Account, batchNum common.BatchNum) (err error) {
210 txn, err := l2db.db.Begin()
211 if err != nil {
212 return err
213 }
214 defer func() {
215 // Rollback the transaction if there was an error.
216 if err != nil {
217 errRollback := txn.Rollback()
218 if errRollback != nil {
219 log.Errorw("Rollback", "err", errRollback)
220 }
221 }
222 }()
223 for i := 0; i < len(updatedAccounts); i++ {
224 _, err = txn.Exec(
225 `UPDATE tx_pool
226 SET state = $1, batch_num = $2
227 WHERE state = $3 AND from_idx = $4 AND nonce <= $5;`,
228 common.PoolL2TxStateInvalid,
229 batchNum,
230 common.PoolL2TxStatePending,
231 updatedAccounts[i].Idx,
232 updatedAccounts[i].Nonce,
233 )
234 if err != nil {
235 return err
236 }
237 }
238 return txn.Commit()
239}
240
241// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
242// The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
243func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
244 _, err := l2db.db.Exec(
245 `UPDATE tx_pool SET batch_num = NULL, state = $1
246 WHERE (state = $2 OR state = $3) AND batch_num > $4`,
247 common.PoolL2TxStatePending,
248 common.PoolL2TxStateForged,
249 common.PoolL2TxStateInvalid,
250 lastValidBatch,
251 )
252 return err
253}
254
255// Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
256// it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
257func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) {
258 txn, err := l2db.db.Begin()
259 if err != nil {
260 return err
261 }
262 defer func() {
263 // Rollback the transaction if there was an error.
264 if err != nil {
265 errRollback := txn.Rollback()
266 if errRollback != nil {
267 log.Errorw("Rollback", "err", errRollback)
268 }
269 }
270 }()
271 // Delete pending txs that have been in the pool after the TTL if maxTxs is reached
272 now := time.Now().UTC().Unix()
273 _, err = txn.Exec(
274 `DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
275 l2db.maxTxs,
276 time.Unix(now-int64(l2db.ttl.Seconds()), 0),
277 )
278 if err != nil {
279 return err
280 }
281 // Delete txs that have been marked as forged / invalid after the safety period
282 _, err = txn.Exec(
283 `DELETE FROM tx_pool
284 WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
285 currentBatchNum-l2db.safetyPeriod,
286 common.PoolL2TxStateForged,
287 common.PoolL2TxStateInvalid,
288 )
289 if err != nil {
290 return err
291 }
292 return txn.Commit()
293}