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