hermez-node/coordinator/txmanager.go
coordinator/txmanager.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "math/big"
7 "time"
8
9 "github.com/ethereum/go-ethereum"
10 "github.com/ethereum/go-ethereum/accounts"
11 "github.com/ethereum/go-ethereum/accounts/abi/bind"
12 "github.com/ethereum/go-ethereum/core/types"
13 "github.com/hermeznetwork/hermez-node/common"
14 "github.com/hermeznetwork/hermez-node/db/l2db"
15 "github.com/hermeznetwork/hermez-node/eth"
16 "github.com/hermeznetwork/hermez-node/log"
17 "github.com/hermeznetwork/hermez-node/synchronizer"
18 "github.com/hermeznetwork/tracerr"
19)
20
21// TxManager handles everything related to ethereum transactions: It makes the
22// call to forge, waits for transaction confirmation, and keeps checking them
23// until a number of confirmed blocks have passed.
24type TxManager struct {
25 cfg Config
26 ethClient eth.ClientInterface
27 l2DB *l2db.L2DB // Used only to mark forged txs as forged in the L2DB
28 coord *Coordinator // Used only to send messages to stop the pipeline
29 batchCh chan *BatchInfo
30 chainID *big.Int
31 account accounts.Account
32 consts synchronizer.SCConsts
33
34 stats synchronizer.Stats
35 vars synchronizer.SCVariables
36 statsVarsCh chan statsVars
37
38 queue []*BatchInfo
39 // lastSuccessBatch stores the last BatchNum that who's forge call was confirmed
40 lastSuccessBatch common.BatchNum
41 lastPendingBatch common.BatchNum
42 lastSuccessNonce uint64
43 lastPendingNonce uint64
44}
45
46// NewTxManager creates a new TxManager
47func NewTxManager(ctx context.Context, cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
48 coord *Coordinator, scConsts *synchronizer.SCConsts, initSCVars *synchronizer.SCVariables) (*TxManager, error) {
49 chainID, err := ethClient.EthChainID()
50 if err != nil {
51 return nil, tracerr.Wrap(err)
52 }
53 address, err := ethClient.EthAddress()
54 if err != nil {
55 return nil, tracerr.Wrap(err)
56 }
57 lastSuccessNonce, err := ethClient.EthNonceAt(ctx, *address, nil)
58 if err != nil {
59 return nil, err
60 }
61 lastPendingNonce, err := ethClient.EthPendingNonceAt(ctx, *address)
62 if err != nil {
63 return nil, err
64 }
65 if lastSuccessNonce != lastPendingNonce {
66 return nil, tracerr.Wrap(fmt.Errorf("lastSuccessNonce (%v) != lastPendingNonce (%v)",
67 lastSuccessNonce, lastPendingNonce))
68 }
69 log.Infow("TxManager started", "nonce", lastSuccessNonce)
70 return &TxManager{
71 cfg: *cfg,
72 ethClient: ethClient,
73 l2DB: l2DB,
74 coord: coord,
75 batchCh: make(chan *BatchInfo, queueLen),
76 statsVarsCh: make(chan statsVars, queueLen),
77 account: accounts.Account{
78 Address: *address,
79 },
80 chainID: chainID,
81 consts: *scConsts,
82
83 vars: *initSCVars,
84
85 lastSuccessNonce: lastSuccessNonce,
86 lastPendingNonce: lastPendingNonce,
87 }, nil
88}
89
90// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
91// the smart contract via the forge call
92func (t *TxManager) AddBatch(ctx context.Context, batchInfo *BatchInfo) {
93 select {
94 case t.batchCh <- batchInfo:
95 case <-ctx.Done():
96 }
97}
98
99// SetSyncStatsVars is a thread safe method to sets the synchronizer Stats
100func (t *TxManager) SetSyncStatsVars(ctx context.Context, stats *synchronizer.Stats, vars *synchronizer.SCVariablesPtr) {
101 select {
102 case t.statsVarsCh <- statsVars{Stats: *stats, Vars: *vars}:
103 case <-ctx.Done():
104 }
105}
106
107func (t *TxManager) syncSCVars(vars synchronizer.SCVariablesPtr) {
108 if vars.Rollup != nil {
109 t.vars.Rollup = *vars.Rollup
110 }
111 if vars.Auction != nil {
112 t.vars.Auction = *vars.Auction
113 }
114 if vars.WDelayer != nil {
115 t.vars.WDelayer = *vars.WDelayer
116 }
117}
118
119// NewAuth generates a new auth object for an ethereum transaction
120func (t *TxManager) NewAuth(ctx context.Context) (*bind.TransactOpts, error) {
121 gasPrice, err := t.ethClient.EthSuggestGasPrice(ctx)
122 if err != nil {
123 return nil, tracerr.Wrap(err)
124 }
125 inc := new(big.Int).Set(gasPrice)
126 const gasPriceDiv = 100
127 inc.Div(inc, new(big.Int).SetUint64(gasPriceDiv))
128 gasPrice.Add(gasPrice, inc)
129 // log.Debugw("TxManager: transaction metadata", "gasPrice", gasPrice)
130
131 auth, err := bind.NewKeyStoreTransactorWithChainID(t.ethClient.EthKeyStore(), t.account, t.chainID)
132 if err != nil {
133 return nil, tracerr.Wrap(err)
134 }
135 auth.Value = big.NewInt(0) // in wei
136 // TODO: Calculate GasLimit based on the contents of the ForgeBatchArgs
137 auth.GasLimit = 1000000
138 auth.GasPrice = gasPrice
139 auth.Nonce = nil
140
141 return auth, nil
142}
143
144func (t *TxManager) sendRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
145 // TODO: Check if we can forge in the next blockNum, abort if we can't
146 batchInfo.Debug.Status = StatusSent
147 batchInfo.Debug.SendBlockNum = t.stats.Eth.LastBlock.Num + 1
148 batchInfo.Debug.SendTimestamp = time.Now()
149 batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
150 batchInfo.Debug.StartTimestamp).Seconds()
151 var ethTx *types.Transaction
152 var err error
153 auth, err := t.NewAuth(ctx)
154 if err != nil {
155 return tracerr.Wrap(err)
156 }
157 auth.Nonce = big.NewInt(int64(t.lastPendingNonce))
158 t.lastPendingNonce++
159 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
160 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs, auth)
161 if err != nil {
162 // if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
163 // log.Errorw("TxManager ethClient.RollupForgeBatch", "err", err,
164 // "block", t.stats.Eth.LastBlock.Num+1)
165 // return tracerr.Wrap(err)
166 // }
167 log.Errorw("TxManager ethClient.RollupForgeBatch",
168 "attempt", attempt, "err", err, "block", t.stats.Eth.LastBlock.Num+1,
169 "batchNum", batchInfo.BatchNum)
170 } else {
171 break
172 }
173 select {
174 case <-ctx.Done():
175 return tracerr.Wrap(common.ErrDone)
176 case <-time.After(t.cfg.EthClientAttemptsDelay):
177 }
178 }
179 if err != nil {
180 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
181 }
182 batchInfo.EthTx = ethTx
183 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
184 t.cfg.debugBatchStore(batchInfo)
185 t.lastPendingBatch = batchInfo.BatchNum
186 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
187 return tracerr.Wrap(err)
188 }
189 return nil
190}
191
192// checkEthTransactionReceipt takes the txHash from the BatchInfo and stores
193// the corresponding receipt if found
194func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
195 txHash := batchInfo.EthTx.Hash()
196 var receipt *types.Receipt
197 var err error
198 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
199 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
200 if ctx.Err() != nil {
201 continue
202 } else if tracerr.Unwrap(err) == ethereum.NotFound {
203 err = nil
204 break
205 } else if err != nil {
206 log.Errorw("TxManager ethClient.EthTransactionReceipt",
207 "attempt", attempt, "err", err)
208 } else {
209 break
210 }
211 select {
212 case <-ctx.Done():
213 return tracerr.Wrap(common.ErrDone)
214 case <-time.After(t.cfg.EthClientAttemptsDelay):
215 }
216 }
217 if err != nil {
218 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
219 }
220 batchInfo.Receipt = receipt
221 t.cfg.debugBatchStore(batchInfo)
222 return nil
223}
224
225func (t *TxManager) handleReceipt(ctx context.Context, batchInfo *BatchInfo) (*int64, error) {
226 receipt := batchInfo.Receipt
227 if receipt != nil {
228 if receipt.Status == types.ReceiptStatusFailed {
229 batchInfo.Debug.Status = StatusFailed
230 t.cfg.debugBatchStore(batchInfo)
231 _, err := t.ethClient.EthCall(ctx, batchInfo.EthTx, receipt.BlockNumber)
232 log.Warnw("TxManager receipt status is failed", "tx", receipt.TxHash.Hex(),
233 "batch", batchInfo.BatchNum, "block", receipt.BlockNumber.Int64(),
234 "err", err)
235 return nil, tracerr.Wrap(fmt.Errorf(
236 "ethereum transaction receipt status is failed: %w", err))
237 } else if receipt.Status == types.ReceiptStatusSuccessful {
238 batchInfo.Debug.Status = StatusMined
239 batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
240 batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
241 batchInfo.Debug.StartBlockNum
242 t.cfg.debugBatchStore(batchInfo)
243 if batchInfo.BatchNum > t.lastSuccessBatch {
244 t.lastSuccessBatch = batchInfo.BatchNum
245 }
246 confirm := t.stats.Eth.LastBlock.Num - receipt.BlockNumber.Int64()
247 return &confirm, nil
248 }
249 }
250 return nil, nil
251}
252
253// Run the TxManager
254func (t *TxManager) Run(ctx context.Context) {
255 next := 0
256 waitDuration := longWaitDuration
257
258 var statsVars statsVars
259 select {
260 case statsVars = <-t.statsVarsCh:
261 case <-ctx.Done():
262 }
263 t.stats = statsVars.Stats
264 t.syncSCVars(statsVars.Vars)
265 log.Infow("TxManager: received initial statsVars",
266 "block", t.stats.Eth.LastBlock.Num, "batch", t.stats.Eth.LastBatch)
267
268 for {
269 select {
270 case <-ctx.Done():
271 log.Info("TxManager done")
272 return
273 case statsVars := <-t.statsVarsCh:
274 t.stats = statsVars.Stats
275 t.syncSCVars(statsVars.Vars)
276 case batchInfo := <-t.batchCh:
277 if err := t.sendRollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
278 continue
279 } else if err != nil {
280 // If we reach here it's because our ethNode has
281 // been unable to send the transaction to
282 // ethereum. This could be due to the ethNode
283 // failure, or an invalid transaction (that
284 // can't be mined)
285 t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch send: %v", err)})
286 continue
287 }
288 t.queue = append(t.queue, batchInfo)
289 waitDuration = t.cfg.TxManagerCheckInterval
290 case <-time.After(waitDuration):
291 if len(t.queue) == 0 {
292 waitDuration = longWaitDuration
293 continue
294 }
295 current := next
296 next = (current + 1) % len(t.queue)
297 batchInfo := t.queue[current]
298 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
299 continue
300 } else if err != nil { //nolint:staticcheck
301 // Our ethNode is giving an error different
302 // than "not found" when getting the receipt
303 // for the transaction, so we can't figure out
304 // if it was not mined, mined and succesfull or
305 // mined and failed. This could be due to the
306 // ethNode failure.
307 t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
308 }
309
310 confirm, err := t.handleReceipt(ctx, batchInfo)
311 if ctx.Err() != nil {
312 continue
313 } else if err != nil { //nolint:staticcheck
314 // Transaction was rejected
315 t.queue = append(t.queue[:current], t.queue[current+1:]...)
316 if len(t.queue) == 0 {
317 next = 0
318 } else {
319 next = current % len(t.queue)
320 }
321 t.coord.SendMsg(ctx, MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
322 }
323 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
324 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
325 "batch", batchInfo.BatchNum)
326 t.queue = append(t.queue[:current], t.queue[current+1:]...)
327 if len(t.queue) == 0 {
328 next = 0
329 } else {
330 next = current % len(t.queue)
331 }
332 }
333 }
334 }
335}
336
337// nolint reason: this function will be used in the future
338//nolint:unused
339func (t *TxManager) canForge(stats *synchronizer.Stats, blockNum int64) bool {
340 return canForge(&t.consts.Auction, &t.vars.Auction,
341 &stats.Sync.Auction.CurrentSlot, &stats.Sync.Auction.NextSlot,
342 t.cfg.ForgerAddress, blockNum)
343}