hermez-node/coordinator/txmanager.go
coordinator/txmanager.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "strings"
7 "time"
8
9 "github.com/ethereum/go-ethereum/core/types"
10 "github.com/hermeznetwork/hermez-node/common"
11 "github.com/hermeznetwork/hermez-node/db/l2db"
12 "github.com/hermeznetwork/hermez-node/eth"
13 "github.com/hermeznetwork/hermez-node/log"
14 "github.com/hermeznetwork/tracerr"
15)
16
17// TxManager handles everything related to ethereum transactions: It makes the
18// call to forge, waits for transaction confirmation, and keeps checking them
19// until a number of confirmed blocks have passed.
20type TxManager struct {
21 cfg Config
22 ethClient eth.ClientInterface
23 l2DB *l2db.L2DB // Used only to mark forged txs as forged in the L2DB
24 coord *Coordinator // Used only to send messages to stop the pipeline
25 batchCh chan *BatchInfo
26 lastBlockCh chan int64
27 queue []*BatchInfo
28 lastBlock int64
29 // lastConfirmedBatch stores the last BatchNum that who's forge call was confirmed
30 lastConfirmedBatch common.BatchNum
31}
32
33// NewTxManager creates a new TxManager
34func NewTxManager(cfg *Config, ethClient eth.ClientInterface, l2DB *l2db.L2DB,
35 coord *Coordinator) *TxManager {
36 return &TxManager{
37 cfg: *cfg,
38 ethClient: ethClient,
39 l2DB: l2DB,
40 coord: coord,
41 batchCh: make(chan *BatchInfo, queueLen),
42 lastBlockCh: make(chan int64, queueLen),
43 lastBlock: -1,
44 }
45}
46
47// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
48// the smart contract via the forge call
49func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
50 t.batchCh <- batchInfo
51}
52
53// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
54func (t *TxManager) SetLastBlock(lastBlock int64) {
55 t.lastBlockCh <- lastBlock
56}
57
58func (t *TxManager) callRollupForgeBatch(ctx context.Context, batchInfo *BatchInfo) error {
59 batchInfo.Debug.Status = StatusSent
60 batchInfo.Debug.SendBlockNum = t.lastBlock + 1
61 batchInfo.Debug.SendTimestamp = time.Now()
62 batchInfo.Debug.StartToSendDelay = batchInfo.Debug.SendTimestamp.Sub(
63 batchInfo.Debug.StartTimestamp).Seconds()
64 var ethTx *types.Transaction
65 var err error
66 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
67 ethTx, err = t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
68 if err != nil {
69 if strings.Contains(err.Error(), common.AuctionErrMsgCannotForge) {
70 log.Debugw("TxManager ethClient.RollupForgeBatch", "err", err,
71 "block", t.lastBlock+1)
72 return tracerr.Wrap(err)
73 }
74 log.Errorw("TxManager ethClient.RollupForgeBatch",
75 "attempt", attempt, "err", err, "block", t.lastBlock+1,
76 "batchNum", batchInfo.BatchNum)
77 } else {
78 break
79 }
80 select {
81 case <-ctx.Done():
82 return tracerr.Wrap(common.ErrDone)
83 case <-time.After(t.cfg.EthClientAttemptsDelay):
84 }
85 }
86 if err != nil {
87 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.RollupForgeBatch: %w", err))
88 }
89 batchInfo.EthTx = ethTx
90 log.Infow("TxManager ethClient.RollupForgeBatch", "batch", batchInfo.BatchNum, "tx", ethTx.Hash().Hex())
91 t.cfg.debugBatchStore(batchInfo)
92 if err := t.l2DB.DoneForging(common.TxIDsFromL2Txs(batchInfo.L2Txs), batchInfo.BatchNum); err != nil {
93 return tracerr.Wrap(err)
94 }
95 return nil
96}
97
98func (t *TxManager) checkEthTransactionReceipt(ctx context.Context, batchInfo *BatchInfo) error {
99 txHash := batchInfo.EthTx.Hash()
100 var receipt *types.Receipt
101 var err error
102 for attempt := 0; attempt < t.cfg.EthClientAttempts; attempt++ {
103 receipt, err = t.ethClient.EthTransactionReceipt(ctx, txHash)
104 if ctx.Err() != nil {
105 continue
106 }
107 if err != nil {
108 log.Errorw("TxManager ethClient.EthTransactionReceipt",
109 "attempt", attempt, "err", err)
110 } else {
111 break
112 }
113 select {
114 case <-ctx.Done():
115 return tracerr.Wrap(common.ErrDone)
116 case <-time.After(t.cfg.EthClientAttemptsDelay):
117 }
118 }
119 if err != nil {
120 return tracerr.Wrap(fmt.Errorf("reached max attempts for ethClient.EthTransactionReceipt: %w", err))
121 }
122 batchInfo.Receipt = receipt
123 t.cfg.debugBatchStore(batchInfo)
124 return nil
125}
126
127func (t *TxManager) handleReceipt(batchInfo *BatchInfo) (*int64, error) {
128 receipt := batchInfo.Receipt
129 if receipt != nil {
130 if receipt.Status == types.ReceiptStatusFailed {
131 batchInfo.Debug.Status = StatusFailed
132 t.cfg.debugBatchStore(batchInfo)
133 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
134 return nil, tracerr.Wrap(fmt.Errorf("ethereum transaction receipt statis is failed"))
135 } else if receipt.Status == types.ReceiptStatusSuccessful {
136 batchInfo.Debug.Status = StatusMined
137 batchInfo.Debug.MineBlockNum = receipt.BlockNumber.Int64()
138 batchInfo.Debug.StartToMineBlocksDelay = batchInfo.Debug.MineBlockNum -
139 batchInfo.Debug.StartBlockNum
140 t.cfg.debugBatchStore(batchInfo)
141 if batchInfo.BatchNum > t.lastConfirmedBatch {
142 t.lastConfirmedBatch = batchInfo.BatchNum
143 }
144 confirm := t.lastBlock - receipt.BlockNumber.Int64()
145 return &confirm, nil
146 }
147 }
148 return nil, nil
149}
150
151// Run the TxManager
152func (t *TxManager) Run(ctx context.Context) {
153 next := 0
154 waitDuration := longWaitDuration
155
156 for {
157 select {
158 case <-ctx.Done():
159 log.Info("TxManager done")
160 return
161 case lastBlock := <-t.lastBlockCh:
162 t.lastBlock = lastBlock
163 case batchInfo := <-t.batchCh:
164 if err := t.callRollupForgeBatch(ctx, batchInfo); ctx.Err() != nil {
165 continue
166 } else if err != nil {
167 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch call: %v", err)})
168 continue
169 }
170 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
171 t.queue = append(t.queue, batchInfo)
172 waitDuration = t.cfg.TxManagerCheckInterval
173 case <-time.After(waitDuration):
174 if len(t.queue) == 0 {
175 continue
176 }
177 current := next
178 next = (current + 1) % len(t.queue)
179 batchInfo := t.queue[current]
180 if err := t.checkEthTransactionReceipt(ctx, batchInfo); ctx.Err() != nil {
181 continue
182 } else if err != nil { //nolint:staticcheck
183 // We can't get the receipt for the
184 // transaction, so we can't confirm if it was
185 // mined
186 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch receipt: %v", err)})
187 }
188
189 confirm, err := t.handleReceipt(batchInfo)
190 if err != nil { //nolint:staticcheck
191 // Transaction was rejected
192 t.coord.SendMsg(MsgStopPipeline{Reason: fmt.Sprintf("forgeBatch reject: %v", err)})
193 }
194 if confirm != nil && *confirm >= t.cfg.ConfirmBlocks {
195 log.Debugw("TxManager tx for RollupForgeBatch confirmed",
196 "batch", batchInfo.BatchNum)
197 t.queue = append(t.queue[:current], t.queue[current+1:]...)
198 if len(t.queue) == 0 {
199 waitDuration = longWaitDuration
200 next = 0
201 } else {
202 next = current % len(t.queue)
203 }
204 }
205 }
206 }
207}