hermez-node/coordinator/coordinator.go
coordinator/coordinator.goRaw
1package coordinator
2
3import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 ethCommon "github.com/ethereum/go-ethereum/common"
10 "github.com/ethereum/go-ethereum/core/types"
11 "github.com/hermeznetwork/hermez-node/batchbuilder"
12 "github.com/hermeznetwork/hermez-node/common"
13 "github.com/hermeznetwork/hermez-node/db/historydb"
14 "github.com/hermeznetwork/hermez-node/eth"
15 "github.com/hermeznetwork/hermez-node/log"
16 "github.com/hermeznetwork/hermez-node/synchronizer"
17 "github.com/hermeznetwork/hermez-node/txselector"
18)
19
20var errTODO = fmt.Errorf("TODO")
21
22// ErrDone is returned when the function is stopped asynchronously via a done
23// (terminated) context. It doesn't indicate an error.
24var ErrDone = fmt.Errorf("done")
25
26// Config contains the Coordinator configuration
27type Config struct {
28 ForgerAddress ethCommon.Address
29 ConfirmBlocks int64
30}
31
32// Coordinator implements the Coordinator type
33type Coordinator struct {
34 // State
35 forging bool
36 batchNum common.BatchNum
37 serverProofPool *ServerProofPool
38 consts synchronizer.SCConsts
39 vars synchronizer.SCVariables
40
41 cfg Config
42
43 hdb *historydb.HistoryDB
44 txsel *txselector.TxSelector
45 batchBuilder *batchbuilder.BatchBuilder
46
47 ethClient eth.ClientInterface
48
49 msgCh chan interface{}
50 ctx context.Context
51 wg sync.WaitGroup
52 cancel context.CancelFunc
53
54 pipelineCtx context.Context
55 pipelineWg sync.WaitGroup
56 pipelineCancel context.CancelFunc
57
58 txManager *TxManager
59}
60
61// NewCoordinator creates a new Coordinator
62func NewCoordinator(cfg Config,
63 hdb *historydb.HistoryDB,
64 txsel *txselector.TxSelector,
65 bb *batchbuilder.BatchBuilder,
66 serverProofs []ServerProofInterface,
67 ethClient eth.ClientInterface,
68 scConsts *synchronizer.SCConsts,
69 initSCVars *synchronizer.SCVariables,
70) *Coordinator { // once synchronizer is ready, synchronizer.Synchronizer will be passed as parameter here
71 serverProofPool := NewServerProofPool(len(serverProofs))
72 for _, serverProof := range serverProofs {
73 serverProofPool.Add(serverProof)
74 }
75
76 txManager := NewTxManager(ethClient, cfg.ConfirmBlocks)
77
78 ctx, cancel := context.WithCancel(context.Background())
79 c := Coordinator{
80 forging: false,
81 batchNum: -1,
82 serverProofPool: serverProofPool,
83 consts: *scConsts,
84 vars: *initSCVars,
85
86 cfg: cfg,
87
88 hdb: hdb,
89 txsel: txsel,
90 batchBuilder: bb,
91
92 ethClient: ethClient,
93
94 msgCh: make(chan interface{}),
95 ctx: ctx,
96 // wg
97 cancel: cancel,
98
99 txManager: txManager,
100 }
101 return &c
102}
103
104// MsgSyncStats indicates an update to the Synchronizer stats
105type MsgSyncStats struct {
106 Stats synchronizer.Stats
107}
108
109// MsgSyncSCVars indicates an update to Smart Contract Vars
110type MsgSyncSCVars struct {
111 Rollup *common.RollupVariables
112 Auction *common.AuctionVariables
113 WDelayer *common.WDelayerVariables
114}
115
116// MsgSyncReorg indicates a reorg
117type MsgSyncReorg struct {
118}
119
120// SendMsg is a thread safe method to pass a message to the Coordinator
121func (c *Coordinator) SendMsg(msg interface{}) {
122 c.msgCh <- msg
123}
124
125func (c *Coordinator) handleMsgSyncSCVars(msg *MsgSyncSCVars) {
126 if msg.Rollup != nil {
127 c.vars.Rollup = *msg.Rollup
128 }
129 if msg.Auction != nil {
130 c.vars.Auction = *msg.Auction
131 }
132 if msg.WDelayer != nil {
133 c.vars.WDelayer = *msg.WDelayer
134 }
135}
136
137func (c *Coordinator) handleMsgSyncStats(stats *synchronizer.Stats) error {
138 if !stats.Synced() {
139 return nil
140 }
141 c.txManager.SetLastBlock(stats.Eth.LastBlock)
142
143 anyoneForge := false
144 if stats.Sync.Auction.CurrentSlot.BatchesLen == 0 &&
145 c.consts.Auction.RelativeBlock(stats.Eth.LastBlock) > int64(c.vars.Auction.SlotDeadline) {
146 log.Debug("Coordinator: anyone can forge in the current slot (slotDeadline passed)")
147 anyoneForge = true
148 }
149 if stats.Sync.Auction.CurrentSlot.Forger != c.cfg.ForgerAddress && !anyoneForge {
150 if c.forging {
151 log.Info("Coordinator: forging state end")
152 c.forging = false
153 c.PipelineStop()
154 }
155 // log.Debug("Coordinator: not in forge time") // DBG
156 return nil
157 }
158 // log.Debug("Coordinator: forge time") // DBG
159 if !c.forging {
160 // Start pipeline from a batchNum state taken from synchronizer
161 log.Info("Coordinator: forging state begin")
162 c.batchNum = common.BatchNum(stats.Sync.LastBatch)
163 err := c.txsel.Reset(c.batchNum)
164 if err != nil {
165 log.Errorw("Coordinator: TxSelector.Reset", "error", err)
166 return err
167 }
168 err = c.batchBuilder.Reset(c.batchNum, true)
169 if err != nil {
170 log.Errorw("Coordinator: BatchBuilder.Reset", "error", err)
171 return err
172 }
173 c.forging = true
174 c.PipelineStart()
175 }
176 return nil
177}
178
179// Start the coordinator
180func (c *Coordinator) Start() {
181 c.wg.Add(1)
182 go func() {
183 c.txManager.Run(c.ctx)
184 c.wg.Done()
185 }()
186
187 c.wg.Add(1)
188 go func() {
189 for {
190 select {
191 case <-c.ctx.Done():
192 log.Info("Coordinator done")
193 c.wg.Done()
194 return
195 case msg := <-c.msgCh:
196 switch msg := msg.(type) {
197 case MsgSyncStats:
198 stats := msg.Stats
199 if err := c.handleMsgSyncStats(&stats); err != nil {
200 log.Errorw("Coordinator.handleMsgSyncStats error", "err", err)
201 }
202 case MsgSyncReorg:
203 if err := c.handleReorg(); err != nil {
204 log.Errorw("Coordinator.handleReorg error", "err", err)
205 }
206 case MsgSyncSCVars:
207 c.handleMsgSyncSCVars(&msg)
208 default:
209 log.Fatalw("Coordinator Unexpected Coordinator msg of type %T: %+v", msg, msg)
210 }
211 }
212 }
213 }()
214}
215
216// Stop the coordinator
217func (c *Coordinator) Stop() {
218 log.Infow("Stopping coordinator...")
219 c.cancel()
220 c.wg.Wait()
221 if c.forging {
222 c.forging = false
223 c.PipelineStop()
224 }
225}
226
227// PipelineStart starts the forging pipeline
228func (c *Coordinator) PipelineStart() {
229 c.pipelineCtx, c.pipelineCancel = context.WithCancel(context.Background())
230
231 queueSize := 1
232 batchChSentServerProof := make(chan *BatchInfo, queueSize)
233
234 c.pipelineWg.Add(1)
235 go func() {
236 for {
237 select {
238 case <-c.pipelineCtx.Done():
239 log.Debug("Pipeline forgeSendServerProof loop done")
240 c.pipelineWg.Done()
241 return
242 default:
243 c.batchNum = c.batchNum + 1
244 batchInfo, err := c.forgeSendServerProof(c.pipelineCtx, c.batchNum)
245 if err == ErrDone {
246 continue
247 }
248 if err != nil {
249 log.Errorw("forgeSendServerProof", "err", err)
250 continue
251 }
252 batchChSentServerProof <- batchInfo
253 }
254 }
255 }()
256
257 c.pipelineWg.Add(1)
258 go func() {
259 for {
260 select {
261 case <-c.pipelineCtx.Done():
262 log.Debug("Pipeline waitServerProofSendEth loop done")
263 c.pipelineWg.Done()
264 return
265 case batchInfo := <-batchChSentServerProof:
266 err := c.waitServerProof(c.pipelineCtx, batchInfo)
267 if err == ErrDone {
268 continue
269 }
270 if err != nil {
271 log.Errorw("waitServerProof", "err", err)
272 continue
273 }
274 c.txManager.AddBatch(batchInfo)
275 }
276 }
277 }()
278}
279
280// PipelineStop stops the forging pipeline
281func (c *Coordinator) PipelineStop() {
282 log.Debug("Stopping pipeline...")
283 c.pipelineCancel()
284 c.pipelineWg.Wait()
285}
286
287// TxManager handles everything related to ethereum transactions: It makes the
288// call to forge, waits for transaction confirmation, and keeps checking them
289// until a number of confirmed blocks have passed.
290type TxManager struct {
291 ethClient eth.ClientInterface
292 batchCh chan *BatchInfo
293 lastBlockCh chan int64
294 queue []*BatchInfo
295 confirmation int64
296 lastBlock int64
297}
298
299// NewTxManager creates a new TxManager
300func NewTxManager(ethClient eth.ClientInterface, confirmation int64) *TxManager {
301 return &TxManager{
302 ethClient: ethClient,
303 // TODO: Find best queue size
304 batchCh: make(chan *BatchInfo, 16), //nolint:gomnd
305 // TODO: Find best queue size
306 lastBlockCh: make(chan int64, 16), //nolint:gomnd
307 confirmation: confirmation,
308 lastBlock: -1,
309 }
310}
311
312// AddBatch is a thread safe method to pass a new batch TxManager to be sent to
313// the smart contract via the forge call
314func (t *TxManager) AddBatch(batchInfo *BatchInfo) {
315 t.batchCh <- batchInfo
316}
317
318// SetLastBlock is a thread safe method to pass the lastBlock to the TxManager
319func (t *TxManager) SetLastBlock(lastBlock int64) {
320 t.lastBlockCh <- lastBlock
321}
322
323const waitTime = 200 * time.Millisecond
324const longWaitTime = 999 * time.Hour
325
326// Run the TxManager
327func (t *TxManager) Run(ctx context.Context) {
328 next := 0
329 d := time.Duration(longWaitTime)
330 for {
331 select {
332 case <-ctx.Done():
333 log.Info("TxManager done")
334 return
335 case lastBlock := <-t.lastBlockCh:
336 t.lastBlock = lastBlock
337 case batchInfo := <-t.batchCh:
338 ethTx, err := t.ethClient.RollupForgeBatch(batchInfo.ForgeBatchArgs)
339 if err != nil {
340 // TODO: Figure out different error cases and handle them properly
341 log.Errorw("TxManager ethClient.RollupForgeBatch", "err", err)
342 continue
343 }
344 log.Debugf("ethClient ForgeCall sent, batchNum: %d", batchInfo.BatchNum)
345 batchInfo.EthTx = ethTx
346 t.queue = append(t.queue, batchInfo)
347 d = waitTime
348 case <-time.After(d):
349 if len(t.queue) == 0 {
350 continue
351 }
352 batchInfo := t.queue[next]
353 txID := batchInfo.EthTx.Hash()
354 receipt, err := t.ethClient.EthTransactionReceipt(ctx, txID)
355 if err != nil {
356 log.Errorw("TxManager ethClient.EthTransactionReceipt", "err", err)
357 // TODO: Figure out different error cases and handle them properly
358 // TODO: Notify the Coordinator to maybe reset the pipeline
359 continue
360 }
361
362 if receipt != nil {
363 if receipt.Status == types.ReceiptStatusFailed {
364 log.Errorw("TxManager receipt status is failed", "receipt", receipt)
365 } else if receipt.Status == types.ReceiptStatusSuccessful {
366 if t.lastBlock-receipt.BlockNumber.Int64() >= t.confirmation {
367 log.Debugw("TxManager tx for RollupForgeBatch confirmed", "batchNum", batchInfo.BatchNum)
368 t.queue = t.queue[1:]
369 if len(t.queue) == 0 {
370 d = longWaitTime
371 }
372 }
373 }
374 }
375 if len(t.queue) == 0 {
376 next = 0
377 } else {
378 next = (next + 1) % len(t.queue)
379 }
380 }
381 }
382}
383
384// forgeSendServerProof the next batch, wait for a proof server to be available and send the
385// circuit inputs to the proof server.
386func (c *Coordinator) forgeSendServerProof(ctx context.Context, batchNum common.BatchNum) (*BatchInfo, error) {
387 // remove transactions from the pool that have been there for too long
388 err := c.purgeRemoveByTimeout()
389 if err != nil {
390 return nil, err
391 }
392
393 batchInfo := BatchInfo{BatchNum: batchNum} // to accumulate metadata of the batch
394
395 var poolL2Txs []common.PoolL2Tx
396 // var feesInfo
397 var l1UserTxsExtra, l1OperatorTxs []common.L1Tx
398 // 1. Decide if we forge L2Tx or L1+L2Tx
399 if c.shouldL1L2Batch() {
400 // 2a: L1+L2 txs
401 // l1UserTxs, toForgeL1TxsNumber := c.hdb.GetNextL1UserTxs() // TODO once HistoryDB is ready, uncomment
402 var l1UserTxs []common.L1Tx = nil // tmp, depends on HistoryDB
403 l1UserTxsExtra, l1OperatorTxs, poolL2Txs, err = c.txsel.GetL1L2TxSelection([]common.Idx{}, batchNum, l1UserTxs) // TODO once feesInfo is added to method return, add the var
404 if err != nil {
405 return nil, err
406 }
407 } else {
408 // 2b: only L2 txs
409 _, poolL2Txs, err = c.txsel.GetL2TxSelection([]common.Idx{}, batchNum) // TODO once feesInfo is added to method return, add the var
410 if err != nil {
411 return nil, err
412 }
413 l1UserTxsExtra = nil
414 l1OperatorTxs = nil
415 }
416
417 // Run purger to invalidate transactions that become invalid beause of
418 // the poolL2Txs selected. Will mark as invalid the txs that have a
419 // (fromIdx, nonce) which already appears in the selected txs (includes
420 // all the nonces smaller than the current one)
421 err = c.purgeInvalidDueToL2TxsSelection(poolL2Txs)
422 if err != nil {
423 return nil, err
424 }
425
426 // 3. Save metadata from TxSelector output for BatchNum
427 // batchInfo.SetTxsInfo(l1UserTxsExtra, l1OperatorTxs, poolL2Txs) // TODO feesInfo
428 batchInfo.L1UserTxsExtra = l1UserTxsExtra
429 batchInfo.L1OperatorTxs = l1OperatorTxs
430 batchInfo.L2Txs = poolL2Txs
431
432 // 4. Call BatchBuilder with TxSelector output
433 configBatch := &batchbuilder.ConfigBatch{
434 ForgerAddress: c.cfg.ForgerAddress,
435 }
436 zkInputs, err := c.batchBuilder.BuildBatch([]common.Idx{}, configBatch, l1UserTxsExtra, l1OperatorTxs, poolL2Txs, nil) // TODO []common.TokenID --> feesInfo
437 if err != nil {
438 return nil, err
439 }
440
441 // 5. Save metadata from BatchBuilder output for BatchNum
442 batchInfo.ZKInputs = zkInputs
443
444 // 6. Wait for an available server proof blocking call
445 serverProof, err := c.serverProofPool.Get(ctx)
446 if err != nil {
447 return nil, err
448 }
449 batchInfo.ServerProof = serverProof
450 defer func() {
451 // If there's an error further on, add the serverProof back to
452 // the pool
453 if err != nil {
454 c.serverProofPool.Add(serverProof)
455 }
456 }()
457
458 // 7. Call the selected idle server proof with BatchBuilder output,
459 // save server proof info for batchNum
460 err = batchInfo.ServerProof.CalculateProof(zkInputs)
461 if err != nil {
462 return nil, err
463 }
464
465 return &batchInfo, nil
466}
467
468// waitServerProof gets the generated zkProof & sends it to the SmartContract
469func (c *Coordinator) waitServerProof(ctx context.Context, batchInfo *BatchInfo) error {
470 proof, err := batchInfo.ServerProof.GetProof(ctx) // blocking call, until not resolved don't continue. Returns when the proof server has calculated the proof
471 if err != nil {
472 return err
473 }
474 c.serverProofPool.Add(batchInfo.ServerProof)
475 batchInfo.ServerProof = nil
476 batchInfo.Proof = proof
477 batchInfo.ForgeBatchArgs = c.prepareForgeBatchArgs(batchInfo)
478 batchInfo.TxStatus = TxStatusPending
479
480 // TODO(FUTURE) once tx data type is defined, store ethTx (returned by ForgeCall)
481 // TBD if use ethTxStore as a disk k-v database, or use a Queue
482 // tx, err := c.ethTxStore.NewTx()
483 // if err != nil {
484 // return err
485 // }
486 // tx.Put(ethTx.Hash(), ethTx.Bytes())
487 // if err := tx.Commit(); err!=nil {
488 // return nil
489 // }
490
491 return nil
492}
493
494func (c *Coordinator) handleReorg() error {
495 return nil // TODO
496}
497
498// isForgeSequence returns true if the node is the Forger in the current ethereum block
499// func (c *Coordinator) isForgeSequence() (bool, error) {
500// // TODO: Consider checking if we can forge by quering the Synchronizer instead of using ethClient
501// blockNum, err := c.ethClient.EthLastBlock()
502// if err != nil {
503// return false, err
504// }
505// addr, err := c.ethClient.EthAddress()
506// if err != nil {
507// return false, err
508// }
509// return c.ethClient.AuctionCanForge(*addr, blockNum+1)
510// }
511
512func (c *Coordinator) purgeRemoveByTimeout() error {
513 return nil // TODO
514}
515
516func (c *Coordinator) purgeInvalidDueToL2TxsSelection(l2Txs []common.PoolL2Tx) error {
517 return nil // TODO
518}
519
520func (c *Coordinator) shouldL1L2Batch() bool {
521 return false // TODO
522}
523
524func (c *Coordinator) prepareForgeBatchArgs(batchInfo *BatchInfo) *eth.RollupForgeBatchArgs {
525 // TODO
526 return ð.RollupForgeBatchArgs{}
527}